From f75b3240abc354bd1a666748e9101220554f5be2 Mon Sep 17 00:00:00 2001 From: Mario Zechner Date: Sat, 17 Oct 2026 22:29:26 +0000 Subject: [PATCH] feat(durable): durable sleeping state replaces runtime.sleep Tasks end a phase asleep with runtime.sleepUntil() or { status: "sleeping", until, checkpoint }: no invocation is held, the deadline survives restarts, and the task resumes at its checkpoint once due. Generation retry/poll and compaction retry sleep this way. Harness.wake() and HarnessOptions.onWake let a host such as a Durable Object wake the process itself. HarnessOptions.setTimer arms every Harness timer, and chord Progress paces on an injectable clock. SQLite migrates to schema version 2 to admit the sleeping status. --- packages/chord/CHANGELOG.md | 1 + packages/chord/src/output.ts | 36 +- packages/chord/test/output.test.ts | 40 +++ packages/durable/CHANGELOG.md | 8 + packages/durable/README.md | 64 ++++ packages/durable/docs/spec.md | 178 ++++++++-- packages/durable/src/codemode/codemode.ts | 8 +- packages/durable/src/harness/compaction.ts | 6 +- packages/durable/src/harness/generation.ts | 20 +- packages/durable/src/harness/harness.ts | 11 +- packages/durable/src/harness/scheduler.ts | 247 +++++++++---- packages/durable/src/harness/task-graph.ts | 6 +- packages/durable/src/harness/tool.ts | 12 + packages/durable/src/harness/types.ts | 39 +++ packages/durable/src/index.ts | 2 + packages/durable/src/session/session.ts | 44 ++- packages/durable/src/storage/memory.ts | 2 +- .../durable/src/storage/sqlite/migrations.ts | 27 +- .../src/testing/storage-conformance.ts | 19 + packages/durable/src/timers.ts | 55 +++ packages/durable/src/types.ts | 50 ++- packages/durable/test/examples/13-recovery.ts | 23 +- .../durable/test/examples/24-child-tasks.ts | 11 +- .../test/harness-generation-recovery.test.ts | 64 +++- packages/durable/test/harness-sleep.test.ts | 330 ++++++++++++++++++ packages/durable/test/harness-tasks.test.ts | 54 +-- .../durable/test/sqlite-cloudflare.test.ts | 2 +- .../durable/test/sqlite-migrations.test.ts | 55 +++ packages/durable/test/task-support.ts | 11 +- 29 files changed, 1209 insertions(+), 216 deletions(-) create mode 100644 packages/durable/src/timers.ts create mode 100644 packages/durable/test/harness-sleep.test.ts diff --git a/packages/chord/CHANGELOG.md b/packages/chord/CHANGELOG.md index 49089b5ca..e9d1c4165 100644 --- a/packages/chord/CHANGELOG.md +++ b/packages/chord/CHANGELOG.md @@ -5,6 +5,7 @@ ### Added - `@earendil-works/chord/sqlite`: portable SQLite interfaces, atomic migration runner, and schema-prefix validation, with Node (`/sqlite/node`) and Cloudflare Durable Object (`/sqlite/cloudflare`) drivers. Moved from Durable, which keeps its storage implementations and re-exports the moved public names. +- `@earendil-works/chord/output`: `Progress` takes an optional `ProgressClock` (`now`, `setTimer`) to pace commits on, by default `Date.now` and `setTimeout`. ## [1.3.0] - 2026-10-15 diff --git a/packages/chord/src/output.ts b/packages/chord/src/output.ts index 54d1f2c71..80804c33e 100644 --- a/packages/chord/src/output.ts +++ b/packages/chord/src/output.ts @@ -302,26 +302,48 @@ function countNewlines(text: string): number { /** Each progress commit also buys a pause proportional to what it wrote. */ export const PROGRESS_BYTES_PER_SECOND = 100 * 1024; +/** The clock and timer progress commits pace on. */ +export type ProgressClock = { + readonly now: () => number; + /** Call `callback` once after about `delayMs`; returns a function that disarms it. */ + readonly setTimer: (callback: () => void, delayMs: number) => () => void; +}; + +const SYSTEM_CLOCK: ProgressClock = { + now: () => Date.now(), + setTimer: (callback, delayMs) => { + const timer = setTimeout(callback, delayMs); + return () => clearTimeout(timer); + }, +}; + /** * Adaptive progress commits: the first change after an idle period commits at once; each commit then delays the next * by at least `minIntervalMs` and by its written size at 100 KiB/s. At most one commit is in flight; changes made - * meanwhile coalesce into the next one. + * meanwhile coalesce into the next one. Paces on `clock`, by default `Date.now` and `setTimeout`. */ export class Progress { readonly #write: () => Promise; readonly #onError: (error: unknown) => void; readonly #minIntervalMs: number; + readonly #clock: ProgressClock; #waiters: PromiseWithResolvers[] = []; - #timer: ReturnType | undefined; + #timer: (() => void) | undefined; #inFlight: Promise | undefined; #nextAt = 0; #dirty = false; #stopped = false; - constructor(write: () => Promise, onError: (error: unknown) => void, minIntervalMs: number) { + constructor( + write: () => Promise, + onError: (error: unknown) => void, + minIntervalMs: number, + clock: ProgressClock = SYSTEM_CLOCK, + ) { this.#write = write; this.#onError = onError; this.#minIntervalMs = minIntervalMs; + this.#clock = clock; } /** Schedule a commit. */ @@ -341,7 +363,7 @@ export class Progress { /** Stop committing and wait for the commit in flight; returns the waiters the final commit must settle. */ async stop(): Promise[]> { this.#stopped = true; - clearTimeout(this.#timer); + this.#timer?.(); this.#timer = undefined; await this.#inFlight; return this.#waiters.splice(0); @@ -349,10 +371,10 @@ export class Progress { #schedule(): void { if (this.#stopped || this.#timer !== undefined || this.#inFlight !== undefined) return; - const wait = this.#nextAt - Date.now(); + const wait = this.#nextAt - this.#clock.now(); if (wait <= 0) this.#flush(); else - this.#timer = setTimeout(() => { + this.#timer = this.#clock.setTimer(() => { this.#timer = undefined; this.#flush(); }, wait); @@ -362,7 +384,7 @@ export class Progress { if (this.#stopped || !this.#dirty) return; this.#dirty = false; const waiters = this.#waiters.splice(0); - const started = Date.now(); + const started = this.#clock.now(); this.#inFlight = this.#write() .then( (bytes) => { diff --git a/packages/chord/test/output.test.ts b/packages/chord/test/output.test.ts index 3109164fd..ae598c026 100644 --- a/packages/chord/test/output.test.ts +++ b/packages/chord/test/output.test.ts @@ -270,6 +270,46 @@ describe("Progress", () => { expect(commits).toEqual([0, 500]); }); + it("paces on the clock it is given", async () => { + let now = 1_000; + const timers: { readonly delayMs: number; readonly callback: () => void; disarmed: boolean }[] = []; + const commits: number[] = []; + const progress = new Progress( + async () => { + commits.push(now); + return 10; + }, + () => {}, + 100, + { + now: () => now, + setTimer: (callback, delayMs) => { + const timer = { delayMs, callback, disarmed: false }; + timers.push(timer); + return () => { + timer.disarmed = true; + }; + }, + }, + ); + await progress.markAndWait(); + expect(commits).toEqual([1_000]); + // Let the first commit finish settling. + await new Promise((resolve) => setTimeout(resolve, 0)); + now = 1_040; + const second = progress.markAndWait(); + expect(timers.map(({ delayMs }) => delayMs)).toEqual([60]); + now = 1_100; + timers[0]!.callback(); + await second; + expect(commits).toEqual([1_000, 1_100]); + await new Promise((resolve) => setTimeout(resolve, 0)); + progress.mark(); + expect(timers.map(({ delayMs }) => delayMs)).toEqual([60, 100]); + await progress.stop(); + expect(timers[1]!.disarmed).toBe(true); + }); + it("rejects the waiters of a failed commit and reports its error", async () => { const errors: unknown[] = []; const failure = new Error("commit failed"); diff --git a/packages/durable/CHANGELOG.md b/packages/durable/CHANGELOG.md index e587541e6..20c287bf8 100644 --- a/packages/durable/CHANGELOG.md +++ b/packages/durable/CHANGELOG.md @@ -2,9 +2,17 @@ ## [Unreleased] +### Breaking Changes + +- `TaskRuntime.sleep()` is removed: it waited inside a phase, which holds an invocation and, after a restart, reruns the phase and restarts the wait. A task now ends its phase asleep with `runtime.sleepUntil(until, checkpoint, context)` or by returning `{ status: "sleeping", until, checkpoint }` from `runtime.commit()`; move the work after the wait into the next phase and carry what it needs in the checkpoint. `TaskRuntime.setTimer()` paces work inside a phase without waiting for it. +- Tasks have a `sleeping` status, so code that handles every `TaskState`, `TaskGraphState`, or `TaskInspection` state needs a case for it. SQLite storage migrates to schema version 2 when opened, rebuilding its task table; older versions refuse a migrated database. + ### Added - `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. +- `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. ### Changed diff --git a/packages/durable/README.md b/packages/durable/README.md index 6e20295a8..797930eed 100644 --- a/packages/durable/README.md +++ b/packages/durable/README.md @@ -125,6 +125,70 @@ const again = await root.submit({ type: "input", content: "Hello", requestId: "g `harness.submission(id)` reacquires a submission by ID, for example to wait for it after a restart. +### Sleeping and Waking Up + +A task waits for time by ending its phase asleep. The sleep is stored, so the task holds no invocation meanwhile, survives restarts, and resumes at the checkpoint once the Harness clock reaches `until`: + +```typescript +const Reminder = defineTask<{ text: string; at: number }, { phase: "wait" } | { phase: "deliver" }, null>({ + name: "app.reminder", + version: 1, + initial: () => ({ phase: "wait" }), + phases: { + // Resolves once the sleep is stored; the phase ends there. + wait: (task, runtime, context) => runtime.sleepUntil(task.input.at, { phase: "deliver" }, context), + deliver: async (task, runtime, context) => { + await notify(task.input.text); + await runtime.commit(() => ({ status: "terminal", outcome: { status: "completed", result: null } }), context); + }, + }, + abort: (_task, runtime, context) => + runtime.commit(() => ({ status: "terminal", outcome: { status: "aborted" } }), context), +}); +``` + +There is no sleep in the middle of a phase: a phase reruns from its top after a restart, so work after the wait goes in the next phase, and what it needs goes in the checkpoint. `{ status: "sleeping", until, checkpoint }` returned from `runtime.commit()` is the same sleep, committed together with other writes. Retry backoffs and deferred-response polls of the built-in tasks sleep the same way. + +Every timer the Harness uses goes through `setTimer` (default `setTimeout`), and every time through `now` (default `Date.now`); pass both to run a Harness on a fake clock. Sleep deadlines are stored on `now`, so every process that opens one storage must use the same clock. While scheduling runs, a timer resumes sleeping tasks when they are due and keeps a Node.js process alive meanwhile. + +A host whose process can be evicted, such as a Cloudflare Durable Object, wakes it itself. With `onWake`, the Harness arms no timer for sleeps; it tells the host what `harness.wake()` says whenever that changes: `active` while some task can run now or is running, and `at`, the earliest sleep. The host keeps its alarm armed and calls `resume()` when it fires: + +```typescript +export class Session extends DurableObject { + #harness!: Harness; + + constructor(ctx: DurableObjectState, env: Env) { + super(ctx, env); + void ctx.blockConcurrencyWhile(async () => { + const storage = await openDurableObjectSqliteStorage(ctx.storage); + this.#harness = await Harness.open(storage, { models, registry, onWake: (wake) => void this.#arm(wake) }, context); + this.#harness.resume(); + }); + } + + override async alarm(): Promise { + this.#harness.resume(); // resumes due sleeps and work a lost process left + await this.#arm(this.#harness.wake()); + } + + async submit(conversation: Conversation, input: string, requestId: string): Promise { + // Arm before admitting work: an early alarm is harmless, a missing one strands the work. + await this.#arm({ ...this.#harness.wake(), active: true }); + await conversation.submit({ type: "input", content: input, requestId }, context); + } + + async #arm(wake: Wake): Promise { + // A recovery heartbeat while work runs, never later than the earliest sleep. + const heartbeat = wake.active ? Date.now() + 60_000 : undefined; + const at = heartbeat === undefined || wake.at === undefined ? (heartbeat ?? wake.at) : Math.min(heartbeat, wake.at); + if (at === undefined) await this.ctx.storage.deleteAlarm(); + else await this.ctx.storage.setAlarm(at); + } +} +``` + +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. + ## Extensions Code the Harness runs, other than its built-in tasks, comes in named extensions installed in a registry your process owns: diff --git a/packages/durable/docs/spec.md b/packages/durable/docs/spec.md index e6747aada..5eb91c392 100644 --- a/packages/durable/docs/spec.md +++ b/packages/durable/docs/spec.md @@ -371,8 +371,26 @@ type HarnessOptions = { ) => ExecutionEnv | undefined | Promise; /** Runs in every commit that creates or forks a conversation, after the built-in creation hook (below). */ readonly conversationCreated?: (tx: Tx, conversation: ConversationRecord) => void | Promise; - /** Wall clock for sleeps, retries, message timestamps, and task lifecycle times. Default `Date.now`. */ + /** + * Wall clock for sleeps, retries, message timestamps, and task lifecycle times, epoch milliseconds. Default + * `Date.now`. Sleeping tasks store deadlines on it, so every process that opens one storage uses one clock. + */ readonly now?: () => number; + /** + * 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. + */ + readonly setTimer?: ( + callback: () => void, + delayMs: number, + options: { readonly background: boolean }, + ) => (() => void) | undefined; + /** + * Replaces the timer for sleeping tasks (section 5.6): called with `Harness.wake()` after open and whenever it + * changes, off the Session line; the host arms its own wakeup and calls `resume()` when it fires. + */ + readonly onWake?: (wake: Wake) => void; /** Failures that do not fail the calling operation, and the Session's failure, once (section 10.1). */ readonly onReport?: (error: unknown) => void; }; @@ -555,6 +573,8 @@ type TaskInspection = { | { readonly kind: "running" } | { readonly kind: "ready"; readonly migrates: boolean } | { readonly kind: "waiting"; readonly on: readonly TaskId[] } + /** A sleep whose `until` the clock has not reached (section 5.6). */ + | { readonly kind: "sleeping"; readonly until: number } /** Outcome held until its ordinary owned work drains (section 5.5). */ | { readonly kind: "completing" } | { @@ -564,6 +584,9 @@ type TaskInspection = { }; }; +/** When the Session next needs a running process (section 5.6). */ +type Wake = { readonly active: boolean; readonly at?: number }; + type HarnessInspection = { readonly scheduling: "paused" | "running" | "closing"; readonly tasks: readonly TaskInspection[]; @@ -613,6 +636,8 @@ interface Conversation { interface Harness extends Session { resume(): void; + /** When the Session next needs a running process, from memory; also while paused (section 5.6). */ + wake(): Wake; root( context: Context, @@ -658,7 +683,7 @@ settlement. 7.1). `createRegistry()` always holds the built-in task definitions (section 8), and open rejects a registry whose snapshot lacks any of them. Open changes surviving `running` tasks to `pending`, keeps `waiting` tasks -waiting, and re-evaluates finalization and `failFast` for `completing` and +waiting and `sleeping` tasks asleep, and re-evaluates finalization and `failFast` for `completing` and `waiting` tasks (section 5.5). It never migrates or terminalizes a task because its definition is missing or unmigratable. Such a task stays `pending` or `waiting` and is **blocked**: the scheduler skips it and reconsiders it whenever the registry changes (section @@ -676,8 +701,10 @@ the root already exists, `agent` and `init` are ignored and no write occurs. Reopen finds the same root by its reserved ID. A conversation whose agent has no model produces a durable `no_model` generation failure. -`resume()` is idempotent while running and only enables scheduling. It does not -repeat open-time reconciliation, and it throws after close. Work that must happen +`resume()` is idempotent while running: it enables scheduling and makes sleeping +tasks whose time has come runnable, which is how a host wakeup resumes them +(section 5.6). It does not repeat open-time reconciliation, and it throws after +close. Work that must happen before any task runs, such as installing extensions or seeding, happens before `resume()`. Calls that ask for progress also enable scheduling, so they never wait on a paused Harness: `Conversation.submit()`, `Conversation.compact()`, @@ -959,7 +986,8 @@ enable scheduling, and runs no task code. Each live task carries its state under the current registry: `running` with an active invocation; `completing` for a held outcome; `waiting` with the live tasks it still waits for, the live part of its `on`, or, when abort-marked, its live ordinary owned work (section -5.5); `ready` when the next scheduling pass would reserve it, with `migrates` +5.5); `sleeping` with its `until` while the clock has not reached it (section +5.6); `ready` when the next scheduling pass would reserve it, with `migrates` when its definition is newer and has `migrate`; or `blocked` (section 5.4). A migration shows as failed only after the scheduler tried it, or when the newer definition has no `migrate`; inspection never runs one to find out. Blocked reasons are derived, never stored. Queued and placed @@ -1607,17 +1635,17 @@ 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 +after each commit and when a background timer (section 5.6) 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 +incarnation retires. Where the host declines background timers, as the default +does 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. +with their task (section 5.5). `createSession(storage, { documentRetentionMs, setTimer })` +sets them 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 @@ -1691,6 +1719,8 @@ type TaskState = readonly on: readonly TaskId[]; readonly policy: JoinPolicy; } + /** Parked without an invocation until the Harness clock reaches `until`; then resumes at `checkpoint` (section 5.6). */ + | { readonly status: "sleeping"; readonly checkpoint: S; readonly until: number } /** Outcome decided; terminal once no ordinary owned work below is live (section 5.5). Runs no more code. */ | { readonly status: "completing"; readonly outcome: TaskOutcome } | { readonly status: "terminal"; readonly outcome: TaskOutcome }; @@ -1793,7 +1823,10 @@ interface TaskRuntime extends DocumentObserver, Docum now(): number; /** Forward a non-fatal failure to `HarnessOptions.onReport`. */ report(error: unknown): void; - sleep(until: number, context: Context): Promise; + /** Commit `{ status: "sleeping", until, checkpoint }`; resolves once stored (section 5.6). */ + sleepUntil(until: number, checkpoint: S, context: Context): Promise; + /** A `HarnessOptions.setTimer` timer for pacing work in a phase; the invocation's end disarms it. */ + setTimer(callback: () => void, delayMs: number): () => void; } type TaskDefinition = { @@ -1879,9 +1912,11 @@ abort: async (task, runtime, context) => { }, ``` `memo(name, candidate)` is one gated commit; `memo(name)` reads the committed -record. `sleep(until)` compares against the Harness `now` clock and rejects when -the invocation is signalled or its context is cancelled. Watches acquired through -the runtime stop when the invocation ends. `snapshot()`/`snapshotAsOf()` read +record. A phase never waits for time in memory: `sleepUntil()` ends the phase +with a durable sleep (section 5.6), and `setTimer()` only paces work inside a +phase, such as progress commits; the invocation's end disarms its timers, and a +call after its end arms nothing. Watches acquired through the runtime stop when +the invocation ends. `snapshot()`/`snapshotAsOf()` read committed documents, for example to supply `PromptInput.read` (section 7.4). `getTask()` and `entry()` read committed records with one lookup each; `waitForTask()` waits for a terminal receipt, for example a child task created by a tool. @@ -1890,14 +1925,14 @@ immutable entries off the line, like `Conversation.context()`. The scheduler kee each conversation's last range and view in memory: a later read by any of its tasks with the same head marker scans and derives only the entries after that range's tail. A conversation keeps them while busy and for `contextRetentionMs` once idle. -Expiry is checked on task changes and by one unreferenced timer, which never keeps -the process alive; where timers cannot be unreferenced, as in Cloudflare Workers, -only task changes check it. Kept entries are frozen and each read returns its own +Expiry is checked on task changes and by one background timer (section 5.6), which +never keeps the process alive; where the host declines it, as the default does in +Cloudflare Workers, only task changes check it. Kept entries are frozen and each read returns its own view arrays. Like every runtime operation, these reject after the invocation ends. -Reservation durably changes `pending`, or `waiting` once it may resume, to -`running`. One invocation runs phase +Reservation durably changes `pending`, `waiting` once it may resume, or +`sleeping` once due, to `running`. One invocation runs phase handlers in sequence; checkpoint commits retain `running`. Before every phase, including the first, the scheduler runs one step callback on the Session line. It reads the committed task and synchronously applies the first matching rule @@ -2282,6 +2317,72 @@ through `outcomes()`, and decides its own outcome. `abortTask(Checkout)` instead marks `Checkout`, the cascade marks the live payments, their abort handlers run first, then `Checkout.abort` runs and sees their final outcomes. +### 5.6 Sleeping, timers, and wakeups + +**Sleeping.** A phase may commit `{ status: "sleeping", checkpoint, until }`, or +call `runtime.sleepUntil(until, checkpoint, context)`, which commits exactly +that. The task stops without an invocation and resumes at `checkpoint` once the +Harness clock reaches `until`, epoch milliseconds on `HarnessOptions.now`. +`sleepUntil()` resolves when the sleep is stored, not when the time comes; the +phase returns right after. `until` must be a finite number, `0` or more; an +abort handler cannot sleep. A sleep adds no wait-for edge. An abort mark ends +it: the abort handler runs once the task's owned work drains, without waiting +for `until`. Open keeps sleeping tasks asleep, and their deadlines survive any +number of reopens. + +There is no sleep inside a phase: a phase reruns from its top after a crash or +close, so an in-memory wait would hold an invocation, and a deadline computed +inside the phase would move on every rerun. Work after a wait belongs in the +next phase, with what it needs in the checkpoint: + +```ts +work: async (task, runtime, context) => { + const a = await stepA(); + await runtime.sleepUntil(runtime.now() + 10 * 60_000, { phase: "afterSleep", a }, context); +}, +afterSleep: async (task, runtime, context) => { + const b = await stepB(task.state.checkpoint.a); + await runtime.commit(() => ({ status: "running", checkpoint: { phase: "next", b } }), context); +}, +``` + +Work before the sleep is not exactly-once: a crash before the sleep is stored +reruns it (section 5.2). A sleep cannot be woken early except by an abort. A +phase that runs before its time, as after an early wake of a built-in retry or +poll, sleeps again with the same checkpoint (section 8.3). + +**Timers.** Every timer of the Session and Harness is armed through +`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 +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 +or late; every caller checks its clock again. + +**Wakeups.** `Harness.wake()` says, from memory and also while scheduling is +paused, when the Session next needs a running process: + +- `active`: some task can run now or is running: a pending task that is not + blocked, a running invocation, a waiting task whose `on` is settled, a due + sleep, or an abort-marked task whose owned work drained. A host keeps a + recovery wakeup armed while it is set, so work cut off by a crash resumes. +- `at`: the earliest sleep not yet due. + +Blocked tasks (section 5.4), `completing` tasks, and waits on live tasks count +for neither; the work they wait for does. With `HarnessOptions.onWake`, the +Harness arms no timer for sleeps. It calls `onWake` with `wake()` after open and +whenever the value changes, off the Session line; a throw is reported. The host +arms its own wakeup and calls `resume()` when it fires, which makes due sleeps +runnable. A host whose process can be evicted, such as a Durable Object with +its one alarm, arms the alarm at `at`, or at `now + heartbeat` while `active` +if that is earlier, and deletes it when neither is set. It arms the recovery wakeup +before admitting work, since an early wakeup is harmless and a missing one +leaves work stranded. + ## 6. Submissions and inbox Submission records back awaitable host objects. Admission is Harness-internal; @@ -2869,6 +2970,8 @@ interface ToolExecutionApi extends Docum readonly models: Models; /** Built by `HarnessOptions.env` for this call. */ readonly env: ExecutionEnv | undefined; + /** The Harness clock, for times a rerun after a restart compares against. */ + now(): number; output(chunk: string | Uint8Array): void; /** The output retained so far, as the model will see it, and whether earlier output was dropped. */ retainedOutput(): { readonly text: string; readonly truncated: boolean }; @@ -4054,17 +4157,20 @@ The run's inputs live in `pi.live.run`, not in the task input. the next threshold check handles silent overflow. - `error` that `isRetryableAssistantError()` accepts while the settings' retry policy allows another attempt (`enabled` and `attempt <= maxRetries`, - so `maxRetries` counts retries after the first attempt): append the error entry and move to - `retry` with `until = now + retryDelayMs(policy, attempt)`. + so `maxRetries` counts retries after the first attempt): append the error entry and sleep + (section 5.6) until `until = now + retryDelayMs(policy, attempt)` at `retry`. - any other `error`, or `aborted` without an abort mark: append the error entry, settle the inputs `unanswered` with `model_error`, remove `run` and `generation`, and fail. - - `deferred`: move to `poll` with `pollAt = now + (handle.pollAfterMs ?? 5000)`. -- `retry` sleeps until `until`, then returns to `prepare` with the next attempt, - so agent and settings changes made during the backoff apply. -- `poll` sleeps until `pollAt` and calls `models.fetchDeferred()`. A still - deferred result moves `pollAt` strictly later; any other result is classified - as above. + - `deferred`: sleep until `pollAt = now + (handle.pollAfterMs ?? 5000)` at `poll`. +- `retry` runs once the backoff's sleep is due and returns to `prepare` with the + next attempt, so agent and settings changes made during the backoff apply. + Run before `until`, as a record of an earlier version that slept inside a + running `retry` phase is, it sleeps again with the same checkpoint. +- `poll` runs once due and calls `models.fetchDeferred()`; run before `pollAt`, + it sleeps again. A still deferred result sleeps until a strictly later + `pollAt`; any other result is classified as above. A model gone after reopen + fails `no_model` once the poll is due. - The abort handler calls `models.cancelDeferred()` in `poll` when the model is known (a failure is reported), converts a committed partial, settles the inputs `unanswered` with `aborted`, removes `run` and `generation`, and ends `aborted`. @@ -4332,12 +4438,12 @@ Phases: - `stop` with non-empty text and no tool call: the text is the summary, placed as below. - `error` that `isRetryableAssistantError()` accepts while the settings' - retry policy, read at classification, allows another attempt: move to `retry` with `until = now + - retryDelayMs(policy, attempt)`, and set the status's `attempt` and `retry`. + retry policy, read at classification, allows another attempt: sleep (section 5.6) until `until = now + + retryDelayMs(policy, attempt)` at `retry`, and set the status's `attempt` and `retry`. - anything else, including a `length` stop, which leaves the summary incomplete: fail with `model_error`. -- `retry` sleeps until `until`, then returns to `summarize` with the next attempt - and the same pinned request. +- `retry` runs once the sleep is due and returns to `summarize` with the next + attempt and the same pinned request; run before `until`, it sleeps again. The built-in prompts are the coding agent's structured checkpoint format. The prompt also asks the model to carry forward an earlier summary at the start of @@ -4791,6 +4897,8 @@ type TaskGraphState = readonly on: readonly TaskId[]; readonly policy: JoinPolicy; } + /** Asleep until the Harness clock reaches `until` (section 5.6). */ + | { readonly status: "sleeping"; readonly phase: string; readonly until: number } /** Outcome held until its ordinary owned work drains (section 5.5). */ | { readonly status: "completing"; readonly outcome: TaskOutcome["status"] }; @@ -4894,7 +5002,7 @@ type EntryQuery = { type TaskQuery = { readonly conversationId?: ConversationId; readonly kind?: string; - readonly status?: "pending" | "running" | "waiting" | "completing" | "terminal"; + readonly status?: "pending" | "running" | "waiting" | "sleeping" | "completing" | "terminal"; readonly abortRequested?: boolean; readonly background?: boolean; readonly order?: ScanOrder; // default ascending @@ -5148,7 +5256,9 @@ One SQL transaction is one Session commit. SQLite stores: - indexed document bases/deltas by document and commit sequence. Live task transitions replace one row. Terminal tasks remain as small records. -Document reads use indexed base-plus-tail ranges. The first implementation stores +Opening applies pending schema migrations in one transaction; schema version 2 +rebuilds the task table to admit the `sleeping` status. A database migrated by a +newer version is refused by an older one. Document reads use indexed base-plus-tail ranges. The first implementation stores Chord records directly; it does not translate generic operations to SQLite JSON functions. diff --git a/packages/durable/src/codemode/codemode.ts b/packages/durable/src/codemode/codemode.ts index 7415071bc..a8bcde21a 100644 --- a/packages/durable/src/codemode/codemode.ts +++ b/packages/durable/src/codemode/codemode.ts @@ -395,7 +395,7 @@ class CodemodeCall { } entry.tools.push(tool.name); } - const now = Date.now(); + const now = api.now(); const seed = crypto.getRandomValues(new Uint32Array(1))[0]!; type Found = | { done: ToolExecutionResult } @@ -448,7 +448,7 @@ class CodemodeCall { ), ); } - const remaining = this.start.deadline === undefined ? undefined : this.start.deadline - Date.now(); + const remaining = this.start.deadline === undefined ? undefined : this.start.deadline - this.api.now(); if (remaining !== undefined && remaining <= 0) { return this.decide(errorResult(`Script timed out: its deadline passed while it was interrupted.`)); } @@ -546,7 +546,7 @@ class CodemodeCall { const details = this.details(); const formatted = await formatResult(result, { generatedImages: this.generatedImages, - wallMs: Date.now() - this.start.startedAt, + wallMs: this.api.now() - this.start.startedAt, maxOutputTokens: this.start.maxOutputTokens, calls: details.calls, limits: OUTPUT_LIMITS, @@ -799,7 +799,7 @@ class CodemodeCall { })); running.then( (done) => { - this.queue.push({ id: call.id, at: Date.now(), ...done }); + this.queue.push({ id: call.id, at: this.api.now(), ...done }); void this.pump(); }, (error: unknown) => { diff --git a/packages/durable/src/harness/compaction.ts b/packages/durable/src/harness/compaction.ts index 4cf1966ca..9196f967b 100644 --- a/packages/durable/src/harness/compaction.ts +++ b/packages/durable/src/harness/compaction.ts @@ -204,7 +204,8 @@ export const CompactionTask = defineTask { const { phase: _, until, ...request } = task.state.checkpoint; - await runtime.sleep(until, context); + if (until > runtime.now()) return runtime.sleepUntil(until, task.state.checkpoint, context); const attempt = request.attempt + 1; await runtime.commit(async (tx) => { const status = compactionStatus(await tx.doc(LiveDoc, runtime.conversationId), runtime.taskId); diff --git a/packages/durable/src/harness/generation.ts b/packages/durable/src/harness/generation.ts index 005436ce5..ffd708c7a 100644 --- a/packages/durable/src/harness/generation.ts +++ b/packages/durable/src/harness/generation.ts @@ -220,9 +220,10 @@ export const GenerationTask = defineTask { const { attempt, compacted, until } = task.state.checkpoint; - await runtime.sleep(until, context); + if (until > runtime.now()) return runtime.sleepUntil(until, task.state.checkpoint, context); await runtime.commit(async (tx) => { (await tx.doc(LiveDoc, runtime.conversationId)).generation = { attempt: attempt + 1 }; const checkpoint: GenerationCheckpoint = { @@ -233,11 +234,12 @@ export const GenerationTask = defineTask { const { attempt, compacted, model: ref, cutoff, handle, pollAt } = task.state.checkpoint; const model = runtime.models.getModel(ref.provider, ref.modelId); if (model === undefined) return failNoModel(runtime, ref, context); - await runtime.sleep(pollAt, context); + if (pollAt > runtime.now()) return runtime.sleepUntil(pollAt, task.state.checkpoint, context); const message = await runtime.models.fetchDeferred(model, handle, { signal: runtime.signal }); const request = { attempt, compacted, model: ref, cutoff, pollAt }; await classify(runtime, request, message, context); @@ -400,7 +402,7 @@ async function streamResponse( ): Promise { const interval = runtime.settings.progress.partialIntervalMs; let pending: AssistantMessage | undefined; - let timer: ReturnType | undefined; + let timer: (() => void) | undefined; let inFlight: Promise | undefined; let stopped = false; const flush = (): void => { @@ -424,7 +426,7 @@ async function streamResponse( }) .finally(() => { inFlight = undefined; - if (pending !== undefined && !stopped) timer = setTimeout(flush, interval); + if (pending !== undefined && !stopped) timer = runtime.setTimer(flush, interval); }); }; try { @@ -434,12 +436,12 @@ async function streamResponse( // never gets past it, so it never leaves a partial. if (event.type === "done" || event.type === "error" || event.partial.content.length === 0) continue; pending = event.partial; - if (timer === undefined && inFlight === undefined) timer = setTimeout(flush, interval); + if (timer === undefined && inFlight === undefined) timer = runtime.setTimer(flush, interval); } return await events.result(); } finally { stopped = true; - clearTimeout(timer); + timer?.(); await inFlight; } } @@ -461,7 +463,7 @@ async function classify( runtime.now() + (handle.pollAfterMs ?? DEFAULT_POLL_AFTER_MS), request.pollAt === undefined ? Number.NEGATIVE_INFINITY : request.pollAt + 1, ); - await runtime.commit(async (tx) => { + await runtime.commit(async (tx): Promise => { (await tx.doc(LiveDoc, conversationId)).generation = { attempt, deferred: { pollAt } }; const checkpoint = { phase: "poll", @@ -472,7 +474,7 @@ async function classify( handle, pollAt, } as const; - return { status: "running", checkpoint }; + return { status: "sleeping", checkpoint, until: pollAt }; }, context); return; } @@ -523,7 +525,7 @@ async function classify( ...(compacted === undefined ? {} : { compacted }), until, } as const; - return { status: "running", checkpoint }; + return { status: "sleeping", checkpoint, until }; } const text = message.errorMessage ?? `Model response ended with stop reason ${message.stopReason}`; endRun(tx, live, runtime.taskId, { status: "unanswered", reason: "model_error", detail: text }); diff --git a/packages/durable/src/harness/harness.ts b/packages/durable/src/harness/harness.ts index 295645299..69069c044 100644 --- a/packages/durable/src/harness/harness.ts +++ b/packages/durable/src/harness/harness.ts @@ -5,6 +5,7 @@ import { ResetEntry } from "../entries.ts"; import type { ExecutionEnv } from "../env/index.ts"; import { contained, SessionImpl } from "../session/session.ts"; import type { Transaction } from "../session/transaction.ts"; +import { hostTimers } from "../timers.ts"; import type { ConversationId, ConversationOwnership, @@ -51,6 +52,7 @@ import type { Submission, SubmissionDraft, ToolRegistration, + Wake, } from "./types.ts"; import { addUsageState, UsageDoc, type UsageState } from "./usage.ts"; import { scanAll } from "./util.ts"; @@ -192,7 +194,7 @@ class HarnessImpl extends SessionImpl implements const now = safeNow(options.now, report); const hostSettings = currentSettings(options.settings); // Read at each use, as all settings are, but only this field: it is read on every commit. - super(rawStorage, now, () => hostSettings()?.documentRetentionMs); + super(rawStorage, now, () => hostSettings()?.documentRetentionMs, options.setTimer); // Every component reads and writes through the Session's failure guard: `rawStorage` serves only `super()`. const storage = this.storage; this.#options = options; @@ -208,6 +210,8 @@ class HarnessImpl extends SessionImpl implements settings, env: (id, callContext) => this.buildEnv(id, callContext), now, + onWake: options.onWake, + timers: hostTimers(options.setTimer, report), report: this.#report, settleOutcome: settleSchedulerOutcome, withdrawInputs: withdrawQueuedInputs, @@ -259,6 +263,11 @@ class HarnessImpl extends SessionImpl implements this.#tasks.resume(); } + wake(): Wake { + this.assertUsable(); + return this.#tasks.wake(); + } + getTask(id: TaskId, context: Context): Promise | undefined> { return this.readOnLine(() => this.storage.task(id, context)) as Promise< TaskRecord | undefined diff --git a/packages/durable/src/harness/scheduler.ts b/packages/durable/src/harness/scheduler.ts index 2fe49e96d..9381b407d 100644 --- a/packages/durable/src/harness/scheduler.ts +++ b/packages/durable/src/harness/scheduler.ts @@ -5,6 +5,7 @@ import type { ExecutionEnv } from "../env/index.ts"; import { errorMessage } from "../errors.ts"; import type { SessionImpl } from "../session/session.ts"; import type { Transaction } from "../session/transaction.ts"; +import type { Timers } from "../timers.ts"; import type { CommitPublication, ConversationId, @@ -35,14 +36,15 @@ import type { Settings, SettledTask, TaskInspection, + Wake, } from "./types.ts"; import { closedError, scanAll, Waiters } from "./util.ts"; type AnyTaskRecord = TaskRecord; -/** A record that can still run code: pending, running, or waiting. */ +/** A record that can still run code: pending, running, waiting, or sleeping. */ type RunnableTaskRecord = Extract< AnyTaskRecord, - { readonly state: { readonly status: "pending" | "running" | "waiting" } } + { readonly state: { readonly status: "pending" | "running" | "waiting" | "sleeping" } } >; type Checkpoint = { readonly phase: string }; type ErasedDefinition = TaskDefinition; @@ -50,9 +52,7 @@ type ErasedRuntime = TaskRuntime; type ErasedRunningTask = RunningTask; const SCAN_PAGE_SIZE = 256; -/** Longest delay `setTimeout` supports; longer sleeps wait in several steps. */ -const MAX_TIMER_DELAY = 2_147_483_647; -const LIVE_STATUSES = ["pending", "running", "waiting", "completing"] as const; +const LIVE_STATUSES = ["pending", "running", "waiting", "sleeping", "completing"] as const; /** Why a pending task cannot be reserved under a registry snapshot. Derived, never persisted. */ type BlockedReason = "missing_task" | "task_too_old" | "migration_failed"; @@ -76,6 +76,8 @@ type Invocation = { readonly context: Context; /** Watches acquired through the runtime; stopped at invocation end. */ readonly watches: Set>; + /** Timers armed through the runtime; disarmed at invocation end. */ + readonly timers: Set<() => void>; ended: boolean; readonly done: Promise; readonly finish: () => void; @@ -161,6 +163,10 @@ export type TaskSchedulerOptions = { /** Build a conversation's environment with `HarnessOptions.env`. */ readonly env: (conversationId: ConversationId, context: Context) => Promise; readonly now: () => number; + /** `HarnessOptions.onWake`: replaces the timer for sleeping tasks. */ + readonly onWake: ((wake: Wake) => void) | undefined; + /** The host's timers (`HarnessOptions.setTimer`). */ + readonly timers: Timers; readonly report: (error: unknown) => void; /** Harness cleanup staged in the commit that makes an outcome the scheduler wrote itself terminal. */ readonly settleOutcome: (tx: Transaction, record: AnyTaskRecord, outcome: SchedulerOutcome) => Promise; @@ -179,8 +185,9 @@ export type TaskSchedulerOptions = { /** * Durable task scheduler of one Harness. * - * `#live` mirrors every committed non-terminal task record: pending, running, waiting, and completing. The synchronous - * commit listener updates it on the Session line, so code running on the line reads exactly the committed state from it. + * `#live` mirrors every committed non-terminal task record: pending, running, waiting, sleeping, and completing. The + * synchronous commit listener updates it on the Session line, so code running on the line reads exactly the committed + * state from it. * * Tasks and conversations form one ownership tree (spec §5.5): a task's parent is its owner task, or its conversation; * a conversation's parent is its owner task, if any. Walks up that tree decide what reaches one task: a cancelling @@ -206,6 +213,8 @@ export class TaskScheduler { readonly #settings: TaskSchedulerOptions["settings"]; readonly #env: TaskSchedulerOptions["env"]; readonly #now: () => number; + readonly #onWake: ((wake: Wake) => void) | undefined; + readonly #timers: Timers; readonly #report: (error: unknown) => void; readonly #settleOutcome: TaskSchedulerOptions["settleOutcome"]; readonly #withdrawInputs: TaskSchedulerOptions["withdrawInputs"]; @@ -218,8 +227,8 @@ export class TaskScheduler { * first idle check after `settings.contextRetentionMs` of idleness, and at close. */ readonly #contexts = new Map(); - /** Timer for the earliest expiry of an idle context, only where timers can be unreferenced (see `#scheduleExpiry`). */ - #expiry: { readonly at: number; readonly timer: ReturnType } | undefined; + /** Background timer for the earliest expiry of an idle context, unless the host declined it (see `#scheduleExpiry`). */ + #expiry: { readonly at: number; readonly disarm: () => void } | undefined; readonly #invocations = new Map(); readonly #taskWaiters = new Waiters>(); /** Idle waiters by conversation; `undefined` waits for the whole Harness. */ @@ -261,6 +270,13 @@ export class TaskScheduler { * them with reason `restart`, so nothing of them or below them runs a phase again. */ readonly #abandoned = new Set(); + /** Sleeping live tasks, due or not: the candidates of `#wakeDue()`. */ + readonly #sleeping = new Set(); + /** Timer for the earliest sleep, without `onWake` while scheduling runs. */ + #timer: { readonly at: number; readonly disarm: () => void } | undefined; + /** Last value given to `onWake`, so an unchanged one is not repeated. */ + #lastWake: Wake | undefined; + #wakeScheduled = false; #reconcileScheduled = false; #sweepScheduled = false; #cascadePending = false; @@ -279,6 +295,8 @@ export class TaskScheduler { this.#settings = options.settings; this.#env = options.env; this.#now = options.now; + this.#onWake = options.onWake; + this.#timers = options.timers; this.#report = options.report; this.#settleOutcome = options.settleOutcome; this.#withdrawInputs = options.withdrawInputs; @@ -313,14 +331,58 @@ export class TaskScheduler { // Derive abort marks a crash left unapplied below cancelled owners, and finalize held outcomes. this.#cascadePending = true; this.#scheduleReconcile(); + this.#scheduleWake(); } - /** Enable scheduling. Idempotent; the kick does nothing once closing. */ + /** + * Enable scheduling, and make sleeping tasks that are due runnable: a host wakeup, such as an alarm, calls this. + * Idempotent; the kick does nothing once closing. + */ resume(): void { this.#enabled = true; + this.#wakeDue(); this.#kick(); } + /** + * When the Session next needs a process: `active` while some task can run now or is running, `at` for the earliest + * sleep not yet due. Synchronous, from memory; a task whose chain is not loaded yet counts as it would alone. + */ + wake(): Wake { + const now = this.#now(); + let active = this.#invocations.size > 0; + let at: number | undefined; + let snapshot: RegistrySnapshot | undefined; + let snapshotFailed = false; + for (const record of this.#live.values()) { + const state = record.state; + if (state.status === "completing") continue; + if (state.status === "waiting" && !record.abortRequested && state.on.some((id) => this.#live.has(id))) + continue; + if (state.status === "sleeping" && !record.abortRequested && state.until > now) { + at = at === undefined ? state.until : Math.min(at, state.until); + continue; + } + if (active) continue; + // An abort invocation waits for the live work below it, which counts itself. + if (record.abortRequested && this.#waitingOn(record, true).length > 0) continue; + // A task no definition can take waits for a registry change, except an abort-marked one, which is orphaned. + if (!record.abortRequested || record.abortReason !== undefined) { + if (snapshot === undefined && !snapshotFailed) { + try { + snapshot = this.#registry.snapshot(); + } catch (error) { + snapshotFailed = true; + this.#report(error); + } + } + if (snapshot !== undefined && "reason" in this.#fit(record, snapshot.task(record.kind))) continue; + } + active = true; + } + return at === undefined ? { active } : { active, at }; + } + /** * Signal every invocation and wait for them; writes nothing. Close calls this after every close listener has run, so * waits, watches, and streams a task holds end with the Session's reason, not as cancelled by its signal. @@ -531,9 +593,9 @@ export class TaskScheduler { } /** - * Run `#settleIdle()` when the earliest idle context expires. The timer is unreferenced, so it never keeps the process - * alive. Where timers cannot be unreferenced, as in Cloudflare Workers, none is kept: a pending timer could keep a - * Durable Object from being evicted, and eviction frees the contexts. There, task changes alone check expiry. + * Run `#settleIdle()` when the earliest idle context expires, on a background timer, which never keeps the process + * alive. Where the host declines it, as the default does in Cloudflare Workers, none is kept: a pending timer could + * keep a Durable Object from being evicted, and eviction frees the contexts. There, task changes alone check expiry. */ #scheduleExpiry(): void { const retention = this.#contextRetentionMs(); @@ -544,23 +606,14 @@ export class TaskScheduler { } } if (this.#expiry !== undefined && this.#expiry.at === at) return; - if (this.#expiry !== undefined) clearTimeout(this.#expiry.timer); + this.#expiry?.disarm(); this.#expiry = undefined; if (at === undefined || this.#closing) return; - const timer = setTimeout( - () => { - this.#expiry = undefined; - this.#settleIdle(); - }, - Math.min(Math.max(0, at - this.#now()), MAX_TIMER_DELAY), - ); - const unref = (timer as { unref?: unknown }).unref; - if (typeof unref !== "function") { - clearTimeout(timer); - return; - } - unref.call(timer); - this.#expiry = { at, timer }; + const disarm = this.#timers.background(() => { + this.#expiry = undefined; + this.#settleIdle(); + }, at - this.#now()); + if (disarm !== undefined) this.#expiry = { at, disarm }; } // ─── Ownership ─────────────────────────────────────────────────────────── @@ -802,6 +855,7 @@ export class TaskScheduler { if (record.state.status === "terminal") { const node = nodeOf(record); this.#live.delete(id); + this.#sleeping.delete(id); this.#intent.delete(id); this.#runnable.delete(id); this.#finalizeChecks.delete(id); @@ -828,6 +882,8 @@ export class TaskScheduler { this.#runnable.add(id); } if (record.state.status === "waiting") for (const on of record.state.on) addTo(this.#waiters, on, id); + if (record.state.status === "sleeping") this.#sleeping.add(id); + else this.#sleeping.delete(id); } /** The nearest live task above ended ordinary work may be free now: one holding its outcome, or an aborting one. */ @@ -1088,11 +1144,69 @@ export class TaskScheduler { this.#taskWaiters.rejectAll(error); this.#idleWaiters.rejectAll(error); this.#contexts.clear(); - if (this.#expiry !== undefined) clearTimeout(this.#expiry.timer); + this.#expiry?.disarm(); this.#expiry = undefined; + this.#timer?.disarm(); + this.#timer = undefined; + } + + /** Recompute the wake in a microtask, off the Session line: arm the built-in timer, or tell `onWake` of a change. */ + #scheduleWake(): void { + if (this.#wakeScheduled || this.#closing) return; + this.#wakeScheduled = true; + queueMicrotask(() => { + this.#wakeScheduled = false; + if (this.#closing) return; + const wake = this.wake(); + if (this.#onWake === undefined) { + this.#armTimer(this.#enabled ? wake.at : undefined); + return; + } + const last = this.#lastWake; + if (last !== undefined && last.active === wake.active && last.at === wake.at) return; + this.#lastWake = wake; + try { + this.#onWake(wake); + } catch (error) { + this.#report(error); + } + }); + } + + /** + * Keep the timer for the earliest sleep. It keeps the process alive, unlike the context expiry timer: a process with a + * sleeping task has work left, such as a retry backoff. Rechecks the clock when it fires, so a wake before `until` + * only arms it again. + */ + #armTimer(at: number | undefined): void { + if (this.#timer !== undefined && this.#timer.at === at) return; + this.#timer?.disarm(); + this.#timer = undefined; + if (at === undefined || this.#closing) return; + const disarm = this.#timers.foreground(() => { + this.#timer = undefined; + this.#wakeDue(); + this.#kick(); + }, at - this.#now()); + this.#timer = { at, disarm }; + } + + /** Make every sleeping task whose time has come a reservation candidate. */ + #wakeDue(): void { + for (const id of this.#sleeping) { + const record = this.#live.get(id); + if (record !== undefined && !this.#notDue(record)) this.#runnable.add(id); + } + } + + /** A sleep whose `until` the clock has not reached; an abort mark ends a sleep. */ + #notDue(record: AnyTaskRecord): boolean { + const state = record.state; + return state.status === "sleeping" && !record.abortRequested && state.until > this.#now(); } #kick(): void { + this.#scheduleWake(); this.#dirty = true; if (this.#draining || !this.#enabled || this.#closing) return; this.#draining = true; @@ -1148,6 +1262,11 @@ export class TaskScheduler { this.#runnable.delete(id); continue; } + // It returns when its time comes: `#wakeDue()` from the timer or `resume()`. + if (this.#notDue(record)) { + this.#runnable.delete(id); + continue; + } const runnable = record as RunnableTaskRecord; const mode = record.abortRequested ? "abort" : "run"; // Work below an owner with cancellation intent waits for its cascade mark instead of running a phase; schedule @@ -1273,6 +1392,8 @@ export class TaskScheduler { if (record.state.status === "completing") return { kind: "completing" }; const on = this.#waitingOn(record); if (on.length > 0) return { kind: "waiting", on }; + if (this.#notDue(record) && record.state.status === "sleeping") + return { kind: "sleeping", until: record.state.until }; const fit = this.#fit(record, snapshot.task(record.kind)); if ("reason" in fit) return { kind: "blocked", ...fit }; if (fit.migrates && fit.task.definition.migrate === undefined) { @@ -1291,6 +1412,7 @@ export class TaskScheduler { controller, context: withAbortSignal(controller.signal, this.#context), watches: new Set(), + timers: new Set(), ended: false, done, finish: () => finish(), @@ -1339,8 +1461,8 @@ export class TaskScheduler { } /** - * Precedence rules for a run invocation, on the line. Rules 1 (terminal, `completing`, or `waiting`) and 2 (closing) - * are applied by `#step`. Returns whether the invocation continues with the next phase. + * Precedence rules for a run invocation, on the line. Rules 1 (terminal, `completing`, `waiting`, or `sleeping`) and 2 + * (closing) are applied by `#step`. Returns whether the invocation continues with the next phase. */ #decide( tx: Transaction, @@ -1401,10 +1523,10 @@ export class TaskScheduler { } /** - * One synchronous decision on the Session line. A task that is no longer running (rule 1: terminal, `completing`, or - * `waiting`) or a closing Harness (rule 2) ends the invocation without a write; otherwise `decide` may stage a write - * and returns whether the invocation continues. Ending happens inside the callback, before a fault's Harness - * cleanup. A rejected step, such as admission after close, also ends the invocation. + * One synchronous decision on the Session line. A task that is no longer running (rule 1: terminal, `completing`, + * `waiting`, or `sleeping`) or a closing Harness (rule 2) ends the invocation without a write; otherwise `decide` may + * stage a write and returns whether the invocation continues. Ending happens inside the callback, before a fault's + * Harness cleanup. A rejected step, such as admission after close, also ends the invocation. */ async #step( invocation: Invocation, @@ -1458,6 +1580,7 @@ export class TaskScheduler { next: NextTaskState, ): Promise { if (next.status === "waiting") await this.#validateWait(tx, invocation, current, next.on, next.policy); + if (next.status === "sleeping") validateSleep(invocation, current.id, next.until); if (next.status === "terminal") { const overlay = overlayOf(tx); await this.#loadScopes(false); @@ -1563,6 +1686,8 @@ export class TaskScheduler { if (this.#live.has(invocation.taskId)) this.#runnable.add(invocation.taskId); } for (const watch of invocation.watches) void watch.stop(); + for (const disarm of invocation.timers) disarm(); + invocation.timers.clear(); // Pending waits bound to the invocation, such as a tool's waitForTask(), reject with it. invocation.controller.abort(endedError(invocation)); } @@ -1657,7 +1782,12 @@ export class TaskScheduler { rest[1] as Context, ); }) as ErasedRuntime["memo"], - sleep: (until, context) => this.#sleep(invocation, until, context), + sleepUntil: (until, checkpoint, context) => + this.#gated( + invocation, + (tx, current) => this.#commitState(tx, invocation, current, { status: "sleeping", checkpoint, until }), + context, + ), watchDoc: ((...args: readonly unknown[]) => this.#watchDoc(invocation, args)) as ErasedRuntime["watchDoc"], snapshot: ((...args: readonly unknown[]) => this.#read(invocation, () => @@ -1774,6 +1904,19 @@ export class TaskScheduler { if (invocation.ended) throw endedError(invocation); return this.#now(); }, + setTimer: (callback, delayMs) => { + // An ended invocation's timers never fire. + if (invocation.ended) return () => {}; + const disarm = this.#timers.foreground(() => { + invocation.timers.delete(disarm); + if (!invocation.ended) callback(); + }, delayMs); + invocation.timers.add(disarm); + return () => { + invocation.timers.delete(disarm); + disarm(); + }; + }, report: (error) => { if (invocation.ended) throw endedError(invocation); this.#report(error); @@ -1812,20 +1955,6 @@ export class TaskScheduler { ); } - /** Wait until the Harness clock reaches `until`, rechecking it after every timer. */ - async #sleep(invocation: Invocation, until: number, context: Context): Promise { - if (invocation.ended) throw endedError(invocation); - const signals = [invocation.controller.signal]; - if (context.abortSignal !== undefined) signals.push(context.abortSignal); - const signal = AbortSignal.any(signals); - for (;;) { - signal.throwIfAborted(); - const remaining = until - this.#now(); - if (remaining <= 0) return; - await delay(Math.min(remaining, MAX_TIMER_DELAY), signal); - } - } - async #watchDoc(invocation: Invocation, args: readonly unknown[]): Promise | undefined> { if (invocation.ended) throw endedError(invocation); const watch = (await sessionMethod(this.#session, "watchDoc")(...args)) as DocumentWatch | undefined; @@ -1929,18 +2058,12 @@ function canReserve(task: AnyTask, record: AnyTaskRecord): boolean { ); } -function delay(ms: number, signal: AbortSignal): Promise { - return new Promise((resolve, reject) => { - const onAbort = (): void => { - clearTimeout(timer); - reject(signal.reason); - }; - const timer = setTimeout(() => { - signal.removeEventListener("abort", onAbort); - resolve(); - }, ms); - signal.addEventListener("abort", onAbort, { once: true }); - }); +/** A sleep's `until` is a time on the Harness clock; an abort handler cannot sleep. */ +function validateSleep(invocation: Invocation, task: TaskId, until: number): void { + if (invocation.mode === "abort") throw new Error(`Abort handler of task ${task} cannot sleep`); + if (typeof until !== "number" || !Number.isFinite(until) || until < 0) { + throw new Error(`Task ${task} cannot sleep until ${until}: not a time in milliseconds`); + } } /** Structural equality of two JSON values; object key order is ignored. */ diff --git a/packages/durable/src/harness/task-graph.ts b/packages/durable/src/harness/task-graph.ts index 9e79aa61a..541c69441 100644 --- a/packages/durable/src/harness/task-graph.ts +++ b/packages/durable/src/harness/task-graph.ts @@ -24,6 +24,8 @@ export type TaskGraphState = readonly on: readonly TaskId[]; readonly policy: JoinPolicy; } + /** Asleep until the Harness clock reaches `until`. */ + | { readonly status: "sleeping"; readonly phase: string; readonly until: number } /** Outcome held until its ordinary owned work drains. */ | { readonly status: "completing"; readonly outcome: TaskOutcome["status"] }; @@ -52,7 +54,7 @@ type AnyTaskRecord = TaskRecord; type Observer = CommittedStateSource | CommittedWatch; type Mount = { value: TaskGraph; readonly observers: Set }; -const LIVE_STATUSES = ["pending", "running", "waiting", "completing"] as const; +const LIVE_STATUSES = ["pending", "running", "waiting", "sleeping", "completing"] as const; const SCAN_PAGE_SIZE = 256; /** @@ -224,6 +226,8 @@ function stateOf(record: AnyTaskRecord): TaskGraphState { return { status: state.status, phase: phaseOf(state.checkpoint) }; case "waiting": return { status: "waiting", phase: phaseOf(state.checkpoint), on: [...state.on], policy: state.policy }; + case "sleeping": + return { status: "sleeping", phase: phaseOf(state.checkpoint), until: state.until }; // Terminal records never reach here: they leave the graph. case "completing": case "terminal": diff --git a/packages/durable/src/harness/tool.ts b/packages/durable/src/harness/tool.ts index 56ff63ed0..eae03bc8c 100644 --- a/packages/durable/src/harness/tool.ts +++ b/packages/durable/src/harness/tool.ts @@ -310,6 +310,7 @@ async function run( get env() { return env; }, + now: () => runtime.now(), output: (chunk, skipped) => { assertLive(); if (reported.output.push(chunk, skipped)) progress.mark(); @@ -492,6 +493,17 @@ function publishProgress(runtime: Runtime, reported: Reported, enabled: boolean, if (!runtime.signal.aborted) runtime.report(error); }, runtime.settings.progress.outputIntervalMs, + { + // After the invocation ends nothing more commits; the clock only has to answer. + now: () => { + try { + return runtime.now(); + } catch { + return Date.now(); + } + }, + setTimer: (callback, delayMs) => runtime.setTimer(callback, delayMs), + }, ); } diff --git a/packages/durable/src/harness/types.ts b/packages/durable/src/harness/types.ts index 05a91420b..99d05983d 100644 --- a/packages/durable/src/harness/types.ts +++ b/packages/durable/src/harness/types.ts @@ -29,6 +29,7 @@ import type { JsonObject, Page, Session, + SetTimer, SubmissionId, SubmissionRecord, Task, @@ -215,6 +216,8 @@ export interface ToolExecutionApi extend readonly models: Models; /** Built by `HarnessOptions.env` for this call; `undefined` without an environment. */ readonly env: ExecutionEnv | undefined; + /** The Harness clock (`HarnessOptions.now`), for times a rerun after a restart compares against. */ + now(): number; /** * Append running output; it becomes the result's `output` when the result omits it. `skipped` counts output * omitted before the chunk, as reported by an environment given `outputWindow` (`ShellOutputInfo.skipped`). @@ -600,7 +603,26 @@ export type HarnessOptions = { * Table reads throw `ReadAfterWrite`, as in `init`; a throw fails the creating commit. */ readonly conversationCreated?: (tx: Tx, conversation: ConversationRecord) => void | Promise; + /** + * The Harness clock, epoch milliseconds; default `Date.now`. Sleeping tasks store deadlines on it, so every process + * that opens the same storage must use the same clock. + */ readonly now?: () => number; + /** + * 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. + */ + readonly setTimer?: SetTimer; + /** + * Wakes the process for sleeping tasks instead of a `setTimer` timer. Without it, while scheduling runs, a timer that + * keeps the process alive resumes sleeping tasks when they are due. With it, the Harness arms no timer for them: it + * calls `onWake` after open and whenever `Harness.wake()` changes, and the host arms its own wakeup, such as a + * Durable Object alarm, and calls `resume()` when it fires. Called off the Session line; it may call `wake()` and + * `resume()`, and must not block. A throw is reported and ignored. + */ + readonly onWake?: (wake: Wake) => void; /** * Receives failures that do not fail the calling operation, such as a throwing hook or listener, and the error that * fails the Session, once. Its own throw or rejection is dropped. @@ -608,6 +630,14 @@ export type HarnessOptions = { readonly onReport?: (error: unknown) => void; }; +/** + * When a Session next needs a running process (`Harness.wake()`). `active`: some task can run now or is running, so a + * process that dies must be woken again soon, for example by a recovery alarm. `at`: the earliest sleep not yet due, + * when there is one. Neither: nothing needs a process until new work is submitted. Tasks blocked on a missing or + * unmigratable definition, `completing` tasks, and waits on live tasks do not count; the work they wait for does. + */ +export type Wake = { readonly active: boolean; readonly at?: number }; + /** Live task and what the scheduler would do with it under the current registry. */ export type TaskInspection = { readonly record: TaskRecord; @@ -620,6 +650,8 @@ export type TaskInspection = { * must end before its abort handler starts. */ | { readonly kind: "waiting"; readonly on: readonly TaskId[] } + /** Sleeps until the Harness clock reaches `until`. */ + | { readonly kind: "sleeping"; readonly until: number } /** Outcome held until its ordinary owned work drains. */ | { readonly kind: "completing" } /** No registered definition can take it; aborting it settles it as `orphaned`. */ @@ -718,6 +750,13 @@ export interface Harness extends Session { * `Harness.waitForIdle()`, and `Conversation.waitForIdle()`. Read-only viewers never do. */ resume(): void; + /** + * When this Session next needs a running process, from the scheduler's memory: `active` while some task can run now + * or is running, `at` for the earliest sleep. Answers also while scheduling is paused. A host that wakes the + * process itself, such as a Durable Object alarm, arms from it and calls `resume()` when woken; see + * `HarnessOptions.onWake`. + */ + wake(): Wake; /** Return the reserved root conversation, creating it with `agent` and `init` in one commit when absent. */ root( diff --git a/packages/durable/src/index.ts b/packages/durable/src/index.ts index 546c6ffa6..dcbe60fc1 100644 --- a/packages/durable/src/index.ts +++ b/packages/durable/src/index.ts @@ -114,6 +114,7 @@ export type { ToolHooks, ToolRegistration, UserInput, + Wake, Wrap, } from "./harness/types.ts"; export { UsageDoc, type UsageState } from "./harness/usage.ts"; @@ -175,6 +176,7 @@ export type { SessionDocFamilyToken, SessionDocToken, SessionEnd, + SetTimer, Storage, StorageWrite, StoredDocument, diff --git a/packages/durable/src/session/session.ts b/packages/durable/src/session/session.ts index 4bb06c85d..930b22ab3 100644 --- a/packages/durable/src/session/session.ts +++ b/packages/durable/src/session/session.ts @@ -11,6 +11,7 @@ import { } from "../documents.ts"; import { SessionFailed, StorageRequestError } from "../errors.ts"; import { idFromNumber } from "../ids.ts"; +import { hostTimers, type Timers } from "../timers.ts"; import type { CommitChange, CommitPublication, @@ -41,6 +42,7 @@ import type { SessionDocFamilyToken, SessionDocToken, SessionEnd, + SetTimer, Storage, StorageWrite, StoredDocument, @@ -63,15 +65,21 @@ import { import { type LoadedDocument, Transaction, type TransactionHost, type TransactionScope } from "./transaction.ts"; /** - * Open a Session kernel over one storage backend. `now` is the wall clock for task times; default `Date.now`. The first + * Open a Session kernel over one storage backend. `now` is the wall clock for task times; default `Date.now`. + * `setTimer` arms the timers that evict idle documents; default `setTimeout` (see `HarnessOptions.setTimer`). The first * error a Storage method throws fails the Session (`SessionFailed`); see `SessionImpl`. */ export function createSession( storage: Storage, - options?: { readonly now?: () => number; readonly documentRetentionMs?: number }, + options?: { readonly now?: () => number; readonly documentRetentionMs?: number; readonly setTimer?: SetTimer }, ): Session { const retention = options?.documentRetentionMs; - return new SessionImpl(storage, options?.now, retention === undefined ? undefined : () => retention); + return new SessionImpl( + storage, + options?.now, + retention === undefined ? undefined : () => retention, + options?.setTimer, + ); } /** Default document retention: ten minutes, as for kept contexts. */ @@ -90,9 +98,6 @@ 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. * @@ -118,10 +123,11 @@ export class SessionImpl implements Session { readonly #attachedDocuments = new Map(); readonly #clock: () => number; readonly #documentRetentionMs: () => number; + readonly #timers: Timers; /** 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 | undefined } | undefined; + #documentSweep: { readonly at: number; disarm: (() => void) | 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. */ + /** Set once the host declined a background timer: 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. */ @@ -153,10 +159,12 @@ export class SessionImpl implements Session { storage: Storage, now: () => number = Date.now, documentRetentionMs: () => number | undefined = () => undefined, + setTimer?: SetTimer, ) { // Waits race it; none has to. this.#failed.promise.catch(() => {}); this.#storage = new GuardedStorage(storage, this); + this.#timers = hostTimers(setTimer, (error) => this.report(error)); // 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 = () => { @@ -518,7 +526,7 @@ 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?.disarm?.(); this.#documentSweep = undefined; const listeners = [...this.#closeListeners]; this.#closeListeners.clear(); @@ -618,8 +626,8 @@ export class SessionImpl implements Session { /** * 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. + * expires, by a background timer (`HarnessOptions.setTimer`), which never keeps the process alive. Where the host + * declines it, as the default does in Cloudflare Workers, none is kept: the end of each commit evicts what expired. */ #scheduleDocumentSweep(): void { if (this.#closing !== undefined) return; @@ -627,9 +635,9 @@ export class SessionImpl implements Session { 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); + this.#documentSweep?.disarm?.(); // 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 | undefined }; + const sweep: { readonly at: number; disarm: (() => void) | undefined } = { at, disarm: undefined }; const run = (): void => { this.#sweepJobs++; this.#enqueue(async () => { @@ -647,16 +655,14 @@ export class SessionImpl implements Session { 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); + // A timer that fires early sweeps nothing and arms the next one. + const disarm = this.#timers.background(run, delay); + if (disarm === undefined) { this.#documentTimersUnavailable = true; this.#documentSweep = undefined; return; } - unref.call(timer); - sweep.timer = timer; + sweep.disarm = disarm; } /** On the line: evict the idle documents whose retention passed, oldest first, then schedule the next eviction. */ diff --git a/packages/durable/src/storage/memory.ts b/packages/durable/src/storage/memory.ts index d6e3b80c4..7f379cc29 100644 --- a/packages/durable/src/storage/memory.ts +++ b/packages/durable/src/storage/memory.ts @@ -247,7 +247,7 @@ export class MemoryStorage implements Storage { entryCommitSeqs: new Map(), tasks: new Map(), taskIds: [], - taskIdsByStatus: { pending: [], running: [], waiting: [], completing: [], terminal: [] }, + taskIdsByStatus: { pending: [], running: [], waiting: [], sleeping: [], completing: [], terminal: [] }, submissions: new Map(), submissionIds: [], submissionIdsByStatus: { queued: [], placed: [], done: [], unanswered: [] }, diff --git a/packages/durable/src/storage/sqlite/migrations.ts b/packages/durable/src/storage/sqlite/migrations.ts index bd2b6dd89..a5ee244ae 100644 --- a/packages/durable/src/storage/sqlite/migrations.ts +++ b/packages/durable/src/storage/sqlite/migrations.ts @@ -88,7 +88,32 @@ export function createSqliteMigrations(schema: SqliteSchema): readonly SqliteMig ) STRICT`, `CREATE INDEX ${indexes.documentRevisionsByKind} ON ${tables.documentRevisions} (document_id, kind, seq DESC)`, ]; - return [{ version: 1, statements: initialSchema }]; + // Version 2 adds the `sleeping` task status. SQLite cannot change a CHECK constraint in place, so the table is rebuilt. + const rebuiltTasks = `${tables.tasks}_v2`; + const sleepingStatus: readonly string[] = [ + `CREATE TABLE ${rebuiltTasks} ( + id INTEGER PRIMARY KEY, + conversation_id INTEGER NOT NULL, + kind TEXT NOT NULL, + status TEXT NOT NULL CHECK (status IN ('pending', 'running', 'waiting', 'sleeping', 'completing', 'terminal')), + abort_requested INTEGER NOT NULL CHECK (abort_requested IN (0, 1)), + background INTEGER NOT NULL CHECK (background IN (0, 1)), + record TEXT NOT NULL CHECK (json_valid(record)) + ) STRICT`, + `INSERT INTO ${rebuiltTasks} (id, conversation_id, kind, status, abort_requested, background, record) + SELECT id, conversation_id, kind, status, abort_requested, background, record FROM ${tables.tasks}`, + `DROP TABLE ${tables.tasks}`, + `ALTER TABLE ${rebuiltTasks} RENAME TO ${tables.tasks}`, + `CREATE INDEX ${indexes.tasksByStatus} ON ${tables.tasks} (status, id)`, + `CREATE INDEX ${indexes.tasksByConversation} ON ${tables.tasks} (conversation_id, id)`, + `CREATE INDEX ${indexes.tasksByKind} ON ${tables.tasks} (kind, id)`, + `CREATE INDEX ${indexes.tasksByAbortRequested} ON ${tables.tasks} (abort_requested, id)`, + `CREATE INDEX ${indexes.tasksByBackground} ON ${tables.tasks} (background, id)`, + ]; + return [ + { version: 1, statements: initialSchema }, + { version: 2, statements: sleepingStatus }, + ]; } export const SQLITE_MIGRATIONS: readonly SqliteMigration[] = createSqliteMigrations(DEFAULT_SQLITE_SCHEMA); diff --git a/packages/durable/src/testing/storage-conformance.ts b/packages/durable/src/testing/storage-conformance.ts index ad0f58aad..12cb75ac9 100644 --- a/packages/durable/src/testing/storage-conformance.ts +++ b/packages/durable/src/testing/storage-conformance.ts @@ -665,6 +665,25 @@ export function createStorageConformance(options: StorageConformanceOptions): re expect(await scan("terminal")).toEqual([terminal]); }), + createCase(options, "stores and scans sleeping tasks by status", async (storage) => { + const rootId = await createRoot(storage); + const sleepingId = await storage.mintId>(); + const sleeping: StoredTask = { + ...pendingTask(sleepingId, rootId), + state: { status: "sleeping", checkpoint: { phase: "next" }, until: 61_000 }, + memos: { kept: true }, + }; + await storage.commit([{ type: "task", value: sleeping }], context); + expect(await storage.task(sleepingId, context)).toEqual(sleeping); + const scan = async (status: StoredTask["state"]["status"]) => + (await storage.scanTasks({ status }, 10, undefined, context)).items; + expect(await scan("sleeping")).toEqual([sleeping]); + const running: StoredTask = { ...sleeping, state: { status: "running", checkpoint: { phase: "next" } } }; + await storage.commit([{ type: "task", value: running }], context); + expect(await scan("sleeping")).toEqual([]); + expect(await scan("running")).toEqual([running]); + }), + createCase( options, "indexes request IDs per conversation and replaces complete submission records", diff --git a/packages/durable/src/timers.ts b/packages/durable/src/timers.ts new file mode 100644 index 000000000..de49d2284 --- /dev/null +++ b/packages/durable/src/timers.ts @@ -0,0 +1,55 @@ +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; + +/** + * 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. + */ +export const nativeTimer: SetTimer = (callback, delayMs, options) => { + const handle = setTimeout(callback, Math.min(Math.max(0, delayMs), MAX_TIMER_DELAY)); + if (options.background) { + const unref = (handle as { unref?: unknown }).unref; + if (typeof unref !== "function") { + clearTimeout(handle); + return undefined; + } + unref.call(handle); + } + return () => clearTimeout(handle); +}; + +/** Timers the Session and Harness arm through the host's `setTimer`. */ +export type Timers = { + /** A timer that paces or resumes work. A host that declines it is reported, and the timer never fires. */ + foreground(callback: () => void, delayMs: number): () => void; + /** A timer that only frees memory; `undefined` when the host declines it. */ + background(callback: () => void, delayMs: number): (() => void) | undefined; +}; + +/** Wrap a host `setTimer`: a throw or a declined foreground timer is reported, and arms nothing. */ +export function hostTimers(setTimer: SetTimer | undefined, report: (error: unknown) => void): Timers { + const arm = setTimer ?? nativeTimer; + return { + foreground: (callback, delayMs) => { + try { + const disarm = arm(callback, delayMs, { background: false }); + if (disarm !== undefined) return disarm; + report(new Error("HarnessOptions.setTimer declined a timer that paces work")); + } catch (error) { + report(error); + } + return () => {}; + }, + background: (callback, delayMs) => { + try { + return arm(callback, delayMs, { background: true }); + } catch (error) { + report(error); + return undefined; + } + }, + }; +} diff --git a/packages/durable/src/types.ts b/packages/durable/src/types.ts index c4d90d77b..11819eb59 100644 --- a/packages/durable/src/types.ts +++ b/packages/durable/src/types.ts @@ -133,16 +133,32 @@ export type TaskDocFamilyToken = DocF DocFamilyDefinition & { readonly scope: "task" } >; +/** + * A host timer, given as `HarnessOptions.setTimer`: call `callback` once, about `delayMs` from now, and return a + * function that disarms it. `background` timers only free memory, and a host may decline one by returning `undefined`, + * for example where a pending timer keeps the process from hibernating; the Harness then frees that memory at later + * commits. Other timers pace or resume work and must be armed. A timer may fire early or late: callers check their + * clock again. + */ +export type SetTimer = ( + callback: () => void, + delayMs: number, + options: { readonly background: boolean }, +) => (() => void) | undefined; + /** Live task record reserved by one invocation. */ export type RunningTask = TaskRecord & { readonly state: Extract, { readonly status: "running" }>; }; /** - * Next state a task commits for itself: a replacement checkpoint, a wait, or its outcome. A returned `terminal` state is + * Next state a task commits for itself: a replacement checkpoint, a wait on tasks, a sleep, or its outcome. A returned `terminal` state is * stored as `completing` while ordinary owned work below the task is live (spec §5.5). */ -export type NextTaskState = Extract, { readonly status: "running" | "waiting" | "terminal" }>; +export type NextTaskState = Extract< + TaskState, + { readonly status: "running" | "waiting" | "sleeping" | "terminal" } +>; /** * Runs one checkpoint phase. It must commit a changed checkpoint or a terminal outcome through `runtime.commit()`; @@ -234,10 +250,21 @@ export interface TaskRuntime extends DocumentObserver context(conversationId: ConversationId, context: Context, options?: { readonly at?: EntryId }): Promise; /** The Harness clock. */ now(): number; + /** + * Call `callback` once after about `delayMs` on the Harness's timer (`HarnessOptions.setTimer`), for pacing work + * inside a phase, such as throttled progress commits. Not durable: the invocation's end disarms it, and after its end + * nothing is armed. Returns a function that disarms it. Wait for time between phases with `sleepUntil()`. + */ + setTimer(callback: () => void, delayMs: number): () => void; /** Forward a non-fatal failure to `HarnessOptions.onReport`. */ report(error: unknown): void; - /** Resolve once the Harness clock reaches `until`; rejects when the invocation or `context` is cancelled. */ - sleep(until: number, context: Context): Promise; + /** + * Commit a sleep: the task holds no invocation until the Harness clock reaches `until`, then resumes at `checkpoint`. + * Resolves once the sleep is stored, not when the time comes; the phase returns right after. Returning + * `{ status: "sleeping", until, checkpoint }` from `commit()` is the same sleep, for one that must commit together + * with other writes. + */ + sleepUntil(until: number, checkpoint: S, context: Context): Promise; } /** Executable durable state machine definition, registered in the registry by `name`. */ @@ -523,6 +550,16 @@ export type TaskState = readonly policy: JoinPolicy; readonly outcome?: never; } + | { + /** + * Parked without an invocation until the Harness clock reaches `until`, in milliseconds on the + * `HarnessOptions.now` clock; then resumes at `checkpoint`. An abort mark does not wait for it. + */ + readonly status: "sleeping"; + readonly checkpoint: S; + readonly until: number; + readonly outcome?: never; + } | { /** Outcome decided; becomes terminal once no ordinary owned work below is live. Runs no more code. */ readonly status: "completing"; @@ -574,7 +611,10 @@ type TaskRecordBase = { export type TaskRecord = TaskRecordBase & ( | { - readonly state: Extract, { readonly status: "pending" | "running" | "waiting" }>; + readonly state: Extract< + TaskState, + { readonly status: "pending" | "running" | "waiting" | "sleeping" } + >; /** Small first-writer-wins values retained while the task can run. */ readonly memos?: Readonly>; } diff --git a/packages/durable/test/examples/13-recovery.ts b/packages/durable/test/examples/13-recovery.ts index 2b3cc4116..2b8c23ca9 100644 --- a/packages/durable/test/examples/13-recovery.ts +++ b/packages/durable/test/examples/13-recovery.ts @@ -34,17 +34,18 @@ const Ticker = defineTask<{ to: number }, { phase: "tick"; n: number }, string>( console.log(`tick ${n}`); } reachedTick(n); - // Save the outcome: the next tick, or the final result. - await runtime.commit( - () => - n === task.input.to - ? { status: "terminal", outcome: { status: "completed", result: `counted to ${n}` } } - : { status: "running", checkpoint: { phase: "tick", n: n + 1 } }, - taskContext, - ); - // Wait a little between ticks. Closing the Harness cancels this wait; - // the checkpoint saved above is where the next Harness continues. - await runtime.sleep(Date.now() + 50, taskContext); + // Save the outcome: the final result, or the next tick after a short + // sleep. The sleep is stored with the checkpoint, so the task holds no + // invocation meanwhile, and a Harness opened later continues it once + // its time has come. + if (n === task.input.to) { + await runtime.commit( + () => ({ status: "terminal", outcome: { status: "completed", result: `counted to ${n}` } }), + taskContext, + ); + return; + } + await runtime.sleepUntil(runtime.now() + 50, { phase: "tick", n: n + 1 }, taskContext); }, }, abort: async (_task, runtime, taskContext) => { diff --git a/packages/durable/test/examples/24-child-tasks.ts b/packages/durable/test/examples/24-child-tasks.ts index e1b49defe..84c747c85 100644 --- a/packages/durable/test/examples/24-child-tasks.ts +++ b/packages/durable/test/examples/24-child-tasks.ts @@ -26,10 +26,11 @@ const charged = new Set(); type PaymentInput = { card: string }; type PaymentResult = { card: string }; -const Payment = defineTask({ +type PaymentState = { phase: "charge" } | { phase: "confirm" }; +const Payment = defineTask({ name: "example.payment", version: 1, - initial: () => ({ phase: "charge", at: Date.now() + 100 }), + initial: () => ({ phase: "charge" }), phases: { charge: async (task, runtime, taskContext) => { const card = task.input.card; @@ -39,7 +40,11 @@ const Payment = defineTask { + const card = task.input.card; await runtime.commit( () => ({ status: "terminal", outcome: { status: "completed", result: { card } } }), taskContext, diff --git a/packages/durable/test/harness-generation-recovery.test.ts b/packages/durable/test/harness-generation-recovery.test.ts index fed804c5e..6b386f049 100644 --- a/packages/durable/test/harness-generation-recovery.test.ts +++ b/packages/durable/test/harness-generation-recovery.test.ts @@ -1,6 +1,7 @@ import { mkdtemp, rm } from "node:fs/promises"; import { tmpdir } from "node:os"; import { join } from "node:path"; +import { DatabaseSync } from "node:sqlite"; import { type AssistantMessage, createModels, fauxAssistantMessage, type Message } from "@earendil-works/pi-ai"; import { AssistantEntry, type Harness, LiveDoc, type TaskId } from "@earendil-works/pi-durable"; import { afterEach, describe, expect, it } from "vitest"; @@ -182,7 +183,16 @@ describe("generation recovery", () => { await opened.harness.close(context); opened = await open(path, setup); - expect(await checkpoint(opened.harness, taskId)).toEqual({ phase: "retry", attempt: 1, until: 61_000 }); + expect((await opened.harness.getTask(taskId, context))?.state).toEqual({ + status: "sleeping", + checkpoint: { phase: "retry", attempt: 1, until: 61_000 }, + until: 61_000, + }); + expect(opened.harness.wake()).toEqual({ active: false, at: 61_000 }); + // An early wake retries nothing. + opened.harness.resume(); + await new Promise((resolve) => setTimeout(resolve, 20)); + expect(setup.faux.state.callCount).toBe(1); expect(await opened.harness.snapshot(LiveDoc, opened.root.id, context)).toEqual({ run: { taskId, inputs: [id] }, generation: { attempt: 1, retry: { at: 61_000, error: "503 Service Unavailable" } }, @@ -198,6 +208,49 @@ describe("generation recovery", () => { await opened.harness.close(context); }); + it("sleeps again for a retry stored as a running phase by an earlier version", async () => { + const path = await sqlitePath(); + const setup = chatSetup(); + let now = 1_000; + setup.now = () => now; + setup.faux.setResponses([ + fauxAssistantMessage([], { stopReason: "error", errorMessage: "503 Service Unavailable" }), + fauxAssistantMessage("recovered"), + ]); + let opened = await open(path, setup); + opened.harness.resume(); + setup.settings.retry = { enabled: true, maxRetries: 2, baseDelayMs: 60_000 }; + const id = (await opened.root.submit({ type: "input", content: "hi" }, context)).id; + await waitFor( + async () => + (await opened.harness.getTask(await runTaskId(opened.harness), context))?.state.status === "sleeping", + ); + const taskId = await runTaskId(opened.harness); + await opened.harness.close(context); + // Earlier versions slept inside a running `retry` phase. + const database = new DatabaseSync(path); + database + .prepare( + "UPDATE tasks SET status = 'running', record = json_set(json_remove(record, '$.state.until'), '$.state.status', 'running') WHERE id = ?", + ) + .run(taskId); + database.close(); + + opened = await open(path, setup); + opened.harness.resume(); + await waitFor(async () => (await opened.harness.getTask(taskId, context))?.state.status === "sleeping"); + expect((await opened.harness.getTask(taskId, context))?.state).toMatchObject({ + status: "sleeping", + until: 61_000, + }); + expect(setup.faux.state.callCount).toBe(1); + now = 61_000; + opened.harness.resume(); + expect((await (await opened.harness.submission(id, context))!.wait(context)).status).toBe("done"); + expect(setup.faux.state.callCount).toBe(2); + await opened.harness.close(context); + }); + it("resumes polling a deferred response after reopen", async () => { const path = await sqlitePath(); const setup = chatSetup({ deferred: { pollAfterMs: 60_000 } }); @@ -217,6 +270,11 @@ describe("generation recovery", () => { opened = await open(path, setup); expect(await checkpoint(opened.harness, taskId)).toMatchObject({ phase: "poll", attempt: 1, pollAt: 61_000 }); + expect((await opened.harness.getTask(taskId, context))?.state).toMatchObject({ + status: "sleeping", + until: 61_000, + }); + expect(opened.harness.wake()).toEqual({ active: false, at: 61_000 }); now = 61_000; opened.harness.resume(); const settled = await (await opened.harness.submission(id, context))!.wait(context); @@ -230,6 +288,8 @@ describe("generation recovery", () => { it("fails no_model when the pinned model is gone after reopen, in request and in poll", async () => { const path = await sqlitePath(); const setup = chatSetup({ deferred: { pollAfterMs: 60_000 } }); + let now = 1_000; + setup.now = () => now; const busy = unanswered(); setup.faux.setResponses([busy.step, fauxAssistantMessage("deferred")]); let opened = await open(path, setup); @@ -258,6 +318,8 @@ describe("generation recovery", () => { }); await opened.harness.close(context); + // The poll sleeps until it is due; then the model is resolved. + now = 61_000; opened = await open(path, empty); opened.harness.resume(); expect(await (await opened.harness.submission(polling, context))!.wait(context)).toMatchObject({ diff --git a/packages/durable/test/harness-sleep.test.ts b/packages/durable/test/harness-sleep.test.ts new file mode 100644 index 000000000..bd98ec2fe --- /dev/null +++ b/packages/durable/test/harness-sleep.test.ts @@ -0,0 +1,330 @@ +import { mkdtemp, rm } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { defineTask, MemoryStorage, type Storage, type TaskId, type Wake } from "@earendil-works/pi-durable"; +import { afterEach, describe, expect, it } from "vitest"; +import { openNodeSqliteStorage } from "../src/storage/sqlite/node.ts"; +import { context, flush } from "./session-support.ts"; +import { aborted, abortedWith, completed, eventually, openTasks } from "./task-support.ts"; + +const directories = new Set(); + +afterEach(async () => { + for (const directory of directories) await rm(directory, { recursive: true, force: true }); + directories.clear(); +}); + +async function sqlitePath(): Promise { + const directory = await mkdtemp(join(tmpdir(), "pi-durable-sleep-")); + directories.add(directory); + return join(directory, "session.sqlite"); +} + +type NapState = { phase: "start" } | { phase: "after"; slept: number }; + +/** Sleeps `delay` ms on the Harness clock, then completes with the time it resumed at. */ +function napTask(delay: number, runs: { after: number } = { after: 0 }) { + return defineTask({ + name: "test.nap", + version: 1, + initial: () => ({ phase: "start" }), + phases: { + start: async (_task, runtime, ctx) => { + await runtime.sleepUntil(runtime.now() + delay, { phase: "after", slept: delay }, ctx); + }, + after: async (_task, runtime, ctx) => { + runs.after++; + const resumed = runtime.now(); + await runtime.commit(() => completed(resumed), ctx); + }, + }, + abort: async (_task, runtime, ctx) => { + await runtime.commit(() => abortedWith("test"), ctx); + }, + }); +} + +async function startNap(harness: Awaited>["harness"], task: ReturnType) { + const root = await harness.root(context); + return root.commit((tx) => tx.createTask(task, null, { ownership: { kind: "conversation" } }), context); +} + +async function stateOf(harness: Awaited>["harness"], id: TaskId) { + return (await harness.getTask(id, context))?.state; +} + +/** Let microtasks and a few timer turns run. */ +async function settle(): Promise { + await flush(); + await new Promise((resolve) => setTimeout(resolve, 20)); + await flush(); +} + +describe("sleeping tasks", () => { + it("sleeps without an invocation and resumes at its checkpoint once due", async () => { + let clock = 1_000; + const wakes: Wake[] = []; + const runs = { after: 0 }; + const Nap = napTask(1_000, runs); + const { harness } = await openTasks(new MemoryStorage(), [Nap], { + now: () => clock, + onWake: (wake) => wakes.push(wake), + }); + const graph = await harness.taskGraph(context); + const id = await startNap(harness, Nap); + harness.resume(); + await eventually(async () => (await stateOf(harness, id))?.status === "sleeping"); + expect(await stateOf(harness, id)).toEqual({ + status: "sleeping", + checkpoint: { phase: "after", slept: 1_000 }, + until: 2_000, + }); + const inspection = await harness.inspect(context); + expect(inspection.tasks.find((task) => task.record.id === id)?.state).toEqual({ kind: "sleeping", until: 2_000 }); + expect(graph.value.tasks[String(id)]?.state).toEqual({ status: "sleeping", phase: "after", until: 2_000 }); + expect(harness.wake()).toEqual({ active: false, at: 2_000 }); + await settle(); + expect(wakes.at(-1)).toEqual({ active: false, at: 2_000 }); + + // Early and repeated wakes run nothing. + clock = 1_999; + harness.resume(); + harness.resume(); + await settle(); + expect(runs.after).toBe(0); + expect((await stateOf(harness, id))?.status).toBe("sleeping"); + + clock = 2_000; + harness.resume(); + harness.resume(); + const receipt = await harness.waitForTask(id, context); + expect(receipt.state.outcome).toEqual({ status: "completed", result: 2_000 }); + expect(runs.after).toBe(1); + await settle(); + expect(harness.wake()).toEqual({ active: false }); + expect(wakes.at(-1)).toEqual({ active: false }); + // Only changes: idle at open, running, asleep until 2000, running again, idle. + for (let index = 1; index < wakes.length; index++) expect(wakes[index]).not.toEqual(wakes[index - 1]); + expect(wakes[0]).toEqual({ active: false }); + expect(wakes).toContainEqual({ active: true }); + expect(wakes).toContainEqual({ active: false, at: 2_000 }); + graph.dispose(); + await harness.close(context); + }); + + it("keeps a sleep across reopen", async () => { + const path = await sqlitePath(); + let clock = 1_000; + const Nap = napTask(60_000); + const open = async (storage: Storage) => openTasks(storage, [Nap], { now: () => clock, onWake: () => {} }); + let { harness } = await open(await openNodeSqliteStorage(path)); + const id = await startNap(harness, Nap); + harness.resume(); + await eventually(async () => (await stateOf(harness, id))?.status === "sleeping"); + await harness.close(context); + + ({ harness } = await open(await openNodeSqliteStorage(path))); + // Known from storage before scheduling starts. + expect(harness.wake()).toEqual({ active: false, at: 61_000 }); + harness.resume(); + await settle(); + expect((await stateOf(harness, id))?.status).toBe("sleeping"); + await harness.close(context); + + clock = 61_000; + ({ harness } = await open(await openNodeSqliteStorage(path))); + expect(harness.wake()).toEqual({ active: true }); + harness.resume(); + expect((await harness.waitForTask(id, context)).state.outcome).toEqual({ status: "completed", result: 61_000 }); + await harness.close(context); + }); + + it("resumes a due sleep with a timer when there is no onWake", async () => { + const Nap = napTask(30); + const { harness } = await openTasks(new MemoryStorage(), [Nap]); + const id = await startNap(harness, Nap); + const started = Date.now(); + // Only this call: the timer wakes the task. + harness.resume(); + const receipt = await harness.waitForTask(id, context); + expect(receipt.state.outcome.status).toBe("completed"); + expect(Date.now() - started).toBeGreaterThanOrEqual(25); + await harness.close(context); + }); + + it("arms its timers through setTimer: a sleep's in the foreground, a declined background one is not needed", async () => { + let clock = 1_000; + const armed: { readonly delayMs: number; readonly background: boolean; readonly callback: () => void }[] = []; + const Nap = napTask(500); + const { harness } = await openTasks(new MemoryStorage(), [Nap], { + now: () => clock, + setTimer: (callback, delayMs, options) => { + armed.push({ delayMs, background: options.background, callback }); + // Declines background timers, as a host that must hibernate would. + return options.background ? undefined : () => {}; + }, + }); + const id = await startNap(harness, Nap); + harness.resume(); + await eventually(() => armed.some((timer) => !timer.background && timer.delayMs === 500)); + expect((await stateOf(harness, id))?.status).toBe("sleeping"); + clock = 1_500; + armed.find((timer) => !timer.background && timer.delayMs === 500)!.callback(); + expect((await harness.waitForTask(id, context)).state.outcome).toEqual({ status: "completed", result: 1_500 }); + await harness.close(context); + }); + + it("disarms an invocation's timers when it ends", async () => { + const disarmed: number[] = []; + const fired: string[] = []; + const Timed = defineTask({ + name: "test.timed", + version: 1, + initial: () => ({ phase: "run" }), + phases: { + run: async (_task, runtime, ctx) => { + runtime.setTimer(() => fired.push("late"), 60_000); + await runtime.commit(() => completed(null), ctx); + }, + }, + abort: async () => {}, + }); + let next = 0; + const { harness } = await openTasks(new MemoryStorage(), [Timed], { + setTimer: (_callback, delayMs) => { + const id = next++; + return delayMs === 60_000 ? () => disarmed.push(id) : () => {}; + }, + }); + const root = await harness.root(context); + const id = await root.commit( + (tx) => tx.createTask(Timed, null, { ownership: { kind: "conversation" } }), + context, + ); + harness.resume(); + await harness.waitForTask(id, context); + await eventually(() => disarmed.length === 1); + expect(fired).toEqual([]); + await harness.close(context); + }); + + it("keeps no timer with onWake", async () => { + const Nap = napTask(10); + const { harness } = await openTasks(new MemoryStorage(), [Nap], { onWake: () => {} }); + const id = await startNap(harness, Nap); + harness.resume(); + await new Promise((resolve) => setTimeout(resolve, 60)); + expect((await stateOf(harness, id))?.status).toBe("sleeping"); + expect(harness.wake()).toEqual({ active: true }); + harness.resume(); + expect((await harness.waitForTask(id, context)).state.outcome.status).toBe("completed"); + await harness.close(context); + }); + + it("runs the abort handler of a sleeping task without waiting for its time", async () => { + const clock = 1_000; + const Nap = napTask(60_000); + const { harness } = await openTasks(new MemoryStorage(), [Nap], { now: () => clock, onWake: () => {} }); + const id = await startNap(harness, Nap); + harness.resume(); + await eventually(async () => (await stateOf(harness, id))?.status === "sleeping"); + await harness.abortTask(id, context); + expect((await harness.waitForTask(id, context)).state.outcome).toEqual({ status: "aborted", reason: "test" }); + await harness.close(context); + }); + + it("rejects a sleep until something that is not a time, and a sleep in an abort handler", async () => { + const errors: string[] = []; + const Invalid = defineTask({ + name: "test.invalid-sleep", + version: 1, + initial: () => ({ phase: "run" }), + phases: { + run: async (_task, runtime, ctx) => { + for (const until of [Number.NaN, -1, Number.POSITIVE_INFINITY]) { + await runtime + .sleepUntil(until, { phase: "next" }, ctx) + .catch((error: unknown) => errors.push((error as Error).message)); + } + await runtime.commit(() => completed(null), ctx); + }, + next: async () => {}, + }, + abort: async () => {}, + }); + const Aborting = defineTask({ + name: "test.sleep-in-abort", + version: 1, + initial: () => ({ phase: "run" }), + phases: { run: (_task, runtime) => aborted(runtime.signal) }, + abort: async (_task, runtime, ctx) => { + await runtime + .sleepUntil(runtime.now() + 10, { phase: "run" }, ctx) + .catch((error: unknown) => errors.push((error as Error).message)); + await runtime.commit(() => abortedWith("test"), ctx); + }, + }); + const { harness } = await openTasks(new MemoryStorage(), [Invalid, Aborting]); + const root = await harness.root(context); + const id = await root.commit( + (tx) => tx.createTask(Invalid, null, { ownership: { kind: "conversation" } }), + context, + ); + const aborting = await root.commit( + (tx) => tx.createTask(Aborting, null, { ownership: { kind: "conversation" } }), + context, + ); + harness.resume(); + expect((await harness.waitForTask(id, context)).state.outcome.status).toBe("completed"); + await harness.abortTask(aborting, context); + expect((await harness.waitForTask(aborting, context)).state.outcome.status).toBe("aborted"); + expect(errors).toEqual([ + `Task ${id} cannot sleep until NaN: not a time in milliseconds`, + `Task ${id} cannot sleep until -1: not a time in milliseconds`, + `Task ${id} cannot sleep until Infinity: not a time in milliseconds`, + `Abort handler of task ${aborting} cannot sleep`, + ]); + await harness.close(context); + }); + + it("does not count a task without a definition as active", async () => { + const Nap = napTask(10); + const Missing = napTask(10); + const { harness } = await openTasks(new MemoryStorage(), []); + const root = await harness.root(context); + // Created through a token the registry does not hold: nothing can run it. + await root.commit((tx) => tx.createTask(Missing, null, { ownership: { kind: "conversation" } }), context); + expect(harness.wake()).toEqual({ active: false }); + harness.resume(); + await settle(); + expect(harness.wake()).toEqual({ active: false }); + void Nap; + await harness.close(context); + }); + + it("reports a throwing onWake and keeps calling it for later changes", async () => { + let clock = 1_000; + const wakes: Wake[] = []; + const failure = new Error("alarm failed"); + const Nap = napTask(1_000); + const { harness, reports } = await openTasks(new MemoryStorage(), [Nap], { + now: () => clock, + onWake: (wake) => { + wakes.push(wake); + if (wake.at !== undefined) throw failure; + }, + }); + const id = await startNap(harness, Nap); + harness.resume(); + await eventually(async () => (await stateOf(harness, id))?.status === "sleeping"); + await settle(); + expect(reports.length).toBeGreaterThan(0); + expect(reports.every((error) => error === failure)).toBe(true); + clock = 2_000; + harness.resume(); + await harness.waitForTask(id, context); + await settle(); + expect(wakes.at(-1)).toEqual({ active: false }); + await harness.close(context); + }); +}); diff --git a/packages/durable/test/harness-tasks.test.ts b/packages/durable/test/harness-tasks.test.ts index 55aabb697..8e9c51ff4 100644 --- a/packages/durable/test/harness-tasks.test.ts +++ b/packages/durable/test/harness-tasks.test.ts @@ -567,7 +567,7 @@ describe("task runtime", () => { await expect(runtime.commit(() => {}, context)).rejects.toThrow("invocation has ended"); await expect(runtime.memo("x", context)).rejects.toThrow("invocation has ended"); await expect(runtime.memo("x", 1, context)).rejects.toThrow("invocation has ended"); - await expect(runtime.sleep(0, context)).rejects.toThrow("invocation has ended"); + await expect(runtime.sleepUntil(0, { phase: "run" }, context)).rejects.toThrow("invocation has ended"); await expect(runtime.watchDoc(Notes, context)).rejects.toThrow("invocation has ended"); // The handler's own context is cancelled by now; the ended invocation still wins. await expect(runtime.agent(handlerContext!)).rejects.toThrow("invocation has ended"); @@ -687,58 +687,6 @@ describe("task runtime", () => { expect((await harness.waitForTask(id, context)).state.outcome).toMatchObject({ status: "faulted" }); await harness.close(context); }); - - it("sleeps until the Harness clock reaches the deadline, rechecking after each timer", async () => { - let clock = 1_000; - const woke = deferred(); - const Sleeper = oneStep("test.sleeper", async (_task, runtime, ctx) => { - await runtime.sleep(900, ctx); - await runtime.sleep(1_005, ctx); - woke.resolve(); - await runtime.commit(() => completed(null), ctx); - }); - const { harness, root } = await openRoot([Sleeper], { now: () => clock }); - const id = await start(root, Sleeper); - harness.resume(); - // The clock stands still, so real timers keep firing without waking the task. - await new Promise((resolve) => setTimeout(resolve, 30)); - expect(await settled(woke.promise)).toBe(false); - clock = 1_005; - await harness.waitForTask(id, context); - await harness.close(context); - }); - - it("rejects a sleep when the invocation is signalled or the sleep's own context is cancelled", async () => { - const results: string[] = []; - const sleeping = deferred(); - const Signalled = oneStep("test.sleep-signalled", async (_task, runtime, ctx) => { - sleeping.resolve(); - await runtime.sleep(Date.now() + 60_000, ctx).catch((error: unknown) => { - results.push(`signalled:${(error as Error).name}`); - throw error; - }); - }); - const Cancelled = oneStep("test.sleep-cancelled", async (_task, runtime, ctx) => { - const controller = new AbortController(); - const sleeping = runtime.sleep(Date.now() + 60_000, withSignal(controller.signal)); - controller.abort(new Error("stop sleeping")); - await sleeping.catch((error: unknown) => results.push(`cancelled:${(error as Error).message}`)); - await runtime.commit(() => completed(null), ctx); - }); - const { harness, root } = await openRoot([Signalled, Cancelled]); - const signalled = await start(root, Signalled); - const cancelled = await start(root, Cancelled); - harness.resume(); - await harness.waitForTask(cancelled, context); - await sleeping.promise; - await harness.abortTask(signalled, context); - expect((await harness.waitForTask(signalled, context)).state.outcome).toEqual({ - status: "aborted", - reason: "test", - }); - expect(results).toEqual(["cancelled:stop sleeping", "signalled:AbortError"]); - await harness.close(context); - }); }); describe("task scheduling", () => { diff --git a/packages/durable/test/sqlite-cloudflare.test.ts b/packages/durable/test/sqlite-cloudflare.test.ts index 4c64772db..0bf9c101d 100644 --- a/packages/durable/test/sqlite-cloudflare.test.ts +++ b/packages/durable/test/sqlite-cloudflare.test.ts @@ -15,7 +15,7 @@ describe("Durable Object SQLite adapter", () => { it("keeps schema names unprefixed by default", async () => { const sqlite = new DatabaseSync(":memory:"); await openDurableObjectSqliteStorage(durableObjectStorage(sqlite)); - expect(sqlite.prepare("SELECT version FROM durable_schema").get()).toEqual({ version: 1 }); + expect(sqlite.prepare("SELECT version FROM durable_schema").get()).toEqual({ version: 2 }); expect(sqlite.prepare("SELECT name FROM sqlite_master WHERE name = 'pi_durable_schema'").get()).toBeUndefined(); }); diff --git a/packages/durable/test/sqlite-migrations.test.ts b/packages/durable/test/sqlite-migrations.test.ts index a1936f626..0006e8a25 100644 --- a/packages/durable/test/sqlite-migrations.test.ts +++ b/packages/durable/test/sqlite-migrations.test.ts @@ -44,6 +44,61 @@ describe("durable SQLite migrations", () => { } }); + it("migrates a version 1 database to store sleeping tasks, keeping its tasks and indexes", async () => { + const path = await databasePath(); + const database = await openNodeSqliteDatabase(path); + await applySqliteMigrations(database, SQLITE_MIGRATIONS.slice(0, 1)); + const waiting = { + id: 5, + conversationId: 1, + kind: "test.task", + version: 1, + input: null, + background: false, + abortRequested: false, + state: { status: "waiting", checkpoint: { phase: "run" }, on: [], policy: "allSettled" }, + }; + await database.run( + "INSERT INTO tasks (id, conversation_id, kind, status, abort_requested, background, record) VALUES (5, 1, 'test.task', 'waiting', 0, 0, ?)", + JSON.stringify(waiting), + ); + await expect( + database.run( + "INSERT INTO tasks (id, conversation_id, kind, status, abort_requested, background, record) VALUES (6, 1, 'test.task', 'sleeping', 0, 0, '{}')", + ), + ).rejects.toThrow(); + await database.close(); + + const storage = await openNodeSqliteStorage(path); + await storage.close(BACKGROUND_CONTEXT); + const migrated = await openNodeSqliteDatabase(path); + try { + expect(await migrated.get("SELECT version FROM durable_schema WHERE singleton = 1")).toEqual({ version: 2 }); + expect(await migrated.get("SELECT status, record FROM tasks WHERE id = 5")).toEqual({ + status: "waiting", + record: JSON.stringify(waiting), + }); + await migrated.run( + "INSERT INTO tasks (id, conversation_id, kind, status, abort_requested, background, record) VALUES (6, 1, 'test.task', 'sleeping', 0, 0, '{}')", + ); + const indexes = await migrated.all<{ name: string }>( + "SELECT name FROM sqlite_schema WHERE type = 'index' AND tbl_name = 'tasks' ORDER BY name", + ); + expect(indexes.map(({ name }) => name)).toEqual([ + "tasks_by_abort_requested", + "tasks_by_background", + "tasks_by_conversation", + "tasks_by_kind", + "tasks_by_status", + ]); + expect(await migrated.get("SELECT count(*) AS count FROM sqlite_schema WHERE name = 'tasks_v2'")).toEqual({ + count: 0, + }); + } finally { + await migrated.close(); + } + }); + it("rejects a database newer than the portable core", async () => { const path = await databasePath(); const database = await openNodeSqliteDatabase(path); diff --git a/packages/durable/test/task-support.ts b/packages/durable/test/task-support.ts index 403fc40d7..fc2006b2b 100644 --- a/packages/durable/test/task-support.ts +++ b/packages/durable/test/task-support.ts @@ -4,6 +4,7 @@ import { createRegistry, defineExtension, Harness, + type HarnessOptions, type HarnessSettings, type Registry, type RegistryReader, @@ -63,7 +64,13 @@ export async function settled(promise: Promise): Promise { export async function openTasks( storage: Storage, tasks: readonly AnyTask[], - options: { readonly registry?: Registry; readonly now?: () => number; readonly settings?: HarnessSettings } = {}, + options: { + readonly registry?: Registry; + readonly now?: () => number; + readonly settings?: HarnessSettings; + readonly onWake?: HarnessOptions["onWake"]; + readonly setTimer?: HarnessOptions["setTimer"]; + } = {}, ): Promise<{ readonly harness: Harness; readonly registry: Registry; readonly reports: unknown[] }> { const registry = options.registry ?? createRegistry(); if (tasks.length > 0) registry.install(defineExtension({ name: "tasks", tasks })); @@ -76,6 +83,8 @@ export async function openTasks( onReport: (error) => reports.push(error), ...(options.now === undefined ? {} : { now: options.now }), ...(options.settings === undefined ? {} : { settings: options.settings }), + ...(options.onWake === undefined ? {} : { onWake: options.onWake }), + ...(options.setTimer === undefined ? {} : { setTimer: options.setTimer }), }, context, );