diff --git a/docs/reference/detail/abort-signal.md b/docs/reference/detail/abort-signal.md index c14a4029fad..dfb430579c7 100644 --- a/docs/reference/detail/abort-signal.md +++ b/docs/reference/detail/abort-signal.md @@ -58,11 +58,12 @@ isolate lock, always in the IoContext current at registration time: - Abort triggered in that context: `fn` runs synchronously during the abort. - Abort triggered elsewhere (another request, or no request): delivery is deferred into the - owning context via `IoCrossContextExecutor::tryExecute` and runs on its next turn. The - deferred task re-takes the registration slot on arrival, so a consumer that went away in - the meantime turns the delivery into a guaranteed no-op. -- Owning context already destroyed: the delivery is silently dropped — everything the action - wanted to touch died with the context. + owning context via `IoCrossContextExecutor::tryExecute` and runs on its next turn (with the + `DEFER_CROSS_CONTEXT_ACTIONS` autogate, an idle actor's next event). The deferred task + re-takes the registration slot on arrival, so a consumer that went away in the meantime + turns the delivery into a guaranteed no-op. +- Owning context destroyed, or its idle actor hibernating, before delivery: the delivery is + dropped — everything the action wanted to touch dies with the context. Dropping the returned handle (safe from any thread) guarantees `fn` never runs again, so `fn` may capture plain references whose lifetime the holder ties to the handle. This is also @@ -110,6 +111,11 @@ registrations (asserted by `crossRequestRegistrationChurn` in Cells hold no JS-heap references and need no GC visitation; abort algorithms do, and are visited by the signal. +Deferred deliveries of one abort share a single copy of its exception, and its RPC clones +share one serialized reason. With the `DEFER_CROSS_CONTEXT_ACTIONS` autogate, deliveries can +wait in an idle actor's queue until its next event, so each copy counts as external memory +until its last delivery is done (asserted in `src/workerd/api/abort-signal-test.c++`). + ## Signals received over RPC A deserialized signal's pending abort reason arrives through a mutex-guarded, atomically @@ -119,11 +125,13 @@ answer correctly even for signals that crossed request boundaries before the abo processed. Actually *reacting* to the abort (running `triggerAbort`) requires the subscription armed via `subscribeToRpcAbort()`, which happens automatically when an `abort` listener, `onabort` handler, or any native registration is added. Only the receiving -request's context can arm it — the underlying promise belongs to that request — so arming +request's context can arm it — the underlying promise belongs to that context — so arming requested from any other context is routed there through the signal's -`IoCrossContextExecutor` and takes effect on the receiving request's next turn. If the -receiving request is already gone, the routing is dropped: no delivery is possible anymore. -Polling stays accurate regardless: the box holds any abort the peer delivered before the -receiving request ended, and a signal whose trigger capability is dropped without an abort -never aborts. Registrations made after that point observe a delivered abort through the -already-aborted pre-checks instead. +`IoCrossContextExecutor` and takes effect on that context's next turn (with the +`DEFER_CROSS_CONTEXT_ACTIONS` autogate, an idle actor's next event). The signal queues at +most one such request. If the receiving IoContext is destroyed or its idle actor hibernates +first, the routing is dropped: no delivery is possible anymore. Polling stays accurate +regardless: the box holds any abort the peer delivered before the receiving request ended, +and a signal whose trigger capability is dropped without an abort never aborts. +Registrations made after that point observe a delivered abort through the already-aborted +pre-checks instead. diff --git a/src/workerd/api/BUILD.bazel b/src/workerd/api/BUILD.bazel index c6332c91baa..8da8f3d0f4f 100644 --- a/src/workerd/api/BUILD.bazel +++ b/src/workerd/api/BUILD.bazel @@ -845,6 +845,15 @@ kj_test( deps = ["//src/workerd/tests:test-fixture"], ) +kj_test( + src = "abort-signal-test.c++", + deps = [ + "//src/workerd/io", + "//src/workerd/io:worker-interface", + "//src/workerd/tests:test-fixture", + ], +) + kj_test( src = "actor-call-retry-test.c++", deps = [ diff --git a/src/workerd/api/abort-signal-test.c++ b/src/workerd/api/abort-signal-test.c++ new file mode 100644 index 00000000000..f27dcd6e3e5 --- /dev/null +++ b/src/workerd/api/abort-signal-test.c++ @@ -0,0 +1,391 @@ +// 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 + +#include +#include +#include +#include +#include + +#include +#include +#include + +namespace workerd::api { +namespace { + +constexpr int64_t REASON_SIZE = 1024 * 1024; +constexpr uint DELIVERIES = 8; + +jsg::JsValue makeReason(jsg::Lock& js) { + return js.str(kj::str(kj::repeat('x', REASON_SIZE))); +} + +// The isolate's external memory, including adjustments not yet applied. +int64_t getExternalMemory(jsg::Lock& js) { + js.getExternalMemoryTarget()->applyDeferredMemoryUpdate(); + return js.v8Isolate->GetExternalMemory(); +} + +// External memory held by an abort's queued deliveries, and still held once they have run. +struct QueuedAbortMemory { + int64_t queued; + int64_t retained; +}; + +// Creates a signal in one non-actor IoContext, has `registerDeliveries` register deliveries on it +// in another, and aborts it from the first, which queues the deliveries in the second. Then runs +// them. +QueuedAbortMemory measureQueuedAbort(TestFixture& fixture, + kj::FunctionParam registerDeliveries) { + auto ownerContext = fixture.newIoContext(); + auto owner = fixture.newIncomingRequest(*ownerContext); + auto aborterContext = fixture.newIoContext(); + auto aborter = fixture.newIncomingRequest(*aborterContext); + // Simulates outstanding I/O, so hang detection doesn't abort the contexts when the loop is idle. + auto ownerPendingEvent = ownerContext->registerPendingEvent(); + auto aborterPendingEvent = aborterContext->registerPendingEvent(); + + kj::Maybe> signal; + fixture.enterContext( + *aborter, [&](TestFixture::Environment& env) { signal = env.js.alloc(); }); + fixture.enterContext(*owner, + [&](TestFixture::Environment& env) { registerDeliveries(env, *KJ_ASSERT_NONNULL(signal)); }); + + QueuedAbortMemory memory{}; + int64_t before = 0; + fixture.enterContext(*aborter, [&](TestFixture::Environment& env) { + before = getExternalMemory(env.js); + KJ_ASSERT_NONNULL(signal)->triggerAbort( + env.js, kj::OneOf(makeReason(env.js))); + memory.queued = getExternalMemory(env.js) - before; + }); + + fixture.pollEventLoop(); + fixture.enterContext(*aborter, [&](TestFixture::Environment& env) { + memory.retained = getExternalMemory(env.js) - before; + signal = kj::none; + }); + + fixture.drainAndDestroy(kj::mv(aborter)); + fixture.drainAndDestroy(kj::mv(owner)); + return memory; +} + +KJ_TEST("queued native abort deliveries share one counted copy of the exception") { + TestFixture fixture({.autogateKeys = kj::arr(util::AutogateKey::DEFER_CROSS_CONTEXT_ACTIONS)}); + uint delivered = 0; + kj::Vector> registrations; + auto memory = + measureQueuedAbort(fixture, [&](TestFixture::Environment& env, AbortSignal& signal) { + for (uint i = 0; i < DELIVERIES; i++) { + registrations.add(signal.addAbortAction( + env.js, [&delivered](jsg::Lock&, const kj::Exception&) { ++delivered; })); + } + }); + + // The copy holds the reason once or twice (stringified, and possibly serialized); a copy per + // delivery would hold it at least DELIVERIES times. + KJ_EXPECT(memory.queued >= REASON_SIZE && memory.queued < 4 * REASON_SIZE, memory.queued); + KJ_EXPECT(delivered == DELIVERIES, delivered); + KJ_EXPECT(memory.retained < REASON_SIZE, memory.retained); +} + +KJ_TEST("queued native abort deliveries aren't counted without the autogate") { + TestFixture fixture({.ignoreAllAutogatesEnv = true}); + uint delivered = 0; + kj::Vector> registrations; + auto memory = + measureQueuedAbort(fixture, [&](TestFixture::Environment& env, AbortSignal& signal) { + for (uint i = 0; i < DELIVERIES; i++) { + registrations.add(signal.addAbortAction( + env.js, [&delivered](jsg::Lock&, const kj::Exception&) { ++delivered; })); + } + }); + + // A counted copy would hold the reason at least once. + KJ_EXPECT(memory.queued < REASON_SIZE, memory.queued); + KJ_EXPECT(delivered == DELIVERIES, delivered); +} + +KJ_TEST("a native abort delivery to an idle actor waits for its next request") { + TestFixture fixture({.autogateKeys = kj::arr(util::AutogateKey::DEFER_CROSS_CONTEXT_ACTIONS), + .actorId = Worker::Actor::Id(kj::str("idle-native-delivery-test"))}); + auto actorContext = fixture.newIoContext(); + auto statelessContext = fixture.newStatelessIoContext(); + auto aborter = fixture.newIncomingRequest(*statelessContext); + // Simulates outstanding I/O, so hang detection doesn't abort the aborter when the loop is idle. + auto aborterPendingEvent = statelessContext->registerPendingEvent(); + + kj::Maybe> signal; + fixture.enterContext( + *aborter, [&](TestFixture::Environment& env) { signal = env.js.alloc(); }); + + // Registered by an actor request that then ends, leaving the actor idle. + uint delivered = 0; + kj::Own registration; + auto registering = fixture.newIncomingRequest(*actorContext); + fixture.enterContext(*registering, [&](TestFixture::Environment& env) { + registration = KJ_ASSERT_NONNULL(signal)->addAbortAction( + env.js, [&delivered](jsg::Lock&, const kj::Exception&) { ++delivered; }); + }); + fixture.drainAndDestroy(kj::mv(registering)); + + fixture.enterContext(*aborter, [&](TestFixture::Environment& env) { + KJ_ASSERT_NONNULL(signal)->triggerAbort( + env.js, kj::OneOf(jsg::JsValue(env.js.str("done"_kj)))); + }); + fixture.pollEventLoop(); + KJ_EXPECT(delivered == 0, delivered); + KJ_EXPECT(actorContext->getAbortReason() == kj::none, actorContext->getAbortReason()); + + auto next = fixture.newIncomingRequest(*actorContext); + fixture.pollEventLoop(); + KJ_EXPECT(delivered == 1, delivered); + KJ_EXPECT(actorContext->getAbortReason() == kj::none, actorContext->getAbortReason()); + + fixture.drainAndDestroy(kj::mv(next)); + fixture.enterContext(*aborter, [&](TestFixture::Environment&) { + registration = nullptr; + signal = kj::none; + }); + fixture.drainAndDestroy(kj::mv(aborter)); +} + +// Counts the aborts sent to a signal's RPC clones. +class CountingAbortTrigger final: public rpc::AbortTrigger::Server { + public: + explicit CountingAbortTrigger(uint& aborts): aborts(aborts) {} + + kj::Promise abort(AbortContext) override { + ++aborts; + return kj::READY_NOW; + } + + private: + uint& aborts; +}; + +// Stands in for the receiving end of RPC, giving each serialized signal a CountingAbortTrigger. +class TestExternalPusher final: public rpc::JsValue::ExternalPusher::Server { + public: + explicit TestExternalPusher(uint& aborts): aborts(aborts) {} + + kj::Promise pushAbortSignal(PushAbortSignalContext context) override { + context.initResults().setTrigger(kj::heap(aborts)); + return kj::READY_NOW; + } + + private: + uint& aborts; +}; + +// Serializes `signal` for RPC to `pusher`'s end, which registers an RPC clone with the signal. +void cloneForRpc(jsg::Lock& js, AbortSignal& signal, rpc::JsValue::ExternalPusher::Client& pusher) { + RpcSerializerExternalHandler handler(RpcSerializerExternalHandler::TRANSFER, pusher, kj::none); + jsg::Serializer serializer(js, {.externalHandler = handler}); + signal.serialize(js, serializer); +} + +KJ_TEST("queued RPC abort deliveries share one counted copy of the serialized reason") { + capnp::MallocMessageBuilder flagsMessage; + auto flags = flagsMessage.initRoot(); + flags.setAbortSignalRpc(true); + uint aborts = 0; + TestFixture fixture({.featureFlags = flags.asReader(), + .autogateKeys = kj::arr(util::AutogateKey::DEFER_CROSS_CONTEXT_ACTIONS)}); + rpc::JsValue::ExternalPusher::Client pusher = kj::heap(aborts); + + int64_t capacity = 0; + auto memory = + measureQueuedAbort(fixture, [&](TestFixture::Environment& env, AbortSignal& signal) { + for (uint i = 0; i < DELIVERIES; i++) { + cloneForRpc(env.js, signal, pusher); + } + + // The abort serializes its reason the same way, into a buffer of this capacity. + jsg::Serializer serializer(env.js); + serializer.write(env.js, makeReason(env.js)); + capacity = serializer.release().dataCapacity; + }); + + // A copy per clone would hold the buffer DELIVERIES times. + KJ_EXPECT(memory.queued >= capacity && memory.queued < 2 * capacity, memory.queued, capacity); + KJ_EXPECT(aborts == DELIVERIES, aborts); + KJ_EXPECT(memory.retained < REASON_SIZE, memory.retained); +} + +KJ_TEST("queued RPC abort deliveries aren't counted without the autogate") { + capnp::MallocMessageBuilder flagsMessage; + auto flags = flagsMessage.initRoot(); + flags.setAbortSignalRpc(true); + uint aborts = 0; + TestFixture fixture({.featureFlags = flags.asReader(), .ignoreAllAutogatesEnv = true}); + rpc::JsValue::ExternalPusher::Client pusher = kj::heap(aborts); + + auto memory = + measureQueuedAbort(fixture, [&](TestFixture::Environment& env, AbortSignal& signal) { + for (uint i = 0; i < DELIVERIES; i++) { + cloneForRpc(env.js, signal, pusher); + } + }); + + // A counted copy would hold the reason at least once. + KJ_EXPECT(memory.queued < REASON_SIZE, memory.queued); + KJ_EXPECT(aborts == DELIVERIES, aborts); +} + +constexpr uint REGISTRATIONS = 100; + +// Sends `signal` from `sender`'s context to `receiver`'s over RPC, as passing it to a JS RPC call +// would, and returns the clone. The receiver's context owns the clone's RPC subscription. +jsg::Ref sendOverRpc(TestFixture& fixture, + IoContext::IncomingRequest& sender, + AbortSignal& signal, + IoContext::IncomingRequest& receiver) { + capnp::MallocMessageBuilder message; + capnp::BuilderCapabilityTable caps; + auto value = caps.imbue(message.initRoot()); + rpc::JsValue::ExternalPusher::Client pusher = receiver.getContext().getExternalPusher(); + fixture.enterContext(sender, [&](TestFixture::Environment& env) { + RpcSerializerExternalHandler handler( + RpcSerializerExternalHandler::TRANSFER, kj::mv(pusher), kj::none); + jsg::Serializer serializer(env.js, {.externalHandler = handler}); + signal.serialize(env.js, serializer); + value.setV8Serialized(serializer.release().data); + value.adoptExternals(handler.build(capnp::Orphanage::getForMessageContaining(value))); + }); + + kj::Maybe> clone; + fixture.enterContext(receiver, [&](TestFixture::Environment& env) { + auto reader = value.asReader(); + RpcStubDisposalGroup disposalGroup; + RpcDeserializerExternalHandler handler(reader.getExternals(), disposalGroup, kj::none); + jsg::Deserializer deserializer(env.js, reader.getV8Serialized(), kj::none, kj::none, + jsg::Deserializer::Options{.externalHandler = handler}); + clone = AbortSignal::deserialize(env.js, rpc::SerializationTag::ABORT_SIGNAL, deserializer); + }); + return KJ_ASSERT_NONNULL(kj::mv(clone)); +} + +// Registers and unregisters an abort action on `signal` REGISTRATIONS times. +void churnAbortActions(jsg::Lock& js, AbortSignal& signal) { + for (uint i = 0; i < REGISTRATIONS; i++) { + auto registration = signal.addAbortAction(js, [](jsg::Lock&, const kj::Exception&) {}); + } +} + +KJ_TEST("an idle actor queues one request to arm a received signal's RPC subscription") { + capnp::MallocMessageBuilder flagsMessage; + auto flags = flagsMessage.initRoot(); + flags.setAbortSignalRpc(true); + TestFixture fixture({.featureFlags = flags.asReader(), + .autogateKeys = kj::arr(util::AutogateKey::DEFER_CROSS_CONTEXT_ACTIONS), + .actorId = Worker::Actor::Id(kj::str("idle-receiver-test"))}); + auto actorContext = fixture.newIoContext(); + auto receiverQueue = actorContext->getCrossContextExecutor(); + auto statelessContext = fixture.newStatelessIoContext(); + auto sender = fixture.newIncomingRequest(*statelessContext); + // Simulates outstanding I/O, so hang detection doesn't abort the sender when the loop is idle. + auto senderPendingEvent = statelessContext->registerPendingEvent(); + + kj::Maybe> signal; + kj::Maybe> clone; + fixture.enterContext( + *sender, [&](TestFixture::Environment& env) { signal = env.js.alloc(); }); + auto receiving = fixture.newIncomingRequest(*actorContext); + clone = sendOverRpc(fixture, *sender, *KJ_ASSERT_NONNULL(signal), *receiving); + fixture.drainAndDestroy(kj::mv(receiving)); + + // The abort reaches the idle receiver, but nothing there observes the clone, so its + // subscription stays unarmed. + fixture.enterContext(*sender, [&](TestFixture::Environment& env) { + KJ_ASSERT_NONNULL(signal)->triggerAbort( + env.js, kj::OneOf(jsg::JsValue(env.js.str("done"_kj)))); + }); + fixture.pollEventLoop(); + + // The sender's registrations queue one request on the receiver to arm it, which stays queued. + uint delivered = 0; + kj::Own observer; + fixture.enterContext(*sender, [&](TestFixture::Environment& env) { + auto& received = *KJ_ASSERT_NONNULL(clone); + churnAbortActions(env.js, received); + observer = received.addAbortAction( + env.js, [&delivered](jsg::Lock&, const kj::Exception&) { ++delivered; }); + }); + fixture.pollEventLoop(); + KJ_EXPECT(receiverQueue.getQueuedActionCountForTest() == 1, + receiverQueue.getQueuedActionCountForTest()); + KJ_EXPECT(delivered == 0, delivered); + + // The receiver's next request arms the subscription, delivering the abort. + auto next = fixture.newIncomingRequest(*actorContext); + fixture.pollEventLoop(); + KJ_EXPECT(delivered == 1, delivered); + KJ_EXPECT(receiverQueue.getQueuedActionCountForTest() == 0, + receiverQueue.getQueuedActionCountForTest()); + KJ_EXPECT(actorContext->getAbortReason() == kj::none, actorContext->getAbortReason()); + + fixture.drainAndDestroy(kj::mv(next)); + fixture.enterContext(*sender, [&](TestFixture::Environment&) { + observer = nullptr; + clone = kj::none; + signal = kj::none; + }); + fixture.drainAndDestroy(kj::mv(sender)); +} + +KJ_TEST("registrations from other contexts queue one request per received signal to arm it") { + capnp::MallocMessageBuilder flagsMessage; + auto flags = flagsMessage.initRoot(); + flags.setAbortSignalRpc(true); + TestFixture fixture({.featureFlags = flags.asReader(), + .actorId = Worker::Actor::Id(kj::str("busy-receiver-test"))}); + auto actorContext = fixture.newIoContext(); + auto receiverQueue = actorContext->getCrossContextExecutor(); + auto statelessContext = fixture.newStatelessIoContext(); + auto sender = fixture.newIncomingRequest(*statelessContext); + // Simulates outstanding I/O, so hang detection doesn't abort the sender when the loop is idle. + auto senderPendingEvent = statelessContext->registerPendingEvent(); + auto receiving = fixture.newIncomingRequest(*actorContext); + + kj::Maybe> signal; + kj::Maybe> clone; + fixture.enterContext( + *sender, [&](TestFixture::Environment& env) { signal = env.js.alloc(); }); + clone = sendOverRpc(fixture, *sender, *KJ_ASSERT_NONNULL(signal), *receiving); + + // Aborted, so that the subscription completes once armed, letting the receiver's request + // finish draining. + fixture.enterContext(*sender, [&](TestFixture::Environment& env) { + KJ_ASSERT_NONNULL(signal)->triggerAbort( + env.js, kj::OneOf(jsg::JsValue(env.js.str("done"_kj)))); + }); + fixture.pollEventLoop(); + + // Holding the receiver's input lock keeps the requests to arm the subscription queued. + auto inputLock = fixture.getActor().getInputGate().wait(nullptr).wait(fixture.getWaitScope()); + fixture.enterContext(*sender, + [&](TestFixture::Environment& env) { churnAbortActions(env.js, *KJ_ASSERT_NONNULL(clone)); }); + fixture.pollEventLoop(); + KJ_EXPECT(receiverQueue.getQueuedActionCountForTest() == 1, + receiverQueue.getQueuedActionCountForTest()); + + { auto released = kj::mv(inputLock); } + fixture.pollEventLoop(); + KJ_EXPECT(receiverQueue.getQueuedActionCountForTest() == 0, + receiverQueue.getQueuedActionCountForTest()); + KJ_EXPECT(actorContext->getAbortReason() == kj::none, actorContext->getAbortReason()); + + fixture.drainAndDestroy(kj::mv(receiving)); + fixture.enterContext(*sender, [&](TestFixture::Environment&) { + clone = kj::none; + signal = kj::none; + }); + fixture.drainAndDestroy(kj::mv(sender)); +} + +} // namespace +} // namespace workerd::api diff --git a/src/workerd/api/basics.c++ b/src/workerd/api/basics.c++ index e2864af8b14..982023d6720 100644 --- a/src/workerd/api/basics.c++ +++ b/src/workerd/api/basics.c++ @@ -10,6 +10,7 @@ #include #include +#include #include #include @@ -542,7 +543,7 @@ class AbortTriggerRpcClient final { public: AbortTriggerRpcClient(rpc::AbortTrigger::Client&& client): client(kj::mv(client)) {} - kj::Promise abort(kj::ArrayPtr reason) { + kj::Promise abort(kj::ArrayPtr reason) { auto req = client.abortRequest(capnp::MessageSize{reason.size() / sizeof(capnp::word) + 8, 0}); auto field = req.initReason(); field.setV8Serialized(reason); @@ -555,17 +556,57 @@ class AbortTriggerRpcClient final { namespace { +// Data released by a jsg::Serializer. Its buffer's capacity can exceed data.size(). +struct SerializedBytes { + kj::Array data; + size_t capacity; +}; + +size_t getOwnedBytes(const SerializedBytes& bytes) { + return bytes.capacity; +} + +// An exception derived from a JavaScript value carries the value twice, stringified in its +// description and serialized in a detail. +size_t getOwnedBytes(const kj::Exception& exception) { + size_t size = exception.getDescription().size() + exception.getRemoteTrace().size(); + for (auto& detail: exception.getDetails()) { + size += detail.value.size(); + } + return size; +} + +// One copy of an abort's payload, shared by its deliveries. With the DEFER_CROSS_CONTEXT_ACTIONS +// autogate, deliveries to an idle actor wait in its queue for its next request (see +// DeleteQueue::scheduleAction()), so the copy counts as external memory. +template +class SharedAbortPayload final: public kj::AtomicRefcounted { + public: + SharedAbortPayload(jsg::Lock& js, T value): value(kj::mv(value)) { + if (util::Autogate::isEnabled(util::AutogateKey::DEFER_CROSS_CONTEXT_ACTIONS)) { + externalMemory = js.getExternalMemoryAdjustment(getOwnedBytes(this->value)); + } + } + + const T value; + + private: + kj::Maybe externalMemory; +}; + +using SerializedAbortReason = SharedAbortPayload; + kj::Promise abortRpcClientTask( - IoContext& ioContext, AbortTriggerRpcClient& client, kj::Array reason) { + IoContext& ioContext, AbortTriggerRpcClient& client, kj::Arc reason) { KJ_IF_SOME(outputLocks, ioContext.waitForOutputLocksIfNecessary()) { co_await outputLocks; } - co_await client.abort(reason); + co_await client.abort(reason->value.data); } // Sends the serialized abort reason to one RPC clone as a task on the current IoContext, // which must be the context that owns the client. -void sendAbortToRpc(IoOwn client, kj::Array reason) { +void sendAbortToRpc(IoOwn client, kj::Arc reason) { auto& ioContext = IoContext::current(); // Dereference the IoOwn here, while the owning context is known to be current; the task // keeps the client alive by holding the IoOwn as an attachment. @@ -1053,24 +1094,30 @@ void AbortSignal::runAbortSteps(jsg::Lock& js) { // 2. Native cancellations (canceler-wrapped promises and abort actions). Each registration // runs in the IoContext that created it: synchronously if that context is the current - // one, otherwise delivered on that context's next turn — or dropped, if that context is + // one, otherwise deferred as addAbortAction() describes — or dropped, if that context is // already gone (in which case everything it wanted to cancel died with it). Taking the // vector up front makes this safe against re-entrant registration. // // A deferred delivery does not take the action with it: it re-takes the cell's slot on // arrival in the owning context, so that a consumer (and the references its action - // captures) that went away in the meantime reliably turns the delivery into a no-op. + // captures) that went away in the meantime reliably turns the delivery into a no-op. The + // deferred deliveries share one copy of the exception. auto cells = kj::mv(nativeRegistrations); + kj::Maybe>> deferredException; for (auto& cell: cells) { if (cell->executor.isCurrent()) { KJ_IF_SOME(action, take(cell->action)) { action(js, exception); } } else { + if (deferredException == kj::none) { + deferredException = kj::arc>(js, exception.clone()); + } cell->executor.tryExecute( - [cell = cell.addRef(), ex = exception.clone()](jsg::Lock& js) mutable { + [cell = cell.addRef(), ex = KJ_ASSERT_NONNULL(deferredException).addRef()]( + jsg::Lock& js) mutable { KJ_IF_SOME(action, take(cell->action)) { - action(js, ex); + action(js, ex->value); } }); } @@ -1080,24 +1127,25 @@ void AbortSignal::runAbortSteps(jsg::Lock& js) { if (!rpcRegistrations.empty()) { auto regs = kj::mv(rpcRegistrations); - // Serialize the reason once; each clone gets its own copy of the bytes. + // Serialize the reason once, for all the clones to share. jsg::Serializer ser(js); KJ_IF_SOME(r, reason) { ser.write(js, r.getHandle(js)); } auto released = ser.release(); + auto serialized = kj::arc( + js, SerializedBytes{.data = kj::mv(released.data), .capacity = released.dataCapacity}); for (auto& reg: regs) { - auto bytes = kj::heapArray(released.data); if (reg->executor.isCurrent()) { KJ_IF_SOME(client, take(reg->client)) { - sendAbortToRpc(kj::mv(client), kj::mv(bytes)); + sendAbortToRpc(kj::mv(client), serialized.addRef()); } } else { reg->executor.tryExecute( - [reg = reg.addRef(), bytes = kj::mv(bytes)](jsg::Lock& js) mutable { + [reg = reg.addRef(), serialized = serialized.addRef()](jsg::Lock& js) mutable { KJ_IF_SOME(client, take(reg->client)) { - sendAbortToRpc(kj::mv(client), kj::mv(bytes)); + sendAbortToRpc(kj::mv(client), kj::mv(serialized)); } }); } @@ -1279,19 +1327,22 @@ void AbortSignal::subscribeToRpcAbort(jsg::Lock& js) { } if (!isRpcReceiverContextCurrent()) { - // Only the request that deserialized this signal can arm the subscription — it owns the + // Only the context that deserialized this signal can arm the subscription — it owns the // underlying promise — but a signal retained across requests may see registrations from // other contexts. Ask the receiving context to arm on its next turn. The queued action // captures only a WeakRef, which is safe to destroy from any thread should that context // be torn down without draining its queue. If the context is already gone, the request // is dropped: no delivery is possible anymore, though the abort itself stays observable // through the pending-reason box (see docs/reference/detail/abort-signal.md). - KJ_IF_SOME(executor, rpcReceiverContext) { - executor.tryExecute([weakSelf = JSG_THIS_WEAK(js)](jsg::Lock& js) { - KJ_IF_SOME(self, weakSelf.tryGet()) { - self.subscribeToRpcAbort(js); - } - }); + if (!rpcSubscriptionRequested) { + KJ_IF_SOME(executor, rpcReceiverContext) { + bool queued = executor.tryExecute([weakSelf = JSG_THIS_WEAK(js)](jsg::Lock& js) { + KJ_IF_SOME(self, weakSelf.tryAddRef(js)) { + self->subscribeToRpcAbort(js); + } + }); + rpcSubscriptionRequested = queued; + } } return; } diff --git a/src/workerd/api/basics.h b/src/workerd/api/basics.h index e337c39850f..efdb7119827 100644 --- a/src/workerd/api/basics.h +++ b/src/workerd/api/basics.h @@ -754,9 +754,9 @@ class AbortSignal final: public EventTarget { // Registers a native callback to be invoked with the abort exception if/when this signal // aborts. The callback runs under the isolate lock in the IoContext that is current at // registration time: synchronously when the abort is triggered within that context, - // otherwise delivered on that context's next turn, and dropped entirely (never invoked) - // once that context has been destroyed. If the signal can never abort, the callback is - // never invoked and no registration is made. + // otherwise as a cross-context action (see DeleteQueue::scheduleAction()), and dropped + // entirely (never invoked) once that context has been destroyed. If the signal can never + // abort, the callback is never invoked and no registration is made. // // Dropping the returned handle (safe from any thread) unregisters the callback: once the // handle is destroyed, the callback is guaranteed to never (again) be invoked, so it may @@ -831,7 +831,7 @@ class AbortSignal final: public EventTarget { // One native abort action, shared between this signal and one consumer. The action is // invoked at most once, only ever in its owning IoContext (synchronously if the abort is - // triggered there; otherwise on that context's next turn), always under the isolate lock, + // triggered there; otherwise as addAbortAction() describes), always under the isolate lock, // and never again after the consumer's RAII handle clears the slot. Because the handle is // held by (or attached to) objects the owning request destroys, an IoContext teardown // reclaims the action without ever touching the signal; the signal side retains only this @@ -945,6 +945,11 @@ class AbortSignal final: public EventTarget { kj::Maybe rpcReceiverContext; bool isRpcReceiverContextCurrent(); + // Whether a request to arm the RPC subscription has been queued on rpcReceiverContext. Queued + // only once, so registrations can't pile requests up on an idle actor; it arms the subscription + // unless the context is aborted or destroyed, or its idle actor hibernates, first. + bool rpcSubscriptionRequested = false; + // A promise that is fulfilled if an abort() message is received over RPC. kj::Maybe>> rpcAbortPromise; diff --git a/src/workerd/api/tests/BUILD.bazel b/src/workerd/api/tests/BUILD.bazel index 0c24145008d..f6453d7a9e7 100644 --- a/src/workerd/api/tests/BUILD.bazel +++ b/src/workerd/api/tests/BUILD.bazel @@ -1394,6 +1394,11 @@ wd_test( data = ["cross-context-promise-test.js"], ) +wd_test( + src = "actor-cross-context-promise-test.wd-test", + 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..bbf30f32e07 --- /dev/null +++ b/src/workerd/api/tests/actor-cross-context-promise-test.js @@ -0,0 +1,127 @@ +// 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 + +// Tests for promises that an actor event creates and another IoContext settles later. The +// settlement runs back in the actor's IoContext, which outlives each event's IncomingRequest, and +// needs one: the continuations use timers and storage. Each case settles while the actor has a +// request, so it behaves the same with or without the DEFER_CROSS_CONTEXT_ACTIONS autogate, which +// the @all-autogates variant enables. Settlements on an idle actor are tested in io-own-test.c++, +// where a test can make an actor idle at a known point. + +import { DurableObject } from 'cloudflare:workers'; +import { deepStrictEqual, strictEqual } from 'node:assert'; + +// Polls until `read()` returns a value, for waiting on progress without a fixed sleep. +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}`); +} + +// Uses the request's timer and the actor's storage, which both need a current request. +async function useRequestScopedApis(ctx, key, value) { + await scheduler.wait(1); + await ctx.storage.put(key, value); + return { value, stored: await ctx.storage.get(key) }; +} + +export class TestActor extends DurableObject { + // Awaits the promise, so its request is still current when the settlement arrives. + async awaitFromActiveRequest() { + const { promise, resolve } = Promise.withResolvers(); + globalThis.activeResolve = resolve; + const value = await promise; + return await useRequestScopedApis(this.ctx, 'active', value); + } + + // Responds at once, but ctx.waitUntil() keeps the request current for the settlement. + 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'; + } + + // Publishes a resolver in blockConcurrencyWhile(), then holds the input lock past the settlement. + async settleDuringCriticalSection() { + const order = []; + const { promise, resolve } = Promise.withResolvers(); + const settled = promise.then(() => order.push('settled')); + await this.ctx.blockConcurrencyWhile(async () => { + globalThis.criticalResolve = resolve; + while (!globalThis.criticalResolved) { + await scheduler.wait(1); + } + await scheduler.wait(10); + order.push('critical section done'); + }); + await settled; + return order; + } +} + +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'); + }, +}; + +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'); + + // With no further event, the settlement runs under the draining request. + const result = await waitFor( + 'the retained continuation to run', + () => globalThis.retainedResult + ); + strictEqual(result.value, 'retained-value'); + strictEqual(result.stored, 'retained-value'); + }, +}; + +export const settlementWaitsForACriticalSectionToEnd = { + async test(_, env) { + const pending = env.ns + .get(env.ns.idFromName('critical')) + .settleDuringCriticalSection(); + const resolve = await waitFor( + 'the critical section to start', + () => globalThis.criticalResolve + ); + globalThis.criticalResolve = undefined; + resolve(); + globalThis.criticalResolved = true; + deepStrictEqual(await pending, ['critical section done', 'settled']); + }, +}; 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..8aa2b773f9a --- /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 = [ + "nodejs_compat", + "rpc", + "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/AGENTS.md b/src/workerd/io/AGENTS.md index 733ecc49dd8..98092f51008 100644 --- a/src/workerd/io/AGENTS.md +++ b/src/workerd/io/AGENTS.md @@ -45,6 +45,7 @@ I/O lifecycle, per-request context, worker/isolate management, actor storage, co - `awaitIo(js, kjPromise, func)` bridges KJ→JS; `func` runs under V8 lock. `awaitIo(js, kjPromise)` for identity - `awaitJs(js, jsPromise)` bridges JS→KJ - `addObject(kj::Own)` returns `IoOwn` — the **only** safe way to store KJ I/O objects reachable from JS heap +- Cross-context actions (`IoCrossContextExecutor::tryExecute()`, `DeleteQueue::scheduleAction()`) can wait in the owning IoContext's queue, keeping what they capture alive: coalesce actions holding just a weak reference, and count native copies of JS data as external memory. See `scheduleAction()` in `io-own.h`. - Two-phase locking: `Worker::AsyncLock` (fair queue) → `Worker::Lock` (V8 isolate lock) - `ActorCacheOps` methods return `kj::OneOf>` — sync when cached, async otherwise - `OutputGate::lockWhile(promise)` blocks outgoing responses until the promise resolves diff --git a/src/workerd/io/BUILD.bazel b/src/workerd/io/BUILD.bazel index f63bcbd9a3d..b4da8906c53 100644 --- a/src/workerd/io/BUILD.bazel +++ b/src/workerd/io/BUILD.bazel @@ -751,6 +751,7 @@ kj_test( deps = [ ":io", "//src/workerd/tests:test-fixture", + "@capnp-cpp//src/kj:kj-async-io", ], ) diff --git a/src/workerd/io/incoming-request-test.c++ b/src/workerd/io/incoming-request-test.c++ index 6df2862fb91..da8daf7086f 100644 --- a/src/workerd/io/incoming-request-test.c++ +++ b/src/workerd/io/incoming-request-test.c++ @@ -9,6 +9,7 @@ #include #include #include +#include #include #include @@ -529,22 +530,20 @@ KJ_TEST("cancelPendingWorkForHibernation() logs a throwing destructor and keeps } // Constructs the fixture's actor from `source`, a Durable Object class with an onHibernate -// handler that appends to `globalThis.events`, then calls `body`. +// handler that appends to `globalThis.events`, then calls `body`. The fixture is set up with +// `params`, apart from its feature flags, module and actor, which this sets. void withOnHibernateActor(kj::StringPtr source, kj::StringPtr actorId, kj::FunctionParam body, - bool flushActorStorage = false) { + TestFixture::SetupParams params = {}) { capnp::MallocMessageBuilder flagsMessage; auto flags = flagsMessage.initRoot(); flags.setWorkerdExperimental(true); - TestFixture fixture({ - .featureFlags = flags.asReader(), - .mainModuleSource = source, - .actorId = Worker::Actor::Id(kj::str(actorId)), - .actorClassName = "default"_kj, - .useRealTimers = false, - .flushActorStorage = flushActorStorage, - }); + params.featureFlags = flags.asReader(); + params.mainModuleSource = source; + params.actorId = Worker::Actor::Id(kj::str(actorId)); + params.actorClassName = "default"_kj; + TestFixture fixture(kj::mv(params)); auto& actor = fixture.getActor(); auto context = fixture.newIoContext(); actor.setIoContext(kj::addRef(*context)); @@ -708,7 +707,7 @@ KJ_TEST("runOnHibernate() rejects when the handler's writes can't be flushed") { KJ_EXPECT_LOG(ERROR, "actorCacheFlush"); KJ_EXPECT_THROW_MESSAGE("broken.outputGateBroken", promise.wait(fixture.getWaitScope())); KJ_EXPECT(observer->outcome == EventOutcome::OK); - }, /*flushActorStorage=*/true); + }, {.flushActorStorage = true}); } KJ_TEST("canceling runOnHibernate() cancels pending work but keeps abort watchers armed") { @@ -768,5 +767,185 @@ KJ_TEST("ambient IoContext is hidden while another V8 isolate is entered") { }); } +// Expects `e` to be run()'s failure for lack of a current request: a JavaScript Error, so +// JavaScript sees its message rather than "internal error", carrying the detail callers check. +void expectNoCurrentIncomingRequestFailure(const kj::Exception& e) { + KJ_EXPECT(e.getDetail(NO_CURRENT_INCOMING_REQUEST_DETAIL_ID) != kj::none, e); + KJ_EXPECT(jsg::isTunneledException(e.getDescription()), e); + KJ_EXPECT(e.getDescription().contains("no request in progress"), e); +} + +// Expects the run() in startRun() to throw for lack of a current request, not return a promise. +template +void expectThrowsForNoCurrentIncomingRequest(StartRun&& startRun) { + KJ_IF_SOME(e, kj::runCatchingExceptions([&]() { auto promise = startRun(); })) { + expectNoCurrentIncomingRequestFailure(e); + } else { + KJ_FAIL_EXPECT("run() returned a promise without a current IncomingRequest"); + } +} + +// Expects a run() whose request ended while it waited for a lock to reject. +void expectRejectsForNoCurrentIncomingRequest(kj::WaitScope& waitScope, kj::Promise run) { + KJ_IF_SOME(e, kj::runCatchingExceptions([&]() { run.wait(waitScope); })) { + expectNoCurrentIncomingRequestFailure(e); + } else { + KJ_FAIL_EXPECT("run() succeeded without a current IncomingRequest"); + } +} + +KJ_TEST("run() fails without running its callback when there is no current request") { + TestFixture fixture({.autogateKeys = kj::arr(util::AutogateKey::DEFER_CROSS_CONTEXT_ACTIONS)}); + auto context = fixture.newIoContext(); + + bool ran = false; + expectThrowsForNoCurrentIncomingRequest( + [&]() { return context->run([&ran](Worker::Lock&) { ran = true; }); }); + KJ_EXPECT(!ran); +} + +KJ_TEST("run() in a critical section fails when there is no current request") { + TestFixture fixture({.autogateKeys = kj::arr(util::AutogateKey::DEFER_CROSS_CONTEXT_ACTIONS), + .actorId = Worker::Actor::Id(kj::str("run-critical-section-test"))}); + auto& waitScope = fixture.getWaitScope(); + auto context = fixture.newIoContext(); + + auto inputLock = fixture.getActor().getInputGate().wait(nullptr).wait(waitScope); + auto criticalSection = inputLock.startCriticalSection(); + kj::Maybe> runIn = criticalSection->clone(); + bool ran = false; + expectThrowsForNoCurrentIncomingRequest( + [&]() { return context->run([&ran](Worker::Lock&) { ran = true; }, kj::mv(runIn)); }); + KJ_EXPECT(!ran); +} + +KJ_TEST("run() fails if the request is gone once the input lock is acquired") { + TestFixture fixture({.autogateKeys = kj::arr(util::AutogateKey::DEFER_CROSS_CONTEXT_ACTIONS), + .actorId = Worker::Actor::Id(kj::str("run-input-lock-test"))}); + auto& waitScope = fixture.getWaitScope(); + auto context = fixture.newIoContext(); + auto request = fixture.newIncomingRequest(*context); + + auto inputLock = fixture.getActor().getInputGate().wait(nullptr).wait(waitScope); + bool ran = false; + auto run = context->run([&ran](Worker::Lock&) { ran = true; }); + KJ_EXPECT(!run.poll(waitScope)); + + fixture.drainAndDestroy(kj::mv(request)); + { auto released = kj::mv(inputLock); } + expectRejectsForNoCurrentIncomingRequest(waitScope, kj::mv(run)); + KJ_EXPECT(!ran); +} + +KJ_TEST("run() fails if the request is gone once the isolate lock is acquired") { + auto io = kj::setupAsyncIo(); + // Each fixture initializes the process's autogates, so both enable it. + TestFixture fixture({.waitScope = io.waitScope, + .autogateKeys = kj::arr(util::AutogateKey::DEFER_CROSS_CONTEXT_ACTIONS), + .useRealTimers = false}); + TestFixture other({.waitScope = io.waitScope, + .autogateKeys = kj::arr(util::AutogateKey::DEFER_CROSS_CONTEXT_ACTIONS), + .useRealTimers = false}); + auto context = fixture.newIoContext(); + auto request = fixture.newIncomingRequest(*context); + + // Holding another isolate's lock holds up the run, since a thread takes one at a time. + auto asyncLock = other.takeAsyncLock(); + bool ran = false; + auto run = context->run([&ran](Worker::Lock&) { ran = true; }); + KJ_EXPECT(!run.poll(io.waitScope)); + + fixture.drainAndDestroy(kj::mv(request)); + { auto released = kj::mv(asyncLock); } + expectRejectsForNoCurrentIncomingRequest(io.waitScope, kj::mv(run)); + KJ_EXPECT(!ran); +} + +KJ_TEST("run() in a critical section fails with the abort reason of an aborted context") { + TestFixture fixture({.autogateKeys = kj::arr(util::AutogateKey::DEFER_CROSS_CONTEXT_ACTIONS), + .actorId = Worker::Actor::Id(kj::str("run-critical-section-abort-test"))}); + auto& waitScope = fixture.getWaitScope(); + auto context = fixture.newIoContext(); + context->abort(KJ_EXCEPTION(FAILED, "test abort")); + + auto inputLock = fixture.getActor().getInputGate().wait(nullptr).wait(waitScope); + auto criticalSection = inputLock.startCriticalSection(); + kj::Maybe> runIn = criticalSection->clone(); + bool ran = false; + KJ_EXPECT_THROW_MESSAGE("test abort", + auto promise = context->run([&ran](Worker::Lock&) { ran = true; }, kj::mv(runIn))); + KJ_EXPECT(!ran); +} + +KJ_TEST("runOnHibernate() drops the cross-context actions queued while the actor was idle") { + constexpr kj::StringPtr source = R"SCRIPT( + import { DurableObject } from "cloudflare:workers"; + import { onHibernate } from "cloudflare:durable-objects"; + globalThis.events = ""; + export default class extends DurableObject { + async [onHibernate]() { + globalThis.events += "onHibernate;"; + } + } + )SCRIPT"_kj; + withOnHibernateActor(source, "on-hibernate-queued-actions-test"_kj, + [](TestFixture& fixture, IoContext& context, Worker::Actor& actor) { + auto executor = context.getCrossContextExecutor(); + uint ran = 0; + KJ_ASSERT(executor.tryExecute([&ran](jsg::Lock&) { ++ran; })); + fixture.pollEventLoop(); + + auto observer = kj::refcounted(); + runOnHibernate(fixture, actor, kj::addRef(*observer)).wait(fixture.getWaitScope()); + + KJ_EXPECT(observer->outcome == EventOutcome::OK); + KJ_EXPECT(readEvents(fixture) == "onHibernate;", readEvents(fixture)); + KJ_EXPECT(ran == 0, ran); + KJ_EXPECT(executor.getQueuedActionCountForTest() == 0, executor.getQueuedActionCountForTest()); + KJ_EXPECT(context.getAbortReason() == kj::none, context.getAbortReason()); + }, {.autogateKeys = kj::arr(util::AutogateKey::DEFER_CROSS_CONTEXT_ACTIONS)}); +} + +KJ_TEST("runOnHibernate() runs the cross-context actions queued during the handler, but not those " + "queued while the actor was idle") { + // The handler waits for the test to call resume() from a cross-context action. + constexpr kj::StringPtr source = R"SCRIPT( + import { DurableObject } from "cloudflare:workers"; + import { onHibernate } from "cloudflare:durable-objects"; + globalThis.events = ""; + export default class extends DurableObject { + async [onHibernate]() { + globalThis.events += "onHibernate;"; + await new Promise((resolve) => { globalThis.resume = resolve; }); + globalThis.events += "resumed;"; + } + } + )SCRIPT"_kj; + withOnHibernateActor(source, "on-hibernate-handler-actions-test"_kj, + [](TestFixture& fixture, IoContext& context, Worker::Actor& actor) { + auto& waitScope = fixture.getWaitScope(); + auto executor = context.getCrossContextExecutor(); + + // Queued just before the handler, so the context hasn't woken up for it yet. + uint idleRan = 0; + KJ_ASSERT(executor.tryExecute([&idleRan](jsg::Lock&) { ++idleRan; })); + auto observer = kj::refcounted(); + auto promise = runOnHibernate(fixture, actor, kj::addRef(*observer)); + KJ_EXPECT(!promise.poll(waitScope)); + KJ_EXPECT(readEvents(fixture) == "onHibernate;", readEvents(fixture)); + + KJ_ASSERT(executor.tryExecute([](jsg::Lock& js) { + auto resume = KJ_ASSERT_NONNULL(js.global().get(js, "resume"_kj).tryCast()); + resume.call(js, js.undefined(), js.undefined()); + })); + promise.wait(waitScope); + + KJ_EXPECT(observer->outcome == EventOutcome::OK); + KJ_EXPECT(readEvents(fixture) == "onHibernate;resumed;", readEvents(fixture)); + KJ_EXPECT(idleRan == 0, idleRan); + KJ_EXPECT(context.getAbortReason() == kj::none, context.getAbortReason()); + }, {.autogateKeys = kj::arr(util::AutogateKey::DEFER_CROSS_CONTEXT_ACTIONS)}); +} + } // namespace } // namespace workerd diff --git a/src/workerd/io/io-context.c++ b/src/workerd/io/io-context.c++ index b2fe3a2244a..1e34c1e9cb3 100644 --- a/src/workerd/io/io-context.c++ +++ b/src/workerd/io/io-context.c++ @@ -12,6 +12,7 @@ #include #include #include +#include #include #include #include @@ -341,12 +342,15 @@ tracing::InvocationSpanContext IoContext::IncomingRequest::getInvocationSpanCont // that sets waitedForWaitUntil). So, we can now safely add the request to // context->incomingRequests, which implies taking responsibility for draining on the way out. void IoContext::IncomingRequest::delivered(kj::SourceLocation location) { - delivered(location.function, location); + delivered(location.function, DropQueuedCrossContextActions::NO, location); } -void IoContext::IncomingRequest::delivered(kj::StringPtr eventType, kj::SourceLocation location) { +void IoContext::IncomingRequest::delivered(kj::StringPtr eventType, + DropQueuedCrossContextActions dropQueuedActions, + kj::SourceLocation location) { KJ_REQUIRE(!wasDelivered, "delivered() can only be called once"); - if (!context->incomingRequests.empty()) { + bool hadRequest = !context->incomingRequests.empty(); + if (hadRequest) { // There is already an IncomingRequest running in this context, and we're going to make it no // longer current. Make sure to attribute accumulated CPU time to it. auto& oldFront = context->incomingRequests.front(); @@ -395,6 +399,11 @@ void IoContext::IncomingRequest::delivered(kj::StringPtr eventType, kj::SourceLo // Record a new incoming request to actor metrics. a.getMetrics().startRequest(); } + + // A stateless context has a single request, before which nothing could reach its queue. + if (!hadRequest && context->actor != kj::none) { + context->handleActionsQueuedWhileIdle(eventType, dropQueuedActions); + } } kj::Date IoContext::IncomingRequest::now(kj::Maybe clockClamp) { @@ -656,6 +665,26 @@ void IoContext::logUncaughtExceptionAsync(UncaughtExceptionSource source, runnable, Worker::Lock::TakeSynchronously(metrics), kj::none, Runnable::Exceptional(true)); } +void IoContext::requireCurrentIncomingRequestForRun(ReportRunWithoutRequest report) { + if (incomingRequests.empty() && + util::Autogate::isEnabled(util::AutogateKey::DEFER_CROSS_CONTEXT_ACTIONS)) { + // As in runSingle(), the abort explains the failure better. + KJ_IF_SOME(ex, abortException) { + kj::throwException(ex.clone()); + } + // Not necessarily a bug (e.g. an idle actor), so throw a JS error, but report it in case it is + // one. + if (report) { + LOG_PERIODICALLY(WARNING, "IoContext::run() without a current IncomingRequest", + lastDeliveredLocation, kj::getStackTrace()); + } + auto e = JSG_KJ_EXCEPTION(FAILED, Error, + "Couldn't complete operation because the execution context has no request in progress."); + e.setDetail(NO_CURRENT_INCOMING_REQUEST_DETAIL_ID, kj::heapArray(0)); + kj::throwException(kj::mv(e)); + } +} + void IoContext::abort(kj::Exception&& e) { if (abortException != kj::none) { return; @@ -1995,25 +2024,93 @@ 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 run the queued actions. try { + if (!util::Autogate::isEnabled(util::AutogateKey::DEFER_CROSS_CONTEXT_ACTIONS)) { + // Without the autogate, every wake runs the actions, aborting the IoContext if that fails. + 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(); + }); + } + } + + 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 first, so that an action scheduled during the run wakes us again. + signal = queue.resetCrossThreadSignal(); + + // Without a request, the actions wait for an actor's next one, whose delivered() runs them. + // A stateless context gets no other request, so they wait until it is destroyed. An earlier + // run may also have taken them already. + if (context->incomingRequests.empty() || !queue.hasQueuedActions()) { + continue; + } + + kj::Promise actionsRun = context->runQueuedActions(); + if (context->actor != kj::none) { + // As in delivered(), keep the request draining until the run is done, so the actions + // aren't stranded until the actor's next request. Other contexts end with their request. + auto forked = actionsRun.fork(); + context->addWaitUntil(forked.addBranch()); + actionsRun = forked.addBranch(); + } + co_await actionsRun; + if (context->abortException != kj::none) { + co_return; + } } } catch (...) { context->abort(kj::getCaughtExceptionAsKj()); } } + +kj::Promise IoContext::runQueuedActions() { + bool tookActions = false; + try { + co_await runSingle([&tookActions](Worker::Lock& lock) { + // Take all the actions out first, since one may schedule another. + tookActions = true; + for (auto& action: IoContext::current().deleteQueue.queue->takeActions()) { + action(lock); + } + }); + } catch (...) { + auto e = kj::getCaughtExceptionAsKj(); + // Only losing the request before the actions were taken leaves them queued for the next one. + // Anything else aborts, including an action's own failure, whatever its exception's details. + if (tookActions || e.getDetail(NO_CURRENT_INCOMING_REQUEST_DETAIL_ID) == kj::none) { + abort(kj::mv(e)); + } + } +} + +void IoContext::handleActionsQueuedWhileIdle( + kj::StringPtr eventType, DropQueuedCrossContextActions drop) { + if (!util::Autogate::isEnabled(util::AutogateKey::DEFER_CROSS_CONTEXT_ACTIONS)) { + return; + } + if (drop) { + // Destroyed at the end of this scope, without the queue's lock held. + auto dropped = deleteQueue.queue->takeActions(); + if (dropped.size() > 0) { + LOG_PERIODICALLY(WARNING, "NOSENTRY IoContext dropped queued cross-context actions", + eventType, dropped.size()); + } + } else if (deleteQueue.queue->hasQueuedActions()) { + // Run them as a waitUntil task, so that the request can't finish draining before they have run. + addWaitUntil(runQueuedActions()); + } +} } // namespace workerd diff --git a/src/workerd/io/io-context.h b/src/workerd/io/io-context.h index 0262b40b10c..02671d72c74 100644 --- a/src/workerd/io/io-context.h +++ b/src/workerd/io/io-context.h @@ -52,6 +52,11 @@ namespace workerd { // Maximum depth to which blockConcurrencyWhile() calls may be nested. constexpr uint MAX_BLOCK_CONCURRENCY_WHILE_DEPTH = 64; +// Marks the exception IoContext::run() fails with when there is no current IncomingRequest, e.g. +// in an actor between events (only with the DEFER_CROSS_CONTEXT_ACTIONS autogate). Not transient: +// a retry needs the context to get another request. +constexpr kj::Exception::DetailTypeId NO_CURRENT_INCOMING_REQUEST_DETAIL_ID = 0x68a64a728b1be5c9ull; + // This wishes it were IoContext::Runnable::Exceptional. WD_STRONG_BOOL(IoContext_Runnable_Exceptional); // Whether the caller awaits a subrequest with its next awaitIo(), so that the subrequest span's @@ -60,6 +65,11 @@ WD_STRONG_BOOL(SpanAwaitedNext); // Whether IoContext::cancelPendingWork() also clears `tasks`, which in an actor holds the // abortWhen() watchers. WD_STRONG_BOOL(ClearTasks); +// Whether IoContext::run() reports a run without a current IncomingRequest as a possible bug. +WD_STRONG_BOOL(ReportRunWithoutRequest); +// Whether IoContext::IncomingRequest::delivered() drops the cross-context actions queued while the +// actor had no request, rather than run them. +WD_STRONG_BOOL(DropQueuedCrossContextActions); [[noreturn]] void throwExceededMemoryLimit(bool isActor); @@ -142,10 +152,16 @@ class IoContext_IncomingRequest final: public kj::PtrTarget { // the IoContext. // // If delivered() is never called, then drain() need not be called. + // + // With the DEFER_CROSS_CONTEXT_ACTIONS autogate, if the actor had no current request, this also + // runs the cross-context actions queued meanwhile (see DeleteQueue::scheduleAction()) as a + // waitUntil task of this request, or drops them if `dropQueuedActions` is YES. void delivered(kj::SourceLocation = kj::SourceLocation()); // As delivered(), naming the event type (e.g. "fetch", "alarm") for async tracing. Without it, // the request's async trace resource is named after the delivering function. - void delivered(kj::StringPtr eventType, kj::SourceLocation location = kj::SourceLocation()); + void delivered(kj::StringPtr eventType, + DropQueuedCrossContextActions dropQueuedActions = DropQueuedCrossContextActions::NO, + kj::SourceLocation location = kj::SourceLocation()); // Continues running the request in the background until it is "done", scheduling the work into // `waitUntilTasks` and keeping `self` alive until work is finished. @@ -525,6 +541,10 @@ class IoContext final: public kj::Refcounted, private kj::TaskSet::ErrorHandler // outlive its last IncomingRequest, so callers reachable from teardown -- destructors and // kj::defer cleanups in particular -- must check hasCurrentIncomingRequest() first and skip // the work rather than call this and fault. + // + // With the DEFER_CROSS_CONTEXT_ACTIONS autogate enabled, the callback doesn't run without a + // request: run() throws, or rejects if the request ends while it waits for the locks, with + // NO_CURRENT_INCOMING_REQUEST_DETAIL_ID. An aborted context fails with its abort reason instead. template auto run(Func&& func, kj::Maybe inputLock = kj::none) KJ_WARN_UNUSED_RESULT { if constexpr (runFuncAcceptsIoContext) { @@ -1427,10 +1447,20 @@ class IoContext final: public kj::Refcounted, private kj::TaskSet::ErrorHandler static constexpr bool runFuncAcceptsIoContext = std::invocable&, Worker::Lock&, IoContext&>; - // Internal implementation of run(), always invoked with a single-arg callback. + // Throws a JavaScript Error carrying NO_CURRENT_INCOMING_REQUEST_DETAIL_ID if there is no + // current IncomingRequest and the DEFER_CROSS_CONTEXT_ACTIONS autogate is enabled. If `report` + // is YES, also logs the miss, at most once an hour, as a possible bug. An aborted context throws + // its abort exception instead, without logging. + void requireCurrentIncomingRequestForRun(ReportRunWithoutRequest report); + + // Internal implementation of run(), always invoked with a single-arg callback. `report` is + // passed to requireCurrentIncomingRequestForRun() for the check made when called. The checks + // made after waiting for a lock never report: the request can end during any wait, so losing + // it there is no sign of a bug. template - kj::PromiseForResult runSingle( - Func&& func, kj::Maybe inputLock = kj::none); + kj::PromiseForResult runSingle(Func&& func, + kj::Maybe inputLock = kj::none, + ReportRunWithoutRequest report = ReportRunWithoutRequest::YES); template kj::PromiseForResult runSingle( @@ -1522,6 +1552,15 @@ class IoContext final: public kj::Refcounted, private kj::TaskSet::ErrorHandler kj::Promise deleteQueueSignalTask; static kj::Promise startDeleteQueueSignalTask(IoContext* context); + // Runs the actions queued on deleteQueue under the input lock and current IncomingRequest. If + // the run loses its request while waiting for a lock, they stay queued; other failures abort. + kj::Promise runQueuedActions(); + + // Called by IncomingRequest::delivered() when an actor gets a request after having none. With + // the DEFER_CROSS_CONTEXT_ACTIONS autogate, runs the cross-context actions queued meanwhile as a + // waitUntil task of the request, or drops them. + void handleActionsQueuedWhileIdle(kj::StringPtr eventType, DropQueuedCrossContextActions drop); + friend class Finalizeable; friend class DeleteQueue; template @@ -1557,11 +1596,13 @@ template kj::PromiseForResult IoContext::runSingle( Func&& func, kj::Maybe> criticalSection) { KJ_IF_SOME(cs, criticalSection) { + requireCurrentIncomingRequestForRun(ReportRunWithoutRequest::YES); return cs.get() ->wait(getCurrentTraceSpan()) .then([self = addWeakToThis(), func = kj::fwd(func)]( InputGate::Lock&& inputLock) mutable { - return self.assertLive().runSingle(kj::fwd(func), kj::mv(inputLock)); + return self.assertLive().runSingle( + kj::fwd(func), kj::mv(inputLock), ReportRunWithoutRequest::NO); }); } else { return runSingle(kj::fwd(func)); @@ -1570,12 +1611,14 @@ kj::PromiseForResult IoContext::runSingle( template kj::PromiseForResult IoContext::runSingle( - Func&& func, kj::Maybe inputLock) { + Func&& func, kj::Maybe inputLock, ReportRunWithoutRequest report) { // Before we try running anything, let's make sure our IoContext hasn't been aborted. If it has // been aborted, there's likely not an active request so later operations will fail anyway. KJ_IF_SOME(ex, abortException) { return ex.clone(); } + // The setup below needs the current request. Re-entry with the input lock checks again. + requireCurrentIncomingRequestForRun(report); kj::Promise asyncLockPromise = nullptr; KJ_IF_SOME(a, actor) { @@ -1584,7 +1627,8 @@ kj::PromiseForResult IoContext::runSingle( .wait(getCurrentTraceSpan()) .then([self = addWeakToThis(), func = kj::fwd(func)]( InputGate::Lock&& inputLock) mutable { - return self.assertLive().runSingle(kj::fwd(func), kj::mv(inputLock)); + return self.assertLive().runSingle( + kj::fwd(func), kj::mv(inputLock), ReportRunWithoutRequest::NO); }); } @@ -1596,10 +1640,11 @@ kj::PromiseForResult IoContext::runSingle( return asyncLockPromise.then([self = addWeakToThis(), inputLock = kj::mv(inputLock), func = kj::fwd(func)](Worker::AsyncLock lock) mutable { auto& context = self.assertLive(); - // Re-check if context was aborted while we waited for the lock. + // Re-check if the context was aborted or lost its request while we waited for the lock. KJ_IF_SOME(ex, context.abortException) { kj::throwException(ex.clone()); } + context.requireCurrentIncomingRequestForRun(ReportRunWithoutRequest::NO); using Result = decltype(func(kj::instance())); diff --git a/src/workerd/io/io-own-test.c++ b/src/workerd/io/io-own-test.c++ index 5d5540f6988..f943ab8f4f2 100644 --- a/src/workerd/io/io-own-test.c++ +++ b/src/workerd/io/io-own-test.c++ @@ -1,6 +1,9 @@ #include #include +#include +#include +#include #include namespace workerd { @@ -94,5 +97,500 @@ KJ_TEST("ReverseIoOwn remains safe after its IoContext is destroyed") { } } +class ExpectNoTaskFailures final: public kj::TaskSet::ErrorHandler { + public: + void taskFailed(kj::Exception&& exception) override { + KJ_FAIL_EXPECT(exception); + } +}; + +KJ_TEST("cross-context actions scheduled while actions run still wake the IoContext") { + TestFixture fixture({.autogateKeys = kj::arr(util::AutogateKey::DEFER_CROSS_CONTEXT_ACTIONS), + .actorId = Worker::Actor::Id(kj::str("rearm-test"))}); + auto context = fixture.newIoContext(); + auto request = fixture.newIncomingRequest(*context); + auto executor = context->getCrossContextExecutor(); + + uint ran = 0; + KJ_ASSERT(executor.tryExecute([&executor, &ran](jsg::Lock&) { + ++ran; + // Scheduling during the run must still wake the context. + KJ_ASSERT(executor.tryExecute([&ran](jsg::Lock&) { ++ran; })); + })); + fixture.pollEventLoop(); + KJ_EXPECT(ran == 2, ran); + fixture.drainAndDestroy(kj::mv(request)); +} + +KJ_TEST("cross-context actions that fail abort the IoContext, whatever details they carry") { + TestFixture fixture({.autogateKeys = kj::arr(util::AutogateKey::DEFER_CROSS_CONTEXT_ACTIONS), + .actorId = Worker::Actor::Id(kj::str("failing-action-test"))}); + auto context = fixture.newIoContext(); + auto request = fixture.newIncomingRequest(*context); + + // An action that fails with the detail that keeps actions queued must still abort the context. + KJ_ASSERT(context->getCrossContextExecutor().tryExecute([](jsg::Lock&) { + auto e = KJ_EXCEPTION(FAILED, "action failed"); + e.setDetail(NO_CURRENT_INCOMING_REQUEST_DETAIL_ID, kj::heapArray(0)); + kj::throwException(kj::mv(e)); + })); + fixture.pollEventLoop(); + KJ_IF_SOME(reason, context->getAbortReason()) { + KJ_EXPECT(reason.getDescription() == "action failed", reason); + } else { + KJ_FAIL_ASSERT("the IoContext wasn't aborted"); + } + fixture.drainAndDestroy(kj::mv(request)); +} + +KJ_TEST("a cross-context action that fails drops the actions taken with it") { + TestFixture fixture({.autogateKeys = kj::arr(util::AutogateKey::DEFER_CROSS_CONTEXT_ACTIONS), + .actorId = Worker::Actor::Id(kj::str("failing-batch-test"))}); + auto context = fixture.newIoContext(); + auto request = fixture.newIncomingRequest(*context); + auto executor = context->getCrossContextExecutor(); + + // Queued together, so one run takes all three. + kj::Vector ran; + KJ_ASSERT(executor.tryExecute([&ran](jsg::Lock&) { ran.add("first"_kj); })); + KJ_ASSERT(executor.tryExecute( + [](jsg::Lock&) { kj::throwException(KJ_EXCEPTION(FAILED, "second failed")); })); + KJ_ASSERT(executor.tryExecute([&ran](jsg::Lock&) { ran.add("third"_kj); })); + fixture.pollEventLoop(); + + KJ_EXPECT(ran.size() == 1 && ran[0] == "first", kj::strArray(ran, ", ")); + KJ_EXPECT(executor.getQueuedActionCountForTest() == 0, executor.getQueuedActionCountForTest()); + KJ_IF_SOME(reason, context->getAbortReason()) { + KJ_EXPECT(reason.getDescription() == "second failed", reason); + } else { + KJ_FAIL_ASSERT("the IoContext wasn't aborted"); + } + fixture.drainAndDestroy(kj::mv(request)); +} + +KJ_TEST("cross-context actions wait for an I/O continuation holding the input lock") { + TestFixture fixture({.autogateKeys = kj::arr(util::AutogateKey::DEFER_CROSS_CONTEXT_ACTIONS), + .actorId = Worker::Actor::Id(kj::str("input-lock-test"))}); + auto context = fixture.newIoContext(); + auto request = fixture.newIncomingRequest(*context); + + kj::Vector order; + auto io = kj::newPromiseAndFulfiller(); + fixture.enterContext(*request, [&](TestFixture::Environment& env) { + env.context + .awaitIoWithInputLock(env.js, kj::mv(io.promise), [&order](jsg::Lock&) { + order.add("continuation"_kj); + }).markAsHandled(env.js); + KJ_ASSERT(env.context.getCrossContextExecutor().tryExecute( + [&order](jsg::Lock&) { order.add("action"_kj); })); + }); + fixture.pollEventLoop(); + KJ_EXPECT(order.empty(), kj::strArray(order, ", ")); + + io.fulfiller->fulfill(); + fixture.pollEventLoop(); + KJ_EXPECT(order.size() == 2 && order[0] == "continuation" && order[1] == "action", + kj::strArray(order, ", ")); + fixture.drainAndDestroy(kj::mv(request)); +} + +KJ_TEST("cross-context actions run while an older request outlives a newer one") { + TestFixture fixture({.autogateKeys = kj::arr(util::AutogateKey::DEFER_CROSS_CONTEXT_ACTIONS), + .actorId = Worker::Actor::Id(kj::str("older-request-test"))}); + auto context = fixture.newIoContext(); + auto older = fixture.newIncomingRequest(*context); + auto newer = fixture.newIncomingRequest(*context); + + // Queued while the newer request is current, which then ends before the signal task wakes, + // leaving the older one current. + uint ran = 0; + KJ_ASSERT(context->getCrossContextExecutor().tryExecute([&ran](jsg::Lock&) { ++ran; })); + newer = nullptr; + KJ_ASSERT(context->hasCurrentIncomingRequest()); + fixture.pollEventLoop(); + KJ_EXPECT(ran == 1, ran); + fixture.drainAndDestroy(kj::mv(older)); +} + +KJ_TEST( + "a request doesn't finish draining while actions queued during it wait for the input lock") { + TestFixture fixture({.autogateKeys = kj::arr(util::AutogateKey::DEFER_CROSS_CONTEXT_ACTIONS), + .actorId = Worker::Actor::Id(kj::str("drain-wait-test"))}); + auto& waitScope = fixture.getWaitScope(); + auto context = fixture.newIoContext(); + auto request = fixture.newIncomingRequest(*context); + + // Queued after delivery, so the signal task runs these, but the input lock holds them up until + // the request is draining. + auto inputLock = fixture.getActor().getInputGate().wait(nullptr).wait(waitScope); + uint ran = 0; + KJ_ASSERT(context->getCrossContextExecutor().tryExecute([&ran](jsg::Lock&) { ++ran; })); + fixture.pollEventLoop(); + ExpectNoTaskFailures errorHandler; + kj::TaskSet drains(errorHandler); + request->drain(drains, kj::mv(request)); + fixture.pollEventLoop(); + KJ_EXPECT(context->hasCurrentIncomingRequest()); + KJ_EXPECT(ran == 0, ran); + + { auto released = kj::mv(inputLock); } + drains.onEmpty().wait(waitScope); + KJ_EXPECT(ran == 1, ran); + KJ_EXPECT(!context->hasCurrentIncomingRequest()); + KJ_EXPECT(context->getAbortReason() == kj::none, context->getAbortReason()); +} + +KJ_TEST( + "cross-context actions stay queued if the request ends while they wait for the input lock") { + TestFixture fixture({.autogateKeys = kj::arr(util::AutogateKey::DEFER_CROSS_CONTEXT_ACTIONS), + .actorId = Worker::Actor::Id(kj::str("input-lock-wait-test"))}); + auto& waitScope = fixture.getWaitScope(); + auto context = fixture.newIoContext(); + auto request = fixture.newIncomingRequest(*context); + + auto inputLock = fixture.getActor().getInputGate().wait(nullptr).wait(waitScope); + uint ran = 0; + KJ_ASSERT(context->getCrossContextExecutor().tryExecute([&ran](jsg::Lock&) { ++ran; })); + fixture.pollEventLoop(); + + // Destroyed without draining, so nothing holds the request open for the actions' run. + request = nullptr; + { auto released = kj::mv(inputLock); } + fixture.pollEventLoop(); + KJ_EXPECT(ran == 0, ran); + KJ_EXPECT(context->getAbortReason() == kj::none, context->getAbortReason()); + + auto next = fixture.newIncomingRequest(*context); + fixture.pollEventLoop(); + KJ_EXPECT(ran == 1, ran); + fixture.drainAndDestroy(kj::mv(next)); +} + +KJ_TEST("cross-context actions stay queued if the actor shuts down while they wait for a lock") { + auto io = kj::setupAsyncIo(); + // Each fixture initializes the process's autogates, so both enable it. + TestFixture fixture({.waitScope = io.waitScope, + .autogateKeys = kj::arr(util::AutogateKey::DEFER_CROSS_CONTEXT_ACTIONS), + .actorId = Worker::Actor::Id(kj::str("isolate-lock-wait-test")), + .useRealTimers = false}); + TestFixture other({.waitScope = io.waitScope, + .autogateKeys = kj::arr(util::AutogateKey::DEFER_CROSS_CONTEXT_ACTIONS), + .useRealTimers = false}); + auto context = fixture.newIoContext(); + + // The request drains until the actor shuts down. + auto request = fixture.newIncomingRequest(*context); + context->addWaitUntil(kj::Promise(kj::NEVER_DONE)); + ExpectNoTaskFailures errorHandler; + kj::TaskSet drains(errorHandler); + request->drain(drains, kj::mv(request)); + + // Holding another isolate's lock holds up the run, since a thread takes one at a time. + auto asyncLock = other.takeAsyncLock(); + uint ran = 0; + auto executor = context->getCrossContextExecutor(); + KJ_ASSERT(executor.tryExecute([&ran](jsg::Lock&) { ++ran; })); + fixture.pollEventLoop(); + + fixture.getActor().shutdown(0); + fixture.pollEventLoop(); + KJ_EXPECT(!context->hasCurrentIncomingRequest()); + { auto released = kj::mv(asyncLock); } + fixture.pollEventLoop(); + KJ_EXPECT(ran == 0, ran); + KJ_EXPECT(executor.getQueuedActionCountForTest() == 1, executor.getQueuedActionCountForTest()); + KJ_EXPECT(context->getAbortReason() == kj::none, context->getAbortReason()); +} + +KJ_TEST("cross-context actions wait for an idle actor's next request") { + TestFixture fixture({.autogateKeys = kj::arr(util::AutogateKey::DEFER_CROSS_CONTEXT_ACTIONS), + .actorId = Worker::Actor::Id(kj::str("idle-actor-test"))}); + auto context = fixture.newIoContext(); + fixture.drainAndDestroy(fixture.newIncomingRequest(*context)); + + uint ran = 0; + KJ_ASSERT(context->getCrossContextExecutor().tryExecute([&ran](jsg::Lock&) { ++ran; })); + fixture.pollEventLoop(); + KJ_EXPECT(ran == 0, ran); + KJ_EXPECT(context->getAbortReason() == kj::none, context->getAbortReason()); + + // Delivering the next request is enough; it doesn't have to run anything itself. + auto request = fixture.newIncomingRequest(*context); + fixture.pollEventLoop(); + KJ_EXPECT(ran == 1, ran); + fixture.drainAndDestroy(kj::mv(request)); +} + +KJ_TEST("an idle actor's queued actions run before its next request's own work") { + TestFixture fixture({.autogateKeys = kj::arr(util::AutogateKey::DEFER_CROSS_CONTEXT_ACTIONS), + .actorId = Worker::Actor::Id(kj::str("idle-actor-order-test"))}); + auto context = fixture.newIoContext(); + fixture.drainAndDestroy(fixture.newIncomingRequest(*context)); + + kj::Vector order; + KJ_ASSERT(context->getCrossContextExecutor().tryExecute( + [&order](jsg::Lock&) { order.add("action"_kj); })); + fixture.pollEventLoop(); + KJ_EXPECT(order.empty(), kj::strArray(order, ", ")); + + // Delivery starts the actions' run, which takes the input lock before the request's own run(). + auto request = fixture.newIncomingRequest(*context); + context->run([&order](Worker::Lock&) { order.add("handler"_kj); }).wait(fixture.getWaitScope()); + KJ_EXPECT(order.size() == 2 && order[0] == "action" && order[1] == "handler", + kj::strArray(order, ", ")); + fixture.drainAndDestroy(kj::mv(request)); +} + +KJ_TEST("a cross-context action that fails when an idle actor's next request runs it aborts the " + "IoContext") { + TestFixture fixture({.autogateKeys = kj::arr(util::AutogateKey::DEFER_CROSS_CONTEXT_ACTIONS), + .actorId = Worker::Actor::Id(kj::str("idle-failing-action-test"))}); + auto context = fixture.newIoContext(); + fixture.drainAndDestroy(fixture.newIncomingRequest(*context)); + + KJ_ASSERT(context->getCrossContextExecutor().tryExecute( + [](jsg::Lock&) { kj::throwException(KJ_EXCEPTION(FAILED, "action failed")); })); + fixture.pollEventLoop(); + KJ_EXPECT(context->getAbortReason() == kj::none, context->getAbortReason()); + + auto request = fixture.newIncomingRequest(*context); + fixture.pollEventLoop(); + KJ_IF_SOME(reason, context->getAbortReason()) { + KJ_EXPECT(reason.getDescription() == "action failed", reason); + } else { + KJ_FAIL_ASSERT("the IoContext wasn't aborted"); + } + fixture.drainAndDestroy(kj::mv(request)); +} + +KJ_TEST("an idle actor's next request doesn't finish draining before its queued actions run") { + TestFixture fixture({.autogateKeys = kj::arr(util::AutogateKey::DEFER_CROSS_CONTEXT_ACTIONS), + .actorId = Worker::Actor::Id(kj::str("short-request-test"))}); + auto& waitScope = fixture.getWaitScope(); + auto context = fixture.newIoContext(); + fixture.drainAndDestroy(fixture.newIncomingRequest(*context)); + + uint ran = 0; + KJ_ASSERT(context->getCrossContextExecutor().tryExecute([&ran](jsg::Lock&) { ++ran; })); + fixture.pollEventLoop(); + KJ_EXPECT(ran == 0, ran); + + // The next request drains at once, like an event that awaits nothing, while the input lock + // holds up the actions. + auto inputLock = fixture.getActor().getInputGate().wait(nullptr).wait(waitScope); + auto request = fixture.newIncomingRequest(*context); + ExpectNoTaskFailures errorHandler; + kj::TaskSet drains(errorHandler); + request->drain(drains, kj::mv(request)); + fixture.pollEventLoop(); + KJ_EXPECT(context->hasCurrentIncomingRequest()); + KJ_EXPECT(ran == 0, ran); + + { auto released = kj::mv(inputLock); } + drains.onEmpty().wait(waitScope); + KJ_EXPECT(ran == 1, ran); + KJ_EXPECT(!context->hasCurrentIncomingRequest()); + KJ_EXPECT(context->getAbortReason() == kj::none, context->getAbortReason()); +} + +KJ_TEST("an idle actor's queued actions stay queued if it shuts down before its next request " + "runs them") { + auto io = kj::setupAsyncIo(); + // Each fixture initializes the process's autogates, so both enable it. + TestFixture fixture({.waitScope = io.waitScope, + .autogateKeys = kj::arr(util::AutogateKey::DEFER_CROSS_CONTEXT_ACTIONS), + .actorId = Worker::Actor::Id(kj::str("next-request-shutdown-test")), + .useRealTimers = false}); + TestFixture other({.waitScope = io.waitScope, + .autogateKeys = kj::arr(util::AutogateKey::DEFER_CROSS_CONTEXT_ACTIONS), + .useRealTimers = false}); + auto context = fixture.newIoContext(); + fixture.drainAndDestroy(fixture.newIncomingRequest(*context)); + + uint ran = 0; + auto executor = context->getCrossContextExecutor(); + KJ_ASSERT(executor.tryExecute([&ran](jsg::Lock&) { ++ran; })); + fixture.pollEventLoop(); + + // Holding another isolate's lock holds up the run that delivering the next request starts, and + // the request drains until the actor shuts down. + auto asyncLock = other.takeAsyncLock(); + auto request = fixture.newIncomingRequest(*context); + ExpectNoTaskFailures errorHandler; + kj::TaskSet drains(errorHandler); + request->drain(drains, kj::mv(request)); + fixture.pollEventLoop(); + KJ_EXPECT(context->hasCurrentIncomingRequest()); + + fixture.getActor().shutdown(0); + fixture.pollEventLoop(); + KJ_EXPECT(!context->hasCurrentIncomingRequest()); + { auto released = kj::mv(asyncLock); } + fixture.pollEventLoop(); + KJ_EXPECT(ran == 0, ran); + KJ_EXPECT(executor.getQueuedActionCountForTest() == 1, executor.getQueuedActionCountForTest()); + KJ_EXPECT(context->getAbortReason() == kj::none, context->getAbortReason()); +} + +KJ_TEST("without the autogate, cross-context actions abort an idle actor") { + TestFixture fixture({.ignoreAllAutogatesEnv = true, + .actorId = Worker::Actor::Id(kj::str("idle-actor-gate-off-test"))}); + auto context = fixture.newIoContext(); + fixture.drainAndDestroy(fixture.newIncomingRequest(*context)); + + uint ran = 0; + KJ_ASSERT(context->getCrossContextExecutor().tryExecute([&ran](jsg::Lock&) { ++ran; })); + fixture.pollEventLoop(); + KJ_EXPECT(ran == 0, ran); + KJ_IF_SOME(reason, context->getAbortReason()) { + KJ_EXPECT(reason.getDescription().contains("no current IncomingRequest"), reason); + } else { + KJ_FAIL_EXPECT("the IoContext wasn't aborted"); + } +} + +KJ_TEST("a stateless context's cross-context actions wait without aborting it once its request " + "has ended") { + TestFixture fixture({.autogateKeys = kj::arr(util::AutogateKey::DEFER_CROSS_CONTEXT_ACTIONS)}); + auto context = fixture.newStatelessIoContext(); + fixture.drainAndDestroy(fixture.newIncomingRequest(*context)); + + uint ran = 0; + auto executor = context->getCrossContextExecutor(); + KJ_ASSERT(executor.tryExecute([&ran](jsg::Lock&) { ++ran; })); + fixture.pollEventLoop(); + KJ_EXPECT(ran == 0, ran); + KJ_EXPECT(executor.getQueuedActionCountForTest() == 1, executor.getQueuedActionCountForTest()); + KJ_EXPECT(context->getAbortReason() == kj::none, context->getAbortReason()); +} + +KJ_TEST("a stateless context's cross-context actions stay queued if its request ends while they " + "wait for the isolate lock") { + auto io = kj::setupAsyncIo(); + // Each fixture initializes the process's autogates, so both enable it. + TestFixture fixture({.waitScope = io.waitScope, + .autogateKeys = kj::arr(util::AutogateKey::DEFER_CROSS_CONTEXT_ACTIONS), + .useRealTimers = false}); + TestFixture other({.waitScope = io.waitScope, + .autogateKeys = kj::arr(util::AutogateKey::DEFER_CROSS_CONTEXT_ACTIONS), + .useRealTimers = false}); + auto context = fixture.newStatelessIoContext(); + auto request = fixture.newIncomingRequest(*context); + + // Holding another isolate's lock holds up the run, since a thread takes one at a time. + auto asyncLock = other.takeAsyncLock(); + uint ran = 0; + auto executor = context->getCrossContextExecutor(); + KJ_ASSERT(executor.tryExecute([&ran](jsg::Lock&) { ++ran; })); + fixture.pollEventLoop(); + + fixture.drainAndDestroy(kj::mv(request)); + { auto released = kj::mv(asyncLock); } + fixture.pollEventLoop(); + KJ_EXPECT(ran == 0, ran); + KJ_EXPECT(executor.getQueuedActionCountForTest() == 1, executor.getQueuedActionCountForTest()); + KJ_EXPECT(context->getAbortReason() == kj::none, context->getAbortReason()); +} + +// An actor whose class publishes its `ctx`. createPromise() creates a promise for another context +// to settle with settle(). The promise's reaction uses the actor's storage, which needs a current +// request, and appends its progress to `globalThis.events`. +constexpr kj::StringPtr PROMISE_ACTOR_SOURCE = R"SCRIPT( + import { DurableObject } from "cloudflare:workers"; + globalThis.events = ""; + export default class extends DurableObject { + constructor(ctx, env) { + super(ctx, env); + globalThis.ctx = ctx; + } + } + globalThis.createPromise = (key) => { + const { promise, resolve } = Promise.withResolvers(); + globalThis.settle = resolve; + promise.then(async (value) => { + globalThis.events += "reaction;"; + await globalThis.ctx.storage.put(key, value); + globalThis.events += `stored:${await globalThis.ctx.storage.get(key)};`; + }); + }; +)SCRIPT"_kj; + +// Calls the global function `name` with the string `argument`. +void callGlobal(jsg::Lock& js, kj::StringPtr name, kj::StringPtr argument) { + auto function = KJ_ASSERT_NONNULL(js.global().get(js, name).tryCast()); + function.call(js, js.undefined(), js.str(argument)); +} + +kj::String readEvents(TestFixture& fixture) { + kj::String events; + fixture.enterWorkerLock([&](Worker::Lock& lock) { + jsg::Lock& js = lock; + js.withinHandleScope([&]() { + v8::Context::Scope contextScope(lock.getContext()); + events = js.global().get(js, "events"_kj).toString(js); + }); + }); + return events; +} + +// Runs `body` with an actor running PROMISE_ACTOR_SOURCE that created a promise in a request that +// has since ended. A stateless context in the same isolate has settled the promise, so its +// reaction is queued on the idle actor. +void withPromiseSettledOnIdleActor( + kj::StringPtr actorId, kj::FunctionParam body) { + capnp::MallocMessageBuilder flagsMessage; + auto flags = flagsMessage.initRoot(); + flags.setHandleCrossRequestPromiseResolution(true); + TestFixture fixture({.featureFlags = flags.asReader(), + .autogateKeys = kj::arr(util::AutogateKey::DEFER_CROSS_CONTEXT_ACTIONS), + .mainModuleSource = PROMISE_ACTOR_SOURCE, + .actorId = Worker::Actor::Id(kj::str(actorId)), + .actorClassName = "default"_kj}); + auto context = fixture.newIoContext(); + fixture.getActor().setIoContext(kj::addRef(*context)); + + auto creator = fixture.newIncomingRequest(*context); + fixture.enterContext(*creator, + [](TestFixture::Environment& env) { callGlobal(env.js, "createPromise"_kj, "key"_kj); }); + fixture.drainAndDestroy(kj::mv(creator)); + + auto statelessContext = fixture.newStatelessIoContext(); + auto settler = fixture.newIncomingRequest(*statelessContext); + fixture.enterContext( + *settler, [](TestFixture::Environment& env) { callGlobal(env.js, "settle"_kj, "value"_kj); }); + fixture.drainAndDestroy(kj::mv(settler)); + fixture.pollEventLoop(); + KJ_EXPECT(context->getCrossContextExecutor().getQueuedActionCountForTest() == 1); + KJ_EXPECT(readEvents(fixture) == "", readEvents(fixture)); + KJ_EXPECT(context->getAbortReason() == kj::none, context->getAbortReason()); + + body(fixture, *context); +} + +KJ_TEST("a promise settled from another context waits for an idle actor's next request") { + withPromiseSettledOnIdleActor( + "idle-promise-test"_kj, [](TestFixture& fixture, IoContext& context) { + auto request = fixture.newIncomingRequest(context); + fixture.pollEventLoop(); + KJ_EXPECT(readEvents(fixture) == "reaction;stored:value;", readEvents(fixture)); + fixture.drainAndDestroy(kj::mv(request)); + }); +} + +KJ_TEST("an idle actor's next request doesn't finish draining before a promise settled from " + "another context runs") { + withPromiseSettledOnIdleActor( + "short-request-promise-test"_kj, [](TestFixture& fixture, IoContext& context) { + // The next request drains at once, like an event that awaits nothing. + auto request = fixture.newIncomingRequest(context); + ExpectNoTaskFailures errorHandler; + kj::TaskSet drains(errorHandler); + request->drain(drains, kj::mv(request)); + drains.onEmpty().wait(fixture.getWaitScope()); + KJ_EXPECT(readEvents(fixture) == "reaction;stored:value;", readEvents(fixture)); + KJ_EXPECT(!context.hasCurrentIncomingRequest()); + }); +} + } // namespace } // namespace workerd diff --git a/src/workerd/io/io-own.c++ b/src/workerd/io/io-own.c++ index 14496432e5a..106997b3586 100644 --- a/src/workerd/io/io-own.c++ +++ b/src/workerd/io/io-own.c++ @@ -3,6 +3,8 @@ #include "io-context.h" #include +#include +#include namespace workerd { @@ -69,6 +71,30 @@ bool DeleteQueue::isDefunct() const { return *crossThreadDeleteQueue.lockShared() == kj::none; } +kj::Array> DeleteQueue::takeActions() const { + auto lock = crossThreadDeleteQueue.lockExclusive(); + KJ_IF_SOME(state, *lock) { + return state.actions.releaseAsArray(); + } + return nullptr; +} + +bool DeleteQueue::hasQueuedActions() const { + auto lock = crossThreadDeleteQueue.lockShared(); + KJ_IF_SOME(state, *lock) { + return !state.actions.empty(); + } + return false; +} + +size_t DeleteQueue::getQueuedActionCountForTest() const { + auto lock = crossThreadDeleteQueue.lockShared(); + KJ_IF_SOME(state, *lock) { + return state.actions.size(); + } + return 0; +} + void DeleteQueue::checkFarGet(const DeleteQueue& deleteQueue, const std::type_info& type) { IoContext::current().checkFarGet(deleteQueue, type); } @@ -93,6 +119,31 @@ kj::Promise DeleteQueue::resetCrossThreadSignal() const { } } +DeleteQueuePtr::~DeleteQueuePtr() noexcept(false) { + auto ptr = queue.get(); + if (ptr != nullptr) { + size_t droppedActions = 0; + { + auto lock = ptr->crossThreadDeleteQueue.lockExclusive(); + KJ_IF_SOME(state, *lock) { + droppedActions = state.actions.size(); + // The delete queue state may include a kj::CrossThreadPromiseFulfiller that + // needs to be destroyed. To do so, we need to allow async destructors here. + // We only want to destroy the crossThreadFulfiller in this scope tho, not + // everything that may be in the queue. + kj::AllowAsyncDestructorsScope scope; + state.crossThreadFulfiller = kj::none; + } + *lock = kj::none; + } + if (droppedActions > 0 && + util::Autogate::isEnabled(util::AutogateKey::DEFER_CROSS_CONTEXT_ACTIONS)) { + LOG_PERIODICALLY(WARNING, + "NOSENTRY IoContext dropped queued cross-context actions when destroyed", droppedActions); + } + } +} + OwnedObjectList::~OwnedObjectList() noexcept(false) { for (;;) { // We want to have the same order of operations as the recursive destructor here. Without this @@ -141,4 +192,8 @@ bool IoCrossContextExecutor::isTargetDestroyed() const { return deleteQueue->isDefunct(); } +size_t IoCrossContextExecutor::getQueuedActionCountForTest() const { + return deleteQueue->getQueuedActionCountForTest(); +} + } // namespace workerd diff --git a/src/workerd/io/io-own.h b/src/workerd/io/io-own.h index 29ab44f0eff..8bf119b72a7 100644 --- a/src/workerd/io/io-own.h +++ b/src/workerd/io/io-own.h @@ -92,10 +92,22 @@ class DeleteQueue: public kj::AtomicRefcounted { void scheduleDeletion(OwnedObject* object) const; - // Schedules the given action to run in the IoContext that owns this queue, the next time - // that context drains its queue (in its own thread, under the isolate lock). If the owning - // IoContext has already been destroyed, the action is dropped and a warning about - // cross-request promise resolution is logged to the current context, if any. + // Schedules the given action to run in the IoContext that owns this queue like one of its + // events: on its thread, under the isolate lock, an actor's input lock and the current + // IncomingRequest, whose limits it counts toward. + // + // With the DEFER_CROSS_CONTEXT_ACTIONS autogate, an action queued while there is no request (an + // actor between events), or whose run loses its request while waiting for a lock, waits for the + // next one; without it, the context is aborted. A stateless context gets no next request, so + // its actions wait until it is destroyed. An actor's onHibernate event drops the actions queued + // before it (see Worker::Actor::runOnHibernate()). + // + // Only the isolate's memory limit bounds the queue, and only through what the actions keep + // alive on its heap, so coalesce actions holding just a weak reference, and count native copies + // of JS data as external memory. + // + // If the context is destroyed first, the action is dropped; if it already was, a cross-request + // promise resolution warning is logged to the current context, if any. void scheduleAction(jsg::Lock& js, kj::Function&& action) const; // Like scheduleAction(), but for callers where a destroyed target context is an expected, @@ -110,17 +122,16 @@ class DeleteQueue: public kj::AtomicRefcounted { // and deletions are dropped. Useful for reclaiming registrations that target this queue. bool isDefunct() const; + // The number of actions queued. For tests. + size_t getQueuedActionCountForTest() 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. See scheduleAction() for when they run. // // In particular, these actions are used to implement cross-context promise resolution. - // - // 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; kj::Maybe>> crossThreadFulfiller; }; @@ -145,8 +156,17 @@ class DeleteQueue: public kj::AtomicRefcounted { template SpecificOwnedObject* addObjectImpl(kj::Own obj, OwnedObjectList& ownedObjects) const; + // Installs a fresh cross-thread signal and returns a promise that resolves once an action is + // scheduled, or never once the queue is defunct. Scheduling requires an installed signal, so the + // owner arms one up front and re-arms after each wake; the previous one must have fired. kj::Promise resetCrossThreadSignal() const; + // Moves the queued actions out of the queue, so they can run without its lock held. + kj::Array> takeActions() const; + + // True if any actions are queued. + bool hasQueuedActions() const; + friend class IoContext; }; @@ -158,8 +178,8 @@ class IoCrossContextExecutor { public: IoCrossContextExecutor(kj::Arc deleteQueue): deleteQueue(kj::mv(deleteQueue)) {} - // Tries to execute the specified action to the owning IoContext. - // The target IoContext will be signaled to run the action as soon as it is able. + // Schedules the specified action to run in the target IoContext. See + // DeleteQueue::scheduleAction() for when it runs. void execute(jsg::Lock& js, kj::Function&& action); // True if the IoContext this executor targets is the calling thread's current IoContext — @@ -175,6 +195,9 @@ class IoCrossContextExecutor { // True if the target IoContext has been destroyed, i.e. tryExecute() would drop the action. bool isTargetDestroyed() const; + // The number of actions queued for the target IoContext. For tests. + size_t getQueuedActionCountForTest() const; + private: friend class IoContext; friend class DeleteQueue; @@ -221,21 +244,7 @@ class DeleteQueuePtr { public: DeleteQueuePtr(kj::Arc queue): queue(kj::mv(queue)) {} KJ_DISALLOW_COPY_AND_MOVE(DeleteQueuePtr); - ~DeleteQueuePtr() noexcept(false) { - auto ptr = queue.get(); - if (ptr != nullptr) { - auto lock = ptr->crossThreadDeleteQueue.lockExclusive(); - KJ_IF_SOME(state, *lock) { - // The delete queue state may include a kj::CrossThreadPromiseFulfiller that - // needs to be destroyed. To do so, we need to allow async destructors here. - // We only want to destroy the crossThreadFulfiller in this scope tho, not - // everything that may be in the queue. - kj::AllowAsyncDestructorsScope scope; - state.crossThreadFulfiller = kj::none; - } - *lock = kj::none; - } - } + ~DeleteQueuePtr() noexcept(false); kj::Arc queue; }; diff --git a/src/workerd/io/worker-entrypoint-test.c++ b/src/workerd/io/worker-entrypoint-test.c++ index 7afa2daaf29..773bf13f743 100644 --- a/src/workerd/io/worker-entrypoint-test.c++ +++ b/src/workerd/io/worker-entrypoint-test.c++ @@ -796,6 +796,68 @@ KJ_TEST("alarm scheduling tags failures after delivery") { KJ_EXPECT(e.getDetail(WORKER_REQUEST_DELIVERED_DETAIL_ID) != kj::none); } +KJ_TEST("an alarm that fails to schedule drains its request") { + RejectingTimerChannel timer; + TestFixture fixture(TestFixture::SetupParams{ + .autogateKeys = kj::arr(util::AutogateKey::DEFER_CROSS_CONTEXT_ACTIONS), + .actorId = Worker::Actor::Id(kj::str("alarm-drain-test")), + .useRealTimers = false, + .ioChannelFactory = kj::Function(TimerChannel&)>( + [&timer](TimerChannel&) -> kj::Rc { + return kj::rc(timer); + }), + }); + auto& waitScope = fixture.getWaitScope(); + auto& actor = fixture.getActor(); + auto context = fixture.newIoContext(); + actor.setIoContext(kj::addRef(*context)); + + // Queued on the idle actor, so delivering the alarm starts a run of it as a waitUntil task of the + // alarm's request. Holding the input lock holds the run up until the alarm has failed. + uint ran = 0; + KJ_ASSERT(context->getCrossContextExecutor().tryExecute([&ran](jsg::Lock&) { ++ran; })); + fixture.pollEventLoop(); + auto inputLock = actor.getInputGate().wait(nullptr).wait(waitScope); + auto entrypoint = fixture.makeWorkerEntrypoint(); + KJ_EXPECT_THROW_MESSAGE("alarm scheduling failed", + entrypoint->runAlarm(kj::UNIX_EPOCH + kj::SECONDS, 0).wait(waitScope)); + KJ_EXPECT(ran == 0, ran); + + // The request is still draining, so it is current when the run gets the lock. + { auto released = kj::mv(inputLock); } + fixture.pollEventLoop(); + KJ_EXPECT(ran == 1, ran); + KJ_EXPECT(context->getAbortReason() == kj::none, context->getAbortReason()); +} + +KJ_TEST("an alarm canceled while it waits for its scheduled time drains its request") { + TestFixture fixture({.autogateKeys = kj::arr(util::AutogateKey::DEFER_CROSS_CONTEXT_ACTIONS), + .actorId = Worker::Actor::Id(kj::str("alarm-cancel-drain-test"))}); + auto& waitScope = fixture.getWaitScope(); + auto& actor = fixture.getActor(); + auto context = fixture.newIoContext(); + actor.setIoContext(kj::addRef(*context)); + + // Queued on the idle actor, so delivering the alarm starts a run of it as a waitUntil task of the + // alarm's request. Holding the input lock holds the run up until the alarm is canceled. + uint ran = 0; + KJ_ASSERT(context->getCrossContextExecutor().tryExecute([&ran](jsg::Lock&) { ++ran; })); + fixture.pollEventLoop(); + auto inputLock = actor.getInputGate().wait(nullptr).wait(waitScope); + auto entrypoint = fixture.makeWorkerEntrypoint(); + // The fixture's timers never fire, so the alarm waits for its scheduled time until canceled. + auto alarm = entrypoint->runAlarm(kj::UNIX_EPOCH + kj::SECONDS, 0); + KJ_EXPECT(!alarm.poll(waitScope)); + alarm = nullptr; + KJ_EXPECT(ran == 0, ran); + + // The request is still draining, so it is current when the run gets the lock. + { auto released = kj::mv(inputLock); } + fixture.pollEventLoop(); + KJ_EXPECT(ran == 1, ran); + KJ_EXPECT(context->getAbortReason() == kj::none, context->getAbortReason()); +} + KJ_TEST("custom events tag failures after delivery") { TestFixture fixture; auto entrypoint = fixture.makeWorkerEntrypoint(); @@ -906,6 +968,103 @@ KJ_TEST("hibernatable WebSocket clean close is reported as a stream disconnect") observer->outcome.orDefault(EventOutcome::OK) == EventOutcome::RESPONSE_STREAM_DISCONNECTED); } +// Like HIBERNATABLE_WEB_SOCKET_SOURCE, but fetch() also creates a promise for another context to +// settle with settle(). Its reaction uses the actor's storage, which needs a current request. +// webSocketMessage() records the message and replies "ack". +constexpr kj::StringPtr PROMISE_WEB_SOCKET_SOURCE = R"SCRIPT( + import { DurableObject } from "cloudflare:workers"; + globalThis.events = ""; + export default class extends DurableObject { + fetch() { + const [client, server] = Object.values(new WebSocketPair()); + this.ctx.acceptWebSocket(server); + const { promise, resolve } = Promise.withResolvers(); + globalThis.settle = resolve; + promise.then(async (value) => { + globalThis.events += "reaction;"; + await this.ctx.storage.put("key", value); + globalThis.events += `stored:${await this.ctx.storage.get("key")};`; + }); + return new Response(null, { status: 101, webSocket: client }); + } + webSocketMessage(ws, message) { + globalThis.events += `message:${message};`; + ws.send("ack"); + } + webSocketClose(ws, code, reason) { + ws.close(code, reason); + } + } +)SCRIPT"_kj; + +KJ_TEST("a promise settled from another context while an actor idles with a hibernatable " + "WebSocket settles in its next WebSocket event") { + capnp::MallocMessageBuilder flagsMessage; + auto flags = flagsMessage.initRoot(); + flags.setHandleCrossRequestPromiseResolution(true); + auto loopback = kj::refcounted(); + TestFixture fixture(TestFixture::SetupParams{ + .featureFlags = flags.asReader(), + .autogateKeys = kj::arr(util::AutogateKey::DEFER_CROSS_CONTEXT_ACTIONS), + .mainModuleSource = PROMISE_WEB_SOCKET_SOURCE, + .actorId = Worker::Actor::Id(kj::str("idle-web-socket-promise-test")), + .actorClassName = "default"_kj, + .useRealTimers = false, + .actorLoopback = kj::addRef(*loopback), + }); + loopback->fixture = fixture; + auto& waitScope = fixture.getWaitScope(); + auto& actor = fixture.getActor(); + actor.setHibernationManager( + kj::refcounted(actor.getLoopback(), /*eventType=*/0)); + + auto entrypoint = fixture.makeWorkerEntrypoint(); + kj::HttpHeaderTable headerTable; + kj::HttpHeaders headers(headerTable); + headers.setPtr(kj::HttpHeaderId::UPGRADE, "websocket"); + kj::NullStream requestBody; + auto pipe = kj::newTwoWayPipe(); + TestResponse response(kj::newWebSocket(kj::mv(pipe.ends[0]), kj::none)); + auto request = + entrypoint->request( + kj::HttpMethod::GET, "https://example.com", headers, requestBody, response) + .eagerlyEvaluate(nullptr); + auto peer = kj::newWebSocket(kj::mv(pipe.ends[1]), kj::none); + fixture.pollEventLoop(); + KJ_ASSERT(response.statusCode == 101); + auto& context = KJ_ASSERT_NONNULL(actor.getIoContext()); + KJ_ASSERT(!context.hasCurrentIncomingRequest()); + + // A stateless request settles the promise, queueing its reaction on the idle actor. + auto statelessContext = fixture.newStatelessIoContext(); + auto settler = fixture.newIncomingRequest(*statelessContext); + fixture.enterContext(*settler, [](TestFixture::Environment& env) { + auto settle = + KJ_ASSERT_NONNULL(env.js.global().get(env.js, "settle"_kj).tryCast()); + settle.call(env.js, env.js.undefined(), env.js.str("value"_kj)); + }); + fixture.drainAndDestroy(kj::mv(settler)); + fixture.pollEventLoop(); + KJ_EXPECT(getJsEvents(fixture) == "", getJsEvents(fixture)); + KJ_EXPECT(context.getAbortReason() == kj::none, context.getAbortReason()); + + // The next WebSocket event runs the reaction before its handler. + peer->send("hi"_kj).wait(waitScope); + auto reply = peer->receive().wait(waitScope); + KJ_EXPECT(reply.is() && reply.get() == "ack"); + fixture.pollEventLoop(); + KJ_EXPECT(getJsEvents(fixture) == "reaction;stored:value;message:hi;", getJsEvents(fixture)); + KJ_EXPECT(context.getAbortReason() == kj::none, context.getAbortReason()); + + // Closing the WebSocket ends the upgrade request. Its outcome is tested above. + peer->close(1000, "done").wait(waitScope); + auto close = peer->receive().wait(waitScope); + KJ_ASSERT(close.is()); + KJ_IF_SOME(e, kj::runCatchingExceptions([&]() { request.wait(waitScope); })) { + KJ_EXPECT(e.getType() == kj::Exception::Type::DISCONNECTED, e); + } +} + KJ_TEST("local WebSocket disconnect is not a script exception") { auto pipe = kj::newTwoWayPipe(); LocalWebSocketRequest test(kj::newWebSocket(kj::mv(pipe.ends[0]), kj::none)); diff --git a/src/workerd/io/worker-entrypoint.c++ b/src/workerd/io/worker-entrypoint.c++ index 03796be5c9e..06ec0362e83 100644 --- a/src/workerd/io/worker-entrypoint.c++ +++ b/src/workerd/io/worker-entrypoint.c++ @@ -13,6 +13,7 @@ #include #include #include +#include #include #include #include @@ -1009,6 +1010,16 @@ kj::Promise WorkerEntrypoint::runAlarmImpl( incomingRequest->delivered("alarm"_kj); + KJ_DEFER({ + // With the DEFER_CROSS_CONTEXT_ACTIONS autogate, delivered() can start a waitUntil task to run + // the cross-context actions queued while the actor was idle. Drain the request even if the + // alarm is canceled or fails, so the task can finish. The AlarmFulfiller case drains it itself. + if (incomingRequest.get() != nullptr && + util::Autogate::isEnabled(util::AutogateKey::DEFER_CROSS_CONTEXT_ACTIONS)) { + incomingRequest->drain(waitUntilTasks, kj::mv(incomingRequest)); + } + }); + auto scheduleAlarmResult = co_await kj::evalNow([&]() { return actor.scheduleAlarm(scheduledTime); }).catch_([](kj::Exception&& exception) -> kj::Promise { diff --git a/src/workerd/io/worker.c++ b/src/workerd/io/worker.c++ index ef6da727737..906ab0fed82 100644 --- a/src/workerd/io/worker.c++ +++ b/src/workerd/io/worker.c++ @@ -1410,16 +1410,10 @@ Worker::Isolate::Isolate(kj::Own apiParam, // refcounted pointer to the DeleteQueue for the correct isolate. // We simply pass the given callback, reactions, and argument to // a function that will be added to the queue inside DeleteQueue. - // The next time the relevant IoContext is entered, this queue will - // be drained and the actions will be run. Adding the task to the - // delete queue will also signal the IoContext that it should wake - // up and drain the queue. Simple, eh? - // - // A word of warning tho! It is possible for the IoContext to be - // destroyed before the promise is resolved. Any actions that have - // already been added to the queue would end up being dropped silently - // on the floor. Actions that are added to the queue now will be run - // immediately in the wrong IoContext. + // Adding the task to the delete queue also signals the IoContext + // that it should wake up and run it. See DeleteQueue::scheduleAction() + // for when it runs, and for what happens if the IoContext is + // destroyed first. auto& ref = jsg::unwrapOpaqueRef>(isolate, tag); ref->execute(js, [reactions = jsg::Data(isolate, reactions), argument = jsg::V8Ref(isolate, argument), @@ -4916,7 +4910,10 @@ kj::Promise Worker::Actor::runOnHibernateImpl(HibernationReason reason, auto incomingRequest = kj::heap(kj::addRef(context), kj::mv(ioChannelFactory), kj::mv(observer), kj::mv(workerTracer), kj::none /* maybeTriggerInvocationSpan */); - incomingRequest->delivered("onHibernate"_kj); + // Cross-context actions queued while the actor was idle are dropped rather than run with the + // handler: they would use up its CPU budget, and the cleanup below would cancel whatever they + // start. Those queued once its request is current still run, since the handler may await them. + incomingRequest->delivered("onHibernate"_kj, DropQueuedCrossContextActions::YES); context.getLimitEnforcer().limitOnHibernateCpu(); { diff --git a/src/workerd/tests/test-fixture.c++ b/src/workerd/tests/test-fixture.c++ index b85ba9bbaa7..920904cba6e 100644 --- a/src/workerd/tests/test-fixture.c++ +++ b/src/workerd/tests/test-fixture.c++ @@ -267,15 +267,19 @@ inline server::config::Worker::Reader buildConfig( modules[0].setName(mainModuleName); modules[0].setEsModule(params.mainModuleSource.orDefault(mainModuleSource)); - // Initialize autogates (with an empty config unless the test supplied gate names). + // Initialize autogates (with an empty config unless the test supplied gates). // // This needs to happen here because `buildConfig` is called early in the construction of // `TestFixture`. - KJ_IF_SOME(gates, params.autogates) { - util::Autogate::initAutogateNamesForTest(gates); - } else { - util::Autogate::initAutogate({}); + kj::Vector gates; + KJ_IF_SOME(names, params.autogates) { + gates.addAll(names); + } + for (auto key: params.autogateKeys) { + gates.add(kj::toCharSequence(key)); } + util::Autogate::initAutogateNamesForTest( + gates.asPtr(), util::IgnoreAllAutogatesEnv(params.ignoreAllAutogatesEnv)); return config; } @@ -506,7 +510,15 @@ void TestFixture::runInIoContext(kj::Function(const Environmen } kj::Own TestFixture::newIoContext() { - return kj::refcounted(threadContext, kj::atomicAddRef(*worker), actor, + return newIoContextFor(actor); +} + +kj::Own TestFixture::newStatelessIoContext() { + return newIoContextFor(kj::none); +} + +kj::Own TestFixture::newIoContextFor(kj::Maybe maybeActor) { + return kj::refcounted(threadContext, kj::atomicAddRef(*worker), maybeActor, kj::heap(checkedSubrequestCount, actorCallReplayMemoryReserver)); } diff --git a/src/workerd/tests/test-fixture.h b/src/workerd/tests/test-fixture.h index 3a022131959..93fe766fac9 100644 --- a/src/workerd/tests/test-fixture.h +++ b/src/workerd/tests/test-fixture.h @@ -10,6 +10,7 @@ #include #include #include +#include #include #include @@ -38,6 +39,11 @@ struct TestFixture { // If set, the named autogates (bare names, without the "workerd-autogate-" prefix) // are enabled for the test. Otherwise autogates are initialized empty. kj::Maybe> autogates; + // Autogates enabled for the test, like those named in `autogates`. + kj::Array autogateKeys; + // If true, only the autogates above are enabled, even under WORKERD_ALL_AUTOGATES (as in the + // @all-autogates test variant), e.g. to test the behavior with a gate disabled. + bool ignoreAllAutogatesEnv; kj::Maybe mainModuleSource; // If set, make a stub of an Actor with the given id. kj::Maybe actorId; @@ -159,6 +165,10 @@ struct TestFixture { // outlives a single IncomingRequest, e.g. to model an actor receiving multiple requests. kj::Own newIoContext(); + // Like newIoContext(), but the IoContext is never bound to the fixture's actor, e.g. to model a + // stateless request in the same isolate as an actor. + kj::Own newStatelessIoContext(); + // Create a new IncomingRequest bound to a fresh IoContext. The returned IncomingRequest is the // sole owner of that IoContext (via kj refcounting): destroying the IR destroys the IoContext. // If you need the IoContext to outlive the IR (e.g. to model multiple sequential or overlapping @@ -198,6 +208,13 @@ struct TestFixture { worker->runInLockScope(asyncLock, [&](Worker::Lock& lock) { callback(lock); }); } + // Takes this fixture's isolate lock without an IoContext, and returns it held. A thread takes + // one isolate lock at a time, so holding it holds up runs in another fixture's isolate on the + // same thread (and event loop), e.g. to test what happens while they wait for the lock. + Worker::AsyncLock takeAsyncLock() { + return worker->takeAsyncLockWithoutRequest(nullptr).wait(getWaitScope()); + } + // Acquire a Worker::Lock synchronously without an IoContext. template void enterWorkerLockSynchronously(Callback&& callback) { @@ -324,6 +341,8 @@ struct TestFixture { // Construct a fresh Worker::Actor with the given id, using the saved Loopback. kj::Own makeActor(Worker::Actor::Id id); + kj::Own newIoContextFor(kj::Maybe maybeActor); + public: // Default IoChannelFactory used by tests. Exposed so tests can subclass it // and override individual methods (e.g. startSubrequest for socket connect tests). diff --git a/src/workerd/util/autogate.h b/src/workerd/util/autogate.h index 3f50d069a69..b8d6a984c9a 100644 --- a/src/workerd/util/autogate.h +++ b/src/workerd/util/autogate.h @@ -127,7 +127,13 @@ namespace workerd::util { arguments do, so that a fraction just past a range limit (2 ** 31 - 0.5 as int, -0.5 as \ uint32_t) converts. When disabled, the range check runs on the untruncated value and such a \ fraction is a TypeError. Applies to C++ JSG and Rust JSG alike. */ \ - V(JSG_INTEGER_ENFORCE_RANGE) + V(JSG_INTEGER_ENFORCE_RANGE) \ + /* Cross-context actions queued on an actor between events wait for its next request instead \ + of aborting it, and IoContext::run() without a current request fails with a typed error. \ + An alarm's request also drains when its alarm is canceled or fails to schedule, and an \ + actor's onHibernate event drops the actions queued before it. Also counts the copies of an \ + abort's reason its queued deliveries share as external memory. */ \ + V(DEFER_CROSS_CONTEXT_ACTIONS) // clang-format on // --------------------------------------------------------------------------------------