From 9ac634f000b2944a1f5e0080acc5df6aeb770ab8 Mon Sep 17 00:00:00 2001 From: Joaquim Silva Date: Wed, 14 Oct 2026 18:54:12 +0100 Subject: [PATCH 1/2] Fail runs without a current IncomingRequest with a typed exception An actor's IoContext outlives its requests, so IoContext::run() can be called while the actor is idle, or lose its request while it waits for a lock. run() then failed with an untyped internal error. Now run() checks for a request when called and after taking each lock. Without one, it skips the callback and fails with a JavaScript Error carrying NO_CURRENT_INCOMING_REQUEST_DETAIL_ID, so callers can tell this case apart. JavaScript sees its message instead of "internal error". An aborted context fails with its abort exception instead, as run() already does when it has a request. To keep a trail for the cases that are bugs, a run called without a request logs a warning, which reaches Sentry, at most once an hour. A run that loses its request while waiting for a lock doesn't log: the request can end during any wait, so that is no sign of a bug. This is behind a new autogate, DEFER_CROSS_CONTEXT_ACTIONS, named for the deferral of cross-context actions that the next commit builds on this. With it off, run() fails as before. The delete queue signal task still aborts the IoContext on this error; the next commit fixes that. For the tests, TestFixture::SetupParams gains autogateKeys, which enables autogates by key rather than by name, and TestFixture gains takeAsyncLock(), which holds up runs in another fixture's isolate on the same thread. --- src/workerd/io/incoming-request-test.c++ | 111 +++++++++++++++++++++++ src/workerd/io/io-context.c++ | 21 +++++ src/workerd/io/io-context.h | 41 +++++++-- src/workerd/tests/test-fixture.c++ | 13 ++- src/workerd/tests/test-fixture.h | 10 ++ src/workerd/util/autogate.h | 4 +- 6 files changed, 187 insertions(+), 13 deletions(-) diff --git a/src/workerd/io/incoming-request-test.c++ b/src/workerd/io/incoming-request-test.c++ index 6df2862fb91..759e5ce182e 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 @@ -768,5 +769,115 @@ 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); +} + } // namespace } // namespace workerd diff --git a/src/workerd/io/io-context.c++ b/src/workerd/io/io-context.c++ index 4c93dcb7a6a..c71d96f39d4 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 @@ -656,6 +657,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; diff --git a/src/workerd/io/io-context.h b/src/workerd/io/io-context.h index 136e61c6aac..fa23e3cf965 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,8 @@ 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); [[noreturn]] void throwExceededMemoryLimit(bool isActor); @@ -525,6 +532,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) { @@ -1419,10 +1430,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( @@ -1549,11 +1570,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)); @@ -1562,12 +1585,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) { @@ -1576,7 +1601,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); }); } @@ -1588,10 +1614,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/tests/test-fixture.c++ b/src/workerd/tests/test-fixture.c++ index b85ba9bbaa7..e96fba5248d 100644 --- a/src/workerd/tests/test-fixture.c++ +++ b/src/workerd/tests/test-fixture.c++ @@ -267,15 +267,18 @@ 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()); return config; } diff --git a/src/workerd/tests/test-fixture.h b/src/workerd/tests/test-fixture.h index 3a022131959..a6d92cb0a2f 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,8 @@ 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; kj::Maybe mainModuleSource; // If set, make a stub of an Actor with the given id. kj::Maybe actorId; @@ -198,6 +201,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) { diff --git a/src/workerd/util/autogate.h b/src/workerd/util/autogate.h index 250aea3f014..4f079a82413 100644 --- a/src/workerd/util/autogate.h +++ b/src/workerd/util/autogate.h @@ -136,7 +136,9 @@ 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) \ + /* IoContext::run() without a current request fails with a typed error. */ \ + V(DEFER_CROSS_CONTEXT_ACTIONS) // clang-format on // -------------------------------------------------------------------------------------- From f2b8d4cf33628974da37ae5f09fd484756cee935 Mon Sep 17 00:00:00 2001 From: Joaquim Silva Date: Thu, 15 Oct 2026 11:48:37 +0100 Subject: [PATCH 2/2] Defer cross-context actions on idle actors instead of aborting Other IoContexts can queue actions on an IoContext, e.g. to settle one of its promises or deliver an AbortSignal's abort. Running them needs a current IncomingRequest. An idle actor, such as one holding hibernatable WebSockets, has none, so the run failed and aborted the IoContext, breaking the actor for all its clients. Now actions queued while an actor has no request wait for its next request, which runs them as a waitUntil task, counting toward its limits. A run that loses its request while waiting for a lock also leaves them queued. Any other failure, including an action's own, still aborts the IoContext. Actions still queued when the IoContext ends are dropped, with a NOSENTRY warning. Actions queued while an actor has a request run as before, on the context's next turn, but the run is now also a waitUntil task of the request, so that the request can't finish draining and strand them until the actor's next request. The delete queue's signal is re-armed before each run, and the run takes the actions out of the queue before running them, so that an action can queue another. An alarm request now drains even if its alarm is canceled or fails to schedule, as other events' requests do. Destroyed undrained, it would log that drain() wasn't called, and its waitUntil task could leave the actions queued for a later request. The onHibernate event drops the actions queued before it, with a similar warning, rather than run them: they would use up its handler's CPU budget, and the cleanup after the handler would cancel whatever they start. Actions queued once its request is current still run, since the handler may await them. The isolate's memory limit bounds the queue, since actions keep what they act on alive on its heap. Two kinds of action escaped it: - A request to arm the RPC subscription of an AbortSignal received over RPC holds only a weak reference. Queue at most one per signal. - An abort copied its reason for each queued delivery and RPC clone, without counting the copies. Now the native deliveries share one copy and the RPC clones another, and both count as external memory. The arm request also now checks its weak reference with tryAddRef(), since a garbage collection may condemn the signal before it runs. handle_cross_request_promise_resolution only guarantees which IoContext a promise settles in, not when, so deferring keeps its contract. The old behavior broke the actor, so we accept the risk of changing it without a compatibility flag. The DEFER_CROSS_CONTEXT_ACTIONS autogate from the previous commit also covers the changes to the delete queue's signal task, alarm requests and the onHibernate event, and the external memory accounting of the shared copies. With it off, they behave as before. The one-per-signal limit, the sharing itself and tryAddRef() apply either way: without the autogate, an arm request runs on the context's next turn or the context is aborted, so a second one would arm nothing; sharing an immutable copy changes no behavior; and tryAddRef() keeps the arm request from using a condemned signal. A new wd-test covers promises that an actor event creates and another IoContext settles while the actor has a request, which behave the same with or without the autogate. For the tests, TestFixture::SetupParams gains ignoreAllAutogatesEnv, to test behavior with an autogate off in the @all-autogates variant, and TestFixture gains newStatelessIoContext(), for a context that isn't bound to the fixture's actor. --- docs/reference/detail/abort-signal.md | 32 +- src/workerd/api/BUILD.bazel | 9 + src/workerd/api/abort-signal-test.c++ | 391 ++++++++++++++ src/workerd/api/basics.c++ | 91 +++- src/workerd/api/basics.h | 13 +- src/workerd/api/tests/BUILD.bazel | 5 + .../tests/actor-cross-context-promise-test.js | 127 +++++ .../actor-cross-context-promise-test.wd-test | 26 + src/workerd/io/AGENTS.md | 1 + src/workerd/io/BUILD.bazel | 1 + src/workerd/io/incoming-request-test.c++ | 90 +++- src/workerd/io/io-context.c++ | 110 +++- src/workerd/io/io-context.h | 20 +- src/workerd/io/io-own-test.c++ | 498 ++++++++++++++++++ src/workerd/io/io-own.c++ | 55 ++ src/workerd/io/io-own.h | 61 ++- src/workerd/io/worker-entrypoint-test.c++ | 159 ++++++ src/workerd/io/worker-entrypoint.c++ | 11 + src/workerd/io/worker.c++ | 19 +- src/workerd/tests/test-fixture.c++ | 13 +- src/workerd/tests/test-fixture.h | 9 + src/workerd/util/autogate.h | 6 +- 22 files changed, 1642 insertions(+), 105 deletions(-) create mode 100644 src/workerd/api/abort-signal-test.c++ create mode 100644 src/workerd/api/tests/actor-cross-context-promise-test.js create mode 100644 src/workerd/api/tests/actor-cross-context-promise-test.wd-test 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 495f6103f94..acfbca7cc3b 100644 --- a/src/workerd/io/AGENTS.md +++ b/src/workerd/io/AGENTS.md @@ -44,6 +44,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 759e5ce182e..da8daf7086f 100644 --- a/src/workerd/io/incoming-request-test.c++ +++ b/src/workerd/io/incoming-request-test.c++ @@ -530,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)); @@ -709,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") { @@ -879,5 +877,75 @@ KJ_TEST("run() in a critical section fails with the abort reason of an aborted c 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 c71d96f39d4..b28659cc6cd 100644 --- a/src/workerd/io/io-context.c++ +++ b/src/workerd/io/io-context.c++ @@ -342,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(); @@ -396,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) { @@ -2023,25 +2031,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 fa23e3cf965..31f137c0c5b 100644 --- a/src/workerd/io/io-context.h +++ b/src/workerd/io/io-context.h @@ -67,6 +67,9 @@ WD_STRONG_BOOL(SpanAwaitedNext); 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); @@ -149,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. @@ -1535,6 +1544,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 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 e96fba5248d..920904cba6e 100644 --- a/src/workerd/tests/test-fixture.c++ +++ b/src/workerd/tests/test-fixture.c++ @@ -278,7 +278,8 @@ inline server::config::Worker::Reader buildConfig( for (auto key: params.autogateKeys) { gates.add(kj::toCharSequence(key)); } - util::Autogate::initAutogateNamesForTest(gates.asPtr()); + util::Autogate::initAutogateNamesForTest( + gates.asPtr(), util::IgnoreAllAutogatesEnv(params.ignoreAllAutogatesEnv)); return config; } @@ -509,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 a6d92cb0a2f..93fe766fac9 100644 --- a/src/workerd/tests/test-fixture.h +++ b/src/workerd/tests/test-fixture.h @@ -41,6 +41,9 @@ struct TestFixture { 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; @@ -162,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 @@ -334,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 4f079a82413..c00fcba12e7 100644 --- a/src/workerd/util/autogate.h +++ b/src/workerd/util/autogate.h @@ -137,7 +137,11 @@ namespace workerd::util { 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) \ - /* IoContext::run() without a current request fails with a typed error. */ \ + /* 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 // --------------------------------------------------------------------------------------