fix(ticker): claim due rows before delivering

README.md

@@ -59,6 +59,8 @@ Paths are relative to where the host mounts the sub-app.
 
 Returns `{ start(), stop() }`.
 
+Delivery is at most once per due minute. Each tick first commits `lastFiredAt` on every due row it claims, then calls `deliver` outside the row lock. A process that dies mid-batch, or a `deliver` that throws, drops that minute's fire rather than repeating it; the schedule fires again at its next matching minute. Stop and waiting callbacks run after their change commits.
+
 | `opts`              | Type                                   | Purpose                                                                                                 |
 | ------------------- | -------------------------------------- | ------------------------------------------------------------------------------------------------------- |
 | `db`                | `CronDb`                               | The hub's drizzle handle.                                                                               |

e2e/ticker.test.ts

@@ -1,7 +1,7 @@
 import { afterAll, beforeAll, expect, test } from "bun:test";
 import { randomUUID } from "node:crypto";
 import { createDB, schema } from "@intx/db";
-import { eq } from "drizzle-orm";
+import { eq, sql } from "drizzle-orm";
 
 import { cronScheduleTable } from "../src/schema.js";
 import { createCronTicker } from "../src/ticker.js";
@@ -14,6 +14,7 @@ import {
   createTestDatabase,
   describeIfDb,
   type TestDatabase,
+  type TestDb,
   DB_SETUP_TIMEOUT_MS,
 } from "./helpers.js";
 import {
@@ -295,63 +296,129 @@ describeIfDb("createCronTicker", () => {
     }
   });
 
-  test("two concurrent tickers deliver a due row exactly once", async () => {
-    const { db: dbA, close: closeA } = createDB(requireDatabase().config);
-    const { db: dbB, close: closeB } = createDB(requireDatabase().config);
-    try {
-      const tenantId = `tnt_cron_race_${randomUUID().slice(0, 8)}`;
-      await seedTenant(dbA, tenantId);
-      await seedDeployment(dbA, tenantId, "agent-race-source");
-
-      const id = `sched_race_${randomUUID().slice(0, 8)}`;
-      await dbA.insert(cronScheduleTable).values({
+  async function seedDueBatch(
+    db: TestDb,
+    label: string,
+    count: number,
+  ): Promise<{ tenantId: string; ids: string[] }> {
+    const tenantId = `tnt_cron_${label}_${randomUUID().slice(0, 8)}`;
+    await seedTenant(db, tenantId);
+    await seedDeployment(db, tenantId, `agent-${label}-source`);
+    const ids = Array.from(
+      { length: count },
+      (_, i) => `sched_${label}_${i}_${randomUUID().slice(0, 8)}`,
+    );
+    // Due once (Jan 1 has passed since creation), and not again next minute,
+    // so a tick that crosses a minute boundary cannot legitimately re-fire.
+    await db.insert(cronScheduleTable).values(
+      ids.map((id) => ({
         id,
         tenantId,
-        expression: "* * * * *",
-        definitionName: "agent-race-source",
-        subject: "race",
-        body: "fire once",
-        createdAt: new Date(Date.now() - 2 * 60_000),
-      });
+        expression: "0 0 1 1 *",
+        definitionName: `agent-${label}-source`,
+        subject: id,
+        body: "b",
+        createdAt: new Date(Date.now() - 400 * 24 * 60 * 60_000),
+      })),
+    );
+    return { tenantId, ids };
+  }
+
+  function countBy(subjects: string[]): Map<string, number> {
+    const counts = new Map<string, number>();
+    for (const subject of subjects) {
+      counts.set(subject, (counts.get(subject) ?? 0) + 1);
+    }
+    return counts;
+  }
 
+  test("a tick killed mid-batch delivers each row at most once", async () => {
+    const config = requireDatabase().config;
+    const crashed = createDB(config);
+    const survivor = createDB(config);
+    try {
+      const { ids } = await seedDueBatch(crashed.db, "crash", 21);
       const delivered: string[] = [];
-      const deliverer: RunTriggerDeliverer = {
-        to: async (address) => {
-          delivered.push(address);
-        },
-      };
-      const tickerA = createCronTicker({
-        db: dbA,
+      let killed = false;
+      // The fifth delivery kills the tick's Postgres sessions and never
+      // returns, like a process dying mid-batch.
+      const crashing = createCronTicker({
+        db: crashed.db,
         intervalMs: 20,
-        deliver: createRunTriggerCronDeliver(deliverer),
+        deliver: async (message) => {
+          if (!ids.includes(message.subject)) return;
+          delivered.push(message.subject);
+          if (delivered.length < 5) return;
+          killed = true;
+          await crashed.db.execute(
+            sql`SELECT pg_terminate_backend(pid) FROM pg_stat_activity WHERE datname = current_database() AND pid <> pg_backend_pid()`,
+          );
+          await new Promise(() => undefined);
+        },
       });
-      const tickerB = createCronTicker({
-        db: dbB,
+      crashing.start();
+      for (let i = 0; i < 250 && !killed; i++) {
+        await new Promise((resolve) => setTimeout(resolve, 20));
+      }
+      crashing.stop();
+
+      const recovering = createCronTicker({
+        db: survivor.db,
         intervalMs: 20,
-        deliver: createRunTriggerCronDeliver(deliverer),
+        deliver: (message) => {
+          if (ids.includes(message.subject)) delivered.push(message.subject);
+        },
       });
-      tickerA.start();
-      tickerB.start();
-      // Polling, not a fixed sleep: tick latency follows DB load.
-      for (let i = 0; i < 1000 && delivered.length === 0; i++) {
+      recovering.start();
+      await new Promise((resolve) => setTimeout(resolve, 1_000));
+      recovering.stop();
+      await new Promise((resolve) => setTimeout(resolve, 300));
+
+      expect(killed).toBe(true);
+      expect([...countBy(delivered).values()].every((n) => n === 1)).toBe(true);
+    } finally {
+      await Promise.race([
+        crashed.close(),
+        new Promise((resolve) => setTimeout(resolve, 1_000)),
+      ]);
+      await survivor.close();
+    }
+  });
+
+  test("three concurrent tickers deliver each row exactly once", async () => {
+    const handles = [1, 2, 3].map(() => createDB(requireDatabase().config));
+    try {
+      const [first] = handles;
+      if (first === undefined) throw new Error("no db handle");
+      const { ids } = await seedDueBatch(first.db, "trio", 21);
+      const delivered: string[] = [];
+      const deliverer: RunTriggerDeliverer = {
+        to: async (_address, _content, _tenantId, subject) => {
+          if (subject === undefined || !ids.includes(subject)) return;
+          await new Promise((resolve) => setTimeout(resolve, 5));
+          delivered.push(subject);
+        },
+      };
+      const tickers = handles.map(({ db }) =>
+        createCronTicker({
+          db,
+          intervalMs: 20,
+          deliver: createRunTriggerCronDeliver(deliverer),
+        }),
+      );
+      for (const ticker of tickers) ticker.start();
+      for (let i = 0; i < 250 && delivered.length < ids.length; i++) {
         await new Promise((resolve) => setTimeout(resolve, 20));
       }
-      // Both tickers keep racing past the first delivery.
+      // All three keep racing past the last delivery.
       await new Promise((resolve) => setTimeout(resolve, 300));
-      tickerA.stop();
-      tickerB.stop();
-      // Let an in-flight tick settle before close() ends the client.
+      for (const ticker of tickers) ticker.stop();
       await new Promise((resolve) => setTimeout(resolve, 300));
 
-      expect(delivered).toHaveLength(1);
-      const [row] = await dbA
-        .select({ lastFiredAt: cronScheduleTable.lastFiredAt })
-        .from(cronScheduleTable)
-        .where(eq(cronScheduleTable.id, id));
-      expect(row?.lastFiredAt).toBeInstanceOf(Date);
+      const counts = countBy(delivered);
+      expect(ids.map((id) => counts.get(id))).toEqual(ids.map(() => 1));
     } finally {
-      await closeA();
-      await closeB();
+      for (const handle of handles) await handle.close();
     }
   });
 

src/ticker.ts

@@ -105,12 +105,21 @@ async function tick<TSchema extends Record<string, unknown>>(
     definitionName: string;
   }) => void,
 ) {
+  const now = new Date();
+  const stopped: Parameters<typeof onScheduleStopped>[0][] = [];
+  const waiting: Parameters<typeof onScheduleWaiting>[0][] = [];
+  const claimed: Array<{
+    row: typeof cronScheduleTable.$inferSelect;
+    address: string;
+  }> = [];
+
+  // Claim, then deliver. One short transaction locks every row (SKIP LOCKED
+  // means concurrent tickers split the due rows rather than double-fire
+  // any of them), decides which are due, and advances lastFiredAt on each
+  // it will deliver. Delivery runs after that commits and outside the
+  // lock, so a slow deliverer blocks no replica and a tick that dies
+  // mid-batch never re-fires what it claimed: at most once per due minute.
   await db.transaction(async (tx) => {
-    const now = new Date();
-    // Every row, locked against a concurrent ticker (SKIP LOCKED means two
-    // tickers racing this table split the due rows rather than double-fire
-    // any of them); which ones are actually due is a JS-side check because
-    // "due" depends on parsing each row's own cron expression.
     const candidates = await tx
       .select()
       .from(cronScheduleTable)
@@ -120,12 +129,17 @@ async function tick<TSchema extends Record<string, unknown>>(
     for (const row of candidates) {
       const due = dueness(row, now);
       if (due === "idle") continue;
+      const schedule = {
+        id: row.id,
+        tenantId: row.tenantId,
+        definitionName: row.definitionName,
+      };
       if (due === "invalid") {
         await tx
           .update(cronScheduleTable)
           .set({ stoppedAt: now, stoppedReason: INVALID_EXPRESSION })
           .where(eq(cronScheduleTable.id, row.id));
-        onScheduleStopped({
+        stopped.push({
           id: row.id,
           tenantId: row.tenantId,
           definitionName: row.definitionName,
@@ -149,54 +163,57 @@ async function tick<TSchema extends Record<string, unknown>>(
               .update(cronScheduleTable)
               .set({ waitingSince: now })
               .where(eq(cronScheduleTable.id, row.id));
-            onScheduleWaiting({
-              id: row.id,
-              tenantId: row.tenantId,
-              definitionName: row.definitionName,
-            });
+            waiting.push(schedule);
           }
           continue;
         }
         await tx
           .update(cronScheduleTable)
           .set({ stoppedAt: now, stoppedReason: AGENT_DELETED })
           .where(eq(cronScheduleTable.id, row.id));
-        onScheduleStopped({
+        stopped.push({
           id: row.id,
           tenantId: row.tenantId,
           definitionName: row.definitionName,
           reason: AGENT_DELETED,
         });
         continue;
       }
-      // One schedule's failed delivery is its own: the tick still advances
-      // every due row, so a permanently undeliverable schedule cannot block
-      // the rest of the table or re-fire every minute forever.
-      try {
-        await deliver({
-          to: [deployment.address],
-          subject: row.subject,
-          body: row.body,
-          tenantId: row.tenantId,
-        });
-      } catch (error) {
-        // The deliverer names the dead run it could not route to. When that
-        // run's sidecar can never come back (its allocation already settled),
-        // fail the stale anchor now: the next tick then sees no live
-        // deployment and waits for the agent to come back instead of
-        // delivering into the dead run every minute. The error is still
-        // reported — it is a real delivery failure, once, not tick noise.
-        if (isUnroutableRunTrigger(error)) {
-          await failStaleAnchorRun(tx, error.runId, now);
-        }
-        onDeliveryError(error, { id: row.id, tenantId: row.tenantId });
-      }
       await tx
         .update(cronScheduleTable)
         .set({ lastFiredAt: now, waitingSince: null })
         .where(eq(cronScheduleTable.id, row.id));
+      claimed.push({ row, address: deployment.address });
     }
   });
+
+  for (const schedule of stopped) onScheduleStopped(schedule);
+  for (const schedule of waiting) onScheduleWaiting(schedule);
+
+  // One schedule's failed delivery is its own: the rest of the batch still
+  // delivers, and a permanently undeliverable schedule fires once per due
+  // minute, not every tick.
+  for (const { row, address } of claimed) {
+    try {
+      await deliver({
+        to: [address],
+        subject: row.subject,
+        body: row.body,
+        tenantId: row.tenantId,
+      });
+    } catch (error) {
+      // The deliverer names the dead run it could not route to. When that
+      // run's sidecar can never come back (its allocation already settled),
+      // fail the stale anchor now: the next tick then sees no live
+      // deployment and waits for the agent to come back instead of
+      // delivering into the dead run every minute. The error is still
+      // reported — it is a real delivery failure, once, not tick noise.
+      if (isUnroutableRunTrigger(error)) {
+        await failStaleAnchorRun(db, error.runId, now);
+      }
+      onDeliveryError(error, { id: row.id, tenantId: row.tenantId });
+    }
+  }
 }
 
 /** Ticks every `intervalMs`, delivering each due schedule as mail. */