From 007c82a1138b4a3a5a0166ad726130b8023d7596 Mon Sep 17 00:00:00 2001 From: georgi-l95 Date: Tue, 7 Oct 2025 18:33:02 +0300 Subject: [PATCH 1/6] improve publisher Signed-off-by: georgi-l95 --- Cargo.lock | 1 + crates/rock-node-core/src/config.rs | 6 + crates/rock-node-core/src/metrics.rs | 12 + crates/rock-node-publish-plugin/Cargo.toml | 1 + crates/rock-node-publish-plugin/src/error.rs | 37 ++ crates/rock-node-publish-plugin/src/lib.rs | 11 +- .../rock-node-publish-plugin/src/service.rs | 2 + .../src/session_manager.rs | 581 +++++++++++++++++- .../src/validation.rs | 158 +++++ .../src/session.rs | 2 + tests/e2e/src/tests/ingestion_suite.rs | 518 +++++++++++++++- 11 files changed, 1325 insertions(+), 4 deletions(-) create mode 100644 crates/rock-node-publish-plugin/src/error.rs create mode 100644 crates/rock-node-publish-plugin/src/validation.rs diff --git a/Cargo.lock b/Cargo.lock index d59fae7..0cfbbc6 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2421,6 +2421,7 @@ dependencies = [ "rock-node-core", "rock-node-protobufs", "tempfile", + "thiserror 1.0.69", "tokio", "tokio-stream", "tonic", diff --git a/crates/rock-node-core/src/config.rs b/crates/rock-node-core/src/config.rs index 5648d35..8447208 100644 --- a/crates/rock-node-core/src/config.rs +++ b/crates/rock-node-core/src/config.rs @@ -105,6 +105,10 @@ pub struct PublishServiceConfig { pub stale_winner_timeout_seconds: u64, pub winner_cleanup_interval_seconds: u64, pub winner_cleanup_threshold_blocks: u64, + /// Maximum gRPC message size in bytes (default 32MB) + pub max_message_size_bytes: usize, + /// Maximum items per BlockItemSet + pub max_items_per_set: usize, } impl Default for PublishServiceConfig { @@ -116,6 +120,8 @@ impl Default for PublishServiceConfig { stale_winner_timeout_seconds: 10, winner_cleanup_interval_seconds: 150, winner_cleanup_threshold_blocks: 1000, + max_message_size_bytes: 32 * 1024 * 1024, // 32MB + max_items_per_set: 10000, } } } diff --git a/crates/rock-node-core/src/metrics.rs b/crates/rock-node-core/src/metrics.rs index bfee66d..e60267b 100644 --- a/crates/rock-node-core/src/metrics.rs +++ b/crates/rock-node-core/src/metrics.rs @@ -35,6 +35,7 @@ pub struct MetricsRegistry { pub publish_header_to_proof_duration_seconds: HistogramVec, pub publish_average_header_to_proof_time_seconds: GaugeVec, pub publish_responses_sent_total: CounterVec, + pub publish_session_duration_seconds: HistogramVec, // --- Persistence Plugin Metrics --- pub persistence_writes_total: CounterVec, @@ -206,6 +207,16 @@ impl MetricsRegistry { )?; registry.register(Box::new(publish_responses_sent_total.clone()))?; + let publish_session_duration_seconds = HistogramVec::new( + Opts::new( + "rocknode_publish_session_duration_seconds", + "Duration of publisher sessions from connection to disconnection.", + ) + .into(), + &[], + )?; + registry.register(Box::new(publish_session_duration_seconds.clone()))?; + let persistence_writes_total = CounterVec::new( Opts::new( "rocknode_persistence_writes_total", @@ -385,6 +396,7 @@ impl MetricsRegistry { publish_header_to_proof_duration_seconds, publish_average_header_to_proof_time_seconds, publish_responses_sent_total, + publish_session_duration_seconds, persistence_writes_total, persistence_write_duration_seconds, persistence_event_duration_seconds, diff --git a/crates/rock-node-publish-plugin/Cargo.toml b/crates/rock-node-publish-plugin/Cargo.toml index 6212b33..69811db 100644 --- a/crates/rock-node-publish-plugin/Cargo.toml +++ b/crates/rock-node-publish-plugin/Cargo.toml @@ -17,6 +17,7 @@ tonic = "0.13.1" # For easy error handling anyhow = "1" +thiserror = "1" # For logging tracing = "0.1" diff --git a/crates/rock-node-publish-plugin/src/error.rs b/crates/rock-node-publish-plugin/src/error.rs new file mode 100644 index 0000000..7b18fbd --- /dev/null +++ b/crates/rock-node-publish-plugin/src/error.rs @@ -0,0 +1,37 @@ +use thiserror::Error; + +/// Errors that can occur during publish operations +#[derive(Error, Debug)] +pub enum PublishError { + #[error("Block item set is empty")] + EmptyBlockItems, + + #[error("Missing block header for new block {block_number}")] + MissingBlockHeader { block_number: i64 }, + + #[error("Too many items in set: {count} (max: {max})")] + TooManyItems { count: usize, max: usize }, + + #[error("Block number mismatch: expected {expected}, got {received}")] + BlockNumberMismatch { expected: i64, received: i64 }, + + #[error("Invalid block number: {block_number}")] + InvalidBlockNumber { block_number: i64 }, + + #[error("Session {session_id} received items without active block")] + NoActiveBlock { session_id: uuid::Uuid }, + + #[error("Failed to encode block: {0}")] + Encoding(#[from] prost::EncodeError), + + #[error("Channel send error")] + ChannelSend, + + #[error("Verification timeout for block {block_number}")] + VerificationTimeout { block_number: u64 }, + + #[error("Verification failed for block {block_number}: {reason}")] + VerificationFailed { block_number: u64, reason: String }, +} + +pub type Result = std::result::Result; diff --git a/crates/rock-node-publish-plugin/src/lib.rs b/crates/rock-node-publish-plugin/src/lib.rs index 8f77e8b..5bfee0a 100644 --- a/crates/rock-node-publish-plugin/src/lib.rs +++ b/crates/rock-node-publish-plugin/src/lib.rs @@ -23,9 +23,11 @@ use std::{ use tonic::service::RoutesBuilder; use tracing::{debug, info, warn}; +mod error; mod service; mod session_manager; mod state; +mod validation; #[derive(Debug, Default)] pub struct PublishPlugin { @@ -134,9 +136,14 @@ impl Plugin for PublishPlugin { context: context.clone(), shared_state: self.shared_state.as_ref().unwrap().clone(), }; - const MAX_MESSAGE_SIZE: usize = 1024 * 1024 * 32; + let max_message_size = context + .config + .plugins + .publish_service + .max_message_size_bytes; let server = BlockStreamPublishServiceServer::new(service) - .max_decoding_message_size(MAX_MESSAGE_SIZE); + .max_decoding_message_size(max_message_size) + .max_encoding_message_size(max_message_size); builder.add_service(server); Ok(true) diff --git a/crates/rock-node-publish-plugin/src/service.rs b/crates/rock-node-publish-plugin/src/service.rs index 91c2603..197a021 100644 --- a/crates/rock-node-publish-plugin/src/service.rs +++ b/crates/rock-node-publish-plugin/src/service.rs @@ -53,6 +53,8 @@ impl BlockStreamPublishService for PublishServiceImpl { } } } + // Log session statistics before cleanup + session_manager.log_session_stats(); info!(%session_id, "Session handler task finished."); // When the session finishes, it may have been the recorded winner for diff --git a/crates/rock-node-publish-plugin/src/session_manager.rs b/crates/rock-node-publish-plugin/src/session_manager.rs index 355f135..d044931 100644 --- a/crates/rock-node-publish-plugin/src/session_manager.rs +++ b/crates/rock-node-publish-plugin/src/session_manager.rs @@ -1,4 +1,8 @@ -use crate::state::{BlockAction, SessionState, SharedState}; +use crate::{ + error::PublishError, + state::{BlockAction, SessionState, SharedState}, + validation::RequestValidator, +}; use anyhow::Result; use prost::Message; use rock_node_core::AppContext; @@ -20,16 +24,24 @@ use uuid::Uuid; pub struct SessionManager { pub id: Uuid, + /// Optional publisher identifier for tracking/metrics + pub publisher_id: Option, context: AppContext, shared_state: Arc, state: SessionState, current_block_number: i64, current_block_action: Option, block_start_time: Option, + session_start_time: Instant, item_buffer: Vec, response_tx: mpsc::Sender>, header_proof_total_duration: f64, header_proof_count: u64, + validator: RequestValidator, + // Metrics tracking + blocks_received: u64, + blocks_accepted: u64, + blocks_skipped: u64, } impl SessionManager { @@ -38,18 +50,25 @@ impl SessionManager { shared_state: Arc, response_tx: mpsc::Sender>, ) -> Self { + let max_items = context.config.plugins.publish_service.max_items_per_set; Self { id: Uuid::new_v4(), + publisher_id: None, context, shared_state, state: SessionState::New, current_block_number: 0, current_block_action: None, block_start_time: None, + session_start_time: Instant::now(), item_buffer: Vec::new(), response_tx, header_proof_total_duration: 0.0, header_proof_count: 0, + validator: RequestValidator::new(max_items), + blocks_received: 0, + blocks_accepted: 0, + blocks_skipped: 0, } } @@ -72,6 +91,8 @@ impl SessionManager { &mut self, block_item_set: rock_node_protobufs::org::hiero::block::api::BlockItemSet, ) -> bool { + self.blocks_received += 1; + let mut block_number: Option = None; // First pass - extract block number from header if present @@ -82,6 +103,27 @@ impl SessionManager { } } + // Determine if this is a new block or continuation + let is_new_block = block_number.is_some(); + + // Validate request format + if let Err(e) = self + .validator + .validate_block_items(&block_item_set, is_new_block) + { + error!( + session_id = %self.id, + error = %e, + "Validation failed for block items" + ); + self.context + .metrics + .publish_blocks_received_total + .with_label_values(&["validation_error"]) + .inc(); + return true; // Terminate session on validation error + } + // Get action for this request let action = if let Some(num) = block_number { // First request of block (has header) @@ -378,6 +420,7 @@ impl SessionManager { /// Handle Skip action - send SkipBlock response and reset async fn handle_skip(&mut self) { + self.blocks_skipped += 1; let response = PublishStreamResponse { response: Some(publish_stream_response::Response::SkipBlock(SkipBlock { block_number: self.current_block_number as u64, @@ -575,6 +618,7 @@ impl SessionManager { /// Complete block: send to verification, wait for result, then persistence, then ACK /// Returns false to continue the session, true to terminate async fn complete_block(&mut self) -> bool { + self.blocks_accepted += 1; info!(session_id = %self.id, block_number = self.current_block_number, "Block is complete. Sending to verification."); let block_proto = Block { @@ -817,6 +861,32 @@ impl SessionManager { Ok(()) } } + + /// Log session statistics (call when session ends) + pub fn log_session_stats(&self) { + let session_duration = self.session_start_time.elapsed(); + info!( + session_id = %self.id, + publisher_id = ?self.publisher_id, + duration_secs = session_duration.as_secs(), + blocks_received = self.blocks_received, + blocks_accepted = self.blocks_accepted, + blocks_skipped = self.blocks_skipped, + avg_header_to_proof_secs = if self.header_proof_count > 0 { + self.header_proof_total_duration / self.header_proof_count as f64 + } else { + 0.0 + }, + "Publisher session completed" + ); + + // Record session duration metric + self.context + .metrics + .publish_session_duration_seconds + .with_label_values::<&str>(&[]) + .observe(session_duration.as_secs_f64()); + } } #[cfg(test)] @@ -884,6 +954,8 @@ mod tests { stale_winner_timeout_seconds: 1, winner_cleanup_interval_seconds: 60, winner_cleanup_threshold_blocks: 100, + max_message_size_bytes: 32 * 1024 * 1024, + max_items_per_set: 10000, }, verification_service: VerificationServiceConfig { enabled: true }, block_access_service: BlockAccessServiceConfig { enabled: true }, @@ -1387,4 +1459,511 @@ mod tests { _ => panic!("Expected ResendBlock to other session"), } } + + // ============================================ + // NEW COMPREHENSIVE MULTI-PUBLISHER TESTS + // ============================================ + + #[tokio::test] + async fn three_publishers_race_for_same_block() { + // Test that when 3 publishers race for the same block, + // only one wins and the other two get SkipBlock + let metrics = create_test_metrics(); + let (context, _rx_items, _rx_verified, _rx_persisted) = make_context_with_registry(metrics); + let shared = Arc::new(SharedState::new()); + shared.set_latest_persisted_block(200); + + let (tx1, mut rx1) = mpsc::channel(4); + let (tx2, mut rx2) = mpsc::channel(4); + let (tx3, mut rx3) = mpsc::channel(4); + + let mut s1 = SessionManager::new(context.clone(), shared.clone(), tx1); + let mut s2 = SessionManager::new(context.clone(), shared.clone(), tx2); + let mut s3 = SessionManager::new(context.clone(), shared.clone(), tx3); + + // Register all three sessions + shared.active_sessions.insert(s1.id, s1.response_tx.clone()); + shared.active_sessions.insert(s2.id, s2.response_tx.clone()); + shared.active_sessions.insert(s3.id, s3.response_tx.clone()); + + // All three send header for block 201 + let header = rock_node_protobufs::com::hedera::hapi::block::stream::BlockItem { + item: Some(BlockItemType::BlockHeader( + rock_node_protobufs::com::hedera::hapi::block::stream::output::BlockHeader { + number: 201, + ..Default::default() + }, + )), + }; + + let req1 = PublishRequestType::BlockItems( + rock_node_protobufs::org::hiero::block::api::BlockItemSet { + block_items: vec![header.clone()], + }, + ); + let req2 = PublishRequestType::BlockItems( + rock_node_protobufs::org::hiero::block::api::BlockItemSet { + block_items: vec![header.clone()], + }, + ); + let req3 = PublishRequestType::BlockItems( + rock_node_protobufs::org::hiero::block::api::BlockItemSet { + block_items: vec![header], + }, + ); + + // s1 wins + assert!(!s1.handle_request(req1).await); + assert_eq!(s1.state, SessionState::Primary); + + // s2 and s3 should get SkipBlock + assert!(!s2.handle_request(req2).await); + assert!(!s3.handle_request(req3).await); + + // Verify both losers got SkipBlock + for (name, rx) in [("s2", &mut rx2), ("s3", &mut rx3)] { + let msg = tokio::time::timeout(Duration::from_millis(200), rx.recv()) + .await + .expect(&format!("{} should receive SkipBlock", name)) + .unwrap() + .unwrap(); + match msg.response.unwrap() { + publish_stream_response::Response::SkipBlock(sk) => { + assert_eq!(sk.block_number, 201); + }, + other => panic!("{} expected SkipBlock, got {:?}", name, other), + } + } + + // Winner should not receive anything yet + assert!(rx1.try_recv().is_err()); + + // Verify block_winners map has correct entry + assert_eq!(shared.block_winners.get(&201).unwrap().value(), &s1.id); + } + + #[tokio::test] + async fn validation_error_empty_block_items() { + // Test that validation rejects empty block items + let metrics = create_test_metrics(); + let (context, _rx_items, _rx_verified, _rx_persisted) = make_context_with_registry(metrics); + let shared = Arc::new(SharedState::new()); + shared.set_latest_persisted_block(100); + + let (tx, _rx) = mpsc::channel(4); + let mut session = SessionManager::new(context, shared, tx); + + // Send empty block items + let req = PublishRequestType::BlockItems( + rock_node_protobufs::org::hiero::block::api::BlockItemSet { + block_items: vec![], + }, + ); + + // Should terminate due to validation error + assert!(session.handle_request(req).await); + } + + #[tokio::test] + async fn validation_error_missing_header_for_new_block() { + // Test that validation rejects new blocks without header + let metrics = create_test_metrics(); + let (context, _rx_items, _rx_verified, _rx_persisted) = make_context_with_registry(metrics); + let shared = Arc::new(SharedState::new()); + shared.set_latest_persisted_block(100); + + let (tx, _rx) = mpsc::channel(4); + let mut session = SessionManager::new(context, shared, tx); + + // Send proof without header (new block should start with header) + let proof_item = rock_node_protobufs::com::hedera::hapi::block::stream::BlockItem { + item: Some(BlockItemType::BlockProof( + rock_node_protobufs::com::hedera::hapi::block::stream::BlockProof { + block: 101, + ..Default::default() + }, + )), + }; + let req = PublishRequestType::BlockItems( + rock_node_protobufs::org::hiero::block::api::BlockItemSet { + block_items: vec![proof_item], + }, + ); + + // Should terminate due to validation error + assert!(session.handle_request(req).await); + } + + #[tokio::test] + async fn validation_error_too_many_items() { + // Test that validation rejects excessive items + let metrics = create_test_metrics(); + let (context, _rx_items, _rx_verified, _rx_persisted) = make_context_with_registry(metrics); + + // Create a new context with modified config + let mut config = (*context.config).clone(); + config.plugins.publish_service.max_items_per_set = 5; + let modified_context = rock_node_core::AppContext { + config: Arc::new(config), + metrics: context.metrics.clone(), + capability_registry: context.capability_registry.clone(), + service_providers: context.service_providers.clone(), + block_data_cache: context.block_data_cache.clone(), + tx_block_items_received: context.tx_block_items_received.clone(), + tx_block_verified: context.tx_block_verified.clone(), + tx_block_verification_failed: context.tx_block_verification_failed.clone(), + tx_block_persisted: context.tx_block_persisted.clone(), + }; + + let shared = Arc::new(SharedState::new()); + shared.set_latest_persisted_block(100); + + let (tx, _rx) = mpsc::channel(4); + let mut session = SessionManager::new(modified_context, shared, tx); + + // Create header + 10 items (exceeds limit of 5) + let header = rock_node_protobufs::com::hedera::hapi::block::stream::BlockItem { + item: Some(BlockItemType::BlockHeader( + rock_node_protobufs::com::hedera::hapi::block::stream::output::BlockHeader { + number: 101, + ..Default::default() + }, + )), + }; + let mut items = vec![header]; + for _ in 0..9 { + items.push( + rock_node_protobufs::com::hedera::hapi::block::stream::BlockItem { + item: Some(BlockItemType::BlockProof( + rock_node_protobufs::com::hedera::hapi::block::stream::BlockProof::default( + ), + )), + }, + ); + } + + let req = PublishRequestType::BlockItems( + rock_node_protobufs::org::hiero::block::api::BlockItemSet { block_items: items }, + ); + + // Should terminate due to validation error + assert!(session.handle_request(req).await); + } + + #[tokio::test] + async fn multiple_consecutive_blocks_from_different_publishers() { + // Test that different publishers can win consecutive blocks + let metrics = create_test_metrics(); + let (context, mut rx_items, _rx_verified, _rx_persisted) = + make_context_with_registry(metrics); + let shared = Arc::new(SharedState::new()); + shared.set_latest_persisted_block(100); + + let (tx1, mut rx1) = mpsc::channel(8); + let (tx2, _rx2) = mpsc::channel(8); + + let mut s1 = SessionManager::new(context.clone(), shared.clone(), tx1); + let mut s2 = SessionManager::new(context.clone(), shared.clone(), tx2); + + shared.active_sessions.insert(s1.id, s1.response_tx.clone()); + shared.active_sessions.insert(s2.id, s2.response_tx.clone()); + + // Publisher 1 wins block 101 + let header_101 = rock_node_protobufs::com::hedera::hapi::block::stream::BlockItem { + item: Some(BlockItemType::BlockHeader( + rock_node_protobufs::com::hedera::hapi::block::stream::output::BlockHeader { + number: 101, + ..Default::default() + }, + )), + }; + let proof_101 = rock_node_protobufs::com::hedera::hapi::block::stream::BlockItem { + item: Some(BlockItemType::BlockProof( + rock_node_protobufs::com::hedera::hapi::block::stream::BlockProof { + block: 101, + ..Default::default() + }, + )), + }; + + let req_101_s1 = PublishRequestType::BlockItems( + rock_node_protobufs::org::hiero::block::api::BlockItemSet { + block_items: vec![header_101, proof_101], + }, + ); + + // Simulate persistence for block 101 + let tx_persisted = s1.context.tx_block_persisted.clone(); + tokio::spawn(async move { + let items_received = rx_items.recv().await.unwrap(); + tokio::time::sleep(Duration::from_millis(50)).await; + let _ = tx_persisted.send(rock_node_core::events::BlockPersisted { + block_number: items_received.block_number, + cache_key: items_received.cache_key, + }); + }); + + assert!(!s1.handle_request(req_101_s1).await); + + // Wait for ACK + let ack = tokio::time::timeout(Duration::from_secs(1), rx1.recv()) + .await + .expect("timeout waiting for ACK") + .unwrap() + .unwrap(); + match ack.response.unwrap() { + publish_stream_response::Response::Acknowledgement(a) => { + assert_eq!(a.block_number, 101); + }, + _ => panic!("Expected Acknowledgement"), + } + + // Now publisher 2 wins block 102 + shared.set_latest_persisted_block(101); + + let header_102 = rock_node_protobufs::com::hedera::hapi::block::stream::BlockItem { + item: Some(BlockItemType::BlockHeader( + rock_node_protobufs::com::hedera::hapi::block::stream::output::BlockHeader { + number: 102, + ..Default::default() + }, + )), + }; + + let req_102_s2 = PublishRequestType::BlockItems( + rock_node_protobufs::org::hiero::block::api::BlockItemSet { + block_items: vec![header_102], + }, + ); + + assert!(!s2.handle_request(req_102_s2).await); + assert_eq!(s2.state, SessionState::Primary); + + // Publisher 1 tries block 102 and should be skipped + let header_102_s1 = rock_node_protobufs::com::hedera::hapi::block::stream::BlockItem { + item: Some(BlockItemType::BlockHeader( + rock_node_protobufs::com::hedera::hapi::block::stream::output::BlockHeader { + number: 102, + ..Default::default() + }, + )), + }; + + let req_102_s1 = PublishRequestType::BlockItems( + rock_node_protobufs::org::hiero::block::api::BlockItemSet { + block_items: vec![header_102_s1], + }, + ); + + assert!(!s1.handle_request(req_102_s1).await); + + // s1 should get SkipBlock for 102 + let skip = tokio::time::timeout(Duration::from_millis(200), rx1.recv()) + .await + .expect("timeout waiting for SkipBlock") + .unwrap() + .unwrap(); + match skip.response.unwrap() { + publish_stream_response::Response::SkipBlock(sk) => { + assert_eq!(sk.block_number, 102); + }, + _ => panic!("Expected SkipBlock"), + } + } + + #[tokio::test] + async fn winner_cleanup_on_session_disconnect() { + // Test that when winner disconnects, block_winners entry is cleaned up + // and other sessions get ResendBlock + let metrics = create_test_metrics(); + let (context, _rx_items, _rx_verified, _rx_persisted) = make_context_with_registry(metrics); + let shared = Arc::new(SharedState::new()); + shared.set_latest_persisted_block(100); + + let (tx1, _rx1) = mpsc::channel(4); + let (tx2, mut rx2) = mpsc::channel(4); + + let s1 = SessionManager::new(context.clone(), shared.clone(), tx1); + let s2 = SessionManager::new(context.clone(), shared.clone(), tx2.clone()); + + let s1_id = s1.id; + + shared.active_sessions.insert(s1.id, s1.response_tx.clone()); + shared.active_sessions.insert(s2.id, tx2.clone()); + + // Manually set s1 as winner for block 101 + shared.block_winners.insert(101, s1_id); + + // Simulate s1 disconnecting (drop it) + drop(s1); + + // Remove s1 from active sessions and clean up affected blocks + shared.active_sessions.remove(&s1_id); + + // Find affected blocks and broadcast ResendBlock + let mut affected_blocks = Vec::new(); + for entry in shared.block_winners.iter() { + if entry.value() == &s1_id { + affected_blocks.push(*entry.key()); + } + } + + // Cleanup and broadcast + for block_num in &affected_blocks { + shared.block_winners.remove(block_num); + + // Broadcast ResendBlock to remaining sessions + for session_entry in shared.active_sessions.iter() { + let resend_msg = rock_node_protobufs::org::hiero::block::api::PublishStreamResponse { + response: Some(publish_stream_response::Response::ResendBlock( + rock_node_protobufs::org::hiero::block::api::publish_stream_response::ResendBlock { + block_number: *block_num, + }, + )), + }; + let _ = session_entry.value().try_send(Ok(resend_msg)); + } + } + + // s2 should receive ResendBlock for 101 + let resend = tokio::time::timeout(Duration::from_millis(200), rx2.recv()) + .await + .expect("timeout waiting for ResendBlock") + .unwrap() + .unwrap(); + + match resend.response.unwrap() { + publish_stream_response::Response::ResendBlock(rb) => { + assert_eq!(rb.block_number, 101); + }, + _ => panic!("Expected ResendBlock"), + } + + // Verify block_winners is cleaned up + assert!(!shared.block_winners.contains_key(&101)); + } + + #[tokio::test] + async fn session_stats_recorded_correctly() { + // Test that session statistics are correctly tracked + let metrics = create_test_metrics(); + let (context, _rx_items, _rx_verified, _rx_persisted) = make_context_with_registry(metrics); + let shared = Arc::new(SharedState::new()); + shared.set_latest_persisted_block(100); + + let (tx, mut rx) = mpsc::channel(8); + let mut session = SessionManager::new(context, shared, tx); + + // Set publisher ID + session.publisher_id = Some("test-publisher-1".to_string()); + + // Send 3 blocks: 2 accepted, 1 skipped + let header_101 = rock_node_protobufs::com::hedera::hapi::block::stream::BlockItem { + item: Some(BlockItemType::BlockHeader( + rock_node_protobufs::com::hedera::hapi::block::stream::output::BlockHeader { + number: 101, + ..Default::default() + }, + )), + }; + + // First block - accepted (wins) + let req1 = PublishRequestType::BlockItems( + rock_node_protobufs::org::hiero::block::api::BlockItemSet { + block_items: vec![header_101.clone()], + }, + ); + assert!(!session.handle_request(req1).await); + assert_eq!(session.state, SessionState::Primary); + + // Manually trigger skip + session.handle_skip().await; + + // Clear the channel + let _ = rx.try_recv(); + + // Another block - accepted + session.reset_for_next_block(); + session.state = SessionState::New; + let header_102 = rock_node_protobufs::com::hedera::hapi::block::stream::BlockItem { + item: Some(BlockItemType::BlockHeader( + rock_node_protobufs::com::hedera::hapi::block::stream::output::BlockHeader { + number: 102, + ..Default::default() + }, + )), + }; + let req2 = PublishRequestType::BlockItems( + rock_node_protobufs::org::hiero::block::api::BlockItemSet { + block_items: vec![header_102], + }, + ); + assert!(!session.handle_request(req2).await); + + // Verify stats + assert_eq!(session.blocks_received, 2); + assert_eq!(session.blocks_skipped, 1); + assert!(session.session_start_time.elapsed().as_secs() < 5); + + // Log stats would normally be called on disconnect + session.log_session_stats(); + } + + #[tokio::test] + async fn concurrent_different_blocks_from_multiple_publishers() { + // Test that different publishers can work on different blocks concurrently + let metrics = create_test_metrics(); + let (context, _rx_items, _rx_verified, _rx_persisted) = make_context_with_registry(metrics); + let shared = Arc::new(SharedState::new()); + shared.set_latest_persisted_block(100); + + let (tx1, _rx1) = mpsc::channel(4); + let (tx2, _rx2) = mpsc::channel(4); + + let mut s1 = SessionManager::new(context.clone(), shared.clone(), tx1); + let mut s2 = SessionManager::new(context.clone(), shared.clone(), tx2); + + shared.active_sessions.insert(s1.id, s1.response_tx.clone()); + shared.active_sessions.insert(s2.id, s2.response_tx.clone()); + + // s1 starts block 101 + let header_101 = rock_node_protobufs::com::hedera::hapi::block::stream::BlockItem { + item: Some(BlockItemType::BlockHeader( + rock_node_protobufs::com::hedera::hapi::block::stream::output::BlockHeader { + number: 101, + ..Default::default() + }, + )), + }; + let req_101 = PublishRequestType::BlockItems( + rock_node_protobufs::org::hiero::block::api::BlockItemSet { + block_items: vec![header_101], + }, + ); + assert!(!s1.handle_request(req_101).await); + assert_eq!(s1.state, SessionState::Primary); + assert_eq!(s1.current_block_number, 101); + + // s2 starts block 102 (ahead of persisted, but allowed) + let header_102 = rock_node_protobufs::com::hedera::hapi::block::stream::BlockItem { + item: Some(BlockItemType::BlockHeader( + rock_node_protobufs::com::hedera::hapi::block::stream::output::BlockHeader { + number: 102, + ..Default::default() + }, + )), + }; + let req_102 = PublishRequestType::BlockItems( + rock_node_protobufs::org::hiero::block::api::BlockItemSet { + block_items: vec![header_102], + }, + ); + assert!(!s2.handle_request(req_102).await); + assert_eq!(s2.state, SessionState::Primary); + assert_eq!(s2.current_block_number, 102); + + // Both should be registered as winners for their respective blocks + assert_eq!(shared.block_winners.get(&101).unwrap().value(), &s1.id); + assert_eq!(shared.block_winners.get(&102).unwrap().value(), &s2.id); + } } diff --git a/crates/rock-node-publish-plugin/src/validation.rs b/crates/rock-node-publish-plugin/src/validation.rs new file mode 100644 index 0000000..4a1e0de --- /dev/null +++ b/crates/rock-node-publish-plugin/src/validation.rs @@ -0,0 +1,158 @@ +use crate::error::{PublishError, Result}; +use rock_node_protobufs::{ + com::hedera::hapi::block::stream::block_item::Item as BlockItemType, + org::hiero::block::api::BlockItemSet, +}; + +/// Validates a BlockItemSet for basic format correctness +pub struct RequestValidator { + max_items_per_set: usize, +} + +impl RequestValidator { + pub fn new(max_items_per_set: usize) -> Self { + Self { max_items_per_set } + } + + /// Validate block item set format + pub fn validate_block_items( + &self, + block_item_set: &BlockItemSet, + is_new_block: bool, + ) -> Result<()> { + // Check not empty + if block_item_set.block_items.is_empty() { + return Err(PublishError::EmptyBlockItems); + } + + // Check item count + let count = block_item_set.block_items.len(); + if count > self.max_items_per_set { + return Err(PublishError::TooManyItems { + count, + max: self.max_items_per_set, + }); + } + + // If this is a new block, first item must be a BlockHeader + if is_new_block { + let first_item = &block_item_set.block_items[0]; + if let Some(item_type) = &first_item.item { + if !matches!(item_type, BlockItemType::BlockHeader(_)) { + // Extract block number if we can + let block_number = self.extract_block_number_from_any_item(item_type); + return Err(PublishError::MissingBlockHeader { + block_number: block_number.unwrap_or(-1), + }); + } + } else { + return Err(PublishError::MissingBlockHeader { block_number: -1 }); + } + } + + Ok(()) + } + + /// Extract block number from any item type (for error reporting) + fn extract_block_number_from_any_item(&self, item: &BlockItemType) -> Option { + match item { + BlockItemType::BlockHeader(header) => Some(header.number as i64), + BlockItemType::BlockProof(proof) => Some(proof.block as i64), + _ => None, + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use rock_node_protobufs::com::hedera::hapi::block::stream::{ + output::BlockHeader, BlockItem, BlockProof, + }; + + fn make_header_item(block_num: u64) -> BlockItem { + BlockItem { + item: Some(BlockItemType::BlockHeader(BlockHeader { + hapi_proto_version: None, + software_version: None, + number: block_num, + block_timestamp: None, + hash_algorithm: 0, + })), + } + } + + fn make_proof_item(block_num: u64) -> BlockItem { + BlockItem { + item: Some(BlockItemType::BlockProof(BlockProof { + block: block_num, + ..Default::default() + })), + } + } + + #[test] + fn test_empty_block_items() { + let validator = RequestValidator::new(1000); + let empty_set = BlockItemSet { + block_items: vec![], + }; + + let result = validator.validate_block_items(&empty_set, true); + assert!(result.is_err()); + assert!(matches!(result.unwrap_err(), PublishError::EmptyBlockItems)); + } + + #[test] + fn test_too_many_items() { + let validator = RequestValidator::new(5); + let items = vec![make_header_item(1); 10]; // 10 items, max is 5 + let set = BlockItemSet { block_items: items }; + + let result = validator.validate_block_items(&set, true); + assert!(result.is_err()); + match result.unwrap_err() { + PublishError::TooManyItems { count, max } => { + assert_eq!(count, 10); + assert_eq!(max, 5); + }, + _ => panic!("Expected TooManyItems error"), + } + } + + #[test] + fn test_new_block_missing_header() { + let validator = RequestValidator::new(1000); + // Start with proof instead of header + let items = vec![make_proof_item(1)]; + let set = BlockItemSet { block_items: items }; + + let result = validator.validate_block_items(&set, true); + assert!(result.is_err()); + assert!(matches!( + result.unwrap_err(), + PublishError::MissingBlockHeader { .. } + )); + } + + #[test] + fn test_valid_new_block() { + let validator = RequestValidator::new(1000); + let items = vec![make_header_item(1), make_proof_item(1)]; + let set = BlockItemSet { block_items: items }; + + let result = validator.validate_block_items(&set, true); + assert!(result.is_ok()); + } + + #[test] + fn test_valid_continuation() { + let validator = RequestValidator::new(1000); + // Continuation doesn't need header + let items = vec![make_proof_item(1)]; + let set = BlockItemSet { block_items: items }; + + let result = validator.validate_block_items(&set, false); + assert!(result.is_ok()); + } +} diff --git a/crates/rock-node-subscriber-plugin/src/session.rs b/crates/rock-node-subscriber-plugin/src/session.rs index cc964eb..a4af142 100644 --- a/crates/rock-node-subscriber-plugin/src/session.rs +++ b/crates/rock-node-subscriber-plugin/src/session.rs @@ -380,6 +380,8 @@ mod tests { stale_winner_timeout_seconds: 0, winner_cleanup_interval_seconds: 0, winner_cleanup_threshold_blocks: 0, + max_message_size_bytes: todo!(), + max_items_per_set: todo!(), }, verification_service: VerificationServiceConfig { enabled: false }, block_access_service: BlockAccessServiceConfig { enabled: false }, diff --git a/tests/e2e/src/tests/ingestion_suite.rs b/tests/e2e/src/tests/ingestion_suite.rs index 0a45e63..b1856ef 100644 --- a/tests/e2e/src/tests/ingestion_suite.rs +++ b/tests/e2e/src/tests/ingestion_suite.rs @@ -2,12 +2,13 @@ use crate::common::{block_builder::BlockBuilder, TestContext}; use anyhow::Result; use prost::Message; use rock_node_protobufs::org::hiero::block::api::{ + block_request::BlockSpecifier, publish_stream_request::Request as PublishRequest, publish_stream_response::{ end_of_stream::Code as EndCode, BlockAcknowledgement, Response as PublishResponse, SkipBlock, }, - BlockItemSet, PublishStreamRequest, + BlockItemSet, BlockRequest, PublishStreamRequest, }; use serial_test::serial; use std::sync::Arc; @@ -304,3 +305,518 @@ async fn test_multi_publisher_broadcast_ack() -> Result<()> { Ok(()) } + +// ============================================ +// CATEGORY 1: BASIC MULTI-PUBLISHER SCENARIOS +// ============================================ + +/// Test Case: Three publishers race for the same block +/// Verifies that with 3+ publishers racing: +/// - Exactly one wins (Primary) +/// - Other two receive SkipBlock +/// - All eventually receive ACK broadcast +#[tokio::test] +#[serial] +async fn test_three_publishers_race_for_same_block() -> Result<()> { + let ctx = TestContext::new().await?; + + // Create 3 publisher clients + let mut client_a = ctx.publisher_client().await?; + let mut client_b = ctx.publisher_client().await?; + let mut client_c = ctx.publisher_client().await?; + + // Create the same block header for all three + let block_bytes = BlockBuilder::new(0).build(); + let block_proto: rock_node_protobufs::com::hedera::hapi::block::stream::Block = + Message::decode(block_bytes.as_slice())?; + let header_only = vec![block_proto.items[0].clone()]; + + // Create channels for all three publishers + let (tx_a, rx_a) = mpsc::channel(8); + let (tx_b, rx_b) = mpsc::channel(8); + let (tx_c, rx_c) = mpsc::channel(8); + + // Open streams for all three FIRST + let mut responses_a = client_a + .publish_block_stream(ReceiverStream::new(rx_a)) + .await? + .into_inner(); + + let mut responses_b = client_b + .publish_block_stream(ReceiverStream::new(rx_b)) + .await? + .into_inner(); + + let mut responses_c = client_c + .publish_block_stream(ReceiverStream::new(rx_c)) + .await? + .into_inner(); + + // All three send the header simultaneously + let header_request = PublishStreamRequest { + request: Some(PublishRequest::BlockItems(BlockItemSet { + block_items: header_only, + })), + }; + + tx_a.send(header_request.clone()).await?; + tx_b.send(header_request.clone()).await?; + tx_c.send(header_request.clone()).await?; + + // Wait for initial responses from all three + // Note: The winner won't get an immediate response (timeout is expected) + // Losers will get SkipBlock immediately + + let response_a = + match tokio::time::timeout(Duration::from_millis(500), responses_a.next()).await { + Ok(Some(Ok(resp))) => Some(resp), + Ok(Some(Err(e))) => return Err(e.into()), + Ok(None) => return Err(anyhow::anyhow!("Client A stream closed")), + Err(_) => None, // Timeout = winner (no immediate response) + }; + + let response_b = + match tokio::time::timeout(Duration::from_millis(500), responses_b.next()).await { + Ok(Some(Ok(resp))) => Some(resp), + Ok(Some(Err(e))) => return Err(e.into()), + Ok(None) => return Err(anyhow::anyhow!("Client B stream closed")), + Err(_) => None, // Timeout = winner (no immediate response) + }; + + let response_c = + match tokio::time::timeout(Duration::from_millis(500), responses_c.next()).await { + Ok(Some(Ok(resp))) => Some(resp), + Ok(Some(Err(e))) => return Err(e.into()), + Ok(None) => return Err(anyhow::anyhow!("Client C stream closed")), + Err(_) => None, // Timeout = winner (no immediate response) + }; + + // Count how many got SkipBlock + let mut skip_count = 0; + let mut no_response_count = 0; + + for (name, resp) in [("A", response_a), ("B", response_b), ("C", response_c)] { + match resp { + Some(resp) => match resp.response { + Some(PublishResponse::SkipBlock(skip)) => { + assert_eq!( + skip.block_number, 0, + "{} got SkipBlock for wrong block", + name + ); + skip_count += 1; + }, + Some(PublishResponse::Acknowledgement(_)) => { + // Might get broadcast ACK from another completed block + // This shouldn't happen in this test, but handle it gracefully + }, + other => { + panic!("{} got unexpected response: {:?}", name, other); + }, + }, + None => { + // This is the winner - no immediate response + no_response_count += 1; + }, + } + } + + // Exactly two should have gotten SkipBlock, and one should have no response (winner) + assert_eq!( + skip_count, 2, + "Expected exactly 2 publishers to get SkipBlock, got {}", + skip_count + ); + assert_eq!( + no_response_count, 1, + "Expected exactly 1 winner (no response), got {}", + no_response_count + ); + + println!("✅ Three publisher race: 1 winner, 2 skip blocks confirmed"); + + Ok(()) +} + +/// Test Case: Five publishers competing with high contention +/// Tests fairness and correctness with 5 publishers racing for 10 blocks +#[tokio::test] +#[serial] +async fn test_five_publishers_high_contention() -> Result<()> { + let ctx = Arc::new(TestContext::new().await?); + + // Track which publisher wins which block + let winners = Arc::new(tokio::sync::Mutex::new(std::collections::HashMap::new())); + + // Create 5 publisher tasks + let mut handles = vec![]; + + for publisher_id in 0..5 { + let ctx = Arc::clone(&ctx); + let winners = Arc::clone(&winners); + + let handle = tokio::spawn(async move { + let mut client = ctx.publisher_client().await.unwrap(); + let (tx, rx) = mpsc::channel(32); + let mut responses = client + .publish_block_stream(ReceiverStream::new(rx)) + .await + .unwrap() + .into_inner(); + + let mut my_wins = 0; + + // Try to publish blocks 0-9 + for block_num in 0u64..10 { + let block_bytes = BlockBuilder::new(block_num).build(); + let block_proto: rock_node_protobufs::com::hedera::hapi::block::stream::Block = + Message::decode(block_bytes.as_slice()).unwrap(); + + // Send just the header to race + let header = vec![block_proto.items[0].clone()]; + tx.send(PublishStreamRequest { + request: Some(PublishRequest::BlockItems(BlockItemSet { + block_items: header, + })), + }) + .await + .unwrap(); + + // Wait for response - might receive broadcast ACKs or SkipBlock + let mut got_skip = false; + let deadline = tokio::time::Instant::now() + Duration::from_millis(500); + + while tokio::time::Instant::now() < deadline { + if let Some(Ok(resp)) = tokio::time::timeout( + deadline - tokio::time::Instant::now(), + responses.next(), + ) + .await + .ok() + .flatten() + { + match resp.response { + Some(PublishResponse::SkipBlock(skip)) => { + // We lost this block + if skip.block_number == block_num { + got_skip = true; + break; + } + // Else: SkipBlock for a future block, shouldn't happen but continue + }, + Some(PublishResponse::Acknowledgement(ack)) => { + // Broadcast ACK from another publisher - ignore and keep waiting + if ack.block_number == block_num { + // This shouldn't happen (we'd get SkipBlock first), but handle it + got_skip = true; + break; + } + // Continue waiting for SkipBlock + }, + _ => {}, + } + } else { + // Timeout - no response + break; + } + } + + if !got_skip { + // No SkipBlock received, we might have won - complete the block + let proof = block_proto.items.last().cloned().unwrap(); + tx.send(PublishStreamRequest { + request: Some(PublishRequest::BlockItems(BlockItemSet { + block_items: vec![proof], + })), + }) + .await + .unwrap(); + + // Wait for ACK, ignoring any broadcast ACKs + let ack_deadline = tokio::time::Instant::now() + Duration::from_secs(3); + while tokio::time::Instant::now() < ack_deadline { + if let Some(Ok(ack_resp)) = tokio::time::timeout( + ack_deadline - tokio::time::Instant::now(), + responses.next(), + ) + .await + .ok() + .flatten() + { + if let Some(PublishResponse::Acknowledgement(ack)) = ack_resp.response { + if ack.block_number == block_num { + my_wins += 1; + winners.lock().await.insert(block_num, publisher_id); + break; + } + // Else: ACK for different block (broadcast), continue waiting + } + } else { + break; + } + } + } + + // Small delay between blocks + tokio::time::sleep(Duration::from_millis(10)).await; + } + + my_wins + }); + + handles.push(handle); + } + + // Wait for all publishers to finish + let mut results = vec![]; + for handle in handles { + results.push(handle.await); + } + + // Collect win counts + let mut total_wins = 0; + for (i, result) in results.iter().enumerate() { + match result { + Ok(wins) => { + println!("Publisher {} won {} blocks", i, wins); + total_wins += wins; + }, + Err(e) => { + println!("Publisher {} task failed: {}", i, e); + }, + } + } + + // Verify we have winners for most blocks (allowing for some contention issues) + let winners_map = winners.lock().await; + println!("Total blocks with confirmed winners: {}", winners_map.len()); + + // At least 7 out of 10 blocks should have been completed successfully + assert!( + winners_map.len() >= 7, + "Expected at least 7 blocks to be completed, got {}", + winners_map.len() + ); + + // Check fairness - no single publisher should win everything + let mut win_counts = std::collections::HashMap::new(); + for winner_id in winners_map.values() { + *win_counts.entry(*winner_id).or_insert(0) += 1; + } + + for (publisher_id, count) in &win_counts { + println!("Publisher {} won {} blocks", publisher_id, count); + } + + // No single publisher should win more than 60% of blocks (fairness check) + for count in win_counts.values() { + assert!( + *count <= 6, + "Single publisher won too many blocks ({}), fairness violated", + count + ); + } + + println!("✅ Five publisher high contention: fairness and correctness verified"); + + Ok(()) +} + +/// Test Case: Different publishers win consecutive blocks +/// Tests that: +/// - Publisher A wins block 0 +/// - Publisher B wins block 1 +/// - Publisher A wins block 2 +/// - Verifies that winner selection isn't sticky to one publisher +#[tokio::test] +#[serial] +async fn test_different_publishers_win_consecutive_blocks() -> Result<()> { + let ctx = Arc::new(TestContext::new().await?); + + // Synchronization channels: + // - block0_done: A signals when block 0 is complete + // - block1_done: B signals when block 1 is complete + let (block0_done_tx, mut block0_done_rx) = mpsc::channel::<()>(1); + let (block1_done_tx, mut block1_done_rx) = mpsc::channel::<()>(1); + + // Publisher A: publishes blocks 0 and 2 + let publisher_a_task = { + let ctx = Arc::clone(&ctx); + tokio::spawn(async move { + let mut client = ctx.publisher_client().await.unwrap(); + let (tx, rx) = mpsc::channel(16); + let mut responses = client + .publish_block_stream(ReceiverStream::new(rx)) + .await + .unwrap() + .into_inner(); + + // Block 0 - should WIN + let block0_bytes = BlockBuilder::new(0).build(); + let block0_proto: rock_node_protobufs::com::hedera::hapi::block::stream::Block = + Message::decode(block0_bytes.as_slice()).unwrap(); + + tx.send(PublishStreamRequest { + request: Some(PublishRequest::BlockItems(BlockItemSet { + block_items: block0_proto.items, + })), + }) + .await + .unwrap(); + + // Wait for ACK for block 0 + let resp0 = responses.next().await.unwrap().unwrap(); + assert!( + matches!( + resp0.response, + Some(PublishResponse::Acknowledgement(BlockAcknowledgement { + block_number: 0, + .. + })) + ), + "Publisher A should get ACK for block 0" + ); + let _ = block0_done_tx.send(()).await; // Signal block 0 complete + + // Wait for block 1 to complete (B's turn) + // First receive the ACK broadcast for block 1 + let resp1_ack = responses.next().await.unwrap().unwrap(); + assert!( + matches!( + resp1_ack.response, + Some(PublishResponse::Acknowledgement(BlockAcknowledgement { + block_number: 1, + .. + })) + ), + "Publisher A should get ACK broadcast for block 1" + ); + + // Wait for signal that block 1 is complete + let _ = block1_done_rx.recv().await; + + // Block 2 - should WIN + let block2_bytes = BlockBuilder::new(2).build(); + let block2_proto: rock_node_protobufs::com::hedera::hapi::block::stream::Block = + Message::decode(block2_bytes.as_slice()).unwrap(); + + tx.send(PublishStreamRequest { + request: Some(PublishRequest::BlockItems(BlockItemSet { + block_items: block2_proto.items, + })), + }) + .await + .unwrap(); + + // Wait for ACK for block 2 + let resp2 = responses.next().await.unwrap().unwrap(); + assert!( + matches!( + resp2.response, + Some(PublishResponse::Acknowledgement(BlockAcknowledgement { + block_number: 2, + .. + })) + ), + "Publisher A should get ACK for block 2" + ); + + println!("✅ Publisher A: Won blocks 0 and 2"); + }) + }; + + // Publisher B: publishes block 1 + let publisher_b_task = { + let ctx = Arc::clone(&ctx); + tokio::spawn(async move { + let mut client = ctx.publisher_client().await.unwrap(); + let (tx, rx) = mpsc::channel(16); + let mut responses = client + .publish_block_stream(ReceiverStream::new(rx)) + .await + .unwrap() + .into_inner(); + + // Wait for block 0 to complete (A's turn) + let _ = block0_done_rx.recv().await; + + // Receive the ACK broadcast for block 0 + let resp0_ack = responses.next().await.unwrap().unwrap(); + assert!( + matches!( + resp0_ack.response, + Some(PublishResponse::Acknowledgement(BlockAcknowledgement { + block_number: 0, + .. + })) + ), + "Publisher B should get ACK broadcast for block 0" + ); + + // Block 1 - should WIN + let block1_bytes = BlockBuilder::new(1).build(); + let block1_proto: rock_node_protobufs::com::hedera::hapi::block::stream::Block = + Message::decode(block1_bytes.as_slice()).unwrap(); + + tx.send(PublishStreamRequest { + request: Some(PublishRequest::BlockItems(BlockItemSet { + block_items: block1_proto.items, + })), + }) + .await + .unwrap(); + + // Wait for ACK for block 1 + let resp1 = responses.next().await.unwrap().unwrap(); + assert!( + matches!( + resp1.response, + Some(PublishResponse::Acknowledgement(BlockAcknowledgement { + block_number: 1, + .. + })) + ), + "Publisher B should get ACK for block 1" + ); + let _ = block1_done_tx.send(()).await; // Signal block 1 complete + + // Wait for block 2 ACK broadcast from A + let resp2_ack = responses.next().await.unwrap().unwrap(); + assert!( + matches!( + resp2_ack.response, + Some(PublishResponse::Acknowledgement(BlockAcknowledgement { + block_number: 2, + .. + })) + ), + "Publisher B should get ACK broadcast for block 2" + ); + + println!("✅ Publisher B: Won block 1"); + }) + }; + + // Wait for both to complete + publisher_a_task.await?; + publisher_b_task.await?; + + // Verify all 3 blocks were persisted + let access_client = ctx.access_client().await?; + for block_num in 0u64..3 { + let request = BlockRequest { + block_specifier: Some(BlockSpecifier::BlockNumber(block_num)), + }; + let response = access_client.clone().get_block(request).await?.into_inner(); + + assert!( + response.block.is_some(), + "Block {} should be persisted", + block_num + ); + } + + println!("✅ Different publishers win consecutive blocks: All blocks persisted correctly"); + + Ok(()) +} From 4f830e9b1b46441a45f7c97a1df6fdcac76b43a2 Mon Sep 17 00:00:00 2001 From: georgi-l95 Date: Tue, 7 Oct 2025 19:31:21 +0300 Subject: [PATCH 2/6] add more e2e tests Signed-off-by: georgi-l95 --- config/config.e2e.toml | 2 + .../src/session_manager.rs | 17 + tests/e2e/src/tests/mod.rs | 2 +- tests/e2e/src/tests/publish/basic.rs | 54 +++ tests/e2e/src/tests/publish/mod.rs | 3 + .../multi_publisher_basic.rs} | 368 ++---------------- .../src/tests/publish/validation_errors.rs | 305 +++++++++++++++ 7 files changed, 416 insertions(+), 335 deletions(-) create mode 100644 tests/e2e/src/tests/publish/basic.rs create mode 100644 tests/e2e/src/tests/publish/mod.rs rename tests/e2e/src/tests/{ingestion_suite.rs => publish/multi_publisher_basic.rs} (61%) create mode 100644 tests/e2e/src/tests/publish/validation_errors.rs diff --git a/config/config.e2e.toml b/config/config.e2e.toml index 1d8aaf1..33343f0 100644 --- a/config/config.e2e.toml +++ b/config/config.e2e.toml @@ -34,6 +34,8 @@ persistence_ack_timeout_seconds = 60 stale_winner_timeout_seconds = 60 winner_cleanup_interval_seconds = 300 winner_cleanup_threshold_blocks = 1000 +max_message_size_bytes = 33554432 # 32MB +max_items_per_set = 10000 [plugins.subscriber_service] enabled = true diff --git a/crates/rock-node-publish-plugin/src/session_manager.rs b/crates/rock-node-publish-plugin/src/session_manager.rs index d044931..34022da 100644 --- a/crates/rock-node-publish-plugin/src/session_manager.rs +++ b/crates/rock-node-publish-plugin/src/session_manager.rs @@ -121,6 +121,23 @@ impl SessionManager { .publish_blocks_received_total .with_label_values(&["validation_error"]) .inc(); + + // Send error response to client before terminating + let response = PublishStreamResponse { + response: Some(publish_stream_response::Response::EndStream( + publish_stream_response::EndOfStream { + status: publish_stream_response::end_of_stream::Code::Error as i32, + block_number: self.current_block_number as u64, + }, + )), + }; + let _ = self.send_response(response).await; + self.context + .metrics + .publish_responses_sent_total + .with_label_values(&["EndStream_ValidationError"]) + .inc(); + return true; // Terminate session on validation error } diff --git a/tests/e2e/src/tests/mod.rs b/tests/e2e/src/tests/mod.rs index 903c973..56d0cf8 100644 --- a/tests/e2e/src/tests/mod.rs +++ b/tests/e2e/src/tests/mod.rs @@ -3,7 +3,7 @@ pub mod backfill_suite; pub mod block_access_suite; pub mod gap_ingestion_suite; pub mod health_check; -pub mod ingestion_suite; +pub mod publish; pub mod query_suites; pub mod resiliency_suite; pub mod server_status_suite; diff --git a/tests/e2e/src/tests/publish/basic.rs b/tests/e2e/src/tests/publish/basic.rs new file mode 100644 index 0000000..92ace68 --- /dev/null +++ b/tests/e2e/src/tests/publish/basic.rs @@ -0,0 +1,54 @@ +use crate::common::{block_builder::BlockBuilder, TestContext}; +use anyhow::Result; +use prost::Message; +use rock_node_protobufs::org::hiero::block::api::{ + publish_stream_request::Request as PublishRequest, + publish_stream_response::{BlockAcknowledgement, Response as PublishResponse}, + BlockItemSet, PublishStreamRequest, +}; +use serial_test::serial; +use std::time::Duration; +use tokio::sync::mpsc; +use tokio_stream::{wrappers::ReceiverStream, StreamExt}; + +/// Test Case: Happy Path – a single publisher sends a full block and receives an acknowledgement. +#[tokio::test] +#[serial] +async fn test_publish_single_block_successfully() -> Result<()> { + let ctx = TestContext::new().await?; + let mut client = ctx.publisher_client().await?; + + let (tx, rx) = mpsc::channel(8); + let mut responses = client + .publish_block_stream(ReceiverStream::new(rx)) + .await? + .into_inner(); + + // Build a complete block + let block_bytes = BlockBuilder::new(0).build(); + let block_proto: rock_node_protobufs::com::hedera::hapi::block::stream::Block = + Message::decode(block_bytes.as_slice())?; + + // Send the complete block + tx.send(PublishStreamRequest { + request: Some(PublishRequest::BlockItems(BlockItemSet { + block_items: block_proto.items, + })), + }) + .await?; + + // Wait for acknowledgement + let response = tokio::time::timeout(Duration::from_secs(5), responses.next()) + .await? + .ok_or(anyhow::anyhow!("No response from server"))??; + + match response.response { + Some(PublishResponse::Acknowledgement(ack)) => { + assert_eq!(ack.block_number, 0, "Expected acknowledgement for block 0"); + }, + other => panic!("Expected Acknowledgement, got {:?}", other), + } + + println!("✅ Single block published and acknowledged successfully"); + Ok(()) +} diff --git a/tests/e2e/src/tests/publish/mod.rs b/tests/e2e/src/tests/publish/mod.rs new file mode 100644 index 0000000..ab712c4 --- /dev/null +++ b/tests/e2e/src/tests/publish/mod.rs @@ -0,0 +1,3 @@ +pub mod basic; +pub mod multi_publisher_basic; +pub mod validation_errors; diff --git a/tests/e2e/src/tests/ingestion_suite.rs b/tests/e2e/src/tests/publish/multi_publisher_basic.rs similarity index 61% rename from tests/e2e/src/tests/ingestion_suite.rs rename to tests/e2e/src/tests/publish/multi_publisher_basic.rs index b1856ef..6bb5385 100644 --- a/tests/e2e/src/tests/ingestion_suite.rs +++ b/tests/e2e/src/tests/publish/multi_publisher_basic.rs @@ -2,319 +2,20 @@ use crate::common::{block_builder::BlockBuilder, TestContext}; use anyhow::Result; use prost::Message; use rock_node_protobufs::org::hiero::block::api::{ - block_request::BlockSpecifier, publish_stream_request::Request as PublishRequest, - publish_stream_response::{ - end_of_stream::Code as EndCode, BlockAcknowledgement, Response as PublishResponse, - SkipBlock, - }, - BlockItemSet, BlockRequest, PublishStreamRequest, + publish_stream_response::{BlockAcknowledgement, Response as PublishResponse, SkipBlock}, + BlockItemSet, PublishStreamRequest, }; use serial_test::serial; +use std::collections::HashMap; use std::sync::Arc; use std::time::Duration; -use tokio::sync::mpsc; +use tokio::sync::{mpsc, Mutex}; use tokio_stream::{wrappers::ReceiverStream, StreamExt}; -/// Test Case: Happy Path – a single publisher sends a full block and receives an acknowledgement. -#[tokio::test] -#[serial] -async fn test_publish_single_block_successfully() -> Result<()> { - let ctx = TestContext::new().await?; - let mut client = ctx.publisher_client().await?; - - let block_bytes = BlockBuilder::new(0).build(); - let block_proto: rock_node_protobufs::com::hedera::hapi::block::stream::Block = - Message::decode(block_bytes.as_slice())?; - - let (tx, rx) = mpsc::channel(1); - tx.send(PublishStreamRequest { - request: Some(PublishRequest::BlockItems(BlockItemSet { - block_items: block_proto.items, - })), - }) - .await?; - - drop(tx); - - let response_stream = client.publish_block_stream(ReceiverStream::new(rx)).await?; - let mut responses = response_stream.into_inner(); - - let response = responses - .next() - .await - .expect("Server should have sent a response")?; - - match response.response { - Some(PublishResponse::Acknowledgement(ack)) => { - assert_eq!( - ack.block_number, 0, - "Acknowledgement has wrong block number" - ); - }, - other => panic!("Expected BlockAcknowledgement, got {:?}", other), - } - - assert!(responses.next().await.is_none()); - Ok(()) -} - -/// Test Case: Duplicate block – sending an already persisted block returns EndStream DuplicateBlock. -#[tokio::test] -#[serial] -async fn test_duplicate_block_rejected() -> Result<()> { - let ctx = TestContext::new().await?; - let mut client = ctx.publisher_client().await?; - - let block_bytes = BlockBuilder::new(0).build(); - let block_proto: rock_node_protobufs::com::hedera::hapi::block::stream::Block = - Message::decode(block_bytes.as_slice())?; - let (tx1, rx1) = mpsc::channel(1); - tx1.send(PublishStreamRequest { - request: Some(PublishRequest::BlockItems(BlockItemSet { - block_items: block_proto.items.clone(), - })), - }) - .await?; - drop(tx1); - - let mut responses = client - .publish_block_stream(ReceiverStream::new(rx1)) - .await? - .into_inner(); - while let Some(resp) = responses.next().await { - let resp = resp?; - if matches!(resp.response, Some(PublishResponse::Acknowledgement(_))) { - break; - } - } - - let mut client2 = ctx.publisher_client().await?; - let header_only = vec![block_proto.items[0].clone()]; - let (tx2, rx2) = mpsc::channel(1); - tx2.send(PublishStreamRequest { - request: Some(PublishRequest::BlockItems(BlockItemSet { - block_items: header_only, - })), - }) - .await?; - drop(tx2); - - let mut dup_resp_stream = client2 - .publish_block_stream(ReceiverStream::new(rx2)) - .await? - .into_inner(); - - let first_resp = dup_resp_stream - .next() - .await - .expect("Should receive response for duplicate block")?; - match first_resp.response { - Some(PublishResponse::EndStream(end)) => { - assert_eq!(end.status, EndCode::DuplicateBlock as i32); - assert_eq!( - end.block_number, 0, - "EndStream should report last persisted block" - ); - }, - other => panic!("Expected EndStream DuplicateBlock, got {:?}", other), - } - - Ok(()) -} - -/// Test Case: Race condition – two publishers send the same header, one wins, one gets SkipBlock. -#[tokio::test] -#[serial] -async fn test_multi_publisher_race_sends_skip_block() -> Result<()> { - let ctx = TestContext::new().await?; - - let mut client_a = ctx.publisher_client().await?; - let mut client_b = ctx.publisher_client().await?; - - let header_request = { - let block_bytes = BlockBuilder::new(0).build(); - let block_proto: rock_node_protobufs::com::hedera::hapi::block::stream::Block = - Message::decode(block_bytes.as_slice())?; - PublishStreamRequest { - request: Some(PublishRequest::BlockItems(BlockItemSet { - block_items: vec![block_proto.items[0].clone()], - })), - } - }; - - let (tx_a, rx_a) = mpsc::channel(4); - tx_a.send(header_request.clone()).await?; - let mut responses_a = client_a - .publish_block_stream(ReceiverStream::new(rx_a)) - .await? - .into_inner(); - - tokio::time::sleep(Duration::from_millis(50)).await; - - let (tx_b, rx_b) = mpsc::channel(4); - tx_b.send(header_request.clone()).await?; - let mut responses_b = client_b - .publish_block_stream(ReceiverStream::new(rx_b)) - .await? - .into_inner(); - - let b_response = tokio::time::timeout(Duration::from_secs(1), responses_b.next()) - .await? - .unwrap(); - assert!( - matches!( - b_response.unwrap().response, - Some(PublishResponse::SkipBlock(SkipBlock { block_number: 0 })) - ), - "Client B should have received SkipBlock" - ); - - let a_result = tokio::time::timeout(Duration::from_millis(200), responses_a.next()).await; - assert!( - a_result.is_err(), - "Client A should not have received a response yet" - ); - - Ok(()) -} - -/// Test Case: Multi-publisher happy path with broadcasted acknowledgements. -#[serial] -async fn test_multi_publisher_broadcast_ack() -> Result<()> { - let ctx = Arc::new(TestContext::new().await?); - - let client_a_task = { - let ctx = Arc::clone(&ctx); - tokio::spawn(async move { - let mut client = ctx.publisher_client().await.unwrap(); - let (tx, rx) = mpsc::channel(4); - let mut responses = client - .publish_block_stream(ReceiverStream::new(rx)) - .await - .unwrap() - .into_inner(); - - let block0_bytes = BlockBuilder::new(0).build(); - let block0_proto: rock_node_protobufs::com::hedera::hapi::block::stream::Block = - Message::decode(block0_bytes.as_slice()).unwrap(); - tx.send(PublishStreamRequest { - request: Some(PublishRequest::BlockItems(BlockItemSet { - block_items: block0_proto.items, - })), - }) - .await - .unwrap(); - - let resp0 = responses.next().await.unwrap().unwrap(); - assert!(matches!( - resp0.response, - Some(PublishResponse::Acknowledgement(BlockAcknowledgement { - block_number: 0, - .. - })) - )); - - let block1_header_bytes = BlockBuilder::new(1).items(); - tx.send(PublishStreamRequest { - request: Some(PublishRequest::BlockItems(BlockItemSet { - block_items: block1_header_bytes, - })), - }) - .await - .unwrap(); - - let resp1_skip = responses.next().await.unwrap().unwrap(); - assert!(matches!( - resp1_skip.response, - Some(PublishResponse::SkipBlock(SkipBlock { block_number: 1 })) - )); - - let resp1_ack = responses.next().await.unwrap().unwrap(); - assert!(matches!( - resp1_ack.response, - Some(PublishResponse::Acknowledgement(BlockAcknowledgement { - block_number: 1, - .. - })) - )); - }) - }; - - let client_b_task = { - let ctx = Arc::clone(&ctx); - tokio::spawn(async move { - let mut client = ctx.publisher_client().await.unwrap(); - let (tx, rx) = mpsc::channel(4); - let mut responses = client - .publish_block_stream(ReceiverStream::new(rx)) - .await - .unwrap() - .into_inner(); - - tokio::time::sleep(Duration::from_millis(100)).await; - - let block0_header_bytes = BlockBuilder::new(0).items(); - tx.send(PublishStreamRequest { - request: Some(PublishRequest::BlockItems(BlockItemSet { - block_items: block0_header_bytes, - })), - }) - .await - .unwrap(); - - let resp0_skip = responses.next().await.unwrap().unwrap(); - assert!(matches!( - resp0_skip.response, - Some(PublishResponse::SkipBlock(SkipBlock { block_number: 0 })) - )); - - let resp0_ack = responses.next().await.unwrap().unwrap(); - assert!(matches!( - resp0_ack.response, - Some(PublishResponse::Acknowledgement(BlockAcknowledgement { - block_number: 0, - .. - })) - )); - - let block1_bytes = BlockBuilder::new(1).build(); - let block1_proto: rock_node_protobufs::com::hedera::hapi::block::stream::Block = - Message::decode(block1_bytes.as_slice()).unwrap(); - tx.send(PublishStreamRequest { - request: Some(PublishRequest::BlockItems(BlockItemSet { - block_items: block1_proto.items, - })), - }) - .await - .unwrap(); - - let resp1 = responses.next().await.unwrap().unwrap(); - assert!(matches!( - resp1.response, - Some(PublishResponse::Acknowledgement(BlockAcknowledgement { - block_number: 1, - .. - })) - )); - }) - }; - - client_a_task.await?; - client_b_task.await?; - - Ok(()) -} - -// ============================================ -// CATEGORY 1: BASIC MULTI-PUBLISHER SCENARIOS -// ============================================ - -/// Test Case: Three publishers race for the same block -/// Verifies that with 3+ publishers racing: -/// - Exactly one wins (Primary) -/// - Other two receive SkipBlock -/// - All eventually receive ACK broadcast +/// Tests that when 3 publishers race for the same block +/// - Exactly one wins (no response initially) +/// - Exactly two lose (get SkipBlock) #[tokio::test] #[serial] async fn test_three_publishers_race_for_same_block() -> Result<()> { @@ -438,26 +139,27 @@ async fn test_three_publishers_race_for_same_block() -> Result<()> { Ok(()) } -/// Test Case: Five publishers competing with high contention -/// Tests fairness and correctness with 5 publishers racing for 10 blocks +/// Tests fairness in winner selection when multiple publishers compete for blocks. +/// Verifies: +/// - No single publisher monopolizes (max 60% wins) +/// - All publishers receive appropriate responses +/// - Most blocks are successfully processed #[tokio::test] #[serial] async fn test_five_publishers_high_contention() -> Result<()> { let ctx = Arc::new(TestContext::new().await?); + let winners: Arc>> = Arc::new(Mutex::new(HashMap::new())); - // Track which publisher wins which block - let winners = Arc::new(tokio::sync::Mutex::new(std::collections::HashMap::new())); - - // Create 5 publisher tasks let mut handles = vec![]; + // Spawn 5 publishers for publisher_id in 0..5 { let ctx = Arc::clone(&ctx); let winners = Arc::clone(&winners); let handle = tokio::spawn(async move { let mut client = ctx.publisher_client().await.unwrap(); - let (tx, rx) = mpsc::channel(32); + let (tx, rx) = mpsc::channel(16); let mut responses = client .publish_block_stream(ReceiverStream::new(rx)) .await @@ -598,36 +300,31 @@ async fn test_five_publishers_high_contention() -> Result<()> { winners_map.len() ); - // Check fairness - no single publisher should win everything - let mut win_counts = std::collections::HashMap::new(); - for winner_id in winners_map.values() { - *win_counts.entry(*winner_id).or_insert(0) += 1; - } - - for (publisher_id, count) in &win_counts { - println!("Publisher {} won {} blocks", publisher_id, count); - } - - // No single publisher should win more than 60% of blocks (fairness check) - for count in win_counts.values() { - assert!( - *count <= 6, - "Single publisher won too many blocks ({}), fairness violated", - count - ); + // Fairness check: No single publisher should win more than 60% of blocks + for (i, result) in results.iter().enumerate() { + if let Ok(wins) = result { + let win_rate = *wins as f64 / winners_map.len() as f64; + assert!( + win_rate <= 0.6, + "Publisher {} won {:.0}% of blocks ({}), exceeds 60% fairness threshold", + i, + win_rate * 100.0, + wins + ); + } } - println!("✅ Five publisher high contention: fairness and correctness verified"); + println!("✅ Five publishers high contention: Fairness and completion verified"); Ok(()) } -/// Test Case: Different publishers win consecutive blocks -/// Tests that: +/// Tests that winner selection isn't sticky to one publisher. +/// Verifies: /// - Publisher A wins block 0 /// - Publisher B wins block 1 /// - Publisher A wins block 2 -/// - Verifies that winner selection isn't sticky to one publisher +/// - All blocks are persisted in order #[tokio::test] #[serial] async fn test_different_publishers_win_consecutive_blocks() -> Result<()> { @@ -802,6 +499,9 @@ async fn test_different_publishers_win_consecutive_blocks() -> Result<()> { publisher_b_task.await?; // Verify all 3 blocks were persisted + use rock_node_protobufs::org::hiero::block::api::{ + block_request::BlockSpecifier, BlockRequest, + }; let access_client = ctx.access_client().await?; for block_num in 0u64..3 { let request = BlockRequest { diff --git a/tests/e2e/src/tests/publish/validation_errors.rs b/tests/e2e/src/tests/publish/validation_errors.rs new file mode 100644 index 0000000..7d9794e --- /dev/null +++ b/tests/e2e/src/tests/publish/validation_errors.rs @@ -0,0 +1,305 @@ +use crate::common::{block_builder::BlockBuilder, TestContext}; +use anyhow::Result; +use prost::Message; +use rock_node_protobufs::org::hiero::block::api::{ + publish_stream_request::Request as PublishRequest, + publish_stream_response::{end_of_stream::Code as EndCode, Response as PublishResponse}, + BlockItemSet, PublishStreamRequest, +}; +use serial_test::serial; +use std::time::Duration; +use tokio::sync::mpsc; +use tokio_stream::{wrappers::ReceiverStream, StreamExt}; + +/// Verifies that when a publisher tries to publish an already-persisted block, +/// they receive EndStream with DuplicateBlock status +#[tokio::test] +#[serial] +async fn test_duplicate_block_rejected() -> Result<()> { + let ctx = TestContext::new().await?; + let mut client = ctx.publisher_client().await?; + + let block_bytes = BlockBuilder::new(0).build(); + let block_proto: rock_node_protobufs::com::hedera::hapi::block::stream::Block = + Message::decode(block_bytes.as_slice())?; + let (tx1, rx1) = mpsc::channel(1); + tx1.send(PublishStreamRequest { + request: Some(PublishRequest::BlockItems(BlockItemSet { + block_items: block_proto.items.clone(), + })), + }) + .await?; + drop(tx1); + + let mut responses = client + .publish_block_stream(ReceiverStream::new(rx1)) + .await? + .into_inner(); + while let Some(resp) = responses.next().await { + let resp = resp?; + if matches!(resp.response, Some(PublishResponse::Acknowledgement(_))) { + break; + } + } + + let mut client2 = ctx.publisher_client().await?; + let header_only = vec![block_proto.items[0].clone()]; + let (tx2, rx2) = mpsc::channel(1); + tx2.send(PublishStreamRequest { + request: Some(PublishRequest::BlockItems(BlockItemSet { + block_items: header_only, + })), + }) + .await?; + drop(tx2); + + let mut dup_resp_stream = client2 + .publish_block_stream(ReceiverStream::new(rx2)) + .await? + .into_inner(); + + let first_resp = dup_resp_stream + .next() + .await + .expect("Should receive response for duplicate block")?; + match first_resp.response { + Some(PublishResponse::EndStream(end)) => { + assert_eq!(end.status, EndCode::DuplicateBlock as i32); + assert_eq!( + end.block_number, 0, + "EndStream should report last persisted block" + ); + }, + other => panic!("Expected EndStream DuplicateBlock, got {:?}", other), + } + + println!("✅ Duplicate block correctly rejected"); + Ok(()) +} + +/// Verifies that empty block item sets are rejected with validation error +#[tokio::test] +#[serial] +async fn test_empty_block_items_rejected() -> Result<()> { + let ctx = TestContext::new().await?; + let mut client = ctx.publisher_client().await?; + + let (tx, rx) = mpsc::channel(1); + + // Send empty block items + tx.send(PublishStreamRequest { + request: Some(PublishRequest::BlockItems(BlockItemSet { + block_items: vec![], + })), + }) + .await?; + drop(tx); + + let mut responses = client + .publish_block_stream(ReceiverStream::new(rx)) + .await? + .into_inner(); + + // Should receive EndStream with error + let response = tokio::time::timeout(Duration::from_secs(2), responses.next()) + .await? + .ok_or(anyhow::anyhow!("No response received"))??; + + match response.response { + Some(PublishResponse::EndStream(end)) => { + assert_eq!( + end.status, + EndCode::Error as i32, + "Expected Error status for empty block items" + ); + }, + other => panic!("Expected EndStream with Error, got {:?}", other), + } + + println!("✅ Empty block items correctly rejected"); + Ok(()) +} + +/// Category 2 - Test 2.2: Missing Header Validation +/// Verifies that block items without header are rejected +#[tokio::test] +#[serial] +async fn test_missing_header_rejected() -> Result<()> { + let ctx = TestContext::new().await?; + let mut client = ctx.publisher_client().await?; + + let (tx, rx) = mpsc::channel(1); + + // Create a block and send only the proof (no header) + let block_bytes = BlockBuilder::new(0).build(); + let block_proto: rock_node_protobufs::com::hedera::hapi::block::stream::Block = + Message::decode(block_bytes.as_slice())?; + + // Send only proof item (last item), skip header + let proof_only = vec![block_proto.items.last().unwrap().clone()]; + + tx.send(PublishStreamRequest { + request: Some(PublishRequest::BlockItems(BlockItemSet { + block_items: proof_only, + })), + }) + .await?; + drop(tx); + + let mut responses = client + .publish_block_stream(ReceiverStream::new(rx)) + .await? + .into_inner(); + + // Should receive EndStream with error + let response = tokio::time::timeout(Duration::from_secs(2), responses.next()) + .await? + .ok_or(anyhow::anyhow!("No response received"))??; + + match response.response { + Some(PublishResponse::EndStream(end)) => { + assert_eq!( + end.status, + EndCode::Error as i32, + "Expected Error status for missing header" + ); + }, + other => panic!("Expected EndStream with Error, got {:?}", other), + } + + println!("✅ Missing header correctly rejected"); + Ok(()) +} + +/// Verifies that blocks behind the persisted chain are rejected +#[tokio::test] +#[serial] +async fn test_behind_block_rejected() -> Result<()> { + let ctx = TestContext::new().await?; + + // First, publish block 0 to establish chain + let mut client1 = ctx.publisher_client().await?; + let (tx1, rx1) = mpsc::channel(1); + + let block0_bytes = BlockBuilder::new(0).build(); + let block0_proto: rock_node_protobufs::com::hedera::hapi::block::stream::Block = + Message::decode(block0_bytes.as_slice())?; + + tx1.send(PublishStreamRequest { + request: Some(PublishRequest::BlockItems(BlockItemSet { + block_items: block0_proto.items, + })), + }) + .await?; + drop(tx1); + + let mut responses1 = client1 + .publish_block_stream(ReceiverStream::new(rx1)) + .await? + .into_inner(); + + // Wait for ACK + while let Some(resp) = responses1.next().await { + let resp = resp?; + if matches!(resp.response, Some(PublishResponse::Acknowledgement(_))) { + break; + } + } + + // Now try to publish block 0 again from different publisher (behind) + let mut client2 = ctx.publisher_client().await?; + let (tx2, rx2) = mpsc::channel(1); + + let block0_again_bytes = BlockBuilder::new(0).build(); + let block0_again_proto: rock_node_protobufs::com::hedera::hapi::block::stream::Block = + Message::decode(block0_again_bytes.as_slice())?; + + tx2.send(PublishStreamRequest { + request: Some(PublishRequest::BlockItems(BlockItemSet { + block_items: vec![block0_again_proto.items[0].clone()], + })), + }) + .await?; + drop(tx2); + + let mut responses2 = client2 + .publish_block_stream(ReceiverStream::new(rx2)) + .await? + .into_inner(); + + // Should receive EndStream with DuplicateBlock + let response = tokio::time::timeout(Duration::from_secs(2), responses2.next()) + .await? + .ok_or(anyhow::anyhow!("No response received"))??; + + match response.response { + Some(PublishResponse::EndStream(end)) => { + assert_eq!( + end.status, + EndCode::DuplicateBlock as i32, + "Expected DuplicateBlock status" + ); + }, + other => panic!("Expected EndStream with DuplicateBlock, got {:?}", other), + } + + println!("✅ Behind block correctly rejected"); + Ok(()) +} + +/// Verifies that oversized block item sets are rejected +/// +/// NOTE: This test requires max_items_per_set to be explicitly set in config.e2e.toml +#[tokio::test] +#[serial] +async fn test_oversized_block_items_rejected() -> Result<()> { + let ctx = TestContext::new().await?; + let mut client = ctx.publisher_client().await?; + + let (tx, rx) = mpsc::channel(1); + + // Create a block with header + let block_bytes = BlockBuilder::new(0).build(); + let block_proto: rock_node_protobufs::com::hedera::hapi::block::stream::Block = + Message::decode(block_bytes.as_slice())?; + + // Create excessive number of items (more than max_items_per_set of 10000) + let mut large_items = vec![block_proto.items[0].clone()]; // header + + // Add items to exceed the limit + for _ in 1..10001 { + large_items.push(block_proto.items.last().unwrap().clone()); + } + + tx.send(PublishStreamRequest { + request: Some(PublishRequest::BlockItems(BlockItemSet { + block_items: large_items, + })), + }) + .await?; + drop(tx); + + let mut responses = client + .publish_block_stream(ReceiverStream::new(rx)) + .await? + .into_inner(); + + // Should receive EndStream with error + let response = tokio::time::timeout(Duration::from_secs(2), responses.next()) + .await? + .ok_or(anyhow::anyhow!("No response received"))??; + + match response.response { + Some(PublishResponse::EndStream(end)) => { + assert_eq!( + end.status, + EndCode::Error as i32, + "Expected Error status for oversized block" + ); + }, + other => panic!("Expected EndStream with Error, got {:?}", other), + } + + println!("✅ Oversized block items correctly rejected"); + Ok(()) +} From 7f2dcdadf22bd68a10ceb2fcdf8d2adfb5bbd087 Mon Sep 17 00:00:00 2001 From: georgi-l95 Date: Wed, 8 Oct 2025 01:52:27 +0300 Subject: [PATCH 3/6] fix unit test Signed-off-by: georgi-l95 --- crates/rock-node-subscriber-plugin/src/session.rs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/crates/rock-node-subscriber-plugin/src/session.rs b/crates/rock-node-subscriber-plugin/src/session.rs index a4af142..c6a07fd 100644 --- a/crates/rock-node-subscriber-plugin/src/session.rs +++ b/crates/rock-node-subscriber-plugin/src/session.rs @@ -380,8 +380,8 @@ mod tests { stale_winner_timeout_seconds: 0, winner_cleanup_interval_seconds: 0, winner_cleanup_threshold_blocks: 0, - max_message_size_bytes: todo!(), - max_items_per_set: todo!(), + max_message_size_bytes: 0, + max_items_per_set: 0, }, verification_service: VerificationServiceConfig { enabled: false }, block_access_service: BlockAccessServiceConfig { enabled: false }, From 45d300e50b64bdba18de20057fcd7cab7edcfa38 Mon Sep 17 00:00:00 2001 From: georgi-l95 Date: Wed, 8 Oct 2025 15:38:18 +0300 Subject: [PATCH 4/6] subscriber hardening Signed-off-by: georgi-l95 --- .../rock-node-subscriber-plugin/src/error.rs | 7 +- .../src/service.rs | 12 +- .../src/session.rs | 447 ++++++++++++++++-- tests/e2e/src/tests/mod.rs | 1 + tests/e2e/src/tests/subscribe/basic.rs | 221 +++++++++ tests/e2e/src/tests/subscribe/edge_cases.rs | 353 ++++++++++++++ .../src/tests/subscribe/failure_scenarios.rs | 177 +++++++ tests/e2e/src/tests/subscribe/mod.rs | 4 + tests/e2e/src/tests/subscribe/validation.rs | 385 +++++++++++++++ 9 files changed, 1575 insertions(+), 32 deletions(-) create mode 100644 tests/e2e/src/tests/subscribe/basic.rs create mode 100644 tests/e2e/src/tests/subscribe/edge_cases.rs create mode 100644 tests/e2e/src/tests/subscribe/failure_scenarios.rs create mode 100644 tests/e2e/src/tests/subscribe/mod.rs create mode 100644 tests/e2e/src/tests/subscribe/validation.rs diff --git a/crates/rock-node-subscriber-plugin/src/error.rs b/crates/rock-node-subscriber-plugin/src/error.rs index b956fae..122cab7 100644 --- a/crates/rock-node-subscriber-plugin/src/error.rs +++ b/crates/rock-node-subscriber-plugin/src/error.rs @@ -43,7 +43,12 @@ impl SubscriberError { /// Maps an internal error to a label for the metrics counter. pub fn to_metric_label(&self) -> &'static str { match self { - SubscriberError::Validation(_, _) => "invalid_request", + SubscriberError::Validation(_, code) => match code { + Code::InvalidStartBlockNumber => "invalid_start_block", + Code::InvalidEndBlockNumber => "invalid_end_block", + Code::NotAvailable => "not_available", + _ => "invalid_request", + }, SubscriberError::Persistence(_) => "persistence_error", SubscriberError::ClientDisconnected => "client_disconnect", SubscriberError::StreamLagged => "stream_lagged", diff --git a/crates/rock-node-subscriber-plugin/src/service.rs b/crates/rock-node-subscriber-plugin/src/service.rs index 7ef6cf7..6532d19 100644 --- a/crates/rock-node-subscriber-plugin/src/service.rs +++ b/crates/rock-node-subscriber-plugin/src/service.rs @@ -9,7 +9,7 @@ use std::sync::Arc; use tokio::sync::{mpsc, Notify}; use tokio_stream::wrappers::ReceiverStream; use tonic::{Request, Response, Status}; -use tracing::{error, info}; +use tracing::{debug, error, info}; use uuid::Uuid; /// Implements the gRPC service for subscribing to block streams. @@ -68,10 +68,16 @@ impl BlockStreamSubscribeService for SubscriberServiceImpl { let active_sessions_clone = self.active_sessions.clone(); tokio::spawn(async move { - info!(session_id = %session.id, "Spawning new session handler task."); + debug!(session_id = %session.id, "Session handler task spawned"); session.run().await; + + // Log failure cause if any + if let Some(cause) = session.get_failure_cause() { + info!(session_id = %session.id, "Session ended with failure: {}", cause); + } + active_sessions_clone.remove(&session_id); - info!(session_id = %session.id, "Session handler task finished."); + debug!(session_id = %session.id, "Session handler task finished"); }); Ok(Response::new(ReceiverStream::new(rx))) diff --git a/crates/rock-node-subscriber-plugin/src/session.rs b/crates/rock-node-subscriber-plugin/src/session.rs index c6a07fd..fccc7fb 100644 --- a/crates/rock-node-subscriber-plugin/src/session.rs +++ b/crates/rock-node-subscriber-plugin/src/session.rs @@ -14,7 +14,7 @@ use std::{ }; use tokio::sync::{mpsc, Notify}; use tonic::Status; -use tracing::{debug, info, warn}; +use tracing::{debug, error, info, trace, warn}; use uuid::Uuid; pub struct SubscriberSession { @@ -26,6 +26,7 @@ pub struct SubscriberSession { server_shutdown_notify: Arc, session_shutdown_notify: Arc, inter_block_metrics: Mutex, + failure_cause: Arc>>, } impl SubscriberSession { @@ -57,35 +58,66 @@ impl SubscriberSession { server_shutdown_notify, session_shutdown_notify, inter_block_metrics: Mutex::new(Default::default()), + failure_cause: Arc::new(Mutex::new(None)), }) } + /// Get the failure cause for this session (for metrics/logging) + pub fn get_failure_cause(&self) -> Option { + self.failure_cause.lock().ok()?.clone() + } + pub async fn run(&mut self) { + info!( + session_id = %self.id, + start_block = self.request.start_block_number, + end_block = self.request.end_block_number, + "Subscriber session started" + ); + self.context.metrics.subscriber_active_sessions.inc(); let _drop_guard = DropGuard::new(self.context.metrics.subscriber_active_sessions.clone()); let result = self.execute_stream().await; - self.finalize_stream(result).await; + let success = result.is_ok(); + self.finalize_stream(&result).await; + + info!( + session_id = %self.id, + success = success, + "Subscriber session completed" + ); } async fn execute_stream(&mut self) -> Result<(), SubscriberError> { - let mut next_block_to_send = self.validate_request().await?; + let mut next_block_to_send = self.validate_and_check_fulfillment().await?; let is_finite_stream = self.request.end_block_number != u64::MAX; + debug!( + session_id = %self.id, + next_block = next_block_to_send, + is_finite = is_finite_stream, + "Starting block streaming loop" + ); + loop { tokio::select! { _ = self.server_shutdown_notify.notified() => { + debug!(session_id = %self.id, "Received server shutdown signal"); return Err(SubscriberError::ServerShutdown); } _ = self.session_shutdown_notify.notified() => { + debug!(session_id = %self.id, "Received session shutdown signal"); return Err(SubscriberError::ServerShutdown); } res = self.process_next_block(next_block_to_send, is_finite_stream) => { match res { Ok(Some(next_block)) => { + trace!(session_id = %self.id, "Advanced to next block: {}", next_block); next_block_to_send = next_block; } Ok(None) => { + debug!(session_id = %self.id, "All requested blocks sent"); return Ok(()); } Err(e) => { @@ -103,11 +135,14 @@ impl SubscriberSession { is_finite: bool, ) -> Result, SubscriberError> { if is_finite && block_to_send > self.request.end_block_number { + trace!(session_id = %self.id, "Reached end of finite stream at block {}", self.request.end_block_number); return Ok(None); } + trace!(session_id = %self.id, "Attempting to read block #{}", block_to_send); match self.block_reader.read_block(block_to_send) { Ok(Some(block_bytes)) => { + trace!(session_id = %self.id, "Block #{} found in persistence, sending", block_to_send); self.send_block(&block_bytes).await?; self.context .metrics @@ -116,8 +151,13 @@ impl SubscriberSession { .inc(); return Ok(Some(block_to_send + 1)); }, - Ok(None) => {}, - Err(e) => return Err(SubscriberError::Persistence(e)), + Ok(None) => { + trace!(session_id = %self.id, "Block #{} not yet persisted, waiting for live stream", block_to_send); + }, + Err(e) => { + error!(session_id = %self.id, "Persistence error reading block #{}: {}", block_to_send, e); + return Err(SubscriberError::Persistence(e)); + }, } self.wait_for_new_block(block_to_send).await?; @@ -163,43 +203,149 @@ impl SubscriberSession { } } - async fn validate_request(&self) -> Result { + /// Validates the request and checks if the node can fulfill it. + /// This is similar to Java's canFulfillRequest() method. + async fn validate_and_check_fulfillment(&self) -> Result { let req_start = self.request.start_block_number; let req_end = self.request.end_block_number; + + trace!( + session_id = %self.id, + req_start = req_start, + req_end = req_end, + "Validating subscription request" + ); + + // Get node's block availability info let earliest = self.block_reader.get_earliest_persisted_block_number()?; - let start_block = match req_start { - u64::MAX => earliest.unwrap_or(0), - literal_start => literal_start, - }; - if req_end != u64::MAX && start_block > req_end { + let latest = self.block_reader.get_latest_persisted_block_number()?; + let max_future_lookahead = self + .context + .config + .plugins + .subscriber_service + .max_future_block_lookahead; + + debug!( + session_id = %self.id, + earliest = ?earliest, + latest = ?latest, + max_future_lookahead = max_future_lookahead, + "Block availability: earliest={:?}, latest={:?}", + earliest, latest + ); + + // Basic validation: end must not be before start + if req_end != u64::MAX && req_start != u64::MAX && req_start > req_end { + let msg = format!( + "Invalid request: start block {} cannot be after end block {}", + req_start, req_end + ); + warn!(session_id = %self.id, "{}", msg); return Err(SubscriberError::Validation( - format!( - "Start block {} cannot be after end block {}", - start_block, req_end - ), + msg, Code::InvalidEndBlockNumber, )); } + + // Determine the actual start block + let start_block = match req_start { + u64::MAX => { + // Start from earliest available, or 0 if no blocks yet + let block = earliest.unwrap_or(0); + debug!(session_id = %self.id, "Request start=MAX, using earliest available: {}", block); + block + }, + literal_start => { + debug!(session_id = %self.id, "Request start={}", literal_start); + literal_start + }, + }; + + // Case 1: Pure live streaming (start=MAX, end=MAX) + if req_start == u64::MAX && req_end == u64::MAX { + info!(session_id = %self.id, "Pure live streaming mode (start=MAX, end=MAX)"); + return Ok(start_block); + } + + // Case 2: From earliest to specific end (start=MAX, end=N) + if req_start == u64::MAX && req_end != u64::MAX { + if let Some(earliest_num) = earliest { + if earliest_num <= req_end { + info!( + session_id = %self.id, + "Streaming from earliest ({}) to end ({})", + earliest_num, req_end + ); + return Ok(earliest_num); + } else { + let msg = format!( + "Cannot fulfill request: earliest available block {} is after requested end block {}", + earliest_num, req_end + ); + warn!(session_id = %self.id, "{}", msg); + return Err(SubscriberError::Validation(msg, Code::NotAvailable)); + } + } else { + let msg = "Cannot fulfill request: no blocks available yet".to_string(); + warn!(session_id = %self.id, "{}", msg); + return Err(SubscriberError::Validation(msg, Code::NotAvailable)); + } + } + + // For cases with explicit start block, validate against available blocks if let Some(earliest_num) = earliest { if start_block < earliest_num { + let msg = format!( + "Requested start block {} is before earliest available block {}", + start_block, earliest_num + ); + warn!(session_id = %self.id, "{}", msg); return Err(SubscriberError::Validation( - format!( - "Requested start block {} is earlier than the first available block {}", - start_block, earliest_num - ), + msg, Code::InvalidStartBlockNumber, )); } } + + // Check if start is too far in the future + if let Some(latest_num) = latest { + let max_permitted_start = latest_num + max_future_lookahead; + if start_block > max_permitted_start { + let msg = format!( + "Requested start block {} is too far in the future (latest: {}, max permitted: {})", + start_block, latest_num, max_permitted_start + ); + warn!(session_id = %self.id, "{}", msg); + return Err(SubscriberError::Validation(msg, Code::NotAvailable)); + } + } + + // Case 3: Specific start to indefinite end (start=N, end=MAX) + if req_end == u64::MAX { + info!( + session_id = %self.id, + "Streaming from block {} indefinitely", start_block + ); + return Ok(start_block); + } + + // Case 4: Specific range (start=N, end=M) + info!( + session_id = %self.id, + "Streaming finite range from {} to {}", start_block, req_end + ); Ok(start_block) } async fn send_block(&self, block_bytes: &[u8]) -> Result<(), SubscriberError> { + let block = + Block::decode(block_bytes).map_err(|e| SubscriberError::Persistence(e.into()))?; + let item_set = BlockItemSet { - block_items: Block::decode(block_bytes) - .map_err(|e| SubscriberError::Persistence(e.into()))? - .items, + block_items: block.items, }; + { // Update inter-block timing statistics let now = Instant::now(); @@ -221,22 +367,62 @@ impl SubscriberSession { } } - info!(session_id = %self.id, "Sending block #{:?} to client", item_set.block_items[0]); + trace!(session_id = %self.id, "Sending block to client"); let response = SubscribeStreamResponse { response: Some(ResponseType::BlockItems(item_set)), }; + if self.response_tx.send(Ok(response)).await.is_err() { + warn!(session_id = %self.id, "Client disconnected while sending block"); Err(SubscriberError::ClientDisconnected) } else { Ok(()) } } - async fn finalize_stream(&self, result: Result<(), SubscriberError>) { + async fn finalize_stream(&self, result: &Result<(), SubscriberError>) { let (final_code, outcome_label) = match result { - Ok(()) => (Code::Success, "completed"), - Err(e) => (e.to_status_code(), e.to_metric_label()), + Ok(()) => { + info!(session_id = %self.id, "Stream completed successfully"); + (Code::Success, "completed") + }, + Err(e) => { + // Store failure cause for inspection + if let Ok(mut cause) = self.failure_cause.lock() { + *cause = Some(e.to_string()); + } + + let code = e.to_status_code(); + let label = e.to_metric_label(); + + match e { + SubscriberError::ClientDisconnected => { + debug!(session_id = %self.id, "Client disconnected"); + }, + SubscriberError::ServerShutdown => { + debug!(session_id = %self.id, "Server shutdown"); + }, + SubscriberError::TimeoutWaitingForBlock(block) => { + warn!(session_id = %self.id, "Timeout waiting for block #{}", block); + }, + SubscriberError::Validation(msg, _) => { + info!(session_id = %self.id, "Validation error: {}", msg); + }, + SubscriberError::Persistence(err) => { + error!(session_id = %self.id, "Persistence error: {}", err); + }, + SubscriberError::StreamLagged => { + warn!(session_id = %self.id, "Stream lagged too far behind"); + }, + SubscriberError::Internal(msg) => { + error!(session_id = %self.id, "Internal error: {}", msg); + }, + } + + (code, label) + }, }; + self.context .metrics .subscriber_sessions_total @@ -248,6 +434,11 @@ impl SubscriberSession { let metrics_lock = self.inter_block_metrics.lock().unwrap(); if metrics_lock.interval_count > 0 { let avg = metrics_lock.total_interval / metrics_lock.interval_count as f64; + debug!( + session_id = %self.id, + avg_inter_block_seconds = avg, + "Average inter-block time recorded" + ); self.context .metrics .subscriber_average_inter_block_time_seconds @@ -260,14 +451,19 @@ impl SubscriberSession { async fn send_final_status(&self, code: Code) { if let Code::Unknown = code { + trace!(session_id = %self.id, "Skipping final status (code=Unknown)"); return; } - info!(session_id = %self.id, "SENDING FINAL STATUS: {:?}", code); + + debug!(session_id = %self.id, "Sending final status: {:?}", code); let response = SubscribeStreamResponse { response: Some(ResponseType::Status(code.into())), }; + if self.response_tx.send(Ok(response)).await.is_err() { - debug!(session_id = %self.id, "Client disconnected before final status could be sent."); + debug!(session_id = %self.id, "Client disconnected before final status could be sent"); + } else { + trace!(session_id = %self.id, "Final status sent successfully"); } } } @@ -609,4 +805,199 @@ mod tests { // The channel should now be closed assert!(rx.recv().await.is_none()); } + + // ===== New Validation Tests ===== + + #[tokio::test] + async fn test_validation_start_too_far_in_future() { + let reader = Arc::new(MockBlockReader::default()); + reader.add_block(0); + reader.add_block(1); + reader.add_block(2); + let (context, _) = create_test_context(reader.clone()); + + // Latest is 2, max_future_lookahead is 100, so max permitted is 102 + let request = SubscribeStreamRequest { + start_block_number: 200, // Too far in future + end_block_number: u64::MAX, + }; + let (tx, mut rx) = mpsc::channel(16); + + let server_shutdown = Arc::new(Notify::new()); + let session_shutdown = Arc::new(Notify::new()); + let mut session = + SubscriberSession::new(context, request, tx, server_shutdown, session_shutdown) + .expect("Failed to create test session"); + + tokio::spawn(async move { + session.run().await; + }); + + let final_msg = rx.recv().await.unwrap().unwrap(); + assert_eq!( + final_msg.response, + Some(ResponseType::Status(Code::NotAvailable.into())) + ); + } + + #[tokio::test] + async fn test_validation_pure_live_streaming() { + let reader = Arc::new(MockBlockReader::default()); + let (context, _) = create_test_context(reader); + + let request = SubscribeStreamRequest { + start_block_number: u64::MAX, + end_block_number: u64::MAX, + }; + let (tx, _rx) = mpsc::channel(16); + + let server_shutdown = Arc::new(Notify::new()); + let session_shutdown = Arc::new(Notify::new()); + let session = + SubscriberSession::new(context, request, tx, server_shutdown, session_shutdown) + .expect("Failed to create test session"); + + // Should accept pure live streaming even with no blocks + let result = session.validate_and_check_fulfillment().await; + assert!(result.is_ok()); + } + + #[tokio::test] + async fn test_validation_from_earliest_to_end() { + let reader = Arc::new(MockBlockReader::default()); + for i in 5..=15 { + reader.add_block(i); + } + let (context, _) = create_test_context(reader); + + let request = SubscribeStreamRequest { + start_block_number: u64::MAX, // From earliest + end_block_number: 20, + }; + let (tx, _rx) = mpsc::channel(16); + + let server_shutdown = Arc::new(Notify::new()); + let session_shutdown = Arc::new(Notify::new()); + let session = + SubscriberSession::new(context, request, tx, server_shutdown, session_shutdown) + .expect("Failed to create test session"); + + let result = session.validate_and_check_fulfillment().await; + assert!(result.is_ok()); + assert_eq!(result.unwrap(), 5); // Should start at earliest (5) + } + + #[tokio::test] + async fn test_validation_from_earliest_but_end_before_earliest() { + let reader = Arc::new(MockBlockReader::default()); + for i in 10..=20 { + reader.add_block(i); + } + let (context, _) = create_test_context(reader); + + let request = SubscribeStreamRequest { + start_block_number: u64::MAX, // From earliest (10) + end_block_number: 5, // Before earliest + }; + let (tx, mut rx) = mpsc::channel(16); + + let server_shutdown = Arc::new(Notify::new()); + let session_shutdown = Arc::new(Notify::new()); + let mut session = + SubscriberSession::new(context, request, tx, server_shutdown, session_shutdown) + .expect("Failed to create test session"); + + tokio::spawn(async move { + session.run().await; + }); + + let final_msg = rx.recv().await.unwrap().unwrap(); + assert_eq!( + final_msg.response, + Some(ResponseType::Status(Code::NotAvailable.into())) + ); + } + + #[tokio::test] + async fn test_validation_no_blocks_available_explicit_start() { + let reader = Arc::new(MockBlockReader::default()); + // No blocks added + let (context, _) = create_test_context(reader); + + let request = SubscribeStreamRequest { + start_block_number: 100, + end_block_number: u64::MAX, + }; + let (tx, _rx) = mpsc::channel(16); + + let server_shutdown = Arc::new(Notify::new()); + let session_shutdown = Arc::new(Notify::new()); + let session = + SubscriberSession::new(context, request, tx, server_shutdown, session_shutdown) + .expect("Failed to create test session"); + + // Should accept since start could be future block + let result = session.validate_and_check_fulfillment().await; + assert!(result.is_ok()); + assert_eq!(result.unwrap(), 100); + } + + #[tokio::test] + async fn test_validation_within_future_lookahead() { + let reader = Arc::new(MockBlockReader::default()); + for i in 0..=10 { + reader.add_block(i); + } + let (context, _) = create_test_context(reader); + + // Latest is 10, max_future_lookahead is 100, so 110 is max permitted + let request = SubscribeStreamRequest { + start_block_number: 50, // Within lookahead + end_block_number: u64::MAX, + }; + let (tx, _rx) = mpsc::channel(16); + + let server_shutdown = Arc::new(Notify::new()); + let session_shutdown = Arc::new(Notify::new()); + let session = + SubscriberSession::new(context, request, tx, server_shutdown, session_shutdown) + .expect("Failed to create test session"); + + let result = session.validate_and_check_fulfillment().await; + assert!(result.is_ok()); + assert_eq!(result.unwrap(), 50); + } + + #[tokio::test] + async fn test_failure_cause_tracking() { + let reader = Arc::new(MockBlockReader::default()); + let (context, _) = create_test_context(reader); + + let request = SubscribeStreamRequest { + start_block_number: 10, + end_block_number: 5, // Invalid: end before start + }; + let (tx, mut rx) = mpsc::channel(16); + + let server_shutdown = Arc::new(Notify::new()); + let session_shutdown = Arc::new(Notify::new()); + let mut session = + SubscriberSession::new(context, request, tx, server_shutdown, session_shutdown) + .expect("Failed to create test session"); + + tokio::spawn(async move { + session.run().await; + // After run completes, failure cause should be set + assert!(session.get_failure_cause().is_some()); + let cause = session.get_failure_cause().unwrap(); + assert!(cause.contains("cannot be after end block")); + }); + + // Should get error status + let final_msg = rx.recv().await.unwrap().unwrap(); + assert_eq!( + final_msg.response, + Some(ResponseType::Status(Code::InvalidEndBlockNumber.into())) + ); + } } diff --git a/tests/e2e/src/tests/mod.rs b/tests/e2e/src/tests/mod.rs index 56d0cf8..e126a23 100644 --- a/tests/e2e/src/tests/mod.rs +++ b/tests/e2e/src/tests/mod.rs @@ -7,4 +7,5 @@ pub mod publish; pub mod query_suites; pub mod resiliency_suite; pub mod server_status_suite; +pub mod subscribe; pub mod subscription_suite; diff --git a/tests/e2e/src/tests/subscribe/basic.rs b/tests/e2e/src/tests/subscribe/basic.rs new file mode 100644 index 0000000..943fb9c --- /dev/null +++ b/tests/e2e/src/tests/subscribe/basic.rs @@ -0,0 +1,221 @@ +use crate::common::{publish_blocks, TestContext}; +use anyhow::Result; +use prost::Message; +use rock_node_protobufs::org::hiero::block::api::{ + block_stream_publish_service_client::BlockStreamPublishServiceClient, + publish_stream_request::Request as PublishRequest, BlockItemSet, PublishStreamRequest, +}; +use rock_node_protobufs::org::hiero::block::api::{ + subscribe_stream_response::{Code as SubCode, Response as SubResponse}, + SubscribeStreamRequest, +}; +use serial_test::serial; +use std::sync::Arc; +use tokio::sync::mpsc; +use tokio::time::Duration; +use tokio_stream::wrappers::ReceiverStream; +use tokio_stream::StreamExt; +use tonic::transport::Channel; + +/// Helper to extract the block number from a `BlockItems` response. +fn header_number(item_set: &rock_node_protobufs::org::hiero::block::api::BlockItemSet) -> u64 { + use rock_node_protobufs::com::hedera::hapi::block::stream::block_item::Item; + item_set + .block_items + .iter() + .find_map(|item| match &item.item { + Some(Item::BlockHeader(h)) => Some(h.number), + _ => None, + }) + .expect("Block items must contain a header") +} + +/// Test Case: Finite historical range subscription. +/// Tests that a subscriber can request a specific range of already-persisted blocks. +#[tokio::test] +#[serial] +async fn test_historical_stream() -> Result<()> { + let ctx = TestContext::new().await?; + + publish_blocks(&ctx, 0, 20).await?; + + let mut sub_client = ctx.subscriber_client().await?; + let request = SubscribeStreamRequest { + start_block_number: 5, + end_block_number: 10, + }; + + let mut stream = sub_client + .subscribe_block_stream(request) + .await? + .into_inner(); + let mut expected = 5u64; + let mut received = 0u64; + + while let Some(msg) = stream.next().await { + let msg = msg?; + match msg.response { + Some(SubResponse::BlockItems(set)) => { + let num = header_number(&set); + assert_eq!(num, expected, "Blocks received out of order"); + expected += 1; + received += 1; + }, + Some(SubResponse::Status(code)) => { + assert_eq!(code, SubCode::Success as i32); + break; + }, + other => panic!("Unexpected response: {:?}", other), + } + } + + assert_eq!(received, 6, "Should have received 6 blocks (5..=10)"); + Ok(()) +} + +/// Test Case: Historical to live streaming transition. +/// Tests that a subscriber can seamlessly transition from reading persisted blocks +/// to waiting for new blocks to arrive. +#[tokio::test] +#[serial] +async fn test_historical_to_live_stream() -> Result<()> { + let ctx = TestContext::new().await?; + + publish_blocks(&ctx, 0, 5).await?; + + let mut sub_client = ctx.subscriber_client().await?; + let request = SubscribeStreamRequest { + start_block_number: 0, + end_block_number: 10, + }; + let mut stream = sub_client + .subscribe_block_stream(request) + .await? + .into_inner(); + + let publish_port = ctx.container.get_host_port_ipv4(50051).await?; + let publish_endpoint = format!("http://localhost:{}", publish_port); + + tokio::spawn(async move { + tokio::time::sleep(Duration::from_millis(200)).await; + let channel = Channel::from_shared(publish_endpoint) + .unwrap() + .connect() + .await + .unwrap(); + let mut client = BlockStreamPublishServiceClient::new(channel); + for i in 6..=10 { + let block_bytes = crate::common::block_builder::BlockBuilder::new(i).build(); + let block_proto: rock_node_protobufs::com::hedera::hapi::block::stream::Block = + Message::decode(block_bytes.as_slice()).unwrap(); + + let (tx, rx) = mpsc::channel(1); + tx.send(PublishStreamRequest { + request: Some(PublishRequest::BlockItems(BlockItemSet { + block_items: block_proto.items, + })), + }) + .await + .unwrap(); + drop(tx); + + let mut responses = client + .publish_block_stream(ReceiverStream::new(rx)) + .await + .unwrap() + .into_inner(); + while responses.next().await.is_some() {} + } + }); + + let mut expected = 0u64; + while let Some(msg) = stream.next().await { + let msg = msg?; + match msg.response { + Some(SubResponse::BlockItems(set)) => { + let num = header_number(&set); + assert_eq!(num, expected); + expected += 1; + }, + Some(SubResponse::Status(code)) => { + assert_eq!(code, SubCode::Success as i32); + break; + }, + other => panic!("Unexpected response: {:?}", other), + } + } + + assert_eq!(expected, 11, "Should have received 11 blocks (0..=10)"); + Ok(()) +} + +/// Test Case: Multiple concurrent subscribers with different ranges. +/// Tests that multiple independent subscribers can operate simultaneously +/// without interfering with each other. +#[tokio::test] +#[serial] +async fn test_multiple_concurrent_subscribers() -> Result<()> { + let ctx = Arc::new(TestContext::new().await?); + publish_blocks(&ctx, 0, 20).await?; + + // --- Subscriber A (historical only) --- + let ctx_a = Arc::clone(&ctx); + let task_a = tokio::spawn(async move { + let mut client = ctx_a.subscriber_client().await.unwrap(); + let request = SubscribeStreamRequest { + start_block_number: 5, + end_block_number: 15, + }; + let mut stream = client + .subscribe_block_stream(request) + .await + .unwrap() + .into_inner(); + let mut received_blocks = Vec::new(); + while let Some(Ok(msg)) = stream.next().await { + if let Some(SubResponse::BlockItems(set)) = msg.response { + received_blocks.push(header_number(&set)); + } + } + received_blocks + }); + + // --- Subscriber B (historical and live) --- + let ctx_b = Arc::clone(&ctx); + let task_b = tokio::spawn(async move { + let mut client = ctx_b.subscriber_client().await.unwrap(); + let request = SubscribeStreamRequest { + start_block_number: 18, + end_block_number: u64::MAX, + }; + let mut stream = client + .subscribe_block_stream(request) + .await + .unwrap() + .into_inner(); + let mut received_blocks = Vec::new(); + while let Some(Ok(msg)) = stream.next().await { + if let Some(SubResponse::BlockItems(set)) = msg.response { + let block_num = header_number(&set); + received_blocks.push(block_num); + if block_num == 22 { + break; + } + } + } + received_blocks + }); + + tokio::time::sleep(Duration::from_millis(200)).await; + + publish_blocks(&ctx, 21, 22).await?; + + // --- Assertions --- + let received_a = task_a.await?; + assert_eq!(received_a, (5..=15).collect::>()); + + let received_b = task_b.await?; + assert_eq!(received_b, (18..=22).collect::>()); + + Ok(()) +} diff --git a/tests/e2e/src/tests/subscribe/edge_cases.rs b/tests/e2e/src/tests/subscribe/edge_cases.rs new file mode 100644 index 0000000..459c7ef --- /dev/null +++ b/tests/e2e/src/tests/subscribe/edge_cases.rs @@ -0,0 +1,353 @@ +use crate::common::{publish_blocks, TestContext}; +use anyhow::Result; +use prost::Message; +use rock_node_protobufs::org::hiero::block::api::{ + block_stream_publish_service_client::BlockStreamPublishServiceClient, + publish_stream_request::Request as PublishRequest, + subscribe_stream_response::{Code as SubCode, Response as SubResponse}, + BlockItemSet, PublishStreamRequest, SubscribeStreamRequest, +}; +use serial_test::serial; +use std::sync::Arc; +use tokio::sync::mpsc; +use tokio_stream::wrappers::ReceiverStream; +use tokio_stream::StreamExt; +use tonic::transport::Channel; + +/// Helper to extract the block number from a `BlockItems` response. +fn header_number(item_set: &rock_node_protobufs::org::hiero::block::api::BlockItemSet) -> u64 { + use rock_node_protobufs::com::hedera::hapi::block::stream::block_item::Item; + item_set + .block_items + .iter() + .find_map(|item| match &item.item { + Some(Item::BlockHeader(h)) => Some(h.number), + _ => None, + }) + .expect("Block items must contain a header") +} + +/// Test Case: Subscribe to range that will have a gap filled live. +/// Tests that a subscriber can successfully receive blocks even when there's +/// a gap in the historical data that gets filled while streaming. +/// +/// Note: This is a simplified version that avoids broadcast timing issues +/// by publishing the gap before the subscriber gets too far ahead. +#[tokio::test] +#[serial] +async fn test_subscribe_across_gap() -> Result<()> { + let ctx = Arc::new(TestContext::new().await?); + + // Publish initial blocks 0-5 + publish_blocks(&ctx, 0, 5).await?; + + // Start subscriber requesting 0-20 + let mut sub_client = ctx.subscriber_client().await?; + let request = SubscribeStreamRequest { + start_block_number: 0, + end_block_number: 20, + }; + + let mut stream = sub_client + .subscribe_block_stream(request) + .await? + .into_inner(); + + // Receive first 6 blocks (0-5) + let mut received = Vec::new(); + for _ in 0..6 { + if let Some(Ok(msg)) = stream.next().await { + if let Some(SubResponse::BlockItems(set)) = msg.response { + received.push(header_number(&set)); + } + } + } + assert_eq!(received, vec![0, 1, 2, 3, 4, 5], "Should receive 0-5"); + + // Now publish the next blocks sequentially (simulating continuous publishing) + // This tests that the subscriber can keep up with live blocks + let ctx_clone = Arc::clone(&ctx); + tokio::spawn(async move { + for i in 6..=20 { + tokio::time::sleep(tokio::time::Duration::from_millis(100)).await; + publish_blocks(&ctx_clone, i, i).await.unwrap(); + } + }); + + // Receive remaining blocks (6-20) + for expected in 6..=20 { + if let Some(Ok(msg)) = + tokio::time::timeout(tokio::time::Duration::from_secs(5), stream.next()).await? + { + if let Some(SubResponse::BlockItems(set)) = msg.response { + let block_num = header_number(&set); + assert_eq!( + block_num, expected, + "Expected block {} but got {}", + expected, block_num + ); + } + } + } + + Ok(()) +} + +/// Test Case: High concurrency with many subscribers. +/// Tests that the subscriber service can handle many concurrent subscribers +/// (50+) without resource exhaustion, deadlocks, or errors. +#[tokio::test] +#[serial] +async fn test_high_concurrency_many_subscribers() -> Result<()> { + let ctx = Arc::new(TestContext::new().await?); + + // Publish sufficient blocks to cover all subscriber ranges + // Subscriber 49: start=98, end=108, so we need at least 0-108 + publish_blocks(&ctx, 0, 120).await?; + + // Spawn 50 concurrent subscribers with different ranges + let mut tasks = Vec::new(); + for i in 0..50 { + let ctx_clone = Arc::clone(&ctx); + let start = i * 2; + let end = start + 10; + + tasks.push(tokio::spawn(async move { + let mut client = ctx_clone.subscriber_client().await.unwrap(); + let request = SubscribeStreamRequest { + start_block_number: start, + end_block_number: end, + }; + + let mut stream = client + .subscribe_block_stream(request) + .await + .unwrap() + .into_inner(); + + let mut count = 0; + while let Some(Ok(msg)) = stream.next().await { + if let Some(SubResponse::BlockItems(_)) = msg.response { + count += 1; + } + if let Some(SubResponse::Status(code)) = msg.response { + assert_eq!(code, SubCode::Success as i32); + break; + } + } + count + })); + } + + // All should complete successfully + for (i, task) in tasks.into_iter().enumerate() { + let count = task.await?; + assert!( + count > 0, + "Subscriber {} received no blocks (expected 11)", + i + ); + assert_eq!( + count, 11, + "Subscriber {} should receive exactly 11 blocks", + i + ); + } + + Ok(()) +} + +/// Test Case: Rapid block publication with live subscriber. +/// Tests that a live subscriber can keep up with rapidly published blocks +/// without data loss or errors. +#[tokio::test] +#[serial] +async fn test_rapid_block_publication() -> Result<()> { + let ctx = Arc::new(TestContext::new().await?); + + // Start live subscriber + let mut sub_client = ctx.subscriber_client().await?; + let request = SubscribeStreamRequest { + start_block_number: 0, + end_block_number: 50, + }; + + let mut stream = sub_client + .subscribe_block_stream(request) + .await? + .into_inner(); + + // Publish blocks rapidly in background + let ctx_clone = Arc::clone(&ctx); + let publish_task = tokio::spawn(async move { + let publish_port = ctx_clone.container.get_host_port_ipv4(50051).await.unwrap(); + let publish_endpoint = format!("http://localhost:{}", publish_port); + let channel = Channel::from_shared(publish_endpoint) + .unwrap() + .connect() + .await + .unwrap(); + let mut client = BlockStreamPublishServiceClient::new(channel); + + for i in 0..=50 { + let block_bytes = crate::common::block_builder::BlockBuilder::new(i).build(); + let block_proto: rock_node_protobufs::com::hedera::hapi::block::stream::Block = + Message::decode(block_bytes.as_slice()).unwrap(); + + let (tx, rx) = mpsc::channel(1); + tx.send(PublishStreamRequest { + request: Some(PublishRequest::BlockItems(BlockItemSet { + block_items: block_proto.items, + })), + }) + .await + .unwrap(); + drop(tx); + + let mut responses = client + .publish_block_stream(ReceiverStream::new(rx)) + .await + .unwrap() + .into_inner(); + + while responses.next().await.is_some() {} + + // Small delay to simulate rapid but not instant + tokio::time::sleep(tokio::time::Duration::from_millis(10)).await; + } + }); + + // Subscriber should receive all blocks + let mut received_count = 0; + while let Some(Ok(msg)) = stream.next().await { + match msg.response { + Some(SubResponse::BlockItems(_)) => { + received_count += 1; + }, + Some(SubResponse::Status(code)) => { + assert_eq!(code, SubCode::Success as i32); + break; + }, + _ => {}, + } + } + + publish_task.await?; + + assert_eq!( + received_count, 51, + "Should receive all 51 blocks (0-50) despite rapid publication" + ); + Ok(()) +} + +/// Test Case: Subscribers with overlapping ranges. +/// Tests that multiple subscribers with overlapping block ranges +/// can operate independently without interfering with each other. +#[tokio::test] +#[serial] +async fn test_overlapping_subscriber_ranges() -> Result<()> { + let ctx = Arc::new(TestContext::new().await?); + + publish_blocks(&ctx, 0, 30).await?; + + // Subscriber A: 0-20 + let ctx_a = Arc::clone(&ctx); + let task_a = tokio::spawn(async move { + let mut client = ctx_a.subscriber_client().await.unwrap(); + let request = SubscribeStreamRequest { + start_block_number: 0, + end_block_number: 20, + }; + let mut stream = client + .subscribe_block_stream(request) + .await + .unwrap() + .into_inner(); + + let mut received = Vec::new(); + while let Some(Ok(msg)) = stream.next().await { + match msg.response { + Some(SubResponse::BlockItems(ref set)) => { + received.push(header_number(set)); + }, + Some(SubResponse::Status(_)) => break, + _ => {}, + } + } + received + }); + + // Subscriber B: 10-30 (overlaps with A on 10-20) + let ctx_b = Arc::clone(&ctx); + let task_b = tokio::spawn(async move { + let mut client = ctx_b.subscriber_client().await.unwrap(); + let request = SubscribeStreamRequest { + start_block_number: 10, + end_block_number: 30, + }; + let mut stream = client + .subscribe_block_stream(request) + .await + .unwrap() + .into_inner(); + + let mut received = Vec::new(); + while let Some(Ok(msg)) = stream.next().await { + match msg.response { + Some(SubResponse::BlockItems(ref set)) => { + received.push(header_number(set)); + }, + Some(SubResponse::Status(_)) => break, + _ => {}, + } + } + received + }); + + let received_a = task_a.await?; + let received_b = task_b.await?; + + assert_eq!(received_a, (0..=20).collect::>()); + assert_eq!(received_b, (10..=30).collect::>()); + + Ok(()) +} + +/// Test Case: Subscribe to single block (start == end). +/// Tests edge case where subscriber requests exactly one block. +#[tokio::test] +#[serial] +async fn test_subscribe_single_block() -> Result<()> { + let ctx = TestContext::new().await?; + + publish_blocks(&ctx, 0, 10).await?; + + let mut sub_client = ctx.subscriber_client().await?; + let request = SubscribeStreamRequest { + start_block_number: 5, + end_block_number: 5, // Single block + }; + + let mut stream = sub_client + .subscribe_block_stream(request) + .await? + .into_inner(); + + let mut received = Vec::new(); + while let Some(Ok(msg)) = stream.next().await { + match msg.response { + Some(SubResponse::BlockItems(set)) => { + received.push(header_number(&set)); + }, + Some(SubResponse::Status(code)) => { + assert_eq!(code, SubCode::Success as i32); + break; + }, + _ => {}, + } + } + + assert_eq!(received, vec![5], "Should receive exactly block 5"); + Ok(()) +} diff --git a/tests/e2e/src/tests/subscribe/failure_scenarios.rs b/tests/e2e/src/tests/subscribe/failure_scenarios.rs new file mode 100644 index 0000000..eec85a3 --- /dev/null +++ b/tests/e2e/src/tests/subscribe/failure_scenarios.rs @@ -0,0 +1,177 @@ +use crate::common::{publish_blocks, TestContext}; +use anyhow::Result; +use rock_node_protobufs::org::hiero::block::api::{ + subscribe_stream_response::{Code as SubCode, Response as SubResponse}, + SubscribeStreamRequest, +}; +use serial_test::serial; +use tokio_stream::StreamExt; + +/// Test Case: Client disconnect mid-stream. +/// Tests that when a client disconnects during an active subscription, +/// the session is cleaned up gracefully and metrics are updated correctly. +#[tokio::test] +#[serial] +async fn test_client_disconnect_tracking() -> Result<()> { + let ctx = TestContext::new().await?; + + // Publish many blocks + publish_blocks(&ctx, 0, 100).await?; + + let mut sub_client = ctx.subscriber_client().await?; + let request = SubscribeStreamRequest { + start_block_number: 0, + end_block_number: 50, + }; + + let mut stream = sub_client + .subscribe_block_stream(request) + .await? + .into_inner(); + + // Receive a few blocks then disconnect + for _ in 0..10 { + stream.next().await; + } + + // Force disconnect by dropping the stream and client + drop(stream); + drop(sub_client); + + // Wait a bit for cleanup + tokio::time::sleep(tokio::time::Duration::from_secs(1)).await; + + // Verify session was cleaned up + // In a real test, we'd check metrics or logs here + // For now, this validates that the disconnect doesn't cause a panic + + Ok(()) +} + +/// Test Case: Server shutdown during active subscription. +/// Tests that when the server initiates shutdown, active subscribers +/// receive a graceful Error status and the stream terminates cleanly. +/// +/// Note: This test is challenging to implement in E2E as it requires +/// shutting down the Docker container mid-test. Marked as placeholder. +#[tokio::test] +#[serial] +#[ignore] // Requires infrastructure to gracefully shutdown server mid-test +async fn test_server_shutdown_during_subscription() -> Result<()> { + // TODO: Implement when we have server shutdown capabilities in TestContext + // Steps: + // 1. Start subscription with infinite range + // 2. Receive a few blocks + // 3. Trigger server shutdown via TestContext.shutdown_server() + // 4. Verify subscriber receives Error status + // 5. Verify metrics show server_shutdown + Ok(()) +} + +/// Test Case: Timeout waiting for block. +/// Tests that when a subscriber is waiting for a block that never arrives, +/// the session timeout triggers and the stream ends with an error status. +/// +/// Note: This requires configuring a short session_timeout_seconds. +/// Currently challenging to test as TestContext uses default config. +#[tokio::test] +#[serial] +#[ignore] // Requires custom config with short timeout +async fn test_timeout_waiting_for_block() -> Result<()> { + // TODO: Implement when TestContext supports custom config + // Steps: + // 1. Create TestContext with session_timeout_seconds = 2 + // 2. Publish blocks 0-5 + // 3. Subscribe requesting blocks 0-20 + // 4. Receive blocks 0-5 + // 5. Wait for block 6 (which never arrives) + // 6. After 2 seconds, should timeout + // 7. Verify Error status received + // 8. Verify metrics show timeout label + Ok(()) +} + +/// Test Case: Multiple subscribers, one disconnects. +/// Tests that when multiple subscribers are active and one disconnects, +/// the others continue operating normally without disruption. +#[tokio::test] +#[serial] +async fn test_multiple_subscribers_one_disconnects() -> Result<()> { + let ctx = TestContext::new().await?; + + publish_blocks(&ctx, 0, 50).await?; + + // Subscriber A: Will disconnect early + let mut client_a = ctx.subscriber_client().await?; + let request_a = SubscribeStreamRequest { + start_block_number: 0, + end_block_number: u64::MAX, + }; + let mut stream_a = client_a + .subscribe_block_stream(request_a) + .await? + .into_inner(); + + // Subscriber B: Will continue to completion + let mut client_b = ctx.subscriber_client().await?; + let request_b = SubscribeStreamRequest { + start_block_number: 0, + end_block_number: 30, + }; + let mut stream_b = client_b + .subscribe_block_stream(request_b) + .await? + .into_inner(); + + // Both receive a few blocks + for _ in 0..5 { + stream_a.next().await; + stream_b.next().await; + } + + // A disconnects + drop(stream_a); + drop(client_a); + + // B should continue without issues + let mut received_b = 5; // Already received 5 + while let Some(Ok(msg)) = stream_b.next().await { + match msg.response { + Some(SubResponse::BlockItems(_)) => { + received_b += 1; + }, + Some(SubResponse::Status(code)) => { + assert_eq!(code, SubCode::Success as i32); + break; + }, + _ => {}, + } + } + + assert_eq!( + received_b, 31, + "Subscriber B should receive all 31 blocks (0-30) despite A's disconnect" + ); + Ok(()) +} + +/// Test Case: Lag detection on slow subscriber. +/// Tests that when a subscriber processes blocks slowly and the broadcast +/// channel lags, the system handles it gracefully by falling back to +/// persistence reads. +/// +/// Note: This is difficult to trigger reliably in E2E as Rust's broadcast +/// channel handles lag automatically. Included as placeholder for awareness. +#[tokio::test] +#[serial] +#[ignore] // Difficult to trigger lag reliably in E2E +async fn test_subscriber_lag_handling() -> Result<()> { + // TODO: Consider implementing with artificial delays + // Steps: + // 1. Create subscriber + // 2. Publish blocks very rapidly (faster than subscriber processes) + // 3. Subscriber should still receive all blocks (may switch to persistence reads) + // 4. Verify no data loss + // 5. Verify logs show lag detection + Ok(()) +} diff --git a/tests/e2e/src/tests/subscribe/mod.rs b/tests/e2e/src/tests/subscribe/mod.rs new file mode 100644 index 0000000..eea00b7 --- /dev/null +++ b/tests/e2e/src/tests/subscribe/mod.rs @@ -0,0 +1,4 @@ +pub mod basic; +pub mod edge_cases; +pub mod failure_scenarios; +pub mod validation; diff --git a/tests/e2e/src/tests/subscribe/validation.rs b/tests/e2e/src/tests/subscribe/validation.rs new file mode 100644 index 0000000..fc7037f --- /dev/null +++ b/tests/e2e/src/tests/subscribe/validation.rs @@ -0,0 +1,385 @@ +use crate::common::{publish_blocks, TestContext}; +use anyhow::Result; +use rock_node_protobufs::org::hiero::block::api::{ + subscribe_stream_response::{Code as SubCode, Response as SubResponse}, + SubscribeStreamRequest, +}; +use serial_test::serial; +use std::sync::Arc; +use tokio_stream::StreamExt; + +/// Helper to extract the block number from a `BlockItems` response. +fn header_number(item_set: &rock_node_protobufs::org::hiero::block::api::BlockItemSet) -> u64 { + use rock_node_protobufs::com::hedera::hapi::block::stream::block_item::Item; + item_set + .block_items + .iter() + .find_map(|item| match &item.item { + Some(Item::BlockHeader(h)) => Some(h.number), + _ => None, + }) + .expect("Block items must contain a header") +} + +/// Test Case: Validation error when start > end. +/// Tests that the new validation logic correctly rejects requests where +/// the start block is after the end block. +#[tokio::test] +#[serial] +async fn test_invalid_end_block_number() -> Result<()> { + let ctx = TestContext::new().await?; + + let mut sub_client = ctx.subscriber_client().await?; + let request = SubscribeStreamRequest { + start_block_number: 10, + end_block_number: 5, + }; + + let mut stream = sub_client + .subscribe_block_stream(request) + .await? + .into_inner(); + + let first = stream.next().await.expect("Expected a response")?; + match first.response { + Some(SubResponse::Status(code)) => { + assert_eq!( + code, + SubCode::InvalidEndBlockNumber as i32, + "Expected InvalidEndBlockNumber error" + ); + }, + other => panic!("Unexpected response: {:?}", other), + } + assert!( + stream.next().await.is_none(), + "Stream should have terminated" + ); + Ok(()) +} + +/// Test Case: Start block too far in the future. +/// Tests that the validation logic rejects requests where the start block +/// exceeds (latest + max_future_block_lookahead). +#[tokio::test] +#[serial] +async fn test_validation_start_too_far_in_future() -> Result<()> { + // Custom config with explicit lookahead to make test deterministic + let custom_config = r#" +[core] +log_level = "info" +database_path = "/app/data" +start_block_number = 0 +grpc_address = "0.0.0.0" +grpc_port = 50051 + +[plugins] + [plugins.observability] + enabled = true + listen_address = "0.0.0.0:8080" + + [plugins.persistence_service] + enabled = true + cold_storage_path = "/app/data/cold" + hot_storage_block_count = 10000 + archive_batch_size = 1000 + + [plugins.verification_service] + enabled = true + + [plugins.state_management_service] + enabled = true + + [plugins.publish_service] + enabled = true + max_concurrent_streams = 10 + + [plugins.subscriber_service] + enabled = true + max_concurrent_streams = 10 + max_future_block_lookahead = 250 + session_timeout_seconds = 60 + + [plugins.block_access_service] + enabled = true + + [plugins.server_status_service] + enabled = true + + [plugins.query_service] + enabled = true +"#; + + let ctx = TestContext::with_config(Some(custom_config), None).await?; + + // Publish blocks 0-10 (latest = 10) + publish_blocks(&ctx, 0, 10).await?; + + let mut sub_client = ctx.subscriber_client().await?; + + // Try to subscribe starting at block 500 + // max_future_block_lookahead = 250 (from custom config) + // Max permitted start = 10 + 250 = 260 + // 500 > 260, so should be rejected + let request = SubscribeStreamRequest { + start_block_number: 500, + end_block_number: u64::MAX, + }; + + let mut stream = sub_client + .subscribe_block_stream(request) + .await? + .into_inner(); + + // Should get NotAvailable error status + let response = tokio::time::timeout(tokio::time::Duration::from_secs(5), stream.next()) + .await? + .expect("Expected response")?; + + match response.response { + Some(SubResponse::Status(code)) => { + assert_eq!( + code, + SubCode::NotAvailable as i32, + "Expected NotAvailable for start beyond lookahead" + ); + }, + Some(SubResponse::BlockItems(_)) => { + panic!("Should not receive blocks when start (500) is beyond lookahead (260)"); + }, + other => panic!("Expected NotAvailable status, got {:?}", other), + } + + // Stream should terminate + assert!(stream.next().await.is_none()); + Ok(()) +} + +/// Test Case: Pure live streaming (start=MAX, end=MAX) from empty node. +/// Tests that pure live mode is accepted even when no blocks exist yet, +/// and that the subscriber receives blocks as they arrive. +#[tokio::test] +#[serial] +async fn test_pure_live_streaming() -> Result<()> { + let ctx = TestContext::new().await?; + + // Start with NO blocks published + let mut sub_client = ctx.subscriber_client().await?; + let request = SubscribeStreamRequest { + start_block_number: u64::MAX, + end_block_number: u64::MAX, + }; + + let mut stream = sub_client + .subscribe_block_stream(request) + .await? + .into_inner(); + + // Spawn task to publish blocks after subscription starts + let ctx_arc = Arc::new(ctx); + let ctx_clone = Arc::clone(&ctx_arc); + tokio::spawn(async move { + tokio::time::sleep(tokio::time::Duration::from_millis(200)).await; + publish_blocks(&ctx_clone, 0, 4).await.unwrap(); + }); + + // Should receive blocks as they arrive + let mut received = Vec::new(); + for _ in 0..5 { + if let Some(Ok(msg)) = + tokio::time::timeout(tokio::time::Duration::from_secs(5), stream.next()).await? + { + if let Some(SubResponse::BlockItems(set)) = msg.response { + received.push(header_number(&set)); + } + } + } + + assert_eq!( + received, + vec![0, 1, 2, 3, 4], + "Should receive blocks 0-4 in order" + ); + Ok(()) +} + +/// Test Case: From earliest to specific end (start=MAX, end=N). +/// Tests that start=MAX resolves to the earliest available block, +/// and that finite range works correctly. +#[tokio::test] +#[serial] +async fn test_from_earliest_to_end() -> Result<()> { + let ctx = TestContext::new().await?; + + // Publish blocks 10-20 (earliest = 10) + publish_blocks(&ctx, 10, 20).await?; + + let mut sub_client = ctx.subscriber_client().await?; + let request = SubscribeStreamRequest { + start_block_number: u64::MAX, // Should resolve to earliest (10) + end_block_number: 15, + }; + + let mut stream = sub_client + .subscribe_block_stream(request) + .await? + .into_inner(); + + let mut received = Vec::new(); + while let Some(Ok(msg)) = stream.next().await { + match msg.response { + Some(SubResponse::BlockItems(set)) => { + received.push(header_number(&set)); + }, + Some(SubResponse::Status(code)) => { + assert_eq!(code, SubCode::Success as i32); + break; + }, + _ => {}, + } + } + + // Should receive blocks 10-15 (earliest to requested end) + assert_eq!( + received, + (10..=15).collect::>(), + "Should receive blocks from earliest (10) to end (15)" + ); + Ok(()) +} + +/// Test Case: Request start before earliest available block. +/// Tests that the validation logic correctly rejects requests where +/// the start block is before the earliest available block. +#[tokio::test] +#[serial] +async fn test_validation_start_before_earliest() -> Result<()> { + let ctx = TestContext::new().await?; + + // Publish blocks 50-60 (earliest = 50) + publish_blocks(&ctx, 50, 60).await?; + + let mut sub_client = ctx.subscriber_client().await?; + + // Try to subscribe from block 10 (before earliest) + let request = SubscribeStreamRequest { + start_block_number: 10, + end_block_number: u64::MAX, + }; + + let mut stream = sub_client + .subscribe_block_stream(request) + .await? + .into_inner(); + + let response = stream.next().await.expect("Expected response")?; + match response.response { + Some(SubResponse::Status(code)) => { + assert_eq!( + code, + SubCode::InvalidStartBlockNumber as i32, + "Expected InvalidStartBlockNumber for start before earliest" + ); + }, + other => panic!("Expected InvalidStartBlockNumber, got {:?}", other), + } + + // Stream should terminate + assert!(stream.next().await.is_none()); + Ok(()) +} + +/// Test Case: Future start with no blocks available. +/// Tests that a future start block is accepted when no blocks exist yet +/// (within lookahead), and the subscriber waits for blocks to arrive. +#[tokio::test] +#[serial] +async fn test_future_start_no_blocks_available() -> Result<()> { + let ctx = TestContext::new().await?; + + // NO blocks published yet + let mut sub_client = ctx.subscriber_client().await?; + let request = SubscribeStreamRequest { + start_block_number: 100, // Future block (within lookahead from 0) + end_block_number: u64::MAX, + }; + + let mut stream = sub_client + .subscribe_block_stream(request) + .await? + .into_inner(); + + // Spawn task to publish blocks starting at 100 + let ctx_arc = Arc::new(ctx); + let ctx_clone = Arc::clone(&ctx_arc); + tokio::spawn(async move { + tokio::time::sleep(tokio::time::Duration::from_millis(200)).await; + publish_blocks(&ctx_clone, 100, 104).await.unwrap(); + }); + + // Should receive blocks starting at 100 + let mut received = Vec::new(); + for _ in 0..5 { + if let Some(Ok(msg)) = + tokio::time::timeout(tokio::time::Duration::from_secs(5), stream.next()).await? + { + if let Some(SubResponse::BlockItems(set)) = msg.response { + received.push(header_number(&set)); + } + } + } + + assert_eq!( + received, + vec![100, 101, 102, 103, 104], + "Should receive blocks starting at future block 100" + ); + Ok(()) +} + +/// Test Case: From earliest but end is before earliest. +/// Tests that when requesting start=MAX with an end block before the earliest +/// available, the request is either rejected or completes with no blocks. +#[tokio::test] +#[serial] +async fn test_validation_from_earliest_but_end_before_earliest() -> Result<()> { + let ctx = TestContext::new().await?; + + // Publish blocks 10-20 (earliest = 10) + publish_blocks(&ctx, 10, 20).await?; + + let mut sub_client = ctx.subscriber_client().await?; + + // Request from earliest (10) to 5 (before earliest) + let request = SubscribeStreamRequest { + start_block_number: u64::MAX, // Will resolve to 10 + end_block_number: 5, // Before earliest + }; + + let mut stream = sub_client + .subscribe_block_stream(request) + .await? + .into_inner(); + + // Should receive an error status (could be NotAvailable or other error) + // The important thing is that it doesn't successfully stream blocks + let response = stream.next().await.expect("Expected response")?; + match response.response { + Some(SubResponse::Status(code)) => { + // Accept any non-success error code + assert_ne!( + code, + SubCode::Success as i32, + "Should not succeed when end < earliest" + ); + }, + Some(SubResponse::BlockItems(_)) => { + panic!("Should not receive blocks when end (5) < earliest (10)"); + }, + other => panic!("Unexpected response: {:?}", other), + } + + // Stream should terminate + assert!(stream.next().await.is_none()); + Ok(()) +} From 21d25b06f4037becbf3f894b4fe013898e0a33b6 Mon Sep 17 00:00:00 2001 From: georgi-l95 Date: Fri, 10 Oct 2025 10:14:13 +0300 Subject: [PATCH 5/6] minor improvements Signed-off-by: georgi-l95 --- .github/workflows/release.yml | 150 ++++++++++++++++++ config/config.toml | 14 +- .../rock-node-backfill-plugin/src/worker.rs | 2 +- .../src/service.rs | 5 +- crates/rock-node-publish-plugin/src/error.rs | 6 + .../src/session_manager.rs | 1 - .../src/session.rs | 1 + 7 files changed, 168 insertions(+), 11 deletions(-) create mode 100644 .github/workflows/release.yml diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml new file mode 100644 index 0000000..ac72ed6 --- /dev/null +++ b/.github/workflows/release.yml @@ -0,0 +1,150 @@ +name: Release + +on: + release: + types: [published] + +env: + REGISTRY: ghcr.io + IMAGE_NAME: ${{ github.repository }} + +jobs: + build-and-publish: + name: Build and Publish Release + runs-on: ubuntu-latest + permissions: + contents: write + packages: write + + steps: + - name: Checkout repository + uses: actions/checkout@v4 + with: + ref: ${{ github.event.release.tag_name }} + + - name: Set up QEMU for multi-arch builds + uses: docker/setup-qemu-action@v3 + + - name: Set up Docker Buildx + uses: docker/setup-buildx-action@v3 + + - name: Log in to GitHub Container Registry + uses: docker/login-action@v3 + with: + registry: ${{ env.REGISTRY }} + username: ${{ github.actor }} + password: ${{ secrets.GITHUB_TOKEN }} + + - name: Extract metadata for Docker + id: meta + uses: docker/metadata-action@v5 + with: + images: ${{ env.REGISTRY }}/${{ env.IMAGE_NAME }} + tags: | + type=semver,pattern={{version}},value=${{ github.event.release.tag_name }} + type=semver,pattern={{major}}.{{minor}},value=${{ github.event.release.tag_name }} + type=semver,pattern={{major}},value=${{ github.event.release.tag_name }} + type=raw,value=latest,enable=${{ !github.event.release.prerelease }} + + - name: Build and push Docker image + uses: docker/build-push-action@v5 + with: + context: . + platforms: linux/amd64,linux/arm64 + push: true + tags: ${{ steps.meta.outputs.tags }} + labels: ${{ steps.meta.outputs.labels }} + cache-from: type=gha + cache-to: type=gha,mode=max + + - name: Install Rust toolchain + uses: dtolnay/rust-toolchain@stable + + - name: Install cross-compilation tools + run: | + cargo install cross --git https://github.com/cross-rs/cross + + - name: Install dependencies (protoc) + run: | + sudo apt-get update -y + sudo apt-get install -y protobuf-compiler + + - name: Build Linux x86_64 binary + run: cross build --release --target x86_64-unknown-linux-gnu --locked + working-directory: app/rock-node + + - name: Build Linux ARM64 binary + run: cross build --release --target aarch64-unknown-linux-gnu --locked + working-directory: app/rock-node + + - name: Package binaries + run: | + mkdir -p release-artifacts + + # Package x86_64 + cd target/x86_64-unknown-linux-gnu/release + tar -czf ../../../release-artifacts/rock-node-${{ github.event.release.tag_name }}-linux-x86_64.tar.gz rock-node + cd - + + # Package ARM64 + cd target/aarch64-unknown-linux-gnu/release + tar -czf ../../../release-artifacts/rock-node-${{ github.event.release.tag_name }}-linux-arm64.tar.gz rock-node + cd - + + # Generate checksums + cd release-artifacts + sha256sum * > checksums.txt + cd - + + - name: Upload Release Assets + uses: softprops/action-gh-release@v1 + with: + tag_name: ${{ github.event.release.tag_name }} + files: | + release-artifacts/rock-node-${{ github.event.release.tag_name }}-linux-x86_64.tar.gz + release-artifacts/rock-node-${{ github.event.release.tag_name }}-linux-arm64.tar.gz + release-artifacts/checksums.txt + append_body: true + body: | + + --- + + ## 🐳 Docker Images + + Multi-architecture images published to GitHub Container Registry: + + ```bash + # Pull specific version + docker pull ${{ env.REGISTRY }}/${{ env.IMAGE_NAME }}:${{ github.event.release.tag_name }} + + # Pull latest + docker pull ${{ env.REGISTRY }}/${{ env.IMAGE_NAME }}:latest + ``` + + **Platforms**: `linux/amd64`, `linux/arm64` + + ## 📦 Standalone Binaries + + Download and verify checksums: + + ```bash + # Set version + VERSION=${{ github.event.release.tag_name }} + + # Download for your platform + wget https://github.com/${{ github.repository }}/releases/download/${VERSION}/rock-node-${VERSION}-linux-x86_64.tar.gz + + # Download checksums + wget https://github.com/${{ github.repository }}/releases/download/${VERSION}/checksums.txt + + # Verify integrity + sha256sum -c checksums.txt --ignore-missing + + # Extract and run + tar -xzf rock-node-${VERSION}-linux-x86_64.tar.gz + ./rock-node --config-path config/config.toml + ``` + + **Available Binaries**: + - `rock-node-${{ github.event.release.tag_name }}-linux-x86_64.tar.gz` (Intel/AMD 64-bit) + - `rock-node-${{ github.event.release.tag_name }}-linux-arm64.tar.gz` (ARM 64-bit) diff --git a/config/config.toml b/config/config.toml index 8e63dcd..ab37528 100644 --- a/config/config.toml +++ b/config/config.toml @@ -5,7 +5,7 @@ log_level = "INFO" database_path = "data/database" start_block_number = 0 -grpc_address = "127.0.0.1" +grpc_address = "0.0.0.0" grpc_port = 8090 # ========================================= @@ -15,7 +15,7 @@ grpc_port = 8090 [plugins.observability] enabled = true - listen_address = "127.0.0.1:9600" + listen_address = "0.0.0.0:9600" [plugins.publish_service] enabled = true @@ -32,7 +32,7 @@ grpc_port = 8090 archive_batch_size = 100 [plugins.state_management_service] - enabled = true + enabled = false [plugins.subscriber_service] enabled = true @@ -42,7 +42,7 @@ grpc_port = 8090 session_timeout_seconds = 60 [plugins.verification_service] - enabled = false + enabled = true [plugins.block_access_service] enabled = true @@ -51,11 +51,11 @@ grpc_port = 8090 enabled = true [plugins.query_service] - enabled = true + enabled = false [plugins.backfill] - enabled = true + enabled = false mode = "GapFill" - peers = ["http://127.0.0.1:8090"] + peers = ["http://0.0.0.0:8090"] check_interval_seconds = 5 max_batch_size = 10 diff --git a/crates/rock-node-backfill-plugin/src/worker.rs b/crates/rock-node-backfill-plugin/src/worker.rs index 704c142..a62877f 100644 --- a/crates/rock-node-backfill-plugin/src/worker.rs +++ b/crates/rock-node-backfill-plugin/src/worker.rs @@ -259,7 +259,7 @@ impl BackfillWorker { } } - async fn run_continuous_stream_cycle(self: &Arc, shuffled_peers: &mut Vec) { + async fn run_continuous_stream_cycle(self: &Arc, shuffled_peers: &mut [String]) { shuffled_peers.shuffle(&mut rand::rng()); for peer_addr in shuffled_peers.iter() { diff --git a/crates/rock-node-block-access-plugin/src/service.rs b/crates/rock-node-block-access-plugin/src/service.rs index 3365995..60edf73 100644 --- a/crates/rock-node-block-access-plugin/src/service.rs +++ b/crates/rock-node-block-access-plugin/src/service.rs @@ -24,7 +24,7 @@ fn code_to_string(code: block_response::Code) -> &'static str { block_response::Code::NotFound => "NotFound", block_response::Code::NotAvailable => "NotAvailable", block_response::Code::Unknown => "Unknown", - block_response::Code::Error => todo!(), + block_response::Code::Error => "Error", } } @@ -207,6 +207,7 @@ impl BlockAccessServiceImpl { } } + #[allow(clippy::result_large_err)] fn record_metrics( &self, response: BlockResponse, @@ -572,6 +573,6 @@ mod tests { "NotAvailable" ); assert_eq!(code_to_string(block_response::Code::Unknown), "Unknown"); - // Note: Error case will panic in debug mode, so we don't test it here + assert_eq!(code_to_string(block_response::Code::Error), "Error"); } } diff --git a/crates/rock-node-publish-plugin/src/error.rs b/crates/rock-node-publish-plugin/src/error.rs index 7b18fbd..56f33e5 100644 --- a/crates/rock-node-publish-plugin/src/error.rs +++ b/crates/rock-node-publish-plugin/src/error.rs @@ -12,24 +12,30 @@ pub enum PublishError { #[error("Too many items in set: {count} (max: {max})")] TooManyItems { count: usize, max: usize }, + #[allow(dead_code)] #[error("Block number mismatch: expected {expected}, got {received}")] BlockNumberMismatch { expected: i64, received: i64 }, + #[allow(dead_code)] #[error("Invalid block number: {block_number}")] InvalidBlockNumber { block_number: i64 }, + #[allow(dead_code)] #[error("Session {session_id} received items without active block")] NoActiveBlock { session_id: uuid::Uuid }, #[error("Failed to encode block: {0}")] Encoding(#[from] prost::EncodeError), + #[allow(dead_code)] #[error("Channel send error")] ChannelSend, + #[allow(dead_code)] #[error("Verification timeout for block {block_number}")] VerificationTimeout { block_number: u64 }, + #[allow(dead_code)] #[error("Verification failed for block {block_number}: {reason}")] VerificationFailed { block_number: u64, reason: String }, } diff --git a/crates/rock-node-publish-plugin/src/session_manager.rs b/crates/rock-node-publish-plugin/src/session_manager.rs index 34022da..384ad4f 100644 --- a/crates/rock-node-publish-plugin/src/session_manager.rs +++ b/crates/rock-node-publish-plugin/src/session_manager.rs @@ -1,5 +1,4 @@ use crate::{ - error::PublishError, state::{BlockAction, SessionState, SharedState}, validation::RequestValidator, }; diff --git a/crates/rock-node-subscriber-plugin/src/session.rs b/crates/rock-node-subscriber-plugin/src/session.rs index fccc7fb..c5aad0f 100644 --- a/crates/rock-node-subscriber-plugin/src/session.rs +++ b/crates/rock-node-subscriber-plugin/src/session.rs @@ -30,6 +30,7 @@ pub struct SubscriberSession { } impl SubscriberSession { + #[allow(clippy::result_large_err)] pub fn new( context: Arc, request: SubscribeStreamRequest, From ba6f5dd13053505f0c3cfbb108e90824ff59382d Mon Sep 17 00:00:00 2001 From: georgi-l95 Date: Fri, 10 Oct 2025 13:51:37 +0300 Subject: [PATCH 6/6] apply coderabbit review comments Signed-off-by: georgi-l95 --- .github/workflows/release.yml | 2 +- config/config.toml | 2 +- crates/rock-node-publish-plugin/src/session_manager.rs | 2 +- 3 files changed, 3 insertions(+), 3 deletions(-) diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml index ac72ed6..70bcae1 100644 --- a/.github/workflows/release.yml +++ b/.github/workflows/release.yml @@ -97,7 +97,7 @@ jobs: cd - - name: Upload Release Assets - uses: softprops/action-gh-release@v1 + uses: softprops/action-gh-release@v2 with: tag_name: ${{ github.event.release.tag_name }} files: | diff --git a/config/config.toml b/config/config.toml index ab37528..b99a1fc 100644 --- a/config/config.toml +++ b/config/config.toml @@ -56,6 +56,6 @@ grpc_port = 8090 [plugins.backfill] enabled = false mode = "GapFill" - peers = ["http://0.0.0.0:8090"] + peers = ["http://127.0.0.1:8090"] check_interval_seconds = 5 max_batch_size = 10 diff --git a/crates/rock-node-publish-plugin/src/session_manager.rs b/crates/rock-node-publish-plugin/src/session_manager.rs index 384ad4f..0ae86ad 100644 --- a/crates/rock-node-publish-plugin/src/session_manager.rs +++ b/crates/rock-node-publish-plugin/src/session_manager.rs @@ -126,7 +126,7 @@ impl SessionManager { response: Some(publish_stream_response::Response::EndStream( publish_stream_response::EndOfStream { status: publish_stream_response::end_of_stream::Code::Error as i32, - block_number: self.current_block_number as u64, + block_number: block_number.unwrap_or(self.current_block_number) as u64, }, )), };