scope collection ownership and replay by database

Author: farhan-syah

nodedb-wal/src/lib.rs

@@ -58,7 +58,7 @@ pub use record::{
     SpatialDeletePayload, SpatialPutPayload, WalRecord, WalRecordArgs,
 };
 pub use recovery::{RecoveryInfo, recover};
-pub use replay::{TombstoneSet, extract_tombstones};
+pub use replay::{DatabaseTombstones, TombstoneSet, extract_tombstones};
 pub use secure_mem::SecureKey;
 pub use segmented::{SegmentedWal, SegmentedWalConfig};
 pub use temporal_purge::{TemporalPurgeEngine, TemporalPurgePayload};

nodedb-wal/src/replay/filter.rs

@@ -2,8 +2,8 @@
 
 //! Collection-tombstone replay filter.
 //!
-//! A [`TombstoneSet`] records, per `(tenant_id, collection)` pair, the
-//! highest `purge_lsn` observed in any `RecordType::CollectionTombstoned`
+//! A [`TombstoneSet`] records, per `(database_id, tenant_id, collection)` tuple,
+//! the highest `purge_lsn` observed in any `RecordType::CollectionTombstoned`
 //! record. Replay consumers query [`TombstoneSet::is_tombstoned`] after
 //! decoding the collection field from their payload; any record with
 //! `lsn < purge_lsn` for a tombstoned pair MUST be skipped.
@@ -20,24 +20,33 @@ use crate::tombstone::CollectionTombstonePayload;
 
 /// In-memory index of active collection tombstones.
 ///
-/// Keyed by `(tenant_id, collection)`; value is the `purge_lsn` written
-/// at tombstone time. If the same pair is tombstoned more than once
+/// Keyed by `(database_id, tenant_id, collection)`; value is the `purge_lsn`
+/// written at tombstone time. If the same object is tombstoned more than once
 /// (re-create → re-drop in the same log), the highest `purge_lsn` wins.
 #[derive(Debug, Default, Clone)]
 pub struct TombstoneSet {
-    entries: HashMap<(u64, String), u64>,
+    entries: HashMap<(u64, u64, String), u64>,
 }
 
 impl TombstoneSet {
     pub fn new() -> Self {
         Self::default()
     }
 
-    /// Record a tombstone. If the pair already has a higher `purge_lsn`,
+    /// Bind replay checks to one database while preserving the concise
+    /// `(tenant, collection, lsn)` predicate used inside engine replay loops.
+    pub fn for_database(&self, database_id: u64) -> DatabaseTombstones<'_> {
+        DatabaseTombstones {
+            set: self,
+            database_id,
+        }
+    }
+
+    /// Record a tombstone. If the object already has a higher `purge_lsn`,
     /// the existing value is kept (idempotent, order-independent).
-    pub fn insert(&mut self, tenant_id: u64, collection: String, purge_lsn: u64) {
+    pub fn insert(&mut self, database_id: u64, tenant_id: u64, collection: String, purge_lsn: u64) {
         self.entries
-            .entry((tenant_id, collection))
+            .entry((database_id, tenant_id, collection))
             .and_modify(|existing| {
                 if purge_lsn > *existing {
                     *existing = purge_lsn;
@@ -46,20 +55,25 @@ impl TombstoneSet {
             .or_insert(purge_lsn);
     }
 
-    /// Return `true` iff a write at `lsn` for `(tenant_id, collection)`
-    /// is shadowed by a later tombstone and therefore must be skipped
-    /// during replay.
-    pub fn is_tombstoned(&self, tenant_id: u64, collection: &str, lsn: u64) -> bool {
+    /// Return `true` iff a write at `lsn` for the database-scoped object
+    /// is shadowed by a later tombstone and therefore must be skipped.
+    pub fn is_tombstoned(
+        &self,
+        database_id: u64,
+        tenant_id: u64,
+        collection: &str,
+        lsn: u64,
+    ) -> bool {
         self.entries
-            .get(&(tenant_id, collection.to_string()))
+            .get(&(database_id, tenant_id, collection.to_string()))
             .is_some_and(|&purge_lsn| lsn < purge_lsn)
     }
 
-    /// Return the `purge_lsn` for a pair, if any. Primarily used by redb
+    /// Return the `purge_lsn` for an object, if any. Primarily used by redb
     /// persistence to serialize the current set after a replay pass.
-    pub fn purge_lsn(&self, tenant_id: u64, collection: &str) -> Option<u64> {
+    pub fn purge_lsn(&self, database_id: u64, tenant_id: u64, collection: &str) -> Option<u64> {
         self.entries
-            .get(&(tenant_id, collection.to_string()))
+            .get(&(database_id, tenant_id, collection.to_string()))
             .copied()
     }
 
@@ -71,22 +85,43 @@ impl TombstoneSet {
         self.entries.is_empty()
     }
 
-    /// Iterate over every `(tenant_id, collection, purge_lsn)` in the set.
-    pub fn iter(&self) -> impl Iterator<Item = (u64, &str, u64)> + '_ {
+    /// Iterate over every `(database_id, tenant_id, collection, purge_lsn)`.
+    pub fn iter(&self) -> impl Iterator<Item = (u64, u64, &str, u64)> + '_ {
         self.entries
             .iter()
-            .map(|((tid, name), lsn)| (*tid, name.as_str(), *lsn))
+            .map(|((db, tid, name), lsn)| (*db, *tid, name.as_str(), *lsn))
     }
 
     /// Merge another tombstone set into this one. Used when loading
     /// persisted tombstones from redb at startup before a fresh WAL pass.
     pub fn extend(&mut self, other: TombstoneSet) {
-        for ((tid, name), lsn) in other.entries {
-            self.insert(tid, name, lsn);
+        for ((db, tid, name), lsn) in other.entries {
+            self.insert(db, tid, name, lsn);
         }
     }
 }
 
+/// Database-bound view over a [`TombstoneSet`].
+pub struct DatabaseTombstones<'a> {
+    set: &'a TombstoneSet,
+    database_id: u64,
+}
+
+impl DatabaseTombstones<'_> {
+    pub fn is_tombstoned(&self, tenant_id: u64, collection: &str, lsn: u64) -> bool {
+        self.set
+            .is_tombstoned(self.database_id, tenant_id, collection, lsn)
+    }
+}
+
+impl std::ops::Deref for DatabaseTombstones<'_> {
+    type Target = TombstoneSet;
+
+    fn deref(&self) -> &Self::Target {
+        self.set
+    }
+}
+
 /// Single-pass extraction: walk `records`, decode every
 /// `CollectionTombstoned` entry, and return the resulting set.
 ///
@@ -109,6 +144,7 @@ pub fn extract_tombstones(records: &[WalRecord]) -> TombstoneSet {
         // this call site changes to `decrypt_payload_ring` first.
         match CollectionTombstonePayload::from_bytes(&record.payload) {
             Ok(payload) => set.insert(
+                record.header.database_id,
                 record.header.tenant_id,
                 payload.collection,
                 payload.purge_lsn,
@@ -129,7 +165,13 @@ mod tests {
     use super::*;
     use crate::record::{WalRecord, WalRecordArgs};
 
-    fn tombstone_record(tenant: u64, name: &str, purge_lsn: u64, record_lsn: u64) -> WalRecord {
+    fn tombstone_record(
+        database: u64,
+        tenant: u64,
+        name: &str,
+        purge_lsn: u64,
+        record_lsn: u64,
+    ) -> WalRecord {
         let payload = CollectionTombstonePayload::new(name, purge_lsn)
             .to_bytes()
             .unwrap();
@@ -138,7 +180,7 @@ mod tests {
             lsn: record_lsn,
             tenant_id: tenant,
             vshard_id: 0,
-            database_id: 0,
+            database_id: database,
             payload,
             encryption_key: None,
             preamble_bytes: None,
@@ -149,42 +191,43 @@ mod tests {
     #[test]
     fn is_tombstoned_shadows_older_writes() {
         let mut set = TombstoneSet::new();
-        set.insert(1, "users".into(), 100);
-        assert!(set.is_tombstoned(1, "users", 50));
-        assert!(!set.is_tombstoned(1, "users", 100));
-        assert!(!set.is_tombstoned(1, "users", 200));
-        assert!(!set.is_tombstoned(1, "other", 50));
-        assert!(!set.is_tombstoned(2, "users", 50));
+        set.insert(7, 1, "users".into(), 100);
+        assert!(set.is_tombstoned(7, 1, "users", 50));
+        assert!(!set.is_tombstoned(7, 1, "users", 100));
+        assert!(!set.is_tombstoned(7, 1, "users", 200));
+        assert!(!set.is_tombstoned(7, 1, "other", 50));
+        assert!(!set.is_tombstoned(7, 2, "users", 50));
+        assert!(!set.is_tombstoned(8, 1, "users", 50));
     }
 
     #[test]
     fn insert_keeps_highest_purge_lsn() {
         let mut set = TombstoneSet::new();
-        set.insert(1, "users".into(), 100);
-        set.insert(1, "users".into(), 50);
-        assert_eq!(set.purge_lsn(1, "users"), Some(100));
-        set.insert(1, "users".into(), 200);
-        assert_eq!(set.purge_lsn(1, "users"), Some(200));
+        set.insert(7, 1, "users".into(), 100);
+        set.insert(7, 1, "users".into(), 50);
+        assert_eq!(set.purge_lsn(7, 1, "users"), Some(100));
+        set.insert(7, 1, "users".into(), 200);
+        assert_eq!(set.purge_lsn(7, 1, "users"), Some(200));
     }
 
     #[test]
     fn extract_from_record_stream() {
         let records = vec![
-            tombstone_record(1, "users", 100, 10),
-            tombstone_record(1, "orders", 150, 11),
-            tombstone_record(2, "users", 200, 12),
+            tombstone_record(7, 1, "users", 100, 10),
+            tombstone_record(7, 1, "orders", 150, 11),
+            tombstone_record(8, 1, "users", 200, 12),
         ];
         let set = extract_tombstones(&records);
         assert_eq!(set.len(), 3);
-        assert_eq!(set.purge_lsn(1, "users"), Some(100));
-        assert_eq!(set.purge_lsn(1, "orders"), Some(150));
-        assert_eq!(set.purge_lsn(2, "users"), Some(200));
+        assert_eq!(set.purge_lsn(7, 1, "users"), Some(100));
+        assert_eq!(set.purge_lsn(7, 1, "orders"), Some(150));
+        assert_eq!(set.purge_lsn(8, 1, "users"), Some(200));
     }
 
     #[test]
     fn extract_ignores_non_tombstone_records() {
         let records = vec![
-            tombstone_record(1, "users", 100, 10),
+            tombstone_record(0, 1, "users", 100, 10),
             WalRecord::new(WalRecordArgs {
                 record_type: RecordType::Put as u32,
                 lsn: 11,
@@ -226,25 +269,25 @@ mod tests {
             preamble_bytes: None,
         })
         .unwrap();
-        let records = vec![bogus, tombstone_record(1, "users", 100, 10)];
+        let records = vec![bogus, tombstone_record(0, 1, "users", 100, 10)];
         let set = extract_tombstones(&records);
         assert_eq!(
             set.len(),
             1,
             "valid tombstone still captured despite peer corruption"
         );
-        assert_eq!(set.purge_lsn(1, "users"), Some(100));
+        assert_eq!(set.purge_lsn(0, 1, "users"), Some(100));
     }
 
     #[test]
     fn extend_merges_sets() {
         let mut a = TombstoneSet::new();
-        a.insert(1, "users".into(), 100);
+        a.insert(7, 1, "users".into(), 100);
         let mut b = TombstoneSet::new();
-        b.insert(1, "users".into(), 150);
-        b.insert(1, "orders".into(), 200);
+        b.insert(7, 1, "users".into(), 150);
+        b.insert(8, 1, "orders".into(), 200);
         a.extend(b);
-        assert_eq!(a.purge_lsn(1, "users"), Some(150));
-        assert_eq!(a.purge_lsn(1, "orders"), Some(200));
+        assert_eq!(a.purge_lsn(7, 1, "users"), Some(150));
+        assert_eq!(a.purge_lsn(8, 1, "orders"), Some(200));
     }
 }

nodedb-wal/src/replay/mod.rs

@@ -9,4 +9,4 @@
 
 pub mod filter;
 
-pub use filter::{TombstoneSet, extract_tombstones};
+pub use filter::{DatabaseTombstones, TombstoneSet, extract_tombstones};

nodedb-wal/tests/wal_collection_tombstone.rs

@@ -98,24 +98,24 @@ fn extract_and_shadow_writes_before_purge_lsn() {
     let set: TombstoneSet = extract_tombstones(&records);
 
     assert_eq!(set.len(), 1);
-    assert_eq!(set.purge_lsn(1, "users"), Some(purge_lsn));
+    assert_eq!(set.purge_lsn(0, 1, "users"), Some(purge_lsn));
 
     for lsn in &put_lsns {
         assert!(
-            set.is_tombstoned(1, "users", *lsn),
+            set.is_tombstoned(0, 1, "users", *lsn),
             "pre-purge write at lsn {lsn} must be shadowed"
         );
     }
     assert!(
-        !set.is_tombstoned(1, "orders", orders_lsn),
+        !set.is_tombstoned(0, 1, "orders", orders_lsn),
         "different collection must not be shadowed"
     );
     assert!(
-        !set.is_tombstoned(2, "users", other_tenant_lsn),
+        !set.is_tombstoned(0, 2, "users", other_tenant_lsn),
         "different tenant must not be shadowed"
     );
     assert!(
-        !set.is_tombstoned(1, "users", post_purge_lsn),
+        !set.is_tombstoned(0, 1, "users", post_purge_lsn),
         "post-purge write (lsn >= purge_lsn) must not be shadowed"
     );
 }
@@ -160,9 +160,9 @@ fn multiple_tombstones_keep_highest_purge_lsn() {
 
     let records = read_all(&path);
     let set = extract_tombstones(&records);
-    assert_eq!(set.purge_lsn(1, "users"), Some(500));
-    assert!(set.is_tombstoned(1, "users", 499));
-    assert!(!set.is_tombstoned(1, "users", 500));
+    assert_eq!(set.purge_lsn(0, 1, "users"), Some(500));
+    assert!(set.is_tombstoned(0, 1, "users", 499));
+    assert!(!set.is_tombstoned(0, 1, "users", 500));
 }
 
 #[test]