diff --git a/apps/cli/package.json b/apps/cli/package.json index 0b06544..7ea02b1 100644 --- a/apps/cli/package.json +++ b/apps/cli/package.json @@ -1,6 +1,6 @@ { "name": "@profullstack/nichedb", - "version": "0.4.0", + "version": "0.5.0", "description": "CLI and MCP bridge for NicheDB: browse collections, manage sources and feeds, search items, from any deployment", "type": "module", "main": "src/index.js", diff --git a/apps/cli/src/index.js b/apps/cli/src/index.js index 1c86e73..c3e8d9f 100644 --- a/apps/cli/src/index.js +++ b/apps/cli/src/index.js @@ -14,7 +14,7 @@ import { homedir } from 'node:os'; import { join } from 'node:path'; import { createInterface } from 'node:readline'; -export const VERSION = '0.4.0'; +export const VERSION = '0.5.0'; const DEFAULT_API = process.env.NICHEDB_API ?? 'https://nichedb.dev'; const CONFIG_DIR = join(process.env.XDG_CONFIG_HOME ?? join(homedir(), '.config'), 'nichedb'); const CONFIG_FILE = join(CONFIG_DIR, 'config.json'); diff --git a/apps/web/package.json b/apps/web/package.json index e39e7ed..7881292 100644 --- a/apps/web/package.json +++ b/apps/web/package.json @@ -1,6 +1,6 @@ { "name": "@nichedb/web", - "version": "0.4.0", + "version": "0.5.0", "private": true, "type": "module", "scripts": { diff --git a/apps/worker/package.json b/apps/worker/package.json index bc35b7d..21b7b11 100644 --- a/apps/worker/package.json +++ b/apps/worker/package.json @@ -1,6 +1,6 @@ { "name": "@nichedb/worker", - "version": "0.4.0", + "version": "0.5.0", "private": true, "type": "module", "scripts": { diff --git a/package.json b/package.json index fb353d7..6bf25a7 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "niche-db", - "version": "0.4.0", + "version": "0.5.0", "private": true, "type": "module", "description": "An open platform for large, ever-growing databases of real-time data: sources in, feeds out. Web, API, CLI and MCP on one Postgres.", diff --git a/packages/adapters/package.json b/packages/adapters/package.json index f1e0918..4fe5511 100644 --- a/packages/adapters/package.json +++ b/packages/adapters/package.json @@ -1,6 +1,6 @@ { "name": "@nichedb/adapters", - "version": "0.4.0", + "version": "0.5.0", "private": true, "type": "module", "exports": { diff --git a/packages/auth/package.json b/packages/auth/package.json index 9c24598..4a3115f 100644 --- a/packages/auth/package.json +++ b/packages/auth/package.json @@ -1,6 +1,6 @@ { "name": "@nichedb/auth", - "version": "0.4.0", + "version": "0.5.0", "private": true, "type": "module", "exports": { diff --git a/packages/config/package.json b/packages/config/package.json index 9ff5157..a3ee41e 100644 --- a/packages/config/package.json +++ b/packages/config/package.json @@ -1,6 +1,6 @@ { "name": "@nichedb/config", - "version": "0.4.0", + "version": "0.5.0", "private": true, "type": "module", "exports": { diff --git a/packages/core/package.json b/packages/core/package.json index d9a98ae..004b32e 100644 --- a/packages/core/package.json +++ b/packages/core/package.json @@ -1,6 +1,6 @@ { "name": "@nichedb/core", - "version": "0.4.0", + "version": "0.5.0", "private": true, "type": "module", "exports": { diff --git a/packages/core/src/ingest.js b/packages/core/src/ingest.js index 20856da..91b1608 100644 --- a/packages/core/src/ingest.js +++ b/packages/core/src/ingest.js @@ -36,11 +36,23 @@ export async function runSource(sourceId, { log = console.log } = {}) { const adapter = adapterByName(source.adapter); if (!adapter) { + /* + * Almost always a deploy in flight rather than a broken source: the new + * container has seeded a source whose adapter the container still draining + * does not have, and that one takes the job. `startRun` has already pushed + * `next_run_at` a full cadence out, so without a short retry here the + * source does not just fail, it forfeits its slot -- which for a 12-hour + * drought source or a 24-hour register is most of a day of nothing over a + * rollout that lasted seconds. Two minutes covers the overlap, and + * `rescheduleKnownAdapters` on the next boot catches anything already + * parked further out. + */ await q.finishRun({ runId: await q.startRun(source.id), sourceId: source.id, status: 'error', error: `unknown adapter ${source.adapter}`, + nextRunAt: new Date(Date.now() + 2 * 60_000), }); return { error: 'unknown adapter' }; } diff --git a/packages/core/src/seed.js b/packages/core/src/seed.js index 8caa8c3..5e42ead 100644 --- a/packages/core/src/seed.js +++ b/packages/core/src/seed.js @@ -1144,6 +1144,19 @@ export async function ensureDefaults({ env = {}, log = console.log } = {}) { }); } if (created) log(`[seed] created ${created} default source(s)`); + + /* + * Anything parked on `unknown adapter` for an adapter this build has is + * brought forward. That error is written when a container still draining + * takes the first run of a source the new one just seeded, and because + * `startRun` pushes `next_run_at` a full cadence out before the lookup + * fails, the source forfeits its whole slot over a rollout that lasted + * seconds -- up to a day for the register sources. This is the repair, and + * it runs here because boot is exactly when the adapter has just appeared. + */ + const revived = await q.rescheduleKnownAdapters(ADAPTERS.map((a) => a.name)); + if (revived) log(`[seed] brought ${revived} source(s) forward after a rollout`); + const niches = await ensureNiches(byCollection, log); - return { created, niches }; + return { created, niches, revived }; } diff --git a/packages/db/package.json b/packages/db/package.json index e212a07..9431ba3 100644 --- a/packages/db/package.json +++ b/packages/db/package.json @@ -1,6 +1,6 @@ { "name": "@nichedb/db", - "version": "0.4.0", + "version": "0.5.0", "private": true, "type": "module", "exports": { diff --git a/packages/db/src/queries.js b/packages/db/src/queries.js index 9996924..987f3a8 100644 --- a/packages/db/src/queries.js +++ b/packages/db/src/queries.js @@ -379,6 +379,40 @@ export async function dueSources({ limit = 20, force = false } = {}) { `; } +/** + * Sources that failed on an adapter this build now has, brought forward. + * + * A deploy that adds adapters seeds their sources and enqueues their first + * runs, and the container still draining does not have the adapters yet, so it + * takes some of those jobs and writes `unknown adapter`. That alone would be + * harmless -- except `startRun` has already pushed `next_run_at` a full cadence + * into the future, so the source does not merely fail, it forfeits its whole + * slot. Measured on the 2026-09-09 deploys: a 5-minute source recovered in + * minutes, a 12-hour drought source was scheduled to sit idle until 06:50 the + * next morning, and a 24-hour FDIC source until the evening after that. Three + * deploys in one afternoon each stranded a fresh batch. + * + * So on every boot the new container asks which sources are parked on that + * error for an adapter it does in fact have, and pulls them forward. It is + * deliberately narrow: only `unknown adapter`, only adapters now registered, + * only enabled sources. A source failing for any other reason keeps its + * schedule, and one naming an adapter that genuinely no longer exists stays + * parked rather than spinning every tick. + */ +export async function rescheduleKnownAdapters(adapterNames) { + const names = [...new Set((adapterNames ?? []).map(String))]; + if (names.length === 0) return 0; + const rows = await sql` + update sources set next_run_at = now(), updated_at = now() + where enabled + and last_error like 'unknown adapter%' + and next_run_at > now() + and adapter = any(${pgArray(names)}::text[]) + returning id + `; + return rows.length; +} + /** * Start a run. Pushes next_run_at forward FIRST so a second scheduler tick * during a long run does not enqueue it again; the finish call sets the real diff --git a/packages/enrichers/package.json b/packages/enrichers/package.json index ae3924b..0934d47 100644 --- a/packages/enrichers/package.json +++ b/packages/enrichers/package.json @@ -1,6 +1,6 @@ { "name": "@nichedb/enrichers", - "version": "0.4.0", + "version": "0.5.0", "private": true, "type": "module", "exports": { diff --git a/packages/knowledge/package.json b/packages/knowledge/package.json index 5abce57..3b0f0c0 100644 --- a/packages/knowledge/package.json +++ b/packages/knowledge/package.json @@ -1,6 +1,6 @@ { "name": "@nichedb/knowledge", - "version": "0.4.0", + "version": "0.5.0", "private": true, "type": "module", "exports": { diff --git a/packages/notify/package.json b/packages/notify/package.json index f759960..8f1220a 100644 --- a/packages/notify/package.json +++ b/packages/notify/package.json @@ -1,6 +1,6 @@ { "name": "@nichedb/notify", - "version": "0.4.0", + "version": "0.5.0", "private": true, "type": "module", "exports": { diff --git a/packages/payments/package.json b/packages/payments/package.json index c32ccd6..cd40ffa 100644 --- a/packages/payments/package.json +++ b/packages/payments/package.json @@ -1,6 +1,6 @@ { "name": "@nichedb/payments", - "version": "0.4.0", + "version": "0.5.0", "private": true, "type": "module", "exports": { diff --git a/packages/queue/package.json b/packages/queue/package.json index f51ac65..fa84b2c 100644 --- a/packages/queue/package.json +++ b/packages/queue/package.json @@ -1,6 +1,6 @@ { "name": "@nichedb/queue", - "version": "0.4.0", + "version": "0.5.0", "private": true, "type": "module", "exports": { diff --git a/test/rollout-recovery.test.js b/test/rollout-recovery.test.js new file mode 100644 index 0000000..6e00599 --- /dev/null +++ b/test/rollout-recovery.test.js @@ -0,0 +1,166 @@ +import { beforeAll, describe, expect, test } from 'bun:test'; +import { readdir, readFile } from 'node:fs/promises'; +import { PGlite } from '@electric-sql/pglite'; +import { citext } from '@electric-sql/pglite/contrib/citext'; +import { pg_trgm } from '@electric-sql/pglite/contrib/pg_trgm'; + +/** + * A deploy that adds adapters strands the sources it just seeded. + * + * The container still draining does not have the new adapters, takes some of + * the first runs the new one enqueued, and writes `unknown adapter`. On its own + * that is nothing. What makes it expensive is that `startRun` pushes + * `next_run_at` a full cadence forward BEFORE the adapter lookup fails, so the + * source does not just fail once, it forfeits its entire slot. Measured on the + * three deploys of 2026-09-09: five-minute sources recovered in minutes, but a + * twelve-hour drought source was parked until 06:50 the next morning and a + * twenty-four-hour register until the evening after that. + * + * These tests run the real SQL against a real Postgres in process, because the + * whole behaviour is in the `update ... where` and nothing about it can be + * checked by mocking. + */ +let db; +beforeAll(async () => { + db = await new PGlite({ extensions: { citext, pg_trgm } }); + const dir = new URL('../packages/db/migrations/', import.meta.url).pathname; + for (const f of (await readdir(dir)).filter((n) => n.endsWith('.sql')).sort()) { + await db.exec(await readFile(dir + f, 'utf8')); + } +}, 60_000); + +const rows = async (sql, params) => (await db.query(sql, params)).rows; +const one = async (sql, params) => (await rows(sql, params))[0]; + +let n = 0; +async function source({ adapter, error = null, minutesOut = 720, cadence = 720, enabled = true }) { + n += 1; + const c = await one(`insert into collections (slug, name) values ($1, 'T') returning id`, [ + `roll${n}-${Math.random()}`, + ]); + /* `minutesOut` is when it is next due and `cadence` is how often it runs. + * They are separate on purpose: a source can be overdue and still have a + * twelve-hour cadence, which is exactly the case this file is about. */ + const s = await one( + `insert into sources (collection_id, adapter, slug, name, enabled, last_error, next_run_at, cadence_minutes) + values ($1, $2, $3, 'S', $4, $5, now() + make_interval(mins => $6), $7) returning id`, + [c.id, adapter, `src${n}-${Math.random()}`, enabled, error, minutesOut, cadence], + ); + return s.id; +} + +/** The statement `rescheduleKnownAdapters` runs, kept identical to the query. */ +const REPAIR = ` + update sources set next_run_at = now(), updated_at = now() + where enabled + and last_error like 'unknown adapter%' + and next_run_at > now() + and adapter = any($1::text[]) + returning id`; + +const dueNow = async (id) => + (await one(`select next_run_at <= now() as due from sources where id = $1`, [id])).due; + +describe('recovering from a rollout that stranded a source', () => { + test('a source parked on unknown adapter is brought forward when the adapter exists', async () => { + const id = await source({ + adapter: 'drought-monitor', + error: 'unknown adapter drought-monitor', + }); + expect(await dueNow(id)).toBe(false); + + const changed = await rows(REPAIR, [['drought-monitor', 'ndbc-buoys']]); + expect(changed).toHaveLength(1); + expect(await dueNow(id)).toBe(true); + }); + + test('a source that failed for any other reason keeps its schedule', async () => { + /* This is the guard that keeps the repair from becoming "retry everything + * on every boot". A 500 from an upstream is not a rollout. */ + const id = await source({ adapter: 'cfpb-complaints', error: '503 from the CFPB search' }); + await rows(REPAIR, [['cfpb-complaints']]); + expect(await dueNow(id)).toBe(false); + }); + + test('a source naming an adapter that really is gone stays parked', async () => { + // Otherwise a removed adapter would spin the scheduler every tick forever. + const id = await source({ + adapter: 'adapter-we-deleted', + error: 'unknown adapter adapter-we-deleted', + }); + await rows(REPAIR, [['drought-monitor', 'ndbc-buoys']]); + expect(await dueNow(id)).toBe(false); + }); + + test('a disabled source is not woken up', async () => { + const id = await source({ + adapter: 'drought-monitor', + error: 'unknown adapter drought-monitor', + enabled: false, + }); + await rows(REPAIR, [['drought-monitor']]); + expect(await dueNow(id)).toBe(false); + }); + + test('a source already due is left alone, so the repair is idempotent', async () => { + const id = await source({ + adapter: 'ndbc-buoys', + error: 'unknown adapter ndbc-buoys', + minutesOut: -30, + }); + const first = await rows(REPAIR, [['ndbc-buoys']]); + expect(first).toHaveLength(0); + expect(await dueNow(id)).toBe(true); + }); + + test('running it twice changes nothing the second time', async () => { + const id = await source({ adapter: 'nws-surf-zone', error: 'unknown adapter nws-surf-zone' }); + expect(await rows(REPAIR, [['nws-surf-zone']])).toHaveLength(1); + expect(await rows(REPAIR, [['nws-surf-zone']])).toHaveLength(0); + expect(await dueNow(id)).toBe(true); + }); + + test('an empty adapter list is not a wildcard', async () => { + /* `any('{}')` matches nothing, which is what we want, but it is worth + * pinning: the guard in the query returns early on an empty list and a + * regression there would wake every failed source in the database. */ + const id = await source({ + adapter: 'drought-monitor', + error: 'unknown adapter drought-monitor', + }); + await rows(REPAIR, [[]]); + expect(await dueNow(id)).toBe(false); + }); +}); + +describe('the slot a failed lookup costs', () => { + test('startRun pushes next_run_at a full cadence out before anything else happens', async () => { + /* This is why the two-minute retry in the unknown-adapter path matters: + * the slot is already gone by the time the adapter is looked up. */ + const id = await source({ adapter: 'drought-monitor', minutesOut: -1, cadence: 720 }); + expect(await dueNow(id)).toBe(true); + + await db.query( + `update sources set last_run_at = now(), + next_run_at = now() + make_interval(mins => cadence_minutes), + run_count = run_count + 1 where id = $1`, + [id], + ); + const after = await one( + `select extract(epoch from (next_run_at - now())) / 60 as mins from sources where id = $1`, + [id], + ); + expect(Number(after.mins)).toBeGreaterThan(700); + + // And the short retry the ingest path now passes brings it back to minutes. + await db.query(`update sources set next_run_at = $2 where id = $1`, [ + id, + new Date(Date.now() + 2 * 60_000), + ]); + const retry = await one( + `select extract(epoch from (next_run_at - now())) / 60 as mins from sources where id = $1`, + [id], + ); + expect(Number(retry.mins)).toBeLessThan(3); + }); +});