From 5cee402784e965849a08011aedce6d9302401004 Mon Sep 17 00:00:00 2001 From: Emirlan Rasulov Date: Fri, 16 Oct 2026 16:31:51 +0600 Subject: [PATCH 1/3] fix(batch): match the Batch client to what the API returns and accepts - Add the expired task status. Bulk downloads skip expired and failed rows that have no body and report each through onSkip, instead of failing partway through. - Add getTaskContentBytes, which returns a Buffer so binary results such as PDFs arrive intact. getTaskContent still returns text. - Run waiters no longer poll a paused run until the timeout: they throw RunPausedError with the run, unless waitWhilePaused is set. - Add verifyWebhookSignature, which decodes the base64 secret, picks the key by kid, compares in constant time and checks the timestamp (300 s by default), plus the WebhookEvent type and header constants. - Add BatchErrorCode with every error code the API sends, and detail and retryAfter on ZenRowsBatchError. - addTasks accepts an idempotencyKey, so a transient failure is retried. - rerun, retryFailed and submitScheduled accept waitForIngest. - putJobWebhook no longer sends signature: false when it is left out; it keeps the job's current setting. - A 429 quota_exceeded is an account limit and is no longer retried. - Requests time out after 30 seconds by default (timeout option); they had no limit before. - The cost estimate counts a flag as on only for true or "true", and mode only for "auto", the values the API accepts. - Types: method and body on tasks, created_at and updated_at on results, trigger on runs. TaskResult.spend and the results limit option are deprecated: the API never returns the first and ignores the second. --- src/batch/client.ts | 130 ++++++++++++-- src/batch/download.ts | 87 +++++++-- src/batch/errors.ts | 109 +++++++++++- src/batch/estimate.ts | 29 +-- src/batch/resources.ts | 49 ++++- src/batch/transport.ts | 30 +++- src/batch/types.ts | 51 +++++- src/batch/waiters.ts | 17 ++ src/batch/webhook.ts | 103 +++++++++++ tests/batch-client.test.ts | 218 ++++++++++++++++++++++- tests/batch-consistency.test.ts | 306 ++++++++++++++++++++++++++++++++ tests/batch-download.test.ts | 157 +++++++++++++++- tests/batch-estimate.test.ts | 44 ++++- tests/batch-webhook.test.ts | 147 +++++++++++++++ 14 files changed, 1408 insertions(+), 69 deletions(-) create mode 100644 src/batch/webhook.ts create mode 100644 tests/batch-consistency.test.ts create mode 100644 tests/batch-webhook.test.ts diff --git a/src/batch/client.ts b/src/batch/client.ts index 0d70c45..34393f2 100644 --- a/src/batch/client.ts +++ b/src/batch/client.ts @@ -39,13 +39,15 @@ import type { Run, StartExportResponse, TaskHistoryResponse, + TaskMethod, TaskResult, TestWebhookResponse, WebhookConfig, } from "./types.js"; -import { pollUntil } from "./waiters.js"; +import { RunPausedError, pollUntil } from "./waiters.js"; const DEFAULT_BATCH_API_URL = "https://async.api.zenrows.com/v1"; +const DEFAULT_TIMEOUT_SECONDS = 30; // Run states that never transition again: the default waiter target. `failed` is an // account-level auto-fail (e.g. `api_key_cap_reached`, `insufficient_credits`). @@ -55,8 +57,17 @@ const TERMINAL_EXPORT_STATUSES = new Set(["completed", "failed"]); export interface BatchClientConfig { /** Override the Batch API base URL. Matches the Go/Python batch clients' base-URL override. */ baseURL?: string; - /** Retries for transient failures (429/502/503/504 + network errors) on idempotent requests. Default 3. */ + /** + * Retries for transient failures (429/502/503/504 + network errors) on idempotent requests. + * Default 3. A 429 `quota_exceeded` (an account limit) is never retried. + */ retries?: number; + /** + * Seconds each API request may take. Default 30. Waiters (`wait`, `waitForRun`) poll with + * many short requests, so this does not limit how long a wait lasts, and result downloads + * from `result_url` are not limited by it either. + */ + timeout?: number; } export interface BatchTask { @@ -64,10 +75,18 @@ export interface BatchTask { external_id?: string; metadata?: Record; zenrows_params?: Record; + /** HTTP method for the target fetch. Default `GET`. */ + method?: TaskMethod; + /** Request body for a `POST` task: a JSON value or a string, up to 16 KiB. */ + body?: unknown; } export interface BatchWebhookConfig { url: string; + /** + * Sign deliveries with the account's HMAC key. On `putJobWebhook`, pass it every time; + * see that method for what happens when it is left out. + */ signature?: boolean; } @@ -106,6 +125,10 @@ export interface ListJobsOptions { export interface GetResultsOptions { runId?: string; cursor?: string; + /** + * @deprecated Ignored and not sent: the results endpoint returns pages of a fixed size + * (about 1,000 rows). Follow `next_cursor`, or use `iterResults`. + */ limit?: number; status?: string; } @@ -126,6 +149,12 @@ export interface WaitForRunOptions { timeout?: number; pollInterval?: number; maxPollInterval?: number; + /** + * A paused run never progresses until someone resumes it, so by default the wait stops + * at once with a `RunPausedError` carrying the run. Set this to keep polling while + * paused, for example when another process will resume it. + */ + waitWhilePaused?: boolean; } type TaskInput = @@ -136,14 +165,13 @@ type TaskInput = metadata?: Record; /** Per-task scrape params, merged over the job-level `zenrowsParams` (task wins). */ zenrows_params?: Record; + /** HTTP method for the target fetch. Default `GET`. */ + method?: TaskMethod; + /** Request body for a `POST` task: a JSON value or a string, up to 16 KiB. */ + body?: unknown; }; -function coerceUrl(item: TaskInput): { - url: string; - external_id?: string; - metadata?: Record; - zenrows_params?: Record; -} { +function coerceUrl(item: TaskInput): BatchTask { return typeof item === "string" ? { url: item } : item; } @@ -167,6 +195,8 @@ export class ZenRowsBatchClient { config.baseURL ?? DEFAULT_BATCH_API_URL, apiKey, config.retries, + undefined, + (config.timeout ?? DEFAULT_TIMEOUT_SECONDS) * 1000, ); } @@ -254,6 +284,7 @@ export class ZenRowsBatchClient { metadata: opts.metadata, webhook: opts.webhook, idempotencyKey: opts.idempotencyKey, + waitForIngest: opts.waitForIngest, // biome-ignore lint/suspicious/noExplicitAny: schedule builder output matches JobSchedule at the wire boundary ...{ schedule: schedule.toRequestBody() as unknown as any }, }); @@ -325,7 +356,7 @@ export class ZenRowsBatchClient { // ===== results / content ===== async getResults(jobId: string, options: GetResultsOptions = {}): Promise { - const { runId, ...query } = options; + const { runId, limit: _ignored, ...query } = options; const path = runId ? `/jobs/${jobId}/runs/${runId}/results` : `/jobs/${jobId}/results`; return this.transport.requestJson("GET", path, { query }); } @@ -343,17 +374,46 @@ export class ZenRowsBatchClient { } } - /** Returns the scraped page's raw content as-is (HTML/text, not JSON). */ + /** + * Returns a task's stored content decoded as text (HTML, Markdown, plain text or JSON). + * Text decoding corrupts binary bodies: for a `pdf` or other binary result, use + * `getTaskContentBytes` instead. + * + * Throws a `ZenRowsBatchError`: 409 `task_not_ready` while the task runs, 422 with the + * task's error for a failed task, and 410 `result_expired` past the retention window. + */ async getTaskContent( jobId: string, taskId: string, options: { runId?: string } = {}, ): Promise { - const path = options.runId + const response = await this.transport.requestRaw( + "GET", + this.contentPath(jobId, taskId, options), + ); + return response.text(); + } + + /** + * Returns a task's stored content as raw bytes, so binary results such as PDFs arrive + * intact. Same errors as `getTaskContent`. + */ + async getTaskContentBytes( + jobId: string, + taskId: string, + options: { runId?: string } = {}, + ): Promise { + const response = await this.transport.requestRaw( + "GET", + this.contentPath(jobId, taskId, options), + ); + return Buffer.from(await response.arrayBuffer()); + } + + private contentPath(jobId: string, taskId: string, options: { runId?: string }): string { + return options.runId ? `/jobs/${jobId}/runs/${options.runId}/tasks/${taskId}/content` : `/jobs/${jobId}/tasks/${taskId}/content`; - const response = await this.transport.requestRaw("GET", path); - return response.text(); } getTaskHistory( @@ -407,12 +467,31 @@ export class ZenRowsBatchClient { return this.transport.requestJson("GET", `/jobs/${jobId}/webhook`); } - putJobWebhook(jobId: string, config: BatchWebhookConfig): Promise { + /** + * Replace a job's webhook. The API needs both `url` and `signature` on this call, so pass + * `signature` whenever you know the setting you want. + * + * When `signature` is left out, the SDK reads the job's current setting first (`false` when + * the job has no webhook yet) and sends that, so changing only the URL never turns signing + * off. That is two requests, not one atomic update: a change someone else makes in between + * is overwritten with the value read. + */ + async putJobWebhook(jobId: string, config: BatchWebhookConfig): Promise { + const signature = config.signature ?? (await this.currentWebhookSignature(jobId)); return this.transport.requestJson("PUT", `/jobs/${jobId}/webhook`, { - body: { signature: false, ...config }, + body: { url: config.url, signature }, }); } + private async currentWebhookSignature(jobId: string): Promise { + try { + return (await this.getJobWebhook(jobId)).signature === true; + } catch (error) { + if (error instanceof ZenRowsBatchError && error.status === 404) return false; + throw error; + } + } + async deleteJobWebhook(jobId: string): Promise { await this.transport.requestJson("DELETE", `/jobs/${jobId}/webhook`); } @@ -526,7 +605,7 @@ export class ZenRowsBatchClient { }); } - /** Start an export, wait for it, and save the zip to `targetPath`. Capped at 1 GiB per run. */ + /** Start an export, wait for it, and save the zip to `targetPath`. The API caps the size of one export. */ async downloadAllResults( jobId: string, runId: string, @@ -633,10 +712,11 @@ export class ZenRowsBatchClient { async _postTasks( jobId: string, tasks: BatchTask[], - options: { lastBatch?: boolean } = {}, + options: { lastBatch?: boolean; idempotencyKey?: string } = {}, ): Promise { return this.transport.requestJson("POST", `/jobs/${jobId}/tasks`, { body: { tasks, last_batch: options.lastBatch ?? false }, + idempotencyKey: options.idempotencyKey, }); } @@ -653,13 +733,26 @@ export class ZenRowsBatchClient { /** @internal */ async _waitForRunRaw(jobId: string, options: WaitForRunOptions): Promise { const target = options.targetStatuses ?? TERMINAL_RUN_STATUSES; - const fetchRun = async (): Promise => { + const loadRun = async (): Promise => { if (!options.runId) { const job = await this._getJobData(jobId); return job.latest_run; } return this._getRunData(jobId, options.runId); }; + const fetchRun = async (): Promise => { + const run = await loadRun(); + if ( + run !== undefined && + run.pause_state === "paused" && + !target.has(run.status) && + !options.failureStatuses?.has(run.status) && + !options.waitWhilePaused + ) { + throw new RunPausedError(run); + } + return run; + }; return pollUntil(fetchRun, { // A status the caller named as a failure throws, even when it is also a target. isDone: (run) => @@ -692,3 +785,4 @@ export class ZenRowsBatchClient { export { CurrentRun, ExportHandle, ExportRef, JobHandle, JobRef, RunHandle, RunRef }; export type { DownloadedResult }; +export * from "./webhook.js"; diff --git a/src/batch/download.ts b/src/batch/download.ts index 73ef72c..d52bdf9 100644 --- a/src/batch/download.ts +++ b/src/batch/download.ts @@ -17,13 +17,63 @@ export interface DownloadedResult { body: Buffer; } +/** + * A result row a bulk download left out because it has no body to fetch: an `expired` row + * (a successful result past the retention window) or a `failed` task. Reported through the + * `onSkip` option of `downloadToDir` / `downloadToMemory`. + */ +export interface SkippedResult { + taskId: string; + externalId?: string; + url: string; + status: "expired" | "failed"; +} + +/** Statuses whose rows have no stored body by design, so a bulk download skips them. */ +const SKIPPABLE_STATUSES = new Set(["expired", "failed"]); + +/** + * Split `results` into rows to fetch and rows to skip, enforcing `limit`. Only `expired` + * and `failed` rows without a `result_url` are skipped; any other row without one (a + * `successful` row missing its URL, or an unfinished task) stays in, so fetching it fails + * loudly instead of being dropped. + */ +async function partition( + results: AsyncIterable, + limit: number, + limitName: string, + onSkip: ((row: SkippedResult) => void) | undefined, +): Promise { + const fetchable: TaskResult[] = []; + for await (const task of results) { + if (!task.result_url && SKIPPABLE_STATUSES.has(task.status)) { + onSkip?.({ + taskId: task.task_id, + externalId: task.external_id, + url: task.url, + status: task.status as SkippedResult["status"], + }); + continue; + } + fetchable.push(task); + if (fetchable.length > limit) { + throw new Error(`result count exceeds ${limitName} (${limit})`); + } + } + return fetchable; +} + /** * Every result body lives at a presigned `result_url` — no auth header, no API content * endpoint in the loop, matching the Python SDK's approach exactly. */ async function fetchResultBody(task: TaskResult, maxBytes?: number): Promise { if (!task.result_url) { - throw new Error(`task ${task.task_id} has no result_url (not a successful task?)`); + const why = + task.status === "expired" + ? "its result is past the retention window (status expired)" + : `it has no stored result (status ${task.status})`; + throw new Error(`task ${task.task_id} has no result_url: ${why}`); } const response = await fetch(task.result_url); if (!response.ok) { @@ -75,9 +125,16 @@ export interface DownloadToDirOptions { concurrency?: number; maxFiles?: number; maxBytesPerFile?: number; + /** Called for each `expired` or `failed` row skipped because it has no body to fetch. */ + onSkip?: (row: SkippedResult) => void; } -/** Write every task body in `results` to `targetDir`, one file per task. */ +/** + * Write every task body in `results` to `targetDir`, one file per task, and return how many + * files it wrote. `expired` and `failed` rows have no body: they are skipped (and passed to + * `onSkip`) instead of failing the download. Any other row without a `result_url` still + * fails it. + */ export async function downloadToDir( results: AsyncIterable, targetDir: string, @@ -87,16 +144,11 @@ export async function downloadToDir( concurrency = 1, maxFiles = DEFAULT_MAX_FILES, maxBytesPerFile = DEFAULT_MAX_BYTES_PER_FILE, + onSkip, }: DownloadToDirOptions = {}, ): Promise { await mkdir(targetDir, { recursive: true }); - const tasks: TaskResult[] = []; - for await (const task of results) { - tasks.push(task); - if (tasks.length > maxFiles) { - throw new Error(`result count exceeds maxFiles (${maxFiles})`); - } - } + const tasks = await partition(results, maxFiles, "maxFiles", onSkip); await runPool(tasks, concurrency, async (task: TaskResult) => { const body = await fetchResultBody(task, maxBytesPerFile); @@ -114,9 +166,15 @@ export interface DownloadToMemoryOptions { maxCount?: number; maxTotalBytes?: number; maxBytesPerFile?: number; + /** Called for each `expired` or `failed` row skipped because it has no body to fetch. */ + onSkip?: (row: SkippedResult) => void; } -/** Load every task body in `results` into memory. */ +/** + * Load every task body in `results` into memory. `expired` and `failed` rows have no body: + * they are skipped (and passed to `onSkip`). Any other row without a `result_url` still + * fails the download. + */ export async function downloadToMemory( results: AsyncIterable, { @@ -124,15 +182,10 @@ export async function downloadToMemory( maxCount = DEFAULT_MAX_COUNT_IN_MEMORY, maxTotalBytes = DEFAULT_MAX_TOTAL_BYTES_IN_MEMORY, maxBytesPerFile = DEFAULT_MAX_BYTES_PER_FILE, + onSkip, }: DownloadToMemoryOptions = {}, ): Promise { - const tasks: TaskResult[] = []; - for await (const task of results) { - tasks.push(task); - if (tasks.length > maxCount) { - throw new Error(`result count exceeds maxCount (${maxCount})`); - } - } + const tasks = await partition(results, maxCount, "maxCount", onSkip); const downloaded: DownloadedResult[] = new Array(tasks.length); let totalBytes = 0; diff --git a/src/batch/errors.ts b/src/batch/errors.ts index 7a2e25b..d7417ee 100644 --- a/src/batch/errors.ts +++ b/src/batch/errors.ts @@ -1,17 +1,118 @@ import type { ProblemJson } from "./types.js"; +/** + * The `code` values the Batch API sends in an error body, by name. Compare + * `ZenRowsBatchError.code` (or a task's `error.code`) against these instead of a bare string. + * The API may add codes, so keep a default branch. + */ +export const BatchErrorCode = { + // Generic + /** 400. The request is invalid; a bad task is listed in `extras.invalid_tasks`. */ + InvalidArgument: "invalid_argument", + /** 404. */ + NotFound: "not_found", + /** 409. */ + Conflict: "conflict", + /** 401. The API key is missing or invalid. */ + Unauthenticated: "unauthenticated", + /** 401. The key is valid but the subscription is not active; fix it in Billing. */ + SubscriptionInactive: "subscription_inactive", + /** 402. The account is out of credits. */ + PaymentRequired: "payment_required", + /** 402. The API key reached one of its credit caps; `detail` names the cap and when it resets. */ + ApiKeyCapReached: "api_key_cap_reached", + /** 403. */ + PermissionDenied: "permission_denied", + /** + * 429. An account limit: active jobs (shared with Crawl), scheduled jobs, or the task cap of + * an open job. It clears only when a job finishes or is deleted, so it is not retried. + */ + QuotaExceeded: "quota_exceeded", + /** 500. */ + Internal: "internal", + /** 501. */ + NotImplemented: "not_implemented", + + // Job, run and task lifecycle (409 unless noted) + JobNotOpen: "job_not_open", + RunNotTerminal: "run_not_terminal", + RunNotStoppable: "run_not_stoppable", + NoPriorRun: "no_prior_run", + /** A rerun with a status filter matched no task. */ + NoMatchingTasks: "no_matching_tasks", + RunHasDescendants: "run_has_descendants", + LastBatchReceived: "last_batch_received", + /** The same `Idempotency-Key` was sent with a different body. */ + IdempotencyKeyConflict: "idempotency_key_conflict", + /** The task is still pending or processing, so it has no content yet. */ + TaskNotReady: "task_not_ready", + /** 410. The result is past the retention window. */ + ResultExpired: "result_expired", + /** A partial rerun (`status` filter) more than 10 days after the run started; a full rerun still works. */ + PartialRerunWindowExpired: "partial_rerun_window_expired", + + // HMAC signing keys (409) + CandidatePending: "candidate_pending", + NoCandidate: "no_candidate", + + // Input validation (400, usually per task in `invalid_tasks`) + UnsupportedScheme: "unsupported_scheme", + MalformedUrl: "malformed_url", + MissingHost: "missing_host", + UrlTooLong: "url_too_long", + MetadataTooLarge: "metadata_too_large", + MetadataKeyInvalid: "metadata_key_invalid", + InvalidExternalId: "invalid_external_id", + + // Task method and body (400) + UnsupportedMethod: "unsupported_method", + BodyNotAllowed: "body_not_allowed", + BodyTooLarge: "body_too_large", + BodyMalformed: "body_malformed", + MethodParamConflict: "method_param_conflict", + + // CSV file inputs + /** 404. */ + FileInputNotFound: "file_input_not_found", + /** 409. */ + FileInputNotUploaded: "file_input_not_uploaded", + /** 409. */ + FileInputUnsupportedForJobType: "file_input_unsupported_for_job_type", + + // Webhooks (400) + CallbackUrlMustBeHttps: "callback_url_must_be_https", + CallbackUrlUnresolvable: "callback_url_unresolvable", + WebhookSigningRequiresActiveKey: "webhook_signing_requires_active_key", + + // Task level: seen in a failed task's `error.code` + GatewayUnreachable: "gateway_unreachable", +} as const; + +/** One of the `BatchErrorCode` values. */ +export type BatchErrorCodeValue = (typeof BatchErrorCode)[keyof typeof BatchErrorCode]; + /** * A non-2xx response from the Batch API, decoded as RFC 7807 `application/problem+json` - * where possible. `code` is the stable Problem `code` member (e.g. `file_input_not_found`) — - * safe to branch on; defaults to `"internal"` when the body wasn't valid Problem JSON. + * where possible. `code` is the stable Problem `code` member (e.g. `file_input_not_found`), + * safe to branch on against `BatchErrorCode`; it is `"internal"` when the body wasn't valid + * Problem JSON. */ export class ZenRowsBatchError extends Error { readonly status: number; readonly problem: ProblemJson | undefined; readonly code: string; readonly extras: Record | undefined; + /** The Problem `detail`: a human-readable explanation, when the body had one. */ + readonly detail: string | undefined; + /** Seconds from the `Retry-After` header, when the response had one. */ + readonly retryAfter: number | undefined; - constructor(status: number, problem: ProblemJson | undefined, extras?: Record) { + constructor( + status: number, + problem: ProblemJson | undefined, + extras?: Record, + retryAfter?: number, + ) { const code = problem?.code ?? "internal"; const message = problem ? `${status} ${problem.title ?? "Error"}: ${problem.detail ?? code}` @@ -22,6 +123,8 @@ export class ZenRowsBatchError extends Error { this.problem = problem && { ...problem, code }; this.code = code; this.extras = extras; + this.detail = problem?.detail; + this.retryAfter = retryAfter; } } diff --git a/src/batch/estimate.ts b/src/batch/estimate.ts index 8848154..94ae1e0 100644 --- a/src/batch/estimate.ts +++ b/src/batch/estimate.ts @@ -21,8 +21,6 @@ const TIER_ORDER: Tier[] = [ "auto", ]; -const TRUTHY_STRINGS = new Set(["true", "1", "yes", "on"]); - export type ParamValue = string | boolean | number | Record; export type ParamMap = Record; @@ -46,19 +44,19 @@ export interface CostEstimate { exact: boolean; } +/** + * A flag is on exactly when the Batch API would read it as on: JSON `true`, or a string + * equal to "true" in any case. The API rejects other spellings ("1", "yes", numbers) with + * a 400 at submit, so they never bill; they count as off here. + */ function truthy(value: ParamValue | undefined): boolean { - if (typeof value === "boolean") return value; - if (typeof value === "number") return value !== 0; - if (typeof value === "string") return TRUTHY_STRINGS.has(value.trim().toLowerCase()); - return false; + if (value === true) return true; + return typeof value === "string" && value.toLowerCase() === "true"; } +/** The API accepts only the exact value "auto" for `mode`. */ function isAuto(params: ParamMap): boolean { - return ( - String(params.mode ?? "") - .trim() - .toLowerCase() === "auto" - ); + return params.mode === "auto"; } function costForParams(params: ParamMap): { tier: Tier; min: number; max: number } { @@ -91,8 +89,13 @@ function taskParams(task: TaskLike): ParamMap { /** * Estimate the credit cost of a job before submitting it. `tasks` accepts the same shapes - * `submitRegular` does: bare URL strings or task objects. Per-task `zenrows_params` override - * the job-level `zenrowsParams` on key collision (task wins), matching the worker's merge. + * `submitRegular` does: bare URL strings or task objects. Each task is priced on the + * job-level `zenrowsParams` overlaid with its own `zenrows_params` (the task wins on a key + * collision), the same merge the API applies. + * + * Scrape params set directly on a task object (a flat `js_render` next to `url`) are not + * counted: the API has no such task field and ignores it, so it doesn't change the price. + * Put per-task params under `zenrows_params`. */ export function estimateCost(tasks: Iterable, zenrowsParams?: ParamMap): CostEstimate { const jobParams = zenrowsParams ?? {}; diff --git a/src/batch/resources.ts b/src/batch/resources.ts index 96a50f5..c490d7a 100644 --- a/src/batch/resources.ts +++ b/src/batch/resources.ts @@ -17,6 +17,7 @@ import type { StartExportResponse, SubmitJobResponse, TaskHistoryResponse, + TaskMethod, TaskResult, WebhookConfig, } from "./types.js"; @@ -26,6 +27,10 @@ interface BatchTaskLike { external_id?: string; metadata?: Record; zenrows_params?: Record; + /** HTTP method for the target fetch. Default `GET`. */ + method?: TaskMethod; + /** Request body for a `POST` task: a JSON value or a string, up to 16 KiB. */ + body?: unknown; } // ======================= specific runs ======================= @@ -96,6 +101,8 @@ export class RunRef { failureStatuses?: Set; timeout?: number; pollInterval?: number; + /** Keep polling while the run is paused instead of throwing `RunPausedError`. */ + waitWhilePaused?: boolean; } = {}, ): Promise { const run = await this.#client.waitForRun(this.jobId, { ...options, runId: this.runId }); @@ -270,6 +277,8 @@ export class CurrentRun { failureStatuses?: Set; timeout?: number; pollInterval?: number; + /** Keep polling while the run is paused instead of throwing `RunPausedError`. */ + waitWhilePaused?: boolean; } = {}, ): Promise { const run = await this.#client.waitForRun(this.jobId, options); @@ -415,24 +424,52 @@ export class JobRef { await this.#client._delete(this.jobId); } + /** + * Start a new run. Without `status` every task runs again; with it (for example + * `["failed", "pending"]`) only tasks in those states run and the rest are inherited. + * + * The API answers every rerun with 202 while it writes the new run's task rows, so + * `results()` read straight after a rerun can be partial. Pass `waitForIngest: true` to + * resolve only once ingestion is done (it polls up to 300 s), or call `waitForIngest()`. + */ async rerun( - options: { status?: string | string[]; idempotencyKey?: string } = {}, + options: { + status?: string | string[]; + idempotencyKey?: string; + waitForIngest?: boolean; + } = {}, ): Promise { - const resp = await this.#client._postRerun(this.jobId, options); - return new RunHandle(this.#client, this.jobId, resp.latest_run.run_id, resp.latest_run); + const { waitForIngest, ...rest } = options; + const resp = await this.#client._postRerun(this.jobId, rest); + let run = resp.latest_run; + if (waitForIngest && run.ingest_status === "pending") { + const job = await this.#client._waitForIngestRaw(this.jobId); + if (job.latest_run?.run_id === run.run_id) run = job.latest_run; + } + return new RunHandle(this.#client, this.jobId, run.run_id, run); } /** Shortcut for `rerun({ status: "failed" })` (or `"failed,pending"` with `includePending`). */ retryFailed( - options: { includePending?: boolean; idempotencyKey?: string } = {}, + options: { includePending?: boolean; idempotencyKey?: string; waitForIngest?: boolean } = {}, ): Promise { const status = options.includePending ? ["failed", "pending"] : "failed"; - return this.rerun({ status, idempotencyKey: options.idempotencyKey }); + return this.rerun({ + status, + idempotencyKey: options.idempotencyKey, + waitForIngest: options.waitForIngest, + }); } + /** + * Append tasks to an open job (up to 10,000 per call). With an `idempotencyKey`, a repeat of + * the same call (same key, same body) returns the original answer instead of adding the + * tasks again, so the SDK retries it on a transient failure. Without one, every call appends + * its tasks and a failed call is not retried. + */ addTasks( tasks: BatchTaskLike[], - options: { lastBatch?: boolean } = {}, + options: { lastBatch?: boolean; idempotencyKey?: string } = {}, ): Promise<{ accepted_tasks: number; job_status: Extensible; diff --git a/src/batch/transport.ts b/src/batch/transport.ts index afd221a..1fe48dc 100644 --- a/src/batch/transport.ts +++ b/src/batch/transport.ts @@ -43,8 +43,27 @@ export type ErrorFactory = ( extras: Record | undefined, ) => Error; -const batchError: ErrorFactory = (response, problem, extras) => - new ZenRowsBatchError(response.status, problem, extras); +const batchError: ErrorFactory = (response, problem, extras) => { + const raw = response.headers.get("Retry-After"); + const seconds = raw ? Number(raw) : Number.NaN; + const retryAfter = Number.isFinite(seconds) && seconds >= 0 ? seconds : undefined; + return new ZenRowsBatchError(response.status, problem, extras, retryAfter); +}; + +/** + * The Batch API's 429 `quota_exceeded` is an account limit (active jobs, scheduled jobs, or + * an open job's task cap). It clears only when a job finishes or is deleted, so retrying it + * only adds delay. Reads a clone, so the body stays available for the error. + */ +async function isQuotaExceeded(response: Response): Promise { + if (response.status !== 429) return false; + try { + const body = (await response.clone().json()) as { code?: unknown } | null; + return body?.code === "quota_exceeded"; + } catch { + return false; + } +} export interface RequestOptions { query?: object; @@ -177,7 +196,12 @@ export class BatchTransport { throw error; } - if (idempotent && attempt < this.retries && retryStatuses.has(response.status)) { + if ( + idempotent && + attempt < this.retries && + retryStatuses.has(response.status) && + !(await isQuotaExceeded(response)) + ) { // An unread body holds its connection until garbage collection. response.body?.cancel().catch(() => {}); clear(); diff --git a/src/batch/types.ts b/src/batch/types.ts index 3a475b3..dde629a 100644 --- a/src/batch/types.ts +++ b/src/batch/types.ts @@ -10,10 +10,19 @@ export type JobType = "regular" | "scheduled"; export type JobStatus = "open" | "closed" | "deleted"; export type ScheduleState = "active" | "paused"; export type RunStatus = "pending" | "running" | "completed" | "failed" | "stopped" | "deleted"; -export type TaskStatus = "pending" | "processing" | "successful" | "failed"; +/** + * `expired`: a successful task whose result is past the retention window (14 days by + * default). The row has no `result_url`, and its content endpoint answers 410 + * `result_expired`. A `status: "successful"` filter still matches expired rows. + */ +export type TaskStatus = "pending" | "processing" | "successful" | "failed" | "expired"; export type PauseState = "active" | "paused"; export type IngestStatus = "pending" | "done"; export type ExportStatus = "pending" | "running" | "completed" | "failed"; +/** What started a run: `manual` (a submit or rerun through the API) or `scheduled` (a schedule fired). */ +export type RunTrigger = "manual" | "scheduled"; +/** HTTP method a task fetches its URL with. `GET` is the default. */ +export type TaskMethod = "GET" | "POST"; /** * Why a run auto-failed (`status == "failed"`). * - `insufficient_credits` — the account is out of credits. @@ -31,6 +40,7 @@ export interface Spend { cost: number; } +/** @deprecated Results listings never return per-task spend; see `TaskHistoryEvent.spend`. */ export interface TaskSpend { total: Spend; last_attempt: Spend; @@ -65,6 +75,8 @@ export interface Run { job_id: string; run_sequence: number; status: Extensible; + /** What started this run. */ + trigger?: Extensible; stats: RunStats; last_batch_received?: boolean; pause_state?: PauseState; @@ -142,12 +154,20 @@ export interface TaskResult { run_id: string; url: string; metadata?: Record; - method?: "GET" | "POST"; + method?: TaskMethod; status: Extensible; type?: string; result_url?: string; error?: ProblemJson; source_run_id?: string; + /** When the task row was created (RFC 3339). */ + created_at?: string; + /** When the task row last changed, for example when an attempt settled (RFC 3339). */ + updated_at?: string; + /** + * @deprecated Never set: results listings do not return per-task spend. Read per-attempt + * spend from `getTaskHistory()`, or the run total from `run.stats.spend`. + */ spend?: TaskSpend; } @@ -225,3 +245,30 @@ export interface TestWebhookResponse { error?: string; elapsed_ms: number; } + +/** Event types a Batch webhook delivers. `webhook.test` comes only from `testWebhook()`. */ +export type WebhookEventType = "run.completed" | "run.failed" | "webhook.test"; + +/** + * The JSON body of a Batch webhook delivery (`schema: "zenrows.webhook/v1"`). Verify the + * raw body with `verifyWebhookSignature()` before parsing it, and deduplicate deliveries on + * `event_id` (also sent as the `X-ZenRows-Event-Id` header): a delivery can repeat. + */ +export interface WebhookEvent { + schema: Extensible<"zenrows.webhook/v1">; + event_id: string; + event_type: Extensible; + /** When the event was queued (RFC 3339). */ + occurred_at: string; + job_id: string; + external_id?: string; + metadata?: Record; + run_id: string; + run_sequence: number; + /** The run's counters when the event was queued. */ + stats?: RunStats; + /** Set on `run.failed` only. */ + failure_reason?: Extensible; + /** Set on `run.failed` only. */ + failure_detail?: string; +} diff --git a/src/batch/waiters.ts b/src/batch/waiters.ts index 417f123..272d2e6 100644 --- a/src/batch/waiters.ts +++ b/src/batch/waiters.ts @@ -1,3 +1,5 @@ +import type { Run } from "./types.js"; + /** Raised when a waiter's `timeout` elapsed before the target state was reached. */ export class WaiterTimeoutError extends Error { constructor(message: string) { @@ -14,6 +16,21 @@ export class WaiterFailureError extends Error { } } +/** + * Raised by a run waiter when the run is paused (`pause_state: "paused"`) and not yet in a + * target status. A paused run makes no progress until it is resumed, so the waiter stops + * instead of polling until its timeout. `run` is the run as last read. + */ +export class RunPausedError extends Error { + readonly run: Run; + + constructor(run: Run) { + super(`waiter: run ${run.run_id} is paused; resume it with job.run.resume() and wait again`); + this.name = "RunPausedError"; + this.run = run; + } +} + export interface PollUntilOptions { isDone(value: T): boolean; isFailure?(value: T): boolean; diff --git a/src/batch/webhook.ts b/src/batch/webhook.ts new file mode 100644 index 0000000..78f7a76 --- /dev/null +++ b/src/batch/webhook.ts @@ -0,0 +1,103 @@ +import { createHmac, timingSafeEqual } from "node:crypto"; + +/** The request header that carries a signed delivery's signature. */ +export const WEBHOOK_SIGNATURE_HEADER = "X-Signature"; + +/** The request header that carries a delivery's event id. Deduplicate deliveries on it. */ +export const WEBHOOK_EVENT_ID_HEADER = "X-ZenRows-Event-Id"; + +/** How far, in seconds, a signature's timestamp may be from now by default. */ +export const DEFAULT_WEBHOOK_TOLERANCE_SECONDS = 300; + +/** The API issues every webhook signing key as 32 random bytes. */ +const WEBHOOK_KEY_BYTES = 32; + +/** + * Decode a base64 signing secret, or return undefined when it isn't one the API could have + * issued. Node's base64 decoder skips characters it doesn't recognise, so "!!!" or "====" + * decode to an empty buffer, and an HMAC keyed with that would accept a delivery anyone can + * sign. Require exactly 32 bytes that re-encode to the same text (ignoring `=` padding). + */ +function decodeSecret(secret: string): Buffer | undefined { + const key = Buffer.from(secret, "base64"); + if (key.length !== WEBHOOK_KEY_BYTES) return undefined; + const canonical = (text: string) => text.replace(/=+$/, ""); + if (canonical(key.toString("base64")) !== canonical(secret)) return undefined; + return key; +} + +export interface VerifyWebhookSignatureOptions { + /** + * Reject a delivery whose `t` is further than this from `now`, in seconds. Default 300. + * `0` turns the check off. Anything that isn't a finite number of 0 or more makes every + * delivery fail to verify. + */ + toleranceSeconds?: number; + /** + * The current time as a Unix timestamp in seconds. Defaults to the system clock. A value + * that isn't a finite number makes every delivery fail to verify. + */ + now?: number; +} + +function parseSignatureHeader(header: string): { t?: string; v1?: string; kid?: string } { + const parts: { t?: string; v1?: string; kid?: string } = {}; + for (const piece of header.split(",")) { + const index = piece.indexOf("="); + if (index <= 0) continue; + const key = piece.slice(0, index).trim(); + const value = piece.slice(index + 1).trim(); + if (key === "t" || key === "v1" || key === "kid") { + parts[key] = value; + } + } + return parts; +} + +/** + * Check a Batch webhook delivery's `X-Signature` header. + * + * The header reads `t=,v1=,kid=`. The signature covers + * `t`, a literal `.`, then the raw request body, keyed with the secret `rotateHmacKey()` + * returned. That secret is base64; this helper decodes it to the key bytes. + * + * Pass the body exactly as received, before any JSON parsing. `secrets` maps each key id to + * its base64 secret, so a receiver accepts both the old and the new key during a rotation; the + * header's `kid` picks the one to check. The comparison is constant-time. + * + * Returns `false` for a missing or malformed header, an unknown `kid`, a secret that isn't a + * 32-byte base64 key, a timestamp outside the tolerance, or a wrong signature. + * Deliveries can repeat: deduplicate on the `X-ZenRows-Event-Id` header. + */ +export function verifyWebhookSignature( + rawBody: string | Uint8Array, + signatureHeader: string | null | undefined, + secrets: Record, + options: VerifyWebhookSignatureOptions = {}, +): boolean { + const { t, v1, kid } = parseSignatureHeader(signatureHeader ?? ""); + // v1 is a hex SHA-256 digest: exactly 64 hex characters. Anything else is malformed + // (Buffer.from(hex) would otherwise drop a trailing odd nibble). + if (!t || !v1 || !kid || !/^\d+$/.test(t) || !/^[0-9a-fA-F]{64}$/.test(v1)) return false; + + const secret = Object.prototype.hasOwnProperty.call(secrets, kid) ? secrets[kid] : undefined; + if (!secret) return false; + + // A bad option fails closed: NaN compares false, so it would otherwise skip the age check. + const tolerance = options.toleranceSeconds ?? DEFAULT_WEBHOOK_TOLERANCE_SECONDS; + if (!Number.isFinite(tolerance) || tolerance < 0) return false; + if (options.now !== undefined && !Number.isFinite(options.now)) return false; + if (tolerance > 0) { + const now = options.now ?? Math.floor(Date.now() / 1000); + if (Math.abs(now - Number(t)) > tolerance) return false; + } + + const key = decodeSecret(secret); + if (!key) return false; + const body = typeof rawBody === "string" ? Buffer.from(rawBody, "utf8") : Buffer.from(rawBody); + const expected = createHmac("sha256", key) + .update(Buffer.concat([Buffer.from(`${t}.`, "utf8"), body])) + .digest(); + const given = Buffer.from(v1, "hex"); + return given.length === expected.length && timingSafeEqual(given, expected); +} diff --git a/tests/batch-client.test.ts b/tests/batch-client.test.ts index 9626478..71854ff 100644 --- a/tests/batch-client.test.ts +++ b/tests/batch-client.test.ts @@ -116,12 +116,90 @@ describe("ZenRowsBatchClient — webhooks", () => { const current = await client.getJobWebhook("job_123"); expect(current).toMatchObject({ url: "https://example.com/hook", signature: true }); - const replaced = await client.putJobWebhook("job_123", { url: "https://example.com/hook2" }); - expect(replaced).toMatchObject({ url: "https://example.com/hook2", signature: false }); + let sent: unknown; + server.use( + http.put("https://async.api.zenrows.com/v1/jobs/job_123/webhook", async ({ request }) => { + sent = await request.json(); + return HttpResponse.json({ url: "https://example.com/hook2", signature: true }); + }), + ); + const replaced = await client.putJobWebhook("job_123", { + url: "https://example.com/hook2", + signature: true, + }); + expect(sent).toEqual({ url: "https://example.com/hook2", signature: true }); + expect(replaced).toMatchObject({ url: "https://example.com/hook2", signature: true }); await expect(client.deleteJobWebhook("job_123")).resolves.toBeUndefined(); }); + test("putJobWebhook without signature keeps the job's current signing setting", async () => { + const sent: unknown[] = []; + let current: { url: string; signature: boolean } | undefined = { + url: "https://example.com/hook", + signature: true, + }; + server.use( + http.get("https://async.api.zenrows.com/v1/jobs/job_wh/webhook", () => + current + ? HttpResponse.json(current) + : HttpResponse.json( + { type: "about:blank", title: "Not found", status: 404, code: "not_found" }, + { status: 404, headers: { "Content-Type": "application/problem+json" } }, + ), + ), + http.put("https://async.api.zenrows.com/v1/jobs/job_wh/webhook", async ({ request }) => { + const body = (await request.json()) as { url: string; signature: boolean }; + sent.push(body); + return HttpResponse.json(body); + }), + ); + + // Signed webhook, URL changed without saying `signature`: signing stays on. + await client.putJobWebhook("job_wh", { url: "https://example.com/new" }); + // No webhook yet: the API's submit-time default, unsigned. + current = undefined; + await client.putJobWebhook("job_wh", { url: "https://example.com/first" }); + // An explicit value is sent as given. + await client.putJobWebhook("job_wh", { url: "https://example.com/x", signature: false }); + + expect(sent).toEqual([ + { url: "https://example.com/new", signature: true }, + { url: "https://example.com/first", signature: false }, + { url: "https://example.com/x", signature: false }, + ]); + }); + + test("putJobWebhook without signature is a read then a write, not an atomic update", async () => { + // Documents a known limit: another caller turns signing on between this call's GET and + // its PUT, and the PUT writes back the value it read. Pass `signature` to avoid it. + const state = { url: "https://example.com/hook", signature: false }; + let gets = 0; + server.use( + http.get("https://async.api.zenrows.com/v1/jobs/job_race/webhook", () => { + gets += 1; + const seen = { ...state }; + state.signature = true; // the concurrent update lands right after this read + return HttpResponse.json(seen); + }), + http.put("https://async.api.zenrows.com/v1/jobs/job_race/webhook", async ({ request }) => { + Object.assign(state, await request.json()); + return HttpResponse.json(state); + }), + ); + + await client.putJobWebhook("job_race", { url: "https://example.com/new" }); + expect(state).toEqual({ url: "https://example.com/new", signature: false }); + expect(gets).toBe(1); + + // With `signature` passed, there is no read, so nothing to race with. + for (const signature of [false, true]) { + await client.putJobWebhook("job_race", { url: "https://example.com/new", signature }); + expect(state.signature).toBe(signature); + } + expect(gets).toBe(1); + }); + test("dispatches a synthetic test event", async () => { const result = await client.testWebhook({ url: "https://example.com/hook" }); expect(result).toMatchObject({ delivered: true, status_code: 200 }); @@ -327,3 +405,139 @@ describe("ZenRowsBatchClient — current-run pause/resume", () => { expect(resumed.data.pause_state).toBe("active"); }); }); + +describe("ZenRowsBatchClient: content, add-tasks idempotency, results paging, timeout", () => { + test("getTaskContentBytes returns the bytes untouched, getTaskContent returns text", async () => { + const pdf = Buffer.from([0x25, 0x50, 0x44, 0x46, 0xff, 0x00, 0xe2, 0x9c]); + server.use( + http.get( + "https://async.api.zenrows.com/v1/jobs/job_pdf/tasks/t1/content", + () => new HttpResponse(pdf, { headers: { "Content-Type": "application/pdf" } }), + ), + ); + const client = new ZenRowsBatchClient("API_KEY"); + const bytes = await client.getTaskContentBytes("job_pdf", "t1"); + expect(Buffer.isBuffer(bytes)).toBe(true); + expect(bytes.equals(pdf)).toBe(true); + + const text = await client.getTaskContent("job_123", "task_1"); + expect(text).toBe("latest run content"); + }); + + test("addTasks sends Idempotency-Key and is retried on a transient failure", async () => { + const keys: (string | null)[] = []; + server.use( + http.post("https://async.api.zenrows.com/v1/jobs/job_open/tasks", ({ request }) => { + keys.push(request.headers.get("Idempotency-Key")); + if (keys.length === 1) return new HttpResponse(null, { status: 503 }); + return HttpResponse.json({ + accepted_tasks: 1, + job_status: "open", + latest_run: { run_id: "r", job_id: "job_open", run_sequence: 1, status: "running" }, + }); + }), + ); + const client = new ZenRowsBatchClient("API_KEY"); + const resp = await client + .job("job_open") + .addTasks([{ url: "https://example.com" }], { idempotencyKey: "batch-7" }); + expect(resp.accepted_tasks).toBe(1); + expect(keys).toEqual(["batch-7", "batch-7"]); + }); + + test("addTasks without a key is not retried", async () => { + let calls = 0; + server.use( + http.post("https://async.api.zenrows.com/v1/jobs/job_open/tasks", () => { + calls += 1; + return new HttpResponse(null, { status: 503 }); + }), + ); + const client = new ZenRowsBatchClient("API_KEY"); + await expect( + client.job("job_open").addTasks([{ url: "https://example.com" }]), + ).rejects.toBeInstanceOf(ZenRowsBatchError); + expect(calls).toBe(1); + }); + + test("getResults does not send limit, which the results endpoint ignores", async () => { + let query: URLSearchParams | undefined; + server.use( + http.get("https://async.api.zenrows.com/v1/jobs/job_q/results", ({ request }) => { + query = new URL(request.url).searchParams; + return HttpResponse.json({ results: [], next_cursor: null }); + }), + ); + const client = new ZenRowsBatchClient("API_KEY"); + await client.getResults("job_q", { limit: 50, status: "failed", cursor: "c1" }); + expect(query?.has("limit")).toBe(false); + expect(query?.get("status")).toBe("failed"); + expect(query?.get("cursor")).toBe("c1"); + }); + + test("requests time out after 30 seconds unless `timeout` says otherwise", () => { + // biome-ignore lint/suspicious/noExplicitAny: reading the private transport to prove it + const timeoutOf = (client: ZenRowsBatchClient) => (client as any).transport.timeoutMs; + expect(timeoutOf(new ZenRowsBatchClient("API_KEY"))).toBe(30_000); + expect(timeoutOf(new ZenRowsBatchClient("API_KEY", { timeout: 5 }))).toBe(5_000); + }); + + test("a request that takes longer than `timeout` seconds is aborted", async () => { + server.use( + http.get("https://async.api.zenrows.com/v1/jobs/job_slow", async () => { + await new Promise((resolve) => setTimeout(resolve, 500)); + return HttpResponse.json({ job_id: "job_slow" }); + }), + ); + const client = new ZenRowsBatchClient("API_KEY", { timeout: 0.05, retries: 0 }); + await expect(client.getJob("job_slow")).rejects.toThrow(/timed out/); + }); +}); + +describe("ZenRowsBatchClient: task inputs, run-scoped content, webhook read errors", () => { + test("submitRegular sends string and object tasks, including method and body", async () => { + let sent: { tasks?: unknown[] } | undefined; + server.use( + http.post("https://async.api.zenrows.com/v1/jobs", async ({ request }) => { + sent = (await request.json()) as { tasks?: unknown[] }; + return HttpResponse.json({ job_id: "job_in", status: "closed", accepted_tasks: 2 }); + }), + ); + const client = new ZenRowsBatchClient("API_KEY"); + await client.submitRegular([ + "https://example.com/a", + { url: "https://example.com/search", method: "POST", body: { q: "shoes" } }, + ]); + expect(sent?.tasks).toEqual([ + { url: "https://example.com/a" }, + { url: "https://example.com/search", method: "POST", body: { q: "shoes" } }, + ]); + }); + + test("getTaskContentBytes reads an explicit run's content", async () => { + const client = new ZenRowsBatchClient("API_KEY"); + const bytes = await client.getTaskContentBytes("job_123", "task_1", { runId: "run_1" }); + expect(bytes.toString("utf8")).toBe("run_1 content"); + }); + + test("putJobWebhook without signature surfaces a read error other than 404", async () => { + let puts = 0; + server.use( + http.get("https://async.api.zenrows.com/v1/jobs/job_denied/webhook", () => + HttpResponse.json( + { type: "about:blank", title: "Forbidden", status: 403, code: "permission_denied" }, + { status: 403, headers: { "Content-Type": "application/problem+json" } }, + ), + ), + http.put("https://async.api.zenrows.com/v1/jobs/job_denied/webhook", () => { + puts += 1; + return HttpResponse.json({}); + }), + ); + const client = new ZenRowsBatchClient("API_KEY"); + await expect( + client.putJobWebhook("job_denied", { url: "https://example.com/hook" }), + ).rejects.toMatchObject({ status: 403 }); + expect(puts).toBe(0); + }); +}); diff --git a/tests/batch-consistency.test.ts b/tests/batch-consistency.test.ts new file mode 100644 index 0000000..f686092 --- /dev/null +++ b/tests/batch-consistency.test.ts @@ -0,0 +1,306 @@ +import { http, HttpResponse } from "msw"; +import { describe, expect, expectTypeOf, test } from "vitest"; +import { + At, + BatchErrorCode, + type BatchErrorCodeValue, + type Run, + RunPausedError, + type TaskResult, + type TaskStatus, + WaiterTimeoutError, + type WebhookEvent, + ZenRowsBatchClient, + ZenRowsBatchError, +} from "../src"; +import { server } from "./_setup"; + +const API = "https://async.api.zenrows.com/v1"; + +function problem(status: number, code: string, headers: Record = {}) { + return HttpResponse.json( + { type: `https://docs.zenrows.com/api-error-codes#${code}`, title: "Error", status, code }, + { status, headers: { "Content-Type": "application/problem+json", ...headers } }, + ); +} + +describe("429 quota_exceeded", () => { + test("is an account limit, so it is not retried", async () => { + let calls = 0; + server.use( + http.get(`${API}/jobs/job_quota`, () => { + calls += 1; + return problem(429, "quota_exceeded"); + }), + ); + const client = new ZenRowsBatchClient("API_KEY"); + const error = await client.getJob("job_quota").catch((e) => e); + expect(error).toBeInstanceOf(ZenRowsBatchError); + expect(error.code).toBe(BatchErrorCode.QuotaExceeded); + expect(calls).toBe(1); + }); + + test("is not retried on a submit that carries an Idempotency-Key either", async () => { + let calls = 0; + server.use( + http.post(`${API}/jobs`, () => { + calls += 1; + return problem(429, "quota_exceeded"); + }), + ); + const client = new ZenRowsBatchClient("API_KEY"); + await expect( + client.submitRegular(["https://example.com"], undefined, { idempotencyKey: "k1" }), + ).rejects.toMatchObject({ status: 429, code: "quota_exceeded" }); + expect(calls).toBe(1); + }); + + test("another 429 is still retried", async () => { + let calls = 0; + server.use( + http.get(`${API}/jobs/job_busy`, () => { + calls += 1; + if (calls === 1) return problem(429, "too_many_requests", { "Retry-After": "0" }); + return HttpResponse.json({ job_id: "job_busy", type: "regular", status: "closed" }); + }), + ); + const client = new ZenRowsBatchClient("API_KEY"); + await expect(client.getJob("job_busy")).resolves.toMatchObject({ jobId: "job_busy" }); + expect(calls).toBe(2); + }); +}); + +describe("ZenRowsBatchError fields", () => { + test("exposes detail and Retry-After seconds", async () => { + server.use( + http.get(`${API}/jobs/job_detail`, () => + HttpResponse.json( + { + type: "https://docs.zenrows.com/api-error-codes#api_key_cap_reached", + title: "Payment Required", + status: 402, + code: "api_key_cap_reached", + detail: "monthly cap reached", + }, + { + status: 402, + headers: { "Content-Type": "application/problem+json", "Retry-After": "120" }, + }, + ), + ), + ); + const client = new ZenRowsBatchClient("API_KEY"); + const error = (await client.getJob("job_detail").catch((e) => e)) as ZenRowsBatchError; + expect(error.code).toBe(BatchErrorCode.ApiKeyCapReached); + expect(error.detail).toBe("monthly cap reached"); + expect(error.retryAfter).toBe(120); + }); + + test("BatchErrorCode names the codes the API sends", () => { + expect(BatchErrorCode.SubscriptionInactive).toBe("subscription_inactive"); + expect(BatchErrorCode.PaymentRequired).toBe("payment_required"); + expect(BatchErrorCode.ResultExpired).toBe("result_expired"); + expect(BatchErrorCode.PartialRerunWindowExpired).toBe("partial_rerun_window_expired"); + expect(BatchErrorCode.IdempotencyKeyConflict).toBe("idempotency_key_conflict"); + // Every value is unique and snake_case. + const values = Object.values(BatchErrorCode); + expect(new Set(values).size).toBe(values.length); + for (const value of values) expect(value).toMatch(/^[a-z]+(_[a-z]+)*$/); + expectTypeOf<"quota_exceeded">().toMatchTypeOf(); + }); +}); + +describe("content of an expired result", () => { + test("getTaskContent throws 410 result_expired", async () => { + server.use( + http.get(`${API}/jobs/job_old/tasks/t1/content`, () => problem(410, "result_expired")), + ); + const client = new ZenRowsBatchClient("API_KEY"); + await expect(client.getTaskContent("job_old", "t1")).rejects.toMatchObject({ + status: 410, + code: BatchErrorCode.ResultExpired, + }); + }); +}); + +describe("waitForIngest on rerun and submitScheduled", () => { + test("rerun({ waitForIngest: true }) returns the run once its rows are written", async () => { + let jobReads = 0; + server.use( + http.post(`${API}/jobs/job_rr/rerun`, () => + HttpResponse.json( + { + job_id: "job_rr", + status: "closed", + latest_run: { run_id: "run_2", status: "pending", ingest_status: "pending" }, + retried_tasks: 3, + inherited_tasks: 0, + }, + { status: 202 }, + ), + ), + http.get(`${API}/jobs/job_rr`, () => { + jobReads += 1; + return HttpResponse.json({ + job_id: "job_rr", + type: "regular", + status: "closed", + latest_run: { run_id: "run_2", status: "running", ingest_status: "done" }, + }); + }), + ); + const client = new ZenRowsBatchClient("API_KEY"); + const run = await client.job("job_rr").retryFailed({ waitForIngest: true }); + expect(run.runId).toBe("run_2"); + expect(run.data.ingest_status).toBe("done"); + expect(jobReads).toBe(1); + + // Without the option, no extra read. + await client.job("job_rr").rerun(); + expect(jobReads).toBe(1); + }); + + test("submitScheduled passes waitForIngest through", async () => { + let jobReads = 0; + server.use( + http.post(`${API}/jobs`, () => + HttpResponse.json( + { + job_id: "job_sched", + status: "closed", + accepted_tasks: 10_000, + latest_run: { run_id: "run_1", status: "pending", ingest_status: "pending" }, + }, + { status: 202 }, + ), + ), + http.get(`${API}/jobs/job_sched`, () => { + jobReads += 1; + return HttpResponse.json({ + job_id: "job_sched", + type: "scheduled", + status: "closed", + latest_run: { run_id: "run_1", status: "running", ingest_status: "done" }, + }); + }), + ); + const client = new ZenRowsBatchClient("API_KEY"); + await client.submitScheduled( + new At("2030-01-01T00:00:00", "UTC"), + ["https://example.com"], + undefined, + { + waitForIngest: true, + }, + ); + expect(jobReads).toBe(1); + }); +}); + +describe("types follow the wire shapes", () => { + test("task, run and webhook event fields", () => { + expectTypeOf<"expired">().toMatchTypeOf(); + expectTypeOf().toEqualTypeOf(); + expectTypeOf().toEqualTypeOf(); + expectTypeOf<"scheduled">().toMatchTypeOf>(); + const event: WebhookEvent = { + schema: "zenrows.webhook/v1", + event_id: "evt_1", + event_type: "run.failed", + occurred_at: "2026-01-01T00:00:00Z", + job_id: "j1", + run_id: "r1", + run_sequence: 1, + failure_reason: "insufficient_credits", + }; + expect(event.event_type).toBe("run.failed"); + }); +}); + +describe("waiting on a paused run", () => { + function pausedJob(jobId: string, counter: { reads: number }) { + return http.get(`${API}/jobs/${jobId}`, () => { + counter.reads += 1; + return HttpResponse.json({ + job_id: jobId, + type: "regular", + status: "closed", + latest_run: { + run_id: "run_p", + job_id: jobId, + run_sequence: 1, + status: "running", + pause_state: "paused", + stats: { total: 2, completed: 1, successful: 1, failed: 0 }, + }, + }); + }); + } + + test("stops at once with RunPausedError carrying the run", async () => { + const counter = { reads: 0 }; + server.use(pausedJob("job_paused", counter)); + const client = new ZenRowsBatchClient("API_KEY"); + const error = await client + .job("job_paused") + .run.wait({ timeout: 5 }) + .catch((e) => e); + expect(error).toBeInstanceOf(RunPausedError); + expect((error as RunPausedError).run.pause_state).toBe("paused"); + expect((error as RunPausedError).run.run_id).toBe("run_p"); + expect(counter.reads).toBe(1); + }); + + test("a specific run's wait stops on pause too", async () => { + server.use( + http.get(`${API}/jobs/job_paused2/runs/run_p`, () => + HttpResponse.json({ + run_id: "run_p", + job_id: "job_paused2", + run_sequence: 1, + status: "pending", + pause_state: "paused", + stats: { total: 1, completed: 0, successful: 0, failed: 0 }, + }), + ), + ); + const client = new ZenRowsBatchClient("API_KEY"); + await expect(client.run("job_paused2", "run_p").wait()).rejects.toBeInstanceOf(RunPausedError); + }); + + test("waitWhilePaused keeps polling until the timeout", async () => { + const counter = { reads: 0 }; + server.use(pausedJob("job_paused3", counter)); + const client = new ZenRowsBatchClient("API_KEY"); + await expect( + client.waitForRun("job_paused3", { + waitWhilePaused: true, + timeout: 0.05, + pollInterval: 0.01, + }), + ).rejects.toBeInstanceOf(WaiterTimeoutError); + expect(counter.reads).toBeGreaterThan(1); + }); + + test("a paused run already in a target status resolves", async () => { + server.use( + http.get(`${API}/jobs/job_paused4`, () => + HttpResponse.json({ + job_id: "job_paused4", + type: "regular", + status: "closed", + latest_run: { + run_id: "run_p", + job_id: "job_paused4", + run_sequence: 1, + status: "stopped", + pause_state: "paused", + stats: { total: 1, completed: 1, successful: 1, failed: 0 }, + }, + }), + ), + ); + const client = new ZenRowsBatchClient("API_KEY"); + await expect(client.waitForRun("job_paused4")).resolves.toMatchObject({ status: "stopped" }); + }); +}); diff --git a/tests/batch-download.test.ts b/tests/batch-download.test.ts index b942c74..5bd9a5f 100644 --- a/tests/batch-download.test.ts +++ b/tests/batch-download.test.ts @@ -1,9 +1,10 @@ import { mkdtemp, readFile, readdir, rm } from "node:fs/promises"; import { tmpdir } from "node:os"; import { join } from "node:path"; +import { http, HttpResponse } from "msw"; import { afterEach, beforeEach, describe, expect, test } from "vitest"; import { ZenRowsBatchClient } from "../src/batch/client"; -import "./_setup"; +import { server } from "./_setup"; describe("downloadToDir / downloadToMemory / single-task download", () => { const client = new ZenRowsBatchClient("API_KEY"); @@ -70,4 +71,158 @@ describe("downloadToDir / downloadToMemory / single-task download", () => { }), ).rejects.toThrow(/no result_url/); }); + + describe("rows without a result_url", () => { + // An `expired` row (past the retention window) still matches `status=successful` but has + // no result_url; a `failed` row has none either when the filter includes it. + const rows = (extra: object[] = []) => + server.use( + http.get("https://async.api.zenrows.com/v1/jobs/job_mixed/results", () => + HttpResponse.json({ + results: [ + { + task_id: "task_a", + run_id: "r", + url: "https://example.com/a", + status: "successful", + result_url: "https://storage.example.test/body-a", + }, + { + task_id: "task_old", + external_id: "ext_old", + run_id: "r", + url: "https://example.com/old", + status: "expired", + }, + { + task_id: "task_bad", + run_id: "r", + url: "https://example.com/bad", + status: "failed", + }, + ...extra, + ], + next_cursor: null, + }), + ), + ); + + test("downloadToDir skips expired and failed rows and reports them to onSkip", async () => { + rows(); + const skipped: unknown[] = []; + const count = await client.downloadToDir("job_mixed", undefined, dir, { + status: "all", + onSkip: (row) => skipped.push(row), + }); + expect(count).toBe(1); + expect(skipped).toEqual([ + { + taskId: "task_old", + externalId: "ext_old", + url: "https://example.com/old", + status: "expired", + }, + { + taskId: "task_bad", + externalId: undefined, + url: "https://example.com/bad", + status: "failed", + }, + ]); + expect(await readdir(dir)).toEqual(["task_a"]); + }); + + test("downloadToMemory skips them and reports them to onSkip", async () => { + rows(); + const skipped: unknown[] = []; + const results = await client.downloadToMemory("job_mixed", undefined, { + status: "all", + onSkip: (row) => skipped.push(row), + }); + expect(results.map((r) => r.taskId)).toEqual(["task_a"]); + expect(skipped).toEqual([ + { + taskId: "task_old", + externalId: "ext_old", + url: "https://example.com/old", + status: "expired", + }, + { + taskId: "task_bad", + externalId: undefined, + url: "https://example.com/bad", + status: "failed", + }, + ]); + }); + + test("onSkip is optional", async () => { + rows(); + const results = await client.downloadToMemory("job_mixed", undefined, { status: "all" }); + expect(results.map((r) => r.taskId)).toEqual(["task_a"]); + }); + + test("a successful row with no result_url still fails the download", async () => { + rows([ + { task_id: "task_odd", run_id: "r", url: "https://example.com/odd", status: "successful" }, + ]); + const skipped: unknown[] = []; + await expect( + client.downloadToDir("job_mixed", undefined, dir, { + status: "all", + onSkip: (row) => skipped.push(row), + }), + ).rejects.toThrow(/task_odd has no result_url/); + expect(skipped).toHaveLength(2); + await expect( + client.downloadToMemory("job_mixed", undefined, { status: "all" }), + ).rejects.toThrow(/task_odd has no result_url/); + }); + + test("an unfinished row with no result_url still fails the download", async () => { + rows([ + { task_id: "task_wip", run_id: "r", url: "https://example.com/wip", status: "processing" }, + ]); + await expect( + client.downloadToMemory("job_mixed", undefined, { status: "all" }), + ).rejects.toThrow(/task_wip has no result_url/); + }); + + test("a single-task download of an expired row says why it has no body", async () => { + await expect( + client.downloadTaskToMemory({ + task_id: "task_old", + run_id: "r", + url: "https://example.com/old", + status: "expired", + }), + ).rejects.toThrow(/retention window/); + }); + }); + + test("maxFiles counts only the rows it will fetch, and still caps them", async () => { + server.use( + http.get("https://async.api.zenrows.com/v1/jobs/job_cap/results", () => + HttpResponse.json({ + results: [ + { task_id: "x", run_id: "r", url: "https://example.com/x", status: "expired" }, + ...["a", "b"].map((id) => ({ + task_id: id, + run_id: "r", + url: `https://example.com/${id}`, + status: "successful", + result_url: `https://storage.example.test/body-${id}`, + })), + ], + next_cursor: null, + }), + ), + ); + await expect(client.downloadToDir("job_cap", undefined, dir, { maxFiles: 1 })).rejects.toThrow( + "result count exceeds maxFiles (1)", + ); + await expect( + client.downloadToMemory("job_cap", undefined, { maxCount: 2 }), + ).resolves.toHaveLength(2); + }); }); diff --git a/tests/batch-estimate.test.ts b/tests/batch-estimate.test.ts index 30d03ae..5c724de 100644 --- a/tests/batch-estimate.test.ts +++ b/tests/batch-estimate.test.ts @@ -33,10 +33,46 @@ describe("estimateCost", () => { expect(result.breakdown[0]?.tier).toBe("auto"); }); - test("truthy string spellings ('1', 'yes', 'ON') count as on", () => { - expect(estimateCost([{ zenrows_params: { js_render: "yes" } }]).min).toBe(5); - expect(estimateCost([{ zenrows_params: { premium_proxy: "1" } }]).min).toBe(10); - expect(estimateCost([{ zenrows_params: { js_render: "ON" } }]).min).toBe(5); + test('a flag is on for JSON true or the string "true" in any case', () => { + expect(estimateCost([{ zenrows_params: { js_render: "true" } }]).min).toBe(5); + expect(estimateCost([{ zenrows_params: { js_render: "TRUE" } }]).min).toBe(5); + expect(estimateCost([{ zenrows_params: { premium_proxy: "True" } }]).min).toBe(10); + }); + + test("spellings the API rejects at submit count as off", () => { + // The API accepts only true/false for these flags and answers 400 for anything else, + // so none of these would ever bill as JS rendering or premium proxy. + for (const value of ["1", "yes", "on", " true", 1, 2]) { + expect(estimateCost([{ zenrows_params: { js_render: value } }]).min).toBe(1); + expect(estimateCost([{ zenrows_params: { premium_proxy: value } }]).min).toBe(1); + } + }); + + test("mode counts as auto only for the exact value the API accepts", () => { + expect(estimateCost([{ zenrows_params: { mode: "auto" } }]).max).toBe(25); + expect(estimateCost([{ zenrows_params: { mode: "AUTO" } }]).max).toBe(1); + }); + + test("scrape params set flat on a task are not counted (the API ignores them)", () => { + expect(estimateCost([{ url: "https://a.com", js_render: true }]).min).toBe(1); + expect(estimateCost([{ url: "https://a.com", premium_proxy: "true" }], {}).min).toBe(1); + }); + + test("job params, then the task's zenrows_params", () => { + const result = estimateCost( + [ + "https://a.com", + { url: "https://b.com", zenrows_params: { premium_proxy: true } }, + { url: "https://c.com", zenrows_params: { js_render: "false" } }, + ], + { js_render: true }, + ); + expect(result.breakdown).toEqual([ + { tier: "base", count: 1, unitMin: 1, unitMax: 1 }, + { tier: "js_render", count: 1, unitMin: 5, unitMax: 5 }, + { tier: "js_render+premium_proxy", count: 1, unitMin: 25, unitMax: 25 }, + ]); + expect(result.min).toBe(31); }); test("falsy string spellings and zero do not count as on", () => { diff --git a/tests/batch-webhook.test.ts b/tests/batch-webhook.test.ts new file mode 100644 index 0000000..9bdf518 --- /dev/null +++ b/tests/batch-webhook.test.ts @@ -0,0 +1,147 @@ +import { createHmac, randomBytes } from "node:crypto"; +import { describe, expect, test } from "vitest"; +import { WEBHOOK_EVENT_ID_HEADER, WEBHOOK_SIGNATURE_HEADER, verifyWebhookSignature } from "../src"; + +// A delivery signed the way the Batch API signs: HMAC-SHA256 over `t + "." + raw body`, keyed +// with the RAW key bytes, while the secret handed to the customer is those bytes in base64. +// The header was produced by a copy of the API's own Go signing code (fmt "%d.%s" into an +// HMAC, then "t=%d,v1=%s,kid=%s"), so the test does not just mirror the helper. +const SECRET_B64 = "AAECAwQFBgcICQoLDA0ODxAREhMUFRYXGBkaGxwdHh8="; +const T = 1760000000; +const BODY = + '{"schema":"zenrows.webhook/v1","event_id":"evt_1","event_type":"run.completed","job_id":"j1","run_id":"r1","run_sequence":1,"stats":{"total":2,"completed":2,"successful":2,"failed":0}}'; +const SIG = "1cb06c874dbfb275a921703f7d5ca5b649ad6543952534feab4cb3b111673b17"; +const HEADER = `t=${T},v1=${SIG},kid=k1`; +const secrets = { k1: SECRET_B64 }; + +describe("verifyWebhookSignature", () => { + test("accepts a genuine delivery (string or bytes body)", () => { + expect(verifyWebhookSignature(BODY, HEADER, secrets, { now: T })).toBe(true); + expect(verifyWebhookSignature(Buffer.from(BODY), HEADER, secrets, { now: T + 10 })).toBe(true); + }); + + test("rejects a tampered body or signature", () => { + expect( + verifyWebhookSignature(BODY.replace('"failed":0', '"failed":1'), HEADER, secrets, { now: T }), + ).toBe(false); + const flipped = `t=${T},v1=${SIG.slice(0, -1)}0,kid=k1`; + expect(verifyWebhookSignature(BODY, flipped, secrets, { now: T })).toBe(false); + }); + + test("rejects the mistake of keying the HMAC with the base64 text itself", () => { + const wrong = createHmac("sha256", SECRET_B64).update(`${T}.${BODY}`).digest("hex"); + expect(verifyWebhookSignature(BODY, `t=${T},v1=${wrong},kid=k1`, secrets, { now: T })).toBe( + false, + ); + }); + + test("picks the secret by kid, so the old and new key both verify during a rotation", () => { + const raw2 = randomBytes(32); + const sig2 = createHmac("sha256", raw2).update(`${T}.${BODY}`).digest("hex"); + const both = { k1: SECRET_B64, k2: raw2.toString("base64") }; + expect(verifyWebhookSignature(BODY, HEADER, both, { now: T })).toBe(true); + expect(verifyWebhookSignature(BODY, `t=${T},v1=${sig2},kid=k2`, both, { now: T })).toBe(true); + expect(verifyWebhookSignature(BODY, `t=${T},v1=${sig2},kid=k3`, both, { now: T })).toBe(false); + }); + + test("rejects a stale timestamp unless the tolerance is turned off", () => { + expect(verifyWebhookSignature(BODY, HEADER, secrets, { now: T + 301 })).toBe(false); + expect(verifyWebhookSignature(BODY, HEADER, secrets, { now: T - 301 })).toBe(false); + expect( + verifyWebhookSignature(BODY, HEADER, secrets, { now: T + 301, toleranceSeconds: 600 }), + ).toBe(true); + expect( + verifyWebhookSignature(BODY, HEADER, secrets, { now: T + 86_400, toleranceSeconds: 0 }), + ).toBe(true); + }); + + test("returns false for a malformed header", () => { + for (const header of [ + "", + "garbage", + `t=${T},kid=k1`, + `v1=${SIG},kid=k1`, + `t=abc,v1=${SIG},kid=k1`, + // A signature with one extra hex digit, or one missing, is not a SHA-256 digest. + `t=${T},v1=${SIG}f,kid=k1`, + `t=${T},v1=${SIG.slice(0, -1)},kid=k1`, + `t=${T},v1=${SIG.slice(0, -2)},kid=k1`, + ]) { + expect(verifyWebhookSignature(BODY, header, secrets, { now: T })).toBe(false); + } + }); + + test("rejects a delivery forged with an empty key when the secret decodes to nothing", () => { + // Node's base64 decoder drops characters it doesn't know, so these decode to zero bytes. + // An HMAC keyed with zero bytes is one anyone can compute. + const forgedBody = '{"forged":true}'; + const forged = createHmac("sha256", Buffer.alloc(0)).update(`${T}.${forgedBody}`).digest("hex"); + for (const bad of ["!!!", "====", "", " "]) { + expect( + verifyWebhookSignature(forgedBody, `t=${T},v1=${forged},kid=k1`, { k1: bad }, { now: T }), + ).toBe(false); + } + }); + + test("rejects a secret that is not exactly 32 bytes of canonical base64", () => { + const short = Buffer.alloc(16, 7); + const shortSig = createHmac("sha256", short).update(`${T}.${BODY}`).digest("hex"); + expect( + verifyWebhookSignature( + BODY, + `t=${T},v1=${shortSig},kid=k1`, + { k1: short.toString("base64") }, + { now: T }, + ), + ).toBe(false); + // The right 32 bytes, but with junk Node would silently skip: not the text the API issued. + expect(verifyWebhookSignature(BODY, HEADER, { k1: `${SECRET_B64}!!` }, { now: T })).toBe(false); + expect(verifyWebhookSignature(BODY, HEADER, { k1: `!${SECRET_B64}` }, { now: T })).toBe(false); + }); + + test("accepts the issued secret with or without its = padding", () => { + expect( + verifyWebhookSignature(BODY, HEADER, { k1: SECRET_B64.replace(/=+$/, "") }, { now: T }), + ).toBe(true); + }); + + test("invalid timestamp options fail closed instead of skipping the age check", () => { + const dayOld = T + 86_400; + expect(verifyWebhookSignature(BODY, HEADER, secrets, { now: dayOld })).toBe(false); + for (const toleranceSeconds of [ + -1, + Number.NaN, + Number.POSITIVE_INFINITY, + Number.NEGATIVE_INFINITY, + ]) { + expect(verifyWebhookSignature(BODY, HEADER, secrets, { now: dayOld, toleranceSeconds })).toBe( + false, + ); + expect(verifyWebhookSignature(BODY, HEADER, secrets, { now: T, toleranceSeconds })).toBe( + false, + ); + } + for (const now of [Number.NaN, Number.POSITIVE_INFINITY, Number.NEGATIVE_INFINITY]) { + expect(verifyWebhookSignature(BODY, HEADER, secrets, { now })).toBe(false); + } + }); + + test("uses the system clock when `now` is not given", () => { + const t = Math.floor(Date.now() / 1000); + const raw = Buffer.from(SECRET_B64, "base64"); + const sig = createHmac("sha256", raw).update(`${t}.${BODY}`).digest("hex"); + expect(verifyWebhookSignature(BODY, `t=${t},v1=${sig},kid=k1`, secrets)).toBe(true); + // The fixed fixture is from 2025, far outside the default 5 minutes of the real clock. + expect(verifyWebhookSignature(BODY, HEADER, secrets)).toBe(false); + }); + + test("a missing header is rejected, not thrown", () => { + expect(verifyWebhookSignature(BODY, undefined, secrets, { now: T })).toBe(false); + expect(verifyWebhookSignature(BODY, null, secrets, { now: T })).toBe(false); + }); + + test("names the headers a delivery carries", () => { + expect(WEBHOOK_SIGNATURE_HEADER).toBe("X-Signature"); + expect(WEBHOOK_EVENT_ID_HEADER).toBe("X-ZenRows-Event-Id"); + }); +}); From 22c1c757650d54db7cbb05c5625bbd5d5e25e1fb Mon Sep 17 00:00:00 2001 From: Emirlan Rasulov Date: Fri, 16 Oct 2026 16:31:51 +0600 Subject: [PATCH 2/3] docs(batch): rewrite the README Batch section against the real client The old example called methods the client does not have (addTasks, closeJob, rerun on client.batch) and read job.job_id from a JobRef, so it did not compile. It also said schedules, webhooks, HMAC keys, CSV uploads and exports were not wrapped. The section now uses the JobRef API, documents retention, errors and retries, paused runs, and webhook verification, and a test typechecks every code block in it against the source. --- README.md | 199 ++++++++++++++++++++++++++++++------- tests/readme-batch.test.ts | 70 +++++++++++++ 2 files changed, 233 insertions(+), 36 deletions(-) create mode 100644 tests/readme-batch.test.ts diff --git a/README.md b/README.md index 3aa3b62..7ff5ee3 100644 --- a/README.md +++ b/README.md @@ -17,7 +17,10 @@ SDK to access [Zenrows](https://www.zenrows.com/) API directly from Node.js. Zen - [POST Requests](#post-requests) - [Extract](#extract) - [Batch](#batch) + - [Results and retention](#results-and-retention) - [Failed runs and extensible values](#failed-runs-and-extensible-values) + - [Errors and retries](#errors-and-retries) + - [Webhooks](#webhooks) - [Extract in a batch](#extract-in-a-batch) - [Crawl](#crawl) - [Concurrency](#concurrency) @@ -201,56 +204,174 @@ A `422` with `code: "no_data_extracted"` means Extract knows this kind of page b ### Batch -The [Batch API](https://docs.zenrows.com) runs many URLs asynchronously as a job, and is reachable via `client.batch`. A job is either **open** (it keeps accepting tasks via `addTasks()` until you `closeJob()` it, useful when you're streaming URLs in over time) or a normal one-shot batch where every task is known upfront. +The [Batch API](https://docs.zenrows.com/batch/introduction) runs many URLs asynchronously as a job, and is reachable via `client.batch`. A job is either a one-shot batch where every task is known upfront (`submitRegular()`), or **open** (`submitOpen()`): it keeps accepting tasks via `addTasks()` until you close it, which is useful when you're streaming URLs in over time. Batch costs the same credits as the regular API. -```javascript -const { ZenRows } = require("zenrows"); +`submitRegular()`, `submitOpen()` and `submitScheduled()` return a `JobRef`. Work with the job through it: `job.addTasks()`, `job.close()`, `job.rerun()`, `job.retryFailed()`, `job.delete()`, and the current run under `job.run` (`wait()`, `results()`, `downloadToDir()`, `stop()`, `pause()`, `resume()`). -const apiKey = "YOUR-API-KEY"; +```ts +import { ZenRows } from "zenrows"; -(async () => { - const client = new ZenRows(apiKey); +const client = new ZenRows("YOUR-API-KEY"); - // One-shot batch: every task known upfront. - const job = await client.batch.submitJob({ - tasks: [{ url: "https://example.com/1" }, { url: "https://example.com/2" }], - }); +// One-shot batch: every task known upfront. Strings and { url, ... } objects both work. +const job = await client.batch.submitRegular( + [ + "https://example.com/1", + { url: "https://example.com/2", external_id: "page-2" }, + // A POST task: `method` plus a `body` (a JSON value or a string, up to 16 KiB). + { url: "https://example.com/search", method: "POST", body: { q: "shoes" } }, + ], + undefined, + // A retried submit with the same key returns the first job instead of creating another. + { idempotencyKey: "catalog-2026-10-16" }, +); - // Streaming batch: keep the job open, add tasks as they arrive, close when done. - const streamingJob = await client.batch.submitJob({ status: "open" }); - await client.batch.addTasks(streamingJob.job_id, [{ url: "https://example.com/3" }]); - // Or close it as part of the last addTasks() call instead of a separate request: - // await client.batch.addTasks(streamingJob.job_id, [...], { lastBatch: true }); - await client.batch.closeJob(streamingJob.job_id); +// Wait for the run to end, then read its results. +const run = await job.run.wait(); +if (run.status === "failed") { + // The run stopped for an account reason, such as running out of credits. + console.log(run.data.failure_reason, run.data.failure_detail); +} +for await (const row of job.run.results({ status: "successful" })) { + console.log(row.task_id, row.url, row.status, row.result_url); +} + +// Or save every successful result to disk. Rows with no body to fetch (an `expired` result, +// or a `failed` task) are skipped and passed to `onSkip`. +const saved = await job.run.downloadToDir("./results", { + onSkip: (row) => console.log("skipped", row.taskId, row.status), +}); +console.log(`saved ${saved} files`); + +// Retry only the tasks that failed, inheriting the rest from the previous run. With no +// failed task there is nothing to retry: the API answers 409 `no_matching_tasks`. +if (run.stats.failed > 0) { + const retry = await job.retryFailed({ waitForIngest: true }); + await retry.wait(); +} + +// Streaming batch: keep the job open, add tasks as they arrive, close with the last call. +const streaming = await client.batch.submitOpen(); +// With an idempotencyKey, a retried call adds its tasks once, so the SDK retries it safely. +await streaming.addTasks([{ url: "https://example.com/3" }], { idempotencyKey: "chunk-1" }); +await streaming.addTasks([{ url: "https://example.com/4" }], { + idempotencyKey: "chunk-2", + lastBatch: true, +}); + +// Pick up an existing job later by its id. +const sameJob = await client.batch.getJob(job.jobId); +console.log(sameJob.status, sameJob.data.latest_run?.stats); +``` - // Poll for progress and page through results once it's done. - const status = await client.batch.getJob(job.job_id); - const { results, next_cursor } = await client.batch.getResults(job.job_id); +The client also wraps the rest of the API: - // Retry only the tasks that failed, inheriting the rest from the previous run. - await client.batch.rerun(job.job_id, { status: "failed" }); -})(); -``` +- Jobs and runs: `getJob()`, `listJobs()` / `iterJobs()`, `job.runs()`, `getRun()`, `run(jobId, runId).delete()`, `waitForRun()`. +- Results: `getResults()` / `iterResults()`, `downloadToMemory()`, `downloadTaskToFile()`, `downloadTaskToMemory()`, `getTaskHistory()`, `getTaskContent()` (the content as text) and `getTaskContentBytes()` (a `Buffer`). Text decoding corrupts binary bodies, so read a `pdf` or other binary result with `getTaskContentBytes()`. +- Scheduling: `submitScheduled()` with the `At`, `Rate` and `Calendar` builders, and `job.schedule.pause()` / `resume()` / `update()`. +- Webhooks: `job.getWebhook()`, `job.setWebhook(url, signature)`, `putJobWebhook()`, `job.deleteWebhook()`, `testWebhook()`, and `verifyWebhookSignature()` for your receiver (see [Webhooks](#webhooks)). +- HMAC signing keys: `listHmacKeys()`, `rotateHmacKey()`, `finalizeHmacKey()`, `cancelHmacRotation()`. +- CSV task uploads: `uploadCsv()`, then `submitRegular(undefined, fileInputId)`. +- Results exports: `job.run.startExport()`, `waitForExport()`, `downloadAllResults()`. +- Cost: `estimateCost(tasks, zenrowsParams)` prices each task on the job params overlaid with its own `zenrows_params`, locally, with no API call. A flag counts as on only when it is `true` or `"true"`, the values the API accepts. + +A submit of 10,000 tasks or more, and every rerun, returns while the API is still writing the run's task rows (`ingest_status: "pending"`). Pass `waitForIngest: true` to `submitRegular()`, `submitOpen()`, `submitScheduled()`, `rerun()` or `retryFailed()` to resolve only once the rows are written; until then `results()` can be partial. -`client.batch` also exposes `listJobs()`, `deleteJob()`, `stopRun()`, `rerun()`, `listRuns()`, `getRun()`, `deleteRun()`, and `getTaskContent()` (returns the scraped page's raw content as a string, not JSON — the endpoint can return HTML or plain text depending on what the target page served). Scheduling, webhook config, HMAC key rotation, CSV task uploads, and results exports aren't wrapped yet — call the [Batch API](https://docs.zenrows.com) directly for those. +#### Results and retention + +A result's `result_url` is valid for 2 hours. To get a fresh one, list the results again or read the content with `getTaskContent()`. Results are kept for 14 days by default. After that a successful task comes back with `status: "expired"` and no `result_url`, and `getTaskContent()` throws a `ZenRowsBatchError` with status 410 and code `result_expired`. A `status: "successful"` filter still matches expired rows, which is why the bulk downloads skip them. #### Failed runs and extensible values -A run the server auto-fails carries `failure_reason` and a human-readable `failure_detail`. Starting, rerunning or resuming a job whose API key hit one of its credit caps throws a `ZenRowsBatchError` with `status` 402 and `code` `"api_key_cap_reached"`; `error.problem.detail` names the cap and when it resets (UTC). +A run that stops for an account reason ends `failed`, with `failure_reason` (`insufficient_credits`, `subscription_inactive` or `api_key_cap_reached`) and a human-readable `failure_detail`. `job.run.wait()` resolves with that run; pass `failureStatuses: new Set(["failed"])` to make it throw instead. A paused run makes no progress until it is resumed, so the wait stops at once and throws a `RunPausedError` whose `run` is the paused run; pass `waitWhilePaused: true` to keep polling, for example when another process will resume it. A wait that runs out of time throws a `WaiterTimeoutError`. + +```ts +import { ZenRowsBatchClient } from "zenrows"; -```js -const run = await client.batch.getRun(jobId, runId); +const batch = new ZenRowsBatchClient("YOUR-API-KEY"); +const run = await batch.getRun("YOUR-JOB-ID", "YOUR-RUN-ID"); if (run.data.failure_reason === "api_key_cap_reached") console.log(run.data.failure_detail); ``` -Enum-valued fields on responses (run, job, task and export `status`, `ingest_status`, `failure_reason`) are typed `Extensible<...>`: the known values plus any `string`, because the server may add values. The exported named types (`RunStatus`, `JobStatus`, `TaskStatus`, `ExportStatus`, `IngestStatus`, `RunFailureReason`) stay closed, so they still work as request input. Give switches on these fields a default branch; an exhaustive `switch` that assigns the leftover to `never` no longer compiles. +Enum-valued fields on responses (run, job, task and export `status`, `ingest_status`, `failure_reason`, run `trigger`) are typed `Extensible<...>`: the known values plus any `string`, because the server may add values. The exported named types (`RunStatus`, `JobStatus`, `TaskStatus`, `ExportStatus`, `IngestStatus`, `RunFailureReason`, `RunTrigger`) stay closed, so they still work as request input. Give switches on these fields a default branch; an exhaustive `switch` that assigns the leftover to `never` no longer compiles. + +#### Errors and retries + +A non-2xx answer throws a `ZenRowsBatchError` with `status`, `code`, `detail` and `retryAfter` (seconds, when the API sent `Retry-After`). `BatchErrorCode` names every code, so you can branch without bare strings: + +```ts +import { BatchErrorCode, ZenRowsBatchClient, ZenRowsBatchError } from "zenrows"; + +const batch = new ZenRowsBatchClient("YOUR-API-KEY"); +try { + await batch.submitRegular(["https://example.com"]); +} catch (error) { + if (!(error instanceof ZenRowsBatchError)) throw error; + switch (error.code) { + case BatchErrorCode.PaymentRequired: // 402: the account is out of credits + case BatchErrorCode.SubscriptionInactive: // 401: the subscription is not active + console.log("Check Billing in your dashboard:", error.detail); + break; + case BatchErrorCode.ApiKeyCapReached: // 402: `detail` names the cap and when it resets (UTC) + console.log(error.detail); + break; + case BatchErrorCode.QuotaExceeded: // 429: an account limit, see below + console.log("Wait for a job to finish, or delete one:", error.detail); + break; + default: + throw error; + } +} +``` + +`quota_exceeded` means one of the account limits: 3 active jobs (shared with Crawl), 5 scheduled jobs, or 100,000 tasks in an open job. It clears only when a job finishes or is deleted, so the client never retries it. Other transient failures (network errors, 502, 503, 504, and any other 429) are retried up to `retries` times (default 3) with backoff, on reads and on the POSTs that carry an `idempotencyKey`: submit, rerun and `addTasks()`. The API keeps a key for 24 hours: the same key with the same body returns the original answer, and the same key with a different body is a 409 `idempotency_key_conflict`. + +#### Webhooks + +A job with a webhook gets one event when a run completes (`run.completed`) or fails on its own (`run.failed`, for example when the account runs out of credits). The event carries the run's stats, see the `WebhookEvent` type. Stopping a run yourself sends no event. A delivery that fails or takes longer than 10 seconds is retried with backoff for up to 3 days, so the same event can arrive more than once: deduplicate on the `X-ZenRows-Event-Id` header. + +With `signature: true`, each delivery has an `X-Signature` header, `t=,v1=,kid=`. `verifyWebhookSignature()` checks it against the secret `rotateHmacKey()` returned (base64, shown once), and rejects deliveries more than 5 minutes old by default. During a key rotation, pass both key ids in `secrets`; the header's `kid` picks the one to check. Pass the raw body, before any JSON parsing: + +```ts +import { createServer } from "node:http"; +import { + WEBHOOK_EVENT_ID_HEADER, + WEBHOOK_SIGNATURE_HEADER, + type WebhookEvent, + verifyWebhookSignature, +} from "zenrows"; + +// The `kid` and base64 `secret` that rotateHmacKey() returned. Load them from your secret store. +const secrets = { "YOUR-KEY-ID": "YOUR-BASE64-SECRET" }; + +createServer((req, res) => { + const chunks: Buffer[] = []; + req.on("data", (chunk: Buffer) => chunks.push(chunk)); + req.on("end", () => { + const body = Buffer.concat(chunks); + const signature = req.headers[WEBHOOK_SIGNATURE_HEADER.toLowerCase()]; + if (typeof signature !== "string" || !verifyWebhookSignature(body, signature, secrets)) { + res.writeHead(401).end(); + return; + } + const event = JSON.parse(body.toString("utf8")) as WebhookEvent; + console.log(req.headers[WEBHOOK_EVENT_ID_HEADER.toLowerCase()], event.event_type, event.stats); + res.writeHead(204).end(); + }); +}).listen(3000); +``` + +Changing a webhook with `putJobWebhook(jobId, { url, signature })`: pass `signature` every time. If you leave it out, the SDK reads the job's current setting and sends it back, so changing only the URL keeps signing as it was; that is a read then a write, so a change someone else makes in between is overwritten. #### Extract in a batch -Set `extract` in the batch params to run tasks through Extract — structured data instead of raw HTML. It works job-wide or per task, and per-task values win on collision. +Set `extract` in the batch params to run tasks through Extract: structured data instead of raw HTML. It works job-wide or per task, and per-task values win on collision. `extract_fields` is accepted the same way. -```js -const job = await client.batch.submitRegular( +```ts +import { ZenRowsBatchClient } from "zenrows"; + +const batch = new ZenRowsBatchClient("YOUR-API-KEY"); +const job = await batch.submitRegular( [ // Inherits the job-level params below. { url: "https://example.com/products", external_id: "p1" }, @@ -260,16 +381,22 @@ const job = await client.batch.submitRegular( undefined, { zenrowsParams: { extract: "auto" } }, ); +console.log(job.jobId); ``` -An Extract task's result carries two keys — `html` (the raw page) and `parsed` (the structured data). It costs the same as a regular task, so `estimateCost` prices it correctly. `extract_fields` is not supported in Batch yet. +An Extract task's result carries two keys: `html` (the raw page) and `parsed` (the structured data). `estimateCost` prices it like a regular task. -The batch client (`ZenRowsBatchClient`) also works standalone, without a `ZenRows` instance — matching the Go and Python SDKs' batch clients: +The batch client (`ZenRowsBatchClient`) also works standalone, without a `ZenRows` instance, matching the Go and Python SDKs' batch clients: -```javascript -const { ZenRowsBatchClient } = require("zenrows"); +```ts +import { ZenRowsBatchClient } from "zenrows"; -const batch = new ZenRowsBatchClient(apiKey, { baseURL: "https://async.api.zenrows.com/v1" }); // baseURL is optional +// Both settings are optional. `timeout` is seconds per API request (default 30). +const batch = new ZenRowsBatchClient("YOUR-API-KEY", { + baseURL: "https://async.api.zenrows.com/v1", + timeout: 30, +}); +console.log(batch.apiKey.length > 0); ``` ### Crawl diff --git a/tests/readme-batch.test.ts b/tests/readme-batch.test.ts new file mode 100644 index 0000000..24540b1 --- /dev/null +++ b/tests/readme-batch.test.ts @@ -0,0 +1,70 @@ +import { mkdtempSync, readFileSync, rmSync, writeFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join, resolve } from "node:path"; +import ts from "typescript"; +import { describe, expect, test } from "vitest"; + +// The README's Batch examples are what people copy. Typecheck every ```ts block in the +// Batch section against the source, with `zenrows` resolving to src/index.ts, so an example +// that calls a method the client doesn't have fails CI. + +const ROOT = resolve(__dirname, ".."); + +function batchSnippets(): string[] { + const readme = readFileSync(join(ROOT, "README.md"), "utf8"); + const start = readme.indexOf("\n### Batch\n"); + const end = readme.indexOf("\n### Crawl\n", start); + if (start < 0 || end < 0) throw new Error("README has no Batch section before Crawl"); + const section = readme.slice(start, end); + return [...section.matchAll(/```ts\n([\s\S]*?)```/g)].map((match) => match[1] ?? ""); +} + +describe("README Batch section", () => { + test("every code block is TypeScript, so all of them get typechecked", () => { + const readme = readFileSync(join(ROOT, "README.md"), "utf8"); + const section = readme.slice(readme.indexOf("\n### Batch\n"), readme.indexOf("\n### Crawl\n")); + const fences = [...section.matchAll(/^```(\w*)$/gm)].map((match) => match[1]); + const openings = fences.filter((_, index) => index % 2 === 0); + expect(openings.length).toBeGreaterThan(0); + expect(openings.every((lang) => lang === "ts")).toBe(true); + }); + + test("its examples compile against the client", () => { + const snippets = batchSnippets(); + const dir = mkdtempSync(join(tmpdir(), "zenrows-readme-")); + try { + const files = snippets.map((code, index) => { + const file = join(dir, `snippet-${index}.ts`); + writeFileSync(file, code); + return file; + }); + const program = ts.createProgram(files, { + strict: true, + noEmit: true, + skipLibCheck: true, + target: ts.ScriptTarget.ES2022, + module: ts.ModuleKind.ESNext, + moduleResolution: ts.ModuleResolutionKind.Bundler, + moduleDetection: ts.ModuleDetectionKind.Force, + resolveJsonModule: true, + esModuleInterop: true, + lib: ["lib.es2022.d.ts"], + types: ["node"], + typeRoots: [join(ROOT, "node_modules", "@types")], + baseUrl: ROOT, + paths: { zenrows: ["src/index.ts"] }, + }); + const diagnostics = ts.getPreEmitDiagnostics(program).map((diagnostic) => { + const where = diagnostic.file + ? `${diagnostic.file.fileName.replace(dir, "README")}:${ + diagnostic.file.getLineAndCharacterOfPosition(diagnostic.start ?? 0).line + 1 + }` + : ""; + return `${where} ${ts.flattenDiagnosticMessageText(diagnostic.messageText, "\n")}`; + }); + expect(diagnostics).toEqual([]); + } finally { + rmSync(dir, { recursive: true, force: true }); + } + }, 60_000); +}); From 2b7d8a62b80c6b5f9b5d159716e7e2a26c8b091a Mon Sep 17 00:00:00 2001 From: Emirlan Rasulov Date: Fri, 16 Oct 2026 18:07:56 +0600 Subject: [PATCH 3/3] fix(batch): wait for the right run's ingestion and time out stalled content reads - An ingestion wait is done only when ingest_status leaves pending. A run stopped mid-ingest is terminal while the API is still settling its rows and totals, so the run status alone no longer ends the wait. - rerun and submit with waitForIngest poll the run they returned, by id, instead of the job's latest run. A replayed request returns its original run, which may no longer be the latest. - getTaskContent and getTaskContentBytes read the body inside the request timeout, so a body that stalls after the headers no longer hangs. - submitScheduled no longer passes waitForIngest: a scheduled submit creates no run, so there is nothing to wait for. The README says so. --- README.md | 2 +- src/batch/client.ts | 45 +++++--- src/batch/resources.ts | 21 +++- src/batch/transport.ts | 14 +++ tests/batch-consistency.test.ts | 175 ++++++++++++++++++++++++++++---- 5 files changed, 215 insertions(+), 42 deletions(-) diff --git a/README.md b/README.md index 7ff5ee3..c3186e2 100644 --- a/README.md +++ b/README.md @@ -275,7 +275,7 @@ The client also wraps the rest of the API: - Results exports: `job.run.startExport()`, `waitForExport()`, `downloadAllResults()`. - Cost: `estimateCost(tasks, zenrowsParams)` prices each task on the job params overlaid with its own `zenrows_params`, locally, with no API call. A flag counts as on only when it is `true` or `"true"`, the values the API accepts. -A submit of 10,000 tasks or more, and every rerun, returns while the API is still writing the run's task rows (`ingest_status: "pending"`). Pass `waitForIngest: true` to `submitRegular()`, `submitOpen()`, `submitScheduled()`, `rerun()` or `retryFailed()` to resolve only once the rows are written; until then `results()` can be partial. +A submit of 10,000 tasks or more, and every rerun, returns while the API is still writing the run's task rows (`ingest_status: "pending"`). Pass `waitForIngest: true` to `submitRegular()`, `submitOpen()`, `rerun()` or `retryFailed()` to resolve only once the rows are written; until then `results()` can be partial. A scheduled submit (`submitScheduled()`) creates no run: it stores the task template, and the first run starts when the schedule fires. #### Results and retention diff --git a/src/batch/client.ts b/src/batch/client.ts index 34393f2..5da9304 100644 --- a/src/batch/client.ts +++ b/src/batch/client.ts @@ -112,6 +112,11 @@ export interface SubmitTypedOptions { metadata?: Record; webhook?: BatchWebhookConfig; idempotencyKey?: string; + /** + * Block until the run's task rows are written for a large (202) submission. No effect on + * `submitScheduled`: a scheduled submit creates no run, the first one starts when the + * schedule fires. + */ waitForIngest?: boolean; } @@ -284,7 +289,6 @@ export class ZenRowsBatchClient { metadata: opts.metadata, webhook: opts.webhook, idempotencyKey: opts.idempotencyKey, - waitForIngest: opts.waitForIngest, // biome-ignore lint/suspicious/noExplicitAny: schedule builder output matches JobSchedule at the wire boundary ...{ schedule: schedule.toRequestBody() as unknown as any }, }); @@ -387,11 +391,7 @@ export class ZenRowsBatchClient { taskId: string, options: { runId?: string } = {}, ): Promise { - const response = await this.transport.requestRaw( - "GET", - this.contentPath(jobId, taskId, options), - ); - return response.text(); + return this.transport.requestText("GET", this.contentPath(jobId, taskId, options)); } /** @@ -403,11 +403,7 @@ export class ZenRowsBatchClient { taskId: string, options: { runId?: string } = {}, ): Promise { - const response = await this.transport.requestRaw( - "GET", - this.contentPath(jobId, taskId, options), - ); - return Buffer.from(await response.arrayBuffer()); + return this.transport.requestBytes("GET", this.contentPath(jobId, taskId, options)); } private contentPath(jobId: string, taskId: string, options: { runId?: string }): string { @@ -770,12 +766,29 @@ export class ZenRowsBatchClient { jobId: string, options: { timeout?: number; pollInterval?: number; maxPollInterval?: number } = {}, ): Promise { + // Ingestion is done only when the API says so. A run stopped mid-ingest is already + // terminal while its rows and totals are still being settled, so the run status alone + // is not enough. return pollUntil(() => this._getJobData(jobId), { - isDone: (job) => { - const run = job.latest_run; - if (!run || TERMINAL_RUN_STATUSES.has(run.status)) return true; - return run.ingest_status !== "pending"; - }, + isDone: (job) => !job.latest_run || job.latest_run.ingest_status !== "pending", + timeout: options.timeout ?? 300, + initialInterval: options.pollInterval ?? 2, + maxInterval: options.maxPollInterval ?? 15, + }); + } + + /** + * @internal Wait for one run's ingestion, by run id. A replayed submit or rerun returns + * the original run, which may no longer be the job's latest, so this never follows + * `latest_run`. + */ + async _waitForRunIngestRaw( + jobId: string, + runId: string, + options: { timeout?: number; pollInterval?: number; maxPollInterval?: number } = {}, + ): Promise { + return pollUntil(() => this._getRunData(jobId, runId), { + isDone: (run) => run.ingest_status !== "pending", timeout: options.timeout ?? 300, initialInterval: options.pollInterval ?? 2, maxInterval: options.maxPollInterval ?? 15, diff --git a/src/batch/resources.ts b/src/batch/resources.ts index c490d7a..9e39681 100644 --- a/src/batch/resources.ts +++ b/src/batch/resources.ts @@ -443,8 +443,8 @@ export class JobRef { const resp = await this.#client._postRerun(this.jobId, rest); let run = resp.latest_run; if (waitForIngest && run.ingest_status === "pending") { - const job = await this.#client._waitForIngestRaw(this.jobId); - if (job.latest_run?.run_id === run.run_id) run = job.latest_run; + // By run id: a replayed rerun returns its original run, which may not be the latest. + run = await this.#client._waitForRunIngestRaw(this.jobId, run.run_id); } return new RunHandle(this.#client, this.jobId, run.run_id, run); } @@ -506,10 +506,25 @@ export class JobRef { await this.#client.deleteJobWebhook(this.jobId); } - /** Block until the current run's async-carrier ingestion has finished writing task rows. */ + /** + * Block until ingestion has finished writing the run's task rows (`ingest_status` is no + * longer `pending`), then return the loaded job. On a ref from a `submit*` call this waits + * for the run that submit returned; otherwise for the job's latest run. A run stopped + * mid-ingest still waits until the API has settled its rows. + */ async waitForIngest( options: { timeout?: number; pollInterval?: number; maxPollInterval?: number } = {}, ): Promise { + const runId = this.submitResponse?.latest_run?.run_id; + if (runId) { + await this.#client._waitForRunIngestRaw(this.jobId, runId, options); + return new JobHandle( + this.#client, + this.jobId, + await this.#client._getJobData(this.jobId), + this.submitResponse, + ); + } return new JobHandle( this.#client, this.jobId, diff --git a/src/batch/transport.ts b/src/batch/transport.ts index 1fe48dc..26a5413 100644 --- a/src/batch/transport.ts +++ b/src/batch/transport.ts @@ -122,6 +122,20 @@ export class BatchTransport { } } + /** + * Send a request and return its body as bytes; throw the factory's error on non-2xx. The + * timeout covers reading the body. + */ + async requestBytes(method: string, path: string, options: RequestOptions = {}): Promise { + const { response, clear } = await this.send(method, path, options); + try { + await this.throwUnlessOk(response); + return Buffer.from(await response.arrayBuffer()); + } finally { + clear(); + } + } + /** * Send a request and return the raw `Response`, for bodies read as a stream. The timeout * stops once the headers of a 2xx arrive; an error body is still read within it. diff --git a/tests/batch-consistency.test.ts b/tests/batch-consistency.test.ts index f686092..4c8287f 100644 --- a/tests/batch-consistency.test.ts +++ b/tests/batch-consistency.test.ts @@ -1,3 +1,5 @@ +import { createServer } from "node:http"; +import type { AddressInfo } from "node:net"; import { http, HttpResponse } from "msw"; import { describe, expect, expectTypeOf, test } from "vitest"; import { @@ -123,8 +125,22 @@ describe("content of an expired result", () => { }); }); -describe("waitForIngest on rerun and submitScheduled", () => { - test("rerun({ waitForIngest: true }) returns the run once its rows are written", async () => { +describe("waiting for ingestion", () => { + function runBody(runId: string, status: string, ingest: string) { + return { + run_id: runId, + job_id: "job_rr", + run_sequence: 2, + status, + ingest_status: ingest, + stats: { total: 3, completed: 0, successful: 0, failed: 0 }, + }; + } + + test("rerun({ waitForIngest: true }) waits on the returned run, not the job's latest", async () => { + // A replayed rerun returns its original run (run_2) while run_3 is now the latest and + // still ingesting. The wait must follow run_2. + const runReads: string[] = []; let jobReads = 0; server.use( http.post(`${API}/jobs/job_rr/rerun`, () => @@ -132,71 +148,186 @@ describe("waitForIngest on rerun and submitScheduled", () => { { job_id: "job_rr", status: "closed", - latest_run: { run_id: "run_2", status: "pending", ingest_status: "pending" }, + latest_run: runBody("run_2", "pending", "pending"), retried_tasks: 3, inherited_tasks: 0, }, { status: 202 }, ), ), + http.get(`${API}/jobs/job_rr/runs/:runId`, ({ params }) => { + runReads.push(String(params.runId)); + return HttpResponse.json(runBody(String(params.runId), "completed", "done")); + }), http.get(`${API}/jobs/job_rr`, () => { jobReads += 1; return HttpResponse.json({ job_id: "job_rr", type: "regular", status: "closed", - latest_run: { run_id: "run_2", status: "running", ingest_status: "done" }, + latest_run: runBody("run_3", "running", "pending"), }); }), ); const client = new ZenRowsBatchClient("API_KEY"); - const run = await client.job("job_rr").retryFailed({ waitForIngest: true }); + const run = await client + .job("job_rr") + .retryFailed({ idempotencyKey: "replayed", waitForIngest: true }); expect(run.runId).toBe("run_2"); expect(run.data.ingest_status).toBe("done"); - expect(jobReads).toBe(1); + expect(runReads).toEqual(["run_2"]); + expect(jobReads).toBe(0); // Without the option, no extra read. await client.job("job_rr").rerun(); - expect(jobReads).toBe(1); + expect(runReads).toEqual(["run_2"]); }); - test("submitScheduled passes waitForIngest through", async () => { - let jobReads = 0; + test("a run stopped mid-ingest is not done until ingest_status leaves pending", async () => { + let reads = 0; + server.use( + http.post(`${API}/jobs/job_rr/rerun`, () => + HttpResponse.json( + { + job_id: "job_rr", + status: "closed", + latest_run: runBody("run_2", "pending", "pending"), + retried_tasks: 3, + inherited_tasks: 0, + }, + { status: 202 }, + ), + ), + http.get(`${API}/jobs/job_rr/runs/run_2`, () => { + reads += 1; + // Stopped at once, but the API settles the rows and totals a poll later. + return HttpResponse.json(runBody("run_2", "stopped", reads < 2 ? "pending" : "done")); + }), + ); + const client = new ZenRowsBatchClient("API_KEY"); + // The rerun path uses the default 2 s poll; drive the same wait directly with a short one. + const resp = await client._postRerun("job_rr", {}); + const run = await client._waitForRunIngestRaw("job_rr", resp.latest_run.run_id, { + pollInterval: 0.01, + }); + expect(run.status).toBe("stopped"); + expect(run.ingest_status).toBe("done"); + expect(reads).toBe(2); + }); + + test("JobRef.waitForIngest on a stopped run with pending ingest keeps polling", async () => { + let reads = 0; + server.use( + http.get(`${API}/jobs/job_stop_ingest`, () => { + reads += 1; + return HttpResponse.json({ + job_id: "job_stop_ingest", + type: "regular", + status: "closed", + latest_run: { + ...runBody("run_1", "stopped", reads < 2 ? "pending" : "done"), + job_id: "job_stop_ingest", + }, + }); + }), + ); + const client = new ZenRowsBatchClient("API_KEY"); + const handle = await client.job("job_stop_ingest").waitForIngest({ pollInterval: 0.01 }); + expect(handle.data.latest_run?.ingest_status).toBe("done"); + expect(reads).toBe(2); + }); + + test("a replayed submit waits on the run it returned, then loads the job", async () => { + const runReads: string[] = []; server.use( http.post(`${API}/jobs`, () => HttpResponse.json( { - job_id: "job_sched", + job_id: "job_big", status: "closed", accepted_tasks: 10_000, - latest_run: { run_id: "run_1", status: "pending", ingest_status: "pending" }, + latest_run: { ...runBody("run_1", "pending", "pending"), job_id: "job_big" }, }, { status: 202 }, ), ), - http.get(`${API}/jobs/job_sched`, () => { - jobReads += 1; + http.get(`${API}/jobs/job_big/runs/:runId`, ({ params }) => { + runReads.push(String(params.runId)); return HttpResponse.json({ - job_id: "job_sched", - type: "scheduled", - status: "closed", - latest_run: { run_id: "run_1", status: "running", ingest_status: "done" }, + ...runBody(String(params.runId), "completed", "done"), + job_id: "job_big", }); }), + http.get(`${API}/jobs/job_big`, () => + HttpResponse.json({ + job_id: "job_big", + type: "regular", + status: "closed", + latest_run: { ...runBody("run_3", "running", "pending"), job_id: "job_big" }, + }), + ), + ); + const client = new ZenRowsBatchClient("API_KEY"); + const job = await client.submitRegular(["https://example.com"], undefined, { + idempotencyKey: "replayed", + waitForIngest: true, + }); + expect(runReads).toEqual(["run_1"]); + expect(job.jobId).toBe("job_big"); + }); + + test("submitScheduled answers 201 with no run, so waitForIngest never polls", async () => { + let jobReads = 0; + server.use( + http.post(`${API}/jobs`, () => + HttpResponse.json( + { job_id: "job_sched", status: "closed", accepted_tasks: 1 }, + { status: 201 }, + ), + ), + http.get(`${API}/jobs/job_sched`, () => { + jobReads += 1; + return HttpResponse.json({ job_id: "job_sched", type: "scheduled", status: "closed" }); + }), ); const client = new ZenRowsBatchClient("API_KEY"); - await client.submitScheduled( + const job = await client.submitScheduled( new At("2030-01-01T00:00:00", "UTC"), ["https://example.com"], undefined, - { - waitForIngest: true, - }, + { waitForIngest: true }, ); - expect(jobReads).toBe(1); + expect(job.jobId).toBe("job_sched"); + expect(jobReads).toBe(0); }); }); +describe("content reads honour the request timeout", () => { + test("getTaskContent and getTaskContentBytes abort a body that stalls after the headers", async () => { + // A real socket: headers and a first chunk arrive, then the body never ends. + const sockets = new Set(); + const stalled = createServer((_req, res) => { + res.writeHead(200, { "Content-Type": "text/html" }); + res.write(""); + }); + stalled.on("connection", (socket) => sockets.add(socket)); + await new Promise((resolve) => stalled.listen(0, "127.0.0.1", resolve)); + const { port } = stalled.address() as AddressInfo; + try { + const client = new ZenRowsBatchClient("API_KEY", { + baseURL: `http://127.0.0.1:${port}/v1`, + timeout: 0.2, + retries: 0, + }); + await expect(client.getTaskContent("job_stall", "t1")).rejects.toThrow(/timed out/); + await expect(client.getTaskContentBytes("job_stall", "t1")).rejects.toThrow(/timed out/); + } finally { + for (const socket of sockets) socket.destroy(); + await new Promise((resolve) => stalled.close(resolve)); + } + }, 5_000); +}); + describe("types follow the wire shapes", () => { test("task, run and webhook event fields", () => { expectTypeOf<"expired">().toMatchTypeOf();