mirror of
https://github.com/block/buzz.git
synced 2026-10-30 05:51:46 +00:00
fix(sync): make the sidebar durable outbox multi-window safe
Sections, stars, mutes, and sort each persist an unpublished edit under one
localStorage outbox key per identity+relay, but generation ownership is only
in-memory per window. Without cross-window ownership, one window's completing
publish could clear a peer window's still-unpublished edit, and (on the merge
lanes) a peer's write could overwrite an edit before it published.
Every outbox write now mints an ownership token and stores a {store, token}
envelope; a completing publish compare-and-clears only when the stored token
still matches its own, so a peer's newer write survives an older window's ACK.
Stars and mutes additionally read-merge-write both the durable outbox and the
main store via their per-entry mergeStores, so two windows editing different
channels both survive; sort and sections replace whole-blob with LWW resolution
matching the relay. Sort also gains the durable outbox + bounded retry it
previously lacked, and stars/mutes/sort now clamp publish created_at inside the
relay's future-drift window like sections. The envelope reader tolerates a
legacy token-less entry so an outbox written by a prior build still resumes.
Co-authored-by: Will Pfleger <pfleger.will@gmail.com>
Signed-off-by: Will Pfleger <pfleger.will@gmail.com>
This commit is contained in:
@@ -1,4 +1,9 @@
|
||||
import { normalizeRelayUrl } from "@/shared/lib/normalizeRelayUrl";
|
||||
import {
|
||||
clearOutboxEntry,
|
||||
readOutboxEntry,
|
||||
writeOutboxEntry,
|
||||
} from "./sidebarSyncWatermark";
|
||||
|
||||
const STORAGE_KEY_PREFIX = "buzz-channel-mutes.v1";
|
||||
export const MAX_CHANNEL_MUTE_ENTRIES = 500;
|
||||
@@ -119,18 +124,26 @@ export function boundMuteStore(
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Read-merge-write the main store: fold `incoming` into whatever is currently
|
||||
* persisted (which a peer window may have advanced since this window last read)
|
||||
* and persist + return the merged result, so a concurrent click in another
|
||||
* window is carried forward instead of clobbered. Returns the merged store, or
|
||||
* `null` on write failure. Uses the same per-entry `mergeStores` as every other
|
||||
* observation path, so the write is order-independent and idempotent.
|
||||
*/
|
||||
export function writeChannelMutesStore(
|
||||
pubkey: string,
|
||||
store: ChannelMuteStore,
|
||||
): boolean {
|
||||
incoming: ChannelMuteStore,
|
||||
preservedKey?: string,
|
||||
): ChannelMuteStore | null {
|
||||
try {
|
||||
window.localStorage.setItem(
|
||||
storageKey(pubkey),
|
||||
JSON.stringify(boundMuteStore(store)),
|
||||
);
|
||||
return true;
|
||||
const persisted = readChannelMutesStore(pubkey);
|
||||
const merged = mergeStores(persisted, incoming, preservedKey);
|
||||
window.localStorage.setItem(storageKey(pubkey), JSON.stringify(merged));
|
||||
return merged;
|
||||
} catch {
|
||||
return false;
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -152,6 +165,7 @@ export function writeChannelMutesStore(
|
||||
export function mergeStores(
|
||||
a: ChannelMuteStore,
|
||||
b: ChannelMuteStore,
|
||||
preservedKey?: string,
|
||||
): ChannelMuteStore {
|
||||
const allIds = new Set([
|
||||
...Object.keys(a.channels),
|
||||
@@ -163,7 +177,7 @@ export function mergeStores(
|
||||
const r = b.channels[id];
|
||||
merged[id] = l && r ? pickMuteEntry(l, r) : ((l ?? r) as ChannelMuteEntry);
|
||||
}
|
||||
return boundMuteStore({ version: 1, channels: merged });
|
||||
return boundMuteStore({ version: 1, channels: merged }, preservedKey);
|
||||
}
|
||||
|
||||
/** The winner of two entries under `updatedAt` → `rev` → `muted` order. */
|
||||
@@ -200,21 +214,25 @@ function outboxKey(pubkey: string, relayUrl: string): string {
|
||||
* 2s publish debounce. Written synchronously on every click; cleared once the
|
||||
* edit is published or found identical to the last published store. Resumed on
|
||||
* next mount so a durable intent is never silently dropped at teardown.
|
||||
*
|
||||
* Read-merge-write: the incoming edit is folded (via `mergeStores`) into
|
||||
* whatever a peer window may have already persisted under the shared key, so
|
||||
* two windows clicking different channels before their `storage` events deliver
|
||||
* both survive in the durable outbox. `token` gives this write cross-window
|
||||
* ownership so a peer's older completing publish cannot clear the merged entry.
|
||||
*/
|
||||
export function writeChannelMutesOutbox(
|
||||
pubkey: string,
|
||||
store: ChannelMuteStore,
|
||||
relayUrl: string,
|
||||
token: string,
|
||||
): void {
|
||||
try {
|
||||
window.localStorage.setItem(
|
||||
outboxKey(pubkey, relayUrl),
|
||||
JSON.stringify(boundMuteStore(store)),
|
||||
);
|
||||
} catch {
|
||||
// Best-effort durability; the in-memory pendingStore still drives this
|
||||
// session's publish even if the persisted copy could not be written.
|
||||
}
|
||||
const key = outboxKey(pubkey, relayUrl);
|
||||
const existing = readOutboxEntry(key, parseMutePayload)?.store;
|
||||
const merged = existing
|
||||
? mergeStores(existing, store)
|
||||
: boundMuteStore(store);
|
||||
writeOutboxEntry(key, merged, token);
|
||||
}
|
||||
|
||||
/** Read a persisted unpublished edit, or null when none/unparseable. */
|
||||
@@ -222,24 +240,22 @@ export function readChannelMutesOutbox(
|
||||
pubkey: string,
|
||||
relayUrl: string,
|
||||
): ChannelMuteStore | null {
|
||||
try {
|
||||
const raw = window.localStorage.getItem(outboxKey(pubkey, relayUrl));
|
||||
if (!raw) return null;
|
||||
return parseMutePayload(JSON.parse(raw));
|
||||
} catch {
|
||||
return null;
|
||||
}
|
||||
return (
|
||||
readOutboxEntry(outboxKey(pubkey, relayUrl), parseMutePayload)?.store ??
|
||||
null
|
||||
);
|
||||
}
|
||||
|
||||
/** Clear the persisted outbox (edit published or a no-op). */
|
||||
/**
|
||||
* Clear the persisted outbox (edit published or a no-op). Compare-and-clear on
|
||||
* `token`: a peer window that overwrote the entry replaced the token, so an
|
||||
* older window's completion no-ops and the peer's edit survives. Omit `token`
|
||||
* to clear unconditionally.
|
||||
*/
|
||||
export function clearChannelMutesOutbox(
|
||||
pubkey: string,
|
||||
relayUrl: string,
|
||||
token?: 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.
|
||||
}
|
||||
clearOutboxEntry(outboxKey(pubkey, relayUrl), token);
|
||||
}
|
||||
|
||||
@@ -16,6 +16,7 @@ import {
|
||||
import {
|
||||
advanceWatermark,
|
||||
clampPublishCreatedAt,
|
||||
mintOutboxToken,
|
||||
readWatermark,
|
||||
runBootstrap,
|
||||
type FetchResult,
|
||||
@@ -62,6 +63,13 @@ export class ChannelMuteSyncManager {
|
||||
// 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;
|
||||
// The ownership token of the durable outbox entry this window most recently
|
||||
// wrote. A completion (success or no-op) clears the shared outbox only when
|
||||
// its stored token still matches this one, so a peer window that read-merge-
|
||||
// wrote the entry (replacing the token) keeps its still-unpublished clicks.
|
||||
// Reset to null once cleared so a stale token can never authorize a later
|
||||
// clear.
|
||||
private pendingOutboxToken: string | null = null;
|
||||
// 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. Serialization guarantees there is never more than one
|
||||
@@ -161,8 +169,16 @@ export class ChannelMuteSyncManager {
|
||||
this.pendingStore = store;
|
||||
++this.pendingGeneration;
|
||||
// Persist synchronously so a click made <2s before quit/community-switch
|
||||
// survives teardown and resumes on next mount (durable outbox).
|
||||
writeChannelMutesOutbox(this.pubkey, store, this.relayUrl);
|
||||
// survives teardown and resumes on next mount (durable outbox). Mint a fresh
|
||||
// ownership token: the write read-merges into a peer window's persisted
|
||||
// clicks, and the token lets a later completion clear only what it still owns.
|
||||
this.pendingOutboxToken = mintOutboxToken();
|
||||
writeChannelMutesOutbox(
|
||||
this.pubkey,
|
||||
store,
|
||||
this.relayUrl,
|
||||
this.pendingOutboxToken,
|
||||
);
|
||||
if (this.debounceTimer !== null) {
|
||||
window.clearTimeout(this.debounceTimer);
|
||||
}
|
||||
@@ -257,7 +273,22 @@ export class ChannelMuteSyncManager {
|
||||
private discardPending(gen: number): void {
|
||||
if (gen !== this.pendingGeneration) return;
|
||||
this.pendingStore = null;
|
||||
clearChannelMutesOutbox(this.pubkey, this.relayUrl);
|
||||
this.clearOwnedOutbox();
|
||||
}
|
||||
|
||||
/**
|
||||
* Compare-and-clear the shared outbox against the token this window wrote, so
|
||||
* a peer window that read-merge-wrote the entry (replacing the token) keeps
|
||||
* its still-unpublished clicks. Reset the token so a stale value cannot
|
||||
* authorize a later clear.
|
||||
*/
|
||||
private clearOwnedOutbox(): void {
|
||||
clearChannelMutesOutbox(
|
||||
this.pubkey,
|
||||
this.relayUrl,
|
||||
this.pendingOutboxToken ?? undefined,
|
||||
);
|
||||
this.pendingOutboxToken = null;
|
||||
}
|
||||
|
||||
/** Schedule a bounded-backoff retry of the retained pending edit. */
|
||||
|
||||
@@ -1,4 +1,9 @@
|
||||
import { normalizeRelayUrl } from "@/shared/lib/normalizeRelayUrl";
|
||||
import {
|
||||
clearOutboxEntry,
|
||||
readOutboxEntry,
|
||||
writeOutboxEntry,
|
||||
} from "./sidebarSyncWatermark";
|
||||
|
||||
const STORAGE_KEY_PREFIX = "buzz-channel-sections.v1";
|
||||
export const MAX_CHANNEL_SECTIONS = 100;
|
||||
@@ -207,21 +212,22 @@ function outboxKey(pubkey: string, relayUrl: string): string {
|
||||
* 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.
|
||||
*
|
||||
* Whole-blob LWW: the write REPLACES the shared entry, and `token` gives the
|
||||
* writing window cross-window ownership so a peer's older completing publish
|
||||
* (which carries a different token) cannot clear this newer edit.
|
||||
*/
|
||||
export function writeChannelSectionsOutbox(
|
||||
pubkey: string,
|
||||
store: ChannelSectionStore,
|
||||
relayUrl: string,
|
||||
token: string,
|
||||
): void {
|
||||
try {
|
||||
window.localStorage.setItem(
|
||||
outboxKey(pubkey, relayUrl),
|
||||
JSON.stringify(boundChannelSectionsStore(store)),
|
||||
);
|
||||
} catch {
|
||||
// Best-effort durability; the in-memory pendingStore still drives this
|
||||
// session's publish even if the persisted copy could not be written.
|
||||
}
|
||||
writeOutboxEntry(
|
||||
outboxKey(pubkey, relayUrl),
|
||||
boundChannelSectionsStore(store),
|
||||
token,
|
||||
);
|
||||
}
|
||||
|
||||
/** Read a persisted unpublished edit, or null when none/unparseable. */
|
||||
@@ -229,22 +235,22 @@ export function readChannelSectionsOutbox(
|
||||
pubkey: string,
|
||||
relayUrl: string,
|
||||
): ChannelSectionStore | null {
|
||||
try {
|
||||
return parseRaw(window.localStorage.getItem(outboxKey(pubkey, relayUrl)));
|
||||
} catch {
|
||||
return null;
|
||||
}
|
||||
return (
|
||||
readOutboxEntry(outboxKey(pubkey, relayUrl), parseChannelSectionPayload)
|
||||
?.store ?? null
|
||||
);
|
||||
}
|
||||
|
||||
/** Clear the persisted outbox (edit published, superseded, or a no-op). */
|
||||
/**
|
||||
* Clear the persisted outbox (edit published, superseded, or a no-op).
|
||||
* Compare-and-clear on `token`: a peer window that overwrote the entry replaced
|
||||
* the token, so an older window's completion no-ops and the peer's edit
|
||||
* survives. Omit `token` to clear unconditionally.
|
||||
*/
|
||||
export function clearChannelSectionsOutbox(
|
||||
pubkey: string,
|
||||
relayUrl: string,
|
||||
token?: 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.
|
||||
}
|
||||
clearOutboxEntry(outboxKey(pubkey, relayUrl), token);
|
||||
}
|
||||
|
||||
@@ -450,7 +450,7 @@ test("overlapping publishes: older completion does not erase a newer queued edit
|
||||
"B is now the pending edit",
|
||||
);
|
||||
assert.equal(
|
||||
JSON.parse(storage.get(outboxKey)).sections[0].id,
|
||||
JSON.parse(storage.get(outboxKey)).store.sections[0].id,
|
||||
"b",
|
||||
"outbox holds B",
|
||||
);
|
||||
|
||||
@@ -16,6 +16,7 @@ import {
|
||||
import {
|
||||
advanceWatermark,
|
||||
clampPublishCreatedAt,
|
||||
mintOutboxToken,
|
||||
readWatermark,
|
||||
runBootstrap,
|
||||
type FetchResult,
|
||||
@@ -142,6 +143,12 @@ export class ChannelSectionSyncManager {
|
||||
// touching shared manager state, so a stale generation can never sign or
|
||||
// publish after a newer edit exists.
|
||||
private publishInFlight = false;
|
||||
// The ownership token of the durable outbox entry this window most recently
|
||||
// wrote. A completion (publish/adopt/no-op) clears the shared outbox only when
|
||||
// its stored token still matches this one, so a peer window that overwrote the
|
||||
// entry (replacing the token) keeps its still-unpublished edit. Reset to null
|
||||
// once cleared so a stale token can never authorize a later clear.
|
||||
private pendingOutboxToken: string | null = null;
|
||||
// Event ids we signed and sent to the relay but whose ACK never arrived (the
|
||||
// publish promise rejected as a timeout/socket error after the frame left).
|
||||
// The relay MAY have accepted such a write, so if a later cycle's pre-publish
|
||||
@@ -256,7 +263,7 @@ export class ChannelSectionSyncManager {
|
||||
this.recordRemoteHead(remote.createdAt, remote.eventId);
|
||||
if (gen !== this.pendingGeneration) return;
|
||||
this.pendingStore = null;
|
||||
clearChannelSectionsOutbox(this.pubkey, this.relayUrl);
|
||||
this.clearOwnedOutbox();
|
||||
this.lastPublishedStore = remote.store;
|
||||
if (this.destroyed) return;
|
||||
this.onRemoteAdopted?.(remote);
|
||||
@@ -271,7 +278,22 @@ export class ChannelSectionSyncManager {
|
||||
private discardPending(gen: number): void {
|
||||
if (gen !== this.pendingGeneration) return;
|
||||
this.pendingStore = null;
|
||||
clearChannelSectionsOutbox(this.pubkey, this.relayUrl);
|
||||
this.clearOwnedOutbox();
|
||||
}
|
||||
|
||||
/**
|
||||
* Compare-and-clear the shared outbox against the token this window wrote, so
|
||||
* a peer window that overwrote the entry (replacing the token) keeps its
|
||||
* still-unpublished edit. Reset the token so a stale value cannot authorize a
|
||||
* later clear.
|
||||
*/
|
||||
private clearOwnedOutbox(): void {
|
||||
clearChannelSectionsOutbox(
|
||||
this.pubkey,
|
||||
this.relayUrl,
|
||||
this.pendingOutboxToken ?? undefined,
|
||||
);
|
||||
this.pendingOutboxToken = null;
|
||||
}
|
||||
|
||||
publishSections(store: ChannelSectionStore): void {
|
||||
@@ -288,8 +310,16 @@ export class ChannelSectionSyncManager {
|
||||
// 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).
|
||||
writeChannelSectionsOutbox(this.pubkey, store, this.relayUrl);
|
||||
// survives teardown and resumes on next mount (fix-3 durable outbox). Mint a
|
||||
// fresh ownership token: whole-blob LWW means this write REPLACES the shared
|
||||
// entry, and the token lets a later completion clear only what it still owns.
|
||||
this.pendingOutboxToken = mintOutboxToken();
|
||||
writeChannelSectionsOutbox(
|
||||
this.pubkey,
|
||||
store,
|
||||
this.relayUrl,
|
||||
this.pendingOutboxToken,
|
||||
);
|
||||
if (this.debounceTimer !== null) {
|
||||
window.clearTimeout(this.debounceTimer);
|
||||
}
|
||||
|
||||
@@ -1,5 +1,10 @@
|
||||
import { normalizeRelayUrl } from "@/shared/lib/normalizeRelayUrl";
|
||||
import type { Channel } from "@/shared/api/types";
|
||||
import {
|
||||
clearOutboxEntry,
|
||||
readOutboxEntry,
|
||||
writeOutboxEntry,
|
||||
} from "./sidebarSyncWatermark";
|
||||
|
||||
const STORAGE_KEY_PREFIX = "buzz-channel-sort.v1";
|
||||
export const MAX_CHANNEL_SORT_GROUPS = 104;
|
||||
@@ -145,21 +150,22 @@ function outboxKey(pubkey: string, relayUrl: string): string {
|
||||
* 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.
|
||||
*
|
||||
* Whole-blob LWW: the write REPLACES the shared entry, and `token` gives the
|
||||
* writing window cross-window ownership so a peer's older completing publish
|
||||
* (which carries a different token) cannot clear this newer edit.
|
||||
*/
|
||||
export function writeChannelSortOutbox(
|
||||
pubkey: string,
|
||||
store: ChannelSortStore,
|
||||
relayUrl: string,
|
||||
token: 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.
|
||||
}
|
||||
writeOutboxEntry(
|
||||
outboxKey(pubkey, relayUrl),
|
||||
boundChannelSortStore(store),
|
||||
token,
|
||||
);
|
||||
}
|
||||
|
||||
/** Read a persisted unpublished sort edit, or null when none/unparseable. */
|
||||
@@ -167,23 +173,24 @@ 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;
|
||||
}
|
||||
return (
|
||||
readOutboxEntry(outboxKey(pubkey, relayUrl), parseChannelSortPayload)
|
||||
?.store ?? 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.
|
||||
}
|
||||
/**
|
||||
* Clear the persisted sort outbox (edit published, superseded, or a no-op).
|
||||
* Compare-and-clear on `token`: a peer window that overwrote the entry replaced
|
||||
* the token, so an older window's completion no-ops and the peer's edit
|
||||
* survives. Omit `token` to clear unconditionally.
|
||||
*/
|
||||
export function clearChannelSortOutbox(
|
||||
pubkey: string,
|
||||
relayUrl: string,
|
||||
token?: string,
|
||||
): void {
|
||||
clearOutboxEntry(outboxKey(pubkey, relayUrl), token);
|
||||
}
|
||||
|
||||
export function sortModeForGroup(
|
||||
|
||||
@@ -314,9 +314,10 @@ test("durable outbox: edit destroyed inside the debounce resumes and publishes o
|
||||
m1.destroy();
|
||||
assert.equal(publishCalls.length, 0, "destroy must not flush");
|
||||
|
||||
// Window 2: resume the persisted outbox (the hook does this after bootstrap).
|
||||
// Window 2: resume the persisted outbox (the hook does this after bootstrap
|
||||
// via readChannelSortOutbox, which unwraps the `{store, token}` envelope).
|
||||
const m2 = new ChannelSortSyncManager("pk-resume", RELAY);
|
||||
m2.publishSortPrefs(JSON.parse(persisted));
|
||||
m2.publishSortPrefs(JSON.parse(persisted).store);
|
||||
fw._fireTimer();
|
||||
await new Promise((r) => setTimeout(r, 20));
|
||||
assert.equal(publishCalls.length, 1, "resumed edit must publish");
|
||||
@@ -429,7 +430,9 @@ test("overlapping publishes: older completion does not erase a newer queued edit
|
||||
"B is now the pending edit",
|
||||
);
|
||||
assert.deepEqual(
|
||||
Object.keys(JSON.parse(storage.get(OUTBOX_KEY("pk-overlap"))).groups),
|
||||
Object.keys(
|
||||
JSON.parse(storage.get(OUTBOX_KEY("pk-overlap"))).store.groups,
|
||||
),
|
||||
["dms"],
|
||||
"outbox holds B",
|
||||
);
|
||||
|
||||
@@ -15,6 +15,7 @@ import {
|
||||
import {
|
||||
advanceWatermark,
|
||||
clampPublishCreatedAt,
|
||||
mintOutboxToken,
|
||||
readWatermark,
|
||||
runBootstrap,
|
||||
type FetchResult,
|
||||
@@ -141,6 +142,12 @@ export class ChannelSortSyncManager {
|
||||
// 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;
|
||||
// The ownership token of the durable outbox entry this window most recently
|
||||
// wrote. A completion (publish/adopt/no-op) clears the shared outbox only when
|
||||
// its stored token still matches this one, so a peer window that overwrote the
|
||||
// entry (replacing the token) keeps its still-unpublished edit. Reset to null
|
||||
// once cleared so a stale token can never authorize a later clear.
|
||||
private pendingOutboxToken: string | null = null;
|
||||
// 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.
|
||||
@@ -252,7 +259,7 @@ export class ChannelSortSyncManager {
|
||||
this.recordRemoteHead(remote.createdAt, remote.eventId);
|
||||
if (gen !== this.pendingGeneration) return;
|
||||
this.pendingStore = null;
|
||||
clearChannelSortOutbox(this.pubkey, this.relayUrl);
|
||||
this.clearOwnedOutbox();
|
||||
this.lastPublishedStore = remote.store;
|
||||
if (this.destroyed) return;
|
||||
this.onRemoteAdopted?.(remote);
|
||||
@@ -267,7 +274,22 @@ export class ChannelSortSyncManager {
|
||||
private discardPending(gen: number): void {
|
||||
if (gen !== this.pendingGeneration) return;
|
||||
this.pendingStore = null;
|
||||
clearChannelSortOutbox(this.pubkey, this.relayUrl);
|
||||
this.clearOwnedOutbox();
|
||||
}
|
||||
|
||||
/**
|
||||
* Compare-and-clear the shared outbox against the token this window wrote, so
|
||||
* a peer window that overwrote the entry (replacing the token) keeps its
|
||||
* still-unpublished edit. Reset the token so a stale value cannot authorize a
|
||||
* later clear.
|
||||
*/
|
||||
private clearOwnedOutbox(): void {
|
||||
clearChannelSortOutbox(
|
||||
this.pubkey,
|
||||
this.relayUrl,
|
||||
this.pendingOutboxToken ?? undefined,
|
||||
);
|
||||
this.pendingOutboxToken = null;
|
||||
}
|
||||
|
||||
publishSortPrefs(store: ChannelSortStore): void {
|
||||
@@ -282,8 +304,16 @@ export class ChannelSortSyncManager {
|
||||
// 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);
|
||||
// survives teardown and resumes on next mount (durable outbox). Mint a fresh
|
||||
// ownership token: whole-blob LWW means this write REPLACES the shared entry,
|
||||
// and the token lets a later completion clear only what it still owns.
|
||||
this.pendingOutboxToken = mintOutboxToken();
|
||||
writeChannelSortOutbox(
|
||||
this.pubkey,
|
||||
store,
|
||||
this.relayUrl,
|
||||
this.pendingOutboxToken,
|
||||
);
|
||||
if (this.debounceTimer !== null) {
|
||||
window.clearTimeout(this.debounceTimer);
|
||||
}
|
||||
|
||||
@@ -1,4 +1,9 @@
|
||||
import { normalizeRelayUrl } from "@/shared/lib/normalizeRelayUrl";
|
||||
import {
|
||||
clearOutboxEntry,
|
||||
readOutboxEntry,
|
||||
writeOutboxEntry,
|
||||
} from "./sidebarSyncWatermark";
|
||||
|
||||
const STORAGE_KEY_PREFIX = "buzz-channel-stars.v1";
|
||||
export const MAX_CHANNEL_STAR_ENTRIES = 500;
|
||||
@@ -119,18 +124,26 @@ export function boundStarStore(
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Read-merge-write the main store: fold `incoming` into whatever is currently
|
||||
* persisted (which a peer window may have advanced since this window last read)
|
||||
* and persist + return the merged result, so a concurrent click in another
|
||||
* window is carried forward instead of clobbered. Returns the merged store, or
|
||||
* `null` on write failure. Uses the same per-entry `mergeStores` as every other
|
||||
* observation path, so the write is order-independent and idempotent.
|
||||
*/
|
||||
export function writeChannelStarsStore(
|
||||
pubkey: string,
|
||||
store: ChannelStarStore,
|
||||
): boolean {
|
||||
incoming: ChannelStarStore,
|
||||
preservedKey?: string,
|
||||
): ChannelStarStore | null {
|
||||
try {
|
||||
window.localStorage.setItem(
|
||||
storageKey(pubkey),
|
||||
JSON.stringify(boundStarStore(store)),
|
||||
);
|
||||
return true;
|
||||
const persisted = readChannelStarsStore(pubkey);
|
||||
const merged = mergeStores(persisted, incoming, preservedKey);
|
||||
window.localStorage.setItem(storageKey(pubkey), JSON.stringify(merged));
|
||||
return merged;
|
||||
} catch {
|
||||
return false;
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -152,6 +165,7 @@ export function writeChannelStarsStore(
|
||||
export function mergeStores(
|
||||
a: ChannelStarStore,
|
||||
b: ChannelStarStore,
|
||||
preservedKey?: string,
|
||||
): ChannelStarStore {
|
||||
const allIds = new Set([
|
||||
...Object.keys(a.channels),
|
||||
@@ -163,7 +177,7 @@ export function mergeStores(
|
||||
const r = b.channels[id];
|
||||
merged[id] = l && r ? pickStarEntry(l, r) : ((l ?? r) as ChannelStarEntry);
|
||||
}
|
||||
return boundStarStore({ version: 1, channels: merged });
|
||||
return boundStarStore({ version: 1, channels: merged }, preservedKey);
|
||||
}
|
||||
|
||||
/** The winner of two entries under `updatedAt` → `rev` → `starred` order. */
|
||||
@@ -202,21 +216,25 @@ function outboxKey(pubkey: string, relayUrl: string): string {
|
||||
* 2s publish debounce. Written synchronously on every click; cleared once the
|
||||
* edit is published or found identical to the last published store. Resumed on
|
||||
* next mount so a durable intent is never silently dropped at teardown.
|
||||
*
|
||||
* Read-merge-write: the incoming edit is folded (via `mergeStores`) into
|
||||
* whatever a peer window may have already persisted under the shared key, so
|
||||
* two windows clicking different channels before their `storage` events deliver
|
||||
* both survive in the durable outbox. `token` gives this write cross-window
|
||||
* ownership so a peer's older completing publish cannot clear the merged entry.
|
||||
*/
|
||||
export function writeChannelStarsOutbox(
|
||||
pubkey: string,
|
||||
store: ChannelStarStore,
|
||||
relayUrl: string,
|
||||
token: string,
|
||||
): void {
|
||||
try {
|
||||
window.localStorage.setItem(
|
||||
outboxKey(pubkey, relayUrl),
|
||||
JSON.stringify(boundStarStore(store)),
|
||||
);
|
||||
} catch {
|
||||
// Best-effort durability; the in-memory pendingStore still drives this
|
||||
// session's publish even if the persisted copy could not be written.
|
||||
}
|
||||
const key = outboxKey(pubkey, relayUrl);
|
||||
const existing = readOutboxEntry(key, parseStarPayload)?.store;
|
||||
const merged = existing
|
||||
? mergeStores(existing, store)
|
||||
: boundStarStore(store);
|
||||
writeOutboxEntry(key, merged, token);
|
||||
}
|
||||
|
||||
/** Read a persisted unpublished edit, or null when none/unparseable. */
|
||||
@@ -224,24 +242,22 @@ export function readChannelStarsOutbox(
|
||||
pubkey: string,
|
||||
relayUrl: string,
|
||||
): ChannelStarStore | null {
|
||||
try {
|
||||
const raw = window.localStorage.getItem(outboxKey(pubkey, relayUrl));
|
||||
if (!raw) return null;
|
||||
return parseStarPayload(JSON.parse(raw));
|
||||
} catch {
|
||||
return null;
|
||||
}
|
||||
return (
|
||||
readOutboxEntry(outboxKey(pubkey, relayUrl), parseStarPayload)?.store ??
|
||||
null
|
||||
);
|
||||
}
|
||||
|
||||
/** Clear the persisted outbox (edit published or a no-op). */
|
||||
/**
|
||||
* Clear the persisted outbox (edit published or a no-op). Compare-and-clear on
|
||||
* `token`: a peer window that overwrote the entry replaced the token, so an
|
||||
* older window's completion no-ops and the peer's edit survives. Omit `token`
|
||||
* to clear unconditionally.
|
||||
*/
|
||||
export function clearChannelStarsOutbox(
|
||||
pubkey: string,
|
||||
relayUrl: string,
|
||||
token?: 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.
|
||||
}
|
||||
clearOutboxEntry(outboxKey(pubkey, relayUrl), token);
|
||||
}
|
||||
|
||||
@@ -16,6 +16,7 @@ import {
|
||||
import {
|
||||
advanceWatermark,
|
||||
clampPublishCreatedAt,
|
||||
mintOutboxToken,
|
||||
readWatermark,
|
||||
runBootstrap,
|
||||
type FetchResult,
|
||||
@@ -62,6 +63,13 @@ export class ChannelStarSyncManager {
|
||||
// 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;
|
||||
// The ownership token of the durable outbox entry this window most recently
|
||||
// wrote. A completion (success or no-op) clears the shared outbox only when
|
||||
// its stored token still matches this one, so a peer window that read-merge-
|
||||
// wrote the entry (replacing the token) keeps its still-unpublished clicks.
|
||||
// Reset to null once cleared so a stale token can never authorize a later
|
||||
// clear.
|
||||
private pendingOutboxToken: string | null = null;
|
||||
// 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. Serialization guarantees there is never more than one
|
||||
@@ -161,8 +169,16 @@ export class ChannelStarSyncManager {
|
||||
this.pendingStore = store;
|
||||
++this.pendingGeneration;
|
||||
// Persist synchronously so a click made <2s before quit/community-switch
|
||||
// survives teardown and resumes on next mount (durable outbox).
|
||||
writeChannelStarsOutbox(this.pubkey, store, this.relayUrl);
|
||||
// survives teardown and resumes on next mount (durable outbox). Mint a fresh
|
||||
// ownership token: the write read-merges into a peer window's persisted
|
||||
// clicks, and the token lets a later completion clear only what it still owns.
|
||||
this.pendingOutboxToken = mintOutboxToken();
|
||||
writeChannelStarsOutbox(
|
||||
this.pubkey,
|
||||
store,
|
||||
this.relayUrl,
|
||||
this.pendingOutboxToken,
|
||||
);
|
||||
if (this.debounceTimer !== null) {
|
||||
window.clearTimeout(this.debounceTimer);
|
||||
}
|
||||
@@ -257,7 +273,22 @@ export class ChannelStarSyncManager {
|
||||
private discardPending(gen: number): void {
|
||||
if (gen !== this.pendingGeneration) return;
|
||||
this.pendingStore = null;
|
||||
clearChannelStarsOutbox(this.pubkey, this.relayUrl);
|
||||
this.clearOwnedOutbox();
|
||||
}
|
||||
|
||||
/**
|
||||
* Compare-and-clear the shared outbox against the token this window wrote, so
|
||||
* a peer window that read-merge-wrote the entry (replacing the token) keeps
|
||||
* its still-unpublished clicks. Reset the token so a stale value cannot
|
||||
* authorize a later clear.
|
||||
*/
|
||||
private clearOwnedOutbox(): void {
|
||||
clearChannelStarsOutbox(
|
||||
this.pubkey,
|
||||
this.relayUrl,
|
||||
this.pendingOutboxToken ?? undefined,
|
||||
);
|
||||
this.pendingOutboxToken = null;
|
||||
}
|
||||
|
||||
/** Schedule a bounded-backoff retry of the retained pending edit. */
|
||||
|
||||
404
desktop/src/features/sidebar/lib/multiWindowOutbox.test.mjs
Normal file
404
desktop/src/features/sidebar/lib/multiWindowOutbox.test.mjs
Normal file
@@ -0,0 +1,404 @@
|
||||
import assert from "node:assert/strict";
|
||||
import test from "node:test";
|
||||
|
||||
// Multi-window durable-outbox safety (Carl's CHANGES_REQUESTED, finding P1).
|
||||
//
|
||||
// All four sidebar-sync lanes share one localStorage outbox key per
|
||||
// (identity, relay), but generation ownership is in-memory per window. Without
|
||||
// a per-write ownership token, window A's completing publish can clear window
|
||||
// B's still-unpublished edit (stale-clear, prong a), and — for the merge lanes
|
||||
// — B's write can overwrite A's still-unpublished payload (overwrite, prong b).
|
||||
//
|
||||
// The fix: every outbox write mints a token and stores a `{store, token}`
|
||||
// envelope; a completing publish compare-and-clears only when the stored token
|
||||
// still matches its own. Merge lanes (stars/mutes) additionally read-merge-write
|
||||
// so two windows editing different channels both survive. Replace lanes
|
||||
// (sort/sections) resolve whole-blob LWW: the last writer's token owns the entry.
|
||||
//
|
||||
// This matrix drives the shared helpers directly through localStorage — no
|
||||
// relay, no timers — so each interleaving is deterministic. Manager-level and
|
||||
// hook-level behavior is covered by their own suites; this file isolates the
|
||||
// cross-window contract the token exists to hold.
|
||||
|
||||
function withFreshStorage(fn) {
|
||||
const store = new Map();
|
||||
const ls = {
|
||||
getItem: (k) => store.get(k) ?? null,
|
||||
setItem: (k, v) => store.set(k, v),
|
||||
removeItem: (k) => store.delete(k),
|
||||
clear: () => store.clear(),
|
||||
};
|
||||
const orig = globalThis.window?.localStorage;
|
||||
if (typeof globalThis.window === "undefined") globalThis.window = {};
|
||||
globalThis.window.localStorage = ls;
|
||||
try {
|
||||
fn(ls);
|
||||
} finally {
|
||||
if (orig !== undefined) globalThis.window.localStorage = orig;
|
||||
else delete globalThis.window.localStorage;
|
||||
}
|
||||
}
|
||||
|
||||
const { mintOutboxToken } = await import("./sidebarSyncWatermark.ts");
|
||||
const stars = await import("./channelStarsStorage.ts");
|
||||
const mutes = await import("./channelMutesStorage.ts");
|
||||
const sort = await import("./channelSortPreference.ts");
|
||||
const sections = await import("./channelSectionsStorage.ts");
|
||||
|
||||
const PK = "pk";
|
||||
const RELAY = "wss://relay.example.com";
|
||||
|
||||
const starStore = (channels) => ({ version: 1, channels });
|
||||
const starEntry = (starred, updatedAt, rev) => ({ starred, updatedAt, rev });
|
||||
const muteStore = (channels) => ({ version: 1, channels });
|
||||
const muteEntry = (muted, updatedAt, rev) => ({ muted, updatedAt, rev });
|
||||
const sortStore = (groups) => ({ version: 1, groups });
|
||||
const sectionStore = (secs, assignments = {}) => ({
|
||||
version: 1,
|
||||
sections: secs,
|
||||
assignments,
|
||||
});
|
||||
|
||||
// ── (a) stale-clear: A queues → B queues → A completes and clears → B survives ─
|
||||
//
|
||||
// Applies to every lane: A's completion carries A's token; B's later write
|
||||
// replaced the token, so A's compare-and-clear must no-op and B's edit stays.
|
||||
|
||||
test("(a) stars: A's completing clear does not erase B's later queued edit", () => {
|
||||
withFreshStorage(() => {
|
||||
const tokenA = mintOutboxToken();
|
||||
stars.writeChannelStarsOutbox(
|
||||
PK,
|
||||
starStore({ a: starEntry(true, 100, 1) }),
|
||||
RELAY,
|
||||
tokenA,
|
||||
);
|
||||
const tokenB = mintOutboxToken();
|
||||
stars.writeChannelStarsOutbox(
|
||||
PK,
|
||||
starStore({ b: starEntry(true, 200, 1) }),
|
||||
RELAY,
|
||||
tokenB,
|
||||
);
|
||||
// A completes and tries to clear with its own (now-stale) token.
|
||||
stars.clearChannelStarsOutbox(PK, RELAY, tokenA);
|
||||
const survived = stars.readChannelStarsOutbox(PK, RELAY);
|
||||
assert.ok(survived, "B's edit must survive A's stale clear");
|
||||
assert.deepEqual(survived.channels.b, starEntry(true, 200, 1));
|
||||
// B's own completion clears cleanly.
|
||||
stars.clearChannelStarsOutbox(PK, RELAY, tokenB);
|
||||
assert.equal(stars.readChannelStarsOutbox(PK, RELAY), null);
|
||||
});
|
||||
});
|
||||
|
||||
test("(a) mutes: A's completing clear does not erase B's later queued edit", () => {
|
||||
withFreshStorage(() => {
|
||||
const tokenA = mintOutboxToken();
|
||||
mutes.writeChannelMutesOutbox(
|
||||
PK,
|
||||
muteStore({ a: muteEntry(true, 100, 1) }),
|
||||
RELAY,
|
||||
tokenA,
|
||||
);
|
||||
const tokenB = mintOutboxToken();
|
||||
mutes.writeChannelMutesOutbox(
|
||||
PK,
|
||||
muteStore({ b: muteEntry(true, 200, 1) }),
|
||||
RELAY,
|
||||
tokenB,
|
||||
);
|
||||
mutes.clearChannelMutesOutbox(PK, RELAY, tokenA);
|
||||
const survived = mutes.readChannelMutesOutbox(PK, RELAY);
|
||||
assert.ok(survived, "B's edit must survive A's stale clear");
|
||||
assert.deepEqual(survived.channels.b, muteEntry(true, 200, 1));
|
||||
mutes.clearChannelMutesOutbox(PK, RELAY, tokenB);
|
||||
assert.equal(mutes.readChannelMutesOutbox(PK, RELAY), null);
|
||||
});
|
||||
});
|
||||
|
||||
test("(a) sort: A's completing clear does not erase B's later queued edit", () => {
|
||||
withFreshStorage(() => {
|
||||
const tokenA = mintOutboxToken();
|
||||
sort.writeChannelSortOutbox(
|
||||
PK,
|
||||
sortStore({ channels: "alpha" }),
|
||||
RELAY,
|
||||
tokenA,
|
||||
);
|
||||
const tokenB = mintOutboxToken();
|
||||
sort.writeChannelSortOutbox(
|
||||
PK,
|
||||
sortStore({ channels: "recent" }),
|
||||
RELAY,
|
||||
tokenB,
|
||||
);
|
||||
sort.clearChannelSortOutbox(PK, RELAY, tokenA);
|
||||
const survived = sort.readChannelSortOutbox(PK, RELAY);
|
||||
assert.ok(survived, "B's edit must survive A's stale clear");
|
||||
assert.equal(survived.groups.channels, "recent");
|
||||
sort.clearChannelSortOutbox(PK, RELAY, tokenB);
|
||||
assert.equal(sort.readChannelSortOutbox(PK, RELAY), null);
|
||||
});
|
||||
});
|
||||
|
||||
test("(a) sections: A's completing clear does not erase B's later queued edit", () => {
|
||||
withFreshStorage(() => {
|
||||
const tokenA = mintOutboxToken();
|
||||
sections.writeChannelSectionsOutbox(
|
||||
PK,
|
||||
sectionStore([{ id: "s1", name: "One", order: 0 }]),
|
||||
RELAY,
|
||||
tokenA,
|
||||
);
|
||||
const tokenB = mintOutboxToken();
|
||||
sections.writeChannelSectionsOutbox(
|
||||
PK,
|
||||
sectionStore([{ id: "s2", name: "Two", order: 0 }]),
|
||||
RELAY,
|
||||
tokenB,
|
||||
);
|
||||
sections.clearChannelSectionsOutbox(PK, RELAY, tokenA);
|
||||
const survived = sections.readChannelSectionsOutbox(PK, RELAY);
|
||||
assert.ok(survived, "B's edit must survive A's stale clear");
|
||||
assert.deepEqual(survived.sections, [{ id: "s2", name: "Two", order: 0 }]);
|
||||
sections.clearChannelSectionsOutbox(PK, RELAY, tokenB);
|
||||
assert.equal(sections.readChannelSectionsOutbox(PK, RELAY), null);
|
||||
});
|
||||
});
|
||||
|
||||
// ── (b) overwrite (merge lanes only): two windows click DIFFERENT channels →
|
||||
// both survive in the durable outbox. This is the read-merge-write that
|
||||
// distinguishes stars/mutes from the replace lanes.
|
||||
|
||||
test("(b) stars: two windows clicking different channels both survive in the outbox", () => {
|
||||
withFreshStorage(() => {
|
||||
stars.writeChannelStarsOutbox(
|
||||
PK,
|
||||
starStore({ a: starEntry(true, 100, 1) }),
|
||||
RELAY,
|
||||
mintOutboxToken(),
|
||||
);
|
||||
// Window B does not read A first; its write must fold, not clobber.
|
||||
stars.writeChannelStarsOutbox(
|
||||
PK,
|
||||
starStore({ b: starEntry(true, 200, 1) }),
|
||||
RELAY,
|
||||
mintOutboxToken(),
|
||||
);
|
||||
const merged = stars.readChannelStarsOutbox(PK, RELAY);
|
||||
assert.deepEqual(
|
||||
merged.channels.a,
|
||||
starEntry(true, 100, 1),
|
||||
"A's click retained",
|
||||
);
|
||||
assert.deepEqual(
|
||||
merged.channels.b,
|
||||
starEntry(true, 200, 1),
|
||||
"B's click retained",
|
||||
);
|
||||
});
|
||||
});
|
||||
|
||||
test("(b) mutes: two windows clicking different channels both survive in the outbox", () => {
|
||||
withFreshStorage(() => {
|
||||
mutes.writeChannelMutesOutbox(
|
||||
PK,
|
||||
muteStore({ a: muteEntry(true, 100, 1) }),
|
||||
RELAY,
|
||||
mintOutboxToken(),
|
||||
);
|
||||
mutes.writeChannelMutesOutbox(
|
||||
PK,
|
||||
muteStore({ b: muteEntry(true, 200, 1) }),
|
||||
RELAY,
|
||||
mintOutboxToken(),
|
||||
);
|
||||
const merged = mutes.readChannelMutesOutbox(PK, RELAY);
|
||||
assert.deepEqual(
|
||||
merged.channels.a,
|
||||
muteEntry(true, 100, 1),
|
||||
"A's click retained",
|
||||
);
|
||||
assert.deepEqual(
|
||||
merged.channels.b,
|
||||
muteEntry(true, 200, 1),
|
||||
"B's click retained",
|
||||
);
|
||||
});
|
||||
});
|
||||
|
||||
test("(b) stars: same-channel concurrent writes resolve by merge order (higher rev wins)", () => {
|
||||
withFreshStorage(() => {
|
||||
stars.writeChannelStarsOutbox(
|
||||
PK,
|
||||
starStore({ c: starEntry(true, 100, 5) }),
|
||||
RELAY,
|
||||
mintOutboxToken(),
|
||||
);
|
||||
// A stale same-second peer write for the same channel folds and loses on rev.
|
||||
stars.writeChannelStarsOutbox(
|
||||
PK,
|
||||
starStore({ c: starEntry(false, 100, 2) }),
|
||||
RELAY,
|
||||
mintOutboxToken(),
|
||||
);
|
||||
const merged = stars.readChannelStarsOutbox(PK, RELAY);
|
||||
assert.deepEqual(
|
||||
merged.channels.c,
|
||||
starEntry(true, 100, 5),
|
||||
"higher rev wins the same-second tie",
|
||||
);
|
||||
});
|
||||
});
|
||||
|
||||
// ── (c) teardown-in-debounce → remount resumes a merged outbox. Window A edits
|
||||
// and tears down before its debounce fires (edit persisted, not cleared);
|
||||
// window B edits a different channel meanwhile. On remount the resume
|
||||
// reads one merged outbox carrying both — no edit is lost across the
|
||||
// teardown boundary. (Merge lanes; replace lanes resolve whole-blob LWW.)
|
||||
|
||||
test("(c) stars: an edit persisted before teardown resumes merged with a peer's concurrent edit", () => {
|
||||
withFreshStorage(() => {
|
||||
// Window A persists its debounce-window edit, then tears down (no clear).
|
||||
stars.writeChannelStarsOutbox(
|
||||
PK,
|
||||
starStore({ a: starEntry(true, 100, 1) }),
|
||||
RELAY,
|
||||
mintOutboxToken(),
|
||||
);
|
||||
// Window B persists a different channel before A's remount.
|
||||
stars.writeChannelStarsOutbox(
|
||||
PK,
|
||||
starStore({ b: starEntry(true, 200, 1) }),
|
||||
RELAY,
|
||||
mintOutboxToken(),
|
||||
);
|
||||
// Remount reads the durable outbox (the hook's resume path); both survive.
|
||||
const resumed = stars.readChannelStarsOutbox(PK, RELAY);
|
||||
assert.deepEqual(resumed.channels.a, starEntry(true, 100, 1));
|
||||
assert.deepEqual(resumed.channels.b, starEntry(true, 200, 1));
|
||||
});
|
||||
});
|
||||
|
||||
test("(c) mutes: an edit persisted before teardown resumes merged with a peer's concurrent edit", () => {
|
||||
withFreshStorage(() => {
|
||||
mutes.writeChannelMutesOutbox(
|
||||
PK,
|
||||
muteStore({ a: muteEntry(true, 100, 1) }),
|
||||
RELAY,
|
||||
mintOutboxToken(),
|
||||
);
|
||||
mutes.writeChannelMutesOutbox(
|
||||
PK,
|
||||
muteStore({ b: muteEntry(true, 200, 1) }),
|
||||
RELAY,
|
||||
mintOutboxToken(),
|
||||
);
|
||||
const resumed = mutes.readChannelMutesOutbox(PK, RELAY);
|
||||
assert.deepEqual(resumed.channels.a, muteEntry(true, 100, 1));
|
||||
assert.deepEqual(resumed.channels.b, muteEntry(true, 200, 1));
|
||||
});
|
||||
});
|
||||
|
||||
// ── (d) adopt-path clear no-ops on a mismatched token (sort/sections). The
|
||||
// adopt path clears the outbox after applying a newer remote head; it
|
||||
// must not erase a peer window's fresher local edit.
|
||||
|
||||
test("(d) sort: adopt-path clear with a stale token leaves the peer's fresher edit", () => {
|
||||
withFreshStorage(() => {
|
||||
// Window A adopted a remote head and holds tokenA (its last write).
|
||||
const tokenA = mintOutboxToken();
|
||||
sort.writeChannelSortOutbox(PK, sortStore({ dms: "alpha" }), RELAY, tokenA);
|
||||
// Window B queues a fresher edit, replacing the token.
|
||||
const tokenB = mintOutboxToken();
|
||||
sort.writeChannelSortOutbox(
|
||||
PK,
|
||||
sortStore({ dms: "recent" }),
|
||||
RELAY,
|
||||
tokenB,
|
||||
);
|
||||
// A's adopt-path clear fires with its now-stale token → must no-op.
|
||||
sort.clearChannelSortOutbox(PK, RELAY, tokenA);
|
||||
assert.equal(sort.readChannelSortOutbox(PK, RELAY).groups.dms, "recent");
|
||||
});
|
||||
});
|
||||
|
||||
test("(d) sections: adopt-path clear with a stale token leaves the peer's fresher edit", () => {
|
||||
withFreshStorage(() => {
|
||||
const tokenA = mintOutboxToken();
|
||||
sections.writeChannelSectionsOutbox(
|
||||
PK,
|
||||
sectionStore([{ id: "s1", name: "One", order: 0 }]),
|
||||
RELAY,
|
||||
tokenA,
|
||||
);
|
||||
const tokenB = mintOutboxToken();
|
||||
sections.writeChannelSectionsOutbox(
|
||||
PK,
|
||||
sectionStore([{ id: "s1", name: "Renamed", order: 0 }]),
|
||||
RELAY,
|
||||
tokenB,
|
||||
);
|
||||
sections.clearChannelSectionsOutbox(PK, RELAY, tokenA);
|
||||
assert.deepEqual(sections.readChannelSectionsOutbox(PK, RELAY).sections, [
|
||||
{ id: "s1", name: "Renamed", order: 0 },
|
||||
]);
|
||||
});
|
||||
});
|
||||
|
||||
// ── (e) legacy token-less resume: an outbox entry written by a prior (pre-token)
|
||||
// build is a bare store, not a `{store, token}` envelope. It must still
|
||||
// parse, resume, and be unconditionally clearable.
|
||||
|
||||
test("(e) stars: a legacy bare-store outbox entry parses and is clearable", () => {
|
||||
withFreshStorage((ls) => {
|
||||
// Simulate a pre-token build's write: bare store under the outbox key.
|
||||
const key = `buzz-channel-stars-outbox.v1:${PK}:${encodeURIComponent(RELAY)}`;
|
||||
ls.setItem(key, JSON.stringify(starStore({ a: starEntry(true, 100, 0) })));
|
||||
const resumed = stars.readChannelStarsOutbox(PK, RELAY);
|
||||
assert.ok(resumed, "legacy bare-store entry must parse");
|
||||
assert.deepEqual(resumed.channels.a, starEntry(true, 100, 0));
|
||||
// A token-carrying clear must still remove a legacy (token-less) entry.
|
||||
stars.clearChannelStarsOutbox(PK, RELAY, mintOutboxToken());
|
||||
assert.equal(stars.readChannelStarsOutbox(PK, RELAY), null);
|
||||
});
|
||||
});
|
||||
|
||||
test("(e) sort: a legacy bare-store outbox entry parses and is clearable", () => {
|
||||
withFreshStorage((ls) => {
|
||||
const key = `buzz-channel-sort-outbox.v1:${PK}:${encodeURIComponent(RELAY)}`;
|
||||
ls.setItem(key, JSON.stringify(sortStore({ forums: "recent" })));
|
||||
const resumed = sort.readChannelSortOutbox(PK, RELAY);
|
||||
assert.ok(resumed, "legacy bare-store entry must parse");
|
||||
assert.equal(resumed.groups.forums, "recent");
|
||||
sort.clearChannelSortOutbox(PK, RELAY, mintOutboxToken());
|
||||
assert.equal(sort.readChannelSortOutbox(PK, RELAY), null);
|
||||
});
|
||||
});
|
||||
|
||||
// ── Token-envelope round-trip + omitted-token unconditional clear ──────────────
|
||||
|
||||
test("write/read/clear: envelope round-trips and an omitted token clears unconditionally", () => {
|
||||
withFreshStorage(() => {
|
||||
const token = mintOutboxToken();
|
||||
stars.writeChannelStarsOutbox(
|
||||
PK,
|
||||
starStore({ a: starEntry(true, 100, 1) }),
|
||||
RELAY,
|
||||
token,
|
||||
);
|
||||
assert.deepEqual(
|
||||
stars.readChannelStarsOutbox(PK, RELAY).channels.a,
|
||||
starEntry(true, 100, 1),
|
||||
);
|
||||
// Omitted token = unconditional clear (bootstrap "nothing to resume" path).
|
||||
stars.clearChannelStarsOutbox(PK, RELAY);
|
||||
assert.equal(stars.readChannelStarsOutbox(PK, RELAY), null);
|
||||
});
|
||||
});
|
||||
|
||||
test("mintOutboxToken: successive mints are distinct", () => {
|
||||
const seen = new Set();
|
||||
for (let i = 0; i < 1000; i++) seen.add(mintOutboxToken());
|
||||
assert.equal(seen.size, 1000, "every minted token must be unique");
|
||||
});
|
||||
@@ -52,6 +52,120 @@ export function clampPublishCreatedAt(
|
||||
);
|
||||
}
|
||||
|
||||
// Monotonic fallback counter so two token mints in the same millisecond within
|
||||
// one window (or in an environment without `crypto.randomUUID`) still differ.
|
||||
let tokenCounter = 0;
|
||||
|
||||
/**
|
||||
* Mint a per-write ownership token for a durable outbox entry.
|
||||
*
|
||||
* Every window writes the outbox under one shared `(pubkey, relay)` key but
|
||||
* holds only an in-memory generation, so a completion in one window must never
|
||||
* clear an entry another window overwrote. The token identifies which write
|
||||
* produced the persisted entry: a completing publish clears only when the
|
||||
* stored token still matches the one it wrote (compare-and-clear), so a peer's
|
||||
* newer write — which replaced the token — survives an older window's ACK.
|
||||
*
|
||||
* Prefers `crypto.randomUUID` for cross-window uniqueness; the counter/time/
|
||||
* random fallback keeps tokens distinct where it is unavailable (older runtimes
|
||||
* and the test harness).
|
||||
*/
|
||||
export function mintOutboxToken(): string {
|
||||
const uuid = globalThis.crypto?.randomUUID?.();
|
||||
if (uuid) return uuid;
|
||||
return `${Date.now().toString(36)}-${(tokenCounter++).toString(36)}-${Math.random().toString(36).slice(2)}`;
|
||||
}
|
||||
|
||||
/**
|
||||
* Read a durable outbox entry, tolerating both the token envelope written by
|
||||
* this build (`{ store, token }`) and the bare-store shape written by a prior
|
||||
* build (token absent). Returns the parsed store plus its ownership token
|
||||
* (`null` for a legacy entry, which is therefore unconditionally clearable), or
|
||||
* `null` when the key is empty/unparseable or the store fails validation.
|
||||
*/
|
||||
export function readOutboxEntry<T>(
|
||||
key: string,
|
||||
parseStore: (json: unknown) => T | null,
|
||||
): { store: T; token: string | null } | null {
|
||||
try {
|
||||
const raw = window.localStorage.getItem(key);
|
||||
if (!raw) return null;
|
||||
const json = JSON.parse(raw);
|
||||
if (
|
||||
json !== null &&
|
||||
typeof json === "object" &&
|
||||
!Array.isArray(json) &&
|
||||
"store" in (json as Record<string, unknown>)
|
||||
) {
|
||||
const env = json as { store: unknown; token?: unknown };
|
||||
const store = parseStore(env.store);
|
||||
if (!store) return null;
|
||||
return {
|
||||
store,
|
||||
token: typeof env.token === "string" ? env.token : null,
|
||||
};
|
||||
}
|
||||
// Legacy bare-store shape from a pre-token build. Token absent ⇒ clearable.
|
||||
const store = parseStore(json);
|
||||
if (!store) return null;
|
||||
return { store, token: null };
|
||||
} catch {
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
/** Persist a durable outbox entry as a token envelope. Best-effort. */
|
||||
export function writeOutboxEntry(
|
||||
key: string,
|
||||
store: unknown,
|
||||
token: string,
|
||||
): void {
|
||||
try {
|
||||
window.localStorage.setItem(key, JSON.stringify({ store, token }));
|
||||
} catch {
|
||||
// Best-effort durability; the in-memory pending edit still drives this
|
||||
// session's publish even if the persisted copy could not be written.
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Clear a durable outbox entry with cross-window ownership.
|
||||
*
|
||||
* When `token` is provided this is a compare-and-clear: the entry is removed
|
||||
* only if its stored token still matches (or is a legacy token-less value).
|
||||
* A peer window that overwrote the entry replaced the token, so an older
|
||||
* window's completing publish no-ops here and the peer's still-unpublished
|
||||
* edit survives. An omitted `token` clears unconditionally (used when the
|
||||
* caller has already established there is nothing worth preserving).
|
||||
*/
|
||||
export function clearOutboxEntry(key: string, token?: string): void {
|
||||
try {
|
||||
const raw = window.localStorage.getItem(key);
|
||||
if (raw === null) return;
|
||||
if (token !== undefined) {
|
||||
let stored: string | null = null;
|
||||
try {
|
||||
const json = JSON.parse(raw);
|
||||
if (
|
||||
json !== null &&
|
||||
typeof json === "object" &&
|
||||
typeof (json as { token?: unknown }).token === "string"
|
||||
) {
|
||||
stored = (json as { token: string }).token;
|
||||
}
|
||||
} catch {
|
||||
// Unparseable ⇒ treat as legacy/foreign and fall through to remove.
|
||||
}
|
||||
// A newer write replaced our entry — leave it for its owner to clear.
|
||||
if (stored !== null && stored !== token) return;
|
||||
}
|
||||
window.localStorage.removeItem(key);
|
||||
} catch {
|
||||
// Ignore — a stale entry is re-evaluated (and re-cleared if identical to
|
||||
// the head) on the next publish attempt.
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Tri-state result returned by every `fetchRemote*()` method.
|
||||
*
|
||||
|
||||
@@ -84,9 +84,15 @@ export function useChannelMutes(
|
||||
(remote: RemoteMutes): ((prev: ChannelMuteStore) => ChannelMuteStore) => {
|
||||
return (prev) => {
|
||||
if (!pubkey) return prev;
|
||||
const merged = mergeStores(prev, remote.store);
|
||||
if (!writeChannelMutesStore(pubkey, merged)) return prev;
|
||||
return merged;
|
||||
// Read-merge-write folds the head into whatever a peer window has
|
||||
// persisted since; use the returned store so a concurrent click there
|
||||
// is carried into this window's state rather than lost.
|
||||
const persisted = writeChannelMutesStore(
|
||||
pubkey,
|
||||
mergeStores(prev, remote.store),
|
||||
);
|
||||
if (!persisted) return prev;
|
||||
return persisted;
|
||||
};
|
||||
},
|
||||
[pubkey],
|
||||
@@ -243,9 +249,15 @@ export function useChannelMutes(
|
||||
},
|
||||
channelId,
|
||||
);
|
||||
if (!writeChannelMutesStore(pubkey, next)) return prev;
|
||||
manager?.publishMutes(next);
|
||||
return next;
|
||||
// Read-merge-write: fold this click into any concurrent peer-window
|
||||
// click already persisted under the shared key, then thread the merged
|
||||
// store into both React state and the publish so neither window's edit
|
||||
// is dropped (Carl prong b). Preserve the clicked channel through the
|
||||
// re-bound so a same-second mutation is not evicted at capacity.
|
||||
const persisted = writeChannelMutesStore(pubkey, next, channelId);
|
||||
if (!persisted) return prev;
|
||||
manager?.publishMutes(persisted);
|
||||
return persisted;
|
||||
});
|
||||
},
|
||||
[pubkey],
|
||||
|
||||
@@ -84,9 +84,15 @@ export function useChannelStars(
|
||||
(remote: RemoteStars): ((prev: ChannelStarStore) => ChannelStarStore) => {
|
||||
return (prev) => {
|
||||
if (!pubkey) return prev;
|
||||
const merged = mergeStores(prev, remote.store);
|
||||
if (!writeChannelStarsStore(pubkey, merged)) return prev;
|
||||
return merged;
|
||||
// Read-merge-write folds the head into whatever a peer window has
|
||||
// persisted since; use the returned store so a concurrent click there
|
||||
// is carried into this window's state rather than lost.
|
||||
const persisted = writeChannelStarsStore(
|
||||
pubkey,
|
||||
mergeStores(prev, remote.store),
|
||||
);
|
||||
if (!persisted) return prev;
|
||||
return persisted;
|
||||
};
|
||||
},
|
||||
[pubkey],
|
||||
@@ -243,9 +249,15 @@ export function useChannelStars(
|
||||
},
|
||||
channelId,
|
||||
);
|
||||
if (!writeChannelStarsStore(pubkey, next)) return prev;
|
||||
manager?.publishStars(next);
|
||||
return next;
|
||||
// Read-merge-write: fold this click into any concurrent peer-window
|
||||
// click already persisted under the shared key, then thread the merged
|
||||
// store into both React state and the publish so neither window's edit
|
||||
// is dropped (Carl prong b). Preserve the clicked channel through the
|
||||
// re-bound so a same-second mutation is not evicted at capacity.
|
||||
const persisted = writeChannelStarsStore(pubkey, next, channelId);
|
||||
if (!persisted) return prev;
|
||||
manager?.publishStars(persisted);
|
||||
return persisted;
|
||||
});
|
||||
},
|
||||
[pubkey],
|
||||
|
||||
Reference in New Issue
Block a user