diff --git a/src/__tests__/commands/snapshot-readiness.test.ts b/src/__tests__/commands/snapshot-readiness.test.ts new file mode 100644 index 0000000..d85737c --- /dev/null +++ b/src/__tests__/commands/snapshot-readiness.test.ts @@ -0,0 +1,73 @@ +import {describe, it, expect} from 'vitest'; +import {snapshot_running_status} from '../../commands/dataset'; +import type {Response_envelope} from '../../utils/client'; + +// The snapshot endpoint answers in two dimensions that vary independently: +// the HTTP status (200 data vs 202 still-building) and the body shape (a +// parsed object under --format json, raw text under csv/ndjson/jsonl). Every +// combination has to resolve correctly, because the failure that matters is +// silent: a not-ready response mistaken for data prints a status stub and +// exits 0, which downstream ETL then consumes as if it were the dataset. +const envelope = (status: number, body: unknown): Response_envelope=>({ + status, + headers: new Headers(), + body, +}); + +describe('commands/dataset.snapshot_running_status', ()=>{ + it('200 + object data is ready', ()=>{ + expect(snapshot_running_status(envelope(200, [{a: 1}]))) + .toBeUndefined(); + }); + + it('200 + text data is ready (csv/jsonl formats)', ()=>{ + expect(snapshot_running_status(envelope(200, 'a,b\n1,2\n'))) + .toBeUndefined(); + }); + + it('200 + object status body is still running', ()=>{ + expect(snapshot_running_status(envelope(200, {status: 'running'}))) + .toBe('running'); + }); + + it('200 + TEXT status body is still running', ()=>{ + // The regression this predicate exists for: under a non-json format + // the client hands back a string, so an object-only check misses it + // and the status stub gets printed as data. + expect(snapshot_running_status( + envelope(200, '{"status":"running"}'))).toBe('running'); + }); + + it('202 is still running even when the body looks like data', ()=>{ + // The protocol is the most authoritative signal available. + expect(snapshot_running_status(envelope(202, [{a: 1}]))) + .toBe('building'); + }); + + it('202 + text status body reports the real status', ()=>{ + expect(snapshot_running_status( + envelope(202, '{"status":"starting"}'))).toBe('starting'); + }); + + it('keeps the real status string rather than a flattened literal', ()=>{ + // Progress output prints this value, so collapsing every running + // state to one token would lose starting -> building -> running. + for (const s of ['starting', 'building', 'running']) + { + expect(snapshot_running_status(envelope(200, {status: s}))) + .toBe(s); + } + }); + + it('treats terminal statuses as ready, not running', ()=>{ + expect(snapshot_running_status(envelope(200, {status: 'ready'}))) + .toBeUndefined(); + expect(snapshot_running_status(envelope(200, {status: 'failed'}))) + .toBeUndefined(); + }); + + it('treats unparseable text as data', ()=>{ + expect(snapshot_running_status(envelope(200, 'not json at all'))) + .toBeUndefined(); + }); +}); diff --git a/src/__tests__/utils/client.request.test.ts b/src/__tests__/utils/client.request.test.ts new file mode 100644 index 0000000..64103f3 --- /dev/null +++ b/src/__tests__/utils/client.request.test.ts @@ -0,0 +1,185 @@ +import {describe, it, expect, beforeEach, afterEach, vi} from 'vitest'; + +// load_config() reads a real file from the user's config dir and can override +// api_url, which would make URL assertions depend on the machine running the +// suite. Pin it. +vi.mock('../../utils/config', ()=>({ + load: ()=>({api_url: 'https://api.brightdata.com'}), +})); + +vi.mock('../../utils/output', ()=>({ + dim: (s: string)=>s, +})); + +import { + request, + get_with_status, + Client_api_error, +} from '../../utils/client'; + +const json_response = (status: number, body: unknown)=>new Response( + JSON.stringify(body), + {status, headers: {'content-type': 'application/json'}} +); + +const text_response = (status: number, body: string)=>new Response( + body, + {status, headers: {'content-type': 'text/plain'}} +); + +// Fast retries — real backoff starts at 500ms and doubles, which would add +// seconds of wall-clock to the suite. +const fast_retry = {retry: {base_ms: 1, max_ms: 2}}; + +describe('utils/client.request', ()=>{ + beforeEach(()=>{ + vi.spyOn(console, 'error').mockImplementation(()=>{}); + }); + + afterEach(()=>{ + vi.unstubAllGlobals(); + vi.restoreAllMocks(); + }); + + it('returns the parsed body only (unchanged contract)', async()=>{ + vi.stubGlobal('fetch', vi.fn(async()=>json_response(200, {ok: 1}))); + await expect(request('key', '/x')).resolves.toEqual({ok: 1}); + }); + + it('returns text when the response is not json', async()=>{ + vi.stubGlobal('fetch', vi.fn(async()=>text_response(200, 'a,b\n'))); + await expect(request('key', '/x')).resolves.toBe('a,b\n'); + }); + + it('sends bearer auth to the resolved url', async()=>{ + const fetch_mock = vi.fn(async()=>json_response(200, {})); + vi.stubGlobal('fetch', fetch_mock); + await request('secret', '/datasets/v3/snapshot/s1'); + const [url, init] = fetch_mock.mock.calls[0] as unknown as + [string, RequestInit]; + expect(url).toBe( + 'https://api.brightdata.com/datasets/v3/snapshot/s1'); + expect((init.headers as Record)['Authorization']) + .toBe('Bearer secret'); + }); + + describe('get_with_status', ()=>{ + it('exposes the status code alongside the body', async()=>{ + vi.stubGlobal('fetch', + vi.fn(async()=>json_response(200, {rows: 1}))); + const env = await get_with_status('key', '/x'); + expect(env.status).toBe(200); + expect(env.body).toEqual({rows: 1}); + }); + + it('surfaces 202 rather than hiding it behind the body', async()=>{ + // 202 is res.ok, so before this the caller could not tell an + // accepted-but-unfinished job from finished data. + vi.stubGlobal('fetch', + vi.fn(async()=>json_response(202, {status: 'running'}))); + const env = await get_with_status('key', '/x'); + expect(env.status).toBe(202); + }); + + it('exposes response headers', async()=>{ + vi.stubGlobal('fetch', + vi.fn(async()=>json_response(200, {}))); + const env = await get_with_status('key', '/x'); + expect(env.headers.get('content-type')) + .toContain('application/json'); + }); + }); + + describe('error typing', ()=>{ + it('throws a Client_api_error carrying the status', async()=>{ + vi.stubGlobal('fetch', + vi.fn(async()=>text_response(404, 'no such dataset'))); + await expect(request('key', '/x', fast_retry)) + .rejects.toBeInstanceOf(Client_api_error); + try { + await request('key', '/x', fast_retry); + } catch(e) { + const err = e as Client_api_error; + expect(err.status).toBe(404); + // message bytes are part of the contract: scraper-studio + // matches on error prose (e.g. 'realtime job limit') + expect(err.message).toBe( + 'Error: no such dataset\n' + +' Status: 404\n' + +' Hint: Resource not found. Check the URL or dataset ' + +'type.' + ); + } + }); + + it('does not retry an API error', async()=>{ + const fetch_mock = vi.fn(async()=>text_response(400, 'bad input')); + vi.stubGlobal('fetch', fetch_mock); + await expect(request('key', '/x', fast_retry)).rejects.toThrow(); + expect(fetch_mock).toHaveBeenCalledTimes(1); + }); + + it('retries a network error whose message starts with "Error:"', + async()=>{ + // Retry used to be decided by message.startsWith('Error:'), + // so a network failure worded this way was misclassified as a + // final API error and never retried. + const fetch_mock = vi.fn(async()=>{ + throw new Error('Error: socket hang up'); + }); + vi.stubGlobal('fetch', fetch_mock); + await expect(request('key', '/x', fast_retry)) + .rejects.toThrow('Network request failed'); + expect(fetch_mock).toHaveBeenCalledTimes(4); + }); + + it('retries transient statuses then surfaces the error', async()=>{ + const fetch_mock = vi.fn(async()=>text_response(503, 'busy')); + vi.stubGlobal('fetch', fetch_mock); + await expect(request('key', '/x', fast_retry)).rejects.toThrow(); + expect(fetch_mock).toHaveBeenCalledTimes(4); + }); + }); + + describe('request timeout', ()=>{ + // A hung connection produces no error at all, so without an abort the + // retry loop never engages and the CLI waits forever. + const hang_until_aborted = (_url: string, init: RequestInit)=> + new Promise((_resolve, reject)=>{ + init.signal?.addEventListener('abort', ()=>{ + const err = new Error('aborted'); + err.name = 'TimeoutError'; + reject(err); + }); + }); + + it('aborts a hung request instead of hanging forever', async()=>{ + vi.stubGlobal('fetch', vi.fn(hang_until_aborted)); + await expect(request('key', '/x', { + timeout_ms: 20, + ...fast_retry, + })).rejects.toThrow('Request timed out after 0s'); + }); + + it('retries a timeout once, not the full retry budget', async()=>{ + // timeout x attempts multiplies the user-visible stall, so the + // generic budget (3 retries) is deliberately not reused here. + const fetch_mock = vi.fn(hang_until_aborted); + vi.stubGlobal('fetch', fetch_mock); + await expect(request('key', '/x', { + timeout_ms: 20, + ...fast_retry, + })).rejects.toThrow(); + expect(fetch_mock).toHaveBeenCalledTimes(2); + }); + + it('passes an abort signal on every attempt', async()=>{ + const fetch_mock = vi.fn(async()=>json_response(200, {})); + vi.stubGlobal('fetch', fetch_mock); + await request('key', '/x'); + const [, init] = fetch_mock.mock.calls[0] as unknown as + [string, RequestInit]; + expect(init.signal).toBeInstanceOf(AbortSignal); + }); + }); +}); diff --git a/src/commands/dataset.ts b/src/commands/dataset.ts index 667db33..0137bec 100644 --- a/src/commands/dataset.ts +++ b/src/commands/dataset.ts @@ -1,6 +1,7 @@ import {Command} from 'commander'; import {ensure_authenticated} from '../utils/auth'; -import {get, post} from '../utils/client'; +import {get_with_status, post} from '../utils/client'; +import type {Response_envelope} from '../utils/client'; import {print, dim, fail} from '../utils/output'; import {start as start_spinner} from '../utils/spinner'; import {parse_timeout, poll_until} from '../utils/polling'; @@ -214,6 +215,36 @@ const extract_status = (result: unknown): string|undefined=>{ return undefined; }; +// A snapshot body only parses to an object when the requested format is json. +// Under csv/ndjson/jsonl the client hands back text, so a status payload would +// arrive as a string and extract_status would miss it. +const parse_if_json_text = (body: unknown): unknown=>{ + if (typeof body != 'string') + return body; + try { + return JSON.parse(body); + } catch(_e) { + return body; + } +}; + +// Is this snapshot still building? Three signals, most authoritative first: +// 1. HTTP 202 — the server says "accepted, not done". Trusted outright. +// 2. a parsed body carrying a running status. +// 3. a *text* body that parses to one (the csv/ndjson/jsonl case above). +// Returns the running status string (so progress output keeps showing +// starting/building/running rather than a flattened literal), or undefined +// when the response is the data. +const snapshot_running_status = ( + env: Response_envelope +): string|undefined=>{ + const status = extract_status(parse_if_json_text(env.body)); + const is_running = !!status && RUNNING_STATUSES.includes(status); + if (env.status == 202) + return is_running ? status : 'building'; + return is_running ? status : undefined; +}; + const handle_pipelines = async( dataset_type_raw: string, params: string[], @@ -266,14 +297,15 @@ const handle_pipelines = async( } console.error(dim(`Triggered collection with snapshot ID:` + `${snapshot_id}`)); - const poll_result = await poll_until({ + const poll_result = await poll_until({ timeout_seconds: timeout, fetch_once: ()=>{ const endpoint = `${SNAPSHOT_ENDPOINT}/${snapshot_id}` +`?format=${format}`; - return get(api_key, endpoint, {timing: opts.timing}); + return get_with_status( + api_key, endpoint, {timing: opts.timing}); }, - get_status: extract_status, + get_status: snapshot_running_status, running_statuses: RUNNING_STATUSES, timeout_label: 'data', on_running: ({attempt, timeout_seconds, status})=>{ @@ -286,7 +318,7 @@ const handle_pipelines = async( console.error(dim( `Data received after ${poll_result.attempts} attempts` )); - const result = poll_result.result; + const result = poll_result.result.body; const cleaned_result = format == 'json' ? strip_nulls(result) : result; print(cleaned_result, { json: opts.json, @@ -333,4 +365,4 @@ add_examples(pipelines_command, [ }, ]); -export {pipelines_command, handle_pipelines}; +export {pipelines_command, handle_pipelines, snapshot_running_status}; diff --git a/src/utils/client.ts b/src/utils/client.ts index f4dd286..ccf10d0 100644 --- a/src/utils/client.ts +++ b/src/utils/client.ts @@ -1,9 +1,20 @@ import {load as load_config} from './config'; +import {dim} from './output'; const TRANSIENT_STATUSES = [429, 500, 502, 503, 504]; const MAX_RETRIES = 3; const RETRY_BASE_MS = 500; const RETRY_MAX_MS_DEFAULT = 16_000; +// Bounds a black-holed connection (VPN drop, hung load balancer), which would +// otherwise hang forever: fetch() has no default timeout, and no error ever +// arrives for the retry loop to react to. Deliberately generous — it must not +// cut off slow-but-alive work (protected-site scrapes, large snapshot bodies). +// Per-request override via Request_opts.timeout_ms. +const REQUEST_TIMEOUT_MS = 120_000; +// A timed-out attempt is retried at most this many times. Retries multiply the +// user-visible wait (timeout x attempts), so the generic retry budget is not +// reused here: 3 retries would mean an 8-minute silent stall. +const TIMEOUT_MAX_RETRIES = 1; const ERROR_HINTS: Record = { 401: 'Invalid or expired API key. Run \'brightdata login\' to re-authenticate.', @@ -54,6 +65,7 @@ type Request_opts = { raw_buffer?: boolean; hints?: Body_hint[]; retry?: Retry_config; + timeout_ms?: number; }; type Api_error = { @@ -62,6 +74,35 @@ type Api_error = { hint?: string; }; +// What the server actually said. request() returns only the body (the shape +// almost every call site wants); callers that must reason about the protocol +// itself — e.g. "is this snapshot still building?" (HTTP 202) vs "this is the +// data" (200) — use get_with_status rather than guess from body shape. +type Response_envelope = { + status: number; + headers: Headers; + body: T; +}; + +// Errors the client itself formatted from an API response, as opposed to +// network-level failures. The retry loop needs to tell those apart: API errors +// are final, network errors are retryable. This used to be decided by testing +// whether the message started with 'Error:', which coupled retry semantics to +// message wording — rewording a template silently changed behavior. +class Client_api_error extends Error { + status: number; + hint?: string; + constructor(message: string, status: number, hint?: string){ + super(message); + this.name = 'Client_api_error'; + this.status = status; + this.hint = hint; + } +} + +const is_timeout_error = (e: unknown): boolean=> + e instanceof Error && (e.name == 'TimeoutError' || e.name == 'AbortError'); + const sleep = (ms: number)=>new Promise(resolve=>setTimeout(resolve, ms)); const format_error = ( @@ -84,11 +125,11 @@ const compute_backoff = ( return Math.floor(exp / 2 + jitter); }; -const request = async( +const request_core = async( api_key: string, endpoint: string, opts: Request_opts = {} -): Promise=>{ +): Promise>=>{ const config = load_config(); const base_url = config.api_url ?? 'https://api.brightdata.com'; const url = endpoint.startsWith('http') ? endpoint : @@ -108,12 +149,19 @@ const request = async( const max_attempts = opts.retry?.max_attempts ?? MAX_RETRIES; const base_ms = opts.retry?.base_ms ?? RETRY_BASE_MS; const max_ms = opts.retry?.max_ms ?? RETRY_MAX_MS_DEFAULT; + const timeout_ms = opts.timeout_ms ?? REQUEST_TIMEOUT_MS; let attempt = 0; + let timeout_retries = 0; let start = opts.timing ? Date.now() : 0; while (attempt <= max_attempts) { try { - const res = await fetch(url, fetch_opts); + // A fresh signal per attempt: AbortSignal.timeout starts counting + // when created, so hoisting it would abort retries instantly. + const res = await fetch(url, { + ...fetch_opts, + signal: AbortSignal.timeout(timeout_ms), + }); if (opts.timing) { console.error(`Timing: ${Date.now()-start}ms @@ -122,16 +170,19 @@ const request = async( const brd_error = res.headers.get('x-brd-error') || res.headers.get('x-luminati-error'); if (brd_error) - throw new Error(`Error: ${brd_error}`); + throw new Client_api_error(`Error: ${brd_error}`, res.status); if (res.ok) { + const envelope = {status: res.status, headers: res.headers}; if (opts.raw_buffer) - return Buffer.from( - await res.arrayBuffer()) as unknown as T; + { + const buffer = Buffer.from(await res.arrayBuffer()); + return {...envelope, body: buffer as unknown as T}; + } const content_type = res.headers.get('content-type') ?? ''; if (content_type.includes('application/json')) - return await res.json() as T; - return await res.text() as unknown as T; + return {...envelope, body: await res.json() as T}; + return {...envelope, body: await res.text() as unknown as T}; } if (TRANSIENT_STATUSES.includes(res.status) && attempt < max_attempts) @@ -160,12 +211,31 @@ const request = async( ]; if (api_err.hint) msg.push(` Hint: ${api_err.hint}`); - throw new Error(msg.join('\n')); + throw new Client_api_error( + msg.join('\n'), api_err.status, api_err.hint); } catch(e) { - if (e instanceof Error && e.message.startsWith('Error:')) + if (e instanceof Client_api_error) throw e; + const timed_out = is_timeout_error(e); + if (timed_out && timeout_retries >= TIMEOUT_MAX_RETRIES) + { + throw new Error( + `Error: Request timed out after ` + +`${Math.round(timeout_ms/1000)}s.\n` + +' The server accepted the connection but sent no ' + +'response. Check your connection and try again.' + ); + } if (attempt < max_attempts) { + if (timed_out) + { + timeout_retries++; + console.error(dim( + `Request timed out after ` + +`${Math.round(timeout_ms/1000)}s — retrying...` + )); + } const delay = compute_backoff(attempt, base_ms, max_ms); opts.retry?.on_retry?.({ attempt: attempt + 1, @@ -186,6 +256,12 @@ const request = async( throw new Error('Error: Max retries exceeded.'); }; +const request = async( + api_key: string, + endpoint: string, + opts: Request_opts = {} +): Promise=>(await request_core(api_key, endpoint, opts)).body; + const post = ( api_key: string, endpoint: string, @@ -199,6 +275,17 @@ const get = ( opts: Omit = {} ): Promise=>request(api_key, endpoint, {method: 'GET', ...opts}); -export {request, post, get, pick_hint, ERROR_HINTS, compute_backoff, - RETRY_BASE_MS, RETRY_MAX_MS_DEFAULT, MAX_RETRIES}; -export type {Request_opts, Api_error, Body_hint, Retry_config, Retry_event}; +// Opt-in protocol-aware GET. Same request path as get(), but hands back the +// status code and headers alongside the body. +const get_with_status = ( + api_key: string, + endpoint: string, + opts: Omit = {} +): Promise>=> + request_core(api_key, endpoint, {method: 'GET', ...opts}); + +export {request, post, get, get_with_status, Client_api_error, pick_hint, + ERROR_HINTS, compute_backoff, RETRY_BASE_MS, RETRY_MAX_MS_DEFAULT, + MAX_RETRIES, REQUEST_TIMEOUT_MS, TIMEOUT_MAX_RETRIES}; +export type {Request_opts, Api_error, Body_hint, Retry_config, Retry_event, + Response_envelope};