diff --git a/crates/buzz-db/src/event.rs b/crates/buzz-db/src/event.rs index 5d682d7843a..21fab218e9c 100644 --- a/crates/buzz-db/src/event.rs +++ b/crates/buzz-db/src/event.rs @@ -152,6 +152,34 @@ pub enum ReactionEventInsertOutcome { }, } +/// Outcome of a NIP-33 parameterized-replaceable coordinate write. +/// +/// Distinguishes the two ways a write can fail to store, which the relay must +/// signal differently on the wire: an exact-id resubmit is a harmless +/// idempotent duplicate (`OK true`), whereas a strictly-losing write against a +/// newer coordinate head is a conflict the client must resolve by refetching +/// (`OK false`). Conflating them — as the pre-fix `bool` did — told a losing +/// device its write succeeded, so it never refetched and diverged silently. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum ParamReplaceOutcome { + /// New coordinate head stored; the event should be fanned out. + Inserted, + /// The exact same event id is already stored (or was previously stored and + /// coordinate-deleted). Idempotent no-op — content already matches intent. + Duplicate, + /// A newer `(created_at, id)` head already dominates this coordinate, so + /// this distinct write lost last-write-wins and was not stored. The + /// client's content differs from the head and must refetch to converge. + Stale, +} + +impl ParamReplaceOutcome { + /// True only when a new head was stored (and must be fanned out). + pub fn was_inserted(self) -> bool { + matches!(self, ParamReplaceOutcome::Inserted) + } +} + /// Maximum length for a `d_tag` value (bytes). NIP-33 d-tags are short identifiers; /// anything beyond this is either a bug or abuse. pub const D_TAG_MAX_LEN: usize = 1024; diff --git a/crates/buzz-db/src/lib.rs b/crates/buzz-db/src/lib.rs index 3ff230f9503..cfea3e390a5 100644 --- a/crates/buzz-db/src/lib.rs +++ b/crates/buzz-db/src/lib.rs @@ -57,7 +57,9 @@ pub mod user; pub mod workflow; pub use error::{DbError, Result}; -pub use event::{EventQuery, ReactionEventInsertOutcome, DEFAULT_MAX_PAGE_LIMIT}; +pub use event::{ + EventQuery, ParamReplaceOutcome, ReactionEventInsertOutcome, DEFAULT_MAX_PAGE_LIMIT, +}; use buzz_datastore_tracing::datastore_span; use chrono::{DateTime, Utc}; @@ -5195,7 +5197,7 @@ impl Db { event: &nostr::Event, d_tag: &str, channel_id: Option, - ) -> Result<(StoredEvent, bool)> { + ) -> Result<(StoredEvent, event::ParamReplaceOutcome)> { let kind_i32 = buzz_core::kind::event_kind_i32(event); let pubkey_bytes = event.pubkey.to_bytes(); let created_at_secs = event.created_at.as_secs() as i64; @@ -5294,9 +5296,23 @@ impl Db { if dominated { tx.rollback().await?; let received_at = chrono::Utc::now(); + // Classify the loss: an exact-id resubmit of a head we already hold + // is idempotent (client's content already matches), but a distinct + // event that merely lost last-write-wins is a conflict the client + // must resolve by refetching. Reported differently on the wire so a + // losing device does not record a stale write as success. + let is_exact_id_resubmit = existing + .iter() + .chain(watermark.iter()) + .any(|(_, accepted_id)| incoming_id == accepted_id.as_slice()); + let outcome = if is_exact_id_resubmit { + event::ParamReplaceOutcome::Duplicate + } else { + event::ParamReplaceOutcome::Stale + }; return Ok(( StoredEvent::with_received_at(event.clone(), received_at, channel_id, false), - false, + outcome, )); } @@ -5367,9 +5383,11 @@ impl Db { let was_inserted = insert_result.rows_affected() > 0; if !was_inserted { tx.rollback().await?; + // The insert hit an existing row with this exact id (e.g. a + // previously soft-deleted coordinate). Idempotent — not a conflict. return Ok(( StoredEvent::with_received_at(event.clone(), received_at, channel_id, false), - false, + event::ParamReplaceOutcome::Duplicate, )); } @@ -5400,7 +5418,7 @@ impl Db { Ok(( StoredEvent::with_received_at(event.clone(), received_at, channel_id, true), - true, + event::ParamReplaceOutcome::Inserted, )) } } @@ -5962,18 +5980,18 @@ mod tests { .sign_with_keys(&keys) .expect("sign new"); - assert!( - db.replace_parameterized_event(community, &old, &d_tag, None) - .await - .expect("insert old") - .1 - ); - assert!( - db.replace_parameterized_event(community, &new, &d_tag, None) - .await - .expect("replace with new") - .1 - ); + assert!(db + .replace_parameterized_event(community, &old, &d_tag, None) + .await + .expect("insert old") + .1 + .was_inserted()); + assert!(db + .replace_parameterized_event(community, &new, &d_tag, None) + .await + .expect("replace with new") + .1 + .was_inserted()); let rows: i64 = sqlx::query_scalar( "SELECT count(*) FROM events WHERE community_id=$1 AND kind=30078 AND pubkey=$2 AND d_tag=$3", @@ -5996,12 +6014,12 @@ mod tests { .await .expect("simulate NIP-09 coordinate deletion"); - assert!( - !db.replace_parameterized_event(community, &old, &d_tag, None) - .await - .expect("replay old") - .1 - ); + assert!(!db + .replace_parameterized_event(community, &old, &d_tag, None) + .await + .expect("replay old") + .1 + .was_inserted()); let live: i64 = sqlx::query_scalar( "SELECT count(*) FROM events WHERE community_id=$1 AND kind=30078 AND pubkey=$2 AND d_tag=$3 AND deleted_at IS NULL", ) @@ -6014,6 +6032,72 @@ mod tests { assert_eq!(live, 0, "watermark must block stale resurrection"); } + #[tokio::test] + #[ignore = "requires Postgres"] + async fn param_replace_distinguishes_stale_conflict_from_exact_id_duplicate() { + use nostr::{EventBuilder, Keys, Kind, Tag, Timestamp}; + + let db = setup_db().await; + let community = CommunityId::from_uuid(make_community(&db.pool).await); + let keys = Keys::generate(); + let d_tag = "channel-sections"; + let tags = vec![Tag::parse(["d", d_tag]).expect("d tag")]; + let base = Timestamp::now().as_secs(); + + let build = |content: &str, ts: u64| { + EventBuilder::new(Kind::Custom(30078), content) + .tags(tags.clone()) + .custom_created_at(Timestamp::from(ts)) + .sign_with_keys(&keys) + .expect("sign") + }; + + // Newer head v2 stored. + let v2 = build("v2", base + 10); + let (_, outcome) = db + .replace_parameterized_event(community, &v2, d_tag, None) + .await + .expect("insert v2"); + assert_eq!(outcome, ParamReplaceOutcome::Inserted); + + // Distinct older write v1 loses last-write-wins → Stale conflict, not a + // duplicate: the relay must reply OK false so the loser refetches. + let v1 = build("v1", base); + let (_, outcome) = db + .replace_parameterized_event(community, &v1, d_tag, None) + .await + .expect("replay v1"); + assert_eq!( + outcome, + ParamReplaceOutcome::Stale, + "a distinct losing write must be a conflict, not an idempotent duplicate" + ); + + // Resubmitting the exact current head v2 is idempotent → Duplicate. + let (_, outcome) = db + .replace_parameterized_event(community, &v2, d_tag, None) + .await + .expect("resubmit v2"); + assert_eq!( + outcome, + ParamReplaceOutcome::Duplicate, + "an exact-id resubmit of the current head must stay an idempotent duplicate" + ); + + // The stored head is still v2 and there is exactly one live row. + let live_content: String = sqlx::query_scalar( + "SELECT content FROM events WHERE community_id=$1 AND kind=30078 AND pubkey=$2 \ + AND d_tag=$3 AND deleted_at IS NULL", + ) + .bind(community.as_uuid()) + .bind(keys.public_key().to_bytes()) + .bind(d_tag) + .fetch_one(&db.pool) + .await + .expect("load live head"); + assert_eq!(live_content, "v2", "v2 must remain the coordinate head"); + } + #[tokio::test] #[ignore = "requires Postgres"] async fn mesh_status_replacement_keeps_one_physical_row() { @@ -6037,12 +6121,12 @@ mod tests { .custom_created_at(Timestamp::from(base + offset)) .sign_with_keys(&keys) .expect("sign mesh status"); - assert!( - db.replace_parameterized_event(community, &event, d_tag, None) - .await - .expect("replace mesh status") - .1 - ); + assert!(db + .replace_parameterized_event(community, &event, d_tag, None) + .await + .expect("replace mesh status") + .1 + .was_inserted()); } let (rows, live): (i64, i64) = sqlx::query_as( @@ -6105,12 +6189,12 @@ mod tests { }; for (content, offset) in [("v1", 0), ("v2", 100)] { - assert!( - db.replace_parameterized_event(community, &version(content, offset), d_tag, None) - .await - .expect("store project version") - .1 - ); + assert!(db + .replace_parameterized_event(community, &version(content, offset), d_tag, None) + .await + .expect("store project version") + .1 + .was_inserted()); } // Tombstone timestamped between V1 and V2: it authorizes deleting V1, @@ -6207,18 +6291,18 @@ mod tests { .sign_with_keys(&keys) .expect("sign new event"); - assert!( - db.replace_parameterized_event(community, &old, &d_tag, None) - .await - .expect("insert old event") - .1 - ); - assert!( - db.replace_parameterized_event(community, &new, &d_tag, None) - .await - .expect("replace with new event") - .1 - ); + assert!(db + .replace_parameterized_event(community, &old, &d_tag, None) + .await + .expect("insert old event") + .1 + .was_inserted()); + assert!(db + .replace_parameterized_event(community, &new, &d_tag, None) + .await + .expect("replace with new event") + .1 + .was_inserted()); let (rows, live): (i64, i64) = sqlx::query_as( "SELECT count(*), count(*) FILTER (WHERE deleted_at IS NULL) FROM events \ @@ -6267,12 +6351,12 @@ mod tests { .custom_created_at(Timestamp::from(base)) .sign_with_keys(&keys) .expect("sign conforming event"); - assert!( - db.replace_parameterized_event(community, &conforming, &conforming_d, None) - .await - .expect("insert conforming event") - .1 - ); + assert!(db + .replace_parameterized_event(community, &conforming, &conforming_d, None) + .await + .expect("insert conforming event") + .1 + .was_inserted()); sqlx::query( "INSERT INTO event_mentions \ (community_id, pubkey_hex, event_id, event_created_at, event_kind) \ @@ -6324,12 +6408,12 @@ mod tests { .custom_created_at(Timestamp::from(base + 1)) .sign_with_keys(&keys) .expect("sign nonconforming event"); - assert!( - db.replace_parameterized_event(community, &nonconforming, &nonconforming_d, None,) - .await - .expect("insert nonconforming event") - .1 - ); + assert!(db + .replace_parameterized_event(community, &nonconforming, &nonconforming_d, None,) + .await + .expect("insert nonconforming event") + .1 + .was_inserted()); let rejected_nonconforming = sqlx::query( "DELETE FROM events WHERE community_id=$1 AND id=$2 AND created_at=to_timestamp($3)", ) @@ -6349,12 +6433,12 @@ mod tests { .custom_created_at(Timestamp::from(base + 2)) .sign_with_keys(&keys) .expect("sign unrelated event"); - assert!( - db.replace_parameterized_event(community, &unrelated, &unrelated_d, None) - .await - .expect("insert unrelated event") - .1 - ); + assert!(db + .replace_parameterized_event(community, &unrelated, &unrelated_d, None) + .await + .expect("insert unrelated event") + .1 + .was_inserted()); let unrelated_delete = sqlx::query( "DELETE FROM events WHERE community_id=$1 AND id=$2 AND created_at=to_timestamp($3)", ) @@ -6571,7 +6655,7 @@ mod tests { .expect("Rust hard delete deadlocked with mention insert") .expect("replacement task panicked") .expect("replace B with C"); - assert!(replaced.1, "C must replace B"); + assert!(replaced.1.was_inserted(), "C must replace B"); let b_mentions: i64 = sqlx::query_scalar( "SELECT count(*) FROM event_mentions WHERE community_id=$1 AND event_id=$2", ) diff --git a/crates/buzz-relay/src/handlers/ingest.rs b/crates/buzz-relay/src/handlers/ingest.rs index dd2fa6e93e0..951dc6b2ac9 100644 --- a/crates/buzz-relay/src/handlers/ingest.rs +++ b/crates/buzz-relay/src/handlers/ingest.rs @@ -3148,11 +3148,23 @@ async fn ingest_event_inner( buzz_db::event::D_TAG_MAX_LEN, ))); } - state + let (stored, outcome) = state .db .replace_parameterized_event(tenant.community(), &event, &d_tag, channel_id) .await - .map_err(|e| IngestError::Internal(format!("error: {e}")))? + .map_err(|e| IngestError::Internal(format!("error: {e}")))?; + // A strictly-losing coordinate write is a conflict, not a duplicate: + // reply OK false so the client refetches the head and converges, + // instead of recording its stale content as successfully synced. An + // exact-id resubmit stays an idempotent OK true duplicate below. + if outcome == buzz_db::ParamReplaceOutcome::Stale { + return Ok(IngestResult { + event_id: event_id_hex, + accepted: false, + message: "conflict: newer version exists".into(), + }); + } + (stored, outcome.was_inserted()) } else { let thread_params = thread_meta.as_ref().map(|m| m.as_params()); match state diff --git a/crates/buzz-relay/src/handlers/side_effects.rs b/crates/buzz-relay/src/handlers/side_effects.rs index 89595fbee17..b9d5486d476 100644 --- a/crates/buzz-relay/src/handlers/side_effects.rs +++ b/crates/buzz-relay/src/handlers/side_effects.rs @@ -3445,11 +3445,11 @@ pub async fn publish_dm_visibility_snapshot( .sign_with_keys(&state.relay_keypair) .map_err(|e| anyhow::anyhow!("failed to sign kind:{KIND_DM_VISIBILITY}: {e}"))?; - let (stored, was_inserted) = state + let (stored, outcome) = state .db .replace_parameterized_event(tenant.community(), &event, &viewer_hex, None) .await?; - if was_inserted { + if outcome.was_inserted() { dispatch_persistent_event( tenant, state, diff --git a/desktop/src/features/sidebar/lib/channelMutesStorage.ts b/desktop/src/features/sidebar/lib/channelMutesStorage.ts index 1bf315d268d..e8de164f784 100644 --- a/desktop/src/features/sidebar/lib/channelMutesStorage.ts +++ b/desktop/src/features/sidebar/lib/channelMutesStorage.ts @@ -111,6 +111,30 @@ export function writeChannelMutesStore( export function mergeStores( local: ChannelMuteStore, remote: ChannelMuteStore, +): ChannelMuteStore { + return mergeStoresWithTie(local, remote, false); +} + +/** + * Merge a remote store that has already won the event-level canonical tie-break + * (`created_at DESC, id ASC`) into the local store, resolving a per-entry + * `updatedAt` tie in favour of the *remote* value. Once the comparator has + * chosen this remote event as the stored winner, its per-entry values must + * survive, or a stale value from a superseded larger-id event delivered first + * would win the merge and silently undo the canonical winner. Strictly-newer + * local per-entry edits (`l.updatedAt > r.updatedAt`) still win. + */ +export function mergeApplyingRemote( + local: ChannelMuteStore, + remote: ChannelMuteStore, +): ChannelMuteStore { + return mergeStoresWithTie(local, remote, true); +} + +function mergeStoresWithTie( + local: ChannelMuteStore, + remote: ChannelMuteStore, + preferRemoteOnTie: boolean, ): ChannelMuteStore { const allIds = new Set([ ...Object.keys(local.channels), @@ -121,7 +145,10 @@ export function mergeStores( const l = local.channels[id]; const r = remote.channels[id]; if (l && r) { - merged[id] = l.updatedAt >= r.updatedAt ? l : r; + const localWins = preferRemoteOnTie + ? l.updatedAt > r.updatedAt + : l.updatedAt >= r.updatedAt; + merged[id] = localWins ? l : r; } else { merged[id] = (l ?? r) as ChannelMuteEntry; } diff --git a/desktop/src/features/sidebar/lib/channelSectionsStorage.ts b/desktop/src/features/sidebar/lib/channelSectionsStorage.ts index f01154751bd..f2b02bd18dd 100644 --- a/desktop/src/features/sidebar/lib/channelSectionsStorage.ts +++ b/desktop/src/features/sidebar/lib/channelSectionsStorage.ts @@ -194,3 +194,57 @@ export function writeChannelSectionsStore( return false; } } + +const OUTBOX_KEY_PREFIX = "buzz-channel-sections-outbox.v1"; + +function outboxKey(pubkey: string, relayUrl: string): string { + return `${OUTBOX_KEY_PREFIX}:${pubkey}:${encodeURIComponent(normalizeRelayUrl(relayUrl))}`; +} + +/** + * Persist an unpublished edit so it survives quit/community-switch within the + * 2s publish debounce. Written synchronously on every edit; cleared once the + * edit is published, superseded by an adopted remote head, or found identical + * to the last published store. Resumed on next mount so a durable intent is + * never silently dropped at teardown. + */ +export function writeChannelSectionsOutbox( + pubkey: string, + store: ChannelSectionStore, + relayUrl: string, +): void { + try { + window.localStorage.setItem( + outboxKey(pubkey, relayUrl), + JSON.stringify(boundChannelSectionsStore(store)), + ); + } catch { + // Best-effort durability; the in-memory pendingStore still drives this + // session's publish even if the persisted copy could not be written. + } +} + +/** Read a persisted unpublished edit, or null when none/unparseable. */ +export function readChannelSectionsOutbox( + pubkey: string, + relayUrl: string, +): ChannelSectionStore | null { + try { + return parseRaw(window.localStorage.getItem(outboxKey(pubkey, relayUrl))); + } catch { + return null; + } +} + +/** Clear the persisted outbox (edit published, superseded, or a no-op). */ +export function clearChannelSectionsOutbox( + pubkey: string, + relayUrl: string, +): void { + try { + window.localStorage.removeItem(outboxKey(pubkey, relayUrl)); + } catch { + // Ignore — a stale outbox entry is re-evaluated (and re-cleared if + // identical to the head) on the next publish attempt. + } +} diff --git a/desktop/src/features/sidebar/lib/channelSectionsSync.test.mjs b/desktop/src/features/sidebar/lib/channelSectionsSync.test.mjs index 904ac1f3f24..4b9c619778c 100644 --- a/desktop/src/features/sidebar/lib/channelSectionsSync.test.mjs +++ b/desktop/src/features/sidebar/lib/channelSectionsSync.test.mjs @@ -179,25 +179,28 @@ test("revert-fix: absent fetch with zero watermark seeds via bootstrap (first-sy } }); -// 4. LWW baseline: newer decryptable pre-publish event still wins after an -// undecryptable head was recorded. -// Mutation test: headBeforeFetch → this.lastRemoteCreatedAt makes comparison -// 200>200=false → local wins instead of remote → wrong content encrypted. -test("revert-fix: sections LWW — newer decryptable pre-publish event selected after undecryptable head recorded", async () => { +// 4. Adopt-winner: a newer remote head at pre-publish time supersedes the local +// edit — the manager must NOT publish, must hand the remote to the adopt +// sink, and must clear the pending/outbox so the loser can't be replayed. +// Mutation test: reverting adopt→republish makes onRemoteAdopted never fire and +// publishEvent fire instead. +test("adopt-winner: newer remote head at pre-publish adopts remote and skips publish", async () => { const REMOTE_ID = "remote-section-from-relay"; - let callCount = 0; - mock.method(relayClient, "fetchEvents", () => { - callCount++; - return Promise.resolve([ + mock.method(relayClient, "fetchEvents", () => + Promise.resolve([ { pubkey: "pk-lww", - content: callCount === 1 ? "bad-cipher" : "good-cipher", - created_at: callCount === 1 ? 100 : 200, - id: `evt-${callCount}`, + content: "good-cipher", + created_at: 200, + id: "evt-remote", }, - ]); + ]), + ); + const publishCalls = []; + mock.method(relayClient, "publishEvent", (...args) => { + publishCalls.push(args); + return Promise.resolve(); }); - mock.method(relayClient, "publishEvent", () => Promise.resolve()); const fw = makeFakeWindow(); const restore = installFakeWindow(fw); const tauri = installTauriMock( @@ -209,25 +212,178 @@ test("revert-fix: sections LWW — newer decryptable pre-publish event selected ); try { const manager = new ChannelSectionSyncManager("pk-lww", RELAY); + const adopted = []; + manager.setOnRemoteAdopted((r) => adopted.push(r)); + manager.publishSections( + makeSectionsStore([{ id: "local-s", name: "Local", order: 0 }]), + ); + // Outbox persisted synchronously on the edit. + assert.ok( + fw.localStorage.getItem( + `buzz-channel-sections-outbox.v1:pk-lww:${RELAY_KEY}`, + ) !== null, + "edit must be persisted to the durable outbox", + ); + fw._fireTimer(); + await new Promise((r) => setTimeout(r, 20)); + assert.equal( + publishCalls.length, + 0, + "must not publish when a newer remote head wins LWW", + ); + assert.equal(adopted.length, 1, "adopt sink must receive the remote"); + assert.ok( + adopted[0].store.sections.some((s) => s.id === REMOTE_ID), + "adopted store must be the remote content", + ); + assert.equal(manager.getPendingStore(), null, "pending must be cleared"); + assert.equal( + fw.localStorage.getItem( + `buzz-channel-sections-outbox.v1:pk-lww:${RELAY_KEY}`, + ), + null, + "outbox must be cleared on adopt so the loser is never replayed", + ); + } finally { + tauri.restore(); + restore(); + mock.reset(); + } +}); + +// 4b. Local edit wins (no newer remote head): publishes and clears the outbox. +test("adopt-winner: local edit at/ahead of head publishes and clears outbox", async () => { + mock.method(relayClient, "fetchEvents", () => Promise.resolve([])); + const publishCalls = []; + mock.method(relayClient, "publishEvent", (...args) => { + publishCalls.push(args); + return Promise.resolve(); + }); + const fw = makeFakeWindow(); + const restore = installFakeWindow(fw); + const tauri = installTauriMock("{}"); + try { + const manager = new ChannelSectionSyncManager("pk-win", RELAY); + manager.publishSections( + makeSectionsStore([{ id: "s1", name: "Work", order: 0 }]), + ); + fw._fireTimer(); + await new Promise((r) => setTimeout(r, 20)); + assert.equal(publishCalls.length, 1, "local edit must be published"); + assert.equal( + fw.localStorage.getItem( + `buzz-channel-sections-outbox.v1:pk-win:${RELAY_KEY}`, + ), + null, + "outbox must be cleared once the edit is published", + ); + } finally { + tauri.restore(); + restore(); + mock.reset(); + } +}); + +// 4c. Timestamp clamp: a remote head far in the future must not make the +// published createdAt walk past the relay's ±15min window. +// Mutation test: removing the Math.min clamp lets createdAt = lastRemote+1 +// (~now+3600), which exceeds now + MAX_PUBLISH_FUTURE_SECS. +test("timestamp clamp: published createdAt stays inside the relay future window", async () => { + const nowSecs = Math.floor(Date.now() / 1000); + const farFutureHead = nowSecs + 3_600; // 1h ahead — beyond the ±15min window + let call = 0; + mock.method(relayClient, "fetchEvents", () => { + call++; + // First call: fetchRemoteSections during a manual head prime; subsequent: + // pre-publish fetch. Return the far-future undecryptable head each time so + // lastRemoteCreatedAt is pushed to farFutureHead but the store still + // publishes (local edit is what we're stamping). + return Promise.resolve([ + { + pubkey: "pk-clamp", + content: "good-cipher", + created_at: call === 1 ? farFutureHead : 0, + id: "evt-clamp", + }, + ]); + }); + let signedCreatedAt = null; + const fw = makeFakeWindow(); + const restore = installFakeWindow(fw); + const tauri = installTauriMock( + JSON.stringify({ version: 1, sections: [], assignments: {} }), + ); + mock.method(relayClient, "publishEvent", (evt) => { + signedCreatedAt = evt.created_at; + return Promise.resolve(); + }); + try { + const manager = new ChannelSectionSyncManager("pk-clamp", RELAY); + // Prime lastRemoteCreatedAt to the far-future head. await manager.fetchRemoteSections(); + manager.publishSections( + makeSectionsStore([{ id: "s1", name: "Work", order: 0 }]), + ); + // Fire debounce; pre-publish fetch returns created_at=0 so local wins. + fw._fireTimer(); + await new Promise((r) => setTimeout(r, 20)); + assert.ok(signedCreatedAt !== null, "publish must have been attempted"); assert.ok( - Number( - fw.localStorage.getItem( - `buzz-sync-watermark.v1:channel-sections:pk-lww:${RELAY_KEY}`, - ) ?? "0", - ) >= 100, + signedCreatedAt <= Math.floor(Date.now() / 1000) + 840, + `createdAt must be clamped inside the future window — got ${signedCreatedAt}`, ); + } finally { + tauri.restore(); + restore(); + mock.reset(); + } +}); + +// 4d. Conflict rejection: relay OK false → refetch head and adopt it. +test("conflict rejection: OK-false conflict refetches head and adopts remote", async () => { + const REMOTE_ID = "remote-after-conflict"; + let fetchCall = 0; + mock.method(relayClient, "fetchEvents", () => { + fetchCall++; + // First fetch (pre-publish): empty → local wins and we publish. + if (fetchCall === 1) return Promise.resolve([]); + // Second fetch (post-conflict refetch): the winning remote head. + return Promise.resolve([ + { + pubkey: "pk-conflict", + content: "good-cipher", + created_at: 500, + id: "evt-winner", + }, + ]); + }); + mock.method(relayClient, "publishEvent", () => + Promise.reject(new Error("conflict: newer version exists")), + ); + const fw = makeFakeWindow(); + const restore = installFakeWindow(fw); + const tauri = installTauriMock( + JSON.stringify({ + version: 1, + sections: [{ id: REMOTE_ID, name: "Remote", order: 0 }], + assignments: {}, + }), + ); + try { + const manager = new ChannelSectionSyncManager("pk-conflict", RELAY); + const adopted = []; + manager.setOnRemoteAdopted((r) => adopted.push(r)); manager.publishSections( makeSectionsStore([{ id: "local-s", name: "Local", order: 0 }]), ); fw._fireTimer(); await new Promise((r) => setTimeout(r, 20)); - const pt = tauri.capturedPlaintext(); - assert.ok(pt !== null, "nip44EncryptToSelf must have been called"); + assert.equal(adopted.length, 1, "conflict must trigger adopt of the head"); assert.ok( - JSON.parse(pt).sections?.some((s) => s.id === REMOTE_ID), - `remote sections must win LWW merge — got: ${pt}`, + adopted[0].store.sections.some((s) => s.id === REMOTE_ID), + "adopted store must be the winning remote content", ); + assert.equal(manager.getPendingStore(), null, "pending cleared on adopt"); } finally { tauri.restore(); restore(); @@ -280,3 +436,397 @@ test("revert-fix: undecryptable live event advances watermark before decrypt att mock.reset(); } }); + +// 6. Overlapping publishes (fix 1): an older in-flight publish must not clear a +// newer edit queued while it was in flight. Regression for the generation +// compare-and-swap on discardPending — reverting the gen guard makes the +// older completion null out B's pendingStore + outbox. +test("overlapping publishes: older completion does not erase a newer queued edit", async () => { + mock.method(relayClient, "fetchEvents", () => Promise.resolve([])); + // First publish blocks until we release it; a second publish is queued while + // the first is in flight. + let releaseFirst = null; + let publishCalls = 0; + mock.method(relayClient, "publishEvent", () => { + publishCalls++; + if (publishCalls === 1) { + return new Promise((res) => { + releaseFirst = res; + }); + } + return Promise.resolve(); + }); + // A multi-slot timer fake: each setTimeout is retained by delay so we can fire + // the debounce independently and inspect what remains. + const storage = new Map(); + const timers = new Map(); + let nextId = 1; + const fakeWindow = { + localStorage: { + getItem: (k) => storage.get(k) ?? null, + setItem: (k, v) => storage.set(k, v), + removeItem: (k) => storage.delete(k), + }, + setTimeout: (fn, ms) => { + const id = nextId++; + timers.set(id, { fn, ms }); + return id; + }, + clearTimeout: (id) => timers.delete(id), + }; + const fireDelay = async (ms) => { + const entry = [...timers.entries()].find(([, v]) => v.ms === ms); + assert.ok(entry, `expected a timer scheduled at ${ms}ms`); + timers.delete(entry[0]); + entry[1].fn(); + await Promise.resolve(); + await Promise.resolve(); + }; + const restore = installFakeWindow(fakeWindow); + const tauri = installTauriMock("{}"); + const outboxKey = `buzz-channel-sections-outbox.v1:pk-overlap:${RELAY_KEY}`; + try { + const manager = new ChannelSectionSyncManager("pk-overlap", RELAY); + const storeA = makeSectionsStore([{ id: "a", name: "A", order: 0 }]); + const storeB = makeSectionsStore([{ id: "b", name: "B", order: 0 }]); + + manager.publishSections(storeA); + await fireDelay(2000); // debounce → doPublish(A) awaits publishEvent + while (releaseFirst === null) await Promise.resolve(); + + // Edit B arrives while A is still in flight. + manager.publishSections(storeB); + assert.deepEqual( + manager.getPendingStore()?.sections.map((s) => s.id), + ["b"], + "B is now the pending edit", + ); + assert.equal( + JSON.parse(storage.get(outboxKey)).sections[0].id, + "b", + "outbox holds B", + ); + + // A completes — its success path must NOT clear B's pending/outbox. + releaseFirst(); + await Promise.resolve(); + await Promise.resolve(); + await Promise.resolve(); + + assert.deepEqual( + manager.getPendingStore()?.sections.map((s) => s.id), + ["b"], + "older completion must leave B pending", + ); + assert.ok( + storage.get(outboxKey) !== undefined, + "older completion must leave B's outbox intact", + ); + assert.ok( + [...timers.values()].some((t) => t.ms === 2000), + "B's debounce timer must survive so it still publishes", + ); + manager.destroy(); + } finally { + tauri.restore(); + restore(); + mock.reset(); + } +}); + +// 7. Live remote during debounce (pass-2 finding 1): a remote head accepted +// while a local edit is debouncing must be adopted at pre-publish, not +// overwritten. The live event advances the watermark before doPublish runs, +// so comparing the fetched head against the mutable watermark would see +// equality and publish over the newer remote. The pre-publish check compares +// against the baseline frozen at publishSections instead. Mutation: comparing +// against lastRemoteCreatedAt rather than publishBaseline republishes local. +test("live remote during debounce is adopted at pre-publish, not overwritten", async () => { + const remoteEvent = { + id: "remote-event", + pubkey: "pk-livedebounce", + content: "good-cipher", + created_at: 1_700_000_100, + kind: 30078, + tags: [["d", "channel-sections"]], + sig: "s", + }; + // Pre-publish fetch returns the same live head that arrived during debounce. + mock.method(relayClient, "fetchEvents", () => Promise.resolve([remoteEvent])); + const publishCalls = []; + mock.method(relayClient, "publishEvent", (...args) => { + publishCalls.push(args); + return Promise.resolve(); + }); + let live = null; + mock.method(relayClient, "subscribeLive", async (_filter, cb) => { + live = cb; + return async () => {}; + }); + + const storage = new Map(); + const timers = new Map(); + let nextId = 1; + const fakeWindow = { + localStorage: { + getItem: (k) => storage.get(k) ?? null, + setItem: (k, v) => storage.set(k, v), + removeItem: (k) => storage.delete(k), + }, + setTimeout: (fn, ms) => { + const id = nextId++; + timers.set(id, { fn, ms }); + return id; + }, + clearTimeout: (id) => timers.delete(id), + }; + const fireDelay = async (ms) => { + const entry = [...timers.entries()].find(([, v]) => v.ms === ms); + assert.ok(entry, `expected a timer scheduled at ${ms}ms`); + timers.delete(entry[0]); + entry[1].fn(); + for (let i = 0; i < 50; i++) await Promise.resolve(); + }; + const restore = installFakeWindow(fakeWindow); + const tauri = installTauriMock( + JSON.stringify({ + version: 1, + sections: [{ id: "remote", name: "Remote", order: 0 }], + assignments: {}, + }), + ); + const outboxKey = `buzz-channel-sections-outbox.v1:pk-livedebounce:${RELAY_KEY}`; + try { + const manager = new ChannelSectionSyncManager("pk-livedebounce", RELAY); + const adopted = []; + manager.setOnRemoteAdopted((remote) => adopted.push(remote.eventId)); + await manager.subscribeToSections(() => {}); + assert.ok(live, "live subscription installed"); + + manager.publishSections( + makeSectionsStore([{ id: "local", name: "Local", order: 0 }]), + ); + // A genuinely later remote head is accepted and delivered while local is + // pending — this advances the watermark past the frozen baseline. + live(remoteEvent); + for (let i = 0; i < 50; i++) await Promise.resolve(); + + await fireDelay(2000); + + assert.equal( + publishCalls.length, + 0, + "later remote head must prevent the local publish", + ); + assert.deepEqual( + adopted, + ["remote-event"], + "the remote accepted after the edit began must be adopted", + ); + assert.equal(manager.getPendingStore(), null, "pending cleared on adopt"); + assert.equal( + storage.get(outboxKey), + undefined, + "outbox cleared on adopt so the loser can't replay", + ); + manager.destroy(); + } finally { + tauri.restore(); + restore(); + mock.reset(); + } +}); + +// 8. Serialized generations (fix round 3, pass-3 finding 1): an older in-flight +// publish that completes after a newer edit is queued must NOT be mistaken +// for a remote that advanced past the newer edit's baseline. A blocks in +// publishEvent; B is queued; A succeeds; B's pre-publish fetch returns A's +// accepted head. Because the baseline is frozen at B's own cycle start — +// after A's completion recorded its head — B publishes above A instead of +// adopting it. Mutation: freezing the baseline at publishSections (before the +// prior cycle completes) makes B see A as a post-baseline remote and adopt it. +test("serialized generations: older completion does not make the newer edit adopt it", async () => { + let releaseFirst = null; + let publishCalls = 0; + let storedHead = []; + mock.method(relayClient, "fetchEvents", () => Promise.resolve(storedHead)); + mock.method(relayClient, "publishEvent", (event) => { + publishCalls++; + if (publishCalls === 1) { + // A's publish blocks; when it resolves, its event becomes the stored head + // the next pre-publish fetch will return. + return new Promise((res) => { + releaseFirst = () => { + storedHead = [ + { + id: "event-a", + pubkey: "pk-serial", + content: "good-cipher", + created_at: event.created_at, + kind: 30078, + tags: [["d", "channel-sections"]], + sig: "s", + }, + ]; + res(); + }; + }); + } + return Promise.resolve(); + }); + const storage = new Map(); + const timers = new Map(); + let nextId = 1; + const fakeWindow = { + localStorage: { + getItem: (k) => storage.get(k) ?? null, + setItem: (k, v) => storage.set(k, v), + removeItem: (k) => storage.delete(k), + }, + setTimeout: (fn, ms) => { + const id = nextId++; + timers.set(id, { fn, ms }); + return id; + }, + clearTimeout: (id) => timers.delete(id), + }; + const fireDelay = async (ms) => { + const entry = [...timers.entries()].find(([, v]) => v.ms === ms); + assert.ok(entry, `expected a timer scheduled at ${ms}ms`); + timers.delete(entry[0]); + entry[1].fn(); + for (let i = 0; i < 100; i++) await Promise.resolve(); + }; + const restore = installFakeWindow(fakeWindow); + // A's decrypted head must parse; the pre-publish check reads its created_at/id. + const tauri = installTauriMock( + JSON.stringify({ + version: 1, + sections: [{ id: "a", name: "A", order: 0 }], + assignments: {}, + }), + ); + try { + const manager = new ChannelSectionSyncManager("pk-serial", RELAY); + const adopted = []; + manager.setOnRemoteAdopted((remote) => adopted.push(remote.eventId)); + + manager.publishSections( + makeSectionsStore([{ id: "a", name: "A", order: 0 }]), + ); + await fireDelay(2000); // A's cycle → publishEvent(A) blocks + while (releaseFirst === null) await Promise.resolve(); + + // B is queued while A is still in flight; its cycle must defer. + manager.publishSections( + makeSectionsStore([{ id: "b", name: "B", order: 0 }]), + ); + assert.deepEqual( + manager.getPendingStore()?.sections.map((s) => s.id), + ["b"], + "B is the pending edit while A is in flight", + ); + + releaseFirst(); // A completes, recording its head; the freed lane drives B + for (let i = 0; i < 100; i++) await Promise.resolve(); + // If B's cycle did not auto-drive on the freed lane, its debounce timer is + // still pending — fire it. + if ([...timers.values()].some((t) => t.ms === 2000)) await fireDelay(2000); + + assert.deepEqual(adopted, [], "B must not adopt the older generation A"); + assert.equal( + publishCalls, + 2, + "B publishes above A rather than adopting A's accepted head", + ); + assert.equal( + manager.getPendingStore(), + null, + "B's pending clears via its own successful publish, not A's completion", + ); + manager.destroy(); + } finally { + tauri.restore(); + restore(); + mock.reset(); + } +}); + +// 9. Serialized generations (fix round 3, pass-3 finding 1): a stale generation +// must never sign/publish after a newer edit is queued. A blocks in the +// pre-publish fetch; B is queued during that await; A must abort before +// signing. Mutation: dropping the post-fetch generation re-check in doPublish +// lets the stale A continue to publishEvent. +test("serialized generations: a stale generation aborts before publishing", async () => { + let releaseFetch = null; + const publishCalls = []; + mock.method( + relayClient, + "fetchEvents", + () => + new Promise((res) => { + releaseFetch = () => res([]); + }), + ); + mock.method(relayClient, "publishEvent", (...args) => { + publishCalls.push(args); + return Promise.resolve(); + }); + const storage = new Map(); + const timers = new Map(); + let nextId = 1; + const fakeWindow = { + localStorage: { + getItem: (k) => storage.get(k) ?? null, + setItem: (k, v) => storage.set(k, v), + removeItem: (k) => storage.delete(k), + }, + setTimeout: (fn, ms) => { + const id = nextId++; + timers.set(id, { fn, ms }); + return id; + }, + clearTimeout: (id) => timers.delete(id), + }; + const fireDelay = async (ms) => { + const entry = [...timers.entries()].find(([, v]) => v.ms === ms); + assert.ok(entry, `expected a timer scheduled at ${ms}ms`); + timers.delete(entry[0]); + entry[1].fn(); + for (let i = 0; i < 100; i++) await Promise.resolve(); + }; + const restore = installFakeWindow(fakeWindow); + const tauri = installTauriMock("{}"); + try { + const manager = new ChannelSectionSyncManager("pk-stale", RELAY); + manager.publishSections( + makeSectionsStore([{ id: "a", name: "A", order: 0 }]), + ); + await fireDelay(2000); // A's cycle → doPublish(A) awaits fetchEvents + while (releaseFetch === null) await Promise.resolve(); + + // B is queued while A is blocked in its pre-publish fetch. + manager.publishSections( + makeSectionsStore([{ id: "b", name: "B", order: 0 }]), + ); + + releaseFetch(); // A's fetch resolves; A must see gen moved and abort + for (let i = 0; i < 100; i++) await Promise.resolve(); + + assert.equal( + publishCalls.length, + 0, + "stale generation A must not sign/publish after B was queued", + ); + assert.deepEqual( + manager.getPendingStore()?.sections.map((s) => s.id), + ["b"], + "B remains the pending edit, owning convergence", + ); + manager.destroy(); + } finally { + tauri.restore(); + restore(); + mock.reset(); + } +}); diff --git a/desktop/src/features/sidebar/lib/channelSectionsSync.ts b/desktop/src/features/sidebar/lib/channelSectionsSync.ts index 858b62430f3..fdf1e4453e3 100644 --- a/desktop/src/features/sidebar/lib/channelSectionsSync.ts +++ b/desktop/src/features/sidebar/lib/channelSectionsSync.ts @@ -7,7 +7,9 @@ import { import type { RelayEvent } from "@/shared/api/types"; import { KIND_CHANNEL_SECTIONS } from "@/shared/constants/kinds"; import { + clearChannelSectionsOutbox, parseChannelSectionPayload, + writeChannelSectionsOutbox, type ChannelSection, type ChannelSectionStore, } from "./channelSectionsStorage"; @@ -22,12 +24,83 @@ const D_TAG = "channel-sections"; const BLOB_TYPE = D_TAG; const DEBOUNCE_MS = 2_000; +// The relay rejects events more than ±15 minutes (900s) from server time +// (`MAX_TIMESTAMP_DRIFT_SECS` in ingest.rs). Clamp our published `created_at` +// well inside that window so a skewed remote head can never make us manufacture +// an unbounded future timestamp that wedges every subsequent publish. 840s +// leaves ~60s of transit margin while still letting us win LWW against any +// legitimately-timestamped head. +const MAX_PUBLISH_FUTURE_SECS = 840; + +// Bounded backoff for a retained pending edit whose publish failed transiently +// (timeout / socket error) on an otherwise-healthy socket, so it does not wait +// for a reconnect that may never fire. +const RETRY_BASE_MS = 2_000; +const RETRY_MAX_MS = 30_000; + export type RemoteSections = { store: ChannelSectionStore; createdAt: number; eventId: string; }; +/** + * Outcome of the pre-publish head check. + * + * - `publish` — local edit is at or ahead of the head; publish it. + * - `adopt` — a newer remote head exists; the local edit lost whole-blob + * LWW and must be discarded in favour of the remote store so UI + * and relay converge (see the fix-2 design note). The manager + * hands the remote back to the hook and never publishes. + */ +type PublishDecision = + | { kind: "publish"; store: ChannelSectionStore } + | { kind: "adopt"; remote: RemoteSections }; + +/** + * The canonical remote head as it stood when an edit was queued. The pre-publish + * check compares the fetched head against this frozen baseline — never against + * the mutable in-memory watermark, which a live event observed during the + * debounce window may already have advanced to that same head (silently + * suppressing the adopt). + */ +type PublishBaseline = { createdAt: number; eventId: string }; + +/** + * True when `head` is the canonical winner over the baseline the edit was queued + * against — i.e. the head advanced since the edit began. Canonical order is + * `created_at DESC, id ASC`: a strictly-later head wins, and a same-second head + * wins only with a strictly-lower id. A same-second head is comparable only once + * the baseline id is known (empty id = no prior head seen → not superseded). + */ +function remoteAdvancedSince( + head: RemoteSections, + baseline: PublishBaseline, +): boolean { + if (head.createdAt > baseline.createdAt) return true; + return ( + head.createdAt === baseline.createdAt && + baseline.eventId !== "" && + head.eventId < baseline.eventId + ); +} + +/** + * True when tuple `a` is the canonical winner over `b` (`created_at DESC, + * id ASC`). An empty id means "no head seen yet" and always loses. + */ +function canonicalGreater(a: PublishBaseline, b: PublishBaseline): boolean { + if (a.eventId === "") return false; + if (b.eventId === "") return true; + if (a.createdAt !== b.createdAt) return a.createdAt > b.createdAt; + return a.eventId < b.eventId; +} + +/** The canonical-greater of two head tuples (`created_at DESC, id ASC`). */ +function canonicalMax(a: PublishBaseline, b: PublishBaseline): PublishBaseline { + return canonicalGreater(a, b) ? a : b; +} + async function decryptAndParse( event: RelayEvent, ): Promise { @@ -45,10 +118,42 @@ export class ChannelSectionSyncManager { private pubkey: string; private relayUrl: string; private debounceTimer: number | null = null; + private retryTimer: number | null = null; + private retryDelayMs = RETRY_BASE_MS; private lastRemoteCreatedAt: number; + // Canonical best head observed so far (`created_at DESC, id ASC`). Frozen into + // a per-edit baseline at publishSections so the pre-publish check can tell + // whether the head advanced *since the edit was queued*, independent of the + // mutable watermark that a live event during the debounce window may advance. + private lastRemoteHead: PublishBaseline = { createdAt: 0, eventId: "" }; + // The canonical head this pending edit is racing against, frozen when the + // edit was queued (publishSections) and advanced ONLY by our own successful + // publishes. Freezing at queue time is what makes a genuine remote observed + // during the debounce window still adopt-worthy (pass-2): the mutable + // watermark advanced to that remote, but the baseline did not. Folding our + // own published head forward is what stops a newer edit from adopting an + // older generation's own accepted write (pass-3): our prior publish is our + // baseline, not a competing remote. + private publishBaseline: PublishBaseline = { createdAt: 0, eventId: "" }; private pendingStore: ChannelSectionStore | null = null; + // Monotonic id for the current pending edit. Every publishSections() bumps + // it; every scheduled publish/retry captures the value it was queued for. + // A completion (success, adopt, or no-op) may only clear pending state via + // compare-and-swap on this generation, so an older in-flight publish can + // never erase a newer edit that arrived while it was in flight. + private pendingGeneration = 0; + // Publish cycles are serialized: at most one runs at a time. A newer edit + // queued while a cycle is in flight does NOT start its own concurrent cycle; + // it defers, and the in-flight cycle's completion schedules it. Serialization + // guarantees there is never more than one baseline/fetch/publish sequence + // touching shared manager state, so a stale generation can never sign or + // publish after a newer edit exists. + private publishInFlight = false; private lastPublishedStore: ChannelSectionStore | null = null; private destroyed = false; + // Set by the hook so an adopted remote head (local edit lost LWW, or a relay + // conflict rejection) is written through to React state + localStorage. + private onRemoteAdopted: ((remote: RemoteSections) => void) | null = null; constructor(pubkey: string, relayUrl: string) { this.pubkey = pubkey; @@ -58,6 +163,11 @@ export class ChannelSectionSyncManager { this.lastRemoteCreatedAt = readWatermark(pubkey, BLOB_TYPE, relayUrl); } + /** Register the hook's adopt-remote sink (write-through to UI + storage). */ + setOnRemoteAdopted(cb: (remote: RemoteSections) => void): void { + this.onRemoteAdopted = cb; + } + async fetchRemoteSections(): Promise> { try { const events = await relayClient.fetchEvents({ @@ -73,7 +183,7 @@ export class ChannelSectionSyncManager { // An event exists — record its created_at regardless of whether we can // decrypt it, so seed-publish is blocked even when the payload is // unreadable (e.g. wrong key). - this.recordRemoteHead(event.created_at); + this.recordRemoteHead(event.created_at, event.id); const result = await decryptAndParse(event); if (!result) { return { status: "failed", createdAt: event.created_at }; @@ -89,11 +199,22 @@ export class ChannelSectionSyncManager { } } - /** Update in-memory + persisted watermark. */ - private recordRemoteHead(createdAt: number): void { + /** Update in-memory + persisted watermark and the canonical head tuple. */ + private recordRemoteHead(createdAt: number, eventId: string): void { if (createdAt > this.lastRemoteCreatedAt) { this.lastRemoteCreatedAt = createdAt; } + // Track the canonical-best head (`created_at DESC, id ASC`): a later head + // always wins; a same-second head wins only with a strictly-lower id. This + // mirrors the relay's stored winner so a frozen baseline reflects reality. + if ( + createdAt > this.lastRemoteHead.createdAt || + (createdAt === this.lastRemoteHead.createdAt && + (this.lastRemoteHead.eventId === "" || + eventId < this.lastRemoteHead.eventId)) + ) { + this.lastRemoteHead = { createdAt, eventId }; + } advanceWatermark(this.pubkey, BLOB_TYPE, this.relayUrl, createdAt); } @@ -102,26 +223,120 @@ export class ChannelSectionSyncManager { window.clearTimeout(this.debounceTimer); this.debounceTimer = null; } + if (this.retryTimer !== null) { + window.clearTimeout(this.retryTimer); + this.retryTimer = null; + } } getPendingStore(): ChannelSectionStore | null { return this.pendingStore; } + /** True while an unpublished local edit is queued (debouncing or retrying). */ + hasPendingEdit(): boolean { + return this.pendingStore !== null; + } + + /** + * Adopt a remote store that superseded a local edit: hand it to the hook for + * write-through, advance the watermark, and drop the losing pending edit — + * including the durable outbox, so the outbox can never replay an edit that + * adopt just decided lost (which would reintroduce divergence). + * + * Compare-and-swap on `gen`: this adopt was decided against the edit queued at + * generation `gen`. If a newer edit arrived while this publish was in flight, + * the generation has moved on and that newer edit is the latest writer — it + * will publish and win LWW — so a stale adopt must not clear its pending state + * or overwrite its optimistic UI. We still advance the watermark (monotonic + * and always safe) so the newer edit stamps above this head. + */ + private adoptRemote(remote: RemoteSections, gen: number): void { + this.recordRemoteHead(remote.createdAt, remote.eventId); + if (gen !== this.pendingGeneration) return; + this.pendingStore = null; + clearChannelSectionsOutbox(this.pubkey, this.relayUrl); + this.lastPublishedStore = remote.store; + if (this.destroyed) return; + this.onRemoteAdopted?.(remote); + } + + /** + * Clear the in-memory pending edit and its durable outbox — but only if the + * completing publish still owns the current generation. A publish for an + * older edit that finishes after a newer edit was queued must leave the newer + * edit (and its retry state) untouched. + */ + private discardPending(gen: number): void { + if (gen !== this.pendingGeneration) return; + this.pendingStore = null; + clearChannelSectionsOutbox(this.pubkey, this.relayUrl); + } + publishSections(store: ChannelSectionStore): void { this.pendingStore = store; + ++this.pendingGeneration; + // Freeze the canonical head this edit is racing against at queue time. The + // pre-publish check compares the fetched head against this baseline, not the + // mutable watermark — a live event applied during the debounce window + // advances the watermark to a new remote head, which would otherwise make + // the pre-publish comparison see equality and fall through to a publish that + // overwrites a remote that became head *after* this edit was queued. The + // baseline only advances via our own successful publishes (see doPublish), + // so a prior generation's own accepted write is folded in rather than + // mistaken for a competing remote. + this.publishBaseline = { ...this.lastRemoteHead }; + // Persist synchronously so an edit made <2s before quit/community-switch + // survives teardown and resumes on next mount (fix-3 durable outbox). + writeChannelSectionsOutbox(this.pubkey, store, this.relayUrl); if (this.debounceTimer !== null) { window.clearTimeout(this.debounceTimer); } + // A fresh edit supersedes any retry scheduled for the previous generation. + if (this.retryTimer !== null) { + window.clearTimeout(this.retryTimer); + this.retryTimer = null; + } + this.retryDelayMs = RETRY_BASE_MS; this.debounceTimer = window.setTimeout(() => { this.debounceTimer = null; - void this.doPublish(store); + this.startCycle(); }, DEBOUNCE_MS); } + /** + * Serialize publish cycles: at most one runs at a time. A debounce/retry + * timer that fires while a cycle is in flight defers — the in-flight cycle's + * completion re-drives if a pending edit still needs publishing. This kills + * the cross-generation race class by construction: a newer edit queued during + * a cycle cannot start its own concurrent cycle, so there is never more than + * one baseline/fetch/publish sequence competing over shared manager state. + */ + private startCycle(): void { + if (this.destroyed || this.pendingStore === null) return; + if (this.publishInFlight) return; + const store = this.pendingStore; + const gen = this.pendingGeneration; + this.publishInFlight = true; + void this.doPublish(store, gen).finally(() => { + this.publishInFlight = false; + // A newer edit queued during the cycle (or a cycle that ended without + // clearing its pending edit) still needs publishing and has no timer + // pending to drive it — drive the next cycle now that the lane is free. + if ( + !this.destroyed && + this.pendingStore !== null && + this.debounceTimer === null && + this.retryTimer === null + ) { + this.startCycle(); + } + }); + } + private async fetchOwnBlobBeforePublish( store: ChannelSectionStore, - ): Promise { + ): Promise { try { const events = await relayClient.fetchEvents({ kinds: [KIND_CHANNEL_SECTIONS], @@ -129,23 +344,25 @@ export class ChannelSectionSyncManager { "#d": [D_TAG], limit: 1, }); - if (events.length === 0 || events[0].pubkey !== this.pubkey) return store; + if (events.length === 0 || events[0].pubkey !== this.pubkey) + return { kind: "publish", store }; const event = events[0]; - // Snapshot the watermark before advancing it: after recordRemoteHead - // runs, lastRemoteCreatedAt equals event.created_at, so the LWW - // comparison remote.createdAt > lastRemoteCreatedAt would always be - // false and silently suppress the merge. - const headBeforeFetch = this.lastRemoteCreatedAt; - this.recordRemoteHead(event.created_at); const remote = await decryptAndParse(event); - if (!remote) return store; - // Sections use whole-blob LWW: take whichever is newer - if (remote.createdAt > headBeforeFetch) { - return remote.store; + // Record the head after decrypt attempt so the watermark/head-tuple + // advance even for an undecryptable payload. + this.recordRemoteHead(event.created_at, event.id); + if (!remote) return { kind: "publish", store }; + // Sections use whole-blob LWW. Compare the fetched head against the + // baseline frozen when this edit was queued — NOT the live watermark, + // which a passive live event during debounce may already have advanced to + // this same head. If the canonical head advanced since the edit began, the + // local edit lost and is adopted-away rather than republished over it. + if (remoteAdvancedSince(remote, this.publishBaseline)) { + return { kind: "adopt", remote }; } - return store; + return { kind: "publish", store }; } catch { - return store; + return { kind: "publish", store }; } } @@ -176,15 +393,45 @@ export class ChannelSectionSyncManager { return true; } - private async doPublish(store: ChannelSectionStore): Promise { + /** Schedule a bounded-backoff retry of the retained pending edit. */ + private scheduleRetry(gen: number): void { + if (this.destroyed || this.pendingStore === null) return; + // A newer edit has superseded this one; its own timer owns the retry. + if (gen !== this.pendingGeneration) return; + if (this.retryTimer !== null) return; + const delay = this.retryDelayMs; + this.retryDelayMs = Math.min(this.retryDelayMs * 2, RETRY_MAX_MS); + this.retryTimer = window.setTimeout(() => { + this.retryTimer = null; + this.startCycle(); + }, delay); + } + + private async doPublish( + store: ChannelSectionStore, + gen: number, + ): Promise { + // A newer edit was queued after this publish was scheduled; it owns the + // pending state and will publish the latest store — abandon this stale run. + if (gen !== this.pendingGeneration) return; try { - const merged = await this.fetchOwnBlobBeforePublish(store); + const decision = await this.fetchOwnBlobBeforePublish(store); // Guard: manager may have been destroyed while fetchOwnBlobBeforePublish // was awaited (community switch during in-flight fetch). If so, abort // before touching the relay. if (this.destroyed) return; + // A newer edit was queued while we awaited the pre-publish fetch. It owns + // convergence now; abort so we neither publish this stale store nor adopt + // over the newer pending edit. The serialized cycle re-drives for the + // newer generation once this one unwinds. + if (gen !== this.pendingGeneration) return; + if (decision.kind === "adopt") { + this.adoptRemote(decision.remote, gen); + return; + } + const merged = decision.store; if (this.isIdenticalToLastPublished(merged)) { - this.pendingStore = null; + this.discardPending(gen); return; } const payload = { @@ -193,9 +440,14 @@ export class ChannelSectionSyncManager { assignments: merged.assignments, }; const ciphertext = await nip44EncryptToSelf(JSON.stringify(payload)); - const createdAt = Math.max( - Math.floor(Date.now() / 1_000), - this.lastRemoteCreatedAt + 1, + const now = Math.floor(Date.now() / 1_000); + // Clamp inside the relay's future-drift window: never manufacture a + // timestamp so far ahead that this or a later publish is rejected for + // drift and wedges. If a skewed remote head sits beyond the window we + // will lose LWW and adopt it on conflict rather than walking past it. + const createdAt = Math.min( + Math.max(now, this.lastRemoteCreatedAt + 1), + now + MAX_PUBLISH_FUTURE_SECS, ); const event = await signRelayEvent({ kind: KIND_CHANNEL_SECTIONS, @@ -208,18 +460,56 @@ export class ChannelSectionSyncManager { }); // Final guard immediately before the network call — sign/encrypt are // synchronous-ish but cheap; the relay socket may have moved to a - // different community by the time we reach this point. - if (this.destroyed) return; + // different community by the time we reach this point, or a newer edit + // may have been queued during the encrypt/sign await (invariant: a stale + // generation never signs/publishes after a newer edit exists). + if (this.destroyed || gen !== this.pendingGeneration) return; await relayClient.publishEvent( event, "Timed out publishing channel sections.", "Failed to publish channel sections.", ); - this.recordRemoteHead(event.created_at); - this.lastPublishedStore = merged; - this.pendingStore = null; + this.recordRemoteHead(event.created_at, event.id); + // Fold our own accepted head into the pending edit's baseline. This is + // unconditional across generations: even a stale generation's own write, + // completing after a newer edit was queued, must advance the current + // pending baseline so the newer edit's pre-publish check does not mistake + // OUR prior publish for a competing remote and adopt it away (pass-3). + // Genuine remotes never fold in here — they only advance the watermark — + // so a remote that became head during the debounce window still adopts + // (pass-2). canonicalMax keeps the advance monotonic (`created_at DESC, + // id ASC`). + this.publishBaseline = canonicalMax(this.publishBaseline, { + createdAt: event.created_at, + eventId: event.id, + }); + // Only claim this store as the published head if it is still the current + // edit; a newer edit queued mid-flight owns lastPublishedStore now. + if (gen === this.pendingGeneration) { + this.lastPublishedStore = merged; + this.retryDelayMs = RETRY_BASE_MS; + } + this.discardPending(gen); } catch (error) { + if (this.destroyed) return; + // The relay rejects a strictly-losing coordinate write with an OK false + // conflict (fix-4). Treat it as a lost race: refetch the head and adopt + // it so we converge instead of retrying a write that can never win. + if (isConflictRejection(error)) { + const head = await this.fetchRemoteSections(); + if (this.destroyed) return; + if (head.status === "found") { + this.adoptRemote(head.data, gen); + } else { + this.scheduleRetry(gen); + } + return; + } + // Transient failure (timeout / socket error): keep the pending edit and + // retry with backoff rather than waiting for a reconnect that a healthy + // socket never fires. console.warn("[channelSectionsSync] publish failed:", error); + this.scheduleRetry(gen); } } @@ -237,7 +527,7 @@ export class ChannelSectionSyncManager { if (event.pubkey !== this.pubkey) return; // Record the raw head before decrypt so an undecryptable live event // still advances the watermark and blocks future seed-publish. - this.recordRemoteHead(event.created_at); + this.recordRemoteHead(event.created_at, event.id); void decryptAndParse(event).then((result) => { if (result) { onUpdate(result); @@ -265,12 +555,19 @@ export class ChannelSectionSyncManager { destroy(): void { // Cancel any pending publish and mark this manager as destroyed so any // in-flight doPublish() calls abort before reaching relayClient. - // Pending debounce-window changes are intentionally dropped: flushing - // could publish relay A's sections to relay B via the shared relayClient - // singleton. On return, bootstrap's found path whole-blob-replaces from - // remote, so any dropped pending edit is lost. + // Debounce-window changes are NOT lost: publishSections persisted them to + // the durable outbox synchronously, and the next mount resumes them. + // Flushing here is still avoided — it could publish relay A's sections to + // relay B via the shared relayClient singleton. this.destroyed = true; this.cancelPendingPublish(); this.pendingStore = null; } } + +/** True when a publish error is the relay's stale-coordinate conflict (fix-4). */ +function isConflictRejection(error: unknown): boolean { + return ( + error instanceof Error && error.message.toLowerCase().includes("conflict") + ); +} diff --git a/desktop/src/features/sidebar/lib/channelStarsStorage.ts b/desktop/src/features/sidebar/lib/channelStarsStorage.ts index 43c845cb3bd..2163dfe6356 100644 --- a/desktop/src/features/sidebar/lib/channelStarsStorage.ts +++ b/desktop/src/features/sidebar/lib/channelStarsStorage.ts @@ -111,6 +111,30 @@ export function writeChannelStarsStore( export function mergeStores( local: ChannelStarStore, remote: ChannelStarStore, +): ChannelStarStore { + return mergeStoresWithTie(local, remote, false); +} + +/** + * Merge a remote store that has already won the event-level canonical tie-break + * (`created_at DESC, id ASC`) into the local store, resolving a per-entry + * `updatedAt` tie in favour of the *remote* value. Once the comparator has + * chosen this remote event as the stored winner, its per-entry values must + * survive, or a stale value from a superseded larger-id event delivered first + * would win the merge and silently undo the canonical winner. Strictly-newer + * local per-entry edits (`l.updatedAt > r.updatedAt`) still win. + */ +export function mergeApplyingRemote( + local: ChannelStarStore, + remote: ChannelStarStore, +): ChannelStarStore { + return mergeStoresWithTie(local, remote, true); +} + +function mergeStoresWithTie( + local: ChannelStarStore, + remote: ChannelStarStore, + preferRemoteOnTie: boolean, ): ChannelStarStore { const allIds = new Set([ ...Object.keys(local.channels), @@ -121,7 +145,10 @@ export function mergeStores( const l = local.channels[id]; const r = remote.channels[id]; if (l && r) { - merged[id] = l.updatedAt >= r.updatedAt ? l : r; + const localWins = preferRemoteOnTie + ? l.updatedAt > r.updatedAt + : l.updatedAt >= r.updatedAt; + merged[id] = localWins ? l : r; } else { merged[id] = (l ?? r) as ChannelStarEntry; } diff --git a/desktop/src/features/sidebar/lib/useChannelMutes.test.mjs b/desktop/src/features/sidebar/lib/useChannelMutes.test.mjs index df41f403114..5691f4b41b8 100644 --- a/desktop/src/features/sidebar/lib/useChannelMutes.test.mjs +++ b/desktop/src/features/sidebar/lib/useChannelMutes.test.mjs @@ -77,3 +77,279 @@ test("same-second mute and unmute mutations survive at capacity", async () => { relayClient.subscribeToReconnects = originalSubscribeToReconnects; } }); + +// Equal-timestamp tie-break must match the relay's canonical winner +// (`created_at DESC, id ASC` → LOWEST id wins). Deliver the larger id first, +// then the lower id at the same timestamp; the lower id is the stored winner +// and must be applied, not rejected. Reverting applyRemote's `>=` back to `<=` +// wrongly ignores the lower id (the actual relay winner). +test("equal-timestamp tie-break applies the lower event id (relay canonical winner)", async () => { + const { act, cleanup, renderHook } = await import("@testing-library/react"); + const { relayClient } = await import("@/shared/api/relayClient"); + const { useChannelMutes } = await import("./useChannelMutes.ts"); + + const origFetch = relayClient.fetchEvents; + const origLive = relayClient.subscribeLive; + const origReconnect = relayClient.subscribeToReconnects; + const origTauri = window.__TAURI_INTERNALS__; + + let live = null; + relayClient.fetchEvents = async () => []; + relayClient.subscribeLive = async (_f, cb) => { + live = cb; + return async () => {}; + }; + relayClient.subscribeToReconnects = () => () => {}; + // Decrypt payload keyed off the event id embedded in the ciphertext so each + // delivered event yields a store muting a distinct channel we can assert on. + window.__TAURI_INTERNALS__ = { + invoke: (cmd, args) => { + if (cmd === "nip44_decrypt_from_self") { + const id = args?.ciphertext ?? ""; + return Promise.resolve( + JSON.stringify({ + version: 1, + channels: { [id]: { muted: true, updatedAt: 0 } }, + }), + ); + } + return Promise.reject(new Error(`unmocked ${cmd}`)); + }, + }; + + const pubkey = "pk-mute-tie"; + const relayUrl = "wss://r.tie"; + let hook = null; + try { + await act(async () => { + hook = renderHook(() => useChannelMutes(pubkey, relayUrl)); + await Promise.resolve(); + await Promise.resolve(); + await Promise.resolve(); + }); + assert.ok(live, "live subscription installed"); + + const deliver = async (id) => { + await act(async () => { + live({ + id, + pubkey, + created_at: 1000, + content: id, // decrypt echoes this into the muted channel id + kind: 30078, + tags: [["d", "channel-mutes"]], + sig: "s", + }); + await Promise.resolve(); + await Promise.resolve(); + }); + }; + + // Larger id first (applied), then the lower id at the same timestamp — the + // relay's canonical winner, which must NOT be rejected. + await deliver("bbbb"); + await deliver("aaaa"); + + assert.ok( + hook.result.current.mutedChannelIds.has("aaaa"), + "lower event id (relay canonical winner) must be applied, not rejected", + ); + hook.unmount(); + } finally { + cleanup(); + relayClient.fetchEvents = origFetch; + relayClient.subscribeLive = origLive; + relayClient.subscribeToReconnects = origReconnect; + window.__TAURI_INTERNALS__ = origTauri; + } +}); + +// Pass-2 finding 2: the comparator admitting the canonical lower id is +// necessary but not sufficient. Mutes are a per-entry store, so applyRemote +// merges the incoming blob into local state. On the SAME channel, a stale +// larger-id event delivered first (muted=true) must not survive the merge once +// the canonical lower-id winner (muted=false) arrives at the same entry +// `updatedAt`. Mutation: reverting the apply path to mergeStores (local/prev +// wins on tie) keeps the stale muted=true value. +test("canonical lower-id unmute replaces a stale larger-id mute at equal entry timestamp", async () => { + const { act, cleanup, renderHook } = await import("@testing-library/react"); + const { relayClient } = await import("@/shared/api/relayClient"); + const { useChannelMutes } = await import("./useChannelMutes.ts"); + + const origFetch = relayClient.fetchEvents; + const origLive = relayClient.subscribeLive; + const origReconnect = relayClient.subscribeToReconnects; + const origTauri = window.__TAURI_INTERNALS__; + + let live = null; + relayClient.fetchEvents = async () => []; + relayClient.subscribeLive = async (_f, cb) => { + live = cb; + return async () => {}; + }; + relayClient.subscribeToReconnects = () => () => {}; + // Both events target the SAME channel `shared` at the same entry updatedAt. + // The larger id `bbbb` says muted; the canonical lower id `aaaa` says not. + window.__TAURI_INTERNALS__ = { + invoke: (cmd, args) => { + if (cmd === "nip44_decrypt_from_self") { + const canonicalLowerId = args?.ciphertext === "aaaa"; + return Promise.resolve( + JSON.stringify({ + version: 1, + channels: { shared: { muted: !canonicalLowerId, updatedAt: 100 } }, + }), + ); + } + return Promise.reject(new Error(`unmocked ${cmd}`)); + }, + }; + + const pubkey = "pk-mute-shared-tie"; + const relayUrl = "wss://r.tie"; + let hook = null; + try { + await act(async () => { + hook = renderHook(() => useChannelMutes(pubkey, relayUrl)); + await Promise.resolve(); + await Promise.resolve(); + await Promise.resolve(); + }); + assert.ok(live, "live subscription installed"); + + const deliver = async (id) => { + await act(async () => { + live({ + id, + pubkey, + created_at: 1000, + content: id, + kind: 30078, + tags: [["d", "channel-mutes"]], + sig: "s", + }); + for (let i = 0; i < 20; i++) await Promise.resolve(); + }); + }; + + await deliver("bbbb"); // stale larger-id head says muted + await deliver("aaaa"); // canonical lower-id winner says unmuted + + assert.equal( + hook.result.current.mutedChannelIds.has("shared"), + false, + "canonical lower-id unmute must replace the stale larger-id mute", + ); + hook.unmount(); + } finally { + cleanup(); + relayClient.fetchEvents = origFetch; + relayClient.subscribeLive = origLive; + relayClient.subscribeToReconnects = origReconnect; + window.__TAURI_INTERNALS__ = origTauri; + } +}); + +// Fix round 3 (pass-3 finding 2): the remote-wins entry-tie merge and the +// pending-publish cancel apply ONLY to a canonical supersession (a lower id at +// the same event timestamp correcting an already-applied larger id). A plain +// live/bootstrap remote must NOT clobber a later same-second local click or +// cancel its pending publish. Entry `updatedAt` is whole seconds, so a click at +// 100.9s and an older remote entry at 100.1s both carry `updatedAt:100`; the +// later local intent must win and keep publishing. Mutation: applying +// mergeApplyingRemote + cancel unconditionally lets the delayed remote overwrite +// the click and drop its publish. +test("delayed same-second remote does not clobber a later local mute or cancel its publish", async () => { + const { act, cleanup, renderHook } = await import("@testing-library/react"); + const { relayClient } = await import("@/shared/api/relayClient"); + const { useChannelMutes } = await import("./useChannelMutes.ts"); + + const origFetch = relayClient.fetchEvents; + const origLive = relayClient.subscribeLive; + const origReconnect = relayClient.subscribeToReconnects; + const origTauri = window.__TAURI_INTERNALS__; + const origSetTimeout = window.setTimeout; + const origClearTimeout = window.clearTimeout; + const origDateNow = Date.now; + + const timers = new Map(); + let nextTimer = 1; + window.setTimeout = (fn, ms) => { + const id = nextTimer++; + timers.set(id, { fn, ms }); + return id; + }; + window.clearTimeout = (id) => timers.delete(id); + // The local click happens later within second 100. + Date.now = () => 100_900; + + let live = null; + relayClient.fetchEvents = async () => []; + relayClient.subscribeLive = async (_f, cb) => { + live = cb; + return async () => {}; + }; + relayClient.subscribeToReconnects = () => () => {}; + // The delayed remote entry sits earlier in the same second and says unmuted. + window.__TAURI_INTERNALS__ = { + invoke: (cmd) => { + if (cmd === "nip44_decrypt_from_self") + return Promise.resolve( + JSON.stringify({ + version: 1, + channels: { shared: { muted: false, updatedAt: 100 } }, + }), + ); + return Promise.reject(new Error(`unmocked ${cmd}`)); + }, + }; + + const pubkey = "pk-mute-same-second"; + const relayUrl = "wss://r.same"; + let hook = null; + try { + await act(async () => { + hook = renderHook(() => useChannelMutes(pubkey, relayUrl)); + for (let i = 0; i < 20; i++) await Promise.resolve(); + }); + assert.ok(live, "live subscription installed"); + + // Local optimistic click: muted=true at updatedAt=100 (Date.now=100.9s). + await act(async () => { + hook.result.current.muteChannel("shared"); + }); + // An older remote entry from the same second decrypts and applies late. + await act(async () => { + live({ + id: "remote-before-click", + pubkey, + created_at: 100, + content: "remote", + kind: 30078, + tags: [["d", "channel-mutes"]], + sig: "s", + }); + for (let i = 0; i < 40; i++) await Promise.resolve(); + }); + + assert.equal( + hook.result.current.mutedChannelIds.has("shared"), + true, + "a later same-second local click must survive a delayed older remote", + ); + assert.ok( + [...timers.values()].some((t) => t.ms === 2000), + "the local pending publish must remain scheduled", + ); + hook.unmount(); + } finally { + cleanup(); + relayClient.fetchEvents = origFetch; + relayClient.subscribeLive = origLive; + relayClient.subscribeToReconnects = origReconnect; + window.__TAURI_INTERNALS__ = origTauri; + window.setTimeout = origSetTimeout; + window.clearTimeout = origClearTimeout; + Date.now = origDateNow; + } +}); diff --git a/desktop/src/features/sidebar/lib/useChannelMutes.ts b/desktop/src/features/sidebar/lib/useChannelMutes.ts index 20b57453254..6f87a77b49f 100644 --- a/desktop/src/features/sidebar/lib/useChannelMutes.ts +++ b/desktop/src/features/sidebar/lib/useChannelMutes.ts @@ -4,6 +4,7 @@ import { relayClient } from "@/shared/api/relayClient"; import { boundMuteStore, DEFAULT_STORE, + mergeApplyingRemote, mergeStores, mutedChannelIdsFromStore, readChannelMutesStore, @@ -73,15 +74,36 @@ export function useChannelMutes( return (prev) => { if (!pubkey) return prev; if (remote.createdAt < lastAppliedRemoteTs.current) return prev; + // Equal timestamps: the relay/database break ties by `id ASC` — the + // LOWEST event id is the canonical winner. Apply a strictly-lower id and + // ignore any id >= the last applied, so the UI converges on the same + // event the relay stored rather than the largest id seen. if ( remote.createdAt === lastAppliedRemoteTs.current && - remote.eventId <= lastAppliedEventId.current + remote.eventId >= lastAppliedEventId.current ) return prev; + // A canonical supersession corrects an already-applied same-timestamp + // LARGER-id head with the true winner: only here may the incoming blob's + // per-entry values win an equal-`updatedAt` tie, and only here does the + // pending publish (which reflected the superseded head) get cancelled. + // Any other application (bootstrap / live / newer timestamp) merges over + // optimistic local state with local-wins `mergeStores` and must NOT + // cancel a pending local publish — otherwise a later same-second local + // click (integer-second `updatedAt`) loses to an older remote entry that + // decrypts late, and its publish is silently dropped. + const isCanonicalSupersession = + remote.createdAt === lastAppliedRemoteTs.current && + lastAppliedEventId.current !== "" && + remote.eventId < lastAppliedEventId.current; lastAppliedRemoteTs.current = remote.createdAt; lastAppliedEventId.current = remote.eventId; - managerRef.current?.cancelPendingMutePublish(); - const merged = mergeStores(prev, remote.store); + const merged = isCanonicalSupersession + ? mergeApplyingRemote(prev, remote.store) + : mergeStores(prev, remote.store); + if (isCanonicalSupersession) { + managerRef.current?.cancelPendingMutePublish(); + } if (!writeChannelMutesStore(pubkey, merged)) return prev; return merged; }; diff --git a/desktop/src/features/sidebar/lib/useChannelSections.test.mjs b/desktop/src/features/sidebar/lib/useChannelSections.test.mjs index 401b59d9c1c..62030348624 100644 --- a/desktop/src/features/sidebar/lib/useChannelSections.test.mjs +++ b/desktop/src/features/sidebar/lib/useChannelSections.test.mjs @@ -76,3 +76,200 @@ test("assignChannel refreshes an existing assignment before the next eviction", relayClient.subscribeToReconnects = originalSubscribeToReconnects; } }); + +// Fix 2 regression: a live remote arriving while a local edit is pending must +// NOT overwrite the optimistic edit or strand its durable outbox. The pending +// edit's own debounced publish owns convergence (publish-or-adopt). Reverting +// applyRemote's hasPendingEdit guard makes the live event clobber the UI and +// leave the outbox replay-eligible. +test("live remote while a local edit is pending defers to the pending edit", async () => { + const { act, cleanup, renderHook } = await import("@testing-library/react"); + const { relayClient } = await import("@/shared/api/relayClient"); + const { useChannelSections } = await import("./useChannelSections.ts"); + + const origFetch = relayClient.fetchEvents; + const origLive = relayClient.subscribeLive; + const origReconnect = relayClient.subscribeToReconnects; + const origPublish = relayClient.publishEvent; + const origTauri = window.__TAURI_INTERNALS__; + + let live = null; + relayClient.fetchEvents = async () => []; + relayClient.subscribeLive = async (_f, cb) => { + live = cb; + return async () => {}; + }; + relayClient.subscribeToReconnects = () => () => {}; + relayClient.publishEvent = async () => {}; + window.__TAURI_INTERNALS__ = { + invoke: (cmd) => { + if (cmd === "nip44_decrypt_from_self") + return Promise.resolve( + JSON.stringify({ + version: 1, + sections: [{ id: "remote", name: "Remote", order: 0 }], + assignments: {}, + }), + ); + if (cmd === "nip44_encrypt_to_self") return Promise.resolve("ct"); + if (cmd === "sign_event") + return Promise.resolve( + JSON.stringify({ + id: "signed", + pubkey: "pk-live-pending", + content: "ct", + created_at: 0, + kind: 30078, + tags: [], + sig: "s", + }), + ); + return Promise.reject(new Error(`unmocked ${cmd}`)); + }, + }; + + const pubkey = "pk-live-pending"; + const relayUrl = "wss://r.live"; + const outboxKey = `buzz-channel-sections-outbox.v1:${pubkey}:${encodeURIComponent(relayUrl)}`; + + let hook = null; + try { + await act(async () => { + hook = renderHook(() => useChannelSections(pubkey, relayUrl)); + await Promise.resolve(); + await Promise.resolve(); + await Promise.resolve(); + }); + assert.ok(live, "live subscription installed"); + + // Make a local edit — it becomes the pending store and persists to outbox. + await act(async () => { + hook.result.current.createSection("Local"); + }); + assert.ok( + window.localStorage.getItem(outboxKey), + "local edit persisted to outbox", + ); + const localSectionIds = hook.result.current.sections.map((s) => s.id); + + // A remote live event arrives while the edit is still pending. + await act(async () => { + live({ + id: "remote-event", + pubkey, + created_at: 500, + content: "cipher", + kind: 30078, + tags: [["d", "channel-sections"]], + sig: "s", + }); + await Promise.resolve(); + await Promise.resolve(); + }); + + assert.deepEqual( + hook.result.current.sections.map((s) => s.id), + localSectionIds, + "pending local edit must NOT be overwritten by the live remote", + ); + assert.ok( + window.localStorage.getItem(outboxKey), + "outbox for the pending edit must survive the live remote", + ); + hook.unmount(); + } finally { + cleanup(); + relayClient.fetchEvents = origFetch; + relayClient.subscribeLive = origLive; + relayClient.subscribeToReconnects = origReconnect; + relayClient.publishEvent = origPublish; + window.__TAURI_INTERNALS__ = origTauri; + } +}); + +// Fix 3 regression: equal-timestamp tie-break must match the relay's canonical +// winner (`created_at DESC, id ASC` → LOWEST id wins). Deliver the larger id +// first, then the lower id at the same timestamp; the lower-id store must win. +// Reverting applyRemote's `>=` back to `<=` converges on the larger id instead. +test("equal-timestamp tie-break applies the lower event id (relay canonical winner)", async () => { + const { act, cleanup, renderHook } = await import("@testing-library/react"); + const { relayClient } = await import("@/shared/api/relayClient"); + const { useChannelSections } = await import("./useChannelSections.ts"); + + const origFetch = relayClient.fetchEvents; + const origLive = relayClient.subscribeLive; + const origReconnect = relayClient.subscribeToReconnects; + const origTauri = window.__TAURI_INTERNALS__; + + let live = null; + relayClient.fetchEvents = async () => []; + relayClient.subscribeLive = async (_f, cb) => { + live = cb; + return async () => {}; + }; + relayClient.subscribeToReconnects = () => () => {}; + // Decrypt payload keyed off the event id embedded in the ciphertext so each + // delivered event yields a distinct store we can assert on. + window.__TAURI_INTERNALS__ = { + invoke: (cmd, args) => { + if (cmd === "nip44_decrypt_from_self") { + const id = args?.ciphertext ?? ""; + return Promise.resolve( + JSON.stringify({ + version: 1, + sections: [{ id, name: id, order: 0 }], + assignments: {}, + }), + ); + } + return Promise.reject(new Error(`unmocked ${cmd}`)); + }, + }; + + const pubkey = "pk-tie"; + const relayUrl = "wss://r.tie"; + let hook = null; + try { + await act(async () => { + hook = renderHook(() => useChannelSections(pubkey, relayUrl)); + await Promise.resolve(); + await Promise.resolve(); + await Promise.resolve(); + }); + assert.ok(live, "live subscription installed"); + + const deliver = async (id) => { + await act(async () => { + live({ + id, + pubkey, + created_at: 1000, + content: id, // decrypt echoes this into the section id + kind: 30078, + tags: [["d", "channel-sections"]], + sig: "s", + }); + await Promise.resolve(); + await Promise.resolve(); + }); + }; + + // Larger id first (would win under the old <= comparator)... + await deliver("bbbb"); + // ...then the lower id at the same timestamp — the relay's canonical winner. + await deliver("aaaa"); + + assert.deepEqual( + hook.result.current.sections.map((s) => s.id), + ["aaaa"], + "lower event id must win the equal-timestamp tie-break", + ); + hook.unmount(); + } finally { + cleanup(); + relayClient.fetchEvents = origFetch; + relayClient.subscribeLive = origLive; + relayClient.subscribeToReconnects = origReconnect; + window.__TAURI_INTERNALS__ = origTauri; + } +}); diff --git a/desktop/src/features/sidebar/lib/useChannelSections.ts b/desktop/src/features/sidebar/lib/useChannelSections.ts index 5a544e82bca..01b3187096b 100644 --- a/desktop/src/features/sidebar/lib/useChannelSections.ts +++ b/desktop/src/features/sidebar/lib/useChannelSections.ts @@ -3,7 +3,9 @@ import * as React from "react"; import { relayClient } from "@/shared/api/relayClient"; import { boundChannelSectionsStore, + clearChannelSectionsOutbox, DEFAULT_STORE, + readChannelSectionsOutbox, readChannelSectionsStore, storageKey, writeChannelSectionsStore, @@ -19,6 +21,13 @@ import type { ChannelSectionStore, } from "./channelSectionsStorage"; +// Reconciliation cadence (fix 1). Steady interval re-fetches the head on a +// healthy socket so divergence self-heals without a reconnect; the retry +// window backs off from base to max while the fetch keeps failing. +const RECONCILE_STEADY_MS = 60_000; +const RECONCILE_RETRY_BASE_MS = 3_000; +const RECONCILE_RETRY_MAX_MS = 60_000; + export function useChannelSections( pubkey: string | undefined, relayUrl?: string, @@ -85,15 +94,26 @@ export function useChannelSections( ): ((prev: ChannelSectionStore) => ChannelSectionStore) => { return (prev) => { if (!pubkey) return prev; + // A pending local edit owns convergence: its debounced publish + // re-checks the head and either wins (publish) or loses (adopt, which + // routes back through onRemoteAdopted with pending already cleared). + // Never let a passive remote arrival clobber the optimistic edit or + // strand its durable outbox — that is the one-convergence-mechanism + // invariant. The adopt path clears pending before calling us, so this + // guard is false there and the winning remote still writes through. + if (managerRef.current?.hasPendingEdit()) return prev; if (remote.createdAt < lastAppliedRemoteTs.current) return prev; + // Equal timestamps: the relay/database break ties by `id ASC` — the + // LOWEST event id is the canonical winner. Apply a strictly-lower id and + // ignore any id >= the last applied, so the UI converges on the same + // event the relay stored rather than the largest id seen. if ( remote.createdAt === lastAppliedRemoteTs.current && - remote.eventId <= lastAppliedEventId.current + remote.eventId >= lastAppliedEventId.current ) return prev; lastAppliedRemoteTs.current = remote.createdAt; lastAppliedEventId.current = remote.eventId; - managerRef.current?.cancelPendingPublish(); if (!writeChannelSectionsStore(pubkey, remote.store, relayUrl)) return prev; return remote.store; @@ -102,6 +122,19 @@ export function useChannelSections( [pubkey, relayUrl], ); + React.useEffect(() => { + if (!pubkey || !relayUrl) return; + const manager = managerRef.current; + if (!manager) return; + // When a local edit loses whole-blob LWW (pre-publish head is newer) or the + // relay rejects it with a conflict, the manager adopts the winning remote + // store. Write it through to React state + localStorage so the UI and relay + // never diverge; applyRemote also advances the applied-ts guard. + manager.setOnRemoteAdopted((remote) => { + setStore(applyRemote(remote)); + }); + }, [pubkey, relayUrl, applyRemote]); + React.useEffect(() => { if (!pubkey || !relayUrl) return; let cancelled = false; @@ -112,13 +145,71 @@ export function useChannelSections( setStore(applyRemote(result.data)); } // "hold": seed already performed by bootstrap (if first-sync), or - // blocked (failed fetch / prior watermark). Hook does nothing. + // blocked (failed fetch / prior watermark). The reconciliation effect + // below retries a failed fetch; here we only resume any edit that was + // persisted to the durable outbox before a prior quit/community-switch. + const outbox = readChannelSectionsOutbox(pubkey, relayUrl); + if (outbox) { + managerRef.current?.publishSections(outbox); + } else { + clearChannelSectionsOutbox(pubkey, relayUrl); + } }); return () => { cancelled = true; }; }, [pubkey, relayUrl, applyRemote]); + // Reconciliation loop (fix 1): a single scheduler that both retries a failed + // bootstrap with bounded backoff and periodically re-fetches the head, so + // stale-at-open state converges without waiting for a reconnect event a + // healthy socket never fires. Also refreshes when the window becomes visible. + React.useEffect(() => { + if (!pubkey || !relayUrl) return; + let cancelled = false; + let timer: number | null = null; + let delayMs = RECONCILE_RETRY_BASE_MS; + + const schedule = (ms: number) => { + if (cancelled) return; + if (timer !== null) window.clearTimeout(timer); + timer = window.setTimeout(tick, ms); + }; + + const tick = () => { + void managerRef.current?.fetchRemoteSections().then((result) => { + if (cancelled) return; + if (result.status === "found") { + // applyRemote defers to a pending local edit (whose own debounced + // publish converges via publish-or-adopt), so a periodic reconcile + // can never drop it — no re-queue needed. + setStore(applyRemote(result.data)); + delayMs = RECONCILE_STEADY_MS; // relay answered → steady cadence + } else if (result.status === "absent") { + delayMs = RECONCILE_STEADY_MS; // answered (no blob) → steady cadence + } else { + delayMs = Math.min(delayMs * 2, RECONCILE_RETRY_MAX_MS); // fetch failed → back off + } + schedule(delayMs); + }); + }; + + const onVisible = () => { + if (document.visibilityState === "visible") { + delayMs = RECONCILE_RETRY_BASE_MS; + tick(); + } + }; + document.addEventListener("visibilitychange", onVisible); + schedule(delayMs); + + return () => { + cancelled = true; + if (timer !== null) window.clearTimeout(timer); + document.removeEventListener("visibilitychange", onVisible); + }; + }, [pubkey, relayUrl, applyRemote]); + React.useEffect(() => { if (!pubkey) return; let unsub: (() => Promise) | null = null; diff --git a/desktop/src/features/sidebar/lib/useChannelSortPreference.test.mjs b/desktop/src/features/sidebar/lib/useChannelSortPreference.test.mjs new file mode 100644 index 00000000000..a9a65754b0c --- /dev/null +++ b/desktop/src/features/sidebar/lib/useChannelSortPreference.test.mjs @@ -0,0 +1,110 @@ +import assert from "node:assert/strict"; +import { after, before, test } from "node:test"; + +import { JSDOM } from "jsdom"; + +const dom = new JSDOM("", { + url: "http://localhost", +}); + +before(() => { + Object.assign(globalThis, { + document: dom.window.document, + HTMLElement: dom.window.HTMLElement, + IS_REACT_ACT_ENVIRONMENT: true, + window: dom.window, + }); +}); + +after(() => dom.window.close()); + +// Equal-timestamp tie-break must match the relay's canonical winner +// (`created_at DESC, id ASC` → LOWEST id wins). Deliver the larger id first, +// then the lower id at the same timestamp; the lower id is the stored winner +// and its whole-blob store must replace the applied state. Reverting +// applyRemote's `>=` back to `<=` wrongly ignores the lower id (the relay winner). +test("equal-timestamp tie-break applies the lower event id (relay canonical winner)", async () => { + const { act, cleanup, renderHook } = await import("@testing-library/react"); + const { relayClient } = await import("@/shared/api/relayClient"); + const { useChannelSortPreference } = await import( + "./useChannelSortPreference.ts" + ); + + const origFetch = relayClient.fetchEvents; + const origLive = relayClient.subscribeLive; + const origReconnect = relayClient.subscribeToReconnects; + const origTauri = window.__TAURI_INTERNALS__; + + let live = null; + relayClient.fetchEvents = async () => []; + relayClient.subscribeLive = async (_f, cb) => { + live = cb; + return async () => {}; + }; + relayClient.subscribeToReconnects = () => () => {}; + // Decrypt payload keyed off the event id embedded in the ciphertext so each + // delivered event yields a store setting a distinct group's mode to "recent". + window.__TAURI_INTERNALS__ = { + invoke: (cmd, args) => { + if (cmd === "nip44_decrypt_from_self") { + const id = args?.ciphertext ?? ""; + return Promise.resolve( + JSON.stringify({ version: 1, groups: { [id]: "recent" } }), + ); + } + return Promise.reject(new Error(`unmocked ${cmd}`)); + }, + }; + + const pubkey = "pk-sort-tie"; + const relayUrl = "wss://r.tie"; + let hook = null; + try { + await act(async () => { + hook = renderHook(() => useChannelSortPreference(pubkey, relayUrl)); + await Promise.resolve(); + await Promise.resolve(); + await Promise.resolve(); + }); + assert.ok(live, "live subscription installed"); + + const deliver = async (id) => { + await act(async () => { + live({ + id, + pubkey, + created_at: 1000, + content: id, // decrypt echoes this into the group key + kind: 30078, + tags: [["d", "channel-sort"]], + sig: "s", + }); + await Promise.resolve(); + await Promise.resolve(); + }); + }; + + // Larger id first (applied), then the lower id at the same timestamp — the + // relay's canonical winner, whose whole-blob store must replace the state. + await deliver("bbbb"); + await deliver("aaaa"); + + assert.equal( + hook.result.current.sortModeFor("aaaa"), + "recent", + "lower event id (relay canonical winner) must be applied, not rejected", + ); + assert.equal( + hook.result.current.sortModeFor("bbbb"), + "alpha", + "larger id's store must be superseded by the lower-id whole-blob winner", + ); + hook.unmount(); + } finally { + cleanup(); + relayClient.fetchEvents = origFetch; + relayClient.subscribeLive = origLive; + relayClient.subscribeToReconnects = origReconnect; + window.__TAURI_INTERNALS__ = origTauri; + } +}); diff --git a/desktop/src/features/sidebar/lib/useChannelSortPreference.ts b/desktop/src/features/sidebar/lib/useChannelSortPreference.ts index 85c07b1c398..a692bbb31a4 100644 --- a/desktop/src/features/sidebar/lib/useChannelSortPreference.ts +++ b/desktop/src/features/sidebar/lib/useChannelSortPreference.ts @@ -86,9 +86,13 @@ export function useChannelSortPreference( return (prev) => { if (!pubkey) return prev; if (remote.createdAt < lastAppliedRemoteTs.current) return prev; + // Equal timestamps: the relay/database break ties by `id ASC` — the + // LOWEST event id is the canonical winner. Apply a strictly-lower id and + // ignore any id >= the last applied, so the UI converges on the same + // event the relay stored rather than the largest id seen. if ( remote.createdAt === lastAppliedRemoteTs.current && - remote.eventId <= lastAppliedEventId.current + remote.eventId >= lastAppliedEventId.current ) return prev; lastAppliedRemoteTs.current = remote.createdAt; diff --git a/desktop/src/features/sidebar/lib/useChannelStars.test.mjs b/desktop/src/features/sidebar/lib/useChannelStars.test.mjs index a8e1b4b4ae5..e2dcf8349b0 100644 --- a/desktop/src/features/sidebar/lib/useChannelStars.test.mjs +++ b/desktop/src/features/sidebar/lib/useChannelStars.test.mjs @@ -77,3 +77,281 @@ test("same-second star and unstar mutations survive at capacity", async () => { relayClient.subscribeToReconnects = originalSubscribeToReconnects; } }); + +// Equal-timestamp tie-break must match the relay's canonical winner +// (`created_at DESC, id ASC` → LOWEST id wins). Deliver the larger id first, +// then the lower id at the same timestamp; the lower id is the stored winner +// and must be applied, not rejected. Reverting applyRemote's `>=` back to `<=` +// wrongly ignores the lower id (the actual relay winner). +test("equal-timestamp tie-break applies the lower event id (relay canonical winner)", async () => { + const { act, cleanup, renderHook } = await import("@testing-library/react"); + const { relayClient } = await import("@/shared/api/relayClient"); + const { useChannelStars } = await import("./useChannelStars.ts"); + + const origFetch = relayClient.fetchEvents; + const origLive = relayClient.subscribeLive; + const origReconnect = relayClient.subscribeToReconnects; + const origTauri = window.__TAURI_INTERNALS__; + + let live = null; + relayClient.fetchEvents = async () => []; + relayClient.subscribeLive = async (_f, cb) => { + live = cb; + return async () => {}; + }; + relayClient.subscribeToReconnects = () => () => {}; + // Decrypt payload keyed off the event id embedded in the ciphertext so each + // delivered event yields a store starring a distinct channel we can assert on. + window.__TAURI_INTERNALS__ = { + invoke: (cmd, args) => { + if (cmd === "nip44_decrypt_from_self") { + const id = args?.ciphertext ?? ""; + return Promise.resolve( + JSON.stringify({ + version: 1, + channels: { [id]: { starred: true, updatedAt: 0 } }, + }), + ); + } + return Promise.reject(new Error(`unmocked ${cmd}`)); + }, + }; + + const pubkey = "pk-star-tie"; + const relayUrl = "wss://r.tie"; + let hook = null; + try { + await act(async () => { + hook = renderHook(() => useChannelStars(pubkey, relayUrl)); + await Promise.resolve(); + await Promise.resolve(); + await Promise.resolve(); + }); + assert.ok(live, "live subscription installed"); + + const deliver = async (id) => { + await act(async () => { + live({ + id, + pubkey, + created_at: 1000, + content: id, // decrypt echoes this into the starred channel id + kind: 30078, + tags: [["d", "channel-stars"]], + sig: "s", + }); + await Promise.resolve(); + await Promise.resolve(); + }); + }; + + // Larger id first (applied), then the lower id at the same timestamp — the + // relay's canonical winner, which must NOT be rejected. + await deliver("bbbb"); + await deliver("aaaa"); + + assert.ok( + hook.result.current.starredChannelIds.has("aaaa"), + "lower event id (relay canonical winner) must be applied, not rejected", + ); + hook.unmount(); + } finally { + cleanup(); + relayClient.fetchEvents = origFetch; + relayClient.subscribeLive = origLive; + relayClient.subscribeToReconnects = origReconnect; + window.__TAURI_INTERNALS__ = origTauri; + } +}); + +// Pass-2 finding 2: the comparator admitting the canonical lower id is +// necessary but not sufficient. Stars are a per-entry store, so applyRemote +// merges the incoming blob into local state. On the SAME channel, a stale +// larger-id event delivered first (starred=true) must not survive the merge +// once the canonical lower-id winner (starred=false) arrives at the same entry +// `updatedAt`. Mutation: reverting the apply path to mergeStores (local/prev +// wins on tie) keeps the stale starred=true value. +test("canonical lower-id unstar replaces a stale larger-id star at equal entry timestamp", async () => { + const { act, cleanup, renderHook } = await import("@testing-library/react"); + const { relayClient } = await import("@/shared/api/relayClient"); + const { useChannelStars } = await import("./useChannelStars.ts"); + + const origFetch = relayClient.fetchEvents; + const origLive = relayClient.subscribeLive; + const origReconnect = relayClient.subscribeToReconnects; + const origTauri = window.__TAURI_INTERNALS__; + + let live = null; + relayClient.fetchEvents = async () => []; + relayClient.subscribeLive = async (_f, cb) => { + live = cb; + return async () => {}; + }; + relayClient.subscribeToReconnects = () => () => {}; + // Both events target the SAME channel `shared` at the same entry updatedAt. + // The larger id `bbbb` says starred; the canonical lower id `aaaa` says not. + window.__TAURI_INTERNALS__ = { + invoke: (cmd, args) => { + if (cmd === "nip44_decrypt_from_self") { + const canonicalLowerId = args?.ciphertext === "aaaa"; + return Promise.resolve( + JSON.stringify({ + version: 1, + channels: { + shared: { starred: !canonicalLowerId, updatedAt: 100 }, + }, + }), + ); + } + return Promise.reject(new Error(`unmocked ${cmd}`)); + }, + }; + + const pubkey = "pk-star-shared-tie"; + const relayUrl = "wss://r.tie"; + let hook = null; + try { + await act(async () => { + hook = renderHook(() => useChannelStars(pubkey, relayUrl)); + await Promise.resolve(); + await Promise.resolve(); + await Promise.resolve(); + }); + assert.ok(live, "live subscription installed"); + + const deliver = async (id) => { + await act(async () => { + live({ + id, + pubkey, + created_at: 1000, + content: id, + kind: 30078, + tags: [["d", "channel-stars"]], + sig: "s", + }); + for (let i = 0; i < 20; i++) await Promise.resolve(); + }); + }; + + await deliver("bbbb"); // stale larger-id head says starred + await deliver("aaaa"); // canonical lower-id winner says unstarred + + assert.equal( + hook.result.current.starredChannelIds.has("shared"), + false, + "canonical lower-id unstar must replace the stale larger-id star", + ); + hook.unmount(); + } finally { + cleanup(); + relayClient.fetchEvents = origFetch; + relayClient.subscribeLive = origLive; + relayClient.subscribeToReconnects = origReconnect; + window.__TAURI_INTERNALS__ = origTauri; + } +}); + +// Fix round 3 (pass-3 finding 2): the remote-wins entry-tie merge and the +// pending-publish cancel apply ONLY to a canonical supersession (a lower id at +// the same event timestamp correcting an already-applied larger id). A plain +// live/bootstrap remote must NOT clobber a later same-second local click or +// cancel its pending publish. Entry `updatedAt` is whole seconds, so a click at +// 100.9s and an older remote entry at 100.1s both carry `updatedAt:100`; the +// later local intent must win and keep publishing. Mutation: applying +// mergeApplyingRemote + cancel unconditionally lets the delayed remote overwrite +// the click and drop its publish. +test("delayed same-second remote does not clobber a later local star or cancel its publish", async () => { + const { act, cleanup, renderHook } = await import("@testing-library/react"); + const { relayClient } = await import("@/shared/api/relayClient"); + const { useChannelStars } = await import("./useChannelStars.ts"); + + const origFetch = relayClient.fetchEvents; + const origLive = relayClient.subscribeLive; + const origReconnect = relayClient.subscribeToReconnects; + const origTauri = window.__TAURI_INTERNALS__; + const origSetTimeout = window.setTimeout; + const origClearTimeout = window.clearTimeout; + const origDateNow = Date.now; + + const timers = new Map(); + let nextTimer = 1; + window.setTimeout = (fn, ms) => { + const id = nextTimer++; + timers.set(id, { fn, ms }); + return id; + }; + window.clearTimeout = (id) => timers.delete(id); + // The local click happens later within second 100. + Date.now = () => 100_900; + + let live = null; + relayClient.fetchEvents = async () => []; + relayClient.subscribeLive = async (_f, cb) => { + live = cb; + return async () => {}; + }; + relayClient.subscribeToReconnects = () => () => {}; + // The delayed remote entry sits earlier in the same second and says unstarred. + window.__TAURI_INTERNALS__ = { + invoke: (cmd) => { + if (cmd === "nip44_decrypt_from_self") + return Promise.resolve( + JSON.stringify({ + version: 1, + channels: { shared: { starred: false, updatedAt: 100 } }, + }), + ); + return Promise.reject(new Error(`unmocked ${cmd}`)); + }, + }; + + const pubkey = "pk-star-same-second"; + const relayUrl = "wss://r.same"; + let hook = null; + try { + await act(async () => { + hook = renderHook(() => useChannelStars(pubkey, relayUrl)); + for (let i = 0; i < 20; i++) await Promise.resolve(); + }); + assert.ok(live, "live subscription installed"); + + // Local optimistic click: starred=true at updatedAt=100 (Date.now=100.9s). + await act(async () => { + hook.result.current.starChannel("shared"); + }); + // An older remote entry from the same second decrypts and applies late. + await act(async () => { + live({ + id: "remote-before-click", + pubkey, + created_at: 100, + content: "remote", + kind: 30078, + tags: [["d", "channel-stars"]], + sig: "s", + }); + for (let i = 0; i < 40; i++) await Promise.resolve(); + }); + + assert.equal( + hook.result.current.starredChannelIds.has("shared"), + true, + "a later same-second local click must survive a delayed older remote", + ); + assert.ok( + [...timers.values()].some((t) => t.ms === 2000), + "the local pending publish must remain scheduled", + ); + hook.unmount(); + } finally { + cleanup(); + relayClient.fetchEvents = origFetch; + relayClient.subscribeLive = origLive; + relayClient.subscribeToReconnects = origReconnect; + window.__TAURI_INTERNALS__ = origTauri; + window.setTimeout = origSetTimeout; + window.clearTimeout = origClearTimeout; + Date.now = origDateNow; + } +}); diff --git a/desktop/src/features/sidebar/lib/useChannelStars.ts b/desktop/src/features/sidebar/lib/useChannelStars.ts index 855c8de8581..a3da782c607 100644 --- a/desktop/src/features/sidebar/lib/useChannelStars.ts +++ b/desktop/src/features/sidebar/lib/useChannelStars.ts @@ -4,6 +4,7 @@ import { relayClient } from "@/shared/api/relayClient"; import { boundStarStore, DEFAULT_STORE, + mergeApplyingRemote, mergeStores, readChannelStarsStore, starredChannelIdsFromStore, @@ -73,15 +74,36 @@ export function useChannelStars( return (prev) => { if (!pubkey) return prev; if (remote.createdAt < lastAppliedRemoteTs.current) return prev; + // Equal timestamps: the relay/database break ties by `id ASC` — the + // LOWEST event id is the canonical winner. Apply a strictly-lower id and + // ignore any id >= the last applied, so the UI converges on the same + // event the relay stored rather than the largest id seen. if ( remote.createdAt === lastAppliedRemoteTs.current && - remote.eventId <= lastAppliedEventId.current + remote.eventId >= lastAppliedEventId.current ) return prev; + // A canonical supersession corrects an already-applied same-timestamp + // LARGER-id head with the true winner: only here may the incoming blob's + // per-entry values win an equal-`updatedAt` tie, and only here does the + // pending publish (which reflected the superseded head) get cancelled. + // Any other application (bootstrap / live / newer timestamp) merges over + // optimistic local state with local-wins `mergeStores` and must NOT + // cancel a pending local publish — otherwise a later same-second local + // click (integer-second `updatedAt`) loses to an older remote entry that + // decrypts late, and its publish is silently dropped. + const isCanonicalSupersession = + remote.createdAt === lastAppliedRemoteTs.current && + lastAppliedEventId.current !== "" && + remote.eventId < lastAppliedEventId.current; lastAppliedRemoteTs.current = remote.createdAt; lastAppliedEventId.current = remote.eventId; - managerRef.current?.cancelPendingStarPublish(); - const merged = mergeStores(prev, remote.store); + const merged = isCanonicalSupersession + ? mergeApplyingRemote(prev, remote.store) + : mergeStores(prev, remote.store); + if (isCanonicalSupersession) { + managerRef.current?.cancelPendingStarPublish(); + } if (!writeChannelStarsStore(pubkey, merged)) return prev; return merged; };