Merge pull request #70 from 0xfnzero/fix-grpc-reconnect-bad-protobuf

# Conflicts:
#	src/streaming/grpc/pool.rs

src/streaming/grpc/pool.rs

@@ -1,5 +1,6 @@
 use super::types::{AccountPretty, BlockMetaPretty, TransactionPretty};
 use crate::streaming::event_parser::common::high_performance_clock::get_high_perf_clock;
+use log::warn;
 use solana_sdk::{pubkey::Pubkey, signature::Signature};
 use std::collections::VecDeque;
 use std::ops::DerefMut;
@@ -95,35 +96,56 @@ impl PooledAccountPretty {
     /// 从 gRPC 更新重置数据
     pub fn reset_from_update(&mut self, account_update: SubscribeUpdateAccount) -> bool {
         let Some(account_info) = account_update.account else {
-            log::debug!("drop account update without account payload");
+            warn!("grpc account update missing nested account, skipping");
             return false;
         };
 
         self.account.slot = account_update.slot;
         self.account.signature = if let Some(txn_signature) = account_info.txn_signature {
+            if txn_signature.len() != 64 {
+                warn!(
+                    "grpc account update has invalid signature length {}, skipping",
+                    txn_signature.len()
+                );
+                return false;
+            }
             match Signature::try_from(txn_signature.as_slice()) {
-                Ok(signature) => signature,
+                Ok(sig) => sig,
                 Err(_) => {
-                    log::debug!("drop account update with invalid transaction signature");
+                    warn!("grpc account update has invalid signature bytes, skipping");
                     return false;
                 }
             }
         } else {
             Signature::default()
         };
+        if account_info.pubkey.len() != 32 {
+            warn!(
+                "grpc account update has invalid pubkey length {}, skipping",
+                account_info.pubkey.len()
+            );
+            return false;
+        }
         self.account.pubkey = match Pubkey::try_from(account_info.pubkey.as_slice()) {
             Ok(pubkey) => pubkey,
             Err(_) => {
-                log::debug!("drop account update with invalid account pubkey");
+                warn!("grpc account update has invalid pubkey bytes, skipping");
                 return false;
             }
         };
         self.account.executable = account_info.executable;
         self.account.lamports = account_info.lamports;
+        if account_info.owner.len() != 32 {
+            warn!(
+                "grpc account update has invalid owner length {}, skipping",
+                account_info.owner.len()
+            );
+            return false;
+        }
         self.account.owner = match Pubkey::try_from(account_info.owner.as_slice()) {
             Ok(owner) => owner,
             Err(_) => {
-                log::debug!("drop account update with invalid owner pubkey");
+                warn!("grpc account update has invalid owner bytes, skipping");
                 return false;
             }
         };
@@ -295,10 +317,18 @@ impl PooledTransactionPretty {
         block_time: Option<Timestamp>,
     ) -> bool {
         let Some(tx) = tx_update.transaction else {
-            log::debug!("drop transaction update without transaction payload");
+            warn!("grpc transaction update missing nested transaction, skipping");
             return false;
         };
 
+        if tx.signature.len() != 64 {
+            warn!(
+                "grpc transaction update has invalid signature length {}, skipping",
+                tx.signature.len()
+            );
+            return false;
+        }
+
         self.transaction.slot = tx_update.slot;
         self.transaction.tx_index = Some(tx.index);
         self.transaction.block_time = block_time;
@@ -306,7 +336,7 @@ impl PooledTransactionPretty {
         self.transaction.signature = match Signature::try_from(tx.signature.as_slice()) {
             Ok(signature) => signature,
             Err(_) => {
-                log::debug!("drop transaction update with invalid signature");
+                warn!("grpc transaction update has invalid signature bytes, skipping");
                 return false;
             }
         };
@@ -400,14 +430,17 @@ impl Default for PoolManager {
 /// 工厂函数用于创建优化的 EventPretty
 impl EventPrettyPool {
     /// 创建账户事件 - 使用对象池优化
-    pub fn create_account_event_optimized(&self, update: SubscribeUpdateAccount) -> AccountPretty {
+    pub fn create_account_event_optimized(
+        &self,
+        update: SubscribeUpdateAccount,
+    ) -> Option<AccountPretty> {
         let mut pooled_account = self.acquire_account();
         if !pooled_account.reset_from_update(update) {
-            return AccountPretty::default();
+            return None;
         }
         // 移动数据而不是克隆，避免多余的内存分配
         let result = std::mem::replace(pooled_account.deref_mut(), AccountPretty::default());
-        result
+        Some(result)
     }
 
     /// 创建区块事件 - 使用对象池优化
@@ -428,14 +461,14 @@ impl EventPrettyPool {
         &self,
         update: SubscribeUpdateTransaction,
         block_time: Option<Timestamp>,
-    ) -> TransactionPretty {
+    ) -> Option<TransactionPretty> {
         let mut pooled_tx = self.acquire_transaction();
         if !pooled_tx.reset_from_update(update, block_time) {
-            return TransactionPretty::default();
+            return None;
         }
         // 移动数据而不是克隆
         let result = std::mem::replace(pooled_tx.deref_mut(), TransactionPretty::default());
-        result
+        Some(result)
     }
 }
 
@@ -448,7 +481,7 @@ pub mod factory {
     use super::*;
 
     /// 使用对象池创建账户事件（推荐用于高性能场景）
-    pub fn create_account_pretty_pooled(update: SubscribeUpdateAccount) -> AccountPretty {
+    pub fn create_account_pretty_pooled(update: SubscribeUpdateAccount) -> Option<AccountPretty> {
         GLOBAL_POOL_MANAGER.get_event_pool().create_account_event_optimized(update)
     }
 
@@ -464,7 +497,7 @@ pub mod factory {
     pub fn create_transaction_pretty_pooled(
         update: SubscribeUpdateTransaction,
         block_time: Option<Timestamp>,
-    ) -> TransactionPretty {
+    ) -> Option<TransactionPretty> {
         GLOBAL_POOL_MANAGER.get_event_pool().create_transaction_event_optimized(update, block_time)
     }
 }

src/streaming/yellowstone_grpc.rs

@@ -155,6 +155,7 @@ impl YellowstoneGrpc {
             return Err(anyhow!("Already subscribed. Use update_subscription() to modify filters"));
         }
 
+        let clear_active_on_drop = ActiveSubscriptionGuard(self.active_subscription.clone());
         let mut metrics_handle = None;
         // 启动自动性能监控（如果启用）
         if self.config.enable_metrics {
@@ -187,7 +188,9 @@ impl YellowstoneGrpc {
         let order_timeout_ms = self.config.order_timeout_ms;
         let micro_batch_us = self.config.micro_batch_us;
 
+        let active_subscription = self.active_subscription.clone();
         let stream_handle = tokio::spawn(async move {
+            let _clear_active = ActiveSubscriptionGuard(active_subscription);
             let mut slot_buffer = SlotBuffer::new();
             let mut micro_batch = MicroBatchBuffer::new();
             let mut last_slot = 0u64;
@@ -218,7 +221,11 @@ impl YellowstoneGrpc {
                                 let created_at = msg.created_at;
                                 match msg.update_oneof {
                                     Some(UpdateOneof::Account(account)) => {
-                                        let account_pretty = factory::create_account_pretty_pooled(account);
+                                        let Some(account_pretty) =
+                                            factory::create_account_pretty_pooled(account)
+                                        else {
+                                            continue;
+                                        };
                                         log::debug!("Received account: {:?}", account_pretty);
                                         if let Err(e) = process_grpc_transaction(
                                             EventPretty::Account(account_pretty),
@@ -248,7 +255,11 @@ impl YellowstoneGrpc {
                                         }
                                     }
                                     Some(UpdateOneof::Transaction(sut)) => {
-                                        let transaction_pretty = factory::create_transaction_pretty_pooled(sut, created_at);
+                                        let Some(transaction_pretty) =
+                                            factory::create_transaction_pretty_pooled(sut, created_at)
+                                        else {
+                                            continue;
+                                        };
                                         log::debug!(
                                             "Received transaction: {} at slot {}",
                                             transaction_pretty.signature,
@@ -343,6 +354,7 @@ impl YellowstoneGrpc {
         let mut handle_guard = self.subscription_handle.lock().await;
         *handle_guard = Some(subscription_handle);
 
+        std::mem::forget(clear_active_on_drop);
         Ok(())
     }
 
@@ -540,6 +552,14 @@ fn has_buffered_events(
     }
 }
 
+struct ActiveSubscriptionGuard(Arc<AtomicBool>);
+
+impl Drop for ActiveSubscriptionGuard {
+    fn drop(&mut self) {
+        self.0.store(false, Ordering::Release);
+    }
+}
+
 // 实现 Clone trait 以支持模块间共享
 impl Clone for YellowstoneGrpc {
     fn clone(&self) -> Self {

src/streaming/yellowstone_sub_system.rs

@@ -59,8 +59,11 @@ impl YellowstoneGrpc {
                         let created_at = msg.created_at;
                         match msg.update_oneof {
                             Some(UpdateOneof::Transaction(sut)) => {
-                                let transaction_pretty =
+                                let Some(transaction_pretty) =
-                                    factory::create_transaction_pretty_pooled(sut, created_at);
+                                    factory::create_transaction_pretty_pooled(sut, created_at)
+                                else {
+                                    continue;
+                                };
                                 let event_pretty = EventPretty::Transaction(transaction_pretty);
                                 if let Err(e) =
                                     Self::process_system_transaction(event_pretty, &*callback).await