diff --git a/desktop/src/features/sidebar/lib/channelSectionsSync.test.mjs b/desktop/src/features/sidebar/lib/channelSectionsSync.test.mjs index 4a156f656f..17248daffe 100644 --- a/desktop/src/features/sidebar/lib/channelSectionsSync.test.mjs +++ b/desktop/src/features/sidebar/lib/channelSectionsSync.test.mjs @@ -436,3 +436,100 @@ 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(); + } +}); diff --git a/desktop/src/features/sidebar/lib/channelSectionsSync.ts b/desktop/src/features/sidebar/lib/channelSectionsSync.ts index 001dd4f4a1..375db4f1d3 100644 --- a/desktop/src/features/sidebar/lib/channelSectionsSync.ts +++ b/desktop/src/features/sidebar/lib/channelSectionsSync.ts @@ -78,6 +78,12 @@ export class ChannelSectionSyncManager { private retryDelayMs = RETRY_BASE_MS; private lastRemoteCreatedAt: number; 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; private lastPublishedStore: ChannelSectionStore | null = null; private destroyed = false; // Set by the hook so an adopted remote head (local edit lost LWW, or a relay @@ -151,37 +157,64 @@ export class ChannelSectionSyncManager { 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): void { + private adoptRemote(remote: RemoteSections, gen: number): void { this.recordRemoteHead(remote.createdAt); - this.discardPending(); + 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 both the in-memory pending edit and its durable outbox copy. */ - private discardPending(): void { + /** + * 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; + const gen = ++this.pendingGeneration; // 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); + void this.doPublish(store, gen); }, DEBOUNCE_MS); } @@ -246,19 +279,27 @@ export class ChannelSectionSyncManager { } /** Schedule a bounded-backoff retry of the retained pending edit. */ - private scheduleRetry(): void { + 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 store = this.pendingStore; const delay = this.retryDelayMs; this.retryDelayMs = Math.min(this.retryDelayMs * 2, RETRY_MAX_MS); this.retryTimer = window.setTimeout(() => { this.retryTimer = null; - void this.doPublish(store); + void this.doPublish(store, gen); }, delay); } - private async doPublish(store: ChannelSectionStore): Promise { + 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 decision = await this.fetchOwnBlobBeforePublish(store); // Guard: manager may have been destroyed while fetchOwnBlobBeforePublish @@ -266,12 +307,12 @@ export class ChannelSectionSyncManager { // before touching the relay. if (this.destroyed) return; if (decision.kind === "adopt") { - this.adoptRemote(decision.remote); + this.adoptRemote(decision.remote, gen); return; } const merged = decision.store; if (this.isIdenticalToLastPublished(merged)) { - this.discardPending(); + this.discardPending(gen); return; } const payload = { @@ -308,9 +349,13 @@ export class ChannelSectionSyncManager { "Failed to publish channel sections.", ); this.recordRemoteHead(event.created_at); - this.lastPublishedStore = merged; - this.discardPending(); - this.retryDelayMs = RETRY_BASE_MS; + // 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 @@ -320,9 +365,9 @@ export class ChannelSectionSyncManager { const head = await this.fetchRemoteSections(); if (this.destroyed) return; if (head.status === "found") { - this.adoptRemote(head.data); + this.adoptRemote(head.data, gen); } else { - this.scheduleRetry(); + this.scheduleRetry(gen); } return; } @@ -330,7 +375,7 @@ export class ChannelSectionSyncManager { // retry with backoff rather than waiting for a reconnect that a healthy // socket never fires. console.warn("[channelSectionsSync] publish failed:", error); - this.scheduleRetry(); + this.scheduleRetry(gen); } } diff --git a/desktop/src/features/sidebar/lib/useChannelSections.test.mjs b/desktop/src/features/sidebar/lib/useChannelSections.test.mjs index 401b59d9c1..6203034862 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 cbef2837a6..01b3187096 100644 --- a/desktop/src/features/sidebar/lib/useChannelSections.ts +++ b/desktop/src/features/sidebar/lib/useChannelSections.ts @@ -94,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; @@ -169,13 +180,10 @@ export function useChannelSections( 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)); - // applyRemote cancels the pending debounce; re-queue any live local - // edit so a periodic reconcile never silently drops it (mirrors the - // reconnect handler). doPublish re-checks the head and adopts if the - // remote is genuinely newer. - const pending = managerRef.current?.getPendingStore(); - if (pending) managerRef.current?.publishSections(pending); delayMs = RECONCILE_STEADY_MS; // relay answered → steady cadence } else if (result.status === "absent") { delayMs = RECONCILE_STEADY_MS; // answered (no blob) → steady cadence