Skip to content

Commit 38a5707

Browse files
apeacock1991apeacock1991
committed
STOR-5489: Exempt actor fetch replays from subrequest limits
Count the initial Durable Object fetch against the per-invocation subrequest budget, but bypass the check for physical replay attempts. This preserves one budget unit per logical fetch even when delivery is retried.
1 parent 69efd7a commit 38a5707

7 files changed

Lines changed: 171 additions & 67 deletions

File tree

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

Lines changed: 105 additions & 39 deletions
Original file line numberDiff line numberDiff line change
@@ -77,7 +77,8 @@ class RetryMetadataOutgoingFactory final: public Fetcher::OutgoingFactory {
7777
}
7878

7979
kj::Own<WorkerInterface> newSingleUseClientWithActorRetryMetadata(kj::Maybe<kj::String>,
80-
kj::Maybe<IoChannelFactory::ActorRetryRequestMetadata> actorRetryRequestMetadata) override {
80+
kj::Maybe<IoChannelFactory::ActorRetryRequestMetadata> actorRetryRequestMetadata,
81+
IsActorFetchRetryAttempt isRetryAttempt) override {
8182
capturedMetadata = kj::mv(actorRetryRequestMetadata);
8283
return kj::heap<MockFetchTarget>();
8384
}
@@ -100,6 +101,7 @@ struct ReplayState {
100101
bool acceptWebSocket = false;
101102
kj::Maybe<kj::Own<kj::WebSocket>> acceptedWebSocket;
102103
kj::Vector<IoChannelFactory::ActorRetryRequestMetadata> metadata;
104+
kj::Vector<IsActorFetchRetryAttempt> retryAttempts;
103105
kj::Vector<kj::Array<kj::byte>> requestBodies;
104106
uint requestCount = 0;
105107
uint webSocketRequestCount = 0;
@@ -133,11 +135,9 @@ class ReplayFetchTarget final: public WorkerInterface {
133135
exception.setDetail(
134136
jsg::REQUEST_NOT_DELIVERED_TO_ACTOR_DETAIL_ID, kj::heapArray<kj::byte>(0));
135137
} else if (failure == ReplayFailure::DELIVERED) {
136-
exception.setDetail(
137-
jsg::REQUEST_DELIVERED_TO_ACTOR_DETAIL_ID, kj::heapArray<kj::byte>(0));
138+
exception.setDetail(jsg::REQUEST_DELIVERED_TO_ACTOR_DETAIL_ID, kj::heapArray<kj::byte>(0));
138139
} else if (failure == ReplayFailure::CLAIM_REJECTED) {
139-
exception.setDetail(
140-
jsg::ACTOR_RETRY_CLAIM_REJECTED_DETAIL_ID, kj::heapArray<kj::byte>(0));
140+
exception.setDetail(jsg::ACTOR_RETRY_CLAIM_REJECTED_DETAIL_ID, kj::heapArray<kj::byte>(0));
141141
}
142142
kj::throwRecoverableException(kj::mv(exception));
143143
}
@@ -192,8 +192,10 @@ class ReplayOutgoingFactory final: public Fetcher::OutgoingFactory {
192192
}
193193

194194
kj::Own<WorkerInterface> newSingleUseClientWithActorRetryMetadata(kj::Maybe<kj::String>,
195-
kj::Maybe<IoChannelFactory::ActorRetryRequestMetadata> actorRetryRequestMetadata) override {
195+
kj::Maybe<IoChannelFactory::ActorRetryRequestMetadata> actorRetryRequestMetadata,
196+
IsActorFetchRetryAttempt isRetryAttempt) override {
196197
state.metadata.add(KJ_REQUIRE_NONNULL(actorRetryRequestMetadata));
198+
state.retryAttempts.add(isRetryAttempt);
197199
return kj::heap<ReplayFetchTarget>(state);
198200
}
199201

@@ -220,8 +222,7 @@ kj::Maybe<kj::Exception> runActorFetch(ReplayState& state,
220222
});
221223
if (enforcement == RetryEnforcement::ENABLED) {
222224
util::Autogate::initAutogateNamesForTest(
223-
{"durable-object-retries-fetch"_kj,
224-
"durable-object-retries-fetch-retry-requests"_kj},
225+
{"durable-object-retries-fetch"_kj, "durable-object-retries-fetch-retry-requests"_kj},
225226
util::IgnoreAllAutogatesEnv::YES);
226227
} else {
227228
util::Autogate::initAutogateNamesForTest(
@@ -230,8 +231,8 @@ kj::Maybe<kj::Exception> runActorFetch(ReplayState& state,
230231
kj::Maybe<kj::Exception> failure;
231232

232233
fixture.runInIoContext([&](const TestFixture::Environment& env) {
233-
auto fetcher = env.js.alloc<Fetcher>(env.context.addObject<Fetcher::OutgoingFactory>(
234-
kj::heap<ReplayOutgoingFactory>(state)),
234+
auto fetcher = env.js.alloc<Fetcher>(
235+
env.context.addObject<Fetcher::OutgoingFactory>(kj::heap<ReplayOutgoingFactory>(state)),
235236
Fetcher::RequiresHostAndProtocol::YES);
236237
RequestInitializerDict init;
237238
KJ_IF_SOME(value, body) {
@@ -248,8 +249,9 @@ kj::Maybe<kj::Exception> runActorFetch(ReplayState& state,
248249
auto promise = fetcher->fetch(env.js, kj::str("http://example.com"), kj::mv(init));
249250
return env.context.awaitJs(env.js, kj::mv(promise))
250251
.ignoreResult()
251-
.catch_([&](kj::Exception&& exception) { failure.emplace(kj::mv(exception)); })
252-
.attach(kj::mv(fetcher));
252+
.catch_([&](kj::Exception&& exception) {
253+
failure.emplace(kj::mv(exception));
254+
}).attach(kj::mv(fetcher));
253255
});
254256

255257
return failure;
@@ -314,6 +316,28 @@ class RecordingActorChannel final: public IoChannelFactory::ActorChannel {
314316
kj::Maybe<IoChannelFactory::ActorRetryRequestMetadata>& capturedMetadata;
315317
};
316318

319+
class ReplayActorChannel final: public IoChannelFactory::ActorChannel {
320+
public:
321+
ReplayActorChannel(ReplayState& state): state(state) {}
322+
323+
kj::Own<WorkerInterface> startRequest(IoChannelFactory::SubrequestMetadata metadata) override {
324+
state.metadata.add(KJ_REQUIRE_NONNULL(metadata.actorRetryRequestMetadata));
325+
return kj::heap<ReplayFetchTarget>(state);
326+
}
327+
328+
void requireAllowsTransfer() override {
329+
KJ_UNIMPLEMENTED("not used in this test");
330+
}
331+
332+
kj::OneOf<kj::Array<byte>, kj::Promise<kj::Array<byte>>> getTokenMaybeSync(
333+
IoChannelFactory::ChannelTokenUsage) override {
334+
KJ_UNIMPLEMENTED("not used in this test");
335+
}
336+
337+
private:
338+
ReplayState& state;
339+
};
340+
317341
struct ActorIoChannelFactory final: public TestFixture::DummyIoChannelFactory {
318342
ActorIoChannelFactory(TimerChannel& timer,
319343
kj::Maybe<IoChannelFactory::ActorRetryRequestMetadata>& capturedMetadata,
@@ -420,11 +444,8 @@ KJ_TEST("fetch omits actor retry metadata for an unsupported outgoing factory")
420444
}
421445

422446
KJ_TEST("actor fetch updates retry metadata and rewinds the body") {
423-
ReplayState state{.failures = kj::arr(
424-
ReplayFailure::NOT_DELIVERED,
425-
ReplayFailure::AMBIGUOUS,
426-
ReplayFailure::NOT_DELIVERED
427-
)};
447+
ReplayState state{.failures = kj::arr(ReplayFailure::NOT_DELIVERED, ReplayFailure::AMBIGUOUS,
448+
ReplayFailure::NOT_DELIVERED)};
428449
KJ_EXPECT(runActorFetch(state, RetryEnforcement::ENABLED, "request body"_kj,
429450
ActorFetchKind::HTTP) == kj::none);
430451

@@ -438,6 +459,11 @@ KJ_TEST("actor fetch updates retry metadata and rewinds the body") {
438459
KJ_EXPECT(state.metadata[1].isRetry == IsActorRetry::NO);
439460
KJ_EXPECT(state.metadata[2].isRetry == IsActorRetry::YES);
440461
KJ_EXPECT(state.metadata[3].isRetry == IsActorRetry::YES);
462+
KJ_ASSERT(state.retryAttempts.size() == 4);
463+
KJ_EXPECT(state.retryAttempts[0] == IsActorFetchRetryAttempt::NO);
464+
KJ_EXPECT(state.retryAttempts[1] == IsActorFetchRetryAttempt::YES);
465+
KJ_EXPECT(state.retryAttempts[2] == IsActorFetchRetryAttempt::YES);
466+
KJ_EXPECT(state.retryAttempts[3] == IsActorFetchRetryAttempt::YES);
441467
KJ_ASSERT(state.requestBodies.size() == 4);
442468
for (auto& body: state.requestBodies) {
443469
KJ_EXPECT(body == "request body"_kj.asBytes());
@@ -447,17 +473,17 @@ KJ_TEST("actor fetch updates retry metadata and rewinds the body") {
447473
KJ_TEST("actor fetch does not retry when enforcement is disabled") {
448474
ReplayState state{.failures = kj::arr(ReplayFailure::AMBIGUOUS)};
449475

450-
KJ_EXPECT(runActorFetch(
451-
state, RetryEnforcement::DISABLED, kj::none, ActorFetchKind::HTTP) != kj::none);
476+
KJ_EXPECT(
477+
runActorFetch(state, RetryEnforcement::DISABLED, kj::none, ActorFetchKind::HTTP) != kj::none);
452478
KJ_EXPECT(state.requestCount == 1);
453479
KJ_EXPECT(state.retryCount == 0);
454480
}
455481

456482
KJ_TEST("actor fetch does not retry a delivered disconnect") {
457483
ReplayState state{.failures = kj::arr(ReplayFailure::DELIVERED)};
458484

459-
KJ_EXPECT(runActorFetch(
460-
state, RetryEnforcement::ENABLED, kj::none, ActorFetchKind::HTTP) != kj::none);
485+
KJ_EXPECT(
486+
runActorFetch(state, RetryEnforcement::ENABLED, kj::none, ActorFetchKind::HTTP) != kj::none);
461487
KJ_EXPECT(state.requestCount == 1);
462488
KJ_EXPECT(state.retryCount == 0);
463489
}
@@ -492,8 +518,8 @@ KJ_TEST("actor WebSocket fetch retries a disconnected handshake") {
492518
.acceptWebSocket = true,
493519
};
494520

495-
KJ_EXPECT(runActorFetch(
496-
state, RetryEnforcement::ENABLED, kj::none, ActorFetchKind::WEB_SOCKET) == kj::none);
521+
KJ_EXPECT(runActorFetch(state, RetryEnforcement::ENABLED, kj::none, ActorFetchKind::WEB_SOCKET) ==
522+
kj::none);
497523
KJ_EXPECT(state.requestCount == 2);
498524
KJ_EXPECT(state.webSocketRequestCount == 2);
499525
KJ_EXPECT(state.retryCount == 1);
@@ -510,8 +536,8 @@ KJ_TEST("actor fetch honors an abort before retrying") {
510536
});
511537

512538
fixture.runInIoContext([&](const TestFixture::Environment& env) {
513-
auto fetcher = env.js.alloc<Fetcher>(env.context.addObject<Fetcher::OutgoingFactory>(
514-
kj::heap<ReplayOutgoingFactory>(state)),
539+
auto fetcher = env.js.alloc<Fetcher>(
540+
env.context.addObject<Fetcher::OutgoingFactory>(kj::heap<ReplayOutgoingFactory>(state)),
515541
Fetcher::RequiresHostAndProtocol::YES);
516542
auto controller = AbortController::constructor(env.js);
517543
RequestInitializerDict init;
@@ -520,8 +546,9 @@ KJ_TEST("actor fetch honors an abort before retrying") {
520546
controller->abort(env.js, kj::none);
521547
return env.context.awaitJs(env.js, kj::mv(promise))
522548
.ignoreResult()
523-
.catch_([&](kj::Exception&& exception) { failure.emplace(kj::mv(exception)); })
524-
.attach(kj::mv(fetcher), kj::mv(controller));
549+
.catch_([&](kj::Exception&& exception) {
550+
failure.emplace(kj::mv(exception));
551+
}).attach(kj::mv(fetcher), kj::mv(controller));
525552
});
526553

527554
auto& exception = KJ_REQUIRE_NONNULL(failure);
@@ -536,8 +563,8 @@ KJ_TEST("actor fetch stops after five attempts") {
536563
ReplayFailure::AMBIGUOUS, ReplayFailure::AMBIGUOUS, ReplayFailure::AMBIGUOUS),
537564
};
538565

539-
KJ_EXPECT(runActorFetch(
540-
state, RetryEnforcement::ENABLED, kj::none, ActorFetchKind::HTTP) != kj::none);
566+
KJ_EXPECT(
567+
runActorFetch(state, RetryEnforcement::ENABLED, kj::none, ActorFetchKind::HTTP) != kj::none);
541568
KJ_EXPECT(state.requestCount == 5);
542569
KJ_EXPECT(state.retryCount == 4);
543570
}
@@ -547,16 +574,44 @@ KJ_TEST("actor fetch stops when the retry budget expires") {
547574
.failures = kj::arr(ReplayFailure::AMBIGUOUS, ReplayFailure::HANG),
548575
};
549576

550-
KJ_EXPECT(runActorFetch(
551-
state, RetryEnforcement::ENABLED, kj::none, ActorFetchKind::HTTP) != kj::none);
577+
KJ_EXPECT(
578+
runActorFetch(state, RetryEnforcement::ENABLED, kj::none, ActorFetchKind::HTTP) != kj::none);
552579
KJ_EXPECT(state.requestCount == 2);
553580
KJ_EXPECT(state.retryCount == 1);
554581
}
555582

556-
// Global actor factories must place the caller's metadata on the actor subrequest.
557-
KJ_TEST("GlobalActorOutgoingFactory places actor retry metadata on the actor subrequest") {
583+
KJ_TEST("actor fetch replay consumes one subrequest limit unit") {
584+
ReplayState state{.failures = kj::arr(ReplayFailure::NOT_DELIVERED)};
585+
uint checkedSubrequestCount = 0;
586+
TestFixture fixture(TestFixture::SetupParams{
587+
.useRealTimers = true,
588+
.checkedSubrequestCount = checkedSubrequestCount,
589+
});
590+
util::Autogate::initAutogateNamesForTest(
591+
{"durable-object-retries-fetch"_kj, "durable-object-retries-fetch-retry-requests"_kj},
592+
util::IgnoreAllAutogatesEnv::YES);
593+
594+
fixture.runInIoContext([&](const TestFixture::Environment& env) {
595+
auto fetcher = env.js.alloc<Fetcher>(
596+
env.context.addObject<Fetcher::OutgoingFactory>(kj::heap<ReplicaActorOutgoingFactory>(
597+
kj::refcounted<ReplayActorChannel>(state), kj::str("actor-id"))),
598+
Fetcher::RequiresHostAndProtocol::YES);
599+
auto promise = fetcher->fetch(env.js, kj::str("http://example.com"), kj::none);
600+
return env.context.awaitJs(env.js, kj::mv(promise)).ignoreResult().attach(kj::mv(fetcher));
601+
});
602+
603+
KJ_EXPECT(state.requestCount == 2);
604+
KJ_ASSERT(state.metadata.size() == 2);
605+
KJ_EXPECT(state.metadata[0].nonce != state.metadata[1].nonce);
606+
KJ_EXPECT(state.metadata[0].isRetry == IsActorRetry::NO);
607+
KJ_EXPECT(state.metadata[1].isRetry == IsActorRetry::NO);
608+
KJ_EXPECT(checkedSubrequestCount == 1);
609+
}
610+
611+
KJ_TEST("GlobalActorOutgoingFactory forwards metadata without counting retries against limits") {
558612
kj::Maybe<IoChannelFactory::ActorRetryRequestMetadata> capturedMetadata;
559613
uint channelCount = 0;
614+
uint checkedSubrequestCount = 0;
560615
kj::Vector<kj::String> locationHints;
561616
kj::Vector<kj::String> cohorts;
562617
TestFixture fixture(TestFixture::SetupParams{
@@ -566,6 +621,7 @@ KJ_TEST("GlobalActorOutgoingFactory places actor retry metadata on the actor sub
566621
return kj::rc<ActorIoChannelFactory>(
567622
timer, capturedMetadata, channelCount, locationHints, cohorts);
568623
}),
624+
.checkedSubrequestCount = checkedSubrequestCount,
569625
});
570626

571627
fixture.runInIoContext([&](const TestFixture::Environment& env) {
@@ -581,7 +637,8 @@ KJ_TEST("GlobalActorOutgoingFactory places actor retry metadata on the actor sub
581637
.nonce = 0x123456789abcdef0,
582638
.createdAt = kj::UNIX_EPOCH + 123 * kj::MILLISECONDS,
583639
.isRetry = IsActorRetry::YES,
584-
});
640+
},
641+
IsActorFetchRetryAttempt::NO);
585642

586643
KJ_IF_SOME(metadata, capturedMetadata) {
587644
KJ_EXPECT(metadata.nonce == 0x123456789abcdef0);
@@ -597,7 +654,9 @@ KJ_TEST("GlobalActorOutgoingFactory places actor retry metadata on the actor sub
597654
.nonce = 0xfedcba9876543210,
598655
.createdAt = kj::UNIX_EPOCH + 456 * kj::MILLISECONDS,
599656
.isRetry = IsActorRetry::YES,
600-
});
657+
},
658+
IsActorFetchRetryAttempt::YES);
659+
KJ_EXPECT(checkedSubrequestCount == 1);
601660
KJ_EXPECT(channelCount == 2);
602661
KJ_ASSERT(locationHints.size() == 2);
603662
KJ_EXPECT(locationHints[0] == "location");
@@ -608,9 +667,13 @@ KJ_TEST("GlobalActorOutgoingFactory places actor retry metadata on the actor sub
608667
});
609668
}
610669

611-
KJ_TEST("ReplicaActorOutgoingFactory places actor retry metadata on the actor subrequest") {
670+
KJ_TEST("ReplicaActorOutgoingFactory forwards metadata without counting retries against limits") {
612671
kj::Maybe<IoChannelFactory::ActorRetryRequestMetadata> capturedMetadata;
613-
TestFixture fixture;
672+
uint checkedSubrequestCount = 0;
673+
TestFixture fixture(TestFixture::SetupParams{
674+
.useRealTimers = false,
675+
.checkedSubrequestCount = checkedSubrequestCount,
676+
});
614677

615678
fixture.runInIoContext([&](const TestFixture::Environment& env) {
616679
ReplicaActorOutgoingFactory factory(
@@ -622,7 +685,8 @@ KJ_TEST("ReplicaActorOutgoingFactory places actor retry metadata on the actor su
622685
.nonce = 0x123456789abcdef0,
623686
.createdAt = kj::UNIX_EPOCH + 123 * kj::MILLISECONDS,
624687
.isRetry = IsActorRetry::YES,
625-
});
688+
},
689+
IsActorFetchRetryAttempt::NO);
626690

627691
KJ_IF_SOME(metadata, capturedMetadata) {
628692
KJ_EXPECT(metadata.nonce == 0x123456789abcdef0);
@@ -638,7 +702,9 @@ KJ_TEST("ReplicaActorOutgoingFactory places actor retry metadata on the actor su
638702
.nonce = 0xfedcba9876543210,
639703
.createdAt = kj::UNIX_EPOCH + 456 * kj::MILLISECONDS,
640704
.isRetry = IsActorRetry::YES,
641-
});
705+
},
706+
IsActorFetchRetryAttempt::YES);
707+
KJ_EXPECT(checkedSubrequestCount == 1);
642708
KJ_IF_SOME(metadata, capturedMetadata) {
643709
KJ_EXPECT(metadata.nonce == 0xfedcba9876543210);
644710
} else {

0 commit comments

Comments
 (0)