feat(storage): record cross-zone deliveries and settle them on finality

lez/storage/src/sequencer/mod.rs

@@ -1,4 +1,8 @@
-use std::{collections::BTreeMap, path::Path, sync::Arc};
+use std::{
+    collections::BTreeMap,
+    path::Path,
+    sync::{Arc, Mutex, MutexGuard, PoisonError},
+};
 
 use borsh::{BorshDeserialize, BorshSerialize};
 use common::{
@@ -18,7 +22,9 @@ use crate::{
     sequencer::sequencer_cells::{
         FinalBlockMetaCellOwned, FinalBlockMetaCellRef, FinalLeeStateCellOwned,
         FinalLeeStateCellRef, LEEStateCellOwned, LEEStateCellRef, LastFinalizedBlockIdCell,
-        LatestBlockMetaCellOwned, LatestBlockMetaCellRef, PendingDepositEventRecord,
+        LatestBlockMetaCellOwned, LatestBlockMetaCellRef, PeerFloorCellOwned, PeerFloorCellRef,
+        PeerZoneKey, PendingCrossZoneDispatchRecord, PendingCrossZoneDispatchesCellOwned,
+        PendingCrossZoneDispatchesCellRef, PendingDepositEventRecord,
         PendingDepositEventsCellOwned, PendingDepositEventsCellRef, UnseenWithdrawCountCell,
         WithdrawalReconciliationKey, ZoneAnchorCell, ZoneAnchorRecord, ZoneSdkCheckpointCellOwned,
         ZoneSdkCheckpointCellRef,
@@ -40,9 +46,25 @@ pub const DB_META_ZONE_CURSOR_KEY: &str = "zone_cursor";
 /// Key base for storing queued deposit events that were not yet
 /// fulfilled on L2.
 pub const DB_META_PENDING_DEPOSIT_EVENTS_KEY: &str = "pending_deposit_events";
+/// Key base for storing a cross-zone watcher's delivery floor on one peer
+/// channel (opaque bytes). Keyed per peer zone.
+pub const DB_META_CROSS_ZONE_PEER_FLOOR_KEY: &str = "cross_zone_peer_floor";
+/// Key base for storing cross-zone deliveries the watcher has recorded but
+/// which are not yet known to be irreversibly delivered.
+pub const DB_META_PENDING_CROSS_ZONE_DISPATCHES_KEY: &str = "pending_cross_zone_dispatches";
+
 /// Key base for counting unseen L2 withdraw intents.
 pub const DB_META_UNSEEN_WITHDRAW_COUNT_KEY: &str = "unseen_withdraw_count";
 
+/// 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.
+pub const MAX_PENDING_CROSS_ZONE_DISPATCHES: usize = 4096;
+
 /// Key base for storing the LEE state.
 pub const DB_LEE_STATE_KEY: &str = "lee_state";
 /// Key base for storing the LEE state at the last L1-finalized block.
@@ -123,6 +145,8 @@ pub struct StoreUpdate<'update> {
     pub new_deposit_events: &'update [PendingDepositEventRecord],
     /// Deposit op ids whose mint finalized: their pending records are dropped.
     pub remove_deposit_records: &'update [HashType],
+    /// Message keys whose delivery finalized: their pending records are dropped.
+    pub remove_dispatch_records: &'update [[u8; 32]],
     /// L1 withdraw events to reconcile against the local unseen counters.
     pub consumed_withdrawals: &'update [WithdrawalReconciliationKey],
     /// L2 withdraw intents this update raises, awaiting their L1 event.
@@ -146,6 +170,7 @@ impl<'update> StoreUpdate<'update> {
             finalized_up_to: None,
             new_deposit_events: &[],
             remove_deposit_records: &[],
+            remove_dispatch_records: &[],
             consumed_withdrawals: &[],
             new_withdraw_intents: &[],
             zone_anchor: None,
@@ -165,8 +190,21 @@ pub struct StoreUpdateOutcome {
     pub unmatched_withdrawals: Vec<WithdrawalReconciliationKey>,
 }
 
+#[expect(
+    clippy::partial_pub_fields,
+    reason = "the pending-record lock is an implementation detail and must stay private"
+)]
 pub struct RocksDBIO {
     pub db: DBWithThreadMode<MultiThreaded>,
+    /// Serializes the read-modify-write cycles over the pending cross-zone
+    /// dispatch list.
+    ///
+    /// 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.
+    pending_records: Mutex<()>,
 }
 
 impl DBIO for RocksDBIO {
@@ -176,6 +214,18 @@ impl DBIO for RocksDBIO {
 }
 
 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.
+    fn lock_pending_records(&self) -> MutexGuard<'_, ()> {
+        self.pending_records
+            .lock()
+            .unwrap_or_else(PoisonError::into_inner)
+    }
+
     pub fn open(path: &Path) -> DbResult<Self> {
         let db_opts = Options::default();
         Self::open_inner(path, &db_opts)
@@ -289,7 +339,10 @@ impl RocksDBIO {
             additional_info: Some("Failed to open or create DB".to_owned()),
         })?;
 
-        let dbio = Self { db };
+        let dbio = Self {
+            db,
+            pending_records: Mutex::new(()),
+        };
         Ok(dbio)
     }
 
@@ -539,6 +592,185 @@ impl RocksDBIO {
         Ok(accepted)
     }
 
+    /// One cross-zone watcher's delivery floor on `peer_zone`'s channel, or
+    /// `None` before it has delivered anything from that peer.
+    pub fn get_cross_zone_peer_floor_bytes(
+        &self,
+        peer_zone: PeerZoneKey,
+    ) -> DbResult<Option<Vec<u8>>> {
+        Ok(self
+            .get_opt::<PeerFloorCellOwned>(peer_zone)?
+            .map(|cell| cell.0))
+    }
+
+    pub fn put_cross_zone_peer_floor_bytes(
+        &self,
+        peer_zone: PeerZoneKey,
+        bytes: &[u8],
+    ) -> DbResult<()> {
+        self.put(&PeerFloorCellRef(bytes), peer_zone)
+    }
+
+    pub fn get_pending_cross_zone_dispatches(
+        &self,
+    ) -> DbResult<Vec<PendingCrossZoneDispatchRecord>> {
+        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)
+    }
+
+    /// Records every delivery one peer block carries, in a single write.
+    ///
+    /// 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.
+    ///
+    /// 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.
+    pub fn add_pending_cross_zone_dispatches(
+        &self,
+        dispatches: Vec<PendingCrossZoneDispatchRecord>,
+    ) -> DbResult<usize> {
+        if dispatches.is_empty() {
+            return Ok(0);
+        }
+
+        let _pending = self.lock_pending_records();
+        let mut records = self.get_pending_cross_zone_dispatches()?;
+        let before = records.len();
+
+        for dispatch in dispatches {
+            if records
+                .iter()
+                .any(|record| record.message_key == dispatch.message_key)
+            {
+                continue;
+            }
+            records.push(dispatch);
+        }
+
+        let accepted = records.len().saturating_sub(before);
+        if accepted == 0 {
+            return Ok(0);
+        }
+        if records.len() > MAX_PENDING_CROSS_ZONE_DISPATCHES {
+            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)?;
+        Ok(accepted)
+    }
+
+    /// Counts a failed production attempt against a delivery, dropping its
+    /// record once it reaches `retire_at`. Returns whether it was dropped.
+    ///
+    /// Dropped rather than flagged: a retired record is one the drain will never
+    /// turn into a block transaction again, so nothing would ever remove it, and
+    /// a peer that can make deliveries fail could grow the list without bound.
+    /// The delivery is given up on either way; this way the cost is a log line
+    /// rather than a permanent entry.
+    ///
+    /// A delivery with no record is already retired as far as this is concerned:
+    /// there is nothing left to count against.
+    pub fn record_dispatch_failure(&self, message_key: [u8; 32], retire_at: u32) -> DbResult<bool> {
+        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 {
+            return Ok(true);
+        };
+
+        let attempts = {
+            let record = &mut records[position];
+            record.failed_attempts = record.failed_attempts.saturating_add(1);
+            record.failed_attempts
+        };
+        let retired = attempts >= retire_at;
+        if retired {
+            records.remove(position);
+        }
+        self.put_pending_cross_zone_dispatches(&records)?;
+        Ok(retired)
+    }
+
+    /// Drops the records of deliveries that are settled for good, outside any
+    /// store update.
+    ///
+    /// The settlement path in [`Self::store_update`] catches a delivery as its
+    /// block becomes irreversible. This catches the ones that path cannot: a
+    /// record re-added after its delivery had already settled, which the watcher
+    /// does whenever it re-reads a slot it has already consumed. Nothing would
+    /// ever put such a key in a block again, so without this it stays for ever.
+    pub fn drop_settled_cross_zone_dispatches(&self, message_keys: &[[u8; 32]]) -> DbResult<usize> {
+        if message_keys.is_empty() {
+            return Ok(0);
+        }
+
+        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());
+
+        if removed > 0 {
+            self.put_pending_cross_zone_dispatches(&records)?;
+        }
+        Ok(removed)
+    }
+
+    /// Drops the pending records of deliveries that just became irreversible,
+    /// staged into `batch` so they go with the update that made them so.
+    ///
+    /// 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
+    /// any store update, so nothing ever adds one here.
+    fn stage_removed_dispatches(
+        &self,
+        remove_keys: &[[u8; 32]],
+        batch: &mut WriteBatch,
+    ) -> DbResult<usize> {
+        if remove_keys.is_empty() {
+            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());
+
+        if removed > 0 {
+            self.put_pending_cross_zone_dispatches_batch(&records, batch)?;
+        }
+        Ok(removed)
+    }
+
     /// Stages the unseen-withdraw decrements for one update into `batch`,
     /// returning one entry per occurrence that matched no local counter.
     ///
@@ -875,6 +1107,7 @@ impl RocksDBIO {
     /// one, most carry nothing else) must not drag a full state serialization
     /// with it.
     pub fn store_update(&self, update: &StoreUpdate<'_>) -> DbResult<StoreUpdateOutcome> {
+        let _pending = self.lock_pending_records();
         let StoreUpdate {
             checkpoint,
             blocks,
@@ -884,6 +1117,7 @@ impl RocksDBIO {
             finalized_up_to,
             new_deposit_events,
             remove_deposit_records,
+            remove_dispatch_records,
             consumed_withdrawals,
             new_withdraw_intents,
             zone_anchor,
@@ -940,6 +1174,7 @@ impl RocksDBIO {
             remove_deposit_records,
             &mut batch,
         )?;
+        self.stage_removed_dispatches(remove_dispatch_records, &mut batch)?;
         let unmatched_withdrawals =
             self.stage_consumed_withdrawals(consumed_withdrawals, &mut batch)?;
         self.stage_new_withdraw_intents(new_withdraw_intents, &mut batch)?;

lez/storage/src/sequencer/sequencer_cells.rs

@@ -8,7 +8,8 @@ use crate::{
     error::DbError,
     sequencer::{
         CF_LEE_STATE_NAME, DB_FINAL_BLOCK_META_KEY, DB_FINAL_LEE_STATE_KEY, DB_LEE_STATE_KEY,
-        DB_META_LAST_FINALIZED_BLOCK_ID, DB_META_LATEST_BLOCK_META_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,
     },
@@ -245,6 +246,82 @@ pub struct PendingDepositEventRecord {
     pub metadata: Vec<u8>,
 }
 
+/// A cross-zone delivery the watcher has read off a peer block but which is not
+/// yet known to be irreversibly delivered.
+///
+/// The watcher's delivery floor is durable, so once it advances past a peer
+/// block that block is never re-read. This record is what stands in its place:
+/// block production drains it every turn, and it survives a restart. Mirrors
+/// [`PendingDepositEventRecord`], which solves the same problem for deposits,
+/// and like it carries no "submitted" mark: the record is dropped when the
+/// delivery itself finalizes, and re-including one meanwhile is harmless
+/// because the inbox no-ops a replay on chain.
+#[derive(Debug, Clone, PartialEq, Eq, BorshSerialize, BorshDeserialize)]
+pub struct PendingCrossZoneDispatchRecord {
+    /// Content-addressed replay key of the delivered message, and this record's
+    /// identity.
+    pub message_key: [u8; 32],
+    /// The borsh-encoded dispatch transaction, so production can re-feed it
+    /// without re-reading the peer channel.
+    pub transaction: Vec<u8>,
+    /// Production attempts that ended in an execution failure.
+    ///
+    /// A dispatch's payload and target accounts are chosen on the peer zone and
+    /// validated by nobody in between, so one can fail for good. A failure can
+    /// equally be a property of the moment, so a single one is not enough to
+    /// give up on a delivery. Once too many accumulate the record is dropped
+    /// rather than flagged, since a delivery nothing will retry is also a
+    /// delivery nothing would ever remove.
+    pub failed_attempts: u32,
+}
+
+impl PendingCrossZoneDispatchRecord {
+    /// A delivery the watcher has just read: never attempted.
+    #[must_use]
+    pub const fn recorded(message_key: [u8; 32], transaction: Vec<u8>) -> Self {
+        Self {
+            message_key,
+            transaction,
+            failed_attempts: 0,
+        }
+    }
+}
+
+#[derive(BorshDeserialize)]
+pub struct PendingCrossZoneDispatchesCellOwned(pub Vec<PendingCrossZoneDispatchRecord>);
+
+impl SimpleStorableCell for PendingCrossZoneDispatchesCellOwned {
+    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 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| {
+            DbError::borsh_cast_message(
+                err,
+                Some("Failed to serialize pending cross-zone dispatches cell".to_owned()),
+            )
+        })
+    }
+}
+
 #[derive(BorshDeserialize)]
 pub struct PendingDepositEventsCellOwned(pub Vec<PendingDepositEventRecord>);
 
@@ -278,6 +355,72 @@ impl SimpleWritableCell for PendingDepositEventsCellRef<'_> {
     }
 }
 
+/// Identifies which peer channel a cross-zone watcher cursor belongs to. The
+/// 32-byte peer channel id doubles as the peer's zone id.
+pub type PeerZoneKey = [u8; 32];
+
+/// Opaque bytes for one peer's cross-zone read cursor. As with the zone-sdk
+/// checkpoint, the caller owns the encoding, since the cursor type derives serde
+/// rather than borsh.
+#[derive(BorshDeserialize)]
+pub struct PeerFloorCellOwned(pub Vec<u8>);
+
+impl SimpleStorableCell for PeerFloorCellOwned {
+    type KeyParams = PeerZoneKey;
+
+    const CELL_NAME: &'static str = DB_META_CROSS_ZONE_PEER_FLOOR_KEY;
+    const CF_NAME: &'static str = CF_META_NAME;
+
+    /// Folds the peer zone into the key so each peer keeps its own cursor.
+    fn key_constructor(peer_zone: Self::KeyParams) -> DbResult<Vec<u8>> {
+        borsh::to_vec(&(Self::CELL_NAME, peer_zone)).map_err(|err| {
+            DbError::borsh_cast_message(
+                err,
+                Some(format!(
+                    "Failed to serialize {:?} key params",
+                    Self::CELL_NAME
+                )),
+            )
+        })
+    }
+}
+
+impl SimpleReadableCell for PeerFloorCellOwned {}
+
+#[derive(BorshSerialize)]
+pub struct PeerFloorCellRef<'bytes>(pub &'bytes [u8]);
+
+impl SimpleStorableCell for PeerFloorCellRef<'_> {
+    type KeyParams = PeerZoneKey;
+
+    const CELL_NAME: &'static str = DB_META_CROSS_ZONE_PEER_FLOOR_KEY;
+    const CF_NAME: &'static str = CF_META_NAME;
+
+    /// Folds the peer zone into the key so each peer keeps its own cursor.
+    fn key_constructor(peer_zone: Self::KeyParams) -> DbResult<Vec<u8>> {
+        borsh::to_vec(&(Self::CELL_NAME, peer_zone)).map_err(|err| {
+            DbError::borsh_cast_message(
+                err,
+                Some(format!(
+                    "Failed to serialize {:?} key params",
+                    Self::CELL_NAME
+                )),
+            )
+        })
+    }
+}
+
+impl SimpleWritableCell for PeerFloorCellRef<'_> {
+    fn value_constructor(&self) -> DbResult<Vec<u8>> {
+        borsh::to_vec(&self).map_err(|err| {
+            DbError::borsh_cast_message(
+                err,
+                Some("Failed to serialize cross-zone peer floor cell".to_owned()),
+            )
+        })
+    }
+}
+
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
 pub struct WithdrawalReconciliationKey {
     pub amount: u64,

lez/storage/src/sequencer/tests.rs

@@ -37,6 +37,17 @@ fn deposit_record(seed: u8) -> PendingDepositEventRecord {
     }
 }
 
+fn dispatch_record(seed: u8) -> PendingCrossZoneDispatchRecord {
+    PendingCrossZoneDispatchRecord::recorded([seed; 32], vec![seed; 4])
+}
+
+/// A distinct message key per index, for filling the pending list.
+fn key_from_index(index: usize) -> [u8; 32] {
+    let mut key = [0_u8; 32];
+    key[..8].copy_from_slice(&u64::try_from(index).expect("test index fits").to_le_bytes());
+    key
+}
+
 fn stored_balance(dbio: &RocksDBIO) -> u128 {
     dbio.get_lee_state()
         .unwrap()
@@ -399,6 +410,171 @@ fn finalized_deposit_records_are_removed_by_op_id() {
     assert_eq!(stored, vec![second]);
 }
 
+#[test]
+fn dispatch_records_round_trip_and_dedupe_by_message_key() {
+    let temp_dir = tempdir().unwrap();
+    let (dbio, _genesis) = dbio_with_genesis(temp_dir.path());
+
+    let record = dispatch_record(1);
+    assert_eq!(
+        dbio.add_pending_cross_zone_dispatches(vec![record.clone()])
+            .unwrap(),
+        1
+    );
+    // The watcher re-reads a slot it stalled on, so the same delivery arrives
+    // again; recording it twice would double-count its failed attempts.
+    assert_eq!(
+        dbio.add_pending_cross_zone_dispatches(vec![record.clone(), dispatch_record(2)])
+            .unwrap(),
+        1,
+        "only the delivery not already held is newly recorded"
+    );
+
+    assert_eq!(
+        dbio.get_pending_cross_zone_dispatches().unwrap(),
+        vec![record, dispatch_record(2)]
+    );
+}
+
+#[test]
+fn recording_past_the_cap_writes_nothing() {
+    let temp_dir = tempdir().unwrap();
+    let (dbio, _genesis) = dbio_with_genesis(temp_dir.path());
+
+    // What fills this list is chosen by peer zones, so the bound is what stops a
+    // peer deciding how large our store gets. Refusing the whole write leaves
+    // the watcher's floor where it is, so the slot is read again later and
+    // nothing is lost.
+    let full: Vec<_> = (0..MAX_PENDING_CROSS_ZONE_DISPATCHES)
+        .map(|seed| PendingCrossZoneDispatchRecord::recorded(key_from_index(seed), vec![0_u8; 4]))
+        .collect();
+    assert_eq!(
+        dbio.add_pending_cross_zone_dispatches(full).unwrap(),
+        MAX_PENDING_CROSS_ZONE_DISPATCHES
+    );
+
+    let over = PendingCrossZoneDispatchRecord::recorded(
+        key_from_index(MAX_PENDING_CROSS_ZONE_DISPATCHES),
+        vec![0_u8; 4],
+    );
+    assert!(
+        dbio.add_pending_cross_zone_dispatches(vec![over]).is_err(),
+        "recording past the cap must fail so the caller holds its floor"
+    );
+    assert_eq!(
+        dbio.get_pending_cross_zone_dispatches().unwrap().len(),
+        MAX_PENDING_CROSS_ZONE_DISPATCHES,
+        "a refused write must leave the list untouched"
+    );
+
+    // Re-offering only what is already held is not growth, so it still succeeds.
+    assert_eq!(
+        dbio.add_pending_cross_zone_dispatches(vec![PendingCrossZoneDispatchRecord::recorded(
+            key_from_index(0),
+            vec![0_u8; 4]
+        )])
+        .unwrap(),
+        0
+    );
+}
+
+#[test]
+fn settled_dispatch_records_are_dropped_outside_an_update() {
+    let temp_dir = tempdir().unwrap();
+    let (dbio, _genesis) = dbio_with_genesis(temp_dir.path());
+
+    // The watcher re-reads a slot it already consumed and re-records a delivery
+    // that settled long ago. Its key will never appear in a future block, so the
+    // store-update path cannot reach it and this is the only thing that does.
+    let first = dispatch_record(1);
+    let second = dispatch_record(2);
+    dbio.add_pending_cross_zone_dispatches(vec![first.clone(), second.clone()])
+        .unwrap();
+
+    assert_eq!(
+        dbio.drop_settled_cross_zone_dispatches(&[first.message_key])
+            .unwrap(),
+        1
+    );
+    assert_eq!(
+        dbio.get_pending_cross_zone_dispatches().unwrap(),
+        vec![second]
+    );
+
+    // Dropping one that is already gone is a no-op, not an error.
+    assert_eq!(
+        dbio.drop_settled_cross_zone_dispatches(&[first.message_key])
+            .unwrap(),
+        0
+    );
+}
+
+#[test]
+fn finalized_dispatch_records_are_removed_by_message_key() {
+    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();
+
+    // Only the finalized delivery's key is dropped. Two deliveries can sit in
+    // the same block, so a record must go by its own identity rather than by
+    // anything about the height its delivery landed at.
+    dbio.store_update(&StoreUpdate {
+        remove_dispatch_records: &[first.message_key],
+        ..StoreUpdate::new(&state_with_balance(100))
+    })
+    .unwrap();
+
+    assert_eq!(
+        dbio.get_pending_cross_zone_dispatches().unwrap(),
+        vec![second]
+    );
+}
+
+#[test]
+fn record_dispatch_failure_drops_the_record_at_the_limit() {
+    let temp_dir = tempdir().unwrap();
+    let (dbio, _genesis) = dbio_with_genesis(temp_dir.path());
+
+    let record = dispatch_record(1);
+    let key = record.message_key;
+    let survivor = dispatch_record(2);
+    dbio.add_pending_cross_zone_dispatches(vec![record, survivor.clone()])
+        .unwrap();
+
+    assert!(!dbio.record_dispatch_failure(key, 3).unwrap());
+    assert_eq!(
+        dbio.get_pending_cross_zone_dispatches().unwrap()[0].failed_attempts,
+        1,
+        "a failure short of the limit is counted, not given up on"
+    );
+    assert!(!dbio.record_dispatch_failure(key, 3).unwrap());
+    assert!(
+        dbio.record_dispatch_failure(key, 3).unwrap(),
+        "the third failure is the one it is given up on"
+    );
+
+    // Dropped rather than flagged: a delivery the drain will never feed into a
+    // block again is one nothing would ever remove, so flagging it would let a
+    // peer that can make deliveries fail grow the list without bound.
+    assert_eq!(
+        dbio.get_pending_cross_zone_dispatches().unwrap(),
+        vec![survivor],
+        "giving up on a delivery drops its record and leaves the others alone"
+    );
+
+    // A key with no record reads as given up on: there is nothing left to count
+    // against, and nothing will feed it into a block.
+    assert!(
+        dbio.record_dispatch_failure(key, 3).unwrap(),
+        "a failure against a dropped delivery must not re-create its record"
+    );
+    assert_eq!(dbio.get_pending_cross_zone_dispatches().unwrap().len(), 1);
+}
+
 #[test]
 fn repeated_withdrawal_key_in_one_update_folds_once_per_occurrence() {
     let temp_dir = tempdir().unwrap();