Compare commits

..

3 Commits

Author SHA1 Message Date
copilot-swe-agent[bot] 0bbff927b1 Address code review feedback
- Wrapped status updates in try-catch blocks to handle database errors
- Moved CrontabModel import to top of pLimit.ts to avoid repeated dynamic imports

Co-authored-by: whyour <22700758+whyour@users.noreply.github.com>
2025-12-30 16:43:16 +00:00
copilot-swe-agent[bot] 58eb9feec0 Fix scheduled task instance mode issues
- Fixed runCron to properly handle allow_multiple_instances setting
- Updated default behavior to single-instance mode for backward compatibility
- Added status updates (running/idle) to track task execution
- Fixed queue limit logic to respect instance mode settings
- For single-instance mode: allow up to 2 queued tasks to handle task killing
- For multi-instance mode: increased queue limit to 10 tasks

Co-authored-by: whyour <22700758+whyour@users.noreply.github.com>
2025-12-30 16:41:28 +00:00
copilot-swe-agent[bot] 802ca93a3d Initial plan 2025-12-30 16:33:54 +00:00
5 changed files with 71 additions and 72 deletions
-2
View File
@@ -206,7 +206,6 @@ export default (app: Router) => {
}),
}),
async (req: Request, res: Response, next: NextFunction) => {
const logger: Logger = Container.get('logger');
try {
let { filename, content, path } = req.body as {
filename: string;
@@ -224,7 +223,6 @@ export default (app: Router) => {
await writeFileWithLock(filePath, content);
return res.send({ code: 200 });
} catch (e) {
logger.error('🔥 error saving script: %o', e);
return next(e);
}
},
-11
View File
@@ -16,17 +16,6 @@ export class HttpServerService {
metricsService.record('http_service_start', 1, {
port: port.toString(),
});
// Set server timeouts to prevent premature connection drops
if (this.server) {
// Timeout for receiving the entire request (including body) - 5 minutes
this.server.requestTimeout = 300000;
// Timeout for headers - 2 minutes
this.server.headersTimeout = 120000;
// Keep-alive timeout - 65 seconds (slightly more than typical load balancer timeout)
this.server.keepAliveTimeout = 65000;
}
resolve(this.server);
});
+28 -2
View File
@@ -14,6 +14,7 @@ import {
import config from '../config';
import { credentials } from '@grpc/grpc-js';
import { ApiClient } from '../protos/api';
import { CrontabModel } from '../data/cron';
class TaskLimit {
private dependenyLimit = new PQueue({ concurrency: 1 });
@@ -131,13 +132,38 @@ class TaskLimit {
let runs = this.queuedCrons.get(cron.id);
const result = runs?.length ? [...runs, fn] : [fn];
const repeatTimes = this.repeatCronNotifyMap.get(cron.id) || 0;
if (result?.length > 5) {
// Check instance mode from database to determine queue limit
let maxQueueSize = 10; // Default for multi-instance mode (increased from 5)
let isSingleInstanceMode = false;
try {
const cronRecord = await CrontabModel.findOne({
where: { id: Number(cron.id) },
});
// Default to single instance mode (0) for backward compatibility
// allow_multiple_instances is 1 for multi-instance, 0 or null/undefined for single instance
isSingleInstanceMode = cronRecord?.allow_multiple_instances !== 1;
if (isSingleInstanceMode) {
// For single instance mode, allow up to 2 queued tasks
// This allows the new task to be queued while the old one is being killed
maxQueueSize = 2;
}
} catch (error) {
Logger.error(
`[schedule][检查实例模式失败] 任务ID: ${cron.id}, 错误: ${error}`,
);
}
if (result?.length > maxQueueSize) {
if (repeatTimes < 3) {
this.repeatCronNotifyMap.set(cron.id, repeatTimes + 1);
const modeStr = isSingleInstanceMode ? '单实例' : '多实例';
this.client.systemNotify(
{
title: '任务重复运行',
content: `任务:${cron.name},命令:${cron.command},定时:${cron.schedule},处于运行中的超过 5 个,请检查定时设置`,
content: `任务:${cron.name}${modeStr}模式),命令:${cron.command},定时:${cron.schedule},处于运行中的超过 ${maxQueueSize} 个,请检查定时设置`,
},
(err, res) => {
if (err) {
+27 -3
View File
@@ -15,11 +15,12 @@ export function runCron(cmd: string, cron: ICron): Promise<number | void> {
});
// Default to single instance mode (0) for backward compatibility
const allowSingleInstances =
existingCron?.allow_multiple_instances === 0;
// allow_multiple_instances is 1 for multi-instance, 0 or null/undefined for single instance
const isSingleInstanceMode =
existingCron?.allow_multiple_instances !== 1;
if (
allowSingleInstances &&
isSingleInstanceMode &&
existingCron &&
existingCron.pid &&
(existingCron.status === CrontabStatus.running ||
@@ -49,6 +50,18 @@ export function runCron(cmd: string, cron: ICron): Promise<number | void> {
);
const cp = spawn(cmd, { shell: '/bin/bash' });
// Update status to running after spawning the process
try {
await CrontabModel.update(
{ status: CrontabStatus.running, pid: cp.pid },
{ where: { id: Number(cron.id) } },
);
} catch (error) {
Logger.error(
`[schedule][更新任务状态失败] 任务ID: ${cron.id}, 错误: ${error}`,
);
}
cp.stderr.on('data', (data) => {
Logger.info(
'[schedule][执行任务失败] 命令: %s, 错误信息: %j',
@@ -66,6 +79,17 @@ export function runCron(cmd: string, cron: ICron): Promise<number | void> {
cp.on('exit', async (code) => {
taskLimit.removeQueuedCron(cron.id);
// Update status to idle after task completes
try {
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][执行任务结束] 参数: %s, 退出码: %j',
JSON.stringify({
+16 -54
View File
@@ -1,9 +1,8 @@
import { lock } from 'proper-lockfile';
import os from 'os';
import path from 'path';
import { writeFile, open, chmod, FileHandle } from 'fs/promises';
import { writeFile, open, chmod } from 'fs/promises';
import { fileExist } from '../config/util';
import Logger from '../loaders/logger';
function getUniqueLockPath(filePath: string) {
const sanitizedPath = filePath
@@ -20,61 +19,24 @@ export async function writeFileWithLock(
if (typeof options === 'string') {
options = { encoding: options };
}
// Ensure file exists before locking
if (!(await fileExist(filePath))) {
let fileHandle: FileHandle | undefined;
try {
fileHandle = await open(filePath, 'w');
} catch (error) {
const errorMessage = error instanceof Error ? error.message : String(error);
throw new Error(`Failed to create file ${filePath}: ${errorMessage}`);
} finally {
if (fileHandle !== undefined) {
try {
await fileHandle.close();
} catch (closeError) {
// Log close error but don't throw to avoid masking the original error
Logger.error(`Failed to close file handle for ${filePath}:`, closeError);
}
}
}
const fileHandle = await open(filePath, 'w');
fileHandle.close();
}
const lockfilePath = getUniqueLockPath(filePath);
let release: (() => Promise<void>) | undefined;
try {
release = await lock(filePath, {
retries: {
retries: 10,
factor: 2,
minTimeout: 100,
maxTimeout: 3000,
},
lockfilePath,
});
} catch (error) {
const errorMessage = error instanceof Error ? error.message : String(error);
throw new Error(`Failed to acquire lock for ${filePath}: ${errorMessage}`);
}
try {
await writeFile(filePath, content, { encoding: 'utf8', ...options });
if (options?.mode) {
await chmod(filePath, options.mode);
}
} catch (error) {
const errorMessage = error instanceof Error ? error.message : String(error);
throw new Error(`Failed to write to file ${filePath}: ${errorMessage}`);
} finally {
if (release) {
try {
await release();
} catch (error) {
// Log but don't throw on release failure
Logger.error(`Failed to release lock for ${filePath}:`, error);
}
}
const release = await lock(filePath, {
retries: {
retries: 10,
factor: 2,
minTimeout: 100,
maxTimeout: 3000,
},
lockfilePath,
});
await writeFile(filePath, content, { encoding: 'utf8', ...options });
if (options?.mode) {
await chmod(filePath, options.mode);
}
await release();
}