diff --git a/crates/apollo_integration_tests/src/flow_test_setup.rs b/crates/apollo_integration_tests/src/flow_test_setup.rs index 9411bcda0da..344861f3f3d 100644 --- a/crates/apollo_integration_tests/src/flow_test_setup.rs +++ b/crates/apollo_integration_tests/src/flow_test_setup.rs @@ -73,6 +73,7 @@ use crate::utils::{ spawn_local_success_recorder, AccumulatedTransactions, NodeDescriptor, + RecorderStats, }; const BUILDER_BASE_ADDRESS: Felt = Felt::from_hex_unchecked("0x42"); @@ -245,6 +246,9 @@ pub struct FlowSequencerSetup { // Monitoring client. pub monitoring_client: MonitoringClient, + // Counters of the cende blobs received by this sequencer's dummy recorder. + pub recorder_stats: Arc, + // Retain clients to avoid closing communication channels, which crashes the server and // subsequently the test. This occurs for components who are wrapped by servers, but no other // component has their client, usually due to these clients being added in a later date. @@ -272,7 +276,7 @@ impl FlowSequencerSetup { let StorageTestSetup { storage_config, storage_handles } = StorageTestSetup::new(accounts, &chain_info, path, PresetTestContracts::new(), None); - let (recorder_url, _join_handle) = + let (recorder_url, _join_handle, recorder_stats) = spawn_local_success_recorder(available_ports.get_next_port()); consensus_manager_config.cende_config.recorder_url = recorder_url; @@ -354,6 +358,7 @@ impl FlowSequencerSetup { storage_handles, node_config, monitoring_client, + recorder_stats, clients, } } diff --git a/crates/apollo_integration_tests/src/integration_test_manager.rs b/crates/apollo_integration_tests/src/integration_test_manager.rs index 14be449e629..0c95261f05b 100644 --- a/crates/apollo_integration_tests/src/integration_test_manager.rs +++ b/crates/apollo_integration_tests/src/integration_test_manager.rs @@ -1356,7 +1356,7 @@ async fn get_sequencer_setup_configs( // TODO(tsabary): Move these to the start of the test and propagate their values when relevant. // All nodes use the same recorder_url and eth_to_strk_oracle_url. - let (recorder_url, _join_handle) = + let (recorder_url, _join_handle, _recorder_stats) = spawn_local_success_recorder(base_layer_ports.get_next_port()); let (eth_to_strk_oracle_url, _join_handle_eth_to_strk_oracle) = spawn_local_eth_to_strk_oracle(base_layer_ports.get_next_port()); diff --git a/crates/apollo_integration_tests/src/utils.rs b/crates/apollo_integration_tests/src/utils.rs index 6b42a1cf44e..4ad00a27813 100644 --- a/crates/apollo_integration_tests/src/utils.rs +++ b/crates/apollo_integration_tests/src/utils.rs @@ -2,7 +2,7 @@ use std::collections::HashMap; use std::future::Future; use std::net::{Ipv4Addr, SocketAddr}; use std::path::PathBuf; -use std::sync::Arc; +use std::sync::{Arc, Mutex}; use std::time::Duration; use apollo_base_layer_tests::anvil_base_layer::AnvilBaseLayer; @@ -94,7 +94,7 @@ use apollo_storage::storage_reader_server::{ StorageReaderServerStaticConfig, }; use apollo_storage::StorageConfig; -use axum::extract::Query; +use axum::extract::{DefaultBodyLimit, Query}; use axum::routing::{get, post}; use axum::{serve, Json, Router}; #[cfg(feature = "cairo_native")] @@ -580,15 +580,86 @@ pub(crate) fn create_consensus_manager_configs_from_network_configs( .collect() } -// Creates a local recorder server that always returns a success status. +/// What a dummy recorder stored from the cende blobs it received. +#[derive(Debug, Default)] +pub struct RecorderStats { + /// The first height whose commitment infos the recorder has not stored yet; `None` while the + /// recorder is empty (the production recorder's height-offset contract). + commitment_infos_height_offset: Mutex>, + /// Heights whose commitment infos did not arrive where the served offset says they should, + /// mapped to the height that was expected instead. + invalid_heights: Mutex>, +} + +impl RecorderStats { + pub fn commitment_infos_height_offset(&self) -> Option { + *self.commitment_infos_height_offset.lock().unwrap() + } + + pub fn invalid_heights(&self) -> HashMap { + self.invalid_heights.lock().unwrap().clone() + } + + /// Stores the heights of a blob's commitment infos, recording every height that did not arrive + /// where the served offset says it should: each blob must start at the offset and ascend + /// contiguously, so every height's witnesses are sent exactly once. + fn store_commitment_infos_heights(&self, heights: &[u64]) { + let Some(first_height) = heights.first() else { return }; + let mut height_offset = self.commitment_infos_height_offset.lock().unwrap(); + // A retried write: the client timed out an attempt the recorder had already stored. + if heights.iter().all(|height| Some(*height) < *height_offset) { + return; + } + // The sender falls back to its whole recent-blocks window while the recorder is empty, so + // the first blob may start at any height; afterwards it starts at the served offset. + let mut expected_height = (*height_offset).unwrap_or(*first_height); + let mut invalid_heights = self.invalid_heights.lock().unwrap(); + for height in heights { + if *height != expected_height { + invalid_heights.insert(*height, format!("expected height {expected_height}")); + } + expected_height = height + 1; + } + // Never regress: a recorder that has stored a height keeps serving past it. + *height_offset = (*height_offset).max(Some(expected_height)); + } +} + +// Creates a local recorder server, discarding what it stored. pub fn spawn_success_recorder(socket_address: SocketAddr) -> JoinHandle<()> { - tokio::spawn(async move { + spawn_success_recorder_with_stats(socket_address).0 +} + +/// Creates a local recorder server that stores the heights of the commitment infos it receives and +/// serves their offset, so the sender ships each height's witnesses once instead of resending its +/// whole recent-blocks window. Heights that do not continue the served offset are recorded as +/// invalid, but still accepted, so the flow reaches the assertion that reports them. +pub fn spawn_success_recorder_with_stats( + socket_address: SocketAddr, +) -> (JoinHandle<()>, Arc) { + let recorder_stats = Arc::new(RecorderStats::default()); + let handler_recorder_stats = recorder_stats.clone(); + let offset_recorder_stats = recorder_stats.clone(); + let join_handle = tokio::spawn(async move { let router = Router::new() .route( RECORDER_WRITE_BLOB_PATH, - post(move || { - async { + post(move |Json(blob): Json| { + let recorder_stats = handler_recorder_stats.clone(); + async move { debug!("Received a request to write a blob."); + // A blob is prepared before its own height's commitment completes, so + // early blobs legitimately carry no commitment infos. + let heights: Vec = blob["recent_state_commitment_infos"] + .as_array() + .map(|state_commitment_infos| { + state_commitment_infos + .iter() + .filter_map(|entry| entry["block_number"].as_u64()) + .collect() + }) + .unwrap_or_default(); + recorder_stats.store_commitment_infos_heights(&heights); StatusCode::OK.to_string() } .instrument(tracing::debug_span!("success recorder write_blob")) @@ -616,28 +687,35 @@ pub fn spawn_success_recorder(socket_address: SocketAddr) -> JoinHandle<()> { ) .route( RECORDER_GET_COMMITMENT_INFOS_HEIGHT_OFFSET_PATH, - get(|| { - async { + get(move || { + let recorder_stats = offset_recorder_stats.clone(); + async move { debug!("Received a request for commitment_infos_height_offset."); - // `null` marks an empty recorder, letting proposals pass the - // retrospective commitment-infos check before any blob is stored. - Json(serde_json::json!({ "block_number": null })) + // `null` while empty: proposals pass the retrospective commitment-infos + // check and the blob sender falls back to its recent-blocks window. + Json(serde_json::json!({ + "block_number": recorder_stats.commitment_infos_height_offset(), + })) } .instrument(tracing::debug_span!( "success recorder commitment_infos_height_offset" )) }), - ); + ) + // Blobs carrying declared classes and execution infos can exceed axum's 2MB default + // body limit; accept any size, as the pre-validation recorder did. + .layer(DefaultBodyLimit::disable()); let listener = TcpListener::bind(socket_address).await.unwrap(); serve(listener, router).await.unwrap(); - }) + }); + (join_handle, recorder_stats) } -pub fn spawn_local_success_recorder(port: u16) -> (Url, JoinHandle<()>) { +pub fn spawn_local_success_recorder(port: u16) -> (Url, JoinHandle<()>, Arc) { let socket_address = SocketAddr::new(Ipv4Addr::LOCALHOST.into(), port); let url = Url::parse(&format!("http://{socket_address}")).unwrap(); - let join_handle = spawn_success_recorder(socket_address); - (url, join_handle) + let (join_handle, recorder_stats) = spawn_success_recorder_with_stats(socket_address); + (url, join_handle, recorder_stats) } /// Fake eth to strk oracle endpoint.