mudhorn/n8n
1
0
Fork 0
mirror of https://github.com/n8n-io/n8n.git synced 2025-01-11 12:57:29 -08:00
Code Issues Actions Packages Projects Releases Wiki Activity

WIP

Browse source
This commit is contained in:
Iván Ovejero 2024-11-19 18:00:20 +01:00
parent 0715ec2512
commit 403b76bdfc
No known key found for this signature in database
3 changed files with 21 additions and 24 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
packages/@n8n/task-runner/src/message-types.ts
View file

@ -21,6 +21,7 @@ export namespace BrokerMessage {
type: 'broker:taskofferaccept'; type: 'broker:taskofferaccept';
taskId: string; taskId: string;
offerId: string; offerId: string;
requestId: string;
} }
export interface TaskCancel { export interface TaskCancel {
@ -188,6 +189,7 @@ export namespace RunnerMessage {
export interface TaskDeferred { export interface TaskDeferred {
type: 'runner:taskdeferred'; type: 'runner:taskdeferred';
taskId: string; taskId: string;
requestId: string;
} }
export interface TaskDone { export interface TaskDone {

6
packages/cli/src/runners/errors.ts
View file

@ -6,10 +6,4 @@ export class TaskRejectError extends ApplicationError {
} }
} }
export class TaskDeferredError extends ApplicationError {
constructor() {
super('Task deferred until runner is ready', { level: 'info' });
}
}
export class TaskError extends ApplicationError {} export class TaskError extends ApplicationError {}

37
packages/cli/src/runners/task-broker.service.ts
View file

@ -13,7 +13,7 @@ import config from '@/config';
import { Time } from '@/constants'; import { Time } from '@/constants';
import { Logger } from '@/logging/logger.service'; import { Logger } from '@/logging/logger.service';
import { TaskDeferredError, TaskRejectError } from './errors'; import { TaskRejectError } from './errors';
import { TaskRunnerTimeoutError } from './errors/task-runner-timeout.error'; import { TaskRunnerTimeoutError } from './errors/task-runner-timeout.error';
import { RunnerLifecycleEvents } from './runner-lifecycle-events'; import { RunnerLifecycleEvents } from './runner-lifecycle-events';
@ -59,7 +59,7 @@ type RunnerAcceptCallback = () => void;
type RequesterAcceptCallback = ( type RequesterAcceptCallback = (
settings: RequesterMessage.ToBroker.TaskSettings['settings'], settings: RequesterMessage.ToBroker.TaskSettings['settings'],
) => void; ) => void;
type TaskRejectCallback = (reason: TaskRejectError | TaskDeferredError) => void; type TaskRejectCallback = (reason: TaskRejectError) => void;
@Service() @Service()
export class TaskBroker { export class TaskBroker {
@ -163,7 +163,7 @@ export class TaskBroker {
this.handleRunnerReject(message.taskId, message.reason); this.handleRunnerReject(message.taskId, message.reason);
break; break;
case 'runner:taskdeferred': case 'runner:taskdeferred':
this.handleRunnerDeferred(message.taskId); this.handleRunnerDeferred(message.taskId, message.requestId);
break; break;
case 'runner:taskoffer': case 'runner:taskoffer':
this.taskOffered({ this.taskOffered({
@ -233,12 +233,15 @@ export class TaskBroker {
} }
} }
handleRunnerDeferred(taskId: Task['id']) { handleRunnerDeferred(taskId: Task['id'], requestId: TaskRequest['requestId']) {
const acceptReject = this.runnerAcceptRejects.get(taskId); const request = this.pendingTaskRequests.find((ptr) => ptr.requestId === requestId);
if (acceptReject) {
acceptReject.reject(new TaskDeferredError()); if (!request) throw new ApplicationError('Failed to find request to defer');
this.runnerAcceptRejects.delete(taskId);
} this.logger.info(`Request for task (${taskId}) deferred until runner is ready`);
request.acceptInProgress = false;
this.pendingTaskRequests.push(request);
} }
async handleDataRequest( async handleDataRequest(
@ -505,16 +508,19 @@ export class TaskBroker {
const acceptPromise = new Promise((resolve, reject) => { const acceptPromise = new Promise((resolve, reject) => {
this.runnerAcceptRejects.set(taskId, { accept: resolve as () => void, reject }); this.runnerAcceptRejects.set(taskId, { accept: resolve as () => void, reject });
// TODO: customisable timeout if (offer.validFor !== -1) {
setTimeout(() => { // TODO: customisable timeout
reject('Runner timed out'); setTimeout(() => {
}, 2000); reject('Runner timed out');
}, 2000);
}
}); });
await this.messageRunner(offer.runnerId, { await this.messageRunner(offer.runnerId, {
type: 'broker:taskofferaccept', type: 'broker:taskofferaccept',
offerId: offer.offerId, offerId: offer.offerId,
taskId, taskId,
requestId: request.requestId,
}); });
await acceptPromise; await acceptPromise;
@ -524,11 +530,6 @@ export class TaskBroker {
this.logger.info(`Task (${taskId}) rejected by Runner with reason "${e.reason}"`); this.logger.info(`Task (${taskId}) rejected by Runner with reason "${e.reason}"`);
return; return;
} }
if (e instanceof TaskDeferredError) {
this.logger.info(`Task (${taskId}) deferred until runner is ready`);
this.pendingTaskRequests.push(request); // will settle on receiving task offer from runner
return;
}
throw e; throw e;
} }