From 2d201f7bacbdd3742251832ce34e37c0bb65424e Mon Sep 17 00:00:00 2001 From: moudyellaz Date: Fri, 31 Jul 2026 10:45:01 +0200 Subject: [PATCH] feat(storage): record cross-zone deliveries and settle them on finality --- lez/storage/src/sequencer/mod.rs | 241 ++++++++++++++++++- lez/storage/src/sequencer/sequencer_cells.rs | 145 ++++++++++- lez/storage/src/sequencer/tests.rs | 176 ++++++++++++++ 3 files changed, 558 insertions(+), 4 deletions(-) diff --git a/lez/storage/src/sequencer/mod.rs b/lez/storage/src/sequencer/mod.rs index 28a9f65e..a559085a 100644 --- a/lez/storage/src/sequencer/mod.rs +++ b/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, } +#[expect( + clippy::partial_pub_fields, + reason = "the pending-record lock is an implementation detail and must stay private" +)] pub struct RocksDBIO { pub db: DBWithThreadMode, + /// 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 { 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>> { + Ok(self + .get_opt::(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> { + Ok(self + .get_opt::(())? + .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, + ) -> DbResult { + 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 { + 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 { + 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 { + 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 { + 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)?; diff --git a/lez/storage/src/sequencer/sequencer_cells.rs b/lez/storage/src/sequencer/sequencer_cells.rs index 8a4223c0..9f3ae9eb 100644 --- a/lez/storage/src/sequencer/sequencer_cells.rs +++ b/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, } +/// 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, + /// 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) -> Self { + Self { + message_key, + transaction, + failed_attempts: 0, + } + } +} + +#[derive(BorshDeserialize)] +pub struct PendingCrossZoneDispatchesCellOwned(pub Vec); + +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> { + 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); @@ -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); + +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> { + 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> { + 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> { + 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, diff --git a/lez/storage/src/sequencer/tests.rs b/lez/storage/src/sequencer/tests.rs index b50a5591..8f0f1ee6 100644 --- a/lez/storage/src/sequencer/tests.rs +++ b/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();