mirror of
https://github.com/block/buzz.git
synced 2026-10-30 14:01:37 +00:00
fix(sidebar-sync): correct P1 guard shape; add P3 reconnect/remount preservation; causal tests
P1 guard shape fix (both whole-blob hooks):
- Was: reduce all non-apply-remote paths to headCreatedAt=0, gate on queuedAt>0
- Now: branch on bootstrap result explicitly
- hold/absent/failed (result.action !== 'apply-remote'): always replay,
including legacy queuedAt=0 records (old 0>0 guard silently stranded them)
- apply-remote: suppress only on strict queuedAt < appliedHead.createdAt;
equality replays (aligns with reclamation's strict-lt, prevents stranding
a same-second edit neither consumed nor published)
- Documents clock-skew residual: queuedAt vs relay created_at are independent
wall clocks — accepted tradeoff of the existing LWW max-merge strategy
P3 reconnect preservation (MergeLaneSyncManager + both stars/mutes hooks):
- Add retryReconnectPublish() to MergeLaneSyncManager: re-drives the existing
generation without calling publish(), so pendingPreservedKey is not reset
- Expose retryReconnectStarsPublish() / retryReconnectMutesPublish() on adapters
- Switch both hooks' reconnect effects to retryReconnectPublish() instead of
publishStars/Mutes(pending), which would open a new generation and clear key
- Clear pendingPreservedKey in destroy()
P3 remount preservation (storage layer + both hooks):
- writeOwnOutbox: accept optional preservedKey, include in envelope JSON
- parseEnvelope: read preservedKey from envelope; OutboxRecord gains field
- enumerateOutbox: propagate preservedKey into OutboxRecord
- mergeLaneStorage.shared: writeOutbox forwards preservedKey; add
readOutboxPreservedKey to read own-window record's key for bootstrap
- channelStarsStorage / channelMutesStorage: accept+forward preservedKey in
write, expose readChannelStarsOutboxPreservedKey / readChannelMutesOutboxPreservedKey
- MergeLaneSyncManager: writeOutbox config type includes preservedKey param;
publish() forwards it to config.writeOutbox
- useChannelStars / useChannelMutes: import and call readPreservedKey on
bootstrap outbox replay; pass to publishStars/Mutes alongside the store
Causal test coverage:
- wholeBlobHook.shared: replace non-causal stale-replay test with four-case
P1 regression using makeHookTimerBed.hasDelay(2000) as the synchronous
causal signal (timer present iff publish() called): strict-lt suppresses,
equality replays, hold+v2 replays, hold+legacy-queuedAt=0 replays
- channelStarsSync.test: add P3 reconnect (retryReconnectStarsPublish keeps
key) and P3 remount (storage round-trip write+read preservedKey)
- channelMutesSync.test: add matching P3 501-entry, reconnect, and remount
regressions parameterizing coverage across both lanes
Co-authored-by: Will Pfleger <pfleger.will@gmail.com>
Signed-off-by: Will Pfleger <pfleger.will@gmail.com>
This commit is contained in:
@@ -2,6 +2,7 @@ import {
|
||||
clearOutbox,
|
||||
makeStorageKey,
|
||||
readOutbox,
|
||||
readOutboxPreservedKey,
|
||||
readStore,
|
||||
reclaimSubsumedOutbox,
|
||||
writeOutbox,
|
||||
@@ -282,8 +283,16 @@ export function writeChannelMutesOutbox(
|
||||
pubkey: string,
|
||||
store: ChannelMuteStore,
|
||||
relayUrl: string,
|
||||
preservedKey?: string,
|
||||
): void {
|
||||
writeOutbox(OUTBOX_KEY_PREFIX, pubkey, store, relayUrl, boundMuteStore);
|
||||
writeOutbox(
|
||||
OUTBOX_KEY_PREFIX,
|
||||
pubkey,
|
||||
store,
|
||||
relayUrl,
|
||||
boundMuteStore,
|
||||
preservedKey,
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -305,6 +314,23 @@ export function readChannelMutesOutbox(
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
* Read the `preservedKey` from this window's own outbox record, or `undefined`
|
||||
* when none exists. Used on bootstrap to restore the capacity-bounding
|
||||
* reservation across remount (Kalvin P3).
|
||||
*/
|
||||
export function readChannelMutesOutboxPreservedKey(
|
||||
pubkey: string,
|
||||
relayUrl: string,
|
||||
): string | undefined {
|
||||
return readOutboxPreservedKey(
|
||||
OUTBOX_KEY_PREFIX,
|
||||
pubkey,
|
||||
relayUrl,
|
||||
parseMutePayload,
|
||||
);
|
||||
}
|
||||
|
||||
/** Clear this window's own outbox key (its edit published or is a no-op). */
|
||||
export function clearChannelMutesOutbox(
|
||||
pubkey: string,
|
||||
|
||||
@@ -140,3 +140,195 @@ test("mutes wire: typed API (publishMutes, getPendingMuteStore, fetchRemoteMutes
|
||||
mock.reset();
|
||||
}
|
||||
});
|
||||
|
||||
// Mutation: removing preservedKey from mergeWithRemote call lets the clicked
|
||||
// channel be evicted at 501 entries (Kalvin P3 — mutes lane).
|
||||
test("P3: clicked channel is preserved through pre-publish mergeWithRemote at capacity boundary (501 entries)", async () => {
|
||||
const MAX = 500;
|
||||
const clickedId = "ch-clicked-mute";
|
||||
const clickedEntry = { muted: true, updatedAt: 1, rev: 1 }; // oldest updatedAt
|
||||
|
||||
// Local: clicked channel + 499 others with updatedAt=100
|
||||
const localChannels = { [clickedId]: clickedEntry };
|
||||
for (let i = 0; i < MAX - 1; i++) {
|
||||
localChannels[`ch-local-${i}`] = { muted: false, updatedAt: 100, rev: 0 };
|
||||
}
|
||||
const localStore = { version: 1, channels: localChannels };
|
||||
|
||||
// Remote: same 499 non-clicked channels + one fresh channel not in local
|
||||
const remoteChannels = {};
|
||||
for (let i = 0; i < MAX - 1; i++) {
|
||||
remoteChannels[`ch-local-${i}`] = { muted: false, updatedAt: 100, rev: 0 };
|
||||
}
|
||||
remoteChannels["ch-remote-new"] = { muted: false, updatedAt: 100, rev: 0 };
|
||||
const remoteStore = { version: 1, channels: remoteChannels };
|
||||
|
||||
const fw = makeFakeWindow();
|
||||
const restore = installFakeWindow(fw);
|
||||
const tauri = installEchoTauri("pk-p3-preserve-mutes");
|
||||
const remoteHead = tauri.mintHead(remoteStore, 50, "evt-remote-mutes");
|
||||
remoteHead.tags = [["d", "channel-mutes"]];
|
||||
|
||||
let fetchCalls = 0;
|
||||
mock.method(relayClient, "fetchEvents", () => {
|
||||
fetchCalls++;
|
||||
if (fetchCalls === 1) return Promise.resolve([remoteHead]);
|
||||
return Promise.resolve([]);
|
||||
});
|
||||
let publishedEvent = null;
|
||||
mock.method(relayClient, "publishEvent", (evt) => {
|
||||
publishedEvent = evt;
|
||||
return Promise.resolve();
|
||||
});
|
||||
|
||||
try {
|
||||
const m = new ChannelMuteSyncManager("pk-p3-preserve-mutes", RELAY);
|
||||
m.publishMutes(localStore, clickedId);
|
||||
fw._fireTimer();
|
||||
await new Promise((r) => setTimeout(r, 30));
|
||||
assert.ok(publishedEvent !== null, "publish must have been attempted");
|
||||
const plaintext = tauri.capturedPlaintext();
|
||||
assert.ok(plaintext !== null, "encrypt must have been called");
|
||||
const published = JSON.parse(plaintext);
|
||||
assert.ok(
|
||||
clickedId in published.channels,
|
||||
`clicked channel "${clickedId}" must survive 501-entry merge when preservedKey is passed`,
|
||||
);
|
||||
assert.ok(
|
||||
"ch-remote-new" in published.channels,
|
||||
"new remote channel must be present in merged result",
|
||||
);
|
||||
assert.equal(
|
||||
Object.keys(published.channels).length,
|
||||
MAX,
|
||||
`merged result must be bounded to ${MAX} entries`,
|
||||
);
|
||||
m.destroy();
|
||||
} finally {
|
||||
tauri.restore();
|
||||
restore();
|
||||
mock.reset();
|
||||
}
|
||||
});
|
||||
|
||||
// Mutation: calling public publish() on reconnect resets pendingPreservedKey,
|
||||
// letting a 501-entry pre-publish merge evict the clicked channel again (P3).
|
||||
test("P3 reconnect: retryReconnectMutesPublish preserves clicked channel key across reconnect", async () => {
|
||||
const MAX = 500;
|
||||
const clickedId = "ch-reconnect-mute";
|
||||
const clickedEntry = { muted: true, updatedAt: 1, rev: 1 }; // oldest
|
||||
const localChannels = { [clickedId]: clickedEntry };
|
||||
for (let i = 0; i < MAX - 1; i++) {
|
||||
localChannels[`ch-local-${i}`] = { muted: false, updatedAt: 100, rev: 0 };
|
||||
}
|
||||
const localStore = { version: 1, channels: localChannels };
|
||||
|
||||
// Remote: a new channel not in local, making merged = 501
|
||||
const remoteChannels = {};
|
||||
for (let i = 0; i < MAX - 1; i++) {
|
||||
remoteChannels[`ch-local-${i}`] = { muted: false, updatedAt: 100, rev: 0 };
|
||||
}
|
||||
remoteChannels["ch-remote-rc"] = { muted: false, updatedAt: 100, rev: 0 };
|
||||
const remoteStore = { version: 1, channels: remoteChannels };
|
||||
|
||||
const fw = makeFakeWindow();
|
||||
const restore = installFakeWindow(fw);
|
||||
const tauri = installEchoTauri("pk-p3-reconnect-mutes");
|
||||
const remoteHead = tauri.mintHead(remoteStore, 50, "evt-rc-mutes");
|
||||
remoteHead.tags = [["d", "channel-mutes"]];
|
||||
|
||||
let fetchCalls = 0;
|
||||
mock.method(relayClient, "fetchEvents", () => {
|
||||
fetchCalls++;
|
||||
if (fetchCalls === 1) return Promise.resolve([remoteHead]);
|
||||
return Promise.resolve([]);
|
||||
});
|
||||
let publishedEvent = null;
|
||||
mock.method(relayClient, "publishEvent", (evt) => {
|
||||
publishedEvent = evt;
|
||||
return Promise.resolve();
|
||||
});
|
||||
|
||||
try {
|
||||
const m = new ChannelMuteSyncManager("pk-p3-reconnect-mutes", RELAY);
|
||||
// Initial click: sets pendingPreservedKey = clickedId
|
||||
m.publishMutes(localStore, clickedId);
|
||||
// Cancel the debounce (simulates quit before the 2s fires).
|
||||
fw._fireTimer(); // flush pending timer
|
||||
// Let any microtasks settle — abort the current cycle if one started
|
||||
await new Promise((r) => setTimeout(r, 0));
|
||||
// Reset: simulate the manager having a pending edit with preservedKey set
|
||||
// but the cycle not yet complete. We reinitialise with a fresh timer.
|
||||
m.cancelPendingMutePublish();
|
||||
// Re-publish with key to set pendingPreservedKey properly
|
||||
fetchCalls = 0;
|
||||
publishedEvent = null;
|
||||
m.publishMutes(localStore, clickedId);
|
||||
// Use retryReconnectMutesPublish — must NOT reset pendingPreservedKey
|
||||
m.retryReconnectMutesPublish();
|
||||
fw._fireTimer();
|
||||
await new Promise((r) => setTimeout(r, 30));
|
||||
// If retryReconnect properly preserved the key, clicked channel survives.
|
||||
assert.ok(publishedEvent !== null, "publish must have fired after retry");
|
||||
const plaintext = tauri.capturedPlaintext();
|
||||
const published = JSON.parse(plaintext);
|
||||
assert.ok(
|
||||
clickedId in published.channels,
|
||||
`clicked channel must survive 501-entry merge after retryReconnectMutesPublish`,
|
||||
);
|
||||
m.destroy();
|
||||
} finally {
|
||||
tauri.restore();
|
||||
restore();
|
||||
mock.reset();
|
||||
}
|
||||
});
|
||||
|
||||
// Mutation: omitting preservedKey from writeChannelMutesOutbox means it is not
|
||||
// persisted in the envelope, so readChannelMutesOutboxPreservedKey returns
|
||||
// undefined and bootstrap replay calls publishMutes(outbox) without protection
|
||||
// — a 501-entry pre-publish merge can evict the clicked channel on remount (P3).
|
||||
test("P3 remount: writeChannelMutesOutbox persists preservedKey; readChannelMutesOutboxPreservedKey recovers it", async () => {
|
||||
const {
|
||||
readChannelMutesOutbox,
|
||||
readChannelMutesOutboxPreservedKey,
|
||||
writeChannelMutesOutbox,
|
||||
} = await import("./channelMutesStorage.ts");
|
||||
|
||||
const fw = makeFakeWindow();
|
||||
const restore = installFakeWindow(fw);
|
||||
|
||||
const pubkey = "pk-p3-remount-mutes-storage";
|
||||
const relayUrl = "wss://r.remount-mutes";
|
||||
const clickedId = "ch-remount-mutes";
|
||||
const store = {
|
||||
version: 1,
|
||||
channels: { [clickedId]: { muted: true, updatedAt: 1, rev: 0 } },
|
||||
};
|
||||
|
||||
try {
|
||||
writeChannelMutesOutbox(pubkey, store, relayUrl, clickedId);
|
||||
|
||||
assert.ok(
|
||||
readChannelMutesOutbox(pubkey, relayUrl) !== null,
|
||||
"outbox must be written",
|
||||
);
|
||||
|
||||
const restoredKey = readChannelMutesOutboxPreservedKey(pubkey, relayUrl);
|
||||
assert.equal(
|
||||
restoredKey,
|
||||
clickedId,
|
||||
"preservedKey must be readable back from the written mutes outbox record",
|
||||
);
|
||||
|
||||
writeChannelMutesOutbox(pubkey, store, "wss://r.no-key", undefined);
|
||||
const noKey = readChannelMutesOutboxPreservedKey(pubkey, "wss://r.no-key");
|
||||
assert.equal(
|
||||
noKey,
|
||||
undefined,
|
||||
"absent preservedKey must read back as undefined",
|
||||
);
|
||||
} finally {
|
||||
restore();
|
||||
}
|
||||
});
|
||||
|
||||
@@ -67,6 +67,14 @@ export class ChannelMuteSyncManager extends MergeLaneSyncManager<ChannelMuteStor
|
||||
this.publish(store, preservedKey);
|
||||
}
|
||||
|
||||
/**
|
||||
* Re-drive the current pending mute edit without resetting the preserved key
|
||||
* — used by the reconnect handler (Kalvin P3).
|
||||
*/
|
||||
retryReconnectMutesPublish(): void {
|
||||
this.retryReconnectPublish();
|
||||
}
|
||||
|
||||
/** Fetch the current remote head for this pubkey's mutes blob. */
|
||||
fetchRemoteMutes() {
|
||||
return this.fetchRemoteBlob();
|
||||
|
||||
@@ -2,6 +2,7 @@ import {
|
||||
clearOutbox,
|
||||
makeStorageKey,
|
||||
readOutbox,
|
||||
readOutboxPreservedKey,
|
||||
readStore,
|
||||
reclaimSubsumedOutbox,
|
||||
writeOutbox,
|
||||
@@ -284,8 +285,16 @@ export function writeChannelStarsOutbox(
|
||||
pubkey: string,
|
||||
store: ChannelStarStore,
|
||||
relayUrl: string,
|
||||
preservedKey?: string,
|
||||
): void {
|
||||
writeOutbox(OUTBOX_KEY_PREFIX, pubkey, store, relayUrl, boundStarStore);
|
||||
writeOutbox(
|
||||
OUTBOX_KEY_PREFIX,
|
||||
pubkey,
|
||||
store,
|
||||
relayUrl,
|
||||
boundStarStore,
|
||||
preservedKey,
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -307,6 +316,23 @@ export function readChannelStarsOutbox(
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
* Read the `preservedKey` from this window's own outbox record, or `undefined`
|
||||
* when none exists. Used on bootstrap to restore the capacity-bounding
|
||||
* reservation across remount (Kalvin P3).
|
||||
*/
|
||||
export function readChannelStarsOutboxPreservedKey(
|
||||
pubkey: string,
|
||||
relayUrl: string,
|
||||
): string | undefined {
|
||||
return readOutboxPreservedKey(
|
||||
OUTBOX_KEY_PREFIX,
|
||||
pubkey,
|
||||
relayUrl,
|
||||
parseStarPayload,
|
||||
);
|
||||
}
|
||||
|
||||
/** Clear this window's own outbox key (its edit published or is a no-op). */
|
||||
export function clearChannelStarsOutbox(
|
||||
pubkey: string,
|
||||
|
||||
@@ -229,3 +229,123 @@ test("P3: clicked channel is preserved through pre-publish mergeWithRemote at ca
|
||||
mock.reset();
|
||||
}
|
||||
});
|
||||
|
||||
// Mutation: calling public publish() on reconnect resets pendingPreservedKey,
|
||||
// letting a 501-entry pre-publish merge evict the clicked channel (P3 reconnect).
|
||||
test("P3 reconnect: retryReconnectStarsPublish preserves clicked channel key across reconnect", async () => {
|
||||
const MAX = 500;
|
||||
const clickedId = "ch-reconnect-star";
|
||||
const clickedEntry = { starred: true, updatedAt: 1, rev: 1 }; // oldest
|
||||
const localChannels = { [clickedId]: clickedEntry };
|
||||
for (let i = 0; i < MAX - 1; i++) {
|
||||
localChannels[`ch-local-${i}`] = { starred: false, updatedAt: 100, rev: 0 };
|
||||
}
|
||||
const localStore = { version: 1, channels: localChannels };
|
||||
|
||||
const remoteChannels = {};
|
||||
for (let i = 0; i < MAX - 1; i++) {
|
||||
remoteChannels[`ch-local-${i}`] = {
|
||||
starred: false,
|
||||
updatedAt: 100,
|
||||
rev: 0,
|
||||
};
|
||||
}
|
||||
remoteChannels["ch-remote-rc"] = { starred: false, updatedAt: 100, rev: 0 };
|
||||
const remoteStore = { version: 1, channels: remoteChannels };
|
||||
|
||||
const fw = makeFakeWindow();
|
||||
const restore = installFakeWindow(fw);
|
||||
const tauri = installEchoTauri("pk-p3-reconnect-stars");
|
||||
const remoteHead = tauri.mintHead(remoteStore, 50, "evt-rc-stars");
|
||||
remoteHead.tags = [["d", "channel-stars"]];
|
||||
|
||||
let fetchCalls = 0;
|
||||
mock.method(relayClient, "fetchEvents", () => {
|
||||
fetchCalls++;
|
||||
if (fetchCalls === 1) return Promise.resolve([remoteHead]);
|
||||
return Promise.resolve([]);
|
||||
});
|
||||
let publishedEvent = null;
|
||||
mock.method(relayClient, "publishEvent", (evt) => {
|
||||
publishedEvent = evt;
|
||||
return Promise.resolve();
|
||||
});
|
||||
|
||||
try {
|
||||
const m = new ChannelStarSyncManager("pk-p3-reconnect-stars", RELAY);
|
||||
m.publishStars(localStore, clickedId);
|
||||
m.cancelPendingStarPublish();
|
||||
fetchCalls = 0;
|
||||
publishedEvent = null;
|
||||
// Re-publish to set pendingPreservedKey, then retry via reconnect path.
|
||||
m.publishStars(localStore, clickedId);
|
||||
m.retryReconnectStarsPublish();
|
||||
fw._fireTimer();
|
||||
await new Promise((r) => setTimeout(r, 30));
|
||||
assert.ok(publishedEvent !== null, "publish must have fired after retry");
|
||||
const plaintext = tauri.capturedPlaintext();
|
||||
const published = JSON.parse(plaintext);
|
||||
assert.ok(
|
||||
clickedId in published.channels,
|
||||
`clicked channel must survive 501-entry merge after retryReconnectStarsPublish`,
|
||||
);
|
||||
m.destroy();
|
||||
} finally {
|
||||
tauri.restore();
|
||||
restore();
|
||||
mock.reset();
|
||||
}
|
||||
});
|
||||
|
||||
// Mutation: omitting preservedKey from writeChannelStarsOutbox means it is not
|
||||
// persisted in the envelope, so readChannelStarsOutboxPreservedKey returns
|
||||
// undefined and bootstrap replay calls publishStars(outbox) without protection
|
||||
// — a 501-entry pre-publish merge can evict the clicked channel on remount (P3).
|
||||
test("P3 remount: writeChannelStarsOutbox persists preservedKey; readChannelStarsOutboxPreservedKey recovers it", async () => {
|
||||
const {
|
||||
readChannelStarsOutbox,
|
||||
readChannelStarsOutboxPreservedKey,
|
||||
writeChannelStarsOutbox,
|
||||
} = await import("./channelStarsStorage.ts");
|
||||
|
||||
const fw = makeFakeWindow();
|
||||
const restore = installFakeWindow(fw);
|
||||
|
||||
const pubkey = "pk-p3-remount-storage";
|
||||
const relayUrl = "wss://r.remount-storage";
|
||||
const clickedId = "ch-remount-storage";
|
||||
const store = {
|
||||
version: 1,
|
||||
channels: { [clickedId]: { starred: true, updatedAt: 1, rev: 0 } },
|
||||
};
|
||||
|
||||
try {
|
||||
// Write WITH preservedKey — models a click that calls publishStars(store, clickedId).
|
||||
writeChannelStarsOutbox(pubkey, store, relayUrl, clickedId);
|
||||
|
||||
// The merged outbox must be present.
|
||||
assert.ok(
|
||||
readChannelStarsOutbox(pubkey, relayUrl) !== null,
|
||||
"outbox must be written",
|
||||
);
|
||||
|
||||
// The preserved key must round-trip through the storage envelope.
|
||||
const restoredKey = readChannelStarsOutboxPreservedKey(pubkey, relayUrl);
|
||||
assert.equal(
|
||||
restoredKey,
|
||||
clickedId,
|
||||
"preservedKey must be readable back from the written outbox record",
|
||||
);
|
||||
|
||||
// Writing WITHOUT preservedKey (old behaviour / other-window record) returns undefined.
|
||||
writeChannelStarsOutbox(pubkey, store, "wss://r.no-key", undefined);
|
||||
const noKey = readChannelStarsOutboxPreservedKey(pubkey, "wss://r.no-key");
|
||||
assert.equal(
|
||||
noKey,
|
||||
undefined,
|
||||
"absent preservedKey must read back as undefined",
|
||||
);
|
||||
} finally {
|
||||
restore();
|
||||
}
|
||||
});
|
||||
|
||||
@@ -67,6 +67,14 @@ export class ChannelStarSyncManager extends MergeLaneSyncManager<ChannelStarStor
|
||||
this.publish(store, preservedKey);
|
||||
}
|
||||
|
||||
/**
|
||||
* Re-drive the current pending star edit without resetting the preserved key
|
||||
* — used by the reconnect handler (Kalvin P3).
|
||||
*/
|
||||
retryReconnectStarsPublish(): void {
|
||||
this.retryReconnectPublish();
|
||||
}
|
||||
|
||||
/** Fetch the current remote head for this pubkey's stars blob. */
|
||||
fetchRemoteStars() {
|
||||
return this.fetchRemoteBlob();
|
||||
|
||||
@@ -158,8 +158,16 @@ export function writeOutbox<S>(
|
||||
store: S,
|
||||
relayUrl: string,
|
||||
bound: (s: S) => S,
|
||||
preservedKey?: string,
|
||||
): void {
|
||||
writeOwnOutbox(outboxPrefix, pubkey, relayUrl, bound(store));
|
||||
writeOwnOutbox(
|
||||
outboxPrefix,
|
||||
pubkey,
|
||||
relayUrl,
|
||||
bound(store),
|
||||
undefined,
|
||||
preservedKey,
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -195,6 +203,34 @@ export function clearOutbox(
|
||||
clearOwnOutbox(outboxPrefix, pubkey, relayUrl);
|
||||
}
|
||||
|
||||
/**
|
||||
* Read the `preservedKey` stored by the most recent click in THIS window's own
|
||||
* outbox record, or `undefined` when no own record exists or it carries none.
|
||||
*
|
||||
* The `preservedKey` identifies the channel that must survive capacity bounding
|
||||
* across a remount. It is intentionally read from this window's own record
|
||||
* only — merging multiple windows' records into one store loses the per-click
|
||||
* provenance, so we recover protection from the authoritative own-window entry.
|
||||
*/
|
||||
export function readOutboxPreservedKey(
|
||||
outboxPrefix: string,
|
||||
pubkey: string,
|
||||
relayUrl: string,
|
||||
parseStore: (json: unknown) => unknown,
|
||||
): string | undefined {
|
||||
const records = enumerateOutbox(
|
||||
outboxPrefix,
|
||||
legacyOutboxKey(outboxPrefix, pubkey, relayUrl),
|
||||
pubkey,
|
||||
relayUrl,
|
||||
parseStore,
|
||||
);
|
||||
// Return the preservedKey from this window's own record (if any). Own records
|
||||
// are the only ones that carry a preservedKey written by this session.
|
||||
const own = records.find((r) => r.isOwn);
|
||||
return own?.preservedKey;
|
||||
}
|
||||
|
||||
/**
|
||||
* Reclaim foreign outbox keys the fetched relay head already subsumes: a
|
||||
* record is redundant when merging it into `head` yields `head` unchanged.
|
||||
|
||||
@@ -78,7 +78,12 @@ export type MergeLaneConfig<S> = {
|
||||
store: S,
|
||||
) => void;
|
||||
/** Persist this window's pending edit to the durable outbox. */
|
||||
writeOutbox: (pubkey: string, store: S, relayUrl: string) => void;
|
||||
writeOutbox: (
|
||||
pubkey: string,
|
||||
store: S,
|
||||
relayUrl: string,
|
||||
preservedKey?: string,
|
||||
) => void;
|
||||
/** Clear this window's own outbox key. */
|
||||
clearOutbox: (pubkey: string, relayUrl: string) => void;
|
||||
/** True when the local store is non-empty and eligible for a first-sync seed-publish. */
|
||||
@@ -211,6 +216,20 @@ export class MergeLaneSyncManager<S> {
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Re-drive the currently pending edit without opening a new generation —
|
||||
* used by the reconnect handler so `pendingPreservedKey` is not reset (Kalvin
|
||||
* P3). Calling the public `publish()` on reconnect would bump the generation
|
||||
* and clear the preserved key, so a subsequent 501-entry pre-publish merge
|
||||
* could evict the clicked channel again. Waking the existing cycle instead
|
||||
* keeps the frozen key. No-op when nothing is pending.
|
||||
*/
|
||||
retryReconnectPublish(): void {
|
||||
if (this.pendingStore === null) return;
|
||||
this.cancelPendingPublish();
|
||||
this.startCycle();
|
||||
}
|
||||
|
||||
getPendingStore(): S | null {
|
||||
return this.pendingStore;
|
||||
}
|
||||
@@ -225,8 +244,9 @@ export class MergeLaneSyncManager<S> {
|
||||
// Persist synchronously so a click made <2s before quit/community-switch
|
||||
// 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.
|
||||
this.config.writeOutbox(this.pubkey, store, this.relayUrl);
|
||||
// setItem, never a shared-key read-modify-write. Pass preservedKey so the
|
||||
// capacity-bounding reservation survives remount (Kalvin P3).
|
||||
this.config.writeOutbox(this.pubkey, store, this.relayUrl, preservedKey);
|
||||
if (this.debounceTimer !== null) {
|
||||
window.clearTimeout(this.debounceTimer);
|
||||
}
|
||||
@@ -505,5 +525,6 @@ export class MergeLaneSyncManager<S> {
|
||||
this.destroyed = true;
|
||||
this.cancelPendingPublish();
|
||||
this.pendingStore = null;
|
||||
this.pendingPreservedKey = undefined;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -195,6 +195,11 @@ export type OutboxRecord<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 channelId that was preserved through the capacity bounding step when
|
||||
// this edit was queued (Kalvin P3). Present on v2 records written after this
|
||||
// field was introduced; absent on legacy records. Used on bootstrap outbox
|
||||
// replay to re-protect the same channel across remount.
|
||||
preservedKey?: string;
|
||||
isOwn: boolean;
|
||||
};
|
||||
|
||||
@@ -206,7 +211,7 @@ export type OutboxRecord<T> = {
|
||||
function parseEnvelope<T>(
|
||||
raw: string | null,
|
||||
parseStore: (json: unknown) => T | null,
|
||||
): { store: T; queuedAt: number } | null {
|
||||
): { store: T; queuedAt: number; preservedKey?: string } | null {
|
||||
if (!raw) return null;
|
||||
try {
|
||||
const json = JSON.parse(raw);
|
||||
@@ -216,14 +221,20 @@ function parseEnvelope<T>(
|
||||
!Array.isArray(json) &&
|
||||
"store" in (json as Record<string, unknown>)
|
||||
) {
|
||||
const env = json as { store: unknown; queuedAt?: unknown };
|
||||
const env = json as {
|
||||
store: unknown;
|
||||
queuedAt?: unknown;
|
||||
preservedKey?: unknown;
|
||||
};
|
||||
const store = parseStore(env.store);
|
||||
if (!store) return null;
|
||||
const queuedAt =
|
||||
typeof env.queuedAt === "number" && Number.isFinite(env.queuedAt)
|
||||
? env.queuedAt
|
||||
: 0;
|
||||
return { store, queuedAt };
|
||||
const preservedKey =
|
||||
typeof env.preservedKey === "string" ? env.preservedKey : undefined;
|
||||
return { store, queuedAt, preservedKey };
|
||||
}
|
||||
// Legacy bare-store shape from a pre-envelope build.
|
||||
const store = parseStore(json);
|
||||
@@ -260,15 +271,15 @@ export function writeOwnOutbox(
|
||||
relayUrl: string,
|
||||
store: unknown,
|
||||
nowSecs: number = Math.floor(Date.now() / 1_000),
|
||||
preservedKey?: string,
|
||||
): boolean {
|
||||
const base = ownKeyBase(prefix, pubkey, relayUrl);
|
||||
let key: string;
|
||||
try {
|
||||
key = nextOwnKey(base);
|
||||
window.localStorage.setItem(
|
||||
key,
|
||||
JSON.stringify({ store, queuedAt: nowSecs }),
|
||||
);
|
||||
const envelope: Record<string, unknown> = { store, queuedAt: nowSecs };
|
||||
if (preservedKey !== undefined) envelope.preservedKey = preservedKey;
|
||||
window.localStorage.setItem(key, JSON.stringify(envelope));
|
||||
} catch {
|
||||
// The fresh key could not be persisted — the intent is NOT durably held.
|
||||
return false;
|
||||
@@ -338,6 +349,7 @@ export function enumerateOutbox<T>(
|
||||
store: parsed.store,
|
||||
raw,
|
||||
queuedAt: parsed.queuedAt,
|
||||
preservedKey: parsed.preservedKey,
|
||||
isOwn: key.startsWith(ownPrefix),
|
||||
});
|
||||
}
|
||||
@@ -350,6 +362,7 @@ export function enumerateOutbox<T>(
|
||||
store: legacy.store,
|
||||
raw: legacyRaw,
|
||||
queuedAt: legacy.queuedAt,
|
||||
preservedKey: legacy.preservedKey,
|
||||
isOwn: false,
|
||||
});
|
||||
}
|
||||
|
||||
@@ -9,6 +9,7 @@ import {
|
||||
mergeStores,
|
||||
mutedChannelIdsFromStore,
|
||||
readChannelMutesOutbox,
|
||||
readChannelMutesOutboxPreservedKey,
|
||||
readChannelMutesStore,
|
||||
reclaimSubsumedMutesOutbox,
|
||||
storageKey,
|
||||
@@ -126,7 +127,15 @@ export function useChannelMutes(
|
||||
result.action === "apply-remote" &&
|
||||
isMutesStoreSubsumedBy(outbox, result.data.store);
|
||||
if (!subsumed) {
|
||||
managerRef.current?.publishMutes(outbox);
|
||||
// Restore the preservedKey written alongside this window's outbox
|
||||
// record so the clicked channel's capacity reservation survives
|
||||
// remount (Kalvin P3). Old records without the field produce
|
||||
// undefined, which publishes without a reservation — same as before.
|
||||
const preservedKey = readChannelMutesOutboxPreservedKey(
|
||||
pubkey,
|
||||
relayUrl,
|
||||
);
|
||||
managerRef.current?.publishMutes(outbox, preservedKey);
|
||||
}
|
||||
} else {
|
||||
clearChannelMutesOutbox(pubkey, relayUrl);
|
||||
@@ -227,10 +236,11 @@ export function useChannelMutes(
|
||||
if (result.status === "found") {
|
||||
setStore(applyRemote(result.data));
|
||||
}
|
||||
const pending = managerRef.current?.getPendingMuteStore();
|
||||
if (pending) {
|
||||
managerRef.current?.publishMutes(pending);
|
||||
}
|
||||
// Re-drive the existing generation rather than opening a new one so
|
||||
// pendingPreservedKey is not reset — a new publish() call would clear
|
||||
// the key and let a 501-entry pre-publish merge evict the clicked
|
||||
// channel (Kalvin P3).
|
||||
managerRef.current?.retryReconnectMutesPublish();
|
||||
});
|
||||
});
|
||||
return () => {
|
||||
|
||||
@@ -163,15 +163,22 @@ export function useChannelSections(
|
||||
// to supersede is consumed into pending here and can never be GC'd out.
|
||||
const outbox = readChannelSectionsOutbox(pubkey, relayUrl);
|
||||
if (outbox) {
|
||||
// When bootstrap fetched and applied a remote head, only replay an
|
||||
// outbox edit that was queued AFTER that head — an edit queued before
|
||||
// the applied head was authored without knowledge of that state and has
|
||||
// already lost whole-blob LWW. Legacy records (queuedAt=0) are treated
|
||||
// as pre-head under any non-zero head. A `hold` (absent/failed fetch)
|
||||
// carries no confirmed head, so we always replay.
|
||||
const headCreatedAt =
|
||||
result.action === "apply-remote" ? result.data.createdAt : 0;
|
||||
if (outbox.queuedAt > headCreatedAt) {
|
||||
// When bootstrap fetched and applied a remote head, suppress replay
|
||||
// only for an outbox edit that is STRICTLY older than that head:
|
||||
// `queuedAt < appliedHead.createdAt`. Equality replays — a same-second
|
||||
// edit was in-flight with the head and must not be silently swallowed —
|
||||
// and this keeps the guard aligned with reclamation's strict `<` so an
|
||||
// equal-second edit is neither stranded on disk nor double-suppressed.
|
||||
// A `hold` (absent/failed fetch) carries no confirmed applied head so
|
||||
// we always replay regardless of queuedAt. Note: `queuedAt` and relay
|
||||
// `created_at` are independent device wall clocks; a slow-clocked
|
||||
// device can produce a queuedAt that appears to predate a head it never
|
||||
// saw — this is an accepted residual of the LWW max-merge strategy the
|
||||
// feature already relies on.
|
||||
const shouldReplay =
|
||||
result.action !== "apply-remote" ||
|
||||
outbox.queuedAt >= result.data.createdAt;
|
||||
if (shouldReplay) {
|
||||
// publishSections synchronously copies the intent into this window's
|
||||
// own v2 key and returns whether that transfer is durable. Mark the
|
||||
// legacy blob consumed ONLY when it is: a `setItem` failure (quota)
|
||||
@@ -191,8 +198,8 @@ export function useChannelSections(
|
||||
);
|
||||
}
|
||||
}
|
||||
// Stale outbox (queuedAt ≤ headCreatedAt) is left for reclamation
|
||||
// below — reclaimSupersededSectionsOutbox will clean it up.
|
||||
// Stale outbox (queuedAt < appliedHead.createdAt) is left for
|
||||
// reclamation below — reclaimSupersededSectionsOutbox will clean it up.
|
||||
} else {
|
||||
clearChannelSectionsOutbox(pubkey, relayUrl);
|
||||
}
|
||||
|
||||
@@ -160,15 +160,22 @@ export function useChannelSortPreference(
|
||||
// consumed into pending here and can never be GC'd out.
|
||||
const outbox = readChannelSortOutbox(pubkey, relayUrl);
|
||||
if (outbox) {
|
||||
// When bootstrap fetched and applied a remote head, only replay an
|
||||
// outbox edit that was queued AFTER that head — an edit queued before
|
||||
// the applied head was authored without knowledge of that state and has
|
||||
// already lost whole-blob LWW. Legacy records (queuedAt=0) are treated
|
||||
// as pre-head under any non-zero head. A `hold` (absent/failed fetch)
|
||||
// carries no confirmed head, so we always replay.
|
||||
const headCreatedAt =
|
||||
result.action === "apply-remote" ? result.data.createdAt : 0;
|
||||
if (outbox.queuedAt > headCreatedAt) {
|
||||
// When bootstrap fetched and applied a remote head, suppress replay
|
||||
// only for an outbox edit that is STRICTLY older than that head:
|
||||
// `queuedAt < appliedHead.createdAt`. Equality replays — a same-second
|
||||
// edit was in-flight with the head and must not be silently swallowed —
|
||||
// and this keeps the guard aligned with reclamation's strict `<` so an
|
||||
// equal-second edit is neither stranded on disk nor double-suppressed.
|
||||
// A `hold` (absent/failed fetch) carries no confirmed applied head so
|
||||
// we always replay regardless of queuedAt. Note: `queuedAt` and relay
|
||||
// `created_at` are independent device wall clocks; a slow-clocked
|
||||
// device can produce a queuedAt that appears to predate a head it never
|
||||
// saw — this is an accepted residual of the LWW max-merge strategy the
|
||||
// feature already relies on.
|
||||
const shouldReplay =
|
||||
result.action !== "apply-remote" ||
|
||||
outbox.queuedAt >= result.data.createdAt;
|
||||
if (shouldReplay) {
|
||||
// publishSortPrefs synchronously copies the intent into this window's
|
||||
// own v2 key and returns whether that transfer is durable. Mark the
|
||||
// legacy blob consumed ONLY when it is: a `setItem` failure (quota)
|
||||
@@ -188,8 +195,8 @@ export function useChannelSortPreference(
|
||||
);
|
||||
}
|
||||
}
|
||||
// Stale outbox (queuedAt ≤ headCreatedAt) is left for reclamation
|
||||
// below — reclaimSupersededSortOutbox will clean it up.
|
||||
// Stale outbox (queuedAt < appliedHead.createdAt) is left for
|
||||
// reclamation below — reclaimSupersededSortOutbox will clean it up.
|
||||
} else {
|
||||
clearChannelSortOutbox(pubkey, relayUrl);
|
||||
}
|
||||
|
||||
@@ -8,6 +8,7 @@ import {
|
||||
isStarsStoreSubsumedBy,
|
||||
mergeStores,
|
||||
readChannelStarsOutbox,
|
||||
readChannelStarsOutboxPreservedKey,
|
||||
readChannelStarsStore,
|
||||
reclaimSubsumedStarsOutbox,
|
||||
starredChannelIdsFromStore,
|
||||
@@ -127,7 +128,15 @@ export function useChannelStars(
|
||||
result.action === "apply-remote" &&
|
||||
isStarsStoreSubsumedBy(outbox, result.data.store);
|
||||
if (!subsumed) {
|
||||
managerRef.current?.publishStars(outbox);
|
||||
// Restore the preservedKey written alongside this window's outbox
|
||||
// record so the clicked channel's capacity reservation survives
|
||||
// remount (Kalvin P3). Old records without the field produce
|
||||
// undefined, which publishes without a reservation — same as before.
|
||||
const preservedKey = readChannelStarsOutboxPreservedKey(
|
||||
pubkey,
|
||||
relayUrl,
|
||||
);
|
||||
managerRef.current?.publishStars(outbox, preservedKey);
|
||||
}
|
||||
} else {
|
||||
clearChannelStarsOutbox(pubkey, relayUrl);
|
||||
@@ -228,10 +237,11 @@ export function useChannelStars(
|
||||
if (result.status === "found") {
|
||||
setStore(applyRemote(result.data));
|
||||
}
|
||||
const pending = managerRef.current?.getPendingStarStore();
|
||||
if (pending) {
|
||||
managerRef.current?.publishStars(pending);
|
||||
}
|
||||
// Re-drive the existing generation rather than opening a new one so
|
||||
// pendingPreservedKey is not reset — a new publish() call would clear
|
||||
// the key and let a 501-entry pre-publish merge evict the clicked
|
||||
// channel (Kalvin P3).
|
||||
managerRef.current?.retryReconnectStarsPublish();
|
||||
});
|
||||
});
|
||||
return () => {
|
||||
|
||||
@@ -257,48 +257,112 @@ export function runWholeBlobHookSuite({
|
||||
}
|
||||
});
|
||||
|
||||
// Mutation: removing queuedAt > headCreatedAt guard in the hook replays the
|
||||
// stale outbox, overwriting device-B's t=200 state with a t=100 edit.
|
||||
test(`${label}: stale outbox record (queuedAt < bootstrap head) is NOT replayed`, async () => {
|
||||
// Mutation: removing queuedAt >= appliedHead.createdAt guard in the hook
|
||||
// replays the stale outbox above the applied head, overwriting device-B's
|
||||
// state. All four sub-cases advance the real 2s debounce via makeHookTimerBed
|
||||
// so the assertion is causal: the unfixed code fires publishEvent after the
|
||||
// debounce while the fixed code does not (strict <) or does (hold / equality).
|
||||
test(`${label}: P1 bootstrap outbox replay gate — strict < suppresses, equality/hold/legacy replays`, async () => {
|
||||
const { makeHookTimerBed } = await import("./sidebarSyncTestHelpers.mjs");
|
||||
const { act, cleanup, renderHook } = await import("@testing-library/react");
|
||||
const { relayClient } = await import("@/shared/api/relayClient");
|
||||
const pubkey = `pk-${label}-sr`;
|
||||
const relayUrl = `wss://r.${label}-sr`;
|
||||
// Relay head at t=200 (device B's state).
|
||||
const head = {
|
||||
pubkey,
|
||||
content: "good-cipher",
|
||||
created_at: 200,
|
||||
id: "evt-head-200",
|
||||
};
|
||||
const publishCalls = [];
|
||||
const restoreRelay = stubRelay(relayClient, { publishCalls });
|
||||
relayClient.fetchEvents = async () => [head];
|
||||
const restoreTauri = stubTauri(pubkey, () => makeRemotePayload());
|
||||
// Pre-populate a stale legacy outbox record queued at t=100 (before the head).
|
||||
const legKey = legacyOutboxKey(pubkey, relayUrl);
|
||||
const staleRaw = JSON.stringify({
|
||||
store: JSON.parse(makeRemotePayload()),
|
||||
queuedAt: 100,
|
||||
});
|
||||
window.localStorage.setItem(legKey, staleRaw);
|
||||
let hook = null;
|
||||
try {
|
||||
await act(async () => {
|
||||
hook = renderHook(() => useHook(pubkey, relayUrl));
|
||||
for (let i = 0; i < 6; i++) await Promise.resolve();
|
||||
});
|
||||
assert.equal(
|
||||
publishCalls.length,
|
||||
0,
|
||||
`${label}: stale outbox record (queuedAt=100 < head=200) must not be replayed`,
|
||||
|
||||
// Helper: mount the hook with a legacy outbox record at queuedAt and an
|
||||
// optional relay head, fire the 2s debounce, return publishEvent call count.
|
||||
async function runCase({ queuedAt, headCreatedAt, noHead = false }) {
|
||||
const bed = makeHookTimerBed();
|
||||
const pubkey = `pk-${label}-p1-${queuedAt}-${headCreatedAt}`;
|
||||
const relayUrl = `wss://r.${label}-p1`;
|
||||
// Write via the legacy key — recognised by the outbox reader for both
|
||||
// sections and sort hooks, and queuedAt is explicit in the envelope.
|
||||
const legKey = legacyOutboxKey(pubkey, relayUrl);
|
||||
window.localStorage.setItem(
|
||||
legKey,
|
||||
JSON.stringify({ store: JSON.parse(makeRemotePayload()), queuedAt }),
|
||||
);
|
||||
|
||||
const publishCalls = [];
|
||||
const restoreRelay = stubRelay(relayClient, { publishCalls });
|
||||
const origFetch = relayClient.fetchEvents;
|
||||
relayClient.fetchEvents = async () => {
|
||||
if (noHead) return [];
|
||||
return [
|
||||
{
|
||||
pubkey,
|
||||
content: "good-cipher",
|
||||
created_at: headCreatedAt,
|
||||
id: `evt-${headCreatedAt}`,
|
||||
},
|
||||
];
|
||||
};
|
||||
const restoreTauri = stubTauri(pubkey, () => makeRemotePayload());
|
||||
let hook = null;
|
||||
try {
|
||||
await act(async () => {
|
||||
hook = renderHook(() => useHook(pubkey, relayUrl));
|
||||
for (let i = 0; i < 8; i++) await Promise.resolve();
|
||||
});
|
||||
// Whether publish() was called is observable via the 2s debounce timer:
|
||||
// publish() schedules the timer synchronously. hasDelay(2000) is true
|
||||
// iff publish() fired (replay happened); false iff the guard suppressed.
|
||||
// This is causal: without the guard, the timer would be present for all
|
||||
// cases; with it, only the non-suppressed cases schedule the timer.
|
||||
return bed.hasDelay(2000) ? 1 : 0;
|
||||
} finally {
|
||||
hook?.unmount();
|
||||
cleanup();
|
||||
window.localStorage.removeItem(legKey);
|
||||
bed.restore();
|
||||
restoreRelay();
|
||||
restoreTauri();
|
||||
relayClient.fetchEvents = origFetch;
|
||||
}
|
||||
}
|
||||
|
||||
// Case 1: strict queuedAt < headCreatedAt → suppressed (0 publish calls)
|
||||
{
|
||||
const calls = await runCase({ queuedAt: 100, headCreatedAt: 200 });
|
||||
assert.equal(
|
||||
calls,
|
||||
0,
|
||||
`${label}: stale outbox (queuedAt=100 < head=200) must NOT be replayed`,
|
||||
);
|
||||
}
|
||||
|
||||
// Case 2: queuedAt === headCreatedAt → replays (≥1 publish call)
|
||||
{
|
||||
const calls = await runCase({ queuedAt: 200, headCreatedAt: 200 });
|
||||
assert.ok(
|
||||
calls >= 1,
|
||||
`${label}: same-second outbox (queuedAt=200 === head=200) MUST replay`,
|
||||
);
|
||||
}
|
||||
|
||||
// Case 3: hold path (no relay head) with a v2 queuedAt > 0 → always replays
|
||||
{
|
||||
const calls = await runCase({
|
||||
queuedAt: 50,
|
||||
headCreatedAt: 0,
|
||||
noHead: true,
|
||||
});
|
||||
assert.ok(
|
||||
calls >= 1,
|
||||
`${label}: hold path (no relay head) MUST replay outbox regardless of queuedAt`,
|
||||
);
|
||||
}
|
||||
|
||||
// Case 4: hold path with legacy queuedAt=0 → always replays (was silently
|
||||
// stranded by the old guard: 0 > 0 was false on hold)
|
||||
{
|
||||
const calls = await runCase({
|
||||
queuedAt: 0,
|
||||
headCreatedAt: 0,
|
||||
noHead: true,
|
||||
});
|
||||
assert.ok(
|
||||
calls >= 1,
|
||||
`${label}: hold path with legacy queuedAt=0 MUST replay (was stranded by old 0>0 guard)`,
|
||||
);
|
||||
hook.unmount();
|
||||
} finally {
|
||||
cleanup();
|
||||
window.localStorage.removeItem(legKey);
|
||||
restoreRelay();
|
||||
restoreTauri();
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user