DO NOT LAND -- Bubbleprof visualization demo

samples/web-streams/config.capnp

@@ -15,4 +15,5 @@ const webStreams :Workerd.Worker = (
     (name = "streams-util", esModule = embed "streams-util.js")
   ],
   compatibilityDate = "2025-12-31",
+  compatibilityFlags = [ "streams_no_default_auto_allocate_chunk_size", "experimental" ],
 );

src/workerd/api/basics.c++

@@ -7,6 +7,7 @@
 #include "actor-state.h"
 #include "global-scope.h"
 
+#include <workerd/io/async-trace.h>
 #include <workerd/io/features.h>
 #include <workerd/io/io-context.h>
 
@@ -1062,6 +1063,17 @@ kj::Promise<void> Scheduler::wait(
     }
   }
 
+  auto& ioContext = IoContext::current();
+
+  // Create async trace resource for this scheduler.wait operation
+  AsyncTraceContext::AsyncId asyncTraceId = AsyncTraceContext::INVALID_ID;
+  if (auto* asyncTrace = ioContext.getAsyncTrace(); asyncTrace != nullptr) {
+    asyncTraceId =
+        asyncTrace->createResource(AsyncTraceContext::ResourceType::kTimer, js.v8Isolate);
+    asyncTrace->annotate(asyncTraceId, "delay"_kj, kj::str(delay));
+    asyncTrace->annotate(asyncTraceId, "type"_kj, "scheduler.wait"_kj);
+  }
+
   // TODO(cleanup): Use jsg promise and resolver to avoid an unlock/relock. However, we need
   //   the abort signal to support wrapping jsg promises.
   auto paf = kj::newPromiseAndFulfiller<void>();
@@ -1070,8 +1082,25 @@ kj::Promise<void> Scheduler::wait(
 
   auto& global =
       jsg::extractInternalPointer<ServiceWorkerGlobalScope, true>(context, context->Global());
-  global.setTimeoutInternal([fulfiller = IoContext::current().addObject(kj::mv(paf.fulfiller))](
-                                jsg::Lock& lock) mutable { fulfiller->fulfill(); },
+  global.setTimeoutInternal(
+      [fulfiller = ioContext.addObject(kj::mv(paf.fulfiller)), asyncTraceId](
+          jsg::Lock& lock) mutable {
+    // Enter async trace callback scope
+    if (asyncTraceId != AsyncTraceContext::INVALID_ID) {
+      if (auto* trace = IoContext::current().getAsyncTrace()) {
+        trace->enterCallback(asyncTraceId);
+      }
+    }
+    KJ_DEFER({
+      // Exit async trace callback scope
+      if (asyncTraceId != AsyncTraceContext::INVALID_ID) {
+        if (auto* trace = IoContext::current().getAsyncTrace()) {
+          trace->exitCallback();
+        }
+      }
+    });
+    fulfiller->fulfill();
+  },
       delay);
 
   auto promise = kj::mv(paf.promise);

src/workerd/api/cache.c++

@@ -6,6 +6,7 @@
 
 #include "util.h"
 
+#include <workerd/io/async-trace.h>
 #include <workerd/io/io-context.h>
 #include <workerd/util/own-util.h>
 
@@ -81,6 +82,12 @@ jsg::Promise<jsg::Optional<jsg::Ref<Response>>> Cache::match(jsg::Lock& js,
   auto userSpan = context.makeUserTraceSpan("cache_match"_kjc);
   TraceContext traceContext(kj::mv(traceSpan), kj::mv(userSpan));
 
+  // Create async trace resource for cache.match
+  if (auto* asyncTrace = context.getAsyncTrace(); asyncTrace != nullptr) {
+    asyncTrace->createResource(AsyncTraceContext::ResourceType::kCacheGet, js.v8Isolate);
+    // Annotation will be added inside evalNow where we have the URL
+  }
+
   KJ_IF_SOME(o, options) {
     KJ_IF_SOME(ignoreMethod, o.ignoreMethod) {
       traceContext.userSpan.setTag("cache.request.ignore_method"_kjc, ignoreMethod);
@@ -288,6 +295,13 @@ jsg::Promise<void> Cache::put(jsg::Lock& js,
     auto userSpan = context.makeUserTraceSpan("cache_put"_kjc);
     TraceContext traceContext(kj::mv(traceSpan), kj::mv(userSpan));
 
+    // Create async trace resource for cache.put
+    if (auto* asyncTrace = context.getAsyncTrace(); asyncTrace != nullptr) {
+      auto asyncId =
+          asyncTrace->createResource(AsyncTraceContext::ResourceType::kCachePut, js.v8Isolate);
+      asyncTrace->annotate(asyncId, "url"_kj, jsRequest->getUrl());
+    }
+
     traceContext.userSpan.setTag("cache.request.url"_kjc, jsRequest->getUrl());
     traceContext.userSpan.setTag("cache.request.method"_kjc, kj::str(jsRequest->getMethodEnum()));
     traceContext.userSpan.setTag(
@@ -562,6 +576,13 @@ jsg::Promise<bool> Cache::delete_(jsg::Lock& js,
   auto userSpan = context.makeUserTraceSpan("cache_delete"_kjc);
   TraceContext traceContext(kj::mv(traceSpan), kj::mv(userSpan));
 
+  // Create async trace resource for cache.delete
+  // Note: URL annotation is added inside evalNow where we have the request
+  if (auto* asyncTrace = context.getAsyncTrace(); asyncTrace != nullptr) {
+    asyncTrace->createResource(
+        AsyncTraceContext::ResourceType::kCacheGet, js.v8Isolate);  // Using kCacheGet as stand-in
+  }
+
   KJ_IF_SOME(o, options) {
     KJ_IF_SOME(ignoreMethod, o.ignoreMethod) {
       traceContext.userSpan.setTag("cache.request.ignore_method"_kjc, ignoreMethod);

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

@@ -18,6 +18,7 @@
 #include <workerd/api/system-streams.h>
 #include <workerd/api/trace.h>
 #include <workerd/api/util.h>
+#include <workerd/io/async-trace.h>
 #include <workerd/io/compatibility-date.h>
 #include <workerd/io/features.h>
 #include <workerd/io/io-context.h>
@@ -776,9 +777,32 @@ jsg::JsString ServiceWorkerGlobalScope::atob(jsg::Lock& js, kj::String data) {
 }
 
 void ServiceWorkerGlobalScope::queueMicrotask(jsg::Lock& js, jsg::Function<void()> task) {
+  auto& context = IoContext::current();
+
+  // Create async trace resource for this microtask
+  AsyncTraceContext::AsyncId asyncTraceId = AsyncTraceContext::INVALID_ID;
+  if (auto* asyncTrace = context.getAsyncTrace(); asyncTrace != nullptr) {
+    asyncTraceId =
+        asyncTrace->createResource(AsyncTraceContext::ResourceType::kMicrotask, js.v8Isolate);
+  }
+
   auto fn = js.wrapSimpleFunction(js.v8Context(),
-      JSG_VISITABLE_LAMBDA((this, fn = kj::mv(task)), (fn),
+      JSG_VISITABLE_LAMBDA((this, fn = kj::mv(task), asyncTraceId), (fn),
           (jsg::Lock& js, const v8::FunctionCallbackInfo<v8::Value>& args) {
+            // Enter async trace callback scope for microtask
+            if (asyncTraceId != AsyncTraceContext::INVALID_ID) {
+            if (auto* trace = IoContext::current().getAsyncTrace()) {
+            trace->enterCallback(asyncTraceId);
+            }
+            }
+            KJ_DEFER({
+              // Exit async trace callback scope
+              if (asyncTraceId != AsyncTraceContext::INVALID_ID) {
+              if (auto* trace = IoContext::current().getAsyncTrace()) {
+              trace->exitCallback();
+              }
+              }
+            });
             js.tryCatch([&] {
               // The function won't be called with any arguments, so we can
               // safely ignore anything passed in to args.
@@ -834,13 +858,39 @@ TimeoutId::NumberType ServiceWorkerGlobalScope::setTimeout(jsg::Lock& js,
     jsg::Function<void(jsg::Arguments<jsg::Value>)> function,
     jsg::Optional<double> msDelay,
     jsg::Arguments<jsg::Value> args) {
+  auto& context = IoContext::current();
+
+  // Create async trace resource for this timer
+  AsyncTraceContext::AsyncId asyncTraceId = AsyncTraceContext::INVALID_ID;
+  if (auto* asyncTrace = context.getAsyncTrace(); asyncTrace != nullptr) {
+    asyncTraceId =
+        asyncTrace->createResource(AsyncTraceContext::ResourceType::kTimer, js.v8Isolate);
+    asyncTrace->annotate(asyncTraceId, "delay"_kj, kj::str(msDelay.orDefault(0)));
+    asyncTrace->annotate(asyncTraceId, "type"_kj, "setTimeout"_kj);
+  }
+
   function.setReceiver(js.v8Ref<v8::Value>(js.v8Context()->Global()));
   auto fn = [function = kj::mv(function), args = kj::mv(args),
-                context = jsg::AsyncContextFrame::currentRef(js)](jsg::Lock& js) mutable {
-    jsg::AsyncContextFrame::Scope scope(js, context);
+                asyncContextFrame = jsg::AsyncContextFrame::currentRef(js),
+                asyncTraceId](jsg::Lock& js) mutable {
+    jsg::AsyncContextFrame::Scope scope(js, asyncContextFrame);
+    // Enter async trace callback scope for timer
+    if (asyncTraceId != AsyncTraceContext::INVALID_ID) {
+      if (auto* trace = IoContext::current().getAsyncTrace()) {
+        trace->enterCallback(asyncTraceId);
+      }
+    }
+    KJ_DEFER({
+      // Exit async trace callback scope
+      if (asyncTraceId != AsyncTraceContext::INVALID_ID) {
+        if (auto* trace = IoContext::current().getAsyncTrace()) {
+          trace->exitCallback();
+        }
+      }
+    });
     function(js, kj::mv(args));
   };
-  auto timeoutId = IoContext::current().setTimeoutImpl(timeoutIdGenerator,
+  auto timeoutId = context.setTimeoutImpl(timeoutIdGenerator,
       /* repeat */ false, [function = kj::mv(fn)](jsg::Lock& js) mutable { function(js); },
       msDelay.orDefault(0));
   return timeoutId.toNumber();
@@ -860,15 +910,41 @@ TimeoutId::NumberType ServiceWorkerGlobalScope::setInterval(jsg::Lock& js,
     jsg::Function<void(jsg::Arguments<jsg::Value>)> function,
     jsg::Optional<double> msDelay,
     jsg::Arguments<jsg::Value> args) {
+  auto& context = IoContext::current();
+
+  // Create async trace resource for this timer
+  AsyncTraceContext::AsyncId asyncTraceId = AsyncTraceContext::INVALID_ID;
+  if (auto* asyncTrace = context.getAsyncTrace(); asyncTrace != nullptr) {
+    asyncTraceId =
+        asyncTrace->createResource(AsyncTraceContext::ResourceType::kTimer, js.v8Isolate);
+    asyncTrace->annotate(asyncTraceId, "delay"_kj, kj::str(msDelay.orDefault(0)));
+    asyncTrace->annotate(asyncTraceId, "type"_kj, "setInterval"_kj);
+  }
+
   function.setReceiver(js.v8Ref<v8::Value>(js.v8Context()->Global()));
   auto fn = [function = kj::mv(function), args = kj::mv(args),
-                context = jsg::AsyncContextFrame::currentRef(js)](jsg::Lock& js) mutable {
-    jsg::AsyncContextFrame::Scope scope(js, context);
+                asyncContextFrame = jsg::AsyncContextFrame::currentRef(js),
+                asyncTraceId](jsg::Lock& js) mutable {
+    jsg::AsyncContextFrame::Scope scope(js, asyncContextFrame);
+    // Enter async trace callback scope for timer
+    if (asyncTraceId != AsyncTraceContext::INVALID_ID) {
+      if (auto* trace = IoContext::current().getAsyncTrace()) {
+        trace->enterCallback(asyncTraceId);
+      }
+    }
+    KJ_DEFER({
+      // Exit async trace callback scope
+      if (asyncTraceId != AsyncTraceContext::INVALID_ID) {
+        if (auto* trace = IoContext::current().getAsyncTrace()) {
+          trace->exitCallback();
+        }
+      }
+    });
     // Because the fn is called multiple times, we will clone the args on each call.
     auto argv = KJ_MAP(i, args) { return i.addRef(js); };
     function(js, jsg::Arguments(kj::mv(argv)));
   };
-  auto timeoutId = IoContext::current().setTimeoutImpl(timeoutIdGenerator,
+  auto timeoutId = context.setTimeoutImpl(timeoutIdGenerator,
       /* repeat */ true, [function = kj::mv(fn)](jsg::Lock& js) mutable { function(js); },
       msDelay.orDefault(0));
   return timeoutId.toNumber();
@@ -1035,9 +1111,34 @@ jsg::Ref<Immediate> ServiceWorkerGlobalScope::setImmediate(jsg::Lock& js,
   // would require a compat flag... but that's OK for now?
 
   auto& context = IoContext::current();
+
+  // Create async trace resource for this immediate
+  AsyncTraceContext::AsyncId asyncTraceId = AsyncTraceContext::INVALID_ID;
+  if (auto* asyncTrace = context.getAsyncTrace(); asyncTrace != nullptr) {
+    asyncTraceId =
+        asyncTrace->createResource(AsyncTraceContext::ResourceType::kTimer, js.v8Isolate);
+    asyncTrace->annotate(asyncTraceId, "delay"_kj, "0"_kj);
+    asyncTrace->annotate(asyncTraceId, "type"_kj, "setImmediate"_kj);
+  }
+
   auto fn = [function = kj::mv(function), args = kj::mv(args),
-                context = jsg::AsyncContextFrame::currentRef(js)](jsg::Lock& js) mutable {
-    jsg::AsyncContextFrame::Scope scope(js, context);
+                asyncContextFrame = jsg::AsyncContextFrame::currentRef(js),
+                asyncTraceId](jsg::Lock& js) mutable {
+    jsg::AsyncContextFrame::Scope scope(js, asyncContextFrame);
+    // Enter async trace callback scope for immediate
+    if (asyncTraceId != AsyncTraceContext::INVALID_ID) {
+      if (auto* trace = IoContext::current().getAsyncTrace()) {
+        trace->enterCallback(asyncTraceId);
+      }
+    }
+    KJ_DEFER({
+      // Exit async trace callback scope
+      if (asyncTraceId != AsyncTraceContext::INVALID_ID) {
+        if (auto* trace = IoContext::current().getAsyncTrace()) {
+          trace->exitCallback();
+        }
+      }
+    });
     function(js, kj::mv(args));
   };
   auto timeoutId = context.setTimeoutImpl(timeoutIdGenerator,

src/workerd/api/http.c++

@@ -14,6 +14,7 @@
 #include "worker-rpc.h"
 #include "workerd/jsg/jsvalue.h"
 
+#include <workerd/io/async-trace.h>
 #include <workerd/io/features.h>
 #include <workerd/io/io-context.h>
 #include <workerd/jsg/ser.h>
@@ -1506,6 +1507,16 @@ jsg::Promise<jsg::Ref<Response>> fetchImplNoOutputLock(jsg::Lock& js,
 
   auto& ioContext = IoContext::current();
 
+  // Create async trace resource for this fetch operation
+  AsyncTraceContext::AsyncId asyncTraceId = AsyncTraceContext::INVALID_ID;
+  if (auto* asyncTrace = ioContext.getAsyncTrace(); asyncTrace != nullptr) {
+    asyncTraceId = asyncTrace->createResource(
+        AsyncTraceContext::ResourceType::kFetch, js.v8Isolate);
+    asyncTrace->annotate(asyncTraceId, "url"_kj, jsRequest->getUrl());
+    asyncTrace->annotate(asyncTraceId, "method"_kj,
+        kj::str(jsRequest->getMethodEnum()));
+  }
+
   auto signal = jsRequest->getSignal();
   KJ_IF_SOME(s, signal) {
     // If the AbortSignal has already been triggered, then we need to stop here.

src/workerd/api/sockets.c++

@@ -7,6 +7,8 @@
 #include "streams/standard.h"
 #include "system-streams.h"
 
+#include <workerd/io/async-trace.h>
+#include <workerd/io/io-context.h>
 #include <workerd/io/worker-interface.h>
 #include <workerd/jsg/exception.h>
 #include <workerd/jsg/url.h>
@@ -255,6 +257,25 @@ jsg::Ref<Socket> connectImplNoOutputLock(jsg::Lock& js,
   }
   kj::Own<kj::TlsStarterCallback> tlsStarter = kj::heap<kj::TlsStarterCallback>();
   httpConnectSettings.tlsStarter = tlsStarter;
+
+  // Create async trace resource for socket connect
+  if (auto* asyncTrace = ioContext.getAsyncTrace(); asyncTrace != nullptr) {
+    auto asyncId =
+        asyncTrace->createResource(AsyncTraceContext::ResourceType::kSocketConnect, js.v8Isolate);
+    asyncTrace->annotate(asyncId, "address"_kj, addressStr);
+    switch (secureTransport) {
+      case SecureTransportKind::OFF:
+        asyncTrace->annotate(asyncId, "secureTransport"_kj, "off"_kj);
+        break;
+      case SecureTransportKind::ON:
+        asyncTrace->annotate(asyncId, "secureTransport"_kj, "on"_kj);
+        break;
+      case SecureTransportKind::STARTTLS:
+        asyncTrace->annotate(asyncId, "secureTransport"_kj, "starttls"_kj);
+        break;
+    }
+  }
+
   auto request = httpClient->connect(addressStr, *headers, httpConnectSettings);
   request.connection = request.connection.attach(kj::mv(httpClient));
 
@@ -283,6 +304,15 @@ jsg::Promise<void> Socket::close(jsg::Lock& js) {
   }
 
   isClosing = true;
+
+  // Create async trace resource for socket close
+  auto& context = IoContext::current();
+  if (auto* asyncTrace = context.getAsyncTrace(); asyncTrace != nullptr) {
+    auto asyncId =
+        asyncTrace->createResource(AsyncTraceContext::ResourceType::kSocketClose, js.v8Isolate);
+    asyncTrace->annotate(asyncId, "address"_kj, remoteAddress);
+  }
+
   writable->getController().setPendingClosure();
   readable->getController().setPendingClosure();
 
@@ -339,6 +369,13 @@ jsg::Ref<Socket> Socket::startTls(jsg::Lock& js, jsg::Optional<TlsOptions> tlsOp
   //
   // Detach the AsyncIoStream from the Writable/Readable streams and make them unusable.
   auto& context = IoContext::current();
+
+  // Create async trace resource for startTls
+  if (auto* asyncTrace = context.getAsyncTrace(); asyncTrace != nullptr) {
+    auto asyncId =
+        asyncTrace->createResource(AsyncTraceContext::ResourceType::kSocketStartTls, js.v8Isolate);
+    asyncTrace->annotate(asyncId, "address"_kj, remoteAddress);
+  }
   auto openedPrPair = js.newPromiseAndResolver<SocketInfo>();
   auto secureStreamPromise = context.awaitJs(js,
       writable->flush(js).then(js,