fix(native-spans): Bun-compatible transport head + UDS client stats

crates/capabilities/src/http.rs

@@ -193,8 +193,10 @@ fn hex_decode(s: &str) -> Option<Vec<u8>> {
 /// Serialize the full HTTP/1.1 request head (request line + Host + Content-Length
 /// + user headers + terminating CRLF) into a contiguous byte buffer.
 ///
-/// The buffer is handed to JS by pointer; JS assigns it to
-/// `req._header`, bypassing Node's `_storeHeader` serialization.
+/// The buffer is handed to JS by pointer; JS parses it (`parseRequestHead` in
+/// http_transport.js) into method/path/headers for `http.request(...)`. (It used
+/// to be assigned to the Node internal `req._header`, but Bun's node:http
+/// ignores that, so the head is parsed into request options instead.)
 ///
 /// `is_socket` requests (unix socket / named pipe) omit the `:port` suffix on
 /// the Host header — there is no TCP port for a socket transport.

crates/capabilities/src/http_transport.js

@@ -107,6 +107,28 @@ function applyEntityHeaders (headView, entity = getEntityHeaders()) {
   return Buffer.from(`${kept.join('\r\n')}\r\n\r\n`, 'latin1')
 }
 
+// Parse the Rust-rendered (+ entity-merged) HTTP/1.1 request head into Node
+// request options `{ method, path, headers }`. We pass these to
+// `http.request(...)` rather than injecting the raw head via the Node-internal
+// `req._header`: that internal is undocumented and Bun's `node:http` ignores it,
+// so under Bun the request went out as `POST /` with no headers and the agent
+// dropped it. Header names/values are ASCII (latin1 round-trips losslessly).
+function parseRequestHead (headBuf) {
+  const head = Buffer.from(headBuf).toString('latin1')
+  const term = head.indexOf('\r\n\r\n')
+  const lines = (term === -1 ? head : head.slice(0, term)).split('\r\n')
+  // Request line: `METHOD request-target HTTP/1.1` (no spaces in the target).
+  const [method, path] = lines[0].split(' ')
+  const headers = {}
+  for (let i = 1; i < lines.length; i++) {
+    const colon = lines[i].indexOf(':')
+    if (colon === -1) continue
+    const name = lines[i].slice(0, colon).trim()
+    if (name) headers[name] = lines[i].slice(colon + 1).trim()
+  }
+  return { method, path, headers }
+}
+
 // A retried write can race a wasm-memory detach; treat that as a transient error.
 function isDetachedBufferError (err) {
   return err instanceof TypeError && /detached/i.test(err.message)
@@ -173,11 +195,15 @@ module.exports.httpRequest = function (host, port, isHttps, socketPath, head_ptr
         const headView = new Uint8Array(wasm_memory.buffer, head_ptr, head_len)
         const bodyView = new Uint8Array(wasm_memory.buffer, body_ptr, body_len)
 
-        // host/port (or socketPath) drive connection selection; method/path/
-        // headers are placeholders because we replace the rendered head below.
+        // The Rust side already rendered the full HTTP/1.1 request head (real
+        // method, `/v0.4/traces` path, Content-Type/Length, datadog-meta-*);
+        // applyEntityHeaders merges in the detected entity headers. Parse it into
+        // request options so the connection uses the correct method/path/headers
+        // on both Node and Bun (host/port or socketPath drive the connection).
+        const { method, path, headers } = parseRequestHead(applyEntityHeaders(headView))
         const requestOptions = useSocket
-          ? { socketPath, method: 'POST', path: '/' }
-          : { host, port, method: 'POST', path: '/' }
+          ? { socketPath, method, path, headers }
+          : { host, port, method, path, headers }
         const req = transport.request(requestOptions, (res) => {
           const chunks = []
           res.on('data', chunk => chunks.push(chunk))
@@ -207,12 +233,10 @@ module.exports.httpRequest = function (host, port, isHttps, socketPath, head_ptr
         })
         req.on('error', reject)
 
-        // Bypass Node's headers: the Rust side has already produced the full
-        // request head in HTTP/1.1 wire format. Setting _header before write()
-        // makes write/end skip _implicitHeader and _send prepends our bytes.
-
+        // The request head (method/path/headers) was supplied via requestOptions
+        // above; just write the body. (No `req._header` injection — that Node
+        // internal is not honored by Bun.)
         try {
-          req._header = applyEntityHeaders(headView)
           req.write(bodyView)
           req.end()
         } catch (error) {

crates/pipeline/Cargo.toml

@@ -15,6 +15,7 @@ serde = { version = "1.0", features = ["derive"] }
 serde_json = "1"
 libdatadog-nodejs-capabilities = { path = "../capabilities" }
 libdd-capabilities = { git = "https://github.com/DataDog/libdatadog.git", tag = "v37.0.0" }
+libdd-common = { git = "https://github.com/DataDog/libdatadog.git", tag = "v37.0.0", default-features = false }
 libdd-data-pipeline = { git = "https://github.com/DataDog/libdatadog.git", tag = "v37.0.0", default-features = false }
 libdd-trace-utils = { git = "https://github.com/DataDog/libdatadog.git", tag = "v37.0.0", default-features = false, features = ["change-buffer"] }
 libdd-trace-stats = { git = "https://github.com/DataDog/libdatadog.git", tag = "v37.0.0", default-features = false }

crates/pipeline/src/stats.rs

@@ -18,6 +18,7 @@ fn now() -> SystemTime {
 
 use bytes::Bytes;
 use libdd_capabilities::http::HttpClientCapability;
+use libdd_common::parse_uri;
 use libdd_trace_protobuf::pb;
 use libdd_trace_stats::span_concentrator::SpanConcentrator;
 use libdatadog_nodejs_capabilities::WasmHttpClient;
@@ -96,19 +97,26 @@ impl StatsCollector {
         let body = rmp_serde::encode::to_vec_named(&payload)
             .map_err(|e| format!("stats msgpack encode error: {e}"))?;
 
-        // `agent_url` typically ends in `/` (Node's `URL.toString()` normalizes a
-        // bare authority that way), so a naive concat yields `//v0.6/stats`. The
-        // agent records the raw request path, so the double slash makes the
-        // test-agent (and stats tooling) miss the request entirely. Trim the
-        // trailing slash so the path is exactly `/v0.6/stats`.
-        let stats_url = format!(
-            "{}{}",
-            self.agent_url.trim_end_matches('/'),
-            STATS_ENDPOINT_PATH
-        );
-        let uri: http::Uri = stats_url
-            .parse()
-            .map_err(|e| format!("invalid stats URL: {e}"))?;
+        // Build the base agent URI exactly like the trace exporter does, via
+        // libdatadog's `parse_uri`. For a `unix://` / `windows:` agent URL that
+        // hex-encodes the socket path into the URI *authority* (there is no
+        // standard URL form for socket paths), which the WASM HTTP client's
+        // `decode_socket_path` reverses to route over the socket. A raw parse
+        // instead leaves the socket path in the URI *path* with an empty/invalid
+        // authority, so the stats request never reaches the socket — client stats
+        // silently never arrive over UDS (dd-trace-js #9139, uds-express4).
+        let base = parse_uri(&self.agent_url).map_err(|e| format!("invalid agent URL: {e}"))?;
+        // Append `/v0.6/stats` to the base path while preserving the (hex)
+        // authority, mirroring libdd-data-pipeline's `add_path`. For `unix://`
+        // the base path is "/" and the authority holds the hex socket path; for
+        // TCP it's `http://host:port/`. Trim a trailing slash so the path is
+        // exactly `/v0.6/stats` (a double slash makes the agent miss the request).
+        let base_path = base.path().strip_suffix('/').unwrap_or_else(|| base.path());
+        let new_path_and_query = format!("{base_path}{STATS_ENDPOINT_PATH}");
+        let mut parts = base.into_parts();
+        parts.path_and_query =
+            Some(new_path_and_query.parse().map_err(|e| format!("invalid stats path: {e}"))?);
+        let uri = http::Uri::from_parts(parts).map_err(|e| format!("invalid stats URL: {e}"))?;
 
         let req = http::Request::builder()
             .method(http::Method::PUT)

test/pipeline.js

@@ -1125,6 +1125,63 @@ describe('pipeline', { skip }, () => {
       const ns = new NativeSpansInterface({ statsEnabled: false })
       assert.strictEqual(await ns.state.flushStats(true), false)
     })
+
+    it('flushes stats to /v0.6/stats over a Unix domain socket', { skip: process.platform === 'win32' }, async () => {
+      // A `unix://` agent URL must route /v0.6/stats over the socket, like
+      // traces do. parse_uri hex-encodes the socket path into the URI authority
+      // (which the transport's decode_socket_path reverses); a raw parse would
+      // leave the path in the URI path and never reach the socket.
+      const http = require('node:http')
+      const os = require('node:os')
+      const fs = require('node:fs')
+      const nodePath = require('node:path')
+      // Keep the path short — AF_UNIX paths are capped (~104 bytes on macOS).
+      const sockPath = nodePath.join(os.tmpdir(), `dd-st-${process.pid}.sock`)
+      try {
+        fs.unlinkSync(sockPath)
+      } catch {
+        // not present
+      }
+      const seen = []
+      const server = http.createServer((req, res) => {
+        const chunks = []
+        req.on('data', c => chunks.push(c))
+        req.on('end', () => {
+          seen.push({ method: req.method, url: req.url, len: Buffer.concat(chunks).length })
+          res.writeHead(200, { 'content-type': 'application/json' })
+          res.end('{}')
+        })
+      })
+      await new Promise((resolve, reject) => {
+        server.once('error', reject)
+        server.listen(sockPath, resolve)
+      })
+
+      const ns = new NativeSpansInterface({ agentUrl: `unix://${sockPath}`, statsEnabled: true })
+      const span = ns.createSpan()
+      span.name = 'stats-span'
+      span.service = 'stats-svc'
+      span.resource = '/stats'
+      span.type = 'web'
+      span.duration = 5_000_000n
+
+      try {
+        await ns.flushSpans(span)
+        const sent = await ns.state.flushStats(true)
+        assert.strictEqual(sent, true, 'flushStats reported a send over the socket')
+        const statsReq = seen.find(r => r.url === '/v0.6/stats')
+        assert.ok(statsReq, 'agent received a /v0.6/stats request over the socket')
+        assert.ok(statsReq.len > 0, 'stats payload is non-empty')
+      } finally {
+        server.closeAllConnections?.()
+        server.close()
+        try {
+          fs.unlinkSync(sockPath)
+        } catch {
+          // already gone
+        }
+      }
+    })
   })
 
   describe('send re-entrancy', () => {