Add scheduler with firing rules and failure dedupe
Claude-Session: https://claude.ai/code/session_01JtuZTCMW9TyogEmBgGf8JT
This commit is contained in:
EugeneTes
164
src/scheduler/scheduler.test.ts
Normal file
164
src/scheduler/scheduler.test.ts
Normal file
@@ -0,0 +1,164 @@
|
||||
import { mkdtemp, rm } from "node:fs/promises";
|
||||
import os from "node:os";
|
||||
import path from "node:path";
|
||||
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
|
||||
import { createLogger } from "../log.js";
|
||||
import type { ScriptOutcome } from "./runner.js";
|
||||
import { Scheduler, validateJobInput } from "./scheduler.js";
|
||||
import { JobStore } from "./store.js";
|
||||
|
||||
const logger = createLogger("error", () => {});
|
||||
const PAST = "2020-01-01T00:00:00.000Z";
|
||||
|
||||
function harness(runScript?: (command: string) => Promise<ScriptOutcome>) {
|
||||
const direct: string[] = [];
|
||||
const agent: string[] = [];
|
||||
const build = (dir: string) =>
|
||||
new Scheduler({
|
||||
store: new JobStore(path.join(dir, "jobs.json")),
|
||||
workspaceDir: dir,
|
||||
timezone: "UTC",
|
||||
logger,
|
||||
deliverDirect: async (t) => void direct.push(t),
|
||||
deliverAgent: async (t) => void agent.push(t),
|
||||
runScript: runScript ? (cmd) => runScript(cmd) : undefined,
|
||||
});
|
||||
return { direct, agent, build };
|
||||
}
|
||||
|
||||
describe("Scheduler", () => {
|
||||
let dir: string;
|
||||
beforeEach(async () => {
|
||||
dir = await mkdtemp(path.join(os.tmpdir(), "secretary-sched-"));
|
||||
});
|
||||
afterEach(async () => {
|
||||
vi.useRealTimers();
|
||||
await rm(dir, { recursive: true, force: true });
|
||||
});
|
||||
|
||||
it("fires an overdue one-shot text job directly and removes it", async () => {
|
||||
const h = harness();
|
||||
const s = h.build(dir);
|
||||
await s.start();
|
||||
const job = await s.create({ name: "dentist", schedule: { at: PAST }, source: { kind: "text", text: "call them" }, deliver: "direct" });
|
||||
await s.drain();
|
||||
expect(h.direct).toEqual(["dentist: call them"]);
|
||||
expect(s.get(job.id)).toBeUndefined();
|
||||
expect(await new JobStore(path.join(dir, "jobs.json")).load()).toEqual([]);
|
||||
await s.stop();
|
||||
});
|
||||
|
||||
it("delivers to the agent with the job prefix", async () => {
|
||||
const h = harness();
|
||||
const s = h.build(dir);
|
||||
await s.start();
|
||||
await s.create({ name: "news", schedule: { at: PAST }, source: { kind: "text", text: "summarize the news" }, deliver: "agent" });
|
||||
await s.drain();
|
||||
expect(h.agent).toEqual(['Scheduled job "news" fired with this output:\n\nsummarize the news']);
|
||||
await s.stop();
|
||||
});
|
||||
|
||||
it("stays silent when a script prints nothing, but records the run", async () => {
|
||||
vi.useFakeTimers({ now: new Date("2026-01-01T12:00:00Z") });
|
||||
const h = harness(async () => ({ kind: "silent" }));
|
||||
const s = h.build(dir);
|
||||
await s.start();
|
||||
const job = await s.create({ name: "watch", schedule: { cron: "* * * * *" }, source: { kind: "script", command: "python w.py" }, deliver: "agent" });
|
||||
await vi.advanceTimersByTimeAsync(60_000);
|
||||
await s.drain();
|
||||
expect(h.agent).toEqual([]);
|
||||
expect(h.direct).toEqual([]);
|
||||
expect(s.get(job.id)?.lastRunAt).toBeDefined();
|
||||
expect(s.get(job.id)?.lastResult?.ok).toBe(true);
|
||||
await s.stop();
|
||||
});
|
||||
|
||||
it("delivers script output to the agent", async () => {
|
||||
const h = harness(async () => ({ kind: "output", text: "price dropped to 9" }));
|
||||
const s = h.build(dir);
|
||||
await s.start();
|
||||
await s.create({ name: "price", schedule: { at: PAST }, source: { kind: "script", command: "python p.py" }, deliver: "agent" });
|
||||
await s.drain();
|
||||
expect(h.agent).toEqual(['Scheduled job "price" fired with this output:\n\nprice dropped to 9']);
|
||||
await s.stop();
|
||||
});
|
||||
|
||||
it("reports a failure once and again only after a success or a different exit code", async () => {
|
||||
const outcomes: ScriptOutcome[] = [
|
||||
{ kind: "failed", exitCode: 1, stderrTail: "no module", timedOut: false },
|
||||
{ kind: "failed", exitCode: 1, stderrTail: "no module", timedOut: false },
|
||||
{ kind: "failed", exitCode: 2, stderrTail: "worse", timedOut: false },
|
||||
{ kind: "silent" },
|
||||
{ kind: "failed", exitCode: 2, stderrTail: "worse again", timedOut: false },
|
||||
];
|
||||
let i = 0;
|
||||
const h = harness(async () => outcomes[i++]);
|
||||
const s = h.build(dir);
|
||||
await s.start();
|
||||
const job = await s.create({ name: "flaky", schedule: { cron: "* * * * *" }, source: { kind: "script", command: "x" }, deliver: "direct" });
|
||||
for (let n = 0; n < outcomes.length; n++) {
|
||||
await s.fireNow(job.id);
|
||||
}
|
||||
expect(h.direct).toHaveLength(3);
|
||||
expect(h.direct[0]).toMatch(/⚠️ Job "flaky" failed \(exit 1\):\nno module/);
|
||||
expect(h.direct[1]).toMatch(/exit 2/);
|
||||
expect(h.direct[2]).toMatch(/worse again/);
|
||||
await s.stop();
|
||||
});
|
||||
|
||||
it("computes the next cron occurrence strictly in the future and fires when the timer elapses", async () => {
|
||||
vi.useFakeTimers({ now: new Date("2026-01-01T12:00:00Z") });
|
||||
const h = harness();
|
||||
const s = h.build(dir);
|
||||
await s.start();
|
||||
const job = await s.create({ name: "tick", schedule: { cron: "*/5 * * * *" }, source: { kind: "text", text: "5 min passed" }, deliver: "direct" });
|
||||
expect(s.get(job.id)?.nextRunAt).toBe("2026-01-01T12:05:00.000Z");
|
||||
await vi.advanceTimersByTimeAsync(5 * 60_000);
|
||||
await s.drain();
|
||||
expect(h.direct).toEqual(["tick: 5 min passed"]);
|
||||
expect(s.get(job.id)?.nextRunAt).toBe("2026-01-01T12:10:00.000Z");
|
||||
await s.stop();
|
||||
});
|
||||
|
||||
it("skips an occurrence while the previous run is still going", async () => {
|
||||
let release!: () => void;
|
||||
const gate = new Promise<void>((r) => (release = r));
|
||||
let runs = 0;
|
||||
const h = harness(async () => {
|
||||
runs++;
|
||||
await gate;
|
||||
return { kind: "silent" };
|
||||
});
|
||||
const s = h.build(dir);
|
||||
await s.start();
|
||||
const job = await s.create({ name: "slow", schedule: { cron: "* * * * *" }, source: { kind: "script", command: "x" }, deliver: "direct" });
|
||||
const first = s.fireNow(job.id);
|
||||
await s.fireNow(job.id);
|
||||
expect(runs).toBe(1);
|
||||
release();
|
||||
await first;
|
||||
await s.stop();
|
||||
});
|
||||
|
||||
it("delete returns false for unknown ids", async () => {
|
||||
const s = harness().build(dir);
|
||||
await s.start();
|
||||
expect(await s.delete("nope1")).toBe(false);
|
||||
await s.stop();
|
||||
});
|
||||
});
|
||||
|
||||
describe("validateJobInput", () => {
|
||||
const ok = { name: "n", schedule: { at: "2030-01-01T00:00:00Z" }, source: { kind: "text" as const, text: "t" }, deliver: "agent" as const };
|
||||
it("accepts a valid job", () => {
|
||||
expect(validateJobInput(ok, "UTC")).toBeNull();
|
||||
});
|
||||
it("rejects missing name, bad schedule, empty text or command, bad deliver", () => {
|
||||
expect(validateJobInput({ ...ok, name: " " }, "UTC")).toMatch(/name/);
|
||||
expect(validateJobInput({ ...ok, schedule: { cron: "bad" } }, "UTC")).toMatch(/cron/);
|
||||
expect(validateJobInput({ ...ok, source: { kind: "text", text: "" } }, "UTC")).toMatch(/text/);
|
||||
expect(validateJobInput({ ...ok, source: { kind: "script", command: " " } }, "UTC")).toMatch(/command/);
|
||||
expect(validateJobInput({ ...ok, source: { kind: "script", command: "x", timeoutSec: 0 } }, "UTC")).toMatch(/timeoutSec/);
|
||||
expect(validateJobInput({ ...ok, deliver: "email" as never }, "UTC")).toMatch(/deliver/);
|
||||
});
|
||||
});
|
||||
233
src/scheduler/scheduler.ts
Normal file
233
src/scheduler/scheduler.ts
Normal file
@@ -0,0 +1,233 @@
|
||||
import type { Logger } from "../log.js";
|
||||
import { runScript as defaultRunScript, type RunScriptOptions, type ScriptOutcome } from "./runner.js";
|
||||
import { generateJobId, type JobStore, nextFireTime, validateSchedule } from "./store.js";
|
||||
import { DEFAULT_SCRIPT_TIMEOUT_SEC, type Job, type NewJobInput } from "./types.js";
|
||||
|
||||
export interface SchedulerDeps {
|
||||
store: JobStore;
|
||||
workspaceDir: string;
|
||||
timezone: string;
|
||||
logger: Logger;
|
||||
deliverDirect: (text: string) => Promise<void>;
|
||||
deliverAgent: (text: string) => Promise<void>;
|
||||
runScript?: (command: string, opts: RunScriptOptions) => Promise<ScriptOutcome>;
|
||||
shellPath?: string;
|
||||
now?: () => Date;
|
||||
}
|
||||
|
||||
export interface JobView extends Job {
|
||||
nextRunAt: string | null;
|
||||
}
|
||||
|
||||
type FailedOutcome = Extract<ScriptOutcome, { kind: "failed" }>;
|
||||
|
||||
const MAX_TIMER_MS = 60 * 60 * 1000;
|
||||
const STOP_GRACE_MS = 10_000;
|
||||
|
||||
export class Scheduler {
|
||||
private jobs: Job[] = [];
|
||||
private nextTimes = new Map<string, Date>();
|
||||
private timer: NodeJS.Timeout | undefined;
|
||||
private running = new Set<string>();
|
||||
private inflight = new Set<Promise<void>>();
|
||||
private stopped = false;
|
||||
private readonly runScript: (command: string, opts: RunScriptOptions) => Promise<ScriptOutcome>;
|
||||
private readonly now: () => Date;
|
||||
|
||||
constructor(private readonly deps: SchedulerDeps) {
|
||||
this.runScript = deps.runScript ?? defaultRunScript;
|
||||
this.now = deps.now ?? (() => new Date());
|
||||
}
|
||||
|
||||
async start(): Promise<void> {
|
||||
this.jobs = await this.deps.store.load();
|
||||
this.stopped = false;
|
||||
await this.tick();
|
||||
}
|
||||
|
||||
async stop(): Promise<void> {
|
||||
this.stopped = true;
|
||||
if (this.timer) clearTimeout(this.timer);
|
||||
this.timer = undefined;
|
||||
await Promise.race([this.drain(), new Promise((r) => setTimeout(r, STOP_GRACE_MS))]);
|
||||
}
|
||||
|
||||
/** Resolves when every in-flight job run has finished. */
|
||||
async drain(): Promise<void> {
|
||||
while (this.inflight.size > 0) await Promise.all([...this.inflight]);
|
||||
}
|
||||
|
||||
list(): JobView[] {
|
||||
const now = this.now();
|
||||
return this.jobs.map((job) => ({
|
||||
...job,
|
||||
nextRunAt: (this.nextTimes.get(job.id) ?? nextFireTime(job, now, this.deps.timezone))?.toISOString() ?? null,
|
||||
}));
|
||||
}
|
||||
|
||||
get(id: string): JobView | undefined {
|
||||
return this.list().find((j) => j.id === id);
|
||||
}
|
||||
|
||||
async create(input: NewJobInput): Promise<JobView> {
|
||||
const problem = validateJobInput(input, this.deps.timezone);
|
||||
if (problem) throw new Error(problem);
|
||||
let id = generateJobId();
|
||||
while (this.jobs.some((j) => j.id === id)) id = generateJobId();
|
||||
const job: Job = {
|
||||
id,
|
||||
name: input.name.trim(),
|
||||
schedule: input.schedule,
|
||||
source: input.source,
|
||||
deliver: input.deliver,
|
||||
createdAt: this.now().toISOString(),
|
||||
};
|
||||
this.jobs.push(job);
|
||||
await this.persist();
|
||||
await this.tick();
|
||||
const view = this.get(id);
|
||||
if (!view) throw new Error(`job ${id} vanished after creation`);
|
||||
return view;
|
||||
}
|
||||
|
||||
async delete(id: string): Promise<boolean> {
|
||||
const before = this.jobs.length;
|
||||
this.jobs = this.jobs.filter((j) => j.id !== id);
|
||||
if (this.jobs.length === before) return false;
|
||||
this.nextTimes.delete(id);
|
||||
await this.persist();
|
||||
await this.tick();
|
||||
return true;
|
||||
}
|
||||
|
||||
/** Fires every due job (without waiting for them) and arms the timer for the next one. */
|
||||
async tick(): Promise<void> {
|
||||
if (this.stopped) return;
|
||||
if (this.timer) clearTimeout(this.timer);
|
||||
this.timer = undefined;
|
||||
const now = this.now();
|
||||
|
||||
for (const job of this.jobs) {
|
||||
if (!this.nextTimes.has(job.id)) {
|
||||
const t = nextFireTime(job, now, this.deps.timezone);
|
||||
if (t) this.nextTimes.set(job.id, t);
|
||||
}
|
||||
}
|
||||
for (const id of [...this.nextTimes.keys()]) {
|
||||
if (!this.jobs.some((j) => j.id === id)) this.nextTimes.delete(id);
|
||||
}
|
||||
|
||||
for (const job of this.jobs) {
|
||||
const due = this.nextTimes.get(job.id);
|
||||
if (!due || due.getTime() > now.getTime()) continue;
|
||||
if ("cron" in job.schedule) {
|
||||
const next = nextFireTime(job, now, this.deps.timezone);
|
||||
if (next) this.nextTimes.set(job.id, next);
|
||||
else this.nextTimes.delete(job.id);
|
||||
} else {
|
||||
this.nextTimes.delete(job.id);
|
||||
}
|
||||
this.track(this.fire(job));
|
||||
}
|
||||
this.arm();
|
||||
}
|
||||
|
||||
/** Runs one job now, waiting for it. Used by tests and by tick(). */
|
||||
async fireNow(id: string): Promise<void> {
|
||||
const job = this.jobs.find((j) => j.id === id);
|
||||
if (!job) throw new Error(`no job with id ${id}`);
|
||||
await this.fire(job);
|
||||
}
|
||||
|
||||
private track(p: Promise<void>): void {
|
||||
const tracked: Promise<void> = p.finally(() => this.inflight.delete(tracked));
|
||||
this.inflight.add(tracked);
|
||||
}
|
||||
|
||||
private arm(): void {
|
||||
if (this.stopped) return;
|
||||
let soonest: Date | undefined;
|
||||
for (const t of this.nextTimes.values()) if (!soonest || t < soonest) soonest = t;
|
||||
if (!soonest) return;
|
||||
const delay = Math.min(Math.max(soonest.getTime() - this.now().getTime(), 0), MAX_TIMER_MS);
|
||||
this.timer = setTimeout(() => void this.tick(), delay);
|
||||
}
|
||||
|
||||
private async fire(job: Job): Promise<void> {
|
||||
if (this.running.has(job.id)) {
|
||||
this.deps.logger.warn(`job ${job.id} (${job.name}) is still running, skipping this occurrence`);
|
||||
return;
|
||||
}
|
||||
this.running.add(job.id);
|
||||
try {
|
||||
const startedAt = this.now().toISOString();
|
||||
const outcome = await this.evaluate(job);
|
||||
const previous = job.lastResult;
|
||||
job.lastRunAt = startedAt;
|
||||
if (outcome.kind === "failed") {
|
||||
job.lastResult = { ok: false, exitCode: outcome.exitCode, stderrTail: outcome.stderrTail, at: startedAt };
|
||||
const repeat = previous !== undefined && !previous.ok && previous.exitCode === outcome.exitCode;
|
||||
if (!repeat) await this.safeDeliver(() => this.deps.deliverDirect(formatFailure(job, outcome)));
|
||||
} else {
|
||||
job.lastResult = { ok: true, at: startedAt };
|
||||
if (outcome.kind === "output") {
|
||||
if (job.deliver === "direct") {
|
||||
await this.safeDeliver(() => this.deps.deliverDirect(`${job.name}: ${outcome.text}`));
|
||||
} else {
|
||||
await this.safeDeliver(() =>
|
||||
this.deps.deliverAgent(`Scheduled job "${job.name}" fired with this output:\n\n${outcome.text}`),
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
if ("at" in job.schedule) this.jobs = this.jobs.filter((j) => j.id !== job.id);
|
||||
await this.persist();
|
||||
} catch (err) {
|
||||
this.deps.logger.error(`job ${job.id} (${job.name}) crashed`, err);
|
||||
} finally {
|
||||
this.running.delete(job.id);
|
||||
}
|
||||
}
|
||||
|
||||
private async evaluate(job: Job): Promise<ScriptOutcome> {
|
||||
if (job.source.kind === "text") return { kind: "output", text: job.source.text };
|
||||
return this.runScript(job.source.command, {
|
||||
cwd: this.deps.workspaceDir,
|
||||
timeoutSec: job.source.timeoutSec ?? DEFAULT_SCRIPT_TIMEOUT_SEC,
|
||||
shellPath: this.deps.shellPath,
|
||||
});
|
||||
}
|
||||
|
||||
private async safeDeliver(deliver: () => Promise<void>): Promise<void> {
|
||||
try {
|
||||
await deliver();
|
||||
} catch (err) {
|
||||
this.deps.logger.error("job delivery failed", err);
|
||||
}
|
||||
}
|
||||
|
||||
private async persist(): Promise<void> {
|
||||
await this.deps.store.save(this.jobs);
|
||||
}
|
||||
}
|
||||
|
||||
export function validateJobInput(input: NewJobInput, timezone: string): string | null {
|
||||
if (!input.name || !input.name.trim()) return "name is required";
|
||||
if (!input.schedule || !("at" in input.schedule || "cron" in input.schedule)) return "schedule must have at or cron";
|
||||
const scheduleProblem = validateSchedule(input.schedule, timezone);
|
||||
if (scheduleProblem) return scheduleProblem;
|
||||
if (!input.source) return "source is required";
|
||||
if (input.source.kind === "text" && !input.source.text.trim()) return "text is required";
|
||||
if (input.source.kind === "script") {
|
||||
if (!input.source.command.trim()) return "command is required";
|
||||
if (input.source.timeoutSec !== undefined && !(input.source.timeoutSec > 0)) return "timeoutSec must be a positive number";
|
||||
}
|
||||
if (input.deliver !== "agent" && input.deliver !== "direct") return "deliver must be agent or direct";
|
||||
return null;
|
||||
}
|
||||
|
||||
export function formatFailure(job: Job, outcome: FailedOutcome): string {
|
||||
const code = outcome.exitCode ?? "?";
|
||||
const suffix = outcome.timedOut ? ", timed out" : "";
|
||||
return `⚠️ Job "${job.name}" failed (exit ${code}${suffix}):\n${outcome.stderrTail}`;
|
||||
}
|
||||
Reference in New Issue
Block a user