qinglong/back/shared/runCron.ts

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<number | void> {
  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 });
      });
    });
  });
}