Task 生命周期

task 是 BlockX 的最小调度和执行单元：Client 生成一个 taskId，向 Worker 提交一份 TaskInput，Worker 在本地跑完 Builder → Calls → Plugin 三个阶段后收敛出一份 TaskResult。本页只讲这条主线，各组件细节见 Worker、Coordinator、Call 执行子系统、Plugin 系统；整体架构见架构总览。主链路依赖以下约束（另见 blockx 仓库 docs/specs/architecture.md §2.1）：

以 task 为中心：状态共享、读缓存、超时都锚定到 task。
单 Worker 执行：一个 task 只在一个 Worker 上跑，Worker 不感知 Coordinator，只暴露 RequestTaskSlot / SubmitTask。
task 内 call 全并行：call 之间没有顺序依赖，由 dispatcher 按窗口逐步放量。
同一 task 函数版本固定：激活时 pin 住一个 taskCodeEpoch（TaskContext.TaskCodeEpoch），期间的代码更新只影响后续 task。
函数只读：用户函数不直接写库，所有写入走 Writer Plugin。
不做 task 级自动重试：Worker 只对单个 call 做有界 attempt 重试（CallRetryPolicy），是否重投 task 由 Client 看 TaskResult.executeResult.retryable 决定。
不支持取消：只有 task 超时（taskTimeoutMs）和可选的 Watch 断连回收。

Task 模型

对外提交的结构是 gRPC TaskInput（api/grpc/worker/worker.proto），Go 侧对应 commontypes.TaskInput（internal/common/types/types.go）：

字段	类型	说明
`task_id`	`string`	Client 自行生成的稳定唯一 ID（建议 uuid4）。幂等键。
`function_call_config`	`{type, config}`	`type` 选 Call Builder，`config` 是不透明 JSON，由对应 Builder 自行解析。
`result_handler`	`{type, config}`	可选。`type` 选 Writer Plugin。每个 task 至多一个。
`task_timeout_ms`	`int64`	可选。`0` 用 Worker 默认；负数拒绝；超过 `maxTaskTimeoutMs` 也拒绝。

当前注册的 Builder type（internal/plugin/callbuilder/types/types.go）：BlockTableCallConfig（dbScan）、InputsCallConfig、BlockBundleCallConfig（bundleScan）。当前注册的 Writer Plugin type（internal/plugin/event/types/types.go）：BlockTableWriteHandler、ReturnValueHandler、NormalTableWriteHandler。一个最小的 JSON 形态示例（按 commontypes.TaskInput 和 call_list.CallListCallConfigBuilderDecl 的字段整理，不是仓库内 fixture）：

{
  "task_id": "3f2b1c9e-8a4d-4c1e-9b0a-6f7e5d4c3b2a",
  "functionCallConfig": {
    "type": "InputsCallConfig",
    "config": {
      "function": "hello_world",
      "callList": [["ct2"], ["blockx"]]
    }
  },
  "resultHandler": {
    "type": "ReturnValueHandler",
    "config": {}
  },
  "taskTimeoutMs": 30000
}

TaskInput 顶层不带区块上下文。区块信息由具体 Builder 的 config 携带（例如 dbscan.DBScanBuilderDecl.Block），再由 Builder 写进每个 call。 Worker 收到请求后，internal/worker/adapters/wire.go 的 submitTaskFromPB 把 proto 转成 commontypes.SubmitTaskRequest（顺带校验 config 是合法 JSON、task_timeout_ms >= 0）；internal/worker/adapters/payload.go 的 parseTaskPayload 再包成 commontypes.TaskCtx：

// TaskCtx carries the immutable context injected when a task is activated.
type TaskCtx struct {
	TaskInput

	Stage Stage `json:"stage"`
	DebugMode bool `json:"debugMode,omitempty"`
	Deadline time.Time          `json:"deadline"`
	Ctx      context.Context    `json:"-"`
	Cancel   context.CancelFunc `json:"-"`
}

TaskCtx 是 Builder、Writer Plugin 和 IO scope 共享的最小上下文；Stage 在阶段推进时被 adapter 改写为 builder / executor / plugin。WorkerCore 只把它作为不透明指针挂在 TaskContext.AdapterTaskCtx 上。

端到端时序

下面按主链路的 16 步展开（另见 docs/specs/architecture.md §3）。

Client 构造 task

Client 生成 task_id，准备 TaskInput。BlockX 不负责 task 生成、DAG 编排或定时触发。

Client 向 Coordinator 申请选址（可选）

调用 CoordinatorService.ReserveWorkerSlot(task_id)（api/grpc/coordinator/coordinator.proto）。Coordinator 从 etcd watch 到的心跳里重建 Worker 视图（CoordinatorCore.ApplyWorkerUpdate），用 SelectCandidate 挑候选。不经过 Coordinator 时，Client 可以直接对 Worker 调 RequestTaskSlot，或直接 SubmitTask 让 Worker 自动补申请。

Coordinator 向 Worker 申请 slot

CoordinatorServer.ReserveWorkerSlot（internal/coordinator/adapters/grpc_server.go）串行尝试候选 Worker，对每个调 WorkerService.RequestTaskSlot(task_id, ttl_ms)，ttl_ms 取 CoordinatorCore.SlotTTLMs()。拒绝就换下一个候选；RPC 超时属于”可能已分配”，会对同一 Worker 同一 task_id 有限次重试。

Worker 原子分配 slot

WorkerCore.HandleRequestTaskSlot（internal/worker/core/worker.go）在 o.mu 下检查 freeCapacity()，把一个 FREE slot 切到 ALLOCATED，写入 LeaseDeadlineMs = now + ttlMs，返回 slot_id 和 state=ALLOCATED。

Coordinator 回传 worker_addr + slot_id

Coordinator 不保存 task payload、不保存结果、不写 slot 真相到 etcd；slot_id 对它只是不透明字符串。

Client 提交 SubmitTask

Client 直连 worker_addr，调 WorkerService.SubmitTask(task, slot_id)。Orchestrator.SubmitTask（internal/worker/adapters/orchestrator.go）先算有效 deadline（WorkerCore.EffectiveTaskDeadlineMs）、解析 payload，再交给 WorkerCore.HandleSubmitTask。

Worker 校验并激活

HandleSubmitTask 依次做：ready/draining 检查；同 taskId 已在 w.tasks 则返回 RUNNING，已有保留结果则返回终态；没带 slot_id 则内联执行一次 HandleRequestTaskSlot；校验 slot 存在、处于 ALLOCATED、绑定的是同一 taskId；通过后创建 TaskContext（Phase = Preparing）并返回命令 PrepareTaskActivation。gRPC 层此时就返回 state=RUNNING。

adapter 完成激活

Orchestrator.activateAndRun（internal/worker/adapters/orchestrator_phases.go）在 goroutine 里解析当前 taskCodeEpoch、pin 住 Function Code View 快照，然后回送 WorkerCore.HandleTaskActivationPrepared：slot 切到 RUNNING，Phase 推进到 Builder，输出 StartBuilderPhase。随后创建 TaskIOScope（WorkerIOScope.NewTaskScope）和 adapter 侧 taskRuntime。任一步失败走 HandleTaskActivationFailed，failureCode 为 ACTIVATION_FAILED 或 IO_SCOPE_FAILED。

Builder 阶段生成 call list

runBuilderPhase 先拿 builderSem，按 FunctionCallConfig.Type 从 BuilderRegistry 查 Builder，调 CallBuilder.Build(ctx, taskCtx, io) 得到 cbtypes.CallList。结果通过 HandleTaskPhaseFinished(PhaseBuilder, ...) 交回 core，core 输出 StartCallPhase。

Calls 阶段并行执行

runCallPhase 拿 executorSem，调 DispatcherCore.PrepareCallPhaseWithDigests 建立 TaskDispatchState，之后由 dispatch loop 按窗口把 call 通过 UDS ExecuteCall 投给 Python Executor。Executor 内的 SDK IO 请求以 CallWaiting 回到 Worker，由 IO scope 处理后 ResumeCall；子函数调用同样回到 dispatcher 重新绑定。详见 Call 执行子系统与 IO 访问子系统。

汇聚 outputs

所有根 call 终态后 DispatcherCore.checkConvergeTask 产出 ConvergeTaskCalls{Success, FailureCode, Outputs}；Outputs 只含成功根 call 的返回值，subcall 返回值不进入。adapter 把它转成 PhaseOutcome 调 HandleTaskPhaseFinished(PhaseCalls, ...)。

Plugin 阶段落库

core 输出 StartWriterPhase{Outputs}；runWriterPhase 拿 pluginSem，若 ResultHandler 非空则从 PluginRegistry 查插件并调 WriterPlugin.Execute(ctx, taskCtx, outputs, io)，产出 []PluginResult。

收敛终态

HandleTaskPhaseFinished(PhasePlugin, ...) 调 convergeTask：Phase = Terminal，taskIndex 从 ActiveSlotID 切成 Result，RetainedUntil = now + ResultRetentionMs，同一临界区内 releaseSlot 把 slot 回收到 FREE，从 w.tasks 删除，输出 ConvergeTask。adapter 的 executePhaseResult 取消未完成的 call、关闭 TaskIOScope、删除 taskRuntime，最后触发 OnTaskTerminal。

Client 获取结果

SubmitTask 不返回 TaskResult。Client 用 WatchTasks 流收终态 TaskUpdate，或轮询 GetTaskResult。OnTaskTerminal 接到 SubscriptionManager.NotifyTerminal（internal/worker/adapters/stream_subscriber.go）向所有订阅该 taskId 的流推一次终态。

心跳反映容量

WorkerCore.DeriveHeartbeat 从 slot 表派生 reservedSlots / runningTasks，EtcdPublisher.RunHeartbeatLoop 周期性写 etcd；Coordinator 由此更新视图。

slot 与准入

RequestTaskSlot 与 SubmitTask 是同一套 Worker 协议的两步；调用方是 Coordinator 还是 Client 对 Worker 没有区别。

RequestTaskSlot(task_id, ttl_ms) 只占位，不执行。ttl_ms 必须为正（<= 0 返回 InvalidArgument），只控制 ALLOCATED 的自动回收，与 task 超时无关。TickTimers 每 500ms 扫一次 ExpiredSlots，过期 slot 由 HandleSlotLeaseExpired 回收，同时清掉还停在 Preparing 的 task 记录。
SubmitTask(task, slot_id?)：带 slot_id 时必须命中一条 ALLOCATED 且绑定同一 taskId 的记录，否则 InvalidSlot；不带时 Worker 用 DefaultTTLMs 内联申请一次，没容量返回 NoSlot。
幂等：同一 taskId 处于 ALLOCATED / RUNNING 时，RequestTaskSlot 返回同一个 slot_id；已进入 RUNNING 或结果仍在保留窗口内时，SubmitTask 返回稳定的当前状态，不会再启动一次执行。旧执行结束且结果过期后，同一 taskId 可以重新分配 slot。
没有 release 接口：拿到 slot 后放弃提交只能等 TTL 回收。

准入错误与执行终态的边界（另见 docs/specs/architecture.md §4.3.1 末尾）：slot 还没从 ALLOCATED 切到 RUNNING 之前的所有失败都以 gRPC status 返回，不进 TaskResult；一旦 HandleSubmitTask 创建了 TaskContext，之后的激活失败、Builder / Calls / Plugin 失败、超时都收敛为 TaskResult 终态。

错误码（`commontypes.ErrorCode`）	gRPC code（`workerCodeToGRPC`）	何时出现
`InvalidArgument`	`InvalidArgument`	`taskId` 为空、`ttl_ms <= 0`、`task_timeout_ms` 非法、`config` 不是合法 JSON
`NoSlot`	`ResourceExhausted`	本地无空闲 slot
`InvalidSlot`	`FailedPrecondition`	`slot_id` 不存在、非 `ALLOCATED`、绑定了别的 `taskId`
`Unavailable`	`Unavailable`	Worker 未 ready（还没有 Executor 心跳）或正在 drain
`NotFound`	`NotFound`	`GetTaskResult` 查不到该 `taskId`

slot 表、taskIndex 和心跳派生细节见 Worker；Coordinator 的候选选择、熔断和不确定态重试见 Coordinator。

三个阶段

WorkerCore 只裁决”是否进入下一阶段、何时写终态”；实际执行都在 adapter。阶段推进靠命令 / 事件往返：core 返回 StartBuilderPhase → adapter 执行 → 回送 HandleTaskPhaseFinished(PhaseBuilder, outcome) → core 返回 StartCallPhase，依此类推（internal/worker/core/types.go 定义了全部命令类型）。

阶段	并发	阶段级准入（`core.AdmissionConfig`，默认）	失败对 task 的影响
Builder	串行执行一次；IO 带 Builder retry scope，由 IO core 对可重试错误做有界重试	`builderSem`（`BuilderSlots = 16`）	立即 `FAILED`，`BUILDER_FAILED` / `BUILDER_NOT_FOUND`，`retryable` 取自 IO 错误
Calls	全并行；每 task 同时 in-flight 上限 `TaskMaxInflightCalls = 1024`；单 call 有界 attempt 重试	`executorSem`（`ExecutorTaskSlots = 32`；空 call list 不占）	`CallFailurePolicy` 目前只有 `fastFail`：任一根 call 终态失败即停止派发、丢弃 outputs，收敛为 `CALL_FAILED`（`retryable=false`）；输出超 `MaxCollectedOutputBytes` 为 `OUTPUT_BYTES_EXCEEDED`
Plugin	串行执行一次；IO 带 Result retry scope	`pluginSem`（`PluginSlots = 16`）	`FAILED`，`PLUGIN_FAILED` / `PLUGIN_NOT_FOUND`，`retryable` 由 `PluginError.Retryable` 决定（未分类错误默认可重试）

其他要点：

每个阶段等准入时如果 task ctx 被取消（例如已超时收敛），以 CANCELLED 失败；进入 Calls 前还会先检查 deadline，已过期直接按 TIMED_OUT 收敛，不再占 executorSem。
Calls 阶段成功后 core 总是输出 StartWriterPhase；ResultHandler 为空时 runWriterPhase 仍会取 pluginSem，只是不执行插件、pluginResults 为空。
流式 Builder（StreamingCallBuilder，仅 BlockBundleCallConfig，StreamBuildConfig.Enabled 默认关闭）把 Builder 阶段缩成 PrepareStream，扫描放到 Calls 阶段由 scanSem（ScanSlots = 8）约束；生产中途失败会导致部分 call 已执行。
超时由 TickTimers 扫 TimedOutTasks 后调 HandleTaskTimedOut 强制收敛为 TIMED_OUT（retryable=true），不强求立即中断 Executor 里正在跑的 call；in-flight call 会收到 CancelCall。

task 对外可见状态只有四个（commontypes.TaskState），内部阶段（core.TaskPhase：Preparing / Builder / Calls / Plugin / Terminal）都折叠在 RUNNING 里：

结果与查询

Worker core 收敛任务并记录终态结果。结果索引独立于 slot 复用保留该结果。Worker adapter 通过单次查询与 Watch 更新提供同一结果。只有启用断连处理且 Watch 曾连接时，HandleWatchDetached 才启动截止时间。定时器可以使任务以 WATCH_DISCONNECTED 收敛；默认禁用该行为。调用方契约与完成边界定义在执行语义与结果。完整结果字段与错误见参考。

时间语义速查

名字	默认值	含义	定义位置
slot TTL（Coordinator 侧）	`SlotTTLMs = 5000`	Coordinator 调 `RequestTaskSlot` 传的 `ttl_ms`，env `COORDINATOR_SLOT_TTL_MS`	`internal/coordinator/core/coordinator.go`
slot TTL（Worker 内联）	`DefaultTTLMs = 5000`	`SubmitTask` 不带 `slot_id` 时自动申请用的 TTL	`internal/worker/core/config.go`
task timeout 默认	`TaskDeadlineMs = 300000`	`task_timeout_ms = 0` 时的有效超时，从 Worker 接受提交起算；env `TASK_DEADLINE_MS`	`internal/worker/core/config.go`
task timeout 上限	`MaxTaskTimeoutMs = 8h`（`DefaultMaxTaskTimeoutMs`）	超过即 `InvalidArgument`；Function Code View 快照保留窗口必须不小于它；env `MAX_TASK_TIMEOUT_MS`	`internal/worker/core/config.go`、`internal/worker/app/config.go`
结果保留	`ResultRetentionMs = 300000`	终态结果可查询 / 可幂等命中的窗口；env `RESULT_RETENTION_MS`	`internal/worker/core/config.go`
心跳间隔	`HeartbeatIntervalMs = 200`	Worker 写 etcd 的频率；生产建议 2000，env `WORKER_HEARTBEAT_INTERVAL`	`internal/worker/app/config.go`
心跳超时（Coordinator）	`HeartbeatTimeoutMs = 10000`	超过即视为 Worker 失联，不再路由	`internal/coordinator/core/coordinator.go`
Watch 断连 grace	`WatchDisconnectGraceMs = 0`（关闭）	最后一条 Watch 断开到判 `WATCH_DISCONNECTED` 的窗口；生产建议 120000，env `WATCH_DISCONNECT_GRACE_MS`	`internal/worker/core/config.go`
gRPC keepalive	`GRPCKeepaliveTimeMs = 30000` / `GRPCKeepaliveTimeoutMs = 10000`	决定静默断连多久后被确认，从而影响 Watch grace 起点	`internal/worker/app/config.go`
定时扫描周期	500ms	`TickTimers`：slot 过期、task 超时、Watch grace、结果清理	`internal/worker/app/app.go`
单 call 执行预算	`CallDeadlineMs = 5000`	只计 CPU + IO backend 时间，不含排队；与 task 超时独立	`internal/worker/core/config.go`

WorkerCore.EffectiveTaskDeadlineMs 是这几个值的收口：task_timeout_ms 为负或超过 MaxTaskTimeoutMs 都直接返回 InvalidArgument，为 0 时取 TaskDeadlineMs。

不适用场景

BlockX 的 task 模型面向”单行状态、轻量函数、一次性写入”，以下场景不适合（docs/specs/architecture.md §5）：

依赖同一张表大量行的计算：大范围聚合、窗口统计、K 线。
需要任意时间窗口状态或大规模有状态计算。
强依赖流式引擎 Exactly Once 状态恢复。
普通在线表的任意增量流式计算。

非 onchain 表的流式需求，折中做法是 Writer Plugin 同时写在线表和 Kafka，业务侧监听 topic 后再按需提交新 task。

开始使用

核心概念

使用指南

参考

架构

组件

开发指南

Task 模型

端到端时序

slot 与准入

三个阶段

结果与查询

时间语义速查

不适用场景

相关文档

​Task 模型

​端到端时序

​slot 与准入

​三个阶段

​结果与查询

​时间语义速查

​不适用场景

​相关文档

Task 模型

端到端时序

slot 与准入

三个阶段

结果与查询

时间语义速查

不适用场景

相关文档