diff --git a/src/workerd/api/actor-call-retry-test.c++ b/src/workerd/api/actor-call-retry-test.c++ index d03ae982b6d..23efa7d4be6 100644 --- a/src/workerd/api/actor-call-retry-test.c++ +++ b/src/workerd/api/actor-call-retry-test.c++ @@ -56,14 +56,18 @@ class RecordingObserver final: public RequestObserver { kj::Vector outcomes; }; -kj::Rc newRetryState(TestTimerChannel& timer, RecordingObserver& observer) { +// A retry state with every gate enabled and a replayable payload. +kj::Rc newRetryState(TestTimerChannel& timer, + RecordingObserver& observer, + ActorRetryPolicy policy = ActorRetryPolicy::systemDefault()) { return kj::rc(timer, observer, ActorCallRetryState::Config{ .callType = ActorRetryCallType::JSRPC, .observationEnabled = ActorRetryGateEnabled::YES, .enforcementEnabled = ActorRetryGateEnabled::YES, .payloadReplayable = ActorCallPayloadReplayable::YES, - }); + }, + policy, timer.nowForLimitTimeout()); } ActorCallRetryState::Attempt startAttempt(ActorCallRetryState& state) { @@ -218,7 +222,7 @@ KJ_TEST("claim rejection after retry commitment returns the original disconnect" KJ_EXPECT(observer->outcomes[0] == ActorRetryOutcome::CLAIM_REJECTED); } -KJ_TEST("actor retries stop after five total attempts") { +KJ_TEST("system retry policy allows five total attempts") { TestTimerChannel timer; auto observer = kj::refcounted(); auto state = newRetryState(timer, *observer); @@ -230,7 +234,8 @@ KJ_TEST("actor retries stop after five total attempts") { if (attempt < 4) { KJ_EXPECT(result.is()); } else { - KJ_EXPECT(result.is()); + auto& failure = KJ_ASSERT_NONNULL(result.tryGet()); + KJ_EXPECT(failure.getDescription().contains("disconnected")); } } @@ -274,6 +279,60 @@ KJ_TEST("actor retries report retry budget exhaustion when the next attempt cann KJ_EXPECT(observer->outcomes[0] == ActorRetryOutcome::RETRY_BUDGET_EXHAUSTED); } +KJ_TEST("user-defined retry count excludes the initial attempt") { + auto checkAttemptCount = [](UserDefinedRetryPolicy policy, uint expectedAttempts) { + TestTimerChannel timer; + auto observer = kj::refcounted(); + auto state = newRetryState(timer, *observer, ActorRetryPolicy::userDefined(policy)); + + uint attemptCount = 1; + startAttempt(*state); + if (policy.maxAttempts == 0) { + KJ_EXPECT(!state->isRetryEnabled()); + KJ_EXPECT(attemptCount == expectedAttempts); + KJ_EXPECT(observer->retryCallTypes.size() == 0); + return; + } + + for (uint retry = 0; retry < policy.maxAttempts; ++retry) { + KJ_EXPECT(handleFailure(*state, makeDisconnect("disconnected"_kj)).is()); + startAttempt(*state); + ++attemptCount; + } + + KJ_EXPECT(handleFailure(*state, makeDisconnect("disconnected"_kj)).is()); + KJ_EXPECT(attemptCount == expectedAttempts); + KJ_EXPECT(observer->retryCallTypes.size() == policy.maxAttempts); + }; + + checkAttemptCount(UserDefinedRetryPolicy{.maxAttempts = 0}, 1); + checkAttemptCount(UserDefinedRetryPolicy{.maxAttempts = 1}, 2); + checkAttemptCount(UserDefinedRetryPolicy{}, ActorRetryPolicy::systemDefault().maxAttempts()); +} + +KJ_TEST("user-defined retry timeout defaults to the system retry timeout") { + KJ_EXPECT(ActorRetryPolicy::userDefined(UserDefinedRetryPolicy{}).timeout() == + ActorRetryPolicy::systemDefault().timeout()); +} + +KJ_TEST("expiring a configured retry timeout returns the original disconnect") { + TestTimerChannel timer; + auto observer = kj::refcounted(); + auto state = newRetryState(timer, *observer, + ActorRetryPolicy::userDefined(UserDefinedRetryPolicy{.timeout = 500 * kj::MILLISECONDS})); + + startAttempt(*state); + KJ_EXPECT(handleFailure(*state, makeDisconnect("original disconnect"_kj)).is()); + startAttempt(*state); + timer.advance(500 * kj::MILLISECONDS); + + auto result = handleFailure(*state, makeDisconnect("second disconnect"_kj)); + auto& failure = KJ_ASSERT_NONNULL(result.tryGet()); + KJ_EXPECT(failure.getDescription().contains("original disconnect")); + KJ_ASSERT(observer->outcomes.size() == 1); + KJ_EXPECT(observer->outcomes[0] == ActorRetryOutcome::RETRY_BUDGET_EXHAUSTED); +} + KJ_TEST("actor retries return the original disconnect when the deadline expires before retry") { TestTimerChannel timer; auto observer = kj::refcounted(); @@ -318,7 +377,8 @@ KJ_TEST("actor calls do not retry when retry requests are disabled") { .observationEnabled = ActorRetryGateEnabled::YES, .enforcementEnabled = ActorRetryGateEnabled::NO, .payloadReplayable = ActorCallPayloadReplayable::YES, - }); + }, + ActorRetryPolicy::systemDefault(), timer.nowForLimitTimeout()); auto first = startAttempt(*state); KJ_EXPECT(KJ_ASSERT_NONNULL(first.getMetadata()).retryGateEnabled == ActorRetryGateEnabled::NO); diff --git a/src/workerd/api/actor-call-retry.c++ b/src/workerd/api/actor-call-retry.c++ index 3e14c302857..2cc89c20343 100644 --- a/src/workerd/api/actor-call-retry.c++ +++ b/src/workerd/api/actor-call-retry.c++ @@ -11,20 +11,20 @@ namespace workerd::api { -ActorCallRetryState::ActorCallRetryState( - TimerChannel& timer, RequestObserver& observer, Config config) +ActorCallRetryState::ActorCallRetryState(TimerChannel& timer, + RequestObserver& observer, + Config config, + ActorRetryPolicy policy, + kj::TimePoint callStart) : timer(timer), observer(kj::addRef(observer)), - config(config) { - auto payloadReplayable = config.payloadReplayable.toBool(); - retriesEnabled = - payloadReplayable && config.observationEnabled.toBool() && config.enforcementEnabled.toBool(); - if (payloadReplayable) { - metadata = generateActorRetryRequestMetadata( - kj::systemCoarseCalendarClock().now(), config.enforcementEnabled); - } - if (retriesEnabled) { - deadline = timer.nowForLimitTimeout() + RETRY_BUDGET; + config(config), + policy(policy), + callStart(callStart), + retriesEnabled(config.payloadReplayable.toBool() && config.observationEnabled.toBool() && + config.enforcementEnabled.toBool() && policy.maxAttempts() > 1) { + if (config.payloadReplayable.toBool()) { + metadata = freshMetadata(); } } @@ -34,14 +34,16 @@ ActorCallRetryState::~ActorCallRetryState() noexcept(false) { } } +IoChannelFactory::ActorRetryRequestMetadata ActorCallRetryState::freshMetadata() const { + return generateActorRetryRequestMetadata( + kj::systemCoarseCalendarClock().now(), config.enforcementEnabled); +} + kj::OneOf ActorCallRetryState::startAttempt() { auto isFirstAttempt = IsFirstActorCallAttempt(attemptCount == 1); if (!isFirstAttempt.toBool()) { - KJ_IF_SOME(deadlineValue, deadline) { - if (timer.nowForLimitTimeout() >= deadlineValue) { - recordOutcome(ActorRetryOutcome::RETRY_BUDGET_EXHAUSTED); - return KJ_ASSERT_NONNULL(originalDisconnect).clone(); - } + if (timer.nowForLimitTimeout() >= retryCutoff()) { + return retryTimeoutExpired(); } ++retryAttemptsStarted; observer->recordActorRetry(config.callType); @@ -49,6 +51,11 @@ kj::OneOf ActorCallRetryState::star return Attempt(metadata, isFirstAttempt); } +kj::Exception ActorCallRetryState::retryTimeoutExpired() { + recordOutcome(ActorRetryOutcome::RETRY_BUDGET_EXHAUSTED); + return KJ_ASSERT_NONNULL(originalDisconnect).clone(); +} + kj::OneOf ActorCallRetryState::handleAttemptFailure( kj::Exception exception) { KJ_REQUIRE(retriesEnabled); @@ -98,28 +105,22 @@ kj::OneOf ActorCallRetryState::checkCanRetry(kj::Ex recordOutcome(ActorRetryOutcome::UNABLE_TO_RETRY); return kj::mv(exception); } - if (attemptCount >= MAX_ATTEMPTS) { + if (originalDisconnect == kj::none) { + originalDisconnect = exception.clone(); + } + if (attemptCount >= policy.maxAttempts()) { recordOutcome(ActorRetryOutcome::ATTEMPTS_EXHAUSTED); return KJ_ASSERT_NONNULL(originalDisconnect).clone(); } auto delay = retryDelay(); - auto deadline = KJ_ASSERT_NONNULL(this->deadline); - if (timer.nowForLimitTimeout() + delay >= deadline) { - recordOutcome(ActorRetryOutcome::RETRY_BUDGET_EXHAUSTED); - KJ_IF_SOME(original, originalDisconnect) { - return original.clone(); - } - return kj::mv(exception); - } - if (attemptCount == 1) { - originalDisconnect = exception.clone(); + if (timer.nowForLimitTimeout() + delay >= retryCutoff()) { + return retryTimeoutExpired(); } if (exception.getDetail(jsg::REQUEST_NOT_DELIVERED_TO_ACTOR_DETAIL_ID) == kj::none) { KJ_ASSERT_NONNULL(metadata).isRetry = IsActorRetry::YES; } else if (KJ_ASSERT_NONNULL(metadata).isRetry == IsActorRetry::NO) { - metadata = generateActorRetryRequestMetadata( - kj::systemCoarseCalendarClock().now(), config.enforcementEnabled); + metadata = freshMetadata(); } ++attemptCount; @@ -142,7 +143,7 @@ kj::Duration ActorCallRetryState::retryDelay() { getEntropy(kj::asBytes(seed)); return std::mt19937_64(seed); }(); - auto maximum = INITIAL_BACKOFF * (1u << (attemptCount - 1)); + auto maximum = kj::min(INITIAL_BACKOFF * (1u << (attemptCount - 1)), MAX_BACKOFF); std::uniform_int_distribution distribution(0, maximum / kj::NANOSECONDS); return distribution(generator) * kj::NANOSECONDS; } diff --git a/src/workerd/api/actor-call-retry.h b/src/workerd/api/actor-call-retry.h index eb7069d5f9d..3e98d41ba62 100644 --- a/src/workerd/api/actor-call-retry.h +++ b/src/workerd/api/actor-call-retry.h @@ -18,6 +18,58 @@ namespace workerd::api { WD_STRONG_BOOL(ActorCallRetriesAllowed); WD_STRONG_BOOL(IsFirstActorCallAttempt); +// Retry behavior carried from a Durable Object namespace binding that configures one. Applied when +// DURABLE_OBJECT_RETRIES_USERLAND is enabled. A binding without a retry policy runs under +// ActorRetryPolicy::systemDefault() instead. +struct UserDefinedRetryPolicy { + // Matches the runtime's default of five attempts in total. + static constexpr uint DEFAULT_MAX_ATTEMPTS = 4; + static constexpr uint MAX_CONFIGURABLE_ATTEMPTS = 10; + // Matches the runtime's default retry timeout. + static constexpr auto DEFAULT_TIMEOUT = 10 * kj::SECONDS; + // The first retry's backoff can be up to 500ms, so a shorter timeout would usually stop the call + // from retrying at all. + static constexpr auto MIN_CONFIGURABLE_TIMEOUT = 500 * kj::MILLISECONDS; + static constexpr auto MAX_CONFIGURABLE_TIMEOUT = 60 * kj::SECONDS; + + // Retries after the initial attempt, so the total attempt count is one more than this. + uint maxAttempts = DEFAULT_MAX_ATTEMPTS; + // See ActorRetryPolicy::timeout(). + kj::Duration timeout = DEFAULT_TIMEOUT; +}; + +// The retry limits a single actor call runs under, resolved from either the runtime's defaults or +// a binding's UserDefinedRetryPolicy. Everything downstream reads this and never asks which. +class ActorRetryPolicy { + public: + static ActorRetryPolicy systemDefault() { + return ActorRetryPolicy(SYSTEM_DEFAULT_MAX_ATTEMPTS, SYSTEM_DEFAULT_TIMEOUT); + } + static ActorRetryPolicy userDefined(const UserDefinedRetryPolicy& policy) { + return ActorRetryPolicy(1 + policy.maxAttempts, policy.timeout); + } + + // Total attempts, including the initial one. + uint maxAttempts() const { + return attempts; + } + // Time from the start of the call after which no retry may start or keep running. + kj::Duration timeout() const { + return retryTimeout; + } + + private: + static constexpr uint SYSTEM_DEFAULT_MAX_ATTEMPTS = 5; + static constexpr auto SYSTEM_DEFAULT_TIMEOUT = 10 * kj::SECONDS; + + ActorRetryPolicy(uint attempts, kj::Duration retryTimeout) + : attempts(attempts), + retryTimeout(retryTimeout) {} + + uint attempts; + kj::Duration retryTimeout; +}; + class ActorCallRetryState final: public kj::Refcounted { public: struct Config { @@ -58,7 +110,13 @@ class ActorCallRetryState final: public kj::Refcounted { IsFirstActorCallAttempt isFirstAttempt; }; - ActorCallRetryState(TimerChannel& timer, RequestObserver& observer, Config config); + // `callStart` is when the logical call began, and the retry timeout runs from it. A redirected + // request passes the original call's start, so one timeout covers the whole chain. + ActorCallRetryState(TimerChannel& timer, + RequestObserver& observer, + Config config, + ActorRetryPolicy policy, + kj::TimePoint callStart); ~ActorCallRetryState() noexcept(false); kj::OneOf startAttempt(); @@ -72,15 +130,34 @@ class ActorCallRetryState final: public kj::Refcounted { bool isRetryEnabled() const { return retriesEnabled; } + kj::TimePoint getCallStart() const { + return callStart; + } + // Rejects `promise` with the original disconnect if the retry timeout expires first. The first + // attempt is never cancelled, so this only affects retries. + template + kj::Promise enforceRetryTimeout(kj::Promise promise) { + if (attemptCount == 1) return kj::mv(promise); + auto timeout = timer.atLimitTimeout(retryCutoff()) + .then([self = addRefToThis()]() mutable -> kj::Promise { + return self->retryTimeoutExpired(); + }); + return kj::mv(promise).exclusiveJoin(kj::mv(timeout)); + } void maybeStartRetryLatencyTimer(const kj::Exception& exception); void recordRecovered(); void recordCanceled(); private: - static constexpr uint MAX_ATTEMPTS = 5; - static constexpr auto RETRY_BUDGET = 10 * kj::SECONDS; static constexpr auto INITIAL_BACKOFF = 500 * kj::MILLISECONDS; + static constexpr auto MAX_BACKOFF = 2 * kj::SECONDS; + IoChannelFactory::ActorRetryRequestMetadata freshMetadata() const; + // No retry may start, or still be running, at or after this point. + kj::TimePoint retryCutoff() const { + return callStart + policy.timeout(); + } + kj::Exception retryTimeoutExpired(); kj::Maybe handleClaimRejection(const kj::Exception& exception); kj::OneOf checkCanRetry(kj::Exception exception); kj::Duration retryDelay(); @@ -89,9 +166,10 @@ class ActorCallRetryState final: public kj::Refcounted { TimerChannel& timer; kj::Own observer; Config config; + ActorRetryPolicy policy; + kj::TimePoint callStart; bool retriesEnabled; kj::Maybe metadata; - kj::Maybe deadline; kj::Maybe originalDisconnect; kj::Maybe retryStartTime; uint attemptCount = 1; diff --git a/src/workerd/api/actor-fetch-retry-test.c++ b/src/workerd/api/actor-fetch-retry-test.c++ index a13fbff66b9..5990945e0c3 100644 --- a/src/workerd/api/actor-fetch-retry-test.c++ +++ b/src/workerd/api/actor-fetch-retry-test.c++ @@ -113,6 +113,8 @@ enum class ReplayAction { NON_RETRYABLE_FAILURE, SLOW_RESPONSE, PAUSE_RESPONSE, + REDIRECT, + SLOW_REDIRECT, PAUSE_NEXT_RETRY, RETRY_DELAY_EXCEEDS_BUDGET, CLIENT_CREATION_FAILURE, @@ -195,6 +197,11 @@ class DeterministicTimerChannel final: public TimerChannel { return timer.now(); } + kj::Promise atLimitTimeout(kj::TimePoint deadline) override { + if (deadline <= timer.now()) return kj::READY_NOW; + return timer.atTime(deadline); + } + private: kj::TimerImpl& timer; kj::Maybe nextDelay; @@ -217,6 +224,7 @@ struct ReplayState { uint observedRetryCount = 0; kj::Vector outcomes; kj::Maybe timerChannel; + kj::Duration retryTimeoutDelay = 11 * kj::SECONDS; }; class RetryRecordingObserver final: public RequestObserver { @@ -254,7 +262,9 @@ class ReplayFetchTarget final: public WorkerInterface { kj::HttpService::Response& response) override { auto attempt = state.requestCount++; kj::Maybe> slowResponseDelay; - if (attempt < state.actions.size() && state.actions[attempt] == ReplayAction::SLOW_RESPONSE) { + if (attempt < state.actions.size() && + (state.actions[attempt] == ReplayAction::SLOW_RESPONSE || + state.actions[attempt] == ReplayAction::SLOW_REDIRECT)) { slowResponseDelay = IoContext::current().afterLimitTimeout(11 * kj::SECONDS); } if (headers.isWebSocket()) { @@ -267,6 +277,16 @@ class ReplayFetchTarget final: public WorkerInterface { case ReplayAction::SLOW_RESPONSE: co_await kj::mv(KJ_ASSERT_NONNULL(slowResponseDelay)); break; + case ReplayAction::SLOW_REDIRECT: + co_await kj::mv(KJ_ASSERT_NONNULL(slowResponseDelay)); + [[fallthrough]]; + case ReplayAction::REDIRECT: { + auto responseHeaders = headers.cloneShallow(); + responseHeaders.clear(); + responseHeaders.setPtr(kj::HttpHeaderId::LOCATION, "/redirected"_kj); + response.send(303, "See Other"_kj, responseHeaders, static_cast(0)); + co_return; + } case ReplayAction::PAUSE_RESPONSE: KJ_REQUIRE_NONNULL(state.pauseStartedFulfiller)->fulfill(); co_await kj::Promise(kj::NEVER_DONE); @@ -286,7 +306,7 @@ class ReplayFetchTarget final: public WorkerInterface { kj::throwRecoverableException(makeReplayFailure(action)); break; case ReplayAction::RETRY_DELAY_EXCEEDS_BUDGET: - KJ_REQUIRE_NONNULL(state.timerChannel).delayNextTimeoutBy(11 * kj::SECONDS); + KJ_REQUIRE_NONNULL(state.timerChannel).delayNextTimeoutBy(state.retryTimeoutDelay); kj::throwRecoverableException(makeReplayFailure(action)); break; } @@ -327,7 +347,10 @@ class ReplayFetchTarget final: public WorkerInterface { class ReplayOutgoingFactory final: public Fetcher::OutgoingFactory { public: - ReplayOutgoingFactory(ReplayState& state): state(state) {} + ReplayOutgoingFactory( + ReplayState& state, kj::Maybe retryPolicy = kj::none) + : state(state), + retryPolicy(retryPolicy) {} Result newSingleUseClient(kj::Maybe, MakeUserSpanParent) override { return {.client = kj::heap(state), .spanParents = kj::none}; @@ -341,6 +364,10 @@ class ReplayOutgoingFactory final: public Fetcher::OutgoingFactory { ++state.acceptedRetryCount; } + kj::Maybe getUserDefinedRetryPolicy() const override { + return retryPolicy; + } + Result newActorCallAttempt( kj::Maybe, ActorCallRetryState::Attempt attempt, MakeUserSpanParent) override { if (!attempt.getIsFirstAttempt().toBool()) { @@ -362,6 +389,7 @@ class ReplayOutgoingFactory final: public Fetcher::OutgoingFactory { private: ReplayState& state; + kj::Maybe retryPolicy; }; enum class ActorFetchKind { @@ -374,11 +402,19 @@ enum class ActorFetchBodyKind { STREAMING, }; +enum class UserDefinedRetryPolicyEnabled { + DISABLED, + ENABLED, +}; + kj::Maybe runActorFetch(ReplayState& state, ActorRetryGateEnabled retryGateEnabled, kj::Maybe body, ActorFetchKind kind, - ActorFetchBodyKind bodyKind = ActorFetchBodyKind::NULL_OR_BUFFERED) { + ActorFetchBodyKind bodyKind = ActorFetchBodyKind::NULL_OR_BUFFERED, + kj::Maybe retryPolicy = kj::none, + UserDefinedRetryPolicyEnabled userDefinedRetryPolicyEnabled = + UserDefinedRetryPolicyEnabled::DISABLED) { kj::TimerImpl timer(kj::origin()); DeterministicTimerChannel timerChannel(timer); state.timerChannel = timerChannel; @@ -393,19 +429,20 @@ kj::Maybe runActorFetch(ReplayState& state, return kj::refcounted(state); }), }); + kj::Vector autogates; + autogates.add("durable-object-retries-fetch"_kj); if (retryGateEnabled.toBool()) { - 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); + autogates.add("durable-object-retries-fetch-retry-requests"_kj); + } + if (userDefinedRetryPolicyEnabled == UserDefinedRetryPolicyEnabled::ENABLED) { + autogates.add("durable-object-retries-userland"_kj); } + util::Autogate::initAutogateNamesForTest(autogates.asPtr(), util::IgnoreAllAutogatesEnv::YES); kj::Maybe failure; fixture.runInIoContext([&](const TestFixture::Environment& env) { - auto fetcher = env.js.alloc( - env.context.addObject(kj::heap(state)), + auto fetcher = env.js.alloc(env.context.addObject( + kj::heap(state, retryPolicy)), Fetcher::RequiresHostAndProtocol::YES); RequestInitializerDict init; if (bodyKind == ActorFetchBodyKind::STREAMING) { @@ -434,6 +471,16 @@ kj::Maybe runActorFetch(ReplayState& state, return failure; } +kj::Maybe runActorFetch(ReplayState& state, + ActorRetryGateEnabled retryGateEnabled, + kj::Maybe body, + ActorFetchKind kind, + kj::Maybe retryPolicy, + UserDefinedRetryPolicyEnabled userDefinedRetryPolicyEnabled) { + return runActorFetch(state, retryGateEnabled, body, kind, ActorFetchBodyKind::NULL_OR_BUFFERED, + retryPolicy, userDefinedRetryPolicyEnabled); +} + void runActorFetchUntilCanceled(ReplayState& state, kj::Promise pauseStarted) { kj::TimerImpl timer(kj::origin()); DeterministicTimerChannel timerChannel(timer); @@ -1016,14 +1063,72 @@ KJ_TEST("dropping actor fetch during a later backoff reports other") { KJ_EXPECT(state.outcomes[0] == ActorRetryOutcome::OTHER); } -KJ_TEST("actor fetch stops after five attempts") { +KJ_TEST("actor fetch starts a new actor call after a redirect") { + ReplayState state{ + .actions = kj::arr(ReplayAction::AMBIGUOUS, ReplayAction::REDIRECT), + }; + + KJ_EXPECT( + runActorFetch(state, ActorRetryGateEnabled::YES, kj::none, ActorFetchKind::HTTP) == kj::none); + KJ_EXPECT(state.requestCount == 3); + KJ_EXPECT(state.retryCount == 1); + KJ_ASSERT(state.metadata.size() == 3); + KJ_EXPECT(state.metadata[0].isRetry == IsActorRetry::NO); + KJ_EXPECT(state.metadata[1].isRetry == IsActorRetry::YES); + KJ_EXPECT(state.metadata[0].nonce == state.metadata[1].nonce); + KJ_EXPECT(state.metadata[2].isRetry == IsActorRetry::NO); + KJ_EXPECT(state.metadata[1].nonce != state.metadata[2].nonce); + KJ_ASSERT(state.countSubrequests.size() == 3); + KJ_EXPECT(state.countSubrequests[0] == CountSubrequest::YES); + KJ_EXPECT(state.countSubrequests[1] == CountSubrequest::NO); + KJ_EXPECT(state.countSubrequests[2] == CountSubrequest::YES); + KJ_ASSERT(state.outcomes.size() == 1); + KJ_EXPECT(state.outcomes[0] == ActorRetryOutcome::RECOVERED); +} + +KJ_TEST("actor fetch retry count resets after a redirect") { ReplayState state{ .actions = kj::arr(ReplayAction::AMBIGUOUS, ReplayAction::AMBIGUOUS, ReplayAction::AMBIGUOUS, - ReplayAction::AMBIGUOUS, ReplayAction::AMBIGUOUS), + ReplayAction::AMBIGUOUS, ReplayAction::REDIRECT, ReplayAction::AMBIGUOUS), }; KJ_EXPECT( - runActorFetch(state, ActorRetryGateEnabled::YES, kj::none, ActorFetchKind::HTTP) != kj::none); + runActorFetch(state, ActorRetryGateEnabled::YES, kj::none, ActorFetchKind::HTTP) == kj::none); + KJ_EXPECT(state.requestCount == 7); + KJ_EXPECT(state.retryCount == 5); + KJ_ASSERT(state.metadata.size() == 7); + KJ_EXPECT(state.metadata[4].nonce != state.metadata[5].nonce); + KJ_EXPECT(state.metadata[5].isRetry == IsActorRetry::NO); + KJ_ASSERT(state.countSubrequests.size() == 7); + KJ_EXPECT(state.countSubrequests[5] == CountSubrequest::YES); + KJ_ASSERT(state.outcomes.size() == 2); + KJ_EXPECT(state.outcomes[0] == ActorRetryOutcome::RECOVERED); + KJ_EXPECT(state.outcomes[1] == ActorRetryOutcome::RECOVERED); +} + +KJ_TEST("actor fetch makes a redirected streaming request retryable") { + ReplayState state{ + .actions = kj::arr(ReplayAction::REDIRECT, ReplayAction::AMBIGUOUS), + }; + + KJ_EXPECT(runActorFetch(state, ActorRetryGateEnabled::YES, kj::none, ActorFetchKind::HTTP, + ActorFetchBodyKind::STREAMING) == kj::none); + KJ_EXPECT(state.requestCount == 3); + KJ_EXPECT(state.retryCount == 1); + KJ_ASSERT(state.metadata.size() == 2); + KJ_EXPECT(state.metadata[0].isRetry == IsActorRetry::NO); + KJ_EXPECT(state.metadata[1].isRetry == IsActorRetry::YES); +} + +KJ_TEST("actor fetch ignores configured retry count while the userland retry gate is disabled") { + ReplayState state{ + .actions = kj::arr(ReplayAction::AMBIGUOUS, ReplayAction::AMBIGUOUS, ReplayAction::AMBIGUOUS, + ReplayAction::AMBIGUOUS, ReplayAction::AMBIGUOUS), + }; + + KJ_EXPECT(runActorFetch(state, ActorRetryGateEnabled::YES, kj::none, ActorFetchKind::HTTP, + UserDefinedRetryPolicy{.maxAttempts = 0}, + UserDefinedRetryPolicyEnabled::DISABLED) != kj::none); KJ_EXPECT(state.requestCount == 5); KJ_EXPECT(state.retryCount == 4); KJ_EXPECT(state.observedRetryCount == 4); @@ -1031,18 +1136,109 @@ KJ_TEST("actor fetch stops after five attempts") { KJ_EXPECT(state.outcomes[0] == ActorRetryOutcome::ATTEMPTS_EXHAUSTED); } -KJ_TEST("actor fetch allows an in-flight retry to finish after the start budget") { +KJ_TEST("actor fetch uses the system default retry count when the binding sets none") { + auto checkFiveAttempts = [](kj::Maybe retryPolicy) { + ReplayState state{ + .actions = kj::arr(ReplayAction::AMBIGUOUS, ReplayAction::AMBIGUOUS, ReplayAction::AMBIGUOUS, + ReplayAction::AMBIGUOUS, ReplayAction::AMBIGUOUS), + }; + + KJ_EXPECT(runActorFetch(state, ActorRetryGateEnabled::YES, kj::none, ActorFetchKind::HTTP, + retryPolicy, UserDefinedRetryPolicyEnabled::ENABLED) != kj::none); + KJ_EXPECT(state.requestCount == 5); + KJ_EXPECT(state.retryCount == 4); + KJ_EXPECT(state.observedRetryCount == 4); + KJ_ASSERT(state.outcomes.size() == 1); + KJ_EXPECT(state.outcomes[0] == ActorRetryOutcome::ATTEMPTS_EXHAUSTED); + }; + + // No policy on the binding falls back to the system default. A policy with only defaults set + // behaves the same, so enabling the userland gate changes nothing for existing bindings. + checkFiveAttempts(kj::none); + checkFiveAttempts(UserDefinedRetryPolicy{}); +} + +KJ_TEST("actor fetch honors the configured retry count") { + ReplayState attemptsDisabled{.actions = kj::arr(ReplayAction::AMBIGUOUS)}; + KJ_EXPECT(runActorFetch(attemptsDisabled, ActorRetryGateEnabled::YES, kj::none, + ActorFetchKind::HTTP, UserDefinedRetryPolicy{.maxAttempts = 0}, + UserDefinedRetryPolicyEnabled::ENABLED) != kj::none); + KJ_EXPECT(attemptsDisabled.requestCount == 1); + KJ_EXPECT(attemptsDisabled.retryCount == 0); + + ReplayState oneRetry{.actions = kj::arr(ReplayAction::AMBIGUOUS, ReplayAction::AMBIGUOUS)}; + KJ_EXPECT(runActorFetch(oneRetry, ActorRetryGateEnabled::YES, kj::none, ActorFetchKind::HTTP, + UserDefinedRetryPolicy{.maxAttempts = 1}, + UserDefinedRetryPolicyEnabled::ENABLED) != kj::none); + KJ_EXPECT(oneRetry.requestCount == 2); + KJ_EXPECT(oneRetry.retryCount == 1); +} + +KJ_TEST("configured retry count resets after a redirect") { + ReplayState state{ + .actions = kj::arr(ReplayAction::AMBIGUOUS, ReplayAction::REDIRECT, ReplayAction::AMBIGUOUS), + }; + + KJ_EXPECT(runActorFetch(state, ActorRetryGateEnabled::YES, kj::none, ActorFetchKind::HTTP, + UserDefinedRetryPolicy{.maxAttempts = 1}, + UserDefinedRetryPolicyEnabled::ENABLED) == kj::none); + KJ_EXPECT(state.requestCount == 4); + KJ_EXPECT(state.retryCount == 2); + KJ_ASSERT(state.outcomes.size() == 2); + KJ_EXPECT(state.outcomes[0] == ActorRetryOutcome::RECOVERED); + KJ_EXPECT(state.outcomes[1] == ActorRetryOutcome::RECOVERED); +} + +KJ_TEST("actor fetch cancels an in-flight retry at the retry timeout") { + for (auto kind: {ActorFetchKind::HTTP, ActorFetchKind::WEB_SOCKET}) { + ReplayState state{ + .actions = kj::arr(ReplayAction::AMBIGUOUS, ReplayAction::SLOW_RESPONSE), + .acceptWebSocket = true, + }; + + auto failure = + KJ_REQUIRE_NONNULL(runActorFetch(state, ActorRetryGateEnabled::YES, kj::none, kind)); + KJ_EXPECT(failure.getType() == kj::Exception::Type::DISCONNECTED, failure); + KJ_EXPECT(state.requestCount == 2); + KJ_EXPECT(state.retryCount == 1); + KJ_EXPECT(state.observedRetryCount == 1); + KJ_ASSERT(state.outcomes.size() == 1); + KJ_EXPECT(state.outcomes[0] == ActorRetryOutcome::RETRY_BUDGET_EXHAUSTED); + } +} + +KJ_TEST("actor fetch lets a slow initial attempt finish after the retry timeout") { + ReplayState state{ + .actions = kj::arr(ReplayAction::SLOW_RESPONSE), + }; + + KJ_EXPECT( + runActorFetch(state, ActorRetryGateEnabled::YES, kj::none, ActorFetchKind::HTTP) == kj::none); + KJ_EXPECT(state.requestCount == 1); + KJ_EXPECT(state.retryCount == 0); +} + +KJ_TEST("actor fetch shares the retry timeout across redirects") { + ReplayState state{ + .actions = kj::arr(ReplayAction::SLOW_REDIRECT, ReplayAction::AMBIGUOUS), + }; + + KJ_EXPECT( + runActorFetch(state, ActorRetryGateEnabled::YES, kj::none, ActorFetchKind::HTTP) != kj::none); + KJ_EXPECT(state.requestCount == 2); + KJ_EXPECT(state.retryCount == 0); + KJ_EXPECT(state.observedRetryCount == 0); +} + +KJ_TEST("actor fetch lets a redirected first attempt finish after the retry timeout") { ReplayState state{ - .actions = kj::arr(ReplayAction::AMBIGUOUS, ReplayAction::SLOW_RESPONSE), + .actions = kj::arr(ReplayAction::SLOW_REDIRECT, ReplayAction::SLOW_RESPONSE), }; KJ_EXPECT( runActorFetch(state, ActorRetryGateEnabled::YES, kj::none, ActorFetchKind::HTTP) == kj::none); KJ_EXPECT(state.requestCount == 2); - KJ_EXPECT(state.retryCount == 1); - KJ_EXPECT(state.observedRetryCount == 1); - KJ_ASSERT(state.outcomes.size() == 1); - KJ_EXPECT(state.outcomes[0] == ActorRetryOutcome::RECOVERED); + KJ_EXPECT(state.retryCount == 0); } KJ_TEST("actor fetch does not start a retry after the start budget") { @@ -1058,6 +1254,102 @@ KJ_TEST("actor fetch does not start a retry after the start budget") { KJ_EXPECT(state.outcomes.size() == 0); } +KJ_TEST( + "actor fetch ignores a configured retry timeout while the userland retry gate is disabled") { + ReplayState state{ + .actions = kj::arr(ReplayAction::RETRY_DELAY_EXCEEDS_BUDGET), + .retryTimeoutDelay = 501 * kj::MILLISECONDS, + }; + + KJ_EXPECT(runActorFetch(state, ActorRetryGateEnabled::YES, kj::none, ActorFetchKind::HTTP, + UserDefinedRetryPolicy{.timeout = 500 * kj::MILLISECONDS}, + UserDefinedRetryPolicyEnabled::DISABLED) == kj::none); + KJ_EXPECT(state.requestCount == 2); + KJ_EXPECT(state.retryCount == 1); +} + +KJ_TEST("actor fetch does not start a retry after a configured retry timeout") { + ReplayState state{ + .actions = kj::arr(ReplayAction::RETRY_DELAY_EXCEEDS_BUDGET), + .retryTimeoutDelay = 501 * kj::MILLISECONDS, + }; + + auto failure = KJ_REQUIRE_NONNULL(runActorFetch(state, ActorRetryGateEnabled::YES, kj::none, + ActorFetchKind::HTTP, UserDefinedRetryPolicy{.timeout = 500 * kj::MILLISECONDS}, + UserDefinedRetryPolicyEnabled::ENABLED)); + KJ_EXPECT(failure.getType() == kj::Exception::Type::DISCONNECTED, failure); + KJ_EXPECT(state.requestCount == 1); + KJ_EXPECT(state.acceptedRetryCount == 1); + KJ_EXPECT(state.retryCount == 0); + KJ_EXPECT(state.observedRetryCount == 0); +} + +KJ_TEST("actor fetch lets a slow initial attempt finish after a configured retry timeout") { + ReplayState state{ + .actions = kj::arr(ReplayAction::SLOW_RESPONSE), + }; + + KJ_EXPECT(runActorFetch(state, ActorRetryGateEnabled::YES, kj::none, ActorFetchKind::HTTP, + UserDefinedRetryPolicy{.timeout = 500 * kj::MILLISECONDS}, + UserDefinedRetryPolicyEnabled::ENABLED) == kj::none); + KJ_EXPECT(state.requestCount == 1); + KJ_EXPECT(state.retryCount == 0); +} + +// Runs a stub fetch while the actor's output gate is held for 11 seconds, longer than both the +// default and any configured retry timeout, then releases it. +void runActorFetchAfterLongGateWait(ReplayState& state, UserDefinedRetryPolicy retryPolicy) { + kj::TimerImpl timer(kj::origin()); + DeterministicTimerChannel timerChannel(timer); + TestFixture fixture(TestFixture::SetupParams{ + .actorId = Worker::Actor::Id(kj::str("test-actor")), + .useRealTimers = false, + .ioChannelFactory = kj::Function(TimerChannel&)>( + [&](TimerChannel&) -> kj::Rc { + return kj::rc(timerChannel); + }), + .requestObserverFactory = + kj::Function()>([&]() -> kj::Own { + return kj::refcounted(state); + }), + }); + util::Autogate::initAutogateNamesForTest( + {"durable-object-retries-fetch"_kj, "durable-object-retries-fetch-retry-requests"_kj, + "durable-object-retries-userland"_kj}, + util::IgnoreAllAutogatesEnv::YES); + + fixture.runInIoContext([&](const TestFixture::Environment& env) { + auto gatePaf = kj::newPromiseAndFulfiller(); + auto gateBlocker = + env.context.getActorOrThrow().getOutputGate().lockWhile(kj::mv(gatePaf.promise), nullptr); + auto fetcher = env.js.alloc(env.context.addObject( + kj::heap(state, retryPolicy)), + Fetcher::RequiresHostAndProtocol::YES); + auto promise = fetcher->fetch(env.js, kj::str("http://example.com"), kj::none); + timer.advanceTo(timer.now() + 11 * kj::SECONDS); + gatePaf.fulfiller->fulfill(); + return env.context.awaitJs(env.js, kj::mv(promise)) + .ignoreResult() + .attach(kj::mv(fetcher), kj::mv(gateBlocker), kj::mv(gatePaf.fulfiller)); + }); +} + +KJ_TEST("actor fetch starts the default retry timeout after the output gate wait") { + ReplayState state{.actions = kj::arr(ReplayAction::AMBIGUOUS)}; + runActorFetchAfterLongGateWait(state, UserDefinedRetryPolicy{}); + KJ_EXPECT(state.requestCount == 2); + KJ_EXPECT(state.retryCount == 1); + KJ_EXPECT(state.observedRetryCount == 1); +} + +KJ_TEST("actor fetch starts a configured retry timeout after the output gate wait") { + ReplayState state{.actions = kj::arr(ReplayAction::AMBIGUOUS)}; + runActorFetchAfterLongGateWait(state, UserDefinedRetryPolicy{.timeout = 500 * kj::MILLISECONDS}); + KJ_EXPECT(state.requestCount == 2); + KJ_EXPECT(state.retryCount == 1); + KJ_EXPECT(state.observedRetryCount == 1); +} + KJ_TEST("replica actor fetch retries a request-level disconnect on its primary channel") { ReplayState state{.actions = kj::arr(ReplayAction::AMBIGUOUS)}; kj::TimerImpl timer(kj::origin()); @@ -1122,8 +1414,15 @@ KJ_TEST("GlobalActorOutgoingFactory forwards metadata and recreates channels for 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")}, ActorCallRetriesAllowed::YES, Persistent::NO); + ActorVersion{.cohort = kj::str("cohort")}, ActorCallRetriesAllowed::YES, Persistent::NO, + UserDefinedRetryPolicy{ + .maxAttempts = 2, + .timeout = 500 * kj::MILLISECONDS, + }); KJ_EXPECT(factory.supportsActorCallRetries()); + auto retryPolicy = KJ_ASSERT_NONNULL(factory.getUserDefinedRetryPolicy()); + KJ_EXPECT(retryPolicy.maxAttempts == 2); + KJ_EXPECT(retryPolicy.timeout == 500 * kj::MILLISECONDS); auto client = factory.newActorCallAttempt(kj::none, ActorCallRetryState::Attempt( @@ -1171,7 +1470,7 @@ KJ_TEST("GlobalActorOutgoingFactory forwards metadata and recreates channels for GlobalActorOutgoingFactory::ChannelIdOrFactory(static_cast(1)), env.js.alloc(kj::heap()), kj::none, ActorGetMode::GET_OR_CREATE, false, ActorRoutingMode::DEFAULT, kj::none, - ActorCallRetriesAllowed::NO, Persistent::NO); + ActorCallRetriesAllowed::NO, Persistent::NO, kj::none); KJ_EXPECT(!retriesDisallowedFactory.supportsActorCallRetries()); }); } diff --git a/src/workerd/api/actor.c++ b/src/workerd/api/actor.c++ index ba78d466ebb..6ddc6026e93 100644 --- a/src/workerd/api/actor.c++ +++ b/src/workerd/api/actor.c++ @@ -275,12 +275,12 @@ jsg::Ref DurableObjectNamespace::getImpl(jsg::Lock& js, KJ_CASE_ONEOF(channelId, uint) { outgoingFactory = kj::heap(channelId, id.addRef(), kj::mv(locationHint), mode, enableReplicaRouting, routingMode, kj::mv(version), - actorCallRetriesAllowed, persistent); + actorCallRetriesAllowed, persistent, userDefinedRetryPolicy); } KJ_CASE_ONEOF(channelFactory, IoOwn) { outgoingFactory = kj::heap(kj::addRef(*channelFactory), id.addRef(), kj::mv(locationHint), mode, enableReplicaRouting, routingMode, - kj::mv(version), actorCallRetriesAllowed, persistent); + kj::mv(version), actorCallRetriesAllowed, persistent, userDefinedRetryPolicy); } } @@ -299,13 +299,13 @@ jsg::Ref DurableObjectNamespace::jurisdiction( // inherits the `persistent` bit. KJ_SWITCH_ONEOF(channel) { KJ_CASE_ONEOF(channelId, uint) { - return js.alloc( - channelId, kj::mv(newIdFactory), actorCallRetriesAllowed, persistent); + return js.alloc(channelId, kj::mv(newIdFactory), + actorCallRetriesAllowed, persistent, userDefinedRetryPolicy); } KJ_CASE_ONEOF(channelFactory, IoOwn) { return js.alloc( IoContext::current().addObject(kj::addRef(*channelFactory)), kj::mv(newIdFactory), - actorCallRetriesAllowed, persistent); + actorCallRetriesAllowed, persistent, userDefinedRetryPolicy); } } diff --git a/src/workerd/api/actor.h b/src/workerd/api/actor.h index cea0102dced..e92996f056b 100644 --- a/src/workerd/api/actor.h +++ b/src/workerd/api/actor.h @@ -182,23 +182,29 @@ class DurableObjectNamespace: public jsg::Object { // `persistent` indicates whether stubs minted from this namespace may be stored in long-term // storage. It is `Persistent::YES` only for `ctx.exports` self-bindings of a worker that has // `allow_irrevocable_stub_storage` enabled (see `LoopbackDurableObjectNamespace`); regular env - // bindings leave it `Persistent::NO`. + // bindings leave it `Persistent::NO`. `userDefinedRetryPolicy` is the binding's configured retry + // policy, applied to minted stubs when user-defined retry policies are enabled; none means the + // runtime's default applies. DurableObjectNamespace(uint channel, kj::Own idFactory, ActorCallRetriesAllowed actorCallRetriesAllowed, - Persistent persistent = Persistent::NO) + Persistent persistent, + kj::Maybe userDefinedRetryPolicy) : channel(channel), idFactory(kj::mv(idFactory)), actorCallRetriesAllowed(actorCallRetriesAllowed), - persistent(persistent) {} + persistent(persistent), + userDefinedRetryPolicy(userDefinedRetryPolicy) {} DurableObjectNamespace(IoOwn factory, kj::Own idFactory, ActorCallRetriesAllowed actorCallRetriesAllowed, - Persistent persistent = Persistent::NO) + Persistent persistent, + kj::Maybe userDefinedRetryPolicy) : channel(kj::mv(factory)), idFactory(kj::mv(idFactory)), actorCallRetriesAllowed(actorCallRetriesAllowed), - persistent(persistent) {} + persistent(persistent), + userDefinedRetryPolicy(userDefinedRetryPolicy) {} struct NewUniqueIdOptions { // Restricts the new unique ID to a set of colos within a jurisdiction. @@ -325,6 +331,8 @@ class DurableObjectNamespace: public jsg::Object { // See doc comment on the constructor. Persistent persistent; + kj::Maybe userDefinedRetryPolicy; + jsg::Ref getImpl(jsg::Lock& js, ActorGetMode mode, jsg::Ref id, @@ -343,7 +351,8 @@ class GlobalActorOutgoingFactory final: public Fetcher::OutgoingFactory { ActorRoutingMode routingMode, kj::Maybe version, ActorCallRetriesAllowed actorCallRetriesAllowed, - Persistent persistent) + Persistent persistent, + kj::Maybe userDefinedRetryPolicy) : channelIdOrFactory(kj::mv(channelIdOrFactory)), id(kj::mv(id)), locationHint(kj::mv(locationHint)), @@ -352,13 +361,17 @@ class GlobalActorOutgoingFactory final: public Fetcher::OutgoingFactory { routingMode(routingMode), version(kj::mv(version)), actorCallRetriesAllowed(actorCallRetriesAllowed), - persistent(persistent) {} + persistent(persistent), + userDefinedRetryPolicy(userDefinedRetryPolicy) {} Result newSingleUseClient( kj::Maybe cfStr, MakeUserSpanParent makeUserSpanParent) override; kj::Maybe getActorTargetRetryability() const override { return ActorCallTargetRetryable(actorCallRetriesAllowed.toBool()); } + kj::Maybe getUserDefinedRetryPolicy() const override { + return userDefinedRetryPolicy; + } void onActorCallRetry() override; Result newActorCallAttempt(kj::Maybe cfStr, ActorCallRetryState::Attempt attempt, @@ -382,6 +395,8 @@ class GlobalActorOutgoingFactory final: public Fetcher::OutgoingFactory { // `persistent` field of `DurableObjectNamespace`. Persistent persistent; + kj::Maybe userDefinedRetryPolicy; + kj::Maybe> actorChannel; // Registered when actorChannel is lazily created, to reflect the cost of holding an open diff --git a/src/workerd/api/export-loopback.h b/src/workerd/api/export-loopback.h index 5f8210112f7..56ef6ce3fe4 100644 --- a/src/workerd/api/export-loopback.h +++ b/src/workerd/api/export-loopback.h @@ -163,11 +163,13 @@ class LoopbackDurableObjectNamespace: public DurableObjectNamespace { kj::Own idFactory, ActorCallRetriesAllowed actorCallRetriesAllowed, jsg::Ref loopbackClass, - CompatibilityFlags::Reader featureFlags) + CompatibilityFlags::Reader featureFlags, + kj::Maybe userDefinedRetryPolicy) : DurableObjectNamespace(nsChannel, kj::mv(idFactory), actorCallRetriesAllowed, - Persistent(featureFlags.getAllowIrrevocableStubStorage())), + Persistent(featureFlags.getAllowIrrevocableStubStorage()), + userDefinedRetryPolicy), loopbackClass(kj::mv(loopbackClass)) {} // getClass() accessor for use from C++ only. diff --git a/src/workerd/api/http.c++ b/src/workerd/api/http.c++ index 2b851c2d604..0f36649fd59 100644 --- a/src/workerd/api/http.c++ +++ b/src/workerd/api/http.c++ @@ -1478,6 +1478,47 @@ struct ActorFetchFailure { template using ActorFetchAttemptResult = kj::OneOf; +ActorRetryGateEnabled fetchRetryObservationEnabled() { + return ActorRetryGateEnabled( + util::Autogate::isEnabled(util::AutogateKey::DURABLE_OBJECT_RETRIES_FETCH)); +} + +ActorRetryGateEnabled fetchRetryEnforcementEnabled() { + return ActorRetryGateEnabled( + util::Autogate::isEnabled(util::AutogateKey::DURABLE_OBJECT_RETRIES_FETCH_RETRY_REQUESTS)); +} + +// The retry policy a fetch through `fetcher` runs under, or none if the target does not support +// actor call retries. A binding's own policy applies only when it configured one and every retry +// gate is on: the fetch gates are prerequisites for all retry behavior, and the userland gate for +// reading the binding's configuration. Otherwise the runtime's default applies. The gates are +// checked before touching the stub. Its I/O objects belong to the context that created it, so +// reaching them can throw, and a disabled rollout must never get that far. +kj::Maybe tryGetActorRetryPolicy(Fetcher& fetcher) { + if (!fetcher.supportsActorCallRetries()) return kj::none; + if (fetchRetryObservationEnabled().toBool() && fetchRetryEnforcementEnabled().toBool() && + util::Autogate::isEnabled(util::AutogateKey::DURABLE_OBJECT_RETRIES_USERLAND)) { + KJ_IF_SOME(userPolicy, fetcher.getUserDefinedRetryPolicy()) { + return ActorRetryPolicy::userDefined(userPolicy); + } + } + return ActorRetryPolicy::systemDefault(); +} + +kj::Maybe> makeActorCallRetryState( + Fetcher& fetcher, Request& request, kj::TimePoint callStart) { + auto policy = KJ_UNWRAP_OR_RETURN(tryGetActorRetryPolicy(fetcher), kj::none); + auto& context = IoContext::current(); + auto config = ActorCallRetryState::Config{ + .callType = ActorRetryCallType::FETCH, + .observationEnabled = fetchRetryObservationEnabled(), + .enforcementEnabled = fetchRetryEnforcementEnabled(), + .payloadReplayable = ActorCallPayloadReplayable(request.canRewindBody()), + }; + return kj::rc( + context.getIoChannelFactory().getTimer(), context.getMetrics(), config, policy, callStart); +} + template kj::Promise> captureActorFetchAttempt(jsg::Lock& js, kj::Maybe>& signal, @@ -1504,17 +1545,21 @@ jsg::Promise> fetchImplNoOutputLockAttempt(jsg::Lock& js, kj::Vector urlList, kj::Maybe> retryState); +// `callStart` is the actor call's start, carried so that a redirect shares the call's retry +// timeout. It is none for fetches that are not actor calls. jsg::Promise> handleHttpResponse(jsg::Lock& js, jsg::Ref fetcher, jsg::Ref jsRequest, kj::Vector urlList, - kj::HttpClient::Response&& response); + kj::HttpClient::Response&& response, + kj::Maybe callStart); jsg::Promise> handleHttpRedirectResponse(jsg::Lock& js, jsg::Ref fetcher, jsg::Ref jsRequest, kj::Vector urlList, uint status, - kj::StringPtr location); + kj::StringPtr location, + kj::Maybe callStart); jsg::Promise> handleWebSocketFetchResponse(jsg::Lock& js, jsg::Ref fetcher, @@ -1522,12 +1567,13 @@ jsg::Promise> handleWebSocketFetchResponse(jsg::Lock& js, kj::Vector urlList, kj::Own client, kj::Maybe> signal, - kj::HttpClient::WebSocketResponse response) { + kj::HttpClient::WebSocketResponse response, + kj::Maybe callStart) { 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)}); + {response.statusCode, response.statusText, response.headers, kj::mv(body)}, callStart); } KJ_CASE_ONEOF(webSocket, kj::Own) { KJ_ASSERT(response.statusCode == 101); @@ -1555,7 +1601,8 @@ jsg::Promise> handleHttpFetchResponse(jsg::Lock& js, kj::Vector urlList, kj::Own client, kj::Maybe& traceContext, - kj::HttpClient::Response response) { + kj::HttpClient::Response response, + kj::Maybe callStart) { response.body = response.body.attach(kj::mv(client)); KJ_IF_SOME(ctx, traceContext) { ctx.setTag("http.response.status_code"_kjc, static_cast(response.statusCode)); @@ -1568,7 +1615,7 @@ jsg::Promise> handleHttpFetchResponse(jsg::Lock& js, } } return handleHttpResponse( - js, kj::mv(fetcher), kj::mv(jsRequest), kj::mv(urlList), kj::mv(response)); + js, kj::mv(fetcher), kj::mv(jsRequest), kj::mv(urlList), kj::mv(response), callStart); } jsg::Promise> rejectFetch(jsg::Lock& js, kj::Exception&& exception) { @@ -1618,27 +1665,16 @@ jsg::Promise> retryActorFetch(jsg::Lock& js, }); } +// `callStart` is set when a redirect continues an earlier call. A new call starts now, after any +// output-gate wait. jsg::Promise> fetchImplNoOutputLock(jsg::Lock& js, jsg::Ref fetcher, jsg::Ref jsRequest, - kj::Vector urlList) { - kj::Maybe> retryState; - if (fetcher->supportsActorCallRetries()) { - auto& context = IoContext::current(); - auto observationEnabled = ActorRetryGateEnabled( - util::Autogate::isEnabled(util::AutogateKey::DURABLE_OBJECT_RETRIES_FETCH)); - auto enforcementEnabled = ActorRetryGateEnabled(util::Autogate::isEnabled( - util::AutogateKey::DURABLE_OBJECT_RETRIES_FETCH_RETRY_REQUESTS)); - retryState = kj::rc(context.getIoChannelFactory().getTimer(), - context.getMetrics(), - ActorCallRetryState::Config{ - .callType = ActorRetryCallType::FETCH, - .observationEnabled = observationEnabled, - .enforcementEnabled = enforcementEnabled, - .payloadReplayable = ActorCallPayloadReplayable(jsRequest->canRewindBody()), - }); - } - + kj::Vector urlList, + kj::Maybe callStart = kj::none) { + auto retryState = makeActorCallRetryState(*fetcher, *jsRequest, + callStart.orDefault( + [] { return IoContext::current().getIoChannelFactory().getTimer().nowForLimitTimeout(); })); return fetchImplNoOutputLockAttempt( js, kj::mv(fetcher), kj::mv(jsRequest), kj::mv(urlList), kj::mv(retryState)); } @@ -1668,6 +1704,7 @@ jsg::Promise> fetchImplNoOutputLockAttempt(jsg::Lock& js, actorCallAttempt = kj::mv(KJ_ASSERT_NONNULL(attemptOrException.tryGet())); } + auto callStart = retryState.map([](auto& state) { return state->getCallStart(); }); // Stash the payload and target retryability before we lose access to the JS-level request. This // is consumed only when the target is an actor; for other fetches the value is overwritten by the @@ -1760,18 +1797,18 @@ jsg::Promise> fetchImplNoOutputLockAttempt(jsg::Lock& js, KJ_IF_SOME(state, retryState) { if (state->isRetryEnabled()) { auto resultPromise = captureActorFetchAttempt( - js, signal, kj::mv(webSocketResponse), state.addRef()); + js, signal, state->enforceRetryTimeout(kj::mv(webSocketResponse)), state.addRef()); return ioContext.awaitIo(js, kj::mv(resultPromise), [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, + retryState = kj::mv(retryState), callStart](jsg::Lock& js, ActorFetchAttemptResult&& result) mutable -> jsg::Promise> { KJ_SWITCH_ONEOF(result) { KJ_CASE_ONEOF(response, kj::HttpClient::WebSocketResponse) { KJ_ASSERT_NONNULL(retryState)->recordRecovered(); return handleWebSocketFetchResponse(js, kj::mv(fetcher), kj::mv(jsRequest), - kj::mv(urlList), kj::mv(client), kj::mv(signal), kj::mv(response)); + kj::mv(urlList), kj::mv(client), kj::mv(signal), kj::mv(response), callStart); } KJ_CASE_ONEOF(failure, ActorFetchFailure) { return retryActorFetch(js, kj::mv(fetcher), kj::mv(jsRequest), kj::mv(urlList), @@ -1785,10 +1822,10 @@ jsg::Promise> fetchImplNoOutputLockAttempt(jsg::Lock& js, webSocketResponse = AbortSignal::maybeCancelWrap(js, signal, kj::mv(webSocketResponse)); 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)]( + client = kj::mv(client), signal = kj::mv(signal), callStart]( 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)); + kj::mv(client), kj::mv(signal), kj::mv(response), callStart); }); } else { kj::Maybe nativeRequest; @@ -1859,19 +1896,19 @@ jsg::Promise> fetchImplNoOutputLockAttempt(jsg::Lock& js, }); KJ_IF_SOME(state, retryState) { if (state->isRetryEnabled()) { - auto resultPromise = - captureActorFetchAttempt(js, signal, kj::mv(responsePromise), state.addRef()); + auto resultPromise = captureActorFetchAttempt( + js, signal, state->enforceRetryTimeout(kj::mv(responsePromise)), state.addRef()); 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)]( + traceContext = kj::mv(traceContext), retryState = kj::mv(retryState), callStart]( jsg::Lock& js, ActorFetchAttemptResult&& result) mutable -> jsg::Promise> { KJ_SWITCH_ONEOF(result) { KJ_CASE_ONEOF(response, kj::HttpClient::Response) { KJ_ASSERT_NONNULL(retryState)->recordRecovered(); return handleHttpFetchResponse(js, kj::mv(fetcher), kj::mv(jsRequest), - kj::mv(urlList), kj::mv(client), traceContext, kj::mv(response)); + kj::mv(urlList), kj::mv(client), traceContext, kj::mv(response), callStart); } KJ_CASE_ONEOF(failure, ActorFetchFailure) { return retryActorFetch(js, kj::mv(fetcher), kj::mv(jsRequest), kj::mv(urlList), @@ -1885,10 +1922,10 @@ jsg::Promise> fetchImplNoOutputLockAttempt(jsg::Lock& js, responsePromise = AbortSignal::maybeCancelWrap(js, signal, kj::mv(responsePromise)); 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)]( + client = kj::mv(client), traceContext = kj::mv(traceContext), callStart]( 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)); + kj::mv(client), traceContext, kj::mv(response), callStart); }); } } @@ -1915,7 +1952,8 @@ jsg::Promise> handleHttpResponse(jsg::Lock& js, jsg::Ref fetcher, jsg::Ref jsRequest, kj::Vector urlList, - kj::HttpClient::Response&& response) { + kj::HttpClient::Response&& response, + kj::Maybe callStart) { auto signal = jsRequest->getSignal(); KJ_IF_SOME(s, signal) { @@ -1936,9 +1974,9 @@ jsg::Promise> handleHttpResponse(jsg::Lock& js, return ioContext.awaitIo(js, response.body->pumpTo(getGlobalNullOutputStream()).ignoreResult().attach(kj::mv(response.body)), [fetcher = kj::mv(fetcher), jsRequest = kj::mv(jsRequest), urlList = kj::mv(urlList), - status = response.statusCode, location = kj::str(l)](jsg::Lock& js) mutable { - return handleHttpRedirectResponse( - js, kj::mv(fetcher), kj::mv(jsRequest), kj::mv(urlList), status, kj::mv(location)); + status = response.statusCode, location = kj::str(l), callStart](jsg::Lock& js) mutable { + return handleHttpRedirectResponse(js, kj::mv(fetcher), kj::mv(jsRequest), kj::mv(urlList), + status, kj::mv(location), callStart); }); } else { // No Location header. That's OK, we just return the response as is. @@ -1958,7 +1996,8 @@ jsg::Promise> handleHttpRedirectResponse(jsg::Lock& js, jsg::Ref jsRequest, kj::Vector urlList, uint status, - kj::StringPtr location) { + kj::StringPtr location, + kj::Maybe callStart) { // Reconstruct the request body stream for retransmission in the face of a redirect. Before // reconstructing the stream, however, this function: // @@ -2080,7 +2119,10 @@ jsg::Promise> handleHttpRedirectResponse(jsg::Lock& js, // No need to wait for output locks again when following a redirect, because we didn't interact // with the app state in any way. - return fetchImplNoOutputLock(js, kj::mv(fetcher), kj::mv(jsRequest), kj::mv(urlList)); + // + // The redirected request is a new actor call with fresh retry state, but it keeps the original + // callStart so one retry timeout covers the whole chain. + return fetchImplNoOutputLock(js, kj::mv(fetcher), kj::mv(jsRequest), kj::mv(urlList), callStart); } } // namespace @@ -2826,6 +2868,15 @@ kj::Maybe Fetcher::getActorTargetRetryability() { return kj::none; } +kj::Maybe Fetcher::getUserDefinedRetryPolicy() { + auto& outgoingFactory = KJ_REQUIRE_NONNULL( + channelOrClientFactory.tryGet>(), + "actor retry policy requested from an unsupported Fetcher"); + KJ_REQUIRE(outgoingFactory->supportsActorCallRetries(), + "actor retry policy requested from an unsupported Fetcher"); + return outgoingFactory->getUserDefinedRetryPolicy(); +} + void Fetcher::onActorCallRetry() { auto& outgoingFactory = KJ_REQUIRE_NONNULL( channelOrClientFactory.tryGet>(), diff --git a/src/workerd/api/http.h b/src/workerd/api/http.h index 279a6e16fcf..cc4d406eb88 100644 --- a/src/workerd/api/http.h +++ b/src/workerd/api/http.h @@ -268,6 +268,12 @@ class Fetcher: public JsRpcClientProvider { KJ_FAIL_REQUIRE("actor call retry requested from an unsupported Fetcher"); } + // The retry policy configured on the binding this factory was minted from, if any. None means + // the runtime's default applies. + virtual kj::Maybe getUserDefinedRetryPolicy() const { + return kj::none; + } + // Factories that support actor call retries override this method. The default rejects the // attempt rather than silently starting a new logical call. virtual Result newActorCallAttempt(kj::Maybe cfStr, @@ -349,6 +355,7 @@ class Fetcher: public JsRpcClientProvider { MakeUserSpanParent makeUserSpanParent); kj::Maybe getActorTargetRetryability() override; + kj::Maybe getUserDefinedRetryPolicy(); void onActorCallRetry() override; // Get a SubrequestChannel representing this Fetcher. diff --git a/src/workerd/api/worker-rpc.c++ b/src/workerd/api/worker-rpc.c++ index b72113f9776..ddb778974c8 100644 --- a/src/workerd/api/worker-rpc.c++ +++ b/src/workerd/api/worker-rpc.c++ @@ -946,14 +946,15 @@ JsRpcRetrySetup setupJsRpcRetries(IoContext& ioContext, result.replayMemoryTracker = kj::refcounted( ioContext.getMetrics().trackActorCallReplayMemory(replayMemoryBytes)); } - result.state = kj::rc(ioContext.getIoChannelFactory().getTimer(), - ioContext.getMetrics(), + auto& timer = ioContext.getIoChannelFactory().getTimer(); + result.state = kj::rc(timer, ioContext.getMetrics(), ActorCallRetryState::Config{ .callType = ActorRetryCallType::JSRPC, .observationEnabled = ActorRetryGateEnabled::YES, .enforcementEnabled = enforcementEnabled, .payloadReplayable = ActorCallPayloadReplayable::YES, - }); + }, + ActorRetryPolicy::systemDefault(), timer.nowForLimitTimeout()); auto attemptOrException = KJ_ASSERT_NONNULL(result.state)->startAttempt(); result.attempt = kj::mv(KJ_ASSERT_NONNULL(attemptOrException.tryGet())); diff --git a/src/workerd/io/io-channels.h b/src/workerd/io/io-channels.h index 83427848bb7..5ea9f4593c9 100644 --- a/src/workerd/io/io-channels.h +++ b/src/workerd/io/io-channels.h @@ -88,6 +88,14 @@ class TimerChannel { virtual kj::TimePoint nowForLimitTimeout() { return kj::systemPreciseMonotonicClock().now(); } + + // Returns a promise that resolves at an absolute deadline on the same clock as + // nowForLimitTimeout(). + virtual kj::Promise atLimitTimeout(kj::TimePoint deadline) { + auto now = nowForLimitTimeout(); + if (deadline <= now) return kj::READY_NOW; + return afterLimitTimeout(deadline - now); + } }; class WorkerStubChannel; diff --git a/src/workerd/server/server-test.c++ b/src/workerd/server/server-test.c++ index 2590947218b..ccfffd38517 100644 --- a/src/workerd/server/server-test.c++ +++ b/src/workerd/server/server-test.c++ @@ -2624,6 +2624,43 @@ KJ_TEST("Server: named images and directory snapshots are not startup sources") conn.httpGet200("/", "Container failed to start; Docker requests: 0"); } +KJ_TEST("Server: rejects Durable Object retry policy values outside the system limits") { + TestServer test(singleWorker(R"(( + compatibilityDate = "2024-10-01", + modules = [ + ( name = "worker", + esModule = `export default { fetch() { return new Response("ok"); } } + ) + ], + bindings = [ + ( name = "attempts", + durableObjectNamespace = ( + className = "MyActorClass", + retryPolicy = (maxAttempts = 11), + ) + ), + ( name = "shortTimeout", + durableObjectNamespace = ( + className = "MyActorClass", + retryPolicy = (timeoutMs = 499), + ) + ), + ( name = "longTimeout", + durableObjectNamespace = ( + className = "MyActorClass", + retryPolicy = (timeoutMs = 60001), + ) + ), + ], + ))"_kj)); + + test.expectErrors(R"( + service hello: Worker "hello"'s binding "attempts" has a Durable Object retry policy outside the system limits. + service hello: Worker "hello"'s binding "shortTimeout" has a Durable Object retry policy outside the system limits. + service hello: Worker "hello"'s binding "longTimeout" has a Durable Object retry policy outside the system limits. + )"_blockquote); +} + KJ_TEST("Server: call queue handler on service binding") { TestServer test(R"(( services = [ diff --git a/src/workerd/server/server.c++ b/src/workerd/server/server.c++ index 46503150f92..6fbb3464a28 100644 --- a/src/workerd/server/server.c++ +++ b/src/workerd/server/server.c++ @@ -4862,6 +4862,24 @@ static kj::Maybe createBinding(kj::StringPtr workerName, case config::Worker::Binding::DURABLE_OBJECT_NAMESPACE: { auto actorBinding = binding.getDurableObjectNamespace(); + kj::Maybe userDefinedRetryPolicy; + if (actorBinding.hasRetryPolicy()) { + auto retryPolicy = actorBinding.getRetryPolicy(); + auto retryTimeout = retryPolicy.getTimeoutMs() * kj::MILLISECONDS; + if (retryPolicy.getMaxAttempts() > api::UserDefinedRetryPolicy::MAX_CONFIGURABLE_ATTEMPTS || + retryTimeout < api::UserDefinedRetryPolicy::MIN_CONFIGURABLE_TIMEOUT || + retryTimeout > api::UserDefinedRetryPolicy::MAX_CONFIGURABLE_TIMEOUT) { + errorReporter.addError(kj::str(errorContext, + " has a Durable Object retry policy outside " + "the system limits.")); + return kj::none; + } + userDefinedRetryPolicy = api::UserDefinedRetryPolicy{ + .maxAttempts = retryPolicy.getMaxAttempts(), + .timeout = retryTimeout, + }; + } + const Server::ActorConfig* actorConfig; if (actorBinding.hasServiceName()) { auto& svcMap = KJ_UNWRAP_OR(actorConfigs.find(actorBinding.getServiceName()), { @@ -4894,7 +4912,10 @@ static kj::Maybe createBinding(kj::StringPtr workerName, KJ_SWITCH_ONEOF(*actorConfig) { KJ_CASE_ONEOF(durable, Server::Durable) { return makeGlobal(Global::DurableActorNamespace{ - .actorChannel = channel, .uniqueKey = durable.uniqueKey}); + .actorChannel = channel, + .uniqueKey = durable.uniqueKey, + .userDefinedRetryPolicy = userDefinedRetryPolicy, + }); } KJ_CASE_ONEOF(_, Server::Ephemeral) { return makeGlobal(Global::EphemeralActorNamespace{.actorChannel = channel}); diff --git a/src/workerd/server/workerd-api.c++ b/src/workerd/server/workerd-api.c++ index 4fe08e6d436..fae456785f6 100644 --- a/src/workerd/server/workerd-api.c++ +++ b/src/workerd/server/workerd-api.c++ @@ -668,13 +668,15 @@ static v8::Local createBindingValue(JsgWorkerdIsolate::Lock& lock, KJ_CASE_ONEOF(ns, Global::DurableActorNamespace) { value = lock.wrap(context, lock.alloc(ns.actorChannel, - kj::heap(ns.uniqueKey), api::ActorCallRetriesAllowed::YES)); + kj::heap(ns.uniqueKey), api::ActorCallRetriesAllowed::YES, + Persistent::NO, ns.userDefinedRetryPolicy)); } KJ_CASE_ONEOF(ns, Global::LoopbackDurableActorNamespace) { value = lock.wrap(context, lock.alloc(ns.actorChannel, kj::heap(ns.uniqueKey), api::ActorCallRetriesAllowed::YES, - lock.alloc(ns.classChannel), featureFlags)); + lock.alloc(ns.classChannel), featureFlags, + /*userDefinedRetryPolicy=*/kj::none)); } KJ_CASE_ONEOF(ae, Global::AnalyticsEngine) { diff --git a/src/workerd/server/workerd-api.h b/src/workerd/server/workerd-api.h index b9689918bbe..83833083e1d 100644 --- a/src/workerd/server/workerd-api.h +++ b/src/workerd/server/workerd-api.h @@ -4,6 +4,7 @@ #pragma once +#include #include #include #include @@ -200,6 +201,7 @@ class WorkerdApi final: public Worker::Api { struct DurableActorNamespace { uint actorChannel; kj::StringPtr uniqueKey; + kj::Maybe userDefinedRetryPolicy; DurableActorNamespace clone() const { return *this; diff --git a/src/workerd/server/workerd.capnp b/src/workerd/server/workerd.capnp index 051316ef48b..cf3fef0ee88 100644 --- a/src/workerd/server/workerd.capnp +++ b/src/workerd/server/workerd.capnp @@ -536,6 +536,25 @@ struct Worker { # # (This is intentionally not a ServiceDesignator because you cannot choose an alternate # entrypoint here; the class name IS the entrypoint.) + + retryPolicy @2 :RetryPolicy; + # Limits on how the runtime retries calls through stubs minted from this binding. When + # absent, the runtime's default retry behavior applies. + + struct RetryPolicy { + maxAttempts @0 :UInt32 = 4; + # Maximum number of retries after the initial attempt. Zero disables retries, and one + # allows a single retry. The default matches the runtime's default of five attempts in + # total. + + timeoutMs @1 :UInt32 = 10000; + # Time in milliseconds, measured from the start of the call, after which no retry may + # start. A retry still running when it expires is cancelled, and the caller gets the + # error that caused the first retry. The initial request, and the first request after each + # redirect, always run to completion. The clock starts after any output-gate wait, and a + # redirect shares the original call's timeout. Must be between 500 and 60,000. The default + # matches the runtime's default. + } } struct CryptoKey { diff --git a/src/workerd/util/autogate.h b/src/workerd/util/autogate.h index eb9cb3d87ac..5bcef443aae 100644 --- a/src/workerd/util/autogate.h +++ b/src/workerd/util/autogate.h @@ -99,6 +99,8 @@ namespace workerd::util { /* Enables Durable Object JSRPC retry requests. Requires both fetch retry gates and the JSRPC \ observe gate. */ \ V(DURABLE_OBJECT_RETRIES_JSRPC_RETRY_REQUESTS) \ + /* Enables user-configured Durable Object retry policy and @retryable dispatch behavior. */ \ + V(DURABLE_OBJECT_RETRIES_USERLAND) \ /* When enabled, the native `node-internal:url` module is provided by the Rust \ implementation (api::node UrlUtil ported to src/rust/api) instead of the \ C++ implementation. The C++ implementation is retained for rollback.*/ \