From f7ca5445036c07d18a861538ac0faf5c6ad7ff14 Mon Sep 17 00:00:00 2001 From: Matthias Seitz Date: Thu, 3 Sep 2026 23:12:38 +0200 Subject: [PATCH] perf(node): stop rebuilding settlement attestations on every retry While a batch boundary waits for its L1 confirmation the leader re-proposed it every 500 ms, and each proposal rebuilt the attestation from scratch: `previous_batch` walks back through a whole batch of blocks reading receipts and decoding logs, followed by a portal multicall and three more L1 reads. Every re-broadcast then made each follower repeat that same work inline on its block-import loop, and each returned signature made the leader repeat it once more to verify a statement it had signed itself. Four changes remove that work: - The leader verifies a follower signature against its own stored proposal at (height, digest) instead of rebuilding it. The precheck already establishes that the leader signed at that digest, and the digest commits to the whole statement, so the comparison is equivalent. `collect_follower_settlement_signatures` no longer needs a chain provider at all. - `previous_batch` is memoized on `AttestationContext` for the boundary height it was computed at. The zone has instant finality, so on the canonical chain the walk is a pure function of that height. - A follower remembers the last (height, digest, signature) it returned and answers a re-broadcast of the same statement from memory. - The retry backs off from 500 ms to at most 5 s while the same boundary stays pending, restarting at the base interval when a new boundary becomes pending, and skips re-proposing once every quorum member has signed. Measured with the mock-based unit tests in replication.rs: a cold build reads 121 blocks of receipts and 2 sealed headers plus 5 L1 requests at the default 120-block batch interval; a retry at the same boundary now reads 1 receipt set and 1 header, and a re-broadcast or a follower-signature verification reads nothing at all. --- crates/node/src/replication.rs | 594 +++++++++++++++++++--- crates/node/src/role.rs | 9 +- crates/node/src/settlement_attestation.rs | 77 ++- crates/sequencer/src/attestation.rs | 33 ++ 4 files changed, 642 insertions(+), 71 deletions(-) diff --git a/crates/node/src/replication.rs b/crates/node/src/replication.rs index a1f8baa22..f912d7d9c 100644 --- a/crates/node/src/replication.rs +++ b/crates/node/src/replication.rs @@ -15,6 +15,7 @@ use reth_provider::HeaderProvider; use reth_storage_api::{BlockNumReader, BlockReader, ReceiptProvider, StateProviderFactory}; use std::{ collections::{BTreeMap, HashMap}, + sync::{Arc, Mutex}, time::Duration, }; use tempo_alloy::TempoNetwork; @@ -41,7 +42,7 @@ use zone_sequencer::{ use alloy_signer_local::PrivateKeySigner; use eyre::{OptionExt as _, WrapErr as _}; -use crate::settlement_attestation::build_settlement_attestation; +use crate::settlement_attestation::{PreviousBatch, build_settlement_attestation}; /// Shared signing and L1-validation context for settlement attestations. #[derive(Clone)] @@ -55,6 +56,11 @@ pub(crate) struct AttestationContext { pub(crate) store: AttestationStore, pub(crate) l1_provider: DynProvider, pub(crate) anchor_config: BatchAnchorConfig, + /// Previous-batch lookup memoized for the boundary height it was computed at. + /// + /// Shared by every clone of the context so a retried proposal, or a second proposal at the + /// same boundary, skips walking a whole batch of receipts again. + pub(crate) previous_batch: Arc>>, } impl AttestationContext { @@ -75,6 +81,7 @@ impl AttestationContext { store, l1_provider, anchor_config, + previous_batch: Arc::default(), } } } @@ -588,16 +595,12 @@ pub(crate) async fn serve_backfill_requests

( /// writer. Backfill requests are served by the process-lifetime /// [`serve_backfill_requests`] task, never by role generations. The loop exits /// when `stop` fires. -async fn store_follower_settlement_signature

( - provider: &P, +fn store_follower_settlement_signature( follower: &P2pPeerId, signature: &[u8], attestation: &AttestationContext, store: &AttestationStore, -) -> eyre::Result<(u64, alloy_primitives::Address, usize)> -where - P: HeaderProvider

+ ReceiptProvider, -{ +) -> eyre::Result<(u64, alloy_primitives::Address, usize)> { let signed = SignedSettlementAttestation::decode(signature)?; let signer = signed.recover_signer(attestation.domain)?; let expected_signer = attestation @@ -622,42 +625,85 @@ where .address(); store.precheck_follower_settlement(height, digest, leader, signer)?; + // The digest commits to the whole statement and the precheck established that the leader holds + // its own signature at (height, digest), so the leader's stored proposal is exactly what + // rebuilding the attestation from receipts and L1 would produce. + let expected = store + .stored_attestation(height, digest, leader) + .ok_or_eyre("leader settlement proposal is no longer stored")?; + eyre::ensure!( + signed.attestation == expected, + "settlement signature does not match leader state" + ); + let signatures = + store.insert_follower_settlement(attestation.domain, leader, signer, signed)?; + Ok((height, signer, signatures)) +} + +/// The settlement signature this follower returned last, keyed by the statement it signed. +struct LastSignedSettlement { + height: u64, + digest: B256, + signature: Vec, +} + +/// Validate and sign one settlement proposal, answering a re-broadcast of an already signed +/// statement from `last_signed`. +/// +/// The leader re-proposes the pending boundary until its batch is confirmed on L1, and validating +/// a proposal walks every block back to the previous boundary and makes several L1 round trips. +/// Signing the same statement again can only produce the same signature, so it is returned as is. +async fn sign_settlement_proposal

( + provider: &P, + attestation: &AttestationContext, + proposal: SettlementAttestation, + height: u64, + last_signed: &mut Option, +) -> eyre::Result> +where + P: HeaderProvider

+ ReceiptProvider, +{ + let digest = attestation.domain.settlement_digest(&proposal); + if let Some(last) = last_signed.as_ref() + && last.height == height + && last.digest == digest + { + return Ok(last.signature.clone()); + } + let expected = build_settlement_attestation( provider, height, attestation, - Some(( - signed.attestation.anchorBlockNumber, - signed.attestation.anchorBlockHash, - )), + Some((proposal.anchorBlockNumber, proposal.anchorBlockHash)), ) .await? - .ok_or_eyre("signed block is not a batch boundary")?; + .ok_or_eyre("proposed block is not a batch boundary")?; eyre::ensure!( - signed.attestation == expected, - "settlement signature does not match leader state" + proposal == expected, + "settlement proposal does not match follower state" ); - let signatures = - store.insert_follower_settlement(attestation.domain, leader, signer, signed)?; - Ok((height, signer, signatures)) + + // Unreachable on an rpc-only member: the P2P layer never routes a proposal to one. Fails + // closed rather than panicking if it ever does. + let signer = attestation.signer.as_ref().ok_or_eyre( + "this node holds no individual secp256k1 key, so it cannot sign a settlement attestation", + )?; + let signature = + SignedSettlementAttestation::sign(proposal, attestation.domain, signer)?.encode(); + *last_signed = Some(LastSignedSettlement { + height, + digest, + signature: signature.clone(), + }); + Ok(signature) } -pub(crate) async fn collect_follower_settlement_signatures

( - provider: P, +pub(crate) async fn collect_follower_settlement_signatures( mut events: mpsc::Receiver, attestation: AttestationContext, stop: sync::CancellationToken, -) where - P: BlockNumReader - + BlockReader - + HeaderProvider

- + StateProviderFactory - + ReceiptProvider - + Clone - + Send - + Sync - + 'static, -{ +) { loop { tokio::select! { biased; @@ -673,15 +719,12 @@ pub(crate) async fn collect_follower_settlement_signatures

( match event { P2pEvent::Started { .. } => {} P2pEvent::SettlementSignatureReceived { follower, signature } => { - let result = async { - store_follower_settlement_signature( - &provider, - &follower, - &signature, - &attestation, - &attestation.store, - ).await - }.await; + let result = store_follower_settlement_signature( + &follower, + &signature, + &attestation, + &attestation.store, + ); match result { Ok((height, signer, signatures)) => info!(target: "zone::p2p", %follower, %signer, height, signatures, "Stored follower settlement signature"), Err(err) => tracing::warn!(target: "zone::p2p", %follower, %err, "Rejected follower settlement signature"), @@ -736,6 +779,7 @@ pub(crate) async fn run_follower_block_sync

( // This is capped to `MAX_PENDING_BLOCKS`. let mut pending = BTreeMap::::new(); let mut backfill = BackfillProgress::new(); + let mut last_signed: Option = None; // Always probe on startup to see if we're behind let mut retry = tokio::time::interval(BACKFILL_RETRY_INTERVAL); @@ -826,31 +870,20 @@ pub(crate) async fn run_follower_block_sync

( height <= persisted_head, "settlement proposal at height {height} is not durable; persisted head is {persisted_head}" ); - let expected = build_settlement_attestation( + let signature = sign_settlement_proposal( &provider, - height, &attestation, - Some((proposal.anchorBlockNumber, proposal.anchorBlockHash)), - ).await?.ok_or_eyre("proposed block is not a batch boundary")?; - eyre::ensure!(proposal == expected, "settlement proposal does not match follower state"); - - // Unreachable on an rpc-only member: the P2P layer never routes a - // proposal to one. Fails closed rather than panicking if it ever does. - let signer = attestation.signer.as_ref().ok_or_eyre( - "this node holds no individual secp256k1 key, so it cannot sign a settlement attestation", - )?; - let signed = SignedSettlementAttestation::sign( proposal, - attestation.domain, - signer, - )?; + height, + &mut last_signed, + ).await?; // Return the signed settlement attestation to the peer that // proposed it. During a scheduled handoff that is the outgoing // leader, not the most recently observed one. commands.send(P2pCommand::SendSettlementSignature { leader: leader.clone(), - signature: signed.encode(), + signature, }) .await .wrap_err("P2P command channel closed")?; @@ -1413,12 +1446,22 @@ mod tests { use tokio_util::sync; use super::{ - AdvanceTempoPortalInputs, BackfillProgress, BroadcasterShutdown, EncodedPersistedBlock, + AdvanceTempoPortalInputs, AttestationContext, AttestationDomain, AttestationStore, + BackfillProgress, BroadcasterShutdown, EncodedPersistedBlock, HashMap, HeaderProvider, MAX_PENDING_BLOCKS, PEER_ANCHOR_WAIT_TIMEOUT, PersistedBlockSource, PersistedTip, - broadcast_persisted_blocks, buffer_pending_block, validate_live_block_sender, + PrivateKeySigner, ReceiptProvider, SettlementAttestation, SignedSettlementAttestation, + TempoHeader, TempoNetwork, broadcast_persisted_blocks, buffer_pending_block, + build_settlement_attestation, sign_settlement_proposal, + store_follower_settlement_signature, validate_live_block_sender, wait_for_validated_peer_anchor, }; - use alloy_primitives::B256; + use alloy_primitives::{Address, B256, Bytes, Log, Sealable as _, U256}; + use alloy_provider::{Provider as _, mock::Asserter}; + use alloy_sol_types::{SolEvent as _, SolValue as _}; + use commonware_cryptography::{Signer as _, ed25519::PrivateKey}; + use reth_provider::{ProviderResult, test_utils::MockEthProvider}; + use tempo_primitives::{TempoPrimitives, TempoReceipt}; + use tempo_zone_contracts as zone_contracts; use zone_l1::{L1BlockTracker, L1PortalEvents}; use zone_p2p::{BackfillCommand, LeadershipSchedule, LeadershipState, P2pCommand}; @@ -2103,4 +2146,439 @@ mod tests { assert_eq!(pending.len(), MAX_PENDING_BLOCKS); assert!(!pending.contains_key(&farther)); } + + /// Zone provider that counts the reads a settlement path performs. + struct CountingZoneProvider { + inner: MockEthProvider, + receipt_reads: Arc, + header_reads: Arc, + } + + impl CountingZoneProvider { + fn new() -> Self { + Self { + inner: MockEthProvider::new(), + receipt_reads: Arc::default(), + header_reads: Arc::default(), + } + } + + /// `(receipt reads, sealed header reads)` observed so far. + fn reads(&self) -> (usize, usize) { + ( + self.receipt_reads.load(Ordering::Relaxed), + self.header_reads.load(Ordering::Relaxed), + ) + } + } + + impl HeaderProvider for CountingZoneProvider { + type Header = TempoHeader; + + fn header(&self, block_hash: B256) -> ProviderResult> { + self.inner.header(block_hash) + } + + fn header_by_number(&self, num: u64) -> ProviderResult> { + self.inner.header_by_number(num) + } + + fn headers_range( + &self, + range: impl std::ops::RangeBounds, + ) -> ProviderResult> { + self.inner.headers_range(range) + } + + fn sealed_header( + &self, + number: u64, + ) -> ProviderResult>> { + self.header_reads.fetch_add(1, Ordering::Relaxed); + self.inner.sealed_header(number) + } + + fn sealed_headers_while( + &self, + range: impl std::ops::RangeBounds, + predicate: impl FnMut(&reth_primitives_traits::SealedHeader) -> bool, + ) -> ProviderResult>> { + self.inner.sealed_headers_while(range, predicate) + } + } + + impl ReceiptProvider for CountingZoneProvider { + type Receipt = TempoReceipt; + + fn receipt(&self, id: u64) -> ProviderResult> { + self.inner.receipt(id) + } + + fn receipt_by_hash(&self, hash: B256) -> ProviderResult> { + self.inner.receipt_by_hash(hash) + } + + fn receipts_by_block( + &self, + block: alloy_eips::BlockHashOrNumber, + ) -> ProviderResult>> { + self.receipt_reads.fetch_add(1, Ordering::Relaxed); + self.inner.receipts_by_block(block) + } + + fn receipts_by_tx_range( + &self, + range: impl std::ops::RangeBounds, + ) -> ProviderResult> { + self.inner.receipts_by_tx_range(range) + } + + fn receipts_by_block_range( + &self, + range: std::ops::RangeInclusive, + ) -> ProviderResult>> { + self.inner.receipts_by_block_range(range) + } + } + + /// Batch boundary the settlement fixture proposes, one full batch after the previous one. + const FIXTURE_BOUNDARY: u64 = 240; + const FIXTURE_PREVIOUS_BOUNDARY: u64 = 120; + /// Receipt reads a cold attestation build performs: the boundary block, then every block back + /// to the previous boundary. + const COLD_RECEIPT_READS: usize = (FIXTURE_BOUNDARY - FIXTURE_PREVIOUS_BOUNDARY + 1) as usize; + /// L1 block the fixture's batch anchors to, which is also the current L1 tip. + const FIXTURE_L1_TIP: u64 = 100; + const FIXTURE_SET_VERSION: u64 = 3; + const FIXTURE_WITHDRAWAL_QUEUE_HASH: B256 = B256::repeat_byte(0x42); + const FIXTURE_VERIFIER: Address = Address::repeat_byte(0x22); + const FIXTURE_PORTAL: Address = Address::repeat_byte(0x11); + + /// A minimal header; `parent_hash` keeps the zone and L1 chains from sharing block hashes. + fn fixture_header(number: u64, parent_hash: B256) -> TempoHeader { + TempoHeader { + inner: alloy_consensus::Header { + number, + parent_hash, + ..Default::default() + }, + ..Default::default() + } + } + + fn deposit_hash(number: u64) -> B256 { + B256::from(U256::from(number)) + } + + fn advance_log(anchor_hash: B256, number: u64) -> Log { + Log { + address: zone_contracts::ZONE_INBOX_ADDRESS, + data: zone_contracts::IZoneInbox::TempoAdvanced { + tempoBlockHash: anchor_hash, + tempoBlockNumber: FIXTURE_L1_TIP, + depositsProcessed: U256::ZERO, + newProcessedDepositQueueHash: deposit_hash(number), + lastProcessedDepositNumber: number, + } + .encode_log_data(), + } + } + + fn batch_finalized_log(index: u64) -> Log { + Log { + address: zone_contracts::ZONE_OUTBOX_ADDRESS, + data: zone_contracts::IZoneOutbox::BatchFinalized { + withdrawalQueueHash: FIXTURE_WITHDRAWAL_QUEUE_HASH, + withdrawalBatchIndex: index, + } + .encode_log_data(), + } + } + + /// Zone chain, L1 mock and signing context for one batch boundary. + struct SettlementFixture { + provider: CountingZoneProvider, + l1: Asserter, + context: AttestationContext, + leader: PrivateKeySigner, + follower: PrivateKeySigner, + follower_peer: zone_p2p::P2pPeerId, + anchor_header: tempo_alloy::rpc::TempoHeaderResponse, + anchor_hash: B256, + previous_tip: B256, + } + + impl SettlementFixture { + /// A zone chain whose batch boundaries are [`FIXTURE_PREVIOUS_BOUNDARY`] and + /// [`FIXTURE_BOUNDARY`], with this node signing as the follower. + fn new() -> Self { + let anchor = fixture_header(FIXTURE_L1_TIP, B256::repeat_byte(0xa1)); + let anchor_hash = anchor.hash_slow(); + let anchor_header = tempo_alloy::rpc::TempoHeaderResponse { + inner: alloy_rpc_types_eth::Header { + hash: anchor_hash, + inner: anchor, + total_difficulty: None, + size: None, + }, + timestamp_millis: 0, + }; + + let provider = CountingZoneProvider::new(); + let mut previous_tip = B256::ZERO; + for number in 1..=FIXTURE_BOUNDARY { + let header = fixture_header(number, B256::ZERO); + let hash = header.hash_slow(); + if number == FIXTURE_PREVIOUS_BOUNDARY { + previous_tip = hash; + } + provider.inner.add_header(hash, header); + + let mut logs = vec![advance_log(anchor_hash, number)]; + match number { + FIXTURE_PREVIOUS_BOUNDARY => logs.push(batch_finalized_log(1)), + FIXTURE_BOUNDARY => logs.push(batch_finalized_log(2)), + _ => {} + } + provider.inner.add_receipts( + number, + vec![TempoReceipt { + tx_type: tempo_primitives::TempoTxType::Legacy, + success: true, + cumulative_gas_used: 0, + logs, + }], + ); + } + + let l1 = Asserter::new(); + let leader = PrivateKeySigner::random(); + let follower = PrivateKeySigner::random(); + let leader_peer = PrivateKey::from_seed(1).public_key(); + let follower_peer = PrivateKey::from_seed(2).public_key(); + let context = AttestationContext::new( + AttestationDomain { + l1_chain_id: 1337, + portal_address: FIXTURE_PORTAL, + zone_id: 7, + }, + Some(FIXTURE_SET_VERSION), + Some(follower.clone()), + HashMap::from([ + (leader_peer, leader.address()), + (follower_peer.clone(), follower.address()), + ]), + AttestationStore::default(), + alloy_provider::ProviderBuilder::new_with_network::() + .connect_mocked_client(l1.clone()) + .erased(), + zone_sequencer::BatchAnchorConfig::default(), + ); + + Self { + provider, + l1, + context, + leader, + follower, + follower_peer, + anchor_header, + anchor_hash, + previous_tip, + } + } + + /// Forget the reads and the cached previous batch, so the next build is cold again. + fn reset_reads(&self) { + self.provider.receipt_reads.store(0, Ordering::Relaxed); + self.provider.header_reads.store(0, Ordering::Relaxed); + *self.context.previous_batch.lock().unwrap() = None; + } + + /// Queue the L1 responses one attestation build consumes: the portal multicall, the tip, + /// and the anchor and Tempo headers. + /// + /// `derives_anchor` covers a leader proposal, which reads the tip once more to pick one. + fn push_l1_responses(&self, derives_anchor: bool) { + let word = |value: u64| Bytes::copy_from_slice(&U256::from(value).to_be_bytes::<32>()); + let returns = vec![ + word(FIXTURE_SET_VERSION), + word(1), + FIXTURE_VERIFIER.abi_encode().into(), + Bytes::copy_from_slice(self.previous_tip.as_slice()), + ]; + self.l1 + .push_success(&Bytes::from((U256::ZERO, returns).abi_encode_params())); + if derives_anchor { + self.l1.push_success(&FIXTURE_L1_TIP); + } + self.l1.push_success(&FIXTURE_L1_TIP); + self.l1.push_success(&self.anchor_header); + self.l1.push_success(&self.anchor_header); + } + } + + /// A retried leader proposal must not walk back through the previous batch again. + #[tokio::test] + async fn retried_settlement_proposal_reuses_the_previous_batch_walk() { + let fixture = SettlementFixture::new(); + fixture.push_l1_responses(true); + fixture.push_l1_responses(true); + + let first = build_settlement_attestation( + &fixture.provider, + FIXTURE_BOUNDARY, + &fixture.context, + None, + ) + .await + .unwrap() + .unwrap(); + assert_eq!(fixture.provider.reads(), (COLD_RECEIPT_READS, 2)); + + let second = build_settlement_attestation( + &fixture.provider, + FIXTURE_BOUNDARY, + &fixture.context, + None, + ) + .await + .unwrap() + .unwrap(); + + assert_eq!(second, first); + assert_eq!( + fixture.provider.reads(), + (COLD_RECEIPT_READS + 1, 3), + "the retry must only read the boundary block itself" + ); + assert!(fixture.l1.read_q().is_empty()); + } + + /// A follower answers a re-broadcast proposal from the signature it already returned. + #[tokio::test] + async fn rebroadcast_settlement_proposal_is_answered_without_revalidating() { + let fixture = SettlementFixture::new(); + fixture.push_l1_responses(true); + let proposal = build_settlement_attestation( + &fixture.provider, + FIXTURE_BOUNDARY, + &fixture.context, + None, + ) + .await + .unwrap() + .unwrap(); + fixture.reset_reads(); + + fixture.push_l1_responses(false); + let mut last_signed = None; + let signature = sign_settlement_proposal( + &fixture.provider, + &fixture.context, + proposal.clone(), + FIXTURE_BOUNDARY, + &mut last_signed, + ) + .await + .unwrap(); + let after_first = fixture.provider.reads(); + assert_eq!(after_first, (COLD_RECEIPT_READS, 2)); + assert!( + fixture.l1.read_q().is_empty(), + "validating a proposal consumes every queued L1 response" + ); + + let resigned = sign_settlement_proposal( + &fixture.provider, + &fixture.context, + proposal, + FIXTURE_BOUNDARY, + &mut last_signed, + ) + .await + .unwrap(); + + // The empty response queue also proves the re-broadcast made no L1 request: one would + // have failed the call. + assert_eq!(resigned, signature); + assert_eq!( + fixture.provider.reads(), + after_first, + "a re-broadcast proposal must not read the zone chain again" + ); + } + + /// The leader checks a follower signature against its own stored proposal. + #[test] + fn follower_settlement_signature_is_verified_without_rebuilding() { + let fixture = SettlementFixture::new(); + let context = { + let mut context = fixture.context.clone(); + context.signer = Some(fixture.leader.clone()); + context + }; + let attestation = SettlementAttestation { + zoneId: 7, + sequencerSetVersion: FIXTURE_SET_VERSION, + zoneHeight: U256::from(FIXTURE_BOUNDARY), + withdrawalBatchIndex: U256::from(2), + verifier: FIXTURE_VERIFIER, + tempoBlockNumber: FIXTURE_L1_TIP, + anchorBlockNumber: FIXTURE_L1_TIP, + anchorBlockHash: fixture.anchor_hash, + blockTransitionHash: B256::repeat_byte(0x31), + depositQueueTransitionHash: B256::repeat_byte(0x32), + withdrawalQueueHash: FIXTURE_WITHDRAWAL_QUEUE_HASH, + verifierConfigHash: B256::repeat_byte(0x33), + }; + + let leader_signed = + SignedSettlementAttestation::sign(attestation.clone(), context.domain, &fixture.leader) + .unwrap(); + context + .store + .insert_settlement(context.domain, fixture.leader.address(), leader_signed); + + let mut forged = attestation.clone(); + forged.withdrawalQueueHash = B256::repeat_byte(0x43); + let forged = + SignedSettlementAttestation::sign(forged, context.domain, &fixture.follower).unwrap(); + let follower_signed = + SignedSettlementAttestation::sign(attestation, context.domain, &fixture.follower) + .unwrap(); + + // Any L1 read would consume this response. + fixture.l1.push_success(&FIXTURE_L1_TIP); + let rejected = store_follower_settlement_signature( + &fixture.follower_peer, + &forged.encode(), + &context, + &context.store, + ) + .expect_err("a signature over another statement must be rejected"); + assert!( + rejected + .to_string() + .contains("settlement response has no active leader proposal"), + "unexpected error: {rejected}" + ); + + let (height, signer, signatures) = store_follower_settlement_signature( + &fixture.follower_peer, + &follower_signed.encode(), + &context, + &context.store, + ) + .unwrap(); + + assert_eq!(height, FIXTURE_BOUNDARY); + assert_eq!(signer, fixture.follower.address()); + assert_eq!(signatures, 2); + assert_eq!( + fixture.l1.read_q().len(), + 1, + "verifying a follower signature must not read L1" + ); + } } diff --git a/crates/node/src/role.rs b/crates/node/src/role.rs index 3d21177d9..aeb570307 100644 --- a/crates/node/src/role.rs +++ b/crates/node/src/role.rs @@ -1017,16 +1017,9 @@ where }); let server_token = token.clone(); - let provider = context.provider.clone(); let attestation = context.attestation.clone(); tasks.spawn(async move { - collect_follower_settlement_signatures( - provider, - sync_rx, - attestation, - server_token, - ) - .await; + collect_follower_settlement_signatures(sync_rx, attestation, server_token).await; TaskEnd::Ended("leader-settlement-signatures") }); diff --git a/crates/node/src/settlement_attestation.rs b/crates/node/src/settlement_attestation.rs index 72f886394..a92ba7332 100644 --- a/crates/node/src/settlement_attestation.rs +++ b/crates/node/src/settlement_attestation.rs @@ -26,6 +26,9 @@ use zone_sequencer::attestation::{SettlementAttestation, SignedSettlementAttesta /// Fallback cadence for transient L1 validation failures or dropped P2P settlement proposals. const SETTLEMENT_RETRY_INTERVAL: Duration = Duration::from_millis(500); +/// Ceiling for the retry backoff applied while the same boundary stays pending. +const MAX_SETTLEMENT_RETRY_INTERVAL: Duration = Duration::from_secs(5); + /// Check the manifest's settlement quorum against `ZonePortal` before any role task starts. /// /// A quorum node the portal has not registered can never settle, and an unreachable threshold @@ -172,10 +175,13 @@ where }) } +/// Previous batch tip hash, its processed deposit-queue hash and processed deposit number. +pub(crate) type PreviousBatch = (B256, B256, u64); + /// Get the previous batch's (i.e the last block in the previous batch) block_hash, /// deposit_hash and processed deposit number. These values /// are used to identify the previous batch while submitting the current batch. -fn previous_batch

(provider: &P, number: u64) -> eyre::Result<(B256, B256, u64)> +fn previous_batch

(provider: &P, number: u64) -> eyre::Result where P: HeaderProvider

+ ReceiptProvider, { @@ -198,6 +204,33 @@ where Ok((B256::ZERO, B256::ZERO, 0)) } +/// [`previous_batch`] memoized on `context` for the boundary it was computed at. +/// +/// The walk reads the receipts of every block back to the previous boundary — a whole batch +/// interval — and the zone's instant finality makes its result a pure function of `number` on the +/// canonical chain. A retried proposal, or a second proposal at the same boundary, reuses it. +fn cached_previous_batch

( + provider: &P, + number: u64, + context: &AttestationContext, +) -> eyre::Result +where + P: HeaderProvider

+ ReceiptProvider, +{ + let mut cached = context + .previous_batch + .lock() + .expect("previous batch cache lock poisoned"); + if let Some((cached_number, previous)) = *cached + && cached_number == number + { + return Ok(previous); + } + let previous = previous_batch(provider, number)?; + *cached = Some((number, previous)); + Ok(previous) +} + /// Build the settlement attestation at a batch boundary in the exact format ZonePortal expects. pub(crate) async fn build_settlement_attestation

( provider: &P, @@ -217,7 +250,7 @@ where .ok_or_eyre(format!("missing batch-tip header {number}"))? .hash(); let (previous_tip, previous_deposit_hash, previous_deposit_number) = - previous_batch(provider, number)?; + cached_previous_batch(provider, number, context)?; let portal = ZonePortal::new(context.domain.portal_address, context.l1_provider.clone()); let (set_version, portal_batch_index, verifier, portal_tip) = context @@ -410,8 +443,13 @@ pub(crate) async fn collect_leader_settlements

( .await; let mut last_scanned = head; - let mut retry = tokio::time::interval(SETTLEMENT_RETRY_INTERVAL); - retry.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); + // Each tick rebuilds the whole attestation, so a boundary that is only waiting for its L1 + // confirmation backs off instead of re-proposing twice a second. A new pending boundary + // restarts at the base interval, keeping recovery from a dropped proposal prompt. + let mut retry_delay = SETTLEMENT_RETRY_INTERVAL; + let mut retry_boundary = pending_boundary; + let retry = tokio::time::sleep(retry_delay); + tokio::pin!(retry); loop { tokio::select! { tip = persisted.next() => { @@ -459,8 +497,18 @@ pub(crate) async fn collect_leader_settlements

( ).await; last_scanned = head; } - _ = retry.tick(), if pending_boundary.is_some() => { + _ = &mut retry, if pending_boundary.is_some() => { let number = pending_boundary.expect("guarded by is_some"); + retry_delay = (retry_delay * 2).min(MAX_SETTLEMENT_RETRY_INTERVAL); + retry.as_mut().reset(tokio::time::Instant::now() + retry_delay); + + // Every quorum member has signed this boundary already, so another proposal can + // only repeat work the batch submitter is waiting on. An unusable certificate is + // dropped by the submitter, which reopens proposing here. + if quorum_fully_signed(&context, number) { + continue; + } + match propose_settlement(&provider, number, &commands, &context).await { Ok(true) => {} Ok(false) => { @@ -513,9 +561,28 @@ pub(crate) async fn collect_leader_settlements

( } } } + + // A different boundary is pending, so restart the backoff. The `continue` paths above only + // bail out on a transient head read and leave `pending_boundary` untouched. + if retry_boundary != pending_boundary { + retry_boundary = pending_boundary; + retry_delay = SETTLEMENT_RETRY_INTERVAL; + retry + .as_mut() + .reset(tokio::time::Instant::now() + retry_delay); + } } } +/// Whether every quorum member has already signed some statement at `height`. +/// +/// The portal's threshold can never exceed the manifest quorum, so a fully signed height cannot +/// gain anything from another proposal. +fn quorum_fully_signed(context: &AttestationContext, height: u64) -> bool { + let quorum = context.addresses.len(); + quorum > 0 && context.store.signature_count(height) >= quorum +} + /// Wait until the submitter or portal resync confirms at least `pending_height`. async fn wait_for_submitted_height( submitted_heights: &mut watch::Receiver, diff --git a/crates/sequencer/src/attestation.rs b/crates/sequencer/src/attestation.rs index aebe74a43..dc5967800 100644 --- a/crates/sequencer/src/attestation.rs +++ b/crates/sequencer/src/attestation.rs @@ -191,6 +191,36 @@ impl AttestationStore { Ok(()) } + /// The statement `signer` attested to at `(height, digest)`, while it remains stored. + /// + /// Lets the leader check an incoming follower signature against the proposal it signed itself, + /// instead of rebuilding that proposal from the zone chain and L1. + pub fn stored_attestation( + &self, + height: u64, + digest: B256, + signer: Address, + ) -> Option { + let all = self + .settlements + .read() + .expect("attestation store lock poisoned"); + all.get(&height)? + .get(&digest)? + .get(&signer) + .map(|signed| signed.attestation.clone()) + } + + /// Most signatures collected for any single statement at `height`. + pub fn signature_count(&self, height: u64) -> usize { + self.settlements + .read() + .expect("attestation store lock poisoned") + .get(&height) + .and_then(|by_digest| by_digest.values().map(BTreeMap::len).max()) + .unwrap_or(0) + } + /// Insert a new follower signature only while its leader proposal remains active. pub fn insert_follower_settlement( &self, @@ -474,8 +504,11 @@ mod tests { ); let certificate = waiting.await.unwrap().unwrap(); assert_eq!(certificate.signatures.len(), 2); + assert_eq!(store.signature_count(10), 2); + assert_eq!(store.signature_count(11), 0); store.remove_submitted(10); assert!(store.settlement_at(10, 1).is_none()); + assert_eq!(store.signature_count(10), 0); } }