Fix Durable Object ordering when mixing RPC and fetch calls

When a caller fires interleaved stub.rpc() and stub.fetch() calls on
the same DO stub without awaiting, all fetch calls were processed before
all RPC calls, regardless of send order. This violated the expected
E-order guarantee for actors.

Root cause: fetch reaches the InputGate in ~1 async hop (via
WorkerEntrypoint::request() -> context.run() -> InputGate::wait()),
while RPC takes ~4+ hops (customEvent -> JsRpcSessionCustomEvent::run()
-> Cap'n Proto session setup -> capability fulfillment -> pipelined
call dispatch -> JsRpcTargetBase::call() -> kj::yield() -> ctx.run()
-> InputGate::wait()). Since all operations originate from the same
synchronous JS execution, fetch calls always reached the FIFO queue
before any RPC calls.

Fix: eagerly acquire the InputGate position in
JsRpcSessionCustomEvent::run() — at the same level where
WorkerEntrypoint::request() acquires it — then thread the lock through
to the first ctx.run() call via the existing
IoContext::run(func, Maybe<InputGate::Lock>) overload. The lock is
consumed by the first RPC method invocation and is NOT held for the
session lifetime. The kj::yield() in JsRpcTargetBase::call() and
ExternalPusher ordering are unaffected.

Fixes #6561

Made-with: Cursor

Author: threepointone

src/workerd/api/tests/js-rpc-socket-test.wd-test

@@ -29,6 +29,7 @@ const unitTests :Workerd.Config = (
           (name = "MyServiceProxy", service = "MyServiceProxy-loop"),
           (name = "MyActor", durableObjectNamespace = "MyActor"),
           (name = "ActorNoExtends", durableObjectNamespace = "ActorNoExtends"),
+          (name = "OrderingActor", durableObjectNamespace = "OrderingActor"),
           (name = "defaultExport", service = "default-loop"),
           (name = "twelve", json = "12"),
           (name = "GreeterFactory", service = "GreeterFactory-loop"),
@@ -37,6 +38,7 @@ const unitTests :Workerd.Config = (
         durableObjectNamespaces = [
           (className = "MyActor", uniqueKey = "foo"),
           (className = "ActorNoExtends", uniqueKey = "bar"),
+          (className = "OrderingActor", uniqueKey = "ordering"),
         ],
 
         durableObjectStorage = (inMemory = void),

src/workerd/api/tests/js-rpc-test.js

@@ -586,6 +586,25 @@ export class ActorNoExtends {
   }
 }
 
+// DO used to test that mixed RPC+fetch calls preserve send order.
+export class OrderingActor extends DurableObject {
+  #log = [];
+
+  record(input) {
+    this.#log.push(input);
+  }
+
+  async fetch(request) {
+    let url = new URL(request.url);
+    this.#log.push(url.pathname);
+    return new Response('ok');
+  }
+
+  getLog() {
+    return this.#log;
+  }
+}
+
 export default class DefaultService extends WorkerEntrypoint {
   async fetch(req) {
     // Test this.env here just to prove omitting the constructor entirely works.
@@ -2117,3 +2136,67 @@ export let eOrderTest = {
     assert.deepEqual(results, [1, 2, 3, 4, 5, 6]);
   },
 };
+
+// Verify that interleaved RPC and fetch calls on a DO stub are delivered in send order.
+export let mixedRpcFetchOrdering = {
+  async test(controller, env, ctx) {
+    let id = env.OrderingActor.idFromName('mixed');
+    let stub = env.OrderingActor.get(id);
+
+    let promises = [];
+    let expected = [];
+    for (let i = 0; i < 20; i++) {
+      if (i % 2 === 0) {
+        promises.push(stub.record(`rpc-${i}`));
+        expected.push(`rpc-${i}`);
+      } else {
+        promises.push(stub.fetch(`http://x/fetch-${i}`));
+        expected.push(`/fetch-${i}`);
+      }
+    }
+    await Promise.all(promises);
+
+    let log = await stub.getLog();
+    assert.deepEqual(log, expected);
+  },
+};
+
+// Verify that pure RPC calls on a DO stub preserve send order.
+export let pureRpcOrdering = {
+  async test(controller, env, ctx) {
+    let id = env.OrderingActor.idFromName('pure-rpc');
+    let stub = env.OrderingActor.get(id);
+
+    let promises = [];
+    for (let i = 0; i < 20; i++) {
+      promises.push(stub.record(`call-${i}`));
+    }
+    await Promise.all(promises);
+
+    let log = await stub.getLog();
+    assert.deepEqual(
+      log,
+      Array.from({ length: 20 }, (_, i) => `call-${i}`)
+    );
+  },
+};
+
+// Verify that pure fetch calls on a DO stub preserve send order.
+export let pureFetchOrdering = {
+  async test(controller, env, ctx) {
+    let id = env.OrderingActor.idFromName('pure-fetch');
+    let stub = env.OrderingActor.get(id);
+
+    let promises = [];
+    for (let i = 0; i < 20; i++) {
+      promises.push(stub.fetch(`http://x/${i}`));
+    }
+    await Promise.all(promises);
+
+    let log = await stub.getLog();
+    assert.deepEqual(
+      log,
+      Array.from({ length: 20 }, (_, i) => `/${i}`)
+    );
+  },
+};

src/workerd/api/tests/js-rpc-test.wd-test

@@ -25,6 +25,7 @@ const unitTests :Workerd.Config = (
           (name = "MyServiceProxy", service = (name = "js-rpc-test", entrypoint = "MyServiceProxy")),
           (name = "MyActor", durableObjectNamespace = "MyActor"),
           (name = "ActorNoExtends", durableObjectNamespace = "ActorNoExtends"),
+          (name = "OrderingActor", durableObjectNamespace = "OrderingActor"),
           (name = "defaultExport", service = "js-rpc-test"),
           (name = "twelve", json = "12"),
           (name = "GreeterFactory", service = (name = "js-rpc-test", entrypoint = "GreeterFactory")),
@@ -33,6 +34,7 @@ const unitTests :Workerd.Config = (
         durableObjectNamespaces = [
           (className = "MyActor", uniqueKey = "foo"),
           (className = "ActorNoExtends", uniqueKey = "bar"),
+          (className = "OrderingActor", uniqueKey = "ordering"),
         ],
 
         durableObjectStorage = (inMemory = void),

src/workerd/api/worker-rpc.c++

@@ -1000,9 +1000,14 @@ class JsRpcTargetBase: public rpc::JsRpcTarget::Server {
       : enterIsolateAndCall([this, &ctx](CallContext callContext) {
           // Note: No need to topUpActor() since this is the start of a top-level request, so the
           // actor will already have been topped up by IncomingRequest::delivered().
+          //
+          // If a pre-acquired InputGate lock is available (set by JsRpcSessionCustomEvent::run()
+          // to preserve ordering with other event types like fetch), consume it here so that
+          // ctx.run() skips InputGate::wait() and uses the already-reserved position.
+          auto inputLock = kj::mv(preAcquiredInputLock);
           return ctx.run([this, &ctx, callContext](Worker::Lock& lock) mutable {
             return callImpl(lock, ctx, callContext);
-          });
+          }, kj::mv(inputLock));
         }),
         externalPusher(ctx.getExternalPusher()) {}
 
@@ -1057,6 +1062,12 @@ class JsRpcTargetBase: public rpc::JsRpcTarget::Server {
     return externalPusher->pushAbortSignal(context);
   }
 
+  // Pre-acquired InputGate lock for the first RPC call in a session. When set by
+  // JsRpcSessionCustomEvent::run(), this ensures the RPC call's position in the InputGate
+  // FIFO matches its arrival order relative to other event types (fetch, connect, etc.).
+  // Consumed on first use by enterIsolateAndCall; subsequent calls acquire normally.
+  kj::Maybe<InputGate::Lock> preAcquiredInputLock;
+
   KJ_DISALLOW_COPY_AND_MOVE(JsRpcTargetBase);
 
  private:
@@ -2174,6 +2185,19 @@ kj::Promise<WorkerInterface::CustomEvent::Result> JsRpcSessionCustomEvent::run(
 
   incomingRequest->delivered();
 
+  // For actors, eagerly reserve our position in the InputGate FIFO. Without this, fetch calls
+  // reach the InputGate in ~1 async hop (via WorkerEntrypoint::request() -> context.run()) while
+  // RPC takes ~4+ hops (session setup, Cap'n Proto dispatch, kj::yield()). By acquiring the lock
+  // here — at the same level as request() — we ensure mixed RPC+fetch calls are ordered by
+  // arrival time, not by how many async hops each code path takes.
+  //
+  // The lock is passed to EntrypointJsRpcTarget and consumed by the first ctx.run() call,
+  // so it is NOT held for the session lifetime.
+  kj::Maybe<InputGate::Lock> inputLock;
+  KJ_IF_SOME(a, ioctx.getActor()) {
+    inputLock = co_await a.getInputGate().wait(ioctx.getCurrentTraceSpan());
+  }
+
   KJ_DEFER({
     // waitUntil() should allow extending execution on the server side even when the client
     // disconnects.
@@ -2182,6 +2206,7 @@ kj::Promise<WorkerInterface::CustomEvent::Result> JsRpcSessionCustomEvent::run(
 
   EntrypointJsRpcTarget target(ioctx, entrypointName, kj::mv(versionInfo), kj::mv(props),
       kj::mv(wrapperModule), mapAddRef(incomingRequest->getWorkerTracer()), isDynamicDispatch);
+  target.preAcquiredInputLock = kj::mv(inputLock);
   capnp::RevocableServer<rpc::JsRpcTarget> revcableTarget(target);
 
   try {