From 6902bfd4c58aea1f7958043e7b6eedc7b84a2ce8 Mon Sep 17 00:00:00 2001 From: Anthony Ettinger Date: Sun, 8 Nov 2026 16:49:18 +0000 Subject: [PATCH 1/2] crawler: search index in Postgres for many parallel writers (0.48.0) SQLite's single writer capped the local search mirror at ~63 pages/s; a million pages an hour needs ~280/s. CRAWL_INDEX_DATABASE_URL now selects a PgIndexStore (src/pg-index-store.ts) that any number of publish processes write at once; unset, the SQLite IndexStore is kept (tests, laptops). - SearchIndex interface; IndexStore implements it (sync), PgIndexStore (async); every caller awaits. /api/v1/search, /api/v1/corpus, /api/v1/corpus/stats and /stats keep their shapes; /stats reads cached figures with a 1 s cap. - Postgres: pages with a stored weighted tsvector ('simple') + GIN (fastupdate), page_policy, page_cards, policy_site_checks, index_meta, in the role's default schema. One INSERT ... ON CONFLICT per batch over unnest() of array literals; the page counter moves in the same statement. - Search ranks a bounded candidate set with ts_rank under a statement timeout; the product path keeps its newest-N style cap (8,000 matches). - Corpus cursors stop at the newest id at least 60 s old, since concurrent writers commit out of id order. - Trim keeps the live pages (count x bytes a page) inside the budget, since Postgres reuses freed space rather than shrinking files. - pit-crawler migrate-index --from-sqlite : streams pages, verdicts, cards and site checks into Postgres with ids and read times kept, resumable from a cursor kept in Postgres. - The page archive names each process's file pages---.jsonl.gz so several publishers never share one; ship-archive.sh's glob still matches. Co-Authored-By: Claude Opus 5.5 --- package.json | 2 +- services/crawler/README.md | 14 + services/crawler/src/archive.ts | 36 +- services/crawler/src/cli.ts | 97 ++- services/crawler/src/config.ts | 6 + services/crawler/src/crawler.ts | 10 +- services/crawler/src/import-nichedb.ts | 18 +- services/crawler/src/index-store.ts | 152 +++- services/crawler/src/migrate-index.ts | 94 +++ services/crawler/src/pg-index-store.ts | 815 +++++++++++++++++++ services/crawler/src/queues.ts | 18 +- services/crawler/test/archive.test.ts | 64 +- services/crawler/test/pg-index-store.test.ts | 350 ++++++++ 13 files changed, 1581 insertions(+), 95 deletions(-) create mode 100644 services/crawler/src/migrate-index.ts create mode 100644 services/crawler/src/pg-index-store.ts create mode 100644 services/crawler/test/pg-index-store.test.ts diff --git a/package.json b/package.json index f492b69..52be3f5 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "pitworld", - "version": "0.47.1", + "version": "0.48.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/README.md b/services/crawler/README.md index 33ea9a0..977e4b7 100644 --- a/services/crawler/README.md +++ b/services/crawler/README.md @@ -43,6 +43,19 @@ truncated the append-only file of every other Redis on the box (2026-11-04). So: An index made before this keeps `auto_vacuum = NONE`: deleting rows still bounds it, but the file keeps its high-water mark, so rebuild it rather than expecting it to shrink. +### In Postgres, for many writers + +SQLite takes one writer, about 63 pages/s here. With **`CRAWL_INDEX_DATABASE_URL`** set the +mirror lives in Postgres instead (`src/pg-index-store.ts`) and any number of publishing +processes write it at once: one `INSERT ... ON CONFLICT (url)` per batch, the page counter +moved in the same statement. Full text is a stored, weighted `tsvector` (`'simple'` config, +pages are in every language) under a GIN index. A search ranks the first 3,000 matches with +`ts_rank` (a product search looks through 8,000) under a ~1 s statement timeout. Postgres files +do not shrink, so the budget is kept on the live pages (pages x bytes a page) and freed space +is reused. Move an existing file in with `pit-crawler migrate-index --from-sqlite ` +(resumable; ids and read times kept). Each publishing process writes its own archive file, +`pages---.jsonl.gz`. + `/stats` reports `index_bytes` against `index_max_bytes`, so /status shows the headroom. Eviction goes by **`fetched_at`**, not by id: a bulk import arrives newest-first, so id order and crawl order are not the same thing. @@ -123,6 +136,7 @@ curl 127.0.0.1:8807/api/v1/corpus/stats | `CRAWL_HEALTH_PORT` | 8797 | `/healthz` with queue depths and counters | | `CRAWL_INDEX_MAX_BYTES` | 64424509440 (60 GB) | the mirror's byte budget; over it the oldest pages go. 0 turns it off, which only a box with its own disk for the index should do | | `CRAWL_INDEX_TRIM_MS` | 300000 | how often the publish role checks the budget | +| `CRAWL_INDEX_DATABASE_URL` | (none) | Postgres for the search mirror instead of the SQLite file at `CRAWL_INDEX_PATH`; tables in the role's default schema. Many publish processes may write it | ## Deploy (dev2) diff --git a/services/crawler/src/archive.ts b/services/crawler/src/archive.ts index 04d684c..93b2528 100644 --- a/services/crawler/src/archive.ts +++ b/services/crawler/src/archive.ts @@ -1,6 +1,6 @@ /** * The page archive: every record the publish role takes from the outbox, full text and - * all, appended to one gzip file per UTC hour (`pages-20261108T15.jsonl.gz`). An index is + * all, appended to one gzip file per UTC hour and process (`pages-20261108T15--.jsonl.gz`). An index is * a cache that forgets its oldest pages; the archive is what a bigger index or another * search engine is rebuilt from. Since 2026-11-08 nothing else keeps pages (they stopped * going to nichedb), so pitbot's publish host writes here and a cron ships finished hours @@ -10,34 +10,58 @@ * stream), so a crash loses at most the batch being written. The current hour is * `.partial`; it is renamed to `` once the hour has passed, and only * renamed files are shipped. + * + * Several publishing processes may share the directory (the index takes many writers since + * it moved to Postgres), so each process writes its own file: the name carries the host and + * pid (`pages-20261108T15--.jsonl.gz`). A process seals its own earlier hours at + * once; another's only once that hour is two behind, by when its writer has long moved on (or + * died, which is how a restart's leftover gets sealed). The `pages-` prefix and `.jsonl.gz` + * suffix stay, so deploy/crawler-remote/ship-archive.sh still matches `pages-*.jsonl.gz`. */ import { appendFileSync, mkdirSync, readdirSync, renameSync } from 'node:fs' +import { hostname } from 'node:os' import { join } from 'node:path' import { gzipSync } from 'node:zlib' -export const hourName = (at: Date) => `pages-${at.toISOString().slice(0, 13).replace(/[-:]/g, '')}.jsonl.gz` +const hourKey = (at: Date) => at.toISOString().slice(0, 13).replace(/[-:]/g, '') + +/** `pages-[-].jsonl.gz`; the tag tells one writing process from another. */ +export const hourName = (at: Date, tag = '') => `pages-${hourKey(at)}${tag ? `-${tag}` : ''}.jsonl.gz` + +/** This process, as a file-name-safe tag: host and pid. */ +export const processTag = (host = hostname(), pid = process.pid) => `${host.replace(/[^A-Za-z0-9.]+/g, '_').slice(0, 64)}-${pid}` export class PageArchive { constructor( readonly dir: string, private now: () => Date = () => new Date(), + readonly tag: string = processTag(), ) { mkdirSync(dir, { recursive: true }) } /** Append records (one JSON per line); seals any earlier hour first. */ write(lines: string[]): void { - const name = hourName(this.now()) + const name = hourName(this.now(), this.tag) this.seal(name) if (!lines.length) return appendFileSync(join(this.dir, `${name}.partial`), gzipSync(`${lines.join('\n')}\n`, { level: 6 })) } - /** Rename every .partial that is not the current hour's: its hour is over. */ - seal(current = hourName(this.now())): string[] { + /** + * Rename every .partial whose hour is over: this process's own as soon as the hour turns, + * any other process's once it is two hours old (it may still be finishing its last batch). + */ + seal(current = hourName(this.now(), this.tag)): string[] { const sealed: string[] = [] + const now = this.now() + const previous = hourKey(new Date(now.getTime() - 3_600_000)) + const suffix = this.tag ? `-${this.tag}.jsonl.gz.partial` : '.jsonl.gz.partial' for (const f of readdirSync(this.dir)) { - if (!f.endsWith('.partial') || f === `${current}.partial`) continue + if (!f.startsWith('pages-') || !f.endsWith('.partial') || f === `${current}.partial`) continue + const hour = f.slice('pages-'.length, 'pages-'.length + 11) + const ours = f === `pages-${hour}${suffix}` + if (!ours && hour >= previous) continue const done = f.slice(0, -'.partial'.length) renameSync(join(this.dir, f), join(this.dir, done)) sealed.push(done) diff --git a/services/crawler/src/cli.ts b/services/crawler/src/cli.ts index 55d5b31..4035bcd 100644 --- a/services/crawler/src/cli.ts +++ b/services/crawler/src/cli.ts @@ -12,6 +12,7 @@ * pit-crawler seed --file one site per line * pit-crawler stats queue depths and counters, as JSON * pit-crawler read [--render] read one page now and print its record (no Redis) + * pit-crawler migrate-index --from-sqlite copy a SQLite index into CRAWL_INDEX_DATABASE_URL (resumable) */ import { readFileSync } from 'node:fs' import { searchMcps } from './mcp.ts' @@ -22,7 +23,8 @@ import { ObscuraClient } from './obscura.ts' import { createRuntime, migrateSitesFromRedis, enqueueSubmissions, report, snapshot, startWorkers, type Runtime } from './queues.ts' import { OPML_SOURCES, opmlText, sitesFromNichedb, sitesFromOpml, sitesFromRing, sitesFromText } from './seed.ts' import { importFromNichedb } from './import-nichedb.ts' -import { IndexStore } from './index-store.ts' +import { openIndex, PgIndexStore } from './pg-index-store.ts' +import { migrateIndexFromSqlite } from './migrate-index.ts' import { normalizeUrl, resolvesPublic, siteOf } from './url.ts' import { trainingFor } from './crawler.ts' import { pathOf } from './robots.ts' @@ -65,14 +67,14 @@ async function worker() { const q = (url.searchParams.get('q') ?? '').slice(0, 512) const limit = Number(url.searchParams.get('limit')) || 10 const kind = url.searchParams.get('kind')?.slice(0, 40) || undefined - const items = q ? runtime.index.search(q, limit, kind) : [] + const items = q ? await runtime.index.search(q, limit, kind) : [] if (kind === 'product' && items.length) { const urls = items.map((i) => i.url) - const known = runtime.index.cards(urls) + const known = await runtime.index.cards(urls) const unread = urls.filter((u) => !known.has(u)) // A shopper waits a moment for pictures, never long: the reads finish and are kept either way. if (unread.length) await Promise.race([runtime.crawler.fillCards(unread), new Promise((done) => setTimeout(done, 2_500))]) - const cards = unread.length ? runtime.index.cards(urls) : known + const cards = unread.length ? await runtime.index.cards(urls) : known return Response.json({ q, count: items.length, items: items.map((i) => ({ ...i, ...cards.get(i.url) })) }) } return Response.json({ q, count: items.length, items }) @@ -82,7 +84,7 @@ async function worker() { const q = (url.searchParams.get('q') ?? '').slice(0, 200) const limit = Math.min(50, Number(url.searchParams.get('limit')) || 20) if (!q) return Response.json({ items: [], unavailable: false }) - return Response.json(await runtime.crawler.searchSocial(q, limit, runtime.index.search(q, limit, 'post'))) + return Response.json(await runtime.crawler.searchSocial(q, limit, await runtime.index.search(q, limit, 'post'))) } if (path === '/api/v1/mcp') { // Proved servers, from Redis: no SQLite in this path. @@ -104,7 +106,7 @@ async function worker() { if (!Number.isFinite(cursor) || cursor < 0) return Response.json({ error: 'cursor must be a cursor this API returned' }, { status: 400 }) const since = p.get('since') ?? '' if (since && Number.isNaN(Date.parse(since))) return Response.json({ error: 'since must be an ISO date, e.g. 2026-11-01' }, { status: 400 }) - const page = runtime.index.corpus({ + const page = await runtime.index.corpus({ q: (p.get('q') ?? '').slice(0, 512), site: (p.get('site') ?? '').slice(0, 253), kind: (p.get('kind') ?? '').slice(0, 40), @@ -117,7 +119,7 @@ async function worker() { return Response.json(page) } if (path === '/api/v1/corpus/stats') { - return Response.json(runtime.index.corpusStats()) + return Response.json(await runtime.index.corpusStats()) } if (path !== '/healthz' && path !== '/stats') return new Response('not found', { status: 404 }) try { @@ -133,7 +135,7 @@ async function worker() { // Pages indexed before training verdicts were recorded: give each a row, then judge each site's // pages on its robots.txt and tdmrep.json (re-crawls add the page's own headers and meta). - const backfill = config.roles.includes('publish') ? startPolicyBackfill(runtime, log, config.indexPath) : null + const backfill = config.roles.includes('publish') ? startPolicyBackfill(runtime, log, config) : null // The mirror inside its byte budget. The publish role is the only writer, so it is the // only one that may delete; a page dropped here is still on nichedb, and comes back on @@ -171,7 +173,9 @@ function startIndexTrim(runtime: Runtime, log: (line: string) => void, config: C 'so deleting rows cannot give the space back. Rebuild it (the records are on nichedb) or give it its own disk.', ) } - if (result.deleted) log(`index trim: dropped ${result.deleted} pages, ${gb(result.bytesBefore)} -> ${gb(result.bytesAfter)} (budget ${gb(config.indexMaxBytes)})${result.done ? '' : ', more next sweep'}`) + // Postgres keeps its files and reuses the space, so it reports what the live pages take. + const after = result.liveBytes !== undefined ? `${gb(result.liveBytes)} live in ${gb(result.bytesAfter)} of files` : gb(result.bytesAfter) + if (result.deleted) log(`index trim: dropped ${result.deleted} pages, ${gb(result.bytesBefore)} -> ${after} (budget ${gb(config.indexMaxBytes)})${result.done ? '' : ', more next sweep'}`) } catch (error) { log(`index trim: ${(error as Error).message}`) } finally { @@ -188,16 +192,37 @@ function startIndexTrim(runtime: Runtime, log: (line: string) => void, config: C } } -function startPolicyBackfill(runtime: Runtime, log: (line: string) => void, indexPath: string) { +function startPolicyBackfill(runtime: Runtime, log: (line: string) => void, config: Config) { let stopped = false let mapping = true - const mapper = new Worker(new URL('./policy-map-worker.ts', import.meta.url).href, { env: { ...process.env, CRAWL_INDEX_PATH: indexPath } } as WorkerOptions) - mapper.onmessage = (event: MessageEvent) => { - const data = event.data as { done?: number; error?: string } - if (data.error) log(`policy map: ${data.error}`) - if (data.done !== undefined) { + let mapper: Worker | null = null + if (config.indexDatabaseUrl) { + // Postgres does not block this thread, so the mapping runs here, a batch at a time. + void (async () => { + let total = 0 + while (!stopped) { + try { + const n = await runtime.index.mapLegacy(2_000) + total += n + if (!n) break + } catch (error) { + log(`policy map: ${(error as Error).message}`) + await Bun.sleep(5_000) + } + await Bun.sleep(50) + } mapping = false - log(`policy map: ${data.done} legacy pages given a row`) + if (total) log(`policy map: ${total} legacy pages given a row`) + })() + } else { + mapper = new Worker(new URL('./policy-map-worker.ts', import.meta.url).href, { env: { ...process.env, CRAWL_INDEX_PATH: config.indexPath } } as WorkerOptions) + mapper.onmessage = (event: MessageEvent) => { + const data = event.data as { done?: number; error?: string } + if (data.error) log(`policy map: ${data.error}`) + if (data.done !== undefined) { + mapping = false + log(`policy map: ${data.done} legacy pages given a row`) + } } } let timer: ReturnType | null = null @@ -207,14 +232,14 @@ function startPolicyBackfill(runtime: Runtime, log: (line: string) => void, inde const run = async () => { try { // Sites are judged as their pages are mapped; the mapper runs on its own thread. - const sites = runtime.index.unknownSites(8) + const sites = await runtime.index.unknownSites(8) if (!sites.length) return next(mapping ? 30_000 : 10 * 60_000) await Promise.all( sites.map(async (site) => { const robots = await runtime.crawler.robotsFor(site).catch(() => 'unavailable' as const) - const pages = runtime.index.unknownPagesOf(site) + const pages = await runtime.index.unknownPagesOf(site) if (robots === 'unavailable' || robots.read === 'error') return runtime.index.setPolicies(site, []) - runtime.index.setPolicies( + await runtime.index.setPolicies( site, pages.map((p) => { const verdict = trainingFor({ robots, path: pathOf(p.url), noai: false, tdmMeta: false }) @@ -234,7 +259,7 @@ function startPolicyBackfill(runtime: Runtime, log: (line: string) => void, inde stop() { stopped = true if (timer) clearTimeout(timer) - mapper.terminate() + mapper?.terminate() }, } } @@ -287,8 +312,8 @@ async function importCmd(args: string[]) { const limit = Number(flag(args, '--limit')) || undefined const maxGb = Number(flag(args, '--max-gb')) || undefined const restart = args.includes('--restart') - const index = new IndexStore(config.indexPath, config.indexMaxBytes) - if (restart) index.setImportCursor(null) + const index = openIndex(config) + if (restart) await index.setImportCursor(null) const gb = (bytes: number) => `${(bytes / 1024 ** 3).toFixed(2)} GB` let lastSaid = 0 try { @@ -307,9 +332,9 @@ async function importCmd(args: string[]) { log(`import: the index is busy at cursor ${cursor} (${message}); waiting 5 s and reading that page again`) }, }) - console.log(JSON.stringify({ ...result, bytes: undefined, size: gb(result.bytes), pages: index.count() })) + console.log(JSON.stringify({ ...result, bytes: undefined, size: gb(result.bytes), pages: await index.count() })) } finally { - index.close() + await index.close() } } @@ -354,12 +379,32 @@ async function migrateSites(args: string[]) { } } +/** + * One-off: every page of an old SQLite index into Postgres (CRAWL_INDEX_DATABASE_URL), ids and + * read times kept so trim stays oldest-first. Resumable: the cursor is kept in Postgres, so a + * stopped run carries on where it left off; --restart begins again (pages already there stay). + */ +async function migrateIndex(args: string[]) { + const config = configFromEnv() + if (!config.indexDatabaseUrl) throw new Error('CRAWL_INDEX_DATABASE_URL is not set') + const from = flag(args, '--from-sqlite') + if (!from) throw new Error('migrate-index --from-sqlite [--batch n] [--restart]') + const batch = Number(flag(args, '--batch')) || undefined + const index = new PgIndexStore(config.indexDatabaseUrl, config.indexMaxBytes) + try { + const result = await migrateIndexFromSqlite(from, index, { ...(batch ? { batch } : {}), restart: args.includes('--restart'), log }) + console.log(JSON.stringify(result)) + } finally { + await index.close() + } +} + const [command, ...rest] = process.argv.slice(2) -const commands: Record Promise> = { worker, seed, social, import: importCmd, stats: () => stats(), read, 'migrate-sites': migrateSites } +const commands: Record Promise> = { worker, seed, social, import: importCmd, stats: () => stats(), read, 'migrate-sites': migrateSites, 'migrate-index': migrateIndex } const run = command ? commands[command] : undefined if (!run) { console.log( - 'usage: pit-crawler worker | seed [--from nichedb|ring] [--file path] [--limit n] | import [--from nichedb] [--limit n] [--max-gb n] [--restart] | stats | read [--render] | migrate-sites [--drop]', + 'usage: pit-crawler worker | seed [--from nichedb|ring] [--file path] [--limit n] | import [--from nichedb] [--limit n] [--max-gb n] [--restart] | stats | read [--render] | migrate-sites [--drop] | migrate-index --from-sqlite [--batch n] [--restart]', ) process.exit(command ? 1 : 0) } diff --git a/services/crawler/src/config.ts b/services/crawler/src/config.ts index 491a1fb..10c8d4b 100644 --- a/services/crawler/src/config.ts +++ b/services/crawler/src/config.ts @@ -61,6 +61,11 @@ export type Config = { healthHost: string /** The local search mirror (SQLite FTS5) the publish role writes and /api/v1/search reads. */ indexPath: string + /** + * Postgres for the search mirror instead of the SQLite file (src/pg-index-store.ts): any + * number of publishing processes write it at once. Unset: SQLite at indexPath. + */ + indexDatabaseUrl: string /** * Bytes the mirror may use. Over it, the publish role drops its oldest pages (they * stay on nichedb, and come back on a recrawl). Unbounded it reached 207 GB in 2.7 @@ -115,6 +120,7 @@ export function configFromEnv(env: Record = process. // Loopback on a host; 0.0.0.0 inside a container, whose port is published on the host's loopback. healthHost: env.CRAWL_HEALTH_HOST ?? '127.0.0.1', indexPath: env.CRAWL_INDEX_PATH ?? './data/index.db', + indexDatabaseUrl: env.CRAWL_INDEX_DATABASE_URL ?? '', indexMaxBytes: num(env.CRAWL_INDEX_MAX_BYTES, 60 * 1024 ** 3, 0), indexTrimMs: num(env.CRAWL_INDEX_TRIM_MS, 300_000, 1_000), userAgent: env.CRAWL_USER_AGENT ?? USER_AGENT, diff --git a/services/crawler/src/crawler.ts b/services/crawler/src/crawler.ts index 73a7580..c7b11df 100644 --- a/services/crawler/src/crawler.ts +++ b/services/crawler/src/crawler.ts @@ -74,7 +74,7 @@ export type Deps = { /** Offer a submitted feed to rssamplifier.com too (best effort). */ forwardFeed?: (feedUrl: string) => Promise /** The local search mirror; every published batch lands here first. */ - index?: { add(entries: string[]): number; setCard?(url: string, image: string | null, price: string | null): void } | null + index?: { add(entries: string[]): number | Promise; setCard?(url: string, image: string | null, price: string | null): void | Promise } | null /** Every published record, full text included, kept for good (src/archive.ts). */ archive?: { write(lines: string[]): void } | null } @@ -302,7 +302,11 @@ export function createCrawler(deps: Deps) { } catch (error) { log(`card ${url}: ${(error as Error).message}`) } - deps.index?.setCard?.(url, image, price) + try { + await deps.index?.setCard?.(url, image, price) + } catch (error) { + log(`card ${url}: ${(error as Error).message}`) + } }), ) } @@ -483,7 +487,7 @@ export function createCrawler(deps: Deps) { await frontier.bump('archived', raw.length) } if (deps.index) { - const kept = deps.index.add(raw) + const kept = await deps.index.add(raw) if (kept) await frontier.bump('indexed_local', kept) } if (!toNichedb) continue diff --git a/services/crawler/src/import-nichedb.ts b/services/crawler/src/import-nichedb.ts index dea23f3..b7c77f3 100644 --- a/services/crawler/src/import-nichedb.ts +++ b/services/crawler/src/import-nichedb.ts @@ -15,7 +15,7 @@ * of /api/corpus until the policy backfill judges their site: an opt-out is respected * by staying closed, which is the right way round to be wrong. */ -import type { IndexedRecord, IndexStore } from './index-store.ts' +import type { IndexedRecord, SearchIndex } from './index-store.ts' type Fetch = typeof fetch @@ -81,7 +81,7 @@ export function recordOf(item: NichedbItem): IndexedRecord | null { * returns (and keeps in the index) lets a later run carry on rather than restart. */ export async function importFromNichedb( - index: IndexStore, + index: SearchIndex, options: { nichedbUrl: string key?: string @@ -106,8 +106,8 @@ export async function importFromNichedb( const ceiling = options.maxBytes ?? index.maxBytes ?? 0 const pause = Math.max(0, Math.trunc(options.pauseMs ?? 50)) const base = options.nichedbUrl.replace(/\/+$/, '') - let cursor = options.cursor ?? Number(index.importCursor() ?? 0) - const progress: ImportProgress = { read: 0, kept: 0, skipped: 0, locked: 0, cursor: cursor || null, bytes: index.sizeBytes(0) } + let cursor = options.cursor ?? Number((await index.importCursor()) ?? 0) + const progress: ImportProgress = { read: 0, kept: 0, skipped: 0, locked: 0, cursor: cursor || null, bytes: await index.sizeBytes(0) } while (progress.read < limit) { const url = `${base}/api/v1/items?collection=sites&limit=${pageSize}${cursor ? `&before=${cursor}` : ''}` @@ -125,7 +125,7 @@ export async function importFromNichedb( const items = body.items ?? [] if (!items.length) { progress.cursor = null - index.setImportCursor(null) + await index.setImportCursor(null) break } const batch: IndexedRecord[] = [] @@ -140,7 +140,7 @@ export async function importFromNichedb( // write lock even after IndexStore's retries must not end a run of millions: the // cursor only moves on a batch that landed, so the page is simply read again. try { - progress.kept += index.add(batch) + progress.kept += await index.add(batch) } catch (error) { const message = String((error as Error)?.message ?? error) if (!/database is locked|SQLITE_BUSY|database table is locked/i.test(message)) throw error @@ -153,13 +153,13 @@ export async function importFromNichedb( } cursor = items[items.length - 1]!.id progress.cursor = cursor - index.setImportCursor(cursor) - progress.bytes = index.sizeBytes(0) + await index.setImportCursor(cursor) + progress.bytes = await index.sizeBytes(0) options.onProgress?.({ ...progress }) if (ceiling && progress.bytes >= ceiling) break if (items.length < pageSize) { progress.cursor = null - index.setImportCursor(null) + await index.setImportCursor(null) break } if (pause) await new Promise((resolve) => setTimeout(resolve, pause)) diff --git a/services/crawler/src/index-store.ts b/services/crawler/src/index-store.ts index d9d3307..2e9d198 100644 --- a/services/crawler/src/index-store.ts +++ b/services/crawler/src/index-store.ts @@ -106,9 +106,102 @@ export type CorpusPage = { export type PolicyRow = { id: number; url: string } -const TOTAL_CAP = 10_000 +export const TOTAL_CAP = 10_000 + +export type TrimOptions = { batch?: number; maxBatches?: number; keepMin?: number; sleep?: (ms: number) => Promise } +export type TrimResult = { + deleted: number + bytesBefore: number + bytesAfter: number + done: boolean + reclaims: boolean + /** Postgres only: what the live pages are estimated to take (its files do not shrink, freed space is reused). */ + liveBytes?: number +} + +type Awaitable = T | Promise -export class IndexStore { +/** + * The local search mirror, whichever store holds it: SQLite (IndexStore, synchronous, one + * writer) or Postgres (PgIndexStore, many writers). Callers await every method, so either fits. + */ +export interface SearchIndex { + /** Bytes the index may use before trim() drops its oldest pages. 0: no budget. */ + readonly maxBytes: number + readonly reclaims: boolean + lastAddAt: string | null + add(entries: Array): Awaitable + sizeBytes(maxAgeMs?: number): Awaitable + trim(options?: TrimOptions): Promise + corpus(query: CorpusQuery): Awaitable + corpusStats(now?: number): Awaitable> + mapLegacy(n?: number): Awaitable + unknownSites(n?: number, now?: number): Awaitable + unknownPagesOf(site: string): Awaitable + setPolicies(site: string, verdicts: Array<{ id: number; allowed: boolean; reason: string }>, now?: number): Awaitable + setCard(url: string, image: string | null, price: string | null): Awaitable + cards(urls: string[]): Awaitable> + search(query: string, limit?: number, kind?: string): Awaitable + count(): Awaitable + importCursor(): Awaitable + setImportCursor(cursor: number | null): Awaitable + tables(): Awaitable + highWaterMark(): Awaitable + close(): Awaitable +} + +/** What a page record becomes in the index: shared by both stores so they keep the same thing. */ +export type PageRow = { + url: string + site: string + kind: string + title: string + description: string + keywords: string + body: string + language: string | null + fetched: string + train: number | null + reason: string | null + source: 'crawl' | 'legacy' + card: { image: string | null; price: string | null } | null +} + +/** An outbox line (or object) as the row both stores write. Null when it cannot be keyed. */ +export function pageRowOf(raw: IndexedRecord | string): PageRow | null { + let entry: IndexedRecord + try { + entry = typeof raw === 'string' ? (JSON.parse(raw) as IndexedRecord) : raw + } catch { + return null + } + const r = entry?.record + if (!r?.canonical) return null + let site = '' + try { + site = new URL(r.canonical).hostname.replace(/^www\./, '') + } catch { + return null + } + const training = entry.enrichment?.training + return { + url: r.canonical, + site, + kind: r.kind ?? 'page', + title: r.title ?? site, + description: r.description ?? '', + keywords: (entry.enrichment?.keywords ?? []).join(' '), + body: entry.enrichment?.text || entry.enrichment?.excerpt || '', + language: entry.enrichment?.language ?? r.language ?? null, + fetched: r.fetched_at ?? new Date().toISOString(), + train: training ? (training.allowed ? 1 : 0) : null, + reason: training ? String(training.reason ?? '').slice(0, 120) : null, + source: training ? 'crawl' : 'legacy', + card: r.kind === 'product' ? { image: entry.record.image?.url ?? entry.enrichment?.images?.[0]?.url ?? null, price: entry.enrichment?.price ?? null } : null, + } +} + +export class IndexStore implements SearchIndex { private db: Database private statsCache: { at: number; value: Record } | null = null /** Bytes the file may use before trim() drops its oldest pages. 0: no budget. */ @@ -264,45 +357,30 @@ export class IndexStore { this.write(() => this.db.transaction(() => { for (const raw of entries) { - let entry: IndexedRecord - try { - entry = typeof raw === 'string' ? (JSON.parse(raw) as IndexedRecord) : raw - } catch { - continue - } - const r = entry?.record - if (!r?.canonical) continue - let site = '' - try { - site = new URL(r.canonical).hostname.replace(/^www\./, '') - } catch { - continue - } - if (r.kind === 'product') this.setCard(r.canonical, entry.record.image?.url ?? entry.enrichment?.images?.[0]?.url ?? null, entry.enrichment?.price ?? null) - const language = entry.enrichment?.language ?? r.language ?? null - const fetched = r.fetched_at ?? new Date().toISOString() + const page = pageRowOf(raw) + if (!page) continue + if (page.card) this.setCard(page.url, page.card.image, page.card.price) const row = upsert.get({ - $url: r.canonical, - $site: site, - $kind: r.kind ?? 'page', - $title: r.title ?? site, - $description: r.description ?? '', - $keywords: (entry.enrichment?.keywords ?? []).join(' '), - $body: entry.enrichment?.text || entry.enrichment?.excerpt || '', - $language: language, - $fetched: fetched, + $url: page.url, + $site: page.site, + $kind: page.kind, + $title: page.title, + $description: page.description, + $keywords: page.keywords, + $body: page.body, + $language: page.language, + $fetched: page.fetched, }) as { id: number } | null - const training = entry.enrichment?.training if (row) policy.run({ $id: row.id, - $site: site, - $kind: r.kind ?? 'page', - $language: language, - $fetched: fetched, - $train: training ? (training.allowed ? 1 : 0) : null, - $reason: training ? String(training.reason ?? '').slice(0, 120) : null, - $source: training ? 'crawl' : 'legacy', + $site: page.site, + $kind: page.kind, + $language: page.language, + $fetched: page.fetched, + $train: page.train, + $reason: page.reason, + $source: page.source, }) // A new row always takes an id above every id ever used, so this tells an // insert from an upsert without asking the database to count anything. @@ -351,7 +429,7 @@ export class IndexStore { * crawl, false the moment a bulk import from nichedb inserts newest-first, which * would have made the newest records the first to be evicted. */ - async trim(options: { batch?: number; maxBatches?: number; keepMin?: number; sleep?: (ms: number) => Promise } = {}): Promise<{ deleted: number; bytesBefore: number; bytesAfter: number; done: boolean; reclaims: boolean }> { + async trim(options: TrimOptions = {}): Promise { const batch = Math.max(1, Math.trunc(options.batch ?? 2_000)) const maxBatches = Math.max(1, Math.trunc(options.maxBatches ?? 50)) // A budget too small for the index's own overhead must leave a working mirror diff --git a/services/crawler/src/migrate-index.ts b/services/crawler/src/migrate-index.ts new file mode 100644 index 0000000..b3d89ad --- /dev/null +++ b/services/crawler/src/migrate-index.ts @@ -0,0 +1,94 @@ +/** + * The SQLite search mirror into Postgres, once (`pit-crawler migrate-index --from-sqlite `). + * + * The file on dev2 holds ~10M pages in 64 GB, so this streams: keyset pages of `batch` rows by + * id, each written in one statement that also moves the cursor kept in Postgres, so a stopped + * run carries on from the last batch that landed. Ids and read times travel with the pages, so + * trim() still drops the oldest first and corpus cursors keep their meaning; the id sequence is + * moved past the highest id brought over, so pages the crawl writes meanwhile never collide. + * A page the crawl has already written to Postgres wins over its older SQLite copy. + */ +import { Database } from 'bun:sqlite' +import type { MigratedPage, PgIndexStore } from './pg-index-store.ts' + +export type MigrateIndexResult = { read: number; inserted: number; cards: number; siteChecks: number; cursor: number; pages: number } + +const PAGES_CURSOR = 'migrate_cursor' +const CARDS_CURSOR = 'migrate_cards_cursor' + +export async function migrateIndexFromSqlite(path: string, index: PgIndexStore, options: { batch?: number; restart?: boolean; log?: (line: string) => void } = {}): Promise { + const batch = Math.min(Math.max(1, Math.trunc(options.batch ?? 500)), 5_000) + const log = options.log ?? (() => {}) + // Read-only: the crawler may still be writing this file, and nothing here should change it. + const db = new Database(path, { readonly: true }) + db.exec('PRAGMA busy_timeout = 30000;') + try { + if (options.restart) { + await index.setCursor(PAGES_CURSOR, null) + await index.setCursor(CARDS_CURSOR, null) + } + const hasPolicy = Boolean(db.query("SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = 'page_policy'").get()) + const pick = db.query( + hasPolicy + ? `SELECT p.id, p.url, p.site, p.kind, p.title, p.description, p.keywords, p.body, p.language, p.fetched_at, pp.train, pp.reason, pp.source + FROM pages p LEFT JOIN page_policy pp ON pp.id = p.id WHERE p.id > ? ORDER BY p.id LIMIT ?` + : `SELECT id, url, site, kind, title, description, keywords, body, language, fetched_at, NULL AS train, NULL AS reason, NULL AS source + FROM pages WHERE id > ? ORDER BY id LIMIT ?`, + ) + // Before the first batch: the crawl may be writing Postgres already, and its new pages + // must take ids above every id coming over. + const top = (db.query('SELECT coalesce(max(id), 0) AS n FROM pages').get() as { n: number }).n + await index.reserveIds(top) + let cursor = (await index.cursor(PAGES_CURSOR)) ?? 0 + const result: MigrateIndexResult = { read: 0, inserted: 0, cards: 0, siteChecks: 0, cursor, pages: 0 } + if (cursor) log(`migrate-index: carrying on after id ${cursor}`) + let lastSaid = 0 + while (true) { + const rows = pick.all(cursor, batch) as MigratedPage[] + if (!rows.length) break + const last = rows[rows.length - 1]!.id + result.inserted += await index.load(rows, { key: PAGES_CURSOR, value: last }) + result.read += rows.length + cursor = last + result.cursor = cursor + if (result.read - lastSaid >= 50_000) { + lastSaid = result.read + log(`migrate-index: ${result.read} read, ${result.inserted} new, at id ${cursor}`) + } + } + + const hasCards = Boolean(db.query("SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = 'page_cards'").get()) + if (hasCards) { + let rowid = (await index.cursor(CARDS_CURSOR)) ?? 0 + const cards = db.query('SELECT rowid AS rowid, url, image, price FROM page_cards WHERE rowid > ? ORDER BY rowid LIMIT ?') + while (true) { + const rows = cards.all(rowid, 2_000) as Array<{ rowid: number; url: string; image: string | null; price: string | null }> + if (!rows.length) break + await index.loadCards(rows) + rowid = rows[rows.length - 1]!.rowid + await index.setCursor(CARDS_CURSOR, rowid) + result.cards += rows.length + } + } + + const hasChecks = Boolean(db.query("SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = 'policy_site_checks'").get()) + if (hasChecks) { + const checks = db.query('SELECT site, checked_at FROM policy_site_checks WHERE site > ? ORDER BY site LIMIT 5000') + let after = '' + while (true) { + const rows = checks.all(after) as Array<{ site: string; checked_at: number }> + if (!rows.length) break + await index.loadSiteChecks(rows) + after = rows[rows.length - 1]!.site + result.siteChecks += rows.length + } + } + + // Once, at the end: the counter made exact (a count(*), fine for a one-off). + result.pages = await index.resync() + log(`migrate-index: done, ${result.read} read, ${result.inserted} new, ${result.cards} cards, ${result.pages} pages in Postgres`) + return result + } finally { + db.close() + } +} diff --git a/services/crawler/src/pg-index-store.ts b/services/crawler/src/pg-index-store.ts new file mode 100644 index 0000000..95df8ae --- /dev/null +++ b/services/crawler/src/pg-index-store.ts @@ -0,0 +1,815 @@ +/** + * The local search mirror in Postgres, for when one SQLite writer is not enough. + * + * SQLite takes one writer at a time, and at ~63 pages/s that writer was the crawl's + * ceiling; a million pages an hour is ~280/s. Here any number of publishing processes + * write at once: a batch is one INSERT ... ON CONFLICT (url) DO UPDATE over unnest()ed + * array literals, and the page counter moves in the same statement. CRAWL_INDEX_DATABASE_URL + * selects it; unset, the crawler keeps the SQLite IndexStore (tests and laptops). + * + * Same tables as the SQLite file, in the role's default schema (nothing here names a + * schema): pages (with a stored, weighted `tsvector` and a GIN index), page_policy, + * page_cards, policy_site_checks and index_meta (counters and cursors). + * + * What differs, on purpose: + * - Search ranks a bounded candidate set (the first SEARCH_SCAN matches the GIN index + * yields) with ts_rank, instead of bm25 over every match: a common word has millions, + * and pit.world gives an engine 2.5 s. Each search also runs under a statement timeout. + * - Ids are handed out by a sequence, and concurrent writers commit out of id order. A + * corpus scan therefore stops at the newest id at least `corpusLagMs` old, so a harvest + * cursor never runs past a page that is still being written. + * - Postgres files do not shrink when rows go: freed space is reused by later inserts. So + * the byte budget is kept on the live pages (count x bytes a page), not the file size; + * see trim(). + */ +import { SQL } from 'bun' +import { pgArray } from './site-store.ts' +import { + IndexStore, + type CorpusItem, + type CorpusPage, + type CorpusQuery, + type IndexedRecord, + type IndexHit, + type PageRow, + pageRowOf, + type PolicyRow, + type SearchIndex, + TOTAL_CAP, + toFtsTerms, + type TrimOptions, + type TrimResult, +} from './index-store.ts' + +/** How many matches a search ranks: GIN hands them over unordered, so this bounds the work. */ +const SEARCH_SCAN = 3_000 +/** How many matches a search for one kind of page looks through (the SQLite store's KIND_SCAN). */ +const KIND_SCAN = 8_000 +/** Each search query's own limit; the AND query and the OR fallback together stay inside pit.world's 2.5 s. */ +const SEARCH_TIMEOUT_MS = 1_100 +/** Longer than any write may run, so an id this old belongs to a committed (or abandoned) row. */ +const WRITE_TIMEOUT_MS = 30_000 +/** A btree entry holds ~2.7 kB; a longer address cannot be keyed, and would fail its whole batch. */ +const MAX_URL_BYTES = 2_048 +/** ts_rank weights {D, C, B, A}: body, description and site, keywords, title (bm25 10/6/4/3/1 in SQLite). */ +const WEIGHTS = '{0.1,0.4,0.6,1.0}' + +/** Postgres text cannot hold NUL; a page with one in it must not fail its batch. */ +const clean = (value: string) => (value.includes('\u0000') ? value.replace(/\u0000/g, '') : value) +const cleanOrNull = (value: string | null) => (value === null ? null : clean(value)) + +/** User words as a websearch_to_tsquery string that means "all of them": each token quoted, so `or` and `-` stay words. */ +export function websearchAll(terms: string[]): string { + return terms.join(' ') +} +export function websearchAny(terms: string[]): string { + return terms.join(' or ') +} + +/** /api/corpus `q` as websearch syntax: commas separate alternatives, words in one part all appear (& binds before |). */ +export function corpusWebsearch(input: string): string | null { + const parts = String(input ?? '') + .split(',') + .map((part) => toFtsTerms(part)) + .filter((terms) => terms.length) + .slice(0, 16) + if (!parts.length) return null + return parts.map((terms) => terms.join(' ')).join(' or ') +} + +const TSV = ` + setweight(to_tsvector('simple'::regconfig, title), 'A') || + setweight(to_tsvector('simple'::regconfig, keywords), 'B') || + setweight(to_tsvector('simple'::regconfig, description || ' ' || site || ' ' || replace(site, '.', ' ')), 'C') || + setweight(to_tsvector('simple'::regconfig, body), 'D')` + +const SCHEMA = [ + `create table if not exists pages ( + id bigint generated by default as identity primary key, + url text not null unique, + site text not null, + kind text not null, + title text not null, + description text not null default '', + keywords text not null default '', + body text not null default '', + language text, + fetched_at text not null, + added_at timestamptz not null default clock_timestamp(), + fts tsvector generated always as (${TSV}) stored + )`, + `create index if not exists pages_fts on pages using gin (fts) with (fastupdate = on)`, + // trim() takes the oldest pages first, by when they were read (see IndexStore.trim). + `create index if not exists pages_fetched_at on pages (fetched_at, id)`, + `create table if not exists page_policy ( + id bigint primary key, + site text not null, + kind text not null, + language text, + fetched_at text not null, + train smallint, + reason text, + source text not null default 'crawl' + )`, + `create index if not exists page_policy_train on page_policy (train, id)`, + `create index if not exists page_policy_site on page_policy (site, id)`, + `create index if not exists page_policy_train_site on page_policy (train, site)`, + `create table if not exists policy_site_checks (site text primary key, checked_at bigint not null)`, + `create table if not exists page_cards (url text primary key, image text, price text)`, + `create table if not exists index_meta (k text primary key, n bigint not null)`, + `insert into index_meta (k, n) values ('pages_live', 0) on conflict do nothing`, + // Deletes from the old end and upserts leave dead rows; vacuum them early so their space is + // reused before the files grow (the default waits for 20% of a 10M-row table). Once only: + // ALTER TABLE takes a lock every starting process would queue behind. + `do $$ begin + if (select reloptions is null from pg_class where oid = 'pages'::regclass) then + alter table pages set (autovacuum_vacuum_scale_factor = 0.01, autovacuum_analyze_scale_factor = 0.02, toast.autovacuum_vacuum_scale_factor = 0.01); + alter table page_policy set (autovacuum_vacuum_scale_factor = 0.02); + end if; + end $$`, +] + +const TABLES = ['pages', 'page_policy', 'page_cards', 'policy_site_checks', 'index_meta'] + +/** One page row from an old SQLite index, id and all (migrate-index). */ +export type MigratedPage = { + id: number + url: string + site: string + kind: string + title: string + description: string + keywords: string + body: string + language: string | null + fetched_at: string + train: number | null + reason: string | null + source: string | null +} + +/** + * Full-text reads go through the GIN index and nothing else. A word in most pages makes a + * sequential scan look cheap to the planner, but every row it reads then decompresses a + * tsvector of several kB to test it (seconds, cold, at 70k pages); a bitmap from the GIN index + * needs no such test unless it turns lossy, which a larger work_mem puts off. Likewise an id + * range is read as a bitmap, not by walking the primary key and testing each row in it. + */ +async function fullTextPlan(tx: SQL) { + await tx.unsafe('set local enable_seqscan = off') + await tx.unsafe('set local enable_indexscan = off') + await tx.unsafe(`set local work_mem = '32MB'`) +} + +const isRetryable = (error: unknown) => /deadlock detected|could not serialize access/i.test(String((error as Error)?.message ?? error)) +const isTimeout = (error: unknown) => /statement timeout|canceling statement/i.test(String((error as Error)?.message ?? error)) + +export class PgIndexStore implements SearchIndex { + private sql: SQL + private ready: Promise | null = null + readonly maxBytes: number + /** Freed space is reused, never returned: trim() bounds the live pages, see there. */ + readonly reclaims = true + lastAddAt: string | null = null + private readonly corpusLagMs: number + private sizeCache: { at: number; value: number } | null = null + private countCache: { at: number; value: number } | null = null + private statsCache: { at: number; value: Record } | null = null + private refreshing: Promise | null = null + + constructor(url: string, maxBytes = 0, options: { max?: number; corpusLagMs?: number } = {}) { + this.maxBytes = maxBytes > 0 ? maxBytes : 0 + this.corpusLagMs = Math.max(0, options.corpusLagMs ?? 60_000) + // prepare: false, so every query is planned with its own values: a generic plan for + // `fts @@ $1` cannot tell a rare word from a common one. + this.sql = new SQL({ url, max: options.max ?? 8, idleTimeout: 60, prepare: false }) + } + + private init() { + this.ready ??= this.sql + .begin(async (tx: SQL) => { + // Several processes start at once; CREATE ... IF NOT EXISTS races itself without this. + await tx.unsafe(`select pg_advisory_xact_lock(hashtext('pitcrawl-index-schema'))`) + for (const statement of SCHEMA) await tx.unsafe(statement) + }) + .then(() => undefined) + .catch((error) => { + this.ready = null + throw error + }) + return this.ready + } + + /** A transaction whose statements each give up after `ms`. */ + private async timed(ms: number, run: (tx: SQL) => Promise, fullText = false): Promise { + await this.init() + return (await this.sql.begin(async (tx: SQL) => { + await tx.unsafe(`set local statement_timeout = ${Math.max(1, Math.trunc(ms))}`) + if (fullText) await fullTextPlan(tx) + return run(tx) + })) as T + } + + /** Writes retry a deadlock or serialization failure: losing the batch means losing the work that made it. */ + private async write(ms: number, run: (tx: SQL) => Promise, attempts = 4): Promise { + for (let attempt = 1; ; attempt++) { + try { + return await this.timed(ms, run) + } catch (error) { + if (attempt >= attempts || !isRetryable(error)) throw error + await Bun.sleep(50 * attempt + Math.floor(Math.random() * 50)) + } + } + } + + private async meta(key: string): Promise { + await this.init() + const rows = (await this.sql`select n from index_meta where k = ${key}`) as { n: string | number }[] + return rows.length ? Number(rows[0]!.n) : null + } + + private async setMeta(key: string, value: number): Promise { + await this.init() + await this.sql`insert into index_meta (k, n) values (${key}, ${value}) on conflict (k) do update set n = excluded.n` + } + + /** Keep a batch: one statement for every page, policy row, card and the page counter. */ + async add(entries: Array): Promise { + const rows: PageRow[] = [] + for (const raw of entries) { + const row = pageRowOf(raw) + if (row && Buffer.byteLength(row.url) <= MAX_URL_BYTES) rows.push(row) + } + if (!rows.length) return 0 + // ON CONFLICT DO UPDATE cannot touch one row twice in a statement: the last copy wins, + // as it would have in a loop. Sorted, so concurrent batches lock rows in one order. + const pages = [...new Map(rows.map((r) => [r.url, r])).values()].sort((a, b) => (a.url < b.url ? -1 : a.url > b.url ? 1 : 0)) + const cards = pages.filter((p) => p.card) + const col = (pick: (p: PageRow) => T) => pgArray(pages.map(pick)) + await this.write(WRITE_TIMEOUT_MS, async (tx) => { + await tx` + with input as ( + select * from unnest( + ${col((p) => clean(p.url))}::text[], ${col((p) => clean(p.site))}::text[], ${col((p) => clean(p.kind))}::text[], + ${col((p) => clean(p.title))}::text[], ${col((p) => clean(p.description))}::text[], ${col((p) => clean(p.keywords))}::text[], + ${col((p) => clean(p.body))}::text[], ${col((p) => cleanOrNull(p.language))}::text[], ${col((p) => clean(p.fetched))}::text[], + ${col((p) => p.train)}::smallint[], ${col((p) => cleanOrNull(p.reason))}::text[], ${col((p) => p.source)}::text[] + ) as t(url, site, kind, title, description, keywords, body, language, fetched, train, reason, source) + ), up as ( + insert into pages (url, site, kind, title, description, keywords, body, language, fetched_at) + select url, site, kind, title, description, keywords, body, language, fetched from input order by url + on conflict (url) do update set site = excluded.site, kind = excluded.kind, title = excluded.title, + description = excluded.description, keywords = excluded.keywords, body = excluded.body, + language = excluded.language, fetched_at = excluded.fetched_at + returning id, url, (xmax = 0) as inserted + ), pol as ( + -- A record without a verdict (queued by an older crawler) keeps whatever was decided before. + insert into page_policy (id, site, kind, language, fetched_at, train, reason, source) + select up.id, i.site, i.kind, i.language, i.fetched, i.train, i.reason, i.source from up join input i using (url) order by up.id + on conflict (id) do update set site = excluded.site, kind = excluded.kind, language = excluded.language, + fetched_at = excluded.fetched_at, + train = case when excluded.source = 'crawl' then excluded.train else page_policy.train end, + reason = case when excluded.source = 'crawl' then excluded.reason else page_policy.reason end, + source = case when excluded.source = 'crawl' then 'crawl' else page_policy.source end + ), cards as ( + insert into page_cards (url, image, price) + select url, image, price from unnest( + ${pgArray(cards.map((p) => clean(p.url)))}::text[], ${pgArray(cards.map((p) => cleanOrNull(p.card!.image)))}::text[], ${pgArray(cards.map((p) => cleanOrNull(p.card!.price)))}::text[] + ) as c(url, image, price) order by url + on conflict (url) do update set image = excluded.image, price = excluded.price + ), live as ( + update index_meta set n = n + (select count(*) from up where inserted) where k = 'pages_live' + ) + select count(*) from up` + }) + this.countCache = null + this.lastAddAt = new Date().toISOString() + return rows.length + } + + /** + * Moves the id sequence past `maxId` (never back), so pages the crawl writes from now on take + * ids above every id a migration will bring. Call before load(). + */ + async reserveIds(maxId: number): Promise { + await this.init() + await this.sql`select setval(pg_get_serial_sequence('pages', 'id'), greatest(nextval(pg_get_serial_sequence('pages', 'id')), ${Math.max(1, Math.trunc(maxId))}::bigint))` + } + + /** + * Bulk load from an old SQLite index with ids kept (migrate-index). A page already here (by + * address) wins. A page whose id was taken by one the crawl wrote here before reserveIds() + * gets a fresh id instead, so nothing is lost to a collision. The cursor moves in the same + * statement, so a batch and its place in the run land together or not at all. + */ + async load(rows: MigratedPage[], cursor?: { key: string; value: number }): Promise { + rows = rows.filter((r) => Buffer.byteLength(r.url) <= MAX_URL_BYTES) + if (!rows.length && !cursor) return 0 + const col = (pick: (p: MigratedPage) => T) => pgArray(rows.map(pick)) + const out = await this.write(WRITE_TIMEOUT_MS * 4, async (tx) => { + return (await tx` + with raw as ( + select * from unnest( + ${col((p) => p.id)}::bigint[], ${col((p) => clean(p.url))}::text[], ${col((p) => clean(p.site))}::text[], ${col((p) => clean(p.kind))}::text[], + ${col((p) => clean(p.title))}::text[], ${col((p) => clean(p.description ?? ''))}::text[], ${col((p) => clean(p.keywords ?? ''))}::text[], + ${col((p) => clean(p.body ?? ''))}::text[], ${col((p) => cleanOrNull(p.language))}::text[], ${col((p) => clean(p.fetched_at))}::text[], + ${col((p) => p.train)}::smallint[], ${col((p) => cleanOrNull(p.reason))}::text[], ${col((p) => p.source ?? 'legacy')}::text[] + ) as t(id, url, site, kind, title, description, keywords, body, language, fetched, train, reason, source) + ), input as materialized ( + select r.*, case when exists (select 1 from pages p where p.id = r.id) then nextval(pg_get_serial_sequence('pages', 'id')) else r.id end as nid + from raw r where not exists (select 1 from pages p where p.url = r.url) + ), ins as ( + insert into pages (id, url, site, kind, title, description, keywords, body, language, fetched_at) + select nid, url, site, kind, title, description, keywords, body, language, fetched from input order by nid + on conflict do nothing + returning id + ), pol as ( + insert into page_policy (id, site, kind, language, fetched_at, train, reason, source) + select i.nid, i.site, i.kind, i.language, i.fetched, i.train, i.reason, i.source from input i join ins on ins.id = i.nid + on conflict (id) do nothing + ), live as ( + update index_meta set n = n + (select count(*) from ins) where k = 'pages_live' + ), cur as ( + insert into index_meta (k, n) select ${cursor?.key ?? 'migrate_cursor'}, ${cursor?.value ?? 0}::bigint where ${Boolean(cursor)}::boolean + on conflict (k) do update set n = excluded.n + ) + select count(*)::int as n from ins`) as { n: number }[] + }) + this.countCache = null + return Number(out[0]?.n ?? 0) + } + + /** Cards from an old SQLite index (migrate-index): a card already here wins. */ + async loadCards(rows: Array<{ url: string; image: string | null; price: string | null }>): Promise { + rows = rows.filter((r) => Buffer.byteLength(r.url) <= MAX_URL_BYTES) + if (!rows.length) return + await this.init() + await this.sql` + insert into page_cards (url, image, price) + select * from unnest(${pgArray(rows.map((r) => clean(r.url)))}::text[], ${pgArray(rows.map((r) => cleanOrNull(r.image)))}::text[], ${pgArray(rows.map((r) => cleanOrNull(r.price)))}::text[]) + on conflict (url) do nothing` + } + + async loadSiteChecks(rows: Array<{ site: string; checked_at: number }>): Promise { + if (!rows.length) return + await this.init() + await this.sql` + insert into policy_site_checks (site, checked_at) + select * from unnest(${pgArray(rows.map((r) => clean(r.site)))}::text[], ${pgArray(rows.map((r) => r.checked_at))}::bigint[]) + on conflict (site) do update set checked_at = greatest(policy_site_checks.checked_at, excluded.checked_at)` + } + + async cursor(key: string): Promise { + return this.meta(key) + } + + async setCursor(key: string, value: number | null): Promise { + if (value === null) { + await this.init() + await this.sql`delete from index_meta where k = ${key}` + } else await this.setMeta(key, value) + } + + // ---- the byte budget --------------------------------------------------------------- + + private async measureSize(): Promise { + await this.init() + const parts = TABLES.map((t) => `pg_total_relation_size('${t}'::regclass)`).join(' + ') + const out = (await this.sql.unsafe(`select (${parts})::bigint as n`)) as { n: string | number }[] + return Number(out[0]?.n ?? 0) + } + + /** + * Bytes on disk, from the catalogue (no scan). /stats asks on every request: a cached + * figure comes back at once and a stale one is refreshed behind it, so a slow database + * costs /stats a slightly old number, never a wait. `0` asks for it fresh. + */ + async sizeBytes(maxAgeMs = 5_000): Promise { + const now = Date.now() + if (maxAgeMs > 0 && this.sizeCache) { + if (now - this.sizeCache.at >= maxAgeMs) this.refresh() + return this.sizeCache.value + } + const value = await this.measureSize() + this.sizeCache = { at: Date.now(), value } + return value + } + + /** Pages kept, from the counter add() and trim() keep: cached like sizeBytes. */ + async count(maxAgeMs = 2_000): Promise { + const now = Date.now() + if (maxAgeMs > 0 && this.countCache) { + if (now - this.countCache.at >= maxAgeMs) this.refresh() + return this.countCache.value + } + const value = Math.max(0, (await this.meta('pages_live')) ?? 0) + this.countCache = { at: Date.now(), value } + return value + } + + private refresh() { + this.refreshing ??= Promise.all([this.measureSize(), this.meta('pages_live')]) + .then(([size, live]) => { + this.sizeCache = { at: Date.now(), value: size } + this.countCache = { at: Date.now(), value: Math.max(0, live ?? 0) } + }) + .catch(() => {}) + .finally(() => { + this.refreshing = null + }) + } + + /** The page counter made exact again (count(*): seconds on millions of rows, never on a hot path). */ + async resync(): Promise { + await this.init() + await this.sql`update index_meta set n = (select count(*) from pages) where k = 'pages_live'` + this.countCache = null + return this.count(0) + } + + /** + * Keeps the live pages inside the budget, oldest (by fetched_at) first, in batches. + * + * Unlike SQLite's incremental vacuum, deleting rows here never shrinks a file: vacuum marks + * the space free and later inserts reuse it. Trimming until the file is under budget would + * therefore delete everything. Instead the budget is applied to the live pages: bytes a page + * = file size / pages, measured whenever the files have grown (growth means no free space + * was left, so the ratio is honest), and pages are dropped until pages x bytes a page fits. + * The files then stop growing at about the budget. A budget lowered below what the files + * already take frees room inside them; only VACUUM FULL or pg_repack gives it back to the disk. + * + * Any number of processes may run this: each batch takes a transaction-level advisory lock, + * re-reads the counter and skips rows a writer holds. + */ + async trim(options: TrimOptions = {}): Promise { + const batch = Math.max(1, Math.trunc(options.batch ?? 2_000)) + const maxBatches = Math.max(1, Math.trunc(options.maxBatches ?? 50)) + const keepMin = Math.max(0, Math.trunc(options.keepMin ?? 1_000)) + const idle = options.sleep ?? ((ms: number) => new Promise((resolve) => setTimeout(resolve, ms))) + const bytesBefore = await this.sizeBytes(0) + const live = await this.count(0) + if (!this.maxBytes) return { deleted: 0, bytesBefore, bytesAfter: bytesBefore, done: true, reclaims: true } + const lastSize = await this.meta('trim_size') + let perPage = await this.meta('trim_bytes_per_page') + if (live > 0 && (perPage === null || lastSize === null || bytesBefore > lastSize)) { + perPage = Math.max(1, Math.ceil(bytesBefore / live)) + await this.setMeta('trim_bytes_per_page', perPage) + await this.setMeta('trim_size', bytesBefore) + } + if (!perPage || live * perPage <= this.maxBytes) return { deleted: 0, bytesBefore, bytesAfter: bytesBefore, done: true, reclaims: true, liveBytes: live * (perPage ?? 0) } + const target = Math.max(keepMin, Math.floor(this.maxBytes / perPage)) + let deleted = 0 + for (let round = 0; round < maxBatches; round++) { + const gone = await this.write(WRITE_TIMEOUT_MS, async (tx) => { + const lock = (await tx`select pg_try_advisory_xact_lock(hashtext('pitcrawl-index-trim')) as ok`) as { ok: boolean }[] + if (!lock[0]?.ok) return -1 + const now = (await tx`select n from index_meta where k = 'pages_live'`) as { n: string | number }[] + const room = Math.min(batch, Number(now[0]?.n ?? 0) - target) + if (room < 1) return 0 + const out = (await tx` + with picked as ( + select id from pages order by fetched_at, id limit ${room} for update skip locked + ), pol as ( + delete from page_policy where id in (select id from picked) + ), del as ( + delete from pages p using picked where p.id = picked.id returning p.id + ), live as ( + update index_meta set n = greatest(0, n - (select count(*) from del)) where k = 'pages_live' + ) + select count(*)::int as n from del`) as { n: number }[] + return Number(out[0]?.n ?? 0) + }) + if (gone < 0) break // another process is trimming + deleted += gone + if (gone < batch) break + await idle(0) + } + this.countCache = null + this.statsCache = null + const left = await this.count(0) + const bytesAfter = await this.sizeBytes(0) + return { deleted, bytesBefore, bytesAfter, done: left * perPage <= this.maxBytes, reclaims: true, liveBytes: left * perPage } + } + + // ---- the training corpus (/api/corpus) --------------------------------------------------- + + /** + * The top of the id space a corpus scan may cover: the newest id at least corpusLagMs old. + * Concurrent writers commit out of id order, so the very newest ids may still have a gap + * below them that a cursor must not run past. Bounded by the rows written in that window. + */ + private async safeTop(tx: SQL): Promise { + const out = (await tx` + select coalesce((select id from pages where added_at < clock_timestamp() - make_interval(secs => ${this.corpusLagMs / 1000}::float8) order by id desc limit 1), 0) as n`) as { + n: string | number + }[] + return Number(out[0]?.n ?? 0) + } + + /** As IndexStore.corpus: id order, open pages only, windows until `limit` or the time budget. */ + async corpus(query: CorpusQuery): Promise { + const limit = Math.min(Math.max(1, Math.trunc(query.limit ?? 20)), 100) + const textMax = Math.min(Math.max(0, Math.trunc(query.textMax ?? 24_000)), 100_000) + const budget = query.budgetMs ?? 800 + const start = Math.max(0, Math.trunc(query.cursor ?? 0)) + const match = query.q ? corpusWebsearch(query.q) : null + if (query.q && !match) return { items: [], next_cursor: null, resume_cursor: String(start), total_matches: 0, total_capped: false } + return this.timed(Math.max(2_000, budget * 4), async (tx) => { + const maxId = await this.safeTop(tx) + // After safeTop, which walks the primary key backwards and must stay an index scan. + if (match) await fullTextPlan(tx) + const filters: string[] = [] + const params: Array = [] + const bind = (value: string | number) => { + params.push(value) + return `$${params.length}` + } + // Fixed leading parameters per query shape, then the filters. + const lead = match ? 4 : 3 + for (let i = 0; i < lead; i++) params.push(0) + if (query.site) filters.push(`pp.site = ${bind(query.site.toLowerCase().replace(/^www\./, ''))}`) + if (query.kind) filters.push(`pp.kind = ${bind(query.kind)}`) + if (query.lang) filters.push(`(pp.language = ${bind(query.lang.toLowerCase())} or pp.language like ${bind(`${query.lang.toLowerCase()}-%`)})`) + if (query.since) filters.push(`pp.fetched_at >= ${bind(query.since)}`) + const where = filters.length ? ` and ${filters.join(' and ')}` : '' + const ids: number[] = [] + let scannedTo = start + const began = performance.now() + const run = async (sqlText: string, values: Array) => ((await tx.unsafe(sqlText, values)) as { id: string | number }[]).map((r) => Number(r.id)) + if (!match && query.site) { + // One site: its own index, no windows needed. + const values = [...params] + values[0] = start + values[1] = maxId + values[2] = limit + // threatcrush-disable-next-line sql-template-interpolation only fixed column filters and "$n" placeholders are interpolated; every value is bound + const found = await run(`select pp.id from page_policy pp where pp.train = 1 and pp.id > $1 and pp.id <= $2${where} order by pp.id limit $3`, values) + ids.push(...found) + // Never behind the cursor it was given: a cursor past the lagged top stays where it is. + scannedTo = found.length === limit ? found[found.length - 1]! : Math.max(start, maxId) + } else { + const sqlText = match + ? // threatcrush-disable-next-line sql-template-interpolation only fixed column filters and "$n" placeholders are interpolated; every value is bound + `select p.id from pages p join page_policy pp on pp.id = p.id where p.fts @@ websearch_to_tsquery('simple'::regconfig, $4) and p.id > $1 and p.id <= $2 and pp.train = 1${where} order by p.id limit $3` + : // threatcrush-disable-next-line sql-template-interpolation only fixed column filters and "$n" placeholders are interpolated; every value is bound + `select pp.id from page_policy pp where pp.train = 1 and pp.id > $1 and pp.id <= $2${where} order by pp.id limit $3` + let window = 50_000 + let lo = start + while (lo < maxId && ids.length < limit) { + const hi = Math.min(maxId, lo + window) + const need = limit - ids.length + const values = [...params] + values[0] = lo + values[1] = hi + values[2] = need + if (match) values[3] = match + const found = await run(sqlText, values) + ids.push(...found) + scannedTo = found.length === need ? found[found.length - 1]! : hi + lo = scannedTo + if (performance.now() - began > budget) break + if (found.length < need) window = Math.min(window * 4, 2_000_000) + } + } + const items: CorpusItem[] = [] + if (ids.length) { + const rows = (await tx` + select id, url, title, site, kind, language, fetched_at, left(body, ${textMax}) as text + from pages where id = any(${pgArray(ids)}::bigint[]) order by id`) as Array + for (const r of rows) items.push({ ...r, id: Number(r.id) }) + } + let total: number | null = null + let capped = false + if (match && start === 0) { + const out = (await tx` + select count(*)::int as n from (select 1 from pages where fts @@ websearch_to_tsquery('simple'::regconfig, ${match}) limit ${TOTAL_CAP}) t`) as { n: number }[] + total = Number(out[0]?.n ?? 0) + capped = total >= TOTAL_CAP + } + const done = scannedTo >= maxId && ids.length < limit + return { items, next_cursor: done ? null : String(scannedTo), resume_cursor: String(scannedTo), total_matches: total, total_capped: capped } + }) + } + + private async computeStats(now: number): Promise> { + return this.timed(60_000, async (tx) => { + const byTrain = (await tx`select train, count(*)::bigint as n from page_policy group by train`) as Array<{ train: number | null; n: string | number }> + const n = (t: number | null) => Number(byTrain.find((r) => (r.train === null ? null : Number(r.train)) === t)?.n ?? 0) + const sites = (await tx`select count(distinct site)::bigint as n from page_policy where train = 1`) as { n: string | number }[] + const languages = (await tx`select coalesce(language, 'unknown') as language, count(*)::int as pages from page_policy where train = 1 group by 1 order by 2 desc limit 25`) as Array<{ + language: string + pages: number + }> + const kinds = (await tx`select kind, count(*)::int as pages from page_policy where train = 1 group by 1 order by 2 desc limit 25`) as Array<{ kind: string; pages: number }> + const reasons = (await tx`select reason, count(*)::int as pages from page_policy where train = 0 group by 1 order by 2 desc limit 25`) as Array<{ reason: string; pages: number }> + const last = (await tx`select max(fetched_at) as t from (select fetched_at from page_policy order by id desc limit 1000) x`) as { t: string | null }[] + const live = await this.count(0) + return { + pages_indexed: live, + pages: n(1), + excluded: n(0), + unknown: n(null), + not_yet_mapped: Math.max(0, live - n(1) - n(0) - n(null)), + sites: Number(sites[0]?.n ?? 0), + languages: languages.map((r) => ({ language: r.language, pages: Number(r.pages) })), + kinds: kinds.map((r) => ({ kind: r.kind, pages: Number(r.pages) })), + excluded_by: reasons.map((r) => ({ reason: r.reason, pages: Number(r.pages) })), + last_crawl: this.lastAddAt ?? last[0]?.t ?? null, + computed_at: new Date(now).toISOString(), + } + }) + } + + /** + * Corpus counts for harvesters, cached for five minutes. Each is a scan of page_policy, so + * once a figure exists a stale one is served while a fresh one is computed behind it. + */ + private statsRefresh: Promise | null = null + async corpusStats(now = Date.now()): Promise> { + if (this.statsCache && now - this.statsCache.at < 300_000) return this.statsCache.value + if (this.statsCache) { + this.statsRefresh ??= this.computeStats(now) + .then((value) => (this.statsCache = { at: now, value })) + .catch(() => {}) + .finally(() => (this.statsRefresh = null)) + return this.statsCache.value + } + const value = await this.computeStats(now) + this.statsCache = { at: now, value } + return value + } + + // ---- backfill: pages indexed before verdicts were recorded --------------------------- + + /** Gives `n` more pages without a page_policy row one (train unknown). 0 once every page has one. */ + async mapLegacy(n = 2000): Promise { + const after = (await this.meta('map_cursor')) ?? 0 + const out = (await this.sql` + with batch as (select id, site, kind, language, fetched_at from pages where id > ${after} order by id limit ${n}), + ins as ( + insert into page_policy (id, site, kind, language, fetched_at, train, reason, source) + select id, site, kind, language, fetched_at, null, null, 'legacy' from batch + on conflict (id) do nothing + ), cur as ( + insert into index_meta (k, n) select 'map_cursor', max(id) from batch having count(*) > 0 + on conflict (k) do update set n = excluded.n + ) + select count(*)::int as n from batch`) as { n: number }[] + return Number(out[0]?.n ?? 0) + } + + async unknownSites(n = 16, now = Date.now()): Promise { + await this.init() + const rows = (await this.sql` + select distinct pp.site from page_policy pp where pp.train is null + and not exists (select 1 from policy_site_checks c where c.site = pp.site and c.checked_at > ${now - 86_400_000}) + limit ${n}`) as Array<{ site: string }> + return rows.map((r) => r.site) + } + + async unknownPagesOf(site: string): Promise { + await this.init() + const rows = (await this.sql`select pp.id, p.url from page_policy pp join pages p on p.id = pp.id where pp.site = ${site} and pp.train is null`) as Array<{ id: string | number; url: string }> + return rows.map((r) => ({ id: Number(r.id), url: r.url })) + } + + /** Records verdicts for unknown pages (never overrides one made at crawl time) and that the site was tried. */ + async setPolicies(site: string, verdicts: Array<{ id: number; allowed: boolean; reason: string }>, now = Date.now()): Promise { + await this.write(WRITE_TIMEOUT_MS, async (tx) => { + if (verdicts.length) + await tx` + update page_policy pp set train = v.train, reason = v.reason, source = 'backfill' + from unnest(${pgArray(verdicts.map((v) => v.id))}::bigint[], ${pgArray(verdicts.map((v) => (v.allowed ? 1 : 0)))}::smallint[], ${pgArray(verdicts.map((v) => clean(v.reason.slice(0, 120))))}::text[]) as v(id, train, reason) + where pp.id = v.id and pp.train is null` + await tx`insert into policy_site_checks (site, checked_at) values (${site}, ${now}) on conflict (site) do update set checked_at = excluded.checked_at` + }) + } + + async setCard(url: string, image: string | null, price: string | null): Promise { + if (Buffer.byteLength(url) > MAX_URL_BYTES) return + await this.init() + await this + .sql`insert into page_cards (url, image, price) values (${clean(url)}, ${cleanOrNull(image)}, ${cleanOrNull(price)}) on conflict (url) do update set image = excluded.image, price = excluded.price` + } + + /** Cards for these pages. A page absent from the map has never been looked at. */ + async cards(urls: string[]): Promise> { + const out = new Map() + if (!urls.length) return out + await this.init() + const rows = (await this.sql`select url, image, price from page_cards where url = any(${pgArray(urls.map(clean))}::text[])`) as Array<{ url: string; image: string | null; price: string | null }> + for (const r of rows) out.set(r.url, { image: r.image, price: r.price }) + return out + } + + /** + * Best matches: every word if any page has them all, else any word. Title weighs most. + * Ranks the first SEARCH_SCAN matches (ts_rank, weighted like the SQLite bm25), so a common + * word costs a bounded amount; `score` is negated so lower is better, as bm25 was. + */ + async search(query: string, limit = 10, kind?: string): Promise { + const terms = toFtsTerms(query) + if (!terms.length) return [] + if (kind) return this.searchKind(terms, kind, limit) + const k = Math.min(Math.max(1, limit), 50) + const run = async (q: string) => { + try { + return await this.timed( + SEARCH_TIMEOUT_MS, + async (tx) => { + const rows = (await tx` + with cand as materialized ( + select id, fts from pages where fts @@ websearch_to_tsquery('simple'::regconfig, ${q}) limit ${SEARCH_SCAN} + ), best as ( + select id, ts_rank(${WEIGHTS}::float4[], fts, websearch_to_tsquery('simple'::regconfig, ${q})) as rank from cand + order by rank desc, id desc limit ${k} + ) + select p.url, p.title, p.site, p.kind, p.fetched_at, + case when p.description <> '' then p.description else left(p.body, 300) end as summary, + -best.rank as score + from best join pages p on p.id = best.id + order by best.rank desc, best.id desc`) as IndexHit[] + return rows.map((r) => ({ ...r, score: Number(r.score) })) + }, + true, + ) + } catch (error) { + // Too slow is no answer, not an error: pit.world has other engines. + if (isTimeout(error)) return [] + throw error + } + } + const all = await run(websearchAll(terms)) + return all.length || terms.length === 1 ? all : run(websearchAny(terms)) + } + + /** As IndexStore.searchKind: KIND_SCAN matches, that kind only, most words in the title first. */ + private async searchKind(terms: string[], kind: string, limit: number): Promise { + const run = async (q: string) => { + try { + return await this.timed( + SEARCH_TIMEOUT_MS, + async (tx) => { + return (await tx` + with cand as materialized ( + select id from pages where fts @@ websearch_to_tsquery('simple'::regconfig, ${q}) limit ${KIND_SCAN} + ) + select p.url, p.title, p.site, p.kind, p.fetched_at, + case when p.description <> '' then p.description else left(p.body, 300) end as summary, 0 as score + from cand join pages p on p.id = cand.id + where p.kind = ${kind} + order by p.id desc limit 200`) as IndexHit[] + }, + true, + ) + } catch (error) { + if (isTimeout(error)) return [] + throw error + } + } + let hits = await run(websearchAll(terms)) + if (!hits.length && terms.length > 1) hits = await run(websearchAny(terms)) + const words = terms.map((t) => t.replace(/"/g, '').toLowerCase()) + const inTitle = (hit: IndexHit) => { + const title = hit.title.toLowerCase() + return words.filter((w) => title.includes(w)).length + } + return hits + .map((hit) => ({ hit: { ...hit, score: Number(hit.score) }, n: inTitle(hit) })) + .sort((a, b) => b.n - a.n) + .slice(0, Math.min(Math.max(1, limit), 50)) + .map(({ hit }) => hit) + } + + async importCursor(): Promise { + const value = await this.meta('import_cursor') + return value ? value : null + } + + async setImportCursor(cursor: number | null): Promise { + await this.setCursor('import_cursor', cursor) + } + + async tables(): Promise { + await this.init() + const rows = (await this.sql`select table_name as name from information_schema.tables where table_schema = current_schema() order by 1`) as Array<{ name: string }> + return rows.map((r) => r.name) + } + + /** The highest id in use (the primary key's last entry: no scan). */ + async highWaterMark(): Promise { + await this.init() + const out = (await this.sql`select coalesce(max(id), 0) as n from pages`) as { n: string | number }[] + return Number(out[0]?.n ?? 0) + } + + async close(): Promise { + await this.refreshing?.catch(() => {}) + await this.sql.close() + } +} + +/** CRAWL_INDEX_DATABASE_URL set: Postgres, many writers. Unset: the SQLite file at indexPath. */ +export function openIndex(config: { indexPath: string; indexMaxBytes: number; indexDatabaseUrl: string }): SearchIndex { + return config.indexDatabaseUrl ? new PgIndexStore(config.indexDatabaseUrl, config.indexMaxBytes) : new IndexStore(config.indexPath, config.indexMaxBytes) +} diff --git a/services/crawler/src/queues.ts b/services/crawler/src/queues.ts index e6743d2..05c1d5a 100644 --- a/services/crawler/src/queues.ts +++ b/services/crawler/src/queues.ts @@ -24,7 +24,8 @@ import { RedisFrontier } from './frontier.ts' import { PageArchive } from './archive.ts' import { MemorySiteStore, PgSiteStore, type SiteStore } from './site-store.ts' import { submitFeed } from '@profullstack/submit-feed/core' -import { IndexStore } from './index-store.ts' +import type { SearchIndex } from './index-store.ts' +import { openIndex } from './pg-index-store.ts' import type { McpJob } from './mcp.ts' import { ObscuraPool } from './obscura.ts' import { urlId } from './url.ts' @@ -68,7 +69,7 @@ export type Runtime = { frontier: RedisFrontier sites: SiteStore redis: Redis - index: IndexStore + index: SearchIndex close(): Promise } @@ -139,7 +140,7 @@ export function createRuntime(config: Config, log: (line: string) => void = cons } const frontier = new RedisFrontier(redis, config.maxPages, Math.min(config.seenTtlMs, config.recrawlDays * 86_400_000)) const pool = config.obscuraUrls.length ? new ObscuraPool(config.obscuraUrls, config.obscuraToken) : null - const index = new IndexStore(config.indexPath, config.indexMaxBytes) + const index = openIndex(config) if (!config.databaseUrl) log('CRAWL_DATABASE_URL unset: the site frontier is in memory and forgets every site on restart') const sites: SiteStore = config.databaseUrl ? new PgSiteStore(config.databaseUrl) : new MemorySiteStore() // Only a publishing process writes the archive; a fetch-only worker never opens it. @@ -193,7 +194,7 @@ export function createRuntime(config: Config, log: (line: string) => void = cons async close() { await Promise.all(Object.values(queues).map((q) => q.close())) redis.disconnect() - index.close() + await index.close() await sites.close() }, } @@ -373,8 +374,13 @@ export async function report(runtime: Runtime, config: Config) { // Redis, like the media counters: this page must never wait on SQLite for it. stats.index_mcps = await runtime.frontier.mcpCount().catch(() => 0) try { - stats.index_pages = runtime.index.count() - stats.index_bytes = runtime.index.sizeBytes() + // A second at most: Postgres answers these from a cache, SQLite from a counter and two pragmas. + const [pages, bytes] = await Promise.race([ + Promise.all([Promise.resolve(runtime.index.count()), Promise.resolve(runtime.index.sizeBytes())]), + new Promise((_, reject) => setTimeout(() => reject(new Error('index figures timed out')), 1_000)), + ]) + stats.index_pages = pages + stats.index_bytes = bytes stats.index_max_bytes = runtime.index.maxBytes } catch { // Leave them out rather than report a wrong zero: /status shows no tile for a figure diff --git a/services/crawler/test/archive.test.ts b/services/crawler/test/archive.test.ts index c09845d..47353e9 100644 --- a/services/crawler/test/archive.test.ts +++ b/services/crawler/test/archive.test.ts @@ -1,28 +1,78 @@ import { describe, expect, test } from 'bun:test' -import { mkdtempSync, readdirSync, readFileSync, rmSync } from 'node:fs' +import { mkdtempSync, readdirSync, readFileSync, rmSync, writeFileSync } from 'node:fs' import { tmpdir } from 'node:os' import { join } from 'node:path' import { gunzipSync } from 'node:zlib' -import { hourName, PageArchive } from '../src/archive.ts' +import { hourName, PageArchive, processTag } from '../src/archive.ts' import { configFromEnv } from '../src/config.ts' import { createCrawler } from '../src/crawler.ts' import { MemoryFrontier } from '../src/frontier.ts' describe('the page archive', () => { - test('one gzip file per UTC hour, batches appended as members, earlier hours sealed', () => { + test('one gzip file per UTC hour and process, batches appended as members, earlier hours sealed', () => { const dir = mkdtempSync(join(tmpdir(), 'pitarchive-')) try { let now = new Date('2026-11-08T15:59:58Z') - const archive = new PageArchive(dir, () => now) + const archive = new PageArchive(dir, () => now, 'hostA-11') archive.write(['{"a":1}', '{"a":2}']) archive.write(['{"a":3}']) - expect(readdirSync(dir)).toEqual(['pages-20261108T15.jsonl.gz.partial']) + expect(readdirSync(dir)).toEqual(['pages-20261108T15-hostA-11.jsonl.gz.partial']) now = new Date('2026-11-08T16:00:01Z') archive.write(['{"b":1}']) - expect(readdirSync(dir).sort()).toEqual(['pages-20261108T15.jsonl.gz', 'pages-20261108T16.jsonl.gz.partial']) + expect(readdirSync(dir).sort()).toEqual(['pages-20261108T15-hostA-11.jsonl.gz', 'pages-20261108T16-hostA-11.jsonl.gz.partial']) // Concatenated gzip members read back as one stream, in order. - expect(gunzipSync(readFileSync(join(dir, 'pages-20261108T15.jsonl.gz'))).toString()).toBe('{"a":1}\n{"a":2}\n{"a":3}\n') + expect(gunzipSync(readFileSync(join(dir, 'pages-20261108T15-hostA-11.jsonl.gz'))).toString()).toBe('{"a":1}\n{"a":2}\n{"a":3}\n') expect(hourName(new Date('2026-01-02T03:04:05Z'))).toBe('pages-20260102T03.jsonl.gz') + expect(hourName(new Date('2026-01-02T03:04:05Z'), 'h-1')).toBe('pages-20260102T03-h-1.jsonl.gz') + // Whatever the tag, the shipper's glob (deploy/crawler-remote/ship-archive.sh) matches a sealed hour, never an open one. + expect(new Bun.Glob('pages-*.jsonl.gz').match('pages-20261108T15-hostA-11.jsonl.gz')).toBe(true) + expect(new Bun.Glob('pages-*.jsonl.gz').match('pages-20261108T16-hostA-11.jsonl.gz.partial')).toBe(false) + expect(processTag('my host/x', 42)).toBe('my_host_x-42') + } finally { + rmSync(dir, { recursive: true, force: true }) + } + }) + + test("several publishing processes write their own files and never seal one another's current hour", () => { + const dir = mkdtempSync(join(tmpdir(), 'pitarchive-')) + try { + let now = new Date('2026-11-08T15:30:00Z') + const a = new PageArchive(dir, () => now, 'box-1') + const b = new PageArchive(dir, () => now, 'box-2') + a.write(['{"a":1}']) + b.write(['{"b":1}']) + a.write(['{"a":2}']) + expect(readdirSync(dir).sort()).toEqual(['pages-20261108T15-box-1.jsonl.gz.partial', 'pages-20261108T15-box-2.jsonl.gz.partial']) + expect(gunzipSync(readFileSync(join(dir, 'pages-20261108T15-box-1.jsonl.gz.partial'))).toString()).toBe('{"a":1}\n{"a":2}\n') + // The hour turns: a seals its own hour 15 at once, but not b's, which may still be finishing a batch there. + now = new Date('2026-11-08T16:00:01Z') + a.write(['{"a":3}']) + expect(readdirSync(dir).sort()).toEqual(['pages-20261108T15-box-1.jsonl.gz', 'pages-20261108T15-box-2.jsonl.gz.partial', 'pages-20261108T16-box-1.jsonl.gz.partial']) + // b writes again and seals its own. + b.write(['{"b":2}']) + expect( + readdirSync(dir) + .filter((f) => f.endsWith('.partial')) + .sort(), + ).toEqual(['pages-20261108T16-box-1.jsonl.gz.partial', 'pages-20261108T16-box-2.jsonl.gz.partial']) + // A process that died (b) leaves a partial; another seals it once that hour is two behind. + now = new Date('2026-11-08T17:10:00Z') + a.write([]) + expect(readdirSync(dir)).toContain('pages-20261108T16-box-2.jsonl.gz.partial') + now = new Date('2026-11-08T18:00:00Z') + a.write([]) + expect(readdirSync(dir).filter((f) => f.endsWith('.partial'))).toEqual([]) + } finally { + rmSync(dir, { recursive: true, force: true }) + } + }) + + test('a leftover partial from before per-process names is still sealed', () => { + const dir = mkdtempSync(join(tmpdir(), 'pitarchive-')) + try { + writeFileSync(join(dir, 'pages-20261108T15.jsonl.gz.partial'), '') + const archive = new PageArchive(dir, () => new Date('2026-11-08T18:00:00Z'), 'box-1') + expect(archive.seal()).toEqual(['pages-20261108T15.jsonl.gz']) } finally { rmSync(dir, { recursive: true, force: true }) } diff --git a/services/crawler/test/pg-index-store.test.ts b/services/crawler/test/pg-index-store.test.ts new file mode 100644 index 0000000..c5bf671 --- /dev/null +++ b/services/crawler/test/pg-index-store.test.ts @@ -0,0 +1,350 @@ +/** + * The search mirror on a real Postgres. Runs when CRAWL_TEST_PG is set, e.g. against + * `docker run -d --rm --name pgidx-test -e POSTGRES_PASSWORD=test -p 127.0.0.1:6493:5432 postgres:17-alpine`. + * + * Each test gets its own login role with a schema of the same name, so the store finds its + * tables through the role's default search_path ("$user") and never touches public. + */ +import { afterAll, afterEach, describe, expect, test } from 'bun:test' +import { SQL } from 'bun' +import { rmSync } from 'node:fs' +import { tmpdir } from 'node:os' +import { IndexStore } from '../src/index-store.ts' +import { migrateIndexFromSqlite } from '../src/migrate-index.ts' +import { corpusWebsearch, PgIndexStore } from '../src/pg-index-store.ts' + +const PG = process.env.CRAWL_TEST_PG +const OPEN = { allowed: true, reason: 'no reservation' } + +const entry = ( + url: string, + title: string, + body: string, + extra: { keywords?: string[]; kind?: string; language?: string; fetched_at?: string; training?: { allowed: boolean; reason: string } | null; description?: string | null } = {}, +) => ({ + record: { + url, + canonical: url, + title, + description: extra.description ?? null, + kind: extra.kind ?? 'page', + language: extra.language ?? 'en', + fetched_at: extra.fetched_at ?? '2026-11-02T00:00:00.000Z', + }, + enrichment: { keywords: extra.keywords ?? [], excerpt: body, text: body, ...(extra.training === null ? {} : { training: extra.training ?? OPEN }) }, +}) + +describe.skipIf(!PG)('the search mirror on a real Postgres', () => { + const admin = PG ? new SQL({ url: PG, max: 2 }) : null + const made: string[] = [] + const stores: PgIndexStore[] = [] + + /** A fresh role + schema; returns its connection string. */ + const fresh = async () => { + const name = `pgidx_${Date.now().toString(36)}_${Math.random().toString(36).slice(2, 8)}` + await admin!.unsafe(`create role ${name} login password '${name}'`) + await admin!.unsafe(`create schema ${name} authorization ${name}`) + made.push(name) + const url = new URL(PG!) + url.username = name + url.password = name + return url.toString() + } + const open = async (maxBytes = 0, options: { corpusLagMs?: number } = {}, url?: string) => { + const store = new PgIndexStore(url ?? (await fresh()), maxBytes, { corpusLagMs: 0, max: 3, ...options }) + stores.push(store) + return store + } + + afterEach(async () => { + for (const s of stores.splice(0)) await s.close().catch(() => {}) + }) + + afterAll(async () => { + for (const name of made) { + await admin!.unsafe(`drop schema if exists ${name} cascade`) + await admin!.unsafe(`drop role if exists ${name}`) + } + await admin?.close() + }) + + test("tables live in the role's own schema; whatever someone types is a valid query", async () => { + const store = await open() + await store.add([entry('https://a.example/', 'Node.js catch-all errors', 'TypeError: x is not a function in C# and node.js 1.3.14')]) + expect(await store.tables()).toEqual(['index_meta', 'page_cards', 'page_policy', 'pages', 'policy_site_checks']) + const inPublic = (await admin!`select count(*)::int as n from information_schema.tables where table_schema = 'public' and table_name = 'pages'`) as { n: number }[] + expect(inPublic[0]!.n).toBe(0) + for (const q of ['catch-all', 'TypeError: x', 'foo(bar', 'node.js', '1.3.14', 'C#', 'why "this fails', '***', '-', 'or', 'a OR b', '!x', "it's", '\\', '_']) + expect(Array.isArray(await store.search(q))).toBe(true) + expect(await store.search('node.js')).toHaveLength(1) + expect(await store.search('1.3.14')).toHaveLength(1) + expect(await store.search('node.js', 10, 'no-such-kind')).toHaveLength(0) + // `or` is a word here, not an operator: nothing has "catch" AND "or". + expect(await store.search('typeerror or')).toHaveLength(1) // the OR fallback finds typeerror + }) + + test('all words first, any word when nothing has them all; title outranks body; a re-crawl replaces', async () => { + const store = await open() + await store.add([ + entry('https://ovens.example/pizza', 'Wood-fired pizza ovens', 'how to build a brick oven', { keywords: ['pizza', 'oven'] }), + entry('https://blog.example/x', 'A post about bread', 'we once ate pizza near an oven'), + entry('https://other.example/', 'Gardening', 'tomatoes'), + ]) + expect((await store.search('pizza oven')).map((h) => h.url)).toEqual(['https://ovens.example/pizza', 'https://blog.example/x']) + expect((await store.search('pizza tomatoes')).map((h) => h.url).sort()).toEqual(['https://blog.example/x', 'https://other.example/', 'https://ovens.example/pizza']) + const [top] = await store.search('pizza') + expect(top!.summary).toBe('how to build a brick oven') + expect(top!.score).toBeLessThan(0) + // The site's own name parts are searchable, as they were in FTS5. + expect((await store.search('ovens example')).map((h) => h.url)).toContain('https://ovens.example/pizza') + await store.add([JSON.stringify(entry('https://other.example/', 'Gardening again', 'cucumbers', { description: 'All about cucumbers.' }))]) + expect(await store.count()).toBe(3) + expect(await store.search('tomatoes')).toEqual([]) + const [cuke] = await store.search('cucumbers') + expect(cuke).toMatchObject({ title: 'Gardening again', summary: 'All about cucumbers.', site: 'other.example', kind: 'page', fetched_at: '2026-11-02T00:00:00.000Z' }) + }) + + test('a search for one kind keeps that kind, most words in the title first; cards come with products', async () => { + const store = await open() + const product = (url: string, title: string, body: string) => { + const e = entry(url, title, body, { kind: 'product' }) + return { ...e, record: { ...e.record, image: { url: `${url}.jpg` } }, enrichment: { ...e.enrichment, price: '$5' } } + } + await store.add( + [ + product('https://a.example/1', 'Trail running shoes', 'light and quick'), + product('https://b.example/2', 'Socks', 'for running in shoes'), + entry('https://c.example/3', 'Why running shoes wear out', 'an essay'), + ].map((e) => JSON.stringify(e)), + ) + expect((await store.search('running shoes', 10, 'product')).map((h) => h.url)).toEqual(['https://a.example/1', 'https://b.example/2']) + expect((await store.search('running shoes', 1, 'product')).map((h) => h.url)).toEqual(['https://a.example/1']) + expect((await store.search('running shoes')).map((h) => h.url)).toContain('https://c.example/3') + expect((await store.cards(['https://a.example/1'])).get('https://a.example/1')).toEqual({ image: 'https://a.example/1.jpg', price: '$5' }) + expect((await store.cards(['https://never.example/'])).size).toBe(0) + await store.setCard('https://bare.example/', null, null) + expect((await store.cards(['https://bare.example/', 'https://a.example/1'])).get('https://bare.example/')).toEqual({ image: null, price: null }) + }) + + test('text Postgres cannot hold, and addresses it cannot key, never fail a batch', async () => { + const store = await open() + const kept = await store.add([ + entry('https://nul.example/', 'Has a \u0000 NUL', 'body with \u0000 in it, and "quotes" and \\backslashes\\ and {braces,} and NULL'), + entry(`https://long.example/${'x'.repeat(3_000)}`, 'Too long to key', 'words'), + entry('https://ok.example/', 'Fine', 'words'), + 'not json', + JSON.stringify({ record: { canonical: 'not a url' } }), + ]) + expect(kept).toBe(2) + expect(await store.count()).toBe(2) + expect((await store.search('braces'))[0]!.url).toBe('https://nul.example/') + expect((await store.search('null'))[0]!.url).toBe('https://nul.example/') + }) + + test('the training corpus: comma lists, filters, opt-outs, the backfill and its stats', async () => { + const store = await open() + expect(corpusWebsearch('open source, self-hosted')).toBe('"open" "source" or "self" "hosted"') + expect(corpusWebsearch(' , ,')).toBeNull() + await store.add([ + entry('https://a.example/1', 'one', 'saas pricing', { language: 'en-us' }), + entry('https://a.example/2', 'two', 'crypto wallets', { language: 'de', kind: 'article' }), + entry('https://b.example/3', 'three', 'gardening', { fetched_at: '2026-10-01T00:00:00.000Z' }), + entry('https://a.example/noai', 'four', 'saas reserved', { training: { allowed: false, reason: 'meta robots: noai' } }), + entry('https://old.example/x', 'five', 'saas legacy', { training: null }), + ]) + for (const q of ['catch-all', 'TypeError: x', 'foo(bar', 'NEAR(a b)', '"', '***', '-x', 'a OR', 'title:x', '^x', ',,,']) expect((await store.corpus({ q })).items).toBeDefined() + expect((await store.corpus({ q: 'saas,crypto' })).items.map((i) => i.url)).toEqual(['https://a.example/1', 'https://a.example/2']) + expect((await store.corpus({ q: 'saas pricing,nothing here' })).items.map((i) => i.url)).toEqual(['https://a.example/1']) + expect((await store.corpus({ q: 'saas' })).total_matches).toBe(3) + expect((await store.corpus({})).items).toHaveLength(3) + expect((await store.corpus({ site: 'www.a.example' })).items.map((i) => i.url)).toEqual(['https://a.example/1', 'https://a.example/2']) + expect((await store.corpus({ lang: 'en' })).items.map((i) => i.url)).toEqual(['https://a.example/1', 'https://b.example/3']) + expect((await store.corpus({ kind: 'article' })).items.map((i) => i.url)).toEqual(['https://a.example/2']) + expect((await store.corpus({ since: '2026-11-01T00:00:00.000Z' })).items).toHaveLength(2) + const first = await store.corpus({ limit: 2 }) + expect(first.items).toHaveLength(2) + expect((await store.corpus({ limit: 2, cursor: Number(first.next_cursor) })).items.map((i) => i.url)).toEqual(['https://b.example/3']) + expect((await store.corpus({ textMax: 2 })).items[0]!.text).toBe('sa') + expect(typeof (await store.corpus({})).items[0]!.id).toBe('number') + + expect(await store.corpusStats(0)).toMatchObject({ pages_indexed: 5, pages: 3, excluded: 1, unknown: 1, sites: 2, excluded_by: [{ reason: 'meta robots: noai', pages: 1 }] }) + expect(await store.unknownSites()).toEqual(['old.example']) + const [legacy] = await store.unknownPagesOf('old.example') + await store.setPolicies('old.example', [{ id: legacy!.id, allowed: true, reason: 'site allows (backfill)' }]) + expect(await store.unknownSites()).toEqual([]) + expect((await store.corpus({ q: 'saas' })).items).toHaveLength(2) + // A record without a verdict (an old outbox line) keeps the one already made. + await store.add([entry('https://a.example/noai', 'four', 'saas reserved', { training: null })]) + expect((await store.corpus({ q: 'saas' })).items).toHaveLength(2) + // Pages without a policy row get one, once. + await (store as unknown as { sql: SQL }).sql.unsafe('delete from page_policy') + expect(await store.mapLegacy(2)).toBe(2) + expect(await store.mapLegacy(10)).toBe(3) + expect(await store.mapLegacy(10)).toBe(0) + expect(await store.corpusStats(Date.now() + 10 * 60_000)).toBeDefined() + }) + + test('cursor paging is stable and complete while pages are added, and never runs past a page still being written', async () => { + const store = await open() + await store.add(Array.from({ length: 25 }, (_, i) => entry(`https://s${i % 3}.example/${i}`, `p${i}`, `widget number ${i}`))) + const seen: string[] = [] + let cursor: string | null = '0' + let pages = 0 + while (cursor !== null) { + const page = await store.corpus({ q: 'widget', limit: 10, cursor: Number(cursor) }) + seen.push(...page.items.map((i) => i.url)) + cursor = page.next_cursor + if (pages++ === 0) await store.add([entry('https://late.example/x', 'late', 'widget late')]) + } + expect(new Set(seen).size).toBe(seen.length) + expect(seen).toHaveLength(26) + expect(seen.at(-1)).toBe('https://late.example/x') + await store.add([entry('https://s0.example/0', 'p0', 'widget again')]) + const after = await store.corpus({ q: 'widget', cursor: Number((await store.corpus({ q: 'widget', limit: 100 })).resume_cursor) }) + expect(after.items).toEqual([]) + expect(after.next_cursor).toBeNull() + + // With the production lag, a page written a moment ago is not handed out yet: the cursor stops below it. + const lagged = await open(0, { corpusLagMs: 60_000 }) + await lagged.add([entry('https://new.example/', 'new', 'widget brand new')]) + const page = await lagged.corpus({ q: 'widget' }) + expect(page.items).toEqual([]) + expect(page.resume_cursor).toBe('0') + }) + + test('many writers at once: every page counted once, the counter equals count(*)', async () => { + const url = await fresh() + const writers = await Promise.all(Array.from({ length: 4 }, () => open(0, {}, url))) + // Overlapping batches from four stores (four pools), several rounds each, in parallel. + const urls = Array.from({ length: 300 }, (_, i) => `https://w${i % 37}.example/${i}`) + await Promise.all( + writers.flatMap((w, n) => + Array.from({ length: 5 }, (_, round) => { + const slice = urls.filter((_, i) => (i + n + round) % 3 !== 0) + return w.add(slice.map((u) => entry(u, `title ${u}`, `round ${round} writer ${n}`))) + }), + ), + ) + expect(await writers[0]!.count(0)).toBe(300) + expect(await writers[1]!.resync()).toBe(300) + const policy = (await (writers[0] as unknown as { sql: SQL }).sql`select count(*)::int as n from page_policy`) as { n: number }[] + expect(policy[0]!.n).toBe(300) + expect(await writers[2]!.highWaterMark()).toBeGreaterThanOrEqual(300) + }, 60_000) + + test('the byte budget: oldest pages go first, in batches, never below keepMin, the counter follows', async () => { + const url = await fresh() + const fill = await open(0, {}, url) + const fat = 'lorem ipsum dolor sit amet '.repeat(200) + for (let batch = 0; batch < 5; batch++) + await fill.add( + Array.from({ length: 200 }, (_, i) => { + const n = batch * 200 + i + // Read times run backwards from the ids, as a newest-first import would make them. + const fetched = new Date(Date.UTC(2026, 10, 5) - n * 60_000).toISOString() + return entry(`https://a.example/${n}`, `Page ${n}`, `${fat}${n === 0 ? ' zebraomega' : ''}${n === 999 ? ' aardvarkzero' : ''}`, { fetched_at: fetched }) + }), + ) + expect(await fill.count()).toBe(1_000) + const grown = await fill.sizeBytes(0) + expect(grown).toBeGreaterThan(0) + // Inside the budget: nothing goes. + expect(await (await open(grown * 2, {}, url)).trim()).toMatchObject({ deleted: 0, done: true }) + expect(await (await open(0, {}, url)).trim()).toMatchObject({ deleted: 0, done: true }) + + const store = await open(Math.floor(grown / 2), {}, url) + const result = await store.trim({ batch: 100, keepMin: 100 }) + expect(result.deleted).toBeGreaterThanOrEqual(490) + expect(result.deleted).toBeLessThanOrEqual(510) + expect(result.done).toBe(true) + expect(result.liveBytes!).toBeLessThanOrEqual(store.maxBytes) + expect(await store.count(0)).toBe(1_000 - result.deleted) + expect(await store.resync()).toBe(1_000 - result.deleted) + // The page read longest ago (the highest id here) went; the newest stayed. + expect(await store.search('zebraomega')).toHaveLength(1) + expect(await store.search('aardvarkzero')).toHaveLength(0) + // A second sweep with nothing new does not keep deleting: the files did not shrink, but the live pages fit. + expect((await store.trim({ batch: 100, keepMin: 100 })).deleted).toBe(0) + + // A budget too small to meet leaves a working index rather than an empty one. + const tiny = await open(4096, {}, url) + const floor = await tiny.trim({ batch: 50, keepMin: 100 }) + expect(floor.done).toBe(false) + expect(await tiny.count(0)).toBe(100) + expect(await tiny.search('zebraomega')).toHaveLength(1) + }, 60_000) + + test('/stats figures are cached and refreshed behind the cache', async () => { + const store = await open(1024 ** 3) + await store.add([entry('https://a.example/', 'First', 'words')]) + const size = await store.sizeBytes() + expect(await store.count()).toBe(1) + await store.add(Array.from({ length: 50 }, (_, i) => entry(`https://b.example/${i}`, `Page ${i}`, 'lorem ipsum '.repeat(300)))) + // The cached size answers at once; a fresh read sees the growth. + expect(await store.sizeBytes()).toBe(size) + expect(await store.sizeBytes(0)).toBeGreaterThanOrEqual(size) + expect(await store.count()).toBe(51) + }) + + test('migrate-index: a SQLite index streams in with ids, read times, verdicts and cards, resumably', async () => { + const path = `${tmpdir()}/pit-migrate-${Date.now()}.db` + const sqlite = new IndexStore(path) + sqlite.add( + Array.from({ length: 30 }, (_, i) => + entry(`https://m${i % 4}.example/${i}`, `Migrated ${i}`, `moved words ${i}`, { + fetched_at: new Date(Date.UTC(2026, 9, 1) + i * 3_600_000).toISOString(), + training: i % 5 === 0 ? { allowed: false, reason: 'robots' } : i % 7 === 0 ? null : OPEN, + }), + ), + ) + sqlite.add([{ ...entry('https://shop.example/p', 'A product', 'buy it', { kind: 'product' }), enrichment: { excerpt: 'buy it', price: '$9', training: OPEN } }]) + sqlite.setPolicies('m0.example', []) + // Gaps in the ids, as trim leaves them. + ;(sqlite as unknown as { db: { run(q: string): void } }).db.run('DELETE FROM pages WHERE id IN (3, 4, 10)') + const ids = (sqlite as unknown as { db: { query(q: string): { all(): Array<{ id: number; url: string; fetched_at: string }> } } }).db + .query('SELECT id, url, fetched_at FROM pages ORDER BY id') + .all() + sqlite.close() + + const url = await fresh() + const pg = await open(0, {}, url) + // The crawl already wrote one of these pages to Postgres, more freshly: it wins. + await pg.add([entry('https://m1.example/1', 'Fresh from the crawl', 'newer words')]) + const result = await migrateIndexFromSqlite(path, pg, { batch: 7 }) + expect(result.read).toBe(ids.length) + expect(result.inserted).toBe(ids.length - 1) + expect(result.pages).toBe(ids.length) + expect(await pg.count(0)).toBe(ids.length) + const rows = (await (pg as unknown as { sql: SQL }).sql`select id::int as id, url, fetched_at from pages order by id`) as Array<{ id: number; url: string; fetched_at: string }> + const byUrl = new Map(rows.map((r) => [r.url, r])) + // Ids and read times kept, except where the crawl's earlier page had taken the id (id 1): + // that page still comes over, under a fresh id above the migrated range. + const maxMigrated = Math.max(...ids.map((r) => r.id)) + for (const r of ids) { + if (r.url === 'https://m1.example/1') continue + if (r.id === 1) expect(byUrl.get(r.url)!.fetched_at).toBe(r.fetched_at) + else expect(byUrl.get(r.url)).toEqual(r) + } + expect(byUrl.get(ids.find((r) => r.id === 1)!.url)!.id).toBeGreaterThan(maxMigrated) + expect((await pg.search('newer'))[0]!.url).toBe('https://m1.example/1') + expect(await pg.search('moved words 29')).not.toHaveLength(0) + // Verdicts came over: reserved pages stay out of the corpus, unknown ones too. + const open1 = (await pg.corpus({ q: 'moved', limit: 100 })).items.map((i) => i.url) + expect(open1).not.toContain('https://m0.example/0') + expect(open1).not.toContain('https://m3.example/7') + expect(open1).toContain('https://m2.example/6') + expect((await pg.cards(['https://shop.example/p'])).get('https://shop.example/p')).toEqual({ image: null, price: '$9' }) + // Run again: the cursor says it is done, nothing is read twice. + const again = await migrateIndexFromSqlite(path, pg, { batch: 7 }) + expect(again.read).toBe(0) + expect(await pg.count(0)).toBe(ids.length) + // A page written after the move takes an id above every migrated one. + await pg.add([entry('https://after.example/', 'After', 'later')]) + const top = (await (pg as unknown as { sql: SQL }).sql`select id::int as id from pages where url = 'https://after.example/'`) as { id: number }[] + expect(top[0]!.id).toBeGreaterThan(Math.max(...ids.map((r) => r.id))) + // --restart reads it all again; pages already there stay as they are. + const restarted = await migrateIndexFromSqlite(path, pg, { batch: 50, restart: true }) + expect(restarted).toMatchObject({ read: ids.length, inserted: 0 }) + for (const suffix of ['', '-wal', '-shm']) rmSync(`${path}${suffix}`, { force: true }) + }, 60_000) +}) From 46b7c1768301af9d24d8b8fbdf3012c9aab8df78 Mon Sep 17 00:00:00 2001 From: Anthony Ettinger Date: Sun, 8 Nov 2026 16:52:46 +0000 Subject: [PATCH 2/2] crawler: mark Bun sql tagged templates in pg-index-store as bound (threatcrush) Each flagged line is Bun's sql tagged template, which binds every ${} as a parameter; nothing is concatenated. Co-Authored-By: Claude Opus 5.5 --- services/crawler/src/pg-index-store.ts | 9 +++++++-- 1 file changed, 7 insertions(+), 2 deletions(-) diff --git a/services/crawler/src/pg-index-store.ts b/services/crawler/src/pg-index-store.ts index 95df8ae..79b43cc 100644 --- a/services/crawler/src/pg-index-store.ts +++ b/services/crawler/src/pg-index-store.ts @@ -224,12 +224,14 @@ export class PgIndexStore implements SearchIndex { private async meta(key: string): Promise { await this.init() + // threatcrush-disable-next-line sql-template-interpolation Bun's sql tagged template binds ${} as parameters; nothing is concatenated const rows = (await this.sql`select n from index_meta where k = ${key}`) as { n: string | number }[] return rows.length ? Number(rows[0]!.n) : null } private async setMeta(key: string, value: number): Promise { await this.init() + // threatcrush-disable-next-line sql-template-interpolation Bun's sql tagged template binds ${} as parameters; nothing is concatenated await this.sql`insert into index_meta (k, n) values (${key}, ${value}) on conflict (k) do update set n = excluded.n` } @@ -366,6 +368,7 @@ export class PgIndexStore implements SearchIndex { async setCursor(key: string, value: number | null): Promise { if (value === null) { await this.init() + // threatcrush-disable-next-line sql-template-interpolation Bun's sql tagged template binds ${} as parameters; nothing is concatenated await this.sql`delete from index_meta where k = ${key}` } else await this.setMeta(key, value) } @@ -669,6 +672,7 @@ export class PgIndexStore implements SearchIndex { async unknownPagesOf(site: string): Promise { await this.init() + // threatcrush-disable-next-line sql-template-interpolation Bun's sql tagged template binds ${} as parameters; nothing is concatenated const rows = (await this.sql`select pp.id, p.url from page_policy pp join pages p on p.id = pp.id where pp.site = ${site} and pp.train is null`) as Array<{ id: string | number; url: string }> return rows.map((r) => ({ id: Number(r.id), url: r.url })) } @@ -688,8 +692,8 @@ export class PgIndexStore implements SearchIndex { async setCard(url: string, image: string | null, price: string | null): Promise { if (Buffer.byteLength(url) > MAX_URL_BYTES) return await this.init() - await this - .sql`insert into page_cards (url, image, price) values (${clean(url)}, ${cleanOrNull(image)}, ${cleanOrNull(price)}) on conflict (url) do update set image = excluded.image, price = excluded.price` + // threatcrush-disable-next-line sql-template-interpolation Bun's sql tagged template binds ${} as parameters; nothing is concatenated + await this.sql`insert into page_cards (url, image, price) values (${clean(url)}, ${cleanOrNull(image)}, ${cleanOrNull(price)}) on conflict (url) do update set image = excluded.image, price = excluded.price` } /** Cards for these pages. A page absent from the map has never been looked at. */ @@ -697,6 +701,7 @@ export class PgIndexStore implements SearchIndex { const out = new Map() if (!urls.length) return out await this.init() + // threatcrush-disable-next-line sql-template-interpolation Bun's sql tagged template binds ${} as parameters; nothing is concatenated const rows = (await this.sql`select url, image, price from page_cards where url = any(${pgArray(urls.map(clean))}::text[])`) as Array<{ url: string; image: string | null; price: string | null }> for (const r of rows) out.set(r.url, { image: r.image, price: r.price }) return out