> ## 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.

# Bundle 集群

> bundle coordinator/worker 双入口、bundlescan 流式扫描 parquet 产 call、blocktablewrite/tableupserts 经 BlockDB BatchWrite 写回，以及 bundle-loadgen 压测工具

历史 EVM 输入、输出上下文与写入作业完成见[回填历史 bundle](/zh/guides/bundle-backfill)。本页说明扫描与写入实现。

Bundle 集群是 BlockX 面向\*\*历史回填（bundle backfill）\*\*的独立部署，由 `cmd/bundle_coordinator` 和 `cmd/bundle_worker` 组成。

它复用 block 集群的 Coordinator / Worker 装配代码，只在入口 profile 上收窄能力、切换 etcd registry 前缀，并默认开启 stream build。系统全貌见 [架构总览](/zh/architecture/overview)。

一个 **block bundle** 是按块高连续切分的一段数据，1000 个区块一段，由 `BundleRef.Number` 标识。数据以 parquet 文件形式落在 S3 / Iceberg 表的 `bundle=N` 分区。

bundle backfill 任务把 parquet 逐行喂给用户函数，再通过 BlockDB BatchWrite 一次性覆写目标表的同一个 bundle。

## 职责与边界

* **负责**：
  * 接受 `BlockBundleCallConfig` / `InputsCallConfig` 类型的 task。
  * 用 DuckDB 从 parquet 流式扫描行，按批产出 call。
  * 用 `BlockTableWriteHandler`（bundle 实现）/ `NormalTableWriteHandler` 把输出经 BlockDB BatchWrite Job 写回。写入流程是 Init → presigned PUT → Commit。
* **不负责**：
  * 不注册单块实时 dbscan（`DBScanBuilder`）、`ReturnValue` handler、devstub 等 block 能力。相关任务会在准入时被拒。
  * 不改变 task 协议、slot 状态机、heartbeat 等公共语义。这些语义归 [Coordinator](/zh/components/coordinator) 与 [Worker](/zh/components/worker)。
  * 不编排上游路由。SDK 何时把流量切到 bundle endpoint 不在本组件内。
* **不变量 1：成员发现隔离**。block worker 只注册到 `workers` 类 registry，bundle worker 只注册到 `bundle-workers` 类 registry；两个 Coordinator 各 watch 自己的精确前缀列表。`Profile.Validate` 在创建任何外部依赖前拒绝错配。
* **不变量 2：同一套代码，不同 profile**。
  * `internal/worker/app.Run` / `internal/coordinator/app.Run` 各只有一份；`cmd/*` 只声明 `Profile`。
  * bundle Coordinator 只暴露 `bundlecoordinator.v1.BundleCoordinatorService`，block Coordinator 只暴露 `coordinator.v1.CoordinatorService`。
  * proxy 按 gRPC method path 区分两类流量。
* **不变量 3：stream build 生产必开**。bundle profile 的 `TuneDefaults` 置 `StreamBuild.Enabled=true`；操作者显式 `STREAM_BUILD_ENABLED=false` 仍可回滚到全量物化 `Build()` 路径。
* **不变量 4：静态错误零执行，动态错误部分执行**。`PrepareStream` 在 Builder 阶段完成全部静态校验，失败则零 call 派发；`Run` 中途失败时先前批次可能已执行。
* **不变量 5：BatchWrite 是唯一写路径，且启动期强校验**。注册 `BundleWrite`（或 bundle profile 的 `TableUpserts`）时 `blockdb.batchWriteAddr` 为空，进程在启动期失败，不等到 Writer 阶段。
* **不变量 6：Commit 结果不确定时不 Cancel、不建替代 Job**，返回不可重调度失败并保留原 `job_id`。

## 代码位置

| 路径 | 用途 |
| - | - |
| `cmd/bundle_worker/main.go` | bundle worker 入口：只声明 `app.Profile`（Builders/Plugins 白名单、registry 前缀、`TuneDefaults`） |
| `cmd/bundle_coordinator/main.go` | bundle coordinator 入口：`Deployment: "bundle"`，watch legacy + prod v2 两个 bundle registry 前缀 |
| `cmd/bundle-loadgen/` | 通过 bundle coordinator 提交合成 task 的压测 CLI，附 `README.md` 与 `cgroup-snapshot.sh` |
| `api/etcd/registry.go` | registry 前缀常量与 `ParseRegistryPrefix`（legacy / v2 格式、worker-kind 校验） |
| `api/grpc/bundlecoordinator/bundle_coordinator.proto` | `BundleCoordinatorService.ReserveWorkerSlot` 独立 wire contract |
| `internal/coordinator/app/` | 两个 coordinator 入口共用的装配（`Profile`、config、gRPC server、watcher） |
| `internal/coordinator/adapters/bundle_grpc_server.go` | `BundleCoordinatorServer`：把 bundle proto 转成共享 `CoordinatorServer` 调用 |
| `internal/worker/app/app.go` | 两个 worker 入口共用的装配；`newBuilderRegistry` / `newPluginRegistry` 按 profile 注册；`validateProfileRuntimeConfig` 强制 `BatchWriteAddr` |
| `internal/plugin/callbuilder/bundlescan/` | `BundleScanBuilder`（`Build` + `PrepareStream`）、DuckDB parquet reader、source 解析、行→call 编码 |
| `internal/duckdb/borrowed/` | DuckDB C bindings 上的低分配基础层：连接池、ctx 中断、零拷贝行视图；不含任何 bundle 业务策略 |
| `internal/plugin/event/batchwrite/` | BatchWrite Job 生命周期的唯一编排方：行计划/编码、presigned PUT、Init/Commit/Cancel |
| `internal/plugin/event/blocktablewrite/bundle/` | `BlockTableWriteHandler` 的 bundle 实现 `bundle.Plugin`：`BlockTableOverwriteStrategy{BlockBundle}`；由 `blocktablewrite/plugin.go` 按 `Deployment` 装配 |
| `internal/plugin/event/tableupserts/` | `TableUpserts` 双模式：block 直接 `UpsertRows`，bundle 走 `NormalTableUpsertStrategy` BatchWrite |
| `internal/sdk/iceberg/` | `tableId` → 当前 snapshot live data file 的 resolver（Glue + Iceberg），经 IO 子系统 system cache |
| `internal/worker/adapters/orchestrator_phases.go` | Worker 侧流式生产：`PrepareStream` 在 builderSem 下、`producer.Run` 在 scanSem 下 |
| `e2e/perf/m3_stream_build_test.go` | `TestPerf_M3_BundleStreamMemory`：stream build 开/关 的 RSS / heap / 阶段耗时对照 |
| `docs/specs/2026-07-21-block-bundle-clusters.md` 等 | 见文末"相关文档" |

## 核心类型与接口

* `app.Profile`（`internal/worker/app/app.go`）：`Deployment` / `Service` / `WorkerRegistryPrefix` / `UsageService`（chaintable-usage 服务名，由入口固定）/ `Builders` / `Plugins` / `TuneDefaults`。bundle 入口的实际值：

```go theme={null}
// cmd/bundle_worker/main.go
return app.Profile{
    Deployment:           "bundle",
    Service:              "blockx-bundle-worker",
    WorkerRegistryPrefix: "/blockx/bundle-workers/",
    UsageService:         usage.ServiceBlockXBundleWorker,
    Builders: []cbtypes.CallBuilderName{
        cbtypes.BuilderBundleScan,
        cbtypes.BuilderCallList,
    },
    Plugins: []evtypes.PluginName{
        evtypes.PluginBlockDBWrite,
        evtypes.PluginTableUpserts,
    },
    TuneDefaults: func(cfg *app.WorkerFullConfig) {
        cfg.StreamBuild.Enabled = true
        cfg.Dispatcher.ExecutorSelectionSampleSize = 3
    },
}
```

* `app.Profile`（`internal/coordinator/app/app.go`）：`Deployment` / `Service` / `WorkerRegistryPrefixes []string`；`registerCoordinatorService` 按 `Deployment` 决定注册哪个 gRPC service。
* registry 前缀（`api/etcd/registry.go`）：
  * `etcdapi.LegacyBundleWorkerRegistryPrefix` = `/blockx/bundle-workers/`
  * `ProdDefaultBundleWorkerRegistryPrefix` = `/blockx/prod/lanes/default/bundle-workers/`
  * `TestDefaultBundleWorkerRegistryPrefix` 是测试环境默认值。
  * `ParseRegistryPrefix` 返回 `RegistryPrefix{Format, Environment, Lane, WorkerKind}`。
* `adapters.BundleCoordinatorServer`（`internal/coordinator/adapters/bundle_grpc_server.go`）：只做 message 转换，调 `CoordinatorServer.ReserveWorkerSlot`。
* `bundlescan.BundleScanBuilderDecl` / `BundleSource`（`internal/plugin/callbuilder/bundlescan/types.go`）：on-wire `functionCallConfig.config`。
  * bundle 定位字段：`blockBundle.number`
  * source 字段：`triggerSources[].tableId|bundleBucket|operator|function|code|param|forceResultCache`
  * 缺文件标记：`NoBundleFile = "NO_BUNDLE_FILE"`
* `bundlescan.BundleScanBuilder`（`bundlescan.go` / `stream.go`）：同时实现 `types.CallBuilder` 与 `types.StreamingCallBuilder`；`SetStreamLimits` 在启动时注入批边界。
* `types.StreamingCallBuilder` / `CallStreamProducer` / `BatchEmit` / `ErrProductionStopped`（`internal/plugin/callbuilder/types/types.go`）：流式两阶段契约。
* `bundlescan.DuckDBParquetReader` / `DuckDBParquetReaderConfig`（`duckdb.go`）：
  * `StreamBundleWithPlan(ctx, objectURI, bundleNumber, plan, yield)` 流式读取 bundle。
  * filter 下推到 `read_parquet(...) WHERE`，dedup 在 Go 侧逐行执行。
  * S3 凭证懒初始化。收到 `ExpiredToken` 且尚未 yield 时，刷新 secret 并重试一次。
* `bundlescan.BundleRowYield`（`parquet.go`）：借用行契约——行与其中的 string/\[]byte 只在回调期间有效。
* `borrowed.Pool` / `borrowed.ScanResult`（`internal/duckdb/borrowed/`）：连接池与 chunk 迭代。
* `icebergsdk.ResolveDataFilesReq` / `IcebergIOAdapter` / `Planner.PlanDataFiles`（`internal/sdk/iceberg/`）：
  * 一次解析整张表当前 snapshot 的全部 live 文件，并编码为 bundle 索引。
  * `LookupBundleFileIndex(data, bundle)` 从索引中选文件。细节归 [后端适配器](/zh/components/backend-adapter)。
* `batchwrite.Writer.Submit(ctx, taskIO, initReq, outputs, columns) (Result, error)`（`internal/plugin/event/batchwrite/writer.go`）：
  * 按计划 → Init → PUT → Commit 的顺序执行。
  * 返回 `Result{JobID, RowCount, ContentBytes}`。
  * `commitStatusUnknownError` 永远满足 `Retryable() == false`。
* `blocktablewrite/bundle` 的 `Plugin` / `Config{TargetTable, BlockBundle}`；`tableupserts.NewPlugin`（block）与 `tableupserts.NewBatchPlugin`（bundle）。

## 数据流 / 执行流程

```mermaid theme={null}
flowchart LR
    caller["bundle 调用方"] -->|"ReserveWorkerSlot (BundleCoordinatorService)"| bcoord["cmd/bundle_coordinator"]
    bcoord -.->|watch| reg[("/blockx/bundle-workers/ + prod v2")]
    bworker["cmd/bundle_worker"] -->|"register / heartbeat"| reg
    caller -->|SubmitTask| bworker
    subgraph W["bundle worker（一个 task）"]
        prep["bundlescan.PrepareStream<br/>静态校验 / Iceberg 解析 / schema"]
        prod["bundleScanProducer.Run<br/>DuckDB read_parquet 流式扫描<br/>按批 emit（8192 行 / 32MiB）"]
        disp["dispatcher / executor<br/>（通用 Call 执行）"]
        wr["blocktablewrite/bundle 或 tableupserts<br/>batchwrite.Writer.Submit"]
        prep --> prod --> disp --> wr
    end
    bworker --> prep
    prep -->|ResolveDataFilesReq| ice["Iceberg / Glue（system cache）"]
    prod -->|"S3 parquet"| s3in[("S3 bundle parquet")]
    wr -->|InitWriteJob / CommitWriteJob| blockdb["BlockDB BatchWriteService"]
    wr -->|"HTTP PUT presigned URL"| s3out[("S3 staging")]
```

<Steps>
  <Step title="Reserve + Submit">
    调用方先向 bundle coordinator 调用 `BundleCoordinatorService.ReserveWorkerSlot`。拿到 `worker_addr` / `slot_id` 后，再向目标 worker 提交 `SubmitTask`。

    调度、slot TTL 和 heartbeat 与 block 集群一致，见 [Coordinator](/zh/components/coordinator)。
  </Step>

  <Step title="Builder 阶段：PrepareStream（builderSem 下，短）">
    `BundleScanBuilder.PrepareStream` 反序列化 `BundleScanBuilderDecl`，并对每个 source 调用 `prepareBundleSource` 执行静态准备：

    * **解析数据源**：`resolveBundleSourceBucket` 在 `bundleBucket` 非空时直接使用该值。否则读 logical-types 判断 event / state；state 表转到 `<tableId>._archive`，再通过 `ResolveDataFilesReq` 解析唯一的 parquet URI。
    * **处理缺失数据**：返回 `NO_BUNDLE_FILE` 或 snapshot 中没有该 bundle 时，跳过当前 source，产出 0 个 call。
    * **归一化调用**：归一化 inline code；archive state 表还会把 operator 中的 `id` 重映射为 `original_id`。
    * **校验扫描计划**：干跑 `validateDuckDBReadPlan` 和 `ToFilterSQLArgs`，然后预建 `ParquetJSONEncoder`。
    * **决定缓存策略**：`markProvablyUniqueSources` 决定哪些 source 的 call 需要设置 `NoResultCache`。

    **失败语义**：任何静态准备失败都收敛为 `BUILDER_FAILED`，且不派发 call。
  </Step>

  <Step title="Calls 阶段：Run（scanSem 下，长）">
    Orchestrator 在 `scanSem` 下启动 `runCallProduction`：

    * **扫描并发**：`bundleScanProducer.Run` 使用 errgroup，最多并发扫描 3 个 source。
    * **构造 call**：每行经 `bundleRowCallBuilder.buildNext` 编码成 `Call`。row args 直接写入 argspool 缓冲区，并设置 `ArgsPooled=true`。
    * **切分批次**：`bundleSourceBatchAppender` 在 `maxRows` 或 `maxArgsBytes` 先到上限时执行 `flush` 和 `emit`。
    * **建立背压**：`emit` 阻塞在 `MaxOutstandingBatches` 信用上，从而限制未完成批次的数量。

    **失败语义**：单行 args 超过字节上限时返回 production error。`emit` 返回 `ErrProductionStopped` 表示 worker 因 fast-fail 受控停止，不算 builder 失败。
  </Step>

  <Step title="Dispatch / Execute">
    通用路径，见 [Call 执行子系统](/zh/components/call-execution) 与 [Plugin 系统](/zh/components/plugins)。
  </Step>

  <Step title="Writer 阶段：BatchWrite">
    Calls 收敛成功后，`bundle.Plugin.Execute` 校验 `targetTable` / `blockBundle`，读取列定义，然后调用 `batchwrite.Writer.Submit`：

    1. **Plan**：`planDelimitedWriteRows` 无保留遍历 outputs，计算精确的行数和字节数。超过 10,000,000 行或 5,000,000,000 bytes 时，在 Init 前失败。
    2. **Init**：`InitWriteJob` 创建 Job，返回 `job_id` 和 `data_url`。
    3. **PUT**：编码器通过 `io.Pipe` 向 presigned URL 发起一次 HTTP `PUT`。请求使用准确的 `Content-Length`，且不跟随重定向。
    4. **Commit**：上传成功后调用 `CommitWriteJob(row_count)`。
    5. **Cancel**：PUT 失败或 Commit 前 ctx 取消时，best-effort 调用 `CancelWriteJob`。该调用使用 `context.WithoutCancel` 和 5s 独立超时。

    成功时返回 `written_rows` / `finish_time` / `job_id`。`DebugMode` 只预览输出，不创建 Job。
  </Step>
</Steps>

<Note>
  `NormalTableWriteHandler` 在两类 worker 上使用不同写路径：

  * **bundle worker**：复用 `batchwrite.Writer`，策略为 `NormalTableUpsertStrategy{Condition, UpdateColumns}`。
  * **Init 前校验**：`planNormalUpsertDelimitedWriteRows` 要求每行的有效字段集合一致、`id` 非空，且至少包含一个非 ID 列。它会把首行推导出的 `update_columns` 显式写入请求。
  * **block worker**：同名 handler 仍调用 `TableWriteService.UpsertRows(sync=false)`。
  * **配置边界**：写路径由 profile 在装配期固定，task 配置中没有 `writeApi`。
</Note>

## 状态与生命周期

* **registry 归属**：worker 启动时先执行 `Profile.Validate`，配置加载后再执行 `validateWorkerRegistryPrefix`。这保证 `bundle` deployment 只会写入 `bundle-workers` 类前缀（legacy 或 v2）。coordinator 端的 `validateRegistryPrefixes` 执行同类校验，并拒绝重复前缀。
* **stream build 两阶段**：
  * Builder 阶段只持有 builderSem，执行 `PrepareStream`。
  * Calls 阶段持有 scanSem 和 executorSem，运行 production。production 结束或失败后释放 scanSem。
  * 一个 scanSem 对应一个 task 的生产过程，内部最多并发扫描 3 个 source。worker 级 DuckDB 扫描并发上限约为 `ScanSlots × 3`。

### BatchWrite 失败分类

分类逻辑位于 `batchwrite/classified_error.go` 和 `presigned_upload.go`。插件把分类结果透传为 `PluginError.Retryable`。

| 分类 | 典型情况 | 可重调度 |
| - | - | - |
| 参数或响应错误 | 配置、校验、编码、非法响应、重定向、其他 4xx | 否 |
| 瞬时下游或传输错误 | Meta / BlockDB 瞬时 IO、HTTP 传输错误、PUT 408 / 425 / 429 / 500 / 502 / 503 / 504 | 是 |
| Commit 状态不确定 | Commit 请求或响应不确定；收敛为 `commitStatusUnknownError` 并保留 `job_id` | 否 |

### Iceberg 解析缓存

`ResolveDataFilesReq.CacheKey()` 只使用逻辑表 ID，请求走 Worker system cache。默认 TTL 为 20 分钟，由 `SystemIOCacheTTLms` 控制。

缓存索引的高水位覆盖 `RequiredBundle` 时，`AcceptCachedRead` 复用缓存；否则刷新一次整表索引。

## 配置

以下是 bundle worker 最相关的配置。结构体字段来自 `internal/worker/app/config.go`、`internal/worker/core/config.go` 和 `internal/worker/adapters/orchestrator.go`。

配置优先级是：内置默认 → `TuneDefaults` → `WORKER_CONFIG` 文件 → 环境变量。

| 字段 / 环境变量 | 默认值 | 说明 |
| - | - | - |
| `workerRegistryPrefix` / `WORKER_REGISTRY_PREFIX` | `/blockx/bundle-workers/`（profile 默认） | 只能是 `bundle-workers` 类前缀；测试环境设 `/blockx/test/lanes/default/bundle-workers/` |
| `streamBuild.enabled` / `STREAM_BUILD_ENABLED` | bundle profile `true`（内置 `false`） | 关闭即回退到全量物化 `Build()`（回滚路径） |
| `streamBuild.batchMaxRows` / `STREAM_BUILD_BATCH_MAX_ROWS` | `8192` | 一批最多行数 |
| `streamBuild.batchMaxArgsBytes` / `STREAM_BUILD_BATCH_MAX_ARGS_BYTES` | `32 << 20`（32MiB） | 一批 `ArgsJSON` 总字节上限；单行超限即失败 |
| `streamBuild.maxOutstandingBatches` / `STREAM_BUILD_MAX_OUTSTANDING_BATCHES` | `2` | 生产者信用窗口；请求侧稳态内存约 `(2 + 3) × 32MiB` |
| `admission.scanSlots` / `SCAN_SLOTS` | `8` | 并发 streaming production 数（每个内部再 ×3 source） |
| `admission.builderSlots` / `executorTaskSlots` / `pluginSlots` | `16` / `32` / `16` | 阶段准入；bundle 未另行调优 |
| `worker.taskSlots` / `TASK_SLOTS` | `50` | task 槽位 |
| `dispatcher.executorSelectionSampleSize` / `EXECUTOR_SELECTION_SAMPLE_SIZE` | bundle profile `3`（内置 `0` = 全池） | 每次 root call 派发只评分 3 个 executor |
| `blockdb.batchWriteAddr` / `BLOCKDB_BATCH_WRITE_ADDR` | 空 | **bundle profile 必填**；空则启动失败 |
| `icebergNamespace` / `ICEBERG_NAMESPACE`、`icebergRegion` / `ICEBERG_REGION` | 空（必填）/ `ap-northeast-1` | Glue namespace；`tableId` 路径依赖它 |
| `io.systemIoCacheTTLms` / `SYSTEM_IO_CACHE_TTL_MS` | 20 分钟 | Iceberg 解析结果缓存 TTL |
| `DUCKDB_MEMORY_LIMIT` / `DUCKDB_THREADS` / `DUCKDB_TEMP_DIRECTORY` / `DUCKDB_MAX_TEMP_DIRECTORY_SIZE` / `DUCKDB_MAX_OPEN_CONNS` / `DUCKDB_MAX_IDLE_CONNS` / `DUCKDB_PRESERVE_INSERTION_ORDER` / `DUCKDB_ENABLE_EXTERNAL_FILE_CACHE` | 空 = DuckDB 默认；`MaxOpenConns` 缺省为 `max(NumCPU, 4)` | 仅环境变量，`duckDBParquetReaderConfigFromEnv` 读取 |
| `memoryDiagnosticsIntervalMs` / `WORKER_MEMORY_DIAGNOSTICS_INTERVAL_MS` | 15 分钟 | 周期打印含 `DuckDBDiagnostics`（内存、临时盘、连接池）的内存日志 |

bundle coordinator：`COORDINATOR_REGISTRY_PREFIXES`（逗号分隔精确列表）整体替换 profile 默认的 `[/blockx/bundle-workers/, /blockx/prod/lanes/default/bundle-workers/]`；其余配置与 block coordinator 相同。

## 扩展点

* **调 bundle profile 默认值**：改 `cmd/bundle_worker/main.go` 的 `TuneDefaults`；只许设默认，不许后处理最终配置（操作者显式值必须能覆盖）。对应更新 `cmd/bundle_worker/profile_test.go`、`internal/worker/app/profile_test.go`。
* **给 bundle worker 增/减 Builder 或 Plugin**：
  * 修改 profile 的 `Builders` / `Plugins`。
  * 在 `internal/worker/app/app.go` 的 `newBuilderRegistry` / `newPluginRegistry` 中增删对应 case。
  * 新 plugin 依赖 BatchWrite 时，还要把它加入 `validateProfileRuntimeConfig`。
* **改 bundlescan 的 source 语义**（新字段、新的解析路径）：`bundlescan/types.go`（decl）→ `source_resolution.go`（解析）→ `bundlescan.go` 的 `prepareBundleSource`（所有静态校验必须在这里，两条路径共用）→ `source_calls.go`（call 拼装）。不要把校验放进 `Run`，否则破坏静态失败零执行。
* **改 DuckDB 读取**：SQL 拼装、S3 secret、filter 下推、dedup 在 `bundlescan/duckdb.go`；连接池 / 行解码 / 类型支持在 `internal/duckdb/borrowed/`（不放业务策略）。
* **改 BatchWrite 生命周期**（重试、上传方式、容量上限）：`internal/plugin/event/batchwrite/`。BlockDB SDK 只保留 `InitWriteJobReq` / `CommitWriteJobReq` / `CancelWriteJobReq` 三个原子请求，不要把编排下沉到 SDK。
* **新增一种 registry 前缀格式**：`api/etcd/registry.go` 的 `ParseRegistryPrefix`，两端 `Validate` 自动生效。
* **新的 bundle coordinator RPC**：改 `api/grpc/bundlecoordinator/bundle_coordinator.proto` → `make`（见 `Makefile` proto 目标）→ `internal/coordinator/adapters/bundle_grpc_server.go` 只做转换，调度逻辑仍在共享 `CoordinatorServer`。

## 测试

```bash theme={null}
# 单元：bundlescan（含 DuckDB 本地 parquet、流式批边界、静态失败零 emit）
go test ./internal/plugin/callbuilder/bundlescan/...
# 单元：写路径
go test ./internal/plugin/event/batchwrite/... ./internal/plugin/event/blocktablewrite/... ./internal/plugin/event/tableupserts/...
# 单元：profile / registry 前缀 / 装配
go test ./internal/worker/app/... ./internal/coordinator/app/... ./internal/coordinator/adapters/... ./api/etcd/...
# 进程级冒烟（构建真实二进制并启动；worker 用例需要 python/.venv 与 blockx_executor 等模块）
go test -v -timeout 120s ./cmd/bundle_worker/...
go test -v -timeout 120s ./cmd/bundle_coordinator/...
# 压测工具自身
go test ./cmd/bundle-loadgen/...
# perf：stream build 开/关 内存对照（默认 200000 行，M3_ROWS 可调）
go test ./e2e/perf/ -run TestPerf_M3_BundleStreamMemory -v -timeout 30m
```

* `internal/plugin/callbuilder/bundlescan/*_test.go`：
  * `bundlescan_test.go`：Build 路径、source 解析、archive 重映射。
  * `stream_test.go` / `TestBundleScanStream_*`：`MaxRows` / `MaxArgsBytes` 边界、单行超限、静态错误零 emit、sibling source 零 emit、emit 拒绝停扫、ctx 取消。
  * `duckdb_test.go`：reader 配置、句柄释放、连接池取消。
  * `parquet_test.go` / `json_encoder_benchmark_test.go`：parquet 编码与性能覆盖。
* `internal/plugin/event/blocktablewrite/bundle/write_test.go` 与 `batchwrite/*_test.go`：Init/PUT/Commit/Cancel 状态机、错误分类、Content-Length 校验；`tableupserts/table_upserts_test.go` 覆盖两种模式。
* `cmd/bundle_worker/bundle_worker_process_test.go`：
  * `TestBundleWorkerProcess_RejectsMissingBatchWriteAddr` 要求缺少 `BLOCKDB_BATCH_WRITE_ADDR` 时启动失败。
  * `TestBundleWorkerProcess_BootsWithBundleProfile` 覆盖启动、就绪和只注册 bundle profile 能力。
  * `profile_test.go` 钉住 deployment / service / registry 值。`cmd/bundle_coordinator/*_test.go` 覆盖 coordinator 侧同类行为。
* `internal/coordinator/adapters/bundle_grpc_server_test.go`：bundle adapter 复用共享调度并透传 gRPC status。
* 测试组织总览见 [测试](/zh/development/testing)。

<Tip>
  `cmd/bundle-loadgen` 的完整用法见仓库内 `cmd/bundle-loadgen/README.md`。

  * `--coordinator` 指向 bundle coordinator。
  * `--input-uri` 指向已验证唯一列的 parquet fixture。
  * `--workload noop|cpu|blockdb|function` 选择负载。
  * `--result-cache disabled|enabled` 与 `--result-shape` 用于 CallCache / 输出编码的控制变量 A/B。
  * `--result-handler-type BlockTableWriteHandler` 可以接入真实 writer，但必须使用隔离测试表。
  * worker 必须设置 `STREAM_BUILD_ENABLED=true`。`blockdb` / `function` 场景还依赖 `STUB_BLOCKDB_DELAY_MS` 等 worker stub。
</Tip>

## 相关文档

blockx 仓库 `docs/specs/`：

* `docs/specs/2026-07-21-block-bundle-clusters.md` — block / bundle 部署拆分：双入口、双 registry、系统不变量（重点）。
* `docs/specs/2026-07-28-registry-prefix-migration.md` — legacy → v2 registry 前缀格式、`WORKER_REGISTRY_PREFIX` / `COORDINATOR_REGISTRY_PREFIXES` 覆盖规则。
* `docs/specs/2026-07-23-bundle-coordinator-api.md` — 独立 `BundleCoordinatorService` 的决策与上线顺序。
* `docs/specs/2026-07-30-bundle-write-batch-api.md` — BatchWrite 写路径的执行流程、契约边界、失败状态机（规范性）。
* `docs/specs/plugin-system.md` §6.1（`StreamingCallBuilder` 契约、`NoResultCache` 四条件）、§10（`BlockBundleCallConfig` / `BlockTableWriteHandler` / `NormalTableWriteHandler` 契约）。
* `docs/specs/2026-07-30-ec2-worker-profiling.md` — 面向 block / bundle 两套 EC2 Worker 集群的采样方案。

站内文档：

* [Plugin 系统](/zh/components/plugins)：Builder / Writer 通用契约。
* [Worker](/zh/components/worker)：阶段准入与 orchestrator。
* [Coordinator](/zh/components/coordinator)：调度与 watcher。
* [Call 执行子系统](/zh/components/call-execution)：通用 Calls 阶段。
* [IO 访问子系统](/zh/components/io-subsystem)：system cache 与 retry 分类。
* [后端适配器](/zh/components/backend-adapter)：BlockDB / Iceberg 客户端。
* [部署概览](/zh/development/deployment)：进程形态与发布方式。


This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.