> ## Documentation Index
> Fetch the complete documentation index at: https://docs.blockx.chaintable.com/llms.txt
> Use this file to discover all available pages before exploring further.

# 提交 task 示例

> 用 blockx-py 提交 task 的可运行示例：一次性计算并取回结果、按区块写 block 表、写普通表、按 bundle 回填；以及不用 SDK、直接调 WorkerService gRPC 的 Go 写法。

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

## 前置

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

    ```bash theme={null}
    cd examples/local_test_service
    uv sync                      # 装 blockx-py 与 blockdb-py
    export PROXY_SOCKET_PATH=/tmp/blockx-proxy.sock
    ```
  </Tab>

  <Tab title="线上环境">
    在 notebook 或函数运行时里，`PROXY_SOCKET_PATH` 由平台注入，直接 `import blockx` 即可。自己装包时：

    ```bash theme={null}
    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
    ```
  </Tab>
</Tabs>

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

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

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

```python theme={null}
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：

```python theme={null}
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`：

```python theme={null}
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 实例化和提交：

```python theme={null}
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：

```python theme={null}
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()
```

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

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

```python theme={null}
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](/concepts/task-and-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` 决定是否重投，不要从错误文本推断：

```python theme={null}
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：

```go theme={null}
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 小时）                             |

## 相关文档

* [快速开始](/quickstart)：本地起 worker 与 proxy 的最短路径。
* [Task 与 Call](/concepts/task-and-call)：`TaskInput` 字段、状态与失败码。
* [Call Builder 与 Writer Plugin](/concepts/builders-and-handlers)：每种 config 的 wire 形状。
* [协议与接口](/architecture/protocols)：`WorkerService` / `CoordinatorService` 的完整字段。
* [测试](/development/testing)：process / system E2E 里的提交辅助函数。
* blockx-py 仓库 `docs/api_reference.md`：SDK 全部参数与 `Pipeline` 的行为细节。