Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions kernel/package-lock.json

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

2 changes: 1 addition & 1 deletion package-lock.json

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

88 changes: 55 additions & 33 deletions src/persist.ts
Original file line number Diff line number Diff line change
Expand Up @@ -373,6 +373,9 @@ export class SessionStore {
/** #3199: post-boot background alias scan (unified modes only) — null in
* legacy mode and until bootBounded defers work. */
private remainderScan: Promise<void> | null = null;
/** #4515: how many state files the last bounded boot DECODED synchronously
* in the warm window before deferring the rest (0 when nothing is warmed). */
private lastBootSyncParsedFiles = 0;
/** #2674: monotonic per-process counter for unique content-store temp
* names (pid + seq makes collisions across processes/rapid saves
* unreachable). */
Expand Down Expand Up @@ -746,8 +749,12 @@ export class SessionStore {
this.diskFiles = [];
this.diskFilesSeeded = true;
this.legacyAliases.clear();
this.lastBootSyncParsedFiles = 0;
const warm: Array<{ file: string; id: string; meta: PersistedSession["meta"]; savedAt: number; bytes: number }> = [];
const canonical = new Set<string>();
// #4515: per-phase startup timing so a slow cold start names which phase ate the spawn budget.
const timings = { walk: 0, parseWindow: 0, indexUnion: 0 };
const tWalk = performance.now();
const filesByRoot = new Map<string, string[]>();
for (const root of [this.sessionsDir, `${this.sessionsDir}-cold`]) {
const files = existsSync(root) ? await walkJsonFiles(root) : [];
Expand All @@ -757,6 +764,7 @@ export class SessionStore {
filesByRoot.set(root, [...(filesByRoot.get(root) ?? []), file]);
}
}
timings.walk = Math.round(performance.now() - tWalk);
// #3199: unified mode (shadow or full) — the listener must not wait
// for the whole corpus to parse. Two stages: (1) parse a bounded
// WINDOW of the newest files (mtime desc, stopped once the window
Expand All @@ -770,40 +778,47 @@ export class SessionStore {
// authoritative" predicate.
if (this.replicasAvailable()) {
this.seedIndexedAliases(canonical);
const hot = (filesByRoot.get(this.sessionsDir) ?? [])
.map((file) => {
try {
const st = statSync(file);
return { file, mtimeMs: st.mtimeMs, size: st.size };
} catch {
return null; // raced deletion between walk and stat
}
})
.filter((c): c is { file: string; mtimeMs: number; size: number } => c !== null)
.sort((a, b) => b.mtimeMs - a.mtimeMs);
const parsedFiles = new Set<string>();
const parsedCands: Array<{ file: string; id: string; meta: PersistedSession["meta"]; savedAt: number; bytes: number }> = [];
let windowBytes = 0;
for (const cand of hot) {
if (parsedCands.length >= limits.maxSessions && windowBytes >= limits.maxBytes) break;
// on-disk size lower-bounds decoded size (#2803) — cheap pre-filter
if (cand.size > limits.maxBytes) continue;
const parsed = await this.readRawFileWithBytes(cand.file);
parsedFiles.add(cand.file);
if (!parsed) continue;
const rec = this.parseBootPayload(parsed.value);
if (!rec || !fileNameMatchesId(path.basename(cand.file), rec.payload.id, rec.meta.protocol, rec.meta.upstreamOrigin)) continue;
const savedAt = rec.payload.savedAt ?? 0;
if (rec.payload.id === rec.meta.label || rec.payload.id.startsWith("pfa-")) this.noteCanonical(rec.payload.id, canonical);
else if (rec.meta.label) this.upsertLegacyAlias(rec.meta.label, { file: cand.file, id: rec.payload.id, savedAt }, canonical);
const bytes = parsed.bytes;
if (bytes > limits.maxBytes) continue;
windowBytes += bytes;
parsedCands.push({ file: cand.file, id: rec.payload.id, meta: rec.meta, savedAt, bytes });
if (limits.maxSessions > 0) {
const tParse = performance.now();
const hot = (filesByRoot.get(this.sessionsDir) ?? [])
.map((file) => {
try {
const st = statSync(file);
return { file, mtimeMs: st.mtimeMs, size: st.size };
} catch {
return null; // raced deletion between walk and stat
}
})
.filter((c): c is { file: string; mtimeMs: number; size: number } => c !== null)
.sort((a, b) => b.mtimeMs - a.mtimeMs);
const parsedCands: Array<{ file: string; id: string; meta: PersistedSession["meta"]; savedAt: number; bytes: number }> = [];
let windowBytes = 0;
for (const cand of hot) {
if (parsedCands.length >= limits.maxSessions && windowBytes >= limits.maxBytes) break;
// on-disk size lower-bounds decoded size (#2803) — cheap pre-filter
if (cand.size > limits.maxBytes) continue;
const parsed = await this.readRawFileWithBytes(cand.file);
parsedFiles.add(cand.file);
if (!parsed) continue;
const rec = this.parseBootPayload(parsed.value);
if (!rec || !fileNameMatchesId(path.basename(cand.file), rec.payload.id, rec.meta.protocol, rec.meta.upstreamOrigin)) continue;
const savedAt = rec.payload.savedAt ?? 0;
if (rec.payload.id === rec.meta.label || rec.payload.id.startsWith("pfa-")) this.noteCanonical(rec.payload.id, canonical);
else if (rec.meta.label) this.upsertLegacyAlias(rec.meta.label, { file: cand.file, id: rec.payload.id, savedAt }, canonical);
const bytes = parsed.bytes;
if (bytes > limits.maxBytes) continue;
windowBytes += bytes;
parsedCands.push({ file: cand.file, id: rec.payload.id, meta: rec.meta, savedAt, bytes });
}
parsedCands.sort((a, b) => b.savedAt - a.savedAt);
while (parsedCands.length > limits.maxSessions || parsedCands.reduce((sum, item) => sum + item.bytes, 0) > limits.maxBytes) parsedCands.pop();
warm.push(...parsedCands);
timings.parseWindow = Math.round(performance.now() - tParse);
} else {
// #4515: maxSessions==0 (the #2803 default) warms NOTHING, so the sync decode window is pure waste on the pre-listen path — the background scan below does all corpus alias/canonical bookkeeping.
}
parsedCands.sort((a, b) => b.savedAt - a.savedAt);
while (parsedCands.length > limits.maxSessions || parsedCands.reduce((sum, item) => sum + item.bytes, 0) > limits.maxBytes) parsedCands.pop();
warm.push(...parsedCands);
this.lastBootSyncParsedFiles = parsedFiles.size;
this.remainderScan = this.scanRemainderInBackground(parsedFiles, canonical);
void this.remainderScan;
} else {
Expand Down Expand Up @@ -853,6 +868,7 @@ export class SessionStore {
// record — the same proxy readRawFileWithBytes gives the file path
// (#2803: the budget is a RAM proxy, not a disk one).
if (this.shadow?.readsUnified) {
const tUnion = performance.now();
let warmBytes = 0;
for (const item of warm) warmBytes += item.bytes;
for (const id of this.shadow.listSessionIds()) {
Expand All @@ -867,8 +883,9 @@ export class SessionStore {
warmBytes += bytes;
out.set(id, session);
}
timings.indexUnion = Math.round(performance.now() - tUnion);
}
this.log("info", `[persist] indexed ${this.diskFiles.length} state files; warmed ${out.size} sessions within ${limits.maxBytes} decoded bytes; older states load on demand`);
this.log("info", `[persist] indexed ${this.diskFiles.length} state files; warmed ${out.size} sessions within ${limits.maxBytes} decoded bytes (walk ${timings.walk}ms, parseWindow ${timings.parseWindow}ms, indexUnion ${timings.indexUnion}ms); older states load on demand (#4515)`);
return out;
}

Expand Down Expand Up @@ -971,6 +988,11 @@ export class SessionStore {
return this.remainderScan ?? Promise.resolve();
}

/** #4515 test hook: state files the last bounded boot decoded synchronously. */
lastBootSyncParsedFilesForTest(): number {
return this.lastBootSyncParsedFiles;
}

private readEnvelopeFile(file: string, id: string): PersistedEnvelope<PersistedSession> | null {
let bytes: Buffer;
try {
Expand Down
6 changes: 6 additions & 0 deletions src/session.ts
Original file line number Diff line number Diff line change
Expand Up @@ -1113,9 +1113,15 @@ let initialized = false;
export async function initSessions(): Promise<void> {
if (initialized) return;
initialized = true;
// #4515: split the pre-listen cost into store-open (index.db open / WAL recovery on a big store) vs bounded boot, so a slow cold start names which phase ate the spawn budget.
const tStore = performance.now();
const store = getStore();
const storeMs = Math.round(performance.now() - tStore);
if (!store.enabled) return;
const tBoot = performance.now();
const loaded = await store.boot({ maxSessions: Math.min(MAX_SESSIONS, sessionWarmMax()), maxBytes: sessionWarmMaxBytes() });
const bootMs = Math.round(performance.now() - tBoot);
loggerLog("info", `[session] startup pre-listen: store open ${storeMs}ms + bounded boot ${bootMs}ms = ${storeMs + bootMs}ms; warmed ${loaded.size} session(s) (#4515)`);
for (const [id, s] of loaded) sessions.set(id, s);
}

Expand Down
157 changes: 157 additions & 0 deletions tests/issue4515-boot-no-sync-warm.test.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,157 @@
import test from "node:test";
import assert from "node:assert/strict";
import { mkdtempSync, writeFileSync } from "node:fs";
import { tmpdir } from "node:os";
import path from "node:path";
import { createInitialState } from "acp-kernel";
import type { CompressionBlock } from "acp-kernel";
import { flatFileNameFor } from "acp-kernel/persist";
import { SessionStore } from "../src/persist.ts";
import type { Session } from "../src/session.ts";
import { rmrf } from "./tmp-rm.ts";

// #4515: at maxSessions==0 (the #2803 default) the bounded boot used to
// decode+parse a whole 64MB-decode window of the newest files synchronously
// only to pop every candidate — pure waste on the pre-listen critical path,
// and the leading contributor to cold starts blowing the native spawn budget.
// Pin the fix: full-mode boot at maxSessions==0 warms NOTHING, yet loses no
// alias bookkeeping — recorded labels resolve from the index synchronously,
// unrecorded labels fill in once the background scan finishes.

// Plain JSON session files so seeds and round-trips stay deterministic (#1080)
process.env.BILI_PERSIST_ZSTD = "0";

// Hermetic config: a developer machine's real billion-context.json must not
// leak storage settings into this suite's tmp-dir stores (same discipline as
// tests/web-sessions.test.ts).
process.env.BILI_CONFIG_FILE = path.join(
mkdtempSync(path.join(tmpdir(), "bili-4515-hermetic-")),
"no-config.json",
);
// Absent everywhere resolves to the shadow DEFAULT (#2815 rollout flip); pin
// the explicit legacy baseline — the single leg flips to full.
process.env.BILI_STORAGE_UNIFIED = "legacy";

interface Stats {
requests: number;
tokensSaved: number;
inputTokens: number;
cachedTokens: number;
outputTokens: number;
cacheSamples: number;
lastInputTokens: number;
contextTokens: number;
retrieveCalls: number;
retrieveHits: number;
retrieveMisses: number;
storedBytes: number;
storeBytesSaved: number;
compressCreditTokens: number;
rangeRestores: number;
}

function zeroStats(): Stats {
return { requests: 0, tokensSaved: 0, inputTokens: 0, cachedTokens: 0, outputTokens: 0, cacheSamples: 0, lastInputTokens: 0, contextTokens: 0, retrieveCalls: 0, retrieveHits: 0, retrieveMisses: 0, storedBytes: 0, storeBytesSaved: 0, compressCreditTokens: 0, rangeRestores: 0 };
}

function makeSession(id: string, meta: Session["meta"] = {}, patch: Partial<Stats> = {}, lastSeen = Date.now()): Session {
return {
id,
meta,
stats: { ...zeroStats(), ...patch },
metadata: {},
state: createInitialState(),
createdAt: lastSeen,
lastSeen,
blockContents: new Map(),
inFlight: 0,
persisted: false,
pendingRetrievals: [],
};
}

/** Full persisted-state shape (kernel initial state + our blocks) so hand-
* written records survive both summary extraction AND buildSession's compat
* pass in the loadSync legs. */
function stateWith(blocks: CompressionBlock[]): unknown {
const s = JSON.parse(JSON.stringify(createInitialState())) as { blocks: unknown[] };
s.blocks = blocks;
return s;
}

const mkBlock = (blockId: string, compressedTokens: number, active: boolean, topic = "T"): CompressionBlock => ({
blockId,
runId: "r1",
tier: 1,
topic,
summary: "S",
directMessageIds: [],
effectiveMessageIds: [],
directBlockIds: [],
compressedTokens,
createdAt: Date.now(),
survivedCount: 1,
generation: "young",
active,
});

test("#4515: full-mode boot at maxSessions==0 warms nothing but keeps every alias", async () => {
const dir = mkdtempSync(path.join(tmpdir(), "bili-4515-"));
const prevDir = process.env.BILI_SESSIONS_DIR;
const prevUnified = process.env.BILI_STORAGE_UNIFIED;
process.env.BILI_SESSIONS_DIR = dir;
process.env.BILI_STORAGE_UNIFIED = "full";
try {
const seeder = new SessionStore({ dir, debounceMs: 0, enabled: true });
assert.ok(seeder.replicasAvailable(), "sidecar must be live for this leg (engine available)");

// Recorded labeled session: writeNow double-ingests it, so an index
// row AND a file both exist. Its alias is SQL-seeded at boot.
const sRec = makeSession("s4515-rec", { protocol: "openai", title: "Recorded", label: "s4515-rec-label" }, { requests: 3, contextTokens: 800 });
sRec.state.blocks.push(mkBlock("b4515-1", 1500, true, "Recorded topic"));
await seeder.writeNow(sRec);
seeder.closeQuietly();

// Unrecorded labeled file: hand-written, NEVER ingested, so no index
// row exists. Its alias can ONLY come from the post-boot background
// scan — exactly the bookkeeping the skipped sync window used to do
// for in-window files.
const strayId = "s4515-stray";
const strayLabel = "s4515-stray-label";
writeFileSync(path.join(dir, flatFileNameFor(strayId)), JSON.stringify({
version: 2,
savedAt: Date.now(),
id: strayId,
meta: { protocol: "openai", title: "Stray", label: strayLabel },
stats: { ...zeroStats(), requests: 7, contextTokens: 700 },
state: stateWith([]),
}), "utf8");

const store = new SessionStore({ dir, debounceMs: 0, enabled: true });
const booted = await store.boot({ maxSessions: 0, maxBytes: 64 * 1024 * 1024 });
assert.equal(booted.size, 0, "maxSessions==0 warms nothing");
// Sharp teeth on the fix itself: NOTHING was decoded synchronously — the
// warm window is skipped entirely at maxSessions==0, not just trimmed.
assert.equal(store.lastBootSyncParsedFilesForTest(), 0, "no state file decoded synchronously at maxSessions==0");

// Recorded label resolves SYNCHRONOUSLY — the alias came from the index
// rows (seedIndexedAliases), independent of the skipped file window.
const rec = store.loadSync("s4515-rec-label");
assert.ok(rec, "SQL-seeded alias resolves before any background work");
assert.equal(rec!.id, "s4515-rec-label");
assert.equal(rec!.stats.contextTokens, 800);

// Unrecorded label fills in once the background scan reaches its file —
// proof the skipped sync window lost no corpus bookkeeping.
await store.backgroundScanComplete();
const stray = store.loadSync(strayLabel);
assert.ok(stray, "background scan recorded the unrecorded label alias");
assert.equal(stray!.id, strayLabel);
assert.equal(stray!.stats.contextTokens, 700);
store.closeQuietly();
} finally {
if (prevDir === undefined) delete process.env.BILI_SESSIONS_DIR; else process.env.BILI_SESSIONS_DIR = prevDir;
if (prevUnified === undefined) delete process.env.BILI_STORAGE_UNIFIED; else process.env.BILI_STORAGE_UNIFIED = prevUnified;
rmrf(dir);
}
});
Loading