From 36cc2fe40d44cae0eed701d34e86b92f9c616106 Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Sat, 26 Sep 2026 19:27:07 -0700 Subject: [PATCH] fix(ticker): claim due rows before delivering One short transaction locks the rows, decides which are due and commits lastFiredAt on each it will deliver. deliver() then runs after commit and outside the row lock, so a slow deliverer blocks no replica and a tick that dies mid-batch never re-fires what it claimed: delivery is at most once per due minute, as the README now says. Stop and waiting callbacks fire after their change commits. New e2e tests kill a tick mid-batch and race three tickers; the three-ticker test replaces the two-ticker one. Closes CL-9448 --- README.md | 2 + e2e/ticker.test.ts | 155 ++++++++++++++++++++++++++++++++------------- src/ticker.ts | 85 +++++++++++++++---------- 3 files changed, 164 insertions(+), 78 deletions(-) diff --git a/README.md b/README.md index def4c9d..203a758 100644 --- a/README.md +++ b/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. | diff --git a/e2e/ticker.test.ts b/e2e/ticker.test.ts index 70af1c4..40c6a96 100644 --- a/e2e/ticker.test.ts +++ b/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 { + const counts = new Map(); + 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(); } }); diff --git a/src/ticker.ts b/src/ticker.ts index 3f30a1a..e5b220d 100644 --- a/src/ticker.ts +++ b/src/ticker.ts @@ -105,12 +105,21 @@ async function tick>( definitionName: string; }) => void, ) { + const now = new Date(); + const stopped: Parameters[0][] = []; + const waiting: Parameters[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>( 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,11 +163,7 @@ async function tick>( .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; } @@ -161,7 +171,7 @@ async function tick>( .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, @@ -169,34 +179,41 @@ async function tick>( }); 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. */