From 61e498f209cfd454f23f7fe96058c5bcd3f454be Mon Sep 17 00:00:00 2001 From: Jules Wiriath Date: Tue, 21 Jul 2026 14:13:59 +0200 Subject: [PATCH 1/9] docs: fix libdatadog's behavior when 0 is passed --- crates/pipeline/src/lib.rs | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/crates/pipeline/src/lib.rs b/crates/pipeline/src/lib.rs index eb25866e..7ad7ac95 100644 --- a/crates/pipeline/src/lib.rs +++ b/crates/pipeline/src/lib.rs @@ -379,7 +379,11 @@ impl WasmSpanState { Ok(()) } - /// Enable telemetry on the lazily-built trace exporter. + /// Enable telemetry on the lazily-built trace exporter. Off by default — + /// dd-trace-js opts in from JS. `heartbeat_ms` sets the metric-flush cadence + /// (0 defers to libdatadog's default interval), `runtime_id` + /// tags telemetry payloads with the tracer's runtime id when provided, and + /// `debug_enabled` toggles libdd-telemetry's verbose logging. /// /// Must be called before the first `sendPreparedChunk`. Later calls have /// no effect. From b0dfe71b35ce565027bf0b77acb1c8dd75e31680 Mon Sep 17 00:00:00 2001 From: Jules Wiriath Date: Tue, 21 Jul 2026 14:19:38 +0200 Subject: [PATCH 2/9] chore: js tomfoolery --- crates/pipeline/src/lib.rs | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/crates/pipeline/src/lib.rs b/crates/pipeline/src/lib.rs index 7ad7ac95..92bee1bc 100644 --- a/crates/pipeline/src/lib.rs +++ b/crates/pipeline/src/lib.rs @@ -380,10 +380,10 @@ impl WasmSpanState { } /// Enable telemetry on the lazily-built trace exporter. Off by default — - /// dd-trace-js opts in from JS. `heartbeat_ms` sets the metric-flush cadence - /// (0 defers to libdatadog's default interval), `runtime_id` - /// tags telemetry payloads with the tracer's runtime id when provided, and - /// `debug_enabled` toggles libdd-telemetry's verbose logging. + /// dd-trace-js opts in from JS. `heartbeat_ms`sets the metric-flush cadence + /// (0 defers to libdatadog's default interval), `runtime_id` tags telemetry + /// payloads with the tracer's runtime id when provided, and `debug_enabled` + /// toggles libdd-telemetry's verbose logging. /// /// Must be called before the first `sendPreparedChunk`. Later calls have /// no effect. From 3a3ba02dbac46b3e137db49f09957ba0339c4c50 Mon Sep 17 00:00:00 2001 From: Jules Wiriath Date: Mon, 27 Jul 2026 12:13:17 +0200 Subject: [PATCH 3/9] fix: comments --- crates/pipeline/src/lib.rs | 6 +----- 1 file changed, 1 insertion(+), 5 deletions(-) diff --git a/crates/pipeline/src/lib.rs b/crates/pipeline/src/lib.rs index 92bee1bc..eb25866e 100644 --- a/crates/pipeline/src/lib.rs +++ b/crates/pipeline/src/lib.rs @@ -379,11 +379,7 @@ impl WasmSpanState { Ok(()) } - /// Enable telemetry on the lazily-built trace exporter. Off by default — - /// dd-trace-js opts in from JS. `heartbeat_ms`sets the metric-flush cadence - /// (0 defers to libdatadog's default interval), `runtime_id` tags telemetry - /// payloads with the tracer's runtime id when provided, and `debug_enabled` - /// toggles libdd-telemetry's verbose logging. + /// Enable telemetry on the lazily-built trace exporter. /// /// Must be called before the first `sendPreparedChunk`. Later calls have /// no effect. From d0ace6f6c1e001c4056f7390ce5aa21394ea267e Mon Sep 17 00:00:00 2001 From: Jules Wiriath Date: Fri, 24 Jul 2026 15:27:39 +0200 Subject: [PATCH 4/9] feat: use libdatadog's css instead of local one --- Cargo.lock | 6 -- crates/pipeline/Cargo.toml | 6 -- crates/pipeline/src/lib.rs | 107 +++---------------- crates/pipeline/src/stats.rs | 201 ----------------------------------- test/pipeline.js | 173 +----------------------------- 5 files changed, 19 insertions(+), 474 deletions(-) delete mode 100644 crates/pipeline/src/stats.rs diff --git a/Cargo.lock b/Cargo.lock index e76eee06..7127688d 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2469,27 +2469,21 @@ checksum = "a89322df9ebe1c1578d689c92318e070967d1042b512afbe49518723f4e6d5cd" name = "pipeline" version = "0.1.0" dependencies = [ - "bytes", "console_error_panic_hook", "getrandom 0.2.17", - "http 1.5.0", "js-sys", "libdatadog-nodejs-capabilities", "libdd-capabilities 3.0.0 (git+https://github.com/DataDog/libdatadog.git?rev=966f921c226b63a1ba84c2e53e3d0f4625f90c75)", "libdd-common 5.2.0 (git+https://github.com/DataDog/libdatadog.git?rev=966f921c226b63a1ba84c2e53e3d0f4625f90c75)", "libdd-data-pipeline", "libdd-shared-runtime 3.0.0", - "libdd-trace-protobuf 4.0.1", - "libdd-trace-stats", "libdd-trace-utils 10.1.0", - "rmp-serde", "serde", "serde_json", "uuid", "wasm-bindgen", "wasm-bindgen-futures", "wasm-bindgen-test", - "web-time", ] [[package]] diff --git a/crates/pipeline/Cargo.toml b/crates/pipeline/Cargo.toml index 67b7aad3..0875b5ba 100644 --- a/crates/pipeline/Cargo.toml +++ b/crates/pipeline/Cargo.toml @@ -18,13 +18,7 @@ libdd-capabilities = { git = "https://github.com/DataDog/libdatadog.git", rev = libdd-common = { git = "https://github.com/DataDog/libdatadog.git", rev = "966f921c226b63a1ba84c2e53e3d0f4625f90c75", default-features = false } libdd-data-pipeline = { git = "https://github.com/DataDog/libdatadog.git", rev = "966f921c226b63a1ba84c2e53e3d0f4625f90c75", default-features = false, features = ["telemetry"] } libdd-trace-utils = { git = "https://github.com/DataDog/libdatadog.git", rev = "966f921c226b63a1ba84c2e53e3d0f4625f90c75", default-features = false, features = ["change-buffer"] } -libdd-trace-stats = { git = "https://github.com/DataDog/libdatadog.git", rev = "966f921c226b63a1ba84c2e53e3d0f4625f90c75", default-features = false } -libdd-trace-protobuf = { git = "https://github.com/DataDog/libdatadog.git", rev = "966f921c226b63a1ba84c2e53e3d0f4625f90c75", default-features = false } libdd-shared-runtime = { git = "https://github.com/DataDog/libdatadog.git", rev = "966f921c226b63a1ba84c2e53e3d0f4625f90c75", default-features = false } -rmp-serde = "1" -bytes = "1" -http = "1" -web-time = "1" console_error_panic_hook = "0.1" [target.'cfg(target_arch = "wasm32")'.dependencies] diff --git a/crates/pipeline/src/lib.rs b/crates/pipeline/src/lib.rs index eb25866e..1fb05c45 100644 --- a/crates/pipeline/src/lib.rs +++ b/crates/pipeline/src/lib.rs @@ -18,8 +18,6 @@ mod span_bytes; mod trace_data; use trace_data::*; -mod stats; - use libdd_trace_utils::change_buffer::{ChangeBuffer, ChangeBufferState}; use libdd_trace_utils::span::v04::{AttributeAnyValue, AttributeArrayValue, SpanEvent}; use span_string::SpanString; @@ -170,7 +168,6 @@ pub struct WasmSpanState { exporter: UnsafeCell>>, builder: UnsafeCell>>, cbs: RefCell>, - stats_collector: RefCell>, /// Chunks staged by `prepareChunk`, one per trace (segment), sent together by /// `sendPreparedChunk` as a single multi-trace request. The exporter groups a /// flush batch by trace and calls `prepareChunk` once per trace so each chunk @@ -224,21 +221,6 @@ impl Drop for InFlightGuard<'_> { } } -fn stats_flush_result(sent: bool, collapsed_spans: u64) -> Result { - let result = js_sys::Object::new(); - js_sys::Reflect::set( - &result, - &JsValue::from_str("sent"), - &JsValue::from_bool(sent), - )?; - js_sys::Reflect::set( - &result, - &JsValue::from_str("collapsedSpans"), - &JsValue::from_f64(collapsed_spans as f64), - )?; - Ok(result.into()) -} - #[wasm_bindgen] impl WasmSpanState { #[wasm_bindgen(constructor)] @@ -268,8 +250,7 @@ impl WasmSpanState { .set_language_interpreter(lang_interpreter) .set_otlp_instrumentation_scope("dd-trace-js", tracer_version) // Populate the payload-level TracerMetadata (service/env/hostname/ - // app_version) the agent receives. These values are already passed - // in for the stats collector; without these calls the trace + // app_version) the agent receives. Without these calls the trace // payload's tracer metadata is sent empty. .set_service(tracer_service) .set_env(env) @@ -278,16 +259,18 @@ impl WasmSpanState { .set_runtime_id(runtime_id) .enable_agent_rates_payload_version(); - // Advertise `Datadog-Client-Computed-Stats` so the agent skips its own - // APM stats/sampling for these traces. This is required in two cases: - // - `stats_enabled`: we build a StatsCollector and send client-side - // stats, so the agent MUST NOT also compute them (double counting); - // - `client_computed_stats`: set independently for APM-standalone - // (apmTracingEnabled=false), where the agent should skip APM stats - // even though we don't compute them client-side. - // Enabling stats therefore always implies the header, so OR the flags - // rather than relying on the caller to keep them in sync. - if client_computed_stats || stats_enabled { + // Client-side stats. Two disjoint modes: + // - `stats_enabled`: libdatadog runs the concentrator + /v0.6/stats + // worker natively (LocalRuntime::spawn_worker), gates activation on + // the agent's /info (`client_drop_p0s` + `/v0.6/stats`), and stamps + // `Datadog-Client-Computed-Stats` per trace request when stats + // actually run. + // - `client_computed_stats` (without `stats_enabled`): APM-standalone + // (apmTracingEnabled=false). We advertise the header so the agent + // skips its own APM stats, even though we don't compute any here. + if stats_enabled { + builder.enable_stats(Duration::from_secs(10)); + } else if client_computed_stats { builder.set_client_computed_stats(); } @@ -311,31 +294,12 @@ impl WasmSpanState { pid, ); - let stats_collector = if stats_enabled { - Some(stats::StatsCollector::new( - Duration::from_secs(10), - url.to_string(), - stats::StatsMeta { - hostname: hostname.to_string(), - env: env.to_string(), - version: app_version.to_string(), - lang: lang.to_string(), - tracer_version: tracer_version.to_string(), - runtime_id: runtime_id.to_string(), - service: tracer_service.to_string(), - }, - )) - } else { - None - }; - Ok(WasmSpanState { change_queue, string_table_input: vec![0u8; string_table_input_size as usize], exporter: UnsafeCell::new(None), builder: UnsafeCell::new(Some(builder)), cbs: RefCell::new(change_buffer_state), - stats_collector: RefCell::new(stats_collector), prepared_spans: RefCell::new(Vec::new()), sending: Cell::new(false), use_v05: Cell::new(false), @@ -439,8 +403,8 @@ impl WasmSpanState { self.string_table_input.len() as u32 } - /// Prepare a chunk of spans for sending. Flushes the change buffer, - /// extracts spans, feeds stats. Returns `true` if a chunk was prepared + /// Prepare a chunk of spans for sending. Flushes the change buffer and + /// extracts spans. Returns `true` if a chunk was prepared /// (there are spans to send) and `false` if there was nothing to send. /// Must be followed by `sendPreparedChunk()` to actually send. #[wasm_bindgen(js_name = "prepareChunk")] @@ -486,10 +450,6 @@ impl WasmSpanState { .flush_chunk(&span_ids, first_is_local_root) .map_err(|e| JsValue::from_str(&e.to_string()))?; - if let Some(collector) = self.stats_collector.borrow_mut().as_mut() { - collector.add_spans(&spans_vec); - } - // Stage this trace's chunk for the subsequent sendPreparedChunk call. // Multiple prepareChunk calls (one per trace) accumulate here and are // sent together as one multi-trace request. An empty result (e.g. every @@ -590,43 +550,6 @@ impl WasmSpanState { .map_err(|e| JsValue::from_str(&format!("{:?}", e))) } - /// Flush aggregated stats to the agent's /v0.6/stats endpoint. - /// - /// Should be called periodically (e.g. every 10s) from JS, and with - /// `force=true` on shutdown. Returns `{ sent, collapsedSpans }` so JS can - /// emit the same collapsed-span health metric as libdd-trace-stats's native - /// exporter. - #[wasm_bindgen(js_name = "flushStats")] - pub async fn flush_stats(&self, force: bool) -> Result { - // Build the stats request under a brief *synchronous* borrow, then drop - // the borrow BEFORE the async send. The collector therefore stays in - // `stats_collector`, so a concurrent `prepareChunk` during the in-flight - // send still reaches `add_spans` and those spans are counted. (Taking - // the collector out for the whole await would silently drop them from - // client-side stats.) No borrow is held across the await, so there is - // no double-borrow hazard from overlapping calls. - let prepared = { - let mut guard = self.stats_collector.borrow_mut(); - match guard.as_mut() { - Some(collector) => collector - .prepare_request(force) - .map_err(|e| JsValue::from_str(&e))?, - None => return stats_flush_result(false, 0), - } - }; - let sent = match prepared.request { - Some(req) => { - stats::StatsCollector::send_request(req) - .await - .map_err(|e| JsValue::from_str(&e))?; - true - } - None => false, - }; - - stats_flush_result(sent, prepared.collapsed_spans) - } - /// Flush the queued change-buffer operations. On success always returns /// `true` (the bool exists only for signature symmetry with the other /// flush methods); failures surface as a thrown error. diff --git a/crates/pipeline/src/stats.rs b/crates/pipeline/src/stats.rs deleted file mode 100644 index 3e65cb46..00000000 --- a/crates/pipeline/src/stats.rs +++ /dev/null @@ -1,201 +0,0 @@ -// Copyright 2026-Present Datadog, Inc. https://www.datadoghq.com/ -// SPDX-License-Identifier: Apache-2.0 - -//! Native stats collection for the pipeline WASM module. -//! -//! Wraps `SpanConcentrator` from `libdd-trace-stats` and provides encoding + -//! HTTP transport for flushing stats to the Datadog agent's `/v0.6/stats` -//! endpoint. - -use web_time::{Duration, SystemTime}; - -/// Wall-clock now() for wasm. Delegates to `web_time::SystemTime::now()`, -/// which routes to JS `Date.now()` on `wasm32-unknown-unknown` (native -/// `std::time::SystemTime::now()` is unimplemented on that target and traps). -fn now() -> SystemTime { - SystemTime::now() -} - -use bytes::Bytes; -use libdatadog_nodejs_capabilities::WasmHttpClient; -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; - -const STATS_ENDPOINT_PATH: &str = "/v0.6/stats"; - -/// Metadata for the stats payload envelope. -pub struct StatsMeta { - pub hostname: String, - pub env: String, - pub version: String, - pub lang: String, - pub tracer_version: String, - pub runtime_id: String, - pub service: String, -} - -/// Stats data prepared by a synchronous concentrator flush. -pub struct PreparedStatsFlush { - pub request: Option>, - pub collapsed_spans: u64, -} - -/// Manages stats aggregation and flushing. -pub struct StatsCollector { - concentrator: SpanConcentrator, - meta: StatsMeta, - agent_url: String, - sequence: u64, -} - -impl StatsCollector { - /// Create a new stats collector. - pub fn new(bucket_size: Duration, agent_url: String, meta: StatsMeta) -> Self { - StatsCollector { - concentrator: SpanConcentrator::new( - bucket_size, - now(), - vec![ - "client".to_string(), - "server".to_string(), - "producer".to_string(), - "consumer".to_string(), - ], - Vec::new(), - None, - Vec::new(), - ), - meta, - agent_url, - sequence: 0, - } - } - - /// Add spans to the concentrator for stats aggregation. - /// - /// 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]) { - for span in spans { - self.concentrator.add_span(span); - } - } - - /// Drain aggregated stats into a ready-to-send request plus flush metadata, - /// **synchronously**. - /// - /// Returns `request: None` when there is no stats payload to send. The - /// concentrator is drained and the sequence advanced as part of this call, - /// so a returned request must be sent (see `send_request`). Kept - /// synchronous and separate from the send so a caller can build the request - /// under a brief borrow and release the collector *before* the async send — - /// leaving it available for `add_spans` while the stats request is in - /// flight. - pub fn prepare_request(&mut self, force: bool) -> Result { - let mut flush = self.concentrator.flush(now(), force); - let collapsed_spans = flush.collapsed_spans; - if !flush.obfuscated_buckets.is_empty() { - // TODO: stats obfuscation is currently disabled. Obfuscated stats - // require the datadog-obfuscation-version header, which - // prepare_request doesn't emit yet. Add that header before enabling - // stats-obfuscation. - return Err( - "stats flush produced obfuscated buckets without obfuscation header support" - .to_string(), - ); - } - if flush.unobfuscated_buckets.is_empty() { - return Ok(PreparedStatsFlush { - request: None, - collapsed_spans, - }); - } - - self.sequence += 1; - let buckets = std::mem::take(&mut flush.unobfuscated_buckets); - let payload = encode_stats_payload(&buckets, &self.meta, self.sequence); - - let body = rmp_serde::encode::to_vec_named(&payload) - .map_err(|e| format!("stats msgpack encode error: {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) - .uri(uri) - .header("Content-Type", "application/msgpack") - .header("Datadog-Meta-Lang", &self.meta.lang) - .header("Datadog-Meta-Tracer-Version", &self.meta.tracer_version) - .body(Bytes::from(body)) - .map_err(|e| format!("failed to build stats request: {e}"))?; - - Ok(PreparedStatsFlush { - request: Some(req), - collapsed_spans, - }) - } - - /// Send a prepared stats request to the agent. Does **not** borrow the - /// collector, so trace export (`add_spans`) can proceed during the await. - pub async fn send_request(req: http::Request) -> Result<(), String> { - let client = WasmHttpClient::new_client(); - client - .request(req) - .await - .map_err(|e| format!("stats send error: {e:?}"))?; - Ok(()) - } -} - -/// Encode flushed stats buckets into a `ClientStatsPayload` for msgpack -/// serialization. -fn encode_stats_payload( - buckets: &[pb::ClientStatsBucket], - meta: &StatsMeta, - sequence: u64, -) -> pb::ClientStatsPayload { - pb::ClientStatsPayload { - hostname: meta.hostname.clone(), - env: meta.env.clone(), - version: meta.version.clone(), - lang: meta.lang.clone(), - tracer_version: meta.tracer_version.clone(), - runtime_id: meta.runtime_id.clone(), - sequence, - stats: buckets.to_vec(), - service: meta.service.clone(), - container_id: String::new(), - tags: Vec::new(), - agent_aggregation: String::new(), - git_commit_sha: String::new(), - image_tag: String::new(), - process_tags: String::new(), - process_tags_hash: 0, - } -} diff --git a/test/pipeline.js b/test/pipeline.js index bfd30a55..7089e81f 100644 --- a/test/pipeline.js +++ b/test/pipeline.js @@ -1081,175 +1081,10 @@ describe('pipeline', { skip }, () => { assert.strictEqual(header, undefined) }) - it('sends the header when stats are enabled (client-side stats imply it)', async () => { - // Enabling client-side stats without clientComputedStats must still send - // the header, otherwise the agent double-counts APM stats. - const { header, sawTraces } = await captureTraceHeader({ statsEnabled: true, clientComputedStats: false }) - assert.ok(sawTraces, 'expected a POST to /v0.4/traces') - assert.strictEqual(header, 'true') - }) - }) - - describe('client-side stats', () => { - it('aggregates and flushes stats to /v0.6/stats', async () => { - const http = require('node:http') - 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 => 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. - 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) - 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') - assert.ok(statsReq, 'agent received a /v0.6/stats request') - assert.strictEqual(statsReq.method, 'PUT') - assert.ok(statsReq.len > 0, 'stats payload is non-empty') - - // Nothing new aggregated -> a second forced flush is a no-op. - assert.deepStrictEqual(await ns.state.flushStats(true), { sent: false, collapsedSpans: 0 }, 'second flush has nothing to send') - } finally { - server.closeAllConnections?.() - server.close() - } - }) - - it('returns collapsed span count when stats cardinality overflows', async () => { - // The resource field has its own cardinality limit (1024) and collapses to a placeholder long - // before the whole-key limit (~7000) is reached, so varying it no longer overflows the key at all. - // `service` has no per-field limit, so distinct services still overflow the whole key -- which is - // what `collapsedSpans` counts. - const http = require('node:http') - 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 => server.listen(0, '127.0.0.1', resolve)) - const { port } = server.address() - - const ns = new NativeSpansInterface({ agentUrl: `http://127.0.0.1:${port}`, statsEnabled: true }) - let batch = [] - - try { - for (let i = 0; i < 15_000; i++) { - const span = ns.createSpan() - span.name = 'stats-span' - span.service = `stats-svc-${i}` - span.resource = '/stats' - span.type = 'web' - span.setTag('span.kind', 'server') - span.duration = 5_000_000n - ns.flushChangeQueue() - batch.push(span) - if (batch.length === 500) { - await ns.flushSpans(...batch) - batch = [] - } - } - if (batch.length > 0) { - await ns.flushSpans(...batch) - } - - const result = await ns.state.flushStats(true) - assert.strictEqual(result.sent, true) - assert.ok(result.collapsedSpans > 0, 'stats cardinality overflow reported collapsed spans') - assert.ok(seen.some(r => r.url === '/v0.6/stats'), 'agent received a /v0.6/stats request') - } finally { - server.closeAllConnections?.() - server.close() - } - }) - - it('flushStats reports no send when stats are disabled', async () => { - const ns = new NativeSpansInterface({ statsEnabled: false }) - assert.deepStrictEqual(await ns.state.flushStats(true), { sent: false, collapsedSpans: 0 }) - }) - - 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 result = await ns.state.flushStats(true) - assert.deepStrictEqual( - result, - { sent: true, collapsedSpans: 0 }, - '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 - } - } - }) + // When statsEnabled=true, libdatadog stamps the header dynamically only + // after the agent /info advertises `client_drop_p0s` + `/v0.6/stats`. + // That's inherently race-y in a unit test, so it's covered end-to-end + // instead of asserted here. }) describe('send re-entrancy', () => { From 4c0bd84e5a5e10f92c628935e5066bfb80075fa6 Mon Sep 17 00:00:00 2001 From: Jules Wiriath Date: Fri, 24 Jul 2026 16:48:01 +0200 Subject: [PATCH 5/9] feat: added force shutdown binding, and trimmed out verbose comments --- crates/pipeline/src/lib.rs | 49 ++++++++++++++++++++++++++++---------- test/pipeline.js | 5 ---- 2 files changed, 37 insertions(+), 17 deletions(-) diff --git a/crates/pipeline/src/lib.rs b/crates/pipeline/src/lib.rs index 1fb05c45..2ddddd09 100644 --- a/crates/pipeline/src/lib.rs +++ b/crates/pipeline/src/lib.rs @@ -249,9 +249,7 @@ impl WasmSpanState { .set_language_version(lang_version) .set_language_interpreter(lang_interpreter) .set_otlp_instrumentation_scope("dd-trace-js", tracer_version) - // Populate the payload-level TracerMetadata (service/env/hostname/ - // app_version) the agent receives. Without these calls the trace - // payload's tracer metadata is sent empty. + // Without these setters the payload-level TracerMetadata is empty. .set_service(tracer_service) .set_env(env) .set_hostname(hostname) @@ -259,15 +257,11 @@ impl WasmSpanState { .set_runtime_id(runtime_id) .enable_agent_rates_payload_version(); - // Client-side stats. Two disjoint modes: - // - `stats_enabled`: libdatadog runs the concentrator + /v0.6/stats - // worker natively (LocalRuntime::spawn_worker), gates activation on - // the agent's /info (`client_drop_p0s` + `/v0.6/stats`), and stamps - // `Datadog-Client-Computed-Stats` per trace request when stats - // actually run. - // - `client_computed_stats` (without `stats_enabled`): APM-standalone - // (apmTracingEnabled=false). We advertise the header so the agent - // skips its own APM stats, even though we don't compute any here. + // `enable_stats` makes libdatadog stamp the client-computed-stats + // header itself when stats run, so the two flags are disjoint: + // `set_client_computed_stats` is only for APM-standalone + // (`apmTracingEnabled=false`), where the header is advertised + // without any stats actually being computed. if stats_enabled { builder.enable_stats(Duration::from_secs(10)); } else if client_computed_stats { @@ -562,6 +556,37 @@ impl WasmSpanState { Ok(true) } + /// Force-flush the current stats bucket and stop the background workers. + /// The stats worker only flushes on its bucket interval, so spans + /// recorded between the last tick and process exit are lost unless this + /// is awaited during tracer shutdown. Consumes the exporter — subsequent + /// sends will error. + /// + /// `timeoutMs` bounds the wait; on timeout returns an error and workers + /// may still be finishing. `None` waits indefinitely. + #[wasm_bindgen(js_name = "shutdown")] + pub async fn shutdown(&self, timeout_ms: Option) -> Result<(), JsValue> { + // `sendPreparedChunk` holds an `&mut` on the exporter across awaits; + // taking it out from under an in-flight send would alias. + if self.sending.get() { + return Err(JsValue::from_str("shutdown: sendPreparedChunk in flight")); + } + self.sending.set(true); + let _in_flight = InFlightGuard(&self.sending); + + // SAFETY: `sending` guard prevents overlapping access; WASM is single-threaded. + let exporter_slot = unsafe { &mut *self.exporter.get() }; + // Idempotent: never-built or already-shut-down state is not an error. + let Some(exporter) = exporter_slot.take() else { + return Ok(()); + }; + let timeout = timeout_ms.map(|ms| Duration::from_millis(u64::from(ms))); + exporter + .shutdown_async(timeout) + .await + .map_err(|e| JsValue::from_str(&format!("shutdown: {e:?}"))) + } + /// Set default meta tags applied to every new span. /// Takes a flat array of key-value pairs: [key1, val1, key2, val2, ...] #[wasm_bindgen(js_name = "setDefaultMeta")] diff --git a/test/pipeline.js b/test/pipeline.js index 7089e81f..08032c07 100644 --- a/test/pipeline.js +++ b/test/pipeline.js @@ -1080,11 +1080,6 @@ describe('pipeline', { skip }, () => { assert.ok(sawTraces, 'expected a POST to /v0.4/traces') assert.strictEqual(header, undefined) }) - - // When statsEnabled=true, libdatadog stamps the header dynamically only - // after the agent /info advertises `client_drop_p0s` + `/v0.6/stats`. - // That's inherently race-y in a unit test, so it's covered end-to-end - // instead of asserted here. }) describe('send re-entrancy', () => { From 8f9a8f21f9c63959307f0d542d5f013d655c756a Mon Sep 17 00:00:00 2001 From: Jules Wiriath Date: Mon, 27 Jul 2026 12:16:12 +0200 Subject: [PATCH 6/9] docs: fix --- crates/pipeline/src/lib.rs | 8 +++----- 1 file changed, 3 insertions(+), 5 deletions(-) diff --git a/crates/pipeline/src/lib.rs b/crates/pipeline/src/lib.rs index 2ddddd09..b3108f15 100644 --- a/crates/pipeline/src/lib.rs +++ b/crates/pipeline/src/lib.rs @@ -556,11 +556,9 @@ impl WasmSpanState { Ok(true) } - /// Force-flush the current stats bucket and stop the background workers. - /// The stats worker only flushes on its bucket interval, so spans - /// recorded between the last tick and process exit are lost unless this - /// is awaited during tracer shutdown. Consumes the exporter — subsequent - /// sends will error. + /// Gracefully shut down the exporter and stop the background workers. + /// Should be awaited during tracer shutdown to avoid losing in-flight + /// data. Consumes the exporter — subsequent sends will error. /// /// `timeoutMs` bounds the wait; on timeout returns an error and workers /// may still be finishing. `None` waits indefinitely. From 8b17010707d45808707f00e9e556f2a569415c1c Mon Sep 17 00:00:00 2001 From: Jules Wiriath Date: Fri, 28 Aug 2026 16:27:42 +0200 Subject: [PATCH 7/9] feat: test and improve shutdown, other nits --- crates/pipeline/src/lib.rs | 29 +++++-- test/pipeline.js | 153 +++++++++++++++++++++++++++++++++++++ 2 files changed, 177 insertions(+), 5 deletions(-) diff --git a/crates/pipeline/src/lib.rs b/crates/pipeline/src/lib.rs index b3108f15..74f6a69b 100644 --- a/crates/pipeline/src/lib.rs +++ b/crates/pipeline/src/lib.rs @@ -205,6 +205,10 @@ pub struct WasmSpanState { /// config. Only takes effect if set before the first send /// (when the exporter is built). telemetry_config: Cell>, + /// Set by `shutdown`. Terminal: the exporter (and the builder, if the + /// exporter was never built) is dropped, so later sends must fail fast + /// instead of lazily building a fresh exporter behind the caller's back. + shut_down: Cell, /// Latched message from a failed lazy `build_async`. Building is one-shot and /// a failure is fatal (bad config), so once set every send returns it (as a /// distinguishable error) instead of a misleading "builder already consumed", @@ -301,6 +305,7 @@ impl WasmSpanState { otlp_protocol: Cell::new(None), otlp_headers: Cell::new(Vec::new()), telemetry_config: Cell::new(None), + shut_down: Cell::new(false), build_error: RefCell::new(None), }) } @@ -468,6 +473,11 @@ impl WasmSpanState { if self.sending.get() { return Err(JsValue::from_str("sendPreparedChunk is already in flight")); } + if self.shut_down.get() { + return Err(JsValue::from_str( + "sendPreparedChunk: exporter has been shut down", + )); + } self.sending.set(true); let _in_flight = InFlightGuard(&self.sending); @@ -491,9 +501,11 @@ impl WasmSpanState { // 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. + // Unreachable in practice: the builder is only taken here (after + // which the exporter exists) or by `shutdown` (rejected above). let mut builder = unsafe { &mut *self.builder.get() } .take() - .ok_or_else(|| JsValue::from_str("exporter builder already consumed"))?; + .ok_or_else(|| JsValue::from_str("exporter builder unavailable"))?; // 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 @@ -522,7 +534,7 @@ impl WasmSpanState { 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:?}"); + let msg = format!("native exporter build failed: {e}"); *self.build_error.borrow_mut() = Some(msg.clone()); return Err(build_failure_error(&msg)); } @@ -541,7 +553,7 @@ impl WasmSpanState { response_str .map(|s| JsValue::from_str(&s)) - .map_err(|e| JsValue::from_str(&format!("{:?}", e))) + .map_err(|e| JsValue::from_str(&e.to_string())) } /// Flush the queued change-buffer operations. On success always returns @@ -558,7 +570,9 @@ impl WasmSpanState { /// Gracefully shut down the exporter and stop the background workers. /// Should be awaited during tracer shutdown to avoid losing in-flight - /// data. Consumes the exporter — subsequent sends will error. + /// data. Terminal and idempotent: it consumes the exporter (and the + /// builder, if the exporter was never built), so later sends error out + /// rather than silently starting a new exporter. /// /// `timeoutMs` bounds the wait; on timeout returns an error and workers /// may still be finishing. `None` waits indefinitely. @@ -571,9 +585,14 @@ impl WasmSpanState { } self.sending.set(true); let _in_flight = InFlightGuard(&self.sending); + self.shut_down.set(true); // SAFETY: `sending` guard prevents overlapping access; WASM is single-threaded. let exporter_slot = unsafe { &mut *self.exporter.get() }; + // Drop the builder too: without it a shutdown that happens before the + // first send would leave the lazy build path armed, and the next send + // would quietly spin up a fresh exporter after shutdown. + unsafe { &mut *self.builder.get() }.take(); // Idempotent: never-built or already-shut-down state is not an error. let Some(exporter) = exporter_slot.take() else { return Ok(()); @@ -582,7 +601,7 @@ impl WasmSpanState { exporter .shutdown_async(timeout) .await - .map_err(|e| JsValue::from_str(&format!("shutdown: {e:?}"))) + .map_err(|e| JsValue::from_str(&format!("shutdown: {e}"))) } /// Set default meta tags applied to every new span. diff --git a/test/pipeline.js b/test/pipeline.js index 08032c07..3e7be0d9 100644 --- a/test/pipeline.js +++ b/test/pipeline.js @@ -1121,4 +1121,157 @@ describe('pipeline', { skip }, () => { } }) }) + + // Minimal agent stub: records every request and advertises the /info + // capabilities libdatadog gates client-side stats on (`client_drop_p0s` plus + // a `/v0.6/stats` endpoint). Without those the stats worker stays disabled. + // + // Every response carries `Datadog-Agent-State`: the info cache is a process + // global shared by every exporter built in this test file, so an earlier test + // may already have populated it from its own stub. The state header makes the + // exporter re-fetch `/info` from *this* stub instead of trusting that cache. + async function startAgentStub () { + const http = require('node:http') + const seen = [] + const state = `state-${process.hrtime.bigint()}` + const info = JSON.stringify({ + version: '7.60.0', + client_drop_p0s: true, + endpoints: ['/v0.4/traces', '/v0.5/traces', '/v0.6/stats', '/info'], + peer_tags: [], + span_kinds_stats_computed: ['server', 'consumer', 'client', 'producer'], + }) + 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, + headers: req.headers, + }) + const headers = { 'content-type': 'application/json', 'datadog-agent-state': state } + res.writeHead(200, headers) + res.end(req.url === '/info' ? info : '{}') + }) + }) + await new Promise(resolve => server.listen(0, '127.0.0.1', resolve)) + return { + seen, + url: `http://127.0.0.1:${server.address().port}`, + close () { + server.closeAllConnections?.() + server.close() + }, + } + } + + function makeSpan (ns, resource) { + const span = ns.createSpan() + span.name = 'stats-span' + span.service = 'stats-svc' + span.resource = resource + span.type = 'web' + span.setTag('span.kind', 'server') + span.duration = 5_000_000n + return span + } + + describe('client-side stats', () => { + it('computes stats natively and flushes them to /v0.6/stats on shutdown', async () => { + const agent = await startAgentStub() + const ns = new NativeSpansInterface({ agentUrl: agent.url, statsEnabled: true }) + + try { + // Stats only start once the /info fetcher has confirmed the agent + // supports them, which happens asynchronously after the exporter is + // built on the first send. Keep sending until libdatadog stamps the + // client-computed-stats header, which it only does once the + // concentrator is actually running. + let tracesWithStatsHeader + for (let i = 0; i < 50; i++) { + await ns.flushSpans(makeSpan(ns, `/stats/${i}`)) + tracesWithStatsHeader = agent.seen.filter( + r => r.url === '/v0.4/traces' && r.headers['datadog-client-computed-stats'], + ) + if (tracesWithStatsHeader.length > 0) break + await new Promise(resolve => setTimeout(resolve, 20)) + } + assert.ok(agent.seen.some(r => r.url === '/info'), 'exporter polled /info') + assert.ok( + tracesWithStatsHeader.length > 0, + 'traces carry Datadog-Client-Computed-Stats once stats are running', + ) + + // The worker only flushes on its bucket interval; shutdown forces the + // last bucket out, which is the point of the binding. + assert.ok(!agent.seen.some(r => r.url === '/v0.6/stats'), 'no stats sent before shutdown') + await ns.state.shutdown(10_000) + + const statsReq = agent.seen.find(r => r.url === '/v0.6/stats') + assert.ok(statsReq, 'agent received a /v0.6/stats request') + assert.strictEqual(statsReq.method, 'POST') + assert.ok(statsReq.len > 0, 'stats payload is non-empty') + } finally { + agent.close() + } + }) + + it('does not send stats when stats are disabled', async () => { + const agent = await startAgentStub() + const ns = new NativeSpansInterface({ agentUrl: agent.url, statsEnabled: false }) + + try { + await ns.flushSpans(makeSpan(ns, '/no-stats')) + await ns.state.shutdown(10_000) + assert.ok(!agent.seen.some(r => r.url === '/v0.6/stats'), 'no /v0.6/stats request') + } finally { + agent.close() + } + }) + }) + + describe('shutdown', () => { + it('is a no-op when the exporter was never built', async () => { + const ns = new NativeSpansInterface({ agentUrl: 'http://127.0.0.1:1' }) + await ns.state.shutdown(1000) + }) + + it('is idempotent', async () => { + const agent = await startAgentStub() + const ns = new NativeSpansInterface({ agentUrl: agent.url }) + try { + await ns.flushSpans(makeSpan(ns, '/shutdown')) + await ns.state.shutdown(10_000) + await ns.state.shutdown(10_000) + } finally { + agent.close() + } + }) + + it('rejects sends after shutdown, including before the first send', async () => { + const agent = await startAgentStub() + try { + // Shutting down before anything was sent must also consume the builder, + // otherwise the next send would lazily build a brand new exporter. + const early = new NativeSpansInterface({ agentUrl: agent.url }) + await early.state.shutdown(1000) + await assert.rejects( + early.flushSpans(makeSpan(early, '/after-shutdown')), + /shut down/, + ) + + const used = new NativeSpansInterface({ agentUrl: agent.url }) + await used.flushSpans(makeSpan(used, '/before-shutdown')) + await used.state.shutdown(10_000) + await assert.rejects( + used.flushSpans(makeSpan(used, '/after-shutdown')), + /shut down/, + ) + } finally { + agent.close() + } + }) + }) }) From d3cd5855b87455d8b8059e44248f0f5fc5bc4f9b Mon Sep 17 00:00:00 2001 From: Jules Wiriath Date: Tue, 1 Sep 2026 11:21:43 +0200 Subject: [PATCH 8/9] fix: return error instead when misconfiguration --- crates/pipeline/src/lib.rs | 11 ++++++----- test/pipeline.js | 7 +++++++ 2 files changed, 13 insertions(+), 5 deletions(-) diff --git a/crates/pipeline/src/lib.rs b/crates/pipeline/src/lib.rs index 74f6a69b..3aa55ebc 100644 --- a/crates/pipeline/src/lib.rs +++ b/crates/pipeline/src/lib.rs @@ -245,6 +245,12 @@ impl WasmSpanState { runtime_id: &str, client_computed_stats: bool, ) -> Result { + if stats_enabled && client_computed_stats { + return Err(JsValue::from_str( + "WasmSpanState: statsEnabled and clientComputedStats are mutually exclusive", + )); + } + let mut builder = TraceExporterBuilder::::new(); builder .set_url(url) @@ -261,11 +267,6 @@ impl WasmSpanState { .set_runtime_id(runtime_id) .enable_agent_rates_payload_version(); - // `enable_stats` makes libdatadog stamp the client-computed-stats - // header itself when stats run, so the two flags are disjoint: - // `set_client_computed_stats` is only for APM-standalone - // (`apmTracingEnabled=false`), where the header is advertised - // without any stats actually being computed. if stats_enabled { builder.enable_stats(Duration::from_secs(10)); } else if client_computed_stats { diff --git a/test/pipeline.js b/test/pipeline.js index 3e7be0d9..165937f2 100644 --- a/test/pipeline.js +++ b/test/pipeline.js @@ -1080,6 +1080,13 @@ describe('pipeline', { skip }, () => { assert.ok(sawTraces, 'expected a POST to /v0.4/traces') assert.strictEqual(header, undefined) }) + + it('rejects statsEnabled together with clientComputedStats', () => { + assert.throws( + () => new NativeSpansInterface({ statsEnabled: true, clientComputedStats: true }), + /mutually exclusive/, + ) + }) }) describe('send re-entrancy', () => { From f1d8e97b25f23b5d8c0a4692ae3cf6885c453fa4 Mon Sep 17 00:00:00 2001 From: Jules Wiriath Date: Tue, 1 Sep 2026 11:40:39 +0200 Subject: [PATCH 9/9] fix: leak --- crates/pipeline/src/lib.rs | 10 +++++++++- test/pipeline.js | 36 +++++++++++++++++++++++++++++++++--- 2 files changed, 42 insertions(+), 4 deletions(-) diff --git a/crates/pipeline/src/lib.rs b/crates/pipeline/src/lib.rs index 3aa55ebc..c87a7ede 100644 --- a/crates/pipeline/src/lib.rs +++ b/crates/pipeline/src/lib.rs @@ -414,6 +414,11 @@ impl WasmSpanState { first_is_local_root: bool, chunk: &[u8], ) -> Result { + if self.shut_down.get() { + return Err(JsValue::from_str( + "prepareChunk: exporter has been shut down", + )); + } // Validate the JS-supplied count against the actual buffer size before // doing any work: each span id is a u64 (8 bytes). This prevents an // out-of-bounds read panic (and a huge `Vec::with_capacity`) when the @@ -573,7 +578,9 @@ impl WasmSpanState { /// Should be awaited during tracer shutdown to avoid losing in-flight /// data. Terminal and idempotent: it consumes the exporter (and the /// builder, if the exporter was never built), so later sends error out - /// rather than silently starting a new exporter. + /// rather than silently starting a new exporter. Staged-but-unsent chunks + /// are dropped and `prepareChunk` errors afterwards, so flush before + /// shutting down. /// /// `timeoutMs` bounds the wait; on timeout returns an error and workers /// may still be finishing. `None` waits indefinitely. @@ -587,6 +594,7 @@ impl WasmSpanState { self.sending.set(true); let _in_flight = InFlightGuard(&self.sending); self.shut_down.set(true); + self.prepared_spans.borrow_mut().clear(); // SAFETY: `sending` guard prevents overlapping access; WASM is single-threaded. let exporter_slot = unsafe { &mut *self.exporter.get() }; diff --git a/test/pipeline.js b/test/pipeline.js index 165937f2..84c75450 100644 --- a/test/pipeline.js +++ b/test/pipeline.js @@ -1257,7 +1257,7 @@ describe('pipeline', { skip }, () => { } }) - it('rejects sends after shutdown, including before the first send', async () => { + it('rejects flushes after shutdown, including before the first send', async () => { const agent = await startAgentStub() try { // Shutting down before anything was sent must also consume the builder, @@ -1266,7 +1266,7 @@ describe('pipeline', { skip }, () => { await early.state.shutdown(1000) await assert.rejects( early.flushSpans(makeSpan(early, '/after-shutdown')), - /shut down/, + /prepareChunk: exporter has been shut down/, ) const used = new NativeSpansInterface({ agentUrl: agent.url }) @@ -1274,11 +1274,41 @@ describe('pipeline', { skip }, () => { await used.state.shutdown(10_000) await assert.rejects( used.flushSpans(makeSpan(used, '/after-shutdown')), - /shut down/, + /prepareChunk: exporter has been shut down/, ) } finally { agent.close() } }) + + it('rejects sendPreparedChunk directly after shutdown', async () => { + const agent = await startAgentStub() + const ns = new NativeSpansInterface({ agentUrl: agent.url }) + try { + await ns.state.shutdown(1000) + await assert.rejects( + ns.state.sendPreparedChunk(), + /sendPreparedChunk: exporter has been shut down/, + ) + } finally { + agent.close() + } + }) + + it('does not accumulate staged chunks once shut down', async () => { + const agent = await startAgentStub() + const ns = new NativeSpansInterface({ agentUrl: agent.url }) + try { + await ns.state.shutdown(1000) + for (let i = 0; i < 3; i++) { + assert.throws( + () => ns.state.prepareChunk(1, true, ns.flushBuffer), + /prepareChunk: exporter has been shut down/, + ) + } + } finally { + agent.close() + } + }) }) })