diff --git a/lez/sequencer/core/src/cross_zone_watcher.rs b/lez/sequencer/core/src/cross_zone_watcher.rs index 038b7e891..a2ef0c689 100644 --- a/lez/sequencer/core/src/cross_zone_watcher.rs +++ b/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))); } diff --git a/lez/storage/src/sequencer/mod.rs b/lez/storage/src/sequencer/mod.rs index 90f074145..c8c2b4626 100644 --- a/lez/storage/src/sequencer/mod.rs +++ b/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, /// 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::(()) + .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::((), &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::(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::(())? + .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::(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> { + 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::(&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> { + 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 { 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) + .get_opt::(())? + .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, @@ -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 = 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::(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 { 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::(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::(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::(*key)? + .is_none() + { + continue; + } + self.del_batch::(*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) } diff --git a/lez/storage/src/sequencer/sequencer_cells.rs b/lez/storage/src/sequencer/sequencer_cells.rs index 8cbfe012d..fb723fccb 100644 --- a/lez/storage/src/sequencer/sequencer_cells.rs +++ b/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); +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> { - 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> { + 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> { + 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> { + 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> { + 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); + +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!( diff --git a/lez/storage/src/sequencer/tests.rs b/lez/storage/src/sequencer/tests.rs index 2d4d2ed34..392ffff93 100644 --- a/lez/storage/src/sequencer/tests.rs +++ b/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, +) -> Vec { + 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::(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 = 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();