From 80a56be5aae702615bf7f50ad892ccce528ee608 Mon Sep 17 00:00:00 2001 From: erhant Date: Thu, 6 Aug 2026 20:42:18 +0300 Subject: [PATCH] fix(sequencer): properly adopt / orphan blocks in false-pos from L1 --- lez/sequencer/core/src/block_store.rs | 11 ++ lez/sequencer/core/src/lib.rs | 169 ++++++++++++++---- lez/sequencer/core/src/tests.rs | 73 ++++++++ .../core/src/tests/reconstruction.rs | 163 ++++++++++++++++- lez/sequencer/service/src/lib.rs | 15 +- lez/storage/src/sequencer/mod.rs | 30 +++- lez/storage/src/sequencer/sequencer_cells.rs | 29 ++- 7 files changed, 441 insertions(+), 49 deletions(-) diff --git a/lez/sequencer/core/src/block_store.rs b/lez/sequencer/core/src/block_store.rs index df61b38b..21b96c7f 100644 --- a/lez/sequencer/core/src/block_store.rs +++ b/lez/sequencer/core/src/block_store.rs @@ -215,6 +215,17 @@ impl SequencerStore { self.dbio.put_zone_anchor(anchor) } + /// The highest block id ever inscribed on the channel by this sequencer, + /// or `None` before it has published anything. + pub fn published_high_water(&self) -> DbResult> { + self.dbio.published_high_water() + } + + /// Raises the published high water mark to `block_id`, never lowering it. + pub fn raise_published_high_water(&self, block_id: u64) -> DbResult<()> { + self.dbio.raise_published_high_water(block_id) + } + pub fn get_pending_deposit_events(&self) -> DbResult> { self.dbio.get_pending_deposit_events() } diff --git a/lez/sequencer/core/src/lib.rs b/lez/sequencer/core/src/lib.rs index e38a19ca..93de58b0 100644 --- a/lez/sequencer/core/src/lib.rs +++ b/lez/sequencer/core/src/lib.rs @@ -21,7 +21,7 @@ use futures::StreamExt as _; use itertools::Itertools as _; use lee::{AccountId, PublicTransaction, public_transaction::Message}; use lee_core::GENESIS_BLOCK_ID; -use log::{error, info, warn}; +use log::{debug, error, info, warn}; use logos_blockchain_key_management_system_service::keys::{ED25519_SECRET_KEY_SIZE, Ed25519Key}; use logos_blockchain_zone_sdk::{ Slot, ZoneMessage, @@ -262,6 +262,21 @@ impl SequencerCore { .await .expect("Failed to verify/reconstruct sequencer state from Bedrock"); + // Seed the high water mark from the tip we are starting on. Every stored + // block reached the store by being published or by being adopted from + // the channel, so the channel holds them all and none is ours to write + // again. Without this the mark is absent until the first publish of this + // run, leaving that window unguarded — which is exactly the window a + // store written before the mark existed starts in. + if let Some(tip) = store + .latest_block_meta() + .expect("Failed to read latest block meta") + { + store + .raise_published_high_water(tip.id) + .expect("Failed to seed published high water mark"); + } + // Publish our blocks only when we are bootstrapping a channel that does // not exist yet (no channel tip). If the channel already exists (another // sequencer created it), we adopted its blocks during reconstruction @@ -293,6 +308,9 @@ impl SequencerCore { ) }); last_checkpoint = Some(outcome.checkpoint); + store + .raise_published_high_water(block.header.block_id) + .expect("Failed to persist published high water mark"); } // These blocks are already stored, so only the sdk's pending set @@ -322,7 +340,8 @@ impl SequencerCore { /// Verifies the local store still belongs to the chain the connected channel /// serves and replays any finalized channel blocks missing locally into /// `state`/`store`, recording each block's L1 inscription slot as the new - /// anchor. Fails (never parks) on any divergence. + /// anchor. Fails (never parks) when the channel proves a different chain: + /// the anchor consistency check, or a block that will not validate. /// /// Returns whether the channel does not exist yet (has no tip), i.e. whether /// this sequencer is the one that must bootstrap-publish its own blocks. @@ -440,8 +459,9 @@ impl SequencerCore { } /// Applies a single channel block during reconstruction: idempotent for - /// blocks we already hold (verifying their hash), a validated continuation - /// for new ones. Advances the persisted anchor to the block's slot. + /// blocks we already hold, ignored when it conflicts at a height the final + /// tier already settled, a validated continuation otherwise. Advances the + /// persisted anchor to the block's slot. fn apply_reconstructed_block( store: &SequencerStore, chain: &mut ChainState, @@ -460,44 +480,44 @@ impl SequencerCore { hash: block_hash, }; - // A block at/below the tip must match what we already stored, otherwise - // the channel is a different chain. + // A block we already hold verbatim needs no replay, but the channel + // serving it is what makes it irreversible, so its deliveries are + // settled and their records are owed nothing. Without this a restart + // leaves a record for every delivery it already published, and nothing + // downstream would ever remove them. if let Some(tip) = &tip && block_id <= tip.id - { - match store + && let Some(stored) = store .get_block_at_id(block_id) .context("Failed to read stored block")? - { - Some(stored) if stored.header.hash == block_hash => { - // Already applied, but the channel serving it is what makes - // it irreversible, so its deliveries are settled and their - // records are owed nothing. Without this a restart leaves a - // record for every delivery it already published, and - // nothing downstream would ever remove them. - settle_reconstructed_deliveries(store, &stored); - store - .set_zone_anchor(&record) - .context("Failed to persist zone anchor")?; - return Ok(()); - } - Some(stored) => { - return Err(anyhow!( - "Channel block {block_id} hash {block_hash} does not match stored hash {}", - stored.header.hash - )); - } - None => { - return Err(anyhow!( - "Channel block {block_id} is at/below local tip {} but is missing locally", - tip.id - )); - } - } + && stored.header.hash == block_hash + { + settle_reconstructed_deliveries(store, &stored); + store + .set_zone_anchor(&record) + .context("Failed to persist zone anchor")?; + return Ok(()); } - // New continuation: channel history is finalized, so it goes through - // the final tier — validation happens inside `apply_finalized`. + // A conflict at a height the final tier already settled: the channel + // carries two inscriptions for one block id — competing sequencers + // around a turn change — and finality already picked one, so the other + // is dropped. `apply_adopted` ignores the same conflict. A genuinely + // foreign channel is caught upstream by the anchor consistency check, + // not here; the anchor stays on the block we hold. + if let Some(final_tip) = chain.final_tip() + && block_id <= final_tip.block_id + { + log::warn!( + "Ignoring channel block {block_id} with hash {block_hash} conflicting with the \ + finalized block at this height" + ); + return Ok(()); + } + + // Above the final tier the head is reorg-able, so finalized history + // wins: `apply_finalized` finalizes the matching prefix and rebases the + // head onto what the channel settled. Validation happens inside it. match chain.apply_finalized(MsgId::from(block.header.hash.0), block, slot) { AcceptOutcome::Applied | AcceptOutcome::AlreadyApplied => {} AcceptOutcome::Parked(err) | AcceptOutcome::RetryableFailure(err) => { @@ -569,6 +589,12 @@ impl SequencerCore { .await .context("Failed to publish block to Bedrock")?; + // The inscription is on L1 from here on, whatever the head does with the + // block below, so this height must never be published again. + self.store + .raise_published_high_water(block.header.block_id) + .context("Failed to persist published high water mark")?; + let withdrawal_reconciliation_keys: Vec<_> = released_notes .iter() .map(withdrawal_reconciliation_key) @@ -1106,6 +1132,37 @@ impl SequencerCore { self.block_publisher.is_our_turn() } + /// The height the next produced block would claim. + #[must_use] + pub fn next_block_height(&self) -> u64 { + self.chain + .lock() + .expect("chain state mutex poisoned") + .head_tip() + .map_or(GENESIS_BLOCK_ID, |tip| { + tip.block_id + .checked_add(1) + .expect("block id should not overflow") + }) + } + + /// `Some(high_water)` when the head has rewound below what we already + /// inscribed, so the next block would be a *second*, different block at a + /// height the channel already carries. Callers must skip their turn. + /// + /// The head alone cannot detect this: an orphan report for our own + /// still-unfinalized blocks rewinds it (`ChainState::apply_channel_update`) + /// and prunes those blocks from the store, so the tip reads as if they were + /// never produced. The mark is kept outside that pruning for exactly this. + /// + /// This is not a stall — the head recovers by itself once the inscriptions + /// we are protecting finalize and the final tier rebases onto them. + #[must_use] + pub fn rewound_below_published(&self) -> Option { + let high_water = self.store.published_high_water().ok().flatten()?; + (self.next_block_height() <= high_water).then_some(high_water) + } + /// Shared handle to the two-tier follow state, for tests to drive the /// follow path directly. #[cfg(all(test, feature = "mock"))] @@ -1201,8 +1258,48 @@ fn apply_follow_update( let (resubmit_txs, outcome, head_height) = { let mut chain = chain.lock().expect("chain state mutex poisoned"); + // An orphan report rewinds the head to the earliest orphaned block and + // prunes the store above it, which is how a run of our own inscriptions + // can silently stop being ours. Loud on the way in: it is the only + // trace, and the rewind it causes is the expensive one. + // Debug, not warn: the sdk orphans our blocks routinely once LIB pruning + // drops them from the lineage, and most of those no longer sit in the + // head. The rewind below is the part that costs something. + let head_before = chain.head_tip().map(|tip| tip.block_id); + if !orphaned.is_empty() { + let ids: Vec = orphaned + .iter() + .map(|(_, block)| block.header.block_id) + .collect(); + debug!( + "Channel orphaned {} block(s) {:?}..={:?}, head tip is {head_before:?}", + ids.len(), + ids.iter().min(), + ids.iter().max(), + ); + } + // Outcomes align with `adopted`. let outcomes = chain.apply_channel_update(&orphaned, &adopted); + + // An adoption that does not apply freezes the head where it is, and + // every later one then fails the same way. Nothing else reports it. + for ((_, block), outcome) in adopted.iter().zip(&outcomes) { + if let AcceptOutcome::Parked(err) | AcceptOutcome::RetryableFailure(err) = outcome { + warn!( + "Adopted block {} did not apply, head stays at {:?}: {err}", + block.header.block_id, + chain.head_tip().map(|tip| tip.block_id), + ); + } + } + + if let (Some(before), Some(after)) = (head_before, chain.head_tip().map(|tip| tip.block_id)) + && after < before + { + warn!("Head rewound from {before} to {after}"); + } + let mut to_persist: Vec<(&Block, bool)> = adopted .iter() .zip(&outcomes) diff --git a/lez/sequencer/core/src/tests.rs b/lez/sequencer/core/src/tests.rs index 8d3f787b..ae0a6c31 100644 --- a/lez/sequencer/core/src/tests.rs +++ b/lez/sequencer/core/src/tests.rs @@ -2011,6 +2011,79 @@ async fn follow_update_persists_the_checkpoint_with_its_effects() { assert!(sequencer.store.get_block_at_id(2).unwrap().is_some()); } +/// The channel orphaning our own still-unfinalized blocks rewinds the head and +/// prunes them from the store, so nothing in the chain state remembers we ever +/// produced them. Producing again there would put a second, different block at +/// a height the channel already carries — the fork that has to be prevented. +#[tokio::test] +async fn head_rewound_below_published_height_blocks_production() { + let config = setup_sequencer_config(); + let (mut sequencer, mempool_handle) = + SequencerCoreWithMockClients::start_from_config(config).await; + + let first = sequencer.produce_new_block().await.unwrap(); + let published_tip = sequencer.produce_new_block().await.unwrap(); + assert_eq!( + sequencer.store.published_high_water().unwrap(), + Some(published_tip), + "publishing records the high water mark" + ); + assert!( + sequencer.rewound_below_published().is_none(), + "an intact head is free to produce" + ); + + let produced: Vec = [first, published_tip] + .into_iter() + .map(|id| sequencer.store.get_block_at_id(id).unwrap().unwrap()) + .collect(); + + // The sdk reports both of them as orphaned. + apply_follow_update( + &sequencer.store.dbio(), + &sequencer.chain(), + &mempool_handle, + FollowUpdate { + orphaned: produced + .iter() + .map(|block| (MsgId::from([0_u8; 32]), block.clone())) + .collect(), + ..empty_follow_update() + }, + ); + + assert!( + sequencer.store.latest_block_meta().unwrap().unwrap().id < published_tip, + "the orphan report rewound the stored tip" + ); + assert_eq!( + sequencer.rewound_below_published(), + Some(published_tip), + "the mark outlives the pruning and blocks the turn" + ); + + // Those inscriptions were on the channel all along: finalizing them rebases + // the head onto them, and production is free again. The guard is a wait. + apply_follow_update( + &sequencer.store.dbio(), + &sequencer.chain(), + &mempool_handle, + FollowUpdate { + finalized: produced + .iter() + .map(|block| (MsgId::from([1_u8; 32]), block.clone())) + .collect(), + ..empty_follow_update() + }, + ); + + assert_eq!(sequencer.next_block_height(), published_tip + 1); + assert!( + sequencer.rewound_below_published().is_none(), + "a recovered head resumes producing" + ); +} + #[tokio::test] async fn follow_update_records_deposits_for_the_production_drain() { let config = setup_sequencer_config(); diff --git a/lez/sequencer/core/src/tests/reconstruction.rs b/lez/sequencer/core/src/tests/reconstruction.rs index 846abe20..6365fca8 100644 --- a/lez/sequencer/core/src/tests/reconstruction.rs +++ b/lez/sequencer/core/src/tests/reconstruction.rs @@ -190,15 +190,17 @@ async fn fails_when_channel_reinscribes_genesis_with_a_different_hash() { } #[tokio::test] -async fn fails_when_a_stored_block_hash_diverges_from_the_channel() { +async fn fails_when_a_below_tip_channel_block_does_not_validate() { // A sequencer that committed blocks past genesis but never recorded an anchor. let config = setup_sequencer_config(); let (mut seq, _handle) = SequencerCoreWithMockClients::start_from_config(config.clone()).await; seq.produce_new_block().await.unwrap(); seq.produce_new_block().await.unwrap(); - // A below-tip block re-served with a corrupted hash: we already hold this id - // with a different hash, so the channel is a different chain. + // A below-tip block re-served with a corrupted hash. Holding a different + // block at that id is not itself grounds to abort — the head tier is + // reorg-able — but this one's header hash does not cover its contents, so it + // parks on validation. let below_tip_id = seq.block_store().genesis_id() + 1; let mut block = seq .block_store() @@ -219,17 +221,18 @@ async fn fails_when_a_stored_block_hash_diverges_from_the_channel() { .await; assert!( result.is_err(), - "a diverging below-tip block hash must abort startup" + "an unverifiable below-tip block must abort startup" ); } #[tokio::test] -async fn fails_when_a_channel_block_is_missing_locally() { +async fn fails_when_a_channel_block_is_numbered_below_genesis() { let config = setup_sequencer_config(); let (store, chain) = fresh_store_and_chain(&config); - // A block numbered below our genesis is at/below the local tip yet absent from - // the store — a foreign chain with a lower numbering. + // A block numbered below our genesis — a foreign chain with a lower + // numbering. Nothing local sits at that id, so it goes straight to + // validation and parks there. let mut foreign = store.get_block_at_id(store.genesis_id()).unwrap().unwrap(); foreign.header.block_id = store.genesis_id() - 1; @@ -272,6 +275,152 @@ async fn fails_when_a_channel_block_does_not_extend_the_tip() { ); } +// The two cases below reproduce the real startup order: zone-sdk's cold-start +// backfill runs inside `BP::new` and populates the store *before* +// `verify_and_reconstruct`, so reconstruction re-reads history it already holds. +// A conflict there is a competing sequencer, not a foreign chain. + +/// The channel carries two inscriptions for one block id — competing sequencers +/// around a turn change — and the final tier already settled that height. +/// Finality is irreversible, so the loser is ignored rather than fatal. +#[tokio::test] +async fn reconstruction_ignores_a_duplicate_height_the_final_tier_settled() { + // Sequencer A's chain is what the channel finalized. + let config_a = setup_sequencer_config(); + let (mut seq_a, _mempool_a) = + SequencerCoreWithMockClients::start_from_config(config_a.clone()).await; + seq_a.produce_new_block().await.unwrap(); + let tip_a = seq_a.block_store().latest_block_meta().unwrap().unwrap(); + let mut messages = channel_from_store(seq_a.block_store(), 10); + let settled_slot = messages.last().unwrap().1; + + // Sequencer B: the cold-start backfill finalizes A's chain into its store. + let (seq_b, mempool_b) = + SequencerCoreWithMockClients::start_from_config(setup_sequencer_config()).await; + let finalized: Vec<(MsgId, Block)> = (seq_b.block_store().genesis_id()..=tip_a.id) + .map(|id| { + let block = seq_a.block_store().get_block_at_id(id).unwrap().unwrap(); + ( + MsgId::from([u8::try_from(id).expect("should be u8"); 32]), + block, + ) + }) + .collect(); + apply_follow_update( + &seq_b.store.dbio(), + &seq_b.chain(), + &mempool_b, + FollowUpdate { + finalized, + ..empty_follow_update() + }, + ); + assert_eq!( + seq_b + .chain() + .lock() + .unwrap() + .final_tip() + .expect("backfill finalized A's chain") + .block_id, + tip_a.id, + ); + + // A competitor published its own block at that same height. + let parent = seq_a + .block_store() + .get_block_at_id(tip_a.id - 1) + .unwrap() + .unwrap(); + let competitor = + common::test_utils::produce_dummy_block(tip_a.id, Some(parent.header.hash), vec![]); + assert_ne!(competitor.header.hash, tip_a.hash); + messages.push(block_to_channel_message(&competitor, 999)); + + let mock_b = MockBlockPublisher::with_canned_channel( + config_a.bedrock_config.channel_id, + Some(Slot::from(999)), + messages, + ); + SequencerCore::::verify_and_reconstruct( + &mock_b, + &seq_b.store, + &seq_b.chain, + true, + ) + .await + .expect("a duplicate height the final tier settled must not abort startup"); + + let tip_b = seq_b.block_store().latest_block_meta().unwrap().unwrap(); + assert_eq!(tip_b.hash, tip_a.hash, "the finalized block stands"); + + // The anchor tracks the block we hold, never the one we dropped. + let anchor = seq_b + .block_store() + .get_zone_anchor() + .unwrap() + .expect("anchor"); + assert_eq!(anchor.slot, settled_slot.into_inner()); + assert_eq!(anchor.hash, tip_a.hash); +} + +/// A block the head tier holds is reorg-able by construction, so finalized +/// channel history at that height wins and the head rebases onto it. +#[tokio::test] +async fn reconstruction_replaces_a_conflicting_head_block_with_finalized_history() { + // Sequencer A's chain is what the channel finalized. + let config_a = setup_sequencer_config(); + let (mut seq_a, _mempool_a) = + SequencerCoreWithMockClients::start_from_config(config_a.clone()).await; + seq_a.produce_new_block().await.unwrap(); + let tip_a = seq_a.block_store().latest_block_meta().unwrap().unwrap(); + let messages = channel_from_store(seq_a.block_store(), 10); + let tip_slot = messages.last().unwrap().1; + + // Sequencer B adopted a competitor at that height and never saw it finalize. + let (seq_b, mempool_b) = + SequencerCoreWithMockClients::start_from_config(setup_sequencer_config()).await; + let genesis_b = seq_b.block_store().latest_block_meta().unwrap().unwrap(); + let competitor = + common::test_utils::produce_dummy_block(tip_a.id, Some(genesis_b.hash), vec![]); + assert_ne!(competitor.header.hash, tip_a.hash); + apply_follow_update( + &seq_b.store.dbio(), + &seq_b.chain(), + &mempool_b, + FollowUpdate { + adopted: vec![(MsgId::from([7_u8; 32]), competitor)], + ..empty_follow_update() + }, + ); + assert_eq!( + seq_b.block_store().latest_block_meta().unwrap().unwrap().id, + tip_a.id, + "the competitor is the head tip going in" + ); + + let mock_b = MockBlockPublisher::with_canned_channel( + config_a.bedrock_config.channel_id, + Some(tip_slot), + messages, + ); + SequencerCore::::verify_and_reconstruct( + &mock_b, + &seq_b.store, + &seq_b.chain, + true, + ) + .await + .expect("finalized history must replace a conflicting head block"); + + let tip_b = seq_b.block_store().latest_block_meta().unwrap().unwrap(); + assert_eq!(tip_b.id, tip_a.id); + assert_eq!( + tip_b.hash, tip_a.hash, + "the finalized block replaces the head competitor" + ); +} + /// A sequencer config whose genesis funds the bridge account, so replayed bridge /// deposit transactions have a source balance to mint from. fn bridge_funded_config() -> SequencerConfig { diff --git a/lez/sequencer/service/src/lib.rs b/lez/sequencer/service/src/lib.rs index 3073823a..683b8ad4 100644 --- a/lez/sequencer/service/src/lib.rs +++ b/lez/sequencer/service/src/lib.rs @@ -5,7 +5,7 @@ use bytesize::ByteSize; use common::transaction::LeeTransaction; use futures::never::Never; use jsonrpsee::server::ServerHandle; -use log::{error, info}; +use log::{error, info, warn}; use mempool::MemPoolHandle; #[cfg(not(feature = "standalone"))] use sequencer_core::SequencerCore; @@ -294,6 +294,19 @@ async fn main_loop(seq_core: Arc>, block_timeout: Duration) continue; } + // Never inscribe a second block at a height we already published: the + // channel would carry two chains from there and nothing resolves that. + // The head rewinds under us when the sdk orphans our own unfinalized + // blocks, and recovers once they finalize, so this is a wait. + if let Some(high_water) = state.rewound_below_published() { + warn!( + "Skipping turn: head rewound to {} but block {high_water} is already inscribed; \ + waiting for the channel to restore it", + state.next_block_height().saturating_sub(1), + ); + continue; + } + info!("Our turn: collecting transactions from mempool, creating block"); let id = state.produce_new_block().await?; info!("Block with id {id} created"); diff --git a/lez/storage/src/sequencer/mod.rs b/lez/storage/src/sequencer/mod.rs index ba6deaf9..ebb31e18 100644 --- a/lez/storage/src/sequencer/mod.rs +++ b/lez/storage/src/sequencer/mod.rs @@ -25,9 +25,9 @@ use crate::{ LatestBlockMetaCellOwned, LatestBlockMetaCellRef, PeerFloorCellOwned, PeerFloorCellRef, PeerZoneKey, PendingCrossZoneDispatchRecord, PendingCrossZoneDispatchesCellOwned, PendingCrossZoneDispatchesCellRef, PendingDepositEventRecord, - PendingDepositEventsCellOwned, PendingDepositEventsCellRef, UnseenWithdrawCountCell, - WithdrawalReconciliationKey, ZoneAnchorCell, ZoneAnchorRecord, ZoneSdkCheckpointCellOwned, - ZoneSdkCheckpointCellRef, + PendingDepositEventsCellOwned, PendingDepositEventsCellRef, PublishedHighWaterCell, + UnseenWithdrawCountCell, WithdrawalReconciliationKey, ZoneAnchorCell, ZoneAnchorRecord, + ZoneSdkCheckpointCellOwned, ZoneSdkCheckpointCellRef, }, }; @@ -56,6 +56,11 @@ pub const DB_META_PENDING_CROSS_ZONE_DISPATCHES_KEY: &str = "pending_cross_zone_ /// Key base for counting unseen L2 withdraw intents. pub const DB_META_UNSEEN_WITHDRAW_COUNT_KEY: &str = "unseen_withdraw_count"; +/// Key base for the highest block id this sequencer has ever inscribed on the +/// channel. Never decreases, and deliberately survives the block pruning a +/// head rewind performs. +pub const DB_META_PUBLISHED_HIGH_WATER_KEY: &str = "published_high_water"; + /// How many cross-zone deliveries may be pending at once. /// /// The whole list is a single value, read on every block and rewritten on every @@ -491,6 +496,25 @@ impl RocksDBIO { self.del::(()) } + /// The highest block id this sequencer has ever inscribed, or `None` if it + /// has never published. Read fresh: a head rewind prunes blocks, so the + /// stored tip is not a safe substitute. + pub fn published_high_water(&self) -> DbResult> { + self.get_opt::(()) + .map(|val| val.map(|cell| cell.0)) + } + + /// Raises the published high water mark to `block_id`, never lowering it. + pub fn raise_published_high_water(&self, block_id: u64) -> DbResult<()> { + if self + .published_high_water()? + .is_some_and(|mark| mark >= block_id) + { + return Ok(()); + } + self.put(&PublishedHighWaterCell(block_id), ()) + } + pub fn get_zone_anchor(&self) -> DbResult> { Ok(self.get_opt::(())?.map(|cell| cell.0)) } diff --git a/lez/storage/src/sequencer/sequencer_cells.rs b/lez/storage/src/sequencer/sequencer_cells.rs index 521fff5e..d7da254b 100644 --- a/lez/storage/src/sequencer/sequencer_cells.rs +++ b/lez/storage/src/sequencer/sequencer_cells.rs @@ -10,8 +10,9 @@ use crate::{ 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_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, }, }; @@ -134,6 +135,30 @@ impl SimpleWritableCell for LastFinalizedBlockIdCell { } } +/// The highest block id ever inscribed on the channel by this sequencer. +#[derive(Debug, BorshSerialize, BorshDeserialize)] +pub struct PublishedHighWaterCell(pub u64); + +impl SimpleStorableCell for PublishedHighWaterCell { + type KeyParams = (); + + const CELL_NAME: &'static str = DB_META_PUBLISHED_HIGH_WATER_KEY; + const CF_NAME: &'static str = CF_META_NAME; +} + +impl SimpleReadableCell for PublishedHighWaterCell {} + +impl SimpleWritableCell for PublishedHighWaterCell { + fn value_constructor(&self) -> DbResult> { + borsh::to_vec(&self).map_err(|err| { + DbError::borsh_cast_message( + err, + Some("Failed to serialize published high water mark".to_owned()), + ) + }) + } +} + #[derive(BorshDeserialize)] pub struct LatestBlockMetaCellOwned(pub BlockMeta);