diff --git a/src/openhuman/subconscious/schemas.rs b/src/openhuman/subconscious/schemas.rs index fcd95f878..6dc1f3e3b 100644 --- a/src/openhuman/subconscious/schemas.rs +++ b/src/openhuman/subconscious/schemas.rs @@ -289,7 +289,7 @@ fn handle_status(_params: Map) -> ControllerFuture { .unwrap_or((None, 0)); Ok((tc, pe, lt, tt)) }) - .map_err(|e| e.to_string())?; + .map_err(|e| format!("{e:#}"))?; let provider_unavailable_reason = if hb.enabled && hb.inference_enabled { super::engine::subconscious_provider_unavailable_reason(&config) @@ -355,7 +355,7 @@ fn handle_tasks_list(params: Map) -> ControllerFuture { let tasks = store::with_connection(&config.workspace_dir, |conn| { store::list_tasks(conn, enabled_only) }) - .map_err(|e| e.to_string())?; + .map_err(|e| format!("{e:#}"))?; to_json(RpcOutcome::single_log(tasks, "tasks listed")) }) } @@ -377,7 +377,7 @@ fn handle_tasks_add(params: Map) -> ControllerFuture { let task = engine .add_task(&title, source) .await - .map_err(|e| e.to_string())?; + .map_err(|e| format!("{e:#}"))?; to_json(RpcOutcome::single_log(task, "task added")) }) } @@ -409,7 +409,7 @@ fn handle_tasks_update(params: Map) -> ControllerFuture { store::with_connection(&config.workspace_dir, |conn| { store::update_task(conn, &task_id, &patch) }) - .map_err(|e| e.to_string())?; + .map_err(|e| format!("{e:#}"))?; to_json(RpcOutcome::single_log( serde_json::json!({"updated": task_id}), "task updated", @@ -428,7 +428,7 @@ fn handle_tasks_remove(params: Map) -> ControllerFuture { store::with_connection(&config.workspace_dir, |conn| { store::remove_task(conn, &task_id) }) - .map_err(|e| e.to_string())?; + .map_err(|e| format!("{e:#}"))?; to_json(RpcOutcome::single_log( serde_json::json!({"removed": task_id}), "task removed", @@ -444,7 +444,7 @@ fn handle_log_list(params: Map) -> ControllerFuture { let entries = store::with_connection(&config.workspace_dir, |conn| { store::list_log_entries(conn, task_id, limit) }) - .map_err(|e| e.to_string())?; + .map_err(|e| format!("{e:#}"))?; to_json(RpcOutcome::single_log(entries, "log entries listed")) }) } @@ -463,7 +463,7 @@ fn handle_escalations_list(params: Map) -> ControllerFuture { let escalations = store::with_connection(&config.workspace_dir, |conn| { store::list_escalations(conn, status_filter.as_ref()) }) - .map_err(|e| e.to_string())?; + .map_err(|e| format!("{e:#}"))?; to_json(RpcOutcome::single_log(escalations, "escalations listed")) }) } @@ -481,7 +481,7 @@ fn handle_escalations_approve(params: Map) -> ControllerFuture { engine .approve_escalation(&escalation_id) .await - .map_err(|e| e.to_string())?; + .map_err(|e| format!("{e:#}"))?; to_json(RpcOutcome::single_log( serde_json::json!({"approved": escalation_id}), "escalation approved and executed", @@ -502,7 +502,7 @@ fn handle_escalations_dismiss(params: Map) -> ControllerFuture { engine .dismiss_escalation(&escalation_id) .await - .map_err(|e| e.to_string())?; + .map_err(|e| format!("{e:#}"))?; to_json(RpcOutcome::single_log( serde_json::json!({"dismissed": escalation_id}), "escalation dismissed", @@ -520,7 +520,7 @@ fn handle_reflections_list(params: Map) -> ControllerFuture { let reflections = store::with_connection(&config.workspace_dir, |conn| { reflection_store::list_recent(conn, limit, since_ts) }) - .map_err(|e| e.to_string())?; + .map_err(|e| format!("{e:#}"))?; to_json(RpcOutcome::single_log(reflections, "reflections listed")) }) } @@ -537,7 +537,7 @@ fn handle_reflections_act(params: Map) -> ControllerFuture { let reflection = store::with_connection(&config.workspace_dir, |conn| { reflection_store::get_reflection(conn, &reflection_id) }) - .map_err(|e| e.to_string())? + .map_err(|e| format!("{e:#}"))? .ok_or_else(|| format!("reflection not found: {reflection_id}"))?; // Spawn a fresh conversation thread for this action. Reflections never @@ -660,7 +660,7 @@ fn handle_reflections_dismiss(params: Map) -> ControllerFuture { store::with_connection(&config.workspace_dir, |conn| { reflection_store::mark_dismissed(conn, &reflection_id, now) }) - .map_err(|e| e.to_string())?; + .map_err(|e| format!("{e:#}"))?; to_json(RpcOutcome::single_log( serde_json::json!({"dismissed": reflection_id}), "reflection dismissed", diff --git a/src/openhuman/subconscious/schemas_tests.rs b/src/openhuman/subconscious/schemas_tests.rs index e645dffc3..200fa4310 100644 --- a/src/openhuman/subconscious/schemas_tests.rs +++ b/src/openhuman/subconscious/schemas_tests.rs @@ -166,3 +166,90 @@ fn field_opt_helper_is_not_required() { let f = field_opt("name", TypeSchema::String, "desc"); assert!(!f.required); } + +// ── Error chain preservation ─────────────────────────────────── +// +// The RPC handlers in this module bridge `anyhow::Result` (from +// `store::with_connection` and the wrapped rusqlite errors) into the +// JSON-RPC `Result` boundary via `map_err(|e| ...)`. +// +// **Critical for observability**: plain `e.to_string()` on an +// `anyhow::Error` returns ONLY the outermost context. For a +// `with_connection` failure the outer wrap is +// `"failed to run subconscious schema DDL"` — the underlying rusqlite +// root (the actual SQLite error code + message) is dropped. That +// stringified message is what `jsonrpc::invoke_method_inner` later +// passes to `report_error_or_expected`, which in turn captures it in +// Sentry. Without the chain, Sentry events for TAURI-RUST-A only +// surface the generic wrapper text and the rusqlite root cause is +// permanently invisible. +// +// All `map_err` sites in `schemas.rs` use `format!("{e:#}")` (anyhow's +// alternate Display walks the cause chain inline joined by `": "`) so +// the rusqlite root reaches Sentry. These guard tests pin the format +// so future contributors don't silently regress to `e.to_string()`. +#[test] +fn anyhow_alternate_display_walks_chain() { + use anyhow::Context; + + let inner = anyhow::anyhow!("database is locked").context("execute_batch failed"); + let outer: anyhow::Result<()> = Err(inner).context("failed to run subconscious schema DDL"); + + let err = outer.unwrap_err(); + + // Plain to_string() — the broken (pre-fix) shape. Only outer + // wrapper reaches the caller, root cause lost. + let lossy = err.to_string(); + assert_eq!(lossy, "failed to run subconscious schema DDL"); + assert!( + !lossy.contains("database is locked"), + "plain Display must drop the root cause — if this changes the chain-formatter \ + is no longer load-bearing, revisit observability assumptions" + ); + + // Alternate Display — what schemas.rs map_err now produces. Every + // layer joined by ": " so the rusqlite root reaches Sentry. + let full = format!("{err:#}"); + assert!( + full.contains("failed to run subconscious schema DDL"), + "chain-formatted message must include outer wrapper, got: {full}" + ); + assert!( + full.contains("execute_batch failed"), + "chain-formatted message must include middle context, got: {full}" + ); + assert!( + full.contains("database is locked"), + "chain-formatted message must include the rusqlite root, got: {full}" + ); +} + +#[test] +fn anyhow_alternate_display_includes_rusqlite_error_chain() { + use anyhow::Context; + + // Simulate the exact shape produced by `with_connection`: + // a real rusqlite Error wrapped in `with_context(...)`. + let raw = rusqlite::Error::SqliteFailure( + rusqlite::ffi::Error { + code: rusqlite::ErrorCode::DatabaseBusy, + extended_code: 5, + }, + Some("database is locked".into()), + ); + let wrapped: anyhow::Result<()> = + Err(anyhow::Error::from(raw)).context("failed to run subconscious schema DDL"); + + let err = wrapped.unwrap_err(); + let chained = format!("{err:#}"); + + // Outer wrapper preserved. + assert!(chained.contains("failed to run subconscious schema DDL")); + // rusqlite-rendered root preserved — this is the signal Sentry + // needs to distinguish a DDL lock-race from a corruption / disk-full + // / permission failure. Without it, all four fingerprint identically. + assert!( + chained.contains("database is locked"), + "rusqlite root must appear in chain-formatted message, got: {chained}" + ); +} diff --git a/src/openhuman/subconscious/store.rs b/src/openhuman/subconscious/store.rs index 2aed761cb..23b3df9c5 100644 --- a/src/openhuman/subconscious/store.rs +++ b/src/openhuman/subconscious/store.rs @@ -6,6 +6,7 @@ use anyhow::{Context, Result}; use rusqlite::{Connection, OptionalExtension}; use std::path::Path; +use std::time::Duration; use uuid::Uuid; use super::types::{ @@ -13,7 +14,46 @@ use super::types::{ TaskPatch, TaskRecurrence, TaskSource, }; +/// Per-connection busy handler window. Tracks the value used by the cron +/// module + other domain stores (`memory_store::unified::init`, 15s) and +/// the higher-throughput whatsapp/memory_queue path (5s). 5 s is enough +/// for the subconscious tick — RPC handlers are user-driven (status +/// polling at 3 s, manual triggers) and we'd rather fail fast than block +/// the UI thread for the full 15 s of contention. +const BUSY_TIMEOUT: Duration = Duration::from_millis(5000); + +/// Maximum number of application-level retries after rusqlite's busy +/// handler is exhausted. The first attempt is "attempt 0" — total +/// attempts = `OPEN_RETRY_ATTEMPTS` + 1. +const OPEN_RETRY_ATTEMPTS: u32 = 3; + +/// Base backoff for application-level retries; per-attempt sleep is +/// `BASE * 3^attempt` so the schedule is `100 ms / 300 ms / 900 ms` +/// totalling ≤ 1.3 s before the final attempt fails through. +const OPEN_RETRY_BASE_MS: u64 = 100; + /// Open the subconscious database and run schema migrations. +/// +/// Three layers of defence against transient `SQLITE_BUSY` / `SQLITE_LOCKED` +/// at the open / DDL boundary, motivated by Sentry TAURI-RUST-A +/// (cross-platform, ~1.3k events / 24 h, RPC paths `subconscious_tasks_list` +/// and `subconscious_status`): +/// +/// 1. **Per-connection busy timeout** (`BUSY_TIMEOUT`, 5 s): SQLite's +/// default is `0` — first lock contention returns `SQLITE_BUSY` +/// immediately. The subconscious domain serialises several RPCs +/// (status poll every 3 s, tasks-list on Intelligence page, manual +/// trigger), each opening its own connection; without a timeout the +/// first concurrent open races and one returns `SQLITE_BUSY` mid-DDL. +/// 2. **Application-level retry** (3 attempts, exponential backoff +/// 100 / 300 / 900 ms): catches the residual case where the busy +/// handler is exhausted (long-running external write txn, AV scan +/// holding the file). Mirrors `whatsapp_data::sqlite_retry` / +/// `memory_queue::worker::is_sqlite_busy`. +/// 3. **Retry classifier** (`is_sqlite_busy`): only retries +/// `DatabaseBusy` / `DatabaseLocked`. Schema / syntax / corruption +/// errors are real bugs or unrecoverable file-state failures — +/// retrying just delays the report. pub fn with_connection( workspace_dir: &Path, f: impl FnOnce(&Connection) -> Result, @@ -24,11 +64,72 @@ pub fn with_connection( .with_context(|| format!("failed to create subconscious dir: {}", parent.display()))?; } - let conn = Connection::open(&db_path) + let conn = open_and_initialize_with_retry(&db_path)?; + f(&conn) +} + +/// Open the SQLite file, set `busy_timeout`, run `SCHEMA_DDL`, and apply +/// the idempotent reflection-store migrations — retrying the whole +/// sequence on `SQLITE_BUSY` / `SQLITE_LOCKED`. Split out so the retry +/// loop has a single failure surface to classify and the happy path +/// stays linear. +fn open_and_initialize_with_retry(db_path: &Path) -> Result { + let mut last_err: Option = None; + + for attempt in 0..=OPEN_RETRY_ATTEMPTS { + match open_and_initialize(db_path) { + Ok(conn) => { + if attempt > 0 { + tracing::debug!( + target: "openhuman::subconscious::store", + attempt = attempt, + db_path = %db_path.display(), + "[subconscious::store] open/DDL succeeded after {attempt} busy retries" + ); + } + return Ok(conn); + } + Err(e) => { + if !is_sqlite_busy(&e) || attempt == OPEN_RETRY_ATTEMPTS { + last_err = Some(e); + break; + } + let sleep_ms = OPEN_RETRY_BASE_MS + .saturating_mul(3u64.saturating_pow(attempt)) + .min(900); + tracing::warn!( + target: "openhuman::subconscious::store", + attempt = attempt + 1, + max_attempts = OPEN_RETRY_ATTEMPTS + 1, + sleep_ms = sleep_ms, + error = %format!("{e:#}"), + "[subconscious::store] SQLite busy/locked on open or DDL; retrying" + ); + std::thread::sleep(Duration::from_millis(sleep_ms)); + last_err = Some(e); + } + } + } + + Err(last_err.expect("OPEN_RETRY_ATTEMPTS >= 0 ensures at least one attempt")) +} + +/// Single-shot open + DDL + migrations. Each invocation returns an +/// owned `Connection`; on failure the partially-initialised connection +/// is dropped before the caller retries. +fn open_and_initialize(db_path: &Path) -> Result { + let conn = Connection::open(db_path) .with_context(|| format!("failed to open subconscious DB: {}", db_path.display()))?; + // Set busy_timeout BEFORE running DDL — the very first PRAGMA / CREATE + // TABLE in SCHEMA_DDL can race with another in-process connection + // (subconscious RPCs each call `with_connection` independently), and + // SQLite's default busy_timeout is 0. + conn.busy_timeout(BUSY_TIMEOUT) + .context("configure subconscious busy_timeout")?; + conn.execute_batch(SCHEMA_DDL) - .with_context(|| "failed to run subconscious schema DDL")?; + .context("failed to run subconscious schema DDL")?; // Drop the legacy `disposition` / `surfaced_at` columns + their index // from previously-migrated DBs. Idempotent — fresh installs and @@ -39,7 +140,33 @@ pub fn with_connection( // Idempotent (duplicate-column errors swallowed). super::reflection_store::migrate_add_source_chunks_column(&conn); - f(&conn) + Ok(conn) +} + +/// Returns true when `err` is transient SQLite contention worth retrying +/// (`SQLITE_BUSY` / `SQLITE_LOCKED`). Schema / syntax / corruption errors +/// are NOT retried — the retry would just delay the same failure. +/// +/// Modelled on [`crate::openhuman::memory_queue::worker::is_sqlite_busy`] +/// and [`crate::openhuman::whatsapp_data::sqlite_retry::is_sqlite_busy`]; +/// kept private to the subconscious store so the retry policy can evolve +/// independently of those sibling domains. +fn is_sqlite_busy(err: &anyhow::Error) -> bool { + if let Some(rusqlite::Error::SqliteFailure(sqlite_err, _)) = + err.downcast_ref::() + { + return matches!( + sqlite_err.code, + rusqlite::ErrorCode::DatabaseBusy | rusqlite::ErrorCode::DatabaseLocked + ); + } + // Fallback for errors wrapped under `.context(...)` layers — the + // rusqlite root may sit a few levels deep after `with_context` + // wraps the open / DDL failure. anyhow's alternate Display joins + // every cause with ": " so the SQLite-rendered phrase remains + // searchable. + let msg = format!("{err:#}").to_ascii_lowercase(); + msg.contains("database is locked") || msg.contains("database table is locked") } const SCHEMA_DDL: &str = " diff --git a/src/openhuman/subconscious/store_tests.rs b/src/openhuman/subconscious/store_tests.rs index f191baa8b..6948c8fa9 100644 --- a/src/openhuman/subconscious/store_tests.rs +++ b/src/openhuman/subconscious/store_tests.rs @@ -156,3 +156,186 @@ fn recurrence_roundtrip() { TaskRecurrence::Cron("0 8 * * *".into()) ); } + +// ── DDL resilience: classifier + retry happy path ────────────── +// +// These guards back Sentry TAURI-RUST-A: the production failure is +// `Connection::open` + `execute_batch(SCHEMA_DDL)` racing against +// another in-process connection that holds the write lock. With +// `BUSY_TIMEOUT` set and the application-level retry loop in place, +// the race resolves on its own; without them the first attempt +// returns `SQLITE_BUSY` and the user sees "failed to run subconscious +// schema DDL" in Sentry with no further context. + +#[test] +fn is_sqlite_busy_matches_database_busy_code() { + let raw = rusqlite::Error::SqliteFailure( + rusqlite::ffi::Error { + code: rusqlite::ErrorCode::DatabaseBusy, + extended_code: 5, // SQLITE_BUSY + }, + Some("database is locked".into()), + ); + let err = anyhow::Error::from(raw); + assert!(is_sqlite_busy(&err)); +} + +#[test] +fn is_sqlite_busy_matches_database_locked_code() { + let raw = rusqlite::Error::SqliteFailure( + rusqlite::ffi::Error { + code: rusqlite::ErrorCode::DatabaseLocked, + extended_code: 6, // SQLITE_LOCKED + }, + Some("database table is locked".into()), + ); + let err = anyhow::Error::from(raw); + assert!(is_sqlite_busy(&err)); +} + +#[test] +fn is_sqlite_busy_does_not_match_constraint_violation() { + let raw = rusqlite::Error::SqliteFailure( + rusqlite::ffi::Error { + code: rusqlite::ErrorCode::ConstraintViolation, + extended_code: 19, + }, + Some("UNIQUE constraint failed".into()), + ); + let err = anyhow::Error::from(raw); + assert!(!is_sqlite_busy(&err)); +} + +#[test] +fn is_sqlite_busy_does_not_match_schema_syntax_error() { + // A genuine bug in `SCHEMA_DDL` (e.g. typo in CREATE TABLE) would + // surface as a `SqliteFailure(Unknown, ...)` with "syntax error" + // in the message — retrying just delays the same failure, so the + // classifier must reject it. Use Unknown + non-busy message. + let raw = rusqlite::Error::SqliteFailure( + rusqlite::ffi::Error { + code: rusqlite::ErrorCode::Unknown, + extended_code: 1, + }, + Some("near \"FOO\": syntax error".into()), + ); + let err = anyhow::Error::from(raw); + assert!(!is_sqlite_busy(&err)); +} + +#[test] +fn is_sqlite_busy_matches_through_context_layers() { + // The production failure shape: a rusqlite error wrapped under + // `with_context("failed to run subconscious schema DDL")` — + // exactly what `open_and_initialize` produces. Downcast must + // still find the rusqlite root. + let raw = rusqlite::Error::SqliteFailure( + rusqlite::ffi::Error { + code: rusqlite::ErrorCode::DatabaseBusy, + extended_code: 5, + }, + Some("database is locked".into()), + ); + let wrapped: anyhow::Result<()> = Err(anyhow::Error::from(raw)) + .with_context(|| "failed to run subconscious schema DDL".to_string()); + let err = wrapped.unwrap_err(); + assert!(is_sqlite_busy(&err)); +} + +#[test] +fn is_sqlite_busy_text_fallback_when_downcast_misses() { + // If a future refactor stringifies the rusqlite error before + // wrapping (e.g. via anyhow!("{e}")), the downcast misses but + // the chain-formatter text still preserves "database is locked". + let err = anyhow::anyhow!("failed to run subconscious schema DDL: database is locked"); + assert!(is_sqlite_busy(&err)); +} + +#[test] +fn with_connection_resolves_external_write_contention() { + use std::sync::mpsc; + use tempfile::TempDir; + + let tmp = TempDir::new().unwrap(); + let workspace = tmp.path().to_path_buf(); + + // First call: prime the DB so the file exists and the schema is + // initialized. Subsequent calls take the fast path. + with_connection(&workspace, |conn| { + add_task(conn, "primer", TaskSource::User, TaskRecurrence::Once)?; + Ok(()) + }) + .expect("prime DB"); + + // Hold an EXCLUSIVE write lock for ~250 ms in a side thread. + // The DDL loop in `open_and_initialize` re-runs PRAGMA journal_mode + // and CREATE TABLE IF NOT EXISTS — both are no-ops on an already + // initialized DB but still acquire the write lock briefly, which + // races against the held lock. The application-level retry + // (100 / 300 / 900 ms) plus the 5 s `busy_timeout` must absorb + // this and let the second `with_connection` succeed. + let db_path = workspace.join("subconscious").join("subconscious.db"); + let (lock_ready_tx, lock_ready_rx) = mpsc::channel::<()>(); + let (release_tx, release_rx) = mpsc::channel::<()>(); + let blocker = std::thread::spawn(move || { + let conn = rusqlite::Connection::open(&db_path).expect("open blocker conn"); + conn.busy_timeout(std::time::Duration::from_millis(100)) + .expect("blocker busy_timeout"); + let tx = conn + .unchecked_transaction() + .expect("begin blocker transaction"); + // Force write-lock acquisition immediately. + tx.execute( + "INSERT INTO subconscious_tasks \ + (id, title, source, recurrence, created_at) \ + VALUES ('blocker', 'blocker', 'user', 'pending', 0.0)", + [], + ) + .expect("blocker insert"); + lock_ready_tx.send(()).expect("signal lock acquired"); + // Wait for the main thread to start contending, then a touch + // longer so the first one or two retries collide. + release_rx + .recv_timeout(std::time::Duration::from_secs(2)) + .expect("release signal"); + std::thread::sleep(std::time::Duration::from_millis(50)); + tx.rollback().expect("rollback blocker txn"); + }); + + // Wait until the blocker actually holds the write lock. + lock_ready_rx + .recv_timeout(std::time::Duration::from_secs(2)) + .expect("blocker never acquired lock"); + + // Contender: should retry through the busy window and succeed + // once the blocker rolls back. We release the blocker after + // ~250 ms so the second / third retry attempt lands in the + // unlocked window. + let release_tx_for_timer = release_tx.clone(); + let timer = std::thread::spawn(move || { + std::thread::sleep(std::time::Duration::from_millis(250)); + let _ = release_tx_for_timer.send(()); + }); + + let result = with_connection(&workspace, |conn| { + // Confirm we can issue a real query through the contended + // connection — proves the open + DDL completed cleanly. + let count: i64 = conn + .query_row("SELECT COUNT(*) FROM subconscious_tasks", [], |row| { + row.get(0) + }) + .unwrap_or(-1); + Ok(count) + }); + + timer.join().expect("timer thread panicked"); + blocker.join().expect("blocker thread panicked"); + + let count = result.expect("contended with_connection must succeed via retry"); + // Primer row is "primer"; blocker's INSERT was rolled back, so + // the count should be exactly 1. + assert_eq!( + count, 1, + "expected only the primer row after blocker rollback, got {count}" + ); +}