n8n-io/n8n
1
1
Fork 0
mirror of https://github.com/n8n-io/n8n.git synced 2024-09-19 08:57:09 +03:00
Code Issues Projects Releases Wiki Activity

fix(core): Ensure graceful shutdown for workers (#9547)

Browse Source
This commit is contained in:
Iván Ovejero 2024-05-31 15:41:00 +02:00 committed by GitHub
parent c5e0cf0137
commit 7fc00d8d10
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
2 changed files with 20 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
packages/cli/src/Queue.ts
View File

@ -109,8 +109,11 @@ export class Queue {
return await this.jobQueue.client.ping();
}
async pause(isLocal?: boolean): Promise<void> {
return await this.jobQueue.pause(isLocal);
async pause({
isLocal,
doNotWaitActive,
}: { isLocal?: boolean; doNotWaitActive?: boolean } = {}): Promise<void> {
return await this.jobQueue.pause(isLocal, doNotWaitActive);
}
getBullObjectInstance(): JobQueue {

20
packages/cli/src/commands/worker.ts
View File

@ -63,23 +63,23 @@ export class Worker extends BaseCommand {
async stopProcess() {
this.logger.info('Stopping n8n...');
// Stop accepting new jobs
await Worker.jobQueue.pause(true);
// Stop accepting new jobs, `doNotWaitActive` allows reporting progress
await Worker.jobQueue.pause({ isLocal: true, doNotWaitActive: true });
try {
await this.externalHooks?.run('n8n.stop', []);
const hardStopTime = Date.now() + this.gracefulShutdownTimeoutInS;
const hardStopTimeMs = Date.now() + this.gracefulShutdownTimeoutInS * 1000;
// Wait for active workflow executions to finish
let count = 0;
while (Object.keys(Worker.runningJobs).length !== 0) {
if (count++ % 4 === 0) {
const waitLeft = Math.ceil((hardStopTime - Date.now()) / 1000);
const waitLeft = Math.ceil((hardStopTimeMs - Date.now()) / 1000);
this.logger.info(
`Waiting for ${
Object.keys(Worker.runningJobs).length
} active executions to finish... (wait ${waitLeft} more seconds)`,
} active executions to finish... (max wait ${waitLeft} more seconds)`,
);
}
@ -483,6 +483,16 @@ export class Worker extends BaseCommand {
await this.setupHealthMonitor();
}
if (process.stdout.isTTY) {
process.stdin.setRawMode(true);
process.stdin.resume();
process.stdin.setEncoding('utf8');
process.stdin.on('data', (key: string) => {
if (key.charCodeAt(0) === 3) process.kill(process.pid, 'SIGINT'); // ctrl+c
});
}
// Make sure that the process does not close
await new Promise(() => {});
}