admin/n8n-enterprise-unlocked
1
0
Fork 0
mirror of https://github.com/Abdulazizzn/n8n-enterprise-unlocked.git synced 2025-12-17 18:12:04 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(core): Add check that queue is defined and remove cyclic dependency (#7404)

Browse Source 
In a rare edge case an undefined queue could be returned - this should
not happen and now an error is thrown.
Also using the opportunity to remove a cyclic dependency from the Queue.
This commit is contained in:
Michael Auerswald
2023-10-13 11:53:59 +02:00
committed by GitHub
parent 609f0837cf
commit 45f2ef373e
5 changed files with 49 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -22,7 +22,7 @@ import * as WorkflowExecuteAdditionalData from '@/WorkflowExecuteAdditionalData'
import { PermissionChecker } from '@/UserManagement/PermissionChecker';
import config from '@/config';
import type { Job, JobId, JobQueue, JobResponse, WebhookResponse } from '@/Queue';
import type { Job, JobId, JobResponse, WebhookResponse } from '@/Queue';
import { Queue } from '@/Queue';
import { generateFailedExecutionFromError } from '@/WorkflowHelpers';
import { N8N_VERSION } from '@/constants';
@@ -61,7 +61,7 @@ export class Worker extends BaseCommand {
[jobId: string]: WorkerJobStatusSummary;
} = {};
static jobQueue: JobQueue;
static jobQueue: Queue;
redisSubscriber: RedisServicePubSubSubscriber;
@@ -335,15 +335,14 @@ export class Worker extends BaseCommand {
`Opening Redis connection to listen to messages with timeout ${redisConnectionTimeoutLimit}`,
);
const queue = Container.get(Queue);
await queue.init();
Worker.jobQueue = Container.get(Queue);
await Worker.jobQueue.init();
this.logger.debug('Queue singleton ready');
Worker.jobQueue = queue.getBullObjectInstance();
void Worker.jobQueue.process(flags.concurrency, async (job) =>
this.runJob(job, this.nodeTypes),
);
Worker.jobQueue.on('global:progress', (jobId: JobId, progress) => {
Worker.jobQueue.getBullObjectInstance().on('global:progress', (jobId: JobId, progress) => {
// Progress of a job got updated which does get used
// to communicate that a job got canceled.
@@ -359,7 +358,7 @@ export class Worker extends BaseCommand {
let lastTimer = 0;
let cumulativeTimeout = 0;
Worker.jobQueue.on('error', (error: Error) => {
Worker.jobQueue.getBullObjectInstance().on('error', (error: Error) => {
if (error.toString().includes('ECONNREFUSED')) {
const now = Date.now();
if (now - lastTimer > 30000) {
@@ -423,7 +422,7 @@ export class Worker extends BaseCommand {
// if it loses the connection to redis
try {
// Redis ping
await Worker.jobQueue.client.ping();
await Worker.jobQueue.ping();
} catch (e) {
LoggerProxy.error('No Redis connection!', e as Error);
const error = new ResponseHelper.ServiceUnavailableError('No Redis connection!');