fix: prevent memory leak in stream abort race (#223)

Closes #222 

- Create a fresh abort promise for each `stream.next()` call.
- Remove the abort listener as soon as either side of the race settles.
- Recheck the signal after registering the listener to handle an
already-aborted signal.

This prevents promise reactions and decoded message buffers from
accumulating on a long-lived pending abort promise.

Author: fracek

change/@apibara-indexer-5ca6e5cc-0d1c-4f01-9714-0ca20c9b3dc9.json

@@ -0,0 +1,7 @@
+{
+  "type": "patch",
+  "comment": "fix: refactor abort handling in stream processing",
+  "packageName": "@apibara/indexer",
+  "email": "jadejajaipal5@gmail.com",
+  "dependentChangeType": "patch"
+}

packages/indexer/src/indexer.ts

@@ -351,17 +351,15 @@ export async function run<TFilter, TBlock>(
 
     let onConnectCalled = false;
 
-    const abortPromise = abortSignal ? waitForAbort(abortSignal) : null;
-
     while (true) {
       let result: IteratorResult<
         StreamDataResponse<TBlock>,
         StreamDataResponse<TBlock>
       >;
 
       try {
-        result = abortPromise
-          ? await Promise.race([stream.next(), abortPromise])
+        result = abortSignal
+          ? await nextWithAbort(stream, abortSignal)
           : await stream.next();
       } catch (e) {
         if (abortSignal?.aborted) {
@@ -578,19 +576,31 @@ export async function run<TFilter, TBlock>(
 }
 
 /**
- * Returns a promise that rejects as soon as the given AbortSignal fires.
- * Used to race against stream.next() so the loop can break on abort.
+ * Races one stream read against an abort signal and removes its listener when
+ * either settles. Using a fresh abort promise prevents reactions from building
+ * up on one long-lived pending promise.
  */
-function waitForAbort(signal: AbortSignal): Promise<never> {
-  return new Promise<never>((_, reject) => {
-    if (signal.aborted) {
-      reject(signal.reason);
-      return;
-    }
-    signal.addEventListener("abort", () => reject(signal.reason), {
+async function nextWithAbort<T>(
+  stream: AsyncIterator<T, T>,
+  signal: AbortSignal,
+): Promise<IteratorResult<T, T>> {
+  let onAbort: () => void = () => {};
+  const abortPromise = new Promise<never>((_, reject) => {
+    onAbort = () => reject(signal.reason);
+    signal.addEventListener("abort", onAbort, {
       once: true,
     });
+
+    if (signal.aborted) {
+      onAbort();
+    }
   });
+
+  try {
+    return await Promise.race([stream.next(), abortPromise]);
+  } finally {
+    signal.removeEventListener("abort", onAbort);
+  }
 }
 
 async function registerMiddleware<TFilter, TBlock>(