- 即时通讯
- 后端
【免费下载链接】WuKongIM
More than just IM 不只是即时通讯(IM)
WuKongIM 的动态节点缩容（scale-in）流程中，Stage 5C 负责为离开中的节点提供 Gateway 排水（drain）写路径：Manager 可以把目标节点的网关置为 drain 模式、关闭新会话准入，并让scale-in status在运行时排水计数未清零前保持 fail closed（SafeToRemove=false）。读完本文，你将掌握该功能的完整实现链路——从 usecase 端口、节点间 manager connection RPC、本地网关排水写入器，到 Manager HTTP 路由与SafeToRemove最终门禁——并能对照源码逐层验证每个设计决策。
一、目标、架构前提与实现边界
Stage 5C 的原始计划文档位于 2026-06-24-internalv2-dynamic-node-stage5c-gateway-drain-mode.md，其 Goal 一句话概括：让 Manager 能把离开节点的 gateway 置入 drain mode，并让 scale-in status 在运行时排水计数未清零时 fail closed。
该子阶段建立在 Stage 4/5B 的基础之上：
- Stage 4 已经提供
NodeRuntimeSummary，其中包含 gateway 会话数、在线数、AcceptingNewSessions与Draining字段； - Stage 5B 已完成 Channel 排水清单（drain inventory），使 scale-in status 能够识别 Channel leader/replica/ISR 阻塞。
Stage 5C 在此基础上补上三块拼图：
- 管理写路径：为 drain 模式增加 usecase 写端口（只读 summary 已存在，但“写入 drain 状态”是新的）；
- 远程写转发：把远端节点的 drain 写操作路由到既有的 manager connection RPC 服务；
- 状态门禁合并：把运行时排水阻塞项并入 scale-in status，并新增
SafeToRemove最终门禁。
计划文档明确划定了边界：本子阶段不得调用MarkNodeRemoved，也不得新增/scale-in/remove路由（移除路由属于后续 Stage 5D）。这一点在当前源码中同样成立：internal/usecase/management/scale_in.go中的MarkNodeRemoved是独立门禁，且必须以SafeToRemove==true为前置（见 scale_in.go#L236-L265）。
技术栈方面，文档标注为：Go、internal 管理 usecase、manager HTTP 路由、节点 RPC 二进制编解码（manager connection codec）、clusterv2 RPC 转发、pkg/gateway准入控制。
说明：计划文档成文早于仓库目录结构调整，文中文件路径写作
internalv2/...；当前仓库中对应实现位于internal/目录下。下文同时给出两者，路径以当前仓库实际布局为准。
入口门禁（Entry Gate）
计划要求开工前确认两个前置条件：Stage 5B 已合入本地main，且 Channel 排水状态测试通过：
GOWORK=off go test ./internalv2/usecase/management -run 'TestNodeChannelDrainInventory|TestScaleInStatus' -count=1当前仓库对应命令（目录更名后）为：
GOWORK=off go test ./internal/usecase/management -run 'TestNodeChannelDrainInventory|TestScaleInStatus' -count=1二、文件结构总览：一个功能横跨五层
计划文档列出的文件清单本身就体现了 WuKongIM 的层次化架构——一个管理功能被拆成 usecase、节点 RPC 编解码、集群转发、app 装配、HTTP 接入五层，每层都有独立测试：
| 层 | 文件（当前仓库路径） | 职责 |
|---|---|---|
| usecase | internal/usecase/management/gateway_drain.go | GatewayDrainWriter端口、SetNodeDrainMode请求/响应 |
| usecase | internal/usecase/management/nodes.go | NodeRuntimeSummary新增PendingActivations |
| usecase | internal/usecase/management/scale_in.go | 运行时排水阻塞项与SafeToRemove |
| 节点 RPC | internal/access/node/manager_connection_codec.go | set_drain_mode操作与Draining请求字段 |
| 节点 RPC | internal/access/node/manager_connection_rpc.go | 服务端 dispatch 与客户端调用 |
| 集群转发 | internal/infra/cluster/management_connections.go | ManagementConnectionReader.SetNodeDrainMode集群路由 |
| app 装配 | internal/app/runtime_summary.go | 本地managementGatewayDrainWriter与运行时计数 |
| HTTP 接入 | internal/access/manager/scale_in.go | drain 路由与 DTO |
配套测试包括 gateway_drain_test.go、scale_in_test.go、manager_connection_codec_test.go、manager_connection_rpc_test.go、scale_in_test.go 等。
三、Task 1：Gateway Drain Usecase 与运行时摘要扩展
3.1 新增运行时计数PendingActivations
计划首先在NodeRuntimeSummary中增加一个字段：
// PendingActivations counts local sessions accepted but not yet authority-active. PendingActivations int这个字段在最终实现中保留了下来，定义见 nodes.go#L527-L555：NodeRuntimeSummary包含ActiveOnline、ClosingOnline、TotalOnline、GatewaySessions、PendingActivations、AcceptingNewSessions、Draining、Unknown等字段。语义上，PendingActivations统计“本地已接受但尚未 authority-active”的会话——这类会话对缩容同样是排水阻碍，因为它们还没有完成权限侧的激活，既不能简单归入ActiveOnline，也不能被忽略。
3.2GatewayDrainWriter端口与SetNodeDrainMode
计划要求新建gateway_drain.go，定义写端口：
// GatewayDrainWriter toggles gateway admission for a target node and returns fresh runtime counters. type GatewayDrainWriter interface { SetNodeDrainMode(context.Context, uint64, bool) (NodeRuntimeSummary, error) }以及请求/响应结构。当前实现位于 gateway_drain.go，响应字段与计划完全一致：
type SetNodeDrainModeResponse struct { NodeID uint64 Draining bool AcceptingNewSessions bool GatewaySessions int ActiveOnline int ClosingOnline int TotalOnline int PendingActivations int Unknown bool }App.SetNodeDrainMode的处理流程（gateway_drain.go#L46-L77）按顺序执行：
ctxErr(ctx)检查上下文；NodeID == 0返回metadb.ErrInvalidArgument；- writer 未装配时返回
ErrNodeScaleInUnavailable（fail closed，而不是静默忽略写请求）； requireDrainModeTarget生命周期校验：读本地 control 快照，确认目标节点存在（否则ErrNodeLifecycleNotFound）、且不携带 Controller 角色、必须携带 Data 角色、且JoinState必须为leaving（否则ErrNodeLifecycleConflict），见 gateway_drain.go#L79-L95；- 委托给
GatewayDrainWriter，把返回的NodeRuntimeSummary映射为响应，summary.NodeID == 0时回填请求中的NodeID。
其中第 4 步是计划之外的增量强化：drain 写只允许作用于“正在离开的 Data 节点”，从 usecase 层面就杜绝了对 active 节点或 Controller 节点误关准入的可能。
3.3 计划中的单元测试设计
计划以 TDD 组织：先写失败测试（RED）再实现（GREEN）。计划的测试意图在当前 gateway_drain_test.go 中完整落地并有所扩展：
TestSetNodeDrainModeDelegatesToWriter：验证请求被正确委托给 writer，且响应字段（Draining、AcceptingNewSessions、GatewaySessions、ActiveOnline、PendingActivations）逐项映射正确；TestSetNodeDrainModeRejectsInvalidInputAndMissingWriter：NodeID=0必须得到ErrInvalidArgument，writer 缺失必须得到ErrNodeScaleInUnavailable；TestSetNodeDrainModeRequiresLeavingDataNode：active data 节点、controller 节点、无角色的 leaving 节点都必须得到ErrNodeLifecycleConflict，且 writer 不得被调用；TestSetNodeDrainModeRejectsMissingTargetNode：目标节点不存在时返回ErrNodeLifecycleNotFound。
验证命令：
GOWORK=off go test ./internalv2/usecase/management -run TestSetNodeDrainMode -count=1四、Task 2：Manager Connection RPC 增加 drain 操作
节点间的 manager 操作复用既有的 manager connection RPC 通道。Stage 5C 在其中新增第四个操作set_drain_mode，实现位于 manager_connection_codec.go：
managerConnectionOpList = "list_connections" managerConnectionOpGet = "get_connection" managerConnectionOpRuntimeSummary = "runtime_summary" managerConnectionOpSetDrainMode = "set_drain_mode"每个操作有稳定的数字 ID（iota + 1），并追加到请求结构（manager_connection_codec.go#L38-L46）：
type managerConnectionRPCRequest struct { Version byte Op string NodeID uint64 SessionID uint64 Limit int Cursor managementusecase.ConnectionListCursor Draining bool }二进制编码遵循“先旧字段、后新字段”的兼容顺序：Draining作为布尔字节紧跟在Limit之后编码（manager_connection_codec.go#L59-L92），codec 支持 V2/V3/V4 三个版本、以 magic 头（如W K V M 4）区分版本，保证滚动升级期间新旧版本节点可以互操作。
服务端 dispatch 在 manager_connection_rpc.go#L50-L54：HandleManagerConnectionRPC收到set_drain_mode后调用a.managerConnections.SetNodeDrainMode(ctx, SetNodeDrainModeRequest{NodeID, Draining})，把响应摘要回填进Summary字段；客户端方法SetManagerDrainMode（manager_connection_rpc.go#L125-L137）则通过callManagerConnection把请求定向到 owner 节点。
集群路由层 management_connections.go#L56-L62 提供了与计划一致的适配器：
// SetNodeDrainMode toggles gateway drain mode on the selected node through manager connection RPC. func (r *ManagementConnectionReader) SetNodeDrainMode(ctx context.Context, nodeID uint64, draining bool) (managementusecase.NodeRuntimeSummary, error) { ... return r.remote.SetManagerDrainMode(ctx, nodeID, draining) }验证命令（计划原文，目录更名后把internalv2换成internal即可）：
GOWORK=off go test ./internalv2/access/node -run 'TestManagerConnection.*Drain|TestManagerConnectionRuntimeSummaryCodecCarriesControlRevision' -count=1 GOWORK=off go test ./internalv2/infra/cluster -run 'TestManagementConnectionReader.*Drain|TestManagementConnectionReaderRoutesRuntimeSummary' -count=1对应测试 manager_connection_codec_test.go 验证set_drain_mode请求的编码/解码 round-trip；manager_connection_rpc_test.go 验证远端 drain RPC 确实调用了服务端的SetNodeDrainMode。
五、Task 3：本地网关排水写入器与 Manager HTTP 路由
5.1 本地/远端分流与 fail closed
app 层的managementGatewayDrainWriter（runtime_summary.go#L30-L132）实现了“本地直接写、远端转发、缺失即失败”的路由策略：
func (w managementGatewayDrainWriter) SetNodeDrainMode(ctx context.Context, nodeID uint64, draining bool) (managementusecase.NodeRuntimeSummary, error) { if nodeID == w.localNodeID || w.localNodeID == 0 { return w.setLocalDrainMode(ctx, nodeID, draining) } if w.remote == nil { return managementusecase.NodeRuntimeSummary{NodeID: nodeID, Unknown: true}, managementusecase.ErrNodeScaleInUnavailable } return w.remote.SetNodeDrainMode(ctx, nodeID, draining) }本地路径的核心是向网关下发准入开关并回读一份新鲜摘要：
gatewayRuntime, ok := w.app.gateway.(gatewayDrainRuntime) // interface { SetAcceptingNewSessions(bool) } if !ok || gatewayRuntime == nil { return managementusecase.NodeRuntimeSummary{NodeID: nodeID, Unknown: true}, managementusecase.ErrNodeScaleInUnavailable } gatewayRuntime.SetAcceptingNewSessions(!draining) return managementRuntimeSummaryReader{app: w.app, localNodeID: w.localNodeID}.localRuntimeSummary(ctx, nodeID), nil计划文档中专门强调了 fail closed 语义：远端目标节点若无 remote drain writer，不得“落回”到本地网关——即请求了 node 4 却只配置了本地 node 1 时，绝不能误关 node 1 的准入。实现用setLocalDrainMode中的防御式检查保证：
if w.localNodeID != 0 && nodeID != w.localNodeID { return managementusecase.NodeRuntimeSummary{NodeID: nodeID, Unknown: true}, managementusecase.ErrNodeScaleInUnavailable }计划中对应的测试TestManagementGatewayDrainWriterFailsClosedWhenRemoteMissing正是用 stub 断言remote drain unexpectedly changed local gateway admission不会发生。
5.2 网关准入的底层实现
SetAcceptingNewSessions最终落在pkg/gateway。网关服务器用一个原子布尔量保存准入开关（pkg/gateway/core/server.go#L1458-L1489）：
// SetAcceptingNewSessions toggles gateway admission for newly opened connections. func (s *Server) SetAcceptingNewSessions(accepting bool) { if s == nil { return } s.accepting.Store(accepting) }由此可以推断：drain 模式是只影响新连接准入的软开关——已建立的存量会话不会被 drain 主动断开，这正符合“排水”语义：先关门、等存量排空、再摘除节点。pkg/gateway侧另有 gateway_lifecycle_contract_test.go 等测试验证SetAcceptingNewSessions(false)确实关闭准入。
5.3 运行时计数的来源
PendingActivations等计数由managementRuntimeSummaryReader.localRuntimeSummary汇总（runtime_summary.go#L56-L94）：
ActiveOnline/PendingActivations/TotalOnline来自在线状态机快照：onlineSummary.Active、onlineSummary.Pending、二者之和；GatewaySessions、AcceptingNewSessions来自网关SessionSummary()，并且Draining被定义为!AcceptingNewSessions；- 任一依赖缺失时返回
Unknown: true的摘要——这正是后续 status 门禁 fail closed 的数据基础。
5.4 Manager HTTP 路由
计划要求注册POST /manager/nodes/:node_id/scale-in/drain，请求体{"draining":true}，要求cluster.node写权限。当前实现：
路由注册在 server.go#L482：
nodeWrites.POST("/nodes/:node_id/scale-in/drain", s.handleNodeScaleInDrain)handler 逻辑见 scale_in.go#L247-L266：解析路径参数node_id→ 绑定 JSON body → 调用s.management.SetNodeDrainMode→ 复用 scale-in 统一的错误映射（writeNodeScaleInError处理 unavailable / invalid argument 等错误）。
DTO 与计划中的字段一一对应（scale_in.go#L79-L105）：
type ManagerNodeScaleInDrainRequest struct { // Draining disables new gateway admission when true. Draining bool `json:"draining"` } type ManagerNodeScaleInDrainResponse struct { NodeID uint64 `json:"node_id"` Draining bool `json:"draining"` AcceptingNewSessions bool `json:"accepting_new_sessions"` GatewaySessions int `json:"gateway_sessions"` ActiveOnline int `json:"active_online"` ClosingOnline int `json:"closing_online"` TotalOnline int `json:"total_online"` PendingActivations int `json:"pending_activations"` Unknown bool `json:"unknown"` }权限行为由 scale_in_test.go 验证：只读 token 访问/manager/nodes/4/scale-in/drain得到 403 且请求不得穿透到 management 层；具备节点写权限的 token 得到 200，且 management 收到的请求恰为{NodeID: 4, Draining: true}。
六、Task 4：把运行时排水阻塞并入 scale-in status
6.1SafeToProceed与SafeToRemove的双门禁设计
计划的核心设计决策是：在NodeScaleInStatusResponse中区分两个布尔量。源码注释把职责写得很清楚（scale_in.go#L86-L89）：
SafeToProceed：没有 Slot、task、Channel 或 control 层面的阻塞——它是排水推进过程的进度门禁，决定PlanNodeScaleIn/AdvanceNodeScaleIn是否还能继续创建 Slot 搬迁任务；SafeToRemove：在SafeToProceed的基础上，再加上运行时排水计数全部清零——它是 Manager 后续执行 remove 路由前的最终门禁。
最终合成逻辑见 scale_in.go#L311-L331：
response.SafeToProceed = !response.BlockedByMissingNode && !response.BlockedByJoinState && !response.BlockedByControlRevision && !response.BlockedByHealth && !response.BlockedByStaleRevision && !response.BlockedByControllerRole && !response.BlockedByDataRole && !response.BlockedBySlots && !response.BlockedBySlotLeadership && !response.BlockedBySlotRuntime && !response.BlockedByTasks && !response.BlockedByChannels && !response.UnknownRuntime && !response.UnknownChannelInventory && response.ChannelLeaderCount == 0 && response.ChannelReplicaCount == 0 && response.ChannelISRCount == 0 response.SafeToRemove = response.SafeToProceed && !response.BlockedByRuntimeDrain && !response.RuntimeUnknown6.2 运行时排水阻塞的判定
markScaleInRuntimeDrainBlockers（scale_in.go#L669-L691）是 Stage 5C 新增的阻塞源：
func (a *App) markScaleInRuntimeDrainBlockers(ctx context.Context, targetNode uint64, response *NodeScaleInStatusResponse) { summary := a.nodeRuntimeSummary(ctx, targetNode) if summary.Unknown { response.UnknownRuntime = true response.RuntimeUnknown = true response.BlockedByRuntimeDrain = true return } response.GatewayDraining = summary.Draining ... response.BlockedByRuntimeDrain = !summary.Draining || summary.AcceptingNewSessions || summary.GatewaySessions > 0 || summary.ActiveOnline > 0 || summary.ClosingOnline > 0 || summary.TotalOnline > 0 || summary.PendingActivations > 0 }对照计划列出的SafeToRemove要求，可以实现与文档逐条对应：
计划要求的SafeToRemove前置条件 | 源码落点 |
|---|---|
Stage 5BSafeToProceed == true | response.SafeToRemove = response.SafeToProceed && ... |
| Channel inventory 已知且清空 | SafeToProceed中的UnknownChannelInventory、三个 Channel 计数 |
| 目标运行时摘要已知 | summary.Unknown→RuntimeUnknown，直接挡掉SafeToRemove |
Draining == true | !summary.Draining触发阻塞 |
AcceptingNewSessions == false | summary.AcceptingNewSessions触发阻塞 |
GatewaySessions == 0 | summary.GatewaySessions > 0触发阻塞 |
ActiveOnline == 0 | summary.ActiveOnline > 0触发阻塞 |
ClosingOnline == 0 | summary.ClosingOnline > 0触发阻塞 |
PendingActivations == 0 | summary.PendingActivations > 0触发阻塞 |
注意一个比计划更严格的实现细节：TotalOnline也必须为 0才会解除阻塞。由于localRuntimeSummary中TotalOnline = Active + Pending，这条与ActiveOnline == 0 && PendingActivations == 0在数值上等价，属于防御性冗余判断。
6.3 计划中的两组关键测试
计划给出的两个测试场景在当前测试中均成立：
TestScaleInStatusBlocksWhenRuntimeDrainIsNotEmpty：目标节点Draining:true, AcceptingNewSessions:false但GatewaySessions:1，断言SafeToRemove==false且BlockedByRuntimeDrain==true——即“已关准入但还有存量会话”时仍然不可移除；TestScaleInStatusSafeToRemoveAfterChannelAndRuntimeDrainClear：四个节点中目标节点处于 leaving、Draining:true且所有计数清零，其余节点健康，断言SafeToRemove==true。
七、端到端验证：从计划到 e2e
7.1 计划中的 Exit Gate
计划文档定义的出口门禁命令（此处保留计划原文的internalv2路径，当前仓库执行时替换为internal）：
GOWORK=off go test ./internalv2/usecase/management ./internalv2/access/node ./internalv2/infra/cluster ./internalv2/access/manager ./internalv2/app -count=1 git diff --check rg -n "SetNodeDrainMode|set_drain_mode|PendingActivations|SafeToRemove|BlockedByRuntimeDrain|scale-in/drain" internalv2期望：全部通过，且SafeToRemove只在 Slot、Channel、runtime drain 三类阻塞项全部清零后才为 true。
7.2 三层测试证据链
usecase 层：gateway_drain_test.go 覆盖委托、校验、生命周期冲突；scale_in_test.go 覆盖排水阻塞与SafeToRemove翻转。
HTTP 层：internal/access/manager/scale_in_test.go 用真实 HTTP 请求验证路由解析、权限拒绝（403 且请求不穿透）、以及 200 时{NodeID: 4, Draining: true}的精确传递与完整 DTO 映射。
e2e 层：test/e2e/cluster/dynamic_node_join/node_remove_drain_test.go 把 Stage 5C 放在真实三节点集群中验证完整链路——第 4 号节点 join、activate、MustStartScaleIn进入leaving后执行MustSetScaleInDrain(t, 4, true)，然后轮询等待SafeToRemove：
drain := manager.MustSetScaleInDrain(t, 4, true) require.True(t, drain.Draining) require.False(t, drain.AcceptingNewSessions) status := manager.EventuallyScaleInSafeToRemove(t, 4, 30*time.Second) require.True(t, status.SafeToRemove) require.False(t, status.BlockedByRuntimeDrain) require.True(t, status.GatewayDraining) require.Zero(t, status.GatewaySessions) require.Zero(t, status.ActiveOnline) require.Zero(t, status.ClosingOnline) require.Zero(t, status.PendingActivations)在故障注入侧，test/e2e/cluster/dynamic_node_faults/下的scale_in_fail_closed_fault_test.go等测试进一步验证故障发生时该流程不会越门（例如排水未清就尝试 remove 会被ErrNodeScaleInUnsafe挡下，见 scale_in.go#L260-L263）。
八、设计要点回顾
- 写路径与读路径分离但共享摘要模型：Stage 4 的只读
NodeRuntimeSummary被 Stage 5C 的写操作直接复用为响应体，保证“改完即回读新鲜计数”，避免客户端二次拉取时状态漂移。 - 全程 fail closed：writer 未装配、远端通道缺失、目标摘要
Unknown，任何一环缺失都返回错误或阻塞位，而不是默认放行；SafeToRemove的推导把RuntimeUnknown也视为阻塞。 - 双层门禁各司其职：
SafeToProceed管“还能不能继续排水”，SafeToRemove管“能不能执行最终移除”，二者在代码注释与测试中都被要求显式区分，避免后续 Stage 5D 的 remove 路由误用进度门禁。 - codec 演进保持兼容：
Draining字段按固定字节序追加在既有字段之后，magic 版本号区分新旧报文，滚动升级中不同版本节点可以互通。 - drain 是软关门：
SetAcceptingNewSessions(false)只原子地翻转新连接准入标志，不主动断开存量会话；“排水”依赖存量自然收敛，status 门禁负责等待收敛完成。
对希望继续深入的读者，建议按以下顺序阅读：Stage 5 索引 2026-06-24-internalv2-dynamic-node-stage5.md、前序子阶段 Stage 5B Channel Drain Inventory、后续 Stage 5D Safe Remove Route，以及本文引用的internal/usecase/management、internal/access/node、internal/access/manager与internal/app源码路径。
- 即时通讯
- 后端
【免费下载链接】WuKongIM
More than just IM 不只是即时通讯(IM)
相关推荐
WuKongIM 动态节点生命周期 Stage 1：持久化生命周期读取模型与 Seed-Join 配置基础实现计划解析
WuKongIM 动态节点生命周期 Stage 1：持久化生命周期读取模型与 Seed Join 配置基础实现计划解析 本文基于 WuKongIM 仓库中 in
即时通讯后端WuKongIM 动态节点生命周期 Stage 4：MarkNodeLeaving、Fail-Closed 安全报告与有界 Slot 缩容排空实战
WuKongIM 动态节点生命周期 Stage 4：MarkNodeLeaving、Fail Closed 安全报告与有界 Slot 缩容排空实战 本篇基于 W
即时通讯后端WuKongIM 动态数据节点加入与激活：Stage 2 以 ControllerV2 为唯一持久成员源的生命周期写入路径
WuKongIM 动态数据节点加入与激活：Stage 2 以 ControllerV2 为唯一持久成员源的生命周期写入路径 本文围绕 WuKongIM 的 in
即时通讯后端
创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
