mirror of
https://github.com/block/buzz.git
synced 2026-11-05 08:51:10 +00:00
fix(b1-store): round-9 — single-owner file recovery, verifiable migration
Fix A: single-owner file recovery ordering and seam test. - lib.rs: run_file_commit_recovery() now runs BEFORE spawn_boot_recovery() is spawned, establishing synchronous pre-admission file repair. - txn.rs: run_boot_recovery_at() (the background publication task) no longer calls recover_interrupted_file_commits — it is publication-only. File repair has exactly one owner: the synchronous run_file_commit_recovery call. - txn.rs: add file_commit_recovery_at_pub() (cfg(test)) for testing the sync path without AppHandle. - Two seam tests in store_journal_fix_tests.rs: test_background_recovery_does_not_repair_files: proves run_boot_recovery_at leaves stage files and phase rows untouched. test_sync_file_recovery_repairs_dangling_intent_row: proves file_commit_recovery_at_pub renames stages and advances phase to committed. Fix B: verifiable schema migration. - schema.rs: replace let _ = conn.execute_batch() with PRAGMA table_info-based column inspection, ADD COLUMN only for missing columns, errors propagated, user_version stamped inside the same transaction as the migration. Post-commit verification ensures required columns are present. - New table_column_names() helper for column introspection. - Three migration tests in store_journal_fix_tests.rs: test_schema_migration_from_v2_adds_all_columns: v2 DB gets all three columns. test_schema_migration_from_partial_v3_adds_missing_d_tag: partial v3 gets d-tag. test_schema_migration_idempotent_on_v4: re-running on v4 is a no-op. Also update existing file-commit tests to use file_commit_recovery_at_pub (store_journal_tests.rs crash test, store_journal_fix_tests.rs hash-verify tests) since run_boot_recovery_at no longer performs file repair. Co-authored-by: Will Pfleger <pfleger.will@gmail.com> Signed-off-by: Will Pfleger <pfleger.will@gmail.com>
This commit is contained in:
npub1mn7jgtj4w2pd0g0zeuhxsa6jy6p0rewxz4kujt98my82ahfmp72sxjexk7
parent
462f4b8554
commit
8ac991e0ef
@@ -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![
|
||||
|
||||
@@ -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;
|
||||
|
||||
|
||||
@@ -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<Vec<String>, 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)
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
|
||||
@@ -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<String> = {
|
||||
let mut stmt = conn
|
||||
.prepare("PRAGMA table_info(file_commit_phases)")
|
||||
.unwrap();
|
||||
stmt.query_map([], |r| r.get::<_, String>(1))
|
||||
.unwrap()
|
||||
.collect::<Result<Vec<_>, _>>()
|
||||
.unwrap()
|
||||
};
|
||||
assert!(fcp_cols.contains(&"agents_content_hash".to_string()));
|
||||
assert!(fcp_cols.contains(&"teams_content_hash".to_string()));
|
||||
let oe_cols: Vec<String> = {
|
||||
let mut stmt = conn.prepare("PRAGMA table_info(outbox_events)").unwrap();
|
||||
stmt.query_map([], |r| r.get::<_, String>(1))
|
||||
.unwrap()
|
||||
.collect::<Result<Vec<_>, _>>()
|
||||
.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<String> = {
|
||||
let mut stmt = conn.prepare("PRAGMA table_info(outbox_events)").unwrap();
|
||||
stmt.query_map([], |r| r.get::<_, String>(1))
|
||||
.unwrap()
|
||||
.collect::<Result<Vec<_>, _>>()
|
||||
.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);
|
||||
}
|
||||
|
||||
@@ -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 {
|
||||
|
||||
Reference in New Issue
Block a user