mirror of
https://github.com/block/buzz.git
synced 2026-11-01 15:01:12 +00:00
fix(desktop): defer inbound events against a pending retention row
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 <pfleger.will@gmail.com> Signed-off-by: Will Pfleger <pfleger.will@gmail.com>
This commit is contained in:
parent
27aa3e0a8c
commit
e10f613a59
@@ -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(());
|
||||
}
|
||||
|
||||
|
||||
@@ -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<InboundOutcome, String> {
|
||||
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)
|
||||
|
||||
@@ -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]
|
||||
|
||||
Reference in New Issue
Block a user