From e10f613a5999cef9ff11d0789547bea24390d511 Mon Sep 17 00:00:00 2001 From: npub1mn7jgtj4w2pd0g0zeuhxsa6jy6p0rewxz4kujt98my82ahfmp72sxjexk7 Date: Thu, 30 Jul 2026 11:20:05 -0400 Subject: [PATCH] fix(desktop): defer inbound events against a pending retention row MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A pending row is durable local intent — an edit this device made and has not yet published. Resolving inbound events by timestamp let a strictly newer event clear `pending_sync` and patch `personas.json`, destroying that edit. Timestamps are not evidence of newer intent: the laundering vector this change targets re-signs stale content at `max(now, head + 1)`, so every laundered revert arrives strictly newer. That made the untrusted input the one that always won. The pending check now precedes every ordering rule, and the new `Deferred` outcome keeps "lost the compare" distinct from "awaiting arbitration" for the boot decision pass. Callers gate on `Applied` so neither non-apply outcome reaches disk. Until that pass lands a pending row shadows genuinely newer remote edits for its coordinate; flush normally clears it within seconds, and an edit that can still be reconciled is worth more than one that cannot be recovered. Co-authored-by: Will Pfleger Signed-off-by: Will Pfleger --- .../src/commands/personas/inbound.rs | 15 ++- .../src-tauri/src/managed_agents/retention.rs | 78 +++++++---- .../src/managed_agents/retention/tests.rs | 122 +++++++++++++++--- 3 files changed, 165 insertions(+), 50 deletions(-) diff --git a/desktop/src-tauri/src/commands/personas/inbound.rs b/desktop/src-tauri/src/commands/personas/inbound.rs index a2b9be7edd..846b0648e3 100644 --- a/desktop/src-tauri/src/commands/personas/inbound.rs +++ b/desktop/src-tauri/src/commands/personas/inbound.rs @@ -39,9 +39,9 @@ mod inbound_tests; /// /// The retention store decides whether the inbound event wins over a pending /// local edit (`retain_inbound_event`): `personas.json` is only patched when the -/// retain reports [`InboundOutcome::Applied`], so an equal-second collision with -/// a pending local edit leaves the local record — and its queued publish — -/// untouched. +/// retain reports [`InboundOutcome::Applied`], so an inbound event colliding +/// with a pending local edit — at any timestamp, newer included — leaves the +/// local record and its queued publish untouched. /// /// `arrival_relay_url` is the relay the calling subscription is bound to. The /// retention store this event belongs to is decided by the community that @@ -134,7 +134,7 @@ fn reconcile_inbound_persona_event_blocking( &conn, &RetainedEvent::inbound(kind, event.pubkey.to_hex(), d_tag.clone(), &event), )?; - if outcome == InboundOutcome::Skipped { + if outcome != InboundOutcome::Applied { return Ok(()); } @@ -250,8 +250,9 @@ fn reconcile_inbound_tombstone( .map_err(|error| error.to_string())?; // Resolve against the retained tombstone row (keyed by the target - // coordinate, F2c) so a re-received tombstone or one older than a pending - // local edit is a no-op. Scoped to the arrival community, so a workspace + // coordinate, F2c) so a re-received tombstone, one that loses the ordering + // compare, or one landing on a pending tombstone of its own is a no-op. + // Scoped to the arrival community, so a workspace // switch since arrival drops the tombstone instead of retaining it — and // deleting a record — in the wrong community's store. let Some(scope) = @@ -269,7 +270,7 @@ fn reconcile_inbound_tombstone( event, ), )?; - if outcome == InboundOutcome::Skipped { + if outcome != InboundOutcome::Applied { return Ok(()); } diff --git a/desktop/src-tauri/src/managed_agents/retention.rs b/desktop/src-tauri/src/managed_agents/retention.rs index f17082c0f1..9476b3a468 100644 --- a/desktop/src-tauri/src/managed_agents/retention.rs +++ b/desktop/src-tauri/src/managed_agents/retention.rs @@ -145,8 +145,11 @@ impl RetainedEvent { /// A row for a locally signed event that still has to reach the relay. /// /// Every event-derived field is read from the one event passed in, so a row - /// can never carry an `event_id` that describes different bytes than its own - /// `raw_event` — the ordering key and the payload cannot drift apart. + /// built this way cannot carry an `event_id` describing different bytes + /// than its own `raw_event`. The fields stay public for test construction, + /// so this is the rule every production writer follows rather than a type + /// guarantee: build rows through these constructors and the ordering key + /// and the payload cannot drift apart. pub fn pending(kind: u32, pubkey: String, d_tag: String, event: &nostr::Event) -> Self { Self::from_signed(kind, pubkey, d_tag, event, true) } @@ -303,15 +306,23 @@ pub fn retain_event(conn: &Connection, event: &RetainedEvent) -> Result<(), Stri /// Outcome of an inbound retain — whether the local store now reflects the /// inbound event, so the caller knows whether to patch `personas.json`. +/// +/// Only [`InboundOutcome::Applied`] authorizes a disk write. Both other +/// variants leave every local file untouched; they differ in what the +/// coordinate is waiting for. #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum InboundOutcome { - /// The inbound event was applied (no row, or it was strictly newer than a - /// non-conflicting local row). The caller patches the local record store. + /// The inbound event was applied (no row, or it beat a non-pending local + /// row). The caller patches the local record store. Applied, - /// The inbound event was NOT applied: either it is older than the retained - /// row, or it collides at the same `created_at` with a pending local edit. - /// The local record store is left untouched and the pending edit republishes. + /// The inbound event lost the ordering compare against a non-pending + /// retained row, or the pair could not be ordered at all. Nothing changes. Skipped, + /// The coordinate carries a pending local edit, so the inbound event is + /// not resolved here at any timestamp. The retained row, its content, and + /// its `pending_sync` flag are left exactly as they were, for the boot + /// decision pass to arbitrate against a writer-consistent head. + Deferred, } /// Retain an event arriving FROM the relay, resolving it against any local row. @@ -319,37 +330,58 @@ pub enum InboundOutcome { /// Inbound events are already on the relay, so they are retained with /// `pending_sync = 0`. The resolution is deliberately narrower than /// [`retain_event`]'s blind newer-or-equal upsert, which would clobber a -/// pending local edit's `pending_sync` flag and silently drop its publish: +/// pending local edit's `pending_sync` flag and silently drop its publish. /// -/// - No local row, or inbound strictly newer (`created_at >`): apply the -/// inbound event, clearing `pending_sync`. Inbound wins; a stale local edit -/// the relay already superseded stops republishing instead of looping. -/// - Equal `created_at`, local row NOT pending: order by event id, matching the +/// Resolution order — **pending is checked before any timestamp compare**: +/// +/// - Local row pending, at ANY inbound `created_at`: defer. Nothing is written +/// and the caller does not patch disk. See the invariant below. +/// - No local row, or inbound strictly newer (`created_at >`) than a +/// non-pending row: apply the inbound event. +/// - Equal `created_at`, non-pending row: order by event id, matching the /// relay's own NIP-33 comparator — lowest id wins (`buzz-db`'s /// `replace_parameterized_event` rejects an incoming event whose id is `>=` /// the accepted one at an equal timestamp). Without this the local cache can /// disagree with the relay about which of two same-second events is the head, /// and every subsequent compare against that head inherits the error. -/// - Equal `created_at`, local row pending: skip regardless of id. A pending row -/// is durable local intent, and intent is arbitrated by the boot decision -/// pass against a writer-consistent head — never dropped by an id compare -/// here. (A re-received echo at equal time is also a no-op.) -/// - Inbound older: skip — nothing to change. +/// - Inbound older than a non-pending row: skip — nothing to change. /// /// An unorderable compare is never resolved by guessing: if either side's /// `event_id` is `None` at an equal timestamp, the local row stands. +/// +/// # Why a newer timestamp does not beat pending +/// +/// `pending_sync = 1` is durable local intent: an edit the user made and this +/// device has not yet published. A newer `created_at` is not evidence that the +/// remote content is newer *intent* — the laundering vector this whole change +/// exists to close mints exactly such events. A stale store re-signs its old +/// content at `monotonic_created_at = max(now, head + 1)`, so every laundered +/// revert arrives strictly newer than whatever it is reverting. Letting +/// strictly-newer inbound clear the flag would therefore destroy an unpublished +/// edit on precisely the input the fix targets. +/// +/// Deferring instead is not the final answer, only a safe one: which side wins +/// is decided by the boot decision pass, which arbitrates the pending row +/// against a writer-consistent head (and its paired tombstone) rather than +/// against whichever event happened to arrive first. Until that pass exists, a +/// pending row shadows genuinely newer remote edits for its coordinate — the +/// flush normally clears it within seconds, and preserving an edit that can +/// still be reconciled beats destroying one that cannot be recovered. pub fn retain_inbound_event( conn: &Connection, event: &RetainedEvent, ) -> Result { let existing = get_retained_event(conn, event.kind, &event.pubkey, &event.d_tag)?; + // Durable local intent outranks every ordering rule below it. + if existing.as_ref().is_some_and(|row| row.pending_sync) { + return Ok(InboundOutcome::Deferred); + } + let apply = match &existing { None => true, Some(row) if event.created_at > row.created_at => true, - Some(row) if event.created_at == row.created_at => { - !row.pending_sync && inbound_wins_equal_timestamp(event, row) - } + Some(row) if event.created_at == row.created_at => inbound_wins_equal_timestamp(event, row), // Older: stale. Some(_) => false, }; @@ -358,9 +390,9 @@ pub fn retain_inbound_event( return Ok(InboundOutcome::Skipped); } - // Inbound is strictly newer (or there was no row): overwrite and clear - // `pending_sync`. No upsert guard is needed — the Rust check above already - // established that this event wins. + // Inbound wins over a non-pending row (or there was no row): overwrite. + // `pending_sync` is written as 0 to state the row's published status + // explicitly; the guard above means it can never have been 1 here. conn.execute( "INSERT INTO persona_events (kind, pubkey, d_tag, content, created_at, raw_event, pending_sync, event_id) VALUES (?1, ?2, ?3, ?4, ?5, ?6, 0, ?7) diff --git a/desktop/src-tauri/src/managed_agents/retention/tests.rs b/desktop/src-tauri/src/managed_agents/retention/tests.rs index b06cc98f5f..d420c9e7a7 100644 --- a/desktop/src-tauri/src/managed_agents/retention/tests.rs +++ b/desktop/src-tauri/src/managed_agents/retention/tests.rs @@ -310,7 +310,7 @@ fn inbound_no_local_row_applies() { } #[test] -fn inbound_equal_second_skips_and_preserves_pending() { +fn inbound_equal_second_defers_and_preserves_pending() { let conn = test_db(); // Pending local edit at t=1000. let local = sample_event(); @@ -324,11 +324,11 @@ fn inbound_equal_second_skips_and_preserves_pending() { }; assert_eq!( retain_inbound_event(&conn, &inbound).unwrap(), - InboundOutcome::Skipped + InboundOutcome::Deferred ); - // Local pending row is untouched: flag preserved, content unchanged so - // the flush republishes and the relay resolves last-writer-wins. + // Local pending row is untouched: flag preserved, content unchanged, for + // the boot decision pass to arbitrate against a writer-consistent head. let row = get_retained_event(&conn, 30175, "abc123", "test-persona") .unwrap() .unwrap(); @@ -337,13 +337,53 @@ fn inbound_equal_second_skips_and_preserves_pending() { } #[test] -fn inbound_strictly_newer_applies_and_clears_pending() { +fn inbound_strictly_newer_defers_to_pending_row() { + // The laundering vector this change exists to close mints STRICTLY NEWER + // events: a stale store re-signs old content at `max(now, head + 1)`. So a + // newer `created_at` is not evidence of newer intent, and applying it here + // would destroy the user's unpublished edit on exactly the input the fix + // targets. Deferral hands the pair to the boot decision pass instead. let conn = test_db(); // Pending local edit at t=1000. let local = sample_event(); retain_event(&conn, &local).unwrap(); // Inbound strictly newer with different content. + let inbound = RetainedEvent { + content: r#"{"display_name":"Remote"}"#.to_string(), + created_at: 2000, + pending_sync: false, + ..sample_event() + }; + assert_eq!( + retain_inbound_event(&conn, &inbound).unwrap(), + InboundOutcome::Deferred + ); + + // Every field of the pending row survives, and the caller does not patch + // disk: `Deferred` is not `Applied`. + let row = get_retained_event(&conn, 30175, "abc123", "test-persona") + .unwrap() + .unwrap(); + assert_eq!(row.created_at, 1000); + assert!(row.pending_sync); + assert!(row.content.contains("Test")); + assert!(!row.content.contains("Remote")); +} + +#[test] +fn inbound_strictly_newer_applies_over_non_pending_row() { + let conn = test_db(); + // Published local row at t=1000 — no local intent to protect. + retain_event( + &conn, + &RetainedEvent { + pending_sync: false, + ..sample_event() + }, + ) + .unwrap(); + let inbound = RetainedEvent { content: r#"{"display_name":"Remote"}"#.to_string(), created_at: 2000, @@ -355,8 +395,6 @@ fn inbound_strictly_newer_applies_and_clears_pending() { InboundOutcome::Applied ); - // Inbound wins: content replaced and pending cleared, so the stale - // local edit stops republishing instead of looping. let row = get_retained_event(&conn, 30175, "abc123", "test-persona") .unwrap() .unwrap(); @@ -368,9 +406,16 @@ fn inbound_strictly_newer_applies_and_clears_pending() { #[test] fn inbound_older_skips() { let conn = test_db(); - let mut local = sample_event(); - local.created_at = 2000; - retain_event(&conn, &local).unwrap(); + // Published local row — the ordering rule decides, not the pending guard. + retain_event( + &conn, + &RetainedEvent { + created_at: 2000, + pending_sync: false, + ..sample_event() + }, + ) + .unwrap(); let inbound = RetainedEvent { content: r#"{"display_name":"Stale"}"#.to_string(), @@ -390,6 +435,41 @@ fn inbound_older_skips() { assert!(!row.content.contains("Stale")); } +#[test] +fn inbound_older_defers_to_pending_row() { + // Third timestamp arm: the guard precedes ordering, so even an inbound + // event that would lose on `created_at` alone reports `Deferred`, not + // `Skipped`. The distinction is what lets the boot decision pass tell + // "lost the compare" apart from "still owes arbitration". + let conn = test_db(); + retain_event( + &conn, + &RetainedEvent { + created_at: 2000, + ..sample_event() + }, + ) + .unwrap(); + + let inbound = RetainedEvent { + content: r#"{"display_name":"Stale"}"#.to_string(), + created_at: 1000, + pending_sync: false, + ..sample_event() + }; + assert_eq!( + retain_inbound_event(&conn, &inbound).unwrap(), + InboundOutcome::Deferred + ); + + let row = get_retained_event(&conn, 30175, "abc123", "test-persona") + .unwrap() + .unwrap(); + assert_eq!(row.created_at, 2000); + assert!(row.pending_sync); + assert!(!row.content.contains("Stale")); +} + /// Two ids at an equal timestamp, ordered as the relay orders them. fn lower_and_higher_ids() -> (String, String) { ("0".repeat(64), "f".repeat(64)) @@ -493,7 +573,7 @@ fn test_equal_second_identical_id_is_a_noop() { } #[test] -fn test_equal_second_lower_id_inbound_still_loses_to_pending_row() { +fn test_equal_second_lower_id_inbound_defers_to_pending_row() { // Pending is durable local intent. Even when the inbound event would win // the relay's id tie-break, applying it here would clear `pending_sync` and // silently drop the user's unpublished edit — the exact class of loss this @@ -519,7 +599,7 @@ fn test_equal_second_lower_id_inbound_still_loses_to_pending_row() { }; assert_eq!( retain_inbound_event(&conn, &inbound).unwrap(), - InboundOutcome::Skipped + InboundOutcome::Deferred ); let row = get_retained_event(&conn, 30175, "abc123", "test-persona") @@ -574,17 +654,18 @@ fn test_equal_second_unorderable_pair_leaves_retained_row_standing() { } #[test] -fn test_strictly_newer_inbound_wins_regardless_of_event_id() { - // The id is only a tie-break. A strictly newer event wins even when its id - // is higher, and even against a pending row — the relay already superseded - // that edit, so republishing it would loop. +fn test_strictly_newer_inbound_defers_regardless_of_event_id() { + // The id is only a tie-break, and it is never reached against a pending + // row: pending outranks every ordering rule, so a strictly newer inbound + // event defers whatever its id. Guards the case where the pending row holds + // the LOWER id — no reading of the tie-break can turn this into an apply. let conn = test_db(); let (lower, higher) = lower_and_higher_ids(); retain_event( &conn, &RetainedEvent { pending_sync: true, - event_id: Some(lower), + event_id: Some(lower.clone()), ..sample_event() }, ) @@ -599,14 +680,15 @@ fn test_strictly_newer_inbound_wins_regardless_of_event_id() { }; assert_eq!( retain_inbound_event(&conn, &inbound).unwrap(), - InboundOutcome::Applied + InboundOutcome::Deferred ); let row = get_retained_event(&conn, 30175, "abc123", "test-persona") .unwrap() .unwrap(); - assert_eq!(row.created_at, 2000); - assert!(!row.pending_sync); + assert_eq!(row.created_at, 1000); + assert!(row.pending_sync); + assert_eq!(row.event_id, Some(lower)); } #[test]