🚀 Feature Request: Allow using an AbortController signal without stopping the Worker execution

[luizzappa]

Describe the solution

I use the Worker as a gateway to call the OpenAI API in streaming mode. Inside the Worker, there is code to track how many tokens are being used by each user (persist in my database) so that I can bill them accordingly.

One of my users' requests is the ability to abort a request (similar to what happens in ChatGPT, where you click the stop button and it stops generating the response).

I need to pass an abort signal from my frontend to the Worker, which will then pass the abort to OpenAI.

However, currently, the Worker implementation for executing the code upon receiving an Abort signal causes me to lose the token tracking and some other processing that I need to perform.

Upon receiving an abort signal, there should be the possibility for the worker to continue execution until completion, leaving it up to the implementation to handle what should happen using request.onabort.

[andyjessop]

I'm wondering if this would be a good use-case for waitUntil? waitUntil should allow you to continue work even after the response has been aborted.

const response = await fetchOpenAI(request);
const stream = response.body;

ctx.waitUntil(trackTokenUsage(stream));

return new Response(stream);

[luizzappa]

Hi @andyjessop ,

Firstly, thank you for your suggestion. It seems you were on the right track, but I encountered a hurdle when trying to implement it.

The fetchOpenAI function you mentioned seems to be the key here. I need to pass the abort signal to the OpenAI endpoint to halt the stream and prevent further completions (thus saving token usage).

async function fetchOpenAI(request) {
  return openai.beta.chat.completions.stream({
          body,
          stream: true
        },{
          signal: request.signal
        });
}

However, upon reviewing the implementation of trackTokensUsage, it seems that the abort signal isn't being transmitted properly to the OpenAI fetch. This means that I'm still receiving chunks, resulting in charges both on the OpenAI side and for our users.

async function trackTokensUsage(stream) {
  stream
      .on('chunk', async (chunk) => {
          calcTokensUsage(chunk);
      })
      .on('abort', async () => { // <-- This part is never called
          console.log('aborted');
      });
}

Upon inspecting the state of request.signal, I noticed that it's never marked as aborted. Even when performing a simple test:

ctx.waitUntil(new Promise(async (res) => {
  await new Promise(res => setTimeout(() => res(''), 3000));
  console.log('request.signal.aborted', request.signal.aborted);
}));

The output is never true, even after attempting to abort. This is the part that puzzles me the most: within waitUntil, I would expect the state to reflect the current situation of the request.

Am I missing any crucial details here?

[andyjessop]

I'm not too familiar with this aspect of OpenAI's API, but in their docs it says:

If you need to cancel a stream, you can break from the loop or call stream.controller.abort().

https://github.com/openai/openai-node?tab=readme-ov-file#streaming-responses

Which suggests that it already has an abort controller. You might, therefore, need a way to listen to your request signal and abort the OpenAI stream manually.

request.signal.addEventListener("abort", () => {
  stream.controller.abort();
});

[luizzappa]

@andyjessop , thank you so much for your assistance thus far. It appears that the issue lies not in the call to OpenAI, but rather within the Cloudflare Worker itself.

I've noticed that the code you suggested is never executed when the worker receives an abort controller.

request.signal.addEventListener('abort', () => console.log('aborted')); // <- never called

Even when wrapped in a waitUntil:

ctx.waitUntil(new Promise(async (resolve) => {
  request.signal.addEventListener('abort', () => console.log('aborted')); // <- never reached
  await new Promise(resolve => setTimeout(() => resolve(''), 3000));
  console.log('request.signal.aborted', request.signal.aborted); // <- never becomes true
}));

The abort signal is reaching the Wrangler, as the connection is being closed:

[jasnell]

The request.signal is currently only used when request is used in a client side fetch where an AbortSignal has been passed in. We can investigate making it usable on server-side request objects but it is likely a lower priority and may not happen for a while

[ricardopolo]

if this is not supported, how can we send a cancellation from a client to cf workers?