fix(test-optimization): retry payload delivery failures

integration-tests/ci-visibility-intake.js

@@ -447,7 +447,10 @@ class FakeCiVisIntake extends FakeAgent {
       })
     })
 
-    app.post('/telemetry/proxy/api/v2/apmtelemetry', express.json(), (req, res) => {
+    app.post([
+      '/api/v2/apmtelemetry',
+      '/telemetry/proxy/api/v2/apmtelemetry',
+    ], express.json(), (req, res) => {
       res.status(200).send()
       if (req.body?.payload?.namespace !== 'civisibility') return
       this.emit('message', {

integration-tests/ci-visibility/playwright-tests-dynamic/dynamic-name-test.js

@@ -1,41 +1,38 @@
 'use strict'
 
-const crypto = require('crypto')
 const { test, expect } = require('@playwright/test')
 
-const uuid = crypto.randomBytes(16).toString('hex')
-  .replace(/(.{8})(.{4})(.{4})(.{4})(.{12})/, '$1-$2-$3-$4-$5')
-
 test.describe('dynamic name suite', () => {
-  test(`can do stuff at ${Date.now()}`, () => {
+  // Playwright loads this file during discovery and again in workers, so test names must match across processes.
+  test('can do stuff at 1750000000000', () => {
     expect(1 + 2).toBe(3)
   })
 
-  test(`connects to localhost:${3000 + Math.floor(Math.random() * 60000)}`, () => {
+  test('connects to localhost:54321', () => {
     expect(2 + 3).toBe(5)
   })
 
-  test(`user session ${uuid}`, () => {
+  test('user session 12345678-1234-1234-1234-123456789abc', () => {
     expect(3 + 4).toBe(7)
   })
 
-  test(`created at ${new Date().toISOString()}`, () => {
+  test('created at 2026-08-31T12:34:56.789Z', () => {
     expect(4 + 5).toBe(9)
   })
 
-  test(`event on ${new Date().toISOString().split('T')[0]}`, () => {
+  test('event on 2026-08-31', () => {
     expect(5 + 6).toBe(11)
   })
 
-  test(`probability ${Math.random()}`, () => {
+  test('probability 0.1234567890', () => {
     expect(6 + 7).toBe(13)
   })
 
-  test(`server at 127.0.0.1:${3000 + Math.floor(Math.random() * 60000)}`, () => {
+  test('server at 127.0.0.1:54322', () => {
     expect(7 + 8).toBe(15)
   })
 
-  test(`bound to 0.0.0.0:${3000 + Math.floor(Math.random() * 60000)}`, () => {
+  test('bound to 0.0.0.0:54323', () => {
     expect(8 + 9).toBe(17)
   })
 })

integration-tests/playwright/playwright-active-test-span.spec.js

@@ -173,9 +173,10 @@ versions.forEach((version) => {
     contextNewVersions('correlation between tests and RUM sessions', () => {
       const getTestAssertions = (receiver, { isRedirecting }) =>
         receiver
-          .gatherPayloadsMaxTimeout(({ url }) => url === '/api/v2/citestcycle', (payloads) => {
+          .gatherPayloadsMaxTimeout(({ url }) => url.endsWith('/api/v2/citestcycle'), (payloads) => {
             const events = payloads.flatMap(({ payload }) => payload.events)
             const tests = events.filter(event => event.type === 'test').map(event => event.content)
+            assert.ok(tests.length > 0, 'test events should be sent')
             tests.forEach(test => {
               if (isRedirecting) {
                 // can't do assertions because playwright has been redirected
@@ -197,7 +198,7 @@ versions.forEach((version) => {
             })
           })
 
-      const runRumTest = async (receiver, { isRedirecting }, extraEnvVars) => {
+      const runRumTest = async (receiver, { isRedirecting }, extraEnvVars, onTelemetryPayloads) => {
         const testAssertionsPromise = getTestAssertions(receiver, { isRedirecting })
         let proc
         try {
@@ -214,7 +215,16 @@ versions.forEach((version) => {
             }
           )
 
-          const [[exitCode]] = await Promise.all([once(proc, 'exit'), testAssertionsPromise])
+          const assertions = [once(proc, 'exit'), testAssertionsPromise]
+          if (onTelemetryPayloads) {
+            assertions.push(receiver.gatherPayloadsUntilChildExit(
+              proc,
+              ({ url }) => url.endsWith('/api/v2/apmtelemetry'),
+              onTelemetryPayloads
+            ))
+          }
+
+          const [[exitCode]] = await Promise.all(assertions)
 
           assert.strictEqual(exitCode, isRedirecting ? 1 : 0)
         } finally {
@@ -241,14 +251,25 @@ versions.forEach((version) => {
       })
 
       it('sends telemetry for RUM browser tests when telemetry is enabled', async (receiver) => {
-        const telemetryPromise = receiver
-          .gatherPayloadsMaxTimeout(({ url }) => url.endsWith('/api/v2/apmtelemetry'), (payloads) => {
+        await runRumTest(
+          receiver,
+          { isRedirecting: false },
+          {
+            ...getCiVisEvpProxyConfig(receiver.port),
+            DD_INSTRUMENTATION_TELEMETRY_ENABLED: 'true',
+          },
+          (payloads) => {
             const telemetryEvents = payloads.flatMap(({ payload }) => payload.payload.series)
-
-            const testSessionMetric = telemetryEvents.find(
-              ({ metric }) => metric === 'test_session'
+            const testSessionPayload = payloads.find(({ payload }) => {
+              return payload.payload.series.some(({ metric }) => metric === 'test_session')
+            })
+            assert.ok(
+              testSessionPayload,
+              `test_session telemetry metric should be sent. Got: ${inspect(
+                telemetryEvents.map(({ metric, tags }) => ({ metric, tags }))
+              )}`
             )
-            assert.ok(testSessionMetric, 'test_session telemetry metric should be sent')
+            assert.strictEqual(testSessionPayload.url, '/telemetry/proxy/api/v2/apmtelemetry')
 
             const eventFinishedTestEvents = telemetryEvents
               .filter(({ metric, tags }) => metric === 'event_finished' && tags.includes('event_type:test'))
@@ -258,19 +279,8 @@ versions.forEach((version) => {
               assert.ok(tags.includes('is_rum'), `Got: ${inspect(tags)}`)
               assert.ok(tags.includes('test_framework:playwright'), `Got: ${inspect(tags)}`)
             })
-          })
-
-        await Promise.all([
-          runRumTest(
-            receiver,
-            { isRedirecting: false },
-            {
-              ...getCiVisEvpProxyConfig(receiver.port),
-              DD_INSTRUMENTATION_TELEMETRY_ENABLED: 'true',
-            }
-          ),
-          telemetryPromise,
-        ])
+          }
+        )
       })
 
       it('do not crash when redirecting and RUM sessions are not active', async (receiver) => {

integration-tests/playwright/playwright-atr.spec.js

@@ -15,6 +15,9 @@ const {
 const { createWebAppServer } = require('../ci-visibility/web-app-server')
 const {
   TEST_STATUS,
+  TEST_NAME,
+  TEST_IS_NEW,
+  TEST_HAS_DYNAMIC_NAME,
   TEST_IS_RETRY,
   TEST_RETRY_REASON,
   TEST_HAS_FAILED_ALL_RETRIES,
@@ -248,6 +251,19 @@ versions.forEach((version) => {
         })
         receiver.setKnownTests({ playwright: {} })
 
+        const eventsPromise = receiver
+          .gatherPayloadsMaxTimeout(({ url }) => url.endsWith('/api/v2/citestcycle'), (payloads) => {
+            const events = payloads.flatMap(({ payload }) => payload.events)
+            const tests = events.filter(event => event.type === 'test').map(event => event.content)
+            const uniqueTests = new Map(tests.map(test => [test.meta[TEST_NAME], test]))
+
+            assert.strictEqual(uniqueTests.size, 8)
+            for (const test of uniqueTests.values()) {
+              assert.strictEqual(test.meta[TEST_IS_NEW], 'true')
+              assert.strictEqual(test.meta[TEST_HAS_DYNAMIC_NAME], 'true')
+            }
+          }, 30000)
+
         const proc = run(
           './node_modules/.bin/playwright test -c playwright.config.js',
           {
@@ -263,14 +279,9 @@ versions.forEach((version) => {
         proc.stdout?.on('data', chunk => { testOutput += chunk.toString() })
         proc.stderr?.on('data', chunk => { testOutput += chunk.toString() })
 
-        const eventsPromise = receiver
-          .gatherPayloadsUntilChildExit(proc, ({ url }) => url.endsWith('/api/v2/citestcycle'), () => {
-            assert.match(testOutput, /detected as new but their names contain dynamic data/)
-          })
-
         const [[exitCode]] = await Promise.all([once(proc, 'exit'), eventsPromise])
-        // Dynamic names differ between discovery and worker processes, so Playwright cannot find these tests.
-        assert.strictEqual(exitCode, 1, testOutput)
+        assert.strictEqual(exitCode, 0, testOutput)
+        assert.match(testOutput, /detected as new but their names contain dynamic data/)
       })
     })
   })

packages/datadog-plugin-cypress/src/cypress-plugin.js

@@ -1327,28 +1327,19 @@ class CypressPlugin {
     }
 
     return new Promise(resolve => {
-      const finishAfterRun = () => {
+      const exporter = this.tracer._tracer._exporter
+      const flushExporter = exporter?.flush?.bind(exporter) || exporter?._writer?.flush?.bind(exporter._writer)
+      let pendingFlushes = flushExporter ? 2 : 1
+      const onFlush = () => {
+        pendingFlushes--
+        if (pendingFlushes !== 0) return
+
         this._isInit = false
-        appClosingTelemetry()
         resolve(null)
       }
+      appClosingTelemetry(onFlush)
 
-      const exporter = this.tracer._tracer._exporter
-      if (!exporter) {
-        finishAfterRun()
-        return
-      }
-      if (exporter.flush) {
-        exporter.flush(() => {
-          finishAfterRun()
-        })
-      } else if (exporter._writer) {
-        exporter._writer.flush(() => {
-          finishAfterRun()
-        })
-      } else {
-        finishAfterRun()
-      }
+      flushExporter?.(onFlush)
     })
   }
 

packages/datadog-plugin-jest/src/index.js

@@ -172,12 +172,13 @@ class JestPlugin extends CiPlugin {
           autoInjected: !!this._tracerConfig.testOptimization.DD_CIVISIBILITY_AUTO_INSTRUMENTATION_PROVIDER,
         })
 
-        appClosingTelemetry()
-        this.tracer._exporter.flush(() => {
-          if (onDone) {
-            onDone()
-          }
-        })
+        let pendingFlushes = 2
+        const onFlush = () => {
+          pendingFlushes--
+          if (pendingFlushes === 0) onDone?.()
+        }
+        appClosingTelemetry(onFlush)
+        this.tracer._exporter.flush(onFlush)
       }
 
       if (this.pendingTestSuiteFinishes.size > 0) {

packages/datadog-plugin-playwright/src/index.js

@@ -182,8 +182,13 @@ class PlaywrightPlugin extends CiPlugin {
           provider: this.ciProviderName,
           autoInjected: !!this._tracerConfig.testOptimization.DD_CIVISIBILITY_AUTO_INSTRUMENTATION_PROVIDER,
         })
-        appClosingTelemetry()
-        this.tracer._exporter.flush(onDone)
+        let pendingFlushes = 2
+        const onFlush = () => {
+          pendingFlushes--
+          if (pendingFlushes === 0) onDone()
+        }
+        appClosingTelemetry(onFlush)
+        this.tracer._exporter.flush(onFlush)
         this.numFailedTests = 0
         this.numFailedSuites = 0
         this.finishSession = undefined

packages/dd-trace/src/ci-visibility/exporters/request.js

@@ -180,6 +180,7 @@ function requestBuffered (data, options, callback, reservedPayloadSize) {
   const timeout = options.timeout || 2000
   const payloadSize = reservedPayloadSize ?? getPayloadSize(data)
   let retryTimer
+  let attemptTimer
   let attemptController
   let settled = false
   let lastError
@@ -197,6 +198,7 @@ function requestBuffered (data, options, callback, reservedPayloadSize) {
     if (settled) return
     settled = true
     clearTimeout(retryTimer)
+    clearTimeout(attemptTimer)
     signal?.removeEventListener('abort', onAbort)
     bufferedBytes -= payloadSize
     callback(error, result, statusCode, headers)
@@ -249,7 +251,6 @@ function requestBuffered (data, options, callback, reservedPayloadSize) {
     }
     waitingForBackpressure = false
 
-    const attemptDeadline = deadline
     const attemptOptions = {
       ...options,
       headers: options.headers ? { ...options.headers } : undefined,
@@ -258,26 +259,44 @@ function requestBuffered (data, options, callback, reservedPayloadSize) {
     if (deadline !== undefined) attemptOptions.timeout = Math.max(1, Math.min(timeout, remaining))
 
     const controller = new AbortController()
+    const attemptTimeout = attemptOptions.timeout || timeout
+    let attemptTimedOut = false
+
     attemptController = controller
     attemptOptions.signal = controller.signal
+    attemptTimer = setTimeout(() => {
+      attemptTimedOut = true
+      controller.abort(createRequestTimeoutError())
+    }, attemptTimeout)
+    attemptTimer.unref?.()
 
     commonRequest(data, attemptOptions, (error, result, statusCode, headers) => {
+      clearTimeout(attemptTimer)
       if (attemptController === controller) attemptController = undefined
       if (settled) return
       if (!error) {
         complete(null, result, statusCode, headers)
         return
       }
 
-      lastError = error
+      const requestError = attemptTimedOut ? createRequestTimeoutError() : error
+      lastError = requestError
 
       const responseStatus = statusCode ?? error.status
-      const isRetriableError = isRetriableNetworkError(error) || isRetriableHttpStatusCode(responseStatus)
-      const deadlineExtended = options.deadline !== undefined &&
-        (attemptDeadline === undefined || options.deadline > attemptDeadline)
-      const reachedAttemptLimit = attemptIndex >= getMaxAttempts(attemptOptions) && !deadlineExtended
-      if (options.retry === false || !isRetriableError || reachedAttemptLimit) {
-        complete(error, result, statusCode, headers)
+      const isUnknownNetworkError = responseStatus === undefined && error.code === undefined
+      const isRetriableError =
+        attemptTimedOut || isRetriableNetworkError(error) || isUnknownNetworkError ||
+          isRetriableHttpStatusCode(responseStatus)
+      const retryUntilDeadline = options.deadline !== undefined && options.retryUntilDeadline !== false
+      const reachedAttemptLimit = !retryUntilDeadline && attemptIndex >= getMaxAttempts(attemptOptions)
+      const reachedUnknownNetworkAttemptLimit = isUnknownNetworkError && attemptIndex >= 2
+      if (
+        options.retry === false ||
+        !isRetriableError ||
+        reachedAttemptLimit ||
+        reachedUnknownNetworkAttemptLimit
+      ) {
+        complete(requestError, result, statusCode, headers)
         return
       }
 
@@ -288,11 +307,11 @@ function requestBuffered (data, options, callback, reservedPayloadSize) {
           if (options.deadline !== undefined) {
             const retryRemaining = options.deadline - Date.now()
             if (resetDelay >= retryRemaining) {
-              complete(error, result, statusCode, headers)
+              complete(requestError, result, statusCode, headers)
               return
             }
           } else if (resetDelay > RATE_LIMIT_MAX_WAIT_MS) {
-            complete(error, result, statusCode, headers)
+            complete(requestError, result, statusCode, headers)
             return
           }
           retryDelay = resetDelay
@@ -303,7 +322,7 @@ function requestBuffered (data, options, callback, reservedPayloadSize) {
       if (options.deadline !== undefined && retryDelay === undefined) {
         const retryRemaining = options.deadline - Date.now()
         if (retryRemaining <= 0) {
-          complete(error, result, statusCode, headers)
+          complete(requestError, result, statusCode, headers)
           return
         }
         const retryAttemptTimeout = timeout < retryRemaining ? timeout : Math.ceil(retryRemaining / 2)

packages/dd-trace/src/ci-visibility/requests/upload-test-screenshot.js

@@ -130,6 +130,7 @@ function uploadTestScreenshot (
     timeout: UPLOAD_TIMEOUT_MS,
     url,
     deadline,
+    retryUntilDeadline: false,
     signal,
   }
 

packages/dd-trace/src/exporters/common/request.js

@@ -212,10 +212,8 @@ function request (data, options, callback) {
             isRetriableNetworkError(error)) {
           settled = true
           finalize()
-          // Unref so a pending retry never keeps the host process alive past
-          // its natural exit point; long-running apps still retry because the
-          // event loop is held open by their own work.
-          setTimeout(attempt, getRetryDelay(options, attemptIndex), attemptIndex + 1).unref?.()
+          const retryTimer = setTimeout(attempt, getRetryDelay(options, attemptIndex), attemptIndex + 1)
+          if (!options.keepRetryTimerReferenced) retryTimer.unref?.()
         } else {
           complete(error)
         }