diff --git a/src/workerd/api/tests/BUILD.bazel b/src/workerd/api/tests/BUILD.bazel index 45c0adf1426..621098a4a82 100644 --- a/src/workerd/api/tests/BUILD.bazel +++ b/src/workerd/api/tests/BUILD.bazel @@ -1040,6 +1040,12 @@ wd_test( data = ["cross-context-promise-test.js"], ) +wd_test( + src = "actor-cross-context-promise-test.wd-test", + args = ["--experimental"], + data = ["actor-cross-context-promise-test.js"], +) + wd_test( src = "error-in-error-event-test.wd-test", args = ["--experimental"], diff --git a/src/workerd/api/tests/actor-cross-context-promise-test.js b/src/workerd/api/tests/actor-cross-context-promise-test.js new file mode 100644 index 00000000000..f3bdcabe805 --- /dev/null +++ b/src/workerd/api/tests/actor-cross-context-promise-test.js @@ -0,0 +1,153 @@ +// Copyright (c) 2026 Cloudflare, Inc. +// Licensed under the Apache 2.0 license found in the LICENSE file or at: +// https://opensource.org/licenses/Apache-2.0 + +// A Durable Object keeps a single IoContext for its whole lifetime, but gets a separate +// IncomingRequest for each event delivered to it. These tests cover promises that an actor event +// creates and that some other IoContext settles later, which routes the settlement back through the +// actor's delete queue. +// +// Settling such a promise runs application JavaScript, which needs the timer, metrics, tracing and +// IoChannelFactory that only a current IncomingRequest provides. Every continuation below therefore +// reaches for those services, so a settlement processed without a request would fail rather than +// quietly appear to work. + +import { DurableObject } from 'cloudflare:workers'; +import { strictEqual } from 'node:assert'; + +// Polls until `read()` returns something. Used instead of a fixed sleep wherever a test is waiting +// for progress rather than for the absence of it. +async function waitFor(what, read) { + for (let i = 0; i < 2000; i++) { + const value = read(); + if (value !== undefined) return value; + await scheduler.wait(1); + } + throw new Error(`timed out waiting for ${what}`); +} + +// Exercises the request-scoped services. Date.now() reads the request's timer, scheduler.wait() +// schedules against it, and storage records through the actor's cache. +async function useRequestScopedApis(ctx, key, value) { + const startedAt = Date.now(); + await scheduler.wait(1); + await ctx.storage.put(key, value); + return { + value, + stored: await ctx.storage.get(key), + clockWorks: Date.now() >= startedAt, + }; +} + +export class TestActor extends DurableObject { + // Holds its event open by awaiting the promise. No further event is coming, so the settlement has + // to be processed under this request. + async awaitFromActiveRequest() { + const { promise, resolve } = Promise.withResolvers(); + globalThis.activeResolve = resolve; + const value = await promise; + return await useRequestScopedApis(this.ctx, 'active', value); + } + + // Responds immediately but retains the continuation with ctx.waitUntil(), so the request outlives + // the response and is still current when the settlement arrives. + async createRetainedPromise() { + const { promise, resolve } = Promise.withResolvers(); + globalThis.retainedResolve = resolve; + this.ctx.waitUntil( + promise.then(async (value) => { + globalThis.retainedResult = await useRequestScopedApis( + this.ctx, + 'retained', + value + ); + }) + ); + return 'created'; + } + + // Responds without retaining the continuation, so this event's IncomingRequest drains as soon as + // the RPC session ends and the actor is left with no current request. + async createDetachedPromise() { + const { promise, resolve } = Promise.withResolvers(); + globalThis.detachedResolve = resolve; + this.detached = promise.then(async (value) => { + // Recorded before the first await, so a test can distinguish "the reaction started" from "the + // reaction finished". + globalThis.detachedReactionRan = true; + return await useRequestScopedApis(this.ctx, 'detached', value); + }); + return 'created'; + } + + async getDetachedResult() { + return await this.detached; + } +} + +export const settlementRunsUnderAnActiveRequest = { + async test(_, env) { + const stub = env.ns.get(env.ns.idFromName('active')); + const pending = stub.awaitFromActiveRequest(); + + // Publishing the resolver is the actor's last act before parking on the await. + const resolve = await waitFor( + 'the actor to publish its resolver', + () => globalThis.activeResolve + ); + globalThis.activeResolve = undefined; + resolve('active-value'); + + const result = await pending; + strictEqual(result.value, 'active-value'); + strictEqual(result.stored, 'active-value'); + strictEqual(result.clockWorks, true); + }, +}; + +export const settlementRunsUnderADrainingRequest = { + async test(_, env) { + const stub = env.ns.get(env.ns.idFromName('retained')); + strictEqual(await stub.createRetainedPromise(), 'created'); + + const resolve = globalThis.retainedResolve; + globalThis.retainedResolve = undefined; + resolve('retained-value'); + + // The actor receives no further event. Its waitUntil task is the only thing keeping the request + // alive, so the settlement has to be processed while that request drains. + const result = await waitFor( + 'the retained continuation to run', + () => globalThis.retainedResult + ); + strictEqual(result.value, 'retained-value'); + strictEqual(result.stored, 'retained-value'); + strictEqual(result.clockWorks, true); + }, +}; + +export const settlementWaitsForTheNextEventWhenTheActorIsIdle = { + async test(_, env) { + const stub = env.ns.get(env.ns.idFromName('detached')); + strictEqual(await stub.createDetachedPromise(), 'created'); + + // Nothing retains that event's request, so it drains once the RPC session ends. Wait long + // enough that it is certainly gone before settling the promise. + await scheduler.wait(100); + + globalThis.detachedResolve('detached-value'); + globalThis.detachedResolve = undefined; + + // With no request to run under, the settlement stays queued rather than executing JavaScript in + // a context that has no timer, metrics or I/O channels. + await scheduler.wait(100); + strictEqual(globalThis.detachedReactionRan, undefined); + + // The actor is still healthy, and its next event picks the settlement up. + const result = await stub.getDetachedResult(); + strictEqual(globalThis.detachedReactionRan, true); + strictEqual(result.value, 'detached-value'); + strictEqual(result.stored, 'detached-value'); + strictEqual(result.clockWorks, true); + }, +}; diff --git a/src/workerd/api/tests/actor-cross-context-promise-test.wd-test b/src/workerd/api/tests/actor-cross-context-promise-test.wd-test new file mode 100644 index 00000000000..3b74d95ed57 --- /dev/null +++ b/src/workerd/api/tests/actor-cross-context-promise-test.wd-test @@ -0,0 +1,26 @@ +using Workerd = import "/workerd/workerd.capnp"; + +const unitTests :Workerd.Config = ( + services = [ + ( name = "actor-cross-context-promise-test", + worker = ( + modules = [ + ( name = "worker", + esModule = embed "actor-cross-context-promise-test.js" ), + ], + compatibilityFlags = [ + "experimental", + "nodejs_compat", + "handle_cross_request_promise_resolution", + ], + durableObjectNamespaces = [ + (className = "TestActor", uniqueKey = "actor-cross-context-promise"), + ], + durableObjectStorage = (inMemory = void), + bindings = [ + (name = "ns", durableObjectNamespace = "TestActor"), + ], + ) + ), + ], +); diff --git a/src/workerd/io/io-context.c++ b/src/workerd/io/io-context.c++ index 73f71c74d34..ba1f0037912 100644 --- a/src/workerd/io/io-context.c++ +++ b/src/workerd/io/io-context.c++ @@ -1331,6 +1331,46 @@ void IoContext::runInContextScope(Worker::LockType lockType, }); } +void IoContext::drainCrossContextActions(Worker::Lock& workerLock) { + // Take the actions out of the queue before running any of them: an action is free to schedule + // another action, which would deadlock against the queue's lock. + auto actions = deleteQueue.queue->takeActions(); + if (actions.size() == 0) return; + + jsg::Lock& js = workerLock; + for (auto& action: actions) { + action(js); + } + + // An action only enqueues the promise's reactions; runImpl() drains the microtask queue before + // it releases the isolate lock, so they still run before we return to the event loop. +} + +kj::Maybe> IoContext::scheduleCrossContextActionDrain() { + if (incomingRequests.empty()) { + // Nothing can own a run right now. For actors the actions stay queued until the next event + // arrives; for a context whose request is already gone they are dropped along with the context. + return kj::none; + } + + auto& request = incomingRequests.front(); + if (request.waitedForWaitUntil && waitUntilTasks.isEmpty()) { + // drain() has already computed waitUntilTasks.onEmpty() and found the set empty, so it is + // committed to destroying this request; registering a task now would not be observed. Leave the + // actions queued rather than starting a run that could lose its request mid-flight. + return kj::none; + } + + auto forked = run([](Worker::Lock&) {}).fork(); + + // Hold the request open until the run completes. drain() either has not looked at the task set + // yet, or is still waiting on it, so this addition is guaranteed to be observed. Errors reach the + // caller through the branch returned below; this branch only provides the keepalive. + addWaitUntil(forked.addBranch().catch_([](kj::Exception&&) {})); + + return forked.addBranch(); +} + void IoContext::runImpl(Runnable& runnable, Worker::LockType lockType, kj::Maybe inputLock, @@ -1447,6 +1487,12 @@ void IoContext::runImpl(Runnable& runnable, v8::TryCatch tryCatch(workerLock.getIsolate()); try { + if (!exceptional) { + // Settle anything another context resolved on our behalf before delivering this event. The + // exceptional path is only logging an already-failed event, so it is not a good place to + // start running application JavaScript. + drainCrossContextActions(workerLock); + } runnable.run(workerLock); } catch (const jsg::JsExceptionThrown&) { if (tryCatch.HasTerminated()) { @@ -1711,22 +1757,25 @@ jsg::JsObject IoContext::getPromiseContextTag(jsg::Lock& js) { kj::Promise IoContext::startDeleteQueueSignalTask(IoContext* context) { // The promise that is returned is held by the IoContext itself, so when the // IoContext is destroyed, the promise will be canceled and the loop will - // end. On each iteration of the loop we want to reset the cross thread - // signal in the delete queue, then wait on the promise. Once the promise - // is fulfilled, we will run an empty task to prompt the IoContext to drain - // the DeleteQueue. + // end. On each iteration of the loop we wait for the delete queue's cross + // thread signal, then prompt the IoContext to drain the queue. try { + auto& queue = *context->deleteQueue.queue; + auto signal = queue.resetCrossThreadSignal(); for (;;) { - co_await context->deleteQueue.queue->resetCrossThreadSignal(); - co_await context->run([](auto& lock) { - auto& context = IoContext::current(); - auto l = context.deleteQueue.queue->crossThreadDeleteQueue.lockExclusive(); - auto& state = KJ_ASSERT_NONNULL(*l); - for (auto& action: state.actions) { - action(lock); - } - state.actions.clear(); - }); + co_await signal; + + // Re-arm before draining, so that an action scheduled while the drain below is in progress + // fulfills this fresh signal instead of the one we just consumed. + signal = queue.resetCrossThreadSignal(); + + // The actions run application JavaScript, which needs a current IncomingRequest to supply + // metrics, tracing, timers, and an IoChannelFactory. An actor's IoContext outlives its + // requests, so there may be no request to run under; in that case the actions stay queued and + // the actor's next event drains them from runImpl(). + KJ_IF_SOME(drained, context->scheduleCrossContextActionDrain()) { + co_await drained; + } } } catch (...) { context->abort(kj::getCaughtExceptionAsKj()); diff --git a/src/workerd/io/io-context.h b/src/workerd/io/io-context.h index b922f72f34e..e164761a89f 100644 --- a/src/workerd/io/io-context.h +++ b/src/workerd/io/io-context.h @@ -1223,6 +1223,17 @@ class IoContext final: public kj::Refcounted, private kj::TaskSet::ErrorHandler kj::Maybe inputLock, Runnable::Exceptional exceptional); + // Runs the actions another IoContext queued on our delete queue. Settling a promise this context + // owns can run arbitrary JavaScript, so this is called from runImpl(), where a current + // IncomingRequest supplies the metrics, tracing, timer, and IoChannelFactory that JavaScript may + // reach for. + void drainCrossContextActions(Worker::Lock& workerLock); + + // Starts a run() whose only purpose is to let runImpl() drain queued cross-context actions, + // returning kj::none if no IncomingRequest can currently own that run. The run is registered as + // a wait-until task so the request stays alive until it completes. + kj::Maybe> scheduleCrossContextActionDrain(); + // Detect whether a callback accepts IoContext& as a second argument. // Used by run() and blockConcurrencyWhile() to optionally pass *this. template diff --git a/src/workerd/io/io-own-test.c++ b/src/workerd/io/io-own-test.c++ index 79aa1f0d557..1f8ff443f05 100644 --- a/src/workerd/io/io-own-test.c++ +++ b/src/workerd/io/io-own-test.c++ @@ -92,5 +92,99 @@ KJ_TEST("ReverseIoOwn remains safe after its IoContext is destroyed") { } } +// Schedules `action` on `queue`, which requires a jsg::Lock but not an IoContext. +void scheduleAction( + TestFixture& fixture, const DeleteQueue& queue, kj::Function&& action) { + fixture.enterWorkerLockSynchronously( + [&](Worker::Lock& lock) { queue.scheduleAction(lock, kj::mv(action)); }); +} + +// Runs everything currently queued on `queue`. +uint runActions(TestFixture& fixture, const DeleteQueue& queue) { + auto actions = queue.takeActions(); + fixture.enterWorkerLockSynchronously([&](Worker::Lock& lock) { + for (auto& action: actions) { + action(lock); + } + }); + return actions.size(); +} + +KJ_TEST("DeleteQueue actions are taken out of the queue before they run") { + TestFixture fixture; + auto queue = kj::arc(); + auto signal = queue->resetCrossThreadSignal(); + + // Settling a promise can run application JavaScript, which is free to settle another promise + // belonging to this same queue. That would deadlock if the queue's lock were still held while + // running actions. + uint ran = 0; + scheduleAction(fixture, *queue, [&queue, &ran](jsg::Lock& js) { + ++ran; + queue->scheduleAction(js, [&ran](jsg::Lock&) { ++ran; }); + }); + + KJ_EXPECT(runActions(fixture, *queue) == 1); + KJ_EXPECT(ran == 1); + + KJ_EXPECT(runActions(fixture, *queue) == 1); + KJ_EXPECT(ran == 2); + + KJ_EXPECT(runActions(fixture, *queue) == 0); +} + +KJ_TEST("DeleteQueue signals its owner for actions scheduled while it is draining") { + TestFixture fixture; + auto queue = kj::arc(); + + auto signal = queue->resetCrossThreadSignal(); + KJ_EXPECT(!signal.poll(fixture.getWaitScope())); + + scheduleAction(fixture, *queue, [](jsg::Lock&) {}); + KJ_EXPECT(signal.poll(fixture.getWaitScope())); + + // Re-arming installs a new signal, so an action scheduled after the previous one was consumed + // fulfills the new signal rather than being left queued with none outstanding. + signal = queue->resetCrossThreadSignal(); + KJ_EXPECT(!signal.poll(fixture.getWaitScope())); + + scheduleAction(fixture, *queue, [](jsg::Lock&) {}); + KJ_EXPECT(signal.poll(fixture.getWaitScope())); + KJ_EXPECT(runActions(fixture, *queue) == 2); +} + +// Returns the handle another IoContext uses to push work onto `context`'s delete queue. This is the +// same object the promise cross-context resolve callback unwraps from a promise's context tag. +IoCrossContextExecutor& getCrossContextExecutor(jsg::Lock& js, IoContext& context) { + return *jsg::unwrapOpaqueRef>( + js.v8Isolate, context.getPromiseContextTag(js)); +} + +KJ_TEST("cross-context actions scheduled during a drain still wake the IoContext") { + TestFixture fixture({.actorId = Worker::Actor::Id(kj::str("drain-rearm-test"))}); + + auto context = fixture.newIoContext(); + // Draining runs application JavaScript, so it needs a request to run under. + auto request = fixture.newIncomingRequest(*context); + + uint ran = 0; + fixture.enterContext(*request, [&](TestFixture::Environment& env) { + // runImpl() already drained the queue on the way in, so this action stays queued until the + // signal task picks it up. + getCrossContextExecutor(env.js, *context).execute(env.js, [&context, &ran](jsg::Lock& js) { + ++ran; + // Schedule a second action from within the first, modelling another IoContext scheduling work + // while this drain is already in progress. Consuming the signal without installing a fresh + // one first would leave this action queued with nothing outstanding to wake the context. + getCrossContextExecutor(js, *context).execute(js, [&ran](jsg::Lock&) { ++ran; }); + }); + }); + + for (uint i = 0; i < 100 && ran < 2; ++i) { + fixture.pollEventLoop(); + } + KJ_EXPECT(ran == 2, ran); +} + } // namespace } // namespace workerd diff --git a/src/workerd/io/io-own.c++ b/src/workerd/io/io-own.c++ index bf1150e5d03..1109fb9564f 100644 --- a/src/workerd/io/io-own.c++ +++ b/src/workerd/io/io-own.c++ @@ -54,6 +54,14 @@ void DeleteQueue::scheduleAction(jsg::Lock& js, kj::Function&& } } +kj::Array> DeleteQueue::takeActions() const { + auto lock = crossThreadDeleteQueue.lockExclusive(); + KJ_IF_SOME(state, *lock) { + return state.actions.releaseAsArray(); + } + return nullptr; +} + void DeleteQueue::checkFarGet(const DeleteQueue& deleteQueue, const std::type_info& type) { IoContext::current().checkFarGet(deleteQueue, type); } diff --git a/src/workerd/io/io-own.h b/src/workerd/io/io-own.h index a7ad8c91121..9a88ddd8802 100644 --- a/src/workerd/io/io-own.h +++ b/src/workerd/io/io-own.h @@ -93,15 +93,28 @@ class DeleteQueue: public kj::AtomicRefcounted { void scheduleDeletion(OwnedObject* object) const; void scheduleAction(jsg::Lock& js, kj::Function&& action) const; + // Moves the queued actions out of the queue so the caller can run them. The caller must not hold + // the queue's lock while running them, because an action may itself schedule another action. + kj::Array> takeActions() const; + + // Installs a fresh cross-thread signal, returning a promise that resolves once an action is + // scheduled. Only one signal is outstanding at a time, so a caller that consumes one must re-arm + // to keep hearing about later actions. + kj::Promise resetCrossThreadSignal() const; + struct State { kj::Vector queue; // Actions that some other IoContext has requested be executed in this IoContext. When // adding an action to this list, crossThreadFulfiller should be fulfilled, signaling the - // target IoContext to wake up and run actions. After draining the actions queue, the target - // IoContext should replace crossThreadFulfiller with a new one which will wake it up again. + // target IoContext to wake up and run actions. // // In particular, these actions are used to implement cross-context promise resolution. // + // Running an action settles a promise owned by the target IoContext, which can run arbitrary + // JavaScript, so it requires a fully-formed request context. The target IoContext therefore + // only takes actions from this list while it has a current IncomingRequest; an actor sitting + // idle between events leaves them queued for its next event to run. + // // Keep in mind the IoContext could be destroyed before the cross-thread signal runs, in // which case the actions will never run. kj::Vector> actions; @@ -128,8 +141,6 @@ class DeleteQueue: public kj::AtomicRefcounted { template SpecificOwnedObject* addObjectImpl(kj::Own obj, OwnedObjectList& ownedObjects) const; - kj::Promise resetCrossThreadSignal() const; - friend class IoContext; };