feat(cron): support configurable run timeouts

apps/api/cron-timeout.test.ts

@@ -0,0 +1,51 @@
+process.env.MOOR_DB_PATH = ":memory:";
+
+import { beforeEach, describe, expect, test } from "bun:test";
+
+let observedTimeout: number | undefined;
+
+const { default: db } = await import("./db");
+const { runCron } = await import("./cron");
+
+describe("cron timeout execution", () => {
+  beforeEach(() => {
+    observedTimeout = undefined;
+    db.query("DELETE FROM runs").run();
+    db.query("DELETE FROM crons").run();
+    db.query("DELETE FROM projects").run();
+  });
+
+  test("passes the cron's configured timeout to the container exec", async () => {
+    const project = db
+      .query("INSERT INTO projects (name) VALUES ('pipeline') RETURNING id")
+      .get() as { id: number };
+    const cron = db
+      .query(
+        `INSERT INTO crons (project_id, name, schedule, command, timeout_ms)
+         VALUES (?, 'refresh', '30 7 * * *', 'run-pipeline', 86400000)
+         RETURNING *`,
+      )
+      .get(project.id) as {
+      id: number;
+      project_id: number;
+      name: string;
+      schedule: string;
+      command: string;
+      timeout_ms: number;
+      enabled: number;
+    };
+
+    await runCron(cron, "container-id", async (_containerId, _command, opts) => {
+      observedTimeout = opts?.timeout_ms;
+      opts?.onExecId?.("exec-1");
+      return { exitCode: 0, stdout: "ok", stderr: "" };
+    });
+
+    expect(observedTimeout).toBe(86_400_000);
+    const run = db.query("SELECT exit_code, stdout FROM runs WHERE cron_id = ?").get(cron.id) as {
+      exit_code: number;
+      stdout: string;
+    };
+    expect(run).toEqual({ exit_code: 0, stdout: "ok" });
+  });
+});

apps/api/cron.ts

@@ -15,6 +15,7 @@ type CronRow = {
   name: string;
   schedule: string;
   command: string;
+  timeout_ms: number;
   enabled: number;
 };
 
@@ -162,7 +163,11 @@ export async function tickInner() {
   }
 }
 
-export async function runCron(cron: CronRow, containerId: string) {
+export async function runCron(
+  cron: CronRow,
+  containerId: string,
+  execute: typeof execInContainer = execInContainer,
+) {
   // #73: set started_at_ms and finished_at_ms so moor_runs' ms-precision
   // ordering (COALESCE(started_at_ms,0) DESC, id DESC) sorts cron runs
   // alongside build runs correctly, and so duration_ms is precise.
@@ -190,8 +195,9 @@ export async function runCron(cron: CronRow, containerId: string) {
   };
 
   try {
-    const result = await execInContainer(containerId, cron.command, {
+    const result = await execute(containerId, cron.command, {
       signal: controller.signal,
+      timeout_ms: cron.timeout_ms,
       onExecId: (id) => {
         entry.execId = id;
       },

apps/api/db-migrations.test.ts

@@ -72,7 +72,7 @@ const BASELINE_SCHEMA_SQL = `
   );
 `;
 
-type MigrationTable = "projects" | "runs" | "exec_runs";
+type MigrationTable = "projects" | "crons" | "runs" | "exec_runs";
 
 function withBaselineDatabase(run: (db: Database) => void): void {
   const db = new Database(":memory:");
@@ -130,6 +130,7 @@ function addLegacyMigrationColumns(db: Database): void {
     ALTER TABLE projects ADD COLUMN source_credential_id INTEGER REFERENCES source_credentials(id);
     ALTER TABLE projects ADD COLUMN command TEXT;
     ALTER TABLE projects ADD COLUMN entrypoint TEXT;
+    ALTER TABLE crons ADD COLUMN timeout_ms INTEGER NOT NULL DEFAULT 600000;
   `);
 }
 
@@ -162,7 +163,18 @@ describe("schema migrations", () => {
         "stdout_total_bytes",
         "stderr_total_bytes",
       ]);
+      expectColumns(db, "crons", ["timeout_ms"]);
       expectColumns(db, "exec_runs", ["started_at_ms", "finished_at_ms"]);
+
+      const project = db
+        .query("INSERT INTO projects (name) VALUES ('cron-default') RETURNING id")
+        .get() as { id: number };
+      const cron = db
+        .query(
+          "INSERT INTO crons (project_id, name, schedule, command) VALUES (?, 'c', '* * * * *', 'echo') RETURNING timeout_ms",
+        )
+        .get(project.id) as { timeout_ms: number };
+      expect(cron.timeout_ms).toBe(600_000);
     });
   });
 

apps/api/db-migrations.ts

@@ -298,6 +298,17 @@ export const schemaMigrations: readonly Migration[] = [
       );
     },
   },
+  {
+    version: 20,
+    up(db) {
+      addColumnIfMissing(
+        db,
+        "crons",
+        "timeout_ms",
+        "ALTER TABLE crons ADD COLUMN timeout_ms INTEGER NOT NULL DEFAULT 600000",
+      );
+    },
+  },
 ];
 
 export const finalSchemaVersion = schemaMigrations[schemaMigrations.length - 1]?.version ?? 0;

apps/api/db.ts

@@ -30,6 +30,7 @@ db.exec(`
     name TEXT NOT NULL,
     schedule TEXT NOT NULL,
     command TEXT NOT NULL,
+    timeout_ms INTEGER NOT NULL DEFAULT 600000,
     enabled INTEGER DEFAULT 1,
     created_at TEXT DEFAULT (datetime('now'))
   );

apps/api/routes/crons.test.ts

@@ -14,8 +14,12 @@ async function errorMessage(res: Response): Promise<string> {
   return ((await res.json()) as { error: string }).error;
 }
 
-async function call(method: string, path: string): Promise<Response> {
-  const req = new Request(`http://localhost${path}`, { method });
+async function call(method: string, path: string, body?: unknown): Promise<Response> {
+  const req = new Request(`http://localhost${path}`, {
+    method,
+    headers: body === undefined ? undefined : { "Content-Type": "application/json" },
+    body: body === undefined ? undefined : JSON.stringify(body),
+  });
   const res = await handleCrons(req, new URL(req.url));
   if (!res) throw new Error(`handleCrons returned null for ${method} ${path}`);
   return res;
@@ -51,3 +55,77 @@ describe("#73 POST /api/crons/:id/run live-check wiring", () => {
     expect(runs.n).toBe(0);
   });
 });
+
+describe("cron timeout configuration", () => {
+  beforeEach(() => {
+    db.query("DELETE FROM runs").run();
+    db.query("DELETE FROM crons").run();
+    db.query("DELETE FROM projects").run();
+  });
+
+  test("create defaults to 10 minutes and accepts a multi-hour timeout", async () => {
+    const project = db
+      .query("INSERT INTO projects (name) VALUES ('timeouts') RETURNING id")
+      .get() as { id: number };
+
+    const defaultRes = await call("POST", `/api/projects/${project.id}/crons`, {
+      name: "default",
+      schedule: "0 3 * * *",
+      command: "echo default",
+    });
+    expect(defaultRes.status).toBe(201);
+    expect((await defaultRes.json()) as { timeout_ms: number }).toMatchObject({
+      timeout_ms: 600_000,
+    });
+
+    const longRes = await call("POST", `/api/projects/${project.id}/crons`, {
+      name: "pipeline",
+      schedule: "30 7 * * *",
+      command: "run-pipeline",
+      timeout_ms: 6 * 60 * 60 * 1000,
+    });
+    expect(longRes.status).toBe(201);
+    expect((await longRes.json()) as { timeout_ms: number }).toMatchObject({
+      timeout_ms: 21_600_000,
+    });
+  });
+
+  test("create and update reject invalid timeouts", async () => {
+    const project = db
+      .query("INSERT INTO projects (name) VALUES ('invalid-timeout') RETURNING id")
+      .get() as { id: number };
+
+    const createRes = await call("POST", `/api/projects/${project.id}/crons`, {
+      name: "bad",
+      schedule: "0 3 * * *",
+      command: "echo bad",
+      timeout_ms: 999,
+    });
+    expect(createRes.status).toBe(400);
+    expect(await errorMessage(createRes)).toContain("timeout_ms must be an integer between");
+
+    const cron = db
+      .query(
+        "INSERT INTO crons (project_id, name, schedule, command) VALUES (?, 'c', '* * * * *', 'echo') RETURNING id",
+      )
+      .get(project.id) as { id: number };
+    const updateRes = await call("PUT", `/api/crons/${cron.id}`, { timeout_ms: 604_800_001 });
+    expect(updateRes.status).toBe(400);
+    expect(await errorMessage(updateRes)).toContain("timeout_ms must be an integer between");
+  });
+
+  test("update persists the timeout", async () => {
+    const project = db
+      .query("INSERT INTO projects (name) VALUES ('update-timeout') RETURNING id")
+      .get() as { id: number };
+    const cron = db
+      .query(
+        "INSERT INTO crons (project_id, name, schedule, command) VALUES (?, 'c', '* * * * *', 'echo') RETURNING id",
+      )
+      .get(project.id) as { id: number };
+
+    const res = await call("PUT", `/api/crons/${cron.id}`, { timeout_ms: 10_800_000 });
+    expect(res.status).toBe(200);
+    expect((await res.json()) as { timeout_ms: number }).toMatchObject({ timeout_ms: 10_800_000 });
+  });
+});

apps/api/routes/crons.ts

@@ -1,3 +1,9 @@
+import {
+  CRON_TIMEOUT_DEFAULT_MS,
+  isJsonObject,
+  validateCronSchedule,
+  validateCronTimeoutMs,
+} from "../../../packages/contract/src/index";
 import { runCron } from "../cron";
 import db from "../db";
 import { requireNotDraining } from "../drain";
@@ -50,6 +56,7 @@ export async function handleCrons(req: Request, url: URL): Promise<Response | nu
       name: string;
       schedule: string;
       command: string;
+      timeout_ms: number;
       enabled: number;
     } | null;
     if (!cron) return errorResponse("Cron not found", 404);
@@ -73,29 +80,62 @@ export async function handleCrons(req: Request, url: URL): Promise<Response | nu
 }
 
 async function handleCreate(req: Request, projectId: number): Promise<Response> {
-  const { name, schedule, command } = await req.json();
-  if (!name || !schedule || !command) {
+  const body: unknown = await req.json();
+  if (!isJsonObject(body)) return errorResponse("Request body must be an object", 400);
+
+  const { name, schedule, command } = body;
+  if (
+    typeof name !== "string" ||
+    typeof schedule !== "string" ||
+    typeof command !== "string" ||
+    !name.trim() ||
+    !schedule.trim() ||
+    !command.trim()
+  ) {
     return errorResponse("name, schedule, and command are required", 400);
   }
+  const scheduleError = validateCronSchedule(schedule);
+  if (scheduleError) return errorResponse(`Invalid schedule: ${scheduleError}`, 400);
+
+  const requestedTimeout = body.timeout_ms ?? CRON_TIMEOUT_DEFAULT_MS;
+  const timeoutError = validateCronTimeoutMs(requestedTimeout);
+  if (timeoutError) return errorResponse(timeoutError, 400);
+  const timeoutMs = requestedTimeout as number;
 
   const row = db
     .query(
-      "INSERT INTO crons (project_id, name, schedule, command) VALUES (?, ?, ?, ?) RETURNING *",
+      "INSERT INTO crons (project_id, name, schedule, command, timeout_ms) VALUES (?, ?, ?, ?, ?) RETURNING *",
     )
-    .get(projectId, name, schedule, command);
+    .get(projectId, name, schedule, command, timeoutMs);
 
   return Response.json(row, { status: 201 });
 }
 
 async function handleUpdate(req: Request, id: number): Promise<Response> {
-  const body = await req.json();
+  const body: unknown = await req.json();
+  if (!isJsonObject(body)) return errorResponse("Request body must be an object", 400);
+
+  if ("schedule" in body) {
+    if (typeof body.schedule !== "string") return errorResponse("schedule must be a string", 400);
+    const scheduleError = validateCronSchedule(body.schedule);
+    if (scheduleError) return errorResponse(`Invalid schedule: ${scheduleError}`, 400);
+  }
+  if ("timeout_ms" in body) {
+    const timeoutError = validateCronTimeoutMs(body.timeout_ms);
+    if (timeoutError) return errorResponse(timeoutError, 400);
+  }
+
   const fields: string[] = [];
   const values: (string | number)[] = [];
 
-  for (const key of ["name", "schedule", "command", "enabled"]) {
+  for (const key of ["name", "schedule", "command", "timeout_ms", "enabled"]) {
     if (key in body) {
       fields.push(`${key} = ?`);
-      values.push(body[key]);
+      const value = body[key];
+      if (typeof value !== "string" && typeof value !== "number" && typeof value !== "boolean") {
+        return errorResponse(`${key} has an invalid value`, 400);
+      }
+      values.push(typeof value === "boolean" ? Number(value) : value);
     }
   }