Worker

​

职责与边界

• 对外提供 gRPC WorkerService（api/grpc/worker/worker.proto）：RequestTaskSlot、SubmitTask、GetTaskResult、WatchTasks。WatchTasks 是 server-streaming RPC，Worker 只监听一个 TCP 端口。协议细节见 协议与接口。
• 维护本地 slot 表和 task 索引，做原子准入和 taskId 级幂等。
• 为每个 task 固定一个 taskCodeEpoch，串行跑 Call Builder，把 call list 交给 dispatcher 并行执行，最后串行跑 Writer Plugin。
• 收敛 TaskResult，推送给 WatchTasks 订阅者，并按 resultRetentionMs 保留。
• 向 etcd 注册并周期上报心跳，供 Coordinator 选址。

• 生成 taskId、构造 payload、DAG 编排或定时触发（这些在上游）。
• 跨 Worker 的去重、全局结果查询或崩溃后的本地恢复。重启后旧 slot、旧结果和订阅全部丢失。
• call 级调度细节（dispatcher、executor adapter）。这部分见 Call 执行子系统。

• WorkerCore 是纯内存、Sans-IO 的决策引擎。它不读时钟，每个方法都显式接收 now。
• slot 表是容量和 task 状态迁移的唯一真相；etcd 心跳只是它的派生视图。
• slotTable 与 taskIndex 在同一次方法调用里一起更新，不存在撕裂状态。
• 一旦 task 成功激活，后续所有失败（Builder、call、Plugin、超时、Watch 断连）都收敛为 TaskResult 里的终态，不再返回提交层错误。
• adapter 不维护第二份 slot 或 phase 真相。它只执行 core 输出的命令，再把结果作为事件回送。

​

代码位置

​

核心类型与接口

• WorkerCore（worker.go）：单线程状态机。持有 slots []SlotEntry、taskIndex map[string]*TaskIndexEntry、tasks map[string]*TaskContext、ready、draining。
• 输入事件（都在 worker.go，除非注明）：HandleRequestTaskSlot(taskID, ttlMs, now)、HandleSubmitTask(req, adapterTaskCtx, now)、HandleTaskActivationPrepared(taskID, slotID, taskCodeEpoch, now)、HandleTaskActivationFailed(taskID, failureCode, now)、HandleTaskPhaseFinished(taskID, phase, outcome, now)、HandleTaskTimedOut(taskID, now)、HandleWatchAttached / HandleWatchDetached / HandleWatchDisconnected(taskID, now)、HandleSlotLeaseExpired(slotID, now)（worker_slots.go）、HandleGetTaskResult(taskID)。
• 定时扫描辅助：ExpiredSlots(now)、TimedOutTasks(now)、WatchDisconnectedTasks(now)、PurgeExpiredResults(now)。adapter 每个 tick 调用它们，再把结果喂回对应 Handle*。
• 输出命令（types.go）：PrepareTaskActivation、StartBuilderPhase、StartCallPhase{CallList, Streaming}、StartWriterPhase{Outputs}、ConvergeTask{State, Result}。它们嵌在结果类型 RequestTaskSlotResult、SubmitTaskResult、ActivationResult、PhaseTransitionResult、LeaseExpiredResult、GetTaskResultOutput 里返回。
• TaskContext（types.go）：TaskID、SlotID、TaskCodeEpoch、Phase、DeadlineMs、WatchAttached / WatchEverAttached / WatchDisconnectDeadlineMs、CallFailurePolicy、CallRetryPolicy、AcceptedAtMs、AdapterTaskCtx，以及三个阶段的 PhaseOutcome。
• DeriveHeartbeat(now) etcd.WorkerHeartbeat：从 slot 表派生 ReservedSlots / RunningTasks。

// internal/worker/core/worker.go
func (w *WorkerCore) HandleSubmitTask(req commontypes.SubmitTaskRequest, adapterTaskCtx *commontypes.TaskCtx, now int64) SubmitTaskResult
func (w *WorkerCore) HandleTaskPhaseFinished(taskID string, phase TaskPhase, outcome PhaseOutcome, now int64) PhaseTransitionResult

• GRPCServer（grpc_server.go）：实现 RequestTaskSlot / SubmitTask / GetTaskResult，把 WatchTasks 委托给 SubscriptionManager。workerErrToStatus（wire.go）把 core.WorkerError 映射为 gRPC status，业务错误码经 workerCodeToGRPC 映射：InvalidArgument → InvalidArgument、NoSlot → ResourceExhausted、Unavailable → Unavailable、InvalidSlot → FailedPrecondition、NotFound → NotFound、SlotUncertain → Aborted。
• Orchestrator（orchestrator.go）：桥接 WorkerCore 和 DispatcherCore。入口方法 RequestTaskSlot、SubmitTask、GetTaskResult、TickTimers、DeriveHeartbeat、SetDraining、WaitDrain、RunDispatchLoop、SyncExecutorSnapshots、CheckWedgedExecutors。
• OrchestratorDeps（orchestrator.go）：构造依赖，包括 BuilderRegistry、PluginRegistry、IOScope、Admission、FunctionCodeViewProvider、EpochResolver、ExecAdapter、AuditGate、StreamBuild。
• taskRuntime（task_runtime.go）：adapter 侧的 task 运行时记录，持有 ioTaskCtx、TaskIOScope、ctx / cancel、functionCodeView、流式生产句柄和 TaskStatistic。它不维护 phase 或 slot 真相。
• SubscriptionManager（stream_subscriber.go）：每条 WatchTasks 流一个 streamSubscriber；把首条 attach / 最后一条 detach 翻译为 Orchestrator.HandleWatchAttached / HandleWatchDetached。
• EtcdPublisher（etcd_publisher.go）：Register、RunHeartbeatLoop、Deregister。key 为 KeyPrefix + workerAddr，lease TTL 默认 10s。
• ExecutorAdapter 接口（exec_adapter.go）：Orchestrator 依赖的 UDS executor 面，生产实现是 executor.Adapter。

• Profile（app.go）：Deployment（block / bundle，只做观测标识）、Service（tracing / Prometheus 服务名）、WorkerRegistryPrefix（etcd 注册前缀）、UsageService、Builders、Plugins、TuneDefaults（只改内置默认值）。
• Run(p Profile)：唯一的装配和启停序。

​

数据流 / 执行流程

• SubmitTask 在返回前只做到 HandleSubmitTask。core 立即创建 TaskContext（Phase = Preparing），使重复提交命中幂等路径；激活和阶段执行在 activateAndRun goroutine 中异步进行。
• 激活失败（epoch 解析失败、Pin 失败、NewTaskScope 失败）走 HandleTaskActivationFailed，失败码为 ACTIVATION_FAILED 或 IO_SCOPE_FAILED。
• Builder 阶段：runBuilderPhase 先取 builderSem，再从 BuilderRegistry 查 FunctionCallConfig.Type。查不到直接以 BUILDER_NOT_FOUND 收敛。若 builder 实现了 StreamingCallBuilder 且 StreamBuild.Enabled，Builder 阶段只跑 PrepareStream，回送 CallStreamMarker，core 输出 StartCallPhase{Streaming: true}，随后 runCallStreamPhase 在 scanSem 下启动生产 goroutine。
• Calls 阶段：runCallPhase 取 executorSem，在锁外用 dispCore.PrepareCallPhaseWithDigests 准备 call 上下文，再投递 evCallPhaseStarted 给 actor。call 级调度、重试、subcall 归 Call 执行子系统。actor 收到 ConvergeTaskCalls 命令时调用 HandleTaskPhaseFinished(PhaseCalls)。
• Plugin 阶段：runWriterPhase 取 pluginSem，按 ResultHandler.Type 查 PluginRegistry。未配置 ResultHandler 时不执行任何插件，直接以成功回送。
• 终态：executePhaseResult 对仍 in-flight 的 call 发 CancelCall，清理 dispatcher 状态，关闭 TaskIOScope，删除 taskRuntime，写 task_finished 日志，并通过 OnTaskTerminal 通知订阅者。slot 释放已在 WorkerCore.convergeTask 内同步完成。
• 定时器：app.Run 每 500ms 调一次 orch.TickTimers()，它依次处理 slot lease 过期、task 超时、Watch grace 到期和结果缓存清理，然后 SyncExecutorSnapshots、CheckWedgedExecutors、execAdapter.CheckHeartbeatTimeouts。
• shadow 转发：配置 ShadowSubmitTargetAddr 后，SubmitTaskShadowForwarder 在主提交成功后按采样率把请求镜像到另一 Worker；接收端通过 isShadowSubmit 识别并剥离 ResultHandler。

​

状态与生命周期

​

slot 状态机

• HandleRequestTaskSlot 先做幂等检查（同 taskId 仍活跃则返回稳定快照；已终态且在保留窗口内则返回终态），再检查 freeCapacity() = TaskSlots - (ALLOCATED + RUNNING)，最后线性扫描第一个 FREE slot 分配。
• SubmitTask 不带 slotId 时，core 内联执行一次 HandleRequestTaskSlot，TTL 用 DefaultTTLMs。
• 若 lease 在 SubmitTask 与 HandleTaskActivationPrepared 之间过期且 slot 被回收，core 把该 task 以 SLOT_LOST 收敛为 FAILED。

​

task 可见状态与内部阶段

• Builder 失败 → FAILED，失败码取 outcome 或 BUILDER_FAILED。成功 → Calls。
• Calls 失败 → FAILED。成功 → 总是输出 StartWriterPhase。
• Plugin 结束 → 终态；失败码取 outcome 或 PLUGIN_FAILED，PluginResults 写入 ExecuteResult。
• 乱序的阶段事件（tc.Phase != phase）被静默丢弃。

​

时间语义

​

启动与关闭顺序

​

配置

​

扩展点

• 新增 Call Builder：在 internal/plugin/callbuilder/ 实现 cbtypes.CallBuilder（可选 StreamingCallBuilder），在 cbtypes 中定义 CallBuilderName，然后在 internal/worker/app/app.go 的 newBuilderRegistry switch 中加一个 case，并把名字加进需要它的 cmd/*/main.go Profile 的 Builders。未列入 Profile 的 builder 不会注册，task 会以 BUILDER_NOT_FOUND 失败。详见 Plugin 系统。
• 新增 Writer Plugin：实现 evtypes.WriterPlugin，在 newPluginRegistry 中加 case，并加进 Profile 的 Plugins。若插件依赖特定后端地址，在 validateProfileRuntimeConfig 加校验。
• 新增入口 / 部署形态：只写一个新的 cmd/<name>/main.go，声明 app.Profile 并调用 app.Run。不要在入口里手搓 wiring。Profile.Validate 目前只接受 block 和 bundle 两种 Deployment。
• 新增配置项：在 WorkerFullConfig 加字段和默认值，在 app.Run 加 envOverride*，必要时在 Validate 加约束。
• 改 slot / 阶段 / 幂等语义：只改 internal/worker/core/worker.go，并在 internal/worker/core/worker*_test.go 补 core unit 测试。adapter 不应出现新的状态判定。
• 改阶段执行副作用：orchestrator_phases.go。任何触碰 dispCore 的新路径必须遵守 orchestrator_actor.go 文件头的 I1–I5 不变量，actor_invariants_test.go 会拦截违规。
• 新增对外 RPC：改 api/grpc/worker/worker.proto，重新生成 workerpb，在 grpc_server.go 加 handler，pb 转换放 wire.go。

​

测试

• core unit（internal/worker/core/）：worker_test.go、worker_activation_test.go、worker_watch_test.go 覆盖 WorkerCore；dispatcher_*_test.go 覆盖 DispatcherCore。不引入 transport 或进程。
• adapter integration（internal/worker/adapters/）：orchestrator_*_test.go（lifecycle、phases、dispatch、stream、budget、deadline_repro、subcall、send_failure 等切片）、grpc_server_test.go、stream_subscriber_test.go、etcd_publisher_test.go、actor_invariants_test.go。使用 fake executor adapter，不起 Python。internal/worker/app/*_test.go 覆盖 Profile、配置解析和 etcd 注册装配。
• process e2e（cmd/worker/worker_process_*_test.go、cmd/bundle_worker/*_test.go）：拉起真实 worker 进程和 Python executor，验证对外 gRPC contract（activation、watch、shutdown、timing、subcall、io、function code 等切片）。

# core + adapters + app（推荐带 -race）
go test -race ./internal/worker/...

# process E2E（需先安装 Python 依赖）
uv sync --project python
go test -v -timeout 300s ./cmd/worker/...
go test -v -timeout 120s ./cmd/bundle_worker/...

​

相关文档

• docs/specs/worker.md：Worker 详细设计。协议、状态模型、执行流程、Executor Pool Manager、心跳与失败语义。
• docs/specs/architecture.md §4.2.1 / §4.2.2：Worker 在整体架构中的职责，以及阶段级准入、task 内放量窗口、backend admission 三层并发控制。
• docs/specs/worker-test-organization.md：Worker 测试组织标准。
• docs/specs/2026-07-16-task-owner-lease-and-cancellation.md：Watch 断连回收方案（watchDisconnectGraceMs）的评审记录。
• docs/specs/2026-07-28-registry-prefix-migration.md：workerRegistryPrefix 的格式与迁移。