From 6e0afb6986695350ab0e3eb5cf6c61ec7f038f63 Mon Sep 17 00:00:00 2001 From: Sawyer Date: Wed, 30 Sep 2026 08:10:50 -0700 Subject: [PATCH] feat: pause and resume a schedule Adds an enabled column, POST /:id/pause and POST /:id/resume. The ticker skips paused rows; resume restarts the schedule's clock at now. --- README.md | 14 ++++---- e2e/routes.test.ts | 64 ++++++++++++++++++++++++++++++++++++ migrations/0001_schedule.sql | 2 ++ src/routes.ts | 26 +++++++++++++++ src/schema.ts | 10 +++++- src/ticker.ts | 9 +++-- 6 files changed, 116 insertions(+), 9 deletions(-) diff --git a/README.md b/README.md index 203a758..9235f21 100644 --- a/README.md +++ b/README.md @@ -2,7 +2,7 @@ A Corbits hub module that wakes Interchange agents on five-field UTC cron schedules, mounted as `@intx/hub-api` routes on the hub (Interchange's multi-tenant control plane) and stored in its Postgres. A ticker mails each due schedule to its agent's current run, the running instance of the agent's workflow definition. -Schedules support create, list and delete. There is no edit: delete and recreate. An expression must be able to fire (no Feb 30), at most 256 characters and 64 comma clauses. +Schedules support create, list, pause, resume and delete. There is no edit: delete and recreate. An expression must be able to fire (no Feb 30), at most 256 characters and 64 comma clauses. ## Why @corbits/cron? @@ -49,11 +49,13 @@ Prints the next 09:00 UTC, the next time a schedule with that expression fires. Paths are relative to where the host mounts the sub-app. -| Method | Path | Grant | Purpose | -| ------ | ------ | ----------------------------- | -------------------------------------------------------------------------------------------------------------------------------------------------- | -| GET | `/` | `cron-schedule:*` `read` | `{ schedules: [...] }` for the caller's tenant. | -| POST | `/` | `cron-schedule:*` `create` | Create from `{ expression, definitionName, subject, body }`. 201 `{ schedule }`; 400 `invalid_body`, `invalid_expression` or `unknown_definition`. | -| DELETE | `/:id` | `cron-schedule:` `manage` | Delete one schedule. `{ ok: true }`; 404 `not_found` when it is not in the caller's tenant. | +| Method | Path | Grant | Purpose | +| ------ | ------------- | ----------------------------- | -------------------------------------------------------------------------------------------------------------------------------------------------- | +| GET | `/` | `cron-schedule:*` `read` | `{ schedules: [...] }` for the caller's tenant. | +| POST | `/` | `cron-schedule:*` `create` | Create from `{ expression, definitionName, subject, body }`. 201 `{ schedule }`; 400 `invalid_body`, `invalid_expression` or `unknown_definition`. | +| POST | `/:id/pause` | `cron-schedule:` `manage` | Stop the ticker firing it. `{ schedule }` with `enabled: false`; 404 `not_found`. | +| POST | `/:id/resume` | `cron-schedule:` `manage` | Fire again from the next matching minute after now. `{ schedule }` with `enabled: true`; 404 `not_found`. | +| DELETE | `/:id` | `cron-schedule:` `manage` | Delete one schedule. `{ ok: true }`; 404 `not_found` when it is not in the caller's tenant. | ### `createCronTicker(opts)` diff --git a/e2e/routes.test.ts b/e2e/routes.test.ts index 8dbb9f3..bd02873 100644 --- a/e2e/routes.test.ts +++ b/e2e/routes.test.ts @@ -4,8 +4,12 @@ import { afterAll, beforeAll, expect, test } from "bun:test"; import { randomUUID } from "node:crypto"; import { createDB } from "@intx/db"; import type { RequireGrant, TenantEnv } from "@intx/hub-api"; +import { eq } from "drizzle-orm"; import type { Hono } from "hono"; +import { cronScheduleTable } from "../src/schema.js"; +import { createCronTicker } from "../src/ticker.js"; + import { createTestDatabase, cronRoutesApp, @@ -109,6 +113,66 @@ describeIfDb("createCronRoutes", () => { } }); + test("a paused schedule is skipped by the ticker; resume restarts its clock", async () => { + const { db, close } = createDB(requireDatabase().config); + try { + const tenantId = `tnt_cron_pause_${randomUUID().slice(0, 8)}`; + await seedTenant(db, tenantId); + await seedDeployment(db, tenantId, "agent-pause-source"); + const app = cronRoutesApp(db, tenantId, allowAll); + const id = `sched_pause_${randomUUID().slice(0, 8)}`; + await db.insert(cronScheduleTable).values({ + id, + tenantId, + expression: "* * * * *", + definitionName: "agent-pause-source", + subject: "s", + body: "b", + createdAt: new Date(Date.now() - 5 * 60_000), + }); + + const paused = await app.request(`/cron/${id}/pause`, { method: "POST" }); + expect(paused.status).toBe(200); + expect( + ((await paused.json()) as { schedule: { enabled: boolean } }).schedule + .enabled, + ).toBe(false); + + const delivered: string[] = []; + const ticker = createCronTicker({ + db, + intervalMs: 50, + deliver: (message) => { + delivered.push(message.subject); + }, + }); + ticker.start(); + await new Promise((resolve) => setTimeout(resolve, 500)); + ticker.stop(); + expect(delivered).toEqual([]); + + const resumed = await app.request(`/cron/${id}/resume`, { + method: "POST", + }); + expect(resumed.status).toBe(200); + const [row] = await db + .select() + .from(cronScheduleTable) + .where(eq(cronScheduleTable.id, id)); + expect(row?.enabled).toBe(true); + // Restarted at now, not due for the five paused minutes. + expect(Date.now() - (row?.lastFiredAt?.getTime() ?? 0)).toBeLessThan( + 30_000, + ); + + expect( + (await app.request("/cron/missing/pause", { method: "POST" })).status, + ).toBe(404); + } finally { + await close(); + } + }); + test("each route is gated by the host's requireGrant", async () => { const { db, close } = createDB(requireDatabase().config); try { diff --git a/migrations/0001_schedule.sql b/migrations/0001_schedule.sql index a814070..87b9624 100644 --- a/migrations/0001_schedule.sql +++ b/migrations/0001_schedule.sql @@ -17,4 +17,6 @@ CREATE TABLE IF NOT EXISTS "cron"."schedule" ( -- Tables created before waiting_since existed. ALTER TABLE "cron"."schedule" ADD COLUMN IF NOT EXISTS "waiting_since" timestamptz; --> statement-breakpoint +ALTER TABLE "cron"."schedule" ADD COLUMN IF NOT EXISTS "enabled" boolean NOT NULL DEFAULT true; +--> statement-breakpoint CREATE INDEX IF NOT EXISTS "cron_schedule_tenant_id_idx" ON "cron"."schedule" ("tenant_id"); diff --git a/src/routes.ts b/src/routes.ts index b1ede82..94763dd 100644 --- a/src/routes.ts +++ b/src/routes.ts @@ -86,5 +86,31 @@ export function createCronRoutes({ }, ); + // Resume restarts the clock at now: a schedule is due from its last fire, + // so keeping the old one would fire at once for the paused minutes. + for (const [path, enabled] of [ + ["pause", false], + ["resume", true], + ] as const) { + app.post( + `/:id/${path}`, + requireGrant(idResource("cron-schedule", "id"), "manage"), + async (c) => { + const [row] = await db + .update(cronScheduleTable) + .set(enabled ? { enabled, lastFiredAt: new Date() } : { enabled }) + .where( + and( + eq(cronScheduleTable.tenantId, c.get("tenant").id), + eq(cronScheduleTable.id, c.req.param("id")), + ), + ) + .returning(); + if (row === undefined) return c.json({ error: "not_found" }, 404); + return c.json({ schedule: row }); + }, + ); + } + return app; } diff --git a/src/schema.ts b/src/schema.ts index 6c390e4..dbaf3be 100644 --- a/src/schema.ts +++ b/src/schema.ts @@ -1,7 +1,13 @@ // The `cron` schema's one table: a tenant's saved cron schedules. Kept on // its own Postgres schema, with a real FK back to Interchange's `tenant` // table, so it never collides with the host's own tables. -import { pgTable, pgSchema, text, timestamp } from "drizzle-orm/pg-core"; +import { + boolean, + pgTable, + pgSchema, + text, + timestamp, +} from "drizzle-orm/pg-core"; const hostTenant = pgTable("tenant", { id: text("id").primaryKey() }); @@ -18,6 +24,8 @@ export const cronScheduleTable = cronSchema.table("schedule", { definitionName: text("definition_name").notNull(), subject: text("subject").notNull(), body: text("body").notNull(), + // False while paused: the ticker skips the row until it is resumed. + enabled: boolean("enabled").notNull().default(true), lastFiredAt: timestamp("last_fired_at", { withTimezone: true }), // Set while the agent exists but has no live run — a hub restart, say. // Cleared by the first tick that delivers again. diff --git a/src/ticker.ts b/src/ticker.ts index e5b220d..d1613a8 100644 --- a/src/ticker.ts +++ b/src/ticker.ts @@ -1,6 +1,6 @@ // Turns due cron schedules into mail. A schedule that missed several ticks // fires once for the most recent due minute, never once per missed tick. -import { eq, isNull } from "drizzle-orm"; +import { and, eq, isNull } from "drizzle-orm"; import type { PostgresJsDatabase } from "drizzle-orm/postgres-js"; import { cronIsDue, parseCronExpression, parsedCronCanFire } from "./cron.js"; @@ -123,7 +123,12 @@ async function tick>( const candidates = await tx .select() .from(cronScheduleTable) - .where(isNull(cronScheduleTable.stoppedAt)) + .where( + and( + isNull(cronScheduleTable.stoppedAt), + eq(cronScheduleTable.enabled, true), + ), + ) .for("update", { skipLocked: true }); for (const row of candidates) {