fix(otlp-exporter-base): drain the fetch response body so that browsers release the keepalive quota

experimental/CHANGELOG.md

@@ -19,6 +19,7 @@ For notes on migrating to 2.x / 0.200.x see [the upgrade guide](doc/upgrade-to-2
 ### :bug: Bug Fixes
 
 * fix(sdk-node): support `headers_list` when creating OTLP exporters from declarative configuration [#6953](https://github.com/open-telemetry/opentelemetry-js/issues/6953) @JacksonWeber
+* fix(otlp-exporter-base): drain the fetch response body so that browsers release the keepalive quota [#7002](https://github.com/open-telemetry/opentelemetry-js/pull/7002) @anneheartrecord
 
 ### :books: Documentation
 

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

@@ -102,6 +102,8 @@ class FetchTransport implements IExporterTransport {
           : 'no-cors',
       });
 
+      await drainResponseBody(response);
+
       if (response.status >= 200 && response.status <= 299) {
         diag.debug(`export response success (status: ${response.status})`);
         return { status: 'success' };
@@ -160,3 +162,47 @@ export function createFetchTransport(
 function isFetchNetworkErrorRetryable(error: unknown): boolean {
   return error instanceof TypeError && !error.cause;
 }
+
+/**
+ * Reads the response body to its end and discards it.
+ *
+ * Chromium gives the request's share of the keepalive quota back only once the
+ * response body has been read to the end, and it skips the buffering consumer
+ * that would otherwise drain the body on its own when the response carries a
+ * `Cache-Control: no-store` header, which collectors commonly send. Leaving the
+ * body unread then leaks the quota until the document goes away and every
+ * following keepalive export stays pending forever.
+ *
+ * @see https://fetch.spec.whatwg.org/#fetch-processresponseendofbody
+ */
+async function drainResponseBody(response: Response): Promise<void> {
+  try {
+    // Empty and opaque responses have no body to read.
+    const body = response.body;
+    if (body == null) {
+      return;
+    }
+
+    // Throws when the body is already locked to another reader, which happens
+    // when the same response is handed to more than one export.
+    const reader = body.getReader();
+    try {
+      // Chunks are dropped as they arrive: the payload is not used, and
+      // buffering it - with `response.arrayBuffer()` for instance - would keep
+      // a response of arbitrary size in memory.
+      let chunk = await reader.read();
+      while (!chunk.done) {
+        chunk = await reader.read();
+      }
+    } finally {
+      // The reader keeps the body locked until it is released, which would
+      // make a later export handed the same response fail to acquire a reader
+      // and skip the drain.
+      reader.releaseLock();
+    }
+  } catch (error) {
+    // The export outcome is decided by the response status, a body that cannot
+    // be read must not change it.
+    diag.debug(`error reading export response body: ${error}`);
+  }
+}

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

@@ -31,6 +31,25 @@ const MAX_KEEPALIVE_BODY_SIZE = 60 * 1024;
 // 9 is the max concurrent keepalive requests
 const MAX_KEEPALIVE_REQUESTS = 9;
 
+/**
+ * A response body that delivers one chunk and then stays open until the
+ * request is aborted, at which point it fails the way a real fetch body does.
+ */
+function neverEndingBodyAbortedBy(
+  signal: AbortSignal | null | undefined
+): ReadableStream<Uint8Array> {
+  return new ReadableStream({
+    start(controller) {
+      controller.enqueue(Uint8Array.from([1, 2, 3]));
+      signal?.addEventListener('abort', () =>
+        controller.error(
+          new DOMException('The user aborted a request.', 'AbortError')
+        )
+      );
+    },
+  });
+}
+
 describe('FetchTransport', function () {
   afterEach(function () {
     sinon.restore();
@@ -231,6 +250,140 @@ describe('FetchTransport', function () {
     });
   });
 
+  describe('response body handling', function () {
+    it('reads the response body of a successful export', async function () {
+      // arrange
+      let cancelled = false;
+      const body = new ReadableStream({
+        start(controller) {
+          controller.enqueue(Uint8Array.from([1, 2, 3]));
+          controller.close();
+        },
+        cancel() {
+          cancelled = true;
+        },
+      });
+      const response = new Response(body, { status: 200 });
+      sinon.stub(globalThis, 'fetch').resolves(response);
+      const transport = createFetchTransport(testTransportParameters);
+
+      // act
+      const result = await transport.send(testPayload, requestTimeout);
+
+      // assert - the body has to be read to its end, cancelling it does not
+      // release the keepalive quota the request holds
+      assert.strictEqual(result.status, 'success');
+      assert.strictEqual(response.bodyUsed, true);
+      assert.strictEqual(cancelled, false);
+    });
+
+    it('reads the response body of a retryable export', async function () {
+      // arrange
+      const response = new Response('test response', {
+        status: 503,
+        headers: { 'Retry-After': '5' },
+      });
+      sinon.stub(globalThis, 'fetch').resolves(response);
+      const transport = createFetchTransport(testTransportParameters);
+
+      // act
+      const result = await transport.send(testPayload, requestTimeout);
+
+      // assert
+      assert.strictEqual(result.status, 'retryable');
+      assert.strictEqual(response.bodyUsed, true);
+    });
+
+    it('releases the reader lock after draining the body', async function () {
+      // arrange - a held reader would keep the body locked, and a later export
+      // handed the same response could then not drain it
+      const response = new Response('test response', { status: 200 });
+      sinon.stub(globalThis, 'fetch').resolves(response);
+      const transport = createFetchTransport(testTransportParameters);
+
+      // act
+      const first = await transport.send(testPayload, requestTimeout);
+      const second = await transport.send(testPayload, requestTimeout);
+
+      // assert
+      assert.strictEqual(first.status, 'success');
+      assert.strictEqual(second.status, 'success');
+      assert.strictEqual(response.bodyUsed, true);
+      assert.strictEqual(response.body?.locked, false);
+    });
+
+    it('returns success when the response body is locked by another reader', async function () {
+      // arrange - a fetch wrapper may hold the body's reader. The export
+      // already reached the collector, so a body that cannot be drained must
+      // not turn into a network error and have the export retried.
+      const response = new Response('test response', { status: 200 });
+      response.body?.getReader();
+      sinon.stub(globalThis, 'fetch').resolves(response);
+      const { debug } = registerMockDiagLogger();
+      const transport = createFetchTransport(testTransportParameters);
+
+      // act
+      const result = await transport.send(testPayload, requestTimeout);
+
+      // assert
+      assert.strictEqual(result.status, 'success');
+      sinon.assert.calledWithMatch(debug, /error reading export response body/);
+    });
+
+    it('returns success when the response body cannot be read', async function () {
+      // arrange
+      const erroringBody = new ReadableStream({
+        start(controller) {
+          controller.error(new Error('body read failed'));
+        },
+      });
+      sinon
+        .stub(globalThis, 'fetch')
+        .resolves(new Response(erroringBody, { status: 200 }));
+      const { debug } = registerMockDiagLogger();
+      const transport = createFetchTransport(testTransportParameters);
+
+      // act
+      const result = await transport.send(testPayload, requestTimeout);
+
+      // assert - the export already reached the collector, the read is best effort
+      assert.strictEqual(result.status, 'success');
+      sinon.assert.calledWithMatch(debug, /error reading export response body/);
+    });
+
+    // The timeout keeps running while the body is drained, so a collector that
+    // holds the response open long enough gets the request aborted in the
+    // middle of the read. The headers are in by then, so the status the
+    // collector sent still decides the export outcome.
+    for (const { status, expected } of [
+      { status: 200, expected: 'success' },
+      { status: 503, expected: 'retryable' },
+    ]) {
+      it(`returns ${expected} when the timeout aborts the export while its ${status} response body is being read`, async function () {
+        // arrange - a body that stays open until the request is aborted, which
+        // is what the reader sees when the timeout fires mid-drain
+        sinon
+          .stub(globalThis, 'fetch')
+          .callsFake(
+            async (_input, init) =>
+              new Response(neverEndingBodyAbortedBy(init?.signal), { status })
+          );
+        const { debug } = registerMockDiagLogger();
+        const transport = createFetchTransport(testTransportParameters);
+
+        // act
+        const result = await transport.send(testPayload, 1);
+
+        // assert
+        assert.strictEqual(result.status, expected);
+        sinon.assert.calledWithMatch(
+          debug,
+          /error reading export response body/
+        );
+      });
+    }
+  });
+
   describe('keepalive queue tracking', function () {
     it('enables keepalive for small requests under limits', async function () {
       // arrange