diff --git a/src/openhuman/memory_queue/worker.rs b/src/openhuman/memory_queue/worker.rs index be73e5c89..e803acc75 100644 --- a/src/openhuman/memory_queue/worker.rs +++ b/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> = 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::() + { + 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(); diff --git a/src/openhuman/memory_store/chunks/connection.rs b/src/openhuman/memory_store/chunks/connection.rs index 486d1ee60..fb7d0ea81 100644 --- a/src/openhuman/memory_store/chunks/connection.rs +++ b/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(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 { + 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-` 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 { + 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 `.corrupt-`. + 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-` 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"); + } } diff --git a/src/openhuman/memory_store/chunks/store.rs b/src/openhuman/memory_store/chunks/store.rs index 9799773b0..7b068c657 100644 --- a/src/openhuman/memory_store/chunks/store.rs +++ b/src/openhuman/memory_store/chunks/store.rs @@ -1414,6 +1414,7 @@ fn ms_to_utc(ms: i64) -> rusqlite::Result> { #[path = "connection.rs"] mod connection; +pub(crate) use connection::recover_corrupt_db; pub use connection::with_connection; #[cfg(test)] #[allow(unused_imports)]