Clean up abort signal listeners

src/client.ts

@@ -246,6 +246,11 @@ import { isEmptyObj } from './internal/utils/values';
 
 const WORKLOAD_IDENTITY_API_KEY_PLACEHOLDER = 'workload-identity-auth';
 
+type FetchWithTimeoutResponse = {
+  response: Response;
+  cleanupAbortSignal: () => void;
+};
+
 export type ApiKeySetter = () => Promise<string>;
 
 export interface ClientOptions {
@@ -741,10 +746,10 @@ 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 retryMessage = `retrying, ${retriesRemaining} attempts remaining`;
       if (options.signal?.aborted) {
         throw new Errors.APIUserAbortError();
@@ -754,8 +759,8 @@ export class OpenAI {
       // undici throws "TypeError: fetch failed" with cause "ConnectTimeoutError: Connect Timeout Error (attempted address: example:443, timeout: 1ms)"
       // others do not provide enough information to distinguish timeouts from other connection errors
       const isTimeout =
-        isAbortError(response) ||
-        /timed? ?out/i.test(String(response) + ('cause' in response ? String(response.cause) : ''));
+        isAbortError(fetchResult) ||
+        /timed? ?out/i.test(String(fetchResult) + ('cause' in fetchResult ? String(fetchResult.cause) : ''));
       if (retriesRemaining) {
         loggerFor(this).info(
           `[${requestLogID}] connection ${isTimeout ? 'timed out' : 'failed'} - ${retryMessage}`,
@@ -766,7 +771,7 @@ export class OpenAI {
             retryOfRequestLogID,
             url,
             durationMs: headersTime - startTime,
-            message: response.message,
+            message: fetchResult.message,
           }),
         );
         return this.retryRequest(options, retriesRemaining, retryOfRequestLogID ?? requestLogID);
@@ -780,21 +785,22 @@ export class OpenAI {
           retryOfRequestLogID,
           url,
           durationMs: headersTime - startTime,
-          message: response.message,
+          message: fetchResult.message,
         }),
       );
-      if (response instanceof OAuthError || response instanceof SubjectTokenProviderError) {
-        throw response;
+      if (fetchResult instanceof OAuthError || fetchResult instanceof SubjectTokenProviderError) {
+        throw fetchResult;
       }
       if (isTimeout) {
         throw new Errors.APIConnectionTimeoutError();
       }
       throw new Errors.APIConnectionError({
-        message: getConnectionErrorMessage(response),
-        cause: response,
+        message: getConnectionErrorMessage(fetchResult),
+        cause: fetchResult,
       });
     }
 
+    const { response, cleanupAbortSignal } = fetchResult;
     const specialHeaders = [...response.headers.entries()]
       .filter(([name]) => name === 'x-request-id')
       .map(([name, value]) => ', ' + name + ': ' + JSON.stringify(value))
@@ -811,7 +817,11 @@ export class OpenAI {
         !options.__metadata?.['hasStreamingBody'] &&
         !options.__metadata?.['workloadIdentityTokenRefreshed']
       ) {
-        await Shims.CancelReadableStream(response.body);
+        try {
+          await Shims.CancelReadableStream(response.body);
+        } finally {
+          cleanupAbortSignal();
+        }
         this._workloadIdentityAuth.invalidateToken();
 
         return this.makeRequest(
@@ -832,7 +842,11 @@ export class OpenAI {
         const retryMessage = `retrying, ${retriesRemaining} attempts remaining`;
 
         // We don't need the body of this response.
-        await Shims.CancelReadableStream(response.body);
+        try {
+          await Shims.CancelReadableStream(response.body);
+        } finally {
+          cleanupAbortSignal();
+        }
         loggerFor(this).info(`${responseInfo} - ${retryMessage}`);
         loggerFor(this).debug(
           `[${requestLogID}] response error (${retryMessage})`,
@@ -856,7 +870,14 @@ export class OpenAI {
 
       loggerFor(this).info(`${responseInfo} - ${retryMessage}`);
 
-      const errText = await response.text().catch((err: any) => castToError(err).message);
+      let errText: string;
+      try {
+        errText = await response.text();
+      } catch (err: any) {
+        errText = castToError(err).message;
+      } finally {
+        cleanupAbortSignal();
+      }
       const errJSON = safeJSON(errText) as any;
       const errMessage = errJSON ? undefined : errText;
 
@@ -888,7 +909,15 @@ export class OpenAI {
       }),
     );
 
-    return { response, options, controller, requestLogID, retryOfRequestLogID, startTime };
+    return {
+      response,
+      options,
+      controller,
+      cleanupAbortSignal,
+      requestLogID,
+      retryOfRequestLogID,
+      startTime,
+    };
   }
 
   getAPIList<Item, PageClass extends Pagination.AbstractPage<Item> = Pagination.AbstractPage<Item>>(
@@ -924,7 +953,7 @@ export class OpenAI {
       bearerAuth: true,
       adminAPIKeyAuth: true,
     },
-  ): Promise<Response> {
+  ): Promise<FetchWithTimeoutResponse> {
     if (this._workloadIdentityAuth && schemes.bearerAuth) {
       const headers = init.headers as Headers;
       const authHeader = headers.get('Authorization');
@@ -944,10 +973,16 @@ export class OpenAI {
     init: RequestInit | undefined,
     ms: number,
     controller: AbortController,
-  ): Promise<Response> {
+  ): Promise<FetchWithTimeoutResponse> {
     const { signal, method, ...options } = init || {};
     const abort = this._makeAbort(controller);
     if (signal) signal.addEventListener('abort', abort, { once: true });
+    let cleanedUp = false;
+    const cleanupAbortSignal = () => {
+      if (!signal || cleanedUp) return;
+      signal.removeEventListener('abort', abort);
+      cleanedUp = true;
+    };
 
     const timeout = setTimeout(abort, ms);
 
@@ -969,7 +1004,11 @@ 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);
+      const response = await this.fetch.call(undefined, url, fetchOptions);
+      return { response, cleanupAbortSignal };
+    } catch (error) {
+      cleanupAbortSignal();
+      throw error;
     } finally {
       clearTimeout(timeout);
     }

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

@@ -10,6 +10,7 @@ export type APIResponseProps = {
   response: Response;
   options: FinalRequestOptions;
   controller: AbortController;
+  cleanupAbortSignal?: () => void;
   requestLogID: string;
   retryOfRequestLogID: string | undefined;
   startTime: number;
@@ -33,6 +34,7 @@ export async function defaultParseResponse<T>(
           props.controller,
           client,
           props.options.__synthesizeEventData,
+          props.cleanupAbortSignal,
         ) as any;
       }
 
@@ -41,34 +43,39 @@ export async function defaultParseResponse<T>(
         props.controller,
         client,
         props.options.__synthesizeEventData,
+        props.cleanupAbortSignal,
       ) as any;
     }
 
-    // fetch refuses to read the body when the status code is 204.
-    if (response.status === 204) {
-      return null as T;
-    }
+    try {
+      // fetch refuses to read the body when the status code is 204.
+      if (response.status === 204) {
+        return null as T;
+      }
 
-    if (props.options.__binaryResponse) {
-      return response as unknown as T;
-    }
+      if (props.options.__binaryResponse) {
+        return response as unknown as T;
+      }
+
+      const contentType = response.headers.get('content-type');
+      const mediaType = contentType?.split(';')[0]?.trim();
+      const isJSON = mediaType?.includes('application/json') || mediaType?.endsWith('+json');
+      if (isJSON) {
+        const contentLength = response.headers.get('content-length');
+        if (contentLength === '0') {
+          // if there is no content we can't do anything
+          return undefined as T;
+        }
 
-    const contentType = response.headers.get('content-type');
-    const mediaType = contentType?.split(';')[0]?.trim();
-    const isJSON = mediaType?.includes('application/json') || mediaType?.endsWith('+json');
-    if (isJSON) {
-      const contentLength = response.headers.get('content-length');
-      if (contentLength === '0') {
-        // if there is no content we can't do anything
-        return undefined as T;
+        const json = await response.json();
+        return addRequestID(json as T, response);
       }
 
-      const json = await response.json();
-      return addRequestID(json as T, response);
+      const text = await response.text();
+      return text as unknown as T;
+    } finally {
+      if (!props.options.__binaryResponse) props.cleanupAbortSignal?.();
     }
-
-    const text = await response.text();
-    return text as unknown as T;
   })();
   loggerFor(client).debug(
     `[${requestLogID}] response parsed`,

tests/index.test.ts

@@ -323,6 +323,112 @@ describe('instantiate client', () => {
     expect(spy).toHaveBeenCalledTimes(1);
   });
 
+  test('removes custom signal abort listener after request finishes', async () => {
+    const controller = new AbortController();
+    const addEventListenerSpy = jest.spyOn(controller.signal, 'addEventListener');
+    const removeEventListenerSpy = jest.spyOn(controller.signal, 'removeEventListener');
+
+    const client = new OpenAI({
+      baseURL: 'http://localhost:5000/',
+      apiKey: 'My API Key',
+      adminAPIKey: 'My Admin API Key',
+      fetch: () => {
+        return Promise.resolve(
+          new Response(JSON.stringify({}), {
+            headers: { 'Content-Type': 'application/json' },
+          }),
+        );
+      },
+    });
+
+    await client.get('/foo', { signal: controller.signal });
+
+    const abortListener = addEventListenerSpy.mock.calls.find(([event]) => event === 'abort')?.[1];
+    expect(abortListener).toBeDefined();
+    expect(addEventListenerSpy).toHaveBeenCalledWith('abort', abortListener, { once: true });
+    expect(removeEventListenerSpy).toHaveBeenCalledWith('abort', abortListener);
+  });
+
+  test('keeps custom signal abort listener until response body is consumed', async () => {
+    const controller = new AbortController();
+    const removeEventListenerSpy = jest.spyOn(controller.signal, 'removeEventListener');
+    let resolveFetch: (() => void) | undefined;
+    const fetchResolved = new Promise<void>((resolve) => {
+      resolveFetch = resolve;
+    });
+    let fetchSignal: AbortSignal | undefined;
+
+    const client = new OpenAI({
+      baseURL: 'http://localhost:5000/',
+      apiKey: 'My API Key',
+      adminAPIKey: 'My Admin API Key',
+      fetch: (_url, init = {}) => {
+        fetchSignal = init.signal as AbortSignal;
+        const body = new ReadableStream<Uint8Array>({
+          start(bodyController) {
+            fetchSignal?.addEventListener('abort', () => bodyController.error(new Error('aborted')));
+          },
+        });
+        resolveFetch?.();
+        return Promise.resolve(
+          new Response(body, {
+            headers: { 'Content-Type': 'application/json' },
+          }),
+        );
+      },
+    });
+
+    const request = client.get('/foo', { signal: controller.signal });
+    const result = request.then(
+      () => 'resolved',
+      () => 'rejected',
+    );
+    await fetchResolved;
+    await new Promise((resolve) => setTimeout(resolve, 0));
+
+    expect(removeEventListenerSpy).not.toHaveBeenCalled();
+
+    controller.abort();
+
+    expect(fetchSignal?.aborted).toBe(true);
+    await expect(
+      Promise.race([result, new Promise((resolve) => setTimeout(() => resolve('pending'), 100))]),
+    ).resolves.toBe('rejected');
+    expect(removeEventListenerSpy).toHaveBeenCalled();
+  });
+
+  test('removes custom signal abort listener after streamed response finishes', async () => {
+    const controller = new AbortController();
+    const addEventListenerSpy = jest.spyOn(controller.signal, 'addEventListener');
+    const removeEventListenerSpy = jest.spyOn(controller.signal, 'removeEventListener');
+
+    const client = new OpenAI({
+      baseURL: 'http://localhost:5000/',
+      apiKey: 'My API Key',
+      adminAPIKey: 'My Admin API Key',
+      fetch: () => {
+        return Promise.resolve(
+          new Response('data: {"id":"chunk"}\n\ndata: [DONE]\n\n', {
+            headers: { 'Content-Type': 'text/event-stream' },
+          }),
+        );
+      },
+    });
+
+    const stream = await client.get<any>('/foo', { stream: true, signal: controller.signal });
+    const abortListener = addEventListenerSpy.mock.calls.find(([event]) => event === 'abort')?.[1];
+    expect(abortListener).toBeDefined();
+    expect(removeEventListenerSpy).not.toHaveBeenCalled();
+
+    const chunks = [];
+    for await (const chunk of stream) {
+      chunks.push(chunk);
+    }
+
+    expect(chunks).toEqual([{ id: 'chunk' }]);
+    expect(removeEventListenerSpy).toHaveBeenCalledWith('abort', abortListener);
+  });
+
   test('normalized method', async () => {
     let capturedRequest: RequestInit | undefined;
     const testFetch = async (url: string | URL | Request, init: RequestInit = {}): Promise<Response> => {