mirror of
https://github.com/whyour/qinglong.git
synced 2026-11-02 06:58:52 +08:00
fix: skip new scheduled triggers while a single-instance task is active (#3101)
* fix: skip active single-instance scheduled executions * fix: recover stale single-instance scheduled tasks * test: remove obsolete single-instance replacement assertions --------- Co-authored-by: Patrick Gu <username@users.noreply.github.com> Co-authored-by: whyour <imwhyour@gmail.com>
This commit is contained in:
Patrick Gu
co-authored by
Patrick Gu
whyour
Patrick Gu
whyour
parent
d895fbba36
commit
da9c6a6bbb
+76
-19
@@ -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,36 +29,94 @@ export function runCron(
|
||||
const allowSingleInstances =
|
||||
existingCron?.allow_multiple_instances === 0;
|
||||
|
||||
if (allowSingleInstances) {
|
||||
// A queued manual run is an active reservation, even if its stored PID
|
||||
// belongs to a previous execution.
|
||||
if (
|
||||
allowSingleInstances &&
|
||||
existingCron &&
|
||||
existingCron.pid &&
|
||||
(existingCron.status === CrontabStatus.running ||
|
||||
existingCron.status === CrontabStatus.queued)
|
||||
existingCron.status === CrontabStatus.queued &&
|
||||
existingCron.pid
|
||||
) {
|
||||
Logger.info(
|
||||
t('[schedule][停止已运行任务] 任务ID: %s, PID: %s'),
|
||||
'[schedule][Skip queued single-instance task] Task 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 },
|
||||
{
|
||||
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,
|
||||
},
|
||||
});
|
||||
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,
|
||||
},
|
||||
},
|
||||
);
|
||||
// Update the status to idle after killing
|
||||
await CrontabModel.update(
|
||||
{ status: CrontabStatus.idle, pid: undefined },
|
||||
{ where: { id: Number(cron.id) } },
|
||||
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(
|
||||
t('[schedule][检查已运行任务失败] 任务ID: %s, 错误: %s'),
|
||||
@@ -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'),
|
||||
|
||||
@@ -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, []);
|
||||
});
|
||||
@@ -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', {
|
||||
|
||||
Reference in New Issue
Block a user