fix(cron): isolate post-persist notification failures from the store write (#120266)

src/cron/service/ops-mutations.ts

@@ -43,6 +43,7 @@ import {
   ensureLoaded,
   persist,
   persistOrRestore,
+  runPostPersistCronNotifications,
   snapshotStoreForRollback,
   type CronRollbackSnapshot,
   warnIfDisabled,
@@ -514,9 +515,7 @@ export async function removeAgentJobsTransactional<T>(
     } catch (error) {
       if (error instanceof AgentDeletionCommitUncertainError) {
         // Uncertain roster writes intentionally keep the cron deletion durable.
-        for (const notify of postPersistNotifications) {
-          notify();
-        }
+        runPostPersistCronNotifications(state, postPersistNotifications);
         armTimer(state);
         for (const job of removedJobs) {
           noteActiveCronJobRemoval(job.id);
@@ -540,9 +539,7 @@ export async function removeAgentJobsTransactional<T>(
       }
       throw error;
     }
-    for (const notify of postPersistNotifications) {
-      notify();
-    }
+    runPostPersistCronNotifications(state, postPersistNotifications);
     for (const job of removedJobs) {
       noteActiveCronJobRemoval(job.id);
       try {

src/cron/service/store.test.ts

@@ -233,7 +233,9 @@ describe("cron service store seam coverage", () => {
     expect(notify).toHaveBeenCalledOnce();
   });
 
-  it("does not restore speculative state after a post-persist notification throws", async () => {
+  it("contains a throwing post-persist notification without dropping siblings or the write", async () => {
+    // A notification failure happens after the durable commit: it must not
+    // reject the persist, skip sibling notifications, or roll back the store.
     const { storePath } = await makeStorePath();
     const nextRunAtMs = STORE_TEST_NOW + 120_000;
     await writeSingleJobStore(storePath, createReloadCronJob());
@@ -242,17 +244,18 @@ describe("cron service store seam coverage", () => {
     const snapshot = snapshotStoreForRollback(state);
     const job = findJobOrThrow(state, "reload-cron-expr-job");
     job.state.nextRunAtMs = nextRunAtMs;
+    const siblingNotify = vi.fn();
 
-    await expect(
-      persistOrRestore(state, snapshot, {
-        postPersistNotifications: [
-          () => {
-            throw new Error("notification failed");
-          },
-        ],
-      }),
-    ).rejects.toThrow("notification failed");
+    await persistOrRestore(state, snapshot, {
+      postPersistNotifications: [
+        () => {
+          throw new Error("notification failed");
+        },
+        siblingNotify,
+      ],
+    });
 
+    expect(siblingNotify).toHaveBeenCalledOnce();
     expect(job.state.nextRunAtMs).toBe(nextRunAtMs);
     expect((await loadCronStore(storePath)).jobs[0]?.state.nextRunAtMs).toBe(nextRunAtMs);
   });

src/cron/service/store.ts

@@ -294,12 +294,32 @@ export async function persist(state: CronServiceState, opts?: PersistOptions) {
     stateOnly,
     suppressScheduledJobId: opts?.suppressScheduledJobId,
   });
-  for (const notify of opts?.postPersistNotifications ?? []) {
-    notify();
-  }
+  runPostPersistCronNotifications(state, opts?.postPersistNotifications);
   return true;
 }
 
+/**
+ * Notifications run after the durable commit; one throwing notify (e.g. an
+ * auto-disable notice for a removed agent) must not drop its siblings or
+ * masquerade as a store-write failure — at startup that keeps the whole
+ * scheduler down.
+ */
+export function runPostPersistCronNotifications(
+  state: CronServiceState,
+  notifications: DeferredCronNotifications | undefined,
+) {
+  for (const notify of notifications ?? []) {
+    try {
+      notify();
+    } catch (err) {
+      state.deps.log.warn(
+        { error: err instanceof Error ? err.message : String(err) },
+        "cron: post-persist notification failed",
+      );
+    }
+  }
+}
+
 /** Captures the live cron state that must stay aligned with the durable store. */
 export function snapshotStoreForRollback(state: CronServiceState): CronRollbackSnapshot {
   return {
@@ -315,23 +335,14 @@ export async function persistOrRestore(
   snapshot: CronRollbackSnapshot,
   opts: Omit<PersistOptions, "stateOnly"> = {},
 ) {
-  let writeCompleted = false;
-  const postPersistNotifications = opts.postPersistNotifications?.map((notify) => () => {
-    // Notification failures happen after commit and must not restore the
-    // speculative snapshot over rows that are already durable.
-    writeCompleted = true;
-    notify();
-  });
   try {
-    const persisted = await persist(state, { ...opts, postPersistNotifications });
+    // Notification failures are contained inside persist(), so a throw here
+    // always means the durable write itself failed and the snapshot must win.
+    const persisted = await persist(state, opts);
     if (!persisted) {
       throw new Error("cron: durable store write did not complete");
     }
-    writeCompleted = true;
   } catch (err) {
-    if (writeCompleted) {
-      throw err;
-    }
     state.store = snapshot.store;
     state.durableNextRunAtMsByJobId = snapshot.durableNextRunAtMsByJobId;
     throw err;