diff --git a/src/workerd/api/BUILD.bazel b/src/workerd/api/BUILD.bazel index bbd222d02d7..44228fca1e2 100644 --- a/src/workerd/api/BUILD.bazel +++ b/src/workerd/api/BUILD.bazel @@ -756,6 +756,15 @@ kj_test( deps = ["//src/workerd/tests:test-fixture"], ) +kj_test( + src = "actor-fetch-retry-test.c++", + deps = [ + "//src/workerd/io", + "//src/workerd/io:worker-interface", + "//src/workerd/tests:test-fixture", + ], +) + kj_test( src = "fetch-body-rewindable-test.c++", deps = [ diff --git a/src/workerd/api/actor-fetch-retry-test.c++ b/src/workerd/api/actor-fetch-retry-test.c++ new file mode 100644 index 00000000000..bfdc51b1639 --- /dev/null +++ b/src/workerd/api/actor-fetch-retry-test.c++ @@ -0,0 +1,741 @@ +// Copyright (c) 2017-2022 Cloudflare, Inc. +// Licensed under the Apache 2.0 license found in the LICENSE file or at: +// https://opensource.org/licenses/Apache-2.0 + +#include "actor.h" +#include "global-scope.h" + +#include +#include +#include +#include + +#include + +namespace workerd::api { +namespace { + +// Minimal WorkerInterface that answers every outgoing request() with an empty 200, draining the +// request body first so a streaming sender doesn't block on backpressure. +class MockFetchTarget final: public WorkerInterface { + public: + kj::Promise request(kj::HttpMethod method, + kj::StringPtr url, + const kj::HttpHeaders& headers, + kj::AsyncInputStream& requestBody, + kj::HttpService::Response& response) override { + co_await requestBody.readAllBytes(); + // Build the response headers on the same HttpHeaderTable as the request headers; the runtime + // reads the response with its own registered header IDs, so a fresh table would mismatch. + auto responseHeaders = headers.cloneShallow(); + responseHeaders.clear(); + response.send(200, "OK"_kj, responseHeaders, static_cast(0)); + } + + kj::Promise connect(kj::StringPtr host, + const kj::HttpHeaders& headers, + kj::AsyncIoStream& connection, + ConnectResponse& response, + kj::HttpConnectSettings settings) override { + KJ_UNIMPLEMENTED("not used in this test"); + } + kj::Promise prewarm(kj::StringPtr url) override { + KJ_UNIMPLEMENTED("not used in this test"); + } + kj::Promise runScheduled(kj::Date scheduledTime, kj::StringPtr cron) override { + KJ_UNIMPLEMENTED("not used in this test"); + } + kj::Promise runAlarm(kj::Date scheduledTime, uint32_t retryCount) override { + KJ_UNIMPLEMENTED("not used in this test"); + } + kj::Promise customEvent(kj::Own event) override { + return event->notSupported(); + } +}; + +class TestStreamSource final: public ReadableStreamSource { + public: + kj::Promise tryRead(void*, size_t, size_t) override { + return static_cast(0); + } +}; + +class RetryMetadataOutgoingFactory final: public Fetcher::OutgoingFactory { + public: + RetryMetadataOutgoingFactory(bool& ordinaryDispatchCalled, + kj::Maybe& capturedMetadata) + : ordinaryDispatchCalled(ordinaryDispatchCalled), + capturedMetadata(capturedMetadata) {} + + kj::Own newSingleUseClient(kj::Maybe) override { + ordinaryDispatchCalled = true; + return kj::heap(); + } + + bool supportsActorFetchRetries() const override { + return true; + } + + kj::Own newSingleUseClientWithActorRetryMetadata(kj::Maybe, + kj::Maybe actorRetryRequestMetadata) override { + capturedMetadata = kj::mv(actorRetryRequestMetadata); + return kj::heap(); + } + + private: + bool& ordinaryDispatchCalled; + kj::Maybe& capturedMetadata; +}; + +enum class ReplayFailure { + AMBIGUOUS, + NOT_DELIVERED, + DELIVERED, + CLAIM_REJECTED, + SLOW_RESPONSE, + RETRY_DELAY_EXCEEDS_BUDGET, +}; + +class DeterministicTimerChannel final: public TimerChannel { + public: + explicit DeterministicTimerChannel(kj::TimerImpl& timer): timer(timer) {} + + void syncTime() override {} + + kj::Date now(kj::Maybe) override { + return kj::UNIX_EPOCH + (timer.now() - kj::origin()); + } + + kj::Promise atTime(kj::Date when) override { + auto target = kj::origin() + (when - kj::UNIX_EPOCH); + if (target <= timer.now()) { + return kj::READY_NOW; + } + auto promise = timer.atTime(target); + timer.advanceTo(target); + return promise; + } + + kj::Promise afterLimitTimeout(kj::Duration delay) override { + auto promise = timer.afterDelay(delay); + timer.advanceTo(timer.now() + nextDelay.orDefault(delay)); + nextDelay = kj::none; + return promise; + } + + void delayNextTimeoutBy(kj::Duration delay) { + nextDelay = delay; + } + + kj::TimePoint nowForLimitTimeout() override { + return timer.now(); + } + + private: + kj::TimerImpl& timer; + kj::Maybe nextDelay; +}; + +struct ReplayState { + kj::Array failures; + bool acceptWebSocket = false; + kj::Maybe> acceptedWebSocket; + kj::Vector metadata; + kj::Vector> requestBodies; + uint requestCount = 0; + uint webSocketRequestCount = 0; + uint retryCount = 0; + kj::Maybe timerChannel; +}; + +class ReplayFetchTarget final: public WorkerInterface { + public: + ReplayFetchTarget(ReplayState& state): state(state) {} + + kj::Promise request(kj::HttpMethod method, + kj::StringPtr url, + const kj::HttpHeaders& headers, + kj::AsyncInputStream& requestBody, + kj::HttpService::Response& response) override { + auto attempt = state.requestCount++; + kj::Maybe> slowResponseDelay; + if (attempt < state.failures.size() && + state.failures[attempt] == ReplayFailure::SLOW_RESPONSE) { + slowResponseDelay = IoContext::current().afterLimitTimeout(11 * kj::SECONDS); + } + if (headers.isWebSocket()) { + ++state.webSocketRequestCount; + } + state.requestBodies.add(co_await requestBody.readAllBytes()); + if (attempt < state.failures.size()) { + auto failure = state.failures[attempt]; + if (failure == ReplayFailure::SLOW_RESPONSE) { + co_await kj::mv(KJ_ASSERT_NONNULL(slowResponseDelay)); + } else { + if (failure == ReplayFailure::RETRY_DELAY_EXCEEDS_BUDGET) { + KJ_REQUIRE_NONNULL(state.timerChannel).delayNextTimeoutBy(11 * kj::SECONDS); + } + auto exception = failure == ReplayFailure::CLAIM_REJECTED + ? KJ_EXCEPTION(FAILED, "actor retry claim rejected") + : KJ_EXCEPTION(DISCONNECTED, "actor fetch disconnected"); + if (failure == ReplayFailure::NOT_DELIVERED) { + exception.setDetail( + jsg::REQUEST_NOT_DELIVERED_TO_ACTOR_DETAIL_ID, kj::heapArray(0)); + } else if (failure == ReplayFailure::DELIVERED) { + exception.setDetail( + jsg::REQUEST_DELIVERED_TO_ACTOR_DETAIL_ID, kj::heapArray(0)); + } else if (failure == ReplayFailure::CLAIM_REJECTED) { + exception.setDetail( + jsg::ACTOR_RETRY_CLAIM_REJECTED_DETAIL_ID, kj::heapArray(0)); + } + kj::throwRecoverableException(kj::mv(exception)); + } + } + + auto responseHeaders = headers.cloneShallow(); + responseHeaders.clear(); + if (headers.isWebSocket() && state.acceptWebSocket) { + state.acceptedWebSocket = response.acceptWebSocket(responseHeaders); + co_return; + } + response.send(200, "OK"_kj, responseHeaders, static_cast(0)); + } + + kj::Promise connect(kj::StringPtr host, + const kj::HttpHeaders& headers, + kj::AsyncIoStream& connection, + ConnectResponse& response, + kj::HttpConnectSettings settings) override { + KJ_UNIMPLEMENTED("not used in this test"); + } + kj::Promise prewarm(kj::StringPtr url) override { + KJ_UNIMPLEMENTED("not used in this test"); + } + kj::Promise runScheduled(kj::Date scheduledTime, kj::StringPtr cron) override { + KJ_UNIMPLEMENTED("not used in this test"); + } + kj::Promise runAlarm(kj::Date scheduledTime, uint32_t retryCount) override { + KJ_UNIMPLEMENTED("not used in this test"); + } + kj::Promise customEvent(kj::Own event) override { + return event->notSupported(); + } + + private: + ReplayState& state; +}; + +class ReplayOutgoingFactory final: public Fetcher::OutgoingFactory { + public: + ReplayOutgoingFactory(ReplayState& state): state(state) {} + + kj::Own newSingleUseClient(kj::Maybe) override { + KJ_FAIL_ASSERT("replay tests should always supply actor retry metadata"); + } + + bool supportsActorFetchRetries() const override { + return true; + } + + void onActorFetchRetry() override { + ++state.retryCount; + } + + kj::Own newSingleUseClientWithActorRetryMetadata(kj::Maybe, + kj::Maybe actorRetryRequestMetadata) override { + state.metadata.add(KJ_REQUIRE_NONNULL(actorRetryRequestMetadata)); + return kj::heap(state); + } + + private: + ReplayState& state; +}; + +enum class RetryEnforcement { + DISABLED, + ENABLED, +}; + +enum class ActorFetchKind { + HTTP, + WEB_SOCKET, +}; + +kj::Maybe runActorFetch(ReplayState& state, + RetryEnforcement enforcement, + kj::Maybe body, + ActorFetchKind kind) { + kj::TimerImpl timer(kj::origin()); + DeterministicTimerChannel timerChannel(timer); + state.timerChannel = timerChannel; + TestFixture fixture(TestFixture::SetupParams{ + .useRealTimers = false, + .ioChannelFactory = kj::Function(TimerChannel&)>( + [&](TimerChannel&) -> kj::Rc { + return kj::rc(timerChannel); + }), + }); + if (enforcement == RetryEnforcement::ENABLED) { + util::Autogate::initAutogateNamesForTest( + {"durable-object-retries-fetch"_kj, "durable-object-retries-fetch-retry-requests"_kj}, + util::IgnoreAllAutogatesEnv::YES); + } else { + util::Autogate::initAutogateNamesForTest( + {"durable-object-retries-fetch"_kj}, util::IgnoreAllAutogatesEnv::YES); + } + kj::Maybe failure; + + fixture.runInIoContext([&](const TestFixture::Environment& env) { + auto fetcher = env.js.alloc( + env.context.addObject(kj::heap(state)), + Fetcher::RequiresHostAndProtocol::YES); + RequestInitializerDict init; + KJ_IF_SOME(value, body) { + init.method = kj::str("POST"); + init.body = kj::Maybe(kj::str(value)); + } + if (kind == ActorFetchKind::WEB_SOCKET) { + jsg::Dict headers; + headers.fields = kj::heapArray::Field>(1); + headers.fields[0].name = kj::str("Upgrade"); + headers.fields[0].value = kj::str("websocket"); + init.headers = kj::mv(headers); + } + auto promise = fetcher->fetch(env.js, kj::str("http://example.com"), kj::mv(init)); + return env.context.awaitJs(env.js, kj::mv(promise)) + .ignoreResult() + .catch_([&](kj::Exception&& exception) { + failure.emplace(kj::mv(exception)); + }).attach(kj::mv(fetcher)); + }); + + return failure; +} + +class UnsupportedOutgoingFactory final: public Fetcher::OutgoingFactory { + public: + UnsupportedOutgoingFactory(bool& called): called(called) {} + + kj::Own newSingleUseClient(kj::Maybe) override { + called = true; + return kj::heap(); + } + + private: + bool& called; +}; + +class MockActorId final: public ActorIdFactory::ActorId { + public: + kj::String toString() const override { + return kj::str("actor-id"); + } + + kj::Maybe getName() const override { + return kj::none; + } + + kj::Maybe getJurisdiction() const override { + return kj::none; + } + + bool equals(const ActorId& other) const override { + return other.toString() == "actor-id"; + } + + kj::Own clone() const override { + return kj::heap(); + } +}; + +class RecordingActorChannel final: public IoChannelFactory::ActorChannel { + public: + RecordingActorChannel(kj::Maybe& capturedMetadata) + : capturedMetadata(capturedMetadata) {} + + kj::Own startRequest(IoChannelFactory::SubrequestMetadata metadata) override { + capturedMetadata = kj::mv(metadata.actorRetryRequestMetadata); + return kj::heap(); + } + + void requireAllowsTransfer() override { + KJ_UNIMPLEMENTED("not used in this test"); + } + + kj::OneOf, kj::Promise>> getTokenMaybeSync( + IoChannelFactory::ChannelTokenUsage) override { + KJ_UNIMPLEMENTED("not used in this test"); + } + + private: + kj::Maybe& capturedMetadata; +}; + +class ReplayActorChannel final: public IoChannelFactory::ActorChannel { + public: + ReplayActorChannel(ReplayState& state): state(state) {} + + kj::Own startRequest(IoChannelFactory::SubrequestMetadata metadata) override { + KJ_IF_SOME(retryMetadata, metadata.actorRetryRequestMetadata) { + state.metadata.add(kj::mv(retryMetadata)); + } + return kj::heap(state); + } + + void requireAllowsTransfer() override { + KJ_UNIMPLEMENTED("not used in this test"); + } + + kj::OneOf, kj::Promise>> getTokenMaybeSync( + IoChannelFactory::ChannelTokenUsage) override { + KJ_UNIMPLEMENTED("not used in this test"); + } + + private: + ReplayState& state; +}; + +struct ActorIoChannelFactory final: public TestFixture::DummyIoChannelFactory { + ActorIoChannelFactory(TimerChannel& timer, + kj::Maybe& capturedMetadata, + uint& channelCount, + kj::Vector& locationHints, + kj::Vector& cohorts) + : DummyIoChannelFactory(timer), + capturedMetadata(capturedMetadata), + channelCount(channelCount), + locationHints(locationHints), + cohorts(cohorts) {} + + kj::Own getGlobalActor(uint, + const ActorIdFactory::ActorId&, + kj::Maybe locationHint, + ActorGetMode, + bool, + ActorRoutingMode, + SpanParent, + kj::Maybe version, + Persistent) override { + ++channelCount; + KJ_IF_SOME(hint, locationHint) { + locationHints.add(kj::mv(hint)); + } + KJ_IF_SOME(v, version) { + KJ_IF_SOME(cohort, v.cohort) { + cohorts.add(kj::mv(cohort)); + } + } + return kj::refcounted(capturedMetadata); + } + + kj::Maybe& capturedMetadata; + uint& channelCount; + kj::Vector& locationHints; + kj::Vector& cohorts; +}; + +KJ_TEST("fetch generates actor retry metadata for a supported outgoing factory") { + bool ordinaryDispatchCalled = false; + kj::Maybe capturedMetadata; + kj::Date beforeFetch = kj::UNIX_EPOCH; + kj::Date afterFetch = kj::UNIX_EPOCH; + TestFixture fixture; + + fixture.runInIoContext([&](const TestFixture::Environment& env) { + auto fetcher = env.js.alloc( + env.context.addObject( + kj::heap(ordinaryDispatchCalled, capturedMetadata)), + Fetcher::RequiresHostAndProtocol::YES); + beforeFetch = kj::systemCoarseCalendarClock().now(); + auto promise = fetcher->fetch(env.js, kj::str("http://example.com"), kj::none); + afterFetch = kj::systemCoarseCalendarClock().now(); + return env.context.awaitJs(env.js, kj::mv(promise)).ignoreResult().attach(kj::mv(fetcher)); + }); + + KJ_EXPECT(!ordinaryDispatchCalled); + KJ_IF_SOME(metadata, capturedMetadata) { + KJ_EXPECT(metadata.createdAt >= beforeFetch); + KJ_EXPECT(metadata.createdAt <= afterFetch); + KJ_EXPECT(metadata.isRetry == IsActorRetry::NO); + } else { + KJ_FAIL_EXPECT("supported fetch did not generate actor retry metadata"); + } +} + +KJ_TEST("fetch omits actor retry metadata for a supported factory with a streaming body") { + bool ordinaryDispatchCalled = false; + kj::Maybe capturedMetadata; + TestFixture fixture; + + fixture.runInIoContext([&](const TestFixture::Environment& env) { + auto fetcher = env.js.alloc( + env.context.addObject( + kj::heap(ordinaryDispatchCalled, capturedMetadata)), + Fetcher::RequiresHostAndProtocol::YES); + RequestInitializerDict init; + init.method = kj::str("POST"); + init.body = kj::Maybe( + JsReadableStream::create(env.js, env.context, kj::heap())); + auto promise = fetcher->fetch(env.js, kj::str("http://example.com"), kj::mv(init)); + return env.context.awaitJs(env.js, kj::mv(promise)).ignoreResult().attach(kj::mv(fetcher)); + }); + + KJ_EXPECT(ordinaryDispatchCalled); + KJ_EXPECT(capturedMetadata == kj::none); +} + +KJ_TEST("fetch omits actor retry metadata for an unsupported outgoing factory") { + bool ordinaryDispatchCalled = false; + TestFixture fixture; + + fixture.runInIoContext([&](const TestFixture::Environment& env) { + auto fetcher = + env.js.alloc(env.context.addObject( + kj::heap(ordinaryDispatchCalled)), + Fetcher::RequiresHostAndProtocol::YES); + auto promise = fetcher->fetch(env.js, kj::str("http://example.com"), kj::none); + return env.context.awaitJs(env.js, kj::mv(promise)).ignoreResult().attach(kj::mv(fetcher)); + }); + + KJ_EXPECT(ordinaryDispatchCalled); +} + +KJ_TEST("actor fetch updates retry metadata and rewinds the body") { + ReplayState state{.failures = kj::arr(ReplayFailure::NOT_DELIVERED, ReplayFailure::AMBIGUOUS, + ReplayFailure::NOT_DELIVERED)}; + KJ_EXPECT(runActorFetch(state, RetryEnforcement::ENABLED, "request body"_kj, + ActorFetchKind::HTTP) == kj::none); + + KJ_ASSERT(state.metadata.size() == 4); + KJ_EXPECT(state.requestCount == 4); + KJ_EXPECT(state.retryCount == 3); + KJ_EXPECT(state.metadata[0].nonce != state.metadata[1].nonce); + KJ_EXPECT(state.metadata[1].nonce == state.metadata[2].nonce); + KJ_EXPECT(state.metadata[1].nonce == state.metadata[3].nonce); + KJ_EXPECT(state.metadata[0].isRetry == IsActorRetry::NO); + KJ_EXPECT(state.metadata[1].isRetry == IsActorRetry::NO); + KJ_EXPECT(state.metadata[2].isRetry == IsActorRetry::YES); + KJ_EXPECT(state.metadata[3].isRetry == IsActorRetry::YES); + KJ_ASSERT(state.requestBodies.size() == 4); + for (auto& body: state.requestBodies) { + KJ_EXPECT(body == "request body"_kj.asBytes()); + } +} + +KJ_TEST("actor fetch does not retry when enforcement is disabled") { + ReplayState state{.failures = kj::arr(ReplayFailure::AMBIGUOUS)}; + + KJ_EXPECT( + runActorFetch(state, RetryEnforcement::DISABLED, kj::none, ActorFetchKind::HTTP) != kj::none); + KJ_EXPECT(state.requestCount == 1); + KJ_EXPECT(state.retryCount == 0); +} + +KJ_TEST("actor fetch does not retry a delivered disconnect") { + ReplayState state{.failures = kj::arr(ReplayFailure::DELIVERED)}; + + KJ_EXPECT( + runActorFetch(state, RetryEnforcement::ENABLED, kj::none, ActorFetchKind::HTTP) != kj::none); + KJ_EXPECT(state.requestCount == 1); + KJ_EXPECT(state.retryCount == 0); +} + +KJ_TEST("actor fetch does not report not-delivered after an ambiguous retry") { + ReplayState state{ + .failures = kj::arr(ReplayFailure::NOT_DELIVERED, ReplayFailure::AMBIGUOUS, + ReplayFailure::NOT_DELIVERED, ReplayFailure::NOT_DELIVERED, ReplayFailure::AMBIGUOUS)}; + + auto failure = KJ_REQUIRE_NONNULL( + runActorFetch(state, RetryEnforcement::ENABLED, kj::none, ActorFetchKind::HTTP)); + + KJ_EXPECT(failure.getDetail(jsg::REQUEST_NOT_DELIVERED_TO_ACTOR_DETAIL_ID) == kj::none, failure); + KJ_EXPECT(state.requestCount == 5); +} + +KJ_TEST("actor fetch stops after a retry claim rejection") { + ReplayState state{ + .failures = kj::arr(ReplayFailure::AMBIGUOUS, ReplayFailure::CLAIM_REJECTED), + }; + auto failure = KJ_REQUIRE_NONNULL( + runActorFetch(state, RetryEnforcement::ENABLED, kj::none, ActorFetchKind::HTTP)); + + KJ_EXPECT(failure.getType() == kj::Exception::Type::DISCONNECTED, failure); + KJ_EXPECT(!failure.getDescription().contains("claim rejected"), failure); + KJ_EXPECT(state.requestCount == 2); + KJ_EXPECT(state.retryCount == 1); +} + +KJ_TEST("actor fetch normalizes an initial retry claim rejection") { + ReplayState state{.failures = kj::arr(ReplayFailure::CLAIM_REJECTED)}; + auto failure = KJ_REQUIRE_NONNULL( + runActorFetch(state, RetryEnforcement::ENABLED, kj::none, ActorFetchKind::HTTP)); + + KJ_EXPECT(failure.getType() == kj::Exception::Type::DISCONNECTED, failure); + KJ_EXPECT(!failure.getDescription().contains("claim rejected"), failure); + KJ_EXPECT(state.requestCount == 1); + KJ_EXPECT(state.retryCount == 0); +} + +KJ_TEST("actor WebSocket fetch retries a disconnected handshake") { + ReplayState state{ + .failures = kj::arr(ReplayFailure::AMBIGUOUS), + .acceptWebSocket = true, + }; + + KJ_EXPECT(runActorFetch(state, RetryEnforcement::ENABLED, kj::none, ActorFetchKind::WEB_SOCKET) == + kj::none); + KJ_EXPECT(state.requestCount == 2); + KJ_EXPECT(state.webSocketRequestCount == 2); + KJ_EXPECT(state.retryCount == 1); + KJ_EXPECT(state.acceptedWebSocket != kj::none); +} + +KJ_TEST("actor fetch honors an abort before retrying") { + ReplayState state{.failures = kj::arr(ReplayFailure::AMBIGUOUS)}; + kj::Maybe failure; + TestFixture fixture(TestFixture::SetupParams{ + .autogates = kj::arr( + "durable-object-retries-fetch"_kj, "durable-object-retries-fetch-retry-requests"_kj), + .useRealTimers = true, + }); + + fixture.runInIoContext([&](const TestFixture::Environment& env) { + auto fetcher = env.js.alloc( + env.context.addObject(kj::heap(state)), + Fetcher::RequiresHostAndProtocol::YES); + auto controller = AbortController::constructor(env.js); + RequestInitializerDict init; + init.signal = kj::Maybe(controller->getSignal()); + auto promise = fetcher->fetch(env.js, kj::str("http://example.com"), kj::mv(init)); + controller->abort(env.js, kj::none); + return env.context.awaitJs(env.js, kj::mv(promise)) + .ignoreResult() + .catch_([&](kj::Exception&& exception) { + failure.emplace(kj::mv(exception)); + }).attach(kj::mv(fetcher), kj::mv(controller)); + }); + + auto& exception = KJ_REQUIRE_NONNULL(failure); + KJ_EXPECT(exception.getDescription().contains("The operation was aborted"), exception); + KJ_EXPECT(state.requestCount == 1); + KJ_EXPECT(state.retryCount == 0); +} + +KJ_TEST("actor fetch stops after five attempts") { + ReplayState state{ + .failures = kj::arr(ReplayFailure::AMBIGUOUS, ReplayFailure::AMBIGUOUS, + ReplayFailure::AMBIGUOUS, ReplayFailure::AMBIGUOUS, ReplayFailure::AMBIGUOUS), + }; + + KJ_EXPECT( + runActorFetch(state, RetryEnforcement::ENABLED, kj::none, ActorFetchKind::HTTP) != kj::none); + KJ_EXPECT(state.requestCount == 5); + KJ_EXPECT(state.retryCount == 4); +} + +KJ_TEST("actor fetch allows an in-flight retry to finish after the start budget") { + ReplayState state{ + .failures = kj::arr(ReplayFailure::AMBIGUOUS, ReplayFailure::SLOW_RESPONSE), + }; + + KJ_EXPECT( + runActorFetch(state, RetryEnforcement::ENABLED, kj::none, ActorFetchKind::HTTP) == kj::none); + KJ_EXPECT(state.requestCount == 2); + KJ_EXPECT(state.retryCount == 1); +} + +KJ_TEST("actor fetch does not start a retry after the start budget") { + ReplayState state{ + .failures = kj::arr(ReplayFailure::RETRY_DELAY_EXCEEDS_BUDGET), + }; + + KJ_EXPECT( + runActorFetch(state, RetryEnforcement::ENABLED, kj::none, ActorFetchKind::HTTP) != kj::none); + KJ_EXPECT(state.requestCount == 1); + KJ_EXPECT(state.retryCount == 1); +} + +KJ_TEST("replica actor fetch does not retry a disconnected primary channel") { + ReplayState state{.failures = kj::arr(ReplayFailure::NOT_DELIVERED)}; + kj::Maybe failure; + TestFixture fixture(TestFixture::SetupParams{ + .autogates = kj::arr( + "durable-object-retries-fetch"_kj, "durable-object-retries-fetch-retry-requests"_kj), + .useRealTimers = false, + }); + + fixture.runInIoContext([&](const TestFixture::Environment& env) { + auto fetcher = env.js.alloc( + env.context.addObject(kj::heap( + kj::refcounted(state), kj::str("actor-id"))), + Fetcher::RequiresHostAndProtocol::YES); + auto promise = fetcher->fetch(env.js, kj::str("http://example.com"), kj::none); + return env.context.awaitJs(env.js, kj::mv(promise)) + .ignoreResult() + .catch_([&](kj::Exception&& exception) { + failure.emplace(kj::mv(exception)); + }).attach(kj::mv(fetcher)); + }); + + KJ_EXPECT(failure != kj::none); + KJ_EXPECT(state.requestCount == 1); + KJ_EXPECT(state.metadata.empty()); +} + +// Global actor factories must place the caller's metadata on the actor subrequest. +KJ_TEST("GlobalActorOutgoingFactory places actor retry metadata on the actor subrequest") { + kj::Maybe capturedMetadata; + uint channelCount = 0; + kj::Vector locationHints; + kj::Vector cohorts; + TestFixture fixture(TestFixture::SetupParams{ + .useRealTimers = false, + .ioChannelFactory = kj::Function(TimerChannel&)>( + [&](TimerChannel& timer) -> kj::Rc { + return kj::rc( + timer, capturedMetadata, channelCount, locationHints, cohorts); + }), + }); + + fixture.runInIoContext([&](const TestFixture::Environment& env) { + GlobalActorOutgoingFactory factory( + GlobalActorOutgoingFactory::ChannelIdOrFactory(static_cast(1)), + env.js.alloc(kj::heap()), kj::str("location"), + ActorGetMode::GET_OR_CREATE, false, ActorRoutingMode::DEFAULT, + ActorVersion{.cohort = kj::str("cohort")}, Persistent::NO); + KJ_EXPECT(factory.supportsActorFetchRetries()); + + auto client = factory.newSingleUseClientWithActorRetryMetadata(kj::none, + IoChannelFactory::ActorRetryRequestMetadata{ + .nonce = 0x123456789abcdef0, + .createdAt = kj::UNIX_EPOCH + 123 * kj::MILLISECONDS, + .isRetry = IsActorRetry::YES, + }); + + KJ_IF_SOME(metadata, capturedMetadata) { + KJ_EXPECT(metadata.nonce == 0x123456789abcdef0); + KJ_EXPECT(metadata.createdAt == kj::UNIX_EPOCH + 123 * kj::MILLISECONDS); + KJ_EXPECT(metadata.isRetry == IsActorRetry::YES); + } else { + KJ_FAIL_EXPECT("actor retry metadata was not forwarded to the actor channel"); + } + + factory.onActorFetchRetry(); + auto retryClient = factory.newSingleUseClientWithActorRetryMetadata(kj::none, + IoChannelFactory::ActorRetryRequestMetadata{ + .nonce = 0xfedcba9876543210, + .createdAt = kj::UNIX_EPOCH + 456 * kj::MILLISECONDS, + .isRetry = IsActorRetry::YES, + }); + KJ_EXPECT(channelCount == 2); + KJ_ASSERT(locationHints.size() == 2); + KJ_EXPECT(locationHints[0] == "location"); + KJ_EXPECT(locationHints[1] == "location"); + KJ_ASSERT(cohorts.size() == 2); + KJ_EXPECT(cohorts[0] == "cohort"); + KJ_EXPECT(cohorts[1] == "cohort"); + }); +} + +} // namespace +} // namespace workerd::api diff --git a/src/workerd/api/actor.c++ b/src/workerd/api/actor.c++ index 80d7e3d306b..0db5a961a23 100644 --- a/src/workerd/api/actor.c++ +++ b/src/workerd/api/actor.c++ @@ -68,6 +68,8 @@ kj::Own LocalActorOutgoingFactory::getSubre IoChannelFactory::ActorChannel& GlobalActorOutgoingFactory::getOrCreateActorChannel( IoContext& context, SpanParent parentSpan) { if (actorChannel == kj::none) { + auto locationHint = this->locationHint.map([](kj::String& hint) { return kj::str(hint); }); + auto version = this->version.map([](ActorVersion& version) { return version.clone(); }); KJ_SWITCH_ONEOF(channelIdOrFactory) { KJ_CASE_ONEOF(channelId, uint) { actorChannel = @@ -91,6 +93,11 @@ IoChannelFactory::ActorChannel& GlobalActorOutgoingFactory::getOrCreateActorChan return *KJ_REQUIRE_NONNULL(actorChannel); } +void GlobalActorOutgoingFactory::onActorFetchRetry() { + actorChannel = kj::none; + channelMemoryAdjustment = kj::none; +} + kj::Own GlobalActorOutgoingFactory::newSingleUseClient( kj::Maybe cfStr) { return newSingleUseClientWithActorRetryMetadata(kj::mv(cfStr), kj::none); @@ -123,14 +130,10 @@ kj::Own GlobalActorOutgoingFactory::getSubr kj::Own ReplicaActorOutgoingFactory::newSingleUseClient( kj::Maybe cfStr) { - return newSingleUseClientWithActorRetryMetadata(kj::mv(cfStr), kj::none); -} - -kj::Own ReplicaActorOutgoingFactory::newSingleUseClientWithActorRetryMetadata( - kj::Maybe cfStr, - kj::Maybe actorRetryRequestMetadata) { auto& context = IoContext::current(); + // Replica-to-primary stubs hold a pre-resolved Pipeline and cannot reroute after it breaks, so + // they do not opt into actor fetch retries. return context.getMetrics().wrapActorSubrequestClient(context.getSubrequest( [&](TraceContext& tracing, IoChannelFactory& ioChannelFactory) { tracing.setTag("objectId"_kjc, actorId.asPtr()); @@ -139,8 +142,7 @@ kj::Own ReplicaActorOutgoingFactory::newSingleUseClientWithActo // already open prior to this DO starting up. return actorChannel->startRequest({.cfBlobJson = kj::mv(cfStr), .parentSpan = tracing.getInternalSpanParent(), - .userSpanParent = tracing.getUserSpanParent(), - .actorRetryRequestMetadata = kj::mv(actorRetryRequestMetadata)}); + .userSpanParent = tracing.getUserSpanParent()}); }, {.inHouse = true, .wrapMetrics = true, diff --git a/src/workerd/api/actor.h b/src/workerd/api/actor.h index a71bfc657f7..cc7eaa02504 100644 --- a/src/workerd/api/actor.h +++ b/src/workerd/api/actor.h @@ -347,9 +347,10 @@ class GlobalActorOutgoingFactory final: public Fetcher::OutgoingFactory { persistent(persistent) {} kj::Own newSingleUseClient(kj::Maybe cfStr) override; - bool supportsActorRetryMetadata() const override { + bool supportsActorFetchRetries() const override { return true; } + void onActorFetchRetry() override; kj::Own newSingleUseClientWithActorRetryMetadata(kj::Maybe cfStr, kj::Maybe actorRetryRequestMetadata) override; kj::Own getSubrequestChannel() override; @@ -411,11 +412,6 @@ class ReplicaActorOutgoingFactory final: public Fetcher::OutgoingFactory { actorId(kj::mv(actorId)) {} kj::Own newSingleUseClient(kj::Maybe cfStr) override; - bool supportsActorRetryMetadata() const override { - return true; - } - kj::Own newSingleUseClientWithActorRetryMetadata(kj::Maybe cfStr, - kj::Maybe actorRetryRequestMetadata) override; kj::Own getSubrequestChannel() override; private: diff --git a/src/workerd/api/fetch-body-rewindable-test.c++ b/src/workerd/api/fetch-body-rewindable-test.c++ index 837302cda50..77feccbd809 100644 --- a/src/workerd/api/fetch-body-rewindable-test.c++ +++ b/src/workerd/api/fetch-body-rewindable-test.c++ @@ -2,7 +2,6 @@ // Licensed under the Apache 2.0 license found in the LICENSE file or at: // https://opensource.org/licenses/Apache-2.0 -#include "actor.h" #include "global-scope.h" #include @@ -66,99 +65,6 @@ class MockFetchTarget final: public WorkerInterface { } }; -class TestStreamSource final: public ReadableStreamSource { - public: - kj::Promise tryRead(void*, size_t, size_t) override { - return static_cast(0); - } -}; - -class RetryMetadataOutgoingFactory final: public Fetcher::OutgoingFactory { - public: - RetryMetadataOutgoingFactory(bool& ordinaryDispatchCalled, - kj::Maybe& capturedMetadata) - : ordinaryDispatchCalled(ordinaryDispatchCalled), - capturedMetadata(capturedMetadata) {} - - kj::Own newSingleUseClient(kj::Maybe) override { - ordinaryDispatchCalled = true; - return kj::heap(); - } - - bool supportsActorRetryMetadata() const override { - return true; - } - - kj::Own newSingleUseClientWithActorRetryMetadata(kj::Maybe, - kj::Maybe actorRetryRequestMetadata) override { - capturedMetadata = kj::mv(actorRetryRequestMetadata); - return kj::heap(); - } - - private: - bool& ordinaryDispatchCalled; - kj::Maybe& capturedMetadata; -}; - -class UnsupportedOutgoingFactory final: public Fetcher::OutgoingFactory { - public: - UnsupportedOutgoingFactory(bool& called): called(called) {} - - kj::Own newSingleUseClient(kj::Maybe) override { - called = true; - return kj::heap(); - } - - private: - bool& called; -}; - -class MockActorId final: public ActorIdFactory::ActorId { - public: - kj::String toString() const override { - return kj::str("actor-id"); - } - - kj::Maybe getName() const override { - return kj::none; - } - - kj::Maybe getJurisdiction() const override { - return kj::none; - } - - bool equals(const ActorId& other) const override { - return other.toString() == "actor-id"; - } - - kj::Own clone() const override { - return kj::heap(); - } -}; - -class RecordingActorChannel final: public IoChannelFactory::ActorChannel { - public: - RecordingActorChannel(kj::Maybe& capturedMetadata) - : capturedMetadata(capturedMetadata) {} - - kj::Own startRequest(IoChannelFactory::SubrequestMetadata metadata) override { - capturedMetadata = kj::mv(metadata.actorRetryRequestMetadata); - return kj::heap(); - } - - void requireAllowsTransfer() override { - KJ_UNIMPLEMENTED("not used in this test"); - } - - kj::OneOf, kj::Promise>> getTokenMaybeSync( - IoChannelFactory::ChannelTokenUsage) override { - KJ_UNIMPLEMENTED("not used in this test"); - } - - private: - kj::Maybe& capturedMetadata; -}; - struct FetchTargetIoChannelFactory final: public TestFixture::DummyIoChannelFactory { FetchTargetIoChannelFactory(TimerChannel& timer): DummyIoChannelFactory(timer) {} @@ -167,27 +73,6 @@ struct FetchTargetIoChannelFactory final: public TestFixture::DummyIoChannelFact } }; -struct ActorIoChannelFactory final: public TestFixture::DummyIoChannelFactory { - ActorIoChannelFactory( - TimerChannel& timer, kj::Maybe& capturedMetadata) - : DummyIoChannelFactory(timer), - capturedMetadata(capturedMetadata) {} - - kj::Own getGlobalActor(uint, - const ActorIdFactory::ActorId&, - kj::Maybe, - ActorGetMode, - bool, - ActorRoutingMode, - SpanParent, - kj::Maybe, - Persistent) override { - return kj::refcounted(capturedMetadata); - } - - kj::Maybe& capturedMetadata; -}; - // fetchImplNoOutputLock forwards Request::canRewindBody() to RequestObserver so that, downstream, // edgeworker can classify retry eligibility for disconnected outgoing actor calls. The subtle // property here is that the stashed signal is per-call, not sticky: a single RequestObserver is @@ -210,7 +95,6 @@ KJ_TEST("fetch reports each outgoing body's rewindability per-call without stale // non-rewindable body. await fetch("http://example.com/stream", { method: "POST", body: request.body, duplex: "half" }); - return new Response("OK"); }, }; @@ -236,204 +120,5 @@ KJ_TEST("fetch reports each outgoing body's rewindability per-call without stale "streamed request body should not be rewindable (no carryover)"); } -KJ_TEST("fetch generates actor retry metadata for a supported outgoing factory") { - bool ordinaryDispatchCalled = false; - kj::Maybe capturedMetadata; - kj::Date beforeFetch = kj::UNIX_EPOCH; - kj::Date afterFetch = kj::UNIX_EPOCH; - TestFixture fixture; - - fixture.runInIoContext([&](const TestFixture::Environment& env) { - auto fetcher = env.js.alloc( - env.context.addObject( - kj::heap(ordinaryDispatchCalled, capturedMetadata)), - Fetcher::RequiresHostAndProtocol::YES); - beforeFetch = kj::systemCoarseCalendarClock().now(); - auto promise = fetcher->fetch(env.js, kj::str("http://example.com"), kj::none); - afterFetch = kj::systemCoarseCalendarClock().now(); - return env.context.awaitJs(env.js, kj::mv(promise)).ignoreResult().attach(kj::mv(fetcher)); - }); - - KJ_EXPECT(!ordinaryDispatchCalled); - KJ_IF_SOME(metadata, capturedMetadata) { - KJ_EXPECT(metadata.createdAt >= beforeFetch); - KJ_EXPECT(metadata.createdAt <= afterFetch); - KJ_EXPECT(metadata.isRetry == IsActorRetry::NO); - } else { - KJ_FAIL_EXPECT("supported fetch did not generate actor retry metadata"); - } -} - -KJ_TEST("fetch omits actor retry metadata for a supported factory with a streaming body") { - bool ordinaryDispatchCalled = false; - kj::Maybe capturedMetadata; - TestFixture fixture; - - fixture.runInIoContext([&](const TestFixture::Environment& env) { - auto fetcher = env.js.alloc( - env.context.addObject( - kj::heap(ordinaryDispatchCalled, capturedMetadata)), - Fetcher::RequiresHostAndProtocol::YES); - RequestInitializerDict init; - init.method = kj::str("POST"); - init.body = kj::Maybe( - JsReadableStream::create(env.js, env.context, kj::heap())); - auto promise = fetcher->fetch(env.js, kj::str("http://example.com"), kj::mv(init)); - return env.context.awaitJs(env.js, kj::mv(promise)).ignoreResult().attach(kj::mv(fetcher)); - }); - - KJ_EXPECT(ordinaryDispatchCalled); - KJ_EXPECT(capturedMetadata == kj::none); -} - -KJ_TEST("fetch omits actor retry metadata for an unsupported outgoing factory") { - bool ordinaryDispatchCalled = false; - TestFixture fixture; - - fixture.runInIoContext([&](const TestFixture::Environment& env) { - auto fetcher = - env.js.alloc(env.context.addObject( - kj::heap(ordinaryDispatchCalled)), - Fetcher::RequiresHostAndProtocol::YES); - auto promise = fetcher->fetch(env.js, kj::str("http://example.com"), kj::none); - return env.context.awaitJs(env.js, kj::mv(promise)).ignoreResult().attach(kj::mv(fetcher)); - }); - - KJ_EXPECT(ordinaryDispatchCalled); -} - -// Isolate Fetcher's dispatch decision from actor routing: a metadata-aware factory must receive the -// caller's logical-call metadata unchanged. -KJ_TEST("Fetcher dispatches actor retry metadata through the metadata-aware factory hook") { - bool ordinaryDispatchCalled = false; - kj::Maybe capturedMetadata; - TestFixture fixture; - - fixture.runInIoContext([&](const TestFixture::Environment& env) { - auto fetcher = env.js.alloc( - env.context.addObject( - kj::heap(ordinaryDispatchCalled, capturedMetadata)), - Fetcher::RequiresHostAndProtocol::YES); - - auto client = fetcher->getClientWithTracing(env.context, kj::none, "fetch"_kjc, - IoChannelFactory::ActorRetryRequestMetadata{ - .nonce = 0x123456789abcdef0, - .createdAt = kj::UNIX_EPOCH + 123 * kj::MILLISECONDS, - .isRetry = IsActorRetry::YES, - }); - - KJ_IF_SOME(metadata, capturedMetadata) { - KJ_EXPECT(metadata.nonce == 0x123456789abcdef0); - KJ_EXPECT(metadata.createdAt == kj::UNIX_EPOCH + 123 * kj::MILLISECONDS); - KJ_EXPECT(metadata.isRetry == IsActorRetry::YES); - } else { - KJ_FAIL_EXPECT("actor retry metadata was not forwarded"); - } - }); -} - -// Never fall back to ordinary dispatch when a factory cannot carry retry metadata. Doing so would -// silently replace the caller's logical-call token at a later seam. -KJ_TEST("Fetcher rejects actor retry metadata instead of using ordinary factory dispatch") { - bool called = false; - TestFixture fixture; - - fixture.runInIoContext([&](const TestFixture::Environment& env) { - auto fetcher = env.js.alloc(env.context.addObject( - kj::heap(called)), - Fetcher::RequiresHostAndProtocol::YES); - - KJ_EXPECT_THROW_MESSAGE("actor retry metadata supplied to an unsupported Fetcher", - fetcher->getClientWithTracing(env.context, kj::none, "fetch"_kjc, - IoChannelFactory::ActorRetryRequestMetadata{ - .nonce = 1, - .createdAt = kj::UNIX_EPOCH, - .isRetry = IsActorRetry::NO, - })); - KJ_EXPECT(!called); - }); -} - -// A numeric subrequest channel has no metadata-aware factory hook, so reject before dispatch rather -// than silently dropping the logical-call metadata. -KJ_TEST("Fetcher rejects actor retry metadata before numeric subrequest-channel dispatch") { - TestFixture fixture; - - fixture.runInIoContext([&](const TestFixture::Environment& env) { - auto fetcher = - env.js.alloc(static_cast(1), Fetcher::RequiresHostAndProtocol::YES); - - KJ_EXPECT_THROW_MESSAGE("actor retry metadata supplied to an unsupported Fetcher", - fetcher->getClientWithTracing(env.context, kj::none, "fetch"_kjc, - IoChannelFactory::ActorRetryRequestMetadata{ - .nonce = 1, - .createdAt = kj::UNIX_EPOCH, - .isRetry = IsActorRetry::NO, - })); - }); -} - -// Exercise the real implementation of the hook tested above: global actor factories must place the -// caller's metadata on the subrequest sent through the actor channel. -KJ_TEST("GlobalActorOutgoingFactory places actor retry metadata on the actor subrequest") { - kj::Maybe capturedMetadata; - TestFixture fixture(TestFixture::SetupParams{ - .useRealTimers = false, - .ioChannelFactory = kj::Function(TimerChannel&)>( - [&](TimerChannel& timer) -> kj::Rc { - return kj::rc(timer, capturedMetadata); - }), - }); - - fixture.runInIoContext([&](const TestFixture::Environment& env) { - GlobalActorOutgoingFactory factory( - GlobalActorOutgoingFactory::ChannelIdOrFactory(static_cast(1)), - env.js.alloc(kj::heap()), kj::none, - ActorGetMode::GET_OR_CREATE, false, ActorRoutingMode::DEFAULT, kj::none, Persistent::NO); - KJ_EXPECT(factory.supportsActorRetryMetadata()); - - auto client = factory.newSingleUseClientWithActorRetryMetadata(kj::none, - IoChannelFactory::ActorRetryRequestMetadata{ - .nonce = 0x123456789abcdef0, - .createdAt = kj::UNIX_EPOCH + 123 * kj::MILLISECONDS, - .isRetry = IsActorRetry::YES, - }); - - KJ_IF_SOME(metadata, capturedMetadata) { - KJ_EXPECT(metadata.nonce == 0x123456789abcdef0); - KJ_EXPECT(metadata.createdAt == kj::UNIX_EPOCH + 123 * kj::MILLISECONDS); - KJ_EXPECT(metadata.isRetry == IsActorRetry::YES); - } else { - KJ_FAIL_EXPECT("actor retry metadata was not forwarded to the actor channel"); - } - }); -} - -KJ_TEST("ReplicaActorOutgoingFactory places actor retry metadata on the actor subrequest") { - kj::Maybe capturedMetadata; - TestFixture fixture; - - fixture.runInIoContext([&](const TestFixture::Environment& env) { - ReplicaActorOutgoingFactory factory( - kj::refcounted(capturedMetadata), kj::str("actor-id")); - KJ_EXPECT(factory.supportsActorRetryMetadata()); - - auto client = factory.newSingleUseClientWithActorRetryMetadata(kj::none, - IoChannelFactory::ActorRetryRequestMetadata{ - .nonce = 0x123456789abcdef0, - .createdAt = kj::UNIX_EPOCH + 123 * kj::MILLISECONDS, - .isRetry = IsActorRetry::YES, - }); - - KJ_IF_SOME(metadata, capturedMetadata) { - KJ_EXPECT(metadata.nonce == 0x123456789abcdef0); - KJ_EXPECT(metadata.createdAt == kj::UNIX_EPOCH + 123 * kj::MILLISECONDS); - KJ_EXPECT(metadata.isRetry == IsActorRetry::YES); - } else { - KJ_FAIL_EXPECT("actor retry metadata was not forwarded to the actor channel"); - } - }); -} - } // namespace } // namespace workerd::api diff --git a/src/workerd/api/http.c++ b/src/workerd/api/http.c++ index 6623ae584a9..097d7e681f7 100644 --- a/src/workerd/api/http.c++ +++ b/src/workerd/api/http.c++ @@ -19,6 +19,7 @@ #include #include #include +#include #include #include #include @@ -33,6 +34,8 @@ #include #include +#include + namespace workerd::api { namespace { @@ -1468,6 +1471,140 @@ namespace { // Fetch spec requires (suggests?) 20: https://fetch.spec.whatwg.org/#http-redirect-fetch constexpr auto MAX_REDIRECT_COUNT = 20; +class ActorFetchRetryState { + public: + static ActorFetchRetryState create(TimerChannel& timer); + + IoChannelFactory::ActorRetryRequestMetadata getMetadata() const { + return metadata; + } + + bool isRetryEnabled() const { + return deadline != kj::none; + } + + kj::Maybe checkDeadline(); + kj::OneOf prepareRetry(kj::Exception exception); + + private: + static constexpr uint MAX_ATTEMPTS = 5; + static constexpr auto RETRY_BUDGET = 10 * kj::SECONDS; + static constexpr auto INITIAL_BACKOFF = 50 * kj::MILLISECONDS; + + ActorFetchRetryState(IoChannelFactory::ActorRetryRequestMetadata metadata, + kj::Maybe deadline, + TimerChannel& timer) + : metadata(kj::mv(metadata)), + deadline(kj::mv(deadline)), + timer(timer) {} + + kj::Duration retryDelay(); + + IoChannelFactory::ActorRetryRequestMetadata metadata; + kj::Maybe deadline; + TimerChannel& timer; + kj::Maybe originalDisconnect; + uint attemptCount = 1; +}; + +struct ActorFetchFailure { + kj::Exception exception; +}; + +template +using ActorFetchAttemptResult = kj::OneOf; + +ActorFetchRetryState ActorFetchRetryState::create(TimerChannel& timer) { + auto metadata = generateActorRetryRequestMetadata(kj::systemCoarseCalendarClock().now()); + kj::Maybe deadline; + // TODO(STOR-5489): Keep sender-side retries disabled until retry-claim enforcement is deployed to + // every receiver. A mixed fleet can otherwise execute both an ambiguous request and its retry. + if (util::Autogate::isEnabled(util::AutogateKey::DURABLE_OBJECT_RETRIES_FETCH) && + util::Autogate::isEnabled( + util::AutogateKey::DURABLE_OBJECT_RETRIES_FETCH_RETRY_REQUESTS)) { + deadline = timer.nowForLimitTimeout() + RETRY_BUDGET; + } + return ActorFetchRetryState(kj::mv(metadata), kj::mv(deadline), timer); +} + +kj::Maybe ActorFetchRetryState::checkDeadline() { + KJ_IF_SOME(deadlineValue, deadline) { + if (attemptCount > 1 && timer.nowForLimitTimeout() >= deadlineValue) { + return KJ_ASSERT_NONNULL(originalDisconnect).clone(); + } + } + return kj::none; +} + +kj::Duration ActorFetchRetryState::retryDelay() { + static thread_local auto generator = [] { + uint64_t seed; + getEntropy(kj::asBytes(seed)); + return std::mt19937_64(seed); + }(); + auto maximum = INITIAL_BACKOFF * (1u << (attemptCount - 1)); + std::uniform_int_distribution distribution(0, maximum / kj::NANOSECONDS); + return distribution(generator) * kj::NANOSECONDS; +} + +kj::OneOf ActorFetchRetryState::prepareRetry( + kj::Exception exception) { + if (!isRetryEnabled()) { + return kj::mv(exception); + } + if (exception.getDetail(jsg::ACTOR_RETRY_CLAIM_REJECTED_DETAIL_ID) != kj::none) { + KJ_IF_SOME(original, originalDisconnect) { + return kj::mv(original); + } + auto normalized = KJ_EXCEPTION(DISCONNECTED, "Durable Object fetch failed"); + for (auto& detail: exception.getDetails()) { + normalized.setDetail(detail.id, kj::heapArray(detail.value.asPtr())); + } + return normalized; + } + + if (exception.getType() != kj::Exception::Type::DISCONNECTED || + exception.getDetail(jsg::REQUEST_DELIVERED_TO_ACTOR_DETAIL_ID) != kj::none) { + return kj::mv(exception); + } + if (attemptCount == 1) { + originalDisconnect = exception.clone(); + } + if (exception.getDetail(jsg::REQUEST_NOT_DELIVERED_TO_ACTOR_DETAIL_ID) == kj::none) { + metadata.isRetry = IsActorRetry::YES; + KJ_ASSERT_NONNULL(originalDisconnect) + .releaseDetail(jsg::REQUEST_NOT_DELIVERED_TO_ACTOR_DETAIL_ID); + } else if (metadata.isRetry == IsActorRetry::NO) { + metadata = generateActorRetryRequestMetadata(kj::systemCoarseCalendarClock().now()); + } + if (attemptCount >= MAX_ATTEMPTS) { + return KJ_ASSERT_NONNULL(originalDisconnect).clone(); + } + + auto delay = retryDelay(); + auto deadline = KJ_ASSERT_NONNULL(this->deadline); + if (timer.nowForLimitTimeout() + delay >= deadline) { + return KJ_ASSERT_NONNULL(originalDisconnect).clone(); + } + ++attemptCount; + return delay; +} + +template +kj::Promise> captureActorFetchAttempt(kj::Promise promise) { + return kj::mv(promise) + .then([](T&& result) -> ActorFetchAttemptResult { return kj::mv(result); }) + .catch_([](kj::Exception&& exception) -> ActorFetchAttemptResult { + return ActorFetchFailure{kj::mv(exception)}; + }); +} + +jsg::Promise> fetchImplNoOutputLockAttempt(jsg::Lock& js, + jsg::Ref fetcher, + jsg::Ref jsRequest, + kj::Vector urlList, + kj::Maybe retryState); + jsg::Promise> handleHttpResponse(jsg::Lock& js, jsg::Ref fetcher, jsg::Ref jsRequest, @@ -1480,19 +1617,129 @@ jsg::Promise> handleHttpRedirectResponse(jsg::Lock& js, uint status, kj::StringPtr location); +jsg::Promise> handleWebSocketFetchResponse(jsg::Lock& js, + jsg::Ref fetcher, + jsg::Ref jsRequest, + kj::Vector urlList, + kj::Own client, + kj::Maybe> signal, + kj::HttpClient::WebSocketResponse response) { + KJ_SWITCH_ONEOF(response.webSocketOrBody) { + KJ_CASE_ONEOF(body, kj::Own) { + body = body.attach(kj::mv(client)); + return handleHttpResponse(js, kj::mv(fetcher), kj::mv(jsRequest), kj::mv(urlList), + {response.statusCode, response.statusText, response.headers, kj::mv(body)}); + } + KJ_CASE_ONEOF(webSocket, kj::Own) { + KJ_ASSERT(response.statusCode == 101); + webSocket = webSocket.attach(kj::mv(client)); + KJ_IF_SOME(s, signal) { + if (s->getAborted(js)) { + return js.rejectedPromise>(s->getReason(js)); + } + webSocket = kj::refcounted(kj::mv(webSocket), s->getCanceler()); + } + return js.resolvedPromise(makeHttpResponse(js, jsRequest->getMethodEnum(), kj::mv(urlList), + response.statusCode, response.statusText, *response.headers, newNullInputStream(), + js.alloc(js, kj::mv(webSocket)), jsRequest->getResponseBodyEncoding(), + kj::mv(signal))); + } + } + KJ_UNREACHABLE; +} + +jsg::Promise> handleHttpFetchResponse(jsg::Lock& js, + jsg::Ref fetcher, + jsg::Ref jsRequest, + kj::Vector urlList, + kj::Own client, + kj::Maybe& traceContext, + kj::HttpClient::Response response) { + response.body = response.body.attach(kj::mv(client)); + KJ_IF_SOME(ctx, traceContext) { + ctx.setTag("http.response.status_code"_kjc, static_cast(response.statusCode)); + KJ_IF_SOME(length, response.body->tryGetLength()) { + ctx.setTag("http.response.body.size"_kjc, static_cast(length)); + } + auto headerIds = IoContext::current().getHeaderIds(); + KJ_IF_SOME(cfRay, response.headers->get(headerIds.cfRay)) { + ctx.setTag("cloudflare.ray_id"_kjc, cfRay); + } + } + return handleHttpResponse( + js, kj::mv(fetcher), kj::mv(jsRequest), kj::mv(urlList), kj::mv(response)); +} + +jsg::Promise> rejectFetch(jsg::Lock& js, kj::Exception&& exception) { + return js.rejectedPromise>(js.exceptionToJsValue(kj::mv(exception))); +} + +kj::Maybe getAbortReason(jsg::Lock& js, Request& request) { + KJ_IF_SOME(signal, request.getSignal()) { + if (signal->getAborted(js)) { + return signal->getReason(js); + } + } + return kj::none; +} + +jsg::Promise> retryActorFetch(jsg::Lock& js, + jsg::Ref fetcher, + jsg::Ref jsRequest, + kj::Vector urlList, + ActorFetchRetryState state, + kj::Exception exception) { + KJ_IF_SOME(reason, getAbortReason(js, *jsRequest)) { + return js.rejectedPromise>(kj::mv(reason)); + } + auto delayOrException = state.prepareRetry(kj::mv(exception)); + KJ_IF_SOME(exception, delayOrException.tryGet()) { + return rejectFetch(js, kj::mv(exception)); + } + auto delay = KJ_ASSERT_NONNULL(delayOrException.tryGet()); + auto& ioContext = IoContext::current(); + jsRequest->rewindBody(js); + fetcher->onActorFetchRetry(); + + auto signal = jsRequest->getSignal(); + return ioContext.awaitIo(js, + AbortSignal::maybeCancelWrap(js, signal, ioContext.afterLimitTimeout(delay)), + [fetcher = kj::mv(fetcher), jsRequest = kj::mv(jsRequest), urlList = kj::mv(urlList), + retryState = kj::Maybe(kj::mv(state))](jsg::Lock& js) mutable { + return fetchImplNoOutputLockAttempt( + js, kj::mv(fetcher), kj::mv(jsRequest), kj::mv(urlList), kj::mv(retryState)); + }); +} + jsg::Promise> fetchImplNoOutputLock(jsg::Lock& js, jsg::Ref fetcher, jsg::Ref jsRequest, kj::Vector urlList) { + kj::Maybe retryState; + if (jsRequest->canRewindBody() && fetcher->supportsActorFetchRetries()) { + retryState = ActorFetchRetryState::create(IoContext::current().getIoChannelFactory().getTimer()); + } + + return fetchImplNoOutputLockAttempt( + js, kj::mv(fetcher), kj::mv(jsRequest), kj::mv(urlList), kj::mv(retryState)); +} + +jsg::Promise> fetchImplNoOutputLockAttempt(jsg::Lock& js, + jsg::Ref fetcher, + jsg::Ref jsRequest, + kj::Vector urlList, + kj::Maybe retryState) { KJ_ASSERT(!urlList.empty()); auto& ioContext = IoContext::current(); auto signal = jsRequest->getSignal(); - KJ_IF_SOME(s, signal) { - // If the AbortSignal has already been triggered, then we need to stop here. - if (s->getAborted(js)) { - return js.rejectedPromise>(s->getReason(js)); + KJ_IF_SOME(reason, getAbortReason(js, *jsRequest)) { + return js.rejectedPromise>(kj::mv(reason)); + } + KJ_IF_SOME(state, retryState) { + KJ_IF_SOME(exception, state.checkDeadline()) { + return rejectFetch(js, kj::mv(exception)); } } @@ -1505,9 +1752,8 @@ jsg::Promise> fetchImplNoOutputLock(jsg::Lock& js, ioContext.getMetrics().setNextSubrequestBodyRewindable(SubrequestBodyRewindable(bodyRewindable)); kj::Maybe actorRetryRequestMetadata; - if (bodyRewindable && fetcher->supportsActorRetryMetadata()) { - actorRetryRequestMetadata = - generateActorRetryRequestMetadata(kj::systemCoarseCalendarClock().now()); + KJ_IF_SOME(state, retryState) { + actorRetryRequestMetadata = state.getMetadata(); } // Get client and trace context (if needed) in one clean call @@ -1582,36 +1828,36 @@ jsg::Promise> fetchImplNoOutputLock(jsg::Lock& js, // subrequest. headers.unset(kj::HttpHeaderId::SEC_WEBSOCKET_EXTENSIONS); } - auto webSocketResponse = client->openWebSocket(url, headers); - return ioContext.awaitIo(js, - AbortSignal::maybeCancelWrap(js, signal, kj::mv(webSocketResponse)), - [fetcher = kj::mv(fetcher), jsRequest = kj::mv(jsRequest), urlList = kj::mv(urlList), - client = kj::mv(client), signal = kj::mv(signal)]( - jsg::Lock& js, kj::HttpClient::WebSocketResponse&& response) mutable - -> jsg::Promise> { - KJ_SWITCH_ONEOF(response.webSocketOrBody) { - KJ_CASE_ONEOF(body, kj::Own) { - body = body.attach(kj::mv(client)); - return handleHttpResponse(js, kj::mv(fetcher), kj::mv(jsRequest), kj::mv(urlList), - {response.statusCode, response.statusText, response.headers, kj::mv(body)}); - } - KJ_CASE_ONEOF(webSocket, kj::Own) { - KJ_ASSERT(response.statusCode == 101); - webSocket = webSocket.attach(kj::mv(client)); - KJ_IF_SOME(s, signal) { - // If the AbortSignal has already been triggered, then we need to stop here. - if (s->getAborted(js)) { - return js.rejectedPromise>(s->getReason(js)); + auto webSocketResponse = + AbortSignal::maybeCancelWrap(js, signal, client->openWebSocket(url, headers)); + KJ_IF_SOME(state, retryState) { + if (state.isRetryEnabled()) { + return ioContext.awaitIo(js, captureActorFetchAttempt(kj::mv(webSocketResponse)), + [fetcher = kj::mv(fetcher), jsRequest = kj::mv(jsRequest), + urlList = kj::mv(urlList), client = kj::mv(client), signal = kj::mv(signal), + retryState = kj::mv(retryState)](jsg::Lock& js, + ActorFetchAttemptResult&& result) mutable + -> jsg::Promise> { + KJ_SWITCH_ONEOF(result) { + KJ_CASE_ONEOF(response, kj::HttpClient::WebSocketResponse) { + return handleWebSocketFetchResponse(js, kj::mv(fetcher), kj::mv(jsRequest), + kj::mv(urlList), kj::mv(client), kj::mv(signal), kj::mv(response)); + } + KJ_CASE_ONEOF(failure, ActorFetchFailure) { + return retryActorFetch(js, kj::mv(fetcher), kj::mv(jsRequest), kj::mv(urlList), + kj::mv(KJ_ASSERT_NONNULL(retryState)), kj::mv(failure.exception)); } - webSocket = kj::refcounted(kj::mv(webSocket), s->getCanceler()); } - return js.resolvedPromise(makeHttpResponse(js, jsRequest->getMethodEnum(), - kj::mv(urlList), response.statusCode, response.statusText, *response.headers, - newNullInputStream(), js.alloc(js, kj::mv(webSocket)), - jsRequest->getResponseBodyEncoding(), kj::mv(signal))); - } + KJ_UNREACHABLE; + }); } - KJ_UNREACHABLE; + } + return ioContext.awaitIo(js, kj::mv(webSocketResponse), + [fetcher = kj::mv(fetcher), jsRequest = kj::mv(jsRequest), urlList = kj::mv(urlList), + client = kj::mv(client), signal = kj::mv(signal)]( + jsg::Lock& js, kj::HttpClient::WebSocketResponse&& response) mutable { + return handleWebSocketFetchResponse(js, kj::mv(fetcher), kj::mv(jsRequest), kj::mv(urlList), + kj::mv(client), kj::mv(signal), kj::mv(response)); }); } else { kj::Maybe nativeRequest; @@ -1670,7 +1916,7 @@ jsg::Promise> fetchImplNoOutputLock(jsg::Lock& js, } else { nativeRequest = client->request(jsRequest->getMethodEnum(), url, headers, static_cast(0)); } - return ioContext.awaitIo(js, + auto responsePromise = AbortSignal::maybeCancelWrap(js, signal, kj::mv(KJ_ASSERT_NONNULL(nativeRequest).response)) .catch_([](kj::Exception&& exception) -> kj::Promise { if (exception.getDescription().startsWith("invalid Content-Length header value")) { @@ -1679,23 +1925,36 @@ jsg::Promise> fetchImplNoOutputLock(jsg::Lock& js, return JSG_KJ_EXCEPTION(FAILED, Error, "Worker not found."); } return kj::mv(exception); - }), - [fetcher = kj::mv(fetcher), jsRequest = kj::mv(jsRequest), urlList = kj::mv(urlList), - client = kj::mv(client), traceContext = kj::mv(traceContext)](jsg::Lock& js, - kj::HttpClient::Response&& response) mutable -> jsg::Promise> { - response.body = response.body.attach(kj::mv(client)); - KJ_IF_SOME(ctx, traceContext) { - ctx.setTag("http.response.status_code"_kjc, static_cast(response.statusCode)); - KJ_IF_SOME(length, response.body->tryGetLength()) { - ctx.setTag("http.response.body.size"_kjc, static_cast(length)); - } - auto headerIds = IoContext::current().getHeaderIds(); - KJ_IF_SOME(cfRay, response.headers->get(headerIds.cfRay)) { - ctx.setTag("cloudflare.ray_id"_kjc, cfRay); - } + }); + KJ_IF_SOME(state, retryState) { + if (state.isRetryEnabled()) { + auto resultPromise = captureActorFetchAttempt(kj::mv(responsePromise)); + return ioContext.awaitIo(js, kj::mv(resultPromise), + [fetcher = kj::mv(fetcher), jsRequest = kj::mv(jsRequest), + urlList = kj::mv(urlList), client = kj::mv(client), + traceContext = kj::mv(traceContext), retryState = kj::mv(retryState)]( + jsg::Lock& js, ActorFetchAttemptResult&& result) mutable + -> jsg::Promise> { + KJ_SWITCH_ONEOF(result) { + KJ_CASE_ONEOF(response, kj::HttpClient::Response) { + return handleHttpFetchResponse(js, kj::mv(fetcher), kj::mv(jsRequest), + kj::mv(urlList), kj::mv(client), traceContext, kj::mv(response)); + } + KJ_CASE_ONEOF(failure, ActorFetchFailure) { + return retryActorFetch(js, kj::mv(fetcher), kj::mv(jsRequest), kj::mv(urlList), + kj::mv(KJ_ASSERT_NONNULL(retryState)), kj::mv(failure.exception)); + } + } + KJ_UNREACHABLE; + }); } - return handleHttpResponse( - js, kj::mv(fetcher), kj::mv(jsRequest), kj::mv(urlList), kj::mv(response)); + } + return ioContext.awaitIo(js, kj::mv(responsePromise), + [fetcher = kj::mv(fetcher), jsRequest = kj::mv(jsRequest), urlList = kj::mv(urlList), + client = kj::mv(client), traceContext = kj::mv(traceContext)]( + jsg::Lock& js, kj::HttpClient::Response&& response) mutable { + return handleHttpFetchResponse(js, kj::mv(fetcher), kj::mv(jsRequest), kj::mv(urlList), + kj::mv(client), traceContext, kj::mv(response)); }); } } @@ -2497,7 +2756,7 @@ Fetcher::ClientWithTracing Fetcher::getClientWithTracing( auto& outgoingFactory = KJ_REQUIRE_NONNULL( channelOrClientFactory.tryGet>(), "actor retry metadata supplied to an unsupported Fetcher"); - KJ_REQUIRE(outgoingFactory->supportsActorRetryMetadata(), + KJ_REQUIRE(outgoingFactory->supportsActorFetchRetries(), "actor retry metadata supplied to an unsupported Fetcher"); auto client = outgoingFactory->newSingleUseClientWithActorRetryMetadata( kj::mv(cfStr), kj::mv(metadata)); @@ -2542,13 +2801,22 @@ Fetcher::ClientWithTracing Fetcher::getClientWithTracing( KJ_UNREACHABLE; } -bool Fetcher::supportsActorRetryMetadata() { +bool Fetcher::supportsActorFetchRetries() { KJ_IF_SOME(outgoingFactory, channelOrClientFactory.tryGet>()) { - return outgoingFactory->supportsActorRetryMetadata(); + return outgoingFactory->supportsActorFetchRetries(); } return false; } +void Fetcher::onActorFetchRetry() { + auto& outgoingFactory = KJ_REQUIRE_NONNULL( + channelOrClientFactory.tryGet>(), + "actor fetch retry requested from an unsupported Fetcher"); + KJ_REQUIRE(outgoingFactory->supportsActorFetchRetries(), + "actor fetch retry requested from an unsupported Fetcher"); + outgoingFactory->onActorFetchRetry(); +} + kj::Own Fetcher::getSubrequestChannel(IoContext& ioContext) { KJ_SWITCH_ONEOF(channelOrClientFactory) { KJ_CASE_ONEOF(channel, uint) { diff --git a/src/workerd/api/http.h b/src/workerd/api/http.h index 55dafa175cc..fbaedb110d8 100644 --- a/src/workerd/api/http.h +++ b/src/workerd/api/http.h @@ -236,10 +236,14 @@ class Fetcher: public JsRpcClientProvider { public: virtual kj::Own newSingleUseClient(kj::Maybe cfStr) = 0; - virtual bool supportsActorRetryMetadata() const { + virtual bool supportsActorFetchRetries() const { return false; } + virtual void onActorFetchRetry() { + KJ_FAIL_REQUIRE("actor fetch retry requested from an unsupported Fetcher"); + } + // Factories that can carry actor retry metadata override this method. The default rejects the // metadata rather than silently starting a new logical call. virtual kj::Own newSingleUseClientWithActorRetryMetadata( @@ -308,7 +312,8 @@ class Fetcher: public JsRpcClientProvider { kj::ConstString operationName, kj::Maybe actorRetryRequestMetadata); - bool supportsActorRetryMetadata(); + bool supportsActorFetchRetries(); + void onActorFetchRetry(); // Get a SubrequestChannel representing this Fetcher. kj::Own getSubrequestChannel(IoContext& ioContext); diff --git a/src/workerd/io/actor-id.h b/src/workerd/io/actor-id.h index e8169b04db2..d7f208ef09f 100644 --- a/src/workerd/io/actor-id.h +++ b/src/workerd/io/actor-id.h @@ -26,6 +26,12 @@ enum class ActorRoutingMode { // Version information for an actor. Used to specify cohort. struct ActorVersion { kj::Maybe cohort; + + ActorVersion clone() const { + return { + .cohort = cohort.map([](const kj::String& value) { return kj::str(value); }), + }; + } }; // An abstract class that implements generation of global actor IDs in a particular namespace. diff --git a/src/workerd/io/io-channels.h b/src/workerd/io/io-channels.h index a4dec88fc5e..c5e5b578dcb 100644 --- a/src/workerd/io/io-channels.h +++ b/src/workerd/io/io-channels.h @@ -79,6 +79,11 @@ class TimerChannel { // time limits on some sort of operation, not for implementing application-driven timing, as it does // not implement any Spectre mitigations. virtual kj::Promise afterLimitTimeout(kj::Duration t) = 0; + + // Returns the precise monotonic time used to calculate deadlines for afterLimitTimeout(). + virtual kj::TimePoint nowForLimitTimeout() { + return kj::systemPreciseMonotonicClock().now(); + } }; class WorkerStubChannel; diff --git a/src/workerd/io/worker.c++ b/src/workerd/io/worker.c++ index 88d32fb3831..91ec2952ffc 100644 --- a/src/workerd/io/worker.c++ +++ b/src/workerd/io/worker.c++ @@ -3959,9 +3959,7 @@ kj::Promise Worker::Actor::ensureConstructedImpl(IoContext& context, Actor auto ctx = js.alloc(js, cloneId(), jsg::JsValue(KJ_ASSERT_NONNULL(lock.getWorker().impl->ctxExports).getHandle(js)), impl->props.toJs(js), kj::mv(storage), kj::mv(impl->container), containerRunning, - impl->facetManager, impl->version.map([](ActorVersion& v) { - return ActorVersion{.cohort = v.cohort.map([](kj::String& s) { return kj::str(s); })}; - })); + impl->facetManager, impl->version.map([](ActorVersion& v) { return v.clone(); })); auto handler = info.cls(lock, ctx.addRef(), KJ_ASSERT_NONNULL(lock.getWorker().impl->env).addRef(js)); diff --git a/src/workerd/server/server.c++ b/src/workerd/server/server.c++ index 10f87828f5f..508d95adbb3 100644 --- a/src/workerd/server/server.c++ +++ b/src/workerd/server/server.c++ @@ -4548,6 +4548,10 @@ class Server::WorkerService final: public Service, return threadContext.getUnsafeTimer().afterDelay(t); } + kj::TimePoint nowForLimitTimeout() override { + return monotonicClock.now(); + } + // --------------------------------------------------------------------------- // implements LimitEnforcer //