A rollout should cost a source two minutes, not a day

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",

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');

apps/web/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@nichedb/web",
-  "version": "0.4.0",
+  "version": "0.5.0",
   "private": true,
   "type": "module",
   "scripts": {

apps/worker/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@nichedb/worker",
-  "version": "0.4.0",
+  "version": "0.5.0",
   "private": true,
   "type": "module",
   "scripts": {

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.",

packages/adapters/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@nichedb/adapters",
-  "version": "0.4.0",
+  "version": "0.5.0",
   "private": true,
   "type": "module",
   "exports": {

packages/auth/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@nichedb/auth",
-  "version": "0.4.0",
+  "version": "0.5.0",
   "private": true,
   "type": "module",
   "exports": {

packages/config/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@nichedb/config",
-  "version": "0.4.0",
+  "version": "0.5.0",
   "private": true,
   "type": "module",
   "exports": {

packages/core/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@nichedb/core",
-  "version": "0.4.0",
+  "version": "0.5.0",
   "private": true,
   "type": "module",
   "exports": {

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' };
   }

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 };
 }

packages/db/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@nichedb/db",
-  "version": "0.4.0",
+  "version": "0.5.0",
   "private": true,
   "type": "module",
   "exports": {

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

packages/enrichers/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@nichedb/enrichers",
-  "version": "0.4.0",
+  "version": "0.5.0",
   "private": true,
   "type": "module",
   "exports": {

packages/knowledge/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@nichedb/knowledge",
-  "version": "0.4.0",
+  "version": "0.5.0",
   "private": true,
   "type": "module",
   "exports": {

packages/notify/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@nichedb/notify",
-  "version": "0.4.0",
+  "version": "0.5.0",
   "private": true,
   "type": "module",
   "exports": {

packages/payments/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@nichedb/payments",
-  "version": "0.4.0",
+  "version": "0.5.0",
   "private": true,
   "type": "module",
   "exports": {

packages/queue/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@nichedb/queue",
-  "version": "0.4.0",
+  "version": "0.5.0",
   "private": true,
   "type": "module",
   "exports": {

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);
+  });
+});