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. */