mirror of
https://github.com/n8n-io/n8n.git
synced 2026-05-30 16:26:59 +02:00
0c6169ee22
# Motivation In Queue mode, finished executions would cause the main instance to always pull all execution data from the database, unflatten it and then use it to send out event log events and telemetry events, as well as required returns to Respond to Webhook nodes etc. This could cause OOM errors when the data was large, since it had to be fully unpacked and transformed on the main instance’s side, using up a lot of memory (and time). This PR attempts to limit this behaviour to only happen in those required cases where the data has to be forwarded to some waiting webhook, for example. # Changes Execution data is only required in cases, where the active execution has a `postExecutePromise` attached to it. These usually forward the data to some other endpoint (e.g. a listening webhook connection). By adding a helper `getPostExecutePromiseCount()`, we can decide that in cases where there is nothing listening at all, there is no reason to pull the data on the main instance. Previously, there would always be postExecutePromises because the telemetry events were called. Now, these have been moved into the workers, which have been given the various InternalHooks calls to their hook function arrays, so they themselves issue these telemetry and event calls. This results in all event log messages to now be logged on the worker’s event log, as well as the worker’s eventbus being the one to send out the events to destinations. The main event log does…pretty much nothing. We are not logging executions on the main event log any more, because this would require all events to be replicated 1:1 from the workers to the main instance(s) (this IS possible and implemented, see the worker’s `replicateToRedisEventLogFunction` - but it is not enabled to reduce the amount of traffic over redis). Partial events in the main log could confuse the recovery process and would result in, ironically, the recovery corrupting the execution data by considering them crashed. # Refactor I have also used the opportunity to reduce duplicate code and move some of the hook functionality into `packages/cli/src/executionLifecycleHooks/shared/sharedHookFunctions.ts` in preparation for a future full refactor of the hooks
108 lines
3.1 KiB
TypeScript
108 lines
3.1 KiB
TypeScript
import type Bull from 'bull';
|
|
import { Service } from 'typedi';
|
|
import type { ExecutionError, IExecuteResponsePromiseData } from 'n8n-workflow';
|
|
import { ActiveExecutions } from '@/ActiveExecutions';
|
|
import * as WebhookHelpers from '@/WebhookHelpers';
|
|
import {
|
|
getRedisClusterClient,
|
|
getRedisClusterNodes,
|
|
getRedisPrefix,
|
|
getRedisStandardClient,
|
|
} from './services/redis/RedisServiceHelper';
|
|
import type { RedisClientType } from './services/redis/RedisServiceBaseClasses';
|
|
import config from '@/config';
|
|
|
|
export type JobId = Bull.JobId;
|
|
export type Job = Bull.Job<JobData>;
|
|
export type JobQueue = Bull.Queue<JobData>;
|
|
|
|
export interface JobData {
|
|
executionId: string;
|
|
loadStaticData: boolean;
|
|
}
|
|
|
|
export interface JobResponse {
|
|
success: boolean;
|
|
error?: ExecutionError;
|
|
}
|
|
|
|
export interface WebhookResponse {
|
|
executionId: string;
|
|
response: IExecuteResponsePromiseData;
|
|
}
|
|
|
|
@Service()
|
|
export class Queue {
|
|
private jobQueue: JobQueue;
|
|
|
|
constructor(private activeExecutions: ActiveExecutions) {}
|
|
|
|
async init() {
|
|
const bullPrefix = config.getEnv('queue.bull.prefix');
|
|
const prefix = getRedisPrefix(bullPrefix);
|
|
const clusterNodes = getRedisClusterNodes();
|
|
const usesRedisCluster = clusterNodes.length > 0;
|
|
// eslint-disable-next-line @typescript-eslint/naming-convention
|
|
const { default: Bull } = await import('bull');
|
|
// eslint-disable-next-line @typescript-eslint/naming-convention
|
|
const { default: Redis } = await import('ioredis');
|
|
// Disabling ready check is necessary as it allows worker to
|
|
// quickly reconnect to Redis if Redis crashes or is unreachable
|
|
// for some time. With it enabled, worker might take minutes to realize
|
|
// redis is back up and resume working.
|
|
// More here: https://github.com/OptimalBits/bull/issues/890
|
|
this.jobQueue = new Bull('jobs', {
|
|
prefix,
|
|
createClient: (type, clientConfig) =>
|
|
usesRedisCluster
|
|
? getRedisClusterClient(Redis, clientConfig, (type + '(bull)') as RedisClientType)
|
|
: getRedisStandardClient(Redis, clientConfig, (type + '(bull)') as RedisClientType),
|
|
});
|
|
|
|
this.jobQueue.on('global:progress', (jobId, progress: WebhookResponse) => {
|
|
this.activeExecutions.resolveResponsePromise(
|
|
progress.executionId,
|
|
WebhookHelpers.decodeWebhookResponse(progress.response),
|
|
);
|
|
});
|
|
}
|
|
|
|
async add(jobData: JobData, jobOptions: object): Promise<Job> {
|
|
return this.jobQueue.add(jobData, jobOptions);
|
|
}
|
|
|
|
async getJob(jobId: JobId): Promise<Job | null> {
|
|
return this.jobQueue.getJob(jobId);
|
|
}
|
|
|
|
async getJobs(jobTypes: Bull.JobStatus[]): Promise<Job[]> {
|
|
return this.jobQueue.getJobs(jobTypes);
|
|
}
|
|
|
|
getBullObjectInstance(): JobQueue {
|
|
return this.jobQueue;
|
|
}
|
|
|
|
/**
|
|
*
|
|
* @param job A Job instance
|
|
* @returns boolean true if we were able to securely stop the job
|
|
*/
|
|
async stopJob(job: Job): Promise<boolean> {
|
|
if (await job.isActive()) {
|
|
// Job is already running so tell it to stop
|
|
await job.progress(-1);
|
|
return true;
|
|
}
|
|
// Job did not get started yet so remove from queue
|
|
try {
|
|
await job.remove();
|
|
return true;
|
|
} catch (e) {
|
|
await job.progress(-1);
|
|
}
|
|
|
|
return false;
|
|
}
|
|
}
|