mirror of
https://github.com/block/buzz.git
synced 2026-11-01 15:01:12 +00:00
fix(sync): key the sidebar durable outbox per window
localStorage has no atomic compare-and-delete or transactional read-modify-write, so a single outbox key shared across every window could not be mutated safely: one window's read-then-write or read-then-remove races a peer's write in the gap and drops its still-unpublished edit. A per-write ownership token narrowed that window but could not close it — the token proves what was read, not what is still stored at the destructive op. Key the outbox per window instead: <prefix>:<pubkey>:<relay>:<nonce>, where the nonce is minted once and parked in sessionStorage. Each window is the sole writer of its own key, so a hot-path write is one unconditional setItem — the write race is designed out, not guarded. Resume enumerates every window's key: merge lanes (stars/mutes) fold all records order-independently; whole-blob lanes (sort/sections) replay the max-queuedAt record with a nonce tiebreak. Redundant foreign keys are reclaimed at boot, gated on durable relay evidence (merge: head subsumes; whole-blob: head created_at supersedes) and re-read immediately before removal so a live peer's fresh write in the recheck gap survives. Reclamation runs only on a successful head fetch, never on a failed one. The token contract is deleted entirely — ownership is the key. Co-authored-by: Will Pfleger <pfleger.will@gmail.com> Signed-off-by: Will Pfleger <pfleger.will@gmail.com>
This commit is contained in:
@@ -1,8 +1,9 @@
|
||||
import { normalizeRelayUrl } from "@/shared/lib/normalizeRelayUrl";
|
||||
import {
|
||||
clearOutboxEntry,
|
||||
readOutboxEntry,
|
||||
writeOutboxEntry,
|
||||
clearOwnOutbox,
|
||||
enumerateOutbox,
|
||||
reclaimOutbox,
|
||||
writeOwnOutbox,
|
||||
} from "./sidebarSyncWatermark";
|
||||
|
||||
const STORAGE_KEY_PREFIX = "buzz-channel-mutes.v1";
|
||||
@@ -125,23 +126,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.
|
||||
* Persist the main store. Writes the passed store as-is (bounded) — no read of
|
||||
* the shared key, so it is never a shared-key read-modify-write. Callers merge
|
||||
* peer state into the window's OWN React state (via the storage-event handler
|
||||
* and applyRemote) before calling here, so the write carries an owned, merged
|
||||
* value. Returns the bounded store, or `null` on write failure.
|
||||
*
|
||||
* Cross-window convergence of the on-disk cache is eventual: a peer's storage
|
||||
* event folds into this window's state, and the relay reconcile writes the
|
||||
* merged head back. Durable no-loss of an unpublished click is held by the
|
||||
* per-window outbox, not this cache.
|
||||
*/
|
||||
export function writeChannelMutesStore(
|
||||
pubkey: string,
|
||||
incoming: ChannelMuteStore,
|
||||
store: ChannelMuteStore,
|
||||
preservedKey?: string,
|
||||
): ChannelMuteStore | null {
|
||||
try {
|
||||
const persisted = readChannelMutesStore(pubkey);
|
||||
const merged = mergeStores(persisted, incoming, preservedKey);
|
||||
window.localStorage.setItem(storageKey(pubkey), JSON.stringify(merged));
|
||||
return merged;
|
||||
const bounded = boundMuteStore(store, preservedKey);
|
||||
window.localStorage.setItem(storageKey(pubkey), JSON.stringify(bounded));
|
||||
return bounded;
|
||||
} catch {
|
||||
return null;
|
||||
}
|
||||
@@ -201,61 +205,95 @@ export function mutedChannelIdsFromStore(store: ChannelMuteStore): Set<string> {
|
||||
|
||||
const OUTBOX_KEY_PREFIX = "buzz-channel-mutes-outbox.v1";
|
||||
|
||||
// The outbox is a per-relay sync-lane structure (like the watermark), so it is
|
||||
// relay-scoped even though the main store stays pubkey-only: an edit made
|
||||
// against relay A must never resume-publish onto relay B after a community
|
||||
// switch.
|
||||
function outboxKey(pubkey: string, relayUrl: string): string {
|
||||
// The single shared key written by builds before the outbox was keyed
|
||||
// per-window. Enumerated as one more record so an edit persisted by a prior
|
||||
// build still resumes, and reclaimed by the same relay-gated rule.
|
||||
function legacyOutboxKey(pubkey: string, relayUrl: string): string {
|
||||
return `${OUTBOX_KEY_PREFIX}:${pubkey}:${encodeURIComponent(normalizeRelayUrl(relayUrl))}`;
|
||||
}
|
||||
|
||||
/**
|
||||
* Persist an unpublished edit so it survives quit/community-switch within the
|
||||
* 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.
|
||||
* Persist this window's unpublished edit under its own outbox key. Written
|
||||
* synchronously on every click as a single unconditional `setItem` (no shared-
|
||||
* key read-modify-write); resumed by merging every window's record on next
|
||||
* mount so a click made <2s before quit/community-switch is never dropped.
|
||||
*/
|
||||
export function writeChannelMutesOutbox(
|
||||
pubkey: string,
|
||||
store: ChannelMuteStore,
|
||||
relayUrl: string,
|
||||
token: string,
|
||||
): void {
|
||||
const key = outboxKey(pubkey, relayUrl);
|
||||
const existing = readOutboxEntry(key, parseMutePayload)?.store;
|
||||
const merged = existing
|
||||
? mergeStores(existing, store)
|
||||
: boundMuteStore(store);
|
||||
writeOutboxEntry(key, merged, token);
|
||||
writeOwnOutbox(OUTBOX_KEY_PREFIX, pubkey, relayUrl, boundMuteStore(store));
|
||||
}
|
||||
|
||||
/** Read a persisted unpublished edit, or null when none/unparseable. */
|
||||
/**
|
||||
* Merge every window's persisted unpublished edit into one store for resume, or
|
||||
* null when none exists. Per-entry `mergeStores` is order-independent, so two
|
||||
* windows' concurrent clicks on different channels both survive.
|
||||
*/
|
||||
export function readChannelMutesOutbox(
|
||||
pubkey: string,
|
||||
relayUrl: string,
|
||||
): ChannelMuteStore | null {
|
||||
return (
|
||||
readOutboxEntry(outboxKey(pubkey, relayUrl), parseMutePayload)?.store ??
|
||||
null
|
||||
const records = enumerateOutbox(
|
||||
OUTBOX_KEY_PREFIX,
|
||||
legacyOutboxKey(pubkey, relayUrl),
|
||||
pubkey,
|
||||
relayUrl,
|
||||
parseMutePayload,
|
||||
);
|
||||
if (records.length === 0) return null;
|
||||
return records.reduce<ChannelMuteStore>(
|
||||
(acc, r) => mergeStores(acc, r.store),
|
||||
DEFAULT_STORE,
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
* 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.
|
||||
*/
|
||||
/** Clear this window's own outbox key (its edit published or is a no-op). */
|
||||
export function clearChannelMutesOutbox(
|
||||
pubkey: string,
|
||||
relayUrl: string,
|
||||
token?: string,
|
||||
): void {
|
||||
clearOutboxEntry(outboxKey(pubkey, relayUrl), token);
|
||||
clearOwnOutbox(OUTBOX_KEY_PREFIX, pubkey, relayUrl);
|
||||
}
|
||||
|
||||
/**
|
||||
* Reclaim foreign outbox keys the fetched relay head already subsumes: a record
|
||||
* is redundant when merging it into `head` yields `head` unchanged (the head
|
||||
* carries an entry at least as new for every channel). Never touches this
|
||||
* window's own key; a still-unpublished peer edit the head does not yet reflect
|
||||
* is kept. Call only after a successful head fetch.
|
||||
*/
|
||||
export function reclaimSubsumedMutesOutbox(
|
||||
pubkey: string,
|
||||
relayUrl: string,
|
||||
head: ChannelMuteStore,
|
||||
): void {
|
||||
reclaimOutbox(
|
||||
OUTBOX_KEY_PREFIX,
|
||||
legacyOutboxKey(pubkey, relayUrl),
|
||||
pubkey,
|
||||
relayUrl,
|
||||
parseMutePayload,
|
||||
(record) => muteStoresEqual(mergeStores(head, record.store), head),
|
||||
);
|
||||
}
|
||||
|
||||
/** Deep per-channel equality of two mute stores (order-independent). */
|
||||
function muteStoresEqual(a: ChannelMuteStore, b: ChannelMuteStore): boolean {
|
||||
const aKeys = Object.keys(a.channels);
|
||||
const bKeys = Object.keys(b.channels);
|
||||
if (aKeys.length !== bKeys.length) return false;
|
||||
for (const id of aKeys) {
|
||||
const l = a.channels[id];
|
||||
const r = b.channels[id];
|
||||
if (
|
||||
!r ||
|
||||
l.muted !== r.muted ||
|
||||
l.updatedAt !== r.updatedAt ||
|
||||
l.rev !== r.rev
|
||||
)
|
||||
return false;
|
||||
}
|
||||
return true;
|
||||
}
|
||||
|
||||
@@ -2,6 +2,7 @@ import assert from "node:assert/strict";
|
||||
import test, { mock } from "node:test";
|
||||
|
||||
import { relayClient } from "@/shared/api/relayClient";
|
||||
import { readChannelMutesOutbox } from "./channelMutesStorage.ts";
|
||||
import { ChannelMuteSyncManager } from "./channelMutesSync.ts";
|
||||
import {
|
||||
installFakeWindow,
|
||||
@@ -28,6 +29,10 @@ function makeMultiTimerWindow() {
|
||||
getItem: (k) => storage.get(k) ?? null,
|
||||
setItem: (k, v) => storage.set(k, v),
|
||||
removeItem: (k) => storage.delete(k),
|
||||
get length() {
|
||||
return storage.size;
|
||||
},
|
||||
key: (i) => [...storage.keys()][i] ?? null,
|
||||
},
|
||||
setTimeout: (fn, ms) => {
|
||||
const id = nextId++;
|
||||
@@ -167,7 +172,6 @@ test("A-in-flight → B-click → A-succeeds: B stays pending and B publishes",
|
||||
const t = makeMultiTimerWindow();
|
||||
const restore = installFakeWindow(t.win);
|
||||
const tauri = installTauriMock("{}");
|
||||
const outboxKey = `buzz-channel-mutes-outbox.v1:pk-ab:${RELAY_KEY}`;
|
||||
try {
|
||||
const manager = new ChannelMuteSyncManager("pk-ab", RELAY);
|
||||
const storeA = makeStore({ a: E(true, 100, 1) });
|
||||
@@ -184,7 +188,7 @@ test("A-in-flight → B-click → A-succeeds: B stays pending and B publishes",
|
||||
["b"],
|
||||
"B is now pending",
|
||||
);
|
||||
assert.ok(t.storage.get(outboxKey), "outbox holds B");
|
||||
assert.ok(readChannelMutesOutbox("pk-ab", RELAY), "outbox holds B");
|
||||
|
||||
// A completes — must NOT clear B.
|
||||
releaseFirst();
|
||||
@@ -194,7 +198,10 @@ test("A-in-flight → B-click → A-succeeds: B stays pending and B publishes",
|
||||
["b"],
|
||||
"older A completion leaves B pending",
|
||||
);
|
||||
assert.ok(t.storage.get(outboxKey), "older A completion leaves B outbox");
|
||||
assert.ok(
|
||||
readChannelMutesOutbox("pk-ab", RELAY),
|
||||
"older A completion leaves B outbox",
|
||||
);
|
||||
|
||||
// B's own debounce fires and B reaches the relay (published) with no kick.
|
||||
const capturedBefore = tauri.capturedPlaintext();
|
||||
@@ -210,7 +217,11 @@ test("A-in-flight → B-click → A-succeeds: B stays pending and B publishes",
|
||||
null,
|
||||
"B cleared after publish",
|
||||
);
|
||||
assert.equal(t.storage.get(outboxKey), undefined, "B outbox cleared");
|
||||
assert.equal(
|
||||
readChannelMutesOutbox("pk-ab", RELAY),
|
||||
null,
|
||||
"B outbox cleared",
|
||||
);
|
||||
manager.destroy();
|
||||
} finally {
|
||||
tauri.restore();
|
||||
@@ -237,7 +248,6 @@ test("A-in-flight → B-click → A-fails: B remains pending and B publishes", a
|
||||
const t = makeMultiTimerWindow();
|
||||
const restore = installFakeWindow(t.win);
|
||||
const tauri = installTauriMock("{}");
|
||||
const outboxKey = `buzz-channel-mutes-outbox.v1:pk-abfail:${RELAY_KEY}`;
|
||||
try {
|
||||
const manager = new ChannelMuteSyncManager("pk-abfail", RELAY);
|
||||
manager.publishMutes(makeStore({ a: E(true, 100, 1) }));
|
||||
@@ -253,7 +263,10 @@ test("A-in-flight → B-click → A-fails: B remains pending and B publishes", a
|
||||
["b"],
|
||||
"B still pending after A's failure",
|
||||
);
|
||||
assert.ok(t.storage.get(outboxKey), "B outbox intact after A's failure");
|
||||
assert.ok(
|
||||
readChannelMutesOutbox("pk-abfail", RELAY),
|
||||
"B outbox intact after A's failure",
|
||||
);
|
||||
|
||||
// B's debounce fires and B publishes successfully.
|
||||
await t.fireDelay(2000);
|
||||
|
||||
@@ -16,7 +16,6 @@ import {
|
||||
import {
|
||||
advanceWatermark,
|
||||
clampPublishCreatedAt,
|
||||
mintOutboxToken,
|
||||
readWatermark,
|
||||
runBootstrap,
|
||||
type FetchResult,
|
||||
@@ -63,13 +62,6 @@ 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
|
||||
@@ -169,16 +161,10 @@ 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). 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,
|
||||
);
|
||||
// survives teardown and resumes on next mount (durable outbox). This
|
||||
// window's own key is the only one written — a single unconditional
|
||||
// setItem, never a shared-key read-modify-write.
|
||||
writeChannelMutesOutbox(this.pubkey, store, this.relayUrl);
|
||||
if (this.debounceTimer !== null) {
|
||||
window.clearTimeout(this.debounceTimer);
|
||||
}
|
||||
@@ -265,30 +251,15 @@ export class ChannelMuteSyncManager {
|
||||
}
|
||||
|
||||
/**
|
||||
* Clear the in-memory pending edit and its durable outbox — but only if the
|
||||
* completing publish still owns the current generation. A publish for an
|
||||
* older edit that finishes after a newer edit was queued must leave the newer
|
||||
* edit (and its retry state) untouched.
|
||||
* Clear the in-memory pending edit and this window's own durable outbox key —
|
||||
* but only if the completing publish still owns the current generation. A
|
||||
* publish for an older edit that finishes after a newer edit was queued must
|
||||
* leave the newer edit (and its retry state) untouched.
|
||||
*/
|
||||
private discardPending(gen: number): void {
|
||||
if (gen !== this.pendingGeneration) return;
|
||||
this.pendingStore = null;
|
||||
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;
|
||||
clearChannelMutesOutbox(this.pubkey, this.relayUrl);
|
||||
}
|
||||
|
||||
/** Schedule a bounded-backoff retry of the retained pending edit. */
|
||||
|
||||
@@ -1,8 +1,9 @@
|
||||
import { normalizeRelayUrl } from "@/shared/lib/normalizeRelayUrl";
|
||||
import {
|
||||
clearOutboxEntry,
|
||||
readOutboxEntry,
|
||||
writeOutboxEntry,
|
||||
clearOwnOutbox,
|
||||
latestOutbox,
|
||||
reclaimOutbox,
|
||||
writeOwnOutbox,
|
||||
} from "./sidebarSyncWatermark";
|
||||
|
||||
const STORAGE_KEY_PREFIX = "buzz-channel-sections.v1";
|
||||
@@ -202,55 +203,79 @@ export function writeChannelSectionsStore(
|
||||
|
||||
const OUTBOX_KEY_PREFIX = "buzz-channel-sections-outbox.v1";
|
||||
|
||||
function outboxKey(pubkey: string, relayUrl: string): string {
|
||||
// The single shared key written by builds before the outbox was keyed
|
||||
// per-window. Enumerated as one more record so an edit persisted by a prior
|
||||
// build still resumes, and reclaimed by the same relay-gated rule.
|
||||
function legacyOutboxKey(pubkey: string, relayUrl: string): string {
|
||||
return `${OUTBOX_KEY_PREFIX}:${pubkey}:${encodeURIComponent(normalizeRelayUrl(relayUrl))}`;
|
||||
}
|
||||
|
||||
/**
|
||||
* Persist an unpublished edit so it survives quit/community-switch within the
|
||||
* 2s publish debounce. Written synchronously on every edit; cleared once 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.
|
||||
* Persist this window's unpublished edit under its own outbox key. Written
|
||||
* synchronously on every edit as a single unconditional `setItem` (no shared-
|
||||
* key read-modify-write); resumed on next mount so an edit made <2s before
|
||||
* quit/community-switch is never dropped. `queuedAt` stamps the write so resume
|
||||
* replays only the newest queued blob (whole-blob LWW).
|
||||
*/
|
||||
export function writeChannelSectionsOutbox(
|
||||
pubkey: string,
|
||||
store: ChannelSectionStore,
|
||||
relayUrl: string,
|
||||
token: string,
|
||||
): void {
|
||||
writeOutboxEntry(
|
||||
outboxKey(pubkey, relayUrl),
|
||||
writeOwnOutbox(
|
||||
OUTBOX_KEY_PREFIX,
|
||||
pubkey,
|
||||
relayUrl,
|
||||
boundChannelSectionsStore(store),
|
||||
token,
|
||||
);
|
||||
}
|
||||
|
||||
/** Read a persisted unpublished edit, or null when none/unparseable. */
|
||||
export function readChannelSectionsOutbox(
|
||||
pubkey: string,
|
||||
relayUrl: string,
|
||||
): ChannelSectionStore | null {
|
||||
return (
|
||||
readOutboxEntry(outboxKey(pubkey, relayUrl), parseChannelSectionPayload)
|
||||
?.store ?? null
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
* 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.
|
||||
* The newest window's persisted unpublished edit for resume, or null when none
|
||||
* exists. Whole-blob LWW: only the max-`queuedAt` record is replayed, so an
|
||||
* older blob a peer queued is superseded by definition, never resurrected.
|
||||
*/
|
||||
export function readChannelSectionsOutbox(
|
||||
pubkey: string,
|
||||
relayUrl: string,
|
||||
): ChannelSectionStore | null {
|
||||
return latestOutbox(
|
||||
OUTBOX_KEY_PREFIX,
|
||||
legacyOutboxKey(pubkey, relayUrl),
|
||||
pubkey,
|
||||
relayUrl,
|
||||
parseChannelSectionPayload,
|
||||
);
|
||||
}
|
||||
|
||||
/** Clear this window's own outbox key (its edit published or is a no-op). */
|
||||
export function clearChannelSectionsOutbox(
|
||||
pubkey: string,
|
||||
relayUrl: string,
|
||||
token?: string,
|
||||
): void {
|
||||
clearOutboxEntry(outboxKey(pubkey, relayUrl), token);
|
||||
clearOwnOutbox(OUTBOX_KEY_PREFIX, pubkey, relayUrl);
|
||||
}
|
||||
|
||||
/**
|
||||
* Reclaim foreign outbox keys the relay head itself supersedes: a whole-blob
|
||||
* record queued at or before the durable head's `created_at` (`queuedAt` ≤
|
||||
* `headCreatedAt`) lost LWW to a blob the relay already holds, so dropping it
|
||||
* matches the relay's own resolution. A record queued AFTER the head is live
|
||||
* intent the head does not yet reflect and is kept — never reclaimed merely for
|
||||
* losing a local max-`queuedAt` comparison. Never touches this window's own key
|
||||
* and rechecks each key before removing. Call only after a successful fetch.
|
||||
*/
|
||||
export function reclaimSupersededSectionsOutbox(
|
||||
pubkey: string,
|
||||
relayUrl: string,
|
||||
headCreatedAt: number,
|
||||
): void {
|
||||
reclaimOutbox(
|
||||
OUTBOX_KEY_PREFIX,
|
||||
legacyOutboxKey(pubkey, relayUrl),
|
||||
pubkey,
|
||||
relayUrl,
|
||||
parseChannelSectionPayload,
|
||||
(record) => record.queuedAt <= headCreatedAt,
|
||||
);
|
||||
}
|
||||
|
||||
@@ -2,6 +2,7 @@ import assert from "node:assert/strict";
|
||||
import test, { mock } from "node:test";
|
||||
|
||||
import { relayClient } from "@/shared/api/relayClient";
|
||||
import { readChannelSectionsOutbox } from "./channelSectionsStorage.ts";
|
||||
import { ChannelSectionSyncManager } from "./channelSectionsSync.ts";
|
||||
import {
|
||||
makeFakeWindow,
|
||||
@@ -219,9 +220,7 @@ test("adopt-winner: newer remote head at pre-publish adopts remote and skips pub
|
||||
);
|
||||
// Outbox persisted synchronously on the edit.
|
||||
assert.ok(
|
||||
fw.localStorage.getItem(
|
||||
`buzz-channel-sections-outbox.v1:pk-lww:${RELAY_KEY}`,
|
||||
) !== null,
|
||||
readChannelSectionsOutbox("pk-lww", RELAY) !== null,
|
||||
"edit must be persisted to the durable outbox",
|
||||
);
|
||||
fw._fireTimer();
|
||||
@@ -238,9 +237,7 @@ test("adopt-winner: newer remote head at pre-publish adopts remote and skips pub
|
||||
);
|
||||
assert.equal(manager.getPendingStore(), null, "pending must be cleared");
|
||||
assert.equal(
|
||||
fw.localStorage.getItem(
|
||||
`buzz-channel-sections-outbox.v1:pk-lww:${RELAY_KEY}`,
|
||||
),
|
||||
readChannelSectionsOutbox("pk-lww", RELAY),
|
||||
null,
|
||||
"outbox must be cleared on adopt so the loser is never replayed",
|
||||
);
|
||||
@@ -271,9 +268,7 @@ test("adopt-winner: local edit at/ahead of head publishes and clears outbox", as
|
||||
await new Promise((r) => setTimeout(r, 20));
|
||||
assert.equal(publishCalls.length, 1, "local edit must be published");
|
||||
assert.equal(
|
||||
fw.localStorage.getItem(
|
||||
`buzz-channel-sections-outbox.v1:pk-win:${RELAY_KEY}`,
|
||||
),
|
||||
readChannelSectionsOutbox("pk-win", RELAY),
|
||||
null,
|
||||
"outbox must be cleared once the edit is published",
|
||||
);
|
||||
@@ -414,6 +409,10 @@ test("overlapping publishes: older completion does not erase a newer queued edit
|
||||
getItem: (k) => storage.get(k) ?? null,
|
||||
setItem: (k, v) => storage.set(k, v),
|
||||
removeItem: (k) => storage.delete(k),
|
||||
get length() {
|
||||
return storage.size;
|
||||
},
|
||||
key: (i) => [...storage.keys()][i] ?? null,
|
||||
},
|
||||
setTimeout: (fn, ms) => {
|
||||
const id = nextId++;
|
||||
@@ -432,7 +431,6 @@ test("overlapping publishes: older completion does not erase a newer queued edit
|
||||
};
|
||||
const restore = installFakeWindow(fakeWindow);
|
||||
const tauri = installTauriMock("{}");
|
||||
const outboxKey = `buzz-channel-sections-outbox.v1:pk-overlap:${RELAY_KEY}`;
|
||||
try {
|
||||
const manager = new ChannelSectionSyncManager("pk-overlap", RELAY);
|
||||
const storeA = makeSectionsStore([{ id: "a", name: "A", order: 0 }]);
|
||||
@@ -450,7 +448,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)).store.sections[0].id,
|
||||
readChannelSectionsOutbox("pk-overlap", RELAY).sections[0].id,
|
||||
"b",
|
||||
"outbox holds B",
|
||||
);
|
||||
@@ -467,7 +465,7 @@ test("overlapping publishes: older completion does not erase a newer queued edit
|
||||
"older completion must leave B pending",
|
||||
);
|
||||
assert.ok(
|
||||
storage.get(outboxKey) !== undefined,
|
||||
readChannelSectionsOutbox("pk-overlap", RELAY) !== null,
|
||||
"older completion must leave B's outbox intact",
|
||||
);
|
||||
assert.ok(
|
||||
@@ -520,6 +518,10 @@ test("live remote during debounce is adopted at pre-publish, not overwritten", a
|
||||
getItem: (k) => storage.get(k) ?? null,
|
||||
setItem: (k, v) => storage.set(k, v),
|
||||
removeItem: (k) => storage.delete(k),
|
||||
get length() {
|
||||
return storage.size;
|
||||
},
|
||||
key: (i) => [...storage.keys()][i] ?? null,
|
||||
},
|
||||
setTimeout: (fn, ms) => {
|
||||
const id = nextId++;
|
||||
@@ -543,7 +545,6 @@ test("live remote during debounce is adopted at pre-publish, not overwritten", a
|
||||
assignments: {},
|
||||
}),
|
||||
);
|
||||
const outboxKey = `buzz-channel-sections-outbox.v1:pk-livedebounce:${RELAY_KEY}`;
|
||||
try {
|
||||
const manager = new ChannelSectionSyncManager("pk-livedebounce", RELAY);
|
||||
const adopted = [];
|
||||
@@ -573,8 +574,8 @@ test("live remote during debounce is adopted at pre-publish, not overwritten", a
|
||||
);
|
||||
assert.equal(manager.getPendingStore(), null, "pending cleared on adopt");
|
||||
assert.equal(
|
||||
storage.get(outboxKey),
|
||||
undefined,
|
||||
readChannelSectionsOutbox("pk-livedebounce", RELAY),
|
||||
null,
|
||||
"outbox cleared on adopt so the loser can't replay",
|
||||
);
|
||||
manager.destroy();
|
||||
|
||||
@@ -16,7 +16,6 @@ import {
|
||||
import {
|
||||
advanceWatermark,
|
||||
clampPublishCreatedAt,
|
||||
mintOutboxToken,
|
||||
readWatermark,
|
||||
runBootstrap,
|
||||
type FetchResult,
|
||||
@@ -143,12 +142,6 @@ 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
|
||||
@@ -263,37 +256,22 @@ export class ChannelSectionSyncManager {
|
||||
this.recordRemoteHead(remote.createdAt, remote.eventId);
|
||||
if (gen !== this.pendingGeneration) return;
|
||||
this.pendingStore = null;
|
||||
this.clearOwnedOutbox();
|
||||
clearChannelSectionsOutbox(this.pubkey, this.relayUrl);
|
||||
this.lastPublishedStore = remote.store;
|
||||
if (this.destroyed) return;
|
||||
this.onRemoteAdopted?.(remote);
|
||||
}
|
||||
|
||||
/**
|
||||
* Clear the in-memory pending edit and its durable outbox — but only if the
|
||||
* completing publish still owns the current generation. A publish for an
|
||||
* older edit that finishes after a newer edit was queued must leave the newer
|
||||
* edit (and its retry state) untouched.
|
||||
* Clear the in-memory pending edit and this window's own durable outbox key —
|
||||
* but only if the completing publish still owns the current generation. A
|
||||
* publish for an older edit that finishes after a newer edit was queued must
|
||||
* leave the newer edit (and its retry state) untouched.
|
||||
*/
|
||||
private discardPending(gen: number): void {
|
||||
if (gen !== this.pendingGeneration) return;
|
||||
this.pendingStore = null;
|
||||
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;
|
||||
clearChannelSectionsOutbox(this.pubkey, this.relayUrl);
|
||||
}
|
||||
|
||||
publishSections(store: ChannelSectionStore): void {
|
||||
@@ -310,16 +288,11 @@ 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). 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,
|
||||
);
|
||||
// survives teardown and resumes on next mount (durable outbox). This
|
||||
// window's own key is the only one written — a single unconditional
|
||||
// setItem, never a shared-key read-modify-write; `queuedAt` stamps it so
|
||||
// resume replays only the newest queued blob (whole-blob LWW).
|
||||
writeChannelSectionsOutbox(this.pubkey, store, this.relayUrl);
|
||||
if (this.debounceTimer !== null) {
|
||||
window.clearTimeout(this.debounceTimer);
|
||||
}
|
||||
|
||||
@@ -1,9 +1,10 @@
|
||||
import { normalizeRelayUrl } from "@/shared/lib/normalizeRelayUrl";
|
||||
import type { Channel } from "@/shared/api/types";
|
||||
import {
|
||||
clearOutboxEntry,
|
||||
readOutboxEntry,
|
||||
writeOutboxEntry,
|
||||
clearOwnOutbox,
|
||||
latestOutbox,
|
||||
reclaimOutbox,
|
||||
writeOwnOutbox,
|
||||
} from "./sidebarSyncWatermark";
|
||||
|
||||
const STORAGE_KEY_PREFIX = "buzz-channel-sort.v1";
|
||||
@@ -140,57 +141,78 @@ export function writeChannelSortStore(
|
||||
|
||||
const OUTBOX_KEY_PREFIX = "buzz-channel-sort-outbox.v1";
|
||||
|
||||
function outboxKey(pubkey: string, relayUrl: string): string {
|
||||
// The single shared key written by builds before the outbox was keyed
|
||||
// per-window. Enumerated as one more record so an edit persisted by a prior
|
||||
// build still resumes, and reclaimed by the same relay-gated rule.
|
||||
function legacyOutboxKey(pubkey: string, relayUrl: string): string {
|
||||
return `${OUTBOX_KEY_PREFIX}:${pubkey}:${encodeURIComponent(normalizeRelayUrl(relayUrl))}`;
|
||||
}
|
||||
|
||||
/**
|
||||
* Persist an unpublished sort edit so it survives quit/community-switch within
|
||||
* the 2s publish debounce. Written synchronously on every edit; cleared once
|
||||
* 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.
|
||||
* Persist this window's unpublished sort edit under its own outbox key. Written
|
||||
* synchronously on every edit as a single unconditional `setItem` (no shared-
|
||||
* key read-modify-write); resumed on next mount so an edit made <2s before
|
||||
* quit/community-switch is never dropped. `queuedAt` stamps the write so resume
|
||||
* replays only the newest queued blob (whole-blob LWW).
|
||||
*/
|
||||
export function writeChannelSortOutbox(
|
||||
pubkey: string,
|
||||
store: ChannelSortStore,
|
||||
relayUrl: string,
|
||||
token: string,
|
||||
): void {
|
||||
writeOutboxEntry(
|
||||
outboxKey(pubkey, relayUrl),
|
||||
writeOwnOutbox(
|
||||
OUTBOX_KEY_PREFIX,
|
||||
pubkey,
|
||||
relayUrl,
|
||||
boundChannelSortStore(store),
|
||||
token,
|
||||
);
|
||||
}
|
||||
|
||||
/** Read a persisted unpublished sort edit, or null when none/unparseable. */
|
||||
export function readChannelSortOutbox(
|
||||
pubkey: string,
|
||||
relayUrl: string,
|
||||
): ChannelSortStore | null {
|
||||
return (
|
||||
readOutboxEntry(outboxKey(pubkey, relayUrl), parseChannelSortPayload)
|
||||
?.store ?? null
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
* 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.
|
||||
* The newest window's persisted unpublished sort edit for resume, or null when
|
||||
* none exists. Whole-blob LWW: only the max-`queuedAt` record is replayed, so
|
||||
* an older blob a peer queued is superseded by definition, never resurrected.
|
||||
*/
|
||||
export function clearChannelSortOutbox(
|
||||
export function readChannelSortOutbox(
|
||||
pubkey: string,
|
||||
relayUrl: string,
|
||||
token?: string,
|
||||
): ChannelSortStore | null {
|
||||
return latestOutbox(
|
||||
OUTBOX_KEY_PREFIX,
|
||||
legacyOutboxKey(pubkey, relayUrl),
|
||||
pubkey,
|
||||
relayUrl,
|
||||
parseChannelSortPayload,
|
||||
);
|
||||
}
|
||||
|
||||
/** Clear this window's own outbox key (its edit published or is a no-op). */
|
||||
export function clearChannelSortOutbox(pubkey: string, relayUrl: string): void {
|
||||
clearOwnOutbox(OUTBOX_KEY_PREFIX, pubkey, relayUrl);
|
||||
}
|
||||
|
||||
/**
|
||||
* Reclaim foreign outbox keys the relay head itself supersedes: a whole-blob
|
||||
* record queued at or before the durable head's `created_at` (`queuedAt` ≤
|
||||
* `headCreatedAt`) lost LWW to a blob the relay already holds, so dropping it
|
||||
* matches the relay's own resolution. A record queued AFTER the head is live
|
||||
* intent the head does not yet reflect and is kept — never reclaimed merely for
|
||||
* losing a local max-`queuedAt` comparison. Never touches this window's own key
|
||||
* and rechecks each key before removing. Call only after a successful fetch.
|
||||
*/
|
||||
export function reclaimSupersededSortOutbox(
|
||||
pubkey: string,
|
||||
relayUrl: string,
|
||||
headCreatedAt: number,
|
||||
): void {
|
||||
clearOutboxEntry(outboxKey(pubkey, relayUrl), token);
|
||||
reclaimOutbox(
|
||||
OUTBOX_KEY_PREFIX,
|
||||
legacyOutboxKey(pubkey, relayUrl),
|
||||
pubkey,
|
||||
relayUrl,
|
||||
parseChannelSortPayload,
|
||||
(record) => record.queuedAt <= headCreatedAt,
|
||||
);
|
||||
}
|
||||
|
||||
export function sortModeForGroup(
|
||||
|
||||
@@ -2,6 +2,7 @@ import assert from "node:assert/strict";
|
||||
import test, { mock } from "node:test";
|
||||
|
||||
import { relayClient } from "@/shared/api/relayClient";
|
||||
import { readChannelSortOutbox } from "./channelSortPreference.ts";
|
||||
import { ChannelSortSyncManager } from "./channelSortSync.ts";
|
||||
import {
|
||||
makeFakeWindow,
|
||||
@@ -15,7 +16,6 @@ function makeStore(groups = {}) {
|
||||
|
||||
const RELAY = "wss://r.test";
|
||||
const RELAY_KEY = encodeURIComponent(RELAY);
|
||||
const OUTBOX_KEY = (pk) => `buzz-channel-sort-outbox.v1:${pk}:${RELAY_KEY}`;
|
||||
|
||||
// ─── destroy() must cancel pending publish, not flush ─────────────────────────
|
||||
test("destroy: cancels pending publish without flushing to the relay", () => {
|
||||
@@ -185,7 +185,7 @@ test("adopt-winner: newer remote head at pre-publish adopts remote and skips pub
|
||||
manager.setOnRemoteAdopted((r) => adopted.push(r));
|
||||
manager.publishSortPrefs(makeStore({ "local-group": "recent" }));
|
||||
assert.ok(
|
||||
fw.localStorage.getItem(OUTBOX_KEY("pk-lww")) !== null,
|
||||
readChannelSortOutbox("pk-lww", RELAY) !== null,
|
||||
"edit must be persisted to the durable outbox",
|
||||
);
|
||||
fw._fireTimer();
|
||||
@@ -202,7 +202,7 @@ test("adopt-winner: newer remote head at pre-publish adopts remote and skips pub
|
||||
);
|
||||
assert.equal(manager.getPendingStore(), null, "pending must be cleared");
|
||||
assert.equal(
|
||||
fw.localStorage.getItem(OUTBOX_KEY("pk-lww")),
|
||||
readChannelSortOutbox("pk-lww", RELAY),
|
||||
null,
|
||||
"outbox must be cleared on adopt so the loser is never replayed",
|
||||
);
|
||||
@@ -231,7 +231,7 @@ test("adopt-winner: local edit at/ahead of head publishes and clears outbox", as
|
||||
await new Promise((r) => setTimeout(r, 20));
|
||||
assert.equal(publishCalls.length, 1, "local edit must be published");
|
||||
assert.equal(
|
||||
fw.localStorage.getItem(OUTBOX_KEY("pk-win")),
|
||||
readChannelSortOutbox("pk-win", RELAY),
|
||||
null,
|
||||
"outbox must be cleared once the edit is published",
|
||||
);
|
||||
@@ -309,20 +309,20 @@ test("durable outbox: edit destroyed inside the debounce resumes and publishes o
|
||||
// Window 1: edit then destroy before the debounce fires.
|
||||
const m1 = new ChannelSortSyncManager("pk-resume", RELAY);
|
||||
m1.publishSortPrefs(makeStore({ channels: "recent" }));
|
||||
const persisted = fw.localStorage.getItem(OUTBOX_KEY("pk-resume"));
|
||||
const persisted = readChannelSortOutbox("pk-resume", RELAY);
|
||||
assert.ok(persisted !== null, "edit must be persisted before teardown");
|
||||
m1.destroy();
|
||||
assert.equal(publishCalls.length, 0, "destroy must not flush");
|
||||
|
||||
// Window 2: resume the persisted outbox (the hook does this after bootstrap
|
||||
// via readChannelSortOutbox, which unwraps the `{store, token}` envelope).
|
||||
// via readChannelSortOutbox, which enumerates every window's outbox key).
|
||||
const m2 = new ChannelSortSyncManager("pk-resume", RELAY);
|
||||
m2.publishSortPrefs(JSON.parse(persisted).store);
|
||||
m2.publishSortPrefs(persisted);
|
||||
fw._fireTimer();
|
||||
await new Promise((r) => setTimeout(r, 20));
|
||||
assert.equal(publishCalls.length, 1, "resumed edit must publish");
|
||||
assert.equal(
|
||||
fw.localStorage.getItem(OUTBOX_KEY("pk-resume")),
|
||||
readChannelSortOutbox("pk-resume", RELAY),
|
||||
null,
|
||||
"outbox must be cleared once the resumed edit publishes",
|
||||
);
|
||||
@@ -399,6 +399,10 @@ test("overlapping publishes: older completion does not erase a newer queued edit
|
||||
getItem: (k) => storage.get(k) ?? null,
|
||||
setItem: (k, v) => storage.set(k, v),
|
||||
removeItem: (k) => storage.delete(k),
|
||||
get length() {
|
||||
return storage.size;
|
||||
},
|
||||
key: (i) => [...storage.keys()][i] ?? null,
|
||||
},
|
||||
setTimeout: (fn, ms) => {
|
||||
const id = nextId++;
|
||||
@@ -430,9 +434,7 @@ 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"))).store.groups,
|
||||
),
|
||||
Object.keys(readChannelSortOutbox("pk-overlap", RELAY)?.groups ?? {}),
|
||||
["dms"],
|
||||
"outbox holds B",
|
||||
);
|
||||
@@ -448,7 +450,7 @@ test("overlapping publishes: older completion does not erase a newer queued edit
|
||||
"older completion must leave B pending",
|
||||
);
|
||||
assert.ok(
|
||||
storage.get(OUTBOX_KEY("pk-overlap")) !== undefined,
|
||||
readChannelSortOutbox("pk-overlap", RELAY) !== null,
|
||||
"older completion must leave B's outbox intact",
|
||||
);
|
||||
manager.destroy();
|
||||
|
||||
@@ -15,7 +15,6 @@ import {
|
||||
import {
|
||||
advanceWatermark,
|
||||
clampPublishCreatedAt,
|
||||
mintOutboxToken,
|
||||
readWatermark,
|
||||
runBootstrap,
|
||||
type FetchResult,
|
||||
@@ -142,12 +141,6 @@ 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.
|
||||
@@ -259,37 +252,22 @@ export class ChannelSortSyncManager {
|
||||
this.recordRemoteHead(remote.createdAt, remote.eventId);
|
||||
if (gen !== this.pendingGeneration) return;
|
||||
this.pendingStore = null;
|
||||
this.clearOwnedOutbox();
|
||||
clearChannelSortOutbox(this.pubkey, this.relayUrl);
|
||||
this.lastPublishedStore = remote.store;
|
||||
if (this.destroyed) return;
|
||||
this.onRemoteAdopted?.(remote);
|
||||
}
|
||||
|
||||
/**
|
||||
* Clear the in-memory pending edit and its durable outbox — but only if the
|
||||
* completing publish still owns the current generation. A publish for an older
|
||||
* edit that finishes after a newer edit was queued must leave the newer edit
|
||||
* (and its retry state) untouched.
|
||||
* Clear the in-memory pending edit and this window's own durable outbox key —
|
||||
* but only if the completing publish still owns the current generation. A
|
||||
* publish for an older edit that finishes after a newer edit was queued must
|
||||
* leave the newer edit (and its retry state) untouched.
|
||||
*/
|
||||
private discardPending(gen: number): void {
|
||||
if (gen !== this.pendingGeneration) return;
|
||||
this.pendingStore = null;
|
||||
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;
|
||||
clearChannelSortOutbox(this.pubkey, this.relayUrl);
|
||||
}
|
||||
|
||||
publishSortPrefs(store: ChannelSortStore): void {
|
||||
@@ -304,16 +282,11 @@ 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). 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,
|
||||
);
|
||||
// survives teardown and resumes on next mount (durable outbox). This
|
||||
// window's own key is the only one written — a single unconditional
|
||||
// setItem, never a shared-key read-modify-write; `queuedAt` stamps it so
|
||||
// resume replays only the newest queued blob (whole-blob LWW).
|
||||
writeChannelSortOutbox(this.pubkey, store, this.relayUrl);
|
||||
if (this.debounceTimer !== null) {
|
||||
window.clearTimeout(this.debounceTimer);
|
||||
}
|
||||
|
||||
@@ -1,8 +1,9 @@
|
||||
import { normalizeRelayUrl } from "@/shared/lib/normalizeRelayUrl";
|
||||
import {
|
||||
clearOutboxEntry,
|
||||
readOutboxEntry,
|
||||
writeOutboxEntry,
|
||||
clearOwnOutbox,
|
||||
enumerateOutbox,
|
||||
reclaimOutbox,
|
||||
writeOwnOutbox,
|
||||
} from "./sidebarSyncWatermark";
|
||||
|
||||
const STORAGE_KEY_PREFIX = "buzz-channel-stars.v1";
|
||||
@@ -125,23 +126,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.
|
||||
* Persist the main store. Writes the passed store as-is (bounded) — no read of
|
||||
* the shared key, so it is never a shared-key read-modify-write. Callers merge
|
||||
* peer state into the window's OWN React state (via the storage-event handler
|
||||
* and applyRemote) before calling here, so the write carries an owned, merged
|
||||
* value. Returns the bounded store, or `null` on write failure.
|
||||
*
|
||||
* Cross-window convergence of the on-disk cache is eventual: a peer's storage
|
||||
* event folds into this window's state, and the relay reconcile writes the
|
||||
* merged head back. Durable no-loss of an unpublished click is held by the
|
||||
* per-window outbox, not this cache.
|
||||
*/
|
||||
export function writeChannelStarsStore(
|
||||
pubkey: string,
|
||||
incoming: ChannelStarStore,
|
||||
store: ChannelStarStore,
|
||||
preservedKey?: string,
|
||||
): ChannelStarStore | null {
|
||||
try {
|
||||
const persisted = readChannelStarsStore(pubkey);
|
||||
const merged = mergeStores(persisted, incoming, preservedKey);
|
||||
window.localStorage.setItem(storageKey(pubkey), JSON.stringify(merged));
|
||||
return merged;
|
||||
const bounded = boundStarStore(store, preservedKey);
|
||||
window.localStorage.setItem(storageKey(pubkey), JSON.stringify(bounded));
|
||||
return bounded;
|
||||
} catch {
|
||||
return null;
|
||||
}
|
||||
@@ -203,61 +207,95 @@ export function starredChannelIdsFromStore(
|
||||
|
||||
const OUTBOX_KEY_PREFIX = "buzz-channel-stars-outbox.v1";
|
||||
|
||||
// The outbox is a per-relay sync-lane structure (like the watermark), so it is
|
||||
// relay-scoped even though the main store stays pubkey-only: an edit made
|
||||
// against relay A must never resume-publish onto relay B after a community
|
||||
// switch.
|
||||
function outboxKey(pubkey: string, relayUrl: string): string {
|
||||
// The single shared key written by builds before the outbox was keyed
|
||||
// per-window. Enumerated as one more record so an edit persisted by a prior
|
||||
// build still resumes, and reclaimed by the same relay-gated rule.
|
||||
function legacyOutboxKey(pubkey: string, relayUrl: string): string {
|
||||
return `${OUTBOX_KEY_PREFIX}:${pubkey}:${encodeURIComponent(normalizeRelayUrl(relayUrl))}`;
|
||||
}
|
||||
|
||||
/**
|
||||
* Persist an unpublished edit so it survives quit/community-switch within the
|
||||
* 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.
|
||||
* Persist this window's unpublished edit under its own outbox key. Written
|
||||
* synchronously on every click as a single unconditional `setItem` (no shared-
|
||||
* key read-modify-write); resumed by merging every window's record on next
|
||||
* mount so a click made <2s before quit/community-switch is never dropped.
|
||||
*/
|
||||
export function writeChannelStarsOutbox(
|
||||
pubkey: string,
|
||||
store: ChannelStarStore,
|
||||
relayUrl: string,
|
||||
token: string,
|
||||
): void {
|
||||
const key = outboxKey(pubkey, relayUrl);
|
||||
const existing = readOutboxEntry(key, parseStarPayload)?.store;
|
||||
const merged = existing
|
||||
? mergeStores(existing, store)
|
||||
: boundStarStore(store);
|
||||
writeOutboxEntry(key, merged, token);
|
||||
writeOwnOutbox(OUTBOX_KEY_PREFIX, pubkey, relayUrl, boundStarStore(store));
|
||||
}
|
||||
|
||||
/** Read a persisted unpublished edit, or null when none/unparseable. */
|
||||
/**
|
||||
* Merge every window's persisted unpublished edit into one store for resume, or
|
||||
* null when none exists. Per-entry `mergeStores` is order-independent, so two
|
||||
* windows' concurrent clicks on different channels both survive.
|
||||
*/
|
||||
export function readChannelStarsOutbox(
|
||||
pubkey: string,
|
||||
relayUrl: string,
|
||||
): ChannelStarStore | null {
|
||||
return (
|
||||
readOutboxEntry(outboxKey(pubkey, relayUrl), parseStarPayload)?.store ??
|
||||
null
|
||||
const records = enumerateOutbox(
|
||||
OUTBOX_KEY_PREFIX,
|
||||
legacyOutboxKey(pubkey, relayUrl),
|
||||
pubkey,
|
||||
relayUrl,
|
||||
parseStarPayload,
|
||||
);
|
||||
if (records.length === 0) return null;
|
||||
return records.reduce<ChannelStarStore>(
|
||||
(acc, r) => mergeStores(acc, r.store),
|
||||
DEFAULT_STORE,
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
* 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.
|
||||
*/
|
||||
/** Clear this window's own outbox key (its edit published or is a no-op). */
|
||||
export function clearChannelStarsOutbox(
|
||||
pubkey: string,
|
||||
relayUrl: string,
|
||||
token?: string,
|
||||
): void {
|
||||
clearOutboxEntry(outboxKey(pubkey, relayUrl), token);
|
||||
clearOwnOutbox(OUTBOX_KEY_PREFIX, pubkey, relayUrl);
|
||||
}
|
||||
|
||||
/**
|
||||
* Reclaim foreign outbox keys the fetched relay head already subsumes: a record
|
||||
* is redundant when merging it into `head` yields `head` unchanged (the head
|
||||
* carries an entry at least as new for every channel). Never touches this
|
||||
* window's own key; a still-unpublished peer edit the head does not yet reflect
|
||||
* is kept. Call only after a successful head fetch.
|
||||
*/
|
||||
export function reclaimSubsumedStarsOutbox(
|
||||
pubkey: string,
|
||||
relayUrl: string,
|
||||
head: ChannelStarStore,
|
||||
): void {
|
||||
reclaimOutbox(
|
||||
OUTBOX_KEY_PREFIX,
|
||||
legacyOutboxKey(pubkey, relayUrl),
|
||||
pubkey,
|
||||
relayUrl,
|
||||
parseStarPayload,
|
||||
(record) => starStoresEqual(mergeStores(head, record.store), head),
|
||||
);
|
||||
}
|
||||
|
||||
/** Deep per-channel equality of two star stores (order-independent). */
|
||||
function starStoresEqual(a: ChannelStarStore, b: ChannelStarStore): boolean {
|
||||
const aKeys = Object.keys(a.channels);
|
||||
const bKeys = Object.keys(b.channels);
|
||||
if (aKeys.length !== bKeys.length) return false;
|
||||
for (const id of aKeys) {
|
||||
const l = a.channels[id];
|
||||
const r = b.channels[id];
|
||||
if (
|
||||
!r ||
|
||||
l.starred !== r.starred ||
|
||||
l.updatedAt !== r.updatedAt ||
|
||||
l.rev !== r.rev
|
||||
)
|
||||
return false;
|
||||
}
|
||||
return true;
|
||||
}
|
||||
|
||||
@@ -2,6 +2,7 @@ import assert from "node:assert/strict";
|
||||
import test, { mock } from "node:test";
|
||||
|
||||
import { relayClient } from "@/shared/api/relayClient";
|
||||
import { readChannelStarsOutbox } from "./channelStarsStorage.ts";
|
||||
import { ChannelStarSyncManager } from "./channelStarsSync.ts";
|
||||
import {
|
||||
installFakeWindow,
|
||||
@@ -28,6 +29,10 @@ function makeMultiTimerWindow() {
|
||||
getItem: (k) => storage.get(k) ?? null,
|
||||
setItem: (k, v) => storage.set(k, v),
|
||||
removeItem: (k) => storage.delete(k),
|
||||
get length() {
|
||||
return storage.size;
|
||||
},
|
||||
key: (i) => [...storage.keys()][i] ?? null,
|
||||
},
|
||||
setTimeout: (fn, ms) => {
|
||||
const id = nextId++;
|
||||
@@ -167,7 +172,6 @@ test("A-in-flight → B-click → A-succeeds: B stays pending and B publishes",
|
||||
const t = makeMultiTimerWindow();
|
||||
const restore = installFakeWindow(t.win);
|
||||
const tauri = installTauriMock("{}");
|
||||
const outboxKey = `buzz-channel-stars-outbox.v1:pk-ab:${RELAY_KEY}`;
|
||||
try {
|
||||
const manager = new ChannelStarSyncManager("pk-ab", RELAY);
|
||||
const storeA = makeStore({ a: E(true, 100, 1) });
|
||||
@@ -184,7 +188,7 @@ test("A-in-flight → B-click → A-succeeds: B stays pending and B publishes",
|
||||
["b"],
|
||||
"B is now pending",
|
||||
);
|
||||
assert.ok(t.storage.get(outboxKey), "outbox holds B");
|
||||
assert.ok(readChannelStarsOutbox("pk-ab", RELAY), "outbox holds B");
|
||||
|
||||
// A completes — must NOT clear B.
|
||||
releaseFirst();
|
||||
@@ -194,7 +198,10 @@ test("A-in-flight → B-click → A-succeeds: B stays pending and B publishes",
|
||||
["b"],
|
||||
"older A completion leaves B pending",
|
||||
);
|
||||
assert.ok(t.storage.get(outboxKey), "older A completion leaves B outbox");
|
||||
assert.ok(
|
||||
readChannelStarsOutbox("pk-ab", RELAY),
|
||||
"older A completion leaves B outbox",
|
||||
);
|
||||
|
||||
// B's own debounce fires and B reaches the relay (published) with no kick.
|
||||
const capturedBefore = tauri.capturedPlaintext();
|
||||
@@ -210,7 +217,11 @@ test("A-in-flight → B-click → A-succeeds: B stays pending and B publishes",
|
||||
null,
|
||||
"B cleared after publish",
|
||||
);
|
||||
assert.equal(t.storage.get(outboxKey), undefined, "B outbox cleared");
|
||||
assert.equal(
|
||||
readChannelStarsOutbox("pk-ab", RELAY),
|
||||
null,
|
||||
"B outbox cleared",
|
||||
);
|
||||
manager.destroy();
|
||||
} finally {
|
||||
tauri.restore();
|
||||
@@ -237,7 +248,6 @@ test("A-in-flight → B-click → A-fails: B remains pending and B publishes", a
|
||||
const t = makeMultiTimerWindow();
|
||||
const restore = installFakeWindow(t.win);
|
||||
const tauri = installTauriMock("{}");
|
||||
const outboxKey = `buzz-channel-stars-outbox.v1:pk-abfail:${RELAY_KEY}`;
|
||||
try {
|
||||
const manager = new ChannelStarSyncManager("pk-abfail", RELAY);
|
||||
manager.publishStars(makeStore({ a: E(true, 100, 1) }));
|
||||
@@ -253,7 +263,10 @@ test("A-in-flight → B-click → A-fails: B remains pending and B publishes", a
|
||||
["b"],
|
||||
"B still pending after A's failure",
|
||||
);
|
||||
assert.ok(t.storage.get(outboxKey), "B outbox intact after A's failure");
|
||||
assert.ok(
|
||||
readChannelStarsOutbox("pk-abfail", RELAY),
|
||||
"B outbox intact after A's failure",
|
||||
);
|
||||
|
||||
// B's debounce fires and B publishes successfully.
|
||||
await t.fireDelay(2000);
|
||||
|
||||
@@ -16,7 +16,6 @@ import {
|
||||
import {
|
||||
advanceWatermark,
|
||||
clampPublishCreatedAt,
|
||||
mintOutboxToken,
|
||||
readWatermark,
|
||||
runBootstrap,
|
||||
type FetchResult,
|
||||
@@ -63,13 +62,6 @@ 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
|
||||
@@ -169,16 +161,10 @@ 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). 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,
|
||||
);
|
||||
// survives teardown and resumes on next mount (durable outbox). This
|
||||
// window's own key is the only one written — a single unconditional
|
||||
// setItem, never a shared-key read-modify-write.
|
||||
writeChannelStarsOutbox(this.pubkey, store, this.relayUrl);
|
||||
if (this.debounceTimer !== null) {
|
||||
window.clearTimeout(this.debounceTimer);
|
||||
}
|
||||
@@ -265,30 +251,15 @@ export class ChannelStarSyncManager {
|
||||
}
|
||||
|
||||
/**
|
||||
* Clear the in-memory pending edit and its durable outbox — but only if the
|
||||
* completing publish still owns the current generation. A publish for an
|
||||
* older edit that finishes after a newer edit was queued must leave the newer
|
||||
* edit (and its retry state) untouched.
|
||||
* Clear the in-memory pending edit and this window's own durable outbox key —
|
||||
* but only if the completing publish still owns the current generation. A
|
||||
* publish for an older edit that finishes after a newer edit was queued must
|
||||
* leave the newer edit (and its retry state) untouched.
|
||||
*/
|
||||
private discardPending(gen: number): void {
|
||||
if (gen !== this.pendingGeneration) return;
|
||||
this.pendingStore = null;
|
||||
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;
|
||||
clearChannelStarsOutbox(this.pubkey, this.relayUrl);
|
||||
}
|
||||
|
||||
/** Schedule a bounded-backoff retry of the retained pending edit. */
|
||||
|
||||
@@ -3,43 +3,88 @@ 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).
|
||||
// All four sidebar-sync lanes persist an unpublished edit so it survives a
|
||||
// quit/community-switch inside the 2s publish debounce. localStorage offers no
|
||||
// atomic compare-and-delete or transactional read-modify-write, so a single key
|
||||
// shared across windows can never be mutated safely: one window's read→write or
|
||||
// read→remove races a peer's write and drops its still-unpublished edit.
|
||||
//
|
||||
// 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.
|
||||
// The fix keys the outbox PER WINDOW: `<prefix>:<pubkey>:<relay>:<nonce>`, where
|
||||
// the nonce is stable per window (sessionStorage). Each window is the sole
|
||||
// writer of its own key, so a hot-path write is one unconditional `setItem` — no
|
||||
// read, no shared-key contention (prong b is designed out, not guarded). Resume
|
||||
// enumerates ALL windows' keys: merge lanes (stars/mutes) fold every record
|
||||
// (order-independent); whole-blob lanes (sort/sections) replay the max-`queuedAt`
|
||||
// record, ties broken by key. Redundant FOREIGN keys are reclaimed at boot,
|
||||
// gated on durable relay evidence and re-read immediately before removal so a
|
||||
// live peer's fresh write in the recheck gap is never destroyed. A window never
|
||||
// removes another window's key on the hot path and never touches its own key
|
||||
// during reclamation.
|
||||
//
|
||||
// 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.
|
||||
// This matrix drives the shared helpers directly through a mock Storage seam —
|
||||
// no relay, no timers — so each interleaving is deterministic. The seam mocks
|
||||
// both localStorage (with a read-then-mutate hook to interpose a foreign write
|
||||
// between the reclaim decision-read and the delete) and sessionStorage (for the
|
||||
// per-window nonce). Manager- and hook-level behavior is covered by their own
|
||||
// suites; this file isolates the cross-window storage contract.
|
||||
|
||||
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(),
|
||||
// A mock Storage. `onReadMutate(key, afterReads, fn)` runs `fn(map)` right after
|
||||
// the Nth getItem of `key` returns its captured value — used to simulate a peer
|
||||
// window rewriting a key in the reclaim recheck gap.
|
||||
function makeStorage(initial = {}) {
|
||||
const map = new Map(Object.entries(initial));
|
||||
const reads = new Map();
|
||||
const hooks = [];
|
||||
return {
|
||||
getItem(k) {
|
||||
const n = (reads.get(k) ?? 0) + 1;
|
||||
reads.set(k, n);
|
||||
const val = map.has(k) ? map.get(k) : null;
|
||||
for (const h of hooks) if (h.key === k && h.afterReads === n) h.fn(map);
|
||||
return val;
|
||||
},
|
||||
setItem(k, v) {
|
||||
map.set(k, String(v));
|
||||
},
|
||||
removeItem(k) {
|
||||
map.delete(k);
|
||||
},
|
||||
clear() {
|
||||
map.clear();
|
||||
},
|
||||
get length() {
|
||||
return map.size;
|
||||
},
|
||||
key(i) {
|
||||
return [...map.keys()][i] ?? null;
|
||||
},
|
||||
onReadMutate(key, afterReads, fn) {
|
||||
hooks.push({ key, afterReads, fn });
|
||||
},
|
||||
has: (k) => map.has(k),
|
||||
};
|
||||
}
|
||||
|
||||
// Run `fn(localStorage)` with fresh mock local + session storage installed.
|
||||
function withStorage(fn) {
|
||||
const ls = makeStorage();
|
||||
const ss = makeStorage();
|
||||
const priorWindow = globalThis.window;
|
||||
globalThis.window = {
|
||||
...(priorWindow ?? {}),
|
||||
localStorage: ls,
|
||||
sessionStorage: ss,
|
||||
};
|
||||
const orig = globalThis.window?.localStorage;
|
||||
if (typeof globalThis.window === "undefined") globalThis.window = {};
|
||||
globalThis.window.localStorage = ls;
|
||||
try {
|
||||
fn(ls);
|
||||
return fn(ls);
|
||||
} finally {
|
||||
if (orig !== undefined) globalThis.window.localStorage = orig;
|
||||
else delete globalThis.window.localStorage;
|
||||
if (priorWindow !== undefined) globalThis.window = priorWindow;
|
||||
else delete globalThis.window;
|
||||
}
|
||||
}
|
||||
|
||||
const { mintOutboxToken } = await import("./sidebarSyncWatermark.ts");
|
||||
const { normalizeRelayUrl } = await import("@/shared/lib/normalizeRelayUrl");
|
||||
const { outboxWindowNonce } = await import("./sidebarSyncWatermark.ts");
|
||||
const stars = await import("./channelStarsStorage.ts");
|
||||
const mutes = await import("./channelMutesStorage.ts");
|
||||
const sort = await import("./channelSortPreference.ts");
|
||||
@@ -47,6 +92,21 @@ const sections = await import("./channelSectionsStorage.ts");
|
||||
|
||||
const PK = "pk";
|
||||
const RELAY = "wss://relay.example.com";
|
||||
const SCOPE = `${PK}:${encodeURIComponent(normalizeRelayUrl(RELAY))}`;
|
||||
|
||||
const PREFIX = {
|
||||
stars: "buzz-channel-stars-outbox.v1",
|
||||
mutes: "buzz-channel-mutes-outbox.v1",
|
||||
sort: "buzz-channel-sort-outbox.v1",
|
||||
sections: "buzz-channel-sections-outbox.v1",
|
||||
};
|
||||
|
||||
// A foreign window's key: same (pubkey, relay) scope, a different nonce than
|
||||
// this process's own. `legacyKey` is the pre-per-window shared key (no nonce).
|
||||
const foreignKey = (lane, nonce) => `${PREFIX[lane]}:${SCOPE}:${nonce}`;
|
||||
const legacyKey = (lane) => `${PREFIX[lane]}:${SCOPE}`;
|
||||
const writeAt = (ls, key, store, queuedAt) =>
|
||||
ls.setItem(key, JSON.stringify({ store, queuedAt }));
|
||||
|
||||
const starStore = (channels) => ({ version: 1, channels });
|
||||
const starEntry = (starred, updatedAt, rev) => ({ starred, updatedAt, rev });
|
||||
@@ -59,240 +119,99 @@ const sectionStore = (secs, assignments = {}) => ({
|
||||
assignments,
|
||||
});
|
||||
|
||||
// ── (a) stale-clear: A queues → B queues → A completes and clears → B survives ─
|
||||
// ── (i) Recheck skips a foreign write injected between decision-read and delete ─
|
||||
//
|
||||
// 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.
|
||||
// reclaimOutbox re-reads each foreign key immediately before removing it and
|
||||
// skips when the value changed since the reclaim decision. A peer that rewrote
|
||||
// its key in that gap owns a fresh edit that must survive. Merge lane (stars):
|
||||
// the head subsumes the enumerated value, so reclaim would fire — but the peer's
|
||||
// interposed write must be preserved.
|
||||
|
||||
test("(a) stars: A's completing clear does not erase B's later queued edit", () => {
|
||||
withFreshStorage(() => {
|
||||
const tokenA = mintOutboxToken();
|
||||
stars.writeChannelStarsOutbox(
|
||||
test("(i) stars: reclaim recheck skips a foreign key rewritten in the decision→delete gap", () => {
|
||||
withStorage((ls) => {
|
||||
const key = foreignKey("stars", "peerB");
|
||||
// Foreign edit the fetched head has already absorbed → decision = reclaim.
|
||||
writeAt(ls, key, starStore({ a: starEntry(true, 100, 1) }), 100);
|
||||
// Between the enumerate read (#1) and the recheck read (#2), the peer writes
|
||||
// a FRESH edit the head does not reflect.
|
||||
ls.onReadMutate(key, 1, (map) =>
|
||||
map.set(
|
||||
key,
|
||||
JSON.stringify({
|
||||
store: starStore({ a: starEntry(false, 300, 9) }),
|
||||
queuedAt: 300,
|
||||
}),
|
||||
),
|
||||
);
|
||||
// Head carries `a` at (100,1) — subsumes the enumerated value.
|
||||
stars.reclaimSubsumedStarsOutbox(
|
||||
PK,
|
||||
RELAY,
|
||||
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);
|
||||
assert.ok(ls.has(key), "peer's fresh edit must survive the recheck");
|
||||
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);
|
||||
assert.deepEqual(survived.channels.a, starEntry(false, 300, 9));
|
||||
});
|
||||
});
|
||||
|
||||
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,
|
||||
test("(i) sort: reclaim recheck skips a foreign key rewritten in the decision→delete gap", () => {
|
||||
withStorage((ls) => {
|
||||
const key = foreignKey("sort", "peerB");
|
||||
writeAt(ls, key, sortStore({ dms: "alpha" }), 100);
|
||||
ls.onReadMutate(key, 1, (map) =>
|
||||
map.set(
|
||||
key,
|
||||
JSON.stringify({ store: sortStore({ dms: "recent" }), queuedAt: 500 }),
|
||||
),
|
||||
);
|
||||
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);
|
||||
// Head created_at 200 supersedes the queuedAt=100 decision, but not the
|
||||
// interposed queuedAt=500 write.
|
||||
sort.reclaimSupersededSortOutbox(PK, RELAY, 200);
|
||||
assert.ok(ls.has(key), "peer's fresh edit must survive the recheck");
|
||||
assert.equal(sort.readChannelSortOutbox(PK, RELAY).groups.dms, "recent");
|
||||
});
|
||||
});
|
||||
|
||||
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);
|
||||
});
|
||||
});
|
||||
// ── (ii) Two windows teardown/remount: every unpublished intent preserved ──────
|
||||
//
|
||||
// Windows A and B each persist an edit and quit; a fresh window remounts and
|
||||
// enumerates both keys. Merge lanes keep BOTH; whole-blob lanes keep the newest
|
||||
// (an older peer blob is LWW-superseded by definition, the documented residual).
|
||||
|
||||
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,
|
||||
test("(ii) stars (merge): both windows' distinct-channel edits resume", () => {
|
||||
withStorage((ls) => {
|
||||
writeAt(
|
||||
ls,
|
||||
foreignKey("stars", "A"),
|
||||
starStore({ a: starEntry(true, 100, 1) }),
|
||||
RELAY,
|
||||
mintOutboxToken(),
|
||||
100,
|
||||
);
|
||||
// Window B does not read A first; its write must fold, not clobber.
|
||||
stars.writeChannelStarsOutbox(
|
||||
PK,
|
||||
writeAt(
|
||||
ls,
|
||||
foreignKey("stars", "B"),
|
||||
starStore({ b: starEntry(true, 200, 1) }),
|
||||
RELAY,
|
||||
mintOutboxToken(),
|
||||
200,
|
||||
);
|
||||
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,
|
||||
test("(ii) mutes (merge): both windows' distinct-channel edits resume", () => {
|
||||
withStorage((ls) => {
|
||||
writeAt(
|
||||
ls,
|
||||
foreignKey("mutes", "A"),
|
||||
muteStore({ a: muteEntry(true, 100, 1) }),
|
||||
RELAY,
|
||||
mintOutboxToken(),
|
||||
100,
|
||||
);
|
||||
mutes.writeChannelMutesOutbox(
|
||||
PK,
|
||||
writeAt(
|
||||
ls,
|
||||
foreignKey("mutes", "B"),
|
||||
muteStore({ b: muteEntry(true, 200, 1) }),
|
||||
RELAY,
|
||||
mintOutboxToken(),
|
||||
200,
|
||||
);
|
||||
const resumed = mutes.readChannelMutesOutbox(PK, RELAY);
|
||||
assert.deepEqual(resumed.channels.a, muteEntry(true, 100, 1));
|
||||
@@ -300,105 +219,202 @@ test("(c) mutes: an edit persisted before teardown resumes merged with a peer's
|
||||
});
|
||||
});
|
||||
|
||||
// ── (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,
|
||||
test("(ii) sort (whole-blob): the newest queued window resumes; older is LWW-superseded", () => {
|
||||
withStorage((ls) => {
|
||||
writeAt(ls, foreignKey("sort", "A"), sortStore({ channels: "alpha" }), 100);
|
||||
writeAt(
|
||||
ls,
|
||||
foreignKey("sort", "B"),
|
||||
sortStore({ channels: "recent" }),
|
||||
200,
|
||||
);
|
||||
assert.equal(
|
||||
sort.readChannelSortOutbox(PK, RELAY).groups.channels,
|
||||
"recent",
|
||||
);
|
||||
// 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,
|
||||
test("(ii) sections (whole-blob): the newest queued window resumes; older is LWW-superseded", () => {
|
||||
withStorage((ls) => {
|
||||
writeAt(
|
||||
ls,
|
||||
foreignKey("sections", "A"),
|
||||
sectionStore([{ id: "s1", name: "One", order: 0 }]),
|
||||
RELAY,
|
||||
tokenA,
|
||||
100,
|
||||
);
|
||||
const tokenB = mintOutboxToken();
|
||||
sections.writeChannelSectionsOutbox(
|
||||
PK,
|
||||
sectionStore([{ id: "s1", name: "Renamed", order: 0 }]),
|
||||
RELAY,
|
||||
tokenB,
|
||||
writeAt(
|
||||
ls,
|
||||
foreignKey("sections", "B"),
|
||||
sectionStore([{ id: "s2", name: "Two", order: 0 }]),
|
||||
200,
|
||||
);
|
||||
sections.clearChannelSectionsOutbox(PK, RELAY, tokenA);
|
||||
assert.deepEqual(sections.readChannelSectionsOutbox(PK, RELAY).sections, [
|
||||
{ id: "s1", name: "Renamed", order: 0 },
|
||||
{ id: "s2", name: "Two", 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.
|
||||
// ── (iii) Legacy v1 shared key: replays, then reclaimed only by relay gating ───
|
||||
//
|
||||
// A pre-per-window build wrote one shared key. It enumerates as one more record
|
||||
// (queuedAt 0 for a bare store), resumes, and is reclaimed by the same relay-
|
||||
// gated rule — never dropped on replay alone (mixed dev/DMG fleet residual).
|
||||
|
||||
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) })));
|
||||
test("(iii) stars: legacy shared key resumes, then is reclaimed once the head subsumes it", () => {
|
||||
withStorage((ls) => {
|
||||
// Legacy bare store (no envelope) from a pre-per-window build.
|
||||
ls.setItem(
|
||||
legacyKey("stars"),
|
||||
JSON.stringify(starStore({ a: starEntry(true, 100, 1) })),
|
||||
);
|
||||
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);
|
||||
assert.deepEqual(
|
||||
resumed.channels.a,
|
||||
starEntry(true, 100, 1),
|
||||
"legacy entry resumes",
|
||||
);
|
||||
// A head that does NOT subsume it (older rev) keeps it.
|
||||
stars.reclaimSubsumedStarsOutbox(
|
||||
PK,
|
||||
RELAY,
|
||||
starStore({ a: starEntry(true, 50, 0) }),
|
||||
);
|
||||
assert.ok(ls.has(legacyKey("stars")), "unsubsumed legacy entry is kept");
|
||||
// A head that subsumes it reclaims it.
|
||||
stars.reclaimSubsumedStarsOutbox(
|
||||
PK,
|
||||
RELAY,
|
||||
starStore({ a: starEntry(true, 100, 1) }),
|
||||
);
|
||||
assert.ok(
|
||||
!ls.has(legacyKey("stars")),
|
||||
"subsumed legacy entry is reclaimed",
|
||||
);
|
||||
});
|
||||
});
|
||||
|
||||
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);
|
||||
test("(iii) sections: legacy shared key resumes, then is reclaimed once the head supersedes it", () => {
|
||||
withStorage((ls) => {
|
||||
// Legacy entry as a {store, queuedAt} envelope from an interim build.
|
||||
writeAt(
|
||||
ls,
|
||||
legacyKey("sections"),
|
||||
sectionStore([{ id: "s1", name: "One", order: 0 }]),
|
||||
100,
|
||||
);
|
||||
assert.deepEqual(sections.readChannelSectionsOutbox(PK, RELAY).sections, [
|
||||
{ id: "s1", name: "One", order: 0 },
|
||||
]);
|
||||
// Head created_at before the queued stamp keeps it (not yet superseded).
|
||||
sections.reclaimSupersededSectionsOutbox(PK, RELAY, 50);
|
||||
assert.ok(
|
||||
ls.has(legacyKey("sections")),
|
||||
"un-superseded legacy entry is kept",
|
||||
);
|
||||
// Head created_at at/after the queued stamp supersedes and reclaims it.
|
||||
sections.reclaimSupersededSectionsOutbox(PK, RELAY, 100);
|
||||
assert.ok(
|
||||
!ls.has(legacyKey("sections")),
|
||||
"superseded legacy entry is reclaimed",
|
||||
);
|
||||
});
|
||||
});
|
||||
|
||||
// ── Token-envelope round-trip + omitted-token unconditional clear ──────────────
|
||||
// ── (iv) Whole-blob replay tie → deterministic nonce (key) tiebreak ────────────
|
||||
|
||||
test("write/read/clear: envelope round-trips and an omitted token clears unconditionally", () => {
|
||||
withFreshStorage(() => {
|
||||
const token = mintOutboxToken();
|
||||
test("(iv) sort: equal-queuedAt records resolve by key so replay is deterministic", () => {
|
||||
withStorage((ls) => {
|
||||
writeAt(ls, foreignKey("sort", "aaa"), sortStore({ forums: "alpha" }), 100);
|
||||
writeAt(
|
||||
ls,
|
||||
foreignKey("sort", "zzz"),
|
||||
sortStore({ forums: "recent" }),
|
||||
100,
|
||||
);
|
||||
// Same queuedAt → the lexicographically-greater key wins (…:zzz).
|
||||
assert.equal(sort.readChannelSortOutbox(PK, RELAY).groups.forums, "recent");
|
||||
});
|
||||
});
|
||||
|
||||
// ── (v) Merge-lane replay is order-independent ─────────────────────────────────
|
||||
|
||||
test("(v) stars: same-channel records fold to the max entry regardless of key order", () => {
|
||||
withStorage((ls) => {
|
||||
// Lower-rev record under a lexicographically-greater key (enumerated later)
|
||||
// must still lose to the higher-rev record — merge is order-independent.
|
||||
writeAt(
|
||||
ls,
|
||||
foreignKey("stars", "aaa"),
|
||||
starStore({ c: starEntry(true, 100, 5) }),
|
||||
100,
|
||||
);
|
||||
writeAt(
|
||||
ls,
|
||||
foreignKey("stars", "zzz"),
|
||||
starStore({ c: starEntry(false, 100, 2) }),
|
||||
200,
|
||||
);
|
||||
const merged = stars.readChannelStarsOutbox(PK, RELAY);
|
||||
assert.deepEqual(
|
||||
merged.channels.c,
|
||||
starEntry(true, 100, 5),
|
||||
"higher rev wins the tie",
|
||||
);
|
||||
});
|
||||
});
|
||||
|
||||
// ── (vi) GC no-op when the head subsumes/supersedes nothing ────────────────────
|
||||
//
|
||||
// The hook calls reclaim only inside the `apply-remote` branch, so a `failed`
|
||||
// head fetch (or `absent`) never invokes it — that guard is structural in the
|
||||
// hook. At the storage layer the matching invariant is that a head which
|
||||
// subsumes/supersedes nothing removes nothing: a foreign edit newer than the
|
||||
// head is live intent and is kept, so a stale/empty head can never over-collect.
|
||||
|
||||
test("(vi) stars: a head that subsumes nothing reclaims nothing", () => {
|
||||
withStorage((ls) => {
|
||||
const key = foreignKey("stars", "B");
|
||||
writeAt(ls, key, starStore({ a: starEntry(true, 300, 2) }), 300);
|
||||
// Empty head subsumes no channel → keep everything.
|
||||
stars.reclaimSubsumedStarsOutbox(PK, RELAY, starStore({}));
|
||||
assert.ok(ls.has(key), "unsubsumed foreign edit is kept");
|
||||
});
|
||||
});
|
||||
|
||||
test("(vi) sort: a head older than the queued edit supersedes nothing", () => {
|
||||
withStorage((ls) => {
|
||||
const key = foreignKey("sort", "B");
|
||||
writeAt(ls, key, sortStore({ dms: "recent" }), 300);
|
||||
// headCreatedAt=0 (absent-equivalent) < queuedAt → keep.
|
||||
sort.reclaimSupersededSortOutbox(PK, RELAY, 0);
|
||||
assert.ok(ls.has(key), "edit queued after the head is kept");
|
||||
});
|
||||
});
|
||||
|
||||
// ── Own-key round trip: write, read, clear (single-window baseline) ────────────
|
||||
|
||||
test("own key: write resumes, clear removes only this window's own key", () => {
|
||||
withStorage((ls) => {
|
||||
stars.writeChannelStarsOutbox(
|
||||
PK,
|
||||
starStore({ a: starEntry(true, 100, 1) }),
|
||||
RELAY,
|
||||
token,
|
||||
);
|
||||
assert.deepEqual(
|
||||
stars.readChannelStarsOutbox(PK, RELAY).channels.a,
|
||||
starEntry(true, 100, 1),
|
||||
const ownKey = `${PREFIX.stars}:${SCOPE}:${outboxWindowNonce()}`;
|
||||
assert.ok(ls.has(ownKey), "own edit is written under this window's nonce");
|
||||
// A foreign peer key is untouched by an own-key clear.
|
||||
writeAt(
|
||||
ls,
|
||||
foreignKey("stars", "peer"),
|
||||
starStore({ z: starEntry(true, 9, 1) }),
|
||||
9,
|
||||
);
|
||||
// Omitted token = unconditional clear (bootstrap "nothing to resume" path).
|
||||
stars.clearChannelStarsOutbox(PK, RELAY);
|
||||
assert.equal(stars.readChannelStarsOutbox(PK, RELAY), null);
|
||||
assert.ok(!ls.has(ownKey), "own key cleared");
|
||||
assert.ok(
|
||||
ls.has(foreignKey("stars", "peer")),
|
||||
"foreign key untouched by own clear",
|
||||
);
|
||||
});
|
||||
});
|
||||
|
||||
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");
|
||||
});
|
||||
|
||||
@@ -7,6 +7,10 @@ export function makeFakeWindow() {
|
||||
setItem: (k, v) => storage.set(k, v),
|
||||
removeItem: (k) => storage.delete(k),
|
||||
clear: () => storage.clear(),
|
||||
get length() {
|
||||
return storage.size;
|
||||
},
|
||||
key: (i) => [...storage.keys()][i] ?? null,
|
||||
};
|
||||
let timerCallback = null;
|
||||
let nextTimerId = 100;
|
||||
|
||||
@@ -52,44 +52,121 @@ 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;
|
||||
// ── Per-window durable outbox ────────────────────────────────────────────────
|
||||
//
|
||||
// Each sidebar-sync lane persists an unpublished edit so it survives a
|
||||
// quit/community-switch inside the 2s publish debounce. The durability boundary
|
||||
// is localStorage, which offers no atomic compare-and-delete or transactional
|
||||
// read-modify-write, so a single key shared across every window can never be
|
||||
// mutated safely: one window's read→write or read→remove can race a peer's
|
||||
// write in the gap and drop its still-unpublished edit (the defect a per-write
|
||||
// ownership token could narrow but not close).
|
||||
//
|
||||
// The outbox is therefore keyed PER WINDOW:
|
||||
// `<prefix>:<pubkey>:<relay>:<nonce>`, where `nonce` is stable for one window's
|
||||
// lifetime (parked in sessionStorage — survives reload, gone on window close).
|
||||
// Each window is the ONLY writer of its own key, so a hot-path write is a single
|
||||
// unconditional `setItem` with no read and no shared-key contention: the write
|
||||
// race is designed out rather than guarded.
|
||||
//
|
||||
// A window clears only its own key once its publish completes. Redundant foreign
|
||||
// keys (a peer that published then crashed, or an edit the relay has since
|
||||
// absorbed) are reclaimed at boot, gated on durable relay evidence so a live
|
||||
// peer's genuinely-unpublished edit is never destroyed:
|
||||
// - merge lanes: delete a record the fetched relay head already subsumes.
|
||||
// - whole-blob lanes: delete a record the head's own timestamp supersedes
|
||||
// (`queuedAt` ≤ head `created_at`) — never one that
|
||||
// merely loses a local comparison.
|
||||
// Reclamation runs only when the head fetch succeeded, never on a failed fetch,
|
||||
// and re-reads each foreign key immediately before removing it so a value that
|
||||
// changed since the decision is left for its owner. This is not atomic —
|
||||
// localStorage cannot be — but it narrows the residual to a recheck→remove gap
|
||||
// that additionally requires the owner to quit before its own debounce publish
|
||||
// (its in-memory pending re-drives otherwise). After any successful publish the
|
||||
// relay head is newest, so foreign keys still drain to zero across boots.
|
||||
|
||||
/**
|
||||
* 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 OUTBOX_WINDOW_NONCE_KEY = "buzz-sidebar-outbox-window.v1";
|
||||
|
||||
// Monotonic fallback counter so two mints in the same millisecond (or in an
|
||||
// environment without `crypto.randomUUID`) still differ.
|
||||
let nonceCounter = 0;
|
||||
let cachedWindowNonce: string | null = null;
|
||||
|
||||
function mintNonce(): string {
|
||||
const uuid = globalThis.crypto?.randomUUID?.();
|
||||
if (uuid) return uuid;
|
||||
return `${Date.now().toString(36)}-${(tokenCounter++).toString(36)}-${Math.random().toString(36).slice(2)}`;
|
||||
return `${Date.now().toString(36)}-${(nonceCounter++).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.
|
||||
* The stable per-window nonce that scopes this window's outbox keys. Minted
|
||||
* once and parked in sessionStorage so a reload re-owns the same key while a
|
||||
* new window gets its own; an unavailable sessionStorage (private mode, test
|
||||
* harness without one) falls back to a process-lifetime in-memory nonce.
|
||||
*/
|
||||
export function readOutboxEntry<T>(
|
||||
key: string,
|
||||
parseStore: (json: unknown) => T | null,
|
||||
): { store: T; token: string | null } | null {
|
||||
export function outboxWindowNonce(): string {
|
||||
if (cachedWindowNonce !== null) return cachedWindowNonce;
|
||||
try {
|
||||
const existing = window.sessionStorage.getItem(OUTBOX_WINDOW_NONCE_KEY);
|
||||
if (existing) {
|
||||
cachedWindowNonce = existing;
|
||||
return existing;
|
||||
}
|
||||
const nonce = mintNonce();
|
||||
window.sessionStorage.setItem(OUTBOX_WINDOW_NONCE_KEY, nonce);
|
||||
cachedWindowNonce = nonce;
|
||||
return nonce;
|
||||
} catch {
|
||||
cachedWindowNonce = mintNonce();
|
||||
return cachedWindowNonce;
|
||||
}
|
||||
}
|
||||
|
||||
function scopeSuffix(pubkey: string, relayUrl: string): string {
|
||||
return `${pubkey}:${encodeURIComponent(normalizeRelayUrl(relayUrl))}`;
|
||||
}
|
||||
|
||||
/** This window's own outbox key — the only key it ever writes or removes. */
|
||||
export function ownOutboxKey(
|
||||
prefix: string,
|
||||
pubkey: string,
|
||||
relayUrl: string,
|
||||
): string {
|
||||
return `${prefix}:${scopeSuffix(pubkey, relayUrl)}:${outboxWindowNonce()}`;
|
||||
}
|
||||
|
||||
/** Prefix every window's key for this (lane, pubkey, relay) starts with. */
|
||||
function outboxScopePrefix(
|
||||
prefix: string,
|
||||
pubkey: string,
|
||||
relayUrl: string,
|
||||
): string {
|
||||
return `${prefix}:${scopeSuffix(pubkey, relayUrl)}:`;
|
||||
}
|
||||
|
||||
/** A durable outbox record enumerated across all windows for a lane. */
|
||||
export type OutboxRecord<T> = {
|
||||
key: string;
|
||||
store: T;
|
||||
// Seconds since epoch when the edit was queued (0 for a legacy entry written
|
||||
// before per-window keys, which therefore never wins a whole-blob tie).
|
||||
queuedAt: number;
|
||||
// The exact stored string at enumeration time, for the pre-delete recheck.
|
||||
raw: string;
|
||||
isOwn: boolean;
|
||||
};
|
||||
|
||||
/**
|
||||
* Parse a stored outbox value, tolerating the per-window envelope
|
||||
* (`{ store, queuedAt }`), the previous token envelope (`{ store, token }` ⇒
|
||||
* `queuedAt` 0), and a bare store from an even older build (⇒ `queuedAt` 0).
|
||||
*/
|
||||
function parseEnvelope<T>(
|
||||
raw: string | null,
|
||||
parseStore: (json: unknown) => T | null,
|
||||
): { store: T; queuedAt: number } | null {
|
||||
if (!raw) return null;
|
||||
try {
|
||||
const raw = window.localStorage.getItem(key);
|
||||
if (!raw) return null;
|
||||
const json = JSON.parse(raw);
|
||||
if (
|
||||
json !== null &&
|
||||
@@ -97,73 +174,185 @@ export function readOutboxEntry<T>(
|
||||
!Array.isArray(json) &&
|
||||
"store" in (json as Record<string, unknown>)
|
||||
) {
|
||||
const env = json as { store: unknown; token?: unknown };
|
||||
const env = json as { store: unknown; queuedAt?: unknown };
|
||||
const store = parseStore(env.store);
|
||||
if (!store) return null;
|
||||
return {
|
||||
store,
|
||||
token: typeof env.token === "string" ? env.token : null,
|
||||
};
|
||||
const queuedAt =
|
||||
typeof env.queuedAt === "number" && Number.isFinite(env.queuedAt)
|
||||
? env.queuedAt
|
||||
: 0;
|
||||
return { store, queuedAt };
|
||||
}
|
||||
// Legacy bare-store shape from a pre-token build. Token absent ⇒ clearable.
|
||||
// Legacy bare-store shape from a pre-envelope build.
|
||||
const store = parseStore(json);
|
||||
if (!store) return null;
|
||||
return { store, token: null };
|
||||
return { store, queuedAt: 0 };
|
||||
} catch {
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
/** Persist a durable outbox entry as a token envelope. Best-effort. */
|
||||
export function writeOutboxEntry(
|
||||
key: string,
|
||||
/**
|
||||
* Persist this window's unpublished edit under its own key. A single
|
||||
* unconditional `setItem` — no read, no merge, no shared-key contention.
|
||||
* Best-effort: the in-memory pending edit still drives this session's publish
|
||||
* even if the persisted copy could not be written.
|
||||
*/
|
||||
export function writeOwnOutbox(
|
||||
prefix: string,
|
||||
pubkey: string,
|
||||
relayUrl: string,
|
||||
store: unknown,
|
||||
token: string,
|
||||
nowSecs: number = Math.floor(Date.now() / 1_000),
|
||||
): void {
|
||||
try {
|
||||
window.localStorage.setItem(key, JSON.stringify({ store, token }));
|
||||
window.localStorage.setItem(
|
||||
ownOutboxKey(prefix, pubkey, relayUrl),
|
||||
JSON.stringify({ store, queuedAt: nowSecs }),
|
||||
);
|
||||
} catch {
|
||||
// Best-effort durability; the in-memory pending edit still drives this
|
||||
// session's publish even if the persisted copy could not be written.
|
||||
// Best-effort durability.
|
||||
}
|
||||
}
|
||||
|
||||
/** Remove this window's own outbox key (its edit published or is a no-op). */
|
||||
export function clearOwnOutbox(
|
||||
prefix: string,
|
||||
pubkey: string,
|
||||
relayUrl: string,
|
||||
): void {
|
||||
try {
|
||||
window.localStorage.removeItem(ownOutboxKey(prefix, pubkey, relayUrl));
|
||||
} catch {
|
||||
// Ignore — a stale own key is re-evaluated on the next publish/boot.
|
||||
}
|
||||
}
|
||||
|
||||
function localStorageKeys(): string[] {
|
||||
const ls = window.localStorage;
|
||||
const keys: string[] = [];
|
||||
for (let i = 0; i < ls.length; i++) {
|
||||
const k = ls.key(i);
|
||||
if (k !== null) keys.push(k);
|
||||
}
|
||||
return keys;
|
||||
}
|
||||
|
||||
/**
|
||||
* Enumerate every window's durable outbox record for a lane, plus the single
|
||||
* legacy shared key from a pre-per-window build (treated as one more record so
|
||||
* an edit persisted by a prior build still resumes and is later reclaimed).
|
||||
*/
|
||||
export function enumerateOutbox<T>(
|
||||
prefix: string,
|
||||
legacyKey: string,
|
||||
pubkey: string,
|
||||
relayUrl: string,
|
||||
parseStore: (json: unknown) => T | null,
|
||||
): OutboxRecord<T>[] {
|
||||
const scopePrefix = outboxScopePrefix(prefix, pubkey, relayUrl);
|
||||
const own = ownOutboxKey(prefix, pubkey, relayUrl);
|
||||
const records: OutboxRecord<T>[] = [];
|
||||
try {
|
||||
for (const key of localStorageKeys()) {
|
||||
if (!key.startsWith(scopePrefix)) continue;
|
||||
const raw = window.localStorage.getItem(key);
|
||||
const parsed = parseEnvelope(raw, parseStore);
|
||||
if (parsed && raw !== null) {
|
||||
records.push({
|
||||
key,
|
||||
store: parsed.store,
|
||||
queuedAt: parsed.queuedAt,
|
||||
raw,
|
||||
isOwn: key === own,
|
||||
});
|
||||
}
|
||||
}
|
||||
const legacyRaw = window.localStorage.getItem(legacyKey);
|
||||
const legacy = parseEnvelope(legacyRaw, parseStore);
|
||||
if (legacy && legacyRaw !== null) {
|
||||
records.push({
|
||||
key: legacyKey,
|
||||
store: legacy.store,
|
||||
queuedAt: legacy.queuedAt,
|
||||
raw: legacyRaw,
|
||||
isOwn: false,
|
||||
});
|
||||
}
|
||||
} catch {
|
||||
// Return whatever parsed before the failure.
|
||||
}
|
||||
return records;
|
||||
}
|
||||
|
||||
/**
|
||||
* Reclaim redundant FOREIGN outbox keys whose edits durable relay evidence
|
||||
* shows are safe to drop (`shouldReclaim`). Never touches this window's own
|
||||
* key. Re-reads each key immediately before removing it and skips if the stored
|
||||
* value changed since enumeration, so a live owner's fresh write is preserved.
|
||||
*/
|
||||
export function reclaimOutbox<T>(
|
||||
prefix: string,
|
||||
legacyKey: string,
|
||||
pubkey: string,
|
||||
relayUrl: string,
|
||||
parseStore: (json: unknown) => T | null,
|
||||
shouldReclaim: (record: OutboxRecord<T>) => boolean,
|
||||
): void {
|
||||
for (const record of enumerateOutbox(
|
||||
prefix,
|
||||
legacyKey,
|
||||
pubkey,
|
||||
relayUrl,
|
||||
parseStore,
|
||||
)) {
|
||||
if (record.isOwn) continue;
|
||||
if (!shouldReclaim(record)) continue;
|
||||
try {
|
||||
// Recheck: only remove if the value is byte-identical to what the
|
||||
// reclaim decision was made against. A peer that rewrote the key in the
|
||||
// gap owns a fresh edit we must not destroy.
|
||||
if (window.localStorage.getItem(record.key) === record.raw) {
|
||||
window.localStorage.removeItem(record.key);
|
||||
}
|
||||
} catch {
|
||||
// Leave for the next boot's reclamation.
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Clear a durable outbox entry with cross-window ownership.
|
||||
* The single winning record among all windows' whole-blob outbox entries for
|
||||
* resume: max `queuedAt`, ties broken by the (nonce-bearing) key string so the
|
||||
* choice is deterministic across windows. Null when no record exists.
|
||||
*
|
||||
* 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).
|
||||
* Whole-blob LWW means only the newest queued intent is replayed; an older
|
||||
* blob a peer queued is superseded by definition and never resurrected.
|
||||
*/
|
||||
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;
|
||||
export function latestOutbox<T>(
|
||||
prefix: string,
|
||||
legacyKey: string,
|
||||
pubkey: string,
|
||||
relayUrl: string,
|
||||
parseStore: (json: unknown) => T | null,
|
||||
): T | null {
|
||||
let best: OutboxRecord<T> | null = null;
|
||||
for (const record of enumerateOutbox(
|
||||
prefix,
|
||||
legacyKey,
|
||||
pubkey,
|
||||
relayUrl,
|
||||
parseStore,
|
||||
)) {
|
||||
if (
|
||||
best === null ||
|
||||
record.queuedAt > best.queuedAt ||
|
||||
(record.queuedAt === best.queuedAt && record.key > best.key)
|
||||
) {
|
||||
best = record;
|
||||
}
|
||||
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.
|
||||
}
|
||||
return best?.store ?? null;
|
||||
}
|
||||
|
||||
/**
|
||||
|
||||
@@ -9,6 +9,7 @@ import {
|
||||
mutedChannelIdsFromStore,
|
||||
readChannelMutesOutbox,
|
||||
readChannelMutesStore,
|
||||
reclaimSubsumedMutesOutbox,
|
||||
storageKey,
|
||||
writeChannelMutesStore,
|
||||
type ChannelMuteEntry,
|
||||
@@ -106,6 +107,11 @@ export function useChannelMutes(
|
||||
if (cancelled) return;
|
||||
if (result.action === "apply-remote") {
|
||||
setStore(applyRemote(result.data));
|
||||
// Head fetch succeeded: reclaim any foreign window's outbox key the
|
||||
// head already subsumes (a peer that published then quit). Gated on the
|
||||
// fetched head and rechecked per key, so a live peer's unpublished edit
|
||||
// is never destroyed. A `hold` (absent/failed head) reclaims nothing.
|
||||
reclaimSubsumedMutesOutbox(pubkey, relayUrl, result.data.store);
|
||||
}
|
||||
// "hold": seed already performed by bootstrap (if first-sync), or blocked.
|
||||
// Resume any edit persisted to the durable outbox before a prior
|
||||
|
||||
@@ -86,6 +86,9 @@ test("live remote while a local edit is pending defers to the pending edit", asy
|
||||
const { act, cleanup, renderHook } = await import("@testing-library/react");
|
||||
const { relayClient } = await import("@/shared/api/relayClient");
|
||||
const { useChannelSections } = await import("./useChannelSections.ts");
|
||||
const { readChannelSectionsOutbox } = await import(
|
||||
"./channelSectionsStorage.ts"
|
||||
);
|
||||
|
||||
const origFetch = relayClient.fetchEvents;
|
||||
const origLive = relayClient.subscribeLive;
|
||||
@@ -130,7 +133,6 @@ test("live remote while a local edit is pending defers to the pending edit", asy
|
||||
|
||||
const pubkey = "pk-live-pending";
|
||||
const relayUrl = "wss://r.live";
|
||||
const outboxKey = `buzz-channel-sections-outbox.v1:${pubkey}:${encodeURIComponent(relayUrl)}`;
|
||||
|
||||
let hook = null;
|
||||
try {
|
||||
@@ -147,7 +149,7 @@ test("live remote while a local edit is pending defers to the pending edit", asy
|
||||
hook.result.current.createSection("Local");
|
||||
});
|
||||
assert.ok(
|
||||
window.localStorage.getItem(outboxKey),
|
||||
readChannelSectionsOutbox(pubkey, relayUrl),
|
||||
"local edit persisted to outbox",
|
||||
);
|
||||
const localSectionIds = hook.result.current.sections.map((s) => s.id);
|
||||
@@ -173,7 +175,7 @@ test("live remote while a local edit is pending defers to the pending edit", asy
|
||||
"pending local edit must NOT be overwritten by the live remote",
|
||||
);
|
||||
assert.ok(
|
||||
window.localStorage.getItem(outboxKey),
|
||||
readChannelSectionsOutbox(pubkey, relayUrl),
|
||||
"outbox for the pending edit must survive the live remote",
|
||||
);
|
||||
hook.unmount();
|
||||
|
||||
@@ -7,6 +7,7 @@ import {
|
||||
DEFAULT_STORE,
|
||||
readChannelSectionsOutbox,
|
||||
readChannelSectionsStore,
|
||||
reclaimSupersededSectionsOutbox,
|
||||
storageKey,
|
||||
writeChannelSectionsStore,
|
||||
} from "./channelSectionsStorage";
|
||||
@@ -143,6 +144,15 @@ export function useChannelSections(
|
||||
if (cancelled) return;
|
||||
if (result.action === "apply-remote") {
|
||||
setStore(applyRemote(result.data));
|
||||
// Head fetch succeeded: reclaim any foreign window's outbox key the
|
||||
// relay head itself supersedes (`queuedAt` ≤ head `created_at`). Gated
|
||||
// on the fetched head and rechecked per key, so a live peer's edit
|
||||
// queued after the head is kept. A `hold` (absent/failed) reclaims none.
|
||||
reclaimSupersededSectionsOutbox(
|
||||
pubkey,
|
||||
relayUrl,
|
||||
result.data.createdAt,
|
||||
);
|
||||
}
|
||||
// "hold": seed already performed by bootstrap (if first-sync), or
|
||||
// blocked (failed fetch / prior watermark). The reconciliation effect
|
||||
|
||||
@@ -29,6 +29,7 @@ test("live remote while a local edit is pending defers to the pending edit", asy
|
||||
const { useChannelSortPreference } = await import(
|
||||
"./useChannelSortPreference.ts"
|
||||
);
|
||||
const { readChannelSortOutbox } = await import("./channelSortPreference.ts");
|
||||
|
||||
const origFetch = relayClient.fetchEvents;
|
||||
const origLive = relayClient.subscribeLive;
|
||||
@@ -72,7 +73,6 @@ test("live remote while a local edit is pending defers to the pending edit", asy
|
||||
|
||||
const pubkey = "pk-sort-pending";
|
||||
const relayUrl = "wss://r.live";
|
||||
const outboxKey = `buzz-channel-sort-outbox.v1:${pubkey}:${encodeURIComponent(relayUrl)}`;
|
||||
|
||||
let hook = null;
|
||||
try {
|
||||
@@ -89,7 +89,7 @@ test("live remote while a local edit is pending defers to the pending edit", asy
|
||||
hook.result.current.setSortModeFor("channels", "recent");
|
||||
});
|
||||
assert.ok(
|
||||
window.localStorage.getItem(outboxKey),
|
||||
readChannelSortOutbox(pubkey, relayUrl),
|
||||
"local edit persisted to outbox",
|
||||
);
|
||||
assert.equal(
|
||||
@@ -124,7 +124,7 @@ test("live remote while a local edit is pending defers to the pending edit", asy
|
||||
"remote store must not have been applied over the pending edit",
|
||||
);
|
||||
assert.ok(
|
||||
window.localStorage.getItem(outboxKey),
|
||||
readChannelSortOutbox(pubkey, relayUrl),
|
||||
"outbox for the pending edit must survive the live remote",
|
||||
);
|
||||
hook.unmount();
|
||||
|
||||
@@ -7,6 +7,7 @@ import {
|
||||
DEFAULT_STORE,
|
||||
readChannelSortOutbox,
|
||||
readChannelSortStore,
|
||||
reclaimSupersededSortOutbox,
|
||||
sortModeForGroup,
|
||||
storageKey,
|
||||
stripOrphanedSectionModes,
|
||||
@@ -141,6 +142,11 @@ export function useChannelSortPreference(
|
||||
if (cancelled) return;
|
||||
if (result.action === "apply-remote") {
|
||||
setStore(applyRemote(result.data));
|
||||
// Head fetch succeeded: reclaim any foreign window's outbox key the
|
||||
// relay head itself supersedes (`queuedAt` ≤ head `created_at`). Gated
|
||||
// on the fetched head and rechecked per key, so a live peer's edit
|
||||
// queued after the head is kept. A `hold` (absent/failed) reclaims none.
|
||||
reclaimSupersededSortOutbox(pubkey, relayUrl, result.data.createdAt);
|
||||
}
|
||||
// "hold": seed already performed by bootstrap (if first-sync), or blocked
|
||||
// (failed fetch / prior watermark). Resume any edit persisted to the
|
||||
|
||||
@@ -8,6 +8,7 @@ import {
|
||||
mergeStores,
|
||||
readChannelStarsOutbox,
|
||||
readChannelStarsStore,
|
||||
reclaimSubsumedStarsOutbox,
|
||||
starredChannelIdsFromStore,
|
||||
storageKey,
|
||||
writeChannelStarsStore,
|
||||
@@ -106,10 +107,15 @@ export function useChannelStars(
|
||||
if (cancelled) return;
|
||||
if (result.action === "apply-remote") {
|
||||
setStore(applyRemote(result.data));
|
||||
// Head fetch succeeded: reclaim any foreign window's outbox key the
|
||||
// head already subsumes (a peer that published then quit). Gated on the
|
||||
// fetched head and rechecked per key, so a live peer's unpublished edit
|
||||
// is never destroyed. A `hold` (absent/failed head) reclaims nothing.
|
||||
reclaimSubsumedStarsOutbox(pubkey, relayUrl, result.data.store);
|
||||
}
|
||||
// "hold": seed already performed by bootstrap (if first-sync), or blocked.
|
||||
// Resume any edit persisted to the durable outbox before a prior
|
||||
// quit/community-switch so a click made <2s before teardown still syncs.
|
||||
// Resume every window's edit persisted to the durable outbox before a
|
||||
// prior quit/community-switch so a click made <2s before teardown still
|
||||
// syncs. Merges all records (order-independent), then publishes.
|
||||
const outbox = readChannelStarsOutbox(pubkey, relayUrl);
|
||||
if (outbox) {
|
||||
managerRef.current?.publishStars(outbox);
|
||||
|
||||
Reference in New Issue
Block a user