diff --git a/src/invoiceBatchProcessor.ts b/src/invoiceBatchProcessor.ts index e9c17a0..026105f 100644 --- a/src/invoiceBatchProcessor.ts +++ b/src/invoiceBatchProcessor.ts @@ -53,9 +53,14 @@ function isRateLimitError(error: unknown): boolean { return error instanceof Error && /429|rate.?limit|too many requests/i.test(error.message); } +function isValidPauseMs(value: number): boolean { + return Number.isInteger(value) && value >= 0; +} + function retryAfterMs(error: unknown, fallbackMs: number): number { const withRetryAfter = error as { retryAfterMs?: number } | undefined; - return typeof withRetryAfter?.retryAfterMs === "number" ? withRetryAfter.retryAfterMs : fallbackMs; + const candidate = withRetryAfter?.retryAfterMs; + return typeof candidate === "number" && isValidPauseMs(candidate) ? candidate : fallbackMs; } export class InvoiceBatchProcessor { @@ -76,14 +81,34 @@ export class InvoiceBatchProcessor { const maxConcurrent = config.maxConcurrent ?? DEFAULT_MAX_CONCURRENT; const rateLimitPauseMs = config.rateLimitPauseMs ?? DEFAULT_RATE_LIMIT_PAUSE_MS; + // Issue #1010 — reject invalid maxConcurrent before any dispatch. + if (!Number.isInteger(maxConcurrent) || maxConcurrent < 1) { + throw new RangeError("maxConcurrent must be a positive integer"); + } + // Issue #1009 — reject invalid rateLimitPauseMs before any dispatch. + if (!isValidPauseMs(rateLimitPauseMs)) { + throw new RangeError("rateLimitPauseMs must be a non-negative integer"); + } + let cursor = 0; let pausedUntil = 0; let slotSeq = 0; const inFlight = new Map>(); + // Issue #1009 — monotonic pause deadline that workers recheck after each + // sleep so a shorter hint from another worker cannot shorten the shared + // pause, and a worker that falls asleep before a longer hint arrives + // does not dispatch immediately on wake. + const waitForGlobalPause = async (): Promise => { + while (true) { + const wait = pausedUntil - Date.now(); + if (wait <= 0) return; + await sleep(wait); + } + }; + const runOne = async (invoiceId: string): Promise => { - const wait = pausedUntil - Date.now(); - if (wait > 0) await sleep(wait); + await waitForGlobalPause(); const amount = config.amounts[invoiceId]; if (amount === undefined) { @@ -103,7 +128,10 @@ export class InvoiceBatchProcessor { return result; } catch (error) { if (isRateLimitError(error)) { - pausedUntil = Date.now() + retryAfterMs(error, rateLimitPauseMs); + // Issue #1009 — take the max so the pause deadline only ever moves + // forward, never backward. + const nextPausedUntil = Date.now() + retryAfterMs(error, rateLimitPauseMs); + pausedUntil = Math.max(pausedUntil, nextPausedUntil); } const result: BatchInvoiceResult = { invoiceId, @@ -125,7 +153,10 @@ export class InvoiceBatchProcessor { ); }; - for (let i = 0; i < maxConcurrent; i++) launch(); + // Issue #1010 — cap initial launches at invoice count so maxConcurrent + // values larger than the batch size don't stall or loop unnecessarily. + const initialLaunches = Math.min(maxConcurrent, invoiceIds.length); + for (let i = 0; i < initialLaunches; i++) launch(); while (inFlight.size > 0) { const { slot, result } = await Promise.race(inFlight.values());