Coordinator

​

职责与边界

• watch etcd 中的 Worker 注册与心跳，维护一份本地、可丢弃的 WorkerView。
• 对每次 ReserveWorkerSlot 做校验、就绪判定、候选过滤与排序，然后串行尝试少量候选。
• 把 Worker 返回的 slotId 原样转交 Client；不解释 slotId 的内部含义。
• 基于最近失败对 Worker 做本地惩罚和短暂熔断。

• 解析 task payload、生成 taskId、接收 SubmitTask 或 TaskResult。
• 维护中心化 slot 账本、pending 队列或全局 taskId 去重。
• 向 etcd 写任何数据；etcd 只承载 Worker 发现与心跳。

• Worker 本地 slot 状态机是唯一真相；Coordinator 重启只丢失视图缓存和路由偏置，不影响正确性。
• CoordinatorCore 是 Sans-IO 的纯决策引擎：不读时钟、不做网络，所有方法显式接收 now（UTC 毫秒）。
• 首轮全量 Worker 视图建立前，以及 watch 断流超过宽限期后，ReserveWorkerSlot 一律返回 Unavailable。
• 单次请求串行尝试候选，不做 hedged request；结果不确定时只对同一 Worker、同一 taskId 重试，绝不切换 Worker。
• 同一套代码用不同 app.Profile 起两套进程：cmd/coordinator（block 集群）和 cmd/bundle_coordinator（bundle 集群）。bundle 集群的细节见 Bundle 集群。

​

代码位置

​

核心类型与接口

​

core

• core.Config（internal/coordinator/core/coordinator.go）：选址与保护参数，DefaultConfig() 给默认值，Validate() 要求 WatchGracePeriodMs <= HeartbeatTimeoutMs。
• core.CoordinatorCore：持有 map[string]*WorkerView、initialized、watchDisconnectedAtMs。事件入口：ApplyFullWorkerList、ApplyWorkerUpdate、ApplyWorkerRemoval、MarkWatchDisconnected、MarkWatchReconnected。决策出口：IsReady、SelectCandidate。反馈入口：RecordSlotSuccess、RecordSlotRejection、RecordSlotBackpressure、RecordSlotTimeout。
• core.WorkerView（internal/coordinator/core/worker_view.go）：内嵌 etcd.WorkerHeartbeat，加三个只在本地存在、不写回 etcd 的字段。

type WorkerView struct {
	WorkerHeartbeat

	// Local-only fields — not from etcd, not written back.
	CircuitBreakerDeadlineMs int64
	RecentFailureCount       int
	Penalty                  float64
}

• WorkerView.UsedTotal() = ReservedSlots + RunningTasks；WorkerView.FreeCapacity() = TaskSlots - UsedTotal()。
• core.CandidateResult{WorkerAddr, Err} 与 core.CoordinatorError{Code, Message}：SelectCandidate 的输出，Code 是 commontypes.ErrorCode。
• core.ValidateReserveRequest(taskID)（internal/coordinator/core/validate.go）：只检查 taskId 非空。

​

adapters

• adapters.CoordinatorServer（grpc_server.go）：实现 coordinatorpb.CoordinatorServiceServer，持有 core、SlotReserver 和一把 sync.RWMutex；Lock/Unlock 暴露给 watcher 用于写 core。
• adapters.HandlerConfig{MaxTimeoutRetries}：不确定结果时对同一 Worker 的重试次数，默认 1。
• adapters.SlotReserver 接口：RequestTaskSlot(ctx, workerAddr, req) (*commontypes.TaskSnapshot, *SlotError)，测试里用 mock 替换。
• adapters.WorkerRPCClient（worker_rpc.go）：SlotReserver 的 gRPC 实现，按 workerAddr 缓存一条 grpc.ClientConn；grpcErrToSlotError 决定 SlotError.MayHaveAllocated。
• adapters.SlotError{Code, Message, MayHaveAllocated}：MayHaveAllocated=true 表示请求可能已到达 Worker，必须留在同一 Worker 重试。
• adapters.EtcdWorkerWatcher / EtcdWatcherConfig{Endpoints, KeyPrefixes, DialTimeout, RetryDelay}（etcd_watcher.go）：Run(ctx) 阻塞运行 list+watch。
• adapters.BundleCoordinatorServer（bundle_grpc_server.go）：bundle proto 到 CoordinatorServer 的薄转换层。

​

app 与 api

• app.Profile{Deployment, Service, WorkerRegistryPrefixes}（internal/coordinator/app/app.go）：入口身份。Deployment 只能是 "block" 或 "bundle"，并决定注册哪个 gRPC service、允许哪种 registry kind。
• app.Run(p Profile)：装配 obs、core、WorkerRPCClient、CoordinatorServer、etcd client、watcher、gRPC server，阻塞到收到 SIGINT/SIGTERM。
• app.CoordinatorRuntimeConfig（config.go）：进程级配置聚合，LoadCoordinatorRuntimeConfigFromEnv(p) 从环境变量覆盖并 Validate()。
• etcd.WorkerHeartbeat（api/etcd/types.go）：Worker 写入 etcd 的 JSON 值，字段包括 workerAddr、taskSlots、reservedSlots、runningTasks、cpuPercent、memoryPercent、memoryAvailableBytes、memoryTotalBytes、processRssBytes、errorRate、lastHeartbeatMs。
• etcd.ParseRegistryPrefix 与常量 LegacyBlockWorkerRegistryPrefix（/blockx/workers/）、LegacyBundleWorkerRegistryPrefix（/blockx/bundle-workers/）、ProdDefaultBlockWorkerRegistryPrefix、ProdDefaultBundleWorkerRegistryPrefix、TestDefaultBundleWorkerRegistryPrefix（api/etcd/registry.go）。

​

数据流 / 执行流程

​

对外错误码

​

状态与生命周期

• initialized：ApplyFullWorkerList 第一次调用后为 true，之后不再回退。watcher 只有在所有配置 prefix 都完成初始 list 后才调用它。
• watchDisconnectedAtMs：任一 prefix 的 watch 断开时由 MarkWatchDisconnected(now) 记录首次断开时间；IsReady 与 SelectCandidate 在 now - watchDisconnectedAtMs > WatchGracePeriodMs 时返回 Unavailable。所有 prefix 重新 list 成功且都处于连接态后，MarkWatchReconnected 清零。

• RecentFailureCount 与 Penalty：RecordSlotRejection/RecordSlotTimeout 各加一，Penalty = float64(RecentFailureCount)。
• CircuitBreakerDeadlineMs：RecentFailureCount >= CircuitBreakerThreshold（默认 3）时设为 now + CircuitBreakerCooldownMs（默认 10s），期间该 Worker 被硬过滤。
• RecordSlotSuccess 把三者一起清零；RecordSlotBackpressure 对三者都不动。
• ApplyFullWorkerList 会保留幸存 Worker 的这三个字段；ApplyWorkerRemoval 删除后重新出现的 Worker 从零开始。

• slot TTL（SlotTTLMs，默认 5s）是 Coordinator 在 RequestTaskSlot 里传给 Worker 的预占窗口，不是 task 执行超时；Coordinator 不再解释它。
• 单次 ReserveWorkerSlot 没有独立的总时间预算；上界由 MaxCandidates × (1 + MaxTimeoutRetries) × WorkerRPCTimeout 和调用方 ctx 共同决定，默认最坏约 3 × 2 × 3s。
• 心跳新鲜度用 Coordinator 本地时钟与 Worker 写入的 lastHeartbeatMs 相减，机器间时钟偏差会直接影响过滤结果。

​

配置

• cmd/coordinator：/blockx/workers/。
• cmd/bundle_coordinator：/blockx/bundle-workers/ 与 /blockx/prod/lanes/default/bundle-workers/。
• 测试集群用 COORDINATOR_REGISTRY_PREFIXES=/blockx/test/lanes/default/bundle-workers/ 整体替换。

​

扩展点

• 改选址策略：只动 internal/coordinator/core/coordinator.go 的 SelectCandidate 与 candidateRanksBefore，并在 coordinator_test.go 补 TestSelectCandidate_*。core 没有 IO，可以纯函数式测试。当前是确定性排序；心跳里的绝对内存字段（memoryAvailableBytes）只被解析，不参与过滤和排序。
• 新增一种 Worker 反馈：在 core 加 RecordSlotXxx，在 adapters/grpc_server.go 的 recordSlotOutcome 里按 SlotError.Code 分派。RecordSlotBackpressure 就是这样加进来的。
• 改 Worker 错误分类：adapters/worker_rpc.go 的 grpcErrToSlotError 与 Worker 侧 workerCodeToGRPC（internal/worker/adapters/wire.go）必须保持互逆；改一边就要看另一边（见 Worker）。
• 新增部署形态：加一个 cmd/<name>/main.go 声明 app.Profile，在 app.registerCoordinatorService 与 validateRegistryPrefixes 里加分支，并在 api/etcd/registry.go 的 ParseRegistryPrefix 里允许新的 worker kind。
• 换 Worker RPC 实现：实现 adapters.SlotReserver 接口即可，CoordinatorServer 不依赖具体 client。当前 WorkerRPCClient 对每个 workerAddr 只建一条 grpc.ClientConn，不使用 internal/sdk/grpcclient 的多连接 balancer：指向单个 Worker 的点对点 client 不启用多连接策略。背景见 blockx 仓库 docs/specs/2026-07-23-grpc-client-balancing.md。
• 新增配置项：在 core.Config / adapters.HandlerConfig / adapters.EtcdWatcherConfig 加字段和 Validate 规则，再在 app/config.go 的 LoadCoordinatorRuntimeConfigFromEnv 加环境变量解析，并补 config_test.go。

​

测试

# core / adapters / app 单元测试，以及 registry prefix 解析
go test ./internal/coordinator/... ./api/etcd/...

# 进程级测试：编译真实二进制、启动内嵌 etcd、验证 gRPC 就绪与错误映射
go test ./cmd/coordinator/... ./cmd/bundle_coordinator/...

# 契约级 E2E（含 coordinator-etcd 与 coordinator-worker）
go test -v -timeout 300s ./e2e/contract/...

• internal/coordinator/core/coordinator_test.go：SelectCandidate 每个过滤条件与排序规则、熔断触发/恢复、IsReady 的宽限期语义、ApplyFullWorkerList 保留本地字段。
• internal/coordinator/adapters/handler_test.go：用 mock SlotReserver 覆盖编排循环（超时后换候选、重试仍超时返回 Aborted、全部拒绝返回 ResourceExhausted、空 SlotID 重试），以及真实 gRPC server 的错误码与压缩行为。
• internal/coordinator/adapters/worker_rpc_test.go：grpcErrToSlotError 的 MayHaveAllocated 分类（连接拒绝 vs 超时/取消/Aborted）。
• internal/coordinator/adapters/etcd_watcher_test.go：内嵌 etcd 上的 list+watch、多 prefix 合并去重、断连/重连的就绪状态、不可解析 value 跳过。
• internal/coordinator/app/config_test.go、profile_test.go：环境变量解析与 Profile 校验。
• e2e/contract/coordinator-etcd/：etcd 晚于 Coordinator 启动时先 Unavailable 后恢复；watch 在宽限期内恢复仍可路由；超过宽限期变 Unavailable。
• e2e/contract/coordinator-worker/：明确传输失败切换下一个 Worker、RequestTaskSlot 超时返回 SlotUncertain（Aborted）、slot 过期后 SubmitTask 被拒、Worker 重启后旧 slotId 失效、统一 slot 池满时拒绝。

​

相关文档

• docs/specs/task-resource-coordinator.md：Coordinator 详细设计，协议、错误语义、选址规则、一致性边界。
• docs/specs/architecture.md §4.1：控制面在整体架构中的定位与三条边界。
• docs/specs/2026-07-28-registry-prefix-migration.md：legacy / v2 registry prefix、多 prefix 合并与发布顺序。
• docs/specs/2026-07-23-bundle-coordinator-api.md：bundle Coordinator 独立 service 身份的由来。
• docs/specs/2026-07-23-grpc-client-balancing.md：多连接 balancer 的适用范围。
• docs/specs/worker.md：RequestTaskSlot 与 slot 状态机的权威定义。

• Worker：RequestTaskSlot 的 Worker 侧实现与 etcd 心跳发布。
• Bundle 集群：bundle 集群与 cmd/bundle_coordinator。
• 协议与接口：gRPC / etcd 协议汇总。
• Task 生命周期：从 ReserveWorkerSlot 到 SubmitTask 的完整链路。
• 可观测性：指标、日志与 tracing 的接入方式。