diff --git a/back/app.ts b/back/app.ts index 7a265da5..78194396 100644 --- a/back/app.ts +++ b/back/app.ts @@ -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((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((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); } } diff --git a/back/config/grpcCerts.ts b/back/config/grpcCerts.ts index 63af59aa..15bd88c5 100644 --- a/back/config/grpcCerts.ts +++ b/back/config/grpcCerts.ts @@ -36,7 +36,7 @@ async function tmpFile(prefix: string): Promise { } async function generateAllCerts(): Promise { - 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 { 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 { 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 { @@ -125,7 +125,7 @@ async function loadExistingCerts(): Promise { 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 }; } diff --git a/back/loaders/app.ts b/back/loaders/app.ts index c9f8e2dc..1c814504 100644 --- a/back/loaders/app.ts +++ b/back/loaders/app.ts @@ -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'); }; diff --git a/back/loaders/db.ts b/back/loaders/db.ts index a8769d38..9a70d91a 100644 --- a/back/loaders/db.ts +++ b/back/loaders/db.ts @@ -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); } }; diff --git a/back/loaders/initFile.ts b/back/loaders/initFile.ts index ac629d63..1c1a93dd 100644 --- a/back/loaders/initFile.ts +++ b/back/loaders/initFile.ts @@ -125,5 +125,5 @@ export default async () => { } } - Logger.info('✌️ Init file down'); + Logger.info('[boot] Init file down'); }; diff --git a/back/loaders/server.ts b/back/loaders/server.ts index dd010777..e3adb624 100644 --- a/back/loaders/server.ts +++ b/back/loaders/server.ts @@ -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); diff --git a/back/schedule/addCron.ts b/back/schedule/addCron.ts index a8e2e9fb..d651de10 100644 --- a/back/schedule/addCron.ts +++ b/back/schedule/addCron.ts @@ -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, callback: sendUnaryData, ) => { + // ===== 第一遍:预校验所有 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); diff --git a/back/services/cron.ts b/back/services/cron.ts index 001eb331..3a7282a5 100644 --- a/back/services/cron.ts +++ b/back/services/cron.ts @@ -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(); } diff --git a/back/services/grpc.ts b/back/services/grpc.ts index 7432ca70..2a69843f 100644 --- a/back/services/grpc.ts +++ b/back/services/grpc.ts @@ -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 diff --git a/back/services/http.ts b/back/services/http.ts index bbce8ff7..a47d49c8 100644 --- a/back/services/http.ts +++ b/back/services/http.ts @@ -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 diff --git a/back/shared/i18n.ts b/back/shared/i18n.ts index 35866c2f..1edd2bb0 100644 --- a/back/shared/i18n.ts +++ b/back/shared/i18n.ts @@ -131,6 +131,16 @@ const messages: Record> = { '执行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)', }, }; diff --git a/back/validation/schedule.ts b/back/validation/schedule.ts index d5ba6a54..db9d31f8 100644 --- a/back/validation/schedule.ts +++ b/back/validation/schedule.ts @@ -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;