fix (cron): skip startup replay for interrupted running jobs

src/cron/service/ops.ts

@@ -22,6 +22,7 @@ export async function start(state: CronServiceState) {
     }
     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(
@@ -29,9 +30,10 @@ export async function start(state: CronServiceState) {
           "cron: clearing stale running marker on startup",
         );
         job.state.runningAtMs = undefined;
+        startupInterruptedJobIds.add(job.id);
       }
     }
-    await runMissedJobs(state);
+    await runMissedJobs(state, { skipJobIds: startupInterruptedJobIds });
     recomputeNextRuns(state);
     await persist(state);
     armTimer(state);

src/cron/service/timer.ts

@@ -357,11 +357,15 @@ function findDueJobs(state: CronServiceState): CronJob[] {
   });
 }
 
-export async function runMissedJobs(state: CronServiceState) {
+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 = state.store.jobs.filter((j) => {
     if (!j.state) {
       j.state = {};
@@ -369,6 +373,9 @@ export async function runMissedJobs(state: CronServiceState) {
     if (!j.enabled) {
       return false;
     }
+    if (skipJobIds?.has(j.id)) {
+      return false;
+    }
     if (typeof j.state.runningAtMs === "number") {
       return false;
     }