Address PR 2230 review: non-blocking integrity_check + replica guard

Three fixes surfaced in code-review of PR 2230:

1. integrity_check (namespace/mod.rs):
   Move the PRAGMA quick_check/integrity_check PRAGMA run from the
   current Tokio worker onto the blocking thread pool via
   spawn_blocking. On a large DB, integrity_check can take seconds
   and would otherwise stall async tasks on that worker. Matches the
   pattern already used for checkpoint() and vacuum_if_needed() on
   LegacyConnection and for block_in_place in admin_shell.

2. reset_replication replica guard (namespace/store.rs):
   Reject calls with 400 NotAPrimary unless db_kind is Primary. On a
   replica, the ReplicaConfigurator does not consume .sentinel, so
   the re-init path would not actually rebuild anything and we'd
   destroy wallog/snapshots the replica still needs.

3. Async filesystem check (namespace/store.rs):
   Replace sync Path::exists() with tokio::fs::try_exists to avoid
   blocking the Tokio worker on a filesystem stat.

Also leaves a TODO comment on the string-based corruption
classification in the checkpoint error path — follow-up tracked to
replace it with typed rusqlite::ErrorCode matching (or a dedicated
Error::DatabaseCorrupt variant) once the Error enum plumbing is
updated.

Verified: cargo check clean, all 7 reset/integrity turmoil tests
pass (reset_replication_preserves_data_on_healthy_namespace,
reset_replication_on_nonexistent_namespace_returns_404,
reset_replication_is_idempotent,
reset_replication_response_includes_elapsed_ms,
integrity_check_on_healthy_namespace,
integrity_check_defaults_to_quick_when_full_omitted,
integrity_check_on_nonexistent_namespace_returns_404).

Author: sle-c

libsql-server/src/namespace/mod.rs

@@ -113,16 +113,27 @@ impl Namespace {
             }
         };
         let pragma = if full { "integrity_check" } else { "quick_check" };
-        let result = conn.with_raw(move |raw| -> rusqlite::Result<Vec<String>> {
-            let mut stmt = raw.prepare(&format!("PRAGMA {pragma}"))?;
-            let mut rows = stmt.query([])?;
-            let mut out = Vec::new();
-            while let Some(row) = rows.next()? {
-                let s: String = row.get(0)?;
-                out.push(s);
-            }
-            Ok(out)
-        });
+        // `with_raw` holds a parking_lot mutex and runs the closure on the
+        // current thread. `PRAGMA integrity_check` scans the full database
+        // and can take seconds on large DBs, which would stall a Tokio
+        // worker. Move the work to the blocking thread pool so Tokio's
+        // async workers keep making progress (consistent with the
+        // `checkpoint` / `vacuum_if_needed` paths on `LegacyConnection`).
+        let pragma_owned = pragma.to_string();
+        let result = tokio::task::spawn_blocking(move || {
+            conn.with_raw(move |raw| -> rusqlite::Result<Vec<String>> {
+                let mut stmt = raw.prepare(&format!("PRAGMA {pragma_owned}"))?;
+                let mut rows = stmt.query([])?;
+                let mut out = Vec::new();
+                while let Some(row) = rows.next()? {
+                    let s: String = row.get(0)?;
+                    out.push(s);
+                }
+                Ok(out)
+            })
+        })
+        .await
+        .map_err(|e| anyhow::anyhow!("integrity_check join failure: {e}"))?;
         match result {
             Ok(rows) => Ok(rows.join("\n")),
             Err(e) => {

libsql-server/src/namespace/store.rs

@@ -267,6 +267,16 @@ impl NamespaceStore {
             return Err(Error::NamespaceStoreShutdown);
         }
 
+        // reset-replication rebuilds the primary's replication log from
+        // its live data file. On a replica node the `.sentinel` sentinel
+        // is not consumed by `ReplicaConfigurator`, so the re-init path
+        // does nothing useful and we'd only destroy wallog/snapshots that
+        // the replica still needs. Reject the call with 400 NotAPrimary
+        // instead of producing confusing behavior.
+        if !matches!(self.inner.db_kind, DatabaseKind::Primary) {
+            return Err(Error::NotAPrimary);
+        }
+
         if !self.inner.metadata.exists(&namespace).await {
             return Err(Error::NamespaceDoesntExist(namespace.to_string()));
         }
@@ -308,6 +318,16 @@ impl NamespaceStore {
             match ns.checkpoint().await {
                 Ok(()) => {}
                 Err(e) => {
+                    // TODO(follow-up): classify corruption via a typed
+                    // `rusqlite::ErrorCode` (DatabaseCorrupt / NotADatabase)
+                    // or a dedicated `Error::DatabaseCorrupt` variant. See
+                    // <https://github.com/tursodatabase/libsql/pull/2230>
+                    // review thread for the design. The current
+                    // `checkpoint()` path boxes through `anyhow::Error`
+                    // which erases the source type, so we match the
+                    // rendered message for now. False-positive risk here
+                    // is low: all substrings below come from SQLite's own
+                    // error text for genuine live-DB corruption.
                     let msg = e.to_string();
                     let is_live_db_corrupt = msg.contains("malformed")
                         || msg.contains("DatabaseCorrupt")
@@ -370,8 +390,11 @@ impl NamespaceStore {
         let sentinel = ns_path.join(".sentinel");
         let _ = tokio::fs::remove_file(&sentinel).await;
         // Ensure parent dir exists (it should, because data still lives
-        // there, but be defensive).
-        if !ns_path.exists() {
+        // there, but be defensive). Use async `try_exists` rather than the
+        // sync `Path::exists()` so we don't block the Tokio worker on a
+        // filesystem stat.
+        let parent_exists = tokio::fs::try_exists(&ns_path).await.unwrap_or(false);
+        if !parent_exists {
             tokio::fs::create_dir_all(&ns_path).await.map_err(|e| {
                 Error::Internal(format!(
                     "reset_replication: create_dir_all {} failed: {e}",