WuKongIM 节点缩容 Stage 5B：用有界 ChannelRuntimeMeta 盘点实现 fail-closed 的 Channel 阻塞判断

• 即时通讯
• 后端

本文围绕 WuKongIM 动态节点生命周期中的 Stage 5B 子阶段展开：如何为缩容（scale-in）状态机补充一份有界（bounded）的ChannelRuntimeMeta盘点（inventory），使系统在移除一个正在离开（leaving）的节点前，能够证明该节点不是任何 Channel 的 leader、replica 或 ISR 成员，并在任何证据缺失时 fail-closed 拒绝移除。读完后你会掌握 WuKongIM 缩放安全判定中 Channel 阻塞字段的完整语义、分页游标扫描的实现约束，以及SafeToProceed计算方式的变化。

1. Stage 5B 在动态节点生命周期中的位置

WuKongIM 的 dynamic node lifecycle 被拆分为多个可独立评审、独立合入的小阶段。Stage 5 的目标是把 Stage 4 已经具备的缩容准备能力，推进为一个最终“可安全移除”（safe-to-remove）的闸门。其阶段索引见 Stage 5 Plan Index，完整的动态节点生命周期规划见 Dynamic Node Lifecycle Plan。

Stage 5 按顺序拆为五个子阶段：

Stage 5B（本文主题）的实施计划原文见 Stage 5B Channel Drain Inventory 计划，其前置子阶段 5A 的计划为 Stage 5A Removed Lifecycle 计划。Stage 5 索引文档还明确了跨子阶段的约束，其中与 5B 直接相关的有：

• removed是墓碑（tombstone）状态，Stage 5 期间不物理删除节点身份；
• 当 Channel inventory、runtime summary、gateway drain 状态、Slot 状态、任务状态或控制面 revision 数据未知时，不得报告safe_to_remove=true；
• SafeToProceed与SafeToRemove必须保持分离：前者表示 leaving 节点没有已知的 Slot/任务/Channel 阻塞；后者额外要求 gateway drain 模式和运行时计数归零；
• 扫描必须保持有界：Channel inventory 扫描是显式的状态/移除安全检查，不是节点列表接口的默认工作。

1.1 本阶段的范围边界

计划文档在 Scope 中明确划定了 5B不做什么：不新增 gateway drain 模式、不新增安全移除（safe remove）路由、不在 usecase 层接线MarkNodeRemoved。这些分别由 Stage 5C/5D 完成。5B 只做一件事：为缩容状态提供一个“该节点是否仍承担 Channel 角色”的只读盘点能力，并把结果合并进缩容状态作为 fail-closed 阻塞项。

这种切分方式的价值在于：每个子阶段都是可独立验证、可独立合入的。Channel 盘点本身不改变任何写路径，它只读取权威元数据并产出计数，因此可以在不触碰移除写路径的前提下先行落地并建立测试基线。

1.2 入口门禁

计划的 Entry Gate 要求：Stage 5A 已合入本地main，且 Stage 4 的缩容状态测试仍然通过：

GOWORK=off go test ./internalv2/usecase/management -run 'TestScaleInStatus|TestAdvanceNodeScaleIn' -count=1

期望结果为 PASS。需要说明：计划编写时该 usecase 包位于internalv2/usecase/management；在当前仓库中，对应的实现与测试位于 internal/usecase/management，运行同类测试应使用当前路径。

2. 核心数据模型：ChannelRuntimeMeta 与分页游标

Stage 5B 的盘点不是引入新存储，而是复用既有的权威 Channel 运行时元数据。理解这一层是理解整个子阶段的前提。

2.1 元数据行结构

ChannelRuntimeMeta定义在 pkg/db/meta/table_runtime_meta.go，它保存 Channel 的权威路由与存活性元数据，与 5B 盘点直接相关的字段包括：

// ChannelRuntimeMeta stores authoritative channel routing and liveness metadata. type ChannelRuntimeMeta struct { ChannelID string ChannelType int64 ChannelEpoch uint64 LeaderEpoch uint64 RouteGeneration uint64 Replicas []uint64 ISR []uint64 Leader uint64 MinISR int64 Status uint8 Features uint64 LeaseUntilMS int64 // ... retention、write fence、directory generation 等字段 }

其中Leader（当前 leader 节点 ID）、Replicas（配置的副本集）、ISR（当前同步副本集）三个字段，正是 Channel drain inventory 要独立统计的三类角色。ChannelRuntimeMeta的主键是(ChannelID, ChannelType)，行归属于某个物理 Slot 的元数据存储，这决定了盘点必须“按物理 Slot 逐 Slot 扫描”。

2.2 分页游标

分页扫描的游标类型定义在同一文件：

// ChannelRuntimeMetaCursor is a page cursor for runtime metadata scans. type ChannelRuntimeMetaCursor struct { // ChannelID is the last returned channel ID. ChannelID string // ChannelType is the last returned channel type. ChannelType int64 }

游标只包含主键二元组，语义是“上一页最后一条记录”。这个结构有两个直接后果：其一，每页扫描的起点是确定性的，测试可以用游标作为 map key 精确模拟分页行为；其二，如果底层存储异常导致游标不前进，扫描会无限循环，因此盘点逻辑必须内置游标停滞检测（见第 4.3 节）。

2.3 读取接口：ChannelRuntimeMetaReader

management usecase 层通过一个最小接口读取权威元数据，定义在 internal/usecase/management/channel_runtime_meta.go：

// ChannelRuntimeMetaReader exposes authoritative channel runtime metadata scans. type ChannelRuntimeMetaReader interface { // ScanChannelRuntimeMetaSlotPage returns one runtime metadata page for a physical Slot. ScanChannelRuntimeMetaSlotPage(ctx context.Context, slotID uint32, after metadb.ChannelRuntimeMetaCursor, limit int) ([]metadb.ChannelRuntimeMeta, metadb.ChannelRuntimeMetaCursor, bool, error) }

返回值四元组为(本页条目, 下一页游标, 是否读完, 错误)。该接口的真实实现方包括 pkg/cluster/node_meta_runtime_scan.go 中集群节点的ScanChannelRuntimeMetaSlotPage方法，以及 pkg/slot/proxy/store.go 中 Slot 代理存储的权威页读取；其分页行为本身有专门的集成测试覆盖，例如 pkg/cluster/node_meta_channel_scan_integration_test.go 验证了逐条游标推进的完整分页。

这个接口通过构造选项注入 management 应用：internal/usecase/management/nodes.go 中Options结构体包含ChannelRuntimeMeta ChannelRuntimeMetaReader字段。接口注入的好处是测试可以用内存 fake 精确构造每一页数据，而不需要真实数据库——Stage 5B 的测试正是这样写的。

3. Task 1：Channel Drain Inventory Usecase

计划将实现拆为两个 Task。Task 1 是新增一个独立的盘点 usecase，对应文件结构：

• 新建channel_drain.go：实现目标节点 Channel drain inventory（当前仓库实际路径为 internal/usecase/management/channel_drain.go）；
• 新建channel_drain_test.go：验证分页、leader/replica/ISR 计数、reader 缺失与扫描错误（当前路径 internal/usecase/management/channel_drain_test.go）。

3.1 请求与响应契约

计划给出的核心类型定义如下（与当前仓库实现一致，当前实现在此基础上增加了MaxPages与ScannedPageCount，见第 5 节）：

const ( DefaultChannelDrainScanLimit = 256 MaxChannelDrainScanLimit = 1024 ) // NodeChannelDrainInventoryRequest configures a target-node Channel drain scan. type NodeChannelDrainInventoryRequest struct { // NodeID is the leaving node being checked. NodeID uint64 // PageLimit bounds each physical Slot metadata scan page. PageLimit int } // NodeChannelDrainInventoryResponse reports Channel blockers for target-node removal. type NodeChannelDrainInventoryResponse struct { NodeID uint64 Safe bool Unknown bool ScannedSlotCount int LeaderCount int ReplicaCount int ISRCount int LastError string }

字段语义需要注意两点：

• Safe与Unknown是两个独立维度。Unknown=true表示“无法证明”（reader 缺失、扫描出错、游标停滞等），此时Safe必须为false；Safe=true的唯一成立条件是 inventory 已知（Unknown=false）且三类计数全部为零；
• ScannedSlotCount记录结果产生前已扫描的物理 Slot 数，用于诊断“盘点进行到哪个 Slot 时中断”。

3.2 实现规则

计划为NodeChannelDrainInventory规定了完整的 fail-closed 行为规则，当前仓库的 channel_drain.go 逐条落实：

1. 非法NodeID（为 0）返回metadb.ErrInvalidArgument；
2. app 实例、cluster 或ChannelRuntimeMetaReader缺失时，返回Unknown=true, Safe=false，不报错——依赖缺失是“证据不足”，不是调用错误；
3. LocalControlSnapshot出错时把错误直接返回给调用方（读快照失败属于调用方需要感知的硬错误）；
4. Slot 扫描顺序为sortedSnapshotSlotIDs(snapshot.Slots)，即按 Slot ID 排序后逐个扫描，保证结果顺序确定、可复现；
5. 每页大小经normalizeChannelDrainLimit(req.PageLimit)归一化；
6. 扫描过程中出错时返回Unknown=true, Safe=false, LastError=err.Error()；
7. 对每条ChannelRuntimeMeta，目标节点作为 leader、replica、ISR 的身份独立计数（一个节点可以在同一个 Channel 上同时命中三类计数）；
8. Safe=true当且仅当 inventory 已知且三个计数全为 0。

核心扫描循环（当前仓库实现节选）：

func (a *App) nodeChannelDrainInventoryFromSnapshot(ctx context.Context, snapshot control.Snapshot, req NodeChannelDrainInventoryRequest) NodeChannelDrainInventoryResponse { resp := NodeChannelDrainInventoryResponse{NodeID: req.NodeID} if a == nil || a.channelRuntimeMeta == nil { resp.Unknown = true return resp } limit := normalizeChannelDrainLimit(req.PageLimit) for _, slotID := range sortedSnapshotSlotIDs(snapshot.Slots) { resp.ScannedSlotCount++ after := metadb.ChannelRuntimeMetaCursor{} for { page, nextCursor, done, err := a.channelRuntimeMeta.ScanChannelRuntimeMetaSlotPage(ctx, slotID, after, limit) if err != nil { resp.Unknown = true resp.Safe = false resp.LastError = err.Error() return resp } for _, meta := range page { countChannelDrainMeta(&resp, req.NodeID, meta) } if done { break } if nextCursor == after { resp.Unknown = true resp.Safe = false resp.LastError = errChannelDrainCursorStalled.Error() return resp } after = nextCursor } } resp.Safe = !resp.Unknown && resp.LeaderCount == 0 && resp.ReplicaCount == 0 && resp.ISRCount == 0 return resp }

3.3 页大小归一化与计数辅助

计划要求添加的两个辅助函数：

func normalizeChannelDrainLimit(limit int) int { if limit <= 0 { return DefaultChannelDrainScanLimit } if limit > MaxChannelDrainScanLimit { return MaxChannelDrainScanLimit } return limit } func countChannelDrainMeta(resp *NodeChannelDrainInventoryResponse, targetNode uint64, meta metadb.ChannelRuntimeMeta) { if meta.Leader == targetNode { resp.LeaderCount++ } if containsUint64(meta.Replicas, targetNode) { resp.ReplicaCount++ } if containsUint64(meta.ISR, targetNode) { resp.ISRCount++ } }

normalizeChannelDrainLimit把操作者传入的页大小钳制到[256, 1024]区间：非正值回落默认值 256，超大值压到上限 1024。这是“有界扫描”原则在参数层的落实——调用方无法通过传一个天文数字的 limit 让单页扫描拖垮存储。

countChannelDrainMeta的三分支互不排斥：leader 必然同时出现在Replicas/ISR中的情况并不少见，因此三类计数分别累计，缩容侧才能看到最完整的角色画像。

3.4 Task 1 的测试设计

计划采用 TDD：先写失败测试（RED），再实现（GREEN）。验证命令：

# RED：类型和方法尚不存在，测试必须失败 GOWORK=off go test ./internalv2/usecase/management -run 'TestNodeChannelDrainInventory' -count=1 # GREEN：实现完成后同一命令必须通过

测试文件用内存 fake 模拟 reader。fake 以map[slotID]map[cursor]page组织数据，可以精确控制“某个 Slot 在某个游标位置返回哪一页”，并记录调用次数：

type channelDrainMetaReader struct { pages map[uint32]map[metadb.ChannelRuntimeMetaCursor]channelDrainMetaPage err error calls int } func (r *channelDrainMetaReader) ScanChannelRuntimeMetaSlotPage(_ context.Context, slotID uint32, after metadb.ChannelRuntimeMetaCursor, _ int) ([]metadb.ChannelRuntimeMeta, metadb.ChannelRuntimeMetaCursor, bool, error) { r.calls++ if r.err != nil { return nil, metadb.ChannelRuntimeMetaCursor{}, false, r.err } page := r.pages[slotID][after] return append([]metadb.ChannelRuntimeMeta(nil), page.items...), page.cursor, page.done, nil }

围绕该 fake，测试覆盖了四类行为（当前仓库 channel_drain_test.go 中还新增了游标停滞用例）：

用例 1：跨 Slot 统计目标节点角色。Slot 1 上构造两条元数据——节点 4 是 Channelleader的 leader（同时在Replicas与ISR中）且是 Channelreplica的副本（ISR不含 4）；Slot 2 上节点 4 仅出现在 Channelisr的ISR中。断言结果为LeaderCount=1, ReplicaCount=2, ISRCount=2, ScannedSlotCount=2，且Unknown=false, Safe=false：

inv, err := app.NodeChannelDrainInventory(context.Background(), NodeChannelDrainInventoryRequest{NodeID: 4, PageLimit: 2}) if err != nil { t.Fatalf("NodeChannelDrainInventory() error = %v", err) } if inv.Unknown || inv.Safe || inv.LeaderCount != 1 || inv.ReplicaCount != 2 || inv.ISRCount != 2 || inv.ScannedSlotCount != 2 { t.Fatalf("inventory = %#v, want counted unsafe target roles", inv) }

注意输入快照故意给出不有序的Slots: []control.SlotAssignment{{SlotID: 2}, {SlotID: 1}}，验证实现确实按排序后的 Slot 顺序扫描。

用例 2：单 Slot 内多页推进。Slot 1 的第一页只有一条记录且done=false，游标推进到{ChannelID: "a", ChannelType: 1}后第二页返回剩余记录并done=true。断言ReplicaCount=1, ISRCount=1且reader.calls == 2，证明游标被正确传递、分页循环终止条件正确。

用例 3：依赖缺失与扫描错误的 fail-closed。构造两种“证据不足”场景：其一，Options不注入ChannelRuntimeMeta（reader 缺失）；其二，reader 的err字段返回errors.New("scan failed")。两种场景都断言Unknown=true, Safe=false，且后者还要求LastError != ""，保证错误信息能透传到诊断输出：

inv, err := missing.NodeChannelDrainInventory(context.Background(), NodeChannelDrainInventoryRequest{NodeID: 4}) if err != nil { t.Fatalf("missing reader inventory error = %v", err) } if !inv.Unknown || inv.Safe { t.Fatalf("missing reader inventory = %#v, want unknown unsafe", inv) }

用例 4（当前仓库新增）：游标停滞 fail-closed。fake 返回done=false但下一页游标与输入游标相同（零值游标），断言盘点结果Unknown=true, Safe=false且LastError非空，对应实现中的nextCursor == after检测分支。

4. 有界扫描的纵深设计

Stage 5B 的“有界”不只是页大小钳制。从当前仓库实现可以看到三层防护，它们共同保证一次盘点不会无限消耗存储资源：

后两层是计划落地后在 channel_drain.go 中补强的（计划原文只定义了单页 limit 归一化）：总页数预算把最坏情况扫描量显式封顶——Slot 数量 × 每 Slot 页数 × 1024 条/页，上界是确定的；游标停滞检测则封住了“存储层 bug 导致死循环”的尾部风险。两者触发时都走同一个 fail-closed 语义：宁可报告“未知”阻塞，也不输出“安全”结论。

此外，channel_drain.go 中保留了 gofail 故障注入标记（wkScaleInChannelDrainInventoryFault），配套有 gofail_markers_test.go 保证注入点存在。这为后续 Stage 7（gofail 故障注入）在真实链路中复现“盘点中途失败 → 状态 fail-closed”留了钩子。

5. Task 2：把 Channel 阻塞项合并进缩容状态

Task 2 修改scale_in.go与scale_in_test.go（当前路径 internal/usecase/management/scale_in.go、scale_in_test.go），把盘点结果并入NodeScaleInStatus。

5.1 状态响应扩展字段

计划要求给NodeScaleInStatusResponse增加五个字段，当前实现（scale_in.go）逐一对应：

// BlockedByChannels reports Channel leader, replica, ISR, or unknown inventory blockers. BlockedByChannels bool // UnknownChannelInventory reports that Channel inventory could not be proven. UnknownChannelInventory bool // ChannelLeaderCount counts Channels led by the target node. ChannelLeaderCount int // ChannelReplicaCount counts Channels where the target is a configured replica. ChannelReplicaCount int // ChannelISRCount counts Channels where the target is in ISR. ChannelISRCount int

BlockedByChannels是布尔汇总位，覆盖“unknown inventory”与三种角色命中两类情形；三个计数则提供可诊断的细节，让管理端能区分“阻塞来自 leader 角色”还是“来自 ISR 成员身份”。

5.2 合并逻辑

计划要求在nodeScaleInStatusFromSnapshot中、Stage 4 的 Slot/任务阻塞项之后合并NodeChannelDrainInventory，缺失 inventory 必须 fail-closed。当前实现见 scale_in.go：

func (a *App) markScaleInChannelBlockers(ctx context.Context, snapshot control.Snapshot, targetNode uint64, response *NodeScaleInStatusResponse) { inventory := a.nodeChannelDrainInventoryFromSnapshot(ctx, snapshot, NodeChannelDrainInventoryRequest{NodeID: targetNode}) response.UnknownChannelInventory = inventory.Unknown response.ChannelLeaderCount = inventory.LeaderCount response.ChannelReplicaCount = inventory.ReplicaCount response.ChannelISRCount = inventory.ISRCount response.BlockedByChannels = inventory.Unknown || inventory.LeaderCount > 0 || inventory.ReplicaCount > 0 || inventory.ISRCount > 0 }

有两个实现细节值得注意：

• 合并走的是nodeChannelDrainInventoryFromSnapshot（基于当前快照计算），而不是再调一次LocalControlSnapshot。缩容状态一次取快照、全部阻塞项基于同一份快照推导，避免了多数据源之间 revision 不一致带来的竞态；
• 在 nodeScaleInStatusFromSnapshot 中，Channel 阻塞项仅在目标节点确实处于合法缩容前置状态（未缺失、join state 正确、角色正确）时才计算，避免对不满足前提的节点做无意义的重扫描。

5.3 SafeToProceed 计算更新

计划给出的新条件是SafeToProceed必须同时满足：

!response.BlockedByChannels && !response.UnknownChannelInventory && response.ChannelLeaderCount == 0 && response.ChannelReplicaCount == 0 && response.ChannelISRCount == 0

当前实现的完整 SafeToProceed 计算 将其与既有阻塞项并在一起：

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.RuntimeUnknown

这里体现了 Stage 5 索引文档中SafeToProceed/SafeToRemove分层约束的最终形态：5B 加入的 Channel 条件只收紧SafeToProceed；SafeToRemove在其基础上再叠加 runtime drain 闸门（gateway 处于 drain 模式、会话与在线连接计数归零等，由markScaleInRuntimeDrainBlockers计算，见 scale_in.go）。计划原文特意说明“本阶段先不新增最终SafeToRemove，Stage 5C 先加 runtime drain 阻塞项”——从当前代码看，这行演进已经按 5C/5D 的规划完成了，且MarkNodeRemoved入口（scale_in.go）只有在status.SafeToRemove为真时才允许写入removed。

5.4 Task 2 的测试

计划为状态合并补充两个测试，断言核心：

• TestScaleInStatusBlocksWhenChannelInventoryReferencesTarget：构造一个就绪快照（节点 4 为 leaving 的 Data 节点，Slot 1 的 desired peers 为 {2,3} 不含 4），再注入一条Leader=4, Replicas={2,4}, ISR={4}的 Channel 元数据。断言SafeToProceed=false、BlockedByChannels=true，且三个 Channel 计数各为 1；
• TestScaleInStatusBlocksWhenChannelInventoryUnknown：同样的快照但不注入ChannelRuntimeMetareader。断言SafeToProceed=false、BlockedByChannels=true且UnknownChannelInventory=true——这是 5B 最核心的 fail-closed 语义的测试化表述：reader 缺失不能等价于“没有 Channel 阻塞”。

辅助构造器（快照、runtime summary、安全的 Slot runtime reader）也在测试文件中随用例一并给出，例如scaleInReadyNoSlotReplicaSnapshot()构造了含 4 个节点（1 个 Controller、3 个 Data，其中节点 4 为NodeJoinStateLeaving）与 1 个 Slot 分配的control.Snapshot。

6. 出口门禁与验收标准

计划定义的 Exit Gate 由三段命令构成：

GOWORK=off go test ./internalv2/usecase/management -run 'TestNodeChannelDrainInventory|TestScaleInStatus|TestAdvanceNodeScaleIn' -count=1 git diff --check rg -n "NodeChannelDrainInventory|BlockedByChannels|UnknownChannelInventory|ChannelLeaderCount|ChannelReplicaCount|ChannelISRCount" internalv2/usecase/management

期望：全部通过，且行为不变量成立——只要 Channel inventory 未知或目标节点出现在任何 Channel 成员关系中，SafeToProceed恒为 false。第三段rg检查本质上是“符号面验收”：确认六个新符号都已进入 usecase 包，防止实现与计划脱节。按当前仓库路径执行时，把./internalv2/usecase/management替换为./internal/usecase/management即可。

7. 小结：5B 的工程质量特征

Stage 5B 虽然只是一次只读盘点的落地，但完整体现了 WuKongIM 动态节点生命周期设计的几个一致原则，值得在类似系统改造中参考：

• 证据导向的 fail-closed：所有“读不到”的路径（依赖缺失、扫描错误、游标停滞、预算耗尽）统一收敛为Unknown语义并计入阻塞判断，代码中不存在“证据缺失则放行”的分支；
• 严格有界：单页 limit、总页数预算、排序后的确定扫描顺序三层约束，使安全检查自身的资源消耗有明确上界；
• 最小接口依赖：usecase 只依赖ChannelRuntimeMetaReader一个方法，真实存储细节被隔离在 pkg/cluster 与 pkg/slot/proxy 实现层，测试可完全用内存 fake 驱动；
• 可独立合入的切分：5B 不触碰任何写路径，为 5C（gateway drain）、5D（safe remove 路由）留下了清晰、已测试的输入契约。

关键代码位置汇总：盘点实现与测试见 channel_drain.go 与 channel_drain_test.go；状态合并与SafeToProceed计算见 scale_in.go；元数据模型见 table_runtime_meta.go；分页读取实现见 node_meta_runtime_scan.go 与 store.go；阶段规划见 Stage 5B 计划 与 Stage 5 索引。

• 即时通讯
• 后端