mudhorn/n8n
1
0
Fork 0
mirror of https://github.com/n8n-io/n8n.git synced 2024-12-24 20:24:05 -08:00
Code Issues Actions 1 Packages Projects Releases Wiki Activity

fix(core): Restore queue listeners for webhook process (#10781)

Browse source
This commit is contained in:
Iván Ovejero 2024-09-12 12:55:24 +02:00 committed by GitHub
parent b2b6190cc0
commit 86f4877bab
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
2 changed files with 26 additions and 10 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

28
packages/cli/src/scaling/__tests__/scaling.service.test.ts
View file

@ -51,7 +51,7 @@ describe('ScalingService', () => {
let scalingService: ScalingService; let scalingService: ScalingService;
let registerMainListenersSpy: jest.SpyInstance; let registerMainOrWebhookListenersSpy: jest.SpyInstance;
let registerWorkerListenersSpy: jest.SpyInstance; let registerWorkerListenersSpy: jest.SpyInstance;
let scheduleQueueRecoverySpy: jest.SpyInstance; let scheduleQueueRecoverySpy: jest.SpyInstance;
let stopQueueRecoverySpy: jest.SpyInstance; let stopQueueRecoverySpy: jest.SpyInstance;
@ -86,8 +86,11 @@ describe('ScalingService', () => {
// @ts-expect-error Private method // @ts-expect-error Private method
ScalingService.prototype.scheduleQueueRecovery = jest.fn(); ScalingService.prototype.scheduleQueueRecovery = jest.fn();
// @ts-expect-error Private method registerMainOrWebhookListenersSpy = jest.spyOn(
registerMainListenersSpy = jest.spyOn(scalingService, 'registerMainListeners'); scalingService,
// @ts-expect-error Private method
'registerMainOrWebhookListeners',
);
// @ts-expect-error Private method // @ts-expect-error Private method
registerWorkerListenersSpy = jest.spyOn(scalingService, 'registerWorkerListeners'); registerWorkerListenersSpy = jest.spyOn(scalingService, 'registerWorkerListeners');
// @ts-expect-error Private method // @ts-expect-error Private method
@ -102,7 +105,7 @@ describe('ScalingService', () => {
await scalingService.setupQueue(); await scalingService.setupQueue();
expect(Bull).toHaveBeenCalledWith(...bullConstructorArgs); expect(Bull).toHaveBeenCalledWith(...bullConstructorArgs);
expect(registerMainListenersSpy).toHaveBeenCalled(); expect(registerMainOrWebhookListenersSpy).toHaveBeenCalled();
expect(registerWorkerListenersSpy).not.toHaveBeenCalled(); expect(registerWorkerListenersSpy).not.toHaveBeenCalled();
expect(scheduleQueueRecoverySpy).toHaveBeenCalled(); expect(scheduleQueueRecoverySpy).toHaveBeenCalled();
}); });
@ -115,7 +118,7 @@ describe('ScalingService', () => {
await scalingService.setupQueue(); await scalingService.setupQueue();
expect(Bull).toHaveBeenCalledWith(...bullConstructorArgs); expect(Bull).toHaveBeenCalledWith(...bullConstructorArgs);
expect(registerMainListenersSpy).toHaveBeenCalled(); expect(registerMainOrWebhookListenersSpy).toHaveBeenCalled();
expect(registerWorkerListenersSpy).not.toHaveBeenCalled(); expect(registerWorkerListenersSpy).not.toHaveBeenCalled();
expect(scheduleQueueRecoverySpy).not.toHaveBeenCalled(); expect(scheduleQueueRecoverySpy).not.toHaveBeenCalled();
}); });
@ -130,7 +133,20 @@ describe('ScalingService', () => {
expect(Bull).toHaveBeenCalledWith(...bullConstructorArgs); expect(Bull).toHaveBeenCalledWith(...bullConstructorArgs);
expect(registerWorkerListenersSpy).toHaveBeenCalled(); expect(registerWorkerListenersSpy).toHaveBeenCalled();
expect(registerMainListenersSpy).not.toHaveBeenCalled(); expect(registerMainOrWebhookListenersSpy).not.toHaveBeenCalled();
});
});
describe('webhook', () => {
it('should set up a queue + listeners', async () => {
// @ts-expect-error Private field
scalingService.instanceType = 'webhook';
await scalingService.setupQueue();
expect(Bull).toHaveBeenCalledWith(...bullConstructorArgs);
expect(registerWorkerListenersSpy).not.toHaveBeenCalled();
expect(registerMainOrWebhookListenersSpy).toHaveBeenCalled();
}); });
}); });
}); });

8
packages/cli/src/scaling/scaling.service.ts
View file

@ -209,8 +209,8 @@ export class ScalingService {
throw error; throw error;
}); });
if (this.instanceType === 'main') { if (this.instanceType === 'main' || this.instanceType === 'webhook') {
this.registerMainListeners(); this.registerMainOrWebhookListeners();
} else if (this.instanceType === 'worker') { } else if (this.instanceType === 'worker') {
this.registerWorkerListeners(); this.registerWorkerListeners();
} }
@ -246,9 +246,9 @@ export class ScalingService {
} }
/** /**
* Register listeners on a `main` process for Bull queue events. * Register listeners on a `main` or `webhook` process for Bull queue events.
*/ */
private registerMainListeners() { private registerMainOrWebhookListeners() {
this.queue.on('global:progress', (_jobId: JobId, msg: unknown) => { this.queue.on('global:progress', (_jobId: JobId, msg: unknown) => {
if (!this.isPubSubMessage(msg)) return; if (!this.isPubSubMessage(msg)) return;