Hold an actor's RPC exceptions behind its output gate

src/workerd/api/jsrpc-call-plan-test.c++

@@ -1213,6 +1213,38 @@ KJ_TEST("a disconnect rethrown by an actor JSRPC method stays DISCONNECTED and d
   session.wait(fixture.getWaitScope());
 }
 
+KJ_TEST("a KJ exception decoding an actor's RPC arguments propagates unchanged") {
+  TestFixture fixture(TestFixture::SetupParams{
+    .mainModuleSource = ACTOR_SOURCE,
+    .actorId = Worker::Actor::Id(kj::str("jsrpc-bad-args-test")),
+    .actorClassName = "default"_kj,
+  });
+  auto entrypoint = fixture.makeWorkerEntrypoint();
+  auto [cap, session] = startSession(*entrypoint);
+
+  // Arguments must decode to an array. A number fails a KJ_REQUIRE before the method runs, so
+  // the exception is not the method's and must not round-trip through JavaScript, which would
+  // reduce it to a tunneled internal error.
+  kj::Array<kj::byte> notAnArray;
+  fixture.runInIoContext([&](const TestFixture::Environment& env) {
+    jsg::Serializer serializer(env.js);
+    serializer.write(env.js, jsg::JsValue(env.js.num(1)));
+    notAnArray = kj::heapArray(serializer.release().data.asPtr());
+  });
+  auto request = cap.callRequest();
+  request.setMethodName("fail");
+  request.getOperation().initCallWithArgs().setV8Serialized(notAnArray);
+  auto e = KJ_ASSERT_NONNULL(
+      kj::runCatchingExceptions([&]() { request.send().wait(fixture.getWaitScope()); }));
+
+  KJ_EXPECT(e.getType() == kj::Exception::Type::FAILED, e);
+  KJ_EXPECT(e.getDescription().contains("expected JsArray when deserializing arguments"), e);
+  KJ_EXPECT(!jsg::isTunneledException(e.getDescription()), e);
+
+  cap = nullptr;
+  session.wait(fixture.getWaitScope());
+}
+
 KJ_TEST("actor JSRPC session abort marks in-flight calls delivered") {
   TestFixture fixture(TestFixture::SetupParams{
     .mainModuleSource = ACTOR_SOURCE,
@@ -1345,5 +1377,76 @@ KJ_TEST("terminal abort signals mark RPC serialization ineligible") {
   });
 }
 
+constexpr kj::StringPtr GATED_RECEIVER_SOURCE = R"JS(
+  import { WorkerEntrypoint } from "cloudflare:workers";
+
+  export default class extends WorkerEntrypoint {
+    succeed() { return "ok"; }
+    fail() { throw new Error("method failed"); }
+    async failLater() {
+      await null;
+      throw new Error("method failed");
+    }
+  }
+)JS"_kj;
+
+// An actor's reply to an RPC call must not leave while its output gate is locked, whether the
+// method returned or threw: a thrown error can carry unconfirmed state as well as a result can.
+void expectReplyWaitsForOutputGate(kj::StringPtr method, bool succeeds) {
+  auto io = kj::setupAsyncIo();
+  capnp::MallocMessageBuilder flagsMessage;
+  auto flags = flagsMessage.initRoot<CompatibilityFlags>();
+  flags.setFetcherRpc(true);
+  TestFixture receiver(TestFixture::SetupParams{
+    .waitScope = io.waitScope,
+    .mainModuleSource = GATED_RECEIVER_SOURCE,
+    .actorId = Worker::Actor::Id(kj::str("gated-receiver")),
+    .useRealTimers = false,
+  });
+  TestFixture sender(TestFixture::SetupParams{
+    .waitScope = io.waitScope,
+    .featureFlags = flags.asReader(),
+    .useRealTimers = false,
+  });
+
+  auto gate = kj::newPromiseAndFulfiller<void>();
+  auto blocker = receiver.getActor().getOutputGate().lockWhile(kj::mv(gate.promise), nullptr);
+
+  kj::Maybe<bool> settled;
+  sender.runInIoContext([&](const TestFixture::Environment& env) -> kj::Promise<void> {
+    auto fetcher = env.js.alloc<Fetcher>(env.context.addObject<Fetcher::OutgoingFactory>(
+                                             kj::heap<ReceiverOutgoingFactory>(receiver)),
+        Fetcher::RequiresHostAndProtocol::YES);
+    auto function = wrapMethod(env.js, *fetcher, method);
+    v8::LocalVector<v8::Value> args(env.js.v8Isolate);
+    auto result =
+        jsg::JsFunction(function.getHandle(env.js)).call(env.js, env.js.undefined(), args);
+    auto reply = env.context.awaitJs(
+        env.js, env.js.toPromise(result).then(env.js, [&settled](jsg::Lock&, jsg::Value) {
+      settled = true;
+    }, [&settled](jsg::Lock&, jsg::Value) { settled = false; }));
+
+    co_await kj::yieldUntilWouldSleep();
+    KJ_EXPECT(settled == kj::none, "the reply left while the output gate was locked", method);
+
+    gate.fulfiller->fulfill();
+    co_await reply;
+    KJ_EXPECT(settled == succeeds, method);
+  });
+  blocker.wait(io.waitScope);
+}
+
+KJ_TEST("an actor's RPC result waits for its output gate") {
+  expectReplyWaitsForOutputGate("succeed"_kj, true);
+}
+
+KJ_TEST("an actor's synchronous RPC exception waits for its output gate") {
+  expectReplyWaitsForOutputGate("fail"_kj, false);
+}
+
+KJ_TEST("an actor's asynchronous RPC exception waits for its output gate") {
+  expectReplyWaitsForOutputGate("failLater"_kj, false);
+}
+
 }  // namespace
 }  // namespace workerd::api

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

@@ -2321,7 +2321,18 @@ class JsRpcTargetBase: public rpc::JsRpcTarget::Server {
         // (makeReentryCallback() ensures this). This does NOT cancel the JavaScript (because JS
         // promises are not RAII-cancelable), but it will cancel this trailing .then(), which is
         // why it's safe to capture `&ctx` here.
-        return result.then([&ctx]() mutable { return ctx.waitForOutputLocks(); });
+        //
+        // An exception must wait for the output gate as well. The method may have read (or
+        // written) state whose write has not been confirmed yet, and the error it throws can carry
+        // that state to the caller as easily as a return value can. This matches
+        // WorkerEntrypoint::request(), which does not let a fetch handler's exception escape the
+        // isolate before the output gate opens. If the gate is broken, its exception replaces
+        // the method's, as it does there.
+        return result.then([&ctx]() mutable { return ctx.waitForOutputLocks(); },
+            [&ctx](kj::Exception&& e) mutable -> kj::Promise<void> {
+          return ctx.waitForOutputLocks().then(
+              [e = kj::mv(e)]() mutable { kj::throwFatalException(kj::mv(e)); });
+        });
       } else {
         return result;
       }
@@ -2359,14 +2370,39 @@ class JsRpcTargetBase: public rpc::JsRpcTarget::Server {
           // Record this call's span on any stubs/callbacks passed as arguments, so that if the
           // callee invokes them, the follow-up jsRpcCall nests under this call's span (mirrors the
           // return-value handling in callImpl). Only recorded when traced.
-          InvocationResult invocationResult;
-          KJ_IF_SOME(envCtx, targetInfo.envCtx) {
-            invocationResult = invokeFnInsertingEnvCtx(js, methodNameForTrace, fn, thisArg, args,
-                envCtx.env, envCtx.ctx, jsRpcCallSpan.getSpanParentsIfObserved());
-          } else {
-            invocationResult =
-                invokeFn(js, fn, thisArg, args, jsRpcCallSpan.getSpanParentsIfObserved());
-          }
+          //
+          // A method that throws synchronously answers with a rejected promise, so that its
+          // exception reaches the caller by the same path as an asynchronous one, behind the
+          // output gate (see handleResult()). Only JavaScript exceptions are caught: a KJ
+          // exception here comes from decoding the arguments, not from the method, and keeps
+          // propagating unchanged rather than being converted to JavaScript and back. This is why
+          // JSG_TRY/JSG_CATCH is not used: JSG_CATCH converts KJ exceptions too.
+          auto invocationResult = [&]() -> InvocationResult {
+            jsg::Value error = nullptr;
+            {
+              v8::TryCatch tryCatch(js.v8Isolate);
+              try {
+                KJ_IF_SOME(envCtx, targetInfo.envCtx) {
+                  return invokeFnInsertingEnvCtx(js, methodNameForTrace, fn, thisArg, args,
+                      envCtx.env, envCtx.ctx, jsRpcCallSpan.getSpanParentsIfObserved());
+                } else {
+                  return invokeFn(js, fn, thisArg, args, jsRpcCallSpan.getSpanParentsIfObserved());
+                }
+              } catch (jsg::JsExceptionThrown&) {
+                // As in jsg::Lock::tryCatch(): a termination cannot become a rejection.
+                if (!tryCatch.CanContinue() || !tryCatch.HasCaught() ||
+                    tryCatch.Exception().IsEmpty()) {
+                  tryCatch.ReThrow();
+                  throw;
+                }
+                error = jsg::Value(js.v8Isolate, tryCatch.Exception());
+              }
+            }
+            // Reject outside the TryCatch, so that it cannot swallow an exception from here.
+            auto resolver = jsg::check(v8::Promise::Resolver::New(js.v8Context()));
+            jsg::check(resolver->Reject(js.v8Context(), error.getHandle(js)));
+            return {.returnValue = resolver->GetPromise()};
+          }();
 
           // We have a function, so let's call it and serialize the result for RPC.
           // If the function returns a promise we will wait for the promise to finish so we can