open-source/qinglong
1
0
Fork 0
mirror of https://github.com/whyour/qinglong.git synced 2025-12-15 08:25:38 +08:00
Code Issues Packages Projects Releases Wiki Activity

Re-register cron jobs when gRPC worker restarts

Browse Source 
When the gRPC worker restarts, the scheduled tasks need to be re-added to the new gRPC server instance. This fix:

1. Tracks the HTTP worker reference in the master process
2. Sends a 'reregister-crons' message to the HTTP worker after gRPC restarts
3. HTTP worker calls autosave_crontab() to re-register all cron jobs with the new gRPC server

This ensures scheduled tasks continue to work after a gRPC worker restart.

Co-authored-by: whyour <22700758+whyour@users.noreply.github.com>
This commit is contained in:
copilot-swe-agent[bot] 2025-11-11 16:48:25 +00:00
parent 6ede8139ce
commit b33d994316
1 changed files with 24 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

26
back/app.ts
View File

@ -24,6 +24,7 @@ class Application {
private grpcServerService?: GrpcServerService; private grpcServerService?: GrpcServerService;
private isShuttingDown = false; private isShuttingDown = false;
private workerMetadataMap = new Map<number, WorkerMetadata>(); private workerMetadataMap = new Map<number, WorkerMetadata>();
private httpWorker?: Worker;
constructor() { constructor() {
this.app = express(); this.app = express();
@ -60,7 +61,7 @@ class Application {
this.waitForWorkerReady(grpcWorker, 30000) this.waitForWorkerReady(grpcWorker, 30000)
.then(() => { .then(() => {
Logger.info('gRPC worker is ready, starting HTTP worker'); Logger.info('gRPC worker is ready, starting HTTP worker');
this.forkWorker('http'); this.httpWorker = this.forkWorker('http');
}) })
.catch((error) => { .catch((error) => {
Logger.error('Failed to wait for gRPC worker:', error); Logger.error('Failed to wait for gRPC worker:', error);
@ -81,6 +82,15 @@ class Application {
this.waitForWorkerReady(newGrpcWorker, 30000) this.waitForWorkerReady(newGrpcWorker, 30000)
.then(() => { .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');
} catch (error) {
Logger.error('Failed to send reregister-crons message:', error);
}
}
}) })
.catch((error) => { .catch((error) => {
Logger.error('Failed to restart gRPC worker:', error); Logger.error('Failed to restart gRPC worker:', error);
@ -89,6 +99,7 @@ class Application {
} else { } else {
// For HTTP worker, just restart it // For HTTP worker, just restart it
const newWorker = this.forkWorker(metadata.serviceType); 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})`);
} }
} }
@ -248,9 +259,20 @@ class Application {
} }
private setupWorkerShutdown(serviceType: string) { private setupWorkerShutdown(serviceType: string) {
process.on('message', (msg) => { process.on('message', async (msg) => {
if (msg === 'shutdown') { if (msg === 'shutdown') {
this.gracefulShutdown(serviceType); this.gracefulShutdown(serviceType);
} 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...');
const CronService = (await import('./services/cron')).default;
const cronService = Container.get(CronService);
await cronService.autosave_crontab();
Logger.info('Cron jobs re-registered successfully');
} catch (error) {
Logger.error('Failed to re-register cron jobs:', error);
}
} }
}); });