From 82bd2b12380a70774ab49b69fea77e0e35d9f24a Mon Sep 17 00:00:00 2001 From: Alex Nahas Date: Sun, 23 Aug 2026 15:53:09 -0700 Subject: [PATCH] Serialize transformed await publication A transformed await previously published its result before __resumeAwait restored the owning actor. In a realm with independently bundled actors, another publication could enter that microtask gap and replace the shared continuation identity, so the first actor resumed under the wrong input gate.\n\nReserve that publication gap across runtime copies, release it at __resumeAwait, and retain a checkpoint-end escape hatch for abandoned transformed results. The browser regression loads two gate module copies to preserve the production failure shape. --- .changeset/calm-actors-resume.md | 5 ++ .../browser/await-publication.smoke.spec.ts | 79 +++++++++++++++++++ src/gate.ts | 45 +++++++++-- 3 files changed, 124 insertions(+), 5 deletions(-) create mode 100644 .changeset/calm-actors-resume.md create mode 100644 conformance/browser/await-publication.smoke.spec.ts diff --git a/.changeset/calm-actors-resume.md b/.changeset/calm-actors-resume.md new file mode 100644 index 0000000..2ed304a --- /dev/null +++ b/.changeset/calm-actors-resume.md @@ -0,0 +1,5 @@ +--- +"@mcp-b/do-runtime": patch +--- + +Serialize transformed await publication until the owning continuation resumes so overlapping actors cannot overwrite each other's ambient identity. diff --git a/conformance/browser/await-publication.smoke.spec.ts b/conformance/browser/await-publication.smoke.spec.ts new file mode 100644 index 0000000..40a63d7 --- /dev/null +++ b/conformance/browser/await-publication.smoke.spec.ts @@ -0,0 +1,79 @@ +import { expect, test } from "vitest"; +import { InputGate, OutputGate } from "../../src/io/io-gate"; +import { IoContext, type Actor, type Timer } from "../../src/io/io-context"; + +const timer: Timer = { + now: () => Date.now(), + afterDelay: (ms) => new Promise((resolve) => setTimeout(resolve, ms)), +}; + +class TestActor implements Actor { + readonly inputGate = new InputGate(); + readonly outputGate = new OutputGate(); + + getInputGate(): InputGate { + return this.inputGate; + } + + getOutputGate(): OutputGate { + return this.outputGate; + } + + shutdownActorCache(): void {} + assertCanSetAlarm(): void {} +} + +function importGateCopy(name: string): Promise { + return import(/* @vite-ignore */ `../../src/gate.ts?${name}`) as Promise< + typeof import("../../src/gate") + >; +} + +function portHop(): Promise { + return new Promise((resolve) => { + const channel = new MessageChannel(); + channel.port1.onmessage = () => { + channel.port1.close(); + channel.port2.close(); + resolve(); + }; + channel.port2.postMessage(undefined); + }); +} + +test("only one actor publishes into the await-to-resume gap", async () => { + const [firstGate, secondGate] = await Promise.all([ + importGateCopy("first-publication"), + importGateCopy("second-publication"), + ]); + const first = new IoContext(new TestActor(), timer); + const second = new IoContext(new TestActor(), timer); + const firstSource = Promise.withResolvers(); + const secondSource = Promise.withResolvers(); + let firstPublication!: Promise; + let secondPublication!: Promise; + + await first.run(() => { + firstPublication = Promise.resolve(firstGate.__gateAwait(firstSource.promise)); + }); + await second.run(() => { + secondPublication = Promise.resolve(secondGate.__gateAwait(secondSource.promise)); + }); + + let secondPublished = false; + void secondPublication.then(() => { + secondPublished = true; + }); + firstSource.resolve(); + secondSource.resolve(); + await Promise.resolve(); + const beforeCheckpointFallback = portHop(); + + const firstResult = await firstPublication; + await beforeCheckpointFallback; + expect(secondPublished).toBe(false); + + firstGate.__resumeAwait(firstResult); + const secondResult = await secondPublication; + secondGate.__resumeAwait(secondResult); +}); diff --git a/src/gate.ts b/src/gate.ts index 5e5bc99..737b5d9 100644 --- a/src/gate.ts +++ b/src/gate.ts @@ -1,6 +1,7 @@ /* @do-runtime-gated */ import { + atCheckpointEnd, tryCurrentContinuation, tryCurrentIoContext, type IoContext, @@ -16,12 +17,24 @@ type Outcome = | { readonly ok: false; readonly exception: unknown }; const TRANSFORMED_AWAIT = Symbol("@mcp-b/do-runtime/transformed-await"); +/** + * Own the gap between publishing an await result and its `__resumeAwait` call. + * This is deliberately separate from the current-continuation ambient: a + * reservation serializes publishers but must never make its actor look current. + * See ยง2.3 and decision 8. + */ +const CURRENT_PUBLICATION = Symbol.for("@mcp-b/do-runtime/current-await-publication"); const warnedUngatedAwaits = new Set(); +type PublicationReservation = { + readonly context: IoContext; +}; + type TransformedAwait = { readonly [TRANSFORMED_AWAIT]: true; readonly context: IoContext; readonly outcome: Outcome; + readonly reservation: PublicationReservation; }; function isThenable(value: unknown): value is PromiseLike { @@ -62,6 +75,7 @@ export function __gateAwait( export function __resumeAwait(value: T | TransformedAwait): T { if (!isTransformedAwait(value)) return value as T; + clearPublication(value.reservation); value.context.restoreContinuation(); if (value.outcome.ok) return value.outcome.value; throw value.outcome.exception; @@ -71,18 +85,37 @@ function isTransformedAwait(value: T | TransformedAwait): value is Transfo return Reflect.get(Object(value), TRANSFORMED_AWAIT) === true; } +function currentPublication(): PublicationReservation | undefined { + return Reflect.get(globalThis, CURRENT_PUBLICATION) as PublicationReservation | undefined; +} + +function reservePublication(context: IoContext): PublicationReservation | undefined { + if (tryCurrentContinuation() !== undefined || currentPublication() !== undefined) return undefined; + const reservation = { context }; + Reflect.set(globalThis, CURRENT_PUBLICATION, reservation); + atCheckpointEnd(() => clearPublication(reservation)); + return reservation; +} + +function clearPublication(reservation: PublicationReservation): void { + if (currentPublication() === reservation) { + Reflect.deleteProperty(globalThis, CURRENT_PUBLICATION); + } +} + function publishOutcome( context: IoContext, promise: Promise, - finish: (outcome: Outcome) => Result, + finish: (outcome: Outcome, reservation: PublicationReservation) => Result, ): Promise { return new Promise((resolve, reject) => { const publish = context.makeTransformReentryCallback((outcome: Outcome) => { - if (tryCurrentContinuation() !== undefined) { + const reservation = reservePublication(context); + if (reservation === undefined) { schedulePublication({ publish: () => publish(outcome), reject }); return; } - resolve(finish(outcome)); + resolve(finish(outcome, reservation)); }); void promise.then( (value) => { @@ -99,15 +132,17 @@ function resumeAwaitWithContext( context: IoContext, promise: Promise, ): Promise> { - return publishOutcome(context, promise, (outcome) => ({ + return publishOutcome(context, promise, (outcome, reservation) => ({ [TRANSFORMED_AWAIT]: true, context, outcome, + reservation, })); } function resumeWithContext(context: IoContext, promise: Promise): Promise { - return publishOutcome(context, promise, (outcome) => { + return publishOutcome(context, promise, (outcome, reservation) => { + clearPublication(reservation); context.restoreContinuation(); if (outcome.ok) return outcome.value; throw outcome.exception;