fix(storage): store pending cross-zone dispatches as per-message entries

lez/sequencer/core/src/cross_zone_watcher.rs

@@ -556,10 +556,8 @@ fn record_block_deliveries(
 ) -> bool {
     let peer_zone = peer.peer_zone;
     let self_zone = peer.self_zone;
-    // Collected and written once. The pending list is a single value, so a write
-    // per delivery would rewrite the whole list once per message, which is
-    // quadratic in a peer block that carries many of them, on a task holding the
-    // lock block production needs.
+    // Collected and written once, so recording a block is all-or-nothing; see
+    // RocksDBIO::add_pending_cross_zone_dispatches.
     let mut deliveries = Vec::new();
     for (index, tx) in block.body.transactions.iter().enumerate() {
         let LeeTransaction::Public(public_tx) = tx else {
@@ -666,7 +664,7 @@ mod tests {
     use logos_blockchain_core::mantle::ops::channel::{MsgId, inscribe::Inscription};
     use logos_blockchain_zone_sdk::ZoneBlock;
     use ping_core::{SenderInstruction, ping_record_pda, receiver_config_account_id};
-    use storage::sequencer::{DB_META_PENDING_CROSS_ZONE_DISPATCHES_KEY, RocksDBIO};
+    use storage::sequencer::{DB_META_PENDING_CROSS_ZONE_DISPATCH_COUNT_KEY, RocksDBIO};
     use tempfile::TempDir;
 
     use super::*;
@@ -792,27 +790,36 @@ mod tests {
         peer_msg(b"not a block".to_vec(), slot)
     }
 
-    /// The message keys recorded so far, in insertion order.
+    /// The message keys recorded so far, sorted: the store keys each record by
+    /// its message key, so no insertion order survives.
     fn recorded_keys(dbio: &RocksDBIO) -> Vec<[u8; 32]> {
-        dbio.get_pending_cross_zone_dispatches()
-            .expect("pending dispatches readable")
-            .into_iter()
-            .map(|record| record.message_key)
-            .collect()
+        sorted(
+            dbio.get_pending_cross_zone_dispatches()
+                .expect("pending dispatches readable")
+                .into_iter()
+                .map(|record| record.message_key)
+                .collect(),
+        )
     }
 
-    /// Makes every later pending-dispatch read fail, standing in for any store
-    /// failure between reading a peer block and the delivery being durable.
-    /// Recording reads the list before it writes it, so a value that will not
-    /// decode is enough.
+    /// `keys` as [`recorded_keys`] reports them.
+    fn sorted(mut keys: Vec<[u8; 32]>) -> Vec<[u8; 32]> {
+        keys.sort_unstable();
+        keys
+    }
+
+    /// Makes every later pending-dispatch write fail, standing in for any
+    /// store failure before a delivery is durable: recording reads the count
+    /// first, so a count that will not decode is enough.
     fn break_the_dispatch_store(dbio: &RocksDBIO) {
         let cf = dbio
             .db
             .cf_handle(storage::CF_META_NAME)
             .expect("meta column family");
-        let key = borsh::to_vec(&DB_META_PENDING_CROSS_ZONE_DISPATCHES_KEY).expect("key encodes");
+        let key =
+            borsh::to_vec(&DB_META_PENDING_CROSS_ZONE_DISPATCH_COUNT_KEY).expect("key encodes");
         dbio.db
-            .put_cf(&cf, key, b"not a pending dispatch list")
+            .put_cf(&cf, key, b"not a pending dispatch count")
             .expect("write");
     }
 
@@ -1071,7 +1078,10 @@ mod tests {
         );
         assert_eq!(
             recorded_keys(&dbio),
-            vec![message_key(&PEER_ZONE, 1, 0), message_key(&PEER_ZONE, 2, 0)]
+            sorted(vec![
+                message_key(&PEER_ZONE, 1, 0),
+                message_key(&PEER_ZONE, 2, 0)
+            ])
         );
     }
 
@@ -1289,11 +1299,11 @@ mod tests {
 
         assert_eq!(
             recorded_keys(&dbio),
-            vec![
+            sorted(vec![
                 message_key(&PEER_ZONE, 1, 0),
                 message_key(&PEER_ZONE, 2, 0),
                 message_key(&PEER_ZONE, 3, 0)
-            ],
+            ]),
             "only the unread block is recorded on the second pass"
         );
         assert_eq!(
@@ -1429,11 +1439,11 @@ mod tests {
         assert_eq!(outcome, PassOutcome::Drained);
         assert_eq!(
             recorded_keys(&dbio),
-            vec![
+            sorted(vec![
                 message_key(&PEER_ZONE, 1, 0),
                 message_key(&PEER_ZONE, 2, 0),
                 message_key(&PEER_ZONE, 3, 0)
-            ],
+            ]),
             "the key the peer aimed to burn is never recorded, and nothing else is held up"
         );
         assert_eq!(tip, Some(tip_at(3)));
@@ -1468,7 +1478,10 @@ mod tests {
         );
         assert_eq!(
             recorded_keys(&dbio),
-            vec![message_key(&PEER_ZONE, 1, 0), message_key(&PEER_ZONE, 2, 0)],
+            sorted(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, Some(tip_at(2)));
@@ -1499,7 +1512,10 @@ mod tests {
         assert_eq!(outcome, PassOutcome::Drained);
         assert_eq!(
             recorded_keys(&dbio),
-            vec![message_key(&PEER_ZONE, 1, 0), message_key(&PEER_ZONE, 2, 0)],
+            sorted(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, Some(tip_at(2)));
@@ -1558,7 +1574,10 @@ mod tests {
         assert_eq!(outcome, PassOutcome::Drained);
         assert_eq!(
             recorded_keys(&dbio),
-            vec![message_key(&PEER_ZONE, 1, 0), message_key(&PEER_ZONE, 2, 0)]
+            sorted(vec![
+                message_key(&PEER_ZONE, 1, 0),
+                message_key(&PEER_ZONE, 2, 0)
+            ])
         );
         assert_eq!(tip, Some(tip_at(2)));
     }

lez/storage/src/sequencer/mod.rs

@@ -10,9 +10,10 @@ use common::{
     block::{BedrockStatus, Block, BlockMeta},
 };
 use lee::V03State;
+use log::info;
 use rocksdb::{
-    BoundColumnFamily, ColumnFamilyDescriptor, DBWithThreadMode, IteratorMode, MultiThreaded,
-    Options, WriteBatch,
+    BoundColumnFamily, ColumnFamilyDescriptor, DBWithThreadMode, Direction, IteratorMode,
+    MultiThreaded, Options, WriteBatch,
 };
 
 use crate::{
@@ -24,12 +25,14 @@ use crate::{
         DeadLetterCrossZoneDispatchesCellRef, DeadLetterDispatchRecord, DispatchOrigin,
         FinalBlockMetaCellOwned, FinalBlockMetaCellRef, FinalLeeStateCellOwned,
         FinalLeeStateCellRef, LEEStateCellOwned, LEEStateCellRef, LastFinalizedBlockIdCell,
-        LatestBlockMetaCellOwned, LatestBlockMetaCellRef, PeerChainTip, PeerFloorCellOwned,
-        PeerFloorCellRef, PeerTipCell, PeerZoneKey, PendingCrossZoneDispatchRecord,
-        PendingCrossZoneDispatchesCellOwned, PendingCrossZoneDispatchesCellRef,
-        PendingDepositEventRecord, PendingDepositEventsCellOwned, PendingDepositEventsCellRef,
-        PublishedHighWaterCell, UnseenWithdrawCountCell, WithdrawalReconciliationKey,
-        ZoneAnchorCell, ZoneAnchorRecord, ZoneSdkCheckpointCellOwned, ZoneSdkCheckpointCellRef,
+        LatestBlockMetaCellOwned, LatestBlockMetaCellRef,
+        LegacyPendingCrossZoneDispatchesCellOwned, PeerChainTip, PeerFloorCellOwned,
+        PeerFloorCellRef, PeerTipCell, PeerZoneKey, PendingCrossZoneDispatchCellOwned,
+        PendingCrossZoneDispatchCellRef, PendingCrossZoneDispatchCountCell,
+        PendingCrossZoneDispatchRecord, PendingDepositEventRecord, PendingDepositEventsCellOwned,
+        PendingDepositEventsCellRef, PublishedHighWaterCell, UnseenWithdrawCountCell,
+        WithdrawalReconciliationKey, ZoneAnchorCell, ZoneAnchorRecord, ZoneSdkCheckpointCellOwned,
+        ZoneSdkCheckpointCellRef,
     },
 };
 
@@ -54,8 +57,13 @@ 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.
+/// Key base for storing one cross-zone delivery the watcher has recorded but
+/// which is not yet known to be irreversibly delivered. Keyed per message.
+pub const DB_META_PENDING_CROSS_ZONE_DISPATCH_KEY: &str = "pending_cross_zone_dispatch";
+/// Key base for counting the pending cross-zone dispatch records.
+pub const DB_META_PENDING_CROSS_ZONE_DISPATCH_COUNT_KEY: &str = "pending_cross_zone_dispatch_count";
+/// Key base under which older stores held the whole pending set as one borsh
+/// blob; kept only for migration on open.
 pub const DB_META_PENDING_CROSS_ZONE_DISPATCHES_KEY: &str = "pending_cross_zone_dispatches";
 /// Key base for storing cross-zone deliveries this node has given up on.
 pub const DB_META_DEAD_LETTER_CROSS_ZONE_DISPATCHES_KEY: &str = "dead_letter_cross_zone_dispatches";
@@ -74,11 +82,10 @@ 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
-/// change, and what fills it is chosen by peer zones rather than by us. Refusing
-/// to record past this bound turns "a peer decides how large our store gets"
-/// into "a peer's messages wait", since a watcher that cannot record holds its
-/// delivery floor and reads the slot again later.
+/// What fills the pending set is chosen by peer zones rather than by us.
+/// Refusing to record past this bound turns "a peer decides how large our
+/// store gets" into "a peer's messages wait", since a watcher that cannot
+/// record holds its delivery floor and reads the slot again later.
 pub const MAX_PENDING_CROSS_ZONE_DISPATCHES: usize = 4096;
 
 /// How many given-up-on cross-zone deliveries are kept for inspection.
@@ -238,13 +245,11 @@ pub struct StoreUpdateOutcome {
 pub struct RocksDBIO {
     pub db: DBWithThreadMode<MultiThreaded>,
     /// Serializes the read-modify-write cycles over the pending cross-zone
-    /// dispatch list.
+    /// dispatch records and their count cell.
     ///
-    /// The list is a single value holding the whole `Vec`, and three tasks
-    /// rewrite it: the watcher recording a delivery, the production loop
-    /// counting a failed attempt, and the publisher's drive task settling
-    /// finalized deliveries. Rocksdb makes the write atomic, not the cycle, so
-    /// without this the writer that read first silently drops the others.
+    /// Three tasks mutate them (watcher, production loop, publisher drive);
+    /// rocksdb makes each staged batch atomic, not the cycle, so without this
+    /// two interleaved writers drift the count away from the entries.
     pending_records: Mutex<()>,
 }
 
@@ -258,9 +263,8 @@ impl RocksDBIO {
     /// Held across a pending-record read-modify-write. See
     /// [`RocksDBIO::pending_records`].
     ///
-    /// A poisoned lock is recovered rather than propagated: the records behind
-    /// it are a plain `Vec` that a panicking writer cannot leave half-written,
-    /// since the write is a single rocksdb put.
+    /// Poison is recovered: every mutation is one rocksdb write, so a panic
+    /// tears nothing.
     fn lock_pending_records(&self) -> MutexGuard<'_, ()> {
         self.pending_records
             .lock()
@@ -357,6 +361,7 @@ impl RocksDBIO {
                 Some("Failed to write dump restore batch".to_owned()),
             )
         })?;
+        dbio.migrate_legacy_pending_dispatches()?;
 
         Ok(dbio)
     }
@@ -384,9 +389,77 @@ impl RocksDBIO {
             db,
             pending_records: Mutex::new(()),
         };
+        dbio.migrate_legacy_pending_dispatches()?;
         Ok(dbio)
     }
 
+    /// Rewrites a legacy whole-vector pending-dispatch blob into per-message
+    /// entries plus the count cell, then drops the blob, in one batch.
+    ///
+    /// Runs on every open, and again after a dump restore, since a restored
+    /// dump lands after the open-time pass. Without the legacy key it is a
+    /// no-op read.
+    fn migrate_legacy_pending_dispatches(&self) -> DbResult<()> {
+        let legacy = self
+            .get_opt::<LegacyPendingCrossZoneDispatchesCellOwned>(())
+            .map_err(|err| {
+                DbError::db_interaction_error(format!(
+                    "Legacy pending-dispatch blob at key {DB_META_PENDING_CROSS_ZONE_DISPATCHES_KEY:?} does not decode; delete that key to start without it: {err}"
+                ))
+            })?;
+        let Some(legacy) = legacy else {
+            return Ok(());
+        };
+        let records = legacy.0;
+
+        let mut batch = WriteBatch::default();
+        self.del_batch::<LegacyPendingCrossZoneDispatchesCellOwned>((), &mut batch)?;
+        // Folded additively: a restored blob may land on a store that already
+        // migrated and drained, so writing the blob's own length (or zero, for
+        // an empty blob) would clobber the live count and the cap would stop
+        // bounding what the store holds.
+        let mut inserted: u64 = 0;
+        if !records.is_empty() {
+            for record in &records {
+                if self
+                    .get_opt::<PendingCrossZoneDispatchCellOwned>(record.message_key)?
+                    .is_some()
+                {
+                    continue;
+                }
+                self.put_batch(
+                    &PendingCrossZoneDispatchCellRef(record),
+                    record.message_key,
+                    &mut batch,
+                )?;
+                inserted = inserted.saturating_add(1);
+            }
+            if inserted > 0 {
+                let existing = self
+                    .get_opt::<PendingCrossZoneDispatchCountCell>(())?
+                    .map_or(0, |cell| cell.0);
+                self.put_batch(
+                    &PendingCrossZoneDispatchCountCell(existing.saturating_add(inserted)),
+                    (),
+                    &mut batch,
+                )?;
+            }
+        }
+        self.db.write(batch).map_err(|rerr| {
+            DbError::rocksdb_cast_message(
+                rerr,
+                Some("Failed to migrate legacy pending cross-zone dispatches".to_owned()),
+            )
+        })?;
+
+        if inserted > 0 {
+            info!(
+                "Migrated {inserted} pending cross-zone dispatch record(s) into per-message entries"
+            );
+        }
+        Ok(())
+    }
+
     pub fn destroy(path: &Path) -> DbResult<()> {
         let mut cf_opts = Options::default();
         cf_opts.set_max_write_buffer_number(16);
@@ -703,27 +776,62 @@ impl RocksDBIO {
         self.del::<PeerFloorCellOwned>(peer_zone)
     }
 
+    /// Every pending cross-zone dispatch record, in message-key byte order:
+    /// no insertion order survives. Lock-free, so a read racing a mutation
+    /// sees either side.
     pub fn get_pending_cross_zone_dispatches(
         &self,
     ) -> DbResult<Vec<PendingCrossZoneDispatchRecord>> {
+        let prefix = Self::pending_dispatch_key_prefix()?;
+        let cf_meta = self.meta_column();
+
+        let mut records = Vec::new();
+        for item in self
+            .db
+            .iterator_cf(&cf_meta, IteratorMode::From(&prefix, Direction::Forward))
+        {
+            let (key, value) = item.map_err(|rerr| {
+                DbError::rocksdb_cast_message(
+                    rerr,
+                    Some("Failed to scan pending cross-zone dispatches".to_owned()),
+                )
+            })?;
+            // Keys sharing the prefix are one contiguous range, so the first
+            // stranger ends the scan.
+            if !key.starts_with(&prefix) {
+                break;
+            }
+            records.push(
+                borsh::from_slice::<PendingCrossZoneDispatchRecord>(&value).map_err(|err| {
+                    DbError::borsh_cast_message(
+                        err,
+                        Some("Failed to deserialize pending cross-zone dispatch".to_owned()),
+                    )
+                })?,
+            );
+        }
+        Ok(records)
+    }
+
+    /// The byte prefix every per-message pending-dispatch key starts with: a
+    /// borsh `(name, message_key)` tuple key opens with the length-prefixed
+    /// name alone, which no other meta cell's key shares (asserted in the cell
+    /// tests).
+    fn pending_dispatch_key_prefix() -> DbResult<Vec<u8>> {
+        borsh::to_vec(&DB_META_PENDING_CROSS_ZONE_DISPATCH_KEY).map_err(|err| {
+            DbError::borsh_cast_message(
+                err,
+                Some("Failed to serialize pending cross-zone dispatch key prefix".to_owned()),
+            )
+        })
+    }
+
+    /// The persisted pending-record count; see
+    /// [`PendingCrossZoneDispatchCountCell`].
+    fn get_pending_cross_zone_dispatch_count(&self) -> DbResult<u64> {
         Ok(self
-            .get_opt::<PendingCrossZoneDispatchesCellOwned>(())?
-            .map_or_else(Vec::new, |cell| cell.0))
-    }
-
-    fn put_pending_cross_zone_dispatches(
-        &self,
-        records: &[PendingCrossZoneDispatchRecord],
-    ) -> DbResult<()> {
-        self.put(&PendingCrossZoneDispatchesCellRef(records), ())
-    }
-
-    fn put_pending_cross_zone_dispatches_batch(
-        &self,
-        records: &[PendingCrossZoneDispatchRecord],
-        batch: &mut WriteBatch,
-    ) -> DbResult<()> {
-        self.put_batch(&PendingCrossZoneDispatchesCellRef(records), (), batch)
+            .get_opt::<PendingCrossZoneDispatchCountCell>(())?
+            .map_or(0, |cell| cell.0))
     }
 
     /// Records every delivery one peer block carries, in a single write.
@@ -731,14 +839,13 @@ impl RocksDBIO {
     /// Returns how many were new. Ones already recorded are skipped, so a slot
     /// the watcher re-reads is not double-tracked.
     ///
-    /// Batched rather than one call per delivery because the whole list is one
-    /// value: recording a block's messages one at a time rewrites the list once
-    /// per message, which is quadratic in a block that carries many.
+    /// All-or-nothing per peer block: recording is what lets the caller move
+    /// its delivery floor past the block, so either every delivery becomes
+    /// durable or none does and the floor holds.
     ///
-    /// Fails without writing anything if the list would exceed
-    /// [`MAX_PENDING_CROSS_ZONE_DISPATCHES`]. The caller's floor then stays put
-    /// and the slot is read again later, which is the difference between
-    /// backpressure and an unbounded list a peer controls the size of.
+    /// Fails without writing anything if the pending set would exceed
+    /// [`MAX_PENDING_CROSS_ZONE_DISPATCHES`]; see the cap for why refusal is
+    /// backpressure.
     pub fn add_pending_cross_zone_dispatches(
         &self,
         dispatches: Vec<PendingCrossZoneDispatchRecord>,
@@ -748,30 +855,52 @@ impl RocksDBIO {
         }
 
         let _pending = self.lock_pending_records();
-        let mut records = self.get_pending_cross_zone_dispatches()?;
-        let before = records.len();
 
+        // Deduped against the store by point-get, never a scan, and against the
+        // offer itself, which may repeat a key.
+        let mut offered_keys = std::collections::HashSet::<[u8; 32]>::new();
+        let mut new_records: Vec<PendingCrossZoneDispatchRecord> = Vec::new();
         for dispatch in dispatches {
-            if records
-                .iter()
-                .any(|record| record.message_key == dispatch.message_key)
+            if !offered_keys.insert(dispatch.message_key) {
+                continue;
+            }
+            if self
+                .get_opt::<PendingCrossZoneDispatchCellOwned>(dispatch.message_key)?
+                .is_some()
             {
                 continue;
             }
-            records.push(dispatch);
+            new_records.push(dispatch);
         }
 
-        let accepted = records.len().saturating_sub(before);
+        let accepted = new_records.len();
         if accepted == 0 {
             return Ok(0);
         }
-        if records.len() > MAX_PENDING_CROSS_ZONE_DISPATCHES {
+
+        let before = self.get_pending_cross_zone_dispatch_count()?;
+        let after = before.saturating_add(u64::try_from(accepted).expect("accepted fits u64"));
+        if after > u64::try_from(MAX_PENDING_CROSS_ZONE_DISPATCHES).expect("cap fits u64") {
             return Err(DbError::db_interaction_error(format!(
                 "Refusing to hold more than {MAX_PENDING_CROSS_ZONE_DISPATCHES} pending cross-zone deliveries; {before} already pending"
             )));
         }
 
-        self.put_pending_cross_zone_dispatches(&records)?;
+        let mut batch = WriteBatch::default();
+        for record in &new_records {
+            self.put_batch(
+                &PendingCrossZoneDispatchCellRef(record),
+                record.message_key,
+                &mut batch,
+            )?;
+        }
+        self.put_batch(&PendingCrossZoneDispatchCountCell(after), (), &mut batch)?;
+        self.db.write(batch).map_err(|rerr| {
+            DbError::rocksdb_cast_message(
+                rerr,
+                Some("Failed to record pending cross-zone dispatches".to_owned()),
+            )
+        })?;
         Ok(accepted)
     }
 
@@ -793,30 +922,23 @@ impl RocksDBIO {
         origin: DispatchOrigin,
     ) -> DbResult<DispatchFailure> {
         let _pending = self.lock_pending_records();
-        let mut records = self.get_pending_cross_zone_dispatches()?;
-        let Some(position) = records
-            .iter()
-            .position(|record| record.message_key == message_key)
-        else {
+        let Some(held) = self.get_opt::<PendingCrossZoneDispatchCellOwned>(message_key)? else {
             return Ok(DispatchFailure::Absent);
         };
 
-        let failed_attempts = {
-            let record = &mut records[position];
-            record.failed_attempts = record.failed_attempts.saturating_add(1);
-            record.failed_attempts
-        };
+        let mut pending = held.0;
+        pending.failed_attempts = pending.failed_attempts.saturating_add(1);
+        let failed_attempts = pending.failed_attempts;
         if failed_attempts < retire_at {
-            self.put_pending_cross_zone_dispatches(&records)?;
+            self.put(&PendingCrossZoneDispatchCellRef(&pending), message_key)?;
             return Ok(DispatchFailure::Retried { failed_attempts });
         }
 
-        let retired = records.remove(position);
         let dead_letter = DeadLetterDispatchRecord {
             message_key,
             origin,
             failed_attempts,
-            transaction_bytes: u32::try_from(retired.transaction.len()).unwrap_or(u32::MAX),
+            transaction_bytes: u32::try_from(pending.transaction.len()).unwrap_or(u32::MAX),
         };
 
         // One entry per delivery, not per retirement. A watcher rebuilding a
@@ -838,12 +960,20 @@ impl RocksDBIO {
         let count = self
             .get_dead_letter_cross_zone_dispatch_count()?
             .saturating_add(1);
+        let pending_count = self
+            .get_pending_cross_zone_dispatch_count()?
+            .saturating_sub(1);
 
         // One batch: a crash between the two halves either loses the message
         // silently or leaves the drain retrying a delivery already recorded as
         // given up on.
         let mut batch = WriteBatch::default();
-        self.put_pending_cross_zone_dispatches_batch(&records, &mut batch)?;
+        self.del_batch::<PendingCrossZoneDispatchCellOwned>(message_key, &mut batch)?;
+        self.put_batch(
+            &PendingCrossZoneDispatchCountCell(pending_count),
+            (),
+            &mut batch,
+        )?;
         self.put_batch(
             &DeadLetterCrossZoneDispatchesCellRef(&dead_letters),
             (),
@@ -888,19 +1018,8 @@ impl RocksDBIO {
         }
 
         let _pending = self.lock_pending_records();
-        let to_remove: std::collections::HashSet<&[u8; 32]> = message_keys.iter().collect();
-        let mut records = self.get_pending_cross_zone_dispatches()?;
-        let before = records.len();
-        records.retain(|record| !to_remove.contains(&record.message_key));
-        let removed = before.saturating_sub(records.len());
-
-        // Both lists in one batch, as in `record_dispatch_failure`: nothing
-        // recomputes these keys on a later pass to fix a torn write.
         let mut batch = WriteBatch::default();
-        if removed > 0 {
-            self.put_pending_cross_zone_dispatches_batch(&records, &mut batch)?;
-        }
-        self.stage_reconciled_dead_letters(&to_remove, &mut batch)?;
+        let removed = self.stage_removed_dispatches(message_keys, &mut batch)?;
         if !batch.is_empty() {
             self.db.write(batch).map_err(|rerr| {
                 DbError::rocksdb_cast_message(
@@ -942,10 +1061,11 @@ impl RocksDBIO {
 
     /// Drops the pending records of deliveries that just became irreversible,
     /// staged into `batch` so they go with the update that made them so.
+    /// Callers hold the pending-record lock; the count cell is staged here.
     ///
-    /// Removal only, unlike [`Self::stage_pending_deposit_events`]: a delivery is
-    /// recorded by the watcher through
-    /// [`Self::add_pending_cross_zone_dispatch`], on its own task and outside
+    /// Removal only, unlike [`Self::stage_pending_deposit_events`]: a delivery
+    /// is recorded by the watcher through
+    /// [`Self::add_pending_cross_zone_dispatches`], on its own task and outside
     /// any store update, so nothing ever adds one here.
     fn stage_removed_dispatches(
         &self,
@@ -956,18 +1076,34 @@ impl RocksDBIO {
             return Ok(0);
         }
 
-        let to_remove: std::collections::HashSet<&[u8; 32]> = remove_keys.iter().collect();
-        let mut records = self.get_pending_cross_zone_dispatches()?;
-        let before = records.len();
-        records.retain(|record| !to_remove.contains(&record.message_key));
-        let removed = before.saturating_sub(records.len());
+        // Point-gets before the deletes: only a key that is actually held may
+        // decrement the count, and a repeated key may do so only once.
+        let mut staged = std::collections::HashSet::<&[u8; 32]>::new();
+        for key in remove_keys {
+            if staged.contains(&key) {
+                continue;
+            }
+            if self
+                .get_opt::<PendingCrossZoneDispatchCellOwned>(*key)?
+                .is_none()
+            {
+                continue;
+            }
+            self.del_batch::<PendingCrossZoneDispatchCellOwned>(*key, batch)?;
+            staged.insert(key);
+        }
 
+        let removed = staged.len();
         if removed > 0 {
-            self.put_pending_cross_zone_dispatches_batch(&records, batch)?;
+            let count = self
+                .get_pending_cross_zone_dispatch_count()?
+                .saturating_sub(u64::try_from(removed).expect("removed fits u64"));
+            self.put_batch(&PendingCrossZoneDispatchCountCell(count), (), batch)?;
         }
 
         // The ordinary case: another sequencer carried a delivery this node gave
         // up on into a block that just became irreversible.
+        let to_remove: std::collections::HashSet<&[u8; 32]> = remove_keys.iter().collect();
         self.stage_reconciled_dead_letters(&to_remove, batch)?;
         Ok(removed)
     }

lez/storage/src/sequencer/sequencer_cells.rs

@@ -11,7 +11,8 @@ use crate::{
         DB_META_CROSS_ZONE_PEER_FLOOR_KEY, DB_META_CROSS_ZONE_PEER_TIP_KEY,
         DB_META_DEAD_LETTER_CROSS_ZONE_DISPATCH_COUNT_KEY,
         DB_META_DEAD_LETTER_CROSS_ZONE_DISPATCHES_KEY, DB_META_LAST_FINALIZED_BLOCK_ID,
-        DB_META_LATEST_BLOCK_META_KEY, DB_META_PENDING_CROSS_ZONE_DISPATCHES_KEY,
+        DB_META_LATEST_BLOCK_META_KEY, DB_META_PENDING_CROSS_ZONE_DISPATCH_COUNT_KEY,
+        DB_META_PENDING_CROSS_ZONE_DISPATCH_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,
@@ -314,41 +315,107 @@ impl PendingCrossZoneDispatchRecord {
     }
 }
 
+/// One pending delivery, held under its own message key so a mutation touches
+/// one entry rather than rewriting the whole set.
 #[derive(BorshDeserialize)]
-pub struct PendingCrossZoneDispatchesCellOwned(pub Vec<PendingCrossZoneDispatchRecord>);
+pub struct PendingCrossZoneDispatchCellOwned(pub PendingCrossZoneDispatchRecord);
 
-impl SimpleStorableCell for PendingCrossZoneDispatchesCellOwned {
-    type KeyParams = ();
+impl SimpleStorableCell for PendingCrossZoneDispatchCellOwned {
+    type KeyParams = [u8; 32];
 
-    const CELL_NAME: &'static str = DB_META_PENDING_CROSS_ZONE_DISPATCHES_KEY;
+    const CELL_NAME: &'static str = DB_META_PENDING_CROSS_ZONE_DISPATCH_KEY;
     const CF_NAME: &'static str = CF_META_NAME;
-}
 
-impl SimpleReadableCell for PendingCrossZoneDispatchesCellOwned {}
-
-#[derive(BorshSerialize)]
-pub struct PendingCrossZoneDispatchesCellRef<'records>(
-    pub &'records [PendingCrossZoneDispatchRecord],
-);
-
-impl SimpleStorableCell for PendingCrossZoneDispatchesCellRef<'_> {
-    type KeyParams = ();
-
-    const CELL_NAME: &'static str = DB_META_PENDING_CROSS_ZONE_DISPATCHES_KEY;
-    const CF_NAME: &'static str = CF_META_NAME;
-}
-
-impl SimpleWritableCell for PendingCrossZoneDispatchesCellRef<'_> {
-    fn value_constructor(&self) -> DbResult<Vec<u8>> {
-        borsh::to_vec(&self).map_err(|err| {
+    /// Folds the message key into the db key so each delivery is its own entry.
+    fn key_constructor(message_key: Self::KeyParams) -> DbResult<Vec<u8>> {
+        borsh::to_vec(&(Self::CELL_NAME, message_key)).map_err(|err| {
             DbError::borsh_cast_message(
                 err,
-                Some("Failed to serialize pending cross-zone dispatches cell".to_owned()),
+                Some(format!(
+                    "Failed to serialize {:?} key params",
+                    Self::CELL_NAME
+                )),
             )
         })
     }
 }
 
+impl SimpleReadableCell for PendingCrossZoneDispatchCellOwned {}
+
+#[derive(BorshSerialize)]
+pub struct PendingCrossZoneDispatchCellRef<'record>(pub &'record PendingCrossZoneDispatchRecord);
+
+impl SimpleStorableCell for PendingCrossZoneDispatchCellRef<'_> {
+    type KeyParams = [u8; 32];
+
+    const CELL_NAME: &'static str = DB_META_PENDING_CROSS_ZONE_DISPATCH_KEY;
+    const CF_NAME: &'static str = CF_META_NAME;
+
+    fn key_constructor(message_key: Self::KeyParams) -> DbResult<Vec<u8>> {
+        borsh::to_vec(&(Self::CELL_NAME, message_key)).map_err(|err| {
+            DbError::borsh_cast_message(
+                err,
+                Some(format!(
+                    "Failed to serialize {:?} key params",
+                    Self::CELL_NAME
+                )),
+            )
+        })
+    }
+}
+
+impl SimpleWritableCell for PendingCrossZoneDispatchCellRef<'_> {
+    fn value_constructor(&self) -> DbResult<Vec<u8>> {
+        borsh::to_vec(&self).map_err(|err| {
+            DbError::borsh_cast_message(
+                err,
+                Some("Failed to serialize pending cross-zone dispatch cell".to_owned()),
+            )
+        })
+    }
+}
+
+/// How many pending dispatch records the store holds, written in the same
+/// batch as every record mutation so the cap check reads one value instead of
+/// scanning the set it bounds.
+#[derive(BorshSerialize, BorshDeserialize)]
+pub struct PendingCrossZoneDispatchCountCell(pub u64);
+
+impl SimpleStorableCell for PendingCrossZoneDispatchCountCell {
+    type KeyParams = ();
+
+    const CELL_NAME: &'static str = DB_META_PENDING_CROSS_ZONE_DISPATCH_COUNT_KEY;
+    const CF_NAME: &'static str = CF_META_NAME;
+}
+
+impl SimpleReadableCell for PendingCrossZoneDispatchCountCell {}
+
+impl SimpleWritableCell for PendingCrossZoneDispatchCountCell {
+    fn value_constructor(&self) -> DbResult<Vec<u8>> {
+        borsh::to_vec(&self).map_err(|err| {
+            DbError::borsh_cast_message(
+                err,
+                Some("Failed to serialize pending cross-zone dispatch count".to_owned()),
+            )
+        })
+    }
+}
+
+/// The whole pending set as one borsh blob, the layout stores held before the
+/// per-message entries. Read-only: opening such a store migrates the blob and
+/// deletes its key, and nothing writes it again.
+#[derive(BorshDeserialize)]
+pub struct LegacyPendingCrossZoneDispatchesCellOwned(pub Vec<PendingCrossZoneDispatchRecord>);
+
+impl SimpleStorableCell for LegacyPendingCrossZoneDispatchesCellOwned {
+    type KeyParams = ();
+
+    const CELL_NAME: &'static str = DB_META_PENDING_CROSS_ZONE_DISPATCHES_KEY;
+    const CF_NAME: &'static str = CF_META_NAME;
+}
+
+impl SimpleReadableCell for LegacyPendingCrossZoneDispatchesCellOwned {}
+
 /// Which peer message a delivery carried, kept so a lost one can be traced back
 /// to the peer block it was in.
 #[derive(Debug, Clone, Copy, PartialEq, Eq, BorshSerialize, BorshDeserialize)]
@@ -644,6 +711,8 @@ mod uniform_tests {
         cells::SimpleStorableCell as _,
         sequencer::sequencer_cells::{
             LEEStateCellOwned, LEEStateCellRef, LatestBlockMetaCellOwned, LatestBlockMetaCellRef,
+            LegacyPendingCrossZoneDispatchesCellOwned, PendingCrossZoneDispatchCellOwned,
+            PendingCrossZoneDispatchCellRef, PendingCrossZoneDispatchCountCell,
             PendingDepositEventsCellOwned, PendingDepositEventsCellRef,
         },
     };
@@ -674,6 +743,46 @@ mod uniform_tests {
         );
     }
 
+    #[test]
+    fn pending_dispatch_ref_and_owned_is_aligned() {
+        assert_eq!(
+            PendingCrossZoneDispatchCellRef::CELL_NAME,
+            PendingCrossZoneDispatchCellOwned::CELL_NAME
+        );
+        assert_eq!(
+            PendingCrossZoneDispatchCellRef::CF_NAME,
+            PendingCrossZoneDispatchCellOwned::CF_NAME
+        );
+        assert_eq!(
+            PendingCrossZoneDispatchCellRef::key_constructor([7; 32]).unwrap(),
+            PendingCrossZoneDispatchCellOwned::key_constructor([7; 32]).unwrap()
+        );
+    }
+
+    #[test]
+    fn pending_dispatch_scan_prefix_covers_only_the_per_message_cells() {
+        // A stray meta cell keyed into this range would decode as a dispatch
+        // record and fail the lock-free scan.
+        let prefix = borsh::to_vec(&PendingCrossZoneDispatchCellOwned::CELL_NAME).unwrap();
+        assert!(
+            PendingCrossZoneDispatchCellOwned::key_constructor([0; 32])
+                .unwrap()
+                .starts_with(&prefix)
+        );
+        assert!(
+            !PendingCrossZoneDispatchCountCell::key_constructor(())
+                .unwrap()
+                .starts_with(&prefix),
+            "the count cell must stay out of the record scan"
+        );
+        assert!(
+            !LegacyPendingCrossZoneDispatchesCellOwned::key_constructor(())
+                .unwrap()
+                .starts_with(&prefix),
+            "the legacy blob must stay out of the record scan"
+        );
+    }
+
     #[test]
     fn pending_deposit_events_ref_and_owned_is_aligned() {
         assert_eq!(

lez/storage/src/sequencer/tests.rs

@@ -57,6 +57,14 @@ fn key_from_index(index: usize) -> [u8; 32] {
     key
 }
 
+/// `records` in message-key order, the order the store reports them in.
+fn sorted_dispatches(
+    mut records: Vec<PendingCrossZoneDispatchRecord>,
+) -> Vec<PendingCrossZoneDispatchRecord> {
+    records.sort_by_key(|record| record.message_key);
+    records
+}
+
 fn stored_balance(dbio: &RocksDBIO) -> u128 {
     dbio.get_lee_state()
         .unwrap()
@@ -510,10 +518,12 @@ fn dispatch_records_round_trip_and_dedupe_by_message_key() {
         "only the delivery not already held is newly recorded"
     );
 
+    // Set equality, not order: no insertion order survives the store.
     assert_eq!(
-        dbio.get_pending_cross_zone_dispatches().unwrap(),
-        vec![record, dispatch_record(2)]
+        sorted_dispatches(dbio.get_pending_cross_zone_dispatches().unwrap()),
+        sorted_dispatches(vec![record, dispatch_record(2)])
     );
+    assert_eq!(dbio.get_pending_cross_zone_dispatch_count().unwrap(), 2);
 }
 
 #[test]
@@ -544,7 +554,12 @@ fn recording_past_the_cap_writes_nothing() {
     assert_eq!(
         dbio.get_pending_cross_zone_dispatches().unwrap().len(),
         MAX_PENDING_CROSS_ZONE_DISPATCHES,
-        "a refused write must leave the list untouched"
+        "a refused write must leave the records untouched"
+    );
+    assert_eq!(
+        dbio.get_pending_cross_zone_dispatch_count().unwrap(),
+        u64::try_from(MAX_PENDING_CROSS_ZONE_DISPATCHES).unwrap(),
+        "and the count cell with them"
     );
 
     // Re-offering only what is already held is not growth, so it still succeeds.
@@ -558,6 +573,186 @@ fn recording_past_the_cap_writes_nothing() {
     );
 }
 
+#[test]
+fn dispatch_records_survive_a_reopen() {
+    let temp_dir = tempdir().unwrap();
+    let (dbio, _genesis) = dbio_with_genesis(temp_dir.path());
+
+    let first = dispatch_record(1);
+    let second = dispatch_record(2);
+    dbio.add_pending_cross_zone_dispatches(vec![first.clone(), second.clone()])
+        .unwrap();
+
+    // On disk, not in memory: the records are what stand between the watcher's
+    // durable read floor and a lost delivery across a restart.
+    drop(dbio);
+    let reopened = RocksDBIO::open(temp_dir.path()).unwrap();
+    assert_eq!(
+        sorted_dispatches(reopened.get_pending_cross_zone_dispatches().unwrap()),
+        sorted_dispatches(vec![first, second])
+    );
+    assert_eq!(reopened.get_pending_cross_zone_dispatch_count().unwrap(), 2);
+}
+
+#[test]
+fn a_legacy_dispatch_blob_is_migrated_into_per_message_entries_on_open() {
+    let temp_dir = tempdir().unwrap();
+    let (dbio, _genesis) = dbio_with_genesis(temp_dir.path());
+
+    // A store written before the per-message layout: the whole set as one borsh
+    // blob under a single fixed key.
+    let records = vec![dispatch_record(1), dispatch_record(2), dispatch_record(3)];
+    let legacy_key = borsh::to_vec(&DB_META_PENDING_CROSS_ZONE_DISPATCHES_KEY).unwrap();
+    dbio.db
+        .put_cf(
+            &dbio.meta_column(),
+            &legacy_key,
+            borsh::to_vec(&records).unwrap(),
+        )
+        .unwrap();
+    drop(dbio);
+
+    let migrated = RocksDBIO::open(temp_dir.path()).unwrap();
+    assert_eq!(
+        sorted_dispatches(migrated.get_pending_cross_zone_dispatches().unwrap()),
+        sorted_dispatches(records.clone()),
+        "every record must come through the migration unchanged"
+    );
+    for record in &records {
+        assert_eq!(
+            migrated
+                .get_opt::<PendingCrossZoneDispatchCellOwned>(record.message_key)
+                .unwrap()
+                .map(|cell| cell.0),
+            Some(record.clone()),
+            "each record must be readable under its own message key"
+        );
+    }
+    assert_eq!(migrated.get_pending_cross_zone_dispatch_count().unwrap(), 3);
+    assert!(
+        migrated
+            .db
+            .get_cf(&migrated.meta_column(), &legacy_key)
+            .unwrap()
+            .is_none(),
+        "the blob must not survive the migration"
+    );
+
+    // An empty blob is deleted without touching the migrated entries or count.
+    let empty: Vec<PendingCrossZoneDispatchRecord> = Vec::new();
+    migrated
+        .db
+        .put_cf(
+            &migrated.meta_column(),
+            &legacy_key,
+            borsh::to_vec(&empty).unwrap(),
+        )
+        .unwrap();
+    drop(migrated);
+
+    let cleaned = RocksDBIO::open(temp_dir.path()).unwrap();
+    assert!(
+        cleaned
+            .db
+            .get_cf(&cleaned.meta_column(), &legacy_key)
+            .unwrap()
+            .is_none()
+    );
+    assert_eq!(
+        cleaned.get_pending_cross_zone_dispatches().unwrap().len(),
+        3
+    );
+    assert_eq!(cleaned.get_pending_cross_zone_dispatch_count().unwrap(), 3);
+}
+
+/// A blob restored over live per-message entries folds additively into the
+/// count.
+#[test]
+fn a_legacy_blob_over_live_entries_folds_additively() {
+    let temp_dir = tempdir().unwrap();
+    let (dbio, _genesis) = dbio_with_genesis(temp_dir.path());
+
+    dbio.add_pending_cross_zone_dispatches(vec![dispatch_record(1), dispatch_record(2)])
+        .unwrap();
+
+    // The blob shares record 2 with the live entries and brings record 3.
+    let blob = vec![dispatch_record(2), dispatch_record(3)];
+    let legacy_key = borsh::to_vec(&DB_META_PENDING_CROSS_ZONE_DISPATCHES_KEY).unwrap();
+    dbio.db
+        .put_cf(
+            &dbio.meta_column(),
+            &legacy_key,
+            borsh::to_vec(&blob).unwrap(),
+        )
+        .unwrap();
+    drop(dbio);
+
+    let merged = RocksDBIO::open(temp_dir.path()).unwrap();
+    assert_eq!(
+        sorted_dispatches(merged.get_pending_cross_zone_dispatches().unwrap()),
+        sorted_dispatches(vec![
+            dispatch_record(1),
+            dispatch_record(2),
+            dispatch_record(3)
+        ]),
+        "the migration must keep the union of blob and live entries"
+    );
+    assert_eq!(
+        merged.get_pending_cross_zone_dispatch_count().unwrap(),
+        3,
+        "the count must be the union's size, not the blob's length"
+    );
+}
+
+#[test]
+fn the_dispatch_count_cell_tracks_the_stored_entries() {
+    let temp_dir = tempdir().unwrap();
+    let (dbio, _genesis) = dbio_with_genesis(temp_dir.path());
+
+    let count_and_entries_agree = |expected: u64| {
+        assert_eq!(
+            dbio.get_pending_cross_zone_dispatch_count().unwrap(),
+            expected
+        );
+        assert_eq!(
+            u64::try_from(dbio.get_pending_cross_zone_dispatches().unwrap().len()).unwrap(),
+            expected,
+            "the count cell and the scanned entries must never disagree"
+        );
+    };
+
+    dbio.add_pending_cross_zone_dispatches(vec![
+        dispatch_record(1),
+        dispatch_record(2),
+        dispatch_record(3),
+    ])
+    .unwrap();
+    count_and_entries_agree(3);
+
+    // A counted retry keeps the record, so the count stands still.
+    dbio.record_dispatch_failure([1; 32], 2, dispatch_origin(1))
+        .unwrap();
+    count_and_entries_agree(3);
+
+    // A retirement into the dead letter takes its record out.
+    dbio.record_dispatch_failure([1; 32], 2, dispatch_origin(1))
+        .unwrap();
+    count_and_entries_agree(2);
+
+    // As does a standalone settled drop, even repeated on a key already gone.
+    dbio.drop_settled_cross_zone_dispatches(&[[2; 32], [2; 32]])
+        .unwrap();
+    count_and_entries_agree(1);
+
+    // And the settlement path inside a store update.
+    dbio.store_update(&StoreUpdate {
+        remove_dispatch_records: &[[3; 32]],
+        ..StoreUpdate::new(&state_with_balance(100))
+    })
+    .unwrap();
+    count_and_entries_agree(0);
+}
+
 #[test]
 fn settled_dispatch_records_are_dropped_outside_an_update() {
     let temp_dir = tempdir().unwrap();