fix: single-core L1 file reclaim and distinguish retry-owned purge failures

Route shared on-disk L1 checkpoint/partition reclaim (vector, spatial,
sparse-vector, timeseries) to a single homing core per collection during
the all-cores UnregisterCollection fan-out, since those paths are keyed
by (database, tenant, collection) rather than by core and concurrent
cores racing remove_dir_all/unlink on the same tree could turn a benign
concurrent-removal errno into a fatal barrier failure. Other cores still
evict their own per-core in-memory state.

Replace the untyped reclaim error with ReclaimFailure, which records
whether a durable pending_reclaim retry record was persisted. Callers
now disarm their lifecycle guard only when a retry owner exists;
otherwise the guard's unwind Drop releases the in-memory drain so a
same-name CREATE can self-heal off the durable inactive catalog row
instead of wedging behind an orphaned hold. Also make catalog existence
checks in collection/materialized-view CREATE abort on a read fault
rather than silently treating it as "does not exist".

Author: farhan-syah

nodedb-physical/src/physical_plan/meta.rs

@@ -131,6 +131,15 @@ pub enum MetaOp {
         tenant_id: u64,
         name: String,
         purge_lsn: u64,
+        /// Whether this core must reclaim the collection's shared on-disk L1
+        /// checkpoint/partition files. Those paths are keyed by
+        /// `(database, tenant, collection)` — not by core — so the Control
+        /// Plane sets this `true` for exactly one core (the collection's
+        /// homing vshard) in the all-cores fan-out. Every core still evicts
+        /// its own per-core in-memory state; only the shared-path unlink /
+        /// `remove_dir_all` is single-cored, so concurrent cores cannot race
+        /// on the same tree.
+        reclaim_l1_files: bool,
     },
 
     /// Reclaim every local Data Plane resource for a single

nodedb/src/bridge/quiesce/drain.rs

@@ -153,27 +153,6 @@ impl CollectionQuiesce {
         }
     }
 
-    /// Wait until no lifecycle drain owns `(database, tenant, collection)`.
-    pub async fn wait_until_not_draining(
-        &self,
-        database_id: u64,
-        tenant_id: u64,
-        collection: &str,
-    ) {
-        loop {
-            let notified = self.notify.notified();
-            tokio::pin!(notified);
-            // Register before checking the state. `notify_waiters` does not
-            // store a permit for a future waiter, so checking first would lose
-            // a clear/forget notification in the check-to-poll gap.
-            notified.as_mut().enable();
-            if !self.is_draining(database_id, tenant_id, collection) {
-                return;
-            }
-            notified.await;
-        }
-    }
-
     /// Returns a future that resolves once every open scan against
     /// `(tenant_id, collection)` has completed. Safe to await from the
     /// Control Plane (tokio) — internally uses [`tokio::sync::Notify`]
@@ -314,19 +293,13 @@ mod tests {
     }
 
     #[tokio::test]
-    async fn create_waiter_resumes_after_lifecycle_drain_is_forgotten() {
+    async fn single_lifecycle_holder_clears_on_forget() {
         let q = CollectionQuiesce::new();
         q.begin_drain(DB, 1, "c");
-
-        let q_clone = Arc::clone(&q);
-        let waiter = tokio::spawn(async move {
-            q_clone.wait_until_not_draining(DB, 1, "c").await;
-        });
-        tokio::task::yield_now().await;
-        assert!(!waiter.is_finished());
+        assert!(q.is_draining(DB, 1, "c"));
 
         q.forget(DB, 1, "c");
-        waiter.await.unwrap();
+        assert!(!q.is_draining(DB, 1, "c"));
     }
 
     #[tokio::test]
@@ -346,23 +319,18 @@ mod tests {
     }
 
     #[tokio::test]
-    async fn create_waiter_requires_every_lifecycle_holder_to_finish() {
+    async fn is_draining_until_every_lifecycle_holder_forgets() {
         let q = CollectionQuiesce::new();
         q.begin_drain(DB, 1, "c");
         q.begin_drain(DB, 1, "c");
 
-        let q_clone = Arc::clone(&q);
-        let waiter = tokio::spawn(async move {
-            q_clone.wait_until_not_draining(DB, 1, "c").await;
-        });
-        tokio::task::yield_now().await;
-
+        // One holder released — still draining while the other holds.
         q.forget(DB, 1, "c");
-        tokio::task::yield_now().await;
-        assert!(!waiter.is_finished());
+        assert!(q.is_draining(DB, 1, "c"));
 
+        // Last holder released — drain clears.
         q.forget(DB, 1, "c");
-        waiter.await.unwrap();
+        assert!(!q.is_draining(DB, 1, "c"));
     }
 
     #[tokio::test]

nodedb/src/control/catalog_entry/post_apply/async_dispatch/collection.rs

@@ -19,6 +19,49 @@ pub async fn put_async(stored: StoredCollection, shared: Arc<SharedState>) {
     collection::put_async(stored, shared).await;
 }
 
+/// Failure outcome of [`reclaim_collection_storage`].
+///
+/// `retry_queued` distinguishes the two failure shapes a lifecycle-guard
+/// holder must handle differently:
+///
+/// - `true` — a durable `_system.pending_reclaim` record was persisted, so a
+///   worker (and the boot-time drain) owns the retry and will release the
+///   lifecycle drain via `forget` once it completes. The holder must `disarm`
+///   its guard so it does NOT also release the drain.
+/// - `false` — no durable owner exists for a retry (the WAL/redb tombstone
+///   writes failed before any record was queued, or queuing the record itself
+///   failed). The holder must let its guard `Drop` release the in-memory drain
+///   so a same-name CREATE can re-acquire the lifecycle and self-heal off the
+///   durable inactive catalog row. Leaking the drain here would wedge every
+///   future same-name CREATE (and the GC sweeper) until the node restarts.
+#[derive(Debug)]
+pub(crate) struct ReclaimFailure {
+    pub(crate) error: crate::Error,
+    pub(crate) retry_queued: bool,
+}
+
+impl ReclaimFailure {
+    pub(crate) fn no_retry(error: impl Into<crate::Error>) -> Self {
+        Self {
+            error: error.into(),
+            retry_queued: false,
+        }
+    }
+
+    fn retry_queued(error: crate::Error) -> Self {
+        Self {
+            error,
+            retry_queued: true,
+        }
+    }
+}
+
+impl std::fmt::Display for ReclaimFailure {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(f, "{}", self.error)
+    }
+}
+
 /// Reclaim every engine's storage for `(tenant_id, name)` on this node — WAL
 /// tombstone, redb tombstone, optional L2 cleanup enqueue, quiesce drain,
 /// `MetaOp::UnregisterCollection` dispatch to the local Data Plane, and Lite
@@ -35,20 +78,27 @@ pub(crate) async fn reclaim_collection_storage(
     name: &str,
     purge_lsn: u64,
     drain_already_held: bool,
-) -> crate::Result<()> {
-    // 1. Persist to redb (every node has its own catalog).
+) -> Result<(), ReclaimFailure> {
+    // 1. Persist to redb (every node has its own catalog). A failure here
+    // leaves no durable retry owner, so it is a `no_retry` failure: the caller
+    // releases its lifecycle guard rather than leaking the drain.
     let catalog = shared.credentials.catalog();
-    catalog.record_wal_tombstone(database_id, tenant_id, name, purge_lsn)?;
+    catalog
+        .record_wal_tombstone(database_id, tenant_id, name, purge_lsn)
+        .map_err(ReclaimFailure::no_retry)?;
 
     // 2. Append to local WAL. Both durable tombstone surfaces are required
     // before storage reclaim; otherwise truncation or catalog loss can replay
     // predecessor writes after a same-name CREATE.
-    shared.wal.append_collection_tombstone(
-        TenantId::new(tenant_id),
-        DatabaseId::new(database_id),
-        name,
-        purge_lsn,
-    )?;
+    shared
+        .wal
+        .append_collection_tombstone(
+            TenantId::new(tenant_id),
+            DatabaseId::new(database_id),
+            name,
+            purge_lsn,
+        )
+        .map_err(ReclaimFailure::no_retry)?;
 
     // 2b. Enqueue an L2 cleanup entry if cold storage is configured.
     // Recorded even when `bytes_pending` is unknown (0) — the worker
@@ -123,49 +173,59 @@ pub(crate) async fn reclaim_collection_storage(
             )
         });
 
-    if let Err(e) = &purge_result {
-        // Keep the lifecycle drain marker set. A same-name CREATE waits until
-        // the durable retry succeeds; releasing it here would let that retry
-        // erase the replacement because engine keys are name-scoped.
-        if let Err(record_error) = record_pending_reclaim(
-            shared,
-            database_id,
-            tenant_id,
-            name,
-            purge_lsn,
-            &e.to_string(),
-        ) {
-            return Err(crate::Error::Storage {
-                engine: "pending-reclaim".into(),
-                detail: format!(
-                    "collection reclaim failed ({e}); durable retry record also failed: {record_error}"
-                ),
-            });
+    match purge_result {
+        Err(e) => {
+            // Keep the lifecycle drain marker set ONLY when a durable retry
+            // record is persisted: a worker then owns the retry and releases
+            // the drain via `forget`. A same-name CREATE waits until that
+            // retry succeeds, because engine keys are name-scoped. If recording
+            // the durable entry itself fails there is no owner to release the
+            // drain, so this is a `no_retry` failure and the caller must let
+            // its guard release the in-memory hold.
+            match record_pending_reclaim(
+                shared,
+                database_id,
+                tenant_id,
+                name,
+                purge_lsn,
+                &e.to_string(),
+            ) {
+                Ok(()) => Err(ReclaimFailure::retry_queued(e)),
+                Err(record_error) => Err(ReclaimFailure::no_retry(crate::Error::Storage {
+                    engine: "pending-reclaim".into(),
+                    detail: format!(
+                        "collection reclaim failed ({e}); durable retry record also failed: {record_error}"
+                    ),
+                })),
+            }
         }
-    } else {
-        // Broadcast only after every core reclaimed the old incarnation.
-        // Saturated per-session channels may drop the notification; offline
-        // replay remains the fallback.
-        shared
-            .crdt_sync_delivery
-            .broadcast_collection_purged(tenant_id, name, purge_lsn);
-
-        // A prior failed attempt may have left a durable entry; a succeeding
-        // purge clears it, then releases CREATE waiters.
-        let catalog = shared.credentials.catalog();
-        catalog.remove_pending_reclaim(database_id, tenant_id, name)?;
-        if !drain_already_held {
-            shared.quiesce.forget(database_id, tenant_id, name);
+        Ok(()) => {
+            // Broadcast only after every core reclaimed the old incarnation.
+            // Saturated per-session channels may drop the notification; offline
+            // replay remains the fallback.
+            shared
+                .crdt_sync_delivery
+                .broadcast_collection_purged(tenant_id, name, purge_lsn);
+
+            // A prior failed attempt may have left a durable entry; a
+            // succeeding purge clears it, then releases CREATE waiters.
+            shared
+                .credentials
+                .catalog()
+                .remove_pending_reclaim(database_id, tenant_id, name)
+                .map_err(ReclaimFailure::no_retry)?;
+            if !drain_already_held {
+                shared.quiesce.forget(database_id, tenant_id, name);
+            }
+            debug!(
+                collection = %name,
+                tenant = tenant_id,
+                purge_lsn,
+                "catalog_entry: UnregisterCollection reclaimed on local Data Plane"
+            );
+            Ok(())
         }
-        debug!(
-            collection = %name,
-            tenant = tenant_id,
-            purge_lsn,
-            "catalog_entry: UnregisterCollection reclaimed on local Data Plane"
-        );
     }
-
-    purge_result
 }
 
 /// Persist a durable `_system.pending_reclaim` entry so the failed

nodedb/src/control/catalog_entry/post_apply/async_dispatch/materialized_view.rs

@@ -24,7 +24,7 @@ pub async fn delete_async(
     name: String,
     purge_lsn: u64,
     shared: Arc<SharedState>,
-) -> crate::Result<()> {
+) -> Result<(), super::collection::ReclaimFailure> {
     super::collection::reclaim_collection_storage(&shared, 0, tenant_id, &name, purge_lsn, false)
         .await
 }

nodedb/src/control/catalog_entry/post_apply/mod.rs

@@ -41,6 +41,6 @@ mod async_dispatch;
 pub(crate) mod gateway_invalidation;
 mod sync;
 
-pub(crate) use async_dispatch::collection::reclaim_collection_storage;
+pub(crate) use async_dispatch::collection::{ReclaimFailure, reclaim_collection_storage};
 pub use async_dispatch::spawn_post_apply_async_side_effects;
 pub use sync::apply_post_apply_side_effects_sync;

nodedb/src/control/server/shared/ddl/neutral/collection/create/build.rs

@@ -152,8 +152,13 @@ pub async fn build_and_persist(
         ));
     }
 
-    // Check if the object already exists.
-    if let Ok(Some(existing)) = catalog.get_collection(database_id, tenant_id.as_u64(), name) {
+    // Check if the object already exists. A catalog-read fault must abort the
+    // CREATE — proceeding as if no row exists could build a fresh collection
+    // over a soft-deleted incarnation's still-present storage.
+    let existing = catalog
+        .get_collection(database_id, tenant_id.as_u64(), name)
+        .map_err(|error| err("XX000", error.to_string()))?;
+    if let Some(existing) = existing {
         if existing.is_active {
             return Err(err(
                 "42P07",
@@ -189,11 +194,16 @@ pub async fn build_and_persist(
                 local_lifecycle.is_some(),
             )
             .await;
-        if let Err(error) = purge_result {
-            if let Some(guard) = local_lifecycle.take() {
+        if let Err(failure) = purge_result {
+            // Only disarm when a durable retry record owns the drain. Otherwise
+            // let the guard release the in-memory hold so this same-name CREATE
+            // can be retried against the durable inactive catalog row.
+            if failure.retry_queued
+                && let Some(guard) = local_lifecycle.take()
+            {
                 guard.disarm();
             }
-            return Err(err("XX000", error.to_string()));
+            return Err(err("XX000", failure.error.to_string()));
         }
     }
 

nodedb/src/control/server/shared/ddl/neutral/collection/drop.rs

@@ -283,11 +283,16 @@ pub fn drop_collection(
                     .await
                 })
             });
-            if let Err(error) = purge_result {
-                if let Some(guard) = local_lifecycle.take() {
+            if let Err(failure) = purge_result {
+                // Disarm only when a durable retry record owns the drain; on a
+                // no-retry failure the guard's unwind Drop releases the hold so
+                // a same-name CREATE is not wedged behind an orphaned drain.
+                if failure.retry_queued
+                    && let Some(guard) = local_lifecycle.take()
+                {
                     guard.disarm();
                 }
-                panic!("local collection reclaim failed: {error}");
+                panic!("local collection reclaim failed: {}", failure.error);
             }
             state
                 .permissions

nodedb/src/control/server/shared/ddl/neutral/collection/purge/dispatch.rs

@@ -46,6 +46,14 @@ pub async fn dispatch_unregister_collection(
             .dispatcher
             .lock()
             .unwrap_or_else(|poisoned| poisoned.into_inner());
+        // The shared on-disk L1 files are keyed by (database, tenant,
+        // collection), so exactly one core reclaims them. Route to the
+        // collection's homing vshard; fall back to core 0 if the router
+        // cannot resolve it, so the files are never orphaned.
+        let homing_core = dispatcher
+            .router()
+            .resolve(VShardId::from_collection_in_database(database, name))
+            .unwrap_or(0);
         for core_id in 0..num_cores {
             let request_id = state.next_request_id();
             let request = Request {
@@ -57,6 +65,7 @@ pub async fn dispatch_unregister_collection(
                     tenant_id,
                     name: name.to_string(),
                     purge_lsn,
+                    reclaim_l1_files: core_id == homing_core,
                 }),
                 deadline: Instant::now() + timeout,
                 priority: Priority::Background,