diff --git a/back/shared/runCron.ts b/back/shared/runCron.ts index e5b9606e..af07debe 100644 --- a/back/shared/runCron.ts +++ b/back/shared/runCron.ts @@ -4,7 +4,6 @@ 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 { observeChildProcess, asError } from './childProcess'; @@ -17,7 +16,7 @@ export function runCron( const execute = async () => { try { if (!isCurrent()) return; - // Check if the cron is already running and stop it (only if multiple instances are not allowed) + // A single-instance trigger must not interrupt work already in progress. try { const existingCron = await CrontabModel.findOne({ where: { id: Number(cron.id) }, @@ -30,35 +29,93 @@ export function runCron( const allowSingleInstances = existingCron?.allow_multiple_instances === 0; - if ( - allowSingleInstances && - existingCron && - existingCron.pid && - (existingCron.status === CrontabStatus.running || - existingCron.status === CrontabStatus.queued) - ) { - Logger.info( - t('[schedule][停止已运行任务] 任务ID: %s, PID: %s'), - cron.id, - existingCron.pid, - ); - await killTask(existingCron.pid, true); - // Mark old running instances as stopped - const stoppedAt = dayjs().unix(); - await RunningInstanceModel.update( - { status: InstanceStatus.stopped, finished_at: stoppedAt }, - { + if (allowSingleInstances) { + // A queued manual run is an active reservation, even if its stored PID + // belongs to a previous execution. + if ( + existingCron.status === CrontabStatus.queued && + existingCron.pid + ) { + Logger.info( + '[schedule][Skip queued single-instance task] Task ID: %s, PID: %s', + cron.id, + existingCron.pid, + ); + return; + } + if ( + existingCron.status === CrontabStatus.running && + existingCron.pid + ) { + let missing = false; + try { + process.kill(existingCron.pid, 0); + } catch (error) { + const code = (error as NodeJS.ErrnoException).code; + if (code === 'ESRCH') missing = true; + else if (code !== 'EPERM') throw error; + } + if (!missing) { + Logger.info( + '[schedule][Skip active single-instance task] Task ID: %s, PID: %s', + cron.id, + existingCron.pid, + ); + return; + } + + // Capture old instance IDs before releasing the stale cron state so + // a concurrent replacement cannot have its instance marked stopped. + const staleInstances = await RunningInstanceModel.findAll({ + attributes: ['id'], where: { cron_id: Number(cron.id), + pid: existingCron.pid, status: InstanceStatus.running, }, - }, - ); - // Update the status to idle after killing - await CrontabModel.update( - { status: CrontabStatus.idle, pid: undefined }, - { where: { id: Number(cron.id) } }, - ); + }); + if (!isCurrent()) return; + const [recovered] = await CrontabModel.update( + { + status: CrontabStatus.idle, + pid: null, + queued_token: null, + } as any, + { + where: { + id: Number(cron.id), + pid: existingCron.pid, + status: CrontabStatus.running, + ...(existingCron.log_path !== undefined + ? { log_path: existingCron.log_path } + : {}), + ...(existingCron.queued_token !== undefined + ? { queued_token: existingCron.queued_token } + : {}), + ...(existingCron.last_execution_time !== undefined + ? { last_execution_time: existingCron.last_execution_time } + : {}), + ...(existingCron.isDisabled !== undefined + ? { isDisabled: existingCron.isDisabled } + : {}), + allow_multiple_instances: 0, + }, + }, + ); + if (recovered !== 1) return; + const staleIds = staleInstances.map((instance) => instance.id!); + if (staleIds.length) { + await RunningInstanceModel.update( + { status: InstanceStatus.stopped, finished_at: dayjs().unix() }, + { where: { id: staleIds, status: InstanceStatus.running } }, + ); + } + Logger.info( + '[schedule][Recovered stale single-instance task] Task ID: %s, PID: %s', + cron.id, + existingCron.pid, + ); + } } } catch (error) { Logger.error( @@ -69,7 +126,7 @@ export function runCron( throw error; } - // In particular, invalidation can happen while stopping a previous run. + // Invalidation can happen during the asynchronous database check. if (!isCurrent()) return; Logger.info( t('[schedule][开始执行任务] 参数 %s'), diff --git a/test/back/single-instance-skip.test.cjs b/test/back/single-instance-skip.test.cjs new file mode 100644 index 00000000..c9ad3e70 --- /dev/null +++ b/test/back/single-instance-skip.test.cjs @@ -0,0 +1,192 @@ +const test = require('node:test'); +const assert = require('node:assert/strict'); +const { spawnSync } = require('node:child_process'); +const load = require('../helpers/load-security-module.cjs'); + +function fixture(row, { claim = true, onSpawn = () => {} } = {}) { + const spawned = [], + cronWrites = [], + instanceWrites = [], + instanceReads = []; + const { runCron } = load('back/shared/runCron.ts', { + './i18n': { t: (s) => s }, + 'cross-spawn': { + spawn: (cmd) => { + spawned.push(cmd); + onSpawn(row); + return { pid: process.pid }; + }, + }, + './pLimit': { + runWithCronLimit: (_cron, execute) => execute(), + removeQueuedCron() {}, + }, + '../loaders/logger': { info() {}, error() {} }, + '../data/cron': { + CrontabModel: { + findOne: async () => row, + update: async (...args) => { + cronWrites.push(args); + if (!claim) return [0]; + Object.assign(row, args[0]); + return [1]; + }, + }, + CrontabStatus: { running: 0, idle: 1, queued: 3 }, + }, + '../data/runningInstance': { + RunningInstanceModel: { + findAll: async (options) => { + instanceReads.push(options); + return [{ id: 7 }]; + }, + update: async (...args) => instanceWrites.push(args), + }, + InstanceStatus: { running: 0, stopped: 2 }, + }, + dayjs: () => ({ unix: () => 123 }), + './childProcess': { + observeChildProcess: () => ({ completed: Promise.resolve({ code: 0 }) }), + asError: (e) => e, + }, + }); + return { + run: () => runCron('task example.py', { id: '1' }), + spawned, + cronWrites, + instanceWrites, + instanceReads, + }; +} + +function exitedPid() { + const child = spawnSync(process.execPath, ['-e', '']); + assert.equal(child.status, 0); + assert.throws(() => process.kill(child.pid, 0), { code: 'ESRCH' }); + return child.pid; +} + +test('running single-instance task stays untouched while its process is alive', async () => { + const f = fixture({ + allow_multiple_instances: 0, + pid: process.pid, + status: 0, + }); + await f.run(); + assert.deepEqual(f.spawned, []); + assert.deepEqual(f.cronWrites, []); + assert.deepEqual(f.instanceWrites, []); +}); + +test('queued single-instance task keeps its reservation even with an exited PID', async () => { + const f = fixture({ + allow_multiple_instances: 0, + pid: exitedPid(), + status: 3, + }); + await f.run(); + assert.deepEqual(f.spawned, []); + assert.deepEqual(f.cronWrites, []); + assert.deepEqual(f.instanceWrites, []); +}); + +test('exited running task is recovered and the next trigger sees the new active run', async () => { + const oldPid = exitedPid(); + const f = fixture( + { + allow_multiple_instances: 0, + pid: oldPid, + status: 0, + log_path: 'old.log', + queued_token: 'old-token', + last_execution_time: 10, + isDisabled: 0, + }, + { + onSpawn: (row) => { + row.status = 0; + row.pid = process.pid; + }, + }, + ); + await f.run(); + await f.run(); + assert.equal(f.spawned.length, 1); + assert.deepEqual(f.instanceReads, [ + { + attributes: ['id'], + where: { cron_id: 1, pid: oldPid, status: 0 }, + }, + ]); + assert.deepEqual(f.cronWrites, [ + [ + { status: 1, pid: null, queued_token: null }, + { + where: { + id: 1, + pid: oldPid, + status: 0, + log_path: 'old.log', + queued_token: 'old-token', + last_execution_time: 10, + isDisabled: 0, + allow_multiple_instances: 0, + }, + }, + ], + ]); + assert.deepEqual(f.instanceWrites, [ + [{ status: 2, finished_at: 123 }, { where: { id: [7], status: 0 } }], + ]); +}); + +test('a lost stale-state claim does not spawn or change running instances', async () => { + const f = fixture( + { allow_multiple_instances: 0, pid: exitedPid(), status: 0 }, + { claim: false }, + ); + await f.run(); + assert.equal(f.cronWrites.length, 1); + assert.deepEqual(f.spawned, []); + assert.deepEqual(f.instanceWrites, []); +}); + +test('EPERM means an existing process must still block a new run', async () => { + const pid = exitedPid(); + const originalKill = process.kill; + process.kill = (target, signal) => { + if (target === pid && signal === 0) + throw Object.assign(new Error('denied'), { code: 'EPERM' }); + return originalKill(target, signal); + }; + try { + const f = fixture({ allow_multiple_instances: 0, pid, status: 0 }); + await f.run(); + assert.deepEqual(f.spawned, []); + assert.deepEqual(f.cronWrites, []); + } finally { + process.kill = originalKill; + } +}); + +test('single-instance still starts idle tasks with a stale stored PID', async () => { + const f = fixture({ + allow_multiple_instances: 0, + pid: exitedPid(), + status: 1, + }); + await f.run(); + assert.equal(f.spawned.length, 1); + assert.deepEqual(f.cronWrites, []); +}); + +test('multi-instance keeps its existing concurrent execution behavior', async () => { + const f = fixture({ + allow_multiple_instances: 1, + pid: process.pid, + status: 0, + }); + await f.run(); + assert.equal(f.spawned.length, 1); + assert.deepEqual(f.cronWrites, []); +}); diff --git a/test/back/stop-failure-feedback.test.cjs b/test/back/stop-failure-feedback.test.cjs index 5b4290af..8f1ecc8e 100644 --- a/test/back/stop-failure-feedback.test.cjs +++ b/test/back/stop-failure-feedback.test.cjs @@ -4,75 +4,6 @@ const load = require('../helpers/load-security-module.cjs'); const logger = { info() {}, error() {} }; -function cronFixture(killTask) { - const events = []; - const { runCron } = load('back/shared/runCron.ts', { - 'cross-spawn': { - spawn: () => { - events.push('spawn'); - return {}; - }, - }, - './childProcess': { - observeChildProcess: () => ({ completed: Promise.resolve({ code: 0 }) }), - asError: (e) => e, - }, - './pLimit': { - runWithCronLimit: async (_, fn) => fn(), - removeQueuedCron: () => events.push('release'), - }, - '../loaders/logger': logger, - '../config/util': { killTask }, - '../data/cron': { - CrontabModel: { - findOne: async () => ({ - pid: 100, - status: 0, - allow_multiple_instances: 0, - }), - update: async () => events.push('idle'), - }, - CrontabStatus: { running: 0, idle: 1, queued: 3 }, - }, - '../data/runningInstance': { - RunningInstanceModel: { update: async () => events.push('stopped') }, - InstanceStatus: { running: 0, stopped: 2 }, - }, - }); - return { run: () => runCron('ignored', { id: '1' }), events }; -} - -test('single-instance replacement does not spawn or finalize when termination fails', async () => { - const f = cronFixture(async () => { - throw Error('EPERM'); - }); - await f.run(); - assert.deepEqual(f.events, ['release']); -}); - -test('single-instance replacement waits for verified exit before spawning', async () => { - let finish, entered, verified; - const pending = new Promise((resolve) => { - entered = resolve; - }); - const gate = new Promise((resolve) => { - finish = resolve; - }); - const f = cronFixture(async (pid, wait) => { - assert.equal(pid, 100); - verified = wait; - entered(); - await gate; - }); - const running = f.run(); - await pending; - assert.deepEqual(f.events, []); - finish(); - await running; - assert.equal(verified, true); - assert.deepEqual(f.events, ['stopped', 'idle', 'spawn', 'release']); -}); - function subscriptionFixture(killTask) { const updates = []; const Service = load('back/services/subscription.ts', {