diff --git a/src/workerd/api/jsrpc-call-plan-test.c++ b/src/workerd/api/jsrpc-call-plan-test.c++ index b5ccb0b9c8c..141d0da964f 100644 --- a/src/workerd/api/jsrpc-call-plan-test.c++ +++ b/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 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(); + 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(); + auto blocker = receiver.getActor().getOutputGate().lockWhile(kj::mv(gate.promise), nullptr); + + kj::Maybe settled; + sender.runInIoContext([&](const TestFixture::Environment& env) -> kj::Promise { + auto fetcher = env.js.alloc(env.context.addObject( + kj::heap(receiver)), + Fetcher::RequiresHostAndProtocol::YES); + auto function = wrapMethod(env.js, *fetcher, method); + v8::LocalVector 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 diff --git a/src/workerd/api/worker-rpc.c++ b/src/workerd/api/worker-rpc.c++ index 1894c5b6fa4..1973a387ee7 100644 --- a/src/workerd/api/worker-rpc.c++ +++ b/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 { + 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