From eadba906b531007054feaa6d3345a0fe1a777f1e Mon Sep 17 00:00:00 2001 From: Anthony Ettinger Date: Sun, 8 Nov 2026 17:03:33 +0000 Subject: [PATCH 1/2] crawler: 10 s fetch, 30 s render, slow or broken targets parked 30 days (0.49.0) Fleet crawler rule (2026-11-08): "crawlers should timeout after 10 seconds for fetch, 30 for browser. Flag to recheck later in a month." - politeGet: one 10 s budget (CRAWL_FETCH_TIMEOUT_MS) for the name lookup, every redirect hop and the body; was 20 s per hop with no body cutoff. - Obscura render: waitUntil load (not networkidle0), a 1.5 s bounded settle, 30 s total (CRAWL_RENDER_TIMEOUT_MS); out of budget rejects RenderTimeout. Obscura's own trouble is ObscuraError and never parks a target. - src/failures.ts: timeout/DNS/TLS/connection/5xx/429 classification. robots.txt failing that way parks the host 30 days (it used to read a timeout as "no rules" and time out every page 3x). Page failures are not thrown to BullMQ: DNS/TLS/connection/429 park the host, a timeout or 5xx parks the page, 5 within an hour park the host. 429 honours Retry-After, capped. - sites gains status, last_error, failed_at, retry_after (added once, in the site store's schema list); takeDue skips retry_after in the future, so a flagged host gets exactly one recheck after it; success clears the flag. - Redis pitcrawl:down:, pitcrawl:pdown:, pitcrawl:fails:, read in the same pipeline as the cached robots.txt (no extra round trip). Co-Authored-By: Claude Opus 5.5 --- package.json | 2 +- services/crawler/src/config.ts | 16 ++ services/crawler/src/crawler.ts | 116 ++++++++-- services/crawler/src/failures.ts | 79 +++++++ services/crawler/src/fetcher.ts | 44 +++- services/crawler/src/frontier.ts | 73 +++++++ services/crawler/src/obscura.ts | 86 ++++++-- services/crawler/src/site-store.ts | 68 +++++- services/crawler/test/failures.test.ts | 287 +++++++++++++++++++++++++ services/crawler/test/redis.test.ts | 50 +++++ 10 files changed, 786 insertions(+), 35 deletions(-) create mode 100644 services/crawler/src/failures.ts create mode 100644 services/crawler/test/failures.test.ts diff --git a/package.json b/package.json index 52be3f5..b072383 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "pitworld", - "version": "0.48.0", + "version": "0.49.0", "mcpName": "world.pit/pit", "description": "pit.world MCP server: private web search and Moshpit name lookups for AI agents (stdio, no API key). Also a web app, API, CLI and TUI.", "license": "MIT", diff --git a/services/crawler/src/config.ts b/services/crawler/src/config.ts index 10c8d4b..6f18bab 100644 --- a/services/crawler/src/config.ts +++ b/services/crawler/src/config.ts @@ -12,6 +12,22 @@ export const VERSION = '0.1.0' export const BOT = 'pitbot' export const USER_AGENT = `Mozilla/5.0 (compatible; ${BOT}/${VERSION}; +https://pit.world/bot)` +/** + * The fleet crawler rule (Anthony, 2026-11-08): "crawlers should timeout after 10 seconds for + * fetch, 30 for browser ... We don't want broken slow shit. Flag to recheck later in a month." + * + * One plain GET of a third-party address (page, robots.txt, sitemap, card read): 10 s for all + * of it, DNS, every redirect hop and the body. Calls to our own services (nichedb, Obscura's + * MCP transport, rssamplifier OPML) keep their own budgets. + */ +export const CRAWL_FETCH_TIMEOUT_MS = 10_000 +/** One browser render through Obscura: navigation to `load`, a short settle, the read back. */ +export const CRAWL_RENDER_TIMEOUT_MS = 30_000 +/** The bounded settle after `load`, for scripts that paint just after it. Never networkidle. */ +export const CRAWL_RENDER_SETTLE_MS = 1_500 +/** A host or page that timed out, would not connect or answered 5xx is left alone this long, then read once more. */ +export const CRAWL_RECHECK_MS = 30 * 86_400_000 + export type Role = 'site' | 'page' | 'render' | 'publish' export type Config = { diff --git a/services/crawler/src/crawler.ts b/services/crawler/src/crawler.ts index c7b11df..160a1d4 100644 --- a/services/crawler/src/crawler.ts +++ b/services/crawler/src/crawler.ts @@ -4,7 +4,8 @@ * The BullMQ wiring (queues.ts) calls these; the tests call them directly * with an in-memory frontier and a fake web. */ -import type { Config } from './config.ts' +import { CRAWL_RECHECK_MS, type Config } from './config.ts' +import { failureKind, HOST_KINDS, HOST_PAGE_FAILURE_WINDOW_MS, HOST_PAGE_FAILURES, parkFor, statusKind, type FailureKind } from './failures.ts' import { extract, needsRender, type Extracted } from './extract.ts' import { politeGet, readSitemapXml } from './fetcher.ts' import { endpointsIn, handshake, mcpRecord, type McpJob, WELL_KNOWN } from './mcp.ts' @@ -135,17 +136,64 @@ export function createCrawler(deps: Deps) { } } - async function robotsFor(site: string): Promise { - const kept = await frontier.getRobots(site) + const errorText = (error: unknown) => (error instanceof Error ? error.message : String(error)).slice(0, 200) + + /** + * Park a host: nothing reads from it until `ms` from now (Redis, which every worker asks + * before a fetch), and the site row carries the flag, so the recrawl skips it until then + * and gives it exactly one recheck after (Postgres). + */ + async function flagHost(site: string, kind: FailureKind, detail: string, ms = CRAWL_RECHECK_MS): Promise { + await frontier.markDown(site, ms) + await siteStore.flag(site, `${kind}: ${detail}`, Date.now() + ms).catch((error) => log(`flag ${site}: ${errorText(error)}`)) + await frontier.bump('sites_flagged') + await frontier.log({ kind: 'fail', site, note: `parked ${ms >= 86_400_000 ? `${Math.round(ms / 86_400_000)} d` : `${Math.round(ms / 60_000)} min`}: ${kind}, ${detail}`.slice(0, 200) }) + } + + /** + * Park one page for 30 days (a timeout or a 5xx may be that page alone). Enough of them + * from one site inside the hour and the host is parked too: a dead host is not read page + * by page for a thousand pages. + */ + async function flagPage(job: PageJob, kind: FailureKind, detail: string, retryAfter?: string): Promise { + if (HOST_KINDS.has(kind)) { + await flagHost(job.site, kind, `${detail} (${job.url})`, parkFor(kind, retryAfter)) + return `host parked: ${kind}` + } + await frontier.markPageDown(job.url, CRAWL_RECHECK_MS) + await frontier.bump('pages_flagged') + await frontier.log({ kind: 'fail', site: job.site, url: job.url, note: `parked 30 d: ${kind}, ${detail}`.slice(0, 200) }) + const n = await frontier.noteFailure(job.site, HOST_PAGE_FAILURE_WINDOW_MS) + if (n === HOST_PAGE_FAILURES) await flagHost(job.site, kind, `${n} pages failed within the hour, last ${detail}`) + return `page parked: ${kind}` + } + + /** `state`: a siteState() the caller already has, saving the round trip. */ + async function robotsFor(site: string, given?: { robots: Robots | null; downMs: number }): Promise { + const state = given ?? (await frontier.siteState(site)) + if (state.downMs > 0) return 'unavailable' + const kept = state.robots // Cached before AI-crawler rules were kept: read again, so a training opt-out is never missed. if (kept && kept.ai) return kept let robots: Robots = EMPTY try { const r = await get(`https://${site}/robots.txt`, site, 'text/plain,*/*;q=0.5') - // 5xx: the site cannot say, so it is read another day (Google's reading). 4xx: no rules. - if (r.status >= 500) return 'unavailable' + // 5xx: the site cannot say (Google's reading), so it is parked and asked again in 30 days. + // 429: parked for its Retry-After (capped). Any other 4xx: no rules. + const failed = statusKind(r.status) + if (failed) { + await flagHost(site, failed, `robots.txt answered HTTP ${r.status}`, parkFor(failed, r.retryAfter)) + return 'unavailable' + } if (r.status < 400 && !/html/.test(r.contentType)) robots = parseRobots(r.body) - } catch { + } catch (error) { + // A host that times out or will not connect on robots.txt would do the same for every + // page: it was read as "no rules" and each page then timed out three times. + const kind = failureKind(error) + if (kind) { + await flagHost(site, kind, `robots.txt: ${errorText(error)}`) + return 'unavailable' + } robots = { ...EMPTY, read: 'error' } } // TDMRep's site-wide file: rarely there, read with robots.txt once a day. @@ -334,16 +382,19 @@ export function createCrawler(deps: Deps) { // Once per cycle per site, however many jobs for it are queued. if (!(await frontier.claimStart(site, config.recrawlDays * 86_400_000))) return { admitted: 0, outcome: 'already started this cycle' } if (!(await resolver(site))) { + // No public address (often no address at all): parked and asked again in 30 days. await frontier.bump('sites_refused') - await frontier.log({ kind: 'skip', site, note: 'not on the public internet' }) + await flagHost(site, 'dns', 'no public address') return { admitted: 0, outcome: 'not public' } } const robots = await robotsFor(site) if (robots === 'unavailable') { await frontier.bump('sites_unavailable') - await frontier.log({ kind: 'skip', site, note: 'robots.txt answered 5xx; trying another day' }) return { admitted: 0, outcome: 'robots.txt unavailable' } } + // Read fine: a flag from an earlier cycle (this was its recheck) is cleared. + await frontier.clearDown(site) + await siteStore.clearFlag(site).catch((error) => log(`unflag ${site}: ${errorText(error)}`)) if (!allows(robots, '/')) { await frontier.bump('sites_disallowed') await frontier.log({ kind: 'skip', site, note: 'robots.txt disallows pitbot' }) @@ -415,7 +466,10 @@ export function createCrawler(deps: Deps) { if ((await frontier.outboxLength()) > config.outboxHigh) { return { delayMs: 20_000 + Math.floor(Math.random() * 20_000), outcome: 'waiting for publishing to catch up' } } - const robots = await robotsFor(job.site) + const state = await frontier.siteState(job.site, job.url) + const parked = parkedOutcome(state) + if (parked) return parked + const robots = await robotsFor(job.site, state) if (robots === 'unavailable') return { outcome: 'robots.txt unavailable' } const gap = gapFor(robots) const slot = await frontier.lease(job.site, gap) @@ -424,11 +478,25 @@ export function createCrawler(deps: Deps) { await frontier.bump('pages_disallowed') return { outcome: 'disallowed' } } - const r = await get(job.url, job.site) + let r: Awaited> + try { + r = await get(job.url, job.site) + } catch (error) { + // Not thrown on to BullMQ: a retry in 30 s of a page that just timed out is the slow + // broken read the rule is about. Anything else (a bug, Redis) still throws and retries. + const kind = failureKind(error) + if (!kind) throw error + return { outcome: await flagPage(job, kind, errorText(error)) } + } if (r.offsite) { if (config.discover) await scheduleSites([r.offsite], 'redirect') return { outcome: `redirects to ${r.offsite}` } } + const failed = statusKind(r.status) + if (failed) { + await frontier.bump('pages_failed') + return { outcome: await flagPage(job, failed, `HTTP ${r.status}`, r.retryAfter) } + } if (r.status >= 400) { await frontier.bump('pages_failed') await frontier.log({ kind: 'fail', site: job.site, url: job.url, note: `HTTP ${r.status}` }) @@ -446,14 +514,38 @@ export function createCrawler(deps: Deps) { return { outcome: await accept(doc, optOut ? { ...job, optOut } : job, robots) } } + /** + * A parked host or page is not read. A short park (a 429's Retry-After) is waited out; + * a long one drops the job, and the site's recheck queues the page again in 30 days. + */ + function parkedOutcome(state: { downMs: number; pageDown: boolean }): PageOutcome | null { + if (state.pageDown) return { outcome: 'page parked' } + if (state.downMs <= 0) return null + if (state.downMs <= MAX_SPREAD_MS) return { delayMs: state.downMs, outcome: 'waiting out a parked host' } + return { outcome: 'host parked' } + } + async function renderPage(job: PageJob): Promise { if (!deps.renderer?.size) return { outcome: 'no renderer' } - const robots = await robotsFor(job.site) + const state = await frontier.siteState(job.site, job.url) + const parked = parkedOutcome(state) + if (parked) return parked + const robots = await robotsFor(job.site, state) if (robots === 'unavailable') return { outcome: 'robots.txt unavailable' } const gap = gapFor(robots) const slot = await frontier.lease(job.site, gap) if (slot.waitMs > 0) return { delayMs: spreadDelay(slot.waitMs, slot.pending, gap), outcome: 'waiting for the site' } - const rendered = await deps.renderer.render(job.url) + let rendered: Rendered + try { + rendered = await deps.renderer.render(job.url) + } catch (error) { + // Out of the 30 s budget, or the browser could not reach the page: flagged, not retried. + // Obscura's own trouble (HTTP errors from our container) still throws and retries. + const kind = failureKind(error) + if (!kind) throw error + await frontier.bump('renders_failed') + return { outcome: await flagPage(job, kind, `render: ${errorText(error)}`) } + } const landed = normalizeUrl(rendered.url) ?? job.url if (!sameSite(landed, job.site)) return { outcome: `rendered onto ${siteOf(landed)}` } const doc = extract(rendered.html, landed, { depth: job.depth, rendered: true, crawler: config.userAgent }) diff --git a/services/crawler/src/failures.ts b/services/crawler/src/failures.ts new file mode 100644 index 0000000..61a8f4f --- /dev/null +++ b/services/crawler/src/failures.ts @@ -0,0 +1,79 @@ +/** + * Slow or broken targets: what counts as one, and how long it is left alone. + * + * A timeout, a name that does not resolve, a bad certificate, a refused or reset + * connection, or a 5xx is not retried within minutes (BullMQ's backoff used to read a + * dead host's every page three times). The host or page is flagged instead and gets + * exactly one recheck after CRAWL_RECHECK_MS: success clears the flag, failure sets it + * another 30 days. A 429 parks the host for its Retry-After, capped at the same 30 days. + * Every other 4xx keeps its old meaning (the page is dropped, robots.txt reads as no rules). + */ +import { CRAWL_RECHECK_MS } from './config.ts' + +export type FailureKind = 'timeout' | 'dns' | 'tls' | 'connection' | 'http5xx' | 'http429' + +/** Kinds that are the host's, not one page's: every page of it would fail the same way. */ +export const HOST_KINDS: ReadonlySet = new Set(['dns', 'tls', 'connection', 'http429']) + +/** Page timeouts or 5xx from one site inside an hour before the whole site is parked. */ +export const HOST_PAGE_FAILURES = 5 +export const HOST_PAGE_FAILURE_WINDOW_MS = 3_600_000 + +/** A 429 without a usable Retry-After waits a day; with one, never less than a minute. */ +const RETRY_AFTER_DEFAULT_MS = 86_400_000 +const RETRY_AFTER_MIN_MS = 60_000 + +/** A thrown fetch (or render) error as a failure kind; null when it is not the target's fault. */ +export function failureKind(error: unknown): FailureKind | null { + if (!error || typeof error !== 'object') return null + const e = error as { name?: unknown; code?: unknown; message?: unknown; ours?: unknown } + // Our own service failed (src/obscura.ts ObscuraError): not the target's doing. + if (e.ours === true) return null + const name = String(e.name ?? '') + const code = String(e.code ?? '') + const message = String(e.message ?? '') + if (name === 'TimeoutError' || name === 'AbortError' || /\btimed? ?out\b|timeout/i.test(message)) return 'timeout' + if (/^(ENOTFOUND|EAI_AGAIN|EAI_NONAME|EAI_NODATA)$/.test(code) || /getaddrinfo|ENOTFOUND|ERR_NAME_NOT_RESOLVED|not on the public internet/i.test(message)) return 'dns' + if (/CERT|SSL|TLS/i.test(code) || /certificate|\bssl\b|\btls\b|ERR_CERT_|ERR_SSL_/i.test(message)) return 'tls' + if (/^(ECONNREFUSED|ECONNRESET|EHOSTUNREACH|ENETUNREACH|EPIPE|ECONNABORTED|ConnectionClosed|ConnectionRefused)$/.test(code) || /ECONNREFUSED|ECONNRESET|EHOSTUNREACH|ENETUNREACH|ERR_CONNECTION_|socket (connection was )?closed|connection (was )?(closed|reset|refused)/i.test(message)) + return 'connection' + return null +} + +/** A status as a failure kind: 5xx and 429 only. */ +export function statusKind(status: number): FailureKind | null { + if (status === 429) return 'http429' + if (status >= 500 && status <= 599) return 'http5xx' + return null +} + +/** Retry-After (seconds or an HTTP date) in ms from now, clamped to [1 min, 30 days]; a day when absent or unreadable. */ +export function retryAfterMs(header: string | null | undefined, now = Date.now(), cap = CRAWL_RECHECK_MS): number { + const raw = (header ?? '').trim() + let ms = RETRY_AFTER_DEFAULT_MS + if (/^\d+$/.test(raw)) ms = Number(raw) * 1000 + else if (raw) { + const at = Date.parse(raw) + if (Number.isFinite(at)) ms = at - now + } + return Math.min(cap, Math.max(RETRY_AFTER_MIN_MS, ms)) +} + +/** How long a failure of this kind keeps its target away. */ +export function parkFor(kind: FailureKind, retryAfter?: string | null, now = Date.now()): number { + return kind === 'http429' ? retryAfterMs(retryAfter, now) : CRAWL_RECHECK_MS +} + +/** The flag a site row carries while it is parked. */ +export type SiteFlag = { status: 'down'; lastError: string; failedAt: number; retryAfter: number } + +export function flagFor(error: string, untilMs: number, now = Date.now()): SiteFlag { + return { status: 'down', lastError: error.slice(0, 300), failedAt: now, retryAfter: untilMs } +} + +/** A scheduled site is due again once its cycle has run out AND any flag on it has. */ +export function isDue(row: { scheduledAt: number | null; retryAfter?: number | null }, cycleMs: number, now = Date.now()): boolean { + if (row.scheduledAt === null) return false + if (row.scheduledAt > now - cycleMs) return false + return !row.retryAfter || row.retryAfter <= now +} diff --git a/services/crawler/src/fetcher.ts b/services/crawler/src/fetcher.ts index 73c0a83..80246f6 100644 --- a/services/crawler/src/fetcher.ts +++ b/services/crawler/src/fetcher.ts @@ -1,7 +1,12 @@ /** * One polite GET: a named user agent, a timeout, a size cap, and redirects * followed by hand so every hop is checked against the private-network guard. + * + * The timeout is one budget for the whole GET (CRAWL_FETCH_TIMEOUT_MS, 10 s): the name + * lookup, every redirect hop and the body. It used to be 20 s per hop, so five redirects + * could hold a worker for over a minute and a slow body was never cut off. */ +import { CRAWL_FETCH_TIMEOUT_MS } from './config.ts' import { normalizeUrl, type Resolver } from './url.ts' export type Fetched = { @@ -15,6 +20,8 @@ export type Fetched = { offsite?: string /** The page's own opt-out headers (X-Robots-Tag, TDMRep's tdm-reservation), lowercased. */ optOut?: { robotsTag: string; tdmReservation: string } + /** Retry-After, as sent (a 429 or 503 may carry one). */ + retryAfter?: string } type Fetch = typeof fetch @@ -22,13 +29,36 @@ type Fetch = typeof fetch export const MAX_BYTES = 2 * 1024 * 1024 const MAX_REDIRECTS = 5 -export async function readCapped(response: Response, max: number): Promise { +/** Reject when the signal fires, whatever `work` is waiting on (a name lookup has no signal of its own). */ +export function withSignal(work: Promise, signal: AbortSignal): Promise { + if (signal.aborted) return Promise.reject(signal.reason) + return new Promise((resolve, reject) => { + const abort = () => reject(signal.reason) + signal.addEventListener('abort', abort, { once: true }) + work.then( + (v) => { + signal.removeEventListener('abort', abort) + resolve(v) + }, + (e) => { + signal.removeEventListener('abort', abort) + reject(e) + }, + ) + }) +} + +export async function readCapped(response: Response, max: number, signal?: AbortSignal): Promise { if (!response.body) return '' const reader = response.body.getReader() const chunks: Uint8Array[] = [] let size = 0 while (true) { - const { done, value } = await reader.read() + // A body that trickles in past the budget is a failure, not a short page. + const { done, value } = signal ? await withSignal(reader.read(), signal).catch(async (error) => { + await reader.cancel().catch(() => {}) + throw error + }) : await reader.read() if (done) break size += value.byteLength if (size > max) { @@ -46,10 +76,11 @@ export async function politeGet( options: { userAgent: string; resolver: Resolver; fetch?: Fetch; timeoutMs?: number; accept?: string; site?: string; maxBytes?: number }, ): Promise { const fetchImpl = options.fetch ?? fetch + const signal = AbortSignal.timeout(options.timeoutMs ?? CRAWL_FETCH_TIMEOUT_MS) let current = url for (let hop = 0; hop <= MAX_REDIRECTS; hop++) { const host = new URL(current).hostname - if (!(await options.resolver(host))) throw new Error(`refused: ${host} is not on the public internet`) + if (!(await withSignal(options.resolver(host), signal))) throw new Error(`refused: ${host} is not on the public internet`) const response = await fetchImpl(current, { redirect: 'manual', headers: { @@ -57,7 +88,7 @@ export async function politeGet( accept: options.accept ?? 'text/html,application/xhtml+xml;q=0.9,*/*;q=0.1', 'accept-language': 'en;q=1, *;q=0.5', }, - signal: AbortSignal.timeout(options.timeoutMs ?? 20_000), + signal, }) if (response.status >= 300 && response.status < 400 && response.headers.get('location')) { await response.body?.cancel().catch(() => {}) @@ -72,10 +103,11 @@ export async function politeGet( } const contentType = (response.headers.get('content-type') ?? '').toLowerCase() const textual = /text\/|html|xml|json/.test(contentType) || contentType === '' - const body = textual ? await readCapped(response, options.maxBytes ?? MAX_BYTES) : '' + const body = textual ? await readCapped(response, options.maxBytes ?? MAX_BYTES, signal) : '' if (!textual) await response.body?.cancel().catch(() => {}) const optOut = { robotsTag: (response.headers.get('x-robots-tag') ?? '').toLowerCase(), tdmReservation: (response.headers.get('tdm-reservation') ?? '').trim() } - return { url, finalUrl: current, status: response.status, contentType, body, optOut } + const retryAfter = response.headers.get('retry-after') + return { url, finalUrl: current, status: response.status, contentType, body, optOut, ...(retryAfter ? { retryAfter } : {}) } } throw new Error('too many redirects') } diff --git a/services/crawler/src/frontier.ts b/services/crawler/src/frontier.ts index d7b15d3..91f2e93 100644 --- a/services/crawler/src/frontier.ts +++ b/services/crawler/src/frontier.ts @@ -6,6 +6,9 @@ * pitcrawl:count: pages admitted this cycle; admission stops at maxPages * pitcrawl:robots: the parsed robots.txt, for a day * pitcrawl:lease: held for the gap between two requests to one site + * pitcrawl:down: a parked host (timeout, no connection, 5xx, 429), expires when it may be read again + * pitcrawl:pdown: a parked page (timed out or 5xx), 30 days + * pitcrawl:fails: page failures this hour; past HOST_PAGE_FAILURES the host is parked * pitcrawl:sites sorted set: site -> when it was last scheduled * pitcrawl:outbox records waiting for nichedb, oldest first * pitcrawl:stats counters @@ -57,6 +60,19 @@ export interface Frontier { lease(site: string, gapMs: number): Promise<{ waitMs: number; pending: number }> getRobots(site: string): Promise setRobots(site: string, robots: Robots): Promise + /** + * One round trip for what a page read needs first: the cached robots.txt, how long the + * host stays parked (0 = not), and whether this page is parked. + */ + siteState(site: string, url?: string): Promise<{ robots: Robots | null; downMs: number; pageDown: boolean }> + /** Park a host for ms: nothing reads from it until then. */ + markDown(site: string, ms: number): Promise + /** Park one page for ms. */ + markPageDown(url: string, ms: number): Promise + /** Count a page failure in a fixed window; returns the count so far in it. */ + noteFailure(site: string, windowMs: number): Promise + /** The host read fine: forget its failures. */ + clearDown(site: string): Promise /** The first start of a site in a cycle claims it; every later one is told no. */ claimStart(site: string, cycleMs: number): Promise /** The first sighting of an MCP address in a cycle claims its probe; every later one is told no. */ @@ -104,6 +120,9 @@ const K = { robots: (s: string) => `pitcrawl:robots:${s}`, lease: (s: string) => `pitcrawl:lease:${s}`, started: (s: string) => `pitcrawl:started:${s}`, + down: (s: string) => `pitcrawl:down:${s}`, + pageDown: (id: string) => `pitcrawl:pdown:${id}`, + fails: (s: string) => `pitcrawl:fails:${s}`, outbox: 'pitcrawl:outbox', stats: 'pitcrawl:stats', mcp: 'pitcrawl:mcp', @@ -199,6 +218,37 @@ export class RedisFrontier implements Frontier { await this.redis.set(K.robots(site), JSON.stringify(robots), 'EX', 86_400) } + async siteState(site: string, url?: string) { + const pipe = this.redis.pipeline().get(K.robots(site)).pttl(K.down(site)) + if (url) pipe.exists(K.pageDown(urlId(url))) + const results = ((await pipe.exec()) ?? []) as [unknown, unknown][] + const raw = results[0]?.[1] + const ttl = results[1]?.[1] + const page = results[2] + let robots: Robots | null = null + if (typeof raw === 'string') { + try { + robots = JSON.parse(raw) as Robots + } catch { + robots = EMPTY + } + } + return { robots, downMs: Math.max(0, Number(ttl ?? 0)), pageDown: Number(page?.[1] ?? 0) > 0 } + } + async markDown(site: string, ms: number) { + await this.redis.set(K.down(site), String(Date.now()), 'PX', Math.max(1_000, Math.floor(ms))) + } + async markPageDown(url: string, ms: number) { + await this.redis.set(K.pageDown(urlId(url)), '1', 'PX', Math.max(1_000, Math.floor(ms))) + } + async noteFailure(site: string, windowMs: number) { + const results = ((await this.redis.multi().incr(K.fails(site)).pexpire(K.fails(site), windowMs, 'NX').exec()) ?? []) as [unknown, unknown][] + return Number(results[0]?.[1] ?? 0) + } + async clearDown(site: string) { + await this.redis.unlink(K.down(site), K.fails(site)) + } + async claimMcp(endpoint: string, cycleMs: number) { return (await this.redis.set(K.mcpSeen(urlId(endpoint)), '1', 'PX', cycleMs, 'NX')) === 'OK' } @@ -388,6 +438,29 @@ export class MemoryFrontier implements Frontier { async setRobots(site: string, robots: Robots) { this.robots.set(site, robots) } + down = new Map() + pageDown = new Map() + fails = new Map() + async siteState(site: string, url?: string) { + const until = this.down.get(site) ?? 0 + return { robots: this.robots.get(site) ?? null, downMs: Math.max(0, until - Date.now()), pageDown: url ? (this.pageDown.get(urlId(url)) ?? 0) > Date.now() : false } + } + async markDown(site: string, ms: number) { + this.down.set(site, Date.now() + ms) + } + async markPageDown(url: string, ms: number) { + this.pageDown.set(urlId(url), Date.now() + ms) + } + async noteFailure(site: string, windowMs: number) { + const f = this.fails.get(site) + const next = f && f.until > Date.now() ? { n: f.n + 1, until: f.until } : { n: 1, until: Date.now() + windowMs } + this.fails.set(site, next) + return next.n + } + async clearDown(site: string) { + this.down.delete(site) + this.fails.delete(site) + } started = new Set() async claimStart(site: string) { if (this.started.has(site)) return false diff --git a/services/crawler/src/obscura.ts b/services/crawler/src/obscura.ts index 11391f5..38fbeed 100644 --- a/services/crawler/src/obscura.ts +++ b/services/crawler/src/obscura.ts @@ -10,8 +10,27 @@ * and put back together as a document the extractor reads like any other. */ +import { CRAWL_RENDER_SETTLE_MS, CRAWL_RENDER_TIMEOUT_MS } from './config.ts' + export type Rendered = { url: string; html: string } +/** + * Trouble on our side of the render (Obscura unreachable, a dropped session, a JSON-RPC + * error): never the target's fault, so never a reason to park it. `ours` tells failureKind. + */ +export class ObscuraError extends Error { + override name = 'ObscuraError' + readonly ours = true +} + +/** A render that ran out of CRAWL_RENDER_TIMEOUT_MS: a failure to record, never a page. */ +export class RenderTimeout extends Error { + override name = 'TimeoutError' + constructor(url: string, ms: number) { + super(`render timed out after ${Math.round(ms / 1000)} s: ${url}`) + } +} + type Fetch = typeof fetch const EXPRESSION = `(() => { @@ -27,6 +46,10 @@ const EXPRESSION = `(() => { }); })()` +/** The read back, after a bounded settle: `load` has fired, a late script gets this long, then we read. */ +export const SETTLED_EXPRESSION = (settleMs: number) => + `new Promise((resolve) => setTimeout(resolve, ${Math.max(0, Math.min(2_000, Math.floor(settleMs)))})).then(() => ${EXPRESSION})` + const esc = (s: string) => s.replace(/&/g, '&').replace(//g, '>').replace(/"/g, '"') /** The evaluate result as HTML: head verbatim, text as paragraphs, links as anchors. */ @@ -55,11 +78,24 @@ export class ObscuraClient { readonly endpoint: string, private readonly token: string, private readonly fetchImpl: Fetch = fetch, - private readonly timeoutMs = 45_000, + /** The whole render: navigation, settle and read back (CRAWL_RENDER_TIMEOUT_MS). */ + private readonly budgetMs = CRAWL_RENDER_TIMEOUT_MS, + private readonly settleMs = CRAWL_RENDER_SETTLE_MS, ) {} + /** When the render in progress must be done by; every call inside it gets only what is left. */ + private deadline = 0 + + private left(): number { + return this.deadline ? this.deadline - Date.now() : this.budgetMs + } + private async rpc(method: string, params: unknown, notify = false): Promise { - const response = await this.fetchImpl(this.endpoint, { + const left = this.left() + if (left <= 0) throw new Error('render budget spent') + let response: Response + try { + response = await this.fetchImpl(this.endpoint, { method: 'POST', headers: { 'content-type': 'application/json', @@ -68,16 +104,21 @@ export class ObscuraClient { ...(this.session ? { 'mcp-session-id': this.session } : {}), }, body: JSON.stringify(notify ? { jsonrpc: '2.0', method, params } : { jsonrpc: '2.0', id: ++this.id, method, params }), - signal: AbortSignal.timeout(this.timeoutMs), - }) + signal: AbortSignal.timeout(left), + }) + } catch (error) { + const name = (error as Error)?.name + if (name === 'TimeoutError' || name === 'AbortError') throw error + throw new ObscuraError(`obscura unreachable: ${(error as Error)?.message ?? error}`) + } this.session = response.headers.get('mcp-session-id') ?? this.session const text = await response.text() if (response.status === 404 || response.status === 400) { // A restarted server forgot the session. this.session = '' - throw new Error(`obscura HTTP ${response.status}`) + throw new ObscuraError(`obscura HTTP ${response.status}`) } - if (!response.ok) throw new Error(`obscura HTTP ${response.status}`) + if (!response.ok) throw new ObscuraError(`obscura HTTP ${response.status}`) if (notify) return null const line = text.includes('data:') ? text @@ -86,8 +127,13 @@ export class ObscuraClient { .pop()! .slice(5) : text - const json = JSON.parse(line) - if (json.error) throw new Error(`obscura ${json.error.message ?? 'error'}`) + let json: any + try { + json = JSON.parse(line) + } catch { + throw new ObscuraError('obscura sent a reply that is not JSON') + } + if (json.error) throw new ObscuraError(`obscura ${json.error.message ?? 'error'}`) return json.result } @@ -113,13 +159,27 @@ export class ObscuraClient { return text } - /** Navigate, let the page settle, read it back. Serialised: one page per server. */ + /** + * Navigate to `load` (never networkidle: a page that keeps a socket open never goes idle), + * settle briefly, read it back, all inside one budget. Out of budget it rejects with + * RenderTimeout, which the crawler records as a slow target. Serialised: one page per + * server, and the budget starts when this page's turn does, not while it queues. + */ render(url: string): Promise { const run = this.busy.then(async () => { - await this.call('browser_navigate', { url, waitUntil: 'networkidle0' }) - const raw = await this.call('browser_evaluate', { expression: EXPRESSION, timeoutMs: 15_000 }) - const parts = JSON.parse(raw) - return { url: String(parts.url || url), html: documentFrom(parts) } + this.deadline = Date.now() + this.budgetMs + try { + await this.call('browser_navigate', { url, waitUntil: 'load' }) + const timeoutMs = Math.max(1, this.left() - 500) + const raw = await this.call('browser_evaluate', { expression: SETTLED_EXPRESSION(Math.min(this.settleMs, timeoutMs / 2)), timeoutMs }) + const parts = JSON.parse(raw) + return { url: String(parts.url || url), html: documentFrom(parts) } + } catch (error) { + if (this.left() <= 0 || (error as Error)?.name === 'TimeoutError' || (error as Error)?.name === 'AbortError') throw new RenderTimeout(url, this.budgetMs) + throw error + } finally { + this.deadline = 0 + } }) this.busy = run.catch(() => undefined) return run diff --git a/services/crawler/src/site-store.ts b/services/crawler/src/site-store.ts index 13933ed..dcb5d20 100644 --- a/services/crawler/src/site-store.ts +++ b/services/crawler/src/site-store.ts @@ -10,6 +10,7 @@ * statements, and recomputed by resync(), since count(*) over 100M rows is seconds. */ import { SQL } from 'bun' +import { flagFor, isDue, type SiteFlag } from './failures.ts' export type SiteRow = { site: string; source: string } export type SiteCounts = { known: number; parked: number } @@ -21,8 +22,14 @@ export interface SiteStore { force(rows: SiteRow[]): Promise /** The oldest parked sites, now marked started. */ takeParked(count: number): Promise - /** Started sites last scheduled more than olderThanMs ago, now rescheduled. */ + /** Started sites last scheduled more than olderThanMs ago and not flagged into the future, now rescheduled. */ takeDue(olderThanMs: number, count: number): Promise + /** Park a site: it timed out, would not connect or answered 5xx/429. Not due again before `untilMs`. */ + flag(site: string, error: string, untilMs: number): Promise + /** A flagged site read fine again: drop the flag. A no-op on an unflagged site. */ + clearFlag(site: string): Promise + /** The flag a site carries, or null. */ + flagOf(site: string): Promise /** From the counters: cheap, slightly behind under concurrent writes until resync(). */ counts(): Promise /** Recompute the counters exactly. */ @@ -55,6 +62,16 @@ const SCHEMA = [ `create index if not exists sites_due on sites (scheduled_at) where scheduled_at is not null`, `create table if not exists counts (k text primary key, n bigint not null)`, `insert into counts (k, n) values ('known', 0), ('parked', 0) on conflict do nothing`, + // 2026-11-08, the 30-day recheck flag (src/failures.ts): status 'down' while parked, why, + // when, and when it may be read again. Added only when missing: ALTER TABLE takes an + // ACCESS EXCLUSIVE lock even with IF NOT EXISTS, and every worker runs this at start. + // Nullable with no default, so adding them rewrites nothing. + `do $$ begin + if not exists (select 1 from information_schema.columns where table_schema = current_schema() and table_name = 'sites' and column_name = 'retry_after') then + alter table sites add column if not exists status text, add column if not exists last_error text, + add column if not exists failed_at timestamptz, add column if not exists retry_after timestamptz; + end if; + end $$`, ] export class PgSiteStore implements SiteStore { @@ -138,12 +155,42 @@ export class PgSiteStore implements SiteStore { with picked as ( select site from sites where scheduled_at is not null and scheduled_at < now() - make_interval(secs => ${olderThanMs / 1000}::float8) + and (retry_after is null or retry_after <= now()) order by scheduled_at limit ${count} for update skip locked ) update sites s set scheduled_at = now() from picked where s.site = picked.site returning s.site` return (out as { site: string }[]).map((r) => r.site) } + async flag(site: string, error: string, untilMs: number) { + await this.init() + const f = flagFor(error, untilMs) + // A site flagged before it was ever recorded (a card read, an MCP probe) is recorded as + // started now, so the recheck finds it in sites_due. + const out = await this.sql` + insert into sites (site, source, scheduled_at, status, last_error, failed_at, retry_after) + values (${site}, 'flag', now(), ${f.status}, ${f.lastError}, now(), to_timestamp(${untilMs / 1000}::float8)) + on conflict (site) do update set status = excluded.status, last_error = excluded.last_error, + failed_at = excluded.failed_at, retry_after = excluded.retry_after, + scheduled_at = coalesce(sites.scheduled_at, excluded.scheduled_at) + returning (xmax = 0) as inserted` + if ((out as { inserted: boolean }[])[0]?.inserted) await this.sql`update counts set n = n + 1 where k = 'known'` + } + + async clearFlag(site: string) { + await this.init() + await this.sql`update sites set status = null, last_error = null, failed_at = null, retry_after = null where site = ${site} and status is not null` + } + + async flagOf(site: string) { + await this.init() + const out = (await this.sql` + select status, last_error, extract(epoch from failed_at) * 1000 as failed_at, extract(epoch from retry_after) * 1000 as retry_after + from sites where site = ${site} and status is not null`) as { status: 'down'; last_error: string; failed_at: string | number; retry_after: string | number }[] + const r = out[0] + return r ? { status: r.status, lastError: r.last_error, failedAt: Number(r.failed_at), retryAfter: Number(r.retry_after) } : null + } + async counts() { // Every discovered link asks, to honour the site limit: a few seconds stale is fine. if (this.cached && Date.now() - this.cached.at < 10_000) return this.cached.counts @@ -183,7 +230,7 @@ export class PgSiteStore implements SiteStore { /** For tests and a laptop without Postgres. Nothing survives a restart. */ export class MemorySiteStore implements SiteStore { - rows = new Map() + rows = new Map() private next = 1 async add(rows: SiteRow[], queueNow: boolean) { @@ -211,13 +258,28 @@ export class MemorySiteStore implements SiteStore { return picked.map(([site, r]) => ({ site, source: r.source })) } async takeDue(olderThanMs: number, count: number) { + const now = Date.now() const due = [...this.rows.entries()] - .filter(([, r]) => r.scheduledAt !== null && r.scheduledAt <= Date.now() - olderThanMs) + .filter(([, r]) => isDue({ scheduledAt: r.scheduledAt, retryAfter: r.flag?.retryAfter ?? null }, olderThanMs, now)) .sort((a, b) => a[1].scheduledAt! - b[1].scheduledAt!) .slice(0, count) for (const [, r] of due) r.scheduledAt = Date.now() return due.map(([site]) => site) } + async flag(site: string, error: string, untilMs: number) { + const row = this.rows.get(site) + if (row) { + row.flag = flagFor(error, untilMs) + row.scheduledAt ??= Date.now() + } else this.rows.set(site, { source: 'flag', scheduledAt: Date.now(), id: this.next++, flag: flagFor(error, untilMs) }) + } + async clearFlag(site: string) { + const row = this.rows.get(site) + if (row) row.flag = null + } + async flagOf(site: string) { + return this.rows.get(site)?.flag ?? null + } async counts() { let parked = 0 for (const r of this.rows.values()) if (r.scheduledAt === null) parked++ diff --git a/services/crawler/test/failures.test.ts b/services/crawler/test/failures.test.ts new file mode 100644 index 0000000..57cc390 --- /dev/null +++ b/services/crawler/test/failures.test.ts @@ -0,0 +1,287 @@ +/** + * The fleet crawler rule (2026-11-08): 10 s per fetch, 30 s per render, and a slow or + * broken target is flagged and left alone for 30 days, then rechecked exactly once. + */ +import { describe, expect, test } from 'bun:test' +import { configFromEnv, CRAWL_FETCH_TIMEOUT_MS, CRAWL_RECHECK_MS, CRAWL_RENDER_SETTLE_MS, CRAWL_RENDER_TIMEOUT_MS } from '../src/config.ts' +import { createCrawler, type PageJob, type SiteJob } from '../src/crawler.ts' +import { failureKind, HOST_PAGE_FAILURES, isDue, parkFor, retryAfterMs, statusKind } from '../src/failures.ts' +import { politeGet } from '../src/fetcher.ts' +import { MemoryFrontier } from '../src/frontier.ts' +import { ObscuraClient, ObscuraError, RenderTimeout, SETTLED_EXPRESSION } from '../src/obscura.ts' +import { MemorySiteStore } from '../src/site-store.ts' + +const DAY = 86_400_000 + +describe('the budgets', () => { + test('10 s per fetch, 30 s per render, a settle of at most 2 s, a 30-day recheck', () => { + expect(CRAWL_FETCH_TIMEOUT_MS).toBe(10_000) + expect(CRAWL_RENDER_TIMEOUT_MS).toBe(30_000) + expect(CRAWL_RENDER_SETTLE_MS).toBeLessThanOrEqual(2_000) + expect(CRAWL_RECHECK_MS).toBe(30 * DAY) + expect(SETTLED_EXPRESSION(60_000)).toContain('setTimeout(resolve, 2000)') + }) + + test('one budget covers every redirect hop and the body, not 20 s a hop', async () => { + let hops = 0 + const slow = (async (input: string | URL | Request, init?: RequestInit) => { + hops++ + await new Promise((resolve, reject) => { + const t = setTimeout(resolve, 40) + init?.signal?.addEventListener('abort', () => (clearTimeout(t), reject(init.signal!.reason))) + }) + return new Response('', { status: 301, headers: { location: `https://a.example/${hops}` } }) + }) as unknown as typeof fetch + const started = Date.now() + const error = await politeGet('https://a.example/', { userAgent: 't', resolver: async () => true, fetch: slow, timeoutMs: 100 }).catch((e) => e) + expect(failureKind(error)).toBe('timeout') + expect(Date.now() - started).toBeLessThan(400) + expect(hops).toBeLessThanOrEqual(4) + }) + + test('a body that trickles past the budget is a timeout, not a short page', async () => { + const trickle = (async () => { + const stream = new ReadableStream({ + async start(controller) { + controller.enqueue(new TextEncoder().encode('')) + await new Promise((r) => setTimeout(r, 500)) + controller.close() + }, + }) + return new Response(stream, { headers: { 'content-type': 'text/html' } }) + }) as unknown as typeof fetch + const error = await politeGet('https://a.example/', { userAgent: 't', resolver: async () => true, fetch: trickle, timeoutMs: 100 }).catch((e) => e) + expect(failureKind(error)).toBe('timeout') + }) + + test('a name lookup that hangs is cut off too', async () => { + const error = await politeGet('https://a.example/', { userAgent: 't', resolver: () => new Promise(() => {}), timeoutMs: 50 }).catch((e) => e) + expect(failureKind(error)).toBe('timeout') + }) +}) + +describe('what counts as slow or broken', () => { + test("Bun's fetch errors, by kind", () => { + expect(failureKind({ name: 'TimeoutError', message: 'The operation timed out.' })).toBe('timeout') + expect(failureKind({ name: 'TypeError', code: 'ENOTFOUND', message: 'getaddrinfo ENOTFOUND x.invalid' })).toBe('dns') + expect(failureKind(new Error('refused: x.example is not on the public internet'))).toBe('dns') + expect(failureKind({ name: 'TypeError', code: 'DEPTH_ZERO_SELF_SIGNED_CERT', message: 'self signed certificate' })).toBe('tls') + expect(failureKind({ name: 'TypeError', code: 'CERT_HAS_EXPIRED', message: 'certificate has expired' })).toBe('tls') + expect(failureKind({ name: 'TypeError', code: 'ECONNREFUSED', message: 'ECONNREFUSED: Unable to connect.' })).toBe('connection') + expect(failureKind(new RenderTimeout('https://a.example/', 30_000))).toBe('timeout') + expect(failureKind(new Error('obscura browser_navigate: net::ERR_NAME_NOT_RESOLVED'))).toBe('dns') + }) + + test('our own trouble and plain bugs are not the target', () => { + expect(failureKind(new ObscuraError('obscura unreachable: ECONNREFUSED'))).toBeNull() + expect(failureKind(new ObscuraError('obscura HTTP 502'))).toBeNull() + expect(failureKind(new Error('too many redirects'))).toBeNull() + expect(failureKind(new TypeError("undefined is not an object"))).toBeNull() + expect(failureKind(null)).toBeNull() + }) + + test('5xx and 429 fail; every other status keeps its meaning', () => { + expect(statusKind(500)).toBe('http5xx') + expect(statusKind(503)).toBe('http5xx') + expect(statusKind(429)).toBe('http429') + expect(statusKind(404)).toBeNull() + expect(statusKind(403)).toBeNull() + expect(statusKind(200)).toBeNull() + }) + + test('Retry-After is honoured but capped at 30 days; absent, a day', () => { + const now = Date.parse('2026-11-08T00:00:00Z') + expect(retryAfterMs('120', now)).toBe(120_000) + expect(retryAfterMs('1', now)).toBe(60_000) + expect(retryAfterMs(String(365 * 86_400), now)).toBe(30 * DAY) + expect(retryAfterMs('Mon, 09 Nov 2026 00:00:00 GMT', now)).toBe(DAY) + expect(retryAfterMs(null, now)).toBe(DAY) + expect(retryAfterMs('soon', now)).toBe(DAY) + expect(parkFor('timeout')).toBe(30 * DAY) + expect(parkFor('http429', '600')).toBe(600_000) + }) + + test('due again only when the cycle AND the flag have run out: exactly one recheck', () => { + const now = Date.now() + expect(isDue({ scheduledAt: now - 31 * DAY, retryAfter: null }, 30 * DAY, now)).toBe(true) + expect(isDue({ scheduledAt: now - 31 * DAY, retryAfter: now + DAY }, 30 * DAY, now)).toBe(false) + expect(isDue({ scheduledAt: now - 31 * DAY, retryAfter: now - 1 }, 30 * DAY, now)).toBe(true) + expect(isDue({ scheduledAt: now - DAY, retryAfter: now - 1 }, 30 * DAY, now)).toBe(false) + expect(isDue({ scheduledAt: null, retryAfter: null }, 30 * DAY, now)).toBe(false) + }) +}) + +describe('the site store flag', () => { + test('a flagged site is skipped by the recrawl until retry_after, rechecked once, cleared on success', async () => { + const store = new MemorySiteStore() + await store.add([{ site: 'dead.example', source: 'x' }, { site: 'fine.example', source: 'x' }], true) + for (const row of store.rows.values()) row.scheduledAt = Date.now() - 31 * DAY + await store.flag('dead.example', 'timeout: robots.txt', Date.now() + 30 * DAY) + expect((await store.flagOf('dead.example'))?.lastError).toBe('timeout: robots.txt') + expect(await store.takeDue(30 * DAY, 10)).toEqual(['fine.example']) + // 30 days on: one recheck, and only one. + store.rows.get('dead.example')!.flag!.retryAfter = Date.now() - 1 + expect(await store.takeDue(30 * DAY, 10)).toEqual(['dead.example']) + expect(await store.takeDue(30 * DAY, 10)).toEqual([]) + await store.clearFlag('dead.example') + expect(await store.flagOf('dead.example')).toBeNull() + }) +}) + +const text = (body: string, status = 200, headers: Record = {}) => new Response(body, { status, headers: { 'content-type': 'text/plain', ...headers } }) +const page = (body: string) => new Response(`t
${body}
`, { headers: { 'content-type': 'text/html' } }) + +function setup(web: (url: string) => Promise | Response, renderer?: { size: number; render(url: string): Promise<{ url: string; html: string }> }) { + const config = configFromEnv({ CRAWL_HOST_DELAY_MS: '0' }) + const frontier = new MemoryFrontier(config.maxPages) + const sites = new MemorySiteStore() + const calls: string[] = [] + const queued = { sites: [] as SiteJob[], pages: [] as PageJob[], render: [] as PageJob[] } + const crawler = createCrawler({ + config, + frontier, + sites, + resolver: async () => true, + fetch: (async (input: string | URL | Request) => { + const url = String(input instanceof Request ? input.url : input) + calls.push(url) + return web(url) + }) as typeof fetch, + renderer: renderer ?? null, + enqueue: { + async sites(jobs) { + queued.sites.push(...jobs) + }, + async pages(jobs) { + queued.pages.push(...jobs) + }, + async render(job) { + queued.render.push(job) + }, + }, + }) + return { crawler, frontier, sites, calls, queued } +} + +const timeout = () => Promise.reject(Object.assign(new Error('The operation timed out.'), { name: 'TimeoutError' })) + +describe('a slow or broken host in the crawl', () => { + test('robots.txt timing out parks the host 30 days instead of reading it as "no rules"', async () => { + const { crawler, frontier, sites, calls, queued } = setup((url) => (url.endsWith('/robots.txt') ? timeout() : page('hi'))) + expect((await crawler.crawlSite({ site: 'slow.example', source: 'x' })).outcome).toBe('robots.txt unavailable') + expect(queued.pages).toEqual([]) + const flag = await sites.flagOf('slow.example') + expect(flag?.lastError).toMatch(/^timeout: robots.txt/) + expect(flag!.retryAfter - Date.now()).toBeGreaterThan(29 * DAY) + expect((await frontier.siteState('slow.example')).downMs).toBeGreaterThan(29 * DAY) + // Its queued pages are dropped without a single request. + calls.length = 0 + expect((await crawler.crawlPage({ url: 'https://slow.example/a', site: 'slow.example', depth: 1 })).outcome).toBe('host parked') + expect(calls).toEqual([]) + }) + + test('robots.txt 5xx parks the host; a 429 parks it for its Retry-After; a 404 is still "no rules"', async () => { + const a = setup((url) => (url.endsWith('/robots.txt') ? text('down', 503) : page('hi'))) + expect((await a.crawler.crawlSite({ site: 'five.example', source: 'x' })).outcome).toBe('robots.txt unavailable') + expect((await a.sites.flagOf('five.example'))?.lastError).toBe('http5xx: robots.txt answered HTTP 503') + + const b = setup((url) => (url.endsWith('/robots.txt') ? text('slow down', 429, { 'retry-after': '600' }) : page('hi'))) + await b.crawler.crawlSite({ site: 'busy.example', source: 'x' }) + const down = (await b.frontier.siteState('busy.example')).downMs + expect(down).toBeGreaterThan(590_000) + expect(down).toBeLessThanOrEqual(600_000) + // A short park is waited out, not dropped. + expect((await b.crawler.crawlPage({ url: 'https://busy.example/a', site: 'busy.example', depth: 1 })).delayMs).toBeGreaterThan(0) + + const c = setup((url) => (url.endsWith('/robots.txt') ? text('nope', 404) : page('hi'))) + expect((await c.crawler.crawlSite({ site: 'ok.example', source: 'x' })).outcome).toBe('started') + expect(await c.sites.flagOf('ok.example')).toBeNull() + }) + + test('a page that times out is flagged, not thrown for BullMQ to retry; enough of them park the host', async () => { + const { crawler, frontier, sites } = setup((url) => (url.endsWith('/robots.txt') ? text('') : url.includes('/slow') ? timeout() : page('hi'))) + const job = (n: number) => ({ url: `https://big.example/slow${n}`, site: 'big.example', depth: 1 }) + expect((await crawler.crawlPage(job(0))).outcome).toBe('page parked: timeout') + expect((await frontier.siteState('big.example', job(0).url)).pageDown).toBe(true) + expect((await crawler.crawlPage(job(0))).outcome).toBe('page parked') + expect(await sites.flagOf('big.example')).toBeNull() + for (let i = 1; i < HOST_PAGE_FAILURES; i++) await crawler.crawlPage(job(i)) + expect((await sites.flagOf('big.example'))?.lastError).toMatch(/pages failed within the hour/) + expect((await crawler.crawlPage({ url: 'https://big.example/fine', site: 'big.example', depth: 1 })).outcome).toBe('host parked') + }) + + test('a connection failure on a page parks the host at once; a 5xx parks the page', async () => { + const refused = () => Promise.reject(Object.assign(new TypeError('ECONNREFUSED: Unable to connect.'), { code: 'ECONNREFUSED' })) + const a = setup((url) => (url.endsWith('/robots.txt') ? text('') : refused())) + expect((await a.crawler.crawlPage({ url: 'https://gone.example/a', site: 'gone.example', depth: 1 })).outcome).toBe('host parked: connection') + expect((await a.sites.flagOf('gone.example'))?.lastError).toMatch(/^connection:/) + + const b = setup((url) => (url.endsWith('/robots.txt') ? text('') : text('oops', 502))) + expect((await b.crawler.crawlPage({ url: 'https://flaky.example/a', site: 'flaky.example', depth: 1 })).outcome).toBe('page parked: http5xx') + }) + + test('a bug is still thrown (and retried): only target failures are swallowed', async () => { + const { crawler } = setup((url) => (url.endsWith('/robots.txt') ? text('') : Promise.reject(new Error('something else')))) + await expect(crawler.crawlPage({ url: 'https://x.example/a', site: 'x.example', depth: 1 })).rejects.toThrow('something else') + }) + + test('the recheck clears the flag when the host answers again', async () => { + let dead = true + const { crawler, frontier, sites } = setup((url) => (dead ? timeout() : url.endsWith('/robots.txt') ? text('') : page('hi'))) + await crawler.crawlSite({ site: 'back.example', source: 'x' }) + expect(await sites.flagOf('back.example')).not.toBeNull() + // 30 days later: the flag, the Redis park and the cycle claim have all run out. + sites.rows.get('back.example')!.scheduledAt = Date.now() - 31 * DAY + sites.rows.get('back.example')!.flag!.retryAfter = Date.now() - 1 + frontier.down.delete('back.example') + frontier.started.delete('back.example') + expect(await crawler.recrawl()).toBe(1) + expect(await crawler.recrawl()).toBe(0) + dead = false + expect((await crawler.crawlSite({ site: 'back.example', source: 'recrawl' })).outcome).toBe('started') + expect(await sites.flagOf('back.example')).toBeNull() + }) +}) + +describe('the browser budget', () => { + test('a render out of its 30 s is a timeout to flag, never a page', async () => { + const hang = (async (_: string | URL | Request, init?: RequestInit) => + new Promise((_, reject) => init?.signal?.addEventListener('abort', () => reject(init.signal!.reason)))) as typeof fetch + const client = new ObscuraClient('http://obscura.test/mcp', 't', hang, 80, 10) + const error = await client.render('https://slow.example/').catch((e) => e) + expect(error).toBeInstanceOf(RenderTimeout) + expect(failureKind(error)).toBe('timeout') + }) + + test("navigates to load, never networkidle, and reads back after a bounded settle", async () => { + const sent: any[] = [] + const ok = (async (_: string | URL | Request, init?: RequestInit) => { + const body = JSON.parse(String(init?.body)) + sent.push(body) + const result = body.method === 'tools/call' && body.params.name === 'browser_evaluate' ? { content: [{ text: JSON.stringify({ url: 'https://a.example/', text: 'hi' }) }] } : { content: [] } + return new Response(JSON.stringify({ jsonrpc: '2.0', id: body.id, result }), { headers: { 'content-type': 'application/json', 'mcp-session-id': 's' } }) + }) as unknown as typeof fetch + await new ObscuraClient('http://obscura.test/mcp', 't', ok).render('https://a.example/') + const nav = sent.find((b) => b.params?.name === 'browser_navigate') + expect(nav.params.arguments.waitUntil).toBe('load') + const evaluate = sent.find((b) => b.params?.name === 'browser_evaluate') + expect(evaluate.params.arguments.timeoutMs).toBeLessThanOrEqual(CRAWL_RENDER_TIMEOUT_MS) + expect(evaluate.params.arguments.expression).toContain('setTimeout(resolve, 1500)') + }) + + test('Obscura being down is our trouble: the page is retried, the host never parked', async () => { + const down = (async () => { + throw Object.assign(new TypeError('ECONNREFUSED: Unable to connect.'), { code: 'ECONNREFUSED' }) + }) as unknown as typeof fetch + const client = new ObscuraClient('http://obscura.test/mcp', 't', down) + const { crawler, sites } = setup((url) => (url.endsWith('/robots.txt') ? text('') : page('')), { size: 1, render: (u) => client.render(u) }) + await expect(crawler.renderPage({ url: 'https://js.example/', site: 'js.example', depth: 0, rendered: true })).rejects.toBeInstanceOf(ObscuraError) + expect(await sites.flagOf('js.example')).toBeNull() + }) + + test('a render timeout parks the page', async () => { + const { crawler, frontier } = setup((url) => (url.endsWith('/robots.txt') ? text('') : page('')), { size: 1, render: (u) => Promise.reject(new RenderTimeout(u, 30_000)) }) + expect((await crawler.renderPage({ url: 'https://js.example/', site: 'js.example', depth: 0, rendered: true })).outcome).toBe('page parked: timeout') + expect((await frontier.siteState('js.example', 'https://js.example/')).pageDown).toBe(true) + }) +}) diff --git a/services/crawler/test/redis.test.ts b/services/crawler/test/redis.test.ts index e1117c4..e88c8ba 100644 --- a/services/crawler/test/redis.test.ts +++ b/services/crawler/test/redis.test.ts @@ -37,6 +37,28 @@ describe.skipIf(!REDIS)('on a real Redis', () => { expect(await redis.pttl('pitcrawl:seen:race.example')).toBeGreaterThan(50_000) }) + test('parking: robots, host park and page park in one round trip; failures count in a fixed window', async () => { + await redis.flushdb() + const frontier = new RedisFrontier(redis, 5, 60_000) + expect(await frontier.siteState('p.example', 'https://p.example/a')).toEqual({ robots: null, downMs: 0, pageDown: false }) + await frontier.setRobots('p.example', { rules: [], crawlDelayMs: null, sitemaps: [], ai: [], tdm: [] }) + await frontier.markDown('p.example', 30 * 86_400_000) + await frontier.markPageDown('https://p.example/a', 30 * 86_400_000) + const state = await frontier.siteState('p.example', 'https://p.example/a') + expect(state.robots?.rules).toEqual([]) + expect(state.downMs).toBeGreaterThan(29 * 86_400_000) + expect(state.pageDown).toBe(true) + expect((await frontier.siteState('p.example', 'https://p.example/b')).pageDown).toBe(false) + expect(await frontier.noteFailure('p.example', 60_000)).toBe(1) + expect(await frontier.noteFailure('p.example', 60_000)).toBe(2) + // Fixed window: a later failure does not push the expiry out. + await redis.pexpire('pitcrawl:fails:p.example', 1_000) + await frontier.noteFailure('p.example', 60_000) + expect(await redis.pttl('pitcrawl:fails:p.example')).toBeLessThanOrEqual(1_000) + await frontier.clearDown('p.example') + expect((await frontier.siteState('p.example')).downMs).toBe(0) + }) + test('admitMany: one call per list, same verdicts as one at a time, and the cap is exact under a race', async () => { await redis.flushdb() const frontier = new RedisFrontier(redis, 5, 60_000) @@ -208,6 +230,34 @@ describe.skipIf(!PG)('on a real Postgres', () => { } }) + test('the 30-day flag: columns added once, skipped by the recrawl until retry_after, one recheck, cleared', async () => { + const store = new PgSiteStore(PG!) + const sql = (store as unknown as { sql: { unsafe(q: string): Promise } }).sql + try { + await store.add([{ site: 'warm.example', source: 'x' }], false) + await sql.unsafe('truncate sites; update counts set n = 0') + await store.add([{ site: 'dead.example', source: 'x' }, { site: 'fine.example', source: 'x' }], true) + await sql.unsafe(`update sites set scheduled_at = now() - interval '31 days'`) + await store.flag('dead.example', 'timeout: robots.txt: The operation timed out.', Date.now() + 30 * 86_400_000) + // Flagging a site never recorded records it, started now. + await store.flag('new.example', 'connection: refused', Date.now() + 30 * 86_400_000) + expect((await store.resync()).known).toBe(3) + const flag = await store.flagOf('dead.example') + expect(flag?.status).toBe('down') + expect(flag?.lastError).toBe('timeout: robots.txt: The operation timed out.') + expect(flag!.retryAfter - Date.now()).toBeGreaterThan(29 * 86_400_000) + expect(await store.takeDue(30 * 86_400_000, 10)).toEqual(['fine.example']) + await sql.unsafe(`update sites set retry_after = now() - interval '1 second', scheduled_at = now() - interval '31 days' where site = 'dead.example'`) + expect(await store.takeDue(30 * 86_400_000, 10)).toEqual(['dead.example']) + expect(await store.takeDue(30 * 86_400_000, 10)).toEqual([]) + await store.clearFlag('dead.example') + expect(await store.flagOf('dead.example')).toBeNull() + expect(await store.flagOf('fine.example')).toBeNull() + } finally { + await store.close() + } + }) + test.skipIf(!REDIS)('the Redis frontier moves into Postgres: parked stay parked and in order, the rest scheduled', async () => { const redis = connect(REDIS!) await redis.flushdb() From eda8ee8260ae08185345bd6c22dfa600df11d86c Mon Sep 17 00:00:00 2001 From: Anthony Ettinger Date: Sun, 8 Nov 2026 17:05:23 +0000 Subject: [PATCH 2/2] crawler: mark the bound sql`` parameter for threatcrush (false positive) Co-Authored-By: Claude Opus 5.5 --- services/crawler/src/site-store.ts | 2 ++ 1 file changed, 2 insertions(+) diff --git a/services/crawler/src/site-store.ts b/services/crawler/src/site-store.ts index dcb5d20..4f0e727 100644 --- a/services/crawler/src/site-store.ts +++ b/services/crawler/src/site-store.ts @@ -179,6 +179,8 @@ export class PgSiteStore implements SiteStore { async clearFlag(site: string) { await this.init() + // A Bun tagged template: ${site} is a bound parameter, not text. + // threatcrush-disable-next-line sql-template-interpolation Bun's sql`` binds every interpolation as a parameter await this.sql`update sites set status = null, last_error = null, failed_at = null, retry_after = null where site = ${site} and status is not null` }