diff --git a/Cargo.lock b/Cargo.lock index b69d0b47..f2fc9e94 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -897,7 +897,7 @@ dependencies = [ "futures-core", "http", "js-sys", - "libdd-capabilities 2.1.0", + "libdd-capabilities 3.0.0", "wasm-bindgen", "wasm-bindgen-futures", "wasm-bindgen-test", @@ -916,11 +916,13 @@ dependencies = [ [[package]] name = "libdd-capabilities" -version = "2.1.0" -source = "git+https://github.com/DataDog/libdatadog.git?rev=3081603d3c74f209be4e3be951f78a1a7469397f#3081603d3c74f209be4e3be951f78a1a7469397f" +version = "3.0.0" +source = "git+https://github.com/DataDog/libdatadog.git?rev=4e4cf372d4e461d43d9061324ecf60946d5dae53#4e4cf372d4e461d43d9061324ecf60946d5dae53" dependencies = [ "anyhow", "bytes", + "futures-channel", + "futures-util", "http", "thiserror", ] @@ -940,15 +942,15 @@ dependencies = [ [[package]] name = "libdd-capabilities-impl" -version = "3.0.0" -source = "git+https://github.com/DataDog/libdatadog.git?rev=3081603d3c74f209be4e3be951f78a1a7469397f#3081603d3c74f209be4e3be951f78a1a7469397f" +version = "4.0.0" +source = "git+https://github.com/DataDog/libdatadog.git?rev=4e4cf372d4e461d43d9061324ecf60946d5dae53#4e4cf372d4e461d43d9061324ecf60946d5dae53" dependencies = [ "anyhow", "bytes", "http", "http-body-util", - "libdd-capabilities 2.1.0", - "libdd-common 5.1.0", + "libdd-capabilities 3.0.0", + "libdd-common 5.2.0", "tokio", ] @@ -989,8 +991,8 @@ dependencies = [ [[package]] name = "libdd-common" -version = "5.1.0" -source = "git+https://github.com/DataDog/libdatadog.git?rev=3081603d3c74f209be4e3be951f78a1a7469397f#3081603d3c74f209be4e3be951f78a1a7469397f" +version = "5.2.0" +source = "git+https://github.com/DataDog/libdatadog.git?rev=4e4cf372d4e461d43d9061324ecf60946d5dae53#4e4cf372d4e461d43d9061324ecf60946d5dae53" dependencies = [ "anyhow", "bytes", @@ -1053,7 +1055,7 @@ dependencies = [ [[package]] name = "libdd-data-pipeline" version = "7.0.0" -source = "git+https://github.com/DataDog/libdatadog.git?rev=3081603d3c74f209be4e3be951f78a1a7469397f#3081603d3c74f209be4e3be951f78a1a7469397f" +source = "git+https://github.com/DataDog/libdatadog.git?rev=4e4cf372d4e461d43d9061324ecf60946d5dae53#4e4cf372d4e461d43d9061324ecf60946d5dae53" dependencies = [ "anyhow", "arc-swap", @@ -1064,16 +1066,17 @@ dependencies = [ "getrandom 0.2.17", "http", "http-body-util", - "libdd-capabilities 2.1.0", - "libdd-capabilities-impl 3.0.0", - "libdd-common 5.1.0", - "libdd-ddsketch 1.1.0", + "libdd-capabilities 3.0.0", + "libdd-capabilities-impl 4.0.0", + "libdd-common 5.2.0", + "libdd-ddsketch 1.1.0 (git+https://github.com/DataDog/libdatadog.git?rev=4e4cf372d4e461d43d9061324ecf60946d5dae53)", "libdd-dogstatsd-client", "libdd-shared-runtime 2.0.0", "libdd-telemetry 6.0.0", "libdd-tinybytes", "libdd-trace-normalization", - "libdd-trace-protobuf 4.0.0 (git+https://github.com/DataDog/libdatadog.git?rev=3081603d3c74f209be4e3be951f78a1a7469397f)", + "libdd-trace-obfuscation", + "libdd-trace-protobuf 4.0.1", "libdd-trace-stats", "libdd-trace-utils", "rmp-serde", @@ -1103,16 +1106,27 @@ dependencies = [ "prost", ] +[[package]] +name = "libdd-ddsketch" +version = "1.1.0" +source = "git+https://github.com/DataDog/libdatadog.git?rev=4e4cf372d4e461d43d9061324ecf60946d5dae53#4e4cf372d4e461d43d9061324ecf60946d5dae53" +dependencies = [ + "prost", +] + [[package]] name = "libdd-dogstatsd-client" version = "4.0.0" -source = "git+https://github.com/DataDog/libdatadog.git?rev=3081603d3c74f209be4e3be951f78a1a7469397f#3081603d3c74f209be4e3be951f78a1a7469397f" +source = "git+https://github.com/DataDog/libdatadog.git?rev=4e4cf372d4e461d43d9061324ecf60946d5dae53#4e4cf372d4e461d43d9061324ecf60946d5dae53" dependencies = [ "anyhow", + "async-trait", "cadence", "http", - "libdd-common 5.1.0", + "libdd-common 5.2.0", + "libdd-shared-runtime 2.0.0", "serde", + "tokio", "tracing", ] @@ -1123,7 +1137,7 @@ source = "git+https://github.com/DataDog/libdatadog.git?rev=0c6e2a5df2a163d34c4f dependencies = [ "anyhow", "libc", - "libdd-trace-protobuf 4.0.0 (git+https://github.com/DataDog/libdatadog.git?rev=0c6e2a5df2a163d34c4f385353ffc5d7257c72f4)", + "libdd-trace-protobuf 4.0.0", "memfd", "prost", "rand", @@ -1164,14 +1178,14 @@ dependencies = [ [[package]] name = "libdd-shared-runtime" version = "2.0.0" -source = "git+https://github.com/DataDog/libdatadog.git?rev=3081603d3c74f209be4e3be951f78a1a7469397f#3081603d3c74f209be4e3be951f78a1a7469397f" +source = "git+https://github.com/DataDog/libdatadog.git?rev=4e4cf372d4e461d43d9061324ecf60946d5dae53#4e4cf372d4e461d43d9061324ecf60946d5dae53" dependencies = [ "async-trait", "futures", "futures-util", - "libdd-capabilities 2.1.0", - "libdd-capabilities-impl 3.0.0", - "libdd-common 5.1.0", + "libdd-capabilities 3.0.0", + "libdd-capabilities-impl 4.0.0", + "libdd-common 5.2.0", "tokio", "tokio-util", "tracing", @@ -1208,7 +1222,7 @@ dependencies = [ [[package]] name = "libdd-telemetry" version = "6.0.0" -source = "git+https://github.com/DataDog/libdatadog.git?rev=3081603d3c74f209be4e3be951f78a1a7469397f#3081603d3c74f209be4e3be951f78a1a7469397f" +source = "git+https://github.com/DataDog/libdatadog.git?rev=4e4cf372d4e461d43d9061324ecf60946d5dae53#4e4cf372d4e461d43d9061324ecf60946d5dae53" dependencies = [ "anyhow", "async-trait", @@ -1219,12 +1233,14 @@ dependencies = [ "hashbrown 0.15.5", "http", "libc", - "libdd-capabilities 2.1.0", - "libdd-common 5.1.0", - "libdd-ddsketch 1.1.0", + "libdd-capabilities 3.0.0", + "libdd-common 5.2.0", + "libdd-ddsketch 1.1.0 (git+https://github.com/DataDog/libdatadog.git?rev=4e4cf372d4e461d43d9061324ecf60946d5dae53)", "libdd-shared-runtime 2.0.0", "serde", "serde_json", + "strum", + "strum_macros", "sys-info", "tokio", "tokio-util", @@ -1236,32 +1252,33 @@ dependencies = [ [[package]] name = "libdd-tinybytes" -version = "1.1.1" -source = "git+https://github.com/DataDog/libdatadog.git?rev=3081603d3c74f209be4e3be951f78a1a7469397f#3081603d3c74f209be4e3be951f78a1a7469397f" +version = "1.1.2" +source = "git+https://github.com/DataDog/libdatadog.git?rev=4e4cf372d4e461d43d9061324ecf60946d5dae53#4e4cf372d4e461d43d9061324ecf60946d5dae53" dependencies = [ "serde", ] [[package]] name = "libdd-trace-normalization" -version = "3.0.0" -source = "git+https://github.com/DataDog/libdatadog.git?rev=3081603d3c74f209be4e3be951f78a1a7469397f#3081603d3c74f209be4e3be951f78a1a7469397f" +version = "3.0.1" +source = "git+https://github.com/DataDog/libdatadog.git?rev=4e4cf372d4e461d43d9061324ecf60946d5dae53#4e4cf372d4e461d43d9061324ecf60946d5dae53" dependencies = [ "anyhow", - "libdd-trace-protobuf 4.0.0 (git+https://github.com/DataDog/libdatadog.git?rev=3081603d3c74f209be4e3be951f78a1a7469397f)", + "libdd-trace-protobuf 4.0.1", + "web-time", ] [[package]] name = "libdd-trace-obfuscation" version = "5.0.0" -source = "git+https://github.com/DataDog/libdatadog.git?rev=3081603d3c74f209be4e3be951f78a1a7469397f#3081603d3c74f209be4e3be951f78a1a7469397f" +source = "git+https://github.com/DataDog/libdatadog.git?rev=4e4cf372d4e461d43d9061324ecf60946d5dae53#4e4cf372d4e461d43d9061324ecf60946d5dae53" dependencies = [ "anyhow", "fluent-uri", - "libdd-common 5.1.0", - "libdd-trace-protobuf 4.0.0 (git+https://github.com/DataDog/libdatadog.git?rev=3081603d3c74f209be4e3be951f78a1a7469397f)", + "libdd-capabilities 3.0.0", + "libdd-common 5.2.0", + "libdd-trace-protobuf 4.0.1", "libdd-trace-utils", - "log", "percent-encoding", "serde", "serde_json", @@ -1279,8 +1296,8 @@ dependencies = [ [[package]] name = "libdd-trace-protobuf" -version = "4.0.0" -source = "git+https://github.com/DataDog/libdatadog.git?rev=3081603d3c74f209be4e3be951f78a1a7469397f#3081603d3c74f209be4e3be951f78a1a7469397f" +version = "4.0.1" +source = "git+https://github.com/DataDog/libdatadog.git?rev=4e4cf372d4e461d43d9061324ecf60946d5dae53#4e4cf372d4e461d43d9061324ecf60946d5dae53" dependencies = [ "prost", "serde", @@ -1290,7 +1307,7 @@ dependencies = [ [[package]] name = "libdd-trace-stats" version = "6.0.0" -source = "git+https://github.com/DataDog/libdatadog.git?rev=3081603d3c74f209be4e3be951f78a1a7469397f#3081603d3c74f209be4e3be951f78a1a7469397f" +source = "git+https://github.com/DataDog/libdatadog.git?rev=4e4cf372d4e461d43d9061324ecf60946d5dae53#4e4cf372d4e461d43d9061324ecf60946d5dae53" dependencies = [ "anyhow", "arc-swap", @@ -1298,15 +1315,15 @@ dependencies = [ "futures", "hashbrown 0.15.5", "http", - "libdd-capabilities 2.1.0", - "libdd-capabilities-impl 3.0.0", - "libdd-common 5.1.0", - "libdd-ddsketch 1.1.0", + "libdd-capabilities 3.0.0", + "libdd-capabilities-impl 4.0.0", + "libdd-common 5.2.0", + "libdd-ddsketch 1.1.0 (git+https://github.com/DataDog/libdatadog.git?rev=4e4cf372d4e461d43d9061324ecf60946d5dae53)", "libdd-dogstatsd-client", "libdd-shared-runtime 2.0.0", "libdd-telemetry 6.0.0", "libdd-trace-obfuscation", - "libdd-trace-protobuf 4.0.0 (git+https://github.com/DataDog/libdatadog.git?rev=3081603d3c74f209be4e3be951f78a1a7469397f)", + "libdd-trace-protobuf 4.0.1", "libdd-trace-utils", "rmp-serde", "serde", @@ -1318,8 +1335,8 @@ dependencies = [ [[package]] name = "libdd-trace-utils" -version = "9.0.0" -source = "git+https://github.com/DataDog/libdatadog.git?rev=3081603d3c74f209be4e3be951f78a1a7469397f#3081603d3c74f209be4e3be951f78a1a7469397f" +version = "10.1.0" +source = "git+https://github.com/DataDog/libdatadog.git?rev=4e4cf372d4e461d43d9061324ecf60946d5dae53#4e4cf372d4e461d43d9061324ecf60946d5dae53" dependencies = [ "anyhow", "base64", @@ -1332,12 +1349,12 @@ dependencies = [ "http-body-util", "indexmap", "itoa", - "libdd-capabilities 2.1.0", - "libdd-capabilities-impl 3.0.0", - "libdd-common 5.1.0", + "libdd-capabilities 3.0.0", + "libdd-capabilities-impl 4.0.0", + "libdd-common 5.2.0", "libdd-tinybytes", "libdd-trace-normalization", - "libdd-trace-protobuf 4.0.0 (git+https://github.com/DataDog/libdatadog.git?rev=3081603d3c74f209be4e3be951f78a1a7469397f)", + "libdd-trace-protobuf 4.0.1", "prost", "rand", "rmp", @@ -1824,11 +1841,11 @@ dependencies = [ "http", "js-sys", "libdatadog-nodejs-capabilities", - "libdd-capabilities 2.1.0", - "libdd-common 5.1.0", + "libdd-capabilities 3.0.0", + "libdd-common 5.2.0", "libdd-data-pipeline", "libdd-shared-runtime 2.0.0", - "libdd-trace-protobuf 4.0.0 (git+https://github.com/DataDog/libdatadog.git?rev=3081603d3c74f209be4e3be951f78a1a7469397f)", + "libdd-trace-protobuf 4.0.1", "libdd-trace-stats", "libdd-trace-utils", "rmp-serde", @@ -1890,7 +1907,7 @@ version = "0.1.0" dependencies = [ "anyhow", "libdd-library-config", - "libdd-trace-protobuf 4.0.0 (git+https://github.com/DataDog/libdatadog.git?rev=0c6e2a5df2a163d34c4f385353ffc5d7257c72f4)", + "libdd-trace-protobuf 4.0.0", "napi", "napi-derive", ] @@ -2368,7 +2385,7 @@ checksum = "703d5c7ef118737c72f1af64ad2f6f8c5e1921f818cdcb97b8fe6fc69bf66214" name = "sketches" version = "0.1.0" dependencies = [ - "libdd-ddsketch 1.1.0", + "libdd-ddsketch 1.1.0 (git+https://github.com/DataDog/libdatadog.git?rev=3081603d3c74f209be4e3be951f78a1a7469397f)", "wasm-bindgen", ] @@ -2406,6 +2423,25 @@ version = "1.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a2eb9349b6444b326872e140eb1cf5e7c522154d69e7a0ffb0fb81c06b37543f" +[[package]] +name = "strum" +version = "0.26.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8fec0f0aef304996cf250b31b5a10dee7980c85da9d759361292b8bca5a18f06" + +[[package]] +name = "strum_macros" +version = "0.26.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4c6bee85a5a24955dc440386795aa378cd9cf82acd5f764469152d2270e581be" +dependencies = [ + "heck", + "proc-macro2", + "quote", + "rustversion", + "syn", +] + [[package]] name = "subtle" version = "2.6.1" diff --git a/README.md b/README.md index 69caebcd..9324c2d0 100644 --- a/README.md +++ b/README.md @@ -6,3 +6,18 @@ Node.js bindings for [libdatadog](https://github.com/DataDog/libdatadog). This project is currently meant to be used only by [dd-trace-js](https://github.com/DataDog/dd-trace-js) and installing it directly is not supported at the moment. + +## Pipeline + +`pipelineApiVersion` is `1` when the package provides encoded trace and agentless pipeline APIs. Consumers can inspect +this value without loading a native or WASM addon; an absent or older value must use the legacy pipeline. + +`WasmSpanState#sendEncodedTraces(data)` accepts an owned `Uint8Array` containing a v0.4 MessagePack array32 payload. +It forwards the payload directly when no configured feature needs span objects. Native stats decode it for aggregation, +and alternate output formats decode it before transforming and sending it through the configured exporter. +Only one trace send may be active per state; an overlapping call rejects with an `already in flight` error. + +`WasmSpanState#setAgentlessEndpoint(url, apiKey)` selects direct trace intake before the first send. +The binding reads obfuscation settings from Node.js `process.env` through its environment capability; +libdatadog normalizes and obfuscates spans before emitting JSON to the configured intake. Structured +AppSec metadata remains attached to the emitted spans. diff --git a/crates/capabilities/Cargo.toml b/crates/capabilities/Cargo.toml index edb40adc..2c83fc2e 100644 --- a/crates/capabilities/Cargo.toml +++ b/crates/capabilities/Cargo.toml @@ -15,7 +15,7 @@ http = "1" bytes = "1.4" futures-core = "0.3" anyhow = "1" -libdd-capabilities = { git = "https://github.com/DataDog/libdatadog.git", rev = "3081603d3c74f209be4e3be951f78a1a7469397f" } +libdd-capabilities = { git = "https://github.com/DataDog/libdatadog.git", rev = "4e4cf372d4e461d43d9061324ecf60946d5dae53" } [dev-dependencies] wasm-bindgen-test = "0.3" diff --git a/crates/capabilities/src/http.rs b/crates/capabilities/src/http.rs index 4248cb30..2ee1bf78 100644 --- a/crates/capabilities/src/http.rs +++ b/crates/capabilities/src/http.rs @@ -59,6 +59,10 @@ impl HttpClientCapability for WasmHttpClient { Self } + fn new_without_connection_pooling() -> Self { + Self + } + #[allow(clippy::manual_async_fn)] fn request( &self, diff --git a/crates/capabilities/src/lib.rs b/crates/capabilities/src/lib.rs index e76a3812..ed807015 100644 --- a/crates/capabilities/src/lib.rs +++ b/crates/capabilities/src/lib.rs @@ -65,6 +65,10 @@ impl HttpClientCapability for WasmCapabilities { Self::new() } + fn new_without_connection_pooling() -> Self { + Self::new() + } + fn request( &self, req: ::http::Request<::bytes::Bytes>, diff --git a/crates/pipeline/Cargo.toml b/crates/pipeline/Cargo.toml index 8056816e..9b1cdea8 100644 --- a/crates/pipeline/Cargo.toml +++ b/crates/pipeline/Cargo.toml @@ -14,13 +14,13 @@ js-sys = "0.3" serde = { version = "1.0", features = ["derive"] } serde_json = "1" libdatadog-nodejs-capabilities = { path = "../capabilities" } -libdd-capabilities = { git = "https://github.com/DataDog/libdatadog.git", rev = "3081603d3c74f209be4e3be951f78a1a7469397f" } -libdd-common = { git = "https://github.com/DataDog/libdatadog.git", rev = "3081603d3c74f209be4e3be951f78a1a7469397f", default-features = false } -libdd-data-pipeline = { git = "https://github.com/DataDog/libdatadog.git", rev = "3081603d3c74f209be4e3be951f78a1a7469397f", default-features = false, features = ["telemetry"] } -libdd-trace-utils = { git = "https://github.com/DataDog/libdatadog.git", rev = "3081603d3c74f209be4e3be951f78a1a7469397f", default-features = false, features = ["change-buffer"] } -libdd-trace-stats = { git = "https://github.com/DataDog/libdatadog.git", rev = "3081603d3c74f209be4e3be951f78a1a7469397f", default-features = false } -libdd-trace-protobuf = { git = "https://github.com/DataDog/libdatadog.git", rev = "3081603d3c74f209be4e3be951f78a1a7469397f", default-features = false } -libdd-shared-runtime = { git = "https://github.com/DataDog/libdatadog.git", rev = "3081603d3c74f209be4e3be951f78a1a7469397f", default-features = false } +libdd-capabilities = { git = "https://github.com/DataDog/libdatadog.git", rev = "4e4cf372d4e461d43d9061324ecf60946d5dae53" } +libdd-common = { git = "https://github.com/DataDog/libdatadog.git", rev = "4e4cf372d4e461d43d9061324ecf60946d5dae53", default-features = false } +libdd-data-pipeline = { git = "https://github.com/DataDog/libdatadog.git", rev = "4e4cf372d4e461d43d9061324ecf60946d5dae53", default-features = false, features = ["agentless", "telemetry"] } +libdd-trace-utils = { git = "https://github.com/DataDog/libdatadog.git", rev = "4e4cf372d4e461d43d9061324ecf60946d5dae53", default-features = false, features = ["change-buffer"] } +libdd-trace-stats = { git = "https://github.com/DataDog/libdatadog.git", rev = "4e4cf372d4e461d43d9061324ecf60946d5dae53", default-features = false } +libdd-trace-protobuf = { git = "https://github.com/DataDog/libdatadog.git", rev = "4e4cf372d4e461d43d9061324ecf60946d5dae53", default-features = false } +libdd-shared-runtime = { git = "https://github.com/DataDog/libdatadog.git", rev = "4e4cf372d4e461d43d9061324ecf60946d5dae53", default-features = false } rmp-serde = "1" bytes = "1" http = "1" diff --git a/crates/pipeline/src/lib.rs b/crates/pipeline/src/lib.rs index eb25866e..e3bc96c6 100644 --- a/crates/pipeline/src/lib.rs +++ b/crates/pipeline/src/lib.rs @@ -1,7 +1,9 @@ -use libdatadog_nodejs_capabilities::WasmCapabilities; +use libdatadog_nodejs_capabilities::{WasmCapabilities, WasmEnvCapability}; use libdd_data_pipeline::trace_exporter::agent_response::AgentResponse; +use libdd_data_pipeline::trace_exporter::error::TraceExporterError; use libdd_data_pipeline::trace_exporter::{ - TelemetryConfig, TraceExporter, TraceExporterBuilder, TraceExporterOutputFormat, + ObfuscationConfig, TelemetryConfig, TraceExporter, TraceExporterBuilder, + TraceExporterOutputFormat, }; use libdd_data_pipeline::OtlpProtocol; use libdd_shared_runtime::LocalRuntime; @@ -33,6 +35,12 @@ fn init() { console_error_panic_hook::set_once(); } +struct AgentlessExporterConfig { + endpoint: String, + api_key: String, + obfuscation: ObfuscationConfig, +} + // --- span event attribute decoding --- // // `addSpanEvent` receives its attributes as a flat little-endian buffer built @@ -156,10 +164,8 @@ fn decode_span_event_attributes( pub struct WasmSpanState { change_queue: Vec, string_table_input: Vec, - /// UnsafeCell because send_trace_chunks_async needs &mut self across an - /// await point. WASM is single-threaded so this is safe — we just need - /// to ensure no overlapping mutable borrows (guaranteed by the JS-side - /// _flushInFlight guard which serializes sendPreparedChunk calls). + /// UnsafeCell because trace sends need &mut access across an await point. + /// WASM is single-threaded and `sending` rejects overlapping sends. /// On wasm the exporter must be built asynchronously (`build_async`), but /// the wasm-bindgen constructor is synchronous. We stash the configured /// builder here and build the exporter lazily on the first send (the only @@ -169,6 +175,10 @@ pub struct WasmSpanState { // ForkSafeRuntime/BasicRuntime are native-only. exporter: UnsafeCell>>, builder: UnsafeCell>>, + /// Agent endpoint applied only after the export mode is selected. + agent_url: Cell>, + /// Direct-intake settings applied instead of the Agent endpoint. + agentless_config: Cell>, cbs: RefCell>, stats_collector: RefCell>, /// Chunks staged by `prepareChunk`, one per trace (segment), sent together by @@ -176,11 +186,8 @@ pub struct WasmSpanState { /// flush batch by trace and calls `prepareChunk` once per trace so each chunk /// carries exactly one segment's spans with correct per-trace tags. prepared_spans: RefCell>>>, - /// Re-entrancy guard for `sendPreparedChunk`. wasm-bindgen async exports - /// can be invoked again from JS before the prior future resolves; without - /// this, two calls would each take `&mut` out of `exporter`/`builder` and - /// alias across the await (UB). The guard makes a re-entrant call return - /// an error instead. + /// Re-entrancy guard shared by both trace-send entry points. wasm-bindgen + /// async exports can be invoked again before the prior future resolves. sending: Cell, /// When true, the lazily-built exporter is configured for v0.5 output /// (`/v0.5/traces`) instead of the default v0.4. v0.5 is a smaller, fixed @@ -224,6 +231,106 @@ impl Drop for InFlightGuard<'_> { } } +impl WasmSpanState { + fn begin_send(&self, operation: &str) -> Result, JsValue> { + if self.sending.get() { + return Err(JsValue::from_str(&format!( + "{operation} is already in flight" + ))); + } + self.sending.set(true); + Ok(InFlightGuard(&self.sending)) + } + + async fn exporter( + &self, + _in_flight: &InFlightGuard<'_>, + ) -> Result<&mut TraceExporter, JsValue> { + // SAFETY: WASM is single-threaded and the `sending` guard held by the + // caller guarantees no overlapping invocation, so this is the only + // live reference to the exporter for the duration of the awaits. + let exporter_slot = unsafe { &mut *self.exporter.get() }; + if exporter_slot.is_none() { + // A previous build attempt failed. Building is one-shot and the + // failure is fatal (bad config won't fix itself), so return it + // consistently rather than a misleading "builder already consumed". + if let Some(message) = self.build_error.borrow().clone() { + return Err(build_failure_error(&message)); + } + // First send: build the exporter asynchronously. `build` is not + // available on wasm (it needs a blocking runtime), so drive + // `build_async` here where there is already an async context. + let mut builder = unsafe { &mut *self.builder.get() } + .take() + .ok_or_else(|| JsValue::from_str("exporter builder already consumed"))?; + if let Some(config) = self.agentless_config.take() { + builder.set_agentless_endpoint( + &config.endpoint, + &config.api_key, + config.obfuscation, + ); + } else if let Some(url) = self.agent_url.take() { + builder.set_url(&url); + } + if self.use_v05.get() { + builder.set_output_format(TraceExporterOutputFormat::V05); + } + if let Some(url) = self.otlp_endpoint.take() { + builder.set_otlp_endpoint(&url); + if let Some(protocol) = self.otlp_protocol.take() { + builder.set_otlp_protocol(protocol); + } + let headers = self.otlp_headers.take(); + if !headers.is_empty() { + builder.set_otlp_headers(headers); + } + } + if let Some(config) = self.telemetry_config.take() { + builder.enable_telemetry(config); + } + match builder.build_async::().await { + Ok(built) => *exporter_slot = Some(built), + Err(error) => { + let message = format!("native exporter build failed: {error:?}"); + *self.build_error.borrow_mut() = Some(message.clone()); + return Err(build_failure_error(&message)); + } + } + } + match exporter_slot.as_mut() { + Some(exporter) => Ok(exporter), + None => Err(build_failure_error("native exporter unavailable")), + } + } +} + +fn exporter_response( + response: Result, +) -> Result { + response + .map(|response| match response { + AgentResponse::Unchanged => JsValue::from_str("unchanged"), + AgentResponse::Changed { body } => JsValue::from_str(&body), + }) + .map_err(|error| JsValue::from_str(&format!("{error:?}"))) +} + +fn encoded_trace_count(data: &[u8]) -> Result { + if data.first() != Some(&0xdd) { + return Err(JsValue::from_str( + "sendEncodedTraces: payload must begin with a MessagePack array32", + )); + } + + if data.len() < 5 { + return Err(JsValue::from_str( + "sendEncodedTraces: incomplete MessagePack array32 header", + )); + } + + Ok(u32::from_be_bytes([data[1], data[2], data[3], data[4]]) as usize) +} + fn stats_flush_result(sent: bool, collapsed_spans: u64) -> Result { let result = js_sys::Object::new(); js_sys::Reflect::set( @@ -261,7 +368,6 @@ impl WasmSpanState { ) -> Result { let mut builder = TraceExporterBuilder::::new(); builder - .set_url(url) .set_tracer_version(tracer_version) .set_language(lang) .set_language_version(lang_version) @@ -334,6 +440,8 @@ impl WasmSpanState { string_table_input: vec![0u8; string_table_input_size as usize], exporter: UnsafeCell::new(None), builder: UnsafeCell::new(Some(builder)), + agent_url: Cell::new(Some(url.to_owned())), + agentless_config: Cell::new(None), cbs: RefCell::new(change_buffer_state), stats_collector: RefCell::new(stats_collector), prepared_spans: RefCell::new(Vec::new()), @@ -359,6 +467,20 @@ impl WasmSpanState { self.use_v05.set(v); } + /// Route trace export directly to the Datadog intake. Must be called before the first send. + /// Obfuscation settings are read through the Node.js environment capability at this boundary. + #[wasm_bindgen(js_name = "setAgentlessEndpoint")] + pub fn set_agentless_endpoint(&self, endpoint: String, api_key: String) -> Result<(), JsValue> { + let obfuscation = ObfuscationConfig::from_env(&WasmEnvCapability) + .map_err(|error| JsValue::from_str(&format!("setAgentlessEndpoint: {error}")))?; + self.agentless_config.set(Some(AgentlessExporterConfig { + endpoint, + api_key, + obfuscation, + })); + Ok(()) + } + /// Route trace export through libdatadog's OTLP HTTP exporter to `url` /// instead of the Datadog agent. Must be called before the first send. /// Takes precedence over `setUseV05` (OTLP bypasses the agent entirely). @@ -511,83 +633,41 @@ impl WasmSpanState { /// rejects that with an error instead of allowing aliasing (UB). #[wasm_bindgen(js_name = "sendPreparedChunk")] pub async fn send_prepared_chunk(&self) -> Result { - if self.sending.get() { - return Err(JsValue::from_str("sendPreparedChunk is already in flight")); - } - self.sending.set(true); - let _in_flight = InFlightGuard(&self.sending); + let in_flight = self.begin_send("sendPreparedChunk")?; let chunks = std::mem::take(&mut *self.prepared_spans.borrow_mut()); if chunks.is_empty() { return Err(JsValue::from_str("no prepared chunk to send")); } - // SAFETY: WASM is single-threaded and the `sending` guard above - // guarantees no overlapping invocation, so this is the only live - // reference to the exporter for the duration of the awaits. - let exporter_slot = unsafe { &mut *self.exporter.get() }; - if exporter_slot.is_none() { - // A previous build attempt failed. Building is one-shot and the - // failure is fatal (bad config won't fix itself), so return it - // consistently — as a distinguishable error the host bails on — - // rather than a misleading "builder already consumed". - if let Some(msg) = self.build_error.borrow().clone() { - return Err(build_failure_error(&msg)); - } - // First send: build the exporter asynchronously. `build` is not - // available on wasm (it needs a blocking runtime), so we drive - // `build_async` here where we already have an async context. - let mut builder = unsafe { &mut *self.builder.get() } - .take() - .ok_or_else(|| JsValue::from_str("exporter builder already consumed"))?; - // Output format is decided here, at first build, and then fixed. - // v0.5 drops meta_struct/span_events/span_links by design (the v0.5 - // schema has no slots for them); dd-trace-js only enables this after - // confirming agent `/v0.5/traces` support via `/info`. - if self.use_v05.get() { - builder.set_output_format(TraceExporterOutputFormat::V05); - } - // When an OTLP endpoint is configured, libdatadog exports traces via - // OTLP HTTP to that endpoint instead of the Datadog agent (mutually - // exclusive with the agent v0.4/v0.5 path). - if let Some(url) = self.otlp_endpoint.take() { - builder.set_otlp_endpoint(&url); - if let Some(protocol) = self.otlp_protocol.take() { - builder.set_otlp_protocol(protocol); - } - let headers = self.otlp_headers.take(); - if !headers.is_empty() { - builder.set_otlp_headers(headers); - } - } - if let Some(cfg) = self.telemetry_config.take() { - builder.enable_telemetry(cfg); - } - match builder.build_async::().await { - Ok(built) => *exporter_slot = Some(built), - Err(e) => { - // Latch the failure: the builder is now consumed and the - // config won't change, so every later send must fail fast. - let msg = format!("native exporter build failed: {e:?}"); - *self.build_error.borrow_mut() = Some(msg.clone()); - return Err(build_failure_error(&msg)); + let exporter = self.exporter(&in_flight).await?; + exporter_response(exporter.send_trace_chunks_async(chunks).await) + } + + /// Send one v0.4 MessagePack array32 payload containing one or more trace chunks. + /// The payload is owned by WASM for the full async send. + #[wasm_bindgen(js_name = "sendEncodedTraces")] + pub async fn send_encoded_traces(&self, data: Vec) -> Result { + let in_flight = self.begin_send("sendEncodedTraces")?; + let trace_count = encoded_trace_count(&data)?; + + if trace_count == 0 { + return Err(JsValue::from_str("no encoded traces to send")); + } + + { + let mut stats_collector = self.stats_collector.borrow_mut(); + if let Some(collector) = stats_collector.as_mut() { + let (chunks, _) = libdd_trace_utils::msgpack_decoder::v04::from_slice(&data) + .map_err(|error| JsValue::from_str(&format!("sendEncodedTraces: {error}")))?; + for chunk in &chunks { + collector.add_spans(chunk); } } } - let exporter = match exporter_slot.as_mut() { - Some(exporter) => exporter, - // Unreachable: the block above either set `Some` or returned early. - None => return Err(build_failure_error("native exporter unavailable")), - }; - let resp = exporter.send_trace_chunks_async(chunks).await; - let response_str = resp.map(|resp| match resp { - AgentResponse::Unchanged => "unchanged".to_string(), - AgentResponse::Changed { body } => body, - }); - response_str - .map(|s| JsValue::from_str(&s)) - .map_err(|e| JsValue::from_str(&format!("{:?}", e))) + let exporter = self.exporter(&in_flight).await?; + exporter_response(exporter.send_v04_payload_async(data, trace_count).await) } /// Flush aggregated stats to the agent's /v0.6/stats endpoint. diff --git a/crates/pipeline/src/stats.rs b/crates/pipeline/src/stats.rs index 3e65cb46..c1caf095 100644 --- a/crates/pipeline/src/stats.rs +++ b/crates/pipeline/src/stats.rs @@ -22,8 +22,7 @@ use libdd_capabilities::http::HttpClientCapability; use libdd_common::parse_uri; use libdd_trace_protobuf::pb; use libdd_trace_stats::span_concentrator::SpanConcentrator; - -use crate::trace_data::WasmTraceData; +use libdd_trace_utils::span::{v04::Span, TraceData}; const STATS_ENDPOINT_PATH: &str = "/v0.6/stats"; @@ -79,7 +78,7 @@ impl StatsCollector { /// /// The spans should already have `_dd.top_level` and `_dd.measured` metrics /// set (done by `ChangeBufferState::flush_chunk`). - pub fn add_spans(&mut self, spans: &[libdd_trace_utils::span::v04::Span]) { + pub fn add_spans(&mut self, spans: &[Span]) { for span in spans { self.concentrator.add_span(span); } diff --git a/load.js b/load.js index 69cdca51..b0b940c5 100644 --- a/load.js +++ b/load.js @@ -10,6 +10,7 @@ const PLATFORM = os.platform() const ARCH = process.arch const LIBC = PLATFORM === 'linux' ? (existsSync('/etc/alpine-release') ? 'musl' : 'glibc') : '' const ABI = process.versions.modules +const PIPELINE_API_VERSION = 1 const inWebpack = typeof __webpack_require__ === 'function' const runtimeRequire = inWebpack ? __non_webpack_require__ : require @@ -91,4 +92,4 @@ function findFile (root, name, binary = false) { || files.find(f => f === `${name}.node`) } -module.exports = { find, load, maybeLoad } +module.exports = { find, load, maybeLoad, pipelineApiVersion: PIPELINE_API_VERSION } diff --git a/test/pipeline.js b/test/pipeline.js index 59511fef..cd81de4f 100644 --- a/test/pipeline.js +++ b/test/pipeline.js @@ -4,7 +4,11 @@ const { describe, it, before, beforeEach } = require('node:test') const assert = require('node:assert') const crypto = require('node:crypto') -const pipeline = require('..').maybeLoad('pipeline') +const libdatadog = require('..') + +assert.strictEqual(libdatadog.pipelineApiVersion, 1) + +const pipeline = libdatadog.maybeLoad('pipeline') // The pipeline binding is wasm-only and is absent in the native // (action-prebuildify) test matrix, where `maybeLoad` returns undefined. Skip // the suite there instead of crashing on the destructure below; the pipeline @@ -13,6 +17,19 @@ const skip = pipeline === undefined const { WasmSpanState } = pipeline ?? {} const OpCode = pipeline ? pipeline.getOpCodes() : {} const wasmMemory = pipeline ? pipeline.getWasmMemory() : undefined +const ENCODED_TRACE_PAYLOAD = Buffer.from([ + 'dd00000001dd000000018ea474797065a3776562a874726163655f6964cf0000000000000001a77370616e5f6964cf', + '0000000000000002a9706172656e745f6964cf0000000000000000a46e616d65ac656e636f6465642d6e616d65a8', + '7265736f75726365b0656e636f6465642d7265736f75726365a773657276696365af656e636f6465642d7365727669', + '6365a56572726f7200a57374617274cf17979cfe362a0000a86475726174696f6ece000f4240a46d657461df000000', + '02ac656e636f6465642e6d657461ad656e636f6465642d76616c7565a97370616e2e6b696e64a6736572766572a7', + '6d657472696373df00000003ae656e636f6465642e6d65747269632aad5f64642e746f705f6c6576656c01ac5f64', + '642e6d6561737572656401ab7370616e5f6576656e7473dd0000000183a46e616d65ad656e636f6465642d657665', + '6e74ae74696d655f756e69785f6e616e6fcf17979cfe362a0000aa61747472696275746573df00000001a46b696e', + '6482a47479706500ac737472696e675f76616c7565b1656e636f6465642d617474726962757465ab6d6574615f73', + '7472756374df00000001a6617070736563c600000023df00000002a7626c6f636b6564c3a576616c7565ae656e63', + '6f6465642d737472756374', +].join(''), 'hex') function getRandomBytes (byteCount) { return new Uint8Array(crypto.randomBytes(byteCount)) @@ -26,6 +43,18 @@ function bytesToBigInt (bytes) { return val } +/** + * @param {string} name + * @param {string | undefined} value + */ +function restoreEnv (name, value) { + if (value === undefined) { + delete process.env[name] + } else { + process.env[name] = value + } +} + // The Span and NativeSpansInterface classes act as a sketch of what should // be implemented in dd-trace-js. @@ -785,9 +814,58 @@ describe('pipeline', { skip }, () => { it('flushSpans with no spans is a no-op returning false', async () => { assert.strictEqual(await nativeSpans.flushSpans(), false) }) + + it('rejects malformed and empty encoded trace payloads', async () => { + await assert.rejects(nativeSpans.state.sendEncodedTraces(Buffer.alloc(0)), /sendEncodedTraces/) + await assert.rejects(nativeSpans.state.sendEncodedTraces(Buffer.from([0x80])), /sendEncodedTraces/) + await assert.rejects(nativeSpans.state.sendEncodedTraces(Buffer.from([0xDC, 0])), /MessagePack array32/) + await assert.rejects(nativeSpans.state.sendEncodedTraces(Buffer.from([0xDD, 0, 0, 0])), /incomplete MessagePack/) + await assert.rejects( + nativeSpans.state.sendEncodedTraces(Buffer.from([0xDD, 0, 0, 0, 0])), + /no encoded traces to send/, + ) + }) }) describe('flush to agent', () => { + it('sends multiple encoded trace chunks without dropping structured fields', async () => { + const http = require('node:http') + const payloads = [] + const server = http.createServer((req, res) => { + const chunks = [] + req.on('data', chunk => chunks.push(chunk)) + req.on('end', () => { + payloads.push({ url: req.url, body: Buffer.concat(chunks) }) + res.writeHead(200, { 'content-type': 'application/json' }) + res.end('{}') + }) + }) + await new Promise(resolve => server.listen(0, '127.0.0.1', resolve)) + const { port } = server.address() + const ns = new NativeSpansInterface({ agentUrl: `http://127.0.0.1:${port}` }) + const payload = Buffer.concat([ + Buffer.from([0xDD, 0, 0, 0, 2]), + ENCODED_TRACE_PAYLOAD.subarray(5), + ENCODED_TRACE_PAYLOAD.subarray(5), + ]) + + try { + assert.ok(await ns.state.sendEncodedTraces(payload)) + const post = payloads.find(payload => payload.url.includes('/v0.4/traces')) + assert.ok(post) + assert.deepStrictEqual(post.body, payload) + assert.strictEqual(msgpackOuterArrayLen(post.body), 2) + assert.ok(post.body.includes(Buffer.from('encoded-name'))) + assert.ok(post.body.includes(Buffer.from('encoded-value'))) + assert.ok(post.body.includes(Buffer.from('encoded-struct'))) + assert.ok(post.body.includes(Buffer.from('encoded-event'))) + assert.ok(post.body.includes(Buffer.from('encoded-attribute'))) + } finally { + server.closeAllConnections?.() + server.close() + } + }) + it('should flush spans to a (mock) agent', async () => { // Stand up a throwaway HTTP server acting as the agent so the flush path // (prepareChunk -> build exporter -> serialize -> send) is exercised @@ -908,14 +986,8 @@ describe('pipeline', { skip }, () => { const { port } = server.address() const ns = new NativeSpansInterface({ agentUrl: `http://127.0.0.1:${port}` }) if (useV05) ns.state.setUseV05(true) - const span = ns.createSpan() - span.name = 'v05-span' - span.service = 'test-service' - span.resource = 'test-resource' - span.type = 'web' - span.duration = 1_000_000n try { - await ns.flushSpans(span) + await ns.state.sendEncodedTraces(ENCODED_TRACE_PAYLOAD) return seen.find(r => r.method === 'POST') } finally { server.closeAllConnections?.() @@ -963,14 +1035,8 @@ describe('pipeline', { skip }, () => { tracerVersion: '7.0.0-pre', }) ns.state.setOtlpEndpoint(`http://127.0.0.1:${port}/v1/traces`) - const span = ns.createSpan() - span.name = 'otlp-span' - span.service = 'test-service' - span.resource = 'test-resource' - span.type = 'web' - span.duration = 1_000_000n try { - await ns.flushSpans(span) + await ns.state.sendEncodedTraces(ENCODED_TRACE_PAYLOAD) const req = seen.find(r => r.method === 'POST') assert.ok(req, 'OTLP endpoint received a POST') assert.strictEqual(req.url, '/v1/traces') @@ -1011,14 +1077,8 @@ describe('pipeline', { skip }, () => { // Two header pairs plus a trailing unpaired element (odd length): both // pairs are applied and the stray 'ignored-no-pair' is dropped. ns.state.setOtlpHeaders(['authorization', 'Bearer test-token', 'x-custom', 'cval', 'ignored-no-pair']) - const span = ns.createSpan() - span.name = 'otlp-span' - span.service = 'test-service' - span.resource = 'test-resource' - span.type = 'web' - span.duration = 1_000_000n try { - await ns.flushSpans(span) + await ns.state.sendEncodedTraces(ENCODED_TRACE_PAYLOAD) assert.ok(captured, 'OTLP endpoint received a POST') assert.match(captured.ct || '', /protobuf/) assert.strictEqual(captured.auth, 'Bearer test-token') @@ -1035,6 +1095,102 @@ describe('pipeline', { skip }, () => { }) }) + describe('agentless output', () => { + it('formats and obfuscates through libdatadog before direct intake', async () => { + const http = require('node:http') + let captured + const server = http.createServer((req, res) => { + const chunks = [] + req.on('data', chunk => chunks.push(chunk)) + req.on('end', () => { + captured = { + apiKey: req.headers['dd-api-key'], + body: Buffer.concat(chunks), + contentType: req.headers['content-type'], + method: req.method, + url: req.url, + } + res.writeHead(200) + res.end() + }) + }) + await new Promise(resolve => server.listen(0, '127.0.0.1', resolve)) + const { port } = server.address() + const previousRules = process.env.DD_APM_REPLACE_TAGS + const nativeSpans = new NativeSpansInterface({ agentUrl: 'http://127.0.0.1:1' }) + + try { + process.env.DD_APM_REPLACE_TAGS = JSON.stringify([{ + name: 'encoded.meta', + pattern: 'encoded-value', + repl: '?', + }]) + nativeSpans.state.setAgentlessEndpoint(`http://127.0.0.1:${port}/v1/input`, 'test-api-key') + restoreEnv('DD_APM_REPLACE_TAGS', previousRules) + + await nativeSpans.state.sendEncodedTraces(ENCODED_TRACE_PAYLOAD) + + assert.ok(captured, 'direct intake received a POST') + assert.strictEqual(captured.body.includes(Buffer.from('encoded-value')), false) + assert.strictEqual(captured.method, 'POST') + assert.strictEqual(captured.url, '/v1/input') + assert.strictEqual(captured.apiKey, 'test-api-key') + assert.match(captured.contentType ?? '', /json/) + const payload = JSON.parse(captured.body) + const span = payload.traces[0].spans[0] + assert.strictEqual(span.start, 1_700_000_000) + assert.strictEqual(span.meta['encoded.meta'], '?') + assert.deepStrictEqual(span.meta_struct.appsec, { + blocked: true, + value: 'encoded-struct', + }) + + const oldStartPayload = Buffer.from(ENCODED_TRACE_PAYLOAD) + const startMarker = Buffer.from('a57374617274cf', 'hex') + const startMarkerOffset = oldStartPayload.indexOf(startMarker) + assert.notStrictEqual(startMarkerOffset, -1) + const startOffset = startMarkerOffset + startMarker.length + oldStartPayload.writeBigUInt64BE(1_000_000_000n, startOffset) + + await nativeSpans.state.sendEncodedTraces(oldStartPayload) + + const normalized = JSON.parse(captured.body).traces[0].spans[0] + assert.strictEqual(Number.isInteger(normalized.start), true) + assert.ok(normalized.start >= 946_684_800) + } finally { + restoreEnv('DD_APM_REPLACE_TAGS', previousRules) + server.closeAllConnections?.() + await new Promise(resolve => server.close(resolve)) + } + }) + + it('rejects malformed replacement rules before export', () => { + const previousRules = process.env.DD_APM_REPLACE_TAGS + const nativeSpans = new NativeSpansInterface() + + try { + process.env.DD_APM_REPLACE_TAGS = '[{"name":"tag","pattern":"[","repl":"?"}]' + assert.throws( + () => nativeSpans.state.setAgentlessEndpoint('http://127.0.0.1:1/v1/input', 'test-api-key'), + /setAgentlessEndpoint/, + ) + } finally { + restoreEnv('DD_APM_REPLACE_TAGS', previousRules) + } + }) + + it('rejects agentless and OTLP trace destinations together', async () => { + const nativeSpans = new NativeSpansInterface() + nativeSpans.state.setAgentlessEndpoint('http://127.0.0.1:1/v1/input', 'test-api-key') + nativeSpans.state.setOtlpEndpoint('http://127.0.0.1:1/v1/traces') + + await assert.rejects( + nativeSpans.state.sendEncodedTraces(ENCODED_TRACE_PAYLOAD), + /OTLP and agentless trace export cannot both be enabled/, + ) + }) + }) + describe('client-computed-stats header', () => { async function captureTraceHeader (nsOptions) { const http = require('node:http') @@ -1106,18 +1262,12 @@ describe('pipeline', { skip }, () => { await new Promise(resolve => server.listen(0, '127.0.0.1', resolve)) const { port } = server.address() - // statsEnabled:true builds the StatsCollector; prepareChunk feeds spans - // into it, and flushStats(true) force-flushes to /v0.6/stats. + // statsEnabled:true builds the StatsCollector; sendEncodedTraces feeds + // decoded spans into it, and flushStats(true) force-flushes to /v0.6/stats. const ns = new NativeSpansInterface({ agentUrl: `http://127.0.0.1:${port}`, 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) + await ns.state.sendEncodedTraces(ENCODED_TRACE_PAYLOAD) const result = await ns.state.flushStats(true) assert.deepStrictEqual(result, { sent: true, collapsedSpans: 0 }, 'flushStats reported a send') const statsReq = seen.find(r => r.url === '/v0.6/stats') @@ -1152,11 +1302,11 @@ describe('pipeline', { skip }, () => { let batch = [] try { - for (let i = 0; i < 15_000; i++) { + for (let i = 0; i < 7001; i++) { const span = ns.createSpan() span.name = 'stats-span' - span.service = 'stats-svc' - span.resource = `/stats/${i}` + span.service = `stats-svc-${i}` + span.resource = '/stats' span.type = 'web' span.setTag('span.kind', 'server') span.duration = 5_000_000n @@ -1173,7 +1323,7 @@ describe('pipeline', { skip }, () => { const result = await ns.state.flushStats(true) assert.strictEqual(result.sent, true) - assert.ok(result.collapsedSpans > 0, 'stats cardinality overflow reported collapsed spans') + assert.strictEqual(result.collapsedSpans, 1, 'stats cardinality overflow reported the first collapsed span') assert.ok(seen.some(r => r.url === '/v0.6/stats'), 'agent received a /v0.6/stats request') } finally { server.closeAllConnections?.() @@ -1249,7 +1399,7 @@ describe('pipeline', { skip }, () => { }) describe('send re-entrancy', () => { - it('rejects an overlapping sendPreparedChunk call', async () => { + it('rejects overlapping prepared and encoded sends', async () => { const http = require('node:http') const server = http.createServer((req, res) => { req.resume() @@ -1281,6 +1431,18 @@ describe('pipeline', { skip }, () => { reasons.some(r => /already in flight/.test(r)), 'one overlapping call rejected as already-in-flight', ) + + const encodedSettled = await Promise.allSettled([ + ns.state.sendEncodedTraces(ENCODED_TRACE_PAYLOAD), + ns.state.sendEncodedTraces(ENCODED_TRACE_PAYLOAD), + ]) + const encodedReasons = encodedSettled + .filter(result => result.status === 'rejected') + .map(result => String(result.reason)) + assert.ok( + encodedReasons.some(reason => /already in flight/.test(reason)), + 'one overlapping encoded call rejected as already-in-flight', + ) } finally { server.closeAllConnections?.() server.close()