From e4dcc392779251f4ea41b27f31f3942101d6802c Mon Sep 17 00:00:00 2001 From: moudyellaz Date: Wed, 5 Aug 2026 19:27:06 +0200 Subject: [PATCH 1/5] feat(storage): add a durable per-peer cross-zone chain tip --- lez/sequencer/core/src/block_store.rs | 7 ++ lez/storage/src/sequencer/mod.rs | 47 +++++++++++-- lez/storage/src/sequencer/sequencer_cells.rs | 60 +++++++++++++++-- lez/storage/src/sequencer/tests.rs | 71 ++++++++++++++++++++ 4 files changed, 175 insertions(+), 10 deletions(-) diff --git a/lez/sequencer/core/src/block_store.rs b/lez/sequencer/core/src/block_store.rs index df61b38b..d9699f2f 100644 --- a/lez/sequencer/core/src/block_store.rs +++ b/lez/sequencer/core/src/block_store.rs @@ -265,6 +265,13 @@ pub fn set_cross_zone_peer_floor( Ok(()) } +/// Drops the stored floor so the watcher reads `peer_zone`'s channel from the +/// peer's genesis again. +pub fn clear_cross_zone_peer_floor(dbio: &RocksDBIO, peer_zone: PeerZoneKey) -> Result<()> { + dbio.delete_cross_zone_peer_floor(peer_zone)?; + Ok(()) +} + #[cfg(test)] mod tests { use common::{block::HashableBlockData, test_utils::sequencer_sign_key_for_testing}; diff --git a/lez/storage/src/sequencer/mod.rs b/lez/storage/src/sequencer/mod.rs index ba6deaf9..7c8fcfad 100644 --- a/lez/storage/src/sequencer/mod.rs +++ b/lez/storage/src/sequencer/mod.rs @@ -22,12 +22,12 @@ use crate::{ sequencer::sequencer_cells::{ FinalBlockMetaCellOwned, FinalBlockMetaCellRef, FinalLeeStateCellOwned, FinalLeeStateCellRef, LEEStateCellOwned, LEEStateCellRef, LastFinalizedBlockIdCell, - LatestBlockMetaCellOwned, LatestBlockMetaCellRef, PeerFloorCellOwned, PeerFloorCellRef, - PeerZoneKey, PendingCrossZoneDispatchRecord, PendingCrossZoneDispatchesCellOwned, - PendingCrossZoneDispatchesCellRef, PendingDepositEventRecord, - PendingDepositEventsCellOwned, PendingDepositEventsCellRef, UnseenWithdrawCountCell, - WithdrawalReconciliationKey, ZoneAnchorCell, ZoneAnchorRecord, ZoneSdkCheckpointCellOwned, - ZoneSdkCheckpointCellRef, + LatestBlockMetaCellOwned, LatestBlockMetaCellRef, PeerChainTip, PeerFloorCellOwned, + PeerFloorCellRef, PeerTipCell, PeerZoneKey, PendingCrossZoneDispatchRecord, + PendingCrossZoneDispatchesCellOwned, PendingCrossZoneDispatchesCellRef, + PendingDepositEventRecord, PendingDepositEventsCellOwned, PendingDepositEventsCellRef, + UnseenWithdrawCountCell, WithdrawalReconciliationKey, ZoneAnchorCell, ZoneAnchorRecord, + ZoneSdkCheckpointCellOwned, ZoneSdkCheckpointCellRef, }, }; @@ -49,6 +49,9 @@ pub const DB_META_PENDING_DEPOSIT_EVENTS_KEY: &str = "pending_deposit_events"; /// Key base for storing a cross-zone watcher's delivery floor on one peer /// channel (opaque bytes). Keyed per peer zone. pub const DB_META_CROSS_ZONE_PEER_FLOOR_KEY: &str = "cross_zone_peer_floor"; +/// Key base for storing the last peer block a cross-zone watcher delivered +/// from, as an id and hash pair. Keyed per peer zone. +pub const DB_META_CROSS_ZONE_PEER_TIP_KEY: &str = "cross_zone_peer_tip"; /// Key base for storing cross-zone deliveries the watcher has recorded but /// which are not yet known to be irreversibly delivered. pub const DB_META_PENDING_CROSS_ZONE_DISPATCHES_KEY: &str = "pending_cross_zone_dispatches"; @@ -611,6 +614,38 @@ impl RocksDBIO { self.put(&PeerFloorCellRef(bytes), peer_zone) } + /// The last peer block one cross-zone watcher delivered from, or `None` + /// before it has delivered anything from that peer. + /// + /// Write it only after that block's deliveries are recorded: a crash in + /// between leaves a tip past deliveries that were never made, and nothing + /// re-reads them. + pub fn get_cross_zone_peer_tip( + &self, + peer_zone: PeerZoneKey, + ) -> DbResult> { + Ok(self.get_opt::(peer_zone)?.map(|cell| cell.0)) + } + + pub fn put_cross_zone_peer_tip( + &self, + peer_zone: PeerZoneKey, + tip: PeerChainTip, + ) -> DbResult<()> { + self.put(&PeerTipCell(tip), peer_zone) + } + + /// Forgets one peer's delivery floor, so its watcher reads that channel from + /// the peer's genesis again. Only sound while that peer has no stored tip: + /// with one, the re-read starts below a tip nothing it reads can link to. + /// + /// A floor above a tip is unusable either way, since the first block read is + /// too far ahead to link, so clearing the floor is what makes rebuilding a + /// tip survive a crash halfway through. + pub fn delete_cross_zone_peer_floor(&self, peer_zone: PeerZoneKey) -> DbResult<()> { + self.del::(peer_zone) + } + pub fn get_pending_cross_zone_dispatches( &self, ) -> DbResult> { diff --git a/lez/storage/src/sequencer/sequencer_cells.rs b/lez/storage/src/sequencer/sequencer_cells.rs index 521fff5e..d3e1e55e 100644 --- a/lez/storage/src/sequencer/sequencer_cells.rs +++ b/lez/storage/src/sequencer/sequencer_cells.rs @@ -8,10 +8,11 @@ use crate::{ error::DbError, sequencer::{ CF_LEE_STATE_NAME, DB_FINAL_BLOCK_META_KEY, DB_FINAL_LEE_STATE_KEY, DB_LEE_STATE_KEY, - DB_META_CROSS_ZONE_PEER_FLOOR_KEY, DB_META_LAST_FINALIZED_BLOCK_ID, - DB_META_LATEST_BLOCK_META_KEY, DB_META_PENDING_CROSS_ZONE_DISPATCHES_KEY, - DB_META_PENDING_DEPOSIT_EVENTS_KEY, DB_META_UNSEEN_WITHDRAW_COUNT_KEY, - DB_META_ZONE_CURSOR_KEY, DB_META_ZONE_SDK_CHECKPOINT_KEY, + DB_META_CROSS_ZONE_PEER_FLOOR_KEY, DB_META_CROSS_ZONE_PEER_TIP_KEY, + DB_META_LAST_FINALIZED_BLOCK_ID, DB_META_LATEST_BLOCK_META_KEY, + DB_META_PENDING_CROSS_ZONE_DISPATCHES_KEY, DB_META_PENDING_DEPOSIT_EVENTS_KEY, + DB_META_UNSEEN_WITHDRAW_COUNT_KEY, DB_META_ZONE_CURSOR_KEY, + DB_META_ZONE_SDK_CHECKPOINT_KEY, }, }; @@ -421,6 +422,57 @@ impl SimpleWritableCell for PeerFloorCellRef<'_> { } } +/// The last peer block a cross-zone watcher delivered from, and the link the +/// next one has to carry. +/// +/// `block_hash` is the recomputed hash, not `header.hash` as read: the +/// signature does not cover that field, so a signed block may carry a bogus one +/// and break the link against the peer's next honest block. +/// +/// Durable, not in-memory: a watcher that re-anchored on restart would accept a +/// block claiming any id. +#[derive(Debug, Clone, Copy, PartialEq, Eq, BorshSerialize, BorshDeserialize)] +pub struct PeerChainTip { + pub block_id: u64, + pub block_hash: HashType, +} + +#[derive(Debug, BorshSerialize, BorshDeserialize)] +pub struct PeerTipCell(pub PeerChainTip); + +impl SimpleStorableCell for PeerTipCell { + type KeyParams = PeerZoneKey; + + const CELL_NAME: &'static str = DB_META_CROSS_ZONE_PEER_TIP_KEY; + const CF_NAME: &'static str = CF_META_NAME; + + /// Folds the peer zone into the key so each peer keeps its own tip. + fn key_constructor(peer_zone: Self::KeyParams) -> DbResult> { + borsh::to_vec(&(Self::CELL_NAME, peer_zone)).map_err(|err| { + DbError::borsh_cast_message( + err, + Some(format!( + "Failed to serialize {:?} key params", + Self::CELL_NAME + )), + ) + }) + } +} + +impl SimpleReadableCell for PeerTipCell {} + +impl SimpleWritableCell for PeerTipCell { + fn value_constructor(&self) -> DbResult> { + borsh::to_vec(&self).map_err(|err| { + DbError::borsh_cast_message( + err, + Some("Failed to serialize cross-zone peer tip cell".to_owned()), + ) + }) + } +} + /// Identity of one withdrawal, shared by the intent recorded when the /// sequencer publishes it and the Bedrock Withdraw event that later reports /// it: the id of the channel note the withdrawal releases. diff --git a/lez/storage/src/sequencer/tests.rs b/lez/storage/src/sequencer/tests.rs index 4f71ab77..385f7ffe 100644 --- a/lez/storage/src/sequencer/tests.rs +++ b/lez/storage/src/sequencer/tests.rs @@ -410,6 +410,77 @@ fn finalized_deposit_records_are_removed_by_op_id() { assert_eq!(stored, vec![second]); } +#[test] +fn peer_chain_tips_round_trip_and_are_kept_per_peer() { + let temp_dir = tempdir().unwrap(); + let (dbio, _genesis) = dbio_with_genesis(temp_dir.path()); + + let peer_a = [1_u8; 32]; + let peer_b = [2_u8; 32]; + assert_eq!(dbio.get_cross_zone_peer_tip(peer_a).unwrap(), None); + + let tip_a = PeerChainTip { + block_id: 7, + block_hash: HashType([9; 32]), + }; + let tip_b = PeerChainTip { + block_id: 3, + block_hash: HashType([4; 32]), + }; + // The floor shares this peer key with the tip. Asserting the tip alone + // passes even when the two cells occupy one key space. + dbio.put_cross_zone_peer_floor_bytes(peer_a, &11_u64.to_le_bytes()) + .unwrap(); + dbio.put_cross_zone_peer_tip(peer_a, tip_a).unwrap(); + dbio.put_cross_zone_peer_tip(peer_b, tip_b).unwrap(); + + // One tip per peer: a shared key would let one peer's chain decide which + // blocks another peer's watcher accepts. + assert_eq!(dbio.get_cross_zone_peer_tip(peer_a).unwrap(), Some(tip_a)); + assert_eq!(dbio.get_cross_zone_peer_tip(peer_b).unwrap(), Some(tip_b)); + assert_eq!( + dbio.get_cross_zone_peer_floor_bytes(peer_a).unwrap(), + Some(11_u64.to_le_bytes().to_vec()), + "the tip must not land in the floor's key space" + ); + + let advanced = PeerChainTip { + block_id: 8, + block_hash: HashType([10; 32]), + }; + dbio.put_cross_zone_peer_tip(peer_a, advanced).unwrap(); + assert_eq!( + dbio.get_cross_zone_peer_tip(peer_a).unwrap(), + Some(advanced) + ); + assert_eq!(dbio.get_cross_zone_peer_tip(peer_b).unwrap(), Some(tip_b)); + + // Clearing the floor is how a watcher with no tip rebuilds one, so it has + // to leave the tip alone: the two share the peer key and differ only in + // their key base. + dbio.delete_cross_zone_peer_floor(peer_a).unwrap(); + assert_eq!(dbio.get_cross_zone_peer_floor_bytes(peer_a).unwrap(), None); + assert_eq!( + dbio.get_cross_zone_peer_tip(peer_a).unwrap(), + Some(advanced) + ); + dbio.delete_cross_zone_peer_floor(peer_a) + .expect("clearing a floor that is already gone is not an error"); + + // On disk, not in memory: a watcher that re-anchored on restart would take + // whatever block reached it first, which is the id the attack picks. + drop(dbio); + let reopened = RocksDBIO::open(temp_dir.path()).unwrap(); + assert_eq!( + reopened.get_cross_zone_peer_tip(peer_a).unwrap(), + Some(advanced) + ); + assert_eq!( + reopened.get_cross_zone_peer_tip(peer_b).unwrap(), + Some(tip_b) + ); +} + #[test] fn dispatch_records_round_trip_and_dedupe_by_message_key() { let temp_dir = tempdir().unwrap(); From 5c28c4cc0c016adfe097a5155c7bf608b64851e4 Mon Sep 17 00:00:00 2001 From: moudyellaz Date: Wed, 5 Aug 2026 20:49:41 +0200 Subject: [PATCH 2/5] fix(sequencer): deliver only from a peer block on its verified chain --- lez/sequencer/core/src/cross_zone_watcher.rs | 1120 ++++++++++++------ 1 file changed, 762 insertions(+), 358 deletions(-) diff --git a/lez/sequencer/core/src/cross_zone_watcher.rs b/lez/sequencer/core/src/cross_zone_watcher.rs index 3b2961f0..d14dfaf0 100644 --- a/lez/sequencer/core/src/cross_zone_watcher.rs +++ b/lez/sequencer/core/src/cross_zone_watcher.rs @@ -1,32 +1,36 @@ use std::{sync::Arc, time::Duration}; -use common::{block::Block, transaction::LeeTransaction}; +use common::{HashType, block::Block, transaction::LeeTransaction}; use cross_zone::{build_dispatch_from_emission, extract_emission}; use cross_zone_inbox_core::{CrossZoneRoute, message_key, routes_permit}; use futures::{Stream, StreamExt as _}; -use lee::PublicKey; +use lee::{GENESIS_BLOCK_ID, PublicKey}; use log::{debug, error, info, warn}; use logos_blockchain_core::mantle::ops::channel::ChannelId; use logos_blockchain_zone_sdk::{ CommonHttpClient, Slot, ZoneMessage, adapter::NodeHttpClient, indexer::ZoneIndexer, }; -use storage::sequencer::{RocksDBIO, sequencer_cells::PendingCrossZoneDispatchRecord}; +use storage::sequencer::{ + RocksDBIO, + sequencer_cells::{PeerChainTip, PendingCrossZoneDispatchRecord}, +}; use crate::{ - block_store::{get_cross_zone_peer_floor, set_cross_zone_peer_floor}, + block_store::{ + clear_cross_zone_peer_floor, get_cross_zone_peer_floor, set_cross_zone_peer_floor, + }, config::{BedrockConfig, CrossZoneConfig}, task_group::TaskGroup, }; -/// Consecutive passes a watcher re-reads the same undecodable slot before giving -/// up and reading past it. +/// Consecutive passes a watcher spends stuck on one slot before it says so as +/// something more than the per-pass failure. /// /// One pass per poll interval, which is the block time, so this is minutes of /// retrying rather than seconds. A transient failure (a truncated read, a peer -/// mid-upgrade) heals well inside that; a block this node genuinely cannot -/// decode does not heal at all, and waiting longer only delays every later -/// message behind it. -const DECODE_RETRY_LIMIT: u32 = 20; +/// mid-upgrade) heals well inside that; anything still stuck after it wants +/// someone to look. +const STUCK_SLOT_ALERT_PASSES: u32 = 20; /// The per-peer settings one watcher pass needs. struct PeerContext { @@ -36,139 +40,78 @@ struct PeerContext { expected_pubkey: Option, } -/// What a pass may do about a slot the watcher cannot decode, and whether it may -/// still move the durable delivery floor. -/// -/// The two are one decision, not two. Past a skipped slot everything is -/// delivered on top of a gap, and persisting past that gap would make the skip -/// survive restarts, so the floor has to stop moving and stay stopped. Holding -/// them in one value is what makes "skipping while still persisting", which -/// would quietly restore that bug, unrepresentable. -#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] -enum SkipPolicy { - /// Nothing has been given up on: deliver everything and move the floor. - #[default] - DeliverAll, - /// Read past this slot, and stop moving the floor. - Skipping(Slot), - /// A slot was skipped earlier in this run. Nothing is being skipped now, but - /// everything read from here sits above the gap, so the floor stays put. - FloorFrozen, -} - /// Why one pass over a peer's stream ended. /// -/// A pass that gave up inside a slot says which kind of failure did it. Only a -/// block this node cannot decode is a reason to eventually read past a slot; -/// a delivery that could not be recorded or handed off is our own problem, and -/// counting it towards the decode budget would read past a slot that is fine. +/// All of them hold the delivery floor at the last slot the watcher consumed +/// whole, bar [`PassOutcome::Drained`] and [`PassOutcome::Stranded`], so the +/// next pass re-reads from there. Only a block that will not deserialize ends a +/// pass; [`link_against`] says why one that decodes never does. #[derive(Clone, Copy, Debug, PartialEq, Eq)] enum PassOutcome { - /// The stream drained. + /// The stream drained, having delivered from at least one block or found + /// nothing to place. Drained, + /// The stream drained having placed nothing, while passing over blocks that + /// were not on the chain this watcher follows. + /// + /// The shape of a tip that no longer tracks the peer: every later block sits + /// above it, is read past, and the floor moves over it, so the peer goes + /// quiet with nothing else to show for it. One pass of this is ordinary (a + /// peer inscribing something that is not its next block), so it is counted + /// rather than acted on. + Stranded, /// Ended inside this slot: its block would not deserialize. Undecodable(Slot), - /// Ended inside this slot: a delivery could not be recorded or enqueued. + /// Ended inside this slot: a delivery could not be recorded, or the chain + /// tip covering it could not be stored. Undelivered(Slot), } -/// The pass-to-pass state of one watcher: what it is stuck on, and what it is -/// allowed to do about it. +/// The pass-to-pass state of one watcher. #[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] struct WatcherState { - /// The slot the watcher is stuck on and how many passes it has spent there. - /// Keyed by slot so a failure at a new slot does not inherit an older - /// slot's count. + /// The slot it is stuck on and how many consecutive passes it has spent + /// there. Keyed by slot so a failure at a new slot does not inherit an + /// older slot's count. stalled: Option<(Slot, u32)>, - skip: SkipPolicy, -} - -impl SkipPolicy { - /// The slot this pass reads past rather than stalling on. - const fn skip_slot(self) -> Option { - match self { - Self::Skipping(slot) => Some(slot), - Self::DeliverAll | Self::FloorFrozen => None, - } - } - - /// Whether this pass may still move the durable delivery floor. - const fn persists_floor(self) -> bool { - matches!(self, Self::DeliverAll) - } - - /// The policy once a pass has read past whatever it was stuck on. - /// - /// Nothing is being skipped any more, but a run that has skipped once keeps - /// its floor frozen: everything from here sits above the gap, and moving the - /// floor over it would make the skip survive a restart. Deliberately not - /// named for clearing: it downgrades, it does not reset. - const fn after_clean_pass(self) -> Self { - match self { - Self::DeliverAll => Self::DeliverAll, - Self::Skipping(_) | Self::FloorFrozen => Self::FloorFrozen, - } - } - - /// Whether a pass that ended at `cursor` actually got past the slot this - /// policy is skipping. - /// - /// A stream can end without reaching it: the zone-sdk ends a stream on a - /// fetch failure exactly as on catching up. Downgrading on such a pass would - /// disarm the skip before it was ever used, and the slot would have to be - /// given up on again from scratch, so a peer endpoint that is flaky around - /// one bad slot would never be read past. - fn used_its_skip(self, cursor: Option) -> bool { - match self { - Self::Skipping(slot) => cursor.is_some_and(|read_to| read_to >= slot), - Self::DeliverAll | Self::FloorFrozen => true, - } - } + /// Consecutive passes that placed nothing while skipping blocks. Not keyed + /// by slot: the peer keeps producing, so every such pass ends at a new slot + /// and a slot-keyed count would reset to one for ever. + stranded: u32, } impl WatcherState { - /// Folds one pass's outcome in, returning a slot the watcher has just given - /// up on so the caller can report it. + /// Folds one pass's outcome in, returning the slot the watcher is stuck on + /// and how long it has been stuck, so the caller can say so. /// - /// `cursor` is the read position after the pass. It is what tells a stream + /// `cursor` is the read position after the pass, and is what tells a stream /// that truncated early apart from one that genuinely drained: the zone-sdk /// ends a stream on a fetch failure exactly as it does on catching up, so - /// without this a flaky peer endpoint would reset the retry count for ever - /// and the watcher would never escape a slot it cannot decode. - fn after_pass(&mut self, outcome: PassOutcome, cursor: Option) -> Option { - match outcome { - PassOutcome::Undecodable(slot) => { - let attempts = match self.stalled { - Some((stuck_on, attempts)) if stuck_on == slot => attempts.saturating_add(1), - _ => 1, - }; - if attempts < DECODE_RETRY_LIMIT { - self.stalled = Some((slot, attempts)); - return None; - } - // Set before the pass that reads past the bad slot, so the - // stored floor stays below it. - self.stalled = None; - self.skip = SkipPolicy::Skipping(slot); - Some(slot) - } - // Ours to fix, not the peer's: retry the slot without spending the - // decode budget on it, or a store outage would read past good blocks. - PassOutcome::Undelivered(_) => None, - PassOutcome::Drained => { + /// without it a flaky peer endpoint resets the count for ever and a watcher + /// stuck for hours never says so. + fn after_pass(&mut self, outcome: PassOutcome, cursor: Option) -> Option<(Slot, u32)> { + let slot = match outcome { + PassOutcome::Drained | PassOutcome::Stranded => { if self.passed_the_stall(cursor) { self.stalled = None; } - // Checked against the skip's own slot, not against `stalled`, - // which arming a skip clears. Otherwise the first truncated - // stream after arming would downgrade the skip before it had - // read past anything. - if self.skip.used_its_skip(cursor) { - self.skip = self.skip.after_clean_pass(); - } - None + self.stranded = match outcome { + PassOutcome::Stranded => self.stranded.saturating_add(1), + PassOutcome::Drained + | PassOutcome::Undecodable(_) + | PassOutcome::Undelivered(_) => 0, + }; + return None; } - } + PassOutcome::Undecodable(slot) | PassOutcome::Undelivered(slot) => slot, + }; + + let attempts = match self.stalled { + Some((stuck_on, attempts)) if stuck_on == slot => attempts.saturating_add(1), + _ => 1, + }; + self.stalled = Some((slot, attempts)); + self.stalled } /// Whether the read position is now past whatever the watcher was stuck on. @@ -179,6 +122,128 @@ impl WatcherState { } } +/// What a starting watcher does with its stored floor and chain tip. +#[derive(Debug, PartialEq, Eq)] +struct Resume { + /// Where to read from. `None` is the peer's genesis. + cursor: Option, + /// Whether the stored floor has to be dropped before anything is read. + clear_floor: bool, +} + +/// Where a peer block sits relative to the chain this watcher has delivered +/// from. +#[derive(Debug, PartialEq, Eq)] +enum Link { + /// The next block on the peer's chain, carrying its recomputed hash. + Next(HashType), + /// At or below the tip, so already delivered from. The ordinary shape of a + /// re-read slot, and how an equivocating second block at one id is refused. + AlreadySeen, + /// Not on the chain this watcher is following, so not deliverable. Read on: + /// the peer's own next block still links to the tip, and treating this as + /// terminal would hand the peer a way to stop its deliveries permanently. + OffChain(String), +} + +/// Whether `block` continues the peer chain pinned by `tip`. +/// +/// This is what closes the suppression. A delivered message's replay key covers +/// `(src_zone, src_block_id, src_tx_index)` and nothing else, so a peer that can +/// get a block delivered under an id of its choosing burns the key an honest +/// block would later use, and the inbox then no-ops the real message as a +/// replay. Off a hash link ids are only claimable in order, so the only id +/// within reach is the one the peer is about to publish anyway. +/// +/// Nothing but [`Link::Next`] is ever delivered from, and nothing but a block +/// that will not decode stops the pass. A peer can inscribe anything it likes on +/// its own channel, so a block this watcher cannot place is read past rather +/// than treated as the end of the chain: the peer's own next honest block still +/// links to the tip. +fn link_against( + tip: Option, + block: &Block, + expected_pubkey: Option<&PublicKey>, +) -> Link { + // The channel authorizes who may write, not what they may claim, so the + // pinned key is what says this node's own sequencer produced the block. + if expected_pubkey.is_some_and(|key| !block.is_signed_by(key)) { + return Link::OffChain("block-signing key does not match the pinned key".to_owned()); + } + + let recomputed = block.recompute_hash(); + if recomputed != block.header.hash { + // The signature does not cover this field, so a correctly signed block + // may still carry a bogus one, and the peer's own next block links + // against the recomputed value rather than this one. + return Link::OffChain(format!( + "block {} carries header hash {} but its contents hash to {recomputed}", + block.header.block_id, block.header.hash + )); + } + + let Some(tip) = tip else { + return if block.header.block_id == GENESIS_BLOCK_ID { + Link::Next(recomputed) + } else { + Link::OffChain(format!( + "block {} is the first one read, but a watcher with no stored chain tip has to start at the peer's genesis block {GENESIS_BLOCK_ID}", + block.header.block_id + )) + }; + }; + + if block.header.block_id <= tip.block_id { + return Link::AlreadySeen; + } + if block.header.block_id > tip.block_id.saturating_add(1) { + return Link::OffChain(format!( + "block {} skips past {}, which is either a hole in what this node read or an id claimed ahead of the peer's chain", + block.header.block_id, + tip.block_id.saturating_add(1) + )); + } + if block.header.prev_block_hash != tip.block_hash { + return Link::OffChain(format!( + "block {} does not follow block {} we delivered from: it links to {} rather than {}", + block.header.block_id, tip.block_id, block.header.prev_block_hash, tip.block_hash + )); + } + Link::Next(recomputed) +} + +/// Whether a watcher stuck for `attempts` passes should say so on this one. +/// +/// Every [`STUCK_SLOT_ALERT_PASSES`], not on the crossing alone: a stall that +/// never clears would otherwise be reported once and then look resolved for as +/// long as it lasts. Not every pass, since that is one line per block time. +const fn alerts_at(attempts: u32) -> bool { + attempts > 0 && attempts.is_multiple_of(STUCK_SLOT_ALERT_PASSES) +} + +/// Where a starting watcher resumes reading a peer's channel. +/// +/// A store holding a floor but no tip predates chain pinning, and its next block +/// would arrive mid-chain with nothing to link against, so it re-reads from the +/// peer's genesis to rebuild the tip. Nothing is delivered twice for that, but +/// every delivery is re-offered to the pending list, a scan and a full rewrite +/// per peer block: a peer with a long history pays for it once. +/// +/// The floor is dropped rather than ignored, or a crash partway through the +/// rebuild resumes above the tip it had just started building. +const fn resume_from(tip: Option, floor: Option) -> Resume { + match tip { + Some(_) => Resume { + cursor: floor, + clear_floor: false, + }, + None => Resume { + cursor: None, + clear_floor: floor.is_some(), + }, + } +} + /// Spawns one watcher task per configured peer. /// /// Each task reads the peer's finalized blocks from Bedrock, recognizes outbound @@ -247,7 +312,7 @@ async fn watch_peer( // key is content-addressed and the inbox no-ops a replay) but re-records // every already-delivered message, so without this a restart replayed the // peer's whole history into the store. - let mut cursor = match get_cross_zone_peer_floor(&dbio, peer_zone) { + let floor = match get_cross_zone_peer_floor(&dbio, peer_zone) { Ok(floor) => floor, Err(err) => { // Falling back to `None` would re-read the peer's whole history and @@ -260,6 +325,36 @@ async fn watch_peer( return; } }; + // The chain this watcher has already delivered from. Without it no block can + // be told apart from one claiming an id it never reached, so a watcher that + // cannot read it delivers nothing rather than guessing. + let mut tip = match dbio.get_cross_zone_peer_tip(peer_zone) { + Ok(tip) => tip, + Err(err) => { + error!( + "Watcher failed to load the stored chain tip for peer {}: {err:#}. Stopping this watcher rather than accepting blocks with nothing to link them against.", + hex::encode(peer_zone) + ); + return; + } + }; + let resume = resume_from(tip, floor); + if resume.clear_floor { + error!( + "Watcher for peer {} holds a delivery floor but no chain tip, so it cannot tell which block continues that peer's chain. Re-reading the channel from the peer's genesis block; deliveries already recorded are deduplicated by message key.", + hex::encode(peer_zone) + ); + // Durably, before reading anything, or a crash partway through the + // rebuild resumes from the stale floor with nothing able to link. + if let Err(err) = clear_cross_zone_peer_floor(&dbio, peer_zone) { + error!( + "Watcher could not clear the stale delivery floor for peer {}: {err:#}. Stopping this watcher rather than rebuilding its chain against a floor a restart would resume from.", + hex::encode(peer_zone) + ); + return; + } + } + let mut cursor = resume.cursor; if let Some(slot) = cursor { info!( "Resuming watcher for peer {} from slot {slot:?}", @@ -267,12 +362,8 @@ async fn watch_peer( ); } - // The slot the watcher is stuck on and how many passes it has spent there, - // and the slot it has given up on. Keyed by slot so a failure at a new slot - // does not inherit an older slot's count. Both stay in memory: a skip must - // not outlive the process, or a peer whose blocks this build cannot decode - // would be skipped past for good and its messages never delivered, even - // after the decoder is fixed. + // In memory, and rebuilt from the store on every start: it says only how + // loud to be about a slot this watcher is stuck on. let mut state = WatcherState::default(); loop { let stream = match zone_indexer.next_messages(cursor).await { @@ -286,38 +377,47 @@ async fn watch_peer( continue; } }; - let outcome = consume_peer_stream(stream, &peer, &dbio, &mut cursor, state.skip).await; + let outcome = consume_peer_stream(stream, &peer, &dbio, &mut cursor, &mut tip).await; - if let Some(slot) = state.after_pass(outcome, cursor) { + if let Some((slot, attempts)) = state.after_pass(outcome, cursor) + && alerts_at(attempts) + { error!( - "Watcher for peer {} could not decode slot {slot:?} after {DECODE_RETRY_LIMIT} attempts; reading past it. Messages in that block are undelivered until this node can decode it, and the delivery floor stops advancing, so every restart re-reads from {:?} onwards.", + "Watcher for peer {} has been stuck at slot {slot:?} for {attempts} passes. Nothing from that peer is being delivered until it clears, and the delivery floor stays at {:?} so the slot keeps coming back.", hex::encode(peer_zone), get_cross_zone_peer_floor(&dbio, peer_zone).ok().flatten() ); } + // Reads on rather than stopping, since one such pass is ordinary, but a + // run of them means the stored tip no longer tracks this peer and every + // block since has been passed over. The floor has moved with them, so + // this does not clear on its own. + if alerts_at(state.stranded) { + error!( + "Watcher for peer {} has read {} consecutive passes without placing a block on the chain it has delivered from, tip {:?}. Nothing from that peer is being delivered, and the blocks passed over are already below the delivery floor.", + hex::encode(peer_zone), + state.stranded, + tip.map(|held| held.block_id) + ); + } // Stream ended (caught up to the peer's last finalized block); poll again. tokio::time::sleep(poll_interval).await; } } -/// Delivers the peer blocks carried by `stream`, moving `cursor` as it goes and -/// persisting the delivery floor behind it. Says why the pass ended, since only -/// a block this node cannot decode counts towards [`DECODE_RETRY_LIMIT`]. +/// Delivers the peer blocks carried by `stream`, moving `cursor` as it goes, +/// persisting the delivery floor behind it and the chain tip as it accepts each +/// block. Says why the pass ended. /// -/// A block that fails to deserialize ends the pass without advancing, so the -/// next poll re-reads it and a transient failure heals. [`SkipPolicy`] names a -/// slot the caller gave up on after [`DECODE_RETRY_LIMIT`] attempts, which is -/// read past so a permanently undecodable inscription cannot wedge the watcher, -/// and says whether the floor may still move: past a skipped slot it may not, -/// because the floor is what a restart resumes from and the skipped messages -/// have to stay reachable. +/// Ending early holds the floor at the last slot consumed whole, so the next +/// poll re-reads from there and a transient failure heals. async fn consume_peer_stream( stream: S, peer: &PeerContext, dbio: &RocksDBIO, cursor: &mut Option, - skip: SkipPolicy, + tip: &mut Option, ) -> PassOutcome where S: Stream, @@ -326,12 +426,17 @@ where // The slot being consumed: every message of it seen so far is handled, but // there may be more to come, so the cursor may not advance onto it yet. let mut in_progress: Option = None; + // What the pass did with the blocks it read, so a peer going quiet behind a + // tip that no longer tracks it is distinguishable from one with nothing to + // say. + let mut placed = 0_usize; + let mut skipped = 0_usize; while let Some((msg, slot)) = stream.next().await { if in_progress != Some(slot) { // A message from a later slot means the previous one completed. if let Some(done) = in_progress { - advance_cursor(dbio, peer.peer_zone, cursor, done, skip.persists_floor()); + advance_cursor(dbio, peer.peer_zone, cursor, done); } in_progress = Some(slot); } @@ -347,43 +452,61 @@ where hex::encode(peer.peer_zone), block.header.block_id ); - // Reject blocks not signed by the pinned peer key (equivocation): - // the channel signer is authenticated by the zone-sdk, but that - // does not prove the peer's honest sequencer produced the block. - if peer - .expected_pubkey - .as_ref() - .is_some_and(|pk| !block.is_signed_by(pk)) - { - warn!( - "Watcher dropping peer {} block {}: block-signing key does not match the pinned key", - hex::encode(peer.peer_zone), - block.header.block_id - ); - continue; + match link_against(*tip, &block, peer.expected_pubkey.as_ref()) { + Link::AlreadySeen => { + debug!( + "Watcher ignoring peer {} block {}: at or below the block it has already delivered from", + hex::encode(peer.peer_zone), + block.header.block_id + ); + } + Link::OffChain(reason) => { + skipped = skipped.saturating_add(1); + warn!( + "Watcher not delivering from peer {} block at slot {slot:?}: {reason}. Reading on; the peer's next block that continues the chain still delivers.", + hex::encode(peer.peer_zone) + ); + } + Link::Next(block_hash) => { + if !record_block_deliveries(&block, peer, dbio) { + // Recording a delivery is what makes it survive the + // mempool. Letting the pass finish here would move + // the floor past this slot on a store that just + // refused the write, and nothing re-reads a slot + // below the floor. + error!( + "Watcher could not record every delivery in peer {} block {}. Holding the floor and retrying the slot.", + hex::encode(peer.peer_zone), + block.header.block_id + ); + return PassOutcome::Undelivered(slot); + } + // After the deliveries, never before: a tip past + // deliveries that were never recorded makes the blocks + // carrying them read as already seen, and nothing looks + // at them again. + let next = PeerChainTip { + block_id: block.header.block_id, + block_hash, + }; + if let Err(err) = dbio.put_cross_zone_peer_tip(peer.peer_zone, next) { + // Advancing only in memory would leave a restart + // resuming from a floor above a tip, and every block + // after it unlinkable. + error!( + "Watcher could not store the chain tip for peer {} at block {}: {err:#}. Holding the floor and retrying the slot.", + hex::encode(peer.peer_zone), + block.header.block_id + ); + return PassOutcome::Undelivered(slot); + } + *tip = Some(next); + placed = placed.saturating_add(1); + } } - - if !record_block_deliveries(&block, peer, dbio) { - // Recording a delivery is what makes it survive the mempool. - // Letting the pass finish here would move the floor past this - // slot on a store that just refused the write, and nothing - // re-reads a slot below the floor. - error!( - "Watcher could not record every delivery in peer {} block {}. Holding the floor and retrying the slot.", - hex::encode(peer.peer_zone), - block.header.block_id - ); - return PassOutcome::Undelivered(slot); - } - } - Err(err) if skip.skip_slot() == Some(slot) => { - debug!( - "Watcher skipping undecodable peer {} block at slot {slot:?}: {err}", - hex::encode(peer.peer_zone) - ); } Err(err) => { - error!( + warn!( "Watcher failed to deserialize peer {} block at slot {slot:?}: {err}. Holding the cursor and retrying.", hex::encode(peer.peer_zone) ); @@ -394,27 +517,21 @@ where // The stream drained cleanly, so the slot in progress completed too. if let Some(done) = in_progress { - advance_cursor(dbio, peer.peer_zone, cursor, done, skip.persists_floor()); + advance_cursor(dbio, peer.peer_zone, cursor, done); + } + if placed == 0 && skipped > 0 { + return PassOutcome::Stranded; } PassOutcome::Drained } -/// Moves the in-memory read cursor past `slot`, and the durable delivery floor -/// with it while `persist_floor` holds. +/// Moves the in-memory read cursor past `slot` and the durable delivery floor +/// with it. /// /// A persist failure is only logged: the worst case is re-reading from the last /// stored slot after a restart, which delivery handles idempotently. -fn advance_cursor( - dbio: &RocksDBIO, - peer_zone: [u8; 32], - cursor: &mut Option, - slot: Slot, - persist_floor: bool, -) { +fn advance_cursor(dbio: &RocksDBIO, peer_zone: [u8; 32], cursor: &mut Option, slot: Slot) { *cursor = Some(slot); - if !persist_floor { - return; - } if let Err(err) = set_cross_zone_peer_floor(dbio, peer_zone, slot) { warn!( "Failed to persist watcher delivery floor for peer {}: {err:#}", @@ -497,8 +614,8 @@ fn record_block_deliveries(block: &Block, peer: &PeerContext, dbio: &RocksDBIO) let offered = deliveries.len(); match dbio.add_pending_cross_zone_dispatches(deliveries) { // Fewer accepted than offered means the rest were recorded by an earlier - // pass over the same slot, which the retry loop does up to - // [`DECODE_RETRY_LIMIT`] times. + // pass over the same slot, which the retry loop repeats for as long as + // the slot stays stuck. Ok(accepted) => { if accepted > 0 { info!( @@ -604,10 +721,46 @@ mod tests { ) } - /// A stream item carrying block `block_id` with one emission for this zone. + /// The peer's chain from its genesis up to and including `block_id`, each + /// block linked to the one before it and carrying one emission for this + /// zone. Empty below [`GENESIS_BLOCK_ID`]. + fn chain_to(block_id: u64) -> Vec { + let mut blocks: Vec = Vec::new(); + for id in GENESIS_BLOCK_ID..=block_id { + let prev = blocks.last().map(|block| block.header.hash); + blocks.push(produce_dummy_block(id, prev, vec![emission()])); + } + blocks + } + + /// The peer's block at `block_id`. + fn chain_block(block_id: u64) -> Block { + chain_to(block_id).pop().expect("chain reaches block_id") + } + + /// The hash the block after `block_id` has to link to. + fn chain_hash(block_id: u64) -> HashType { + chain_block(block_id).header.hash + } + + /// A block continuing the peer's chain at `block_id`, whose one emission + /// targets `target_program_id`. + fn chain_block_to(block_id: u64, target_program_id: lee_core::program::ProgramId) -> Block { + let prefix = chain_to(block_id.saturating_sub(1)); + produce_dummy_block( + block_id, + prefix.last().map(|block| block.header.hash), + vec![emission_to(target_program_id)], + ) + } + + fn block_msg(block: &Block, slot: u64) -> (ZoneMessage, Slot) { + peer_msg(borsh::to_vec(block).expect("block serializes"), slot) + } + + /// A stream item carrying the peer's block `block_id`. fn peer_block_msg(block_id: u64, slot: u64) -> (ZoneMessage, Slot) { - let block = produce_dummy_block(block_id, None, vec![emission()]); - peer_msg(borsh::to_vec(&block).expect("block serializes"), slot) + block_msg(&chain_block(block_id), slot) } /// A stream item carrying a block whose one emission targets @@ -617,8 +770,15 @@ mod tests { slot: u64, target_program_id: lee_core::program::ProgramId, ) -> (ZoneMessage, Slot) { - let block = produce_dummy_block(block_id, None, vec![emission_to(target_program_id)]); - peer_msg(borsh::to_vec(&block).expect("block serializes"), slot) + block_msg(&chain_block_to(block_id, target_program_id), slot) + } + + /// The tip a watcher holds after delivering up to `block_id`. + fn tip_at(block_id: u64) -> Option { + Some(PeerChainTip { + block_id, + block_hash: chain_hash(block_id), + }) } fn undecodable_msg(slot: u64) -> (ZoneMessage, Slot) { @@ -659,51 +819,33 @@ mod tests { state } - fn retry_limit() -> usize { - usize::try_from(DECODE_RETRY_LIMIT).expect("retry limit fits in usize") - } - fn stall(slot: u64, cursor: Option) -> (PassOutcome, Option) { (PassOutcome::Undecodable(Slot::from(slot)), cursor) } #[test] - fn a_slot_is_skipped_only_after_the_retry_limit() { - let limit = retry_limit(); - let almost = vec![stall(4, Some(3)); limit - 1]; + fn a_stuck_slot_is_counted_but_never_read_past() { + // The watcher used to give up on a slot and read past it. Counting is + // now only how loud to be about one it is stuck on. + let passes = vec![stall(4, Some(3)); 3]; + assert_eq!(run_passes(&passes).stalled, Some((Slot::from(4), 3))); + + let long = vec![ + stall(4, Some(3)); + usize::try_from(STUCK_SLOT_ALERT_PASSES).expect("alert threshold fits") * 2 + ]; assert_eq!( - run_passes(&almost).skip, - SkipPolicy::DeliverAll, - "a slot must not be given up on before the limit" + run_passes(&long).stalled, + Some((Slot::from(4), STUCK_SLOT_ALERT_PASSES.saturating_mul(2))), + "a slot is retried for as long as it stays stuck" ); - - let enough = vec![stall(4, Some(3)); limit]; - assert_eq!( - run_passes(&enough).skip, - SkipPolicy::Skipping(Slot::from(4)) - ); - } - - #[test] - fn the_floor_stays_frozen_for_the_rest_of_the_run_after_a_skip() { - // Twenty failures at slot 4, then the pass that reads past it, then - // clean passes: the floor must never be persistable again, or the skip - // survives the next restart and those messages are gone for good. - let mut passes = vec![stall(4, Some(3)); retry_limit()]; - passes.push((PassOutcome::Drained, Some(9))); - passes.push((PassOutcome::Drained, Some(12))); - let state = run_passes(&passes); - - assert_eq!(state.skip, SkipPolicy::FloorFrozen); - assert!(!state.skip.persists_floor()); - assert_eq!(state.stalled, None); } #[test] fn a_stream_that_ended_before_the_stalled_slot_does_not_reset_the_count() { // The zone-sdk ends a stream on a fetch failure exactly as it does on - // catching up. Treating that as a clean pass would reset the retry count - // for ever, and the watcher would never escape a slot it cannot decode. + // catching up. Treating that as a clean pass would reset the count for + // ever, and a watcher stuck for hours would never say so. let mut passes = vec![stall(4, Some(3)); 5]; passes.push((PassOutcome::Drained, Some(3))); let state = run_passes(&passes); @@ -713,42 +855,45 @@ mod tests { "the count survives a pass that never reached the stalled slot" ); - // Reading past it is what actually clears the stall. + // Getting past it is what actually clears the stall. let mut read_past = vec![stall(4, Some(3)); 5]; read_past.push((PassOutcome::Drained, Some(7))); assert_eq!(run_passes(&read_past).stalled, None); } #[test] - fn a_failed_handoff_does_not_spend_the_decode_budget() { - // A store or mempool failure is ours, not the peer's. Counting it here - // would read past a block that decodes perfectly well. - let passes = vec![(PassOutcome::Undelivered(Slot::from(4)), Some(3)); retry_limit() * 2]; - let state = run_passes(&passes); - assert_eq!(state.skip, SkipPolicy::DeliverAll); - assert_eq!(state.stalled, None); + fn a_stall_says_so_on_a_cadence_rather_than_once() { + // Reporting only on the crossing leaves a watcher that never recovers + // looking resolved, which is the failure this whole commit is about. + assert!(!alerts_at(0)); + assert!(!alerts_at(1)); + assert!(!alerts_at(STUCK_SLOT_ALERT_PASSES - 1)); + assert!(alerts_at(STUCK_SLOT_ALERT_PASSES)); + assert!(!alerts_at(STUCK_SLOT_ALERT_PASSES + 1)); + assert!(alerts_at(STUCK_SLOT_ALERT_PASSES * 3)); } #[test] - fn a_truncated_pass_does_not_disarm_a_skip_before_it_is_used() { - // Arming a skip clears `stalled`, so a `Drained` pass that never reached - // the bad slot passes the stall check vacuously. Downgrading on that - // would disarm the skip before it read past anything, and the slot would - // have to be given up on again from scratch, so a peer endpoint that is - // flaky around one bad slot would never be read past. - let mut passes = vec![stall(4, Some(3)); retry_limit()]; - passes.push((PassOutcome::Drained, Some(3))); - let state = run_passes(&passes); - assert_eq!( - state.skip, - SkipPolicy::Skipping(Slot::from(4)), - "a pass that ended before the skipped slot must leave the skip armed" - ); + fn passes_that_place_nothing_while_skipping_blocks_are_counted() { + // A tip that stops tracking the peer is silent by construction: every + // later block sits above it, is read past, and the floor moves over it, + // so there is no stuck slot to count and nothing else to notice. The + // count is not keyed by slot, because the peer keeps producing and each + // such pass ends at a new one. + let stranded = [ + (PassOutcome::Stranded, Some(4)), + (PassOutcome::Stranded, Some(9)), + (PassOutcome::Stranded, Some(14)), + ]; + assert_eq!(run_passes(&stranded).stranded, 3); - // The pass that actually gets past it is what downgrades. - let mut used = vec![stall(4, Some(3)); retry_limit()]; - used.push((PassOutcome::Drained, Some(7))); - assert_eq!(run_passes(&used).skip, SkipPolicy::FloorFrozen); + // Placing anything at all means the tip still tracks the peer. + let mut recovered = stranded.to_vec(); + recovered.push((PassOutcome::Drained, Some(19))); + assert_eq!(run_passes(&recovered).stranded, 0); + + // And an ordinary pass over a peer with nothing to say is not this. + assert_eq!(run_passes(&[(PassOutcome::Drained, Some(4))]).stranded, 0); } #[test] @@ -758,40 +903,155 @@ mod tests { } #[test] - fn a_run_that_skipped_once_never_moves_its_floor_again() { - // The state that makes a skip recoverable: after the bad slot is read - // past, later passes decode cleanly, and the floor still must not move - // over the gap or the skip survives the next restart. - assert_eq!( - SkipPolicy::Skipping(Slot::from(4)).after_clean_pass(), - SkipPolicy::FloorFrozen - ); - assert_eq!( - SkipPolicy::FloorFrozen.after_clean_pass(), - SkipPolicy::FloorFrozen - ); - assert!(!SkipPolicy::FloorFrozen.persists_floor()); - assert_eq!(SkipPolicy::FloorFrozen.skip_slot(), None); + fn every_way_of_ending_early_keeps_the_slot_coming_back() { + // Undecodable and undelivered differ in whose problem they are, not in + // what the watcher does about them: hold the floor and read the slot + // again. + for outcome in [ + PassOutcome::Undecodable(Slot::from(4)), + PassOutcome::Undelivered(Slot::from(4)), + ] { + assert_eq!( + run_passes(&[(outcome, Some(3))]).stalled, + Some((Slot::from(4), 1)) + ); + } + } + + #[test] + fn only_the_next_block_off_the_tip_links() { + let tip = tip_at(2); - // A run that has never skipped keeps moving. assert_eq!( - SkipPolicy::DeliverAll.after_clean_pass(), - SkipPolicy::DeliverAll + link_against(tip, &chain_block(3), None), + Link::Next(chain_hash(3)), + "the block that continues the chain is the one delivered from" + ); + + // The #677 suppression. The peer's chain is public, so the version that + // matters is the block linking correctly and lying only about the id: + // one with no link at all is caught by the check below and proves + // nothing about this one. + assert!(matches!( + link_against( + tip, + &produce_dummy_block(5, Some(chain_hash(2)), vec![emission()]), + None + ), + Link::OffChain(_) + )); + assert!(matches!( + link_against(tip, &produce_dummy_block(5, None, vec![emission()]), None), + Link::OffChain(_) + )); + + // Two blocks claiming one id collapse to one key on chain, so + // delivering from both delivers one message twice. + assert_eq!(link_against(tip, &chain_block(2), None), Link::AlreadySeen); + assert_eq!( + link_against( + tip, + &produce_dummy_block(2, Some(HashType([9; 32])), vec![emission()]), + None + ), + Link::AlreadySeen + ); + + // Right id, wrong ancestry: the peer forked at our tip, or reset it. + assert!(matches!( + link_against( + tip, + &produce_dummy_block(3, Some(HashType([9; 32])), vec![emission()]), + None + ), + Link::OffChain(_) + )); + } + + #[test] + fn a_watcher_with_no_tip_starts_at_the_peers_genesis() { + assert_eq!( + link_against(None, &chain_block(GENESIS_BLOCK_ID), None), + Link::Next(chain_hash(GENESIS_BLOCK_ID)) + ); + // Anchoring on whatever arrived first is the whole attack: the peer + // would pick the id, and every key below it with one block. + assert!(matches!( + link_against(None, &chain_block(GENESIS_BLOCK_ID + 1), None), + Link::OffChain(_) + )); + } + + #[test] + fn a_block_whose_header_hash_is_not_its_contents_is_off_chain() { + // A correctly signed block can still carry any value in `header.hash`. + let mut tampered = chain_block(3); + tampered.header.hash = HashType([9; 32]); + assert!(matches!( + link_against(tip_at(2), &tampered, None), + Link::OffChain(_) + )); + } + + #[test] + fn a_block_not_signed_by_the_pinned_key_is_not_delivered_from() { + let signer = lee::PublicKey::new_from_private_key( + &lee::PrivateKey::try_new([37; 32]).expect("test key"), + ); + assert_eq!( + link_against(None, &chain_block(GENESIS_BLOCK_ID), Some(&signer)), + Link::Next(chain_hash(GENESIS_BLOCK_ID)), + "produce_dummy_block signs with this key, so the pin must accept it" + ); + + let other = lee::PublicKey::try_new([42; 32]).expect("test key"); + assert!(matches!( + link_against(None, &chain_block(GENESIS_BLOCK_ID), Some(&other)), + Link::OffChain(_) + )); + } + + #[test] + fn a_floor_without_a_tip_resumes_from_the_peers_genesis() { + // A store written before chain pinning. The floor is cleared rather + // than ignored so a crash mid-rebuild does not resume from it either. + assert_eq!( + resume_from(None, Some(Slot::from(7))), + Resume { + cursor: None, + clear_floor: true + } + ); + assert_eq!( + resume_from(tip_at(2), Some(Slot::from(7))), + Resume { + cursor: Some(Slot::from(7)), + clear_floor: false + }, + "an ordinary restart resumes where it left off and keeps its floor" + ); + assert_eq!( + resume_from(None, None), + Resume { + cursor: None, + clear_floor: false + }, + "a first start has no floor to clear" ); - assert!(SkipPolicy::DeliverAll.persists_floor()); } #[tokio::test] async fn watcher_persists_its_cursor_as_it_consumes() { let (_dir, dbio) = store(); let mut cursor = None; + let mut tip = None; let outcome = consume_peer_stream( stream::iter(vec![peer_block_msg(1, 0), peer_block_msg(2, 1)]), &peer_context(), &dbio, &mut cursor, - SkipPolicy::DeliverAll, + &mut tip, ) .await; @@ -818,6 +1078,7 @@ mod tests { // from becoming a record production feeds in and gives up on. let (_dir, dbio) = store(); let mut cursor = None; + let mut tip = None; let outcome = consume_peer_stream( stream::iter(vec![peer_block_msg_to( @@ -828,7 +1089,7 @@ mod tests { &peer_context(), &dbio, &mut cursor, - SkipPolicy::DeliverAll, + &mut tip, ) .await; @@ -852,13 +1113,14 @@ mod tests { async fn watcher_records_every_delivery_it_reads() { let (_dir, dbio) = store(); let mut cursor = None; + let mut tip = None; consume_peer_stream( stream::iter(vec![peer_block_msg(1, 0)]), &peer_context(), &dbio, &mut cursor, - SkipPolicy::DeliverAll, + &mut tip, ) .await; @@ -888,13 +1150,14 @@ mod tests { let (_dir, dbio) = store(); break_the_dispatch_store(&dbio); let mut cursor = None; + let mut tip = None; let outcome = consume_peer_stream( stream::iter(vec![peer_block_msg(1, 0)]), &peer_context(), &dbio, &mut cursor, - SkipPolicy::DeliverAll, + &mut tip, ) .await; @@ -907,28 +1170,40 @@ mod tests { None, "the slot must stay re-readable" ); + // The tip is written after the deliveries, never before. Ahead of them a + // crash in between makes the re-read see the block as already delivered + // from, and its messages are never looked at again. + assert_eq!(tip, None); + assert_eq!(dbio.get_cross_zone_peer_tip(PEER_ZONE).unwrap(), None); } #[tokio::test] async fn watcher_resumes_from_the_persisted_cursor_without_rereading() { let (_dir, dbio) = store(); let mut cursor = None; + let mut tip = None; consume_peer_stream( stream::iter(vec![peer_block_msg(1, 0), peer_block_msg(2, 1)]), &peer_context(), &dbio, &mut cursor, - SkipPolicy::DeliverAll, + &mut tip, ) .await; assert_eq!(recorded_keys(&dbio).len(), 2); - // Restart: a fresh watcher seeds its cursor from the store rather than - // starting at `None`, which is what stops it re-reading the peer channel - // from genesis. + // Restart: a fresh watcher seeds both its cursor and its chain tip from + // the store. One that had to rebuild the tip in memory would accept + // whatever block arrived first. let resumed = get_cross_zone_peer_floor(&dbio, PEER_ZONE).unwrap(); assert_eq!(resumed, Some(Slot::from(1))); + let mut resumed_tip = dbio.get_cross_zone_peer_tip(PEER_ZONE).unwrap(); + assert_eq!( + resumed_tip, + tip_at(2), + "the tip is durable, not just in memory" + ); // The sdk resumes the stream at cursor + 1, so only block 3 arrives. let mut resumed_cursor = resumed; @@ -937,7 +1212,7 @@ mod tests { &peer_context(), &dbio, &mut resumed_cursor, - SkipPolicy::DeliverAll, + &mut resumed_tip, ) .await; @@ -960,6 +1235,7 @@ mod tests { async fn watcher_does_not_persist_past_an_undecodable_block() { let (_dir, dbio) = store(); let mut cursor = None; + let mut tip = None; let outcome = consume_peer_stream( stream::iter(vec![ @@ -970,7 +1246,7 @@ mod tests { &peer_context(), &dbio, &mut cursor, - SkipPolicy::DeliverAll, + &mut tip, ) .await; @@ -995,13 +1271,14 @@ mod tests { // failed is never re-read and its delivery is lost for good. let (_dir, dbio) = store(); let mut cursor = None; + let mut tip = None; let outcome = consume_peer_stream( stream::iter(vec![peer_block_msg(1, 4), undecodable_msg(4)]), &peer_context(), &dbio, &mut cursor, - SkipPolicy::DeliverAll, + &mut tip, ) .await; @@ -1012,101 +1289,228 @@ mod tests { } #[tokio::test] - async fn watcher_reads_past_a_slot_it_has_given_up_on() { + async fn an_undecodable_block_stops_the_peer() { + // This used to be read past after twenty attempts, which advanced the + // floor over the hole and lost those messages rather than delaying + // them: nothing after a hole can link. Stopping keeps the slot readable. let (_dir, dbio) = store(); let mut cursor = None; + let mut tip = None; + + for _ in 0..3 { + let outcome = consume_peer_stream( + stream::iter(vec![ + peer_block_msg(1, 0), + undecodable_msg(1), + peer_block_msg(3, 2), + ]), + &peer_context(), + &dbio, + &mut cursor, + &mut tip, + ) + .await; + assert_eq!(outcome, PassOutcome::Undecodable(Slot::from(1))); + } + + assert_eq!( + recorded_keys(&dbio), + vec![message_key(&PEER_ZONE, 1, 0)], + "no pass reads past the slot it cannot decode" + ); + assert_eq!( + get_cross_zone_peer_floor(&dbio, PEER_ZONE).unwrap(), + Some(Slot::from(0)), + "the floor stays below it, so a fixed decoder recovers the messages" + ); + assert_eq!(tip, tip_at(1)); + } + + #[tokio::test] + async fn a_block_claiming_an_id_ahead_of_the_chain_is_not_delivered_from() { + // The #677 suppression, end to end. The peer inscribes a block claiming + // id 5 while its chain is at 1. Delivered, it would burn + // message_key(PEER_ZONE, 5, 0), and the honest block 5 carrying a real + // message at index 0 would then be no-oped by the inbox as a replay, + // with the funds behind it already escrowed on the peer. + // + // The honest blocks behind it still deliver: stopping here would cost + // the peer one inscription to end its own deliveries for good. + let (_dir, dbio) = store(); + let mut cursor = None; + let mut tip = None; + let pre_burn = produce_dummy_block(5, None, vec![emission()]); let outcome = consume_peer_stream( stream::iter(vec![ peer_block_msg(1, 0), - undecodable_msg(1), - peer_block_msg(3, 2), + block_msg(&pre_burn, 1), + peer_block_msg(2, 2), + peer_block_msg(3, 3), ]), &peer_context(), &dbio, &mut cursor, - SkipPolicy::Skipping(Slot::from(1)), + &mut tip, ) .await; - assert_eq!(outcome, PassOutcome::Drained, "the pass drains"); - assert_eq!( - recorded_keys(&dbio), - vec![message_key(&PEER_ZONE, 1, 0), message_key(&PEER_ZONE, 3, 0)], - "only the skipped block goes unrecorded" - ); - - // The cursor moves so later blocks are still read, but the durable floor - // does not follow it past the gap. - assert_eq!( - cursor, - Some(Slot::from(2)), - "the pass keeps reading forward" - ); - assert_eq!( - get_cross_zone_peer_floor(&dbio, PEER_ZONE).unwrap(), - None, - "the floor must not move past a slot this node could not decode" - ); - } - - #[tokio::test] - async fn a_restart_re_reads_a_skipped_slot() { - let (_dir, dbio) = store(); - let mut cursor = None; - - // Slot 0 is recorded, slot 1 is undecodable and eventually skipped, slot - // 2 is recorded on top of the gap. - consume_peer_stream( - stream::iter(vec![peer_block_msg(1, 0)]), - &peer_context(), - &dbio, - &mut cursor, - SkipPolicy::DeliverAll, - ) - .await; - consume_peer_stream( - stream::iter(vec![undecodable_msg(1), peer_block_msg(3, 2)]), - &peer_context(), - &dbio, - &mut cursor, - SkipPolicy::Skipping(Slot::from(1)), - ) - .await; - assert_eq!(recorded_keys(&dbio).len(), 2); - - // A fresh watcher seeds from the floor, so slot 1 comes back around - // rather than being skipped for the life of the store. That is what - // makes a decoder fix recover the messages instead of a store reset. - let resumed = get_cross_zone_peer_floor(&dbio, PEER_ZONE).unwrap(); - assert_eq!(resumed, Some(Slot::from(0))); - - let mut resumed_cursor = resumed; - consume_peer_stream( - stream::iter(vec![peer_block_msg(2, 1), peer_block_msg(3, 2)]), - &peer_context(), - &dbio, - &mut resumed_cursor, - SkipPolicy::DeliverAll, - ) - .await; - - // Three records, not four: the block at slot 2 was recorded on the - // earlier pass and the re-read does not double-track it, while the block - // at slot 1, skipped before, is recorded for the first time. + assert_eq!(outcome, PassOutcome::Drained); assert_eq!( recorded_keys(&dbio), vec![ message_key(&PEER_ZONE, 1, 0), - message_key(&PEER_ZONE, 3, 0), - message_key(&PEER_ZONE, 2, 0) + message_key(&PEER_ZONE, 2, 0), + message_key(&PEER_ZONE, 3, 0) ], - "the previously skipped block must be recorded after a restart, and nothing re-recorded" + "the key the peer aimed to burn is never recorded, and nothing else is held up" + ); + assert_eq!(tip, tip_at(3)); + } + + #[tokio::test] + async fn a_second_block_at_a_delivered_id_is_not_delivered_from() { + // Both claim id 2, so on chain both deliveries key on (PEER_ZONE, 2, 0) + // and the second is a replay the inbox no-ops. + let (_dir, dbio) = store(); + let mut cursor = None; + let mut tip = None; + let equivocation = produce_dummy_block(2, Some(HashType([9; 32])), vec![emission()]); + + let outcome = consume_peer_stream( + stream::iter(vec![ + peer_block_msg(1, 0), + peer_block_msg(2, 1), + block_msg(&equivocation, 2), + ]), + &peer_context(), + &dbio, + &mut cursor, + &mut tip, + ) + .await; + + assert_eq!( + outcome, + PassOutcome::Drained, + "a peer equivocating about its own chain is not this node's failure" ); + assert_eq!( + recorded_keys(&dbio), + vec![message_key(&PEER_ZONE, 1, 0), message_key(&PEER_ZONE, 2, 0)], + "one delivery per id, whatever the peer publishes under it" + ); + assert_eq!(tip, tip_at(2)); + } + + #[tokio::test] + async fn a_block_that_does_not_link_to_the_tip_is_not_delivered_from() { + // Not on the chain we verified, so nothing is delivered from it, and + // the honest block at that id still is when it lands. + let (_dir, dbio) = store(); + let mut cursor = None; + let mut tip = None; + let forked = produce_dummy_block(2, Some(HashType([9; 32])), vec![emission()]); + + let outcome = consume_peer_stream( + stream::iter(vec![ + peer_block_msg(1, 0), + block_msg(&forked, 1), + peer_block_msg(2, 2), + ]), + &peer_context(), + &dbio, + &mut cursor, + &mut tip, + ) + .await; + + assert_eq!(outcome, PassOutcome::Drained); + assert_eq!( + recorded_keys(&dbio), + vec![message_key(&PEER_ZONE, 1, 0), message_key(&PEER_ZONE, 2, 0)], + "the fork is passed over and the peer's own chain continues" + ); + assert_eq!(tip, tip_at(2)); + } + + #[tokio::test] + async fn a_watcher_with_no_tip_delivers_nothing_below_the_peers_genesis() { + // A fresh watcher handed a mid-chain block has nothing to link it + // against. Adopting it would let the peer choose where the chain starts + // and burn every key below it with one block. + let (_dir, dbio) = store(); + let mut cursor = None; + let mut tip = None; + + let outcome = consume_peer_stream( + stream::iter(vec![peer_block_msg(2, 0)]), + &peer_context(), + &dbio, + &mut cursor, + &mut tip, + ) + .await; + + assert_eq!( + outcome, + PassOutcome::Stranded, + "a pass that placed nothing while passing blocks over is how a peer goes quiet" + ); + assert!(recorded_keys(&dbio).is_empty()); + assert_eq!(tip, None); + } + + #[tokio::test] + async fn a_tampered_header_hash_is_not_delivered_from() { + // As correctly signed as any other block, since the signature does not + // cover `header.hash`. Block 2 arriving behind it still delivers. + let (_dir, dbio) = store(); + let mut cursor = None; + let mut tip = None; + let mut tampered = chain_block(2); + tampered.header.hash = HashType([9; 32]); + + let outcome = consume_peer_stream( + stream::iter(vec![ + peer_block_msg(1, 0), + block_msg(&tampered, 1), + peer_block_msg(2, 2), + ]), + &peer_context(), + &dbio, + &mut cursor, + &mut tip, + ) + .await; + + assert_eq!(outcome, PassOutcome::Drained); + assert_eq!( + recorded_keys(&dbio), + vec![message_key(&PEER_ZONE, 1, 0), message_key(&PEER_ZONE, 2, 0)] + ); + assert_eq!(tip, tip_at(2)); + } + + #[tokio::test] + async fn rebuilding_a_missing_tip_clears_the_stale_floor_first() { + // The tip is written per block and the floor per slot, so a crash + // partway through the rebuild would otherwise leave a floor far above a + // tip of 1, and nothing read after that restart could link. + let (_dir, dbio) = store(); + set_cross_zone_peer_floor(&dbio, PEER_ZONE, Slot::from(5000)).unwrap(); + + let floor = get_cross_zone_peer_floor(&dbio, PEER_ZONE).unwrap(); + let tip = dbio.get_cross_zone_peer_tip(PEER_ZONE).unwrap(); + let resume = resume_from(tip, floor); + assert_eq!(resume.cursor, None, "the rebuild reads from genesis"); + assert!(resume.clear_floor); + + clear_cross_zone_peer_floor(&dbio, PEER_ZONE).unwrap(); assert_eq!( get_cross_zone_peer_floor(&dbio, PEER_ZONE).unwrap(), - Some(Slot::from(2)), - "with the gap filled the floor moves again" + None, + "and a crash mid-rebuild resumes from genesis too, not from slot 5000" ); } } From c7739733dbf068e428f763fdb9eaed23eccb7a90 Mon Sep 17 00:00:00 2001 From: moudyellaz Date: Thu, 6 Aug 2026 10:17:12 +0200 Subject: [PATCH 3/5] fix(indexer): deliver only from peer blocks inside the verified run --- lez/indexer/core/src/cross_zone_verifier.rs | 482 +++++++++++++------ lez/sequencer/core/src/cross_zone_watcher.rs | 24 +- 2 files changed, 346 insertions(+), 160 deletions(-) diff --git a/lez/indexer/core/src/cross_zone_verifier.rs b/lez/indexer/core/src/cross_zone_verifier.rs index 3ada2a73..9dcfe2eb 100644 --- a/lez/indexer/core/src/cross_zone_verifier.rs +++ b/lez/indexer/core/src/cross_zone_verifier.rs @@ -36,9 +36,9 @@ const PEER_BLOCK_WAIT_TIMEOUT: Duration = Duration::from_secs(300); /// sleep can overshoot, understates it. const PEER_BLOCK_POLL_INTERVAL: Duration = Duration::from_secs(1); -/// Consecutive passes a peer reader re-reads the same undecodable slot before -/// giving up and reading past it. -const DECODE_RETRY_LIMIT: u32 = 3; +/// Consecutive passes a peer reader spends stuck on one slot before it says so +/// as something more than the per-pass failure. It never reads past the slot. +const STUCK_SLOT_ALERT_PASSES: u32 = 3; /// Why a cross-zone dispatch could not be verified. /// @@ -111,6 +111,7 @@ impl PeerChain { /// What one consistent look at the peer cache says about a referenced block. enum PeerLookup { + /// Held, and inside the run verified from the peer's genesis. Cached(Box), /// Inside the verified run but not held, so it is not on the peer chain. InsideRun, @@ -126,33 +127,70 @@ struct PeerBlocks { } impl PeerBlocks { - /// Caches `block` unless a different block is already held at its id, and - /// says whether it was newly cached. + /// Caches `block` if it is the next one this reader needs, and says whether + /// it was newly cached. /// - /// First write wins. An identical re-read is a no-op, which the reader does - /// on every slot retry. A differing block at a held id is equivocation, and - /// replacing the entry is what makes it a remote halt: the prefix already - /// certified the old value, so the next dispatch naming that id re-derives - /// against the new one and reads as forged. + /// Sequential, exactly like the watcher on the sequencer side. Caching ahead + /// of the run looks harmless, since an id that does not continue the run + /// cannot advance it, but it is not: when the predecessor later arrives the + /// prefix walks straight through the block held ahead, while the watcher, + /// which reads strictly in order and never reconsiders what it passed over, + /// has already discarded it. The two then hold different blocks at one id, + /// and the next dispatch naming it re-derives against the wrong one and + /// halts ingestion. Refusing to look ahead is what keeps the two in step. /// - /// A peer resetting its chain is indistinguishable from this and is refused - /// the same way. The cache is process-local, so a restart adopts it. + /// First write wins at every id but one. An identical re-read is a no-op, + /// which the reader does on every slot retry. A differing block inside the + /// run is equivocation, and replacing it is the remote halt from #648: the + /// prefix certified the old value, so the next dispatch naming that id + /// re-derives against the new one and reads as forged. + /// + /// The exception is the one id that would extend the run. Holding the first + /// arrival there is its own trap: a peer inscribing a block that claims that + /// id and does not continue the chain locks it out for good, since the + /// honest block is refused when it lands and the run can never walk past it, + /// and the channel is append-only so a restart replays the same order and + /// holds the same block. There, and only there, a block that continues the + /// run displaces one that does not. async fn insert(&self, zone: ZoneId, block: Block) -> bool { let mut chains = self.chains.write().await; let chain = chains.entry(zone).or_default(); + let next = chain.next_expected(); + + // Below the peer's genesis as well as ahead of the run: an id under + // GENESIS_BLOCK_ID is on no chain the run can ever walk, and cached it + // would resolve as the peer's own block for ever after. + if block.header.block_id > next || block.header.block_id < GENESIS_BLOCK_ID { + debug!( + "Peer reader for {} not caching block {}: only block {next} continues the run verified from that peer's genesis.", + hex::encode(zone), + block.header.block_id + ); + return false; + } if let Some(held) = chain.blocks.get(&block.header.block_id) { if held.header.hash == block.header.hash { return false; } - error!( - "Peer zone {} equivocated at block {}: holding {}, refusing {}. Restart the indexer if this peer legitimately reset its chain.", + if block.header.block_id != next || !Self::extends_the_run(chain, &block) { + error!( + "Peer zone {} equivocated at block {}: holding {}, refusing {}. Nothing at or above block {} can be delivered from until that peer inscribes a block continuing the run verified from its genesis.", + hex::encode(zone), + block.header.block_id, + held.header.hash, + block.header.hash, + block.header.block_id + ); + return false; + } + info!( + "Peer zone {} block {}: replacing held block {} with {}, which continues the verified run where the held one never could.", hex::encode(zone), block.header.block_id, held.header.hash, block.header.hash ); - return false; } chain.blocks.insert(block.header.block_id, block); @@ -160,29 +198,47 @@ impl PeerBlocks { true } + /// Whether `block` links to the block at the head of the verified run. + /// + /// False before the peer's genesis has been read, so the first block at that + /// id wins and is never displaced, which is how the watcher anchors too. + fn extends_the_run(chain: &PeerChain, block: &Block) -> bool { + chain + .verified_prefix + .and_then(|prefix| chain.blocks.get(&prefix)) + .is_some_and(|tip| block.header.prev_block_hash == tip.header.hash) + } + /// Resolves `block_id` under a single read lock. /// - /// Answering "is it cached?" and "is it inside the verified run?" under two - /// separate locks races with the peer reader: an insert landing between them + /// Cached is not the same as verified, and only the second may be delivered + /// from. A peer writes its own block ids, and a block enters the cache on + /// its own hash and signature alone, so one claiming an id its chain never + /// reached is cached like any other. The run walked from the peer's genesis + /// is what says the peer built it. A block outside that run therefore reads + /// as one the reader has not got to, which stalls the dispatch naming it + /// rather than certifying it, and a peer inscribing a block ahead of its + /// chain cannot get a message delivered under an id it has not reached. + /// + /// One lock, because answering "is it cached?" and "is it inside the run?" + /// separately races with the peer reader: an insert landing between them /// reads as absent-and-inside-the-run, which is the forgery signal, for a /// block that is in fact cached. That is the normal steady state, a waiting - /// verifier and the block it waits for arriving, so it must be one look. + /// verifier and the block it waits for arriving. async fn resolve(&self, zone: ZoneId, block_id: u64) -> PeerLookup { let chains = self.chains.read().await; let Some(chain) = chains.get(&zone) else { return PeerLookup::Behind; }; - if let Some(block) = chain.blocks.get(&block_id) { - return PeerLookup::Cached(Box::new(block.clone())); - } - if chain - .verified_prefix - .is_some_and(|prefix| prefix >= block_id) - { - PeerLookup::InsideRun - } else { - PeerLookup::Behind + if chain.verified_prefix.is_none_or(|prefix| prefix < block_id) { + return PeerLookup::Behind; } + chain + .blocks + .get(&block_id) + .map_or(PeerLookup::InsideRun, |block| { + PeerLookup::Cached(Box::new(block.clone())) + }) } #[cfg(test)] @@ -528,10 +584,9 @@ async fn read_peer( let mut cursor = None; // The slot the reader is stuck on and how many passes it has spent there. - // Keyed by slot: the retry budget is per slot, so a failure at a new slot - // must not inherit an older slot's count and be skipped on its first try. + // Keyed by slot so a failure at a new slot does not inherit an older slot's + // count, and used only to say so once rather than every pass. let mut stalled: Option<(Slot, u32)> = None; - let mut skip_slot = None; loop { match zone_indexer.next_messages(cursor).await { Ok(stream) => { @@ -541,7 +596,6 @@ async fn read_peer( expected_pubkey.as_ref(), &peers, cursor, - skip_slot, ) .await; cursor = pass.cursor; @@ -550,22 +604,18 @@ async fn read_peer( Some((prev, attempts)) if prev == slot => attempts.saturating_add(1), _ => 1, }; - if attempts >= DECODE_RETRY_LIMIT { - // Reading on leaves a hole: dispatches referencing the - // skipped block can no longer be verified, but every - // later block stays readable. + stalled = Some((slot, attempts)); + // Every threshold rather than on the crossing alone: a stall + // that never clears would otherwise be reported once and + // then look resolved for as long as it lasts. + if attempts > 0 && attempts.is_multiple_of(STUCK_SLOT_ALERT_PASSES) { error!( - "Peer reader for {} could not decode slot {slot:?} after {attempts} attempts; reading past it.", + "Peer reader for {} has been stuck at slot {slot:?} for {attempts} passes. The run verified from that peer's genesis stops below it, so every dispatch naming a later block stalls until this slot can be read.", hex::encode(peer_zone) ); - skip_slot = Some(slot); - stalled = None; - } else { - stalled = Some((slot, attempts)); } } else { stalled = None; - skip_slot = None; } } Err(err) => error!( @@ -582,10 +632,11 @@ async fn read_peer( /// /// A block that fails to deserialize ends the pass and holds the cursor at the /// last fully-consumed slot, so the next poll re-reads it and a transient -/// failure heals itself. `skip_slot` names a slot the caller gave up on after -/// [`DECODE_RETRY_LIMIT`] attempts, which is read past instead so a permanently -/// undecodable inscription cannot wedge the reader. Skipping only leaves a hole, -/// which cannot advance [`PeerChain::verified_prefix`] past itself. +/// failure heals itself. It is never read past, however long it stays stuck: +/// a hole stops [`PeerChain::verified_prefix`] below it, and since only blocks +/// inside that run may be delivered from, reading on would cache blocks that can +/// never be used while the reader claimed to have caught up. The watcher on the +/// sequencer side stops at the same hole for the same reason. /// /// The cursor advances only on a slot boundary, since one slot can carry several /// messages and resuming mid-slot would skip the ones after the failure. This @@ -598,7 +649,6 @@ async fn consume_peer_stream( expected_pubkey: Option<&PublicKey>, peers: &PeerBlocks, resume_from: Option, - skip_slot: Option, ) -> PeerPass where S: Stream, @@ -625,12 +675,6 @@ where peers.insert(peer_zone, block).await; } } - Err(err) if skip_slot == Some(slot) => { - debug!( - "Peer reader skipping undecodable block from {} at slot {slot:?}: {err}", - hex::encode(peer_zone) - ); - } Err(err) => { error!( "Peer reader failed to deserialize block from {} at slot {slot:?}: {err}. Holding the cursor and retrying.", @@ -652,7 +696,7 @@ where #[cfg(test)] mod tests { - use common::test_utils::produce_dummy_block; + use common::{HashType, test_utils::produce_dummy_block}; use futures::stream; use lee::{ PrivateKey, PublicKey, PublicTransaction, @@ -667,6 +711,9 @@ mod tests { const SELF_ZONE: ZoneId = [1; 32]; const PEER_ZONE: ZoneId = [2; 32]; const PEER_BLOCK_ID: u64 = 5; + /// The peer's run has to start at its genesis, or every test built on + /// [`peer_chain`] stalls for the full peer-block timeout before failing. + const _: () = assert!(PEER_BLOCK_ID >= GENESIS_BLOCK_ID); fn verifier() -> CrossZoneVerifier { verifier_with_pinned_keys(HashMap::new()) @@ -733,6 +780,27 @@ mod tests { peer_msg(borsh::to_vec(block).expect("block serializes"), slot) } + /// A hash-linked run from the peer's genesis whose last block, + /// `PEER_BLOCK_ID`, carries a `payload` emission. The run is what makes that + /// block deliverable. + fn peer_chain(payload: &[u8]) -> Vec { + let mut chain = linked_chain(PEER_BLOCK_ID.saturating_sub(GENESIS_BLOCK_ID)); + let prev = chain.last().map(|block| block.header.hash); + chain.push(produce_dummy_block( + PEER_BLOCK_ID, + prev, + vec![emission(payload)], + )); + chain + } + + /// Caches a run so its last block sits inside the verified prefix. + async fn cache_chain(verifier: &CrossZoneVerifier, chain: Vec) { + for block in chain { + verifier.peers.insert(PEER_ZONE, block).await; + } + } + /// A peer-stream item whose inscription is not a decodable block. fn undecodable_msg(slot: u64) -> (ZoneMessage, Slot) { peer_msg(b"not a block".to_vec(), slot) @@ -755,13 +823,7 @@ mod tests { #[tokio::test] async fn verifies_dispatch_matching_a_peer_emission() { let verifier = verifier(); - verifier - .peers - .insert( - PEER_ZONE, - produce_dummy_block(PEER_BLOCK_ID, None, vec![emission(b"hi")]), - ) - .await; + cache_chain(&verifier, peer_chain(b"hi")).await; let block = produce_dummy_block(9, None, vec![dispatch(b"hi")]); verifier @@ -775,13 +837,7 @@ mod tests { let verifier = verifier(); // The peer block carries the real emission, but the block claims a // different payload, so re-derivation does not reproduce it. - verifier - .peers - .insert( - PEER_ZONE, - produce_dummy_block(PEER_BLOCK_ID, None, vec![emission(b"real")]), - ) - .await; + cache_chain(&verifier, peer_chain(b"real")).await; let block = produce_dummy_block(9, None, vec![dispatch(b"forged")]); let err = verifier.verify_block(&block).await.unwrap_err(); @@ -798,13 +854,7 @@ mod tests { let mut keys = HashMap::new(); keys.insert(PEER_ZONE, signer); let verifier = verifier_with_pinned_keys(keys); - verifier - .peers - .insert( - PEER_ZONE, - produce_dummy_block(PEER_BLOCK_ID, None, vec![emission(b"hi")]), - ) - .await; + cache_chain(&verifier, peer_chain(b"hi")).await; let block = produce_dummy_block(9, None, vec![dispatch(b"hi")]); verifier @@ -819,13 +869,7 @@ mod tests { let mut keys = HashMap::new(); keys.insert(PEER_ZONE, PublicKey::try_new([42; 32]).unwrap()); let verifier = verifier_with_pinned_keys(keys); - verifier - .peers - .insert( - PEER_ZONE, - produce_dummy_block(PEER_BLOCK_ID, None, vec![emission(b"hi")]), - ) - .await; + cache_chain(&verifier, peer_chain(b"hi")).await; let block = produce_dummy_block(9, None, vec![dispatch(b"hi")]); let err = verifier.verify_block(&block).await.unwrap_err(); @@ -838,13 +882,7 @@ mod tests { #[tokio::test] async fn accepts_replayed_dispatch_as_noop() { let verifier = verifier(); - verifier - .peers - .insert( - PEER_ZONE, - produce_dummy_block(PEER_BLOCK_ID, None, vec![emission(b"hi")]), - ) - .await; + cache_chain(&verifier, peer_chain(b"hi")).await; let first = produce_dummy_block(9, None, vec![dispatch(b"hi")]); let keys = verifier @@ -854,19 +892,12 @@ mod tests { // Mark the delivery seen, as the ingest loop does once the block applies. verifier.record_seen(keys).await; - // Replace the peer block with a different emission so re-deriving the - // replay would mismatch. The replay must still be accepted, proving it is - // the seen-key short-circuit (the inbox no-ops it on chain) and not a - // successful re-derivation. - verifier - .peers - .insert( - PEER_ZONE, - produce_dummy_block(PEER_BLOCK_ID, None, vec![emission(b"different")]), - ) - .await; - - let replay = produce_dummy_block(10, None, vec![dispatch(b"hi")]); + // A payload that cannot re-derive, under the key just recorded. Accepted + // only by the seen-key short circuit, since the inbox no-ops it on + // chain; `unaccepted_dispatch_does_not_poison_seen` asserts the same + // input is rejected when the key was never recorded, which is what makes + // this one about the short circuit rather than re-derivation. + let replay = produce_dummy_block(10, None, vec![dispatch(b"forged")]); verifier .verify_block(&replay) .await @@ -880,13 +911,7 @@ mod tests { // its key to skip re-derivation, while the inbox, never having recorded // the key on chain, would deliver the forgery. let verifier = verifier(); - verifier - .peers - .insert( - PEER_ZONE, - produce_dummy_block(PEER_BLOCK_ID, None, vec![emission(b"hi")]), - ) - .await; + cache_chain(&verifier, peer_chain(b"hi")).await; // The dispatch verifies, but the block is not applied, so record_seen is // not called (the ingest loop records only after an Applied outcome). @@ -916,7 +941,7 @@ mod tests { peer_block_msg(&chain[1], 1), ]); - let pass = consume_peer_stream(stream, PEER_ZONE, None, &peers, None, None).await; + let pass = consume_peer_stream(stream, PEER_ZONE, None, &peers, None).await; assert_eq!(pass.cursor, Some(Slot::from(1))); assert_eq!(pass.stalled_at, None); @@ -972,7 +997,7 @@ mod tests { peer_block_msg(&chain[2], 2), ]); - let pass = consume_peer_stream(stream, PEER_ZONE, None, &peers, None, None).await; + let pass = consume_peer_stream(stream, PEER_ZONE, None, &peers, None).await; assert_eq!(pass.cursor, Some(Slot::from(0))); assert_eq!(pass.stalled_at, Some(Slot::from(1))); @@ -987,17 +1012,17 @@ mod tests { // One slot can carry several messages; the second one fails. let stream = stream::iter(vec![peer_block_msg(&chain[0], 7), undecodable_msg(7)]); - let pass = - consume_peer_stream(stream, PEER_ZONE, None, &peers, Some(Slot::from(6)), None).await; + let pass = consume_peer_stream(stream, PEER_ZONE, None, &peers, Some(Slot::from(6))).await; // Slot 7 is re-read whole next pass, not resumed past the failure. assert_eq!(pass.cursor, Some(Slot::from(6))); } #[tokio::test] - async fn peer_reader_reads_past_a_slot_it_has_given_up_on() { - // After DECODE_RETRY_LIMIT attempts the caller nominates the slot to - // skip, so a permanently undecodable inscription cannot wedge the reader. + async fn peer_reader_never_reads_past_a_slot_it_cannot_decode() { + // It used to give up after DECODE_RETRY_LIMIT attempts and read on, + // caching blocks the stalled run could never use. The watcher stops at + // the same hole, so neither side delivers across it. let peers = PeerBlocks::default(); let chain = linked_chain(3); let stream = stream::iter(vec![ @@ -1006,14 +1031,14 @@ mod tests { peer_block_msg(&chain[2], 2), ]); - let pass = - consume_peer_stream(stream, PEER_ZONE, None, &peers, None, Some(Slot::from(1))).await; + let pass = consume_peer_stream(stream, PEER_ZONE, None, &peers, None).await; - assert_eq!(pass.cursor, Some(Slot::from(2)), "the pass drains"); - assert_eq!(pass.stalled_at, None); - // Block 3 is cached and servable, so dispatches referencing it verify. - assert!(peers.get(PEER_ZONE, 3).await.is_some()); - // But the hole stops the verified run, so block 2 is never called forged. + assert_eq!(pass.cursor, Some(Slot::from(0)), "the slot is held"); + assert_eq!(pass.stalled_at, Some(Slot::from(1))); + assert!( + peers.get(PEER_ZONE, 3).await.is_none(), + "nothing past the hole is even read" + ); assert_eq!(peers.verified_prefix(PEER_ZONE).await, Some(1)); } @@ -1026,19 +1051,12 @@ mod tests { undecodable_msg(1), peer_block_msg(&chain[2], 2), ]); - consume_peer_stream( - stream, - PEER_ZONE, - None, - &verifier.peers, - None, - Some(Slot::from(1)), - ) - .await; + consume_peer_stream(stream, PEER_ZONE, None, &verifier.peers, None).await; - // Regression: the reader cached block 3, so the old `max(cached ids)` - // high-water mark reached 3 and a dispatch referencing block 2 was - // rejected as forged, halting ingestion permanently. + // Regression: block 2 used to be reported as forged, halting ingestion + // permanently, because a `max(cached ids)` high-water mark counted + // blocks read past the undecodable slot. The reader now stops at that + // slot, so block 2 is simply unread, which is lag. let err = verifier .wait_for_peer_block(PEER_ZONE, 2) .await @@ -1049,12 +1067,38 @@ mod tests { ); } + #[tokio::test(start_paused = true)] + async fn a_block_ahead_of_the_run_is_not_cached_and_not_delivered_from() { + // The other half of #677. A peer inscribes a block claiming an id its + // chain has not reached; it is well formed and correctly signed, so + // nothing about the block itself refuses it. Delivered, its message + // would burn the replay key the honest block at that id would later + // need, and the inbox would no-op the real message. + let verifier = verifier(); + cache_chain(&verifier, linked_chain(2)).await; + let claimed = produce_dummy_block(9, None, vec![emission(b"hi")]); + assert!( + !verifier.peers.insert(PEER_ZONE, claimed).await, + "the reader takes the next block on the run, never one ahead of it" + ); + assert!(verifier.peers.get(PEER_ZONE, 9).await.is_none()); + + let err = verifier + .wait_for_peer_block(PEER_ZONE, 9) + .await + .expect_err("a block off the verified run must not resolve"); + assert!( + matches!(err, CrossZoneVerifyError::PeerUnavailable { .. }), + "a claimed id and a reader that is behind read alike, so this stalls rather than halting: {err}" + ); + } + #[tokio::test(start_paused = true)] async fn a_high_block_id_cannot_poison_the_forgery_test() { // A peer picks its own block ids, so one inscribed block claiming a huge // id would drive a `max(cached ids)` high-water mark past every real id - // and make each later dispatch look forged. It cannot extend the - // verified run, so it is inert. + // and make each later dispatch look forged. It is not the block that + // would continue the run, so it is not cached at all. let verifier = verifier(); let chain = linked_chain(2); verifier.peers.insert(PEER_ZONE, chain[0].clone()).await; @@ -1089,7 +1133,6 @@ mod tests { None, &verifier.peers, None, - None, ) .await; assert_eq!(pass.cursor, None, "the failed slot is not skipped"); @@ -1104,7 +1147,6 @@ mod tests { None, &verifier.peers, pass.cursor, - None, ) .await; assert_eq!(pass.stalled_at, None); @@ -1122,13 +1164,19 @@ mod tests { #[tokio::test] async fn an_equivocating_peer_cannot_replace_a_cached_block() { let verifier = verifier(); - let real = produce_dummy_block(PEER_BLOCK_ID, None, vec![emission(b"hi")]); - let impostor = produce_dummy_block(PEER_BLOCK_ID, None, vec![emission(b"forged")]); + let mut chain = peer_chain(b"hi"); + let real = chain.pop().expect("chain has a last block"); + let impostor = produce_dummy_block( + PEER_BLOCK_ID, + Some(real.header.prev_block_hash), + vec![emission(b"forged")], + ); assert_ne!( real.header.hash, impostor.header.hash, "the two blocks must differ, or this proves nothing" ); + cache_chain(&verifier, chain).await; assert!(verifier.peers.insert(PEER_ZONE, real.clone()).await); assert!( !verifier.peers.insert(PEER_ZONE, impostor).await, @@ -1154,12 +1202,149 @@ mod tests { .expect("equivocation must not halt ingestion of an honest dispatch"); } + #[tokio::test] + async fn a_non_linking_block_at_the_run_head_does_not_lock_that_id_out() { + // The other side of first-write-wins: refusing the honest block 3 when + // it lands behind a claimed one would pin the run at 2 for the life of + // the store, and every later message from that peer would stall. + let peers = PeerBlocks::default(); + let chain = linked_chain(3); + for block in chain.iter().take(2).cloned() { + peers.insert(PEER_ZONE, block).await; + } + let claimed = produce_dummy_block(3, Some(HashType([9; 32])), vec![emission(b"claimed")]); + assert!(peers.insert(PEER_ZONE, claimed).await); + assert_eq!( + peers.verified_prefix(PEER_ZONE).await, + Some(2), + "it cannot extend the run, which is what makes it inert but sticky" + ); + + let honest = chain[2].clone(); + assert!( + peers.insert(PEER_ZONE, honest.clone()).await, + "the block that continues the run displaces the one that never could" + ); + assert_eq!(peers.verified_prefix(PEER_ZONE).await, Some(3)); + assert_eq!( + peers.get(PEER_ZONE, 3).await.unwrap().header.hash, + honest.header.hash + ); + } + + #[tokio::test] + async fn a_block_ahead_of_the_run_cannot_win_an_id_the_watcher_gave_to_another() { + // The two sides tie-break differently unless this reader stays strictly + // sequential. The peer is its own sequencer, so it knows block 3's hash + // before publishing it: it inscribes a block claiming id 4 and linking + // to 3, then 3, then its honest 4. + // + // Caching ahead, the prefix would walk 3 and then straight through the + // block held at 4, and the honest 4 would be refused when it landed. + // The watcher reads in order, so at tip 2 it passes over the block + // claiming 4 and never reconsiders it, then delivers from the honest 4. + // Two different blocks at one id, and the dispatch naming it re-derives + // against the wrong one and halts ingestion for good. + let peers = PeerBlocks::default(); + let chain = linked_chain(4); + for block in chain.iter().take(2).cloned() { + peers.insert(PEER_ZONE, block).await; + } + let ahead = produce_dummy_block(4, Some(chain[2].header.hash), vec![emission(b"ahead")]); + assert!(!peers.insert(PEER_ZONE, ahead).await); + + peers.insert(PEER_ZONE, chain[2].clone()).await; + peers.insert(PEER_ZONE, chain[3].clone()).await; + + assert_eq!(peers.verified_prefix(PEER_ZONE).await, Some(4)); + assert_eq!( + peers.get(PEER_ZONE, 4).await.unwrap().header.hash, + chain[3].header.hash, + "the run holds the block the watcher delivered from" + ); + } + + #[tokio::test] + async fn a_block_below_the_peers_genesis_is_not_cached() { + // It is on no chain the run can walk, so nothing would ever certify it, + // and cached it would answer every later lookup at that id as though + // the peer had built it. + let peers = PeerBlocks::default(); + let below = produce_dummy_block(0, None, vec![emission(b"below")]); + assert!(!peers.insert(PEER_ZONE, below).await); + + for block in linked_chain(3) { + peers.insert(PEER_ZONE, block).await; + } + assert_eq!(peers.verified_prefix(PEER_ZONE).await, Some(3)); + assert!(peers.get(PEER_ZONE, 0).await.is_none()); + // Under the run and unheld, which is the forgery signal, and the right + // one: the peer's chain starts at its genesis, so only a dispatch that + // invented the coordinate could name a block below it. + assert!(matches!( + peers.resolve(PEER_ZONE, 0).await, + PeerLookup::InsideRun + )); + } + + #[tokio::test] + async fn a_peer_whose_genesis_is_unread_resolves_to_behind() { + // Before the run has a first block there is nothing to place anything + // against, so every id is lag rather than forgery. Reading it the other + // way round is the original hole: any inscribed block would resolve as + // the peer's own, and any absent one as a forgery that halts. + let peers = PeerBlocks::default(); + assert_eq!(peers.verified_prefix(PEER_ZONE).await, None); + assert!(matches!( + peers.resolve(PEER_ZONE, GENESIS_BLOCK_ID).await, + PeerLookup::Behind + )); + assert!(matches!( + peers.resolve(PEER_ZONE, 9).await, + PeerLookup::Behind + )); + } + + #[tokio::test] + async fn a_block_inside_the_verified_run_is_never_displaced() { + // #648 in the other direction: once the run has walked a block, a later + // arrival at that id must not replace it, whatever it links to, or a + // dispatch naming it re-derives against the new one and halts ingestion. + let peers = PeerBlocks::default(); + let chain = linked_chain(2); + for block in chain.iter().cloned() { + peers.insert(PEER_ZONE, block).await; + } + assert_eq!(peers.verified_prefix(PEER_ZONE).await, Some(2)); + + let impostor = + produce_dummy_block(2, Some(chain[0].header.hash), vec![emission(b"forged")]); + assert!( + !peers.insert(PEER_ZONE, impostor).await, + "it links to block 1 just as the held block does, and the run has certified the held one" + ); + + // The one that would slip through if displacement were allowed anywhere + // below the id that extends the run: it links to the run's own head, so + // every test but the id guard says take it. + let plausible = + produce_dummy_block(2, Some(chain[1].header.hash), vec![emission(b"forged")]); + assert!( + !peers.insert(PEER_ZONE, plausible).await, + "displacement fires only at the id that would extend the run, never inside it" + ); + assert_eq!( + peers.get(PEER_ZONE, 2).await.unwrap().header.hash, + chain[1].header.hash + ); + } + /// The reader re-reads a slot on every retry, so caching the same block /// twice must be a quiet no-op rather than equivocation. #[tokio::test] async fn re_reading_the_same_block_is_not_equivocation() { let verifier = verifier(); - let block = produce_dummy_block(PEER_BLOCK_ID, None, vec![emission(b"hi")]); + let block = linked_chain(1).pop().expect("genesis block"); assert!(verifier.peers.insert(PEER_ZONE, block.clone()).await); assert!( @@ -1169,7 +1354,7 @@ mod tests { assert_eq!( verifier .peers - .get(PEER_ZONE, PEER_BLOCK_ID) + .get(PEER_ZONE, GENESIS_BLOCK_ID) .await .unwrap() .header @@ -1192,7 +1377,6 @@ mod tests { None, &verifier.peers, None, - None, ) .await; @@ -1211,7 +1395,7 @@ mod tests { #[tokio::test] async fn a_block_not_signed_by_the_pinned_key_is_not_cached() { let verifier = verifier(); - let block = produce_dummy_block(PEER_BLOCK_ID, None, vec![emission(b"hi")]); + let block = linked_chain(1).pop().expect("genesis block"); let wrong_key = PublicKey::try_new([42; 32]).unwrap(); let pass = consume_peer_stream( @@ -1220,7 +1404,6 @@ mod tests { Some(&wrong_key), &verifier.peers, None, - None, ) .await; @@ -1239,11 +1422,14 @@ mod tests { Some(&signer), &verifier.peers, None, - None, ) .await; assert!( - verifier.peers.get(PEER_ZONE, PEER_BLOCK_ID).await.is_some(), + verifier + .peers + .get(PEER_ZONE, GENESIS_BLOCK_ID) + .await + .is_some(), "the pinned signer's own block is cached" ); } diff --git a/lez/sequencer/core/src/cross_zone_watcher.rs b/lez/sequencer/core/src/cross_zone_watcher.rs index d14dfaf0..e21641a6 100644 --- a/lez/sequencer/core/src/cross_zone_watcher.rs +++ b/lez/sequencer/core/src/cross_zone_watcher.rs @@ -774,11 +774,11 @@ mod tests { } /// The tip a watcher holds after delivering up to `block_id`. - fn tip_at(block_id: u64) -> Option { - Some(PeerChainTip { + fn tip_at(block_id: u64) -> PeerChainTip { + PeerChainTip { block_id, block_hash: chain_hash(block_id), - }) + } } fn undecodable_msg(slot: u64) -> (ZoneMessage, Slot) { @@ -920,7 +920,7 @@ mod tests { #[test] fn only_the_next_block_off_the_tip_links() { - let tip = tip_at(2); + let tip = Some(tip_at(2)); assert_eq!( link_against(tip, &chain_block(3), None), @@ -988,7 +988,7 @@ mod tests { let mut tampered = chain_block(3); tampered.header.hash = HashType([9; 32]); assert!(matches!( - link_against(tip_at(2), &tampered, None), + link_against(Some(tip_at(2)), &tampered, None), Link::OffChain(_) )); } @@ -1023,7 +1023,7 @@ mod tests { } ); assert_eq!( - resume_from(tip_at(2), Some(Slot::from(7))), + resume_from(Some(tip_at(2)), Some(Slot::from(7))), Resume { cursor: Some(Slot::from(7)), clear_floor: false @@ -1201,7 +1201,7 @@ mod tests { let mut resumed_tip = dbio.get_cross_zone_peer_tip(PEER_ZONE).unwrap(); assert_eq!( resumed_tip, - tip_at(2), + Some(tip_at(2)), "the tip is durable, not just in memory" ); @@ -1323,7 +1323,7 @@ mod tests { Some(Slot::from(0)), "the floor stays below it, so a fixed decoder recovers the messages" ); - assert_eq!(tip, tip_at(1)); + assert_eq!(tip, Some(tip_at(1))); } #[tokio::test] @@ -1365,7 +1365,7 @@ mod tests { ], "the key the peer aimed to burn is never recorded, and nothing else is held up" ); - assert_eq!(tip, tip_at(3)); + assert_eq!(tip, Some(tip_at(3))); } #[tokio::test] @@ -1400,7 +1400,7 @@ mod tests { vec![message_key(&PEER_ZONE, 1, 0), message_key(&PEER_ZONE, 2, 0)], "one delivery per id, whatever the peer publishes under it" ); - assert_eq!(tip, tip_at(2)); + assert_eq!(tip, Some(tip_at(2))); } #[tokio::test] @@ -1431,7 +1431,7 @@ mod tests { vec![message_key(&PEER_ZONE, 1, 0), message_key(&PEER_ZONE, 2, 0)], "the fork is passed over and the peer's own chain continues" ); - assert_eq!(tip, tip_at(2)); + assert_eq!(tip, Some(tip_at(2))); } #[tokio::test] @@ -1489,7 +1489,7 @@ mod tests { recorded_keys(&dbio), vec![message_key(&PEER_ZONE, 1, 0), message_key(&PEER_ZONE, 2, 0)] ); - assert_eq!(tip, tip_at(2)); + assert_eq!(tip, Some(tip_at(2))); } #[tokio::test] From 3d5f083e786eb5fe2339bb7caea1e77781b0ecd3 Mon Sep 17 00:00:00 2001 From: Sergio Chouhy Date: Thu, 6 Aug 2026 16:40:39 -0300 Subject: [PATCH 4/5] fix(lez): duplicated lines --- lez/storage/src/sequencer/mod.rs | 8 ++------ lez/storage/src/sequencer/sequencer_cells.rs | 7 ++----- 2 files changed, 4 insertions(+), 11 deletions(-) diff --git a/lez/storage/src/sequencer/mod.rs b/lez/storage/src/sequencer/mod.rs index 7a112f42..c771ea55 100644 --- a/lez/storage/src/sequencer/mod.rs +++ b/lez/storage/src/sequencer/mod.rs @@ -26,12 +26,8 @@ use crate::{ PeerFloorCellRef, PeerTipCell, PeerZoneKey, PendingCrossZoneDispatchRecord, PendingCrossZoneDispatchesCellOwned, PendingCrossZoneDispatchesCellRef, PendingDepositEventRecord, PendingDepositEventsCellOwned, PendingDepositEventsCellRef, - LatestBlockMetaCellOwned, LatestBlockMetaCellRef, PeerFloorCellOwned, PeerFloorCellRef, - PeerZoneKey, PendingCrossZoneDispatchRecord, PendingCrossZoneDispatchesCellOwned, - PendingCrossZoneDispatchesCellRef, PendingDepositEventRecord, - PendingDepositEventsCellOwned, PendingDepositEventsCellRef, PublishedHighWaterCell, - UnseenWithdrawCountCell, WithdrawalReconciliationKey, ZoneAnchorCell, ZoneAnchorRecord, - ZoneSdkCheckpointCellOwned, ZoneSdkCheckpointCellRef, + PublishedHighWaterCell, UnseenWithdrawCountCell, WithdrawalReconciliationKey, + ZoneAnchorCell, ZoneAnchorRecord, ZoneSdkCheckpointCellOwned, ZoneSdkCheckpointCellRef, }, }; diff --git a/lez/storage/src/sequencer/sequencer_cells.rs b/lez/storage/src/sequencer/sequencer_cells.rs index 08125af8..098d561b 100644 --- a/lez/storage/src/sequencer/sequencer_cells.rs +++ b/lez/storage/src/sequencer/sequencer_cells.rs @@ -11,11 +11,8 @@ use crate::{ DB_META_CROSS_ZONE_PEER_FLOOR_KEY, DB_META_CROSS_ZONE_PEER_TIP_KEY, DB_META_LAST_FINALIZED_BLOCK_ID, DB_META_LATEST_BLOCK_META_KEY, DB_META_PENDING_CROSS_ZONE_DISPATCHES_KEY, DB_META_PENDING_DEPOSIT_EVENTS_KEY, - DB_META_CROSS_ZONE_PEER_FLOOR_KEY, DB_META_LAST_FINALIZED_BLOCK_ID, - DB_META_LATEST_BLOCK_META_KEY, DB_META_PENDING_CROSS_ZONE_DISPATCHES_KEY, - DB_META_PENDING_DEPOSIT_EVENTS_KEY, DB_META_PUBLISHED_HIGH_WATER_KEY, - DB_META_UNSEEN_WITHDRAW_COUNT_KEY, DB_META_ZONE_CURSOR_KEY, - DB_META_ZONE_SDK_CHECKPOINT_KEY, + DB_META_PUBLISHED_HIGH_WATER_KEY, DB_META_UNSEEN_WITHDRAW_COUNT_KEY, + DB_META_ZONE_CURSOR_KEY, DB_META_ZONE_SDK_CHECKPOINT_KEY, }, }; From 245c63d190d18528b82ea50f1faabaee3ccfb19c Mon Sep 17 00:00:00 2001 From: Sergio Chouhy Date: Thu, 6 Aug 2026 16:41:06 -0300 Subject: [PATCH 5/5] chore(workspace): bump bedrock rev --- Cargo.lock | 80 +++++++++++++++++++++++++++--------------------------- Cargo.toml | 16 +++++------ 2 files changed, 48 insertions(+), 48 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 8f8344f1..d4c52e89 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -5757,7 +5757,7 @@ checksum = "113b30b4cd05f7c06868fdb2854f66a7b9fece9a48425351cd532e810d74024f" [[package]] name = "logos-blockchain-blake2btree" version = "0.0.0" -source = "git+https://github.com/logos-blockchain/logos-blockchain.git?rev=0dc34e2c5a6e2ff772140378659489a602ee06bc#0dc34e2c5a6e2ff772140378659489a602ee06bc" +source = "git+https://github.com/logos-blockchain/logos-blockchain.git?rev=e2a1c3b7ef2191c224f998b94332c5926c789f9d#e2a1c3b7ef2191c224f998b94332c5926c789f9d" dependencies = [ "blake2", "logos-blockchain-dynamic-merkle", @@ -5767,7 +5767,7 @@ dependencies = [ [[package]] name = "logos-blockchain-blend-crypto" version = "0.0.0" -source = "git+https://github.com/logos-blockchain/logos-blockchain.git?rev=0dc34e2c5a6e2ff772140378659489a602ee06bc#0dc34e2c5a6e2ff772140378659489a602ee06bc" +source = "git+https://github.com/logos-blockchain/logos-blockchain.git?rev=e2a1c3b7ef2191c224f998b94332c5926c789f9d#e2a1c3b7ef2191c224f998b94332c5926c789f9d" dependencies = [ "blake2", "logos-blockchain-groth16", @@ -5781,7 +5781,7 @@ dependencies = [ [[package]] name = "logos-blockchain-blend-message" version = "0.0.0" -source = "git+https://github.com/logos-blockchain/logos-blockchain.git?rev=0dc34e2c5a6e2ff772140378659489a602ee06bc#0dc34e2c5a6e2ff772140378659489a602ee06bc" +source = "git+https://github.com/logos-blockchain/logos-blockchain.git?rev=e2a1c3b7ef2191c224f998b94332c5926c789f9d#e2a1c3b7ef2191c224f998b94332c5926c789f9d" dependencies = [ "blake2", "derivative", @@ -5806,7 +5806,7 @@ dependencies = [ [[package]] name = "logos-blockchain-blend-proofs" version = "0.0.0" -source = "git+https://github.com/logos-blockchain/logos-blockchain.git?rev=0dc34e2c5a6e2ff772140378659489a602ee06bc#0dc34e2c5a6e2ff772140378659489a602ee06bc" +source = "git+https://github.com/logos-blockchain/logos-blockchain.git?rev=e2a1c3b7ef2191c224f998b94332c5926c789f9d#e2a1c3b7ef2191c224f998b94332c5926c789f9d" dependencies = [ "ed25519-dalek", "generic-array 1.4.3", @@ -5827,7 +5827,7 @@ dependencies = [ [[package]] name = "logos-blockchain-chain-broadcast-service" version = "0.0.0" -source = "git+https://github.com/logos-blockchain/logos-blockchain.git?rev=0dc34e2c5a6e2ff772140378659489a602ee06bc#0dc34e2c5a6e2ff772140378659489a602ee06bc" +source = "git+https://github.com/logos-blockchain/logos-blockchain.git?rev=e2a1c3b7ef2191c224f998b94332c5926c789f9d#e2a1c3b7ef2191c224f998b94332c5926c789f9d" dependencies = [ "async-trait", "derivative", @@ -5841,7 +5841,7 @@ dependencies = [ [[package]] name = "logos-blockchain-chain-service" version = "0.0.0" -source = "git+https://github.com/logos-blockchain/logos-blockchain.git?rev=0dc34e2c5a6e2ff772140378659489a602ee06bc#0dc34e2c5a6e2ff772140378659489a602ee06bc" +source = "git+https://github.com/logos-blockchain/logos-blockchain.git?rev=e2a1c3b7ef2191c224f998b94332c5926c789f9d#e2a1c3b7ef2191c224f998b94332c5926c789f9d" dependencies = [ "async-trait", "bytes", @@ -5922,7 +5922,7 @@ dependencies = [ [[package]] name = "logos-blockchain-circuits-prover" version = "0.0.0" -source = "git+https://github.com/logos-blockchain/logos-blockchain.git?rev=0dc34e2c5a6e2ff772140378659489a602ee06bc#0dc34e2c5a6e2ff772140378659489a602ee06bc" +source = "git+https://github.com/logos-blockchain/logos-blockchain.git?rev=e2a1c3b7ef2191c224f998b94332c5926c789f9d#e2a1c3b7ef2191c224f998b94332c5926c789f9d" dependencies = [ "rust-rapidsnark", ] @@ -5949,7 +5949,7 @@ dependencies = [ [[package]] name = "logos-blockchain-codec" version = "0.0.0" -source = "git+https://github.com/logos-blockchain/logos-blockchain.git?rev=0dc34e2c5a6e2ff772140378659489a602ee06bc#0dc34e2c5a6e2ff772140378659489a602ee06bc" +source = "git+https://github.com/logos-blockchain/logos-blockchain.git?rev=e2a1c3b7ef2191c224f998b94332c5926c789f9d#e2a1c3b7ef2191c224f998b94332c5926c789f9d" dependencies = [ "hex", "logos-blockchain-codec-macros", @@ -5961,7 +5961,7 @@ dependencies = [ [[package]] name = "logos-blockchain-codec-macros" version = "0.0.0" -source = "git+https://github.com/logos-blockchain/logos-blockchain.git?rev=0dc34e2c5a6e2ff772140378659489a602ee06bc#0dc34e2c5a6e2ff772140378659489a602ee06bc" +source = "git+https://github.com/logos-blockchain/logos-blockchain.git?rev=e2a1c3b7ef2191c224f998b94332c5926c789f9d#e2a1c3b7ef2191c224f998b94332c5926c789f9d" dependencies = [ "hex", "proc-macro2", @@ -5972,7 +5972,7 @@ dependencies = [ [[package]] name = "logos-blockchain-common-http-client" version = "0.0.0" -source = "git+https://github.com/logos-blockchain/logos-blockchain.git?rev=0dc34e2c5a6e2ff772140378659489a602ee06bc#0dc34e2c5a6e2ff772140378659489a602ee06bc" +source = "git+https://github.com/logos-blockchain/logos-blockchain.git?rev=e2a1c3b7ef2191c224f998b94332c5926c789f9d#e2a1c3b7ef2191c224f998b94332c5926c789f9d" dependencies = [ "futures", "hex", @@ -5995,7 +5995,7 @@ dependencies = [ [[package]] name = "logos-blockchain-core" version = "0.0.0" -source = "git+https://github.com/logos-blockchain/logos-blockchain.git?rev=0dc34e2c5a6e2ff772140378659489a602ee06bc#0dc34e2c5a6e2ff772140378659489a602ee06bc" +source = "git+https://github.com/logos-blockchain/logos-blockchain.git?rev=e2a1c3b7ef2191c224f998b94332c5926c789f9d#e2a1c3b7ef2191c224f998b94332c5926c789f9d" dependencies = [ "ark-ff", "bincode", @@ -6028,7 +6028,7 @@ dependencies = [ [[package]] name = "logos-blockchain-cryptarchia-engine" version = "0.0.0" -source = "git+https://github.com/logos-blockchain/logos-blockchain.git?rev=0dc34e2c5a6e2ff772140378659489a602ee06bc#0dc34e2c5a6e2ff772140378659489a602ee06bc" +source = "git+https://github.com/logos-blockchain/logos-blockchain.git?rev=e2a1c3b7ef2191c224f998b94332c5926c789f9d#e2a1c3b7ef2191c224f998b94332c5926c789f9d" dependencies = [ "logos-blockchain-codec", "logos-blockchain-pol", @@ -6045,7 +6045,7 @@ dependencies = [ [[package]] name = "logos-blockchain-cryptarchia-sync" version = "0.0.0" -source = "git+https://github.com/logos-blockchain/logos-blockchain.git?rev=0dc34e2c5a6e2ff772140378659489a602ee06bc#0dc34e2c5a6e2ff772140378659489a602ee06bc" +source = "git+https://github.com/logos-blockchain/logos-blockchain.git?rev=e2a1c3b7ef2191c224f998b94332c5926c789f9d#e2a1c3b7ef2191c224f998b94332c5926c789f9d" dependencies = [ "bytes", "futures", @@ -6064,7 +6064,7 @@ dependencies = [ [[package]] name = "logos-blockchain-dynamic-merkle" version = "0.0.0" -source = "git+https://github.com/logos-blockchain/logos-blockchain.git?rev=0dc34e2c5a6e2ff772140378659489a602ee06bc#0dc34e2c5a6e2ff772140378659489a602ee06bc" +source = "git+https://github.com/logos-blockchain/logos-blockchain.git?rev=e2a1c3b7ef2191c224f998b94332c5926c789f9d#e2a1c3b7ef2191c224f998b94332c5926c789f9d" dependencies = [ "rpds", "serde", @@ -6073,7 +6073,7 @@ dependencies = [ [[package]] name = "logos-blockchain-groth16" version = "0.0.0" -source = "git+https://github.com/logos-blockchain/logos-blockchain.git?rev=0dc34e2c5a6e2ff772140378659489a602ee06bc#0dc34e2c5a6e2ff772140378659489a602ee06bc" +source = "git+https://github.com/logos-blockchain/logos-blockchain.git?rev=e2a1c3b7ef2191c224f998b94332c5926c789f9d#e2a1c3b7ef2191c224f998b94332c5926c789f9d" dependencies = [ "ark-bn254", "ark-ec", @@ -6092,7 +6092,7 @@ dependencies = [ [[package]] name = "logos-blockchain-http-api-common" version = "0.0.0" -source = "git+https://github.com/logos-blockchain/logos-blockchain.git?rev=0dc34e2c5a6e2ff772140378659489a602ee06bc#0dc34e2c5a6e2ff772140378659489a602ee06bc" +source = "git+https://github.com/logos-blockchain/logos-blockchain.git?rev=e2a1c3b7ef2191c224f998b94332c5926c789f9d#e2a1c3b7ef2191c224f998b94332c5926c789f9d" dependencies = [ "axum 0.7.9", "logos-blockchain-core", @@ -6113,7 +6113,7 @@ dependencies = [ [[package]] name = "logos-blockchain-key-management-system-keys" version = "0.0.0" -source = "git+https://github.com/logos-blockchain/logos-blockchain.git?rev=0dc34e2c5a6e2ff772140378659489a602ee06bc#0dc34e2c5a6e2ff772140378659489a602ee06bc" +source = "git+https://github.com/logos-blockchain/logos-blockchain.git?rev=e2a1c3b7ef2191c224f998b94332c5926c789f9d#e2a1c3b7ef2191c224f998b94332c5926c789f9d" dependencies = [ "async-trait", "bytes", @@ -6141,7 +6141,7 @@ dependencies = [ [[package]] name = "logos-blockchain-key-management-system-macros" version = "0.0.0" -source = "git+https://github.com/logos-blockchain/logos-blockchain.git?rev=0dc34e2c5a6e2ff772140378659489a602ee06bc#0dc34e2c5a6e2ff772140378659489a602ee06bc" +source = "git+https://github.com/logos-blockchain/logos-blockchain.git?rev=e2a1c3b7ef2191c224f998b94332c5926c789f9d#e2a1c3b7ef2191c224f998b94332c5926c789f9d" dependencies = [ "proc-macro2", "quote", @@ -6151,7 +6151,7 @@ dependencies = [ [[package]] name = "logos-blockchain-key-management-system-operators" version = "0.0.0" -source = "git+https://github.com/logos-blockchain/logos-blockchain.git?rev=0dc34e2c5a6e2ff772140378659489a602ee06bc#0dc34e2c5a6e2ff772140378659489a602ee06bc" +source = "git+https://github.com/logos-blockchain/logos-blockchain.git?rev=e2a1c3b7ef2191c224f998b94332c5926c789f9d#e2a1c3b7ef2191c224f998b94332c5926c789f9d" dependencies = [ "async-trait", "logos-blockchain-blend-proofs", @@ -6169,7 +6169,7 @@ dependencies = [ [[package]] name = "logos-blockchain-key-management-system-service" version = "0.0.0" -source = "git+https://github.com/logos-blockchain/logos-blockchain.git?rev=0dc34e2c5a6e2ff772140378659489a602ee06bc#0dc34e2c5a6e2ff772140378659489a602ee06bc" +source = "git+https://github.com/logos-blockchain/logos-blockchain.git?rev=e2a1c3b7ef2191c224f998b94332c5926c789f9d#e2a1c3b7ef2191c224f998b94332c5926c789f9d" dependencies = [ "async-trait", "logos-blockchain-key-management-system-keys", @@ -6186,7 +6186,7 @@ dependencies = [ [[package]] name = "logos-blockchain-ledger" version = "0.0.0" -source = "git+https://github.com/logos-blockchain/logos-blockchain.git?rev=0dc34e2c5a6e2ff772140378659489a602ee06bc#0dc34e2c5a6e2ff772140378659489a602ee06bc" +source = "git+https://github.com/logos-blockchain/logos-blockchain.git?rev=e2a1c3b7ef2191c224f998b94332c5926c789f9d#e2a1c3b7ef2191c224f998b94332c5926c789f9d" dependencies = [ "derivative", "logos-blockchain-blend-crypto", @@ -6212,7 +6212,7 @@ dependencies = [ [[package]] name = "logos-blockchain-libp2p" version = "0.0.0" -source = "git+https://github.com/logos-blockchain/logos-blockchain.git?rev=0dc34e2c5a6e2ff772140378659489a602ee06bc#0dc34e2c5a6e2ff772140378659489a602ee06bc" +source = "git+https://github.com/logos-blockchain/logos-blockchain.git?rev=e2a1c3b7ef2191c224f998b94332c5926c789f9d#e2a1c3b7ef2191c224f998b94332c5926c789f9d" dependencies = [ "async-trait", "backon", @@ -6241,7 +6241,7 @@ dependencies = [ [[package]] name = "logos-blockchain-log-targets" version = "0.0.0" -source = "git+https://github.com/logos-blockchain/logos-blockchain.git?rev=0dc34e2c5a6e2ff772140378659489a602ee06bc#0dc34e2c5a6e2ff772140378659489a602ee06bc" +source = "git+https://github.com/logos-blockchain/logos-blockchain.git?rev=e2a1c3b7ef2191c224f998b94332c5926c789f9d#e2a1c3b7ef2191c224f998b94332c5926c789f9d" dependencies = [ "logos-blockchain-log-targets-macros", ] @@ -6249,7 +6249,7 @@ dependencies = [ [[package]] name = "logos-blockchain-log-targets-macros" version = "0.0.0" -source = "git+https://github.com/logos-blockchain/logos-blockchain.git?rev=0dc34e2c5a6e2ff772140378659489a602ee06bc#0dc34e2c5a6e2ff772140378659489a602ee06bc" +source = "git+https://github.com/logos-blockchain/logos-blockchain.git?rev=e2a1c3b7ef2191c224f998b94332c5926c789f9d#e2a1c3b7ef2191c224f998b94332c5926c789f9d" dependencies = [ "proc-macro2", "quote", @@ -6259,7 +6259,7 @@ dependencies = [ [[package]] name = "logos-blockchain-merkle-tree" version = "0.0.0" -source = "git+https://github.com/logos-blockchain/logos-blockchain.git?rev=0dc34e2c5a6e2ff772140378659489a602ee06bc#0dc34e2c5a6e2ff772140378659489a602ee06bc" +source = "git+https://github.com/logos-blockchain/logos-blockchain.git?rev=e2a1c3b7ef2191c224f998b94332c5926c789f9d#e2a1c3b7ef2191c224f998b94332c5926c789f9d" dependencies = [ "logos-blockchain-dynamic-merkle", "rpds", @@ -6270,7 +6270,7 @@ dependencies = [ [[package]] name = "logos-blockchain-mmr" version = "0.0.0" -source = "git+https://github.com/logos-blockchain/logos-blockchain.git?rev=0dc34e2c5a6e2ff772140378659489a602ee06bc#0dc34e2c5a6e2ff772140378659489a602ee06bc" +source = "git+https://github.com/logos-blockchain/logos-blockchain.git?rev=e2a1c3b7ef2191c224f998b94332c5926c789f9d#e2a1c3b7ef2191c224f998b94332c5926c789f9d" dependencies = [ "ark-ff", "logos-blockchain-groth16", @@ -6284,7 +6284,7 @@ dependencies = [ [[package]] name = "logos-blockchain-network-service" version = "0.0.0" -source = "git+https://github.com/logos-blockchain/logos-blockchain.git?rev=0dc34e2c5a6e2ff772140378659489a602ee06bc#0dc34e2c5a6e2ff772140378659489a602ee06bc" +source = "git+https://github.com/logos-blockchain/logos-blockchain.git?rev=e2a1c3b7ef2191c224f998b94332c5926c789f9d#e2a1c3b7ef2191c224f998b94332c5926c789f9d" dependencies = [ "async-trait", "futures", @@ -6306,7 +6306,7 @@ dependencies = [ [[package]] name = "logos-blockchain-poc" version = "0.0.0" -source = "git+https://github.com/logos-blockchain/logos-blockchain.git?rev=0dc34e2c5a6e2ff772140378659489a602ee06bc#0dc34e2c5a6e2ff772140378659489a602ee06bc" +source = "git+https://github.com/logos-blockchain/logos-blockchain.git?rev=e2a1c3b7ef2191c224f998b94332c5926c789f9d#e2a1c3b7ef2191c224f998b94332c5926c789f9d" dependencies = [ "logos-blockchain-circuits-poc-sys", "logos-blockchain-circuits-prover", @@ -6323,7 +6323,7 @@ dependencies = [ [[package]] name = "logos-blockchain-pol" version = "0.0.0" -source = "git+https://github.com/logos-blockchain/logos-blockchain.git?rev=0dc34e2c5a6e2ff772140378659489a602ee06bc#0dc34e2c5a6e2ff772140378659489a602ee06bc" +source = "git+https://github.com/logos-blockchain/logos-blockchain.git?rev=e2a1c3b7ef2191c224f998b94332c5926c789f9d#e2a1c3b7ef2191c224f998b94332c5926c789f9d" dependencies = [ "astro-float", "logos-blockchain-circuits-pol-sys", @@ -6343,7 +6343,7 @@ dependencies = [ [[package]] name = "logos-blockchain-poq" version = "0.0.0" -source = "git+https://github.com/logos-blockchain/logos-blockchain.git?rev=0dc34e2c5a6e2ff772140378659489a602ee06bc#0dc34e2c5a6e2ff772140378659489a602ee06bc" +source = "git+https://github.com/logos-blockchain/logos-blockchain.git?rev=e2a1c3b7ef2191c224f998b94332c5926c789f9d#e2a1c3b7ef2191c224f998b94332c5926c789f9d" dependencies = [ "logos-blockchain-circuits-poq-sys", "logos-blockchain-circuits-prover", @@ -6362,7 +6362,7 @@ dependencies = [ [[package]] name = "logos-blockchain-poseidon2" version = "0.0.0" -source = "git+https://github.com/logos-blockchain/logos-blockchain.git?rev=0dc34e2c5a6e2ff772140378659489a602ee06bc#0dc34e2c5a6e2ff772140378659489a602ee06bc" +source = "git+https://github.com/logos-blockchain/logos-blockchain.git?rev=e2a1c3b7ef2191c224f998b94332c5926c789f9d#e2a1c3b7ef2191c224f998b94332c5926c789f9d" dependencies = [ "ark-bn254", "ark-ff", @@ -6373,7 +6373,7 @@ dependencies = [ [[package]] name = "logos-blockchain-proofs-error" version = "0.0.0" -source = "git+https://github.com/logos-blockchain/logos-blockchain.git?rev=0dc34e2c5a6e2ff772140378659489a602ee06bc#0dc34e2c5a6e2ff772140378659489a602ee06bc" +source = "git+https://github.com/logos-blockchain/logos-blockchain.git?rev=e2a1c3b7ef2191c224f998b94332c5926c789f9d#e2a1c3b7ef2191c224f998b94332c5926c789f9d" dependencies = [ "logos-blockchain-circuits-types", "logos-blockchain-groth16", @@ -6384,7 +6384,7 @@ dependencies = [ [[package]] name = "logos-blockchain-services-utils" version = "0.0.0" -source = "git+https://github.com/logos-blockchain/logos-blockchain.git?rev=0dc34e2c5a6e2ff772140378659489a602ee06bc#0dc34e2c5a6e2ff772140378659489a602ee06bc" +source = "git+https://github.com/logos-blockchain/logos-blockchain.git?rev=e2a1c3b7ef2191c224f998b94332c5926c789f9d#e2a1c3b7ef2191c224f998b94332c5926c789f9d" dependencies = [ "async-trait", "bytes", @@ -6400,7 +6400,7 @@ dependencies = [ [[package]] name = "logos-blockchain-storage-service" version = "0.0.0" -source = "git+https://github.com/logos-blockchain/logos-blockchain.git?rev=0dc34e2c5a6e2ff772140378659489a602ee06bc#0dc34e2c5a6e2ff772140378659489a602ee06bc" +source = "git+https://github.com/logos-blockchain/logos-blockchain.git?rev=e2a1c3b7ef2191c224f998b94332c5926c789f9d#e2a1c3b7ef2191c224f998b94332c5926c789f9d" dependencies = [ "async-trait", "bytes", @@ -6421,7 +6421,7 @@ dependencies = [ [[package]] name = "logos-blockchain-time-service" version = "0.0.0" -source = "git+https://github.com/logos-blockchain/logos-blockchain.git?rev=0dc34e2c5a6e2ff772140378659489a602ee06bc#0dc34e2c5a6e2ff772140378659489a602ee06bc" +source = "git+https://github.com/logos-blockchain/logos-blockchain.git?rev=e2a1c3b7ef2191c224f998b94332c5926c789f9d#e2a1c3b7ef2191c224f998b94332c5926c789f9d" dependencies = [ "async-trait", "futures", @@ -6444,7 +6444,7 @@ dependencies = [ [[package]] name = "logos-blockchain-tracing" version = "0.0.0" -source = "git+https://github.com/logos-blockchain/logos-blockchain.git?rev=0dc34e2c5a6e2ff772140378659489a602ee06bc#0dc34e2c5a6e2ff772140378659489a602ee06bc" +source = "git+https://github.com/logos-blockchain/logos-blockchain.git?rev=e2a1c3b7ef2191c224f998b94332c5926c789f9d#e2a1c3b7ef2191c224f998b94332c5926c789f9d" dependencies = [ "flate2", "logos-blockchain-log-targets", @@ -6470,7 +6470,7 @@ dependencies = [ [[package]] name = "logos-blockchain-utils" version = "0.0.0" -source = "git+https://github.com/logos-blockchain/logos-blockchain.git?rev=0dc34e2c5a6e2ff772140378659489a602ee06bc#0dc34e2c5a6e2ff772140378659489a602ee06bc" +source = "git+https://github.com/logos-blockchain/logos-blockchain.git?rev=e2a1c3b7ef2191c224f998b94332c5926c789f9d#e2a1c3b7ef2191c224f998b94332c5926c789f9d" dependencies = [ "async-trait", "blake2", @@ -6495,7 +6495,7 @@ dependencies = [ [[package]] name = "logos-blockchain-utxotree" version = "0.0.0" -source = "git+https://github.com/logos-blockchain/logos-blockchain.git?rev=0dc34e2c5a6e2ff772140378659489a602ee06bc#0dc34e2c5a6e2ff772140378659489a602ee06bc" +source = "git+https://github.com/logos-blockchain/logos-blockchain.git?rev=e2a1c3b7ef2191c224f998b94332c5926c789f9d#e2a1c3b7ef2191c224f998b94332c5926c789f9d" dependencies = [ "ark-ff", "logos-blockchain-dynamic-merkle", @@ -6508,7 +6508,7 @@ dependencies = [ [[package]] name = "logos-blockchain-zksign" version = "0.0.0" -source = "git+https://github.com/logos-blockchain/logos-blockchain.git?rev=0dc34e2c5a6e2ff772140378659489a602ee06bc#0dc34e2c5a6e2ff772140378659489a602ee06bc" +source = "git+https://github.com/logos-blockchain/logos-blockchain.git?rev=e2a1c3b7ef2191c224f998b94332c5926c789f9d#e2a1c3b7ef2191c224f998b94332c5926c789f9d" dependencies = [ "logos-blockchain-circuits-prover", "logos-blockchain-circuits-signature-sys", @@ -6528,7 +6528,7 @@ dependencies = [ [[package]] name = "logos-blockchain-zone-sdk" version = "0.0.0" -source = "git+https://github.com/logos-blockchain/logos-blockchain.git?rev=0dc34e2c5a6e2ff772140378659489a602ee06bc#0dc34e2c5a6e2ff772140378659489a602ee06bc" +source = "git+https://github.com/logos-blockchain/logos-blockchain.git?rev=e2a1c3b7ef2191c224f998b94332c5926c789f9d#e2a1c3b7ef2191c224f998b94332c5926c789f9d" dependencies = [ "async-trait", "futures", diff --git a/Cargo.toml b/Cargo.toml index d5ebbe48..dc5df07d 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -182,14 +182,14 @@ schemars = "1.2" async-stream = "0.3.6" strum = { version = "0.28.0", features = ["derive"] } -logos-blockchain-common-http-client = { git = "https://github.com/logos-blockchain/logos-blockchain.git", rev = "0dc34e2c5a6e2ff772140378659489a602ee06bc" } -logos-blockchain-key-management-system-service = { git = "https://github.com/logos-blockchain/logos-blockchain.git", rev = "0dc34e2c5a6e2ff772140378659489a602ee06bc" } -logos-blockchain-codec = { git = "https://github.com/logos-blockchain/logos-blockchain.git", rev = "0dc34e2c5a6e2ff772140378659489a602ee06bc" } -logos-blockchain-core = { git = "https://github.com/logos-blockchain/logos-blockchain.git", rev = "0dc34e2c5a6e2ff772140378659489a602ee06bc" } -logos-blockchain-chain-broadcast-service = { git = "https://github.com/logos-blockchain/logos-blockchain.git", rev = "0dc34e2c5a6e2ff772140378659489a602ee06bc" } -logos-blockchain-chain-service = { git = "https://github.com/logos-blockchain/logos-blockchain.git", rev = "0dc34e2c5a6e2ff772140378659489a602ee06bc" } -logos-blockchain-zone-sdk = { git = "https://github.com/logos-blockchain/logos-blockchain.git", rev = "0dc34e2c5a6e2ff772140378659489a602ee06bc" } -logos-blockchain-http-api-common = { git = "https://github.com/logos-blockchain/logos-blockchain.git", rev = "0dc34e2c5a6e2ff772140378659489a602ee06bc" } +logos-blockchain-common-http-client = { git = "https://github.com/logos-blockchain/logos-blockchain.git", rev = "e2a1c3b7ef2191c224f998b94332c5926c789f9d" } +logos-blockchain-key-management-system-service = { git = "https://github.com/logos-blockchain/logos-blockchain.git", rev = "e2a1c3b7ef2191c224f998b94332c5926c789f9d" } +logos-blockchain-codec = { git = "https://github.com/logos-blockchain/logos-blockchain.git", rev = "e2a1c3b7ef2191c224f998b94332c5926c789f9d" } +logos-blockchain-core = { git = "https://github.com/logos-blockchain/logos-blockchain.git", rev = "e2a1c3b7ef2191c224f998b94332c5926c789f9d" } +logos-blockchain-chain-broadcast-service = { git = "https://github.com/logos-blockchain/logos-blockchain.git", rev = "e2a1c3b7ef2191c224f998b94332c5926c789f9d" } +logos-blockchain-chain-service = { git = "https://github.com/logos-blockchain/logos-blockchain.git", rev = "e2a1c3b7ef2191c224f998b94332c5926c789f9d" } +logos-blockchain-zone-sdk = { git = "https://github.com/logos-blockchain/logos-blockchain.git", rev = "e2a1c3b7ef2191c224f998b94332c5926c789f9d" } +logos-blockchain-http-api-common = { git = "https://github.com/logos-blockchain/logos-blockchain.git", rev = "e2a1c3b7ef2191c224f998b94332c5926c789f9d" } keycard-rs = { git = "https://github.com/keycard-tech/keycard-rs", rev = "9535a657ba04b1e6916de51777e22b4837c1a84d" }