⚗✨ [RUM-253] use deflate encoders for RUM batches

This commit is not pretty. To avoid having inflated packages while
dogfooding, we pass some functions from rum to rum-core when creating
the RUM public API.

In the future, those functions will be moved to the core package, and
rum-core will be able to import them directly.

packages/core/src/domain/deflate/types.ts

@@ -50,4 +50,6 @@ export type DeflateEncoder = Encoder<Uint8Array>
 
 export const enum DeflateEncoderStreamId {
   REPLAY = 1,
+  RUM = 2,
+  RUM_REPLICA = 3,
 }

packages/core/src/tools/experimentalFeatures.ts

@@ -20,6 +20,7 @@ export enum ExperimentalFeature {
   SCROLLMAP = 'scrollmap',
   INTERACTION_TO_NEXT_PAINT = 'interaction_to_next_paint',
   DISABLE_REPLAY_INLINE_CSS = 'disable_replay_inline_css',
+  COMPRESS_BATCH = 'compress_batch',
 }
 
 const enabledExperimentalFeatures: Set<ExperimentalFeature> = new Set()

packages/rum-core/src/boot/rumPublicApi.spec.ts

@@ -1,5 +1,13 @@
-import type { RelativeTime, TimeStamp, Context } from '@datadog/browser-core'
-import { ONE_SECOND, getTimeStamp, display, DefaultPrivacyLevel } from '@datadog/browser-core'
+import type { RelativeTime, TimeStamp, Context, DeflateWorker } from '@datadog/browser-core'
+import {
+  ONE_SECOND,
+  getTimeStamp,
+  display,
+  DefaultPrivacyLevel,
+  resetExperimentalFeatures,
+  ExperimentalFeature,
+  noop,
+} from '@datadog/browser-core'
 import {
   initEventBridgeStub,
   deleteEventBridgeStub,
@@ -149,6 +157,69 @@ describe('rum public api', () => {
         expect(startRumSpy).toHaveBeenCalled()
       })
     })
+
+    describe('deflate worker', () => {
+      let rumPublicApi: RumPublicApi
+      let startDeflateWorkerSpy: jasmine.Spy
+
+      beforeEach(() => {
+        startDeflateWorkerSpy = jasmine.createSpy().and.returnValue({} as DeflateWorker)
+
+        rumPublicApi = makeRumPublicApi(startRumSpy, noopRecorderApi, {
+          startDeflateWorker: startDeflateWorkerSpy,
+          createDeflateEncoder: noop as any,
+        })
+      })
+
+      afterEach(() => {
+        resetExperimentalFeatures()
+        deleteEventBridgeStub()
+      })
+
+      describe('without the COMPRESS_BATCH experimental flag', () => {
+        it('does not create a deflate worker', () => {
+          rumPublicApi.init(DEFAULT_INIT_CONFIGURATION)
+
+          expect(startDeflateWorkerSpy).not.toHaveBeenCalled()
+          expect(startRumSpy.calls.mostRecent().args[6]).toBeUndefined()
+        })
+      })
+
+      describe('with the COMPRESS_BATCH experimental flag', () => {
+        it('creates a deflate worker instance', () => {
+          rumPublicApi.init({
+            ...DEFAULT_INIT_CONFIGURATION,
+            enableExperimentalFeatures: [ExperimentalFeature.COMPRESS_BATCH],
+          })
+
+          expect(startDeflateWorkerSpy).toHaveBeenCalledTimes(1)
+          expect(startRumSpy.calls.mostRecent().args[6]).toEqual(jasmine.any(Function))
+        })
+
+        it('aborts the initialization if it fails to create a deflate worker', () => {
+          startDeflateWorkerSpy.and.returnValue(undefined)
+
+          rumPublicApi.init({
+            ...DEFAULT_INIT_CONFIGURATION,
+            enableExperimentalFeatures: [ExperimentalFeature.COMPRESS_BATCH],
+          })
+
+          expect(startRumSpy).not.toHaveBeenCalled()
+        })
+
+        it('if message bridge is present, does not create a deflate worker instance', () => {
+          initEventBridgeStub()
+
+          rumPublicApi.init({
+            ...DEFAULT_INIT_CONFIGURATION,
+            enableExperimentalFeatures: [ExperimentalFeature.COMPRESS_BATCH],
+          })
+
+          expect(startDeflateWorkerSpy).not.toHaveBeenCalled()
+          expect(startRumSpy).toHaveBeenCalledTimes(1)
+        })
+      })
+    })
   })
 
   describe('getInternalContext', () => {

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

@@ -1,4 +1,14 @@
-import type { Context, InitConfiguration, TimeStamp, RelativeTime, User, Observable } from '@datadog/browser-core'
+import type {
+  Context,
+  InitConfiguration,
+  TimeStamp,
+  RelativeTime,
+  User,
+  Observable,
+  DeflateWorker,
+  DeflateEncoderStreamId,
+  DeflateEncoder,
+} from '@datadog/browser-core'
 import {
   noop,
   CustomerDataType,
@@ -18,6 +28,8 @@ import {
   checkUser,
   sanitizeUser,
   sanitize,
+  isExperimentalFeatureEnabled,
+  ExperimentalFeature,
 } from '@datadog/browser-core'
 import type { LifeCycle } from '../domain/lifeCycle'
 import type { ViewContexts } from '../domain/contexts/viewContexts'
@@ -57,12 +69,18 @@ export interface RecorderApi {
 }
 interface RumPublicApiOptions {
   ignoreInitIfSyntheticsWillInjectRum?: boolean
+  startDeflateWorker?: (configuration: RumConfiguration, source: string) => DeflateWorker | undefined
+  createDeflateEncoder?: (
+    configuration: RumConfiguration,
+    worker: DeflateWorker,
+    streamId: DeflateEncoderStreamId
+  ) => DeflateEncoder
 }
 
 export function makeRumPublicApi(
   startRumImpl: StartRum,
   recorderApi: RecorderApi,
-  { ignoreInitIfSyntheticsWillInjectRum = true }: RumPublicApiOptions = {}
+  { ignoreInitIfSyntheticsWillInjectRum = true, startDeflateWorker, createDeflateEncoder }: RumPublicApiOptions = {}
 ) {
   let isAlreadyInitialized = false
 
@@ -98,6 +116,8 @@ export function makeRumPublicApi(
     bufferApiCalls.add(() => addFeatureFlagEvaluationStrategy(key, value))
   }
 
+  let deflateWorker: DeflateWorker | undefined
+
   function initRum(initConfiguration: RumInitConfiguration) {
     // This function should be available, regardless of initialization success.
     getInitConfigurationStrategy = () => deepClone<InitConfiguration>(initConfiguration)
@@ -129,6 +149,17 @@ export function makeRumPublicApi(
       return
     }
 
+    if (
+      isExperimentalFeatureEnabled(ExperimentalFeature.COMPRESS_BATCH) &&
+      !eventBridgeAvailable &&
+      startDeflateWorker
+    ) {
+      deflateWorker = startDeflateWorker(configuration, 'Datadog RUM')
+      if (!deflateWorker) {
+        return
+      }
+    }
+
     if (!configuration.trackViewsManually) {
       doStartRum(initConfiguration, configuration)
     } else {
@@ -158,7 +189,10 @@ export function makeRumPublicApi(
       recorderApi,
       globalContextManager,
       userContextManager,
-      initialViewOptions
+      initialViewOptions,
+      deflateWorker &&
+        createDeflateEncoder &&
+        ((streamId) => createDeflateEncoder(configuration, deflateWorker!, streamId))
     )
     getSessionReplayLinkStrategy = () =>
       recorderApi.getSessionReplayLink(configuration, startRumResults.session, startRumResults.viewContexts)

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

@@ -1,4 +1,11 @@
-import type { Observable, TelemetryEvent, RawError, ContextManager } from '@datadog/browser-core'
+import type {
+  Observable,
+  TelemetryEvent,
+  RawError,
+  ContextManager,
+  DeflateEncoderStreamId,
+  DeflateEncoder,
+} from '@datadog/browser-core'
 import {
   sendToExtension,
   createPageExitObservable,
@@ -45,7 +52,8 @@ export function startRum(
   recorderApi: RecorderApi,
   globalContextManager: ContextManager,
   userContextManager: ContextManager,
-  initialViewOptions?: ViewOptions
+  initialViewOptions?: ViewOptions,
+  createDeflateEncoder?: (streamId: DeflateEncoderStreamId) => DeflateEncoder
 ) {
   const lifeCycle = new LifeCycle()
 
@@ -86,7 +94,8 @@ export function startRum(
       telemetry.observable,
       reportError,
       pageExitObservable,
-      session.expireObservable
+      session.expireObservable,
+      createDeflateEncoder
     )
     startCustomerDataTelemetry(
       configuration,

packages/rum-core/src/transport/startRumBatch.ts

@@ -1,5 +1,17 @@
-import type { Context, TelemetryEvent, Observable, RawError, PageExitEvent } from '@datadog/browser-core'
-import { combine, isTelemetryReplicationAllowed, startBatchWithReplica } from '@datadog/browser-core'
+import type {
+  Context,
+  TelemetryEvent,
+  Observable,
+  RawError,
+  PageExitEvent,
+  DeflateEncoder,
+} from '@datadog/browser-core'
+import {
+  DeflateEncoderStreamId,
+  combine,
+  isTelemetryReplicationAllowed,
+  startBatchWithReplica,
+} from '@datadog/browser-core'
 import type { RumConfiguration } from '../domain/configuration'
 import type { LifeCycle } from '../domain/lifeCycle'
 import { LifeCycleEventType } from '../domain/lifeCycle'
@@ -12,18 +24,21 @@ export function startRumBatch(
   telemetryEventObservable: Observable<TelemetryEvent & Context>,
   reportError: (error: RawError) => void,
   pageExitObservable: Observable<PageExitEvent>,
-  sessionExpireObservable: Observable<void>
+  sessionExpireObservable: Observable<void>,
+  createDeflateEncoder?: (streamId: DeflateEncoderStreamId) => DeflateEncoder
 ) {
   const replica = configuration.replica
 
   const batch = startBatchWithReplica(
     configuration,
     {
       endpoint: configuration.rumEndpointBuilder,
+      encoder: createDeflateEncoder && createDeflateEncoder(DeflateEncoderStreamId.RUM),
     },
     replica && {
       endpoint: replica.rumEndpointBuilder,
       transformMessage: (message) => combine(message, { application: { id: replica.applicationId } }),
+      encoder: createDeflateEncoder && createDeflateEncoder(DeflateEncoderStreamId.RUM_REPLICA),
     },
     reportError,
     pageExitObservable,

packages/rum/src/domain/deflate/deflateWorker.ts

@@ -47,7 +47,7 @@ let state: DeflateWorkerState = { status: DeflateWorkerStatus.Nil }
 export function startDeflateWorker(
   configuration: RumConfiguration,
   source: string,
-  onInitializationFailure: () => void,
+  onInitializationFailure?: () => void,
   createDeflateWorkerImpl = createDeflateWorker
 ) {
   if (state.status === DeflateWorkerStatus.Nil) {
@@ -57,7 +57,9 @@ export function startDeflateWorker(
 
   switch (state.status) {
     case DeflateWorkerStatus.Loading:
-      state.initializationFailureCallbacks.push(onInitializationFailure)
+      if (onInitializationFailure) {
+        state.initializationFailureCallbacks.push(onInitializationFailure)
+      }
       return state.worker
     case DeflateWorkerStatus.Initialized:
       return state.worker

packages/rum/src/entries/main.ts

@@ -5,6 +5,7 @@ import { makeRumPublicApi, startRum } from '@datadog/browser-rum-core'
 
 import { startRecording } from '../boot/startRecording'
 import { makeRecorderApi } from '../boot/recorderApi'
+import { createDeflateEncoder, startDeflateWorker } from '../domain/deflate'
 
 export {
   CommonProperties,
@@ -30,7 +31,7 @@ export {
 export { DefaultPrivacyLevel } from '@datadog/browser-core'
 
 const recorderApi = makeRecorderApi(startRecording)
-export const datadogRum = makeRumPublicApi(startRum, recorderApi)
+export const datadogRum = makeRumPublicApi(startRum, recorderApi, { startDeflateWorker, createDeflateEncoder })
 
 interface BrowserWindow extends Window {
   DD_RUM?: RumPublicApi