[skip ci] Merge branch into staging-14

Author: gh-worker-dd-devflow-36fce6[bot]

packages/core/src/transport/flushController.spec.ts

@@ -100,6 +100,19 @@ describe('flushController', () => {
   })
 
   describe('bytes limit', () => {
+    it('uses the page exit reason as flush reason for intermediate flushes during page exit', () => {
+      flushController.notifyBeforeAddMessage(SMALL_MESSAGE_BYTE_COUNT)
+      flushController.notifyAfterAddMessage()
+
+      flushController.preparePageExitFlushObservable.subscribe(() => {
+        flushController.notifyBeforeAddMessage(BYTES_LIMIT)
+      })
+
+      pageMayExitObservable.notify({ reason: 'before_unload' })
+
+      expect(flushSpy.calls.first().args[0].reason).toBe('before_unload')
+    })
+
     it('notifies when the bytes limit is reached after adding a message', () => {
       flushController.notifyBeforeAddMessage(BYTES_LIMIT)
       flushController.notifyAfterAddMessage()

packages/core/src/transport/flushController.ts

@@ -39,7 +39,17 @@ interface FlushControllerOptions {
  * but relies on invariants described in each method documentation to keep a coherent state.
  */
 export function createFlushController({ pageMayExitObservable, sessionExpireObservable }: FlushControllerOptions) {
-  const pageMayExitSubscription = pageMayExitObservable.subscribe((event) => flush(event.reason))
+  let forcedFlushReason: FlushReason | undefined
+  const preparePageExitFlushObservable = new Observable<PageExitReason>()
+  const pageMayExitSubscription = pageMayExitObservable.subscribe((event) => {
+    forcedFlushReason = event.reason
+    try {
+      preparePageExitFlushObservable.notify(event.reason)
+    } finally {
+      forcedFlushReason = undefined
+    }
+    flush(event.reason)
+  })
   const sessionExpireSubscription = sessionExpireObservable.subscribe(() => flush('session_expire'))
 
   const flushObservable = new Observable<FlushEvent>(() => () => {
@@ -85,6 +95,7 @@ export function createFlushController({ pageMayExitObservable, sessionExpireObse
 
   return {
     flushObservable,
+    preparePageExitFlushObservable,
     get messagesCount() {
       return currentMessagesCount
     },
@@ -100,7 +111,7 @@ export function createFlushController({ pageMayExitObservable, sessionExpireObse
      */
     notifyBeforeAddMessage(estimatedMessageBytesCount: number) {
       if (currentBytesCount + estimatedMessageBytesCount >= RECOMMENDED_REQUEST_BYTES_LIMIT) {
-        flush('bytes_limit')
+        flush(forcedFlushReason ?? 'bytes_limit')
       }
       // Consider the message to be added now rather than in `notifyAfterAddMessage`, because if no
       // message was added yet and `notifyAfterAddMessage` is called asynchronously, we still want
@@ -123,9 +134,9 @@ export function createFlushController({ pageMayExitObservable, sessionExpireObse
       currentBytesCount += messageBytesCountDiff
 
       if (currentMessagesCount >= MESSAGES_LIMIT) {
-        flush('messages_limit')
+        flush(forcedFlushReason ?? 'messages_limit')
       } else if (currentBytesCount >= RECOMMENDED_REQUEST_BYTES_LIMIT) {
-        flush('bytes_limit')
+        flush(forcedFlushReason ?? 'bytes_limit')
       }
     },
 

packages/core/test/emulate/mockFlushController.ts

@@ -1,10 +1,12 @@
 import { Observable } from '../../src/tools/observable'
+import type { PageExitReason } from '../../src/browser/pageMayExitObservable'
 import type { FlushEvent, FlushController, FlushReason } from '../../src/transport'
 
 export type MockFlushController = ReturnType<typeof createMockFlushController>
 
 export function createMockFlushController() {
   const flushObservable = new Observable<FlushEvent>()
+  const preparePageExitFlushObservable = new Observable<PageExitReason>()
   let currentMessagesCount = 0
   let currentBytesCount = 0
 
@@ -33,6 +35,7 @@ export function createMockFlushController() {
       return currentBytesCount
     },
     flushObservable,
+    preparePageExitFlushObservable,
     notifyFlush(reason: FlushReason = 'bytes_limit') {
       if (currentMessagesCount === 0) {
         throw new Error(

packages/rum-core/src/boot/startRum.ts

@@ -88,10 +88,6 @@ export function startRum(
   }
 
   const pageMayExitObservable = createPageMayExitObservable(configuration)
-  const pageMayExitSubscription = pageMayExitObservable.subscribe((event) => {
-    lifeCycle.notify(LifeCycleEventType.PAGE_MAY_EXIT, event)
-  })
-  cleanupTasks.push(() => pageMayExitSubscription.unsubscribe())
 
   const session = !canUseEventBridge()
     ? startRumSessionManager(configuration, lifeCycle, trackingConsentState)
@@ -106,10 +102,18 @@ export function startRum(
       session.expireObservable,
       createEncoder
     )
+    const preparePageExitSubscription = batch.flushController.preparePageExitFlushObservable.subscribe((reason) => {
+      lifeCycle.notify(LifeCycleEventType.PAGE_MAY_EXIT, { reason })
+    })
+    cleanupTasks.push(() => preparePageExitSubscription.unsubscribe())
     cleanupTasks.push(() => batch.stop())
     startCustomerDataTelemetry(telemetry, lifeCycle, batch.flushController.flushObservable)
   } else {
     startRumEventBridge(lifeCycle)
+    const pageMayExitSubscription = pageMayExitObservable.subscribe((event) => {
+      lifeCycle.notify(LifeCycleEventType.PAGE_MAY_EXIT, event)
+    })
+    cleanupTasks.push(() => pageMayExitSubscription.unsubscribe())
   }
 
   startTrackingConsentContext(hooks, trackingConsentState)

test/e2e/lib/framework/intakeProxyMiddleware.ts

@@ -11,6 +11,7 @@ import type { TelemetryEvent } from '@datadog/browser-core/src/domain/telemetry/
 interface BaseIntakeRequest {
   isBridge: boolean
   encoding: string | null
+  transport: string | null
 }
 
 export type LogsIntakeRequest = {
@@ -56,6 +57,7 @@ interface IntakeRequestInfos {
   isBridge: boolean
   intakeType: IntakeRequest['intakeType']
   encoding: string | null
+  transport: string | null
 }
 
 interface IntakeProxyOptions {
@@ -87,12 +89,14 @@ function computeIntakeRequestInfos(req: express.Request): IntakeRequestInfos {
   const { pathname, searchParams } = new URL(ddforward, 'https://example.org')
 
   const encoding = req.headers['content-encoding'] || searchParams.get('dd-evp-encoding')
+  const transport = searchParams.get('_dd.api')
 
   if (req.query.bridge === 'true') {
     const eventType = req.query.event_type
     return {
       isBridge: true,
       encoding,
+      transport,
       intakeType: eventType === 'log' ? 'logs' : eventType === 'record' ? 'replay' : 'rum',
     }
   }
@@ -108,6 +112,7 @@ function computeIntakeRequestInfos(req: express.Request): IntakeRequestInfos {
   return {
     isBridge: false,
     encoding,
+    transport,
     intakeType,
   }
 }

test/e2e/scenario/transport.scenario.ts

@@ -2,6 +2,50 @@ import { test, expect } from '@playwright/test'
 import { createTest } from '../lib/framework'
 
 test.describe('transport', () => {
+  test.describe('batch flushing on beforeunload', () => {
+    createTest('use sendBeacon for a batch flushed by bytes_limit during beforeunload')
+      // This test reproduces a bug where a batch near the size limit is flushed using fetch
+      // instead of sendBeacon when the beforeunload event fires.
+      //
+      // Scenario:
+      // 1. A batch is almost full (close to RECOMMENDED_REQUEST_BYTES_LIMIT = 16 KiB)
+      // 2. beforeunload fires and triggers a final view update
+      // 3. Adding the view update exceeds the limit → the batch is flushed due to bytes_limit
+      // 4. BUG: this flush uses fetch instead of sendBeacon, so it may be cancelled during unload
+      // 5. A new batch is created for the view update and flushed via sendBeacon
+      //
+      // Expected: the bytes_limit flush should use sendBeacon when triggered in the context of a
+      // page exit, so that it is not cancelled by the browser.
+      .withRum({ telemetrySampleRate: 0 })
+      .run(async ({ page, flushEvents, intakeRegistry }) => {
+        // Fill the batch close to the 16 KiB limit using a custom action. The action name is sized
+        // so that action event is almost at the limit 16KB limit → no flush yet.
+        await page.evaluate(() => {
+          window.DD_RUM!.addAction('x'.repeat(15000))
+        })
+
+        // Navigating away fires beforeunload, which triggers a final view update. Adding the view
+        // update (~2KB) to the near-full batch tips it over the limit and causes a bytes_limit
+        // flush, which the SDK issues via fetch — and that fetch gets cancelled on page unload.
+        await flushEvents()
+
+        // We expect two last RUM batches:
+        // 1. The near-full batch containing the large action, flushed due to bytes_limit
+        // 2. The final view update batch, flushed due to beforeunload
+        const [penultimateBatch, finalBatch] = intakeRegistry.rumRequests.slice(-2)
+
+        // The action event should be present in one of the batches
+        expect(
+          penultimateBatch.events.some((e) => e.type === 'action') || finalBatch.events.some((e) => e.type === 'action')
+        ).toBe(true)
+
+        // Both batches should use sendBeacon so they are not cancelled during page unload.
+        // With the bug, penultimateBatch.transport is 'fetch' instead of 'beacon'.
+        expect(penultimateBatch.transport).toBe('beacon')
+        expect(finalBatch.transport).toBe('beacon')
+      })
+  })
+
   test.describe('data compression', () => {
     createTest('send RUM data compressed')
       .withRum({