Skip to content

Commit f4e54c9

Browse files
markoubmarkoub
committed
fix(cron): preserve seeded store revision
1 parent 0017321 commit f4e54c9

6 files changed

Lines changed: 45 additions & 32 deletions

File tree

‎src/cron/service.issue-66019-unresolved-next-run.test.ts‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -153,7 +153,8 @@ describe("#66019 unresolved next-run repro", () => {
153153
error: "synthetic failure",
154154
});
155155
const naturalNext = scheduledAt + 5_000;
156-
const backoffNext = scheduledAt + 30_000;
156+
const claimedAt = scheduledAt + 1;
157+
const backoffNext = claimedAt + 30_000;
157158
const nextRunSpy = vi
158159
.spyOn(schedule, "computeNextRunAtMs")
159160
.mockReturnValueOnce(undefined)

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

Lines changed: 5 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -152,6 +152,7 @@ describe("CronService restart catch-up", () => {
152152

153153
it("executes an overdue recurring job immediately on start", async () => {
154154
const dueAt = Date.parse("2025-12-13T15:00:00.000Z");
155+
const claimedAt = Date.parse("2025-12-13T17:00:00.001Z");
155156
const lastRunAt = Date.parse("2025-12-12T15:00:00.000Z");
156157

157158
await withRestartedCron(
@@ -180,8 +181,8 @@ describe("CronService restart catch-up", () => {
180181
const listedJobs = await cron.list({ includeDisabled: true });
181182
const updated = listedJobs.find((job) => job.id === "restart-overdue-job");
182183
expect(updated?.state.lastStatus).toBe("ok");
183-
expect(updated?.state.lastRunAtMs).toBe(Date.parse("2025-12-13T17:00:00.000Z"));
184-
expect(updated?.state.nextRunAtMs).toBeGreaterThan(Date.parse("2025-12-13T17:00:00.000Z"));
184+
expect(updated?.state.lastRunAtMs).toBe(claimedAt);
185+
expect(updated?.state.nextRunAtMs).toBeGreaterThan(claimedAt);
185186
},
186187
);
187188
});
@@ -344,12 +345,13 @@ describe("CronService restart catch-up", () => {
344345
},
345346
],
346347
async ({ cron, enqueueSystemEvent, requestHeartbeat }) => {
348+
const claimedAt = Date.parse("2025-12-13T04:02:00.001Z");
347349
expectQueuedSystemEvent(enqueueSystemEvent, "catch missed slot");
348350
expect(requestHeartbeat).toHaveBeenCalled();
349351

350352
const listedJobs = await cron.list({ includeDisabled: true });
351353
const updated = listedJobs.find((job) => job.id === "restart-missed-slot");
352-
expect(updated?.state.lastRunAtMs).toBe(Date.parse("2025-12-13T04:02:00.000Z"));
354+
expect(updated?.state.lastRunAtMs).toBe(claimedAt);
353355
},
354356
);
355357
});

‎src/cron/service/ops.test.ts‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -593,11 +593,12 @@ describe("cron service ops seam coverage", () => {
593593
storePath,
594594
now,
595595
});
596+
const claimedAt = now + 1;
596597

597598
await runMissedJobs(state);
598599

599600
expectTaskRun({
600-
runId: `cron:startup-timeout:${now}`,
601+
runId: `cron:startup-timeout:${claimedAt}`,
601602
runtime: "cron",
602603
status: "timed_out",
603604
sourceId: "startup-timeout",

‎src/cron/service/store.ts‎

Lines changed: 13 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -202,12 +202,19 @@ export async function ensureLoaded(
202202
const storeRevision = getCronJobStoreRevision(state.deps.storePath);
203203
// Fast path: trust the in-memory copy unless this process or another SQLite
204204
// connection advanced the store revision.
205-
if (
206-
state.store &&
207-
!opts?.forceReload &&
208-
loadedStoreRevisionByState.get(state) === storeRevision
209-
) {
210-
return;
205+
if (state.store && !opts?.forceReload) {
206+
const loadedRevision = loadedStoreRevisionByState.get(state);
207+
if (loadedRevision === undefined) {
208+
loadedStoreRevisionByState.set(state, storeRevision);
209+
persistedDeliveryTargetSnapshotByState.set(
210+
state,
211+
snapshotPersistedDeliveryTargets(storeRevision, state.store),
212+
);
213+
return;
214+
}
215+
if (loadedRevision === storeRevision) {
216+
return;
217+
}
211218
}
212219
const previousJobsById = new Map<string, CronJob>();
213220
for (const job of state.store?.jobs ?? []) {

‎src/cron/service/timer.regression.test.ts‎

Lines changed: 11 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -887,7 +887,8 @@ describe("cron service timer regressions", () => {
887887
});
888888
await runnerStarted.promise;
889889

890-
const runId = `cron:no-timeout-cancel:${scheduledAt}`;
890+
const claimedAt = scheduledAt + 1;
891+
const runId = `cron:no-timeout-cancel:${claimedAt}`;
891892
const task = listTaskRecords().find(
892893
(entry) => entry.runtime === "cron" && entry.runId === runId,
893894
);
@@ -1072,7 +1073,7 @@ describe("cron service timer regressions", () => {
10721073

10731074
const timerPromise = onTimer(state);
10741075
const observedAbortSignal = await runnerStarted.promise;
1075-
const runId = `cron:cancel-before-timeout:${scheduledAt}`;
1076+
const runId = `cron:cancel-before-timeout:${scheduledAt + 1}`;
10761077
let timerSettled = false;
10771078
void timerPromise.then(() => {
10781079
timerSettled = true;
@@ -1150,7 +1151,7 @@ describe("cron service timer regressions", () => {
11501151
now += Math.ceil(FAST_TIMEOUT_SECONDS * 1_000) + 10;
11511152
await cleanupStarted.promise;
11521153

1153-
const runId = `cron:late-cancel-after-timeout:${scheduledAt}`;
1154+
const runId = `cron:late-cancel-after-timeout:${scheduledAt + 1}`;
11541155
const task = listTaskRecords().find(
11551156
(entry) => entry.runtime === "cron" && entry.runId === runId,
11561157
);
@@ -1687,7 +1688,7 @@ describe("cron service timer regressions", () => {
16871688
});
16881689

16891690
const timerPromise = onTimer(state);
1690-
const runId = `cron:main-session-cancel-boundary:${scheduledAt}`;
1691+
const runId = `cron:main-session-cancel-boundary:${scheduledAt + 1}`;
16911692
for (
16921693
let attempt = 0;
16931694
attempt < 10 && runHeartbeatOnce.mock.calls.length === 0;
@@ -1723,7 +1724,7 @@ describe("cron service timer regressions", () => {
17231724
await vi.advanceTimersByTimeAsync(0);
17241725
await timerPromise;
17251726

1726-
const expectedSessionKey = `agent:main:cron:main-session-cancel-boundary:run:${scheduledAt}`;
1727+
const expectedSessionKey = `agent:main:cron:main-session-cancel-boundary:run:${scheduledAt + 1}`;
17271728
expect(enqueueSystemEvent).toHaveBeenCalledWith(
17281729
"queued downstream work",
17291730
expect.objectContaining({
@@ -1888,7 +1889,7 @@ describe("cron service timer regressions", () => {
18881889
const persisted = await loadCronStore(store.storePath);
18891890
const persistedJob = persisted.jobs.find((job) => job.id === cronJob.id);
18901891
expect(persistedJob?.state.lastStatus).not.toBe("ok");
1891-
expect(persistedJob?.state.runningAtMs).toBe(scheduledAt);
1892+
expect(persistedJob?.state.runningAtMs).toBe(scheduledAt + 1);
18921893
});
18931894

18941895
it("persists the due-job start timestamp before creating the cron task run", async () => {
@@ -2303,11 +2304,11 @@ describe("cron service timer regressions", () => {
23032304
.filter((evt) => evt.action === "started")
23042305
.map((evt) => evt.runAtMs);
23052306

2306-
expect(firstDone?.state.lastRunAtMs).toBe(dueAt);
2307-
expect(firstDone?.state.lastDurationMs).toBe(50);
2307+
expect(firstDone?.state.lastRunAtMs).toBe(dueAt + 1);
2308+
expect(firstDone?.state.lastDurationMs).toBe(49);
23082309
expect(secondDone?.state.lastRunAtMs).toBe(dueAt + 50);
23092310
expect(secondDone?.state.lastDurationMs).toBe(20);
2310-
expect(startedAtEvents).toEqual([dueAt, dueAt + 50]);
2311+
expect(startedAtEvents).toEqual([dueAt + 1, dueAt + 50]);
23112312
});
23122313

23132314
it("honors cron maxConcurrentRuns for due jobs", async () => {
@@ -2730,7 +2731,7 @@ describe("cron service timer regressions", () => {
27302731
const persistedReplacementClaimedJob = persisted.jobs.find(
27312732
(entry) => entry.id === replacementClaimedJob.id,
27322733
);
2733-
expect(persistedJob?.state.runningAtMs).toBe(scheduledAt);
2734+
expect(persistedJob?.state.runningAtMs).toBe(scheduledAt + 1);
27342735
expect(persistedJob?.state.lastStatus).toBeUndefined();
27352736
expect(persistedUnstartedJob?.state.runningAtMs).toBeUndefined();
27362737
expect(persistedUnstartedJob?.state.lastStatus).toBeUndefined();

‎src/cron/service/timer.test.ts‎

Lines changed: 12 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -148,7 +148,8 @@ describe("cron service timer seam coverage", () => {
148148

149149
await onTimer(state);
150150

151-
const cronRunSessionKey = `agent:main:cron:main-heartbeat-job:run:${now}`;
151+
const claimedAt = now + 1;
152+
const cronRunSessionKey = `agent:main:cron:main-heartbeat-job:run:${claimedAt}`;
152153
expect(enqueueSystemEvent).toHaveBeenCalledWith("heartbeat seam tick", {
153154
agentId: undefined,
154155
sessionKey: cronRunSessionKey,
@@ -170,8 +171,8 @@ describe("cron service timer seam coverage", () => {
170171
}
171172
expect(job.state.lastStatus).toBe("ok");
172173
expect(job.state.runningAtMs).toBeUndefined();
173-
expect(job.state.nextRunAtMs).toBe(now + 60_000);
174-
const task = findTaskByRunId(`cron:main-heartbeat-job:${now}`);
174+
expect(job.state.nextRunAtMs).toBe(claimedAt + 60_000);
175+
const task = findTaskByRunId(`cron:main-heartbeat-job:${claimedAt}`);
175176
if (!task) {
176177
throw new Error("expected cron task ledger record");
177178
}
@@ -180,16 +181,16 @@ describe("cron service timer seam coverage", () => {
180181
expect(task.ownerKey).toBe("");
181182
expect(task.scopeKind).toBe("system");
182183
expect(task.childSessionKey).toBe(cronRunSessionKey);
183-
expect(task.runId).toBe(`cron:main-heartbeat-job:${now}`);
184+
expect(task.runId).toBe(`cron:main-heartbeat-job:${claimedAt}`);
184185
expect(task.label).toBe("main heartbeat job");
185186
expect(task.task).toBe("main heartbeat job");
186187
expect(task.status).toBe("succeeded");
187188
expect(task.deliveryStatus).toBe("not_applicable");
188189
expect(task.notifyPolicy).toBe("silent");
189-
expect(task.startedAt).toBe(now);
190-
expect(task.lastEventAt).toBe(now);
191-
expect(task.endedAt).toBe(now);
192-
expect(task?.cleanupAfter).toBe(now + 7 * 24 * 60 * 60_000);
190+
expect(task.startedAt).toBe(claimedAt);
191+
expect(task.lastEventAt).toBe(claimedAt);
192+
expect(task.endedAt).toBe(claimedAt);
193+
expect(task?.cleanupAfter).toBe(claimedAt + 7 * 24 * 60 * 60_000);
193194

194195
const delays = timeoutSpy.mock.calls
195196
.map(([, delay]) => delay)
@@ -264,7 +265,7 @@ describe("cron service timer seam coverage", () => {
264265
message: "run isolated cron",
265266
}),
266267
);
267-
const task = findTaskByRunId(`cron:isolated-agent-job:${now}`);
268+
const task = findTaskByRunId(`cron:isolated-agent-job:${now + 1}`);
268269
if (!task) {
269270
throw new Error("expected isolated cron task ledger record");
270271
}
@@ -306,7 +307,7 @@ describe("cron service timer seam coverage", () => {
306307
expect(runIsolatedAgentJob).toHaveBeenCalledTimes(1);
307308
});
308309

309-
const task = findTaskByRunId(`cron:isolated-agent-job:${now}`);
310+
const task = findTaskByRunId(`cron:isolated-agent-job:${now + 1}`);
310311
if (!task) {
311312
throw new Error("expected active cron task ledger record");
312313
}
@@ -352,7 +353,7 @@ describe("cron service timer seam coverage", () => {
352353
{ jobId: "main-heartbeat-job", error: ledgerError },
353354
"cron: failed to create task ledger record",
354355
);
355-
const cronRunSessionKey = `agent:main:cron:main-heartbeat-job:run:${now}`;
356+
const cronRunSessionKey = `agent:main:cron:main-heartbeat-job:run:${now + 1}`;
356357
expect(enqueueSystemEvent).toHaveBeenCalledWith("heartbeat seam tick", {
357358
agentId: undefined,
358359
sessionKey: cronRunSessionKey,

0 commit comments

Comments
 (0)