diff --git a/desktop/src-tauri/src/lib.rs b/desktop/src-tauri/src/lib.rs index 2219cb4494..255362b212 100644 --- a/desktop/src-tauri/src/lib.rs +++ b/desktop/src-tauri/src/lib.rs @@ -62,6 +62,7 @@ use huddle::{ start_huddle, start_stt_pipeline, HuddlePhase, }; use initial_window::*; +use managed_agents::store_journal::{run_recovery_gate, store_anchor_dir}; use managed_agents::{ backfill_persona_snapshots, ensure_nest, list_managed_agent_runtimes, put_managed_agent_runtime_lifecycle, reconcile_managed_agent_runtimes, @@ -340,31 +341,32 @@ pub fn run() { return Ok(()); } - // File-commit recovery MUST run before any migration, backfill, or - // canonical-store reader/writer. A stale stage from a pre-crash - // commit must not overwrite migration output. One owner, one site. - // Fail closed: a repair error sets store_recovery_failed and skips - // all store-touching setup. + // run_recovery_gate: file-commit recovery before migrations; fail closed. { let state = app_handle.state::(); - if let Err(e) = - crate::managed_agents::store_journal::run_file_commit_recovery(&app_handle) - { + let anchor = store_anchor_dir(&app_handle).unwrap_or_else(|_| { + app_handle + .path() + .app_data_dir() + .unwrap_or_default() + .join("agents") + }); + let _ = std::fs::create_dir_all(&anchor); + let reset_done = reset_outcome.completed; + if let Err(e) = run_recovery_gate(&anchor, || { + if reset_done { + migration::run_boot_migrations_after_reset(&app_handle); + } else { + migration::run_boot_migrations(&app_handle); + } + Ok(()) + }) { eprintln!("buzz-desktop: file-commit-recovery: {e}"); - state - .store_recovery_failed - .store(true, std::sync::atomic::Ordering::Release); + state.store_recovery_failed.store(true, Ordering::Release); return Ok(()); } } - // Run all pre-identity data migrations before state loads from disk. - if reset_outcome.completed { - migration::run_boot_migrations_after_reset(&app_handle); - } else { - migration::run_boot_migrations(&app_handle); - } - // Resolve persisted identity key (env var → file → generate+save). // This is fatal — the app should not start with an ephemeral identity // that will be lost on restart, as that silently breaks channel diff --git a/desktop/src-tauri/src/managed_agents/store_journal/mod.rs b/desktop/src-tauri/src/managed_agents/store_journal/mod.rs index 7746151c11..abcb46c59f 100644 --- a/desktop/src-tauri/src/managed_agents/store_journal/mod.rs +++ b/desktop/src-tauri/src/managed_agents/store_journal/mod.rs @@ -62,8 +62,8 @@ pub use schema::apply_journal_schema_pub; pub use schema::open_journal; pub use txn::{ - advance_to_committed, mutate_store, prepare_publication, run_boot_recovery, - run_file_commit_recovery, StoreState, + advance_to_committed, mutate_store, prepare_publication, run_boot_recovery, run_recovery_gate, + StoreState, }; #[cfg(test)] #[allow(unused_imports)] diff --git a/desktop/src-tauri/src/managed_agents/store_journal/txn.rs b/desktop/src-tauri/src/managed_agents/store_journal/txn.rs index 541b46f20d..43f490d919 100644 --- a/desktop/src-tauri/src/managed_agents/store_journal/txn.rs +++ b/desktop/src-tauri/src/managed_agents/store_journal/txn.rs @@ -26,6 +26,7 @@ use std::sync::MutexGuard; use rusqlite::params; use tauri::AppHandle; +use tauri::Manager; use crate::managed_agents::{ManagedAgentRecord, TeamRecord}; @@ -96,6 +97,29 @@ fn advance_file_commit_phase( Ok(()) } +/// Run file-commit recovery and, only if it succeeds with zero unresolved +/// commits, execute `store_work`. +/// +/// This is the structural seam that enforces "recovery before canonical-store +/// access". Both production setup and tests drive the same function, so the +/// ordering invariant lives in one place and cannot drift. +/// +/// Returns `Err` if recovery fails (`Err` or unresolved count > 0) or if +/// `store_work` returns `Err`. On recovery failure `store_work` is never +/// called. +pub fn run_recovery_gate(anchor: &std::path::Path, store_work: F) -> Result<(), String> +where + F: FnOnce() -> Result<(), String>, +{ + let unresolved = file_commit_recovery_at(anchor)?; + if unresolved > 0 { + return Err(format!( + "file-commit recovery: {unresolved} nonterminal commit(s) could not be resolved" + )); + } + store_work() +} + /// Mutate the store under the full lock sequence. /// /// The closure runs inside a real `rusqlite::Transaction` — if it returns @@ -104,6 +128,10 @@ fn advance_file_commit_phase( /// /// On any decode error inside this function no file is written, no journal /// transition occurs, and the error is propagated with `?`. +/// +/// Returns `Err` immediately if the `store_recovery_failed` flag is set on +/// `AppState` — a recovery failure at boot means the store is in an uncertain +/// state and no mutations are safe until the user relaunches. pub fn mutate_store<'g, F, T>( app: &AppHandle, store_mutex_guard: MutexGuard<'g, ()>, @@ -112,6 +140,20 @@ pub fn mutate_store<'g, F, T>( where F: FnOnce(StoreState<'_>) -> Result<(Vec, Vec, T), String>, { + // Guard: a boot recovery failure means the store is in an uncertain state. + // Reject all mutations until the user relaunches and recovery succeeds. + { + use std::sync::atomic::Ordering; + let state = app.state::(); + if state.store_recovery_failed.load(Ordering::Acquire) { + return Err( + "store mutation rejected: boot file-commit recovery failed; \ + relaunch to retry" + .to_string(), + ); + } + } + let anchor = store_anchor_dir(app)?; std::fs::create_dir_all(&anchor).map_err(|e| format!("create anchor dir: {e}"))?; @@ -415,9 +457,9 @@ pub fn run_boot_recovery( /// Path-level boot recovery — **publication-only** (no file repair). /// /// File-commit recovery (phase 1) is handled exclusively by the synchronous -/// `run_file_commit_recovery` call in app setup, which runs BEFORE this -/// background task is spawned. This function runs only phase 2: publication -/// recovery (outbox re-drive into the retention DB). +/// `run_recovery_gate` call in app setup, which runs BEFORE this background +/// task is spawned. This function runs only phase 2: publication recovery +/// (outbox re-drive into the retention DB). /// /// `retention_db_path` is the path to the anchor's active scoped retention DB. /// When `Some`, recovery re-inserts missing `pending_sync` retention rows from @@ -428,30 +470,19 @@ pub(crate) fn run_boot_recovery_at( retention_db_path: Option<&std::path::Path>, ) -> Result<(), String> { // Publication recovery only — no file I/O, no lock required. - // File-commit repair is the sole responsibility of run_file_commit_recovery. + // File-commit repair is the sole responsibility of run_recovery_gate. let journal = open_journal(anchor)?; recover_nonterminal_operations(&journal, retention_db_path)?; Ok(()) } -/// File-commit-only recovery: acquire the advisory lock, replay any interrupted -/// two-phase file commits, then release the lock. -/// -/// Called synchronously during app setup (before commands are admitted) so the -/// store is always in a consistent state by the time the first command runs. -pub fn run_file_commit_recovery(app: &AppHandle) -> Result<(), String> { - let anchor = store_anchor_dir(app)?; - std::fs::create_dir_all(&anchor).map_err(|e| format!("boot-recovery create anchor: {e}"))?; - file_commit_recovery_at(&anchor) -} - /// Path-level file-commit recovery, extracted for testing without an AppHandle. #[cfg(test)] -pub(crate) fn file_commit_recovery_at_pub(anchor: &std::path::Path) -> Result<(), String> { +pub(crate) fn file_commit_recovery_at_pub(anchor: &std::path::Path) -> Result { file_commit_recovery_at(anchor) } -fn file_commit_recovery_at(anchor: &std::path::Path) -> Result<(), String> { +fn file_commit_recovery_at(anchor: &std::path::Path) -> Result { let _advisory = super::lock::JournalLockGuard::acquire(anchor)?; let journal = open_journal(anchor)?; recover_interrupted_file_commits(&journal) @@ -497,8 +528,14 @@ fn sha256_hex_of_file(path: &std::path::Path) -> Option { Some(hex::encode(sha2::Sha256::digest(&bytes))) } +/// Recover any interrupted two-phase file commits. +/// +/// Returns the number of nonterminal commits that could NOT be resolved (any +/// branch that previously `continue`d without completing the rename sequence). +/// A non-zero count means the store is in an uncertain state and callers must +/// treat it as a recovery failure. #[allow(clippy::type_complexity)] -fn recover_interrupted_file_commits(journal: &rusqlite::Connection) -> Result<(), String> { +fn recover_interrupted_file_commits(journal: &rusqlite::Connection) -> Result { // Read commit_id, phase, stage paths, AND content hashes. let rows: Vec<(String, String, String, String, String, String)> = { let mut stmt = journal @@ -528,6 +565,8 @@ fn recover_interrupted_file_commits(journal: &rusqlite::Connection) -> Result<() .map_err(|e| format!("read file_commit_phases row: {e}"))? }; + let mut unresolved: usize = 0; + for (commit_id, phase_str, agents_stage_str, teams_stage_str, agents_hash, teams_hash) in rows { let phase = FileCommitPhase::from_str(&phase_str).unwrap_or(FileCommitPhase::Intent); let agents_stage = std::path::PathBuf::from(&agents_stage_str); @@ -540,6 +579,15 @@ fn recover_interrupted_file_commits(journal: &rusqlite::Connection) -> Result<() let agents_canonical = canonical_from_stage(&agents_stage); let teams_canonical = canonical_from_stage(&teams_stage); + // Macro that counts a skipped commit as unresolved and moves to the next. + macro_rules! skip_unresolved { + ($msg:expr) => {{ + eprintln!("{}", $msg); + unresolved += 1; + continue; + }}; + } + match phase { FileCommitPhase::Intent => { // Both stage files should exist for a full intent commit. @@ -554,7 +602,8 @@ fn recover_interrupted_file_commits(journal: &rusqlite::Connection) -> Result<() && sha256_hex_of_file(&teams_canonical).as_deref() == Some(&teams_hash); if agents_ok && teams_ok { eprintln!( - "buzz-desktop: boot-recovery: {commit_id}: both stage files absent, canonicals verified — advancing to committed" + "buzz-desktop: boot-recovery: {commit_id}: both stage files absent, \ + canonicals verified — advancing to committed" ); advance_file_commit_phase(journal, &commit_id, &FileCommitPhase::Committed) .unwrap_or_else(|e| { @@ -563,25 +612,28 @@ fn recover_interrupted_file_commits(journal: &rusqlite::Connection) -> Result<() ) }); } else { - eprintln!( - "buzz-desktop: boot-recovery: {commit_id}: both stage files absent in intent phase and canonicals do not match recorded hashes — failing closed; manual recovery may be needed" - ); + skip_unresolved!(format!( + "buzz-desktop: boot-recovery: {commit_id}: both stage files absent \ + in intent phase and canonicals do not match recorded hashes — \ + failing closed; manual recovery may be needed" + )); } continue; } if !agents_exists { // Agents stage absent, teams stage present: ambiguous — // cannot determine if agents rename completed without proof. - // Fail closed: skip this commit; operator intervention needed. - eprintln!( - "buzz-desktop: boot-recovery: {commit_id}: agents stage absent in intent phase (teams stage present) — cannot safely complete; skipping (manual recovery may be needed)" - ); - continue; + skip_unresolved!(format!( + "buzz-desktop: boot-recovery: {commit_id}: agents stage absent in \ + intent phase (teams stage present) — cannot safely complete; \ + skipping (manual recovery may be needed)" + )); } // Agents stage exists — replay agents rename. if let Err(e) = rename_staged(&agents_stage, &agents_canonical) { - eprintln!("buzz-desktop: boot-recovery: agents rename failed: {e}"); - continue; + skip_unresolved!(format!( + "buzz-desktop: boot-recovery: agents rename failed: {e}" + )); } if let Err(e) = advance_file_commit_phase(journal, &commit_id, &FileCommitPhase::FirstRenamed) @@ -591,20 +643,20 @@ fn recover_interrupted_file_commits(journal: &rusqlite::Connection) -> Result<() // Teams rename. if teams_exists { if let Err(e) = rename_staged(&teams_stage, &teams_canonical) { - eprintln!("buzz-desktop: boot-recovery: teams rename failed: {e}"); - continue; + skip_unresolved!(format!( + "buzz-desktop: boot-recovery: teams rename failed: {e}" + )); } } else { // Teams stage absent after agents rename. Verify via hash. let teams_ok = !teams_hash.is_empty() && sha256_hex_of_file(&teams_canonical).as_deref() == Some(&teams_hash); if !teams_ok { - eprintln!( + skip_unresolved!(format!( "buzz-desktop: boot-recovery: {commit_id}: teams stage absent \ in intent phase (after agents rename) and canonical does not \ match recorded hash — failing closed" - ); - continue; + )); } } advance_file_commit_phase(journal, &commit_id, &FileCommitPhase::Committed) @@ -616,10 +668,9 @@ fn recover_interrupted_file_commits(journal: &rusqlite::Connection) -> Result<() // Agents already renamed; only teams stage remains. if teams_stage.exists() { if let Err(e) = rename_staged(&teams_stage, &teams_canonical) { - eprintln!( + skip_unresolved!(format!( "buzz-desktop: boot-recovery: teams rename (first_renamed) failed: {e}" - ); - continue; + )); } // Teams rename succeeded — advance to committed. } else { @@ -627,10 +678,11 @@ fn recover_interrupted_file_commits(journal: &rusqlite::Connection) -> Result<() let teams_ok = !teams_hash.is_empty() && sha256_hex_of_file(&teams_canonical).as_deref() == Some(&teams_hash); if !teams_ok { - eprintln!( - "buzz-desktop: boot-recovery: {commit_id}: teams stage absent in first_renamed phase and canonical does not match recorded hash — failing closed; manual recovery may be needed" - ); - continue; + skip_unresolved!(format!( + "buzz-desktop: boot-recovery: {commit_id}: teams stage absent in \ + first_renamed phase and canonical does not match recorded hash — \ + failing closed; manual recovery may be needed" + )); } // Canonical matches — teams rename already completed. } @@ -643,7 +695,7 @@ fn recover_interrupted_file_commits(journal: &rusqlite::Connection) -> Result<() } } - Ok(()) + Ok(unresolved) } /// Recover nonterminal operations from the journal outbox. diff --git a/desktop/src-tauri/src/managed_agents/store_journal_fix_tests.rs b/desktop/src-tauri/src/managed_agents/store_journal_fix_tests.rs index ba58313e4d..fd0649b6da 100644 --- a/desktop/src-tauri/src/managed_agents/store_journal_fix_tests.rs +++ b/desktop/src-tauri/src/managed_agents/store_journal_fix_tests.rs @@ -6,7 +6,7 @@ use super::operations::insert_operation; use super::{ apply_journal_schema_pub, file_commit_recovery_at_pub, insert_outbox_event, open_journal, - run_boot_recovery_at, Generation, + run_boot_recovery_at, run_recovery_gate, Generation, }; use crate::managed_agents::retention::{ get_pending_sync, open_retention_db, tombstone_retention_d_tag, @@ -674,44 +674,24 @@ fn test_sync_file_recovery_repairs_dangling_intent_row() { ); } -/// Fix A — production-order seam test: file-commit recovery must complete -/// BEFORE any migration or canonical-store reader observes the canonical files. -/// -/// Seam used: `file_commit_recovery_at_pub` is the exact function that -/// `run_file_commit_recovery` delegates to in production — not a -/// reimplementation. The "migration read" is a direct filesystem read of the -/// canonical path, matching what every real migration/backfill does. -/// -/// Scenario: a process crashed after writing a stage file but before the rename -/// completed. The canonical (`managed-agents.json`) still holds pre-crash -/// ("stale") content. The stage file holds the intended post-crash ("repaired") -/// content. After recovery runs, the canonical must hold the repaired content — -/// so any subsequent migration or backfill reader sees the repaired state. +/// Fix A — ordered seam: `run_recovery_gate` runs recovery before invoking the +/// closure; the closure (= migrations) observes the repaired canonical. +/// Both production and this test drive the same function. #[test] fn test_setup_order_recovery_completes_before_migration_reads_canonical() { let dir = tmp_dir(); let anchor = dir.path().to_path_buf(); - - // "Stale" content — what canonical holds before recovery. let stale_content = b"[{\"pubkey\":\"stale-pre-crash\"}]"; - // "Repaired" content — what the stage file holds (the intended commit). let repaired_content = b"[{\"pubkey\":\"repaired-post-crash\"}]"; let teams_content = b"[]"; - let a_can = anchor.join("managed-agents.json"); let t_can = anchor.join("teams.json"); let a_stage = anchor.join("managed-agents.order_test.stage"); let t_stage = anchor.join("teams.order_test.stage"); - - // Write stale canonical — this is what migrations would see WITHOUT recovery. std::fs::write(&a_can, stale_content).unwrap(); std::fs::write(&t_can, teams_content).unwrap(); - - // Write stage files containing the repaired (intended) state. std::fs::write(&a_stage, repaired_content).unwrap(); std::fs::write(&t_stage, teams_content).unwrap(); - - // Plant the intent row — recovery will detect and replay it. { let j = open_journal(&anchor).unwrap(); insert_phase_row( @@ -724,25 +704,124 @@ fn test_setup_order_recovery_completes_before_migration_reads_canonical() { &sha256_hex(teams_content), ); } - - // Run recovery — the exact same code path production calls via - // run_file_commit_recovery → file_commit_recovery_at. - file_commit_recovery_at_pub(&anchor).unwrap(); - - // Now simulate a "migration read" — read the canonical directly, as every - // real migration/backfill does. - let observed = std::fs::read(&a_can).unwrap(); + let mut closure_saw: Option> = None; + let saw = &mut closure_saw; + run_recovery_gate(&anchor, || { + *saw = Some(std::fs::read(&a_can).unwrap()); + Ok(()) + }) + .expect("run_recovery_gate must succeed"); + let observed = closure_saw.expect("closure must have run"); assert_eq!( observed, repaired_content, - "migration read must observe the repaired canonical, not the stale pre-crash state" + "closure must see repaired canonical" ); - // Stale content must not be present. assert_ne!( observed, stale_content, - "stale canonical content must have been replaced by recovery" + "stale canonical must be replaced first" ); } +/// Fix A + GAP 3 — fail-closed: unresolved recovery → closure never runs, +/// `run_recovery_gate` returns `Err`. +#[test] +fn test_recovery_gate_failure_skips_closure_and_returns_err() { + let dir = tmp_dir(); + let anchor = dir.path().to_path_buf(); + let a_stage = anchor.join("managed-agents.gate_fail.stage"); + let t_stage = anchor.join("teams.gate_fail.stage"); + // No stage files + empty hashes → hash-mismatch → 1 unresolved. + { + let j = open_journal(&anchor).unwrap(); + insert_phase_row( + &j, + "gate_fail", + "intent", + a_stage.to_str().unwrap(), + t_stage.to_str().unwrap(), + "", + "", + ); + } + let mut closure_ran = false; + let result = run_recovery_gate(&anchor, || { + closure_ran = true; + Ok(()) + }); + assert!( + result.is_err(), + "run_recovery_gate must return Err on unresolved commits" + ); + assert!( + !closure_ran, + "closure must not run when recovery has unresolved commits" + ); +} + +/// GAP 2 — hash-mismatch: absent stages + non-matching hashes → unresolved count > 0. +#[test] +fn test_recovery_hash_mismatch_returns_unresolved_count() { + let dir = tmp_dir(); + let anchor = dir.path().to_path_buf(); + let a_stage = anchor.join("managed-agents.hmm.stage"); + let t_stage = anchor.join("teams.hmm.stage"); + std::fs::write( + anchor.join("managed-agents.json"), + b"[{\"pubkey\":\"actual\"}]", + ) + .unwrap(); + std::fs::write(anchor.join("teams.json"), b"[]").unwrap(); + { + let j = open_journal(&anchor).unwrap(); + insert_phase_row( + &j, + "hmm", + "intent", + a_stage.to_str().unwrap(), + t_stage.to_str().unwrap(), + "deadbeefdeadbeef", + "deadbeefdeadbeef", + ); + } + let unresolved = file_commit_recovery_at_pub(&anchor).expect("must not Err"); + assert_eq!( + unresolved, 1, + "hash-mismatch with absent stages → 1 unresolved" + ); +} + +/// GAP 2 — rename-failure: stage exists but canonical target is a directory +/// (rename-over-directory is rejected by the OS) → unresolved count > 0. +#[test] +fn test_recovery_rename_failure_returns_unresolved_count() { + let dir = tmp_dir(); + let anchor = dir.path().to_path_buf(); + let repaired = b"[{\"pubkey\":\"rename-fail\"}]"; + let teams = b"[]"; + let a_stage = anchor.join("managed-agents.rfail.stage"); + let t_stage = anchor.join("teams.rfail.stage"); + let a_can = anchor.join("managed-agents.json"); + std::fs::write(&a_stage, repaired).unwrap(); + std::fs::write(&t_stage, teams).unwrap(); + // canonical is a directory → rename(file, dir) → EISDIR + std::fs::create_dir_all(&a_can).unwrap(); + std::fs::write(anchor.join("teams.json"), teams).unwrap(); + { + let j = open_journal(&anchor).unwrap(); + insert_phase_row( + &j, + "rfail", + "intent", + a_stage.to_str().unwrap(), + t_stage.to_str().unwrap(), + &sha256_hex(repaired), + &sha256_hex(teams), + ); + } + let unresolved = file_commit_recovery_at_pub(&anchor).expect("must not Err"); + assert_eq!(unresolved, 1, "rename failure → 1 unresolved"); +} + // ── Fix B: migration tests ─────────────────────────────────────────────────── /// Helper: create a v2-shaped in-memory journal (no hash/d-tag columns).