Skip to content
13 changes: 12 additions & 1 deletion src/persist.ts
Original file line number Diff line number Diff line change
Expand Up @@ -2855,7 +2855,14 @@ export class SessionStore {
this.drainFlushWaiters(id, this.lastSaveLanded.get(id) === true);
});
}, this.debounceMs);
timer.unref?.();
// #4788: the debounce timer is host-owned periodic upkeep and must not
// pin a process open on its own -- so it is unref'd. But `afterNextFlush`
// makes it a HOST: a caller that awaits a verdict has no other handle
// guaranteed to fire, and an unref'd timer is not a reason for Node to
// stay alive. Awaiting then exits the loop with the promise unresolved
// (`Promise resolution is still pending but the event loop has already
// resolved`). Keep the timer referenced exactly while someone waits.
if (!this.flushWaiters.has(id)) timer.unref?.();
this.saveTimers.set(id, timer);
}

Expand All @@ -2876,6 +2883,10 @@ export class SessionStore {
const ws = this.flushWaiters.get(sessionId) ?? [];
ws.push((ok: boolean): void => { resolve(ok); });
this.flushWaiters.set(sessionId, ws);
// #4788: the timer was armed unref'd (no waiter existed then). The
// waiter now holds the promise open, so the timer must hold the loop
// open too -- otherwise the process drains and the verdict is lost.
this.saveTimers.get(sessionId)?.ref?.();
});
}

Expand Down
4 changes: 2 additions & 2 deletions src/storage/shadow.ts
Original file line number Diff line number Diff line change
Expand Up @@ -329,12 +329,12 @@ export class UnifiedShadow {
: "; delete the file to reset the index";
loggerLog(
"warn",
`[storage] unified sidecar failed to open ${dir}/index.db (${errStr}) — staying legacy-only${tail}`,
`[storage] unified sidecar failed to open ${path.join(dir, "index.db")} (${errStr}) — staying legacy-only${tail}`,
);
recordSidecarOpenFailure(errStr);
return null;
}
const shadow = new UnifiedShadow(mode, store, legacyLoad ?? (() => null), undefined, `${dir}/index.db`);
const shadow = new UnifiedShadow(mode, store, legacyLoad ?? (() => null), undefined, path.join(dir, "index.db"));
recordSidecarInstance(shadow);
const engineTag = ` [engine: ${store.engine}/${store.engineSource}]`;
const walTag = ` [wal-cap: ${walLimit}]`;
Expand Down
8 changes: 6 additions & 2 deletions tests/inc-canonical-pointers.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -105,7 +105,11 @@ test("cold canonical rebuild: identical resync keeps the shared prefix; delta as
// #5155: the wrap-up defers the durable flush (debounced) — let the
// rows settle before dropping the resident caches, or the cold
// rebuild reads a one-turn-stale table and honestly 409s.
await new Promise((resolve) => setTimeout(resolve, 20));
// Await the settle itself (afterNextFlush resolves once the armed save
// lands) instead of a fixed sleep: the settle is a real condition, and
// a magic millisecond budget is a deadline a loaded windows-latest
// runner can miss — the 409 flake this replaces.
assert.equal(await getStore().afterNextFlush("conv-ptrs"), true, "the turn's durable save must land before the cold reset");
__resetIncrementalLaneForTests();

const delta = await h.turn({ sinceSeq: 3, messages: [core("u4", "four")] });
Expand Down Expand Up @@ -135,7 +139,7 @@ test("cold canonical rebuild: edited-tail resync keeps the shared prefix, drops
await h.turn({ sinceSeq: 0, resync: true, messages: [core("u1", "one"), core("u2", "two-EDITED")] });
await __incrementalPtrChainForTests("conv-ptrs");
// #5155: settle the debounced flush before the cold reset (see above).
await new Promise((resolve) => setTimeout(resolve, 20));
assert.equal(await getStore().afterNextFlush("conv-ptrs"), true, "the resync's durable save must land before the cold reset");
__resetIncrementalLaneForTests();
const delta = await h.turn({ sinceSeq: 2, messages: [core("u3", "three")] });
assert.equal(delta.status, 200);
Expand Down
4 changes: 2 additions & 2 deletions tests/issue4739-bounded-session-load.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -139,7 +139,7 @@ test("full mode: viewFor decodes only pointer rows and matches the full-log deco
assert.deepEqual(via, reference, "bounded decode byte-equal to the full-log decode");
assert.equal(delta, expectedPointers, `viewFor read ${delta} message rows for ${expectedPointers} pointer seqs (full log = ${liveSeqs.size})`);
} finally {
store.cancelAll();
store.closeQuietly();
if (prev === undefined) delete process.env.BILI_STORAGE_UNIFIED;
else process.env.BILI_STORAGE_UNIFIED = prev;
rmrf(dir);
Expand Down Expand Up @@ -178,7 +178,7 @@ test("full mode: dropped metadata restores from the index alone (narrow read, no
assert.ok(again, "envelope rewritten by later save");
assert.deepEqual(Object.keys(JSON.parse(fs.readFileSync(again, "utf8")).payload.metadata as object).sort(), persistedKeys);
} finally {
store.cancelAll();
store.closeQuietly();
if (prev === undefined) delete process.env.BILI_STORAGE_UNIFIED;
else process.env.BILI_STORAGE_UNIFIED = prev;
rmrf(dir);
Expand Down
17 changes: 13 additions & 4 deletions tests/issue4788-resident-minimal.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -86,6 +86,10 @@ async function turn(store: SessionStore, s: Session, midFlight?: boolean): Promi
if (midFlight) s.inFlight = 1;
s.revisionEpoch = (s.revisionEpoch ?? 0) + 1;
store.scheduleSave(s);
// #4788: while a caller awaits afterNextFlush, persist.ts keeps the pending
// debounce timer ref'd, so this settle is deterministic even in a node:test
// child whose only surviving handles are unref'd sockets — no test-side
// keep-alive needed here anymore.
const landed = await store.afterNextFlush(s.id);
if (midFlight) s.inFlight = 0;
return landed;
Expand Down Expand Up @@ -129,7 +133,12 @@ test("(a) between-turns drop: every landed request-end boundary discharges dropp
assert.ok((s.blockContents as LazyBlockContents).isDropped, "still folded after the second boundary");
assert.ok(isDroppedMetadata(s.metadata));
} finally {
store.cancelAll();
// #4739/#5140 family: the suite arms a real SessionStorage under
// BILI_STORAGE_UNIFIED=full, so the sidecar handle is still open here.
// cancelAll() clears timers and the build map but never closes the
// shadow; Windows then refuses to unlink the directory (EBUSY/EPERM).
// Close the short-lived store first: cancel, then close.
store.closeQuietly();
rmrf(dir);
}
});
Expand Down Expand Up @@ -164,7 +173,7 @@ test("(b) narrow-read reconstruction deepEquals the pre-drop reference", async (
const rec = (parsed.payload ?? parsed) as Record<string, unknown>;
assert.equal(Object.keys((rec.blockContents ?? {}) as object).length, Object.keys(refBlocks).length, "full set round-trips through the envelope");
} finally {
store.cancelAll();
store.closeQuietly();
rmrf(dir);
}
});
Expand Down Expand Up @@ -203,7 +212,7 @@ test("(c) hard cap under sustained activity: boundaries drop, in-flight skips, w
process.env.BILI_SESSION_RESIDENT_HARD_MAX_BYTES = saved;
}
} finally {
store.cancelAll();
store.closeQuietly();
rmrf(dir);
}
});
Expand Down Expand Up @@ -246,7 +255,7 @@ test("(d) envelope-frozen routing is unchanged AND gets equivalent drops", async
assert.equal(s.blockContents.size, 21, "all entries resident again after reconstruction");
} finally {
process.env.BILI_STORAGE_UNIFIED = "legacy";
store.cancelAll();
store.closeQuietly();
rmrf(dir);
}
});
10 changes: 9 additions & 1 deletion tests/plugin-fork.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -1477,6 +1477,10 @@ test("HTTP a mid-history rewrite replaces the snapshot wholesale with an exact b
});

test("HTTP a cold-restored non-retained parent serves its intact raw tail immediately (#5140)", { timeout: 60_000 }, async () => {
// Pin the mode this raw-tail proof requires: under the runner's legacy
// baseline no unified rows exist and forkSnapshot fails closed (#4599).
const oldMode = process.env.BILI_STORAGE_UNIFIED;
process.env.BILI_STORAGE_UNIFIED = "full";
const h = await harness(true);
try {
const parent = resolveConversation("parent").session!;
Expand All @@ -1495,7 +1499,11 @@ test("HTTP a cold-restored non-retained parent serves its intact raw tail immedi
await sendModel(h, "parent", h.messages);
assert.equal((await h.request("/__bili/plugin/snapshot?conversationId=parent")).status, 200);
assert.equal(resolveConversation("parent").session!.pluginSnapshot!.length, 3);
} finally { await h.close(); }
} finally {
await h.close();
if (oldMode === undefined) delete process.env.BILI_STORAGE_UNIFIED;
else process.env.BILI_STORAGE_UNIFIED = oldMode;
}
});

test("fork protocol v2: snapshot serves the light revision, a stale v1 whole-store revision is still accepted (#2671 Phase 2b)", { timeout: 60_000 }, async () => {
Expand Down
Loading