Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion packages/adapters/src/buyvps.js
Original file line number Diff line number Diff line change
Expand Up @@ -188,7 +188,7 @@ export const buyvps = defineAdapter({
});
if (html) via = 'obscura';
}
if (!html) html = await http.text(url, { headers: { accept: 'text/html' }, timeoutMs: 30_000 });
if (!html) html = await http.text(url, { headers: { accept: 'text/html' } });
const fetchedAt = new Date();
const plans = parsePlans(html);
const items = plans.map((p) => toItem(p, fetchedAt)).filter(Boolean);
Expand Down
4 changes: 2 additions & 2 deletions packages/adapters/src/openvuln.js
Original file line number Diff line number Diff line change
Expand Up @@ -86,7 +86,7 @@ export async function readIndex(doc, fetchedFrom, http, { limit = 500 } = {}) {
if (!entry?.summary && entry?.url) {
if (!sameHost(entry.url, fetchedFrom)) continue;
try {
record = await http.json(entry.url, { timeoutMs: 20_000 });
record = await http.json(entry.url);
} catch {
continue;
}
Expand Down Expand Up @@ -142,7 +142,7 @@ export const openvuln = defineAdapter({
for (const url of urls) {
if (Date.now() > deadline) break;
try {
const doc = await http.json(url, { timeoutMs: 20_000 });
const doc = await http.json(url);
const { items: got, rejected } = await readIndex(doc, url, http);
if (rejected) notes.push(`${hostOf(url)} rejected (${rejected})`);
else {
Expand Down
2 changes: 1 addition & 1 deletion packages/adapters/src/statuspage.js
Original file line number Diff line number Diff line change
Expand Up @@ -96,7 +96,7 @@ export const statuspage = defineAdapter({
for (const host of hosts) {
if (Date.now() > deadline) break;
try {
const res = await http.json(`https://${host}/api/v2/incidents.json`, { timeoutMs: 15_000 });
const res = await http.json(`https://${host}/api/v2/incidents.json`);
for (const inc of (res.incidents ?? []).slice(0, 50))
items.push(toItem(host, res.page, inc));
} catch (err) {
Expand Down
57 changes: 41 additions & 16 deletions packages/adapters/src/storefront.js
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
import { readFileSync } from 'node:fs';
import { defineAdapter } from '@nichedb/core/adapter';
import { BROWSER_SETTLE_MS, CRAWL_BROWSER_TIMEOUT_MS } from '@nichedb/core/crawl';
import { offer, offerItem } from './hosting.js';
import {
kindOf,
Expand Down Expand Up @@ -173,10 +174,27 @@ export function productItem(product, { platform, domain, provider, providerName,
return item;
}

/** Ask an Obscura MCP server for a page's rendered HTML. */
export async function renderWithObscura(mcpUrl, url, { timeoutMs = 60_000 } = {}) {
const call = async (id, name, args) => {
const res = await fetch(mcpUrl, {
/**
* Ask an Obscura MCP server for a page's rendered HTML.
*
* The fleet crawler rule: 30 s for the whole render (navigate, settle, read),
* waiting for `load` plus a bounded settle, never `networkidle` (a page with a
* chat widget or an analytics beacon never goes idle). Running out of time
* rejects, so the caller records a failure instead of reading half a page.
*/
export async function renderWithObscura(
mcpUrl,
url,
{
timeoutMs = CRAWL_BROWSER_TIMEOUT_MS,
settleMs = BROWSER_SETTLE_MS,
sleep = Bun.sleep,
fetchImpl = fetch,
} = {},
) {
const deadline = AbortSignal.timeout(timeoutMs);
const call = async (id, name, args, signal = deadline) => {
const res = await fetchImpl(mcpUrl, {
method: 'POST',
headers: {
'content-type': 'application/json',
Expand All @@ -188,20 +206,30 @@ export async function renderWithObscura(mcpUrl, url, { timeoutMs = 60_000 } = {}
method: 'tools/call',
params: { name, arguments: args },
}),
signal: AbortSignal.timeout(timeoutMs),
signal,
});
const text = await res.text();
const body = text.includes('data:') ? text.split('data:').pop() : text;
const json = JSON.parse(body.trim());
if (json.error) throw new Error(json.error.message ?? 'obscura error');
return json.result?.content?.find((c) => c.type === 'text')?.text ?? '';
};
await call(1, 'browser_navigate', { url, waitUntil: 'networkidle' });
const html = await call(2, 'browser_evaluate', {
expression: 'document.documentElement.outerHTML',
});
await call(3, 'browser_close', {}).catch(() => {});
return html;
try {
await call(1, 'browser_navigate', { url, waitUntil: 'load', timeout: timeoutMs });
// Late scripts get a moment, bounded and inside the same budget.
await Promise.race([
sleep(Math.min(settleMs, 2_000)),
new Promise((_, reject) =>
deadline.addEventListener('abort', () => reject(deadline.reason), { once: true }),
),
]);
return await call(2, 'browser_evaluate', {
expression: 'document.documentElement.outerHTML',
});
} finally {
// Close the tab whatever happened, on its own short budget.
await call(3, 'browser_close', {}, AbortSignal.timeout(5_000)).catch(() => {});
}
}

/**
Expand All @@ -220,7 +248,7 @@ export async function readHost(
const u = new URL(url);
if (!robots.has(u.origin)) {
requests++;
const txt = await http.request(`${u.origin}/robots.txt`, { timeoutMs: 15_000 }).then(
const txt = await http.request(`${u.origin}/robots.txt`).then(
(r) => (r.ok ? r.text() : ''),
() => '',
);
Expand All @@ -231,10 +259,7 @@ export async function readHost(
if (requests >= maxRequests || Date.now() > deadline) return null;
if (requests > 1) await sleep(1000);
requests++;
const res = await http.request(url, {
headers: { accept: 'text/html,*/*' },
timeoutMs: 15_000,
});
const res = await http.request(url, { headers: { accept: 'text/html,*/*' } });
const html = (await res.text()).slice(0, 600_000);
return {
status: res.status,
Expand Down
1 change: 1 addition & 0 deletions packages/core/package.json
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@
"exports": {
".": "./src/index.js",
"./adapter": "./src/adapter.js",
"./crawl": "./src/crawl.js",
"./http": "./src/http.js",
"./names": "./src/names.js",
"./opensite": "./src/opensite.js",
Expand Down
104 changes: 104 additions & 0 deletions packages/core/src/crawl.js
Original file line number Diff line number Diff line change
@@ -0,0 +1,104 @@
/**
* The fleet crawler rule (2026-11-08): "crawlers should timeout after 10
* seconds for fetch, 30 for browser (js, images, css). We don't want broken
* slow shit. Flag to recheck later in a month."
*
* One place for the numbers, and the pure function that decides whether a
* failed run was the upstream being slow or broken. On its own, with no
* imports, like retry.js: http.js, ingest.js and the adapters all need it.
*/

/** One plain HTTP request to a third-party URL: connect, headers and body. */
export const CRAWL_FETCH_TIMEOUT_MS = 10_000;

/** One headless-browser render (JS, images, CSS): navigation plus settle. */
export const CRAWL_BROWSER_TIMEOUT_MS = 30_000;

/** The bounded settle after `load` in a browser render; never networkidle. */
export const BROWSER_SETTLE_MS = 1_500;

/**
* A bulk dataset download (a dump of gigabytes, `http.download`). Not a page
* or an API call, so not the 10 s rule: an hour for the whole transfer.
*/
export const BULK_DOWNLOAD_TIMEOUT_MS = 60 * 60_000;

/** How long a source whose upstream was slow or broken is left alone. */
export const FLAG_DAYS = 30;
export const FLAG_MS = FLAG_DAYS * 86_400_000;

// Timeouts anywhere between us and the upstream (Bun: TimeoutError; undici: UND_ERR_*).
const SLOW_CODES = new Set([
'TimeoutError',
'ETIMEDOUT',
'ESOCKETTIMEDOUT',
'UND_ERR_CONNECT_TIMEOUT',
'UND_ERR_HEADERS_TIMEOUT',
'UND_ERR_BODY_TIMEOUT',
]);
// The upstream cannot be reached or spoken to: no such host, refused, reset, TLS.
const BROKEN_CODES = new Set([
'ENOTFOUND',
'ECONNREFUSED',
'ECONNRESET',
'EHOSTUNREACH',
'ENETUNREACH',
'EHOSTDOWN',
'EPIPE',
'EPROTO',
'ConnectionRefused',
'ConnectionClosed',
'FailedToOpenSocket',
'UND_ERR_SOCKET',
]);
const TLS_CODE =
/^(ERR_TLS_|ERR_SSL_|CERT_|UNABLE_TO_|DEPTH_ZERO_SELF_SIGNED_CERT|SELF_SIGNED_CERT_IN_CHAIN|HOSTNAME_MISMATCH)/;

function codesOf(err) {
const out = [];
for (let e = err, depth = 0; e && depth < 4; e = e.cause, depth++) {
if (typeof e !== 'object' && typeof e !== 'function') break;
for (const c of [e.code, e.name]) if (typeof c === 'string') out.push(c);
}
return out;
}

/**
* 'slow' (a timeout), 'broken' (DNS, TLS, refused/reset connection, HTTP
* 5xx) or null for anything else: a 4xx, a parse error, a missing key, a bug
* of ours, a database error. Those keep the short retry in retry.js.
*
* EAI_AGAIN is null on purpose: it is our own resolver failing for a moment
* (dev2's Docker DNS has failed every host at once), not the upstream. An
* error can force the answer with `flag: 'slow' | 'broken' | null`.
*/
export function failureClass(err) {
if (!err || (typeof err !== 'object' && typeof err !== 'function')) return null;
if ('flag' in err && (err.flag === 'slow' || err.flag === 'broken' || err.flag === null))
return err.flag;
const codes = codesOf(err);
if (codes.includes('EAI_AGAIN')) return null;
// Our own database failing is never the upstream's fault: Bun SQL errors
// carry ERR_POSTGRES_* codes, and a refused Postgres must not park sources.
if (codes.some((c) => c.startsWith('ERR_POSTGRES') || c === 'PostgresError' || c === 'SQLError'))
return null;
// A system code only says "upstream" on a fetch error: Bun's fetch throws a
// TypeError carrying it, and makeHttp marks what it throws `upstream`.
const fromFetch =
err.upstream === true || err.name === 'TypeError' || err.name === 'TimeoutError';
if (codes.includes('TimeoutError')) return 'slow';
if (fromFetch && codes.some((c) => SLOW_CODES.has(c))) return 'slow';
const status = Number(err.status ?? err.statusCode);
if (Number.isFinite(status) && status >= 500 && status <= 599) return 'broken';
if (fromFetch && codes.some((c) => BROKEN_CODES.has(c) || TLS_CODE.test(c))) return 'broken';
const message = String(err.message ?? '');
// httpError's message, for an error rethrown without its status.
if (/^5\d\d from https?:\/\//.test(message)) return 'broken';
if (/^The operation timed out\.?$/.test(message)) return 'slow';
return null;
}

/** When a source flagged now is looked at again. */
export function flaggedUntil(now = Date.now()) {
return new Date(now + FLAG_MS);
}
41 changes: 37 additions & 4 deletions packages/core/src/http.js
Original file line number Diff line number Diff line change
Expand Up @@ -5,14 +5,32 @@
* police the User-Agent (SEC refuses a blank one outright) and the rate. So this
* always identifies the deployment, always has a timeout, and turns a 429 into
* one polite wait rather than a retry storm.
*
* The timeout is the fleet crawler rule's 10 s for a page or API call
* (CRAWL_FETCH_TIMEOUT_MS); an adapter reading a big document passes its own,
* and `download` keeps an hour for a dump. What it throws is marked
* `upstream` (and an HTTP error carries `status`), so ingest can tell a slow
* or broken upstream, flagged for 30 days, from anything else.
*/

import { mkdir, open, stat, unlink } from 'node:fs/promises';
import { dirname } from 'node:path';
import { BULK_DOWNLOAD_TIMEOUT_MS, CRAWL_FETCH_TIMEOUT_MS } from './crawl.js';
import { requestMusicBrainz, retryDelay } from './musicbrainz-http.js';

export function makeHttp({ userAgent, log = () => {} }) {
async function request(url, { headers = {}, timeoutMs = 30_000, method = 'GET', body } = {}) {
async function request(url, opts = {}) {
try {
return await send(url, opts);
} catch (err) {
throw markUpstream(err);
}
}

async function send(
url,
{ headers = {}, timeoutMs = CRAWL_FETCH_TIMEOUT_MS, method = 'GET', body } = {},
) {
const doFetch = () =>
fetch(url, {
method,
Expand All @@ -39,7 +57,7 @@ export function makeHttp({ userAgent, log = () => {} }) {
*
* The dumps this serves are 1.7 to 11 GB, so the body is streamed chunk by
* chunk onto an open file handle and never held in memory; the timeout covers
* the whole transfer and defaults to an hour, because `request`'s 30 s is a
* the whole transfer and defaults to an hour, because `request`'s 10 s is a
* ceiling on an API call, not on a download. The user agent goes with it:
* Podcast Index answers 403 without one.
*
Expand All @@ -59,7 +77,7 @@ export function makeHttp({ userAgent, log = () => {} }) {
async function download(
url,
filePath,
{ timeoutMs = 60 * 60_000, headers = {}, onProgress } = {},
{ timeoutMs = BULK_DOWNLOAD_TIMEOUT_MS, headers = {}, onProgress } = {},
) {
await mkdir(dirname(filePath), { recursive: true });
const have = (await stat(filePath).catch(() => null))?.size ?? 0;
Expand Down Expand Up @@ -165,7 +183,22 @@ async function httpError(res, url) {
} catch {
// An unreadable body is no worse than the message without it.
}
return new Error(`${res.status} from ${String(url).slice(0, 400)}${detail}`);
return Object.assign(new Error(`${res.status} from ${String(url).slice(0, 400)}${detail}`), {
status: res.status,
upstream: true,
});
}

/** Mark a fetch failure as the upstream's, keeping the error itself (its name, code and cause). */
function markUpstream(err) {
if (err && typeof err === 'object') {
try {
err.upstream = true;
} catch {
// A frozen error stays as it is.
}
}
return err;
}

/** The total out of `Content-Range: bytes 0-99/1000` (or the `bytes` star form of a 416), or null. */
Expand Down
7 changes: 7 additions & 0 deletions packages/core/src/index.js
Original file line number Diff line number Diff line change
@@ -1,5 +1,12 @@
export * from './adapter.js';
export { computeCollectionStats, refreshCollectionStats } from './collection-stats.js';
export {
CRAWL_BROWSER_TIMEOUT_MS,
CRAWL_FETCH_TIMEOUT_MS,
FLAG_DAYS,
failureClass,
flaggedUntil,
} from './crawl.js';
export * from './dump.js';
export { describeEnrichers, enrichPending } from './enrich.js';
export { scanFeeds } from './feedscan.js';
Expand Down
23 changes: 19 additions & 4 deletions packages/core/src/ingest.js
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@ import * as profiles from '@nichedb/db/profiles';
import * as q from '@nichedb/db/queries';
import { retryTransient } from '@nichedb/db/resilience';
import { normaliseItem } from './adapter.js';
import { failureClass, flaggedUntil } from './crawl.js';
import { makeHttp } from './http.js';
import { profileItem } from './profiles.js';
import { retryMinutes } from './retry.js';
Expand Down Expand Up @@ -158,7 +159,11 @@ export async function runSource(
return { ...totals };
} catch (err) {
const message = String(err?.message ?? err).slice(0, 1000);
const nextRunAt = await retryAt(source).catch(() => null);
// The fleet crawler rule: an upstream that timed out, failed DNS/TLS or a
// connection, or answered 5xx is flagged and left alone for 30 days, then
// tried once; anything else takes the short doubling retry.
const flag = failureClass(err);
const nextRunAt = flag ? flaggedUntil() : await retryAt(source).catch(() => null);
/*
* A run that died because Postgres went away must still be closed once it
* is back. Left 'running', the row parks the source: `dueSources` treats it
Expand All @@ -169,11 +174,21 @@ export async function runSource(
* most, before giving up and leaving the row to the reaper as before.
*/
await retryTransient(
() => q.finishRun({ runId, sourceId: source.id, status: 'error', error: message, nextRunAt }),
() =>
q.finishRun({
runId,
sourceId: source.id,
status: 'error',
error: message,
nextRunAt,
flag,
}),
{ label: `${tag} closing the run`, log: l, budgetMs: 3 * 60_000, ...closeRetry },
);
l(`failed: ${message}${nextRunAt ? `; retrying at ${nextRunAt.toISOString()}` : ''}`);
return { error: message };
l(
`failed${flag ? ` (${flag} upstream, flagged)` : ''}: ${message}${nextRunAt ? `; retrying at ${nextRunAt.toISOString()}` : ''}`,
);
return flag ? { error: message, flag } : { error: message };
}
}

Expand Down
Loading
Loading