jerry/buzz
1
0
Fork 0
mirror of https://github.com/block/buzz.git synced 2026-11-08 10:21:46 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(sync): clamp publish timestamps and give sort prefs the durable lane

Browse Source 
Carl's re-review found three edit-loss paths remaining after the
stars/mutes rev-merge landed.

[P2] Stars, mutes, and sort stamped createdAt = lastRemoteCreatedAt + 1
uncapped, so a self-authored head accepted near the relay's +900s drift
boundary wedged every later publish until wall time caught up. Extract
the sections clamp into a shared clampPublishCreatedAt in
sidebarSyncWatermark.ts (all four surfaces already import that module)
and wire sections/stars/mutes/sort to it.

[P1] Sort preferences never got the durable lane: doPublish only logged
on failure, destroy() discarded the pending edit, and a live remote
cancelled the debounce with the edit stranded. Port the sections lane —
durable outbox + bootstrap resume, generation/CAS ownership, single-
flight + completion re-drive, 2s->30s backoff, 60s reconcile loop, and
pending-aware remote application. Sort stays whole-blob LWW; a lost head
is adopted at pre-publish rather than republished.

Tests: a clamp test per surface, and sort lifecycle coverage — outbox
resume after destroy-inside-debounce, retry without a later edit,
overlapping-generation safety, live-remote-during-debounce adopt, and a
hook-level pending-defer test.

Co-authored-by: Will Pfleger <pfleger.will@gmail.com>
Signed-off-by: Will Pfleger <pfleger.will@gmail.com>
This commit is contained in:
Duncan
2026-08-26 11:35:16 -04:00
parent 16aeab68b0
commit c7933fd9da
11 changed files with 1054 additions and 106 deletions
Download Patch File Download Diff File Expand all files Collapse all files

47
desktop/src/features/sidebar/lib/channelMutesSync.test.mjs
View File

@@ -393,3 +393,50 @@ test("revert-fix: relay-A watermark does not suppress first-sync seed on relay-B
mock.reset();
}
});
// ─── Timestamp clamp (Carl P2): a far-future remote head must not push our
// published createdAt past the relay's ±15min future-drift window.
// Mutation test: removing the clamp lets createdAt = lastRemote+1 (~now+3600),
// which exceeds now + MAX_PUBLISH_FUTURE_SECS.
test("timestamp clamp: published createdAt stays inside the relay future window", async () => {
const nowSecs = Math.floor(Date.now() / 1000);
const farFutureHead = nowSecs + 3_600; // 1h ahead — beyond the ±15min window
let call = 0;
mock.method(relayClient, "fetchEvents", () => {
call++;
// First call primes lastRemoteCreatedAt to farFutureHead (undecryptable so
// it does not merge into the store); pre-publish call returns created_at=0.
return Promise.resolve([
{
pubkey: "pk-clamp",
content: "bad-cipher",
created_at: call === 1 ? farFutureHead : 0,
id: "evt-clamp",
},
]);
});
let signedCreatedAt = null;
const fw = makeFakeWindow();
const restore = installFakeWindow(fw);
const tauri = installTauriMock(JSON.stringify({ version: 1, channels: {} }));
mock.method(relayClient, "publishEvent", (evt) => {
signedCreatedAt = evt.created_at;
return Promise.resolve();
});
try {
const manager = new ChannelMuteSyncManager("pk-clamp", RELAY);
await manager.fetchRemoteMutes(); // prime lastRemoteCreatedAt
manager.publishMutes(makeStore({ ch1: E(true, 100, 1) }));
fw._fireTimer();
await new Promise((r) => setTimeout(r, 20));
assert.ok(signedCreatedAt !== null, "publish must have been attempted");
assert.ok(
signedCreatedAt <= Math.floor(Date.now() / 1000) + 840,
`createdAt must be clamped inside the future window — got ${signedCreatedAt}`,
);
} finally {
tauri.restore();
restore();
mock.reset();
}
});

9
desktop/src/features/sidebar/lib/channelMutesSync.ts
View File

@@ -15,6 +15,7 @@ import {
} from "./channelMutesStorage";
import {
advanceWatermark,
clampPublishCreatedAt,
readWatermark,
runBootstrap,
type FetchResult,
@@ -295,10 +296,10 @@ export class ChannelMuteSyncManager {
channels: merged.channels,
};
const ciphertext = await nip44EncryptToSelf(JSON.stringify(payload));
const createdAt = Math.max(
Math.floor(Date.now() / 1_000),
this.lastRemoteCreatedAt + 1,
);
// Clamp inside the relay's future-drift window so a skewed remote head
// can never make us stamp an unbounded future timestamp that wedges every
// subsequent publish; we adopt such a head on the next fetch instead.
const createdAt = clampPublishCreatedAt(this.lastRemoteCreatedAt);
const event = await signRelayEvent({
kind: KIND_CHANNEL_MUTES,
content: ciphertext,

15
desktop/src/features/sidebar/lib/channelSectionsSync.ts
View File

@@ -15,6 +15,7 @@ import {
} from "./channelSectionsStorage";
import {
advanceWatermark,
clampPublishCreatedAt,
readWatermark,
runBootstrap,
type FetchResult,
@@ -24,14 +25,6 @@ const D_TAG = "channel-sections";
const BLOB_TYPE = D_TAG;
const DEBOUNCE_MS = 2_000;
// The relay rejects events more than ±15 minutes (900s) from server time
// (`MAX_TIMESTAMP_DRIFT_SECS` in ingest.rs). Clamp our published `created_at`
// well inside that window so a skewed remote head can never make us manufacture
// an unbounded future timestamp that wedges every subsequent publish. 840s
// leaves ~60s of transit margin while still letting us win LWW against any
// legitimately-timestamped head.
const MAX_PUBLISH_FUTURE_SECS = 840;
// Bounded backoff for a retained pending edit whose publish failed transiently
// (timeout / socket error) on an otherwise-healthy socket, so it does not wait
// for a reconnect that may never fire.
@@ -460,16 +453,12 @@ export class ChannelSectionSyncManager {
assignments: merged.assignments,
};
const ciphertext = await nip44EncryptToSelf(JSON.stringify(payload));
const now = Math.floor(Date.now() / 1_000);
// Clamp inside the relay's future-drift window: never manufacture a
// timestamp so far ahead that this or a later publish is rejected for
// drift and wedges. If a skewed remote head sits beyond the window we
// will lose LWW and adopt it on the next pre-publish fetch rather than
// walking past it.
const createdAt = Math.min(
Math.max(now, this.lastRemoteCreatedAt + 1),
now + MAX_PUBLISH_FUTURE_SECS,
);
const createdAt = clampPublishCreatedAt(this.lastRemoteCreatedAt);
const event = await signRelayEvent({
kind: KIND_CHANNEL_SECTIONS,
content: ciphertext,

53
desktop/src/features/sidebar/lib/channelSortPreference.ts
View File

@@ -133,6 +133,59 @@ export function writeChannelSortStore(
}
}
const OUTBOX_KEY_PREFIX = "buzz-channel-sort-outbox.v1";
function outboxKey(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.
*/
export function writeChannelSortOutbox(
pubkey: string,
store: ChannelSortStore,
relayUrl: string,
): void {
try {
window.localStorage.setItem(
outboxKey(pubkey, relayUrl),
JSON.stringify(boundChannelSortStore(store)),
);
} catch {
// Best-effort durability; the in-memory pendingStore still drives this
// session's publish even if the persisted copy could not be written.
}
}
/** Read a persisted unpublished sort edit, or null when none/unparseable. */
export function readChannelSortOutbox(
pubkey: string,
relayUrl: string,
): ChannelSortStore | null {
try {
const raw = window.localStorage.getItem(outboxKey(pubkey, relayUrl));
if (!raw) return null;
return parseChannelSortPayload(JSON.parse(raw));
} catch {
return null;
}
}
/** Clear the persisted sort outbox (edit published, superseded, or a no-op). */
export function clearChannelSortOutbox(pubkey: string, relayUrl: string): void {
try {
window.localStorage.removeItem(outboxKey(pubkey, relayUrl));
} catch {
// Ignore — a stale outbox entry is re-evaluated (and re-cleared if
// identical to the head) on the next publish attempt.
}
}
export function sortModeForGroup(
store: ChannelSortStore,
group: ChannelSortGroupKey,

397
desktop/src/features/sidebar/lib/channelSortSync.test.mjs
View File

@@ -15,6 +15,7 @@ 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", () => {
@@ -92,8 +93,7 @@ test("destroy: is safe to call with no pending publish", () => {
// Wiring tests 1-3 drive the production bootstrap() path; policy tested once
// in sidebarSyncWatermark.test.mjs.
// 1. fetch failed (error/timeout) + local non-empty → hold, zero publish calls
// Mutation: removing the failed guard causes bootstrap to call publishSortPrefs → pendingStore set.
// 1. fetch failed → hold, pendingStore null (mutation: remove failed guard → seed queued)
test("revert-fix: fetch failed (error) does not trigger seed-publish via bootstrap", async () => {
mock.method(relayClient, "fetchEvents", () =>
Promise.reject(new Error("relay timeout")),
@@ -112,8 +112,7 @@ test("revert-fix: fetch failed (error) does not trigger seed-publish via bootstr
}
});
// 2. absent + persisted head > 0 → hold, zero publish calls (the dev-build stale-copy case)
// Mutation: setting watermark to 0 in localStorage causes bootstrap to seed.
// 2. absent + persisted head > 0 → hold, zero publish calls (dev-build stale-copy case)
test("revert-fix: absent fetch with prior watermark blocks seed-publish via bootstrap", async () => {
mock.method(relayClient, "fetchEvents", () => Promise.resolve([]));
mock.method(relayClient, "publishEvent", () => Promise.resolve());
@@ -125,13 +124,6 @@ test("revert-fix: absent fetch with prior watermark blocks seed-publish via boot
const restore = installFakeWindow(fw);
try {
const manager = new ChannelSortSyncManager("pk-stale", RELAY);
assert.ok(
Number(
fw.localStorage.getItem(
`buzz-sync-watermark.v1:channel-sort:pk-stale:${RELAY_KEY}`,
) ?? "0",
) > 0,
);
const result = await manager.bootstrap(makeStore({ channels: "recent" }));
assert.equal(result.action, "hold");
assert.equal(manager.getPendingStore(), null);
@@ -142,7 +134,6 @@ test("revert-fix: absent fetch with prior watermark blocks seed-publish via boot
});
// 3. absent + head 0 + local non-empty → seed-publish queued (first-sync preserved)
// Mutation: removing the absent+head-0 seed call leaves pendingStore null.
test("revert-fix: absent fetch with zero watermark seeds via bootstrap (first-sync preserved)", async () => {
mock.method(relayClient, "fetchEvents", () => Promise.resolve([]));
mock.method(relayClient, "publishEvent", () => Promise.resolve());
@@ -150,12 +141,6 @@ test("revert-fix: absent fetch with zero watermark seeds via bootstrap (first-sy
const restore = installFakeWindow(fw);
try {
const manager = new ChannelSortSyncManager("pk-fresh", RELAY);
assert.equal(
fw.localStorage.getItem(
`buzz-sync-watermark.v1:channel-sort:pk-fresh:${RELAY_KEY}`,
),
null,
);
const result = await manager.bootstrap(makeStore({ channels: "recent" }));
assert.equal(result.action, "hold");
assert.ok(manager.getPendingStore() !== null);
@@ -165,25 +150,30 @@ test("revert-fix: absent fetch with zero watermark seeds via bootstrap (first-sy
}
});
// 4. LWW baseline: newer decryptable pre-publish event still wins after an
// undecryptable head was recorded.
// Mutation test: headBeforeFetch → this.lastRemoteCreatedAt makes comparison
// 200>200=false → local wins instead of remote → wrong content encrypted.
test("revert-fix: sort LWW — newer decryptable pre-publish event selected after undecryptable head recorded", async () => {
// ─── Whole-blob LWW: adopt-winner and local-winner ────────────────────────────
// 4a. Adopt-winner: a newer remote head at pre-publish time supersedes the local
// edit — the manager must NOT publish, must hand the remote to the adopt
// sink, and must clear pending/outbox so the loser can't be replayed.
// Mutation: reverting adopt→republish makes onRemoteAdopted never fire and
// publishEvent fire instead.
test("adopt-winner: newer remote head at pre-publish adopts remote and skips publish", async () => {
const REMOTE_KEY = "remote-group-from-relay";
let callCount = 0;
mock.method(relayClient, "fetchEvents", () => {
callCount++;
return Promise.resolve([
mock.method(relayClient, "fetchEvents", () =>
Promise.resolve([
{
pubkey: "pk-lww",
content: callCount === 1 ? "bad-cipher" : "good-cipher",
created_at: callCount === 1 ? 100 : 200,
id: `evt-${callCount}`,
content: "good-cipher",
created_at: 200,
id: "evt-remote",
},
]);
]),
);
const publishCalls = [];
mock.method(relayClient, "publishEvent", (...args) => {
publishCalls.push(args);
return Promise.resolve();
});
mock.method(relayClient, "publishEvent", () => Promise.resolve());
const fw = makeFakeWindow();
const restore = installFakeWindow(fw);
const tauri = installTauriMock(
@@ -191,22 +181,30 @@ test("revert-fix: sort LWW — newer decryptable pre-publish event selected afte
);
try {
const manager = new ChannelSortSyncManager("pk-lww", RELAY);
await manager.fetchRemoteSortPrefs();
assert.ok(
Number(
fw.localStorage.getItem(
`buzz-sync-watermark.v1:channel-sort:pk-lww:${RELAY_KEY}`,
) ?? "0",
) >= 100,
);
const adopted = [];
manager.setOnRemoteAdopted((r) => adopted.push(r));
manager.publishSortPrefs(makeStore({ "local-group": "recent" }));
assert.ok(
fw.localStorage.getItem(OUTBOX_KEY("pk-lww")) !== null,
"edit must be persisted to the durable outbox",
);
fw._fireTimer();
await new Promise((r) => setTimeout(r, 20));
const pt = tauri.capturedPlaintext();
assert.ok(pt !== null, "nip44EncryptToSelf must have been called");
assert.equal(
publishCalls.length,
0,
"must not publish when a newer remote head wins LWW",
);
assert.equal(adopted.length, 1, "adopt sink must receive the remote");
assert.ok(
JSON.parse(pt).groups && REMOTE_KEY in JSON.parse(pt).groups,
`remote groups must win LWW merge — got: ${pt}`,
REMOTE_KEY in adopted[0].store.groups,
"adopted store must be the remote content",
);
assert.equal(manager.getPendingStore(), null, "pending must be cleared");
assert.equal(
fw.localStorage.getItem(OUTBOX_KEY("pk-lww")),
null,
"outbox must be cleared on adopt so the loser is never replayed",
);
} finally {
tauri.restore();
@@ -215,7 +213,316 @@ test("revert-fix: sort LWW — newer decryptable pre-publish event selected afte
}
});
// 5. live-sub: undecryptable event on live path records head before decrypt
// 4b. Local edit wins (no newer remote head): publishes and clears the outbox.
test("adopt-winner: local edit at/ahead of head publishes and clears outbox", async () => {
mock.method(relayClient, "fetchEvents", () => Promise.resolve([]));
const publishCalls = [];
mock.method(relayClient, "publishEvent", (...args) => {
publishCalls.push(args);
return Promise.resolve();
});
const fw = makeFakeWindow();
const restore = installFakeWindow(fw);
const tauri = installTauriMock("{}");
try {
const manager = new ChannelSortSyncManager("pk-win", RELAY);
manager.publishSortPrefs(makeStore({ channels: "recent" }));
fw._fireTimer();
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")),
null,
"outbox must be cleared once the edit is published",
);
} finally {
tauri.restore();
restore();
mock.reset();
}
});
// 4c. Timestamp clamp: a remote head far in the future must not make the
// published createdAt walk past the relay's ±15min window.
// Mutation test: removing the clamp lets createdAt = lastRemote+1 (~now+3600),
// which exceeds now + MAX_PUBLISH_FUTURE_SECS.
test("timestamp clamp: published createdAt stays inside the relay future window", async () => {
const nowSecs = Math.floor(Date.now() / 1000);
const farFutureHead = nowSecs + 3_600; // 1h ahead — beyond the ±15min window
let call = 0;
mock.method(relayClient, "fetchEvents", () => {
call++;
// First call primes lastRemoteCreatedAt to farFutureHead (undecryptable so
// no adopt); pre-publish call returns created_at=0 so the local edit wins.
return Promise.resolve([
{
pubkey: "pk-clamp",
content: "bad-cipher",
created_at: call === 1 ? farFutureHead : 0,
id: "evt-clamp",
},
]);
});
let signedCreatedAt = null;
const fw = makeFakeWindow();
const restore = installFakeWindow(fw);
const tauri = installTauriMock(JSON.stringify({ version: 1, groups: {} }));
mock.method(relayClient, "publishEvent", (evt) => {
signedCreatedAt = evt.created_at;
return Promise.resolve();
});
try {
const manager = new ChannelSortSyncManager("pk-clamp", RELAY);
await manager.fetchRemoteSortPrefs(); // prime lastRemoteCreatedAt
manager.publishSortPrefs(makeStore({ channels: "recent" }));
fw._fireTimer();
await new Promise((r) => setTimeout(r, 20));
assert.ok(signedCreatedAt !== null, "publish must have been attempted");
assert.ok(
signedCreatedAt <= Math.floor(Date.now() / 1000) + 840,
`createdAt must be clamped inside the future window — got ${signedCreatedAt}`,
);
} finally {
tauri.restore();
restore();
mock.reset();
}
});
// ─── Durable lane: outbox resume, retry, and cross-generation safety ──────────
// 5. Durable outbox resume: an edit made <2s before quit (destroy inside the
// debounce) is persisted, and a fresh manager resuming from that outbox
// publishes it — the edit is not silently dropped at teardown.
// Mutation: dropping writeChannelSortOutbox leaves the outbox null → no resume.
test("durable outbox: edit destroyed inside the debounce resumes and publishes on remount", async () => {
mock.method(relayClient, "fetchEvents", () => Promise.resolve([]));
const publishCalls = [];
mock.method(relayClient, "publishEvent", (...args) => {
publishCalls.push(args);
return Promise.resolve();
});
const fw = makeFakeWindow();
const restore = installFakeWindow(fw);
const tauri = installTauriMock("{}");
try {
// 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"));
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).
const m2 = new ChannelSortSyncManager("pk-resume", RELAY);
m2.publishSortPrefs(JSON.parse(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")),
null,
"outbox must be cleared once the resumed edit publishes",
);
} finally {
tauri.restore();
restore();
mock.reset();
}
});
// 6. Failed publish retries with no later edit: a transient publish rejection
// keeps the pending edit and schedules a bounded-backoff retry that succeeds,
// rather than logging-and-dropping (the pre-fix behaviour).
// Mutation: reverting scheduleRetry to a bare console.warn leaves pending null
// and never re-publishes.
test("failed publish retries the retained edit without a later edit", async () => {
mock.method(relayClient, "fetchEvents", () => Promise.resolve([]));
let attempts = 0;
mock.method(relayClient, "publishEvent", () => {
attempts++;
if (attempts === 1) return Promise.reject(new Error("socket timeout"));
return Promise.resolve();
});
const fw = makeFakeWindow();
const restore = installFakeWindow(fw);
const tauri = installTauriMock("{}");
try {
const manager = new ChannelSortSyncManager("pk-retry", RELAY);
manager.publishSortPrefs(makeStore({ channels: "recent" }));
fw._fireTimer(); // debounce → doPublish → publish rejects → scheduleRetry
await new Promise((r) => setTimeout(r, 20));
assert.equal(attempts, 1, "first publish attempt rejected");
assert.ok(
manager.getPendingStore() !== null,
"failed publish must retain the pending edit",
);
assert.ok(fw._hasTimer(), "a bounded-backoff retry must be scheduled");
fw._fireTimer(); // retry → doPublish → publish resolves
await new Promise((r) => setTimeout(r, 20));
assert.equal(attempts, 2, "retry must re-attempt the publish");
assert.equal(
manager.getPendingStore(),
null,
"successful retry must clear the pending edit",
);
} finally {
tauri.restore();
restore();
mock.reset();
}
});
// 7. Overlapping publishes: an older completion must not erase a newer queued
// edit or its outbox (generation compare-and-swap).
// Mutation: dropping the gen guard in discardPending clears B on A's completion.
test("overlapping publishes: older completion does not erase a newer queued edit", async () => {
mock.method(relayClient, "fetchEvents", () => Promise.resolve([]));
let releaseFirst = null;
let publishCalls = 0;
mock.method(relayClient, "publishEvent", () => {
publishCalls++;
if (publishCalls === 1) {
return new Promise((res) => {
releaseFirst = res;
});
}
return Promise.resolve();
});
const storage = new Map();
const timers = new Map();
let nextId = 1;
const fakeWindow = {
localStorage: {
getItem: (k) => storage.get(k) ?? null,
setItem: (k, v) => storage.set(k, v),
removeItem: (k) => storage.delete(k),
},
setTimeout: (fn, ms) => {
const id = nextId++;
timers.set(id, { fn, ms });
return id;
},
clearTimeout: (id) => timers.delete(id),
};
const fireDelay = async (ms) => {
const entry = [...timers.entries()].find(([, v]) => v.ms === ms);
assert.ok(entry, `expected a timer scheduled at ${ms}ms`);
timers.delete(entry[0]);
entry[1].fn();
await Promise.resolve();
await Promise.resolve();
};
const restore = installFakeWindow(fakeWindow);
const tauri = installTauriMock("{}");
try {
const manager = new ChannelSortSyncManager("pk-overlap", RELAY);
manager.publishSortPrefs(makeStore({ channels: "recent" })); // A
await fireDelay(2000); // debounce → doPublish(A) awaits publishEvent
while (releaseFirst === null) await Promise.resolve();
manager.publishSortPrefs(makeStore({ dms: "alpha" })); // B while A in flight
assert.deepEqual(
Object.keys(manager.getPendingStore()?.groups ?? {}),
["dms"],
"B is now the pending edit",
);
assert.deepEqual(
Object.keys(JSON.parse(storage.get(OUTBOX_KEY("pk-overlap"))).groups),
["dms"],
"outbox holds B",
);
releaseFirst(); // A completes — must NOT clear B's pending/outbox
await Promise.resolve();
await Promise.resolve();
await Promise.resolve();
assert.deepEqual(
Object.keys(manager.getPendingStore()?.groups ?? {}),
["dms"],
"older completion must leave B pending",
);
assert.ok(
storage.get(OUTBOX_KEY("pk-overlap")) !== undefined,
"older completion must leave B's outbox intact",
);
manager.destroy();
} finally {
tauri.restore();
restore();
mock.reset();
}
});
// 8. Live remote during debounce is adopted at pre-publish, not overwritten. The
// live event advances the mutable watermark before doPublish runs, so
// comparing the fetched head against the watermark would see equality and
// publish over the newer remote. The pre-publish check compares against the
// baseline frozen at publishSortPrefs instead.
// Mutation: comparing against lastRemoteCreatedAt rather than publishBaseline
// republishes local over the newer remote.
test("live remote during debounce is adopted at pre-publish, not overwritten", async () => {
const REMOTE_KEY = "remote-during-debounce";
let liveCallback = null;
mock.method(relayClient, "subscribeLive", (_filter, onEvent) => {
liveCallback = onEvent;
return Promise.resolve(async () => {});
});
// Pre-publish fetch returns the same newer head the live event delivered.
mock.method(relayClient, "fetchEvents", () =>
Promise.resolve([
{
pubkey: "pk-live-deb",
content: "good-cipher",
created_at: 500,
id: "evt-live",
},
]),
);
const publishCalls = [];
mock.method(relayClient, "publishEvent", (...args) => {
publishCalls.push(args);
return Promise.resolve();
});
const fw = makeFakeWindow();
const restore = installFakeWindow(fw);
const tauri = installTauriMock(
JSON.stringify({ version: 1, groups: { [REMOTE_KEY]: "recent" } }),
);
try {
const manager = new ChannelSortSyncManager("pk-live-deb", RELAY);
const adopted = [];
manager.setOnRemoteAdopted((r) => adopted.push(r));
await manager.subscribeToSortPrefs(() => {});
// Local edit queued first, freezing the baseline at the empty head.
manager.publishSortPrefs(makeStore({ "local-group": "alpha" }));
// A newer remote head arrives during the debounce window (advances the
// mutable watermark to created_at=500).
liveCallback({
pubkey: "pk-live-deb",
content: "good-cipher",
created_at: 500,
id: "evt-live",
});
await new Promise((r) => setTimeout(r, 0));
fw._fireTimer(); // debounce → doPublish → pre-publish sees advanced head
await new Promise((r) => setTimeout(r, 20));
assert.equal(
publishCalls.length,
0,
"must not publish over a remote that became head during the debounce",
);
assert.equal(adopted.length, 1, "the newer remote must be adopted");
} finally {
tauri.restore();
restore();
mock.reset();
}
});
// 9. live-sub: undecryptable event on live path records head before decrypt.
// Mutation test: removing recordRemoteHead before decrypt in the live callback
// leaves watermark at 0 after a live event.
test("revert-fix: undecryptable live event advances watermark before decrypt attempt", async () => {

344
desktop/src/features/sidebar/lib/channelSortSync.ts
View File

@@ -7,11 +7,14 @@ import {
import type { RelayEvent } from "@/shared/api/types";
import { KIND_CHANNEL_SORT } from "@/shared/constants/kinds";
import {
clearChannelSortOutbox,
parseChannelSortPayload,
writeChannelSortOutbox,
type ChannelSortStore,
} from "./channelSortPreference";
import {
advanceWatermark,
clampPublishCreatedAt,
readWatermark,
runBootstrap,
type FetchResult,
@@ -21,12 +24,75 @@ const D_TAG = "channel-sort";
const BLOB_TYPE = D_TAG;
const DEBOUNCE_MS = 2_000;
// Bounded backoff for a retained pending edit whose publish failed transiently
// (timeout / socket error) on an otherwise-healthy socket, so it does not wait
// for a reconnect that may never fire.
const RETRY_BASE_MS = 2_000;
const RETRY_MAX_MS = 30_000;
export type RemoteSortPrefs = {
store: ChannelSortStore;
createdAt: number;
eventId: string;
};
/**
* Outcome of the pre-publish head check.
*
* - `publish` — local edit is at or ahead of the head; publish it.
* - `adopt` — a newer remote head exists; the local edit lost whole-blob LWW
* and must be discarded in favour of the remote store so UI and
* relay converge. The manager hands the remote back to the hook
* and never publishes.
*/
type PublishDecision =
| { kind: "publish"; store: ChannelSortStore }
| { kind: "adopt"; remote: RemoteSortPrefs };
/**
* The canonical remote head as it stood when an edit was queued. The pre-publish
* check compares the fetched head against this frozen baseline — never against
* the mutable in-memory watermark, which a live event observed during the
* debounce window may already have advanced to that same head (silently
* suppressing the adopt).
*/
type PublishBaseline = { createdAt: number; eventId: string };
/**
* True when `head` is the canonical winner over the baseline the edit was queued
* against — i.e. the head advanced since the edit began. Canonical order is
* `created_at DESC, id ASC`: a strictly-later head wins, and a same-second head
* wins only with a strictly-lower id. A same-second head is comparable only once
* the baseline id is known (empty id = no prior head seen → not superseded).
*/
function remoteAdvancedSince(
head: RemoteSortPrefs,
baseline: PublishBaseline,
): boolean {
if (head.createdAt > baseline.createdAt) return true;
return (
head.createdAt === baseline.createdAt &&
baseline.eventId !== "" &&
head.eventId < baseline.eventId
);
}
/**
* True when tuple `a` is the canonical winner over `b` (`created_at DESC,
* id ASC`). An empty id means "no head seen yet" and always loses.
*/
function canonicalGreater(a: PublishBaseline, b: PublishBaseline): boolean {
if (a.eventId === "") return false;
if (b.eventId === "") return true;
if (a.createdAt !== b.createdAt) return a.createdAt > b.createdAt;
return a.eventId < b.eventId;
}
/** The canonical-greater of two head tuples (`created_at DESC, id ASC`). */
function canonicalMax(a: PublishBaseline, b: PublishBaseline): PublishBaseline {
return canonicalGreater(a, b) ? a : b;
}
async function decryptAndParse(
event: RelayEvent,
): Promise<RemoteSortPrefs | null> {
@@ -48,15 +114,48 @@ async function decryptAndParse(
* low-frequency preference blob, so whole-blob LWW (like sections) is
* sufficient — per-key merge (like stars/mutes) would be unnecessary
* complexity here.
*
* The durable lane (outbox + generation/CAS + bounded retry) is mirrored from
* sections so a sort edit made just before quit/community-switch, or one whose
* publish times out, self-heals rather than being silently dropped.
*/
export class ChannelSortSyncManager {
private pubkey: string;
private relayUrl: string;
private debounceTimer: number | null = null;
private retryTimer: number | null = null;
private retryDelayMs = RETRY_BASE_MS;
private lastRemoteCreatedAt: number;
// Canonical best head observed so far (`created_at DESC, id ASC`). Frozen into
// a per-edit baseline at publishSortPrefs so the pre-publish check can tell
// whether the head advanced *since the edit was queued*, independent of the
// mutable watermark that a live event during the debounce window may advance.
private lastRemoteHead: PublishBaseline = { createdAt: 0, eventId: "" };
// The canonical head this pending edit is racing against, frozen when the edit
// was queued and advanced ONLY by our own successful publishes.
private publishBaseline: PublishBaseline = { createdAt: 0, eventId: "" };
private pendingStore: ChannelSortStore | null = null;
// Monotonic id for the current pending edit. Every publishSortPrefs() bumps
// it; every scheduled publish/retry captures the value it was queued for. A
// completion (success, adopt, or no-op) may only clear pending state via
// compare-and-swap on this generation, so an older in-flight publish can
// never erase a newer edit that arrived while it was in flight.
private pendingGeneration = 0;
// 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.
private publishInFlight = false;
// Event ids we signed and sent to the relay but whose ACK never arrived. The
// relay MAY have accepted such a write, so if a later cycle's pre-publish
// fetch returns a head whose id is in this set, that head is OUR OWN accepted
// predecessor — fold it forward and publish above it, rather than adopting it
// and erasing the queued edit.
private ambiguousAttemptIds = new Set<string>();
private lastPublishedStore: ChannelSortStore | null = null;
private destroyed = false;
// Set by the hook so an adopted remote head (local edit lost whole-blob LWW)
// is written through to React state + localStorage.
private onRemoteAdopted: ((remote: RemoteSortPrefs) => void) | null = null;
constructor(pubkey: string, relayUrl: string) {
this.pubkey = pubkey;
@@ -64,6 +163,11 @@ export class ChannelSortSyncManager {
this.lastRemoteCreatedAt = readWatermark(pubkey, BLOB_TYPE, relayUrl);
}
/** Register the hook's adopt-remote sink (write-through to UI + storage). */
setOnRemoteAdopted(cb: (remote: RemoteSortPrefs) => void): void {
this.onRemoteAdopted = cb;
}
async fetchRemoteSortPrefs(): Promise<FetchResult<RemoteSortPrefs>> {
try {
const events = await relayClient.fetchEvents({
@@ -76,7 +180,10 @@ export class ChannelSortSyncManager {
return { status: "absent" };
}
const event = events[0];
this.recordRemoteHead(event.created_at);
// An event exists — record its created_at regardless of whether we can
// decrypt it, so seed-publish is blocked even when the payload is
// unreadable (e.g. wrong key).
this.recordRemoteHead(event.created_at, event.id);
const result = await decryptAndParse(event);
if (!result) {
return { status: "failed", createdAt: event.created_at };
@@ -92,10 +199,21 @@ export class ChannelSortSyncManager {
}
}
private recordRemoteHead(createdAt: number): void {
/** Update in-memory + persisted watermark and the canonical head tuple. */
private recordRemoteHead(createdAt: number, eventId: string): void {
if (createdAt > this.lastRemoteCreatedAt) {
this.lastRemoteCreatedAt = createdAt;
}
// Track the canonical-best head (`created_at DESC, id ASC`): a later head
// always wins; a same-second head wins only with a strictly-lower id.
if (
createdAt > this.lastRemoteHead.createdAt ||
(createdAt === this.lastRemoteHead.createdAt &&
(this.lastRemoteHead.eventId === "" ||
eventId < this.lastRemoteHead.eventId))
) {
this.lastRemoteHead = { createdAt, eventId };
}
advanceWatermark(this.pubkey, BLOB_TYPE, this.relayUrl, createdAt);
}
@@ -104,26 +222,110 @@ export class ChannelSortSyncManager {
window.clearTimeout(this.debounceTimer);
this.debounceTimer = null;
}
if (this.retryTimer !== null) {
window.clearTimeout(this.retryTimer);
this.retryTimer = null;
}
}
getPendingStore(): ChannelSortStore | null {
return this.pendingStore;
}
/** True while an unpublished local edit is queued (debouncing or retrying). */
hasPendingEdit(): boolean {
return this.pendingStore !== null;
}
/**
* Adopt a remote store that superseded a local edit: hand it to the hook for
* write-through, advance the watermark, and drop the losing pending edit —
* including the durable outbox, so the outbox can never replay an edit that
* adopt just decided lost.
*
* Compare-and-swap on `gen`: if a newer edit arrived while this publish was in
* flight, the generation has moved on and that newer edit is the latest writer
* — a stale adopt must not clear its pending state or overwrite its optimistic
* UI. We still advance the watermark (monotonic and always safe).
*/
private adoptRemote(remote: RemoteSortPrefs, gen: number): void {
this.recordRemoteHead(remote.createdAt, remote.eventId);
if (gen !== this.pendingGeneration) return;
this.pendingStore = null;
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.
*/
private discardPending(gen: number): void {
if (gen !== this.pendingGeneration) return;
this.pendingStore = null;
clearChannelSortOutbox(this.pubkey, this.relayUrl);
}
publishSortPrefs(store: ChannelSortStore): void {
this.pendingStore = store;
++this.pendingGeneration;
// Freeze the canonical head this edit is racing against at queue time so a
// live event applied during the debounce window (which advances the mutable
// watermark) cannot make the pre-publish comparison see equality and fall
// through to a publish that overwrites a remote that became head after this
// edit was queued. The baseline only advances via our own successful
// publishes, so a prior generation's own accepted write is folded in rather
// than mistaken for a competing remote.
this.publishBaseline = { ...this.lastRemoteHead };
// Persist synchronously so an edit made <2s before quit/community-switch
// survives teardown and resumes on next mount (durable outbox).
writeChannelSortOutbox(this.pubkey, store, this.relayUrl);
if (this.debounceTimer !== null) {
window.clearTimeout(this.debounceTimer);
}
// A fresh edit supersedes any retry scheduled for the previous generation.
if (this.retryTimer !== null) {
window.clearTimeout(this.retryTimer);
this.retryTimer = null;
}
this.retryDelayMs = RETRY_BASE_MS;
this.debounceTimer = window.setTimeout(() => {
this.debounceTimer = null;
void this.doPublish(store);
this.startCycle();
}, DEBOUNCE_MS);
}
/**
* Serialize publish cycles: at most one runs at a time. A debounce/retry timer
* that fires while a cycle is in flight defers — the in-flight cycle's
* completion re-drives if a pending edit still needs publishing.
*/
private startCycle(): void {
if (this.destroyed || this.pendingStore === null) return;
if (this.publishInFlight) return;
const store = this.pendingStore;
const gen = this.pendingGeneration;
this.publishInFlight = true;
void this.doPublish(store, gen).finally(() => {
this.publishInFlight = false;
if (
!this.destroyed &&
this.pendingStore !== null &&
this.debounceTimer === null &&
this.retryTimer === null
) {
this.startCycle();
}
});
}
private async fetchOwnBlobBeforePublish(
store: ChannelSortStore,
): Promise<ChannelSortStore> {
): Promise<PublishDecision> {
try {
const events = await relayClient.fetchEvents({
kinds: [KIND_CHANNEL_SORT],
@@ -131,23 +333,35 @@ export class ChannelSortSyncManager {
"#d": [D_TAG],
limit: 1,
});
if (events.length === 0 || events[0].pubkey !== this.pubkey) return store;
if (events.length === 0 || events[0].pubkey !== this.pubkey)
return { kind: "publish", store };
const event = events[0];
// Snapshot the watermark before advancing it: after recordRemoteHead
// runs, lastRemoteCreatedAt equals event.created_at, so the LWW
// comparison remote.createdAt > lastRemoteCreatedAt would always be
// false and silently suppress the merge.
const headBeforeFetch = this.lastRemoteCreatedAt;
this.recordRemoteHead(event.created_at);
const remote = await decryptAndParse(event);
if (!remote) return store;
// Sort prefs use whole-blob LWW: take whichever is newer
if (remote.createdAt > headBeforeFetch) {
return remote.store;
// Record the head after decrypt attempt so the watermark/head-tuple
// advance even for an undecryptable payload.
this.recordRemoteHead(event.created_at, event.id);
if (!remote) return { kind: "publish", store };
// Sort prefs use whole-blob LWW. Compare the fetched head against the
// baseline frozen when this edit was queued — NOT the live watermark. If
// the canonical head advanced since the edit began, the local edit lost
// and is adopted-away rather than republished over it.
if (remoteAdvancedSince(remote, this.publishBaseline)) {
// Unless the advancing head is a prior publish of OURS whose ACK was
// lost: it is our own accepted predecessor, not a competing remote.
// Fold it into the baseline and publish above it so the queued edit
// survives instead of adopting our own stale write away.
if (this.ambiguousAttemptIds.has(remote.eventId)) {
this.publishBaseline = canonicalMax(this.publishBaseline, {
createdAt: remote.createdAt,
eventId: remote.eventId,
});
return { kind: "publish", store };
}
return { kind: "adopt", remote };
}
return store;
return { kind: "publish", store };
} catch {
return store;
return { kind: "publish", store };
}
}
@@ -164,15 +378,40 @@ export class ChannelSortSyncManager {
return true;
}
private async doPublish(store: ChannelSortStore): Promise<void> {
/** Schedule a bounded-backoff retry of the retained pending edit. */
private scheduleRetry(gen: number): void {
if (this.destroyed || this.pendingStore === null) return;
// A newer edit has superseded this one; its own timer owns the retry.
if (gen !== this.pendingGeneration) return;
if (this.retryTimer !== null) return;
const delay = this.retryDelayMs;
this.retryDelayMs = Math.min(this.retryDelayMs * 2, RETRY_MAX_MS);
this.retryTimer = window.setTimeout(() => {
this.retryTimer = null;
this.startCycle();
}, delay);
}
private async doPublish(store: ChannelSortStore, gen: number): Promise<void> {
// A newer edit was queued after this publish was scheduled; it owns the
// pending state and will publish the latest store — abandon this stale run.
if (gen !== this.pendingGeneration) return;
try {
const merged = await this.fetchOwnBlobBeforePublish(store);
const decision = await this.fetchOwnBlobBeforePublish(store);
// Guard: manager may have been destroyed while fetchOwnBlobBeforePublish
// was awaited (community switch during in-flight fetch). If so, abort
// before touching the relay.
// was awaited (community switch during in-flight fetch).
if (this.destroyed) return;
// A newer edit was queued while we awaited the pre-publish fetch. It owns
// convergence now; the serialized cycle re-drives for it once this run
// unwinds.
if (gen !== this.pendingGeneration) return;
if (decision.kind === "adopt") {
this.adoptRemote(decision.remote, gen);
return;
}
const merged = decision.store;
if (this.isIdenticalToLastPublished(merged)) {
this.pendingStore = null;
this.discardPending(gen);
return;
}
const payload = {
@@ -180,10 +419,10 @@ export class ChannelSortSyncManager {
groups: merged.groups,
};
const ciphertext = await nip44EncryptToSelf(JSON.stringify(payload));
const createdAt = Math.max(
Math.floor(Date.now() / 1_000),
this.lastRemoteCreatedAt + 1,
);
// Clamp inside the relay's future-drift window so a skewed remote head
// can never make us stamp an unbounded future timestamp that wedges every
// subsequent publish; we adopt such a head on the next fetch instead.
const createdAt = clampPublishCreatedAt(this.lastRemoteCreatedAt);
const event = await signRelayEvent({
kind: KIND_CHANNEL_SORT,
content: ciphertext,
@@ -193,20 +432,49 @@ export class ChannelSortSyncManager {
["t", D_TAG], // relay discoverability; not used in our filters
],
});
// Final guard immediately before the network call — sign/encrypt are
// synchronous-ish but cheap; the relay socket may have moved to a
// different community by the time we reach this point.
if (this.destroyed) return;
// Final guard immediately before the network call — a newer edit may have
// been queued during the encrypt/sign await, or the manager destroyed.
if (this.destroyed || gen !== this.pendingGeneration) return;
// Record this signed id as an in-flight attempt of unknown fate before we
// send it. If the ACK is lost below, a later cycle that fetches this id as
// the head recognises it as our own accepted write and folds it forward
// rather than adopting it away.
this.ambiguousAttemptIds.add(event.id);
await relayClient.publishEvent(
event,
"Timed out publishing channel sort preferences.",
"Failed to publish channel sort preferences.",
);
this.recordRemoteHead(event.created_at);
this.lastPublishedStore = merged;
this.pendingStore = null;
this.recordRemoteHead(event.created_at, event.id);
// This write is now the confirmed accepted head; it dominates every prior
// attempt (`created_at DESC, id ASC`), so no earlier ambiguous id can ever
// be the canonical head again. Clear the set to keep it bounded.
this.ambiguousAttemptIds.clear();
// Fold our own accepted head into the pending edit's baseline —
// unconditional across generations so a newer edit's pre-publish check
// does not mistake OUR prior publish for a competing remote and adopt it
// away. Genuine remotes never fold in here.
this.publishBaseline = canonicalMax(this.publishBaseline, {
createdAt: event.created_at,
eventId: event.id,
});
// Only claim this store as the published head if it is still the current
// edit; a newer edit queued mid-flight owns lastPublishedStore now.
if (gen === this.pendingGeneration) {
this.lastPublishedStore = merged;
this.retryDelayMs = RETRY_BASE_MS;
}
this.discardPending(gen);
} catch (error) {
if (this.destroyed) return;
// Ambiguous outcome: the publish promise rejected (timeout / socket
// error), but the relay may already have accepted the write before the
// ACK was lost. Keep the pending edit and retry with backoff. The attempt
// id stays in ambiguousAttemptIds: if the relay did accept it, a later
// cycle that fetches this id as the head folds it forward as our own
// accepted predecessor instead of adopting it away.
console.warn("[channelSortSync] publish failed:", error);
this.scheduleRetry(gen);
}
}
@@ -224,7 +492,7 @@ export class ChannelSortSyncManager {
if (event.pubkey !== this.pubkey) return;
// Record the raw head before decrypt so an undecryptable live event
// still advances the watermark and blocks future seed-publish.
this.recordRemoteHead(event.created_at);
this.recordRemoteHead(event.created_at, event.id);
void decryptAndParse(event).then((result) => {
if (result) {
onUpdate(result);
@@ -252,10 +520,10 @@ export class ChannelSortSyncManager {
destroy(): void {
// Cancel any pending publish and mark this manager as destroyed so any
// in-flight doPublish() calls abort before reaching relayClient.
// Pending debounce-window changes are intentionally dropped: flushing
// could publish relay A's sort prefs to relay B via the shared relayClient
// singleton. On return, bootstrap's found path whole-blob-replaces from
// remote, so any dropped pending edit is lost.
// Debounce-window changes are NOT lost: publishSortPrefs persisted them to
// the durable outbox synchronously, and the next mount resumes them.
// Flushing here is still avoided — it could publish relay A's sort prefs to
// relay B via the shared relayClient singleton.
this.destroyed = true;
this.cancelPendingPublish();
this.pendingStore = null;

47
desktop/src/features/sidebar/lib/channelStarsSync.test.mjs
View File

@@ -393,3 +393,50 @@ test("revert-fix: relay-A watermark does not suppress first-sync seed on relay-B
mock.reset();
}
});
// ─── Timestamp clamp (Carl P2): a far-future remote head must not push our
// published createdAt past the relay's ±15min future-drift window.
// Mutation test: removing the clamp lets createdAt = lastRemote+1 (~now+3600),
// which exceeds now + MAX_PUBLISH_FUTURE_SECS.
test("timestamp clamp: published createdAt stays inside the relay future window", async () => {
const nowSecs = Math.floor(Date.now() / 1000);
const farFutureHead = nowSecs + 3_600; // 1h ahead — beyond the ±15min window
let call = 0;
mock.method(relayClient, "fetchEvents", () => {
call++;
// First call primes lastRemoteCreatedAt to farFutureHead (undecryptable so
// it does not merge into the store); pre-publish call returns created_at=0.
return Promise.resolve([
{
pubkey: "pk-clamp",
content: "bad-cipher",
created_at: call === 1 ? farFutureHead : 0,
id: "evt-clamp",
},
]);
});
let signedCreatedAt = null;
const fw = makeFakeWindow();
const restore = installFakeWindow(fw);
const tauri = installTauriMock(JSON.stringify({ version: 1, channels: {} }));
mock.method(relayClient, "publishEvent", (evt) => {
signedCreatedAt = evt.created_at;
return Promise.resolve();
});
try {
const manager = new ChannelStarSyncManager("pk-clamp", RELAY);
await manager.fetchRemoteStars(); // prime lastRemoteCreatedAt
manager.publishStars(makeStore({ ch1: E(true, 100, 1) }));
fw._fireTimer();
await new Promise((r) => setTimeout(r, 20));
assert.ok(signedCreatedAt !== null, "publish must have been attempted");
assert.ok(
signedCreatedAt <= Math.floor(Date.now() / 1000) + 840,
`createdAt must be clamped inside the future window — got ${signedCreatedAt}`,
);
} finally {
tauri.restore();
restore();
mock.reset();
}
});

9
desktop/src/features/sidebar/lib/channelStarsSync.ts
View File

@@ -15,6 +15,7 @@ import {
} from "./channelStarsStorage";
import {
advanceWatermark,
clampPublishCreatedAt,
readWatermark,
runBootstrap,
type FetchResult,
@@ -295,10 +296,10 @@ export class ChannelStarSyncManager {
channels: merged.channels,
};
const ciphertext = await nip44EncryptToSelf(JSON.stringify(payload));
const createdAt = Math.max(
Math.floor(Date.now() / 1_000),
this.lastRemoteCreatedAt + 1,
);
// Clamp inside the relay's future-drift window so a skewed remote head
// can never make us stamp an unbounded future timestamp that wedges every
// subsequent publish; we adopt such a head on the next fetch instead.
const createdAt = clampPublishCreatedAt(this.lastRemoteCreatedAt);
const event = await signRelayEvent({
kind: KIND_CHANNEL_STARS,
content: ciphertext,

29
desktop/src/features/sidebar/lib/sidebarSyncWatermark.ts
View File

@@ -23,6 +23,35 @@ import { normalizeRelayUrl } from "@/shared/lib/normalizeRelayUrl";
const PREFIX = "buzz-sync-watermark.v1";
// The relay rejects events more than ±15 minutes (900s) from server time
// (`MAX_TIMESTAMP_DRIFT_SECS` in ingest.rs). Clamp every sidebar-sync publish's
// `created_at` well inside that window so a skewed remote head can never make a
// manager manufacture an unbounded future timestamp that wedges every
// subsequent publish. 840s leaves ~60s of transit margin while still letting a
// publish win LWW against any legitimately-timestamped head.
const MAX_PUBLISH_FUTURE_SECS = 840;
/**
* Compute the `created_at` for a sidebar-sync publish.
*
* Stamps one second past the newest observed remote head so the write wins
* last-write-wins, but never further ahead than the relay's future-drift
* window. If a skewed remote head sits beyond the window the manager loses LWW
* and adopts it on the next fetch rather than walking past it and wedging.
*
* `nowSecs` defaults to the current wall clock in seconds; callers pass it
* explicitly only to keep a single clock reading across a publish.
*/
export function clampPublishCreatedAt(
lastRemoteCreatedAt: number,
nowSecs: number = Math.floor(Date.now() / 1_000),
): number {
return Math.min(
Math.max(nowSecs, lastRemoteCreatedAt + 1),
nowSecs + MAX_PUBLISH_FUTURE_SECS,
);
}
/**
* Tri-state result returned by every `fetchRemote*()` method.
*

120
desktop/src/features/sidebar/lib/useChannelSortPreference.test.mjs
View File

@@ -18,6 +18,126 @@ before(() => {
after(() => dom.window.close());
// Carl P1-sort regression: a live remote arriving while a local sort edit is
// still pending must NOT overwrite the optimistic edit or strand its durable
// outbox. Reverting applyRemote's `hasPendingEdit()` guard (or restoring the
// old `cancelPendingPublish()` on remote arrival) lets the remote clobber the
// pending edit and drop it.
test("live remote while a local edit is pending defers to the pending edit", async () => {
const { act, cleanup, renderHook } = await import("@testing-library/react");
const { relayClient } = await import("@/shared/api/relayClient");
const { useChannelSortPreference } = await import(
"./useChannelSortPreference.ts"
);
const origFetch = relayClient.fetchEvents;
const origLive = relayClient.subscribeLive;
const origReconnect = relayClient.subscribeToReconnects;
const origPublish = relayClient.publishEvent;
const origTauri = window.__TAURI_INTERNALS__;
let live = null;
relayClient.fetchEvents = async () => [];
relayClient.subscribeLive = async (_f, cb) => {
live = cb;
return async () => {};
};
relayClient.subscribeToReconnects = () => () => {};
relayClient.publishEvent = async () => {};
window.__TAURI_INTERNALS__ = {
invoke: (cmd) => {
if (cmd === "nip44_decrypt_from_self")
return Promise.resolve(
JSON.stringify({
version: 1,
groups: { "remote-group": "recent" },
}),
);
if (cmd === "nip44_encrypt_to_self") return Promise.resolve("ct");
if (cmd === "sign_event")
return Promise.resolve(
JSON.stringify({
id: "signed",
pubkey: "pk-sort-pending",
content: "ct",
created_at: 0,
kind: 30078,
tags: [],
sig: "s",
}),
);
return Promise.reject(new Error(`unmocked ${cmd}`));
},
};
const pubkey = "pk-sort-pending";
const relayUrl = "wss://r.live";
const outboxKey = `buzz-channel-sort-outbox.v1:${pubkey}:${encodeURIComponent(relayUrl)}`;
let hook = null;
try {
await act(async () => {
hook = renderHook(() => useChannelSortPreference(pubkey, relayUrl));
await Promise.resolve();
await Promise.resolve();
await Promise.resolve();
});
assert.ok(live, "live subscription installed");
// Make a local edit — it becomes the pending store and persists to outbox.
await act(async () => {
hook.result.current.setSortModeFor("channels", "recent");
});
assert.ok(
window.localStorage.getItem(outboxKey),
"local edit persisted to outbox",
);
assert.equal(
hook.result.current.sortModeFor("channels"),
"recent",
"optimistic local edit applied",
);
// A remote live event arrives while the edit is still pending.
await act(async () => {
live({
id: "remote-event",
pubkey,
created_at: 500,
content: "cipher",
kind: 30078,
tags: [["d", "channel-sort"]],
sig: "s",
});
await Promise.resolve();
await Promise.resolve();
});
assert.equal(
hook.result.current.sortModeFor("channels"),
"recent",
"pending local edit must NOT be overwritten by the live remote",
);
assert.equal(
hook.result.current.sortModeFor("remote-group"),
"alpha",
"remote store must not have been applied over the pending edit",
);
assert.ok(
window.localStorage.getItem(outboxKey),
"outbox for the pending edit must survive the live remote",
);
hook.unmount();
} finally {
cleanup();
relayClient.fetchEvents = origFetch;
relayClient.subscribeLive = origLive;
relayClient.subscribeToReconnects = origReconnect;
relayClient.publishEvent = origPublish;
window.__TAURI_INTERNALS__ = origTauri;
}
});
// Equal-timestamp tie-break must match the relay's canonical winner
// (`created_at DESC, id ASC` → LOWEST id wins). Deliver the larger id first,
// then the lower id at the same timestamp; the lower id is the stored winner

90
desktop/src/features/sidebar/lib/useChannelSortPreference.ts
View File

@@ -3,7 +3,9 @@ import * as React from "react";
import { relayClient } from "@/shared/api/relayClient";
import {
boundChannelSortStore,
clearChannelSortOutbox,
DEFAULT_STORE,
readChannelSortOutbox,
readChannelSortStore,
sortModeForGroup,
storageKey,
@@ -16,6 +18,13 @@ import {
import { ChannelSortSyncManager } from "./channelSortSync";
import type { RemoteSortPrefs } from "./channelSortSync";
// Reconciliation cadence (mirrors sections). Steady interval re-fetches the head
// on a healthy socket so divergence self-heals without a reconnect; the retry
// window backs off from base to max while the fetch keeps failing.
const RECONCILE_STEADY_MS = 60_000;
const RECONCILE_RETRY_BASE_MS = 3_000;
const RECONCILE_RETRY_MAX_MS = 60_000;
/**
* Persistent per-group sidebar sort preferences, scoped by pubkey + relay so
* they don't bleed across identities or communities (same scoping as channel
@@ -85,6 +94,13 @@ export function useChannelSortPreference(
): ((prev: ChannelSortStore) => ChannelSortStore) => {
return (prev) => {
if (!pubkey) return prev;
// A pending local edit owns convergence: its debounced publish re-checks
// the head and either wins (publish) or loses (adopt, which routes back
// through onRemoteAdopted with pending already cleared). Never let a
// passive remote arrival clobber the optimistic edit or strand its
// durable outbox. The adopt path clears pending before calling us, so
// this guard is false there and the winning remote still writes through.
if (managerRef.current?.hasPendingEdit()) return prev;
if (remote.createdAt < lastAppliedRemoteTs.current) return prev;
// Equal timestamps: the relay/database break ties by `id ASC` — the
// LOWEST event id is the canonical winner. Apply a strictly-lower id and
@@ -97,7 +113,6 @@ export function useChannelSortPreference(
return prev;
lastAppliedRemoteTs.current = remote.createdAt;
lastAppliedEventId.current = remote.eventId;
managerRef.current?.cancelPendingPublish();
if (!writeChannelSortStore(pubkey, remote.store, relayUrl)) return prev;
return remote.store;
};
@@ -105,6 +120,19 @@ export function useChannelSortPreference(
[pubkey, relayUrl],
);
React.useEffect(() => {
if (!pubkey || !relayUrl) return;
const manager = managerRef.current;
if (!manager) return;
// When a local edit loses whole-blob LWW (pre-publish head is newer), the
// manager adopts the winning remote store. Write it through to React state +
// localStorage so the UI and relay never diverge; applyRemote also advances
// the applied-ts guard.
manager.setOnRemoteAdopted((remote) => {
setStore(applyRemote(remote));
});
}, [pubkey, relayUrl, applyRemote]);
React.useEffect(() => {
if (!pubkey || !relayUrl) return;
let cancelled = false;
@@ -114,13 +142,71 @@ export function useChannelSortPreference(
if (result.action === "apply-remote") {
setStore(applyRemote(result.data));
}
// "hold": seed already performed by bootstrap (if first-sync), or blocked.
// "hold": seed already performed by bootstrap (if first-sync), or blocked
// (failed fetch / prior watermark). Resume any edit persisted to the
// durable outbox before a prior quit/community-switch.
const outbox = readChannelSortOutbox(pubkey, relayUrl);
if (outbox) {
managerRef.current?.publishSortPrefs(outbox);
} else {
clearChannelSortOutbox(pubkey, relayUrl);
}
});
return () => {
cancelled = true;
};
}, [pubkey, relayUrl, applyRemote]);
// Reconciliation loop (mirrors sections): a single scheduler that both retries
// a failed bootstrap with bounded backoff and periodically re-fetches the
// head, so stale-at-open state converges without waiting for a reconnect event
// a healthy socket never fires. Also refreshes when the window becomes visible.
React.useEffect(() => {
if (!pubkey || !relayUrl) return;
let cancelled = false;
let timer: number | null = null;
let delayMs = RECONCILE_RETRY_BASE_MS;
const schedule = (ms: number) => {
if (cancelled) return;
if (timer !== null) window.clearTimeout(timer);
timer = window.setTimeout(tick, ms);
};
const tick = () => {
void managerRef.current?.fetchRemoteSortPrefs().then((result) => {
if (cancelled) return;
if (result.status === "found") {
// applyRemote defers to a pending local edit (whose own debounced
// publish converges via publish-or-adopt), so a periodic reconcile
// can never drop it — no re-queue needed.
setStore(applyRemote(result.data));
delayMs = RECONCILE_STEADY_MS; // relay answered → steady cadence
} else if (result.status === "absent") {
delayMs = RECONCILE_STEADY_MS; // answered (no blob) → steady cadence
} else {
delayMs = Math.min(delayMs * 2, RECONCILE_RETRY_MAX_MS); // fetch failed → back off
}
schedule(delayMs);
});
};
const onVisible = () => {
if (document.visibilityState === "visible") {
delayMs = RECONCILE_RETRY_BASE_MS;
tick();
}
};
document.addEventListener("visibilitychange", onVisible);
schedule(delayMs);
return () => {
cancelled = true;
if (timer !== null) window.clearTimeout(timer);
document.removeEventListener("visibilitychange", onVisible);
};
}, [pubkey, relayUrl, applyRemote]);
React.useEffect(() => {
if (!pubkey) return;
let unsub: (() => Promise<void>) | null = null;