Rivet Actors Runner 停机机制：从 ToServerStopping 到 ToClientClose 的完整 Shutdown 流程解析

Rivet Actors Runner 停机机制从 ToServerStopping 到 ToClientClose 的完整 Shutdown 流程解析【免费下载链接】actorsRivet Actors are the primitive for stateful workloads. Built for AI agents, collaborative apps, and durable execution.项目地址: https://gitcode.com/GitHub_Trending/riv/actorsRivet Actors 的 Runner 是承载 Actor 工作负载的计算节点其生命周期中最关键的环节之一是停机Shutdown——即如何在停机期间既保证正在运行的 Actor 有机会优雅收尾又能确保 Pegboard 控制面最终一定完成资源回收。本文基于内部设计文档 RUNNER_SHUTDOWN.md结合pegboard工作流与 TypeScript Runner SDK 的实际源码完整拆解Runner 自发起停机与Rivet 发起停机两条路径的每一步信号流转、draining 状态语义、runner_lost_threshold超时机制以及 Actor 侧的 Lost/GoingAway 补偿逻辑。两种停机路径概览内部文档 RUNNER_SHUTDOWN.md 将停机分为两大类核心区别只在于第一个信号从哪里来1. Runner 自发起停机Self initiated shutdownRunner 向其 Runner WS 发送ToServerStoppingRunner WS 将ToServerStopping代理转发给 Runner WFRunner WorkflowRunner WF 将自己置为 draining 状态阻止后续再有新的 Actor 分配到该 RunnerRunner WF 向所有 Actor WF 发送GoingAway信号一旦超过 runner lost thresholdRunner WF 向 Runner WS 发送ToClientCloseRunner WS 关闭与 Runner 的连接并明确告知 Runner 不要尝试重连。2. Rivet 发起停机Rivet initiated shutdownRunner WF 直接收到Stop信号不经过ToServerStoppingRunner WF 向所有 Actor WF 发送GoingAway信号一旦超过 runner lost thresholdRunner WF 向 Runner WS 发送ToClientCloseRunner WS 关闭与 Runner 的连接并告知 Runner 不要重连。两条路径从第 3 步自发起/第 2 步Rivet 发起开始完全汇合共享同一套 draining → 等待 → 关闭逻辑。这个设计的意义在于无论是计算节点自己优雅退出如SIGTERM还是控制面主动驱逐如运维操作、驱逐工作流Actor 的善后处理路径都完全一致。从配套的生命周期文档 RUNNER_LIFECYCLE.md 中的 Shutdown 时序图可以看到完整图景Runner 端SDK在发送ToServerStopping后会进入waitForActorsToStop最长 120 秒、每 100ms 轮询而 Pegboard 侧的 Runner Workflow 则在超过runner_lost_threshold后退出 drain 循环标记 Actor 为 Lost 并发送ToClientClose。服务端实现Runner Workflow 的信号处理Runner Workflow 位于 runner.rs其主循环通过listen_with_timeout::Main(runner_lost_threshold)监听信号其中runner_lost_threshold直接来自 Pegboard 配置见 pegboard.rs默认值15_000即 15 秒。ToServerStopping进入 draining 状态主循环中对该协议消息的处理只有三行见 runner.rsprotocol::ToServer::ToServerStopping { handle_stopping(ctx, input, state, false).await?; }handle_stoppingrunner.rs是两种停机路径的共同入口源码注释精确对应了文档中sets itself as draining的描述if !state.draining { // The workflow will enter a draining state where it can still process signals if // needed. After the runner lost threshold it will exit this loop and stop. state.draining true; ctx.activity(ClearDbInput { runner_id: input.runner_id, ... update_state: RunnerState::Draining, // 数据库中标记为 Draining }) .await?; let actors ctx .activity(FetchRemainingActorsInput { runner_id: input.runner_id }) .await?; // Set all remaining actors as going away immediately for (actor_id, generation) in actors { ctx.signal(crate::workflows::actor::GoingAway { generation: *generation, reset_rescheduling: reset_actor_rescheduling, }) .to_workflow::crate::workflows::actor::Workflow() .tag(actor_id, actor_id) .graceful_not_found() .send() .await?; } }这里有三个值得注意的实现细节幂等保护if !state.draining保证即使ToServerStopping和Stop信号先后到达例如 Runner 先发停止消息随后控制面又下发驱逐draining 副作用也只执行一次。数据库状态更新ClearDbInput { update_state: RunnerState::Draining }会同时把 Runner 从分配索引allocation indexes中移除并写入drain_ts这正是文档所说的preventing future actor allocations to it的底层实现——分配器后续查不到该 Runner。FetchRemainingActorsGoingAway广播逐个查询该 Runner 上仍未停止的 Actor 并向其 Workflow 发送GoingAway { generation, reset_rescheduling }。reset_rescheduling参数在两条路径下有区别自发起路径传falseRivet 发起路径传sig.reset_actor_rescheduling由Stop信号的载荷决定见下文。Rivet 发起停机Stop 信号Stop信号的处理runner.rsSome(Main::Stop(sig)) { handle_stopping(ctx, input, state, sig.reset_actor_rescheduling).await?; }也就是说 Rivet 发起停机走的是与ToServerStopping完全相同的handle_stopping函数唯一差别是跳过了ToServerStopping网络消息、并且可以携带reset_actor_rescheduling标志。这也解释了为什么文档中两条路径的步骤列表从发送 GoingAway开始重合。draining 期间的分配竞争处理仅从从分配索引中移除不足以杜绝所有竞态一个 Actor 分配请求可能在 Runner 进入 draining 之前已被入队。主循环对CommandStartActor有专门防线runner.rs// If draining, ignore start actor command and inform the actor wf that it is lost if let ( protocol::Command::CommandStartActor(protocol::CommandStartActor { actor_id, generation, .. }), true, // state.draining ) (command.inner, state.draining) { tracing::warn!(?actor_id, attempt to schedule actor to draining runner, reallocating); ctx.signal(crate::workflows::actor::Lost { generation: *generation, // Because this is a race condition, we want the actor to reschedule // regardless of its crash policy force_reschedule: true, reset_rescheduling: true, ... }) ... }这段代码体现了一个关键语义对 draining Runner 的CommandStartActor不下发而是直接给该 Actor 发Lost { force_reschedule: true, ... }——源码注释明确说明因为是竞态条件无论 Actor 的 crash policy 如何都强制重调度。这保证了停机过程中没有任何 Actor 会丢失只是被重新调度到健康的 Runner。runner lost threshold退出 drain 循环的计时器文档第 5 步Once the runner lost threshold is passed对应的实现是主循环的超时分支runner.rsNone { let expired ctx.activity(CheckExpiredInput { runner_id: input.runner_id }).await?; if state.draining || expired { return Ok(Loop::Break(())); } }CheckExpiredactivityrunner.rs用last_ping_ts now - runner_lost_threshold判断 Runner 是否失联。也就是说一旦进入 draining下一次主循环超时检查即触发Loop::BreakRunner Workflow 退出主循环。这里的listen_with_timeout(runner_lost_threshold)同时充当等待 Actor 收尾的等待器与Runner 失联的看门狗默认 15 秒。停机收尾标记剩余 Actor 为 Lost 并发送 ToClientClose主循环退出后Runner Workflow 依次执行三件事runner.rs与文档中ToClientClose → 关闭连接步骤对应清理数据库ClearDbInput { update_state: RunnerState::Stopped }将 Runner 标记为 Stopped 并从活跃索引中移除。将剩余 Actor 标记为 Lostlet actors ctx.activity(FetchRemainingActorsInput { runner_id: input.runner_id }).await?; // Set all remaining actors as lost for (actor_id, generation) in actors { ctx.signal(crate::workflows::actor::Lost { generation, force_reschedule: false, reset_rescheduling: false, ... }).to_workflow::crate::workflows::actor::Workflow() .graceful_not_found() .send() .await?; }注意此处的Lost信号参数与 draining 竞态下的force_reschedule: true不同这里是force_reschedule: false即是否重新调度遵循 Actor 自身的 crash policy——因为 Runner 是被有意停机的Actor 应按其声明的故障策略处理可推断该策略支持不再重启的语义避免停机演练变成无限重调度风暴。 3.发送 ToClientCloseSendMessageToRunnerInput { message: protocol::ToRunner::ToClientClose }。源码注释标注 Close websocket connection (its unlikely to be open)——在自发起路径中Runner SDK 通常已主动关闭连接此消息更多是兜底而在 Rivet 发起路径中它正是让 Runner 端关闭且不要重连的那条指令对应文档步骤 4 informing it not to attempt reconnection。ToClientClose本身定义在 Runner 协议 schema 中见 runner-protocol v7.bare并有跨版本转换器如 v8_to_v7.rs保证新旧版本 Runner/引擎之间该消息的兼容转发。Runner SDK 侧优雅停机与 120 秒等待从 Runner 视角自发起停机的完整实现在 TypeScript SDK 的 mod.ts 中。核心是shutdown(immediate, exit)方法mod.ts其流程与文档/时序图逐条对应用#shutdown布尔量防止并发重复 shutdown若 WebSocket 仍处于CONNECTING状态无法发送停止消息直接关闭先调用stopActors让所有 Actor 优雅停止再发送ToServerStoppingsending stopping message最后以pegboardWebSocket.close(1000, pegboard.runner_shutdown)关闭连接。其中等待 Actor 停止由waitForActorsToStop实现mod.ts三个硬性边界值得注意const shutdownTimeout 120_000; // 总超时 120 秒 const shutdownCheckInterval 100; // 每 100ms 轮询一次所有 Actor 已停止 → 继续关闭流程WebSocket 已关闭说明 Pegboard 侧已主动断连→ 强制继续达到 120 秒超时 → 强制继续源码注释 shutdown timeout reached, forcing close。此外SDK 内部还有一条独立的runner lost路径当 Runner 与 Pegboard 断连且超过runner_lost_threshold时会执行 stopping all actors due to runner lost thresholdmod.ts把 Actor 强制停止。这说明阈值概念是双向的Pegboard 用CheckExpired判定 Runner 失联Runner 用同一阈值判定 Pegboard 失联两侧都以 15 秒为默认基准。另一个 SDK 细节是 Actor 级别的stop方法mod.ts中的注释If a drain (GoingAway) occurs after this is called but before [the actor is removed]...——即 Actor 停止过程中若恰好收到GoingAwaySDK 会保证停止动作幂等这与 Pegboard 侧对GoingAway/Lost信号统一使用graceful_not_found()的容错信号到达但 Actor Workflow 已不存在时仅打 warning 日志形成呼应。新旧两代 Workflow 的差异仓库中同时存在runner.rslegacy与 runner2.rs 两代 Runner Workflow。runner2.rs中的handle_stopping逻辑与 legacy 版本一致同样设置state.draining、遍历FetchRemainingActors发送GoingAway、由CheckExpiredstate.draining决定退出循环但增加了 legacy 版本缺失的drain_on_version_upgrade能力——runner.rs中有明确注释This intentionally does not implement drain_on_version_upgrade like runner2.rs since this workflow is legacyrunner.rs。从源码结构看runner_lost_threshold超时同时被用作listen_with_timeout的监听间隔这一超时兼等待器的设计在两代实现中均保持。关键参数与可验证依据汇总参数 / 信号默认值 / 语义源码依据runner_lost_threshold默认 15 秒15_000ms可配置为Optioni64pegboard.rsToServerStoppingRunner → Runner WS → Runner WF触发 drainingrunner.rsStop信号控制面直接下发可携带reset_actor_reschedulingrunner.rsGoingAway广播给所有剩余 Actor WF开始优雅停止runner.rsLostdraining 竞态force_reschedule: true无视 crash policy 强制重调度runner.rsLost停机收尾force_reschedule: false遵循 Actor crash policyrunner.rsSDK 停机总超时120 秒每 100ms 轮询一次 Actor 状态mod.tsWS 关闭code 1000reasonpegboard.runner_shutdown不重连mod.ts小结Rivet Actors 的 Runner 停机设计可以用三条原则概括入口收敛无论是 Runner 自发起ToServerStopping还是 Rivet 发起Stop信号最终都汇入同一个handle_stopping保证 draining、GoingAway广播、索引清理等行为完全一致且幂等draining 是核心状态进入 draining 后 Runner 立即从分配索引移除但 Workflow 本身继续存活并处理信号对迟到分配请求的CommandStartActor一律转为Lost { force_reschedule: true }确保停机期间零 Actor 丢失超时是最终兜底runner_lost_threshold默认 15 秒同时承担等待 Actor 收尾和判定 Runner 失联双重角色到期后强制退出 drain 循环、把剩余 Actor 按 crash policy 标记为Lost、发送ToClientClose并告知 Runner 端不要重连——即使 Actor 永远无法优雅停止整个停机流程也有确定的终止点。理解这套机制后你可以在自托管部署中通过runner_lost_threshold调整停机容忍度在编写自定义 Runner/SDK 集成时正确实现ToServerStopping发送时机与 120 秒 Actor 等待窗口并在调试 Actor 频繁重调度问题时定位到 draining 竞态这条特定路径。【免费下载链接】actorsRivet Actors are the primitive for stateful workloads. Built for AI agents, collaborative apps, and durable execution.项目地址: https://gitcode.com/GitHub_Trending/riv/actors创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考