> ## 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 clusters
> Dual bundle coordinator/worker entry points, bundlescan streaming parquet scans that produce calls, bundlewrite/tableupserts write-back through BlockDB BatchWrite, and the bundle-loadgen load-testing tool
The Bundle cluster is BlockX's independent deployment for **historical backfill (bundle backfill)**. It consists of `cmd/bundle_coordinator` and `cmd/bundle_worker`.
It reuses the block cluster's Coordinator / Worker assembly code. Only the entry-point profile narrows its capabilities, switches the etcd registry prefix, and enables stream build by default. See [Architecture overview](/en/architecture/overview) for the full system.
A **block bundle** is a contiguous slice of data split by block height. `BundleRef.Number = N` covers heights `N*1000+1 … (N+1)*1000`; bundle 0 contains only height 0. The data is stored as parquet files in the `bundle=N` partition of an S3 / Iceberg table.
A bundle backfill task feeds the parquet rows to the user function, then uses BlockDB BatchWrite to overwrite the same bundle in the target table.
## 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](/en/components/coordinator) and [Worker](/en/components/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
| Path | Purpose |
| ----------------------------------------------------------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| `cmd/bundle_worker/main.go` | bundle worker entry point: only declares `app.Profile` (Builders/Plugins allowlist, registry prefix, `TuneDefaults`) |
| `cmd/bundle_coordinator/main.go` | bundle coordinator entry point: `Deployment: "bundle"`, watches the legacy + prod v2 bundle registry prefixes |
| `cmd/bundle-loadgen/` | Load-testing CLI that submits synthetic tasks through the bundle coordinator, with `README.md` and `cgroup-snapshot.sh` |
| `api/etcd/registry.go` | Registry prefix constants and `ParseRegistryPrefix` (legacy / v2 formats, worker-kind validation) |
| `api/grpc/bundlecoordinator/bundle_coordinator.proto` | Standalone wire contract for `BundleCoordinatorService.ReserveWorkerSlot` |
| `internal/coordinator/app/` | Assembly shared by both coordinator entry points (`Profile`, config, gRPC server, watcher) |
| `internal/coordinator/adapters/bundle_grpc_server.go` | `BundleCoordinatorServer`: converts the bundle proto into shared `CoordinatorServer` calls |
| `internal/worker/app/app.go` | Assembly shared by both worker entry points; `newBuilderRegistry` / `newPluginRegistry` register by profile; `validateProfileRuntimeConfig` enforces `BatchWriteAddr` |
| `internal/plugin/callbuilder/bundlescan/` | `BundleScanBuilder` (`Build` + `PrepareStream`), DuckDB parquet reader, source resolution, row→call encoding |
| `internal/duckdb/borrowed/` | Low-allocation foundation layer over the DuckDB C bindings: connection pool, ctx interruption, zero-copy row views; contains no bundle business policy |
| `internal/plugin/event/batchwrite/` | Sole orchestrator of the BatchWrite Job lifecycle: row planning/encoding, presigned PUT, Init/Commit/Cancel |
| `internal/plugin/event/bundlewrite/` | `BundleWritePlugin`: `BlockTableOverwriteStrategy{BlockBundle}` |
| `internal/plugin/event/tableupserts/` | `TableUpserts` dual mode: block calls `UpsertRows` directly, bundle goes through `NormalTableUpsertStrategy` BatchWrite |
| `internal/sdk/iceberg/` | Resolver from `tableId` to the current snapshot's live data files (Glue + Iceberg), via the IO subsystem system cache |
| `internal/worker/adapters/orchestrator_phases.go` | Worker-side streaming production: `PrepareStream` under builderSem, `producer.Run` under scanSem |
| `e2e/perf/m3_stream_build_test.go` | `TestPerf_M3_BundleStreamMemory`: RSS / heap / phase-duration comparison with stream build on/off |
| `docs/specs/2026-07-21-block-bundle-clusters.md` and others | See "Related docs" at the end |
## 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:
```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.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](/en/components/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
```mermaid theme={null}
flowchart LR
caller["bundle caller"] -->|"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 (one task)"]
prep["bundlescan.PrepareStream
static validation / Iceberg resolution / schema"]
prod["bundleScanProducer.Run
DuckDB read_parquet streaming scan
emit in batches (8192 rows / 32MiB)"]
disp["dispatcher / executor
(generic Call execution)"]
wr["bundlewrite or tableupserts
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")]
```
The caller first invokes `BundleCoordinatorService.ReserveWorkerSlot` on the bundle coordinator. After receiving `worker_addr` / `slot_id`, it submits `SubmitTask` to the target worker.
Scheduling, slot TTL, and heartbeat match the block cluster. See [Coordinator](/en/components/coordinator).
`BundleScanBuilder.PrepareStream` deserializes `BundleScanBuilderDecl`. For each source, it calls `prepareBundleSource` to perform static preparation:
* **Resolve the data source**: `resolveBundleSourceBucket` uses `bundleBucket` directly when it is non-empty. Otherwise, it reads logical-types to identify an event or state table. State tables map to `._archive`, then `ResolveDataFilesReq` resolves the unique parquet URI.
* **Handle missing data**: when the result is `NO_BUNDLE_FILE` or the snapshot does not contain the bundle, it skips the source and produces 0 calls.
* **Normalize the call**: normalize inline code. For archived state tables, also remap `id` in the operator to `original_id`.
* **Validate the scan plan**: dry-run `validateDuckDBReadPlan` and `ToFilterSQLArgs`, then prebuild `ParquetJSONEncoder`.
* **Choose the cache policy**: `markProvablyUniqueSources` decides which sources' calls set `NoResultCache`.
**Failure semantics**: any static preparation failure becomes `BUILDER_FAILED`, and no calls are dispatched.
The Orchestrator starts `runCallProduction` under `scanSem`:
* **Scan concurrently**: `bundleScanProducer.Run` uses an errgroup to scan at most 3 sources concurrently.
* **Build calls**: `bundleRowCallBuilder.buildNext` encodes each row into a `Call`. Row args are written directly into the argspool buffer with `ArgsPooled=true`.
* **Split batches**: `bundleSourceBatchAppender` runs `flush` and `emit` when either `maxRows` or `maxArgsBytes` reaches its limit.
* **Apply backpressure**: `emit` blocks on `MaxOutstandingBatches` credits, limiting the number of unfinished batches.
**Failure semantics**: args for a single row that exceed the byte limit return a production error. If `emit` returns `ErrProductionStopped`, the worker stopped in a controlled fast-fail path; this is not a builder failure.
This is the shared path. See [Call execution subsystem](/en/components/call-execution) and [Plugin system](/en/components/plugins).
After the Calls converge successfully, `BundleWritePlugin.Execute` validates `targetTable` / `blockBundle`, reads the column definitions, and calls `batchwrite.Writer.Submit`:
1. **Plan**: `planDelimitedWriteRows` walks outputs without retaining them and computes the exact row and byte counts. More than 10,000,000 rows or 5,000,000,000 bytes fails before Init.
2. **Init**: `InitWriteJob` creates a Job and returns `job_id` and `data_url`.
3. **PUT**: the encoder streams through `io.Pipe` in one HTTP `PUT` to the presigned URL. The request uses an exact `Content-Length` and does not follow redirects.
4. **Commit**: after a successful upload, call `CommitWriteJob(row_count)`.
5. **Cancel**: if PUT fails or ctx is canceled before Commit, call `CancelWriteJob` on a best-effort basis. This call uses `context.WithoutCancel` and an independent 5s timeout.
On success, it returns `written_rows` / `finish_time` / `job_id`. `DebugMode` only previews output and does not create a Job.
`TableUpsertsResultHandler` uses different write paths on the two worker types:
* **bundle worker**: reuses `batchwrite.Writer` with `NormalTableUpsertStrategy{Condition, UpdateColumns}`.
* **Validation before Init**: `planNormalUpsertDelimitedWriteRows` requires every row to have the same valid-field set, a non-empty `id`, and at least one non-ID column. It writes the `update_columns` derived from the first row explicitly into the request.
* **block worker**: the handler with the same name still calls `TableWriteService.UpsertRows(sync=false)`.
* **Configuration boundary**: the profile fixes the write path at assembly time. The task configuration has no `writeApi`.
## 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
Classification lives in `batchwrite/classified_error.go` and `presigned_upload.go`. The plugin passes the classification through as `PluginError.Retryable`.
| Classification | Typical case | Reschedulable |
| --------------------------------------- | -------------------------------------------------------------------------------------------------- | ------------- |
| Parameter or response error | Configuration, validation, encoding, invalid response, redirect, other 4xx | No |
| Transient downstream or transport error | Transient Meta / BlockDB IO, HTTP transport error, PUT 408 / 425 / 429 / 500 / 502 / 503 / 504 | Yes |
| Uncertain Commit state | Commit request or response is uncertain; becomes `commitStatusUnknownError` and preserves `job_id` | No |
### Iceberg resolution cache
`ResolveDataFilesReq.CacheKey()` uses only the logical table ID. Requests go through the Worker system cache. The default TTL is 20 minutes and is controlled by `SystemIOCacheTTLms`.
When the cached index's high-water mark covers `RequiredBundle`, `AcceptCachedRead` reuses it. Otherwise, it refreshes the full table index once.
## Configuration
The following configuration is most relevant to the bundle worker. Struct fields come from `internal/worker/app/config.go`, `internal/worker/core/config.go`, and `internal/worker/adapters/orchestrator.go`.
Configuration precedence is: built-in defaults → `TuneDefaults` → `WORKER_CONFIG` file → environment variables.
| Field / environment variable | Default | Description |
| ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | -------------------------------------------------------------------- | ---------------------------------------------------------------------------------------------------------- |
| `workerRegistryPrefix` / `WORKER_REGISTRY_PREFIX` | `/blockx/bundle-workers/` (profile default) | Must be a `bundle-workers`-kind prefix; test environments set `/blockx/test/lanes/default/bundle-workers/` |
| `streamBuild.enabled` / `STREAM_BUILD_ENABLED` | bundle profile `true` (built-in `false`) | Disabling falls back to the fully materialized `Build()` (rollback path) |
| `streamBuild.batchMaxRows` / `STREAM_BUILD_BATCH_MAX_ROWS` | `8192` | Max rows per batch |
| `streamBuild.batchMaxArgsBytes` / `STREAM_BUILD_BATCH_MAX_ARGS_BYTES` | `32 << 20` (32MiB) | Cap on total `ArgsJSON` bytes per batch; a single row over the cap fails |
| `streamBuild.maxOutstandingBatches` / `STREAM_BUILD_MAX_OUTSTANDING_BATCHES` | `2` | Producer credit window; steady-state request-side memory is about `(2 + 3) × 32MiB` |
| `admission.scanSlots` / `SCAN_SLOTS` | `8` | Number of concurrent streaming productions (each ×3 sources internally) |
| `admission.builderSlots` / `executorTaskSlots` / `pluginSlots` | `16` / `32` / `16` | Phase admission; not separately tuned for bundle |
| `worker.taskSlots` / `TASK_SLOTS` | `50` | task slots |
| `dispatcher.executorSelectionSampleSize` / `EXECUTOR_SELECTION_SAMPLE_SIZE` | bundle profile `3` (built-in `0` = whole pool) | Each root call dispatch scores only 3 executors |
| `blockdb.batchWriteAddr` / `BLOCKDB_BATCH_WRITE_ADDR` | empty | **Required for the bundle profile**; empty fails startup |
| `icebergNamespace` / `ICEBERG_NAMESPACE`, `icebergRegion` / `ICEBERG_REGION` | empty (required) / `ap-northeast-1` | Glue namespace; the `tableId` path depends on it |
| `io.systemIoCacheTTLms` / `SYSTEM_IO_CACHE_TTL_MS` | 20 minutes | TTL for cached Iceberg resolution results |
| `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` | empty = DuckDB defaults; `MaxOpenConns` defaults to `max(NumCPU, 4)` | Environment variables only, read by `duckDBParquetReaderConfigFromEnv` |
| `memoryDiagnosticsIntervalMs` / `WORKER_MEMORY_DIAGNOSTICS_INTERVAL_MS` | 15 minutes | Periodically logs memory including `DuckDBDiagnostics` (memory, temp disk, connection pool) |
bundle coordinator: `COORDINATOR_REGISTRY_PREFIXES` (a comma-separated exact list) replaces the profile default `[/blockx/bundle-workers/, /blockx/prod/lanes/default/bundle-workers/]` as a whole; all other configuration is the same as the block coordinator.
## 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
```bash theme={null}
# 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](/en/development/testing) for an overview of test organization.
See `cmd/bundle-loadgen/README.md` in the repository for full `cmd/bundle-loadgen` usage.
* `--coordinator` points to the bundle coordinator.
* `--input-uri` points to a parquet fixture with a verified unique column.
* `--workload noop|cpu|blockdb|function` selects the workload.
* `--result-cache disabled|enabled` and `--result-shape` are control variables for CallCache / output-encoding A/B tests.
* `--result-handler-type BlockBundleWriteResultHandler` connects a real writer, but you must use an isolated test table.
* The worker must set `STREAM_BUILD_ENABLED=true`. The `blockdb` / `function` scenarios also depend on worker stubs such as `STUB_BLOCKDB_DELAY_MS`.
## Related docs
`docs/specs/` in the blockx repo:
* `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.
On this site:
* [Plugin system](/en/components/plugins): shared Builder / Writer contracts.
* [Worker](/en/components/worker): phase admission and orchestrator.
* [Coordinator](/en/components/coordinator): scheduling and watcher.
* [Call execution subsystem](/en/components/call-execution): shared Calls phase.
* [IO access subsystem](/en/components/io-subsystem): system cache and retry classification.
* [Backend Adapter](/en/components/backend-adapter): BlockDB / Iceberg clients.
* [Deployment overview](/en/development/deployment): process shapes and release methods.