diff --git a/src/cron/service/ops-mutations.ts b/src/cron/service/ops-mutations.ts index 93c0d8e62231..8298af540406 100644 --- a/src/cron/service/ops-mutations.ts +++ b/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( } 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( } throw error; } - for (const notify of postPersistNotifications) { - notify(); - } + runPostPersistCronNotifications(state, postPersistNotifications); for (const job of removedJobs) { noteActiveCronJobRemoval(job.id); try { diff --git a/src/cron/service/store.test.ts b/src/cron/service/store.test.ts index e91959d1f3ba..597c4b49701d 100644 --- a/src/cron/service/store.test.ts +++ b/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); }); diff --git a/src/cron/service/store.ts b/src/cron/service/store.ts index 158307e9841b..f5554ecd3aff 100644 --- a/src/cron/service/store.ts +++ b/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 = {}, ) { - 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;