Serialize job persistence and bound scheduler drain

Claude-Session: https://claude.ai/code/session_01JtuZTCMW9TyogEmBgGf8JT

src/scheduler/scheduler.test.ts

@@ -6,6 +6,7 @@ import { createLogger } from "../log.js";
 import type { ScriptOutcome } from "./runner.js";
 import { Scheduler, validateJobInput } from "./scheduler.js";
 import { JobStore } from "./store.js";
+import type { Job } from "./types.js";
 
 const logger = createLogger("error", () => {});
 const PAST = "2020-01-01T00:00:00.000Z";
@@ -146,6 +147,72 @@ describe("Scheduler", () => {
     expect(await s.delete("nope1")).toBe(false);
     await s.stop();
   });
+
+  it("serializes persistence when jobs fire concurrently", async () => {
+    class OverlapStore extends JobStore {
+      active = 0;
+      maxActive = 0;
+      async save(jobs: Job[]) {
+        this.active++;
+        this.maxActive = Math.max(this.maxActive, this.active);
+        await new Promise((r) => setTimeout(r, 5));
+        await super.save(jobs);
+        this.active--;
+      }
+    }
+    const storePath = path.join(dir, "jobs.json");
+    const jobA: Job = {
+      id: "job01",
+      name: "a",
+      schedule: { at: PAST },
+      source: { kind: "text", text: "hello a" },
+      deliver: "direct",
+      createdAt: PAST,
+    };
+    const jobB: Job = {
+      id: "job02",
+      name: "b",
+      schedule: { at: PAST },
+      source: { kind: "text", text: "hello b" },
+      deliver: "direct",
+      createdAt: PAST,
+    };
+    await new JobStore(storePath).save([jobA, jobB]);
+
+    const direct: string[] = [];
+    const overlapStore = new OverlapStore(storePath);
+    const s = new Scheduler({
+      store: overlapStore,
+      workspaceDir: dir,
+      timezone: "UTC",
+      logger,
+      deliverDirect: async (t) => void direct.push(t),
+      deliverAgent: async () => {},
+    });
+    await s.start();
+    await s.drain();
+    expect([...direct].sort()).toEqual(["a: hello a", "b: hello b"]);
+    expect(overlapStore.maxActive).toBe(1);
+    expect(await new JobStore(storePath).load()).toEqual([]);
+    await s.stop();
+  });
+
+  it("drain() gives up after its timeout instead of hanging forever", async () => {
+    const s = new Scheduler({
+      store: new JobStore(path.join(dir, "jobs.json")),
+      workspaceDir: dir,
+      timezone: "UTC",
+      logger,
+      deliverDirect: () => new Promise(() => {}),
+      deliverAgent: async () => {},
+    });
+    await s.start();
+    await s.create({ name: "hangs", schedule: { at: PAST }, source: { kind: "text", text: "x" }, deliver: "direct" });
+    const started = Date.now();
+    await expect(s.drain(50)).resolves.toBeUndefined();
+    expect(Date.now() - started).toBeLessThan(1000);
+    await s.stop();
+  });
 });
 
 describe("validateJobInput", () => {

src/scheduler/scheduler.ts

@@ -31,6 +31,7 @@ export class Scheduler {
   private running = new Set<string>();
   private inflight = new Set<Promise<void>>();
   private stopped = false;
+  private persistChain: Promise<void> = Promise.resolve();
   private readonly runScript: (command: string, opts: RunScriptOptions) => Promise<ScriptOutcome>;
   private readonly now: () => Date;
 
@@ -49,12 +50,12 @@ export class Scheduler {
     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))]);
+    await this.drain();
   }
 
-  /** Resolves when every in-flight job run has finished. */
-  async drain(): Promise<void> {
-    while (this.inflight.size > 0) await Promise.all([...this.inflight]);
+  /** Resolves when every in-flight job run has finished, or after `timeoutMs`, whichever comes first. */
+  async drain(timeoutMs: number = STOP_GRACE_MS): Promise<void> {
+    await this.withTimeout(this.waitForInflight(), timeoutMs);
   }
 
   list(): JobView[] {
@@ -144,6 +145,22 @@ export class Scheduler {
     this.inflight.add(tracked);
   }
 
+  private async waitForInflight(): Promise<void> {
+    while (this.inflight.size > 0) await Promise.all([...this.inflight]);
+  }
+
+  private async withTimeout<T>(promise: Promise<T>, ms: number): Promise<T | undefined> {
+    let timer: NodeJS.Timeout | undefined;
+    const timeout = new Promise<undefined>((resolve) => {
+      timer = setTimeout(() => resolve(undefined), ms);
+    });
+    try {
+      return await Promise.race([promise, timeout]);
+    } finally {
+      if (timer) clearTimeout(timer);
+    }
+  }
+
   private arm(): void {
     if (this.stopped) return;
     let soonest: Date | undefined;
@@ -167,6 +184,8 @@ export class Scheduler {
       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;
+        // Failure reports always go directly to the owner, regardless of the job's deliver mode —
+        // alerts about broken jobs should never depend on a model call.
         if (!repeat) await this.safeDeliver(() => this.deps.deliverDirect(formatFailure(job, outcome)));
       } else {
         job.lastResult = { ok: true, at: startedAt };
@@ -206,8 +225,13 @@ export class Scheduler {
     }
   }
 
-  private async persist(): Promise<void> {
-    await this.deps.store.save(this.jobs);
+  private persist(): Promise<void> {
+    this.persistChain = this.persistChain
+      .then(() => this.deps.store.save(this.jobs))
+      .catch((err) => {
+        this.deps.logger.error("failed to persist jobs", err);
+      });
+    return this.persistChain;
   }
 }