⚗ [RUMF-823] monitor deflate worker

packages/core/src/domain/internalMonitoring.ts

@@ -132,7 +132,7 @@ export function addMonitoringMessage(message: string, context?: Context) {
   })
 }
 
-function addErrorToMonitoringBatch(e: unknown) {
+export function addErrorToMonitoringBatch(e: unknown) {
   addToMonitoringBatch({
     ...formatError(e),
     status: StatusType.error,

packages/core/src/index.ts

@@ -16,6 +16,7 @@ export {
   monitored,
   monitor,
   addMonitoringMessage,
+  addErrorToMonitoringBatch,
   setDebugMode,
 } from './domain/internalMonitoring'
 export { Observable } from './tools/observable'

packages/rum-recorder/src/domain/deflateSegmentWriter.ts

@@ -1,4 +1,4 @@
-import { addMonitoringMessage } from '@datadog/browser-core'
+import { addMonitoringMessage, addErrorToMonitoringBatch, monitor } from '@datadog/browser-core'
 import { SegmentMeta } from '../types'
 import { DeflateWorker } from './deflateWorker'
 import { SegmentWriter } from './segment'
@@ -12,28 +12,33 @@ export class DeflateSegmentWriter implements SegmentWriter {
     private onWrote: (size: number) => void,
     private onFlushed: (data: Uint8Array, meta: SegmentMeta) => void
   ) {
-    worker.addEventListener('message', ({ data }) => {
-      if ('result' in data) {
-        let pendingMeta = this.pendingMeta.shift()!
+    worker.addEventListener(
+      'message',
+      monitor(({ data }) => {
+        if ('error' in data) {
+          addErrorToMonitoringBatch(data.error)
+        } else if ('result' in data) {
+          let pendingMeta = this.pendingMeta.shift()!
 
-        // Messages should be received in the same order as they are sent, so the first
-        // 'pendingMeta' of the list should be the one corresponding to the handled message.
-        // But if something goes wrong in the worker and a response is lost, we need to avoid
-        // associating an incorrect meta to the flushed segment. Remove any pending meta with an id
-        // inferior to the one being waited for.
-        if (pendingMeta.id !== data.id) {
-          let lostCount = 0
-          while (pendingMeta.id !== data.id) {
-            pendingMeta = this.pendingMeta.shift()!
-            lostCount += 1
+          // Messages should be received in the same order as they are sent, so the first
+          // 'pendingMeta' of the list should be the one corresponding to the handled message.
+          // But if something goes wrong in the worker and a response is lost, we need to avoid
+          // associating an incorrect meta to the flushed segment. Remove any pending meta with an id
+          // inferior to the one being waited for.
+          if (pendingMeta.id !== data.id) {
+            let lostCount = 0
+            while (pendingMeta.id !== data.id) {
+              pendingMeta = this.pendingMeta.shift()!
+              lostCount += 1
+            }
+            addMonitoringMessage(`${lostCount} deflate worker responses have been lost`)
           }
-          addMonitoringMessage(`${lostCount} deflate worker responses have been lost`)
+          this.onFlushed(data.result, pendingMeta.meta)
+        } else {
+          this.onWrote(data.size)
         }
-        this.onFlushed(data.result, pendingMeta.meta)
-      } else {
-        this.onWrote(data.size)
-      }
-    })
+      })
+    )
   }
 
   write(data: string): void {

packages/rum-recorder/src/domain/deflateWorker.d.ts

@@ -30,3 +30,4 @@ export type DeflateWorkerResponse =
       id: number
       result: Uint8Array
     }
+  | { error: Error | string }

packages/rum-recorder/src/domain/deflateWorker.js

@@ -10,32 +10,52 @@ export function createDeflateWorker() {
 }
 
 function workerCodeFn() {
-  const { Deflate, constants } = makePakoDeflate()
-
-  let deflate = new Deflate()
-  self.addEventListener('message', (event) => {
-    const data = event.data
-    switch (data.action) {
-      case 'write':
-        deflate.push(data.data, constants.Z_SYNC_FLUSH)
-        self.postMessage({
-          id: data.id,
-          size: deflate.chunks.reduce((total, chunk) => total + chunk.length, 0),
-        })
-        break
-      case 'flush':
-        if (data.data) {
-          deflate.push(data.data, constants.Z_SYNC_FLUSH)
+  monitor(function () {
+    const { Deflate, constants } = makePakoDeflate()
+
+    let deflate = new Deflate()
+    self.addEventListener(
+      'message',
+      monitor((event) => {
+        const data = event.data
+        switch (data.action) {
+          case 'write':
+            deflate.push(data.data, constants.Z_SYNC_FLUSH)
+            self.postMessage({
+              id: data.id,
+              size: deflate.chunks.reduce((total, chunk) => total + chunk.length, 0),
+            })
+            break
+          case 'flush':
+            if (data.data) {
+              deflate.push(data.data, constants.Z_SYNC_FLUSH)
+            }
+            deflate.push('', constants.Z_FINISH)
+            self.postMessage({
+              id: data.id,
+              result: deflate.result,
+            })
+            deflate = new Deflate()
+            break
         }
-        deflate.push('', constants.Z_FINISH)
-        self.postMessage({
-          id: data.id,
-          result: deflate.result,
-        })
-        deflate = new Deflate()
-        break
+      })
+    )
+  })()
+
+  function monitor(fn) {
+    return function () {
+      try {
+        return fn.apply(this, arguments)
+      } catch (e) {
+        try {
+          self.postMessage({ 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 })
+        }
+      }
     }
-  })
+  }
 
   // https://github.com/nodeca/pako/blob/034669ba0f1a4c0590e45f7c2820128200f972b3/dist/pako_deflate.es5.js
   function makePakoDeflate() {