mirror of
https://github.com/block/buzz.git
synced 2026-10-29 05:21:31 +00:00
fix(sidebar): serialize section publishes and scope remote-tie merge
Each prior round patched one cross-generation interleaving and opened another a layer deeper. Kill the race class structurally instead. Sections: serialize publish cycles (one in-flight at a time; a newer edit queued mid-cycle defers and the completion re-drives it). The per-edit pre-publish baseline is frozen at queue time, so a genuine remote observed during the debounce window still adopts, while our own accepted head is folded forward via canonicalMax so a stale generation's own write is never mistaken for a competing remote and adopted away. Dual generation guards in doPublish (post-fetch and pre-publish) stop a stale generation signing or publishing after a newer edit exists. Stars/mutes: scope mergeApplyingRemote (remote-wins on entry-tie) and the pending-publish cancel to fire only on a canonical supersession of an already-applied same-timestamp larger-id head. Every other application (bootstrap/live/newer-timestamp) keeps local-wins mergeStores and does not cancel the pending publish, so a later same-second local click is no longer clobbered by an older remote entry that decrypts late. Co-authored-by: Will Pfleger <pfleger.will@gmail.com> Signed-off-by: Will Pfleger <pfleger.will@gmail.com>
This commit is contained in:
@@ -636,3 +636,197 @@ test("live remote during debounce is adopted at pre-publish, not overwritten", a
|
||||
mock.reset();
|
||||
}
|
||||
});
|
||||
|
||||
// 8. Serialized generations (fix round 3, pass-3 finding 1): an older in-flight
|
||||
// publish that completes after a newer edit is queued must NOT be mistaken
|
||||
// for a remote that advanced past the newer edit's baseline. A blocks in
|
||||
// publishEvent; B is queued; A succeeds; B's pre-publish fetch returns A's
|
||||
// accepted head. Because the baseline is frozen at B's own cycle start —
|
||||
// after A's completion recorded its head — B publishes above A instead of
|
||||
// adopting it. Mutation: freezing the baseline at publishSections (before the
|
||||
// prior cycle completes) makes B see A as a post-baseline remote and adopt it.
|
||||
test("serialized generations: older completion does not make the newer edit adopt it", async () => {
|
||||
let releaseFirst = null;
|
||||
let publishCalls = 0;
|
||||
let storedHead = [];
|
||||
mock.method(relayClient, "fetchEvents", () => Promise.resolve(storedHead));
|
||||
mock.method(relayClient, "publishEvent", (event) => {
|
||||
publishCalls++;
|
||||
if (publishCalls === 1) {
|
||||
// A's publish blocks; when it resolves, its event becomes the stored head
|
||||
// the next pre-publish fetch will return.
|
||||
return new Promise((res) => {
|
||||
releaseFirst = () => {
|
||||
storedHead = [
|
||||
{
|
||||
id: "event-a",
|
||||
pubkey: "pk-serial",
|
||||
content: "good-cipher",
|
||||
created_at: event.created_at,
|
||||
kind: 30078,
|
||||
tags: [["d", "channel-sections"]],
|
||||
sig: "s",
|
||||
},
|
||||
];
|
||||
res();
|
||||
};
|
||||
});
|
||||
}
|
||||
return Promise.resolve();
|
||||
});
|
||||
const storage = new Map();
|
||||
const timers = new Map();
|
||||
let nextId = 1;
|
||||
const fakeWindow = {
|
||||
localStorage: {
|
||||
getItem: (k) => storage.get(k) ?? null,
|
||||
setItem: (k, v) => storage.set(k, v),
|
||||
removeItem: (k) => storage.delete(k),
|
||||
},
|
||||
setTimeout: (fn, ms) => {
|
||||
const id = nextId++;
|
||||
timers.set(id, { fn, ms });
|
||||
return id;
|
||||
},
|
||||
clearTimeout: (id) => timers.delete(id),
|
||||
};
|
||||
const fireDelay = async (ms) => {
|
||||
const entry = [...timers.entries()].find(([, v]) => v.ms === ms);
|
||||
assert.ok(entry, `expected a timer scheduled at ${ms}ms`);
|
||||
timers.delete(entry[0]);
|
||||
entry[1].fn();
|
||||
for (let i = 0; i < 100; i++) await Promise.resolve();
|
||||
};
|
||||
const restore = installFakeWindow(fakeWindow);
|
||||
// A's decrypted head must parse; the pre-publish check reads its created_at/id.
|
||||
const tauri = installTauriMock(
|
||||
JSON.stringify({
|
||||
version: 1,
|
||||
sections: [{ id: "a", name: "A", order: 0 }],
|
||||
assignments: {},
|
||||
}),
|
||||
);
|
||||
try {
|
||||
const manager = new ChannelSectionSyncManager("pk-serial", RELAY);
|
||||
const adopted = [];
|
||||
manager.setOnRemoteAdopted((remote) => adopted.push(remote.eventId));
|
||||
|
||||
manager.publishSections(
|
||||
makeSectionsStore([{ id: "a", name: "A", order: 0 }]),
|
||||
);
|
||||
await fireDelay(2000); // A's cycle → publishEvent(A) blocks
|
||||
while (releaseFirst === null) await Promise.resolve();
|
||||
|
||||
// B is queued while A is still in flight; its cycle must defer.
|
||||
manager.publishSections(
|
||||
makeSectionsStore([{ id: "b", name: "B", order: 0 }]),
|
||||
);
|
||||
assert.deepEqual(
|
||||
manager.getPendingStore()?.sections.map((s) => s.id),
|
||||
["b"],
|
||||
"B is the pending edit while A is in flight",
|
||||
);
|
||||
|
||||
releaseFirst(); // A completes, recording its head; the freed lane drives B
|
||||
for (let i = 0; i < 100; i++) await Promise.resolve();
|
||||
// If B's cycle did not auto-drive on the freed lane, its debounce timer is
|
||||
// still pending — fire it.
|
||||
if ([...timers.values()].some((t) => t.ms === 2000)) await fireDelay(2000);
|
||||
|
||||
assert.deepEqual(adopted, [], "B must not adopt the older generation A");
|
||||
assert.equal(
|
||||
publishCalls,
|
||||
2,
|
||||
"B publishes above A rather than adopting A's accepted head",
|
||||
);
|
||||
assert.equal(
|
||||
manager.getPendingStore(),
|
||||
null,
|
||||
"B's pending clears via its own successful publish, not A's completion",
|
||||
);
|
||||
manager.destroy();
|
||||
} finally {
|
||||
tauri.restore();
|
||||
restore();
|
||||
mock.reset();
|
||||
}
|
||||
});
|
||||
|
||||
// 9. Serialized generations (fix round 3, pass-3 finding 1): a stale generation
|
||||
// must never sign/publish after a newer edit is queued. A blocks in the
|
||||
// pre-publish fetch; B is queued during that await; A must abort before
|
||||
// signing. Mutation: dropping the post-fetch generation re-check in doPublish
|
||||
// lets the stale A continue to publishEvent.
|
||||
test("serialized generations: a stale generation aborts before publishing", async () => {
|
||||
let releaseFetch = null;
|
||||
const publishCalls = [];
|
||||
mock.method(
|
||||
relayClient,
|
||||
"fetchEvents",
|
||||
() =>
|
||||
new Promise((res) => {
|
||||
releaseFetch = () => res([]);
|
||||
}),
|
||||
);
|
||||
mock.method(relayClient, "publishEvent", (...args) => {
|
||||
publishCalls.push(args);
|
||||
return Promise.resolve();
|
||||
});
|
||||
const storage = new Map();
|
||||
const timers = new Map();
|
||||
let nextId = 1;
|
||||
const fakeWindow = {
|
||||
localStorage: {
|
||||
getItem: (k) => storage.get(k) ?? null,
|
||||
setItem: (k, v) => storage.set(k, v),
|
||||
removeItem: (k) => storage.delete(k),
|
||||
},
|
||||
setTimeout: (fn, ms) => {
|
||||
const id = nextId++;
|
||||
timers.set(id, { fn, ms });
|
||||
return id;
|
||||
},
|
||||
clearTimeout: (id) => timers.delete(id),
|
||||
};
|
||||
const fireDelay = async (ms) => {
|
||||
const entry = [...timers.entries()].find(([, v]) => v.ms === ms);
|
||||
assert.ok(entry, `expected a timer scheduled at ${ms}ms`);
|
||||
timers.delete(entry[0]);
|
||||
entry[1].fn();
|
||||
for (let i = 0; i < 100; i++) await Promise.resolve();
|
||||
};
|
||||
const restore = installFakeWindow(fakeWindow);
|
||||
const tauri = installTauriMock("{}");
|
||||
try {
|
||||
const manager = new ChannelSectionSyncManager("pk-stale", RELAY);
|
||||
manager.publishSections(
|
||||
makeSectionsStore([{ id: "a", name: "A", order: 0 }]),
|
||||
);
|
||||
await fireDelay(2000); // A's cycle → doPublish(A) awaits fetchEvents
|
||||
while (releaseFetch === null) await Promise.resolve();
|
||||
|
||||
// B is queued while A is blocked in its pre-publish fetch.
|
||||
manager.publishSections(
|
||||
makeSectionsStore([{ id: "b", name: "B", order: 0 }]),
|
||||
);
|
||||
|
||||
releaseFetch(); // A's fetch resolves; A must see gen moved and abort
|
||||
for (let i = 0; i < 100; i++) await Promise.resolve();
|
||||
|
||||
assert.equal(
|
||||
publishCalls.length,
|
||||
0,
|
||||
"stale generation A must not sign/publish after B was queued",
|
||||
);
|
||||
assert.deepEqual(
|
||||
manager.getPendingStore()?.sections.map((s) => s.id),
|
||||
["b"],
|
||||
"B remains the pending edit, owning convergence",
|
||||
);
|
||||
manager.destroy();
|
||||
} finally {
|
||||
tauri.restore();
|
||||
restore();
|
||||
mock.reset();
|
||||
}
|
||||
});
|
||||
|
||||
@@ -85,6 +85,22 @@ function remoteAdvancedSince(
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
* 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<RemoteSections | null> {
|
||||
@@ -110,6 +126,14 @@ export class ChannelSectionSyncManager {
|
||||
// whether the head advanced *since the edit was queued*, independent of the
|
||||
// mutable watermark that a live event during the debounce window may advance.
|
||||
private lastRemoteHead: PublishBaseline = { createdAt: 0, eventId: "" };
|
||||
// The canonical head this pending edit is racing against, frozen when the
|
||||
// edit was queued (publishSections) and advanced ONLY by our own successful
|
||||
// publishes. Freezing at queue time is what makes a genuine remote observed
|
||||
// during the debounce window still adopt-worthy (pass-2): the mutable
|
||||
// watermark advanced to that remote, but the baseline did not. Folding our
|
||||
// own published head forward is what stops a newer edit from adopting an
|
||||
// older generation's own accepted write (pass-3): our prior publish is our
|
||||
// baseline, not a competing remote.
|
||||
private publishBaseline: PublishBaseline = { createdAt: 0, eventId: "" };
|
||||
private pendingStore: ChannelSectionStore | null = null;
|
||||
// Monotonic id for the current pending edit. Every publishSections() bumps
|
||||
@@ -118,6 +142,13 @@ export class ChannelSectionSyncManager {
|
||||
// compare-and-swap on this generation, so an older in-flight publish can
|
||||
// never erase a newer edit that arrived while it was in flight.
|
||||
private pendingGeneration = 0;
|
||||
// Publish cycles are serialized: at most one runs at a time. A newer edit
|
||||
// queued while a cycle is in flight does NOT start its own concurrent cycle;
|
||||
// it defers, and the in-flight cycle's completion schedules it. Serialization
|
||||
// guarantees there is never more than one baseline/fetch/publish sequence
|
||||
// touching shared manager state, so a stale generation can never sign or
|
||||
// publish after a newer edit exists.
|
||||
private publishInFlight = false;
|
||||
private lastPublishedStore: ChannelSectionStore | null = null;
|
||||
private destroyed = false;
|
||||
// Set by the hook so an adopted remote head (local edit lost LWW, or a relay
|
||||
@@ -244,13 +275,16 @@ 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.pendingGeneration;
|
||||
// Freeze the canonical head this edit is racing against at queue time. The
|
||||
// pre-publish check compares the fetched head against this baseline, not the
|
||||
// mutable watermark — a live event applied during the debounce window
|
||||
// advances the watermark to a new remote head, which would otherwise make
|
||||
// the pre-publish comparison see equality and fall through to a publish that
|
||||
// overwrites a remote that became head *after* this edit was queued. The
|
||||
// baseline only advances via our own successful publishes (see doPublish),
|
||||
// so a prior generation's own accepted write is folded in rather than
|
||||
// mistaken for a competing remote.
|
||||
this.publishBaseline = { ...this.lastRemoteHead };
|
||||
// Persist synchronously so an edit made <2s before quit/community-switch
|
||||
// survives teardown and resumes on next mount (fix-3 durable outbox).
|
||||
@@ -266,10 +300,40 @@ export class ChannelSectionSyncManager {
|
||||
this.retryDelayMs = RETRY_BASE_MS;
|
||||
this.debounceTimer = window.setTimeout(() => {
|
||||
this.debounceTimer = null;
|
||||
void this.doPublish(store, gen);
|
||||
this.startCycle();
|
||||
}, DEBOUNCE_MS);
|
||||
}
|
||||
|
||||
/**
|
||||
* Serialize publish cycles: at most one runs at a time. A debounce/retry
|
||||
* timer that fires while a cycle is in flight defers — the in-flight cycle's
|
||||
* completion re-drives if a pending edit still needs publishing. This kills
|
||||
* the cross-generation race class by construction: a newer edit queued during
|
||||
* a cycle cannot start its own concurrent cycle, so there is never more than
|
||||
* one baseline/fetch/publish sequence competing over shared manager state.
|
||||
*/
|
||||
private startCycle(): void {
|
||||
if (this.destroyed || this.pendingStore === null) return;
|
||||
if (this.publishInFlight) return;
|
||||
const store = this.pendingStore;
|
||||
const gen = this.pendingGeneration;
|
||||
this.publishInFlight = true;
|
||||
void this.doPublish(store, gen).finally(() => {
|
||||
this.publishInFlight = false;
|
||||
// A newer edit queued during the cycle (or a cycle that ended without
|
||||
// clearing its pending edit) still needs publishing and has no timer
|
||||
// pending to drive it — drive the next cycle now that the lane is free.
|
||||
if (
|
||||
!this.destroyed &&
|
||||
this.pendingStore !== null &&
|
||||
this.debounceTimer === null &&
|
||||
this.retryTimer === null
|
||||
) {
|
||||
this.startCycle();
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
private async fetchOwnBlobBeforePublish(
|
||||
store: ChannelSectionStore,
|
||||
): Promise<PublishDecision> {
|
||||
@@ -335,12 +399,11 @@ export class ChannelSectionSyncManager {
|
||||
// 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, gen);
|
||||
this.startCycle();
|
||||
}, delay);
|
||||
}
|
||||
|
||||
@@ -357,6 +420,11 @@ export class ChannelSectionSyncManager {
|
||||
// was awaited (community switch during in-flight fetch). If so, abort
|
||||
// before touching the relay.
|
||||
if (this.destroyed) return;
|
||||
// A newer edit was queued while we awaited the pre-publish fetch. It owns
|
||||
// convergence now; abort so we neither publish this stale store nor adopt
|
||||
// over the newer pending edit. The serialized cycle re-drives for the
|
||||
// newer generation once this one unwinds.
|
||||
if (gen !== this.pendingGeneration) return;
|
||||
if (decision.kind === "adopt") {
|
||||
this.adoptRemote(decision.remote, gen);
|
||||
return;
|
||||
@@ -392,14 +460,29 @@ export class ChannelSectionSyncManager {
|
||||
});
|
||||
// Final guard immediately before the network call — sign/encrypt are
|
||||
// synchronous-ish but cheap; the relay socket may have moved to a
|
||||
// different community by the time we reach this point.
|
||||
if (this.destroyed) return;
|
||||
// different community by the time we reach this point, or a newer edit
|
||||
// may have been queued during the encrypt/sign await (invariant: a stale
|
||||
// generation never signs/publishes after a newer edit exists).
|
||||
if (this.destroyed || gen !== this.pendingGeneration) return;
|
||||
await relayClient.publishEvent(
|
||||
event,
|
||||
"Timed out publishing channel sections.",
|
||||
"Failed to publish channel sections.",
|
||||
);
|
||||
this.recordRemoteHead(event.created_at, event.id);
|
||||
// Fold our own accepted head into the pending edit's baseline. This is
|
||||
// unconditional across generations: even a stale generation's own write,
|
||||
// completing after a newer edit was queued, must advance the current
|
||||
// pending baseline so the newer edit's pre-publish check does not mistake
|
||||
// OUR prior publish for a competing remote and adopt it away (pass-3).
|
||||
// Genuine remotes never fold in here — they only advance the watermark —
|
||||
// so a remote that became head during the debounce window still adopts
|
||||
// (pass-2). canonicalMax keeps the advance monotonic (`created_at DESC,
|
||||
// id ASC`).
|
||||
this.publishBaseline = canonicalMax(this.publishBaseline, {
|
||||
createdAt: event.created_at,
|
||||
eventId: event.id,
|
||||
});
|
||||
// Only claim this store as the published head if it is still the current
|
||||
// edit; a newer edit queued mid-flight owns lastPublishedStore now.
|
||||
if (gen === this.pendingGeneration) {
|
||||
|
||||
@@ -249,3 +249,107 @@ test("canonical lower-id unmute replaces a stale larger-id mute at equal entry t
|
||||
window.__TAURI_INTERNALS__ = origTauri;
|
||||
}
|
||||
});
|
||||
|
||||
// Fix round 3 (pass-3 finding 2): the remote-wins entry-tie merge and the
|
||||
// pending-publish cancel apply ONLY to a canonical supersession (a lower id at
|
||||
// the same event timestamp correcting an already-applied larger id). A plain
|
||||
// live/bootstrap remote must NOT clobber a later same-second local click or
|
||||
// cancel its pending publish. Entry `updatedAt` is whole seconds, so a click at
|
||||
// 100.9s and an older remote entry at 100.1s both carry `updatedAt:100`; the
|
||||
// later local intent must win and keep publishing. Mutation: applying
|
||||
// mergeApplyingRemote + cancel unconditionally lets the delayed remote overwrite
|
||||
// the click and drop its publish.
|
||||
test("delayed same-second remote does not clobber a later local mute or cancel its publish", async () => {
|
||||
const { act, cleanup, renderHook } = await import("@testing-library/react");
|
||||
const { relayClient } = await import("@/shared/api/relayClient");
|
||||
const { useChannelMutes } = await import("./useChannelMutes.ts");
|
||||
|
||||
const origFetch = relayClient.fetchEvents;
|
||||
const origLive = relayClient.subscribeLive;
|
||||
const origReconnect = relayClient.subscribeToReconnects;
|
||||
const origTauri = window.__TAURI_INTERNALS__;
|
||||
const origSetTimeout = window.setTimeout;
|
||||
const origClearTimeout = window.clearTimeout;
|
||||
const origDateNow = Date.now;
|
||||
|
||||
const timers = new Map();
|
||||
let nextTimer = 1;
|
||||
window.setTimeout = (fn, ms) => {
|
||||
const id = nextTimer++;
|
||||
timers.set(id, { fn, ms });
|
||||
return id;
|
||||
};
|
||||
window.clearTimeout = (id) => timers.delete(id);
|
||||
// The local click happens later within second 100.
|
||||
Date.now = () => 100_900;
|
||||
|
||||
let live = null;
|
||||
relayClient.fetchEvents = async () => [];
|
||||
relayClient.subscribeLive = async (_f, cb) => {
|
||||
live = cb;
|
||||
return async () => {};
|
||||
};
|
||||
relayClient.subscribeToReconnects = () => () => {};
|
||||
// The delayed remote entry sits earlier in the same second and says unmuted.
|
||||
window.__TAURI_INTERNALS__ = {
|
||||
invoke: (cmd) => {
|
||||
if (cmd === "nip44_decrypt_from_self")
|
||||
return Promise.resolve(
|
||||
JSON.stringify({
|
||||
version: 1,
|
||||
channels: { shared: { muted: false, updatedAt: 100 } },
|
||||
}),
|
||||
);
|
||||
return Promise.reject(new Error(`unmocked ${cmd}`));
|
||||
},
|
||||
};
|
||||
|
||||
const pubkey = "pk-mute-same-second";
|
||||
const relayUrl = "wss://r.same";
|
||||
let hook = null;
|
||||
try {
|
||||
await act(async () => {
|
||||
hook = renderHook(() => useChannelMutes(pubkey, relayUrl));
|
||||
for (let i = 0; i < 20; i++) await Promise.resolve();
|
||||
});
|
||||
assert.ok(live, "live subscription installed");
|
||||
|
||||
// Local optimistic click: muted=true at updatedAt=100 (Date.now=100.9s).
|
||||
await act(async () => {
|
||||
hook.result.current.muteChannel("shared");
|
||||
});
|
||||
// An older remote entry from the same second decrypts and applies late.
|
||||
await act(async () => {
|
||||
live({
|
||||
id: "remote-before-click",
|
||||
pubkey,
|
||||
created_at: 100,
|
||||
content: "remote",
|
||||
kind: 30078,
|
||||
tags: [["d", "channel-mutes"]],
|
||||
sig: "s",
|
||||
});
|
||||
for (let i = 0; i < 40; i++) await Promise.resolve();
|
||||
});
|
||||
|
||||
assert.equal(
|
||||
hook.result.current.mutedChannelIds.has("shared"),
|
||||
true,
|
||||
"a later same-second local click must survive a delayed older remote",
|
||||
);
|
||||
assert.ok(
|
||||
[...timers.values()].some((t) => t.ms === 2000),
|
||||
"the local pending publish must remain scheduled",
|
||||
);
|
||||
hook.unmount();
|
||||
} finally {
|
||||
cleanup();
|
||||
relayClient.fetchEvents = origFetch;
|
||||
relayClient.subscribeLive = origLive;
|
||||
relayClient.subscribeToReconnects = origReconnect;
|
||||
window.__TAURI_INTERNALS__ = origTauri;
|
||||
window.setTimeout = origSetTimeout;
|
||||
window.clearTimeout = origClearTimeout;
|
||||
Date.now = origDateNow;
|
||||
}
|
||||
});
|
||||
|
||||
@@ -5,6 +5,7 @@ import {
|
||||
boundMuteStore,
|
||||
DEFAULT_STORE,
|
||||
mergeApplyingRemote,
|
||||
mergeStores,
|
||||
mutedChannelIdsFromStore,
|
||||
readChannelMutesStore,
|
||||
storageKey,
|
||||
@@ -82,10 +83,27 @@ export function useChannelMutes(
|
||||
remote.eventId >= lastAppliedEventId.current
|
||||
)
|
||||
return prev;
|
||||
// A canonical supersession corrects an already-applied same-timestamp
|
||||
// LARGER-id head with the true winner: only here may the incoming blob's
|
||||
// per-entry values win an equal-`updatedAt` tie, and only here does the
|
||||
// pending publish (which reflected the superseded head) get cancelled.
|
||||
// Any other application (bootstrap / live / newer timestamp) merges over
|
||||
// optimistic local state with local-wins `mergeStores` and must NOT
|
||||
// cancel a pending local publish — otherwise a later same-second local
|
||||
// click (integer-second `updatedAt`) loses to an older remote entry that
|
||||
// decrypts late, and its publish is silently dropped.
|
||||
const isCanonicalSupersession =
|
||||
remote.createdAt === lastAppliedRemoteTs.current &&
|
||||
lastAppliedEventId.current !== "" &&
|
||||
remote.eventId < lastAppliedEventId.current;
|
||||
lastAppliedRemoteTs.current = remote.createdAt;
|
||||
lastAppliedEventId.current = remote.eventId;
|
||||
managerRef.current?.cancelPendingMutePublish();
|
||||
const merged = mergeApplyingRemote(prev, remote.store);
|
||||
const merged = isCanonicalSupersession
|
||||
? mergeApplyingRemote(prev, remote.store)
|
||||
: mergeStores(prev, remote.store);
|
||||
if (isCanonicalSupersession) {
|
||||
managerRef.current?.cancelPendingMutePublish();
|
||||
}
|
||||
if (!writeChannelMutesStore(pubkey, merged)) return prev;
|
||||
return merged;
|
||||
};
|
||||
|
||||
@@ -251,3 +251,107 @@ test("canonical lower-id unstar replaces a stale larger-id star at equal entry t
|
||||
window.__TAURI_INTERNALS__ = origTauri;
|
||||
}
|
||||
});
|
||||
|
||||
// Fix round 3 (pass-3 finding 2): the remote-wins entry-tie merge and the
|
||||
// pending-publish cancel apply ONLY to a canonical supersession (a lower id at
|
||||
// the same event timestamp correcting an already-applied larger id). A plain
|
||||
// live/bootstrap remote must NOT clobber a later same-second local click or
|
||||
// cancel its pending publish. Entry `updatedAt` is whole seconds, so a click at
|
||||
// 100.9s and an older remote entry at 100.1s both carry `updatedAt:100`; the
|
||||
// later local intent must win and keep publishing. Mutation: applying
|
||||
// mergeApplyingRemote + cancel unconditionally lets the delayed remote overwrite
|
||||
// the click and drop its publish.
|
||||
test("delayed same-second remote does not clobber a later local star or cancel its publish", async () => {
|
||||
const { act, cleanup, renderHook } = await import("@testing-library/react");
|
||||
const { relayClient } = await import("@/shared/api/relayClient");
|
||||
const { useChannelStars } = await import("./useChannelStars.ts");
|
||||
|
||||
const origFetch = relayClient.fetchEvents;
|
||||
const origLive = relayClient.subscribeLive;
|
||||
const origReconnect = relayClient.subscribeToReconnects;
|
||||
const origTauri = window.__TAURI_INTERNALS__;
|
||||
const origSetTimeout = window.setTimeout;
|
||||
const origClearTimeout = window.clearTimeout;
|
||||
const origDateNow = Date.now;
|
||||
|
||||
const timers = new Map();
|
||||
let nextTimer = 1;
|
||||
window.setTimeout = (fn, ms) => {
|
||||
const id = nextTimer++;
|
||||
timers.set(id, { fn, ms });
|
||||
return id;
|
||||
};
|
||||
window.clearTimeout = (id) => timers.delete(id);
|
||||
// The local click happens later within second 100.
|
||||
Date.now = () => 100_900;
|
||||
|
||||
let live = null;
|
||||
relayClient.fetchEvents = async () => [];
|
||||
relayClient.subscribeLive = async (_f, cb) => {
|
||||
live = cb;
|
||||
return async () => {};
|
||||
};
|
||||
relayClient.subscribeToReconnects = () => () => {};
|
||||
// The delayed remote entry sits earlier in the same second and says unstarred.
|
||||
window.__TAURI_INTERNALS__ = {
|
||||
invoke: (cmd) => {
|
||||
if (cmd === "nip44_decrypt_from_self")
|
||||
return Promise.resolve(
|
||||
JSON.stringify({
|
||||
version: 1,
|
||||
channels: { shared: { starred: false, updatedAt: 100 } },
|
||||
}),
|
||||
);
|
||||
return Promise.reject(new Error(`unmocked ${cmd}`));
|
||||
},
|
||||
};
|
||||
|
||||
const pubkey = "pk-star-same-second";
|
||||
const relayUrl = "wss://r.same";
|
||||
let hook = null;
|
||||
try {
|
||||
await act(async () => {
|
||||
hook = renderHook(() => useChannelStars(pubkey, relayUrl));
|
||||
for (let i = 0; i < 20; i++) await Promise.resolve();
|
||||
});
|
||||
assert.ok(live, "live subscription installed");
|
||||
|
||||
// Local optimistic click: starred=true at updatedAt=100 (Date.now=100.9s).
|
||||
await act(async () => {
|
||||
hook.result.current.starChannel("shared");
|
||||
});
|
||||
// An older remote entry from the same second decrypts and applies late.
|
||||
await act(async () => {
|
||||
live({
|
||||
id: "remote-before-click",
|
||||
pubkey,
|
||||
created_at: 100,
|
||||
content: "remote",
|
||||
kind: 30078,
|
||||
tags: [["d", "channel-stars"]],
|
||||
sig: "s",
|
||||
});
|
||||
for (let i = 0; i < 40; i++) await Promise.resolve();
|
||||
});
|
||||
|
||||
assert.equal(
|
||||
hook.result.current.starredChannelIds.has("shared"),
|
||||
true,
|
||||
"a later same-second local click must survive a delayed older remote",
|
||||
);
|
||||
assert.ok(
|
||||
[...timers.values()].some((t) => t.ms === 2000),
|
||||
"the local pending publish must remain scheduled",
|
||||
);
|
||||
hook.unmount();
|
||||
} finally {
|
||||
cleanup();
|
||||
relayClient.fetchEvents = origFetch;
|
||||
relayClient.subscribeLive = origLive;
|
||||
relayClient.subscribeToReconnects = origReconnect;
|
||||
window.__TAURI_INTERNALS__ = origTauri;
|
||||
window.setTimeout = origSetTimeout;
|
||||
window.clearTimeout = origClearTimeout;
|
||||
Date.now = origDateNow;
|
||||
}
|
||||
});
|
||||
|
||||
@@ -5,6 +5,7 @@ import {
|
||||
boundStarStore,
|
||||
DEFAULT_STORE,
|
||||
mergeApplyingRemote,
|
||||
mergeStores,
|
||||
readChannelStarsStore,
|
||||
starredChannelIdsFromStore,
|
||||
storageKey,
|
||||
@@ -82,10 +83,27 @@ export function useChannelStars(
|
||||
remote.eventId >= lastAppliedEventId.current
|
||||
)
|
||||
return prev;
|
||||
// A canonical supersession corrects an already-applied same-timestamp
|
||||
// LARGER-id head with the true winner: only here may the incoming blob's
|
||||
// per-entry values win an equal-`updatedAt` tie, and only here does the
|
||||
// pending publish (which reflected the superseded head) get cancelled.
|
||||
// Any other application (bootstrap / live / newer timestamp) merges over
|
||||
// optimistic local state with local-wins `mergeStores` and must NOT
|
||||
// cancel a pending local publish — otherwise a later same-second local
|
||||
// click (integer-second `updatedAt`) loses to an older remote entry that
|
||||
// decrypts late, and its publish is silently dropped.
|
||||
const isCanonicalSupersession =
|
||||
remote.createdAt === lastAppliedRemoteTs.current &&
|
||||
lastAppliedEventId.current !== "" &&
|
||||
remote.eventId < lastAppliedEventId.current;
|
||||
lastAppliedRemoteTs.current = remote.createdAt;
|
||||
lastAppliedEventId.current = remote.eventId;
|
||||
managerRef.current?.cancelPendingStarPublish();
|
||||
const merged = mergeApplyingRemote(prev, remote.store);
|
||||
const merged = isCanonicalSupersession
|
||||
? mergeApplyingRemote(prev, remote.store)
|
||||
: mergeStores(prev, remote.store);
|
||||
if (isCanonicalSupersession) {
|
||||
managerRef.current?.cancelPendingStarPublish();
|
||||
}
|
||||
if (!writeChannelStarsStore(pubkey, merged)) return prev;
|
||||
return merged;
|
||||
};
|
||||
|
||||
Reference in New Issue
Block a user