From 1a3985593e2324dab4dbc80cf25a9c4710b1df9b Mon Sep 17 00:00:00 2001 From: Anthony Ettinger Date: Wed, 9 Sep 2026 19:27:28 +0000 Subject: [PATCH 1/2] 0.5.0: buoys, surf forecasts and aircraft on watch Since 0.4.0: #40 added three keyless sources. NDBC's every-buoy file into water, with wave height, period and direction read out the way a surf report says them; the National Weather Service surf zone forecast from every coastal office beside it; and the ADS-B emergency squawk and military watch lists into aviation, held as episodes rather than as positions. Twenty-seven collections, seventy-nine adapters. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01CjTtJPEpyJbvPeQcVfYPPs --- apps/cli/package.json | 2 +- apps/cli/src/index.js | 2 +- apps/web/package.json | 2 +- apps/worker/package.json | 2 +- package.json | 2 +- packages/adapters/package.json | 2 +- packages/auth/package.json | 2 +- packages/config/package.json | 2 +- packages/core/package.json | 2 +- packages/db/package.json | 2 +- packages/enrichers/package.json | 2 +- packages/knowledge/package.json | 2 +- packages/notify/package.json | 2 +- packages/payments/package.json | 2 +- packages/queue/package.json | 2 +- 15 files changed, 15 insertions(+), 15 deletions(-) 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/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/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": { From 7dfc0c397e06ee6a93688c1a2724b533ea7776fc Mon Sep 17 00:00:00 2001 From: Anthony Ettinger Date: Wed, 9 Sep 2026 19:39:53 +0000 Subject: [PATCH 2/2] A rollout should cost a source two minutes, not a day Three deploys this afternoon each stranded a batch of freshly seeded sources, and the reason is worse than the error text suggests. When a deploy adds adapters, the new container seeds their sources and enqueues the first runs, and the container still draining takes some of those jobs. It does not have the new adapters, so `runSource` writes `unknown adapter`. That part is harmless and self-correcting. What is not harmless is that `startRun` pushes `next_run_at` a full cadence forward BEFORE the adapter is looked up. So the source does not fail and retry. It forfeits its whole slot. Measured on the live database just now: every 5m recovered within minutes (aircraft-emergency, run 2, 1 item) every 60m next attempt 19:50 (~1 hour of nothing) every 180m next attempt 21:50 (3 hours) every 720m next attempt 06:50 TOMORROW (11 hours) every 1440m next attempt 18:50 TOMORROW (23 hours) Twenty-four of thirty-six new sources were sitting in that state, most of a day of silence bought by a rollout that lasted seconds. "It heals on the next run" was true and almost useless. Two changes. The unknown-adapter path now asks to be called back in two minutes instead of letting the forfeited slot stand. Two minutes covers a rollout overlap and costs nothing if the adapter really is gone, because the run after that parks it again. And `ensureDefaults` now brings forward anything already parked on that error for an adapter this build actually has. Boot is exactly the moment the adapter appears, so it is the right place to ask. The query is deliberately narrow: only `unknown adapter`, only adapters now registered, only enabled sources, only ones not already due. A source that failed for any other reason keeps its schedule, and one naming an adapter that genuinely no longer exists stays parked rather than spinning the scheduler every tick. Both are tested against a real Postgres in process rather than mocked, because the entire behaviour lives in an `update ... where` and there is nothing else to check. Eight tests: the repair works, it is idempotent, it ignores other errors, it ignores deleted adapters, it ignores disabled sources, an empty adapter list is not a wildcard, and the full-cadence forfeit is pinned so nobody removes the two-minute retry without a test going red. 604 tests pass, biome clean. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01CjTtJPEpyJbvPeQcVfYPPs --- packages/core/src/ingest.js | 12 +++ packages/core/src/seed.js | 15 ++- packages/db/src/queries.js | 34 +++++++ test/rollout-recovery.test.js | 166 ++++++++++++++++++++++++++++++++++ 4 files changed, 226 insertions(+), 1 deletion(-) create mode 100644 test/rollout-recovery.test.js 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/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/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); + }); +});