diff --git a/src/streaming/event_parser/common/types.rs b/src/streaming/event_parser/common/types.rs index f82f51e..f8a6a4f 100755 --- a/src/streaming/event_parser/common/types.rs +++ b/src/streaming/event_parser/common/types.rs @@ -359,7 +359,7 @@ fn extract_swap_context(event: &DexEvent) -> ( fn extract_swap_data_from_instructions( event: &DexEvent, instructions: impl Iterator, - current_index: i8, + current_index: i32, accounts: &[Pubkey], ) -> Option { let (mut swap_data, fm, tm, uft, utt, fv, tv) = extract_swap_context(event); @@ -371,8 +371,13 @@ fn extract_swap_data_from_instructions( let to_mint = tm.unwrap_or_default(); let from_mint = fm.unwrap_or_default(); - for instruction in instructions.skip((current_index + 1) as usize) { - let program_id = accounts[instruction.program_id_index()]; + let skip_count = (current_index + 1).max(0) as usize; + for instruction in instructions.skip(skip_count) { + let program_id_index = instruction.program_id_index(); + let program_id = match accounts.get(program_id_index) { + Some(&pid) => pid, + None => break, + }; if !SYSTEM_PROGRAMS.contains(&program_id) { break; } @@ -383,19 +388,31 @@ fn extract_swap_data_from_instructions( continue; } - let get_pubkey = |i: usize| accounts[accs[i] as usize]; + let get_pubkey = |i: usize| -> Option { + let idx = accs.get(i).copied().map(|b| b as usize)?; + accounts.get(idx).copied() + }; let (source, destination, amount) = match data[0] { - 12 if accs.len() >= 4 => { + 12 if accs.len() >= 4 && data.len() >= 9 => { let amt = u64::from_le_bytes(data[1..9].try_into().unwrap()); - (get_pubkey(0), get_pubkey(2), amt) + match (get_pubkey(0), get_pubkey(2)) { + (Some(s), Some(d)) => (s, d, amt), + _ => continue, + } } - 3 if accs.len() >= 3 => { + 3 if accs.len() >= 3 && data.len() >= 9 => { let amt = u64::from_le_bytes(data[1..9].try_into().unwrap()); - (get_pubkey(0), get_pubkey(1), amt) + match (get_pubkey(0), get_pubkey(1)) { + (Some(s), Some(d)) => (s, d, amt), + _ => continue, + } } - 2 if accs.len() >= 2 => { + 2 if accs.len() >= 2 && data.len() >= 12 => { let amt = u64::from_le_bytes(data[4..12].try_into().unwrap()); - (get_pubkey(0), get_pubkey(1), amt) + match (get_pubkey(0), get_pubkey(1)) { + (Some(s), Some(d)) => (s, d, amt), + _ => continue, + } } _ => continue, }; @@ -450,7 +467,7 @@ fn extract_swap_data_from_instructions( pub fn parse_swap_data_from_next_instructions( event: &DexEvent, inner_instruction: &solana_transaction_status::InnerInstructions, - current_index: i8, + current_index: i32, accounts: &[Pubkey], ) -> Option { extract_swap_data_from_instructions( @@ -465,7 +482,7 @@ pub fn parse_swap_data_from_next_instructions( pub fn parse_swap_data_from_next_grpc_instructions( event: &DexEvent, inner_instruction: &yellowstone_grpc_proto::prelude::InnerInstructions, - current_index: i8, + current_index: i32, accounts: &[Pubkey], ) -> Option { extract_swap_data_from_instructions( diff --git a/src/streaming/event_parser/core/event_parser.rs b/src/streaming/event_parser/core/event_parser.rs index db3b436..0c64eb0 100644 --- a/src/streaming/event_parser/core/event_parser.rs +++ b/src/streaming/event_parser/core/event_parser.rs @@ -82,7 +82,7 @@ impl EventParser { .collect(); // 解析指令事件 let instructions = &message.instructions; - let recent_blockhash = if message.recent_blockhash.is_empty() { + let recent_blockhash = if message.recent_blockhash.len() != 32 { None } else { Some(solana_sdk::bs58::encode(&message.recent_blockhash).into_string()) @@ -434,7 +434,7 @@ impl EventParser { if let Some(swap_data) = parse_swap_data_from_next_grpc_instructions( &event, inner_instructions_ref, - current_inner_idx as i8, + current_inner_idx, accounts, ) { event.metadata_mut().set_swap_data(swap_data); @@ -602,7 +602,7 @@ impl EventParser { parse_swap_data_from_next_instructions( &event, inner_instructions_ref, - current_inner_idx as i8, + current_inner_idx, accounts, ) } else { diff --git a/src/streaming/event_parser/core/global_state.rs b/src/streaming/event_parser/core/global_state.rs index 415d7d0..2c64538 100644 --- a/src/streaming/event_parser/core/global_state.rs +++ b/src/streaming/event_parser/core/global_state.rs @@ -55,10 +55,11 @@ impl GlobalState { .map(|entry| *entry.key()) .collect(); - // Remove old signatures atomically + // Remove old signatures atomically; only decrement count when entry was present for signature in signatures_to_remove { - self.signature_data.remove(&signature); - self.signature_count.fetch_sub(1, Ordering::Relaxed); + if self.signature_data.remove(&signature).is_some() { + self.signature_count.fetch_sub(1, Ordering::Relaxed); + } } }