From 897a98fcea391c3a93083739f10179b64ee9a4e0 Mon Sep 17 00:00:00 2001 From: Aditya Kulkarni Date: Fri, 17 Jul 2026 11:29:43 -0500 Subject: [PATCH] cleanup --- crates/l1/src/lib.rs | 2 +- crates/l1/src/subscriber.rs | 289 ++++++++++++++++++++++++++------- crates/l1/src/tests.rs | 115 ++++++++++++- crates/node/src/node.rs | 56 ++++--- crates/node/src/replication.rs | 230 ++++++++++++++++++++++++-- crates/node/tests/it/e2e.rs | 131 ++++++++++++++- crates/node/tests/it/utils.rs | 23 ++- 7 files changed, 753 insertions(+), 93 deletions(-) diff --git a/crates/l1/src/lib.rs b/crates/l1/src/lib.rs index ad8033386..9254c8a18 100644 --- a/crates/l1/src/lib.rs +++ b/crates/l1/src/lib.rs @@ -94,7 +94,7 @@ pub use event::{EnabledToken, L1PortalEvents, L1SequencerEvent}; pub use ext::{ChainTempoStateExt, TempoStateExt}; pub use queue::DepositQueue; pub use state::{L1StateCache, PolicyCache, PolicyProvider}; -pub use subscriber::{L1Subscriber, L1SubscriberConfig}; +pub use subscriber::{L1BlockObserver, L1Subscriber, L1SubscriberConfig}; pub(crate) use event::EnqueueOutcome; diff --git a/crates/l1/src/subscriber.rs b/crates/l1/src/subscriber.rs index d6cab03a1..6dbca6c4a 100644 --- a/crates/l1/src/subscriber.rs +++ b/crates/l1/src/subscriber.rs @@ -1,5 +1,116 @@ use super::*; +use std::collections::BTreeMap; + +/// A handle to L1 blocks whose headers and receipts have been independently +/// validated and whose L1-derived events have been applied to the local caches. +/// +/// Followers use this handle to gate zone-block import on the exact L1 anchor +/// embedded in `advanceTempo`. Recording happens only after cache updates, so a +/// successful wait also establishes that execution-visible L1 data is ready. +#[derive(Debug, Clone)] +pub struct L1BlockObserver { + observed: Arc>>, + changed: tokio::sync::watch::Sender<()>, +} + +impl Default for L1BlockObserver { + fn default() -> Self { + let (changed, _) = tokio::sync::watch::channel(()); + Self { + observed: Default::default(), + changed, + } + } +} + +impl L1BlockObserver { + /// Return the independently observed hash at `number`, if available. + pub fn observed_hash(&self, number: u64) -> Option { + self.observed.read().get(&number).copied() + } + + /// Return the highest independently observed L1 anchor. + pub fn latest(&self) -> Option { + self.observed + .read() + .last_key_value() + .map(|(&number, &hash)| NumHash::new(number, hash)) + } + + /// Wait until the exact L1 block has been validated and applied locally. + /// + /// Fails immediately if a different hash is observed at the requested + /// height (the requested block is not canonical in this observer) or if + /// the height has already been pruned below the retained range. + pub async fn wait_for(&self, block: NumHash) -> eyre::Result<()> { + let mut changed = self.changed.subscribe(); + loop { + { + let observed = self.observed.read(); + match observed.get(&block.number) { + Some(&hash) if hash == block.hash => return Ok(()), + Some(&hash) => { + eyre::bail!( + "observed different L1 hash at block {}: expected {}, got {}", + block.number, + block.hash, + hash + ) + } + None => { + if let Some((&earliest, _)) = observed.first_key_value() + && earliest > block.number + { + eyre::bail!( + "L1 block {} is below the observer's retained range \ + (earliest {earliest})", + block.number + ) + } + } + } + } + changed + .changed() + .await + .expect("observer sender is retained"); + } + } + + /// Record an independently validated and applied L1 anchor. + /// + /// Only the L1 subscriber should record anchors. + /// recording promises that all L1-derived cache updates through this block + /// are visible. + pub fn record(&self, block: NumHash) { + let mut observed = self.observed.write(); + if observed.get(&block.number) == Some(&block.hash) { + return; + } + + // Replacing an anchor invalidates all descendants recorded from the + // previous branch. + observed.split_off(&block.number); + observed.insert(block.number, block.hash); + drop(observed); + self.changed.send_replace(()); + } + + /// Drop anchors below `number`; imports have advanced past them so they + /// can no longer be waited on. Bounds the observer's memory. + pub fn prune_below(&self, number: u64) { + let mut observed = self.observed.write(); + let retained = observed.split_off(&number); + *observed = retained; + } + + fn clear(&self) { + self.observed.write().clear(); + self.changed.send_replace(()); + } +} + /// Poll interval for the HTTP block filter fallback (500ms, matching L1 block time). const HTTP_POLL_INTERVAL: std::time::Duration = std::time::Duration::from_millis(500); @@ -21,6 +132,9 @@ pub struct L1SubscriberConfig { /// Shared L1 state cache. The subscriber updates the cache anchor on each /// confirmed block and clears it on reorgs. pub l1_state_cache: crate::state::cache::L1StateCache, + /// Shared record of validated and applied L1 anchors. Followers gate + /// zone-block import on it; the subscriber records each processed block. + pub block_observer: L1BlockObserver, /// Maximum number of concurrent L1 RPC receipt fetches. Used directly for /// the live stream and halved for backfill (which sends 2 requests per block). pub l1_fetch_concurrency: usize, @@ -51,7 +165,9 @@ where pub struct L1Subscriber { pub(crate) config: L1SubscriberConfig, pub(crate) local_state: Arc, - pub(crate) deposit_queue: DepositQueue, + /// Leader-only sink. Followers observe and apply L1 data without retaining + /// deposit payloads. + pub(crate) deposit_queue: Option, /// Mutable set of token addresses tracked for TIP-403 policy events. /// Initialized from config, grows dynamically when `TokenEnabled` events are seen. pub(crate) tracked_tokens: Vec
, @@ -74,6 +190,35 @@ impl L1Subscriber { task_executor: reth_tasks::Runtime, ) where P: StateProviderFactory + Clone + Send + Sync + 'static, + { + Self::spawn_inner( + config, + local_state_provider, + Some(deposit_queue), + task_executor, + ); + } + + /// Spawn the shared L1 observation path without the leader-only deposit + /// enqueueing sink. Followers gate zone-block import on the config's + /// [`L1BlockObserver`]. + pub fn spawn_observer

( + config: L1SubscriberConfig, + local_state_provider: P, + task_executor: reth_tasks::Runtime, + ) where + P: StateProviderFactory + Clone + Send + Sync + 'static, + { + Self::spawn_inner(config, local_state_provider, None, task_executor) + } + + fn spawn_inner

( + config: L1SubscriberConfig, + local_state_provider: P, + deposit_queue: Option, + task_executor: reth_tasks::Runtime, + ) where + P: StateProviderFactory + Clone + Send + Sync + 'static, { let tracked_tokens = config.policy_cache.read().tracked_tokens(); let subscriber = Self { @@ -88,7 +233,7 @@ impl L1Subscriber { }; task_executor.spawn_critical_task( - "l1-deposit-subscriber", + "l1-block-subscriber", Box::pin(async move { loop { if let Err(e) = subscriber.clone().run().await { @@ -278,7 +423,7 @@ impl L1Subscriber { Ok(Some(on_chain + 1)) } - /// Backfill deposit events from the starting block to the current L1 tip. + /// Backfill validated L1 data from the starting block to the current L1 tip. #[instrument(skip(self, l1_provider))] async fn sync_to_l1_tip( &mut self, @@ -290,17 +435,21 @@ impl L1Subscriber { }; // Skip past blocks already in the queue from a previous `run()`. - if let Some(last) = self.deposit_queue.last_enqueued() { - let adjusted = last.number + 1; - if adjusted > from { - info!( - portal_from = from, - queue_last = last.number, - adjusted_from = adjusted, - "Skipping blocks already in deposit queue" - ); + if let Some(deposit_queue) = &self.deposit_queue { + if let Some(last) = deposit_queue.last_enqueued() { + let adjusted = last.number + 1; + if adjusted > from { + info!( + portal_from = from, + queue_last = last.number, + adjusted_from = adjusted, + "Skipping blocks already in deposit queue" + ); + } + from = from.max(adjusted); } - from = from.max(adjusted); + } else if let Some(last) = self.config.block_observer.latest() { + from = from.max(last.number + 1); } let tip = l1_provider.get_block_number().await?; @@ -311,12 +460,7 @@ impl L1Subscriber { return Ok(()); } - info!( - from, - tip, - blocks = tip - from + 1, - "Backfilling deposit events" - ); + info!(from, tip, blocks = tip - from + 1, "Backfilling L1 blocks"); let start = std::time::Instant::now(); let result = self.backfill(l1_provider, from, tip).await; self.subscriber_metrics @@ -332,8 +476,9 @@ impl L1Subscriber { /// /// Fetches headers and receipts for up to `l1_fetch_concurrency` blocks in /// parallel, then processes them sequentially (event extraction, policy - /// application, enqueue). Receipts are fetched by the corresponding block - /// hash and validated against the header's receipts root before processing. + /// application, and optional leader enqueue). Receipts are fetched by the + /// corresponding block hash and validated against the header's receipts + /// root before processing. #[instrument(skip(self, l1_provider), fields(from, to))] async fn backfill( &mut self, @@ -393,7 +538,7 @@ impl L1Subscriber { }) .buffered(concurrency); - let mut enqueued = 0u64; + let mut processed = 0u64; let backfill_start = std::time::Instant::now(); while let Some((header, receipts)) = fetched.try_next().await? { @@ -405,16 +550,19 @@ impl L1Subscriber { self.update_l1_state_anchor(block_number, sealed.hash(), sealed.parent_hash()); self.apply_policy_events(block_number, &policy_events); self.apply_portal_state_events(block_number, &events); - self.deposit_queue - .enqueue_sealed(sealed, events, policy_events); - enqueued += 1; - self.subscriber_metrics.blocks_enqueued.increment(1); + self.config.block_observer.record(sealed.num_hash()); + processed += 1; + if let Some(deposit_queue) = &self.deposit_queue { + deposit_queue.enqueue_sealed(sealed, events, policy_events); + self.subscriber_metrics.blocks_enqueued.increment(1); + } - if enqueued.is_multiple_of(100) { + if processed.is_multiple_of(100) { + self.prune_observed_anchors(); let elapsed = backfill_start.elapsed(); - let blocks_per_sec = enqueued as f64 / elapsed.as_secs_f64().max(0.001); + let blocks_per_sec = processed as f64 / elapsed.as_secs_f64().max(0.001); info!( - enqueued, + processed, current_block = block_number, target = to, remaining = to - block_number, @@ -437,10 +585,10 @@ impl L1Subscriber { /// Run the L1 subscriber until the stream ends or an error occurs. /// - /// Connects to the L1 node (HTTP or WebSocket), backfills deposit events - /// to the current L1 tip, then listens for new block headers. Each block — - /// with or without deposits — is enqueued so the zone engine sees a strict - /// sequential chain. + /// Connects to the L1 node (HTTP or WebSocket), backfills validated L1 data + /// to the current tip, then listens for new block headers. Leaders enqueue + /// each block for the zone engine; observers only update L1-derived caches + /// and exact-hash anchors. /// /// Live-streamed blocks are buffered one block behind: a block is only /// flushed to the deposit queue once the next block arrives with a @@ -492,35 +640,51 @@ impl L1Subscriber { // Confirmed — update the L1 state anchor, apply events, and // flush to the queue. let tip_number = tip_header.number(); + if let Some(observed) = self.config.block_observer.latest() + && tip_number > observed.number + 1 + { + let from = observed.number + 1; + warn!( + from, + to = tip_number, + "Backfilling gap in independently observed L1 blocks" + ); + self.backfill(&provider, from, tip_number).await?; + unconfirmed_tip = Some((sealed, events, policy_events)); + continue; + } let tip_hash = tip_header.hash(); let tip_parent = tip_header.parent_hash(); self.update_l1_state_anchor(tip_number, tip_hash, tip_parent); self.apply_policy_events(tip_number, &tip_policy_events); self.apply_portal_state_events(tip_number, &tip_events); - match self - .deposit_queue - .try_enqueue(tip_header, tip_events, tip_policy_events) - { - EnqueueOutcome::Accepted => { - self.subscriber_metrics.blocks_enqueued.increment(1); - } - EnqueueOutcome::Duplicate => {} - EnqueueOutcome::NeedBackfill { from, to } => { - // Gap between queue head and confirmed tip — backfill - // the missing range including the tip (re-fetched from - // the provider since try_enqueue consumed ownership). - warn!( - from, - to, - tip = tip_number, - "Backfilling gap before confirmed tip" - ); - self.backfill(&provider, from, tip_number).await?; + self.config.block_observer.record(tip_header.num_hash()); + self.prune_observed_anchors(); + if let Some(deposit_queue) = &self.deposit_queue { + match deposit_queue.try_enqueue(tip_header, tip_events, tip_policy_events) { + EnqueueOutcome::Accepted => { + self.subscriber_metrics.blocks_enqueued.increment(1); + } + EnqueueOutcome::Duplicate => {} + EnqueueOutcome::NeedBackfill { from, to } => { + // Gap between queue head and confirmed tip — backfill + // the missing range including the tip (re-fetched from + // the provider since try_enqueue consumed ownership). + warn!( + from, + to, + tip = tip_number, + "Backfilling gap before confirmed tip" + ); + self.backfill(&provider, from, tip_number).await?; + } } } } else { - // Reorg — discard the buffered tip and clear L1 state and - // policy caches. + // Reorg of the unconfirmed tip. None of that tip's events + // were applied, so retain the confirmed caches/anchors. If + // the replacement creates a height gap, the next confirmed + // header takes the backfill path above. self.subscriber_metrics.reorgs_detected.increment(1); warn!( discarded_block = tip_header.number(), @@ -529,8 +693,6 @@ impl L1Subscriber { new_parent = %sealed.parent_hash(), "Discarding unconfirmed L1 block (reorg)" ); - self.config.l1_state_cache.write().clear(); - self.config.policy_cache.write().clear(); } } @@ -693,9 +855,24 @@ impl L1Subscriber { ); guard.clear(); self.config.policy_cache.write().clear(); + self.config.block_observer.clear(); } guard.update_anchor(NumHash::new(number, hash)); } + + /// Drop observed anchors below the locally executed Tempo checkpoint. + /// + /// The zone's `tempoBlockNumber` only advances once the corresponding L1 + /// anchor has been consumed (block production on leaders, block import on + /// followers), so anchors below it can no longer be waited on. + fn prune_observed_anchors(&self) { + match self.local_state.latest_tempo_block_number() { + Ok(checkpoint) => self.config.block_observer.prune_below(checkpoint), + Err(err) => { + debug!(%err, "Skipping observer pruning; local Tempo checkpoint unavailable") + } + } + } } /// Fetch receipts for the L1 header by block hash and verify they match the diff --git a/crates/l1/src/tests.rs b/crates/l1/src/tests.rs index 4c792ea78..7580e3b56 100644 --- a/crates/l1/src/tests.rs +++ b/crates/l1/src/tests.rs @@ -149,11 +149,12 @@ fn test_subscriber( genesis_tempo_block_number, policy_cache: crate::PolicyCache::default(), l1_state_cache: crate::L1StateCache::new(HashSet::from([portal_address])), + block_observer: L1BlockObserver::default(), l1_fetch_concurrency: 1, retry_connection_interval: Duration::from_secs(1), }, local_state, - deposit_queue: DepositQueue::default(), + deposit_queue: Some(DepositQueue::default()), tracked_tokens: vec![], tip403_metrics: Default::default(), subscriber_metrics: Default::default(), @@ -451,6 +452,118 @@ fn update_l1_state_anchor_reorg_clears_stale_policy_state() { ); } +#[tokio::test] +async fn l1_block_observer_waits_for_exact_hash_and_invalidates_descendants() { + let observer = L1BlockObserver::default(); + let hash_10 = B256::with_last_byte(0x10); + let hash_11 = B256::with_last_byte(0x11); + observer.record(NumHash::new(10, hash_10)); + observer.record(NumHash::new(11, hash_11)); + + observer.wait_for(NumHash::new(11, hash_11)).await.unwrap(); + assert!( + observer + .wait_for(NumHash::new(11, B256::with_last_byte(0xff))) + .await + .unwrap_err() + .to_string() + .contains("observed different L1 hash") + ); + + let replacement_10 = B256::with_last_byte(0xa0); + observer.record(NumHash::new(10, replacement_10)); + assert_eq!(observer.observed_hash(10), Some(replacement_10)); + assert_eq!(observer.observed_hash(11), None); +} + +#[tokio::test] +async fn l1_block_observer_prune_drops_consumed_anchors() { + let observer = L1BlockObserver::default(); + for number in 10..=14 { + observer.record(NumHash::new(number, B256::with_last_byte(number as u8))); + } + + observer.prune_below(12); + assert_eq!(observer.observed_hash(11), None); + assert_eq!(observer.observed_hash(12), Some(B256::with_last_byte(12))); + assert_eq!( + observer.latest(), + Some(NumHash::new(14, B256::with_last_byte(14))) + ); + + // Waiting on a pruned height fails instead of hanging. + assert!( + observer + .wait_for(NumHash::new(11, B256::with_last_byte(11))) + .await + .unwrap_err() + .to_string() + .contains("below the observer's retained range") + ); +} + +#[tokio::test] +async fn observed_policy_change_is_visible_at_its_exact_l1_height() { + use crate::state::tip403::AuthRole; + use tempo_contracts::precompiles::ITIP403Registry::PolicyType; + + let subscriber = test_subscriber( + Arc::new(SequenceLocalTempoCheckpointReader::new([0])), + Some(0), + ); + let token = address!("0x0000000000000000000000000000000000000011"); + let user = address!("0x0000000000000000000000000000000000000022"); + { + let mut cache = subscriber.config.policy_cache.write(); + cache.set_policy_type(2, PolicyType::WHITELIST); + cache.set_token_policy(token, 10, 2); + } + + subscriber.apply_policy_events( + 10, + &[PolicyEvent::MembershipChanged { + policy_id: 2, + account: user, + in_set: true, + }], + ); + let hash_10 = B256::with_last_byte(0x10); + subscriber + .config + .block_observer + .record(NumHash::new(10, hash_10)); + + subscriber.apply_policy_events( + 11, + &[PolicyEvent::MembershipChanged { + policy_id: 2, + account: user, + in_set: false, + }], + ); + let hash_11 = B256::with_last_byte(0x11); + subscriber + .config + .block_observer + .record(NumHash::new(11, hash_11)); + subscriber + .config + .block_observer + .wait_for(NumHash::new(11, hash_11)) + .await + .unwrap(); + + let cache = subscriber.config.policy_cache.read(); + assert_eq!( + cache.is_authorized(token, user, 10, AuthRole::Transfer), + Some(true) + ); + assert_eq!( + cache.is_authorized(token, user, 11, AuthRole::Transfer), + Some(false) + ); +} + /// Confirm the front of the queue, panicking if it fails. fn confirm(queue: &mut PendingDeposits) -> L1BlockDeposits { let num_hash = queue.peek().expect("queue is empty").header.num_hash(); diff --git a/crates/node/src/node.rs b/crates/node/src/node.rs index 0f1038e9e..f212ed8fd 100644 --- a/crates/node/src/node.rs +++ b/crates/node/src/node.rs @@ -62,7 +62,7 @@ use tracing::{debug, info, warn}; use zone_chainspec::ZoneChainSpec; use zone_evm::ZoneEvmConfig; use zone_l1::{ - DepositQueue, L1Subscriber, L1SubscriberConfig, PolicyCache, TempoStateExt, + DepositQueue, L1BlockObserver, L1Subscriber, L1SubscriberConfig, PolicyCache, TempoStateExt, state::{ L1StateCache, L1StateProvider, L1StateProviderConfig, PolicyProvider, spawn_policy_resolution_task, spawn_pool_prefetch_task, @@ -214,6 +214,7 @@ impl ZoneNode { genesis_tempo_block_number, policy_cache: policy_cache.clone(), l1_state_cache: l1_state_cache.clone(), + block_observer: L1BlockObserver::default(), l1_fetch_concurrency, retry_connection_interval, }; @@ -309,6 +310,11 @@ impl ZoneNode { self.policy_cache.clone() } + /// Returns the shared record of independently observed L1 anchors. + pub fn l1_block_observer(&self) -> L1BlockObserver { + self.l1_config.block_observer.clone() + } + /// Returns a [`ComponentsBuilder`] configured for a Zone node. pub fn components( executor_builder: ZoneExecutorBuilder, @@ -467,17 +473,7 @@ where .erased(); self.resolve_and_seed_tokens(&l1_provider).await?; - let p2p_role = self.p2p_config.as_ref().map(P2pConfig::role); - if p2p_role == Some(Role::Follower) { - // TODO(multi-sequencer): Split L1 observation/cache updates from deposit - // enqueueing. Followers import complete blocks from the leader and do not consume - // DepositQueue; starting the unified subscriber here would grow that queue forever. - // On promotion/restart the subscriber resumes from the tempoBlockNumber persisted in - // the follower's imported zone state. - info!(target: "reth::cli", "Skipping L1 deposit subscriber on follower"); - } else { - self.spawn_l1_subscriber(&ctx); - } + self.spawn_l1_subscriber(&ctx); self.spawn_policy_tasks(&l1_provider, &ctx); let task_executor = ctx.node.task_executor().clone(); @@ -490,6 +486,8 @@ where &task_executor, ctx.node.provider().clone(), ctx.beacon_engine_handle.clone(), + self.l1_config.block_observer.clone(), + self.policy_cache.clone(), )?; } @@ -546,6 +544,8 @@ where task_executor: &reth_tasks::TaskExecutor, provider: N::Provider, engine: reth_node_builder::ConsensusEngineHandle, + l1_observer: L1BlockObserver, + policy_cache: PolicyCache, ) -> eyre::Result<()> { let role = config.role(); let handle = spawn_p2p(config, network_id)?; @@ -572,7 +572,14 @@ where // for later ACK/backfill commands even though followers send nothing in this PR. task_executor.spawn_critical_task( "zone-p2p-block-import", - import_leader_blocks(provider, engine, events, commands), + import_leader_blocks( + provider, + engine, + events, + commands, + l1_observer, + policy_cache, + ), ); } } @@ -671,13 +678,22 @@ where /// Spawn the L1 subscriber. Listens for new blocks and deposit events. fn spawn_l1_subscriber(&mut self, ctx: &AddOnsContext<'_, N>) { - L1Subscriber::spawn( - self.l1_config.clone(), - ctx.node.provider().clone(), - self.deposit_queue.clone(), - ctx.node.task_executor().clone(), - ); - info!(target: "reth::cli", "Unified L1 subscriber started"); + if self.p2p_config.as_ref().map(P2pConfig::role) == Some(Role::Follower) { + L1Subscriber::spawn_observer( + self.l1_config.clone(), + ctx.node.provider().clone(), + ctx.node.task_executor().clone(), + ); + info!(target: "reth::cli", "L1 observer started without deposit enqueueing"); + } else { + L1Subscriber::spawn( + self.l1_config.clone(), + ctx.node.provider().clone(), + self.deposit_queue.clone(), + ctx.node.task_executor().clone(), + ); + info!(target: "reth::cli", "Leader L1 observer and deposit sink started"); + } } /// Spawn TIP-403 policy resolution and pool prefetch tasks. diff --git a/crates/node/src/replication.rs b/crates/node/src/replication.rs index da3ecd453..0201c10a0 100644 --- a/crates/node/src/replication.rs +++ b/crates/node/src/replication.rs @@ -4,17 +4,23 @@ use alloy_consensus::BlockHeader as _; use alloy_primitives::B256; use alloy_rlp::Decodable as _; use alloy_rpc_types_engine::ForkchoiceState; +use alloy_sol_types::SolCall as _; use futures::{StreamExt as _, stream::BoxStream}; use reth_chain_state::PersistedBlockSubscriptions; use reth_node_api::PayloadTypes as _; use reth_node_builder::ConsensusEngineHandle; -use reth_primitives_traits::SealedBlock; -use reth_storage_api::{BlockNumReader, BlockReader, HeaderProvider}; -use tempo_primitives::{Block, TempoHeader}; +use reth_primitives_traits::{SealedBlock, SealedHeader}; +use reth_storage_api::{BlockNumReader, BlockReader, HeaderProvider, StateProviderFactory}; +use std::time::Duration; +use tempo_primitives::{Block, TempoHeader, TempoTxEnvelope}; use tokio::sync::mpsc; -use tracing::{debug, info}; +use tracing::{debug, info, warn}; +use zone_l1::{L1BlockObserver, PolicyCache, TempoStateExt as _}; use zone_p2p::{P2pCommand, P2pEvent}; -use zone_payload::ZonePayloadTypes; +use zone_payload::{ + ZonePayloadTypes, + abi::{ZONE_INBOX_ADDRESS, ZoneInbox}, +}; #[derive(Debug, Clone, Copy)] pub(crate) struct PersistedTip { @@ -162,14 +168,24 @@ pub(crate) async fn import_leader_blocks

( engine: ConsensusEngineHandle, mut events: mpsc::Receiver, _commands: mpsc::Sender, + l1_observer: L1BlockObserver, + policy_cache: PolicyCache, ) where - P: BlockNumReader + HeaderProvider

+ Clone + Send + Sync + 'static, + P: StateProviderFactory + + BlockNumReader + + HeaderProvider
+ + Clone + + Send + + Sync + + 'static, { while let Some(event) = events.recv().await { let P2pEvent::BlockReceived { block, .. } = event else { continue; }; - if let Err(err) = import_leader_block(&provider, &engine, &block).await { + if let Err(err) = + import_leader_block(&provider, &engine, &l1_observer, &policy_cache, &block).await + { tracing::error!(target: "zone::p2p", %err, "Rejected leader block"); } } @@ -179,10 +195,18 @@ pub(crate) async fn import_leader_blocks

( async fn import_leader_block

( provider: &P, engine: &ConsensusEngineHandle, + l1_observer: &L1BlockObserver, + policy_cache: &PolicyCache, encoded: &[u8], ) -> eyre::Result<()> where - P: BlockNumReader + HeaderProvider

+ Clone + Send + Sync + 'static, + P: StateProviderFactory + + BlockNumReader + + HeaderProvider
+ + Clone + + Send + + Sync + + 'static, { // Check the received block let mut input = encoded; @@ -231,14 +255,57 @@ where ); } - // 3. All txns in the block execute properly + // 3. The block's advanceTempo system tx advances the local Tempo + // checkpoint by exactly one L1 block, and that L1 block has been + // independently observed (header + receipts validated, policy and L1 + // state caches updated) before we execute against it. + let l1_header = decode_advance_tempo_header(&block)?; + let local = provider.latest()?.tempo_num_hash()?; + if l1_header.number() != local.number + 1 { + eyre::bail!( + "leader block {block_number} advances Tempo to L1 block {}, but the local checkpoint \ + is {}; expected {}", + l1_header.number(), + local.number, + local.number + 1 + ); + } + if l1_header.parent_hash() != local.hash { + eyre::bail!( + "advanceTempo L1 header {} does not extend the local Tempo checkpoint: \ + embedded parent {}, local hash {}", + l1_header.number(), + l1_header.parent_hash(), + local.hash + ); + } + let anchor = l1_header.num_hash(); + loop { + match tokio::time::timeout(Duration::from_secs(30), l1_observer.wait_for(anchor)).await { + Ok(observed) => break observed?, + Err(_elapsed) => warn!( + target: "zone::p2p", + block_number, + l1_block = anchor.number, + l1_hash = ?anchor.hash, + "Leader block import is waiting for local L1 observation of its anchor" + ), + } + } + + // 4. All txns in the block execute properly let payload = ZonePayloadTypes::block_to_payload(block, None); let status = engine.new_payload(payload).await?; if !status.is_valid() { eyre::bail!("execution engine rejected leader block {block_number} ({hash}): {status:?}"); } - // 4. Forkchoice + // Mirror the leader engine: fold observed policy deltas up to the consumed + // anchor so the next block's execution resolves policy state at its + // parent's L1 height. + policy_cache.advance(anchor.number); + + // 5. Forkchoice let forkchoice = ForkchoiceState::same_hash(hash); let result = engine.fork_choice_updated(forkchoice, None).await?; if !result.is_valid() { @@ -251,6 +318,38 @@ where Ok(()) } +/// Decode the L1 header embedded in the block's first system transaction +/// (`ZoneInbox.advanceTempo`). +fn decode_advance_tempo_header( + block: &SealedBlock, +) -> eyre::Result> { + // Do some basic checks on the `advanceTempo` txn + let first_tx = block.body().transactions().next().ok_or_else(|| { + eyre::eyre!("leader block has no transactions; expected an advanceTempo system tx") + })?; + let TempoTxEnvelope::Legacy(signed) = first_tx else { + eyre::bail!("first transaction in leader block is not a legacy system transaction"); + }; + if !first_tx.is_system_tx() { + eyre::bail!("first transaction in leader block is not a Tempo system transaction"); + } + if signed.tx().to != ZONE_INBOX_ADDRESS.into() { + eyre::bail!("first Tempo system transaction is not sent to ZoneInbox"); + } + let call = ZoneInbox::advanceTempoCall::abi_decode(signed.tx().input.as_ref()) + .map_err(|err| eyre::eyre!("first transaction does not decode as advanceTempo: {err}"))?; + let mut header_rlp = call.header.as_ref(); + let header = TempoHeader::decode(&mut header_rlp) + .map_err(|err| eyre::eyre!("invalid RLP-encoded L1 header in advanceTempo: {err}"))?; + if !header_rlp.is_empty() { + eyre::bail!( + "advanceTempo L1 header has {} trailing bytes", + header_rlp.len() + ); + } + Ok(SealedHeader::seal_slow(header)) +} + #[cfg(test)] mod tests { use std::sync::{ @@ -296,6 +395,117 @@ mod tests { } } + #[test] + fn decodes_advance_tempo_header_from_first_system_tx() { + use alloy_consensus::BlockHeader as _; + use reth_primitives_traits::{SealedBlock, SealedHeader}; + use tempo_primitives::{Block, TempoHeader}; + + let l1_header = TempoHeader { + inner: alloy_consensus::Header { + number: 7, + parent_hash: B256::repeat_byte(0x42), + ..Default::default() + }, + ..Default::default() + }; + let prepared = zone_l1::PreparedL1Block { + header: SealedHeader::seal_slow(l1_header), + queued_deposits: vec![], + decryptions: vec![], + enabled_tokens: vec![], + }; + let tx = zone_payload::build_advance_tempo_tx(&prepared); + + let block = SealedBlock::seal_slow(Block { + header: TempoHeader::default(), + body: alloy_consensus::BlockBody { + transactions: vec![tx.into_inner()], + ommers: vec![], + withdrawals: None, + }, + }); + + let decoded = super::decode_advance_tempo_header(&block).unwrap(); + assert_eq!(decoded.number(), 7); + assert_eq!(decoded.parent_hash(), B256::repeat_byte(0x42)); + assert_eq!(decoded.hash(), prepared.header.hash()); + } + + #[test] + fn rejects_advance_tempo_calldata_not_sent_to_zone_inbox() { + use alloy_consensus::{Signed, TxLegacy}; + use alloy_primitives::{Address, Bytes, U256}; + use alloy_rlp::Encodable as _; + use alloy_sol_types::SolCall as _; + use reth_primitives_traits::SealedBlock; + use tempo_primitives::{ + Block, TempoHeader, TempoTxEnvelope, transaction::envelope::TEMPO_SYSTEM_TX_SIGNATURE, + }; + + let l1_header = TempoHeader { + inner: alloy_consensus::Header { + number: 7, + parent_hash: B256::repeat_byte(0x42), + ..Default::default() + }, + ..Default::default() + }; + let mut header_rlp = Vec::new(); + l1_header.encode(&mut header_rlp); + let calldata = zone_payload::abi::ZoneInbox::advanceTempoCall { + header: Bytes::from(header_rlp), + deposits: vec![], + decryptions: vec![], + enabledTokens: vec![], + } + .abi_encode(); + + let tx = TxLegacy { + chain_id: None, + nonce: 0, + gas_price: 0, + gas_limit: 100_000, + to: Address::repeat_byte(0x99).into(), + value: U256::ZERO, + input: calldata.into(), + }; + + let block = SealedBlock::seal_slow(Block { + header: TempoHeader::default(), + body: alloy_consensus::BlockBody { + transactions: vec![TempoTxEnvelope::Legacy(Signed::new_unhashed( + tx, + TEMPO_SYSTEM_TX_SIGNATURE, + ))], + ommers: vec![], + withdrawals: None, + }, + }); + + let err = super::decode_advance_tempo_header(&block) + .expect_err("advanceTempo calldata not sent to ZoneInbox must be rejected"); + assert!(err.to_string().contains("ZoneInbox")); + } + + #[test] + fn rejects_leader_block_without_advance_tempo_tx() { + use reth_primitives_traits::SealedBlock; + use tempo_primitives::{Block, TempoHeader}; + + let block = SealedBlock::seal_slow(Block { + header: TempoHeader::default(), + body: alloy_consensus::BlockBody { + transactions: vec![], + ommers: vec![], + withdrawals: None, + }, + }); + + let err = super::decode_advance_tempo_header(&block).unwrap_err(); + assert!(err.to_string().contains("no transactions")); + } + #[tokio::test] async fn broadcasts_block_persisted_during_startup_reconciliation_once() { let source = StartupRaceSource { diff --git a/crates/node/tests/it/e2e.rs b/crates/node/tests/it/e2e.rs index 89b94203e..e18a7f5a5 100644 --- a/crates/node/tests/it/e2e.rs +++ b/crates/node/tests/it/e2e.rs @@ -14,12 +14,14 @@ use alloy_provider::{DynProvider, Provider}; use alloy_rpc_types_eth::TransactionRequest; use alloy_sol_types::SolCall; use tempo_chainspec::spec::TEMPO_T0_BASE_FEE; +use tempo_contracts::precompiles::ITIP403Registry; use tempo_precompiles::PATH_USD_ADDRESS; use tempo_zone_contracts::{ TEMPO_STATE_ADDRESS, TempoState, Withdrawal, ZONE_INBOX_ADDRESS, ZONE_OUTBOX_ADDRESS, ZoneInbox, ZoneOutbox, }; use zone_l1::ChainTempoStateExt; +use zone_l1::state::tip403::PolicyEvent; use crate::utils::{ DEFAULT_POLL, DEFAULT_TIMEOUT, L1Fixture, WITHDRAWAL_TX_GAS, ZoneTestNode, approve_outbox, @@ -41,7 +43,8 @@ async fn test_p2p_follower_tracks_leader_balance() -> eyre::Result<()> { // first block. tokio::time::sleep(Duration::from_secs(1)).await; - fixture.inject_empty_block(leader.deposit_queue()); + let anchor = fixture.inject_empty_block(leader.deposit_queue()); + follower.l1_block_observer().record(anchor); leader.wait_for_block_number(1, DEFAULT_TIMEOUT).await?; follower.wait_for_block_number(1, DEFAULT_TIMEOUT).await?; @@ -49,7 +52,8 @@ async fn test_p2p_follower_tracks_leader_balance() -> eyre::Result<()> { let recipient = address!("0x0000000000000000000000000000000000005678"); let amount = 1_000_000_u128; let deposit = fixture.make_deposit(PATH_USD_ADDRESS, depositor, recipient, amount); - fixture.inject_deposits(leader.deposit_queue(), vec![deposit]); + let anchor = fixture.inject_deposits(leader.deposit_queue(), vec![deposit]); + follower.l1_block_observer().record(anchor); leader .wait_for_balance( @@ -72,6 +76,120 @@ async fn test_p2p_follower_tracks_leader_balance() -> eyre::Result<()> { Ok(()) } +/// TIP-403 policy changes between imported leader blocks affect follower +/// validation: the follower executes each leader block with the policy state +/// of that block's L1 anchor, not stale (or future) state. +/// +/// The whitelist switch at L1 height 2 makes the block-3 deposit to Bob a +/// refused mint. A follower still validating with the old allow-all state +/// would mint, diverge on state root, and fail the import — so the follower +/// reaching block 3 with Bob's balance at zero proves it applied the change. +/// The membership update at height 3 then proves the follow-up change lands +/// at its exact height too. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn test_p2p_follower_validates_with_policy_state_at_anchor() -> eyre::Result<()> { + reth_tracing::init_test_tracing(); + + let (leader, follower, mut fixture) = start_local_p2p_pair(10).await?; + + // Wait for peer dial/handshake before producing the first block. + tokio::time::sleep(Duration::from_secs(1)).await; + + let alice = address!("0x000000000000000000000000000000000000a11c"); + let bob = address!("0x0000000000000000000000000000000000000b0b"); + let amount = 1_000_000_u128; + + // Block 1: pathUSD still uses the seeded allow-all policy — the deposit + // mints to Alice on both nodes. + let deposit = fixture.make_deposit(PATH_USD_ADDRESS, alice, alice, amount); + let anchor = fixture.inject_deposits(leader.deposit_queue(), vec![deposit]); + follower.l1_block_observer().record(anchor); + leader + .wait_for_balance(PATH_USD_ADDRESS, alice, U256::from(amount), DEFAULT_TIMEOUT) + .await?; + follower + .wait_for_balance(PATH_USD_ADDRESS, alice, U256::from(amount), DEFAULT_TIMEOUT) + .await?; + + // At L1 height 2, switch pathUSD to whitelist policy 9 with Alice in and + // Bob explicitly out. Blocks anchored above height 2 execute with it. + let policy_change = [ + PolicyEvent::PolicyCreated { + policy_id: 9, + policy_type: ITIP403Registry::PolicyType::WHITELIST, + }, + PolicyEvent::MembershipChanged { + policy_id: 9, + account: alice, + in_set: true, + }, + PolicyEvent::MembershipChanged { + policy_id: 9, + account: bob, + in_set: false, + }, + PolicyEvent::TokenPolicyChanged { + token: PATH_USD_ADDRESS, + policy_id: 9, + }, + ]; + leader + .policy_cache() + .write() + .apply_events(2, &policy_change); + follower + .policy_cache() + .write() + .apply_events(2, &policy_change); + let anchor = fixture.inject_empty_block(leader.deposit_queue()); + follower.l1_block_observer().record(anchor); + leader.wait_for_block_number(2, DEFAULT_TIMEOUT).await?; + follower.wait_for_block_number(2, DEFAULT_TIMEOUT).await?; + + // At L1 height 3, whitelist Bob. Like the L1 subscriber, apply the events + // before the zone block anchored at height 3 exists; they stay invisible + // to that block's execution, which reads policy state at height 2. + let bob_whitelisted = [PolicyEvent::MembershipChanged { + policy_id: 9, + account: bob, + in_set: true, + }]; + leader + .policy_cache() + .write() + .apply_events(3, &bob_whitelisted); + follower + .policy_cache() + .write() + .apply_events(3, &bob_whitelisted); + + // Block 3 executes with the height-2 whitelist: the deposit to Bob is a + // refused mint on the leader, and the follower must reproduce that. + let refused = fixture.make_deposit(PATH_USD_ADDRESS, alice, bob, amount); + let anchor = fixture.inject_deposits(leader.deposit_queue(), vec![refused]); + follower.l1_block_observer().record(anchor); + leader.wait_for_block_number(3, DEFAULT_TIMEOUT).await?; + follower.wait_for_block_number(3, DEFAULT_TIMEOUT).await?; + assert_eq!(leader.balance_of(PATH_USD_ADDRESS, bob).await?, U256::ZERO); + assert_eq!( + follower.balance_of(PATH_USD_ADDRESS, bob).await?, + U256::ZERO + ); + + // Block 4 executes with the height-3 membership and the same deposit now + // mints — on the follower too. + let accepted = fixture.make_deposit(PATH_USD_ADDRESS, alice, bob, amount); + let anchor = fixture.inject_deposits(leader.deposit_queue(), vec![accepted]); + follower.l1_block_observer().record(anchor); + leader + .wait_for_balance(PATH_USD_ADDRESS, bob, U256::from(amount), DEFAULT_TIMEOUT) + .await?; + follower + .wait_for_balance(PATH_USD_ADDRESS, bob, U256::from(amount), DEFAULT_TIMEOUT) + .await?; + Ok(()) +} + /// A P2P bind failure is fatal rather than leaving the node running without P2P. #[tokio::test(flavor = "multi_thread", worker_threads = 4)] async fn test_p2p_listener_failure_stops_node() -> eyre::Result<()> { @@ -80,7 +198,14 @@ async fn test_p2p_listener_failure_stops_node() -> eyre::Result<()> { let occupied_listener = TcpListener::bind("127.0.0.1:0")?; let p2p_config = leader_p2p_config(occupied_listener.local_addr()?)?; let l1_rpc_url = start_chain_id_rpc(1337).await?; - let mut zone = ZoneTestNode::start_local_with_p2p(l1_rpc_url.to_string(), p2p_config).await?; + let mut zone = + match ZoneTestNode::start_local_with_p2p(l1_rpc_url.to_string(), p2p_config).await { + Ok(zone) => zone, + // The P2P bind failure can abort node launch itself when startup is + // slow (parallel test load) — that is equally fatal, which is what + // this test asserts. + Err(_) => return Ok(()), + }; let _exit = tokio::time::timeout(DEFAULT_TIMEOUT, zone.wait_for_node_exit()) .await diff --git a/crates/node/tests/it/utils.rs b/crates/node/tests/it/utils.rs index ccdc2e70b..0a6677c63 100644 --- a/crates/node/tests/it/utils.rs +++ b/crates/node/tests/it/utils.rs @@ -378,6 +378,7 @@ pub(crate) struct ZoneTestNode { deposit_queue: DepositQueue, l1_state_cache: L1StateCache, policy_cache: zone_l1::PolicyCache, + l1_block_observer: zone_l1::L1BlockObserver, rpc_api_factory: Arc, node_handle: Box, _tasks: Runtime, @@ -411,6 +412,14 @@ impl ZoneTestNode { &self.policy_cache } + /// Returns the node's record of independently observed L1 anchors. + /// + /// Follower imports wait on it; fixtures without a real L1 record anchors + /// here in place of the L1 subscriber. + pub(crate) fn l1_block_observer(&self) -> &zone_l1::L1BlockObserver { + &self.l1_block_observer + } + /// Builds the real private RPC API backed by the node's EthHandlers. pub(crate) async fn rpc_api( &self, @@ -847,6 +856,7 @@ impl ZoneTestNode { let deposit_queue = zone_node.deposit_queue(); let l1_state_cache = zone_node.l1_state_cache(); let policy_cache = zone_node.policy_cache(); + let l1_block_observer = zone_node.l1_block_observer(); if is_local_dummy_l1 { seed_local_policy_cache(&policy_cache); } @@ -915,6 +925,7 @@ impl ZoneTestNode { http_url, l1_state_cache, policy_cache, + l1_block_observer, rpc_api_factory, node_handle: Box::new(node_handle), _tasks: tasks, @@ -3655,9 +3666,11 @@ impl L1Fixture { } /// Inject an empty L1 block (no deposits) into the queue. - pub(crate) fn inject_empty_block(&mut self, queue: &DepositQueue) { + pub(crate) fn inject_empty_block(&mut self, queue: &DepositQueue) -> NumHash { let header = self.next_header(); + let anchor = NumHash::new(header.inner.number, self.last_hash); queue.enqueue(header, L1PortalEvents::default(), vec![]); + anchor } /// Inject `n` empty L1 blocks (no deposits) into the queue. @@ -3668,11 +3681,17 @@ impl L1Fixture { } /// Inject an L1 block with the given deposits into the queue. - pub(crate) fn inject_deposits(&mut self, queue: &DepositQueue, deposits: Vec) { + pub(crate) fn inject_deposits( + &mut self, + queue: &DepositQueue, + deposits: Vec, + ) -> NumHash { let header = self.next_header(); + let anchor = NumHash::new(header.inner.number, self.last_hash); let l1_deposits = deposits.into_iter().map(L1Deposit::Regular).collect(); let events = L1PortalEvents::from_deposits(l1_deposits); queue.enqueue(header, events, vec![]); + anchor } /// Inject an L1 block with mixed regular and encrypted deposits.