STOR-5489: Measure Durable Object RPC replay memory

src/workerd/api/BUILD.bazel

@@ -841,6 +841,14 @@ kj_test(
     ],
 )
 
+kj_test(
+    src = "jsrpc-call-observation-test.c++",
+    deps = [
+        "//src/workerd/io",
+        "//src/workerd/tests:test-fixture",
+    ],
+)
+
 kj_test(
     src = "fetch-retry-claim-hook-test.c++",
     deps = [

src/workerd/api/actor-call-retry.h

@@ -15,7 +15,6 @@
 
 namespace workerd::api {
 
-WD_STRONG_BOOL(ActorCallPayloadReplayable);
 WD_STRONG_BOOL(ActorCallRetriesAllowed);
 WD_STRONG_BOOL(IsFirstActorCallAttempt);
 

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

@@ -87,8 +87,8 @@ class RetryMetadataOutgoingFactory final: public Fetcher::OutgoingFactory {
     return {.client = kj::heap<MockFetchTarget>(), .spanParents = kj::none};
   }
 
-  bool supportsActorCallRetries() const override {
-    return true;
+  kj::Maybe<ActorCallTargetRetryable> getActorTargetRetryability() const override {
+    return ActorCallTargetRetryable::YES;
   }
 
   void onActorCallRetry() override {}
@@ -333,8 +333,8 @@ class ReplayOutgoingFactory final: public Fetcher::OutgoingFactory {
     return {.client = kj::heap<ReplayFetchTarget>(state), .spanParents = kj::none};
   }
 
-  bool supportsActorCallRetries() const override {
-    return state.retriesAllowed.toBool();
+  kj::Maybe<ActorCallTargetRetryable> getActorTargetRetryability() const override {
+    return ActorCallTargetRetryable(state.retriesAllowed.toBool());
   }
 
   void onActorCallRetry() override {

src/workerd/api/actor-state.c++

@@ -1010,6 +1010,10 @@ class FacetOutgoingFactory final: public Fetcher::OutgoingFactory {
     return {.client = kj::mv(client), .spanParents = kj::mv(spanParents)};
   }
 
+  kj::Maybe<ActorCallTargetRetryable> getActorTargetRetryability() const override {
+    return ActorCallTargetRetryable::NO;
+  }
+
   kj::Own<IoChannelFactory::SubrequestChannel> getSubrequestChannel() override {
     return kj::addRef(getOrCreateActorChannel());
   }

src/workerd/api/actor.h

@@ -356,8 +356,8 @@ class GlobalActorOutgoingFactory final: public Fetcher::OutgoingFactory {
 
   Result newSingleUseClient(
       kj::Maybe<kj::String> cfStr, MakeUserSpanParent makeUserSpanParent) override;
-  bool supportsActorCallRetries() const override {
-    return actorCallRetriesAllowed.toBool();
+  kj::Maybe<ActorCallTargetRetryable> getActorTargetRetryability() const override {
+    return ActorCallTargetRetryable(actorCallRetriesAllowed.toBool());
   }
   void onActorCallRetry() override;
   Result newActorCallAttempt(kj::Maybe<kj::String> cfStr,
@@ -399,6 +399,9 @@ class LocalActorOutgoingFactory final: public Fetcher::OutgoingFactory {
 
   Result newSingleUseClient(
       kj::Maybe<kj::String> cfStr, MakeUserSpanParent makeUserSpanParent) override;
+  kj::Maybe<ActorCallTargetRetryable> getActorTargetRetryability() const override {
+    return ActorCallTargetRetryable::NO;
+  }
   kj::Own<IoChannelFactory::SubrequestChannel> getSubrequestChannel() override;
 
  private:
@@ -425,8 +428,8 @@ class ReplicaActorOutgoingFactory final: public Fetcher::OutgoingFactory {
 
   Result newSingleUseClient(
       kj::Maybe<kj::String> cfStr, MakeUserSpanParent makeUserSpanParent) override;
-  bool supportsActorCallRetries() const override {
-    return true;
+  kj::Maybe<ActorCallTargetRetryable> getActorTargetRetryability() const override {
+    return ActorCallTargetRetryable::YES;
   }
   void onActorCallRetry() override {
     // Keep the pre-resolved primary channel on retries. Reconnecting a broken channel requires

src/workerd/api/bench-jsrpc-sender.c++

@@ -90,8 +90,8 @@ class LocalOutgoingFactory final: public Fetcher::OutgoingFactory {
     return {.client = receiver.makeWorkerEntrypoint(), .spanParents = kj::none};
   }
 
-  bool supportsActorCallRetries() const override {
-    return true;
+  kj::Maybe<ActorCallTargetRetryable> getActorTargetRetryability() const override {
+    return ActorCallTargetRetryable::YES;
   }
 
  private:
@@ -155,13 +155,13 @@ class JsRpcSenderHarness {
     int64_t externalWhilePending = 0;
     kj::Maybe<kj::Promise<void>> pending;
     sender->enterContext(*request, [&](const TestFixture::Environment& env) {
-      externalBefore = static_cast<int64_t>(env.js.v8Isolate->GetExternalMemory());
+      externalBefore = env.js.v8Isolate->GetExternalMemory();
       auto promises = kj::heapArrayBuilder<kj::Promise<void>>(callCount);
       for (size_t i = 0; i < callCount; ++i) {
         promises.add(call(env, KJ_ASSERT_NONNULL(accept), ArgumentKind::PLAIN));
       }
       pending = kj::joinPromises(promises.finish());
-      externalWhilePending = static_cast<int64_t>(env.js.v8Isolate->GetExternalMemory());
+      externalWhilePending = env.js.v8Isolate->GetExternalMemory();
     });
     auto allocatedWhilePending = currentAllocatedBytes();
     callGate.release(callCount);

src/workerd/api/fetch-body-rewindable-test.c++

@@ -14,20 +14,22 @@
 namespace workerd::api {
 namespace {
 
-// Records, in call order, every body-rewindable value passed to setNextSubrequestRetryEligibility().
+struct RetryEligibility {
+  bool bodyRewindable;
+  ActorCallTargetRetryable targetRetryable;
+};
+
 class RecordingRequestObserver final: public RequestObserver {
  public:
-  RecordingRequestObserver(kj::Vector<bool>& calls): calls(calls) {}
+  RecordingRequestObserver(kj::Vector<RetryEligibility>& calls): calls(calls) {}
 
   void setNextSubrequestRetryEligibility(
       SubrequestBodyRewindable bodyRewindable, ActorCallTargetRetryable targetRetryable) override {
-    // A service binding is not a retryable actor target.
-    KJ_EXPECT(targetRetryable == ActorCallTargetRetryable::NO);
-    calls.add(bodyRewindable.toBool());
+    calls.add(RetryEligibility{bodyRewindable.toBool(), targetRetryable});
   }
 
  private:
-  kj::Vector<bool>& calls;
+  kj::Vector<RetryEligibility>& calls;
 };
 
 // Minimal WorkerInterface that answers every outgoing request() with an empty 200, draining the
@@ -76,17 +78,12 @@ struct FetchTargetIoChannelFactory final: public TestFixture::DummyIoChannelFact
   }
 };
 
-// fetchImplNoOutputLock forwards Request::canRewindBody() and the target's retry support 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
-// shared across every outgoing subrequest in an IoContext, so the value set for one call must not
-// carry over into the next. We issue two fetches in one invocation -- a rewindable (buffered) body
-// then a non-rewindable (stream) body -- to exercise that shared observer across consecutive calls
-// and verify the per-body mapping, the per-call sequencing, and the absence of stale attribution all
-// at once (the no-staleness behaviour can only be observed across more than one fetch).
-KJ_TEST("fetch reports each outgoing body's rewindability per-call without staleness") {
-  kj::Vector<bool> bodyRewindableCalls;
+// fetchImplNoOutputLock forwards payload and target retryability to RequestObserver so edgeworker
+// can classify disconnected outgoing actor calls. The stashed signal is per-call, not sticky: one
+// RequestObserver is shared across every outgoing subrequest in an IoContext, so the value set for
+// one call must not carry over into the next. Two fetches exercise consecutive values.
+KJ_TEST("fetch reports each outgoing call's retry eligibility without staleness") {
+  kj::Vector<RetryEligibility> calls;
 
   TestFixture fixture(TestFixture::SetupParams{
     .mainModuleSource = R"SCRIPT(
@@ -109,19 +106,20 @@ KJ_TEST("fetch reports each outgoing body's rewindability per-call without stale
   }),
     .requestObserverFactory =
         kj::Function<kj::Own<RequestObserver>()>([&]() -> kj::Own<RequestObserver> {
-    return kj::refcounted<RecordingRequestObserver>(bodyRewindableCalls);
+    return kj::refcounted<RecordingRequestObserver>(calls);
   }),
   });
 
   auto result =
       fixture.runRequest(kj::HttpMethod::POST, "http://www.example.com"_kj, "incoming-body"_kj);
   KJ_EXPECT(result.statusCode == 200);
 
-  KJ_ASSERT(bodyRewindableCalls.size() == 2,
-      "expected exactly one rewindability signal per outgoing fetch");
-  KJ_EXPECT(bodyRewindableCalls[0] == true, "buffered request body should be rewindable");
-  KJ_EXPECT(bodyRewindableCalls[1] == false,
-      "streamed request body should not be rewindable (no carryover)");
+  KJ_ASSERT(calls.size() == 2, "expected exactly one retry eligibility signal per outgoing fetch");
+  KJ_EXPECT(calls[0].bodyRewindable, "buffered request body should be rewindable");
+  KJ_EXPECT(
+      !calls[1].bodyRewindable, "streamed request body should not be rewindable (no carryover)");
+  KJ_EXPECT(calls[0].targetRetryable == ActorCallTargetRetryable::NO);
+  KJ_EXPECT(calls[1].targetRetryable == ActorCallTargetRetryable::NO);
 }
 
 }  // namespace

src/workerd/api/http.c++

@@ -1669,15 +1669,13 @@ jsg::Promise<jsg::Ref<Response>> fetchImplNoOutputLockAttempt(jsg::Lock& js,
         kj::mv(KJ_ASSERT_NONNULL(attemptOrException.tryGet<ActorCallRetryState::Attempt>()));
   }
 
-  // Stash whether this request's body can be rewound (and so the request re-sent) and whether the
-  // target supports retries, before we lose access to the JS-level request. This is currently
-  // consumed only when the target is an actor (Durable Object), to classify retry eligibility for
-  // disconnected calls; for other fetches the values are simply overwritten by the next call and
-  // never read. The set->getClientWithTracing->wrap*SubrequestClient sequence is synchronous, so
-  // there is no stale-attribution risk.
+  // 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
+  // next call. The set->getClientWithTracing->wrap*SubrequestClient sequence is synchronous.
+  auto targetRetryable = fetcher->getActorTargetRetryability()
+                             .orDefault(ActorCallTargetRetryable::NO);
   ioContext.getMetrics().setNextSubrequestRetryEligibility(
-      SubrequestBodyRewindable(jsRequest->canRewindBody()),
-      ActorCallTargetRetryable(fetcher->supportsActorCallRetries()));
+      SubrequestBodyRewindable(jsRequest->canRewindBody()), targetRetryable);
 
   // Get client and trace context (if needed) in one clean call.
   auto cfBlobJson = jsRequest->serializeCfBlobJson(js);
@@ -2317,7 +2315,8 @@ kj::LiteralStringConst Fetcher::getRpcTargetKind() {
   return "fetcher"_kjc;
 }
 
-JsRpcClientProvider::ClientForOneCall Fetcher::getClientForOneCall(jsg::Lock& js) {
+JsRpcClientProvider::ClientForOneCall Fetcher::getClientForOneCall(
+    jsg::Lock& js, kj::Maybe<ActorCallRetryState::Attempt> actorCallAttempt) {
   auto& ioContext = IoContext::current();
 
   kj::Maybe<TraceContext> callSpan;
@@ -2327,17 +2326,28 @@ JsRpcClientProvider::ClientForOneCall Fetcher::getClientForOneCall(jsg::Lock& js
     // The "jsRpcSession" trace context is attached to the customEvent task below so it covers the
     // whole session. The first jsRpcCall span is opened before the session client so its user span
     // can also become the callee invocation's parent.
-    clientWithTracing = buildClient(ioContext, kj::none, "jsRpcSession"_kjc,
-        [&](TraceContext& sessionSpan) -> kj::Maybe<SpanParent> {
+    auto makeUserSpanParent = [&](TraceContext& sessionSpan) -> kj::Maybe<SpanParent> {
       callSpan = sessionSpan.getSpanParents().newChild("jsRpcCall"_kjc);
       return KJ_ASSERT_NONNULL(callSpan).getUserSpanParent();
-    });
+    };
+    KJ_IF_SOME(attempt, actorCallAttempt) {
+      clientWithTracing = getClientForActorCallAttempt(ioContext, kj::none, "jsRpcSession"_kjc,
+          kj::mv(attempt), kj::mv(makeUserSpanParent));
+    } else {
+      clientWithTracing =
+          buildClient(ioContext, kj::none, "jsRpcSession"_kjc, kj::mv(makeUserSpanParent));
+    }
     callSpanParents = clientWithTracing.traceContext.map(
         [](TraceContext& tc) { return tc.getSpanParents(); });
   } else {
-    clientWithTracing = ClientWithTracing{
-      .client = getClient(ioContext, kj::none, "jsRpcSession"_kjc),
-    };
+    KJ_IF_SOME(attempt, actorCallAttempt) {
+      clientWithTracing = getClientForActorCallAttempt(
+          ioContext, kj::none, "jsRpcSession"_kjc, kj::mv(attempt));
+    } else {
+      clientWithTracing = ClientWithTracing{
+        .client = getClient(ioContext, kj::none, "jsRpcSession"_kjc),
+      };
+    }
   }
   auto worker = kj::mv(clientWithTracing.client);
   auto event = kj::heap<api::JsRpcSessionCustomEvent>(JsRpcSessionCustomEvent::WORKER_RPC_EVENT_TYPE);
@@ -2717,8 +2727,20 @@ Fetcher::ClientWithTracing Fetcher::getClientForActorCallAttempt(IoContext& ioCo
     kj::Maybe<kj::String> cfStr,
     kj::ConstString operationName,
     ActorCallRetryState::Attempt attempt) {
+  return getClientForActorCallAttempt(ioContext, kj::mv(cfStr), kj::mv(operationName),
+      kj::mv(attempt), [](TraceContext& traceContext) -> kj::Maybe<SpanParent> {
+    return traceContext.getUserSpanParent();
+  });
+}
+
+Fetcher::ClientWithTracing Fetcher::getClientForActorCallAttempt(IoContext& ioContext,
+    kj::Maybe<kj::String> cfStr,
+    kj::ConstString operationName,
+    ActorCallRetryState::Attempt attempt,
+    MakeUserSpanParent makeUserSpanParent) {
   if (!attempt.hasMetadata()) {
-    return buildClient(ioContext, kj::mv(cfStr), kj::mv(operationName));
+    return buildClient(
+        ioContext, kj::mv(cfStr), kj::mv(operationName), kj::mv(makeUserSpanParent));
   }
 
   auto& outgoingFactory = KJ_REQUIRE_NONNULL(
@@ -2736,7 +2758,7 @@ Fetcher::ClientWithTracing Fetcher::getClientForActorCallAttempt(IoContext& ioCo
       [&](TraceContext& outerTraceContext) -> kj::Maybe<SpanParent> {
     if (!outerTraceContext.isObserved()) return kj::none;
     traceContext = outerTraceContext.getSpanParents().newChild(operationName.clone());
-    return KJ_ASSERT_NONNULL(traceContext).getUserSpanParent();
+    return makeUserSpanParent(KJ_ASSERT_NONNULL(traceContext));
   });
   return ClientWithTracing{kj::mv(result.client), kj::mv(traceContext)};
 }
@@ -2797,11 +2819,11 @@ Fetcher::ClientWithTracing Fetcher::buildClient(IoContext& ioContext,
   KJ_UNREACHABLE;
 }
 
-bool Fetcher::supportsActorCallRetries() {
+kj::Maybe<ActorCallTargetRetryable> Fetcher::getActorTargetRetryability() {
   KJ_IF_SOME(outgoingFactory, channelOrClientFactory.tryGet<IoOwn<OutgoingFactory>>()) {
-    return outgoingFactory->supportsActorCallRetries();
+    return outgoingFactory->getActorTargetRetryability();
   }
-  return false;
+  return kj::none;
 }
 
 void Fetcher::onActorCallRetry() {

src/workerd/api/http.h

@@ -254,8 +254,14 @@ class Fetcher: public JsRpcClientProvider {
     virtual Result newSingleUseClient(
         kj::Maybe<kj::String> cfStr, MakeUserSpanParent makeUserSpanParent) = 0;
 
-    virtual bool supportsActorCallRetries() const {
-      return false;
+    // Whether this factory dispatches to a Durable Object, and whether that target can create fresh
+    // retry attempts. Actor calls are observed whether or not the target supports retries.
+    virtual kj::Maybe<ActorCallTargetRetryable> getActorTargetRetryability() const {
+      return kj::none;
+    }
+
+    bool supportsActorCallRetries() const {
+      return getActorTargetRetryability().orDefault(ActorCallTargetRetryable::NO).toBool();
     }
 
     virtual void onActorCallRetry() {
@@ -336,7 +342,13 @@ class Fetcher: public JsRpcClientProvider {
       kj::ConstString operationName,
       ActorCallRetryState::Attempt attempt);
 
-  bool supportsActorCallRetries() override;
+  [[nodiscard]] ClientWithTracing getClientForActorCallAttempt(IoContext& ioContext,
+      kj::Maybe<kj::String> cfStr,
+      kj::ConstString operationName,
+      ActorCallRetryState::Attempt attempt,
+      MakeUserSpanParent makeUserSpanParent);
+
+  kj::Maybe<ActorCallTargetRetryable> getActorTargetRetryability() override;
   void onActorCallRetry();
 
   // Get a SubrequestChannel representing this Fetcher.
@@ -430,7 +442,8 @@ class Fetcher: public JsRpcClientProvider {
     return getRpcMethod(js, kj::mv(name));
   }
 
-  ClientForOneCall getClientForOneCall(jsg::Lock& js) override;
+  ClientForOneCall getClientForOneCall(
+      jsg::Lock& js, kj::Maybe<ActorCallRetryState::Attempt> actorCallAttempt) override;
 
   kj::LiteralStringConst getRpcTargetKind() override;