Skip to content

Commit fde60c1

Browse files
waleedlatif1waleedlatif1
committed
improvement(data-drains): consolidate parseRetryAfter; add Datadog NDJSON line context
- Extract a single parseRetryAfter helper (capped at 30s, returns number | null) into lib/data-drains/destinations/utils.ts and remove the five local copies in bigquery, datadog, gcs, snowflake, and webhook. - Datadog parseNdjson now wraps JSON.parse in try/catch and surfaces the failing line index, matching BigQuery's parser.
1 parent 0e58703 commit fde60c1

6 files changed

Lines changed: 50 additions & 83 deletions

File tree

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

Lines changed: 1 addition & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -4,6 +4,7 @@ import { JWT } from 'google-auth-library'
44
import { z } from 'zod'
55
import {
66
type ParsedServiceAccount,
7+
parseRetryAfter,
78
parseServiceAccount,
89
refineServiceAccountJson,
910
sleepUntilAborted,
@@ -159,18 +160,6 @@ const RETRYABLE_STATUSES = new Set([408, 429, 500, 502, 503, 504])
159160
const MAX_RETRY_ATTEMPTS = 3
160161
const BASE_RETRY_DELAY_MS = 250
161162

162-
function parseRetryAfter(header: string | null): number | null {
163-
if (!header) return null
164-
const seconds = Number(header)
165-
if (Number.isFinite(seconds) && seconds >= 0) return Math.min(seconds * 1000, 30_000)
166-
const dateMs = Date.parse(header)
167-
if (Number.isFinite(dateMs)) {
168-
const delta = dateMs - Date.now()
169-
if (delta > 0) return Math.min(delta, 30_000)
170-
}
171-
return null
172-
}
173-
174163
/**
175164
* Streams a chunk of rows to `tabledata.insertAll`.
176165
*

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

Lines changed: 12 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -2,7 +2,7 @@ import { gzipSync } from 'node:zlib'
22
import { createLogger } from '@sim/logger'
33
import { toError } from '@sim/utils/errors'
44
import { z } from 'zod'
5-
import { sleepUntilAborted } from '@/lib/data-drains/destinations/utils'
5+
import { parseRetryAfter, sleepUntilAborted } from '@/lib/data-drains/destinations/utils'
66
import type { DeliveryMetadata, DrainDestination } from '@/lib/data-drains/types'
77

88
const logger = createLogger('DataDrainDatadogDestination')
@@ -71,9 +71,15 @@ function buildEndpoint(site: DatadogSite): string {
7171
function parseNdjson(body: Buffer): unknown[] {
7272
const text = body.toString('utf8')
7373
const rows: unknown[] = []
74-
for (const line of text.split(/\r?\n/)) {
74+
const lines = text.split(/\r?\n/)
75+
for (let i = 0; i < lines.length; i++) {
76+
const line = lines[i]
7577
if (line.length === 0) continue
76-
rows.push(JSON.parse(line))
78+
try {
79+
rows.push(JSON.parse(line))
80+
} catch (error) {
81+
throw new Error(`NDJSON parse failed at line ${i}: ${toError(error).message}`)
82+
}
7783
}
7884
return rows
7985
}
@@ -110,20 +116,8 @@ function isRetryableStatus(status: number): boolean {
110116
return status === 408 || status === 429 || status >= 500
111117
}
112118

113-
function parseRetryAfter(header: string | null): number | undefined {
114-
if (!header) return undefined
115-
const seconds = Number.parseInt(header, 10)
116-
if (!Number.isNaN(seconds) && seconds >= 0) return seconds * 1000
117-
const dateMs = Date.parse(header)
118-
if (!Number.isNaN(dateMs)) {
119-
const delta = dateMs - Date.now()
120-
return delta > 0 ? delta : 0
121-
}
122-
return undefined
123-
}
124-
125-
function backoffWithJitter(attempt: number, retryAfterMs?: number): number {
126-
if (retryAfterMs !== undefined) {
119+
function backoffWithJitter(attempt: number, retryAfterMs: number | null): number {
120+
if (retryAfterMs !== null) {
127121
return Math.min(Math.max(retryAfterMs, BASE_BACKOFF_MS), MAX_BACKOFF_MS)
128122
}
129123
const exponential = Math.min(BASE_BACKOFF_MS * 2 ** (attempt - 1), MAX_BACKOFF_MS)
@@ -173,7 +167,7 @@ async function postWithRetries(input: PostInput): Promise<Response> {
173167
for (let attempt = 1; attempt <= MAX_ATTEMPTS; attempt++) {
174168
if (input.signal.aborted) throw input.signal.reason ?? new Error('Aborted')
175169
const perAttempt = AbortSignal.any([input.signal, AbortSignal.timeout(PER_ATTEMPT_TIMEOUT_MS)])
176-
let retryAfterMs: number | undefined
170+
let retryAfterMs: number | null = null
177171
let response: Response | undefined
178172
try {
179173
response = await fetch(input.url, {

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

Lines changed: 4 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -7,6 +7,7 @@ import {
77
buildObjectKey,
88
normalizePrefix,
99
type ParsedServiceAccount,
10+
parseRetryAfter,
1011
parseServiceAccount,
1112
refineServiceAccountJson,
1213
sleepUntilAborted,
@@ -99,23 +100,14 @@ function isRetryableStatus(status: number): boolean {
99100
)
100101
}
101102

102-
function backoffMs(attempt: number, retryAfterMs?: number): number {
103-
if (retryAfterMs !== undefined) {
103+
function backoffMs(attempt: number, retryAfterMs: number | null): number {
104+
if (retryAfterMs !== null) {
104105
return Math.min(Math.max(retryAfterMs, BASE_BACKOFF_MS), MAX_BACKOFF_MS)
105106
}
106107
const exp = Math.min(BASE_BACKOFF_MS * 2 ** (attempt - 1), MAX_BACKOFF_MS)
107108
return exp * (0.8 + Math.random() * 0.4)
108109
}
109110

110-
function parseRetryAfter(header: string | null): number | undefined {
111-
if (!header) return undefined
112-
const seconds = Number.parseInt(header, 10)
113-
if (!Number.isNaN(seconds) && seconds >= 0) return seconds * 1000
114-
const dateMs = Date.parse(header)
115-
if (!Number.isNaN(dateMs)) return Math.max(0, dateMs - Date.now())
116-
return undefined
117-
}
118-
119111
interface RetryRequestInput {
120112
action: string
121113
bucket: string
@@ -149,7 +141,7 @@ async function fetchWithRetry(input: RetryRequestInput): Promise<void> {
149141
error: toError(error).message,
150142
})
151143
if (attempt < MAX_ATTEMPTS) {
152-
await sleepUntilAborted(backoffMs(attempt), input.signal)
144+
await sleepUntilAborted(backoffMs(attempt, null), input.signal)
153145
continue
154146
}
155147
throw error

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

Lines changed: 1 addition & 25 deletions
Original file line numberDiff line numberDiff line change
@@ -3,7 +3,7 @@ import { createLogger } from '@sim/logger'
33
import { toError } from '@sim/utils/errors'
44
import { importPKCS8, SignJWT } from 'jose'
55
import { z } from 'zod'
6-
import { sleepUntilAborted } from '@/lib/data-drains/destinations/utils'
6+
import { parseRetryAfter, sleepUntilAborted } from '@/lib/data-drains/destinations/utils'
77
import type { DrainDestination } from '@/lib/data-drains/types'
88

99
const logger = createLogger('DataDrainSnowflakeDestination')
@@ -175,30 +175,6 @@ function parseNdjson(body: Buffer): string[] {
175175
return lines
176176
}
177177

178-
/**
179-
* Parses a Retry-After header. Supports both delta-seconds and HTTP-date
180-
* formats. Returns the delay in milliseconds, or `null` if not parseable.
181-
*/
182-
/** Cap Retry-After to keep cancellation latency bounded, matching the other destinations. */
183-
const RETRY_AFTER_MAX_MS = 30_000
184-
185-
function parseRetryAfter(header: string | null): number | null {
186-
if (!header) return null
187-
const trimmed = header.trim()
188-
if (trimmed.length === 0) return null
189-
const seconds = Number(trimmed)
190-
if (Number.isFinite(seconds) && seconds >= 0) {
191-
return Math.min(Math.floor(seconds * 1000), RETRY_AFTER_MAX_MS)
192-
}
193-
const dateMs = Date.parse(trimmed)
194-
if (!Number.isNaN(dateMs)) {
195-
const delta = dateMs - Date.now()
196-
if (delta <= 0) return 0
197-
return Math.min(delta, RETRY_AFTER_MAX_MS)
198-
}
199-
return null
200-
}
201-
202178
function isRetryableStatus(status: number): boolean {
203179
return status === 408 || status === 429 || (status >= 500 && status <= 599)
204180
}

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

Lines changed: 28 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -27,6 +27,34 @@ export function sleepUntilAborted(ms: number, signal: AbortSignal): Promise<void
2727
* (it produces an empty-name segment), and we want exactly one boundary
2828
* between prefix and the rest of the key.
2929
*/
30+
/**
31+
* Maximum HTTP Retry-After value we honor. A server requesting >30s is treated
32+
* as a 30s delay so a misconfigured upstream can't stall a drain run.
33+
*/
34+
const RETRY_AFTER_MAX_MS = 30_000
35+
36+
/**
37+
* Parses an HTTP `Retry-After` header (either delta-seconds or HTTP-date) into
38+
* a millisecond delay, capped at 30s. Returns `null` when the header is
39+
* absent or unparseable so callers can fall back to their own backoff.
40+
*/
41+
export function parseRetryAfter(header: string | null): number | null {
42+
if (!header) return null
43+
const trimmed = header.trim()
44+
if (trimmed.length === 0) return null
45+
const seconds = Number(trimmed)
46+
if (Number.isFinite(seconds) && seconds >= 0) {
47+
return Math.min(Math.floor(seconds * 1000), RETRY_AFTER_MAX_MS)
48+
}
49+
const dateMs = Date.parse(trimmed)
50+
if (!Number.isNaN(dateMs)) {
51+
const delta = dateMs - Date.now()
52+
if (delta <= 0) return 0
53+
return Math.min(delta, RETRY_AFTER_MAX_MS)
54+
}
55+
return null
56+
}
57+
3058
export function normalizePrefix(raw: string | undefined): string {
3159
if (!raw) return ''
3260
const trimmed = raw.replace(/^\/+/, '').replace(/\/+$/, '')

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

Lines changed: 4 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -7,7 +7,7 @@ import {
77
secureFetchWithPinnedIP,
88
validateUrlWithDNS,
99
} from '@/lib/core/security/input-validation.server'
10-
import { sleepUntilAborted } from '@/lib/data-drains/destinations/utils'
10+
import { parseRetryAfter, sleepUntilAborted } from '@/lib/data-drains/destinations/utils'
1111
import type { DeliveryMetadata, DrainDestination } from '@/lib/data-drains/types'
1212

1313
const logger = createLogger('DataDrainWebhookDestination')
@@ -92,8 +92,8 @@ function sign(body: Buffer, secret: string, timestamp: number): string {
9292
return `t=${timestamp},${SIGNATURE_VERSION}=${hmac}`
9393
}
9494

95-
function backoffWithJitter(attempt: number, retryAfterMs?: number): number {
96-
if (retryAfterMs !== undefined) {
95+
function backoffWithJitter(attempt: number, retryAfterMs: number | null): number {
96+
if (retryAfterMs !== null) {
9797
// Floor at 500ms so a misbehaving server returning Retry-After: 0 cannot
9898
// pin us in a tight retry loop.
9999
return Math.min(Math.max(retryAfterMs, BASE_BACKOFF_MS), MAX_BACKOFF_MS)
@@ -103,18 +103,6 @@ function backoffWithJitter(attempt: number, retryAfterMs?: number): number {
103103
return exponential * (0.8 + Math.random() * 0.4)
104104
}
105105

106-
function parseRetryAfter(header: string | null): number | undefined {
107-
if (!header) return undefined
108-
const seconds = Number.parseInt(header, 10)
109-
if (!Number.isNaN(seconds) && seconds >= 0) return seconds * 1000
110-
const dateMs = Date.parse(header)
111-
if (!Number.isNaN(dateMs)) {
112-
const delta = dateMs - Date.now()
113-
return delta > 0 ? delta : 0
114-
}
115-
return undefined
116-
}
117-
118106
function isRetryableStatus(status: number): boolean {
119107
return status === 408 || status === 429 || status >= 500
120108
}
@@ -206,7 +194,7 @@ export const webhookDestination: DrainDestination<
206194
// fresh (otherwise long backoffs would push us outside the
207195
// verifier's skew window).
208196
const headers = buildHeaders({ config, credentials, body, contentType, metadata })
209-
let retryAfterMs: number | undefined
197+
let retryAfterMs: number | null = null
210198
let response: Awaited<ReturnType<typeof secureFetchWithPinnedIP>> | undefined
211199
try {
212200
response = await secureFetchWithPinnedIP(config.url, resolvedIP, {

0 commit comments

Comments
 (0)