Bundle clusters

​

Responsibilities and boundaries

• Responsible for:
  • Accepting tasks of type BlockBundleCallConfig / CallListCallConfig.
  • Streaming rows from parquet with DuckDB and producing calls in batches.
  • Writing output back through a BlockDB BatchWrite Job with BlockBundleWriteResultHandler / TableUpsertsResultHandler. The write flow is Init → presigned PUT → Commit.
• Not responsible for:
  • Registering block capabilities such as single-block real-time dbscan (DBScanBuilder), the ReturnValue handler, or devstub. Related tasks are rejected at admission.
  • Changing shared semantics such as the task protocol, slot state machine, or heartbeat. These belong to Coordinator and Worker.
  • Orchestrating upstream routing. When the SDK shifts traffic to the bundle endpoint is outside this component.
• Invariant 1: membership discovery isolation. block workers register only under workers-kind registries, bundle workers only under bundle-workers-kind registries; each Coordinator watches its own exact prefix list. Profile.Validate rejects mismatches before any external dependency is created.
• Invariant 2: one codebase, different profiles.
  • There is one internal/worker/app.Run and one internal/coordinator/app.Run; cmd/* only declares a Profile.
  • The bundle Coordinator exposes only bundlecoordinator.v1.BundleCoordinatorService; the block Coordinator exposes only coordinator.v1.CoordinatorService.
  • The proxy separates the two traffic types by gRPC method path.
• Invariant 3: stream build must be on in production. The bundle profile’s TuneDefaults sets StreamBuild.Enabled=true; an operator can still roll back to the fully materialized Build() path with an explicit STREAM_BUILD_ENABLED=false.
• Invariant 4: static errors execute nothing, dynamic errors execute partially. PrepareStream completes all static validation in the Builder phase; on failure zero calls are dispatched. If Run fails midway, earlier batches may already have executed.
• Invariant 5: BatchWrite is the only write path, and it is strictly validated at startup. If blockdb.batchWriteAddr is empty when BundleWrite (or the bundle profile’s TableUpserts) is registered, the process fails at startup rather than waiting for the Writer phase.
• Invariant 6: when the Commit result is uncertain, do not call Cancel or create a replacement Job; return a non-reschedulable failure and keep the original job_id.

​

Code location

​

Core types and interfaces

• app.Profile (internal/worker/app/app.go): Deployment / Service / WorkerRegistryPrefix / UsageService (chaintable-usage service name, fixed by the entry point) / Builders / Plugins / TuneDefaults. The actual values for the bundle entry point:

// 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 decides which gRPC service to register based on Deployment.
• Registry prefixes (api/etcd/registry.go):
  • etcdapi.LegacyBundleWorkerRegistryPrefix = /blockx/bundle-workers/
  • ProdDefaultBundleWorkerRegistryPrefix = /blockx/prod/lanes/default/bundle-workers/
  • TestDefaultBundleWorkerRegistryPrefix is the default for test environments.
  • ParseRegistryPrefix returns RegistryPrefix{Format, Environment, Lane, WorkerKind}.
• adapters.BundleCoordinatorServer (internal/coordinator/adapters/bundle_grpc_server.go): only converts messages and calls CoordinatorServer.ReserveWorkerSlot.
• bundlescan.BundleScanBuilderDecl / BundleSource (internal/plugin/callbuilder/bundlescan/types.go): the on-wire functionCallConfig.config.
  • Bundle location field: blockBundle.number
  • Source fields: triggerSources[].tableId|bundleBucket|operator|function|code|param|forceResultCache
  • Missing-file marker: NoBundleFile = "NO_BUNDLE_FILE"
• bundlescan.BundleScanBuilder (bundlescan.go / stream.go): implements both types.CallBuilder and types.StreamingCallBuilder; SetStreamLimits injects the batch boundaries at startup.
• types.StreamingCallBuilder / CallStreamProducer / BatchEmit / ErrProductionStopped (internal/plugin/callbuilder/types/types.go): the two-phase streaming contract.
• bundlescan.DuckDBParquetReader / DuckDBParquetReaderConfig (duckdb.go):
  • StreamBundleWithPlan(ctx, objectURI, bundleNumber, plan, yield) streams a bundle.
  • Filters are pushed down into read_parquet(...) WHERE; dedup runs row by row on the Go side.
  • S3 credentials are initialized lazily. On ExpiredToken before anything has been yielded, the reader refreshes the secret and retries once.
• bundlescan.BundleRowYield (parquet.go): the borrowed-row contract — the row and the string/[]byte values inside it are valid only during the callback.
• borrowed.Pool / borrowed.ScanResult (internal/duckdb/borrowed/): connection pool and chunk iteration.
• icebergsdk.ResolveDataFilesReq / IcebergIOAdapter / Planner.PlanDataFiles (internal/sdk/iceberg/):
  • Resolves all live files in the table’s current snapshot in one pass and encodes them as a bundle index.
  • LookupBundleFileIndex(data, bundle) selects a file from the index. See Backend Adapter for details.
• batchwrite.Writer.Submit(ctx, taskIO, initReq, outputs, columns) (Result, error) (internal/plugin/event/batchwrite/writer.go):
  • Runs in this order: plan → Init → PUT → Commit.
  • Returns Result{JobID, RowCount, ContentBytes}.
  • commitStatusUnknownError always satisfies Retryable() == false.
• bundlewrite.BundleWritePlugin / BundleWriteConfig{TargetTable, BlockBundle}; tableupserts.NewPlugin (block) and tableupserts.NewBatchPlugin (bundle).

​

Data flow / execution flow

​

State and lifecycle

• Registry ownership: at worker startup, Profile.Validate runs, then validateWorkerRegistryPrefix runs again after config loading, guaranteeing that a bundle deployment can only write to a bundle-workers-kind prefix (legacy or v2). On the coordinator side, validateRegistryPrefixes does the same and also rejects duplicate prefixes.
• Two-phase stream build:
  • The Builder phase holds only builderSem and runs PrepareStream.
  • The Calls phase holds scanSem and executorSem while production runs. It releases scanSem when production finishes or fails.
  • One scanSem covers one task’s production process, which scans at most 3 sources concurrently. The worker-level DuckDB scan concurrency limit is about ScanSlots × 3.

​

BatchWrite failure classification

​

Iceberg resolution cache

​

Configuration

​

Extension points

• Tuning bundle profile defaults: change TuneDefaults in cmd/bundle_worker/main.go; it may only set defaults and must not post-process the final config (an operator’s explicit value must be able to override). Update cmd/bundle_worker/profile_test.go and internal/worker/app/profile_test.go accordingly.
• Adding or removing a Builder or Plugin for the bundle worker:
  • Change the profile’s Builders / Plugins.
  • Add or remove the corresponding case in newBuilderRegistry / newPluginRegistry in internal/worker/app/app.go.
  • If a new plugin depends on BatchWrite, also add it to validateProfileRuntimeConfig.
• Changing bundlescan source semantics (new fields, new resolution paths): bundlescan/types.go (decl) → source_resolution.go (resolution) → prepareBundleSource in bundlescan.go (all static validation must live here, shared by both paths) → source_calls.go (call assembly). Do not put validation into Run, or you break “static failure executes nothing”.
• Changing DuckDB reads: SQL assembly, S3 secrets, filter pushdown, and dedup are in bundlescan/duckdb.go; connection pool / row decoding / type support are in internal/duckdb/borrowed/ (no business policy there).
• Changing the BatchWrite lifecycle (retries, upload method, capacity caps): internal/plugin/event/batchwrite/. The BlockDB SDK keeps only the three atomic requests InitWriteJobReq / CommitWriteJobReq / CancelWriteJobReq; do not push orchestration down into the SDK.
• Adding a registry prefix format: ParseRegistryPrefix in api/etcd/registry.go; Validate on both sides picks it up automatically.
• Adding a bundle coordinator RPC: change api/grpc/bundlecoordinator/bundle_coordinator.proto → make (see the proto target in the Makefile) → internal/coordinator/adapters/bundle_grpc_server.go only does conversion; scheduling logic stays in the shared CoordinatorServer.

​

Testing

# Unit: bundlescan (includes DuckDB local parquet, streaming batch boundaries, static failure zero emit)
go test ./internal/plugin/callbuilder/bundlescan/...
# Unit: write path
go test ./internal/plugin/event/batchwrite/... ./internal/plugin/event/bundlewrite/... ./internal/plugin/event/tableupserts/...
# Unit: profile / registry prefix / assembly
go test ./internal/worker/app/... ./internal/coordinator/app/... ./internal/coordinator/adapters/... ./api/etcd/...
# Process-level smoke (builds the real binary and starts it; worker cases need python/.venv and modules such as blockx_executor)
go test -v -timeout 120s ./cmd/bundle_worker/...
go test -v -timeout 120s ./cmd/bundle_coordinator/...
# The load-testing tool itself
go test ./cmd/bundle-loadgen/...
# perf: memory comparison with stream build on/off (default 200000 rows, tunable via M3_ROWS)
go test ./e2e/perf/ -run TestPerf_M3_BundleStreamMemory -v -timeout 30m

• internal/plugin/callbuilder/bundlescan/*_test.go:
  • bundlescan_test.go: Build path, source resolution, archive remapping.
  • stream_test.go / TestBundleScanStream_*: MaxRows / MaxArgsBytes boundaries, single row over the limit, zero emits on static errors, zero emits from sibling sources, emit rejection stopping the scan, ctx cancellation.
  • duckdb_test.go: reader configuration, handle release, connection pool cancellation.
  • parquet_test.go / json_encoder_benchmark_test.go: parquet encoding and performance coverage.
• internal/plugin/event/bundlewrite/bundle_write_test.go and batchwrite/*_test.go: Init/PUT/Commit/Cancel state machine, error classification, Content-Length validation; tableupserts/table_upserts_test.go covers both modes.
• cmd/bundle_worker/bundle_worker_process_test.go:
  • TestBundleWorkerProcess_RejectsMissingBatchWriteAddr requires startup to fail without BLOCKDB_BATCH_WRITE_ADDR.
  • TestBundleWorkerProcess_BootsWithBundleProfile covers startup, readiness, and registration of bundle-profile capabilities only.
  • profile_test.go pins deployment / service / registry values. cmd/bundle_coordinator/*_test.go covers the corresponding coordinator behavior.
• internal/coordinator/adapters/bundle_grpc_server_test.go: the bundle adapter reuses shared scheduling and passes gRPC status through.
• See Testing for an overview of test organization.

​

Related docs

• docs/specs/2026-07-21-block-bundle-clusters.md — block / bundle deployment split: dual entry points, dual registries, system invariants (key).
• docs/specs/2026-07-28-registry-prefix-migration.md — legacy → v2 registry prefix format, WORKER_REGISTRY_PREFIX / COORDINATOR_REGISTRY_PREFIXES override rules.
• docs/specs/2026-07-23-bundle-coordinator-api.md — the decision on a standalone BundleCoordinatorService and its rollout order.
• docs/specs/2026-07-30-bundle-write-batch-api.md — the BatchWrite write path’s execution flow, contract boundaries, failure state machine (normative).
• docs/specs/plugin-system.md §6.1 (StreamingCallBuilder contract, the four NoResultCache conditions), §10 (BlockBundleCallConfig / BlockBundleWriteResultHandler / TableUpsertsResultHandler contracts).
• docs/specs/2026-07-30-ec2-worker-profiling.md — sampling plan for the two EC2 Worker clusters, block and bundle.

• Plugin system: shared Builder / Writer contracts.
• Worker: phase admission and orchestrator.
• Coordinator: scheduling and watcher.
• Call execution subsystem: shared Calls phase.
• IO access subsystem: system cache and retry classification.
• Backend Adapter: BlockDB / Iceberg clients.
• Deployment overview: process shapes and release methods.