Cron: keep list/status responsive during startup catch-up

src/cron/service/ops.ts

@@ -28,14 +28,15 @@ async function ensureLoadedForRead(state: CronServiceState) {
 }
 
 export async function start(state: CronServiceState) {
+  if (!state.deps.cronEnabled) {
+    state.deps.log.info({ enabled: false }, "cron: disabled");
+    return;
+  }
+
+  const startupInterruptedJobIds = new Set<string>();
   await locked(state, async () => {
-    if (!state.deps.cronEnabled) {
-      state.deps.log.info({ enabled: false }, "cron: disabled");
-      return;
-    }
     await ensureLoaded(state, { skipRecompute: true });
     const jobs = state.store?.jobs ?? [];
-    const startupInterruptedJobIds = new Set<string>();
     for (const job of jobs) {
       if (typeof job.state.runningAtMs === "number") {
         state.deps.log.warn(
@@ -46,7 +47,13 @@ export async function start(state: CronServiceState) {
         startupInterruptedJobIds.add(job.id);
       }
     }
-    await runMissedJobs(state, { skipJobIds: startupInterruptedJobIds });
+    await persist(state);
+  });
+
+  await runMissedJobs(state, { skipJobIds: startupInterruptedJobIds });
+
+  await locked(state, async () => {
+    await ensureLoaded(state, { forceReload: true, skipRecompute: true });
     recomputeNextRuns(state);
     await persist(state);
     armTimer(state);

src/cron/service/timer.ts

@@ -458,22 +458,97 @@ export async function runMissedJobs(
   state: CronServiceState,
   opts?: { skipJobIds?: ReadonlySet<string> },
 ) {
-  if (!state.store) {
-    return;
-  }
-  const now = state.deps.nowMs();
-  const skipJobIds = opts?.skipJobIds;
-  const missed = collectRunnableJobs(state, now, { skipJobIds, skipAtIfAlreadyRan: true });
-
-  if (missed.length > 0) {
+  const startupCandidates = await locked(state, async () => {
+    await ensureLoaded(state, { skipRecompute: true });
+    if (!state.store) {
+      return [] as Array<{ jobId: string; job: CronJob }>;
+    }
+    const now = state.deps.nowMs();
+    const skipJobIds = opts?.skipJobIds;
+    const missed = collectRunnableJobs(state, now, { skipJobIds, skipAtIfAlreadyRan: true });
+    if (missed.length === 0) {
+      return [] as Array<{ jobId: string; job: CronJob }>;
+    }
     state.deps.log.info(
       { count: missed.length, jobIds: missed.map((j) => j.id) },
       "cron: running missed jobs after restart",
     );
     for (const job of missed) {
-      await executeJob(state, job, now, { forced: false });
+      job.state.runningAtMs = now;
+      job.state.lastError = undefined;
+    }
+    await persist(state);
+    return missed.map((job) => ({ jobId: job.id, job }));
+  });
+
+  if (startupCandidates.length === 0) {
+    return;
+  }
+
+  const outcomes: Array<TimedCronRunOutcome> = [];
+  for (const candidate of startupCandidates) {
+    const startedAt = state.deps.nowMs();
+    emit(state, { jobId: candidate.job.id, action: "started", runAtMs: startedAt });
+    try {
+      const result = await executeJobCore(state, candidate.job);
+      outcomes.push({
+        jobId: candidate.jobId,
+        status: result.status,
+        error: result.error,
+        summary: result.summary,
+        delivered: result.delivered,
+        sessionId: result.sessionId,
+        sessionKey: result.sessionKey,
+        model: result.model,
+        provider: result.provider,
+        usage: result.usage,
+        startedAt,
+        endedAt: state.deps.nowMs(),
+      });
+    } catch (err) {
+      outcomes.push({
+        jobId: candidate.jobId,
+        status: "error",
+        error: String(err),
+        startedAt,
+        endedAt: state.deps.nowMs(),
+      });
     }
   }
+
+  await locked(state, async () => {
+    await ensureLoaded(state, { forceReload: true, skipRecompute: true });
+    if (!state.store) {
+      return;
+    }
+
+    for (const result of outcomes) {
+      const job = state.store.jobs.find((entry) => entry.id === result.jobId);
+      if (!job) {
+        continue;
+      }
+      const shouldDelete = applyJobResult(state, job, {
+        status: result.status,
+        error: result.error,
+        delivered: result.delivered,
+        startedAt: result.startedAt,
+        endedAt: result.endedAt,
+      });
+
+      emitJobFinished(state, job, result, result.startedAt);
+
+      if (shouldDelete) {
+        state.store.jobs = state.store.jobs.filter((entry) => entry.id !== job.id);
+        emit(state, { jobId: job.id, action: "removed" });
+      }
+    }
+
+    // Preserve any new past-due nextRunAtMs values that became due while
+    // startup catch-up was running. They should execute on a future tick
+    // instead of being silently advanced.
+    recomputeNextRunsForMaintenance(state);
+    await persist(state);
+  });
 }
 
 export async function runDueJobs(state: CronServiceState) {