Merge pull request #205 from NodeDB-Lab/fix/purge-teardown-and-cluster-harness-superuser

Fix cluster harness superuser identity and single-node tenant teardown storage reclaim

Author: farhan-syah

Cargo.toml

@@ -265,6 +265,13 @@ fluxbench = "0.1"
 
 [profile.dev]
 debug = "line-tables-only"
+# Strip symbol tables from dev/test binaries. With --all-features every test
+# binary statically links the full dependency closure, and there are hundreds of
+# them, so an unstripped target/ runs to ~100 GB. Stripping removes better than a
+# third of each binary. This unsymbolises panics/backtraces on ordinary dev
+# builds; when you need to debug, build with `--profile debugging`, which keeps
+# full symbols (see below).
+strip = true
 
 [profile.dev.package."*"]
 debug = false
@@ -286,3 +293,5 @@ strip = true
 [profile.debugging]
 inherits = "dev"
 debug = true
+# dev strips symbols; override it here so the debugging profile stays symbolised.
+strip = false

nodedb-cluster-tests/tests/native_gather_join_in_txn_occ.rs

@@ -92,6 +92,13 @@ fn pinned_native_client(node: &TestClusterNode) -> NativeClient {
     NativeClient::new(PoolConfig {
         addr: format!("127.0.0.1:{}", node.native_port),
         max_size: 1,
+        // The cluster harness runs trust auth with `nodedb` as the sole
+        // materialized superuser; the `PoolConfig` default user is `admin`,
+        // which trust mode rejects as a non-existent identity. Authenticate as
+        // the harness superuser, matching the pgwire path (`user=nodedb`).
+        auth: nodedb_types::protocol::AuthMethod::Trust {
+            username: "nodedb".into(),
+        },
         ..Default::default()
     })
 }

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

@@ -107,6 +107,11 @@ impl TestClusterNode {
                 &data_dir_path.join("system.redb"),
             )?,
         );
+        // Mirror production/pgwire-harness bootstrap: both listeners run in
+        // `AuthMode::Trust`, which resolves — but never fabricates — a durable
+        // stored identity. Without this every node rejects the harness connect
+        // with `trust auth: user 'nodedb' does not exist`.
+        credentials.bootstrap_trust_superuser("nodedb")?;
         let mut shared =
             SharedState::new_with_credentials(dispatcher, Arc::clone(&wal), credentials)?;
 

nodedb/src/control/catalog_entry/tests/collection.rs

@@ -102,6 +102,10 @@ fn purge_collection_is_scoped_to_database() {
     apply_to(&CatalogEntry::PutCollection(Box::new(default)), catalog);
     apply_to(&CatalogEntry::PutCollection(Box::new(other)), catalog);
 
+    // A `PurgeCollection` apply only deactivates the catalog row (the
+    // crash-durable same-name barrier); the row/owner/surrogate deletion is the
+    // `finalize_purge` half that runs once storage reclaim succeeds. Drive both
+    // to assert the delete is scoped to the target database.
     apply_to(
         &CatalogEntry::PurgeCollection {
             database_id: 9,
@@ -110,6 +114,8 @@ fn purge_collection_is_scoped_to_database() {
         },
         catalog,
     );
+    crate::control::catalog_entry::apply::collection::finalize_purge(9, 1, "shared", catalog)
+        .expect("finalize purge for database 9");
 
     assert!(
         catalog

nodedb/src/control/server/shared/ddl/neutral/user/tenant_purge.rs

@@ -41,6 +41,34 @@ pub(super) fn purge_owned_for_tenant_teardown(
         crate::control::catalog_entry::apply::local::apply_locally_if_needed(
             state, &entry, log_index,
         );
+        // A `PurgeCollection` apply only deactivates the catalog row — the
+        // durable owner/collection deletion and storage reclaim are the
+        // post-apply half. On the clustered path the metadata applier schedules
+        // that reclaim on every node; on the single-node path (`log_index == 0`)
+        // there is no applier, so drive the same reclaim `drop.rs` runs inline.
+        // Without this the teardown leaves the owner row and never reclaims the
+        // collection's storage. Other owner kinds delete fully in their apply.
+        if kind == OwnerKind::Collection && log_index == 0 {
+            let purge_lsn = state.wal.next_lsn().as_u64();
+            let reclaim = tokio::task::block_in_place(|| {
+                tokio::runtime::Handle::current().block_on(
+                    crate::control::server::shared::ddl::neutral::collection::purge::hard_purge_collection(
+                        state,
+                        owner.database_id,
+                        tenant.as_u64(),
+                        &owner.object_name,
+                        purge_lsn,
+                        false,
+                    ),
+                )
+            });
+            reclaim.map_err(|failure| {
+                ddl_err(format!(
+                    "tenant teardown collection reclaim failed for '{}': {}",
+                    owner.object_name, failure.error
+                ))
+            })?;
+        }
         if log_index == 0 {
             state
                 .permissions

nodedb/tests/crash_harness/mod.rs

@@ -233,34 +233,54 @@ impl CrashHarness {
         self.wait_ready(Duration::from_secs(20));
     }
 
+    /// Open a fresh pgwire connection, run one statement, and return the
+    /// resulting messages. Panics on connect/exec error.
+    ///
+    /// Retries the transient "no sequencer leader elected yet" startup
+    /// condition: `/healthz` intentionally reports ready before the Calvin
+    /// sequencer group has elected a leader (the sequencer is deliberately not
+    /// a data group in the readiness gate), so a cross-shard write issued in
+    /// the first moments of uptime can race the election and get a clean,
+    /// retryable error. On a loaded machine that window is wide enough to lose.
+    /// A real client retries; so does the harness, bounded, before writing.
+    async fn simple_query_ready(&self, sql: &str) -> Vec<tokio_postgres::SimpleQueryMessage> {
+        let deadline = Instant::now() + Duration::from_secs(20);
+        loop {
+            let (client, connection) =
+                tokio_postgres::connect(&self.pgwire_conn_str(), tokio_postgres::NoTls)
+                    .await
+                    .expect("connect for exec");
+            let conn_handle = tokio::spawn(async move {
+                let _ = connection.await;
+            });
+            let result = client.simple_query(sql).await;
+            drop(client);
+            let _ = conn_handle.await;
+            match result {
+                Ok(messages) => return messages,
+                Err(e)
+                    if Instant::now() < deadline
+                        && e.as_db_error().is_some_and(|db| {
+                            db.message().contains("no sequencer leader elected yet")
+                        }) =>
+                {
+                    tokio::time::sleep(Duration::from_millis(100)).await;
+                }
+                Err(e) => panic!("exec: {e}"),
+            }
+        }
+    }
+
     /// Open a fresh pgwire connection, run one statement, and drop the
     /// connection. Panics on connect/exec error.
     pub async fn exec(&self, sql: &str) {
-        let (client, connection) =
-            tokio_postgres::connect(&self.pgwire_conn_str(), tokio_postgres::NoTls)
-                .await
-                .expect("connect for exec");
-        let conn_handle = tokio::spawn(async move {
-            let _ = connection.await;
-        });
-        client.simple_query(sql).await.expect("exec");
-        drop(client);
-        let _ = conn_handle.await;
+        let _ = self.simple_query_ready(sql).await;
     }
 
     /// Run a query and return column `col` from every returned row, as text
     /// (via `simple_query`, so the value survives regardless of its type OID).
     pub async fn query_col(&self, sql: &str, col: &str) -> Vec<String> {
-        let (client, connection) =
-            tokio_postgres::connect(&self.pgwire_conn_str(), tokio_postgres::NoTls)
-                .await
-                .expect("connect for query");
-        let conn_handle = tokio::spawn(async move {
-            let _ = connection.await;
-        });
-        let messages = client.simple_query(sql).await.expect("query");
-        drop(client);
-        let _ = conn_handle.await;
+        let messages = self.simple_query_ready(sql).await;
         messages
             .iter()
             .filter_map(|m| match m {
@@ -278,16 +298,7 @@ impl CrashHarness {
     /// `COUNT(*) AS n` — does not surface a usable column name through the
     /// pgwire row description, so aggregates must be read by index.
     pub async fn query_col_idx(&self, sql: &str, idx: usize) -> Vec<String> {
-        let (client, connection) =
-            tokio_postgres::connect(&self.pgwire_conn_str(), tokio_postgres::NoTls)
-                .await
-                .expect("connect for query");
-        let conn_handle = tokio::spawn(async move {
-            let _ = connection.await;
-        });
-        let messages = client.simple_query(sql).await.expect("query");
-        drop(client);
-        let _ = conn_handle.await;
+        let messages = self.simple_query_ready(sql).await;
         messages
             .iter()
             .filter_map(|m| match m {

nodedb/tests/graph_drop_hides_edges.rs

@@ -105,7 +105,10 @@ async fn soft_drop_named_collection_stats_still_deactivated() {
 
 /// `DROP ... PURGE` must still fully remove edges and stats — the hard-purge
 /// path is untouched by this fix and must keep working exactly as before.
-#[tokio::test]
+///
+/// Multi-threaded runtime: the single-node purge path reclaims storage inline
+/// via `block_in_place`, which panics on the current-thread runtime.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
 async fn hard_purge_removes_edges_and_stats() {
     let srv = TestServer::start().await;
     srv.exec("CREATE COLLECTION g_hard_purge").await.unwrap();