Merge branch 'dev' into moudy/cross-zone-content-bound-key

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<Option<u64>> {
+        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<Vec<PendingDepositEventRecord>> {
         self.dbio.get_pending_deposit_events()
     }

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<BP: BlockPublisherTrait> SequencerCore<BP> {
                 .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<BP: BlockPublisherTrait> SequencerCore<BP> {
                         )
                     });
                 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<BP: BlockPublisherTrait> SequencerCore<BP> {
     /// 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<BP: BlockPublisherTrait> SequencerCore<BP> {
     }
 
     /// 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<BP: BlockPublisherTrait> SequencerCore<BP> {
             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<BP: BlockPublisherTrait> SequencerCore<BP> {
             .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<BP: BlockPublisherTrait> SequencerCore<BP> {
         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<u64> {
+        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<u64> = 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)

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<Block> = [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();

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::<MockBlockPublisher>::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::<MockBlockPublisher>::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 {

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<Mutex<SequencerCore>>, 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");

lez/storage/src/sequencer/mod.rs

@@ -26,6 +26,10 @@ 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,
     },
@@ -59,6 +63,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
@@ -494,6 +503,25 @@ impl RocksDBIO {
         self.del::<ZoneSdkCheckpointCellOwned>(())
     }
 
+    /// 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<Option<u64>> {
+        self.get_opt::<PublishedHighWaterCell>(())
+            .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<Option<ZoneAnchorRecord>> {
         Ok(self.get_opt::<ZoneAnchorCell>(())?.map(|cell| cell.0))
     }

lez/storage/src/sequencer/sequencer_cells.rs

@@ -11,6 +11,9 @@ 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,
     },
@@ -135,6 +138,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<Vec<u8>> {
+        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);