import { spawn } from 'cross-spawn'; import taskLimit from './pLimit'; import Logger from '../loaders/logger'; import { ICron } from '../protos/cron'; import { CrontabModel, CrontabStatus } from '../data/cron'; import { killTask } from '../config/util'; import { RunningInstanceModel, InstanceStatus, } from '../data/runningInstance'; import dayjs from 'dayjs'; import { observeLegacyCancellation, observeLegacyExecution, } from '../runtime/compatibility/legacyExecutionBridge'; import { observeLegacyChildProcess } from '../runtime/compatibility/observeLegacyChildProcess'; import { createLegacyTaskRevision } from '../runtime/compatibility/legacyTaskRevision'; export function runCron(cmd: string, cron: ICron): Promise { return taskLimit.runWithCronLimit(cron, () => { return new Promise(async (resolve: any) => { // Check if the cron is already running and stop it (only if multiple instances are not allowed) try { const existingCron = await CrontabModel.findOne({ where: { id: Number(cron.id) }, }); // Default to single instance mode (0) for backward compatibility const allowSingleInstances = existingCron?.allow_multiple_instances === 0; if ( allowSingleInstances && existingCron && existingCron.pid && (existingCron.status === CrontabStatus.running || existingCron.status === CrontabStatus.queued) ) { observeLegacyCancellation({ legacyCronId: Number(cron.id), pid: existingCron.pid, ...(existingCron.log_path ? { logPath: existingCron.log_path } : {}), atMs: Date.now(), scope: 'all', reason: 'policy', }); Logger.info( `[schedule][停止已运行任务] 任务ID: ${cron.id}, PID: ${existingCron.pid}`, ); await killTask(existingCron.pid); // Mark old running instances as stopped const stoppedAt = dayjs().unix(); await RunningInstanceModel.update( { status: InstanceStatus.stopped, finished_at: stoppedAt }, { where: { cron_id: Number(cron.id), status: InstanceStatus.running } }, ); // Update the status to idle after killing await CrontabModel.update( { status: CrontabStatus.idle, pid: undefined }, { where: { id: Number(cron.id) } }, ); } } catch (error) { Logger.error( `[schedule][检查已运行任务失败] 任务ID: ${cron.id}, 错误: ${error}`, ); } Logger.info( `[schedule][开始执行任务] 参数 ${JSON.stringify({ ...cron, command: cmd, })}`, ); const legacyCronId = Number(cron.id); const observation = observeLegacyExecution('scheduled_node', () => ({ origin: 'scheduled_node', projectId: 'default', taskId: `legacy-cron:${cron.id}`, taskRevision: createLegacyTaskRevision({ command: cmd, schedule: cron.schedule, extraSchedules: cron.extra_schedules.map((item) => item.schedule), }), taskName: cron.name, ...(Number.isSafeInteger(legacyCronId) && legacyCronId > 0 ? { legacyCronId } : {}), triggerType: 'scheduled', triggeredBy: 'legacy:scheduler', acceptedAtMs: Date.now(), })); const cp = spawn(cmd, { shell: '/bin/bash' }); if (observation) observeLegacyChildProcess(cp, observation); cp.stderr.on('data', (data) => { Logger.info( '[schedule][执行任务失败] 命令: %s, 错误信息: %j', cmd, data.toString(), ); }); cp.on('error', (err) => { Logger.error( '[schedule][创建任务失败] 命令: %s, 错误信息: %j', cmd, err, ); }); cp.on('exit', async (code) => { taskLimit.removeQueuedCron(cron.id); Logger.info( '[schedule][执行任务结束] 参数: %s, 退出码: %j', JSON.stringify({ ...cron, command: cmd, }), code, ); resolve({ ...cron, command: cmd, pid: cp.pid, code }); }); }); }); }