fix(data-drains): extract sleepUntilAborted, honor abort across all destinations

Author: waleedlatif1

apps/sim/lib/data-drains/destinations/bigquery.ts

@@ -2,6 +2,7 @@ import { createLogger } from '@sim/logger'
 import { toError } from '@sim/utils/errors'
 import { JWT } from 'google-auth-library'
 import { z } from 'zod'
+import { sleepUntilAborted } from '@/lib/data-drains/destinations/utils'
 import type { DeliveryMetadata, DrainDestination } from '@/lib/data-drains/types'
 
 const logger = createLogger('DataDrainBigQueryDestination')
@@ -215,21 +216,6 @@ const RETRYABLE_STATUSES = new Set([408, 429, 500, 502, 503, 504])
 const MAX_RETRY_ATTEMPTS = 3
 const BASE_RETRY_DELAY_MS = 250
 
-function sleepUntilAborted(ms: number, signal: AbortSignal): Promise<void> {
-  if (signal.aborted) return Promise.resolve()
-  return new Promise((resolve) => {
-    const onAbort = () => {
-      clearTimeout(timeoutId)
-      resolve()
-    }
-    const timeoutId = setTimeout(() => {
-      signal.removeEventListener('abort', onAbort)
-      resolve()
-    }, ms)
-    signal.addEventListener('abort', onAbort, { once: true })
-  })
-}
-
 function parseRetryAfter(header: string | null): number | null {
   if (!header) return null
   const seconds = Number(header)

apps/sim/lib/data-drains/destinations/datadog.ts

@@ -2,6 +2,7 @@ import { gzipSync } from 'node:zlib'
 import { createLogger } from '@sim/logger'
 import { toError } from '@sim/utils/errors'
 import { z } from 'zod'
+import { sleepUntilAborted } from '@/lib/data-drains/destinations/utils'
 import type { DeliveryMetadata, DrainDestination } from '@/lib/data-drains/types'
 
 const logger = createLogger('DataDrainDatadogDestination')
@@ -70,7 +71,7 @@ function buildEndpoint(site: DatadogSite): string {
 function parseNdjson(body: Buffer): unknown[] {
   const text = body.toString('utf8')
   const rows: unknown[] = []
-  for (const line of text.split('\n')) {
+  for (const line of text.split(/\r?\n/)) {
     if (line.length === 0) continue
     rows.push(JSON.parse(line))
   }
@@ -129,21 +130,6 @@ function backoffWithJitter(attempt: number, retryAfterMs?: number): number {
   return exponential * (0.8 + Math.random() * 0.4)
 }
 
-function sleepUntilAborted(ms: number, signal: AbortSignal): Promise<void> {
-  if (signal.aborted) return Promise.resolve()
-  return new Promise((resolve) => {
-    const onAbort = () => {
-      clearTimeout(timeoutId)
-      resolve()
-    }
-    const timeoutId = setTimeout(() => {
-      signal.removeEventListener('abort', onAbort)
-      resolve()
-    }, ms)
-    signal.addEventListener('abort', onAbort, { once: true })
-  })
-}
-
 interface PreparedBody {
   body: Uint8Array | string
   headers: Record<string, string>

apps/sim/lib/data-drains/destinations/gcs.ts

@@ -1,9 +1,9 @@
 import { createLogger } from '@sim/logger'
 import { toError } from '@sim/utils/errors'
-import { sleep } from '@sim/utils/helpers'
 import { generateShortId } from '@sim/utils/id'
 import { JWT } from 'google-auth-library'
 import { z } from 'zod'
+import { sleepUntilAborted } from '@/lib/data-drains/destinations/utils'
 import type { DrainDestination } from '@/lib/data-drains/types'
 
 const logger = createLogger('DataDrainGCSDestination')
@@ -229,7 +229,7 @@ async function fetchWithRetry(input: RetryRequestInput): Promise<void> {
         error: toError(error).message,
       })
       if (attempt < MAX_ATTEMPTS) {
-        await sleep(backoffMs(attempt))
+        await sleepUntilAborted(backoffMs(attempt), input.signal)
         continue
       }
       throw error
@@ -249,7 +249,7 @@ async function fetchWithRetry(input: RetryRequestInput): Promise<void> {
     }
     lastError = new Error(`GCS ${input.action} responded with HTTP ${response.status}`)
     const retryAfterMs = parseRetryAfter(response.headers.get('retry-after'))
-    await sleep(backoffMs(attempt, retryAfterMs))
+    await sleepUntilAborted(backoffMs(attempt, retryAfterMs), input.signal)
   }
   throw lastError instanceof Error
     ? lastError

apps/sim/lib/data-drains/destinations/snowflake.ts

@@ -1,9 +1,9 @@
 import { createHash, createPublicKey } from 'node:crypto'
 import { createLogger } from '@sim/logger'
 import { toError } from '@sim/utils/errors'
-import { sleep } from '@sim/utils/helpers'
 import { importPKCS8, SignJWT } from 'jose'
 import { z } from 'zod'
+import { sleepUntilAborted } from '@/lib/data-drains/destinations/utils'
 import type { DrainDestination } from '@/lib/data-drains/types'
 
 const logger = createLogger('DataDrainSnowflakeDestination')
@@ -179,16 +179,22 @@ function parseNdjson(body: Buffer): string[] {
  * Parses a Retry-After header. Supports both delta-seconds and HTTP-date
  * formats. Returns the delay in milliseconds, or `null` if not parseable.
  */
+/** Cap Retry-After to keep cancellation latency bounded, matching the other destinations. */
+const RETRY_AFTER_MAX_MS = 30_000
+
 function parseRetryAfter(header: string | null): number | null {
   if (!header) return null
   const trimmed = header.trim()
   if (trimmed.length === 0) return null
   const seconds = Number(trimmed)
-  if (Number.isFinite(seconds) && seconds >= 0) return Math.floor(seconds * 1000)
+  if (Number.isFinite(seconds) && seconds >= 0) {
+    return Math.min(Math.floor(seconds * 1000), RETRY_AFTER_MAX_MS)
+  }
   const dateMs = Date.parse(trimmed)
   if (!Number.isNaN(dateMs)) {
     const delta = dateMs - Date.now()
-    return delta > 0 ? delta : 0
+    if (delta <= 0) return 0
+    return Math.min(delta, RETRY_AFTER_MAX_MS)
   }
   return null
 }
@@ -254,7 +260,7 @@ async function executeStatement(input: ExecuteInput): Promise<void> {
         error: toError(error).message,
       })
       if (input.signal.aborted || attempt === EXECUTE_MAX_ATTEMPTS) throw error
-      await sleep(computeBackoffDelay(attempt))
+      await sleepUntilAborted(computeBackoffDelay(attempt), input.signal)
       continue
     }
     if (response.status === 202) {
@@ -282,7 +288,7 @@ async function executeStatement(input: ExecuteInput): Promise<void> {
       status: response.status,
       delayMs: delay,
     })
-    await sleep(delay)
+    await sleepUntilAborted(delay, input.signal)
   }
   throw lastError ?? new Error('Snowflake request failed after retries')
 }
@@ -309,7 +315,7 @@ async function pollStatement(input: PollInput): Promise<void> {
   let interval = POLL_INITIAL_INTERVAL_MS
   while (Date.now() < deadline) {
     if (input.signal.aborted) throw input.signal.reason ?? new Error('Aborted')
-    await sleep(interval)
+    await sleepUntilAborted(interval, input.signal)
     const response = await fetch(url, {
       headers: {
         Authorization: `Bearer ${input.jwt}`,

apps/sim/lib/data-drains/destinations/utils.ts

@@ -0,0 +1,19 @@
+/**
+ * Sleep for `ms` milliseconds, resolving early if `signal` aborts. Used by
+ * destination retry/poll loops so cancelled drain runs do not hang waiting on
+ * a `setTimeout` that ignores the abort signal.
+ */
+export function sleepUntilAborted(ms: number, signal: AbortSignal): Promise<void> {
+  if (signal.aborted) return Promise.resolve()
+  return new Promise((resolve) => {
+    const onAbort = () => {
+      clearTimeout(timeoutId)
+      resolve()
+    }
+    const timeoutId = setTimeout(() => {
+      signal.removeEventListener('abort', onAbort)
+      resolve()
+    }, ms)
+    signal.addEventListener('abort', onAbort, { once: true })
+  })
+}

apps/sim/lib/data-drains/destinations/webhook.ts

@@ -7,6 +7,7 @@ import {
   secureFetchWithPinnedIP,
   validateUrlWithDNS,
 } from '@/lib/core/security/input-validation.server'
+import { sleepUntilAborted } from '@/lib/data-drains/destinations/utils'
 import type { DeliveryMetadata, DrainDestination } from '@/lib/data-drains/types'
 
 const logger = createLogger('DataDrainWebhookDestination')
@@ -91,26 +92,6 @@ function sign(body: Buffer, secret: string, timestamp: number): string {
   return `t=${timestamp},${SIGNATURE_VERSION}=${hmac}`
 }
 
-/**
- * Resolves after `ms` or as soon as `signal` aborts, whichever happens first.
- * The caller checks `signal.aborted` at the top of the next iteration to
- * surface the abort — keeping resolution side-effect-free here.
- */
-function sleepUntilAborted(ms: number, signal: AbortSignal): Promise<void> {
-  if (signal.aborted) return Promise.resolve()
-  return new Promise((resolve) => {
-    const onAbort = () => {
-      clearTimeout(timeoutId)
-      resolve()
-    }
-    const timeoutId = setTimeout(() => {
-      signal.removeEventListener('abort', onAbort)
-      resolve()
-    }, ms)
-    signal.addEventListener('abort', onAbort, { once: true })
-  })
-}
-
 function backoffWithJitter(attempt: number, retryAfterMs?: number): number {
   if (retryAfterMs !== undefined) {
     // Floor at 500ms so a misbehaving server returning Retry-After: 0 cannot