Bundle 集群

Bundle 集群是 BlockX 面向**历史回填（bundle backfill）**的一套独立部署：cmd/bundle_coordinator + cmd/bundle_worker。它复用与 block 集群完全相同的 Coordinator / Worker 装配代码，只在入口 profile 上收窄能力、切换 etcd registry 前缀并默认开启 stream build。系统全貌见架构总览。一个 block bundle 是按块高连续切分的一段数据（BundleRef.Number = N 覆盖高度 N*1000+1 … (N+1)*1000，bundle 0 只含高度 0），以 parquet 文件形式落在 S3 / Iceberg 表的 bundle=N 分区。bundle backfill 任务把这份 parquet 逐行喂给用户函数，再把全部输出用 BlockDB BatchWrite 一次性覆写目标表的同一个 bundle。

职责与边界

负责：接受 BlockBundleCallConfig / CallListCallConfig 类型的 task；用 DuckDB 从 parquet 流式扫描行、按批产出 call；用 BlockBundleWriteResultHandler / TableUpsertsResultHandler 把输出经 BlockDB BatchWrite Job（Init → presigned PUT → Commit）写回。
不负责：单块实时 dbscan（DBScanBuilder）、ReturnValue handler、devstub 等 block 能力——bundle profile 不注册它们，任务在准入时被拒；不改变 task 协议、slot 状态机、heartbeat 等公共语义（这些归 Coordinator 与 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.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/bundlewrite/`	`BundleWritePlugin`：`BlockTableOverwriteStrategy{BlockBundle}`
`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 入口的实际值：

// 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.PluginBundleWrite,
        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。
etcdapi.LegacyBundleWorkerRegistryPrefix = /blockx/bundle-workers/、ProdDefaultBundleWorkerRegistryPrefix = /blockx/prod/lanes/default/bundle-workers/、TestDefaultBundleWorkerRegistryPrefix（api/etcd/registry.go）；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，字段 blockBundle.number、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)；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) 选文件。细节归后端适配器。
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。
bundlewrite.BundleWritePlugin / BundleWriteConfig{TargetTable, BlockBundle}；tableupserts.NewPlugin（block）与 tableupserts.NewBatchPlugin（bundle）。

数据流 / 执行流程

Reserve + Submit

调用方对 bundle coordinator 调 BundleCoordinatorService.ReserveWorkerSlot，拿到 worker_addr / slot_id 后向该 worker SubmitTask。调度、slot TTL、heartbeat 与 block 完全一致，见 Coordinator。

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 则 skip、产 0 call）→ inline code 归一化 → archive state 表把 operator 里的 id 重映射为 original_id → validateDuckDBReadPlan + ToFilterSQLArgs 干跑 → 预建 ParquetJSONEncoder。最后 markProvablyUniqueSources 决定哪些 source 的 call 置 NoResultCache。任何一步失败即 BUILDER_FAILED，零 call 派发。

Calls 阶段：Run（scanSem 下，长）

Orchestrator 在 scanSem 下起 runCallProduction，bundleScanProducer.Run 用 errgroup 限 3 并发扫 source。每行经 bundleRowCallBuilder.buildNext 编码成 Call（row-args 直接写进 argspool 缓冲区，ArgsPooled=true），bundleSourceBatchAppender 在 maxRows 或 maxArgsBytes 先到者处 flush → emit。emit 阻塞在 MaxOutstandingBatches 信用上，形成背压。单行 args 超过字节上限是 production error。emit 返回 ErrProductionStopped 表示 worker 受控停止（fast-fail），不算 builder 失败。

Dispatch / Execute

通用路径，见 Call 执行子系统与 Plugin 系统。

Writer 阶段：BatchWrite

Calls 收敛成功后，BundleWritePlugin.Execute 校验 targetTable / blockBundle，读列定义，调 batchwrite.Writer.Submit：先无保留遍历 outputs 算精确行数与字节数（planDelimitedWriteRows；超过 10,000,000 行或 5,000,000,000 bytes 在 Init 前失败）→ InitWriteJob 拿 job_id + data_url → 边编码边通过 io.Pipe 单次 HTTP PUT（准确 Content-Length，不跟随重定向）→ CommitWriteJob(row_count)。PUT 失败或 Commit 前 ctx 取消则 best-effort CancelWriteJob（5s 独立超时，context.WithoutCancel）。成功返回 written_rows / finish_time / job_id。DebugMode 只预览输出，不建 Job。

TableUpsertsResultHandler 在 bundle worker 上走同一个 batchwrite.Writer，只是策略换成 NormalTableUpsertStrategy{Condition, UpdateColumns}；planNormalUpsertDelimitedWriteRows 在 Init 前要求每行有效字段集合一致、id 非空、至少一个非 ID 列，并把首行推导出的 update_columns 显式写入请求。block worker 上的同名 handler 仍调 TableWriteService.UpsertRows(sync=false)。写路径由 profile 在装配期固定，task 配置里没有 writeApi。

状态与生命周期

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）：配置/校验/编码/非法响应/重定向/其他 4xx → 不可重调度；Meta/BlockDB 瞬时 IO、HTTP 传输错误、PUT 408/425/429/500/502/503/504 → 可重调度；Commit 请求或响应不确定 → commitStatusUnknownError，不可重调度且保留 job_id。插件把分类透传为 PluginError.Retryable。
Iceberg 解析缓存：ResolveDataFilesReq.CacheKey() 只用逻辑表 ID，走 Worker system cache（默认 TTL 20 分钟，SystemIOCacheTTLms）；AcceptCachedRead 在缓存索引的高水位覆盖 RequiredBundle 时复用，否则整表刷新一次。

配置

以下是 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。

测试

# 单元：bundlescan（含 DuckDB 本地 parquet、流式批边界、静态失败零 emit）
go test ./internal/plugin/callbuilder/bundlescan/...
# 单元：写路径
go test ./internal/plugin/event/batchwrite/... ./internal/plugin/event/bundlewrite/... ./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。
internal/plugin/event/bundlewrite/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 同理。
internal/coordinator/adapters/bundle_grpc_server_test.go：bundle adapter 复用共享调度并透传 gRPC status。
测试组织总览见测试。

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 BlockBundleWriteResultHandler 可接真实 writer（必须用隔离测试表）。worker 侧必须 STREAM_BUILD_ENABLED=true；blockdb / function 场景依赖 worker 打开 STUB_BLOCKDB_DELAY_MS 等 stub。

开始使用

架构

组件

开发指南

职责与边界

代码位置

核心类型与接口

数据流 / 执行流程

状态与生命周期

配置

扩展点

测试

相关文档

​职责与边界

​代码位置

​核心类型与接口

​数据流 / 执行流程

​状态与生命周期

​配置

​扩展点

​测试

​相关文档

职责与边界

代码位置

核心类型与接口

数据流 / 执行流程

状态与生命周期

配置

扩展点

测试

相关文档