fix: clean up fetch timeout abort listener

Author: meaqua9420

src/client.ts

@@ -246,6 +246,8 @@ import { isEmptyObj } from './internal/utils/values';
 
 const WORKLOAD_IDENTITY_API_KEY_PLACEHOLDER = 'workload-identity-auth';
 
+type FetchWithTimeoutResult = { response: Response; cleanup: () => void };
+
 export type ApiKeySetter = () => Promise<string>;
 
 export interface ClientOptions {
@@ -741,10 +743,11 @@ export class OpenAI {
 
     const security = options.__security ?? { bearerAuth: true };
     const controller = new AbortController();
-    const response = await this.fetchWithAuth(url, req, timeout, controller, security).catch(castToError);
+    const fetchResult = await this.fetchWithAuth(url, req, timeout, controller, security).catch(castToError);
     const headersTime = Date.now();
 
-    if (response instanceof globalThis.Error) {
+    if (fetchResult instanceof globalThis.Error) {
+      const response = fetchResult;
       const retryMessage = `retrying, ${retriesRemaining} attempts remaining`;
       if (options.signal?.aborted) {
         throw new Errors.APIUserAbortError();
@@ -795,6 +798,8 @@ export class OpenAI {
       });
     }
 
+    const { response, cleanup } = fetchResult;
+
     const specialHeaders = [...response.headers.entries()]
       .filter(([name]) => name === 'x-request-id')
       .map(([name, value]) => ', ' + name + ': ' + JSON.stringify(value))
@@ -812,6 +817,7 @@ export class OpenAI {
         !options.__metadata?.['workloadIdentityTokenRefreshed']
       ) {
         await Shims.CancelReadableStream(response.body);
+        cleanup();
         this._workloadIdentityAuth.invalidateToken();
 
         return this.makeRequest(
@@ -833,6 +839,7 @@ export class OpenAI {
 
         // We don't need the body of this response.
         await Shims.CancelReadableStream(response.body);
+        cleanup();
         loggerFor(this).info(`${responseInfo} - ${retryMessage}`);
         loggerFor(this).debug(
           `[${requestLogID}] response error (${retryMessage})`,
@@ -856,7 +863,10 @@ export class OpenAI {
 
       loggerFor(this).info(`${responseInfo} - ${retryMessage}`);
 
-      const errText = await response.text().catch((err: any) => castToError(err).message);
+      const errText = await response
+        .text()
+        .catch((err: any) => castToError(err).message)
+        .finally(cleanup);
       const errJSON = safeJSON(errText) as any;
       const errMessage = errJSON ? undefined : errText;
 
@@ -888,7 +898,7 @@ export class OpenAI {
       }),
     );
 
-    return { response, options, controller, requestLogID, retryOfRequestLogID, startTime };
+    return { response, options, controller, requestLogID, retryOfRequestLogID, startTime, cleanup };
   }
 
   getAPIList<Item, PageClass extends Pagination.AbstractPage<Item> = Pagination.AbstractPage<Item>>(
@@ -924,7 +934,7 @@ export class OpenAI {
       bearerAuth: true,
       adminAPIKeyAuth: true,
     },
-  ): Promise<Response> {
+  ): Promise<FetchWithTimeoutResult> {
     if (this._workloadIdentityAuth && schemes.bearerAuth) {
       const headers = init.headers as Headers;
       const authHeader = headers.get('Authorization');
@@ -944,12 +954,19 @@ export class OpenAI {
     init: RequestInit | undefined,
     ms: number,
     controller: AbortController,
-  ): Promise<Response> {
+  ): Promise<FetchWithTimeoutResult> {
     const { signal, method, ...options } = init || {};
     const abort = this._makeAbort(controller);
     if (signal) signal.addEventListener('abort', abort, { once: true });
 
     const timeout = setTimeout(abort, ms);
+    let cleanedUp = false;
+    const cleanup = () => {
+      if (cleanedUp) return;
+      cleanedUp = true;
+      clearTimeout(timeout);
+      if (signal) signal.removeEventListener('abort', abort);
+    };
 
     const isReadableBody =
       ((globalThis as any).ReadableStream && options.body instanceof (globalThis as any).ReadableStream) ||
@@ -969,9 +986,13 @@ export class OpenAI {
 
     try {
       // use undefined this binding; fetch errors if bound to something else in browser/cloudflare
-      return await this.fetch.call(undefined, url, fetchOptions);
-    } finally {
+      const response = await this.fetch.call(undefined, url, fetchOptions);
       clearTimeout(timeout);
+      if (!response.body) cleanup();
+      return { response, cleanup };
+    } catch (err) {
+      cleanup();
+      throw err;
     }
   }
 

src/core/streaming.ts

@@ -36,6 +36,7 @@ export class Stream<Item> implements AsyncIterable<Item> {
     controller: AbortController,
     client?: OpenAI,
     synthesizeEventData?: boolean,
+    cleanup?: () => void,
   ): Stream<Item> {
     let consumed = false;
     const logger = client ? loggerFor(client) : console;
@@ -95,6 +96,7 @@ export class Stream<Item> implements AsyncIterable<Item> {
       } finally {
         // If the user `break`s, abort the ongoing request.
         if (!done) controller.abort();
+        cleanup?.();
       }
     }
 

src/internal/parse.ts

@@ -13,37 +13,40 @@ export type APIResponseProps = {
   requestLogID: string;
   retryOfRequestLogID: string | undefined;
   startTime: number;
+  cleanup?: () => void;
 };
 
 export async function defaultParseResponse<T>(
   client: OpenAI,
   props: APIResponseProps,
 ): Promise<WithRequestID<T>> {
-  const { response, requestLogID, retryOfRequestLogID, startTime } = props;
-  const body = await (async () => {
-    if (props.options.stream) {
-      loggerFor(client).debug('response', response.status, response.url, response.headers, response.body);
-
-      // Note: there is an invariant here that isn't represented in the type system
-      // that if you set `stream: true` the response type must also be `Stream<T>`
+  const { response, requestLogID, retryOfRequestLogID, startTime, cleanup } = props;
+  if (props.options.stream) {
+    loggerFor(client).debug('response', response.status, response.url, response.headers, response.body);
 
-      if (props.options.__streamClass) {
-        return props.options.__streamClass.fromSSEResponse(
-          response,
-          props.controller,
-          client,
-          props.options.__synthesizeEventData,
-        ) as any;
-      }
+    // Note: there is an invariant here that isn't represented in the type system
+    // that if you set `stream: true` the response type must also be `Stream<T>`
 
-      return Stream.fromSSEResponse(
+    if (props.options.__streamClass) {
+      return props.options.__streamClass.fromSSEResponse(
         response,
         props.controller,
         client,
         props.options.__synthesizeEventData,
+        cleanup,
       ) as any;
     }
 
+    return Stream.fromSSEResponse(
+      response,
+      props.controller,
+      client,
+      props.options.__synthesizeEventData,
+      cleanup,
+    ) as any;
+  }
+
+  const body = await (async () => {
     // fetch refuses to read the body when the status code is 204.
     if (response.status === 204) {
       return null as T;
@@ -69,7 +72,7 @@ export async function defaultParseResponse<T>(
 
     const text = await response.text();
     return text as unknown as T;
-  })();
+  })().finally(() => cleanup?.());
   loggerFor(client).debug(
     `[${requestLogID}] response parsed`,
     formatRequestDetails({
@@ -80,7 +83,7 @@ export async function defaultParseResponse<T>(
       durationMs: Date.now() - startTime,
     }),
   );
-  return body;
+  return body as WithRequestID<T>;
 }
 
 export type WithRequestID<T> =

tests/index.test.ts

@@ -685,6 +685,125 @@ describe('retries', () => {
     expect(count).toEqual(3);
   });
 
+  test('removes caller abort listener after successful response parsing', async () => {
+    const callerController = new AbortController();
+    const addEventListenerSpy = jest.spyOn(callerController.signal, 'addEventListener');
+    const removeEventListenerSpy = jest.spyOn(callerController.signal, 'removeEventListener');
+    const testFetch = async (): Promise<Response> =>
+      new Response(JSON.stringify({ a: 1 }), { headers: { 'Content-Type': 'application/json' } });
+
+    const client = new OpenAI({
+      apiKey: 'My API Key',
+      adminAPIKey: 'My Admin API Key',
+      timeout: 1000,
+      fetch: testFetch,
+    });
+
+    expect(
+      await client.request({
+        path: '/foo',
+        method: 'get',
+        signal: callerController.signal,
+      }),
+    ).toEqual({ a: 1 });
+
+    const abortListener = addEventListenerSpy.mock.calls[0]?.[1];
+    expect(addEventListenerSpy).toHaveBeenCalledWith('abort', abortListener, { once: true });
+    expect(removeEventListenerSpy).toHaveBeenCalledWith('abort', abortListener);
+  });
+
+  test('keeps caller abort forwarding until response body parsing settles', async () => {
+    const callerController = new AbortController();
+    const addEventListenerSpy = jest.spyOn(callerController.signal, 'addEventListener');
+    const removeEventListenerSpy = jest.spyOn(callerController.signal, 'removeEventListener');
+    const encoder = new TextEncoder();
+    let fetchSignal: AbortSignal | undefined;
+
+    const testFetch = async (url: string | URL | Request, init: RequestInit = {}): Promise<Response> => {
+      fetchSignal = init.signal as AbortSignal;
+      return new Response(
+        new ReadableStream({
+          start(controller) {
+            controller.enqueue(encoder.encode('{"a":'));
+            fetchSignal?.addEventListener('abort', () => controller.error(new Error('body aborted')), {
+              once: true,
+            });
+          },
+        }),
+        { headers: { 'Content-Type': 'application/json' } },
+      );
+    };
+
+    const client = new OpenAI({
+      apiKey: 'My API Key',
+      adminAPIKey: 'My Admin API Key',
+      timeout: 1000,
+      maxRetries: 0,
+      fetch: testFetch,
+    });
+
+    const parsePromise = client
+      .request({
+        path: '/foo',
+        method: 'get',
+        signal: callerController.signal,
+      })
+      .then(
+        () => undefined,
+        (err) => err,
+      );
+
+    for (let i = 0; i < 5 && !fetchSignal; i++) {
+      await new Promise((resolve) => setTimeout(resolve, 0));
+    }
+
+    expect(fetchSignal?.aborted).toBe(false);
+    expect(removeEventListenerSpy).not.toHaveBeenCalled();
+
+    callerController.abort();
+
+    expect(fetchSignal?.aborted).toBe(true);
+    await expect(parsePromise).resolves.toBeInstanceOf(Error);
+
+    const abortListener = addEventListenerSpy.mock.calls[0]?.[1];
+    expect(addEventListenerSpy).toHaveBeenCalledWith('abort', abortListener, { once: true });
+    expect(removeEventListenerSpy).toHaveBeenCalledWith('abort', abortListener);
+  });
+
+  test('removes caller abort listener after streaming response is consumed', async () => {
+    const callerController = new AbortController();
+    const addEventListenerSpy = jest.spyOn(callerController.signal, 'addEventListener');
+    const removeEventListenerSpy = jest.spyOn(callerController.signal, 'removeEventListener');
+    const testFetch = async (): Promise<Response> =>
+      new Response('data: {"a":1}\n\ndata: [DONE]\n\n', {
+        headers: { 'Content-Type': 'text/event-stream' },
+      });
+
+    const client = new OpenAI({
+      apiKey: 'My API Key',
+      adminAPIKey: 'My Admin API Key',
+      timeout: 1000,
+      fetch: testFetch,
+    });
+
+    const stream = await client.request<any>({
+      path: '/foo',
+      method: 'get',
+      stream: true,
+      signal: callerController.signal,
+    });
+    const chunks: unknown[] = [];
+
+    for await (const chunk of stream as AsyncIterable<unknown>) {
+      chunks.push(chunk);
+    }
+
+    expect(chunks).toEqual([{ a: 1 }]);
+    const abortListener = addEventListenerSpy.mock.calls[0]?.[1];
+    expect(addEventListenerSpy).toHaveBeenCalledWith('abort', abortListener, { once: true });
+    expect(removeEventListenerSpy).toHaveBeenCalledWith('abort', abortListener);
+  });
+
   test('retry count header', async () => {
     let count = 0;
     let capturedRequest: RequestInit | undefined;