ralyodio · Nov 8, 2026 · Nov 8, 2026 · Nov 8, 2026
diff --git 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
@@ -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 <path>`
+(resumable; ids and read times kept). Each publishing process writes its own archive file,
+`pages-<hour>-<host>-<pid>.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
@@ -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-<host>-<pid>.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
  * `<name>.partial`; it is renamed to `<name>` 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-<host>-<pid>.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-<hour>[-<tag>].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
@@ -12,6 +12,7 @@
  *   pit-crawler seed --file <path>           one site per line
  *   pit-crawler stats                        queue depths and counters, as JSON
  *   pit-crawler read <url> [--render]        read one page now and print its record (no Redis)
+ *   pit-crawler migrate-index --from-sqlite <path>   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<typeof setTimeout> | 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 <path> [--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<string, (args: string[]) => Promise<void>> = { worker, seed, social, import: importCmd, stats: () => stats(), read, 'migrate-sites': migrateSites }
+const commands: Record<string, (args: string[]) => Promise<void>> = { 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 <site...> [--from nichedb|ring] [--file path] [--limit n] | import [--from nichedb] [--limit n] [--max-gb n] [--restart] | stats | read <url> [--render] | migrate-sites [--drop]',
+    'usage: pit-crawler worker | seed <site...> [--from nichedb|ring] [--file path] [--limit n] | import [--from nichedb] [--limit n] [--max-gb n] [--restart] | stats | read <url> [--render] | migrate-sites [--drop] | migrate-index --from-sqlite <path> [--batch n] [--restart]',
   )
   process.exit(command ? 1 : 0)
 }

diff --git 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<string, string | undefined> = 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,