Skip to content

Commit e728957

Browse files
joshavantjoshavant
authored
Fix disabled heartbeat one-shot cron retries (#92225)
* fix: retry disabled cron wake one-shots * fix: satisfy cron retry CI checks
1 parent d9124c9 commit e728957

7 files changed

Lines changed: 411 additions & 31 deletions

File tree

‎src/cron/service.restart-catchup.test.ts‎

Lines changed: 61 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -74,6 +74,28 @@ describe("CronService restart catch-up", () => {
7474
};
7575
}
7676

77+
function createOverdueDisabledHeartbeatOneShotRetry(id: string, nextRunAtMs: number): CronJob {
78+
return {
79+
id,
80+
name: `disabled-heartbeat-retry-${id}`,
81+
enabled: true,
82+
createdAtMs: nextRunAtMs - 60_000,
83+
updatedAtMs: nextRunAtMs - 30_000,
84+
deleteAfterRun: true,
85+
schedule: { kind: "at", at: new Date(nextRunAtMs - 30_000).toISOString() },
86+
sessionTarget: "main",
87+
wakeMode: "now",
88+
payload: { kind: "systemEvent", text: `retry-${id}` },
89+
state: {
90+
nextRunAtMs,
91+
lastRunAtMs: nextRunAtMs - 30_000,
92+
lastRunStatus: "skipped",
93+
lastError: "disabled",
94+
consecutiveSkipped: 1,
95+
},
96+
};
97+
}
98+
7799
function expectQueuedSystemEvent(
78100
enqueueSystemEvent: ReturnType<typeof vi.fn>,
79101
expectedText: string,
@@ -658,4 +680,43 @@ describe("CronService restart catch-up", () => {
658680

659681
await store.cleanup();
660682
});
683+
684+
it("stagger-limits overdue disabled-heartbeat one-shot retries after restart", async () => {
685+
const store = await makeStorePath();
686+
const startNow = Date.parse("2025-12-13T17:00:00.000Z");
687+
688+
await writeStoreJobs(store.storePath, [
689+
createOverdueDisabledHeartbeatOneShotRetry("disabled-retry-0", startNow - 60_000),
690+
createOverdueDisabledHeartbeatOneShotRetry("disabled-retry-1", startNow - 45_000),
691+
]);
692+
693+
const enqueueSystemEvent = vi.fn();
694+
const requestHeartbeat = vi.fn();
695+
const state = createCronServiceState({
696+
cronEnabled: true,
697+
storePath: store.storePath,
698+
log: noopLogger,
699+
nowMs: () => startNow,
700+
enqueueSystemEvent,
701+
requestHeartbeat,
702+
runIsolatedAgentJob: vi.fn(async () => ({ status: "ok" as const })),
703+
maxMissedJobsPerRestart: 1,
704+
missedJobStaggerMs: 5_000,
705+
});
706+
707+
await runMissedJobs(state);
708+
709+
expectQueuedSystemEvent(enqueueSystemEvent, "retry-disabled-retry-0");
710+
expect(requestHeartbeat).toHaveBeenCalledTimes(1);
711+
712+
const listedJobs = state.store?.jobs ?? [];
713+
expect(listedJobs.find((job) => job.id === "disabled-retry-0")).toBeUndefined();
714+
const deferred = listedJobs.find((job) => job.id === "disabled-retry-1");
715+
expect(deferred?.enabled).toBe(true);
716+
expect(deferred?.state.lastRunStatus).toBe("skipped");
717+
expect(deferred?.state.lastError).toBe("disabled");
718+
expect(deferred?.state.nextRunAtMs).toBe(startNow + 5_000);
719+
720+
await store.cleanup();
721+
});
661722
});

‎src/cron/service.runs-one-shot-main-job-disables-it.test.ts‎

Lines changed: 137 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -5,6 +5,13 @@ import {
55
HEARTBEAT_SKIP_REQUESTS_IN_FLIGHT,
66
type HeartbeatRunResult,
77
} from "../infra/heartbeat-wake.js";
8+
import {
9+
consumeSelectedSystemEventEntries,
10+
drainSystemEventEntries,
11+
enqueueSystemEventEntry,
12+
peekSystemEventEntries,
13+
resetSystemEventsForTest,
14+
} from "../infra/system-events.js";
815
import type { CronEvent, CronServiceDeps } from "./service.js";
916
import { CronService } from "./service.js";
1017
import {
@@ -60,14 +67,34 @@ type CronHarnessOptions = {
6067
runIsolatedAgentJob?: CronServiceDeps["runIsolatedAgentJob"];
6168
runHeartbeatOnce?: NonNullable<CronServiceDeps["runHeartbeatOnce"]>;
6269
nowMs?: () => number;
70+
cronConfig?: CronServiceDeps["cronConfig"];
71+
useRemovableSystemEventQueue?: boolean;
6372
wakeNowHeartbeatBusyMaxWaitMs?: number;
6473
wakeNowHeartbeatBusyRetryDelayMs?: number;
6574
withEvents?: boolean;
6675
};
6776

6877
async function createCronHarness(options: CronHarnessOptions = {}) {
6978
const store = await makeStorePath();
70-
const enqueueSystemEvent = vi.fn();
79+
const enqueueSystemEvent = options.useRemovableSystemEventQueue
80+
? vi.fn((text: string, opts?: Parameters<CronServiceDeps["enqueueSystemEvent"]>[1]) => {
81+
if (!opts?.sessionKey) {
82+
throw new Error("test removable queue requires a sessionKey");
83+
}
84+
const event = enqueueSystemEventEntry(text, {
85+
sessionKey: opts.sessionKey,
86+
contextKey: opts.contextKey,
87+
deliveryContext: opts.deliveryContext,
88+
});
89+
return event
90+
? {
91+
accepted: true,
92+
remove: () =>
93+
consumeSelectedSystemEventEntries(opts.sessionKey as string, [event]).length > 0,
94+
}
95+
: { accepted: false };
96+
})
97+
: vi.fn();
7198
const requestHeartbeat = vi.fn();
7299
const events = options.withEvents === false ? undefined : createCronEventHarness();
73100

@@ -76,6 +103,7 @@ async function createCronHarness(options: CronHarnessOptions = {}) {
76103
cronEnabled: true,
77104
log: noopLogger,
78105
...(options.nowMs ? { nowMs: options.nowMs } : {}),
106+
...(options.cronConfig ? { cronConfig: options.cronConfig } : {}),
79107
...(options.wakeNowHeartbeatBusyMaxWaitMs !== undefined
80108
? { wakeNowHeartbeatBusyMaxWaitMs: options.wakeNowHeartbeatBusyMaxWaitMs }
81109
: {}),
@@ -236,10 +264,23 @@ function expectQueuedCronHeartbeat(
236264
expectCronRunSessionKey(request?.sessionKey, params.jobId);
237265
}
238266

267+
function getPostedSystemEventSessionKeys(enqueueSystemEvent: ReturnType<typeof vi.fn>) {
268+
return enqueueSystemEvent.mock.calls
269+
.map(([, options]) => (options as { sessionKey?: string } | undefined)?.sessionKey)
270+
.filter((sessionKey): sessionKey is string => Boolean(sessionKey));
271+
}
272+
273+
function expectNoQueuedEvents(sessionKeys: readonly string[]) {
274+
for (const sessionKey of sessionKeys) {
275+
expect(peekSystemEventEntries(sessionKey)).toHaveLength(0);
276+
}
277+
}
278+
239279
async function stopCronAndCleanup(cron: CronService, store: { cleanup: () => Promise<void> }) {
240280
await cron.status();
241281
cron.stop();
242282
await store.cleanup();
283+
resetSystemEventsForTest();
243284
}
244285

245286
function createStartedCronService(
@@ -449,6 +490,101 @@ describe("CronService", () => {
449490
await stopCronAndCleanup(cron, store);
450491
});
451492

493+
it("retries disabled one-shot main wakes without leaving failed-attempt system events", async () => {
494+
resetSystemEventsForTest();
495+
const atMs = Date.parse("2025-12-13T00:00:02.000Z");
496+
let now = atMs;
497+
const consumedTexts: string[] = [];
498+
const runHeartbeatOnce = vi.fn(
499+
async (opts?: Parameters<NonNullable<CronServiceDeps["runHeartbeatOnce"]>>[0]) => {
500+
if (runHeartbeatOnce.mock.calls.length < 3) {
501+
return { status: "skipped" as const, reason: "disabled" };
502+
}
503+
const sessionKey = opts?.sessionKey;
504+
if (sessionKey) {
505+
consumedTexts.push(...drainSystemEventEntries(sessionKey).map((event) => event.text));
506+
}
507+
return { status: "ran" as const, durationMs: 1 };
508+
},
509+
);
510+
const { store, cron, enqueueSystemEvent, requestHeartbeat } = await createCronHarness({
511+
runHeartbeatOnce,
512+
nowMs: () => now,
513+
useRemovableSystemEventQueue: true,
514+
withEvents: false,
515+
});
516+
const job = await addMainOneShotHelloJob(cron, {
517+
atMs,
518+
name: "one-shot disabled heartbeat retries cleanly",
519+
});
520+
521+
await cron.run(job.id, "due");
522+
let jobs = await cron.list({ includeDisabled: true });
523+
let updated = jobs.find((j) => j.id === job.id);
524+
expect(updated?.enabled).toBe(true);
525+
expect(updated?.state.lastStatus).toBe("skipped");
526+
expect(updated?.state.lastError).toBe("disabled");
527+
expect(updated?.state.consecutiveSkipped).toBe(1);
528+
expect(updated?.state.nextRunAtMs).toBe(atMs + 30_000);
529+
expectNoQueuedEvents(getPostedSystemEventSessionKeys(enqueueSystemEvent));
530+
531+
now = updated?.state.nextRunAtMs ?? now;
532+
await cron.run(job.id, "due");
533+
jobs = await cron.list({ includeDisabled: true });
534+
updated = jobs.find((j) => j.id === job.id);
535+
expect(updated?.enabled).toBe(true);
536+
expect(updated?.state.consecutiveSkipped).toBe(2);
537+
expect(updated?.state.nextRunAtMs).toBe(atMs + 90_000);
538+
expectNoQueuedEvents(getPostedSystemEventSessionKeys(enqueueSystemEvent));
539+
540+
now = updated?.state.nextRunAtMs ?? now;
541+
await cron.run(job.id, "due");
542+
543+
jobs = await cron.list({ includeDisabled: true });
544+
expect(jobs.find((j) => j.id === job.id)).toBeUndefined();
545+
expect(runHeartbeatOnce).toHaveBeenCalledTimes(3);
546+
expect(requestHeartbeat).not.toHaveBeenCalled();
547+
expect(consumedTexts).toEqual(["hello"]);
548+
expectNoQueuedEvents(getPostedSystemEventSessionKeys(enqueueSystemEvent));
549+
550+
await stopCronAndCleanup(cron, store);
551+
});
552+
553+
it("disables exhausted disabled-heartbeat one-shots without leaving queued events", async () => {
554+
resetSystemEventsForTest();
555+
const atMs = Date.parse("2025-12-13T00:00:02.000Z");
556+
const runHeartbeatOnce = vi.fn(async () => ({
557+
status: "skipped" as const,
558+
reason: "disabled",
559+
}));
560+
const { store, cron, enqueueSystemEvent, requestHeartbeat } = await createCronHarness({
561+
runHeartbeatOnce,
562+
nowMs: () => atMs,
563+
cronConfig: { retry: { maxAttempts: 0, backoffMs: [30_000] } },
564+
useRemovableSystemEventQueue: true,
565+
withEvents: false,
566+
});
567+
const job = await addMainOneShotHelloJob(cron, {
568+
atMs,
569+
name: "one-shot disabled heartbeat exhausted",
570+
});
571+
572+
await cron.run(job.id, "due");
573+
574+
const jobs = await cron.list({ includeDisabled: true });
575+
const updated = jobs.find((j) => j.id === job.id);
576+
expect(updated?.enabled).toBe(false);
577+
expect(updated?.state.lastStatus).toBe("skipped");
578+
expect(updated?.state.lastError).toBe("disabled");
579+
expect(updated?.state.consecutiveSkipped).toBe(1);
580+
expect(updated?.state.nextRunAtMs).toBeUndefined();
581+
expect(runHeartbeatOnce).toHaveBeenCalledTimes(1);
582+
expect(requestHeartbeat).not.toHaveBeenCalled();
583+
expectNoQueuedEvents(getPostedSystemEventSessionKeys(enqueueSystemEvent));
584+
585+
await stopCronAndCleanup(cron, store);
586+
});
587+
452588
it("runs an isolated job without posting a fallback summary to main", async () => {
453589
const runIsolatedAgentJob = vi.fn(async () => ({ status: "ok" as const, summary: "done" }));
454590
const { store, cron, enqueueSystemEvent, requestHeartbeat, events } =

‎src/cron/service/state.ts‎

Lines changed: 9 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -51,6 +51,14 @@ export type Logger = {
5151
error: (obj: unknown, msg?: string) => void;
5252
};
5353

54+
export type CronSystemEventEnqueueResult =
55+
| boolean
56+
| void
57+
| {
58+
accepted?: boolean;
59+
remove?: () => boolean | void;
60+
};
61+
5462
/** Dependency injection surface for the cron service runtime. */
5563
export type CronServiceDeps = {
5664
nowMs?: () => number;
@@ -90,7 +98,7 @@ export type CronServiceDeps = {
9098
contextKey?: string;
9199
deliveryContext?: DeliveryContext;
92100
},
93-
) => void;
101+
) => CronSystemEventEnqueueResult;
94102
/**
95103
* Resolve the channel-correct origin delivery context for a session key (the
96104
* value the channel's send expects, e.g. Telegram message_thread_id), sourced

0 commit comments

Comments
 (0)