diff --git a/README.md b/README.md index 1f7180f..f1b6379 100644 --- a/README.md +++ b/README.md @@ -132,12 +132,35 @@ validation + credit estimation always work (no key needed). ```bash zenrows batch estimate jobs.jsonl # validate the spec + estimate credits (local) -zenrows batch create jobs.jsonl --wait # submit a job and wait for it to finish +zenrows batch create jobs.jsonl --follow # submit a job and wait for it to finish zenrows batch status # run status + progress stats +zenrows batch wait --timeout 600000 # wait up to 600000 ms (or --timeout 600s) zenrows batch results --out results.jsonl # collect results (paginated) zenrows batch cancel # stop an in-flight run +zenrows batch retry-failed --include-pending # rerun failed and unfinished tasks ``` +`--wait` is an alias of `--follow`. `create` and `retry-failed` send an +`Idempotency-Key` and reuse it for their own automatic retries, so a dropped +connection never starts a second job or run. Running a command again sends a +new key unless you pass one: a repeated `create` creates another job, and a +repeated `retry-failed`, once the previous run has ended, may start another +run of the tasks that still match, re-scraping and billing them again. With +the same `--idempotency-key `, the same file and flags, within 24 hours +of the first request, a repeat cannot create a duplicate (after 24 hours the +key is forgotten). Its answer is not guaranteed to be the original: the API +may replay it, answer 409 (no recorded answer yet, or a rerun whose run is +still in progress), or refuse it before looking at the key, for example with +a 402 when the account is out of credits. Unless the repeat returns the job, +check your jobs before doing anything else. When a submit or rerun ends without a usable answer, the +error carries the key and prints the exact command to repeat with it. + +Before submitting, `create` checks the run against the local policy: +`max_pages_per_run` (default 1000 URLs) and `max_credits_per_run` (default +10000 estimated credits, where each Adaptive Stealth Mode task counts as 25). +Raise them for large jobs with `zenrows policy set max_pages_per_run ` and +`zenrows policy set max_credits_per_run `. + `jobs.jsonl` is one JSON object per line, each with a `url` (plus optional per-line overrides like `js_render`, `premium_proxy`, `proxy_country`, `extract`). Validate and estimate locally first, then submit with `create`. @@ -150,8 +173,8 @@ Extract — structured data instead of raw HTML: {"url": "https://example.com/reviews", "extract": "auto", "external_id": "r1"} ``` -An Extract task costs the same as a regular one (1 credit at base tier), so -`batch estimate` prices it correctly. Results for an Extract task carry two +Extract adds no charge of its own: an Extract task costs what the same task +costs without it (1 credit at base tier), so `batch estimate` prices it correctly. Results for an Extract task carry two keys: `html` (the raw page) and `parsed` (the structured data) — validate a sample with `zenrows extract ` before running the full batch. diff --git a/agent-plugin/skills/batch-jobs/SKILL.md b/agent-plugin/skills/batch-jobs/SKILL.md index db3ef13..7d40862 100644 --- a/agent-plugin/skills/batch-jobs/SKILL.md +++ b/agent-plugin/skills/batch-jobs/SKILL.md @@ -21,19 +21,23 @@ anti-bot advantage becomes obvious — thousands to millions of URLs. | `batch_create` | Submit a job (list of URL tasks + optional scrape/extract params) | | `batch_status` | Poll run status + progress stats | | `batch_results` | Page results (cursor pagination; filter successful/failed/all) | +| `batch_wait` | Block until the run reaches a terminal status | | `batch_cancel` | Stop an in-flight run | -Optional if exposed by the server: `batch_wait` to block until terminal status. - ## Workflow 1. Validate the workflow on **one** URL with `scrape` or `extract`. -2. `batch_create` with the full task list (each task: `url` + optional - overrides like `js_render`, `premium_proxy`, `proxy_country`, `mode`, - `autoparse`, `extract`, `external_id`, `metadata`). `extract: "auto"` - returns structured data instead of raw HTML, at the same credit cost as a - regular task; the result carries `html` and `parsed` keys. -3. Poll `batch_status` until terminal (`completed` / `stopped` / …). +2. `batch_create` with the full task list. Each task is `url` plus optional + `external_id` and `metadata`, and per-task scrape overrides like + `js_render`, `premium_proxy`, `proxy_country`, `mode`, `autoparse` or + `extract`. Put those overrides under the task's `zenrows_params` + (`{"url": "...", "zenrows_params": {"extract": "auto"}}`): the MCP server + reads them there, and some versions silently ignore the same keys set flat + on the task. `extract: "auto"` returns structured data instead of + raw HTML, at the same credit cost as a regular task; the result carries + `html` and `parsed` keys. +3. Call `batch_wait`, or poll `batch_status`, until terminal (`completed`, + `failed`, `stopped` or `deleted`). A `failed` run carries `failure_reason`. 4. Collect with `batch_results`. 5. `batch_cancel` if the user aborts or the run should stop early. diff --git a/agent-plugin/skills/cost-control/SKILL.md b/agent-plugin/skills/cost-control/SKILL.md index d6a76f0..39720e0 100644 --- a/agent-plugin/skills/cost-control/SKILL.md +++ b/agent-plugin/skills/cost-control/SKILL.md @@ -18,9 +18,10 @@ These multipliers apply when you enable the flags (on `scrape` or `extract`): - Premium proxies (`premium_proxy`): **10×** - Both: **25×** -`extract` `mode=auto` is currently free; billing may apply -later. Do not promise permanent free extract pricing. Autoparse / CSS still -follow the multipliers above when stealth flags are set. +Extract adds no charge of its own: an `extract` request costs what the +underlying fetch costs, multipliers included. Do not promise that Extract +pricing will never change. Autoparse / CSS follow the same multipliers when +stealth flags are set. ## Browser sessions (`browser_*`) diff --git a/agent-plugin/skills/extract/SKILL.md b/agent-plugin/skills/extract/SKILL.md index 4d08cc9..b46c28a 100644 --- a/agent-plugin/skills/extract/SKILL.md +++ b/agent-plugin/skills/extract/SKILL.md @@ -21,7 +21,8 @@ extract({ url, proxy_country: "US" }) # geo-targeted ``` `mode=auto` (`extract=auto`) works on any domain Zenrows has prepared, with any -key, and is currently free; billing may apply later. If the +key. Extract adds no charge of its own: a request costs what the underlying +fetch costs, multipliers included. If the domain is not enabled (AUTH010) or not prepared yet (REQS007), the tool retries once with Autoparse by default (`fallback_autoparse`). diff --git a/agent-plugin/skills/zenrows/SKILL.md b/agent-plugin/skills/zenrows/SKILL.md index 1cae7d3..1c0eff7 100644 --- a/agent-plugin/skills/zenrows/SKILL.md +++ b/agent-plugin/skills/zenrows/SKILL.md @@ -66,8 +66,8 @@ If the user needs login, clicks, forms, sessions, or persistent state: | Primitive | MCP tools | Notes | | --- | --- | --- | | Protected Fetch | `scrape` | Default markdown; HTML / plaintext / PDF / screenshot options | -| Extract | `extract` | Structured JSON (`extract=auto`, autoparse, css_extractor). `extract=auto` is currently free; billing may apply later | -| Batch | `batch_create`, `batch_status`, `batch_results`, `batch_cancel` | Same credits as Fetch; a 403 maps to `BATCH_ACCESS_DENIED` | +| Extract | `extract` | Structured JSON (`extract=auto`, autoparse, css_extractor). Extract adds no charge of its own: a request costs what the underlying fetch costs | +| Batch | `batch_create`, `batch_status`, `batch_results`, `batch_wait`, `batch_cancel` | Same credits as Fetch; a 403 maps to `BATCH_ACCESS_DENIED` | | Browser | `browser_*` (start with `browser_navigate`) | Escalation-only; same backend as Browser Sessions | ## Cost notes diff --git a/recipes/batch-estimate/RECIPE.md b/recipes/batch-estimate/RECIPE.md index 84d4bd4..3632dcd 100644 --- a/recipes/batch-estimate/RECIPE.md +++ b/recipes/batch-estimate/RECIPE.md @@ -28,6 +28,6 @@ URL; cost multipliers are basic 1x / JS 5x / proxy 10x / both 25x. Once the spec is clean, submit it: ``` -zenrows batch create jobs.jsonl --wait +zenrows batch create jobs.jsonl --follow ``` See [[batch-jobs]] for the full cloud flow. diff --git a/skills/batch-jobs/SKILL.md b/skills/batch-jobs/SKILL.md index 79d1fa2..b1d1050 100644 --- a/skills/batch-jobs/SKILL.md +++ b/skills/batch-jobs/SKILL.md @@ -34,22 +34,52 @@ structured data instead of raw HTML. {"url": "https://example.com/products", "extract": "auto", "external_id": "p1"} ``` -An Extract task costs the same as a regular one, so `batch estimate` prices it -correctly. The result carries two keys: `html` (the +Extract adds no charge of its own: an Extract task costs what the same task +costs without it, so `batch estimate` prices it correctly. The result carries two keys: `html` (the raw page) and `parsed` (the structured data). Validate on a single URL with `zenrows extract ` before running the full batch. ## Cloud (needs a key) ``` -zenrows batch create jobs.jsonl [--proxy-country cc] [--wait] +zenrows batch create jobs.jsonl [--proxy-country cc] [--follow [--timeout ms]] zenrows batch create jobs.jsonl --manual [--js-render] [--premium-proxy] zenrows batch status # run status + stats -zenrows batch results [--status successful|failed|all] [--out results.jsonl] -zenrows batch wait [--timeout ms] +zenrows batch results [--status successful|failed|processing|all] [--out results.jsonl] +zenrows batch wait [--timeout ms] # default 600000 ms (600s also works); fails if paused zenrows batch cancel # stop an in-flight run -zenrows batch retry-failed # rerun only the failed tasks (new run) +zenrows batch retry-failed [--include-pending] # rerun failed (and unfinished) tasks ``` +`create` and `retry-failed` send an `Idempotency-Key` and reuse it for their +own automatic retries, so a dropped connection never starts a second job or +run. Running a command again sends a new key unless you pass one: a repeated +`create` creates another job, and a repeated `retry-failed`, once the +previous run has ended, may start another run of the tasks that still match. +Either re-scrapes and bills those tasks again. With the same +`--idempotency-key `, the same file and flags, within 24 hours of the +first request, a repeat cannot create a duplicate (after 24 hours the key is +forgotten). Its answer is not guaranteed to be the original: the API may +replay it, answer 409 (no recorded answer yet, or a rerun whose run is still +in progress), or refuse it before looking at the key (for example 402, out of +credits). When a submit or rerun ends without a +usable answer (no response, a 5xx, a refusal after a lost attempt, or a +success answer that can't be read), the error carries `idempotency_key` and +prints the exact command to repeat; run that, not a fresh command. If it +does not return the job, or reports that an earlier attempt may have gone +through, check `zenrows batch status ` (or the dashboard) before running +anything else. With a workspace, `create` and `retry-failed` also record the +key in their run artifact. If `create --follow` fails after the job was +created, the error names the job: follow it with `zenrows batch wait `, +do not create it again. Status reads retry transient errors on their own. +`--timeout` takes milliseconds, as before; `600s` and `600000ms` also work. +`--wait` still works as an alias of `--follow`. + +`--include-pending` reruns every task that did not finish: `failed`, `pending` +(never started) and `processing` (cut off mid-flight). `wait` fails right +away when a run is paused (`BATCH_FAILED`, `reason: run_paused`, exit 1, the +same code and exit as the timeout it would otherwise reach): a paused run +picks up no new task, though tasks already started may still finish. + Every task runs in Adaptive Stealth Mode (`mode=auto`) by default, like `fetch` and `extract`: Zenrows enables JS rendering and premium proxies only on the pages that need them, and bills only the configuration that succeeds. A task that sets @@ -68,7 +98,29 @@ A run ends in one of `completed`, `failed`, `stopped`, or `deleted`. `status`, `failure_reason` is `api_key_cap_reached`, the key hit its own credit cap (`KEY_CREDIT_CAP_REACHED`): the account still has credits, other keys keep working, and the tasks not yet run stay pending. Wait for the cap to reset -(`zenrows usage`) or raise the cap; do not resubmit the job with the same key. +(`zenrows usage`) or raise the cap, then run what is left with +`zenrows batch retry-failed --include-pending`; do not resubmit the job. +A run that failed with `insufficient_credits` keeps `BATCH_FAILED` and works +the same way once credits are added. Errors carry the Batch API's own code as +`server_code`, and a finer `reason` (`insufficient_credits`, +`api_key_cap_reached`, `too_many_active_jobs`, `too_many_scheduled_jobs`, +`run_paused`, `wait_timeout`) next to the unchanged `code`. A `429` +(`BATCH_QUOTA_EXCEEDED`) means an account limit: 3 active jobs and 5 +scheduled jobs by default. A closed job counts as active while its run is in +progress (wait for it, or stop it), an open job until it is deleted, or +closed and its run has ended, and a scheduled job counts toward its limit +until it is deleted. + +## Local policy caps + +`create` checks the run against `.zenrows/policy.json` before anything is sent: +`max_pages_per_run` (default 1000 URLs) and `max_credits_per_run` (default +10000 estimated credits). The estimate is an upper bound and counts each +Adaptive Stealth Mode task as 25, so with the defaults a run of more than 400 +auto-mode URLs stops with `POLICY_LIMIT_EXCEEDED`. Raise the caps for large +jobs, e.g. `zenrows policy set max_pages_per_run 50000` and +`zenrows policy set max_credits_per_run 1250000`; the error names the exact +command for the run you tried. Deferred (documented, not yet in the CLI): CSV upload, open/queue jobs, scheduled jobs, webhooks/HMAC, and ZIP export. diff --git a/skills/trace-debug/SKILL.md b/skills/trace-debug/SKILL.md index 0948509..bf2d08b 100644 --- a/skills/trace-debug/SKILL.md +++ b/skills/trace-debug/SKILL.md @@ -34,7 +34,7 @@ zenrows trace export # JSON for sharing connectivity and `zenrows config show`. - `AUTH_INVALID` → re-check the key, `zenrows login --api-key …`. - `KEY_CREDIT_CAP_REACHED` → this API key hit one of its own credit caps (HTTP - 402, gateway code `AUTH014`, Batch `api_key_cap_reached`). The account still + 402, code `AUTH014`). The account still has credits and its other keys keep working. Do not retry and do not escalate: `zenrows usage` shows the cap and when it resets. Wait for the reset, or have the account owner raise or remove the cap at @@ -43,8 +43,23 @@ zenrows trace export # JSON for sharing local policy credit limit was hit). Do not retry-loop; `zenrows usage` shows when credits renew. Top up or upgrade, or claim the account if it is an auto-created Free plan. -- `BATCH_FAILED` → the batch run ended `failed`; `failure_reason` and - `failure_detail` in `zenrows batch status --json` say why. +- Batch's 402 keeps the codes above: `POLICY_MAX_CREDITS_EXCEEDED` (out of + credits, `reason: insufficient_credits`) or `KEY_CREDIT_CAP_REACHED` + (`reason: api_key_cap_reached`). A run that stopped for lack of credits ends + `failed` with `BATCH_FAILED` and `reason: insufficient_credits`. Afterwards + run the tasks that did not finish with + `zenrows batch retry-failed --include-pending`. +- `BATCH_QUOTA_EXCEEDED` → Batch's 429 (`reason: too_many_active_jobs` or + `too_many_scheduled_jobs`): the account reached an active-job (3) + or scheduled-job (5) limit, shared with Crawl. A closed job frees its slot + when its run ends (wait, or stop it), an open job when it is deleted, or + closed and its run has ended, and a scheduled job only when it is deleted. +- `BATCH_FAILED` → any other Batch error (400, 409, 5xx), a run that ended + `failed`, or a `wait` that timed out (`reason: wait_timeout`) or found the + run paused (`reason: run_paused`); `server_code` holds the API's own code, and `failure_reason` / + `failure_detail` in `zenrows batch status --json` say why a run failed. + A Batch 401, 403 and 404 keep their own codes: `AUTH_INVALID`, + `BATCH_ACCESS_DENIED` and `BATCH_NOT_FOUND`. - `CRAWL_NOT_ENABLED` → the API answered 403 `REQS008`: Crawl is not enabled for this account. Do not retry; ask Zenrows support for access. - `CRAWL_INSUFFICIENT_CREDITS` / `CRAWL_KEY_CAP_REACHED` → Crawl's 402: the diff --git a/skills/zenrows/SKILL.md b/skills/zenrows/SKILL.md index c782a36..21bcb92 100644 --- a/skills/zenrows/SKILL.md +++ b/skills/zenrows/SKILL.md @@ -41,8 +41,9 @@ If the user has a start page (listing, index, section) but not the item URLs: If the user has many URLs (bulk): → Use Batch. (validate one page with fetch/extract first, then `zenrows batch create `) - Batch is the cheap path at scale — do NOT - fan out fetch/extract per URL for bulk work. + Batch is the reliable path at scale (same + per-request credits as fetch); do NOT fan + out fetch/extract per URL for bulk work. If the user needs login, clicks, forms, sessions, or persistent state: → Use Interact / Browser Sessions.(zenrows browser) [escalation-only] @@ -79,7 +80,7 @@ Run `zenrows status` for the live capability matrix. As of this toolkit: | MCP | `zenrows mcp` | **available**: `https://mcp.zenrows.com/mcp` | - Extract: `extract=auto` falls back to Autoparse on a domain not prepared yet. -- Batch: the cheapest path at scale (create, status, results). Estimate specs locally with no key. +- Batch: the reliable path at scale, at the same per-request credits as Fetch (create, status, results, wait, retry-failed). Estimate specs locally with no key. - Crawl: collects the URLs behind one start page, optionally with each page's HTML. - Browser: escalation-only, for pages Fetch can't handle; same backend as the MCP `browser_*` tools. diff --git a/src/adapters/batch.ts b/src/adapters/batch.ts index b32bf7d..aa90891 100644 --- a/src/adapters/batch.ts +++ b/src/adapters/batch.ts @@ -16,8 +16,8 @@ export interface BatchJob { url: string; /** Optional stable id echoed back on each result row. */ external_id?: string; - /** Opaque per-task metadata carried through to results. */ - metadata?: unknown; + /** Per-task string-to-string metadata carried through to results. */ + metadata?: Record; /** Optional nested per-task scrape params (merged with the flat keys below). */ zenrows_params?: Record; /** Optional per-job overrides mirroring fetch options. */ @@ -65,6 +65,11 @@ export function validateJsonl(file: string): BatchValidation { result.errors.push({ line: idx + 1, reason: `invalid url: ${obj.url}` }); return; } + const metaError = metadataError(obj.metadata); + if (metaError) { + result.errors.push({ line: idx + 1, reason: metaError }); + return; + } result.validJobs++; result.jobs.push(obj); } catch { @@ -75,10 +80,105 @@ export function validateJsonl(file: string): BatchValidation { } /** - * Rough local credit estimate using documented multipliers - * (basic 1x, js 5x, proxies 10x, both 25x). Counts in abstract "credits". + * The API accepts `metadata` only as an object whose values are strings; any + * other shape fails the whole submit. Returns the reason, or null when valid. + */ +function metadataError(metadata: unknown): string | null { + if (metadata === undefined || metadata === null) return null; + if (typeof metadata !== "object" || Array.isArray(metadata)) { + return "`metadata` must be an object of string values"; + } + for (const [k, v] of Object.entries(metadata as Record)) { + if (typeof v !== "string") return `\`metadata.${k}\` must be a string`; + } + return null; +} + +/** + * Scrape params the API reads as booleans. It accepts JSON `true`/`false` or the + * strings "true"/"false" in any case (it lowercases them), and rejects anything + * else, `1` and "yes" included, with a 400. + */ +const BOOLEAN_PARAMS = new Set(["js_render", "premium_proxy", "original_status", "json_response", "autoparse"]); + +/** + * Canonical spelling of a boolean param: JSON `true` and "true" in any case + * become "true", JSON `false` and "false" in any case become "false". Any other + * value is returned unchanged (see `assertBooleanParams`, which rejects it). */ -export function estimateCredits(jobs: BatchJob[]): { credits: number; perJob: number[] } { +function canonicalBool(v: unknown): unknown { + if (v === true) return "true"; + if (v === false) return "false"; + if (typeof v === "string") { + const s = v.trim().toLowerCase(); + if (s === "true" || s === "false") return s; + } + return v; +} + +/** Throw before submitting if a boolean param holds a value the API rejects. */ +function assertBooleanParams(params: Record, where: string): void { + for (const k of BOOLEAN_PARAMS) { + const v = params[k]; + if (v !== undefined && v !== "true" && v !== "false") { + throw new ToolkitError({ + code: "INVALID_USAGE", + message: `${k} must be true or false (got ${JSON.stringify(v)}) in ${where}.`, + likely_cause: `The Batch API accepts only true or false for ${k}; other values, 1 and "yes" included, fail the whole job.`, + next_action: `Set ${k} to true or false on that line, then retry.`, + }); + } + } +} + +/** A task's own params: its nested `zenrows_params`, then its flat keys (later wins). */ +function taskParamsOf(job: BatchJob): Record { + const flat: Record = {}; + for (const [k, v] of Object.entries(job)) { + if (!TASK_RESERVED.has(k)) flat[k] = v; + } + return { ...(job.zenrows_params ?? {}), ...flat }; +} + +/** + * The params a task actually runs with: the job-level params and the task's own, + * each canonicalized exactly as they are submitted (`normalizeParams`), merged + * the way the API merges them (the task wins on a shared key). A task key set + * to null is not submitted, so the job's value applies. + */ +export function effectiveParams(job: BatchJob, jobParams: Record = {}): Record { + return { ...normalizeParams(jobParams), ...normalizeParams(taskParamsOf(job)) }; +} + +/** + * Local credit estimate, an upper bound, from the documented rate card: basic + * 1, js_render 5, premium_proxy 10, both 25. `mode=auto` costs 1 to 25 + * depending on what each page needs, so it counts as 25. Each task is priced on + * the params it is submitted with (`effectiveParams`), so job-level flags count. + */ +export function estimateCredits( + jobs: BatchJob[], + jobParams: Record = {}, +): { credits: number; perJob: number[] } { + const perJob = jobs.map((j) => { + const p = effectiveParams(j, jobParams); + if (p.mode === "auto") return 25; + const js = p.js_render === "true"; + const proxy = p.premium_proxy === "true"; + if (js && proxy) return 25; + if (proxy) return 10; + if (js) return 5; + return 1; + }); + return { credits: perJob.reduce((a, b) => a + b, 0), perJob }; +} + +/** + * The estimate earlier versions used, kept only for the policy credit cap so a + * run that passed the cap before still passes: flat per-line flags only, with + * JavaScript truthiness. Use `estimateCredits` for the real estimate. + */ +export function legacyEstimateCredits(jobs: BatchJob[]): { credits: number; perJob: number[] } { const perJob = jobs.map((j) => { const js = j.js_render || j.mode === "auto"; const proxy = j.premium_proxy || j.mode === "auto"; @@ -116,7 +216,7 @@ export interface JobBody { tasks: Array<{ url: string; external_id?: string; - metadata?: unknown; + metadata?: Record; zenrows_params?: Record; }>; } @@ -137,15 +237,14 @@ export function toJobBody(jobs: BatchJob[], jobParams: Record = // No separate job-level check: a job-level proxy_country is valid for each task // in Adaptive Stealth Mode, and the per-task check below sees job ⊕ task params. const tasks = jobs.map((job, idx) => { - const { url, external_id, metadata, zenrows_params } = job; - const flat: Record = {}; - for (const [k, v] of Object.entries(job)) { - if (!TASK_RESERVED.has(k)) flat[k] = v; - } - const taskParams: Record = { ...(zenrows_params ?? {}), ...flat }; - // Validate on the effective (job ⊕ task) params so a job-level premium_proxy - // covers a per-task proxy_country, matching how the API merges them. - assertProxyCountryPremium({ ...jobParams, ...taskParams }, `line ${idx + 1} (${url})`); + const { url, external_id, metadata } = job; + const taskParams = taskParamsOf(job); + // Validate on the effective (job + task) params as submitted, so a job-level + // premium_proxy covers a per-task proxy_country, and a task key set to null + // (not submitted) leaves the job's value in force, matching the API's merge. + const effective = effectiveParams(job, jobParams); + assertBooleanParams(effective, `line ${idx + 1} (${url})`); + assertProxyCountryPremium(effective, `line ${idx + 1} (${url})`); const task: JobBody["tasks"][number] = { url }; if (external_id !== undefined) task.external_id = String(external_id); @@ -164,12 +263,14 @@ export function toJobBody(jobs: BatchJob[], jobParams: Record = /** * Coerce a flat options object into the string-valued map the Batch API expects * (its docs pass `zenrows_params` values as strings, e.g. `"true"`, `"us"`). - * Booleans/numbers are stringified; objects are JSON-encoded; nullish is dropped. + * Boolean params are canonicalized to "true"/"false" (see `canonicalBool`); + * other booleans/numbers are stringified; objects are JSON-encoded; nullish is dropped. */ function normalizeParams(obj: Record): Record { const out: Record = {}; - for (const [k, v] of Object.entries(obj)) { - if (v === undefined || v === null) continue; + for (const [k, raw] of Object.entries(obj)) { + if (raw === undefined || raw === null) continue; + const v = BOOLEAN_PARAMS.has(k) ? canonicalBool(raw) : raw; if (typeof v === "string") out[k] = v; else if (typeof v === "boolean" || typeof v === "number") out[k] = String(v); else out[k] = JSON.stringify(v); @@ -182,7 +283,7 @@ function isAutoMode(v: unknown): boolean { return v === "auto" || v === true; } function isTruthyFlag(v: unknown): boolean { - return v === true || v === "true"; + return canonicalBool(v) === "true"; } /** Throw if proxy_country is set without premium_proxy and outside auto mode. */ diff --git a/src/cli/commands/batch.ts b/src/cli/commands/batch.ts index d109cf9..f5a341e 100644 --- a/src/cli/commands/batch.ts +++ b/src/cli/commands/batch.ts @@ -9,16 +9,37 @@ import { statSync } from "node:fs"; import { join } from "node:path"; import { log, ANSI, c } from "../../core/logger.ts"; -import { estimateCredits, toJobBody, validateJsonl, withAdaptiveStealth, type BatchJob } from "../../adapters/batch.ts"; +import { + estimateCredits, + legacyEstimateCredits, + toJobBody, + validateJsonl, + withAdaptiveStealth, + type BatchJob, +} from "../../adapters/batch.ts"; import { loadConfig } from "../../core/config.ts"; import { requireApiKey } from "../../core/auth.ts"; import { ensureApiKey } from "../../core/ensure-key.ts"; import { assertUsable } from "../../core/capabilities.ts"; import { assertDomainAllowed, assertWithinLimits, loadPolicy } from "../../core/policy.ts"; import { newRunId, writeRun } from "../../core/artifacts.ts"; -import { createJob, downloadResults, getJob, listResults, rerunJob, stopJob, waitForJob, type Job } from "../../core/batch-api.ts"; +import { + createJob, + downloadResults, + getJob, + assertIdempotencyKey, + isPaused, + listResults, + newIdempotencyKey, + rerunJob, + stopJob, + waitForJob, + type Job, + type ResultStatus, +} from "../../core/batch-api.ts"; +import { parseArgs } from "node:util"; import { asNumber, asString, parse, type Command, type RunContext } from "../command.ts"; -import { ToolkitError, isKeyCapReached, keyCapReached } from "../../core/errors.ts"; +import { ToolkitError, isKeyCapReached, keyCapReached, quotaExhausted } from "../../core/errors.ts"; import { printError, writeOut } from "../output.ts"; export const batch: Command = { @@ -38,14 +59,22 @@ export const batch: Command = { " --proxy-country job-level: geo-target (auto mode, or with --premium-proxy)", " --output job-level response_type (markdown|plaintext|pdf|html)", " --follow poll until the run finishes (alias: --wait)", + " --timeout with --follow: stop waiting after ms milliseconds (default 600000;", + " `600s` or `600000ms` also work)", + " --idempotency-key repeat a create safely: same key + same file and flags within 24 h", + " never creates a second job (the answer may be a 409 to look into)", " --no-signup do not auto-create a Free plan account if no key exists", " status show run status + stats", - " results [--status s] list results (successful|failed|all); paginated", + " results [--status s] list results (successful|failed|processing|all); paginated", " --out write results as JSONL instead of printing", " --download fetch each result body into (+ _manifest.jsonl)", " cancel stop an in-flight run", - " wait [--timeout ] poll until the run finishes", + " wait [--timeout ] poll until the run finishes (default 600000 ms; `600s` also works);", + " fails right away if the run is paused", " retry-failed rerun only the failed tasks (new run)", + " --include-pending also run tasks that did not finish (pending or processing)", + " --idempotency-key repeat a rerun safely: same key + same flags within 24 h never starts", + " a duplicate run (the answer may be a 409 to look into)", " --json print structured output", ].join("\n"), async run(argv: string[], ctx: RunContext): Promise { @@ -117,6 +146,8 @@ async function createCmd(rest: string[], ctx: RunContext): Promise { output: { type: "string" }, follow: { type: "boolean" }, wait: { type: "boolean" }, // back-compat alias for --follow + timeout: { type: "string" }, + "idempotency-key": { type: "string" }, "no-signup": { type: "boolean" }, json: { type: "boolean" }, }); @@ -126,6 +157,10 @@ async function createCmd(rest: string[], ctx: RunContext): Promise { const json = ctx.json || values.json === true; const file = positionals[0]; if (!file) throw needFile(); + // Chosen once, up front: checked before anything else happens, recorded in the + // run artifact, and shown in the repeat command if the outcome is unknown. + const idempotencyKey = asString(values["idempotency-key"]) ?? newIdempotencyKey(); + assertIdempotencyKey(idempotencyKey); // Capability gate first — a beta-disabled backend never gets a spec read. assertUsable("batch"); @@ -158,8 +193,20 @@ async function createCmd(rest: string[], ctx: RunContext): Promise { const policy = loadPolicy(); for (const job of jobs) assertDomainAllowed(job.url, policy); - const est = estimateCredits(jobs); - assertWithinLimits({ pages: v.validJobs, credits: est.credits }, policy, "batch"); + const est = estimateCredits(jobs, jobParams); + // The policy credit cap keeps enforcing the estimate earlier versions used, + // so a run that passed the cap before still passes. That estimate ignored + // job-level flags and nested params; when the full estimate is over the cap, + // say so instead of blocking a run that used to go through. + const legacy = legacyEstimateCredits(jobs); + assertWithinLimits({ pages: v.validJobs, credits: legacy.credits }, policy, "batch"); + if (est.credits > policy.max_credits_per_run) { + log.warn( + `This run is estimated at up to ${est.credits} credits, over the policy cap of ${policy.max_credits_per_run}. ` + + `It is still submitted: the cap checks an older estimate (${legacy.credits}) that leaves out job-level flags. ` + + `To have the cap cover this run, lower the job's cost or raise the cap: zenrows policy set max_credits_per_run ${est.credits}`, + ); + } // toJobBody validates proxy_country/premium_proxy BEFORE any HTTP call. const body = toJobBody(jobs, jobParams); @@ -177,8 +224,13 @@ async function createCmd(rest: string[], ctx: RunContext): Promise { const startedAt = new Date().toISOString(); log.step(`Submitting batch job (${body.tasks.length} tasks, ~${est.credits} credits)…`); try { - const job = await createJob(body, { apiKey }); - const finished = follow ? await waitForJob(job.job_id, { apiKey }) : job; + const timeoutMs = timeoutFlag(values.timeout); + const job = await createJob(body, { apiKey, idempotencyKey }).catch((err: unknown) => { + throw withRepeatCommand(err, "create", rest); + }); + const finished = follow ? await waitForJob(job.job_id, { apiKey, timeoutMs }).catch((err: unknown) => { + throw jobAlreadyCreated(err, job.job_id); + }) : job; const runError = runFailure(finished); const runDir = writeRun({ runId, @@ -187,7 +239,7 @@ async function createCmd(rest: string[], ctx: RunContext): Promise { startedAt, finishedAt: new Date().toISOString(), status: runError ? "error" : "ok", - request: { file, tasks: body.tasks.length, estimatedCredits: est.credits, jobParams }, + request: { file, tasks: body.tasks.length, estimatedCredits: est.credits, jobParams, idempotencyKey }, result: { jobId: job.job_id, status: finished.latest_run?.status ?? "unknown", ...failureFields(finished) }, ...(runError ? { error: runError.toJSON() } : {}), }); @@ -202,7 +254,7 @@ async function createCmd(rest: string[], ctx: RunContext): Promise { startedAt, finishedAt: new Date().toISOString(), status: "error", - request: { file, tasks: body.tasks.length, estimatedCredits: est.credits, jobParams }, + request: { file, tasks: body.tasks.length, estimatedCredits: est.credits, jobParams, idempotencyKey }, error: err instanceof ToolkitError ? err.toJSON() : { message: String(err) }, }); throw err; @@ -293,8 +345,10 @@ async function cancelCmd(rest: string[], ctx: RunContext): Promise { const id = requireId(positionals[0]); assertUsable("batch"); const apiKey = requireApiKey(); - const job = await stopJob(id, { apiKey }); - return printJob(job, ctx.json, `Stopped job ${id}`); + const run = await stopJob(id, { apiKey }); + // A successful stop call exits 0 with ok:true, as it always did, whatever the + // run's status; the status, stats and failure fields are added to the output. + return printJob({ job_id: id, latest_run: run }, ctx.json, `Stopped job ${id}`, { runFailureIsError: false }); } async function waitCmd(rest: string[], ctx: RunContext): Promise { @@ -303,19 +357,149 @@ async function waitCmd(rest: string[], ctx: RunContext): Promise { const id = requireId(positionals[0]); assertUsable("batch"); const apiKey = requireApiKey(); - const job = await waitForJob(id, { apiKey, timeoutMs: asNumber(values.timeout) }); + const job = await waitForJob(id, { apiKey, timeoutMs: timeoutFlag(values.timeout) }); return printJob(job, json, `Job ${id} finished`); } async function retryCmd(rest: string[], ctx: RunContext): Promise { - const { positionals } = parse(rest, {}); + const { values, positionals } = parse(rest, { + "include-pending": { type: "boolean" }, + "idempotency-key": { type: "string" }, + }); const id = requireId(positionals[0]); assertUsable("batch"); const apiKey = requireApiKey(); - // "Reruns and retrying failures": POST /jobs/{id}/rerun?status=failed replays - // only the failures; already-successful tasks carry over. - const job = await rerunJob(id, { apiKey, status: "failed" }); - return printJob(job, ctx.json, `Reran failed tasks for job ${id}`); + // POST /jobs/{id}/rerun?status=failed replays only the failures. With + // --include-pending it also runs every task that did not finish: a stopped or + // credit-failed run leaves unstarted tasks `pending`, and tasks cut off in + // flight can stay `processing`. A partial rerun only re-enqueues the statuses + // it names, so leaving `processing` out would carry those rows over unfinished + // and the new run could never complete. Already-successful tasks carry over. + const includePending = values["include-pending"] === true; + const status = includePending ? "failed,pending,processing" : "failed"; + // Chosen here so it is recorded in the run artifact and shown in the repeat + // command if the outcome of the rerun is unknown. + const idempotencyKey = asString(values["idempotency-key"]) ?? newIdempotencyKey(); + assertIdempotencyKey(idempotencyKey); + const runId = newRunId(); + const startedAt = new Date().toISOString(); + const request = { jobId: id, status, idempotencyKey }; + let job: Job; + try { + job = await rerunJob(id, { apiKey, status, idempotencyKey }); + } catch (err) { + const final = withRepeatCommand(err, "retry-failed", rest); + writeRun({ + runId, + command: "zenrows batch retry-failed", + capability: "batch", + startedAt, + finishedAt: new Date().toISOString(), + status: "error", + request, + error: final instanceof ToolkitError ? final.toJSON() : { message: String(final) }, + }); + throw final; + } + writeRun({ + runId, + command: "zenrows batch retry-failed", + capability: "batch", + startedAt, + finishedAt: new Date().toISOString(), + status: "ok", + request, + result: { jobId: job.job_id, status: job.latest_run?.status ?? "unknown" }, + }); + return printJob(job, ctx.json, includePending ? `Reran unfinished tasks for job ${id}` : `Reran failed tasks for job ${id}`); +} + +/** + * The job was created but following it failed (a timeout, or status reads that + * kept failing). Say so, so nobody runs `create` again for a job that exists. + */ +function jobAlreadyCreated(err: unknown, jobId: string): unknown { + if (!(err instanceof ToolkitError)) return err; + const e = err.toJSON(); + const follow = `zenrows batch wait ${jobId}`; + return new ToolkitError({ + ...e, + next_action: `Job ${jobId} was created, so do not run \`create\` again: only following it failed. Follow the job with \`${follow}\`.`, + suggested_commands: [follow, ...(e.suggested_commands ?? []).filter((c) => c !== follow)], + }); +} + +/** + * The arguments that repeat `sub` with `key`: the key option goes right after + * the subcommand, ahead of everything the caller typed, so it is read as an + * option even when the caller used a `--` terminator. Any key option the caller + * gave (before `--`; after it, it was a positional) is dropped, so the only one + * left is `key`, the key that was actually sent. + */ +export function repeatArgs(sub: string, rest: string[], key: string): string[] { + const { tokens } = parseArgs({ + args: rest, + options: { "idempotency-key": { type: "string" } }, + strict: false, + allowPositionals: true, + tokens: true, + }); + const drop = new Set(); + for (const t of tokens ?? []) { + if (t.kind !== "option" || t.name !== "idempotency-key") continue; + drop.add(t.index); + if (!t.inlineValue && t.value !== undefined) drop.add(t.index + 1); + } + return [sub, "--idempotency-key", key, ...rest.filter((_, i) => !drop.has(i))]; +} + +/** Quote one argument for a POSIX shell, so a copied command passes it unchanged. */ +export function shellQuote(a: string): string { + return /^[\w./:=@,+-]+$/.test(a) ? a : `'${a.replace(/'/g, "'\\''")}'`; +} + +/** + * When a submit or rerun ended with an unknown outcome (the error carries the + * Idempotency-Key it was sent with), add the exact command that repeats it with + * that key: same arguments, so the API sees the identical request and returns + * the original answer instead of creating a second job or run. + */ +function withRepeatCommand(err: unknown, sub: string, rest: string[]): unknown { + if (!(err instanceof ToolkitError) || !err.idempotency_key) return err; + const cmd = ["zenrows", "batch", ...repeatArgs(sub, rest, err.idempotency_key)].map(shellQuote).join(" "); + const e = err.toJSON(); + return new ToolkitError({ + ...e, + next_action: `${e.next_action} Run: ${cmd}`, + suggested_commands: [cmd, ...(e.suggested_commands ?? []).filter((c) => c !== cmd)], + }); +} + +/** + * `--timeout` in milliseconds, as it always was: a bare number is read exactly + * as before. `s` (seconds) and `ms` are also accepted, and must be + * positive. A value that is none of these was silently ignored before (the + * 600000 ms default applied); it still is, with a warning, so no script breaks. + * Returns undefined when unset or ignored. + */ +export function timeoutFlag(v: unknown): number | undefined { + if (v === undefined) return undefined; + const raw = String(v).trim(); + const bare = asNumber(raw); + if (bare !== undefined) return bare; + const m = /^(\d+(?:\.\d+)?)(ms|s)$/.exec(raw); + if (m) { + const n = Number(m[1]) * (m[2] === "s" ? 1000 : 1); + if (Number.isFinite(n) && n > 0) return n; + throw new ToolkitError({ + code: "INVALID_USAGE", + message: `Invalid --timeout value '${raw}'.`, + likely_cause: "--timeout with a unit must be a positive number of seconds (`s`) or milliseconds (`ms`).", + next_action: "Pass milliseconds (--timeout 600000), or a unit: --timeout 600s or --timeout 600000ms.", + }); + } + log.warn(`--timeout '${raw}' is not a number of milliseconds (or s / ms); using the default of 600000 ms.`); + return undefined; } /** The run's failure_reason / failure_detail, when the API reports them. */ @@ -336,16 +520,39 @@ export function runFailure(job: Job): ToolkitError | null { if (job.latest_run?.status !== "failed") return null; const { failure_reason, failure_detail } = failureFields(job); const where = `batch job ${job.job_id}`; + const retry = `zenrows batch retry-failed ${job.job_id} --include-pending`; if (isKeyCapReached(failure_reason)) { - return keyCapReached(where, { status: null, detail: failure_detail }); + const e = keyCapReached(where, { status: null, detail: failure_detail }).toJSON(); + return new ToolkitError({ + ...e, + server_code: failure_reason, + reason: "api_key_cap_reached", + next_action: `${e.next_action} Then run the tasks that did not finish: \`${retry}\`.`, + suggested_commands: [...(e.suggested_commands ?? []), retry], + }); + } + if (failure_reason === "insufficient_credits") { + // Same code as before (BATCH_FAILED); `reason` and the credit advice are added. + const credits = quotaExhausted(where, undefined, { status: 402, detail: failure_detail }).next_action; + return new ToolkitError({ + code: "BATCH_FAILED", + server_code: failure_reason, + reason: "insufficient_credits", + message: `Batch job ${job.job_id} failed.`, + likely_cause: `insufficient_credits${failure_detail ? `: ${failure_detail}` : ""}`, + next_action: `The run stopped when the account ran out of credits. ${credits} Then run the tasks that did not finish: \`${retry}\`.`, + suggested_commands: ["zenrows usage", retry], + }); } const why = [failure_reason, failure_detail].filter(Boolean).join(": "); return new ToolkitError({ code: "BATCH_FAILED", + ...(failure_reason ? { server_code: failure_reason } : {}), message: `Batch job ${job.job_id} failed.`, likely_cause: why || "The run ended in status failed without a reason.", - next_action: "Inspect the per-task results, fix the cause, then rerun the failed tasks.", - suggested_commands: [`zenrows batch results ${job.job_id} --status failed`, `zenrows batch retry-failed ${job.job_id}`], + next_action: + "Inspect the per-task results, fix the cause, then rerun the tasks that failed or never started.", + suggested_commands: [`zenrows batch results ${job.job_id} --status failed`, retry], }); } @@ -355,43 +562,60 @@ export function runFailure(job: Job): ToolkitError | null { * deliberate outcomes (someone cancelled or removed the run), so they exit 0 * but print as a warning, not a success. */ -function printJob(job: Job, json: boolean, headline: string): number { +function printJob(job: Job, json: boolean, headline: string, opts: { runFailureIsError?: boolean } = {}): number { const run = job.latest_run ?? ({} as Job["latest_run"]); const stats = run.stats; const failure = failureFields(job); - const err = runFailure(job); + const err = opts.runFailureIsError === false ? null : runFailure(job); + const paused = isPaused(run); if (json) { log.out( JSON.stringify( - { ok: !err, jobId: job.job_id, status: run.status, stats, ...failure, ...(err ? { error: err.toJSON() } : {}) }, + { + ok: !err, + jobId: job.job_id, + status: run.status, + ...(paused ? { paused: true } : {}), + stats, + ...failure, + ...(err ? { error: err.toJSON() } : {}), + }, null, 2, ), ); return err ? 1 : 0; } - const line = `${headline} · status: ${run.status ?? "unknown"}`; + const line = `${headline} · status: ${run.status ?? "unknown"}${paused ? " (paused)" : ""}`; if (err) log.error(line); - else if (run.status === "stopped" || run.status === "deleted") log.warn(line); + else if (paused || run.status === "stopped" || run.status === "deleted" || run.status === "failed") log.warn(line); else log.success(line); if (stats) { log.info(` ${stats.completed}/${stats.total} completed · ${stats.successful} successful · ${stats.failed} failed`); } + if (paused) { + log.dim(" The run is paused: no new task is picked up, though tasks already started may still finish."); + } + if (run.status === "stopped" && stats && stats.completed < stats.total) { + log.dim( + ` ${stats.total - stats.completed} task(s) did not complete. Run them: zenrows batch retry-failed ${job.job_id} --include-pending`, + ); + } if (failure.failure_reason) log.info(` failure_reason: ${failure.failure_reason}`); if (failure.failure_detail) log.info(` failure_detail: ${failure.failure_detail}`); if (err) printError(err, false); return err ? 1 : 0; } -function normalizeResultStatus(v?: string): "successful" | "failed" | "all" | undefined { +function normalizeResultStatus(v?: string): ResultStatus | undefined { if (!v) return undefined; const s = v.toLowerCase(); - if (s === "successful" || s === "failed" || s === "all") return s; + if (s === "successful" || s === "failed" || s === "processing" || s === "all") return s; throw new ToolkitError({ code: "INVALID_USAGE", message: `Invalid --status: ${v}`, - likely_cause: "Only successful | failed | all are supported.", - next_action: "Use --status successful | failed | all (default: all).", + likely_cause: "Only successful | failed | processing | all are supported.", + next_action: "Use --status successful | failed | processing | all (default: all).", }); } diff --git a/src/core/batch-api.ts b/src/core/batch-api.ts index cb0fad9..c1a8e7f 100644 --- a/src/core/batch-api.ts +++ b/src/core/batch-api.ts @@ -9,14 +9,25 @@ * redacted from any logged output. `fetchImpl` is injectable for tests, mirroring * `usage.ts` / `agent-account.ts`. * + * GETs retry transient failures (429/502/503/504, honoring `Retry-After`) with + * backoff; a POST retries only when it carries an `Idempotency-Key`, which + * `createJob` and `rerunJob` send. The key is reused across the automatic + * retries of one call, so a retry within the key's 24 hours never starts a + * second job or run. Its answer is not guaranteed to be the original: the API + * may replay it, answer 409 (no recorded answer yet, or for a rerun a run the + * first attempt already started), or refuse before checking the key (e.g. 402 + * out of credits). After a lost attempt, any of these is reported as "an + * earlier attempt may have gone through", never as a reason to resubmit. + * * Deferred (documented, not wired here): CSV upload (`/job_inputs` + presigned * PUT), open/queue jobs (`/tasks`, `last_batch`), scheduled jobs - * (`schedule` / `PUT /schedule` / `schedule/state`), webhooks + HMAC, ZIP - * archive export, and the `Idempotency-Key` header. + * (`schedule` / `PUT /schedule` / `schedule/state`), webhooks + HMAC, and ZIP + * archive export. */ import { mkdirSync, writeFileSync } from "node:fs"; import { join } from "node:path"; -import { ToolkitError, isKeyCapReached, keyCapReached, quotaExhausted } from "./errors.ts"; +import { randomUUID } from "node:crypto"; +import { ToolkitError, isKeyCapReached, keyCapReached, quotaExhausted, type ErrorCode } from "./errors.ts"; import { readAccount } from "./agent-account.ts"; import { registerSecret } from "./logger.ts"; import { CLI_USER_AGENT } from "./config.ts"; @@ -49,6 +60,8 @@ export interface JobRun { failure_reason?: string; /** Human-readable detail for `failure_reason`. */ failure_detail?: string; + /** `paused` while the run is paused: no new task is picked up; tasks already started may still finish. */ + pause_state?: string; [k: string]: unknown; } @@ -84,19 +97,90 @@ interface ProblemJson { /** Terminal run states — a run stops progressing once it reaches one of these. */ export const TERMINAL_STATUSES: ReadonlySet = new Set(["completed", "stopped", "failed", "deleted"]); +/** Result filters the API accepts on `GET /jobs/{id}/results`. */ +export type ResultStatus = "successful" | "failed" | "processing" | "all"; + interface RequestOpts { apiKey: string; body?: unknown; query?: Record; + headers?: Record; fetchImpl?: typeof fetch; timeoutMs?: number; + sleepImpl?: (ms: number) => Promise; + /** Shape check for a 2xx answer; a failed check is treated like an unreadable answer. */ + validate?: (v: unknown) => boolean; +} + +/** True for an answer that names a job, as `POST /jobs` and `POST /jobs/{id}/rerun` do. */ +function hasJobId(v: unknown): boolean { + return typeof v === "object" && v !== null && typeof (v as { job_id?: unknown }).job_id === "string" && (v as { job_id: string }).job_id !== ""; +} + +const MAX_RETRIES = 3; +/** Never wait longer than this on a `Retry-After`, in seconds. */ +const MAX_RETRY_AFTER_SECONDS = 30; +const READ_RETRY_STATUSES: ReadonlySet = new Set([429, 502, 503, 504]); +// 429 on a POST is an account limit (e.g. active jobs): retrying only waits for a slot that may never free. +const WRITE_RETRY_STATUSES: ReadonlySet = new Set([502, 503, 504]); + +/** + * The statuses to retry, or undefined when the request must run once: a GET is + * safe to repeat, a POST only with an Idempotency-Key. + */ +function retryStatuses(method: string, headers: Record): ReadonlySet | undefined { + if (method === "GET") return READ_RETRY_STATUSES; + return headers["Idempotency-Key"] ? WRITE_RETRY_STATUSES : undefined; +} + +/** 250 ms x 2^attempt, capped at 10 s, +/-20% jitter. */ +function backoffMs(attempt: number): number { + return Math.min(250 * 2 ** attempt, 10_000) * (0.8 + Math.random() * 0.4); +} + +/** `Retry-After` in whole seconds, as milliseconds, at most `MAX_RETRY_AFTER_SECONDS`. */ +function retryAfterMs(value: string | null): number | undefined { + return value && /^\d+$/.test(value.trim()) ? Math.min(Number(value), MAX_RETRY_AFTER_SECONDS) * 1000 : undefined; +} + +const defaultSleep = (ms: number) => new Promise((r) => setTimeout(r, ms)); + +/** A fresh Idempotency-Key for one logical submit or rerun. */ +export function newIdempotencyKey(): string { + return randomUUID(); +} + +/** The longest Idempotency-Key the Batch API documents. */ +export const MAX_IDEMPOTENCY_KEY_LENGTH = 128; + +/** + * Throw before any request if `key` cannot work as an Idempotency-Key. HTTP + * strips a header value's leading and trailing whitespace, so a blank key + * reaches the API empty (no deduplication at all) and a padded key reaches it + * as a different key than the one shown to the user. Control characters can't + * be sent in a header. + */ +export function assertIdempotencyKey(key: string): void { + let reason: string | undefined; + if (key.trim() === "") reason = "it is empty or only whitespace"; + else if (key !== key.trim()) reason = "it starts or ends with whitespace"; + else if (/[\u0000-\u001f\u007f]/.test(key)) reason = "it contains a control character"; + else if (key.length > MAX_IDEMPOTENCY_KEY_LENGTH) reason = `it is longer than ${MAX_IDEMPOTENCY_KEY_LENGTH} characters`; + if (!reason) return; + throw new ToolkitError({ + code: "INVALID_USAGE", + message: `Invalid --idempotency-key: ${reason}.`, + likely_cause: + "An Idempotency-Key must be 1 to 128 characters, with no leading or trailing whitespace and no control characters; otherwise the API would not see the key you meant.", + next_action: "Pass a non-empty key such as a UUID, or leave --idempotency-key out to have one generated.", + }); } /** * Perform a Batch API request. Sets `X-API-Key`, adds a JSON content-type when a - * body is present, parses the JSON response, and on a non-2xx parses the - * problem+json body into a normalized ToolkitError. Returns `null` for empty - * (204 / no-body) responses. + * body is present, retries transient failures as `retryStatuses` allows, parses + * the JSON response, and on a non-2xx parses the problem+json body into a + * normalized ToolkitError. Returns `null` for empty (204 / no-body) responses. */ export async function batchRequest(method: string, path: string, opts: RequestOpts): Promise { registerSecret(opts.apiKey); @@ -105,57 +189,185 @@ export async function batchRequest(method: string, path: string, opts: Reques if (v !== undefined && v !== null) url.searchParams.set(k, v); } const doFetch = opts.fetchImpl ?? fetch; - const controller = new AbortController(); - const timeout = setTimeout(() => controller.abort(), opts.timeoutMs ?? 60_000); + const sleep = opts.sleepImpl ?? defaultSleep; const headers: Record = { "X-API-Key": opts.apiKey, Accept: "application/json", "User-Agent": CLI_USER_AGENT, ...agentClientHeader(), + ...(opts.headers ?? {}), }; if (opts.body !== undefined) headers["Content-Type"] = "application/json"; - - let res: Response; - try { - res = await doFetch(url.toString(), { - method, - headers, - body: opts.body !== undefined ? JSON.stringify(opts.body) : undefined, - signal: controller.signal, - }); - } catch (err) { + const retryOn = retryStatuses(method, headers); + // A write sent with an Idempotency-Key. Only these are retried, so only these + // can end with an attempt whose result is unknown. + const key = headers["Idempotency-Key"]; + + // Set once an attempt got no usable answer (no response, or a 5xx): the + // server may have done the work, so every later failure of this call is an + // unknown outcome too, whatever its status (a 402 or 409 included). + let earlierMayHaveLanded = false; + for (let n = 0; ; n++) { + const canRetry = retryOn !== undefined && n < MAX_RETRIES; + const controller = new AbortController(); + const timeout = setTimeout(() => controller.abort(), opts.timeoutMs ?? 60_000); + let res: Response; + let text: string; + try { + res = await doFetch(url.toString(), { + method, + headers, + body: opts.body !== undefined ? JSON.stringify(opts.body) : undefined, + signal: controller.signal, + }); + text = await res.text(); + } catch (err) { + clearTimeout(timeout); + if (canRetry) { + earlierMayHaveLanded = true; + await sleep(backoffMs(n)); + continue; + } + const failure = new ToolkitError({ + code: "BACKEND_UNAVAILABLE", + message: "Could not reach the Zenrows Batch API.", + likely_cause: err instanceof Error ? err.message : String(err), + next_action: + "Check connectivity and retry. Override the host with ZENROWS_BATCH_API_BASE if you are testing against staging.", + suggested_commands: ["zenrows status"], + }); + // Unknown: the request may have reached the server before the connection failed. + throw key ? outcomeUnknown(failure, key) : failure; + } clearTimeout(timeout); - throw new ToolkitError({ - code: "BACKEND_UNAVAILABLE", - message: "Could not reach the Zenrows Batch API.", - likely_cause: err instanceof Error ? err.message : String(err), - next_action: - "Check connectivity and retry. Override the host with ZENROWS_BATCH_API_BASE if you are testing against staging.", - suggested_commands: ["zenrows status"], - }); + + if (canRetry && retryOn.has(res.status)) { + if (res.status >= 500) earlierMayHaveLanded = true; + await sleep(retryAfterMs(res.headers.get("retry-after")) ?? backoffMs(n)); + continue; + } + if (res.status < 200 || res.status >= 300) { + const failure = problemToError(res.status, text, method, path, earlierMayHaveLanded); + if (key && res.status >= 500) throw outcomeUnknown(failure, key); + // A definite refusal of this attempt, but an earlier one may have been accepted. + if (key && earlierMayHaveLanded) throw outcomeUnknown(failure, key, { keepAdvice: true, earlier: res.status !== 409 }); + throw failure; + } + + // A 2xx: the server accepted this attempt. If its answer can't be read, a + // write's result (the job or run it created) is unknown, not failed. + let parsed: unknown = null; + let readable = true; + if (text) { + try { + parsed = JSON.parse(text); + } catch { + readable = false; + } + } + if (readable && opts.validate && !opts.validate(parsed)) readable = false; + if (!readable) { + const failure = new ToolkitError({ + code: "BATCH_FAILED", + message: `The Batch API accepted ${method} ${path} (HTTP ${res.status}) but its answer could not be read.`, + likely_cause: text ? `Unexpected response: ${text.slice(0, 240)}` : "The response body was empty.", + next_action: "Retry, or inspect the job in the Zenrows dashboard.", + }); + throw key ? outcomeUnknown(failure, key, { accepted: true }) : failure; + } + return parsed as T; } - clearTimeout(timeout); +} - const text = await res.text(); - if (res.status < 200 || res.status >= 300) { - throw problemToError(res.status, text, method, path); +/** + * A write (POST with an Idempotency-Key) whose outcome is unknown: the server + * may have done the work even though no usable answer arrived. Attach the key + * and say what is safe; the command layer adds the exact command to repeat. + * + * - `accepted`: this attempt was answered 2xx but the answer was unreadable. + * - `keepAdvice`: this attempt was refused with its own advice (402, 409, ...), + * kept in front; `earlier` adds that an earlier attempt may have gone through. + */ +function outcomeUnknown( + err: ToolkitError, + key: string, + opts: { keepAdvice?: boolean; earlier?: boolean; accepted?: boolean } = {}, +): ToolkitError { + const e = err.toJSON(); + let lead: string; + if (opts.accepted) { + lead = "The API accepted the request, so the job or run probably exists, but its answer could not be read."; + } else if (opts.keepAdvice) { + lead = opts.earlier + ? `${e.next_action} An earlier attempt of this request got no usable answer and may have been accepted.` + : e.next_action; + } else { + lead = "The request may have been accepted even though no usable answer arrived."; } - if (!text) return null as T; - try { - return JSON.parse(text) as T; - } catch { - throw new ToolkitError({ - code: "BATCH_FAILED", - message: "The Batch API response was not valid JSON.", - likely_cause: text.slice(0, 240), - next_action: "Retry, or inspect the job in the Zenrows dashboard.", - }); + return new ToolkitError({ + ...e, + idempotency_key: key, + next_action: + `${lead} Do not resubmit with a new key: that can create a second job or run. ` + + `Repeating the identical request with the same Idempotency-Key (${key}) within 24 hours of the first attempt cannot create a duplicate; after that the key is forgotten and counts as new. ` + + "The answer to a repeat is not guaranteed to be the original: the API may replay it, answer 409 (no recorded answer yet, or a rerun whose run is already in progress), " + + "or refuse before it looks at the key (for example 402 when the account is out of credits). " + + "Unless the repeat comes back with the job, check your jobs (`zenrows batch status ` or the dashboard) before doing anything else.", + }); +} + +/** + * How to free the slot a 429 names. Each kind frees differently: a scheduled job + * counts until it is deleted, whatever its runs do; an open job counts while it + * is open; a closed job counts while its run is in progress. + */ +function quotaAdvice(detail: string): string { + if (/scheduled job/i.test(detail)) { + return "Scheduled jobs count toward their limit until they are deleted, whatever their runs are doing. Delete a scheduled job you no longer need, then retry."; + } + return "Active jobs count until they are done: a closed job while its run is in progress (wait for it, or stop it with `zenrows batch cancel `), and an open job until it is deleted, or closed and its run has ended. Free one, then retry."; +} + +/** Add `server_code` and `reason` to a shared error; its `code` and advice stay as they are. */ +function withDetail(err: ToolkitError, serverCode: string, reason: string): ToolkitError { + return new ToolkitError({ ...err.toJSON(), ...(serverCode ? { server_code: serverCode } : {}), reason }); +} + +/** + * Specific guidance for the 409s a create, rerun or stop can answer with. + * `retried` is true when an earlier attempt of the same call (same + * Idempotency-Key) failed in transit, so it may have been accepted. + */ +function conflictAdvice(serverCode: string, detail: string, path: string, retried: boolean): string | undefined { + const isRerun = /\/rerun$/.test(path); + switch (serverCode) { + case "no_matching_tasks": + return "No task in the previous run has the statuses you asked to retry. If tasks did not finish (a stopped or credit-failed run), retry them with `zenrows batch retry-failed --include-pending`, or check the run with `zenrows batch status `."; + case "run_not_terminal": + if (retried && isRerun) { + return "An earlier attempt of this rerun failed in transit and may have started the new run, which is why the job now has a run in progress. Check it with `zenrows batch status ` before rerunning again."; + } + if (/ingestion did not complete/i.test(detail)) { + return "The previous run never finished receiving its tasks, so it cannot be rerun. Delete the job and submit the URLs again as a new job."; + } + return "The current run is still in progress. Wait for it to finish with `zenrows batch wait `, or stop it with `zenrows batch cancel `, then retry."; + case "partial_rerun_window_expired": + return "Retrying only some tasks is allowed for 10 days after a run starts. Rerun everything with a full rerun, or submit the URLs again as a new job."; + case "run_not_stoppable": + return "The run already finished, so there is nothing to stop. Check it with `zenrows batch status `."; + case "idempotency_key_conflict": + if (retried) { + return "An earlier attempt of this request failed in transit, and the API holds that attempt under the same Idempotency-Key without a recorded answer: it may still be processing, or it may have failed partway."; + } + return "This Idempotency-Key was already used, either for a different request, or for this one without a recorded answer (still processing, or failed partway). Retrying with the same key within 24 hours of the first request cannot create a duplicate, but it may keep answering 409: check your jobs (`zenrows batch status ` or the dashboard) first. Use a new key only for a deliberately new job, after checking that the earlier one is not the job you want."; + default: + return undefined; } } /** Map an RFC 7807 problem body + HTTP status to a normalized ToolkitError. */ -function problemToError(status: number, body: string, method: string, path: string): ToolkitError { +function problemToError(status: number, body: string, method: string, path: string, retried = false): ToolkitError { let problem: ProblemJson = {}; try { problem = JSON.parse(body) as ProblemJson; @@ -163,11 +375,13 @@ function problemToError(status: number, body: string, method: string, path: stri // non-JSON error body — fall through with an empty problem } const serverCode = problem.code ?? ""; - const detail = problem.detail || problem.title || body.slice(0, 240) || `HTTP ${status}`; + const detail = (problem.detail || problem.title || body.slice(0, 240) || `HTTP ${status}`).replace(/\.\s*$/, ""); const cause = `HTTP ${status}${serverCode ? ` (${serverCode})` : ""} for ${method} ${path}: ${detail}`; + const sc = serverCode ? { server_code: serverCode } : {}; if (status === 403) { return new ToolkitError({ + ...sc, code: "BATCH_ACCESS_DENIED", message: "The Batch API rejected this request (access denied).", likely_cause: `${cause}. Batch is not enabled for this account or API key.`, @@ -178,6 +392,7 @@ function problemToError(status: number, body: string, method: string, path: stri } if (status === 401) { return new ToolkitError({ + ...sc, code: "AUTH_INVALID", message: "Zenrows rejected the API key for the Batch API.", likely_cause: cause, @@ -187,6 +402,7 @@ function problemToError(status: number, body: string, method: string, path: stri } if (status === 404) { return new ToolkitError({ + ...sc, code: "BATCH_NOT_FOUND", message: "Batch job, run, or task not found.", likely_cause: `${cause}. The id may be wrong or the job is not owned by this account.`, @@ -194,17 +410,23 @@ function problemToError(status: number, body: string, method: string, path: stri }); } if (status === 429) { + const scheduled = /scheduled job/i.test(detail); return new ToolkitError({ + ...sc, code: "BATCH_QUOTA_EXCEEDED", + reason: scheduled ? "too_many_scheduled_jobs" : "too_many_active_jobs", message: "Batch quota exceeded.", - likely_cause: `${cause}. An account limit was reached (e.g. the maximum of 3 concurrent active jobs).`, - next_action: - "Wait for an in-flight job to finish (or stop one with `zenrows batch cancel `), then retry.", + likely_cause: `${cause}. An account limit was reached: 3 active jobs and 5 scheduled jobs by default.`, + next_action: quotaAdvice(detail), suggested_commands: ["zenrows batch status "], }); } if (status === 402 && isKeyCapReached(serverCode)) { - return keyCapReached(`${method} ${path}`, { status: 402, detail: problem.detail || problem.title || undefined }); + return withDetail( + keyCapReached(`${method} ${path}`, { status: 402, detail: problem.detail || problem.title || undefined }), + serverCode, + "api_key_cap_reached", + ); } if (status === 402) { // Out of credits ("Subscription has no credit available") — the same @@ -212,10 +434,14 @@ function problemToError(status: number, body: string, method: string, path: stri // a claim link (unclaimed Free plan) or dashboard/upgrade pointer, not a // generic "fix the problem and retry". const acct = readAccount(); - return quotaExhausted(`${method} ${path}`, acct?.unclaimed ? acct.claimUrl : undefined, { - status: 402, - detail: problem.detail || problem.title || undefined, - }); + return withDetail( + quotaExhausted(`${method} ${path}`, acct?.unclaimed ? acct.claimUrl : undefined, { + status: 402, + detail: problem.detail || problem.title || undefined, + }), + serverCode, + "insufficient_credits", + ); } // 400 invalid_argument, 409 conflict, 5xx, etc. @@ -226,13 +452,15 @@ function problemToError(status: number, body: string, method: string, path: stri .join("; ")}` : ""; return new ToolkitError({ + ...sc, code: "BATCH_FAILED", message: `Batch request failed (HTTP ${status}).`, likely_cause: `${cause}.${invalid}`, next_action: - status >= 500 - ? "This is transient — retry with a short backoff." - : "Fix the reported problem and retry. For validation errors, correct the flagged tasks.", + conflictAdvice(serverCode, detail, path, retried) ?? + (status >= 500 + ? "This is transient: retry with a short backoff." + : "Fix the reported problem and retry. For validation errors, correct the flagged tasks."), }); } @@ -240,41 +468,70 @@ interface CallOpts { apiKey: string; fetchImpl?: typeof fetch; timeoutMs?: number; + sleepImpl?: (ms: number) => Promise; } -/** Create a job. Accepts both 201 (<10k tasks) and 202 (≥10k) as success. */ -export function createJob(body: unknown, opts: CallOpts): Promise { - return batchRequest("POST", "/jobs", { apiKey: opts.apiKey, body, fetchImpl: opts.fetchImpl, timeoutMs: opts.timeoutMs }); +/** + * Create a job. Accepts both 201 (tasks stored before answering) and 202 + * (accepted for asynchronous ingestion, used for large submits) as success. + * Sends an `Idempotency-Key`, a fresh one unless given, that the automatic + * retries of this call reuse; repeating the command later only dedupes when + * the caller passes the same key again. + */ +export async function createJob(body: unknown, opts: CallOpts & { idempotencyKey?: string }): Promise { + if (opts.idempotencyKey !== undefined) assertIdempotencyKey(opts.idempotencyKey); + return batchRequest("POST", "/jobs", { + apiKey: opts.apiKey, + body, + validate: hasJobId, + headers: { "Idempotency-Key": opts.idempotencyKey ?? newIdempotencyKey() }, + fetchImpl: opts.fetchImpl, + timeoutMs: opts.timeoutMs, + sleepImpl: opts.sleepImpl, + }); } /** Read a job (`latest_run.status` + `latest_run.stats`). */ export function getJob(id: string, opts: CallOpts): Promise { - return batchRequest("GET", `/jobs/${encodeURIComponent(id)}`, { apiKey: opts.apiKey, fetchImpl: opts.fetchImpl, timeoutMs: opts.timeoutMs }); + return batchRequest("GET", `/jobs/${encodeURIComponent(id)}`, { + apiKey: opts.apiKey, + validate: hasJobId, + fetchImpl: opts.fetchImpl, + timeoutMs: opts.timeoutMs, + sleepImpl: opts.sleepImpl, + }); } -/** Abort an in-flight run. */ -export function stopJob(id: string, opts: CallOpts): Promise { - return batchRequest("POST", `/jobs/${encodeURIComponent(id)}/stop`, { apiKey: opts.apiKey, fetchImpl: opts.fetchImpl, timeoutMs: opts.timeoutMs }); +/** Abort an in-flight run. The API answers with the refreshed run, not the job. */ +export function stopJob(id: string, opts: CallOpts): Promise { + return batchRequest("POST", `/jobs/${encodeURIComponent(id)}/stop`, { apiKey: opts.apiKey, fetchImpl: opts.fetchImpl, timeoutMs: opts.timeoutMs }); } /** - * Replay a job as a new run. `status="failed"` re-runs only the failures - * (append `,pending` to also include tasks that never started). Already- - * successful tasks carry over, so you only re-scrape what failed. + * Replay a job as a new run. `status="failed"` re-runs only the failures; + * `failed,pending,processing` also includes every task that did not finish + * (after a stopped or credit-failed run, unstarted tasks stay `pending` and + * tasks cut off mid-flight can stay `processing`). Already-successful tasks + * carry over, so you only re-scrape what failed. Sends an `Idempotency-Key` + * like `createJob`. */ -export function rerunJob(id: string, opts: CallOpts & { status?: string }): Promise { +export async function rerunJob(id: string, opts: CallOpts & { status?: string; idempotencyKey?: string }): Promise { + if (opts.idempotencyKey !== undefined) assertIdempotencyKey(opts.idempotencyKey); return batchRequest("POST", `/jobs/${encodeURIComponent(id)}/rerun`, { apiKey: opts.apiKey, + validate: hasJobId, query: opts.status ? { status: opts.status } : undefined, + headers: { "Idempotency-Key": opts.idempotencyKey ?? newIdempotencyKey() }, fetchImpl: opts.fetchImpl, timeoutMs: opts.timeoutMs, + sleepImpl: opts.sleepImpl, }); } /** List all result rows for a run, paginating over the `cursor` until `next_cursor` is null. */ export async function listResults( id: string, - opts: CallOpts & { status?: "successful" | "failed" | "all" }, + opts: CallOpts & { status?: ResultStatus }, ): Promise { const all: ResultRow[] = []; let cursor: string | undefined; @@ -284,6 +541,7 @@ export async function listResults( query: { status: opts.status, cursor }, fetchImpl: opts.fetchImpl, timeoutMs: opts.timeoutMs, + sleepImpl: opts.sleepImpl, }); if (page?.results) all.push(...page.results); cursor = page?.next_cursor ?? undefined; @@ -397,10 +655,18 @@ export async function downloadResults( return outcome; } +/** True when the run is paused: no new task is picked up, though tasks already started may still finish. */ +export function isPaused(run: JobRun | undefined): boolean { + // A stopped run can keep `pause_state: paused`; only a live run is paused. + return run?.pause_state === "paused" && (run.status === "running" || run.status === "pending"); +} + /** * Poll `GET /jobs/{id}` until `latest_run.status` is terminal, backing off - * 2s → ×1.5 → capped at 15s. Throws BATCH_FAILED on timeout. `sleepImpl` is - * injectable for tests. + * 2s, x1.5, capped at 15s. Throws BATCH_FAILED on timeout (`reason: + * wait_timeout`), and right away when the run is paused (`reason: run_paused`), + * since a paused run picks up no new task and the wait would only time out. + * `sleepImpl` is injectable for tests. */ export async function waitForJob( id: string, @@ -410,11 +676,24 @@ export async function waitForJob( const deadline = Date.now() + (opts.timeoutMs ?? 600_000); let delay = 2000; for (;;) { - const job = await getJob(id, { apiKey: opts.apiKey, fetchImpl: opts.fetchImpl }); + const job = await getJob(id, { apiKey: opts.apiKey, fetchImpl: opts.fetchImpl, sleepImpl: opts.sleepImpl }); if (job.latest_run && TERMINAL_STATUSES.has(job.latest_run.status)) return job; + if (isPaused(job.latest_run)) { + // A paused run picks up no new task, so waiting on would only end in the + // timeout error below. Fail the same way (same code, exit 1), now. + throw new ToolkitError({ + code: "BATCH_FAILED", + reason: "run_paused", + message: `Batch job ${id} is paused, so it will not finish while it stays paused.`, + likely_cause: `The run is ${job.latest_run!.status} with pause_state paused: no new task is picked up, though tasks already started may still finish.`, + next_action: "Resume the run (dashboard or API), then wait again with `zenrows batch wait `.", + suggested_commands: [`zenrows batch status ${id}`], + }); + } if (Date.now() > deadline) { throw new ToolkitError({ code: "BATCH_FAILED", + reason: "wait_timeout", message: `Timed out waiting for batch job ${id} to finish.`, likely_cause: `The run did not reach a terminal state within ${Math.round((opts.timeoutMs ?? 600_000) / 1000)}s.`, next_action: "Re-check progress with `zenrows batch status `, or raise --timeout.", diff --git a/src/core/errors.ts b/src/core/errors.ts index 1c4da84..a59a21a 100644 --- a/src/core/errors.ts +++ b/src/core/errors.ts @@ -61,10 +61,21 @@ export interface ToolkitErrorShape { suggested_commands?: string[]; /** The API's own `code`, when the error came from an API answer that had one. */ server_code?: string; + /** + * A finer, stable reason within `code`, e.g. `insufficient_credits` or + * `too_many_scheduled_jobs`. Added alongside `code`, which keeps its value. + */ + reason?: string; /** The crawl the error is about, once it exists. */ crawl_id?: string; /** Seconds to wait before a retry (the `Retry-After` header). */ retry_after?: number; + /** + * The Idempotency-Key a write was sent with, when its outcome is unknown. Repeat + * the identical request with this key to get the original answer instead of a + * second job or run. + */ + idempotency_key?: string; } export class ToolkitError extends Error { @@ -73,8 +84,10 @@ export class ToolkitError extends Error { readonly next_action: string; readonly suggested_commands: string[]; readonly server_code?: string; + readonly reason?: string; readonly crawl_id?: string; readonly retry_after?: number; + readonly idempotency_key?: string; constructor(shape: ToolkitErrorShape) { super(shape.message); @@ -84,8 +97,10 @@ export class ToolkitError extends Error { this.next_action = shape.next_action; this.suggested_commands = shape.suggested_commands ?? []; this.server_code = shape.server_code; + this.reason = shape.reason; this.crawl_id = shape.crawl_id; this.retry_after = shape.retry_after; + this.idempotency_key = shape.idempotency_key; } toJSON(): ToolkitErrorShape { @@ -96,8 +111,10 @@ export class ToolkitError extends Error { next_action: this.next_action, suggested_commands: this.suggested_commands, ...(this.server_code ? { server_code: this.server_code } : {}), + ...(this.reason ? { reason: this.reason } : {}), ...(this.crawl_id ? { crawl_id: this.crawl_id } : {}), ...(this.retry_after !== undefined ? { retry_after: this.retry_after } : {}), + ...(this.idempotency_key ? { idempotency_key: this.idempotency_key } : {}), }; } } diff --git a/src/core/policy.ts b/src/core/policy.ts index fd6badd..121ec7f 100644 --- a/src/core/policy.ts +++ b/src/core/policy.ts @@ -111,8 +111,8 @@ export function assertWithinLimits( code: "POLICY_LIMIT_EXCEEDED", message: `This ${context} submits ${requested.pages} page(s), over the policy cap of ${policy.max_pages_per_run}.`, likely_cause: "policy.max_pages_per_run bounds how many URLs a single run may submit.", - next_action: `Split the job into runs of ≤${policy.max_pages_per_run} page(s), or raise max_pages_per_run in policy.json.`, - suggested_commands: ["zenrows policy show"], + next_action: `Split the job into runs of at most ${policy.max_pages_per_run} page(s), or raise the cap: \`zenrows policy set max_pages_per_run ${requested.pages}\`. The default is ${defaultPolicy().max_pages_per_run}.`, + suggested_commands: [`zenrows policy set max_pages_per_run ${requested.pages}`, "zenrows policy show"], }); } if (requested.credits !== undefined && requested.credits > policy.max_credits_per_run) { @@ -120,8 +120,8 @@ export function assertWithinLimits( code: "POLICY_LIMIT_EXCEEDED", message: `This ${context} is estimated at ${requested.credits} credit(s), over the policy cap of ${policy.max_credits_per_run}.`, likely_cause: "policy.max_credits_per_run bounds the estimated credit cost of a single run.", - next_action: "Reduce the job size or per-task cost (e.g. drop premium_proxy/js_render), or raise max_credits_per_run in policy.json.", - suggested_commands: ["zenrows policy show"], + next_action: `Reduce the job size or per-task cost (e.g. drop premium_proxy/js_render), or raise the cap: \`zenrows policy set max_credits_per_run ${requested.credits}\`. The default is ${defaultPolicy().max_credits_per_run}; the estimate is an upper bound (an Adaptive Stealth Mode task counts as 25).`, + suggested_commands: [`zenrows policy set max_credits_per_run ${requested.credits}`, "zenrows policy show"], }); } } diff --git a/templates/batch-jsonl-pipeline/README.md b/templates/batch-jsonl-pipeline/README.md index cc98586..0b1b21b 100644 --- a/templates/batch-jsonl-pipeline/README.md +++ b/templates/batch-jsonl-pipeline/README.md @@ -22,10 +22,11 @@ The cloud subcommands run with any active API key and spend the same credits as Fetch. ```bash -zenrows batch create jobs.example.jsonl --wait # submit + wait for the run to finish +zenrows batch create jobs.example.jsonl --follow # submit + wait for the run to finish zenrows batch status # run status + progress stats zenrows batch results --out results.jsonl # collect results (paginated) zenrows batch retry-failed # rerun only the failed tasks +zenrows batch retry-failed --include-pending # also run tasks that did not finish zenrows batch cancel # stop an in-flight run ``` diff --git a/tests/batch-api.test.ts b/tests/batch-api.test.ts index a85c6e6..6334daf 100644 --- a/tests/batch-api.test.ts +++ b/tests/batch-api.test.ts @@ -90,13 +90,16 @@ test("listResults paginates across >=2 cursor pages until next_cursor is null", assert.deepEqual(seenCursors, [null, "c1", "c2"]); }); -test("stopJob posts /jobs/{id}/stop", async () => { +test("stopJob posts /jobs/{id}/stop and returns the run", async () => { const { impl, calls } = jsonFetch(200, { + run_id: "run_1", job_id: "job_1", - latest_run: { status: "stopped", stats: { total: 5, completed: 3, successful: 3, failed: 0 } }, + status: "stopped", + stats: { total: 5, completed: 3, successful: 3, failed: 0 }, }); - const job = await stopJob("job_1", { apiKey: "k", fetchImpl: impl }); - assert.equal(job.latest_run.status, "stopped"); + const run = await stopJob("job_1", { apiKey: "k", fetchImpl: impl }); + assert.equal(run.status, "stopped"); + assert.equal(run.stats.completed, 3); assert.equal(calls[0]!.init?.method, "POST"); assert.match(calls[0]!.url, /\/jobs\/job_1\/stop$/); }); @@ -121,7 +124,11 @@ test("problem+json 429 quota_exceeded maps to BATCH_QUOTA_EXCEEDED", async () => ); await assert.rejects( () => createJob({ type: "regular", status: "closed", tasks: [] }, { apiKey: "k", fetchImpl: impl }), - (e: unknown) => e instanceof ToolkitError && e.code === "BATCH_QUOTA_EXCEEDED", + (e: unknown) => + e instanceof ToolkitError && + e.code === "BATCH_QUOTA_EXCEEDED" && + e.server_code === "quota_exceeded" && + e.reason === "too_many_active_jobs", ); }); @@ -133,7 +140,12 @@ test("problem+json 402 (no credit available) maps to POLICY_MAX_CREDITS_EXCEEDED ); await assert.rejects( () => createJob({ type: "regular", status: "closed", tasks: [] }, { apiKey: "k", fetchImpl: impl }), - (e: unknown) => e instanceof ToolkitError && e.code === "POLICY_MAX_CREDITS_EXCEEDED", + (e: unknown) => + e instanceof ToolkitError && + e.code === "POLICY_MAX_CREDITS_EXCEEDED" && + e.server_code === "payment_required" && + e.reason === "insufficient_credits" && + /out of Zenrows credits/.test(e.next_action), ); }); @@ -145,7 +157,12 @@ test("problem+json 402 api_key_cap_reached maps to KEY_CREDIT_CAP_REACHED", asyn ); await assert.rejects( () => createJob({ type: "regular", status: "closed", tasks: [] }, { apiKey: "k", fetchImpl: impl }), - (e: unknown) => e instanceof ToolkitError && e.code === "KEY_CREDIT_CAP_REACHED" && /daily cap of 200/.test(e.likely_cause), + (e: unknown) => + e instanceof ToolkitError && + e.code === "KEY_CREDIT_CAP_REACHED" && + e.server_code === "api_key_cap_reached" && + e.reason === "api_key_cap_reached" && + /daily cap of 200/.test(e.likely_cause), ); }); diff --git a/tests/batch-command.test.ts b/tests/batch-command.test.ts index 44ea664..3e475b1 100644 --- a/tests/batch-command.test.ts +++ b/tests/batch-command.test.ts @@ -191,6 +191,8 @@ test("batch status on a run the key cap stopped exits 1, ok:false, with reason, assert.equal(j.failure_reason, "api_key_cap_reached"); assert.match(j.failure_detail, /weekly credit cap/); assert.equal(j.error.code, "KEY_CREDIT_CAP_REACHED"); + assert.equal(j.error.server_code, "api_key_cap_reached"); + assert.equal(j.error.reason, "api_key_cap_reached"); assert.match(j.error.next_action, /settings\/api-keys/); // The status call itself succeeded, so the cause must not cite an HTTP 402. assert.doesNotMatch(j.error.likely_cause, /HTTP 402/); @@ -244,6 +246,22 @@ test("batch status on completed and stopped runs exits 0 with ok:true", async () } }); +test("batch cancel reads the run that /stop returns", async () => { + const run = { run_id: "r1", job_id: "j_stop", status: "stopped", stats: { total: 5, completed: 3, successful: 3, failed: 0 } }; + await withJobResponse(run, async () => { + let code = -1; + const out = await captureOut(async () => { + code = await batch.run(["cancel", "j_stop"], ctx); + }); + assert.equal(code, 0); + const j = JSON.parse(out) as Record; + assert.equal(j.ok, true); + assert.equal(j.jobId, "j_stop"); + assert.equal(j.status, "stopped"); + assert.deepEqual(j.stats, run.stats); + }); +}); + test("batch create --wait whose run the cap stops exits 1 with ok:false", async () => { await withJobResponse(capFailedJob, async () => { const file = writeSpec(["https://ok.example/a"]); diff --git a/tests/batch-hardening.test.ts b/tests/batch-hardening.test.ts new file mode 100644 index 0000000..cbff97b --- /dev/null +++ b/tests/batch-hardening.test.ts @@ -0,0 +1,596 @@ +import { test } from "node:test"; +import assert from "node:assert/strict"; +import { writeFileSync } from "node:fs"; +import { join } from "node:path"; +import { createJob, getJob, listResults, rerunJob, stopJob, waitForJob } from "../src/core/batch-api.ts"; +import { ToolkitError } from "../src/core/errors.ts"; +import { estimateCredits, toJobBody, validateJsonl, type BatchJob } from "../src/adapters/batch.ts"; +import { assertWithinLimits, defaultPolicy, savePolicy } from "../src/core/policy.ts"; +import { batch, runFailure, timeoutFlag } from "../src/cli/commands/batch.ts"; +import { createWorkspace } from "../src/core/workspace.ts"; +import { saveApiKey } from "../src/core/auth.ts"; +import { tempRoot } from "./helpers.ts"; + +const noSleep = async () => {}; + +/** A fetch stub that answers each call from `responses` in order, recording the calls. */ +function seqFetch(responses: Array<{ status: number; body?: unknown; headers?: Record }>) { + const calls: Array<{ url: string; init?: RequestInit }> = []; + let i = 0; + const impl = (async (url: string, init?: RequestInit) => { + calls.push({ url, init }); + const r = responses[Math.min(i++, responses.length - 1)]!; + return new Response(r.body === undefined ? "" : JSON.stringify(r.body), { + status: r.status, + headers: { "content-type": "application/json", ...(r.headers ?? {}) }, + }); + }) as unknown as typeof fetch; + return { impl, calls }; +} + +const runningJob = { job_id: "j1", latest_run: { status: "running", stats: { total: 1, completed: 0, successful: 0, failed: 0 } } }; + +test("GETs retry a transient 503 and then succeed", async () => { + const { impl, calls } = seqFetch([{ status: 503, body: { code: "unavailable" } }, { status: 200, body: runningJob }]); + const job = await getJob("j1", { apiKey: "k", fetchImpl: impl, sleepImpl: noSleep }); + assert.equal(job.job_id, "j1"); + assert.equal(calls.length, 2); +}); + +test("GET pagination survives a transient 502 mid-way without losing pages", async () => { + const { impl, calls } = seqFetch([ + { status: 200, body: { results: [{ task_id: "a" }], next_cursor: "c2" } }, + { status: 502 }, + { status: 200, body: { results: [{ task_id: "b" }], next_cursor: null } }, + ]); + const rows = await listResults("j1", { apiKey: "k", fetchImpl: impl, sleepImpl: noSleep }); + assert.deepEqual(rows.map((r) => r.task_id), ["a", "b"]); + assert.equal(calls.length, 3); +}); + +test("createJob sends an Idempotency-Key and retries a 502 with the same key", async () => { + const { impl, calls } = seqFetch([{ status: 502 }, { status: 201, body: runningJob }]); + await createJob({ type: "regular", status: "closed", tasks: [] }, { apiKey: "k", fetchImpl: impl, sleepImpl: noSleep }); + assert.equal(calls.length, 2); + const k1 = (calls[0]!.init?.headers as Record)["Idempotency-Key"]; + const k2 = (calls[1]!.init?.headers as Record)["Idempotency-Key"]; + assert.ok(k1 && k1.length > 0); + assert.equal(k1, k2); +}); + +test("createJob does not retry a 429 (an account limit, not a blip)", async () => { + const { impl, calls } = seqFetch([{ status: 429, body: { code: "quota_exceeded", status: 429 } }]); + await assert.rejects( + () => createJob({ tasks: [] }, { apiKey: "k", fetchImpl: impl, sleepImpl: noSleep }), + (e: unknown) => e instanceof ToolkitError && e.code === "BATCH_QUOTA_EXCEEDED", + ); + assert.equal(calls.length, 1); +}); + +test("createJob honors a caller-supplied Idempotency-Key", async () => { + const { impl, calls } = seqFetch([{ status: 201, body: runningJob }]); + await createJob({ tasks: [] }, { apiKey: "k", fetchImpl: impl, idempotencyKey: "my-key" }); + assert.equal((calls[0]!.init?.headers as Record)["Idempotency-Key"], "my-key"); +}); + +test("rerunJob sends an Idempotency-Key and the status filter", async () => { + const { impl, calls } = seqFetch([{ status: 202, body: runningJob }]); + await rerunJob("j1", { apiKey: "k", fetchImpl: impl, status: "failed,pending" }); + assert.match(calls[0]!.url, /\/jobs\/j1\/rerun\?status=failed%2Cpending$/); + assert.ok((calls[0]!.init?.headers as Record)["Idempotency-Key"]); +}); + +test("409 no_matching_tasks points at --include-pending and keeps server_code", async () => { + const { impl } = seqFetch([ + { status: 409, body: { status: 409, code: "no_matching_tasks", detail: "No tasks in the previous run match the requested statuses." } }, + ]); + await assert.rejects( + () => rerunJob("j1", { apiKey: "k", fetchImpl: impl, status: "failed" }), + (e: unknown) => + e instanceof ToolkitError && + e.code === "BATCH_FAILED" && + e.server_code === "no_matching_tasks" && + /--include-pending/.test(e.next_action), + ); +}); + +test("409 run_not_terminal and partial_rerun_window_expired get their own guidance", async () => { + for (const [code, re] of [ + ["run_not_terminal", /still in progress/], + ["partial_rerun_window_expired", /10 days/], + ] as const) { + const { impl } = seqFetch([{ status: 409, body: { status: 409, code } }]); + await assert.rejects( + () => rerunJob("j1", { apiKey: "k", fetchImpl: impl, status: "failed" }), + (e: unknown) => e instanceof ToolkitError && e.server_code === code && re.test(e.next_action), + ); + } +}); + +test("listResults passes the processing filter through", async () => { + const { impl, calls } = seqFetch([{ status: 200, body: { results: [], next_cursor: null } }]); + await listResults("j1", { apiKey: "k", fetchImpl: impl, status: "processing" }); + assert.match(calls[0]!.url, /status=processing/); +}); + +test("waitForJob fails right away on a paused run, with the timeout's code and reason run_paused", async () => { + const paused = { job_id: "j1", latest_run: { status: "running", pause_state: "paused", stats: { total: 2, completed: 1, successful: 1, failed: 0 } } }; + const { impl, calls } = seqFetch([{ status: 200, body: paused }]); + await assert.rejects( + () => waitForJob("j1", { apiKey: "k", fetchImpl: impl, sleepImpl: noSleep, timeoutMs: 600_000 }), + (e: unknown) => e instanceof ToolkitError && e.code === "BATCH_FAILED" && e.reason === "run_paused", + ); + assert.equal(calls.length, 1); +}); + +test("waitForJob still times out with BATCH_FAILED, now with reason wait_timeout", async () => { + const { impl } = seqFetch([{ status: 200, body: runningJob }]); + await assert.rejects( + () => waitForJob("j1", { apiKey: "k", fetchImpl: impl, sleepImpl: noSleep, timeoutMs: 0 }), + (e: unknown) => e instanceof ToolkitError && e.code === "BATCH_FAILED" && e.reason === "wait_timeout" && /Timed out/.test(e.message), + ); +}); + +test("estimateCredits prices job-level flags: 2 URLs with --premium-proxy cost 20, not 2", () => { + const jobs: BatchJob[] = [{ url: "https://a.com" }, { url: "https://b.com" }]; + assert.equal(estimateCredits(jobs).credits, 2); + assert.equal(estimateCredits(jobs, { premium_proxy: true }).credits, 20); + assert.equal(estimateCredits(jobs, { premium_proxy: true, js_render: true }).credits, 50); +}); + +test("estimateCredits reads nested zenrows_params and canonical flag spellings", () => { + assert.equal(estimateCredits([{ url: "https://a.com", zenrows_params: { js_render: true } }]).credits, 5); + // The API lowercases boolean strings, so "TRUE" is on; it rejects "1", so that is never priced as on. + assert.equal(estimateCredits([{ url: "https://a.com", zenrows_params: { premium_proxy: "TRUE" } }]).credits, 10); + assert.equal(estimateCredits([{ url: "https://a.com", zenrows_params: { premium_proxy: "1" } }]).credits, 1); + assert.equal(estimateCredits([{ url: "https://a.com", js_render: "false" } as unknown as BatchJob]).credits, 1); + // A task's own params win over the job's. + assert.equal(estimateCredits([{ url: "https://a.com", js_render: false }], { js_render: true }).credits, 1); + // Adaptive Stealth Mode counts at its upper bound. + assert.equal(estimateCredits([{ url: "https://a.com" }], { mode: "auto" }).credits, 25); +}); + +test("a task flag set to null is not submitted, so the job's value applies and is priced", () => { + const jobs = [{ url: "https://a.com", js_render: null } as unknown as BatchJob]; + assert.equal(estimateCredits(jobs, { js_render: true }).credits, 5); + const body = toJobBody(jobs, { js_render: true }); + assert.equal(body.tasks[0]!.zenrows_params, undefined); + assert.equal(body.zenrows_params?.js_render, "true"); +}); + +test("a null premium_proxy on a task keeps the job's premium_proxy for the proxy_country check", () => { + const jobs = [{ url: "https://a.com", premium_proxy: null, proxy_country: "us" } as unknown as BatchJob]; + assert.equal(estimateCredits(jobs, { premium_proxy: true }).credits, 10); + const body = toJobBody(jobs, { premium_proxy: true }); + assert.deepEqual(body.tasks[0]!.zenrows_params, { proxy_country: "us" }); + assert.equal(body.zenrows_params?.premium_proxy, "true"); +}); + +test("boolean params are sent as true/false; any case is normalized, other values are rejected locally", () => { + const body = toJobBody([{ url: "https://a.com", zenrows_params: { js_render: "TRUE", autoparse: false } }], { json_response: true }); + assert.deepEqual(body.tasks[0]!.zenrows_params, { js_render: "true", autoparse: "false" }); + assert.equal(body.zenrows_params?.json_response, "true"); + for (const bad of ["1", 1, "yes", 0]) { + assert.throws( + () => toJobBody([{ url: "https://a.com", zenrows_params: { premium_proxy: bad } }]), + (e: unknown) => e instanceof ToolkitError && e.code === "INVALID_USAGE" && /premium_proxy must be true or false/.test(e.message), + String(bad), + ); + } +}); + +test("validateJsonl rejects metadata that is not string-to-string", () => { + const { root, cleanup } = tempRoot(); + try { + const file = join(root, "jobs.jsonl"); + writeFileSync( + file, + [ + JSON.stringify({ url: "https://a.com", metadata: { sku: "1" } }), + JSON.stringify({ url: "https://b.com", metadata: { sku: 1 } }), + JSON.stringify({ url: "https://c.com", metadata: ["x"] }), + ].join("\n") + "\n", + ); + const v = validateJsonl(file); + assert.equal(v.validJobs, 1); + assert.deepEqual( + v.errors.map((e) => e.line), + [2, 3], + ); + assert.match(v.errors[0]!.reason, /metadata\.sku/); + } finally { + cleanup(); + } +}); + +test("a policy cap names the exact `zenrows policy set` command", () => { + const policy = defaultPolicy(); + assert.throws( + () => assertWithinLimits({ pages: 1500 }, policy, "batch"), + (e: unknown) => + e instanceof ToolkitError && + e.code === "POLICY_LIMIT_EXCEEDED" && + e.suggested_commands.includes("zenrows policy set max_pages_per_run 1500") && + /default is 1000/.test(e.next_action), + ); + assert.throws( + () => assertWithinLimits({ credits: 12500 }, policy, "batch"), + (e: unknown) => e instanceof ToolkitError && e.suggested_commands.includes("zenrows policy set max_credits_per_run 12500"), + ); +}); + +test("a run that ran out of credits keeps BATCH_FAILED, adds reason insufficient_credits, and suggests --include-pending", () => { + const err = runFailure({ + job_id: "j9", + latest_run: { status: "failed", stats: { total: 3, completed: 1, successful: 1, failed: 0 }, failure_reason: "insufficient_credits" }, + }); + assert.ok(err); + assert.equal(err.code, "BATCH_FAILED"); + assert.equal(err.reason, "insufficient_credits"); + assert.equal(err.server_code, "insufficient_credits"); + assert.ok(err.suggested_commands.includes("zenrows batch retry-failed j9 --include-pending")); +}); + +/** Run `fn` in a workspace with a key, routing fetch through `impl`. */ +async function inWorkspace(impl: typeof fetch, fn: () => Promise): Promise { + const { root, cleanup } = tempRoot(); + const cwd = process.cwd(); + createWorkspace(root); + savePolicy(defaultPolicy(), root); + saveApiKey("0".repeat(41), root); + process.chdir(root); + const orig = globalThis.fetch; + globalThis.fetch = impl; + try { + await fn(); + } finally { + globalThis.fetch = orig; + process.chdir(cwd); + cleanup(); + } +} + +/** + * Capture the CLI's stdout (the logger writes strings) and silence stderr while + * `fn` runs. Binary writes pass through: under `node --test` they are the test + * runner's own reports, and a slow test (real retry backoff) must not swallow them. + */ +async function quiet(fn: () => unknown): Promise { + const o = process.stdout.write.bind(process.stdout); + const e = process.stderr.write.bind(process.stderr); + let out = ""; + process.stdout.write = ((s: string | Uint8Array, ...rest: unknown[]) => { + if (typeof s !== "string") return (o as (...a: unknown[]) => boolean)(s, ...rest); + out += s; + return true; + }) as typeof process.stdout.write; + process.stderr.write = (() => true) as typeof process.stderr.write; + try { + await fn(); + } finally { + process.stdout.write = o; + process.stderr.write = e; + } + return out; +} + +test("batch retry-failed --include-pending reruns failed and pending tasks", async () => { + const { impl, calls } = seqFetch([{ status: 202, body: runningJob }]); + await inWorkspace(impl, async () => { + let code = -1; + await quiet(async () => { + code = await batch.run(["retry-failed", "j1", "--include-pending"], { json: true, yes: false }); + }); + assert.equal(code, 0); + }); + assert.match(calls[0]!.url, /status=failed%2Cpending%2Cprocessing$/); +}); + +test("batch retry-failed without the flag reruns only failed tasks", async () => { + const { impl, calls } = seqFetch([{ status: 202, body: runningJob }]); + await inWorkspace(impl, async () => { + await quiet(() => batch.run(["retry-failed", "j1"], { json: true, yes: false })); + }); + assert.match(calls[0]!.url, /status=failed$/); +}); + +test("batch wait on a paused run exits 1 with BATCH_FAILED, as the timeout it would reach did", async () => { + const paused = { job_id: "j1", latest_run: { status: "running", pause_state: "paused", stats: { total: 2, completed: 1, successful: 1, failed: 0 } } }; + const { impl } = seqFetch([{ status: 200, body: paused }]); + await inWorkspace(impl, async () => { + let code = -1; + const out = await quiet(async () => { + code = await batch.run(["wait", "j1", "--timeout", "600000"], { json: true, yes: false }); + }); + assert.equal(code, 1); + const j = JSON.parse(out) as { ok: boolean; error: { code: string; reason: string } }; + assert.equal(j.ok, false); + assert.equal(j.error.code, "BATCH_FAILED"); + assert.equal(j.error.reason, "run_paused"); + }); +}); + +test("batch status shows a paused run with paused:true and exits 0", async () => { + const paused = { job_id: "j1", latest_run: { status: "running", pause_state: "paused", stats: { total: 2, completed: 1, successful: 1, failed: 0 } } }; + const { impl } = seqFetch([{ status: 200, body: paused }]); + await inWorkspace(impl, async () => { + let code = -1; + const out = await quiet(async () => { + code = await batch.run(["status", "j1"], { json: true, yes: false }); + }); + assert.equal(code, 0); + const j = JSON.parse(out) as Record; + assert.equal(j.paused, true); + assert.equal(j.status, "running"); + }); +}); + +test("batch wait --timeout 0 is read as 0 ms, as before: it times out at once with exit 1", async () => { + const { impl } = seqFetch([{ status: 200, body: runningJob }]); + await inWorkspace(impl, async () => { + let code = -1; + const out = await quiet(async () => { + code = await batch.run(["wait", "j1", "--timeout", "0"], { json: true, yes: false }); + }); + assert.equal(code, 1); + assert.equal((JSON.parse(out) as { error: { reason: string } }).error.reason, "wait_timeout"); + }); +}); + +test("batch cancel exits 0 with ok:true even when the stopped run is failed, as before", async () => { + const run = { run_id: "r1", job_id: "j1", status: "failed", failure_reason: "insufficient_credits", stats: { total: 2, completed: 1, successful: 1, failed: 0 } }; + const { impl } = seqFetch([{ status: 200, body: run }]); + await inWorkspace(impl, async () => { + let code = -1; + const out = await quiet(async () => { + code = await batch.run(["cancel", "j1"], { json: true, yes: false }); + }); + assert.equal(code, 0); + const j = JSON.parse(out) as Record; + assert.equal(j.ok, true); + assert.equal(j.jobId, "j1"); + assert.equal(j.status, "failed"); + assert.equal(j.failure_reason, "insufficient_credits"); + }); +}); + +test("batch results --status processing is accepted", async () => { + const { impl, calls } = seqFetch([{ status: 200, body: { results: [], next_cursor: null } }]); + await inWorkspace(impl, async () => { + await quiet(() => batch.run(["results", "j1", "--status", "processing"], { json: true, yes: false })); + }); + assert.match(calls[0]!.url, /status=processing/); +}); + +test("a rerun whose first attempt was lost and whose retry finds a run in progress is reported as possibly accepted", async () => { + const { impl, calls } = seqFetch([ + { status: 502 }, + { status: 409, body: { status: 409, code: "run_not_terminal", detail: "Previous run is still in flight; call /stop before /rerun." } }, + ]); + await assert.rejects( + () => rerunJob("j1", { apiKey: "k", fetchImpl: impl, sleepImpl: noSleep, status: "failed" }), + (e: unknown) => + e instanceof ToolkitError && + e.server_code === "run_not_terminal" && + /may have started the new run/.test(e.next_action) && + !/then retry/.test(e.next_action), + ); + assert.equal(calls.length, 2); + const k = (c: { init?: RequestInit }) => (c.init?.headers as Record)["Idempotency-Key"]; + assert.equal(k(calls[0]!), k(calls[1]!)); +}); + +test("a create whose first attempt was lost and is still processing never advises a new key", async () => { + const { impl } = seqFetch([{ status: 503 }, { status: 409, body: { status: 409, code: "idempotency_key_conflict" } }]); + await assert.rejects( + () => createJob({ tasks: [] }, { apiKey: "k", fetchImpl: impl, sleepImpl: noSleep }), + (e: unknown) => + e instanceof ToolkitError && + e.server_code === "idempotency_key_conflict" && + /Do not resubmit with a new key/.test(e.next_action), + ); +}); + +test("an idempotency conflict on a first attempt only suggests a new key for a deliberately new job", async () => { + const { impl } = seqFetch([{ status: 409, body: { status: 409, code: "idempotency_key_conflict" } }]); + await assert.rejects( + () => createJob({ tasks: [] }, { apiKey: "k", fetchImpl: impl, idempotencyKey: "k1" }), + (e: unknown) => + e instanceof ToolkitError && /Retrying with the same key within 24 hours of the first request cannot create a duplicate/.test(e.next_action) && /deliberately new job/.test(e.next_action), + ); +}); + +test("run_not_terminal for a run whose ingestion did not complete advises a new job, not waiting", async () => { + const { impl } = seqFetch([ + { + status: 409, + body: { + status: 409, + code: "run_not_terminal", + detail: "Previous run's task ingestion did not complete; DELETE the job and resubmit instead of rerunning a partial run.", + }, + }, + ]); + await assert.rejects( + () => rerunJob("j1", { apiKey: "k", fetchImpl: impl, status: "failed" }), + (e: unknown) => e instanceof ToolkitError && /Delete the job/.test(e.next_action) && !/batch wait/.test(e.next_action), + ); +}); + +test("timeoutFlag reads a bare number as milliseconds, as before, and accepts s and ms units", () => { + assert.equal(timeoutFlag(undefined), undefined); + assert.equal(timeoutFlag("5"), 5); + assert.equal(timeoutFlag("600000"), 600_000); + assert.equal(timeoutFlag("0"), 0); + assert.equal(timeoutFlag("-1"), -1); + assert.equal(timeoutFlag("600s"), 600_000); + assert.equal(timeoutFlag("1.5s"), 1500); + assert.equal(timeoutFlag("250ms"), 250); + for (const bad of ["0s", "0ms", `${"9".repeat(400)}s`, `${"9".repeat(400)}ms`]) { + assert.throws(() => timeoutFlag(bad), (e: unknown) => e instanceof ToolkitError && e.code === "INVALID_USAGE", bad); + } + // Anything else was ignored before (the default applied); it still is, with a warning. + const warn = process.stderr.write.bind(process.stderr); + let warned = ""; + process.stderr.write = ((t: string) => { + warned += t; + return true; + }) as typeof process.stderr.write; + try { + assert.equal(timeoutFlag("10m"), undefined); + assert.equal(timeoutFlag("abc"), undefined); + assert.equal(timeoutFlag("9".repeat(400)), undefined); + } finally { + process.stderr.write = warn; + } + assert.match(warned, /using the default of 600000 ms/); +}); + +test("a create whose retries run out keeps its Idempotency-Key on the error", async () => { + const { impl, calls } = seqFetch([{ status: 502 }]); + await assert.rejects( + () => createJob({ tasks: [] }, { apiKey: "k", fetchImpl: impl, sleepImpl: noSleep, idempotencyKey: "key-1" }), + (e: unknown) => + e instanceof ToolkitError && + e.idempotency_key === "key-1" && + /may have been accepted/.test(e.next_action) && + /Do not resubmit with a new key/.test(e.next_action) && + /key-1/.test(e.next_action), + ); + assert.equal(calls.length, 4); +}); + +test("a create that never reaches the API keeps its Idempotency-Key on the error", async () => { + const impl = (async () => { + throw new Error("connect ECONNRESET"); + }) as unknown as typeof fetch; + await assert.rejects( + () => createJob({ tasks: [] }, { apiKey: "k", fetchImpl: impl, sleepImpl: noSleep, idempotencyKey: "key-2" }), + (e: unknown) => e instanceof ToolkitError && e.code === "BACKEND_UNAVAILABLE" && e.idempotency_key === "key-2", + ); +}); + +test("a GET whose retries run out carries no Idempotency-Key", async () => { + const { impl } = seqFetch([{ status: 503 }]); + await assert.rejects( + () => getJob("j1", { apiKey: "k", fetchImpl: impl, sleepImpl: noSleep }), + (e: unknown) => e instanceof ToolkitError && e.idempotency_key === undefined, + ); +}); + +test("batch create with exhausted retries prints the exact command to repeat with the same key", async () => { + const { impl, calls } = seqFetch([{ status: 502 }]); + await inWorkspace(impl, async () => { + writeFileSync(join(process.cwd(), "jobs.jsonl"), JSON.stringify({ url: "https://a.com" }) + "\n"); + const out = await quiet(() => batch.run(["create", "jobs.jsonl", "--manual", "--no-signup"], { json: true, yes: false })); + const err = (JSON.parse(out) as { error: Record }).error; + const sent = (calls[0]!.init?.headers as Record)["Idempotency-Key"]!; + assert.ok(sent); + assert.equal(err.idempotency_key, sent); + assert.equal((err.suggested_commands as string[])[0], `zenrows batch create --idempotency-key ${sent} jobs.jsonl --manual --no-signup`); + }); + assert.ok(calls.length >= 2); +}); + +test("batch retry-failed with exhausted retries repeats with the caller's key, not a new one", async () => { + const { impl } = seqFetch([{ status: 503 }]); + await inWorkspace(impl, async () => { + const out = await quiet(() => + batch.run(["retry-failed", "j1", "--include-pending", "--idempotency-key", "mine"], { json: true, yes: false }), + ); + const err = (JSON.parse(out) as { error: Record }).error; + assert.equal((err.suggested_commands as string[])[0], "zenrows batch retry-failed --idempotency-key mine j1 --include-pending"); + }); +}); + +test("a run the policy cap allowed before still goes through when the full estimate is over the cap, with a warning", async () => { + const { impl, calls } = seqFetch([{ status: 201, body: runningJob }]); + await inWorkspace(impl, async () => { + const lines = Array.from({ length: 500 }, (_, i) => JSON.stringify({ url: `https://a.com/${i}` })).join("\n") + "\n"; + writeFileSync(join(process.cwd(), "jobs.jsonl"), lines); + let code = -1; + let warned = ""; + await quiet(async () => { + // quiet() silences stderr; record what is written there instead. + const silent = process.stderr.write; + process.stderr.write = ((t: string | Uint8Array) => { + warned += typeof t === "string" ? t : ""; + return true; + }) as typeof process.stderr.write; + try { + code = await batch.run(["create", "jobs.jsonl", "--premium-proxy", "--js-render", "--no-signup"], { json: true, yes: false }); + } finally { + process.stderr.write = silent; + } + }); + assert.equal(code, 0); + assert.match(warned, /estimated at up to 12500 credits, over the policy cap of 10000/); + }); + assert.equal(calls.length, 1); +}); + +/** Run a batch subcommand with human output, returning its stderr text. */ +async function human(impl: typeof fetch, argv: string[]): Promise<{ code: number; err: string }> { + let code = -1; + let err = ""; + await inWorkspace(impl, async () => { + await quiet(async () => { + const silent = process.stderr.write; + process.stderr.write = ((t: string | Uint8Array) => { + err += typeof t === "string" ? t : ""; + return true; + }) as typeof process.stderr.write; + try { + code = await batch.run(argv, { json: false, yes: false }); + } finally { + process.stderr.write = silent; + } + }); + }); + return { code, err }; +} + +test("human status: a paused run is a warning with the pause hint", async () => { + const paused = { job_id: "j1", latest_run: { status: "running", pause_state: "paused", stats: { total: 2, completed: 1, successful: 1, failed: 0 } } }; + const { code, err } = await human(seqFetch([{ status: 200, body: paused }]).impl, ["status", "j1"]); + assert.equal(code, 0); + assert.match(err, /status: running \(paused\)/); + assert.match(err, /no new task is picked up/); + assert.doesNotMatch(err, /✓/); +}); + +test("human status: a stopped run with unfinished tasks points at --include-pending", async () => { + const stopped = { job_id: "j1", latest_run: { status: "stopped", stats: { total: 5, completed: 3, successful: 3, failed: 0 } } }; + const { code, err } = await human(seqFetch([{ status: 200, body: stopped }]).impl, ["status", "j1"]); + assert.equal(code, 0); + assert.match(err, /2 task\(s\) did not complete\. Run them: zenrows batch retry-failed j1 --include-pending/); +}); + +test("human cancel: a failed run is shown as a warning, exit 0 as before", async () => { + const run = { run_id: "r1", job_id: "j1", status: "failed", failure_reason: "x", stats: { total: 1, completed: 1, successful: 0, failed: 1 } }; + const { code, err } = await human(seqFetch([{ status: 200, body: run }]).impl, ["cancel", "j1"]); + assert.equal(code, 0); + assert.match(err, /Stopped job j1 · status: failed/); + assert.doesNotMatch(err, /✓/); +}); + +test("batch results rejects an unknown --status before any request", async () => { + const { impl, calls } = seqFetch([{ status: 200, body: { results: [], next_cursor: null } }]); + const { code, err } = await human(impl, ["results", "j1", "--status", "nope"]); + assert.equal(code, 1); + assert.match(err, /Invalid --status: nope/); + assert.equal(calls.length, 0); +}); + +test("401 keeps AUTH_INVALID and adds server_code; 409 run_not_stoppable gets its own advice", async () => { + const a = seqFetch([{ status: 401, body: { status: 401, code: "unauthenticated", detail: "Invalid API key." } }]); + await assert.rejects( + () => getJob("j1", { apiKey: "k", fetchImpl: a.impl }), + (e: unknown) => e instanceof ToolkitError && e.code === "AUTH_INVALID" && e.server_code === "unauthenticated", + ); + const b = seqFetch([{ status: 409, body: { status: 409, code: "run_not_stoppable" } }]); + await assert.rejects( + () => stopJob("j1", { apiKey: "k", fetchImpl: b.impl }), + (e: unknown) => e instanceof ToolkitError && e.code === "BATCH_FAILED" && /nothing to stop/.test(e.next_action), + ); +}); diff --git a/tests/batch-uncertain.test.ts b/tests/batch-uncertain.test.ts new file mode 100644 index 0000000..0d2dddd --- /dev/null +++ b/tests/batch-uncertain.test.ts @@ -0,0 +1,391 @@ +/** + * Every way a Batch write (create, rerun) can end without a known result, and + * what the CLI must do then: keep the Idempotency-Key on the error, never + * suggest a new key, and print the exact command that repeats the request. + * + * A. no answer on every attempt (connection error) + * B. no answer on every attempt (client-side timeout) + * C. a 5xx on the last attempt + * D. a refusal (402, 409, 429, 400) after an attempt that got no usable answer + * E. a 2xx whose answer is empty, not JSON, or has no job_id + * F. create succeeded but `--follow` failed afterwards (job id known) + * + * A definite refusal on the first attempt, and every GET, carry no key. + * Stop sends no key: the API treats a repeated stop as a no-op. + */ +import { test } from "node:test"; +import assert from "node:assert/strict"; +import { execFileSync } from "node:child_process"; +import { readdirSync, readFileSync, writeFileSync } from "node:fs"; +import { join } from "node:path"; +import { createJob, getJob, isPaused, rerunJob, stopJob } from "../src/core/batch-api.ts"; +import { ToolkitError } from "../src/core/errors.ts"; +import { batch, repeatArgs, shellQuote } from "../src/cli/commands/batch.ts"; +import { createWorkspace } from "../src/core/workspace.ts"; +import { defaultPolicy, savePolicy } from "../src/core/policy.ts"; +import { saveApiKey } from "../src/core/auth.ts"; +import { tempRoot } from "./helpers.ts"; + +const noSleep = async () => {}; + +type Reply = { status: number; body?: unknown; raw?: string } | "throw"; + +/** A fetch stub answering each call from `replies` in order (the last repeats). */ +function seqFetch(replies: Reply[]) { + const calls: Array<{ url: string; init?: RequestInit }> = []; + let i = 0; + const impl = (async (url: string, init?: RequestInit) => { + calls.push({ url, init }); + const r = replies[Math.min(i++, replies.length - 1)]!; + if (r === "throw") throw new Error("socket hang up"); + const text = r.raw ?? (r.body === undefined ? "" : JSON.stringify(r.body)); + return new Response(text, { status: r.status, headers: { "content-type": "application/json" } }); + }) as unknown as typeof fetch; + const keys = () => calls.map((c) => (c.init?.headers as Record)["Idempotency-Key"]); + return { impl, calls, keys }; +} + +const job = { job_id: "j1", latest_run: { status: "running", stats: { total: 1, completed: 0, successful: 0, failed: 0 } } }; + +/** The error of an unknown outcome: carries the key, forbids a new key, explains a 409. */ +function isUnknownOutcome(key: string) { + return (e: unknown) => + e instanceof ToolkitError && + e.idempotency_key === key && + /Do not resubmit with a new key/.test(e.next_action) && + /within 24 hours of the first attempt cannot create a duplicate/.test(e.next_action) && + /not guaranteed to be the original/.test(e.next_action) && + /402/.test(e.next_action) && + e.next_action.includes(key); +} + +const create = (impl: typeof fetch, key: string) => + createJob({ tasks: [] }, { apiKey: "k", fetchImpl: impl, sleepImpl: noSleep, idempotencyKey: key }); +const rerun = (impl: typeof fetch, key: string) => + rerunJob("j1", { apiKey: "k", fetchImpl: impl, sleepImpl: noSleep, idempotencyKey: key, status: "failed" }); + +// A. no answer on every attempt +test("A: create and rerun with no answer on every attempt keep the key", async () => { + for (const [name, call] of [["create", create], ["rerun", rerun]] as const) { + const { impl, calls, keys } = seqFetch(["throw"]); + await assert.rejects(() => call(impl, `a-${name}`), isUnknownOutcome(`a-${name}`), name); + assert.equal(calls.length, 4, name); + assert.deepEqual(new Set(keys()), new Set([`a-${name}`]), name); + } +}); + +// B. client-side timeout on every attempt +test("B: a write that times out on every attempt keeps the key", async () => { + const impl = ((_url: string, init?: RequestInit) => + new Promise((_resolve, reject) => { + init?.signal?.addEventListener("abort", () => reject(new Error("The operation was aborted"))); + })) as unknown as typeof fetch; + await assert.rejects( + () => createJob({ tasks: [] }, { apiKey: "k", fetchImpl: impl, sleepImpl: noSleep, idempotencyKey: "b-1", timeoutMs: 5 }), + (e: unknown) => isUnknownOutcome("b-1")(e) && (e as ToolkitError).code === "BACKEND_UNAVAILABLE", + ); +}); + +// C. a 5xx on the last attempt +test("C: create and rerun whose last answer is a 5xx keep the key", async () => { + for (const [name, call] of [["create", create], ["rerun", rerun]] as const) { + const { impl, calls } = seqFetch([{ status: 502 }, { status: 503 }, { status: 504 }, { status: 500, body: { code: "internal" } }]); + await assert.rejects(() => call(impl, `c-${name}`), isUnknownOutcome(`c-${name}`), name); + assert.equal(calls.length, 4, name); + } +}); + +// D. a refusal after an attempt that got no usable answer +test("D: a 402 after a lost attempt keeps the key and the credit advice (create and rerun)", async () => { + for (const [name, call] of [["create", create], ["rerun", rerun]] as const) { + const { impl } = seqFetch([{ status: 502 }, { status: 402, body: { status: 402, code: "payment_required", detail: "No credit available." } }]); + await assert.rejects( + () => call(impl, `d-${name}`), + (e: unknown) => + isUnknownOutcome(`d-${name}`)(e) && + (e as ToolkitError).code === "POLICY_MAX_CREDITS_EXCEEDED" && + (e as ToolkitError).reason === "insufficient_credits" && + /may have been accepted/.test((e as ToolkitError).next_action), + name, + ); + } +}); + +test("D: a 402 key cap after a connection error keeps the key", async () => { + const { impl } = seqFetch(["throw", { status: 402, body: { status: 402, code: "api_key_cap_reached" } }]); + await assert.rejects( + () => create(impl, "d-cap"), + (e: unknown) => isUnknownOutcome("d-cap")(e) && (e as ToolkitError).code === "KEY_CREDIT_CAP_REACHED", + ); +}); + +test("D: a 429 or 400 after a lost attempt keeps the key", async () => { + for (const status of [429, 400]) { + const { impl } = seqFetch([{ status: 503 }, { status, body: { status, code: status === 429 ? "quota_exceeded" : "invalid_argument" } }]); + await assert.rejects(() => create(impl, `d-${status}`), isUnknownOutcome(`d-${status}`), String(status)); + } +}); + +test("D: a 409 conflict after a lost attempt keeps the key and explains it", async () => { + const { impl } = seqFetch([{ status: 503 }, { status: 409, body: { status: 409, code: "idempotency_key_conflict" } }]); + await assert.rejects( + () => create(impl, "d-409"), + (e: unknown) => + isUnknownOutcome("d-409")(e) && + /without a recorded answer/.test((e as ToolkitError).next_action) && + !/is still processing it/.test((e as ToolkitError).next_action), + ); +}); + +test("D: a rerun 409 run_not_terminal after a lost attempt keeps the key", async () => { + const { impl } = seqFetch([{ status: 502 }, { status: 409, body: { status: 409, code: "run_not_terminal", detail: "Previous run is still in flight." } }]); + await assert.rejects( + () => rerun(impl, "d-rnt"), + (e: unknown) => isUnknownOutcome("d-rnt")(e) && /may have started the new run/.test((e as ToolkitError).next_action), + ); +}); + +// E. a 2xx whose answer can't be used +test("E: a 2xx with an empty, truncated or job_id-less answer keeps the key (create and rerun)", async () => { + const bad: Reply[] = [{ status: 201, raw: "" }, { status: 201, raw: '{"job_id":"j1","lat' }, { status: 202, body: { latest_run: {} } }, { status: 201, body: { job_id: "" } }]; + for (const [name, call] of [["create", create], ["rerun", rerun]] as const) { + for (const [i, reply] of bad.entries()) { + const { impl, calls } = seqFetch([reply]); + await assert.rejects( + () => call(impl, `e-${name}-${i}`), + (e: unknown) => + isUnknownOutcome(`e-${name}-${i}`)(e) && /answer could not be read/.test((e as ToolkitError).message) && /probably exists/.test((e as ToolkitError).next_action), + `${name} #${i}`, + ); + // An accepted answer is not retried: a retry would only replay the same unreadable answer. + assert.equal(calls.length, 1, `${name} #${i}`); + } + } +}); + +test("E: a GET with an unreadable answer fails without a key", async () => { + const { impl } = seqFetch([{ status: 200, raw: "not json" }]); + await assert.rejects( + () => getJob("j1", { apiKey: "k", fetchImpl: impl }), + (e: unknown) => e instanceof ToolkitError && e.idempotency_key === undefined && /could not be read/.test(e.message), + ); +}); + +// Definite outcomes carry no key. +test("a definite refusal on the first attempt carries no key", async () => { + const { impl } = seqFetch([{ status: 402, body: { status: 402, code: "payment_required" } }]); + await assert.rejects( + () => create(impl, "none"), + (e: unknown) => e instanceof ToolkitError && e.code === "POLICY_MAX_CREDITS_EXCEEDED" && e.idempotency_key === undefined, + ); +}); + +test("a readable 2xx returns the job", async () => { + const { impl } = seqFetch([{ status: 201, body: job }]); + assert.equal((await create(impl, "ok")).job_id, "j1"); +}); + +test("stop sends no key and is not retried (a repeated stop is a no-op on the API)", async () => { + const { impl, calls, keys } = seqFetch(["throw"]); + await assert.rejects( + () => stopJob("j1", { apiKey: "k", fetchImpl: impl, sleepImpl: noSleep }), + (e: unknown) => e instanceof ToolkitError && e.code === "BACKEND_UNAVAILABLE" && e.idempotency_key === undefined, + ); + assert.equal(calls.length, 1); + assert.deepEqual(keys(), [undefined]); +}); + +// Shell quoting of the repeat command. +test("shellQuote keeps plain words and round-trips spaces and quotes through sh", () => { + assert.equal(shellQuote("jobs.jsonl"), "jobs.jsonl"); + assert.equal(shellQuote("--idempotency-key"), "--idempotency-key"); + for (const v of ["key with spaces", "it's a key", 'say "hi"', "a;b $(x) `y`", ""]) { + const out = execFileSync("sh", ["-c", `printf '%s' ${shellQuote(v)}`], { encoding: "utf8" }); + assert.equal(out, v, v); + } +}); + +/** Run `fn` in a workspace with a key, routing fetch through `impl`. */ +async function inWorkspace(impl: typeof fetch, fn: (root: string) => Promise): Promise { + const { root, cleanup } = tempRoot(); + const cwd = process.cwd(); + createWorkspace(root); + savePolicy(defaultPolicy(), root); + saveApiKey("0".repeat(41), root); + process.chdir(root); + const orig = globalThis.fetch; + globalThis.fetch = impl; + try { + await fn(root); + } finally { + globalThis.fetch = orig; + process.chdir(cwd); + cleanup(); + } +} + +/** Capture the CLI's string stdout; binary writes (the test runner's) pass through. */ +async function quiet(fn: () => unknown): Promise { + const o = process.stdout.write.bind(process.stdout); + const e = process.stderr.write.bind(process.stderr); + let out = ""; + process.stdout.write = ((s: string | Uint8Array, ...rest: unknown[]) => { + if (typeof s !== "string") return (o as (...a: unknown[]) => boolean)(s, ...rest); + out += s; + return true; + }) as typeof process.stdout.write; + process.stderr.write = (() => true) as typeof process.stderr.write; + try { + await fn(); + } finally { + process.stdout.write = o; + process.stderr.write = e; + } + return out; +} + +test("the repeat command quotes a caller key with spaces and quotes, and sh reads it back intact", async () => { + const { impl } = seqFetch([{ status: 502 }, { status: 402, body: { status: 402, code: "payment_required" } }]); + const key = `my "key" it's`; + await inWorkspace(impl, async () => { + const out = await quiet(() => batch.run(["retry-failed", "j1", "--idempotency-key", key], { json: true, yes: false })); + const err = (JSON.parse(out) as { error: Record }).error; + const cmd = (err.suggested_commands as string[])[0]!; + assert.equal(cmd, `zenrows batch retry-failed --idempotency-key ${shellQuote(key)} j1`); + const argv = execFileSync("sh", ["-c", `for a in ${cmd}; do printf '%s\\n' "$a"; done`], { encoding: "utf8" }); + const parts = argv.trim().split("\n"); + assert.equal(parts[parts.indexOf("--idempotency-key") + 1], key); + }); +}); + +test("retry-failed records its key in the run artifact, on success and on an unknown outcome", async () => { + for (const replies of [[{ status: 202, body: job }], [{ status: 502 }]] as Reply[][]) { + const { impl, keys } = seqFetch(replies); + await inWorkspace(impl, async (root) => { + await quiet(() => batch.run(["retry-failed", "j1"], { json: true, yes: false })); + const runs = join(root, ".zenrows", "runs"); + const meta = JSON.parse(readFileSync(join(runs, readdirSync(runs)[0]!, "run.json"), "utf8")) as Record; + assert.equal(meta.command, "zenrows batch retry-failed"); + assert.equal(meta.request.idempotencyKey, keys()[0]); + }); + } +}); + +// F. create succeeded, following it failed +test("F: create --follow that fails after the job exists names the job and says not to create again", async () => { + const replies: Reply[] = [{ status: 201, body: job }, { status: 503 }]; + const { impl } = seqFetch(replies); + await inWorkspace(impl, async () => { + writeFileSync(join(process.cwd(), "jobs.jsonl"), JSON.stringify({ url: "https://a.com" }) + "\n"); + const out = await quiet(() => batch.run(["create", "jobs.jsonl", "--manual", "--no-signup", "--follow"], { json: true, yes: false })); + const err = (JSON.parse(out) as { error: Record }).error; + assert.match(String(err.next_action), /Job j1 was created, so do not run `create` again/); + assert.equal((err.suggested_commands as string[])[0], "zenrows batch wait j1"); + assert.equal(err.idempotency_key, undefined); + }); +}); + +test("isPaused needs a live run: a stopped run that kept pause_state is not paused", () => { + assert.equal(isPaused({ status: "running", pause_state: "paused", stats: { total: 1, completed: 0, successful: 0, failed: 0 } }), true); + assert.equal(isPaused({ status: "pending", pause_state: "paused", stats: { total: 1, completed: 0, successful: 0, failed: 0 } }), true); + for (const status of ["stopped", "completed", "failed", "deleted"]) { + assert.equal(isPaused({ status, pause_state: "paused", stats: { total: 1, completed: 0, successful: 0, failed: 0 } }), false, status); + } +}); + +test("429 advice names how each kind of slot is freed", async () => { + const cases: Array<[string, RegExp]> = [ + ["Account already has 5 scheduled jobs; max 5 allowed. DELETE one before submitting a new one.", /until they are deleted/], + ["Account already has 3 active jobs; max 3 allowed concurrently.", /closed job while its run is in progress.*open job until it is deleted, or closed and its run has ended/], + ]; + for (const [detail, re] of cases) { + const { impl } = seqFetch([{ status: 429, body: { status: 429, code: "quota_exceeded", detail } }]); + await assert.rejects( + () => createJob({ tasks: [] }, { apiKey: "k", fetchImpl: impl }), + (e: unknown) => e instanceof ToolkitError && e.code === "BATCH_QUOTA_EXCEEDED" && re.test(e.next_action), + detail, + ); + } +}); + +// Keys the API would not see as sent. +test("blank, padded, control-character and over-long keys are rejected before any request", async () => { + // Why: HTTP drops a header value's surrounding whitespace, so " " would reach the API as no key at all. + assert.equal(new Headers({ "Idempotency-Key": " " }).get("Idempotency-Key"), ""); + assert.equal(new Headers({ "Idempotency-Key": " k " }).get("Idempotency-Key"), "k"); + assert.equal(new Headers({ "Idempotency-Key": `my "key" it's` }).get("Idempotency-Key"), `my "key" it's`); + for (const bad of ["", " ", "\t", " k", "k ", "a\nb", "x".repeat(129)]) { + for (const [name, call] of [["create", create], ["rerun", rerun]] as const) { + const { impl, calls } = seqFetch([{ status: 201, body: job }]); + await assert.rejects( + () => call(impl, bad), + (e: unknown) => e instanceof ToolkitError && e.code === "INVALID_USAGE" && /Invalid --idempotency-key/.test(e.message), + `${name} ${JSON.stringify(bad)}`, + ); + assert.equal(calls.length, 0, `${name} ${JSON.stringify(bad)}`); + } + } + const { impl } = seqFetch([{ status: 201, body: job }]); + assert.equal((await create(impl, "x".repeat(128))).job_id, "j1"); +}); + +test("batch create and retry-failed reject a whitespace-only --idempotency-key before any request", async () => { + for (const argv of [["create", "jobs.jsonl", "--manual", "--no-signup", "--idempotency-key", " "], ["retry-failed", "j1", "--idempotency-key", " "]]) { + const { impl, calls } = seqFetch([{ status: 201, body: job }]); + await inWorkspace(impl, async () => { + writeFileSync(join(process.cwd(), "jobs.jsonl"), JSON.stringify({ url: "https://a.com" }) + "\n"); + let code = -1; + const out = await quiet(async () => { + code = await batch.run(argv, { json: true, yes: false }); + }); + assert.equal(code, 1, argv[0]); + assert.equal((JSON.parse(out) as { error: { code: string } }).error.code, "INVALID_USAGE", argv[0]); + }); + assert.equal(calls.length, 0, argv[0]); + } +}); + +// The repeat command carries exactly the key that was sent. +test("repeatArgs puts the sent key ahead of a -- terminator and drops the caller's own key option", () => { + assert.deepEqual(repeatArgs("create", ["--manual", "--", "jobs.jsonl"], "K"), ["create", "--idempotency-key", "K", "--manual", "--", "jobs.jsonl"]); + assert.deepEqual(repeatArgs("create", ["jobs.jsonl", "--idempotency-key", "K"], "K"), ["create", "--idempotency-key", "K", "jobs.jsonl"]); + assert.deepEqual(repeatArgs("create", ["--idempotency-key=K", "jobs.jsonl"], "K"), ["create", "--idempotency-key", "K", "jobs.jsonl"]); + // After --, "--idempotency-key" is a positional (a file name), so it stays. + assert.deepEqual(repeatArgs("create", ["--", "--idempotency-key"], "K"), ["create", "--idempotency-key", "K", "--", "--idempotency-key"]); +}); + +/** Split a printed command the way sh does, dropping the leading "zenrows". */ +function shArgv(cmd: string): string[] { + const out = execFileSync("sh", ["-c", `for a in ${cmd}; do printf '%s\\0' "$a"; done`], { encoding: "utf8" }); + return out.split("\0").slice(0, -1).slice(1); +} + +test("running the printed repeat command sends exactly the key of the lost request, also after a -- terminator", async () => { + for (const [argv, label] of [ + [["create", "--manual", "--no-signup", "--", "jobs.jsonl"], "terminator, generated key"], + [["create", "jobs.jsonl", "--manual", "--no-signup", "--idempotency-key", `a "b" c's`], "caller key with quotes"], + ] as const) { + // First invocation: every attempt answered 502. Second (the repeat): accepted. + const first = seqFetch([{ status: 502 }]); + let printed = ""; + await inWorkspace(first.impl, async () => { + writeFileSync(join(process.cwd(), "jobs.jsonl"), JSON.stringify({ url: "https://a.com" }) + "\n"); + const out = await quiet(() => batch.run([...argv], { json: true, yes: false })); + printed = ((JSON.parse(out) as { error: { suggested_commands: string[] } }).error.suggested_commands)[0]!; + }); + const sent = first.keys()[0]!; + assert.ok(sent, label); + const replay = seqFetch([{ status: 201, body: job }]); + await inWorkspace(replay.impl, async () => { + writeFileSync(join(process.cwd(), "jobs.jsonl"), JSON.stringify({ url: "https://a.com" }) + "\n"); + const args = shArgv(printed); + assert.equal(args[0], "batch", label); + let code = -1; + await quiet(async () => { + code = await batch.run(args.slice(1), { json: true, yes: false }); + }); + assert.equal(code, 0, label); + }); + assert.deepEqual(replay.keys(), [sent], label); + } +});