From 76ee33a58d33d06beadda95b8b610390685d65a3 Mon Sep 17 00:00:00 2001 From: Bryan English Date: Mon, 6 Jul 2026 10:37:45 -0400 Subject: [PATCH] fix(pipeline): stage one chunk per trace in a single request prepareChunk overwrote the single prepared chunk on each call, so a flush carrying spans from multiple traces could only send the last one. Callers worked around that by passing every span to one prepareChunk, which lumps distinct trace_ids into a single chunk: flush_chunk then treats them as one segment and stamps trace-level tags (sampling priority, _dd.p.dm, origin, top_level) onto only the first span, corrupting sampling and grouping under load. Stage chunks in a Vec instead: prepareChunk pushes one chunk per call and sendPreparedChunk sends them all as a single multi-trace request. The exporter calls prepareChunk once per trace, so each chunk is one segment with correct per-trace tags. send_trace_chunks_async already accepts multiple chunks, so the request shape is unchanged. --- Cargo.lock | 8 ++--- crates/pipeline/src/lib.rs | 42 ++++++++++++------------ test/pipeline.js | 66 ++++++++++++++++++++++++++++++++++++++ 3 files changed, 90 insertions(+), 26 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index e973b89..9b0f46f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1071,7 +1071,7 @@ source = "git+https://github.com/DataDog/libdatadog.git?rev=7cdeb7896e92d1ba38bd dependencies = [ "anyhow", "libc", - "libdd-trace-protobuf 3.0.2 (git+https://github.com/DataDog/libdatadog.git?tag=v35.0.0)", + "libdd-trace-protobuf 3.0.2 (git+https://github.com/DataDog/libdatadog.git?rev=7cdeb7896e92d1ba38bde495934e112dac2eda25)", "memfd", "prost", "rand", @@ -1172,7 +1172,7 @@ dependencies = [ [[package]] name = "libdd-trace-protobuf" version = "3.0.2" -source = "git+https://github.com/DataDog/libdatadog.git?rev=7cdeb7896e92d1ba38bde495934e112dac2eda25#7cdeb7896e92d1ba38bde495934e112dac2eda25" +source = "git+https://github.com/DataDog/libdatadog.git?tag=v37.0.0#86b7f5700d3db4e58076794b5e8073a40d780083" dependencies = [ "prost", "serde", @@ -1182,7 +1182,7 @@ dependencies = [ [[package]] name = "libdd-trace-protobuf" version = "3.0.2" -source = "git+https://github.com/DataDog/libdatadog.git?tag=v37.0.0#86b7f5700d3db4e58076794b5e8073a40d780083" +source = "git+https://github.com/DataDog/libdatadog.git?rev=7cdeb7896e92d1ba38bde495934e112dac2eda25#7cdeb7896e92d1ba38bde495934e112dac2eda25" dependencies = [ "prost", "serde", @@ -1784,7 +1784,7 @@ version = "0.1.0" dependencies = [ "anyhow", "libdd-library-config 2.0.0", - "libdd-trace-protobuf", + "libdd-trace-protobuf 3.0.2 (git+https://github.com/DataDog/libdatadog.git?rev=7cdeb7896e92d1ba38bde495934e112dac2eda25)", "napi", "napi-derive", ] diff --git a/crates/pipeline/src/lib.rs b/crates/pipeline/src/lib.rs index 9385cbb..28cd78b 100644 --- a/crates/pipeline/src/lib.rs +++ b/crates/pipeline/src/lib.rs @@ -169,7 +169,11 @@ pub struct WasmSpanState { builder: UnsafeCell>>, cbs: RefCell>, stats_collector: RefCell>, - prepared_spans: 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 + /// 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 @@ -291,7 +295,7 @@ impl WasmSpanState { builder: UnsafeCell::new(Some(builder)), cbs: RefCell::new(change_buffer_state), stats_collector: RefCell::new(stats_collector), - prepared_spans: RefCell::new(None), + prepared_spans: RefCell::new(Vec::new()), sending: Cell::new(false), use_v05: Cell::new(false), otlp_endpoint: RefCell::new(None), @@ -389,11 +393,8 @@ impl WasmSpanState { )); } if len == 0 { - // Nothing to send: drop any previously prepared-but-unsent chunk so - // a caller that ignores this `false` cannot later resend a stale one. - if let Some(old_spans) = self.prepared_spans.borrow_mut().take() { - self.cbs.borrow_mut().recycle_spans(old_spans); - } + // Nothing to prepare for this trace; leave any chunks already staged + // for other traces in this same flush untouched. return Ok(false); } @@ -420,16 +421,14 @@ impl WasmSpanState { collector.add_spans(&spans_vec); } - // Recycle any previously prepared spans that were never sent (e.g. - // if the prior send was skipped by JS back-pressure). Reusing the - // pre-allocated HashMaps avoids allocator fragmentation in WASM. - if let Some(old_spans) = self.prepared_spans.borrow_mut().take() { - self.cbs.borrow_mut().recycle_spans(old_spans); - } - - // Store prepared spans for the subsequent sendPreparedChunk call + // 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 + // span already extracted) is not staged. let has_spans = !spans_vec.is_empty(); - *self.prepared_spans.borrow_mut() = Some(spans_vec); + if has_spans { + self.prepared_spans.borrow_mut().push(spans_vec); + } Ok(has_spans) } @@ -449,11 +448,10 @@ impl WasmSpanState { self.sending.set(true); let _in_flight = InFlightGuard(&self.sending); - let spans_vec = self - .prepared_spans - .borrow_mut() - .take() - .ok_or_else(|| JsValue::from_str("no prepared chunk to send"))?; + 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 @@ -510,7 +508,7 @@ impl WasmSpanState { None => return Err(build_failure_error("native exporter unavailable")), }; let resp = exporter - .send_trace_chunks_async(vec![spans_vec]) + .send_trace_chunks_async(chunks) .await; let response_str = resp.map(|resp| match resp { AgentResponse::Unchanged => "unchanged".to_string(), diff --git a/test/pipeline.js b/test/pipeline.js index 6827305..da8fda5 100644 --- a/test/pipeline.js +++ b/test/pipeline.js @@ -373,6 +373,16 @@ function encodeSpanEventAttrs (attributes) { return new Uint8Array(Buffer.concat(chunks)) } +// Read just the length of the outer msgpack array (the v0.4 trace payload is an +// array of trace chunks). Enough to assert how many separate traces were sent. +function msgpackOuterArrayLen (buf) { + const b = buf[0] + if (b >= 0x90 && b <= 0x9F) return b & 0x0F // fixarray + if (b === 0xDC) return buf.readUInt16BE(1) // array16 + if (b === 0xDD) return buf.readUInt32BE(1) // array32 + throw new Error('payload is not a msgpack array: 0x' + b.toString(16)) +} + describe('pipeline', { skip }, () => { let nativeSpans @@ -808,6 +818,62 @@ describe('pipeline', { skip }, () => { server.close() } }) + + it('accumulates one chunk per trace into a single multi-trace request', async () => { + // prepareChunk stages one chunk per call; a single sendPreparedChunk sends + // them all as one request. Before accumulation, a second prepareChunk + // overwrote the first, so only one trace shipped per request. Here two + // spans in two DISTINCT traces must arrive as two separate trace chunks. + const http = require('node:http') + const payloads = [] + const server = http.createServer((req, res) => { + const chunks = [] + req.on('data', c => chunks.push(c)) + 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 mk = (name) => { + const s = ns.createSpan() // distinct random trace id per span + s.name = name + s.service = 'svc' + s.resource = 'res' + s.type = 'web' + s.duration = 1_000_000n + return s + } + const a = mk('trace-a') + const b = mk('trace-b') + + // Prepare one chunk per trace (span id written LE), then send once. + const prepareOne = (span) => { + const buf = Buffer.alloc(8) + for (let i = 0; i < 8; i++) buf[i] = span.spanId[7 - i] + return ns.state.prepareChunk(1, true, buf) + } + + try { + assert.strictEqual(prepareOne(a), true, 'first chunk staged') + assert.strictEqual(prepareOne(b), true, 'second chunk staged') + const result = await ns.state.sendPreparedChunk() + assert(result, 'agent responded') + const post = payloads.find(p => p.url.includes('/v0.4/traces')) + assert.ok(post, 'received a v0.4 POST') + assert.strictEqual( + msgpackOuterArrayLen(post.body), 2, + 'payload carries two separate trace chunks, not one lumped chunk', + ) + } finally { + server.closeAllConnections?.() + server.close() + } + }) }) describe('v0.5 output format', () => {