From 35dfa757378bd7a8570670e99c2687bdd1039a06 Mon Sep 17 00:00:00 2001 From: Mario Zechner Date: Fri, 30 Oct 2026 23:17:56 +0000 Subject: [PATCH] fix(durable): idle checks and wake() without per-commit scans After every commit that changed a task, the scheduler re-checked every conversation with a kept context range (every conversation read in the last contextRetentionMs) by walking its subtree, scanned them again for the next expiry, and wake() scanned every live task. CPU per run grew linearly with the number of conversations. Now a commit re-checks only the conversations above tasks that became live or ended (collected before the sweep; a chain load that resolves an unknown for a live unloaded task falls back to a full check), idle context ranges expire from an index ordered by idleSince (re-sorted when the clock goes back; zero retention drops all idle ranges; reconcile passes still expire for hosts that decline timers), and wake() takes the earliest sleep from the sleeping tasks and scans live tasks only while nothing runs. One Session, one-tool runs, no storage latency: 2048 conversations 100 -> 805 runs/s, 4096: 46 -> 772; local SQLite at 2048: 88 -> 459; CPU per run flat at about 1.6 ms. --- packages/durable/CHANGELOG.md | 1 + packages/durable/src/harness/harness.ts | 8 + packages/durable/src/harness/scheduler.ts | 293 ++++++--- .../durable/test/harness-idle-scopes.test.ts | 591 ++++++++++++++++++ 4 files changed, 821 insertions(+), 72 deletions(-) create mode 100644 packages/durable/test/harness-idle-scopes.test.ts diff --git a/packages/durable/CHANGELOG.md b/packages/durable/CHANGELOG.md index 96d53480d..009e0367f 100644 --- a/packages/durable/CHANGELOG.md +++ b/packages/durable/CHANGELOG.md @@ -163,6 +163,7 @@ - A tool slot's `running` status and a nested call's final arguments, and a generation's attempt, are stored with the next commit, like streamed progress: a run with one tool call stores 12 commits instead of 14. After a crash a slot may show `pending` while its task's checkpoint is at execution; the rerun sets it again. - A wait whose commit is still being stored when the Harness closes ends with the closed error; the commit is still stored and published before close completes. - Cancellation is checked when a job starts, which can be earlier than before: an invocation-bound submit queued behind an abort mark that is not stored yet lands, and the abort's cascade then aborts its run. +- The scheduler's per-commit work no longer grows with the number of conversations: after a commit it re-checks idleness only for the conversations above the tasks that became live or ended, expires idle context ranges from an expiry-ordered index, and `wake()` takes the earliest sleep from the sleeping tasks and scans live tasks only while nothing runs. Before, every commit re-walked every conversation read in the last `contextRetentionMs` (default 10 minutes). One Session with 2048 conversations went from 100 to 805 agent runs per second (4096: 46 to 772; local SQLite at 2048: 88 to 459), with CPU per run flat at about 1.6 ms instead of growing linearly. - `subscribeCommits()` listeners are read when a commit's dispatch runs: a listener added between publication and the dispatch gets the commit only when one was registered at adoption; one added after the dispatch ran misses it. - The abort signal of a newly marked running task fires after `subscribeCommits()` listeners instead of before them. - A throwing Durable commit observer fails the Session after `Storage.commit()` returned, after the other observers and the host listeners got the commit; the commit still stands. When Storage also rejects that commit after adopting it, the storage error fails the Session and the observer's error is reported as `listener_error`. diff --git a/packages/durable/src/harness/harness.ts b/packages/durable/src/harness/harness.ts index 5ef69ef23..8fb91bcc0 100644 --- a/packages/durable/src/harness/harness.ts +++ b/packages/durable/src/harness/harness.ts @@ -106,6 +106,9 @@ type ConversationHost = { /** Key of the Harness's internal index-size getter, for tests. */ export const SCHEDULER_INDEX_SIZES = Symbol.for("pi-durable.schedulerIndexSizes"); +/** Key of the Harness's internal idle-state getter (kept contexts against a full recompute), for tests. */ +export const SCHEDULER_IDLE_STATE = Symbol.for("pi-durable.schedulerIdleState"); + class ConversationImpl implements Conversation { readonly id: ConversationId; readonly #host: ConversationHost; @@ -318,6 +321,11 @@ class HarnessImpl extends SessionImpl implements return this.#tasks.indexSizes; } + /** Internal, for tests: the kept contexts' idle state and its disagreements with a full recompute. */ + get [SCHEDULER_IDLE_STATE](): TaskScheduler["idleState"] { + return this.#tasks.idleState; + } + inspect(context: Context): Promise { return this.readOnLine(async (storage) => { const { scheduling, tasks } = await this.#tasks.inspect(this.#options.registry.snapshot()); diff --git a/packages/durable/src/harness/scheduler.ts b/packages/durable/src/harness/scheduler.ts index c0bd82ef4..63cbe89ac 100644 --- a/packages/durable/src/harness/scheduler.ts +++ b/packages/durable/src/harness/scheduler.ts @@ -314,6 +314,20 @@ export class TaskScheduler { ConversationId, { range: ContextRange; idleSince: number | undefined; expiresAt?: number } >(); + /** + * The idle kept contexts by `idleSince`, oldest first: the expiry order, so neither an expiry nor finding the next one + * scans `#contexts`. Kept sorted also when the Harness clock goes back (see `#markIdle`). + */ + readonly #idleOrder = new Map(); + /** At least the newest `idleSince` in `#idleOrder` (drops do not lower it): an older one added means the clock went back. */ + #idleNewest = Number.NEGATIVE_INFINITY; + /** + * A chain load resolved what `#idle()` treats as unknown for a live task whose chain is not loaded, which can make a + * scope idle without a task change below it; the next idle check then covers every scope. + */ + #unknownsChanged = false; + /** Idle checks that covered every scope (`idleState`, a test diagnostic). */ + #fullIdleChecks = 0; /** 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(); @@ -503,8 +517,22 @@ export class TaskScheduler { const now = this.#now(); // A reconcile pass with work to do is work too: it finalizes held outcomes and marks cascades, at open as // recovery. A process that dies before it lands must be woken again. A pass with nothing to do is not. - let active = this.#invocations.size > 0 || this.#reconcilingWork > 0 || this.#reconcileWork(); + const active = + this.#invocations.size > 0 || this.#reconcilingWork > 0 || this.#reconcileWork() || this.#runsNow(now); + // The earliest sleep not yet due: only `#sleeping` holds sleeping tasks, so no pass visits every live task. let at: number | undefined; + for (const id of this.#sleeping) { + const record = this.#live.get(id); + const state = record?.state; + if (state?.status === "sleeping" && !record!.abortRequested && state.until > now) { + at = at === undefined ? state.until : Math.min(at, state.until); + } + } + return at === undefined ? { active } : { active, at }; + } + + /** Whether a live task can run now: the scan `wake()` needs only while no invocation runs. */ + #runsNow(now: number): boolean { let snapshot: RegistrySnapshot | undefined; let snapshotFailed = false; for (const record of this.#live.values()) { @@ -512,11 +540,7 @@ export class TaskScheduler { 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; + if (state.status === "sleeping" && !record.abortRequested && state.until > now) 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. @@ -531,9 +555,9 @@ export class TaskScheduler { } if (snapshot !== undefined && "reason" in this.#fit(record, snapshot.task(record.kind))) continue; } - active = true; + return true; } - return at === undefined ? { active } : { active, at }; + return false; } /** @@ -744,6 +768,8 @@ export class TaskScheduler { const updated: AnyTaskRecord[] = []; const failed: TaskId[] = []; let changed = false; + // The conversations whose idleness the commit's task changes can change; undefined when a chain is not known. + let dirty: Set | undefined = new Set(); // Edges first, so the tasks a commit creates in a new conversation hang from it. for (const change of publication.changes) { if (change.type === "conversation" && !this.#edges.has(change.value.id)) { @@ -757,6 +783,10 @@ export class TaskScheduler { const previous = this.#live.get(record.id); if (failedOutcome(record) && (previous === undefined || !failedOutcome(previous))) failed.push(record.id); this.#track(previous, record); + // Idleness counts live tasks, so only a task that became live or ended can change it. Collected before `#sweep()`, + // which may forget the chain of a task that ended. + const liveness = previous === undefined || record.state.status === "terminal"; + if (liveness && dirty !== undefined && !this.#addScopesAbove(record, dirty)) dirty = undefined; if (record.state.status === "terminal") { this.#failedMigrations.delete(record.id); this.#failFastChecks.delete(record.id); @@ -817,10 +847,24 @@ export class TaskScheduler { if (this.#cascadePending) this.#scheduleReconcile(); this.#sweep(); if (!changed) return; - this.#settleIdle(); + this.#settleIdle(dirty); this.#kick(); } + /** + * Add the conversations above `record` to `scopes`: those whose idleness a change of it can change. False when a node + * on the way up is not known. A conversation already in `scopes` has everything above it there too. + */ + #addScopesAbove(record: AnyTaskRecord, scopes: Set): boolean { + for (const step of this.#above(parentOf(record))) { + if ("unknown" in step) return false; + if (!("conversation" in step)) continue; + if (scopes.has(step.conversation)) return true; + scopes.add(step.conversation); + } + return true; + } + /** * Extend each kept context range with the entries a stored commit appended to its conversation, so a later read * needs no Storage scan. Entries are published in ID order. @@ -841,11 +885,11 @@ export class TaskScheduler { // An idle range past its retention is dropped here too: entry-only commits do not run `#settleIdle()`, and a // host may decline the expiry timer. `expiresAt` was fixed when it went idle, so no settings are read here. if (kept.idleSince !== undefined && kept.expiresAt !== undefined && kept.expiresAt <= this.#now()) { - this.#contexts.delete(conversationId); + this.#dropContext(conversationId); continue; } const range = advanceRange(kept.range, entries); - if (range === undefined) this.#contexts.delete(conversationId); + if (range === undefined) this.#dropContext(conversationId); else kept.range = range; } } @@ -863,50 +907,98 @@ export class TaskScheduler { } } - /** Resolve idle waiters, and drop each kept context whose conversation has been idle for the retention period. */ - #settleIdle(): void { - for (const conversationId of this.#idleWaiters.keys()) { - if (this.#idle(conversationId)) this.#idleWaiters.resolve(conversationId); - } + /** + * After task changes below the conversations in `scopes` (every conversation when undefined): resolve their idle + * waiters and the Harness-wide ones, drop the kept contexts whose idle period passed the retention, then start or end + * the idle periods of the kept contexts in `scopes`. A task change can change the idleness only of the conversations + * above it, so no check visits every conversation. + */ + #settleIdle(scopes: ReadonlySet | undefined): void { + const all = scopes === undefined || this.#unknownsChanged; + this.#unknownsChanged = false; + if (all) this.#fullIdleChecks++; + this.#resolveIdleWaiters(all ? undefined : scopes); const now = this.#now(); const retention = this.#contextRetentionMs(); - for (const [conversationId, kept] of this.#contexts) { - if (kept.idleSince !== undefined && now - kept.idleSince >= retention) { - this.#contexts.delete(conversationId); - } else if (!this.#idle(conversationId)) { + // Expired first: a context past its retention is dropped even when the same commit makes it busy again. + this.#expireContexts(now, retention); + for (const conversationId of all ? [...this.#contexts.keys()] : scopes) { + const kept = this.#contexts.get(conversationId); + if (kept === undefined) continue; + if (!this.#idle(conversationId)) { + if (kept.idleSince === undefined) continue; kept.idleSince = undefined; - } else if (retention > 0) { - if (kept.idleSince === undefined) { - kept.idleSince = now; - kept.expiresAt = now + retention; - } - } else { - this.#contexts.delete(conversationId); + this.#idleOrder.delete(conversationId); + } else if (retention <= 0) { + this.#dropContext(conversationId); + } else if (kept.idleSince === undefined) { + kept.idleSince = now; + kept.expiresAt = now + retention; + this.#markIdle(conversationId, now); } } this.#scheduleExpiry(); } + /** Resolve the idle waiters of the conversations in `scopes` (all when undefined) and the Harness-wide ones. */ + #resolveIdleWaiters(scopes: ReadonlySet | undefined): void { + for (const conversationId of this.#idleWaiters.keys()) { + if (conversationId !== undefined && scopes !== undefined && !scopes.has(conversationId)) continue; + if (this.#idle(conversationId)) this.#idleWaiters.resolve(conversationId); + } + } + /** - * 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. + * Drop the kept contexts idle for at least `retention`, oldest first, with the retention setting of now; with no + * retention, every idle one, also those whose idle period a clock that went back makes start after now. + */ + #expireContexts(now: number, retention: number): void { + for (const [conversationId, idleSince] of this.#idleOrder) { + if (retention > 0 && now - idleSince < retention) return; + this.#dropContext(conversationId); + } + } + + /** + * Start the idle period of a kept context in the expiry order. An idle period older than the newest one kept, which + * a clock that went back starts, sorts the order again, and the newest is taken from what is kept then. + */ + #markIdle(conversationId: ConversationId, idleSince: number): void { + this.#idleOrder.delete(conversationId); + this.#idleOrder.set(conversationId, idleSince); + if (this.#idleOrder.size === 1 || idleSince >= this.#idleNewest) { + this.#idleNewest = idleSince; + return; + } + const sorted = [...this.#idleOrder].sort((a, b) => a[1] - b[1]); + this.#idleOrder.clear(); + for (const [id, since] of sorted) this.#idleOrder.set(id, since); + this.#idleNewest = sorted.at(-1)![1]; + } + + /** Forget a kept context, idle or not, and its place in the expiry order. */ + #dropContext(conversationId: ConversationId): void { + this.#contexts.delete(conversationId); + this.#idleOrder.delete(conversationId); + } + + /** + * Drop the expired contexts 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(); - let at: number | undefined; - for (const kept of this.#contexts.values()) { - if (kept.idleSince !== undefined && (at === undefined || kept.idleSince + retention < at)) { - at = kept.idleSince + retention; - } - } + const oldest = this.#idleOrder.values().next(); + const at = oldest.done === true ? undefined : oldest.value + this.#contextRetentionMs(); if (this.#expiry !== undefined && this.#expiry.at === at) return; this.#expiry?.disarm(); this.#expiry = undefined; if (at === undefined || this.#closing) return; const disarm = this.#timers.background(() => { this.#expiry = undefined; - this.#settleIdle(); + this.#expireContexts(this.#now(), this.#contextRetentionMs()); + this.#scheduleExpiry(); }, at - this.#now()); if (disarm !== undefined) this.#expiry = { at, disarm }; } @@ -954,7 +1046,16 @@ export class TaskScheduler { } if (work) this.#reconcilingWork--; this.#scheduleWake(); - this.#settleIdle(); + // Its commit's task changes were settled at their publication; a waiter that asked for this pass checks again, and + // what its chain loads resolved can make any scope idle. + if (this.#unknownsChanged) { + this.#settleIdle(undefined); + return; + } + this.#resolveIdleWaiters(undefined); + // A pass checks expiry too, for hosts that decline the expiry timer. + this.#expireContexts(this.#now(), this.#contextRetentionMs()); + this.#scheduleExpiry(); } /** @@ -1138,46 +1239,65 @@ export class TaskScheduler { */ async #loadChain(start: Up, overlay?: Overlay): Promise { const passed: TaskId[] = []; + let installed = false; let at: Up | undefined = start; - while (at !== undefined) { - if ("task" in at) { - const id: TaskId = at.task; - let node: TaskNode | undefined = this.#live.get(id) ?? this.#settled.get(id); - // Created in this commit: the load from its own record covers what is above it. - if (node === undefined && overlay?.tasks.has(id) === true) break; - if (node !== undefined && !this.#unloaded.has(id)) break; - if (node === undefined) { - // Through the line's Storage: a task whose creation or end is not published yet waits for it. - const record = await this.#session.lineStorage.task(id, this.#context); - if (record === undefined) return; - node = nodeOf(record); - if (record.state.status === "terminal") { - this.#settle(id, node); - this.#pin(id); + try { + while (at !== undefined) { + if ("task" in at) { + const id: TaskId = at.task; + let node: TaskNode | undefined = this.#live.get(id) ?? this.#settled.get(id); + // Created in this commit: the load from its own record covers what is above it. + if (node === undefined && overlay?.tasks.has(id) === true) break; + if (node !== undefined && !this.#unloaded.has(id)) break; + if (node === undefined) { + // Through the line's Storage: a task whose creation or end is not published yet waits for it. + const record = await this.#session.lineStorage.task(id, this.#context); + if (record === undefined) return; + node = nodeOf(record); + if (record.state.status === "terminal") { + this.#settle(id, node); + this.#pin(id); + installed = true; + } + // A loaded node can reveal a cancelling owner: the next reconcile derives marks again. + this.#scheduleReconcile(); } - // A loaded node can reveal a cancelling owner: the next reconcile derives marks again. - this.#scheduleReconcile(); + passed.push(id); + at = parentOf(node); + } else { + let edge = this.#edge(at.conversation, overlay); + if (edge === undefined) { + // Through the line's Storage: a conversation whose creation is not published yet waits for it, so a missing + // record is never mistaken for an ownerless one. + const record = await this.#session.lineStorage.conversation(at.conversation, this.#context); + edge = record?.owner?.taskId ?? null; + this.#setEdge(at.conversation, edge); + this.#pin(conversationNode(at.conversation)); + installed = true; + // A loaded node can reveal a cancelling owner: the next reconcile derives marks again. + this.#scheduleReconcile(); + } + at = edge === null ? undefined : { task: edge }; } - passed.push(id); - at = parentOf(node); - } else { - let edge = this.#edge(at.conversation, overlay); - if (edge === undefined) { - // Through the line's Storage: a conversation whose creation is not published yet waits for it, so a missing - // record is never mistaken for an ownerless one. - const record = await this.#session.lineStorage.conversation(at.conversation, this.#context); - edge = record?.owner?.taskId ?? null; - this.#setEdge(at.conversation, edge); - this.#pin(conversationNode(at.conversation)); - // A loaded node can reveal a cancelling owner: the next reconcile derives marks again. - this.#scheduleReconcile(); - } - at = edge === null ? undefined : { task: edge }; + } + } finally { + // `#idle()` counts a live task whose chain is not loaded as inside every scope it may be in: what this load + // resolved, also when it stopped early, can make such a scope idle without a task change below it. + if (passed.some((id) => this.#live.has(id)) || (installed && this.#liveUnloaded())) { + this.#unknownsChanged = true; + // A waiter such a task held is checked again at the pass's end, also when this load was not a reconcile's. + this.#scheduleReconcile(); } } for (const id of passed) this.#unloaded.delete(id); } + /** Whether a live task's chain is not loaded yet: one `#idle()` counts as inside every scope it may be in. */ + #liveUnloaded(): boolean { + for (const id of this.#unloaded) if (this.#live.has(id)) return true; + return false; + } + #setEdge(conversationId: ConversationId, owner: TaskId | null): void { this.#edges.set(conversationId, owner); // The conversation now hangs from its owner, or is a root; with nothing below it, it waits for the sweep. @@ -1629,6 +1749,7 @@ export class TaskScheduler { this.#taskWaiters.rejectAll(error); this.#idleWaiters.rejectAll(error); this.#contexts.clear(); + this.#idleOrder.clear(); this.#expiry?.disarm(); this.#expiry = undefined; this.#timer?.disarm(); @@ -1969,6 +2090,32 @@ export class TaskScheduler { }; } + /** + * The kept contexts' idle state, for tests (a diagnostic, computed on demand): how many are kept and how many are + * idle, how many disagree with a full recompute (an idle period without `#idle()`, or `#idle()` without one; an + * expiry order that differs from the idle periods, or is not sorted), and how many idle checks covered every scope. + */ + get idleState(): Readonly> { + let mismatches = 0; + let idle = 0; + for (const [conversationId, kept] of this.#contexts) { + if (kept.idleSince !== undefined) idle++; + if ((kept.idleSince !== undefined) !== this.#idle(conversationId)) mismatches++; + if (this.#idleOrder.get(conversationId) !== kept.idleSince) mismatches++; + } + let previous = Number.NEGATIVE_INFINITY; + for (const [conversationId, idleSince] of this.#idleOrder) { + if (idleSince < previous || this.#contexts.get(conversationId)?.idleSince !== idleSince) mismatches++; + previous = idleSince; + } + return { + contexts: this.#contexts.size, + idleContexts: idle, + idleMismatches: mismatches, + fullIdleChecks: this.#fullIdleChecks, + }; + } + /** * Scheduling state and every live task with its derived state, read on the Session line. Runs no task code: a * pending migration shows as `ready` with `migrates`, and only a migration the scheduler already tried, or one that @@ -2713,14 +2860,16 @@ export class TaskScheduler { // A read of another, idle conversation starts or continues its retention period. if (!this.#idle(conversationId)) { this.#contexts.set(conversationId, { range, idleSince: undefined }); + this.#idleOrder.delete(conversationId); } else if (this.#contextRetentionMs() > 0) { - // An idle period that already started keeps its expiry; a new one starts now. + // An idle period that already started keeps its expiry and its place; a new one starts now. const idleSince = kept?.idleSince ?? this.#now(); const expiresAt = kept?.idleSince !== undefined && kept.expiresAt !== undefined ? kept.expiresAt : idleSince + this.#contextRetentionMs(); this.#contexts.set(conversationId, { range, idleSince, expiresAt }); + if (kept?.idleSince === undefined) this.#markIdle(conversationId, idleSince); this.#scheduleExpiry(); } } diff --git a/packages/durable/test/harness-idle-scopes.test.ts b/packages/durable/test/harness-idle-scopes.test.ts new file mode 100644 index 000000000..b034150b0 --- /dev/null +++ b/packages/durable/test/harness-idle-scopes.test.ts @@ -0,0 +1,591 @@ +// Idle checks without per-commit scans: a task change settles only the conversations above it (idle waiters and kept +// context ranges), expiry runs oldest first from an ordered index, and `wake()` reads only sleeping tasks while work +// is active. Every case checks the incremental idle state against a full recompute (`SCHEDULER_IDLE_STATE`). +import { createModels } from "@earendil-works/pi-ai"; +import { + type Conversation, + type ConversationId, + createRegistry, + defineExtension, + defineTask, + Harness, + MemoryStorage, + type SetTimer, + type Storage, + type TaskId, +} from "@earendil-works/pi-durable"; +import { afterEach, describe, expect, it } from "vitest"; +import { SCHEDULER_IDLE_STATE } from "../src/harness/harness.ts"; +import { commitWrites } from "../src/testing/commit-writes.ts"; +import { ROOT_CONVERSATION_ID, type StorageWrite } from "../src/types.ts"; +import { ControlledStorage, context, flush } from "./session-support.ts"; +import { aborted, type Deferred, deferred, eventually, openTasks, settled } from "./task-support.ts"; + +const gates = new Map>(); +/** Resolves once the named task's context read returned. */ +const reads = new Map>(); + +function named(map: Map>, name: string): Deferred { + let found = map.get(name); + if (found === undefined) { + found = deferred(); + map.set(name, found); + } + return found; +} + +afterEach(() => { + gates.clear(); + reads.clear(); +}); + +/** + * Reads the context of conversation `read` (when given), then holds until its gate opens and completes; its abort + * handler ends it `aborted`. + */ +const Work = defineTask<{ name: string; read?: number }, { phase: "run" }, null>({ + name: "test.idle.work", + version: 1, + initial: () => ({ phase: "run" }), + phases: { + run: async (task, runtime, ctx) => { + if (task.input.read !== undefined) { + await runtime.context(task.input.read as ConversationId, ctx); + named(reads, task.input.name).resolve(); + } + const opened = await Promise.race([ + named(gates, task.input.name).promise.then(() => true), + aborted(runtime.signal), + ]); + if (opened !== true) return; + await runtime.commit(() => ({ status: "terminal", outcome: { status: "completed", result: null } }), ctx); + }, + }, + abort: async (_task, runtime, ctx) => + runtime.commit(() => ({ status: "terminal", outcome: { status: "aborted" } }), ctx), +}); + +type IdleState = Record<"contexts" | "idleContexts" | "idleMismatches" | "fullIdleChecks", number>; + +function idleState(harness: Harness): IdleState { + return (harness as unknown as Record)[SCHEDULER_IDLE_STATE]!; +} + +/** Background timers declined, as Cloudflare Workers do; foreground ones on `setTimeout`. */ +const declineBackground: SetTimer = (callback, delayMs, { background }) => { + if (background) return undefined; + const timer = setTimeout(callback, delayMs); + return () => clearTimeout(timer); +}; + +/** + * A resumed Harness running `Work`, over MemoryStorage counting the entry rows each scan returns, with `clock` as its + * time, the retention `retention` returns, and background timers declined unless `timers`. + */ +async function open(options: { clock?: () => number; retention?: () => number; timers?: boolean } = {}) { + const scanned = { rows: 0 }; + const memory = new MemoryStorage(); + const storage = new Proxy(memory, { + get(target, key) { + const value: unknown = Reflect.get(target, key, target); + if (key === "scanEntries") { + return async (...args: Parameters) => { + const page = await target.scanEntries(...args); + scanned.rows += page.items.length; + return page; + }; + } + return typeof value === "function" ? value.bind(target) : value; + }, + }); + const registry = createRegistry(); + registry.install(defineExtension({ name: "tasks", tasks: [Work] })); + const retention = options.retention ?? (() => 600_000); + const harness = await Harness.open( + storage, + { + models: createModels(), + registry, + settings: () => ({ contextRetentionMs: retention() }), + ...(options.clock === undefined ? {} : { now: options.clock }), + ...(options.timers === true ? {} : { setTimer: declineBackground }), + }, + context, + ); + harness.resume(); + const root = await harness.root(context); + /** A new ownerless conversation with `entries` entries. */ + const conversation = async (entries = 2): Promise => { + const created = await harness.createConversation({ ownership: { kind: "ownerless" } }, context); + for (let index = 0; index < entries; index++) { + await created.commit((tx) => tx.appendEntry(created.id, { kind: "note", data: { index } }), context); + } + return created; + }; + /** Start `Work` named `name` in `into`, reading conversation `read`, and wait for its read. */ + const work = async (name: string, into: Conversation, read?: ConversationId): Promise => { + const input = read === undefined ? { name } : { name, read: read as number }; + const id = await into.commit( + (tx) => tx.createTask(Work, input, { ownership: { kind: "conversation" }, conversationId: into.id }), + context, + ); + if (read !== undefined) await named(reads, name).promise; + return id; + }; + /** Open the gate of `name` and wait for its task, `id`, to end. */ + const end = async (name: string, id: TaskId): Promise => { + named(gates, name).resolve(); + await harness.waitForTask(id, context); + }; + /** The entry rows a read of `read` from a new task in `into` scans. */ + const rowsOf = async (name: string, into: Conversation, read: ConversationId): Promise => { + const before = scanned.rows; + const id = await work(name, into, read); + const rows = scanned.rows - before; + await end(name, id); + return rows; + }; + return { harness, root, conversation, work, end, rowsOf }; +} + +/** Flush until scheduler passes started by the last changes settled. */ +async function quiet(): Promise { + for (let round = 0; round < 4; round++) await flush(); +} + +describe("idle waiters by scope", () => { + it("resolve through owner chains, owned conversations and ended tasks between, ignoring background work", async () => { + const { harness, root } = await open(); + const tree = await root.commit(async (tx) => { + const p = await tx.createTask(Work, { name: "p" }, { ownership: { kind: "conversation" } }); + const b = await tx.createTask(Work, { name: "b" }, { ownership: { kind: "task", taskId: p } }); + const a = await tx.createTask(Work, { name: "a" }, { ownership: { kind: "task", taskId: b } }); + const c = await tx.createConversation({ ownership: { kind: "task", taskId: a } }); + const d = await tx.createTask( + Work, + { name: "d" }, + { ownership: { kind: "conversation" }, conversationId: c.id }, + ); + const g = await tx.createTask( + Work, + { name: "g" }, + { ownership: { kind: "conversation" }, conversationId: c.id, background: true }, + ); + return { p, b, a, c: c.id, d, g }; + }, context); + const c = (await harness.conversation(tree.c, context))!; + const idleC = c.waitForIdle(context); + const idleRoot = root.waitForIdle(context); + const idleAll = harness.waitForIdle(context); + // The owners A and B hold for D: live, though their handlers returned. + named(gates, "a").resolve(); + named(gates, "b").resolve(); + await eventually(async () => (await harness.getTask(tree.b, context))!.state.status === "completing"); + expect([await settled(idleC), await settled(idleRoot), await settled(idleAll)]).toEqual([false, false, false]); + // D ends: C is idle (G is background); A and B finalize; P still holds the root. + named(gates, "d").resolve(); + await idleC; + await harness.waitForTask(tree.b, context); + expect([await settled(idleRoot), await settled(idleAll)]).toEqual([false, false]); + named(gates, "p").resolve(); + await idleRoot; + await idleAll; + expect((await harness.getTask(tree.g, context))!.state.status).not.toBe("terminal"); + await harness.close(context); + }); + + it("count work whose chain is not loaded yet as inside, until the open reconcile loads it", async () => { + const storage = new ControlledStorage(); + const first = await openTasks(storage, [Work]); + const firstRoot = await first.harness.root(context); + const tree = await firstRoot.commit(async (tx) => { + const p = await tx.createTask(Work, { name: "p" }, { ownership: { kind: "conversation" } }); + const c = await tx.createConversation({ ownership: { kind: "task", taskId: p } }); + const d = await tx.createTask( + Work, + { name: "d" }, + { ownership: { kind: "conversation" }, conversationId: c.id }, + ); + const other = await tx.createConversation({ ownership: { kind: "ownerless" } }); + return { p, c: c.id, d, other: other.id }; + }, context); + await first.harness.close(context); + // Reopened, no chain is loaded: P and D count as inside every scope until the reconcile pass loads theirs. + const { harness } = await openTasks(storage.reopen(), [Work]); + const idleOther = (await harness.conversation(tree.other, context))!.waitForIdle(context); + const idleRoot = (await harness.root(context)).waitForIdle(context); + await idleOther; + expect(await settled(idleRoot)).toBe(false); + harness.resume(); + named(gates, "d").resolve(); + await harness.waitForTask(tree.d, context); + expect(await settled(idleRoot)).toBe(false); + named(gates, "p").resolve(); + await idleRoot; + await harness.close(context); + }); + + it("resolve a waiter on an unrelated conversation without settling the busy one", async () => { + const { harness, conversation, work, end } = await open(); + const [busy, other] = [await conversation(0), await conversation(0)]; + const held = await work("held", busy); + const quick = await work("quick", other); + const idleBusy = busy.waitForIdle(context); + const idleOther = other.waitForIdle(context); + await end("quick", quick); + await idleOther; + expect(await settled(idleBusy)).toBe(false); + await end("held", held); + await idleBusy; + await harness.close(context); + }); +}); + +describe("kept context ranges", () => { + it("settles a run end whose sweep drops the chain without a check of every scope", async () => { + const { harness, root } = await open(); + const tree = await root.commit(async (tx) => { + const p = await tx.createTask(Work, { name: "p" }, { ownership: { kind: "conversation" } }); + const t = await tx.createTask(Work, { name: "t" }, { ownership: { kind: "task", taskId: p } }); + const c = await tx.createConversation({ ownership: { kind: "task", taskId: t } }); + // A context range keeps entries: an empty conversation has none to keep. + await tx.appendEntry(c.id, { kind: "note" }); + return { p, t, c: c.id }; + }, context); + const c = (await harness.conversation(tree.c, context))!; + const d = await c.commit( + (tx) => + tx.createTask( + Work, + { name: "d", read: tree.c as number }, + { ownership: { kind: "conversation" }, conversationId: tree.c }, + ), + context, + ); + await named(reads, "d").promise; + named(gates, "t").resolve(); + await eventually(async () => (await harness.getTask(tree.t, context))!.state.status === "completing"); + await quiet(); + const before = idleState(harness); + expect(before).toMatchObject({ contexts: 1, idleContexts: 0, idleMismatches: 0 }); + // D's end makes C idle, T finalizes, and the sweep forgets C's chain: none of it checks every scope. + named(gates, "d").resolve(); + await harness.waitForTask(d, context); + await harness.waitForTask(tree.t, context); + await quiet(); + expect(idleState(harness)).toEqual({ ...before, idleContexts: 1 }); + named(gates, "p").resolve(); + await harness.waitForTask(tree.p, context); + await harness.close(context); + }); + + it("drops a range past its retention when a task change makes it busy again, with background timers declined", async () => { + let clock = 1_000; + const { harness, conversation, work, end } = await open({ clock: () => clock, retention: () => 1_000 }); + const [x, z] = [await conversation(), await conversation(0)]; + await end("reader", await work("reader", z, x.id)); + expect(idleState(harness)).toMatchObject({ contexts: 1, idleContexts: 1, idleMismatches: 0 }); + clock = 2_500; + // No timer expired it; the task change that makes X busy checks expiry first. + const busy = await work("busy", x); + await quiet(); + expect(idleState(harness)).toMatchObject({ contexts: 0, idleContexts: 0, idleMismatches: 0 }); + await end("busy", busy); + await harness.close(context); + }); + + it("drops a range past its retention at a reconcile pass with nothing to do, with background timers declined (review)", async () => { + let clock = 1_000; + const { harness, conversation, work, end } = await open({ clock: () => clock, retention: () => 1_000 }); + const [x, z] = [await conversation(), await conversation(0)]; + await end("reader", await work("reader", z, x.id)); + const held = await work("held", z); + // Held runs: no task changes from here. + await quiet(); + expect(idleState(harness)).toMatchObject({ contexts: 1, idleContexts: 1 }); + clock = 2_500; + // A wait on busy Z asks for a reconcile pass, which has no work and changes no task: it checks expiry. + const idle = z.waitForIdle(context); + await quiet(); + expect(idleState(harness)).toMatchObject({ contexts: 0, idleMismatches: 0 }); + await end("held", held); + await idle; + await harness.close(context); + }); + + it("expires in idle order when the clock goes back", async () => { + let clock = 1_000; + const { harness, conversation, work, end, rowsOf } = await open({ + clock: () => clock, + retention: () => 1_000, + }); + const [a, b, z] = [await conversation(3), await conversation(5), await conversation(0)]; + await end("read a", await work("read a", z, a.id)); + clock = 500; + await end("read b", await work("read b", z, b.id)); + expect(idleState(harness)).toMatchObject({ contexts: 2, idleContexts: 2, idleMismatches: 0 }); + // B went idle at 500 and expires first, though it went idle last. + clock = 1_600; + await end("unrelated", await work("unrelated", z)); + expect(idleState(harness)).toMatchObject({ contexts: 1, idleMismatches: 0 }); + expect(await rowsOf("again b", z, b.id)).toBe(5); + expect(await rowsOf("again a", z, a.id)).toBe(0); + await harness.close(context); + }); + + it("drops every idle range when retention becomes zero, also one that went idle after the clock's now (review)", async () => { + let clock = 2_000; + let retention = 1_000; + const { harness, conversation, work, end } = await open({ clock: () => clock, retention: () => retention }); + const [a, z] = [await conversation(), await conversation(0)]; + await end("read", await work("read", z, a.id)); + expect(idleState(harness)).toMatchObject({ contexts: 1, idleContexts: 1 }); + // The clock goes back below A's idle start; then retention 0 and an unrelated task change. + clock = 1_000; + retention = 0; + await end("unrelated", await work("unrelated", z)); + expect(idleState(harness)).toMatchObject({ contexts: 0, idleMismatches: 0 }); + await harness.close(context); + }); + + it("keeps an idle range's expiry when another task reads it again", async () => { + let clock = 1_000; + const { harness, conversation, work, end, rowsOf } = await open({ + clock: () => clock, + retention: () => 1_000, + }); + const [a, b, z] = [await conversation(3), await conversation(5), await conversation(0)]; + await end("read a", await work("read a", z, a.id)); + clock = 1_500; + await end("read b", await work("read b", z, b.id)); + clock = 1_800; + expect(await rowsOf("again a", z, a.id)).toBe(0); + clock = 2_100; + await end("unrelated", await work("unrelated", z)); + expect(idleState(harness)).toMatchObject({ contexts: 1, idleMismatches: 0 }); + expect(await rowsOf("late b", z, b.id)).toBe(0); + await harness.close(context); + }); + + it("drops a range from the expiry order when an entry commit finds it expired", async () => { + let clock = 1_000; + const { harness, conversation, work, end } = await open({ clock: () => clock, retention: () => 1_000 }); + const [a, z] = [await conversation(), await conversation(0)]; + await end("read", await work("read", z, a.id)); + clock = 2_500; + // An entry-only commit settles no idleness: the range is dropped where it would be extended. + await a.commit((tx) => tx.appendEntry(a.id, { kind: "note", data: { late: true } }), context); + expect(idleState(harness)).toMatchObject({ contexts: 0, idleContexts: 0, idleMismatches: 0 }); + await harness.close(context); + }); + + it("drops a range from the expiry order when an entry commit cannot extend it", async () => { + const { harness, conversation, work, end } = await open(); + const [a, z] = [await conversation(0), await conversation(0)]; + const first = await a.commit((tx) => tx.appendEntry(a.id, { kind: "note" }), context); + const second = await a.commit((tx) => tx.appendEntry(a.id, { kind: "note" }), context); + await a.commit((tx) => tx.appendEntry(a.id, { kind: "summary", head: second.id }), context); + await end("read", await work("read", z, a.id)); + expect(idleState(harness)).toMatchObject({ contexts: 1, idleContexts: 1, idleMismatches: 0 }); + // A marker whose head is older than the range's: the range cannot be extended and is dropped. + await a.commit((tx) => tx.appendEntry(a.id, { kind: "summary", head: first.id }), context); + expect(idleState(harness)).toMatchObject({ contexts: 0, idleContexts: 0, idleMismatches: 0 }); + await harness.close(context); + }); + + it.each([ + ["shorter", 400, 0], + ["longer", 10_000, 1], + ["zero", 0, 0], + ] as const)("applies a %s retention to idle ranges at an unrelated commit", async (_name, changed, kept) => { + let clock = 1_000; + let retention = 1_000; + const { harness, conversation, work, end } = await open({ clock: () => clock, retention: () => retention }); + const [a, z] = [await conversation(), await conversation(0)]; + await end("read", await work("read", z, a.id)); + clock = 1_500; + retention = changed; + await end("unrelated", await work("unrelated", z)); + expect(idleState(harness)).toMatchObject({ contexts: kept, idleMismatches: 0 }); + await harness.close(context); + }); + + it("drops a range that goes idle with zero retention", async () => { + const { harness, conversation, work, end } = await open({ retention: () => 0 }); + const a = await conversation(); + const own = await work("own", a, a.id); + await quiet(); + expect(idleState(harness)).toMatchObject({ contexts: 1, idleContexts: 0, idleMismatches: 0 }); + await end("own", own); + await quiet(); + expect(idleState(harness)).toMatchObject({ contexts: 0, idleMismatches: 0 }); + await harness.close(context); + }); +}); + +/** A random source with a fixed seed. */ +function seeded(seed: number): () => number { + let state = seed; + return () => { + state = (state * 1103515245 + 12345) % 2147483648; + return state / 2147483648; + }; +} + +describe("incremental idle state", () => { + it("equals a full recompute after every step of random nested work", async () => { + const random = seeded(42); + const pick = (items: readonly T[]): T => items[Math.floor(random() * items.length)]!; + const { harness, root } = await open(); + // A context range keeps entries: every conversation gets one. + await root.commit((tx) => tx.appendEntry(root.id, { kind: "note" }), context); + const conversations: ConversationId[] = [root.id]; + const live = new Map(); + let count = 0; + let fullChecks = idleState(harness).fullIdleChecks; + const handle = async (id: ConversationId) => (await harness.conversation(id, context))!; + for (let step = 0; step < 200; step++) { + const action = random(); + const name = `t${count++}`; + if (action < 0.1) { + const created = await harness.createConversation({ ownership: { kind: "ownerless" } }, context); + await created.commit((tx) => tx.appendEntry(created.id, { kind: "note" }), context); + conversations.push(created.id); + } else if (action < 0.45 || live.size === 0) { + // A task in any conversation, also one whose owner chain ended and was swept; it may read a context. + const into = pick(conversations); + const read = random() < 0.6 ? pick(conversations) : undefined; + const background = random() < 0.15; + const input = read === undefined ? { name } : { name, read: read as number }; + const target = await handle(into); + const id = await target.commit( + (tx) => + tx.createTask(Work, input, { ownership: { kind: "conversation" }, conversationId: into, background }), + context, + ); + live.set(name, { id, conversation: into }); + if (read !== undefined) await named(reads, name).promise; + } else if (action < 0.6) { + // Work a live task owns: a child task, or an owned conversation with a task that reads it. + const [ownerName, owner] = pick([...live]); + const state = (await harness.getTask(owner.id, context))?.state.status; + if (state === undefined || state === "terminal" || state === "completing") { + live.delete(ownerName); + continue; + } + const conversation = random() < 0.5; + const created = await (await handle(owner.conversation)).commit(async (tx) => { + if (!conversation) { + const id = await tx.createTask(Work, { name }, { ownership: { kind: "task", taskId: owner.id } }); + return { id, into: owner.conversation, read: false }; + } + const child = await tx.createConversation({ ownership: { kind: "task", taskId: owner.id } }); + await tx.appendEntry(child.id, { kind: "note" }); + const id = await tx.createTask( + Work, + { name, read: child.id as number }, + { ownership: { kind: "conversation" }, conversationId: child.id }, + ); + return { id, into: child.id, read: true }; + }, context); + if (created.read) conversations.push(created.into); + live.set(name, { id: created.id, conversation: created.into }); + if (created.read) await named(reads, name).promise; + } else { + const [ended] = pick([...live]); + live.delete(ended); + named(gates, ended).resolve(); + } + await quiet(); + const state = idleState(harness); + expect(state.idleMismatches, `step ${step}`).toBe(0); + fullChecks = state.fullIdleChecks; + } + const state = idleState(harness); + expect(state.contexts).toBeGreaterThan(5); + expect(state.idleContexts).toBeGreaterThan(0); + expect(state.idleContexts).toBeLessThan(state.contexts); + expect(fullChecks).toBe(0); + for (const gate of gates.values()) gate.resolve(); + await harness.close(context); + }); +}); + +describe("wake()", () => { + const kind = Work.definition.name; + const record = (id: number, state: unknown, extra: Record = {}): StorageWrite => ({ + type: "task", + value: { + id, + conversationId: ROOT_CONVERSATION_ID, + kind, + version: 1, + input: { name: `seeded${id}` }, + state, + background: false, + abortRequested: false, + ...extra, + } as never, + }); + const sleeping = (until: number) => ({ status: "sleeping", checkpoint: { phase: "run" }, until }); + const SLEEPS = record(10, sleeping(5_000)); + + it.each([ + ["a sleeping task", [SLEEPS], { active: false, at: 5_000 }], + [ + "an abort-marked sleeping task, which runs its handler", + [SLEEPS, record(11, sleeping(3_000), { abortRequested: true })], + { active: true, at: 5_000 }, + ], + ["a sleeping task that is due", [record(11, sleeping(500))], { active: true }], + [ + "a task waiting on a live one", + [SLEEPS, record(11, { status: "waiting", checkpoint: { phase: "run" }, on: [10], policy: "allSettled" })], + { active: false, at: 5_000 }, + ], + [ + "a completing task holding for a sleeping child", + [ + record(11, { status: "completing", outcome: { status: "completed", result: null } }), + record(10, sleeping(5_000), { owner: 11 }), + ], + { active: false, at: 5_000 }, + ], + [ + "a task no definition can take", + [SLEEPS, record(11, { status: "pending", checkpoint: { phase: "run" } }, { kind: "test.idle.missing" })], + { active: false, at: 5_000 }, + ], + [ + "a pending task", + [SLEEPS, record(11, { status: "pending", checkpoint: { phase: "run" } })], + { active: true, at: 5_000 }, + ], + ] as const)("with %s", async (_name, writes, expected) => { + const storage = new MemoryStorage(); + await commitWrites(storage, [{ type: "conversation", value: { id: ROOT_CONVERSATION_ID } }, ...writes], context); + const { harness } = await openTasks(storage, [Work], { now: () => 1_000 }); + await quiet(); + expect(harness.wake()).toEqual(expected); + await harness.close(context); + }); + + it("with an invocation running: active, and the earliest sleep still found", async () => { + const storage = new MemoryStorage(); + await commitWrites(storage, [{ type: "conversation", value: { id: ROOT_CONVERSATION_ID } }, SLEEPS], context); + const { harness } = await openTasks(storage, [Work], { now: () => 1_000 }); + harness.resume(); + const root = await harness.root(context); + const id = await root.commit( + (tx) => tx.createTask(Work, { name: "running" }, { ownership: { kind: "conversation" } }), + context, + ); + await eventually(async () => (await harness.getTask(id, context))!.state.status === "running"); + expect(harness.wake()).toEqual({ active: true, at: 5_000 }); + named(gates, "running").resolve(); + await harness.waitForTask(id, context); + await quiet(); + expect(harness.wake()).toEqual({ active: false, at: 5_000 }); + await harness.close(context); + }); +});