🐛 [RUM-11439] Fix deflate encoder sending data twice in some cases (#3806)

When calling `finishSync()`, pending data was not cleared, which lead to
those problematic scenarios:

* Pending data can be send  multiple times:

```
encoder.write('data1')
encoder.finishSync().pendingData === 'data1'
encoder.finishSync().pendingData === 'data1' // duplicated data
```

* Pending data can be corrupted:

```
encoder.write('{"type":"view"}')
encoder.finishSync().pendingData === '{"type":"view"}'
encoder.write('{"type":"action"}')
encoder.finishSync().pendingData === '{"type":"view"}{"type":"action"}' // corrupted data, missing newline between events
```

* Compressed data can be corrupted:

```
encoder.write('data1')
encoder.finishSync()
encoder.write('data2')
encoder.finish(({ result }) => {
  // result contains compressed bytes corresponding to 'data1', which
  // might lack the zlib header.
})
```

By clearing any pending data/write action when calling `finishSync()`, we
ensure that data is taken into account only once.

Author: BenoitZugmeyer

packages/rum/src/domain/deflate/deflateEncoder.spec.ts

@@ -213,4 +213,44 @@ describe('createDeflateEncoder', () => {
       },
     ])
   })
+
+  it('do not notify data twice when calling finishSync() then finish()', () => {
+    const encoder = createDeflateEncoder(configuration, worker, DeflateEncoderStreamId.REPLAY)
+    const finishCallbackSpy = jasmine.createSpy<(result: EncoderResult<Uint8ArrayBuffer>) => void>()
+
+    encoder.write('foo')
+    encoder.finishSync()
+
+    encoder.write('bar')
+    encoder.finish(finishCallbackSpy)
+
+    worker.processAllMessages()
+
+    expect(finishCallbackSpy).toHaveBeenCalledOnceWith({
+      rawBytesCount: 3,
+      output: new Uint8Array([...ENCODED_BAR, ...TRAILER]),
+      outputBytesCount: 4,
+      encoding: 'deflate',
+    })
+  })
+
+  it('do not notify data twice when calling finishSync() then finishSync()', () => {
+    const encoder = createDeflateEncoder(configuration, worker, DeflateEncoderStreamId.REPLAY)
+
+    encoder.write('foo')
+    encoder.finishSync()
+
+    encoder.write('bar')
+    expect(encoder.finishSync().pendingData).toBe('bar')
+  })
+
+  it('does not unsubscribe when there is no pending action', () => {
+    const encoder = createDeflateEncoder(configuration, worker, DeflateEncoderStreamId.REPLAY)
+
+    encoder.write('foo')
+    encoder.finishSync()
+    worker.processAllMessages()
+
+    expect(worker.messageListenersCount).toBe(1)
+  })
 })

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

@@ -18,6 +18,7 @@ export function createDeflateEncoder(
   let compressedData: Uint8ArrayBuffer[] = []
   let compressedDataTrailer: Uint8ArrayBuffer
 
+  let isEmpty = true
   let nextWriteActionId = 0
   const pendingWriteActions: Array<{
     writeCallback?: (additionalEncodedBytesCount: number) => void
@@ -35,20 +36,24 @@ export function createDeflateEncoder(
         return
       }
 
-      rawBytesCount += workerResponse.additionalBytesCount
-      compressedData.push(workerResponse.result)
-      compressedDataTrailer = workerResponse.trailer
-
-      const nextPendingAction = pendingWriteActions.shift()
-      if (nextPendingAction && nextPendingAction.id === workerResponse.id) {
-        if (nextPendingAction.writeCallback) {
-          nextPendingAction.writeCallback(workerResponse.result.byteLength)
-        } else if (nextPendingAction.finishCallback) {
-          nextPendingAction.finishCallback()
+      const nextPendingAction = pendingWriteActions[0]
+      if (nextPendingAction) {
+        if (nextPendingAction.id === workerResponse.id) {
+          pendingWriteActions.shift()
+
+          rawBytesCount += workerResponse.additionalBytesCount
+          compressedData.push(workerResponse.result)
+          compressedDataTrailer = workerResponse.trailer
+
+          if (nextPendingAction.writeCallback) {
+            nextPendingAction.writeCallback(workerResponse.result.byteLength)
+          } else if (nextPendingAction.finishCallback) {
+            nextPendingAction.finishCallback()
+          }
+        } else if (nextPendingAction.id < workerResponse.id) {
+          removeMessageListener()
+          addTelemetryDebug('Worker responses received out of order.')
         }
-      } else {
-        removeMessageListener()
-        addTelemetryDebug('Worker responses received out of order.')
       }
     }
   )
@@ -68,20 +73,20 @@ export function createDeflateEncoder(
   }
 
   function sendResetIfNeeded() {
-    if (nextWriteActionId > 0) {
+    if (!isEmpty) {
       worker.postMessage({
         action: 'reset',
         streamId,
       })
-      nextWriteActionId = 0
+      isEmpty = true
     }
   }
 
   return {
     isAsync: true,
 
     get isEmpty() {
-      return nextWriteActionId === 0
+      return isEmpty
     },
 
     write(data, callback) {
@@ -96,6 +101,7 @@ export function createDeflateEncoder(
         writeCallback: callback,
         data,
       })
+      isEmpty = false
       nextWriteActionId += 1
     },
 
@@ -117,16 +123,9 @@ export function createDeflateEncoder(
 
     finishSync() {
       sendResetIfNeeded()
-
-      const pendingData = pendingWriteActions
-        .map((pendingWriteAction) => {
-          // Make sure we do not call any write or finish callback
-          delete pendingWriteAction.writeCallback
-          delete pendingWriteAction.finishCallback
-          return pendingWriteAction.data
-        })
-        .join('')
-
+      const pendingData = pendingWriteActions.map((pendingWriteAction) => pendingWriteAction.data).join('')
+      // Ignore all pending write actions responses from the worker
+      pendingWriteActions.length = 0
       return { ...consumeResult(), pendingData }
     },