diff --git a/src/core/observability.rs b/src/core/observability.rs index eaf6d04a33..fd9565c466 100644 --- a/src/core/observability.rs +++ b/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 { 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 @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 @@ -2097,6 +2153,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 @@ -4330,6 +4394,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 [ diff --git a/src/openhuman/whatsapp_data/sqlite_retry.rs b/src/openhuman/whatsapp_data/sqlite_retry.rs index d8ebf0e721..7f041cdeec 100644 --- a/src/openhuman/whatsapp_data/sqlite_retry.rs +++ b/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::() + { + 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(op_name: &str, mut f: F) -> Result 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; diff --git a/src/openhuman/whatsapp_data/store.rs b/src/openhuman/whatsapp_data/store.rs index 671d96fc60..6025f8ed36 100644 --- a/src/openhuman/whatsapp_data/store.rs +++ b/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 { 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( + &self, + op_name: &str, + f: impl Fn() -> Result, + ) -> Result { + 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-` 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 { + // 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 `.corrupt-`. + 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-` 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 { + 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 { + #[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::>() @@ -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 { Ok(WhatsAppChat { account_id: row.get(0)?, diff --git a/src/openhuman/whatsapp_data/store_tests.rs b/src/openhuman/whatsapp_data/store_tests.rs index 4ccfd77ef4..b20617f07a 100644 --- a/src/openhuman/whatsapp_data/store_tests.rs +++ b/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();