Submit a task: examples

This page turns Task and Call and Call Builder and Writer Plugin into code. The first four examples use blockx-py (python/pyproject.toml in the blockx repository pins v0.1.72); the last section shows how to call WorkerService directly from Go without the SDK.

Prerequisites

Local worker
Production environment

Start the worker and examples/local_test_service/local_proxy.py as described in the Quickstart. Example 1 runs locally as-is; examples 2 to 4 read and write BlockDB tables, so point BLOCKDB_*_ADDR at a reachable BlockDB and leave STUB_BLOCKDB_DELAY_MS unset.

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

In a notebook or function runtime, PROXY_SOCKET_PATH is injected by the platform; just import blockx. To install the packages yourself:

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 only connects through the UDS proxy specified by PROXY_SOCKET_PATH. If the variable is not set, it raises RuntimeError immediately rather than silently connecting elsewhere.

Example 1: run once and get the result back

InputsCallConfig + ReturnValueHandler. This is 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

# The entry function is named `_`; each call receives one row of callList as positional arguments.
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"]],   # two rows → the function is called twice
    ),
    ReturnValueHandler(),
)

# timeout is the task's execution limit on the worker (seconds), not how long the client waits.
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'}]

Key points:

LocalTestService is a sample capability backend built into the worker; no BlockDB is needed locally. Any self-contained pure-computation function works just as well.
Results are collected in the completion order of the individual calls and are not guaranteed to match the row order of callList. When you need to map results back to inputs, have the return value carry the input.
A task with ReturnValueHandler is routed to the block cluster by the SDK, because the bundle worker does not register this handler.

Example 2: compute per block and write a block table

BlockTableCallConfig + BlockTableWriteHandler(block=...). Subscribe to new blocks of the source table and submit one task per block:

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):
    # The function body must be self-contained: define constants inside the function and import modules inside it.
    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                      # return None: this row is not written
    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,     # a callable has its source captured and `_ = trace_to_token_transfer` appended
                "params": ["eth", SOURCE_ROW],       # the SOURCE_ROW position is replaced with one row of the source table
            }],
        ),
        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}")

Key points:

When you pass a callable, the SDK captures the function body with inspect.getsource and appends the line _ = trace_to_token_transfer; you can also pass Function(source_code=...) directly, or a registered function ID string.
params determines the function’s positional arguments: here chain_id="eth", and record is one row of the source table as a dict.
BlockTableWriteHandler projects only the columns of the returned dict that belong to the target table; written_rows is the number of rows written this time.
You can add "operator": filter(...) to triggerSources to filter and deduplicate in the Builder phase first and reduce the number of calls.

Example 3: write a normal table

Swap the handler in example 2 for NormalTableWriteHandler and the target changes from a block table to a normal table (L1). It does not need block:

from blockx import NormalTableWriteHandler

# Plain upsert: overwrite by primary key id
handler = NormalTableWriteHandler(targetTable="token.token_transfer.eth")

# Conditional update: overwrite only when the new value is smaller in first_seen_at; out-of-order / replayed writes are naturally idempotent
handler = NormalTableWriteHandler(
    targetTable="token.token_transfer.eth",
    condition=("first_seen_at", "UpdateIfSmaller"),   # or "UpdateIfLarger"
)

condition.column cannot be the primary key id; an invalid value raises ValueError at construction time. This handler carries no block context, so the same instance can be handed directly to the backfill in example 4.

Example 4: backfill per bundle

Historical backfill submits one task per bundle (a segment of 1000 blocks): BlockBundleCallConfig reads the parquet of the bundle partition, and BlockTableWriteHandler(block_bundle=...) overwrites the same bundle of the target table. backfill_block_bundles instantiates and submits bundle by bundle over a range for you:

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

config = BlockBundleCallConfigTemplate(triggerSources=[{
    "table": "chain.trace.eth",
    "func": trace_to_token_transfer,      # the same function as in example 2
    "params": ["eth", SOURCE_ROW],
}])
handler = BlockTableWriteHandler(targetTable="token.token_transfer.eth")   # template form: no block / block_bundle

summary = backfill_block_bundles(18001, 18100, config, handler)
print(summary)                            # succeeded / failed bundle counts, failure code distribution, written_rows

To send a single bundle, construct the config and handler with the number directly:

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()

The shape of BlockTableWriteHandler must follow the cluster: the bundle cluster needs block_bundle=, the block cluster needs block=. If you send the wrong shape, the server only rejects it in the Writer phase after all calls have run, so the SDK validates before submit() and raises ValueError.

For day-to-day use, prefer blockx-py’s higher-level Pipeline: one set of triggers + target_table, backfill(block_start, block_end) for historical backfill, and update() to fill gaps first and then switch to a real-time subscription. Bundle conversion, cluster routing, slot backoff, and multi-table alignment are all handled for you:

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()   # long-running: fills gaps, then switches to real-time

Reading the result

task.submit() returns a TaskResult:

Field	Type	Meaning
`task_result.success`	`bool`	Whether the task succeeded. `True` only means the worker accepted it and ran it to completion; the table write is asynchronous in the background
`task_result.failure_code`	`str`	Failure code, see Task and Call
`task_result.retryable`	`bool`	Whether resubmitting the same task is worthwhile
`handler_result`	`dict / list / None`	The Writer Plugin’s return value: the output array for `ReturnValueHandler`, `{"written_rows": n, "finish_time": ...}` for table-writing handlers
`raw`	gRPC response	`GetTaskResultResponse` or an adapted `TaskUpdate`

Decide whether to resubmit based on retryable; do not infer it from the error text:

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)   # new task_id
    else:
        raise RuntimeError(f"{res.task_result.failure_code}: {res.error}")

Task.task_id is a fresh uuid on every build; log it when troubleshooting, since the worker’s task finished log entries are searchable by it.

Without the SDK: call gRPC directly

Go-side contributors often need to submit tasks directly to a Worker in tests or tools. The protocol is WorkerService in api/grpc/worker/worker.proto, with Go bindings in api/grpc/worker/workerpb. The program below goes through the three steps RequestTaskSlot → SubmitTask → WatchTasks against a local worker and submits the same task as example 1:

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. Reserve a slot. In production, the Coordinator's ReserveWorkerSlot does this step and returns worker_addr + slot_id.
	slot, err := client.RequestTaskSlot(ctx, &workerpb.RequestTaskSlotRequest{TaskId: taskID, TtlMs: 5000})
	if err != nil {
		log.Fatal(err)
	}

	// 2. Submit the task. config is opaque JSON, parsed by the corresponding 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) // admission errors: InvalidArgument / ResourceExhausted / FailedPrecondition / Unavailable
	}
	fmt.Println("submitted:", submit.State) // RUNNING

	// 3. Subscribe to the terminal state. The stream first sends the current snapshot, then one update with is_terminal=true when the task converges.
	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 can be omitted: when SubmitTask carries no slot_id, the Worker requests a slot inline and returns ResourceExhausted when there is no capacity.
When going through the Coordinator, first call coordinator.v1.CoordinatorService.ReserveWorkerSlot(task_id) (api/grpc/coordinator/coordinator.proto), then open a connection to the returned worker_addr and submit with slot_id.
Repeating SubmitTask with the same task_id is idempotent: it returns RUNNING while the task is still running, and the terminal state while the result is still within the retention window.
In tests you can use the RequestTaskSlot / SubmitTask / PollUntilTerminal helpers from internal/testutil directly; they wrap exactly the calls above.

Common errors

Symptom	Cause and fix
`RuntimeError: PROXY_SOCKET_PATH not set` or `UNAVAILABLE`	The proxy is not running, or the socket path differs from the one in `local_proxy.py`
`ValueError: ... BlockTableWriteHandler carries no block_bundle`	The handler shape does not match the target cluster, see the note in example 4
`InvalidArgument: plugin not registered: ReturnValueHandler`	`cluster="bundle"` forced a task with `ReturnValueHandler` onto the bundle cluster; remove the `cluster` argument
Queueing messages keep printing / `ResourceExhausted`	No idle slot. The SDK backs off and retries; when calling gRPC directly, back off yourself
`FAILED` with `failure_code` `CALL_FAILED`	The function raised an exception, or the code was rejected by the audit in `enforce` mode. Check the `call failed` entries in the worker log
`FAILED` with `failure_code` `BUILDER_FAILED`	The trigger table does not exist, `operator` validation failed, or reading BlockDB failed
`FAILED` with `failure_code` `PLUGIN_FAILED` and message `block deployment requires config.block`	The `BlockTableWriteHandler` shape was sent to the wrong cluster
`FAILED` with `failure_code` `TIMED_OUT`	Exceeded `timeout` or the worker’s default `TASK_DEADLINE_MS`; `retryable=true`
`InvalidArgument: task_timeout_ms ...`	`timeout` is negative or exceeds the worker’s `MAX_TASK_TIMEOUT_MS` (default 8 hours)

Quickstart: the shortest path to a local worker and proxy.
Task and Call: TaskInput fields, states, and failure codes.
Call Builder and Writer Plugin: the wire shape of every config.
Protocols and interfaces: the complete fields of WorkerService / CoordinatorService.
Testing: submission helpers in the process / system E2E tests.
docs/api_reference.md in the blockx-py repository: all SDK parameters and the behavior details of Pipeline.

Get started

Core concepts

Architecture

Components

Development guide

Prerequisites

Example 1: run once and get the result back

Example 2: compute per block and write a block table

Example 3: write a normal table

Example 4: backfill per bundle

Reading the result

Without the SDK: call gRPC directly

Common errors

​Prerequisites

​Example 1: run once and get the result back

​Example 2: compute per block and write a block table

​Example 3: write a normal table

​Example 4: backfill per bundle

​Reading the result

​Without the SDK: call gRPC directly

​Common errors

​Related docs

Prerequisites

Example 1: run once and get the result back

Example 2: compute per block and write a block table

Example 3: write a normal table

Example 4: backfill per bundle

Reading the result

Without the SDK: call gRPC directly

Common errors

Related docs