WuKongIM Slot Raft 日志手动压缩实战：从 Manager HTTP 路由到 Multi-Raft 底层实现

• 即时通讯
• 后端

导读

本文以 WuKongIM 仓库内《Internalv2 Slot Raft Compaction Implementation Plan》及其配套设计文档为核心，完整讲解"按节点、按 Slot 手动触发本地 Raft 日志压缩"这一运维能力的落地过程：包括POST /manager/nodes/:node_id/slots/:slot_id/compact管理端接口的契约形态、HTTP → management usecase → infra cluster 适配器 → 本地 Slot Multi-Raft 运行时（或远端节点 RPC）的完整调用链，以及每一步对应的真实源码位置。读完本文，你将掌握如何在 WuKongIM 中通过 Manager API 触发单个节点上指定 Slot 的日志压缩、理解压缩结果 DTO 中每个字段的业务含义、看懂compacted/skipped_reason的判定逻辑，并能复现该功能的单元测试与 web 端契约验证流程。

说明：计划文档中的目标路径写作internalv2/...，仓库实际落地代码位于internal/...，本文以当前仓库真实路径为准。

一、功能背景：为什么要给 Slot Raft 提供手动压缩

WuKongIM 的存储层按 Slot 划分数据分片，每个 Slot 由一组节点通过 Multi-Raft 组维护本地日志。长期运行后，单个 Slot 的 Raft 日志会不断累积已提交、已应用的 entry，占用本地存储并拖慢日志读取与重放。系统自带基于阈值（TriggerEntries、CheckInterval）的自动压缩器（见 pkg/slot/multiraft/compaction.go 中的logCompactor），但运维人员仍需要一种面向管理界面的、按需触发的操作入口——这正是本文所讲功能的出发点。

该功能的原始目标（来自 设计文档）非常明确：

让internalv2manager API 支持手动的节点本地 Slot Raft 日志压缩，使已有的 web Slot 日志页面可以通过wukongimv2触发压缩。

压缩本质上是一个运维写操作，因此它与既有"只读"的分布式日志查询 RPC 必须分离，需要一套独立的 Manager Slot Raft RPC 服务。这就是整个架构设计的原点。

二、API 契约：接口形态与响应结构

2.1 路由与权限

功能对外暴露的接口为：

POST /manager/nodes/:node_id/slots/:slot_id/compact

在真实源码 internal/access/manager/server.go 中，该路由注册在独立的slotWrites分组下，并强制要求写权限cluster.slot:w（与只读分组cluster.slot:r严格区分）：

slots := s.engine.Group("/manager") if s.auth.enabled() { slots.Use(s.requirePermission("cluster.slot", "r")) // 只读：列表、日志查询 } slots.GET("/slots", s.handleSlots) slots.GET("/slots/:slot_id/logs", s.handleSlotLogs) slotWrites := s.engine.Group("/manager") if s.auth.enabled() { slotWrites.Use(s.requirePermission("cluster.slot", "w")) // 写：压缩、Leader 转移 } slotWrites.POST("/nodes/:node_id/slots/:slot_id/compact", s.handleCompactSlotRaftLog)

权限模型由 internal/access/manager/server.go 中的PermissionConfig/UserConfig定义：每个 manager 静态用户通过resource + actions获得授权，资源为cluster.slot、动作码为r/w。也就是说，没有cluster.slot:w权限的账号调用该接口会收到 403 拒绝，这是本功能的第一道安全边界。

2.2 成功响应：兼容 web 既有契约

压缩触发成功后，接口返回与既有 web Slot 日志页面契约一致的 JSON（来自 设计文档 的 API Shape 一节）：

{ "generated_at": "2026-06-19T00:00:00Z", "total": 1, "succeeded": 1, "failed": 0, "items": [{ "node_id": 1, "slot_id": 9, "success": true, "applied_index": 10, "before_snapshot_index": 4, "after_snapshot_index": 10, "compacted": true, "skipped_reason": "", "error": "" }] }

各字段含义如下表（字段与注释均可在 internal/access/manager/slot_raft_compaction.go 的 DTO 定义中找到对应）：

虽然该接口当前只接收单个 node/slot 目标，但响应结构从设计上就是"摘要 + 逐项结果"的批量形态，为将来多目标批量压缩保留了契约空间。

2.3 错误语义

设计文档 的 Error Handling 一节规定了完整错误映射，源码在 internal/access/manager/slot_raft_compaction.go 的writeSlotRaftCompactError中落地：

其中参数校验的规则可在 usecase 层看到：nodeID == 0 || slotID == 0直接返回metadb.ErrInvalidArgument（见 internal/usecase/management/slot_raft_compaction.go），HTTP 层再将其翻译为 400。Slot 不存在（ErrSlotNotFound）则来自多级错误映射：multiraft 的ErrSlotNotFound→mapSlotLogRuntimeError→ 最终透传。

三、架构分层：一次压缩请求的完整调用链

3.1 分层设计原则

按计划文档的 Architecture 一节，该功能镜像了 legacy manager 契约，同时保持了 internalv2 的分层约束：

HTTP adapter (access/manager) ↓ management usecase (usecase/management) ↓ infra cluster adapter (infra/cluster) ↓ 本地: clusterv2 facade → 本地 Slot 运行时 远端: node RPC → 目标节点本地执行

设计上刻意让 HTTP 路由保持"薄适配器"角色：它只负责解析node_id/slot_id、执行cluster.slot:w权限校验、委托给 usecase、最后把 DTO 序列化成 JSON。真正的参数校验、结果组装由 usecase 负责；"本地执行还是远端执行"由 infra 适配器根据目标节点 ID 决定。

3.2 各层真实源码对应

四、分层源码逐层剖析

4.1 Management Usecase：契约与汇总逻辑

usecase 层定义了功能的核心抽象——SlotRaftOperator接口（internal/usecase/management/slot_raft_compaction.go）：

type SlotRaftOperator interface { SlotRaftStatus(context.Context, uint64, uint32) (SlotNodeLogStatus, error) CompactSlotRaftLog(context.Context, uint64, uint32) (SlotRaftCompactionResult, error) }

接口同时暴露状态查询与压缩两个操作，由 infra 层的ManagementSlotRaftOperator实现（见下文）。当应用未接线该 operator 时，返回哨兵错误：

var ErrSlotRaftOperatorUnavailable = errors.New("internal/usecase/management: slot raft operator unavailable")

App.CompactSlotRaftLog的处理流程（internal/usecase/management/slot_raft_compaction.go）：

1. 校验ctx未取消（ctxErr）；
2. 校验nodeID != 0 && slotID != 0，否则返回metadb.ErrInvalidArgument；
3. 校验a.slotRaft != nil，否则返回ErrSlotRaftOperatorUnavailable；
4. 调用a.slotRaft.CompactSlotRaftLog(ctx, nodeID, slotID)；
5. 通过slotRaftCompactNodeResult组装单个 item，并计算Total/Succeeded/Failed摘要；
6. GeneratedAt使用a.now()（可注入的时钟），保证测试可确定性断言时间。

注意slotRaftCompactNodeResult的一个细节（internal/usecase/management/slot_raft_compaction.go）：如果底层结果里已带NodeID/SlotID，则以底层实际处理节点为准覆盖请求参数；如果 operator 返回了 error 而 item 的Error字段为空，则把err.Error()填入 item。这保证了"节点已接受但执行失败"时错误能透传到items[].error，而不是被吞掉。

4.2 Infra 集群适配器：本地 vs 远端路由

ManagementSlotRaftOperator（internal/infra/cluster/management_slot_raft.go）持有两个关键依赖：

• node ManagementSlotRaftNode：本地节点能力接口，暴露NodeID()、CallRPC(...)、LocalSlotRaftStatus(...)、LocalCompactSlotRaftLog(...)；
• remote *accessnode.Client：节点 RPC 客户端，由NewManagementSlotRaftOperator内部用accessnode.NewClient(node)构造。

压缩路由的核心判断是目标节点是否为本节点：

func (o *ManagementSlotRaftOperator) CompactSlotRaftLog(ctx context.Context, nodeID uint64, slotID uint32) (managementusecase.SlotRaftCompactionResult, error) { if o == nil || o.node == nil || o.remote == nil { return managementusecase.SlotRaftCompactionResult{}, managementusecase.ErrSlotRaftOperatorUnavailable } if nodeID == o.node.NodeID() { result, err := o.node.LocalCompactSlotRaftLog(ctx, slotID) // 本地：直接调用 facade ... return slotRaftCompactionFromCluster(result), nil } return o.remote.CompactManagerSlotRaftLog(ctx, nodeID, slotID) // 远端：走节点 RPC }

• nodeID == 本机 ID：走LocalCompactSlotRaftLog本地路径；
• 否则：走CompactManagerSlotRaftLogRPC 远端路径，目标节点收到 RPC 后执行的是完全相同的本地 usecase/operator 路径——这正是设计文档强调的"远端节点运行同一套本地逻辑"。

slotRaftCompactionFromCluster负责把pkg/cluster.SlotRaftCompactionResult映射为 management 层的SlotRaftCompactionResult（字段一一对应）。

4.3 本地节点 Facade：对接默认 Slot Multi-Raft 运行时

Node.LocalCompactSlotRaftLog（pkg/cluster/node_logs.go）是本地执行的最终入口：

func (n *Node) LocalCompactSlotRaftLog(ctx context.Context, slotID uint32) (SlotRaftCompactionResult, error) { if err := ctxErr(ctx); err != nil { ... } if slotID == 0 { return SlotRaftCompactionResult{}, ErrSlotNotFound } if err := n.ensureForeground(); err != nil { ... } if n.defaultSlotRuntime == nil { return SlotRaftCompactionResult{}, ErrNotStarted } result, err := n.defaultSlotRuntime.CompactLog(ctx, multiraft.SlotID(slotID)) if err != nil { return SlotRaftCompactionResult{}, mapSlotLogRuntimeError(err) } out := slotRaftCompactionResultFromRuntime(result) if out.NodeID == 0 { out.NodeID = n.NodeID() } if out.SlotID == 0 { out.SlotID = slotID } return out, nil }

几个关键点：

• slotID == 0直接返回ErrSlotNotFound；
• 依赖n.defaultSlotRuntime（默认 Slot Multi-Raft 运行时），未启动时返回ErrNotStarted；
• mapSlotLogRuntimeError（pkg/cluster/node_logs.go）把 multiraft 的ErrSlotNotFound/ErrRuntimeClosed归一为ErrSlotNotFound，其余错误原样透传；
• slotRaftCompactionResultFromRuntime（pkg/cluster/node_logs.go）把multiraft.LogCompactionResult映射为稳定的 clusterv2 结果；若运行时未填NodeID/SlotID，由 facade 补上本机 ID 与请求的 slotID。

与本地路径对应的测试见 pkg/cluster/node_logs_test.go（搜索LocalCompactSlotRaftLog相关用例），计划文档 Task 2 要求"使用 fake Slot compactor/runtime seam 或既有默认运行时 fixture"证明 facade 确实驱动了默认 Slot 运行时。

4.4 节点 RPC：独立的 Manager Slot Raft 服务

由于压缩是运维写操作，而既有 manager 日志 RPC 是只读的，因此计划文档明确要求"使用一个专用的 manager Slot Raft RPC 服务"。真实实现：

服务 ID 注册在 pkg/cluster/net/ids.go：

// RPCManagerSlotRaft serves internal node-local manager Slot Raft operations. RPCManagerSlotRaft

并登记到 RPC 服务名解析表（同文件serviceNameForID的 case 分支），internal/access/node/manager_slot_raft_rpc.go中通过const ManagerSlotRaftRPCServiceID uint8 = clusternet.RPCManagerSlotRaft引用。

RPC 编解码在 internal/access/node/manager_slot_raft_codec.go：

• 请求/响应都以 5 字节 magic 前缀开头：请求{'W','K','V','S',1}，响应{'W','K','V','s',1}，随后是 JSON body；
• 请求结构{op, node_id, slot_id}，op取值"status"或"compact"（常量managerSlotRaftOpStatus/managerSlotRaftOpCompact）；
• 响应结构{status, raft, compaction}，其中compaction字段承载managementusecase.SlotRaftCompactionResult。

服务端 handler（internal/access/node/manager_slot_raft_rpc.go）按req.Op分派：status走SlotRaftStatus，compact走CompactSlotRaftLog；若 adapter 未接线managerSlotRaft，则直接返回rpcStatusRejected。未知 op 记录结构化日志并返回错误。

客户端方法（同文件Client.CompactManagerSlotRaftLog，internal/access/node/manager_slot_raft_rpc.go）：

1. 编码{op: "compact", node_id, slot_id}请求；
2. 通过c.node.CallRPC(ctx, nodeID, ManagerSlotRaftRPCServiceID, body)调用远端；
3. 解码响应并校验Status（managerSlotRaftRPCErrorForStatus：not_found→ErrSlotNotFound，rejected→ErrSlotRaftOperatorUnavailable，context_canceled/context_deadline_exceeded原样映射）；
4. 返回resp.Compaction。

RPC 状态映射（managerSlotRaftRPCStatusForError）对错误做了精细归约：ErrNotFound/ErrSlotNotFound→not_found；ErrInvalidArgument/ErrSlotRaftOperatorUnavailable→rejected；context 取消/超时单独映射。拒绝类错误还会写结构化告警日志（internal.access.node.manager_slot_raft_rejected），便于排障。

4.5 底层 Multi-Raft：手动压缩的真正执行点

整个链条最终落在Runtime.CompactLog（pkg/slot/multiraft/api.go）：

1. 校验运行时未关闭、Slot 存在（不存在返回ErrSlotNotFound）；
2. 构造logCompactionRequest{ctx, resp: make(chan logCompactionResponse, 1)}，投递controlCompactLog控制动作到该 Slot 的控制队列（g.enqueueControl）；
3. r.scheduler.enqueue(slotID)唤醒调度器；
4. 在req.resp与ctx.Done()之间做 select：要么拿到压缩结果，要么被调用方 context 取消/超时打断。

Slot 内部由compactLogManually（pkg/slot/multiraft/compaction.go）执行判定与真正压缩：

result := {NodeID: g.nodeID(), SlotID: g.id, AppliedIndex: applied} snapshotIndex := g.compactor.lastSnapshotIdx if compactor 未配置或 cfg.Enabled == false: SkippedReason = "disabled" // 压缩被配置关闭 if applied == 0: SkippedReason = "no_applied_index" // 尚无已应用 entry snapshotUpToDate := applied <= beforeSnapshotIndex logUpToDate := firstIndex > applied if snapshotUpToDate && logUpToDate: SkippedReason = "up_to_date" // 快照已覆盖，日志已裁剪 否则: compactLogAt(ctx, applied, false) // 真正创建快照并裁剪日志 AfterSnapshotIndex = applied; Compacted = true

由此可以总结skipped_reason的三种取值（常量定义在 pkg/slot/multiraft/compaction.go）：

compactLogAt（pkg/slot/multiraft/compaction.go）是压缩的落地函数，步骤包括：

1. 若状态机实现了DurableAppliedStateMachine，先storage.MarkApplied(ctx, applied)把应用水位镜像进 Raft 存储；
2. 若实现了CheckpointStateMachine，先采集RecoveryState作为压缩前检查点证明；
3. stateMachine.Snapshot(ctx)生成业务状态快照，取storageView.memory.Term(applied)与当前 confState，组装raftpb.Snapshot；
4. 通过storage.Save(ctx, PersistentState{Snapshot: &snap})（或实现了ExternalSnapshotStorage时用ReplaceSnapshot）持久化快照；
5. memory.CreateSnapshot(applied, ...)+memory.Compact(applied)更新内存视图并裁剪已压缩 entry；
6. resetCheckpointAnchor复位检查点锚点。

这解释了响应字段before_snapshot_index/after_snapshot_index的来源：before来自compactor.lastSnapshotIdx，after在成功压缩后等于applied。

五、应用接线（App Wiring）：把整条链串起来

上述各层组件并非自动生效，需要应用层显式接线。真实接线在 internal/app/wiring.go：

• 构建 infra operator 的前提是a.cluster实现了clusterinfra.ManagementSlotRaftNode接口（internal/app/wiring.go）：

node, hasNode := a.cluster.(clusterinfra.ManagementSlotRaftNode) ... operator := clusterinfra.NewManagementSlotRaftOperator(node)

• 接线管理选项时再次以类型断言方式装配SlotRaftoperator（internal/app/wiring.go）：

if slotRaftNode, ok := a.cluster.(clusterinfra.ManagementSlotRaftNode); ok { opts.SlotRaft = clusterinfra.NewManagementSlotRaftOperator(slotRaftNode) opts.SlotRuntimeStatus = clusterinfra.NewManagementSlotRuntimeStatusReader(slotRaftNode) }

• 同时，RPCManagerSlotRaft服务的 handler 需要在节点 access adapter 上注册（对应计划文档 Task 5 的"app RPC registration"），使远端节点能够响应compact/status操作。

若接线缺失会发生什么？按设计文档的 Error Handling，manager 未配置或SlotRaftoperator 未接线时，接口返回503 service_unavailable（分别对应 HTTP 层的management == nil检查和 usecase 层的a.slotRaft == nil检查）。这也解释了为什么计划文档 Task 5 专门要求"写失败的 app wiring 测试来证明缺失路由/接线"，以及 Task 1/3 要求先写失败测试再补实现——接线缺失必须表现为可观测的 503，而不是静默 200。

六、测试与验证：如何复现完整的质量保障流程

计划文档 Task 6 给出了完整的验证矩阵，与仓库中已落地的测试一一对应：

6.1 Go 单元测试（后端分层验证）

go test ./internal/usecase/management ./pkg/cluster ./internal/access/manager ./internal/access/node ./internal/infra/cluster ./pkg/cluster/net ./internal/app -count=1

各层对应测试文件：

计划文档中每个 Task 都遵循 TDD 节奏：先写失败测试 → 确认 API 缺失 → 补实现 → 跑通定向测试（如go test ./internal/usecase/management -run 'TestSlotRaft' -count=1、go test ./pkg/cluster -run TestLocalSlot -count=1）。这种"红绿循环"保证了每个分层契约都有独立测试背书。

6.2 Web 契约验证（前端回归）

cd web && yarn test src/pages/slot-logs/page.test.tsx src/lib/manager-api.test.ts

web 端测试以既有 Slot 日志页面与 manager API 客户端为契约锚点（文件位于 web/src/pages/slot-logs 与 web/src/lib），后端改完接口后前端用例保持不变，用于证明响应结构没有破坏既有页面契约。

6.3 代码格式与收尾

计划文档要求对涉及的 Go 文件执行gofmt后重复失败/定向测试。此外，计划文档 Task 5 还要求在功能落地后同步更新各层FLOW.md（internal/access/manager、internal/usecase/management、internal/app、internal/infra/cluster、pkg/cluster下的流程文档），记录新路由与 operator 路径——这正是仓库维护这类功能时"文档与代码同步演进"的惯例。

七、实战小结：从需求到可运维能力的要点回顾

1. 接口形态：POST /manager/nodes/:node_id/slots/:slot_id/compact，写权限cluster.slot:w，响应为"摘要 + 逐项"结构，与 web Slot 日志页既有契约完全兼容；
2. 分层纪律：HTTP 薄适配器 → management usecase（校验/汇总）→ infra cluster 适配器（本地/远端路由）→ 本地 facade 或节点 RPC → 目标节点本地 Slot Multi-Raft 运行时；
3. 读写分离：压缩是运维写操作，使用独立RPCManagerSlotRaft服务，与只读日志 RPC 严格隔离；拒绝/缺失接线必须表现为 503；
4. 错误分级：参数非法 400、未接线 503、Slot 不存在 404、无法启动 500、节点接受后失败透传为items[].error；
5. 底层语义：compacted=true表示真正创建了新快照并裁剪日志；skipped_reason的disabled/no_applied_index/up_to_date三种取值对应压缩被关闭、无已应用索引、快照已最新三种情形；
6. 质量保障：每层都有 TDD 式失败测试先行（usecase/infra/RPC/facade/HTTP），web 端用既有页面测试做契约回归，落地后同步更新 FLOW.md。

对于部署 WuKongIM 的团队，这条能力链路的意义在于：当某个 Slot 的本地 Raft 日志长期膨胀、磁盘水位偏高时，运维可以直接通过 Manager API（或 web Slot 日志页面）精准地针对单个节点、单个 Slot 发起压缩，并在响应中看到before_snapshot_index → after_snapshot_index的推进证据，而不必重启节点或依赖不可控的自动压缩时机。

参考文档：实现计划 | 设计文档 | 源码实现见文内各链接。

• 即时通讯
• 后端