本文是专栏「n8n 工作流引擎剖析」第 05 章组件深度剖析的第 04/11 篇承接上一篇《Webhook 接入层 Webhook Ingress》。核心是WorkflowRunner与ActiveExecutions。你在这里读完本文你会知道一次执行是如何被创建出来的、n8n 如何决定它在哪里运行以及围绕着一次正在运行的执行有哪些簿记工作超时、取消、响应 Promise。缩写DBDatabase数据库、IDIdentifier标识符、HTTPHypertext Transfer Protocol超文本传输协议、JSONJavaScript Object NotationJavaScript 对象表示法。角色回顾负责把用这些首批数据项运行这个工作流变成一条持久化的执行记录和一次正在进行中的运行——可以在本地跑也可以经由队列跑。掌握执行 ID、设置超时时间、正在等待的 HTTP 调用方的响应 Promise、取消句柄。不做不知道节点内部如何工作不在节点之间搬运数据。出现于S5 → S6以及每一次手动运行、每一次由触发器发起的运行。内部设计cli/src/下两个相互协作的类类文件角色WorkflowRunnerworkflow-runner.ts编排的配方run、runMainProcess、enqueueExecutionActiveExecutionsactive-executions.ts正在运行的执行的内存登记表创建数据库记录add持有PCancelable句柄、响应 Promise以及执行后置PromiseWorkflowRunner.run()第 407 行是否拒绝通过是否run(data, loadStaticData, realtime, existingExecution, responsePromise)是否走 engine v2engineV2Dispatcher.start超出本文范围establishContextForPersistencecredentialsPermissionChecker.check(workflow nodes)activeExecutions.add failExecution→ 返回 idprepareNewExecution → WorkflowactiveExecutions.add→ 执行记录状态 new绑定响应 Promise启动流式心跳executions.mode queue且非 manualenqueueExecution→ Bull 任务runMainProcess→ 进程内运行WorkflowExecute执行后置 Promise图注运行器是一棵决策树最终走到两条运行路径之一无论走哪条都会返回一个执行 ID。持久化上下文establishContextForPersistence和凭据访问检查credentialsPermissionChecker.check排在最前面——如果一个工作流引用了当前执行方所在项目无权使用的凭据它会作为一次执行而失败因而在历史记录里可见而不是直接把请求丢弃。prepareNewExecution构建Workflow对象在手动/评估模式下会解析出 pin data。ActiveExecutions.add第 71 行通过executionPersistence.create({ data, mode, finished:false, workflowData, status:new, workflowId, … })插入记录——此时整个IRunExecutionData包括预先填好的nodeExecutionStack就已经存进去了——并预留并发容量ConcurrencyCapacityReservationevaluation模式下跳过。在常规模式下会立即执行setRunning。决定在哪里运行。constshouldEnqueueOFFLOAD_MANUAL_EXECUTIONS_TO_WORKERStrue?modequeue:modequeuedata.executionMode!manual;所以在队列模式下生产运行会进工作进程手动编辑器运行留在主进程除非某个环境变量开关另有规定。runMainProcess第 569 行设置一个软超时workflowSettings.executionTimeout或全局默认值上限由maxTimeout决定构建带执行超时时间戳的additionalData调用setRunning(executionId)组装生命周期钩子getLifecycleHooksForRegularMain加上sendResponse解析 HTTP 响应 Promise和sendChunk流式输出两个处理器然后调用new WorkflowExecute(additionalData, mode, data.executionData).processRunExecutionData(workflow)——一个PCancelableIRun——并把它挂到登记表上。完成时调用finalizeExecution(executionId, fullRunData)。enqueueExecution第 740 行→ 见下一篇《伸缩队列 Scaling Queue》。超时是软的。计时器触发时stopExecution(id, TimeoutExecutionCancelledError)会在当前节点跑完之后才取消——引擎会在节点之间检查shouldStopExecuting()。ActiveExecutions的职责方法用途add第 71 行创建记录或续接一条已存在的等待中记录、预留容量、返回 idattachWorkflowExecution第 199 行存下可取消的 Promise这样stopExecution才能取消它attachResponsePromise/resolveResponsePromise连通到在lastNode/responseNode模式下等待中的 HTTP 调用方stopExecution第 225 行以一个带类型的ExecutionCancelledError取消手动、超时、关闭finalizeExecution第 255 行解析执行后置 Promise、释放容量、从登记表中移除getPostExecutePromise第 295 行让调用方可以等待完成供lastNode模式和子工作流使用交互关系对象契约Webhook 接入层、触发器、编辑器手动运行、子工作流节点调用run(IWorkflowExecutionDataProcess, …)得到一个执行 ID引擎本系列后续文章构造WorkflowExecute收到IRun伸缩队列下一篇调用addJob监听任务消息钩子本系列后续文章根据当前进程角色选用哪一套钩子并发控制容量预留可能在执行开始前就先把它延后⚓ 回到示例 —— S5 → S6接入层调用run(...)带上executionMode: webhook、包含预填好的栈Order Webhook 该数据项的executionDataworkflowData 已发布版本。credentialsPermissionChecker.check(wf123, nodes)——httpHeaderAuth凭据cred42属于 Ada 的项目 → 通过。ActiveExecutions.add插入execution_entity状态new模式webhook及其execution_data负载假设 id 为1042。队列模式且模式 ≠manual→enqueueExecution(1042, wf123, data, loadStaticDatafalse, realtimefalse, …)。realtime为假因为onReceived会延迟回复!didSendResponse !shouldDeferOnReceivedResponse→false所以任务优先级是100而不是50。run返回1042接入层发出此前延迟的200 {message:Workflow was started}。在常规模式下同样的调用会直接走到runMainProcessS7–S12 会在这个进程里原地发生。失败行为失败情形结果凭据访问被拒绝创建执行记录并立即失败failExecution返回 id在历史记录中可见无法建立持久化上下文同上——是一次失败的执行而不是一个直接抛出的 HTTP 错误enqueueExecution抛出异常Redis 挂了用工作进程式的钩子跑一遍processError让执行以error结束然后重新抛出异常超时软取消引擎会在下一次循环迭代时注意到shouldStopExecuting把运行标记为canceled结果中带有一个TimeoutExecutionCancelledError关闭对正在运行的执行抛出SystemShutdownExecutionCancelledError执行后置 Promise 被拒绝记录日志并上报取消类错误会被忽略下一篇《伸缩队列 Scaling Queue队列模式》讲清楚任务是怎么从主进程搬到工作进程的。 返回专栏目录