open-source/qinglong
1
0
Fork 0
mirror of https://github.com/whyour/qinglong.git synced 2026-07-17 09:34:31 +08:00
Code Issues Packages Projects Releases Wiki Activity

修复添加任务有可能失败

Browse Source
This commit is contained in:
whyour
2026-07-01 00:37:42 +08:00
parent a0d57c78c5
commit e92ddb0aed
12 changed files with 194 additions and 58 deletions
Download Patch File Download Diff File Expand all files Collapse all files
back/app.ts
+19 -19
View File
@@ -60,11 +60,11 @@ class Application {
// Wait for gRPC worker to signal it's ready before starting HTTP worker
this.waitForWorkerReady(grpcWorker, 30000)
.then(() => {
Logger.info('✌️ gRPC worker is ready, starting HTTP worker');
Logger.info('[boot] gRPC worker is ready, starting HTTP worker');
this.httpWorker = this.forkWorker('http');
})
.catch((error) => {
Logger.error('✌️ Failed to wait for gRPC worker:', error);
Logger.error('[boot] Failed to wait for gRPC worker:', error);
process.exit(1);
});
@@ -73,7 +73,7 @@ class Application {
if (metadata) {
if (!this.isShuttingDown) {
Logger.error(
`✌️ ${metadata.serviceType} worker ${worker.process.pid} died (${signal || code
`${metadata.serviceType} worker ${worker.process.pid} died (${signal || code
}). Restarting...`,
);
// If gRPC worker died, restart it and wait for it to be ready
@@ -81,26 +81,26 @@ class Application {
const newGrpcWorker = this.forkWorker('grpc');
this.waitForWorkerReady(newGrpcWorker, 30000)
.then(() => {
Logger.info('✌️ gRPC worker restarted and ready');
Logger.info('gRPC worker restarted and ready');
// Re-register cron jobs by notifying the HTTP worker
if (this.httpWorker) {
try {
this.httpWorker.send('reregister-crons');
Logger.info('✌️ Sent reregister-crons message to HTTP worker');
Logger.info('Sent reregister-crons message to HTTP worker');
} catch (error) {
Logger.error('✌️ Failed to send reregister-crons message:', error);
Logger.error('Failed to send reregister-crons message:', error);
}
}
})
.catch((error) => {
Logger.error('✌️ Failed to restart gRPC worker:', error);
Logger.error('Failed to restart gRPC worker:', error);
process.exit(1);
});
} else {
// For HTTP worker, just restart it
const newWorker = this.forkWorker(metadata.serviceType);
this.httpWorker = newWorker;
Logger.info(`✌️ Restarted ${metadata.serviceType} worker (PID: ${newWorker.process.pid})`);
Logger.info(`Restarted ${metadata.serviceType} worker (PID: ${newWorker.process.pid})`);
}
}
@@ -169,7 +169,7 @@ class Application {
if (worker) {
const exitPromise = new Promise<void>((resolve) => {
worker.once('exit', () => {
Logger.info(`✌️ Worker ${worker.process.pid} exited`);
Logger.info(`Worker ${worker.process.pid} exited`);
resolve();
});
@@ -177,7 +177,7 @@ class Application {
worker.send('shutdown');
} catch (error) {
Logger.warn(
`✌️ Failed to send shutdown to worker ${worker.process.pid}:`,
`Failed to send shutdown to worker ${worker.process.pid}:`,
error,
);
}
@@ -192,14 +192,14 @@ class Application {
Promise.all(workerPromises),
new Promise<void>((resolve) => {
setTimeout(() => {
Logger.warn('✌️ Worker shutdown timeout reached');
Logger.warn('Worker shutdown timeout reached');
resolve();
}, 10000);
}),
]);
process.exit(0);
} catch (error) {
Logger.error('✌️ Error during worker shutdown:', error);
Logger.error('Error during worker shutdown:', error);
process.exit(1);
}
};
@@ -211,11 +211,11 @@ class Application {
private async startWorkerProcess() {
const serviceType = process.env.SERVICE_TYPE;
if (!serviceType || !['http', 'grpc'].includes(serviceType)) {
Logger.error('✌️ Invalid SERVICE_TYPE:', serviceType);
Logger.error('[boot] Invalid SERVICE_TYPE:', serviceType);
process.exit(1);
}
Logger.info(`✌️ ${serviceType} worker started (PID: ${process.pid})`);
Logger.info(`[boot] ${serviceType} worker started (PID: ${process.pid})`);
try {
if (serviceType === 'http') {
@@ -226,7 +226,7 @@ class Application {
process.send?.('ready');
} catch (error) {
Logger.error(`✌️ ${serviceType} worker failed:`, error);
Logger.error(`[boot] ${serviceType} worker failed:`, error);
process.exit(1);
}
}
@@ -269,13 +269,13 @@ class Application {
} else if (msg === 'reregister-crons' && serviceType === 'http') {
// Re-register cron jobs when gRPC worker restarts
try {
Logger.info('✌️ Received reregister-crons message, re-registering cron jobs...');
Logger.info('[boot] Received reregister-crons message, re-registering cron jobs...');
const CronService = (await import('./services/cron')).default;
const cronService = Container.get(CronService);
await cronService.autosave_crontab();
Logger.info('✌️ Cron jobs re-registered successfully');
Logger.info('[boot] Cron jobs re-registered successfully');
} catch (error) {
Logger.error('✌️ Failed to re-register cron jobs:', error);
Logger.error('[boot] Failed to re-register cron jobs:', error);
}
}
});
@@ -297,7 +297,7 @@ class Application {
}
process.exit(0);
} catch (error) {
Logger.error(`✌️ [${serviceType}] Error during shutdown:`, error);
Logger.error(`[${serviceType}] Error during shutdown:`, error);
process.exit(1);
}
}
back/config/grpcCerts.ts
+4 -4
View File
@@ -36,7 +36,7 @@ async function tmpFile(prefix: string): Promise<string> {
}
async function generateAllCerts(): Promise<GrpcTlsConfig> {
Logger.info('Generating gRPC mTLS certificates...');
Logger.info('[boot] Generating gRPC mTLS certificates...');
const caKeyTmp = await tmpFile('ca_key');
const caCertTmp = await tmpFile('ca_cert');
@@ -83,7 +83,7 @@ async function generateAllCerts(): Promise<GrpcTlsConfig> {
const clientKey = await fs.readFile(clientKeyTmp, 'utf-8');
await cleanup();
Logger.info('gRPC mTLS certificates generated successfully');
Logger.info('[boot] gRPC mTLS certificates generated successfully');
return { caCert, serverCert, serverKey, clientCert, clientKey };
} catch (e) {
@@ -101,7 +101,7 @@ async function saveCerts(tlsConfig: GrpcTlsConfig): Promise<void> {
await fs.writeFile(clientCertPath, tlsConfig.clientCert, { mode: 0o644 });
await fs.writeFile(clientKeyPath, tlsConfig.clientKey, { mode: 0o600 });
Logger.info(`gRPC mTLS certificates saved to ${certDir}`);
Logger.info(`[boot] gRPC mTLS certificates saved to ${certDir}`);
}
async function loadExistingCerts(): Promise<GrpcTlsConfig | null> {
@@ -125,7 +125,7 @@ async function loadExistingCerts(): Promise<GrpcTlsConfig | null> {
fs.readFile(clientKeyPath, 'utf-8'),
]);
Logger.info('Loaded existing gRPC mTLS certificates from disk');
Logger.info('[boot] Loaded existing gRPC mTLS certificates from disk');
return { caCert, serverCert, serverKey, clientCert, clientKey };
}
back/loaders/app.ts
+6 -6
View File
@@ -9,20 +9,20 @@ import initFile from './initFile';
export default async ({ app }: { app: Application }) => {
depInjectorLoader();
Logger.info('✌️ Dependency loaded');
Logger.info('[boot] Dependency loaded');
await linkDeps();
Logger.info('✌️ Link deps loaded');
Logger.info('[boot] Link deps loaded');
await initFile();
Logger.info('✌️ Init file loaded');
Logger.info('[boot] Init file loaded');
await initData();
Logger.info('✌️ Init data loaded');
Logger.info('[boot] Init data loaded');
initTask();
Logger.info('✌️ Init task loaded');
Logger.info('[boot] Init task loaded');
expressLoader({ app });
Logger.info('✌️ Express loaded');
Logger.info('[boot] Express loaded');
};
back/loaders/db.ts
+2 -2
View File
@@ -58,8 +58,8 @@ export default async () => {
}
}
Logger.info('✌️ DB loaded');
Logger.info('[boot] DB loaded');
} catch (error) {
Logger.error('✌️ DB load failed', error);
Logger.error('[boot] DB load failed', error);
}
};
back/loaders/initFile.ts
+1 -1
View File
@@ -125,5 +125,5 @@ export default async () => {
}
}
Logger.info('✌️ Init file down');
Logger.info('[boot] Init file down');
};
back/loaders/server.ts
+1 -1
View File
@@ -4,7 +4,7 @@ import Sock from './sock';
export default async ({ server }: { server: Server }) => {
await Sock({ server });
Logger.info('✌️ Sock loaded');
Logger.info('[boot] Sock loaded');
process.on('uncaughtException', (error) => {
Logger.error('Uncaught exception:', error);
back/schedule/addCron.ts
+99 -15
View File
@@ -1,16 +1,79 @@
import { ServerUnaryCall, sendUnaryData } from '@grpc/grpc-js';
import { ServerUnaryCall, sendUnaryData, status } from '@grpc/grpc-js';
import { AddCronRequest, AddCronResponse } from '../protos/cron';
import nodeSchedule from 'node-schedule';
import { scheduleStacks } from './data';
import { runCron } from '../shared/runCron';
import Logger from '../loaders/logger';
import { tf } from '../shared/i18n';
/**
* 预校验 cron 表达式，检测 node-schedule 会拒绝但 cron-parser 会接受的 pattern。
* node-schedule 对 bare /N（字段以 / 开头，如前无星号/数字前缀的 /6）返回 null，
* 提前拦截避免走 scheduleJob 后才发现无效。
*/
const isValidCronField = (cron: string): boolean => {
// 检测 bare /N 模式：字段以 / 开头如 "/6"，或空格后紧跟 "/6"
// node-schedule 会对这种字段返回 null
if (/\s\/\d/.test(cron) || /^\/\d/.test(cron)) {
return false;
}
// 检测 ? 字符：Quartz cron 语法，node-schedule 在大多数位置返回 null
// cron-parser 接受但 node-schedule 拒绝，提前拦截
if (/\?/.test(cron)) {
return false;
}
return true;
};
const addCron = (
call: ServerUnaryCall<AddCronRequest, AddCronResponse>,
callback: sendUnaryData<AddCronResponse>,
) => {
// ===== 第一遍：预校验所有 cron 表达式 =====
const validationErrors: string[] = [];
for (const item of call.request.crons) {
const { id, schedule, extra_schedules } = item;
if (!isValidCronField(schedule)) {
validationErrors.push(
tf(
'任务ID %s: 无效的 cron 表达式 "%s"（不支持裸 /N 步长和 ? 字符）',
String(id),
schedule,
),
);
}
if (extra_schedules?.length) {
extra_schedules.forEach((x) => {
if (!isValidCronField(x.schedule)) {
validationErrors.push(
tf(
'任务ID %s (extra_schedule): 无效的 cron 表达式 "%s"（不支持裸 /N 步长和 ? 字符）',
String(id),
x.schedule,
),
);
}
});
}
}
if (validationErrors.length > 0) {
const details = validationErrors.join('\n');
const err: any = new Error(details);
err.code = status.INVALID_ARGUMENT;
err.details = details;
callback(err, null);
return;
}
// ===== 第二遍：注册所有任务 =====
for (const item of call.request.crons) {
const { id, schedule, command, extra_schedules, name } = item;
// 取消该 id 已有的旧任务
if (scheduleStacks.has(id)) {
scheduleStacks.get(id)?.forEach((x) => x.cancel());
}
@@ -35,20 +98,41 @@ const addCron = (
});
}
scheduleStacks.set(id, [
nodeSchedule.scheduleJob(id, schedule, async () => {
Logger.info(`[schedule][准备运行任务] 命令: ${command}`);
runCron(command, item);
}),
...(extra_schedules?.length
? extra_schedules.map((x) =>
nodeSchedule.scheduleJob(id, x.schedule, async () => {
Logger.info(`[schedule][准备运行任务] 命令: ${command}`);
runCron(command, item);
}),
)
: []),
]);
const mainJob = nodeSchedule.scheduleJob(id, schedule, async () => {
Logger.info(`[schedule][准备运行任务] 命令: ${command}`);
runCron(command, item);
});
if (!mainJob) {
Logger.warn(
'[schedule][创建定时任务] scheduleJob 返回 null（不符合预期，已通过预校验）: 任务ID: %s, cron: %s',
id,
schedule,
);
}
const extraJobs = extra_schedules?.length
? extra_schedules.map((x) => {
const job = nodeSchedule.scheduleJob(id, x.schedule, async () => {
Logger.info(`[schedule][准备运行任务] 命令: ${command}`);
runCron(command, item);
});
if (!job) {
Logger.warn(
'[schedule][创建定时任务] scheduleJob 返回 null（不符合预期，已通过预校验）: 任务ID: %s, cron: %s',
id,
x.schedule,
);
}
return job;
})
: [];
// 过滤 null（兜底保护，正常情况下预校验已拦截）
const jobs = [mainJob, ...extraJobs].filter((x) => x != null);
if (jobs.length > 0) {
scheduleStacks.set(id, jobs);
}
}
callback(null, null);
back/services/cron.ts
+40 -8
View File
@@ -117,12 +117,16 @@ export default class CronService {
},
]);
} catch (error: any) {
// 调度器注册为尽力而为，失败时不阻断 crontab.list 与系统 crontab 的同步，
// 调度器重启后会重新注册（见 autosave_crontab）
this.logger.warn(
'[crontab] Failed to register cron job in scheduler:',
// gRPC 注册失败时回滚 DB 记录，避免产生"僵尸任务"
// （DB 和 crontab.list 有记录但调度器永远不会执行）
await CrontabModel.destroy({ where: { id: doc.id } });
this.logger.error(
'[crontab] Failed to register cron job in scheduler, task creation rolled back:',
error?.message || error,
);
throw new Error(
`${t('调度器注册失败，任务创建已回滚')}: ${(error as any)?.details || error?.message}`,
);
}
}
@@ -166,10 +170,33 @@ export default class CronService {
},
]);
} catch (error: any) {
this.logger.warn(
'[crontab] Failed to register cron job in scheduler:',
// gRPC 注册新任务失败 → 回滚 DB 到旧数据，并尝试恢复旧调度注册
await CrontabModel.update(doc, { where: { id: doc.id } });
if (this.shouldUseCronClient(doc)) {
try {
await cronClient.addCron([
{
name: doc.name || '',
id: String(doc.id),
schedule: doc.schedule!,
command: this.makeCommand(doc),
extra_schedules: doc.extra_schedules || [],
},
]);
} catch (_recoveryError: any) {
this.logger.warn(
'[crontab] Failed to restore old cron job in scheduler after rollback:',
_recoveryError?.message || _recoveryError,
);
}
}
this.logger.error(
'[crontab] Failed to register updated cron job in scheduler, update rolled back:',
error?.message || error,
);
throw new Error(
`${t('调度器注册失败，任务更新已回滚')}: ${(error as any)?.details || error?.message}`,
);
}
}
@@ -726,10 +753,15 @@ export default class CronService {
try {
await cronClient.addCron(crons);
} catch (error: any) {
this.logger.warn(
'[crontab] Failed to register cron job in scheduler:',
// gRPC 注册失败 → 回滚启用状态，避免 DB 显示已启用但调度器未注册
await CrontabModel.update({ isDisabled: 1 }, { where: { id: ids } });
this.logger.error(
'[crontab] Failed to register cron job in scheduler, enable rolled back:',
error?.message || error,
);
throw new Error(
`${t('调度器注册失败，任务启用已回滚')}: ${(error as any)?.details || error?.message}`,
);
}
await this.setCrontab();
}
back/services/grpc.ts
+1 -1
View File
@@ -50,7 +50,7 @@ export class GrpcServerService {
try {
const address = this.formatGrpcAddress(host, grpcPort);
await bindAsync(address, credentials);
Logger.debug(`✌️ gRPC service started successfully on ${address}`);
Logger.debug(`[boot] gRPC service started successfully on ${address}`);
metricsService.record('grpc_service_start', 1, {
port: grpcPort.toString(),
host
back/services/http.ts
+1 -1
View File
@@ -20,7 +20,7 @@ export class HttpServerService {
for (const host of hostsToTry) {
try {
const server = await this.tryListen(expressApp, port, host);
Logger.debug(`✌️ HTTP service started successfully on ${host}:${port}`);
Logger.debug(`[boot] HTTP service started successfully on ${host}:${port}`);
metricsService.record('http_service_start', 1, {
port: port.toString(),
host
back/shared/i18n.ts
+10
View File
@@ -131,6 +131,16 @@ const messages: Record<string, Record<string, string>> = {
'执行before命令...': 'Execute before command...',
'执行after命令...': 'Execute after command...',
'## 执行结束... %s 耗时 %s 秒': '## Execution finished... %s elapsed %s seconds',
'调度器注册失败，任务创建已回滚':
'Scheduler registration failed, task creation rolled back',
'调度器注册失败，任务更新已回滚':
'Scheduler registration failed, task update rolled back',
'调度器注册失败，任务启用已回滚':
'Scheduler registration failed, task enable rolled back',
'任务ID %s: 无效的 cron 表达式 "%s"（不支持裸 /N 步长和 ? 字符）':
'Task ID %s: invalid cron expression "%s" (bare /N steps and ? character not supported)',
'任务ID %s (extra_schedule): 无效的 cron 表达式 "%s"（不支持裸 /N 步长和 ? 字符）':
'Task ID %s (extra_schedule): invalid cron expression "%s" (bare /N steps and ? character not supported)',
},
};
back/validation/schedule.ts
+10
View File
@@ -12,6 +12,16 @@ const validateSchedule = (value: string, helpers: any) => {
return value;
}
// 检测裸 /N 模式：cron-parser 会接受，但 node-schedule 会返回 null
// 提前拦截，避免任务入库后调度器注册失败
if (/\s\/\d/.test(value) || /^\/\d/.test(value)) {
return helpers.error('any.invalid');
}
// 检测 ? 字符：Quartz cron 语法，node-schedule 在大多数字段上返回 null
if (/\?/.test(value)) {
return helpers.error('any.invalid');
}
try {
if (CronExpressionParser.parse(value).hasNext()) {
return value;