fix(sidebar): close pre-publish and per-entry merge convergence holes

Two convergence holes one layer under the pass-1 fixes:

Sections: the pre-publish head check compared the fetched head against
the mutable lastRemoteCreatedAt, which a live event observed during the
debounce window already advanced to that same head — equality fell
through to publish and the local blob overwrote a remote that became
head after the edit was queued. Freeze a canonical head baseline
(created_at, id) at publishSections and compare the fetched head against
that generation baseline instead, adopting when the head advanced.

Stars/mutes: applyRemote admits the canonical lower-id winner but then
mergeStores resolved equal per-entry updatedAt as local/prev-wins, so a
stale larger-id value delivered first survived and undid the winner. Add
mergeApplyingRemote which resolves an entry-timestamp tie toward the
canonical incoming blob while keeping strictly-newer local entries.

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 21:12:39 -04:00
committed by Will Pfleger
parent d5d83e0909
commit 53ba5ab4b1
8 changed files with 404 additions and 22 deletions

View File

@@ -111,6 +111,30 @@ export function writeChannelMutesStore(
export function mergeStores(
local: ChannelMuteStore,
remote: ChannelMuteStore,
): ChannelMuteStore {
return mergeStoresWithTie(local, remote, false);
}
/**
* Merge a remote store that has already won the event-level canonical tie-break
* (`created_at DESC, id ASC`) into the local store, resolving a per-entry
* `updatedAt` tie in favour of the *remote* value. Once the comparator has
* chosen this remote event as the stored winner, its per-entry values must
* survive, or a stale value from a superseded larger-id event delivered first
* would win the merge and silently undo the canonical winner. Strictly-newer
* local per-entry edits (`l.updatedAt > r.updatedAt`) still win.
*/
export function mergeApplyingRemote(
local: ChannelMuteStore,
remote: ChannelMuteStore,
): ChannelMuteStore {
return mergeStoresWithTie(local, remote, true);
}
function mergeStoresWithTie(
local: ChannelMuteStore,
remote: ChannelMuteStore,
preferRemoteOnTie: boolean,
): ChannelMuteStore {
const allIds = new Set([
...Object.keys(local.channels),
@@ -121,7 +145,10 @@ export function mergeStores(
const l = local.channels[id];
const r = remote.channels[id];
if (l && r) {
merged[id] = l.updatedAt >= r.updatedAt ? l : r;
const localWins = preferRemoteOnTie
? l.updatedAt > r.updatedAt
: l.updatedAt >= r.updatedAt;
merged[id] = localWins ? l : r;
} else {
merged[id] = (l ?? r) as ChannelMuteEntry;
}

View File

@@ -533,3 +533,106 @@ test("overlapping publishes: older completion does not erase a newer queued edit
mock.reset();
}
});
// 7. Live remote during debounce (pass-2 finding 1): a remote head accepted
// while a local edit is debouncing must be adopted at pre-publish, not
// overwritten. The live event advances the watermark before doPublish runs,
// so comparing the fetched head against the mutable watermark would see
// equality and publish over the newer remote. The pre-publish check compares
// against the baseline frozen at publishSections instead. Mutation: comparing
// against lastRemoteCreatedAt rather than publishBaseline republishes local.
test("live remote during debounce is adopted at pre-publish, not overwritten", async () => {
const remoteEvent = {
id: "remote-event",
pubkey: "pk-livedebounce",
content: "good-cipher",
created_at: 1_700_000_100,
kind: 30078,
tags: [["d", "channel-sections"]],
sig: "s",
};
// Pre-publish fetch returns the same live head that arrived during debounce.
mock.method(relayClient, "fetchEvents", () => Promise.resolve([remoteEvent]));
const publishCalls = [];
mock.method(relayClient, "publishEvent", (...args) => {
publishCalls.push(args);
return Promise.resolve();
});
let live = null;
mock.method(relayClient, "subscribeLive", async (_filter, cb) => {
live = cb;
return async () => {};
});
const storage = new Map();
const timers = new Map();
let nextId = 1;
const fakeWindow = {
localStorage: {
getItem: (k) => storage.get(k) ?? null,
setItem: (k, v) => storage.set(k, v),
removeItem: (k) => storage.delete(k),
},
setTimeout: (fn, ms) => {
const id = nextId++;
timers.set(id, { fn, ms });
return id;
},
clearTimeout: (id) => timers.delete(id),
};
const fireDelay = async (ms) => {
const entry = [...timers.entries()].find(([, v]) => v.ms === ms);
assert.ok(entry, `expected a timer scheduled at ${ms}ms`);
timers.delete(entry[0]);
entry[1].fn();
for (let i = 0; i < 50; i++) await Promise.resolve();
};
const restore = installFakeWindow(fakeWindow);
const tauri = installTauriMock(
JSON.stringify({
version: 1,
sections: [{ id: "remote", name: "Remote", order: 0 }],
assignments: {},
}),
);
const outboxKey = `buzz-channel-sections-outbox.v1:pk-livedebounce:${RELAY_KEY}`;
try {
const manager = new ChannelSectionSyncManager("pk-livedebounce", RELAY);
const adopted = [];
manager.setOnRemoteAdopted((remote) => adopted.push(remote.eventId));
await manager.subscribeToSections(() => {});
assert.ok(live, "live subscription installed");
manager.publishSections(
makeSectionsStore([{ id: "local", name: "Local", order: 0 }]),
);
// A genuinely later remote head is accepted and delivered while local is
// pending — this advances the watermark past the frozen baseline.
live(remoteEvent);
for (let i = 0; i < 50; i++) await Promise.resolve();
await fireDelay(2000);
assert.equal(
publishCalls.length,
0,
"later remote head must prevent the local publish",
);
assert.deepEqual(
adopted,
["remote-event"],
"the remote accepted after the edit began must be adopted",
);
assert.equal(manager.getPendingStore(), null, "pending cleared on adopt");
assert.equal(
storage.get(outboxKey),
undefined,
"outbox cleared on adopt so the loser can't replay",
);
manager.destroy();
} finally {
tauri.restore();
restore();
mock.reset();
}
});

View File

@@ -57,6 +57,34 @@ type PublishDecision =
| { kind: "publish"; store: ChannelSectionStore }
| { kind: "adopt"; remote: RemoteSections };
/**
* The canonical remote head as it stood when an edit was queued. The pre-publish
* check compares the fetched head against this frozen baseline — never against
* the mutable in-memory watermark, which a live event observed during the
* debounce window may already have advanced to that same head (silently
* suppressing the adopt).
*/
type PublishBaseline = { createdAt: number; eventId: string };
/**
* True when `head` is the canonical winner over the baseline the edit was queued
* against — i.e. the head advanced since the edit began. Canonical order is
* `created_at DESC, id ASC`: a strictly-later head wins, and a same-second head
* wins only with a strictly-lower id. A same-second head is comparable only once
* the baseline id is known (empty id = no prior head seen → not superseded).
*/
function remoteAdvancedSince(
head: RemoteSections,
baseline: PublishBaseline,
): boolean {
if (head.createdAt > baseline.createdAt) return true;
return (
head.createdAt === baseline.createdAt &&
baseline.eventId !== "" &&
head.eventId < baseline.eventId
);
}
async function decryptAndParse(
event: RelayEvent,
): Promise<RemoteSections | null> {
@@ -77,6 +105,12 @@ export class ChannelSectionSyncManager {
private retryTimer: number | null = null;
private retryDelayMs = RETRY_BASE_MS;
private lastRemoteCreatedAt: number;
// Canonical best head observed so far (`created_at DESC, id ASC`). Frozen into
// a per-edit baseline at publishSections so the pre-publish check can tell
// whether the head advanced *since the edit was queued*, independent of the
// mutable watermark that a live event during the debounce window may advance.
private lastRemoteHead: PublishBaseline = { createdAt: 0, eventId: "" };
private publishBaseline: PublishBaseline = { createdAt: 0, eventId: "" };
private pendingStore: ChannelSectionStore | null = null;
// Monotonic id for the current pending edit. Every publishSections() bumps
// it; every scheduled publish/retry captures the value it was queued for.
@@ -118,7 +152,7 @@ export class ChannelSectionSyncManager {
// An event exists — record its created_at regardless of whether we can
// decrypt it, so seed-publish is blocked even when the payload is
// unreadable (e.g. wrong key).
this.recordRemoteHead(event.created_at);
this.recordRemoteHead(event.created_at, event.id);
const result = await decryptAndParse(event);
if (!result) {
return { status: "failed", createdAt: event.created_at };
@@ -134,11 +168,22 @@ export class ChannelSectionSyncManager {
}
}
/** Update in-memory + persisted watermark. */
private recordRemoteHead(createdAt: number): void {
/** Update in-memory + persisted watermark and the canonical head tuple. */
private recordRemoteHead(createdAt: number, eventId: string): void {
if (createdAt > this.lastRemoteCreatedAt) {
this.lastRemoteCreatedAt = createdAt;
}
// Track the canonical-best head (`created_at DESC, id ASC`): a later head
// always wins; a same-second head wins only with a strictly-lower id. This
// mirrors the relay's stored winner so a frozen baseline reflects reality.
if (
createdAt > this.lastRemoteHead.createdAt ||
(createdAt === this.lastRemoteHead.createdAt &&
(this.lastRemoteHead.eventId === "" ||
eventId < this.lastRemoteHead.eventId))
) {
this.lastRemoteHead = { createdAt, eventId };
}
advanceWatermark(this.pubkey, BLOB_TYPE, this.relayUrl, createdAt);
}
@@ -176,7 +221,7 @@ export class ChannelSectionSyncManager {
* and always safe) so the newer edit stamps above this head.
*/
private adoptRemote(remote: RemoteSections, gen: number): void {
this.recordRemoteHead(remote.createdAt);
this.recordRemoteHead(remote.createdAt, remote.eventId);
if (gen !== this.pendingGeneration) return;
this.pendingStore = null;
clearChannelSectionsOutbox(this.pubkey, this.relayUrl);
@@ -200,6 +245,13 @@ export class ChannelSectionSyncManager {
publishSections(store: ChannelSectionStore): void {
this.pendingStore = store;
const gen = ++this.pendingGeneration;
// Freeze the canonical head this edit is racing against. The pre-publish
// check compares the fetched head against this baseline, not the mutable
// watermark — a live event applied during the debounce window advances the
// watermark to the new head, which would otherwise make the pre-publish
// comparison see equality and fall through to a publish that overwrites a
// remote that became head *after* this edit was queued.
this.publishBaseline = { ...this.lastRemoteHead };
// Persist synchronously so an edit made <2s before quit/community-switch
// survives teardown and resumes on next mount (fix-3 durable outbox).
writeChannelSectionsOutbox(this.pubkey, store, this.relayUrl);
@@ -231,18 +283,17 @@ export class ChannelSectionSyncManager {
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 adopt.
const headBeforeFetch = this.lastRemoteCreatedAt;
this.recordRemoteHead(event.created_at);
const remote = await decryptAndParse(event);
// Record the head after decrypt attempt so the watermark/head-tuple
// advance even for an undecryptable payload.
this.recordRemoteHead(event.created_at, event.id);
if (!remote) return { kind: "publish", store };
// Sections use whole-blob LWW: a newer remote head wins, and the local
// edit is adopted-away rather than silently republished as remote content
// while the UI keeps showing the edit.
if (remote.createdAt > headBeforeFetch) {
// Sections use whole-blob LWW. Compare the fetched head against the
// baseline frozen when this edit was queued — NOT the live watermark,
// which a passive live event during debounce may already have advanced to
// this same head. If the canonical head advanced since the edit began, the
// local edit lost and is adopted-away rather than republished over it.
if (remoteAdvancedSince(remote, this.publishBaseline)) {
return { kind: "adopt", remote };
}
return { kind: "publish", store };
@@ -348,7 +399,7 @@ export class ChannelSectionSyncManager {
"Timed out publishing channel sections.",
"Failed to publish channel sections.",
);
this.recordRemoteHead(event.created_at);
this.recordRemoteHead(event.created_at, 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) {
@@ -393,7 +444,7 @@ export class ChannelSectionSyncManager {
if (event.pubkey !== this.pubkey) return;
// Record the raw head before decrypt so an undecryptable live event
// still advances the watermark and blocks future seed-publish.
this.recordRemoteHead(event.created_at);
this.recordRemoteHead(event.created_at, event.id);
void decryptAndParse(event).then((result) => {
if (result) {
onUpdate(result);

View File

@@ -111,6 +111,30 @@ export function writeChannelStarsStore(
export function mergeStores(
local: ChannelStarStore,
remote: ChannelStarStore,
): ChannelStarStore {
return mergeStoresWithTie(local, remote, false);
}
/**
* Merge a remote store that has already won the event-level canonical tie-break
* (`created_at DESC, id ASC`) into the local store, resolving a per-entry
* `updatedAt` tie in favour of the *remote* value. Once the comparator has
* chosen this remote event as the stored winner, its per-entry values must
* survive, or a stale value from a superseded larger-id event delivered first
* would win the merge and silently undo the canonical winner. Strictly-newer
* local per-entry edits (`l.updatedAt > r.updatedAt`) still win.
*/
export function mergeApplyingRemote(
local: ChannelStarStore,
remote: ChannelStarStore,
): ChannelStarStore {
return mergeStoresWithTie(local, remote, true);
}
function mergeStoresWithTie(
local: ChannelStarStore,
remote: ChannelStarStore,
preferRemoteOnTie: boolean,
): ChannelStarStore {
const allIds = new Set([
...Object.keys(local.channels),
@@ -121,7 +145,10 @@ export function mergeStores(
const l = local.channels[id];
const r = remote.channels[id];
if (l && r) {
merged[id] = l.updatedAt >= r.updatedAt ? l : r;
const localWins = preferRemoteOnTie
? l.updatedAt > r.updatedAt
: l.updatedAt >= r.updatedAt;
merged[id] = localWins ? l : r;
} else {
merged[id] = (l ?? r) as ChannelStarEntry;
}

View File

@@ -163,3 +163,89 @@ test("equal-timestamp tie-break applies the lower event id (relay canonical winn
window.__TAURI_INTERNALS__ = origTauri;
}
});
// Pass-2 finding 2: the comparator admitting the canonical lower id is
// necessary but not sufficient. Mutes are a per-entry store, so applyRemote
// merges the incoming blob into local state. On the SAME channel, a stale
// larger-id event delivered first (muted=true) must not survive the merge once
// the canonical lower-id winner (muted=false) arrives at the same entry
// `updatedAt`. Mutation: reverting the apply path to mergeStores (local/prev
// wins on tie) keeps the stale muted=true value.
test("canonical lower-id unmute replaces a stale larger-id mute at equal entry timestamp", async () => {
const { act, cleanup, renderHook } = await import("@testing-library/react");
const { relayClient } = await import("@/shared/api/relayClient");
const { useChannelMutes } = await import("./useChannelMutes.ts");
const origFetch = relayClient.fetchEvents;
const origLive = relayClient.subscribeLive;
const origReconnect = relayClient.subscribeToReconnects;
const origTauri = window.__TAURI_INTERNALS__;
let live = null;
relayClient.fetchEvents = async () => [];
relayClient.subscribeLive = async (_f, cb) => {
live = cb;
return async () => {};
};
relayClient.subscribeToReconnects = () => () => {};
// Both events target the SAME channel `shared` at the same entry updatedAt.
// The larger id `bbbb` says muted; the canonical lower id `aaaa` says not.
window.__TAURI_INTERNALS__ = {
invoke: (cmd, args) => {
if (cmd === "nip44_decrypt_from_self") {
const canonicalLowerId = args?.ciphertext === "aaaa";
return Promise.resolve(
JSON.stringify({
version: 1,
channels: { shared: { muted: !canonicalLowerId, updatedAt: 100 } },
}),
);
}
return Promise.reject(new Error(`unmocked ${cmd}`));
},
};
const pubkey = "pk-mute-shared-tie";
const relayUrl = "wss://r.tie";
let hook = null;
try {
await act(async () => {
hook = renderHook(() => useChannelMutes(pubkey, relayUrl));
await Promise.resolve();
await Promise.resolve();
await Promise.resolve();
});
assert.ok(live, "live subscription installed");
const deliver = async (id) => {
await act(async () => {
live({
id,
pubkey,
created_at: 1000,
content: id,
kind: 30078,
tags: [["d", "channel-mutes"]],
sig: "s",
});
for (let i = 0; i < 20; i++) await Promise.resolve();
});
};
await deliver("bbbb"); // stale larger-id head says muted
await deliver("aaaa"); // canonical lower-id winner says unmuted
assert.equal(
hook.result.current.mutedChannelIds.has("shared"),
false,
"canonical lower-id unmute must replace the stale larger-id mute",
);
hook.unmount();
} finally {
cleanup();
relayClient.fetchEvents = origFetch;
relayClient.subscribeLive = origLive;
relayClient.subscribeToReconnects = origReconnect;
window.__TAURI_INTERNALS__ = origTauri;
}
});

View File

@@ -4,7 +4,7 @@ import { relayClient } from "@/shared/api/relayClient";
import {
boundMuteStore,
DEFAULT_STORE,
mergeStores,
mergeApplyingRemote,
mutedChannelIdsFromStore,
readChannelMutesStore,
storageKey,
@@ -85,7 +85,7 @@ export function useChannelMutes(
lastAppliedRemoteTs.current = remote.createdAt;
lastAppliedEventId.current = remote.eventId;
managerRef.current?.cancelPendingMutePublish();
const merged = mergeStores(prev, remote.store);
const merged = mergeApplyingRemote(prev, remote.store);
if (!writeChannelMutesStore(pubkey, merged)) return prev;
return merged;
};

View File

@@ -163,3 +163,91 @@ test("equal-timestamp tie-break applies the lower event id (relay canonical winn
window.__TAURI_INTERNALS__ = origTauri;
}
});
// Pass-2 finding 2: the comparator admitting the canonical lower id is
// necessary but not sufficient. Stars are a per-entry store, so applyRemote
// merges the incoming blob into local state. On the SAME channel, a stale
// larger-id event delivered first (starred=true) must not survive the merge
// once the canonical lower-id winner (starred=false) arrives at the same entry
// `updatedAt`. Mutation: reverting the apply path to mergeStores (local/prev
// wins on tie) keeps the stale starred=true value.
test("canonical lower-id unstar replaces a stale larger-id star at equal entry timestamp", async () => {
const { act, cleanup, renderHook } = await import("@testing-library/react");
const { relayClient } = await import("@/shared/api/relayClient");
const { useChannelStars } = await import("./useChannelStars.ts");
const origFetch = relayClient.fetchEvents;
const origLive = relayClient.subscribeLive;
const origReconnect = relayClient.subscribeToReconnects;
const origTauri = window.__TAURI_INTERNALS__;
let live = null;
relayClient.fetchEvents = async () => [];
relayClient.subscribeLive = async (_f, cb) => {
live = cb;
return async () => {};
};
relayClient.subscribeToReconnects = () => () => {};
// Both events target the SAME channel `shared` at the same entry updatedAt.
// The larger id `bbbb` says starred; the canonical lower id `aaaa` says not.
window.__TAURI_INTERNALS__ = {
invoke: (cmd, args) => {
if (cmd === "nip44_decrypt_from_self") {
const canonicalLowerId = args?.ciphertext === "aaaa";
return Promise.resolve(
JSON.stringify({
version: 1,
channels: {
shared: { starred: !canonicalLowerId, updatedAt: 100 },
},
}),
);
}
return Promise.reject(new Error(`unmocked ${cmd}`));
},
};
const pubkey = "pk-star-shared-tie";
const relayUrl = "wss://r.tie";
let hook = null;
try {
await act(async () => {
hook = renderHook(() => useChannelStars(pubkey, relayUrl));
await Promise.resolve();
await Promise.resolve();
await Promise.resolve();
});
assert.ok(live, "live subscription installed");
const deliver = async (id) => {
await act(async () => {
live({
id,
pubkey,
created_at: 1000,
content: id,
kind: 30078,
tags: [["d", "channel-stars"]],
sig: "s",
});
for (let i = 0; i < 20; i++) await Promise.resolve();
});
};
await deliver("bbbb"); // stale larger-id head says starred
await deliver("aaaa"); // canonical lower-id winner says unstarred
assert.equal(
hook.result.current.starredChannelIds.has("shared"),
false,
"canonical lower-id unstar must replace the stale larger-id star",
);
hook.unmount();
} finally {
cleanup();
relayClient.fetchEvents = origFetch;
relayClient.subscribeLive = origLive;
relayClient.subscribeToReconnects = origReconnect;
window.__TAURI_INTERNALS__ = origTauri;
}
});

View File

@@ -4,7 +4,7 @@ import { relayClient } from "@/shared/api/relayClient";
import {
boundStarStore,
DEFAULT_STORE,
mergeStores,
mergeApplyingRemote,
readChannelStarsStore,
starredChannelIdsFromStore,
storageKey,
@@ -85,7 +85,7 @@ export function useChannelStars(
lastAppliedRemoteTs.current = remote.createdAt;
lastAppliedEventId.current = remote.eventId;
managerRef.current?.cancelPendingStarPublish();
const merged = mergeStores(prev, remote.store);
const merged = mergeApplyingRemote(prev, remote.store);
if (!writeChannelStarsStore(pubkey, merged)) return prev;
return merged;
};