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;