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 与 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。

​

代码位置

​

核心类型与接口

• 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.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) 从索引中选文件。细节归 后端适配器。
• 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）。

​

数据流 / 执行流程

​

状态与生命周期

• 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 失败分类

​

Iceberg 解析缓存

​

配置

​

扩展点

• 调 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/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。
• 测试组织总览见 测试。

​

相关文档

• 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 系统：Builder / Writer 通用契约。
• Worker：阶段准入与 orchestrator。
• Coordinator：调度与 watcher。
• Call 执行子系统：通用 Calls 阶段。
• IO 访问子系统：system cache 与 retry 分类。
• 后端适配器：BlockDB / Iceberg 客户端。
• 部署概览：进程形态与发布方式。