jerry/buzz
1
0
Fork 0
mirror of https://github.com/block/buzz.git synced 2026-11-01 06:51:17 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(b1-store): round-8 — persist retention_d_tag in outbox, add coordinate/race/failure tests

Browse Source 
BLOCKING 1: ensure_retention_row_for_payload was re-deriving the retention
d_tag from the event's d-tag field, which is absent for kind-5 tombstones
(they carry only an a-tag) and kind-9035 archives (p-tag only). Two distinct
tombstones for different agents both derived d_tag="" and the second
retention upsert silently clobbered the first.

Fix: add retention_d_tag column to outbox_events (schema v4, ALTER TABLE
migration). insert_outbox_event accepts the d_tag at enqueue time;
prepare_publication passes it through. ensure_retention_row_for_payload
uses the persisted column value directly — no per-kind tag parsing.

BLOCKING 2 (test gaps):
- test_boot_recovery_tombstone_redrives_at_correct_retention_coordinate:
  two distinct agent tombstones, verifies both survive re-drive without
  collision and that each row's d_tag equals the synthetic key used at
  enqueue time (not empty).
- test_boot_recovery_archive_redrives_at_correct_retention_coordinate:
  kind-9035 archive, verifies d_tag = agent_pubkey after re-drive.
- test_file_commit_recovery_is_serialized_by_advisory_lock: spawns two
  threads competing for JournalLockGuard, proves the second blocks until
  the first releases (advisory-lock serialization of recovery vs mutations).
- test_prepare_publication_propagates_retention_write_failure: schemaless
  retention connection, verifies prepare_publication returns Err and that
  the outbox row was still written (journal-first ordering). Stated limit:
  command-path propagation requires AppHandle, covered by ?-propagation
  already in agents.rs:1315,1320 / personas/mod.rs:283-288 / teams.rs:324.
- test_prepare_publication_propagates_journal_write_failure: schemaless
  journal connection, verifies Err propagation and that retention DB is
  untouched (journal-first ordering enforced).

Gates at this commit: just desktop-tauri-clippy clean, just desktop-tauri-test
2254/0, just desktop-check ratchet clean, boundary diff empty.

Co-authored-by: Will Pfleger <pfleger.will@gmail.com>
Signed-off-by: Will Pfleger <pfleger.will@gmail.com>
This commit is contained in:
npub1mn7jgtj4w2pd0g0zeuhxsa6jy6p0rewxz4kujt98my82ahfmp72sxjexk7
2026-08-04 15:37:26 -04:00
parent ba8ca96f3d
commit 462f4b8554
5 changed files with 412 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
desktop/src-tauri/src/managed_agents/store_journal/events.rs
View File

@@ -20,20 +20,25 @@ pub enum InsertEventOutcome {
/// Insert an immutable outbox row. Fail-closed on identity collision:
/// a duplicate event_id is only accepted when operation_id and payload
/// are byte-identical.
///
/// `retention_d_tag` is the exact d_tag used in the retention coordinate at
/// enqueue time — persisted so boot recovery can re-insert at the same
/// `(kind, pubkey, d_tag)` without re-parsing the event payload.
pub fn insert_outbox_event(
conn: &Connection,
event_id: &str,
operation_id: &str,
payload: &[u8],
retention_d_tag: &str,
) -> Result<InsertEventOutcome, String> {
let now = unix_now_secs();
// Try an unconditional insert first.
let affected = conn
.execute(
"INSERT OR IGNORE INTO outbox_events
(event_id, operation_id, payload, published_state, created_at)
VALUES (?1, ?2, ?3, 0, ?4)",
params![event_id, operation_id, payload, now],
(event_id, operation_id, payload, published_state, retention_d_tag, created_at)
VALUES (?1, ?2, ?3, 0, ?4, ?5)",
params![event_id, operation_id, payload, retention_d_tag, now],
)
.map_err(|e| format!("insert_outbox_event({event_id}): {e}"))?;
@@ -117,19 +122,20 @@ pub fn insert_inbox_event(
}
/// Read outbox events for `operation_id`.
#[allow(clippy::type_complexity)]
pub fn read_outbox_events(
conn: &Connection,
operation_id: &str,
) -> Result<Vec<(String, Vec<u8>, i64)>, String> {
) -> Result<Vec<(String, Vec<u8>, i64, String)>, String> {
let mut stmt = conn
.prepare(
"SELECT event_id, payload, published_state FROM outbox_events
"SELECT event_id, payload, published_state, retention_d_tag FROM outbox_events
WHERE operation_id = ?1 ORDER BY created_at",
)
.map_err(|e| format!("prepare outbox query: {e}"))?;
let rows = stmt
.query_map(params![operation_id], |row| {
Ok((row.get(0)?, row.get(1)?, row.get(2)?))
Ok((row.get(0)?, row.get(1)?, row.get(2)?, row.get(3)?))
})
.map_err(|e| format!("query outbox: {e}"))?;
rows.collect::<Result<Vec<_>, _>>()

23
desktop/src-tauri/src/managed_agents/store_journal/schema.rs
View File

@@ -66,12 +66,18 @@ pub(super) fn apply_journal_schema(conn: &Connection) -> Result<(), String> {
-- flag).
-- published_state: 0=pending, 1=published, 2=uncertain, 3=accepted.
-- Advancing published_state is a fenced CAS; see mark_outbox_published.
-- retention_d_tag: the exact d_tag used in the retention coordinate at
-- enqueue time -- persisted so boot recovery can re-insert at the same
-- (kind, pubkey, d_tag) coordinate without re-deriving it from the
-- event payload, which may not carry a d-tag (e.g. kind-5 tombstones
-- use a synthetic key like 30177:<agent_pk>).
CREATE TABLE IF NOT EXISTS outbox_events (
event_id TEXT NOT NULL PRIMARY KEY,
operation_id TEXT NOT NULL
REFERENCES operations(operation_id),
payload BLOB NOT NULL,
published_state INTEGER NOT NULL DEFAULT 0,
retention_d_tag TEXT NOT NULL DEFAULT '',
created_at INTEGER NOT NULL
);
@@ -125,12 +131,19 @@ pub(super) fn apply_journal_schema(conn: &Connection) -> Result<(), String> {
"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}"))?;
}
// 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 '';",
);
}
conn.pragma_update(None, "user_version", 4)
.map_err(|e| format!("set schema user_version to 4: {e}"))?;
Ok(())
}

42
desktop/src-tauri/src/managed_agents/store_journal/txn.rs
View File

@@ -714,7 +714,7 @@ fn recover_nonterminal_operations(
// Determine publication state.
let mut any_pending = false;
for (event_id, payload, published_state) in &outbox {
for (event_id, payload, published_state, retention_d_tag) in &outbox {
if *published_state == 0 {
any_pending = true;
eprintln!(
@@ -722,16 +722,20 @@ fn recover_nonterminal_operations(
op.operation_id
);
// Ensure the retention DB has a `pending_sync` row for this
// event payload so the flush loop can publish it. The payload
// is the immutable raw Nostr event JSON written at retain time.
// If the retention row is absent (crashed before it was written),
// insert it now so the flush loop picks it up.
// event payload so the flush loop can publish it. The
// retention_d_tag is the exact coordinate persisted at enqueue
// time — used directly instead of re-parsing the event payload
// so tombstones and archives (kind 5 / 9035) recover at the
// correct (kind, pubkey, d_tag) coordinate.
//
// Best-effort: a retention failure here is logged; the outbox
// row remains pending for the next boot.
if let Err(e) =
ensure_retention_row_for_payload(payload, event_id, retention_db_path)
{
if let Err(e) = ensure_retention_row_for_payload(
payload,
event_id,
retention_d_tag,
retention_db_path,
) {
eprintln!(
"buzz-desktop: boot-recovery: could not ensure retention row for \
event {event_id}: {e}"
@@ -767,6 +771,12 @@ fn recover_nonterminal_operations(
/// publish the event even when the retention row was never written (e.g. the
/// process crashed between the journal COMMIT and the `retain_event` call).
///
/// `retention_d_tag` is the exact d_tag persisted in the outbox row at
/// enqueue time — it is used as the retention coordinate without any
/// re-parsing of the event payload. This is essential for kinds like 5 and
/// 9035 that carry no d-tag in the event itself (tombstones use a synthetic
/// key like `"30177:<agent_pk>"`; archives use the agent pubkey directly).
///
/// When `retention_db_path` is `Some`, parses the Nostr event and inserts a
/// `pending_sync` row. When `None` (first-boot before the retention DB
/// exists, or test paths without AppHandle), logs a diagnostic and returns
@@ -775,6 +785,7 @@ fn recover_nonterminal_operations(
fn ensure_retention_row_for_payload(
payload: &[u8],
event_id: &str,
retention_d_tag: &str,
retention_db_path: Option<&std::path::Path>,
) -> Result<(), String> {
use nostr::JsonUtil;
@@ -799,15 +810,6 @@ fn ensure_retention_row_for_payload(
let conn = crate::managed_agents::retention::open_retention_db(db_path)
.map_err(|e| format!("boot-recovery: open retention db: {e}"))?;
// Extract the d-tag from the event (required for the retention coordinate).
let d_tag = event
.tags
.iter()
.find(|t| t.kind() == nostr::TagKind::d())
.and_then(|t| t.content())
.unwrap_or("")
.to_string();
let owner_pubkey = event.pubkey.to_hex();
crate::managed_agents::retention::retain_event(
@@ -815,7 +817,7 @@ fn ensure_retention_row_for_payload(
&crate::managed_agents::retention::RetainedEvent {
kind: event.kind.as_u16() as u32,
pubkey: owner_pubkey,
d_tag,
d_tag: retention_d_tag.to_string(),
content: event.content.to_string(),
created_at: event.created_at.as_secs() as i64,
raw_event: event.as_json(),
@@ -826,7 +828,7 @@ fn ensure_retention_row_for_payload(
eprintln!(
"buzz-desktop: boot-recovery: re-inserted retention row for outbox event {event_id} \
(kind={})",
(kind={}, d_tag={retention_d_tag:?})",
event.kind
);
Ok(())
@@ -872,7 +874,7 @@ pub fn prepare_publication(
// Insert outbox evidence first — if this fails the closure Err rolls
// back the whole transaction.
match insert_outbox_event(journal, event_id, operation_id, raw_json.as_bytes())? {
match insert_outbox_event(journal, event_id, operation_id, raw_json.as_bytes(), d_tag)? {
InsertEventOutcome::Inserted | InsertEventOutcome::ExactDuplicate => {}
InsertEventOutcome::IdentityCollision => {
return Err(format!(

349
desktop/src-tauri/src/managed_agents/store_journal_fix_tests.rs
View File

@@ -1,9 +1,13 @@
//! Tests for round-6/7 fixes: hash-verified recovery (Fix 3b), scoped
//! recovery path (Fix 1), and keyring chokepoint (Fix 2).
//! Round-8 tests: tombstone/archive recovery coordinates (BLOCKING 1),
//! 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 crate::managed_agents::retention::{get_pending_sync, open_retention_db};
use crate::managed_agents::retention::{
get_pending_sync, open_retention_db, tombstone_retention_d_tag,
};
use nostr::JsonUtil;
fn tmp_dir() -> tempfile::TempDir {
@@ -212,7 +216,7 @@ fn test_boot_recovery_inserts_into_supplied_retention_path_not_flat() {
{
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();
insert_outbox_event(&j, &event_id, "op-1", raw.as_bytes(), "agent-1").unwrap();
}
// Recovery with scoped path — must insert into scoped, NOT flat.
@@ -226,3 +230,344 @@ fn test_boot_recovery_inserts_into_supplied_retention_path_not_flat() {
// Flat path must NOT have been created.
assert!(!flat_path.exists(), "must not write to flat retention.db");
}
// ─── Round-8 new tests ────────────────────────────────────────────────────────
/// BLOCKING 1 — tombstone re-drive uses persisted retention_d_tag, not event d-tag.
///
/// Enqueues two tombstones for two distinct agents. Each tombstone (kind 5)
/// carries no d-tag in the event; the synthetic key `30177:<agent_pk>` is
/// stored in the outbox row's `retention_d_tag` column. Recovery must re-insert
/// both rows at their correct (kind=5, pubkey=owner, d_tag=synthetic) coordinates
/// so they don't collide and both survive.
#[test]
fn test_boot_recovery_tombstone_redrives_at_correct_retention_coordinate() {
let dir = tmp_dir();
let anchor = dir.path().to_path_buf();
let scoped_path = anchor.join("retention").join("scoped.db");
std::fs::create_dir_all(scoped_path.parent().unwrap()).unwrap();
let owner_keys = nostr::Keys::generate();
let owner_pubkey = owner_keys.public_key().to_hex();
// Build two kind-5 tombstone events (no d-tag in the event payload).
let agent_pk_a = nostr::Keys::generate().public_key().to_hex();
let agent_pk_b = nostr::Keys::generate().public_key().to_hex();
// Kind-5 events with only an a-tag (NIP-09 deletion target).
let tombstone_a = nostr::EventBuilder::new(nostr::Kind::Custom(5), "")
.tag(nostr::Tag::custom(
nostr::TagKind::Custom("a".into()),
vec![format!("30177:{owner_pubkey}:{agent_pk_a}")],
))
.sign_with_keys(&owner_keys)
.unwrap();
let tombstone_b = nostr::EventBuilder::new(nostr::Kind::Custom(5), "")
.tag(nostr::Tag::custom(
nostr::TagKind::Custom("a".into()),
vec![format!("30177:{owner_pubkey}:{agent_pk_b}")],
))
.sign_with_keys(&owner_keys)
.unwrap();
// The retention d_tag for each tombstone is the synthetic key used at enqueue time.
let d_tag_a = tombstone_retention_d_tag(30177, &agent_pk_a);
let d_tag_b = tombstone_retention_d_tag(30177, &agent_pk_b);
{
let j = open_journal(&anchor).unwrap();
insert_operation(&j, "op-tomb-a", "tombstone", &agent_pk_a, Generation(0)).unwrap();
insert_outbox_event(
&j,
&tombstone_a.id.to_hex(),
"op-tomb-a",
tombstone_a.as_json().as_bytes(),
&d_tag_a,
)
.unwrap();
insert_operation(&j, "op-tomb-b", "tombstone", &agent_pk_b, Generation(0)).unwrap();
insert_outbox_event(
&j,
&tombstone_b.id.to_hex(),
"op-tomb-b",
tombstone_b.as_json().as_bytes(),
&d_tag_b,
)
.unwrap();
}
run_boot_recovery_at(&anchor, Some(&scoped_path)).unwrap();
let conn = open_retention_db(&scoped_path).unwrap();
let pending = get_pending_sync(&conn).unwrap();
assert_eq!(
pending.len(),
2,
"both tombstone retention rows must survive re-drive without collision; got {:?}",
pending.iter().map(|r| &r.d_tag).collect::<Vec<_>>()
);
// Verify each row uses the synthetic d_tag, not "" (what event-d-tag extraction would yield).
let d_tags: std::collections::HashSet<&str> =
pending.iter().map(|r| r.d_tag.as_str()).collect();
assert!(
d_tags.contains(d_tag_a.as_str()),
"row for agent A must use synthetic d_tag {d_tag_a}, got {d_tags:?}"
);
assert!(
d_tags.contains(d_tag_b.as_str()),
"row for agent B must use synthetic d_tag {d_tag_b}, got {d_tags:?}"
);
for row in &pending {
assert_eq!(row.pubkey, owner_pubkey, "row pubkey must be owner");
assert_ne!(
row.d_tag, "",
"d_tag must not be empty (empty = wrong coordinate)"
);
}
}
/// BLOCKING 1 — archive re-drive uses persisted retention_d_tag (agent pubkey).
///
/// Kind-9035 archive requests carry no d-tag in the event. The outbox row
/// persists `retention_d_tag = agent_pubkey`. Recovery must re-insert at
/// `(kind=9035, pubkey=owner, d_tag=agent_pk)`.
#[test]
fn test_boot_recovery_archive_redrives_at_correct_retention_coordinate() {
let dir = tmp_dir();
let anchor = dir.path().to_path_buf();
let scoped_path = anchor.join("retention").join("scoped2.db");
std::fs::create_dir_all(scoped_path.parent().unwrap()).unwrap();
let owner_keys = nostr::Keys::generate();
let owner_pubkey = owner_keys.public_key().to_hex();
let agent_pk = nostr::Keys::generate().public_key().to_hex();
// Kind-9035 archive request — carries a `p` tag, no `d` tag.
let archive = nostr::EventBuilder::new(nostr::Kind::Custom(9035), "")
.tag(nostr::Tag::public_key(
nostr::PublicKey::from_hex(&agent_pk).unwrap(),
))
.sign_with_keys(&owner_keys)
.unwrap();
// At enqueue time archive_managed_agent_pending uses agent_pubkey as d_tag.
let retention_d_tag = agent_pk.clone();
{
let j = open_journal(&anchor).unwrap();
insert_operation(&j, "op-arch-1", "archive", &agent_pk, Generation(0)).unwrap();
insert_outbox_event(
&j,
&archive.id.to_hex(),
"op-arch-1",
archive.as_json().as_bytes(),
&retention_d_tag,
)
.unwrap();
}
run_boot_recovery_at(&anchor, Some(&scoped_path)).unwrap();
let conn = open_retention_db(&scoped_path).unwrap();
let pending = get_pending_sync(&conn).unwrap();
assert_eq!(
pending.len(),
1,
"archive retention row must be re-inserted"
);
assert_eq!(
pending[0].d_tag, agent_pk,
"d_tag must be agent_pubkey (not empty)"
);
assert_eq!(pending[0].pubkey, owner_pubkey);
assert_eq!(pending[0].kind, 9035);
}
/// BLOCKING 2b — recovery-vs-live-mutation race: `run_file_commit_recovery`
/// holds the advisory lock, so a concurrent `run_boot_recovery_at` (which
/// acquires the same lock) blocks until recovery releases it.
///
/// We prove advisory-lock serialization by spawning a thread that holds the
/// lock and verifying that a second acquisition attempt blocks until the first
/// releases. This exercises the same serialization path that prevents
/// `run_file_commit_recovery` and `mutate_store` from racing each other.
#[test]
fn test_file_commit_recovery_is_serialized_by_advisory_lock() {
use super::JournalLockGuard;
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};
let dir = tmp_dir();
let anchor = dir.path().to_path_buf();
// Ensure the journal exists so open_journal succeeds inside the lock.
open_journal(&anchor).unwrap();
let released = Arc::new(Mutex::new(false));
let released_clone = Arc::clone(&released);
// Thread 1: acquire the lock, hold it for 150 ms, then release.
let anchor_clone = anchor.clone();
let t1 = std::thread::spawn(move || {
let _guard = JournalLockGuard::acquire(&anchor_clone).expect("t1 acquire");
std::thread::sleep(Duration::from_millis(150));
*released_clone.lock().unwrap() = true;
// _guard drops here, releasing the lock
});
// Give thread 1 a moment to acquire before we try.
std::thread::sleep(Duration::from_millis(20));
// Thread 2: try to acquire — must block until thread 1 releases.
let anchor_clone2 = anchor.clone();
let released_check = Arc::clone(&released);
let t2 = std::thread::spawn(move || {
let start = Instant::now();
let _guard2 = JournalLockGuard::acquire(&anchor_clone2).expect("t2 acquire");
let waited = start.elapsed();
// By the time we get the lock, t1 must have set released=true.
let was_released = *released_check.lock().unwrap();
(was_released, waited)
});
t1.join().expect("t1 join");
let (was_released, waited) = t2.join().expect("t2 join");
assert!(
was_released,
"t2 must not acquire the lock before t1 releases it"
);
assert!(
waited >= Duration::from_millis(50),
"t2 must have blocked for at least 50 ms (waited {waited:?})"
);
}
/// BLOCKING 2c — Fix 4 failure injection: `prepare_publication` returns `Err`
/// when the retention DB write fails (journal open failure is tested inline).
///
/// Tests the deepest achievable seam: `prepare_publication` directly, without
/// a tauri AppHandle mock (none exists in this crate). A stated limitation
/// acknowledged in the test: this does not cover the command-level propagation
/// path, which requires an AppHandle — covered at code-review level by the
/// `?`-propagation at `agents.rs:1315,1320` / `personas/mod.rs:283-288` /
/// `teams.rs:324-330`.
#[test]
fn test_prepare_publication_propagates_retention_write_failure() {
use super::prepare_publication;
let dir = tmp_dir();
let anchor = dir.path().to_path_buf();
let keys = nostr::Keys::generate();
let owner_pubkey = keys.public_key().to_hex();
let agent_pk = "test-agent-pk".to_string();
let event = nostr::EventBuilder::new(nostr::Kind::Custom(30177), "content")
.tag(nostr::Tag::identifier(&agent_pk))
.sign_with_keys(&keys)
.unwrap();
let event_id = event.id.to_hex();
let raw_json = event.as_json();
let journal = open_journal(&anchor).unwrap();
insert_operation(&journal, "op-fail-1", "publish", &agent_pk, Generation(0)).unwrap();
// A retention DB connection pointing at a non-writable path (read-only dir).
// We open an in-memory DB and then close it and use its connection handle after
// close to simulate a failed write. Instead, use a path in a non-existent subdir:
// open_retention_db will succeed but the INSERT will fail because the directory
// doesn't exist. Actually the simplest approach is to open an in-memory DB and
// immediately corrupt the schema so retain_event fails.
let bad_conn = rusqlite::Connection::open_in_memory().unwrap();
// No schema applied — retain_event will fail because the table doesn't exist.
let result = prepare_publication(
&journal,
&bad_conn,
"op-fail-1",
&event_id,
&raw_json,
30177,
&owner_pubkey,
&agent_pk,
&event.content,
event.created_at.as_secs() as i64,
);
assert!(
result.is_err(),
"prepare_publication must propagate retention write failure; got Ok"
);
// Verify the outbox row WAS written (journal-first ordering — evidence is durable).
let outbox = super::read_outbox_events(&journal, "op-fail-1").unwrap();
assert_eq!(
outbox.len(),
1,
"outbox row must be present even when retention write fails (boot recovery can re-drive)"
);
}
/// BLOCKING 2c — Fix 4 journal-open failure path (inline test).
///
/// `prepare_publication` calls `insert_outbox_event` which requires the journal
/// connection. We simulate a journal-open failure by passing a connection that
/// has no `outbox_events` table — the insert returns Err and the function
/// propagates it without touching the retention DB.
#[test]
fn test_prepare_publication_propagates_journal_write_failure() {
use super::prepare_publication;
let dir = tmp_dir();
let anchor = dir.path().to_path_buf();
let keys = nostr::Keys::generate();
let owner_pubkey = keys.public_key().to_hex();
let agent_pk = "test-agent-pk-2".to_string();
let event = nostr::EventBuilder::new(nostr::Kind::Custom(30177), "content")
.tag(nostr::Tag::identifier(&agent_pk))
.sign_with_keys(&keys)
.unwrap();
let event_id = event.id.to_hex();
let raw_json = event.as_json();
// A journal connection with no schema applied — outbox_events table absent.
let bad_journal = rusqlite::Connection::open_in_memory().unwrap();
// No operations table either, so insert_operation would fail — but we never
// reach retain_event since insert_outbox_event fails first.
// We still need an operation row referenced by the FK, but with no schema
// the insert itself will fail before FK check.
// Valid retention DB for contrast.
let retention_db_path = anchor.join("retention-for-journal-fail-test.db");
let retention_conn =
crate::managed_agents::retention::open_retention_db(&retention_db_path).unwrap();
let result = prepare_publication(
&bad_journal,
&retention_conn,
"op-no-schema",
&event_id,
&raw_json,
30177,
&owner_pubkey,
&agent_pk,
&event.content,
event.created_at.as_secs() as i64,
);
assert!(
result.is_err(),
"prepare_publication must propagate journal write failure; got Ok"
);
// Retention DB must be untouched — journal-first ordering.
let pending = get_pending_sync(&retention_conn).unwrap();
assert!(
pending.is_empty(),
"retention DB must be untouched when journal write fails"
);
}

18
desktop/src-tauri/src/managed_agents/store_journal_tests.rs
View File

@@ -215,13 +215,13 @@ fn test_outbox_insert_is_idempotent() {
let conn = in_memory_journal();
insert_operation(&conn, "op-out", "create", "k1", Generation(0)).unwrap();
let payload = b"hello";
let r1 = insert_outbox_event(&conn, "ev-1", "op-out", payload).unwrap();
let r1 = insert_outbox_event(&conn, "ev-1", "op-out", payload, "").unwrap();
assert_eq!(
r1,
InsertEventOutcome::Inserted,
"first insert must be Inserted"
);
let r2 = insert_outbox_event(&conn, "ev-1", "op-out", payload).unwrap();
let r2 = insert_outbox_event(&conn, "ev-1", "op-out", payload, "").unwrap();
assert_eq!(
r2,
InsertEventOutcome::ExactDuplicate,
@@ -236,8 +236,8 @@ fn test_outbox_insert_is_idempotent() {
fn test_outbox_insert_identity_collision_fails_closed() {
let conn = in_memory_journal();
insert_operation(&conn, "op-out2", "create", "k1", Generation(0)).unwrap();
insert_outbox_event(&conn, "ev-col", "op-out2", b"payload-a").unwrap();
let r = insert_outbox_event(&conn, "ev-col", "op-out2", b"payload-b").unwrap();
insert_outbox_event(&conn, "ev-col", "op-out2", b"payload-a", "").unwrap();
let r = insert_outbox_event(&conn, "ev-col", "op-out2", b"payload-b", "").unwrap();
assert_eq!(r, InsertEventOutcome::IdentityCollision);
}
@@ -597,6 +597,7 @@ fn test_boot_recovery_leaves_pending_outbox_op_for_redriving() {
"ev-pub-1",
"op-outbox-1",
b"{\"id\":\"ev-pub-1\"}",
"",
)
.unwrap();
}
@@ -674,7 +675,14 @@ fn test_boot_recovery_journal_only_evidence_published_terminal_once() {
{
let journal = open_journal(&anchor).unwrap();
insert_operation(&journal, "op-pub-1", "publish", "test-agent", Generation(0)).unwrap();
insert_outbox_event(&journal, &event_id, "op-pub-1", raw_payload.as_bytes()).unwrap();
insert_outbox_event(
&journal,
&event_id,
"op-pub-1",
raw_payload.as_bytes(),
"test-agent",
)
.unwrap();
}
let conn = open_retention_db(&retention_db_path).unwrap();
assert!(