fix(native-spans): send one request per trace chunk again

The previous commit staged every group and sent the whole flush as one
multi-trace payload. The native chunk Vec does accumulate, so that works
at the protocol level, but it changes what a consumer observes per
payload: `traces[0]` is then whichever chunk happened to be staged
first, not the only chunk.

That broke 19 plugin suites which were green on the parent commit. rhea,
for example, reads `traces[0][0]` and got `amqp.send` where it expected
`amqp.receive`. The legacy AgentWriter produces one trace per payload at
the flush intervals these paths run with, and the test agent plus every
`traces[0][0]` assertion depends on that shape.

Restore the per-group prepare/send chain, and with it the per-chunk
`.requests`/`.responses` and OTLP export counters. Drop the payload-size
split in `flush()`: one chunk per request already bounds a payload to a
single trace. The `export()` soft limit stays, since it bounds how much
is buffered rather than how much one request carries.

Keep the reclamation work from the parent commit, which is unaffected:
`setAgentUrl` still frees the state it replaces, and rebuilds are still
amortized over dropped spans.

Batching remains possible and is worth revisiting, but it needs the test
agent to become chunk-oriented rather than request-oriented, which is a
change to shared infrastructure that 80 spec files depend on.

Author: bengl

packages/dd-trace/src/exporters/native/index.js

@@ -194,20 +194,21 @@ class NativeExporter {
    * Mirror the deleted OtlpHttpTraceExporter's per-request telemetry counters.
    * No-op on the agent path.
    *
-   * `attempts`/`successes` measure export *pushes*, so they are incremented once
-   * per HTTP request. The deleted JS exporter's `export()` was invoked per trace
-   * chunk and issued one request each, so its `spans:` tag was that chunk's span
-   * count; `flushSpansGrouped` coalesces the whole flush into one multi-chunk
-   * request, so the equivalent tag is the payload's total span count.
+   * The deleted JS exporter's `export()` was invoked per trace chunk and issued
+   * one HTTP request each, tagged with that chunk's span count.
+   * `flushSpansGrouped` also sends one request per chunk, so emit once per group
+   * with that group's span count: a single per-flush increment would under-count
+   * attempts by the number of chunks and would turn `spans:` into an unbounded
+   * whole-flush total (the telemetry namespace map never evicts keys).
    *
    * @param {string} metric `otel.traces_export_attempts` or `..._successes`
    * @param {Array<{spanIds: Uint8Array[]}>} groups Groups in this flush
    */
   #recordOtlpTelemetry (metric, groups) {
     if (this.#otlpTelemetryTags === null) return
-    let spans = 0
-    for (const group of groups) spans += group.spanIds.length
-    tracerMetrics.count(metric, [...this.#otlpTelemetryTags, `spans:${spans}`]).inc(1)
+    for (const group of groups) {
+      tracerMetrics.count(metric, [...this.#otlpTelemetryTags, `spans:${group.spanIds.length}`]).inc(1)
+    }
   }
 
   /**
@@ -560,36 +561,12 @@ class NativeExporter {
       return
     }
 
-    // One flush is one HTTP request, so cap what a single payload carries. The
-    // soft-limit trigger in `export()` bounds how much is buffered while idle, but
-    // it cannot bound this: sends are serialized, so while one is in flight
-    // `flush()` only records `#flushRequested` and `_pendingSpanChunks` keeps
-    // growing for the whole round trip. Take whole chunks up to the limit and
-    // leave the rest for the send `#finishSend` will start immediately after.
-    let spanChunks
-    if (this._pendingSpans.length > SOFT_LIMIT_SPANS) {
-      let taken = 0
-      let i = 0
-      // Never split a chunk - chunk boundaries are the processor's trace
-      // boundaries. Always take at least one, even if it alone exceeds the limit.
-      while (i < this._pendingSpanChunks.length &&
-        (taken === 0 || taken + this._pendingSpanChunks[i].length <= SOFT_LIMIT_SPANS)) {
-        taken += this._pendingSpanChunks[i].length
-        i++
-      }
-      spanChunks = this._pendingSpanChunks.slice(0, i)
-      this._pendingSpanChunks = this._pendingSpanChunks.slice(i)
-      // `_pendingSpans` is the in-order concatenation of the chunks, so the
-      // remainder is exactly the tail past what this payload took.
-      this._pendingSpans = this._pendingSpans.slice(taken)
-      // Guarantee the remainder ships right after this send instead of waiting
-      // out another flushInterval.
-      this.#flushRequested = true
-    } else {
-      spanChunks = this._pendingSpanChunks
-      this._pendingSpans = []
-      this._pendingSpanChunks = []
-    }
+    // Each chunk becomes its own HTTP request (see flushSpansGrouped), so payload
+    // size is bounded by one trace and there is nothing to split here. The
+    // soft-limit trigger in `export()` still bounds how much is buffered.
+    const spanChunks = this._pendingSpanChunks
+    this._pendingSpans = []
+    this._pendingSpanChunks = []
 
     // Convert each SpanProcessor export call into one or more native chunks,
     // splitting only traces that happen to share one export call. Never group
@@ -598,11 +575,12 @@ class NativeExporter {
     // when flushInterval coalesces HTTP sends.
     const groups = this.#groupsFromSpanChunks(spanChunks, true)
 
-    // `flushSpansGrouped` stages every chunk synchronously and issues exactly one
-    // HTTP request for the whole flush, so `.requests`/`.responses` are counted
-    // once here - the same per-request scale as `.errors` and as the legacy
-    // AgentWriter's `_sendPayload`.
-    runtimeMetrics.increment(`${METRIC_PREFIX}.requests`, true)
+    // `flushSpansGrouped` sends one request per trace chunk, so count per chunk:
+    // a single per-flush increment reported 1/N of the real request volume and
+    // left `.requests` on a different scale from `.errors`, which is per-attempt.
+    for (let i = 0; i < groups.length; i++) {
+      runtimeMetrics.increment(`${METRIC_PREFIX}.requests`, true)
+    }
     this.#recordOtlpTelemetry('otel.traces_export_attempts', groups)
     // Self-guarded (`integrationsAlreadyRan`), so the repeat cost is one boolean.
     // Without this the on-by-default `INTEGRATIONS LOADED` startup line never
@@ -619,9 +597,8 @@ class NativeExporter {
       this.#firstFlushSent = true
       firstFlushChannel.publish()
     }
-    // One request carrying one chunk per trace: `prepareChunk` appends to a
-    // native chunk Vec and `sendPreparedChunk` drains all of it into a single
-    // multi-trace payload, which is the shape the legacy AgentWriter sent.
+    // One request per trace chunk, sequentially, preserving the legacy writer's
+    // one-trace-per-payload shape that `traces[0]` consumers rely on.
     let sendGrouped
     try {
       sendGrouped = this._nativeSpans.flushSpansGrouped(groups)
@@ -633,7 +610,9 @@ class NativeExporter {
     sendGrouped
       .then((response) => {
         this.#flushInFlight = false
-        runtimeMetrics.increment(`${METRIC_PREFIX}.responses`, true)
+        for (let i = 0; i < groups.length; i++) {
+          runtimeMetrics.increment(`${METRIC_PREFIX}.responses`, true)
+        }
         this.#recordOtlpTelemetry('otel.traces_export_successes', groups)
         // The agent's response carries per-service sampling rates. Feed them
         // back into the priority sampler so adaptive (agent-driven) sampling

packages/dd-trace/src/native/native_spans.js

@@ -1209,65 +1209,47 @@ class NativeSpansInterface {
    * local root. Passing many traces as one chunk would lump distinct trace_ids
    * together and stamp only the first — corrupting sampling/grouping under load.
    *
-   * `prepareChunk` appends to a native chunk Vec and `sendPreparedChunk` drains
-   * all of it into a single multi-trace request (libdatadog-nodejs #159), which
-   * is the same shape the legacy writer sends. So stage the whole flush, then
-   * send once: one HTTP request per flush carrying one chunk per trace.
-   *
-   * Staging synchronously also matters for correctness. `prepareChunk` is what
-   * drains the change buffer and removes spans from the WASM map, so with no
-   * await between groups nothing can finish into a half-staged flush, and the
-   * string table can be evicted exactly once at a provably drained point.
+   * One request per group, sequentially. The native `prepared_spans` Vec does
+   * accumulate (libdatadog-nodejs #159) and `sendPreparedChunk` would drain all of
+   * it into a single multi-trace payload, so batching the whole flush into one
+   * request is possible - but it changes what a consumer sees per payload, and
+   * `traces[0]` consumers (the test agent among them) rely on one trace per
+   * payload, which is also what the legacy AgentWriter produced at the flush
+   * intervals these paths run with. Sending per group also keeps a failed send
+   * from taking unrelated traces down with it.
    *
    * @param {Array<{spanIds: Uint8Array[], firstIsLocalRoot: boolean}>} groups
    * @returns {Promise<string>} The agent response body, or a no-op marker
    */
   flushSpansGrouped (groups) {
     // Drain all pending ops (creates, tags, per-trace sampling/trace tags) once
-    // up front so every chunk staged below sees a fully-applied span map.
+    // up front so every chunk prepared below sees a fully-applied span map.
     this.flushChangeQueue()
 
-    let staged = 0
-    try {
-      for (const group of groups) {
-        if (!group.spanIds?.length) continue
-        if (this.#prepareGroup(group)) staged++
-      }
-    } catch (e) {
-      // prepareChunk may throw partway through (`flush_chunk` errors on an
-      // absent span id), after consuming some of the change queue or growing
-      // WASM memory. Reset JS-side queue state and refresh views so the next
-      // caller starts from a known-good baseline. Groups staged before the
-      // throw stay staged and ship with the next flush - they are real spans we
-      // wanted to send, so delaying beats dropping them.
-      this.resetChangeQueue()
-      this.#checkDetach()
-      log.error('Error preparing spans to flush:', e)
-      return Promise.reject(e)
+    const pending = []
+    for (const group of groups) {
+      if (group.spanIds?.length) pending.push(group)
     }
 
-    // Safe here and only here: `prepareChunk` drained the change buffer and
-    // resolved every interned id into the staged spans, and nothing can have
-    // queued an op since (staging above is synchronous). `#evictStringTable(true)`
-    // also resets `_stringIdCounter`, so running it while ops are queued - e.g.
-    // from a `.finally()` after the async send - would re-issue live ids to
-    // different strings and silently mis-tag exported spans.
-    this.#evictStringTable(true)
-
-    if (staged === 0) return Promise.resolve('no spans to flush')
+    if (pending.length === 0) {
+      this.#evictStringTable(true)
+      return Promise.resolve('no spans to flush')
+    }
 
-    const send = this._state.sendPreparedChunk()
-    this.#sendInFlight = send
-    const clearSend = () => {
-      if (this.#sendInFlight === send) this.#sendInFlight = null
+    // Sequential: `sendPreparedChunk` also guards against async re-entrancy, and
+    // staging the next chunk while one is in flight would put it in the same
+    // payload as the current one.
+    let chain = Promise.resolve('no spans to flush')
+    for (let i = 0; i < pending.length; i++) {
+      const group = pending[i]
+      const isLast = i === pending.length - 1
+      chain = chain.then(() => this.#prepareAndSend(group, isLast))
     }
-    send.then(clearSend, clearSend)
 
-    return send
+    return chain
       .catch(e => {
         // A send failure is a *network* fault for the already-serialized chunks;
-        // `sendPreparedChunk` took them out of the native Vec before sending, so
-        // they are lost, which is expected on a transient agent outage.
+        // those are lost, which is expected on a transient agent outage.
         //
         // Crucially, do NOT resetChangeQueue() here. sendPreparedChunk is async,
         // so by the time this rejection lands, ops for *other* spans (including
@@ -1288,6 +1270,55 @@ class NativeSpansInterface {
       })
   }
 
+  /**
+   * Stage one trace's chunk and send it. Rejects if staging throws, after
+   * restoring JS-side queue state.
+   *
+   * @param {{spanIds: Uint8Array[], firstIsLocalRoot: boolean}} group
+   * @param {boolean} isLast Whether this is the final group of the flush
+   * @returns {Promise<string>}
+   */
+  #prepareAndSend (group, isLast) {
+    // Every group after the first runs in a `.then()` after an HTTP response, so
+    // spans that finished during that send have queued ops. `prepareChunk` calls
+    // `flush_change_buffer` internally, which zeroes the WASM header without
+    // touching our `_cqbIndex`/`_cqbCount` — drain through our own path first so
+    // the two stay in sync (same hazard `setMetaStruct` documents). A no-op when
+    // the queue is already empty.
+    this.flushChangeQueue()
+
+    let staged
+    try {
+      staged = this.#prepareGroup(group)
+    } catch (e) {
+      // prepareChunk may throw partway through, after consuming some of the
+      // change queue or growing WASM memory. Reset JS-side queue state and
+      // refresh views so the next caller starts from a known-good baseline.
+      this.resetChangeQueue()
+      this.#checkDetach()
+      log.error('Error preparing spans to flush:', e)
+      return Promise.reject(e)
+    }
+
+    // Evict only here, and only on the last group: `prepareChunk` has just
+    // drained the change buffer, so no queued op can still reference an id we
+    // are about to evict — and `#evictStringTable(true)` also resets
+    // `_stringIdCounter`, so running it while ops are queued (e.g. from a
+    // `.finally()` after the async send) would re-issue live ids to different
+    // strings and silently mis-tag exported spans.
+    if (isLast) this.#evictStringTable(true)
+
+    if (!staged) return Promise.resolve('no spans to flush')
+
+    const send = this._state.sendPreparedChunk()
+    this.#sendInFlight = send
+    const clearSend = () => {
+      if (this.#sendInFlight === send) this.#sendInFlight = null
+    }
+    send.then(clearSend, clearSend)
+    return send
+  }
+
   // Note: sample() is not available in the WASM pipeline module.
   // Sampling is handled by the JS-side priority sampler.
 }

packages/dd-trace/test/native/exporter.spec.js

@@ -192,14 +192,14 @@ describe('NativeExporter', () => {
       sinon.assert.calledOnce(logWarn)
     })
 
-    it('counts one export attempt and success per flush, tagged with the payload span total', async () => {
-      // `flushSpansGrouped` issues ONE request for the whole flush, so these
-      // mirror the deleted JS OTLP exporter's per-request counters: one increment
-      // each, tagged with every span in the payload - not one per trace chunk.
+    it('counts an export attempt and success per chunk, tagged with that chunk span count', async () => {
+      // `flushSpansGrouped` issues one HTTP request per trace chunk, so these
+      // mirror the deleted JS OTLP exporter exactly: its `export()` was invoked
+      // per chunk and emitted one increment tagged with that chunk's span count.
       exporter = new NativeExporter(config, prioritySampler, nativeSpans)
 
-      // Two traces of 2 and 3 spans: 5 spans in 2 groups, so a `spans:` tag built
-      // from the group count is distinguishable from the real span total.
+      // Two traces of 2 and 3 spans, so a `spans:` tag built from the whole-flush
+      // total is distinguishable from the per-chunk counts.
       const traceA = [createMockSpan(1n), createMockSpan(2n)]
       const traceB = [createMockSpan(3n), createMockSpan(4n), createMockSpan(5n)]
       for (const span of traceA) span.context()._trace = traceA[0].context()._trace
@@ -211,8 +211,10 @@ describe('NativeExporter', () => {
 
       assert.strictEqual(nativeSpans.flushSpansGrouped.getCall(0).args[0].length, 2)
       assert.deepStrictEqual(telemetryCounts, [
-        { metric: 'otel.traces_export_attempts', tags: ['protocol:http', 'encoding:json', 'spans:5'] },
-        { metric: 'otel.traces_export_successes', tags: ['protocol:http', 'encoding:json', 'spans:5'] },
+        { metric: 'otel.traces_export_attempts', tags: ['protocol:http', 'encoding:json', 'spans:2'] },
+        { metric: 'otel.traces_export_attempts', tags: ['protocol:http', 'encoding:json', 'spans:3'] },
+        { metric: 'otel.traces_export_successes', tags: ['protocol:http', 'encoding:json', 'spans:2'] },
+        { metric: 'otel.traces_export_successes', tags: ['protocol:http', 'encoding:json', 'spans:3'] },
       ])
     })
 
@@ -394,41 +396,6 @@ describe('NativeExporter', () => {
       assert.strictEqual(exporter._pendingSpans.length, 0)
     })
 
-    it('splits an oversized payload across sends instead of one unbounded request', async () => {
-      // Sends are serialized, so while one is in flight flush() only records
-      // #flushRequested and the pending queue keeps growing for the whole round
-      // trip - the export()-time trigger cannot bound the payload here.
-      let release
-      nativeSpans.flushSpansGrouped = sinon.stub().returns(new Promise(resolve => { release = resolve }))
-
-      // First flush takes the whole (small) batch and is now in flight.
-      exporter.export([createMockSpan(1n)])
-      exporter.flush()
-      sinon.assert.calledOnce(nativeSpans.flushSpansGrouped)
-
-      // 12_000 spans arrive as 12 chunks of 1000 while that send is in flight.
-      for (let c = 0; c < 12; c++) {
-        const chunk = []
-        for (let i = 0; i < 1000; i++) chunk.push(createMockSpan(BigInt(c * 1000 + i + 2)))
-        exporter.export(chunk)
-      }
-      assert.strictEqual(exporter._pendingSpans.length, 12_000)
-      sinon.assert.calledOnce(nativeSpans.flushSpansGrouped)
-
-      // The 12_000 backlog must not go out as one request: the next send carries
-      // 10 whole chunks (10_000 spans) and the 2_000-span remainder follows in its
-      // own request, without waiting out another flushInterval.
-      nativeSpans.flushSpansGrouped = sinon.stub().resolves('OK')
-      release('OK')
-      await clock.tickAsync(0)
-
-      const sizes = nativeSpans.flushSpansGrouped.getCalls()
-        .map(call => call.args[0].reduce((total, group) => total + group.spanIds.length, 0))
-      assert.deepStrictEqual(sizes, [10_000, 2000])
-      assert.strictEqual(exporter._pendingSpans.length, 0)
-      assert.strictEqual(exporter._pendingSpanChunks.length, 0)
-    })
-
     it('resets native state immediately when explicitly requested while idle', () => {
       exporter._resetNativeStateWhenIdle()
 
@@ -958,10 +925,10 @@ describe('NativeExporter', () => {
   describe('health metrics', () => {
     const P = 'datadog.tracer.node.exporter.agent'
 
-    it('increments request + response counters once per flush, not once per trace', async () => {
-      // Two traces coalesce into two chunks but ONE request, so these counters
-      // must fire once - the same per-request scale as `.errors`. Counting per
-      // chunk multiplies every native user's request rate by traces-per-flush.
+    it('increments request + response counters once per trace chunk', async () => {
+      // `flushSpansGrouped` issues one HTTP request per chunk, so these counters
+      // must scale with chunks to stay on the same per-attempt footing as
+      // `.errors`. A single per-flush increment reports 1/N of the real volume.
       exporter = new NativeExporter(config, prioritySampler, nativeSpans)
       exporter.export([createMockSpan(1n)])
       exporter.export([createMockSpan(2n)])
@@ -970,10 +937,9 @@ describe('NativeExporter', () => {
 
       const groups = nativeSpans.flushSpansGrouped.getCall(0).args[0]
       assert.strictEqual(groups.length, 2)
-      sinon.assert.calledOnce(nativeSpans.flushSpansGrouped)
       const counted = metric => metricsIncrement.args.filter(([name]) => name === metric).length
-      assert.strictEqual(counted(`${P}.requests`), 1)
-      assert.strictEqual(counted(`${P}.responses`), 1)
+      assert.strictEqual(counted(`${P}.requests`), 2)
+      assert.strictEqual(counted(`${P}.responses`), 2)
     })
 
     it('increments error counters (name + code) on a failed flush', async () => {