fix: isolate drop recreate collection state

Author: farhan-syah

nodedb-test-support/src/cluster_harness/node/lifecycle/spawn_full.rs

@@ -96,7 +96,7 @@ impl TestClusterNode {
             &data_dir_path.join("test.wal"),
         )?);
         let wal_records: Arc<[nodedb_wal::WalRecord]> = Arc::from(wal.replay()?.into_boxed_slice());
-        let replay_tombstones = nodedb_wal::extract_tombstones(&wal_records);
+        let replay_tombstones = nodedb_wal::extract_tombstones(&wal_records).unwrap();
         let (dispatcher, data_sides) = Dispatcher::new(num_cores, 1024);
         let (event_producers, event_consumers) = create_event_bus(num_cores);
 

nodedb-test-support/src/pgwire_harness/restart.rs

@@ -172,7 +172,7 @@ impl TestServer {
         let wal = Arc::new(WalManager::open_for_testing(&wal_path).unwrap());
         let wal_records: Arc<[nodedb_wal::WalRecord]> =
             Arc::from(wal.replay().unwrap().into_boxed_slice());
-        let replay_tombstones = nodedb_wal::extract_tombstones(&wal_records);
+        let replay_tombstones = nodedb_wal::extract_tombstones(&wal_records).unwrap();
 
         let (dispatcher, data_sides) = Dispatcher::new(1, 64);
         let (event_producers, event_consumers) = create_event_bus(1);

nodedb-wal/src/replay/filter.rs

@@ -125,11 +125,9 @@ impl std::ops::Deref for DatabaseTombstones<'_> {
 /// Single-pass extraction: walk `records`, decode every
 /// `CollectionTombstoned` entry, and return the resulting set.
 ///
-/// Records that fail to decode are logged and skipped — replay callers
-/// that need hard failure on corrupt tombstones should validate CRCs
-/// upstream; a CRC-passing but structurally corrupt payload is an
-/// out-of-band programmer bug, not a run-time recoverable condition.
-pub fn extract_tombstones(records: &[WalRecord]) -> TombstoneSet {
+/// A malformed tombstone fails extraction. Skipping one would replay writes
+/// below an unknown purge boundary and can resurrect a dropped collection.
+pub fn extract_tombstones(records: &[WalRecord]) -> crate::Result<TombstoneSet> {
     let mut set = TombstoneSet::new();
     for record in records {
         let Some(kind) = RecordType::from_raw(record.logical_record_type()) else {
@@ -142,22 +140,15 @@ pub fn extract_tombstones(records: &[WalRecord]) -> TombstoneSet {
         // they carry only a collection name and an LSN, no secrets.
         // If payload-level encryption is later extended to tombstones,
         // 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,
-            ),
-            Err(e) => tracing::warn!(
-                lsn = record.header.lsn,
-                tenant_id = record.header.tenant_id,
-                error = %e,
-                "skipping malformed CollectionTombstoned record during tombstone extraction",
-            ),
-        }
+        let payload = CollectionTombstonePayload::from_bytes(&record.payload)?;
+        set.insert(
+            record.header.database_id,
+            record.header.tenant_id,
+            payload.collection,
+            payload.purge_lsn,
+        );
     }
-    set
+    Ok(set)
 }
 
 #[cfg(test)]
@@ -217,7 +208,7 @@ mod tests {
             tombstone_record(7, 1, "orders", 150, 11),
             tombstone_record(8, 1, "users", 200, 12),
         ];
-        let set = extract_tombstones(&records);
+        let set = extract_tombstones(&records).unwrap();
         assert_eq!(set.len(), 3);
         assert_eq!(set.purge_lsn(7, 1, "users"), Some(100));
         assert_eq!(set.purge_lsn(7, 1, "orders"), Some(150));
@@ -251,12 +242,12 @@ mod tests {
             })
             .unwrap(),
         ];
-        let set = extract_tombstones(&records);
+        let set = extract_tombstones(&records).unwrap();
         assert_eq!(set.len(), 1);
     }
 
     #[test]
-    fn extract_tolerates_corrupt_payload() {
+    fn extract_rejects_corrupt_payload() {
         // Build a tombstone-typed record whose payload is too short to decode.
         let bogus = WalRecord::new(WalRecordArgs {
             record_type: RecordType::CollectionTombstoned as u32,
@@ -270,13 +261,10 @@ mod tests {
         })
         .unwrap();
         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!(
+            extract_tombstones(&records).is_err(),
+            "malformed tombstone must fail replay rather than lose a purge boundary"
         );
-        assert_eq!(set.purge_lsn(0, 1, "users"), Some(100));
     }
 
     #[test]

nodedb-wal/tests/wal_collection_tombstone.rs

@@ -95,7 +95,7 @@ fn extract_and_shadow_writes_before_purge_lsn() {
     writer.sync().unwrap();
 
     let records = read_all(&path);
-    let set: TombstoneSet = extract_tombstones(&records);
+    let set: TombstoneSet = extract_tombstones(&records).unwrap();
 
     assert_eq!(set.len(), 1);
     assert_eq!(set.purge_lsn(0, 1, "users"), Some(purge_lsn));
@@ -159,7 +159,7 @@ fn multiple_tombstones_keep_highest_purge_lsn() {
     writer.sync().unwrap();
 
     let records = read_all(&path);
-    let set = extract_tombstones(&records);
+    let set = extract_tombstones(&records).unwrap();
     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));
@@ -179,6 +179,6 @@ fn extract_ignores_unrelated_record_types() {
 
     let records = read_all(&path);
     assert_eq!(records.len(), 4);
-    let set = extract_tombstones(&records);
+    let set = extract_tombstones(&records).unwrap();
     assert!(set.is_empty());
 }

nodedb/src/bootstrap/cluster_ready.rs

@@ -150,25 +150,20 @@ pub async fn await_cluster_ready(
         return Err(e);
     }
 
+    // A pending name-scoped reclaim must complete before the gateway opens.
+    // Otherwise a same-name CREATE can install a replacement that the delayed
+    // retry subsequently erases. Fail readiness and let the operator restart
+    // after the underlying storage fault is resolved.
+    if let Err(error) = crate::event::collection_gc::pending_reclaim::drain_once(shared).await {
+        data_groups_gate.fail(format!("pending collection reclaim failed: {error}"));
+        return Err(anyhow::anyhow!(
+            "pending collection reclaim failed during startup: {error}"
+        ));
+    }
+
     data_groups_gate.fire();
     transport_gate.fire();
 
-    // Boot-repair outstanding engine purges: a node that crashed with a
-    // dropped collection's catalog row already removed but its redb +
-    // versioned engine purge incomplete recorded a `_system.pending_reclaim`
-    // entry (or would have, had it stayed up). Drain that table once at
-    // boot — re-running the engine purge for each pending entry — so the
-    // reclaim completes promptly instead of waiting for the first worker
-    // tick. NOT a readiness gate: a purge hiccup must never wedge boot, so
-    // this is spawned and any per-entry failure is left for the
-    // pending-reclaim worker to retry.
-    {
-        let drain_shared = Arc::clone(shared);
-        tokio::spawn(async move {
-            crate::event::collection_gc::pending_reclaim::drain_once(&drain_shared).await;
-        });
-    }
-
     // Warm the QUIC peer cache so the first replicated request
     // after boot doesn't pay a cold dial.
     if let (Some(transport), Some(topology)) = (

nodedb/src/bootstrap/wal_init.rs

@@ -68,43 +68,29 @@ pub fn init_wal(
          for at-rest catalog encryption"
     );
 
-    let tombstones = load_tombstones(config, &wal_records);
+    let tombstones = load_tombstones(config, &wal_records)?;
     Ok((wal, wal_records, tombstones))
 }
 
 fn load_tombstones(
     config: &ServerConfig,
     wal_records: &Arc<[nodedb_wal::WalRecord]>,
-) -> nodedb_wal::TombstoneSet {
+) -> anyhow::Result<nodedb_wal::TombstoneSet> {
     let catalog_path = config.catalog_path();
-    let mut set = nodedb_wal::extract_tombstones(wal_records);
-    match crate::control::security::catalog::SystemCatalog::open(&catalog_path) {
-        Ok(catalog) => match catalog.load_wal_tombstones() {
-            Ok(persisted) => {
-                if !persisted.is_empty() {
-                    info!(
-                        persisted = persisted.len(),
-                        in_wal = set.len(),
-                        "merging persisted collection tombstones into replay set"
-                    );
-                }
-                set.extend(persisted);
-            }
-            Err(e) => {
-                tracing::warn!(
-                    error = %e,
-                    "failed to load _system.wal_tombstones at startup — \
-                     replay will see WAL-extracted tombstones only"
-                );
-            }
-        },
-        Err(e) => {
-            tracing::warn!(
-                error = %e,
-                "could not open system catalog to load persisted WAL tombstones — \
-                 falling back to WAL-extracted set"
-            );
-        }
+    let mut set = nodedb_wal::extract_tombstones(wal_records)
+        .map_err(|error| anyhow::anyhow!("extract WAL tombstones: {error}"))?;
+    let catalog = crate::control::security::catalog::SystemCatalog::open(&catalog_path)
+        .map_err(|error| anyhow::anyhow!("open catalog for WAL tombstones: {error}"))?;
+    let persisted = catalog
+        .load_wal_tombstones()
+        .map_err(|error| anyhow::anyhow!("load persisted WAL tombstones: {error}"))?;
+    if !persisted.is_empty() {
+        info!(
+            persisted = persisted.len(),
+            in_wal = set.len(),
+            "merging persisted collection tombstones into replay set"
+        );
     }
-    set
+    set.extend(persisted);
+    Ok(set)
 }