From 3b58ad0f4d297c9316f6ec9c5fbcb8d7625ffbc4 Mon Sep 17 00:00:00 2001 From: Mega Mind <146339422+M3gA-Mind@users.noreply.github.com> Date: Wed, 20 May 2026 01:11:48 +0530 Subject: [PATCH] fix(memory): connection pooling + circuit breaker for memory_tree SQLite (#2206) (#2212) --- src/openhuman/doctor/core.rs | 65 ++++ src/openhuman/doctor/core_tests.rs | 60 ++++ src/openhuman/memory/tree/jobs/worker.rs | 114 ++++++- src/openhuman/memory/tree/store.rs | 397 +++++++++++++++++++---- src/openhuman/memory/tree/store_tests.rs | 115 +++++++ 5 files changed, 687 insertions(+), 64 deletions(-) diff --git a/src/openhuman/doctor/core.rs b/src/openhuman/doctor/core.rs index 29a1325be..9dc9727db 100644 --- a/src/openhuman/doctor/core.rs +++ b/src/openhuman/doctor/core.rs @@ -72,6 +72,7 @@ pub fn run(config: &Config) -> Result { check_workspace(config, &mut items); check_daemon_state(config, &mut items); check_environment(&mut items); + check_memory_tree_db(config, &mut items); let errors = items .iter() @@ -771,6 +772,70 @@ fn check_command_available( } } +// ── Memory-tree DB health ──────────────────────────────────────── + +/// Probe the memory-tree SQLite database and push a [`DiagnosticItem`]. +/// +/// - If the DB directory / file does not exist yet: `Warn` (not yet created). +/// - If a stale `.db-shm` file is present alongside the DB: `Warn`. +/// - If we can open the DB and run a basic probe query: `Ok`. +/// - If the probe fails: `Error`. +fn check_memory_tree_db(config: &Config, items: &mut Vec) { + let cat = "memory_tree_db"; + let db_path = config.workspace_dir.join("memory_tree").join("chunks.db"); + + // ── Stale side-files (checked even when chunks.db is absent) ──── + let base_name = db_path + .file_name() + .unwrap_or_default() + .to_string_lossy() + .into_owned(); + let shm = db_path.with_file_name(format!("{base_name}-shm")); + let wal = db_path.with_file_name(format!("{base_name}-wal")); + for sidecar in [&shm, &wal] { + if sidecar.exists() { + items.push(DiagnosticItem::warn( + cat, + format!( + "stale SQLite side-file present (may indicate unclean shutdown): {}", + sidecar.display() + ), + )); + } + } + + // ── File existence ────────────────────────────────────────────── + if !db_path.exists() { + items.push(DiagnosticItem::warn( + cat, + format!( + "DB not yet created (first ingest will initialise it): {}", + db_path.display() + ), + )); + return; + } + + // ── Probe connection ───────────────────────────────────────────── + match crate::openhuman::memory::tree::store::with_connection(config, |conn| { + let n: i64 = conn.query_row("SELECT COUNT(*) FROM mem_tree_chunks", [], |r| r.get(0))?; + Ok(n) + }) { + Ok(count) => { + items.push(DiagnosticItem::ok( + cat, + format!("DB accessible at {} ({count} chunks)", db_path.display()), + )); + } + Err(err) => { + items.push(DiagnosticItem::error( + cat, + format!("DB probe failed at {}: {err:#}", db_path.display()), + )); + } + } +} + // ── Helpers ────────────────────────────────────────────────────── fn parse_rfc3339(input: &str) -> Option> { diff --git a/src/openhuman/doctor/core_tests.rs b/src/openhuman/doctor/core_tests.rs index 5bde24a2a..fcf9d8b92 100644 --- a/src/openhuman/doctor/core_tests.rs +++ b/src/openhuman/doctor/core_tests.rs @@ -1,4 +1,11 @@ use super::*; +use tempfile::TempDir; + +fn test_config_in(tmp: &TempDir) -> Config { + let mut cfg = Config::default(); + cfg.workspace_dir = tmp.path().to_path_buf(); + cfg +} #[test] fn config_validation_warns_no_channels() { @@ -51,3 +58,56 @@ fn embedding_provider_validation_rejects_malformed_url() { let err = embedding_provider_validation_error("custom:not a url").expect("should fail"); assert!(err.contains("invalid custom provider URL"), "{err}"); } + +// ── check_memory_tree_db tests (#2206) ─────────────────────────────────────── + +/// When the workspace exists but the DB file has never been created, +/// `check_memory_tree_db` should push exactly one `Warn` item mentioning +/// "not yet created". +#[test] +fn check_memory_tree_db_warns_when_db_missing() { + let tmp = TempDir::new().expect("tempdir"); + let cfg = test_config_in(&tmp); + + let mut items = vec![]; + check_memory_tree_db(&cfg, &mut items); + + assert_eq!(items.len(), 1, "expected exactly one diagnostic item"); + assert_eq!(items[0].severity, Severity::Warn); + assert!( + items[0].message.contains("not yet created"), + "unexpected message: {}", + items[0].message + ); +} + +/// After `with_connection` has successfully initialised the DB, the probe +/// should push an `Ok` item. +#[test] +fn check_memory_tree_db_ok_when_accessible() { + let tmp = TempDir::new().expect("tempdir"); + let cfg = test_config_in(&tmp); + + // Trigger DB creation. + crate::openhuman::memory::tree::store::with_connection(&cfg, |_conn| Ok(())) + .expect("DB init must succeed"); + + let mut items = vec![]; + check_memory_tree_db(&cfg, &mut items); + + // There may be a Warn about the SHM file on some platforms; there must + // be at least one Ok item about the DB being accessible. + let ok_items: Vec<_> = items + .iter() + .filter(|i| i.severity == Severity::Ok && i.category == "memory_tree_db") + .collect(); + assert!( + !ok_items.is_empty(), + "expected at least one Ok memory_tree_db item; got: {items:?}" + ); + assert!( + ok_items[0].message.contains("accessible"), + "unexpected ok message: {}", + ok_items[0].message + ); +} diff --git a/src/openhuman/memory/tree/jobs/worker.rs b/src/openhuman/memory/tree/jobs/worker.rs index 14e814059..9c9117849 100644 --- a/src/openhuman/memory/tree/jobs/worker.rs +++ b/src/openhuman/memory/tree/jobs/worker.rs @@ -96,8 +96,21 @@ pub fn start(config: Config) { if is_sqlite_busy(&err) { log::warn!( "[memory_tree::jobs] worker {idx} hit SQLite busy/locked, \ - backing off: {err:#}" + backing off 1s: {err:#}" ); + tokio::time::sleep(Duration::from_secs(1)).await; + } else if is_sqlite_io_transient(&err) { + // I/O errors (IOERR_TRUNCATE 1546, IOERR_SHMMAP 4874, + // CANTOPEN 14) or circuit breaker open — transient + // filesystem / WAL condition. Back off 30 s and let the + // connection cache try a fresh open on next poll. These + // are NOT reported to Sentry (they are transient and were + // flooding ~19K events/4 days, see #2206). + log::warn!( + "[memory_tree::jobs] worker {idx} hit transient I/O error, \ + backing off 30s: {err:#}" + ); + tokio::time::sleep(Duration::from_secs(30)).await; } else { crate::core::observability::report_error( &err, @@ -105,8 +118,8 @@ pub fn start(config: Config) { "tree_jobs_worker", &[("worker_idx", &idx.to_string())], ); + tokio::time::sleep(Duration::from_secs(1)).await; } - tokio::time::sleep(Duration::from_secs(1)).await; } } } @@ -226,6 +239,37 @@ pub async fn run_once(config: &Config) -> Result { Ok(true) } +/// Classify whether an error is a transient I/O failure that should be +/// silently backed off without a Sentry report (#2206). +/// +/// Covers: +/// - `SQLITE_IOERR_TRUNCATE` (extended code 1546): WAL truncation failed — +/// usually a transient filesystem hiccup. +/// - `SQLITE_IOERR_SHMMAP` (extended code 4874): shared-memory mapping +/// failed — WAL side-file temporarily unavailable. +/// - `SQLITE_CANTOPEN` / `CannotOpen` (extended code 14): DB file temporarily +/// inaccessible. +/// - Text fallback: circuit breaker message, or rusqlite phrases that don't +/// downcast cleanly after multiple `.context()` layers. +fn is_sqlite_io_transient(err: &anyhow::Error) -> bool { + if let Some(rusqlite::Error::SqliteFailure(f, _)) = err.downcast_ref::() { + if matches!(f.extended_code, 1546 | 4874 | 14) { + return true; + } + if f.code == rusqlite::ErrorCode::CannotOpen { + return true; + } + } + // Text fallback for errors wrapped under `.context()` layers or + // emitted as plain `anyhow!` strings (e.g. circuit breaker message). + let msg = format!("{err:#}").to_ascii_lowercase(); + msg.contains("circuit breaker open") + || msg.contains("disk i/o error") + || msg.contains("unable to open database file") + || msg.contains("xshmmap") + || msg.contains("truncate file") +} + /// Classify whether an error from `run_once` is a transient SQLite /// write-lock contention (`SQLITE_BUSY` or `SQLITE_LOCKED`). /// @@ -335,4 +379,70 @@ mod tests { let err = anyhow::anyhow!("upstream returned 500: internal server error"); assert!(!is_sqlite_busy(&err)); } + + // ── is_sqlite_io_transient tests (#2206) ───────────────────────────── + + /// SQLITE_IOERR_TRUNCATE (extended code 1546) must be classified as + /// transient so the worker backs off without hitting Sentry. + #[test] + fn is_sqlite_io_transient_matches_ioerr_truncate() { + let raw = rusqlite::Error::SqliteFailure( + rusqlite::ffi::Error { + code: rusqlite::ErrorCode::SystemIoFailure, + extended_code: 1546, // SQLITE_IOERR_TRUNCATE + }, + Some("disk I/O error".into()), + ); + assert!(is_sqlite_io_transient(&anyhow::Error::from(raw))); + } + + /// SQLITE_IOERR_SHMMAP (extended code 4874) must be classified as + /// transient — WAL shared-memory mapping is a filesystem hiccup. + #[test] + fn is_sqlite_io_transient_matches_ioerr_shmmap() { + let raw = rusqlite::Error::SqliteFailure( + rusqlite::ffi::Error { + code: rusqlite::ErrorCode::SystemIoFailure, + extended_code: 4874, // SQLITE_IOERR_SHMMAP + }, + Some("xshmmap failed".into()), + ); + assert!(is_sqlite_io_transient(&anyhow::Error::from(raw))); + } + + /// SQLITE_CANTOPEN (code CannotOpen, extended code 14) must be + /// classified as transient — temporary inability to open the file. + #[test] + fn is_sqlite_io_transient_matches_cantopen() { + let raw = rusqlite::Error::SqliteFailure( + rusqlite::ffi::Error { + code: rusqlite::ErrorCode::CannotOpen, + extended_code: 14, // SQLITE_CANTOPEN + }, + Some("unable to open database file".into()), + ); + assert!(is_sqlite_io_transient(&anyhow::Error::from(raw))); + } + + /// The circuit breaker error message produced by `get_or_init_connection` + /// must be classified as transient via the text fallback. + #[test] + fn is_sqlite_io_transient_text_fallback() { + let err = anyhow::anyhow!("memory_tree_db circuit breaker open: too many init failures"); + assert!(is_sqlite_io_transient(&err)); + } + + /// UNIQUE constraint violation must NOT be reclassified as a transient + /// I/O error — those are genuine bugs. + #[test] + fn is_sqlite_io_transient_negative_constraint_violation() { + let raw = 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_io_transient(&anyhow::Error::from(raw))); + } } diff --git a/src/openhuman/memory/tree/store.rs b/src/openhuman/memory/tree/store.rs index 4b03e3a1a..0912bdd6e 100644 --- a/src/openhuman/memory/tree/store.rs +++ b/src/openhuman/memory/tree/store.rs @@ -6,11 +6,30 @@ //! //! Upsert semantics: writes are idempotent on `chunk.id` so re-ingesting the //! same raw source yields no duplicates. +//! +//! ## Connection cache (#2206) +//! +//! `with_connection()` previously opened a new SQLite connection and re-ran +//! the full schema init (8 tables, 15+ indexes, 8+ migrations) on **every** +//! call. With 4 workers polling every 5 s this amounted to ~69K connection +//! opens/day, and three I/O error codes (1546 IOERR_TRUNCATE, 4874 +//! IOERR_SHMMAP, 14 CANTOPEN) flooded Sentry with ~19K events in 4 days. +//! +//! Fix: a process-level `ConnectionCache` keyed by DB path. Each entry holds +//! one `parking_lot::Mutex` that is initialised once (schema + +//! migrations + legacy-embedding migration) and then reused for all subsequent +//! calls. A per-entry `CircuitBreaker` stops retrying after 3 consecutive +//! init failures for 30 s so a broken install does not busy-loop. use anyhow::{Context, Result}; use chrono::{DateTime, TimeZone, Utc}; +use parking_lot::Mutex as PMutex; use rusqlite::{params, Connection, OptionalExtension, Transaction}; -use std::time::Duration; +use std::collections::HashMap; +use std::path::PathBuf; +use std::sync::atomic::{AtomicBool, AtomicU32, Ordering}; +use std::sync::{Arc, OnceLock}; +use std::time::{Duration, Instant}; use crate::openhuman::config::Config; use crate::openhuman::memory::tree::content_store::StagedChunk; @@ -694,20 +713,136 @@ fn ms_to_utc(ms: i64) -> rusqlite::Result> { }) } -/// Open the memory_tree SQLite DB and run a closure against it. +// ── Connection cache (#2206) ───────────────────────────────────────────────── + +/// How many consecutive init failures before the circuit breaker trips. +const CB_THRESHOLD: u32 = 3; +/// How long the circuit breaker holds the DB closed after tripping. +const CB_COOLDOWN: Duration = Duration::from_secs(30); + +/// Per-path circuit breaker: after [`CB_THRESHOLD`] consecutive init failures +/// the breaker trips and `get_or_init_connection` returns an error immediately +/// until [`CB_COOLDOWN`] elapses. On the first success it resets to zero. +struct CircuitBreaker { + consecutive_failures: AtomicU32, + tripped: AtomicBool, + last_trip: PMutex>, +} + +impl CircuitBreaker { + fn new() -> Self { + Self { + consecutive_failures: AtomicU32::new(0), + tripped: AtomicBool::new(false), + last_trip: PMutex::new(None), + } + } + + fn record_success(&self) { + self.consecutive_failures.store(0, Ordering::Relaxed); + self.tripped.store(false, Ordering::Relaxed); + *self.last_trip.lock() = None; + } + + /// Records one more failure. Returns `true` if this call just tripped the + /// breaker (i.e. the threshold was crossed right now). + fn record_failure(&self) -> bool { + let prev = self.consecutive_failures.fetch_add(1, Ordering::Relaxed); + let count = prev + 1; + if count >= CB_THRESHOLD && !self.tripped.swap(true, Ordering::Relaxed) { + *self.last_trip.lock() = Some(Instant::now()); + return true; + } + // Re-arm the cooldown on each subsequent failure while already tripped + // so that a failed post-cooldown retry restarts the 30 s window instead + // of leaving the stale timestamp in place (which would let `is_open` + // return false immediately and allow unlimited retries). + if self.tripped.load(Ordering::Relaxed) { + *self.last_trip.lock() = Some(Instant::now()); + } + false + } + + /// Returns `true` when the breaker is open AND the cooldown has not yet + /// elapsed. Returns `false` (allowing a retry) once the cooldown passes. + fn is_open(&self) -> bool { + if !self.tripped.load(Ordering::Relaxed) { + return false; + } + let guard = self.last_trip.lock(); + match *guard { + Some(t) if t.elapsed() < CB_COOLDOWN => true, + _ => false, + } + } +} + +/// Process-level cache — two separate maps so that a failing init cannot +/// accidentally serve a dummy connection on the fast path. /// -/// Visible to sibling modules (e.g. `score::store`) so Phase 2 can reuse -/// the same connection setup / schema initialisation without duplication. -pub(crate) fn with_connection( - config: &Config, - f: impl FnOnce(&Connection) -> Result, -) -> Result { - let dir = config.workspace_dir.join(DB_DIR); - std::fs::create_dir_all(&dir) - .with_context(|| format!("Failed to create memory_tree dir: {}", dir.display()))?; - let db_path = dir.join(DB_FILE); - let conn = Connection::open(&db_path) - .with_context(|| format!("Failed to open memory_tree DB: {}", db_path.display()))?; +/// `connections`: only fully-initialised (schema + migrations run) entries. +/// `breakers`: persists across failed init attempts so the circuit breaker +/// survives even when `connections` has no entry for that path. +struct ConnectionCache { + connections: PMutex>>>, + breakers: PMutex>>, +} + +static CONN_CACHE: OnceLock = OnceLock::new(); + +fn conn_cache() -> &'static ConnectionCache { + CONN_CACHE.get_or_init(|| ConnectionCache { + connections: PMutex::new(HashMap::new()), + breakers: PMutex::new(HashMap::new()), + }) +} + +/// Compute the canonical DB path from `config`. +fn db_path_for(config: &Config) -> PathBuf { + config.workspace_dir.join(DB_DIR).join(DB_FILE) +} + +/// Delete stale WAL/SHM side-files (`-shm`, `-wal`) that can block a +/// clean DB open after a crash. Logs what was deleted and returns `true` if +/// anything was removed. +/// +/// SQLite names these files `-shm` and `-wal`. +/// For `chunks.db` that is `chunks.db-shm` / `chunks.db-wal`. +pub(crate) fn try_cleanup_stale_files(db_path: &std::path::Path) -> bool { + let mut cleaned = false; + for suffix in &["-shm", "-wal"] { + // Build the side-file path: append suffix to the full db filename. + let side = { + let mut p = db_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 + }; + if side.exists() { + match std::fs::remove_file(&side) { + Ok(()) => { + log::warn!("[memory_tree] removed stale side-file: {}", side.display()); + cleaned = true; + } + Err(e) => { + log::warn!( + "[memory_tree] failed to remove stale side-file {}: {e}", + side.display() + ); + } + } + } + } + cleaned +} + +/// Run the full one-time DB initialisation (WAL, schema, migrations) against +/// an already-open `Connection`. Used by `get_or_init_connection`. +fn init_db(conn: &Connection, config: &Config) -> Result<()> { conn.busy_timeout(SQLITE_BUSY_TIMEOUT) .context("Failed to configure memory_tree busy timeout")?; if let Err(wal_err) = conn.execute_batch("PRAGMA journal_mode=WAL;") { @@ -718,28 +853,20 @@ pub(crate) fn with_connection( conn.execute_batch(SCHEMA) .context("Failed to initialize memory_tree schema")?; // Phase 2 migrations — additive, idempotent. - add_column_if_missing(&conn, "mem_tree_chunks", "embedding", "BLOB")?; + add_column_if_missing(conn, "mem_tree_chunks", "embedding", "BLOB")?; // Phase 2 LLM-NER follow-up: per-chunk LLM importance signal + // human-readable reason. Both nullable; absence is treated as // "no LLM signal available" by readers. - add_column_if_missing(&conn, "mem_tree_score", "llm_importance", "REAL")?; - add_column_if_missing(&conn, "mem_tree_score", "llm_importance_reason", "TEXT")?; - // Phase 3a (#709): parent-summary backlink on leaves. Populated when - // the L0 buffer seals into an L1 summary so traversal can walk - // leaf → parent without scanning `mem_tree_summaries.child_ids_json`. - add_column_if_missing(&conn, "mem_tree_chunks", "parent_summary_id", "TEXT")?; + add_column_if_missing(conn, "mem_tree_score", "llm_importance", "REAL")?; + add_column_if_missing(conn, "mem_tree_score", "llm_importance_reason", "TEXT")?; + // Phase 3a (#709): parent-summary backlink on leaves. + add_column_if_missing(conn, "mem_tree_chunks", "parent_summary_id", "TEXT")?; // Phase 4 (#710): sealed-summary embeddings for semantic rerank. - // Blob layout matches `mem_tree_chunks.embedding` — see - // `score::embed::{pack_embedding, unpack_embedding}`. Nullable so - // legacy summaries from Phases 1-3 read back as None; retrieval - // tolerates NULL by dropping the row to the bottom of a rerank. - add_column_if_missing(&conn, "mem_tree_summaries", "embedding", "BLOB")?; + add_column_if_missing(conn, "mem_tree_summaries", "embedding", "BLOB")?; // Async-pipeline lifecycle flag. Default 'admitted' so chunks ingested - // before the queue migration stay queryable. New writes start at - // 'pending_extraction'; the extract handler advances them to 'admitted' - // (then 'buffered' / 'sealed') or 'dropped'. + // before the queue migration stay queryable. add_column_if_missing( - &conn, + conn, "mem_tree_chunks", "lifecycle_status", "TEXT NOT NULL DEFAULT 'admitted'", @@ -749,43 +876,189 @@ pub(crate) fn with_connection( ON mem_tree_chunks(lifecycle_status);", ) .context("Failed to create mem_tree_chunks lifecycle index")?; - // Phase MD-content (#TBD): pointer + integrity hash. Body lives at - // / as a .md file. Both nullable so chunks - // ingested before this migration read back with NULL (body still in - // `content`). New writes populate both columns. The `content` column - // stores a 500-char plain-text preview instead of the full body. - add_column_if_missing(&conn, "mem_tree_chunks", "content_path", "TEXT")?; - add_column_if_missing(&conn, "mem_tree_chunks", "content_sha256", "TEXT")?; - // Phase MD-content (summaries): same pointer pattern for summary nodes. - // `content_path` is the relative path to the .md file under - // `/summaries/...`. `content_sha256` is the SHA-256 hex - // of the body bytes only (front-matter excluded). Both nullable so - // legacy rows (from before this migration) read back with NULL — callers - // fall back to the `content` column for those rows. - add_column_if_missing(&conn, "mem_tree_summaries", "content_path", "TEXT")?; - add_column_if_missing(&conn, "mem_tree_summaries", "content_sha256", "TEXT")?; - // Raw-archive pointer column. JSON array of {path, start, end} — - // used by chunks whose body comes from one or more files under - // `/raw/...` (today: email). When set, `read_chunk_body` - // reads + concatenates those byte ranges instead of fetching from - // disk via `content_path` or falling back to the SQL `content` - // preview. Nullable so legacy chunks keep working unchanged. - add_column_if_missing(&conn, "mem_tree_chunks", "raw_refs_json", "TEXT")?; - // #1365: is_user flag on indexed entity rows. Set at write time by - // running the canonical id through the Composio identity registry - // (`is_self_identity_any_toolkit`). Default 0 so legacy rows read - // back as "not user" until the backfill job re-tags them. + // Phase MD-content (#TBD): pointer + integrity hash. + add_column_if_missing(conn, "mem_tree_chunks", "content_path", "TEXT")?; + add_column_if_missing(conn, "mem_tree_chunks", "content_sha256", "TEXT")?; + // Phase MD-content (summaries). + add_column_if_missing(conn, "mem_tree_summaries", "content_path", "TEXT")?; + add_column_if_missing(conn, "mem_tree_summaries", "content_sha256", "TEXT")?; + // Raw-archive pointer column. + add_column_if_missing(conn, "mem_tree_chunks", "raw_refs_json", "TEXT")?; + // #1365: is_user flag on indexed entity rows. add_column_if_missing( - &conn, + conn, "mem_tree_entity_index", "is_user", "INTEGER NOT NULL DEFAULT 0", )?; - // #1574 §7: one-shot, version-gated copy of legacy embedding columns - // into the per-model sidecar at the active signature. Runs once - // (PRAGMA user_version gate); cheap no-op on every subsequent open. - migrate_legacy_embeddings_to_sidecar(&conn, config)?; - f(&conn) + // #1574 §7: one-shot, version-gated legacy→sidecar embedding migration. + migrate_legacy_embeddings_to_sidecar(conn, config)?; + Ok(()) +} + +/// Whether `err` looks like one of the I/O error codes that warrant a +/// stale-file cleanup + single retry before giving up. +fn is_io_open_error(err: &anyhow::Error) -> bool { + if let Some(rusqlite::Error::SqliteFailure(f, _)) = err.downcast_ref::() { + // 1546 = SQLITE_IOERR_TRUNCATE, 4874 = SQLITE_IOERR_SHMMAP, 14 = SQLITE_CANTOPEN + return matches!(f.extended_code, 1546 | 4874 | 14) + || f.code == rusqlite::ErrorCode::CannotOpen; + } + let msg = format!("{err:#}").to_ascii_lowercase(); + msg.contains("disk i/o error") + || msg.contains("unable to open database file") + || msg.contains("xshmmap") + || msg.contains("truncate file") +} + +/// Obtain (or lazily create) a cached connection for the workspace described +/// by `config`. Returns `Err` immediately when the circuit breaker is open. +fn get_or_init_connection(config: &Config) -> Result>> { + let db_path = db_path_for(config); + + // ── Fast path: reject immediately if the breaker is open ───────────── + { + let breakers = conn_cache().breakers.lock(); + if let Some(breaker) = breakers.get(&db_path) { + if breaker.is_open() { + anyhow::bail!( + "[memory_tree] circuit breaker open for {}: too many consecutive init failures", + db_path.display() + ); + } + } + } + // ── Fast path: return cached connection if already initialised ──────── + { + let guard = conn_cache().connections.lock(); + if let Some(conn) = guard.get(&db_path) { + return Ok(Arc::clone(conn)); + } + } + + // ── Slow path: initialise DB, then insert into cache ───────────────── + log::debug!( + "[memory_tree] opening and initialising DB at {}", + db_path.display() + ); + + // Attempt to open + init the connection (dir creation is inside + // `open_and_init` so every failure — including EEXIST on the dir — + // reaches the circuit-breaker recording logic below). On certain I/O + // errors (#2206) we clean up stale WAL/SHM side-files and retry once. + let conn = open_and_init(&db_path, config).or_else(|first_err| { + if is_io_open_error(&first_err) { + log::warn!( + "[memory_tree] I/O error on first open attempt ({}), cleaning stale files and retrying", + first_err + ); + try_cleanup_stale_files(&db_path); + open_and_init(&db_path, config) + } else { + Err(first_err) + } + }); + + match conn { + Ok(conn) => { + let arc_conn = Arc::new(PMutex::new(conn)); + conn_cache() + .connections + .lock() + .insert(db_path.clone(), Arc::clone(&arc_conn)); + // Reset any prior failure counter now that init succeeded. + if let Some(breaker) = conn_cache().breakers.lock().get(&db_path) { + breaker.record_success(); + } + log::debug!("[memory_tree] DB connection cached and ready"); + Ok(arc_conn) + } + Err(err) => { + // Persist the breaker so the failure count accumulates across + // calls even though no connection entry exists yet. + let breaker = { + let mut guard = conn_cache().breakers.lock(); + guard + .entry(db_path.clone()) + .or_insert_with(|| Arc::new(CircuitBreaker::new())) + .clone() + }; + let just_tripped = breaker.record_failure(); + if just_tripped { + log::error!( + "[memory_tree] circuit breaker tripped for {}: {} consecutive init failures", + db_path.display(), + CB_THRESHOLD + ); + let _ = crate::core::event_bus::publish_global( + crate::core::event_bus::DomainEvent::HealthChanged { + component: "memory_tree_db".to_string(), + healthy: false, + message: Some(format!( + "Schema init failed {CB_THRESHOLD} consecutive times" + )), + }, + ); + } + Err(err) + } + } +} + +/// Ensure the DB directory exists, open the SQLite file, and run the full +/// schema init sequence. All errors (dir creation, file open, schema init) +/// are returned as `Err` so callers can funnel them through the circuit +/// breaker logic in a single place. +fn open_and_init(db_path: &std::path::Path, config: &Config) -> Result { + let dir = db_path.parent().expect("db_path always has a parent"); + std::fs::create_dir_all(dir) + .with_context(|| format!("Failed to create memory_tree dir: {}", dir.display()))?; + let conn = Connection::open(db_path) + .with_context(|| format!("Failed to open memory_tree DB: {}", db_path.display()))?; + init_db(&conn, config) + .with_context(|| format!("Failed to init memory_tree schema: {}", db_path.display()))?; + Ok(conn) +} + +/// Remove the cached connection for `config`'s workspace (forces a fresh open +/// on the next `with_connection` call). Also clears the breaker so the next +/// open attempt is not immediately rejected. Does nothing if no entry exists. +#[allow(dead_code)] +pub(crate) fn invalidate_connection(config: &Config) { + let db_path = db_path_for(config); + conn_cache().connections.lock().remove(&db_path); + conn_cache().breakers.lock().remove(&db_path); + log::debug!( + "[memory_tree] connection invalidated for {}", + db_path.display() + ); +} + +/// Clear the entire connection cache. For test isolation only. +#[cfg(test)] +pub(crate) fn clear_connection_cache() { + conn_cache().connections.lock().clear(); + conn_cache().breakers.lock().clear(); +} + +/// Open the memory_tree SQLite DB and run a closure against it. +/// +/// Visible to sibling modules (e.g. `score::store`) so Phase 2 can reuse +/// the same connection setup / schema initialisation without duplication. +/// +/// # Connection caching (#2206) +/// +/// The underlying connection is initialised once per workspace path and then +/// reused from a process-level cache. Schema migrations run exactly once on +/// the first call for a given `config.workspace_dir`. Subsequent calls pay +/// only the cost of a `parking_lot::Mutex` lock and the closure itself. +pub(crate) fn with_connection( + config: &Config, + f: impl FnOnce(&Connection) -> Result, +) -> Result { + let conn_arc = get_or_init_connection(config)?; + let guard = conn_arc.lock(); + f(&guard) } /// One-shot migration (#1574 §7, vN): copy legacy `mem_tree_chunks.embedding` diff --git a/src/openhuman/memory/tree/store_tests.rs b/src/openhuman/memory/tree/store_tests.rs index ff77d4364..287c66fef 100644 --- a/src/openhuman/memory/tree/store_tests.rs +++ b/src/openhuman/memory/tree/store_tests.rs @@ -1,5 +1,14 @@ //! Unit tests for [`super`] — chunk upsert / list / lifecycle / embedding / //! content-pointer accessors against a tempdir-backed SQLite store. +//! +//! ## Test isolation for the connection cache +//! +//! Because the connection cache is a process-level singleton, tests that want +//! to exercise cache behaviour (same Arc, independent workspaces, circuit +//! breaker, cleanup) must call `clear_connection_cache()` at the start — or +//! be careful to use unique tempdirs that cannot collide with other tests. +//! The call is cheap (a mutex lock + HashMap clear) and harmless for tests +//! that don't need it. use super::*; use crate::openhuman::memory::tree::types::chunk_id; @@ -296,6 +305,10 @@ fn legacy_embeddings_migrate_to_sidecar_once() { }) .unwrap(); + // Evict the cached connection so the next open sees user_version = 0 + // and re-runs migrate_legacy_embeddings_to_sidecar. + invalidate_connection(&cfg); + // The next store open (this getter's `with_connection`) sees // user_version = 0 and runs the migration before returning. assert_eq!( @@ -340,3 +353,105 @@ fn legacy_embeddings_migrate_to_sidecar_once() { Some(match_vec) ); } + +// ── Connection cache tests (#2206) ─────────────────────────────────────────── + +/// Two `with_connection` calls for the same workspace must return the same +/// cached `Arc` (pointer identity proves no re-init happened). +#[test] +fn connection_cache_returns_same_arc_for_same_workspace() { + clear_connection_cache(); + let (_tmp, cfg) = test_config(); + + let arc1 = get_or_init_connection(&cfg).expect("first get_or_init"); + let arc2 = get_or_init_connection(&cfg).expect("second get_or_init"); + assert!( + Arc::ptr_eq(&arc1, &arc2), + "expected the same Arc from the connection cache on the second call" + ); +} + +/// Two configs pointing at different tempdirs must produce independent +/// connections (separate Arc pointers, no cross-contamination). +#[test] +fn connection_cache_uses_separate_connections_for_different_workspaces() { + clear_connection_cache(); + let (_tmp1, cfg1) = test_config(); + let (_tmp2, cfg2) = test_config(); + + let arc1 = get_or_init_connection(&cfg1).expect("workspace 1"); + let arc2 = get_or_init_connection(&cfg2).expect("workspace 2"); + assert!( + !Arc::ptr_eq(&arc1, &arc2), + "different workspaces must have independent connections" + ); + + // Sanity: each DB is usable independently. + let c = sample_chunk("s", 0, 1_700_000_000_000); + upsert_chunks(&cfg1, &[c.clone()]).unwrap(); + assert_eq!(count_chunks(&cfg1).unwrap(), 1); + assert_eq!(count_chunks(&cfg2).unwrap(), 0); +} + +/// Pointing the DB path at a *file* (not a directory) makes it impossible to +/// create the DB, so `get_or_init_connection` must fail. After +/// `CB_THRESHOLD` failures the circuit breaker trips and subsequent calls +/// return an error immediately without touching the filesystem. +#[test] +fn circuit_breaker_trips_after_threshold() { + clear_connection_cache(); + let tmp = TempDir::new().expect("tempdir"); + + // Create a regular file where the memory_tree *directory* would be — + // this prevents `create_dir_all` from succeeding. + let blocker = tmp.path().join(DB_DIR); + std::fs::write(&blocker, b"not a dir").expect("write blocker file"); + + let mut cfg = Config::default(); + cfg.workspace_dir = tmp.path().to_path_buf(); + + // First CB_THRESHOLD calls should all fail (can't create dir over a file). + for i in 0..CB_THRESHOLD { + let result = get_or_init_connection(&cfg); + assert!( + result.is_err(), + "call {i}: expected error before breaker trips" + ); + } + + // The CB_THRESHOLD+1'th call should be rejected immediately by the + // circuit breaker (error message contains "circuit breaker"). + let cb_err = get_or_init_connection(&cfg) + .expect_err("expected circuit breaker error on call after threshold"); + let msg = format!("{cb_err:#}").to_ascii_lowercase(); + assert!( + msg.contains("circuit breaker"), + "expected circuit breaker message, got: {msg}" + ); +} + +/// `try_cleanup_stale_files` removes `.db-shm` and `.db-wal` side-files that +/// exist alongside the main DB file. +#[test] +fn stale_shm_cleanup_removes_files() { + let tmp = TempDir::new().expect("tempdir"); + let db_path = tmp.path().join("chunks.db"); + + // Create the main DB file and the two stale side-files. + std::fs::write(&db_path, b"").expect("create db file"); + let shm = tmp.path().join("chunks.db-shm"); + let wal = tmp.path().join("chunks.db-wal"); + std::fs::write(&shm, b"stale shm").expect("create shm"); + std::fs::write(&wal, b"stale wal").expect("create wal"); + + assert!(shm.exists(), "shm must exist before cleanup"); + assert!(wal.exists(), "wal must exist before cleanup"); + + let cleaned = try_cleanup_stale_files(&db_path); + assert!( + cleaned, + "cleanup should return true when files were removed" + ); + assert!(!shm.exists(), "shm must be removed"); + assert!(!wal.exists(), "wal must be removed"); +}