fix(pipeline): stage one chunk per trace in a single request

crates/pipeline/src/lib.rs

@@ -169,7 +169,11 @@ pub struct WasmSpanState {
     builder: UnsafeCell<Option<TraceExporterBuilder<LocalRuntime>>>,
     cbs: RefCell<ChangeBufferState<WasmTraceData>>,
     stats_collector: RefCell<Option<stats::StatsCollector>>,
-    prepared_spans: RefCell<Option<Vec<libdd_trace_utils::span::v04::Span<WasmTraceData>>>>,
+    /// 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<Vec<Vec<libdd_trace_utils::span::v04::Span<WasmTraceData>>>>,
     /// 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(),

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', () => {