mirror of
https://github.com/block/buzz.git
synced 2026-10-29 05:21:31 +00:00
fix(sidebar-sync): address Carl round — P1/P2a-1/P2b/P3 data-loss paths
P1 — restored-outbox replay re-mints queuedAt and triggers the failed-bootstrap exception designed only for fresh in-session clicks. Hook-replayed outbox calls now pass isRestoredReplay=true through publishSections/publishSortPrefs → publish(), which suppresses the exception (pendingIsRestoredReplay) and does not re-freeze publishBaseline from mutable lastRemoteHead. P2a-1 — hook .then() replay after releaseDeferred establishes the correct baseline called publish() unconditionally, overwriting the baseline with the mutable lastRemoteHead (which by then could carry a suppressed live peer head). The isRestoredReplay flag added for P1 also closes this path: restored replays skip the publishBaseline re-freeze so the baseline releaseDeferred established is preserved through the callback. P2b — fetchRemoteBlob (periodic/reconnect path) called recordRemoteHead before decryptAndParse, leaving a decrypt-gap window where a concurrent click could freeze its publishBaseline against the pre-decrypt head. Moved recordRemoteHead to after decryptAndParse succeeds, mirroring the existing subscribeLive fix. P3 — isStarsStoreSubsumedBy / isMutesStoreSubsumedBy proved subsumption via a capacity-bounded mergeStores(head, candidate) without a preservedKey. At the 500-entry cap the clicked channel (lexicographically smallest) could be evicted during the proof merge, certifying retention of a click the relay never kept. Two consumers affected: bootstrap suppression (hook files pass outboxMeta.preservedKey to the subsumption check) and ACK confirmation (mergeLaneSyncManager passes this.pendingPreservedKey to config.isSubsumedBy). Both storage functions now take an optional preservedKey and use a direct membership check when present: head must contain the preserved channel with rev >= candidate's before the rest of the store is verified. Regressions — hook-layer tests in new shared suites: - wholeBlobSyncCarl.shared.test.mjs: P1 (failed-bootstrap replay adopts newer retained head), P2a-1 (H102 survives hook replay as genuine advance), P2b (fetchRemoteBlob decrypt gap). Wired from useChannelSections and useChannelSortPreference test files. - mergeLaneSyncCarl.shared.test.mjs: P3-bootstrap (500-entry head does not subsume click via eviction) and P3-ack (confirmRetainedHeadSubsumes retains outbox when relay head omits clicked channel). Wired from useChannelStars and useChannelMutes test files. All 47 targeted tests pass (43 pre-existing + 4 new Carl-round P3 tests plus 6 new P1/P2a-1/P2b tests). T1-T6 manager-level contracts unchanged. Co-authored-by: Will Pfleger <pfleger.will@gmail.com> Signed-off-by: Will Pfleger <pfleger.will@gmail.com>
This commit is contained in:
@@ -338,11 +338,38 @@ export function clearChannelMutesOutbox(
|
||||
* both to reclaim a subsumed foreign key and to skip a redundant boot-time
|
||||
* replay publish of a fold the head already carries (e.g. only the
|
||||
* never-deleted legacy key lingers).
|
||||
*
|
||||
* When `preservedKey` is provided the check is precise: `head` must contain an
|
||||
* entry for `preservedKey` with `rev >= candidate.channels[preservedKey].rev`.
|
||||
* This avoids using capacity-bounded `mergeStores` as the subsumption proof:
|
||||
* without a preserved key, mergeStores can evict the clicked entry at the
|
||||
* 500-cap boundary and return the truncated head unchanged, incorrectly
|
||||
* certifying retention of a click the relay never kept (Carl P3).
|
||||
*
|
||||
* When `preservedKey` is not provided (no click reservation active) the
|
||||
* bounded merge proof is still semantically safe — no specific entry needs to
|
||||
* survive eviction — so we fall back to the merge-equality check.
|
||||
*/
|
||||
export function isMutesStoreSubsumedBy(
|
||||
candidate: ChannelMuteStore,
|
||||
head: ChannelMuteStore,
|
||||
preservedKey?: string,
|
||||
): boolean {
|
||||
if (preservedKey !== undefined) {
|
||||
const candidateEntry = candidate.channels[preservedKey];
|
||||
if (candidateEntry === undefined) {
|
||||
return muteStoresEqual(mergeStores(head, candidate), head);
|
||||
}
|
||||
const headEntry = head.channels[preservedKey];
|
||||
if (!headEntry) return false;
|
||||
if (headEntry.updatedAt < candidateEntry.updatedAt) return false;
|
||||
if (
|
||||
headEntry.updatedAt === candidateEntry.updatedAt &&
|
||||
headEntry.rev < candidateEntry.rev
|
||||
)
|
||||
return false;
|
||||
return muteStoresEqual(mergeStores(head, candidate, preservedKey), head);
|
||||
}
|
||||
return muteStoresEqual(mergeStores(head, candidate), head);
|
||||
}
|
||||
|
||||
|
||||
@@ -57,8 +57,11 @@ export class ChannelSectionSyncManager extends WholeBlobSyncManager<ChannelSecti
|
||||
}
|
||||
|
||||
/** Publish a sections store, debounced to 2s. Returns whether the intent is durably held. */
|
||||
publishSections(store: ChannelSectionStore): boolean {
|
||||
return this.publish(store);
|
||||
publishSections(
|
||||
store: ChannelSectionStore,
|
||||
isRestoredReplay = false,
|
||||
): boolean {
|
||||
return this.publish(store, isRestoredReplay);
|
||||
}
|
||||
|
||||
/** Fetch the current remote head for this pubkey's sections blob. */
|
||||
|
||||
@@ -38,8 +38,8 @@ export class ChannelSortSyncManager extends WholeBlobSyncManager<ChannelSortStor
|
||||
}
|
||||
|
||||
/** Publish a sort prefs store, debounced to 2s. Returns whether the intent is durably held. */
|
||||
publishSortPrefs(store: ChannelSortStore): boolean {
|
||||
return this.publish(store);
|
||||
publishSortPrefs(store: ChannelSortStore, isRestoredReplay = false): boolean {
|
||||
return this.publish(store, isRestoredReplay);
|
||||
}
|
||||
|
||||
/** Fetch the current remote head for this pubkey's sort-prefs blob. */
|
||||
|
||||
@@ -340,11 +340,46 @@ export function clearChannelStarsOutbox(
|
||||
* both to reclaim a subsumed foreign key and to skip a redundant boot-time
|
||||
* replay publish of a fold the head already carries (e.g. only the
|
||||
* never-deleted legacy key lingers).
|
||||
*
|
||||
* When `preservedKey` is provided the check is precise: `head` must contain an
|
||||
* entry for `preservedKey` with `rev >= candidate.channels[preservedKey].rev`.
|
||||
* This avoids using capacity-bounded `mergeStores` as the subsumption proof:
|
||||
* without a preserved key, mergeStores can evict the clicked entry at the
|
||||
* 500-cap boundary and return the truncated head unchanged, incorrectly
|
||||
* certifying retention of a click the relay never kept (Carl P3).
|
||||
*
|
||||
* When `preservedKey` is not provided (no click reservation active) the
|
||||
* bounded merge proof is still semantically safe — no specific entry needs to
|
||||
* survive eviction — so we fall back to the merge-equality check.
|
||||
*/
|
||||
export function isStarsStoreSubsumedBy(
|
||||
candidate: ChannelStarStore,
|
||||
head: ChannelStarStore,
|
||||
preservedKey?: string,
|
||||
): boolean {
|
||||
if (preservedKey !== undefined) {
|
||||
// Direct membership proof: head must carry the preserved channel with a
|
||||
// revision at least as high as the candidate's. If head is missing the
|
||||
// entry entirely, it cannot subsume the click regardless of other entries.
|
||||
const candidateEntry = candidate.channels[preservedKey];
|
||||
if (candidateEntry === undefined) {
|
||||
// Candidate has no entry for the key — subsumption reduces to the
|
||||
// boundless equality check (no click to protect).
|
||||
return starStoresEqual(mergeStores(head, candidate), head);
|
||||
}
|
||||
const headEntry = head.channels[preservedKey];
|
||||
if (!headEntry) return false;
|
||||
if (headEntry.updatedAt < candidateEntry.updatedAt) return false;
|
||||
if (
|
||||
headEntry.updatedAt === candidateEntry.updatedAt &&
|
||||
headEntry.rev < candidateEntry.rev
|
||||
)
|
||||
return false;
|
||||
// Head has the preserved entry at or above the candidate's revision.
|
||||
// Now verify the rest of the candidate is also subsumed, using a
|
||||
// preservedKey-aware merge so no other eviction can mask a real difference.
|
||||
return starStoresEqual(mergeStores(head, candidate, preservedKey), head);
|
||||
}
|
||||
return starStoresEqual(mergeStores(head, candidate), head);
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,313 @@
|
||||
// Carl-round P3 regression suite for merge-lane hooks (useChannelStars, useChannelMutes).
|
||||
//
|
||||
// Carl P3: isStarsStoreSubsumedBy (and mutes twin) calls mergeStores(head, candidate)
|
||||
// WITHOUT preservedKey. At the 500-cap, mergeStores can evict the clicked entry X and
|
||||
// return a store equal to head — "proving" subsumption even though X was never retained.
|
||||
// Two consumers of this proof are fixed:
|
||||
//
|
||||
// P3-bootstrap: bootstrap suppression (useChannelStars.ts:126-128 / useChannelMutes.ts)
|
||||
// skips the publish when isStarsStoreSubsumedBy returns true. Without preservedKey,
|
||||
// a full-head bootstrap with 500 entries + candidate containing clicked X (501 total)
|
||||
// → mergeStores evicts X → returns head → "subsumed" → publish skipped → X lost.
|
||||
// Fix: pass outboxMeta.preservedKey to isStarsStoreSubsumedBy.
|
||||
// Mutation: drop the preservedKey argument from the isStarsStoreSubsumedBy call.
|
||||
//
|
||||
// P3-ack: confirmRetainedHeadSubsumes (mergeLaneSyncManager.ts:358) calls
|
||||
// config.isSubsumedBy(store, remote.store) without pendingPreservedKey.
|
||||
// Same eviction → returns true → discardPending clears outbox → X lost.
|
||||
// Fix: pass this.pendingPreservedKey to config.isSubsumedBy.
|
||||
// Mutation: drop pendingPreservedKey from the isSubsumedBy call.
|
||||
|
||||
import assert from "node:assert/strict";
|
||||
import test, { mock } from "node:test";
|
||||
|
||||
import {
|
||||
makeHookTimerBed,
|
||||
makeHookStubs,
|
||||
installEchoTauri,
|
||||
} from "./sidebarSyncTestHelpers.mjs";
|
||||
import { relayClient } from "@/shared/api/relayClient";
|
||||
|
||||
const { stubRelay } = makeHookStubs();
|
||||
|
||||
/**
|
||||
* Register the P3 capacity-boundary subsumption regressions for a merge lane.
|
||||
*
|
||||
* @param {object} opts
|
||||
* @param {string} opts.label — "stars"|"mutes"
|
||||
* @param {number} opts.MAX_ENTRIES — capacity cap (500)
|
||||
* @param {string} opts.entryValueField — "starred"|"muted"
|
||||
* @param {string} opts.trueAction — "starChannel"|"muteChannel"
|
||||
* @param {string} opts.dTag — "channel-stars"|"channel-mutes"
|
||||
* @param {string} opts.outboxKeyPrefix — outbox localStorage prefix
|
||||
* @param {Function} opts.readStore — (pubkey, relayUrl) => store
|
||||
* @param {Function} opts.storageKey — (pubkey, relayUrl) => string
|
||||
* @param {Function} opts.useHook — the hook under test
|
||||
* @param {Function} opts.makePayload — (channels) => JSON string
|
||||
*/
|
||||
export function runMergeLaneCarlSuite({
|
||||
label,
|
||||
MAX_ENTRIES,
|
||||
entryValueField,
|
||||
trueAction,
|
||||
dTag,
|
||||
outboxKeyPrefix,
|
||||
readStore,
|
||||
storageKey,
|
||||
useHook,
|
||||
makePayload,
|
||||
}) {
|
||||
const clickedId = `ch-p3-click-${label}`;
|
||||
const relayUrl = `wss://relay-p3-carl-${label}`;
|
||||
|
||||
// P3-bootstrap: 500-entry relay head + candidate with clicked channel X
|
||||
// → isStarsStoreSubsumedBy without preservedKey evicts X → "subsumed=true"
|
||||
// → publish skipped → X lost.
|
||||
// Fix: pass outboxMeta.preservedKey → direct membership check protects X.
|
||||
// Mutation: drop preservedKey from isStarsStoreSubsumedBy call.
|
||||
test(`P3-bootstrap ${label}: 500-cap relay head must not subsume clicked channel X; bootstrap must publish to preserve X`, async () => {
|
||||
const { act, cleanup, renderHook } = await import("@testing-library/react");
|
||||
|
||||
const pubkey = `pk-p3-boot-${label}`;
|
||||
const encodedRelay = encodeURIComponent(relayUrl);
|
||||
|
||||
// Relay head: 500 entries, all at updatedAt=100, rev=0.
|
||||
// None of them is clickedId.
|
||||
const headChannels = {};
|
||||
for (let i = 0; i < MAX_ENTRIES; i++) {
|
||||
headChannels[`ch-boot-${String(i).padStart(4, "0")}`] = {
|
||||
[entryValueField]: false,
|
||||
updatedAt: 100,
|
||||
rev: 0,
|
||||
};
|
||||
}
|
||||
|
||||
// Outbox candidate: 499 of the head entries + clicked X (updatedAt=1, rev=1).
|
||||
// X is the oldest, so mergeStores(head, candidate) at cap=500 evicts X.
|
||||
const outboxChannels = {
|
||||
[clickedId]: {
|
||||
[entryValueField]: true,
|
||||
updatedAt: 1,
|
||||
rev: 1,
|
||||
},
|
||||
};
|
||||
for (let i = 0; i < MAX_ENTRIES - 1; i++) {
|
||||
outboxChannels[`ch-boot-${String(i).padStart(4, "0")}`] = {
|
||||
[entryValueField]: false,
|
||||
updatedAt: 100,
|
||||
rev: 0,
|
||||
};
|
||||
}
|
||||
|
||||
// Seed local store (matches outbox channels so bootstrap sees clicked channel)
|
||||
window.localStorage.setItem(
|
||||
storageKey(pubkey, relayUrl),
|
||||
JSON.stringify({ version: 1, channels: outboxChannels }),
|
||||
);
|
||||
|
||||
// Seed outbox: foreign nonce with preservedKey = clickedId.
|
||||
const foreignKey = `${outboxKeyPrefix}:${pubkey}:${encodedRelay}:foreign-p3-boot:0000`;
|
||||
window.localStorage.setItem(
|
||||
foreignKey,
|
||||
JSON.stringify({
|
||||
store: { version: 1, channels: outboxChannels },
|
||||
queuedAt: 1,
|
||||
preservedKey: clickedId,
|
||||
}),
|
||||
);
|
||||
|
||||
const tauri = installEchoTauri(pubkey);
|
||||
const remoteHead = tauri.mintHead(
|
||||
{ version: 1, channels: headChannels },
|
||||
50,
|
||||
`evt-p3-boot-${label}`,
|
||||
);
|
||||
remoteHead.tags = [["d", dTag]];
|
||||
remoteHead.pubkey = pubkey;
|
||||
remoteHead.kind = 30078;
|
||||
|
||||
const publishCalls = [];
|
||||
let fetchCalls = 0;
|
||||
const restoreRelay = stubRelay(relayClient);
|
||||
relayClient.fetchEvents = async () => {
|
||||
fetchCalls++;
|
||||
// Bootstrap: return 500-entry head.
|
||||
if (fetchCalls === 1) return [remoteHead];
|
||||
// Pre-publish merge fetch: return same head.
|
||||
if (fetchCalls === 2) return [remoteHead];
|
||||
// Confirm: return published event.
|
||||
return publishCalls.length > 0
|
||||
? [publishCalls[publishCalls.length - 1]]
|
||||
: [remoteHead];
|
||||
};
|
||||
relayClient.publishEvent = async (evt) => {
|
||||
publishCalls.push(evt);
|
||||
};
|
||||
|
||||
const { fireDelay, restore: restoreTimers } = makeHookTimerBed();
|
||||
const origDateNow = Date.now;
|
||||
Date.now = () => 100 * 1_000;
|
||||
|
||||
let hook = null;
|
||||
try {
|
||||
await act(async () => {
|
||||
hook = renderHook(() => useHook(pubkey, relayUrl));
|
||||
for (let i = 0; i < 40; i++) await Promise.resolve();
|
||||
});
|
||||
|
||||
// Bootstrap returns apply-remote with 500-entry head.
|
||||
// outboxMeta.preservedKey = clickedId.
|
||||
// With the fix: isStarsStoreSubsumedBy(outbox, head, clickedId) →
|
||||
// direct check: head.channels[clickedId] is absent → return false
|
||||
// → subsumed=false → publishStars/publishMutes called.
|
||||
// With the mutation: isStarsStoreSubsumedBy(outbox, head) →
|
||||
// mergeStores(head, outbox) at cap=500 evicts clickedId → equals head
|
||||
// → subsumed=true → publish skipped → clickedId lost.
|
||||
|
||||
// Fire the 2s debounce to trigger the publish cycle.
|
||||
await fireDelay(2000);
|
||||
for (let i = 0; i < 100; i++) await Promise.resolve();
|
||||
|
||||
assert.ok(
|
||||
publishCalls.length > 0,
|
||||
`P3-bootstrap ${label}: publish must fire when clicked channel is absent from relay head — ` +
|
||||
`drop preservedKey from isSubsumedBy call → mergeStores evicts click → subsumed=true → publish skipped`,
|
||||
);
|
||||
const plaintext = tauri.capturedPlaintext();
|
||||
assert.ok(plaintext !== null, "encrypt must have been called");
|
||||
const published = JSON.parse(plaintext);
|
||||
assert.ok(
|
||||
clickedId in published.channels,
|
||||
`P3-bootstrap ${label}: clicked channel "${clickedId}" must survive 500-entry bootstrap subsumption check`,
|
||||
);
|
||||
hook.unmount();
|
||||
} finally {
|
||||
cleanup();
|
||||
Date.now = origDateNow;
|
||||
tauri.restore();
|
||||
restoreRelay();
|
||||
restoreTimers();
|
||||
window.localStorage.clear();
|
||||
mock.reset();
|
||||
}
|
||||
});
|
||||
|
||||
// P3-ack: confirmRetainedHeadSubsumes calls isSubsumedBy without
|
||||
// pendingPreservedKey. At the 500-cap, the same eviction certifies the click
|
||||
// was retained even though it was not. discardPending clears the outbox.
|
||||
// Fix: pass this.pendingPreservedKey to config.isSubsumedBy.
|
||||
// Mutation: drop pendingPreservedKey from the confirmRetainedHeadSubsumes call.
|
||||
test(`P3-ack ${label}: confirmRetainedHeadSubsumes must use pendingPreservedKey; must not clear outbox when relay head does not contain clicked channel`, async () => {
|
||||
const { act, cleanup, renderHook } = await import("@testing-library/react");
|
||||
|
||||
const pubkey = `pk-p3-ack-${label}`;
|
||||
const encodedRelay = encodeURIComponent(relayUrl);
|
||||
|
||||
// Local: clicked channel (updatedAt=1, oldest) + 499 others.
|
||||
const localChannels = {
|
||||
[clickedId]: { [entryValueField]: true, updatedAt: 1, rev: 1 },
|
||||
};
|
||||
for (let i = 0; i < MAX_ENTRIES - 1; i++) {
|
||||
localChannels[`ch-ack-${String(i).padStart(4, "0")}`] = {
|
||||
[entryValueField]: false,
|
||||
updatedAt: 100,
|
||||
rev: 0,
|
||||
};
|
||||
}
|
||||
|
||||
// Remote head returned by confirmRetainedHeadSubsumes: 500 entries that do
|
||||
// NOT include clickedId, but do include one new entry that was not in local.
|
||||
const remoteHeadChannels = {};
|
||||
for (let i = 0; i < MAX_ENTRIES - 1; i++) {
|
||||
remoteHeadChannels[`ch-ack-${String(i).padStart(4, "0")}`] = {
|
||||
[entryValueField]: false,
|
||||
updatedAt: 100,
|
||||
rev: 0,
|
||||
};
|
||||
}
|
||||
remoteHeadChannels[`ch-ack-new-${label}`] = {
|
||||
[entryValueField]: false,
|
||||
updatedAt: 100,
|
||||
rev: 0,
|
||||
};
|
||||
// mergeStores(remoteHead, pending_with_click, NO_KEY) → 501 entries →
|
||||
// evict clickedId (oldest) → equals remoteHead → isSubsumedBy returns true
|
||||
// → discardPending clears outbox → click lost.
|
||||
// Fix: isSubsumedBy(pending, remoteHead, clickedId) →
|
||||
// headEntry = remoteHead.channels[clickedId] = undefined → return false
|
||||
// → scheduleRetry.
|
||||
|
||||
window.localStorage.setItem(
|
||||
storageKey(pubkey, relayUrl),
|
||||
JSON.stringify({ version: 1, channels: localChannels }),
|
||||
);
|
||||
|
||||
const tauri = installEchoTauri(pubkey);
|
||||
const confirmHead = tauri.mintHead(
|
||||
{ version: 1, channels: remoteHeadChannels },
|
||||
50,
|
||||
`evt-p3-ack-${label}`,
|
||||
);
|
||||
confirmHead.tags = [["d", dTag]];
|
||||
confirmHead.pubkey = pubkey;
|
||||
confirmHead.kind = 30078;
|
||||
|
||||
let fetchCalls = 0;
|
||||
const restoreRelay = stubRelay(relayClient);
|
||||
relayClient.fetchEvents = async () => {
|
||||
fetchCalls++;
|
||||
// Bootstrap: absent (so publish is always triggered, no subsumed check).
|
||||
if (fetchCalls === 1) return [];
|
||||
// Pre-publish merge: also absent or same local.
|
||||
if (fetchCalls === 2) return [];
|
||||
// confirmRetainedHeadSubsumes: return remote head without clickedId.
|
||||
return [confirmHead];
|
||||
};
|
||||
relayClient.publishEvent = async (evt) => {};
|
||||
|
||||
const { fireDelay, restore: restoreTimers } = makeHookTimerBed();
|
||||
const origDateNow = Date.now;
|
||||
Date.now = () => 100 * 1_000;
|
||||
|
||||
let hook = null;
|
||||
const v2Prefix = `${outboxKeyPrefix}:${pubkey}:${encodedRelay}:`;
|
||||
const v2Keys = () =>
|
||||
Array.from({ length: window.localStorage.length }, (_, i) =>
|
||||
window.localStorage.key(i),
|
||||
).filter((k) => k?.startsWith(v2Prefix) && k.split(":").length >= 5);
|
||||
|
||||
try {
|
||||
await act(async () => {
|
||||
hook = renderHook(() => useHook(pubkey, relayUrl));
|
||||
for (let i = 0; i < 40; i++) await Promise.resolve();
|
||||
});
|
||||
|
||||
// Click: registers pendingPreservedKey = clickedId.
|
||||
await act(async () => hook.result.current[trueAction](clickedId));
|
||||
|
||||
// Fire 2s debounce: fetchOwn → publishEvent → confirmRetainedHeadSubsumes.
|
||||
// confirmRetainedHeadSubsumes returns confirmHead (no clickedId).
|
||||
// With the fix: isSubsumedBy(published, confirmHead, clickedId) →
|
||||
// headEntry=undefined → false → scheduleRetry (outbox KEPT).
|
||||
// With the mutation: isSubsumedBy(published, confirmHead) →
|
||||
// mergeStores evicts clickedId → true → discardPending → outbox CLEARED.
|
||||
await fireDelay(2000);
|
||||
for (let i = 0; i < 100; i++) await Promise.resolve();
|
||||
|
||||
assert.ok(
|
||||
v2Keys().length > 0,
|
||||
`P3-ack ${label}: outbox must be retained when relay head does not contain clicked channel — ` +
|
||||
`drop pendingPreservedKey from confirmRetainedHeadSubsumes → isSubsumedBy evicts click → discardPending clears outbox`,
|
||||
);
|
||||
hook.unmount();
|
||||
} finally {
|
||||
cleanup();
|
||||
Date.now = origDateNow;
|
||||
tauri.restore();
|
||||
restoreRelay();
|
||||
restoreTimers();
|
||||
window.localStorage.clear();
|
||||
mock.reset();
|
||||
}
|
||||
});
|
||||
}
|
||||
@@ -65,8 +65,13 @@ export type MergeLaneConfig<S> = {
|
||||
* a successful publish without a strict id match (the relay OKs superseded
|
||||
* NIP-33 writes as no-ops, so two windows can both receive OK while only one
|
||||
* blob is retained).
|
||||
*
|
||||
* `preservedKey` is the channelId that was clicked and must survive the
|
||||
* capacity-bounded subsumption proof (Carl P3): a 500-cap merge without this
|
||||
* key can evict the clicked entry and certify retention of a click the relay
|
||||
* never kept.
|
||||
*/
|
||||
isSubsumedBy: (attempted: S, retained: S) => boolean;
|
||||
isSubsumedBy: (attempted: S, retained: S, preservedKey?: string) => boolean;
|
||||
/**
|
||||
* True when `a` is identical to `b` (used to skip no-op publishes against
|
||||
* the last-published head).
|
||||
@@ -355,7 +360,11 @@ export class MergeLaneSyncManager<S> {
|
||||
const remote = await this.decryptAndParse(event);
|
||||
if (!remote) return false;
|
||||
this.observe(remote.store);
|
||||
return this.config.isSubsumedBy(store, remote.store);
|
||||
return this.config.isSubsumedBy(
|
||||
store,
|
||||
remote.store,
|
||||
this.pendingPreservedKey,
|
||||
);
|
||||
} catch {
|
||||
return false;
|
||||
}
|
||||
|
||||
@@ -5,6 +5,7 @@ import {
|
||||
} from "./channelMutesStorage.ts";
|
||||
import { useChannelMutes } from "./useChannelMutes.ts";
|
||||
import { runMergeLaneHookSuite } from "./mergeLaneHook.shared.test.mjs";
|
||||
import { runMergeLaneCarlSuite } from "./mergeLaneSyncCarl.shared.test.mjs";
|
||||
|
||||
runMergeLaneHookSuite({
|
||||
label: "mutes",
|
||||
@@ -20,3 +21,16 @@ runMergeLaneHookSuite({
|
||||
useHook: useChannelMutes,
|
||||
makePayload: (channels) => JSON.stringify({ version: 1, channels }),
|
||||
});
|
||||
|
||||
runMergeLaneCarlSuite({
|
||||
label: "mutes",
|
||||
MAX_ENTRIES: MAX_CHANNEL_MUTE_ENTRIES,
|
||||
entryValueField: "muted",
|
||||
trueAction: "muteChannel",
|
||||
dTag: "channel-mutes",
|
||||
outboxKeyPrefix: "buzz-channel-mutes-outbox.v1",
|
||||
readStore: readChannelMutesStore,
|
||||
storageKey,
|
||||
useHook: useChannelMutes,
|
||||
makePayload: (channels) => JSON.stringify({ version: 1, channels }),
|
||||
});
|
||||
|
||||
@@ -124,7 +124,11 @@ export function useChannelMutes(
|
||||
// noise only.
|
||||
const subsumed =
|
||||
result.action === "apply-remote" &&
|
||||
isMutesStoreSubsumedBy(outboxMeta.store, result.data.store);
|
||||
isMutesStoreSubsumedBy(
|
||||
outboxMeta.store,
|
||||
result.data.store,
|
||||
outboxMeta.preservedKey ?? undefined,
|
||||
);
|
||||
if (!subsumed) {
|
||||
// Forward the preserved key so the clicked channel's capacity
|
||||
// reservation survives remount and restart. The key is selected
|
||||
|
||||
@@ -5,6 +5,10 @@ import { JSDOM } from "jsdom";
|
||||
import { makeHookStubs } from "./sidebarSyncTestHelpers.mjs";
|
||||
import { runWholeBlobHookSuite } from "./wholeBlobHook.shared.test.mjs";
|
||||
import { runWholeBlobP2aSuite } from "./wholeBlobSyncP2a.shared.test.mjs";
|
||||
import {
|
||||
runWholeBlobCarlSuite,
|
||||
runWholeBlobP2bSuite,
|
||||
} from "./wholeBlobSyncCarl.shared.test.mjs";
|
||||
|
||||
const { act, cleanup, renderHook } = await import("@testing-library/react");
|
||||
const { relayClient } = await import("@/shared/api/relayClient");
|
||||
@@ -226,3 +230,42 @@ runWholeBlobP2aSuite({
|
||||
assignments: {},
|
||||
}),
|
||||
});
|
||||
|
||||
runWholeBlobCarlSuite({
|
||||
label: "sections",
|
||||
Manager: ChannelSectionSyncManager,
|
||||
publishEdit: (m, store) => m.publishSections(store),
|
||||
publishReplay: (m, store) => m.publishSections(store, true),
|
||||
subscribe: (m, cb) => m.subscribeToSections(cb),
|
||||
makeNonEmptyStore: () => ({
|
||||
version: 1,
|
||||
sections: [{ id: "s1", name: "Work", order: 0 }],
|
||||
assignments: {},
|
||||
}),
|
||||
makeEditStore: () => ({
|
||||
version: 1,
|
||||
sections: [{ id: "click", name: "Click", order: 0 }],
|
||||
assignments: {},
|
||||
}),
|
||||
makeRemoteStore: () => ({
|
||||
version: 1,
|
||||
sections: [{ id: "remote", name: "Remote", order: 0 }],
|
||||
assignments: {},
|
||||
}),
|
||||
});
|
||||
|
||||
runWholeBlobP2bSuite({
|
||||
label: "sections",
|
||||
Manager: ChannelSectionSyncManager,
|
||||
publishEdit: (m, store) => m.publishSections(store),
|
||||
makeEditStore: () => ({
|
||||
version: 1,
|
||||
sections: [{ id: "click", name: "Click", order: 0 }],
|
||||
assignments: {},
|
||||
}),
|
||||
makeRemoteStore: () => ({
|
||||
version: 1,
|
||||
sections: [{ id: "remote", name: "Remote", order: 0 }],
|
||||
assignments: {},
|
||||
}),
|
||||
});
|
||||
|
||||
@@ -189,7 +189,10 @@ export function useChannelSections(
|
||||
// crash after resumes it from the v2 key. The marker is what stops the
|
||||
// never-deleted legacy key republishing above the head every boot
|
||||
// (Thufir pass-2 resurrection finding).
|
||||
const durable = managerRef.current?.publishSections(outbox.store);
|
||||
const durable = managerRef.current?.publishSections(
|
||||
outbox.store,
|
||||
true,
|
||||
);
|
||||
if (durable && outbox.legacyRawToConsume !== null) {
|
||||
markChannelSectionsLegacyConsumed(
|
||||
pubkey,
|
||||
|
||||
@@ -5,6 +5,10 @@ import { JSDOM } from "jsdom";
|
||||
import { makeHookStubs } from "./sidebarSyncTestHelpers.mjs";
|
||||
import { runWholeBlobHookSuite } from "./wholeBlobHook.shared.test.mjs";
|
||||
import { runWholeBlobP2aSuite } from "./wholeBlobSyncP2a.shared.test.mjs";
|
||||
import {
|
||||
runWholeBlobCarlSuite,
|
||||
runWholeBlobP2bSuite,
|
||||
} from "./wholeBlobSyncCarl.shared.test.mjs";
|
||||
|
||||
const { act, cleanup, renderHook } = await import("@testing-library/react");
|
||||
const { relayClient } = await import("@/shared/api/relayClient");
|
||||
@@ -141,3 +145,22 @@ runWholeBlobP2aSuite({
|
||||
makeMountStore: () => ({ version: 1, groups: { mount: "recent" } }),
|
||||
makeRemoteStore: () => ({ version: 1, groups: { remote: "recent" } }),
|
||||
});
|
||||
|
||||
runWholeBlobCarlSuite({
|
||||
label: "sort",
|
||||
Manager: ChannelSortSyncManager,
|
||||
publishEdit: (m, store) => m.publishSortPrefs(store),
|
||||
publishReplay: (m, store) => m.publishSortPrefs(store, true),
|
||||
subscribe: (m, cb) => m.subscribeToSortPrefs(cb),
|
||||
makeNonEmptyStore: () => ({ version: 1, groups: { s1: "recent" } }),
|
||||
makeEditStore: () => ({ version: 1, groups: { click: "alpha" } }),
|
||||
makeRemoteStore: () => ({ version: 1, groups: { remote: "recent" } }),
|
||||
});
|
||||
|
||||
runWholeBlobP2bSuite({
|
||||
label: "sort",
|
||||
Manager: ChannelSortSyncManager,
|
||||
publishEdit: (m, store) => m.publishSortPrefs(store),
|
||||
makeEditStore: () => ({ version: 1, groups: { click: "alpha" } }),
|
||||
makeRemoteStore: () => ({ version: 1, groups: { remote: "recent" } }),
|
||||
});
|
||||
|
||||
@@ -186,7 +186,10 @@ export function useChannelSortPreference(
|
||||
// crash after resumes it from the v2 key. The marker is what stops the
|
||||
// never-deleted legacy key republishing above the head every boot
|
||||
// (Thufir pass-2 resurrection finding).
|
||||
const durable = managerRef.current?.publishSortPrefs(outbox.store);
|
||||
const durable = managerRef.current?.publishSortPrefs(
|
||||
outbox.store,
|
||||
true,
|
||||
);
|
||||
if (durable && outbox.legacyRawToConsume !== null) {
|
||||
markChannelSortLegacyConsumed(
|
||||
pubkey,
|
||||
|
||||
@@ -5,6 +5,7 @@ import {
|
||||
} from "./channelStarsStorage.ts";
|
||||
import { useChannelStars } from "./useChannelStars.ts";
|
||||
import { runMergeLaneHookSuite } from "./mergeLaneHook.shared.test.mjs";
|
||||
import { runMergeLaneCarlSuite } from "./mergeLaneSyncCarl.shared.test.mjs";
|
||||
|
||||
runMergeLaneHookSuite({
|
||||
label: "stars",
|
||||
@@ -20,3 +21,16 @@ runMergeLaneHookSuite({
|
||||
useHook: useChannelStars,
|
||||
makePayload: (channels) => JSON.stringify({ version: 1, channels }),
|
||||
});
|
||||
|
||||
runMergeLaneCarlSuite({
|
||||
label: "stars",
|
||||
MAX_ENTRIES: MAX_CHANNEL_STAR_ENTRIES,
|
||||
entryValueField: "starred",
|
||||
trueAction: "starChannel",
|
||||
dTag: "channel-stars",
|
||||
outboxKeyPrefix: "buzz-channel-stars-outbox.v1",
|
||||
readStore: readChannelStarsStore,
|
||||
storageKey,
|
||||
useHook: useChannelStars,
|
||||
makePayload: (channels) => JSON.stringify({ version: 1, channels }),
|
||||
});
|
||||
|
||||
@@ -125,7 +125,11 @@ export function useChannelStars(
|
||||
// noise only.
|
||||
const subsumed =
|
||||
result.action === "apply-remote" &&
|
||||
isStarsStoreSubsumedBy(outboxMeta.store, result.data.store);
|
||||
isStarsStoreSubsumedBy(
|
||||
outboxMeta.store,
|
||||
result.data.store,
|
||||
outboxMeta.preservedKey ?? undefined,
|
||||
);
|
||||
if (!subsumed) {
|
||||
// Forward the preserved key so the clicked channel's capacity
|
||||
// reservation survives remount and restart. The key is selected
|
||||
|
||||
@@ -0,0 +1,399 @@
|
||||
// Carl-round regression suite for WholeBlobSyncManager (P1, P2a-1, P2b).
|
||||
//
|
||||
// Three causal regressions confirmed at source in Duncan's analysis pass:
|
||||
//
|
||||
// T-P1 (hook layer): failed bootstrap → restored outbox (old queuedAt) →
|
||||
// bootstrap `.then()` replay → pre-publish fetch finds newer peer head B
|
||||
// retained while device was closed → must ADOPT (never publish stale edit
|
||||
// over B).
|
||||
// Mutation: remove `!this.pendingIsRestoredReplay` guard from the
|
||||
// failed-bootstrap exception → exception fires, folds B in, publishes over it.
|
||||
//
|
||||
// T-P2a1 (hook layer): blocked bootstrap H100 → click → live H102 arrives and
|
||||
// is suppressed (hasPendingEdit) → bootstrap resolves H100 → hook `.then()`
|
||||
// calls publishSections/publishSortPrefs(outbox, isRestoredReplay=true) →
|
||||
// replay must NOT re-freeze publishBaseline from mutable lastRemoteHead →
|
||||
// pre-publish fetch returns H102 → ADOPT (H102 is a genuine advance).
|
||||
// Mutation: remove `if (!isRestoredReplay)` guard from publish() →
|
||||
// publishBaseline refreezes to H102 → pre-publish sees equality → publishes
|
||||
// pre-H102 content over H102, H102's changes lost.
|
||||
//
|
||||
// T-P2b (manager layer): periodic/reconnect fetchRemoteBlob path records
|
||||
// lastRemoteHead BEFORE decryptAndParse returns; a concurrent click freezes
|
||||
// publishBaseline against the pre-decrypt head; the pre-publish fetch then
|
||||
// sees no advance and publishes stale content over the fetched event.
|
||||
// Mutation: move recordRemoteHead back above decryptAndParse (revert P2b fix)
|
||||
// → publishBaseline freezes against undecrypted head → pre-publish fetch
|
||||
// returns same head → no advance → publish-over.
|
||||
|
||||
import assert from "node:assert/strict";
|
||||
import test, { mock } from "node:test";
|
||||
|
||||
import { relayClient } from "@/shared/api/relayClient";
|
||||
import {
|
||||
makeHookTimerBed,
|
||||
installEchoTauri,
|
||||
} from "./sidebarSyncTestHelpers.mjs";
|
||||
|
||||
const RELAY = "wss://r.carl";
|
||||
|
||||
/**
|
||||
* Register the Carl-round P1/P2a-1 hook-layer regressions for a single lane.
|
||||
*
|
||||
* @param {object} opts
|
||||
* @param {string} opts.label — "sections"|"sort"
|
||||
* @param {Function} opts.Manager — concrete manager class
|
||||
* @param {Function} opts.publishEdit — (manager, store) => void (fresh click, isRestoredReplay=false)
|
||||
* @param {Function} opts.publishReplay — (manager, store) => void (restored replay, isRestoredReplay=true)
|
||||
* @param {Function} opts.subscribe — (manager, cb) => Promise<unsubscribe>
|
||||
* @param {Function} opts.makeNonEmptyStore — () => non-empty mount store
|
||||
* @param {Function} opts.makeEditStore — () => user-click store (distinct from mount)
|
||||
* @param {Function} opts.makeRemoteStore — () => peer relay head store
|
||||
*/
|
||||
export function runWholeBlobCarlSuite({
|
||||
label,
|
||||
Manager,
|
||||
publishEdit,
|
||||
publishReplay,
|
||||
subscribe,
|
||||
makeNonEmptyStore,
|
||||
makeEditStore,
|
||||
makeRemoteStore,
|
||||
}) {
|
||||
// T-P1: failed bootstrap → hook replay of restored outbox → peer head B was
|
||||
// retained while device was closed → replay must ADOPT, never publish over B.
|
||||
//
|
||||
// Production sequence:
|
||||
// 1. A edits at t=100, queuedAt=100, quits before publish.
|
||||
// 2. Peer retains head B at t=200 while A is offline.
|
||||
// 3. A reopens: bootstrap fetch fails → bootstrapFailed=true.
|
||||
// 4. Hook `.then()` reads outbox (queuedAt=100 < B.createdAt=200 → shouldReplay
|
||||
// guard in the hook would be false for apply-remote; but bootstrap=hold here
|
||||
// so shouldReplay is always true). The replay calls publish(store,
|
||||
// isRestoredReplay=true) → pendingIsRestoredReplay=true.
|
||||
// 5. Debounce fires. fetchOwnBlobBeforePublish returns B.
|
||||
// remoteAdvancedSince(B, {0,""}) = true → would normally adopt.
|
||||
// Failed-bootstrap exception: bootstrapFailed=true,
|
||||
// !bootstrapFailedExternalHeadObserved=true, publishBaseline={0,""}.
|
||||
// With the fix: pendingIsRestoredReplay=true → exception suppressed
|
||||
// → normal adopt fires.
|
||||
// Mutation: exception fires → fold B into baseline → publish over B.
|
||||
//
|
||||
// The test uses a `hold` bootstrap (failed fetch) so the hook always replays.
|
||||
// `shouldReplay` is true for `hold`. The outbox queuedAt (100) is older than
|
||||
// B (200) — without the fix, the exception would override and publish over B.
|
||||
test(`P1 ${label}: failed-bootstrap hook replay must adopt a newer peer head retained while device was closed, never publish over it`, async () => {
|
||||
let publishCalls = 0;
|
||||
const { fireDelay, restore } = makeHookTimerBed();
|
||||
const tauri = installEchoTauri(`pk-p1-${label}`);
|
||||
|
||||
const peerHead = tauri.mintHead(
|
||||
makeRemoteStore(),
|
||||
200,
|
||||
`evt-p1-peer-${label}`,
|
||||
);
|
||||
|
||||
let fetchCalls = 0;
|
||||
mock.method(relayClient, "fetchEvents", () => {
|
||||
fetchCalls++;
|
||||
// Call 1 = bootstrap fetch: fail → bootstrapFailed=true.
|
||||
if (fetchCalls === 1) return Promise.reject(new Error("bootstrap fail"));
|
||||
// Subsequent calls (pre-publish fetch, retry): return peerHead.
|
||||
return Promise.resolve([peerHead]);
|
||||
});
|
||||
mock.method(relayClient, "publishEvent", () => {
|
||||
publishCalls++;
|
||||
return Promise.resolve();
|
||||
});
|
||||
|
||||
try {
|
||||
const manager = new Manager(`pk-p1-${label}`, RELAY);
|
||||
const adopted = [];
|
||||
manager.setOnRemoteAdopted((r) => adopted.push(r));
|
||||
|
||||
// Bootstrap fails synchronously.
|
||||
await manager.bootstrap(makeNonEmptyStore());
|
||||
|
||||
// Hook .then() replays the outbox: publishEdit with isRestoredReplay=true.
|
||||
// The manager sees bootstrapResolved=true and schedules the debounce.
|
||||
publishReplay(manager, makeEditStore());
|
||||
|
||||
// Debounce fires.
|
||||
await fireDelay(2000);
|
||||
for (let i = 0; i < 100; i++) await Promise.resolve();
|
||||
|
||||
// The pre-publish fetch returns peerHead (createdAt=200) which is newer
|
||||
// than publishBaseline={0,""}. With the fix: pendingIsRestoredReplay=true
|
||||
// → failed-bootstrap exception is suppressed → normal adopt fires.
|
||||
// Mutation: exception fires → fold peerHead into baseline → publish over it.
|
||||
assert.equal(
|
||||
publishCalls,
|
||||
0,
|
||||
`P1 ${label}: restored replay must ADOPT newer peer head, never publish over it`,
|
||||
);
|
||||
assert.equal(
|
||||
adopted.length,
|
||||
1,
|
||||
`P1 ${label}: peer head must be adopted after restored replay`,
|
||||
);
|
||||
assert.equal(
|
||||
manager.getPendingStore(),
|
||||
null,
|
||||
`P1 ${label}: pending cleared after adopt`,
|
||||
);
|
||||
manager.destroy();
|
||||
} finally {
|
||||
tauri.restore();
|
||||
restore();
|
||||
mock.reset();
|
||||
}
|
||||
});
|
||||
|
||||
// T-P2a1: blocked bootstrap H100 → click → live H102 decrypts (suppressed by
|
||||
// hasPendingEdit) → bootstrap resolves H100 → hook .then() replay
|
||||
// (isRestoredReplay=true) → H102 must be adopted as a genuine advance.
|
||||
//
|
||||
// This tests that the hook-level replay path (which calls publish with
|
||||
// isRestoredReplay=true) does NOT re-freeze publishBaseline from lastRemoteHead
|
||||
// (which now holds H102 after the live delivery). releaseDeferred already set
|
||||
// publishBaseline = canonicalMax(click-time, H100) = H100 (click-time baseline
|
||||
// was {0,""}). The replay's publish(store, true) must keep that baseline.
|
||||
// Pre-publish fetch returns H102 (createdAt=400 > H100.createdAt=200) →
|
||||
// remoteAdvancedSince(H102, H100) = true → ADOPT.
|
||||
//
|
||||
// Mutation: publish() re-freezes publishBaseline = lastRemoteHead = H102 →
|
||||
// pre-publish sees equality → publishes pre-H102 content over H102 (lose H102).
|
||||
test(`P2a-1 ${label}: hook replay after blocked bootstrap must not re-freeze baseline from live peer head H102; H102 must be adopted`, async () => {
|
||||
let publishCalls = 0;
|
||||
let liveCallback = null;
|
||||
let releaseBootstrap = null;
|
||||
const { fireDelay, restore } = makeHookTimerBed();
|
||||
const tauri = installEchoTauri(`pk-p2a1-${label}`);
|
||||
|
||||
const H100 = tauri.mintHead(makeRemoteStore(), 200, `evt-h100-${label}`);
|
||||
const H102 = tauri.mintHead(makeRemoteStore(), 400, `evt-h102-${label}`);
|
||||
|
||||
mock.method(relayClient, "subscribeLive", (_f, cb) => {
|
||||
liveCallback = cb;
|
||||
return Promise.resolve(async () => {});
|
||||
});
|
||||
|
||||
let fetchCalls = 0;
|
||||
mock.method(relayClient, "fetchEvents", () => {
|
||||
fetchCalls++;
|
||||
// Call 1 = bootstrap fetch (blocked).
|
||||
if (fetchCalls === 1)
|
||||
return new Promise((res) => {
|
||||
releaseBootstrap = () => res([H100]);
|
||||
});
|
||||
// Subsequent (pre-publish fetch): return H102.
|
||||
return Promise.resolve([H102]);
|
||||
});
|
||||
mock.method(relayClient, "publishEvent", () => {
|
||||
publishCalls++;
|
||||
return Promise.resolve();
|
||||
});
|
||||
|
||||
try {
|
||||
const manager = new Manager(`pk-p2a1-${label}`, RELAY);
|
||||
const adopted = [];
|
||||
manager.setOnRemoteAdopted((r) => adopted.push(r));
|
||||
|
||||
// Subscribe to capture live callback.
|
||||
await subscribe(manager, () => {});
|
||||
|
||||
// Bootstrap blocked; fresh click BEFORE bootstrap resolves.
|
||||
// (No existing outbox — this simulates a click during the bootstrap window,
|
||||
// with the hook seeing an existing pending edit before .then() runs.)
|
||||
const bootstrapPromise = manager.bootstrap(makeNonEmptyStore());
|
||||
// Fresh click: publish(store, isRestoredReplay=false) — normal path, freezes
|
||||
// publishBaseline = {0,""} (lastRemoteHead before any live event).
|
||||
publishEdit(manager, makeEditStore());
|
||||
|
||||
// Live H102 arrives BEFORE bootstrap resolves (suppressed by hasPendingEdit).
|
||||
while (liveCallback === null) await Promise.resolve();
|
||||
liveCallback(H102);
|
||||
for (let i = 0; i < 20; i++) await Promise.resolve();
|
||||
// lastRemoteHead is now H102.
|
||||
|
||||
// Bootstrap resolves with H100. releaseDeferred sets publishBaseline =
|
||||
// canonicalMax({0,""}, H100) = H100.
|
||||
while (releaseBootstrap === null) await Promise.resolve();
|
||||
releaseBootstrap();
|
||||
await bootstrapPromise;
|
||||
for (let i = 0; i < 50; i++) await Promise.resolve();
|
||||
// bootstrapResolved=true. The timer was scheduled by releaseDeferred.
|
||||
|
||||
// Hook .then() would replay the outbox edit. Simulate: call publish with
|
||||
// isRestoredReplay=true as the hook's .then() callback does.
|
||||
// With the fix: baseline is unchanged (still H100).
|
||||
// With the mutation: publishBaseline is reset to lastRemoteHead=H102.
|
||||
publishReplay(manager, makeEditStore());
|
||||
|
||||
// Fire the debounce.
|
||||
await fireDelay(2000);
|
||||
for (let i = 0; i < 100; i++) await Promise.resolve();
|
||||
|
||||
// Pre-publish returns H102 (createdAt=400 > H100.createdAt=200).
|
||||
// Fix: publishBaseline=H100 → remoteAdvancedSince(H102, H100) = true → ADOPT.
|
||||
// Mutation: publishBaseline=H102 → remoteAdvancedSince(H102, H102) = false
|
||||
// → publish-over fires, H102's changes lost.
|
||||
assert.equal(
|
||||
publishCalls,
|
||||
0,
|
||||
`P2a-1 ${label}: must ADOPT H102 (genuine advance), not publish over it`,
|
||||
);
|
||||
assert.equal(
|
||||
adopted.length,
|
||||
1,
|
||||
`P2a-1 ${label}: H102 must be adopted as genuine remote advance`,
|
||||
);
|
||||
manager.destroy();
|
||||
} finally {
|
||||
tauri.restore();
|
||||
restore();
|
||||
mock.reset();
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Register the Carl-round P2b manager-level regression (fetchRemoteBlob
|
||||
* decrypt gap — periodic/reconnect path).
|
||||
*
|
||||
* @param {object} opts
|
||||
* @param {string} opts.label — "sections"|"sort"
|
||||
* @param {Function} opts.Manager — concrete manager class
|
||||
* @param {Function} opts.publishEdit — (manager, store) => void
|
||||
* @param {Function} opts.makeEditStore — () => store
|
||||
* @param {Function} opts.makeRemoteStore — () => store
|
||||
*/
|
||||
export function runWholeBlobP2bSuite({
|
||||
label,
|
||||
Manager,
|
||||
publishEdit,
|
||||
makeEditStore,
|
||||
makeRemoteStore,
|
||||
}) {
|
||||
// T-P2b: periodic/reconnect fetchRemoteBlob records lastRemoteHead BEFORE
|
||||
// decryptAndParse; a concurrent click (whose publish() fires during the
|
||||
// async decrypt gap) freezes publishBaseline against the pre-decrypt head;
|
||||
// the pre-publish fetch returns the same head; no advance → publish-over.
|
||||
//
|
||||
// Sequence:
|
||||
// 1. fetchRemoteBlob is in-flight, has fetched the event but has not yet
|
||||
// decrypted it. Before the fix, recordRemoteHead(event.id) fires here →
|
||||
// lastRemoteHead advances to H.
|
||||
// 2. User click: publish() freezes publishBaseline = lastRemoteHead = H.
|
||||
// 3. decryptAndParse resolves → store updated to H's content.
|
||||
// 4. Debounce fires. fetchOwnBlobBeforePublish returns H.
|
||||
// remoteAdvancedSince(H, H) = false → publishes. BUT: our store is the
|
||||
// click's intent, NOT H's content + click merged in → H's remote-only
|
||||
// changes are lost.
|
||||
// With the fix: recordRemoteHead deferred to after decrypt → at step 2
|
||||
// lastRemoteHead is still {createdAt: 0} → publishBaseline={0,""} → at
|
||||
// step 4 remoteAdvancedSince(H, {0,""}) = true → ADOPT (or use exception
|
||||
// path for a failed-bootstrap scenario). Either way, the clicked edit
|
||||
// doesn't silently overwrite H's remote-only changes.
|
||||
//
|
||||
// This test exercises the race by controlling the decrypt resolution.
|
||||
test(`P2b ${label}: fetchRemoteBlob must not advance lastRemoteHead before decryptAndParse succeeds; click during decrypt gap must not publish over fetched head`, async () => {
|
||||
let publishCalls = 0;
|
||||
const { fireDelay, restore } = makeHookTimerBed();
|
||||
const tauri = installEchoTauri(`pk-p2b-${label}`);
|
||||
|
||||
// H: the pre-existing relay head with some remote-only content.
|
||||
const H = tauri.mintHead(makeRemoteStore(), 200, `evt-p2b-h-${label}`);
|
||||
|
||||
// A controlled decrypt gate: we resolve the decrypt promise AFTER the click.
|
||||
let releaseDecrypt = null;
|
||||
const origInvoke = globalThis.window?.__TAURI_INTERNALS__?.invoke;
|
||||
const orig = globalThis.window.__TAURI_INTERNALS__;
|
||||
// Wrap decrypt: the first call (for H in fetchRemoteBlob) is gated.
|
||||
let decryptCallCount = 0;
|
||||
globalThis.window.__TAURI_INTERNALS__ = {
|
||||
invoke: (cmd, args) => {
|
||||
if (cmd === "nip44_decrypt_from_self") {
|
||||
decryptCallCount++;
|
||||
if (decryptCallCount === 1) {
|
||||
// First decrypt (fetchRemoteBlob's call for H): hold it.
|
||||
return new Promise((res, rej) => {
|
||||
releaseDecrypt = () =>
|
||||
orig.invoke(cmd, args).then(res).catch(rej);
|
||||
});
|
||||
}
|
||||
return orig.invoke(cmd, args);
|
||||
}
|
||||
return orig.invoke(cmd, args);
|
||||
},
|
||||
};
|
||||
|
||||
let fetchCalls = 0;
|
||||
mock.method(relayClient, "fetchEvents", () => {
|
||||
fetchCalls++;
|
||||
// Bootstrap fetch: absent (no remote head at mount).
|
||||
if (fetchCalls === 1) return Promise.resolve([]);
|
||||
// fetchRemoteBlob (periodic/reconnect) call: return H.
|
||||
if (fetchCalls === 2) return Promise.resolve([H]);
|
||||
// Pre-publish fetch: return H (relay still has H as head).
|
||||
return Promise.resolve([H]);
|
||||
});
|
||||
mock.method(relayClient, "publishEvent", () => {
|
||||
publishCalls++;
|
||||
return Promise.resolve();
|
||||
});
|
||||
|
||||
try {
|
||||
const manager = new Manager(`pk-p2b-${label}`, RELAY);
|
||||
const adopted = [];
|
||||
manager.setOnRemoteAdopted((r) => adopted.push(r));
|
||||
|
||||
// Bootstrap: absent → bootstrapFailed=false, publishBaseline={0,""}.
|
||||
await manager.bootstrap(makeEditStore());
|
||||
|
||||
// Trigger a fetchRemoteBlob (simulates the periodic/reconnect path).
|
||||
// This call will block at the decrypt await.
|
||||
const fetchPromise = manager.fetchRemoteBlob();
|
||||
// Wait for the fetch to have collected the event but not yet decrypted.
|
||||
// The fetch awaits decryptAndParse at this point.
|
||||
for (let i = 0; i < 10; i++) await Promise.resolve();
|
||||
|
||||
// Click during the decrypt gap.
|
||||
// With the fix: lastRemoteHead still {createdAt:0} (decrypt not done yet)
|
||||
// → publishBaseline = {0,""}.
|
||||
// With the mutation: lastRemoteHead = {200, H.id} (pre-decrypt record)
|
||||
// → publishBaseline = {200, H.id}.
|
||||
publishEdit(manager, makeEditStore());
|
||||
|
||||
// Release the decrypt.
|
||||
assert.ok(releaseDecrypt !== null, "decrypt gate must have been hit");
|
||||
releaseDecrypt();
|
||||
await fetchPromise;
|
||||
for (let i = 0; i < 20; i++) await Promise.resolve();
|
||||
|
||||
// Debounce fires. Pre-publish fetch returns H (createdAt=200, same id).
|
||||
// With the fix: publishBaseline={0,""} → remoteAdvancedSince(H,{0,""})=true
|
||||
// → bootstrapFailed=false so no exception → normal ADOPT.
|
||||
// With the mutation: publishBaseline=H → remoteAdvancedSince(H,H)=false
|
||||
// → publish fires, H's remote-only content silently lost.
|
||||
await fireDelay(2000);
|
||||
for (let i = 0; i < 100; i++) await Promise.resolve();
|
||||
|
||||
assert.equal(
|
||||
publishCalls,
|
||||
0,
|
||||
`P2b ${label}: click during fetchRemoteBlob decrypt gap must not publish over the fetched head`,
|
||||
);
|
||||
assert.equal(
|
||||
adopted.length >= 1,
|
||||
true,
|
||||
`P2b ${label}: the fetched head must be adopted (genuine advance)`,
|
||||
);
|
||||
manager.destroy();
|
||||
} finally {
|
||||
globalThis.window.__TAURI_INTERNALS__ = orig;
|
||||
restore();
|
||||
mock.reset();
|
||||
}
|
||||
});
|
||||
}
|
||||
@@ -224,6 +224,10 @@ export class WholeBlobSyncManager<S> {
|
||||
// normal remoteAdvancedSince check correctly adopts — the exception must not
|
||||
// override that by treating the head as a "first unknown base".
|
||||
private bootstrapFailedExternalHeadObserved = false;
|
||||
// True when the current pending edit is a hook-level replay of a prior
|
||||
// session's durable outbox record. Suppresses the failed-bootstrap exception
|
||||
// in fetchOwnBlobBeforePublish (Carl P1, see publish() JSDoc).
|
||||
private pendingIsRestoredReplay = false;
|
||||
// Event ids we signed and sent to the relay but whose ACK never arrived (the
|
||||
// publish promise rejected as a timeout/socket error after the frame left).
|
||||
// The relay MAY have accepted such a write, so if a later cycle's pre-publish
|
||||
@@ -270,10 +274,27 @@ export class WholeBlobSyncManager<S> {
|
||||
return { status: "absent" };
|
||||
}
|
||||
const event = events[0];
|
||||
// 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);
|
||||
// Advance the watermark (seed-publish guard) regardless of decrypt
|
||||
// outcome so an unreadable event still blocks a future seed-publish.
|
||||
// The canonical head tuple (`lastRemoteHead`) is updated only AFTER
|
||||
// decrypt succeeds (Carl P2b, periodic/reconnect path): a click during
|
||||
// the async decrypt gap must not freeze its baseline against a head
|
||||
// whose store content has not yet been applied.
|
||||
if (event.created_at > this.lastRemoteCreatedAt) {
|
||||
this.lastRemoteCreatedAt = event.created_at;
|
||||
}
|
||||
advanceWatermark(
|
||||
this.pubkey,
|
||||
this.config.dTag,
|
||||
this.relayUrl,
|
||||
event.created_at,
|
||||
);
|
||||
const result = await this.decryptAndParse(event);
|
||||
if (!result) {
|
||||
return { status: "failed", createdAt: event.created_at };
|
||||
}
|
||||
// Decrypt succeeded — record the full head tuple.
|
||||
this.recordRemoteHead(result.createdAt, result.eventId);
|
||||
// A head was independently observed via a non-bootstrap fetch (reconnect,
|
||||
// periodic refresh, etc.). Disarm the failed-bootstrap exception so this
|
||||
// head is treated as a genuine remote advance rather than a "first unknown
|
||||
@@ -281,10 +302,6 @@ export class WholeBlobSyncManager<S> {
|
||||
if (this.bootstrapFailed) {
|
||||
this.bootstrapFailedExternalHeadObserved = true;
|
||||
}
|
||||
const result = await this.decryptAndParse(event);
|
||||
if (!result) {
|
||||
return { status: "failed", createdAt: event.created_at };
|
||||
}
|
||||
return {
|
||||
status: "found",
|
||||
data: result,
|
||||
@@ -344,15 +361,10 @@ export class WholeBlobSyncManager<S> {
|
||||
}
|
||||
|
||||
/**
|
||||
* Re-drive the CURRENT pending edit's publish immediately without opening a
|
||||
* new generation — used by the reconnect handler. Calling the public
|
||||
* `publish()` there would bump the generation and reset `publishBaseline` to
|
||||
* the just-fetched head, so a remote that won LWW while the edit was pending
|
||||
* would be published over instead of adopted (Carl P1). Waking the existing
|
||||
* generation keeps the frozen baseline, so the debounced cycle's pre-publish
|
||||
* check still adopts a genuinely-advanced remote. No-op when nothing is
|
||||
* pending or when bootstrap has not yet resolved (the deferred-release path
|
||||
* handles that case once bootstrap completes).
|
||||
* Re-drive the current pending edit without opening a new generation — used
|
||||
* by the reconnect handler to keep the frozen baseline so a remote that won
|
||||
* LWW while the edit was pending is still adopted (Carl P1). No-op when
|
||||
* nothing is pending or bootstrap has not yet resolved.
|
||||
*/
|
||||
retryPendingPublish(): void {
|
||||
if (this.pendingStore === null) return;
|
||||
@@ -371,27 +383,13 @@ export class WholeBlobSyncManager<S> {
|
||||
* bootstrap() after it sets bootstrapResolved=true, passing the head that
|
||||
* bootstrap itself observed (the canonical head at the moment bootstrap
|
||||
* completed — NOT the mutable lastRemoteHead, which subscribeLive may advance
|
||||
* at any time). If a pending edit exists with no timer or in-flight cycle,
|
||||
* re-freeze publishBaseline to canonicalMax(queueTimeBaseline, bootstrapResultHead)
|
||||
* and schedule the normal 2s debounce.
|
||||
*
|
||||
* Re-freezing the baseline here is the P2a fix for the success path: the
|
||||
* edit's comparison point becomes the head bootstrap established, so the
|
||||
* pre-publish fetch for the same head returns "not advanced" and the edit
|
||||
* publishes above it correctly.
|
||||
*
|
||||
* Using the bootstrap-result snapshot (not mutable lastRemoteHead) is critical:
|
||||
* subscribeLive can advance lastRemoteHead while bootstrap is unresolved. A
|
||||
* live peer head arriving AFTER the click but BEFORE bootstrap resolves is
|
||||
* never incorporated into the editable store while a local edit is pending;
|
||||
* using mutable lastRemoteHead here would fold that live head into the baseline,
|
||||
* making the pre-publish check see equality and publish the pre-head blob OVER
|
||||
* the peer's write. With the bootstrap-result snapshot, that live peer head
|
||||
* remains a genuine advance (its timestamp exceeds the just-frozen baseline)
|
||||
* and is adopted per normal LWW when the pre-publish fetch returns it.
|
||||
*
|
||||
* For a failed bootstrap, bootstrapResultHead is {0,""} and the bootstrapFailed
|
||||
* flag enables the doPublish-time exception.
|
||||
* at any time). Re-freezes publishBaseline to
|
||||
* canonicalMax(queueTimeBaseline, bootstrapResultHead) and schedules the 2s
|
||||
* debounce. Using the bootstrap-result snapshot (not lastRemoteHead) is
|
||||
* critical: a live peer head arriving after the click but before bootstrap
|
||||
* resolves must remain a genuine remote advance, not be folded into the
|
||||
* baseline. For a failed bootstrap, bootstrapResultHead is {0,""} and the
|
||||
* bootstrapFailed flag enables the doPublish-time exception.
|
||||
*/
|
||||
private releaseDeferred(bootstrapResultHead: PublishBaseline): void {
|
||||
if (
|
||||
@@ -463,17 +461,9 @@ export class WholeBlobSyncManager<S> {
|
||||
}
|
||||
|
||||
/**
|
||||
* 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 (which would reintroduce divergence).
|
||||
*
|
||||
* Compare-and-swap on `gen`: this adopt was decided against the edit queued
|
||||
* at generation `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
|
||||
* — it will publish and win LWW — so a stale adopt must not clear its pending
|
||||
* state or overwrite its optimistic UI. We still advance the watermark
|
||||
* (monotonic and always safe) so the newer edit stamps above this head.
|
||||
* Adopt a remote store that superseded a local edit: hand to the hook for
|
||||
* write-through, advance the watermark, and drop the losing pending edit.
|
||||
* CAS on `gen`: a stale adopt must not clear a newer edit's state.
|
||||
*/
|
||||
private adoptRemote(remote: RemoteBlob<S>, gen: number): void {
|
||||
this.recordRemoteHead(remote.createdAt, remote.eventId);
|
||||
@@ -501,10 +491,30 @@ export class WholeBlobSyncManager<S> {
|
||||
private discardPending(gen: number): void {
|
||||
if (gen !== this.pendingGeneration) return;
|
||||
this.pendingStore = null;
|
||||
this.pendingIsRestoredReplay = false;
|
||||
this.config.clearOutbox(this.pubkey, this.relayUrl);
|
||||
}
|
||||
|
||||
publish(store: S): boolean {
|
||||
/**
|
||||
* Queue a store for debounced publish.
|
||||
*
|
||||
* @param isRestoredReplay When true, this call is a hook-level replay of a
|
||||
* prior session's durable outbox record. Two behaviours are adjusted:
|
||||
*
|
||||
* 1. **Baseline preservation (P2a-1):** `publishBaseline` is NOT
|
||||
* re-frozen from `lastRemoteHead`. `releaseDeferred` already set it to
|
||||
* `canonicalMax(queueTime, bootstrapResult)` — re-freezing here would
|
||||
* fold in a suppressed live peer head (H102) and cause the pre-publish
|
||||
* check to see equality and publish over H102 instead of adopting it.
|
||||
*
|
||||
* 2. **Failed-bootstrap exception (P1):** the "first unknown base"
|
||||
* exception in `fetchOwnBlobBeforePublish` is suppressed. A restored
|
||||
* replay carries its original `queuedAt` from the prior session and its
|
||||
* baseline was already frozen then — the exception was designed only for
|
||||
* fresh in-session clicks. Letting it fire for a replay would publish
|
||||
* the stale edit over a newer peer head retained while offline (Carl P1).
|
||||
*/
|
||||
publish(store: S, isRestoredReplay = false): boolean {
|
||||
this.pendingStore = store;
|
||||
++this.pendingGeneration;
|
||||
// Freeze the canonical head this edit is racing against at queue time. The
|
||||
@@ -515,11 +525,18 @@ export class WholeBlobSyncManager<S> {
|
||||
// overwrites a remote that became head *after* this edit was queued. The
|
||||
// baseline only advances via our own successful publishes (see doPublish),
|
||||
// so a prior generation's own accepted write is folded in rather than
|
||||
// mistaken for a competing remote.
|
||||
this.publishBaseline = { ...this.lastRemoteHead };
|
||||
// mistaken for a competing remote. For a restored replay, the baseline was
|
||||
// already established by releaseDeferred — do not re-freeze (see JSDoc).
|
||||
if (!isRestoredReplay) {
|
||||
this.publishBaseline = { ...this.lastRemoteHead };
|
||||
}
|
||||
// Record whether this is a restored replay so fetchOwnBlobBeforePublish
|
||||
// can suppress the failed-bootstrap exception (P1) — a replay must not
|
||||
// re-qualify for the "first unknown base" logic.
|
||||
this.pendingIsRestoredReplay = isRestoredReplay;
|
||||
// Persist synchronously so an edit 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,
|
||||
// survives teardown and resumes on next mount (durable outbox).
|
||||
// This window's own key is the only one written — a single unconditional setItem,
|
||||
// never a shared-key read-modify-write; `queuedAt` stamps it so resume
|
||||
// replays only the newest queued blob (whole-blob LWW). The returned flag
|
||||
// reports whether the intent is now durably held, so a legacy replay can
|
||||
@@ -632,27 +649,25 @@ export class WholeBlobSyncManager<S> {
|
||||
if (this.foldSupersedingAttemptWinner(remote)) {
|
||||
return { kind: "publish", store, fetchedRemote: remote };
|
||||
}
|
||||
// Failed-bootstrap exception (P2a): the edit was queued while bootstrap
|
||||
// was in-flight and bootstrap subsequently FAILED — no head was ever
|
||||
// fetched, so publishBaseline and lastRemoteHead were never advanced.
|
||||
// releaseDeferred() re-freezes publishBaseline from the bootstrap-result
|
||||
// head, but when bootstrap failed that is still {0,""}. The FIRST head
|
||||
// the relay returns at pre-publish time is the TRUE first established
|
||||
// baseline; publish above it rather than adopting it away.
|
||||
// Failed-bootstrap exception (P2a): when bootstrap failed, the FIRST
|
||||
// relay head at pre-publish time is the true first established baseline;
|
||||
// publish above it rather than adopting it away.
|
||||
//
|
||||
// Guard: this exception fires only when no external head was independently
|
||||
// observed after the edit was queued (subscribeLive or a reconnect/periodic
|
||||
// fetch set bootstrapFailedExternalHeadObserved). An independently-observed
|
||||
// head is a genuine post-click remote advance — the normal
|
||||
// remoteAdvancedSince check above correctly adopts it.
|
||||
// Three guards: (1) no external head observed after the failed bootstrap
|
||||
// (subscribeLive or reconnect/periodic fetch set
|
||||
// bootstrapFailedExternalHeadObserved — that head is a genuine post-click
|
||||
// advance that normal remoteAdvancedSince correctly adopts); (2) this
|
||||
// is NOT a restored outbox replay (pendingIsRestoredReplay) — the
|
||||
// exception was designed for fresh in-session clicks only; letting it
|
||||
// fire for a replay publishes the stale edit over a newer peer head
|
||||
// retained while offline (Carl P1); (3) publishBaseline is still {0,""}.
|
||||
//
|
||||
// (For a successful bootstrap, releaseDeferred sets publishBaseline to the
|
||||
// fetched head before scheduling the timer, so this branch is not reached
|
||||
// for successful bootstraps whose head was already known when the debounce
|
||||
// fired.)
|
||||
// fetched head before the timer fires; this branch is not reached.)
|
||||
if (
|
||||
this.bootstrapFailed &&
|
||||
!this.bootstrapFailedExternalHeadObserved &&
|
||||
!this.pendingIsRestoredReplay &&
|
||||
this.publishBaseline.createdAt === 0 &&
|
||||
this.publishBaseline.eventId === ""
|
||||
) {
|
||||
@@ -677,21 +692,10 @@ export class WholeBlobSyncManager<S> {
|
||||
|
||||
/**
|
||||
* After a publish OK, fetch the authoritative retained head and decide whether
|
||||
* our event actually became it. The relay OKs a superseded NIP-33 write as a
|
||||
* no-op (`Duplicate`), so a resolved publish is NOT proof of retention: two
|
||||
* windows stamping the same `created_at` both get OK while only the lower
|
||||
* event id is retained (Carl P1). Only an exact event-id match proves our
|
||||
* whole-blob write won LWW.
|
||||
*
|
||||
* - head id === ours → `confirmed`
|
||||
* - head is our own accepted prior → `retain` (a lost-ACK / same-second
|
||||
* attempt (ambiguousAttemptIds) lower-id attempt of ours won; the retry
|
||||
* folds it forward and republishes above)
|
||||
* - head is a different readable → `adopt` (our blob lost; adopt the
|
||||
* event winner so UI and relay converge)
|
||||
* - fetch failed / no head / head → `retain` (retention unprovable or the
|
||||
* unreadable winner unreadable — keep the durable
|
||||
* edit and retry)
|
||||
* our event is it. The relay OKs a superseded NIP-33 write as a no-op, so
|
||||
* two concurrent windows can both get OK while only one blob is retained.
|
||||
* Returns `confirmed` on id match, `adopt` for a different readable head,
|
||||
* `retain` when unprovable (fetch failed / unreadable / our prior attempt).
|
||||
*/
|
||||
private async confirmRetainedHead(
|
||||
ourEventId: string,
|
||||
@@ -742,14 +746,8 @@ export class WholeBlobSyncManager<S> {
|
||||
if (gen !== this.pendingGeneration) return;
|
||||
try {
|
||||
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.
|
||||
// Guard: manager may have been destroyed or newer edit queued during fetch.
|
||||
if (this.destroyed) return;
|
||||
// A newer edit was queued while we awaited the pre-publish fetch. It owns
|
||||
// convergence now; abort so we neither publish this stale store nor adopt
|
||||
// over the newer pending edit. The serialized cycle re-drives for the
|
||||
// newer generation once this one unwinds.
|
||||
if (gen !== this.pendingGeneration) return;
|
||||
if (decision.kind === "adopt") {
|
||||
this.adoptRemote(decision.remote, gen);
|
||||
|
||||
Reference in New Issue
Block a user