@@ -7,6 +7,7 @@ import {
77createRunningTaskRun,
88failTaskRunByRunId,
99} from "../../tasks/detached-task-runtime.js";
10+import { clearCronJobActive, markCronJobActive } from "../active-jobs.js";
1011import { createCronRunDiagnosticsFromError } from "../run-diagnostics.js";
1112import { createCronExecutionId } from "../run-id.js";
1213import type { CronJob, CronJobCreate, CronJobPatch } from "../types.js";
@@ -686,6 +687,7 @@ async function prepareManualRun(
686687 job,
687688startedAt: preflight.now,
688689});
690+markCronJobActive(job.id);
689691const executionJob = structuredClone(job);
690692return {
691693ok: true,
@@ -710,92 +712,96 @@ async function finishPreparedManualRun(
710712const taskRunId = prepared.taskRunId;
711713const runId = prepared.runId;
712714713-let coreResult: Awaited<ReturnType<typeof executeJobCoreWithTimeout>>;
714715try {
715-coreResult = await executeJobCoreWithTimeout(state, executionJob);
716-} catch (err) {
717-coreResult = { status: "error", error: normalizeCronRunErrorText(err) };
718-}
719-const endedAt = state.deps.nowMs();
720-tryFinishManualTaskRun(state, {
721- taskRunId,
722- coreResult,
723- endedAt,
724-});
725-726-await locked(state, async () => {
727-await ensureLoaded(state, { skipRecompute: true });
728-const job = state.store?.jobs.find((entry) => entry.id === jobId);
729-if (!job) {
730-return;
716+let coreResult: Awaited<ReturnType<typeof executeJobCoreWithTimeout>>;
717+try {
718+coreResult = await executeJobCoreWithTimeout(state, executionJob);
719+} catch (err) {
720+coreResult = { status: "error", error: normalizeCronRunErrorText(err) };
731721}
722+const endedAt = state.deps.nowMs();
723+tryFinishManualTaskRun(state, {
724+ taskRunId,
725+ coreResult,
726+ endedAt,
727+});
732728733-const shouldDelete = applyJobResult(
734-state,
735-job,
736-{
729+await locked(state, async () => {
730+await ensureLoaded(state, { skipRecompute: true });
731+const job = state.store?.jobs.find((entry) => entry.id === jobId);
732+if (!job) {
733+return;
734+}
735+736+const shouldDelete = applyJobResult(
737+state,
738+job,
739+{
740+status: coreResult.status,
741+error: coreResult.error,
742+diagnostics: coreResult.diagnostics,
743+delivered: coreResult.delivered,
744+ startedAt,
745+ endedAt,
746+},
747+{ preserveSchedule: mode === "force" },
748+);
749+750+emit(state, {
751+jobId: job.id,
752+action: "finished",
753+ job,
737754status: coreResult.status,
738755error: coreResult.error,
756+summary: coreResult.summary,
739757diagnostics: coreResult.diagnostics,
740758delivered: coreResult.delivered,
741- startedAt,
742- endedAt,
743-},
744-{ preserveSchedule: mode === "force" },
745-);
746-747-emit(state, {
748-jobId: job.id,
749-action: "finished",
750- job,
751-status: coreResult.status,
752-error: coreResult.error,
753-summary: coreResult.summary,
754-diagnostics: coreResult.diagnostics,
755-delivered: coreResult.delivered,
756-deliveryStatus: job.state.lastDeliveryStatus,
757-deliveryError: job.state.lastDeliveryError,
758-delivery: coreResult.delivery,
759-sessionId: coreResult.sessionId,
760-sessionKey: coreResult.sessionKey,
761- runId,
762-runAtMs: startedAt,
763-durationMs: job.state.lastDurationMs,
764-nextRunAtMs: job.state.nextRunAtMs,
765-model: coreResult.model,
766-provider: coreResult.provider,
767-usage: coreResult.usage,
768-});
759+deliveryStatus: job.state.lastDeliveryStatus,
760+deliveryError: job.state.lastDeliveryError,
761+delivery: coreResult.delivery,
762+sessionId: coreResult.sessionId,
763+sessionKey: coreResult.sessionKey,
764+ runId,
765+runAtMs: startedAt,
766+durationMs: job.state.lastDurationMs,
767+nextRunAtMs: job.state.nextRunAtMs,
768+model: coreResult.model,
769+provider: coreResult.provider,
770+usage: coreResult.usage,
771+});
769772770-if (shouldDelete && state.store) {
771-state.store.jobs = state.store.jobs.filter((entry) => entry.id !== job.id);
772-emit(state, { jobId: job.id, action: "removed", job });
773-}
773+ if (shouldDelete && state.store) {
774+ state.store.jobs = state.store.jobs.filter((entry) => entry.id !== job.id);
775+ emit(state, { jobId: job.id, action: "removed", job });
776+ }
774777775-// Manual runs should not advance other due jobs without executing them.
776-// Use maintenance-only recompute to repair missing values while
777-// preserving existing past-due nextRunAtMs entries for future timer ticks.
778-const postRunSnapshot = shouldDelete
779- ? null
780- : {
781-enabled: job.enabled,
782-updatedAtMs: job.updatedAtMs,
783-state: structuredClone(job.state),
784-};
785-const postRunRemoved = shouldDelete;
786-// Isolated Telegram send can persist target writeback directly to disk.
787-// Reload before final persist so manual `cron run` keeps those changes.
788-await ensureLoaded(state, { forceReload: true, skipRecompute: true });
789-mergeManualRunSnapshotAfterReload({
790- state,
791- jobId,
792-snapshot: postRunSnapshot,
793-removed: postRunRemoved,
778+// Manual runs should not advance other due jobs without executing them.
779+// Use maintenance-only recompute to repair missing values while
780+// preserving existing past-due nextRunAtMs entries for future timer ticks.
781+const postRunSnapshot = shouldDelete
782+ ? null
783+ : {
784+enabled: job.enabled,
785+updatedAtMs: job.updatedAtMs,
786+state: structuredClone(job.state),
787+};
788+const postRunRemoved = shouldDelete;
789+// Isolated Telegram send can persist target writeback directly to disk.
790+// Reload before final persist so manual `cron run` keeps those changes.
791+await ensureLoaded(state, { forceReload: true, skipRecompute: true });
792+mergeManualRunSnapshotAfterReload({
793+ state,
794+ jobId,
795+snapshot: postRunSnapshot,
796+removed: postRunRemoved,
797+});
798+recomputeNextRunsForMaintenance(state, { recomputeExpired: true });
799+await persist(state);
800+armTimer(state);
794801});
795-recomputeNextRunsForMaintenance(state, { recomputeExpired: true });
796-await persist(state);
797-armTimer(state);
798-});
802+} finally {
803+clearCronJobActive(jobId);
804+}
799805}
800806801807export async function run(
此内容由惯性聚合(RSS阅读器)自动聚合整理，仅供阅读参考。 原文来自 — 版权归原作者所有。