05.03 · n8n 源码剖析：Webhook 接入层 Webhook Ingress

本文是专栏「n8n 工作流引擎剖析」第 05 章组件深度剖析的第 03/11 篇承接上一篇《触发器注册中心 Trigger Registry》。讲的是从一个原始 HTTP 请求到调用WorkflowRunner.run之间发生的一切。你在这里读完本文你能从 Express 一路追踪请求直到第一个数据项能解释各种响应模式并能说清为什么 Webhook 节点会在执行记录存在之前就先运行。缩写HTTPHypertext Transfer Protocol超文本传输协议、CORSCross-Origin Resource Sharing跨域资源共享、IDIdentifier标识符、DBDatabase数据库、JSONJavaScript Object NotationJavaScript 对象表示法、V8Google’s JavaScript engine谷歌的 JavaScript 引擎。角色回顾负责HTTP 请求 →找到对应 webhook、加载已发布工作流、生成首个数据项、给出 HTTP 响应。掌握路由 → 数据行的映射通过WebhookService、响应模式、请求/响应对象。不做不遍历图不决定跑在主进程还是工作进程不持久化执行记录。出现于S3、S4、S5以及 S6 末尾那个被延迟发出的回复。内部设计存在 workflowData没有 workflowDataExpressapp.all(/webhook/*path)createWebhookHandlerFor(liveWebhooks,webhook)WebhookRequestHandler.handleRequest方法检查 · CORSOPTIONS→204LiveWebhooks.executeWebhookfindWebhook(path, method)缓存 → DB 静态 → DB 动态加载已发布版本→ new Workflow(...)sanitizeWebhookRequest除非节点在认证白名单中WebhookHelpers.executeWebhookevaluateResponseOptions, parseRequestBodyinvokeWebhook → WebhookService.runWebhook → node.webhook(ctx)handleImmediateWebhookResponseprepareExecutionData → WorkflowRunner.run(...)直接响应不创建执行延迟的 onReceived 回复图注接入层的处理流水线——注意节点自己的webhook()是先执行的它甚至可能决定根本不启动任何执行。路由。AbstractServer把app.all(/endpointWebhook/*path, createWebhookHandlerFor(liveWebhooks, webhook))挂载在请求体解析器之前这样 Webhook 节点才能自己流式处理二进制请求体cli/src/abstract-server.ts第 240–252 行。表单Form以及等待中的 webhook/表单恢复接口有对应的兄弟路由测试 webhook 由TestWebhooks在webhook-test前缀下提供服务。处理器。WebhookRequestHandler.handleRequest拒绝不支持的方法只在存在origin请求头时才应用 CORS用204响应OPTIONS然后把请求交给webhookManager.executeWebhook。错误会被转换成 HTTP 错误响应未知的 webhook →WebhookNotFoundError。查找。WebhookService.findWebhook→ 静态缓存webhook:${method}-${path}→ 静态数据库记录 → 动态路径探测路径中带:param段的情况路径参数会被复制进request.params。加载已发布的版本。loadWebhookExecutionData使用workflow.activeVersion或者在某个开关后面的新发布服务——nodes/connections来自已发布的版本绝不是草稿。构建出一个WorkflowgetBase(...)创建additionalData凭据辅助对象、钩子占位、设置、生产调用下的userId 发布者。表达式隔离实例——仅在需要时才创建。webhookPhaseNeedsIsolate会跳过创建 V8 隔离实例前提是满足一个很常见的场景Webhook 节点 v2 及以上版本、参数是静态的、描述字段能原生解析。只要有任何一点无法证明是静态的就会去获取一个隔离实例。WebhookHelpers.executeWebhookwebhook-helpers.ts第 804 行通过evaluateResponseOptions解析出responseMode等不支持的模式 → HTTP 500。支持的模式有onReceived、lastNode、responseNode、formPage、streaming、hostedChatparseRequestBodyinvokeWebhook→WebhookService.runWebhookwebhook.service.ts第 646 行构建一个WebhookContext调用节点的webhook()。这一步出错会被上报并以一个通用错误作答不会创建任何执行记录。由节点来决定。Webhook 节点的webhook()Webhook.node.ts第 225 行会检查 IP 白名单、机器人过滤、认证方式基础认证/请求头/JWT/n8n OAuth、一个可选的仅在满足条件时运行表达式然后构建出{json:{headers:req.headers,params:req.params,query:req.query,body:req.body}}并返回{ webhookResponse, workflowData: [[item]] }。如果认证失败它会自己写出403/401并返回{ noWebhookResponse: true }没有workflowData→ 不会创建执行。响应模式决定调用方在等什么模式调用方会收到……onReceived默认一旦执行记录存在就立刻收到一个 JSON 回复如果配置了responseData就用它否则是{ message: Workflow was started }lastNode运行结束时最后一个执行节点的输出responseNodeRespond to Webhook节点发送的任何内容streaming运行过程中的分块流sendChunk钩子formPage/hostedChatForm / Chat 触发器对应的界面页面流程启动运行。prepareExecutionData构建出初始的IRunExecutionData——关键是executionData.nodeExecutionStack [{ node: 起始节点, data: { main: workflowData }, source: null }]——然后executionIdawaitContainer.get(WorkflowRunner).run(runData,/*loadStaticData*/true,/*realtime*/!didSendResponse!shouldDeferOnReceivedResponse,existingExecution/* 只有在恢复一次 Wait 时才有值 */,responsePromise);对于onReceived模式回复会延迟到run返回之后这样responseData表达式里就能用上$execution.idwebhook-helpers.ts约第 1262–1290 行。交互关系对象契约触发器注册中心读取它写入的数据行启动时会预先填充静态缓存populateCache节点调用webhook(ctx)期望得到IWebhookResponseDataworkflowData、webhookResponse、noWebhookResponse工作流运行器下一篇run(IWorkflowExecutionDataProcess, …)→ 返回一个执行 ID队列模式对于responseNode/lastNode工作进程会把响应转发回这个进程伸缩队列是本系列后续文章⚓ 回到示例 —— S3 → S5POST /webhook/orders内容为{customer:ACME,amount:100}Content-Type: application/jsoncreateWebhookHandlerFor拼出params.pathorders。调试日志Received webhook POST for path orders。findWebhook(POST,orders)→ 缓存未命中 → 数据库命中(orders, POST)→ 写入缓存。加载已发布版本webhookPhaseNeedsIsolate返回假Webhook v2、参数是静态的→ 不构建隔离实例。runWebhook调用Webhook.webhook(ctx)没有配置认证validateAuth通过结果{workflowData:[[{json:{headers:{content-type:application/json,…:…},params:{},query:{},body:{customer:ACME,amount:100}}}]]}prepareExecutionData把这个数据项放进初始栈里WorkflowRunner.run被调用S6 开始。它返回例如1042直到这时接入层才发出HTTP/1.1 200 OK {message:Workflow was started}失败行为失败情形结果找不到(path, method)对应的记录抛出WebhookNotFoundError→ 404错误信息里会列出这个路径已注册的其他方法webhook()中认证失败节点自己写出 401/403返回noWebhookResponse不创建执行webhook()抛出异常上报给错误报告器返回一个通用错误响应不创建执行不支持的响应模式500提示The response mode … is not valid!调用方断开连接对onReceived模式没有影响——执行记录已经创建好了WorkflowRunner.run抛出异常例如执行前被阻断异常会一路传到 HTTP 响应PreExecuteBlockedError会在运行器里被解包下一篇《工作流运行器 Workflow Runner》讲清楚一次执行是怎么被创建、又被决定跑在哪里的。 返回专栏目录