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); } }