fix(native-spans): restore the parent commit's payload composition

Two commits ago I attributed a batch of plugin failures to sending the
whole flush as one multi-trace payload, and switched back to one request
per trace chunk. That diagnosis was wrong: the parent commit already
batched, and it was green. Sending per chunk then broke the Azure
Functions integration tests, which assert `payload.length === 2` and so
require both traces in a single payload.

The actual cause was the post-send drain. `#finishSend` had been gated
on "is something waiting on this send", which let chunks accumulate
across a send window instead of going out as soon as the previous send
resolved. That changes how many traces a payload carries, and
`traces[0]` consumers — the plugin test agent among them — depend on the
payload holding the trace they just produced. rhea, for instance, read
`traces[0][0]` and got `amqp.send` where it expected `amqp.receive`.

So: restore batching, and restore the unconditional drain along with it.
`#flushRequested` goes away with the gate; the oversized-payload split
now relies on `#finishSend` draining the remainder, which it does.

The gate was a throughput optimization (it stopped a 2s batch from
degenerating into one request per round trip when spans trickle in
during a send). Worth revisiting, but not at the cost of changing what a
payload contains.

Author: bengl

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

@@ -61,9 +61,6 @@ function formatSpansForDebug (spans) {
 class NativeExporter {
   #timer
   #flushInFlight = false
-  // An explicit flush() arrived while a send was in flight, so the send's
-  // completion must drain rather than wait for the next batching timer.
-  #flushRequested = false
   #firstFlushSent = false
   #flushCallbacks = []
   #activeSpans = 0
@@ -194,21 +191,20 @@ class NativeExporter {
    * Mirror the deleted OtlpHttpTraceExporter's per-request telemetry counters.
    * No-op on the agent path.
    *
-   * 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).
+   * `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.
    *
    * @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
-    for (const group of groups) {
-      tracerMetrics.count(metric, [...this.#otlpTelemetryTags, `spans:${group.spanIds.length}`]).inc(1)
-    }
+    let spans = 0
+    for (const group of groups) spans += group.spanIds.length
+    tracerMetrics.count(metric, [...this.#otlpTelemetryTags, `spans:${spans}`]).inc(1)
   }
 
   /**
@@ -465,33 +461,17 @@ class NativeExporter {
   }
 
   #finishSend () {
-    // Only drain eagerly when something is actually waiting on this send.
-    // Draining unconditionally defeated flushInterval entirely: any span that
-    // finished inside a send window triggered another send the moment the
-    // previous one resolved, turning a 2s batch into one request per round trip.
-    const waiting = this.#flushRequested ||
-      this.#flushCallbacks.length > 0 ||
-      this.#urlUpdateCallbacks.length > 0
-    this.#flushRequested = false
-
-    if (this._pendingSpanChunks.length > 0 && waiting) {
+    // Drain unconditionally. Gating this on "is something waiting" lets chunks
+    // accumulate across a send window, which changes how many traces a payload
+    // carries - and `traces[0]` consumers (the plugin test agent among them)
+    // depend on a payload holding the trace they just produced.
+    if (this._pendingSpanChunks.length > 0) {
       this.flush()
       return
     }
 
     this.#finishFlushCallbacks()
     this.#finishUrlUpdateCallbacks()
-
-    // An explicit flush() during the send cleared the batching timer; re-arm it
-    // so spans buffered in the meantime still go out on the normal interval.
-    const { flushInterval } = this._config
-    if (this._pendingSpanChunks.length > 0 && flushInterval > 0 && this.#timer === undefined) {
-      this.#timer = setTimeout(() => {
-        this.flush()
-        this.#timer = undefined
-      }, flushInterval)
-      this.#timer.unref?.()
-    }
   }
 
   #handleSendError (err) {
@@ -552,7 +532,6 @@ class NativeExporter {
     // on this to observe spans that finished while a previous payload was still
     // being sent.
     if (this.#flushInFlight) {
-      this.#flushRequested = true
       return
     }
 
@@ -561,12 +540,35 @@ class NativeExporter {
       return
     }
 
-    // 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 = []
+    // 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()` returns early and `_pendingSpanChunks` keeps growing for the whole
+    // round trip. Take whole chunks up to the limit and leave the rest, which
+    // `#finishSend` drains as soon as this send resolves.
+    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)
+      // The remainder ships from `#finishSend`, which drains whatever is still
+      // pending as soon as this send resolves.
+    } else {
+      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
@@ -575,12 +577,11 @@ class NativeExporter {
     // when flushInterval coalesces HTTP sends.
     const groups = this.#groupsFromSpanChunks(spanChunks, 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)
-    }
+    // `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)
     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
@@ -597,8 +598,9 @@ class NativeExporter {
       this.#firstFlushSent = true
       firstFlushChannel.publish()
     }
-    // One request per trace chunk, sequentially, preserving the legacy writer's
-    // one-trace-per-payload shape that `traces[0]` consumers rely on.
+    // 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.
     let sendGrouped
     try {
       sendGrouped = this._nativeSpans.flushSpansGrouped(groups)
@@ -610,9 +612,7 @@ class NativeExporter {
     sendGrouped
       .then((response) => {
         this.#flushInFlight = false
-        for (let i = 0; i < groups.length; i++) {
-          runtimeMetrics.increment(`${METRIC_PREFIX}.responses`, true)
-        }
+        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,47 +1209,65 @@ 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.
    *
-   * 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.
+   * `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.
    *
    * @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 prepared below sees a fully-applied span map.
+    // up front so every chunk staged below sees a fully-applied span map.
     this.flushChangeQueue()
 
-    const pending = []
-    for (const group of groups) {
-      if (group.spanIds?.length) pending.push(group)
+    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)
     }
 
-    if (pending.length === 0) {
-      this.#evictStringTable(true)
-      return Promise.resolve('no spans to flush')
-    }
+    // 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)
 
-    // 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))
+    if (staged === 0) 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 chain
+    return send
       .catch(e => {
         // A send failure is a *network* fault for the already-serialized chunks;
-        // those are lost, which is expected on a transient agent outage.
+        // `sendPreparedChunk` took them out of the native Vec before sending, so
+        // they 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
@@ -1270,55 +1288,6 @@ 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.
 }