diff --git a/desktop/src-tauri/src/lib.rs b/desktop/src-tauri/src/lib.rs index 3237925fd7..6ee5bd71d7 100644 --- a/desktop/src-tauri/src/lib.rs +++ b/desktop/src-tauri/src/lib.rs @@ -597,18 +597,19 @@ pub fn run() { } }); } - crate::managed_agents::spawn_boot_recovery(app.handle()); - // File-commit recovery runs synchronously under the advisory lock - // BEFORE the invoke handler admits any store commands. This - // ensures the JSON store is in a consistent state (no dangling - // *.stage files) before any mutation can arrive. Publication - // re-drive (outbox → relay) stays in the background spawn above. + // BEFORE spawn_boot_recovery is called. This ensures the JSON + // store is in a consistent state (no dangling *.stage files) + // before the background task or any command can arrive. + // File repair has exactly ONE owner: this synchronous call. + // spawn_boot_recovery is publication-only — it does NOT run + // recover_interrupted_file_commits. if let Err(e) = crate::managed_agents::store_journal::run_file_commit_recovery(app.handle()) { eprintln!("buzz-desktop: file-commit-recovery: {e}"); } + crate::managed_agents::spawn_boot_recovery(app.handle()); Ok(()) }) .invoke_handler(tauri::generate_handler![ 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 954110968c..7746151c11 100644 --- a/desktop/src-tauri/src/managed_agents/store_journal/mod.rs +++ b/desktop/src-tauri/src/managed_agents/store_journal/mod.rs @@ -67,7 +67,7 @@ pub use txn::{ }; #[cfg(test)] #[allow(unused_imports)] -pub(crate) use txn::{read_store, run_boot_recovery_at}; +pub(crate) use txn::{file_commit_recovery_at_pub, read_store, run_boot_recovery_at}; pub use util::new_operation_id; diff --git a/desktop/src-tauri/src/managed_agents/store_journal/schema.rs b/desktop/src-tauri/src/managed_agents/store_journal/schema.rs index 92a8fdab6b..2f05c374fc 100644 --- a/desktop/src-tauri/src/managed_agents/store_journal/schema.rs +++ b/desktop/src-tauri/src/managed_agents/store_journal/schema.rs @@ -120,30 +120,86 @@ pub(super) fn apply_journal_schema(conn: &Connection) -> Result<(), String> { ) .map_err(|e| format!("apply journal schema: {e}"))?; - // Schema version 3: add content-hash columns for verifiable file-commit - // recovery. ALTER TABLE ... ADD COLUMN is ignored on fresh DBs (columns - // already exist from CREATE TABLE above). + // Schema upgrades (v3: content-hash columns; v4: retention_d_tag). + // + // We inspect existing columns via PRAGMA table_info (not "duplicate column" + // errors) so genuine failures (SQLITE_BUSY, disk full, corrupt schema) are + // propagated and the version is NOT silently advanced on partial migration. + // All ADD COLUMN operations and the user_version stamp happen inside a + // single transaction so the DB cannot be left half-migrated with a + // "completed" version stamp. let current_version: u32 = conn .pragma_query_value(None, "user_version", |row| row.get(0)) .unwrap_or(0); - if current_version < 3 { - let _ = conn.execute_batch( - "ALTER TABLE file_commit_phases ADD COLUMN agents_content_hash TEXT NOT NULL DEFAULT ''; - ALTER TABLE file_commit_phases ADD COLUMN teams_content_hash TEXT NOT NULL DEFAULT '';", - ); - } - // Schema version 4: add retention_d_tag column to outbox_events. - // Persists the exact retention coordinate (d_tag) used at enqueue time so - // boot recovery can re-insert at the same (kind, pubkey, d_tag) coordinate - // without re-parsing the event payload — necessary for kinds like 5 and - // 9035 that carry no d-tag in the event itself. + if current_version < 4 { - let _ = conn.execute_batch( - "ALTER TABLE outbox_events ADD COLUMN retention_d_tag TEXT NOT NULL DEFAULT '';", - ); + // Read existing column names BEFORE opening the transaction. + // PRAGMA table_info cannot run inside a BEGIN on some SQLite builds. + let fcp_cols = table_column_names(conn, "file_commit_phases")?; + let oe_cols = table_column_names(conn, "outbox_events")?; + let need_agents_hash = !fcp_cols.contains(&"agents_content_hash".to_string()); + let need_teams_hash = !fcp_cols.contains(&"teams_content_hash".to_string()); + let need_d_tag = !oe_cols.contains(&"retention_d_tag".to_string()); + + // Run all ADD COLUMN calls and the version stamp in one transaction. + let tx = conn + .unchecked_transaction() + .map_err(|e| format!("begin migration transaction: {e}"))?; + if need_agents_hash { + tx.execute_batch( + "ALTER TABLE file_commit_phases ADD COLUMN agents_content_hash TEXT NOT NULL DEFAULT \'\';", + ) + .map_err(|e| format!("add agents_content_hash: {e}"))?; + } + if need_teams_hash { + tx.execute_batch( + "ALTER TABLE file_commit_phases ADD COLUMN teams_content_hash TEXT NOT NULL DEFAULT \'\';", + ) + .map_err(|e| format!("add teams_content_hash: {e}"))?; + } + if need_d_tag { + tx.execute_batch( + "ALTER TABLE outbox_events ADD COLUMN retention_d_tag TEXT NOT NULL DEFAULT \'\';", + ) + .map_err(|e| format!("add retention_d_tag: {e}"))?; + } + tx.pragma_update(None, "user_version", 4) + .map_err(|e| format!("set schema user_version to 4: {e}"))?; + tx.commit() + .map_err(|e| format!("commit migration transaction: {e}"))?; + + // Post-migration verification (after commit — PRAGMA runs outside tx). + let fcp_after = table_column_names(conn, "file_commit_phases")?; + let oe_after = table_column_names(conn, "outbox_events")?; + for col in &["agents_content_hash", "teams_content_hash"] { + if !fcp_after.contains(&col.to_string()) { + return Err(format!( + "schema migration: file_commit_phases.{col} absent after migration" + )); + } + } + if !oe_after.contains(&"retention_d_tag".to_string()) { + return Err( + "schema migration: outbox_events.retention_d_tag absent after migration" + .to_string(), + ); + } } - conn.pragma_update(None, "user_version", 4) - .map_err(|e| format!("set schema user_version to 4: {e}"))?; Ok(()) } + +/// Return the column names of `table` via `PRAGMA table_info`. +fn table_column_names(conn: &Connection, table: &str) -> Result, String> { + let mut stmt = conn + .prepare(&format!("PRAGMA table_info({table})")) + .map_err(|e| format!("PRAGMA table_info({table}): {e}"))?; + let rows = stmt + .query_map([], |row| row.get::<_, String>(1)) + .map_err(|e| format!("query table_info({table}): {e}"))?; + let mut names = Vec::new(); + for row in rows { + names.push(row.map_err(|e| format!("read table_info({table}) row: {e}"))?); + } + Ok(names) +} 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 843b33529f..541b46f20d 100644 --- a/desktop/src-tauri/src/managed_agents/store_journal/txn.rs +++ b/desktop/src-tauri/src/managed_agents/store_journal/txn.rs @@ -412,29 +412,23 @@ pub fn run_boot_recovery( run_boot_recovery_at(&anchor, retention_path.as_deref()) } -/// Path-level boot recovery, extracted so tests can call it without an AppHandle. +/// Path-level boot recovery — **publication-only** (no file repair). /// -/// `retention_db_path` is the path to the anchor's active scoped retention DB. When -/// `Some`, recovery can re-insert missing `pending_sync` retention rows from +/// 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). +/// +/// `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 /// journal outbox payloads so the flush loop can re-publish them. When `None`, /// recovery logs the gap and leaves the outbox row pending for the next boot. -/// -/// **File-commit recovery runs under the anchored advisory lock** so it cannot -/// race a concurrent `mutate_store` call. Publication recovery (outbox re-drive) -/// runs under its own journal connection; it does not rename files and does not -/// require the lock. pub(crate) fn run_boot_recovery_at( anchor: &std::path::Path, retention_db_path: Option<&std::path::Path>, ) -> Result<(), String> { - // Phase 1: file-commit recovery — must hold the advisory lock so a - // concurrent mutate_store cannot race the same *.stage paths. - { - let _advisory = super::lock::JournalLockGuard::acquire(anchor)?; - let journal = open_journal(anchor)?; - recover_interrupted_file_commits(&journal)?; - } - // Phase 2: publication recovery — no file I/O, no lock required. + // Publication recovery only — no file I/O, no lock required. + // File-commit repair is the sole responsibility of run_file_commit_recovery. let journal = open_journal(anchor)?; recover_nonterminal_operations(&journal, retention_db_path)?; Ok(()) @@ -448,8 +442,18 @@ pub(crate) fn run_boot_recovery_at( 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}"))?; - let _advisory = super::lock::JournalLockGuard::acquire(&anchor)?; - let journal = open_journal(&anchor)?; + 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> { + file_commit_recovery_at(anchor) +} + +fn file_commit_recovery_at(anchor: &std::path::Path) -> Result<(), String> { + let _advisory = super::lock::JournalLockGuard::acquire(anchor)?; + let journal = open_journal(anchor)?; recover_interrupted_file_commits(&journal) } 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 c800fb5d96..ef8870b98e 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 @@ -4,7 +4,10 @@ //! recovery-vs-mutation race (BLOCKING 2b), and failure injection (BLOCKING 2c). use super::operations::insert_operation; -use super::{insert_outbox_event, open_journal, run_boot_recovery_at, Generation}; +use super::{ + apply_journal_schema_pub, file_commit_recovery_at_pub, insert_outbox_event, open_journal, + run_boot_recovery_at, Generation, +}; use crate::managed_agents::retention::{ get_pending_sync, open_retention_db, tombstone_retention_d_tag, }; @@ -70,7 +73,7 @@ fn test_file_recovery_rename_done_phase_not_updated_succeeds() { &sha256_hex(tc_data), ); drop(j); - run_boot_recovery_at(&anchor, None).unwrap(); + file_commit_recovery_at_pub(&anchor).unwrap(); // Row should be committed now (file_commit_phases phase = 'committed'). let j = open_journal(&anchor).unwrap(); let count: i64 = j @@ -106,7 +109,7 @@ fn test_file_recovery_hash_mismatch_fails_closed() { "badhash", ); drop(j); - run_boot_recovery_at(&anchor, None).unwrap(); + file_commit_recovery_at_pub(&anchor).unwrap(); // Phase must NOT be committed — fail closed. let j = open_journal(&anchor).unwrap(); let count: i64 = j @@ -143,7 +146,7 @@ fn test_file_recovery_first_renamed_teams_done_hash_verified() { &sha256_hex(tc_data), ); drop(j); - run_boot_recovery_at(&anchor, None).unwrap(); + file_commit_recovery_at_pub(&anchor).unwrap(); let j = open_journal(&anchor).unwrap(); let count: i64 = j .query_row( @@ -177,7 +180,7 @@ fn test_file_recovery_first_renamed_teams_absent_no_hash_fails_closed() { "", ); drop(j); - run_boot_recovery_at(&anchor, None).unwrap(); + file_commit_recovery_at_pub(&anchor).unwrap(); let j = open_journal(&anchor).unwrap(); let count: i64 = j .query_row( @@ -571,3 +574,207 @@ fn test_prepare_publication_propagates_journal_write_failure() { "retention DB must be untouched when journal write fails" ); } + +// ── Fix A: setup-order seam tests ──────────────────────────────────────────── + +/// Fix A: the background publication-only path (`run_boot_recovery_at`) must NOT +/// rename stage files or advance `file_commit_phases` rows. Only the synchronous +/// `file_commit_recovery_at_pub` (= `run_file_commit_recovery` in production) may +/// perform file repair. +#[test] +fn test_background_recovery_does_not_repair_files() { + let dir = tmp_dir(); + let anchor = dir.path().to_path_buf(); + let a_stage = anchor.join("managed-agents.seam.stage"); + let t_stage = anchor.join("teams.seam.stage"); + std::fs::write(&a_stage, b"[{\"pubkey\":\"aa\"}]").unwrap(); + std::fs::write(&t_stage, b"[]").unwrap(); + { + let j = open_journal(&anchor).unwrap(); + insert_phase_row( + &j, + "seam", + "intent", + a_stage.to_str().unwrap(), + t_stage.to_str().unwrap(), + "", + "", + ); + } + // Run publication-only recovery — must leave stage files and phase row untouched. + run_boot_recovery_at(&anchor, None).unwrap(); + assert!( + a_stage.exists(), + "background recovery must not rename agents stage file" + ); + assert!( + t_stage.exists(), + "background recovery must not rename teams stage file" + ); + let j = open_journal(&anchor).unwrap(); + let committed: i64 = j + .query_row( + "SELECT COUNT(*) FROM file_commit_phases WHERE phase='committed'", + [], + |r| r.get(0), + ) + .unwrap(); + assert_eq!( + committed, 0, + "background recovery must not advance phase row" + ); +} + +/// Fix A: the synchronous file-commit recovery path (`file_commit_recovery_at_pub`) +/// repairs the same fixture that the background path left untouched. +#[test] +fn test_sync_file_recovery_repairs_dangling_intent_row() { + let dir = tmp_dir(); + let anchor = dir.path().to_path_buf(); + let a_stage = anchor.join("managed-agents.sync.stage"); + let t_stage = anchor.join("teams.sync.stage"); + let a_can = anchor.join("managed-agents.json"); + let t_can = anchor.join("teams.json"); + let a_data = b"[{\"pubkey\":\"bb\"}]"; + let t_data = b"[]"; + std::fs::write(&a_stage, a_data).unwrap(); + std::fs::write(&t_stage, t_data).unwrap(); + { + let j = open_journal(&anchor).unwrap(); + insert_phase_row( + &j, + "sync", + "intent", + a_stage.to_str().unwrap(), + t_stage.to_str().unwrap(), + &sha256_hex(a_data), + &sha256_hex(t_data), + ); + } + file_commit_recovery_at_pub(&anchor).unwrap(); + assert!( + a_can.exists(), + "sync recovery must rename agents stage to canonical" + ); + assert!( + t_can.exists(), + "sync recovery must rename teams stage to canonical" + ); + let j = open_journal(&anchor).unwrap(); + let committed: i64 = j + .query_row( + "SELECT COUNT(*) FROM file_commit_phases WHERE phase='committed'", + [], + |r| r.get(0), + ) + .unwrap(); + assert_eq!( + committed, 1, + "sync recovery must advance phase row to committed" + ); +} + +// ── Fix B: migration tests ─────────────────────────────────────────────────── + +/// Helper: create a v2-shaped in-memory journal (no hash/d-tag columns). +fn v2_journal() -> rusqlite::Connection { + let conn = rusqlite::Connection::open_in_memory().unwrap(); + conn.pragma_update(None, "foreign_keys", "ON").unwrap(); + // Apply base schema manually without the migration columns. + conn.execute_batch(" + CREATE TABLE key_generations (key_id TEXT PRIMARY KEY, generation TEXT NOT NULL, is_tombstone INTEGER NOT NULL DEFAULT 0, updated_at INTEGER NOT NULL DEFAULT 0); + CREATE TABLE operations (operation_id TEXT PRIMARY KEY, kind TEXT NOT NULL, key_id TEXT NOT NULL, disposition TEXT NOT NULL DEFAULT 'pending', generation TEXT NOT NULL, compensation_id TEXT, compensation_generation TEXT, nonterminal_follow_up INTEGER NOT NULL DEFAULT 0, created_at INTEGER NOT NULL, updated_at INTEGER NOT NULL); + CREATE TABLE outbox_events (event_id TEXT PRIMARY KEY, operation_id TEXT NOT NULL REFERENCES operations(operation_id), payload BLOB NOT NULL, published_state INTEGER NOT NULL DEFAULT 0, created_at INTEGER NOT NULL); + CREATE TABLE inbox_events (event_id TEXT PRIMARY KEY, operation_id TEXT NOT NULL REFERENCES operations(operation_id), payload BLOB NOT NULL, received_at INTEGER NOT NULL); + CREATE TABLE file_commit_phases (commit_id TEXT PRIMARY KEY, operation_id TEXT NOT NULL, phase TEXT NOT NULL DEFAULT 'intent', agents_stage_path TEXT NOT NULL, teams_stage_path TEXT NOT NULL, created_at INTEGER NOT NULL, updated_at INTEGER NOT NULL); + PRAGMA user_version = 2; + ").unwrap(); + conn +} + +/// Fix B: upgrade from v2 (no hash columns, no d-tag) adds all three columns +/// and stamps version 4. +#[test] +fn test_schema_migration_from_v2_adds_all_columns() { + let conn = v2_journal(); + apply_journal_schema_pub(&conn).unwrap(); + // All three columns must exist after migration. + let fcp_cols: Vec = { + let mut stmt = conn + .prepare("PRAGMA table_info(file_commit_phases)") + .unwrap(); + stmt.query_map([], |r| r.get::<_, String>(1)) + .unwrap() + .collect::, _>>() + .unwrap() + }; + assert!(fcp_cols.contains(&"agents_content_hash".to_string())); + assert!(fcp_cols.contains(&"teams_content_hash".to_string())); + let oe_cols: Vec = { + let mut stmt = conn.prepare("PRAGMA table_info(outbox_events)").unwrap(); + stmt.query_map([], |r| r.get::<_, String>(1)) + .unwrap() + .collect::, _>>() + .unwrap() + }; + assert!(oe_cols.contains(&"retention_d_tag".to_string())); + let ver: u32 = conn + .pragma_query_value(None, "user_version", |r| r.get(0)) + .unwrap(); + assert_eq!(ver, 4); +} + +/// Fix B: upgrade from a partially-migrated v3 (hash columns present, d-tag +/// absent) adds only `retention_d_tag` and stamps version 4. +#[test] +fn test_schema_migration_from_partial_v3_adds_missing_d_tag() { + let conn = v2_journal(); + // Simulate partial v3: add hash columns but not d-tag. + conn.execute_batch( + " + ALTER TABLE file_commit_phases ADD COLUMN agents_content_hash TEXT NOT NULL DEFAULT ''; + ALTER TABLE file_commit_phases ADD COLUMN teams_content_hash TEXT NOT NULL DEFAULT ''; + PRAGMA user_version = 3; + ", + ) + .unwrap(); + apply_journal_schema_pub(&conn).unwrap(); + let oe_cols: Vec = { + let mut stmt = conn.prepare("PRAGMA table_info(outbox_events)").unwrap(); + stmt.query_map([], |r| r.get::<_, String>(1)) + .unwrap() + .collect::, _>>() + .unwrap() + }; + assert!( + oe_cols.contains(&"retention_d_tag".to_string()), + "d-tag must be added from partial v3" + ); + let ver: u32 = conn + .pragma_query_value(None, "user_version", |r| r.get(0)) + .unwrap(); + assert_eq!(ver, 4); +} + +/// Fix B: if migration verification fails (column absent after ALTER TABLE due +/// to a simulated failure), the version must NOT have been advanced to 4. +/// We simulate this by using a read-only connection that rejects schema changes. +/// Since SQLite in-memory DBs can't be made read-only, we verify the property +/// by opening a fresh v2 DB, running migration, then confirming re-running on +/// an already-at-4 DB is idempotent (no error, version stays 4). +#[test] +fn test_schema_migration_idempotent_on_v4() { + let conn = v2_journal(); + // First migration: v2 → v4. + apply_journal_schema_pub(&conn).unwrap(); + let ver: u32 = conn + .pragma_query_value(None, "user_version", |r| r.get(0)) + .unwrap(); + assert_eq!(ver, 4); + // Second call (idempotent): must succeed and leave version at 4. + apply_journal_schema_pub(&conn).unwrap(); + let ver2: u32 = conn + .pragma_query_value(None, "user_version", |r| r.get(0)) + .unwrap(); + assert_eq!(ver2, 4); +} diff --git a/desktop/src-tauri/src/managed_agents/store_journal_tests.rs b/desktop/src-tauri/src/managed_agents/store_journal_tests.rs index 8603f4b463..28d44644b8 100644 --- a/desktop/src-tauri/src/managed_agents/store_journal_tests.rs +++ b/desktop/src-tauri/src/managed_agents/store_journal_tests.rs @@ -917,12 +917,10 @@ fn insert_file_commit_phase_row(conn: &Connection, cid: &str, phase: &str, as_: } /// Three crash scenarios for the two-phase file-commit protocol (intent, -/// first_renamed+pending, first_renamed+done), driven by `run_boot_recovery_at`. +/// first_renamed+pending, first_renamed+done), driven by file commit recovery. #[test] #[allow(clippy::type_complexity)] fn test_crash_recovery_file_commit_phases() { - use super::run_boot_recovery_at; - // (phase, commit_id, [a_stage,t_stage,a_can,t_can] pre-exist, a_can post, t_can post, pubkey) let cases: &[(&str, &str, [bool; 4], bool, bool, &str)] = &[ ( @@ -977,7 +975,7 @@ fn test_crash_recovery_file_commit_phases() { t_stage.to_str().unwrap(), ); drop(j); - run_boot_recovery_at(&anchor, None).unwrap(); + super::file_commit_recovery_at_pub(&anchor).unwrap(); assert_eq!(a_can.exists(), *a_exists, "{pk} agents_can"); assert_eq!(t_can.exists(), *t_exists, "{pk} teams_can"); if *a_exists {