Move retry claim before actor construction

Claim retry tokens in WorkerEntrypoint before delivered() constructs the actor. Keep the hook synchronous so capability pipelining is unchanged.

Update the test to pin the claim-before-delivery ordering.

Author: apeacock1991

src/workerd/api/BUILD.bazel

@@ -812,14 +812,6 @@ kj_test(
     ],
 )
 
-kj_test(
-    src = "fetch-retry-claim-hook-test.c++",
-    deps = [
-        "//src/workerd/io",
-        "//src/workerd/tests:test-fixture",
-    ],
-)
-
 wd_test(
     src = "streams/streams-test.wd-test",
     args = ["--experimental"],

src/workerd/api/fetch-retry-claim-hook-test.c++

@@ -378,11 +378,6 @@ kj::Promise<DeferredProxy<void>> ServiceWorkerGlobalScope::request(kj::HttpMetho
   bool useDefaultHandling;
   KJ_IF_SOME(h, exportedHandler) {
     KJ_IF_SOME(f, h.fetch) {
-      // Immediately before dispatching into user code, give the observer a chance to claim the
-      // request's retry-token nonce. No-op unless the request is to an actor and the observer
-      // overrides the hook. Only the exported-handler path is hooked: Durable Objects are always
-      // class-based, so actor fetches never take the service-worker dispatchEventImpl() path below.
-      ioContext.getMetrics().claimRetryTokenBeforeUserCode();
       auto promise = f(lock, event->getRequest(), h.env.addRef(js), h.getCtx());
       event->respondWith(lock, kj::mv(promise));
       useDefaultHandling = false;

src/workerd/api/global-scope.c++

@@ -143,14 +143,9 @@ class RequestObserver: public kj::Refcounted {
   // the base observer; edgeworker overrides it to feed retry classification.
   virtual void setNextSubrequestBodyRewindable(SubrequestBodyRewindable bodyRewindable) {}
 
-  // Fired immediately before an actor fetch dispatches into user code, so an observer can claim the
-  // request's retry-token nonce against the actor's claim store. No-op in the base observer;
-  // edgeworker overrides it. Gating and fetch-only scoping are the override's concern.
-  //
-  // This is only fired on the exported-handler (ES modules) dispatch path, not the service-worker
-  // addEventListener('fetch') path. That is deliberate, not an omission: Durable Objects must be
-  // class-based and so are always invoked via an exported handler, meaning an actor fetch never
-  // reaches the service-worker path -- where this hook would be a guaranteed no-op anyway.
+  // Fired before a fetch request is delivered, so an observer can claim an actor request's
+  // retry-token nonce before actor construction or user code. This also fires for non-actor and
+  // service-worker fetches; observers are responsible for treating those as no-ops.
   virtual void claimRetryTokenBeforeUserCode() {}
 
   // Used to record when a worker has used a dynamic dispatch binding.

src/workerd/io/observer.h

@@ -80,12 +80,34 @@ class TestResponse final: public kj::HttpService::Response {
       kj::StringPtr statusText,
       const kj::HttpHeaders& headers,
       kj::Maybe<uint64_t> expectedBodySize) override {
+    this->statusCode = statusCode;
     return kj::heap<kj::NullStream>();
   }
 
   kj::Own<kj::WebSocket> acceptWebSocket(const kj::HttpHeaders& headers) override {
     KJ_FAIL_ASSERT("request unexpectedly returned a WebSocket");
   }
+
+  uint statusCode = 0;
+};
+
+class RetryClaimObserver final: public RequestObserver {
+ public:
+  void claimRetryTokenBeforeUserCode() override {
+    KJ_EXPECT(stage == 0);
+    stage = 1;
+    if (rejectClaim) {
+      kj::throwFatalException(KJ_EXCEPTION(FAILED, "claim rejected"));
+    }
+  }
+
+  void delivered() override {
+    KJ_EXPECT(stage == 1);
+    stage = 2;
+  }
+
+  uint stage = 0;
+  bool rejectClaim = false;
 };
 
 class ThrowingResponse final: public kj::HttpService::Response {
@@ -195,6 +217,71 @@ class RecordingObserver final: public RequestObserver, public WorkerInterface {
   kj::Maybe<WorkerInterface&> inner;
 };
 
+KJ_TEST("retry claim fires synchronously before fetch delivery") {
+  auto observer = kj::refcounted<RetryClaimObserver>();
+  TestFixture fixture(TestFixture::SetupParams{
+    .mainModuleSource = R"SCRIPT(
+        export default {
+          async fetch() {
+            return new Response("OK");
+          },
+        };
+      )SCRIPT"_kj,
+    .requestObserverFactory = kj::Function<kj::Own<RequestObserver>()>(
+        [&observer]() -> kj::Own<RequestObserver> { return kj::addRef(*observer); }),
+  });
+
+  auto entrypoint = fixture.makeWorkerEntrypoint();
+  kj::HttpHeaderTable headerTable;
+  kj::HttpHeaders headers(headerTable);
+  kj::NullStream requestBody;
+  TestResponse response;
+
+  auto request = entrypoint->request(
+      kj::HttpMethod::GET, "https://example.com", headers, requestBody, response);
+  KJ_EXPECT(observer->stage == 2, "claim and delivery hooks did not fire synchronously");
+  request.wait(fixture.getWaitScope());
+
+  KJ_EXPECT(response.statusCode == 200);
+  KJ_EXPECT(observer->stage == 2, "claim hook fired more than once");
+}
+
+KJ_TEST("rejected retry claim prevents actor construction and delivery") {
+  auto observer = kj::refcounted<RetryClaimObserver>();
+  observer->rejectClaim = true;
+  TestFixture fixture(TestFixture::SetupParams{
+    .mainModuleSource = R"SCRIPT(
+        export default class {
+          constructor() {
+            throw new Error("actor was constructed");
+          }
+
+          async fetch() {
+            return new Response("unexpected");
+          }
+        }
+      )SCRIPT"_kj,
+    .actorId = Worker::Actor::Id(kj::str("retry-claim-test")),
+    .requestObserverFactory = kj::Function<kj::Own<RequestObserver>()>(
+        [&observer]() -> kj::Own<RequestObserver> { return kj::addRef(*observer); }),
+  });
+
+  auto entrypoint = fixture.makeWorkerEntrypoint();
+  kj::HttpHeaderTable headerTable;
+  kj::HttpHeaders headers(headerTable);
+  kj::NullStream requestBody;
+  TestResponse response;
+
+  auto exception = kj::runCatchingExceptions([&]() {
+    entrypoint->request(kj::HttpMethod::GET, "https://example.com", headers, requestBody, response)
+        .wait(fixture.getWaitScope());
+  });
+
+  KJ_EXPECT(KJ_ASSERT_NONNULL(exception).getDescription().contains("claim rejected"));
+  KJ_EXPECT(observer->stage == 1, "request was delivered after its retry claim was rejected");
+  KJ_EXPECT(response.statusCode == 0);
+}
+
 KJ_TEST("connect pass-through tags failures after delivery") {
   capnp::MallocMessageBuilder flagsMessage;
   auto flags = flagsMessage.initRoot<CompatibilityFlags>();

src/workerd/io/worker-entrypoint-test.c++

@@ -399,6 +399,9 @@ kj::Promise<void> WorkerEntrypoint::requestImpl(kj::HttpMethod method,
     workerTracer = t;
   }
 
+  // Claim before delivered() constructs an actor. This introduces no asynchronous boundary, so
+  // capability pipelining remains unchanged.
+  incomingRequest->getMetrics().claimRetryTokenBeforeUserCode();
   incomingRequest->delivered();
 
   auto metricsForCatch = kj::addRef(incomingRequest->getMetrics());