STOR-5615: Add configurable Durable Object retries

src/workerd/api/actor-call-retry-test.c++

@@ -56,14 +56,18 @@ class RecordingObserver final: public RequestObserver {
   kj::Vector<ActorRetryOutcome> outcomes;
 };
 
-kj::Rc<ActorCallRetryState> newRetryState(TestTimerChannel& timer, RecordingObserver& observer) {
+// A retry state with every gate enabled and a replayable payload.
+kj::Rc<ActorCallRetryState> newRetryState(TestTimerChannel& timer,
+    RecordingObserver& observer,
+    ActorRetryPolicy policy = ActorRetryPolicy::systemDefault()) {
   return kj::rc<ActorCallRetryState>(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<RecordingObserver>();
   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<kj::Duration>());
     } else {
-      KJ_EXPECT(result.is<kj::Exception>());
+      auto& failure = KJ_ASSERT_NONNULL(result.tryGet<kj::Exception>());
+      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<RecordingObserver>();
+    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<kj::Duration>());
+      startAttempt(*state);
+      ++attemptCount;
+    }
+
+    KJ_EXPECT(handleFailure(*state, makeDisconnect("disconnected"_kj)).is<kj::Exception>());
+    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<RecordingObserver>();
+  auto state = newRetryState(timer, *observer,
+      ActorRetryPolicy::userDefined(UserDefinedRetryPolicy{.timeout = 500 * kj::MILLISECONDS}));
+
+  startAttempt(*state);
+  KJ_EXPECT(handleFailure(*state, makeDisconnect("original disconnect"_kj)).is<kj::Duration>());
+  startAttempt(*state);
+  timer.advance(500 * kj::MILLISECONDS);
+
+  auto result = handleFailure(*state, makeDisconnect("second disconnect"_kj));
+  auto& failure = KJ_ASSERT_NONNULL(result.tryGet<kj::Exception>());
+  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<RecordingObserver>();
@@ -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);

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,21 +34,28 @@ ActorCallRetryState::~ActorCallRetryState() noexcept(false) {
   }
 }
 
+IoChannelFactory::ActorRetryRequestMetadata ActorCallRetryState::freshMetadata() const {
+  return generateActorRetryRequestMetadata(
+      kj::systemCoarseCalendarClock().now(), config.enforcementEnabled);
+}
+
 kj::OneOf<ActorCallRetryState::Attempt, kj::Exception> 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);
   }
   return Attempt(metadata, isFirstAttempt);
 }
 
+kj::Exception ActorCallRetryState::retryTimeoutExpired() {
+  recordOutcome(ActorRetryOutcome::RETRY_BUDGET_EXHAUSTED);
+  return KJ_ASSERT_NONNULL(originalDisconnect).clone();
+}
+
 kj::OneOf<kj::Duration, kj::Exception> ActorCallRetryState::handleAttemptFailure(
     kj::Exception exception) {
   KJ_REQUIRE(retriesEnabled);
@@ -98,28 +105,22 @@ kj::OneOf<kj::Duration, kj::Exception> 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<uint64_t> distribution(0, maximum / kj::NANOSECONDS);
   return distribution(generator) * kj::NANOSECONDS;
 }

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<Attempt, kj::Exception> 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 <typename T>
+  kj::Promise<T> enforceRetryTimeout(kj::Promise<T> promise) {
+    if (attemptCount == 1) return kj::mv(promise);
+    auto timeout = timer.atLimitTimeout(retryCutoff())
+                       .then([self = addRefToThis()]() mutable -> kj::Promise<T> {
+      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<kj::Exception> handleClaimRejection(const kj::Exception& exception);
   kj::OneOf<kj::Duration, kj::Exception> checkCanRetry(kj::Exception exception);
   kj::Duration retryDelay();
@@ -89,9 +166,10 @@ class ActorCallRetryState final: public kj::Refcounted {
   TimerChannel& timer;
   kj::Own<RequestObserver> observer;
   Config config;
+  ActorRetryPolicy policy;
+  kj::TimePoint callStart;
   bool retriesEnabled;
   kj::Maybe<IoChannelFactory::ActorRetryRequestMetadata> metadata;
-  kj::Maybe<kj::TimePoint> deadline;
   kj::Maybe<kj::Exception> originalDisconnect;
   kj::Maybe<kj::TimePoint> retryStartTime;
   uint attemptCount = 1;