From c7933fd9daf2f7401c2b5db6f63c431d95c3b0b4 Mon Sep 17 00:00:00 2001 From: Duncan Date: Wed, 26 Aug 2026 11:35:16 -0400 Subject: [PATCH] fix(sync): clamp publish timestamps and give sort prefs the durable lane MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Carl's re-review found three edit-loss paths remaining after the stars/mutes rev-merge landed. [P2] Stars, mutes, and sort stamped createdAt = lastRemoteCreatedAt + 1 uncapped, so a self-authored head accepted near the relay's +900s drift boundary wedged every later publish until wall time caught up. Extract the sections clamp into a shared clampPublishCreatedAt in sidebarSyncWatermark.ts (all four surfaces already import that module) and wire sections/stars/mutes/sort to it. [P1] Sort preferences never got the durable lane: doPublish only logged on failure, destroy() discarded the pending edit, and a live remote cancelled the debounce with the edit stranded. Port the sections lane — durable outbox + bootstrap resume, generation/CAS ownership, single- flight + completion re-drive, 2s->30s backoff, 60s reconcile loop, and pending-aware remote application. Sort stays whole-blob LWW; a lost head is adopted at pre-publish rather than republished. Tests: a clamp test per surface, and sort lifecycle coverage — outbox resume after destroy-inside-debounce, retry without a later edit, overlapping-generation safety, live-remote-during-debounce adopt, and a hook-level pending-defer test. Co-authored-by: Will Pfleger Signed-off-by: Will Pfleger --- .../sidebar/lib/channelMutesSync.test.mjs | 47 +++ .../features/sidebar/lib/channelMutesSync.ts | 9 +- .../sidebar/lib/channelSectionsSync.ts | 15 +- .../sidebar/lib/channelSortPreference.ts | 53 +++ .../sidebar/lib/channelSortSync.test.mjs | 397 ++++++++++++++++-- .../features/sidebar/lib/channelSortSync.ts | 344 +++++++++++++-- .../sidebar/lib/channelStarsSync.test.mjs | 47 +++ .../features/sidebar/lib/channelStarsSync.ts | 9 +- .../sidebar/lib/sidebarSyncWatermark.ts | 29 ++ .../lib/useChannelSortPreference.test.mjs | 120 ++++++ .../sidebar/lib/useChannelSortPreference.ts | 90 +++- 11 files changed, 1054 insertions(+), 106 deletions(-) diff --git a/desktop/src/features/sidebar/lib/channelMutesSync.test.mjs b/desktop/src/features/sidebar/lib/channelMutesSync.test.mjs index 1ef2cdd1b5..5f6ed2678b 100644 --- a/desktop/src/features/sidebar/lib/channelMutesSync.test.mjs +++ b/desktop/src/features/sidebar/lib/channelMutesSync.test.mjs @@ -393,3 +393,50 @@ test("revert-fix: relay-A watermark does not suppress first-sync seed on relay-B mock.reset(); } }); + +// ─── Timestamp clamp (Carl P2): a far-future remote head must not push our +// published createdAt past the relay's ±15min future-drift window. +// Mutation test: removing the 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 primes lastRemoteCreatedAt to farFutureHead (undecryptable so + // it does not merge into the store); pre-publish call returns created_at=0. + return Promise.resolve([ + { + pubkey: "pk-clamp", + content: "bad-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, channels: {} })); + mock.method(relayClient, "publishEvent", (evt) => { + signedCreatedAt = evt.created_at; + return Promise.resolve(); + }); + try { + const manager = new ChannelMuteSyncManager("pk-clamp", RELAY); + await manager.fetchRemoteMutes(); // prime lastRemoteCreatedAt + manager.publishMutes(makeStore({ ch1: E(true, 100, 1) })); + fw._fireTimer(); + await new Promise((r) => setTimeout(r, 20)); + assert.ok(signedCreatedAt !== null, "publish must have been attempted"); + assert.ok( + signedCreatedAt <= Math.floor(Date.now() / 1000) + 840, + `createdAt must be clamped inside the future window — got ${signedCreatedAt}`, + ); + } finally { + tauri.restore(); + restore(); + mock.reset(); + } +}); diff --git a/desktop/src/features/sidebar/lib/channelMutesSync.ts b/desktop/src/features/sidebar/lib/channelMutesSync.ts index ac88f3c4bc..f534b50fa1 100644 --- a/desktop/src/features/sidebar/lib/channelMutesSync.ts +++ b/desktop/src/features/sidebar/lib/channelMutesSync.ts @@ -15,6 +15,7 @@ import { } from "./channelMutesStorage"; import { advanceWatermark, + clampPublishCreatedAt, readWatermark, runBootstrap, type FetchResult, @@ -295,10 +296,10 @@ export class ChannelMuteSyncManager { channels: merged.channels, }; const ciphertext = await nip44EncryptToSelf(JSON.stringify(payload)); - const createdAt = Math.max( - Math.floor(Date.now() / 1_000), - this.lastRemoteCreatedAt + 1, - ); + // Clamp inside the relay's future-drift window so a skewed remote head + // can never make us stamp an unbounded future timestamp that wedges every + // subsequent publish; we adopt such a head on the next fetch instead. + const createdAt = clampPublishCreatedAt(this.lastRemoteCreatedAt); const event = await signRelayEvent({ kind: KIND_CHANNEL_MUTES, content: ciphertext, diff --git a/desktop/src/features/sidebar/lib/channelSectionsSync.ts b/desktop/src/features/sidebar/lib/channelSectionsSync.ts index b736aa812c..d8a125e981 100644 --- a/desktop/src/features/sidebar/lib/channelSectionsSync.ts +++ b/desktop/src/features/sidebar/lib/channelSectionsSync.ts @@ -15,6 +15,7 @@ import { } from "./channelSectionsStorage"; import { advanceWatermark, + clampPublishCreatedAt, readWatermark, runBootstrap, type FetchResult, @@ -24,14 +25,6 @@ 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. @@ -460,16 +453,12 @@ export class ChannelSectionSyncManager { assignments: merged.assignments, }; const ciphertext = await nip44EncryptToSelf(JSON.stringify(payload)); - 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 the next pre-publish fetch rather than // walking past it. - const createdAt = Math.min( - Math.max(now, this.lastRemoteCreatedAt + 1), - now + MAX_PUBLISH_FUTURE_SECS, - ); + const createdAt = clampPublishCreatedAt(this.lastRemoteCreatedAt); const event = await signRelayEvent({ kind: KIND_CHANNEL_SECTIONS, content: ciphertext, diff --git a/desktop/src/features/sidebar/lib/channelSortPreference.ts b/desktop/src/features/sidebar/lib/channelSortPreference.ts index 4de9768d84..755e3b3385 100644 --- a/desktop/src/features/sidebar/lib/channelSortPreference.ts +++ b/desktop/src/features/sidebar/lib/channelSortPreference.ts @@ -133,6 +133,59 @@ export function writeChannelSortStore( } } +const OUTBOX_KEY_PREFIX = "buzz-channel-sort-outbox.v1"; + +function outboxKey(pubkey: string, relayUrl: string): string { + return `${OUTBOX_KEY_PREFIX}:${pubkey}:${encodeURIComponent(normalizeRelayUrl(relayUrl))}`; +} + +/** + * Persist an unpublished sort 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 writeChannelSortOutbox( + pubkey: string, + store: ChannelSortStore, + relayUrl: string, +): void { + try { + window.localStorage.setItem( + outboxKey(pubkey, relayUrl), + JSON.stringify(boundChannelSortStore(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 sort edit, or null when none/unparseable. */ +export function readChannelSortOutbox( + pubkey: string, + relayUrl: string, +): ChannelSortStore | null { + try { + const raw = window.localStorage.getItem(outboxKey(pubkey, relayUrl)); + if (!raw) return null; + return parseChannelSortPayload(JSON.parse(raw)); + } catch { + return null; + } +} + +/** Clear the persisted sort outbox (edit published, superseded, or a no-op). */ +export function clearChannelSortOutbox(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. + } +} + export function sortModeForGroup( store: ChannelSortStore, group: ChannelSortGroupKey, diff --git a/desktop/src/features/sidebar/lib/channelSortSync.test.mjs b/desktop/src/features/sidebar/lib/channelSortSync.test.mjs index 28159eedd3..06fe665f7a 100644 --- a/desktop/src/features/sidebar/lib/channelSortSync.test.mjs +++ b/desktop/src/features/sidebar/lib/channelSortSync.test.mjs @@ -15,6 +15,7 @@ function makeStore(groups = {}) { const RELAY = "wss://r.test"; const RELAY_KEY = encodeURIComponent(RELAY); +const OUTBOX_KEY = (pk) => `buzz-channel-sort-outbox.v1:${pk}:${RELAY_KEY}`; // ─── destroy() must cancel pending publish, not flush ───────────────────────── test("destroy: cancels pending publish without flushing to the relay", () => { @@ -92,8 +93,7 @@ test("destroy: is safe to call with no pending publish", () => { // Wiring tests 1-3 drive the production bootstrap() path; policy tested once // in sidebarSyncWatermark.test.mjs. -// 1. fetch failed (error/timeout) + local non-empty → hold, zero publish calls -// Mutation: removing the failed guard causes bootstrap to call publishSortPrefs → pendingStore set. +// 1. fetch failed → hold, pendingStore null (mutation: remove failed guard → seed queued) test("revert-fix: fetch failed (error) does not trigger seed-publish via bootstrap", async () => { mock.method(relayClient, "fetchEvents", () => Promise.reject(new Error("relay timeout")), @@ -112,8 +112,7 @@ test("revert-fix: fetch failed (error) does not trigger seed-publish via bootstr } }); -// 2. absent + persisted head > 0 → hold, zero publish calls (the dev-build stale-copy case) -// Mutation: setting watermark to 0 in localStorage causes bootstrap to seed. +// 2. absent + persisted head > 0 → hold, zero publish calls (dev-build stale-copy case) test("revert-fix: absent fetch with prior watermark blocks seed-publish via bootstrap", async () => { mock.method(relayClient, "fetchEvents", () => Promise.resolve([])); mock.method(relayClient, "publishEvent", () => Promise.resolve()); @@ -125,13 +124,6 @@ test("revert-fix: absent fetch with prior watermark blocks seed-publish via boot const restore = installFakeWindow(fw); try { const manager = new ChannelSortSyncManager("pk-stale", RELAY); - assert.ok( - Number( - fw.localStorage.getItem( - `buzz-sync-watermark.v1:channel-sort:pk-stale:${RELAY_KEY}`, - ) ?? "0", - ) > 0, - ); const result = await manager.bootstrap(makeStore({ channels: "recent" })); assert.equal(result.action, "hold"); assert.equal(manager.getPendingStore(), null); @@ -142,7 +134,6 @@ test("revert-fix: absent fetch with prior watermark blocks seed-publish via boot }); // 3. absent + head 0 + local non-empty → seed-publish queued (first-sync preserved) -// Mutation: removing the absent+head-0 seed call leaves pendingStore null. test("revert-fix: absent fetch with zero watermark seeds via bootstrap (first-sync preserved)", async () => { mock.method(relayClient, "fetchEvents", () => Promise.resolve([])); mock.method(relayClient, "publishEvent", () => Promise.resolve()); @@ -150,12 +141,6 @@ test("revert-fix: absent fetch with zero watermark seeds via bootstrap (first-sy const restore = installFakeWindow(fw); try { const manager = new ChannelSortSyncManager("pk-fresh", RELAY); - assert.equal( - fw.localStorage.getItem( - `buzz-sync-watermark.v1:channel-sort:pk-fresh:${RELAY_KEY}`, - ), - null, - ); const result = await manager.bootstrap(makeStore({ channels: "recent" })); assert.equal(result.action, "hold"); assert.ok(manager.getPendingStore() !== null); @@ -165,25 +150,30 @@ 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: sort LWW — newer decryptable pre-publish event selected after undecryptable head recorded", async () => { +// ─── Whole-blob LWW: adopt-winner and local-winner ──────────────────────────── + +// 4a. 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 pending/outbox so the loser can't be replayed. +// Mutation: 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_KEY = "remote-group-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( @@ -191,22 +181,30 @@ test("revert-fix: sort LWW — newer decryptable pre-publish event selected afte ); try { const manager = new ChannelSortSyncManager("pk-lww", RELAY); - await manager.fetchRemoteSortPrefs(); - assert.ok( - Number( - fw.localStorage.getItem( - `buzz-sync-watermark.v1:channel-sort:pk-lww:${RELAY_KEY}`, - ) ?? "0", - ) >= 100, - ); + const adopted = []; + manager.setOnRemoteAdopted((r) => adopted.push(r)); manager.publishSortPrefs(makeStore({ "local-group": "recent" })); + assert.ok( + fw.localStorage.getItem(OUTBOX_KEY("pk-lww")) !== null, + "edit must be persisted to the durable outbox", + ); fw._fireTimer(); await new Promise((r) => setTimeout(r, 20)); - const pt = tauri.capturedPlaintext(); - assert.ok(pt !== null, "nip44EncryptToSelf must have been called"); + 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( - JSON.parse(pt).groups && REMOTE_KEY in JSON.parse(pt).groups, - `remote groups must win LWW merge — got: ${pt}`, + REMOTE_KEY in adopted[0].store.groups, + "adopted store must be the remote content", + ); + assert.equal(manager.getPendingStore(), null, "pending must be cleared"); + assert.equal( + fw.localStorage.getItem(OUTBOX_KEY("pk-lww")), + null, + "outbox must be cleared on adopt so the loser is never replayed", ); } finally { tauri.restore(); @@ -215,7 +213,316 @@ test("revert-fix: sort LWW — newer decryptable pre-publish event selected afte } }); -// 5. live-sub: undecryptable event on live path records head before decrypt +// 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 ChannelSortSyncManager("pk-win", RELAY); + manager.publishSortPrefs(makeStore({ channels: "recent" })); + fw._fireTimer(); + await new Promise((r) => setTimeout(r, 20)); + assert.equal(publishCalls.length, 1, "local edit must be published"); + assert.equal( + fw.localStorage.getItem(OUTBOX_KEY("pk-win")), + 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 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 primes lastRemoteCreatedAt to farFutureHead (undecryptable so + // no adopt); pre-publish call returns created_at=0 so the local edit wins. + return Promise.resolve([ + { + pubkey: "pk-clamp", + content: "bad-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, groups: {} })); + mock.method(relayClient, "publishEvent", (evt) => { + signedCreatedAt = evt.created_at; + return Promise.resolve(); + }); + try { + const manager = new ChannelSortSyncManager("pk-clamp", RELAY); + await manager.fetchRemoteSortPrefs(); // prime lastRemoteCreatedAt + manager.publishSortPrefs(makeStore({ channels: "recent" })); + fw._fireTimer(); + await new Promise((r) => setTimeout(r, 20)); + assert.ok(signedCreatedAt !== null, "publish must have been attempted"); + assert.ok( + signedCreatedAt <= Math.floor(Date.now() / 1000) + 840, + `createdAt must be clamped inside the future window — got ${signedCreatedAt}`, + ); + } finally { + tauri.restore(); + restore(); + mock.reset(); + } +}); + +// ─── Durable lane: outbox resume, retry, and cross-generation safety ────────── + +// 5. Durable outbox resume: an edit made <2s before quit (destroy inside the +// debounce) is persisted, and a fresh manager resuming from that outbox +// publishes it — the edit is not silently dropped at teardown. +// Mutation: dropping writeChannelSortOutbox leaves the outbox null → no resume. +test("durable outbox: edit destroyed inside the debounce resumes and publishes on remount", 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 { + // Window 1: edit then destroy before the debounce fires. + const m1 = new ChannelSortSyncManager("pk-resume", RELAY); + m1.publishSortPrefs(makeStore({ channels: "recent" })); + const persisted = fw.localStorage.getItem(OUTBOX_KEY("pk-resume")); + assert.ok(persisted !== null, "edit must be persisted before teardown"); + m1.destroy(); + assert.equal(publishCalls.length, 0, "destroy must not flush"); + + // Window 2: resume the persisted outbox (the hook does this after bootstrap). + const m2 = new ChannelSortSyncManager("pk-resume", RELAY); + m2.publishSortPrefs(JSON.parse(persisted)); + fw._fireTimer(); + await new Promise((r) => setTimeout(r, 20)); + assert.equal(publishCalls.length, 1, "resumed edit must publish"); + assert.equal( + fw.localStorage.getItem(OUTBOX_KEY("pk-resume")), + null, + "outbox must be cleared once the resumed edit publishes", + ); + } finally { + tauri.restore(); + restore(); + mock.reset(); + } +}); + +// 6. Failed publish retries with no later edit: a transient publish rejection +// keeps the pending edit and schedules a bounded-backoff retry that succeeds, +// rather than logging-and-dropping (the pre-fix behaviour). +// Mutation: reverting scheduleRetry to a bare console.warn leaves pending null +// and never re-publishes. +test("failed publish retries the retained edit without a later edit", async () => { + mock.method(relayClient, "fetchEvents", () => Promise.resolve([])); + let attempts = 0; + mock.method(relayClient, "publishEvent", () => { + attempts++; + if (attempts === 1) return Promise.reject(new Error("socket timeout")); + return Promise.resolve(); + }); + const fw = makeFakeWindow(); + const restore = installFakeWindow(fw); + const tauri = installTauriMock("{}"); + try { + const manager = new ChannelSortSyncManager("pk-retry", RELAY); + manager.publishSortPrefs(makeStore({ channels: "recent" })); + fw._fireTimer(); // debounce → doPublish → publish rejects → scheduleRetry + await new Promise((r) => setTimeout(r, 20)); + assert.equal(attempts, 1, "first publish attempt rejected"); + assert.ok( + manager.getPendingStore() !== null, + "failed publish must retain the pending edit", + ); + assert.ok(fw._hasTimer(), "a bounded-backoff retry must be scheduled"); + fw._fireTimer(); // retry → doPublish → publish resolves + await new Promise((r) => setTimeout(r, 20)); + assert.equal(attempts, 2, "retry must re-attempt the publish"); + assert.equal( + manager.getPendingStore(), + null, + "successful retry must clear the pending edit", + ); + } finally { + tauri.restore(); + restore(); + mock.reset(); + } +}); + +// 7. Overlapping publishes: an older completion must not erase a newer queued +// edit or its outbox (generation compare-and-swap). +// Mutation: dropping the gen guard in discardPending clears B on A's completion. +test("overlapping publishes: older completion does not erase a newer queued edit", async () => { + mock.method(relayClient, "fetchEvents", () => Promise.resolve([])); + let releaseFirst = null; + let publishCalls = 0; + mock.method(relayClient, "publishEvent", () => { + publishCalls++; + if (publishCalls === 1) { + return new Promise((res) => { + releaseFirst = 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(); + await Promise.resolve(); + await Promise.resolve(); + }; + const restore = installFakeWindow(fakeWindow); + const tauri = installTauriMock("{}"); + try { + const manager = new ChannelSortSyncManager("pk-overlap", RELAY); + manager.publishSortPrefs(makeStore({ channels: "recent" })); // A + await fireDelay(2000); // debounce → doPublish(A) awaits publishEvent + while (releaseFirst === null) await Promise.resolve(); + + manager.publishSortPrefs(makeStore({ dms: "alpha" })); // B while A in flight + assert.deepEqual( + Object.keys(manager.getPendingStore()?.groups ?? {}), + ["dms"], + "B is now the pending edit", + ); + assert.deepEqual( + Object.keys(JSON.parse(storage.get(OUTBOX_KEY("pk-overlap"))).groups), + ["dms"], + "outbox holds B", + ); + + releaseFirst(); // A completes — must NOT clear B's pending/outbox + await Promise.resolve(); + await Promise.resolve(); + await Promise.resolve(); + + assert.deepEqual( + Object.keys(manager.getPendingStore()?.groups ?? {}), + ["dms"], + "older completion must leave B pending", + ); + assert.ok( + storage.get(OUTBOX_KEY("pk-overlap")) !== undefined, + "older completion must leave B's outbox intact", + ); + manager.destroy(); + } finally { + tauri.restore(); + restore(); + mock.reset(); + } +}); + +// 8. Live remote during debounce is adopted at pre-publish, not overwritten. The +// live event advances the mutable watermark before doPublish runs, so +// comparing the fetched head against the watermark would see equality and +// publish over the newer remote. The pre-publish check compares against the +// baseline frozen at publishSortPrefs instead. +// Mutation: comparing against lastRemoteCreatedAt rather than publishBaseline +// republishes local over the newer remote. +test("live remote during debounce is adopted at pre-publish, not overwritten", async () => { + const REMOTE_KEY = "remote-during-debounce"; + let liveCallback = null; + mock.method(relayClient, "subscribeLive", (_filter, onEvent) => { + liveCallback = onEvent; + return Promise.resolve(async () => {}); + }); + // Pre-publish fetch returns the same newer head the live event delivered. + mock.method(relayClient, "fetchEvents", () => + Promise.resolve([ + { + pubkey: "pk-live-deb", + content: "good-cipher", + created_at: 500, + id: "evt-live", + }, + ]), + ); + const publishCalls = []; + mock.method(relayClient, "publishEvent", (...args) => { + publishCalls.push(args); + return Promise.resolve(); + }); + const fw = makeFakeWindow(); + const restore = installFakeWindow(fw); + const tauri = installTauriMock( + JSON.stringify({ version: 1, groups: { [REMOTE_KEY]: "recent" } }), + ); + try { + const manager = new ChannelSortSyncManager("pk-live-deb", RELAY); + const adopted = []; + manager.setOnRemoteAdopted((r) => adopted.push(r)); + await manager.subscribeToSortPrefs(() => {}); + // Local edit queued first, freezing the baseline at the empty head. + manager.publishSortPrefs(makeStore({ "local-group": "alpha" })); + // A newer remote head arrives during the debounce window (advances the + // mutable watermark to created_at=500). + liveCallback({ + pubkey: "pk-live-deb", + content: "good-cipher", + created_at: 500, + id: "evt-live", + }); + await new Promise((r) => setTimeout(r, 0)); + fw._fireTimer(); // debounce → doPublish → pre-publish sees advanced head + await new Promise((r) => setTimeout(r, 20)); + assert.equal( + publishCalls.length, + 0, + "must not publish over a remote that became head during the debounce", + ); + assert.equal(adopted.length, 1, "the newer remote must be adopted"); + } finally { + tauri.restore(); + restore(); + mock.reset(); + } +}); + +// 9. live-sub: undecryptable event on live path records head before decrypt. // Mutation test: removing recordRemoteHead before decrypt in the live callback // leaves watermark at 0 after a live event. test("revert-fix: undecryptable live event advances watermark before decrypt attempt", async () => { diff --git a/desktop/src/features/sidebar/lib/channelSortSync.ts b/desktop/src/features/sidebar/lib/channelSortSync.ts index fe71fe62df..1fe944a2d4 100644 --- a/desktop/src/features/sidebar/lib/channelSortSync.ts +++ b/desktop/src/features/sidebar/lib/channelSortSync.ts @@ -7,11 +7,14 @@ import { import type { RelayEvent } from "@/shared/api/types"; import { KIND_CHANNEL_SORT } from "@/shared/constants/kinds"; import { + clearChannelSortOutbox, parseChannelSortPayload, + writeChannelSortOutbox, type ChannelSortStore, } from "./channelSortPreference"; import { advanceWatermark, + clampPublishCreatedAt, readWatermark, runBootstrap, type FetchResult, @@ -21,12 +24,75 @@ const D_TAG = "channel-sort"; const BLOB_TYPE = D_TAG; const DEBOUNCE_MS = 2_000; +// 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 RemoteSortPrefs = { store: ChannelSortStore; 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. The manager hands the remote back to the hook + * and never publishes. + */ +type PublishDecision = + | { kind: "publish"; store: ChannelSortStore } + | { kind: "adopt"; remote: RemoteSortPrefs }; + +/** + * 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: RemoteSortPrefs, + 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 { @@ -48,15 +114,48 @@ async function decryptAndParse( * low-frequency preference blob, so whole-blob LWW (like sections) is * sufficient — per-key merge (like stars/mutes) would be unnecessary * complexity here. + * + * The durable lane (outbox + generation/CAS + bounded retry) is mirrored from + * sections so a sort edit made just before quit/community-switch, or one whose + * publish times out, self-heals rather than being silently dropped. */ export class ChannelSortSyncManager { 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 publishSortPrefs 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 and advanced ONLY by our own successful publishes. + private publishBaseline: PublishBaseline = { createdAt: 0, eventId: "" }; private pendingStore: ChannelSortStore | null = null; + // Monotonic id for the current pending edit. Every publishSortPrefs() 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 defers; the in-flight cycle's completion + // re-drives it. This kills the cross-generation race class by construction. + private publishInFlight = false; + // Event ids we signed and sent to the relay but whose ACK never arrived. The + // relay MAY have accepted such a write, so if a later cycle's pre-publish + // fetch returns a head whose id is in this set, that head is OUR OWN accepted + // predecessor — fold it forward and publish above it, rather than adopting it + // and erasing the queued edit. + private ambiguousAttemptIds = new Set(); private lastPublishedStore: ChannelSortStore | null = null; private destroyed = false; + // Set by the hook so an adopted remote head (local edit lost whole-blob LWW) + // is written through to React state + localStorage. + private onRemoteAdopted: ((remote: RemoteSortPrefs) => void) | null = null; constructor(pubkey: string, relayUrl: string) { this.pubkey = pubkey; @@ -64,6 +163,11 @@ export class ChannelSortSyncManager { this.lastRemoteCreatedAt = readWatermark(pubkey, BLOB_TYPE, relayUrl); } + /** Register the hook's adopt-remote sink (write-through to UI + storage). */ + setOnRemoteAdopted(cb: (remote: RemoteSortPrefs) => void): void { + this.onRemoteAdopted = cb; + } + async fetchRemoteSortPrefs(): Promise> { try { const events = await relayClient.fetchEvents({ @@ -76,7 +180,10 @@ export class ChannelSortSyncManager { return { status: "absent" }; } const event = events[0]; - this.recordRemoteHead(event.created_at); + // 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, event.id); const result = await decryptAndParse(event); if (!result) { return { status: "failed", createdAt: event.created_at }; @@ -92,10 +199,21 @@ export class ChannelSortSyncManager { } } - 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. + 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); } @@ -104,26 +222,110 @@ export class ChannelSortSyncManager { window.clearTimeout(this.debounceTimer); this.debounceTimer = null; } + if (this.retryTimer !== null) { + window.clearTimeout(this.retryTimer); + this.retryTimer = null; + } } getPendingStore(): ChannelSortStore | 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. + * + * Compare-and-swap on `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 + * — a stale adopt must not clear its pending state or overwrite its optimistic + * UI. We still advance the watermark (monotonic and always safe). + */ + private adoptRemote(remote: RemoteSortPrefs, gen: number): void { + this.recordRemoteHead(remote.createdAt, remote.eventId); + if (gen !== this.pendingGeneration) return; + this.pendingStore = null; + clearChannelSortOutbox(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; + clearChannelSortOutbox(this.pubkey, this.relayUrl); + } + publishSortPrefs(store: ChannelSortStore): void { this.pendingStore = store; + ++this.pendingGeneration; + // Freeze the canonical head this edit is racing against at queue time so a + // live event applied during the debounce window (which advances the mutable + // watermark) cannot 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, 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 (durable outbox). + writeChannelSortOutbox(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. + */ + 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; + if ( + !this.destroyed && + this.pendingStore !== null && + this.debounceTimer === null && + this.retryTimer === null + ) { + this.startCycle(); + } + }); + } + private async fetchOwnBlobBeforePublish( store: ChannelSortStore, - ): Promise { + ): Promise { try { const events = await relayClient.fetchEvents({ kinds: [KIND_CHANNEL_SORT], @@ -131,23 +333,35 @@ export class ChannelSortSyncManager { "#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; - // Sort prefs 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 }; + // Sort prefs use whole-blob LWW. Compare the fetched head against the + // baseline frozen when this edit was queued — NOT the live watermark. 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)) { + // Unless the advancing head is a prior publish of OURS whose ACK was + // lost: it is our own accepted predecessor, not a competing remote. + // Fold it into the baseline and publish above it so the queued edit + // survives instead of adopting our own stale write away. + if (this.ambiguousAttemptIds.has(remote.eventId)) { + this.publishBaseline = canonicalMax(this.publishBaseline, { + createdAt: remote.createdAt, + eventId: remote.eventId, + }); + return { kind: "publish", store }; + } + return { kind: "adopt", remote }; } - return store; + return { kind: "publish", store }; } catch { - return store; + return { kind: "publish", store }; } } @@ -164,15 +378,40 @@ export class ChannelSortSyncManager { return true; } - private async doPublish(store: ChannelSortStore): 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: ChannelSortStore, 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. + // was awaited (community switch during in-flight fetch). if (this.destroyed) return; + // A newer edit was queued while we awaited the pre-publish fetch. It owns + // convergence now; the serialized cycle re-drives for it once this run + // 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 = { @@ -180,10 +419,10 @@ export class ChannelSortSyncManager { groups: merged.groups, }; const ciphertext = await nip44EncryptToSelf(JSON.stringify(payload)); - const createdAt = Math.max( - Math.floor(Date.now() / 1_000), - this.lastRemoteCreatedAt + 1, - ); + // Clamp inside the relay's future-drift window so a skewed remote head + // can never make us stamp an unbounded future timestamp that wedges every + // subsequent publish; we adopt such a head on the next fetch instead. + const createdAt = clampPublishCreatedAt(this.lastRemoteCreatedAt); const event = await signRelayEvent({ kind: KIND_CHANNEL_SORT, content: ciphertext, @@ -193,20 +432,49 @@ export class ChannelSortSyncManager { ["t", D_TAG], // relay discoverability; not used in our filters ], }); - // 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; + // Final guard immediately before the network call — a newer edit may have + // been queued during the encrypt/sign await, or the manager destroyed. + if (this.destroyed || gen !== this.pendingGeneration) return; + // Record this signed id as an in-flight attempt of unknown fate before we + // send it. If the ACK is lost below, a later cycle that fetches this id as + // the head recognises it as our own accepted write and folds it forward + // rather than adopting it away. + this.ambiguousAttemptIds.add(event.id); await relayClient.publishEvent( event, "Timed out publishing channel sort preferences.", "Failed to publish channel sort preferences.", ); - this.recordRemoteHead(event.created_at); - this.lastPublishedStore = merged; - this.pendingStore = null; + this.recordRemoteHead(event.created_at, event.id); + // This write is now the confirmed accepted head; it dominates every prior + // attempt (`created_at DESC, id ASC`), so no earlier ambiguous id can ever + // be the canonical head again. Clear the set to keep it bounded. + this.ambiguousAttemptIds.clear(); + // Fold our own accepted head into the pending edit's baseline — + // unconditional across generations so a newer edit's pre-publish check + // does not mistake OUR prior publish for a competing remote and adopt it + // away. Genuine remotes never fold in here. + 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; + // Ambiguous outcome: the publish promise rejected (timeout / socket + // error), but the relay may already have accepted the write before the + // ACK was lost. Keep the pending edit and retry with backoff. The attempt + // id stays in ambiguousAttemptIds: if the relay did accept it, a later + // cycle that fetches this id as the head folds it forward as our own + // accepted predecessor instead of adopting it away. console.warn("[channelSortSync] publish failed:", error); + this.scheduleRetry(gen); } } @@ -224,7 +492,7 @@ export class ChannelSortSyncManager { 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); @@ -252,10 +520,10 @@ export class ChannelSortSyncManager { 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 sort prefs 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: publishSortPrefs persisted them to + // the durable outbox synchronously, and the next mount resumes them. + // Flushing here is still avoided — it could publish relay A's sort prefs to + // relay B via the shared relayClient singleton. this.destroyed = true; this.cancelPendingPublish(); this.pendingStore = null; diff --git a/desktop/src/features/sidebar/lib/channelStarsSync.test.mjs b/desktop/src/features/sidebar/lib/channelStarsSync.test.mjs index d238f3c0fa..58d9ce9aa9 100644 --- a/desktop/src/features/sidebar/lib/channelStarsSync.test.mjs +++ b/desktop/src/features/sidebar/lib/channelStarsSync.test.mjs @@ -393,3 +393,50 @@ test("revert-fix: relay-A watermark does not suppress first-sync seed on relay-B mock.reset(); } }); + +// ─── Timestamp clamp (Carl P2): a far-future remote head must not push our +// published createdAt past the relay's ±15min future-drift window. +// Mutation test: removing the 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 primes lastRemoteCreatedAt to farFutureHead (undecryptable so + // it does not merge into the store); pre-publish call returns created_at=0. + return Promise.resolve([ + { + pubkey: "pk-clamp", + content: "bad-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, channels: {} })); + mock.method(relayClient, "publishEvent", (evt) => { + signedCreatedAt = evt.created_at; + return Promise.resolve(); + }); + try { + const manager = new ChannelStarSyncManager("pk-clamp", RELAY); + await manager.fetchRemoteStars(); // prime lastRemoteCreatedAt + manager.publishStars(makeStore({ ch1: E(true, 100, 1) })); + fw._fireTimer(); + await new Promise((r) => setTimeout(r, 20)); + assert.ok(signedCreatedAt !== null, "publish must have been attempted"); + assert.ok( + signedCreatedAt <= Math.floor(Date.now() / 1000) + 840, + `createdAt must be clamped inside the future window — got ${signedCreatedAt}`, + ); + } finally { + tauri.restore(); + restore(); + mock.reset(); + } +}); diff --git a/desktop/src/features/sidebar/lib/channelStarsSync.ts b/desktop/src/features/sidebar/lib/channelStarsSync.ts index 695cfc3ebc..7daf991829 100644 --- a/desktop/src/features/sidebar/lib/channelStarsSync.ts +++ b/desktop/src/features/sidebar/lib/channelStarsSync.ts @@ -15,6 +15,7 @@ import { } from "./channelStarsStorage"; import { advanceWatermark, + clampPublishCreatedAt, readWatermark, runBootstrap, type FetchResult, @@ -295,10 +296,10 @@ export class ChannelStarSyncManager { channels: merged.channels, }; const ciphertext = await nip44EncryptToSelf(JSON.stringify(payload)); - const createdAt = Math.max( - Math.floor(Date.now() / 1_000), - this.lastRemoteCreatedAt + 1, - ); + // Clamp inside the relay's future-drift window so a skewed remote head + // can never make us stamp an unbounded future timestamp that wedges every + // subsequent publish; we adopt such a head on the next fetch instead. + const createdAt = clampPublishCreatedAt(this.lastRemoteCreatedAt); const event = await signRelayEvent({ kind: KIND_CHANNEL_STARS, content: ciphertext, diff --git a/desktop/src/features/sidebar/lib/sidebarSyncWatermark.ts b/desktop/src/features/sidebar/lib/sidebarSyncWatermark.ts index 8f4cb321bd..2daf230a34 100644 --- a/desktop/src/features/sidebar/lib/sidebarSyncWatermark.ts +++ b/desktop/src/features/sidebar/lib/sidebarSyncWatermark.ts @@ -23,6 +23,35 @@ import { normalizeRelayUrl } from "@/shared/lib/normalizeRelayUrl"; const PREFIX = "buzz-sync-watermark.v1"; +// The relay rejects events more than ±15 minutes (900s) from server time +// (`MAX_TIMESTAMP_DRIFT_SECS` in ingest.rs). Clamp every sidebar-sync publish's +// `created_at` well inside that window so a skewed remote head can never make a +// manager manufacture an unbounded future timestamp that wedges every +// subsequent publish. 840s leaves ~60s of transit margin while still letting a +// publish win LWW against any legitimately-timestamped head. +const MAX_PUBLISH_FUTURE_SECS = 840; + +/** + * Compute the `created_at` for a sidebar-sync publish. + * + * Stamps one second past the newest observed remote head so the write wins + * last-write-wins, but never further ahead than the relay's future-drift + * window. If a skewed remote head sits beyond the window the manager loses LWW + * and adopts it on the next fetch rather than walking past it and wedging. + * + * `nowSecs` defaults to the current wall clock in seconds; callers pass it + * explicitly only to keep a single clock reading across a publish. + */ +export function clampPublishCreatedAt( + lastRemoteCreatedAt: number, + nowSecs: number = Math.floor(Date.now() / 1_000), +): number { + return Math.min( + Math.max(nowSecs, lastRemoteCreatedAt + 1), + nowSecs + MAX_PUBLISH_FUTURE_SECS, + ); +} + /** * Tri-state result returned by every `fetchRemote*()` method. * diff --git a/desktop/src/features/sidebar/lib/useChannelSortPreference.test.mjs b/desktop/src/features/sidebar/lib/useChannelSortPreference.test.mjs index a9a65754b0..1a8172d1a5 100644 --- a/desktop/src/features/sidebar/lib/useChannelSortPreference.test.mjs +++ b/desktop/src/features/sidebar/lib/useChannelSortPreference.test.mjs @@ -18,6 +18,126 @@ before(() => { after(() => dom.window.close()); +// Carl P1-sort regression: a live remote arriving while a local sort edit is +// still pending must NOT overwrite the optimistic edit or strand its durable +// outbox. Reverting applyRemote's `hasPendingEdit()` guard (or restoring the +// old `cancelPendingPublish()` on remote arrival) lets the remote clobber the +// pending edit and drop it. +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 { useChannelSortPreference } = await import( + "./useChannelSortPreference.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, + groups: { "remote-group": "recent" }, + }), + ); + if (cmd === "nip44_encrypt_to_self") return Promise.resolve("ct"); + if (cmd === "sign_event") + return Promise.resolve( + JSON.stringify({ + id: "signed", + pubkey: "pk-sort-pending", + content: "ct", + created_at: 0, + kind: 30078, + tags: [], + sig: "s", + }), + ); + return Promise.reject(new Error(`unmocked ${cmd}`)); + }, + }; + + const pubkey = "pk-sort-pending"; + const relayUrl = "wss://r.live"; + const outboxKey = `buzz-channel-sort-outbox.v1:${pubkey}:${encodeURIComponent(relayUrl)}`; + + 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"); + + // Make a local edit — it becomes the pending store and persists to outbox. + await act(async () => { + hook.result.current.setSortModeFor("channels", "recent"); + }); + assert.ok( + window.localStorage.getItem(outboxKey), + "local edit persisted to outbox", + ); + assert.equal( + hook.result.current.sortModeFor("channels"), + "recent", + "optimistic local edit applied", + ); + + // 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-sort"]], + sig: "s", + }); + await Promise.resolve(); + await Promise.resolve(); + }); + + assert.equal( + hook.result.current.sortModeFor("channels"), + "recent", + "pending local edit must NOT be overwritten by the live remote", + ); + assert.equal( + hook.result.current.sortModeFor("remote-group"), + "alpha", + "remote store must not have been applied over the pending edit", + ); + 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; + } +}); + // 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 diff --git a/desktop/src/features/sidebar/lib/useChannelSortPreference.ts b/desktop/src/features/sidebar/lib/useChannelSortPreference.ts index a692bbb31a..25ad01ae2e 100644 --- a/desktop/src/features/sidebar/lib/useChannelSortPreference.ts +++ b/desktop/src/features/sidebar/lib/useChannelSortPreference.ts @@ -3,7 +3,9 @@ import * as React from "react"; import { relayClient } from "@/shared/api/relayClient"; import { boundChannelSortStore, + clearChannelSortOutbox, DEFAULT_STORE, + readChannelSortOutbox, readChannelSortStore, sortModeForGroup, storageKey, @@ -16,6 +18,13 @@ import { import { ChannelSortSyncManager } from "./channelSortSync"; import type { RemoteSortPrefs } from "./channelSortSync"; +// Reconciliation cadence (mirrors sections). 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; + /** * Persistent per-group sidebar sort preferences, scoped by pubkey + relay so * they don't bleed across identities or communities (same scoping as channel @@ -85,6 +94,13 @@ export function useChannelSortPreference( ): ((prev: ChannelSortStore) => ChannelSortStore) => { 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. 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 @@ -97,7 +113,6 @@ export function useChannelSortPreference( return prev; lastAppliedRemoteTs.current = remote.createdAt; lastAppliedEventId.current = remote.eventId; - managerRef.current?.cancelPendingPublish(); if (!writeChannelSortStore(pubkey, remote.store, relayUrl)) return prev; return remote.store; }; @@ -105,6 +120,19 @@ export function useChannelSortPreference( [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), 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; @@ -114,13 +142,71 @@ export function useChannelSortPreference( if (result.action === "apply-remote") { setStore(applyRemote(result.data)); } - // "hold": seed already performed by bootstrap (if first-sync), or blocked. + // "hold": seed already performed by bootstrap (if first-sync), or blocked + // (failed fetch / prior watermark). Resume any edit persisted to the + // durable outbox before a prior quit/community-switch. + const outbox = readChannelSortOutbox(pubkey, relayUrl); + if (outbox) { + managerRef.current?.publishSortPrefs(outbox); + } else { + clearChannelSortOutbox(pubkey, relayUrl); + } }); return () => { cancelled = true; }; }, [pubkey, relayUrl, applyRemote]); + // Reconciliation loop (mirrors sections): 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?.fetchRemoteSortPrefs().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;