Keep cron fire history out of scheduler scans

src/api/app-messaging.ts

@@ -31,6 +31,7 @@ export function createMessagingMethods(
   App,
   | "createCron"
   | "getCron"
+  | "getCronRuns"
   | "listCrons"
   | "listCronsForViewer"
   | "updateCron"
@@ -113,6 +114,9 @@ export function createMessagingMethods(
     getCron(id) {
       return deps.crons.get(id);
     },
+    getCronRuns(id, limit) {
+      return deps.crons.getRuns(id, limit);
+    },
     listCrons() {
       return deps.crons.list();
     },

src/api/app-types.ts

@@ -41,6 +41,7 @@ import type { CapabilityClaims } from "../auth/capability-token.ts";
 import type { ScopedConfigStore } from "../resolution/config-store.ts";
 import { type AdminService } from "../admin/admin-service.ts";
 import type { CronStore, CreateCronInput, CronPatch } from "../cron/cron-store.ts";
+import type { CronFirePage } from "../cron/cron-fire-store.ts";
 import type { WebhookStore, CreateWebhookInput } from "../webhooks/webhook-store.ts";
 import type { DeliveryStore } from "../delivery/delivery-store.ts";
 import type {
@@ -348,6 +349,7 @@ export interface App {
   ): Promise<number>;
   createCron(input: CreateCronInput): Promise<Cron>;
   getCron(id: string): Promise<Cron | null>;
+  getCronRuns(id: string, limit?: number): Promise<CronFirePage>;
   listCrons(): Promise<Cron[]>;
   listCronsForViewer(principalId: string): Promise<{ owned: Cron[]; visible: VisibleCron[] }>;
   updateCron(id: string, patch: CronPatch): Promise<Cron | null>;

src/api/control-service.ts

@@ -564,9 +564,8 @@ export function createControlService(app: App, scheduler?: Scheduler, admin?: Ad
       if (req.limit !== undefined && (!Number.isInteger(req.limit) || req.limit < 1)) {
         return { ok: false, code: "bad_request", message: "limit must be a positive integer" };
       }
-      const fireLog = cron.fireLog ?? [];
-      const runs = req.limit !== undefined ? fireLog.slice(-req.limit) : fireLog;
-      return { ok: true, cron, runs, total: fireLog.length };
+      const { runs, total } = await app.getCronRuns(id, req.limit);
+      return { ok: true, cron, runs, total };
     },
 
     async patchCron(id, req, capability) {

src/api/routes/crons.ts

@@ -332,7 +332,7 @@ async function runCronNow(ctx: ApiCtx): Promise<void> {
 }
 
 async function cronRuns(ctx: ApiCtx): Promise<void> {
-  const { res, url, capability } = ctx;
+  const { res, url, capability, app } = ctx;
   const id = ctx.params.id!;
   const rawLimit = url.searchParams.get("limit");
   const limit = rawLimit === null ? undefined : Number(rawLimit);
@@ -346,9 +346,8 @@ async function cronRuns(ctx: ApiCtx): Promise<void> {
   if (limit !== undefined && (!Number.isInteger(limit) || limit < 1)) {
     return sendJson(res, 400, { error: "bad_request", message: "limit must be a positive integer" });
   }
-  const fireLog = cron.fireLog ?? [];
-  const runs = limit !== undefined ? fireLog.slice(-limit) : fireLog;
-  return sendJson(res, 200, { cron: withoutFireLog(cron), runs, total: fireLog.length });
+  const { runs, total } = await app.getCronRuns(id, limit);
+  return sendJson(res, 200, { cron: withoutFireLog(cron), runs, total });
 }
 
 const CRON_PATCH_BAD_REQUEST =

src/cron/cron-fire-store.ts

@@ -0,0 +1,146 @@
+import type { CronFireLogEntry } from "../types.ts";
+import { jsonbStringify } from "../persistence/durable-map.ts";
+import type { PgPool } from "../persistence/pg-pool.ts";
+
+export interface CronFirePage {
+  runs: CronFireLogEntry[];
+  total: number;
+}
+
+export interface CronFireStore {
+  ready(): Promise<void>;
+  drainInline?(cronId?: string): Promise<void>;
+  record(cronId: string, entry: CronFireLogEntry): Promise<void>;
+  import(cronId: string, entries: CronFireLogEntry[]): Promise<void>;
+  list(cronId: string, limit?: number): Promise<CronFirePage>;
+  delete(cronId: string): Promise<void>;
+}
+
+export function createMemoryCronFireStore(): CronFireStore {
+  const byCron = new Map<string, Map<string, CronFireLogEntry>>();
+  const record = (cronId: string, entry: CronFireLogEntry) => {
+    let entries = byCron.get(cronId);
+    if (!entries) {
+      entries = new Map();
+      byCron.set(cronId, entries);
+    }
+    entries.set(entry.fireKey, { ...entries.get(entry.fireKey), ...entry });
+  };
+  return {
+    async ready() {},
+    async record(cronId, entry) {
+      record(cronId, entry);
+    },
+    async import(cronId, entries) {
+      for (const entry of entries) record(cronId, entry);
+    },
+    async list(cronId, limit) {
+      const all = [...(byCron.get(cronId)?.values() ?? [])].sort(
+        (a, b) => a.firedAt - b.firedAt || a.fireKey.localeCompare(b.fireKey),
+      );
+      return { runs: limit === undefined ? all : all.slice(-limit), total: all.length };
+    },
+    async delete(cronId) {
+      byCron.delete(cronId);
+    },
+  };
+}
+
+export function createPostgresCronFireStore(
+  pg: PgPool,
+  cronTable = "crons",
+  fireTable = "cron_fire_log",
+): CronFireStore {
+  for (const table of [cronTable, fireTable]) {
+    if (!/^[a-z_][a-z0-9_]*$/i.test(table)) throw new Error(`invalid table name: ${table}`);
+  }
+  const schema = [
+    `CREATE TABLE IF NOT EXISTS ${fireTable} (
+        cron_id TEXT NOT NULL,
+        fire_key TEXT NOT NULL,
+        fired_at BIGINT NOT NULL,
+        json JSONB NOT NULL,
+        PRIMARY KEY (cron_id, fire_key),
+        FOREIGN KEY (cron_id) REFERENCES ${cronTable}(id) ON DELETE CASCADE
+      )`,
+    `CREATE INDEX IF NOT EXISTS ${fireTable}_cron_fired_idx ON ${fireTable} (cron_id, fired_at DESC, fire_key DESC)`,
+    `CREATE INDEX CONCURRENTLY IF NOT EXISTS ${fireTable}_inline_cron_idx ON ${cronTable} (id) WHERE json ? 'fireLog'`,
+  ];
+  let readyP: Promise<void> | undefined;
+  const ready = () =>
+    (readyP ??= (async () => {
+      try {
+        if (!pg.schema) throw new Error("cron fire history requires schema-capable Postgres storage");
+        for (const statement of schema) await pg.schema(statement);
+      } catch (error) {
+        readyP = undefined;
+        throw error;
+      }
+    })());
+  const importEntries = async (cronId: string, entries: CronFireLogEntry[]) => {
+    if (!entries.length) return;
+    await ready();
+    await pg.query(
+      `INSERT INTO ${fireTable} (cron_id, fire_key, fired_at, json)
+       SELECT $1, entry->>'fireKey', (entry->>'firedAt')::BIGINT, entry
+       FROM jsonb_array_elements($2::jsonb) entry
+       ON CONFLICT (cron_id, fire_key) DO UPDATE
+       SET fired_at = EXCLUDED.fired_at, json = ${fireTable}.json || EXCLUDED.json`,
+      [cronId, jsonbStringify(entries)],
+    );
+  };
+  return {
+    ready,
+    async drainInline(cronId) {
+      await ready();
+      await pg.query(
+        `WITH source AS (
+           SELECT id, json FROM ${cronTable}
+           WHERE json ? 'fireLog'${cronId === undefined ? "" : " AND id = $1"} FOR UPDATE
+         ), copied AS (
+           INSERT INTO ${fireTable} (cron_id, fire_key, fired_at, json)
+           SELECT source.id, entry->>'fireKey', (entry->>'firedAt')::BIGINT, entry
+           FROM source
+           CROSS JOIN LATERAL jsonb_array_elements(
+             CASE WHEN jsonb_typeof(source.json->'fireLog') = 'array' THEN source.json->'fireLog' ELSE '[]'::jsonb END
+           ) AS entry
+           WHERE entry ? 'fireKey' AND entry ? 'firedAt'
+           ON CONFLICT (cron_id, fire_key) DO UPDATE
+           SET fired_at = EXCLUDED.fired_at, json = ${fireTable}.json || EXCLUDED.json
+         )
+         UPDATE ${cronTable} target SET json = target.json - 'fireLog'
+         FROM source WHERE target.id = source.id`,
+        cronId === undefined ? [] : [cronId],
+      );
+    },
+    async record(cronId, entry) {
+      await importEntries(cronId, [entry]);
+    },
+    async import(cronId, entries) {
+      await importEntries(cronId, entries);
+    },
+    async list(cronId, limit) {
+      await ready();
+      const rows =
+        limit === undefined
+          ? await pg.q(
+              `SELECT json, COUNT(*) OVER()::BIGINT AS total
+               FROM ${fireTable} WHERE cron_id = $1 ORDER BY fired_at, fire_key`,
+              [cronId],
+            )
+          : await pg.q(
+              `SELECT json, total FROM (
+                 SELECT fired_at, fire_key, json, COUNT(*) OVER()::BIGINT AS total
+                 FROM ${fireTable}
+                 WHERE cron_id = $1 ORDER BY fired_at DESC, fire_key DESC LIMIT $2
+               ) recent ORDER BY fired_at, fire_key`,
+              [cronId, limit],
+            );
+      return { runs: rows.map((row) => row.json as CronFireLogEntry), total: Number(rows[0]?.total ?? 0) };
+    },
+    async delete(cronId) {
+      await ready();
+      await pg.query(`DELETE FROM ${fireTable} WHERE cron_id = $1`, [cronId]);
+    },
+  };
+}

src/cron/cron-store.ts

@@ -10,6 +10,7 @@ import {
 } from "../triggers/trigger-store.ts";
 import { hashId } from "../util/crypto.ts";
 import { advanceNextFireAt, isCalendarSchedule, normalizeSchedule, recoverNextFireAt } from "./schedule.ts";
+import { createMemoryCronFireStore, type CronFirePage, type CronFireStore } from "./cron-fire-store.ts";
 
 export interface CreateCronInput extends CreateTriggerInput {
   schedule: Cron["schedule"];
@@ -44,6 +45,7 @@ export interface CronStore {
   setDestination(id: string, destination: Destination | undefined): Promise<void>;
   setRecipientConsent(id: string, recipientConsent: RecipientConsent): Promise<void>;
   recordFire(id: string, entry: CronFireLogEntry): Promise<void>;
+  getRuns(id: string, limit?: number): Promise<CronFirePage>;
   markFired(id: string, at: number, scheduledAt?: number): Promise<void>;
   markAttempted(id: string, at: number): Promise<void>;
   claimSlot(id: string, scheduledAt: number, at: number): Promise<boolean>;
@@ -57,7 +59,33 @@ function normalizeTitle(title: string | undefined): string | undefined {
   return trimmed.length > 80 ? `${trimmed.slice(0, 79)}...` : trimmed;
 }
 
-export function createCronStore(backing: DurableMap<Cron> = createMemoryMap<Cron>()): CronStore {
+export function createCronStore(
+  backing: DurableMap<Cron> = createMemoryMap<Cron>(),
+  fires: CronFireStore = createMemoryCronFireStore(),
+): CronStore {
+  let readyP: Promise<void> | undefined;
+  const ready = () =>
+    (readyP ??= (async () => {
+      try {
+        await backing.get("__cron_fire_log_migration__");
+        await fires.ready();
+      } catch (error) {
+        readyP = undefined;
+        throw error;
+      }
+    })());
+  const withoutFireLog = async (cron: Cron | null): Promise<Cron | null> => {
+    if (!cron) return null;
+    const { fireLog, ...rest } = cron;
+    if (fireLog?.length) {
+      if (fires.drainInline) await fires.drainInline(cron.id);
+      else {
+        await fires.import(cron.id, fireLog);
+        await backing.merge(cron.id, { fireLog: undefined });
+      }
+    }
+    return rest;
+  };
   return {
     async create(input) {
       assertNoEscalation(input);
@@ -88,8 +116,16 @@ export function createCronStore(backing: DurableMap<Cron> = createMemoryMap<Cron
         ...(input.unattendedGrants ? { unattendedGrants: input.unattendedGrants } : {}),
       }));
     },
-    get: (id) => backing.get(id),
-    list: () => backing.all(),
+    async get(id) {
+      await ready();
+      if (fires.drainInline) await fires.drainInline(id);
+      return withoutFireLog(await backing.get(id));
+    },
+    async list() {
+      await ready();
+      if (fires.drainInline) await fires.drainInline();
+      return (await Promise.all((await backing.all()).map(withoutFireLog))) as Cron[];
+    },
     async update(id, patch) {
       const fields: Partial<Cron> = {};
       if (patch.title !== undefined) fields.title = normalizeTitle(patch.title);
@@ -109,7 +145,9 @@ export function createCronStore(backing: DurableMap<Cron> = createMemoryMap<Cron
       if (patch.unattendedGrants !== undefined) fields.unattendedGrants = patch.unattendedGrants;
       return backing.merge(id, fields);
     },
-    delete: (id) => backing.delete(id),
+    async delete(id) {
+      await Promise.all([backing.delete(id), fires.delete(id)]);
+    },
     async setEnabled(id, enabled) {
       await backing.merge(id, { enabled, ...(enabled ? { archived: false } : {}) });
     },
@@ -120,18 +158,15 @@ export function createCronStore(backing: DurableMap<Cron> = createMemoryMap<Cron
       return setTriggerRecipientConsent(backing, id, recipientConsent);
     },
     async recordFire(id, entry) {
-      const addEntry = (cron: Cron): Cron => {
-        const byKey = new Map((cron.fireLog ?? []).map((e) => [e.fireKey, e]));
-        byKey.set(entry.fireKey, { ...byKey.get(entry.fireKey), ...entry });
-        return { ...cron, fireLog: [...byKey.values()].sort((a, b) => a.firedAt - b.firedAt) };
-      };
-      if (backing.update) {
-        await backing.update(id, addEntry);
-        return;
-      }
-      const cron = await backing.get(id);
-      if (!cron) return;
-      await backing.merge(id, { fireLog: addEntry(cron).fireLog });
+      await ready();
+      if (fires.drainInline) await fires.drainInline(id);
+      if (!(await withoutFireLog(await backing.get(id)))) return;
+      await fires.record(id, entry);
+    },
+    async getRuns(id, limit) {
+      await ready();
+      await withoutFireLog(await backing.get(id));
+      return fires.list(id, limit);
     },
     async markFired(id, at, scheduledAt) {
       const cron = await backing.get(id);
@@ -186,7 +221,10 @@ export function createCronStore(backing: DurableMap<Cron> = createMemoryMap<Cron
     },
     async due(now) {
       const due: Array<Cron & { scheduledAt: number }> = [];
-      for (const c of await backing.all()) {
+      await ready();
+      if (fires.drainInline) await fires.drainInline();
+      for (const row of await backing.all()) {
+        const c = (await withoutFireLog(row))!;
         if (c.archived || !c.enabled) continue;
         const scheduledAt = recoverNextFireAt(c.schedule, c.createdAt, c.lastFiredAt, c.nextFireAt);
         if (scheduledAt !== undefined && now >= scheduledAt) due.push({ ...c, nextFireAt: scheduledAt, scheduledAt });

src/wiring.ts

@@ -59,6 +59,7 @@ import { createPostgresRateLimiter } from "./ratelimit/postgres-rate-limiter.ts"
 import { createBudgetTracker } from "./ratelimit/budget.ts";
 import { createPostgresBudgetTracker } from "./ratelimit/postgres-budget.ts";
 import { createCronStore, type CronStore } from "./cron/cron-store.ts";
+import { createMemoryCronFireStore, createPostgresCronFireStore } from "./cron/cron-fire-store.ts";
 import { createDeliveryStore, type DeliveryStore } from "./delivery/delivery-store.ts";
 import { createPostgresDeliveryStore } from "./delivery/postgres-delivery-store.ts";
 import { wireRunResultDeliveries } from "./delivery/run-result-delivery.ts";
@@ -933,7 +934,10 @@ export function buildApp(
     : createMemoryEnvironmentStore();
   const monitors = createMonitorStore(artifactMap<Monitor>("monitors"));
   const cronChanged: { notify?: (id: string) => void } = {};
-  const cronsBase = createCronStore(artifactMap<Cron>("crons"));
+  const cronsBase = createCronStore(
+    artifactMap<Cron>("crons"),
+    pgArtifactMap ? createPostgresCronFireStore(pgArtifactMap.pool) : createMemoryCronFireStore(),
+  );
   const crons: CronStore = {
     ...cronsBase,
     async create(input) {

test/cron-queue.test.ts

@@ -3,6 +3,7 @@ import assert from "node:assert/strict";
 import { createScheduler, type Scheduler } from "../src/cron/scheduler.ts";
 import { createPgBossCronQueue } from "../src/cron/job-queue.ts";
 import { createCronStore, type CronStore } from "../src/cron/cron-store.ts";
+import { createPostgresCronFireStore } from "../src/cron/cron-fire-store.ts";
 import { createDeliveryStore } from "../src/delivery/delivery-store.ts";
 import { createIdempotencyStore, type IdempotencyRecord } from "../src/idempotency/idempotency-store.ts";
 import { createIdentityService } from "../src/identity/identity-service.ts";
@@ -15,13 +16,14 @@ const skip = URL ? false : "set DATABASE_URL (a Postgres) to run the cron queue
 const SCHEMA = "pgboss_cron_queue_test";
 const CRONS_TABLE = "cron_queue_test_crons";
 const IDEM_TABLE = "cron_queue_test_idempotency";
+const FIRES_TABLE = "cron_queue_test_fires";
 
 before(async () => {
   if (!URL) return;
   const pg = (await import("pg")).default;
   const p = new pg.Pool({ connectionString: URL });
   await p.query(`DROP SCHEMA IF EXISTS ${SCHEMA} CASCADE`);
-  await p.query(`DROP TABLE IF EXISTS ${CRONS_TABLE}, ${IDEM_TABLE}`);
+  await p.query(`DROP TABLE IF EXISTS ${CRONS_TABLE}, ${IDEM_TABLE}, ${FIRES_TABLE}`);
   await p.end();
 });
 
@@ -32,7 +34,10 @@ async function until(cond: () => boolean, ms: number): Promise<void> {
 
 function instance(calls: TurnRequest[], turnMs = 0): { scheduler: Scheduler; crons: CronStore } {
   const maps = createPostgresMapFactory(URL!);
-  const crons = createCronStore(maps.map<Cron>(CRONS_TABLE));
+  const crons = createCronStore(
+    maps.map<Cron>(CRONS_TABLE),
+    createPostgresCronFireStore(maps.pool, CRONS_TABLE, FIRES_TABLE),
+  );
   const run = async (req: TurnRequest): Promise<TurnResult> => {
     calls.push(req);
     if (turnMs) await new Promise((r) => setTimeout(r, turnMs));
@@ -71,7 +76,7 @@ test(
       await new Promise((r) => setTimeout(r, 16_000));
       assert.equal(calls.length, 1, "no sibling or reconcile re-run while (or after) the slow turn runs");
       assert.equal((await b.crons.get(cron.id))?.enabled, false, "the one-shot ends disabled");
-      assert.equal((await b.crons.get(cron.id))?.fireLog?.length, 1, "one fire recorded");
+      assert.equal((await b.crons.getRuns(cron.id)).runs.length, 1, "one fire recorded");
     } finally {
       a.scheduler.stop();
       b.scheduler.stop();