Implement async trace mechanism

src/rust/AGENTS.md

@@ -21,6 +21,7 @@ _Snapshot — the set drifts as crates come and go; `bazel query //src/rust/...`
 | `encoding/`             | WHATWG legacy text decoders via `encoding_rs`; an opaque streaming `Decoder` emits UTF-16 over the CXX bridge                                                                                                                 |
 | `kj/`                   | Rust bindings for KJ library (`http`, `io`, `own` submodules); `Result<T>` = `Result<T, cxx::KjError>`                                                                                                                        |
 | `worker/`               | Rust counterpart of `workerd::WorkerInterface`: the `worker::Interface` trait (`into_kj` hands an implementation to C++ as `KjOwn<WorkerInterface>`; `not_supported` answers a `CustomEvent` as C++'s `event->notSupported()` does), `PromisedInterface` (an `Interface` whose target is still starting, as C++'s `PromisedWorkerInterface`) plus FFI bindings; multi-bridge crate |
+| `async-trace/`          | Async activity tracking core (`async_hooks`-style): per-isolate `IsolateState`, per-IoContext `Tracker` (turns, scopes, binding adoption, stats), `Sink` trait, NDJSON and recording sinks; opaque `Isolate`/`Tracker`/`Writer` over CXX |
 | `cxx-integration/`      | Tokio runtime init; called from C++ `main()` before anything else                                                                                                                                                             |
 | `cxx-integration-test/` | Non-production crate exercising Rust/C++ integration: callbacks, shared structs, `Result` error mapping                                                                                                                       |
 | `transpiler/`           | TS type stripping via SWC (`ts_strip()`, `StripOnly` mode)                                                                                                                                                                    |

src/rust/async-trace/BUILD.bazel

@@ -0,0 +1,11 @@
+load("//:build/wd_rust_crate.bzl", "wd_rust_crate")
+
+wd_rust_crate(
+    name = "async-trace",
+    cxx_bridge_src = "ffi.rs",
+    visibility = ["//visibility:public"],
+    deps = [
+        "@crates_vendor//:serde",
+        "@crates_vendor//:serde_json",
+    ],
+)

src/rust/async-trace/clock-test.rs

@@ -0,0 +1,24 @@
+// 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
+
+use super::*;
+
+#[test]
+fn monotonic_clock_does_not_go_backwards() {
+    let clock = MonotonicClock;
+    let first = clock.now();
+    let second = clock.now();
+    assert!(second >= first);
+}
+
+#[test]
+fn epoch_wall_clock_is_plausible() {
+    // 2020-01-01T00:00:00Z. Anything earlier means the conversion is broken.
+    assert!(epoch_unix_ms() > 1_577_836_800_000);
+}
+
+#[test]
+fn epoch_is_fixed() {
+    assert_eq!(epoch_unix_ms(), epoch_unix_ms());
+}

src/rust/async-trace/clock.rs

@@ -0,0 +1,59 @@
+// 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
+
+//! Trace timestamps.
+//!
+//! Timestamps are nanoseconds on the monotonic clock since a process-wide epoch, fixed the first
+//! time any trace clock is read. They are precise (not Spectre-coarsened like `IoContext::now()`),
+//! which is acceptable only because they never reach JavaScript.
+
+use std::sync::OnceLock;
+use std::time::Instant;
+use std::time::SystemTime;
+
+/// Nanoseconds since the process trace epoch.
+pub type Nanos = u64;
+
+struct Epoch {
+    instant: Instant,
+    unix_ms: u64,
+}
+
+static EPOCH: OnceLock<Epoch> = OnceLock::new();
+
+fn epoch() -> &'static Epoch {
+    EPOCH.get_or_init(|| Epoch {
+        instant: Instant::now(),
+        unix_ms: SystemTime::now()
+            .duration_since(SystemTime::UNIX_EPOCH)
+            .map_or(0, |d| u64::try_from(d.as_millis()).unwrap_or(u64::MAX)),
+    })
+}
+
+/// Wall-clock time of the trace epoch, in milliseconds since the Unix epoch. For display only.
+#[must_use]
+pub fn epoch_unix_ms() -> u64 {
+    epoch().unix_ms
+}
+
+/// A source of trace timestamps. Production code uses [`MonotonicClock`]; tests substitute a
+/// manual clock.
+pub trait Clock {
+    fn now(&self) -> Nanos;
+}
+
+/// The monotonic clock, relative to the process trace epoch.
+#[derive(Debug, Clone, Copy, Default)]
+pub struct MonotonicClock;
+
+impl Clock for MonotonicClock {
+    fn now(&self) -> Nanos {
+        let elapsed = Instant::now().saturating_duration_since(epoch().instant);
+        u64::try_from(elapsed.as_nanos()).unwrap_or(u64::MAX)
+    }
+}
+
+#[cfg(test)]
+#[path = "clock-test.rs"]
+mod tests;

src/rust/async-trace/ffi-test.rs

@@ -0,0 +1,126 @@
+// 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
+
+use super::*;
+use crate::Event;
+use crate::RecordingSink;
+
+#[test]
+fn trackers_share_the_isolate() {
+    let isolate = new_isolate();
+    let mut a = new_tracker(&isolate, b"a", b"");
+    let mut b = new_tracker(&isolate, b"b", b"");
+    let x = a.create_ffi(3, b"x", 0, 0, 0);
+    let y = b.create_ffi(3, b"y", 0, 0, 0);
+    assert!(y > x);
+}
+
+#[test]
+fn empty_actor_means_none() {
+    let isolate = new_isolate();
+    for (actor, expected) in [(&b""[..], None), (&b"id"[..], Some("id".to_owned()))] {
+        let mut tracker = new_tracker(&isolate, b"w", actor);
+        let sink = RecordingSink::new();
+        tracker.add_sink(Box::new(sink.clone()));
+        assert!(matches!(
+            &sink.events()[0],
+            Event::ContextBegin { actor: a, .. } if *a == expected
+        ));
+    }
+}
+
+#[test]
+fn create_and_settle_translate_their_arguments() {
+    let isolate = new_isolate();
+    let mut tracker = new_tracker(&isolate, b"w", b"");
+    let sink = RecordingSink::new();
+    tracker.add_sink(Box::new(sink.clone()));
+    let _ = sink.take();
+
+    tracker.begin_stack();
+    tracker.push_frame_ffi(b"f", b"s.js", 1, 2, 3);
+    let stack = tracker.end_stack_ffi();
+    assert_ne!(stack, 0);
+    let id = tracker.create_ffi(5, b"kv_get", 0, stack, 0);
+    let unstacked = tracker.create_ffi(200, b"other", id, 0, 0);
+    tracker.settle_ffi(id, 2);
+
+    let events = sink.take();
+    assert!(matches!(&events[0], Event::Stack { id: s, .. } if *s == stack));
+    assert!(matches!(
+        &events[1],
+        Event::Init { kind: Kind::Operation, stack: Some(s), .. } if *s == stack
+    ));
+    assert!(matches!(
+        &events[2],
+        Event::Init { id: i, kind: Kind::Other, trigger, stack: None, .. }
+            if *i == unstacked && *trigger == id
+    ));
+    assert!(matches!(
+        &events[3],
+        Event::Settle {
+            outcome: Outcome::Canceled,
+            ..
+        }
+    ));
+}
+
+#[test]
+fn empty_stack_is_zero() {
+    let isolate = new_isolate();
+    let mut tracker = new_tracker(&isolate, b"w", b"");
+    tracker.begin_stack();
+    assert_eq!(tracker.end_stack_ffi(), 0);
+}
+
+#[test]
+fn ndjson_writer_round_trip() {
+    let dir = std::env::var("TEST_TMPDIR").map_or_else(|_| std::env::temp_dir(), Into::into);
+    let path = dir.join("async-trace-ffi-test.ndjson");
+    let path = path.to_str().unwrap();
+    let writer = open_ndjson_writer(path.as_bytes(), b"0.0.0-test").unwrap();
+    let isolate = new_isolate();
+    let mut tracker = new_tracker(&isolate, b"w", b"");
+    tracker.add_ndjson_sink(&writer);
+    tracker.create_ffi(0, b"fetch", 0, 0, 0);
+    tracker.close();
+    assert!(!writer.failed());
+
+    let text = std::fs::read_to_string(path).unwrap();
+    let lines: Vec<serde_json::Value> = text
+        .lines()
+        .map(|line| serde_json::from_str(line).unwrap())
+        .collect();
+    assert_eq!(lines[0]["producer"], "workerd");
+    assert_eq!(lines[0]["version"], "0.0.0-test");
+    assert_eq!(lines[2]["kind"], "request");
+    assert_eq!(lines.last().unwrap()["e"], "ctx_end");
+}
+
+#[test]
+fn open_ndjson_writer_reports_errors() {
+    assert!(open_ndjson_writer(b"/nonexistent-dir/x/y.ndjson", b"1").is_err());
+}
+
+#[test]
+fn invalid_utf8_is_replaced_not_rejected() {
+    let isolate = new_isolate();
+    let mut tracker = new_tracker(&isolate, b"w\xff", b"");
+    let sink = RecordingSink::new();
+    tracker.add_sink(Box::new(sink.clone()));
+    let id = tracker.create_ffi(5, b"kv\xfe", 0, 0, 0);
+    tracker.annotate_ffi(id, b"k\xff", b"v\xff");
+    let events = sink.take();
+    assert!(matches!(&events[0], Event::ContextBegin { worker, .. } if worker == "w\u{fffd}"));
+    assert!(matches!(&events[1], Event::Init { name, .. } if name == "kv\u{fffd}"));
+    assert!(matches!(
+        &events[2],
+        Event::Annotate { key, value, .. } if key == "k\u{fffd}" && value == "v\u{fffd}"
+    ));
+}
+
+#[test]
+fn writer_path_must_be_utf8() {
+    assert!(open_ndjson_writer(b"/tmp/\xff.ndjson", b"1").is_err());
+}