Ensure cross-context promise settlements run under an IncomingRequest

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"],

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);
+  },
+};

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"),
+        ],
+      )
+    ),
+  ],
+);

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<kj::Promise<void>> 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<InputGate::Lock> 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<void> 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());

src/workerd/io/io-context.h

@@ -1223,6 +1223,17 @@ class IoContext final: public kj::Refcounted, private kj::TaskSet::ErrorHandler
       kj::Maybe<InputGate::Lock> 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<kj::Promise<void>> scheduleCrossContextActionDrain();
+
   // Detect whether a callback accepts IoContext& as a second argument.
   // Used by run() and blockConcurrencyWhile() to optionally pass *this.
   template <typename Func>