diff --git a/NEW-ARCH.md b/NEW-ARCH.md new file mode 100644 index 0000000..b6431aa --- /dev/null +++ b/NEW-ARCH.md @@ -0,0 +1,122 @@ +# All-quiescence scheduling: a simpler architecture, not (yet) adopted + +This document writes up a simplification of the scheduler that came out of +fixing cross-task promise sharing (a task awaiting a promise settled by +another task used to hang the simulation silently). It describes the +current two-path architecture, the proposed single-path alternative, and +the trade-offs that kept the alternative out of the codebase for now. + +## Background: what the bookkeeping can and cannot know + +The scheduler tracks each task in one of three states (`TaskInfo.resolve`): + +| State | Meaning | Nature | +| --- | --- | --- | +| a function | parked at a checkpoint/failpoint; the function is its continuation | **fact** — the scheduler holds the wake mechanism | +| `false` | blocked (sleeping on a timer, waiting on a mutex/CV) | **fact** — a timer fire or a notify routes the wake back through the scheduler | +| `undefined` | "running" | **prediction** — "this task will re-enter the scheduler at its next park or finish" | + +The first two states are exact because nothing outside the scheduler can +invalidate them. The third is a bet, and JavaScript provides no hook at +the moment a task awaits something, so the scheduler cannot observe +suspension — it can only assume re-entry. The bet is wrong in exactly one +situation: the task awaited a promise the scheduler doesn't manage +(another task's promise, a bare deferred). And the wrongness is +observable at exactly one kind of moment: *quiescence*, when the +microtask queue has drained and the predicted re-entry has demonstrably +not happened. + +## The current architecture: two paths + +1. **Synchronous path** (`unlockIfNecessary` → `scheduleNext`): runs + inside park/finish calls. If any task is marked running, it does + nothing (someone will re-enter). Otherwise every state is a fact, and + it is safe to make scheduling decisions immediately and to fire + several timers in a row until a task becomes schedulable. +2. **Quiescence probe** (`armQuiescenceProbe` → `onQuiescence`): armed + whenever the scheduler hands control to user code; fires as a + macrotask, i.e. strictly after the whole microtask queue has drained. + A task still marked running at that point is provably suspended on an + unmanaged promise, and the probe treats it as blocked. Unlike the + synchronous path it fires at most **one** timer per probe and then + re-arms: a fired timer may settle promises via deadline abort + listeners, and those wake-ups sit in the microtask queue until the + probe returns — firing further timers would advance time past a + wake-up already in flight, or misreport a deadlock. Errors raised in a + probe have no task stack, so they fail `runTasks` through an + out-of-band rejection channel (`Promise.race`). + +One-line summary: **bookkeeping where it's fact, observation where it's +prediction.** + +## The proposed simplification: quiescence as the only scheduling point + +Delete the synchronous path. Park calls only register their continuation +and arm the probe; *every* scheduling decision — picking a parked task, +firing a timer, declaring deadlock — happens in a probe, at quiescence. + +What this removes: + +- `unlockIfNecessary` and its someone-is-running early return. The + "running" state stops mattering entirely: at quiescence nobody is + running, so the scheduler never needs to ask. +- The dual firing rule. Fire-one-then-yield becomes the only rule, and + it is the safer one everywhere. +- `unlockIfNecessaryAfterPark` and its rollback contract. It exists only + because the synchronous path can throw *through* a parking task's + stack, which corrupts the park slot unless carefully rolled back. With + no synchronous scheduling, park calls that don't themselves consume + entropy cannot throw, and the subtlest exception-path code in the + scheduler disappears. (`sleep`'s rollback survives in reduced form: + timer registration can still raise a trace-divergence error in task + context.) +- The two-channel error story. All scheduling errors go through the + out-of-band channel; only task-attributable errors (failpoint draws, + `task.random`, timer-creation divergence) still throw in task context. + +Scheduling decisions would depend on the same scheduler states in the +same order — a decision made "immediately when the last task parks" +observes the same candidates and draws the same entropy as the same +decision deferred to the quiescence that immediately follows — so +recorded traces are expected to replay unchanged for closed-world +workloads. This must be verified (golden traces recorded on the current +scheduler, replayed on the new one) before adopting. + +## Why it hasn't been adopted + +1. **It changes the error-injection contract.** Today an entropy source + that throws from a scheduling pick (a DST resource guard tripping) + throws synchronously into whichever task happened to park last, and + that task can catch it and recover — documented, tested behavior. In + the all-quiescence design the pick happens in a macrotask, so every + scheduling error fails the run out-of-band; no task can intercept it. + Arguably cleaner (the current attribution — "whoever parked last eats + the throw" — is accidental), but it is a deliberate breaking change, + not a refactor, and it deletes a feature: recoverable entropy-guard + trips. +2. **Every scheduling step costs a macrotask hop.** The current + synchronous path schedules thousands of steps per run purely in + microtasks. Deferring every decision to `setImmediate` adds a real + constant factor to step-dense simulations, and DST workloads run many + iterations. +3. **The win is smaller than it looks.** The probe machinery (arming, + run tokens, the out-of-band channel) is needed in both designs; the + synchronous path that would be deleted is the simple, well-tested + part. The genuinely subtle deletion — the park rollback contract — + only shrinks, because timer-registration divergence still throws in + task context. + +## When to adopt it + +In a major version, if/when breaking the "tasks can catch scheduler +throws from park calls" contract is acceptable. The migration should: + +- move all scheduling-pick entropy draws and deadlock/budget errors to + the out-of-band channel, keeping task-context throws only for + failpoint draws, `task.random`, and timer-creation divergence; +- verify trace compatibility with golden traces recorded on the old + scheduler (same scenario, replay must fully consume); +- benchmark step-dense simulations to size the macrotask-hop cost; +- re-run the existing suite expecting failures *only* in tests that + assert the sync error-injection contract, and rewrite those as + out-of-band assertions. diff --git a/README.md b/README.md index 916bc3a..f5d1d7d 100644 --- a/README.md +++ b/README.md @@ -75,6 +75,10 @@ Returns `Result` — either the array of results (in spec order) or **Scheduling algorithm**: When all running tasks have reached a checkpoint or blockpoint, the scheduler picks one of the checkpointed tasks using `sample()` (entropy-driven). Blocked tasks are excluded. If no tasks are checkpointed and all are blocked, a deadlock error is raised. +**Cross-task promises**: A task may await a promise that another task will settle — the singleflight/coalescing shape, or a `Promise.all` spanning tasks. The scheduler's task states are facts except one: "parked" and "blocked" are exact because the scheduler itself holds the wake mechanism, but "running" is a *prediction* — "this task will re-enter the scheduler at its next park or finish" — and awaiting a promise the scheduler doesn't manage breaks exactly that prediction. So whenever the scheduler hands control to user code, it arms a one-shot *quiescence probe* (a macrotask, which runs only once the entire microtask queue has drained). A task still marked running when the probe fires is provably suspended on an unmanaged promise, and the probe treats it as blocked: parked tasks keep running, pending timers fire one at a time — yielding after each fire so that settling cascades (e.g. a deadline abort listener resolving a deferred) drain before the next decision — and virtual time advances until the awaited promise settles. If nothing can progress — a task awaits a promise nobody will ever settle and no timers are pending — the run fails with a deadlock report naming the task as awaiting a promise not managed by the simulation, instead of hanging silently. + +Two consequences to be aware of: tasks resumed by a promise settling run in native promise-reaction order, not entropy order — deterministic and replayable, but that resumption order is not part of the explored schedule space; and the quiescence inference assumes a closed world — a task awaiting real async work (I/O, real timers) looks identical to one awaiting an unmanaged promise, so the scheduler may advance virtual time while that work is still in flight. Real async work inside simulation tasks is outside the library's contract regardless, since it breaks determinism on its own. + #### Deterministic time The simulation has a virtual monotonic clock, starting at 0 per run. Time never passes for real: `task.sleep(3_600_000, "one hour")` completes instantly in real time. @@ -381,9 +385,10 @@ npm run typecheck See [TIMERS-SPEC.md](./TIMERS-SPEC.md) for the full requirements and semantics of deterministic time. -- All concurrency is cooperative, not preemptive. Tasks only yield control at explicit `checkpoint`, `failpoint`, `blockpoint`, or `sleep` calls. +- All concurrency is cooperative, not preemptive. Tasks only yield control at explicit `checkpoint`, `failpoint`, `blockpoint`, or `sleep` calls — or by awaiting a promise another task settles (see "Cross-task promises" above). - A task is a single sequential coroutine: never race two parking operations (checkpoints, failpoints, sleeps, mutex/condition-variable waits) within one task via `Promise.race`/`Promise.all`. The scheduler models exactly one park per task. Concurrency is expressed with multiple tasks; bounding work with a deadline is expressed with `withTimedSignal` and cooperative cancellation. - Timer durations are lower bounds, and firing order is entropy-controlled — never depend on the relative firing order of pending timers. -- The simulation runs in a single JS event loop turn between scheduling decisions. There is no actual parallelism. +- The simulation makes scheduling decisions synchronously (in microtasks) while its bookkeeping is provably exact, and defers to a macrotask-based quiescence probe only when a task might be suspended on an unmanaged promise. There is no actual parallelism. [NEW-ARCH.md](./NEW-ARCH.md) writes up a simpler all-quiescence alternative and why it hasn't been adopted. +- Scheduling errors (a deadlock, an exhausted step budget, an entropy source that throws from a scheduling pick) normally propagate synchronously into the task whose park call triggered the decision — which is what lets a task catch and recover from a transient entropy-guard trip. Errors raised from a quiescence probe have no task stack, so they fail `runTasks` directly through an out-of-band channel. - `SimulationImpl` should be treated as single-use per `runTasks` call. After a failed run (error or deadlock), the instance is permanently poisoned (`abortedWithError` is never reset) and subsequent `runTasks` calls will immediately fail. - The `sample()` function's "no entropy for single item" optimization is critical for replay correctness — it ensures the entropy consumption sequence doesn't depend on transient pool sizes. diff --git a/package-lock.json b/package-lock.json index 80defa4..dad08cb 100644 --- a/package-lock.json +++ b/package-lock.json @@ -1,12 +1,12 @@ { "name": "determined", - "version": "0.4.0", + "version": "0.4.1", "lockfileVersion": 3, "requires": true, "packages": { "": { "name": "determined", - "version": "0.4.0", + "version": "0.4.1", "license": "MIT", "dependencies": { "@glideapps/ts-necessities": "^2.4.0", diff --git a/package.json b/package.json index 6bdf107..3ac8483 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "determined", - "version": "0.4.0", + "version": "0.4.1", "type": "module", "license": "MIT", "repository": { diff --git a/simulation.ts b/simulation.ts index f296732..c4cbc9a 100644 --- a/simulation.ts +++ b/simulation.ts @@ -5,6 +5,14 @@ import { ApplicationFailure, CancellationError } from "./errors.ts"; import { isTimerTraceSink, type TimerTraceSink } from "./trace.ts"; import { err, ok, type Result } from "neverthrow"; +/** + * Schedules a macrotask: runs strictly after the entire microtask queue — + * including all promise-continuation chains — has drained. `setImmediate` + * where available (Node, Bun), `setTimeout(0)` otherwise. + */ +const scheduleMacrotask: (f: () => void) => void = + typeof setImmediate === "function" ? setImmediate : (f) => setTimeout(f, 0); + export interface Logger { log(...log: readonly unknown[]): void; error(...log: readonly unknown[]): void; @@ -339,6 +347,17 @@ export class SimulationImpl implements Simulation { * stability, so they fail descriptively instead. */ private inUserAbortListener = false; + /** Bumped per `runTasks` call so quiescence probes from a previous run no-op. */ + private runToken = 0; + /** True while a quiescence probe is scheduled for the current run. */ + private probeArmed = false; + /** + * Rejects the current run's out-of-band failure channel. Errors raised + * from a quiescence probe (deadlock, exhausted budgets, trace + * divergence) have no task stack to propagate through, so they fail + * `runTasks` directly via this instead. + */ + private rejectRun: ((e: unknown) => void) | undefined; constructor( logger: Logger, @@ -394,17 +413,20 @@ export class SimulationImpl implements Simulation { } } + private recordAbort(e: unknown): void { + if (this.abortedWithError !== undefined) return; + this.abortedWithError = e; + // Cancellation cleanup: pending timers must not survive an + // aborted run. (unlockIfNecessary rethrows the abort error + // before ever firing a timer, but there's no reason to keep + // them around.) + this.timers.clear(); + this.activeDeadlines.clear(); + this.logger.error(`Aborting simulation: ${exceptionToString(e)}`); + } + private abort(e: unknown): never { - if (this.abortedWithError === undefined) { - this.abortedWithError = e; - // Cancellation cleanup: pending timers must not survive an - // aborted run. (unlockIfNecessary rethrows the abort error - // before ever firing a timer, but there's no reason to keep - // them around.) - this.timers.clear(); - this.activeDeadlines.clear(); - this.logger.error(`Aborting simulation: ${exceptionToString(e)}`); - } + this.recordAbort(e); throw e; } @@ -503,7 +525,15 @@ export class SimulationImpl implements Simulation { private makeDeadlockError(infos: readonly TaskInfo[]): Error { const blocked = infos - .map((i) => `${i.name}${i.parkReason !== undefined ? ` (${i.parkReason})` : ""}`) + .map((i) => { + // A task still marked running at a quiescence probe is + // suspended on a promise the simulation doesn't manage — + // name that, since it has no parkReason of its own. + const reason = + i.parkReason ?? + (i.resolve === undefined ? "awaiting a promise not managed by the simulation" : undefined); + return `${i.name}${reason !== undefined ? ` (${reason})` : ""}`; + }) .join(", "); let message = `Deadlock at t=${this.monotonic}ms: all tasks are blocked and no timers are pending. Blocked tasks: ${blocked}.`; // A blocked task holding an already-aborted signal is the typical @@ -527,9 +557,46 @@ export class SimulationImpl implements Simulation { const infos = Array.from(this.taskInfos.values()); if (infos.some((i) => i.resolve === undefined)) { - // Some tasks are still running, so there's nothing to do yet. + // Some tasks are still running, so there's nothing to do yet — + // but "running" is trusted bookkeeping, not observed fact: a + // task that awaits a promise the simulation doesn't manage + // stays marked running without ever re-entering the scheduler. + // The quiescence probe re-checks once the runtime has actually + // gone idle. + this.armQuiescenceProbe(); return; } + this.scheduleNext(infos); + } + + /** + * Picks and resumes an entropy-chosen task among `candidates` (all + * parked at a checkpoint). Tasks marked running (`resolve === + * undefined`) are never candidates: callers guarantee they are either + * genuinely running (unlockIfNecessary, which requires none) or + * provably suspended on unmanaged promises (the quiescence probe). + */ + private resumeCandidate(infos: readonly TaskInfo[], candidates: readonly TaskInfo[]): void { + this.countStep(); + const info = this.pickTask(candidates); + this.logger.log(`${info.name} UNBLOCKED at t=${this.monotonic}ms from ${infos.map((i) => i.name).join(", ")}`); + const { resolve } = info; + assert(resolve !== undefined && resolve !== false); + info.resolve = undefined; + info.parkReason = undefined; + resolve(); + // The resumed task may suspend on an unmanaged promise instead of + // re-entering the scheduler; the probe catches that. + this.armQuiescenceProbe(); + } + + /** + * The synchronous scheduling path: every task is parked or blocked + * (none marked running), so timer fires can only flip scheduler + * bookkeeping and it is safe to fire several in a row until a task + * becomes schedulable. + */ + private scheduleNext(infos: readonly TaskInfo[]): void { const checkpointed = () => infos.filter((i) => i.resolve !== undefined && i.resolve !== false); let candidates = checkpointed(); while (candidates.length === 0) { @@ -543,14 +610,75 @@ export class SimulationImpl implements Simulation { this.fireNextTimer(); candidates = checkpointed(); } - this.countStep(); - const info = this.pickTask(candidates); - this.logger.log(`${info.name} UNBLOCKED at t=${this.monotonic}ms from ${infos.map((i) => i.name).join(", ")}`); - const { resolve } = info; - assert(resolve !== undefined && resolve !== false); - info.resolve = undefined; - info.parkReason = undefined; - resolve(); + this.resumeCandidate(infos, candidates); + } + + /** + * Arms a one-shot probe that runs when the JS runtime goes quiescent. + * A macrotask runs only after the whole microtask queue — every + * pending task continuation — has drained, so state observed by the + * probe is settled fact, not in-flight bookkeeping. One pending probe + * covers an entire synchronous/microtask cascade, however many + * scheduler entries it contains. + */ + private armQuiescenceProbe(): void { + if (this.probeArmed) return; + if (this.abortedWithError !== undefined) return; + if (this.taskInfos.size === 0) return; + const token = this.runToken; + this.probeArmed = true; + scheduleMacrotask(() => { + // A stale probe from an earlier run must not touch the current + // run's state — including its probeArmed flag. + if (token !== this.runToken) return; + this.probeArmed = false; + try { + this.onQuiescence(); + } catch (e) { + // No task stack to throw into: record the abort and fail + // the run through the out-of-band channel. + this.recordAbort(e); + this.rejectRun?.(e); + } + }); + } + + /** + * Runs at quiescence: no task code on the stack, microtask queue + * empty. A task still marked running now is not running — it is + * suspended on a promise the simulation doesn't manage (another + * task's promise, a bare deferred). Treat those tasks as blocked and + * schedule one step: parked tasks keep running, pending timers fire + * and advance virtual time (settling cross-task promise shapes like + * singleflight), and if nothing can progress the run fails loudly + * with a deadlock report instead of hanging silently. + */ + private onQuiescence(): void { + if (this.abortedWithError !== undefined) return; + if (this.taskInfos.size === 0) return; + const infos = Array.from(this.taskInfos.values()); + if (!infos.some((i) => i.resolve === undefined)) { + // Nobody is suspended on an unmanaged promise; the scheduler + // already handled this state synchronously. + return; + } + const candidates = infos.filter((i) => i.resolve !== undefined && i.resolve !== false); + if (candidates.length > 0) { + this.resumeCandidate(infos, candidates); + return; + } + if (this.timers.size === 0) { + throw this.makeDeadlockError(infos); + } + // Unlike the synchronous path, fire exactly ONE timer per probe: a + // deadline timer's abort listeners may settle promises whose + // cascades resume suspended tasks, and those continuations sit in + // the microtask queue until this macrotask returns. Firing further + // timers here could advance time past a wake-up already in flight, + // or misreport a deadlock. Re-arming yields until the cascade has + // drained, then the next probe re-evaluates. + this.fireNextTimer(); + this.armQuiescenceProbe(); } public async runTasks[]>( @@ -574,6 +702,13 @@ export class SimulationImpl implements Simulation { this.activeDeadlines.clear(); this.steps = 0; this.lastAdvanceStep = 0; + this.runToken++; + this.probeArmed = false; + let rejectRun!: (e: unknown) => void; + const runFailure = new Promise((_resolve, reject) => { + rejectRun = reject; + }); + this.rejectRun = rejectRun; const tasksAndInfos = specs.map((s) => { const info: TaskInfo = { name: s.name, resolve: undefined, parkReason: undefined }; @@ -820,23 +955,28 @@ export class SimulationImpl implements Simulation { // In the try so that a replay divergence on the recorded epoch // fails the run like any other divergence. this.timerTrace?.runStart(this.wallClockEpoch); - const results = (await Promise.all( - tasksAndInfos.map(([s, task]) => { - return task - .checkpoint("START") - .then(() => s.f(task)) - .catch((e) => this.abort(e)) - .finally(() => { - this.taskInfos.delete(task); - this.logger.log( - `FINISHED ${s.name}, still left ${Array.from(this.taskInfos.values()) - .map((i) => i.name) - .join(", ")}`, - ); - this.unlockIfNecessary(); - }); - }), - )) as any; // I wish we could type this better + // Racing against `runFailure` lets a quiescence probe fail the + // run even though the stuck tasks' promises never settle. + const results = (await Promise.race([ + Promise.all( + tasksAndInfos.map(([s, task]) => { + return task + .checkpoint("START") + .then(() => s.f(task)) + .catch((e) => this.abort(e)) + .finally(() => { + this.taskInfos.delete(task); + this.logger.log( + `FINISHED ${s.name}, still left ${Array.from(this.taskInfos.values()) + .map((i) => i.name) + .join(", ")}`, + ); + this.unlockIfNecessary(); + }); + }), + ), + runFailure, + ])) as any; // I wish we could type this better return ok(results); } catch (e: unknown) { // A synchronous throw out of a START checkpoint (e.g. an entropy diff --git a/time.test.ts b/time.test.ts index b86cd45..4a9aa29 100644 --- a/time.test.ts +++ b/time.test.ts @@ -3,6 +3,7 @@ import assert from "node:assert"; import { defined } from "@glideapps/ts-necessities"; import { NoSimulationTask, noSimulation, SimulationImpl, type SimulationTask } from "./simulation.ts"; import { RecordingEntropySource, ReplayingEntropySource, SimpleEntropySource } from "./entropy.ts"; +import { RecordingTraceSource, ReplayingTraceSource } from "./trace.ts"; import { isApplicationFailure, isCancellation } from "./errors.ts"; import { Mutex } from "./mutex.ts"; import { ConditionVariable } from "./condition-variable.ts"; @@ -1097,3 +1098,549 @@ describe("timer pick policy", () => { assert.deepStrictEqual(result.value, [1_000]); }); }); + +describe("cross-task promise sharing", () => { + // These tests cover DETERMINED-BUG.md: a task awaiting a promise that + // will be settled by another task (the singleflight/coalescing shape) + // must complete — the scheduler must advance time to pending timers + // even while a task is suspended on a promise it doesn't manage, and + // must fail loudly (never hang silently) when nothing can progress. + // + // A hang can't fail a test by itself, so every runTasks is raced + // against a real-time timeout that turns a hang into an assertion + // failure instead of a stuck test runner. + const HANG = Symbol("hang"); + async function withHangGuard(run: Promise): Promise { + let timer: NodeJS.Timeout | undefined; + const timeout = new Promise((resolve) => { + timer = setTimeout(() => resolve(HANG), 2_000); + }); + const raced = await Promise.race([run, timeout]); + clearTimeout(timer); + if (raced === HANG) assert.fail("simulation hung: runTasks never settled"); + return raced; + } + + function makeCrossTaskSim(entropy: ConstructorParameters[1]): SimulationImpl { + return new SimulationImpl(new ArrayLogger(), entropy, () => 0, { + maxSchedulingSteps: 10_000, + maxVirtualDurationMs: 60_000, + }); + } + + // The direct repro from DETERMINED-BUG.md. Always-zero entropy forces + // the deadlocking schedule: the owner runs first, publishes its + // promise, and sleeps; the waiter then awaits the foreign promise + // directly. (Under random entropy the scheduler can happen to fire the + // owner's timer before the waiter awaits, masking the bug.) + it("a task awaiting a promise settled by another task's timer completes", async () => { + const sim = makeCrossTaskSim({ random: () => 0 }); + let shared: Promise | undefined; + let settled = false; + let waiterSawSettled: boolean | undefined; + const result = await withHangGuard( + sim.runTasks([ + { + name: "owner", + f: async (task: SimulationTask) => { + shared = task.sleep(10, "work").then(() => { + settled = true; + return "done"; + }); + return await shared; + }, + }, + { + name: "waiter", + f: async (task: SimulationTask) => { + // Wait until the owner has published its promise, then await it. + while (shared === undefined) await task.sleep(1, "poll"); + waiterSawSettled = settled; + return await shared; + }, + }, + ]), + ); + assert.ok(result.isOk(), `expected ok, got err: ${result.isErr() ? result.error.message : ""}`); + assert.deepStrictEqual(result.value, ["done", "done"]); + // Guard the repro itself: the waiter must have suspended on the + // promise while it was still pending — otherwise this test isn't + // exercising the cross-task wait at all. + assert.strictEqual(waiterSawSettled, false); + }); + + // The motivating use case: singleflight. Several waiters coalesce on + // one in-flight operation and all receive its result. Waiters resumed + // by the shared promise settling run in promise-reaction order, not + // entropy order — so also assert the whole run is deterministic by + // running it twice and comparing the recorded resume orders. + it("multiple waiters coalescing on one in-flight promise all complete deterministically", async () => { + async function runScenario(): Promise<{ values: readonly string[]; order: string[] }> { + const sim = makeCrossTaskSim({ random: () => 0 }); + const order: string[] = []; + let inFlight: Promise | undefined; + const fetchOnce = (task: SimulationTask): Promise => { + inFlight ??= task.sleep(50, "fetch").then(() => "payload"); + return inFlight; + }; + const makeWaiter = (name: string) => ({ + name, + f: async (task: SimulationTask) => { + while (inFlight === undefined) await task.sleep(1, "poll"); + const value = await inFlight; + order.push(name); + return value; + }, + }); + const result = await withHangGuard( + sim.runTasks([ + { + name: "fetcher", + f: async (task: SimulationTask) => { + const value = await fetchOnce(task); + order.push("fetcher"); + return value; + }, + }, + makeWaiter("waiterA"), + makeWaiter("waiterB"), + makeWaiter("waiterC"), + ]), + ); + assert.ok(result.isOk(), `expected ok, got err: ${result.isErr() ? result.error.message : ""}`); + return { values: result.value, order }; + } + + const first = await runScenario(); + assert.deepStrictEqual(first.values, ["payload", "payload", "payload", "payload"]); + assert.strictEqual(first.order.length, 4); + const second = await runScenario(); + assert.deepStrictEqual(second.order, first.order); + }); + + // A task suspended on a foreign promise must not stall the rest of the + // simulation: a checkpoint-parked task keeps running, at the current + // virtual time, before any timer fires. + it("a checkpointing task keeps running while another task awaits a foreign promise", async () => { + const sim = makeCrossTaskSim({ random: () => 0 }); + let shared: Promise | undefined; + const workerTimes: number[] = []; + const result = await withHangGuard( + sim.runTasks([ + { + name: "owner", + f: async (task: SimulationTask) => { + shared = task.sleep(10, "work").then(() => "done"); + return await shared; + }, + }, + { + name: "waiter", + f: async (task: SimulationTask) => { + while (shared === undefined) await task.sleep(1, "poll"); + return await shared; + }, + }, + { + name: "worker", + f: async (task: SimulationTask) => { + workerTimes.push(task.monotonicNow()); + await task.checkpoint("mid"); + workerTimes.push(task.monotonicNow()); + return "worked"; + }, + }, + ]), + ); + assert.ok(result.isOk(), `expected ok, got err: ${result.isErr() ? result.error.message : ""}`); + assert.deepStrictEqual(result.value, ["done", "done", "worked"]); + // The worker never sleeps, so it must have run to completion at + // t=0 — before the scheduler advanced time to the owner's timer. + assert.deepStrictEqual(workerTimes, [0, 0]); + }); + + // A foreign promise settled directly by another running task — no + // timer involved. The waiter is listed first so always-zero entropy + // schedules it first and it suspends on the still-unsettled promise; + // the scheduler must keep running the resolver and must not misreport + // a deadlock while the settling cascade is still in the microtask + // queue. + it("a foreign promise settled by another task without a timer completes", async () => { + const sim = makeCrossTaskSim({ random: () => 0 }); + let resolveShared!: (value: string) => void; + const shared = new Promise((resolve) => { + resolveShared = resolve; + }); + const result = await withHangGuard( + sim.runTasks([ + { + name: "waiter", + f: async () => { + return await shared; + }, + }, + { + name: "resolver", + f: async (task: SimulationTask) => { + await task.checkpoint("before resolve"); + resolveShared("payload"); + return "resolved"; + }, + }, + ]), + ); + assert.ok(result.isOk(), `expected ok, got err: ${result.isErr() ? result.error.message : ""}`); + assert.deepStrictEqual(result.value, ["payload", "resolved"]); + }); + + // Promise.all spanning tasks: the waiter joins on two promises settled + // by two different tasks' timers. + it("Promise.all over promises from different tasks completes", async () => { + const sim = makeCrossTaskSim({ random: () => 0 }); + let sharedA: Promise | undefined; + let sharedB: Promise | undefined; + const result = await withHangGuard( + sim.runTasks([ + { + name: "ownerA", + f: async (task: SimulationTask) => { + sharedA = task.sleep(10, "work A").then(() => "A"); + return await sharedA; + }, + }, + { + name: "ownerB", + f: async (task: SimulationTask) => { + sharedB = task.sleep(20, "work B").then(() => "B"); + return await sharedB; + }, + }, + { + name: "waiter", + f: async (task: SimulationTask) => { + while (sharedA === undefined || sharedB === undefined) await task.sleep(1, "poll"); + return await Promise.all([sharedA, sharedB]); + }, + }, + ]), + ); + assert.ok(result.isOk(), `expected ok, got err: ${result.isErr() ? result.error.message : ""}`); + assert.deepStrictEqual(result.value, ["A", "B", ["A", "B"]]); + }); + + // A timer whose only effect is settling a foreign promise: a deadline + // whose abort listener resolves a deferred (listeners are documented + // as allowed to "settle a low-level promise"). Firing it produces no + // schedulable task synchronously — the wake-up sits in the microtask + // queue — so the scheduler must wait for the cascade instead of + // misreporting a deadlock. + it("a timer that settles only a foreign promise does not deadlock", async () => { + const sim = makeCrossTaskSim({ random: () => 0 }); + let resolveShared!: (value: string) => void; + const shared = new Promise((resolve) => { + resolveShared = resolve; + }); + const result = await withHangGuard( + sim.runTasks([ + { + name: "creator", + f: async (task: SimulationTask) => { + const deadline = task.createDeadline(5, "settle shared"); + deadline.signal.addEventListener("abort", () => resolveShared("settled"), { once: true }); + return "created"; + }, + }, + { + name: "waiter", + f: async (task: SimulationTask) => { + // A deeper reaction chain than a bare await: the + // whole cascade must drain before the scheduler + // decides anything. + const value = await shared.then((v) => v).then((v) => v); + return `${value} at t=${task.monotonicNow()}ms`; + }, + }, + ]), + ); + assert.ok(result.isOk(), `expected ok, got err: ${result.isErr() ? result.error.message : ""}`); + assert.deepStrictEqual(result.value, ["created", "settled at t=5ms"]); + }); + + // Same shape plus a later timer: after the settling timer fires, the + // scheduler must let the wake-up cascade drain rather than firing the + // later timer too — the waiter must observe t=5, not t=1000. + it("a later timer does not fire before a settling cascade has drained", async () => { + const sim = makeCrossTaskSim({ random: () => 0 }); + let resolveShared!: (value: string) => void; + const shared = new Promise((resolve) => { + resolveShared = resolve; + }); + const result = await withHangGuard( + sim.runTasks([ + { + // Runs first (always-zero entropy), so its timer is + // created first and the always-zero timer pick fires it + // first. + name: "creator", + f: async (task: SimulationTask) => { + const deadline = task.createDeadline(5, "settle shared"); + deadline.signal.addEventListener("abort", () => resolveShared("settled"), { once: true }); + return "created"; + }, + }, + { + name: "sleeper", + f: async (task: SimulationTask) => { + await task.sleep(1_000, "long nap"); + return `woke at t=${task.monotonicNow()}ms`; + }, + }, + { + name: "waiter", + f: async (task: SimulationTask) => { + const value = await shared; + return `${value} at t=${task.monotonicNow()}ms`; + }, + }, + ]), + ); + assert.ok(result.isOk(), `expected ok, got err: ${result.isErr() ? result.error.message : ""}`); + assert.deepStrictEqual(result.value, ["created", "woke at t=1000ms", "settled at t=5ms"]); + }); + + // Promises settled by the runtime itself — already resolved, or via + // queueMicrotask — resume their awaiters during the microtask drain, + // before the scheduler's quiescence check. None of these may be + // misreported as a deadlock. + it("promises settled outside the scheduler do not false-deadlock", async () => { + const sim = makeCrossTaskSim({ random: () => 0 }); + const result = await withHangGuard( + sim.runTasks([ + { + name: "resolved", + f: async (task: SimulationTask) => { + await Promise.resolve(); + await task.checkpoint("mid"); + return await Promise.resolve("a"); + }, + }, + { + name: "microtask", + f: async () => { + await new Promise((resolve) => queueMicrotask(resolve)); + return "b"; + }, + }, + { + name: "nested", + f: async () => { + return await Promise.resolve("c") + .then((v) => v) + .then((v) => Promise.resolve(v)) + .then((v) => v); + }, + }, + ]), + ); + assert.ok(result.isOk(), `expected ok, got err: ${result.isErr() ? result.error.message : ""}`); + assert.deepStrictEqual(result.value, ["a", "b", "c"]); + }); + + // A completed run may leave a quiescence probe pending. It must be + // inert: no entropy draws, no log output, no state mutation, and no + // poisoning of the instance. + it("pending probes from a completed run are inert", async () => { + const logger = new ArrayLogger(); + // FixedEntropySource throws on any draw, so an extra draw from a + // post-completion probe would surface as a failure below. + const sim = new SimulationImpl(logger, new FixedEntropySource([]), () => 0); + const solo = { + name: "solo", + f: async (task: SimulationTask) => { + await task.sleep(1, "nap"); + return "one"; + }, + }; + const first = await withHangGuard(sim.runTasks([solo])); + assert.ok(first.isOk(), `expected ok, got err: ${first.isErr() ? first.error.message : ""}`); + const logCount = logger.logs.length; + // Let any pending probes fire. + await new Promise((resolve) => setImmediate(() => setImmediate(resolve))); + assert.strictEqual(logger.logs.length, logCount, "a post-completion probe did something"); + // Not poisoned: the instance is still usable. + const again = await withHangGuard(sim.runTasks([solo])); + assert.ok(again.isOk(), `expected ok, got err: ${again.isErr() ? again.error.message : ""}`); + }); + + // Back-to-back runs on one instance, where the second run starts while + // the first run's probe is still pending: the stale probe must not + // touch the new run's state, and the new run's own probes must work. + it("a stale probe from an earlier run does not affect a cross-task run started immediately", async () => { + const sim = makeCrossTaskSim({ random: () => 0 }); + const first = await withHangGuard( + sim.runTasks([ + { + name: "solo", + f: async (task: SimulationTask) => { + await task.sleep(1, "nap"); + return "one"; + }, + }, + ]), + ); + assert.ok(first.isOk(), `expected ok, got err: ${first.isErr() ? first.error.message : ""}`); + + // No event-loop turn in between: run 2 starts with run 1's probe + // still in flight. + let shared: Promise | undefined; + const second = await withHangGuard( + sim.runTasks([ + { + name: "owner", + f: async (task: SimulationTask) => { + shared = task.sleep(10, "work").then(() => "done"); + return await shared; + }, + }, + { + name: "waiter", + f: async (task: SimulationTask) => { + while (shared === undefined) await task.sleep(1, "poll"); + return await shared; + }, + }, + ]), + ); + assert.ok(second.isOk(), `expected ok, got err: ${second.isErr() ? second.error.message : ""}`); + assert.deepStrictEqual(second.value, ["done", "done"]); + }); + + // Fail-loudly half of the contract: a task suspended on a promise + // nobody will ever settle, with no pending timers, is a genuine + // deadlock. It must abort with a diagnostic naming the stuck task — + // never hang silently. + it("a task awaiting a promise nobody settles fails loudly instead of hanging", async () => { + const sim = makeCrossTaskSim({ random: () => 0 }); + let resolveLate!: (value: string) => void; + const late = new Promise((resolve) => { + resolveLate = resolve; + }); + let zombieRan = false; + const result = await withHangGuard( + sim.runTasks([ + { + name: "finisher", + f: async (task: SimulationTask) => { + await task.checkpoint("step"); + return "finished"; + }, + }, + { + name: "stuck", + f: async () => { + await late; + zombieRan = true; + return "unreachable"; + }, + }, + ]), + ); + assert.ok(result.isErr(), "expected the simulation to report the deadlock"); + assert.match(result.error.message, /[Dd]eadlock/); + assert.match(result.error.message, /stuck/); + + // Like any failed run, a probe-detected deadlock poisons the + // instance. + const reused = await withHangGuard(sim.runTasks([{ name: "again", f: async () => "again" }])); + assert.ok(reused.isErr(), "expected the poisoned instance to fail"); + + // Settling the promise after the run failed resumes the abandoned + // task's code — that cannot be prevented, but it must not crash the + // process or produce an unhandled rejection. + resolveLate("too late"); + await new Promise((resolve) => setImmediate(() => setImmediate(resolve))); + assert.strictEqual(zombieRan, true, "the abandoned task's continuation should have run"); + }); + + // A budget error raised while the scheduler advances time on behalf of + // a foreign-suspended task has no task stack to propagate through; it + // must still fail the run instead of hanging. + it("exceeding maxVirtualDurationMs while a task awaits a foreign promise fails the run", async () => { + const sim = new SimulationImpl(new ArrayLogger(), { random: () => 0 }, () => 0, { + maxVirtualDurationMs: 5, + }); + let shared: Promise | undefined; + const result = await withHangGuard( + sim.runTasks([ + { + name: "owner", + f: async (task: SimulationTask) => { + shared = task.sleep(10, "too long").then(() => "done"); + return await shared; + }, + }, + { + name: "waiter", + f: async (task: SimulationTask) => { + while (shared === undefined) await task.sleep(1, "poll"); + return await shared; + }, + }, + ]), + ); + assert.ok(result.isErr(), "expected the simulation to report the budget violation"); + assert.match(result.error.message, /Maximum virtual duration/); + }); + + // The cross-task shape must stay deterministic: a recorded run replays + // identically, including the timer firings that resume foreign-blocked + // tasks. + it("cross-task awaits record and replay deterministically", async () => { + async function runScenario( + entropy: ConstructorParameters[1], + ): Promise { + const events: string[] = []; + let shared: Promise | undefined; + const sim = makeCrossTaskSim(entropy); + const result = await withHangGuard( + sim.runTasks([ + { + name: "owner", + f: async (task: SimulationTask) => { + shared = task.sleep(10, "work").then(() => "done"); + const value = await shared; + events.push(`owner got ${value} at t=${task.monotonicNow()}ms`); + }, + }, + { + name: "waiterA", + f: async (task: SimulationTask) => { + while (shared === undefined) await task.sleep(1, "poll"); + const value = await shared; + events.push(`waiterA got ${value} at t=${task.monotonicNow()}ms`); + }, + }, + { + name: "waiterB", + f: async (task: SimulationTask) => { + await task.sleep(3, "delay"); + const value = await defined(shared); + events.push(`waiterB got ${value} at t=${task.monotonicNow()}ms`); + }, + }, + ]), + ); + assert.ok(result.isOk(), `expected ok, got err: ${result.isErr() ? result.error.message : ""}`); + return events; + } + + for (let iteration = 0; iteration < 10; iteration++) { + const recording = new RecordingTraceSource(new SimpleEntropySource()); + const recorded = await runScenario(recording); + const replaying = new ReplayingTraceSource(recording.getTrace()); + const replayed = await runScenario(replaying); + assert.deepStrictEqual(replayed, recorded); + replaying.assertFullyConsumed(); + } + }); +});