fix(memory): recover from SQLITE_CORRUPT in mem_tree jobs worker (#4048) (#4067)

src/openhuman/memory_queue/worker.rs

@@ -8,6 +8,7 @@
 //! worker itself just calls `wait_for_capacity()`; non-LLM jobs
 //! (`AppendBuffer`, `FlushStale`) run without acquiring a permit.
 
+use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::{Arc, OnceLock};
 use std::time::Duration;
 
@@ -46,6 +47,14 @@ const POLL_INTERVAL: Duration = Duration::from_secs(5);
 static WORKER_NOTIFY: OnceLock<Arc<Notify>> = OnceLock::new();
 static STARTED: std::sync::Once = std::sync::Once::new();
 
+/// Process-wide latch so a `SQLITE_CORRUPT` flood is reported to Sentry **once**,
+/// not on every poll from every worker. Set on the first malformed-image
+/// detection; cleared after a recovery attempt settles (quarantine+rebuild or a
+/// quick_check that now passes) so a genuinely-new, later corruption can still
+/// page once. Without this, 4 workers polling a wedged DB re-page ~1/sec
+/// (Sentry TAURI-RUST-E93: 1,633 events in ~17 min from one host).
+static CORRUPT_REPORTED: AtomicBool = AtomicBool::new(false);
+
 /// Notify any idle workers so they re-poll immediately instead of waiting
 /// out [`POLL_INTERVAL`]. Cheap no-op before [`start`] has run.
 pub fn wake_workers() {
@@ -159,6 +168,22 @@ pub fn start(config: Config) {
                                      backing off 300s without reporting: {err:#}"
                                 );
                                 tokio::time::sleep(Duration::from_secs(300)).await;
+                            } else if is_sqlite_corrupt(&err) {
+                                // SQLITE_CORRUPT (code 11): the on-disk mem_tree
+                                // image is malformed. Unlike busy/io-transient/
+                                // disk-full, this NEVER clears on its own — the
+                                // claim UPDATE fails forever, so re-polling every
+                                // second and paging Sentry each time turns one
+                                // unrecoverable file into a flood (TAURI-RUST-E93:
+                                // 1,633 events in ~17 min, one host). Report once,
+                                // drive quarantine+rebuild recovery (factored into
+                                // `recover_corrupt_db_once` so it is unit-testable
+                                // without spinning the live loop), then back off
+                                // long so a failed recovery never re-floods.
+                                // `notify` still wakes us on new enqueues once the
+                                // rebuild succeeds.
+                                recover_corrupt_db_once(idx, &err, &cfg);
+                                tokio::time::sleep(Duration::from_secs(300)).await;
                             } else {
                                 crate::core::observability::report_error(
                                     &err,
@@ -411,6 +436,83 @@ fn is_sqlite_disk_full(err: &anyhow::Error) -> bool {
         || msg.contains("insertion failed because database is full")
 }
 
+/// Classify whether an error from `claim_next` 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 `SQLITE_BUSY`/`LOCKED`, the transient I/O family, or `SQLITE_FULL`,
+/// a malformed image is **persistent on-disk damage**: the claim `UPDATE` can
+/// never succeed, so re-polling every second and paging Sentry on each failure
+/// turns one corrupt file into an infinite flood (Sentry TAURI-RUST-E93:
+/// ~1.6k events in ~17 min from a single host). The worker reports once, drives
+/// a quarantine+rebuild recovery (`recover_corrupt_db`), and backs off long.
+///
+/// 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).
+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("database disk image is malformed") || msg.contains("file is not a database")
+}
+
+/// Handle a confirmed `SQLITE_CORRUPT` failure from the worker loop: report it
+/// to Sentry **once** (process-wide [`CORRUPT_REPORTED`] latch, not per-poll
+/// across the workers) and drive the quarantine+rebuild recovery in
+/// [`recover_corrupt_db`](crate::openhuman::memory_store::chunks::store::recover_corrupt_db).
+///
+/// Factored out of [`start`]'s error arm so the report-once + recovery decision
+/// logic is unit-testable without spinning the live worker loop. The caller
+/// applies the long backoff after this returns.
+fn recover_corrupt_db_once(idx: usize, err: &anyhow::Error, config: &Config) {
+    if !CORRUPT_REPORTED.swap(true, Ordering::Relaxed) {
+        crate::core::observability::report_error(
+            err,
+            "memory",
+            "tree_jobs_worker_corrupt",
+            &[("worker_idx", &idx.to_string())],
+        );
+    }
+    log::error!(
+        "[memory::jobs] worker {idx} hit SQLITE_CORRUPT (malformed DB image), \
+         attempting quarantine + rebuild recovery: {err:#}"
+    );
+    match crate::openhuman::memory_store::chunks::store::recover_corrupt_db(config) {
+        Ok(true) => {
+            log::warn!(
+                "[memory::jobs] worker {idx} quarantined corrupt mem_tree DB and rebuilt \
+                 empty schema; queue will resume"
+            );
+            // Recovery settled — allow a future, genuinely-new corruption to
+            // page once.
+            CORRUPT_REPORTED.store(false, Ordering::Relaxed);
+        }
+        Ok(false) => {
+            log::info!(
+                "[memory::jobs] worker {idx} corruption recovery: quick_check now passes, \
+                 no quarantine needed"
+            );
+            CORRUPT_REPORTED.store(false, Ordering::Relaxed);
+        }
+        Err(rec_err) => {
+            log::error!(
+                "[memory::jobs] worker {idx} corruption recovery FAILED, retrying after \
+                 backoff: {rec_err:#}"
+            );
+        }
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -666,6 +768,138 @@ mod tests {
         )));
     }
 
+    // ── is_sqlite_corrupt tests (#4048 / Sentry TAURI-RUST-E93) ──────────────
+
+    /// `SQLITE_CORRUPT` (primary code `DatabaseCorrupt`, code 11) is the
+    /// malformed-image signal from `claim_next`; it must classify so the worker
+    /// quarantines + rebuilds instead of paging Sentry every second.
+    #[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)));
+    }
+
+    /// `SQLITE_NOTADB` (code `NotADatabase`, 26 — header unreadable) is the
+    /// same broad on-disk-damage class and must classify too.
+    #[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 a few `.context()` layers deep when it bubbles
+    /// out of `claim_next` → `with_connection`; the downcast must still find
+    /// the `DatabaseCorrupt` code.
+    #[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("Failed to claim next mem_tree_jobs row")
+            .context("with_connection closure failed");
+        assert!(is_sqlite_corrupt(&wrapped));
+    }
+
+    /// Text fallback: the exact flattened Sentry string (TAURI-RUST-E93) must
+    /// classify even when no rusqlite error is available to downcast.
+    #[test]
+    fn is_sqlite_corrupt_text_fallback() {
+        let err = anyhow::anyhow!(
+            "Failed to claim next mem_tree_jobs row: database disk image is malformed: \
+             Error code 11: The database disk image is malformed"
+        );
+        assert!(is_sqlite_corrupt(&err));
+    }
+
+    /// Busy/locked, disk-full, constraint violations, and unrelated errors must
+    /// NOT be swallowed as corruption — quarantining on those would destroy a
+    /// perfectly good DB.
+    #[test]
+    fn is_sqlite_corrupt_does_not_match_other_errors() {
+        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 disk_full = rusqlite::Error::SqliteFailure(
+            rusqlite::ffi::Error {
+                code: rusqlite::ErrorCode::DiskFull,
+                extended_code: 13,
+            },
+            Some("database or disk is full".into()),
+        );
+        assert!(!is_sqlite_corrupt(&anyhow::Error::from(disk_full)));
+
+        let constraint = rusqlite::Error::SqliteFailure(
+            rusqlite::ffi::Error {
+                code: rusqlite::ErrorCode::ConstraintViolation,
+                extended_code: 19,
+            },
+            Some("UNIQUE constraint failed: mem_tree_jobs.dedupe_key".into()),
+        );
+        assert!(!is_sqlite_corrupt(&anyhow::Error::from(constraint)));
+
+        assert!(!is_sqlite_corrupt(&anyhow::anyhow!(
+            "upstream returned 500: internal server error"
+        )));
+    }
+
+    /// The worker's corruption arm must quarantine a malformed image and rebuild
+    /// an empty, queryable schema so the queue resumes — exercising the
+    /// report-once + recover path the live loop runs.
+    #[tokio::test]
+    async fn recover_corrupt_db_once_quarantines_and_rebuilds() {
+        let (_tmp, cfg) = test_config();
+        // Lay down a malformed `chunks.db` (garbage header) at the canonical path.
+        let db_path = cfg.workspace_dir.join("memory_tree").join("chunks.db");
+        std::fs::create_dir_all(db_path.parent().unwrap()).unwrap();
+        std::fs::write(&db_path, b"not a sqlite database, just garbage bytes").unwrap();
+
+        let err = anyhow::anyhow!(
+            "Failed to claim next mem_tree_jobs row: database disk image is malformed"
+        );
+        recover_corrupt_db_once(0, &err, &cfg);
+
+        // Corrupt bytes are preserved alongside (never silently dropped) ...
+        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("chunks.db.corrupt-")
+            });
+        assert!(
+            quarantined,
+            "corrupt image must be quarantined, not deleted"
+        );
+
+        // ... and the rebuilt queue DB is healthy and empty.
+        let processed = run_once(&cfg).await.unwrap();
+        assert!(!processed, "rebuilt queue starts empty");
+    }
+
     #[tokio::test]
     async fn wake_workers_is_noop_before_start() {
         wake_workers();

src/openhuman/memory_store/chunks/connection.rs

@@ -1,4 +1,5 @@
 use anyhow::{Context, Result};
+use chrono::Utc;
 use parking_lot::Mutex as PMutex;
 use rusqlite::Connection;
 use std::collections::HashMap;
@@ -597,6 +598,131 @@ pub fn with_connection<T>(config: &Config, f: impl FnOnce(&Connection) -> Result
     f(&guard)
 }
 
+/// Append `suffix` to the *file name* of `path` (so `chunks.db` + `-wal`
+/// = `chunks.db-wal`, and `chunks.db` + `.corrupt-…` = `chunks.db.corrupt-…`).
+/// 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
+}
+
+/// Run `PRAGMA quick_check(1)` against `db_path` on a fresh, short-lived
+/// connection. Returns `Ok(true)` when the structural scan reports `"ok"`,
+/// `Ok(false)` when it reports any 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(db_path: &Path) -> Result<bool> {
+    let conn = Connection::open(db_path)
+        .with_context(|| format!("open for quick_check: {}", db_path.display()))?;
+    let _ = conn.busy_timeout(SQLITE_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"))
+}
+
+/// Recover from a `SQLITE_CORRUPT` (malformed image) on the memory_tree DB.
+///
+/// Unlike the transient/contention/disk-full classes, a malformed on-disk
+/// image never heals on its own — every query fails forever and the worker
+/// re-pages Sentry on each poll (Sentry TAURI-RUST-E93: ~1.6k events in ~17 min
+/// from a single host). This is the recovery lever the sibling suppressors
+/// lack: it quarantines the damaged file (and its WAL/SHM side-files) to a
+/// timestamped `.corrupt-<ts>` copy — **preserved, not deleted**, so the bytes
+/// can still be inspected or salvaged — then rebuilds an empty schema so the
+/// memory-tree queue resumes instead of wedging indefinitely.
+///
+/// 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 (caller backs off and retries).
+pub(crate) fn recover_corrupt_db(config: &Config) -> Result<bool> {
+    let db_path = db_path_for(config);
+
+    // 1. Drop any cached (corrupt) connection + breaker so the OS file handle
+    //    is closed before we rename, and the next open re-inits cleanly.
+    conn_cache().connections.lock().remove(&db_path);
+    conn_cache().breakers.lock().remove(&db_path);
+
+    // 2. 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 out without quarantining.
+    if db_path.exists() {
+        match quick_check_ok(&db_path) {
+            Ok(true) => {
+                log::info!(
+                    "[memory_tree] quick_check passed for {} — no quarantine needed",
+                    db_path.display()
+                );
+                return Ok(false);
+            }
+            Ok(false) => {
+                log::warn!(
+                    "[memory_tree] quick_check confirms corruption for {}, quarantining",
+                    db_path.display()
+                );
+            }
+            Err(e) => {
+                // The check couldn't even run (unopenable / unreadable header).
+                // That is itself a malformed-image signal — treat as corrupt.
+                log::warn!(
+                    "[memory_tree] quick_check could not run for {} ({e:#}); treating as corrupt",
+                    db_path.display()
+                );
+            }
+        }
+    } else {
+        log::warn!(
+            "[memory_tree] corrupt-recovery: {} is missing; rebuilding fresh schema",
+            db_path.display()
+        );
+    }
+
+    // 3. Quarantine the main DB + WAL/SHM side-files to `<name>.corrupt-<ts>`.
+    let ts = Utc::now().format("%Y%m%dT%H%M%SZ").to_string();
+    let mut quarantined = 0usize;
+    for suffix in &["", "-wal", "-shm"] {
+        let src = with_name_suffix(&db_path, suffix);
+        if !src.exists() {
+            continue;
+        }
+        let dst = with_name_suffix(&src, &format!(".corrupt-{ts}"));
+        std::fs::rename(&src, &dst).with_context(|| {
+            format!(
+                "failed to quarantine corrupt memory_tree file {} -> {}",
+                src.display(),
+                dst.display()
+            )
+        })?;
+        log::warn!(
+            "[memory_tree] quarantined {} -> {}",
+            src.display(),
+            dst.display()
+        );
+        quarantined += 1;
+    }
+
+    // 4. Rebuild an empty schema by forcing a fresh open. The damaged rows are
+    //    not silently dropped — they live on in the `.corrupt-<ts>` copy.
+    get_or_init_connection(config)
+        .context("failed to rebuild memory_tree schema after quarantining corrupt DB")?;
+
+    log::warn!(
+        "[memory_tree] corruption recovery complete: quarantined {quarantined} file(s), \
+         rebuilt empty schema at {}",
+        db_path.display()
+    );
+    Ok(true)
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -634,4 +760,79 @@ mod tests {
         assert!(!cb.mark_startup_emitted());
         assert!(!cb.mark_startup_emitted());
     }
+
+    // ── recover_corrupt_db tests (TAURI-RUST-E93 / #4048) ────────────────────
+
+    fn corrupt_test_config() -> (tempfile::TempDir, Config) {
+        let tmp = tempfile::TempDir::new().unwrap();
+        let mut cfg = Config::default();
+        cfg.workspace_dir = tmp.path().to_path_buf();
+        (tmp, cfg)
+    }
+
+    /// A malformed on-disk image must be quarantined (not deleted) and replaced
+    /// by a fresh, queryable schema so the memory-tree queue resumes.
+    #[test]
+    fn recover_corrupt_db_quarantines_and_rebuilds() {
+        clear_connection_cache();
+        let (_tmp, cfg) = corrupt_test_config();
+        let db_path = db_path_for(&cfg);
+        std::fs::create_dir_all(db_path.parent().unwrap()).unwrap();
+        // Garbage bytes → not a valid SQLite header → corrupt image.
+        std::fs::write(&db_path, b"this is not a sqlite database, it is garbage").unwrap();
+
+        let recovered = recover_corrupt_db(&cfg).expect("recovery should succeed");
+        assert!(recovered, "garbage image must be quarantined + rebuilt");
+
+        // The corrupt bytes are preserved alongside, not 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("chunks.db.corrupt-")
+            })
+            .collect();
+        assert_eq!(
+            quarantined.len(),
+            1,
+            "exactly one quarantined copy should exist"
+        );
+
+        // The rebuilt DB is healthy and the jobs table is queryable + empty.
+        clear_connection_cache();
+        let count: i64 = with_connection(&cfg, |conn| {
+            conn.query_row("SELECT COUNT(*) FROM mem_tree_jobs", [], |r| r.get(0))
+                .context("count jobs")
+        })
+        .expect("rebuilt DB must be queryable");
+        assert_eq!(count, 0, "rebuilt jobs table starts empty");
+    }
+
+    /// A healthy DB must NOT be quarantined — `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() {
+        clear_connection_cache();
+        let (_tmp, cfg) = corrupt_test_config();
+        // Force a healthy DB into existence.
+        with_connection(&cfg, |conn| {
+            conn.query_row("SELECT COUNT(*) FROM mem_tree_jobs", [], |r| {
+                r.get::<_, i64>(0)
+            })
+            .context("seed healthy db")
+        })
+        .unwrap();
+
+        let recovered = recover_corrupt_db(&cfg).expect("recovery should succeed");
+        assert!(!recovered, "healthy DB must not be quarantined");
+
+        let db_path = db_path_for(&cfg);
+        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");
+    }
 }

src/openhuman/memory_store/chunks/store.rs

@@ -1414,6 +1414,7 @@ fn ms_to_utc(ms: i64) -> rusqlite::Result<DateTime<Utc>> {
 
 #[path = "connection.rs"]
 mod connection;
+pub(crate) use connection::recover_corrupt_db;
 pub use connection::with_connection;
 #[cfg(test)]
 #[allow(unused_imports)]