fix(unread): centralize allocateGeneration, fix rebase boundary, add finally to drainPendingIntents

Round-8 fixes for #3966:

1. allocateGeneration() primitive (pendingOverrideIntents.ts)
   Preflights BOTH the emitted gen AND _nextGen+1 before any mutation.
   If either is unsafe, throws without touching any queue/transaction map.
   Replaces both post-increment isSafeInteger guards in enqueue() and
   promoteDeferred(). promoteDeferred() now allocates FIRST, then mutates
   deferredEnqueues/promotedDeferredPayloads — so an allocation failure
   leaves the deferred payload untouched and abortTransaction can safely
   re-buffer it.

2. Hydration rebase boundary (readStateStorage.ts)
   Condition order swapped: rebase checked before clamp. New trigger:
   !isSafeInteger(requiredNextGen) || requiredNextGen >= MAX_SAFE_INTEGER
   This catches maxObservedGen = MAX_SAFE_INTEGER-1 (requiredNextGen =
   MAX_SAFE_INTEGER), which previously passed the isSafeInteger check and
   took the clamp path, leaving nextGen=MAX_SAFE_INTEGER and one tick from
   poisoning _nextGen. Added isSafeInteger guard to ng serialization.

3. Promotion allocation failure → total-closure exit (readStateDrain.ts)
   promoteDeferred() call in step-3 is now wrapped in try-catch. If
   allocateGeneration throws, the catch calls abortTransaction +
   scheduleAbortRetry and continues, same as other non-success exits.

4. drainPendingIntents finally (readStateManager.ts)
   Wrapped in try-finally so drainScheduled=false always clears, even on
   an escaped exception from the impl. Condensed JSDoc for scheduleDrain
   to stay within the 1000-line ratchet.

5. Two new witnesses (Tests 60, 61):
   - Test 60: pi.gen=MAX-1, ng=MAX-1 → hydrate → rebase fires → two
     enqueues get safe distinct gens, neither throws, persisted ng safe,
     both intents survive fresh-manager hydration.
   - Test 61: _nextGen forced to MAX_SAFE_INTEGER via restoreFromStorage
     with deferred gen2 mid-transaction → drain → allocation throws in
     promoteDeferred → catch routes to abortTransaction+scheduleAbortRetry
     → drainScheduled false, drainRetryTimer set, gen1 live, deferred
     payload preserved → hydrateFromLocalStorage normalizes _nextGen to 2
     → retry drain completes gen1+gen2, frontier ≥ 999.

Optional MINOR (explicit typed pass outcome): skipped. The change would
require threading a return value through drainPendingIntentsImpl and its
caller, adding several lines to files already near the ratchet. The current
drainRetryTimer === null discrimination is sound for the synchronous impl.

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 18:40:08 -04:00
parent da0461a090
commit 0d7fe32a9e
5 changed files with 404 additions and 46 deletions

View File

@@ -196,14 +196,7 @@ export class PendingOverrideIntentStore {
...(priorForcedEntry !== undefined ? { priorForcedEntry } : {}),
};
}
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 gen = this.allocateGeneration();
const intent: PendingIntent = {
gen,
op,
@@ -215,6 +208,29 @@ export class PendingOverrideIntentStore {
return intent;
}
/**
* Allocate the next generation number.
*
* Preflights BOTH the generation that will be emitted (the current `_nextGen`)
* AND the resulting `_nextGen` after increment. Both must be safe integers
* before any mutation occurs — this ensures the allocator can never enter a
* state where it has emitted the last safe value and left `_nextGen` poisoned.
*
* Throws if either value is unsafe (allocator exhausted). The rebase path in
* `readStateStorage.ts` prevents this during normal operation.
*/
private allocateGeneration(): number {
const gen = this._nextGen;
const next = this._nextGen + 1;
if (!Number.isSafeInteger(gen) || !Number.isSafeInteger(next)) {
throw new Error(
"[PendingOverrideIntentStore] generation counter exhausted (unsafe)",
);
}
this._nextGen = next;
return gen;
}
/**
* Open a drain transaction for `channelId`.
* While open, `enqueue()` for this channel buffers rather than applying.
@@ -291,16 +307,14 @@ export class PendingOverrideIntentStore {
promoteDeferred(channelId: string): boolean {
const deferred = this.deferredEnqueues.get(channelId);
if (deferred === undefined) return false;
// Preflight allocation BEFORE mutating any map. If the allocator is
// exhausted, the throw propagates without touching deferredEnqueues or
// promotedDeferredPayloads, so abortTransaction can safely re-buffer.
const gen = this.allocateGeneration();
// Allocation succeeded — now mutate the maps.
this.deferredEnqueues.delete(channelId);
// 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,

View File

@@ -349,8 +349,20 @@ export async function drainPendingIntents(
// • On restart: blob still has gen1 intent + receipt; receipt prevents a
// re-bump and cleanup runs next drain. Gen2 is preserved for retry.
const capturedReceipt = ctx.appliedReceipts.get(channelId);
const gen2Promoted =
pendingOverrideIntentStore.promoteDeferred(channelId);
let gen2Promoted: boolean;
try {
gen2Promoted = pendingOverrideIntentStore.promoteDeferred(channelId);
} catch (err) {
// Allocation exhausted — treat the same as a storage-failure abort.
// promoteDeferred() preflights before mutating, so no map was touched.
console.warn(
`[ReadStateManager] drain: allocateGeneration failed in promoteDeferred for ${channelId}:`,
err,
);
pendingOverrideIntentStore.abortTransaction(channelId);
if (!ctx.destroyed) ctx.scheduleAbortRetry();
continue; // transaction closed via abort; finally is a no-op
}
ctx.appliedReceipts.delete(channelId);
pendingOverrideIntentStore.compareAndDelete(channelId, capturedGen);
if (!ctx.persistLocalState()) {

View File

@@ -6426,3 +6426,331 @@ test("identitySwap_aDeferredMidTransaction_bHydrates_bSeesNoAState", async () =>
mgrA.destroy();
mgrB.destroy();
});
// ── Test 60: MAX-1 boundary — two enqueues from valid one-below-ceiling hydration ──
test("hydrate_maxMinusOneState_twoEnqueues_safeGensNeitherThrows_ngPersistSafe", async () => {
// Thufir's exact repro: blob with pi.gen = MAX_SAFE_INTEGER-1 and ng = MAX_SAFE_INTEGER-1.
// Before the fix, hydration accepted this verbatim (requiredNextGen = MAX_SAFE_INTEGER,
// which isSafeInteger), set nextGen=MAX_SAFE_INTEGER via clamp, and the first enqueue
// minted gen=MAX_SAFE_INTEGER while leaving _nextGen=MAX_SAFE_INTEGER+1 (unsafe).
// The second enqueue then threw synchronously.
//
// After the fix: requiredNextGen >= MAX_SAFE_INTEGER triggers rebase instead of clamp.
// The single pending intent is rebased to gen=1, nextGen=2. Both enqueues get safe
// distinct gens; neither throws; the persisted ng after each enqueue is safe.
//
// Failure branch: rebase boundary (maxObservedGen = MAX_SAFE_INTEGER-1, requiredNextGen
// = MAX_SAFE_INTEGER triggers rebase rather than clamp).
// Durable surface: v2 blob after each enqueue has safe ng; fresh-manager hydration
// succeeds and sees both intents.
globalThis.window.localStorage = makeLocalStorage();
const pubkey = "60".repeat(32);
const channelId = `max-minus-one-ch-${"\u006d".repeat(47)}`;
const v2Key = `buzz.nip-rs.override-state.v2:${pubkey}`;
const ls = globalThis.window.localStorage;
const MAX = Number.MAX_SAFE_INTEGER;
const fakeRelay = {
fetchEvents: async () => [],
publishEvent: async () => {},
subscribeFenced: async (_f, _h) => makeFenceHandle({ eose: true }),
subscribeLive: async (_f, _h) => () => {},
subscribeToReconnects: () => () => {},
getConnectionGeneration: () => 0,
};
// Plant blob: one intent at gen=MAX_SAFE_INTEGER-1, ng=MAX_SAFE_INTEGER-1.
ls.setItem(
v2Key,
JSON.stringify({
pi: { [channelId]: { gen: MAX - 1, op: "unread" } },
ng: MAX - 1,
}),
);
const mgr = new ReadStateManager(pubkey, fakeRelay);
mgr.hydrateFromLocalStorage();
// After hydration: rebase must have fired.
// maxObservedGen = MAX-1 → requiredNextGen = MAX → >= MAX_SAFE_INTEGER → rebase.
// One intent rebased to gen=1; nextGen=2.
const ngAfterHydrate = pendingOverrideIntentStore.nextGen;
assert.ok(
Number.isSafeInteger(ngAfterHydrate),
`nextGen must be safe after hydration; got ${ngAfterHydrate}`,
);
assert.ok(
ngAfterHydrate < MAX,
`nextGen must be well below MAX_SAFE_INTEGER after rebase; got ${ngAfterHydrate}`,
);
assert.equal(ngAfterHydrate, 2, "nextGen must be 2 after rebase of 1 intent");
const ch2 = `max-minus-one-ch2-${"n".repeat(46)}`;
// First enqueue — must NOT throw, must produce safe gen.
mgr.isLoadComplete = false;
let r1;
assert.doesNotThrow(() => {
r1 = mgr.markChannelUnread(ch2);
}, "first enqueue must not throw");
assert.equal(r1.status, "queued", "first enqueue must be queued");
const intent1 = pendingOverrideIntentStore.get(ch2);
assert.ok(intent1, "first intent must be in store");
assert.ok(
Number.isSafeInteger(intent1.gen),
`first intent gen must be safe; got ${intent1.gen}`,
);
assert.ok(intent1.gen > 0, "first intent gen must be positive");
// Persisted ng after first enqueue must be safe.
const blob1Raw = ls.getItem(v2Key);
assert.ok(blob1Raw, "v2 blob must be present after first enqueue");
const blob1 = JSON.parse(blob1Raw);
const ng1 = blob1.ng ?? 1; // ng omitted when === 1 (the `ng > 1` condition)
assert.ok(
Number.isSafeInteger(ng1),
`persisted ng after first enqueue must be safe; got ${ng1}`,
);
assert.ok(
ng1 < MAX,
`persisted ng must be below MAX_SAFE_INTEGER; got ${ng1}`,
);
const ch3 = `max-minus-one-ch3-${"o".repeat(46)}`;
// Second enqueue — must NOT throw, must produce distinct safe gen.
let r2;
assert.doesNotThrow(() => {
r2 = mgr.markChannelRead(ch3, undefined, 500);
}, "second enqueue must not throw");
assert.equal(r2.status, "queued", "second enqueue must be queued");
const intent2 = pendingOverrideIntentStore.get(ch3);
assert.ok(intent2, "second intent must be in store");
assert.ok(
Number.isSafeInteger(intent2.gen),
`second intent gen must be safe; got ${intent2.gen}`,
);
assert.notEqual(
intent1.gen,
intent2.gen,
"both enqueues must have distinct gens",
);
// Persisted ng after second enqueue must be safe.
const blob2Raw = ls.getItem(v2Key);
assert.ok(blob2Raw, "v2 blob must be present after second enqueue");
const blob2 = JSON.parse(blob2Raw);
const ng2 = blob2.ng ?? 1;
assert.ok(
Number.isSafeInteger(ng2),
`persisted ng after second enqueue must be safe; got ${ng2}`,
);
assert.ok(
ng2 < MAX,
`persisted ng must be below MAX_SAFE_INTEGER; got ${ng2}`,
);
// Both intents survive hydration on a fresh manager.
const mgr2 = new ReadStateManager(pubkey, fakeRelay);
mgr2.hydrateFromLocalStorage();
assert.ok(
pendingOverrideIntentStore.get(ch2),
"ch2 intent must survive hydration",
);
assert.ok(
pendingOverrideIntentStore.get(ch3),
"ch3 intent must survive hydration",
);
mgr2.destroy();
mgr.destroy();
});
// ── Test 61: promotion allocation failure — no stranded lock, retry scheduled ──
test("drain_promoteDeferredAllocExhausted_noStrandedLock_retryScheduled_drainScheduledCleared", async () => {
// Forces allocateGeneration() to throw during promoteDeferred() inside the drain.
// Achieved by: restoreFromStorage with _nextGen=MAX_SAFE_INTEGER (artificially exhausted),
// plus a gen1 intent and a deferred gen2 enqueue buffered mid-transaction.
// The promoteDeferred() call preflights BEFORE mutating — when it throws:
// • deferredEnqueues still has the gen2 payload (not deleted before throw)
// • abortTransaction() is called: gen1 restored as live intent, channel unlocked
// • scheduleAbortRetry() fires: drainRetryTimer set
// • continue → finally clears drainScheduled via the manager's try-finally
//
// After the drain:
// • drainScheduled = false (future scheduleDrain() calls will not be lost)
// • drainRetryTimer != null (bounded retry scheduled)
// • Channel is unlocked (gen1 live in store, gen > 0)
// • Deferred payload not lost (still in deferredEnqueues; confirmed by
// hydrateFromLocalStorage re-normalizing _nextGen then re-draining successfully)
//
// Failure branch: promoteDeferred() allocation exhaustion → catch in step-3.
// Durable surface: after fresh hydration (normalizes _nextGen) + retry drain,
// gen1+gen2 committed; frontier ≥ 999; no pending intent.
globalThis.window.localStorage = makeLocalStorage();
const pubkey = "61".repeat(32);
const channelId = `promo-alloc-ch-${"p".repeat(50)}`;
const v2Key = `buzz.nip-rs.override-state.v2:${pubkey}`;
const ls = globalThis.window.localStorage;
const MAX = Number.MAX_SAFE_INTEGER;
const origSetTimeout = globalThis.window.setTimeout;
const origClearTimeout = globalThis.window.clearTimeout;
const scheduledTimers = new Map(); // id → fn
let nextFakeId = 61000;
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);
}
};
const fakeRelay = {
fetchEvents: async () => [],
publishEvent: async () => {},
subscribeFenced: async (_f, _h) => makeFenceHandle({ eose: true }),
subscribeLive: async (_f, _h) => () => {},
subscribeToReconnects: () => () => {},
getConnectionGeneration: () => 0,
};
try {
const mgr = new ReadStateManager(pubkey, fakeRelay);
mgr.effectiveState.set(channelId, 50);
// Plant a gen1 intent and set ng = MAX_SAFE_INTEGER directly via restoreFromStorage.
// This bypasses hydration normalization to simulate an exhausted allocator.
// (In production, the rebase boundary prevents _nextGen from ever reaching MAX in-session.)
const gen1Intent = { gen: 1, op: /** @type {"unread"} */ ("unread") };
pendingOverrideIntentStore.restoreFromStorage(
new Map([[channelId, gen1Intent]]),
MAX,
);
// Write a matching v2 blob so hydrateFromLocalStorage (used for restart witness) can
// normalize _nextGen via the readStateStorage normalizer.
ls.setItem(
v2Key,
JSON.stringify({
pi: { [channelId]: { gen: 1, op: "unread" } },
ng: MAX,
}),
);
// Wire the drain callback: enqueue gen2 (read, readTarget=999) from inside the
// outcome callback. The channel is locked at callback time, so the enqueue is
// buffered as deferred. The promoteDeferred() call will then exhaust the allocator.
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;
}
};
// Drain with _nextGen = MAX_SAFE_INTEGER: step1 succeeds (register + receipt written),
// step2 callback fires (deferred enqueue buffered), step3 promoteDeferred() throws.
mgr.isLoadComplete = true;
await mgr.drainPendingIntents(mgr.loadGeneration);
// ── drainScheduled must be false (manager's try-finally always clears it) ──
assert.equal(
mgr.drainScheduled,
false,
"drainScheduled must be false after allocation throw (finally path)",
);
// ── drainRetryTimer must be set (scheduleAbortRetry called from catch) ──
assert.ok(
mgr.drainRetryTimer !== null,
"drainRetryTimer must be set after allocation failure",
);
assert.ok(
scheduledTimers.has(mgr.drainRetryTimer),
"abort retry timer ID must be in scheduledTimers",
);
// ── Channel is unlocked: gen1 is the live intent (abortTransaction ran) ──
const intentAfterFail = pendingOverrideIntentStore.get(channelId);
assert.ok(
intentAfterFail,
"gen1 must be live in store after allocation failure",
);
assert.equal(
intentAfterFail.gen,
gen1Intent.gen,
"live intent gen must be gen1 after abort",
);
assert.ok(
intentAfterFail.gen > 0,
"live intent must not be the -1 placeholder",
);
// ── Deferred payload not lost ──
// The deferred gen2 payload was NOT deleted by promoteDeferred() (it threw before
// touching deferredEnqueues). Verify indirectly: after re-normalizing _nextGen via
// hydrateFromLocalStorage, fire the retry timer, and confirm gen2 promotes + drains.
//
// Hydrate: readStateStorage normalizer sees ng=MAX and gen1 intent → clamp or rebase.
// maxObservedGen = max(gen1=1, receipt.intentGen from step1 write).
// After step1 in the failed drain, v2 blob was updated with register+receipt+pi+ng=MAX.
// On hydration: maxObservedGen = 1, requiredNextGen = 2, 2 < MAX → clamp → _nextGen=2.
mgr.hydrateFromLocalStorage();
// Restore callback on the original mgr (hydration replaces the singleton).
mgr.onDrainOutcome = (outcome) => {
if (
outcome.kind === "applied-unread" &&
outcome.channelId === channelId
) {
const prev = mgr.isLoadComplete;
mgr.isLoadComplete = false;
mgr.markChannelRead(channelId, undefined, 999);
mgr.isLoadComplete = prev;
}
};
// Fire the retry timer — _nextGen is now 2 (normalized), allocation will succeed.
const retryTimerId = mgr.drainRetryTimer;
const retryFn = scheduledTimers.get(retryTimerId);
assert.ok(retryFn, "retry timer fn must be in scheduledTimers");
scheduledTimers.clear();
await retryFn();
await new Promise((r) => origSetTimeout(r, 0));
// ── After retry + auto-drain: gen1+gen2 committed, frontier ≥ 999 ──
const frontierAfterRetry = mgr.effectiveState.get(channelId) ?? 0;
assert.ok(
frontierAfterRetry >= 999,
`frontier must be ≥ 999 after gen1+gen2 drain; got ${frontierAfterRetry}`,
);
assert.equal(
pendingOverrideIntentStore.get(channelId),
undefined,
"no pending intent after successful retry drain",
);
// drainScheduled must still be false (no new drain was needed).
assert.equal(
mgr.drainScheduled,
false,
"drainScheduled must be false after successful retry",
);
mgr.destroy();
} finally {
globalThis.window.setTimeout = origSetTimeout;
globalThis.window.clearTimeout = origClearTimeout;
}
});

View File

@@ -395,28 +395,27 @@ export class ReadStateManager {
}
async drainPendingIntents(drainGen: number): Promise<void> {
await drainPendingIntentsImpl(createDrainContext(this), drainGen);
// Reset the attempt counter only on a successful pass (no abort-retry timer
// was scheduled during the impl). Abort paths call scheduleAbortRetry() which
// sets drainRetryTimer; if it's still null here the pass committed cleanly.
if (this.drainRetryTimer === null) {
this.drainRetryAttempt = 0;
}
this.drainScheduled = false;
if (this.pendingFreshDrain && !this.destroyed) {
this.pendingFreshDrain = false;
if (!this.drainScheduled && this.isLoadComplete) {
this.drainScheduled = true;
void this.drainPendingIntents(this.loadGeneration);
try {
await drainPendingIntentsImpl(createDrainContext(this), drainGen);
// Reset only on a successful pass; abort paths set drainRetryTimer first.
if (this.drainRetryTimer === null) {
this.drainRetryAttempt = 0;
}
this.drainScheduled = false;
if (this.pendingFreshDrain && !this.destroyed) {
this.pendingFreshDrain = false;
if (!this.drainScheduled && this.isLoadComplete) {
this.drainScheduled = true;
void this.drainPendingIntents(this.loadGeneration);
}
}
} finally {
// Always clear drainScheduled even if the impl throws.
this.drainScheduled = false;
}
}
/**
* 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.
*/
/** Schedule a fresh drain pass; sets `pendingFreshDrain` if already in-flight. */
scheduleDrain(): void {
if (this.destroyed || !this.isLoadComplete) return;
if (this.drainScheduled) {

View File

@@ -373,21 +373,18 @@ function readOverrideState(pubkey: string): OverrideStateBlob {
}
const requiredNextGen = maxObservedGen + 1;
if (
Number.isSafeInteger(requiredNextGen) &&
nextGen !== requiredNextGen
!Number.isSafeInteger(requiredNextGen) ||
requiredNextGen >= Number.MAX_SAFE_INTEGER
) {
// Clamp nextGen to exactly the required value — both upward (corrupt
// ng too low, e.g. ng=7 with existing gen-7 receipt) and downward
// (corrupt ng too high, e.g. ng=MAX_SAFE_INTEGER with low gens, which
// would cause the first enqueue to mint gen=MAX_SAFE_INTEGER and the
// second to throw from isSafeInteger guard). No retained identity
// sits at or above requiredNextGen, so no collision is possible.
nextGen = requiredNextGen;
} else if (!Number.isSafeInteger(requiredNextGen)) {
// 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.
//
// The ceiling check (`requiredNextGen >= MAX_SAFE_INTEGER`) catches the
// valid one-below-ceiling state (`maxObservedGen = MAX_SAFE_INTEGER - 1`)
// before `allocateGeneration()` can mint `MAX_SAFE_INTEGER` and leave
// `_nextGen` at the unsafe value.
const sorted = [...pendingIntents.entries()].sort(
([, a], [, b]) => a.gen - b.gen,
);
@@ -404,6 +401,14 @@ function readOverrideState(pubkey: string): OverrideStateBlob {
newGen++;
}
nextGen = newGen; // first free gen after the rebased range
} else if (nextGen !== requiredNextGen) {
// Clamp nextGen to exactly the required value — both upward (corrupt
// ng too low, e.g. ng=7 with existing gen-7 receipt) and downward
// (corrupt ng too high, e.g. ng=MAX_SAFE_INTEGER with low gens, which
// would cause the first enqueue to mint gen=MAX_SAFE_INTEGER and the
// second to throw from the allocator guard). No retained identity
// sits at or above requiredNextGen, so no collision is possible.
nextGen = requiredNextGen;
}
}
}
@@ -557,7 +562,7 @@ export function writeStoredReadState(
r: registersObj,
...(Object.keys(receiptsObj).length > 0 ? { receipts: receiptsObj } : {}),
...(Object.keys(piObj).length > 0 ? { pi: piObj } : {}),
...(nextGen > 1 ? { ng: nextGen } : {}),
...(nextGen > 1 && Number.isSafeInteger(nextGen) ? { ng: nextGen } : {}),
}),
);