fix(sidebar): close pending-edit convergence races in channel sections

Three cross-layer races defeated the one-convergence-mechanism design:

- An older in-flight publish unconditionally cleared pending state on
  completion, erasing a newer edit queued mid-flight. Each pending edit now
  carries a monotonic generation; a completion clears pending/outbox/retry only
  via compare-and-swap on the generation it published.
- Hook-level remote application (bootstrap/live/periodic) cancelled the pending
  publish's timers without deciding supersession, stranding the durable outbox
  and clobbering the optimistic edit. applyRemote now defers entirely to a
  pending edit, whose own debounced publish converges via publish-or-adopt; the
  manager's adopt path clears pending before write-through so the winning remote
  still applies.
- The equal-timestamp tie-break kept the largest event id, opposite the
  relay/database canonical order (created_at DESC, id ASC → lowest id wins).
  applyRemote now applies a strictly-lower id and ignores ids >= the last
  applied, so the UI converges on the event the relay actually stored.

Co-authored-by: Will Pfleger <pfleger.will@gmail.com>
Signed-off-by: Will Pfleger <pfleger.will@gmail.com>
This commit is contained in:
Duncan
2026-08-21 20:10:09 -04:00
committed by Will Pfleger
parent a05280f164
commit 1046e88904
4 changed files with 371 additions and 24 deletions

View File

@@ -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();
}
});

View File

@@ -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<void> {
private async doPublish(
store: ChannelSectionStore,
gen: number,
): Promise<void> {
// 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);
}
}

View File

@@ -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;
}
});

View File

@@ -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