feat(durable): workerd tests for Durable Object hosting

Runs the storage conformance suite, the schema migration, an alarm-woken sleep across an eviction, and a
turn recovered through the recovery alarm in workerd via @cloudflare/vitest-plugin (npm run test:workers).

fix(durable): decline background timers in Cloudflare Workers. With nodejs_compat, timers have an unref(),
so the idle-memory timers were armed and a pending one kept a Durable Object from hibernating or being evicted.
This commit is contained in:
Mario Zechner 2026-10-17 23:02:48 +00:00
parent f75b3240ab
commit ea7c648430
15 changed files with 1439 additions and 13 deletions

View file

@ -12,6 +12,7 @@
- `settings.compaction.modelOverrides` sets compaction thresholds per model, keyed by exact `"provider/modelId"`, so one harness can serve small- and large-window models; the threshold check, overflow recovery, and the compaction task use the entry of the model they act for ([#10942](https://github.com/earendil-works/pi/issues/10942)).
- Durable sleeps: a sleeping task holds no invocation, survives restarts, and resumes at its checkpoint once the Harness clock reaches `until`. Generation retry backoffs, deferred-response polls, and compaction retry backoffs sleep this way; a record that slept inside a running phase sleeps again when it resumes early.
- `HarnessOptions.setTimer` arms every timer of the Session and Harness (progress throttles, idle-memory eviction, sleeps), and `ToolExecutionApi.now()` reads the Harness clock, which code mode now uses for its deadlines; with `now`, a Harness runs entirely on a host or fake clock.
- Tests that run in workerd (`npm run test:workers`): the storage conformance suite and the schema migration on real Durable Object SQLite, a sleep resumed by the object's alarm after an eviction, and a turn recovered through the recovery alarm after an eviction mid-stream.
- `Harness.wake()` and `HarnessOptions.onWake` tell a host when the Session next needs a process: `active` while work can run, `at` for the earliest sleep. With `onWake`, the Harness keeps no timer for sleeps, so a Cloudflare Durable Object can hibernate and wake on its alarm, calling `resume()`.
- `HarnessOptions.models` and `HarnessOptions.settings` also accept a function returning the current value, like `env`. Generation, compaction, hooks and tools call it at each use (`api.models` reads the current value at each access), so a host can replace its model configuration or settings without reopening the Harness. A plain object still works.
@ -23,6 +24,7 @@
### Fixed
- In Cloudflare Workers with `nodejs_compat`, the timers that evict idle documents and contexts were armed, since timers there have an `unref()`; a pending one kept a Durable Object from hibernating and blocked its eviction. They are now declined in Workers, and memory is freed at later commits.
- The README's Errors section now limits the `StorageRequestError` exemption to reads: a `StorageRequestError` thrown by `commit()` or `mintId()` fails the Harness like any other storage error ([#11014](https://github.com/earendil-works/pi/issues/11014)).
## [1.3.0] - 2026-10-15

View file

@ -187,7 +187,7 @@ export class Session extends DurableObject<Env> {
}
```
The default `setTimer` declines the timers that only free memory where timers cannot be unreferenced, as in Workers, so no pending timer keeps the object from hibernating.
The default `setTimer` declines the timers that only free memory in Workers, so no pending timer keeps the object from hibernating or being evicted; memory is then freed at later commits.
## Extensions
@ -383,7 +383,7 @@ const harness = await Harness.open(storage, {
compaction: { reserveTokens: 16384 },
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)
contextRetentionMs: 600_000, // idle conversations keep their context in memory this long (checked lazily where background timers are declined, as in Workers)
documentRetentionMs: 600_000, // loaded documents nothing observes stay in memory this long after their last use; observed ones always
get followUpMode() {
return userSettings.followUpMode;

View file

@ -379,7 +379,7 @@ type HarnessOptions<Tool extends ToolRegistration = ToolRegistration> = {
/**
* Arms every timer of the Session and Harness (section 5.6). `background` timers only free memory and may be
* declined by returning `undefined`; the others pace or resume work. Default `setTimeout`, background timers
* unreferenced, or declined where they cannot be.
* unreferenced, or declined in Cloudflare Workers and where they cannot be unreferenced.
*/
readonly setTimer?: (
callback: () => void,
@ -2355,9 +2355,10 @@ poll, sleeps again with the same checkpoint (section 8.3).
`HarnessOptions.setTimer` (default `setTimeout`). Background timers only free
memory: document eviction (section 4) and kept contexts (section 5.1). A host
may decline one by returning `undefined`; that memory is then freed at later
commits and task changes. The default declines them where timers cannot be
unreferenced, as in Cloudflare Workers, since a pending timer keeps a Durable
Object from hibernating. Foreground timers pace or resume work: throttled
commits and task changes. The default declines them in Cloudflare Workers, since
a pending timer keeps a Durable Object from hibernating or being evicted even
when `nodejs_compat` gives it an `unref()`, and wherever timers cannot be
unreferenced. Foreground timers pace or resume work: throttled
progress commits through `TaskRuntime.setTimer()`, and, without `onWake`, the
timer for the earliest sleep while scheduling runs. That one keeps a Node.js
process alive, since a sleeping task is work left to do. A timer may fire early

View file

@ -113,7 +113,8 @@
"scripts": {
"clean": "shx rm -rf dist",
"build": "tsc -p tsconfig.build.json",
"test": "vitest --run && PI_DURABLE_DOCUMENT_RETENTION_MS=0 vitest --run",
"test": "vitest --run && PI_DURABLE_DOCUMENT_RETENTION_MS=0 vitest --run && npm run test:workers",
"test:workers": "vitest --run --config test-workers/vitest.config.ts",
"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",
@ -151,6 +152,7 @@
"typebox": "1.3.27"
},
"devDependencies": {
"@cloudflare/vitest-plugin": "1.4.4",
"shx": "0.3.4",
"vitest": "4.1.11"
}

View file

@ -611,8 +611,8 @@ export type HarnessOptions<Tool extends ToolRegistration = ToolRegistration> = {
/**
* Arms every timer the Harness uses: pacing progress commits, freeing idle memory (`background`), and, without
* `onWake`, resuming sleeping tasks; and the timers of `TaskRuntime.setTimer()`. Default `setTimeout`, with background
* timers unreferenced, or declined where they cannot be. Tests can pass fake timers; a host can keep timers off a
* process that should hibernate.
* timers unreferenced, or declined in Cloudflare Workers and where they cannot be unreferenced. Tests can pass fake
* timers; a host can keep timers off a process that should hibernate.
*/
readonly setTimer?: SetTimer;
/**

View file

@ -3,12 +3,18 @@ import type { SetTimer } from "./types.ts";
/** Longest delay `setTimeout` supports; a longer one fires early, and its caller arms again. */
const MAX_TIMER_DELAY = 2_147_483_647;
/**
* Cloudflare Workers: a pending timer keeps a Durable Object from hibernating or being evicted, even an unreferenced
* one (`nodejs_compat` gives timers an `unref()` that changes nothing).
*/
const inWorkers = typeof navigator !== "undefined" && navigator.userAgent === "Cloudflare-Workers";
/**
* The default `HarnessOptions.setTimer`: `setTimeout`. A background timer is unreferenced, so it never keeps the
* process alive; where timers cannot be unreferenced, as in Cloudflare Workers, it is declined, since a pending timer
* could keep a Durable Object from hibernating.
* process alive. It is declined in Cloudflare Workers and wherever timers cannot be unreferenced.
*/
export const nativeTimer: SetTimer = (callback, delayMs, options) => {
if (options.background && inWorkers) return undefined;
const handle = setTimeout(callback, Math.min(Math.max(0, delayMs), MAX_TIMER_DELAY));
if (options.background) {
const unref = (handle as { unref?: unknown }).unref;

View file

@ -0,0 +1,40 @@
// The parts of the Workers runtime these tests use, typed for the repository's Node program. The tests run in workerd
// itself (`vitest.config.ts`); full Workers types would conflict with Node's. `DurableObject` itself is declared by
// packages/codemode/src/cloudflare-modules.d.ts, with an untyped `ctx`.
declare module "cloudflare:workers" {
import type { DurableObjectSqliteStorage } from "@earendil-works/chord/sqlite/cloudflare";
export type DurableObjectStorage = DurableObjectSqliteStorage & {
getAlarm(): Promise<number | null>;
setAlarm(scheduledTime: number): Promise<void>;
deleteAlarm(): Promise<void>;
/** Synchronous key-value storage of a SQLite-backed Durable Object. */
readonly kv: { get<T>(key: string): T | undefined; put<T>(key: string, value: T): void };
};
export type DurableObjectState = {
readonly storage: DurableObjectStorage;
blockConcurrencyWhile<T>(callback: () => Promise<T>): Promise<T>;
};
export type DurableObjectId = { toString(): string };
export type DurableObjectStub = { readonly id: DurableObjectId };
export type DurableObjectNamespace = { newUniqueId(): DurableObjectId; get(id: DurableObjectId): DurableObjectStub };
export const env: { readonly SESSIONS: DurableObjectNamespace };
}
declare module "cloudflare:test" {
import type { DurableObjectState, DurableObjectStub } from "cloudflare:workers";
/** Run `callback` inside the object, with its instance and state. */
export function runInDurableObject<O, R>(
stub: DurableObjectStub,
callback: (instance: O, state: DurableObjectState) => R | Promise<R>,
): Promise<R>;
/** Run the object's alarm now if one is set; whether one ran. */
export function runDurableObjectAlarm(stub: DurableObjectStub): Promise<boolean>;
/** Tear the object's instance down, keeping its storage. */
export function evictDurableObject(stub: DurableObjectStub): Promise<void>;
}

View file

@ -0,0 +1,54 @@
import { runInDurableObject } from "cloudflare:test";
import { env } from "cloudflare:workers";
import { DurableObjectSqliteDatabase } from "@earendil-works/chord/sqlite/cloudflare";
import { registerStorageConformance } from "@earendil-works/pi-durable/testing";
import { describe, expect, it } from "vitest";
import { openDurableObjectSqliteStorage } from "../src/storage/sqlite/cloudflare.ts";
import { applySqliteMigrationsForSchema, createSqliteMigrations } from "../src/storage/sqlite/migrations.ts";
import { createSqliteSchema } from "../src/storage/sqlite/schema.ts";
const fresh = () => env.SESSIONS.get(env.SESSIONS.newUniqueId());
let suite = 0;
registerStorageConformance({ describe, expect, it }, "SqliteStorage on Durable Object SQLite in workerd", (use) =>
runInDurableObject(fresh(), async (_instance, state) => {
// Tables of their own, next to the ones the object's Harness opened.
await use(await openDurableObjectSqliteStorage(state.storage, { prefix: `conformance${suite++}_` }));
}),
);
describe("Durable Object SQLite schema", () => {
it("migrates a version 1 schema to admit sleeping tasks, keeping tasks and indexes", async () => {
await runInDurableObject(fresh(), async (_instance, state) => {
const database = new DurableObjectSqliteDatabase(state.storage);
const schema = createSqliteSchema("legacy_");
const tasks = schema.tables.tasks;
const insert = (id: number, status: string) =>
database.run(
`INSERT INTO ${tasks} (id, conversation_id, kind, status, abort_requested, background, record) VALUES (?, 1, 'test', ?, 0, 0, '{}')`,
id,
status,
);
// A database written by a version that knew schema version 1 only.
await applySqliteMigrationsForSchema(database, schema, createSqliteMigrations(schema).slice(0, 1));
await insert(5, "waiting");
await expect(insert(6, "sleeping")).rejects.toThrow();
await openDurableObjectSqliteStorage(state.storage, { prefix: "legacy_" });
expect(await database.get(`SELECT version FROM ${schema.tables.durableSchema}`)).toEqual({ version: 2 });
expect(await database.get(`SELECT status FROM ${tasks} WHERE id = 5`)).toEqual({ status: "waiting" });
await insert(6, "sleeping");
const indexes = await database.all<{ name: string }>(
`SELECT name FROM sqlite_master WHERE type = 'index' AND tbl_name = ? ORDER BY name`,
tasks,
);
expect(indexes.map(({ name }) => name)).toEqual([
schema.indexes.tasksByAbortRequested,
schema.indexes.tasksByBackground,
schema.indexes.tasksByConversation,
schema.indexes.tasksByKind,
schema.indexes.tasksByStatus,
]);
});
});
});

View file

@ -0,0 +1,24 @@
import { fileURLToPath } from "node:url";
import { cloudflareTest } from "@cloudflare/vitest-plugin";
import { defineConfig } from "vitest/config";
const source = (path: string) => fileURLToPath(new URL(path, import.meta.url));
// Runs in workerd: the Durable Object storage, alarms, and eviction are Cloudflare's own.
export default defineConfig({
root: source("."),
plugins: [cloudflareTest({ wrangler: { configPath: source("./wrangler.jsonc") } })],
test: { include: ["*.test.ts"] },
resolve: {
conditions: ["source"],
alias: [
{ find: /^@earendil-works\/pi-durable$/, replacement: source("../src/index.ts") },
{ find: /^@earendil-works\/pi-durable\/testing$/, replacement: source("../src/testing/index.ts") },
{
find: /^@earendil-works\/pi-durable\/storage\/sqlite\/cloudflare$/,
replacement: source("../src/storage/sqlite/cloudflare.ts"),
},
],
},
ssr: { resolve: { conditions: ["source"] } },
});

View file

@ -0,0 +1,96 @@
import { evictDurableObject, runDurableObjectAlarm, runInDurableObject } from "cloudflare:test";
import { type DurableObjectState, type DurableObjectStub, env } from "cloudflare:workers";
import { BACKGROUND_CONTEXT } from "@earendil-works/chord/context";
import { AssistantEntry, type TaskId } from "@earendil-works/pi-durable";
import { describe, expect, it } from "vitest";
import { nativeTimer } from "../src/timers.ts";
import type { TestSession } from "./worker.ts";
const context = BACKGROUND_CONTEXT;
const fresh = () => env.SESSIONS.get(env.SESSIONS.newUniqueId());
/** Poll `check` inside the object until it holds. */
async function eventually(
stub: DurableObjectStub,
check: (instance: TestSession, state: DurableObjectState) => Promise<boolean>,
): Promise<void> {
for (let attempt = 0; attempt < 200; attempt++) {
if (await runInDurableObject(stub, check)) return;
await new Promise((resolve) => setTimeout(resolve, 10));
}
throw new Error("Condition not reached");
}
describe("a Harness woken by its Durable Object alarm", () => {
it("sleeps through an eviction and resumes the task from its alarm", async () => {
const stub = fresh();
const sleep = 600_000;
const { id, at } = await runInDurableObject(stub, async (instance: TestSession) => {
const at = instance.now() + sleep;
return { id: await instance.remind(at), at };
});
await eventually(stub, async (instance, state) => {
const record = await instance.harness.getTask(id, context);
return record?.state.status === "sleeping" && (await state.storage.getAlarm()) === at;
});
await runInDurableObject(stub, async (instance: TestSession) => {
expect(instance.harness.wake()).toEqual({ active: false, at });
expect(instance.wakes.at(-1)).toEqual({ active: false, at });
});
await evictDurableObject(stub);
// A new instance: the sleep is read from storage, and the alarm stays.
await runInDurableObject(stub, async (instance: TestSession, state) => {
expect(instance.harness.wake()).toEqual({ active: false, at });
expect(await state.storage.getAlarm()).toBe(at);
// Time passes while the object sleeps.
instance.advance(sleep);
});
await evictDurableObject(stub);
expect(await runDurableObjectAlarm(stub)).toBe(true);
const woke = await runInDurableObject(stub, async (instance: TestSession) => {
const settled = await instance.harness.waitForTask(id as TaskId<number>, context);
return settled.state.outcome;
});
expect(woke).toMatchObject({ status: "completed" });
expect(woke.status === "completed" && woke.result).toBeGreaterThanOrEqual(at);
// Nothing is left to do: the alarm is gone.
await eventually(stub, async (_instance, state) => (await state.storage.getAlarm()) === null);
});
it("recovers a turn an eviction cut off mid-stream through the heartbeat alarm", async () => {
const stub = fresh();
await runInDurableObject(stub, async (instance: TestSession, state) => {
instance.script(["hang", "recovered"]);
await instance.submit("hi", "turn-1");
await instance.hanging.promise;
expect(instance.harness.wake().active).toBe(true);
// The heartbeat alarm armed before the input was admitted.
expect(await state.storage.getAlarm()).not.toBeNull();
});
await evictDurableObject(stub);
expect(await runDurableObjectAlarm(stub)).toBe(true);
const answer = await runInDurableObject(stub, async (instance: TestSession) => {
const root = await instance.root();
// The same request ID finds the submission admitted before the eviction.
const submission = await root.submit({ type: "input", content: "hi", requestId: "turn-1" }, context);
const settled = await submission.wait(context);
if (settled.status !== "done" || settled.type !== "input") throw new Error(`Unexpected ${settled.status}`);
const entry = await root.commit((tx) => tx.entry(AssistantEntry, settled.answer), context);
const message = entry?.model?.[0];
return message?.role === "assistant" ? message.content : undefined;
});
expect(answer).toEqual([{ type: "text", text: "recovered" }]);
await eventually(stub, async (_instance, state) => (await state.storage.getAlarm()) === null);
});
it("declines background timers by default, so no pending timer keeps an object from hibernating", () => {
expect(nativeTimer(() => {}, 10, { background: true })).toBeUndefined();
const disarm = nativeTimer(() => {}, 10, { background: false });
expect(typeof disarm).toBe("function");
disarm?.();
});
});

View file

@ -0,0 +1,148 @@
// A Durable Object that hosts a Harness the way an application would: opened in the constructor, woken by its alarm
// through `onWake`. Tests drive it with `runInDurableObject()`, evict it, and run its alarm.
import { DurableObject, type DurableObjectState } from "cloudflare:workers";
import { BACKGROUND_CONTEXT } from "@earendil-works/chord/context";
import { createModels, type FauxResponseStep, fauxAssistantMessage, fauxProvider } from "@earendil-works/pi-ai";
import {
type Conversation,
createRegistry,
defineExtension,
defineTask,
Harness,
type TaskId,
type Wake,
} from "@earendil-works/pi-durable";
import { openDurableObjectSqliteStorage } from "@earendil-works/pi-durable/storage/sqlite/cloudflare";
const context = BACKGROUND_CONTEXT;
/** Sleeps until `at`, then completes with the Harness time it woke at. */
export const Reminder = defineTask<{ at: number }, { phase: "wait" } | { phase: "deliver" }, number>({
name: "test.reminder",
version: 1,
initial: () => ({ phase: "wait" }),
phases: {
wait: (task, runtime, ctx) => runtime.sleepUntil(task.input.at, { phase: "deliver" }, ctx),
deliver: async (_task, runtime, ctx) => {
const woke = runtime.now();
await runtime.commit(() => ({ status: "terminal", outcome: { status: "completed", result: woke } }), ctx);
},
},
abort: (_task, runtime, ctx) => runtime.commit(() => ({ status: "terminal", outcome: { status: "aborted" } }), ctx),
});
/** Scripted model responses, kept in storage so a new instance continues the script: `hang` waits for an abort. */
type Script = { responses: string[]; next: number };
export const HEARTBEAT_MS = 60_000;
export class TestSession extends DurableObject {
readonly #state: DurableObjectState;
#harness!: Harness;
/** Added to `Date.now()`, kept in storage, so a test can move the Harness clock across evictions. */
#offset = 0;
/** The wakes `onWake` received, in order. */
readonly wakes: Wake[] = [];
/** Resolves when this instance's model request starts and hangs. */
readonly hanging = Promise.withResolvers<void>();
constructor(ctx: DurableObjectState, env: unknown) {
super(ctx, env);
this.#state = ctx;
void ctx.blockConcurrencyWhile(async () => {
this.#offset = ctx.storage.kv.get<number>("offset") ?? 0;
const faux = fauxProvider();
faux.setResponses(Array.from({ length: 8 }, () => this.#step()));
const models = createModels();
models.setProvider(faux.provider);
const registry = createRegistry();
registry.install(defineExtension({ name: "test", tasks: [Reminder] }));
this.#harness = await Harness.open(
await openDurableObjectSqliteStorage(ctx.storage),
{
models,
registry,
now: () => this.now(),
onWake: (wake) => {
this.wakes.push(wake);
void this.arm(wake);
},
},
context,
);
this.#harness.resume();
});
}
get harness(): Harness {
return this.#harness;
}
now(): number {
return Date.now() + this.#offset;
}
/** Move the Harness clock forward, for this instance and the ones after it. */
advance(ms: number): void {
this.#offset += ms;
this.#state.storage.kv.put("offset", this.#offset);
}
script(responses: string[]): void {
this.#state.storage.kv.put("script", { responses, next: 0 } satisfies Script);
}
async alarm(): Promise<void> {
this.#harness.resume();
await this.arm(this.#harness.wake());
}
/** The alarm a host arms for `wake`: the earliest sleep, or a recovery heartbeat while work can run. */
async arm(wake: Wake): Promise<void> {
const heartbeat = wake.active ? Date.now() + HEARTBEAT_MS : undefined;
const sleep = wake.at === undefined ? undefined : wake.at - this.#offset;
const at = heartbeat === undefined || sleep === undefined ? (heartbeat ?? sleep) : Math.min(heartbeat, sleep);
if (at === undefined) await this.#state.storage.deleteAlarm();
else await this.#state.storage.setAlarm(at);
}
async root(): Promise<Conversation> {
return this.#harness.root(context, { agent: { model: { provider: "faux", modelId: "faux-1" } } });
}
async submit(text: string, requestId: string): Promise<void> {
// Arm before admitting work: a process that dies after the commit is still woken.
await this.arm({ ...this.#harness.wake(), active: true });
await (await this.root()).submit({ type: "input", content: text, requestId }, context);
}
async remind(at: number): Promise<TaskId<number>> {
const root = await this.root();
const id = await root.commit(
(tx) => tx.createTask(Reminder, { at }, { ownership: { kind: "conversation" } }),
context,
);
this.#harness.resume();
return id;
}
/** The next scripted response; the script position is stored before the response is used. */
#step(): FauxResponseStep {
return (_request, options) => {
const script = this.#state.storage.kv.get<Script>("script") ?? { responses: [], next: 0 };
const text = script.responses[script.next] ?? "unscripted";
this.#state.storage.kv.put("script", { ...script, next: script.next + 1 });
if (text !== "hang") return fauxAssistantMessage(text);
this.hanging.resolve();
return new Promise((_resolve, reject) => {
const abort = () => reject(new DOMException("Request interrupted", "AbortError"));
if (options?.signal?.aborted) abort();
else options?.signal?.addEventListener("abort", abort, { once: true });
});
};
}
}
export default {
fetch: () => new Response("pi-durable test worker"),
};

View file

@ -0,0 +1,10 @@
{
"name": "pi-durable-test-workers",
"main": "./worker.ts",
"compatibility_date": "2026-10-01",
"compatibility_flags": ["nodejs_compat"],
"durable_objects": {
"bindings": [{ "name": "SESSIONS", "class_name": "TestSession" }]
},
"migrations": [{ "tag": "v1", "new_sqlite_classes": ["TestSession"] }]
}

View file

@ -1,5 +1,5 @@
import { fileURLToPath } from "node:url";
import { defineConfig } from "vitest/config";
import { configDefaults, defineConfig } from "vitest/config";
const durableSrcIndex = fileURLToPath(new URL("./src/index.ts", import.meta.url));
const durableSrcTesting = fileURLToPath(new URL("./src/testing/index.ts", import.meta.url));
@ -7,6 +7,8 @@ const durableSrcTesting = fileURLToPath(new URL("./src/testing/index.ts", import
export default defineConfig({
test: {
environment: "node",
// Run in workerd by their own config: `npm run test:workers`.
exclude: [...configDefaults.exclude, "test-workers/**"],
setupFiles: ["./test/setup-document-retention.ts"],
},
resolve: {