Server: abortable delay

src/server/trpc/trpc.fetchers.retrier.ts

@@ -1,4 +1,5 @@
 import { TRPCFetcherError } from '~/server/trpc/trpc.router.fetchers';
+import { abortableDelay } from '~/server/wire';
 
 
 const AIX_DEBUG_SERVER_RETRY = true;
@@ -138,24 +139,7 @@ export function createRetryablePromise<T>(operationFn: () => Promise<T>, abortSi
           console.log(`[fetchers.retrier] 🔄 Retrying attempt ${attemptNumber - 1}/${rp.maxAttempts - 1} after ${delayMs}ms delay`);
 
         // abortable wait
-        await new Promise<void>((resolveDelay) => {
-          if (abortSignal.aborted || delayMs <= 0) {
-            resolveDelay();
-            return;
-          }
-
-          const timer = setTimeout(resolveDelay, delayMs);
-
-          const onAbort = () => {
-            clearTimeout(timer);
-            resolveDelay();
-          };
-
-          abortSignal.addEventListener('abort', onAbort, { once: true });
-        });
-
-        // check if the wait was aborted
-        if (abortSignal.aborted) {
+        if (await abortableDelay(delayMs, abortSignal)) {
           reject(error);
           return;
         }

src/server/wire.ts

@@ -134,6 +134,30 @@ export function createEmptyReadableStream<T = Uint8Array>(): ReadableStream<T> {
 }
 
 
+/**
+ * Used in retry logic to wait between attempts while respecting abort signals.
+ * @returns True if aborted, false if completed normally
+ */
+export function abortableDelay(delayMs: number, abortSignal: AbortSignal): Promise<boolean> {
+  return new Promise<boolean>((resolve) => {
+    // pre-check: already aborted or invalid delay
+    if (abortSignal.aborted || delayMs <= 0) {
+      resolve(abortSignal.aborted);
+      return;
+    }
+
+    const timer = setTimeout(() => resolve(false), delayMs);
+
+    const onAbort = () => {
+      clearTimeout(timer);
+      resolve(true);
+    };
+
+    abortSignal.addEventListener('abort', onAbort, { once: true });
+  });
+}
+
+
 /**
  * Small debugging utility to log train of events, used on the server-side
  * for incoming packets (e.g. SSE).