From 0481b200065d985b41dabbf73b6b37d0d8f248b8 Mon Sep 17 00:00:00 2001 From: Farhan Syah Date: Tue, 21 Jul 2026 13:54:15 +0000 Subject: [PATCH 1/2] fix: isolate drop recreate collection state --- .../node/lifecycle/spawn_full.rs | 2 +- .../src/pgwire_harness/restart.rs | 2 +- nodedb-wal/src/replay/filter.rs | 46 ++-- nodedb-wal/tests/wal_collection_tombstone.rs | 6 +- nodedb/src/bootstrap/cluster_ready.rs | 27 +-- nodedb/src/bootstrap/wal_init.rs | 48 ++--- nodedb/src/bridge/quiesce/drain.rs | 204 ++++++++++++++++-- nodedb/src/bridge/quiesce/refcount.rs | 10 +- .../control/catalog_entry/apply/collection.rs | 66 +++--- .../catalog_entry/apply/materialized_view.rs | 23 +- nodedb/src/control/catalog_entry/apply/mod.rs | 31 ++- .../src/control/catalog_entry/apply/owner.rs | 22 ++ nodedb/src/control/catalog_entry/entry.rs | 7 +- .../post_apply/async_dispatch/collection.rs | 175 ++++++--------- .../post_apply/async_dispatch/dispatcher.rs | 49 +++-- .../async_dispatch/materialized_view.rs | 100 ++------- .../catalog_entry/post_apply/collection.rs | 4 +- .../ddl/neutral/collection/create/build.rs | 53 ++++- .../shared/ddl/neutral/collection/drop.rs | 56 ++++- .../ddl/neutral/collection/purge/dispatch.rs | 150 ++++++++----- .../ddl/neutral/collection/purge/hard.rs | 20 +- .../shared/ddl/neutral/collection/undrop.rs | 18 ++ .../ddl/neutral/materialized_view/create.rs | 130 +++++------ .../ddl/neutral/materialized_view/drop.rs | 77 ++++--- .../shared/ddl/neutral/timeseries/create.rs | 23 +- .../executor/handlers/aggregate/invalidate.rs | 25 ++- .../src/data/executor/handlers/reclaim/mod.rs | 26 ++- .../handlers/reclaim/sparse_vector.rs | 123 ++++++----- .../data/executor/handlers/reclaim/spatial.rs | 59 ++--- .../executor/handlers/reclaim/timeseries.rs | 37 ++-- .../data/executor/handlers/reclaim/vector.rs | 72 ++++--- .../handlers/unregister_collection.rs | 125 +++++++++-- .../handlers/unregister_materialized_view.rs | 12 +- .../event/collection_gc/pending_reclaim.rs | 66 ++++-- nodedb/src/event/collection_gc/sweeper.rs | 37 +++- nodedb/src/event/crdt_sync/delivery.rs | 4 +- nodedb/src/event/wal_replay.rs | 2 +- nodedb/tests/collection_hard_delete.rs | 19 +- nodedb/tests/sql_drop_collection.rs | 47 ++++ nodedb/tests/sql_materialized_view_refresh.rs | 104 +++++++++ 40 files changed, 1353 insertions(+), 754 deletions(-) diff --git a/nodedb-test-support/src/cluster_harness/node/lifecycle/spawn_full.rs b/nodedb-test-support/src/cluster_harness/node/lifecycle/spawn_full.rs index 9513aca8c..4f7eec7f7 100644 --- a/nodedb-test-support/src/cluster_harness/node/lifecycle/spawn_full.rs +++ b/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); diff --git a/nodedb-test-support/src/pgwire_harness/restart.rs b/nodedb-test-support/src/pgwire_harness/restart.rs index bfafdcede..ee0b43efe 100644 --- a/nodedb-test-support/src/pgwire_harness/restart.rs +++ b/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); diff --git a/nodedb-wal/src/replay/filter.rs b/nodedb-wal/src/replay/filter.rs index c184bd4c6..ea860ffd8 100644 --- a/nodedb-wal/src/replay/filter.rs +++ b/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 { 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] diff --git a/nodedb-wal/tests/wal_collection_tombstone.rs b/nodedb-wal/tests/wal_collection_tombstone.rs index 2a86dd864..0ca95cd87 100644 --- a/nodedb-wal/tests/wal_collection_tombstone.rs +++ b/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()); } diff --git a/nodedb/src/bootstrap/cluster_ready.rs b/nodedb/src/bootstrap/cluster_ready.rs index 9ed629ff7..69e59f19f 100644 --- a/nodedb/src/bootstrap/cluster_ready.rs +++ b/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)) = ( diff --git a/nodedb/src/bootstrap/wal_init.rs b/nodedb/src/bootstrap/wal_init.rs index 1227e104b..ed889c80e 100644 --- a/nodedb/src/bootstrap/wal_init.rs +++ b/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 { 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) } diff --git a/nodedb/src/bridge/quiesce/drain.rs b/nodedb/src/bridge/quiesce/drain.rs index 41523fb7b..9954613b2 100644 --- a/nodedb/src/bridge/quiesce/drain.rs +++ b/nodedb/src/bridge/quiesce/drain.rs @@ -9,20 +9,45 @@ use std::task::{Context, Poll}; use super::refcount::CollectionQuiesce; +/// Exclusive per-name lifecycle hold used by no-Raft DDL. +/// +/// Dropping the guard releases one drain hold and wakes CREATE waiters. Call +/// [`disarm`](LifecycleDrainGuard::disarm) after ownership is transferred to a +/// durable pending-reclaim record. +pub struct LifecycleDrainGuard { + registry: Arc, + database_id: u64, + tenant_id: u64, + collection: String, + active: bool, +} + +impl LifecycleDrainGuard { + pub fn disarm(mut self) { + self.active = false; + } +} + +impl Drop for LifecycleDrainGuard { + fn drop(&mut self) { + if self.active { + self.registry + .clear_drain(self.database_id, self.tenant_id, &self.collection); + } + } +} + impl CollectionQuiesce { - /// Mark `(tenant_id, collection)` as draining. After this call: - /// - `try_start_scan` returns `ScanStartError::Draining`. - /// - `wait_until_drained` can be awaited to block until open scans - /// reach zero. - /// - /// Idempotent: calling twice is a no-op on the second call. + /// Acquire one lifecycle drain hold for `(database, tenant, collection)`. + /// New scans and same-name CREATE operations remain blocked until every + /// matching hold is released by `clear_drain` or `forget`. pub fn begin_drain(&self, database_id: u64, tenant_id: u64, collection: &str) { let mut inner = self.inner_mut(); let entry = inner .states .entry((database_id, tenant_id, collection.to_string())) .or_default(); - entry.draining = true; + entry.drain_holders = entry.drain_holders.saturating_add(1); } /// Stop the drain marker, allowing new scans again. Only called when @@ -32,12 +57,18 @@ impl CollectionQuiesce { /// is garbage-collected via `forget`. pub fn clear_drain(&self, database_id: u64, tenant_id: u64, collection: &str) { let mut inner = self.inner_mut(); - if let Some(state) = inner - .states - .get_mut(&(database_id, tenant_id, collection.to_string())) - { - state.draining = false; + let key = (database_id, tenant_id, collection.to_string()); + let remove = if let Some(state) = inner.states.get_mut(&key) { + state.drain_holders = state.drain_holders.saturating_sub(1); + state.drain_holders == 0 && state.open_scans == 0 + } else { + false + }; + if remove { + inner.states.remove(&key); } + drop(inner); + self.notify.notify_waiters(); } /// Drop the entry entirely once reclaim has completed. After this, @@ -45,9 +76,102 @@ impl CollectionQuiesce { /// the purge handler right before emitting the reclaim ack. pub fn forget(&self, database_id: u64, tenant_id: u64, collection: &str) { let mut inner = self.inner_mut(); - inner + let key = (database_id, tenant_id, collection.to_string()); + let remove = if let Some(state) = inner.states.get_mut(&key) { + state.drain_holders = state.drain_holders.saturating_sub(1); + state.drain_holders == 0 + } else { + false + }; + if remove { + inner.states.remove(&key); + } + drop(inner); + self.notify.notify_waiters(); + } + + /// Try to acquire the exclusive lifecycle drain without waiting. + /// Synchronous DDL uses this form because it may run on a current-thread + /// Tokio runtime where blocking on an async waiter would panic. + pub fn try_acquire_lifecycle( + self: &Arc, + database_id: u64, + tenant_id: u64, + collection: &str, + ) -> Option { + let mut inner = self.inner_mut(); + let entry = inner .states - .remove(&(database_id, tenant_id, collection.to_string())); + .entry((database_id, tenant_id, collection.to_string())) + .or_default(); + if entry.drain_holders > 0 { + return None; + } + entry.drain_holders = 1; + Some(LifecycleDrainGuard { + registry: Arc::clone(self), + database_id, + tenant_id, + collection: collection.to_string(), + active: true, + }) + } + + /// Exclusively acquire the lifecycle drain for one collection name. + /// + /// The check-and-acquire is performed under the registry mutex, so two + /// concurrent local DDL operations cannot both enter the destructive + /// lifecycle section. + pub async fn acquire_lifecycle( + self: &Arc, + database_id: u64, + tenant_id: u64, + collection: &str, + ) -> LifecycleDrainGuard { + loop { + let notified = self.notify.notified(); + tokio::pin!(notified); + notified.as_mut().enable(); + { + let mut inner = self.inner_mut(); + let entry = inner + .states + .entry((database_id, tenant_id, collection.to_string())) + .or_default(); + if entry.drain_holders == 0 { + entry.drain_holders = 1; + return LifecycleDrainGuard { + registry: Arc::clone(self), + database_id, + tenant_id, + collection: collection.to_string(), + active: true, + }; + } + } + notified.await; + } + } + + /// 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 @@ -189,6 +313,58 @@ mod tests { drain_task.await.unwrap(); } + #[tokio::test] + async fn create_waiter_resumes_after_lifecycle_drain_is_forgotten() { + 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()); + + q.forget(DB, 1, "c"); + waiter.await.unwrap(); + } + + #[tokio::test] + async fn lifecycle_acquisition_is_exclusive() { + let q = CollectionQuiesce::new(); + let first = q.acquire_lifecycle(DB, 1, "c").await; + + let q_clone = Arc::clone(&q); + let second = tokio::spawn(async move { q_clone.acquire_lifecycle(DB, 1, "c").await }); + tokio::task::yield_now().await; + assert!(!second.is_finished()); + + drop(first); + let second = second.await.unwrap(); + drop(second); + assert!(!q.is_draining(DB, 1, "c")); + } + + #[tokio::test] + async fn create_waiter_requires_every_lifecycle_holder_to_finish() { + 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; + + q.forget(DB, 1, "c"); + tokio::task::yield_now().await; + assert!(!waiter.is_finished()); + + q.forget(DB, 1, "c"); + waiter.await.unwrap(); + } + #[tokio::test] async fn forget_clears_state() { let q = CollectionQuiesce::new(); diff --git a/nodedb/src/bridge/quiesce/refcount.rs b/nodedb/src/bridge/quiesce/refcount.rs index 9de5ab82a..94df7716f 100644 --- a/nodedb/src/bridge/quiesce/refcount.rs +++ b/nodedb/src/bridge/quiesce/refcount.rs @@ -12,9 +12,9 @@ use tokio::sync::Notify; pub(super) struct CollectionState { /// Number of scans currently open against this collection. pub(super) open_scans: usize, - /// `true` once `begin_drain` has been called; new scans are rejected - /// and the collection is waiting for `open_scans` to reach 0. - pub(super) draining: bool, + /// Number of lifecycle operations currently holding the collection drain. + /// CREATE remains blocked until every holder completes. + pub(super) drain_holders: usize, } /// Why `try_start_scan` refused a new scan. @@ -72,7 +72,7 @@ impl CollectionQuiesce { .states .entry((database_id, tenant_id, collection.to_string())) .or_default(); - if entry.draining { + if entry.drain_holders > 0 { return Err(ScanStartError::Draining); } entry.open_scans += 1; @@ -100,7 +100,7 @@ impl CollectionQuiesce { inner .states .get(&(database_id, tenant_id, collection.to_string())) - .is_some_and(|s| s.draining) + .is_some_and(|s| s.drain_holders > 0) } pub(super) fn release_scan(&self, database_id: u64, tenant_id: u64, collection: &str) { diff --git a/nodedb/src/control/catalog_entry/apply/collection.rs b/nodedb/src/control/catalog_entry/apply/collection.rs index 08bb559f6..8211c3323 100644 --- a/nodedb/src/control/catalog_entry/apply/collection.rs +++ b/nodedb/src/control/catalog_entry/apply/collection.rs @@ -59,58 +59,52 @@ pub fn put_if_absent(stored: &StoredCollection, catalog: &SystemCatalog) { } } -/// Hard-delete the catalog metadata for a collection: primary -/// `StoredCollection` row, owner row, and surrogate ↔ PK map. -/// -/// Returns `Err` if any storage step that actually removes data -/// fails. Callers with different failure semantics decide what to do: -/// the raft applier (`apply_to_inner`) runs symmetrically on every -/// node and log-and-continues, while the interactive re-CREATE -/// hard-purge propagates so a `CREATE COLLECTION` never builds over -/// data that was not actually purged. Reporting the error rather than -/// swallowing it is what lets each caller make that choice. -pub fn purge( +/// Persist the fail-closed catalog half of a purge before touching storage. +/// The inactive row survives crashes and prevents same-name CREATE/UNDROP from +/// crossing an incomplete reclaim. +pub fn prepare_purge( database_id: u64, tenant_id: u64, name: &str, catalog: &SystemCatalog, ) -> crate::Result<()> { let database_id = DatabaseId::new(database_id); - // Hard delete of the primary StoredCollection row. Symmetric - // with `apply/function.rs::delete` and the other hard-delete - // peers: remove the primary, then remove the owner row. - // - // The two-step DROP → retention-expiry → PURGE flow means the - // record may already be absent when this runs (operator-driven - // PURGE on a record the GC sweeper already reclaimed). The - // `delete_collection` helper is idempotent and returns `false` - // in that case, which is fine — we still call - // `delete_parent_owner` because the owner row may linger - // independently. - let removed = catalog.delete_collection(database_id, tenant_id, name)?; - debug!( - collection = %name, - tenant = tenant_id, - removed, - "catalog_entry: purge_collection primary row removed" - ); - super::owner::delete_parent_owner_in_database( + if let Some(mut stored) = catalog.get_collection(database_id, tenant_id, name)? { + stored.is_active = false; + catalog.put_collection(database_id, &stored)?; + } + Ok(()) +} + +/// Remove catalog metadata only after every persistent engine surface has been +/// reclaimed. The primary inactive row is deleted last, so any intermediate +/// failure continues to block same-name lifecycle operations across restart. +pub fn finalize_purge( + database_id: u64, + tenant_id: u64, + name: &str, + catalog: &SystemCatalog, +) -> crate::Result<()> { + let database_id = DatabaseId::new(database_id); + super::owner::delete_parent_owner_in_database_checked( object_type::COLLECTION, database_id.as_u64(), tenant_id, name, catalog, - ); - // Wipe the surrogate ↔ PK map for this collection. Surrogates are - // collection-scoped; once the primary row is gone they can never - // be observed again, so leaving the catalog rows behind would - // just be allocator-bloat. Mirrors the array-drop cleanup in - // `array_convert::convert_drop_array`. + )?; catalog.delete_all_surrogates_for_collection( database_id, nodedb_types::TenantId::new(tenant_id), name, )?; + let removed = catalog.delete_collection(database_id, tenant_id, name)?; + debug!( + collection = %name, + tenant = tenant_id, + removed, + "catalog_entry: purge_collection finalized" + ); Ok(()) } diff --git a/nodedb/src/control/catalog_entry/apply/materialized_view.rs b/nodedb/src/control/catalog_entry/apply/materialized_view.rs index 130089cf2..bec2b8d3c 100644 --- a/nodedb/src/control/catalog_entry/apply/materialized_view.rs +++ b/nodedb/src/control/catalog_entry/apply/materialized_view.rs @@ -25,14 +25,17 @@ pub fn put(stored: &StoredMaterializedView, catalog: &SystemCatalog) { ); } -pub fn delete(tenant_id: u64, name: &str, catalog: &SystemCatalog) { - if let Err(e) = catalog.delete_materialized_view(tenant_id, name) { - warn!( - view = %name, - tenant = tenant_id, - error = %e, - "catalog_entry: delete_materialized_view failed" - ); - } - super::owner::delete_parent_owner(object_type::MATERIALIZED_VIEW, tenant_id, name, catalog); +pub fn delete(tenant_id: u64, name: &str, catalog: &SystemCatalog) -> crate::Result<()> { + catalog.delete_materialized_view(tenant_id, name)?; + super::owner::delete_parent_owner_checked( + object_type::MATERIALIZED_VIEW, + tenant_id, + name, + catalog, + )?; + + // Preserve the target as inactive until synchronous post-apply reclaim + // succeeds. Its row is the restart-durable ownership/lifecycle barrier. + super::collection::prepare_purge(0, tenant_id, name, catalog)?; + Ok(()) } diff --git a/nodedb/src/control/catalog_entry/apply/mod.rs b/nodedb/src/control/catalog_entry/apply/mod.rs index ebce23cc5..2c1cc22f6 100644 --- a/nodedb/src/control/catalog_entry/apply/mod.rs +++ b/nodedb/src/control/catalog_entry/apply/mod.rs @@ -35,10 +35,11 @@ pub mod wal_tombstone; use crate::control::catalog_entry::entry::CatalogEntry; use crate::control::security::catalog::SystemCatalog; -/// Apply `entry` to `catalog`. Best-effort: per-variant errors are -/// logged + swallowed inside the family handlers so a single write -/// failure doesn't stall the raft apply path. Startup replay will -/// re-run the entry if needed. +/// Apply `entry` to `catalog`. Most per-variant errors are logged and +/// swallowed so startup replay can retry them. Compound lifecycle mutations +/// whose partial application would expose stale object state (currently +/// materialized-view definition + target deletion) fail closed by panicking +/// the applying node. /// /// Debug builds run the full referential-integrity verifier after /// every apply and panic on any violation. This catches the @@ -100,19 +101,11 @@ fn apply_to_inner(entry: &CatalogEntry, catalog: &SystemCatalog) { tenant_id, name, } => { - // Raft apply runs symmetrically on every node and must not - // abort on a per-node catalog storage hiccup — log and - // continue, exactly as before. The swallow is now an - // explicit choice at the call site rather than hidden - // inside `purge`; the interactive re-CREATE caller makes - // the opposite choice and propagates. - if let Err(e) = collection::purge(*database_id, *tenant_id, name, catalog) { - tracing::warn!( - collection = %name, - tenant = tenant_id, - error = %e, - "catalog_entry: purge_collection apply failed" - ); + // Preserve an inactive catalog row until synchronous post-apply + // storage reclaim succeeds. This row is the restart-durable + // same-name lifecycle barrier. + if let Err(error) = collection::prepare_purge(*database_id, *tenant_id, name, catalog) { + panic!("collection catalog purge preparation failed: {error}"); } } CatalogEntry::PutSequence(stored) => sequence::put(stored, catalog), @@ -148,7 +141,9 @@ fn apply_to_inner(entry: &CatalogEntry, catalog: &SystemCatalog) { CatalogEntry::RevokeApiKey { key_id } => api_key::revoke(key_id, catalog), CatalogEntry::PutMaterializedView(stored) => materialized_view::put(stored, catalog), CatalogEntry::DeleteMaterializedView { tenant_id, name } => { - materialized_view::delete(*tenant_id, name, catalog) + if let Err(error) = materialized_view::delete(*tenant_id, name, catalog) { + panic!("materialized-view catalog deletion failed: {error}"); + } } CatalogEntry::PutContinuousAggregate(stored) => continuous_aggregate::put(stored, catalog), CatalogEntry::DeleteContinuousAggregate { diff --git a/nodedb/src/control/catalog_entry/apply/owner.rs b/nodedb/src/control/catalog_entry/apply/owner.rs index decc93560..80d4c8c9b 100644 --- a/nodedb/src/control/catalog_entry/apply/owner.rs +++ b/nodedb/src/control/catalog_entry/apply/owner.rs @@ -136,3 +136,25 @@ pub(super) fn delete_parent_owner_in_database( ); } } + +/// Fail-closed owner deletion for compound lifecycle mutations whose primary +/// and owner rows must advance atomically. +pub(super) fn delete_parent_owner_checked( + object_type: &'static str, + tenant_id: u64, + object_name: &str, + catalog: &SystemCatalog, +) -> crate::Result<()> { + delete_parent_owner_in_database_checked(object_type, 0, tenant_id, object_name, catalog) +} + +pub(super) fn delete_parent_owner_in_database_checked( + object_type: &'static str, + database_id: u64, + tenant_id: u64, + object_name: &str, + catalog: &SystemCatalog, +) -> crate::Result<()> { + catalog.delete_owner(object_type, database_id, tenant_id, object_name)?; + Ok(()) +} diff --git a/nodedb/src/control/catalog_entry/entry.rs b/nodedb/src/control/catalog_entry/entry.rs index a9891b1de..f18d3a775 100644 --- a/nodedb/src/control/catalog_entry/entry.rs +++ b/nodedb/src/control/catalog_entry/entry.rs @@ -174,9 +174,10 @@ pub enum CatalogEntry { /// refresh loop picks up the new definition on its next tick /// and starts materializing rows from source → target. PutMaterializedView(Box), - /// Delete a materialized view definition. The target - /// collection is NOT deleted — operators drop it separately - /// with `DROP COLLECTION` if desired. + /// Delete a materialized-view definition and its implementation-owned + /// target collection as one replicated catalog mutation. Post-apply waits + /// for collection-wide Data Plane reclaim before advancing the applied + /// index, so a same-name re-CREATE starts from a fresh incarnation. DeleteMaterializedView { tenant_id: u64, name: String }, // ── Continuous Aggregate ─────────────────────────────────────── diff --git a/nodedb/src/control/catalog_entry/post_apply/async_dispatch/collection.rs b/nodedb/src/control/catalog_entry/post_apply/async_dispatch/collection.rs index 9b72e229b..6a27338fa 100644 --- a/nodedb/src/control/catalog_entry/post_apply/async_dispatch/collection.rs +++ b/nodedb/src/control/catalog_entry/post_apply/async_dispatch/collection.rs @@ -19,84 +19,36 @@ pub async fn put_async(stored: StoredCollection, shared: Arc) { collection::put_async(stored, shared).await; } -/// Dispatch `MetaOp::UnregisterCollection` to this node's local Data -/// Plane so every engine reclaims storage for the purged collection. -/// Called on every node (not just the leader), so each node's Data -/// Plane reclaims its own L1 segment files, memtables, compaction -/// debt, and WAL tombstone entries locally. +/// 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 +/// `CollectionPurged` broadcast. /// -/// Order of operations: -/// -/// 1. Persist the tombstone into the local `_system.wal_tombstones` -/// redb table. Makes startup replay O(1) and survives process -/// crashes between steps 2 and 3. -/// 2. Append a `CollectionTombstoned` record to the local WAL. Replay -/// constructs the same in-memory set whether or not step 1 committed -/// (belt + suspenders — if the redb write failed but WAL succeeded, -/// replay still sees the tombstone). -/// 3. Dispatch `MetaOp::UnregisterCollection` into the Data Plane to -/// reclaim engine-local storage. -pub async fn purge_async( - database_id: u64, - tenant_id: u64, - name: String, - purge_lsn: u64, - shared: Arc, -) { - // The Result is already durably captured inside - // `reclaim_collection_storage` (failure → `_system.pending_reclaim` - // for at-least-once retry). This raft post-apply path runs on every - // node and must not itself fail the apply, so it consumes the Result - // here. There is deliberately NO warn-and-forget of the engine-purge - // error on this path — the durable record IS the handling. - let _ = reclaim_collection_storage(&shared, database_id, tenant_id, &name, purge_lsn).await; -} - -/// Borrowed core of [`purge_async`]: 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 `CollectionPurged` broadcast. -/// -/// Split out from `purge_async` so the synchronous re-CREATE hard-purge -/// (`shared::ddl::neutral::collection::purge::hard_purge_collection`) can reuse the exact -/// same reclaim body against a `&SharedState` without an owned `Arc`, and -/// so the raft post-apply path and the re-CREATE path share one -/// implementation rather than a copy. +/// Shared by the synchronous replicated post-apply barrier, materialized-view +/// target deletion, and the interactive re-CREATE hard-purge. All callers use +/// this one result-checked implementation rather than duplicating lifecycle +/// cleanup. pub(crate) async fn reclaim_collection_storage( shared: &SharedState, database_id: u64, tenant_id: u64, name: &str, purge_lsn: u64, + drain_already_held: bool, ) -> crate::Result<()> { // 1. Persist to redb (every node has its own catalog). let catalog = shared.credentials.catalog(); - if let Err(e) = catalog.record_wal_tombstone(database_id, tenant_id, name, purge_lsn) { - warn!( - collection = %name, - tenant = tenant_id, - purge_lsn, - error = %e, - "failed to persist WAL tombstone to _system.wal_tombstones — \ - replay will fall back to WAL extraction" - ); - } + catalog.record_wal_tombstone(database_id, tenant_id, name, purge_lsn)?; - // 2. Append to local WAL. - if let Err(e) = shared.wal.append_collection_tombstone( + // 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, - ) { - warn!( - collection = %name, - tenant = tenant_id, - purge_lsn, - error = %e, - "failed to append CollectionTombstoned WAL record" - ); - } + )?; // 2b. Enqueue an L2 cleanup entry if cold storage is configured. // Recorded even when `bytes_pending` is unknown (0) — the worker @@ -138,7 +90,9 @@ pub(crate) async fn reclaim_collection_storage( // files while a scan is touching an mmap page faults the // whole TPC reactor — drain ordering is a correctness, not // performance, requirement. - shared.quiesce.begin_drain(database_id, tenant_id, name); + if !drain_already_held { + shared.quiesce.begin_drain(database_id, tenant_id, name); + } shared .quiesce .wait_until_drained(database_id, tenant_id, name) @@ -159,41 +113,49 @@ pub(crate) async fn reclaim_collection_storage( crate::control::server::shared::ddl::neutral::collection::purge::dispatch_unregister_collection( shared, database_id, tenant_id, name, purge_lsn, ) - .await; - - // 4b. Broadcast `CollectionPurged` to every connected Lite - // session subscribed to this collection. Fire-and-forget; - // each session's control-frame channel is bounded to 32 and - // any saturated channel drops the notification (the client - // picks it up on reconnect via the offline-replay path). - shared - .crdt_sync_delivery - .broadcast_collection_purged(tenant_id, name, purge_lsn); - - // 5. Drop the quiesce entry. From here on, the catalog has no - // record of the collection; queries return `collection_not_found`. - shared.quiesce.forget(database_id, tenant_id, name); + .await + .and_then(|()| { + crate::control::catalog_entry::apply::collection::finalize_purge( + database_id, + tenant_id, + name, + shared.credentials.catalog(), + ) + }); if let Err(e) = &purge_result { - record_pending_reclaim( + // 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}" + ), + }); + } } else { - // A prior failed attempt may have left a durable entry; a - // succeeding purge clears it so the worker stops retrying. + // 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(); - if let Err(rm) = catalog.remove_pending_reclaim(database_id, tenant_id, name) { - warn!( - collection = %name, - tenant = tenant_id, - error = %rm, - "failed to reap _system.pending_reclaim entry after successful engine purge" - ); + catalog.remove_pending_reclaim(database_id, tenant_id, name)?; + if !drain_already_held { + shared.quiesce.forget(database_id, tenant_id, name); } debug!( collection = %name, @@ -218,7 +180,7 @@ fn record_pending_reclaim( name: &str, purge_lsn: u64, last_error: &str, -) { +) -> crate::Result<()> { let catalog = shared.credentials.catalog(); let now_ns = std::time::SystemTime::now() .duration_since(std::time::UNIX_EPOCH) @@ -233,27 +195,14 @@ fn record_pending_reclaim( last_error: last_error.to_string(), attempts: 0, }; - if let Err(e) = catalog.enqueue_pending_reclaim(&entry) { - // The durable record itself failed. This is the one place we - // cannot make durable — log loudly at error so operators see it. - tracing::error!( - collection = %name, - tenant = tenant_id, - purge_lsn, - purge_error = %last_error, - record_error = %e, - "engine purge failed AND recording the pending-reclaim entry failed — \ - collection storage may survive behind a removed catalog row until the \ - DROP is re-proposed or the node reboots and re-drains" - ); - } else { - warn!( - collection = %name, - tenant = tenant_id, - purge_lsn, - error = %last_error, - "engine purge failed — recorded _system.pending_reclaim entry for \ - at-least-once retry by the pending-reclaim worker" - ); - } + catalog.enqueue_pending_reclaim(&entry)?; + warn!( + collection = %name, + tenant = tenant_id, + purge_lsn, + error = %last_error, + "engine purge failed — recorded _system.pending_reclaim entry for \ + at-least-once retry by the pending-reclaim worker" + ); + Ok(()) } diff --git a/nodedb/src/control/catalog_entry/post_apply/async_dispatch/dispatcher.rs b/nodedb/src/control/catalog_entry/post_apply/async_dispatch/dispatcher.rs index 5d1f511b3..effa2ba92 100644 --- a/nodedb/src/control/catalog_entry/post_apply/async_dispatch/dispatcher.rs +++ b/nodedb/src/control/catalog_entry/post_apply/async_dispatch/dispatcher.rs @@ -18,9 +18,14 @@ //! synchronously on the calling tokio worker thread. The raft tick loop always //! runs on a tokio worker thread, so `block_in_place` is valid here. //! -//! All other variants (purge, MV delete, etc.) are fire-and-forget and are -//! spawned as background tasks — their correctness does not depend on -//! completing before the watcher bumps. +//! Collection purge and materialized-view deletion have the same ordering +//! requirement: all local Data Plane cores must reclaim the old incarnation +//! before the applied-index watcher advances, because a same-name re-CREATE may +//! immediately follow. Reclaim failure is fatal to the applying node; the +//! durable pending-reclaim record is drained on restart before stale state can +//! be served. +//! +//! Variants without a read-after-apply dependency remain fire-and-forget. use std::sync::Arc; @@ -93,19 +98,39 @@ pub fn spawn_post_apply_async_side_effects( tenant_id, name, } => { - tokio::spawn(async move { - collection::purge_async(database_id, tenant_id, name, raft_index, shared).await; + let result = tokio::task::block_in_place(|| { + tokio::runtime::Handle::current().block_on(async move { + collection::reclaim_collection_storage( + &shared, + database_id, + tenant_id, + &name, + raft_index, + false, + ) + .await + }) }); + if let Err(error) = result { + panic!("collection post-apply reclaim failed: {error}"); + } } - // `DeleteMaterializedView` now has async follow-up: every - // node dispatches `MetaOp::UnregisterMaterializedView` to its - // local Data Plane so the MV's columnar segment files + - // per-core in-memory state get reclaimed on followers, not - // just on the leader. Runs on every node; idempotent. + // SYNCHRONOUS: every node must clear the view target's per-core state + // before its applied-index watcher advances. Otherwise a same-name + // re-CREATE can observe cached aggregates from the dropped target. + // A failure is fatal: the metadata deletion is already committed, so + // continuing would serve an inconsistent catalog/Data Plane pair; + // restart safely reconstructs the in-memory cache from empty state. CatalogEntry::DeleteMaterializedView { tenant_id, name } => { - tokio::spawn(async move { - super::materialized_view::delete_async(tenant_id, name, shared).await; + let result = tokio::task::block_in_place(|| { + tokio::runtime::Handle::current().block_on(async move { + super::materialized_view::delete_async(tenant_id, name, raft_index, shared) + .await + }) }); + if let Err(error) = result { + panic!("materialized-view post-apply reclaim failed: {error}"); + } } // `PutContinuousAggregate` dispatches register to every core on // this node so the local `continuous_agg_mgr` picks up the new diff --git a/nodedb/src/control/catalog_entry/post_apply/async_dispatch/materialized_view.rs b/nodedb/src/control/catalog_entry/post_apply/async_dispatch/materialized_view.rs index 63812bcea..179a50f01 100644 --- a/nodedb/src/control/catalog_entry/post_apply/async_dispatch/materialized_view.rs +++ b/nodedb/src/control/catalog_entry/post_apply/async_dispatch/materialized_view.rs @@ -1,88 +1,30 @@ // SPDX-License-Identifier: BUSL-1.1 -//! Async post-apply for `CatalogEntry::DeleteMaterializedView`. +//! Synchronous post-apply reclaim for `CatalogEntry::DeleteMaterializedView`. //! -//! Runs on **every node** so each follower reclaims its local copy -//! of the MV's columnar segment files + in-memory refresh state — -//! mirroring the `UnregisterCollection` pattern one level up. +//! The replicated catalog entry removes both the materialized-view definition +//! and its implementation-owned target collection. Every node must then +//! reclaim that target through the same collection-wide Data Plane path before +//! advancing its metadata applied index. This prevents a same-name re-CREATE +//! from observing predecessor rows, indexes, or derived cache entries. use std::sync::Arc; -use tracing::debug; - -use crate::bridge::envelope::{PhysicalPlan, Priority, Request, Status}; use crate::control::state::SharedState; -use crate::types::{DatabaseId, ReadConsistency, TenantId, TraceId, VShardId}; -use nodedb_physical::physical_plan::MetaOp; - -/// Dispatch `MetaOp::UnregisterMaterializedView` to every core on -/// this node. Fire-and-forget: any core that fails or times out -/// logs at debug — reclaim is idempotent and the next MV drop on -/// the same `(tenant, name)` picks up where this one left off. -pub async fn delete_async(tenant_id: u64, name: String, shared: Arc) { - let num_cores = { - let d = shared.dispatcher.lock().unwrap_or_else(|p| p.into_inner()); - d.num_cores() - }; - let timeout = std::time::Duration::from_secs(30); - let mut receivers = Vec::with_capacity(num_cores); - - { - let mut d = shared.dispatcher.lock().unwrap_or_else(|p| p.into_inner()); - for core_id in 0..num_cores { - let request_id = shared.next_request_id(); - let request = Request { - request_id, - tenant_id: TenantId::new(tenant_id), - database_id: DatabaseId::DEFAULT, - vshard_id: VShardId::new(core_id as u32), - plan: PhysicalPlan::Meta(MetaOp::UnregisterMaterializedView { - tenant_id, - name: name.clone(), - }), - deadline: std::time::Instant::now() + timeout, - priority: Priority::Background, - trace_id: TraceId::generate(), - consistency: ReadConsistency::Eventual, - idempotency_key: None, - event_source: crate::event::EventSource::User, - user_roles: Vec::new(), - user_id: None, - statement_digest: None, - txn_id: None, - wal_lsn: None, - resolved_now_ms: None, - admission: crate::bridge::envelope::Admission::Exempt( - crate::bridge::envelope::ExemptReason::AlreadyOrdered, - ), - }; - let rx = shared.tracker.register(request_id); - if d.dispatch_to_core(core_id, request).is_err() { - shared.tracker.cancel(&request_id); - continue; - } - receivers.push((core_id, rx)); - } - } - for (core_id, mut rx) in receivers { - match tokio::time::timeout(timeout, async { rx.recv().await.ok_or(()) }).await { - Ok(Ok(resp)) if resp.status == Status::Ok => { - debug!( - tenant_id, - mv = %name, - core_id, - "materialized view reclaim ack" - ); - } - _ => { - debug!( - tenant_id, - mv = %name, - core_id, - "materialized view reclaim: core did not ack (will retry idempotently on next drop)" - ); - } - } - } +/// Reclaim the dropped view's target collection on this node. +/// +/// `reclaim_collection_storage` provides the durable pending-reclaim fallback +/// and dispatches `UnregisterCollection` to every local Data Plane core. The +/// caller treats an error as fatal because the catalog deletion is already +/// committed; serving through an incomplete reclaim would violate object +/// incarnation isolation. +pub async fn delete_async( + tenant_id: u64, + name: String, + purge_lsn: u64, + shared: Arc, +) -> crate::Result<()> { + super::collection::reclaim_collection_storage(&shared, 0, tenant_id, &name, purge_lsn, false) + .await } diff --git a/nodedb/src/control/catalog_entry/post_apply/collection.rs b/nodedb/src/control/catalog_entry/post_apply/collection.rs index bd28214e1..42bbb4162 100644 --- a/nodedb/src/control/catalog_entry/post_apply/collection.rs +++ b/nodedb/src/control/catalog_entry/post_apply/collection.rs @@ -61,8 +61,8 @@ pub async fn put_async(stored: StoredCollection, shared: Arc) { /// in-memory owner entry + any permission-cache entries keyed on /// the purged collection. The primary `StoredCollection` redb row /// is already gone at this point (removed by `apply/collection.rs::purge`). -/// The Data Plane `UnregisterCollection` dispatch is the async half -/// and lives in `async_dispatch/collection.rs::purge_async`. +/// The result-checked Data Plane `UnregisterCollection` barrier lives in +/// `async_dispatch/collection.rs::reclaim_collection_storage`. pub fn purge_sync(database_id: u64, tenant_id: u64, name: String, shared: Arc) { let owner_removed = shared .permissions diff --git a/nodedb/src/control/server/shared/ddl/neutral/collection/create/build.rs b/nodedb/src/control/server/shared/ddl/neutral/collection/create/build.rs index 807b0baf4..e3b951871 100644 --- a/nodedb/src/control/server/shared/ddl/neutral/collection/create/build.rs +++ b/nodedb/src/control/server/shared/ddl/neutral/collection/create/build.rs @@ -124,8 +124,35 @@ pub async fn build_and_persist( let tenant_id = identity.tenant_id; - // Check if the object already exists. + // Metadata Raft serializes clustered DDL. Without it, hold an exclusive + // per-name lifecycle guard across validation, any predecessor reclaim, + // catalog creation, and Data Plane registration. + let mut local_lifecycle = if state.metadata_raft.get().is_none() { + Some( + state + .quiesce + .acquire_lifecycle(database_id.as_u64(), tenant_id.as_u64(), name) + .await, + ) + } else { + None + }; + + // A materialized-view definition durably owns its same-name target even if + // a crash occurred between definition and target registration. let catalog = state.credentials.catalog(); + if catalog + .get_materialized_view(tenant_id.as_u64(), name) + .map_err(|error| err("XX000", error.to_string()))? + .is_some() + { + return Err(err( + "42P07", + format!("materialized view '{name}' already owns this collection name"), + )); + } + + // Check if the object already exists. if let Ok(Some(existing)) = catalog.get_collection(database_id, tenant_id.as_u64(), name) { if existing.is_active { return Err(err( @@ -152,14 +179,22 @@ pub async fn build_and_persist( // catalog row, ABORT the CREATE rather than build a new // collection over un-purged data (which would resurrect the // stale rows). Surface as an internal error to the client. - crate::control::server::shared::ddl::neutral::collection::purge::hard_purge_collection( - state, - tenant_id.as_u64(), - name, - purge_lsn, - ) - .await - .map_err(|e| err("XX000", e.to_string()))?; + let purge_result = + crate::control::server::shared::ddl::neutral::collection::purge::hard_purge_collection( + state, + database_id.as_u64(), + tenant_id.as_u64(), + name, + purge_lsn, + local_lifecycle.is_some(), + ) + .await; + if let Err(error) = purge_result { + if let Some(guard) = local_lifecycle.take() { + guard.disarm(); + } + return Err(err("XX000", error.to_string())); + } } let now = std::time::SystemTime::now() diff --git a/nodedb/src/control/server/shared/ddl/neutral/collection/drop.rs b/nodedb/src/control/server/shared/ddl/neutral/collection/drop.rs index bfbbe0c0d..ced131098 100644 --- a/nodedb/src/control/server/shared/ddl/neutral/collection/drop.rs +++ b/nodedb/src/control/server/shared/ddl/neutral/collection/drop.rs @@ -183,6 +183,18 @@ pub fn drop_collection( // joins them on the absent-name branch. { let catalog = state.credentials.catalog(); + if catalog + .get_materialized_view(tenant_id.as_u64(), name) + .map_err(|error| err("XX000", error.to_string()))? + .is_some() + { + return Err(err( + "2BP01", + format!( + "collection '{name}' is owned by a materialized view; drop the materialized view" + ), + )); + } match catalog.get_collection(database_id, tenant_id.as_u64(), name) { Ok(Some(coll)) if coll.is_active => {} Ok(Some(_)) if purge => {} @@ -240,21 +252,43 @@ pub fn drop_collection( name: name.to_string(), } }; + // Without metadata Raft, acquire the per-name lifecycle guard before the + // catalog mutation and hold it through local reclaim. + let mut local_lifecycle = if state.metadata_raft.get().is_none() { + Some( + state + .quiesce + .try_acquire_lifecycle(database_id.as_u64(), tenant_id.as_u64(), name) + .ok_or_else(|| err("55006", format!("collection '{name}' lifecycle is busy")))?, + ) + } else { + None + }; let log_index = crate::control::metadata_proposer::propose_catalog_entry(state, &entry) - .map_err(|e| err("XX000", e.to_string()))?; + .map_err(|error| err("XX000", error.to_string()))?; if log_index == 0 { let catalog = state.credentials.catalog(); - // Single-node / no-cluster fallback: apply the catalog mutation - // directly, matching what the applier would have done on a - // clustered deployment. if purge { - crate::control::catalog_entry::apply::collection::purge( - database_id.as_u64(), - tenant_id.as_u64(), - name, - catalog, - ) - .map_err(|e| err("XX000", e.to_string()))?; + let purge_lsn = state.wal.next_lsn().as_u64(); + let purge_result = tokio::task::block_in_place(|| { + tokio::runtime::Handle::current().block_on(async { + crate::control::server::shared::ddl::neutral::collection::purge::hard_purge_collection( + state, + database_id.as_u64(), + tenant_id.as_u64(), + name, + purge_lsn, + local_lifecycle.is_some(), + ) + .await + }) + }); + if let Err(error) = purge_result { + if let Some(guard) = local_lifecycle.take() { + guard.disarm(); + } + panic!("local collection reclaim failed: {error}"); + } state .permissions .install_replicated_remove_owner_in_database( diff --git a/nodedb/src/control/server/shared/ddl/neutral/collection/purge/dispatch.rs b/nodedb/src/control/server/shared/ddl/neutral/collection/purge/dispatch.rs index b395678bb..0dd8617bb 100644 --- a/nodedb/src/control/server/shared/ddl/neutral/collection/purge/dispatch.rs +++ b/nodedb/src/control/server/shared/ddl/neutral/collection/purge/dispatch.rs @@ -1,35 +1,29 @@ // SPDX-License-Identifier: BUSL-1.1 -//! Dispatch `MetaOp::UnregisterCollection` to the local Data Plane. +//! Result-checked `UnregisterCollection` fan-out to every local Data Plane core. //! -//! Called from `catalog_entry::post_apply::async_dispatch::collection::purge_async` -//! on **every node** (leader and followers) so each node's Data -//! Plane reclaims its own local L1/L2 storage for the purged -//! collection symmetrically with the metadata row removal. +//! Collection-scoped state is present on multiple cores (including aggregate +//! caches and engine-local segments), so purge completion means every core has +//! acknowledged reclaim. All requests are dispatched before responses are +//! awaited; any send, timeout, channel, or handler failure aborts the barrier. -use crate::bridge::envelope::{PhysicalPlan, Status}; -use crate::control::state::SharedState; -use crate::types::{DatabaseId, TenantId, TraceId, VShardId}; +use std::time::{Duration, Instant}; + +use futures::future::join_all; use nodedb_physical::physical_plan::MetaOp; -/// Dispatch `MetaOp::UnregisterCollection { tenant_id, name, purge_lsn }` -/// to this node's Data Plane so `clear_collection_all_engines` reclaims -/// every engine's redb + versioned storage for the purged collection. -/// -/// **Result-checked.** This is the correctness-critical half of a DROP: -/// the catalog row has already been removed at apply, so if the engine -/// purge does not actually succeed, engine rows survive behind a gone -/// catalog row (the resurrection hole on re-CREATE). The Data-Plane -/// handler is fail-closed and returns a `Status::Error` response on a -/// failed engine purge, but the shared `dispatch_to_data_plane` hands -/// that back as `Ok(response)` regardless of `response.status` (its -/// contract is used broadly and must not change). We therefore inspect -/// `response.status` HERE and turn an error response — or a transport -/// error, or a deadline — into a propagated `Err`. The caller records a -/// durable pending-reclaim entry so the purge is retried at-least-once -/// rather than lost to a warn log. +use crate::bridge::envelope::{PhysicalPlan, Priority, Request, Status}; +use crate::control::state::SharedState; +use crate::types::{DatabaseId, ReadConsistency, TenantId, TraceId, VShardId}; + +/// Dispatch `MetaOp::UnregisterCollection` to every local core and require an +/// `Ok` acknowledgement from each one. /// -/// Idempotent: safe to re-dispatch after a partial or failed attempt. +/// The catalog row has already been removed when this runs. Returning success +/// after only a subset of cores reclaimed would allow a same-name re-CREATE to +/// observe predecessor state, so partial success is always an error. The caller +/// records a durable pending-reclaim entry and the applied-index barrier fails +/// closed. pub async fn dispatch_unregister_collection( state: &SharedState, database_id: u64, @@ -39,36 +33,86 @@ pub async fn dispatch_unregister_collection( ) -> crate::Result<()> { let tenant = TenantId::new(tenant_id); let database = DatabaseId::new(database_id); - let vshard = VShardId::from_collection_in_database(database, name); - let plan = PhysicalPlan::Meta(MetaOp::UnregisterCollection { - tenant_id, - name: name.to_string(), - purge_lsn, - }); - - let response = crate::control::server::dispatch_utils::dispatch_to_data_plane( - state, - tenant, - database, - vshard, - plan, - TraceId::ZERO, - ) - .await?; + let timeout = Duration::from_secs(state.tuning.network.default_deadline_secs); + let num_cores = state + .dispatcher + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + .num_cores(); + let mut receivers = Vec::with_capacity(num_cores); - if response.status == Status::Error { - let detail = match response.error_code { - Some(code) => format!("{code:?}"), - None => "engine purge returned Status::Error with no error_code".to_string(), - }; - return Err(crate::Error::Storage { - engine: "collection-purge".into(), - detail: format!( - "UnregisterCollection for tenant {tenant_id} collection '{name}' \ - failed on the local Data Plane: {detail}" - ), - }); + { + let mut dispatcher = state + .dispatcher + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()); + for core_id in 0..num_cores { + let request_id = state.next_request_id(); + let request = Request { + request_id, + tenant_id: tenant, + database_id: database, + vshard_id: VShardId::new(core_id as u32), + plan: PhysicalPlan::Meta(MetaOp::UnregisterCollection { + tenant_id, + name: name.to_string(), + purge_lsn, + }), + deadline: Instant::now() + timeout, + priority: Priority::Background, + trace_id: TraceId::ZERO, + consistency: ReadConsistency::Strong, + idempotency_key: None, + event_source: crate::event::EventSource::User, + user_roles: Vec::new(), + user_id: None, + statement_digest: None, + txn_id: None, + wal_lsn: None, + resolved_now_ms: None, + admission: crate::bridge::envelope::Admission::Exempt( + crate::bridge::envelope::ExemptReason::AlreadyOrdered, + ), + }; + let receiver = state.tracker.register(request_id); + if let Err(error) = dispatcher.dispatch_to_core(core_id, request) { + state.tracker.cancel(&request_id); + for (registered_id, _, _) in &receivers { + state.tracker.cancel(registered_id); + } + return Err(error); + } + receivers.push((request_id, core_id, receiver)); + } } + let responses = join_all(receivers.into_iter().map( + |(_request_id, core_id, mut receiver)| async move { + let response = tokio::time::timeout(timeout, receiver.recv()) + .await + .map_err(|_| crate::Error::Dispatch { + detail: format!("collection reclaim timed out on core {core_id}"), + })? + .ok_or_else(|| crate::Error::Dispatch { + detail: format!("collection reclaim channel closed on core {core_id}"), + })?; + if response.status != Status::Ok { + return Err(crate::Error::Storage { + engine: "collection-purge".into(), + detail: format!( + "UnregisterCollection for tenant {tenant_id} collection '{name}' \ + failed on core {core_id}: {:?}", + response.error_code + ), + }); + } + Ok(()) + }, + )) + .await; + + for response in responses { + response?; + } Ok(()) } diff --git a/nodedb/src/control/server/shared/ddl/neutral/collection/purge/hard.rs b/nodedb/src/control/server/shared/ddl/neutral/collection/purge/hard.rs index 20cec46b9..3e1be3198 100644 --- a/nodedb/src/control/server/shared/ddl/neutral/collection/purge/hard.rs +++ b/nodedb/src/control/server/shared/ddl/neutral/collection/purge/hard.rs @@ -14,8 +14,8 @@ //! This composes the two halves the `DROP COLLECTION ... PURGE` //! applier runs on every node — the catalog-row removal //! (`apply::collection::purge`) and the storage reclaim -//! (`post_apply::async_dispatch::collection::reclaim_collection_storage`, -//! the borrowed core of `purge_async`) — into one awaitable the +//! (`post_apply::async_dispatch::collection::reclaim_collection_storage`) — +//! into one awaitable the //! Control Plane can drive inline before it proceeds to persist the //! new collection. No engine is special-cased: the reclaim dispatch //! covers every engine exactly as `DROP ... PURGE` does. @@ -48,9 +48,11 @@ use crate::control::state::SharedState; /// Lite broadcast) still log-and-continue. pub(crate) async fn hard_purge_collection( state: &SharedState, + database_id: u64, tenant_id: u64, name: &str, purge_lsn: u64, + drain_already_held: bool, ) -> Result<(), NodeDbError> { // 1. Remove the catalog metadata row (primary StoredCollection, // owner row, surrogate map) — the synchronous half of the @@ -58,7 +60,12 @@ pub(crate) async fn hard_purge_collection( // survives, the new collection must NOT register over it. { let catalog = state.credentials.catalog(); - crate::control::catalog_entry::apply::collection::purge(0, tenant_id, name, catalog)?; + crate::control::catalog_entry::apply::collection::prepare_purge( + database_id, + tenant_id, + name, + catalog, + )?; } // 2. Reclaim engine-local storage on the Data Plane (WAL tombstone, @@ -66,7 +73,12 @@ pub(crate) async fn hard_purge_collection( // dispatch, Lite `CollectionPurged` broadcast) — the async half // of the `PurgeCollection` post-apply, shared verbatim. crate::control::catalog_entry::post_apply::reclaim_collection_storage( - state, 0, tenant_id, name, purge_lsn, + state, + database_id, + tenant_id, + name, + purge_lsn, + drain_already_held, ) .await?; diff --git a/nodedb/src/control/server/shared/ddl/neutral/collection/undrop.rs b/nodedb/src/control/server/shared/ddl/neutral/collection/undrop.rs index ce8478dcb..e1e59c1cf 100644 --- a/nodedb/src/control/server/shared/ddl/neutral/collection/undrop.rs +++ b/nodedb/src/control/server/shared/ddl/neutral/collection/undrop.rs @@ -44,6 +44,24 @@ pub fn undrop_collection( let name = name_lower.as_str(); let tenant_id = identity.tenant_id; + // Metadata Raft serializes clustered lifecycle mutations. The local + // fallback must acquire the same exclusive name guard as CREATE/DROP + // before reading the preserved descriptor, otherwise it can restore an + // incarnation that a concurrent purge has already superseded. + let _local_lifecycle = if state.metadata_raft.get().is_none() { + Some( + state + .quiesce + .try_acquire_lifecycle(database_id.as_u64(), tenant_id.as_u64(), name) + .ok_or_else(|| DdlError { + sqlstate: "55006".to_string(), + message: format!("collection '{name}' lifecycle is busy"), + })?, + ) + } else { + None + }; + let catalog = state.credentials.catalog(); // Look up the soft-deleted record. Three distinct failures: diff --git a/nodedb/src/control/server/shared/ddl/neutral/materialized_view/create.rs b/nodedb/src/control/server/shared/ddl/neutral/materialized_view/create.rs index 077c46643..a044ed9fb 100644 --- a/nodedb/src/control/server/shared/ddl/neutral/materialized_view/create.rs +++ b/nodedb/src/control/server/shared/ddl/neutral/materialized_view/create.rs @@ -51,6 +51,20 @@ pub async fn create_materialized_view( return create_streaming_mv(state, identity, &name, &query_sql).await; } + // Metadata Raft serializes clustered DDL. Without it, hold an exclusive + // name lifecycle guard through definition+target creation and Data Plane + // registration so DROP or another CREATE cannot interleave. + let _local_lifecycle = if state.metadata_raft.get().is_none() { + Some( + state + .quiesce + .acquire_lifecycle(0, tenant_id.as_u64(), &name) + .await, + ) + } else { + None + }; + // Validate source collection exists. { let catalog = state.credentials.catalog(); @@ -70,6 +84,10 @@ pub async fn create_materialized_view( format!("materialized view '{name}' already exists"), )); } + if let Ok(Some(_)) = catalog.get_collection(DatabaseId::DEFAULT, tenant_id.as_u64(), &name) + { + return Err(err("42P07", format!("collection '{name}' already exists"))); + } } let now = std::time::SystemTime::now() @@ -98,68 +116,58 @@ pub async fn create_materialized_view( crate::control::catalog_entry::CatalogEntry::PutMaterializedView(Box::new(view.clone())); propose_and_apply(state, &entry)?; - // Create the view's target collection so REFRESH can insert into it - // and clients can SELECT from it like any other collection. Skipped - // when a collection of the same name already exists (idempotent). - let target_exists = { - let catalog = state.credentials.catalog(); - matches!( - catalog.get_collection(DatabaseId::DEFAULT, tenant_id.as_u64(), &name), - Ok(Some(c)) if c.is_active - ) + // Create the implementation-owned target collection so REFRESH can insert + // into it and clients can SELECT from it. The pre-check rejects every + // same-name collection; DROP may therefore purge this target without ever + // deleting a user-owned collection. + let target = StoredCollection { + tenant_id: tenant_id.as_u64(), + name: name.clone(), + owner: identity.username.clone(), + created_at: now, + descriptor_version: 0, + constraint_version: 0, + modification_hlc: nodedb_types::Hlc::ZERO, + fields: Vec::new(), + field_defs: Vec::new(), + event_defs: Vec::new(), + collection_type: nodedb_types::CollectionType::document(), + timeseries_config: None, + conflict_policy: None, + is_active: true, + append_only: false, + hash_chain: false, + balanced: None, + last_chain_hash: None, + period_lock: None, + retention_period: None, + legal_holds: Vec::new(), + state_constraints: Vec::new(), + transition_checks: Vec::new(), + type_guards: Vec::new(), + check_constraints: Vec::new(), + materialized_sums: Vec::new(), + lvc_enabled: false, + bitemporal: false, + crdt: false, + permission_tree_def: None, + indexes: Vec::new(), + size_bytes_estimate: 0, + primary: nodedb_types::PrimaryEngine::Document, + vector_primary: None, + partition_strategy: nodedb_types::PartitionStrategy::CollectionHomed, + database_id: nodedb_types::DatabaseId::DEFAULT, + cloned_from: None, + clone_status: nodedb_types::CloneStatus::default(), + has_implicit_edges: false, + declared_primary_key: None, }; - if !target_exists { - let target = StoredCollection { - tenant_id: tenant_id.as_u64(), - name: name.clone(), - owner: identity.username.clone(), - created_at: now, - descriptor_version: 0, - constraint_version: 0, - modification_hlc: nodedb_types::Hlc::ZERO, - fields: Vec::new(), - field_defs: Vec::new(), - event_defs: Vec::new(), - collection_type: nodedb_types::CollectionType::document(), - timeseries_config: None, - conflict_policy: None, - is_active: true, - append_only: false, - hash_chain: false, - balanced: None, - last_chain_hash: None, - period_lock: None, - retention_period: None, - legal_holds: Vec::new(), - state_constraints: Vec::new(), - transition_checks: Vec::new(), - type_guards: Vec::new(), - check_constraints: Vec::new(), - materialized_sums: Vec::new(), - lvc_enabled: false, - bitemporal: false, - crdt: false, - permission_tree_def: None, - indexes: Vec::new(), - size_bytes_estimate: 0, - primary: nodedb_types::PrimaryEngine::Document, - vector_primary: None, - partition_strategy: nodedb_types::PartitionStrategy::CollectionHomed, - database_id: nodedb_types::DatabaseId::DEFAULT, - cloned_from: None, - clone_status: nodedb_types::CloneStatus::default(), - has_implicit_edges: false, - declared_primary_key: None, - }; - let coll_entry = - crate::control::catalog_entry::CatalogEntry::PutCollection(Box::new(target.clone())); - propose_and_apply(state, &coll_entry)?; - // Register the target with this node's Data Plane so writes - // encode correctly and scans can find the collection. - super::super::collection::dispatch_register_from_stored(state, &target) - .await - .map_err(|e| err("XX000", e.to_string()))?; - } + let coll_entry = + crate::control::catalog_entry::CatalogEntry::PutCollection(Box::new(target.clone())); + propose_and_apply(state, &coll_entry)?; + super::super::collection::dispatch_register_from_stored(state, &target) + .await + .map_err(|e| err("XX000", e.to_string()))?; tracing::info!( view = name, diff --git a/nodedb/src/control/server/shared/ddl/neutral/materialized_view/drop.rs b/nodedb/src/control/server/shared/ddl/neutral/materialized_view/drop.rs index e7c8dcdb4..f75e50831 100644 --- a/nodedb/src/control/server/shared/ddl/neutral/materialized_view/drop.rs +++ b/nodedb/src/control/server/shared/ddl/neutral/materialized_view/drop.rs @@ -3,15 +3,10 @@ //! Protocol-neutral `DROP MATERIALIZED VIEW [IF EXISTS]` handler. //! //! Ported from the pgwire `ddl::materialized_view::drop` handler. The DIRECT -//! catalog path (`propose_catalog_entry` for the `DeleteMaterializedView` entry -//! with a manual `catalog.delete_materialized_view` fallback on the `log_index -//! == 0` bypass branch, then a `DeactivateCollection` propose for the view's -//! target collection), the token-based name / IF EXISTS extraction, and the -//! pre-check existence gate are preserved verbatim; only the result construction -//! changed from pgwire `Response` / `PgWireError` to the protocol-neutral -//! [`DdlResult`] / [`DdlError`]. - -use nodedb_types::DatabaseId; +//! catalog path (`propose_catalog_entry` for the compound +//! `DeleteMaterializedView` definition+target deletion, with synchronous local +//! apply/reclaim when metadata Raft is absent), the token-based name / IF EXISTS +//! extraction, and the pre-check existence gate are shared by every protocol. use crate::control::security::identity::AuthenticatedIdentity; use crate::control::state::SharedState; @@ -108,31 +103,49 @@ pub fn drop_materialized_view( tenant_id: tenant_id.as_u64(), name: name.clone(), }; + let mut local_lifecycle = if state.metadata_raft.get().is_none() { + Some( + state + .quiesce + .try_acquire_lifecycle(0, tenant_id.as_u64(), &name) + .ok_or_else(|| { + err( + "55006", + format!("materialized view '{name}' lifecycle is busy"), + ) + })?, + ) + } else { + None + }; let log_index = crate::control::metadata_proposer::propose_catalog_entry(state, &entry) - .map_err(|e| err("XX000", format!("metadata propose: {e}")))?; + .map_err(|error| err("XX000", format!("metadata propose: {error}")))?; if log_index == 0 { - let catalog = state.credentials.catalog(); - catalog - .delete_materialized_view(tenant_id.as_u64(), &name) - .map_err(|e| err("XX000", e.to_string()))?; - } - - // Also drop the view's target collection created by CREATE MATERIALIZED VIEW. - // The target lives as a normal collection to support INSERT...SELECT refresh - // and SELECT reads; leaving it behind would leak storage and shadow any - // later CREATE COLLECTION with the same name. - let catalog = state.credentials.catalog(); - if matches!( - catalog.get_collection(DatabaseId::DEFAULT, tenant_id.as_u64(), &name), - Ok(Some(_)) - ) { - let coll_entry = crate::control::catalog_entry::CatalogEntry::DeactivateCollection { - database_id: DatabaseId::DEFAULT.as_u64(), - tenant_id: tenant_id.as_u64(), - name: name.clone(), - }; - let _ = crate::control::metadata_proposer::propose_catalog_entry(state, &coll_entry) - .map_err(|e| err("XX000", format!("metadata propose: {e}")))?; + // No metadata Raft is active, so apply the same compound catalog + // deletion locally and synchronously reclaim the implementation-owned + // target collection. A reclaim failure after catalog deletion is + // fatal: continuing would permit a same-name CREATE over stale rows. + crate::control::catalog_entry::apply::apply_to(&entry, state.credentials.catalog()); + let purge_lsn = state.wal.next_lsn().as_u64(); + let purge_result = tokio::task::block_in_place(|| { + tokio::runtime::Handle::current().block_on(async { + crate::control::server::shared::ddl::neutral::collection::purge::hard_purge_collection( + state, + 0, + tenant_id.as_u64(), + &name, + purge_lsn, + local_lifecycle.is_some(), + ) + .await + }) + }); + if let Err(error) = purge_result { + if let Some(guard) = local_lifecycle.take() { + guard.disarm(); + } + panic!("local materialized-view target reclaim failed: {error}"); + } } tracing::info!(view = name, "materialized view dropped"); diff --git a/nodedb/src/control/server/shared/ddl/neutral/timeseries/create.rs b/nodedb/src/control/server/shared/ddl/neutral/timeseries/create.rs index 312f1bf73..1ea962aec 100644 --- a/nodedb/src/control/server/shared/ddl/neutral/timeseries/create.rs +++ b/nodedb/src/control/server/shared/ddl/neutral/timeseries/create.rs @@ -2,8 +2,6 @@ //! `CREATE TIMESERIES [WITH (key = 'value', ...)]` -use nodedb_types::DatabaseId; - use crate::control::security::catalog::StoredCollection; use crate::control::security::identity::AuthenticatedIdentity; use crate::control::state::SharedState; @@ -28,16 +26,19 @@ pub fn create_timeseries( let name = parts[2].to_lowercase(); let tenant_id = identity.tenant_id; - if let Ok(Some(_)) = - state - .credentials - .catalog() - .get_collection(DatabaseId::DEFAULT, tenant_id.as_u64(), &name) + match state + .credentials + .catalog() + .get_collection(database_id, tenant_id.as_u64(), &name) { - return Err(ddl_err( - "42P07", - format!("collection '{name}' already exists"), - )); + Ok(Some(_)) => { + return Err(ddl_err( + "42P07", + format!("collection '{name}' already exists"), + )); + } + Ok(None) => {} + Err(error) => return Err(ddl_err("XX000", error.to_string())), } let config_json = parse_with_clause(parts); diff --git a/nodedb/src/data/executor/handlers/aggregate/invalidate.rs b/nodedb/src/data/executor/handlers/aggregate/invalidate.rs index d23e6b50d..a289eb967 100644 --- a/nodedb/src/data/executor/handlers/aggregate/invalidate.rs +++ b/nodedb/src/data/executor/handlers/aggregate/invalidate.rs @@ -1,24 +1,23 @@ // SPDX-License-Identifier: BUSL-1.1 -//! Shared aggregate-cache invalidation sweep, used by every write path that -//! changes row counts for a collection (PUT, point DELETE, bulk DELETE, -//! TRUNCATE, columnar insert) and therefore must not leave a stale -//! `COUNT(*)` / `GROUP BY` result cached. +//! Shared derived-result-cache invalidation sweep, used by every write and +//! object-lifecycle path that can make cached collection results stale. The +//! cache stores both aggregate and facet payloads. //! -//! Keyed on `(tenant, "{collection}\0...")` — every cached entry for a given -//! collection shares that null-byte-terminated prefix (see -//! `aggregate_cache_key` in `cache_key.rs`), so a prefix sweep evicts exactly -//! that collection's entries without touching other collections' cached -//! aggregates for the same tenant. +//! Keyed on `(database, tenant, "{collection}\0...")` — every cached entry for +//! a given collection shares that null-byte-terminated prefix (see +//! `aggregate_cache_key` in `cache_key.rs` and `facet_cache_key` in +//! `facet.rs`). Routing all invalidation through this method keeps the encoded +//! key invariant out of callers and evicts exactly one collection's entries. use crate::data::executor::core_loop::CoreLoop; impl CoreLoop { - /// Evict every cached aggregate result for `(tid, collection)`. + /// Evict every cached aggregate or facet result for + /// `(database_id, tid, collection)`. /// - /// Callers gate this on "did this operation actually change the - /// collection's rows" (e.g. `prior.is_some()`, `affected > 0`) — the - /// sweep itself is unconditional once called. + /// Write callers gate this on whether rows changed. Lifecycle callers run + /// it unconditionally when an object is unregistered. pub(in crate::data::executor) fn invalidate_aggregate_cache_for_collection( &mut self, database_id: u64, diff --git a/nodedb/src/data/executor/handlers/reclaim/mod.rs b/nodedb/src/data/executor/handlers/reclaim/mod.rs index b124ac9b4..7db4737d0 100644 --- a/nodedb/src/data/executor/handlers/reclaim/mod.rs +++ b/nodedb/src/data/executor/handlers/reclaim/mod.rs @@ -15,14 +15,34 @@ //! engines that write per-collection checkpoint or partition files //! under `{data_dir}/...`. +use std::path::PathBuf; + +use thiserror::Error; + pub mod sparse_vector; pub mod spatial; pub mod timeseries; pub mod vector; -/// Summary of a single engine's reclaim pass. Byte counts are -/// best-effort — missing files count as 0, IO errors are warn-logged -/// and skipped (idempotent). +/// A persistent L1 surface could not be fully reclaimed. Callers must not +/// release the collection lifecycle barrier after this error. +#[derive(Debug, Error)] +pub enum ReclaimError { + #[error("{operation} failed for '{}': {source}", path.display())] + Io { + operation: &'static str, + path: PathBuf, + #[source] + source: std::io::Error, + }, + #[error("sparse-vector manifest at '{}' is unreadable: {detail}", path.display())] + SparseManifest { path: PathBuf, detail: String }, +} + +pub type Result = std::result::Result; + +/// Summary of a single engine's reclaim pass. Missing files count as zero; +/// actual I/O failures are returned to the lifecycle barrier. #[derive(Debug, Default, Clone, Copy)] pub struct ReclaimStats { pub files_unlinked: u32, diff --git a/nodedb/src/data/executor/handlers/reclaim/sparse_vector.rs b/nodedb/src/data/executor/handlers/reclaim/sparse_vector.rs index 6323293e1..030015aaa 100644 --- a/nodedb/src/data/executor/handlers/reclaim/sparse_vector.rs +++ b/nodedb/src/data/executor/handlers/reclaim/sparse_vector.rs @@ -23,9 +23,9 @@ use std::path::Path; -use tracing::{debug, warn}; +use tracing::debug; -use super::ReclaimStats; +use super::{ReclaimError, ReclaimStats, Result}; use crate::data::executor::sparse_vector_checkpoint::{ read_sparse_vector_manifest_at, sparse_vector_checkpoint_prefix, sparse_vector_ckpt_gen_dir, }; @@ -38,30 +38,33 @@ pub fn reclaim_sparse_vector_checkpoints( database_id: u64, tenant_id: u64, collection: &str, -) -> ReclaimStats { +) -> Result { let root = data_dir.join("sparse-vector-ckpt"); - if !root.exists() { - return ReclaimStats::default(); - } + let cores = match std::fs::read_dir(&root) { + Ok(entries) => entries, + Err(source) if source.kind() == std::io::ErrorKind::NotFound => { + return Ok(ReclaimStats::default()); + } + Err(source) => { + return Err(ReclaimError::Io { + operation: "read sparse-vector checkpoint root", + path: root, + source, + }); + } + }; // Build the prefix via the shared encoder so it always matches the // filenames the write path produced. let prefix = sparse_vector_checkpoint_prefix(database_id, tenant_id, collection); - let cores = match std::fs::read_dir(&root) { - Ok(e) => e, - Err(e) => { - warn!( - dir = %root.display(), - error = %e, - "sparse-vector reclaim: failed to read ckpt root" - ); - return ReclaimStats::default(); - } - }; - let mut stats = ReclaimStats::default(); - for core_entry in cores.flatten() { + for core_entry in cores { + let core_entry = core_entry.map_err(|source| ReclaimError::Io { + operation: "read sparse-vector core entry", + path: root.clone(), + source, + })?; let core_dir = core_entry.path(); if !core_dir.is_dir() { continue; @@ -76,44 +79,37 @@ pub fn reclaim_sparse_vector_checkpoints( }; // No manifest means no reachable generation for this core, so there is // nothing a boot could restore and nothing to reclaim. A corrupt - // manifest is left for the boot path to fail loudly on; reclaim is - // best-effort disk cleanup, not a restore, so it just skips this core - // rather than aborting the whole reclaim pass. + // manifest is fail-closed: skipping it could release same-name CREATE + // while predecessor index files remain reachable. let manifest = match read_sparse_vector_manifest_at(&core_dir, core_id) { Ok(Some(m)) => m, Ok(None) => continue, - Err(e) => { - warn!( - dir = %core_dir.display(), - error = %e, - "sparse-vector reclaim: manifest unreadable; skipping this core" - ); - continue; + Err(error) => { + return Err(ReclaimError::SparseManifest { + path: core_dir.join("MANIFEST"), + detail: error.to_string(), + }); } }; let gen_dir = sparse_vector_ckpt_gen_dir(&core_dir, manifest.generation); - reclaim_generation(&gen_dir, &prefix, &mut stats); + reclaim_generation(&gen_dir, &prefix, &mut stats)?; } - stats + Ok(stats) } /// Unlink every file in `gen_dir` whose name carries `prefix`. -fn reclaim_generation(gen_dir: &Path, prefix: &str, stats: &mut ReclaimStats) { - let entries = match std::fs::read_dir(gen_dir) { - Ok(e) => e, - Err(e) => { - // A live manifest naming a missing generation dir is corruption, not - // routine absence, so it is worth a warning rather than a silent - // skip. - warn!( - dir = %gen_dir.display(), - error = %e, - "sparse-vector reclaim: failed to read live generation dir" - ); - return; - } - }; - for entry in entries.flatten() { +fn reclaim_generation(gen_dir: &Path, prefix: &str, stats: &mut ReclaimStats) -> Result<()> { + let entries = std::fs::read_dir(gen_dir).map_err(|source| ReclaimError::Io { + operation: "read sparse-vector live generation", + path: gen_dir.to_path_buf(), + source, + })?; + for entry in entries { + let entry = entry.map_err(|source| ReclaimError::Io { + operation: "read sparse-vector generation entry", + path: gen_dir.to_path_buf(), + source, + })?; let path = entry.path(); let Some(name) = path.file_name().and_then(|s| s.to_str()) else { continue; @@ -133,13 +129,17 @@ fn reclaim_generation(gen_dir: &Path, prefix: &str, stats: &mut ReclaimStats) { stats.bytes_freed = stats.bytes_freed.saturating_add(size); debug!(path = %path.display(), size, "sparse-vector reclaim: unlinked"); } - Err(e) => warn!( - path = %path.display(), - error = %e, - "sparse-vector reclaim: unlink failed" - ), + Err(source) if source.kind() == std::io::ErrorKind::NotFound => {} + Err(source) => { + return Err(ReclaimError::Io { + operation: "unlink sparse-vector checkpoint", + path, + source, + }); + } } } + Ok(()) } #[cfg(test)] @@ -192,7 +192,7 @@ mod tests { ], ); - let stats = reclaim_sparse_vector_checkpoints(tmp.path(), 0, 1, "docs"); + let stats = reclaim_sparse_vector_checkpoints(tmp.path(), 0, 1, "docs").unwrap(); assert_eq!( stats.files_unlinked, 3, "both of core 0's files and core 1's one file must be unlinked" @@ -217,12 +217,23 @@ mod tests { let tmp = TempDir::new().expect("tempdir"); publish(tmp.path(), 0, 0, &["0_1_docs%5Farchive_title.ckpt"]); - let stats = reclaim_sparse_vector_checkpoints(tmp.path(), 0, 1, "docs"); + let stats = reclaim_sparse_vector_checkpoints(tmp.path(), 0, 1, "docs").unwrap(); assert_eq!(stats.files_unlinked, 0); let gen_dir = sparse_vector_ckpt_gen_dir(&sparse_vector_ckpt_dir(tmp.path(), 0), 0); assert!(gen_dir.join("0_1_docs%5Farchive_title.ckpt").exists()); } + #[test] + fn corrupt_manifest_is_returned_to_lifecycle_barrier() { + let tmp = TempDir::new().expect("tempdir"); + let core_dir = sparse_vector_ckpt_dir(tmp.path(), 0); + std::fs::create_dir_all(&core_dir).expect("mkdir"); + std::fs::write(core_dir.join("MANIFEST"), b"not-a-manifest").expect("manifest"); + + let error = reclaim_sparse_vector_checkpoints(tmp.path(), 0, 1, "docs").unwrap_err(); + assert!(error.to_string().contains("manifest")); + } + /// Files under a SUPERSEDED generation are already unreachable: the manifest /// alone decides what a boot restores, so reclaim leaves them to the write /// path's own cleanup rather than walking directories nothing can read. @@ -234,7 +245,7 @@ mod tests { std::fs::create_dir_all(&stale).expect("mkdir"); std::fs::write(stale.join("0_1_docs_title.ckpt"), b"old").expect("write"); - let stats = reclaim_sparse_vector_checkpoints(tmp.path(), 0, 1, "docs"); + let stats = reclaim_sparse_vector_checkpoints(tmp.path(), 0, 1, "docs").unwrap(); assert_eq!(stats.files_unlinked, 1, "only the live generation's file"); assert!(stale.join("0_1_docs_title.ckpt").exists()); } @@ -244,7 +255,7 @@ mod tests { #[test] fn absent_root_is_a_noop() { let tmp = TempDir::new().expect("tempdir"); - let stats = reclaim_sparse_vector_checkpoints(tmp.path(), 0, 1, "docs"); + let stats = reclaim_sparse_vector_checkpoints(tmp.path(), 0, 1, "docs").unwrap(); assert_eq!(stats.files_unlinked, 0); } } diff --git a/nodedb/src/data/executor/handlers/reclaim/spatial.rs b/nodedb/src/data/executor/handlers/reclaim/spatial.rs index 2b986762e..e3bd4e19d 100644 --- a/nodedb/src/data/executor/handlers/reclaim/spatial.rs +++ b/nodedb/src/data/executor/handlers/reclaim/spatial.rs @@ -11,9 +11,9 @@ use std::path::Path; -use tracing::{debug, warn}; +use tracing::debug; -use super::ReclaimStats; +use super::{ReclaimError, ReclaimStats, Result}; use crate::data::executor::spatial_checkpoint::spatial_checkpoint_prefix; /// Unlink every spatial checkpoint + docmap file for @@ -23,29 +23,33 @@ pub fn reclaim_spatial_checkpoints( database_id: u64, tenant_id: u64, collection: &str, -) -> ReclaimStats { +) -> Result { let ckpt_dir = data_dir.join("spatial-ckpt"); - if !ckpt_dir.exists() { - return ReclaimStats::default(); - } + let entries = match std::fs::read_dir(&ckpt_dir) { + Ok(entries) => entries, + Err(source) if source.kind() == std::io::ErrorKind::NotFound => { + return Ok(ReclaimStats::default()); + } + Err(source) => { + return Err(ReclaimError::Io { + operation: "read spatial checkpoint directory", + path: ckpt_dir, + source, + }); + } + }; // Build the prefix via the shared encoder so it always matches the // filenames produced by `checkpoint_spatial_indexes`. let prefix = spatial_checkpoint_prefix(database_id, tenant_id, collection); let mut stats = ReclaimStats::default(); - let entries = match std::fs::read_dir(&ckpt_dir) { - Ok(e) => e, - Err(e) => { - warn!( - dir = %ckpt_dir.display(), - error = %e, - "spatial reclaim: failed to read ckpt dir" - ); - return stats; - } - }; - for entry in entries.flatten() { + for entry in entries { + let entry = entry.map_err(|source| ReclaimError::Io { + operation: "read spatial checkpoint entry", + path: ckpt_dir.clone(), + source, + })?; let path = entry.path(); let Some(name) = path.file_name().and_then(|s| s.to_str()) else { continue; @@ -68,14 +72,17 @@ pub fn reclaim_spatial_checkpoints( stats.bytes_freed = stats.bytes_freed.saturating_add(size); debug!(path = %path.display(), size, "spatial reclaim: unlinked"); } - Err(e) => warn!( - path = %path.display(), - error = %e, - "spatial reclaim: unlink failed" - ), + Err(source) if source.kind() == std::io::ErrorKind::NotFound => {} + Err(source) => { + return Err(ReclaimError::Io { + operation: "unlink spatial checkpoint", + path, + source, + }); + } } } - stats + Ok(stats) } #[cfg(test)] @@ -106,7 +113,7 @@ mod tests { // Keep: different database. write(&ckpt.join("1_1_places_geom.ckpt"), b"keep3"); - let stats = reclaim_spatial_checkpoints(base, 0, 1, "places"); + let stats = reclaim_spatial_checkpoints(base, 0, 1, "places").unwrap(); assert_eq!(stats.files_unlinked, 3); assert_eq!(stats.bytes_freed, 1 + 2 + 3); assert!(ckpt.join("0_1_stores_geom.ckpt").exists()); @@ -117,7 +124,7 @@ mod tests { #[test] fn empty_dir_is_noop() { let tmp = TempDir::new().unwrap(); - let s = reclaim_spatial_checkpoints(tmp.path(), 0, 1, "x"); + let s = reclaim_spatial_checkpoints(tmp.path(), 0, 1, "x").unwrap(); assert_eq!(s.files_unlinked, 0); } } diff --git a/nodedb/src/data/executor/handlers/reclaim/timeseries.rs b/nodedb/src/data/executor/handlers/reclaim/timeseries.rs index d52193d2b..e8b0c2db8 100644 --- a/nodedb/src/data/executor/handlers/reclaim/timeseries.rs +++ b/nodedb/src/data/executor/handlers/reclaim/timeseries.rs @@ -11,9 +11,9 @@ use std::path::Path; -use tracing::{debug, warn}; +use tracing::debug; -use super::ReclaimStats; +use super::{ReclaimError, ReclaimStats, Result}; use crate::data::executor::handlers::timeseries::paths::ts_collection_dir; /// Recursively remove the partition directory for `collection`. @@ -24,24 +24,23 @@ pub fn reclaim_timeseries_partitions( database_id: u64, tenant_id: u64, collection: &str, -) -> ReclaimStats { +) -> Result { let partition_dir = ts_collection_dir(data_dir, database_id, tenant_id, collection); - if !partition_dir.exists() { - return ReclaimStats::default(); - } - let mut stats = ReclaimStats::default(); tally_tree(&partition_dir, &mut stats); - if let Err(e) = std::fs::remove_dir_all(&partition_dir) { - warn!( - dir = %partition_dir.display(), - error = %e, - "timeseries reclaim: remove_dir_all failed; partial state left for next attempt" - ); - // Return whatever we tallied; the next idempotent purge will - // pick up the rest. - return stats; + match std::fs::remove_dir_all(&partition_dir) { + Ok(()) => {} + Err(source) if source.kind() == std::io::ErrorKind::NotFound => { + return Ok(ReclaimStats::default()); + } + Err(source) => { + return Err(ReclaimError::Io { + operation: "remove timeseries partition directory", + path: partition_dir, + source, + }); + } } debug!( dir = %partition_dir.display(), @@ -49,7 +48,7 @@ pub fn reclaim_timeseries_partitions( bytes = stats.bytes_freed, "timeseries reclaim: partition directory removed" ); - stats + Ok(stats) } /// Walk the tree rooted at `root`, accumulating file count and byte @@ -96,7 +95,7 @@ mod tests { std::fs::create_dir_all(&other).unwrap(); std::fs::write(other.join("x.bin"), b"keep").unwrap(); - let stats = reclaim_timeseries_partitions(base, 0, 1, "metrics"); + let stats = reclaim_timeseries_partitions(base, 0, 1, "metrics").unwrap(); assert_eq!(stats.files_unlinked, 3); assert_eq!(stats.bytes_freed, 4 + 2 + 1); assert!(!ts_collection_dir(base, 0, 1, "metrics").exists()); @@ -106,7 +105,7 @@ mod tests { #[test] fn missing_dir_is_noop() { let tmp = TempDir::new().unwrap(); - let s = reclaim_timeseries_partitions(tmp.path(), 0, 1, "nope"); + let s = reclaim_timeseries_partitions(tmp.path(), 0, 1, "nope").unwrap(); assert_eq!(s.files_unlinked, 0); } } diff --git a/nodedb/src/data/executor/handlers/reclaim/vector.rs b/nodedb/src/data/executor/handlers/reclaim/vector.rs index ad4edd372..d8c31825b 100644 --- a/nodedb/src/data/executor/handlers/reclaim/vector.rs +++ b/nodedb/src/data/executor/handlers/reclaim/vector.rs @@ -11,9 +11,9 @@ use std::path::Path; -use tracing::{debug, warn}; +use tracing::debug; -use super::ReclaimStats; +use super::{ReclaimError, ReclaimStats, Result}; /// Unlink every vector checkpoint file for `(database_id, tenant_id, collection)`. /// Returns stats; idempotent (missing files count as 0). @@ -22,27 +22,31 @@ pub fn reclaim_vector_checkpoints( database_id: u64, tenant_id: u64, collection: &str, -) -> ReclaimStats { +) -> Result { let ckpt_dir = data_dir.join("vector-ckpt"); - if !ckpt_dir.exists() { - return ReclaimStats::default(); - } + let entries = match std::fs::read_dir(&ckpt_dir) { + Ok(entries) => entries, + Err(source) if source.kind() == std::io::ErrorKind::NotFound => { + return Ok(ReclaimStats::default()); + } + Err(source) => { + return Err(ReclaimError::Io { + operation: "read vector checkpoint directory", + path: ckpt_dir, + source, + }); + } + }; let prefix_exact = format!("{database_id}:{tenant_id}:{collection}"); let prefix_field = format!("{database_id}:{tenant_id}:{collection}:"); let mut stats = ReclaimStats::default(); - let entries = match std::fs::read_dir(&ckpt_dir) { - Ok(e) => e, - Err(e) => { - warn!( - dir = %ckpt_dir.display(), - error = %e, - "vector reclaim: failed to read ckpt dir" - ); - return stats; - } - }; - for entry in entries.flatten() { + for entry in entries { + let entry = entry.map_err(|source| ReclaimError::Io { + operation: "read vector checkpoint entry", + path: ckpt_dir.clone(), + source, + })?; let path = entry.path(); let Some(stem) = path.file_stem().and_then(|s| s.to_str()) else { continue; @@ -68,14 +72,17 @@ pub fn reclaim_vector_checkpoints( stats.bytes_freed = stats.bytes_freed.saturating_add(size); debug!(path = %path.display(), size, "vector reclaim: unlinked ckpt"); } - Err(e) => warn!( - path = %path.display(), - error = %e, - "vector reclaim: unlink failed" - ), + Err(source) if source.kind() == std::io::ErrorKind::NotFound => {} + Err(source) => { + return Err(ReclaimError::Io { + operation: "unlink vector checkpoint", + path, + source, + }); + } } } - stats + Ok(stats) } #[cfg(test)] @@ -105,7 +112,7 @@ mod tests { // Different database: must not touch. write(&ckpt.join("1:1:users.ckpt"), b"keepdb"); - let stats = reclaim_vector_checkpoints(base, 0, 1, "users"); + let stats = reclaim_vector_checkpoints(base, 0, 1, "users").unwrap(); assert_eq!(stats.files_unlinked, 3); assert_eq!(stats.bytes_freed, 1 + 2 + 3); assert!(ckpt.join("0:1:orders.ckpt").exists()); @@ -114,10 +121,21 @@ mod tests { assert!(!ckpt.join("0:1:users.ckpt").exists()); } + #[test] + fn unlink_failure_is_returned_to_lifecycle_barrier() { + let tmp = TempDir::new().unwrap(); + let invalid_target = tmp.path().join("vector-ckpt/0:1:users.ckpt"); + std::fs::create_dir_all(&invalid_target).unwrap(); + + let error = reclaim_vector_checkpoints(tmp.path(), 0, 1, "users").unwrap_err(); + assert!(error.to_string().contains("unlink vector checkpoint")); + assert!(invalid_target.exists()); + } + #[test] fn empty_dir_is_noop() { let tmp = TempDir::new().unwrap(); - let stats = reclaim_vector_checkpoints(tmp.path(), 0, 1, "x"); + let stats = reclaim_vector_checkpoints(tmp.path(), 0, 1, "x").unwrap(); assert_eq!(stats.files_unlinked, 0); } @@ -127,7 +145,7 @@ mod tests { let ckpt = tmp.path().join("vector-ckpt"); // Prefix overlap but distinct collection name. write(&ckpt.join("0:1:users_archive.ckpt"), b"keep"); - let stats = reclaim_vector_checkpoints(tmp.path(), 0, 1, "users"); + let stats = reclaim_vector_checkpoints(tmp.path(), 0, 1, "users").unwrap(); assert_eq!(stats.files_unlinked, 0); assert!(ckpt.join("0:1:users_archive.ckpt").exists()); } diff --git a/nodedb/src/data/executor/handlers/unregister_collection.rs b/nodedb/src/data/executor/handlers/unregister_collection.rs index 2c2807548..dec7a9f58 100644 --- a/nodedb/src/data/executor/handlers/unregister_collection.rs +++ b/nodedb/src/data/executor/handlers/unregister_collection.rs @@ -369,41 +369,71 @@ impl CoreLoop { // with no per-collection persistent file (KV hash index, // CRDT — per-tenant checkpoint) are N/A. let mut l1 = reclaim::ReclaimStats::default(); - l1.merge(reclaim::vector::reclaim_vector_checkpoints( - &self.data_dir, - db_raw, + l1.merge(retry_reclaim( + "vector checkpoints", tid_raw, collection, - )); - l1.merge(reclaim::spatial::reclaim_spatial_checkpoints( - &self.data_dir, - db_raw, + || { + reclaim::vector::reclaim_vector_checkpoints( + &self.data_dir, + db_raw, + tid_raw, + collection, + ) + }, + )?); + l1.merge(retry_reclaim( + "spatial checkpoints", tid_raw, collection, - )); - l1.merge(reclaim::sparse_vector::reclaim_sparse_vector_checkpoints( - &self.data_dir, - db_raw, + || { + reclaim::spatial::reclaim_spatial_checkpoints( + &self.data_dir, + db_raw, + tid_raw, + collection, + ) + }, + )?); + l1.merge(retry_reclaim( + "sparse-vector checkpoints", tid_raw, collection, - )); - l1.merge(reclaim::timeseries::reclaim_timeseries_partitions( - &self.data_dir, - db_raw, + || { + reclaim::sparse_vector::reclaim_sparse_vector_checkpoints( + &self.data_dir, + db_raw, + tid_raw, + collection, + ) + }, + )?); + l1.merge(retry_reclaim( + "timeseries partitions", tid_raw, collection, - )); + || { + reclaim::timeseries::reclaim_timeseries_partitions( + &self.data_dir, + db_raw, + tid_raw, + collection, + ) + }, + )?); - // Doc configs + chain hashes + aggregate cache: the collection's schema - // and derived metadata. Preserved for clear-then-install (the snapshot - // carries row data, not the collection definition); removed on a full DROP. + // Doc configs + chain hashes + derived-result cache: the collection's + // schema and metadata. Preserved for clear-then-install (the snapshot + // carries row data, not the collection definition); removed on a full + // DROP. Cache entries encode query shape after a NUL-terminated + // collection prefix, so lifecycle eviction must use the canonical + // invalidation API rather than compare the encoded key to `coll`. if !preserve_collection_metadata { self.doc_configs .retain(|(d, t, c), _| !(*d == db && *t == tid && c == &coll)); self.chain_hashes .retain(|(d, t, c), _| !(*d == db && *t == tid && c == &coll)); - self.aggregate_cache - .retain(|(d, t, c), _| !(*d == db && *t == tid && c == &coll)); + self.invalidate_aggregate_cache_for_collection(db_raw, tid_raw, collection); } Ok(ClearCollectionStats { @@ -420,3 +450,56 @@ impl CoreLoop { }) } } + +#[cfg(test)] +mod tests { + use std::sync::Arc; + + use nodedb_bridge::buffer::RingBuffer; + + use super::*; + use crate::bridge::dispatch::{BridgeRequest, BridgeResponse}; + + fn open_core() -> (CoreLoop, tempfile::TempDir) { + let dir = tempfile::tempdir().expect("tempdir"); + let (_request_tx, request_rx) = RingBuffer::channel::(64); + let (response_tx, _response_rx) = RingBuffer::channel::(64); + let core = CoreLoop::open( + 0, + request_rx, + response_tx, + dir.path(), + Arc::new(nodedb_types::OrdinalClock::new()), + ) + .expect("open core loop"); + (core, dir) + } + + #[test] + fn full_clear_removes_encoded_derived_cache_entries() { + let (mut core, _dir) = open_core(); + let database = DatabaseId::DEFAULT; + let tenant = TenantId::new(1); + + core.aggregate_cache.insert( + (database, tenant, "products\0count(*)".to_string()), + vec![1], + ); + core.aggregate_cache.insert( + (database, tenant, "products\0facet:brand".to_string()), + vec![2], + ); + core.aggregate_cache + .insert((database, tenant, "other\0count(*)".to_string()), vec![3]); + + core.clear_collection_all_engines(database, tenant, "products", false) + .expect("full collection clear"); + + assert_eq!(core.aggregate_cache.len(), 1); + assert!( + core.aggregate_cache + .contains_key(&(database, tenant, "other\0count(*)".to_string())), + "collection-scoped clear must preserve another collection's cache" + ); + } +} diff --git a/nodedb/src/data/executor/handlers/unregister_materialized_view.rs b/nodedb/src/data/executor/handlers/unregister_materialized_view.rs index e6b67ad8a..08436101b 100644 --- a/nodedb/src/data/executor/handlers/unregister_materialized_view.rs +++ b/nodedb/src/data/executor/handlers/unregister_materialized_view.rs @@ -66,10 +66,14 @@ impl CoreLoop { self.doc_cache .evict_collection(task.request.database_id.as_u64(), tenant_id, name); - // Aggregate + chain-hash + doc-config caches are all keyed - // `(tenant, collection_or_mv_name)` — same retain sweep. - self.aggregate_cache - .retain(|(d, t, c), _| !(*d == db && *t == tid && c == &nm)); + // Derived-result entries encode query shape after a NUL-terminated + // object-name prefix. Use the canonical lifecycle sweep rather than + // comparing that encoded key to the bare materialized-view name. + self.invalidate_aggregate_cache_for_collection( + task.request.database_id.as_u64(), + tenant_id, + name, + ); self.chain_hashes .retain(|(d, t, c), _| !(*d == db && *t == tid && c == &nm)); self.doc_configs diff --git a/nodedb/src/event/collection_gc/pending_reclaim.rs b/nodedb/src/event/collection_gc/pending_reclaim.rs index fdff5db80..10b89d11c 100644 --- a/nodedb/src/event/collection_gc/pending_reclaim.rs +++ b/nodedb/src/event/collection_gc/pending_reclaim.rs @@ -49,28 +49,28 @@ async fn run_loop(shared: Arc) { ); loop { tokio::time::sleep(TICK_INTERVAL).await; - drain_once(&shared).await; + if let Err(error) = drain_once(&shared).await { + warn!(error = %error, "pending-reclaim worker pass incomplete"); + } } } /// One worker pass: re-run the engine purge for every queued entry. /// Public for the boot-time drain and for testing. -pub async fn drain_once(shared: &SharedState) { +pub async fn drain_once(shared: &SharedState) -> crate::Result<()> { let catalog = shared.credentials.catalog(); - - let queue = match catalog.load_pending_reclaim_queue() { - Ok(q) => q, - Err(e) => { - warn!(error = %e, "pending-reclaim: failed to load queue"); - return; - } - }; - - if queue.is_empty() { - return; - } + let queue = catalog.load_pending_reclaim_queue()?; + let mut last_error = None; for entry in queue { + if !shared + .quiesce + .is_draining(entry.database_id, entry.tenant_id, &entry.name) + { + shared + .quiesce + .begin_drain(entry.database_id, entry.tenant_id, &entry.name); + } match crate::control::server::shared::ddl::neutral::collection::purge::dispatch_unregister_collection( shared, entry.database_id, @@ -81,15 +81,40 @@ pub async fn drain_once(shared: &SharedState) { .await { Ok(()) => { - if let Err(e) = catalog.remove_pending_reclaim(entry.database_id, entry.tenant_id, &entry.name) { + if let Err(error) = + crate::control::catalog_entry::apply::collection::finalize_purge( + entry.database_id, + entry.tenant_id, + &entry.name, + catalog, + ) + { + warn!( + tenant = entry.tenant_id, + collection = %entry.name, + error = %error, + "pending-reclaim: engine rows purged but catalog finalization failed" + ); + last_error = Some(error.to_string()); + continue; + } + if let Err(error) = catalog.remove_pending_reclaim( + entry.database_id, + entry.tenant_id, + &entry.name, + ) { warn!( tenant = entry.tenant_id, collection = %entry.name, - error = %e, + error = %error, "pending-reclaim: purged engine rows but failed to reap queue entry" ); + last_error = Some(error.to_string()); continue; } + shared + .quiesce + .forget(entry.database_id, entry.tenant_id, &entry.name); debug!( tenant = entry.tenant_id, collection = %entry.name, @@ -121,7 +146,16 @@ pub async fn drain_once(shared: &SharedState) { error = %msg, "pending-reclaim: engine purge failed; will retry next tick" ); + last_error = Some(msg); } } } + + if let Some(detail) = last_error { + return Err(crate::Error::Storage { + engine: "pending-reclaim".into(), + detail, + }); + } + Ok(()) } diff --git a/nodedb/src/event/collection_gc/sweeper.rs b/nodedb/src/event/collection_gc/sweeper.rs index be3be73ed..c78cdc6b0 100644 --- a/nodedb/src/event/collection_gc/sweeper.rs +++ b/nodedb/src/event/collection_gc/sweeper.rs @@ -70,7 +70,7 @@ async fn run_loop(shared: Arc) { // per-tenant overrides are resolved inside `sweep_once` via // `effective_retention_for_tenant`. let (interval, retention, _) = load_settings(&shared); - if let Err(e) = sweep_once(&shared, retention) { + if let Err(e) = sweep_once(&shared, retention).await { warn!(error = %e, "collection-gc sweep failed; will retry next tick"); } tokio::time::sleep(interval).await; @@ -104,7 +104,7 @@ fn effective_retention_for_tenant( /// Single sweep pass. Public for testability — callers can drive the /// sweeper synchronously in tests without waiting on the interval. -pub fn sweep_once(shared: &SharedState, retention: Duration) -> crate::Result<()> { +pub async fn sweep_once(shared: &SharedState, retention: Duration) -> crate::Result<()> { let catalog = shared.credentials.catalog(); let now_ns = SystemTime::now() @@ -145,7 +145,38 @@ pub fn sweep_once(shared: &SharedState, retention: Duration) -> crate::Result<() name: coll.name.clone(), }; match crate::control::metadata_proposer::propose_catalog_entry(shared, &entry) { - Ok(_) => { + Ok(log_index) => { + if log_index == 0 { + let mut lifecycle = Some( + shared + .quiesce + .acquire_lifecycle( + coll.database_id.as_u64(), + coll.tenant_id, + &coll.name, + ) + .await, + ); + let purge_lsn = shared.wal.next_lsn().as_u64(); + if let Err(error) = crate::control::server::shared::ddl::neutral::collection::purge::hard_purge_collection( + shared, + coll.database_id.as_u64(), + coll.tenant_id, + &coll.name, + purge_lsn, + true, + ) + .await + { + if let Some(guard) = lifecycle.take() { + guard.disarm(); + } + return Err(crate::Error::Storage { + engine: "collection-gc".into(), + detail: error.to_string(), + }); + } + } proposed += 1; debug!( tenant = coll.tenant_id, diff --git a/nodedb/src/event/crdt_sync/delivery.rs b/nodedb/src/event/crdt_sync/delivery.rs index fa1d0c596..acbd11147 100644 --- a/nodedb/src/event/crdt_sync/delivery.rs +++ b/nodedb/src/event/crdt_sync/delivery.rs @@ -98,8 +98,8 @@ impl CrdtSyncDelivery { } /// Broadcast a `CollectionPurged` control frame to every session - /// currently subscribed to `(tenant_id, collection)`. Runs when - /// `purge_async` finishes the per-node reclaim, so every online + /// currently subscribed to `(tenant_id, collection)`. Runs when the + /// per-node reclaim barrier finishes, so every online /// Lite client learns about the hard-delete within one network /// round-trip and can drop local state. Offline clients pick it /// up on reconnect via the `last_seen_lsn` replay path. diff --git a/nodedb/src/event/wal_replay.rs b/nodedb/src/event/wal_replay.rs index b778e7cad..39db9e765 100644 --- a/nodedb/src/event/wal_replay.rs +++ b/nodedb/src/event/wal_replay.rs @@ -85,7 +85,7 @@ fn convert_records_to_events( // Collection tombstones shadow any prior write in the same stream. // Extract once, then drop events whose `(tenant, collection, lsn)` // is covered. - let tombstones = nodedb_wal::extract_tombstones(records); + let tombstones = nodedb_wal::extract_tombstones(records)?; for record in records { let vshard_id = record.header.vshard_id as usize; diff --git a/nodedb/tests/collection_hard_delete.rs b/nodedb/tests/collection_hard_delete.rs index 2bb85e475..f0d876f31 100644 --- a/nodedb/tests/collection_hard_delete.rs +++ b/nodedb/tests/collection_hard_delete.rs @@ -174,10 +174,11 @@ fn reclaim_handlers_are_idempotent_on_missing_files() { // All four reclaim helpers must return default stats on a fresh // empty data dir. - let vector = reclaim::vector::reclaim_vector_checkpoints(base, 0, TENANT, "x"); - let spatial = reclaim::spatial::reclaim_spatial_checkpoints(base, 0, TENANT, "x"); - let sparse = reclaim::sparse_vector::reclaim_sparse_vector_checkpoints(base, 0, TENANT, "x"); - let ts = reclaim::timeseries::reclaim_timeseries_partitions(base, 0, TENANT, "x"); + let vector = reclaim::vector::reclaim_vector_checkpoints(base, 0, TENANT, "x").unwrap(); + let spatial = reclaim::spatial::reclaim_spatial_checkpoints(base, 0, TENANT, "x").unwrap(); + let sparse = + reclaim::sparse_vector::reclaim_sparse_vector_checkpoints(base, 0, TENANT, "x").unwrap(); + let ts = reclaim::timeseries::reclaim_timeseries_partitions(base, 0, TENANT, "x").unwrap(); assert_eq!(vector.files_unlinked, 0); assert_eq!(spatial.files_unlinked, 0); @@ -185,10 +186,10 @@ fn reclaim_handlers_are_idempotent_on_missing_files() { assert_eq!(ts.files_unlinked, 0); // Re-running must still succeed (no "already deleted" error). - let _ = reclaim::vector::reclaim_vector_checkpoints(base, 0, TENANT, "x"); - let _ = reclaim::spatial::reclaim_spatial_checkpoints(base, 0, TENANT, "x"); - let _ = reclaim::sparse_vector::reclaim_sparse_vector_checkpoints(base, 0, TENANT, "x"); - let _ = reclaim::timeseries::reclaim_timeseries_partitions(base, 0, TENANT, "x"); + reclaim::vector::reclaim_vector_checkpoints(base, 0, TENANT, "x").unwrap(); + reclaim::spatial::reclaim_spatial_checkpoints(base, 0, TENANT, "x").unwrap(); + reclaim::sparse_vector::reclaim_sparse_vector_checkpoints(base, 0, TENANT, "x").unwrap(); + reclaim::timeseries::reclaim_timeseries_partitions(base, 0, TENANT, "x").unwrap(); } /// The reclaim handlers don't touch other tenants' or other @@ -204,7 +205,7 @@ fn reclaim_is_scoped_to_tenant_and_collection() { std::fs::write(vec_dir.join("0:1:orders.ckpt"), b"b").unwrap(); std::fs::write(vec_dir.join("0:2:users.ckpt"), b"c").unwrap(); - let stats = reclaim::vector::reclaim_vector_checkpoints(base, 0, 1, "users"); + let stats = reclaim::vector::reclaim_vector_checkpoints(base, 0, 1, "users").unwrap(); assert_eq!(stats.files_unlinked, 1); assert!(!vec_dir.join("0:1:users.ckpt").exists()); // Siblings untouched. diff --git a/nodedb/tests/sql_drop_collection.rs b/nodedb/tests/sql_drop_collection.rs index 5c9348721..c5249f199 100644 --- a/nodedb/tests/sql_drop_collection.rs +++ b/nodedb/tests/sql_drop_collection.rs @@ -167,6 +167,15 @@ async fn drop_then_recreate_document_strict_starts_empty() { 7, "7 rows must exist before DROP, got {before:?}" ); + let cached_before = srv + .query_text("SELECT COUNT(*) FROM recycled_strict") + .await + .unwrap(); + assert_eq!( + cached_before, + vec!["7".to_string()], + "COUNT(*) must reflect the predecessor before DROP, got {cached_before:?}" + ); srv.exec("DROP COLLECTION recycled_strict").await.unwrap(); srv.exec( @@ -196,3 +205,41 @@ async fn drop_then_recreate_document_strict_starts_empty() { "COUNT(*) on a freshly re-created strict collection must be 0, got {count:?}" ); } + +/// Grouped aggregates over a same-name re-CREATE must observe the new +/// collection incarnation. With no rows in the replacement, SQL aggregate +/// semantics require zero groups rather than groups cached from its predecessor. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn drop_then_recreate_discards_predecessor_grouped_aggregates() { + let srv = TestServer::start().await; + + srv.exec("CREATE COLLECTION recycled_groups").await.unwrap(); + srv.exec("INSERT INTO recycled_groups { id: 'a', bucket: 'old' }") + .await + .unwrap(); + srv.exec("INSERT INTO recycled_groups { id: 'b', bucket: 'old' }") + .await + .unwrap(); + + let predecessor = srv + .query_rows("SELECT bucket, COUNT(*) AS n FROM recycled_groups GROUP BY bucket") + .await + .unwrap(); + assert_eq!( + predecessor.len(), + 1, + "predecessor must produce one cached group, got {predecessor:?}" + ); + + srv.exec("DROP COLLECTION recycled_groups").await.unwrap(); + srv.exec("CREATE COLLECTION recycled_groups").await.unwrap(); + + let replacement = srv + .query_rows("SELECT bucket, COUNT(*) AS n FROM recycled_groups GROUP BY bucket") + .await + .unwrap(); + assert!( + replacement.is_empty(), + "empty replacement must not inherit predecessor groups: {replacement:?}" + ); +} diff --git a/nodedb/tests/sql_materialized_view_refresh.rs b/nodedb/tests/sql_materialized_view_refresh.rs index 6575d94b3..5e9d66a36 100644 --- a/nodedb/tests/sql_materialized_view_refresh.rs +++ b/nodedb/tests/sql_materialized_view_refresh.rs @@ -147,6 +147,110 @@ async fn refresh_applies_group_by_aggregate() { ); } +/// Dropping and re-creating a materialized view under the same name creates a +/// new target. Before its first refresh, a scalar aggregate over that empty +/// target must return the zero-row identity rather than the predecessor's +/// cached result. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn drop_then_recreate_materialized_view_discards_cached_aggregates() { + let server = TestServer::start().await; + + server.exec("CREATE COLLECTION mv_reuse_src").await.unwrap(); + server + .exec("INSERT INTO mv_reuse_src { id: 'r1', value: 1 }") + .await + .unwrap(); + server + .exec("INSERT INTO mv_reuse_src { id: 'r2', value: 2 }") + .await + .unwrap(); + server + .exec( + "CREATE MATERIALIZED VIEW mv_reuse ON mv_reuse_src AS \ + SELECT id, value FROM mv_reuse_src", + ) + .await + .unwrap(); + server + .exec("REFRESH MATERIALIZED VIEW mv_reuse") + .await + .unwrap(); + + let predecessor = server + .query_text("SELECT COUNT(*) FROM mv_reuse") + .await + .unwrap(); + assert_eq!(predecessor, vec!["2".to_string()]); + + assert!( + server.exec("DROP COLLECTION mv_reuse PURGE").await.is_err(), + "a materialized-view target cannot be dropped independently of its owner" + ); + + server + .exec("DROP MATERIALIZED VIEW mv_reuse") + .await + .unwrap(); + server + .exec( + "CREATE MATERIALIZED VIEW mv_reuse ON mv_reuse_src AS \ + SELECT id, value FROM mv_reuse_src", + ) + .await + .unwrap(); + + let replacement_rows = server.query_rows("SELECT id FROM mv_reuse").await.unwrap(); + assert!( + replacement_rows.is_empty(), + "new materialized-view target must start without predecessor rows: {replacement_rows:?}" + ); + + let replacement = server + .query_text("SELECT COUNT(*) FROM mv_reuse") + .await + .unwrap(); + assert_eq!( + replacement, + vec!["0".to_string()], + "new materialized-view target must not inherit predecessor aggregates" + ); +} + +/// A materialized view owns and purges its same-name target collection. +/// Creation must therefore reject a pre-existing user collection rather than +/// silently adopting storage that DROP MATERIALIZED VIEW would later erase. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn create_rejects_user_collection_with_same_name() { + let server = TestServer::start().await; + + server.exec("CREATE COLLECTION mv_owner_src").await.unwrap(); + server + .exec("CREATE COLLECTION mv_owned_name") + .await + .unwrap(); + server + .exec("INSERT INTO mv_owned_name { id: 'keep', value: 7 }") + .await + .unwrap(); + + let result = server + .exec( + "CREATE MATERIALIZED VIEW mv_owned_name ON mv_owner_src AS \ + SELECT id FROM mv_owner_src", + ) + .await; + assert!( + result.is_err(), + "materialized view must not adopt a user-owned same-name collection" + ); + + let rows = server + .query_rows("SELECT id FROM mv_owned_name") + .await + .unwrap(); + assert_eq!(rows, vec![vec!["keep".to_string()]]); +} + /// A view whose SELECT joins two collections must materialize the /// joined rows, not copies of one source's raw docs. #[tokio::test(flavor = "multi_thread", worker_threads = 4)] From d5c45c2595795770247e0c819211e16993b2cfc8 Mon Sep 17 00:00:00 2001 From: Farhan Syah Date: Tue, 21 Jul 2026 22:57:52 +0800 Subject: [PATCH 2/2] 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". --- nodedb-physical/src/physical_plan/meta.rs | 9 + nodedb/src/bridge/quiesce/drain.rs | 48 +----- .../post_apply/async_dispatch/collection.rs | 158 ++++++++++++------ .../async_dispatch/materialized_view.rs | 2 +- .../control/catalog_entry/post_apply/mod.rs | 2 +- .../ddl/neutral/collection/create/build.rs | 20 ++- .../shared/ddl/neutral/collection/drop.rs | 11 +- .../ddl/neutral/collection/purge/dispatch.rs | 9 + .../ddl/neutral/collection/purge/hard.rs | 12 +- .../ddl/neutral/materialized_view/create.rs | 14 +- .../ddl/neutral/materialized_view/drop.rs | 13 +- .../predicate/txn_buffering.rs | 1 + nodedb/src/data/executor/dispatch/meta.rs | 9 +- .../snapshot/restore/tenant_snapshot.rs | 3 + .../handlers/unregister_collection.rs | 123 ++++++++------ nodedb/src/event/collection_gc/sweeper.rs | 8 +- 16 files changed, 277 insertions(+), 165 deletions(-) diff --git a/nodedb-physical/src/physical_plan/meta.rs b/nodedb-physical/src/physical_plan/meta.rs index b1ed27f41..3830b5729 100644 --- a/nodedb-physical/src/physical_plan/meta.rs +++ b/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 diff --git a/nodedb/src/bridge/quiesce/drain.rs b/nodedb/src/bridge/quiesce/drain.rs index 9954613b2..24fbe33a6 100644 --- a/nodedb/src/bridge/quiesce/drain.rs +++ b/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] diff --git a/nodedb/src/control/catalog_entry/post_apply/async_dispatch/collection.rs b/nodedb/src/control/catalog_entry/post_apply/async_dispatch/collection.rs index 6a27338fa..a61153804 100644 --- a/nodedb/src/control/catalog_entry/post_apply/async_dispatch/collection.rs +++ b/nodedb/src/control/catalog_entry/post_apply/async_dispatch/collection.rs @@ -19,6 +19,49 @@ pub async fn put_async(stored: StoredCollection, shared: Arc) { 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) -> 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 diff --git a/nodedb/src/control/catalog_entry/post_apply/async_dispatch/materialized_view.rs b/nodedb/src/control/catalog_entry/post_apply/async_dispatch/materialized_view.rs index 179a50f01..adc5d918f 100644 --- a/nodedb/src/control/catalog_entry/post_apply/async_dispatch/materialized_view.rs +++ b/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, -) -> crate::Result<()> { +) -> Result<(), super::collection::ReclaimFailure> { super::collection::reclaim_collection_storage(&shared, 0, tenant_id, &name, purge_lsn, false) .await } diff --git a/nodedb/src/control/catalog_entry/post_apply/mod.rs b/nodedb/src/control/catalog_entry/post_apply/mod.rs index 02f6769a5..a71c216e9 100644 --- a/nodedb/src/control/catalog_entry/post_apply/mod.rs +++ b/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; diff --git a/nodedb/src/control/server/shared/ddl/neutral/collection/create/build.rs b/nodedb/src/control/server/shared/ddl/neutral/collection/create/build.rs index e3b951871..abceaece2 100644 --- a/nodedb/src/control/server/shared/ddl/neutral/collection/create/build.rs +++ b/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())); } } diff --git a/nodedb/src/control/server/shared/ddl/neutral/collection/drop.rs b/nodedb/src/control/server/shared/ddl/neutral/collection/drop.rs index ced131098..b60444b8a 100644 --- a/nodedb/src/control/server/shared/ddl/neutral/collection/drop.rs +++ b/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 diff --git a/nodedb/src/control/server/shared/ddl/neutral/collection/purge/dispatch.rs b/nodedb/src/control/server/shared/ddl/neutral/collection/purge/dispatch.rs index 0dd8617bb..b8b550691 100644 --- a/nodedb/src/control/server/shared/ddl/neutral/collection/purge/dispatch.rs +++ b/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, diff --git a/nodedb/src/control/server/shared/ddl/neutral/collection/purge/hard.rs b/nodedb/src/control/server/shared/ddl/neutral/collection/purge/hard.rs index 3e1be3198..e9b6b07a9 100644 --- a/nodedb/src/control/server/shared/ddl/neutral/collection/purge/hard.rs +++ b/nodedb/src/control/server/shared/ddl/neutral/collection/purge/hard.rs @@ -20,8 +20,7 @@ //! new collection. No engine is special-cased: the reclaim dispatch //! covers every engine exactly as `DROP ... PURGE` does. -use nodedb_types::error::NodeDbError; - +use crate::control::catalog_entry::post_apply::ReclaimFailure; use crate::control::state::SharedState; /// Hard-purge `(tenant_id, name)`: remove the catalog metadata row @@ -53,11 +52,13 @@ pub(crate) async fn hard_purge_collection( name: &str, purge_lsn: u64, drain_already_held: bool, -) -> Result<(), NodeDbError> { +) -> Result<(), ReclaimFailure> { // 1. Remove the catalog metadata row (primary StoredCollection, // owner row, surrogate map) — the synchronous half of the // `PurgeCollection` applier. Propagate failure: if the old row - // survives, the new collection must NOT register over it. + // survives, the new collection must NOT register over it. This runs + // before any durable retry record is queued, so it is a `no_retry` + // failure — the caller releases its lifecycle guard. { let catalog = state.credentials.catalog(); crate::control::catalog_entry::apply::collection::prepare_purge( @@ -65,7 +66,8 @@ pub(crate) async fn hard_purge_collection( tenant_id, name, catalog, - )?; + ) + .map_err(ReclaimFailure::no_retry)?; } // 2. Reclaim engine-local storage on the Data Plane (WAL tombstone, diff --git a/nodedb/src/control/server/shared/ddl/neutral/materialized_view/create.rs b/nodedb/src/control/server/shared/ddl/neutral/materialized_view/create.rs index a044ed9fb..01c8c4e98 100644 --- a/nodedb/src/control/server/shared/ddl/neutral/materialized_view/create.rs +++ b/nodedb/src/control/server/shared/ddl/neutral/materialized_view/create.rs @@ -78,13 +78,23 @@ pub async fn create_materialized_view( } } - if let Ok(Some(_)) = catalog.get_materialized_view(tenant_id.as_u64(), &name) { + // A catalog-read fault must abort the CREATE: proceeding could adopt a + // same-name target over an object whose existence check transiently + // failed. + if catalog + .get_materialized_view(tenant_id.as_u64(), &name) + .map_err(|error| err("XX000", error.to_string()))? + .is_some() + { return Err(err( "42P07", format!("materialized view '{name}' already exists"), )); } - if let Ok(Some(_)) = catalog.get_collection(DatabaseId::DEFAULT, tenant_id.as_u64(), &name) + if catalog + .get_collection(DatabaseId::DEFAULT, tenant_id.as_u64(), &name) + .map_err(|error| err("XX000", error.to_string()))? + .is_some() { return Err(err("42P07", format!("collection '{name}' already exists"))); } diff --git a/nodedb/src/control/server/shared/ddl/neutral/materialized_view/drop.rs b/nodedb/src/control/server/shared/ddl/neutral/materialized_view/drop.rs index f75e50831..464b28bfd 100644 --- a/nodedb/src/control/server/shared/ddl/neutral/materialized_view/drop.rs +++ b/nodedb/src/control/server/shared/ddl/neutral/materialized_view/drop.rs @@ -140,11 +140,18 @@ pub fn drop_materialized_view( .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; a + // no-retry failure releases the hold via the guard's unwind Drop. + if failure.retry_queued + && let Some(guard) = local_lifecycle.take() + { guard.disarm(); } - panic!("local materialized-view target reclaim failed: {error}"); + panic!( + "local materialized-view target reclaim failed: {}", + failure.error + ); } } diff --git a/nodedb/src/control/server/shared/write_admission/predicate/txn_buffering.rs b/nodedb/src/control/server/shared/write_admission/predicate/txn_buffering.rs index 570898255..5118660ab 100644 --- a/nodedb/src/control/server/shared/write_admission/predicate/txn_buffering.rs +++ b/nodedb/src/control/server/shared/write_admission/predicate/txn_buffering.rs @@ -1697,6 +1697,7 @@ mod tests { tenant_id: 1, name: "c".into(), purge_lsn: 0, + reclaim_l1_files: true, }), PhysicalPlan::Meta(MetaOp::UnregisterMaterializedView { tenant_id: 1, diff --git a/nodedb/src/data/executor/dispatch/meta.rs b/nodedb/src/data/executor/dispatch/meta.rs index 46a12df11..0da158848 100644 --- a/nodedb/src/data/executor/dispatch/meta.rs +++ b/nodedb/src/data/executor/dispatch/meta.rs @@ -92,7 +92,14 @@ impl CoreLoop { tenant_id, name, purge_lsn, - } => self.execute_unregister_collection(task, *tenant_id, name, *purge_lsn), + reclaim_l1_files, + } => self.execute_unregister_collection( + task, + *tenant_id, + name, + *purge_lsn, + *reclaim_l1_files, + ), MetaOp::UnregisterMaterializedView { tenant_id, name } => { self.execute_unregister_materialized_view(task, *tenant_id, name) diff --git a/nodedb/src/data/executor/handlers/snapshot/restore/tenant_snapshot.rs b/nodedb/src/data/executor/handlers/snapshot/restore/tenant_snapshot.rs index 5ebea6575..20e09ae48 100644 --- a/nodedb/src/data/executor/handlers/snapshot/restore/tenant_snapshot.rs +++ b/nodedb/src/data/executor/handlers/snapshot/restore/tenant_snapshot.rs @@ -45,6 +45,9 @@ impl CoreLoop { crate::types::TenantId::new(*tid_raw), coll, true, + // Single-core clear-then-install: reclaim this collection's + // shared L1 files here (no concurrent-core race to avoid). + true, ) { return self.response_error( task, diff --git a/nodedb/src/data/executor/handlers/unregister_collection.rs b/nodedb/src/data/executor/handlers/unregister_collection.rs index dec7a9f58..33aedfbe9 100644 --- a/nodedb/src/data/executor/handlers/unregister_collection.rs +++ b/nodedb/src/data/executor/handlers/unregister_collection.rs @@ -123,6 +123,7 @@ impl CoreLoop { tenant_id: u64, collection: &str, purge_lsn: u64, + reclaim_l1_files: bool, ) -> Response { info!( core = self.core_id, @@ -145,6 +146,7 @@ impl CoreLoop { TenantId::new(tenant_id), collection, false, + reclaim_l1_files, ) { Ok(stats) => stats, // Fail-closed: a failed engine purge must surface as an error so @@ -233,12 +235,20 @@ impl CoreLoop { /// Fail-closed: any persistent-engine reclaim that fails after its bounded /// retries propagates as `Err`; the caller must abort the purge rather than /// let a partially-cleared collection have its catalog row removed. + /// + /// `reclaim_l1_files` gates the shared on-disk L1 unlink pass. Those paths + /// are keyed by `(database, tenant, collection)` — not by core — so in the + /// all-cores `UnregisterCollection` fan-out only the homing core passes + /// `true`; other cores still evict their own per-core in-memory state but + /// must not race `remove_dir_all`/`unlink` on the same tree. Single-core + /// callers (snapshot restore) pass `true`. pub(in crate::data::executor) fn clear_collection_all_engines( &mut self, database_id: DatabaseId, tenant_id: TenantId, collection: &str, preserve_collection_metadata: bool, + reclaim_l1_files: bool, ) -> crate::Result { let db = database_id; let tid = tenant_id; @@ -368,59 +378,66 @@ impl CoreLoop { // strict, FTS, graph edges) already reclaimed above; engines // with no per-collection persistent file (KV hash index, // CRDT — per-tenant checkpoint) are N/A. + // Shared on-disk L1 files are keyed by (database, tenant, collection), + // not by core. Only the homing core reclaims them so concurrent cores + // in the all-cores fan-out cannot race `remove_dir_all`/`unlink` on the + // same tree and turn a benign concurrent-removal errno into a fatal + // barrier failure. let mut l1 = reclaim::ReclaimStats::default(); - l1.merge(retry_reclaim( - "vector checkpoints", - tid_raw, - collection, - || { - reclaim::vector::reclaim_vector_checkpoints( - &self.data_dir, - db_raw, - tid_raw, - collection, - ) - }, - )?); - l1.merge(retry_reclaim( - "spatial checkpoints", - tid_raw, - collection, - || { - reclaim::spatial::reclaim_spatial_checkpoints( - &self.data_dir, - db_raw, - tid_raw, - collection, - ) - }, - )?); - l1.merge(retry_reclaim( - "sparse-vector checkpoints", - tid_raw, - collection, - || { - reclaim::sparse_vector::reclaim_sparse_vector_checkpoints( - &self.data_dir, - db_raw, - tid_raw, - collection, - ) - }, - )?); - l1.merge(retry_reclaim( - "timeseries partitions", - tid_raw, - collection, - || { - reclaim::timeseries::reclaim_timeseries_partitions( - &self.data_dir, - db_raw, - tid_raw, - collection, - ) - }, - )?); + if reclaim_l1_files { + l1.merge(retry_reclaim( + "vector checkpoints", + tid_raw, + collection, + || { + reclaim::vector::reclaim_vector_checkpoints( + &self.data_dir, + db_raw, + tid_raw, + collection, + ) + }, + )?); + l1.merge(retry_reclaim( + "spatial checkpoints", + tid_raw, + collection, + || { + reclaim::spatial::reclaim_spatial_checkpoints( + &self.data_dir, + db_raw, + tid_raw, + collection, + ) + }, + )?); + l1.merge(retry_reclaim( + "sparse-vector checkpoints", + tid_raw, + collection, + || { + reclaim::sparse_vector::reclaim_sparse_vector_checkpoints( + &self.data_dir, + db_raw, + tid_raw, + collection, + ) + }, + )?); + l1.merge(retry_reclaim( + "timeseries partitions", + tid_raw, + collection, + || { + reclaim::timeseries::reclaim_timeseries_partitions( + &self.data_dir, + db_raw, + tid_raw, + collection, + ) + }, + )?); + } // Doc configs + chain hashes + derived-result cache: the collection's // schema and metadata. Preserved for clear-then-install (the snapshot @@ -492,7 +509,7 @@ mod tests { core.aggregate_cache .insert((database, tenant, "other\0count(*)".to_string()), vec![3]); - core.clear_collection_all_engines(database, tenant, "products", false) + core.clear_collection_all_engines(database, tenant, "products", false, true) .expect("full collection clear"); assert_eq!(core.aggregate_cache.len(), 1); diff --git a/nodedb/src/event/collection_gc/sweeper.rs b/nodedb/src/event/collection_gc/sweeper.rs index c78cdc6b0..f4b57249e 100644 --- a/nodedb/src/event/collection_gc/sweeper.rs +++ b/nodedb/src/event/collection_gc/sweeper.rs @@ -168,12 +168,16 @@ pub async fn sweep_once(shared: &SharedState, retention: Duration) -> crate::Res ) .await { - if let Some(guard) = lifecycle.take() { + // Disarm only when a durable retry record owns + // the drain; a no-retry failure releases the + // hold via the guard's Drop so the next sweep + // (or a same-name CREATE) is not wedged. + if error.retry_queued && let Some(guard) = lifecycle.take() { guard.disarm(); } return Err(crate::Error::Storage { engine: "collection-gc".into(), - detail: error.to_string(), + detail: error.error.to_string(), }); } }