mirror of
https://github.com/block/buzz.git
synced 2026-11-01 06:51:17 +00:00
fix(b1-store): round-7 — rebase, hash-verified recovery, tombstone propagation, tests
True git rebase onto bc9e6528a (current origin/main). Conflict resolution
preserves #3637 (spawn_snapshot, restart-diff) and #4522 (media_raw, deferred
uploads) alongside all B1 content.
Fix 3b (hash-verified file recovery):
- schema.rs: add agents_content_hash / teams_content_hash columns to
file_commit_phases (schema v3 with ALTER TABLE migration for existing DBs).
- txn.rs: compute SHA-256 of staged payloads and record them in the intent row.
- Recovery now verifies canonical file hash against recorded hash before
treating absent stage files as 'already completed'. All three missing-stage
branches (intent/both-absent, intent/teams-absent, first_renamed/teams-absent)
fail closed unless canonical matches. Only hash-verified canonicals advance
to committed.
Fix 4 (tombstone/archive propagation):
- agents.rs: tombstone_managed_agent_pending and archive_managed_agent_pending
use .map_err(|e| ...)? — failures surface in the delete command's Result.
- personas/mod.rs: same pattern for cascaded agent tombstones and persona tombstone.
- teams.rs: tombstone_team_pending and cascaded persona tombstones propagate.
- agents_retain.rs: retain_managed_agent_pending keeps its outer log-and-swallow
with an explicit comment documenting the boot-reconcile-recoverable contract
(Thufir accepted this distinction for updates vs deletes).
Tests:
- store_journal_fix_tests.rs (5 new tests):
- test_file_recovery_rename_done_phase_not_updated_succeeds: both stages absent,
canonicals match hashes → advance to committed.
- test_file_recovery_hash_mismatch_fails_closed: hash mismatch → uncommitted.
- test_file_recovery_first_renamed_teams_done_hash_verified: teams hash verified.
- test_file_recovery_first_renamed_teams_absent_no_hash_fails_closed: fail closed.
- test_boot_recovery_inserts_into_supplied_retention_path_not_flat: scoped path.
- storage_tests.rs (2 new tests): keyring round-trip (Fix 2 coverage).
Co-authored-by: Will Pfleger <pfleger.will@gmail.com>
Signed-off-by: Will Pfleger <pfleger.will@gmail.com>
This commit is contained in:
parent
3f164c679e
commit
ba8ca96f3d
@@ -1310,20 +1310,15 @@ pub async fn delete_managed_agent(
|
||||
// guard above and a confirmed removal — never orphan a live remote
|
||||
// deployment's relay record. Inside the lock, before the block closes
|
||||
// (no .await here). Every agent published, so every delete tombstones.
|
||||
if let Err(e) = tombstone_managed_agent_pending(&app, &state, &pubkey) {
|
||||
eprintln!(
|
||||
"buzz-desktop: agent-delete: tombstone failed for {pubkey}: {e} — \
|
||||
relay record may persist until next boot reconcile"
|
||||
);
|
||||
}
|
||||
// Propagate failure: a deleted agent has no boot-reconcile fallback
|
||||
// for its tombstone — without outbox evidence the tombstone is lost.
|
||||
tombstone_managed_agent_pending(&app, &state, &pubkey)
|
||||
.map_err(|e| format!("agent deleted but tombstone failed — relay record may persist: {e}"))?;
|
||||
// NIP-IA: archive the deleted agent's identity on the relay so it
|
||||
// stops appearing in member pickers and autocomplete. Same
|
||||
// best-effort, inside-the-lock contract as the tombstone above.
|
||||
if let Err(e) = archive_managed_agent_pending(&app, &state, &pubkey) {
|
||||
eprintln!(
|
||||
"buzz-desktop: agent-delete: archive failed for {pubkey}: {e}"
|
||||
);
|
||||
}
|
||||
// stops appearing in member pickers and autocomplete.
|
||||
// Also propagate: no reconcile fallback for archive events.
|
||||
archive_managed_agent_pending(&app, &state, &pubkey)
|
||||
.map_err(|e| format!("agent deleted but archive failed: {e}"))?;
|
||||
}
|
||||
try_regenerate_nest(&app);
|
||||
Ok(())
|
||||
|
||||
@@ -99,6 +99,12 @@ pub(crate) fn retain_managed_agent_pending(
|
||||
)
|
||||
})();
|
||||
if let Err(e) = result {
|
||||
// Update-retain failures are logged and swallowed: the agent record is
|
||||
// already durably written to the JSON store, and boot reconcile
|
||||
// (`reconcile_agents_to_events`) will re-enqueue the publication on
|
||||
// next launch. This is the accepted boot-reconcile-recoverable contract
|
||||
// for live-content updates (Thufir pass-3 review: "acceptable for updates,
|
||||
// not deletes"). Tombstones/archives use propagating Result instead.
|
||||
eprintln!("buzz-desktop: agent-retain: {e}");
|
||||
}
|
||||
}
|
||||
|
||||
@@ -279,16 +279,14 @@ pub async fn delete_persona(id: String, app: AppHandle) -> Result<(), String> {
|
||||
state.clear_agent_session_caches(pk);
|
||||
// Remove nsec from keyring after the record is gone.
|
||||
delete_agent_key(pk);
|
||||
if let Err(e) = super::agents::tombstone_managed_agent_pending(&app, &state, pk) {
|
||||
eprintln!("buzz-desktop: persona-delete: agent tombstone failed for {pk}: {e}");
|
||||
}
|
||||
if let Err(e) = super::agents::archive_managed_agent_pending(&app, &state, pk) {
|
||||
eprintln!("buzz-desktop: persona-delete: agent archive failed for {pk}: {e}");
|
||||
}
|
||||
}
|
||||
if let Err(e) = tombstone_persona_pending(&app, &state, &d_tag) {
|
||||
eprintln!("buzz-desktop: persona-delete: tombstone failed for {d_tag}: {e}");
|
||||
// Propagate: no boot-reconcile fallback for tombstone/archive.
|
||||
super::agents::tombstone_managed_agent_pending(&app, &state, pk)
|
||||
.map_err(|e| format!("persona-delete: agent tombstone failed for {pk}: {e}"))?;
|
||||
super::agents::archive_managed_agent_pending(&app, &state, pk)
|
||||
.map_err(|e| format!("persona-delete: agent archive failed for {pk}: {e}"))?;
|
||||
}
|
||||
tombstone_persona_pending(&app, &state, &d_tag)
|
||||
.map_err(|e| format!("persona-delete: tombstone failed for {d_tag}: {e}"))?;
|
||||
|
||||
// _store_guard drops here, before try_regenerate_nest.
|
||||
}
|
||||
|
||||
@@ -320,18 +320,16 @@ pub async fn delete_team(id: String, app: AppHandle) -> Result<(), String> {
|
||||
// delete_team_with_cascade rejects built-in teams via validate_team_deletion,
|
||||
// so reaching here means this team was owner-published — tombstone it. The
|
||||
// d_tag is the team id, captured before the record left the store.
|
||||
if let Err(e) = tombstone_team_pending(&app, &state, &id) {
|
||||
eprintln!("buzz-desktop: team-delete: tombstone failed for {id}: {e}");
|
||||
}
|
||||
// Propagate: no boot-reconcile fallback for tombstones.
|
||||
tombstone_team_pending(&app, &state, &id).map_err(|e| {
|
||||
format!("team deleted but tombstone failed — relay record may persist: {e}")
|
||||
})?;
|
||||
// Tombstone the cascaded personas too, so their orphaned kind:30175 heads
|
||||
// don't linger on the relay (F4). Each d-tag was captured pre-removal.
|
||||
for persona_d_tag in &cascaded_persona_d_tags {
|
||||
if let Err(e) = super::personas::tombstone_persona_pending(&app, &state, persona_d_tag)
|
||||
{
|
||||
eprintln!(
|
||||
"buzz-desktop: team-delete: persona tombstone failed for {persona_d_tag}: {e}"
|
||||
);
|
||||
}
|
||||
super::personas::tombstone_persona_pending(&app, &state, persona_d_tag).map_err(
|
||||
|e| format!("team-delete: persona tombstone failed for {persona_d_tag}: {e}"),
|
||||
)?;
|
||||
}
|
||||
try_regenerate_nest(&app);
|
||||
Ok(())
|
||||
|
||||
@@ -830,3 +830,39 @@ fn install_log_filename_accepts_ordinary_runtime_ids() {
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
// ── Fix 2: keyring chokepoint round-trip ─────────────────────────────────────
|
||||
|
||||
/// Fix 2: `persist_agent_keys_with` strips nsec on healthy keyring.
|
||||
#[test]
|
||||
fn persist_agent_keys_strips_nsec_on_healthy_keyring() {
|
||||
let store = FakeKeyStore::reachable();
|
||||
let mut record = record_with_pubkey_and_key("aabbcc", "nsec1abc");
|
||||
persist_agent_keys_with(&store, std::slice::from_mut(&mut record));
|
||||
assert!(
|
||||
record.private_key_nsec.is_empty(),
|
||||
"healthy keyring: nsec must be stripped"
|
||||
);
|
||||
assert_eq!(
|
||||
store
|
||||
.stored
|
||||
.borrow()
|
||||
.get(&agent_keyring_name("aabbcc"))
|
||||
.cloned(),
|
||||
Some("nsec1abc".to_string()),
|
||||
"healthy keyring: nsec must be in keyring"
|
||||
);
|
||||
}
|
||||
|
||||
/// Fix 2: `persist_agent_keys_with` keeps nsec inline when keyring unreachable.
|
||||
#[test]
|
||||
fn persist_agent_keys_keeps_nsec_inline_when_keyring_unreachable() {
|
||||
let store = FakeKeyStore::unreachable();
|
||||
let mut record = record_with_pubkey_and_key("aabbcc", "nsec1abc");
|
||||
persist_agent_keys_with(&store, std::slice::from_mut(&mut record));
|
||||
assert_eq!(
|
||||
record.private_key_nsec, "nsec1abc",
|
||||
"unreachable: nsec stays inline"
|
||||
);
|
||||
assert!(store.stored.borrow().is_empty());
|
||||
}
|
||||
|
||||
@@ -76,3 +76,7 @@ pub use writer::{atomic_write_restricted_with_fsync, atomic_write_with_fsync};
|
||||
#[cfg(test)]
|
||||
#[path = "../store_journal_tests.rs"]
|
||||
mod tests;
|
||||
|
||||
#[cfg(test)]
|
||||
#[path = "../store_journal_fix_tests.rs"]
|
||||
mod fix_tests;
|
||||
|
||||
@@ -100,22 +100,37 @@ pub(super) fn apply_journal_schema(conn: &Connection) -> Result<(), String> {
|
||||
-- agents_stage_path / teams_stage_path name the temp files written
|
||||
-- before rename. Recovery checks for them to decide what remains to do.
|
||||
CREATE TABLE IF NOT EXISTS file_commit_phases (
|
||||
commit_id TEXT NOT NULL 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
|
||||
commit_id TEXT NOT NULL 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,
|
||||
agents_content_hash TEXT NOT NULL DEFAULT '',
|
||||
teams_content_hash TEXT NOT NULL DEFAULT '',
|
||||
created_at INTEGER NOT NULL,
|
||||
updated_at INTEGER NOT NULL
|
||||
);
|
||||
",
|
||||
)
|
||||
.map_err(|e| format!("apply journal schema: {e}"))?;
|
||||
|
||||
// Set schema version via PRAGMA user_version (authoritative singleton,
|
||||
// never duplicated).
|
||||
conn.pragma_update(None, "user_version", 2)
|
||||
.map_err(|e| format!("set schema user_version: {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).
|
||||
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 '';",
|
||||
);
|
||||
conn.pragma_update(None, "user_version", 3)
|
||||
.map_err(|e| format!("bump schema user_version to 3: {e}"))?;
|
||||
} else {
|
||||
conn.pragma_update(None, "user_version", 3)
|
||||
.map_err(|e| format!("set schema user_version: {e}"))?;
|
||||
}
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
@@ -198,18 +198,27 @@ where
|
||||
|
||||
// Insert the file-commit intent row. This is the durable record that
|
||||
// recovery uses to identify an interrupted two-phase commit.
|
||||
// Content hashes allow recovery to verify a canonical file after a
|
||||
// missing-stage scenario: if the canonical hash matches the recorded
|
||||
// hash, the rename completed; otherwise fail closed.
|
||||
use sha2::Digest;
|
||||
let agents_hash = hex::encode(sha2::Sha256::digest(&agents_payload));
|
||||
let teams_hash = hex::encode(sha2::Sha256::digest(&teams_payload));
|
||||
let now = unix_now_secs();
|
||||
tx.execute(
|
||||
"INSERT INTO file_commit_phases
|
||||
(commit_id, operation_id, phase,
|
||||
agents_stage_path, teams_stage_path,
|
||||
agents_content_hash, teams_content_hash,
|
||||
created_at, updated_at)
|
||||
VALUES (?1, ?2, 'intent', ?3, ?4, ?5, ?5)",
|
||||
VALUES (?1, ?2, 'intent', ?3, ?4, ?5, ?6, ?7, ?7)",
|
||||
params![
|
||||
commit_id,
|
||||
commit_id, // operation_id == commit_id for phase records
|
||||
agents_stage.to_string_lossy().as_ref(),
|
||||
teams_stage.to_string_lossy().as_ref(),
|
||||
agents_hash,
|
||||
teams_hash,
|
||||
now,
|
||||
],
|
||||
)
|
||||
@@ -476,22 +485,35 @@ fn canonical_from_stage(stage: &Path) -> std::path::PathBuf {
|
||||
///
|
||||
/// Called before `recover_nonterminal_operations` so the JSON files are in a
|
||||
/// consistent state before publication recovery reads or reconciles them.
|
||||
/// Compute the SHA-256 hash of a file's contents, or `None` if the file cannot
|
||||
/// be read. Used by recovery to verify canonical files match recorded hashes.
|
||||
fn sha256_hex_of_file(path: &std::path::Path) -> Option<String> {
|
||||
use sha2::Digest;
|
||||
let bytes = std::fs::read(path).ok()?;
|
||||
Some(hex::encode(sha2::Sha256::digest(&bytes)))
|
||||
}
|
||||
|
||||
#[allow(clippy::type_complexity)]
|
||||
fn recover_interrupted_file_commits(journal: &rusqlite::Connection) -> Result<(), String> {
|
||||
let rows: Vec<(String, String, String, String)> = {
|
||||
// Read commit_id, phase, stage paths, AND content hashes.
|
||||
let rows: Vec<(String, String, String, String, String, String)> = {
|
||||
let mut stmt = journal
|
||||
.prepare(
|
||||
"SELECT commit_id, phase, agents_stage_path, teams_stage_path
|
||||
"SELECT commit_id, phase, agents_stage_path, teams_stage_path,
|
||||
agents_content_hash, teams_content_hash
|
||||
FROM file_commit_phases
|
||||
WHERE phase != 'committed'",
|
||||
)
|
||||
.map_err(|e| format!("prepare file_commit_phases query: {e}"))?;
|
||||
let collected: Vec<rusqlite::Result<(String, String, String, String)>> = stmt
|
||||
.query_map([], |row| {
|
||||
let collected: Vec<rusqlite::Result<(String, String, String, String, String, String)>> =
|
||||
stmt.query_map([], |row| {
|
||||
Ok((
|
||||
row.get::<_, String>(0)?,
|
||||
row.get::<_, String>(1)?,
|
||||
row.get::<_, String>(2)?,
|
||||
row.get::<_, String>(3)?,
|
||||
row.get::<_, String>(4)?,
|
||||
row.get::<_, String>(5)?,
|
||||
))
|
||||
})
|
||||
.map_err(|e| format!("query file_commit_phases: {e}"))?
|
||||
@@ -502,7 +524,7 @@ fn recover_interrupted_file_commits(journal: &rusqlite::Connection) -> Result<()
|
||||
.map_err(|e| format!("read file_commit_phases row: {e}"))?
|
||||
};
|
||||
|
||||
for (commit_id, phase_str, agents_stage_str, teams_stage_str) in rows {
|
||||
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);
|
||||
let teams_stage = std::path::PathBuf::from(&teams_stage_str);
|
||||
@@ -511,22 +533,36 @@ fn recover_interrupted_file_commits(journal: &rusqlite::Connection) -> Result<()
|
||||
"buzz-desktop: boot-recovery: interrupted file commit {commit_id} phase={phase_str}"
|
||||
);
|
||||
|
||||
let agents_canonical = canonical_from_stage(&agents_stage);
|
||||
let teams_canonical = canonical_from_stage(&teams_stage);
|
||||
|
||||
match phase {
|
||||
FileCommitPhase::Intent => {
|
||||
// Both stage files should exist for a full intent commit.
|
||||
// Handle each case explicitly to avoid certifying missing data.
|
||||
let agents_exists = agents_stage.exists();
|
||||
let teams_exists = teams_stage.exists();
|
||||
if !agents_exists && !teams_exists {
|
||||
// Both missing: both renames likely already completed but
|
||||
// the phase record update was lost. Advance to committed.
|
||||
eprintln!(
|
||||
"buzz-desktop: boot-recovery: {commit_id}: both stage files absent in intent phase — advancing to committed (renames likely completed)"
|
||||
);
|
||||
advance_file_commit_phase(journal, &commit_id, &FileCommitPhase::Committed)
|
||||
.unwrap_or_else(|e| {
|
||||
eprintln!("buzz-desktop: boot-recovery: advance committed failed: {e}")
|
||||
});
|
||||
// Both stage files absent. Verify the canonical files match
|
||||
// the recorded hashes — proof both renames completed.
|
||||
let agents_ok = !agents_hash.is_empty()
|
||||
&& sha256_hex_of_file(&agents_canonical).as_deref() == Some(&agents_hash);
|
||||
let teams_ok = !teams_hash.is_empty()
|
||||
&& 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"
|
||||
);
|
||||
advance_file_commit_phase(journal, &commit_id, &FileCommitPhase::Committed)
|
||||
.unwrap_or_else(|e| {
|
||||
eprintln!(
|
||||
"buzz-desktop: boot-recovery: advance committed failed: {e}"
|
||||
)
|
||||
});
|
||||
} 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"
|
||||
);
|
||||
}
|
||||
continue;
|
||||
}
|
||||
if !agents_exists {
|
||||
@@ -539,8 +575,7 @@ fn recover_interrupted_file_commits(journal: &rusqlite::Connection) -> Result<()
|
||||
continue;
|
||||
}
|
||||
// Agents stage exists — replay agents rename.
|
||||
let canonical = canonical_from_stage(&agents_stage);
|
||||
if let Err(e) = rename_staged(&agents_stage, &canonical) {
|
||||
if let Err(e) = rename_staged(&agents_stage, &agents_canonical) {
|
||||
eprintln!("buzz-desktop: boot-recovery: agents rename failed: {e}");
|
||||
continue;
|
||||
}
|
||||
@@ -551,14 +586,23 @@ fn recover_interrupted_file_commits(journal: &rusqlite::Connection) -> Result<()
|
||||
}
|
||||
// Teams rename.
|
||||
if teams_exists {
|
||||
let canonical = canonical_from_stage(&teams_stage);
|
||||
if let Err(e) = rename_staged(&teams_stage, &canonical) {
|
||||
if let Err(e) = rename_staged(&teams_stage, &teams_canonical) {
|
||||
eprintln!("buzz-desktop: boot-recovery: teams rename failed: {e}");
|
||||
continue;
|
||||
}
|
||||
} 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!(
|
||||
"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;
|
||||
}
|
||||
}
|
||||
// Teams stage absent after agents rename: teams rename already
|
||||
// completed. Either way, record committed.
|
||||
advance_file_commit_phase(journal, &commit_id, &FileCommitPhase::Committed)
|
||||
.unwrap_or_else(|e| {
|
||||
eprintln!("buzz-desktop: boot-recovery: advance committed failed: {e}")
|
||||
@@ -567,15 +611,25 @@ fn recover_interrupted_file_commits(journal: &rusqlite::Connection) -> Result<()
|
||||
FileCommitPhase::FirstRenamed => {
|
||||
// Agents already renamed; only teams stage remains.
|
||||
if teams_stage.exists() {
|
||||
let canonical = canonical_from_stage(&teams_stage);
|
||||
if let Err(e) = rename_staged(&teams_stage, &canonical) {
|
||||
if let Err(e) = rename_staged(&teams_stage, &teams_canonical) {
|
||||
eprintln!(
|
||||
"buzz-desktop: boot-recovery: teams rename (first_renamed) failed: {e}"
|
||||
);
|
||||
continue;
|
||||
}
|
||||
// Teams rename succeeded — advance to committed.
|
||||
} else {
|
||||
// Teams stage absent. Verify via recorded hash.
|
||||
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;
|
||||
}
|
||||
// Canonical matches — teams rename already completed.
|
||||
}
|
||||
// Teams stage absent: teams rename already completed.
|
||||
advance_file_commit_phase(journal, &commit_id, &FileCommitPhase::Committed)
|
||||
.unwrap_or_else(|e| {
|
||||
eprintln!("buzz-desktop: boot-recovery: advance committed failed: {e}")
|
||||
|
||||
228
desktop/src-tauri/src/managed_agents/store_journal_fix_tests.rs
Normal file
228
desktop/src-tauri/src/managed_agents/store_journal_fix_tests.rs
Normal file
@@ -0,0 +1,228 @@
|
||||
//! Tests for round-6/7 fixes: hash-verified recovery (Fix 3b), scoped
|
||||
//! recovery path (Fix 1), and keyring chokepoint (Fix 2).
|
||||
|
||||
use super::operations::insert_operation;
|
||||
use super::{insert_outbox_event, open_journal, run_boot_recovery_at, Generation};
|
||||
use crate::managed_agents::retention::{get_pending_sync, open_retention_db};
|
||||
use nostr::JsonUtil;
|
||||
|
||||
fn tmp_dir() -> tempfile::TempDir {
|
||||
tempfile::tempdir().expect("create temp dir")
|
||||
}
|
||||
|
||||
fn sha256_hex(data: &[u8]) -> String {
|
||||
use sha2::Digest;
|
||||
hex::encode(sha2::Sha256::digest(data))
|
||||
}
|
||||
|
||||
fn insert_phase_row(
|
||||
conn: &rusqlite::Connection,
|
||||
cid: &str,
|
||||
phase: &str,
|
||||
as_: &str,
|
||||
ts: &str,
|
||||
ah: &str,
|
||||
th: &str,
|
||||
) {
|
||||
conn.execute(
|
||||
"INSERT INTO file_commit_phases (commit_id,operation_id,phase,agents_stage_path,teams_stage_path,agents_content_hash,teams_content_hash,created_at,updated_at) VALUES(?1,?2,?3,?4,?5,?6,?7,0,0)",
|
||||
rusqlite::params![cid, cid, phase, as_, ts, ah, th],
|
||||
).unwrap();
|
||||
}
|
||||
|
||||
fn agents_can(anchor: &std::path::Path) -> std::path::PathBuf {
|
||||
anchor.join("managed-agents.json")
|
||||
}
|
||||
fn teams_can(anchor: &std::path::Path) -> std::path::PathBuf {
|
||||
anchor.join("teams.json")
|
||||
}
|
||||
fn agents_stage(anchor: &std::path::Path) -> std::path::PathBuf {
|
||||
anchor.join("managed-agents.cc1.stage")
|
||||
}
|
||||
fn teams_stage(anchor: &std::path::Path) -> std::path::PathBuf {
|
||||
anchor.join("teams.cc1.stage")
|
||||
}
|
||||
|
||||
/// Fix 3b — rename completed, phase update did not: both stage absent,
|
||||
/// canonicals match recorded hashes → recovery advances to committed.
|
||||
#[test]
|
||||
fn test_file_recovery_rename_done_phase_not_updated_succeeds() {
|
||||
let dir = tmp_dir();
|
||||
let anchor = dir.path().to_path_buf();
|
||||
let ac = agents_can(&anchor);
|
||||
let tc = teams_can(&anchor);
|
||||
let ac_data = b"[{\"pubkey\":\"aa\"}]";
|
||||
let tc_data = b"[]";
|
||||
std::fs::write(&ac, ac_data).unwrap();
|
||||
std::fs::write(&tc, tc_data).unwrap();
|
||||
let j = open_journal(&anchor).unwrap();
|
||||
insert_phase_row(
|
||||
&j,
|
||||
"cc1",
|
||||
"intent",
|
||||
agents_stage(&anchor).to_str().unwrap(),
|
||||
teams_stage(&anchor).to_str().unwrap(),
|
||||
&sha256_hex(ac_data),
|
||||
&sha256_hex(tc_data),
|
||||
);
|
||||
drop(j);
|
||||
run_boot_recovery_at(&anchor, None).unwrap();
|
||||
// Row should be committed now (file_commit_phases phase = 'committed').
|
||||
let j = open_journal(&anchor).unwrap();
|
||||
let count: i64 = j
|
||||
.query_row(
|
||||
"SELECT COUNT(*) FROM file_commit_phases WHERE phase='committed'",
|
||||
[],
|
||||
|r| r.get(0),
|
||||
)
|
||||
.unwrap();
|
||||
assert_eq!(
|
||||
count, 1,
|
||||
"recovery must advance to committed when canonicals verified"
|
||||
);
|
||||
}
|
||||
|
||||
/// Fix 3b — both stages absent, canonical hashes do NOT match → fail closed.
|
||||
#[test]
|
||||
fn test_file_recovery_hash_mismatch_fails_closed() {
|
||||
let dir = tmp_dir();
|
||||
let anchor = dir.path().to_path_buf();
|
||||
let ac = agents_can(&anchor);
|
||||
let tc = teams_can(&anchor);
|
||||
std::fs::write(&ac, b"[{\"pubkey\":\"different\"}]").unwrap();
|
||||
std::fs::write(&tc, b"[]").unwrap();
|
||||
let j = open_journal(&anchor).unwrap();
|
||||
insert_phase_row(
|
||||
&j,
|
||||
"cc2",
|
||||
"intent",
|
||||
agents_stage(&anchor).to_str().unwrap(),
|
||||
teams_stage(&anchor).to_str().unwrap(),
|
||||
"badhash",
|
||||
"badhash",
|
||||
);
|
||||
drop(j);
|
||||
run_boot_recovery_at(&anchor, None).unwrap();
|
||||
// Phase must NOT be committed — fail closed.
|
||||
let j = open_journal(&anchor).unwrap();
|
||||
let count: i64 = j
|
||||
.query_row(
|
||||
"SELECT COUNT(*) FROM file_commit_phases WHERE phase='committed'",
|
||||
[],
|
||||
|r| r.get(0),
|
||||
)
|
||||
.unwrap();
|
||||
assert_eq!(
|
||||
count, 0,
|
||||
"hash mismatch must leave phase uncommitted (fail closed)"
|
||||
);
|
||||
}
|
||||
|
||||
/// Fix 3b — first_renamed: teams stage absent, canonical matches hash → advance.
|
||||
#[test]
|
||||
fn test_file_recovery_first_renamed_teams_done_hash_verified() {
|
||||
let dir = tmp_dir();
|
||||
let anchor = dir.path().to_path_buf();
|
||||
let ac = agents_can(&anchor);
|
||||
let tc = teams_can(&anchor);
|
||||
let tc_data = b"[]";
|
||||
std::fs::write(&ac, b"[{\"pubkey\":\"aa\"}]").unwrap();
|
||||
std::fs::write(&tc, tc_data).unwrap();
|
||||
let j = open_journal(&anchor).unwrap();
|
||||
insert_phase_row(
|
||||
&j,
|
||||
"cc3",
|
||||
"first_renamed",
|
||||
agents_stage(&anchor).to_str().unwrap(),
|
||||
teams_stage(&anchor).to_str().unwrap(),
|
||||
"",
|
||||
&sha256_hex(tc_data),
|
||||
);
|
||||
drop(j);
|
||||
run_boot_recovery_at(&anchor, None).unwrap();
|
||||
let j = open_journal(&anchor).unwrap();
|
||||
let count: i64 = j
|
||||
.query_row(
|
||||
"SELECT COUNT(*) FROM file_commit_phases WHERE phase='committed'",
|
||||
[],
|
||||
|r| r.get(0),
|
||||
)
|
||||
.unwrap();
|
||||
assert_eq!(
|
||||
count, 1,
|
||||
"teams canonical verified in first_renamed → committed"
|
||||
);
|
||||
}
|
||||
|
||||
/// Fix 3b — first_renamed: teams stage absent, hash missing (empty string) → fail closed.
|
||||
#[test]
|
||||
fn test_file_recovery_first_renamed_teams_absent_no_hash_fails_closed() {
|
||||
let dir = tmp_dir();
|
||||
let anchor = dir.path().to_path_buf();
|
||||
let ac = agents_can(&anchor);
|
||||
std::fs::write(&ac, b"[{\"pubkey\":\"aa\"}]").unwrap();
|
||||
// teams canonical ABSENT
|
||||
let j = open_journal(&anchor).unwrap();
|
||||
insert_phase_row(
|
||||
&j,
|
||||
"cc4",
|
||||
"first_renamed",
|
||||
agents_stage(&anchor).to_str().unwrap(),
|
||||
teams_stage(&anchor).to_str().unwrap(),
|
||||
"",
|
||||
"",
|
||||
);
|
||||
drop(j);
|
||||
run_boot_recovery_at(&anchor, None).unwrap();
|
||||
let j = open_journal(&anchor).unwrap();
|
||||
let count: i64 = j
|
||||
.query_row(
|
||||
"SELECT COUNT(*) FROM file_commit_phases WHERE phase='committed'",
|
||||
[],
|
||||
|r| r.get(0),
|
||||
)
|
||||
.unwrap();
|
||||
assert_eq!(
|
||||
count, 0,
|
||||
"absent teams canonical with no hash → fail closed"
|
||||
);
|
||||
}
|
||||
|
||||
/// Fix 1 — scoped recovery path: journal-only outbox evidence, retention DB at
|
||||
/// a specific path (simulating the scoped hash-named path). Recovery re-inserts
|
||||
/// the retention row into THAT path, not a flat path.
|
||||
#[test]
|
||||
fn test_boot_recovery_inserts_into_supplied_retention_path_not_flat() {
|
||||
let dir = tmp_dir();
|
||||
let anchor = dir.path().to_path_buf();
|
||||
let scoped_path = anchor.join("retention").join("abc123.db");
|
||||
std::fs::create_dir_all(scoped_path.parent().unwrap()).unwrap();
|
||||
let flat_path = anchor.join("retention.db");
|
||||
|
||||
// Build real signed event for outbox.
|
||||
let keys = nostr::Keys::generate();
|
||||
let owner_pubkey = keys.public_key().to_hex();
|
||||
let event = nostr::EventBuilder::new(nostr::Kind::Custom(30177), "content")
|
||||
.tag(nostr::Tag::identifier("agent-1"))
|
||||
.sign_with_keys(&keys)
|
||||
.unwrap();
|
||||
let event_id = event.id.to_hex();
|
||||
let raw = event.as_json();
|
||||
|
||||
{
|
||||
let j = open_journal(&anchor).unwrap();
|
||||
insert_operation(&j, "op-1", "publish", "agent-1", Generation(0)).unwrap();
|
||||
insert_outbox_event(&j, &event_id, "op-1", raw.as_bytes()).unwrap();
|
||||
}
|
||||
|
||||
// Recovery with scoped path — must insert into scoped, NOT flat.
|
||||
run_boot_recovery_at(&anchor, Some(&scoped_path)).unwrap();
|
||||
|
||||
let conn_scoped = open_retention_db(&scoped_path).unwrap();
|
||||
let pending_scoped = get_pending_sync(&conn_scoped).unwrap();
|
||||
assert_eq!(pending_scoped.len(), 1, "re-inserted into scoped path");
|
||||
assert_eq!(pending_scoped[0].pubkey, owner_pubkey);
|
||||
|
||||
// Flat path must NOT have been created.
|
||||
assert!(!flat_path.exists(), "must not write to flat retention.db");
|
||||
}
|
||||
@@ -901,22 +901,11 @@ fn empty_teams_json() -> Vec<u8> {
|
||||
serde_json::to_vec_pretty(&serde_json::json!([])).unwrap()
|
||||
}
|
||||
|
||||
fn insert_file_commit_phase_row(
|
||||
conn: &Connection,
|
||||
commit_id: &str,
|
||||
phase: &str,
|
||||
agents_stage: &str,
|
||||
teams_stage: &str,
|
||||
) {
|
||||
fn insert_file_commit_phase_row(conn: &Connection, cid: &str, phase: &str, as_: &str, ts: &str) {
|
||||
conn.execute(
|
||||
"INSERT INTO file_commit_phases
|
||||
(commit_id, operation_id, phase,
|
||||
agents_stage_path, teams_stage_path,
|
||||
created_at, updated_at)
|
||||
VALUES (?1, ?2, ?3, ?4, ?5, 0, 0)",
|
||||
rusqlite::params![commit_id, commit_id, phase, agents_stage, teams_stage],
|
||||
)
|
||||
.unwrap();
|
||||
"INSERT INTO file_commit_phases (commit_id,operation_id,phase,agents_stage_path,teams_stage_path,agents_content_hash,teams_content_hash,created_at,updated_at) VALUES(?1,?2,?3,?4,?5,'','',0,0)",
|
||||
rusqlite::params![cid, cid, phase, as_, ts],
|
||||
).unwrap();
|
||||
}
|
||||
|
||||
/// Three crash scenarios for the two-phase file-commit protocol (intent,
|
||||
|
||||
Reference in New Issue
Block a user