fix(whatsapp_data): recover from SQLite corruption instead of re-flooding Sentry (#4733) (#4740)

src/core/observability.rs

@@ -151,6 +151,23 @@ pub enum ExpectedErrorKind {
     /// SQLite lock text, so unrelated DB lock errors in other domains still
     /// reach Sentry.
     WhatsAppDataSqliteBusy,
+    /// WhatsApp structured-ingest write hit a `SQLITE_CORRUPT` malformed on-disk
+    /// image ("database disk image is malformed" / "file is not a database").
+    ///
+    /// This is **defense-in-depth after** the store's own quarantine + rebuild
+    /// recovery (`whatsapp_data::store::WhatsAppDataStore::recover_corrupt_db`),
+    /// never instead of it: the store detects the corrupt image, reports it to
+    /// Sentry exactly once (process-wide latch), quarantines the damaged file,
+    /// and rebuilds an empty schema so ingest resumes. This classifier only
+    /// demotes the residual noise that can leak in the narrow window between
+    /// detection and a successful rebuild, or when a rebuild keeps failing on a
+    /// wedged host filesystem the app can't fix. Without both, one corrupt file
+    /// re-pages on every 2–30s scan tick (Sentry TAURI-RUST-KNH: 1,813
+    /// escalating events from a single host).
+    ///
+    /// Anchored to the whatsapp ingest failure envelope plus the malformed-image
+    /// text, so unrelated corruption in other domains still reaches Sentry.
+    WhatsAppDataSqliteCorrupt,
     /// Host disk is full — the filesystem returned `ENOSPC` to a write,
     /// `mkdir`, or `open` syscall. The user cannot recover from this without
     /// freeing space on their machine, and Sentry has no remediation path
@@ -588,6 +605,12 @@ pub fn expected_error_kind(message: &str) -> Option<ExpectedErrorKind> {
     if is_memory_store_breaker_open(&lower) {
         return Some(ExpectedErrorKind::MemoryStoreBreakerOpen);
     }
+    // Corruption is checked before the busy matcher: the two envelopes are
+    // mutually exclusive by their SQLite text (malformed-image vs locked), but
+    // ordering keeps the more-specific on-disk-damage signal unambiguous.
+    if is_whatsapp_data_sqlite_corrupt_message(&lower) {
+        return Some(ExpectedErrorKind::WhatsAppDataSqliteCorrupt);
+    }
     if is_whatsapp_data_sqlite_busy_message(&lower) {
         return Some(ExpectedErrorKind::WhatsAppDataSqliteBusy);
     }
@@ -791,6 +814,39 @@ fn is_whatsapp_data_sqlite_busy_message(lower: &str) -> bool {
         || lower.contains("error code 5")
 }
 
+/// Match whatsapp structured-ingest failures caused by a `SQLITE_CORRUPT`
+/// malformed on-disk image. Scoped to the whatsapp ingest envelope plus an
+/// upsert write frame, so unrelated malformed-image errors in other domains
+/// still reach Sentry.
+///
+/// This is defense-in-depth **after** the store's quarantine + rebuild recovery
+/// (see [`ExpectedErrorKind::WhatsAppDataSqliteCorrupt`]) — the store already
+/// reports the first hit once and rebuilds the DB; this only demotes residual
+/// noise. The `upsert wa_chat` / `upsert wa_message` frames are matched because
+/// the observed Sentry symptom (TAURI-RUST-KNH) fired on the
+/// `upsert wa_chat <jid>@lid` path; the `prune` frame is matched because the
+/// same ingest RPC also runs the 90-day auto-prune under the same corrupt-DB
+/// recovery wrapper, and a malformed image surfaced from the prune step (its
+/// error context carries the word `prune`, e.g. `prune old wa_messages`) would
+/// otherwise reach Sentry unfiltered. All three still require the
+/// `[whatsapp_data] ingest failed:` envelope so unrelated malformed-image
+/// errors in other domains keep paging.
+fn is_whatsapp_data_sqlite_corrupt_message(lower: &str) -> bool {
+    if !lower.contains("[whatsapp_data] ingest failed:") {
+        return false;
+    }
+    let has_write_frame = lower.contains("upsert wa_message")
+        || lower.contains("upsert wa_chat")
+        || lower.contains("prune");
+    if !has_write_frame {
+        return false;
+    }
+    lower.contains("disk image is malformed")
+        || lower.contains("file is not a database")
+        || lower.contains("error code 11")
+        || lower.contains("error code 26")
+}
+
 /// Match subconscious-engine SQLite schema-init failures caused by the host
 /// filesystem being unable to open the DB file (`SQLITE_CANTOPEN` /
 /// `SQLITE_IOERR_SHMMAP`). Anchored to the subconscious open/DDL envelope so it
@@ -2099,6 +2155,14 @@ fn report_expected_message(kind: ExpectedErrorKind, message: &str, domain: &str,
                 "[observability] {domain}.{operation} skipped expected whatsapp_data sqlite busy/locked error"
             );
         }
+        ExpectedErrorKind::WhatsAppDataSqliteCorrupt => {
+            tracing::warn!(
+                domain = domain,
+                operation = operation,
+                kind = "whatsapp_data_sqlite_corrupt",
+                "[observability] {domain}.{operation} skipped expected whatsapp_data sqlite corrupt error (store quarantines + rebuilds the DB and reports once)"
+            );
+        }
         ExpectedErrorKind::FilesystemUserPathInvalid => {
             // User-input validation failure surfaced at the RPC
             // boundary — e.g. `openhuman.vault_create` called with a
@@ -4332,6 +4396,81 @@ mod tests {
         }
     }
 
+    #[test]
+    fn classifies_whatsapp_data_sqlite_corrupt_errors() {
+        for raw in [
+            // The observed TAURI-RUST-KNH symptom: corruption on the wa_chat path.
+            r#"[whatsapp_data] ingest failed: upsert wa_chat 207897942335683@lid: database disk image is malformed: Error code 11: The database disk image is malformed"#,
+            // The wa_message path — same on-disk damage class.
+            r#"[whatsapp_data] ingest failed: upsert wa_message chat=120363402402350155@g.us msg=abc: file is not a database: Error code 26: File opened that is not a database file"#,
+            // Wrapped in outer RPC context — classifier runs on the full chain.
+            r#"rpc.invoke_method failed: [whatsapp_data] ingest failed: upsert wa_chat 207897942335683@lid: database disk image is malformed"#,
+        ] {
+            assert_eq!(
+                expected_error_kind(raw),
+                Some(ExpectedErrorKind::WhatsAppDataSqliteCorrupt),
+                "should classify whatsapp_data sqlite corrupt: {raw}"
+            );
+        }
+    }
+
+    #[test]
+    fn classifies_whatsapp_data_prune_path_sqlite_corrupt_errors() {
+        // The same ingest RPC runs the 90-day auto-prune under the store's
+        // corrupt-DB recovery wrapper. A malformed image surfaced from the prune
+        // step carries a `prune` frame (e.g. `prune old wa_messages`) instead of
+        // an `upsert` frame, and must still be demoted — otherwise a boot-time
+        // prune corruption re-floods Sentry on every scan tick (Finding 3).
+        for raw in [
+            r#"[whatsapp_data] ingest failed: prune old wa_messages: database disk image is malformed: Error code 11: The database disk image is malformed"#,
+            r#"[whatsapp_data] ingest failed: prune old wa_messages: scan affected chats: database disk image is malformed"#,
+            r#"[whatsapp_data] ingest failed: refresh chat stats after prune: chat@c.us: file is not a database: Error code 26"#,
+            r#"rpc.invoke_method failed: [whatsapp_data] ingest failed: prune old wa_messages: database disk image is malformed"#,
+        ] {
+            assert_eq!(
+                expected_error_kind(raw),
+                Some(ExpectedErrorKind::WhatsAppDataSqliteCorrupt),
+                "should classify whatsapp_data prune-path sqlite corrupt: {raw}"
+            );
+        }
+    }
+
+    #[test]
+    fn does_not_classify_unrelated_prune_errors_as_whatsapp_corrupt() {
+        for raw in [
+            // A `prune` word outside the whatsapp ingest envelope must still page.
+            "memory queue prune failed: database disk image is malformed",
+            // whatsapp prune-path lock contention is the *busy* bucket, not corrupt.
+            "[whatsapp_data] ingest failed: prune old wa_messages: database is locked",
+            // whatsapp prune-path constraint/logic error is a real bug, not on-disk damage.
+            "[whatsapp_data] ingest failed: prune old wa_messages: no such table: wa_messages",
+        ] {
+            assert_ne!(
+                expected_error_kind(raw),
+                Some(ExpectedErrorKind::WhatsAppDataSqliteCorrupt),
+                "must not classify as whatsapp_data sqlite corrupt: {raw}"
+            );
+        }
+    }
+
+    #[test]
+    fn does_not_classify_unrelated_corrupt_messages_as_whatsapp_corrupt() {
+        for raw in [
+            // Malformed image outside the whatsapp ingest envelope must still page.
+            "memory queue write failed: database disk image is malformed",
+            // Read-path whatsapp failure (no upsert frame) is not the ingest write.
+            "[whatsapp_data] list_messages failed: database disk image is malformed",
+            // whatsapp ingest lock contention is the *busy* bucket, not corrupt.
+            "[whatsapp_data] ingest failed: upsert wa_message chat=x msg=y: database is locked",
+        ] {
+            assert_ne!(
+                expected_error_kind(raw),
+                Some(ExpectedErrorKind::WhatsAppDataSqliteCorrupt),
+                "must not classify as whatsapp_data sqlite corrupt: {raw}"
+            );
+        }
+    }
+
     #[test]
     fn classifies_subconscious_schema_unavailable_errors() {
         for raw in [

src/openhuman/whatsapp_data/sqlite_retry.rs

@@ -7,7 +7,9 @@
 use std::thread;
 use std::time::Duration;
 
-use anyhow::{Context, Result};
+// `anyhow::Error::context` (used in `retry_on_sqlite_busy`) is the inherent
+// method on `Error`, not the `Context` trait, so only `Result` is imported.
+use anyhow::Result;
 
 /// Per-connection busy handler window (issue #2077).
 pub const BUSY_TIMEOUT: Duration = Duration::from_millis(5000);
@@ -30,6 +32,40 @@ pub fn is_sqlite_busy(err: &anyhow::Error) -> bool {
     msg.contains("database is locked") || msg.contains("database table is locked")
 }
 
+/// Returns true when `err` is a `SQLITE_CORRUPT` malformed-image condition
+/// (primary code `DatabaseCorrupt`, code 11) or the closely-related
+/// `NotADatabase` (code 26 — the header itself is unreadable).
+///
+/// Unlike [`is_sqlite_busy`], a malformed on-disk image is **persistent
+/// damage**: the upsert can never succeed, so every scan poll (2–30s)
+/// re-opens the dead file, re-hits the error, and re-reports to Sentry —
+/// turning one corrupt file into an escalating flood (Sentry TAURI-RUST-KNH:
+/// 1,813 events from a single host). Detecting it here is what lets the store
+/// drive its quarantine + rebuild recovery
+/// ([`WhatsAppDataStore::recover_corrupt_db`](crate::openhuman::whatsapp_data::store::WhatsAppDataStore))
+/// instead of retrying forever.
+///
+/// Matching on the error **code** is rusqlite-version-stable. The text
+/// fallback covers the case where the rusqlite error was flattened to a plain
+/// `anyhow!` string across `.context()` layers — SQLite renders these as
+/// "database disk image is malformed" (code 11) and "file is not a database"
+/// (code 26). The `"disk image is malformed"` substring matches both the bare
+/// and `Error code 11: The database disk image is malformed` envelope shapes.
+pub fn is_sqlite_corrupt(err: &anyhow::Error) -> bool {
+    if let Some(rusqlite::Error::SqliteFailure(sqlite_err, _)) =
+        err.downcast_ref::<rusqlite::Error>()
+    {
+        if matches!(
+            sqlite_err.code,
+            rusqlite::ErrorCode::DatabaseCorrupt | rusqlite::ErrorCode::NotADatabase
+        ) {
+            return true;
+        }
+    }
+    let msg = format!("{err:#}").to_ascii_lowercase();
+    msg.contains("disk image is malformed") || msg.contains("file is not a database")
+}
+
 /// Run `f` up to [`WRITE_RETRY_ATTEMPTS`] times when SQLite reports busy/locked.
 pub fn retry_on_sqlite_busy<T, F>(op_name: &str, mut f: F) -> Result<T>
 where
@@ -101,6 +137,87 @@ mod tests {
         assert!(!is_sqlite_busy(&err));
     }
 
+    #[test]
+    fn is_sqlite_corrupt_matches_database_corrupt_code() {
+        let raw = rusqlite::Error::SqliteFailure(
+            rusqlite::ffi::Error {
+                code: rusqlite::ErrorCode::DatabaseCorrupt,
+                extended_code: 11,
+            },
+            Some("database disk image is malformed".into()),
+        );
+        assert!(is_sqlite_corrupt(&anyhow::Error::from(raw)));
+    }
+
+    #[test]
+    fn is_sqlite_corrupt_matches_not_a_database_code() {
+        let raw = rusqlite::Error::SqliteFailure(
+            rusqlite::ffi::Error {
+                code: rusqlite::ErrorCode::NotADatabase,
+                extended_code: 26,
+            },
+            Some("file is not a database".into()),
+        );
+        assert!(is_sqlite_corrupt(&anyhow::Error::from(raw)));
+    }
+
+    /// The rusqlite error sits under the `[whatsapp_data] ingest failed:` +
+    /// `upsert wa_chat …` context layers when it bubbles out of the store; the
+    /// downcast must still find the `DatabaseCorrupt` code (regression guard:
+    /// don't rely on the top-level error type).
+    #[test]
+    fn is_sqlite_corrupt_matches_through_context_layers() {
+        let raw = rusqlite::Error::SqliteFailure(
+            rusqlite::ffi::Error {
+                code: rusqlite::ErrorCode::DatabaseCorrupt,
+                extended_code: 11,
+            },
+            Some("database disk image is malformed".into()),
+        );
+        let wrapped = anyhow::Error::from(raw)
+            .context("upsert wa_chat 207897942335683@lid")
+            .context("[whatsapp_data] ingest failed");
+        assert!(is_sqlite_corrupt(&wrapped));
+    }
+
+    /// Text fallback: the exact flattened Sentry string (TAURI-RUST-KNH) must
+    /// classify even when no rusqlite error is available to downcast.
+    #[test]
+    fn is_sqlite_corrupt_text_fallback() {
+        let err = anyhow::anyhow!(
+            "[whatsapp_data] ingest failed: upsert wa_chat 207897942335683@lid: \
+             database disk image is malformed: Error code 11: The database disk image is malformed"
+        );
+        assert!(is_sqlite_corrupt(&err));
+    }
+
+    /// Busy/locked and constraint violations must NOT be swallowed as
+    /// corruption — quarantining on those would destroy a perfectly good DB.
+    #[test]
+    fn is_sqlite_corrupt_does_not_match_busy_or_constraint() {
+        let busy = rusqlite::Error::SqliteFailure(
+            rusqlite::ffi::Error {
+                code: rusqlite::ErrorCode::DatabaseBusy,
+                extended_code: 5,
+            },
+            Some("database is locked".into()),
+        );
+        assert!(!is_sqlite_corrupt(&anyhow::Error::from(busy)));
+
+        let constraint = rusqlite::Error::SqliteFailure(
+            rusqlite::ffi::Error {
+                code: rusqlite::ErrorCode::ConstraintViolation,
+                extended_code: 19,
+            },
+            Some("UNIQUE constraint failed".into()),
+        );
+        assert!(!is_sqlite_corrupt(&anyhow::Error::from(constraint)));
+
+        assert!(!is_sqlite_corrupt(&anyhow::anyhow!(
+            "upstream returned 500: internal server error"
+        )));
+    }
+
     #[test]
     fn retry_on_sqlite_busy_succeeds_after_transient_busy() {
         let mut calls = 0u32;

src/openhuman/whatsapp_data/store.rs

@@ -6,18 +6,44 @@
 //! This store is local-only; no data is transmitted to external services.
 
 use std::collections::HashMap;
-use std::path::Path;
+use std::path::{Path, PathBuf};
+use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::Mutex;
 
 use anyhow::{Context, Result};
 use rusqlite::{params, Connection};
 
-use crate::openhuman::whatsapp_data::sqlite_retry::{retry_on_sqlite_busy, BUSY_TIMEOUT};
+use crate::openhuman::whatsapp_data::sqlite_retry::{
+    is_sqlite_corrupt, retry_on_sqlite_busy, BUSY_TIMEOUT,
+};
 use crate::openhuman::whatsapp_data::types::{
     ChatMeta, IngestMessage, ListChatsRequest, ListMessagesRequest, SearchMessagesRequest,
     WhatsAppChat, WhatsAppMessage,
 };
 
+/// Process-wide latch so a `SQLITE_CORRUPT` flood is reported to Sentry
+/// **once**, not on every scan tick. The whatsapp_scanner re-ingests every
+/// 2–30s, so a wedged DB re-hits the malformed-image error on each poll — one
+/// corrupt file produced 1,813 escalating Sentry events from a single host
+/// (TAURI-RUST-KNH). Set on the first detection; cleared once a recovery
+/// attempt settles (quarantine + rebuild, or a quick_check that now passes) so
+/// a genuinely-new, later corruption can still page exactly once.
+static CORRUPT_REPORTED: AtomicBool = AtomicBool::new(false);
+
+/// Test-only override: when set, [`WhatsAppDataStore::integrity_check_ok`]
+/// reports the (rebuilt) DB as failing its integrity check. This is the seam
+/// that lets tests drive the "rebuild still fails integrity_check" branch of
+/// [`WhatsAppDataStore::recover_corrupt_db`] without forging a file that both
+/// survives quarantine and yet fails `PRAGMA integrity_check`.
+#[cfg(test)]
+static FORCE_INTEGRITY_CHECK_FAIL: AtomicBool = AtomicBool::new(false);
+
+/// Test-only serialization guard for the recovery-episode tests. `CORRUPT_REPORTED`
+/// and `FORCE_INTEGRITY_CHECK_FAIL` are process-wide, so tests that mutate them
+/// must not run concurrently or they clobber each other's latch observations.
+#[cfg(test)]
+static CORRUPT_TEST_GUARD: Mutex<()> = Mutex::new(());
+
 /// SQLite-backed store for WhatsApp chats and messages.
 pub struct WhatsAppDataStore {
     db_path: std::path::PathBuf,
@@ -40,10 +66,49 @@ impl WhatsAppDataStore {
             db_path,
             write_lock: Mutex::new(()),
         };
-        store.init_schema()?;
+        store.init_schema_or_recover()?;
         Ok(store)
     }
 
+    /// Initialize the schema, self-healing a DB that is already corrupt **at
+    /// process startup**.
+    ///
+    /// Without this, a `whatsapp_data.db` that is malformed before the first
+    /// write RPC arrives makes `init_schema()` fail, `global::init` leaves the
+    /// store singleton unset, and every subsequent ingest RPC fails with
+    /// "store accessed before init" — the corruption never reaches the
+    /// quarantine + rebuild path (which only guards the *write* wrapper), so it
+    /// survives restarts and re-pages Sentry on every scanner tick. Recovering
+    /// here makes a boot-time corrupt DB heal exactly as a mid-run one does.
+    ///
+    /// The `CORRUPT_REPORTED` latch semantics are reused verbatim via
+    /// [`Self::report_and_recover`] (report once per episode, reset only on a
+    /// confirmed-healthy rebuild), so a boot-time corruption pages at most once.
+    fn init_schema_or_recover(&self) -> Result<()> {
+        match self.init_schema() {
+            Ok(()) => Ok(()),
+            Err(e) if is_sqlite_corrupt(&e) => {
+                log::error!(
+                    "[whatsapp_data] init_schema hit SQLITE_CORRUPT at startup for {} — \
+                     driving quarantine + rebuild before ingest can begin: {e:#}",
+                    self.db_path.display()
+                );
+                // Reports once (latch), quarantines the malformed image, and
+                // rebuilds the schema. `recover_corrupt_db` is safe to call on
+                // the just-constructed store: it only needs `db_path` +
+                // `init_schema`, both available before the store is published.
+                self.report_and_recover("init_schema", &e);
+                // Confirm the store is now usable. If recovery failed, this
+                // re-init returns Err and `global::init` correctly leaves the
+                // singleton unset — but now only when the DB is genuinely
+                // unrecoverable, not merely corrupt-on-boot.
+                self.init_schema()
+                    .context("re-init whatsapp_data schema after boot-time corrupt-DB recovery")
+            }
+            Err(e) => Err(e),
+        }
+    }
+
     /// Initialize the schema. Idempotent — safe to call on every startup.
     fn init_schema(&self) -> Result<()> {
         let conn = self.open_conn()?;
@@ -84,6 +149,22 @@ impl WhatsAppDataStore {
         Ok(())
     }
 
+    /// Open a fresh connection to the DB file.
+    ///
+    /// Read paths (`list_chats` / `list_messages` / `search_messages`) call this
+    /// **without** taking `write_lock`, so a read can race the brief file-rename
+    /// window inside [`Self::recover_corrupt_db`] (which quarantines the corrupt
+    /// image then rebuilds under `write_lock`). We deliberately do NOT serialize
+    /// reads behind the write lock: the race is not a correctness or
+    /// data-integrity problem, only a rare transient read error that self-heals
+    /// on the next poll. Three outcomes are possible and all are safe —
+    /// (1) the read opens before the rename and reads valid (old) data via its
+    /// still-live fd; (2) it opens after the rebuild and reads the fresh schema;
+    /// (3) it opens in the sub-millisecond gap after the rename but before the
+    /// rebuild, gets an empty auto-created file, and returns a benign
+    /// "no such table" error the caller retries. Guarding this hot path with a
+    /// global read/write lock would trade that vanishingly-rare transient for
+    /// permanent read/write contention on every list/search — not worth it.
     fn open_conn(&self) -> Result<Connection> {
         let conn = Connection::open(&self.db_path)
             .with_context(|| format!("open whatsapp_data db: {}", self.db_path.display()))?;
@@ -115,6 +196,222 @@ impl WhatsAppDataStore {
             .unwrap_or(0)
     }
 
+    /// Run a write op through the busy-retry loop, and on a confirmed
+    /// `SQLITE_CORRUPT` malformed-image error drive a **once-per-call**
+    /// quarantine + rebuild recovery, then retry the op a single time against
+    /// the rebuilt schema.
+    ///
+    /// The recovery is bounded: at most one quarantine attempt and one retry
+    /// per public write call. A genuinely unrecoverable file therefore returns
+    /// the corrupt error to the caller after one recovery pass instead of
+    /// spinning — the process-wide report latch (`report_and_recover`) then
+    /// keeps Sentry from re-flooding on every scan tick.
+    fn write_with_corrupt_recovery<T>(
+        &self,
+        op_name: &str,
+        f: impl Fn() -> Result<T>,
+    ) -> Result<T> {
+        match retry_on_sqlite_busy(op_name, &f) {
+            Ok(val) => Ok(val),
+            Err(e) if is_sqlite_corrupt(&e) => {
+                self.report_and_recover(op_name, &e);
+                // Retry once against the (now rebuilt) DB. If this still fails,
+                // the error propagates — no further recovery, so a wedged
+                // filesystem can't loop.
+                retry_on_sqlite_busy(op_name, &f)
+            }
+            Err(e) => Err(e),
+        }
+    }
+
+    /// Report a confirmed `SQLITE_CORRUPT` failure to Sentry and drive the
+    /// quarantine + rebuild recovery. Factored out of
+    /// [`Self::write_with_corrupt_recovery`] so the report + recovery decision
+    /// is unit-testable without a live scanner loop.
+    fn report_and_recover(&self, op_name: &str, err: &anyhow::Error) {
+        // Report to Sentry at most once per corruption episode. Without this
+        // latch the scanner's 2–30s poll re-hits the wedged DB and re-pages on
+        // every tick (TAURI-RUST-KNH: 1,813 events from one host).
+        if !CORRUPT_REPORTED.swap(true, Ordering::Relaxed) {
+            crate::core::observability::report_error(
+                err,
+                "whatsapp_data",
+                "ingest_corrupt",
+                &[("op", op_name)],
+            );
+        }
+        log::error!(
+            "[whatsapp_data] {op_name} hit SQLITE_CORRUPT (malformed DB image), \
+             attempting quarantine + rebuild recovery: {err:#}"
+        );
+        match self.recover_corrupt_db() {
+            Ok(true) => {
+                log::warn!(
+                    "[whatsapp_data] {op_name} quarantined corrupt DB and rebuilt empty schema; \
+                     ingest will resume"
+                );
+                // Recovery settled — allow a future, genuinely-new corruption
+                // to page once more.
+                CORRUPT_REPORTED.store(false, Ordering::Relaxed);
+            }
+            Ok(false) => {
+                log::info!(
+                    "[whatsapp_data] {op_name} corruption recovery: integrity check now passes, \
+                     no quarantine needed"
+                );
+                CORRUPT_REPORTED.store(false, Ordering::Relaxed);
+            }
+            Err(rec_err) => log::error!(
+                "[whatsapp_data] {op_name} corruption recovery FAILED, ingest stays degraded: \
+                 {rec_err:#}"
+            ),
+        }
+    }
+
+    /// Recover from a `SQLITE_CORRUPT` (malformed image) on the whatsapp_data DB.
+    ///
+    /// A malformed on-disk image never heals on its own — every write fails
+    /// forever and the scanner re-pages Sentry on each poll (Sentry
+    /// TAURI-RUST-KNH: 1,813 events from a single host). This quarantines the
+    /// damaged file (and its WAL/SHM side-files) to a timestamped
+    /// `.corrupt-<ts>` copy — **preserved, not deleted**, so the bytes can be
+    /// inspected or salvaged — then rebuilds an empty schema so ingest resumes.
+    ///
+    /// Returns `Ok(true)` when a quarantine + rebuild happened, `Ok(false)`
+    /// when a fresh `PRAGMA quick_check` now passes (the earlier failure was
+    /// transient and quarantining would have destroyed good data), and `Err`
+    /// when the quarantine rename or the schema rebuild failed.
+    ///
+    /// The store opens a fresh `Connection` per call (no cached handle), so no
+    /// connection needs to be dropped before the rename — the next `open_conn`
+    /// naturally picks up the rebuilt file.
+    pub(crate) fn recover_corrupt_db(&self) -> Result<bool> {
+        // 1. Re-confirm corruption against the on-disk file. `quick_check` is
+        //    the cheap structural scan; if it now reports "ok" the image is
+        //    actually healthy (e.g. the original error was a transient mmap
+        //    fault) and we must NOT destroy good data — bail without quarantine.
+        if self.db_path.exists() {
+            match self.quick_check_ok() {
+                Ok(true) => {
+                    log::info!(
+                        "[whatsapp_data] quick_check passed for {} — no quarantine needed",
+                        self.db_path.display()
+                    );
+                    return Ok(false);
+                }
+                Ok(false) => {
+                    log::warn!(
+                        "[whatsapp_data] quick_check confirms corruption for {}, quarantining",
+                        self.db_path.display()
+                    );
+                }
+                Err(e) => {
+                    // The check couldn't even run (unopenable / unreadable
+                    // header). That is itself a malformed-image signal.
+                    log::warn!(
+                        "[whatsapp_data] quick_check could not run for {} ({e:#}); \
+                         treating as corrupt",
+                        self.db_path.display()
+                    );
+                }
+            }
+        } else {
+            log::warn!(
+                "[whatsapp_data] corrupt-recovery: {} is missing; rebuilding fresh schema",
+                self.db_path.display()
+            );
+        }
+
+        // 2. Quarantine the main DB + WAL/SHM side-files to `<name>.corrupt-<ts>`.
+        let ts = Self::now_secs();
+        let mut quarantined = 0usize;
+        for suffix in &["", "-wal", "-shm"] {
+            let src = with_name_suffix(&self.db_path, suffix);
+            if !src.exists() {
+                continue;
+            }
+            let dst = with_name_suffix(&src, &format!(".corrupt-{ts}"));
+            std::fs::rename(&src, &dst).with_context(|| {
+                format!(
+                    "quarantine corrupt whatsapp_data file {} -> {}",
+                    src.display(),
+                    dst.display()
+                )
+            })?;
+            log::warn!(
+                "[whatsapp_data] quarantined {} -> {}",
+                src.display(),
+                dst.display()
+            );
+            quarantined += 1;
+        }
+
+        // 3. Rebuild an empty schema by re-running init on a fresh file. The
+        //    damaged rows are not silently dropped — they live on in the
+        //    `.corrupt-<ts>` copy.
+        self.init_schema()
+            .context("rebuild whatsapp_data schema after quarantining corrupt DB")?;
+
+        // 4. Confirm the rebuilt image is structurally sound. This result is
+        //    load-bearing: `report_and_recover` only resets the process-wide
+        //    `CORRUPT_REPORTED` latch (and logs "ingest will resume") on
+        //    `Ok(true)`. If the rebuild still fails integrity_check — or the
+        //    check itself can't run — we MUST surface `Err`, otherwise the
+        //    latch resets, re-arming Sentry to page on the next scan tick and
+        //    breaking the report-once-per-episode guarantee this recovery
+        //    exists to protect.
+        match self.integrity_check_ok() {
+            Ok(true) => {
+                log::warn!(
+                    "[whatsapp_data] corruption recovery complete: quarantined {quarantined} file(s), \
+                     rebuilt empty schema, integrity_check=ok at {}",
+                    self.db_path.display()
+                );
+                Ok(true)
+            }
+            Ok(false) => {
+                log::error!(
+                    "[whatsapp_data] rebuilt DB still fails integrity_check at {}",
+                    self.db_path.display()
+                );
+                Err(anyhow::anyhow!(
+                    "rebuilt whatsapp_data db still fails integrity_check at {}",
+                    self.db_path.display()
+                ))
+            }
+            Err(e) => Err(e.context("integrity_check after rebuild could not run")),
+        }
+    }
+
+    /// Run `PRAGMA quick_check(1)` on a fresh, short-lived connection. Returns
+    /// `Ok(true)` when the structural scan reports `"ok"`, `Ok(false)` on any
+    /// reported corruption, and `Err` when the check itself can't run (file
+    /// unopenable / header unreadable — itself a corruption signal the caller
+    /// treats as malformed).
+    fn quick_check_ok(&self) -> Result<bool> {
+        let conn = Connection::open(&self.db_path)
+            .with_context(|| format!("open for quick_check: {}", self.db_path.display()))?;
+        let _ = conn.busy_timeout(BUSY_TIMEOUT);
+        let result: String = conn
+            .query_row("PRAGMA quick_check(1)", [], |row| row.get(0))
+            .context("running PRAGMA quick_check")?;
+        Ok(result.eq_ignore_ascii_case("ok"))
+    }
+
+    /// Run `PRAGMA integrity_check(1)` against the (rebuilt) DB to confirm it is
+    /// structurally sound. Returns `Ok(true)` when it reports `"ok"`.
+    fn integrity_check_ok(&self) -> Result<bool> {
+        #[cfg(test)]
+        if FORCE_INTEGRITY_CHECK_FAIL.load(Ordering::Relaxed) {
+            return Ok(false);
+        }
+        let conn = self.open_conn()?;
+        let result: String = conn
+            .query_row("PRAGMA integrity_check(1)", [], |row| row.get(0))
+            .context("running PRAGMA integrity_check")?;
+        Ok(result.eq_ignore_ascii_case("ok"))
+    }
+
     /// Upsert chat metadata rows.  Returns the number of rows inserted or updated.
     pub fn upsert_chats(
         &self,
@@ -128,7 +425,7 @@ impl WhatsAppDataStore {
             .write_lock
             .lock()
             .map_err(|e| anyhow::anyhow!("whatsapp_data write lock poisoned: {e}"))?;
-        retry_on_sqlite_busy("upsert_chats", || {
+        self.write_with_corrupt_recovery("upsert_chats", || {
             self.upsert_chats_inner(account_id, chats)
         })
     }
@@ -172,7 +469,7 @@ impl WhatsAppDataStore {
             .write_lock
             .lock()
             .map_err(|e| anyhow::anyhow!("whatsapp_data write lock poisoned: {e}"))?;
-        retry_on_sqlite_busy("upsert_messages", || {
+        self.write_with_corrupt_recovery("upsert_messages", || {
             self.upsert_messages_inner(account_id, msgs)
         })
     }
@@ -265,7 +562,7 @@ impl WhatsAppDataStore {
             .write_lock
             .lock()
             .map_err(|e| anyhow::anyhow!("whatsapp_data write lock poisoned: {e}"))?;
-        retry_on_sqlite_busy("prune_old_messages", || {
+        self.write_with_corrupt_recovery("prune_old_messages", || {
             self.prune_old_messages_inner(cutoff_ts)
         })
     }
@@ -275,10 +572,17 @@ impl WhatsAppDataStore {
         let now = Self::now_secs();
 
         // Collect affected (account_id, chat_id) pairs before deleting.
-        let mut stmt = conn.prepare(
-            "SELECT DISTINCT account_id, chat_id FROM wa_messages
-             WHERE timestamp > 0 AND timestamp < ?1",
-        )?;
+        // Contextualized so a `SQLITE_CORRUPT` surfaced while compiling this
+        // scan still carries a `prune` frame marker the observability
+        // classifier keys on (otherwise a boot-time prune corruption would
+        // reach Sentry unfiltered — the prepare of a plain SELECT still reads
+        // the schema/root page where damage commonly lives).
+        let mut stmt = conn
+            .prepare(
+                "SELECT DISTINCT account_id, chat_id FROM wa_messages
+                 WHERE timestamp > 0 AND timestamp < ?1",
+            )
+            .context("prune old wa_messages: scan affected chats")?;
         let affected: Vec<(String, String)> = stmt
             .query_map(params![cutoff_ts], |row| Ok((row.get(0)?, row.get(1)?)))?
             .collect::<rusqlite::Result<_>>()
@@ -501,6 +805,22 @@ impl WhatsAppDataStore {
     }
 }
 
+/// Append `suffix` to the *file name* of `path` (so `whatsapp_data.db` + `-wal`
+/// = `whatsapp_data.db-wal`, and `whatsapp_data.db` + `.corrupt-123` =
+/// `whatsapp_data.db.corrupt-123`). SQLite names its side-files this way (not
+/// as a new extension), and the quarantine keeps the corrupt image alongside
+/// the original for inspection.
+fn with_name_suffix(path: &Path, suffix: &str) -> PathBuf {
+    let mut p = path.to_path_buf();
+    let name = p
+        .file_name()
+        .unwrap_or_default()
+        .to_string_lossy()
+        .into_owned();
+    p.set_file_name(format!("{name}{suffix}"));
+    p
+}
+
 fn map_chat_row(row: &rusqlite::Row<'_>) -> rusqlite::Result<WhatsAppChat> {
     Ok(WhatsAppChat {
         account_id: row.get(0)?,

src/openhuman/whatsapp_data/store_tests.rs

@@ -462,6 +462,255 @@ fn open_conn_configures_busy_timeout_and_wal() {
 
 /// Messages with an empty message_id or chat_id must be silently skipped,
 /// never causing a panic or spurious database error.
+/// A malformed on-disk image must be detected on the upsert write path,
+/// quarantined (never deleted), the schema rebuilt, and the *same* upsert call
+/// must then SUCCEED against the fresh DB — proving ingest self-heals instead of
+/// re-hitting the dead file on every scan tick (Sentry TAURI-RUST-KNH: 1,813
+/// events from a single host). The process-wide report latch must reset after a
+/// successful recovery so it fires at most once per corruption episode.
+#[test]
+fn upsert_recovers_from_corrupt_database() {
+    use std::sync::atomic::Ordering;
+
+    // Serialize against the other latch-touching recovery tests: the report
+    // latch + integrity-fail seam are process-wide statics.
+    let _guard = super::CORRUPT_TEST_GUARD
+        .lock()
+        .unwrap_or_else(|e| e.into_inner());
+
+    let (store, tmp) = make_store();
+    let workspace = tmp.path().to_path_buf();
+    let db_path = db_path_for(&tmp);
+
+    // Reset the process-wide latch so this test observes a clean episode.
+    super::CORRUPT_REPORTED.store(false, Ordering::Relaxed);
+
+    // Corrupt the freshly-created DB: drop any WAL side-files, then overwrite the
+    // main file with garbage so the next open reads a malformed image.
+    for suffix in ["-wal", "-shm"] {
+        let side = db_path.with_file_name(format!("whatsapp_data.db{suffix}"));
+        let _ = std::fs::remove_file(&side);
+    }
+    std::fs::write(
+        &db_path,
+        b"this is not a sqlite database, just garbage bytes",
+    )
+    .unwrap();
+
+    let mut chats = HashMap::new();
+    chats.insert("chat@c.us".to_string(), chat_meta("Alice"));
+
+    // First upsert hits the malformed image → detect → quarantine → rebuild →
+    // retry against the fresh DB, so the call SUCCEEDS.
+    let count = store
+        .upsert_chats("acct1", &chats)
+        .expect("upsert must succeed after corrupt-DB recovery");
+    assert_eq!(count, 1);
+
+    // The corrupt bytes are preserved alongside, never silently dropped.
+    let quarantined: Vec<_> = std::fs::read_dir(db_path.parent().unwrap())
+        .unwrap()
+        .filter_map(|e| e.ok())
+        .filter(|e| {
+            e.file_name()
+                .to_string_lossy()
+                .contains("whatsapp_data.db.corrupt-")
+        })
+        .collect();
+    assert_eq!(
+        quarantined.len(),
+        1,
+        "exactly one quarantined copy of the corrupt image should exist"
+    );
+
+    // The rebuilt DB is healthy and queryable — the chat we just wrote is there.
+    let rows = store
+        .list_chats(&ListChatsRequest {
+            account_id: Some("acct1".to_string()),
+            limit: None,
+            offset: None,
+        })
+        .unwrap();
+    assert_eq!(rows.len(), 1);
+    assert_eq!(rows[0].chat_id, "chat@c.us");
+
+    // A fresh store over the same workspace also sees the rebuilt, healthy DB.
+    let reopened = WhatsAppDataStore::new(&workspace).expect("reopen store");
+    let rows2 = reopened
+        .list_chats(&ListChatsRequest {
+            account_id: Some("acct1".to_string()),
+            limit: None,
+            offset: None,
+        })
+        .unwrap();
+    assert_eq!(rows2.len(), 1);
+
+    // Recovery settled, so the report latch was reset: a genuinely-new later
+    // corruption can page once more (the report fired at most once this episode).
+    assert!(
+        !super::CORRUPT_REPORTED.load(Ordering::Relaxed),
+        "report latch must reset after a successful recovery"
+    );
+}
+
+/// Finding 1 regression: when the quarantine + rebuild leaves a DB that STILL
+/// fails its integrity check, `recover_corrupt_db` must return `Err` (not
+/// swallow it as `Ok(true)`). Consequently `report_and_recover` must NOT reset
+/// the process-wide report latch — preserving the report-once-per-episode
+/// guarantee (a spurious reset would re-arm Sentry to page on the next scan
+/// tick against a DB that never actually recovered).
+#[test]
+fn recover_corrupt_db_errors_and_keeps_latch_when_rebuild_fails_integrity() {
+    use std::sync::atomic::Ordering;
+
+    let _guard = super::CORRUPT_TEST_GUARD
+        .lock()
+        .unwrap_or_else(|e| e.into_inner());
+
+    let (store, tmp) = make_store();
+    let db_path = db_path_for(&tmp);
+
+    // Force the post-rebuild integrity_check to report failure via the test seam.
+    super::FORCE_INTEGRITY_CHECK_FAIL.store(true, Ordering::Relaxed);
+
+    let corrupt = |path: &std::path::Path| {
+        for suffix in ["-wal", "-shm"] {
+            let side = path.with_file_name(format!("whatsapp_data.db{suffix}"));
+            let _ = std::fs::remove_file(&side);
+        }
+        std::fs::write(path, b"not a sqlite database, just garbage bytes").unwrap();
+    };
+
+    // (a) Direct call: recovery quarantines + rebuilds, but the forced
+    //     integrity_check failure makes it return Err instead of Ok(true).
+    corrupt(&db_path);
+    let direct = store.recover_corrupt_db();
+
+    // (b) report_and_recover path: on that recovery failure the report latch,
+    //     set by the initial report, must remain set (not reset to false).
+    corrupt(&db_path);
+    super::CORRUPT_REPORTED.store(false, Ordering::Relaxed);
+    let err = anyhow::anyhow!(
+        "[whatsapp_data] ingest failed: upsert wa_chat 1@lid: database disk image is malformed"
+    );
+    store.report_and_recover("upsert_chats", &err);
+    let latch_after = super::CORRUPT_REPORTED.load(Ordering::Relaxed);
+
+    // Clear the seam + latch BEFORE asserting so a failing assert cannot leak
+    // the forced-fail state into other tests.
+    super::FORCE_INTEGRITY_CHECK_FAIL.store(false, Ordering::Relaxed);
+    super::CORRUPT_REPORTED.store(false, Ordering::Relaxed);
+
+    assert!(
+        direct.is_err(),
+        "recover_corrupt_db must return Err when the rebuilt DB still fails integrity_check"
+    );
+    assert!(
+        latch_after,
+        "report latch must stay set when recovery fails (report-once-per-episode must hold)"
+    );
+}
+
+/// Finding 2 regression: a `whatsapp_data.db` that is already malformed at
+/// process startup must self-heal during store construction. Before the fix,
+/// `WhatsAppDataStore::new()` propagated the `init_schema` corruption error, the
+/// store singleton was never set, and every ingest RPC failed forever — the
+/// corruption survived restarts and re-paged Sentry on every scanner tick.
+#[test]
+fn new_recovers_from_corruption_at_startup() {
+    use std::sync::atomic::Ordering;
+
+    let _guard = super::CORRUPT_TEST_GUARD
+        .lock()
+        .unwrap_or_else(|e| e.into_inner());
+    super::CORRUPT_REPORTED.store(false, Ordering::Relaxed);
+
+    // Pre-create the whatsapp_data dir and plant a malformed DB file BEFORE any
+    // store exists, simulating a corrupt image left behind by a prior run.
+    let tmp = tempdir().expect("tempdir");
+    let workspace = tmp.path().to_path_buf();
+    let db_path = db_path_for(&tmp);
+    std::fs::create_dir_all(db_path.parent().unwrap()).unwrap();
+    std::fs::write(
+        &db_path,
+        b"this is not a sqlite database, just garbage bytes",
+    )
+    .unwrap();
+
+    // Construction must succeed by quarantining + rebuilding, not error out.
+    let store = WhatsAppDataStore::new(&workspace)
+        .expect("new() must self-heal a boot-time corrupt DB, not propagate the error");
+
+    // The corrupt bytes are preserved alongside, never silently dropped.
+    let quarantined = std::fs::read_dir(db_path.parent().unwrap())
+        .unwrap()
+        .filter_map(|e| e.ok())
+        .filter(|e| {
+            e.file_name()
+                .to_string_lossy()
+                .contains("whatsapp_data.db.corrupt-")
+        })
+        .count();
+    assert_eq!(
+        quarantined, 1,
+        "boot-time corrupt image must be quarantined once"
+    );
+
+    // The rebuilt DB is fully usable — a subsequent upsert lands and reads back.
+    let mut chats = HashMap::new();
+    chats.insert("chat@c.us".to_string(), chat_meta("Alice"));
+    let count = store
+        .upsert_chats("acct1", &chats)
+        .expect("upsert must succeed against the rebuilt DB");
+    assert_eq!(count, 1);
+
+    let rows = store
+        .list_chats(&ListChatsRequest {
+            account_id: Some("acct1".to_string()),
+            limit: None,
+            offset: None,
+        })
+        .unwrap();
+    assert_eq!(rows.len(), 1);
+    assert_eq!(rows[0].chat_id, "chat@c.us");
+
+    super::CORRUPT_REPORTED.store(false, Ordering::Relaxed);
+}
+
+/// A *healthy* DB must never be quarantined even if `recover_corrupt_db` is
+/// invoked — `quick_check` passes, so good data is preserved and recovery is a
+/// no-op returning `Ok(false)`.
+#[test]
+fn recover_corrupt_db_is_noop_on_healthy_db() {
+    let (store, tmp) = make_store();
+    let db_path = db_path_for(&tmp);
+    // Seed a real row so there is genuine data that must survive.
+    let mut chats = HashMap::new();
+    chats.insert("chat@c.us".to_string(), chat_meta("Alice"));
+    store.upsert_chats("acct1", &chats).unwrap();
+
+    let recovered = store
+        .recover_corrupt_db()
+        .expect("recovery on a healthy DB must not error");
+    assert!(!recovered, "healthy DB must not be quarantined");
+
+    let quarantined = std::fs::read_dir(db_path.parent().unwrap())
+        .unwrap()
+        .filter_map(|e| e.ok())
+        .any(|e| e.file_name().to_string_lossy().contains(".corrupt-"));
+    assert!(!quarantined, "no quarantine file should be created");
+
+    // Data survives untouched.
+    let rows = store
+        .list_chats(&ListChatsRequest {
+            account_id: Some("acct1".to_string()),
+            limit: None,
+            offset: None,
+        })
+        .unwrap();
+    assert_eq!(rows.len(), 1);
+}
+
 #[test]
 fn upsert_messages_skips_rows_with_empty_ids() {
     let (store, _tmp) = make_store();