diff --git a/.changeset/resolve-loopback-caller.md b/.changeset/resolve-loopback-caller.md new file mode 100644 index 0000000..c2b7e1f --- /dev/null +++ b/.changeset/resolve-loopback-caller.md @@ -0,0 +1,7 @@ +--- +"@mcp-b/do-runtime": patch +--- + +Add `ActorContainer.resolveLoopback()` so in-realm actor bindings use the raw instance only for exact self-calls. Other calls enter the target gate and resume through the exact current, transformed, or structurally supplied caller, including separately bundled runtime copies. + +Verify await-transform coverage against final build modules and warn on development fail-open paths. Reject failed critical sections with `BrokenActorError`, expose queued-entry cancellation, align the `cloudflare-workers` declaration path with its JavaScript entry, and remove the ineffective scheduler WAL pragma. diff --git a/README.md b/README.md index 5417f36..de7c808 100644 --- a/README.md +++ b/README.md @@ -181,19 +181,23 @@ The lifecycle: 1. `await createActorContainer(options)`. 2. `container.start((ctx, env) => new ActorClass(ctx, env))` once, under boot semantics (input gate held for the constructor, deletion receipts replayed first). -3. Expose `container.entry(instance)` to callers. Its `ActorEntry` type makes every method return a promise because each call is one gated event. -4. Use `container.run(fn)` for events that are not method calls: a WebSocket frame, a host callback. +3. Expose `container.entry(instance, signal?)` to callers. Its `ActorEntry` type makes every method return a promise because each call is one gated event. The optional signal is bound to the proxy and cancels only calls still queued for admission. +4. Use `container.run(fn, signal?)` for events that are not method calls: a WebSocket frame, a host callback. Its signal likewise stops only a queued event, not one already running. 5. Reach the platform through `container.globals` (or install it with `installActorScope`). For a host-provided promise an actor must await, wrap it once in `container.awaitIo()`. -6. Watch `container.onBroken`; dispose the placement; recreate it on the next event over the same storage. +6. Watch `container.onBroken`; dispose the placement; recreate it on the next event over the same storage. A failed `blockConcurrencyWhile()` rejects its caller with `BrokenActorError` and breaks the placement with that same error. For a standard Durable Object binding, call `createDurableObjectNamespace(uniqueKey, channel)` and put the result in `env` and `ctx.exports`. The channel maps each routed id to a placed `Fetcher`; that binding works directly with Agents SDK `routeAgentRequest()` and -`getAgentByName()`. When an actor uses the binding to call another actor, wrap -the transport promise with the caller's `container.awaitIo()` so its continuation -re-enters the owning input gate. The extension example shows both the external -router binding and the per-container actor binding. +`getAgentByName()`. For an in-realm binding, pass the raw and entered call +thunks to the target's `container.resolveLoopback()`; it invokes the raw +instance only when the exact caller is that target and otherwise owns the callee +entry and caller `awaitIo`. Current slices and transformed continuations resolve +automatically; pass the still-lock-holding structural caller as the third +argument from untransformed post-await code. For an external transport, wrap +its promise with the caller's `container.awaitIo()` so the continuation +re-enters the owning input gate. ### Storage @@ -217,6 +221,8 @@ On workerd every awaitable thing is an io-context primitive, so "resuming from a `container.globals` is the complete gated set, bound to that container: `setTimeout`/`clearTimeout`/`setInterval`/`clearInterval` capture the critical section when armed and re-enter when fired; `scheduler.wait()` and `scheduler.yield()` resume under the actor; `fetch()` waits for output locks and releases the input gate while in flight; `crypto` re-enters on async completion; accepted WebSocket frames enter through the captured context. Install it as the worker's globals (`installActorScope`) when one worker hosts one root, or hand it to application code explicitly when it must not. +Actor bundles can also install `doRuntimeAwaitTransform()` from `@mcp-b/do-runtime/vite`. A production build checks the final module graph and fails with transformed/total counts for any included module with an uncovered await; the development transform warns once per module if a transformed await reaches its fail-open path without an actor lock. + ## What is not supported The browser cannot reproduce every workerd facility. Where it cannot, the runtime **fails closed**: the API exists, throws a named error that the conformance suite asserts on every lane, and never silently does less. diff --git a/docs/decisions.md b/docs/decisions.md index 7f2717b..06f9dfc 100644 --- a/docs/decisions.md +++ b/docs/decisions.md @@ -53,7 +53,9 @@ Internal `_cf_` names remain reserved to the runtime. `blockConcurrencyWhile()` uses a real nested `CriticalSection`. It blocks peer entries, inherits through supported reentry callbacks, has the workerd -deadline, and permanently breaks the input gate on failure. +deadline, and permanently breaks the input gate on failure. Workerd discards +the caller with its isolate; this same-realm runtime rejects it with the same +typed `BrokenActorError` used to break and abort the placement. ### §1.6 Abort and breakage @@ -113,10 +115,11 @@ MessagePort session. ### §2.1 One application-entry boundary External methods enter through `ActorContainer.entry()` and non-method events -through `run()`. A genuine self-loopback reuses the current slice -(`isCurrentSlice()`). Cross-actor calls pass through the callee's entry -surface and actor-owned I/O. There is no serialized event tail, method-family -dispatch table, or off-tail exception list. +through `run()`. In-realm bindings use `resolveLoopback()`, which invokes the +raw instance only when the exact caller is the target; every other actor call +passes through the callee's entry and the exact current, transformed, or +structurally supplied caller's `awaitIo`. There is no serialized event tail, +method-family dispatch table, or off-tail exception list. ### §2.2 Worker placement is not actor identity @@ -130,8 +133,9 @@ Timers, fetch, crypto, and host I/O are reached through the owning actor's scope — `container.globals`, installed ambiently only where one realm hosts one root (`installActorScope`). A raw host promise that application code must await is wrapped once at its shared owner with `awaitIo`; it is not repaired -at every caller or by patching the realm. No mutable global or ambient field -may select an actor across an `await`. +at every caller or by patching the realm. The await transform's tokenized +continuation marker is the sole exception: it republishes the context captured +from the exact slice, and readers ignore it once that context's lock is gone. ### §2.4 Storage contract @@ -176,20 +180,23 @@ owns only placement and physical storage operations. 3. Use workerd-shaped `CriticalSection` semantics rather than a mutex or queue. 4. Implement `blockConcurrencyWhile()` with that critical section and its - deadline. + deadline; reject its reachable same-realm caller with `BrokenActorError` when + the section breaks the actor. 5. Use a promise-chain output gate and wait before externally observable I/O. 6. Treat a broken gate as fatal to the current container; rebuild from committed storage. 7. Bound implicit transactions by the same hand-off that releases the input gate. 8. Carry explicit actor scope through application-owned code and route raw - host promises through `awaitIo`. Do not restore a one-slot async-context - shim, patch the realm with zones, or infer an actor from a realm-wide - global that can mean more than one. + host promises through `awaitIo`. Do not add a generic async-context shim or + patch the realm with zones. The compile-time await transform may republish + only the exact context it captured, tokenized for one checkpoint and valid + only while that context still holds an input lock. 9. No stream pump or generic remote-facet protocol in the runtime. Facet placement is the host's; direct actor hops use native capabilities. -10. Store per-connection host bridges by connection id. Do not save and - restore an ambient across an `await`. +10. Store per-connection host bridges by connection id. Never use the + transform's actor marker to select a connection or save and restore a host + bridge ambient across an `await`. 11. Run one conformance suite against real workerd, the Node backend, and the browser backend; assert unavailable substrate behavior rather than skipping it. diff --git a/docs/gating-coverage.md b/docs/gating-coverage.md index a787cb6..f05cc0f 100644 --- a/docs/gating-coverage.md +++ b/docs/gating-coverage.md @@ -37,7 +37,7 @@ enumerates it. Every row is one of: | `scheduler.wait()` / `scheduler.yield()` | scoped `Scheduler` over the same timer path | `api/global-scope.ts` | | `crypto.subtle.*` | every method's promise gated; sync members pass through | `api/global-scope.ts` | | `WebSocket` | frames each take a fresh input lock at the `accept()` loop; `send` carries its own output-gate promise (§1.8) | `api/web-socket.ts` | -| storage / `sql` / alarms / `blockConcurrencyWhile` / `awaitIo` / `makeReentryCallback` / entry dispatch | the runtime's own primitives | `io/io-context.ts`, `server/actor-container.ts` | +| storage / `sql` / alarms / `blockConcurrencyWhile` / `awaitIo` / `makeReentryCallback` / entry and loopback dispatch | the runtime's own primitives | `io/io-context.ts`, `server/actor-container.ts` | ## Transform @@ -46,18 +46,27 @@ Vite plugin for actor-bundled modules. It rewrites every `await value` to route through `@mcp-b/do-runtime/gate`, and wraps every `for await` source so `next()`, `return()`, and `throw()` settlements re-enter the owning actor. -The gate helper fails open outside actor code. Inside an actor it publishes each -continuation through a fresh input-gated slice, including awaits of plain values. +The gate helper fails open outside actor code. A development transform supplies +the module id and warns once if that path is reached; production keeps the helper +silent. Inside an actor it publishes each continuation through a fresh +input-gated slice, including awaits of plain values. The slice preserves a surrounding `blockConcurrencyWhile` critical section so -the section can await its own continuation without deadlocking. The actor identity -exists only for that continuation's microtask and is cleared before another -publication. Publications are serialized across actors so two promises settling -in the same checkpoint cannot overwrite each other's identity. - -The transform covers every syntactic await in modules selected by the consumer's -include policy, including top-level await and async generators. It does not cover -bare `.then()` chains on foreign promises or code outside the filter. The runtime -itself is excluded: its internal promise machinery must keep using raw awaits. +the section can await its own continuation without deadlocking. The tokenized +actor identity is realm-shared so separately bundled actor and host copies agree. +It exists only while the captured context still holds its input lock, the +synchronous current slice always wins, and the marker clears at that context's +checkpoint boundary. Publications are serialized across actors so two promises +settling in the same checkpoint cannot overwrite each other's identity. + +At build end the plugin reads the final Rollup module graph, after later +transforms, and compares fully wrapped awaits with total awaits per included +module. It logs the aggregate and fails the build with each incomplete module's +transformed/total count. `await using` is counted as uncovered rather than +silently claimed. The transform covers ordinary syntactic awaits selected by the +consumer's include policy, including top-level await and async generators. It +does not cover bare `.then()` chains on foreign promises, null-byte virtual +modules, or code outside the filter. The runtime itself is excluded: its +internal promise machinery must keep using raw awaits. Every seam row above remains defense in depth for untransformed consumers and for promise continuations that do not pass through syntax the transform can rewrite. diff --git a/examples/extension/tsconfig.worker.json b/examples/extension/tsconfig.worker.json index eb0e3b8..d687da0 100644 --- a/examples/extension/tsconfig.worker.json +++ b/examples/extension/tsconfig.worker.json @@ -18,12 +18,7 @@ "noUnusedParameters": true, "verbatimModuleSyntax": true, "skipLibCheck": true, - "noEmit": true, - // `cloudflare:workers` is a platform specifier no resolver knows. `vite.config.ts` - // aliases it for the build; this is the same statement for the checker. - "paths": { - "cloudflare:workers": ["../../src/api/cloudflare-workers.ts"] - } + "noEmit": true }, "include": ["src/worker/**/*.ts", "src/protocol.ts"] } diff --git a/package.json b/package.json index 20e98a1..1592c3a 100644 --- a/package.json +++ b/package.json @@ -54,7 +54,7 @@ "import": "./dist/backends/node-sqlite.js" }, "./cloudflare-workers": { - "types": "./dist/src/api/cloudflare-workers.d.ts", + "types": "./dist/cloudflare-workers.d.ts", "import": "./dist/cloudflare-workers.js" }, "./gate": { diff --git a/scripts/check-package.mjs b/scripts/check-package.mjs index d00c98f..459c3d1 100644 --- a/scripts/check-package.mjs +++ b/scripts/check-package.mjs @@ -25,6 +25,7 @@ for (const required of [ "CHANGELOG.md", "dist/index.js", "dist/src/index.d.ts", + "dist/cloudflare-workers.d.ts", "dist/backends/node-sqlite.js", "dist/backends/sqlite-wasm.js", "dist/gate.js", @@ -62,6 +63,9 @@ const vite = modules.get("./vite"); if (typeof runtime.createActorContainer !== "function") { throw new Error("packed root entry does not export createActorContainer"); } +if (typeof runtime.BrokenActorError !== "function" || typeof runtime.CanceledError !== "function") { + throw new Error("packed root entry does not export its actor lifecycle errors"); +} if (typeof nodeBackend.createNodeSqlProvider !== "function") { throw new Error("packed Node backend does not export createNodeSqlProvider"); } diff --git a/scripts/fix-declaration-imports.mjs b/scripts/fix-declaration-imports.mjs index 9a501ec..ad93fe8 100644 --- a/scripts/fix-declaration-imports.mjs +++ b/scripts/fix-declaration-imports.mjs @@ -17,3 +17,8 @@ for (const name of await readdir(dist, { recursive: true })) { if (fixed !== source) await writeFile(file, fixed); } + +await writeFile( + new URL("cloudflare-workers.d.ts", dist), + 'export * from "./src/api/cloudflare-workers.js";\n', +); diff --git a/src/api/global-scope.ts b/src/api/global-scope.ts index 97b0860..081dcf6 100644 --- a/src/api/global-scope.ts +++ b/src/api/global-scope.ts @@ -26,7 +26,7 @@ * "wrap it in awaitIo" would be wrong three ways: * * - **Timers** capture the critical section at the ARMING call and re-enter - * through `ctx.run(callback, cs)` when they fire. Not `awaitIo`, deliberately + * through `ctx.run(callback, { input: cs })` when they fire. Not `awaitIo`, deliberately * — see `TimeoutManager` in `io/io-context.ts` for upstream's own reason. * - **`fetch`** is `awaitIo` (`http.c++` has ten of them and zero * `awaitIoWithInputLock`), preceded by an output-gate wait so nothing departs diff --git a/src/api/web-socket.ts b/src/api/web-socket.ts index f28bdec..901d393 100644 --- a/src/api/web-socket.ts +++ b/src/api/web-socket.ts @@ -136,7 +136,7 @@ export class AcceptedWebSocket extends EventTarget { // so both are called, exactly as `WebSocketFacade` does for the same reason. const handler = this[`on${type}`] as ((event: Event) => void) | null; handler?.(delivered); - }, this.#criticalSection), + }, { input: this.#criticalSection }), ); } diff --git a/src/gate.test.ts b/src/gate.test.ts index 7a8b0db..52f510a 100644 --- a/src/gate.test.ts +++ b/src/gate.test.ts @@ -1,7 +1,13 @@ -import { describe, expect, test } from "vitest"; +import { describe, expect, test, vi } from "vitest"; import { __gate, __gateAsyncIterable, __gateAwait, __resumeAwait } from "./gate"; import { InputGate, OutputGate } from "./io/io-gate"; -import { IoContext, requireInputLock, type Actor, type Timer } from "./io/io-context"; +import { + BrokenActorError, + IoContext, + requireInputLock, + type Actor, + type Timer, +} from "./io/io-context"; const timer: Timer = { now: () => Date.now(), @@ -53,6 +59,25 @@ describe("__gate", () => { expect(__gate(thenable)).toBe(thenable); }); + test("warns once only when a development transform reaches a lockless continuation", async () => { + const warn = vi.spyOn(console, "warn").mockImplementation(() => {}); + const value = Promise.resolve("done"); + try { + expect(__gateAwait(value)).toBe(value); + const context = newContext(); + await context.run(async () => { + __resumeAwait(await __gateAwait(value, "/actor-with-a-lock.js")); + }); + await portHop(); + expect(__gateAwait(value, "/actor-with-a-gap.js")).toBe(value); + expect(__gateAwait(value, "/actor-with-a-gap.js")).toBe(value); + expect(warn).toHaveBeenCalledOnce(); + expect(warn).toHaveBeenCalledWith(expect.stringContaining("/actor-with-a-gap.js")); + } finally { + warn.mockRestore(); + } + }); + test("re-enters the same actor after sequential foreign awaits", async () => { const context = newContext(); @@ -194,6 +219,54 @@ describe("__gate", () => { }); describe("transformed await resume", () => { + test("preserves BrokenActorError when a failed section cannot re-enter", async () => { + const context = newContext(); + const cause = new Error("section failed"); + + const exception = await context.run(async () => { + try { + __resumeAwait( + await __gateAwait( + context.blockConcurrencyWhile(() => { + throw cause; + }), + ), + ); + return undefined; + } catch (error) { + return error; + } + }); + + expect(exception).toBeInstanceOf(BrokenActorError); + expect(exception).toHaveProperty("cause", cause); + }); + + test("ignores a continuation marker after its input lock is gone", () => { + const context = newContext(); + const key = Symbol.for("@mcp-b/do-runtime/current-continuation"); + const value = Promise.resolve("outside"); + Reflect.set(globalThis, key, { context, token: {} }); + + try { + expect(__gate(value)).toBe(value); + } finally { + Reflect.deleteProperty(globalThis, key); + } + }); + + test("publishes continuation identity for separately bundled runtime copies", async () => { + const context = newContext(); + const key = Symbol.for("@mcp-b/do-runtime/current-continuation"); + + await context.run(async () => { + __resumeAwait(await __gateAwait(portHop())); + expect(Reflect.get(globalThis, key)).toMatchObject({ context }); + }); + await portHop(); + expect(Reflect.has(globalThis, key)).toBe(false); + }); + test("restores context at the first instruction after fulfillment", async () => { const context = newContext(); diff --git a/src/gate.ts b/src/gate.ts index ca9fadc..5e5bc99 100644 --- a/src/gate.ts +++ b/src/gate.ts @@ -1,13 +1,10 @@ /* @do-runtime-gated */ -import { atCheckpointEnd, tryCurrentSlice, type IoContext } from "./io/io-context"; - -type ContinuationContext = { - readonly context: IoContext; - readonly token: object; -}; - -let continuationContext: ContinuationContext | undefined; +import { + tryCurrentContinuation, + tryCurrentIoContext, + type IoContext, +} from "./io/io-context"; type Publication = { readonly publish: () => Promise; @@ -19,6 +16,7 @@ type Outcome = | { readonly ok: false; readonly exception: unknown }; const TRANSFORMED_AWAIT = Symbol("@mcp-b/do-runtime/transformed-await"); +const warnedUngatedAwaits = new Set(); type TransformedAwait = { readonly [TRANSFORMED_AWAIT]: true; @@ -35,16 +33,28 @@ function isThenable(value: unknown): value is PromiseLike { /** Re-enter the actor that owns this transformed await; fail open outside actors. */ export function __gate(value: T): T | Promise> { - const context = tryCurrentSlice() ?? continuationContext?.context; + const context = tryCurrentIoContext(); if (!isThenable(value) && context === undefined) return value; if (context === undefined) return value; return resumeWithContext(context, Promise.resolve(value)); } /** Capture an actor await without publishing its context before the continuation runs. */ -export function __gateAwait(value: T): T | Promise>> { - const context = tryCurrentSlice() ?? continuationContext?.context; - if (context === undefined) return value; +export function __gateAwait( + value: T, + developmentSource?: string, +): T | Promise>> { + const context = tryCurrentIoContext(); + if (context === undefined) { + if (developmentSource !== undefined && !warnedUngatedAwaits.has(developmentSource)) { + warnedUngatedAwaits.add(developmentSource); + console.warn( + `do-runtime: transformed await in ${developmentSource} ran without an actor input lock; ` + + "an earlier await or entry path is not gated", + ); + } + return value; + } return resumeAwaitWithContext(context, Promise.resolve(value)); } @@ -52,7 +62,7 @@ export function __gateAwait(value: T): T | Promise(value: T | TransformedAwait): T { if (!isTransformedAwait(value)) return value as T; - restoreContinuation(value.context); + value.context.restoreContinuation(); if (value.outcome.ok) return value.outcome.value; throw value.outcome.exception; } @@ -61,14 +71,6 @@ function isTransformedAwait(value: T | TransformedAwait): value is Transfo return Reflect.get(Object(value), TRANSFORMED_AWAIT) === true; } -function restoreContinuation(context: IoContext): void { - const token = {}; - continuationContext = { context, token }; - atCheckpointEnd(() => { - if (continuationContext?.token === token) continuationContext = undefined; - }); -} - function publishOutcome( context: IoContext, promise: Promise, @@ -76,7 +78,7 @@ function publishOutcome( ): Promise { return new Promise((resolve, reject) => { const publish = context.makeTransformReentryCallback((outcome: Outcome) => { - if (continuationContext !== undefined) { + if (tryCurrentContinuation() !== undefined) { schedulePublication({ publish: () => publish(outcome), reject }); return; } @@ -106,7 +108,7 @@ function resumeAwaitWithContext( function resumeWithContext(context: IoContext, promise: Promise): Promise { return publishOutcome(context, promise, (outcome) => { - restoreContinuation(context); + context.restoreContinuation(); if (outcome.ok) return outcome.value; throw outcome.exception; }); diff --git a/src/index.ts b/src/index.ts index a19e51a..6b820c7 100644 --- a/src/index.ts +++ b/src/index.ts @@ -33,7 +33,8 @@ export type { } from "./util/sqlite"; export type { ReadOptions, WriteOptions } from "./io/actor-cache"; export type { AlarmOutlet } from "./io/actor-sqlite"; -export type { Timer } from "./io/io-context"; +export { BrokenActorError, type Timer } from "./io/io-context"; +export { CanceledError } from "./io/io-gate"; /** * The Worker Loader (§1.11, decision 15). Exported where the `api/` classes are * not, and for the reason `AlarmScheduler` is: this one is a **binding**, so a diff --git a/src/io/io-context.test.ts b/src/io/io-context.test.ts index 5de67bb..3c6ff83 100644 --- a/src/io/io-context.test.ts +++ b/src/io/io-context.test.ts @@ -22,6 +22,7 @@ import { } from "./io-gate"; import { BLOCK_CONCURRENCY_WHILE_TIMEOUT_MESSAGE, + BrokenActorError, type Actor, IoContext, type Timer, @@ -528,6 +529,7 @@ it("§1.5 blockConcurrencyWhile times out at 30 seconds against the Timer port", const never = new Promise(() => {}); const blocking = blockConcurrencyWhile(ctx, () => never); + const failure = blocking.catch((exception: unknown) => exception); await quiesce(); expect(timer.pendingCount).toBe(1); @@ -536,8 +538,12 @@ it("§1.5 blockConcurrencyWhile times out at 30 seconds against the Timer port", timer.advance(1); await expect(onBroken).rejects.toThrow(BLOCK_CONCURRENCY_WHILE_TIMEOUT_MESSAGE); - // The deadline is a failure like any other, so the returned promise stays unsettled. - expect(await poll(blocking)).toBe(false); + const exception = await failure; + expect(exception).toBeInstanceOf(BrokenActorError); + expect(exception).toHaveProperty( + "cause", + new Error(BLOCK_CONCURRENCY_WHILE_TIMEOUT_MESSAGE), + ); }); it("§1.5 the deadline timer is cancelled when the callback wins", async () => { @@ -547,22 +553,53 @@ it("§1.5 the deadline timer is cancelled when the callback wins", async () => { expect(timer.pendingCount).toBe(0); }); -it("§1.5 a failed critical section breaks the gate and never settles its promise", async () => { - // "we don't even bother calling resolver.reject() because it's meaningless at this - // point" — the actor is aborted instead, which is what the caller observes. +it("§1.5 a failed critical section breaks the gate and rejects its same-realm caller", async () => { const { ctx, actor } = newContext(); - const onBroken = actor.inputGate.onBroken(); - const onAbort = ctx.onAbort(); + const onBroken = actor.inputGate.onBroken().catch((exception: unknown) => exception); + const onAbort = ctx.onAbort().catch((exception: unknown) => exception); + const cause = new Error("boot failed"); const blocking = blockConcurrencyWhile(ctx, () => { - throw new Error("boot failed"); + throw cause; }); + const failure = blocking.catch((exception: unknown) => exception); - await expect(onBroken).rejects.toThrow("boot failed"); - await expect(onAbort).rejects.toThrow("boot failed"); - expect(await poll(blocking)).toBe(false); + const exception = await failure; + expect(exception).toBeInstanceOf(BrokenActorError); + expect(exception).toHaveProperty("cause", cause); + expect(await onBroken).toBe(exception); + expect(await onAbort).toBe(exception); // Every future wait rejects, forever. - await expect(ctx.run(() => "later")).rejects.toThrow("boot failed"); + expect(await ctx.run(() => "later").catch((error: unknown) => error)).toBe(exception); +}); + +it("§1.5 nested critical-section failures reuse one BrokenActorError", async () => { + const { ctx } = newContext(); + const cause = new Error("nested failure"); + + const exception = await blockConcurrencyWhile( + ctx, + async () => + await ctx.blockConcurrencyWhile(() => { + throw cause; + }), + ).catch((error: unknown) => error); + + expect(exception).toBeInstanceOf(BrokenActorError); + expect(exception).toHaveProperty("cause", cause); +}); + +it("§1.5 an ignored failed section does not create an unhandled rejection", async () => { + const { ctx } = newContext(); + const onAbort = ctx.onAbort().catch((exception: unknown) => exception); + + await ctx.run(() => { + void ctx.blockConcurrencyWhile(() => { + throw new Error("ignored failure"); + }); + }); + + await expect(onAbort).resolves.toBeInstanceOf(BrokenActorError); }); it("§1.5 a failure is annotated broken.inputGateBroken exactly once", async () => { @@ -573,8 +610,10 @@ it("§1.5 a failure is annotated broken.inputGateBroken exactly once", async () throw new Error("boom"); }); - expect(await poll(blocking)).toBe(false); - await expect(onBroken).rejects.toThrow("broken.inputGateBroken; boom"); + await expect(blocking).rejects.toBeInstanceOf(BrokenActorError); + await expect(onBroken).rejects.toThrow( + "broken.inputGateBroken; The Durable Object was reset after its input gate broke: boom", + ); }); it("§1.5 an abandoned critical section hands its parent lock back", async () => { @@ -588,7 +627,7 @@ it("§1.5 an abandoned critical section hands its parent lock back", async () => throw new Error("boom"); }); - expect(await poll(blocking)).toBe(false); + await expect(blocking).rejects.toBeInstanceOf(BrokenActorError); await quiesce(); expect(counts.locked).toBe(counts.released); @@ -841,8 +880,7 @@ it("§1.2 a timer armed inside a critical section runs inside that section", asy const section = blockConcurrencyWhile(ctx, async (lock) => { // Read synchronously: a `Lock` is released at the end of the slice that holds it, and // `getCriticalSection()` on a released one throws. That throw would fail the critical - // section, and a failed one is never settled at all (§1.5) — so the mistake presents as a - // hung test rather than as an assertion, which is worth knowing before making it. + // section and reject with BrokenActorError (§1.5). const own = lock.getCriticalSection(); ctx.setTimeoutImpl( false, diff --git a/src/io/io-context.ts b/src/io/io-context.ts index 271cd49..c3ad9db 100644 --- a/src/io/io-context.ts +++ b/src/io/io-context.ts @@ -35,7 +35,7 @@ * Upstream is the same: `getCriticalSection()` (`io-context.c++:362`) does * not touch `currentInputLock`, and `:1214` is the only place that clears * it. The difference between the two forms is entirely on the far side — - * `awaitIo` re-enters through `run(func, criticalSection)` and queues for a + * `awaitIo` re-enters through `run(func, { input: criticalSection })` and queues for a * fresh lock, `awaitIoWithInputLock` re-enters holding the ref it took. * 4. Removal from the stack is by identity, not by popping, because entries do * overlap — three deep in the unit tests. One invocation can have several @@ -97,10 +97,10 @@ * makes impossible; hang detection and `registerPendingEvent`, which need the * isolate's own idea of pending work; and the thread-local * `IoContext::current()` static, whose lock-resolving half the invocation stack - * replaces — its *identity* half is `currentSlice` below, narrowed to the - * synchronous slice, with one consumer and no resolver. `EventOutcome` and - * `RequestObserver` are metrics types with no port, so `waitUntilStatus()` - * returns the first exception instead. + * replaces — its *identity* half is `currentSlice` below for synchronous code + * and the await transform's captured continuation for post-await code. Neither + * identity resolves a lock. `EventOutcome` and `RequestObserver` are metrics + * types with no port, so `waitUntilStatus()` returns the first exception instead. */ import { @@ -163,6 +163,16 @@ export const BLOCK_CONCURRENCY_WHILE_TIMEOUT_MESSAGE = "A call to blockConcurrencyWhile() in a Durable Object waited for too long. " + "The call was canceled and the Durable Object was reset."; +/** A critical-section failure reset the actor before its caller could continue. */ +export class BrokenActorError extends Error { + override readonly name = "BrokenActorError"; + + constructor(cause: unknown) { + const detail = cause instanceof Error ? `: ${cause.message}` : ""; + super(`The Durable Object was reset after its input gate broke${detail}`, { cause }); + } +} + /** ← `jsg::annotateBroken(msg, "broken.inputGateBroken")`. */ const INPUT_GATE_BROKEN_PREFIX = "broken.inputGateBroken; "; @@ -328,12 +338,10 @@ const pendingCheckpointEnds: (() => void)[] = []; /** * ← `static thread_local IoContext* threadLocalRequest` (`io-context.c++:25`). * - * **This is NOT an async context and must not become one.** It is set on entry - * to a slice's SYNCHRONOUS body and restored the instant that body returns — - * which for an `async` function is its first `await`. It propagates through - * exactly nothing. The package's "no async context is required" property - * (README, Part 4 mechanic 1) is undisturbed: nothing resolves a lock through - * this, and deleting it would change no gate behaviour. + * `CURRENT_SLICE` is NOT an async context. It is set on entry to a slice's + * SYNCHRONOUS body and restored the instant that body returns — which for an + * `async` function is its first `await`. It propagates through exactly nothing, + * and neither identity below resolves a lock. * * Upstream's scope is wider and cannot be matched. `runInContextScope` saves the * previous context, installs itself, and restores at the end of the isolate run @@ -346,21 +354,42 @@ const pendingCheckpointEnds: (() => void)[] = []; * a parent against its child), and the value would be wrong with nothing to say * so. * - * So the port keeps only the half that is exact, and the one consumer is a - * tripwire that refuses on mismatch and stays quiet on `undefined` — never a - * resolver. See `requireOwnSlice` in `api/global-scope.ts`. + * So the port keeps that exact synchronous half. The await transform separately + * restores the context it captured at the await's first continuation instruction; + * `tryCurrentIoContext()` combines the two exact windows for loopback routing. + * Both values are realm-shared so separately bundled host and actor copies agree. */ const CURRENT_SLICE = Symbol.for("@mcp-b/do-runtime/current-slice"); +const CURRENT_CONTINUATION = Symbol.for("@mcp-b/do-runtime/current-continuation"); + +type CurrentContinuation = { + readonly context: IoContext; +}; function currentSlice(): IoContext | undefined { return Reflect.get(globalThis, CURRENT_SLICE) as IoContext | undefined; } +function currentContinuation(): CurrentContinuation | undefined { + return Reflect.get(globalThis, CURRENT_CONTINUATION) as CurrentContinuation | undefined; +} + /** ← `IoContext::tryCurrent()` (`io-context.c++:1416-1422`), over the narrowed scope above. */ export function tryCurrentSlice(): IoContext | undefined { return currentSlice(); } +/** The actor continuation restored by the await transform, if one is running. */ +export function tryCurrentContinuation(): IoContext | undefined { + const context = currentContinuation()?.context; + return context?.hasCurrent() === true ? context : undefined; +} + +/** The exact actor calling now, across both a synchronous slice and a transformed continuation. */ +export function tryCurrentIoContext(): IoContext | undefined { + return currentSlice() ?? tryCurrentContinuation(); +} + /** * ← the `SuppressIoContextScope` constructor's `threadLocalRequest = this` half * (`io-context.c++:1208`), as a function rather than an assignment in `#runImpl`. @@ -497,7 +526,7 @@ type TimeoutState = { * ugly, but using awaitIo() doesn't work here because we need the ability to * cancel the timer, so we don't want to addTask() it, which awaitIo() does * implicitly." So the shape is `cs = ctx.getCriticalSection()` captured at the - * call, then `ctx.run(callback, cs)` when it fires. The captured section is what + * call, then `ctx.run(callback, { input: cs })` when it fires. The captured section is what * makes a timer armed inside `blockConcurrencyWhile` run INSIDE that section * rather than queueing on the root gate behind it. * @@ -629,7 +658,7 @@ class TimeoutManager { } finally { if (state.params.repeat && !state.isCanceled) this.#arm(ctx, id, state); } - }, criticalSection); + }, { input: criticalSection }); } } @@ -772,6 +801,17 @@ export class IoContext { return currentSlice() === this; } + /** Publish this transformed continuation through the same checkpoint queue as its lock. */ + restoreContinuation(): void { + const current: CurrentContinuation = { context: this }; + Reflect.set(globalThis, CURRENT_CONTINUATION, current); + atCheckpointEnd(() => { + if (currentContinuation() === current) { + Reflect.deleteProperty(globalThis, CURRENT_CONTINUATION); + } + }); + } + /** * Record that user code just engaged this context's gate — an `awaitIo`, an * `entry` dispatch, a re-entry callback firing. No upstream analogue, because @@ -988,11 +1028,15 @@ export class IoContext { * ← the two `IoContext::run()` overloads: given a CriticalSection it waits on that, given * an already-held Lock it runs under it, and given neither it takes a fresh lock from the * gate. The third case is what a new external event does, and it is the reason inheritance - * cannot be read from gate state — see `makeReentryCallback`. + * cannot be read from gate state — see `makeReentryCallback`. `signal` cancels only the wait + * for admission; once a lock is acquired the slice runs normally. */ async run( func: (lock: Lock) => T | PromiseLike, - ilOrCs?: Lock | CriticalSection, + options?: { + readonly input?: Lock | CriticalSection | undefined; + readonly signal?: AbortSignal | undefined; + }, ): Promise { // Before we try running anything, let's make sure our IoContext hasn't been aborted. If it // has been aborted, there's likely not an active request so later operations will fail @@ -1002,13 +1046,14 @@ export class IoContext { throw aborted.exception; } + const input = options?.input; let lock: Lock; - if (ilOrCs === undefined) { - lock = await this.#actor.getInputGate().wait(); - } else if (ilOrCs instanceof CriticalSection) { - lock = await ilOrCs.wait(); + if (input === undefined) { + lock = await this.#actor.getInputGate().wait(options?.signal); + } else if (input instanceof CriticalSection) { + lock = await input.wait(options?.signal); } else { - lock = ilOrCs; + lock = input; } return await this.#runImpl(func, lock); @@ -1032,7 +1077,7 @@ export class IoContext { * * It does not route through `io-gate.ts`'s `makeReentryCallback`, which is the same idea * expressed at the gate. Upstream's `IoContext::makeReentryCallback` is literally - * `ctx.run(func, cs)`, and going through the gate helper instead would take a lock this + * `ctx.run(func, { input: cs })`, and going through the gate helper instead would take a lock this * file then has to make current a second time. The gate copy stays: it is the shape a * consumer holding only a gate needs, and Section 1's tests cover it. */ @@ -1049,7 +1094,7 @@ export class IoContext { return async (...args: Args): Promise => { this.noteGateUse("a re-entry callback registered at the site below", registrationStack); - const call = this.run((lock) => func(lock, ...args), criticalSection); + const call = this.run((lock) => func(lock, ...args), { input: criticalSection }); // ← the `addTask()` + `registerPendingEvent()` pair, which keeps the context live while // a callback is outstanding. Upstream scopes that to the callback's lifetime via a @@ -1074,7 +1119,7 @@ export class IoContext { * Waits for some background I/O to complete, then executes `func` on the result. * * The input lock is NOT held across the wait: the resumption re-enters through - * `run(func, criticalSection)` and takes a fresh lock, so it queues behind whatever + * `run(func, { input: criticalSection })` and takes a fresh lock, so it queues behind whatever * arrived in the meantime. This is what makes a Durable Object awaiting another Durable * Object fully re-entrant (§1.3). * @@ -1110,7 +1155,7 @@ export class IoContext { }; return (...args: Args): Promise => { - const call = this.run(() => func(...args), criticalSection); + const call = this.run(() => func(...args), { input: criticalSection }); this.addTask( call.then( () => {}, @@ -1161,14 +1206,15 @@ export class IoContext { * * Three behaviours live here rather than in `io-gate.ts`, which has no timer, and rather * than in `api/actor-state.ts`, whose own `blockConcurrencyWhile` is a one-line forward: - * the 30-second deadline, the brokenness annotation, and the fact that on failure the - * returned promise is never settled at all. + * the 30-second deadline, the brokenness annotation, and the typed rejection returned to + * the same-realm caller after the actor is broken. */ blockConcurrencyWhile(callback: (lock: Lock) => T | PromiseLike): Promise { const lock = this.getInputLock(); this.noteGateUse("blockConcurrencyWhile", captureGateStack()); const criticalSection = lock.startCriticalSection(); - const { promise: result, resolve } = Promise.withResolvers(); + const { promise: result, resolve, reject } = Promise.withResolvers(); + void result.catch(() => {}); this.addTask( (async () => { @@ -1183,15 +1229,13 @@ export class IoContext { } catch (exception) { // Annotate as broken for periodic metrics. If we already set up a brokenness reason, // we shouldn't override it. - annotateInputGateBroken(exception); + const broken = + exception instanceof BrokenActorError ? exception : new BrokenActorError(exception); + annotateInputGateBroken(broken); + criticalSection.failed(broken); + reject(broken); - // Note that on failure, no further InputLocks will be obtainable and the actor will - // shut down, so don't worry about holding a lock until we get back to application - // code -- we won't! In fact, we don't even bother calling resolver.reject() because - // it's meaningless at this point. - criticalSection.failed(exception); - - throw exception; + throw broken; } finally { // ← `~CriticalSection`. A no-op after `succeeded()`; on the failure path it is what // hands the parent lock back, since `failed()` does not. @@ -1265,7 +1309,8 @@ export class IoContext { * ← `awaitIoImpl()`. * * The KJ-side rejection is merged into the value so a single continuation handles both, the - * continuation re-enters through `run(func, ilOrCs)`, and the whole thing rides `addTask()`. + * continuation re-enters through `run(func, { input: ilOrCs })`, and the whole thing rides + * `addTask()`. * When `ilOrCs` is a Lock this is `awaitIoWithInputLock` and the gate never opened; when it * is a CriticalSection or nothing this is `awaitIo` and the resumption queues for a fresh * lock like any other event. @@ -1291,7 +1336,7 @@ export class IoContext { } else { reject(outcome.exception); } - }, ilOrCs); + }, { input: ilOrCs }); } catch (exception) { // `run()` refuses to re-enter an aborted context, and both of its throws happen // before the lock reaches the invocation stack. Upstream would destroy the whole diff --git a/src/io/io-gate.test.ts b/src/io/io-gate.test.ts index d25690c..03b687a 100644 --- a/src/io/io-gate.test.ts +++ b/src/io/io-gate.test.ts @@ -15,7 +15,7 @@ * upstream only exercises from `io-context.h`, which is Section 2. */ -import { expect, it } from "vitest"; +import { expect, it, vi } from "vitest"; import { CanceledError, CriticalSection, @@ -785,7 +785,9 @@ it("InputGate hooks balance across fulfilled, cancelled and broken waiters", asy const first = await gate.wait(); const second = first.addRef(); - const queued = gate.wait(); + const controller = new AbortController(); + const removeAbort = vi.spyOn(controller.signal, "removeEventListener"); + const queued = gate.wait(controller.signal); expect(counts).toEqual({ locked: 1, released: 0, waiterAdded: 1, waiterRemoved: 0 }); first.release(); @@ -794,6 +796,8 @@ it("InputGate hooks balance across fulfilled, cancelled and broken waiters", asy second.release(); (await queued).release(); + expect(removeAbort).toHaveBeenCalledWith("abort", expect.any(Function)); + controller.abort(); expect(counts).toEqual({ locked: 2, released: 2, waiterAdded: 1, waiterRemoved: 1 }); } @@ -979,7 +983,7 @@ it("OutputGate hooks balance on every path a lock can leave by", async () => { } // Settle, then abort. One AbortController covering a batch of writes, aborted during cleanup - // after the writes landed: the handler still fires, and must do nothing at all. + // after the writes landed: its settled listener is removed and the gate stays healthy. { const counts = newHookCounts(); const gate = new OutputGate(recordingOutputGateHooks(counts)); diff --git a/src/io/io-gate.ts b/src/io/io-gate.ts index a55133d..6e7aa23 100644 --- a/src/io/io-gate.ts +++ b/src/io/io-gate.ts @@ -63,13 +63,14 @@ function makeCanceledError(): CanceledError { * here, and every one of them first rejects a pre-aborted wait before touching gate state, so * "cancelled" always means "left the gate exactly as it found it". */ -function onAbort(signal: AbortSignal | undefined, run: () => void): void { - if (signal === undefined) return; +function onAbort(signal: AbortSignal | undefined, run: () => void): () => void { + if (signal === undefined) return () => {}; if (signal.aborted) { run(); - return; + return () => {}; } signal.addEventListener("abort", run, { once: true }); + return () => signal.removeEventListener("abort", run); } /** @@ -235,12 +236,13 @@ export class InputGate { newWaiterPromise(isChildWaiter: boolean, signal?: AbortSignal): Promise { const { promise, resolve, reject } = Promise.withResolvers(); const waiter = new Waiter(this, isChildWaiter, resolve, reject); - onAbort(signal, () => { + const removeAbort = onAbort(signal, () => { // ← `~Waiter` on the cancellation path. A waiter that already settled is unlinked, and // cancelling it is the no-op that dropping a settled promise is. if (!waiter.linked) return; waiter.reject(makeCanceledError()); }); + void promise.then(removeAbort, removeAbort); return promise; } @@ -601,7 +603,7 @@ export class CriticalSection extends InputGate { /** * ← the gate half of `IoContext::makeReentryCallback()` (`io-context.h:1507`), which is - * `ctx.run(func, cs)` with the critical section captured here rather than looked up later. + * `ctx.run(func, { input: cs })` with the critical section captured here rather than looked up later. * * Upstream, on why the critical section travels with the callback at all: * @@ -716,7 +718,7 @@ export class OutputGate { // ← the `kj::defer(rejectIfCanceled)` arm that runs when the coroutine is destroyed. // Upstream leaves the dropped promise unobservable; here the caller still holds it, and a // promise that never settles is a hang nobody can see, so it takes the same exception. - onAbort(signal, () => { + const removeAbort = onAbort(signal, () => { // The guard comes first, as it does in `Waiter`. Upstream can call the hook before its // own check because `kj::defer` runs once ever, on whichever path exits the scope; an // abort listener can fire after the lock already settled, so the invariant to preserve @@ -731,6 +733,7 @@ export class OutputGate { void raced.then( (value) => { + removeAbort(); // kj would have destroyed this frame on cancellation; there is nothing left to settle. if (!fulfiller.isWaiting()) return; fulfiller.fulfill(); @@ -738,6 +741,7 @@ export class OutputGate { resolve(value); }, (exception: unknown) => { + removeAbort(); if (!fulfiller.isWaiting()) return; this.#setBroken(exception); fulfiller.reject(exception); diff --git a/src/server/actor-container.test.ts b/src/server/actor-container.test.ts index 79415f6..c418285 100644 --- a/src/server/actor-container.test.ts +++ b/src/server/actor-container.test.ts @@ -11,10 +11,12 @@ import { describe, expect, expectTypeOf, test, vi } from "vitest"; import { createNodeSqlProvider } from "../../backends/node-sqlite"; +import { __gateAwait, __resumeAwait } from "../gate"; import { DurableObjectClass } from "../api/actor"; import { FACET_TREE_MAX_DEPTH } from "../api/actor-state"; import type { AlarmInvocationInfo } from "../api/global-scope"; import type { Timer } from "../io/io-context"; +import { CanceledError } from "../io/io-gate"; import type { SqlDatabase, SqlDatabaseProvider } from "../util/sqlite"; import { FacetDeletionReceiptStore } from "./facet-deletion"; import type { ActorClassChannel } from "../io/io-channels"; @@ -403,6 +405,37 @@ describe("the composition", () => { expect(await entry.increment(1)).toBe(2); }); + test("entry and run signals cancel queued admission without breaking the actor", async () => { + const { container } = await counterContainer(); + await portHop(); + + const started = Promise.withResolvers(); + const release = Promise.withResolvers(); + const holding = container.run(() => + container.state.blockConcurrencyWhile(async () => { + started.resolve(); + await release.promise; + }), + ); + await started.promise; + + const controller = new AbortController(); + const runBody = vi.fn(() => "run"); + const entryBody = vi.fn(() => "entry"); + const queuedRun = container.run(runBody, controller.signal); + const queuedEntry = container.entry({ call: entryBody }, controller.signal).call(); + controller.abort(); + + await expect(queuedRun).rejects.toBeInstanceOf(CanceledError); + await expect(queuedEntry).rejects.toBeInstanceOf(CanceledError); + expect(runBody).not.toHaveBeenCalled(); + expect(entryBody).not.toHaveBeenCalled(); + + release.resolve(); + await holding; + await expect(container.run(() => "still alive")).resolves.toBe("still alive"); + }); + test("isCurrentSlice identifies this container's synchronous body only", async () => { const first = await counterContainer(); const second = await counterContainer(); @@ -447,6 +480,148 @@ describe("the composition", () => { expect(first.container.hasCurrent()).toBe(false); }); + test("resolveLoopback preserves direct call return types", async () => { + const { container } = await counterContainer(); + + const syncResult = container.resolveLoopback(() => 1, () => Promise.resolve(1)); + expectTypeOf(syncResult).toEqualTypeOf>(); + const asyncResult = container.resolveLoopback( + async () => 1, + () => Promise.resolve(1), + ); + expectTypeOf(asyncResult).toEqualTypeOf>(); + }); + + test("resolveLoopback invokes direct only for an exact self-call", async () => { + const actor = await counterContainer(); + await portHop(); + + const direct = vi.fn(() => "direct"); + const entered = actor.container.entry({ call: () => "entered" }); + const enter = vi.fn(() => entered.call()); + + await expect( + actor.container.run(() => actor.container.resolveLoopback(direct, enter)), + ).resolves.toBe("direct"); + await portHop(); + await expect(actor.container.resolveLoopback(direct, enter)).resolves.toBe("entered"); + + expect(direct).toHaveBeenCalledOnce(); + expect(enter).toHaveBeenCalledOnce(); + }); + + test("resolveLoopback resumes through a structural caller after an untransformed await", async () => { + const parent = await counterContainer(); + const facet = await counterContainer(); + await portHop(); + + const direct = vi.fn(() => "direct"); + const entered = parent.container.entry({ call: () => "entered" }); + const enter = vi.fn(() => entered.call()); + + await expect( + parent.container.run(async () => { + await Promise.resolve(); + return parent.container.resolveLoopback(direct, enter, parent.container); + }), + ).resolves.toBe("direct"); + + const order: string[] = []; + const outputWait = vi.spyOn(facet.container, "waitOutputLocks").mockImplementation(() => { + order.push("output"); + return Promise.resolve(); + }); + await expect( + facet.container.run(async () => { + await Promise.resolve(); + const value = await parent.container.resolveLoopback( + direct, + () => { + order.push("entry"); + return enter(); + }, + facet.container, + ); + await facet.instance.ctx.storage.get("still-gated"); + return value; + }), + ).resolves.toBe("entered"); + outputWait.mockRestore(); + expect(order).toEqual(["output", "entry"]); + + await portHop(); + expect(() => parent.container.resolveLoopback(direct, enter, parent.container)).toThrow( + "resolveLoopback() caller has no current input lock", + ); + expect(direct).toHaveBeenCalledOnce(); + expect(enter).toHaveBeenCalledOnce(); + }); + + test("resolveLoopback prefers the exact current slice over another held lock", async () => { + const parent = await counterContainer(); + const facet = await counterContainer(); + await portHop(); + + const direct = vi.fn(() => "direct"); + const entered = parent.container.entry({ call: () => "entered" }); + const enter = vi.fn(() => entered.call()); + + await expect( + parent.container.run(async () => { + __resumeAwait(await __gateAwait(portHop())); + return facet.container.run(async () => { + expect(parent.container.hasCurrent()).toBe(true); + expect(facet.container.isCurrentSlice()).toBe(true); + const value = await parent.container.resolveLoopback(direct, enter); + await facet.instance.ctx.storage.get("still-gated"); + return value; + }); + }), + ).resolves.toBe("entered"); + expect(direct).not.toHaveBeenCalled(); + expect(enter).toHaveBeenCalledOnce(); + }); + + test("resolveLoopback follows a transformed caller continuation", async () => { + const caller = await counterContainer(); + const target = await counterContainer(); + await portHop(); + + const direct = vi.fn(() => "direct"); + const entered = target.container.entry({ call: () => "entered" }); + + await expect( + caller.container.run(async () => { + __resumeAwait(await __gateAwait(portHop())); + expect(caller.container.isCurrentSlice()).toBe(false); + const value = await target.container.resolveLoopback(direct, () => entered.call()); + await caller.instance.ctx.storage.get("still-gated"); + return value; + }), + ).resolves.toBe("entered"); + expect(direct).not.toHaveBeenCalled(); + await portHop(); + }); + + test("resolveLoopback keeps a transformed same-actor continuation direct", async () => { + const actor = await counterContainer(); + await portHop(); + + const direct = vi.fn(() => "direct"); + const entered = vi.fn(() => Promise.resolve("entered")); + + await expect( + actor.container.run(async () => { + __resumeAwait(await __gateAwait(portHop())); + expect(actor.container.isCurrentSlice()).toBe(false); + return await actor.container.resolveLoopback(direct, entered); + }), + ).resolves.toBe("direct"); + expect(direct).toHaveBeenCalledOnce(); + expect(entered).not.toHaveBeenCalled(); + await portHop(); + }); + test("transformed actor code re-enters before storage after a foreign await", async () => { const { container, instance } = await counterContainer(); const entry = container.entry({ diff --git a/src/server/actor-container.ts b/src/server/actor-container.ts index 1282010..06790c3 100644 --- a/src/server/actor-container.ts +++ b/src/server/actor-container.ts @@ -57,7 +57,7 @@ import type { AlarmOutlet } from "../io/actor-sqlite"; import { ActorSqlite, DEFAULT_ALARM_OUTLET } from "../io/actor-sqlite"; import type { AlarmResult } from "./alarm-scheduler"; import type { Actor, Timer } from "../io/io-context"; -import { IoContext, captureGateStack } from "../io/io-context"; +import { IoContext, captureGateStack, tryCurrentIoContext } from "../io/io-context"; import { InputGate, OutputGate } from "../io/io-gate"; import type { FacetManager, FacetStartInfo } from "../io/worker"; import { asFacetStub } from "../io/worker"; @@ -401,15 +401,28 @@ export interface ActorContainer { * when its synchronous body returns, but the lock it took drains the whole * microtask checkpoint (§1.2), so actor code chained one promise past a gated * resumption is lock-holding without being slice-current. That window is - * where a host stub still has a caller to identify: an outbound call made - * there must resume through the caller's `awaitIo`, or the code after it - * comes back with no input lock and its next storage call throws. A host - * that resolves callers with `isCurrentSlice()` alone routes exactly those - * calls ungated, which is how the loss stays invisible until three layers - * later. + * where an outbound call must resume through the caller's `awaitIo`, or the + * code after it comes back with no input lock and its next storage call + * throws. Lock state is not caller identity: a parent and its running facet + * can both return true. `resolveLoopback()` owns that decision. */ hasCurrent(): boolean; + /** + * Resolve an in-realm actor call without making the host infer caller identity. + * A call from this exact actor context uses the raw instance; every other call + * uses its gated entry, routed through the caller's output gate and `awaitIo` + * when there is one. The runtime resolves a current slice or + * transformed continuation first; `caller` is the lock-holding structural + * fallback for untransformed post-await code. Both callbacks are lazy: only + * the selected invocation runs. + */ + resolveLoopback( + invokeDirect: () => Result, + invokeEntry: () => Promise>, + caller?: ActorContainer, + ): Result | Promise>; + /** * Construct the instance under workerd's boot semantics: the input gate is * held for the constructor's synchronous slice, and boot-time @@ -426,19 +439,21 @@ export interface ActorContainer { * lock across an await. That is §1.2's whole content and the suite pins it: a * second event posted while a storage await holds the gate must not interleave, * and a door that reused the held lock could not tell that event apart from a - * call the actor made to itself. Telling them apart needs to know WHO is - * calling, which is a host's question rather than a container's — see the - * extension host's `loopbackStub`, where an actor reaching its own - * `DurableObjectNamespace` binding skips this door entirely because the lock it - * would take is the one it is already holding. + * call the actor made to itself. `resolveLoopback()` makes that distinction: + * an actor reaching its own `DurableObjectNamespace` binding from its exact + * current context skips this door because the lock it would take is the one + * it holds. `signal` is bound to the returned proxy and cancels invocations + * that are still waiting for admission; it does not interrupt an admitted + * method or its output-gate drain. */ - entry(target: T): ActorEntry; + entry(target: T, signal?: AbortSignal): ActorEntry; /** * The door for events that are not method calls — one WebSocket frame, one - * host-originated callback. Upstream: `IoContext::run`. + * host-originated callback. Upstream: `IoContext::run`. `signal` cancels only + * while the event is queued for an input lock. */ - run(event: () => T | PromiseLike): Promise; + run(event: () => T | PromiseLike, signal?: AbortSignal): Promise; /** * ← `IoContext::awaitIo`. The form a HOST-PROVIDED async primitive must take, @@ -1337,6 +1352,24 @@ class ActorContainerImpl implements ActorContainer { return this.#ctx.hasCurrent(); } + resolveLoopback( + invokeDirect: () => Result, + invokeEntry: () => Promise>, + caller?: ActorContainer, + ): Result | Promise> { + const context = tryCurrentIoContext(); + if (context === this.#ctx) return invokeDirect(); + if (context !== undefined) { + return context.awaitIo(context.waitForOutputLocks().then(invokeEntry)); + } + if (caller === undefined) return invokeEntry(); + if (!caller.hasCurrent()) { + throw new Error("resolveLoopback() caller has no current input lock"); + } + if (caller === this) return invokeDirect(); + return caller.awaitIo(caller.waitOutputLocks().then(invokeEntry)); + } + /** * ← `ActorContainer::start` (`server.c++:2854-2957`) as far as the class * instance, plus decision 4's boot semantics. @@ -1363,7 +1396,7 @@ class ActorContainerImpl implements ActorContainer { } } - entry(target: T): ActorEntry { + entry(target: T, signal?: AbortSignal): ActorEntry { const bound = new Map(); return new Proxy(target, { @@ -1379,10 +1412,12 @@ class ActorContainerImpl implements ActorContainer { // The dispatch is the one moment that knows both the method name and // the caller's frames — the provenance `describeLostLock` reports. this.#ctx.noteGateUse(`entry ${String(property)}()`, captureGateStack()); - const result = await this.#ctx.run(() => - this.#withExternalEntry(() => - (value as (...rest: unknown[]) => unknown).apply(subject, args), - ), + const result = await this.#ctx.run( + () => + this.#withExternalEntry(() => + (value as (...rest: unknown[]) => unknown).apply(subject, args), + ), + { signal }, ); // ← the reply being piped through `waitForOutputLocks()`. This is §1.1's whole point: // a method that returns without awaiting its own write still must not answer before @@ -1431,8 +1466,11 @@ class ActorContainerImpl implements ActorContainer { }); } - run(event: () => T | PromiseLike): Promise { - return this.#ctx.run(() => this.#withExternalEntry(event)); + run(event: () => T | PromiseLike, signal?: AbortSignal): Promise { + return this.#ctx.run( + () => this.#withExternalEntry(event), + { signal }, + ); } #withExternalEntry(body: () => T): T { @@ -1460,7 +1498,7 @@ class ActorContainerImpl implements ActorContainer { * * "Alarms enter with no lock and no critical section, so an alarm queues behind * any held lock and takes a fresh top-level lock" (§1.8) — which is exactly - * `ctx.run(func)` with no third argument. The retry ladder and the watchdog are + * `ctx.run(func)` with no input option. The retry ladder and the watchdog are * `server/alarm-scheduler.ts`'s; what is here is one delivery, and the * serialization of one delivery against the next, which is the property * `_cf_executingScheduleRowId` upstream depends on. diff --git a/src/server/alarm-scheduler.ts b/src/server/alarm-scheduler.ts index 4177c0d..c1eec5c 100644 --- a/src/server/alarm-scheduler.ts +++ b/src/server/alarm-scheduler.ts @@ -890,8 +890,5 @@ function requireRange(actorId: string, column: string, value: number, max: numbe /** ← `ensureInitialized` (`alarm-scheduler.c++:50-60`). */ function ensureInitialized(db: SqliteDatabase): void { hasCurrentSqliteTable(db, "_cf_ALARM", STMT.createTable); - // TODO(sqlite): Do this automatically at a lower layer? - db.run("PRAGMA journal_mode=WAL"); - db.run(STMT.createTable); } diff --git a/src/vite.test.ts b/src/vite.test.ts index 4f6b8e0..6fb8444 100644 --- a/src/vite.test.ts +++ b/src/vite.test.ts @@ -1,16 +1,15 @@ -import { parseSync, type FilterPattern } from "vite"; +import { build, parseSync, type Plugin } from "vite"; import { describe, expect, test } from "vitest"; -import { doRuntimeAwaitTransform } from "./vite"; +import { + doRuntimeAwaitTransform, + type DoRuntimeAwaitTransformOptions, +} from "./vite"; const HEADER = '/* @do-runtime-gated */\nimport { __gateAsyncIterable, __gateAwait, __resumeAwait } from "@mcp-b/do-runtime/gate";\n'; -async function transform( - code: string, - id = "/actor.js", - options?: { include?: FilterPattern; exclude?: FilterPattern }, -): Promise { - const hook = doRuntimeAwaitTransform(options).transform; +async function transformWith(plugin: Plugin, code: string, id: string): Promise { + const hook = plugin.transform; if (hook === undefined) throw new Error("await transform has no transform hook"); const handler = typeof hook === "function" ? hook : hook.handler; const result = await Reflect.apply( @@ -26,6 +25,22 @@ async function transform( return typeof result === "string" ? result : result.code; } +async function transform( + code: string, + id = "/actor.js", + options?: DoRuntimeAwaitTransformOptions, + command?: "build" | "serve", +): Promise { + const plugin = doRuntimeAwaitTransform(options); + if (command !== undefined) { + const hook = plugin.configResolved; + if (hook === undefined) throw new Error("await transform has no configResolved hook"); + const handler = typeof hook === "function" ? hook : hook.handler; + await Reflect.apply(handler, {}, [{ command }]); + } + return await transformWith(plugin, code, id); +} + describe("doRuntimeAwaitTransform", () => { test("gates a plain await", async () => { const source = "async function run() { return await task; }\n"; @@ -35,6 +50,14 @@ describe("doRuntimeAwaitTransform", () => { ); }); + test("identifies lockless transformed awaits in development", async () => { + const source = "async function run() { return await task; }\n"; + + await expect(transform(source, "/actor.js", undefined, "serve")).resolves.toBe( + `${HEADER}async function run() { return __resumeAwait((await __gateAwait((task), "/actor.js"))); }\n`, + ); + }); + test("gates each operation of a for-await iterator", async () => { const source = "async function run() { for await (const value of values) consume(value); }\n"; @@ -98,4 +121,39 @@ describe("doRuntimeAwaitTransform", () => { transform(source, "/project/node_modules/@mcp-b/do-runtime/dist/index.js"), ).resolves.toBe(source); }); + + test("asserts coverage against code added by a later transform", async () => { + const actorId = "/actor.js"; + let actorSource = + "export async function run(values) { await first; for await (const value of values) consume(value); }\n"; + const virtualActor: Plugin = { + name: "virtual-actor", + resolveId: (id) => (id === "actor-entry" ? actorId : null), + load: (id) => (id === actorId ? actorSource : null), + }; + const lateAwait: Plugin = { + name: "late-await", + enforce: "post", + transform: (code, id) => (id === actorId ? `${code}\nawait second;\n` : null), + }; + const actorBuild = (plugins: Plugin[]) => + build({ + configFile: false, + logLevel: "silent", + plugins: [virtualActor, doRuntimeAwaitTransform(), ...plugins], + build: { + write: false, + target: "esnext", + rollupOptions: { + input: "actor-entry", + external: ["@mcp-b/do-runtime/gate"], + }, + }, + }); + + await expect(actorBuild([])).resolves.toBeDefined(); + await expect(actorBuild([lateAwait])).rejects.toThrow("/actor.js: 2/3"); + actorSource = "export async function run() { await using resource = open(); }\n"; + await expect(actorBuild([])).rejects.toThrow("/actor.js: 0/1"); + }); }); diff --git a/src/vite.ts b/src/vite.ts index 155789c..73f7739 100644 --- a/src/vite.ts +++ b/src/vite.ts @@ -2,6 +2,7 @@ import MagicString from "magic-string"; import { createFilter, Visitor, + type ESTree, type FilterPattern, type Plugin, } from "vite"; @@ -20,6 +21,41 @@ function patterns(pattern: FilterPattern | undefined): readonly (string | RegExp return typeof pattern === "string" || pattern instanceof RegExp ? [pattern] : pattern; } +type AwaitCoverage = { + readonly total: number; + readonly transformed: number; +}; + +function directCallName(node: ESTree.Node | null | undefined): string | undefined { + if (node?.type !== "CallExpression" || node.callee.type !== "Identifier") return undefined; + return node.callee.name; +} + +function countAwaitCoverage(program: ESTree.Program): AwaitCoverage { + let total = 0; + let transformed = 0; + new Visitor({ + AwaitExpression() { + total += 1; + }, + CallExpression(node) { + if (directCallName(node) !== "__resumeAwait") return; + const argument = node.arguments[0]; + if (argument?.type !== "AwaitExpression") return; + if (directCallName(argument.argument) === "__gateAwait") transformed += 1; + }, + ForOfStatement(node) { + if (!node.await) return; + total += 1; + if (directCallName(node.right) === "__gateAsyncIterable") transformed += 1; + }, + VariableDeclaration(node) { + if (node.kind === "await using") total += 1; + }, + }).visit(program); + return { total, transformed }; +} + /** Rewrite syntactic awaits in selected actor-bundled modules to re-enter their input gate. */ export function doRuntimeAwaitTransform(options?: DoRuntimeAwaitTransformOptions): Plugin { const filter = createFilter(options?.include, [ @@ -27,10 +63,14 @@ export function doRuntimeAwaitTransform(options?: DoRuntimeAwaitTransformOptions "**/@mcp-b/do-runtime/gate", ...patterns(options?.exclude), ]); + let development = false; return { name: "do-runtime-await-transform", enforce: "post", + configResolved(config) { + development = config.command === "serve"; + }, transform(code, id) { if (!code.includes("await") || code.includes(MARKER) || !filter(id)) return null; @@ -41,7 +81,10 @@ export function doRuntimeAwaitTransform(options?: DoRuntimeAwaitTransformOptions AwaitExpression(node) { source.prependLeft(node.start, "__resumeAwait(("); source.prependLeft(node.argument.start, "__gateAwait(("); - source.appendRight(node.argument.end, "))"); + source.appendRight( + node.argument.end, + development ? `), ${JSON.stringify(id)})` : "))", + ); source.appendRight(node.end, "))"); transformed = true; }, @@ -61,5 +104,37 @@ export function doRuntimeAwaitTransform(options?: DoRuntimeAwaitTransformOptions map: source.generateMap({ hires: "boundary", includeContent: true, source: id }), }; }, + buildEnd(error) { + if (error !== undefined || development) return; + const incomplete: string[] = []; + let total = 0; + let transformed = 0; + let modules = 0; + for (const id of this.getModuleIds()) { + if (!filter(id)) continue; + const code = this.getModuleInfo(id)?.code; + if (!code?.includes("await")) continue; + + const coverage = countAwaitCoverage(this.parse(code)); + if (coverage.total === 0) continue; + total += coverage.total; + transformed += coverage.transformed; + modules += 1; + if (coverage.transformed !== coverage.total) { + incomplete.push(`${id}: ${coverage.transformed}/${coverage.total}`); + } + } + + if (incomplete.length > 0) { + incomplete.sort(); + this.error(`do-runtime await transform missed included awaits:\n${incomplete.join("\n")}`); + } + + if (total > 0) { + this.info( + `do-runtime await transform: ${transformed}/${total} awaits gated in ${modules} await-bearing included modules`, + ); + } + }, }; }