Bug Report: Abort Signal Not Working for Streaming Server Functions #4651

// PROBLEMATIC CODE:
request.signal.removeEventListener('abort', abort)

if (isRaw) {
  return response  // <-- Signal disconnected too early for streaming!
}

import { createServerFn } from "@tanstack/react-start";

export const streamSearchFn = createServerFn({
  method: "GET",
  response: "raw",
})
  .validator((input: { query: string }) => input)
  .handler(async ({ data, signal }) => {
    console.log(`🚀 Server: Starting stream for query: "${data.query}"`);
    
    // This should receive abort events but doesn't
    signal.addEventListener("abort", () => {
      console.log(`🛑 Server: Abort signal received for "${data.query}"`);
    });

    const stream = new ReadableStream({
      async start(controller) {
        let count = 0;
        const intervalId = setInterval(() => {
          // This check never triggers because signal.aborted stays false
          if (signal.aborted) {
            console.log(`🛑 Server: Stream cancelled for "${data.query}"`);
            clearInterval(intervalId);
            controller.close();
            return;
          }

          const result = JSON.stringify({
            id: `result-${count++}`,
            name: `Result ${count} for "${data.query}"`,
            timestamp: new Date().toISOString(),
          });

          controller.enqueue(new TextEncoder().encode(result + "\n"));

          if (count >= 10) {
            clearInterval(intervalId);
            controller.close();
          }
        }, 1000);
      },
    });

    return new Response(stream, {
      headers: {
        "Content-Type": "text/event-stream",
        "Cache-Control": "no-cache",
        Connection: "keep-alive",
      },
    });
  });

function useStreamingSearch(query: string) {
  const abortControllerRef = useRef<AbortController | null>(null);

  useEffect(() => {
    // Cancel previous stream
    if (abortControllerRef.current) {
      console.log("🔄 Client: Cancelling previous stream");
      abortControllerRef.current.abort(); // This should cancel the server stream
    }

    const abortController = new AbortController();
    abortControllerRef.current = abortController;

    async function fetchStream() {
      const response = await streamSearchFn({
        data: { query },
        signal: abortController.signal, // Passed to server but not propagated
      });

      const reader = response.body?.getReader();
      // ... streaming logic
    }

    fetchStream();
  }, [query]);
}

const regularServerFn = createServerFn()
  .validator((input: { query: string }) => input)
  .handler(async ({ data, signal }) => {
    console.log(`🚀 Server: Starting regular function for query: "${data.query}"`);
    
    // This DOES work for regular functions
    signal.addEventListener("abort", () => {
      console.log(`🛑 Server: Abort signal received for "${data.query}"`);
    });

    return new Promise((resolve, reject) => {
      const timeout = setTimeout(() => {
        resolve(`Result for ${data.query}`);
      }, 2000);

      signal.addEventListener("abort", () => {
        clearTimeout(timeout);
        reject(new Error("Aborted"));
      });
    });
  });

// CURRENT (BROKEN):
request.signal.removeEventListener('abort', abort)

if (isRaw) {
  return response
}

// PROPOSED FIX:
if (isRaw) {
  // Keep abort listener connected for streaming responses
  return response
}

// Only remove listener for non-streaming responses
request.signal.removeEventListener('abort', abort)

test('streaming server function receives abort signal', async () => {
  const abortController = new AbortController();
  let abortReceived = false;
  
  const streamFn = createServerFn({ response: 'raw' })
    .handler(async ({ signal }) => {
      signal.addEventListener('abort', () => {
        abortReceived = true;
      });
      
      return new Response(new ReadableStream({
        start(controller) {
          const interval = setInterval(() => {
            if (signal.aborted) {
              clearInterval(interval);
              controller.close();
              return;
            }
            controller.enqueue(new TextEncoder().encode('data\n'));
          }, 100);
        }
      }));
    });

  const responsePromise = streamFn({ signal: abortController.signal });
  
  setTimeout(() => abortController.abort(), 50);
  
  await responsePromise;
  expect(abortReceived).toBe(true);
});

test('cancelling previous stream before starting new one', async () => {
  const streams = [];
  
  for (let i = 0; i < 3; i++) {
    const controller = new AbortController();
    if (streams.length > 0) {
      streams[streams.length - 1].controller.abort();
    }
    
    streams.push({
      controller,
      response: streamFn({ signal: controller.signal })
    });
  }
  
  // Only the last stream should still be active
  // Previous streams should be cancelled
});