fix: post-merge robustness - global_state cleanup count, swap bounds, blockhash & i32 index

- global_state: only decrement signature_count when remove() actually removed entry
- types: bounds-check program_id_index and account indices in swap extraction
- types: require data.len() >= 9/12 before slicing token instruction data
- types: use i32 for current_index and safe skip count to avoid i8 truncation/UB
- event_parser: encode recent_blockhash only when len() == 32; pass i32 to swap parsers

Made-with: Cursor

src/streaming/event_parser/common/types.rs

@@ -359,7 +359,7 @@ fn extract_swap_context(event: &DexEvent) -> (
 fn extract_swap_data_from_instructions<I: InnerInstructionLike>(
     event: &DexEvent,
     instructions: impl Iterator<Item = I>,
-    current_index: i8,
+    current_index: i32,
     accounts: &[Pubkey],
 ) -> Option<SwapData> {
     let (mut swap_data, fm, tm, uft, utt, fv, tv) = extract_swap_context(event);
@@ -371,8 +371,13 @@ fn extract_swap_data_from_instructions<I: InnerInstructionLike>(
     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<I: InnerInstructionLike>(
             continue;
         }
 
-        let get_pubkey = |i: usize| accounts[accs[i] as usize];
+        let get_pubkey = |i: usize| -> Option<Pubkey> {
+            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<I: InnerInstructionLike>(
 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<SwapData> {
     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<SwapData> {
     extract_swap_data_from_instructions(

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 {

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);
+            }
         }
     }