Add POST /v1/namespaces/:ns/integrity-check admin endpoint

Runs PRAGMA quick_check (or integrity_check if {full:true}) on a
namespace's live data file. Lets callers (cloud-sync-streamer,
operators) classify the failure mode BEFORE attempting recovery:

  ok:true  -> live DB is fine; corruption must be in wallog/snapshots
              (Mode A) -> use reset-replication (~1s recovery).
  ok:false -> live DB itself is corrupt (Mode B) -> reset-replication
              would propagate the corruption; fall back to delete +
              bulk-import.

Implementation detail: SQLite surfaces severe corruption as
prepare/connect errors rather than PRAGMA rows. The endpoint normalizes
those into the same {ok:false, message:...} response shape so the caller
gets a uniform classification signal (HTTP 200) rather than a server
error (HTTP 500).

- libsql-server/src/namespace/mod.rs: Namespace::integrity_check()
- libsql-server/src/namespace/store.rs: NamespaceStore::integrity_check()
- libsql-server/src/http/admin/mod.rs: handler + route

Verified with /tmp/test_integrity_check.sh against a healthy namespace
(ok), a namespace with a poisoned data file (Mode B detected), and a
non-existent namespace (proper 404).

Author: sle-c

libsql-server/src/http/admin/mod.rs

@@ -162,6 +162,10 @@ where
             "/v1/namespaces/:namespace/reset-replication",
             post(handle_reset_replication),
         )
+        .route(
+            "/v1/namespaces/:namespace/integrity-check",
+            post(handle_integrity_check),
+        )
         .route("/v1/namespaces/:namespace", delete(handle_delete_namespace))
         .route("/v1/namespaces/:namespace/stats", get(stats::handle_stats))
         .route(
@@ -576,6 +580,53 @@ async fn handle_reset_replication<C>(
     Ok(())
 }
 
+#[derive(serde::Deserialize, Default)]
+struct IntegrityCheckReq {
+    /// If true, run full `PRAGMA integrity_check` (O(DB size), thorough).
+    /// Default is `PRAGMA quick_check` which is fast and catches the
+    /// critical corruption classes.
+    #[serde(default)]
+    full: bool,
+}
+
+#[derive(serde::Serialize)]
+struct IntegrityCheckResp {
+    ok: bool,
+    /// Raw SQLite diagnostic text. `"ok"` on success, otherwise one or
+    /// more messages describing integrity issues.
+    message: String,
+    /// "quick" or "full", mirrors the `full` request field.
+    check: &'static str,
+}
+
+/// Run `PRAGMA quick_check` (default) or `PRAGMA integrity_check` on a
+/// namespace's live data file without touching other namespaces.
+///
+/// Use this to classify the failure mode before recovery:
+/// - `ok` → live DB is fine, any corruption is in wallog/snapshots (Mode A)
+/// → caller should use `POST /v1/namespaces/:ns/reset-replication`.
+/// - non-"ok" → live DB itself is corrupt (Mode B)
+/// → caller should restore from backup, not reset-replication.
+///
+/// Cheap: ~10ms for quick_check on small-to-medium namespaces.
+async fn handle_integrity_check<C>(
+    State(app_state): State<Arc<AppState<C>>>,
+    Path(namespace): Path<NamespaceName>,
+    payload: Option<Json<IntegrityCheckReq>>,
+) -> crate::Result<Json<IntegrityCheckResp>> {
+    let full = payload.map(|p| p.0.full).unwrap_or(false);
+    let message = app_state
+        .namespaces
+        .integrity_check(namespace, full)
+        .await?;
+    let ok = message.trim() == "ok";
+    Ok(Json(IntegrityCheckResp {
+        ok,
+        message,
+        check: if full { "full" } else { "quick" },
+    }))
+}
+
 #[derive(serde::Deserialize)]
 struct EnableHeapProfileRequest {
     #[serde(default)]

libsql-server/src/namespace/mod.rs

@@ -91,6 +91,48 @@ impl Namespace {
         Ok(())
     }
 
+    /// Run `PRAGMA quick_check` (or `integrity_check` if `full=true`) on
+    /// the namespace's live DB file and return the result string.
+    ///
+    /// For a healthy DB this returns `"ok"`. Anything else is an
+    /// integrity diagnostic message from SQLite.
+    ///
+    /// A catastrophically corrupt DB can fail before PRAGMA runs (e.g.
+    /// `malformed database schema` raised while the prepared statement
+    /// is parsing the schema). We normalize that into the same
+    /// `Ok(String)` return path so callers get a uniform classification
+    /// signal instead of a server error.
+    async fn integrity_check(&self, full: bool) -> anyhow::Result<String> {
+        // Even creating a connection can fail ("malformed database schema")
+        // when the DB is badly corrupt — that IS an integrity signal so we
+        // surface it as `Ok(String)` rather than an Err that becomes a 500.
+        let conn = match self.db.connection_maker().create().await {
+            Ok(c) => c,
+            Err(e) => {
+                return Ok(format!("connection failed: {e}"));
+            }
+        };
+        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)
+        });
+        match result {
+            Ok(rows) => Ok(rows.join("\n")),
+            Err(e) => {
+                // SQLite surfaces integrity failures as prepare/query errors
+                // rather than PRAGMA rows. Treat those as integrity signals.
+                Ok(format!("{e}"))
+            }
+        }
+    }
+
     async fn shutdown(mut self, should_checkpoint: bool) -> anyhow::Result<()> {
         self.tasks.shutdown().await;
         if should_checkpoint {

libsql-server/src/namespace/store.rs

@@ -153,6 +153,44 @@ impl NamespaceStore {
         Ok(())
     }
 
+    /// Run `PRAGMA quick_check` (or `integrity_check` if `full=true`) on
+    /// the namespace's live DB and return the result string. Takes only
+    /// a read lock on the namespace; other operations (including other
+    /// namespaces on this pod) are unaffected.
+    ///
+    /// A healthy DB returns `"ok"`. Any other text is diagnostic output
+    /// from SQLite. Use this to classify Mode A (wallog/snapshot
+    /// corruption, live DB OK) vs Mode B (live DB corruption) before
+    /// deciding on a recovery procedure.
+    pub async fn integrity_check(
+        &self,
+        namespace: NamespaceName,
+        full: bool,
+    ) -> crate::Result<String> {
+        if !self.inner.metadata.exists(&namespace).await {
+            return Err(Error::NamespaceDoesntExist(namespace.to_string()));
+        }
+        // Force-load the namespace so we can run a query against it.
+        let db_config = self.inner.metadata.handle(namespace.clone()).await;
+        let _ = self
+            .load_namespace(&namespace, db_config, RestoreOption::Latest)
+            .await?;
+
+        let entry = self
+            .inner
+            .store
+            .get_with(namespace.clone(), async { Default::default() })
+            .await;
+        let lock = entry.read().await;
+        if let Some(ns) = &*lock {
+            ns.integrity_check(full)
+                .await
+                .map_err(|e| Error::Internal(format!("integrity_check: {e}")))
+        } else {
+            Err(Error::NamespaceDoesntExist(namespace.to_string()))
+        }
+    }
+
     pub async fn reset(
         &self,
         namespace: NamespaceName,