Skip to content

Accept span completion after request teardown #7016

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
apeacock1991 wants to merge 2 commits into
base: main
Choose a base branch
from
Open

Accept span completion after request teardown #7016

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
ee0396f
Restore JSRPC tracing changes
apeacock1991 Aug 16, 2026
e4db9fb
Accept span completion after request teardown
apeacock1991 Aug 14, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 3 additions & 2 deletions src/cloudflare/internal/test/instrumentation-test-helper.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -170,7 +170,7 @@ export function findSpanByName(state, name, filterFn = () => true) {
* @param {Array} expectedSpans - The expected spans to compare against
* @param {Object} options - Options for the test
* @param {Function} options.mapFn - Map function to transform spans before comparison (default: x => x)
* @param {Function} options.filterFn - Filter function for spans (default: filters out jsRpcSession)
* @param {Function} options.filterFn - Filter function for spans (default: filters out jsRpcSession and jsRpcCall)
* @param {string} options.testName - Name for the test (default: 'instrumentation')
* @param {boolean} options.logReceived - Log received spans for debugging (default: false)
*
Expand All @@ -185,7 +185,8 @@ export async function runInstrumentationTest(
) {
const {
mapFn = (x) => x,
filterFn = (span) => span.name !== 'jsRpcSession',
filterFn = (span) =>
span.name !== 'jsRpcSession' && span.name !== 'jsRpcCall',
testName = 'instrumentation',
logReceived = false,
} = options;
Expand Down
14 changes: 11 additions & 3 deletions src/workerd/api/actor-state.c++
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -986,20 +986,28 @@ class FacetOutgoingFactory final: public Fetcher::OutgoingFactory {
name(kj::mv(name)),
getStartInfo(kj::mv(getStartInfo)) {}

kj::Own<WorkerInterface> newSingleUseClient(kj::Maybe<kj::String> cfStr) override {
Result newSingleUseClient(
kj::Maybe<kj::String> cfStr, MakeUserSpanParent makeUserSpanParent) override {
auto& context = IoContext::current();

return context.getMetrics().wrapActorSubrequestClient(context.getSubrequest(
kj::Maybe<TraceContextParent> spanParents;
auto client = context.getMetrics().wrapActorSubrequestClient(context.getSubrequest(
[&](TraceContext& tracing, IoChannelFactory& ioChannelFactory) {
tracing.setTag("facet_name"_kjc, name.asPtr());
spanParents = tracing.getSpanParents();
auto userSpanParent = tracing.getUserSpanParent();
KJ_IF_SOME(parent, makeUserSpanParent(tracing)) {
userSpanParent = kj::mv(parent);
}

return getOrCreateActorChannel().startRequest({.cfBlobJson = kj::mv(cfStr),
.parentSpan = tracing.getInternalSpanParent(),
.userSpanParent = tracing.getUserSpanParent()});
.userSpanParent = kj::mv(userSpanParent)});
},
{.inHouse = true,
.wrapMetrics = true,
.operationName = kj::ConstString("facet_subrequest"_kjc)}));
return {.client = kj::mv(client), .spanParents = kj::mv(spanParents)};
}

kj::Own<IoChannelFactory::SubrequestChannel> getSubrequestChannel() override {
Expand Down
63 changes: 43 additions & 20 deletions src/workerd/api/actor.c++
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -42,22 +42,29 @@ IoChannelFactory::ActorChannel& LocalActorOutgoingFactory::getOrCreateActorChann
return *KJ_REQUIRE_NONNULL(actorChannel);
}

kj::Own<WorkerInterface> LocalActorOutgoingFactory::newSingleUseClient(
kj::Maybe<kj::String> cfStr) {
Fetcher::OutgoingFactory::Result LocalActorOutgoingFactory::newSingleUseClient(
kj::Maybe<kj::String> cfStr, MakeUserSpanParent makeUserSpanParent) {
auto& context = IoContext::current();

return context.getMetrics().wrapActorSubrequestClient(context.getSubrequest(
kj::Maybe<TraceContextParent> spanParents;
auto client = context.getMetrics().wrapActorSubrequestClient(context.getSubrequest(
[&](TraceContext& tracing, IoChannelFactory& ioChannelFactory) {
tracing.setTag("objectId"_kjc, actorId.asPtr());
spanParents = tracing.getSpanParents();
auto userSpanParent = tracing.getUserSpanParent();
KJ_IF_SOME(parent, makeUserSpanParent(tracing)) {
userSpanParent = kj::mv(parent);
}

return getOrCreateActorChannel(context, tracing.getInternalSpanParent())
.startRequest({.cfBlobJson = kj::mv(cfStr),
.parentSpan = tracing.getInternalSpanParent(),
.userSpanParent = tracing.getUserSpanParent()});
.userSpanParent = kj::mv(userSpanParent)});
},
{.inHouse = true,
.wrapMetrics = true,
.operationName = kj::ConstString("durable_object_subrequest"_kjc)}));
return {.client = kj::mv(client), .spanParents = kj::mv(spanParents)};
}

kj::Own<IoChannelFactory::SubrequestChannel> LocalActorOutgoingFactory::getSubrequestChannel() {
Expand Down Expand Up @@ -91,60 +98,76 @@ IoChannelFactory::ActorChannel& GlobalActorOutgoingFactory::getOrCreateActorChan
return *KJ_REQUIRE_NONNULL(actorChannel);
}

kj::Own<WorkerInterface> GlobalActorOutgoingFactory::newSingleUseClient(
kj::Maybe<kj::String> cfStr) {
return newSingleUseClientWithActorRetryMetadata(kj::mv(cfStr), kj::none);
Fetcher::OutgoingFactory::Result GlobalActorOutgoingFactory::newSingleUseClient(
kj::Maybe<kj::String> cfStr, MakeUserSpanParent makeUserSpanParent) {
return newSingleUseClientWithActorRetryMetadata(kj::mv(cfStr), kj::none, makeUserSpanParent);
}

kj::Own<WorkerInterface> GlobalActorOutgoingFactory::newSingleUseClientWithActorRetryMetadata(
kj::Maybe<kj::String> cfStr,
kj::Maybe<IoChannelFactory::ActorRetryRequestMetadata> actorRetryRequestMetadata) {
Fetcher::OutgoingFactory::Result GlobalActorOutgoingFactory::
newSingleUseClientWithActorRetryMetadata(kj::Maybe<kj::String> cfStr,
kj::Maybe<IoChannelFactory::ActorRetryRequestMetadata> actorRetryRequestMetadata,
MakeUserSpanParent makeUserSpanParent) {
auto& context = IoContext::current();

return context.getMetrics().wrapActorSubrequestClient(context.getSubrequest(
kj::Maybe<TraceContextParent> spanParents;
auto client = context.getMetrics().wrapActorSubrequestClient(context.getSubrequest(
[&](TraceContext& tracing, IoChannelFactory& ioChannelFactory) {
tracing.setTag("objectId"_kjc, id->toString());
spanParents = tracing.getSpanParents();
auto userSpanParent = tracing.getUserSpanParent();
KJ_IF_SOME(parent, makeUserSpanParent(tracing)) {
userSpanParent = kj::mv(parent);
}

return getOrCreateActorChannel(context, tracing.getInternalSpanParent())
.startRequest({.cfBlobJson = kj::mv(cfStr),
.parentSpan = tracing.getInternalSpanParent(),
.userSpanParent = tracing.getUserSpanParent(),
.userSpanParent = kj::mv(userSpanParent),
.actorRetryRequestMetadata = kj::mv(actorRetryRequestMetadata)});
},
{.inHouse = true,
.wrapMetrics = true,
.operationName = kj::ConstString("durable_object_subrequest"_kjc)}));
return {.client = kj::mv(client), .spanParents = kj::mv(spanParents)};
}

kj::Own<IoChannelFactory::SubrequestChannel> GlobalActorOutgoingFactory::getSubrequestChannel() {
auto& context = IoContext::current();
return kj::addRef(getOrCreateActorChannel(context, context.getCurrentTraceSpan()));
}

kj::Own<WorkerInterface> ReplicaActorOutgoingFactory::newSingleUseClient(
kj::Maybe<kj::String> cfStr) {
return newSingleUseClientWithActorRetryMetadata(kj::mv(cfStr), kj::none);
Fetcher::OutgoingFactory::Result ReplicaActorOutgoingFactory::newSingleUseClient(
kj::Maybe<kj::String> cfStr, MakeUserSpanParent makeUserSpanParent) {
return newSingleUseClientWithActorRetryMetadata(kj::mv(cfStr), kj::none, makeUserSpanParent);
}

kj::Own<WorkerInterface> ReplicaActorOutgoingFactory::newSingleUseClientWithActorRetryMetadata(
kj::Maybe<kj::String> cfStr,
kj::Maybe<IoChannelFactory::ActorRetryRequestMetadata> actorRetryRequestMetadata) {
Fetcher::OutgoingFactory::Result ReplicaActorOutgoingFactory::
newSingleUseClientWithActorRetryMetadata(kj::Maybe<kj::String> cfStr,
kj::Maybe<IoChannelFactory::ActorRetryRequestMetadata> actorRetryRequestMetadata,
MakeUserSpanParent makeUserSpanParent) {
auto& context = IoContext::current();

return context.getMetrics().wrapActorSubrequestClient(context.getSubrequest(
kj::Maybe<TraceContextParent> spanParents;
auto client = context.getMetrics().wrapActorSubrequestClient(context.getSubrequest(
[&](TraceContext& tracing, IoChannelFactory& ioChannelFactory) {
tracing.setTag("objectId"_kjc, actorId.asPtr());
spanParents = tracing.getSpanParents();
auto userSpanParent = tracing.getUserSpanParent();
KJ_IF_SOME(parent, makeUserSpanParent(tracing)) {
userSpanParent = kj::mv(parent);
}

// Unlike in `GlobalActorOutgoingFactory`, we do not create this lazily, since our channel was
// already open prior to this DO starting up.
return actorChannel->startRequest({.cfBlobJson = kj::mv(cfStr),
.parentSpan = tracing.getInternalSpanParent(),
.userSpanParent = tracing.getUserSpanParent(),
.userSpanParent = kj::mv(userSpanParent),
.actorRetryRequestMetadata = kj::mv(actorRetryRequestMetadata)});
},
{.inHouse = true,
.wrapMetrics = true,
.operationName = kj::ConstString("durable_object_subrequest"_kjc)}));
return {.client = kj::mv(client), .spanParents = kj::mv(spanParents)};
}

kj::Own<IoChannelFactory::SubrequestChannel> ReplicaActorOutgoingFactory::getSubrequestChannel() {
Expand Down
19 changes: 12 additions & 7 deletions src/workerd/api/actor.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -346,12 +346,14 @@ class GlobalActorOutgoingFactory final: public Fetcher::OutgoingFactory {
version(kj::mv(version)),
persistent(persistent) {}

kj::Own<WorkerInterface> newSingleUseClient(kj::Maybe<kj::String> cfStr) override;
Result newSingleUseClient(
kj::Maybe<kj::String> cfStr, MakeUserSpanParent makeUserSpanParent) override;
bool supportsActorRetryMetadata() const override {
return true;
}
kj::Own<WorkerInterface> newSingleUseClientWithActorRetryMetadata(kj::Maybe<kj::String> cfStr,
kj::Maybe<IoChannelFactory::ActorRetryRequestMetadata> actorRetryRequestMetadata) override;
Result newSingleUseClientWithActorRetryMetadata(kj::Maybe<kj::String> cfStr,
kj::Maybe<IoChannelFactory::ActorRetryRequestMetadata> actorRetryRequestMetadata,
MakeUserSpanParent makeUserSpanParent) override;
kj::Own<IoChannelFactory::SubrequestChannel> getSubrequestChannel() override;

private:
Expand Down Expand Up @@ -385,7 +387,8 @@ class LocalActorOutgoingFactory final: public Fetcher::OutgoingFactory {
: channelId(channelId),
actorId(kj::mv(actorId)) {}

kj::Own<WorkerInterface> newSingleUseClient(kj::Maybe<kj::String> cfStr) override;
Result newSingleUseClient(
kj::Maybe<kj::String> cfStr, MakeUserSpanParent makeUserSpanParent) override;
kj::Own<IoChannelFactory::SubrequestChannel> getSubrequestChannel() override;

private:
Expand All @@ -410,12 +413,14 @@ class ReplicaActorOutgoingFactory final: public Fetcher::OutgoingFactory {
: actorChannel(kj::mv(channel)),
actorId(kj::mv(actorId)) {}

kj::Own<WorkerInterface> newSingleUseClient(kj::Maybe<kj::String> cfStr) override;
Result newSingleUseClient(
kj::Maybe<kj::String> cfStr, MakeUserSpanParent makeUserSpanParent) override;
bool supportsActorRetryMetadata() const override {
return true;
}
kj::Own<WorkerInterface> newSingleUseClientWithActorRetryMetadata(kj::Maybe<kj::String> cfStr,
kj::Maybe<IoChannelFactory::ActorRetryRequestMetadata> actorRetryRequestMetadata) override;
Result newSingleUseClientWithActorRetryMetadata(kj::Maybe<kj::String> cfStr,
kj::Maybe<IoChannelFactory::ActorRetryRequestMetadata> actorRetryRequestMetadata,
MakeUserSpanParent makeUserSpanParent) override;
kj::Own<IoChannelFactory::SubrequestChannel> getSubrequestChannel() override;

private:
Expand Down
8 changes: 6 additions & 2 deletions src/workerd/api/bench-container-ingress.c++
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -228,10 +228,14 @@ class DirectWorkerInterface final: public WorkerInterface {
class DirectOutgoingFactory final: public Fetcher::OutgoingFactory {
public:
explicit DirectOutgoingFactory(kj::HttpClient& client): client(client) {}
kj::Own<WorkerInterface> newSingleUseClient(kj::Maybe<kj::String> cfStr) override {
return IoContext::current().getSubrequestNoChecks([this](auto& tracing, auto& channelFactory) {
Result newSingleUseClient(
kj::Maybe<kj::String> cfStr, MakeUserSpanParent makeUserSpanParent) override {
auto result = IoContext::current().getSubrequestNoChecks(
[this, &makeUserSpanParent](auto& tracing, auto& channelFactory) {
makeUserSpanParent(tracing);
return kj::heap<DirectWorkerInterface>(client);
}, {.inHouse = false, .wrapMetrics = false});
return {.client = kj::mv(result), .spanParents = kj::none};
}

private:
Expand Down
9 changes: 6 additions & 3 deletions src/workerd/api/container.c++
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1447,12 +1447,15 @@ class Container::TcpPortOutgoingFactory final: public Fetcher::OutgoingFactory {
headerTable(headerTable),
portState(kj::mv(portState)) {}

kj::Own<WorkerInterface> newSingleUseClient(kj::Maybe<kj::String> cfStr) override {
// At present we have no use for `cfStr`.
return IoContext::current().getSubrequestNoChecks(
Result newSingleUseClient(
kj::Maybe<kj::String> cfStr, MakeUserSpanParent makeUserSpanParent) override {
// At present we have no use for `cfStr`. This factory creates no operation span.
auto client = IoContext::current().getSubrequestNoChecks(
[&](auto& tracing, auto& channelFactory) -> kj::Own<WorkerInterface> {
makeUserSpanParent(tracing);
return kj::heap<TcpPortWorkerInterface>(entropySource, headerTable, portState.addRef());
}, {.inHouse = false, .wrapMetrics = false});
return {.client = kj::mv(client), .spanParents = kj::none};
}

private:
Expand Down
21 changes: 12 additions & 9 deletions src/workerd/api/fetch-body-rewindable-test.c++
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -80,19 +80,20 @@ class RetryMetadataOutgoingFactory final: public Fetcher::OutgoingFactory {
: ordinaryDispatchCalled(ordinaryDispatchCalled),
capturedMetadata(capturedMetadata) {}

kj::Own<WorkerInterface> newSingleUseClient(kj::Maybe<kj::String>) override {
Result newSingleUseClient(kj::Maybe<kj::String>, MakeUserSpanParent makeUserSpanParent) override {
ordinaryDispatchCalled = true;
return kj::heap<MockFetchTarget>();
return {.client = kj::heap<MockFetchTarget>(), .spanParents = kj::none};
}

bool supportsActorRetryMetadata() const override {
return true;
}

kj::Own<WorkerInterface> newSingleUseClientWithActorRetryMetadata(kj::Maybe<kj::String>,
kj::Maybe<IoChannelFactory::ActorRetryRequestMetadata> actorRetryRequestMetadata) override {
Result newSingleUseClientWithActorRetryMetadata(kj::Maybe<kj::String>,
kj::Maybe<IoChannelFactory::ActorRetryRequestMetadata> actorRetryRequestMetadata,
MakeUserSpanParent makeUserSpanParent) override {
capturedMetadata = kj::mv(actorRetryRequestMetadata);
return kj::heap<MockFetchTarget>();
return {.client = kj::heap<MockFetchTarget>(), .spanParents = kj::none};
}

private:
Expand All @@ -104,9 +105,9 @@ class UnsupportedOutgoingFactory final: public Fetcher::OutgoingFactory {
public:
UnsupportedOutgoingFactory(bool& called): called(called) {}

kj::Own<WorkerInterface> newSingleUseClient(kj::Maybe<kj::String>) override {
Result newSingleUseClient(kj::Maybe<kj::String>, MakeUserSpanParent makeUserSpanParent) override {
called = true;
return kj::heap<MockFetchTarget>();
return {.client = kj::heap<MockFetchTarget>(), .spanParents = kj::none};
}

private:
Expand Down Expand Up @@ -396,7 +397,8 @@ KJ_TEST("GlobalActorOutgoingFactory places actor retry metadata on the actor sub
.nonce = 0x123456789abcdef0,
.createdAt = kj::UNIX_EPOCH + 123 * kj::MILLISECONDS,
.isRetry = IsActorRetry::YES,
});
},
[](TraceContext&) -> kj::Maybe<SpanParent> { return kj::none; });

KJ_IF_SOME(metadata, capturedMetadata) {
KJ_EXPECT(metadata.nonce == 0x123456789abcdef0);
Expand All @@ -422,7 +424,8 @@ KJ_TEST("ReplicaActorOutgoingFactory places actor retry metadata on the actor su
.nonce = 0x123456789abcdef0,
.createdAt = kj::UNIX_EPOCH + 123 * kj::MILLISECONDS,
.isRetry = IsActorRetry::YES,
});
},
[](TraceContext&) -> kj::Maybe<SpanParent> { return kj::none; });

KJ_IF_SOME(metadata, capturedMetadata) {
KJ_EXPECT(metadata.nonce == 0x123456789abcdef0);
Expand Down
Loading
Loading