mirror of
https://github.com/badlogic/pi-mono.git
synced 2026-11-05 23:37:25 +00:00
feat(durable): bound the Session's document cache by use (documentRetentionMs)
The Session kept every document it loaded or created in memory for the life of the process: every conversation and session document ever touched, built-in and third-party. Now a cached document that no watch or document state observes leaves the cache once it was not read or written for `documentRetentionMs` (HarnessSettings and createSession options; default ten minutes, 0 drops it right after use, Infinity keeps documents; anything that is not a number of milliseconds means the default), and loads from Storage again on its next access. An observer pins the incarnation it observes and releases it when that incarnation retires. Eviction runs on the mutation line, after every commit outcome (published, no-write, rejected, failed settlement) and from one unreferenced timer for the next expiry, so a transaction never loses a tracker it holds. Where timers cannot be unreferenced only commits evict. Cache bookkeeping is contained: a throwing clock or setting never changes an operation's outcome. A hot document moves in the eviction order at most once per granule (a tenth of the retention, up to 1 s), keeping the fan-out bench at HEAD's speed. `npm test` runs the durable suite a second time with every unattached document evicted right after use. 5,000 conversations kept 26,000 documents and about 3.8 KB each; with a 50 ms retention, after waiting, none (bench:document-cache).
This commit is contained in:
Mario Zechner
parent
ea448f454a
commit
f44707e7f7
16 changed files with 994 additions and 23 deletions
|
|
@ -27,6 +27,7 @@
|
|||
- A task whose fault write failed no longer runs again in the same process.
|
||||
- The scheduler no longer visits every live task on each pass: abort cascades start only from tasks with cancellation intent, owned live work and idle scopes are walks down an in-memory ownership index, and reservation, finalization, and failFast checks keep their own candidate sets. A chain of 500 owned tasks that took 12 s to settle takes 50 ms, 20,000 take 1.5 s; a tool making 8000 parallel nested calls finishes in 2.7 s instead of 18.8 s, with one document per nested call instead of one growing index (`npm run bench:scheduler-fanout`).
|
||||
- The scheduler's ownership indexes are bounded by live work: it keeps a conversation's owner and an ended task's ownership only while live work is below them, instead of every conversation and subagent chain it ever saw. Work that later appears in an old conversation loads its chain from storage again. After 10,000 ended subagent calls under a live parent it kept 10,000 ended tasks and 10,001 conversation edges; now none of either beyond the parent's conversation (`npm run bench:scheduler-memory`). A queued input no longer makes the scheduler scan every queued submission when nothing is being cancelled.
|
||||
- The Session's document cache is bounded by use: a loaded document that no watch or document state observes leaves memory once it was not read or written for `documentRetentionMs` (new in `HarnessSettings` and `createSession()` options; default ten minutes, `0` drops it right after use) and loads from Storage again on its next access. It held every conversation and session document ever touched, built-in and third-party, for the life of the process: 5,000 conversations kept 26,000 documents and about 3.8 KB each; with retention 0 they keep none (`npm run bench:document-cache`).
|
||||
|
||||
## [1.1.0] - 2026-10-07
|
||||
|
||||
|
|
|
|||
|
|
@ -302,6 +302,7 @@ const harness = await Harness.open(storage, {
|
|||
progress: { partialIntervalMs: 100, outputIntervalMs: 100 },
|
||||
toolExecution: "parallel",
|
||||
contextRetentionMs: 600_000, // idle conversations keep their context in memory this long (checked lazily where timers cannot be unreferenced)
|
||||
documentRetentionMs: 600_000, // loaded documents nothing observes stay in memory this long after their last use; observed ones always
|
||||
get followUpMode() {
|
||||
return userSettings.followUpMode;
|
||||
},
|
||||
|
|
@ -616,7 +617,7 @@ const usage = await harness.usage(context); // { models: { "openai/gpt-6-sol": U
|
|||
|---|---|---|
|
||||
| Memory | `MemoryStorage` from the package root | Nothing is persisted. |
|
||||
| SQLite | `openNodeSqliteStorage(file)` from `@earendil-works/pi-durable/storage/sqlite/node` | One database file. WAL mode with `synchronous = NORMAL`: commits survive process crashes; the newest may be lost on power or host failure. |
|
||||
| JSONL | `openNodeJsonlStorage(directory, context)` from `@earendil-works/pi-durable/storage/jsonl/node` | Append-only files in one directory. Pass `{ fsync: true }` to flush before each commit marker. |
|
||||
| JSONL | `openNodeJsonlStorage(directory, context)` from `@earendil-works/pi-durable/storage/jsonl/node` | Append-only files in one directory. Pass `{ fsync: true }` to flush before each commit marker. Keeps the whole history in memory too, so memory grows with it; use SQLite for long-running agents. |
|
||||
| Cloudflare Durable Object | `openDurableObjectSqliteStorage(ctx.storage)` from `@earendil-works/pi-durable/storage/sqlite/cloudflare` | The SQLite storage of one SQLite-backed Durable Object; commits are its storage transactions. |
|
||||
|
||||
One process owns a storage at a time; there is no cross-process locking. The portable SQLite and JSONL cores (`/storage/sqlite`, `/storage/jsonl`) run without Node APIs, for example on Bun or in Cloudflare Durable Objects, given an asynchronous `SqliteDatabase` facade or a `FileSystem` from `@earendil-works/pi-durable/env`.
|
||||
|
|
|
|||
|
|
@ -415,6 +415,8 @@ type HarnessSettings = {
|
|||
readonly followUpMode?: QueueMode;
|
||||
/** How long an idle conversation keeps its last context read in memory; busy ones always keep it. */
|
||||
readonly contextRetentionMs?: number;
|
||||
/** How long a loaded document no watch or document state observes stays in memory after its last use (section 4). */
|
||||
readonly documentRetentionMs?: number;
|
||||
};
|
||||
|
||||
/** Resolved: every field over its built-in default, object fields merged. */
|
||||
|
|
@ -427,6 +429,7 @@ type Settings = {
|
|||
readonly steeringMode: QueueMode; // "one-at-a-time"
|
||||
readonly followUpMode: QueueMode; // "one-at-a-time"
|
||||
readonly contextRetentionMs: number; // 600000
|
||||
readonly documentRetentionMs: number; // 600000
|
||||
};
|
||||
|
||||
/** Stored choices of one conversation; names, not objects. Unset fields follow the host. */
|
||||
|
|
@ -1567,6 +1570,24 @@ Session-line operation so it never exposes a mixture from one commit.
|
|||
|
||||
## 4. Transactions and storage ownership
|
||||
|
||||
The Session keeps the documents it loads or creates in memory, as Chord
|
||||
trackers, so reads and changes need no Storage access. The cache is bounded by
|
||||
use, not by history: a document no watch or document state observes leaves it
|
||||
once it was not read or written for `documentRetentionMs` (default ten
|
||||
minutes; `0` drops it right after use), and the next access loads it from
|
||||
Storage. An observed document never leaves. Eviction runs on the mutation line,
|
||||
after each commit and when an unreferenced timer for the next expiry fires
|
||||
(the oldest idle document's, or sooner once the retention is shortened), so a
|
||||
transaction never loses a tracker it holds. A retention that is not a number of
|
||||
milliseconds, `0` or more, means the default; `Infinity` keeps documents. An
|
||||
observer pins the incarnation it observes, and releases it when that
|
||||
incarnation retires. Where timers cannot be
|
||||
unreferenced, as in Cloudflare Workers, only commits evict. A document that is
|
||||
used again and again moves to the back of the eviction order at most once per
|
||||
granule, a tenth of the retention up to one second. Task documents also retire
|
||||
with their task (section 5.5). `createSession(storage, { documentRetentionMs })`
|
||||
sets it for a plain Session; the Harness reads its settings at each use.
|
||||
|
||||
A Session commit callback may be asynchronous. It owns the Session mutation
|
||||
line through callback execution, preparation, storage settlement, committed
|
||||
baseline adoption, and publication enqueue. Commit observers run on the
|
||||
|
|
|
|||
|
|
@ -101,14 +101,15 @@
|
|||
"scripts": {
|
||||
"clean": "shx rm -rf dist",
|
||||
"build": "tsc -p tsconfig.build.json",
|
||||
"test": "vitest --run",
|
||||
"test": "vitest --run && PI_DURABLE_DOCUMENT_RETENTION_MS=0 vitest --run",
|
||||
"bench:storage": "vitest bench --config vitest.benchmark.config.ts",
|
||||
"bench:storage:memory": "node --conditions=source --experimental-strip-types test/storage-memory.ts",
|
||||
"bench:tool-output": "node --conditions=source --experimental-strip-types --expose-gc test/tool-output-bench.ts",
|
||||
"bench:nested-tools": "node --conditions=source --experimental-strip-types --expose-gc test/nested-tools-bench.ts",
|
||||
"bench:nested-images": "node --conditions=source --experimental-strip-types --expose-gc test/nested-images-bench.ts",
|
||||
"bench:scheduler-fanout": "node --conditions=source --experimental-strip-types test/scheduler-fanout-bench.ts",
|
||||
"bench:scheduler-memory": "node --conditions=source --experimental-strip-types --expose-gc test/scheduler-memory-bench.ts",
|
||||
"bench:scheduler-memory": "node --conditions=source --experimental-strip-types test/scheduler-memory-bench.ts",
|
||||
"bench:document-cache": "node --conditions=source --experimental-strip-types test/session-document-cache-bench.ts",
|
||||
"prepublishOnly": "npm run clean && npm run build"
|
||||
},
|
||||
"keywords": [
|
||||
|
|
|
|||
|
|
@ -1,5 +1,6 @@
|
|||
import { copyJson, type Draft } from "@earendil-works/chord";
|
||||
import { defineDoc } from "../documents.ts";
|
||||
import { documentRetention } from "../session/session.ts";
|
||||
import type { ConversationId, ConversationRecord, Tx } from "../types.ts";
|
||||
import type {
|
||||
Agent,
|
||||
|
|
@ -69,6 +70,7 @@ export function resolveSettings(settings: HarnessSettings | undefined): Settings
|
|||
steeringMode: settings?.steeringMode ?? "one-at-a-time",
|
||||
followUpMode: settings?.followUpMode ?? "one-at-a-time",
|
||||
contextRetentionMs: settings?.contextRetentionMs ?? DEFAULT_CONTEXT_RETENTION_MS,
|
||||
documentRetentionMs: documentRetention(settings?.documentRetentionMs),
|
||||
};
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -177,7 +177,8 @@ class HarnessImpl<Tool extends ToolRegistration> extends SessionImpl implements
|
|||
constructor(rawStorage: Storage, options: HarnessOptions<Tool>, context: Context) {
|
||||
const report = safeReport(options.onReport);
|
||||
const now = safeNow(options.now, report);
|
||||
super(rawStorage, now);
|
||||
// Read at each use, as all settings are, but only this field: it is read on every commit.
|
||||
super(rawStorage, now, () => options.settings?.documentRetentionMs);
|
||||
// Every component reads and writes through the Session's failure guard: `rawStorage` serves only `super()`.
|
||||
const storage = this.storage;
|
||||
this.#options = options;
|
||||
|
|
|
|||
|
|
@ -518,6 +518,12 @@ export type HarnessSettings = {
|
|||
* Busy conversations always keep it; `0` drops it once the conversation is idle.
|
||||
*/
|
||||
readonly contextRetentionMs?: number;
|
||||
/**
|
||||
* How long a loaded document that no watch or document state observes stays in memory after its last read or write;
|
||||
* the next access loads it from Storage. Observed documents always stay. Default ten minutes; `0` drops one right
|
||||
* after use; `Infinity` keeps them. Anything else that is not a number of milliseconds means the default.
|
||||
*/
|
||||
readonly documentRetentionMs?: number;
|
||||
};
|
||||
|
||||
/** Resolved settings: every field over its built-in default, object fields merged. */
|
||||
|
|
@ -532,6 +538,7 @@ export type Settings = {
|
|||
readonly steeringMode: QueueMode;
|
||||
readonly followUpMode: QueueMode;
|
||||
readonly contextRetentionMs: number;
|
||||
readonly documentRetentionMs: number;
|
||||
};
|
||||
|
||||
/** What `HarnessOptions.env` builds an environment for. */
|
||||
|
|
|
|||
|
|
@ -3,6 +3,7 @@ import { awaitWithContext, BACKGROUND_CONTEXT, withoutAbortSignal } from "@earen
|
|||
import { type Op, track } from "@earendil-works/chord/delta";
|
||||
import {
|
||||
type AnyDocToken,
|
||||
addressId,
|
||||
checkRecordScope,
|
||||
checkRecordVersion,
|
||||
materializeDocument,
|
||||
|
|
@ -65,13 +66,40 @@ import { type LoadedDocument, Transaction, type TransactionHost, type Transactio
|
|||
* Open a Session kernel over one storage backend. `now` is the wall clock for task times; default `Date.now`. The first
|
||||
* error a Storage method throws fails the Session (`SessionFailed`); see `SessionImpl`.
|
||||
*/
|
||||
export function createSession(storage: Storage, options?: { readonly now?: () => number }): Session {
|
||||
return new SessionImpl(storage, options?.now);
|
||||
export function createSession(
|
||||
storage: Storage,
|
||||
options?: { readonly now?: () => number; readonly documentRetentionMs?: number },
|
||||
): Session {
|
||||
const retention = options?.documentRetentionMs;
|
||||
return new SessionImpl(storage, options?.now, retention === undefined ? undefined : () => retention);
|
||||
}
|
||||
|
||||
/** Default document retention: ten minutes, as for kept contexts. */
|
||||
let defaultRetention = 600_000;
|
||||
|
||||
/** Internal, for tests: the retention of every Session that keeps the default, as a suite run with eviction at once. */
|
||||
export function setDefaultDocumentRetention(ms: number): void {
|
||||
defaultRetention = ms;
|
||||
}
|
||||
|
||||
/**
|
||||
* A retention as given, when it is a number of milliseconds, `0` or more (`Infinity` keeps documents); otherwise the
|
||||
* default.
|
||||
*/
|
||||
export function documentRetention(ms: number | undefined): number {
|
||||
return typeof ms === "number" && ms >= 0 ? ms : defaultRetention;
|
||||
}
|
||||
|
||||
/** Longest delay `setTimeout` supports; a longer retention waits in several steps. */
|
||||
const MAX_TIMER_DELAY = 2_147_483_647;
|
||||
|
||||
/**
|
||||
* Session kernel: one mutation line, the loaded document tracker cache, and committed publication.
|
||||
*
|
||||
* The cache is bounded by use: a document with no watch or document state attached leaves it once it was not read or
|
||||
* written for the retention period, and the next access loads it from Storage. Eviction runs as a job on the mutation
|
||||
* line, so no transaction ever holds a tracker that leaves.
|
||||
*
|
||||
* Only committed state is observable. Every commit callback, preparation, Storage settlement, adoption, and
|
||||
* publication enqueue runs while the line is held; listeners run later.
|
||||
*
|
||||
|
|
@ -84,6 +112,17 @@ export function createSession(storage: Storage, options?: { readonly now?: () =>
|
|||
export class SessionImpl implements Session {
|
||||
readonly #storage: Storage;
|
||||
readonly #documents = new Map<string, LoadedDocument>();
|
||||
/** Cached documents with no observer attached, by address, oldest use first, with the time of that use. */
|
||||
readonly #idleDocuments = new Map<string, number>();
|
||||
/** Observers attached per document incarnation (record ID); a cached incarnation with any is never evicted. */
|
||||
readonly #attachedDocuments = new Map<DocumentId, number>();
|
||||
readonly #clock: () => number;
|
||||
readonly #documentRetentionMs: () => number;
|
||||
/** The pending eviction: a timer for the oldest idle document, or a line job when the retention is 0. */
|
||||
#documentSweep: { readonly at: number; readonly timer: ReturnType<typeof setTimeout> | undefined } | undefined;
|
||||
#sweepJobs = 0;
|
||||
/** Set once a timer could not be unreferenced: then none is kept, and the end of each commit evicts what expired. */
|
||||
#documentTimersUnavailable = false;
|
||||
readonly #commitListeners = new Set<(publication: CommitPublication, context: Context) => void>();
|
||||
/** The Session's own listeners, such as its scheduler's: they keep memory in step with storage, so a throw fails it. */
|
||||
readonly #internalListeners = new Set<(publication: CommitPublication, context: Context) => void>();
|
||||
|
|
@ -110,20 +149,45 @@ export class SessionImpl implements Session {
|
|||
return this.#failed.promise;
|
||||
}
|
||||
|
||||
constructor(storage: Storage, now: () => number = Date.now) {
|
||||
constructor(
|
||||
storage: Storage,
|
||||
now: () => number = Date.now,
|
||||
documentRetentionMs: () => number | undefined = () => undefined,
|
||||
) {
|
||||
// Waits race it; none has to.
|
||||
this.#failed.promise.catch(() => {});
|
||||
this.#storage = new GuardedStorage(storage, this);
|
||||
// Cache bookkeeping must never change the outcome of the operation it follows: a clock or a retention setting
|
||||
// that throws counts as now, and as the default.
|
||||
this.#clock = () => {
|
||||
try {
|
||||
return now();
|
||||
} catch {
|
||||
return Date.now();
|
||||
}
|
||||
};
|
||||
this.#documentRetentionMs = () => {
|
||||
try {
|
||||
return documentRetention(documentRetentionMs());
|
||||
} catch {
|
||||
return documentRetention(undefined);
|
||||
}
|
||||
};
|
||||
this.#host = {
|
||||
storage: this.#storage,
|
||||
now,
|
||||
cached: (id) => this.#documents.get(id),
|
||||
cached: (id) => {
|
||||
const cached = this.#documents.get(id);
|
||||
if (cached !== undefined) this.#touchDocument(id);
|
||||
return cached;
|
||||
},
|
||||
load: (definition, addressId, address, context) => this.#loadDocument(definition, addressId, address, context),
|
||||
install: (document) => {
|
||||
this.#documents.set(document.addressId, document);
|
||||
this.#touchDocument(document.addressId);
|
||||
},
|
||||
evict: (id, recordId) => {
|
||||
if (this.#documents.get(id)?.record.id === recordId) this.#documents.delete(id);
|
||||
if (this.#documents.get(id)?.record.id === recordId) this.#dropDocument(id);
|
||||
},
|
||||
conversationCreated: (tx, record) => this.conversationCreated(tx, record),
|
||||
};
|
||||
|
|
@ -232,6 +296,7 @@ export class SessionImpl implements Session {
|
|||
const resolved = resolveAddress(definition, args);
|
||||
const context = args[resolved.nextArgument] as Context;
|
||||
const cached = this.#documents.get(resolved.id);
|
||||
if (cached?.valueVersion === definition.version) this.#touchDocument(resolved.id);
|
||||
const loaded =
|
||||
cached?.valueVersion === definition.version
|
||||
? cached
|
||||
|
|
@ -426,6 +491,8 @@ export class SessionImpl implements Session {
|
|||
this.#commitListeners.clear();
|
||||
this.#internalListeners.clear();
|
||||
this.#documents.clear();
|
||||
this.#idleDocuments.clear();
|
||||
this.#attachedDocuments.clear();
|
||||
try {
|
||||
await this.#storage.close(cleanup);
|
||||
} catch (error) {
|
||||
|
|
@ -451,6 +518,8 @@ export class SessionImpl implements Session {
|
|||
|
||||
/** Run the close listeners once, each on its own: one that fails is reported, and the rest still run. */
|
||||
#notifyClose(): void {
|
||||
if (this.#documentSweep?.timer !== undefined) clearTimeout(this.#documentSweep.timer);
|
||||
this.#documentSweep = undefined;
|
||||
const listeners = [...this.#closeListeners];
|
||||
this.#closeListeners.clear();
|
||||
for (const listener of listeners) contained(listener, (error) => this.report(error));
|
||||
|
|
@ -503,9 +572,121 @@ export class SessionImpl implements Session {
|
|||
unloadDocuments(): Promise<void> {
|
||||
return this.#enqueue(async () => {
|
||||
this.#documents.clear();
|
||||
this.#idleDocuments.clear();
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Internal, for tests: how many documents the cache holds, how many are idle, how many incarnations are pinned, and how
|
||||
* many sweep jobs were queued on the line so far.
|
||||
*/
|
||||
get documentCacheSizes(): {
|
||||
readonly cached: number;
|
||||
readonly idle: number;
|
||||
readonly attached: number;
|
||||
readonly sweeps: number;
|
||||
} {
|
||||
return {
|
||||
cached: this.#documents.size,
|
||||
idle: this.#idleDocuments.size,
|
||||
attached: this.#attachedDocuments.size,
|
||||
sweeps: this.#sweepJobs,
|
||||
};
|
||||
}
|
||||
|
||||
/** Note a use of a cached document: an unattached one becomes the newest idle one, and an eviction is scheduled. */
|
||||
#touchDocument(addressId: string): void {
|
||||
const cached = this.#documents.get(addressId);
|
||||
if (cached === undefined || this.#attachedDocuments.has(cached.record.id)) return;
|
||||
const now = this.#clock();
|
||||
const since = this.#idleDocuments.get(addressId);
|
||||
// A hot document moves to the back at most once per granule, a tenth of the retention up to 1 s: it may then leave
|
||||
// up to a granule early, which costs a reload, while a commit-heavy workload skips the reordering.
|
||||
if (since !== undefined && now - since < Math.min(1_000, this.#documentRetentionMs() / 10)) return;
|
||||
this.#idleDocuments.delete(addressId);
|
||||
this.#idleDocuments.set(addressId, now);
|
||||
// A pending sweep for the oldest document comes first unless the retention was shortened since; then it moves up.
|
||||
if (this.#documentSweep === undefined || this.#documentSweep.at > now + this.#documentRetentionMs()) {
|
||||
this.#scheduleDocumentSweep();
|
||||
}
|
||||
}
|
||||
|
||||
#dropDocument(addressId: string): void {
|
||||
this.#documents.delete(addressId);
|
||||
this.#idleDocuments.delete(addressId);
|
||||
}
|
||||
|
||||
/**
|
||||
* Schedule the eviction of the oldest idle document: on the line at once when the retention is 0, otherwise when it
|
||||
* expires, by a timer that never keeps the process alive. Where timers cannot be unreferenced, as in Cloudflare
|
||||
* Workers, none is kept: the end of each commit evicts what expired.
|
||||
*/
|
||||
#scheduleDocumentSweep(): void {
|
||||
if (this.#closing !== undefined) return;
|
||||
const oldest = this.#idleDocuments.values().next();
|
||||
if (oldest.done === true) return;
|
||||
const at = oldest.value + this.#documentRetentionMs();
|
||||
if (this.#documentSweep !== undefined && this.#documentSweep.at <= at) return;
|
||||
if (this.#documentSweep?.timer !== undefined) clearTimeout(this.#documentSweep.timer);
|
||||
// The pending sweep stays set until its job runs, so touches while the line is busy queue no more.
|
||||
const sweep = { at, timer: undefined as ReturnType<typeof setTimeout> | undefined };
|
||||
const run = (): void => {
|
||||
this.#sweepJobs++;
|
||||
this.#enqueue(async () => {
|
||||
if (this.#documentSweep === sweep) this.#documentSweep = undefined;
|
||||
if (this.#closing === undefined) this.#sweepDocuments();
|
||||
}).catch(() => {});
|
||||
};
|
||||
const delay = at - this.#clock();
|
||||
this.#documentSweep = sweep;
|
||||
if (delay <= 0) {
|
||||
queueMicrotask(run);
|
||||
return;
|
||||
}
|
||||
if (this.#documentTimersUnavailable) {
|
||||
this.#documentSweep = undefined;
|
||||
return;
|
||||
}
|
||||
const timer = setTimeout(run, Math.min(delay, MAX_TIMER_DELAY));
|
||||
const unref = (timer as { unref?: unknown }).unref;
|
||||
if (typeof unref !== "function") {
|
||||
clearTimeout(timer);
|
||||
this.#documentTimersUnavailable = true;
|
||||
this.#documentSweep = undefined;
|
||||
return;
|
||||
}
|
||||
unref.call(timer);
|
||||
sweep.timer = timer;
|
||||
}
|
||||
|
||||
/** On the line: evict the idle documents whose retention passed, oldest first, then schedule the next eviction. */
|
||||
#sweepDocuments(): void {
|
||||
const expired = this.#clock() - this.#documentRetentionMs();
|
||||
for (const [addressId, since] of this.#idleDocuments) {
|
||||
if (since > expired) break;
|
||||
this.#dropDocument(addressId);
|
||||
}
|
||||
this.#scheduleDocumentSweep();
|
||||
}
|
||||
|
||||
/** Pin a cached incarnation for one observer; the returned release unpins it once, idle again after the last. */
|
||||
#pinDocument(addressId: string, recordId: DocumentId): () => void {
|
||||
this.#attachedDocuments.set(recordId, (this.#attachedDocuments.get(recordId) ?? 0) + 1);
|
||||
this.#idleDocuments.delete(addressId);
|
||||
let pinned = true;
|
||||
return () => {
|
||||
if (!pinned) return;
|
||||
pinned = false;
|
||||
const count = (this.#attachedDocuments.get(recordId) ?? 1) - 1;
|
||||
if (count > 0) {
|
||||
this.#attachedDocuments.set(recordId, count);
|
||||
return;
|
||||
}
|
||||
this.#attachedDocuments.delete(recordId);
|
||||
if (this.#documents.get(addressId)?.record.id === recordId) this.#touchDocument(addressId);
|
||||
};
|
||||
}
|
||||
|
||||
async #runCommit<T>(
|
||||
change: (tx: Transaction) => T | Promise<T>,
|
||||
context: Context,
|
||||
|
|
@ -521,11 +702,20 @@ export class SessionImpl implements Session {
|
|||
this.#assertHealthy();
|
||||
} catch (error) {
|
||||
await tx.settleFailure();
|
||||
this.#sweepDocuments();
|
||||
throw error;
|
||||
}
|
||||
let writes: readonly StorageWrite[];
|
||||
try {
|
||||
writes = await tx.settleSuccess();
|
||||
} catch (error) {
|
||||
// Settlement aborted every change already.
|
||||
this.#sweepDocuments();
|
||||
throw error;
|
||||
}
|
||||
const writes = await tx.settleSuccess();
|
||||
if (writes.length === 0) {
|
||||
tx.discard();
|
||||
this.#sweepDocuments();
|
||||
return result;
|
||||
}
|
||||
let seq: Seq;
|
||||
|
|
@ -545,7 +735,9 @@ export class SessionImpl implements Session {
|
|||
this.fail(error);
|
||||
throw error;
|
||||
}
|
||||
for (const document of documents) this.#touchDocument(addressId(document.record));
|
||||
this.#publish(seq, writes, documents, context);
|
||||
this.#sweepDocuments();
|
||||
return result;
|
||||
}
|
||||
|
||||
|
|
@ -596,11 +788,20 @@ export class SessionImpl implements Session {
|
|||
checkRecordVersion(definition, loaded.record, loaded.storedVersion);
|
||||
let unsubscribeCommit = (): void => {};
|
||||
let unsubscribeClose = (): void => {};
|
||||
// Attached, the incarnation stays cached; once its last observer detaches, or it retires, it is idle again.
|
||||
const unpin = this.#pinDocument(loaded.addressId, loaded.record.id);
|
||||
const detach = (): void => {
|
||||
unsubscribeCommit();
|
||||
unsubscribeClose();
|
||||
unpin();
|
||||
};
|
||||
const observer = create(loaded.tracker.value, detach);
|
||||
let observer: O;
|
||||
try {
|
||||
observer = create(loaded.tracker.value, detach);
|
||||
} catch (error) {
|
||||
unpin();
|
||||
throw error;
|
||||
}
|
||||
const observed = { version: loaded.valueVersion };
|
||||
unsubscribeCommit = this.observeCommits((publication, context) => {
|
||||
for (const change of publication.changes) {
|
||||
|
|
@ -610,6 +811,8 @@ export class SessionImpl implements Session {
|
|||
const ops = observedOperations(observed, change);
|
||||
// A migration-only base changes nothing for an observer of the new version.
|
||||
if (ops.length === 0) continue;
|
||||
// A retired incarnation never changes again: its observers keep their last value but pin nothing.
|
||||
if (change.value === null) unpin();
|
||||
observer.advance(change.value, ops, frameContext);
|
||||
}
|
||||
});
|
||||
|
|
@ -625,8 +828,11 @@ export class SessionImpl implements Session {
|
|||
): Promise<LoadedDocument | undefined> {
|
||||
const cached = this.#documents.get(addressId);
|
||||
// A tracker serves only tokens of the version its value was materialized for; others reload from Storage.
|
||||
if (cached?.valueVersion === definition.version) return cached;
|
||||
if (cached !== undefined) this.#documents.delete(addressId);
|
||||
if (cached?.valueVersion === definition.version) {
|
||||
this.#touchDocument(addressId);
|
||||
return cached;
|
||||
}
|
||||
if (cached !== undefined) this.#dropDocument(addressId);
|
||||
const record = await this.#storage.findDocument(address, "current", context);
|
||||
if (record === undefined) return undefined;
|
||||
const stored = await this.#storage.document(record.id, "current", context);
|
||||
|
|
@ -641,6 +847,7 @@ export class SessionImpl implements Session {
|
|||
tracker: track(value),
|
||||
};
|
||||
this.#documents.set(addressId, loaded);
|
||||
this.#touchDocument(addressId);
|
||||
return loaded;
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -494,6 +494,7 @@ describe("generation", () => {
|
|||
steeringMode: "one-at-a-time",
|
||||
followUpMode: "one-at-a-time",
|
||||
contextRetentionMs: 600_000,
|
||||
documentRetentionMs: expect.any(Number),
|
||||
});
|
||||
expect(resolveSettings({ retry: { enabled: false }, compaction: { backgroundTokens: 0 } })).toMatchObject({
|
||||
retry: { enabled: false, maxRetries: 3, baseDelayMs: 2000 },
|
||||
|
|
|
|||
26
packages/durable/test/heap-retained.ts
Normal file
26
packages/durable/test/heap-retained.ts
Normal file
|
|
@ -0,0 +1,26 @@
|
|||
// For benchmarks: the bytes live objects hold, from a heap snapshot. `heapUsed` after `gc()` also counts what the
|
||||
// collector has not compacted yet, which hides a small leak or invents a large one.
|
||||
import { Session } from "node:inspector/promises";
|
||||
|
||||
/** Sum of the self sizes of every object in a fresh heap snapshot, which collects garbage first. */
|
||||
export async function retainedBytes(): Promise<number> {
|
||||
const inspector = new Session();
|
||||
inspector.connect();
|
||||
const chunks: string[] = [];
|
||||
inspector.on("HeapProfiler.addHeapSnapshotChunk", (message) => chunks.push(message.params.chunk));
|
||||
try {
|
||||
await inspector.post("HeapProfiler.takeHeapSnapshot", { reportProgress: false });
|
||||
} finally {
|
||||
inspector.disconnect();
|
||||
}
|
||||
const snapshot = JSON.parse(chunks.join("")) as {
|
||||
snapshot: { meta: { node_fields: string[] } };
|
||||
nodes: number[];
|
||||
};
|
||||
const fields = snapshot.snapshot.meta.node_fields;
|
||||
const width = fields.length;
|
||||
const size = fields.indexOf("self_size");
|
||||
let total = 0;
|
||||
for (let index = size; index < snapshot.nodes.length; index += width) total += snapshot.nodes[index]!;
|
||||
return total;
|
||||
}
|
||||
|
|
@ -1,10 +1,10 @@
|
|||
/**
|
||||
* Scheduler memory benchmark. Under one live parent task, N subagent cycles run and end: a tool task owned by the
|
||||
* parent, a conversation owned by the tool task, and a task in that conversation. Reports the scheduler's ownership
|
||||
* index sizes and the heap after GC once all cycles ended, then the storage reads a later message into an old subagent
|
||||
* index sizes and the bytes live objects retain (from heap snapshots) once all cycles ended, then the storage reads a later message into an old subagent
|
||||
* conversation costs. On SQLite, so durable history stays out of the heap.
|
||||
*
|
||||
* node --conditions=source --experimental-strip-types --expose-gc test/scheduler-memory-bench.ts [cycles...]
|
||||
* node --conditions=source --experimental-strip-types test/scheduler-memory-bench.ts [cycles...]
|
||||
*/
|
||||
import { mkdtempSync, rmSync } from "node:fs";
|
||||
import { tmpdir } from "node:os";
|
||||
|
|
@ -21,9 +21,7 @@ import {
|
|||
type TaskId,
|
||||
} from "../src/index.ts";
|
||||
import { openNodeSqliteStorage } from "../src/storage/sqlite/node.ts";
|
||||
|
||||
const gc = (globalThis as { gc?: () => void }).gc;
|
||||
if (gc === undefined) throw new Error("Run with --expose-gc");
|
||||
import { retainedBytes } from "./heap-retained.ts";
|
||||
|
||||
const waiting = new Map<string, () => void>();
|
||||
/** Holds until released by name, then completes. */
|
||||
|
|
@ -73,8 +71,7 @@ async function run(cycles: number) {
|
|||
(tx) => tx.createTask(Hold, { name: "parent" }, { ownership: { kind: "conversation" } }),
|
||||
context,
|
||||
);
|
||||
gc!();
|
||||
const heapBefore = process.memoryUsage().heapUsed;
|
||||
const heapBefore = await retainedBytes();
|
||||
let first: ConversationId | undefined;
|
||||
const started = performance.now();
|
||||
for (let n = 0; n < cycles; n++) {
|
||||
|
|
@ -99,8 +96,7 @@ async function run(cycles: number) {
|
|||
first ??= child;
|
||||
}
|
||||
const ms = Math.round(performance.now() - started);
|
||||
gc!();
|
||||
const heapAfter = process.memoryUsage().heapUsed;
|
||||
const heapAfter = await retainedBytes();
|
||||
const sizes = (harness as unknown as Record<symbol, Sizes>)[SCHEDULER_INDEX_SIZES]!;
|
||||
// A later message into an old subagent conversation: a task created there, run, and ended.
|
||||
const readsBefore = reads.count;
|
||||
|
|
|
|||
105
packages/durable/test/session-document-cache-bench.ts
Normal file
105
packages/durable/test/session-document-cache-bench.ts
Normal file
|
|
@ -0,0 +1,105 @@
|
|||
/**
|
||||
* Session document cache benchmark. A Harness on SQLite creates N conversations and runs one short task in each, as a
|
||||
* long-running agent with subagents does; then, at a few retention settings, reports the documents left in the cache,
|
||||
* the bytes live objects retain (from heap snapshots) per conversation, and the cost of reading a document of an old conversation again.
|
||||
*
|
||||
* node --conditions=source --experimental-strip-types test/session-document-cache-bench.ts [N...]
|
||||
*/
|
||||
import { mkdtempSync, rmSync } from "node:fs";
|
||||
import { tmpdir } from "node:os";
|
||||
import { join } from "node:path";
|
||||
import { BACKGROUND_CONTEXT as context } from "@earendil-works/chord/context";
|
||||
import { createModels } from "@earendil-works/pi-ai/models";
|
||||
import {
|
||||
type ConversationId,
|
||||
createRegistry,
|
||||
defineExtension,
|
||||
defineTask,
|
||||
Harness,
|
||||
LiveDoc,
|
||||
UsageDoc,
|
||||
} from "../src/index.ts";
|
||||
import type { SessionImpl } from "../src/session/session.ts";
|
||||
import { openNodeSqliteStorage } from "../src/storage/sqlite/node.ts";
|
||||
import { retainedBytes } from "./heap-retained.ts";
|
||||
|
||||
/** Writes `pi.usage` of its conversation once, then completes. */
|
||||
const Work = defineTask<null, { phase: "work" }, null>({
|
||||
name: "bench.work",
|
||||
version: 1,
|
||||
initial: () => ({ phase: "work" }),
|
||||
phases: {
|
||||
work: async (_task, runtime, ctx) =>
|
||||
runtime.commit(async (tx) => {
|
||||
(await tx.doc(UsageDoc, runtime.conversationId)).tools.bench = { input: 1 } as never;
|
||||
return { status: "terminal", outcome: { status: "completed", result: null } };
|
||||
}, ctx),
|
||||
},
|
||||
abort: async (_task, runtime, ctx) =>
|
||||
runtime.commit(() => ({ status: "terminal", outcome: { status: "aborted" } }), ctx),
|
||||
});
|
||||
|
||||
async function run(conversations: number, documentRetentionMs: number) {
|
||||
const directory = mkdtempSync(join(tmpdir(), "pi-durable-doc-cache-bench-"));
|
||||
try {
|
||||
const storage = await openNodeSqliteStorage(join(directory, "session.sqlite"));
|
||||
const registry = createRegistry();
|
||||
registry.install(defineExtension({ name: "bench", tasks: [Work] }));
|
||||
const harness = await Harness.open(
|
||||
storage,
|
||||
{ models: createModels(), registry, settings: { documentRetentionMs } },
|
||||
context,
|
||||
);
|
||||
harness.resume();
|
||||
let first: ConversationId | undefined;
|
||||
const cycle = async (): Promise<void> => {
|
||||
const conversation = await harness.createConversation({ ownership: { kind: "ownerless" } }, context);
|
||||
first ??= conversation.id;
|
||||
const task = await conversation.commit(
|
||||
(tx) => tx.createTask(Work, null, { ownership: { kind: "conversation" } }),
|
||||
context,
|
||||
);
|
||||
const settled = await harness.waitForTask(task, context);
|
||||
if (settled.state.outcome.status !== "completed")
|
||||
throw new Error(`task ${task} ${settled.state.outcome.status}`);
|
||||
};
|
||||
// Warm up first: compiled code and storage caches would otherwise count as growth.
|
||||
for (let n = 0; n < 200; n++) await cycle();
|
||||
await (harness as unknown as SessionImpl).readOnLine(async () => {});
|
||||
const heapBefore = await retainedBytes();
|
||||
const started = performance.now();
|
||||
for (let n = 0; n < conversations; n++) await cycle();
|
||||
const ms = Math.round(performance.now() - started);
|
||||
// Wait past a short retention without committing, so only the timer can evict; then let its job run.
|
||||
if (documentRetentionMs > 0 && documentRetentionMs < 1_000) {
|
||||
await new Promise((resolve) => setTimeout(resolve, documentRetentionMs * 4));
|
||||
}
|
||||
await (harness as unknown as SessionImpl).readOnLine(async () => {});
|
||||
const heap = (await retainedBytes()) - heapBefore;
|
||||
const cached = (harness as unknown as SessionImpl).documentCacheSizes.cached;
|
||||
// One read of an old conversation's document: from the cache, or from storage after eviction.
|
||||
const readStarted = performance.now();
|
||||
for (let n = 0; n < 100; n++) await harness.snapshot(LiveDoc, first!, context);
|
||||
const readUs = Math.round(((performance.now() - readStarted) / 100) * 1000);
|
||||
await harness.close(context);
|
||||
return { ms, heap, cached, readUs };
|
||||
} finally {
|
||||
rmSync(directory, { recursive: true, force: true });
|
||||
}
|
||||
}
|
||||
|
||||
const sizes = process.argv.slice(2).map(Number);
|
||||
console.log("conversations | retention | ms | cached docs | retained bytes | old-doc read");
|
||||
for (const n of sizes.length > 0 ? sizes : [1000, 5000]) {
|
||||
for (const [label, retention] of [
|
||||
["infinite", Number.MAX_SAFE_INTEGER],
|
||||
["10 min", 600_000],
|
||||
["50 ms", 50],
|
||||
["0", 0],
|
||||
] as const) {
|
||||
const { ms, heap, cached, readUs } = await run(n, retention);
|
||||
console.log(
|
||||
`${String(n).padEnd(13)} | ${label.padEnd(11)} | ${String(ms).padEnd(6)} | ${String(cached).padEnd(11)} | ${`${(heap / 1024 / 1024).toFixed(1)} MiB, ${Math.round(heap / n)} B/conv`.padEnd(21)} | ${readUs} µs`,
|
||||
);
|
||||
}
|
||||
}
|
||||
593
packages/durable/test/session-document-retention.test.ts
Normal file
593
packages/durable/test/session-document-retention.test.ts
Normal file
|
|
@ -0,0 +1,593 @@
|
|||
// The Session's document cache is bounded by use: unattached documents leave it after the retention period and load
|
||||
// from Storage again; documents with a watch or a document state never leave it.
|
||||
import type { Context } from "@earendil-works/chord";
|
||||
import { fauxAssistantMessage, fauxText, fauxToolCall, Type } from "@earendil-works/pi-ai";
|
||||
import {
|
||||
type AgentEvent,
|
||||
type ConversationId,
|
||||
defineDoc,
|
||||
defineDocFamily,
|
||||
defineTool,
|
||||
type HarnessSettings,
|
||||
MemoryStorage,
|
||||
type StorageWrite,
|
||||
watchEvents,
|
||||
} from "@earendil-works/pi-durable";
|
||||
import { describe, expect, it, onTestFinished } from "vitest";
|
||||
import { SessionImpl, setDefaultDocumentRetention } from "../src/session/session.ts";
|
||||
import { chatSetup, openChat } from "./chat-support.ts";
|
||||
import { addTool } from "./harness-support.ts";
|
||||
import { ControlledStorage, context, createConversation } from "./session-support.ts";
|
||||
import { deferred, eventually } from "./task-support.ts";
|
||||
|
||||
type Counter = { count: number };
|
||||
const CounterDoc = defineDoc<Counter>({
|
||||
kind: "test.retention.counter",
|
||||
version: 1,
|
||||
scope: "conversation",
|
||||
history: "latest",
|
||||
fork: "initial",
|
||||
initial: () => ({ count: 0 }),
|
||||
});
|
||||
const RewindableDoc = defineDoc<Counter>({
|
||||
kind: "test.retention.rewindable",
|
||||
version: 1,
|
||||
scope: "conversation",
|
||||
history: "rewindable",
|
||||
fork: "asOf",
|
||||
initial: () => ({ count: 0 }),
|
||||
});
|
||||
/** A third-party conversation family: one member per event. */
|
||||
const EventDoc = defineDocFamily<{ seq: number }, number>({
|
||||
kind: "test.retention.event",
|
||||
version: 1,
|
||||
scope: "conversation",
|
||||
history: "latest",
|
||||
fork: "initial",
|
||||
family: true,
|
||||
initial: (seq) => ({ seq }),
|
||||
});
|
||||
const SessionCounterDoc = defineDoc<Counter>({
|
||||
kind: "test.retention.session-counter",
|
||||
version: 1,
|
||||
scope: "session",
|
||||
initial: () => ({ count: 0 }),
|
||||
});
|
||||
|
||||
/** A session document whose checkpoint predicate throws, so the commit that changes it fails while it settles. */
|
||||
const Failing = defineDoc<Counter>({
|
||||
kind: "test.retention.failing",
|
||||
version: 1,
|
||||
scope: "session",
|
||||
initial: () => ({ count: 0 }),
|
||||
checkpointWhen: () => {
|
||||
throw new Error("checkpoint failed");
|
||||
},
|
||||
});
|
||||
|
||||
/** The default retention of this run, as the setup file set it. */
|
||||
const defaultForRun =
|
||||
process.env.PI_DURABLE_DOCUMENT_RETENTION_MS === undefined
|
||||
? 600_000
|
||||
: Number(process.env.PI_DURABLE_DOCUMENT_RETENTION_MS);
|
||||
|
||||
/** A conversation document any Harness test can read. */
|
||||
const TestDoc = defineDoc<Counter>({
|
||||
kind: "test.retention.harness",
|
||||
version: 1,
|
||||
scope: "conversation",
|
||||
history: "latest",
|
||||
fork: "initial",
|
||||
initial: () => ({ count: 0 }),
|
||||
});
|
||||
|
||||
/** A Session with a controllable clock and retention. */
|
||||
function open(retentionMs: number) {
|
||||
const clock = { now: 1_000 };
|
||||
const storage = new ControlledStorage();
|
||||
const session = new SessionImpl(
|
||||
storage,
|
||||
() => clock.now,
|
||||
() => retentionMs,
|
||||
);
|
||||
const sizes = () => session.documentCacheSizes;
|
||||
return { storage, session, clock, sizes };
|
||||
}
|
||||
|
||||
/** Let the line run what is queued on it: an eviction job, then return. */
|
||||
async function settleLine(session: SessionImpl): Promise<void> {
|
||||
await session.readOnLine(async () => {});
|
||||
}
|
||||
|
||||
async function bump(session: SessionImpl, conversationId: ConversationId, by = 1): Promise<void> {
|
||||
await session.commit(async (tx) => {
|
||||
(await tx.doc(CounterDoc, conversationId)).count += by;
|
||||
}, context);
|
||||
}
|
||||
|
||||
describe("document cache retention", () => {
|
||||
it("evicts an unattached document once its retention passed, and loads it again with the same value", async () => {
|
||||
const { session, storage, clock, sizes } = open(60_000);
|
||||
const conversationId = await createConversation(session);
|
||||
await bump(session, conversationId, 3);
|
||||
expect(sizes().cached).toBeGreaterThan(0);
|
||||
// Not yet: inside the retention period.
|
||||
clock.now += 59_000;
|
||||
await bump(session, conversationId, 0);
|
||||
await settleLine(session);
|
||||
expect(sizes().cached).toBeGreaterThan(0);
|
||||
// Past it, with no use since: the next commit evicts it.
|
||||
clock.now += 61_000;
|
||||
await session.commit(async () => {}, context);
|
||||
await session.commit(async (tx) => {
|
||||
await tx.doc(SessionCounterDoc);
|
||||
}, context);
|
||||
await settleLine(session);
|
||||
const reads = storage.documentReadCount;
|
||||
expect(await session.snapshot(CounterDoc, conversationId, context)).toEqual({ count: 3 });
|
||||
expect(storage.documentReadCount).toBeGreaterThan(reads);
|
||||
await session.close(context);
|
||||
});
|
||||
|
||||
it("evicts with retention 0 right after use, on the line", async () => {
|
||||
const { session, sizes } = open(0);
|
||||
const conversationId = await createConversation(session);
|
||||
await bump(session, conversationId);
|
||||
await eventually(async () => {
|
||||
await settleLine(session);
|
||||
return sizes().cached === 0;
|
||||
});
|
||||
expect(await session.snapshot(CounterDoc, conversationId, context)).toEqual({ count: 1 });
|
||||
await session.close(context);
|
||||
});
|
||||
|
||||
it("never evicts a document a watch or a document state observes, and evicts it once the last detaches", async () => {
|
||||
const { session, storage, clock, sizes } = open(1_000);
|
||||
const conversationId = await createConversation(session);
|
||||
await bump(session, conversationId);
|
||||
const watch = (await session.watchDoc(CounterDoc, conversationId, context))!;
|
||||
const state = (await session.documentState(CounterDoc, conversationId, context))!;
|
||||
const seen: number[] = [];
|
||||
watch.start(async (value) => {
|
||||
seen.push(value?.count ?? -1);
|
||||
});
|
||||
clock.now += 1_000_000;
|
||||
await bump(session, conversationId);
|
||||
await settleLine(session);
|
||||
// Still cached: a read long after its retention passed loads nothing from Storage.
|
||||
expect(state.value?.count).toBe(2);
|
||||
await eventually(() => seen.includes(2));
|
||||
const reads = storage.documentReadCount;
|
||||
clock.now += 1_000_000;
|
||||
await session.commit(async (tx) => {
|
||||
await tx.doc(SessionCounterDoc);
|
||||
}, context);
|
||||
await settleLine(session);
|
||||
expect(await session.snapshot(CounterDoc, conversationId, context)).toEqual({ count: 2 });
|
||||
expect(storage.documentReadCount).toBe(reads);
|
||||
await watch.stop();
|
||||
clock.now += 1_000_000;
|
||||
await bump(session, conversationId);
|
||||
await settleLine(session);
|
||||
expect(state.value?.count).toBe(3);
|
||||
state.dispose();
|
||||
clock.now += 1_000_000;
|
||||
await session.commit(async () => {}, context);
|
||||
await session.commit(async (tx) => {
|
||||
await tx.doc(SessionCounterDoc);
|
||||
}, context);
|
||||
await settleLine(session);
|
||||
// The session document was just used; the counter is gone.
|
||||
expect(sizes().cached).toBe(1);
|
||||
await session.close(context);
|
||||
});
|
||||
|
||||
it("never takes a tracker from a transaction that holds it", async () => {
|
||||
const { session, clock } = open(0);
|
||||
const conversationId = await createConversation(session);
|
||||
await bump(session, conversationId);
|
||||
const inside = deferred();
|
||||
const release = deferred();
|
||||
const commit = session.commit(async (tx) => {
|
||||
const draft = await tx.doc(CounterDoc, conversationId);
|
||||
inside.resolve();
|
||||
await release.promise;
|
||||
draft.count += 10;
|
||||
}, context);
|
||||
await inside.promise;
|
||||
// Retention passes while the transaction holds the document; eviction waits on the line behind it.
|
||||
clock.now += 1_000_000;
|
||||
release.resolve();
|
||||
await commit;
|
||||
expect(await session.snapshot(CounterDoc, conversationId, context)).toEqual({ count: 11 });
|
||||
await bump(session, conversationId);
|
||||
expect(await session.snapshot(CounterDoc, conversationId, context)).toEqual({ count: 12 });
|
||||
await session.close(context);
|
||||
});
|
||||
|
||||
it("writes after a reload continue the stored deltas and bases", async () => {
|
||||
const { session, storage } = open(0);
|
||||
const conversationId = await createConversation(session);
|
||||
const writes: StorageWrite[][] = [];
|
||||
const commit = storage.commit.bind(storage);
|
||||
storage.commit = (batch: readonly StorageWrite[], commitContext: Context) => {
|
||||
writes.push([...batch]);
|
||||
return commit(batch, commitContext);
|
||||
};
|
||||
for (let n = 0; n < 5; n++) {
|
||||
await bump(session, conversationId);
|
||||
await settleLine(session);
|
||||
}
|
||||
expect(await session.snapshot(CounterDoc, conversationId, context)).toEqual({ count: 5 });
|
||||
// Every bump after the first is a change of the same incarnation, never a second creation.
|
||||
const creations = writes
|
||||
.flat()
|
||||
.filter((write) => write.type === "document.create" && write.record.kind === CounterDoc.definition.kind);
|
||||
expect(creations).toHaveLength(1);
|
||||
await session.close(context);
|
||||
// And a fresh Session over the same storage reads the same value.
|
||||
const reopened = new SessionImpl(storage.reopen(), Date.now, () => 0);
|
||||
expect(await reopened.snapshot(CounterDoc, conversationId, context)).toEqual({ count: 5 });
|
||||
await reopened.close(context);
|
||||
});
|
||||
|
||||
it("keeps history and forks of rewindable documents after eviction", async () => {
|
||||
const { session } = open(0);
|
||||
const conversationId = await createConversation(session);
|
||||
const first = await session.commit(async (tx) => {
|
||||
(await tx.doc(RewindableDoc, conversationId)).count = 1;
|
||||
return (await tx.appendEntry(conversationId, { kind: "note", data: null })).id;
|
||||
}, context);
|
||||
await settleLine(session);
|
||||
const second = await session.commit(async (tx) => {
|
||||
(await tx.doc(RewindableDoc, conversationId)).count = 2;
|
||||
return (await tx.appendEntry(conversationId, { kind: "note", data: null })).id;
|
||||
}, context);
|
||||
await settleLine(session);
|
||||
expect(await session.snapshotAsOf(RewindableDoc, conversationId, first, context)).toEqual({ count: 1 });
|
||||
expect(await session.snapshotAsOf(RewindableDoc, conversationId, second, context)).toEqual({ count: 2 });
|
||||
const fork = await session.commit(
|
||||
async (tx) => (await tx.forkConversation(conversationId, first, { ownership: { kind: "ownerless" } })).id,
|
||||
context,
|
||||
);
|
||||
await settleLine(session);
|
||||
expect(await session.snapshot(RewindableDoc, fork, context)).toEqual({ count: 1 });
|
||||
expect(await session.snapshot(RewindableDoc, conversationId, context)).toEqual({ count: 2 });
|
||||
await session.close(context);
|
||||
});
|
||||
|
||||
it("bounds a third-party family with a member per event", async () => {
|
||||
const { session, sizes } = open(0);
|
||||
const conversationId = await createConversation(session);
|
||||
for (let seq = 0; seq < 200; seq++) {
|
||||
await session.commit(async (tx) => {
|
||||
await tx.doc(EventDoc, conversationId, String(seq), seq);
|
||||
}, context);
|
||||
}
|
||||
await eventually(async () => {
|
||||
await settleLine(session);
|
||||
return sizes().cached === 0;
|
||||
});
|
||||
expect(await session.snapshot(EventDoc, conversationId, "123", context)).toEqual({ seq: 123 });
|
||||
await session.close(context);
|
||||
});
|
||||
|
||||
it("keeps the cache of a long-lived Session bounded across many conversations", async () => {
|
||||
const { session, clock, sizes } = open(10_000);
|
||||
for (let n = 0; n < 300; n++) {
|
||||
const conversationId = await createConversation(session);
|
||||
await bump(session, conversationId);
|
||||
clock.now += 1_000;
|
||||
}
|
||||
await settleLine(session);
|
||||
// Only what was used in the last 10 s, about ten conversations' worth.
|
||||
expect(sizes().cached).toBeLessThanOrEqual(12);
|
||||
await session.close(context);
|
||||
});
|
||||
|
||||
it("never changes the outcome of a commit when the clock throws", async () => {
|
||||
const clock = { now: 1_000, fail: false };
|
||||
const storage = new ControlledStorage();
|
||||
const session = new SessionImpl(
|
||||
storage,
|
||||
() => {
|
||||
if (clock.fail) throw new Error("clock");
|
||||
return clock.now;
|
||||
},
|
||||
() => 0,
|
||||
);
|
||||
const conversationId = await createConversation(session);
|
||||
clock.fail = true;
|
||||
await bump(session, conversationId, 42);
|
||||
clock.fail = false;
|
||||
expect(await session.snapshot(CounterDoc, conversationId, context)).toEqual({ count: 42 });
|
||||
await session.close(context);
|
||||
});
|
||||
|
||||
it("evicts after a commit whose settlement fails", async () => {
|
||||
const { session, clock, sizes } = open(10_000);
|
||||
const conversationId = await createConversation(session);
|
||||
await bump(session, conversationId);
|
||||
// Created first: the predicate runs only for changes to an existing document.
|
||||
await session.commit((tx) => tx.doc(Failing).then(() => undefined), context);
|
||||
await settleLine(session);
|
||||
// Retention passes inside a commit whose checkpoint predicate throws while it settles.
|
||||
await expect(
|
||||
session.commit(async (tx) => {
|
||||
(await tx.doc(Failing)).count++;
|
||||
clock.now += 1_000_000;
|
||||
}, context),
|
||||
).rejects.toThrow("checkpoint failed");
|
||||
expect(sizes().cached).toBe(0);
|
||||
await session.close(context);
|
||||
});
|
||||
|
||||
it("evicts promptly once the retention is shortened", async () => {
|
||||
const retention = { ms: 600_000 };
|
||||
const storage = new ControlledStorage();
|
||||
const clock = { now: 1_000 };
|
||||
const session = new SessionImpl(
|
||||
storage,
|
||||
() => clock.now,
|
||||
() => retention.ms,
|
||||
);
|
||||
const conversationId = await createConversation(session);
|
||||
await bump(session, conversationId);
|
||||
retention.ms = 0;
|
||||
clock.now += 2_000;
|
||||
// A read, no commit: the sweep pending for ten minutes from now moves up.
|
||||
await session.snapshot(CounterDoc, conversationId, context);
|
||||
await eventually(async () => {
|
||||
await settleLine(session);
|
||||
return session.documentCacheSizes.cached === 0;
|
||||
});
|
||||
await session.close(context);
|
||||
});
|
||||
|
||||
it("does not keep a replacement cached for an observer of the retired incarnation", async () => {
|
||||
const { session, sizes } = open(0);
|
||||
const conversationId = await createConversation(session);
|
||||
await bump(session, conversationId);
|
||||
const state = (await session.documentState(CounterDoc, conversationId, context))!;
|
||||
await session.commit((tx) => tx.retireDoc(CounterDoc, conversationId), context);
|
||||
expect(state.value).toBeNull();
|
||||
await bump(session, conversationId, 5);
|
||||
await eventually(async () => {
|
||||
await settleLine(session);
|
||||
return sizes().cached === 0;
|
||||
});
|
||||
expect(await session.snapshot(CounterDoc, conversationId, context)).toEqual({ count: 5 });
|
||||
state.dispose();
|
||||
await session.close(context);
|
||||
});
|
||||
|
||||
it("leaves no pin behind when an observer's load loses a race with close", async () => {
|
||||
const { session, storage } = open(0);
|
||||
const conversationId = await createConversation(session);
|
||||
await bump(session, conversationId);
|
||||
await eventually(async () => {
|
||||
await settleLine(session);
|
||||
return session.documentCacheSizes.cached === 0;
|
||||
});
|
||||
const gate = storage.holdFindDocument();
|
||||
const attaching = session.documentState(CounterDoc, conversationId, context);
|
||||
await gate.entered;
|
||||
const closing = session.close(context);
|
||||
gate.release();
|
||||
await expect(attaching).rejects.toThrow();
|
||||
await closing;
|
||||
expect(session.documentCacheSizes).toMatchObject({ cached: 0, idle: 0, attached: 0 });
|
||||
});
|
||||
|
||||
it("evicts on its timer without a commit, and tolerates a detach called twice", async () => {
|
||||
const storage = new ControlledStorage();
|
||||
const session = new SessionImpl(storage, Date.now, () => 20);
|
||||
const conversationId = await createConversation(session);
|
||||
await bump(session, conversationId);
|
||||
const state = (await session.documentState(CounterDoc, conversationId, context))!;
|
||||
state.dispose();
|
||||
state.dispose();
|
||||
const until = Date.now() + 2_000;
|
||||
while (session.documentCacheSizes.cached > 0 && Date.now() < until)
|
||||
await new Promise((resolve) => setTimeout(resolve, 10));
|
||||
expect(session.documentCacheSizes.cached).toBe(0);
|
||||
await session.close(context);
|
||||
});
|
||||
|
||||
it("migrates a document again after its eviction", async () => {
|
||||
const V1 = defineDoc<{ count: number }>({
|
||||
kind: "test.retention.migrating",
|
||||
version: 1,
|
||||
scope: "session",
|
||||
initial: () => ({ count: 1 }),
|
||||
});
|
||||
let migrations = 0;
|
||||
const V2 = defineDoc<{ count: number; migrated: boolean }>({
|
||||
kind: "test.retention.migrating",
|
||||
version: 2,
|
||||
scope: "session",
|
||||
initial: () => ({ count: 0, migrated: true }),
|
||||
migrate: (value) => {
|
||||
migrations++;
|
||||
return { count: (value as { count: number }).count, migrated: true };
|
||||
},
|
||||
});
|
||||
const { session } = open(0);
|
||||
await session.commit((tx) => tx.doc(V1).then(() => undefined), context);
|
||||
await settleLine(session);
|
||||
expect(await session.snapshot(V2, context)).toEqual({ count: 1, migrated: true });
|
||||
await settleLine(session);
|
||||
expect(await session.snapshot(V2, context)).toEqual({ count: 1, migrated: true });
|
||||
// Not written back, so each cold load migrates again.
|
||||
expect(migrations).toBe(2);
|
||||
await session.close(context);
|
||||
});
|
||||
|
||||
it("treats a retention that is not a number of milliseconds as the default, and Infinity as never", async () => {
|
||||
// The default as shipped, also in the suite run that forces another.
|
||||
setDefaultDocumentRetention(600_000);
|
||||
onTestFinished(() => setDefaultDocumentRetention(defaultForRun));
|
||||
for (const retention of [Number.NaN, -5, "soon" as unknown as number]) {
|
||||
const { session, clock, sizes } = open(retention);
|
||||
const conversationId = await createConversation(session);
|
||||
await bump(session, conversationId);
|
||||
clock.now += 60_000;
|
||||
await session.commit(async () => {}, context);
|
||||
await settleLine(session);
|
||||
// Not dropped at once, as a retention of 0 would: the ten-minute default holds it.
|
||||
expect(sizes().cached).toBe(1);
|
||||
await session.close(context);
|
||||
}
|
||||
const { session, clock, sizes } = open(Number.POSITIVE_INFINITY);
|
||||
const conversationId = await createConversation(session);
|
||||
await bump(session, conversationId);
|
||||
clock.now += 1e15;
|
||||
await session.commit(async () => {}, context);
|
||||
await settleLine(session);
|
||||
expect(sizes().cached).toBe(1);
|
||||
await session.close(context);
|
||||
});
|
||||
|
||||
it("keeps no timer where timers cannot be unreferenced, and evicts on commits", async () => {
|
||||
const setTimer = globalThis.setTimeout;
|
||||
let timers = 0;
|
||||
globalThis.setTimeout = ((callback: () => void, ms?: number) => {
|
||||
timers++;
|
||||
const timer = setTimer(callback, ms);
|
||||
// As in Cloudflare Workers: a plain number, no `unref`.
|
||||
return Number(timer as unknown as number) as unknown as ReturnType<typeof setTimeout>;
|
||||
}) as typeof setTimeout;
|
||||
try {
|
||||
const { session, clock, sizes } = open(1_000);
|
||||
const conversationId = await createConversation(session);
|
||||
await bump(session, conversationId);
|
||||
await bump(session, conversationId);
|
||||
// Tried once, then never again.
|
||||
expect(timers).toBeLessThanOrEqual(1);
|
||||
clock.now += 10_000;
|
||||
await session.commit(async () => {}, context);
|
||||
await settleLine(session);
|
||||
expect(sizes().cached).toBe(0);
|
||||
await session.close(context);
|
||||
} finally {
|
||||
globalThis.setTimeout = setTimer;
|
||||
}
|
||||
});
|
||||
|
||||
it("follows a Harness settings change at the next use", async () => {
|
||||
setDefaultDocumentRetention(600_000);
|
||||
onTestFinished(() => setDefaultDocumentRetention(defaultForRun));
|
||||
const setup = chatSetup();
|
||||
const settings: { -readonly [K in keyof HarnessSettings]: HarnessSettings[K] } = setup.settings;
|
||||
const { harness, root } = await openChat(new MemoryStorage(), setup);
|
||||
await root.commit(async (tx) => {
|
||||
(await tx.doc(TestDoc, root.id)).count++;
|
||||
}, context);
|
||||
const sizesOf = () => (harness as unknown as SessionImpl).documentCacheSizes;
|
||||
expect(sizesOf().cached).toBeGreaterThan(0);
|
||||
settings.documentRetentionMs = 0;
|
||||
await harness.snapshot(TestDoc, root.id, context);
|
||||
await eventually(async () => {
|
||||
await (harness as unknown as SessionImpl).readOnLine(async () => {});
|
||||
return sizesOf().cached === 0;
|
||||
});
|
||||
await harness.close(context);
|
||||
});
|
||||
|
||||
it("gives the same events with documents evicted right after use as with documents kept", async () => {
|
||||
const run = async (documentRetentionMs: number): Promise<string[]> => {
|
||||
const setup = chatSetup();
|
||||
setup.settings.documentRetentionMs = documentRetentionMs;
|
||||
setup.settings.progress = { outputIntervalMs: 0 };
|
||||
addTool(
|
||||
setup.registry,
|
||||
defineTool({
|
||||
name: "steps",
|
||||
description: "Streams three steps",
|
||||
parameters: Type.Object({}),
|
||||
execute: async (_args, api) => {
|
||||
for (const step of ["a\n", "b\n", "c\n"]) {
|
||||
api.output(step);
|
||||
await new Promise((resolve) => setTimeout(resolve, 5));
|
||||
}
|
||||
return {};
|
||||
},
|
||||
}),
|
||||
);
|
||||
setup.faux.setResponses([
|
||||
fauxAssistantMessage([fauxToolCall("steps", {}, { id: "s1" })], { stopReason: "toolUse" }),
|
||||
fauxAssistantMessage([fauxText("done")]),
|
||||
]);
|
||||
const { harness, root } = await openChat(new MemoryStorage(), setup);
|
||||
const stream = await watchEvents(harness, root.id, context);
|
||||
const events: AgentEvent[] = [];
|
||||
stream.start(async (batch) => {
|
||||
events.push(...batch);
|
||||
});
|
||||
await (await root.submit({ type: "input", content: "go" }, context)).wait(context);
|
||||
await harness.waitForIdle(context);
|
||||
await stream.stop();
|
||||
await harness.close(context);
|
||||
// Ids and times differ between runs; the sequence of event types and tool outputs must not.
|
||||
return events.map((event) =>
|
||||
event.type === "tool_execution_update" ? `update:${JSON.stringify(event.output)}` : event.type,
|
||||
);
|
||||
};
|
||||
expect(await run(0)).toEqual(await run(600_000));
|
||||
});
|
||||
|
||||
it("keeps a reloaded newer version cached while an observer of the older one is attached", async () => {
|
||||
const V1 = defineDoc<Counter>({
|
||||
kind: "test.retention.versioned",
|
||||
version: 1,
|
||||
scope: "session",
|
||||
initial: () => ({ count: 3 }),
|
||||
});
|
||||
const V2 = defineDoc<Counter & { v2: true }>({
|
||||
kind: "test.retention.versioned",
|
||||
version: 2,
|
||||
scope: "session",
|
||||
initial: () => ({ count: 0, v2: true }),
|
||||
migrate: (value) => ({ count: (value as Counter).count, v2: true }),
|
||||
});
|
||||
const { session, sizes } = open(0);
|
||||
await session.commit((tx) => tx.doc(V1).then(() => undefined), context);
|
||||
const state = (await session.documentState(V1, context))!;
|
||||
expect(await session.snapshot(V2, context)).toEqual({ count: 3, v2: true });
|
||||
await settleLine(session);
|
||||
expect(sizes().cached).toBe(1);
|
||||
state.dispose();
|
||||
await eventually(async () => {
|
||||
await settleLine(session);
|
||||
return sizes().cached === 0;
|
||||
});
|
||||
await session.close(context);
|
||||
});
|
||||
|
||||
it("queues one sweep job, not one per document, while a commit holds the line", async () => {
|
||||
const { session } = open(0);
|
||||
const conversationIds: ConversationId[] = [];
|
||||
for (let n = 0; n < 50; n++) {
|
||||
const id = await createConversation(session);
|
||||
await bump(session, id);
|
||||
conversationIds.push(id);
|
||||
}
|
||||
await eventually(async () => {
|
||||
await settleLine(session);
|
||||
return session.documentCacheSizes.cached === 0;
|
||||
});
|
||||
const before = session.documentCacheSizes.sweeps;
|
||||
// One commit loads 50 evicted documents from storage, each touched as it loads; the sweep job stays pending while
|
||||
// the line is held.
|
||||
await session.commit(async (tx) => {
|
||||
for (const id of conversationIds) {
|
||||
(await tx.doc(CounterDoc, id)).count++;
|
||||
await new Promise((resolve) => setTimeout(resolve, 0));
|
||||
}
|
||||
}, context);
|
||||
await settleLine(session);
|
||||
expect(session.documentCacheSizes.cached).toBe(0);
|
||||
expect(session.documentCacheSizes.sweeps - before).toBeLessThanOrEqual(2);
|
||||
await session.close(context);
|
||||
});
|
||||
});
|
||||
|
|
@ -125,7 +125,8 @@ export function openTestSession(): {
|
|||
readonly publications: CommitPublication[];
|
||||
} {
|
||||
const storage = new ControlledStorage();
|
||||
const session = new SessionImpl(storage);
|
||||
// These tests pin the cache's identities and loads, so documents stay as long as the default keeps them.
|
||||
const session = new SessionImpl(storage, Date.now, () => 600_000);
|
||||
const publications: CommitPublication[] = [];
|
||||
session.subscribeCommits((publication) => {
|
||||
publications.push(publication);
|
||||
|
|
|
|||
7
packages/durable/test/setup-document-retention.ts
Normal file
7
packages/durable/test/setup-document-retention.ts
Normal file
|
|
@ -0,0 +1,7 @@
|
|||
// Loaded by vitest for every test file: `PI_DURABLE_DOCUMENT_RETENTION_MS` sets the document retention of every Session
|
||||
// that keeps the default, so the suite can run with each unattached document evicted right after use
|
||||
// (`npm run test:evicting`).
|
||||
import { setDefaultDocumentRetention } from "../src/session/session.ts";
|
||||
|
||||
const forced = process.env.PI_DURABLE_DOCUMENT_RETENTION_MS;
|
||||
if (forced !== undefined) setDefaultDocumentRetention(Number(forced));
|
||||
|
|
@ -7,6 +7,7 @@ const durableSrcTesting = fileURLToPath(new URL("./src/testing/index.ts", import
|
|||
export default defineConfig({
|
||||
test: {
|
||||
environment: "node",
|
||||
setupFiles: ["./test/setup-document-retention.ts"],
|
||||
},
|
||||
resolve: {
|
||||
conditions: ["source"],
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue