jerry/buzz
1
0
Fork 0
mirror of https://github.com/block/buzz.git synced 2026-11-01 06:51:17 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(read-state): close round-7 findings — bounded abort retry, gen exhaustion, total tx closure

Browse Source 
Round 7 fixes for PR #3966 (Thufir's BLOCKED verdict at 35873afea3):

1. Bounded drain-retry controller: scheduleAbortRetry() gives abort retries
   their own escalating timer (1s→2s→4s…60s), coalesced per manager, cancelled
   on destroy() and identity change. Abort paths call scheduleAbortRetry(); only
   the success-path gen2 promotion uses scheduleDrain() (immediate). No microtask
   spin loop under persistent storage failure.

2. Generation exhaustion without reuse: deleted nextGen=1 ceiling reset.
   Hydration now compacts/rebases all pending intent gens and their matching
   receipt gens into a fresh safe range [1..N] when maxObservedGen is at or
   above MAX_SAFE_INTEGER; nextGen = N+1 after rebase. Both enqueue() and
   promoteDeferred() guard the _nextGen++ with Number.isSafeInteger(gen).

3. Total transaction closure: onDrainOutcome wrapped in try/catch — thrown
   callback triggers the same receipt-aware abortTransaction + scheduleAbortRetry
   path as the storage-failure abort, not a silent latch-forever exit.
   restoreFromStorage() clears all transaction-only maps (clearTransactionState)
   so no lock/snapshot/deferred/promoted state survives an identity swap.
   destroy() calls clearTransactionState() for the same reason.

4. Witnesses to contract (Tests 51/52/37):
   - Tests 51/52: enqueue gen2 from onDrainOutcome (isLoadComplete toggle forces
     deferred latch path); fail exactly one write; let real bounded scheduler fire
     retry (no manual intercept); assert intermediate blob, recovered combined
     gen1-cleanup+gen2-pi commit, restart parity.
   - Test 37: Phase 5 sends TWO reconnects while gen-2 held; Phase 6 releases
     secondBarrier, asserts gen advances to gen1+2, fetchCallCount=3 before
     destroy. All assertions unconditional.

Co-authored-by: Will Pfleger <pfleger.will@gmail.com>
Signed-off-by: Will Pfleger <pfleger.will@gmail.com>
This commit is contained in:
Duncan
2026-08-08 16:26:45 -04:00
parent 35873afea3
commit 4002067de3
5 changed files with 543 additions and 283 deletions
Download Patch File Download Diff File Expand all files Collapse all files

34
desktop/src/features/channels/pendingOverrideIntents.ts
View File

@@ -142,6 +142,9 @@ export class PendingOverrideIntentStore {
/**
* Restore in-memory state from deserialized storage (called by manager
* during `hydrateFromLocalStorage`). Replaces any in-memory state.
*
* Clears ALL transaction-only maps so no lock, snapshot, deferred payload,
* or promoted payload from a prior identity can survive a hydration/swap.
*/
restoreFromStorage(
intents: ReadonlyMap<string, PendingIntent>,
@@ -149,6 +152,8 @@ export class PendingOverrideIntentStore {
): void {
this.intents = new Map(intents);
this._nextGen = nextGen;
// Clear transaction-only state — identity swap must not inherit A's locks.
this.clearTransactionState();
}
/**
@@ -192,6 +197,13 @@ export class PendingOverrideIntentStore {
};
}
const gen = this._nextGen++;
if (!Number.isSafeInteger(gen)) {
// Allocation exhausted — this should never happen in practice (the store
// is rebased at hydration when near the ceiling), but guard defensively.
throw new Error(
"[PendingOverrideIntentStore] generation counter exhausted (unsafe)",
);
}
const intent: PendingIntent = {
gen,
op,
@@ -283,6 +295,12 @@ export class PendingOverrideIntentStore {
// Save the payload before promoting so abortTransaction can re-buffer it.
this.promotedDeferredPayloads.set(channelId, deferred);
const gen = this._nextGen++;
if (!Number.isSafeInteger(gen)) {
// Guard: rebase at hydration prevents this in practice, but fail safely.
throw new Error(
"[PendingOverrideIntentStore] generation counter exhausted (unsafe)",
);
}
const intent: PendingIntent = {
gen,
op: deferred.op,
@@ -368,6 +386,22 @@ export class PendingOverrideIntentStore {
get size(): number {
return this.intents.size;
}
/**
* Clear all transaction-only state.
*
* Called by `restoreFromStorage` (identity swap) and `ReadStateManager.destroy()`
* so no lock, snapshot, deferred payload, or promoted payload can escape past a
* manager lifecycle boundary on the shared module singleton.
*
* Does NOT touch the `intents` map or `_nextGen` — caller owns those.
*/
clearTransactionState(): void {
this.lockedChannels.clear();
this.intentSnapshots.clear();
this.deferredEnqueues.clear();
this.promotedDeferredPayloads.clear();
}
}
/** Singleton store — one per renderer process, shared across all manager instances. */

29
desktop/src/features/channels/readState/readStateDrain.ts
View File

@@ -73,6 +73,13 @@ export interface DrainContext {
tryCandidatePlan(rawCtxId: string, reg: OverrideRegister): boolean;
restoreExtraSlotIds(prev: string[]): void;
scheduleDrain(): void;
/**
* Schedule a bounded-backoff abort retry drain pass.
* Used on storage-failure and thrown-callback abort paths — these must NOT
* use the immediate `pendingFreshDrain` path to prevent a hot spin loop
* under persistent storage failure. Mirrors the load retry controller.
*/
scheduleAbortRetry(): void;
/** Drain outcome callback — typed discriminated union for exhaustive handling
* in the hook layer. */
onDrainOutcome: ((outcome: DrainOutcome) => void) | null;
@@ -100,6 +107,7 @@ interface ManagerInternals {
tryCandidatePlan(rawCtxId: string, reg: OverrideRegister): boolean;
restoreExtraSlotIds(prev: string[]): void;
scheduleDrain(): void;
scheduleAbortRetry(): void;
}
/**
@@ -144,6 +152,7 @@ export function createDrainContext(mgr: ManagerInternals): DrainContext {
tryCandidatePlan: (id, reg) => mgr.tryCandidatePlan(id, reg),
restoreExtraSlotIds: (prev) => mgr.restoreExtraSlotIds(prev),
scheduleDrain: () => mgr.scheduleDrain(),
scheduleAbortRetry: () => mgr.scheduleAbortRetry(),
};
}
@@ -253,7 +262,7 @@ export async function drainPendingIntents(
ctx.restoreExtraSlotIds(prevExtraSlotIds);
// Abort: gen1 restored as live intent, gen2 stays buffered for retry.
pendingOverrideIntentStore.abortTransaction(channelId);
if (!ctx.destroyed) ctx.scheduleDrain();
if (!ctx.destroyed) ctx.scheduleAbortRetry();
continue; // transaction closed via abort; finally is a no-op (transactionCommitted=false)
}
ctx.schedulePublish();
@@ -306,8 +315,22 @@ export async function drainPendingIntents(
// Fire the typed callback.
// The transaction latch ensures no enqueue() for this channel can change
// the generation inside this callback — Amendment A holds structurally.
// Wrap in try-catch: a thrown callback must not leave the channel latched
// forever. On throw, treat the same as a storage-failure abort: restore
// gen1 as live intent, keep gen2 buffered, schedule a bounded retry.
if (outcome !== null && ctx.onDrainOutcome !== null) {
ctx.onDrainOutcome(outcome);
try {
ctx.onDrainOutcome(outcome);
} catch (err) {
console.warn(
`[ReadStateManager] drain: onDrainOutcome threw for ${channelId}:`,
err,
);
// Abort the transaction: gen1 restored, gen2 stays buffered.
pendingOverrideIntentStore.abortTransaction(channelId);
if (!ctx.destroyed) ctx.scheduleAbortRetry();
continue; // transaction closed via abort; finally is a no-op
}
}
// Step 3: atomic cleanup commit — promote any deferred gen2 enqueue, then
@@ -341,7 +364,7 @@ export async function drainPendingIntents(
ctx.appliedReceipts.set(channelId, capturedReceipt);
}
pendingOverrideIntentStore.abortTransaction(channelId);
if (!ctx.destroyed) ctx.scheduleDrain();
if (!ctx.destroyed) ctx.scheduleAbortRetry();
continue; // transaction closed via abort; finally is a no-op
}
transactionCommitted = true;

629
desktop/src/features/channels/readState/readStateManager.test.mjs
View File

@@ -4075,25 +4075,47 @@ test("retryLoad_reconnectDuringLoadInFlight_coalescesToPendingRetry", async () =
);
assert.equal(fetchCallCount, 2, "exactly two fetches must have occurred");
// ── Phase 5: gen-2 fetch is in-flight — reconnect coalesces unconditionally ─
// ── Phase 5: gen-2 fetch is in-flight — multiple reconnects coalesce ─────────
// loadInFlight must still be true because secondBarrier has not been released.
assert.equal(
mgr.loadInFlight,
true,
"gen-2 load must still be in-flight (secondBarrier not released)",
);
// This reconnect MUST always be asserted — no `if (loadInFlight)` guard.
// First reconnect — unconditional, no `if (loadInFlight)` guard.
mgr.retryLoad();
assert.equal(
mgr.pendingRetryOnComplete,
true,
"reconnect during gen-2 load must unconditionally set pendingRetryOnComplete",
"first reconnect during gen-2 load must unconditionally set pendingRetryOnComplete",
);
// Second reconnect — must also coalesce (not start a third fetch immediately).
mgr.retryLoad();
assert.equal(
mgr.pendingRetryOnComplete,
true,
"second reconnect during gen-2 load must keep pendingRetryOnComplete set",
);
assert.equal(
mgr.loadGeneration,
gen1 + 1,
"gen must not advance on coalesced reconnects during gen-2 in-flight",
);
// ── Phase 6: release gen-2 fetch and destroy ──────────────────────────────
// ── Phase 6: release gen-2 fetch — pending retry fires gen-3 load ─────────
// The pending retry means the finally-block fires retryLoad() immediately after
// gen-2 resolves, starting a gen-3 fetch. Assert gen-3 fires BEFORE destroy.
resolveSecond();
await new Promise((r) => origSetTimeout(r, 0));
// Gen-3 fetch must be in-flight (started by the pending retry on gen-2 complete).
assert.equal(
mgr.loadGeneration,
gen1 + 2,
"gen must advance to gen1+2 when pending retry fires after gen-2 complete",
);
assert.equal(fetchCallCount, 3, "exactly three fetches must have occurred");
mgr.destroy();
assert.equal(
mgr.retryBackoffTimer,
@@ -5181,29 +5203,51 @@ test("markChannelReadDirect_uint32overflow_frontierDeactivated_silentSuccess", a
// ── Test 51: abort/retry on step-1 storage failure — gen1 survives, retry drains ──
test("drain_step1StorageFailure_abortRetry_gen1SurvivesAndRetrains", async () => {
// If the step-1 register+receipt persist fails, the drain aborts the transaction:
// gen1 stays live in the store, deferred gen2 stays buffered, and scheduleDrain()
// fires a retry pass. On the retry, storage is healthy and the drain completes.
// gen1 stays live, no outcome callback fires (step-1 abort skips steps 2-3 via `continue`),
// and scheduleAbortRetry() fires a bounded-backoff timer. When the timer fires, storage is
// healthy: gen1 drains fully; the outcome callback enqueues gen2 (deferred while channel is
// locked); promoteDeferred() commits gen1-cleanup + gen2-pi atomically; scheduleDrain() fires
// gen2 drain immediately; gen2 applies (frontier=999).
//
// Failure branches covered:
// - step-1 failure path: abort + bounded retry timer (no immediate spin)
//
// Assertions (durable surfaces):
// - blob has gen1 pi at the start (from enqueue persist before drain)
// - restart witness: hydrating after step-1 failure sees gen1 (not gen2)
// - after the retry pass: register committed, gen1+gen2 cleaned up, queue empty,
// blob has no pi
//
// Timing note: the abort path schedules a retry drain as a microtask continuation
// inside the same awaited drain call. We suppress the auto-retry by momentarily
// clearing pendingFreshDrain after the failing drain, verify durability and restart,
// then manually fire the retry to prove gen1 actually drains successfully.
// - blob pre-drain: gen1 pi (no register)
// - post-fail: blob unchanged (step-1 abort rolled back register), gen1 live in memory
// - restart witness 1: blob has gen1 pi; fresh mgr hydrates gen1 (gen2 not yet durable)
// - scheduleAbortRetry sets mgr.drainRetryTimer (bounded backoff, NOT immediate pendingFreshDrain)
// - retry fires: gen1 drains + gen2 auto-drains (scheduleDrain path)
// - post-retry: register committed (from gen1), frontier ≥ 999 (gen2 applied), no pending intent
// - restart witness 2: fresh mgr hydrates frontier ≥ 999; no stale gen1 intent
globalThis.window.localStorage = makeLocalStorage();
const pubkey = "51".repeat(32);
const channelId = `abort-step1-ch-${"w".repeat(49)}`;
const v2Key = `buzz.nip-rs.override-state.v2:${pubkey}`;
const ls = globalThis.window.localStorage;
// Block ALL writes to the v2 key until `unblockV2Write()` is called.
// setLocalStorageItemWithRecovery retries after catching, so we must block
// both the initial write AND the recovery retry — a single-shot flag is not
// enough. A persistent-block flag is cleared explicitly by the test.
// Fake timer: intercept ms>0 timers for scheduleAbortRetry control.
const origSetTimeout = globalThis.window.setTimeout;
const origClearTimeout = globalThis.window.clearTimeout;
const scheduledTimers = new Map(); // id → fn
let nextFakeId = 51000;
globalThis.window.setTimeout = (fn, ms) => {
if (ms > 0) {
const id = nextFakeId++;
scheduledTimers.set(id, fn);
return id;
}
return origSetTimeout(fn, ms);
};
globalThis.window.clearTimeout = (id) => {
if (scheduledTimers.has(id)) {
scheduledTimers.delete(id);
} else {
origClearTimeout(id);
}
};
// Block ALL v2 writes until explicitly unblocked (step-1 fail simulation).
let blockV2Writes = false;
const origSetItem = ls.setItem.bind(ls);
ls.setItem = (key, value) => {
@@ -5222,150 +5266,231 @@ test("drain_step1StorageFailure_abortRetry_gen1SurvivesAndRetrains", async () =>
getConnectionGeneration: () => 0,
};
const mgr = new ReadStateManager(pubkey, fakeRelay);
mgr.effectiveState.set(channelId, 50);
mgr.isLoadComplete = false;
try {
const mgr = new ReadStateManager(pubkey, fakeRelay);
mgr.effectiveState.set(channelId, 50);
mgr.isLoadComplete = false;
// Queue gen1 (unread).
mgr.markChannelUnread(channelId);
const gen1Intent = pendingOverrideIntentStore.get(channelId);
assert.ok(gen1Intent, "gen1 must be queued");
const gen1 = gen1Intent.gen;
// Queue gen1 (unread) — also persists initial pi (write 0).
mgr.markChannelUnread(channelId);
const gen1 = pendingOverrideIntentStore.get(channelId).gen;
// The v2 blob must have gen1 pi from the initial enqueue persist (write 0).
// Check this BEFORE the failing drain so we know the baseline durable state.
const blobBeforeDrain = JSON.parse(ls.getItem(v2Key) ?? "null") ?? {};
assert.ok(
blobBeforeDrain.pi?.[channelId],
"v2 blob must carry gen1 pi before drain",
);
assert.equal(
blobBeforeDrain.pi[channelId].gen,
gen1,
"blob pi gen must match gen1",
);
// Intercept scheduleDrain to capture any auto-retry fired by the abort path.
// The step-1 abort calls scheduleDrain(), which the manager honors by firing a
// fire-and-forget retry drain. We capture it here to prevent it from running
// before we check intermediate state.
let capturedRetryFn = null;
const origScheduleDrain = mgr.scheduleDrain.bind(mgr);
mgr.scheduleDrain = () => {
// Save the retry request — fire it manually after intermediate assertions.
capturedRetryFn = () => {
mgr.scheduleDrain = origScheduleDrain;
origScheduleDrain();
};
};
// Complete load, but fail the step-1 register+receipt write by blocking ALL
// v2 writes (setLocalStorageItemWithRecovery would retry after eviction, so
// a single-shot flag is insufficient — a persistent block is required).
mgr.isLoadComplete = true;
blockV2Writes = true;
await mgr.drainPendingIntents(mgr.loadGeneration);
// Unblock writes so subsequent operations can persist.
blockV2Writes = false;
// The v2 blob must still have gen1 pi (step-1 write failed — blob unchanged from write 0).
const blobAfterFail = JSON.parse(ls.getItem(v2Key) ?? "null") ?? {};
assert.ok(
blobAfterFail.pi?.[channelId],
"v2 blob must still carry gen1 pi after step-1 failure",
);
assert.equal(
blobAfterFail.pi[channelId].gen,
gen1,
"blob gen must still be gen1 after failure",
);
// The register must NOT be committed (step-1 failed before persist).
assert.equal(
mgr.overrideRegisters.get(channelId),
undefined,
"register must not be committed after step-1 failure",
);
// Restart witness: a fresh manager sees gen1 (not a stale or promoted gen2).
{
const mgr2 = new ReadStateManager(pubkey, fakeRelay);
mgr2.hydrateFromLocalStorage();
const hydratedIntent = pendingOverrideIntentStore.get(channelId);
assert.ok(hydratedIntent, "gen1 must be hydrated from blob on restart");
// Confirm v2 blob has gen1 pi from write 0.
const blobBefore = JSON.parse(ls.getItem(v2Key) ?? "null") ?? {};
assert.equal(
hydratedIntent.gen,
blobBefore.pi?.[channelId]?.gen,
gen1,
"hydrated gen must be gen1 (abort preserved durable state)",
"blob must carry gen1 pi before drain",
);
mgr2.destroy();
// Wire drain callback: enqueue gen2 (read, readTarget=999) from inside the transaction.
// Force isLoadComplete=false temporarily so markChannelRead goes through the latch buffer
// (deferred enqueue path) — the channel is locked at callback time, so the enqueue is
// buffered as deferred rather than applied immediately. This is the scenario the latch
// is designed for: a user action arrives mid-transaction while the drain holds the lock.
mgr.onDrainOutcome = (outcome) => {
if (
outcome.kind === "applied-unread" &&
outcome.channelId === channelId
) {
const prev = mgr.isLoadComplete;
mgr.isLoadComplete = false; // route through deferred latch path
mgr.markChannelRead(channelId, undefined, 999);
mgr.isLoadComplete = prev;
// The enqueue returned gen:-1 (buffered placeholder); real gen assigned on promote.
}
};
// NOTE: step-1 failure skips the outcome callback entirely (abort + continue before step 2).
// The callback only fires during the retry drain (when step-1 succeeds).
// Drain with step-1 blocked.
mgr.isLoadComplete = true;
blockV2Writes = true;
await mgr.drainPendingIntents(mgr.loadGeneration);
blockV2Writes = false;
// After failing drain: register NOT committed (step-1 rolled back), gen1 live.
assert.equal(
mgr.overrideRegisters.get(channelId),
undefined,
"register must not be committed after step-1 failure",
);
// v2 blob must still be write-0 state: gen1 pi, no register.
const blobAfterFail = JSON.parse(ls.getItem(v2Key) ?? "null") ?? {};
assert.equal(
blobAfterFail.pi?.[channelId]?.gen,
gen1,
"blob must still have gen1 pi after step-1 failure",
);
assert.equal(
blobAfterFail.r?.[channelId],
undefined,
"register must not be in blob after step-1 failure",
);
// Gen1 must be live in the store (abort restored it).
const intentAfterFail = pendingOverrideIntentStore.get(channelId);
assert.ok(intentAfterFail, "gen1 must be live after abort");
assert.equal(
intentAfterFail.gen,
gen1,
"live intent must be gen1 after step-1 abort",
);
// scheduleAbortRetry must have set mgr.drainRetryTimer (bounded backoff timer).
// Step-1 fail skips schedulePublish (step-1 is the register write — on failure,
// the register is rolled back and no publish is scheduled). Exactly one timer.
assert.equal(
scheduledTimers.size,
1,
"exactly one abort retry timer after step-1 failure",
);
assert.ok(
mgr.drainRetryTimer !== null,
"mgr.drainRetryTimer must be set by scheduleAbortRetry",
);
assert.ok(
scheduledTimers.has(mgr.drainRetryTimer),
"abort retry timer ID must be in scheduledTimers",
);
// Restart witness 1: blob has gen1 pi; hydrating sees gen1 (gen2 not yet durable).
{
const mgr2 = new ReadStateManager(pubkey, fakeRelay);
mgr2.hydrateFromLocalStorage();
const hydratedIntent = pendingOverrideIntentStore.get(channelId);
assert.ok(
hydratedIntent,
"gen1 must hydrate from blob (step-1 fail leaves gen1 pi in blob)",
);
assert.equal(hydratedIntent.gen, gen1, "hydrated intent must be gen1");
mgr2.destroy();
}
// Restore for original mgr (hydration above replaced singleton state).
mgr.hydrateFromLocalStorage();
// Fire the abort retry timer — storage is healthy now.
// Use mgr.drainRetryTimer to get the specific abort retry timer (not publish debounce).
const drainRetryTimerId = mgr.drainRetryTimer;
const retryFn = scheduledTimers.get(drainRetryTimerId);
assert.ok(retryFn, "abort retry timer fn must be in scheduledTimers");
scheduledTimers.clear(); // clear all before firing
await retryFn(); // fires the drain retry
// The retry drain is async but has no awaits internally — drainPendingIntentsImpl runs
// synchronously. The gen2 auto-drain (via scheduleDrain/pendingFreshDrain) also runs
// synchronously within the outer drainPendingIntents wrapper microtask.
// One setTimeout(0) macrotask is sufficient to let all microtasks settle.
await new Promise((r) => origSetTimeout(r, 0));
// After retry + auto-drain:
// - gen1 register committed (S-bumped from gen1 unread drain)
// - gen2 (read, readTarget=999) promoted, committed atomically with gen1 cleanup,
// then auto-drained via scheduleDrain → frontier advances to 999
// - no pending intent remains (both gen1 and gen2 consumed)
// - blob: register with C-bumped (from gen2 read drain), no pi, no receipt
const regAfterRetry = mgr.overrideRegisters.get(channelId);
assert.ok(regAfterRetry, "register must be committed after retry");
assert.ok(regAfterRetry.s > 0, "S must be bumped (gen1 unread applied)");
// Receipt must be gone (step-3 cleanup succeeded).
const blobAfterRetry = JSON.parse(ls.getItem(v2Key) ?? "null") ?? {};
assert.equal(
blobAfterRetry.receipts?.[channelId],
undefined,
"receipt must be cleaned up after retry",
);
// Frontier must have advanced to 999 (gen2 read applied).
const frontierAfterRetry = mgr.effectiveState.get(channelId) ?? 0;
assert.ok(
frontierAfterRetry >= 999,
`frontier must be ≥ 999 after gen2 drain; got ${frontierAfterRetry}`,
);
// No pending intent (both gen1 and gen2 consumed by drain).
assert.equal(
pendingOverrideIntentStore.get(channelId),
undefined,
"no pending intent after gen1+gen2 drain",
);
// Restart witness 2: fresh manager sees frontier ≥ 999 (gen2 durably applied);
// no stale gen1 intent. Restart parity: durable state reflects what memory holds.
{
const mgr3 = new ReadStateManager(pubkey, fakeRelay);
mgr3.hydrateFromLocalStorage();
// No pending intent (both drained).
assert.equal(
pendingOverrideIntentStore.get(channelId),
undefined,
"fresh manager: no pending intent after full drain",
);
// Frontier must be persisted at ≥ 999 (stored in blob.r[channelId].f).
const f3 = mgr3.effectiveState.get(channelId) ?? 0;
assert.ok(
f3 >= 999,
`fresh manager must hydrate frontier ≥ 999; got ${f3}`,
);
mgr3.destroy();
}
mgr.destroy();
} finally {
globalThis.window.setTimeout = origSetTimeout;
globalThis.window.clearTimeout = origClearTimeout;
}
// Manually fire the retry drain (storage is healthy now — no more blocks).
assert.ok(
capturedRetryFn,
"scheduleDrain must have been called by the abort path",
);
mgr.scheduleDrain = origScheduleDrain; // restore before firing
mgr.isLoadComplete = true;
await mgr.drainPendingIntents(mgr.loadGeneration);
// After retry: register committed, queue empty, blob cleaned.
const regAfterRetry = mgr.overrideRegisters.get(channelId);
assert.ok(regAfterRetry, "register must be committed after retry drain");
assert.ok(regAfterRetry.s > 0, "S must be bumped after retry");
assert.equal(
pendingOverrideIntentStore.get(channelId),
undefined,
"gen1 must be consumed after retry drain",
);
const blobAfterRetry = JSON.parse(ls.getItem(v2Key) ?? "null") ?? {};
assert.equal(
blobAfterRetry.pi?.[channelId],
undefined,
"v2 blob must have no pi entry after retry cleanup",
);
mgr.destroy();
});
// ── Test 52: abort/retry on step-3 cleanup failure — gen1 stays live, retry cleans up ──
test("drain_step3CleanupFailure_abortRetry_gen1StaysLiveAndRetrains", async () => {
// If the step-3 cleanup write fails (after step-1 register+receipt committed),
// abortTransaction restores gen1 as the live intent and keeps gen2 buffered.
// The in-memory receipt is also restored so the retry drain's alreadyApplied
// check fires (cleanup-only, no re-bump). scheduleDrain fires a retry.
// abortTransaction restores gen1 as the live intent and re-buffers gen2 (enqueued
// from onDrainOutcome during the transaction). The in-memory receipt is also
// restored so the retry drain's alreadyApplied check fires (cleanup-only, no re-bump).
// scheduleAbortRetry fires a bounded-backoff timer. On retry: combined gen1-cleanup
// + gen2-pi committed atomically. Restart must see gen2.
//
// Assertions (durable surfaces):
// - after failing pass (auto-retry suppressed): register in blob (step-1 ok),
// gen1 pi still in blob (cleanup write failed), gen1 pi+receipt in blob
// - gen1 restored as live in-memory intent (abortTransaction)
// - restart witness: gen1 pi+receipt in blob; hydration sees gen1 as pending
// - after retry pass: pi cleaned, queue empty, blob has no pi
//
// Timing: same as Test 51 — we intercept scheduleDrain to prevent the auto-retry
// from running before intermediate assertions, then manually fire the retry.
// - after failing pass: register in blob (step-1 ok), gen1 pi + receipt still in blob
// (cleanup write failed), gen1 live in memory
// - restart witness: gen1 + receipt in blob; hydration sees gen1 pending
// - timer fires: retry drain runs alreadyApplied (no re-bump), gen1-cleanup + gen2-pi atomic
// - post-retry: pi cleaned, receipt cleaned, queue has gen2, blob has gen2 pi
// - restart witness 2: fresh manager hydrates gen2 from blob
globalThis.window.localStorage = makeLocalStorage();
const pubkey = "52".repeat(32);
const channelId = `abort-step3-ch-${"x".repeat(49)}`;
const v2Key = `buzz.nip-rs.override-state.v2:${pubkey}`;
const ls = globalThis.window.localStorage;
// We need to fail exactly write 2 (step-3 cleanup) while writes 0 and 1 succeed.
// setLocalStorageItemWithRecovery catches throws and retries once after cache eviction.
// In our clean test environment eviction returns 0 entries, so no retry occurs —
// a single throw per write attempt is sufficient for the specific numbered-write approach.
// BUT: the throw-on-count approach only works because our test localStorage has no
// pure-cache keys (evictPureCacheEntries returns 0).
//
// Write ordering:
// Write 0: enqueue persist (markChannelUnread, incomplete load) ← succeeds
// Write 1: step-1 register+receipt commit (drain) ← succeeds
// Write 2: step-3 cleanup commit (drain) ← FAILS (persistent block from this index)
// Fake timer for scheduleAbortRetry.
const origSetTimeout = globalThis.window.setTimeout;
const origClearTimeout = globalThis.window.clearTimeout;
const scheduledTimers = new Map();
let nextFakeId = 52000;
globalThis.window.setTimeout = (fn, ms) => {
if (ms > 0) {
const id = nextFakeId++;
scheduledTimers.set(id, fn);
return id;
}
return origSetTimeout(fn, ms);
};
globalThis.window.clearTimeout = (id) => {
if (scheduledTimers.has(id)) {
scheduledTimers.delete(id);
} else {
origClearTimeout(id);
}
};
// Write 0: enqueue persist (succeeds)
// Write 1: step-1 register+receipt commit (succeeds)
// Write 2: step-3 cleanup commit (FAILS once, then succeeds)
let v2WriteCount = 0;
let blockFromWriteIndex = -1; // block all v2 writes from this index onward; -1 = unblocked
let blockFromWriteIndex = -1;
const origSetItem = ls.setItem.bind(ls);
ls.setItem = (key, value) => {
if (key.startsWith("buzz.nip-rs.override-state.v2:")) {
@@ -5386,95 +5511,153 @@ test("drain_step3CleanupFailure_abortRetry_gen1StaysLiveAndRetrains", async () =
getConnectionGeneration: () => 0,
};
const mgr = new ReadStateManager(pubkey, fakeRelay);
mgr.effectiveState.set(channelId, 50);
mgr.isLoadComplete = false;
try {
const mgr = new ReadStateManager(pubkey, fakeRelay);
mgr.effectiveState.set(channelId, 50);
mgr.isLoadComplete = false;
// Queue gen1 (unread). This persists the initial pi (write 0).
mgr.markChannelUnread(channelId);
const gen1 = pendingOverrideIntentStore.get(channelId).gen;
// Queue gen1 (unread) — write 0 (enqueue persist).
mgr.markChannelUnread(channelId);
const gen1 = pendingOverrideIntentStore.get(channelId).gen;
// Intercept scheduleDrain to prevent the auto-retry from running before we
// check intermediate state. Same pattern as Test 51.
let capturedRetryFn = null;
const origScheduleDrain = mgr.scheduleDrain.bind(mgr);
mgr.scheduleDrain = () => {
capturedRetryFn = () => {
mgr.scheduleDrain = origScheduleDrain;
origScheduleDrain();
// Wire callback: enqueue gen2 (read, readTarget=999) from inside the transaction.
// Force isLoadComplete=false temporarily so markChannelRead goes through the
// deferred latch path (enqueue() with locked channel → deferredEnqueues buffer).
// Without this toggle, markChannelRead uses the direct path (isLoadComplete=true),
// bypasses enqueue(), and applies the register change immediately outside the latch.
mgr.onDrainOutcome = (outcome) => {
if (
outcome.kind === "applied-unread" &&
outcome.channelId === channelId
) {
const prev = mgr.isLoadComplete;
mgr.isLoadComplete = false; // route through deferred latch path
mgr.markChannelRead(channelId, undefined, 999);
mgr.isLoadComplete = prev;
// The enqueue returned gen:-1 (buffered placeholder); real gen assigned on promote.
}
};
};
// Enable failure from write index 2 (step-3 cleanup).
// Write 1 (step-1 register+receipt) must still succeed.
blockFromWriteIndex = 2;
mgr.isLoadComplete = true;
await mgr.drainPendingIntents(mgr.loadGeneration);
// Unblock writes so subsequent operations can persist.
blockFromWriteIndex = -1;
// Block writes starting at write index 2 (step-3 cleanup).
blockFromWriteIndex = 2;
mgr.isLoadComplete = true;
await mgr.drainPendingIntents(mgr.loadGeneration);
blockFromWriteIndex = -1;
// After failing pass: register must be committed (step-1 succeeded).
const reg = mgr.overrideRegisters.get(channelId);
assert.ok(reg, "register must be committed after step-1 succeeds");
assert.ok(reg.s > 0, "S must be bumped");
// After failing pass: register committed (step-1 ok), gen1 pi + receipt in blob (cleanup failed).
const reg = mgr.overrideRegisters.get(channelId);
assert.ok(reg, "register must be committed after step-1 succeeds");
assert.ok(reg.s > 0, "S must be bumped");
// Blob must have the register but gen1 pi must still be present (cleanup failed).
// The blob reflects write 1 (register + receipt + pi) since write 2 was blocked.
const blobAfterFail = JSON.parse(ls.getItem(v2Key) ?? "null") ?? {};
assert.ok(
blobAfterFail.r?.[channelId],
"register must be in blob after step-1 success",
);
assert.ok(
blobAfterFail.pi?.[channelId],
"gen1 pi must still be in blob (cleanup write failed)",
);
// The receipt must also be in the blob (write 1 included it; write 2 failed to remove it).
assert.ok(
blobAfterFail.receipts?.[channelId],
"receipt must still be in blob (cleanup write failed)",
);
const blobAfterFail = JSON.parse(ls.getItem(v2Key) ?? "null") ?? {};
assert.ok(
blobAfterFail.r?.[channelId],
"register must be in blob after step-1 success",
);
assert.ok(
blobAfterFail.pi?.[channelId],
"gen1 pi must still be in blob (cleanup write failed)",
);
assert.ok(
blobAfterFail.receipts?.[channelId],
"receipt must still be in blob (cleanup write failed)",
);
// Gen1 must be restored as the live intent (abortTransaction).
const intentAfterFail = pendingOverrideIntentStore.get(channelId);
assert.ok(intentAfterFail, "gen1 must be live after step-3 abort");
assert.equal(intentAfterFail.gen, gen1, "live intent must be gen1");
// Gen1 must be live in memory (abortTransaction restored it).
const intentAfterFail = pendingOverrideIntentStore.get(channelId);
assert.ok(intentAfterFail, "gen1 must be live after step-3 abort");
assert.equal(intentAfterFail.gen, gen1, "live intent must be gen1");
// Restart witness: hydrating sees gen1 pi + receipt; next drain runs Amendment C
// cleanup without re-bumping (register already committed).
{
const mgr2 = new ReadStateManager(pubkey, fakeRelay);
mgr2.hydrateFromLocalStorage();
const hydratedIntent = pendingOverrideIntentStore.get(channelId);
assert.ok(hydratedIntent, "gen1 must be hydrated from blob on restart");
assert.equal(hydratedIntent.gen, gen1, "hydrated gen must be gen1");
mgr2.destroy();
// scheduleAbortRetry must have set the drain retry timer.
// Note: schedulePublish() also fires after step-1 success, so scheduledTimers
// may contain a debounce timer in addition to the abort retry timer.
// Use mgr.drainRetryTimer to identify the specific abort retry timer.
const drainRetryTimerId = mgr.drainRetryTimer;
assert.ok(
drainRetryTimerId !== null,
"scheduleAbortRetry must set mgr.drainRetryTimer",
);
assert.ok(
scheduledTimers.has(drainRetryTimerId),
"abort retry timer ID must be in scheduledTimers",
);
// Restart witness 1: gen1 pi + receipt in blob; hydration sees gen1.
{
const mgr2 = new ReadStateManager(pubkey, fakeRelay);
mgr2.hydrateFromLocalStorage();
const hydratedIntent = pendingOverrideIntentStore.get(channelId);
assert.ok(hydratedIntent, "gen1 must hydrate from blob");
assert.equal(hydratedIntent.gen, gen1, "hydrated must be gen1");
mgr2.destroy();
}
// Restore mgr's intent store state (hydration above replaced singleton).
mgr.hydrateFromLocalStorage();
// Fire the abort retry timer specifically (not the publish debounce timer).
// Storage is healthy (blockFromWriteIndex = -1).
const retryFn = scheduledTimers.get(drainRetryTimerId);
assert.ok(retryFn, "retry timer fn must still be pending");
scheduledTimers.clear(); // clear all (publish debounce + retry) before firing
await retryFn();
await new Promise((r) => origSetTimeout(r, 0));
// After retry: gen1 cleaned up (alreadyApplied path), gen2 promoted+committed atomically,
// gen2 auto-drained via scheduleDrain (frontier advances to 999), then gen2 cleaned up.
// The blob now has: register (C-bumped from gen2 read drain), no pi, no receipt.
// Gen2 pi was in the blob briefly (after atomic gen1-cleanup+gen2-pi commit) but was
// consumed synchronously by the gen2 auto-drain within the same microtask chain.
const blobAfterRetry = JSON.parse(ls.getItem(v2Key) ?? "null") ?? {};
assert.equal(
blobAfterRetry.pi?.[channelId],
undefined,
"pi must be cleaned after retry + gen2 auto-drain",
);
assert.equal(
blobAfterRetry.receipts?.[channelId],
undefined,
"receipt must be cleaned after retry",
);
// Frontier must have advanced to 999 (gen2 read applied by auto-drain).
const frontierAfterRetry = mgr.effectiveState.get(channelId) ?? 0;
assert.ok(
frontierAfterRetry >= 999,
`frontier must be ≥ 999 after gen2 auto-drain; got ${frontierAfterRetry}`,
);
// No pending intent (gen1 + gen2 both consumed).
assert.equal(
pendingOverrideIntentStore.get(channelId),
undefined,
"no pending intent after gen1+gen2 drain",
);
// Restart witness 2: fresh manager sees frontier ≥ 999 (gen2 durably applied);
// no stale gen1 intent. Restart parity: durable state reflects what memory holds.
{
const mgr3 = new ReadStateManager(pubkey, fakeRelay);
mgr3.hydrateFromLocalStorage();
// No pending intent (both drained).
assert.equal(
pendingOverrideIntentStore.get(channelId),
undefined,
"fresh manager: no pending intent after full drain",
);
// Frontier must be persisted at ≥ 999 (stored in blob.r[channelId].f).
const f3 = mgr3.effectiveState.get(channelId) ?? 0;
assert.ok(
f3 >= 999,
`fresh manager must hydrate frontier ≥ 999; got ${f3}`,
);
mgr3.destroy();
}
mgr.destroy();
} finally {
globalThis.window.setTimeout = origSetTimeout;
globalThis.window.clearTimeout = origClearTimeout;
}
// Manually fire the retry drain (storage healthy — block is lifted).
assert.ok(
capturedRetryFn,
"scheduleDrain must have been called by the step-3 abort path",
);
mgr.scheduleDrain = origScheduleDrain; // restore before firing
mgr.isLoadComplete = true;
await mgr.drainPendingIntents(mgr.loadGeneration);
// After retry: gen1 cleaned up, queue empty.
assert.equal(
pendingOverrideIntentStore.get(channelId),
undefined,
"gen1 must be consumed after retry cleanup drain",
);
const blobAfterRetry = JSON.parse(ls.getItem(v2Key) ?? "null") ?? {};
assert.equal(
blobAfterRetry.pi?.[channelId],
undefined,
"v2 blob must have no pi entry after retry",
);
mgr.destroy();
});
// ── Test 53: ng normalization — gen7 collision cannot discard a fresh action ──

99
desktop/src/features/channels/readState/readStateManager.ts
View File

@@ -69,7 +69,6 @@ export type MarkResult =
| "storage_failed";
};
/** Explicit delta returned by `ingest()`. */
export type IngestDelta = {
changedContexts: Set<string>;
canonicalChanged: boolean;
@@ -190,9 +189,10 @@ export class ReadStateManager {
retryAttempt = 0; // bounded exponential backoff attempt counter
pendingRetryOnComplete = false; // reconnect arrived during loadInFlight
drainScheduled = false;
/** Set by scheduleDrain() when a drain is in-progress; honoured after the pass ends. */
pendingFreshDrain = false;
/** Called after each drain intent is settled; wired by the hook for outcome routing. */
pendingFreshDrain = false; // set by scheduleDrain() when a drain is in-progress
/** Bounded-backoff drain-abort retry timer — distinct from scheduleDrain. */
drainRetryTimer: number | null = null;
drainRetryAttempt = 0;
onDrainOutcome: ((outcome: DrainOutcome) => void) | null = null;
constructor(pubkey: string, relayClient: RelayClient) {
@@ -206,7 +206,6 @@ export class ReadStateManager {
generateHex(16),
);
this.extraSlotIds = loadExtraSlotIds(pubkey);
// pendingOverrideIntentStore is restored from v2 blob in hydrateFromLocalStorage.
}
async initialize(): Promise<void> {
@@ -224,13 +223,11 @@ export class ReadStateManager {
this.drainScheduled = true;
void this.drainPendingIntents(gen);
} else if (!this.isLoadComplete && this.retryBackoffTimer === null) {
// Incomplete — schedule first automatic retry (attempt 1 = 1 s delay).
this.retryAttempt = 1;
const delayMs = 1_000;
this.retryBackoffTimer = window.setTimeout(() => {
this.retryBackoffTimer = null;
if (!this.destroyed) void this.retryLoad();
}, delayMs);
}, 1_000);
}
await this.startLiveSubscription();
if (this.destroyed) return;
@@ -312,6 +309,12 @@ export class ReadStateManager {
window.clearTimeout(this.retryBackoffTimer);
this.retryBackoffTimer = null;
}
if (this.drainRetryTimer !== null) {
window.clearTimeout(this.drainRetryTimer);
this.drainRetryTimer = null;
}
this.drainRetryAttempt = 0;
pendingOverrideIntentStore.clearTransactionState();
if (this.debounceTimer !== null) {
window.clearTimeout(this.debounceTimer);
this.debounceTimer = null;
@@ -341,27 +344,14 @@ export class ReadStateManager {
this.notifyListeners();
}
/**
* Retry load: generation-fenced, coalesces in-flight loads.
* Always starts immediately (no entry delay). After an incomplete verdict
* this method schedules itself via `retryBackoffTimer` with bounded
* exponential backoff (1 s, 2 s, 4 s … 60 s) so pending intents eventually
* drain even without an external reconnect signal.
*
* A reconnect during loadInFlight sets pendingRetryOnComplete so the next
* complete/incomplete verdict triggers exactly one fresh generation rather
* than dropping the reconnect signal.
*/
/** Retry load: generation-fenced, coalesces in-flight loads; bounded backoff on incomplete. */
async retryLoad(): Promise<void> {
if (this.destroyed) return;
// Cancel any pending backoff timer — this call supersedes it.
if (this.retryBackoffTimer !== null) {
window.clearTimeout(this.retryBackoffTimer);
this.retryBackoffTimer = null;
}
if (this.loadInFlight) {
// Reconnect arrived while a load is in flight — coalesce into a required
// fresh generation rather than dropping the signal.
this.pendingRetryOnComplete = true;
return;
}
@@ -378,25 +368,22 @@ export class ReadStateManager {
void this.drainPendingIntents(gen);
}
} else {
// Incomplete verdict — schedule automatic retry with bounded backoff
// so pending intents drain eventually even without a reconnect signal.
// Backoff: 1 s, 2 s, 4 s, … capped at 60 s.
// Incomplete verdict — schedule automatic retry with bounded backoff.
this.retryAttempt++;
const nextDelayMs = Math.min(
1_000 * 2 ** (this.retryAttempt - 1),
60_000,
this.retryBackoffTimer = window.setTimeout(
() => {
this.retryBackoffTimer = null;
if (!this.destroyed) void this.retryLoad();
},
Math.min(1_000 * 2 ** (this.retryAttempt - 1), 60_000),
);
this.retryBackoffTimer = window.setTimeout(() => {
this.retryBackoffTimer = null;
if (!this.destroyed) void this.retryLoad();
}, nextDelayMs);
}
} finally {
if (gen === this.loadGeneration) {
this.loadInFlight = false;
if (this.pendingRetryOnComplete && !this.destroyed) {
this.pendingRetryOnComplete = false;
// Cancel any scheduled backoff timer — reconnect triggers immediate retry.
// Cancel scheduled backoff — reconnect triggers immediate retry.
if (this.retryBackoffTimer !== null) {
window.clearTimeout(this.retryBackoffTimer);
this.retryBackoffTimer = null;
@@ -407,11 +394,10 @@ export class ReadStateManager {
}
}
/** Drain pending override intents. See readStateDrain.ts for the full implementation. */
async drainPendingIntents(drainGen: number): Promise<void> {
await drainPendingIntentsImpl(createDrainContext(this), drainGen);
this.drainRetryAttempt = 0; // successful pass resets the abort-retry counter
this.drainScheduled = false;
// Honor a scheduleDrain() request that arrived during the impl (deferred gen2 promotion).
if (this.pendingFreshDrain && !this.destroyed) {
this.pendingFreshDrain = false;
if (!this.drainScheduled && this.isLoadComplete) {
@@ -422,8 +408,8 @@ export class ReadStateManager {
}
/**
* Schedule a fresh drain pass if none is in-flight. Called by the drain after
* a deferred gen2 promotion so the new intent drains immediately.
* Schedule a fresh drain pass if none is in-flight. Called after deferred gen2
* promotion so the new intent drains immediately without waiting for a reconnect.
* While a drain is in-progress, sets `pendingFreshDrain` instead.
*/
scheduleDrain(): void {
@@ -436,6 +422,31 @@ export class ReadStateManager {
void this.drainPendingIntents(this.loadGeneration);
}
/**
* Schedule a bounded-backoff abort-retry drain pass (on storage failure or thrown callback).
* One pending timer, escalating backoff (1s→2s→4s…60s), coalesced per manager,
* cancelled on destroy() and identity change. Only abort paths call this; success-path
* gen2 promotion uses scheduleDrain() (immediate pendingFreshDrain) for no-delay draining.
*/
scheduleAbortRetry(): void {
if (this.destroyed || !this.isLoadComplete) return;
if (this.drainRetryTimer !== null) {
window.clearTimeout(this.drainRetryTimer);
this.drainRetryTimer = null;
}
this.drainRetryAttempt++;
this.drainRetryTimer = window.setTimeout(
() => {
this.drainRetryTimer = null;
if (!this.destroyed && this.isLoadComplete && !this.drainScheduled) {
this.drainScheduled = true;
void this.drainPendingIntents(this.loadGeneration);
}
},
Math.min(1_000 * 2 ** (this.drainRetryAttempt - 1), 60_000),
);
}
markChannelUnread(
channelId: string,
priorForcedEntry?: ForcedUnreadEntry,
@@ -460,14 +471,12 @@ export class ReadStateManager {
);
}
/** Shared ingest: accept pre-merged state and return semantic delta (changed contexts, canonical flip). */
private async ingest(merged: MergedReadState): Promise<IngestDelta> {
const delta: IngestDelta = {
changedContexts: new Set(),
canonicalChanged: false,
};
// Snapshot canonical forms before frontier merge.
const prevCanonicalByCtx = new Map<string, string>();
for (const rawCtx of merged.frontiers.keys()) {
const reg = this.overrideRegisters.get(rawCtx);
@@ -526,7 +535,6 @@ export class ReadStateManager {
return delta;
}
/** Process already-parsed events for metadata, conflict rotation, and maxFetchedCreatedAt. */
private processLoadedParsedEvents(events: ParsedReadStateEvent[]): void {
for (const parsed of events) {
this.maxFetchedCreatedAt = Math.max(
@@ -538,7 +546,6 @@ export class ReadStateManager {
if (parsed.createdAt > src)
this.contextSourceCreatedAt.set(rawCtx, parsed.createdAt);
}
// Slot-conflict rotation (NIP-RS.md:55-64,402-406).
if (
parsed.dTag === `read-state:${this.slotId}` &&
parsed.blob.client_id !== this.clientId
@@ -564,7 +571,6 @@ export class ReadStateManager {
}
}
/** Ingest parsed events: process metadata then merge into frontier/override maps. */
private async ingestParsedEvents(
events: ParsedReadStateEvent[],
): Promise<IngestDelta> {
@@ -765,7 +771,6 @@ export class ReadStateManager {
return true;
}
/** Effective frontier including parent resolver. */
channelFrontier(channelId: string): number {
return (
resolveEffectiveTimestamp({
@@ -776,7 +781,6 @@ export class ReadStateManager {
);
}
/** Build the single-slot contexts record; returns null if it doesn't fit. */
private currentContexts(): Record<string, number> | null {
const contexts: Record<string, number> = {};
for (const [ctx, ts] of this.effectiveState) {
@@ -857,7 +861,6 @@ export class ReadStateManager {
};
}
/** Coherent snapshot of completeness + frontiers + overrides. */
getProjection(): ReadStateProjection {
return {
loadComplete: this.isLoadComplete,
@@ -866,7 +869,6 @@ export class ReadStateManager {
};
}
/** Trial candidate plan; may mutate extraSlotIds — caller must restore on failure. */
tryCandidatePlan(rawCtxId: string, reg: OverrideRegister): boolean {
const prev = this.overrideRegisters.get(rawCtxId);
const wasPublishable = this.publishableContextIds.has(rawCtxId);
@@ -883,7 +885,6 @@ export class ReadStateManager {
return fits;
}
/** Restore extra slot IDs on rollback. Removes localStorage key when array empties. */
restoreExtraSlotIds(prev: string[]): void {
const changed =
prev.length !== this.extraSlotIds.length ||
@@ -918,14 +919,12 @@ export class ReadStateManager {
ctx,
Math.max(this.effectiveState.get(ctx) ?? 0, e.f),
);
this.publishableContextIds.add(ctx); // v2 entry is inherently publishable
this.publishableContextIds.add(ctx);
}
// Restore pending intents from v2 blob (consolidation: no separate key).
pendingOverrideIntentStore.restoreFromStorage(
stored.pendingIntents,
stored.nextGen,
);
// Amendment C hydration: load receipts; orphan sweep for receipts with no matching intent.
for (const [channelId, receipt] of stored.appliedReceipts) {
const intent = pendingOverrideIntentStore.get(channelId);
if (

35
desktop/src/features/channels/readState/readStateStorage.ts
View File

@@ -354,9 +354,14 @@ function readOverrideState(pubkey: string): OverrideStateBlob {
// which would silently discard the fresh action via the Amendment C
// already-applied check.
//
// If the increment would be unrepresentable (> Number.MAX_SAFE_INTEGER),
// fall back to a fresh safe value — any safe value above existing gens
// is acceptable; we prefer the smallest representable one.
// If all loaded generations are at or near MAX_SAFE_INTEGER (allocator
// exhausted), rebase: atomically compact all intent gens and their
// matching receipt gens into a fresh range [1..N] and persist the
// normalized blob. N is bounded by the number of channels in the map
// (at most one intent per channel), so rebasing is always deterministic.
//
// This guarantees no allocation can produce a gen that matches a
// retained receipt, since all receipts are renumbered in the same step.
{
let maxObservedGen = 0;
for (const intent of pendingIntents.values()) {
@@ -373,10 +378,26 @@ function readOverrideState(pubkey: string): OverrideStateBlob {
) {
nextGen = requiredNextGen;
} else if (!Number.isSafeInteger(requiredNextGen)) {
// maxObservedGen is already at or near MAX_SAFE_INTEGER — reset to
// a fresh safe value. Existing gens become unreachable for amendment-C
// matching (intentionally: the corrupt blob is untrustworthy).
nextGen = 1;
// Allocator exhausted — rebase all intent gens and matching receipt
// gens into a compact safe range [1..N]. intents is keyed by
// channelId with at most one entry per channel, so sorting is
// deterministic and produces a stable assignment.
const sorted = [...pendingIntents.entries()].sort(
([, a], [, b]) => a.gen - b.gen,
);
let newGen = 1;
for (const [chId, intent] of sorted) {
const oldGen = intent.gen;
const rebasedIntent: PendingIntent = { ...intent, gen: newGen };
pendingIntents.set(chId, rebasedIntent);
// Re-key the matching receipt (same channelId + old gen).
const receipt = receipts.get(chId);
if (receipt !== undefined && receipt.intentGen === oldGen) {
receipts.set(chId, { ...receipt, intentGen: newGen });
}
newGen++;
}
nextGen = newGen; // first free gen after the rebased range
}
}
}