提交 task 示例

这一页把 Task 与 Call 和 Call Builder 与 Writer Plugin 落成代码。前四个示例用 blockx-py（blockx 仓库 python/pyproject.toml 锁定 v0.1.72），最后一节给出不依赖 SDK、直接调 WorkerService 的 Go 写法。

前置

本地 worker
线上环境

按快速开始启动 worker 和 examples/local_test_service/local_proxy.py。示例 1 在本地就能跑通；示例 2 到 4 要读写 BlockDB 表，需要把 BLOCKDB_*_ADDR 指向可访问的 BlockDB，并且不设 STUB_BLOCKDB_DELAY_MS。

cd examples/local_test_service
uv sync                      # 装 blockx-py 与 blockdb-py
export PROXY_SOCKET_PATH=/tmp/blockx-proxy.sock

在 notebook 或函数运行时里，PROXY_SOCKET_PATH 由平台注入，直接 import blockx 即可。自己装包时：

pip install git+https://github.com/Chaintable/blockx-py.git@v0.1.72
pip install git+https://github.com/Chaintable/blockdb-py.git@v0.1.26

blockx-py 只通过 PROXY_SOCKET_PATH 指定的 UDS proxy 连接。没设这个变量会直接 RuntimeError，不会静默连到别处。

示例 1：一次性计算并取回结果

InputsCallConfig + ReturnValueHandler。这就是 examples/local_test_service/submit.py：

import os
os.environ.setdefault("PROXY_SOCKET_PATH", "/tmp/blockx-proxy.sock")

from blockx import Function, InputsCallConfig, ReturnValueHandler, TaskBuilder

# 入口函数名是 `_`；每次调用收到 callList 的一行作为位置参数。
source_code = """
from blockx import LocalTestService

def _(name):
    return {"message": LocalTestService().helloworld(name)}
"""

task = TaskBuilder.build(
    InputsCallConfig(
        func=Function(source_code=source_code),
        callList=[["ct2"], ["blockx"]],   # 两行 → 函数被调用两次
    ),
    ReturnValueHandler(),
)

# timeout 是 task 在 worker 上的执行上限（秒），不是客户端等待时长。
result = task.submit(timeout=30)
if not result.task_result.success:
    raise RuntimeError(result.task_result.failure_code)
print(result.handler_result)
# [{'message': 'hello, ct2'}, {'message': 'hello, blockx'}]

要点：

LocalTestService 是 worker 内置的示例能力后端，本地不需要 BlockDB。换成任何自包含的纯计算函数也一样能跑。
结果按各次调用的完成序收集，不保证与 callList 行序一致。需要对应关系时让返回值带上入参。
带 ReturnValueHandler 的 task 被 SDK 路由到 block 集群，因为 bundle worker 没有注册这个 handler。

示例 2：按区块计算并写入 block 表

BlockTableCallConfig + BlockTableWriteHandler(block=...)。订阅源表的新区块，每个区块提交一个 task：

from blockdb import BlockTable, Subscribe
from blockx import SOURCE_ROW, BlockTableCallConfig, BlockTableWriteHandler, TaskBuilder

SOURCE = "chain.trace.eth"
TARGET = "token.token_transfer.eth"


def trace_to_token_transfer(chain_id, record):
    # 函数体要自包含：常量写在函数里，模块在函数里 import。
    NATIVE = "0x000000000000000000000000000000000000eeee"
    try:
        value = float(record.get("value") or 0)
    except (TypeError, ValueError):
        return None
    if value <= 0 or not record.get("tx_id"):
        return None                      # 返回 None：这一行不写
    return {
        "id": record["id"],
        "token_id": NATIVE,
        "from_addr": record["from_addr"],
        "to_addr": record["to_addr"],
        "value": value,
        "tx_id": record["tx_id"],
    }


src = BlockTable(SOURCE)
for table_id, block in Subscribe(tables=[src]).listen():
    task = TaskBuilder.build(
        BlockTableCallConfig(
            block=block,
            triggerSources=[{
                "table": src,
                "func": trace_to_token_transfer,     # callable 会被抓源码并补上 `_ = trace_to_token_transfer`
                "params": ["eth", SOURCE_ROW],       # SOURCE_ROW 的位置换成源表的一行
            }],
        ),
        BlockTableWriteHandler(targetTable=TARGET, block=block),
    )
    res = task.submit(timeout=120)
    if res.task_result.success:
        print(f"h={block.height} written_rows={res.handler_result['written_rows']}")
    else:
        print(f"h={block.height} failed: {res.task_result.failure_code} retryable={res.task_result.retryable}")

要点：

传 callable 时 SDK 用 inspect.getsource 抓函数体并补一行 _ = trace_to_token_transfer；也可以直接传 Function(source_code=...) 或注册函数 ID 字符串。
params 决定函数的位置参数：这里 chain_id="eth"，record 是源表的一行 dict。
BlockTableWriteHandler 只投影返回 dict 中属于目标表的列，written_rows 是这次写入的行数。
triggerSources 里可以加 "operator": filter(...) 在 Builder 阶段先过滤、去重，减少 call 数量。

示例 3：写入普通表

把示例 2 的 handler 换成 NormalTableWriteHandler，目标就从 block 表变成普通表（L1）。它不需要 block：

from blockx import NormalTableWriteHandler

# 普通 upsert：按主键 id 覆盖
handler = NormalTableWriteHandler(targetTable="token.token_transfer.eth")

# 条件更新：只有新值在 first_seen_at 上更小才覆盖，乱序 / 重放写入天然幂等
handler = NormalTableWriteHandler(
    targetTable="token.token_transfer.eth",
    condition=("first_seen_at", "UpdateIfSmaller"),   # 或 "UpdateIfLarger"
)

condition.column 不能是主键 id，非法值在构造时就抛 ValueError。这个 handler 不带区块上下文，同一个实例可以直接交给示例 4 的回填。

示例 4：按 bundle 回填

历史回填按 bundle（1000 个区块一段）提交 task，用 BlockBundleCallConfig 读 bundle 分区的 parquet，用 BlockTableWriteHandler(block_bundle=...) 覆盖写目标表的同一个 bundle。backfill_block_bundles 帮你按区间逐 bundle 实例化和提交：

from blockx import (
    SOURCE_ROW,
    BlockBundleCallConfigTemplate,
    BlockTableWriteHandler,
    backfill_block_bundles,
)

config = BlockBundleCallConfigTemplate(triggerSources=[{
    "table": "chain.trace.eth",
    "func": trace_to_token_transfer,      # 与示例 2 同一个函数
    "params": ["eth", SOURCE_ROW],
}])
handler = BlockTableWriteHandler(targetTable="token.token_transfer.eth")   # 模板态：不带 block / block_bundle

summary = backfill_block_bundles(18001, 18100, config, handler)
print(summary)                            # 成功 / 失败 bundle 数、失败码分布、written_rows

只发一个 bundle 时直接构造带号码的 config 和 handler：

from blockx import BlockBundleCallConfig, BlockTableWriteHandler, TaskBuilder

task = TaskBuilder.build(
    BlockBundleCallConfig(number=18001, triggerSources=[{
        "table": "chain.trace.eth",
        "func": trace_to_token_transfer,
        "params": ["eth", SOURCE_ROW],
    }]),
    BlockTableWriteHandler(targetTable="token.token_transfer.eth", block_bundle=18001),
)
result = task.submit()

BlockTableWriteHandler 的形状必须跟集群走：bundle 集群要 block_bundle=，block 集群要 block=。发错形状服务端要等所有 call 算完才在 Writer 阶段拒绝，所以 SDK 在 submit() 前就校验并抛 ValueError。

日常使用推荐 blockx-py 的上层封装 Pipeline：一份 triggers + target_table，backfill(block_start, block_end) 做历史回填，update() 先补齐缺口再转实时订阅。bundle 换算、集群路由、slot 退避和多表对齐都由它处理：

from blockx import Pipeline, Trigger, SOURCE_ROW

pipe = Pipeline(
    triggers=[Trigger(table="chain.trace.eth", func=trace_to_token_transfer, params=["eth", SOURCE_ROW])],
    target_table="token.token_transfer.eth",
)
pipe.backfill(block_start=18_000_001, block_end=18_100_000)
# pipe.update()   # 长驻：补齐缺口后转实时

读结果

task.submit() 返回 TaskResult：

字段	类型	含义
`task_result.success`	`bool`	task 是否成功。`True` 只代表 worker 受理并跑完；写表是后台异步的
`task_result.failure_code`	`str`	失败码，见 Task 与 Call
`task_result.retryable`	`bool`	是否值得用同一份 task 重投
`handler_result`	`dict / list / None`	Writer Plugin 的返回：`ReturnValueHandler` 是输出数组，写表 handler 是 `{"written_rows": n, "finish_time": ...}`
`raw`	gRPC 响应	`GetTaskResultResponse` 或适配后的 `TaskUpdate`

按 retryable 决定是否重投，不要从错误文本推断：

res = task.submit(timeout=60)
if not res.task_result.success:
    if res.task_result.retryable:
        res = TaskBuilder.build(task.call_config, task.handler).submit(timeout=60)   # 新 task_id
    else:
        raise RuntimeError(f"{res.task_result.failure_code}: {res.error}")

Task.task_id 每次 build 都是新的 uuid；排查时把它记到日志里，worker 的 task finished 日志按它检索。

不用 SDK：直接调 gRPC

Go 侧的贡献者经常需要在测试或工具里直接对 Worker 提交 task。协议是 api/grpc/worker/worker.proto 的 WorkerService，Go 绑定在 api/grpc/worker/workerpb。下面的程序对本地 worker 走 RequestTaskSlot → SubmitTask → WatchTasks 三步，提交的是示例 1 同样的 task：

package main

import (
	"context"
	"fmt"
	"io"
	"log"
	"time"

	"github.com/google/uuid"
	"google.golang.org/grpc"
	"google.golang.org/grpc/credentials/insecure"

	workerpb "github.com/Chaintable/blockx/api/grpc/worker/workerpb"
)

func main() {
	conn, err := grpc.NewClient("passthrough:///127.0.0.1:9221",
		grpc.WithTransportCredentials(insecure.NewCredentials()))
	if err != nil {
		log.Fatal(err)
	}
	defer conn.Close()
	client := workerpb.NewWorkerServiceClient(conn)
	ctx, cancel := context.WithTimeout(context.Background(), time.Minute)
	defer cancel()

	taskID := uuid.NewString()

	// 1. 预占 slot。生产环境由 Coordinator 的 ReserveWorkerSlot 完成这一步并返回 worker_addr + slot_id。
	slot, err := client.RequestTaskSlot(ctx, &workerpb.RequestTaskSlotRequest{TaskId: taskID, TtlMs: 5000})
	if err != nil {
		log.Fatal(err)
	}

	// 2. 提交 task。config 是不透明 JSON，由对应的 Builder / Plugin 解析。
	submit, err := client.SubmitTask(ctx, &workerpb.SubmitTaskRequest{
		SlotId: slot.SlotId,
		Task: &workerpb.TaskInput{
			TaskId: taskID,
			FunctionCallConfig: &workerpb.FunctionCallConfig{
				Type: "InputsCallConfig",
				Config: []byte(`{"function":"","code":{"sourceCode":"def _(name):\n    return {\"message\": f\"hello, {name}\"}\n"},"callList":[["ct2"],["blockx"]]}`),
			},
			ResultHandler: &workerpb.ResultHandler{Type: "ReturnValueHandler", Config: []byte(`{}`)},
			TaskTimeoutMs: 30000,
		},
	})
	if err != nil {
		log.Fatal(err) // 准入错误：InvalidArgument / ResourceExhausted / FailedPrecondition / Unavailable
	}
	fmt.Println("submitted:", submit.State) // RUNNING

	// 3. 订阅终态。stream 先发一次当前快照，再在 task 收敛时发一条 is_terminal=true 的更新。
	stream, err := client.WatchTasks(ctx, &workerpb.WatchTasksRequest{TaskIds: []string{taskID}})
	if err != nil {
		log.Fatal(err)
	}
	for {
		upd, err := stream.Recv()
		if err == io.EOF {
			return
		}
		if err != nil {
			log.Fatal(err)
		}
		if !upd.IsTerminal {
			continue
		}
		fmt.Println("state:", upd.State)
		for _, p := range upd.Result.GetExecuteResult().GetPluginResults() {
			fmt.Printf("%s success=%v result=%s\n", p.PluginName, p.Success, p.Result)
		}
	}
}

RequestTaskSlot 可以省略：SubmitTask 不带 slot_id 时 Worker 会内联申请一次，没有容量返回 ResourceExhausted。
经 Coordinator 时先调 coordinator.v1.CoordinatorService.ReserveWorkerSlot(task_id)（api/grpc/coordinator/coordinator.proto），再用返回的 worker_addr 建连接、用 slot_id 提交。
同一 task_id 重复 SubmitTask 是幂等的：task 仍在跑返回 RUNNING，结果仍在保留窗口内返回终态。
测试里可以直接用 internal/testutil 的 RequestTaskSlot / SubmitTask / PollUntilTerminal 辅助函数，它们封装的就是上面的调用。

常见错误

现象	原因与处理
`RuntimeError: PROXY_SOCKET_PATH not set` 或 `UNAVAILABLE`	proxy 没起，或 socket 路径与 `local_proxy.py` 不一致
`ValueError: ... BlockTableWriteHandler carries no block_bundle`	handler 形状与目标集群不匹配，见示例 4 的说明
`InvalidArgument: plugin not registered: ReturnValueHandler`	用 `cluster="bundle"` 强制把带 `ReturnValueHandler` 的 task 发到了 bundle 集群；去掉 `cluster` 参数
一直打印排队提示 / `ResourceExhausted`	没有空闲 slot。SDK 会退避重试；直接调 gRPC 时自行退避
`FAILED` 且 `failure_code` 为 `CALL_FAILED`	函数抛异常，或审计 `enforce` 下代码被拒。看 worker 日志的 `call failed` 条目
`FAILED` 且 `failure_code` 为 `BUILDER_FAILED`	触发表不存在、`operator` 校验失败或读 BlockDB 失败
`FAILED` 且 `failure_code` 为 `PLUGIN_FAILED`，消息 `block deployment requires config.block`	`BlockTableWriteHandler` 形状发错了集群
`FAILED` 且 `failure_code` 为 `TIMED_OUT`	超过 `timeout` 或 worker 默认 `TASK_DEADLINE_MS`；`retryable=true`
`InvalidArgument: task_timeout_ms ...`	`timeout` 为负或超过 worker 的 `MAX_TASK_TIMEOUT_MS`（默认 8 小时）

开始使用

核心概念

架构

组件

开发指南

前置

示例 1：一次性计算并取回结果

示例 2：按区块计算并写入 block 表

示例 3：写入普通表

示例 4：按 bundle 回填

读结果

不用 SDK：直接调 gRPC

常见错误

相关文档

​前置

​示例 1：一次性计算并取回结果

​示例 2：按区块计算并写入 block 表

​示例 3：写入普通表

​示例 4：按 bundle 回填

​读结果

​不用 SDK：直接调 gRPC

​常见错误

​相关文档

前置

示例 1：一次性计算并取回结果

示例 2：按区块计算并写入 block 表

示例 3：写入普通表

示例 4：按 bundle 回填

读结果

不用 SDK：直接调 gRPC

常见错误

相关文档