From 63af06edd3e41da7f55938cdc632362fe48e6a08 Mon Sep 17 00:00:00 2001 From: Wood Date: Sat, 7 Mar 2026 16:59:07 +0800 Subject: [PATCH 1/2] feat(events): add recent_blockhash to EventMetadata - EventMetadata: add recent_blockhash (Option>) - gRPC path: extract from message.recent_blockhash, thread through parse_instruction_events_from_grpc_transaction and parse_events_from_grpc_instruction - VersionedTransaction path: get from message.recent_blockhash(), pass into parse_events_from_instruction - BlockMeta event: pass None for recent_blockhash Made-with: Cursor --- src/streaming/event_parser/common/types.rs | 5 +++++ src/streaming/event_parser/core/event_parser.rs | 16 ++++++++++++++++ .../protocols/block/block_meta_event.rs | 1 + 3 files changed, 22 insertions(+) diff --git a/src/streaming/event_parser/common/types.rs b/src/streaming/event_parser/common/types.rs index f33d78f..088bcf4 100755 --- a/src/streaming/event_parser/common/types.rs +++ b/src/streaming/event_parser/common/types.rs @@ -312,6 +312,9 @@ pub struct EventMetadata { pub swap_data: Option, pub outer_index: i64, pub inner_index: Option, + /// Transaction message recent blockhash (32 bytes), when available. + #[serde(default)] + pub recent_blockhash: Option>, } impl EventMetadata { @@ -328,6 +331,7 @@ impl EventMetadata { inner_index: Option, recv_us: i64, tx_index: Option, + recent_blockhash: Option>, ) -> Self { Self { signature, @@ -343,6 +347,7 @@ impl EventMetadata { outer_index, inner_index, tx_index, + recent_blockhash, } } diff --git a/src/streaming/event_parser/core/event_parser.rs b/src/streaming/event_parser/core/event_parser.rs index 782427a..a4f576f 100644 --- a/src/streaming/event_parser/core/event_parser.rs +++ b/src/streaming/event_parser/core/event_parser.rs @@ -82,6 +82,11 @@ impl EventParser { .collect(); // 解析指令事件 let instructions = &message.instructions; + let recent_blockhash = if message.recent_blockhash.is_empty() { + None + } else { + Some(message.recent_blockhash.clone()) + }; Self::parse_instruction_events_from_grpc_transaction( protocols, event_type_filter, @@ -94,6 +99,7 @@ impl EventParser { &inner_instructions, bot_wallet, tx_index, + recent_blockhash, adapter_callback, ) .await?; @@ -128,6 +134,7 @@ impl EventParser { }); // 获取交易的指令和账户 let compiled_instructions = transaction.message.instructions(); + let recent_blockhash = Some(transaction.message.recent_blockhash().to_bytes().to_vec()); let mut accounts: Vec = accounts.to_vec(); // 检查交易中是否包含程序 let has_program = accounts @@ -160,6 +167,7 @@ impl EventParser { None, bot_wallet, tx_index, + recent_blockhash.as_deref(), inner_instructions, adapter_callback.clone(), )?; @@ -182,6 +190,7 @@ impl EventParser { Some(inner_index as i64), bot_wallet, tx_index, + recent_blockhash.as_deref(), Some(&inner_instructions), adapter_callback.clone(), )?; @@ -214,6 +223,7 @@ impl EventParser { inner_instructions: &[yellowstone_grpc_proto::prelude::InnerInstructions], bot_wallet: Option, tx_index: Option, + recent_blockhash: Option>, callback: Arc Fn(&'a DexEvent) + Send + Sync>, ) -> anyhow::Result<()> { // 获取交易的指令和账户 @@ -249,6 +259,7 @@ impl EventParser { None, bot_wallet, tx_index, + recent_blockhash.as_deref(), inner_instructions, callback.clone(), )?; @@ -279,6 +290,7 @@ impl EventParser { Some(inner_index as i64), bot_wallet, tx_index, + recent_blockhash.as_deref(), Some(&inner_instructions), callback.clone(), )?; @@ -308,6 +320,7 @@ impl EventParser { inner_index: Option, bot_wallet: Option, tx_index: Option, + recent_blockhash: Option<&[u8]>, inner_instructions: Option<&yellowstone_grpc_proto::prelude::InnerInstructions>, callback: Arc Fn(&'a DexEvent) + Send + Sync>, ) -> anyhow::Result<()> { @@ -347,6 +360,7 @@ impl EventParser { inner_index, recv_us, tx_index, + recent_blockhash.map(|s| s.to_vec()), ); if is_cu_program { @@ -465,6 +479,7 @@ impl EventParser { inner_index: Option, bot_wallet: Option, tx_index: Option, + recent_blockhash: Option<&[u8]>, inner_instructions: Option<&InnerInstructions>, callback: Arc Fn(&'a DexEvent) + Send + Sync>, ) -> anyhow::Result<()> { @@ -505,6 +520,7 @@ impl EventParser { inner_index, recv_us, tx_index, + recent_blockhash.map(|s| s.to_vec()), ); if is_cu_program { diff --git a/src/streaming/event_parser/protocols/block/block_meta_event.rs b/src/streaming/event_parser/protocols/block/block_meta_event.rs index 6d606c6..fb42319 100644 --- a/src/streaming/event_parser/protocols/block/block_meta_event.rs +++ b/src/streaming/event_parser/protocols/block/block_meta_event.rs @@ -31,6 +31,7 @@ impl BlockMetaEvent { None, recv_us, None, + None, // recent_blockhash not applicable for block meta ); Self { metadata, slot, block_hash } } From 2da2b06e1ff42ba018d00f471d9ac1678cf4fd87 Mon Sep 17 00:00:00 2001 From: Wood Date: Sat, 7 Mar 2026 20:24:46 +0800 Subject: [PATCH 2/2] refactor(events): recent_blockhash as base58 string (same as signature) - EventMetadata.recent_blockhash: Option> -> Option - Add bs58 dependency; gRPC path: bs58::encode message bytes - VersionedTransaction path: use Hash::to_string() (base58) - parse_events_from_grpc_instruction/parse_events_from_instruction: accept Option<&str> Made-with: Cursor --- Cargo.toml | 1 + src/streaming/event_parser/common/types.rs | 6 +++--- src/streaming/event_parser/core/event_parser.rs | 14 +++++++------- 3 files changed, 11 insertions(+), 10 deletions(-) diff --git a/Cargo.toml b/Cargo.toml index 93478f8..8c5ee44 100755 --- a/Cargo.toml +++ b/Cargo.toml @@ -25,6 +25,7 @@ serde-big-array = "0.5.1" futures = "0.3.32" bincode = "1.3" anyhow = "1.0.102" +bs58 = "0.5.1" yellowstone-grpc-client = { version = "10.2.0" } yellowstone-grpc-proto = { version = "10.1.1" } tokio = { version = "1.49.0", features = ["full", "rt-multi-thread"]} diff --git a/src/streaming/event_parser/common/types.rs b/src/streaming/event_parser/common/types.rs index 088bcf4..b5fedaa 100755 --- a/src/streaming/event_parser/common/types.rs +++ b/src/streaming/event_parser/common/types.rs @@ -312,9 +312,9 @@ pub struct EventMetadata { pub swap_data: Option, pub outer_index: i64, pub inner_index: Option, - /// Transaction message recent blockhash (32 bytes), when available. + /// Transaction message recent blockhash as base58 string (same encoding as signature), when available. #[serde(default)] - pub recent_blockhash: Option>, + pub recent_blockhash: Option, } impl EventMetadata { @@ -331,7 +331,7 @@ impl EventMetadata { inner_index: Option, recv_us: i64, tx_index: Option, - recent_blockhash: Option>, + recent_blockhash: Option, ) -> Self { Self { signature, diff --git a/src/streaming/event_parser/core/event_parser.rs b/src/streaming/event_parser/core/event_parser.rs index a4f576f..d7703a0 100644 --- a/src/streaming/event_parser/core/event_parser.rs +++ b/src/streaming/event_parser/core/event_parser.rs @@ -85,7 +85,7 @@ impl EventParser { let recent_blockhash = if message.recent_blockhash.is_empty() { None } else { - Some(message.recent_blockhash.clone()) + Some(bs58::encode(&message.recent_blockhash).into_string()) }; Self::parse_instruction_events_from_grpc_transaction( protocols, @@ -134,7 +134,7 @@ impl EventParser { }); // 获取交易的指令和账户 let compiled_instructions = transaction.message.instructions(); - let recent_blockhash = Some(transaction.message.recent_blockhash().to_bytes().to_vec()); + let recent_blockhash = Some(transaction.message.recent_blockhash().to_string()); let mut accounts: Vec = accounts.to_vec(); // 检查交易中是否包含程序 let has_program = accounts @@ -223,7 +223,7 @@ impl EventParser { inner_instructions: &[yellowstone_grpc_proto::prelude::InnerInstructions], bot_wallet: Option, tx_index: Option, - recent_blockhash: Option>, + recent_blockhash: Option, callback: Arc Fn(&'a DexEvent) + Send + Sync>, ) -> anyhow::Result<()> { // 获取交易的指令和账户 @@ -320,7 +320,7 @@ impl EventParser { inner_index: Option, bot_wallet: Option, tx_index: Option, - recent_blockhash: Option<&[u8]>, + recent_blockhash: Option<&str>, inner_instructions: Option<&yellowstone_grpc_proto::prelude::InnerInstructions>, callback: Arc Fn(&'a DexEvent) + Send + Sync>, ) -> anyhow::Result<()> { @@ -360,7 +360,7 @@ impl EventParser { inner_index, recv_us, tx_index, - recent_blockhash.map(|s| s.to_vec()), + recent_blockhash.map(|s| s.to_string()), ); if is_cu_program { @@ -479,7 +479,7 @@ impl EventParser { inner_index: Option, bot_wallet: Option, tx_index: Option, - recent_blockhash: Option<&[u8]>, + recent_blockhash: Option<&str>, inner_instructions: Option<&InnerInstructions>, callback: Arc Fn(&'a DexEvent) + Send + Sync>, ) -> anyhow::Result<()> { @@ -520,7 +520,7 @@ impl EventParser { inner_index, recv_us, tx_index, - recent_blockhash.map(|s| s.to_vec()), + recent_blockhash.map(|s| s.to_string()), ); if is_cu_program {