diff --git a/crates/tinymemory-api/src/host/events.rs b/crates/tinymemory-api/src/host/events.rs index a843db4..d2864fd 100644 --- a/crates/tinymemory-api/src/host/events.rs +++ b/crates/tinymemory-api/src/host/events.rs @@ -196,6 +196,27 @@ pub enum MemoryEvent { /// distinguishable in the log without a correlation id. origin: String, }, + /// The primary memory-tree store (`chunks.db`) was found corrupt; the + /// damaged file was quarantined to a timestamped `.corrupt-` sibling + /// (preserved, never deleted) and an empty schema was rebuilt in its + /// place. + /// + /// The rebuilt store works, but it is empty: the ingested-source registry + /// rows went with the old file, so every previously synced source must + /// re-sync before tree-backed recall recovers. The host surfaces this in + /// its durable user-error centre — see [`STORE_CORRUPT_KIND`] — naming the + /// quarantined path so the user's indexed history is recoverable rather + /// than silently stranded on disk (openhuman#5820). + StoreCorruptQuarantined { + /// Short, non-sensitive tag naming the detecting path + /// (`jobs worker N` / `composio tree ingest` / `startup integrity + /// check`), so the paths stay distinguishable in the log. + origin: String, + /// Filesystem path of the quarantined main DB file, when the rename's + /// result could be located. Local path, shown to the workspace's own + /// user only. + quarantined_path: Option, + }, } /// Stable `error_type` token for the local-embedding-runtime user error. @@ -206,6 +227,14 @@ pub enum MemoryEvent { /// either side drops the signal silently. pub const LOCAL_MODEL_UNAVAILABLE_KIND: &str = "local_model_unavailable"; +/// Stable `error_type` token for the corrupt-store-quarantined user error. +/// +/// Mirrors the frontend `UserErrorKind` discriminator of the same name, like +/// [`LOCAL_MODEL_UNAVAILABLE_KIND`] above: the host builds the wire payload +/// from it, and tests on both sides assert on it, so a drift on either side +/// drops the signal silently. +pub const STORE_CORRUPT_KIND: &str = "memory_store_corrupt"; + /// `error_source` for the memory subsystem's user errors. Drives the panel's /// scope grouping (`socketService` maps it to the `memory` `UserErrorScope`). pub const MEMORY_USER_ERROR_SOURCE: &str = "memory"; diff --git a/crates/tinymemory-api/src/host/mod.rs b/crates/tinymemory-api/src/host/mod.rs index e3b713f..1f0e158 100644 --- a/crates/tinymemory-api/src/host/mod.rs +++ b/crates/tinymemory-api/src/host/mod.rs @@ -69,7 +69,7 @@ pub use embeddings::{format_embedding_signature, EmbeddingProvider, NoopEmbeddin pub use error_reporter::ErrorReporter; pub use events::{ EmbeddingHealthReason, MemoryEvent, MemoryEventSink, NoopEventSink, SyncTrigger, - LOCAL_MODEL_UNAVAILABLE_KIND, MEMORY_USER_ERROR_SOURCE, + LOCAL_MODEL_UNAVAILABLE_KIND, MEMORY_USER_ERROR_SOURCE, STORE_CORRUPT_KIND, }; pub use local_ai::{LocalAiConfig, LocalAiUsage}; pub use nlp::{SpacyEntity, SpacyResponse}; diff --git a/crates/tinymemory-bus/src/provider/sync.rs b/crates/tinymemory-bus/src/provider/sync.rs index 83cc4f3..382d729 100644 --- a/crates/tinymemory-bus/src/provider/sync.rs +++ b/crates/tinymemory-bus/src/provider/sync.rs @@ -201,6 +201,21 @@ pub struct SyncAuditEntry { /// Why it did not, when it did not. Never memory content. #[serde(default, skip_serializing_if = "Option::is_none")] pub error: Option, + /// Items fetched-and-stored whose memory-tree ingest failed + /// (openhuman#5820). A non-zero count with `success: false` is the + /// "fetch succeeded, tree did not" partial verdict; rows written before + /// the field existed read back as `0`. + #[serde(default, skip_serializing_if = "is_zero_u32")] + pub tree_ingest_failures: u32, + /// Why the tree half failed, when it did. Never memory content. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub tree_error: Option, +} + +/// `skip_serializing_if` gate for the additive counter above. +#[allow(clippy::trivially_copy_pass_by_ref)] // serde's contract is a reference +fn is_zero_u32(value: &u32) -> bool { + *value == 0 } impl SyncAuditEntry { diff --git a/crates/tinymemory-bus/src/provider/sync_tests.rs b/crates/tinymemory-bus/src/provider/sync_tests.rs index 2d50f8b..9a0b5f4 100644 --- a/crates/tinymemory-bus/src/provider/sync_tests.rs +++ b/crates/tinymemory-bus/src/provider/sync_tests.rs @@ -29,6 +29,8 @@ fn entry() -> SyncAuditEntry { duration_ms: 4_200, success: true, error: None, + tree_ingest_failures: 0, + tree_error: None, } } diff --git a/crates/tinymemory-core/src/corruption/mod.rs b/crates/tinymemory-core/src/corruption/mod.rs new file mode 100644 index 0000000..cd04378 --- /dev/null +++ b/crates/tinymemory-core/src/corruption/mod.rs @@ -0,0 +1,263 @@ +//! One classification and one recovery path for a corrupt chunk store. +//! +//! `SQLITE_CORRUPT` used to be handled per call site, and the sites disagreed: +//! the queue worker treated it as fatal (report once, quarantine + rebuild, +//! long backoff) while the tree-ingest paths logged it at `warn` as +//! "non-fatal" and carried on — which let a malformed `chunks.db` fail every +//! ingest for 34 minutes while the sync surfaces reported success, until the +//! job-claim path finally hit the same damage and quarantined the file +//! (openhuman#5820). This module is the single answer both kinds of site call: +//! [`is_sqlite_corrupt`] to classify, [`report_and_recover`] to escalate. +//! +//! Recovery is deliberately the queue worker's proven sequence: report to the +//! host once per corruption episode (process-wide latch), mark the tree +//! degraded so status surfaces stop reading healthy, quarantine + rebuild via +//! [`recover_corrupt_db`](crate::store::chunks::store::recover_corrupt_db), +//! and announce the outcome as a [`MemoryEvent`] so a host can tell the user +//! what happened and where the quarantined file is. + +use std::path::PathBuf; +use std::sync::atomic::{AtomicBool, Ordering}; + +use crate::events::{self, MemoryEvent}; +use crate::tree::health::{clear_storage_degraded, mark_storage_degraded, FailureCode}; +use crate::Config; + +/// Process-wide latch so a `SQLITE_CORRUPT` flood is reported to the host +/// **once** per corruption episode, not once per failing call. One corrupt +/// file fails every ingest and every queue poll until recovery settles, so +/// without the latch a single episode pages hundreds of times (Sentry +/// TAURI-RUST-E93: ~1.6k events in ~17 min from one host). Cleared when a +/// recovery attempt settles (quarantine + rebuild, or a quick_check that now +/// passes) so a genuinely-new, later corruption can page again. +static CORRUPT_REPORTED: AtomicBool = AtomicBool::new(false); + +/// Classify whether an error 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 busy/locked, the transient I/O family, or `SQLITE_FULL`, a malformed +/// image is **persistent on-disk damage**: no retry of the failing call can +/// ever succeed, so callers must escalate through [`report_and_recover`] +/// rather than logging and continuing. +/// +/// Matching on the error code is rusqlite-version-stable and, because anyhow +/// downcasts through `context` layers, survives wrapping. The text fallback +/// covers the case where the rusqlite error was flattened into a plain +/// `anyhow!("…: {error}")` string at a module boundary — SQLite renders these +/// as "database disk image is malformed" (code 11) and "file is not a +/// database" (code 26). +pub(crate) 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; + } + } + is_corrupt_text(&format!("{err:#}")) +} + +/// The text half of [`is_sqlite_corrupt`], for errors that only exist as +/// strings (a pipeline failure message, a wire error body). +pub(crate) fn is_corrupt_text(message: &str) -> bool { + let msg = message.to_ascii_lowercase(); + msg.contains("database disk image is malformed") || msg.contains("file is not a database") +} + +/// Handle a confirmed `SQLITE_CORRUPT` on the chunk store, from any path. +/// +/// Reports to the host once per episode (see [`CORRUPT_REPORTED`]), marks the +/// tree storage-degraded so status surfaces read `error` instead of healthy, +/// then drives the quarantine + rebuild recovery. On a settled recovery the +/// degraded flag and the latch clear — the rebuilt store works, and the +/// durable "your memory tree was quarantined" message is the +/// [`MemoryEvent::StoreCorruptQuarantined`] this publishes, not a stuck +/// banner. A failed recovery leaves both set: the store really is unusable. +/// +/// `origin` names the detecting path for logs and event payloads +/// (`"jobs worker 0"`, `"composio tree ingest"`, `"startup integrity check"`); +/// `report_key` is the host-facing operation tag, kept caller-chosen so the +/// queue worker's long-standing `tree_jobs_worker_corrupt` Sentry grouping +/// survives the consolidation. +pub(crate) fn report_and_recover( + origin: &str, + report_key: &str, + err: &anyhow::Error, + config: &Config, +) { + if !CORRUPT_REPORTED.swap(true, Ordering::Relaxed) { + crate::observability::report_error(err, "memory", report_key, &[("origin", origin)]); + } + mark_storage_degraded(FailureCode::StorageUnavailable); + log::error!( + "[memory:corruption] {origin} hit SQLITE_CORRUPT (malformed chunk DB image), \ + attempting quarantine + rebuild recovery: {err:#}" + ); + match crate::store::chunks::store::recover_corrupt_db(config) { + Ok(true) => { + let quarantined = latest_quarantined_path(config); + match quarantined.as_deref() { + Some(path) => log::error!( + "[memory:corruption] {origin}: quarantined corrupt mem_tree DB to \ + {path} and rebuilt an empty schema. The quarantined file is preserved, \ + not deleted; previously ingested sources must re-sync to repopulate \ + the tree", + path = path.display() + ), + None => log::error!( + "[memory:corruption] {origin}: quarantined corrupt mem_tree DB and \ + rebuilt an empty schema; previously ingested sources must re-sync" + ), + } + events::publish(MemoryEvent::StoreCorruptQuarantined { + origin: origin.to_string(), + quarantined_path: quarantined.map(|path| path.display().to_string()), + }); + // Recovery settled: the rebuilt store is usable again, so the + // degraded flag must not outlive the damage, and a future, + // genuinely-new corruption may page once more. + clear_storage_degraded(); + CORRUPT_REPORTED.store(false, Ordering::Relaxed); + } + Ok(false) => { + log::info!( + "[memory:corruption] {origin}: corruption recovery ran but quick_check \ + now passes; no quarantine needed" + ); + clear_storage_degraded(); + CORRUPT_REPORTED.store(false, Ordering::Relaxed); + } + Err(rec_err) => { + log::error!( + "[memory:corruption] {origin}: corruption recovery FAILED, store stays \ + degraded: {rec_err:#}" + ); + } + } +} + +/// The tree-ingest sinks' shared error policy: escalate corruption, count and +/// tolerate everything else (openhuman#5820). +/// +/// `Ok(())` means the failure was tolerated — logged by the caller, recorded +/// in `counter` for the run's verdict, sync continues. `Err` means the store +/// is corrupt: the shared recovery has run and the caller must abort its run, +/// because every later item fails identically against a malformed image. +/// Lives here rather than on each sink so `PipelineHost` and +/// `HostSyncAdapter` cannot drift apart on the classification again — the +/// drift IS the incident this module exists for. +pub(crate) fn escalate_or_count( + origin: &str, + config: &Config, + error: anyhow::Error, + counter: &std::sync::atomic::AtomicU32, +) -> anyhow::Result<()> { + if is_sqlite_corrupt(&error) { + report_and_recover(origin, "tree_ingest_corrupt", &error, config); + return Err(error.context( + "memory-tree store is corrupt; aborting this sync run \ + (the store was quarantined and rebuilt — re-sync to repopulate)", + )); + } + counter.fetch_add(1, Ordering::Relaxed); + Ok(()) +} + +/// The most recent quarantined chunk-DB copy in this workspace, if any. +/// +/// The quarantine renames `chunks.db` to `chunks.db.corrupt-` +/// (`%Y%m%dT%H%M%SZ`), so the lexically greatest matching name is the newest. +/// Side files quarantine as `chunks.db-wal.corrupt-` and never match the +/// main file's prefix. +pub(crate) fn latest_quarantined_path(config: &Config) -> Option { + let dir = config.workspace_dir().join("memory_tree"); + let entries = std::fs::read_dir(&dir).ok()?; + entries + .filter_map(|entry| entry.ok()) + .filter(|entry| { + entry + .file_name() + .to_str() + .is_some_and(|name| name.starts_with("chunks.db.corrupt-")) + }) + .max_by_key(std::fs::DirEntry::file_name) + .map(|entry| entry.path()) +} + +/// Startup integrity check for the chunk store (openhuman#5820 item 5). +/// +/// Workspaces written before the two-engines-over-one-file fix +/// (openhuman#5725) can carry latent page damage that only surfaces when some +/// later call happens to walk a damaged b-tree — in the incident, 10 hours +/// after boot, via whichever path hit it first. Running `PRAGMA +/// quick_check(1)` once at queue start moves that discovery to a defined +/// moment with a defined owner: damage found here goes straight through +/// [`report_and_recover`] instead of failing arbitrary calls first. +/// +/// A missing file is healthy (first boot creates it). A failing pragma is +/// treated as corrupt only when the failure itself classifies as corruption +/// (`NotADatabase` is what a destroyed header raises) — a plain open failure +/// can be a lock or a permission problem, and quarantining on those would +/// rename a healthy file. The scan reads the whole file, so callers run this +/// on a blocking thread, off the async workers. +pub(crate) fn startup_integrity_check(config: &Config) { + let db_path = config.workspace_dir().join("memory_tree").join("chunks.db"); + if !db_path.exists() { + return; + } + let verdict = (|| -> anyhow::Result { + let conn = rusqlite::Connection::open(&db_path)?; + let _ = conn.busy_timeout(std::time::Duration::from_secs(15)); + Ok(conn.query_row("PRAGMA quick_check(1)", [], |row| row.get(0))?) + })(); + match verdict { + Ok(result) if result.eq_ignore_ascii_case("ok") => { + log::debug!( + "[memory:corruption] startup quick_check passed for {}", + db_path.display() + ); + } + Ok(result) => { + let err = anyhow::anyhow!( + "startup quick_check found a malformed chunk DB image at {}: {result}", + db_path.display() + ); + report_and_recover( + "startup integrity check", + "tree_startup_corrupt", + &err, + config, + ); + } + Err(error) => { + let err = error.context(format!( + "startup quick_check could not scan {}", + db_path.display() + )); + if is_sqlite_corrupt(&err) { + report_and_recover( + "startup integrity check", + "tree_startup_corrupt", + &err, + config, + ); + } else { + // A lock, a permission problem, a dying disk — not proven + // corruption. Quarantining here would rename a file that may + // be fine; leave it for the runtime classifiers to judge from + // a real call's error. + log::warn!( + "[memory:corruption] startup quick_check could not scan the chunk DB \ + (not classified as corruption, leaving the file in place): {err:#}" + ); + } + } + } +} + +#[cfg(test)] +mod test; diff --git a/crates/tinymemory-core/src/corruption/test.rs b/crates/tinymemory-core/src/corruption/test.rs new file mode 100644 index 0000000..0f94930 --- /dev/null +++ b/crates/tinymemory-core/src/corruption/test.rs @@ -0,0 +1,290 @@ +//! Tests for the surrounding module. +//! +//! The classifier table moved here with `is_sqlite_corrupt` (it grew up in +//! `queue::worker` for #4048 / Sentry TAURI-RUST-E93); the recovery tests +//! exercise the shared `report_and_recover` every detecting path now calls. + +use super::*; +use crate::events::MemoryEvent; +use tempfile::TempDir; +use tinymemory_api::host::test_support::TestHostConfig; + +fn test_config() -> (TempDir, TestHostConfig) { + crate::test_seams::init(); + let tmp = TempDir::new().unwrap(); + let mut cfg = TestHostConfig::default(); + cfg.workspace_dir = tmp.path().to_path_buf(); + cfg.memory_tree.embedding_endpoint = None; + cfg.memory_tree.embedding_model = None; + cfg.memory_tree.embedding_strict = false; + (tmp, cfg) +} + +// ── is_sqlite_corrupt (#4048 / Sentry TAURI-RUST-E93) ──────────────────── + +/// `SQLITE_CORRUPT` (primary code `DatabaseCorrupt`, code 11) is the +/// malformed-image signal; it must classify so detectors escalate through +/// quarantine + rebuild instead of retrying or paging forever. +#[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)); +} + +/// The tree-ingest boundary flattens the engine error into a plain +/// `anyhow!("memory-tree ingest failed for source `…`: {error}")` string — +/// the exact shape of the openhuman#5820 incident's 747 warns. The classifier +/// must see through that flattening, because this path is why corruption ran +/// as "non-fatal" for 34 minutes. +#[test] +fn is_sqlite_corrupt_matches_the_flattened_ingest_shape() { + let err = anyhow::anyhow!( + "memory-tree ingest failed for source `github:owner/repo:42`: \ + 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 string half classifies the same two SQLite phrases, for errors that +/// only exist as text (pipeline failure messages, wire error bodies). +#[test] +fn is_corrupt_text_matches_both_phrases_and_nothing_else() { + assert!(is_corrupt_text( + "composio sync failed: database disk image is malformed" + )); + assert!(is_corrupt_text("open failed: File is NOT a Database")); + assert!(!is_corrupt_text("database or disk is full")); + assert!(!is_corrupt_text("connection refused")); +} + +// ── report_and_recover ─────────────────────────────────────────────────── + +/// The shared recovery must quarantine a malformed image, rebuild an empty +/// queryable schema, publish `StoreCorruptQuarantined` naming the quarantined +/// file, and clear the storage degradation once recovery settles — exercising +/// the path every detector (worker, ingest, startup) now runs. +#[tokio::test] +async fn report_and_recover_quarantines_rebuilds_and_announces() { + let (_tmp, cfg) = test_config(); + let sink = crate::events::RecordingSink::install(); + // 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"); + report_and_recover("jobs worker 0", "tree_jobs_worker_corrupt", &err, &cfg); + + // Corrupt bytes are preserved alongside (never silently dropped) ... + let quarantined = latest_quarantined_path(&cfg).expect("quarantined copy exists"); + assert!(quarantined + .file_name() + .unwrap() + .to_string_lossy() + .starts_with("chunks.db.corrupt-")); + + // ... the event names that file so a host can tell the user ... + let events = sink.drain(); + let announced = events.iter().any(|event| { + matches!( + event, + MemoryEvent::StoreCorruptQuarantined { origin, quarantined_path } + if origin == "jobs worker 0" + && quarantined_path.as_deref() + == Some(quarantined.display().to_string().as_str()) + ) + }); + assert!( + announced, + "StoreCorruptQuarantined must be published with the quarantined path; got {events:?}" + ); + + // ... recovery settled, so the storage degradation does not outlive it ... + assert!( + !crate::tree::health::current_degraded_state().storage, + "a settled recovery must clear the storage degradation" + ); + + // ... and the rebuilt queue DB is healthy and empty. + let processed = crate::queue::worker::run_once(&cfg).await.unwrap(); + assert!(!processed, "rebuilt queue starts empty"); +} + +/// `latest_quarantined_path` picks the newest timestamped copy and ignores +/// side-file quarantines (`chunks.db-wal.corrupt-…`). +#[test] +fn latest_quarantined_path_picks_newest_main_copy() { + let (_tmp, cfg) = test_config(); + let dir = cfg.workspace_dir.join("memory_tree"); + std::fs::create_dir_all(&dir).unwrap(); + assert!(latest_quarantined_path(&cfg).is_none()); + std::fs::write(dir.join("chunks.db.corrupt-20260101T000000Z"), b"old").unwrap(); + std::fs::write(dir.join("chunks.db.corrupt-20260827T120000Z"), b"new").unwrap(); + std::fs::write(dir.join("chunks.db-wal.corrupt-20261231T235959Z"), b"wal").unwrap(); + let newest = latest_quarantined_path(&cfg).expect("a main quarantined copy"); + assert_eq!( + newest.file_name().unwrap().to_string_lossy(), + "chunks.db.corrupt-20260827T120000Z" + ); +} + +// ── startup_integrity_check (openhuman#5820 item 5) ────────────────────── + +/// A garbage `chunks.db` found at startup is quarantined immediately instead +/// of surfacing hours later through whichever call walks the damage first. +#[test] +fn startup_integrity_check_quarantines_a_corrupt_db() { + let (_tmp, cfg) = test_config(); + 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(); + + startup_integrity_check(&cfg); + + assert!( + latest_quarantined_path(&cfg).is_some(), + "startup check must quarantine a corrupt image" + ); +} + +/// A healthy DB passes untouched, and a missing DB (first boot) is a no-op — +/// the check must never quarantine what it cannot prove corrupt. +#[test] +fn startup_integrity_check_leaves_healthy_and_missing_dbs_alone() { + let (_tmp, cfg) = test_config(); + // Missing: no-op. + startup_integrity_check(&cfg); + assert!(latest_quarantined_path(&cfg).is_none()); + + // Healthy: create a real empty SQLite DB, check, and expect it in place. + let db_path = cfg.workspace_dir.join("memory_tree").join("chunks.db"); + std::fs::create_dir_all(db_path.parent().unwrap()).unwrap(); + let conn = rusqlite::Connection::open(&db_path).unwrap(); + conn.execute_batch("CREATE TABLE probe (id INTEGER PRIMARY KEY);") + .unwrap(); + drop(conn); + + startup_integrity_check(&cfg); + + assert!(db_path.exists(), "healthy DB must stay in place"); + assert!(latest_quarantined_path(&cfg).is_none()); +} + +// ── escalate_or_count (the tree-ingest sinks' shared arm) ──────────────── + +/// Non-corrupt failures are tolerated and counted; corruption aborts with the +/// recovery run and does NOT count — the two sinks (`PipelineHost`, +/// `HostSyncAdapter`) share this arm precisely so they cannot disagree again. +#[test] +fn escalate_or_count_splits_corrupt_from_tolerated() { + let (_tmp, cfg) = test_config(); + let counter = std::sync::atomic::AtomicU32::new(0); + + // Tolerated: an ordinary ingest failure returns Ok and increments. + let plain = anyhow::anyhow!("memory-tree ingest failed for source `x`: no such directory"); + assert!(escalate_or_count("test ingest", &cfg, plain, &counter).is_ok()); + assert_eq!(counter.load(std::sync::atomic::Ordering::Relaxed), 1); + + // Corrupt: the flattened incident shape returns Err and does not count. + let corrupt = anyhow::anyhow!( + "memory-tree ingest failed for source `github:o/r:42`: database disk image is malformed" + ); + let err = escalate_or_count("test ingest", &cfg, corrupt, &counter) + .expect_err("corruption must abort the run"); + assert!( + format!("{err:#}").contains("corrupt"), + "the abort must say why: {err:#}" + ); + assert_eq!( + counter.load(std::sync::atomic::Ordering::Relaxed), + 1, + "corruption is fatal, not a tolerated count" + ); +} diff --git a/crates/tinymemory-core/src/engine/mod.rs b/crates/tinymemory-core/src/engine/mod.rs index aec048c..63e1155 100644 --- a/crates/tinymemory-core/src/engine/mod.rs +++ b/crates/tinymemory-core/src/engine/mod.rs @@ -71,6 +71,8 @@ pub use sync::{ sync_context, HostSyncAdapter, RawCoverage, RawFileRef, RealCostAccumulator, RebuildOutcome, SourcePipelineFailure, HOST_SYNC_STATE_NAMESPACE, }; +// Crate-private seam for `crate::sources::sync` (openhuman#5820); not host surface. +pub(crate) use sync::run_source_pipeline_core; // The audit type, under the seam path OpenHuman already names // (`memory::tinycortex::SyncAuditEntry` embeds it in an RPC response type). // The type itself is core-owned (#18 §B1a); only the address is preserved. diff --git a/crates/tinymemory-core/src/engine/sync.rs b/crates/tinymemory-core/src/engine/sync.rs index a93783f..c8372d8 100644 --- a/crates/tinymemory-core/src/engine/sync.rs +++ b/crates/tinymemory-core/src/engine/sync.rs @@ -27,6 +27,11 @@ pub use tinycortex::memory::sync::{RawCoverage, RawFileRef, RealCostAccumulator, pub struct HostSyncAdapter { memory: MemoryClientRef, config: Option>, + /// Items whose skill-store write committed but whose (non-corrupt) tree + /// ingest failed during this adapter's run — the tolerated warns in + /// `store()`. Read back by [`run_source_pipeline_core`] so the run's + /// verdict can report "fetched, not tree-ingested" (openhuman#5820). + tree_ingest_failures: std::sync::atomic::AtomicU32, } #[derive(Debug)] @@ -57,6 +62,7 @@ impl HostSyncAdapter { Self { memory, config: None, + tree_ingest_failures: std::sync::atomic::AtomicU32::new(0), } } @@ -64,9 +70,16 @@ impl HostSyncAdapter { Self { memory, config: Some(config), + tree_ingest_failures: std::sync::atomic::AtomicU32::new(0), } } + /// Tolerated (non-corrupt) tree-ingest failures recorded so far. + fn tree_ingest_failure_count(&self) -> u32 { + self.tree_ingest_failures + .load(std::sync::atomic::Ordering::Relaxed) + } + /// Reconnect a synced Composio document to the memory tree (#5473). /// /// The TinyCortex migration (#4794) dropped the per-provider tree-ingest @@ -290,8 +303,13 @@ pub fn sync_context(memory: MemoryClientRef) -> SyncContext { } } -fn source_sync_context(memory: MemoryClientRef, config: &Config, local: bool) -> SyncContext { - let adapter = std::sync::Arc::new(HostSyncAdapter::with_config(memory, config.to_arc())); +/// [`source_sync_context`] over a caller-held adapter, so the caller can read +/// the adapter's per-run counters after the pipeline finishes. +fn context_over_adapter( + adapter: std::sync::Arc, + config: &Config, + local: bool, +) -> SyncContext { SyncContext { events: adapter.clone(), documents: adapter.clone(), @@ -309,6 +327,31 @@ pub async fn run_source_pipeline( source: &MemorySourceEntry, config: &Config, ) -> Result { + // Engine-typed view over `run_source_pipeline_core` for callers that speak + // the engine's `SyncOutcome`. The conversion drops `tree_ingest_failures` + // (the engine type has no field for it) — a caller that must see the tree + // half's verdict calls the `_core` variant instead. + let outcome = run_source_pipeline_core(source, config).await?; + Ok(SyncOutcome { + records_ingested: outcome.records_ingested, + more_pending: outcome.more_pending, + actions_called: outcome.actions_called, + provider_cost_usd: outcome.provider_cost_usd, + note: outcome.note, + }) +} + +/// [`run_source_pipeline`] returning the core pipelines' own +/// [`crate::sync::pipelines::traits::SyncOutcome`], which additionally carries +/// `tree_ingest_failures` — the "fetch committed, tree ingest did not" count a +/// sync verdict must not launder into success (openhuman#5820). The engine's +/// outcome type stays untouched; this is the boundary where the richer count +/// would otherwise be dropped. Crate-private: it is the seam +/// `crate::sources::sync` reads through, not host surface. +pub(crate) async fn run_source_pipeline_core( + source: &MemorySourceEntry, + config: &Config, +) -> Result { // Composio sources run on the engine-free pipelines (#18 §B1); this seam // keeps only the tree-coupled kinds (folder/repo/rss/web — they summarise // into the engine tree by design) and converts at the boundary for its @@ -329,7 +372,7 @@ pub async fn run_source_pipeline( .ok_or_else(|| { SourcePipelineFailure::without_usage("composio source missing connection_id") })?; - let outcome = crate::sync::pipelines::host::run_composio_connection_with_caps( + return crate::sync::pipelines::host::run_composio_connection_with_caps( &toolkit, connection_id, config, @@ -340,13 +383,6 @@ pub async fn run_source_pipeline( message: failure.message, actions_called: failure.actions_called, provider_cost_usd: failure.provider_cost_usd, - })?; - return Ok(SyncOutcome { - records_ingested: outcome.records_ingested, - more_pending: outcome.more_pending, - actions_called: outcome.actions_called, - provider_cost_usd: outcome.provider_cost_usd, - note: outcome.note, }); } @@ -366,12 +402,14 @@ pub async fn run_source_pipeline( dispatcher .register(pipeline) .map_err(|error| SourcePipelineFailure::without_usage(error.to_string()))?; - dispatcher - .tick( - &pipeline_id, - &memory_config, - &source_sync_context(memory, config, source.kind != SourceKind::Composio), - ) + // Built from an adapter handle this fn keeps, rather than through + // `source_sync_context`, so the tolerated tree-ingest failure count can be + // read back after the run. + let adapter = std::sync::Arc::new(HostSyncAdapter::with_config(memory, config.to_arc())); + let context = + context_over_adapter(adapter.clone(), config, source.kind != SourceKind::Composio); + let outcome = dispatcher + .tick(&pipeline_id, &memory_config, &context) .await .map_err(|error| { let usage = error.downcast_ref::(); @@ -380,7 +418,15 @@ pub async fn run_source_pipeline( actions_called: usage.map_or(0, |error| error.actions_called), provider_cost_usd: usage.map_or(0.0, |error| error.provider_cost_usd), } - }) + })?; + Ok(crate::sync::pipelines::traits::SyncOutcome { + records_ingested: outcome.records_ingested, + more_pending: outcome.more_pending, + actions_called: outcome.actions_called, + provider_cost_usd: outcome.provider_cost_usd, + note: outcome.note, + tree_ingest_failures: adapter.tree_ingest_failure_count(), + }) } /// Run a Composio connection through tinycortex, preserving any source-level @@ -590,13 +636,17 @@ impl SkillDocSink for HostSyncAdapter { // #5473: additively reconnect the synced item to the memory tree. This // is a best-effort secondary index over the skill store, which is the - // source of truth and has already committed above. A failure here must - // NOT abort the connector sync: most providers do not tolerate scope - // errors, so the orchestrator turns a `store` error into a run-aborting - // `Err` — propagating would let one deterministically-poisonous item - // stall the whole connection and re-fetch the page (Composio spend) on - // every retry. Log and continue; the per-item source gate re-attempts - // the item on a later sync, and an operator rebuild can backfill. + // source of truth and has already committed above. An ordinary failure + // here must NOT abort the connector sync: most providers do not + // tolerate scope errors, so the orchestrator turns a `store` error + // into a run-aborting `Err` — propagating would let one + // deterministically-poisonous item stall the whole connection and + // re-fetch the page (Composio spend) on every retry. Log, count, and + // continue; the per-item source gate re-attempts the item on a later + // sync, and an operator rebuild can backfill. Corruption is the + // exception (openhuman#5820): a malformed `chunks.db` fails every + // later item identically, so it escalates through the shared recovery + // and aborts the run — there is nothing per-item about it. // The config-less adapter (`sync_context`) has no ingest pipeline and is // not on the connector sync path, so it skips tree ingest entirely. if let Some(config) = self.config.as_deref() { @@ -604,11 +654,18 @@ impl SkillDocSink for HostSyncAdapter { .ingest_document_into_memory_tree(config, &document) .await { + let rendered = format!("{error:#}"); + crate::corruption::escalate_or_count( + "connector tree ingest", + config, + error, + &self.tree_ingest_failures, + )?; tracing::warn!( toolkit = %document.toolkit, connection_id = %document.connection_id, document_id = %document.document_id, - %error, + error = %rendered, "[tinycortex:sync] memory-tree ingest failed; skill store retained" ); } diff --git a/crates/tinymemory-core/src/engine/sync_tests.rs b/crates/tinymemory-core/src/engine/sync_tests.rs index 36459f3..940a0a5 100644 --- a/crates/tinymemory-core/src/engine/sync_tests.rs +++ b/crates/tinymemory-core/src/engine/sync_tests.rs @@ -8,6 +8,18 @@ use crate::sources::MemorySourceEntry; use crate::sync::composio::{get_composio_sync_provider, init_default_composio_sync_providers}; use crate::sync::pipelines::host::{is_composio_toolkit_syncable, syncable_composio_toolkits}; +/// The context the production path used to build inline; kept here since +/// `run_source_pipeline_core` took over that call site with a caller-held +/// adapter (see `context_over_adapter`). +fn source_sync_context( + memory: crate::store::MemoryClientRef, + config: &crate::Config, + local: bool, +) -> tinycortex::memory::sync::SyncContext { + let adapter = std::sync::Arc::new(super::HostSyncAdapter::with_config(memory, config.to_arc())); + super::context_over_adapter(adapter, config, local) +} + fn memory_fixture() -> ( tempfile::TempDir, tinymemory_api::host::test_support::TestHostConfig, @@ -50,12 +62,12 @@ async fn failure_and_context_helpers_preserve_contract_state() { assert!(context.external_sources.is_none()); assert!(context.summariser.is_none()); - let local = super::source_sync_context(client.clone(), &config, true); + let local = source_sync_context(client.clone(), &config, true); assert!(local.local_documents.is_some()); assert!(local.external_sources.is_some()); assert!(local.summariser.is_some()); - let remote = super::source_sync_context(client, &config, false); + let remote = source_sync_context(client, &config, false); assert!(remote.local_documents.is_none()); assert!(remote.external_sources.is_none()); assert!(remote.summariser.is_none()); @@ -672,11 +684,18 @@ async fn tree_ingest_failure_is_tolerated_and_skill_store_is_retained() { "the broken tree-ingest workspace must make ingest fail" ); - // `store` must swallow that tree-ingest failure and still succeed. + // `store` must swallow that tree-ingest failure and still succeed — + // and count it, so the run's verdict can report the tree half honestly + // (openhuman#5820). adapter .store(document) .await .expect("store must tolerate a memory-tree ingest failure (best-effort tree)"); + assert_eq!( + adapter.tree_ingest_failure_count(), + 1, + "a tolerated tree-ingest failure must be counted for the run's verdict" + ); // The skill store, committed before the tree half, still holds the item — // best-effort tree ingest must never cost the durable skill write. diff --git a/crates/tinymemory-core/src/lib.rs b/crates/tinymemory-core/src/lib.rs index 63a97c4..8d9b5fb 100644 --- a/crates/tinymemory-core/src/lib.rs +++ b/crates/tinymemory-core/src/lib.rs @@ -36,6 +36,7 @@ pub mod chat_host; pub mod composio_host; pub mod config_loader; pub mod conversations; +pub(crate) mod corruption; pub mod diff; pub mod embedding_adapter; pub mod embedding_host; diff --git a/crates/tinymemory-core/src/queue/worker.rs b/crates/tinymemory-core/src/queue/worker.rs index b0822f5..a7cacf2 100644 --- a/crates/tinymemory-core/src/queue/worker.rs +++ b/crates/tinymemory-core/src/queue/worker.rs @@ -28,6 +28,7 @@ use crate::Config; // legacy `handlers`, per-job settle (`mark_*`/`scrub_for_log`), and claim // helpers are gone from this module. Only startup lock recovery + the loop's // storage-degraded signalling remain host. +use crate::corruption::is_sqlite_corrupt; use crate::queue::store::{recover_stale_locks, release_running_locks}; use crate::tree::health::{clear_storage_degraded, mark_storage_degraded, FailureCode}; @@ -53,14 +54,6 @@ 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); - /// Process-wide latch so a persistent host-filesystem failure (EIO/ENOSPC/ /// EROFS on the memory_tree dir/DB path) is reported to Sentry **once**, not on /// every poll from every worker. Set on the first host-I/O failure; cleared on @@ -95,6 +88,18 @@ pub fn start(config: Arc) { log::warn!("[memory::jobs] recover_stale_locks failed at startup: {err:#}"); } + // One-shot integrity check of the chunk DB (openhuman#5820 item 5): + // workspaces written by pre-#5725 builds can carry latent page damage + // that otherwise surfaces hours later through whichever call walks a + // damaged b-tree first. `quick_check` reads the whole file, so it runs + // on a blocking thread while the workers start normally — a corrupt + // verdict quarantines + rebuilds via the same recovery the runtime + // classifiers use, and the workers' next poll sees the fresh store. + let integrity_cfg = config.to_arc(); + tokio::task::spawn_blocking(move || { + crate::corruption::startup_integrity_check(&*integrity_cfg); + }); + // Release in-flight locks on graceful shutdown so a clean restart // re-claims the work immediately instead of waiting out the lease // (which surfaced as a stale-lock recovery warn on every launch). @@ -207,13 +212,18 @@ pub fn start(config: Arc) { // 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); + // drive quarantine+rebuild recovery (shared with + // the ingest and startup paths in + // `crate::corruption` so every detector escalates + // identically), then back off long so a failed + // recovery never re-floods. `notify` still wakes + // us on new enqueues once the rebuild succeeds. + crate::corruption::report_and_recover( + &format!("jobs worker {idx}"), + "tree_jobs_worker_corrupt", + &err, + &*cfg, + ); tokio::time::sleep(Duration::from_secs(300)).await; } else if is_host_io_error(&err) { // Persistent host-filesystem failure (EIO 5 / @@ -382,36 +392,6 @@ 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") -} - /// Classify whether an error is a **persistent host-filesystem failure** — /// `std::fs::create_dir_all` / file open returning an OS-level I/O error on the /// memory_tree path. Matches the three persistent, user-only-fixable POSIX @@ -446,53 +426,6 @@ fn is_host_io_error(err: &anyhow::Error) -> bool { msg.contains("(os error 5)") || msg.contains("(os error 28)") || msg.contains("(os error 30)") } -/// 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::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::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::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)] #[path = "worker_tests.rs"] mod tests; diff --git a/crates/tinymemory-core/src/queue/worker_tests.rs b/crates/tinymemory-core/src/queue/worker_tests.rs index 2f96f3b..72c9751 100644 --- a/crates/tinymemory-core/src/queue/worker_tests.rs +++ b/crates/tinymemory-core/src/queue/worker_tests.rs @@ -250,103 +250,6 @@ fn is_sqlite_disk_full_does_not_match_other_errors() { ))); } -// ── 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" - ))); -} - // ── is_host_io_error tests (CORE-RUST-19J) ─────────────────────────────── /// EIO (`os error 5`) is the CORE-RUST-19J signal: `create_dir_all` on a @@ -427,40 +330,6 @@ fn is_host_io_error_does_not_match_other_errors() { ))); } -/// 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/crates/tinymemory-core/src/sources/sync.rs b/crates/tinymemory-core/src/sources/sync.rs index 5f7cf71..a51aa2b 100644 --- a/crates/tinymemory-core/src/sources/sync.rs +++ b/crates/tinymemory-core/src/sources/sync.rs @@ -87,13 +87,20 @@ pub async fn sync_source(source: MemorySourceEntry, config: Arc) -> Resu // Composio billable-action usage for this run, populated by // `sync_composio` (#3111). Stays zero for non-Composio kinds. let mut composio_usage = ComposioUsage::default(); + // Every kind runs through `run_source_pipeline_core`, whose + // outcome carries `tree_ingest_failures` — the count of items that + // were fetched-and-stored but never reached the memory tree. The + // engine-typed `run_source_pipeline` would drop it (openhuman#5820). let outcome = match source.kind { SourceKind::Composio => { - match crate::engine::run_source_pipeline(&source, &*config).await { + match crate::engine::run_source_pipeline_core(&source, &*config).await { Ok(outcome) => { composio_usage.actions_called = outcome.actions_called; composio_usage.cost_usd = outcome.provider_cost_usd; - Ok(outcome.records_ingested as usize) + Ok(( + outcome.records_ingested as usize, + outcome.tree_ingest_failures, + )) } Err(error) => { composio_usage.actions_called = error.actions_called; @@ -102,43 +109,59 @@ pub async fn sync_source(source: MemorySourceEntry, config: Arc) -> Resu } } } - SourceKind::Conversation | SourceKind::Folder => { - crate::engine::run_source_pipeline(&source, &*config) - .await - .map(|outcome| outcome.records_ingested as usize) - .map_err(|error| error.to_string()) - } - SourceKind::GithubRepo => crate::engine::run_source_pipeline(&source, &*config) + SourceKind::Conversation + | SourceKind::Folder + | SourceKind::GithubRepo + | SourceKind::RssFeed + | SourceKind::WebPage => crate::engine::run_source_pipeline_core(&source, &*config) .await - .map(|outcome| outcome.records_ingested as usize) + .map(|outcome| { + ( + outcome.records_ingested as usize, + outcome.tree_ingest_failures, + ) + }) .map_err(|error| error.to_string()), - SourceKind::RssFeed | SourceKind::WebPage => { - crate::engine::run_source_pipeline(&source, &*config) - .await - .map(|outcome| outcome.records_ingested as usize) - .map_err(|error| error.to_string()) - } SourceKind::TwitterQuery => Err( "Twitter sync not yet configured. Provide bearer token in settings." .to_string(), ), }; - let duration_ms = sync_start.elapsed().as_millis() as u64; match outcome { - Ok(items) => { + Ok((items, pipeline_tree_failures)) => { + // Auto-rebuild BEFORE the verdict (openhuman#5820): if raw + // files exist but the tree has no summaries, build the + // tree now — its failures are part of this run's truth, + // and stamping the audit line first is how a corrupt store + // ran for 34 minutes behind a column of green rows. + let reconcile_errors = check_and_rebuild_tree(&source, &*config).await; + let duration_ms = sync_start.elapsed().as_millis() as u64; + + let verdict = run_verdict(items, pipeline_tree_failures, &reconcile_errors); tracing::debug!( source_id = %source.id, kind = %source.kind.as_str(), items = items, - "[memory_sources:sync] completed" + tree_failures = verdict.tree_failures, + "[memory_sources:sync] pipeline finished" ); + if !verdict.success { + tracing::warn!( + source_id = %source.id, + kind = %source.kind.as_str(), + items = items, + tree_failures = verdict.tree_failures, + "[memory_sources:sync] fetch succeeded but the memory-tree \ + half failed; reporting the run as failed" + ); + } emit_sync_stage( MemorySyncTrigger::Manual, - MemorySyncStage::Completed, + verdict.stage, Some(source.kind.as_str()), Some(&source.id), - Some(format!("ingested {items} item(s)")), + Some(verdict.detail), Some(&source.id), ); @@ -163,18 +186,18 @@ pub async fn sync_source(source: MemorySourceEntry, config: Arc) -> Resu composio_cost_usd: composio_usage.cost_usd, actual_charged_usd: None, duration_ms, - success: true, + success: verdict.success, error: None, + tree_ingest_failures: verdict.tree_failures, + tree_error: verdict.tree_error, }, ) { tracing::warn!(%error, "[memory_sync:audit] append failed"); } - // Auto-rebuild: if raw files exist but the tree has - // no summaries, build the tree now. - check_and_rebuild_tree(&source, &*config).await; - // Auto-snapshot: capture post-sync state for diff tracking. + // The raw archive committed even when the tree half failed, + // so the snapshot stays correct either way. if let Err(e) = crate::diff::ops::auto_snapshot_after_sync(&source, &*config).await { @@ -186,6 +209,7 @@ pub async fn sync_source(source: MemorySourceEntry, config: Arc) -> Resu } } Err(error) => { + let duration_ms = sync_start.elapsed().as_millis() as u64; // Audit failed syncs too. use crate::sync::audit::{append_audit_entry, SyncAuditEntry}; if let Err(error) = append_audit_entry( @@ -210,6 +234,8 @@ pub async fn sync_source(source: MemorySourceEntry, config: Arc) -> Resu duration_ms, success: false, error: Some(error.clone()), + tree_ingest_failures: 0, + tree_error: None, }, ) { tracing::warn!(%error, "[memory_sync:audit] append failed"); @@ -268,10 +294,78 @@ pub async fn sync_source(source: MemorySourceEntry, config: Arc) -> Resu Ok(()) } +/// What a finished (fetch-successful) run reports, folding the tree half in. +#[derive(Clone, Debug, PartialEq)] +struct RunVerdict { + stage: MemorySyncStage, + detail: String, + success: bool, + /// Items fetched-and-stored whose tree ingest failed — an item count, the + /// unit `SyncAuditEntry::tree_ingest_failures` is defined in. Reconcile + /// failures are per scope and are NOT folded into this number. + tree_failures: u32, + tree_error: Option, +} + +/// Fold the tree half into the run's verdict (openhuman#5820). +/// +/// A run whose fetch and skill-store committed but whose tree ingest dropped +/// items, or whose post-run reconcile failed, must NOT read as success — that +/// is the exact shape that hid a corrupt store behind 34 minutes of green +/// sync rows. Reporting it failed with the fetch count intact is the +/// recoverable direction: the next sync retries the tree half, nothing +/// fetched is lost. +/// +/// Item failures and reconcile failures are different units (items vs +/// scopes) and both diagnostics are kept: the item count is what the audit +/// row stores, and `tree_error` / `detail` name whichever halves failed. +fn run_verdict( + items: usize, + pipeline_tree_failures: u32, + reconcile_errors: &[String], +) -> RunVerdict { + let mut problems: Vec = Vec::new(); + if pipeline_tree_failures > 0 { + problems.push(format!( + "{pipeline_tree_failures} item(s) fetched but not ingested into the memory tree" + )); + } + problems.extend(reconcile_errors.iter().cloned()); + if problems.is_empty() { + return RunVerdict { + stage: MemorySyncStage::Completed, + detail: format!("ingested {items} item(s)"), + success: true, + tree_failures: 0, + tree_error: None, + }; + } + let tree_error = problems.join("; "); + RunVerdict { + stage: MemorySyncStage::Failed, + detail: format!( + "fetched {items} item(s) but the memory-tree half failed ({tree_error}); \ + tree-backed recall is missing these items" + ), + success: false, + tree_failures: pipeline_tree_failures, + tree_error: Some(tree_error), + } +} + /// Reconcile raw files that are not yet covered by tree summaries. -pub(crate) async fn check_and_rebuild_tree(source: &MemorySourceEntry, config: &Config) { +/// +/// Returns one message per failed scope so the caller can fold reconcile +/// failures into the run's verdict instead of the run reading green over a +/// tree that received nothing (openhuman#5820). A corrupt store additionally +/// escalates through the shared recovery before being returned. +pub(crate) async fn check_and_rebuild_tree( + source: &MemorySourceEntry, + config: &Config, +) -> Vec { use crate::engine::{needs_rebuild, rebuild_tree_from_raw}; + let mut failures = Vec::new(); for scope in derive_scopes(source, config) { if !needs_rebuild(config, &scope.tree_scope, &scope.archive_source_id) { continue; @@ -291,13 +385,28 @@ pub(crate) async fn check_and_rebuild_tree(source: &MemorySourceEntry, config: & cost_is_actual = outcome.actual_charged_usd.is_some(), "[memory_sources:sync] reconcile complete" ), - Err(error) => tracing::warn!( - scope = %scope.tree_scope, - error = %format!("{error:#}"), - "[memory_sources:sync] reconcile failed" - ), + Err(error) => { + if crate::corruption::is_sqlite_corrupt(&error) { + crate::corruption::report_and_recover( + "tree reconcile", + "tree_ingest_corrupt", + &error, + config, + ); + } + tracing::warn!( + scope = %scope.tree_scope, + error = %format!("{error:#}"), + "[memory_sources:sync] reconcile failed" + ); + failures.push(format!( + "reconcile failed for scope `{}`: {error:#}", + scope.tree_scope + )); + } } } + failures } /// A source's tree scope paired with its raw-archive source id. The two diff --git a/crates/tinymemory-core/src/sources/sync_tests.rs b/crates/tinymemory-core/src/sources/sync_tests.rs index ba2c4ca..67b1c3b 100644 --- a/crates/tinymemory-core/src/sources/sync_tests.rs +++ b/crates/tinymemory-core/src/sources/sync_tests.rs @@ -202,7 +202,7 @@ fn derive_scopes_fails_closed_and_reads_only_valid_gmail_archives() { #[tokio::test] async fn rebuild_check_is_a_noop_for_sources_without_archive_scopes() { let config = TestHostConfig::default(); - check_and_rebuild_tree( + let failures = check_and_rebuild_tree( &source( "folder", "folder-no-rebuild", @@ -211,4 +211,74 @@ async fn rebuild_check_is_a_noop_for_sources_without_archive_scopes() { &config, ) .await; + assert!(failures.is_empty(), "a no-op reconcile reports no failures"); +} + +/// The #5820 verdict table: a clean tree half completes; any dropped item — +/// from the pipeline's tolerated ingest failures or from a failed reconcile — +/// flips the run to Failed with the fetch count intact, and carries a +/// tree_error for the audit row. Item failures (an item count) and reconcile +/// failures (per scope) stay separate units, and both diagnostics survive +/// when they coexist. A false ✓ here is the unrecoverable direction (the user +/// never learns recall is missing items); a false ✗ costs one re-sync. +#[test] +fn run_verdict_folds_the_tree_half_into_the_outcome() { + use crate::sync_events::MemorySyncStage; + + let clean = run_verdict(250, 0, &[]); + assert!(clean.success); + assert_eq!(clean.stage, MemorySyncStage::Completed); + assert_eq!(clean.tree_failures, 0); + assert!(clean.tree_error.is_none()); + assert_eq!(clean.detail, "ingested 250 item(s)"); + + let dropped = run_verdict(250, 3, &[]); + assert!(!dropped.success); + assert_eq!(dropped.stage, MemorySyncStage::Failed); + assert_eq!(dropped.tree_failures, 3); + assert!(dropped + .tree_error + .as_deref() + .is_some_and(|error| error.contains("3 item(s) fetched but not ingested"))); + assert!(dropped.detail.contains("fetched 250 item(s)")); + + let reconcile_failed = run_verdict( + 10, + 0, + &["reconcile failed for scope `gmail:user`: database disk image is malformed".to_string()], + ); + assert!(!reconcile_failed.success); + assert_eq!(reconcile_failed.stage, MemorySyncStage::Failed); + assert_eq!( + reconcile_failed.tree_failures, 0, + "a failed reconcile scope is not an item count" + ); + assert!(reconcile_failed + .tree_error + .as_deref() + .is_some_and(|error| error.contains("reconcile failed for scope"))); + + // Both halves failing: the item count stays an item count and neither + // diagnostic is dropped. + let both = run_verdict( + 10, + 2, + &["reconcile failed for scope `gmail:user`: boom".to_string()], + ); + assert!(!both.success); + assert_eq!(both.tree_failures, 2); + let error = both.tree_error.as_deref().expect("combined diagnostics"); + assert!( + error.contains("2 item(s) fetched but not ingested"), + "{error}" + ); + assert!( + error.contains("reconcile failed for scope `gmail:user`: boom"), + "{error}" + ); + assert!( + both.detail.contains("2 item(s)") && both.detail.contains("boom"), + "{}", + both.detail + ); } diff --git a/crates/tinymemory-core/src/sync/audit.rs b/crates/tinymemory-core/src/sync/audit.rs index f084af2..04101bf 100644 --- a/crates/tinymemory-core/src/sync/audit.rs +++ b/crates/tinymemory-core/src/sync/audit.rs @@ -46,6 +46,25 @@ pub struct SyncAuditEntry { pub success: bool, #[serde(skip_serializing_if = "Option::is_none")] pub error: Option, + /// Items fetched-and-stored whose memory-tree ingest failed + /// (openhuman#5820). Both tree-ingest sinks (`PipelineHost` and the + /// engine-side `HostSyncAdapter`) count tolerated failures, and the + /// source-sync and periodic writers in this crate store that count here; + /// only the legacy engine-typed `run_source_pipeline` conversion drops it, + /// and the engine's own rebuild writer never sets it. `0` is skipped on + /// the wire so those rows stay byte-identical to this writer's healthy + /// rows. + #[serde(default, skip_serializing_if = "is_zero_u32")] + pub tree_ingest_failures: u32, + /// Why the tree half failed, when it did. Never memory content. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub tree_error: Option, +} + +/// `skip_serializing_if` gate for the additive counters above. +#[allow(clippy::trivially_copy_pass_by_ref)] // serde's contract is a reference +fn is_zero_u32(value: &u32) -> bool { + *value == 0 } impl SyncAuditEntry { diff --git a/crates/tinymemory-core/src/sync/audit_tests.rs b/crates/tinymemory-core/src/sync/audit_tests.rs index 986310b..ac83aaa 100644 --- a/crates/tinymemory-core/src/sync/audit_tests.rs +++ b/crates/tinymemory-core/src/sync/audit_tests.rs @@ -21,6 +21,8 @@ fn entry() -> SyncAuditEntry { duration_ms: 1234, success: true, error: None, + tree_ingest_failures: 0, + tree_error: None, } } @@ -50,6 +52,31 @@ fn audit_line_format_is_pinned() { ); } +/// The #5820 fields are skip-if-empty precisely so the healthy line above +/// stays byte-identical to the engine writer's; a run with a failed tree half +/// serialises them, and a reader of engine-written rows (which never carry +/// them) defaults both. This pins the failure shape and the tolerant read. +#[test] +fn tree_failure_fields_serialise_only_when_set_and_default_on_read() { + let mut failed = entry(); + failed.success = false; + failed.tree_ingest_failures = 5; + failed.tree_error = Some("database disk image is malformed".into()); + let line = serde_json::to_string(&failed).unwrap(); + assert!(line.contains("\"tree_ingest_failures\":5")); + assert!(line.contains("\"tree_error\":\"database disk image is malformed\"")); + + // An engine-written row (no #5820 fields) reads back with defaults. + let legacy: SyncAuditEntry = serde_json::from_str( + "{\"timestamp\":\"2026-01-02T03:04:05Z\",\"source_id\":\"s\",\"source_kind\":\"k\",\ + \"scope\":\"u\",\"items_fetched\":1,\"batches\":0,\"input_tokens\":0,\ + \"output_tokens\":0,\"estimated_cost_usd\":0.0,\"duration_ms\":1,\"success\":true}", + ) + .unwrap(); + assert_eq!(legacy.tree_ingest_failures, 0); + assert!(legacy.tree_error.is_none()); +} + #[test] fn append_then_read_round_trips_newest_first() { let tmp = tempfile::tempdir().unwrap(); diff --git a/crates/tinymemory-core/src/sync/composio/periodic.rs b/crates/tinymemory-core/src/sync/composio/periodic.rs index 42b74fb..5d92717 100644 --- a/crates/tinymemory-core/src/sync/composio/periodic.rs +++ b/crates/tinymemory-core/src/sync/composio/periodic.rs @@ -571,13 +571,24 @@ pub(crate) async fn run_one_tick() -> Result<(), String> { actions_called: outcome.actions_called, cost_usd: outcome.provider_cost_usd, }; - tracing::debug!( - toolkit = %conn.toolkit, - connection_id = %conn.id, - items = outcome.records_ingested, - composio_actions = usage.actions_called, - "[composio:periodic] sync ok" - ); + if outcome.tree_ingest_failures > 0 { + tracing::warn!( + toolkit = %conn.toolkit, + connection_id = %conn.id, + items = outcome.records_ingested, + tree_failures = outcome.tree_ingest_failures, + "[composio:periodic] fetch ok but the memory-tree half dropped \ + items; auditing the run as failed" + ); + } else { + tracing::debug!( + toolkit = %conn.toolkit, + connection_id = %conn.id, + items = outcome.records_ingested, + composio_actions = usage.actions_called, + "[composio:periodic] sync ok" + ); + } let entry = build_periodic_audit_entry( &toolkit, &conn.id, @@ -585,6 +596,7 @@ pub(crate) async fn run_one_tick() -> Result<(), String> { outcome.records_ingested as usize, duration_ms, None, + outcome.tree_ingest_failures, ); if let Err(error) = append_audit_entry(config.workspace_dir(), &entry) { tracing::warn!(%error, "[memory_sync:audit] append failed"); @@ -612,6 +624,7 @@ pub(crate) async fn run_one_tick() -> Result<(), String> { 0, duration_ms, Some(e.to_string()), + 0, ); if let Err(error) = append_audit_entry(config.workspace_dir(), &entry) { tracing::warn!(%error, "[memory_sync:audit] append failed"); @@ -693,7 +706,14 @@ fn build_periodic_audit_entry( items_ingested: usize, duration_ms: u64, error: Option, + tree_ingest_failures: u32, ) -> SyncAuditEntry { + // A run whose fetch committed but whose tree half dropped items must not + // read as success in Sync History (openhuman#5820). + let success = error.is_none() && tree_ingest_failures == 0; + let tree_error = (tree_ingest_failures > 0).then(|| { + format!("{tree_ingest_failures} item(s) fetched but not ingested into the memory tree") + }); SyncAuditEntry { timestamp: chrono::Utc::now(), source_id: connection_id.to_string(), @@ -708,8 +728,10 @@ fn build_periodic_audit_entry( composio_cost_usd: usage.cost_usd, actual_charged_usd: None, duration_ms, - success: error.is_none(), + success, error, + tree_ingest_failures, + tree_error, } } diff --git a/crates/tinymemory-core/src/sync/composio/periodic_tests.rs b/crates/tinymemory-core/src/sync/composio/periodic_tests.rs index cdde4ab..d911676 100644 --- a/crates/tinymemory-core/src/sync/composio/periodic_tests.rs +++ b/crates/tinymemory-core/src/sync/composio/periodic_tests.rs @@ -260,6 +260,8 @@ fn audit_entry( duration_ms: 10, success, error: None, + tree_ingest_failures: 0, + tree_error: None, } } @@ -380,7 +382,7 @@ fn periodic_audit_entry_records_composio_cost_on_success() { actions_called: 3, cost_usd: 0.042, }; - let entry = build_periodic_audit_entry("gmail", "cmp-123", &usage, 17, 1234, None); + let entry = build_periodic_audit_entry("gmail", "cmp-123", &usage, 17, 1234, None, 0); assert_eq!(entry.source_kind, "composio"); assert_eq!(entry.source_id, "cmp-123"); @@ -413,6 +415,7 @@ fn periodic_audit_entry_preserves_partial_cost_on_failure() { 0, 500, Some("fetch timed out".to_string()), + 0, ); assert!(!entry.success); diff --git a/crates/tinymemory-core/src/sync/pipelines/composio/orchestrator.rs b/crates/tinymemory-core/src/sync/pipelines/composio/orchestrator.rs index 1f19dbf..f036e31 100644 --- a/crates/tinymemory-core/src/sync/pipelines/composio/orchestrator.rs +++ b/crates/tinymemory-core/src/sync/pipelines/composio/orchestrator.rs @@ -425,7 +425,14 @@ async fn run_pages( tokens_ingested = tokens_ingested.saturating_add((document.content.len() / 4) as u64); if let Err(error) = context.documents.store(document).await { - if source.tolerate_scope_errors() { + // Scope tolerance exists for per-item flakiness. A corrupt + // store is not that: every later item fails identically in + // every scope, so tolerating it here re-buys the + // openhuman#5820 flood one scope at a time. Corruption + // always aborts the run. + if source.tolerate_scope_errors() + && !crate::corruption::is_sqlite_corrupt(&error) + { tracing::warn!(toolkit = source.toolkit(), connection_id, scope = %scope.label, %error, "[sync:orchestrator] scope document store failed; continuing"); scope_failed = true; break; @@ -494,6 +501,9 @@ async fn run_pages( actions_called: state.run_requests, provider_cost_usd: state.run_provider_cost_usd, note: None, + // Stamped by the runner from the sink's counter; the orchestrator only + // sees store()'s Ok/Err and the tolerated failures return Ok. + tree_ingest_failures: 0, }) } diff --git a/crates/tinymemory-core/src/sync/pipelines/composio/providers/slack.rs b/crates/tinymemory-core/src/sync/pipelines/composio/providers/slack.rs index 5a7ea59..2b032fb 100644 --- a/crates/tinymemory-core/src/sync/pipelines/composio/providers/slack.rs +++ b/crates/tinymemory-core/src/sync/pipelines/composio/providers/slack.rs @@ -188,6 +188,7 @@ impl SlackSearchBackfillPipeline { note: Some(format!( "slack search-backfill: pages={page} records={stored}" )), + tree_ingest_failures: 0, }) } } diff --git a/crates/tinymemory-core/src/sync/pipelines/dispatcher_tests.rs b/crates/tinymemory-core/src/sync/pipelines/dispatcher_tests.rs index a5b40d8..fe85502 100644 --- a/crates/tinymemory-core/src/sync/pipelines/dispatcher_tests.rs +++ b/crates/tinymemory-core/src/sync/pipelines/dispatcher_tests.rs @@ -40,6 +40,7 @@ impl SyncPipeline for FakePipeline { actions_called: 0, provider_cost_usd: 0.0, note: None, + tree_ingest_failures: 0, }) } } diff --git a/crates/tinymemory-core/src/sync/pipelines/host.rs b/crates/tinymemory-core/src/sync/pipelines/host.rs index bb0abf3..c1fa1b1 100644 --- a/crates/tinymemory-core/src/sync/pipelines/host.rs +++ b/crates/tinymemory-core/src/sync/pipelines/host.rs @@ -59,6 +59,12 @@ impl PipelineFailure { pub struct PipelineHost { memory: MemoryClientRef, config: Option>, + /// Items whose skill-store write committed but whose (non-corrupt) tree + /// ingest failed during this adapter's run. Read back into + /// [`SyncOutcome::tree_ingest_failures`] by the runners, because the + /// orchestrator only sees `store()`'s `Ok`/`Err` and the tolerated + /// failures deliberately return `Ok` (openhuman#5820). + tree_ingest_failures: std::sync::atomic::AtomicU32, } impl PipelineHost { @@ -68,6 +74,7 @@ impl PipelineHost { Self { memory, config: Some(config), + tree_ingest_failures: std::sync::atomic::AtomicU32::new(0), } } @@ -77,6 +84,7 @@ impl PipelineHost { Self { memory, config: None, + tree_ingest_failures: std::sync::atomic::AtomicU32::new(0), } } @@ -88,6 +96,12 @@ impl PipelineHost { state: self.clone(), } } + + /// Tolerated (non-corrupt) tree-ingest failures recorded so far. + pub fn tree_ingest_failures(&self) -> u32 { + self.tree_ingest_failures + .load(std::sync::atomic::Ordering::Relaxed) + } } #[async_trait] @@ -117,13 +131,25 @@ impl SkillDocSink for PipelineHost { // #5473: additively reconnect the synced item to the memory tree — a // best-effort secondary index; the skill store above is the source of - // truth and has committed. A failure here must NOT abort the sync (one - // poisonous item would stall the connection and re-buy the page on - // every retry). The config-less adapter skips tree ingest entirely. + // truth and has committed. An ordinary failure here must NOT abort the + // sync (one poisonous item would stall the connection and re-buy the + // page on every retry), but it is COUNTED so the run's verdict can say + // "fetched, not tree-ingested" instead of success. Corruption is the + // exception (openhuman#5820): a malformed `chunks.db` fails every + // later item identically — 747 warns in 34 minutes in the incident — + // so it escalates through the shared recovery and aborts the run. + // The config-less adapter skips tree ingest entirely. if let Some(config) = self.config.as_deref() { if let Err(error) = ingest_into_tree(config, &document).await { + let rendered = format!("{error:#}"); + crate::corruption::escalate_or_count( + "composio tree ingest", + config, + error, + &self.tree_ingest_failures, + )?; tracing::warn!( - %error, + error = %rendered, document_id = %document.document_id, "[memory_sync] tree ingest failed; skill store remains authoritative" ); @@ -380,14 +406,16 @@ pub async fn run_composio_connection_with_caps( max_cost_per_sync_usd: caps.max_cost_per_sync_usd, }; let host = Arc::new(PipelineHost::new(memory, config.to_arc())); - run_pipeline( + let mut outcome = run_pipeline( pipeline, toolkit, connection_id, &pipeline_config, &host.context(), ) - .await + .await?; + outcome.tree_ingest_failures = host.tree_ingest_failures(); + Ok(outcome) } /// Run a bounded Gmail backfill through the engine-free pipelines. @@ -410,14 +438,16 @@ pub async fn run_gmail_backfill( // The backfill drives the Gmail pipeline, which keys its `SyncState` on // `"gmail"`; naming the same toolkit here puts it behind the same guard as // a periodic or RPC Gmail sync of this connection. - run_pipeline( + let mut outcome = run_pipeline( pipeline, "gmail", connection_id, &PipelineConfig::default(), &host.context(), ) - .await + .await?; + outcome.tree_ingest_failures = host.tree_ingest_failures(); + Ok(outcome) } /// Run the Slack search backfill through the engine-free pipelines. @@ -439,14 +469,16 @@ pub async fn run_slack_search_backfill( // `SlackSearchBackfillPipeline` loads and saves the same // `("slack", connection_id)` state the Slack sync pipeline does, so the two // must share one guard or they clobber each other's cursor and budget. - run_pipeline( + let mut outcome = run_pipeline( pipeline, "slack", connection_id, &PipelineConfig::default(), &host.context(), ) - .await + .await?; + outcome.tree_ingest_failures = host.tree_ingest_failures(); + Ok(outcome) } /// The note a run carries when another run already holds its connection. diff --git a/crates/tinymemory-core/src/sync/pipelines/host_tests.rs b/crates/tinymemory-core/src/sync/pipelines/host_tests.rs index a81e087..a84d040 100644 --- a/crates/tinymemory-core/src/sync/pipelines/host_tests.rs +++ b/crates/tinymemory-core/src/sync/pipelines/host_tests.rs @@ -452,3 +452,45 @@ async fn pipeline_host_state_event_and_delete_capabilities_round_trip() { assert!(!sink.drain().is_empty()); host.delete("gmail", "missing-document").await.unwrap(); } + +/// An ordinary (non-corrupt) tree-ingest failure is tolerated — `store` +/// returns `Ok` and the skill store keeps the item — but it is COUNTED, so the +/// run's verdict can say "fetched, not tree-ingested" instead of reading as +/// full success (openhuman#5820). The broken-workspace lever is the same one +/// the engine adapter's tolerance test uses: the tree config's workspace sits +/// under a regular file, so the tree store cannot be created. +#[tokio::test] +async fn a_tolerated_tree_ingest_failure_is_counted() { + use tinymemory_api::host::test_support::TestHostConfig; + use tinymemory_api::host::MemoryHostConfig; + + crate::test_seams::init(); + let workspace = tempfile::tempdir().expect("workspace"); + let client: MemoryClientRef = Arc::new( + crate::store::MemoryClient::from_workspace_dir(workspace.path().join("skill-store")) + .expect("memory client initialises against a fresh workspace"), + ); + let blocker = workspace.path().join("blocker"); + std::fs::write(&blocker, b"not a directory").expect("write blocker file"); + let mut host_config = TestHostConfig::default(); + host_config.workspace_dir = blocker.join("workspace"); + let host = Arc::new(PipelineHost::new(client, host_config.to_arc())); + + assert_eq!(host.tree_ingest_failures(), 0); + host.store(SkillDocument { + namespace_skill_id: "gmail".into(), + connection_id: "conn-1".into(), + document_id: "gmail:msg-1".into(), + title: "Quarterly planning".into(), + content: "Let's finalise the Q3 roadmap.".into(), + toolkit: "gmail".into(), + metadata: serde_json::json!({ "source": "composio-provider-incremental" }), + }) + .await + .expect("a non-corrupt tree-ingest failure must stay tolerated"); + assert_eq!( + host.tree_ingest_failures(), + 1, + "the tolerated failure must be counted for the run's verdict" + ); +} diff --git a/crates/tinymemory-core/src/sync/pipelines/traits.rs b/crates/tinymemory-core/src/sync/pipelines/traits.rs index a3b76d3..18efc98 100644 --- a/crates/tinymemory-core/src/sync/pipelines/traits.rs +++ b/crates/tinymemory-core/src/sync/pipelines/traits.rs @@ -163,6 +163,13 @@ pub struct SyncOutcome { pub provider_cost_usd: f64, #[serde(default, skip_serializing_if = "Option::is_none")] pub note: Option, + /// Items whose skill-store write committed but whose memory-tree ingest + /// failed (non-corrupt failures — corruption aborts the run instead). + /// `records_ingested` counts those items as fetched-and-stored, so a + /// non-zero value here is the "fetch succeeded, tree did not" signal the + /// sync verdict must not report as full success (openhuman#5820). + #[serde(default)] + pub tree_ingest_failures: u32, } #[derive(Debug, thiserror::Error)] diff --git a/crates/tinymemory-core/src/sync/workspace/periodic_tests.rs b/crates/tinymemory-core/src/sync/workspace/periodic_tests.rs index 1d24ee4..f16be12 100644 --- a/crates/tinymemory-core/src/sync/workspace/periodic_tests.rs +++ b/crates/tinymemory-core/src/sync/workspace/periodic_tests.rs @@ -19,6 +19,8 @@ fn entry(source_id: &str, kind: &str, success: bool, ts: DateTime) -> SyncA duration_ms: 10, success, error: None, + tree_ingest_failures: 0, + tree_error: None, } } diff --git a/crates/tinymemory-tinycortex/src/engine/mod.rs b/crates/tinymemory-tinycortex/src/engine/mod.rs index 0966ec7..744bd45 100644 --- a/crates/tinymemory-tinycortex/src/engine/mod.rs +++ b/crates/tinymemory-tinycortex/src/engine/mod.rs @@ -2398,6 +2398,8 @@ fn audit_entry(entry: tinymemory_core::sync::audit::SyncAuditEntry) -> SyncAudit duration_ms, success, error, + tree_ingest_failures, + tree_error, } = entry; SyncAuditEntry { timestamp, @@ -2415,6 +2417,8 @@ fn audit_entry(entry: tinymemory_core::sync::audit::SyncAuditEntry) -> SyncAudit duration_ms, success, error, + tree_ingest_failures, + tree_error, } } diff --git a/crates/tinymemory-tinycortex/src/engine/test.rs b/crates/tinymemory-tinycortex/src/engine/test.rs index a2aa311..7d26441 100644 --- a/crates/tinymemory-tinycortex/src/engine/test.rs +++ b/crates/tinymemory-tinycortex/src/engine/test.rs @@ -404,6 +404,8 @@ fn an_audit_row_crosses_field_for_field_and_keeps_its_price() { duration_ms: 4_200, success: true, error: None, + tree_ingest_failures: 0, + tree_error: None, }; let crossed = audit_entry(entry); assert_eq!(crossed.source_id, "composio:gmail:conn-1");