🐛 [RUMF-1060] fix failing worker detection in Firefox

packages/core/src/domain/internalMonitoring.ts

@@ -93,6 +93,20 @@ function startMonitoringBatch(configuration: Configuration) {
   }
 }
 
+export function startFakeInternalMonitoring() {
+  const messages: MonitoringMessage[] = []
+  assign(monitoringConfiguration, {
+    batch: {
+      add(message: MonitoringMessage) {
+        messages.push(message)
+      },
+    },
+    maxMessagesPerPage: Infinity,
+    sentMessageCount: 0,
+  })
+  return messages
+}
+
 export function resetInternalMonitoring() {
   monitoringConfiguration.batch = undefined
 }

packages/core/src/index.ts

@@ -31,6 +31,8 @@ export {
   callMonitored,
   addMonitoringMessage,
   addErrorToMonitoringBatch,
+  startFakeInternalMonitoring,
+  resetInternalMonitoring,
   setDebugMode,
 } from './domain/internalMonitoring'
 export { Observable, Subscription } from './tools/observable'

packages/rum/src/boot/recorderApi.spec.ts

@@ -1,4 +1,4 @@
-import { Configuration } from '@datadog/browser-core'
+import { Configuration, noop } from '@datadog/browser-core'
 import {
   RecorderApi,
   ParentContexts,
@@ -10,6 +10,7 @@ import { createNewEvent } from '@datadog/browser-core/test/specHelper'
 import { createRumSessionMock, RumSessionMock } from '../../../rum-core/test/mockRumSession'
 import { setup, TestSetupBuilder } from '../../../rum-core/test/specHelper'
 import { DeflateWorker } from '../domain/segmentCollection/deflateWorker'
+import { startDeflateWorker } from '../domain/segmentCollection/startDeflateWorker'
 import { makeRecorderApi, StartRecording } from './recorderApi'
 
 const DEFAULT_INIT_CONFIGURATION = { applicationId: 'xxx', clientToken: 'xxx' }
@@ -19,7 +20,23 @@ describe('makeRecorderApi', () => {
   let recorderApi: RecorderApi
   let startRecordingSpy: jasmine.Spy<StartRecording>
   let stopRecordingSpy: jasmine.Spy<() => void>
-  let getDeflateWorkerSingletonSpy: jasmine.Spy<() => DeflateWorker | undefined>
+  let startDeflateWorkerSpy: jasmine.Spy<typeof startDeflateWorker>
+  const FAKE_WORKER = {} as DeflateWorker
+
+  function startDeflateWorkerWith(worker?: DeflateWorker) {
+    if (!startDeflateWorkerSpy) {
+      startDeflateWorkerSpy = jasmine.createSpy<typeof startDeflateWorker>('startDeflateWorker')
+    }
+    startDeflateWorkerSpy.and.callFake((callback) => callback(worker))
+  }
+
+  function callLastRegisteredInitialisationCallback() {
+    startDeflateWorkerSpy.calls.mostRecent().args[0](FAKE_WORKER)
+  }
+
+  function stopDeflateWorker() {
+    startDeflateWorkerSpy.and.callFake(noop)
+  }
 
   let rumInit: (initConfiguration: RumInitConfiguration) => void
 
@@ -29,8 +46,9 @@ describe('makeRecorderApi', () => {
       startRecordingSpy = jasmine.createSpy('startRecording').and.callFake(() => ({
         stop: stopRecordingSpy,
       }))
-      getDeflateWorkerSingletonSpy = jasmine.createSpy('getDeflateWorkerSingleton').and.returnValue({})
-      recorderApi = makeRecorderApi(startRecordingSpy, getDeflateWorkerSingletonSpy)
+
+      startDeflateWorkerWith(FAKE_WORKER)
+      recorderApi = makeRecorderApi(startRecordingSpy, startDeflateWorkerSpy)
       rumInit = (initConfiguration) => {
         recorderApi.onRumStart(lifeCycle, initConfiguration, {} as Configuration, session, {} as ParentContexts)
       }
@@ -104,10 +122,23 @@ describe('makeRecorderApi', () => {
     it('do not start recording if worker fails to be instantiated', () => {
       setupBuilder.build()
       rumInit(DEFAULT_INIT_CONFIGURATION)
-      getDeflateWorkerSingletonSpy.and.returnValue(undefined)
+      startDeflateWorkerWith(undefined)
       recorderApi.start()
       expect(startRecordingSpy).not.toHaveBeenCalled()
     })
+
+    it('does not start recording multiple times if restarted before worker is initialized', () => {
+      setupBuilder.build()
+      rumInit(DEFAULT_INIT_CONFIGURATION)
+      stopDeflateWorker()
+      recorderApi.start()
+      recorderApi.stop()
+
+      callLastRegisteredInitialisationCallback()
+      recorderApi.start()
+      callLastRegisteredInitialisationCallback()
+      expect(startRecordingSpy).toHaveBeenCalledTimes(1)
+    })
   })
 
   describe('stopSessionReplayRecording()', () => {

packages/rum/src/boot/recorderApi.ts

@@ -8,7 +8,7 @@ import {
   RecorderApi,
 } from '@datadog/browser-rum-core'
 import { getReplayStats } from '../domain/replayStats'
-import { getDeflateWorkerSingleton } from '../domain/segmentCollection/deflateWorkerSingleton'
+import { startDeflateWorker } from '../domain/segmentCollection/startDeflateWorker'
 
 import { startRecording } from './startRecording'
 
@@ -42,7 +42,7 @@ type RecorderState =
 
 export function makeRecorderApi(
   startRecordingImpl: StartRecording,
-  getDeflateWorkerSingletonImpl = getDeflateWorkerSingleton
+  startDeflateWorkerImpl = startDeflateWorker
 ): RecorderApi {
   let state: RecorderState = {
     status: RecorderStatus.Stopped,
@@ -96,26 +96,31 @@ export function makeRecorderApi(
             return
           }
 
-          const worker = getDeflateWorkerSingletonImpl()
-          if (!worker) {
-            state = {
-              status: RecorderStatus.Stopped,
+          startDeflateWorkerImpl((worker) => {
+            if (state.status !== RecorderStatus.Starting) {
+              return
             }
-            return
-          }
 
-          const { stop: stopRecording } = startRecordingImpl(
-            lifeCycle,
-            initConfiguration.applicationId,
-            configuration,
-            session,
-            parentContexts,
-            worker
-          )
-          state = {
-            status: RecorderStatus.Started,
-            stopRecording,
-          }
+            if (!worker) {
+              state = {
+                status: RecorderStatus.Stopped,
+              }
+              return
+            }
+
+            const { stop: stopRecording } = startRecordingImpl(
+              lifeCycle,
+              initConfiguration.applicationId,
+              configuration,
+              session,
+              parentContexts,
+              worker
+            )
+            state = {
+              status: RecorderStatus.Started,
+              stopRecording,
+            }
+          })
         })
       }
 

packages/rum/src/boot/startRecording.spec.ts

@@ -9,7 +9,7 @@ import { collectAsyncCalls } from '../../test/utils'
 import { setMaxSegmentSize } from '../domain/segmentCollection/segmentCollection'
 
 import { Segment, RecordType } from '../types'
-import { getDeflateWorkerSingleton } from '../domain/segmentCollection/deflateWorkerSingleton'
+import { doStartDeflateWorker } from '../domain/segmentCollection/startDeflateWorker'
 import { startRecording } from './startRecording'
 
 describe('startRecording', () => {
@@ -37,6 +37,12 @@ describe('startRecording', () => {
     textField = document.createElement('input')
     sandbox.appendChild(textField)
 
+    const requestSendSpy = spyOn(HttpRequest.prototype, 'send')
+    ;({
+      waitAsyncCalls: waitRequestSendCalls,
+      expectNoExtraAsyncCall: expectNoExtraRequestSendCalls,
+    } = collectAsyncCalls(requestSendSpy))
+
     setupBuilder = setup()
       .withParentContexts({
         findView() {
@@ -61,17 +67,11 @@ describe('startRecording', () => {
           configuration,
           session,
           parentContexts,
-          getDeflateWorkerSingleton()!
+          doStartDeflateWorker()!
         )
         stopRecording = recording ? recording.stop : noop
         return { stop: stopRecording }
       })
-
-    const requestSendSpy = spyOn(HttpRequest.prototype, 'send')
-    ;({
-      waitAsyncCalls: waitRequestSendCalls,
-      expectNoExtraAsyncCall: expectNoExtraRequestSendCalls,
-    } = collectAsyncCalls(requestSendSpy))
   })
 
   afterEach(() => {

packages/rum/src/domain/segmentCollection/deflateWorker.d.ts

@@ -2,35 +2,60 @@ export function createDeflateWorker(): DeflateWorker
 
 export interface DeflateWorker {
   addEventListener(name: 'message', listener: DeflateWorkerListener): void
+  addEventListener(name: 'error', listener: (error: ErrorEvent) => void): void
   removeEventListener(name: 'message', listener: DeflateWorkerListener): void
+  removeEventListener(name: 'error', listener: (error: ErrorEvent) => void): void
+
   postMessage(message: DeflateWorkerAction): void
   terminate(): void
 }
 
 export type DeflateWorkerListener = (event: { data: DeflateWorkerResponse }) => void
 
 export type DeflateWorkerAction =
+  // Action to send when creating the worker to check if the communication is working correctly.
+  // The worker should respond with a 'initialized' response.
+  | {
+      action: 'init'
+    }
+  // Action to send when writing some unfinished data. The worker will respond with a 'wrote'
+  // response, with the same id and measurements of the wrote data size.
   | {
-      id: number
       action: 'write'
+      id: number
       data: string
     }
+  // Action to send when finishing to write some data. The worker will respond with a 'flushed'
+  // response, with the same id, measurements of the wrote data size and the complete deflate
+  // data.
   | {
-      id: number
       action: 'flush'
+      id: number
       data?: string
     }
 
 export type DeflateWorkerResponse =
+  // Response to 'init' action
+  | {
+      type: 'initialized'
+    }
+  // Response to 'write' action
   | {
+      type: 'wrote'
       id: number
       compressedSize: number
       additionalRawSize: number
     }
+  // Response to 'flush' action
   | {
+      type: 'flushed'
       id: number
       result: Uint8Array
       additionalRawSize: number
       rawSize: number
     }
-  | { error: Error | string }
+  // Could happen at any time when something goes wrong in the worker
+  | {
+      type: 'errored'
+      error: Error | string
+    }

packages/rum/src/domain/segmentCollection/deflateWorker.js

@@ -20,9 +20,15 @@ function workerCodeFn() {
       monitor((event) => {
         const data = event.data
         switch (data.action) {
+          case 'init':
+            self.postMessage({
+              type: 'initialized',
+            })
+            break
           case 'write':
             const additionalRawSize = pushData(data.data)
             self.postMessage({
+              type: 'wrote',
               id: data.id,
               compressedSize: deflate.chunks.reduce((total, chunk) => total + chunk.length, 0),
               additionalRawSize,
@@ -32,6 +38,7 @@ function workerCodeFn() {
             const additionalRawSize = data.data ? pushData(data.data) : 0
             deflate.push('', constants.Z_FINISH)
             self.postMessage({
+              type: 'flushed',
               id: data.id,
               result: deflate.result,
               additionalRawSize,
@@ -58,10 +65,16 @@ function workerCodeFn() {
         return fn.apply(this, arguments)
       } catch (e) {
         try {
-          self.postMessage({ error: e })
+          self.postMessage({
+            type: 'errored',
+            error: e,
+          })
         } catch (_) {
           // DATA_CLONE_ERR, cf https://developer.mozilla.org/en-US/docs/Web/API/Web_Workers_API/Structured_clone_algorithm
-          self.postMessage({ error: '' + e })
+          self.postMessage({
+            type: 'errored',
+            error: '' + e,
+          })
         }
       }
     }

packages/rum/src/domain/segmentCollection/deflateWorker.spec.ts

@@ -11,9 +11,9 @@ describe('deflateWorker', () => {
     const deflateWorker = createDeflateWorker()
     listen(deflateWorker, 3, (events) => {
       expect(events).toEqual([
-        { id: 0, compressedSize: 11, additionalRawSize: 3 },
-        { id: 1, compressedSize: 20, additionalRawSize: 3 },
-        { id: 2, compressedSize: 29, additionalRawSize: 3 },
+        { type: 'wrote', id: 0, compressedSize: 11, additionalRawSize: 3 },
+        { type: 'wrote', id: 1, compressedSize: 20, additionalRawSize: 3 },
+        { type: 'wrote', id: 2, compressedSize: 29, additionalRawSize: 3 },
       ])
       done()
     })
@@ -26,8 +26,9 @@ describe('deflateWorker', () => {
     const deflateWorker = createDeflateWorker()
     listen(deflateWorker, 2, (events) => {
       expect(events).toEqual([
-        { id: 0, compressedSize: 11, additionalRawSize: 3 },
+        { type: 'wrote', id: 0, compressedSize: 11, additionalRawSize: 3 },
         {
+          type: 'flushed',
           id: 1,
           result: new Uint8Array([120, 156, 74, 203, 207, 7, 0, 0, 0, 255, 255, 3, 0, 2, 130, 1, 69]),
           additionalRawSize: 0,
@@ -45,6 +46,7 @@ describe('deflateWorker', () => {
     listen(deflateWorker, 1, (events) => {
       expect(events).toEqual([
         {
+          type: 'flushed',
           id: 0,
           result: new Uint8Array([120, 156, 74, 203, 207, 7, 0, 0, 0, 255, 255, 3, 0, 2, 130, 1, 69]),
           additionalRawSize: 3,
@@ -61,22 +63,26 @@ describe('deflateWorker', () => {
     listen(deflateWorker, 4, (events) => {
       expect(events).toEqual([
         {
+          type: 'wrote',
           id: 0,
           compressedSize: 11,
           additionalRawSize: 3,
         },
         {
+          type: 'flushed',
           id: 1,
           result: new Uint8Array([120, 156, 74, 203, 207, 7, 0, 0, 0, 255, 255, 3, 0, 2, 130, 1, 69]),
           additionalRawSize: 0,
           rawSize: 3,
         },
         {
+          type: 'wrote',
           id: 2,
           compressedSize: 11,
           additionalRawSize: 3,
         },
         {
+          type: 'flushed',
           id: 3,
           result: new Uint8Array([120, 156, 74, 74, 44, 2, 0, 0, 0, 255, 255, 3, 0, 2, 93, 1, 54]),
           additionalRawSize: 0,