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
4 changed files with 79 additions and 175 deletions
-160
View File
@@ -10,7 +10,6 @@ import Logger from '../loaders/logger';
import { writeFileWithLock } from '../shared/utils';
import { DependenceTypes } from '../data/dependence';
import { FormData } from 'undici';
import os from 'os';
export * from './share';
@@ -591,162 +590,3 @@ export function getUninstallCommand(
export function isDemoEnv() {
return process.env.DeployEnv === 'demo';
}
// OS detection for Linux mirror configuration
let osType: 'Debian' | 'Ubuntu' | 'Alpine' | undefined;
async function getOSReleaseInfo(): Promise<string> {
try {
const osRelease = await fs.readFile('/etc/os-release', 'utf8');
return osRelease;
} catch (error) {
Logger.error(`Failed to read /etc/os-release: ${error}`);
return '';
}
}
function isDebian(osReleaseInfo: string): boolean {
return osReleaseInfo.includes('Debian');
}
function isUbuntu(osReleaseInfo: string): boolean {
return osReleaseInfo.includes('Ubuntu');
}
function isAlpine(osReleaseInfo: string): boolean {
return osReleaseInfo.includes('Alpine');
}
export async function detectOS(): Promise<
'Debian' | 'Ubuntu' | 'Alpine' | undefined
> {
if (osType) return osType;
const platform = os.platform();
if (platform === 'linux') {
const osReleaseInfo = await getOSReleaseInfo();
// Check Ubuntu before Debian since Ubuntu is based on Debian
if (isUbuntu(osReleaseInfo)) {
osType = 'Ubuntu';
} else if (isDebian(osReleaseInfo)) {
osType = 'Debian';
} else if (isAlpine(osReleaseInfo)) {
osType = 'Alpine';
} else {
Logger.error(`Unknown Linux Distribution: ${osReleaseInfo}`);
console.error(`Unknown Linux Distribution: ${osReleaseInfo}`);
}
} else if (platform === 'darwin') {
osType = undefined;
} else {
Logger.error(`Unsupported platform: ${platform}`);
console.error(`Unsupported platform: ${platform}`);
}
return osType;
}
async function getCurrentMirrorDomain(
filePath: string,
): Promise<string | null> {
try {
const fileContent = await fs.readFile(filePath, 'utf8');
const lines = fileContent.split('\n');
for (const line of lines) {
if (line.trim().startsWith('#')) {
continue;
}
const match = line.match(/https?:\/\/[^\/]+/);
if (match) {
return match[0];
}
}
return null;
} catch (error) {
Logger.error(`Failed to read mirror configuration file ${filePath}: ${error}`);
return null;
}
}
function escapeRegExp(string: string): string {
return string.replace(/[.*+?^${}()|[\]\\]/g, '\\$&');
}
async function replaceDomainInFile(
filePath: string,
oldDomainWithScheme: string,
newDomainWithScheme: string,
): Promise<void> {
// Ensure the new domain has a trailing slash before replacement
if (!newDomainWithScheme.endsWith('/')) {
newDomainWithScheme += '/';
}
let fileContent = await fs.readFile(filePath, 'utf8');
// Escape special regex characters in the old domain
const escapedOldDomain = escapeRegExp(oldDomainWithScheme);
let updatedContent = fileContent.replace(
new RegExp(escapedOldDomain, 'g'),
newDomainWithScheme,
);
await writeFileWithLock(filePath, updatedContent);
}
async function _updateLinuxMirror(
osType: string,
mirrorDomainWithScheme: string,
): Promise<string> {
let filePath: string, currentDomainWithScheme: string | null;
switch (osType) {
case 'Debian':
filePath = '/etc/apt/sources.list.d/debian.sources';
currentDomainWithScheme = await getCurrentMirrorDomain(filePath);
if (currentDomainWithScheme) {
await replaceDomainInFile(
filePath,
currentDomainWithScheme,
mirrorDomainWithScheme || 'http://deb.debian.org',
);
return 'apt-get update';
} else {
throw Error(`Current mirror domain not found.`);
}
case 'Ubuntu':
filePath = '/etc/apt/sources.list.d/ubuntu.sources';
currentDomainWithScheme = await getCurrentMirrorDomain(filePath);
if (currentDomainWithScheme) {
await replaceDomainInFile(
filePath,
currentDomainWithScheme,
mirrorDomainWithScheme || 'http://archive.ubuntu.com',
);
return 'apt-get update';
} else {
throw Error(`Current mirror domain not found.`);
}
case 'Alpine':
filePath = '/etc/apk/repositories';
currentDomainWithScheme = await getCurrentMirrorDomain(filePath);
if (currentDomainWithScheme) {
await replaceDomainInFile(
filePath,
currentDomainWithScheme,
mirrorDomainWithScheme || 'http://dl-cdn.alpinelinux.org',
);
return 'apk update';
} else {
throw Error(`Current mirror domain not found.`);
}
default:
throw Error('Unsupported OS type for updating mirrors.');
}
}
export async function updateLinuxMirrorFile(mirror: string): Promise<string> {
const detectedOS = await detectOS();
if (!detectedOS) {
throw Error(`Unknown Linux Distribution`);
}
return await _updateLinuxMirror(detectedOS, mirror);
}
+24 -10
View File
@@ -17,7 +17,6 @@ import {
readDirs,
rmPath,
setSystemTimezone,
updateLinuxMirrorFile,
} from '../config/util';
import {
DependenceModel,
@@ -215,11 +214,33 @@ export default class SystemService {
onEnd?: () => void,
) {
const oDoc = await this.getSystemConfig();
await this.updateAuthDb({
...oDoc,
info: { ...oDoc.info, ...info },
});
let defaultDomain = 'https://dl-cdn.alpinelinux.org';
let targetDomain = 'https://dl-cdn.alpinelinux.org';
if (os.platform() !== 'linux') {
return;
}
const command = await updateLinuxMirrorFile(info.linuxMirror || '');
let hasError = false;
const content = await fs.promises.readFile('/etc/apk/repositories', {
encoding: 'utf-8',
});
const domainMatch = content.match(/(http.*)\/alpine\/.*/);
if (domainMatch) {
defaultDomain = domainMatch[1];
}
if (info.linuxMirror) {
targetDomain = info.linuxMirror;
}
const command = `sed -i 's/${defaultDomain.replace(
/\//g,
'\\/',
)}/${targetDomain.replace(
/\//g,
'\\/',
)}/g' /etc/apk/repositories && apk update -f`;
this.scheduleService.runTask(
command,
{
@@ -233,15 +254,8 @@ export default class SystemService {
message: 'update linux mirror end',
});
onEnd?.();
if (!hasError) {
await this.updateAuthDb({
...oDoc,
info: { ...oDoc.info, ...info },
});
}
},
onError: async (message: string) => {
hasError = true;
this.sockService.sendMessage({ type: 'updateLinuxMirror', message });
},
onLog: async (message: string) => {
+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({