From dac5f2848a41489ba79cdbbff6614c55c8d68f21 Mon Sep 17 00:00:00 2001 From: Stephen Chen <20940639+stephenctw@users.noreply.github.com> Date: Wed, 22 Jul 2026 13:55:43 +0800 Subject: [PATCH 1/4] feat(sequencer): enrich websocket transaction context --- README.md | 6 +- sequencer-core/src/broadcast.rs | 62 +++++++++++---- sequencer/src/egress/api/subscribe.rs | 10 +-- sequencer/src/egress/l2_tx_feed/mod.rs | 50 ++++++------ sequencer/src/egress/l2_tx_feed/tests.rs | 46 ++++++++--- sequencer/src/storage/egress.rs | 98 +++++++++++++++++++++--- sequencer/src/storage/mod.rs | 1 + sequencer/tests/e2e_sequencer.rs | 3 + sequencer/tests/ws_broadcaster.rs | 7 +- tests/harness/src/replay.rs | 12 +-- 10 files changed, 220 insertions(+), 75 deletions(-) diff --git a/README.md b/README.md index 065171f..af40487 100644 --- a/README.md +++ b/README.md @@ -157,16 +157,16 @@ Notes: - messages are JSON text frames. - binary fields are hex-encoded (`0x`-prefixed). - the current runtime enforces a subscriber cap of `64` and a catch-up cap of `50000` events. -- if the requested catch-up window exceeds that cap, the server upgrades and then immediately closes the socket with close code `1008` (`POLICY`) and reason `catch-up window exceeded`. +- if the requested catch-up window exceeds that cap, the server upgrades and then immediately closes the socket with close code `1008` (`POLICY`) and reason `catch-up window exceeded: live_start_offset=`; reconnecting at that offset starts from the current live head. Message shapes: ```json -{ "kind": "user_op", "offset": 10, "sender": "0x...", "fee": 1, "data": "0x..." } +{ "kind": "user_op", "offset": 10, "sender": "0x...", "nonce": 7, "fee": 1, "data": "0x...", "safe_block": 123, "batch_nonce": 4 } ``` ```json -{ "kind": "direct_input", "offset": 11, "sender": "0x...", "block_number": 123, "payload": "0x..." } +{ "kind": "direct_input", "offset": 11, "sender": "0x...", "block_number": 123, "payload": "0x...", "input_index": 42, "batch_nonce": 4 } ``` Success response: diff --git a/sequencer-core/src/broadcast.rs b/sequencer-core/src/broadcast.rs index 5185f4d..cb0ce16 100644 --- a/sequencer-core/src/broadcast.rs +++ b/sequencer-core/src/broadcast.rs @@ -1,24 +1,39 @@ // (c) Cartesi and individual authors (see AUTHORS) // SPDX-License-Identifier: Apache-2.0 (see LICENSE) -use crate::l2_tx::SequencedL2Tx; +use crate::l2_tx::{DirectInput, ValidUserOp}; use serde::{Deserialize, Serialize}; +/// One transaction from the sequencer's replay-then-live feed. +/// +/// In addition to application inputs, each variant carries enough persisted +/// ordering context for a subscriber to rebuild the soft suffix and reconcile +/// it with L1 settlement. #[derive(Debug, Clone, Serialize, Deserialize)] #[serde(tag = "kind", rename_all = "snake_case")] pub enum BroadcastTxMessage { UserOp { offset: u64, sender: String, + /// Signed replay-protection nonce of the user operation. + nonce: u32, /// Log-space fee exponent (base 129/128). See [`crate::fee`]. fee: u16, data: String, + /// Safe L1 block committed by the covering frame. + safe_block: u64, + /// Nonce of the batch containing the covering frame. + batch_nonce: u64, }, DirectInput { offset: u64, sender: String, block_number: u64, payload: String, + /// Per-application InputBox index of this direct input. + input_index: u64, + /// Nonce of the batch that drained this direct input. + batch_nonce: u64, }, } @@ -30,20 +45,37 @@ impl BroadcastTxMessage { } } - pub fn from_offset_and_tx(offset: u64, tx: SequencedL2Tx) -> Self { - match tx { - SequencedL2Tx::UserOp(user_op) => Self::UserOp { - offset, - sender: user_op.sender.to_string(), - fee: user_op.fee, - data: alloy_primitives::hex::encode_prefixed(user_op.data.as_slice()), - }, - SequencedL2Tx::Direct(direct) => Self::DirectInput { - offset, - sender: direct.sender.to_string(), - block_number: direct.block_number, - payload: alloy_primitives::hex::encode_prefixed(direct.payload.as_slice()), - }, + pub fn from_user_op( + offset: u64, + user_op: ValidUserOp, + nonce: u32, + safe_block: u64, + batch_nonce: u64, + ) -> Self { + Self::UserOp { + offset, + sender: user_op.sender.to_string(), + nonce, + fee: user_op.fee, + data: alloy_primitives::hex::encode_prefixed(user_op.data.as_slice()), + safe_block, + batch_nonce, + } + } + + pub fn from_direct_input( + offset: u64, + direct: DirectInput, + input_index: u64, + batch_nonce: u64, + ) -> Self { + Self::DirectInput { + offset, + sender: direct.sender.to_string(), + block_number: direct.block_number, + payload: alloy_primitives::hex::encode_prefixed(direct.payload.as_slice()), + input_index, + batch_nonce, } } } diff --git a/sequencer/src/egress/api/subscribe.rs b/sequencer/src/egress/api/subscribe.rs index 897f73f..ab2abb3 100644 --- a/sequencer/src/egress/api/subscribe.rs +++ b/sequencer/src/egress/api/subscribe.rs @@ -72,12 +72,10 @@ async fn run_ws_session( max_catchup_events, "ws catch-up window exceeded; closing subscriber" ); - close_with_frame( - &mut socket, - close_code::POLICY, - WS_CATCHUP_WINDOW_EXCEEDED_REASON, - ) - .await; + let reason = format!( + "{WS_CATCHUP_WINDOW_EXCEEDED_REASON}: live_start_offset={live_start_offset}" + ); + close_with_frame(&mut socket, close_code::POLICY, reason.as_str()).await; return; } Err(SubscribeError::OpenStorage { source }) => { diff --git a/sequencer/src/egress/l2_tx_feed/mod.rs b/sequencer/src/egress/l2_tx_feed/mod.rs index 0b65a74..8e35db4 100644 --- a/sequencer/src/egress/l2_tx_feed/mod.rs +++ b/sequencer/src/egress/l2_tx_feed/mod.rs @@ -14,11 +14,10 @@ pub use sequencer_core::broadcast::BroadcastTxMessage; use std::time::Duration; use alloy_primitives::Address; -use sequencer_core::l2_tx::SequencedL2Tx; use tokio::sync::mpsc; use crate::runtime::shutdown::ShutdownSignal; -use crate::storage::Storage; +use crate::storage::{OrderedL2TxRow, Storage}; #[derive(Debug, Clone, Copy)] pub struct L2TxFeedConfig { @@ -182,7 +181,7 @@ fn run_subscription( } let txs = storage - .ordered_l2_txs_page_from(next_offset, page_size) + .ordered_l2_tx_rows_page_from(next_offset, page_size) .map_err(|source| SubscriptionError::LoadReplay { offset: next_offset, source, @@ -193,35 +192,36 @@ fn run_subscription( continue; } - // The frame safe_block (third element) is a replay-only concern; - // the WS message shape doesn't carry it (review F7/WP5 owns any - // feed-protocol extension). - for (db_offset, tx, _frame_safe_block) in txs { + for row in txs { if shutdown.is_shutdown_requested() || events_tx.is_closed() { return Ok(()); } - next_offset = db_offset; - - if should_filter_from_broadcast(&tx, batch_submitter_address) { - continue; - } - - let event = BroadcastTxMessage::from_offset_and_tx(db_offset, tx); + next_offset = row.offset(); + let event = match row { + OrderedL2TxRow::UserOp { + offset, + tx, + nonce, + safe_block, + batch_nonce, + } => BroadcastTxMessage::from_user_op(offset, tx, nonce, safe_block, batch_nonce), + OrderedL2TxRow::DirectInput { + offset, + tx, + input_index, + batch_nonce, + .. + } => { + if batch_submitter_address == Some(tx.sender) { + continue; + } + BroadcastTxMessage::from_direct_input(offset, tx, input_index, batch_nonce) + } + }; if events_tx.blocking_send(event).is_err() { return Ok(()); } } } } - -fn should_filter_from_broadcast( - tx: &SequencedL2Tx, - batch_submitter_address: Option
, -) -> bool { - matches!( - (tx, batch_submitter_address), - (SequencedL2Tx::Direct(direct), Some(batch_submitter_address)) - if direct.sender == batch_submitter_address - ) -} diff --git a/sequencer/src/egress/l2_tx_feed/tests.rs b/sequencer/src/egress/l2_tx_feed/tests.rs index 3f996a4..81a9225 100644 --- a/sequencer/src/egress/l2_tx_feed/tests.rs +++ b/sequencer/src/egress/l2_tx_feed/tests.rs @@ -11,35 +11,43 @@ use crate::ingress::inclusion_lane::{PendingUserOp, SequencerError}; use crate::runtime::shutdown::ShutdownSignal; use crate::storage::test_helpers::temp_db; use crate::storage::{SafeInputRange, Storage, StoredSafeInput}; -use sequencer_core::l2_tx::{DirectInput, SequencedL2Tx, ValidUserOp}; +use sequencer_core::l2_tx::{DirectInput, ValidUserOp}; use sequencer_core::user_op::UserOp; #[test] fn broadcast_user_op_serializes_with_hex_data() { - let msg = BroadcastTxMessage::from_offset_and_tx( + let msg = BroadcastTxMessage::from_user_op( 7, - SequencedL2Tx::UserOp(ValidUserOp { + ValidUserOp { sender: Address::from_slice(&[0x11; 20]), fee: 3, data: vec![0xaa, 0xbb], - }), + }, + 11, + 1_234, + 5, ); let json = serde_json::to_string(&msg).expect("serialize"); assert!(json.contains("\"kind\":\"user_op\"")); assert!(json.contains("\"offset\":7")); + assert!(json.contains("\"nonce\":11")); assert!(json.contains("\"fee\":3")); assert!(json.contains("\"data\":\"0xaabb\"")); + assert!(json.contains("\"safe_block\":1234")); + assert!(json.contains("\"batch_nonce\":5")); } #[test] fn broadcast_direct_input_serializes_with_hex_payload() { - let msg = BroadcastTxMessage::from_offset_and_tx( + let msg = BroadcastTxMessage::from_direct_input( 9, - SequencedL2Tx::Direct(DirectInput { + DirectInput { sender: Address::ZERO, block_number: 42, payload: vec![0xcc, 0xdd], - }), + }, + 3, + 5, ); let json = serde_json::to_string(&msg).expect("serialize"); assert!(json.contains("\"kind\":\"direct_input\"")); @@ -47,6 +55,8 @@ fn broadcast_direct_input_serializes_with_hex_payload() { assert!(json.contains("\"sender\":\"0x0000000000000000000000000000000000000000\"")); assert!(json.contains("\"block_number\":42")); assert!(json.contains("\"payload\":\"0xccdd\"")); + assert!(json.contains("\"input_index\":3")); + assert!(json.contains("\"batch_nonce\":5")); } #[tokio::test(flavor = "multi_thread", worker_threads = 2)] @@ -98,9 +108,25 @@ async fn subscription_replays_existing_rows_in_order() { .expect("wait second event") .expect("second event"); - // DB offsets (SQLite rowid) start at 1. - assert_eq!(first.offset(), 1); - assert_eq!(second.offset(), 2); + assert!(matches!( + first, + BroadcastTxMessage::UserOp { + offset: 1, + nonce: 0, + safe_block: 0, + batch_nonce: 0, + .. + } + )); + assert!(matches!( + second, + BroadcastTxMessage::DirectInput { + offset: 2, + input_index: 0, + batch_nonce: 0, + .. + } + )); subscription.finish().await.expect("finish subscription"); } diff --git a/sequencer/src/storage/egress.rs b/sequencer/src/storage/egress.rs index d182d17..c46a66e 100644 --- a/sequencer/src/storage/egress.rs +++ b/sequencer/src/storage/egress.rs @@ -13,21 +13,79 @@ use rusqlite::{Result, params}; use super::Storage; use super::convert::{i64_to_u64, u64_to_i64, usize_to_i64}; use super::queries::decode_l2_tx_row; -use sequencer_core::l2_tx::SequencedL2Tx; +use sequencer_core::l2_tx::{DirectInput, SequencedL2Tx, ValidUserOp}; + +/// One persisted L2 transaction and the ordering context of its covering frame. +#[derive(Debug, Clone)] +pub(crate) enum OrderedL2TxRow { + UserOp { + offset: u64, + tx: ValidUserOp, + nonce: u32, + safe_block: u64, + batch_nonce: u64, + }, + DirectInput { + offset: u64, + tx: DirectInput, + input_index: u64, + safe_block: u64, + batch_nonce: u64, + }, +} + +impl OrderedL2TxRow { + pub(crate) fn offset(&self) -> u64 { + match self { + Self::UserOp { offset, .. } | Self::DirectInput { offset, .. } => *offset, + } + } + + fn into_replay_parts(self) -> (u64, SequencedL2Tx, u64) { + match self { + Self::UserOp { + offset, + tx, + safe_block, + .. + } => (offset, SequencedL2Tx::UserOp(tx), safe_block), + Self::DirectInput { + offset, + tx, + safe_block, + .. + } => (offset, SequencedL2Tx::Direct(tx), safe_block), + } + } +} impl Storage { /// Load a page of ordered L2 transactions starting after the given offset. /// Returns `(db_offset, tx, frame_safe_block)` triples — the third element /// is the covering frame's `safe_block`. Catch-up replay feeds it to /// `execute_valid_user_op` so the app's safe-block clock advances exactly - /// as it did live (directs use their own `block_number` instead; the feed - /// ignores it). Callers should track `db_offset` of the last item as their - /// cursor, not increment a counter. + /// as it did live (directs use their own `block_number` instead). Callers + /// should track `db_offset` of the last item as their cursor, not increment + /// a counter. pub fn ordered_l2_txs_page_from( &mut self, offset: u64, limit: usize, ) -> Result> { + self.ordered_l2_tx_rows_page_from(offset, limit) + .map(|rows| { + rows.into_iter() + .map(OrderedL2TxRow::into_replay_parts) + .collect() + }) + } + + /// Load feed rows with all persisted execution and settlement context. + pub(crate) fn ordered_l2_tx_rows_page_from( + &mut self, + offset: u64, + limit: usize, + ) -> Result> { if limit == 0 { return Ok(Vec::new()); } @@ -45,7 +103,10 @@ impl Storage { CASE WHEN s.user_op_pos_in_frame IS NOT NULL THEN f.fee ELSE NULL END AS fee, CASE WHEN s.safe_input_index IS NOT NULL THEN d.payload ELSE NULL END AS payload, CASE WHEN s.safe_input_index IS NOT NULL THEN d.block_number ELSE NULL END AS block_number, - f.safe_block + f.safe_block, + b.nonce, + s.safe_input_index, + CASE WHEN s.user_op_pos_in_frame IS NOT NULL THEN u.nonce ELSE NULL END AS op_nonce FROM valid_sequenced_l2_txs s LEFT JOIN user_ops u ON u.batch_index = s.batch_index @@ -56,6 +117,8 @@ impl Storage { AND f.frame_in_batch = s.frame_in_batch LEFT JOIN safe_inputs d ON d.safe_input_index = s.safe_input_index + LEFT JOIN batches b + ON b.batch_index = s.batch_index WHERE s.offset > ?1 ORDER BY s.offset ASC LIMIT ?2 @@ -71,10 +134,27 @@ impl Storage { row.get(5)?, row.get(6)?, ); - // Non-NULL for every sequenced row: the frames row is inserted - // when the frame opens, before anything is sequenced into it. - let frame_safe_block: i64 = row.get(7)?; - Ok((i64_to_u64(db_offset), tx, i64_to_u64(frame_safe_block))) + // Non-NULL for every sequenced row: batches and frames exist before + // anything can be sequenced into them. + let safe_block = i64_to_u64(row.get(7)?); + let batch_nonce = i64_to_u64(row.get(8)?); + match tx { + SequencedL2Tx::UserOp(tx) => Ok(OrderedL2TxRow::UserOp { + offset: i64_to_u64(db_offset), + tx, + nonce: u32::try_from(row.get::<_, i64>(10)?) + .expect("persisted user op nonce must fit u32"), + safe_block, + batch_nonce, + }), + SequencedL2Tx::Direct(tx) => Ok(OrderedL2TxRow::DirectInput { + offset: i64_to_u64(db_offset), + tx, + input_index: i64_to_u64(row.get(9)?), + safe_block, + batch_nonce, + }), + } })?; rows.collect::>>() } diff --git a/sequencer/src/storage/mod.rs b/sequencer/src/storage/mod.rs index 20d2f6d..8990ba6 100644 --- a/sequencer/src/storage/mod.rs +++ b/sequencer/src/storage/mod.rs @@ -44,6 +44,7 @@ pub(crate) mod test_helpers; use std::time::SystemTime; use thiserror::Error; +pub(crate) use egress::OrderedL2TxRow; pub use open::Storage; pub use recovery::DangerStatus; pub use snapshot_dumps::{ diff --git a/sequencer/tests/e2e_sequencer.rs b/sequencer/tests/e2e_sequencer.rs index eaaa5e5..b62b670 100644 --- a/sequencer/tests/e2e_sequencer.rs +++ b/sequencer/tests/e2e_sequencer.rs @@ -259,6 +259,7 @@ async fn e2e_submit_tx_ack_and_broadcast() { sender: ws_sender, fee, data, + .. } => { assert_eq!(offset, 2); assert_eq!(ws_sender, sender.to_string()); @@ -1367,6 +1368,7 @@ fn assert_ws_message_matches_tx( sender, fee, data, + .. }, SequencedL2Tx::UserOp(expected), ) => { @@ -1384,6 +1386,7 @@ fn assert_ws_message_matches_tx( sender, block_number, payload, + .. }, SequencedL2Tx::Direct(expected), ) => { diff --git a/sequencer/tests/ws_broadcaster.rs b/sequencer/tests/ws_broadcaster.rs index 2c7a0d7..817471b 100644 --- a/sequencer/tests/ws_broadcaster.rs +++ b/sequencer/tests/ws_broadcaster.rs @@ -237,7 +237,10 @@ async fn ws_subscribe_closes_when_catchup_window_exceeds_limit() { close_frame.code, tokio_tungstenite::tungstenite::protocol::frame::coding::CloseCode::Policy ); - assert_eq!(close_frame.reason, WS_CATCHUP_WINDOW_EXCEEDED_REASON); + assert_eq!( + close_frame.reason, + format!("{WS_CATCHUP_WINDOW_EXCEEDED_REASON}: live_start_offset=2") + ); } other => panic!("expected close frame for catch-up limit, got {other:?}"), } @@ -562,6 +565,7 @@ fn assert_ws_message_matches_tx( sender, fee, data, + .. }, SequencedL2Tx::UserOp(expected), ) => { @@ -579,6 +583,7 @@ fn assert_ws_message_matches_tx( sender, block_number, payload, + .. }, SequencedL2Tx::Direct(expected), ) => { diff --git a/tests/harness/src/replay.rs b/tests/harness/src/replay.rs index 6bd3bfc..747dc19 100644 --- a/tests/harness/src/replay.rs +++ b/tests/harness/src/replay.rs @@ -55,19 +55,19 @@ pub(crate) fn apply_ws_message( })?; } WsTxMessage::UserOp { - sender, fee, data, .. + sender, + fee, + data, + safe_block, + .. } => { - // The WS feed does not carry the covering frame's safe_block - // yet (feed-protocol work, review F7/WP5), so the replayed - // app's safe-block clock lags the live one. Fine here: these - // replays assert balances/nonces, never the clock. app.execute_valid_user_op( &ValidUserOp { sender: decode_address(sender.as_str()), fee, data: decode_hex_prefixed(data.as_str()), }, - 0, + safe_block, )?; } } From cc949442a779401859e38d0e6218d8f5fc7302bd Mon Sep 17 00:00:00 2001 From: Stephen Chen <20940639+stephenctw@users.noreply.github.com> Date: Wed, 22 Jul 2026 23:21:51 +0800 Subject: [PATCH 2/4] feat(sequencer): add direct input L1 provenance --- README.md | 3 +- sequencer-core/src/broadcast.rs | 8 ++ sequencer/src/egress/l2_tx_feed/mod.rs | 11 +- sequencer/src/egress/l2_tx_feed/tests.rs | 20 ++- sequencer/src/l1/reader.rs | 16 ++- sequencer/src/storage/egress.rs | 10 +- sequencer/src/storage/l1_inputs.rs | 119 ++++++++++++++++-- .../src/storage/migrations/0001_schema.sql | 6 +- sequencer/src/storage/mod.rs | 10 ++ 9 files changed, 177 insertions(+), 26 deletions(-) diff --git a/README.md b/README.md index af40487..71da28b 100644 --- a/README.md +++ b/README.md @@ -156,6 +156,7 @@ Notes: - `from_offset` is optional and defaults to `0`. - messages are JSON text frames. - binary fields are hex-encoded (`0x`-prefixed). +- direct-input `block_timestamp` values are Unix seconds. - the current runtime enforces a subscriber cap of `64` and a catch-up cap of `50000` events. - if the requested catch-up window exceeds that cap, the server upgrades and then immediately closes the socket with close code `1008` (`POLICY`) and reason `catch-up window exceeded: live_start_offset=`; reconnecting at that offset starts from the current live head. @@ -166,7 +167,7 @@ Message shapes: ``` ```json -{ "kind": "direct_input", "offset": 11, "sender": "0x...", "block_number": 123, "payload": "0x...", "input_index": 42, "batch_nonce": 4 } +{ "kind": "direct_input", "offset": 11, "sender": "0x...", "block_number": 123, "block_timestamp": 1700000000, "transaction_hash": "0x...", "payload": "0x...", "input_index": 42, "batch_nonce": 4 } ``` Success response: diff --git a/sequencer-core/src/broadcast.rs b/sequencer-core/src/broadcast.rs index cb0ce16..641cab2 100644 --- a/sequencer-core/src/broadcast.rs +++ b/sequencer-core/src/broadcast.rs @@ -34,6 +34,10 @@ pub enum BroadcastTxMessage { input_index: u64, /// Nonce of the batch that drained this direct input. batch_nonce: u64, + /// Unix timestamp, in seconds, of the L1 block containing this direct input. + block_timestamp: u64, + /// Hash of the L1 transaction that carried this direct input. + transaction_hash: String, }, } @@ -68,6 +72,8 @@ impl BroadcastTxMessage { direct: DirectInput, input_index: u64, batch_nonce: u64, + block_timestamp: u64, + transaction_hash: alloy_primitives::B256, ) -> Self { Self::DirectInput { offset, @@ -76,6 +82,8 @@ impl BroadcastTxMessage { payload: alloy_primitives::hex::encode_prefixed(direct.payload.as_slice()), input_index, batch_nonce, + block_timestamp, + transaction_hash: alloy_primitives::hex::encode_prefixed(transaction_hash), } } } diff --git a/sequencer/src/egress/l2_tx_feed/mod.rs b/sequencer/src/egress/l2_tx_feed/mod.rs index 8e35db4..6820b35 100644 --- a/sequencer/src/egress/l2_tx_feed/mod.rs +++ b/sequencer/src/egress/l2_tx_feed/mod.rs @@ -211,12 +211,21 @@ fn run_subscription( tx, input_index, batch_nonce, + block_timestamp, + transaction_hash, .. } => { if batch_submitter_address == Some(tx.sender) { continue; } - BroadcastTxMessage::from_direct_input(offset, tx, input_index, batch_nonce) + BroadcastTxMessage::from_direct_input( + offset, + tx, + input_index, + batch_nonce, + block_timestamp, + transaction_hash, + ) } }; if events_tx.blocking_send(event).is_err() { diff --git a/sequencer/src/egress/l2_tx_feed/tests.rs b/sequencer/src/egress/l2_tx_feed/tests.rs index 81a9225..02f357a 100644 --- a/sequencer/src/egress/l2_tx_feed/tests.rs +++ b/sequencer/src/egress/l2_tx_feed/tests.rs @@ -3,14 +3,14 @@ use std::time::{Duration, SystemTime}; -use alloy_primitives::{Address, Signature}; +use alloy_primitives::{Address, B256, Signature}; use tokio::sync::oneshot; use super::{BroadcastTxMessage, L2TxFeed, L2TxFeedConfig, SubscribeError}; use crate::ingress::inclusion_lane::{PendingUserOp, SequencerError}; use crate::runtime::shutdown::ShutdownSignal; use crate::storage::test_helpers::temp_db; -use crate::storage::{SafeInputRange, Storage, StoredSafeInput}; +use crate::storage::{FrontierMode, IngestedSafeInput, SafeInputRange, Storage, StoredSafeInput}; use sequencer_core::l2_tx::{DirectInput, ValidUserOp}; use sequencer_core::user_op::UserOp; @@ -48,6 +48,8 @@ fn broadcast_direct_input_serializes_with_hex_payload() { }, 3, 5, + 1_700_000_000, + B256::repeat_byte(0xab), ); let json = serde_json::to_string(&msg).expect("serialize"); assert!(json.contains("\"kind\":\"direct_input\"")); @@ -57,6 +59,8 @@ fn broadcast_direct_input_serializes_with_hex_payload() { assert!(json.contains("\"payload\":\"0xccdd\"")); assert!(json.contains("\"input_index\":3")); assert!(json.contains("\"batch_nonce\":5")); + assert!(json.contains("\"block_timestamp\":1700000000")); + assert!(json.contains(&format!("\"transaction_hash\":\"0x{}\"", "ab".repeat(32)))); } #[tokio::test(flavor = "multi_thread", worker_threads = 2)] @@ -124,8 +128,10 @@ async fn subscription_replays_existing_rows_in_order() { offset: 2, input_index: 0, batch_nonce: 0, + block_timestamp: 1_700_000_000, + transaction_hash, .. - } + } if transaction_hash == B256::repeat_byte(0xcd).to_string() )); subscription.finish().await.expect("finish subscription"); @@ -383,12 +389,15 @@ fn seed_ordered_txs_with_sender(db_path: &str, direct_sender: Address) { .append_user_ops_chunk(&mut head, &[pending]) .expect("append user-op chunk"); storage - .append_safe_inputs( + .append_ingested_safe_inputs_with_timestamp( 10, - &[StoredSafeInput { + 10, + &[IngestedSafeInput { sender: direct_sender, payload: vec![0xaa], block_number: 10, + block_timestamp: 1_700_000_000, + transaction_hash: B256::repeat_byte(0xcd), }], Address::ZERO, &sequencer_core::protocol::ProtocolTiming { @@ -397,6 +406,7 @@ fn seed_ordered_txs_with_sender(db_path: &str, direct_sender: Address) { l1_read_stale_after_blocks: 900, seconds_per_block: 12, }, + FrontierMode::Populate, ) .expect("append direct input"); storage diff --git a/sequencer/src/l1/reader.rs b/sequencer/src/l1/reader.rs index 2915ce6..612aa14 100644 --- a/sequencer/src/l1/reader.rs +++ b/sequencer/src/l1/reader.rs @@ -20,7 +20,7 @@ use tracing::info; use crate::l1::partition::{decode_evm_advance_input, get_input_added_events_ordered}; use crate::runtime::shutdown::ShutdownSignal; -use crate::storage::{FrontierMode, Storage, StorageOpenError, StoredSafeInput}; +use crate::storage::{FrontierMode, IngestedSafeInput, Storage, StorageOpenError}; use sequencer_core::protocol::ProtocolTiming; #[derive(Debug, Clone)] @@ -290,6 +290,9 @@ impl InputReader { let block_number = log.block_number.ok_or_else(|| { InputReaderError::Provider("InputAdded log missing block_number".to_string()) })?; + let transaction_hash = log.transaction_hash.ok_or_else(|| { + InputReaderError::Provider("InputAdded log missing transaction_hash".to_string()) + })?; let evm_advance = decode_evm_advance_input(event.input.as_ref()) .map_err(InputReaderError::Provider)?; assert_eq!( @@ -298,11 +301,16 @@ impl InputReader { "InputAdded block number mismatch: log={block_number}, payload={}", evm_advance.blockNumber ); + let block_timestamp = u64::try_from(evm_advance.blockTimestamp).map_err(|_| { + InputReaderError::Provider("EvmAdvance block timestamp exceeds u64".to_string()) + })?; - batch.push(StoredSafeInput { + batch.push(IngestedSafeInput { sender: evm_advance.msgSender, payload: evm_advance.payload.into(), block_number, + block_timestamp, + transaction_hash, }); } @@ -391,7 +399,7 @@ impl InputReader { async fn append_safe_inputs( &self, current_safe_head: SafeHead, - batch: Vec, + batch: Vec, ) -> Result<(), InputReaderError> { let db_path = self.db_path.clone(); let batch_submitter = self.batch_submitter; @@ -400,7 +408,7 @@ impl InputReader { tokio::task::spawn_blocking(move || { let mut storage = Storage::open(&db_path)?; storage - .append_safe_inputs_with_timestamp( + .append_ingested_safe_inputs_with_timestamp( current_safe_head.block_number, current_safe_head.block_timestamp, &batch, diff --git a/sequencer/src/storage/egress.rs b/sequencer/src/storage/egress.rs index c46a66e..d2692e6 100644 --- a/sequencer/src/storage/egress.rs +++ b/sequencer/src/storage/egress.rs @@ -7,7 +7,7 @@ //! or counts over it. The view encapsulates the exclusion of invalidated batches //! so callers don't repeat the filter. -use alloy_primitives::Address; +use alloy_primitives::{Address, B256}; use rusqlite::{Result, params}; use super::Storage; @@ -31,6 +31,8 @@ pub(crate) enum OrderedL2TxRow { input_index: u64, safe_block: u64, batch_nonce: u64, + block_timestamp: u64, + transaction_hash: B256, }, } @@ -106,7 +108,9 @@ impl Storage { f.safe_block, b.nonce, s.safe_input_index, - CASE WHEN s.user_op_pos_in_frame IS NOT NULL THEN u.nonce ELSE NULL END AS op_nonce + CASE WHEN s.user_op_pos_in_frame IS NOT NULL THEN u.nonce ELSE NULL END AS op_nonce, + CASE WHEN s.safe_input_index IS NOT NULL THEN d.block_timestamp ELSE NULL END AS block_timestamp, + CASE WHEN s.safe_input_index IS NOT NULL THEN d.transaction_hash ELSE NULL END AS transaction_hash FROM valid_sequenced_l2_txs s LEFT JOIN user_ops u ON u.batch_index = s.batch_index @@ -153,6 +157,8 @@ impl Storage { input_index: i64_to_u64(row.get(9)?), safe_block, batch_nonce, + block_timestamp: i64_to_u64(row.get(11)?), + transaction_hash: B256::from_slice(row.get::<_, Vec>(12)?.as_slice()), }), } })?; diff --git a/sequencer/src/storage/l1_inputs.rs b/sequencer/src/storage/l1_inputs.rs index fc4b3f2..944e374 100644 --- a/sequencer/src/storage/l1_inputs.rs +++ b/sequencer/src/storage/l1_inputs.rs @@ -7,7 +7,7 @@ //! Also exposes the read-side queries the input reader and other callers need //! (current safe block, safe-input bounds, last safe-progress timestamp). -use alloy_primitives::Address; +use alloy_primitives::{Address, B256}; use rusqlite::{OptionalExtension, Result, Transaction, params}; use super::Storage; @@ -17,9 +17,62 @@ use super::queries::{ query_latest_safe_input_index_exclusive, }; use super::safe_accepted_batches::populate_safe_accepted_batches; -use super::{DeploymentIdentity, FrontierMode, StoredSafeInput}; +use super::{DeploymentIdentity, FrontierMode, IngestedSafeInput, StoredSafeInput}; use sequencer_core::protocol::ProtocolTiming; +trait SafeInputRecord { + fn sender(&self) -> Address; + fn payload(&self) -> &[u8]; + fn block_number(&self) -> u64; + fn block_timestamp(&self) -> u64; + fn transaction_hash(&self) -> B256; +} + +impl SafeInputRecord for StoredSafeInput { + fn sender(&self) -> Address { + self.sender + } + + fn payload(&self) -> &[u8] { + self.payload.as_slice() + } + + fn block_number(&self) -> u64 { + self.block_number + } + + // Synthetic storage fixtures do not model L1 provenance. + fn block_timestamp(&self) -> u64 { + 0 + } + + fn transaction_hash(&self) -> B256 { + B256::ZERO + } +} + +impl SafeInputRecord for IngestedSafeInput { + fn sender(&self) -> Address { + self.sender + } + + fn payload(&self) -> &[u8] { + self.payload.as_slice() + } + + fn block_number(&self) -> u64 { + self.block_number + } + + fn block_timestamp(&self) -> u64 { + self.block_timestamp + } + + fn transaction_hash(&self) -> B256 { + self.transaction_hash + } +} + impl Storage { /// `MAX(safe_input_index) + 1` (or 0 if empty). The exclusive bound on the /// `safe_inputs` table — the next index a fresh row would receive. @@ -146,9 +199,8 @@ impl Storage { } /// Same as [`Storage::append_safe_inputs`], but records the L1 timestamp - /// of `safe_block`. Production input-reader code should use this path; - /// the shorter helper exists for tests that only need a fresh synthetic - /// safe head. + /// of `safe_block`. Synthetic inputs receive zero provenance; production + /// input-reader code uses [`Storage::append_ingested_safe_inputs_with_timestamp`]. /// /// `frontier` gates the `safe_accepted_batches` update — see /// [`FrontierMode`]. Everything except `setup --recovery`'s interim syncs @@ -161,6 +213,46 @@ impl Storage { batch_submitter: Address, timing: &ProtocolTiming, frontier: FrontierMode, + ) -> Result<()> { + self.append_safe_input_records_with_timestamp( + safe_block, + safe_block_timestamp, + inputs, + batch_submitter, + timing, + frontier, + ) + } + + /// Production input-reader path. Persists per-input L1 provenance together + /// with the safe-input row and safe-head advance. + pub(crate) fn append_ingested_safe_inputs_with_timestamp( + &mut self, + safe_block: u64, + safe_block_timestamp: u64, + inputs: &[IngestedSafeInput], + batch_submitter: Address, + timing: &ProtocolTiming, + frontier: FrontierMode, + ) -> Result<()> { + self.append_safe_input_records_with_timestamp( + safe_block, + safe_block_timestamp, + inputs, + batch_submitter, + timing, + frontier, + ) + } + + fn append_safe_input_records_with_timestamp( + &mut self, + safe_block: u64, + safe_block_timestamp: u64, + inputs: &[T], + batch_submitter: Address, + timing: &ProtocolTiming, + frontier: FrontierMode, ) -> Result<()> { self.write(|tx| { if let Some(current) = current_safe_block(tx)? { @@ -312,24 +404,27 @@ fn query_deployment_identity(conn: &rusqlite::Connection) -> Result( tx: &Transaction<'_>, start_index: u64, - inputs: &[StoredSafeInput], + inputs: &[T], ) -> Result<()> { if inputs.is_empty() { return Ok(()); } let mut stmt = tx.prepare_cached( - "INSERT INTO safe_inputs (safe_input_index, sender, payload, block_number) \ - VALUES (?1, ?2, ?3, ?4)", + "INSERT INTO safe_inputs \ + (safe_input_index, sender, payload, block_number, block_timestamp, transaction_hash) \ + VALUES (?1, ?2, ?3, ?4, ?5, ?6)", )?; for (offset, input) in inputs.iter().enumerate() { stmt.execute(params![ u64_to_i64(start_index.saturating_add(offset as u64)), - input.sender.as_slice(), - input.payload.as_slice(), - u64_to_i64(input.block_number), + input.sender().as_slice(), + input.payload(), + u64_to_i64(input.block_number()), + u64_to_i64(input.block_timestamp()), + input.transaction_hash().as_slice(), ])?; } Ok(()) diff --git a/sequencer/src/storage/migrations/0001_schema.sql b/sequencer/src/storage/migrations/0001_schema.sql index 20a5da4..49bd6f5 100644 --- a/sequencer/src/storage/migrations/0001_schema.sql +++ b/sequencer/src/storage/migrations/0001_schema.sql @@ -251,7 +251,11 @@ CREATE TABLE IF NOT EXISTS safe_inputs ( sender BLOB NOT NULL CHECK (length(sender) = 20), payload BLOB NOT NULL, -- Block number of the chain block where this direct input was included (e.g. InputAdded event block). - block_number INTEGER NOT NULL CHECK (block_number >= 0) + block_number INTEGER NOT NULL CHECK (block_number >= 0), + -- Timestamp of the carrying L1 block. + block_timestamp INTEGER NOT NULL CHECK (block_timestamp >= 0), + -- Hash of the L1 transaction that carried this input. + transaction_hash BLOB NOT NULL CHECK (length(transaction_hash) = 32) ); CREATE INDEX IF NOT EXISTS idx_safe_inputs_sender diff --git a/sequencer/src/storage/mod.rs b/sequencer/src/storage/mod.rs index 8990ba6..49d0f14 100644 --- a/sequencer/src/storage/mod.rs +++ b/sequencer/src/storage/mod.rs @@ -61,6 +61,16 @@ pub struct StoredSafeInput { pub block_number: u64, } +/// One InputBox event with the L1 provenance persisted for feed consumers. +#[derive(Debug, Clone, PartialEq, Eq)] +pub(crate) struct IngestedSafeInput { + pub sender: alloy_primitives::Address, + pub payload: Vec, + pub block_number: u64, + pub block_timestamp: u64, + pub transaction_hash: alloy_primitives::B256, +} + /// Whether a sync also maintains the scheduler-accepted gold frontier /// (`safe_accepted_batches`). /// From ddf8172dd9581dd23f3f8c9c7083e966d5889120 Mon Sep 17 00:00:00 2001 From: Stephen Chen <20940639+stephenctw@users.noreply.github.com> Date: Wed, 22 Jul 2026 23:48:38 +0800 Subject: [PATCH 3/4] test(sequencer): fix provenance checks and recovery polling --- sequencer/src/storage/recovery_tests.rs | 44 ++++++++++++++++++++++--- tests/e2e/src/test_cases.rs | 39 +++++++++++++--------- 2 files changed, 64 insertions(+), 19 deletions(-) diff --git a/sequencer/src/storage/recovery_tests.rs b/sequencer/src/storage/recovery_tests.rs index 956cd5f..43349db 100644 --- a/sequencer/src/storage/recovery_tests.rs +++ b/sequencer/src/storage/recovery_tests.rs @@ -2039,8 +2039,9 @@ mod schema_invariants { let db = temp_db("schema-safe-input-sender-len"); let storage = Storage::open(db.path.as_str()).expect("open storage"); let err = storage.conn.execute( - "INSERT INTO safe_inputs (safe_input_index, sender, payload, block_number) \ - VALUES (0, X'DEADBEEF', X'00', 10)", + "INSERT INTO safe_inputs \ + (safe_input_index, sender, payload, block_number, block_timestamp, transaction_hash) \ + VALUES (0, X'DEADBEEF', X'00', 10, 0, zeroblob(32))", [], ); assert!( @@ -2152,8 +2153,9 @@ mod schema_invariants { let storage = Storage::open(db.path.as_str()).expect("open storage"); let sender = vec![0u8; 20]; let err = storage.conn.execute( - "INSERT INTO safe_inputs (safe_input_index, sender, payload, block_number) \ - VALUES (0, ?1, X'00', -1)", + "INSERT INTO safe_inputs \ + (safe_input_index, sender, payload, block_number, block_timestamp, transaction_hash) \ + VALUES (0, ?1, X'00', -1, 0, zeroblob(32))", params![sender], ); assert!( @@ -2161,6 +2163,40 @@ mod schema_invariants { "expected CHECK constraint error on block_number >= 0, got: {err:?}", ); } + + #[test] + fn schema_rejects_safe_input_with_negative_block_timestamp() { + let db = temp_db("schema-safe-input-neg-block-timestamp"); + let storage = Storage::open(db.path.as_str()).expect("open storage"); + let sender = vec![0u8; 20]; + let err = storage.conn.execute( + "INSERT INTO safe_inputs \ + (safe_input_index, sender, payload, block_number, block_timestamp, transaction_hash) \ + VALUES (0, ?1, X'00', 10, -1, zeroblob(32))", + params![sender], + ); + assert!( + format!("{err:?}").contains("CHECK constraint failed"), + "expected CHECK constraint error on block_timestamp >= 0, got: {err:?}", + ); + } + + #[test] + fn schema_rejects_safe_input_with_wrong_transaction_hash_length() { + let db = temp_db("schema-safe-input-transaction-hash-len"); + let storage = Storage::open(db.path.as_str()).expect("open storage"); + let sender = vec![0u8; 20]; + let err = storage.conn.execute( + "INSERT INTO safe_inputs \ + (safe_input_index, sender, payload, block_number, block_timestamp, transaction_hash) \ + VALUES (0, ?1, X'00', 10, 0, X'DEADBEEF')", + params![sender], + ); + assert!( + format!("{err:?}").contains("CHECK constraint failed"), + "expected CHECK constraint error on transaction_hash length, got: {err:?}", + ); + } } mod tree_invariants { diff --git a/tests/e2e/src/test_cases.rs b/tests/e2e/src/test_cases.rs index b4b79fe..d7874a4 100644 --- a/tests/e2e/src/test_cases.rs +++ b/tests/e2e/src/test_cases.rs @@ -38,6 +38,13 @@ const PROMOTION_POLL_ATTEMPTS: usize = 40; /// tick and the promoter room to run between mines. const PROMOTION_POLL_INTERVAL: Duration = Duration::from_secs(1); +/// Poll budget for waiting until a landed batch reaches the reader's +/// `safe_accepted_batches` frontier. +const ACCEPTANCE_POLL_ATTEMPTS: usize = 40; + +/// Per-attempt pause while waiting for batch submission and safe-head ingestion. +const ACCEPTANCE_POLL_INTERVAL: Duration = Duration::from_secs(1); + // ── Zone-math constants for the outage-matrix and recovery tests ───────── // // These derive from the sequencer's default config so a change to @@ -429,6 +436,20 @@ async fn mine_until_finalized_advances( ) } +async fn mine_until_batch_is_safe_accepted( + runtime: &ManagedSequencer, +) -> ScenarioResult<(u64, Option)> { + for _ in 0..ACCEPTANCE_POLL_ATTEMPTS { + runtime.mine_l1_blocks(2).await?; + tokio::time::sleep(ACCEPTANCE_POLL_INTERVAL).await; + let accepted = runtime.count_safe_accepted_batches()?; + if accepted.0 > 0 { + return Ok(accepted); + } + } + Err("timed out waiting for a batch to reach safe_accepted_batches".into()) +} + /// Close the open batch, land it on L1, and wait for snapshot promotion so /// `/finalized_state` reports a new `inclusion_block`. async fn drive_finalized_gold_batch_for_watchdog( @@ -3576,21 +3597,9 @@ async fn run_nonce_zero_recovery_invalidates_then_accepts_at_nonce_zero_test( replay_after.apply(ws_after.expect_user_op_from(alice_address).await?)?; } - // Wait for the submitter to fire a tick + submit the batch. Anvil's - // instamine puts the submission at 1 confirmation; the submitter's - // `wait_for_confirmations` needs `confirmation_depth + 1 = 3`. We - // explicitly mine the remaining 2 blocks below to unblock it without - // having to wait the full 72s timeout. - tokio::time::sleep(Duration::from_secs(7)).await; - runtime.mine_l1_blocks(2).await?; - - // After confirmations land, the submitter's tick loop continues: - // next iteration runs `refresh_recovery_metadata` → - // `populate_safe_accepted_batches`, which appends the batch - // to `safe_accepted_batches` at its expected nonce (0, reused). - tokio::time::sleep(Duration::from_secs(10)).await; - - let (accepted_count, min_accepted_nonce) = runtime.count_safe_accepted_batches()?; + // Mine and poll until submission, confirmation, and the input-reader sync + // have all completed. Fixed sleeps made this assertion race under CI load. + let (accepted_count, min_accepted_nonce) = mine_until_batch_is_safe_accepted(runtime).await?; assert!( accepted_count >= 1, "expected at least one batch to land in safe_accepted_batches \ From 30ab6fdf5bbc737ecec7b61bc66093d247f90854 Mon Sep 17 00:00:00 2001 From: Stephen Chen <20940639+stephenctw@users.noreply.github.com> Date: Thu, 23 Jul 2026 12:04:45 +0800 Subject: [PATCH 4/4] test(sequencer): harden WS enrichment review fixtures Pin non-zero feed context, disambiguate live_start_offset, and assert mirror nonce/safe-block fidelity so regressions cannot pass silently. --- examples/app-core/src/application/wallet.rs | 2 +- sequencer-core/src/broadcast.rs | 6 +- sequencer/src/egress/l2_tx_feed/tests.rs | 62 ++++++++++++++++----- sequencer/src/http.rs | 4 +- sequencer/tests/ws_broadcaster.rs | 17 ++++-- tests/e2e/src/test_cases.rs | 10 ++++ tests/harness/src/replay.rs | 14 +++++ 7 files changed, 95 insertions(+), 20 deletions(-) diff --git a/examples/app-core/src/application/wallet.rs b/examples/app-core/src/application/wallet.rs index 1afca01..a91d365 100644 --- a/examples/app-core/src/application/wallet.rs +++ b/examples/app-core/src/application/wallet.rs @@ -138,7 +138,7 @@ impl WalletApp { self.executed_input_count } - pub(crate) fn last_executed_safe_block(&self) -> u64 { + pub fn last_executed_safe_block(&self) -> u64 { self.last_executed_safe_block } diff --git a/sequencer-core/src/broadcast.rs b/sequencer-core/src/broadcast.rs index 641cab2..54ffa20 100644 --- a/sequencer-core/src/broadcast.rs +++ b/sequencer-core/src/broadcast.rs @@ -16,6 +16,8 @@ pub enum BroadcastTxMessage { offset: u64, sender: String, /// Signed replay-protection nonce of the user operation. + /// A mirror must apply the op under exactly this nonce; synthesizing + /// it from mirror state mis-executes ops across feed gaps. nonce: u32, /// Log-space fee exponent (base 129/128). See [`crate::fee`]. fee: u16, @@ -32,7 +34,9 @@ pub enum BroadcastTxMessage { payload: String, /// Per-application InputBox index of this direct input. input_index: u64, - /// Nonce of the batch that drained this direct input. + /// Nonce of the batch that drained this direct input. Lets a mirror + /// distinguish a direct already executed by the settled chain from one + /// still parked in its scheduler fridge (both have settled L1 inputs). batch_nonce: u64, /// Unix timestamp, in seconds, of the L1 block containing this direct input. block_timestamp: u64, diff --git a/sequencer/src/egress/l2_tx_feed/tests.rs b/sequencer/src/egress/l2_tx_feed/tests.rs index 02f357a..e1e389e 100644 --- a/sequencer/src/egress/l2_tx_feed/tests.rs +++ b/sequencer/src/egress/l2_tx_feed/tests.rs @@ -67,15 +67,16 @@ fn broadcast_direct_input_serializes_with_hex_payload() { async fn subscribe_from_rejects_catchup_window() { let db = temp_db("catchup-window"); seed_ordered_txs(db.path.as_str()); + append_direct_input(db.path.as_str()); let feed = test_feed(db.path.as_str(), ShutdownSignal::default()); - let result = feed.subscribe_from(0, 1); + let result = feed.subscribe_from(1, 1); assert!(matches!( result, Err(SubscribeError::CatchUpWindowExceeded { - requested_offset: 0, - live_start_offset: 2, + requested_offset: 1, + live_start_offset: 3, max_catchup_events: 1, }) )); @@ -116,9 +117,9 @@ async fn subscription_replays_existing_rows_in_order() { first, BroadcastTxMessage::UserOp { offset: 1, - nonce: 0, - safe_block: 0, - batch_nonce: 0, + nonce: 7, + safe_block: 123, + batch_nonce: 1, .. } )); @@ -127,7 +128,7 @@ async fn subscription_replays_existing_rows_in_order() { BroadcastTxMessage::DirectInput { offset: 2, input_index: 0, - batch_nonce: 0, + batch_nonce: 1, block_timestamp: 1_700_000_000, transaction_hash, .. @@ -367,8 +368,11 @@ fn seed_ordered_txs(db_path: &str) { fn seed_ordered_txs_with_sender(db_path: &str, direct_sender: Address) { let mut storage = Storage::open(db_path).expect("open storage"); let mut head = storage - .initialize_open_state(0, SafeInputRange::empty_at(0)) + .initialize_open_state(123, SafeInputRange::empty_at(0)) .expect("initialize open state"); + storage + .close_frame_and_batch(&mut head, 123) + .expect("advance to batch nonce 1"); let (respond_to, _recv) = oneshot::channel::>(); let pending = PendingUserOp { @@ -376,7 +380,7 @@ fn seed_ordered_txs_with_sender(db_path: &str, direct_sender: Address) { sender: Address::from_slice(&[0x11; 20]), signature: Signature::test_signature(), user_op: UserOp { - nonce: 0, + nonce: 7, max_fee: 3, data: vec![0x42].into(), }, @@ -390,12 +394,12 @@ fn seed_ordered_txs_with_sender(db_path: &str, direct_sender: Address) { .expect("append user-op chunk"); storage .append_ingested_safe_inputs_with_timestamp( - 10, - 10, + 456, + 456, &[IngestedSafeInput { sender: direct_sender, payload: vec![0xaa], - block_number: 10, + block_number: 456, block_timestamp: 1_700_000_000, transaction_hash: B256::repeat_byte(0xcd), }], @@ -410,6 +414,38 @@ fn seed_ordered_txs_with_sender(db_path: &str, direct_sender: Address) { ) .expect("append direct input"); storage - .close_frame_only(&mut head, 10, SafeInputRange::new(0, 1)) + .close_frame_only(&mut head, 456, SafeInputRange::new(0, 1)) .expect("close frame with one drained direct input"); } + +fn append_direct_input(db_path: &str) { + let mut storage = Storage::open(db_path).expect("open storage"); + let mut head = storage + .open_state() + .expect("load open state") + .expect("open state exists"); + storage + .append_ingested_safe_inputs_with_timestamp( + 789, + 789, + &[IngestedSafeInput { + sender: Address::ZERO, + payload: vec![0xbb], + block_number: 789, + block_timestamp: 1_700_000_001, + transaction_hash: B256::repeat_byte(0xef), + }], + Address::ZERO, + &sequencer_core::protocol::ProtocolTiming { + max_wait_blocks: sequencer_core::MAX_WAIT_BLOCKS, + preemptive_margin_blocks: 75, + l1_read_stale_after_blocks: 900, + seconds_per_block: 12, + }, + FrontierMode::Populate, + ) + .expect("append second direct input"); + storage + .close_frame_only(&mut head, 789, SafeInputRange::new(1, 2)) + .expect("close frame with second direct input"); +} diff --git a/sequencer/src/http.rs b/sequencer/src/http.rs index f9147f0..7aa4127 100644 --- a/sequencer/src/http.rs +++ b/sequencer/src/http.rs @@ -144,8 +144,10 @@ const DEFAULT_WS_MAX_SUBSCRIBERS: usize = 64; const DEFAULT_WS_MAX_CATCHUP_EVENTS: u64 = 50_000; const DEFAULT_MAX_BODY_BYTES: usize = TxRequest::MAX_JSON_BYTES_RECOMMENDED; -/// Reason returned in the WS Close frame when the subscriber's requested +/// Stable prefix of the WS Close-frame reason when the subscriber's requested /// `from_offset` is too old for the catch-up window to bridge. +/// +/// The full reason is `{WS_CATCHUP_WINDOW_EXCEEDED_REASON}: live_start_offset=`. pub const WS_CATCHUP_WINDOW_EXCEEDED_REASON: &str = "catch-up window exceeded"; pub type ApiServerTask = tokio::task::JoinHandle>; diff --git a/sequencer/tests/ws_broadcaster.rs b/sequencer/tests/ws_broadcaster.rs index 817471b..2bb2c91 100644 --- a/sequencer/tests/ws_broadcaster.rs +++ b/sequencer/tests/ws_broadcaster.rs @@ -219,12 +219,13 @@ async fn ws_subscribe_rejects_when_subscriber_limit_is_reached() { async fn ws_subscribe_closes_when_catchup_window_exceeds_limit() { let db = temp_db("ws-catchup-limit"); seed_ordered_txs(db.path.as_str()); + append_drained_direct_input(db.path.as_str(), vec![0xbb]); let Some(runtime) = start_test_server_with_limits(db.path.as_str(), 64, 1).await else { return; }; - let url = ws_subscribe_url(runtime.addr, 0); + let url = ws_subscribe_url(runtime.addr, 1); let (mut ws, _) = tokio::time::timeout(Duration::from_secs(5), connect_async(url)) .await .expect("timeout connecting websocket") @@ -237,10 +238,18 @@ async fn ws_subscribe_closes_when_catchup_window_exceeds_limit() { close_frame.code, tokio_tungstenite::tungstenite::protocol::frame::coding::CloseCode::Policy ); - assert_eq!( - close_frame.reason, - format!("{WS_CATCHUP_WINDOW_EXCEEDED_REASON}: live_start_offset=2") + let prefix = format!("{WS_CATCHUP_WINDOW_EXCEEDED_REASON}: live_start_offset="); + let reason = close_frame.reason.as_str(); + assert!( + reason.starts_with(WS_CATCHUP_WINDOW_EXCEEDED_REASON), + "close reason must retain the stable prefix: {reason}" ); + let live_start_offset = reason + .strip_prefix(prefix.as_str()) + .expect("close reason carries live_start_offset") + .parse::() + .expect("live_start_offset is a u64"); + assert_eq!(live_start_offset, 3); } other => panic!("expected close frame for catch-up limit, got {other:?}"), } diff --git a/tests/e2e/src/test_cases.rs b/tests/e2e/src/test_cases.rs index d7874a4..e16a905 100644 --- a/tests/e2e/src/test_cases.rs +++ b/tests/e2e/src/test_cases.rs @@ -715,6 +715,11 @@ async fn run_restart_and_replay_test(runtime: &mut ManagedSequencer) -> Scenario }, 3, ); + assert_eq!( + replay_after_restart.last_executed_safe_block(), + replay_before_restart.last_executed_safe_block(), + "mirror safe-block clock must match the pre-restart live replay clock", + ); Ok(()) } @@ -3267,6 +3272,11 @@ async fn run_replay_matches_live_for_mixed_workload_test( replay_post.executed_input_count(), replay_live.executed_input_count(), ); + assert_eq!( + replay_post.last_executed_safe_block(), + replay_live.last_executed_safe_block(), + "mirror safe-block clock must match the live replay clock", + ); Ok(()) } diff --git a/tests/harness/src/replay.rs b/tests/harness/src/replay.rs index 747dc19..ed88ff5 100644 --- a/tests/harness/src/replay.rs +++ b/tests/harness/src/replay.rs @@ -21,6 +21,16 @@ impl ReplayWalletApp { } pub fn apply(&mut self, message: WsTxMessage) -> HarnessResult<()> { + if let WsTxMessage::UserOp { sender, nonce, .. } = &message { + let sender = decode_address(sender.as_str()); + let expected = self.app.current_user_nonce(sender); + if *nonce != expected { + return Err(std::io::Error::other(format!( + "WS user-op nonce mismatch for {sender}: expected {expected}, got {nonce}" + )) + .into()); + } + } apply_ws_message(&mut self.app, message) } @@ -35,6 +45,10 @@ impl ReplayWalletApp { pub fn executed_input_count(&self) -> u64 { self.app.executed_input_count() } + + pub fn last_executed_safe_block(&self) -> u64 { + self.app.last_executed_safe_block() + } } pub(crate) fn apply_ws_message(