diff --git a/src/cloudflare/internal/test/instrumentation-test-helper.js b/src/cloudflare/internal/test/instrumentation-test-helper.js index 7f50fc4c689..a7bf373136e 100644 --- a/src/cloudflare/internal/test/instrumentation-test-helper.js +++ b/src/cloudflare/internal/test/instrumentation-test-helper.js @@ -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) * @@ -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; diff --git a/src/workerd/api/actor-state.c++ b/src/workerd/api/actor-state.c++ index 4284aae3578..c8f35ff286d 100644 --- a/src/workerd/api/actor-state.c++ +++ b/src/workerd/api/actor-state.c++ @@ -986,20 +986,28 @@ class FacetOutgoingFactory final: public Fetcher::OutgoingFactory { name(kj::mv(name)), getStartInfo(kj::mv(getStartInfo)) {} - kj::Own newSingleUseClient(kj::Maybe cfStr) override { + Result newSingleUseClient( + kj::Maybe cfStr, MakeUserSpanParent makeUserSpanParent) override { auto& context = IoContext::current(); - return context.getMetrics().wrapActorSubrequestClient(context.getSubrequest( + kj::Maybe 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 getSubrequestChannel() override { diff --git a/src/workerd/api/actor.c++ b/src/workerd/api/actor.c++ index 80d7e3d306b..b9b36925306 100644 --- a/src/workerd/api/actor.c++ +++ b/src/workerd/api/actor.c++ @@ -42,22 +42,29 @@ IoChannelFactory::ActorChannel& LocalActorOutgoingFactory::getOrCreateActorChann return *KJ_REQUIRE_NONNULL(actorChannel); } -kj::Own LocalActorOutgoingFactory::newSingleUseClient( - kj::Maybe cfStr) { +Fetcher::OutgoingFactory::Result LocalActorOutgoingFactory::newSingleUseClient( + kj::Maybe cfStr, MakeUserSpanParent makeUserSpanParent) { auto& context = IoContext::current(); - return context.getMetrics().wrapActorSubrequestClient(context.getSubrequest( + kj::Maybe 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 LocalActorOutgoingFactory::getSubrequestChannel() { @@ -91,29 +98,37 @@ IoChannelFactory::ActorChannel& GlobalActorOutgoingFactory::getOrCreateActorChan return *KJ_REQUIRE_NONNULL(actorChannel); } -kj::Own GlobalActorOutgoingFactory::newSingleUseClient( - kj::Maybe cfStr) { - return newSingleUseClientWithActorRetryMetadata(kj::mv(cfStr), kj::none); +Fetcher::OutgoingFactory::Result GlobalActorOutgoingFactory::newSingleUseClient( + kj::Maybe cfStr, MakeUserSpanParent makeUserSpanParent) { + return newSingleUseClientWithActorRetryMetadata(kj::mv(cfStr), kj::none, makeUserSpanParent); } -kj::Own GlobalActorOutgoingFactory::newSingleUseClientWithActorRetryMetadata( - kj::Maybe cfStr, - kj::Maybe actorRetryRequestMetadata) { +Fetcher::OutgoingFactory::Result GlobalActorOutgoingFactory:: + newSingleUseClientWithActorRetryMetadata(kj::Maybe cfStr, + kj::Maybe actorRetryRequestMetadata, + MakeUserSpanParent makeUserSpanParent) { auto& context = IoContext::current(); - return context.getMetrics().wrapActorSubrequestClient(context.getSubrequest( + kj::Maybe 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 GlobalActorOutgoingFactory::getSubrequestChannel() { @@ -121,30 +136,38 @@ kj::Own GlobalActorOutgoingFactory::getSubr return kj::addRef(getOrCreateActorChannel(context, context.getCurrentTraceSpan())); } -kj::Own ReplicaActorOutgoingFactory::newSingleUseClient( - kj::Maybe cfStr) { - return newSingleUseClientWithActorRetryMetadata(kj::mv(cfStr), kj::none); +Fetcher::OutgoingFactory::Result ReplicaActorOutgoingFactory::newSingleUseClient( + kj::Maybe cfStr, MakeUserSpanParent makeUserSpanParent) { + return newSingleUseClientWithActorRetryMetadata(kj::mv(cfStr), kj::none, makeUserSpanParent); } -kj::Own ReplicaActorOutgoingFactory::newSingleUseClientWithActorRetryMetadata( - kj::Maybe cfStr, - kj::Maybe actorRetryRequestMetadata) { +Fetcher::OutgoingFactory::Result ReplicaActorOutgoingFactory:: + newSingleUseClientWithActorRetryMetadata(kj::Maybe cfStr, + kj::Maybe actorRetryRequestMetadata, + MakeUserSpanParent makeUserSpanParent) { auto& context = IoContext::current(); - return context.getMetrics().wrapActorSubrequestClient(context.getSubrequest( + kj::Maybe 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 ReplicaActorOutgoingFactory::getSubrequestChannel() { diff --git a/src/workerd/api/actor.h b/src/workerd/api/actor.h index a71bfc657f7..7ff7e71dcd7 100644 --- a/src/workerd/api/actor.h +++ b/src/workerd/api/actor.h @@ -346,12 +346,14 @@ class GlobalActorOutgoingFactory final: public Fetcher::OutgoingFactory { version(kj::mv(version)), persistent(persistent) {} - kj::Own newSingleUseClient(kj::Maybe cfStr) override; + Result newSingleUseClient( + kj::Maybe cfStr, MakeUserSpanParent makeUserSpanParent) override; bool supportsActorRetryMetadata() const override { return true; } - kj::Own newSingleUseClientWithActorRetryMetadata(kj::Maybe cfStr, - kj::Maybe actorRetryRequestMetadata) override; + Result newSingleUseClientWithActorRetryMetadata(kj::Maybe cfStr, + kj::Maybe actorRetryRequestMetadata, + MakeUserSpanParent makeUserSpanParent) override; kj::Own getSubrequestChannel() override; private: @@ -385,7 +387,8 @@ class LocalActorOutgoingFactory final: public Fetcher::OutgoingFactory { : channelId(channelId), actorId(kj::mv(actorId)) {} - kj::Own newSingleUseClient(kj::Maybe cfStr) override; + Result newSingleUseClient( + kj::Maybe cfStr, MakeUserSpanParent makeUserSpanParent) override; kj::Own getSubrequestChannel() override; private: @@ -410,12 +413,14 @@ class ReplicaActorOutgoingFactory final: public Fetcher::OutgoingFactory { : actorChannel(kj::mv(channel)), actorId(kj::mv(actorId)) {} - kj::Own newSingleUseClient(kj::Maybe cfStr) override; + Result newSingleUseClient( + kj::Maybe cfStr, MakeUserSpanParent makeUserSpanParent) override; bool supportsActorRetryMetadata() const override { return true; } - kj::Own newSingleUseClientWithActorRetryMetadata(kj::Maybe cfStr, - kj::Maybe actorRetryRequestMetadata) override; + Result newSingleUseClientWithActorRetryMetadata(kj::Maybe cfStr, + kj::Maybe actorRetryRequestMetadata, + MakeUserSpanParent makeUserSpanParent) override; kj::Own getSubrequestChannel() override; private: diff --git a/src/workerd/api/bench-container-ingress.c++ b/src/workerd/api/bench-container-ingress.c++ index bf76425acc2..bdcd4972a8e 100644 --- a/src/workerd/api/bench-container-ingress.c++ +++ b/src/workerd/api/bench-container-ingress.c++ @@ -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 newSingleUseClient(kj::Maybe cfStr) override { - return IoContext::current().getSubrequestNoChecks([this](auto& tracing, auto& channelFactory) { + Result newSingleUseClient( + kj::Maybe cfStr, MakeUserSpanParent makeUserSpanParent) override { + auto result = IoContext::current().getSubrequestNoChecks( + [this, &makeUserSpanParent](auto& tracing, auto& channelFactory) { + makeUserSpanParent(tracing); return kj::heap(client); }, {.inHouse = false, .wrapMetrics = false}); + return {.client = kj::mv(result), .spanParents = kj::none}; } private: diff --git a/src/workerd/api/container.c++ b/src/workerd/api/container.c++ index 806b30bfcf9..40cabfdefe7 100644 --- a/src/workerd/api/container.c++ +++ b/src/workerd/api/container.c++ @@ -1447,12 +1447,15 @@ class Container::TcpPortOutgoingFactory final: public Fetcher::OutgoingFactory { headerTable(headerTable), portState(kj::mv(portState)) {} - kj::Own newSingleUseClient(kj::Maybe cfStr) override { - // At present we have no use for `cfStr`. - return IoContext::current().getSubrequestNoChecks( + Result newSingleUseClient( + kj::Maybe 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 { + makeUserSpanParent(tracing); return kj::heap(entropySource, headerTable, portState.addRef()); }, {.inHouse = false, .wrapMetrics = false}); + return {.client = kj::mv(client), .spanParents = kj::none}; } private: diff --git a/src/workerd/api/fetch-body-rewindable-test.c++ b/src/workerd/api/fetch-body-rewindable-test.c++ index 89fd4b4bf53..cfe7ba8de44 100644 --- a/src/workerd/api/fetch-body-rewindable-test.c++ +++ b/src/workerd/api/fetch-body-rewindable-test.c++ @@ -80,19 +80,20 @@ class RetryMetadataOutgoingFactory final: public Fetcher::OutgoingFactory { : ordinaryDispatchCalled(ordinaryDispatchCalled), capturedMetadata(capturedMetadata) {} - kj::Own newSingleUseClient(kj::Maybe) override { + Result newSingleUseClient(kj::Maybe, MakeUserSpanParent makeUserSpanParent) override { ordinaryDispatchCalled = true; - return kj::heap(); + return {.client = kj::heap(), .spanParents = kj::none}; } bool supportsActorRetryMetadata() const override { return true; } - kj::Own newSingleUseClientWithActorRetryMetadata(kj::Maybe, - kj::Maybe actorRetryRequestMetadata) override { + Result newSingleUseClientWithActorRetryMetadata(kj::Maybe, + kj::Maybe actorRetryRequestMetadata, + MakeUserSpanParent makeUserSpanParent) override { capturedMetadata = kj::mv(actorRetryRequestMetadata); - return kj::heap(); + return {.client = kj::heap(), .spanParents = kj::none}; } private: @@ -104,9 +105,9 @@ class UnsupportedOutgoingFactory final: public Fetcher::OutgoingFactory { public: UnsupportedOutgoingFactory(bool& called): called(called) {} - kj::Own newSingleUseClient(kj::Maybe) override { + Result newSingleUseClient(kj::Maybe, MakeUserSpanParent makeUserSpanParent) override { called = true; - return kj::heap(); + return {.client = kj::heap(), .spanParents = kj::none}; } private: @@ -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 { return kj::none; }); KJ_IF_SOME(metadata, capturedMetadata) { KJ_EXPECT(metadata.nonce == 0x123456789abcdef0); @@ -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 { return kj::none; }); KJ_IF_SOME(metadata, capturedMetadata) { KJ_EXPECT(metadata.nonce == 0x123456789abcdef0); diff --git a/src/workerd/api/http.c++ b/src/workerd/api/http.c++ index c34cc889181..bc656cffbf7 100644 --- a/src/workerd/api/http.c++ +++ b/src/workerd/api/http.c++ @@ -2118,26 +2118,44 @@ kj::Maybe> Fetcher::getRpcMethodInternal(jsg::Lock& js, return js.alloc(JSG_THIS, kj::mv(name)); } -rpc::JsRpcTarget::Client Fetcher::getClientForOneCall( +kj::LiteralStringConst Fetcher::getRpcTargetKind() { + return "fetcher"_kjc; +} + +JsRpcClientProvider::ClientForOneCall Fetcher::getClientForOneCall( jsg::Lock& js, kj::Vector& path) { auto& ioContext = IoContext::current(); - auto worker = getClient(ioContext, kj::none, "jsRpcSession"_kjc); - auto event = kj::heap( - JsRpcSessionCustomEvent::WORKER_RPC_EVENT_TYPE); + + // 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. + kj::Maybe callSpan; + auto clientWithTracing = buildClient(ioContext, kj::none, "jsRpcSession"_kjc, + [&](TraceContext& sessionSpan) -> kj::Maybe { + callSpan = sessionSpan.getSpanParents().newChild("jsRpcCall"_kjc); + return KJ_ASSERT_NONNULL(callSpan).getUserSpanParent(); + }); + kj::Maybe callSpanParents = clientWithTracing.traceContext.map( + [](TraceContext& tc) { return tc.getSpanParents(); }); + auto worker = kj::mv(clientWithTracing.client); + auto event = kj::heap(JsRpcSessionCustomEvent::WORKER_RPC_EVENT_TYPE); auto result = event->getCap(); // Arrange to cancel the CustomEvent if our I/O context is destroyed. But otherwise, we don't // actually care about the result of the event. If it throws, the membrane will already have - // propagated the exception to any RPC calls that we're waiting on, so we even ignore errors - // here -- otherwise they'll end up logged as "uncaught exceptions" even if they were, in fact, - // caught elsewhere. - ioContext.addTask(worker->customEvent(kj::mv(event)).attach(kj::mv(worker)).then([](auto&&) { - }, [](kj::Exception&&) {})); + // propagated the exception to any RPC calls that we're waiting on, so we even ignore errors here + // -- otherwise they'll end up logged as "uncaught exceptions" even if they were, in fact, caught + // elsewhere. + ioContext.addTask(worker->customEvent(kj::mv(event)) + .attach(kj::mv(worker), kj::mv(clientWithTracing.traceContext)) + .then([](auto&&) {}, [](kj::Exception&&) {})); // (Don't extend `path` because we're the root.) - return result; + return {.client = kj::mv(result), + .callSpanParents = kj::mv(callSpanParents), + .callSpan = kj::mv(callSpan)}; } void Fetcher::serialize(jsg::Lock& js, jsg::Serializer& serializer) { @@ -2489,8 +2507,7 @@ kj::Own Fetcher::getClient( return clientWithTracing.client.attach(kj::mv(clientWithTracing.traceContext)); } -Fetcher::ClientWithTracing Fetcher::getClientWithTracing( - IoContext& ioContext, +Fetcher::ClientWithTracing Fetcher::getClientWithTracing(IoContext& ioContext, kj::Maybe cfStr, kj::ConstString operationName, kj::Maybe actorRetryRequestMetadata) { @@ -2500,14 +2517,37 @@ Fetcher::ClientWithTracing Fetcher::getClientWithTracing( "actor retry metadata supplied to an unsupported Fetcher"); KJ_REQUIRE(outgoingFactory->supportsActorRetryMetadata(), "actor retry metadata supplied to an unsupported Fetcher"); - auto client = outgoingFactory->newSingleUseClientWithActorRetryMetadata( - kj::mv(cfStr), kj::mv(metadata)); - return ClientWithTracing{kj::mv(client), kj::none}; + kj::Maybe traceContext; + auto result = outgoingFactory->newSingleUseClientWithActorRetryMetadata(kj::mv(cfStr), + kj::mv(metadata), [&](TraceContext& outerTraceContext) -> kj::Maybe { + if (!outerTraceContext.isObserved()) return kj::none; + traceContext = outerTraceContext.getSpanParents().newChild(operationName.clone()); + return KJ_ASSERT_NONNULL(traceContext).getUserSpanParent(); + }); + return ClientWithTracing{kj::mv(result.client), kj::mv(traceContext)}; + } + + return buildClient(ioContext, kj::mv(cfStr), kj::mv(operationName)); +} + +Fetcher::ClientWithTracing Fetcher::wrapWithInnerSpan( + OutgoingFactory::Result result, kj::ConstString operationName) { + KJ_IF_SOME(parents, result.spanParents) { + // Factories populate `spanParents` unconditionally. Only build the inner span when tracing is + // actually observed; otherwise returning a (non-recording) TraceContext would still force the + // fetch hot path to eagerly evaluate span tags (method, URL, etc.) that end up discarded. + if (parents.isObserved()) { + return ClientWithTracing{kj::mv(result.client), parents.newChild(kj::mv(operationName))}; + } } + return ClientWithTracing{kj::mv(result.client), kj::none}; +} +Fetcher::ClientWithTracing Fetcher::buildClient(IoContext& ioContext, + kj::Maybe cfStr, + kj::ConstString operationName) { KJ_SWITCH_ONEOF(channelOrClientFactory) { KJ_CASE_ONEOF(channel, uint) { - // For channels, create trace context auto traceContext = ioContext.makeUserTraceSpan(kj::mv(operationName)); auto client = ioContext.getSubrequestChannel(channel, isInHouse, kj::mv(cfStr), traceContext); return ClientWithTracing{kj::mv(client), kj::mv(traceContext)}; @@ -2527,17 +2567,17 @@ Fetcher::ClientWithTracing Fetcher::getClientWithTracing( return ClientWithTracing{kj::mv(client), kj::mv(traceContext)}; } KJ_CASE_ONEOF(outgoingFactory, IoOwn) { - // Outgoing factories are responsible for routing through getSubrequestNoChecks() (or - // getSubrequest()) internally if they create HTTP connections, to ensure external memory - // adjustment and other subrequest accounting are applied. - auto client = outgoingFactory->newSingleUseClient(kj::mv(cfStr)); - return ClientWithTracing{kj::mv(client), kj::none}; + // The factory creates its own outer dispatch span (e.g. durable_object_subrequest) + // and exposes it as `result.spanParents`. Nest our inner span under it so the trace + // tree shows `outerSpan -> operationName`. + auto result = outgoingFactory->newSingleUseClient(kj::mv(cfStr), + [](TraceContext&) -> kj::Maybe { return kj::none; }); + return wrapWithInnerSpan(kj::mv(result), kj::mv(operationName)); } KJ_CASE_ONEOF(outgoingFactory, kj::Own) { - // Same as OutgoingFactory above -- the factory is responsible for routing through - // getSubrequestNoChecks() internally. - auto client = outgoingFactory->newSingleUseClient(ioContext, kj::mv(cfStr)); - return ClientWithTracing{kj::mv(client), kj::none}; + auto result = outgoingFactory->newSingleUseClient(ioContext, kj::mv(cfStr), + [](TraceContext&) -> kj::Maybe { return kj::none; }); + return wrapWithInnerSpan(kj::mv(result), kj::mv(operationName)); } } KJ_UNREACHABLE; @@ -2550,6 +2590,65 @@ bool Fetcher::supportsActorRetryMetadata() { return false; } +Fetcher::ClientWithTracing Fetcher::buildClient(IoContext& ioContext, + kj::Maybe cfStr, + kj::ConstString operationName, + MakeUserSpanParent makeUserSpanParent) { + KJ_SWITCH_ONEOF(channelOrClientFactory) { + KJ_CASE_ONEOF(channel, uint) { + auto traceContext = ioContext.makeUserTraceSpan(kj::mv(operationName)); + auto userSpanParent = makeUserSpanParent(traceContext); + kj::Own client; + KJ_IF_SOME(parent, userSpanParent) { + client = ioContext.getSubrequestChannel( + channel, isInHouse, kj::mv(cfStr), traceContext, kj::mv(parent)); + } else { + client = ioContext.getSubrequestChannel(channel, isInHouse, kj::mv(cfStr), traceContext); + } + return ClientWithTracing{kj::mv(client), kj::mv(traceContext)}; + } + KJ_CASE_ONEOF(channel, IoOwn) { + auto traceContext = ioContext.makeUserTraceSpan(kj::mv(operationName)); + auto propagatedUserSpanParent = traceContext.getUserSpanParent(); + KJ_IF_SOME(parent, makeUserSpanParent(traceContext)) { + propagatedUserSpanParent = kj::mv(parent); + } + auto client = ioContext.getSubrequest( + [&](TraceContext& tracing, IoChannelFactory& ioChannelFactory) { + return channel->startRequest({.cfBlobJson = kj::mv(cfStr), + .parentSpan = tracing.getInternalSpanParent(), + .userSpanParent = kj::mv(propagatedUserSpanParent)}); + }, { + .inHouse = isInHouse, + .wrapMetrics = !isInHouse, + .existingTraceContext = traceContext, + }); + return ClientWithTracing{kj::mv(client), kj::mv(traceContext)}; + } + KJ_CASE_ONEOF(outgoingFactory, IoOwn) { + kj::Maybe traceContext; + auto result = outgoingFactory->newSingleUseClient(kj::mv(cfStr), + [&](TraceContext& outerTraceContext) -> kj::Maybe { + if (!outerTraceContext.isObserved()) return kj::none; + traceContext = outerTraceContext.getSpanParents().newChild(operationName.clone()); + return makeUserSpanParent(KJ_ASSERT_NONNULL(traceContext)); + }); + return ClientWithTracing{kj::mv(result.client), kj::mv(traceContext)}; + } + KJ_CASE_ONEOF(outgoingFactory, kj::Own) { + kj::Maybe traceContext; + auto result = outgoingFactory->newSingleUseClient(ioContext, kj::mv(cfStr), + [&](TraceContext& outerTraceContext) -> kj::Maybe { + if (!outerTraceContext.isObserved()) return kj::none; + traceContext = outerTraceContext.getSpanParents().newChild(operationName.clone()); + return makeUserSpanParent(KJ_ASSERT_NONNULL(traceContext)); + }); + return ClientWithTracing{kj::mv(result.client), kj::mv(traceContext)}; + } + } + KJ_UNREACHABLE; +} + kj::Own Fetcher::getSubrequestChannel(IoContext& ioContext) { KJ_SWITCH_ONEOF(channelOrClientFactory) { KJ_CASE_ONEOF(channel, uint) { diff --git a/src/workerd/api/http.h b/src/workerd/api/http.h index 55dafa175cc..66eb6718e9d 100644 --- a/src/workerd/api/http.h +++ b/src/workerd/api/http.h @@ -187,6 +187,12 @@ using AnySocketAddress = kj::OneOf; // renamed, though I haven't heard any great suggestions for what the name should be. class Fetcher: public JsRpcClientProvider { public: + // Called synchronously while constructing a WorkerInterface, after its outer dispatch span has + // been opened but before SubrequestMetadata is consumed. Returning kj::none preserves the + // factory's usual user span parent. + using MakeUserSpanParent = + kj::FunctionParam(TraceContext& outerTraceContext)>; + // Should we use a fake https base url if we lack a scheme+authority? enum class RequiresHostAndProtocol { YES, NO }; @@ -234,7 +240,18 @@ class Fetcher: public JsRpcClientProvider { // is almost the same thing. class OutgoingFactory { public: - virtual kj::Own newSingleUseClient(kj::Maybe cfStr) = 0; + using MakeUserSpanParent = Fetcher::MakeUserSpanParent; + + struct Result { + kj::Own client; + // Parents of the dispatch-site span (e.g. durable_object_subrequest) that the + // caller can use to nest an inner operation span underneath. SpanParent holds an + // owning refcount on the underlying SpanObserver, so these are independently + // valid regardless of `client`'s lifetime. kj::none if no span was created. + kj::Maybe spanParents; + }; + virtual Result newSingleUseClient( + kj::Maybe cfStr, MakeUserSpanParent makeUserSpanParent) = 0; virtual bool supportsActorRetryMetadata() const { return false; @@ -242,9 +259,9 @@ class Fetcher: public JsRpcClientProvider { // Factories that can carry actor retry metadata override this method. The default rejects the // metadata rather than silently starting a new logical call. - virtual kj::Own newSingleUseClientWithActorRetryMetadata( - kj::Maybe cfStr, - kj::Maybe actorRetryRequestMetadata) { + virtual Result newSingleUseClientWithActorRetryMetadata(kj::Maybe cfStr, + kj::Maybe actorRetryRequestMetadata, + MakeUserSpanParent makeUserSpanParent) { KJ_FAIL_REQUIRE("actor retry metadata supplied to an unsupported Fetcher"); } @@ -264,8 +281,10 @@ class Fetcher: public JsRpcClientProvider { // IoContext::getSubrequestNoChecks() internally. class CrossContextOutgoingFactory { public: - virtual kj::Own newSingleUseClient( - IoContext& context, kj::Maybe cfStr) = 0; + using MakeUserSpanParent = Fetcher::MakeUserSpanParent; + + virtual OutgoingFactory::Result newSingleUseClient( + IoContext& context, kj::Maybe cfStr, MakeUserSpanParent makeUserSpanParent) = 0; virtual kj::Own getSubrequestChannel(IoContext& context) { // TODO(soon): Update all implementations and remove this default implementation. @@ -291,7 +310,7 @@ class Fetcher: public JsRpcClientProvider { requiresHost(requiresHost), isInHouse(isInHouse) {} - // Returns an `WorkerInterface` that is only valid for the lifetime of the current + // Returns a `WorkerInterface` that is only valid for the lifetime of the current // `IoContext`. kj::Own getClient( IoContext& ioContext, kj::Maybe cfStr, kj::ConstString operationName); @@ -302,8 +321,9 @@ class Fetcher: public JsRpcClientProvider { kj::Maybe traceContext; }; - // Get client and optionally create trace context, all in one call - ClientWithTracing getClientWithTracing(IoContext& ioContext, + // Get client and optionally create trace context, all in one call. + // + [[nodiscard]] ClientWithTracing getClientWithTracing(IoContext& ioContext, kj::Maybe cfStr, kj::ConstString operationName, kj::Maybe actorRetryRequestMetadata); @@ -401,8 +421,9 @@ class Fetcher: public JsRpcClientProvider { return getRpcMethod(js, kj::mv(name)); } - rpc::JsRpcTarget::Client getClientForOneCall( - jsg::Lock& js, kj::Vector& path) override; + ClientForOneCall getClientForOneCall(jsg::Lock& js, kj::Vector& path) override; + + kj::LiteralStringConst getRpcTargetKind() override; JSG_RESOURCE_TYPE(Fetcher, CompatibilityFlags::Reader flags) { // WARNING: New JSG_METHODs on Fetcher must be gated via compatibility flag to prevent @@ -516,6 +537,19 @@ class Fetcher: public JsRpcClientProvider { jsg::Deserializer& deserializer, RpcCompatGateBypassed rpcCompatGateBypassed); + [[nodiscard]] ClientWithTracing buildClient( + IoContext& ioContext, kj::Maybe cfStr, kj::ConstString operationName); + [[nodiscard]] ClientWithTracing buildClient(IoContext& ioContext, + kj::Maybe cfStr, + kj::ConstString operationName, + MakeUserSpanParent makeUserSpanParent); + + // Wraps an OutgoingFactory result, nesting an inner operation span under the factory's outer + // dispatch span when it created one. Factories that create no dispatch span + // (result.spanParents == kj::none) yield no inner span and no trace context. + [[nodiscard]] static ClientWithTracing wrapWithInnerSpan( + OutgoingFactory::Result result, kj::ConstString operationName); + kj::OneOf, kj::Own, diff --git a/src/workerd/api/sockets.c++ b/src/workerd/api/sockets.c++ index 4dfd17227cc..3690063983c 100644 --- a/src/workerd/api/sockets.c++ +++ b/src/workerd/api/sockets.c++ @@ -1060,7 +1060,8 @@ class StreamOutgoingFactory final: public Fetcher::OutgoingFactory, public kj::R httpClient( kj::newHttpClient(headerTable, *this->stream, {.entropySource = entropySource})) {} - kj::Own newSingleUseClient(kj::Maybe cfStr) override; + Result newSingleUseClient( + kj::Maybe cfStr, MakeUserSpanParent makeUserSpanParent) override; private: kj::Own stream; @@ -1125,14 +1126,19 @@ class StreamWorkerInterface final: public WorkerInterface { kj::Own factory; }; -kj::Own StreamOutgoingFactory::newSingleUseClient(kj::Maybe cfStr) { +Fetcher::OutgoingFactory::Result StreamOutgoingFactory::newSingleUseClient( + kj::Maybe cfStr, MakeUserSpanParent makeUserSpanParent) { + // This factory creates no operation span. JSG_ASSERT(stream.get() != nullptr, Error, "Fetcher created from internalNewHttpClient can only be used once"); // Create a WorkerInterface that wraps the stream, routing through getSubrequestNoChecks to apply // external memory adjustment for GC pressure. - return IoContext::current().getSubrequestNoChecks([&](auto& tracing, auto& channelFactory) { + auto client = IoContext::current().getSubrequestNoChecks( + [&](auto& tracing, auto& channelFactory) -> kj::Own { + makeUserSpanParent(tracing); return kj::heap(kj::addRef(*this)); }, {.inHouse = false, .wrapMetrics = false}); + return {.client = kj::mv(client), .spanParents = kj::none}; } jsg::Promise> SocketsModule::internalNewHttpClient( diff --git a/src/workerd/api/tests/BUILD.bazel b/src/workerd/api/tests/BUILD.bazel index 3ff58861ce4..0120fe694d6 100644 --- a/src/workerd/api/tests/BUILD.bazel +++ b/src/workerd/api/tests/BUILD.bazel @@ -187,6 +187,30 @@ wd_test( tags = ["no-coverage"], ) +# Regression test: a callback passed as an RPC argument, when invoked by the callee, must +# produce a jsRpcCall span nested under the method's server-side jsRpcCall span. +wd_test( + src = "jsrpc-callback-trace-test.wd-test", + args = ["--experimental"], + data = [ + "jsrpc-callback-trace-test.js", + "jsrpc-callback-trace-test-tail.js", + ], + tags = ["no-coverage"], +) + +# Test: a callee invocation is parented to the first caller jsRpcCall, and each dispatch links to +# its corresponding caller span. +wd_test( + src = "jsrpc-pipelined-trace-test.wd-test", + args = ["--experimental"], + data = [ + "jsrpc-pipelined-trace-test.js", + "jsrpc-pipelined-trace-test-tail.js", + ], + tags = ["no-coverage"], +) + # Regression test: jsrpc onset.info emitted into a streaming tail worker must remain # serializable when forwarded over service-binding RPC, just like fetch onset.info. wd_test( diff --git a/src/workerd/api/tests/actor-kv-test-tail.js b/src/workerd/api/tests/actor-kv-test-tail.js index 06423acd250..7121b597fee 100644 --- a/src/workerd/api/tests/actor-kv-test-tail.js +++ b/src/workerd/api/tests/actor-kv-test-tail.js @@ -20,6 +20,16 @@ export const test = { objectId: 'aa299662980ce671dbcb09a5d7ab26ab30e45465bcd12f263f2bdd7d5edd804a', }, + { + name: 'fetch', + closed: true, + 'network.protocol.name': 'http', + 'network.protocol.version': 'HTTP/1.1', + 'http.request.method': 'GET', + 'url.full': 'http://test.example/kv-test', + 'http.response.status_code': 200n, + 'http.response.body.size': 34n, + }, { name: 'durable_object_storage_put', closed: true }, { name: 'durable_object_storage_put', closed: true }, { name: 'durable_object_storage_get', closed: true }, diff --git a/src/workerd/api/tests/jsrpc-callback-trace-test-tail.js b/src/workerd/api/tests/jsrpc-callback-trace-test-tail.js new file mode 100644 index 00000000000..fb8577bf46f --- /dev/null +++ b/src/workerd/api/tests/jsrpc-callback-trace-test-tail.js @@ -0,0 +1,119 @@ +// Copyright (c) 2025 Cloudflare, Inc. +// Licensed under the Apache 2.0 license found in the LICENSE file or at: +// https://opensource.org/licenses/Apache-2.0 + +// Streaming tail worker that validates jsRpcCall span nesting for callback arguments. +// +// Within the CallbackService (jsrpc) invocation we expect: +// onset +// jsRpcCall (jsrpc.method = "invokeCallback", target_kind = "entrypoint") <- server dispatch +// jsRpcCall (target_kind = "stub") <- server invokes cb +// +// The inner (callback) jsRpcCall must be a child of the method's jsRpcCall span, proving the +// argument stub recorded the correct originating call. + +import * as assert from 'node:assert'; + +// Per-invocation span data: invocationId -> { onset, rootSpanId, spans: Map(spanId -> span) }. +// Each span records its name, its parent span ID (from the spanOpen's spanContext), and any +// attributes merged from subsequent attributes events targeting that span. +let invocations = new Map(); + +export default { + tailStream(onsetEvent, env, ctx) { + const invocationId = onsetEvent.invocationId; + const rootSpanId = onsetEvent.event.spanId; + const data = { + onset: { + info: onsetEvent.event.info?.type, + entrypoint: onsetEvent.event.entrypoint, + }, + rootSpanId, + spans: new Map(), + }; + invocations.set(invocationId, data); + + return (event) => { + const type = event.event.type; + if (type === 'spanOpen') { + data.spans.set(event.event.spanId, { + name: event.event.name, + parentId: event.spanContext.spanId, + attrs: {}, + }); + } else if (type === 'attributes') { + const span = data.spans.get(event.spanContext.spanId); + if (span) { + for (const { name, value } of event.event.info) { + span.attrs[name] = value; + } + } + } + }; + }, +}; + +// Find the server-side CallbackService JSRPC invocation and its two jsRpcCall spans, or return +// null if they haven't all arrived yet. Tail events are delivered asynchronously, so callers poll. +function findCallbackSpans() { + let target = null; + for (const data of invocations.values()) { + if ( + data.onset.info === 'jsrpc' && + data.onset.entrypoint === 'CallbackService' + ) { + target = data; + break; + } + } + if (!target) return { target: null, methodSpan: null, callbackSpan: null }; + + const jsRpcCalls = [...target.spans.entries()] + .filter(([, s]) => s.name === 'jsRpcCall') + .map(([spanId, s]) => ({ spanId, ...s })); + + // The method dispatch span (server-side jsRpcCall for invokeCallback) and the callback + // invocation span (the server calling back into the client stub, target_kind=stub). + const methodSpan = jsRpcCalls.find( + (s) => s.attrs['jsrpc.method'] === 'invokeCallback' + ); + const callbackSpan = jsRpcCalls.find( + (s) => s.attrs['jsrpc.target_kind'] === 'stub' + ); + return { target, methodSpan, callbackSpan }; +} + +export const test = { + async test() { + // Poll until the invocation and both spans have arrived rather than relying on a fixed delay. + const deadline = Date.now() + 5000; + let found = findCallbackSpans(); + while (!(found.methodSpan && found.callbackSpan) && Date.now() < deadline) { + await scheduler.wait(10); + found = findCallbackSpans(); + } + + const { target, methodSpan, callbackSpan } = found; + assert.ok( + target, + 'Could not find the CallbackService JSRPC invocation in tail events' + ); + assert.ok(methodSpan, 'Missing jsRpcCall span for invokeCallback'); + assert.ok( + callbackSpan, + 'Missing jsRpcCall span for the callback invocation (target_kind=stub)' + ); + + // The core assertion: the callback jsRpcCall nests under the method jsRpcCall, not the onset. + assert.strictEqual( + callbackSpan.parentId, + methodSpan.spanId, + 'Callback jsRpcCall should nest under the method jsRpcCall span, not the onset' + ); + assert.notStrictEqual( + callbackSpan.parentId, + target.rootSpanId, + 'Callback jsRpcCall must not be parented directly under the onset span' + ); + }, +}; diff --git a/src/workerd/api/tests/jsrpc-callback-trace-test.js b/src/workerd/api/tests/jsrpc-callback-trace-test.js new file mode 100644 index 00000000000..caa4477e213 --- /dev/null +++ b/src/workerd/api/tests/jsrpc-callback-trace-test.js @@ -0,0 +1,28 @@ +// Copyright (c) 2025 Cloudflare, Inc. +// Licensed under the Apache 2.0 license found in the LICENSE file or at: +// https://opensource.org/licenses/Apache-2.0 + +// Regression test for JSRPC tracing: when an RPC method receives a callback/stub as an argument +// and invokes it, the follow-up jsRpcCall (the server calling back into the client) must nest +// under the server-side jsRpcCall span of the method that received the callback -- not become a +// sibling of it under the onset. The parent/child relationship is asserted in the tail worker's +// test() handler (see jsrpc-callback-trace-test-tail.js). + +import { WorkerEntrypoint } from 'cloudflare:workers'; + +export class CallbackService extends WorkerEntrypoint { + // Receives a callback (serialized as an RPC stub) and invokes it. Invoking `cb` makes an RPC + // back to the caller, producing a client-side jsRpcCall span within this invocation's trace. + async invokeCallback(cb) { + return await cb(); + } +} + +export default { + async test(controller, env, ctx) { + const result = await env.CallbackService.invokeCallback(() => 42); + if (result !== 42) { + throw new Error(`Expected callback result 42, got ${result}`); + } + }, +}; diff --git a/src/workerd/api/tests/jsrpc-callback-trace-test.wd-test b/src/workerd/api/tests/jsrpc-callback-trace-test.wd-test new file mode 100644 index 00000000000..f8822b05f1c --- /dev/null +++ b/src/workerd/api/tests/jsrpc-callback-trace-test.wd-test @@ -0,0 +1,30 @@ +using Workerd = import "/workerd/workerd.capnp"; + +# Regression test for JSRPC tracing: a callback passed as an RPC argument, when invoked by the +# callee, must produce a jsRpcCall span nested under the method's server-side jsRpcCall span. +# The nesting is asserted in the tail worker's test() handler. + +const unitTests :Workerd.Config = ( + services = [ + (name = "jsrpc-callback-trace-test", worker = .mainWorker), + (name = "tail", worker = .tailWorker), + ], +); + +const mainWorker :Workerd.Worker = ( + modules = [ + (name = "worker", esModule = embed "jsrpc-callback-trace-test.js") + ], + compatibilityFlags = ["nodejs_compat", "experimental", "rpc"], + bindings = [ + (name = "CallbackService", service = (name = "jsrpc-callback-trace-test", entrypoint = "CallbackService")), + ], + streamingTails = ["tail"], +); + +const tailWorker :Workerd.Worker = ( + modules = [ + (name = "worker", esModule = embed "jsrpc-callback-trace-test-tail.js") + ], + compatibilityFlags = ["experimental", "nodejs_compat"], +); diff --git a/src/workerd/api/tests/jsrpc-pipelined-trace-test-tail.js b/src/workerd/api/tests/jsrpc-pipelined-trace-test-tail.js new file mode 100644 index 00000000000..3c43158d9ee --- /dev/null +++ b/src/workerd/api/tests/jsrpc-pipelined-trace-test-tail.js @@ -0,0 +1,186 @@ +// Copyright (c) 2026 Cloudflare, Inc. +// Licensed under the Apache 2.0 license found in the LICENSE file or at: +// https://opensource.org/licenses/Apache-2.0 + +// Streaming tail worker asserting that a callee's per-call jsRpcCall span is attributed to the +// caller's per-call jsRpcCall span rather than only to the session. +// +// Two invocations take part: +// +// caller (the test() handler) +// jsRpcSession <- covers the whole session +// jsRpcCall (getCounter) <- opens the session +// jsRpcCall (increment) <- reuses the session +// +// callee (CounterService, jsrpc onset) +// onset parent <- caller's getCounter jsRpcCall +// jsRpcCall (getCounter dispatch) <- jsrpc.caller_span_id = caller's getCounter jsRpcCall +// jsRpcCall (increment dispatch) <- jsrpc.caller_span_id = caller's increment jsRpcCall +// +// Both callee spans belong to a single invocation (one onset), so the onset's parent alone +// can identify only the call that opened the session. Each dispatch therefore also carries a link +// to its specific caller call while remaining a child of its own invocation root. + +import * as assert from 'node:assert'; + +// invocationId -> { onset, rootSpanId, spans: Map(spanId -> {name, parentId, attrs}) } +let invocations = new Map(); + +export default { + tailStream(onsetEvent, env, ctx) { + const data = { + onset: { + info: onsetEvent.event.info?.type, + entrypoint: onsetEvent.event.entrypoint, + parentId: onsetEvent.spanContext.spanId, + }, + rootSpanId: onsetEvent.event.spanId, + spans: new Map(), + }; + invocations.set(onsetEvent.invocationId, data); + + return (event) => { + const type = event.event.type; + if (type === 'spanOpen') { + data.spans.set(event.event.spanId, { + name: event.event.name, + parentId: event.spanContext.spanId, + attrs: {}, + }); + } else if (type === 'attributes') { + const span = data.spans.get(event.spanContext.spanId); + if (span) { + for (const { name, value } of event.event.info) { + span.attrs[name] = value; + } + } + } + }; + }, +}; + +function spansNamed(data, name) { + return [...data.spans.entries()] + .filter(([, s]) => s.name === name) + .map(([spanId, s]) => ({ spanId, ...s })); +} + +// Returns the caller and callee invocations once both have reported the spans we need, else nulls. +// Tail events arrive asynchronously, so callers poll. +function findInvocations() { + let callee = null; + let caller = null; + for (const data of invocations.values()) { + if ( + data.onset.info === 'jsrpc' && + data.onset.entrypoint === 'CounterService' + ) { + callee = data; + } else if (spansNamed(data, 'jsRpcSession').length > 0) { + caller = data; + } + } + if (!caller || !callee) return { caller: null, callee: null }; + + // Expect both dispatches on the callee and both client-side calls on the caller. + if (spansNamed(callee, 'jsRpcCall').length < 2) + return { caller: null, callee: null }; + if (spansNamed(caller, 'jsRpcCall').length < 2) + return { caller: null, callee: null }; + return { caller, callee }; +} + +export const test = { + async test() { + const deadline = Date.now() + 5000; + let found = findInvocations(); + while (!found.callee && Date.now() < deadline) { + await scheduler.wait(10); + found = findInvocations(); + } + + const { caller, callee } = found; + assert.ok(caller, 'Could not find the caller invocation in tail events'); + assert.ok( + callee, + 'Could not find the CounterService JSRPC invocation in tail events' + ); + + const callerCalls = spansNamed(caller, 'jsRpcCall'); + const calleeCalls = spansNamed(callee, 'jsRpcCall'); + const callerCallIds = new Set(callerCalls.map((s) => s.spanId)); + const callerCallsByMethod = new Map( + callerCalls.map((span) => [span.attrs['jsrpc.method'], span]) + ); + + const sessionSpans = spansNamed(caller, 'jsRpcSession'); + assert.strictEqual( + sessionSpans.length, + 1, + 'Expected exactly one jsRpcSession span' + ); + const sessionSpanId = sessionSpans[0].spanId; + + // Sanity check that the caller nests its own calls as expected: the increment call was made on + // a stub returned by getCounter, so it nests under getCounter's span. + const callerGetCounter = callerCalls.find( + (s) => s.attrs['jsrpc.method'] === 'getCounter' + ); + assert.ok(callerGetCounter, "Missing caller's getCounter jsRpcCall span"); + assert.strictEqual( + callerGetCounter.parentId, + sessionSpanId, + "The caller's first call should nest under the jsRpcSession span" + ); + assert.strictEqual( + callee.onset.parentId, + callerGetCounter.spanId, + 'The callee onset should be parented to the jsRpcCall that opened the session' + ); + + // The core assertions: every callee dispatch span is linked to a specific caller call, while + // remaining contained within its own invocation. + for (const span of calleeCalls) { + const method = span.attrs['jsrpc.method']; + const link = span.attrs['jsrpc.caller_span_id']; + + assert.ok( + link, + `Callee jsRpcCall (${method}) should carry a jsrpc.caller_span_id attribute` + ); + assert.notStrictEqual( + link, + sessionSpanId, + `Callee jsRpcCall (${method}) should link to a specific call, not the session span` + ); + assert.ok( + callerCallIds.has(link), + `Callee jsRpcCall (${method}) should link to one of the caller's jsRpcCall spans ` + + `(jsrpc.caller_span_id=${link}, caller call spans=${[...callerCallIds].join(',')})` + ); + assert.strictEqual( + link, + callerCallsByMethod.get(method)?.spanId, + `Callee jsRpcCall (${method}) should link to the caller's matching method span` + ); + + // The span itself must stay within its own invocation so this tail stream is self-contained. + assert.ok( + span.parentId === callee.rootSpanId || callee.spans.has(span.parentId), + `Callee jsRpcCall (${method}) must be parented within its own invocation ` + + `(parentId=${span.parentId})` + ); + } + + // The two dispatches must link to *different* caller calls, proving the link tracks the + // individual call rather than something session-wide. + const distinctLinks = new Set( + calleeCalls.map((s) => s.attrs['jsrpc.caller_span_id']) + ); + assert.strictEqual( + distinctLinks.size, + calleeCalls.length, + 'Each callee jsRpcCall should link to a distinct caller call' + ); + }, +}; diff --git a/src/workerd/api/tests/jsrpc-pipelined-trace-test.js b/src/workerd/api/tests/jsrpc-pipelined-trace-test.js new file mode 100644 index 00000000000..38192094f51 --- /dev/null +++ b/src/workerd/api/tests/jsrpc-pipelined-trace-test.js @@ -0,0 +1,46 @@ +// Copyright (c) 2026 Cloudflare, Inc. +// Licensed under the Apache 2.0 license found in the LICENSE file or at: +// https://opensource.org/licenses/Apache-2.0 + +// Test for JSRPC tracing: the callee invocation must be parented to the caller's first per-call +// jsRpcCall span, and every callee call must link to its corresponding caller span. +// +// A single JSRPC session carries many calls: `getCounter()` opens the session, and the calls on +// the returned stub reuse it. The callee therefore has one invocation (one onset) covering every +// call, so the onset's parent alone can't attribute work to an individual call. The caller sends +// its per-call span identity with each call so the callee can attribute every dispatch precisely. +// +// The parent/child relationships are asserted in the tail worker's test() handler (see +// jsrpc-pipelined-trace-test-tail.js). + +import { WorkerEntrypoint, RpcTarget } from 'cloudflare:workers'; + +class Counter extends RpcTarget { + #value = 0; + + increment(amount) { + this.#value += amount; + return this.#value; + } +} + +export class CounterService extends WorkerEntrypoint { + // Returns a stub. Calls the caller subsequently makes on it reuse this same session, and so are + // delivered to this same invocation. + async getCounter() { + return new Counter(); + } +} + +export default { + async test(controller, env, ctx) { + // Opens the session. + const counter = await env.CounterService.getCounter(); + + // Reuses the session opened above, so this is delivered to the same callee invocation. + const result = await counter.increment(5); + if (result !== 5) { + throw new Error(`Expected 5, got ${result}`); + } + }, +}; diff --git a/src/workerd/api/tests/jsrpc-pipelined-trace-test.wd-test b/src/workerd/api/tests/jsrpc-pipelined-trace-test.wd-test new file mode 100644 index 00000000000..afe766347ab --- /dev/null +++ b/src/workerd/api/tests/jsrpc-pipelined-trace-test.wd-test @@ -0,0 +1,31 @@ +using Workerd = import "/workerd/workerd.capnp"; + +# Test for JSRPC tracing: the callee invocation is parented to the caller's first per-call jsRpcCall +# span, and every callee call links to its corresponding caller span. Multiple calls share one +# session (and hence one callee invocation), so both forms of attribution are needed. +# Asserted in the tail worker's test() handler. + +const unitTests :Workerd.Config = ( + services = [ + (name = "jsrpc-pipelined-trace-test", worker = .mainWorker), + (name = "tail", worker = .tailWorker), + ], +); + +const mainWorker :Workerd.Worker = ( + modules = [ + (name = "worker", esModule = embed "jsrpc-pipelined-trace-test.js") + ], + compatibilityFlags = ["nodejs_compat", "experimental", "rpc"], + bindings = [ + (name = "CounterService", service = (name = "jsrpc-pipelined-trace-test", entrypoint = "CounterService")), + ], + streamingTails = ["tail"], +); + +const tailWorker :Workerd.Worker = ( + modules = [ + (name = "worker", esModule = embed "jsrpc-pipelined-trace-test-tail.js") + ], + compatibilityFlags = ["experimental", "nodejs_compat"], +); diff --git a/src/workerd/api/tests/sql-test-tail.js b/src/workerd/api/tests/sql-test-tail.js index 323441315d6..87a141e4dc5 100644 --- a/src/workerd/api/tests/sql-test-tail.js +++ b/src/workerd/api/tests/sql-test-tail.js @@ -29,6 +29,13 @@ export const test = { durable_object_storage_get: 18, durable_object_storage_transaction: 8, durable_object_subrequest: 48, + // Inner spans nested under durable_object_subrequest: jsRpcSession for RPC + // dispatches (one per call, client-side only -- the server's jsrpc-typed + // onset is its equivalent), fetch for fetch dispatches. + jsRpcSession: 36, + fetch: 12, + // jsRpcCall: 36 client-side + 36 server-side. + jsRpcCall: 72, durable_object_storage_deleteAll: 1, createStringTable: 4, runActorFunc: 4, diff --git a/src/workerd/api/tests/tail-worker-test.js b/src/workerd/api/tests/tail-worker-test.js index 3044f47ce44..fed0a74eeaf 100644 --- a/src/workerd/api/tests/tail-worker-test.js +++ b/src/workerd/api/tests/tail-worker-test.js @@ -83,13 +83,78 @@ export default { }, }; +// Split a concatenated string of top-level JSON objects into individual event +// JSON strings. Events are emitted back-to-back as `}{`, and individual events +// may themselves contain nested objects/arrays (e.g. attributes `info` arrays), +// so we split on brace depth while respecting string literals and escapes. +function splitEvents(str) { + const events = []; + let depth = 0; + let start = 0; + let inStr = false; + let esc = false; + for (let i = 0; i < str.length; i++) { + const c = str[i]; + if (esc) { + esc = false; + continue; + } + if (inStr) { + if (c === '\\') esc = true; + else if (c === '"') inStr = false; + continue; + } + if (c === '"') { + inStr = true; + } else if (c === '{') { + if (depth === 0) start = i; + depth++; + } else if (c === '}') { + depth--; + if (depth === 0) events.push(str.slice(start, i + 1)); + } + } + return events; +} + +// Produce a canonical form of an invocation's event stream that is tolerant of the one +// source of ordering nondeterminism: asynchronous span closes. +// +// The client-side jsRpcSession span closes asynchronously (when its background customEvent task +// settles, which is an async RPC round-trip), so the arrival order of that spanClose relative to +// subsequent synchronous events (e.g. the next spanOpen) is platform-dependent. spanClose events +// carry no span identity, so we can't reorder only that one close; instead we only normalize +// invocations that actually open a jsRpcSession. For those, we keep every non-spanClose event in +// its original order -- so ordering regressions in onset/spanOpen/attributes/log/return/outcome +// are still caught -- and gather the spanClose events into a sorted group (count preserved, so a +// missing close is still detected). All other invocations keep strict ordering, so their span +// lifecycles are checked exactly. Cross-invocation nesting is verified separately by buildTree +// via span IDs. +function canonicalizeEvents(eventsStr) { + if (!eventsStr.includes('"name":"jsRpcSession"')) return eventsStr; + const events = splitEvents(eventsStr); + if (events.length <= 1) return eventsStr; + const ordered = []; + const closes = []; + for (const e of events) { + if (JSON.parse(e).type === 'spanClose') closes.push(e); + else ordered.push(e); + } + closes.sort(); + return ordered.join('') + closes.join(''); +} + // Build a tree from the flat invocations array. // Each invocation tracks allSpanIds (its root span + all child spans from spanOpen events). // A child invocation's parentSpanId is matched against allSpanIds from invocations in the // same trace (same traceId) to find its parent. This handles both: // - Subrequests: parentSpanId is the caller's user span ID (sequential, from spanOpen) // - Hibernation: parentSpanId is the caller's invocation root span ID (from fromEntropy) -// Scoping by traceId avoids false matches from sequential span IDs that reset per invocation. +// Scoping by traceId avoids false matches across different traces. It does NOT disambiguate +// span IDs within a single trace: workerd-standalone assigns sequential span IDs that reset per +// invocation, so sibling callee invocations in the same trace reuse the same IDs and can be +// mis-linked under one another (see the jsrpcDoSubrequest expectation and its TODO). Production +// span IDs are random 64-bit and don't collide. // Stack frames in trace events name modules differently between the module // registries: the original registry uses the bare module name ('worker'), // while the new module registry uses the module's canonical URL @@ -104,7 +169,10 @@ function buildTree(invocations) { const byTraceId = new Map(); const nodes = []; for (const inv of invocations) { - const node = { events: normalizeStackFilenames(inv.events), children: [] }; + const node = { + events: canonicalizeEvents(normalizeStackFilenames(inv.events)), + children: [], + }; nodes.push({ inv, node }); if (!byTraceId.has(inv.traceId)) { byTraceId.set(inv.traceId, []); @@ -165,9 +233,11 @@ function verifyTraceIds(invocations) { } } -// Helper to create a tree node. +// Helper to create a tree node. Expected event strings are canonicalized the same +// way as the collected ones so the comparison is insensitive to async span-close +// ordering (see canonicalizeEvents). function n(events, children = []) { - return { events, children }; + return { events: canonicalizeEvents(events), children }; } // Regression check (WO-1436) for the "large-log" subject worker, which is tailed by this same @@ -254,33 +324,35 @@ const E = { wsThrow: '{"type":"onset","executionModel":"durableObject","spanId":"0000000000000000","entrypoint":"DurableObjectExample","durableObjectId":"DO_ID","scriptTags":[],"info":{"type":"fetch","method":"GET","url":"http://example.com/throw","headers":[{"name":"upgrade","value":"websocket"}]}}{"type":"exception","name":"Error","message":"boom","stack":" at DurableObjectExample.fetch (worker:25:13)"}{"type":"outcome","outcome":"exception","cpuTime":0,"wallTime":0}', - // jsrpc + // jsrpc -- the jsrpc-typed onset on the server already represents the session + // (delivered() to outcome), so no separate jsRpcSession user span is emitted + // server-side. Each method dispatch gets a jsRpcCall span. myActorJsrpc: - '{"type":"onset","executionModel":"durableObject","spanId":"0000000000000000","entrypoint":"MyActor","durableObjectId":"DO_ID","scriptTags":[],"info":{"type":"jsrpc"}}{"type":"log","level":"log","message":["baz"]}{"type":"attributes","info":[{"name":"jsrpc.method","value":"functionProperty"}]}{"type":"return"}{"type":"outcome","outcome":"ok","cpuTime":0,"wallTime":0}', + '{"type":"onset","executionModel":"durableObject","spanId":"0000000000000000","entrypoint":"MyActor","durableObjectId":"DO_ID","scriptTags":[],"info":{"type":"jsrpc"}}{"type":"log","level":"log","message":["baz"]}{"type":"spanOpen","name":"jsRpcCall","spanId":"0000000000000001"}{"type":"attributes","info":[{"name":"jsrpc.method","value":"functionProperty"}]}{"type":"attributes","info":[{"name":"jsrpc.method","value":"functionProperty"},{"name":"jsrpc.target_kind","value":"entrypoint"},{"name":"jsrpc.operation","value":"call"},{"name":"jsrpc.caller_span_id","value":"0000000000000005"}]}{"type":"spanClose","outcome":"ok"}{"type":"return"}{"type":"outcome","outcome":"ok","cpuTime":0,"wallTime":0}', jsrpcNonFunction: - '{"type":"onset","executionModel":"stateless","spanId":"0000000000000000","entrypoint":"MyService","scriptTags":[],"info":{"type":"jsrpc"}}{"type":"attributes","info":[{"name":"jsrpc.method","value":"nonFunctionProperty"}]}{"type":"log","level":"log","message":["bar"]}{"type":"log","level":"log","message":["foo"]}{"type":"outcome","outcome":"ok","cpuTime":0,"wallTime":0}', + '{"type":"onset","executionModel":"stateless","spanId":"0000000000000000","entrypoint":"MyService","scriptTags":[],"info":{"type":"jsrpc"}}{"type":"spanOpen","name":"jsRpcCall","spanId":"0000000000000001"}{"type":"attributes","info":[{"name":"jsrpc.method","value":"nonFunctionProperty"}]}{"type":"log","level":"log","message":["bar"]}{"type":"log","level":"log","message":["foo"]}{"type":"attributes","info":[{"name":"jsrpc.method","value":"nonFunctionProperty"},{"name":"jsrpc.target_kind","value":"entrypoint"},{"name":"jsrpc.operation","value":"call"},{"name":"jsrpc.caller_span_id","value":"0000000000000002"}]}{"type":"spanClose","outcome":"ok"}{"type":"outcome","outcome":"ok","cpuTime":0,"wallTime":0}', jsrpcGetCounter: - '{"type":"onset","executionModel":"stateless","spanId":"0000000000000000","entrypoint":"MyService","scriptTags":[],"info":{"type":"jsrpc"}}{"type":"attributes","info":[{"name":"jsrpc.method","value":"getCounter"}]}{"type":"log","level":"log","message":["bar"]}{"type":"log","level":"log","message":["getCounter called"]}{"type":"return"}{"type":"log","level":"log","message":["increment called on transient"]}{"type":"log","level":"log","message":["getValue called on transient"]}{"type":"outcome","outcome":"ok","cpuTime":0,"wallTime":0}', + '{"type":"onset","executionModel":"stateless","spanId":"0000000000000000","entrypoint":"MyService","scriptTags":[],"info":{"type":"jsrpc"}}{"type":"spanOpen","name":"jsRpcCall","spanId":"0000000000000001"}{"type":"attributes","info":[{"name":"jsrpc.method","value":"getCounter"}]}{"type":"log","level":"log","message":["bar"]}{"type":"log","level":"log","message":["getCounter called"]}{"type":"attributes","info":[{"name":"jsrpc.method","value":"getCounter"},{"name":"jsrpc.target_kind","value":"entrypoint"},{"name":"jsrpc.operation","value":"call"},{"name":"jsrpc.caller_span_id","value":"0000000000000007"}]}{"type":"spanClose","outcome":"ok"}{"type":"return"}{"type":"spanOpen","name":"jsRpcCall","spanId":"0000000000000002"}{"type":"log","level":"log","message":["increment called on transient"]}{"type":"attributes","info":[{"name":"jsrpc.method","value":"(this)"},{"name":"jsrpc.target_kind","value":"transient"},{"name":"jsrpc.operation","value":"call"},{"name":"jsrpc.caller_span_id","value":"0000000000000008"}]}{"type":"spanClose","outcome":"ok"}{"type":"spanOpen","name":"jsRpcCall","spanId":"0000000000000003"}{"type":"log","level":"log","message":["getValue called on transient"]}{"type":"attributes","info":[{"name":"jsrpc.method","value":"(this)"},{"name":"jsrpc.target_kind","value":"transient"},{"name":"jsrpc.operation","value":"call"},{"name":"jsrpc.caller_span_id","value":"0000000000000009"}]}{"type":"spanClose","outcome":"ok"}{"type":"outcome","outcome":"ok","cpuTime":0,"wallTime":0}', jsrpcDoSubrequest: - '{"type":"onset","executionModel":"stateless","spanId":"0000000000000000","scriptTags":[],"info":{"type":"custom"}}{"type":"spanOpen","name":"jsRpcSession","spanId":"0000000000000001"}{"type":"spanOpen","name":"durable_object_subrequest","spanId":"0000000000000002"}{"type":"spanOpen","name":"jsRpcSession","spanId":"0000000000000003"}{"type":"attributes","info":[{"name":"objectId","value":"af6dd8b6678e07bac992dae1bbbb3f385af19ebae7e5ea8c66d6341b246d3328"}]}{"type":"spanClose","outcome":"ok"}{"type":"spanClose","outcome":"ok"}{"type":"spanClose","outcome":"ok"}{"type":"outcome","outcome":"ok","cpuTime":0,"wallTime":0}', + '{"type":"onset","executionModel":"stateless","spanId":"0000000000000000","scriptTags":[],"info":{"type":"custom"}}{"type":"spanOpen","name":"jsRpcSession","spanId":"0000000000000001"}{"type":"spanOpen","name":"jsRpcCall","spanId":"0000000000000002"}{"type":"spanOpen","name":"durable_object_subrequest","spanId":"0000000000000003"}{"type":"spanOpen","name":"jsRpcSession","spanId":"0000000000000004"}{"type":"spanOpen","name":"jsRpcCall","spanId":"0000000000000005"}{"type":"attributes","info":[{"name":"jsrpc.target_kind","value":"fetcher"},{"name":"jsrpc.operation","value":"call"},{"name":"jsrpc.method","value":"nonFunctionProperty"}]}{"type":"spanClose","outcome":"ok"}{"type":"spanOpen","name":"jsRpcSession","spanId":"0000000000000006"}{"type":"spanOpen","name":"jsRpcCall","spanId":"0000000000000007"}{"type":"attributes","info":[{"name":"jsrpc.target_kind","value":"fetcher"},{"name":"jsrpc.operation","value":"call"},{"name":"jsrpc.method","value":"functionProperty"}]}{"type":"spanClose","outcome":"ok"}{"type":"spanClose","outcome":"ok"}{"type":"attributes","info":[{"name":"objectId","value":"af6dd8b6678e07bac992dae1bbbb3f385af19ebae7e5ea8c66d6341b246d3328"}]}{"type":"spanClose","outcome":"ok"}{"type":"spanOpen","name":"jsRpcCall","spanId":"0000000000000008"}{"type":"attributes","info":[{"name":"jsrpc.target_kind","value":"fetcher"},{"name":"jsrpc.operation","value":"call"},{"name":"jsrpc.method","value":"getCounter"}]}{"type":"spanClose","outcome":"ok"}{"type":"spanClose","outcome":"ok"}{"type":"spanOpen","name":"jsRpcCall","spanId":"0000000000000009"}{"type":"attributes","info":[{"name":"jsrpc.target_kind","value":"stub"},{"name":"jsrpc.operation","value":"call"},{"name":"jsrpc.method","value":"(this)"}]}{"type":"spanClose","outcome":"ok"}{"type":"attributes","info":[{"name":"jsrpc.target_kind","value":"stub"},{"name":"jsrpc.operation","value":"call"},{"name":"jsrpc.method","value":"(this)"}]}{"type":"spanClose","outcome":"ok"}{"type":"spanClose","outcome":"ok"}{"type":"outcome","outcome":"ok","cpuTime":0,"wallTime":0}', jsrpcDisposal: - '{"type":"onset","executionModel":"stateless","spanId":"0000000000000000","entrypoint":"disposal","scriptTags":[],"info":{"type":"custom"}}{"type":"spanOpen","name":"jsRpcSession","spanId":"0000000000000001"}{"type":"spanClose","outcome":"ok"}{"type":"spanOpen","name":"jsRpcSession","spanId":"0000000000000002"}{"type":"spanOpen","name":"jsRpcSession","spanId":"0000000000000003"}{"type":"spanClose","outcome":"ok"}{"type":"spanClose","outcome":"ok"}{"type":"outcome","outcome":"ok","cpuTime":0,"wallTime":0}', + '{"type":"onset","executionModel":"stateless","spanId":"0000000000000000","entrypoint":"disposal","scriptTags":[],"info":{"type":"custom"}}{"type":"spanOpen","name":"jsRpcSession","spanId":"0000000000000001"}{"type":"spanOpen","name":"jsRpcCall","spanId":"0000000000000002"}{"type":"spanOpen","name":"jsRpcCall","spanId":"0000000000000003"}{"type":"spanOpen","name":"jsRpcCall","spanId":"0000000000000004"}{"type":"attributes","info":[{"name":"jsrpc.target_kind","value":"fetcher"},{"name":"jsrpc.operation","value":"call"},{"name":"jsrpc.method","value":"testDispose"}]}{"type":"spanClose","outcome":"ok"}{"type":"attributes","info":[{"name":"jsrpc.target_kind","value":"stub"},{"name":"jsrpc.operation","value":"call"},{"name":"jsrpc.method","value":"(this)"}]}{"type":"spanClose","outcome":"ok"}{"type":"attributes","info":[{"name":"jsrpc.target_kind","value":"promise"},{"name":"jsrpc.operation","value":"call"},{"name":"jsrpc.method","value":"(this)"}]}{"type":"spanClose","outcome":"ok"}{"type":"spanClose","outcome":"ok"}{"type":"spanOpen","name":"jsRpcSession","spanId":"0000000000000005"}{"type":"spanOpen","name":"jsRpcCall","spanId":"0000000000000006"}{"type":"spanOpen","name":"jsRpcCall","spanId":"0000000000000007"}{"type":"attributes","info":[{"name":"jsrpc.target_kind","value":"fetcher"},{"name":"jsrpc.operation","value":"call"},{"name":"jsrpc.method","value":"leak"}]}{"type":"spanClose","outcome":"ok"}{"type":"attributes","info":[{"name":"jsrpc.target_kind","value":"stub"},{"name":"jsrpc.operation","value":"call"},{"name":"jsrpc.method","value":"(this)"}]}{"type":"spanClose","outcome":"ok"}{"type":"spanOpen","name":"jsRpcSession","spanId":"0000000000000008"}{"type":"spanOpen","name":"jsRpcCall","spanId":"0000000000000009"}{"type":"attributes","info":[{"name":"jsrpc.target_kind","value":"fetcher"},{"name":"jsrpc.operation","value":"call"},{"name":"jsrpc.method","value":"leakButReturnPlainObject"}]}{"type":"spanClose","outcome":"ok"}{"type":"spanClose","outcome":"ok"}{"type":"spanClose","outcome":"ok"}{"type":"outcome","outcome":"ok","cpuTime":0,"wallTime":0}', jsrpcNamedServiceBinding: - '{"type":"onset","executionModel":"stateless","spanId":"0000000000000000","entrypoint":"namedServiceBinding","scriptTags":[],"info":{"type":"custom"}}{"type":"spanOpen","name":"jsRpcSession","spanId":"0000000000000001"}{"type":"spanClose","outcome":"ok"}{"type":"outcome","outcome":"ok","cpuTime":0,"wallTime":0}', + '{"type":"onset","executionModel":"stateless","spanId":"0000000000000000","entrypoint":"namedServiceBinding","scriptTags":[],"info":{"type":"custom"}}{"type":"spanOpen","name":"jsRpcSession","spanId":"0000000000000001"}{"type":"spanOpen","name":"jsRpcCall","spanId":"0000000000000002"}{"type":"spanClose","outcome":"ok"}{"type":"attributes","info":[{"name":"jsrpc.target_kind","value":"fetcher"},{"name":"jsrpc.operation","value":"call"},{"name":"jsrpc.method","value":"neverReturn"}]}{"type":"spanClose","outcome":"ok"}{"type":"outcome","outcome":"ok","cpuTime":0,"wallTime":0}', jsrpcPortAbortCall: - '{"type":"onset","executionModel":"stateless","spanId":"0000000000000000","entrypoint":"portAbortCall","scriptTags":[],"info":{"type":"custom"}}{"type":"spanOpen","name":"durable_object_subrequest","spanId":"0000000000000001"}{"type":"attributes","info":[{"name":"objectId","value":"000000000000000000000000000000002aae183609eb9745ae9a5bb18ffb4793"}]}{"type":"spanClose","outcome":"ok"}{"type":"spanOpen","name":"durable_object_subrequest","spanId":"0000000000000002"}{"type":"attributes","info":[{"name":"objectId","value":"0100000000000000000000000000000067d1138346e9d456110d9e5cdfb6d564"}]}{"type":"spanClose","outcome":"ok"}{"type":"outcome","outcome":"ok","cpuTime":0,"wallTime":0}', + '{"type":"onset","executionModel":"stateless","spanId":"0000000000000000","entrypoint":"portAbortCall","scriptTags":[],"info":{"type":"custom"}}{"type":"spanOpen","name":"durable_object_subrequest","spanId":"0000000000000001"}{"type":"spanOpen","name":"jsRpcSession","spanId":"0000000000000002"}{"type":"spanOpen","name":"jsRpcCall","spanId":"0000000000000003"}{"type":"spanOpen","name":"jsRpcCall","spanId":"0000000000000004"}{"type":"spanOpen","name":"jsRpcCall","spanId":"0000000000000005"}{"type":"attributes","info":[{"name":"jsrpc.target_kind","value":"fetcher"},{"name":"jsrpc.operation","value":"call"},{"name":"jsrpc.method","value":"makePostAbortCallTester"}]}{"type":"spanClose","outcome":"ok"}{"type":"spanClose","outcome":"ok"}{"type":"attributes","info":[{"name":"objectId","value":"000000000000000000000000000000002aae183609eb9745ae9a5bb18ffb4793"}]}{"type":"spanClose","outcome":"ok"}{"type":"attributes","info":[{"name":"jsrpc.target_kind","value":"stub"},{"name":"jsrpc.operation","value":"call"},{"name":"jsrpc.method","value":"abort"}]}{"type":"spanClose","outcome":"ok"}{"type":"spanOpen","name":"durable_object_subrequest","spanId":"0000000000000006"}{"type":"spanOpen","name":"jsRpcSession","spanId":"0000000000000007"}{"type":"spanOpen","name":"jsRpcCall","spanId":"0000000000000008"}{"type":"attributes","info":[{"name":"jsrpc.target_kind","value":"promise"},{"name":"jsrpc.operation","value":"call"},{"name":"jsrpc.method","value":"(this)"}]}{"type":"spanClose","outcome":"ok"}{"type":"spanOpen","name":"jsRpcCall","spanId":"0000000000000009"}{"type":"spanOpen","name":"jsRpcCall","spanId":"000000000000000a"}{"type":"attributes","info":[{"name":"jsrpc.target_kind","value":"fetcher"},{"name":"jsrpc.operation","value":"call"},{"name":"jsrpc.method","value":"makePostAbortCallTester"}]}{"type":"spanClose","outcome":"ok"}{"type":"spanClose","outcome":"ok"}{"type":"attributes","info":[{"name":"objectId","value":"0100000000000000000000000000000067d1138346e9d456110d9e5cdfb6d564"}]}{"type":"spanClose","outcome":"ok"}{"type":"attributes","info":[{"name":"jsrpc.target_kind","value":"stub"},{"name":"jsrpc.operation","value":"call"},{"name":"jsrpc.method","value":"failCriticalSection"}]}{"type":"spanClose","outcome":"ok"}{"type":"attributes","info":[{"name":"jsrpc.target_kind","value":"promise"},{"name":"jsrpc.operation","value":"call"},{"name":"jsrpc.method","value":"(this)"}]}{"type":"spanClose","outcome":"ok"}{"type":"outcome","outcome":"ok","cpuTime":0,"wallTime":0}', jsrpcTestDispose: - '{"type":"onset","executionModel":"stateless","spanId":"0000000000000000","entrypoint":"MyService","scriptTags":[],"info":{"type":"jsrpc"}}{"type":"attributes","info":[{"name":"jsrpc.method","value":"leak"}]}{"type":"log","level":"log","message":["bar"]}{"type":"return"}{"type":"outcome","outcome":"ok","cpuTime":0,"wallTime":0}', + '{"type":"onset","executionModel":"stateless","spanId":"0000000000000000","entrypoint":"MyService","scriptTags":[],"info":{"type":"jsrpc"}}{"type":"spanOpen","name":"jsRpcCall","spanId":"0000000000000001"}{"type":"attributes","info":[{"name":"jsrpc.method","value":"leak"}]}{"type":"log","level":"log","message":["bar"]}{"type":"attributes","info":[{"name":"jsrpc.method","value":"leak"},{"name":"jsrpc.target_kind","value":"entrypoint"},{"name":"jsrpc.operation","value":"call"},{"name":"jsrpc.caller_span_id","value":"0000000000000006"}]}{"type":"spanClose","outcome":"ok"}{"type":"return"}{"type":"spanOpen","name":"jsRpcCall","spanId":"0000000000000002"}{"type":"attributes","info":[{"name":"jsrpc.method","value":"(this)"},{"name":"jsrpc.target_kind","value":"transient"},{"name":"jsrpc.operation","value":"call"},{"name":"jsrpc.caller_span_id","value":"0000000000000007"}]}{"type":"spanClose","outcome":"ok"}{"type":"outcome","outcome":"ok","cpuTime":0,"wallTime":0}', jsrpcExceptionI: - '{"type":"onset","executionModel":"stateless","spanId":"0000000000000000","entrypoint":"MyService","scriptTags":[],"info":{"type":"jsrpc"}}{"type":"attributes","info":[{"name":"jsrpc.method","value":"neverReturn"}]}{"type":"log","level":"log","message":["bar"]}{"type":"exception","name":"Error","message":"The Workers runtime canceled this request because it detected that your Worker\'s code had hung and would never generate a response. Refer to: https://developers.cloudflare.com/workers/observability/errors/"}{"type":"outcome","outcome":"exception","cpuTime":0,"wallTime":0}', + '{"type":"onset","executionModel":"stateless","spanId":"0000000000000000","entrypoint":"MyService","scriptTags":[],"info":{"type":"jsrpc"}}{"type":"spanOpen","name":"jsRpcCall","spanId":"0000000000000001"}{"type":"attributes","info":[{"name":"jsrpc.method","value":"neverReturn"}]}{"type":"log","level":"log","message":["bar"]}{"type":"exception","name":"Error","message":"The Workers runtime canceled this request because it detected that your Worker\'s code had hung and would never generate a response. Refer to: https://developers.cloudflare.com/workers/observability/errors/"}{"type":"attributes","info":[{"name":"jsrpc.method","value":"neverReturn"},{"name":"jsrpc.target_kind","value":"entrypoint"},{"name":"jsrpc.operation","value":"call"},{"name":"jsrpc.caller_span_id","value":"0000000000000002"}]}{"type":"spanClose","outcome":"ok"}{"type":"outcome","outcome":"exception","cpuTime":0,"wallTime":0}', jsrpcExceptionII: - '{"type":"onset","executionModel":"stateless","spanId":"0000000000000000","entrypoint":"MyService","scriptTags":[],"info":{"type":"jsrpc"}}{"type":"attributes","info":[{"name":"jsrpc.method","value":"leakButReturnPlainObject"}]}{"type":"log","level":"log","message":["bar"]}{"type":"return"}{"type":"exception","name":"Error","message":"The Workers runtime canceled this request because it detected that your Worker\'s code had hung and would never generate a response. Refer to: https://developers.cloudflare.com/workers/observability/errors/"}{"type":"outcome","outcome":"exception","cpuTime":0,"wallTime":0}', + '{"type":"onset","executionModel":"stateless","spanId":"0000000000000000","entrypoint":"MyService","scriptTags":[],"info":{"type":"jsrpc"}}{"type":"spanOpen","name":"jsRpcCall","spanId":"0000000000000001"}{"type":"attributes","info":[{"name":"jsrpc.method","value":"leakButReturnPlainObject"}]}{"type":"log","level":"log","message":["bar"]}{"type":"attributes","info":[{"name":"jsrpc.method","value":"leakButReturnPlainObject"},{"name":"jsrpc.target_kind","value":"entrypoint"},{"name":"jsrpc.operation","value":"call"},{"name":"jsrpc.caller_span_id","value":"0000000000000009"}]}{"type":"spanClose","outcome":"ok"}{"type":"return"}{"type":"exception","name":"Error","message":"The Workers runtime canceled this request because it detected that your Worker\'s code had hung and would never generate a response. Refer to: https://developers.cloudflare.com/workers/observability/errors/"}{"type":"outcome","outcome":"exception","cpuTime":0,"wallTime":0}', jsrpcExceptionIII: - '{"type":"onset","executionModel":"stateless","spanId":"0000000000000000","entrypoint":"MyService","scriptTags":[],"info":{"type":"jsrpc"}}{"type":"attributes","info":[{"name":"jsrpc.method","value":"testDispose"}]}{"type":"log","level":"log","message":["bar"]}{"type":"return"}{"type":"exception","name":"Error","message":"The Workers runtime canceled this request because it detected that your Worker\'s code had hung and would never generate a response. Refer to: https://developers.cloudflare.com/workers/observability/errors/"}{"type":"outcome","outcome":"exception","cpuTime":0,"wallTime":0}', + '{"type":"onset","executionModel":"stateless","spanId":"0000000000000000","entrypoint":"MyService","scriptTags":[],"info":{"type":"jsrpc"}}{"type":"spanOpen","name":"jsRpcCall","spanId":"0000000000000001"}{"type":"attributes","info":[{"name":"jsrpc.method","value":"testDispose"}]}{"type":"log","level":"log","message":["bar"]}{"type":"attributes","info":[{"name":"jsrpc.method","value":"testDispose"},{"name":"jsrpc.target_kind","value":"entrypoint"},{"name":"jsrpc.operation","value":"call"},{"name":"jsrpc.caller_span_id","value":"0000000000000002"}]}{"type":"spanClose","outcome":"ok"}{"type":"return"}{"type":"spanOpen","name":"jsRpcCall","spanId":"0000000000000002"}{"type":"spanOpen","name":"jsRpcCall","spanId":"0000000000000003"}{"type":"attributes","info":[{"name":"jsrpc.target_kind","value":"stub"},{"name":"jsrpc.operation","value":"call"},{"name":"jsrpc.method","value":"increment"}]}{"type":"spanClose","outcome":"ok"}{"type":"attributes","info":[{"name":"jsrpc.method","value":"(this)"},{"name":"jsrpc.target_kind","value":"transient"},{"name":"jsrpc.operation","value":"call"},{"name":"jsrpc.caller_span_id","value":"0000000000000003"}]}{"type":"spanClose","outcome":"ok"}{"type":"exception","name":"Error","message":"The Workers runtime canceled this request because it detected that your Worker\'s code had hung and would never generate a response. Refer to: https://developers.cloudflare.com/workers/observability/errors/"}{"type":"outcome","outcome":"exception","cpuTime":0,"wallTime":0}', jsrpcExceptionIV: - '{"type":"onset","executionModel":"durableObject","spanId":"0000000000000000","entrypoint":"MyActor","durableObjectId":"DO_ID","scriptTags":[],"info":{"type":"jsrpc"}}{"type":"log","level":"log","message":["baz"]}{"type":"attributes","info":[{"name":"jsrpc.method","value":"makePostAbortCallTester"}]}{"type":"return"}{"type":"exception","name":"Error","message":"test aborted by abort()"}{"type":"outcome","outcome":"exception","cpuTime":0,"wallTime":0}', + '{"type":"onset","executionModel":"durableObject","spanId":"0000000000000000","entrypoint":"MyActor","durableObjectId":"DO_ID","scriptTags":[],"info":{"type":"jsrpc"}}{"type":"log","level":"log","message":["baz"]}{"type":"spanOpen","name":"jsRpcCall","spanId":"0000000000000001"}{"type":"attributes","info":[{"name":"jsrpc.method","value":"makePostAbortCallTester"}]}{"type":"attributes","info":[{"name":"jsrpc.method","value":"makePostAbortCallTester"},{"name":"jsrpc.target_kind","value":"entrypoint"},{"name":"jsrpc.operation","value":"call"},{"name":"jsrpc.caller_span_id","value":"0000000000000003"}]}{"type":"spanClose","outcome":"ok"}{"type":"return"}{"type":"spanOpen","name":"jsRpcCall","spanId":"0000000000000002"}{"type":"attributes","info":[{"name":"jsrpc.method","value":"abort"},{"name":"jsrpc.target_kind","value":"transient"},{"name":"jsrpc.operation","value":"call"},{"name":"jsrpc.caller_span_id","value":"0000000000000004"}]}{"type":"spanClose","outcome":"ok"}{"type":"exception","name":"Error","message":"test aborted by abort()"}{"type":"outcome","outcome":"exception","cpuTime":0,"wallTime":0}', jsrpcExceptionV: - '{"type":"onset","executionModel":"durableObject","spanId":"0000000000000000","entrypoint":"MyActor","durableObjectId":"DO_ID","scriptTags":[],"info":{"type":"jsrpc"}}{"type":"log","level":"log","message":["baz"]}{"type":"attributes","info":[{"name":"jsrpc.method","value":"makePostAbortCallTester"}]}{"type":"return"}{"type":"exception","name":"Error","message":"test broken critical section","stack":" at worker:144:13"}{"type":"exception","name":"Error","message":"test broken critical section","stack":" at worker:144:13"}{"type":"outcome","outcome":"exception","cpuTime":0,"wallTime":0}', + '{"type":"onset","executionModel":"durableObject","spanId":"0000000000000000","entrypoint":"MyActor","durableObjectId":"DO_ID","scriptTags":[],"info":{"type":"jsrpc"}}{"type":"log","level":"log","message":["baz"]}{"type":"spanOpen","name":"jsRpcCall","spanId":"0000000000000001"}{"type":"attributes","info":[{"name":"jsrpc.method","value":"makePostAbortCallTester"}]}{"type":"attributes","info":[{"name":"jsrpc.method","value":"makePostAbortCallTester"},{"name":"jsrpc.target_kind","value":"entrypoint"},{"name":"jsrpc.operation","value":"call"},{"name":"jsrpc.caller_span_id","value":"0000000000000008"}]}{"type":"spanClose","outcome":"ok"}{"type":"return"}{"type":"spanOpen","name":"jsRpcCall","spanId":"0000000000000002"}{"type":"exception","name":"Error","message":"test broken critical section","stack":" at worker:144:13"}{"type":"exception","name":"Error","message":"test broken critical section","stack":" at worker:144:13"}{"type":"attributes","info":[{"name":"jsrpc.method","value":"failCriticalSection"},{"name":"jsrpc.target_kind","value":"transient"},{"name":"jsrpc.operation","value":"call"},{"name":"jsrpc.caller_span_id","value":"0000000000000009"}]}{"type":"spanClose","outcome":"ok"}{"type":"outcome","outcome":"exception","cpuTime":0,"wallTime":0}', // cacheMode cacheMode: @@ -366,17 +438,35 @@ const expected = [ n(E.localAddressViaServiceBinding), n(E.connectTarget), - // jsrpc DO subrequest test: caller has children (MyService + MyActor DO calls) + // JSRPC callee invocations are children of the per-call jsRpcCall spans that opened their + // sessions. The exact nesting below still contains a workerd-standalone artifact: sibling callee + // invocations re-use sequential span IDs starting at 0000000000000001, so buildTree's + // spanId-based parent lookup can mis-link them under each other. Production span IDs are random + // 64-bit and don't collide. + // TODO: Once buildTree (in instrumentation-test-helper.js) disambiguates spans + // by (traceId, spanId) or otherwise handles workerd-standalone's sequential + // span-ID reuse, replace these with the natural shape, in which every callee is a + // direct child of its caller: + // n(E.jsrpcDoSubrequest, [ + // n(E.myActorJsrpc), + // n(E.jsrpcGetCounter), + // n(E.jsrpcNonFunction), + // ]), + // n(E.jsrpcNamedServiceBinding, [n(E.jsrpcExceptionI)]), + // n(E.jsrpcDisposal, [ + // n(E.jsrpcExceptionII), + // n(E.jsrpcExceptionIII), + // n(E.jsrpcTestDispose), + // ]), + // n(E.jsrpcPortAbortCall, [n(E.jsrpcExceptionIV), n(E.jsrpcExceptionV)]), n(E.jsrpcDoSubrequest, [ n(E.myActorJsrpc), - n(E.jsrpcGetCounter), - n(E.jsrpcNonFunction), + n(E.jsrpcGetCounter, [n(E.jsrpcNonFunction)]), ]), n(E.jsrpcNamedServiceBinding, [n(E.jsrpcExceptionI)]), n(E.jsrpcDisposal, [ + n(E.jsrpcTestDispose, [n(E.jsrpcExceptionIII)]), n(E.jsrpcExceptionII), - n(E.jsrpcExceptionIII), - n(E.jsrpcTestDispose), ]), n(E.jsrpcPortAbortCall, [n(E.jsrpcExceptionIV), n(E.jsrpcExceptionV)]), diff --git a/src/workerd/api/worker-rpc.c++ b/src/workerd/api/worker-rpc.c++ index 932a55291c0..64308ed887b 100644 --- a/src/workerd/api/worker-rpc.c++ +++ b/src/workerd/api/worker-rpc.c++ @@ -120,10 +120,12 @@ struct DeserializeResult { }; // Call to construct a JS value from an `rpc::JsValue`. -DeserializeResult deserializeJsValue(jsg::Lock& js, rpc::JsValue::Reader reader) { +DeserializeResult deserializeJsValue( + jsg::Lock& js, rpc::JsValue::Reader reader, kj::Maybe originatingCall) { auto disposalGroup = kj::heap(); - RpcDeserializerExternalHandler externalHandler(reader.getExternals(), *disposalGroup); + RpcDeserializerExternalHandler externalHandler( + reader.getExternals(), *disposalGroup, kj::mv(originatingCall)); jsg::Deserializer deserializer(js, reader.getV8Serialized(), kj::none, kj::none, jsg::Deserializer::Options{ @@ -148,9 +150,11 @@ DeserializeResult deserializeJsValue(jsg::Lock& js, rpc::JsValue::Reader reader) // Does deserializeJsValue() and then adds a `dispose()` method to the returned object (if it is // an object) which disposes all stubs therein. -jsg::JsValue deserializeRpcReturnValue( - jsg::Lock& js, rpc::JsRpcTarget::CallResults::Reader callResults) { - auto [value, disposalGroup] = deserializeJsValue(js, callResults.getResult()); +jsg::JsValue deserializeRpcReturnValue(jsg::Lock& js, + rpc::JsRpcTarget::CallResults::Reader callResults, + kj::Maybe originatingCall) { + auto [value, disposalGroup] = + deserializeJsValue(js, callResults.getResult(), kj::mv(originatingCall)); // If the object had a disposer on the callee side, it will run when we discard the callPipeline, // so attach that to the disposal group on the caller side. If the returned object did NOT have @@ -218,13 +222,23 @@ void tryCallDisposeMethod(jsg::Lock& js, jsg::JsValue value) { }); } +kj::Maybe> ownOriginatingCall( + kj::Maybe originatingCall) { + KJ_IF_SOME(call, originatingCall) { + return IoContext::current().addObject(kj::heap(kj::mv(call))); + } + return kj::none; +} + } // namespace JsRpcPromise::JsRpcPromise(jsg::JsRef inner, kj::Own weakRefParam, - IoOwn pipeline) + IoOwn pipeline, + kj::Maybe originatingCall) : inner(kj::mv(inner)), weakRef(kj::mv(weakRefParam)), + originatingCall(ownOriginatingCall(kj::mv(originatingCall))), state(Pending{kj::mv(pipeline)}) { KJ_REQUIRE(weakRef->ref == kj::none); weakRef->ref = *this; @@ -259,13 +273,17 @@ void JsRpcPromise::dispose(jsg::Lock& js) { static rpc::JsRpcTarget::Client makeJsRpcTargetForSingleLoopbackCall( jsg::Lock& js, jsg::JsObject obj); -rpc::JsRpcTarget::Client JsRpcPromise::getClientForOneCall( +JsRpcClientProvider::ClientForOneCall JsRpcPromise::getClientForOneCall( jsg::Lock& js, kj::Vector& path) { // (Don't extend `path` because we're the root.) - + auto callSpanParents = + originatingCall.map([](IoOwn& p) { return p->addRef(); }); KJ_SWITCH_ONEOF(state) { KJ_CASE_ONEOF(pending, Pending) { - return pending.pipeline->getCallPipeline(); + return { + .client = pending.pipeline->getCallPipeline(), + .callSpanParents = kj::mv(callSpanParents), + }; } KJ_CASE_ONEOF(resolved, Resolved) { // Dereference `ctxCheck` just to verify we're running in the correct context. (If not, @@ -290,7 +308,8 @@ rpc::JsRpcTarget::Client JsRpcPromise::getClientForOneCall( // The easiest way to make this all just work is... to actually wrap the value in a one-off // RPC stub, and make a real RPC on it. - return js.withinHandleScope([&]() -> rpc::JsRpcTarget::Client { + return { + .client = js.withinHandleScope([&]() -> rpc::JsRpcTarget::Client { auto value = jsg::JsValue(resolved.result.getHandle(js)); KJ_IF_SOME(obj, value.tryCast()) { @@ -304,16 +323,18 @@ rpc::JsRpcTarget::Client JsRpcPromise::getClientForOneCall( } else { JSG_FAIL_REQUIRE(TypeError, "Can't pipeline on RPC that did not return an object."); } - }); + }), + .callSpanParents = kj::mv(callSpanParents), + }; } KJ_CASE_ONEOF(disposed, Disposed) { - return JSG_KJ_EXCEPTION(FAILED, Error, "RPC promise used after being disposed."); + return {.client = JSG_KJ_EXCEPTION(FAILED, Error, "RPC promise used after being disposed.")}; } } KJ_UNREACHABLE; } -rpc::JsRpcTarget::Client JsRpcProperty::getClientForOneCall( +JsRpcClientProvider::ClientForOneCall JsRpcProperty::getClientForOneCall( jsg::Lock& js, kj::Vector& path) { auto result = parent->getClientForOneCall(js, path); path.add(name); @@ -327,12 +348,68 @@ struct JsRpcPromiseAndPipeline { kj::Own weakRef; rpc::JsRpcTarget::CallResults::Pipeline pipeline; + // The jsRpcCall of the call that produced this promise, so calls pipelined on the promise + // nest under it. Absent when untraced, and on the error paths where no call span was opened. + kj::Maybe originatingCall; + jsg::Ref asJsRpcPromise(jsg::Lock& js) && { return js.alloc(jsg::JsRef(js, promise), kj::mv(weakRef), - IoContext::current().addObject(kj::heap(kj::mv(pipeline)))); + IoContext::current().addObject(kj::heap(kj::mv(pipeline))), kj::mv(originatingCall)); } }; +enum class JsRpcOperation { + CALL, + GET_PROPERTY, +}; + +static void setJsRpcCallSpanTags(TraceContext& span, + JsRpcClientProvider& parent, + kj::Maybe name, + kj::ArrayPtr path, + JsRpcOperation operation) { + // Guard tag population on observation: `setTag` no-ops when unobserved, but its arguments + // (notably the `kj::strArray` join below) are evaluated eagerly, so skip them entirely on + // the untraced hot path. + if (span.isObserved()) { + span.setTag("jsrpc.target_kind"_kjc, parent.getRpcTargetKind()); + span.setTag( + "jsrpc.operation"_kjc, operation == JsRpcOperation::CALL ? "call"_kjc : "getProperty"_kjc); + // Empty path and no name means the stub itself is being invoked as a function. + if (path.size() == 0 && name == kj::none) { + span.setTag("jsrpc.method"_kjc, "(this)"_kjc); + } else if (path.size() == 0) { + KJ_IF_SOME(n, name) { + span.setTag("jsrpc.method"_kjc, n.asPtr()); + } + } else { + kj::Vector fullPath(path.size() + 1); + fullPath.addAll(path); + KJ_IF_SOME(n, name) { + fullPath.add(n); + } + span.setTag("jsrpc.method"_kjc, kj::strArray(fullPath, ".")); + } + } +} + +// Creates the per-call client-side `jsRpcCall` span, nested under `callSpanParents` when +// provided (set by Fetcher for root sessions and by JsRpcStub for follow-up calls on returned +// stubs/promises) or under the current async context's spans otherwise. +static TraceContext makeJsRpcCallSpan(IoContext& ioContext, + JsRpcClientProvider& parent, + kj::Maybe name, + kj::ArrayPtr path, + kj::Maybe callSpanParents, + JsRpcOperation operation) { + TraceContextParent parents = kj::mv(callSpanParents).orDefault([&] { + return TraceContextParent(ioContext.getCurrentTraceSpan(), ioContext.getCurrentUserTraceSpan()); + }); + TraceContext span = parents.newChild("jsRpcCall"_kjc); + setJsRpcCallSpanTags(span, parent, name, path, operation); + return span; +} + // Core implementation of making an RPC call, reusable for many cases below. JsRpcPromiseAndPipeline callImpl(jsg::Lock& js, JsRpcClientProvider& parent, @@ -358,13 +435,25 @@ JsRpcPromiseAndPipeline callImpl(jsg::Lock& js, try { return js.tryCatch([&]() -> JsRpcPromiseAndPipeline { - // `path` will be filled in with the path of property names leading from the stub represented by - // `client` to the specific property / method that we're trying to invoke. - kj::Vector path; - auto client = parent.getClientForOneCall(js, path); - auto& ioContext = IoContext::current(); + // `path` is filled in with the chain of property names leading to the method. + kj::Vector path; + auto oneCall = parent.getClientForOneCall(js, path); + auto client = kj::mv(oneCall.client); + + // Per-call dispatch span, captured into the awaitIo callback below so it stays + // open until the response settles. + auto operation = maybeArgs != kj::none ? JsRpcOperation::CALL : JsRpcOperation::GET_PROPERTY; + TraceContext jsRpcCallSpan; + KJ_IF_SOME(span, oneCall.callSpan) { + jsRpcCallSpan = kj::mv(span); + setJsRpcCallSpanTags(jsRpcCallSpan, parent, name, path.asPtr(), operation); + } else { + jsRpcCallSpan = makeJsRpcCallSpan( + ioContext, parent, name, path.asPtr(), kj::mv(oneCall.callSpanParents), operation); + } + KJ_IF_SOME(lock, ioContext.waitForOutputLocksIfNecessary()) { // Replace the client with a promise client that will delay the call until the output gate // is open. @@ -373,6 +462,13 @@ JsRpcPromiseAndPipeline callImpl(jsg::Lock& js, auto builder = client.callRequest(); + // Tell the callee which caller span corresponds to this dispatch. A session carries many + // calls (e.g. calls pipelined on a returned stub), so the context propagated when the session + // opened identifies only the first call. Yields kj::none when untraced. + KJ_IF_SOME(callerSpanContext, jsRpcCallSpan.getUserSpanParent().toSpanContext()) { + callerSpanContext.toCapnp(builder.initCallerSpanContext()); + } + // This code here is slightly overcomplicated in order to avoid pushing anything to the // kj::Vector in the common case that the parent path is empty. I'm probably trying too hard // but oh well. @@ -434,12 +530,21 @@ JsRpcPromiseAndPipeline callImpl(jsg::Lock& js, // here, which is filled in later on to point at the JsRpcPromise, if and when one is created. auto weakRef = kj::atomicRefcounted(); + // Capture this call's span parents (adds refs, doesn't consume the span) so that calls + // pipelined on the returned promise nest under it, mirroring the stub handling below. Only + // retained when traced, so the untraced path holds no span state. + auto originatingCall = jsRpcCallSpan.getSpanParentsIfObserved(); + // RemotePromise lets us consume its pipeline and promise portions independently; we consume // the promise here and we consume the pipeline below, both via kj::mv(). auto jsPromise = ioContext.awaitIo(js, kj::mv(callResult), - [weakRef = kj::atomicAddRef(*weakRef)](jsg::Lock& js, + [weakRef = kj::atomicAddRef(*weakRef), jsRpcCallSpan = kj::mv(jsRpcCallSpan)]( + jsg::Lock& js, capnp::Response response) mutable -> jsg::Value { - auto jsResult = deserializeRpcReturnValue(js, response); + // Stubs in the response record this call as their originating call so that + // follow-up calls on those stubs nest under it (only when traced). + auto jsResult = + deserializeRpcReturnValue(js, response, jsRpcCallSpan.getSpanParentsIfObserved()); if (weakRef->disposed) { // The promise was explicitly disposed before it even resolved. This means we must dispose @@ -458,6 +563,7 @@ JsRpcPromiseAndPipeline callImpl(jsg::Lock& js, .promise = jsg::JsPromise(js.wrapSimplePromise(kj::mv(jsPromise))), .weakRef = kj::mv(weakRef), .pipeline = kj::mv(callResult), + .originatingCall = kj::mv(originatingCall), }; }, [&](jsg::Value error) -> JsRpcPromiseAndPipeline { // Probably a serialization error. Need to convert to an async error since we never throw @@ -585,21 +691,25 @@ kj::Maybe> JsRpcPromise::getProperty(jsg::Lock& js, kj:: JsRpcStub::JsRpcStub(IoOwn capnpClient, RpcStubDisposalGroup& disposalGroup, - jsg::ExternalMemoryAdjustment externalMemoryAdjustment) + jsg::ExternalMemoryAdjustment externalMemoryAdjustment, + kj::Maybe originatingCall) : capnpClient(kj::mv(capnpClient)), disposalGroup(disposalGroup), - externalMemoryAdjustment(kj::mv(externalMemoryAdjustment)) { + externalMemoryAdjustment(kj::mv(externalMemoryAdjustment)), + originatingCall(ownOriginatingCall(kj::mv(originatingCall))) { disposalGroup.list.add(*this); } JsRpcStub::JsRpcStub(IoOwn capnpClient, IoOwn rpcChannel, RpcStubDisposalGroup& disposalGroup, - jsg::ExternalMemoryAdjustment externalMemoryAdjustment) + jsg::ExternalMemoryAdjustment externalMemoryAdjustment, + kj::Maybe originatingCall) : capnpClient(kj::mv(capnpClient)), rpcChannel(kj::mv(rpcChannel)), disposalGroup(disposalGroup), - externalMemoryAdjustment(kj::mv(externalMemoryAdjustment)) { + externalMemoryAdjustment(kj::mv(externalMemoryAdjustment)), + originatingCall(ownOriginatingCall(kj::mv(originatingCall))) { disposalGroup.list.add(*this); } @@ -713,10 +823,14 @@ kj::Maybe> JsRpcStub::getRpcChannel(IoCont } } -rpc::JsRpcTarget::Client JsRpcStub::getClientForOneCall( +JsRpcClientProvider::ClientForOneCall JsRpcStub::getClientForOneCall( jsg::Lock& js, kj::Vector& path) { // (Don't extend `path` because we're the root.) - return getClient(); + return { + .client = getClient(), + .callSpanParents = + originatingCall.map([](IoOwn& p) { return p->addRef(); }), + }; } jsg::Ref JsRpcStub::dup(jsg::Lock& js) { @@ -931,10 +1045,12 @@ jsg::Ref JsRpcStub::deserialize( KJ_IF_SOME(c, channel) { return js.alloc(ioctx.addObject(kj::heap(rpcTarget.getCap())), - ioctx.addObject(kj::mv(c)), externalHandler.getDisposalGroup(), kj::mv(externalMemory)); + ioctx.addObject(kj::mv(c)), externalHandler.getDisposalGroup(), kj::mv(externalMemory), + externalHandler.getOriginatingCall()); } else { return js.alloc(ioctx.addObject(kj::heap(rpcTarget.getCap())), - externalHandler.getDisposalGroup(), kj::mv(externalMemory)); + externalHandler.getDisposalGroup(), kj::mv(externalMemory), + externalHandler.getOriginatingCall()); } } else KJ_IF_SOME(storedHandler, kj::tryDowncast(handler)) { @@ -1101,6 +1217,10 @@ class JsRpcTargetBase: public rpc::JsRpcTarget::Server { // Returns true if the given name cannot be used as a method on this type. virtual bool isReservedName(kj::StringPtr name) = 0; + // Tracing tag value for jsrpc.target_kind on the server-side per-call span + // (see JsRpcClientProvider::getRpcTargetKind for the client-side equivalent). + virtual kj::LiteralStringConst getTargetKind() = 0; + kj::Promise callImpl(Worker::Lock& lock, IoContext& ctx, CallContext callContext) { jsg::Lock& js = lock; auto params = callContext.getParams(); @@ -1128,6 +1248,25 @@ class JsRpcTargetBase: public rpc::JsRpcTarget::Server { } } + // Server-side jsRpcCall, attached to the dispatch promise below so it stays + // open through JS invocation and result serialization. + auto jsRpcCallSpan = ctx.makeUserTraceSpan("jsRpcCall"_kjc); + jsRpcCallSpan.setTag("jsrpc.method"_kjc, methodNameForTrace.asPtr()); + jsRpcCallSpan.setTag("jsrpc.target_kind"_kjc, getTargetKind()); + jsRpcCallSpan.setTag("jsrpc.operation"_kjc, + params.getOperation().isGetProperty() ? "getProperty"_kjc : "call"_kjc); + + // Link this dispatch to the caller's per-call span. This span stays a child of its own + // invocation root, so that a consumer reading this invocation's tail stream can always resolve + // the parent. The link is what attributes the work to an individual call: one session (and so + // one invocation) carries many calls, e.g. calls pipelined on a returned stub. + if (jsRpcCallSpan.isObserved() && params.hasCallerSpanContext()) { + auto callerContext = tracing::SpanContext::fromCapnp(params.getCallerSpanContext()); + KJ_IF_SOME(callerSpanId, callerContext.getSpanId()) { + jsRpcCallSpan.setTag("jsrpc.caller_span_id"_kjc, callerSpanId.toGoString()); + } + } + maybeSetJsRpcInfo(ctx, methodNameForTrace); auto targetInfo = getTargetInfo(lock, ctx); @@ -1254,42 +1393,50 @@ class JsRpcTargetBase: public rpc::JsRpcTarget::Server { } }; - switch (op.which()) { - case rpc::JsRpcTarget::CallParams::Operation::CALL_WITH_ARGS: { - // Note that using isFunctionForRpc(js, propHandle) here would be incorrect, since that - // decides whether it is a function *that can be serialized as a stub*. JsRpcProperty - // is (at present) considered non-serializable in itself, but when traversing the - // pipeline path, we may have descended into a stub and its properties, thus we could - // actually be invoking a JsRpcProperty here. As long as it is in fact callable, we will - // allow it. - JSG_REQUIRE(propHandle->IsFunction(), TypeError, - kj::str("\"", methodNameForTrace, "\" is not a function.")); - auto fn = propHandle.As(); - - kj::Maybe args; - if (op.hasCallWithArgs()) { - args = op.getCallWithArgs(); - } + auto dispatch = [&]() -> kj::Promise { + switch (op.which()) { + case rpc::JsRpcTarget::CallParams::Operation::CALL_WITH_ARGS: { + // Note that using isFunctionForRpc(js, propHandle) here would be incorrect, since that + // decides whether it is a function *that can be serialized as a stub*. JsRpcProperty + // is (at present) considered non-serializable in itself, but when traversing the + // pipeline path, we may have descended into a stub and its properties, thus we could + // actually be invoking a JsRpcProperty here. As long as it is in fact callable, we will + // allow it. + JSG_REQUIRE(propHandle->IsFunction(), TypeError, + kj::str("\"", methodNameForTrace, "\" is not a function.")); + auto fn = propHandle.As(); + + kj::Maybe args; + if (op.hasCallWithArgs()) { + args = op.getCallWithArgs(); + } - InvocationResult invocationResult; - KJ_IF_SOME(envCtx, targetInfo.envCtx) { - invocationResult = invokeFnInsertingEnvCtx( - js, methodNameForTrace, fn, thisArg, args, envCtx.env, envCtx.ctx); - } else { - invocationResult = invokeFn(js, fn, thisArg, args); + // Record this call's span on any stubs/callbacks passed as arguments, so that if the + // callee invokes them, the follow-up jsRpcCall nests under this call's span (mirrors the + // return-value handling in callImpl). Only recorded when traced. + InvocationResult invocationResult; + KJ_IF_SOME(envCtx, targetInfo.envCtx) { + invocationResult = invokeFnInsertingEnvCtx(js, methodNameForTrace, fn, thisArg, args, + envCtx.env, envCtx.ctx, jsRpcCallSpan.getSpanParentsIfObserved()); + } else { + invocationResult = + invokeFn(js, fn, thisArg, args, jsRpcCallSpan.getSpanParentsIfObserved()); + } + + // We have a function, so let's call it and serialize the result for RPC. + // If the function returns a promise we will wait for the promise to finish so we can + // serialize the result. + return handleResult(kj::mv(invocationResult)); } - // We have a function, so let's call it and serialize the result for RPC. - // If the function returns a promise we will wait for the promise to finish so we can - // serialize the result. - return handleResult(kj::mv(invocationResult)); + case rpc::JsRpcTarget::CallParams::Operation::GET_PROPERTY: + return handleResult({.returnValue = propHandle}); } - case rpc::JsRpcTarget::CallParams::Operation::GET_PROPERTY: - return handleResult({.returnValue = propHandle}); - } + KJ_FAIL_ASSERT("unknown JsRpcTarget::CallParams::Operation", (uint)op.which()); + }; - KJ_FAIL_ASSERT("unknown JsRpcTarget::CallParams::Operation", (uint)op.which()); + return dispatch().attach(kj::mv(jsRpcCallSpan)); } struct GetPropResult { @@ -1442,10 +1589,11 @@ class JsRpcTargetBase: public rpc::JsRpcTarget::Server { static InvocationResult invokeFn(jsg::Lock& js, v8::Local fn, v8::Local thisArg, - kj::Maybe args) { + kj::Maybe args, + kj::Maybe originatingCall) { // We received arguments from the client, deserialize them back to JS. KJ_IF_SOME(a, args) { - auto [value, disposalGroup] = deserializeJsValue(js, a); + auto [value, disposalGroup] = deserializeJsValue(js, a, kj::mv(originatingCall)); auto args = KJ_REQUIRE_NONNULL( value.tryCast(), "expected JsArray when deserializing arguments."); // Call() expects a `Local []`... so we populate an array. @@ -1476,7 +1624,8 @@ class JsRpcTargetBase: public rpc::JsRpcTarget::Server { v8::Local thisArg, kj::Maybe args, v8::Local env, - jsg::JsObject ctx) { + jsg::JsObject ctx, + kj::Maybe originatingCall) { // Determine the function arity (how many parameters it was declared to accept) by reading the // `.length` attribute. auto arity = js.withinHandleScope([&]() { @@ -1503,7 +1652,7 @@ class JsRpcTargetBase: public rpc::JsRpcTarget::Server { kj::Maybe argsArrayFromClient; size_t argCountFromClient = 0; KJ_IF_SOME(a, args) { - auto [value, disposalGroup] = deserializeJsValue(js, a); + auto [value, disposalGroup] = deserializeJsValue(js, a, kj::mv(originatingCall)); auto array = KJ_REQUIRE_NONNULL( value.tryCast(), "expected JsArray when deserializing arguments."); @@ -1658,6 +1807,10 @@ class TransientJsRpcTarget final: public JsRpcTargetBase { return false; } + kj::LiteralStringConst getTargetKind() override { + return "transient"_kjc; + } + void maybeSetJsRpcInfo(IoContext& ctx, const kj::ConstString& methodNameForTrace) override {} }; @@ -2112,6 +2265,10 @@ class EntrypointJsRpcTarget final: public JsRpcTargetBase { return false; } + kj::LiteralStringConst getTargetKind() override { + return "entrypoint"_kjc; + } + void maybeSetJsRpcInfo(IoContext& ctx, const kj::ConstString& methodNameForTrace) override { KJ_IF_SOME(tracer, ctx.getWorkerTracer()) { tracer.setJsRpcInfo(ctx.getInvocationSpanContext(), ctx.now(), methodNameForTrace); @@ -2136,6 +2293,11 @@ kj::Promise JsRpcSessionCustomEvent::run( incomingRequest->drain(waitUntilTasks, kj::mv(incomingRequest)); }); + // No server-side user span: the jsrpc-typed onset already represents the session + // (delivered() to outcome). The internal span is still emitted for the legacy + // buffered tail. + auto jsRpcSessionInternalSpan = ioctx.makeTraceSpan("jsRpcSession"_kjc); + EntrypointJsRpcTarget target(ioctx, entrypointName, kj::mv(versionInfo), kj::mv(props), kj::mv(wrapperModule), mapAddRef(incomingRequest->getWorkerTracer()), isDynamicDispatch); capnp::RevocableServer revocableTarget(target); diff --git a/src/workerd/api/worker-rpc.h b/src/workerd/api/worker-rpc.h index 6cc4a4cb226..4cd303a0746 100644 --- a/src/workerd/api/worker-rpc.h +++ b/src/workerd/api/worker-rpc.h @@ -110,10 +110,14 @@ class RpcStubDisposalGroup; // handle RPC specially should use this. class RpcDeserializerExternalHandler final: public jsg::Deserializer::ExternalHandler { public: - RpcDeserializerExternalHandler( - capnp::List::Reader externals, RpcStubDisposalGroup& disposalGroup) + // `originatingCall`, when present, is recorded on any stubs deserialized so that + // follow-up calls on those stubs nest under the call that returned them. + RpcDeserializerExternalHandler(capnp::List::Reader externals, + RpcStubDisposalGroup& disposalGroup, + kj::Maybe originatingCall) : externals(externals), - disposalGroup(disposalGroup) {} + disposalGroup(disposalGroup), + originatingCall(kj::mv(originatingCall)) {} ~RpcDeserializerExternalHandler() noexcept(false); // Read and return the next external. @@ -125,12 +129,17 @@ class RpcDeserializerExternalHandler final: public jsg::Deserializer::ExternalHa return disposalGroup; } + kj::Maybe getOriginatingCall() { + return originatingCall.map([](TraceContextParent& p) { return p.addRef(); }); + } + private: capnp::List::Reader externals; uint i = 0; kj::UnwindDetector unwindDetector; RpcStubDisposalGroup& disposalGroup; + kj::Maybe originatingCall; }; // Base class for objects which can be sent over RPC, but doing so actually sends a stub which @@ -155,12 +164,28 @@ class JsRpcTarget: public jsg::Object { // it's only a C++ class used to abstract how to get a capnp client out of the object. class JsRpcClientProvider: public jsg::Object { public: + // Result of resolving a stub for one call's worth of dispatch. + struct ClientForOneCall { + rpc::JsRpcTarget::Client client; + // Spans to parent the per-call jsRpcCall under. Root Fetchers use the newly-opened + // jsRpcSession as the parent; follow-up calls on returned stubs/promises use the call that + // produced them. + kj::Maybe callSpanParents; + + // The per-call span may be opened while resolving a root Fetcher so its user span can be + // propagated as the callee invocation's parent before the session client is constructed. + kj::Maybe callSpan; + }; + // Get a capnp client that can be used to dispatch one call. // // If this isn't the root object (i.e. this is a JsRpcProperty), the property path starting from // the root object will be appended to `path`. - virtual rpc::JsRpcTarget::Client getClientForOneCall( - jsg::Lock& js, kj::Vector& path) = 0; + virtual ClientForOneCall getClientForOneCall(jsg::Lock& js, kj::Vector& path) = 0; + + // Tracing tag value for jsrpc.target_kind on the client-side per-call span + // (see JsRpcTargetBase::getTargetKind for the server-side equivalent). + virtual kj::LiteralStringConst getRpcTargetKind() = 0; }; class JsRpcProperty; @@ -186,14 +211,18 @@ class JsRpcPromise: public JsRpcClientProvider { JsRpcPromise(jsg::JsRef inner, kj::Own weakRef, - IoOwn pipeline); + IoOwn pipeline, + kj::Maybe originatingCall); ~JsRpcPromise() noexcept(false); void resolve(jsg::Lock& js, jsg::JsValue result); void dispose(jsg::Lock& js); - rpc::JsRpcTarget::Client getClientForOneCall( - jsg::Lock& js, kj::Vector& path) override; + ClientForOneCall getClientForOneCall(jsg::Lock& js, kj::Vector& path) override; + + kj::LiteralStringConst getRpcTargetKind() override { + return "promise"_kjc; + } // Expect that the call is itself going to return a function... and call that. jsg::Ref call(const v8::FunctionCallbackInfo& args); @@ -233,6 +262,10 @@ class JsRpcPromise: public JsRpcClientProvider { jsg::JsRef inner; kj::Own weakRef; + // The jsRpcCall of the call that produced this promise, used to parent follow-up calls + // pipelined on the promise under it (mirrors JsRpcStub::originatingCall). Only set when traced. + kj::Maybe> originatingCall; + struct Pending { IoOwn pipeline; }; @@ -277,8 +310,13 @@ class JsRpcProperty: public JsRpcClientProvider { name(kj::mv(name)), depth(depth) {} - rpc::JsRpcTarget::Client getClientForOneCall( - jsg::Lock& js, kj::Vector& path) override; + ClientForOneCall getClientForOneCall(jsg::Lock& js, kj::Vector& path) override; + + // Forward to parent: a property chain dispatches to the root's target, and + // the property path itself is captured separately by jsrpc.method. + kj::LiteralStringConst getRpcTargetKind() override { + return parent->getRpcTargetKind(); + } // Call the property as a method. jsg::Ref call(const v8::FunctionCallbackInfo& args); @@ -359,11 +397,13 @@ class JsRpcStub: public JsRpcClientProvider { JsRpcStub(IoOwn rpcChannel): rpcChannel(kj::mv(rpcChannel)) {} JsRpcStub(IoOwn capnpClient, RpcStubDisposalGroup& disposalGroup, - jsg::ExternalMemoryAdjustment externalMemoryAdjustment); + jsg::ExternalMemoryAdjustment externalMemoryAdjustment, + kj::Maybe originatingCall); JsRpcStub(IoOwn capnpClient, IoOwn rpcChannel, RpcStubDisposalGroup& disposalGroup, - jsg::ExternalMemoryAdjustment externalMemoryAdjustment); + jsg::ExternalMemoryAdjustment externalMemoryAdjustment, + kj::Maybe originatingCall); explicit JsRpcStub(uint channelNumber): channelNumber(channelNumber) {} ~JsRpcStub() noexcept(false); @@ -372,8 +412,11 @@ class JsRpcStub: public JsRpcClientProvider { // If the stub is backed by a persistable RpcChannel, return it. kj::Maybe> getRpcChannel(IoContext& ioctx); - rpc::JsRpcTarget::Client getClientForOneCall( - jsg::Lock& js, kj::Vector& path) override; + ClientForOneCall getClientForOneCall(jsg::Lock& js, kj::Vector& path) override; + + kj::LiteralStringConst getRpcTargetKind() override { + return "stub"_kjc; + } jsg::Ref dup(jsg::Lock& js); void dispose(); @@ -427,6 +470,19 @@ class JsRpcStub: public JsRpcClientProvider { kj::ListLink disposalGroupLink; kj::Maybe externalMemoryAdjustment; + // The jsRpcCall of the call that returned this stub (set on stubs received via + // deserialization), used to parent follow-up calls on the stub under it. Only set when the + // originating call was traced; kj::none otherwise, so untraced stubs retain no span state. When + // set, it holds a refcount on the originating SpanObserver through the request's IoContext. + // + // Note: the originating jsRpcCall span is normally already closed by the time we open + // a child here (the originating SpanBuilder is destroyed when the awaitIo callback that + // produced this stub returns). That's OK: SpanParent holds a refcount on the underlying + // refcounted SpanObserver, so the parent identity remains valid for newChild() even after + // the parent has been reported closed. Observers must tolerate children opening after + // their parent's onClose(). + kj::Maybe> originatingCall; + friend class RpcStubDisposalGroup; }; diff --git a/src/workerd/io/incoming-request-test.c++ b/src/workerd/io/incoming-request-test.c++ index 5974ee7bcb9..02f73fb2571 100644 --- a/src/workerd/io/incoming-request-test.c++ +++ b/src/workerd/io/incoming-request-test.c++ @@ -23,6 +23,66 @@ class ErrorHandlerImpl: public kj::TaskSet::ErrorHandler { } }; +class SpanWarningCapture final: public kj::ExceptionCallback { + public: + void logMessage(kj::LogSeverity severity, + const char* file, + int line, + int contextDepth, + kj::String&& text) override { + if (severity == kj::LogSeverity::WARNING && + text.contains("reported span without current request"_kj)) { + sawWarning = true; + return; + } + kj::ExceptionCallback::logMessage(severity, file, line, contextDepth, kj::mv(text)); + } + + bool sawWarning = false; +}; + +class RecordingTracer final: public BaseTracer { + public: + void setContext(IoContext& context) { + weakIoContext = context.getWeakRef(); + } + + kj::Date getSpanEndTime() { + return KJ_ASSERT_NONNULL(spanEndTime); + } + + void addLog(const tracing::InvocationSpanContext&, + kj::Date, + LogLevel, + kj::String, + tracing::LogErrorInfo) override {} + void addSpanOpen(tracing::SpanId, tracing::SpanId, kj::ConstString, kj::Date) override {} + void addSpanClose(tracing::SpanEndData&& span, kj::Maybe maybeStartTime) override { + adjustSpanTime(span, maybeStartTime); + spanEndTime = span.endTime; + } + void addException(const tracing::InvocationSpanContext&, + kj::Date, + kj::String, + kj::String, + kj::Maybe) override {} + void addDiagnosticChannelEvent( + const tracing::InvocationSpanContext&, kj::Date, kj::String, kj::Array) override {} + void setEventInfo(IoContext::IncomingRequest& request, tracing::EventInfo&&) override { + setContext(request.getContext()); + } + void setReturn(kj::Maybe, kj::Maybe) override {} + void setOutcome(EventOutcome, kj::Duration, kj::Duration) override {} + void recordTimestamp(kj::Date timestamp) override { + completeTime = timestamp; + } + void setJsRpcInfo( + const tracing::InvocationSpanContext&, kj::Date, const kj::ConstString&) override {} + + private: + kj::Maybe spanEndTime; +}; + class FrozenTimerChannel final: public TimerChannel { public: void syncTime() override { @@ -87,6 +147,83 @@ KJ_TEST("trace onset synchronizes an idle actor's clock before reading it") { fixture.drainAndDestroy(kj::mv(request)); } +KJ_TEST("span close after request teardown uses the recorded completion time") { + FrozenTimerChannel timer; + kj::Function(TimerChannel&)> makeChannelFactory = + [&timer](TimerChannel&) -> kj::Rc { + return kj::rc(timer); + }; + TestFixture fixture({ + .actorId = Worker::Actor::Id(kj::str("trace-timing-test")), + .useRealTimers = false, + .ioChannelFactory = kj::mv(makeChannelFactory), + }); + + auto tracer = kj::refcounted(); + auto tracerRef = kj::addRef(*tracer); + + auto context = fixture.newIoContext(); + auto request = fixture.newUndeliveredIncomingRequest(*context, kj::mv(tracer)); + timer.advance(1 * kj::SECONDS); + tracerRef->setEventInfo(*request, tracing::CustomEventInfo()); + request->delivered(); + request = nullptr; + + SpanWarningCapture warningCapture; + tracerRef->addSpanClose( + tracing::SpanEndData(tracing::SpanId(1), kj::UNIX_EPOCH + 10 * kj::SECONDS), + kj::UNIX_EPOCH + 500 * kj::MILLISECONDS); + KJ_EXPECT(tracerRef->getSpanEndTime() == timer.getWallTime()); + KJ_EXPECT(!warningCapture.sawWarning); +} + +KJ_TEST("current request provides span close time after an earlier request completed") { + FrozenTimerChannel timer; + kj::Function(TimerChannel&)> makeChannelFactory = + [&timer](TimerChannel&) -> kj::Rc { + return kj::rc(timer); + }; + TestFixture fixture({ + .actorId = Worker::Actor::Id(kj::str("trace-timing-test")), + .useRealTimers = false, + .ioChannelFactory = kj::mv(makeChannelFactory), + }); + + auto tracer = kj::refcounted(); + auto tracerRef = kj::addRef(*tracer); + auto context = fixture.newIoContext(); + auto request = fixture.newUndeliveredIncomingRequest(*context, kj::mv(tracer)); + timer.advance(1 * kj::SECONDS); + tracerRef->setEventInfo(*request, tracing::CustomEventInfo()); + request->delivered(); + request = nullptr; + + timer.advance(1 * kj::SECONDS); + auto replacementRequest = fixture.newIncomingRequest(*context); + SpanWarningCapture warningCapture; + tracerRef->addSpanClose( + tracing::SpanEndData(tracing::SpanId(1), kj::UNIX_EPOCH + 10 * kj::SECONDS), + kj::UNIX_EPOCH + 1500 * kj::MILLISECONDS); + KJ_EXPECT(tracerRef->getSpanEndTime() == timer.getWallTime()); + KJ_EXPECT(!warningCapture.sawWarning); + + fixture.drainAndDestroy(kj::mv(replacementRequest)); +} + +KJ_TEST("span close without a current request or completion time still warns") { + TestFixture fixture({.actorId = Worker::Actor::Id(kj::str("trace-timing-test"))}); + auto context = fixture.newIoContext(); + auto tracer = kj::refcounted(); + tracer->setContext(*context); + + SpanWarningCapture warningCapture; + auto startTime = kj::UNIX_EPOCH + 1 * kj::SECONDS; + tracer->addSpanClose( + tracing::SpanEndData(tracing::SpanId(1), kj::UNIX_EPOCH + 2 * kj::SECONDS), startTime); + KJ_EXPECT(tracer->getSpanEndTime() == startTime); + KJ_EXPECT(warningCapture.sawWarning); +} + // Regression test: two IncomingRequests share a single actor IoContext, as happens when a Durable // Object receives overlapping requests. Draining the older, superseded request hits drain()'s // "a newer request has taken over" early return. diff --git a/src/workerd/io/io-context.c++ b/src/workerd/io/io-context.c++ index e0decf407d4..c785b634f4b 100644 --- a/src/workerd/io/io-context.c++ +++ b/src/workerd/io/io-context.c++ @@ -1098,6 +1098,23 @@ kj::Own IoContext::getSubrequestChannel( }); } +kj::Own IoContext::getSubrequestChannel(uint channel, + bool isInHouse, + kj::Maybe cfBlobJson, + TraceContext& traceContext, + SpanParent userSpanParent) { + return getSubrequest( + [&](TraceContext& tracing, IoChannelFactory& channelFactory) { + return getSubrequestChannelImpl( + channel, isInHouse, kj::mv(cfBlobJson), tracing, channelFactory, kj::mv(userSpanParent)); + }, + SubrequestOptions{ + .inHouse = isInHouse, + .wrapMetrics = !isInHouse, + .existingTraceContext = traceContext, + }); +} + kj::Own IoContext::getSubrequestChannelNoChecks(uint channel, bool isInHouse, kj::Maybe cfBlobJson, @@ -1118,11 +1135,16 @@ kj::Own IoContext::getSubrequestChannelImpl(uint channel, bool isInHouse, kj::Maybe cfBlobJson, TraceContext& tracing, - IoChannelFactory& channelFactory) { + IoChannelFactory& channelFactory, + kj::Maybe userSpanParent) { + auto propagatedUserSpanParent = tracing.getUserSpanParent(); + KJ_IF_SOME(parent, userSpanParent) { + propagatedUserSpanParent = kj::mv(parent); + } IoChannelFactory::SubrequestMetadata metadata{ .cfBlobJson = kj::mv(cfBlobJson), .parentSpan = tracing.getInternalSpanParent(), - .userSpanParent = tracing.getUserSpanParent(), + .userSpanParent = kj::mv(propagatedUserSpanParent), .featureFlagsForFl = mapCopyString(worker->getIsolate().getFeatureFlagsForFl()), }; diff --git a/src/workerd/io/io-context.h b/src/workerd/io/io-context.h index d1f41acf866..6c5d6810d51 100644 --- a/src/workerd/io/io-context.h +++ b/src/workerd/io/io-context.h @@ -972,6 +972,12 @@ class IoContext final: public kj::Refcounted, private kj::TaskSet::ErrorHandler kj::Own getSubrequestChannel( uint channel, bool isInHouse, kj::Maybe cfBlobJson, TraceContext& traceContext); + kj::Own getSubrequestChannel(uint channel, + bool isInHouse, + kj::Maybe cfBlobJson, + TraceContext& traceContext, + SpanParent userSpanParent); + // Like getSubrequestChannel() but doesn't enforce limits. Use for trusted paths only. kj::Own getSubrequestChannelNoChecks(uint channel, bool isInHouse, @@ -1184,7 +1190,8 @@ class IoContext final: public kj::Refcounted, private kj::TaskSet::ErrorHandler bool isInHouse, kj::Maybe cfBlobJson, TraceContext& tracing, - IoChannelFactory& channelFactory); + IoChannelFactory& channelFactory, + kj::Maybe userSpanParent = kj::none); friend class IoContext_IncomingRequest; template diff --git a/src/workerd/io/trace.h b/src/workerd/io/trace.h index c0aca4514c9..e33c78f66c9 100644 --- a/src/workerd/io/trace.h +++ b/src/workerd/io/trace.h @@ -1374,6 +1374,35 @@ inline SpanBuilder SpanBuilder::newChild( kj::mv(operationName), startTime); } +class TraceContext; + +// Pair of span parents (internal + user) used to express "open new spans nested +// under these". Doesn't own SpanBuilders, unlike TraceContext. +class TraceContextParent { + public: + TraceContextParent(SpanParent internalSpan, SpanParent userSpan) + : internalSpan(kj::mv(internalSpan)), + userSpan(kj::mv(userSpan)) {} + TraceContextParent(TraceContextParent&& other) = default; + TraceContextParent& operator=(TraceContextParent&& other) = default; + KJ_DISALLOW_COPY(TraceContextParent); + + TraceContextParent addRef() { + return TraceContextParent(internalSpan.addRef(), userSpan.addRef()); + } + + // Useful to skip unnecessary work (e.g. creating child spans) when not observed. + bool isObserved() { + return internalSpan.isObserved() || userSpan.isObserved(); + } + + [[nodiscard]] TraceContext newChild(kj::ConstString operationName); + + private: + SpanParent internalSpan; + SpanParent userSpan; +}; + // TraceContext to keep track of user tracing/existing tracing better class TraceContext { public: @@ -1398,11 +1427,32 @@ class TraceContext { return SpanParent(userSpan); } + TraceContextParent getSpanParents() { + return TraceContextParent(SpanParent(span), SpanParent(userSpan)); + } + + // Like getSpanParents(), but returns kj::none when neither span is observed. Use this when the + // parents may be retained beyond the current call (e.g. stored on a returned stub/promise): + // storing nothing on the untraced path avoids holding span state, while the traced path still + // nests follow-up calls correctly. + kj::Maybe getSpanParentsIfObserved() { + if (!isObserved()) return kj::none; + return getSpanParents(); + } + private: SpanBuilder span; SpanBuilder userSpan; }; +inline TraceContext TraceContextParent::newChild(kj::ConstString operationName) { + // newChild() consumes its operationName argument, so clone it for the internal child and move + // the original into the user child. + auto internalChild = internalSpan.newChild(operationName.clone()); + auto userChild = userSpan.newChild(kj::mv(operationName)); + return TraceContext(kj::mv(internalChild), kj::mv(userChild)); +} + // RAII object that measures the time duration over its lifetime. It tags this duration onto a // given request span using a specified tag name. Ideal for automatically tracking and logging // execution times within a scoped block. diff --git a/src/workerd/io/tracer.c++ b/src/workerd/io/tracer.c++ index e0e1d32a6b5..2db3195e3b2 100644 --- a/src/workerd/io/tracer.c++ +++ b/src/workerd/io/tracer.c++ @@ -479,20 +479,17 @@ void BaseTracer::adjustSpanTime(tracing::SpanEndData& span, kj::Maybe if (context.hasCurrentIncomingRequest()) { span.endTime = context.now(); } else { - // We have an IOContext, but there's no current IncomingRequest. Always log a warning here, - // this should not be happening. Still report completeTime as a useful timestamp if - // available. - bool hasCompleteTime = false; + // Tasks can finish after their IncomingRequest has been destroyed. Use the timestamp + // recorded during request teardown when it is available. if (completeTime != kj::UNIX_EPOCH) { span.endTime = completeTime; - hasCompleteTime = true; } else { span.endTime = startTime; - } - if (isPredictableModeForTest()) { - KJ_FAIL_ASSERT("reported span without current request", hasCompleteTime); - } else { - LOG_WARNING_PERIODICALLY("reported span without current request"); + if (isPredictableModeForTest()) { + KJ_FAIL_ASSERT("reported span without current request or completeTime"); + } else { + LOG_WARNING_PERIODICALLY("reported span without current request or completeTime"); + } } } }); diff --git a/src/workerd/io/worker-interface.capnp b/src/workerd/io/worker-interface.capnp index 171f60a7a03..6cf097fe5b7 100644 --- a/src/workerd/io/worker-interface.capnp +++ b/src/workerd/io/worker-interface.capnp @@ -808,6 +808,14 @@ interface JsRpcTarget extends(JsValue.ExternalPusher) $Cxx.allowCancellation { # ExternalPusher object which will push into the caller's isolate. Use this to push externals # that will be included in the results. } + + callerSpanContext @6 :SpanContext; + # Identity of the caller's per-call `jsRpcCall` span. The callee records this as a link on its + # own per-call span. This is needed because a single session carries many calls (e.g. calls + # pipelined on a returned stub or promise), while the context propagated when the session was + # opened identifies only the first call. + # + # Absent when the caller is not being traced. } struct CallResults {