vrr/openclaw
2
0
Fork 0
mirror of https://github.com/openclaw/openclaw.git synced 2026-09-03 06:38:40 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 21

test(cron): settle deferred runs before fixture teardown (#136901)

Browse source 
Preallocate held provider completions and finish them in local fixture
cleanup. Stop scheduled work or abort trigger evaluation before releasing
its gate, retain the actual timer and manual-run promises, and join core
and tick settlement before clocks or stores are cleaned up.

Keep all 43 cases, normal completion values, success-path awaits, and
progress, read responsiveness, reschedule, ABA and cancellation assertions.
No production, shared-harness, proxy or routing change.

Validation: both original and final 43-case suites pass. Four matching
intentional-failure controls show the original unfinished work and the
candidate's completed cleanup before diagnostic rescue. Each control keeps
its intended failure with 42 other cases passing; final rescue is unused.

Final validation: normal two-path changed checks pass and independent
Codex review has no actionable P0-P2 findings. The final lint correction
omits one explicit default type argument; emitted JavaScript is identical
to the candidate used for the 43-case and four fault-control runs.
This commit is contained in:
Peter Steinberger 2026-09-02 20:31:09 -07:00 • committed by GitHub
parent 9478b3c816
commit c5bde6ac93
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
2 changed files with 76 additions and 43 deletions
Download patch file Download diff file Expand all files Collapse all files

42
src/cron/service.read-ops-nonblocking.test.ts
View file

@ -3,9 +3,12 @@ import fs from "node:fs/promises";
import os from "node:os";
import path from "node:path";
import { describe, expect, it, vi } from "vitest";
import { createDeferred } from "../../test/helpers/promise.js";
import { getActiveGatewayRootWorkCount } from "../process/gateway-work-admission.js";
import { withTimeout } from "../utils/with-timeout.js";
import { CronService } from "./service.js";
import { writeCronStoreSnapshot } from "./service.test-harness.js";
import { getSuspensionVisibleCronTaskRunCount } from "./service/active-run-cancellation.js";
import type { CronJob } from "./types.js";
const sqliteTransactionLabels = vi.hoisted(() => [] as string[]);
@ -58,22 +61,27 @@ async function makeStorePath() {
}
function createDeferredIsolatedRun() {
let resolveRun: ((value: IsolatedRunResult) => void) | undefined;
let resolveRunStarted: (() => void) | undefined;
const runStarted = new Promise<void>((resolve) => {
resolveRunStarted = resolve;
});
const result = createDeferred<IsolatedRunResult>();
const started = createDeferred();
const runIsolatedAgentJob = vi.fn(async () => {
resolveRunStarted?.();
return await new Promise<IsolatedRunResult>((resolve) => {
resolveRun = resolve;
});
started.resolve();
return await result.promise;
});
return {
runIsolatedAgentJob,
runStarted,
completeRun: (result: IsolatedRunResult) => {
resolveRun?.(result);
runStarted: started.promise,
completeRun: result.resolve,
settle: async (run?: Promise<unknown>) => {
// The caller stops scheduling first; storage must outlive the admitted core and tick.
result.resolve({ status: "ok", summary: "done" });
try {
await run;
} finally {
await vi.waitFor(() => {
expect(getSuspensionVisibleCronTaskRunCount()).toBe(0);
expect(getActiveGatewayRootWorkCount()).toBe(0);
});
}
},
};
}
@ -259,6 +267,7 @@ describe("CronService read ops while job is running", () => {
} finally {
cron.stop();
restartedCron?.stop();
await isolatedRun.settle();
vi.clearAllTimers();
vi.useRealTimers();
await store.cleanup();
@ -344,6 +353,7 @@ describe("CronService read ops while job is running", () => {
expect(internal.state?.running).toBe(false);
} finally {
cron.stop();
await isolatedRun.settle();
vi.clearAllTimers();
vi.useRealTimers();
await store.cleanup();
@ -366,6 +376,7 @@ describe("CronService read ops while job is running", () => {
runIsolatedAgentJob: isolatedRun.runIsolatedAgentJob,
});
let restartedCron: CronService | undefined;
let run: ReturnType<CronService["run"]> | undefined;
try {
await cron.start();
@ -380,7 +391,7 @@ describe("CronService read ops while job is running", () => {
delivery: { mode: "none" },
});
const run = cron.run(job.id, "force");
run = cron.run(job.id, "force");
await isolatedRun.runStarted;
await cron.update(job.id, {
schedule: { kind: "at", at: new Date(intermediateAt).toISOString() },
@ -418,6 +429,7 @@ describe("CronService read ops while job is running", () => {
} finally {
cron.stop();
restartedCron?.stop();
await isolatedRun.settle(run);
await store.cleanup();
}
},
@ -437,6 +449,7 @@ describe("CronService read ops while job is running", () => {
requestHeartbeat,
runIsolatedAgentJob: isolatedRun.runIsolatedAgentJob,
});
let runPromise: ReturnType<CronService["run"]> | undefined;
try {
await cron.start();
@ -454,7 +467,7 @@ describe("CronService read ops while job is running", () => {
delivery: { mode: "none" },
});
const runPromise = cron.run(job.id, "force");
runPromise = cron.run(job.id, "force");
await isolatedRun.runStarted;
await expect(
@ -479,6 +492,7 @@ describe("CronService read ops while job is running", () => {
expect(completed[0]?.state.runningAtMs).toBeUndefined();
} finally {
cron.stop();
await isolatedRun.settle(runPromise);
await store.cleanup();
}
});

77
src/cron/service/timer.test.ts
View file

@ -8,12 +8,15 @@ import { createCronServiceState as createCronServiceStateBase } from "../../cron
import { executeJobCore, onTimer } from "../../cron/service/timer.test-support.js";
import { loadCronStore } from "../../cron/store.js";
import type { CronJob } from "../../cron/types.js";
import { getActiveGatewayRootWorkCount } from "../../process/gateway-work-admission.js";
import { openOpenClawStateDatabase } from "../../state/openclaw-state-db.js";
import * as taskExecutor from "../../tasks/task-executor.js";
import { findTaskByRunId, listTaskRecordsUnsorted } from "../../tasks/task-registry.js";
import { resetTaskRegistryForTests } from "../../tasks/task-runtime.test-helpers.js";
import { formatTaskStatusDetail } from "../../tasks/task-status.js";
import { normalizeSessionDeliveryState } from "../../utils/delivery-context.shared.js";
import { getSuspensionVisibleCronTaskRunCount } from "./active-run-cancellation.js";
import { stop } from "./ops-lifecycle.js";
const { logger, makeStorePath } = setupCronServiceSuite({
prefix: "cron-service-timer-seam",
@ -365,17 +368,24 @@ describe("cron service timer seam coverage", () => {
const controller = new AbortController();
const result = executeJobCore(state, job, controller.signal);
expect(evaluateCronTrigger).toHaveBeenCalledOnce();
controller.abort(new Error("operator cancelled the scheduled run"));
evaluation.resolve({ kind: "evaluated", fire: true, state: { revision: 2 } });
try {
expect(evaluateCronTrigger).toHaveBeenCalledOnce();
controller.abort(new Error("operator cancelled the scheduled run"));
evaluation.resolve({ kind: "evaluated", fire: true, state: { revision: 2 } });
await expect(result).resolves.toMatchObject({ status: "error" });
expect(enqueueSystemEvent).not.toHaveBeenCalled();
expect(requestHeartbeat).not.toHaveBeenCalled();
expect(runCommandJob).not.toHaveBeenCalled();
expect(runScriptJob).not.toHaveBeenCalled();
expect(runSkillCollectionReview).not.toHaveBeenCalled();
expect(runIsolatedAgentJob).not.toHaveBeenCalled();
await expect(result).resolves.toMatchObject({ status: "error" });
expect(enqueueSystemEvent).not.toHaveBeenCalled();
expect(requestHeartbeat).not.toHaveBeenCalled();
expect(runCommandJob).not.toHaveBeenCalled();
expect(runScriptJob).not.toHaveBeenCalled();
expect(runSkillCollectionReview).not.toHaveBeenCalled();
expect(runIsolatedAgentJob).not.toHaveBeenCalled();
} finally {
// Abort before releasing evaluation so failed assertions cannot start payload work.
controller.abort(new Error("operator cancelled the scheduled run"));
evaluation.resolve({ kind: "evaluated", fire: true, state: { revision: 2 } });
await result;
}
},
);
@ -918,13 +928,8 @@ describe("cron service timer seam coverage", () => {
const now = Date.parse("2026-03-23T12:00:00.000Z");
const enqueueSystemEvent = vi.fn();
const requestHeartbeat = vi.fn();
let resolveRun: ((value: { status: "ok"; summary: string }) => void) | undefined;
const runIsolatedAgentJob = vi.fn(
() =>
new Promise<{ status: "ok"; summary: string }>((resolve) => {
resolveRun = resolve;
}),
);
const runResult = createDeferred<{ status: "ok"; summary: string }>();
const runIsolatedAgentJob = vi.fn(() => runResult.promise);
await writeCronStoreSnapshot({
storePath,
@ -942,20 +947,34 @@ describe("cron service timer seam coverage", () => {
});
const timerRun = onTimer(state);
await vi.waitFor(() => {
expect(runIsolatedAgentJob).toHaveBeenCalledTimes(1);
});
try {
await vi.waitFor(() => {
expect(runIsolatedAgentJob).toHaveBeenCalledTimes(1);
});
const task = findCronTaskByBaseRunId(`cron:isolated-agent-job:${now}`);
if (!task) {
throw new Error("expected active cron task ledger record");
const task = findCronTaskByBaseRunId(`cron:isolated-agent-job:${now}`);
if (!task) {
throw new Error("expected active cron task ledger record");
}
expect(task.status).toBe("running");
expect(task.progressSummary).toBe("Running automation.");
expect(formatTaskStatusDetail(task)).toBe("Running automation.");
runResult.resolve({ status: "ok", summary: "done" });
await timerRun;
} finally {
// Stop new ticks and settle this core before the shared hooks reset its state.
stop(state);
runResult.resolve({ status: "ok", summary: "done" });
try {
await timerRun;
} finally {
await vi.waitFor(() => {
expect(getSuspensionVisibleCronTaskRunCount()).toBe(0);
expect(getActiveGatewayRootWorkCount()).toBe(0);
});
}
}
expect(task.status).toBe("running");
expect(task.progressSummary).toBe("Running automation.");
expect(formatTaskStatusDetail(task)).toBe("Running automation.");
resolveRun?.({ status: "ok", summary: "done" });
await timerRun;
});
it("keeps scheduler progress when task ledger creation fails", async () => {