fix(otlp-exporter-base): cover compression with the export timeout

- Start the abort timer before compressing, so compression time counts
  toward the timeout
- Pass the timeout signal to the compression pipe, so a stalled
  compressor fails the export instead of hanging

Author: YangJonghun

experimental/packages/otlp-exporter-base/src/platform/browser/transport/fetch-transport.ts

@@ -77,15 +77,16 @@ class FetchTransport implements IExporterTransport {
     // would not survive.
     const suppressedContext = suppressTracing(context.active());
 
+    const abortController = new AbortController();
+    const timeout = setTimeout(() => abortController.abort(), timeoutMillis);
+
     // Compressed first so that the keepalive budget counts the bytes sent.
     const compressed =
       this._compression === 'none'
         ? undefined
-        : await this._compress(data, this._compression);
+        : await this._compress(data, this._compression, abortController.signal);
     const body = compressed ?? data;
 
-    const abortController = new AbortController();
-    const timeout = setTimeout(() => abortController.abort(), timeoutMillis);
     // Fetch API may be wrapped by an instrumentation like `@opentelemetry/instrumentation-fetch`.
     // In that case the instrumentation would create a new Span for this request,
     // creating an indirect endless loop Export -> Span -> Export.
@@ -212,14 +213,19 @@ class FetchTransport implements IExporterTransport {
 
   private async _compress(
     data: Uint8Array,
-    format: CompressionFormat
+    format: CompressionFormat,
+    signal: AbortSignal
   ): Promise<Blob | undefined> {
     try {
-      return await compressWithCompressionStream(data, format);
+      return await compressWithCompressionStream(data, format, signal);
     } catch (error) {
-      diag.warn(
-        `${format} compression failed, sending export uncompressed: ${error}`
-      );
+      // Not warned on timeout: `fetch` rejects on the aborted signal, so nothing
+      // is sent.
+      if (!signal.aborted) {
+        diag.warn(
+          `${format} compression failed, sending export uncompressed: ${error}`
+        );
+      }
       return undefined;
     }
   }
@@ -318,25 +324,26 @@ function isCompressionStreamFormatSupported(
 
 async function compressWithCompressionStream(
   data: Uint8Array,
-  format: CompressionFormat
+  format: CompressionFormat,
+  signal: AbortSignal
 ): Promise<Blob> {
-  // Read with a stream reader: `Response.arrayBuffer()` waits for a task in
-  // some engines, which never runs once the page unloads.
-  const reader = new ReadableStream<Uint8Array>({
+  // Read with `pipeTo`: `Response.arrayBuffer()` waits for a task in some
+  // engines, which never runs once the page unloads.
+  const chunks: Uint8Array[] = [];
+  await new ReadableStream<Uint8Array>({
     start(controller) {
       controller.enqueue(data);
       controller.close();
     },
   })
     .pipeThrough(new globalThis.CompressionStream(format))
-    .getReader();
-
-  const chunks: Uint8Array[] = [];
-  for (;;) {
-    const { done, value } = await reader.read();
-    if (done) {
-      return new Blob(chunks);
-    }
-    chunks.push(value);
-  }
+    .pipeTo(
+      new WritableStream({
+        write(chunk) {
+          chunks.push(chunk);
+        },
+      }),
+      { signal }
+    );
+  return new Blob(chunks);
 }

experimental/packages/otlp-exporter-base/test/browser/fetch-transport.test.ts

@@ -1334,6 +1334,65 @@ describe('FetchTransport', function () {
       sinon.assert.alwaysCalledWithMatch(warn, 'gzip compression failed');
     });
 
+    it('fails when compression takes longer than the timeout', async function () {
+      // arrange
+      // Barely compressible, so gzip tasks far longer than the timeout.
+      const slowPayload = new Uint8Array(4 * 1024 * 1024); // 4MB
+      for (let i = 0; i < slowPayload.length; i += 65536) {
+        crypto.getRandomValues(slowPayload.subarray(i, i + 65536));
+      }
+      // Settles after a task, as a real request does, so a due timer runs first.
+      sinon.stub(globalThis, 'fetch').callsFake(
+        (_input, init) =>
+          new Promise((resolve, reject) => {
+            init?.signal?.addEventListener('abort', () => {
+              reject(init.signal?.reason);
+            });
+            setTimeout(() => resolve(new Response('', { status: 200 })), 0);
+          })
+      );
+      const transport = createFetchTransport(gzipTransportParameters);
+      const shortTimeout = 10;
+
+      // act
+      const result = await transport.send(slowPayload, shortTimeout);
+
+      // assert
+      assert.strictEqual(result.status, 'failure');
+    });
+
+    it('aborts a stalled compression and fails without sending when the timeout fires', async function () {
+      // arrange
+      const cancel = sinon.spy();
+      sinon.stub(globalThis, 'CompressionStream').value(
+        class {
+          readonly writable = new WritableStream<Uint8Array>();
+          readonly readable = new ReadableStream<Uint8Array>({ cancel });
+        }
+      );
+      const fetchStub = sinon
+        .stub(globalThis, 'fetch')
+        .callsFake(async (_input, init) => {
+          init?.signal?.throwIfAborted();
+          return new Response('', { status: 200 });
+        });
+      const { warn } = registerMockDiagLogger();
+      const transport = createFetchTransport(gzipTransportParameters);
+      const shortTimeout = 10;
+
+      // act
+      const result = await transport.send(testPayload, shortTimeout);
+
+      // assert
+      assert.strictEqual(result.status, 'failure');
+      assert.strictEqual(
+        (fetchStub.firstCall.args[1] as RequestInit).signal?.aborted,
+        true
+      );
+      sinon.assert.calledOnce(cancel);
+      sinon.assert.notCalled(warn);
+    });
+
     it('compresses the body again on retry', async function () {
       // arrange
       const fetchStub = sinon.stub(globalThis, 'fetch');