> ## 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.
# IO access subsystem
> The IO layer inside the Worker that fronts all BlockDB / NodeRPC / capability backend access: WorkerIOScope, TaskIOScope, caching, singleflight, adaptive admission, and backend adapter assembly
The IO access subsystem is a layer of resource-governance code inside the Worker process. Every BlockDB / RPC / capability backend access from the Builder, from user functions in the Executor (relayed back through the Python SDK and UDS), and from Writer plugins goes through it before the backend adapters make the real network calls. See [Architecture overview](/en/architecture/overview) for where it sits in the system.
It is not organized in the core / adapter event-command pattern; instead it uses a **Scoped Resource Context**: task-level scopes (`TaskIOScope`) nest inside the Worker-level shared resources (`WorkerIOScope`), and the request lifecycle nests inside the task lifecycle.
## Responsibilities and boundaries
Responsible for:
* Providing the single synchronous entry point `TaskIOScope.Read / Write`, shared by the Builder, the Executor IO relay, and plugins.
* A fixed request path: Worker-level read cache → task-local singleflight → task IO window → backend admission → adapter.
* Holding the Worker-level shared read cache (IO cache) and a separate system cache.
* Bounded retries for requests that carry a Builder / Result retry scope.
* Binding the request lifecycle to the task: once the scope closes, in-flight requests are canceled, new requests are rejected, and late results are discarded.
* A unified error model (`model.IOError`) and per-task IO statistics.
Not responsible for:
* Deciding whether a call gets dispatched (that belongs to the Dispatcher; see [Call execution subsystem](/en/components/call-execution)).
* The concrete logic of real protocol calls, connection pools, and error classification — these live in the adapters under `internal/sdk/*`; see [Backend Adapter](/en/components/backend-adapter).
* The call result cache (owned by the Call execution subsystem).
* Cross-task write idempotency or transaction recovery.
Core invariants:
* Each backend kind has exactly one admission controller, wrapped unconditionally by `assembly.Build` (`adaptive.WrapBackend`); there is no second shared quota layer in front of `TaskIOScope`.
* Write requests are never cached and never go through singleflight, but they do pass through the task window and backend admission.
* `req.Timeout()` only bounds the real adapter call after a permit is obtained; waiting for the task window and admission has no local timeout and is bounded only by the request context and the scope lifecycle.
* An empty `CacheKey()` means "not cacheable": the cache and singleflight are skipped and the request goes straight to the window + adapter.
* `TaskIOScope.Close` is idempotent; after closing, nothing may be written back into the cache or singleflight.
* IO cache keys consist of a `(backend, operation)` namespace plus a normalized request key, so different backends / interfaces never collide.
## Code location
| Path | Purpose |
| ---------------------------------------------------------------------- | -------------------------------------------------------------------------------------------------------------------------------------------------- |
| `internal/io/core/worker_scope.go` | `WorkerIOScope`: adapter registry, IO cache / system cache, active task scope table, cache namespaces |
| `internal/io/core/task_scope.go` | `TaskIOScope`: `Read / Write / Close`, singleflight leader logic, late-result checks |
| `internal/io/core/retry.go` | `RetryScope`, `WithRetryScope`, `runIORetry` bounded retry and backoff |
| `internal/io/core/task_scope_stats.go` and others | `Snapshot / IOStats / BackendHits`, wait-error classification, `classifyBackendError` |
| `internal/io/core/model/` | `BackendAdapter`, `ReadReq / WriteReq`, `IOError`, `WorkerIOConfig`, `BackendRegistration`, result and statistics types |
| `internal/io/core/stats/` | Hit counters, per-call IO durations, per-(backend, operation) CKMS quantiles, IO error drain clustering |
| `internal/io/adaptive/` | Adaptive concurrency `Limiter` (windowed AIMD + soft-limit probing + fair lanes), `Backend` wrapper, `AIMDConfig`, transport-layer error detection |
| `internal/io/adaptive/observed/` | `WrapBackendRegistration`: wraps with the adaptive layer and registers the `blockx_io_adaptive_*` metrics |
| `internal/io/assembly/` | `Module` declaration + `Build`: builds `[]BackendRegistration` from a module list, enforces admission wrapping, stub fallback |
| `internal/io/cache/` | Generic TTL + LRU + byte-accounting cache `cache.Cache[K]` |
| `internal/io/sflight/` | Generic singleflight `sflight.Table[K]` |
| `internal/io/quota/` | Channel semaphore `quota.Semaphore` used for the task IO window |
| `internal/io/capabilitywire/` | `DecodeBinaryRequest`: unpacks the UDS binary envelope of gRPC-style capability requests |
| `internal/worker/app/app.go`, `internal/worker/app/config.go` | Worker-side module assembly table, `DefaultIOConfig`, env overrides |
| `cmd/syncinvoker/io.go` | Sync Invoker-side module assembly (shared read cache disabled) |
| `internal/worker/adapters/orchestrator_dispatch.go` | `Orchestrator.HandleIO`: the entry point where Executor IO requests enter `TaskIOScope` |
| `internal/worker/adapters/orchestrator_phases.go` | `NewTaskScope` on task activation, `Close("task_terminal")` on terminal state, retry scope tagging for the Builder / Writer phases |
| `internal/sdk/localtestservice/` | Live template for a capability backend (adapter + service + proto) |
| `internal/io/**/*_test.go` | Unit tests and benchmarks |
| `docs/specs/io-subsystem.md` | Design spec (blockx repo) |
| `docs/io-backend-module-design.md`, `docs/capability-backend-guide.md` | Backend module design and the how-to guide for adding a capability backend (blockx repo) |
## Core types and interfaces
* `model.BackendAdapter` (`internal/io/core/model/adapter.go`): the minimal interface an adapter must implement. `Write` returns `(*WriteResult, error)`.
```go theme={null}
type BackendAdapter interface {
Read(ctx context.Context, req ReadReq) (*ReadResult, error)
Write(ctx context.Context, req WriteReq) (*WriteResult, error)
Close() error
}
```
* `model.ReadReq` / `model.WriteReq` (same file): the request interfaces. `ReadReq` has `Backend() BackendKind`, `Operation() string`, `CacheKey() string`, and `Timeout() time.Duration`; `WriteReq` has no `CacheKey`. Concrete request types are defined by the individual SDK packages (e.g. `internal/sdk/blockdb/types.go`); requests relayed from the Executor are implemented by `executor.IORequest`.
* `model.ErrorClassifier`: adapters may optionally implement `ClassifyError(err) *IOError`; without it, all unknown errors are classified as `retryable_io`.
* `model.CachedReadValidator`: an optional capability for cacheable reads, `AcceptCachedRead(data []byte) bool`; returning false bypasses the cached value and joins the singleflight refresh for the same key (the current user is `internal/sdk/iceberg/io_adapter.go`).
* `model.IOError` / `model.IOErrorKind`: the unified error model. `Kind` is one of `system_error`, `param_error`, `retryable_io`, `non_retryable`, `timeout`, `scope_closed`; `Code` is a fine-grained observability code (e.g. `io:no_adapter_registered`, `context:timeout`); `StructuredError()` produces the `errorKind / retryable / detailCode` returned to the Python SDK.
* `model.BackendRegistration`: `Kind`, `Adapter`, `SystemCache bool` (reads go through the system cache), `ReadOnly bool` (`Write` is rejected outright with `io:write_not_supported`).
* `core.WorkerIOScope` (`worker_scope.go`): `NewWorkerIOScope(WorkerIOScopeParams)`, `NewTaskScope(*commontypes.TaskCtx)`, `Close()`, `Snapshot()`. `WorkerIOScopeParams.DisableSharedReadCache` lets the Sync Invoker turn off cross-call result reuse.
* `core.TaskIOScope` (`task_scope.go`): `Read`, `Write`, `Close(reason)`, `Snapshot()`, `IOStats()`, `BackendHits()`, `AddCallIO / FinalizeCallIO`, `BackendIOStats()`, `DrainIOErrorClusters()`.
* `core.RetryScope` / `core.WithRetryScope` (`retry.go`): only the two framework tags `RetryScopeBuilder` and `RetryScopeResult` enable IO-local retry; unknown tags fail closed.
* `adaptive.Backend` / `adaptive.WrapBackend(next, Config)` (`adaptive/backend.go`): wraps any adapter with a dedicated `Limiter`; it also implements `ErrorClassifier`, mapping admission errors to the canonical `context:timeout` / `context:canceled` / `io:temporarily_unavailable`.
* `adaptive.Limiter` / `adaptive.Permit` / `adaptive.Outcome` (`adaptive/limiter.go`): `Acquire(ctx) (*Permit, error)`, `Permit.Done(outcome)` (idempotent); `Outcome` has three values, `OutcomeSuccess / OutcomeOverloaded / OutcomeIgnore`.
* `adaptive.OutcomeClassifier`: adapters may optionally implement `ClassifyAdaptiveOutcome(err) Outcome`; without it, success is recorded as `Success` and errors as `Ignore`. Implementations: `internal/sdk/noderpc/adaptive.go`, `internal/sdk/blockdb/adaptive.go`, `internal/sdk/logicaltypes/adaptive.go`, `internal/sdk/meta/adaptive.go`.
* `adaptive.AIMDConfig` (`adaptive/config.go`): operator-facing AIMD configuration; `ResolveAdmission(initialLimit)` degrades to `FixedConfig` when `Enabled=false`.
* `assembly.Module` / `assembly.Build` (`assembly/assembly.go`): see "Extension points".
* `capabilitywire.DecodeBinaryRequest(kind, req) (method, payload, error)`: see "Extension points".
## Data flow / execution flow
A read request coming from the Executor (Python SDK → UDS → Worker → backend):
```mermaid theme={null}
flowchart TD
A["Python SDK read_req / BridgeChannel"] -->|"UDS CallWaiting, waitKind=io"| B["executor.Adapter.processWaitRequest builds executor.IORequest"]
B --> C["Orchestrator.HandleIO looks up taskRuntime, injects clientid"]
C --> D["TaskIOScope.Read"]
D --> E{"scope closing?"}
E -->|"yes"| Z1["return scope_closed"]
E -->|"no"| F{"CacheKey empty?"}
F -->|"yes, non-cacheable bypass"| L
F -->|"no"| G{"Worker IO cache hit?"}
G -->|"yes, and validator accepts"| Z2["return FromCache=true"]
G -->|"no"| H{"sflight.Join is leader?"}
H -->|"no, as waiter"| I["waitForLeader waits on entry.Done / scopeCtx / ctx"]
H -->|"yes"| L["runIORetry starts one attempt"]
L --> M["quota.Semaphore.Acquire takes the task IO window"]
M --> N["adaptive.Backend.Read: Limiter.Acquire obtains a permit"]
N --> O["backendCallContext applies req.Timeout, calls raw adapter.Read"]
O --> P["permit.Done(outcome), release the window"]
P --> Q{"result"}
Q -->|"retryable_io with a retry scope"| L
Q -->|"other error"| Z3["completeSingleflight broadcasts the error"]
Q -->|"success"| R{"scope already closed?"}
R -->|"yes"| Z4["discard the result, return scope_closed"]
R -->|"no"| S["cache.Set, completeSingleflight broadcasts the data"]
S --> T["ReadResult returns to the Executor via ResumeCall, BudgetUsedMs takes BackendLatencyMs"]
```
Key points:
* The entry point is `processWaitRequest` in `internal/worker/adapters/executor/handlers.go`: it builds an `executor.IORequest` from the `Mode / Operation / Backend / CacheKey / TimeoutMs / Request` fields of `CallWaitingPayload` (using `NewIORequestWithBinary` when a binary sidecar is present, and using the sidecar bytes directly as the cache key when `DeriveCacheKeyFromBinary=true`), then calls `IOHandler.HandleIO`.
* `Orchestrator.HandleIO` (`orchestrator_dispatch.go`) finds the `taskRuntime` by `taskID`, writes the instance id into the ctx (for fair lanes), rejects Executor-originated `blockdb` writes here, then calls `rt.io.Read / Write`; on success it records `BackendLatencyMs` into the per-call IO statistics.
* Return path: `ResumeCall.BudgetUsedMs` takes `ReadResult.BudgetChargedMs()` — 0 for cache / singleflight hits, and the backend duration measured by the adapter for real reads; admission and window queueing time are not charged to the budget. For requests that arrived through the binary sidecar, the result `Data` goes back through the sidecar as well (binary-in → binary-out).
* The Builder and Writer do not go through UDS; they hold the `TaskIOScope` directly (the Builder gets a `cbtypes.TaskIOReader`, plugins get a `types.TaskIO`), with the ctx carrying `RetryScopeBuilder` / `RetryScopeResult` respectively.
* The write path differs from the read path only in that it skips the cache and singleflight and checks `ReadOnly` first; everything else (window, admission, retry, error classification) is the same.
### Cache and namespaces
`WorkerIOScope` holds two `cache.Cache[ioCacheKey]` instances:
* IO cache: 4096 entries / 16 MiB by default, fixed TTL of 1 minute (`workerIOCacheTTL`), shared across tasks, purged with `Purge` when the Worker shuts down.
* system cache: 64 entries / 512 MiB / 20 minutes by default; it only serves backends registered with `SystemCache: true` (currently `iceberg`'s `resolve_data_files`).
In `ioCacheKey{namespace, request}`, `namespace` is the numeric ID of `(backend, operation)`: the known operations of the two BlockDB kinds, `logicalTypes`, `meta`, `router`, `localtestservice`, and `iceberg` have compile-time IDs in `knownIOCacheNamespace`; NodeRPC JSON-RPC methods are an open set and, together with other unknown combinations, go through a bounded (4096) dynamic registry; when the registry is exhausted, the request degrades to a non-cacheable read. The third kind of cache, the call result cache, is not part of this subsystem.
### Adaptive admission
`adaptive.Backend` is the single process-level admission gate of each backend. `Limiter` implements windowed AIMD:
* Within each sampling window (default 1s), the first `Overloaded` triggers an immediate multiplicative decrease (`BackoffRatio` default 0.5, floored at `MinLimit`); a further decrease within the same window requires the new-generation sample count to reach `RepeatBackoffMinSamples` and the overload ratio to reach `RepeatBackoffOverloadRatio`, without exceeding `MaxDecreasesPerWindow`.
* When the `Ignore` ratio reaches `IgnoreRatioThreshold` (0.10), the window loses its eligibility for growth; when it reaches `IgnoreBackoffRatioThreshold` (0.30), the limit shrinks once when the window settles.
* Growth only happens at the end of a window and requires no overload, enough samples, saturation (max in-flight reached the limit), and `IncreaseAfterHealthyWindows` consecutive healthy windows; the step is `IncreaseStep` (0 means 1% of the initial limit, at least 1).
* Above the learned `softLimit`, growth turns into probing: if the probe generation overloads, the limit falls back exactly to `probeBaseLimit`, and consecutive failures back off exponentially in window count (capped at `ProbeBackoffMaxWindows`).
* When the limit drops below in-flight, existing requests are not canceled; the difference is repaid as shrink debt.
* No goroutines are started; windows advance lazily on `Acquire / Done` events.
When `Config.LaneKeyFromCtx` is non-nil, fair lane mode is enabled (`adaptive/lanes.go`): waiters are split into lanes by key, and freed capacity goes to the lane with the fewest in-flight requests; the AIMD policy is unchanged. The Worker enables it for specific backends through the `IOFairQueueBackends` configuration, with the key taken from `clientid.FromContext`.
`observed.WrapBackendRegistration` wraps the backend and, at the same time, registers `blockx_io_adaptive_limit / _pressure / _overloads_total / _ignored_total / _limit_transitions_total / _admission_waits_total / _admission_wait_seconds_total` by `BackendKind`. The remaining IO-layer metrics are `blockx_task_io_ops_total`, `blockx_task_io_backend_duration_milliseconds`, `blockx_task_io_cache_hits_total`, and `blockx_task_io_singleflight_hits_total`, defined in `internal/obs/metrics.go` and `internal/obs/metrics_io_adaptive.go`.
## State and lifecycle
`TaskIOScope` has only two states, "active" and "closing":
* Creation: on task activation, the `Orchestrator` calls `ioScope.NewTaskScope(ioTaskCtx)` (`orchestrator_phases.go`); a failure is recorded as `task_activation_failed`. The scope holds a `scopeCtx` derived from the task ctx.
* Requests: each `Read / Write` derives an `ioCtx`, and `context.AfterFunc(scopeCtx, ioCancel)` lets a scope close cancel in-flight adapter calls; each attempt does its own `Acquire / Release` of the task window, and neither the window nor a permit is held during backoff.
* Close: on task terminal state, `rt.io.Close("task_terminal")`. `Close` uses `CompareAndSwap` to guarantee idempotency, then, in order: cancels `scopeCtx`, calls `sfTable.DrainAll(scope_closed)` to wake all waiters, and removes itself from `WorkerIOScope.activeScopes`.
* Late results: after `runIORetry` returns successfully, the leader checks `closing` once more; if already closed, it returns `scope_closed` without writing to the cache or broadcasting; `sflight.Table.Complete` and `DrainAll` perform an idempotent close under the same lock to avoid double-close panics.
* Worker shutdown: `WorkerIOScope.Close` first closes all active task scopes, purges both caches with `Purge`, then calls `adapter.Close()` one by one (the adaptive wrapper first closes the limiter to wake waiters, then closes the raw adapter).
Error classification and retry:
* Wait-phase errors are normalized by `classifyWaitContextError`: scope already closing → `scope_closed` / `io:temporarily_unavailable`; `DeadlineExceeded` → `timeout` / `context:timeout`; `Canceled` → `scope_closed` / `context:canceled`; anything else → `system_error`. There is no separate "queue timeout" error.
* Backend errors are normalized by `classifyBackendError`: a `*IOError` passes through as-is → the adapter's `ClassifyError` → otherwise `retryable_io`.
* `shouldRetryIO` only allows a retry when `ioErr.Retryable()`, `RetryMaxRetries` is not exhausted, the ctx carries a known retry scope, and the ctx is not done. Backoff uses `backoff/v5`: by default three windows of 250–500ms, 500ms–1s, and 1–2s, capped at 3s per wait.
* Write retry is at-least-once: IO core simply replays the current typed write request. All BlockDB APIs (including BundleWrite's `InitWriteJob / CommitWriteJob / CancelWriteJob`) guarantee idempotency, so BlockDB writes inside the Result scope remain retryable; the earlier `WithoutRetryScope` opt-out mechanism was removed in commit `aae6b77a`. When onboarding a write API that does not satisfy the idempotency contract, adjust the retry scope or the backend contract first.
* The singleflight leader owns the entire retry loop; waiters only wait for the final result; the cache is only backfilled with the final successful value.
## Configuration
Worker-side IO configuration lives in `WorkerFullConfig.IO` (type `iocore.WorkerIOConfig`, defined in `internal/io/core/model/adapter.go`; `internal/io/core/types.go` only aliases it; defaults come from `DefaultIOConfig` in `internal/worker/app/config.go`):
| Field (JSON) | Default | Description |
| ------------------------------------ | ------- | ----------------------------------------------------------------------------------------------------------------------------------------------------- |
| `io.taskMaxInflightIO` | 1024 | Per-task in-flight IO limit (task window). No env override |
| `io.retryMaxRetries` | 3 | Maximum number of additional retries after the first attempt for logical IO carrying a retry scope; 0 disables. env `WORKER_IO_RETRY_MAX_RETRIES` |
| `io.workerIoCacheMaxEntries` | 4096 | IO cache entry limit. env `WORKER_IO_CACHE_MAX_ENTRIES` (`TASK_IO_CACHE_MAX_ENTRIES` accepted for compatibility) |
| `io.workerIoCacheMaxBytes` | 16 MiB | Estimated resident byte limit of the IO cache. env `WORKER_IO_CACHE_MAX_BYTES` (`TASK_IO_CACHE_MAX_BYTES` accepted for compatibility) |
| `io.systemIoCacheMaxEntries` | 64 | system cache entry limit. env `SYSTEM_IO_CACHE_MAX_ENTRIES` |
| `io.systemIoCacheMaxBytes` | 512 MiB | system cache byte limit. env `SYSTEM_IO_CACHE_MAX_BYTES` |
| `io.systemIoCacheTtlMs` | 1200000 | system cache TTL; a negative value disables expiry. env `SYSTEM_IO_CACHE_TTL_MS` |
| `io.systemIoCacheRefreshConcurrency` | 2 | Concurrency limit for Iceberg full-table planning; also the fixed admission limit of the `iceberg` backend. env `SYSTEM_IO_CACHE_REFRESH_CONCURRENCY` |
Backend admission configuration lives at the top level of `WorkerFullConfig`:
| Field (JSON) | Default | Description |
| -------------------------------------------------------------------------- | ------------------------------ | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| `nodeRpcInitialLimit` / `blockdbInitialLimit` / `logicalTypesInitialLimit` | 1024 / 16 / 16 | Startup concurrency of each backend (not a static quota). env `NODE_RPC_INITIAL_LIMIT`, `BLOCKDB_INITIAL_LIMIT`, `LOGICAL_TYPES_INITIAL_LIMIT`; `blockdb` and `blockdb_executor` share the same BlockDB policy but each has its own limiter |
| `nodeRpcAimd` / `blockdbAimd` / `logicalTypesAimd` | `adaptive.DefaultAIMDConfig()` | `enabled=true`, `minLimit=1`, `maxLimit=0` (no upper bound), `increaseAfterHealthyWindows=2`, `backoffRatio=0.5`, `windowMs=1000`, `minSamples=20`, `repeatBackoffOverloadRatio=0.30`, `repeatBackoffMinSamples=20`, `maxDecreasesPerWindow=2`, `ignoreRatioThreshold=0.10`, `ignoreBackoffRatioThreshold=0.30`, `probeBackoffMaxWindows=64`. env prefixes `NODE_RPC_AIMD_*`, `BLOCKDB_AIMD_*`, `LOGICAL_TYPES_AIMD_*` (see `applyAIMDEnvOverrides`) |
| `ioFairQueueBackends` | empty | List of backend kinds with fair lanes enabled. env `IO_FAIR_QUEUE_BACKENDS` |
| `localTestServiceDisabled` / `routerDisabled` | false | Left out of the assembly table; calls get `io:no_adapter_registered`. env `LOCAL_TEST_SERVICE_DISABLED`, `ROUTER_DISABLED` |
| `phaseIoTimeoutMs` | 10000 | `req.Timeout()` for Builder / Writer phase IO, independent of the call budget. env `PHASE_IO_TIMEOUT_MS` |
The backend kinds registered by the Worker and Sync Invoker assembly tables are `rpc`, `blockdb`, `blockdb_executor`, `logicalTypes`, `iceberg`, `localtestservice`, and `router`; `internal/sdk/meta` defines `BackendMeta`, but neither assembly table registers it.
The Sync Invoker (`cmd/syncinvoker/io.go`) only sets `TaskMaxInflightIO: 1024` and constructs `WorkerIOScope` with `DisableSharedReadCache: true`.
## Extension points
### Adding a capability backend
Use `internal/sdk/localtestservice/` as the template (`docs/capability-backend-guide.md` is the file-by-file guide); no platform code changes are needed. On the Go side you need to:
Under `internal/sdk//`, declare `const Backend iocore.BackendKind = ""` (never reuse an existing kind), write `proto/.proto` as the gRPC service definition, and generate code into `gen/`.
`service.go` implements the generated server interface; all business logic lives there. `adapter.go` implements `model.BackendAdapter`: `Read` uses `capabilitywire.DecodeBinaryRequest(string(Backend), req)` to extract `(grpc_method, protobuf bytes)`, calls the service after `proto.Unmarshal`, and marshals the response with `proto.Marshal` into `ReadResult.Data`; unknown methods return `IOErrParam` + `:unsupported_method`, and a malformed envelope / proto returns `:bad_request`. `Write` of a read-only capability is never called (the module declares `ReadOnly: true`). Backends with external dependencies should also implement `model.ErrorClassifier` and `adaptive.OutcomeClassifier`.
Export `Module()` (optionally taking dependencies as parameters) returning `assembly.Module{Kind, ReadOnly, Build, Stub, Admission}`. Lightweight in-process backends use `adaptive.FixedConfig(n)`; backends with external dependencies use `AIMDConfig.ResolveAdmission`, with `Enabled` checking whether the configuration is complete and `Stub` providing the dev fallback (without one, startup is refused when unconfigured). To borrow another backend, inject the raw adapter the way `localtestservice.BlockDBReader` does: borrowing does not own it (`Close` does not close it), admission is the borrower's own responsibility, and errors caused by the local ctx ending are passed through as-is before delegating classification to the dependency.
Add a switch field in `internal/worker/app/config.go`, add the env override in `internal/worker/app/app.go` and `backendModules = append(backendModules, .Module(...))`; append likewise in `cmd/syncinvoker/io.go`. Add a line to `deploy/env/worker.env.example`.
`adapter_test.go` covers: the happy path, parameter defaults, unknown operation, malformed envelope, malformed proto, `Write` rejection, and the `Module()` declaration; composite backends additionally cover typed request forwarding, clean failure when the dependency is not injected, and pass-through of the dependency's error classification. For e2e, in `cmd/worker/` follow `worker_process_localtestservice_test.go` to write a positive case (task SUCCEEDED under `FUNCTION_CODE_AUDIT_MODE=enforce`) and a `_DISABLED=true` negative case (task FAILED); devstub functions are registered in `internal/worker/app/function_code.go`.
For the Python side (the blockx-py class, the `connection.py` channel getter, the executor `BridgeChannel` wiring, and the five table additions in `python/blockx_audit/tables.py`), see [Python Executor](/en/components/python-executor) and `docs/capability-backend-guide.md` §3–§4.
The fields of `assembly.Module`:
```go theme={null}
type Module struct {
Kind iocoremodel.BackendKind
SystemCache bool
ReadOnly bool
Enabled func() bool
Build func(adm adaptive.Config) (iocoremodel.BackendAdapter, error)
Stub func() iocoremodel.BackendAdapter
Admission func() (adaptive.Config, error)
OuterWrap func(inner iocoremodel.BackendAdapter) (iocoremodel.BackendAdapter, error)
}
```
`OuterWrap` is the only layer allowed outside admission (currently only the NodeRPC semantic batching `noderpc.NewBatchingBackend`, gated by `NODE_RPC_SEMANTIC_BATCH_MODE`, default `off`).
### Other common changes
* Adding a compile-time cache namespace for a hot `(backend, operation)`: modify `knownIOCacheNamespace` and the corresponding enum in `worker_scope.go`, and update `TestKnownIOCacheNamespacesAreDistinctEnums`.
* Tuning a backend's AIMD feedback: modify `adaptive.go` (`ClassifyAdaptiveOutcome`) in that SDK package; do not touch `internal/io/adaptive`; use `adaptive.IsTransportFailure / IsTransportFailureMessage` to detect transport-layer errors.
* Adding a cacheable read that needs per-request validation: have the request type implement `model.CachedReadValidator`.
## Testing
```bash theme={null}
go test ./internal/io/...
go test ./internal/io/core/ -run 'TestTaskScope|TestWorkerScope'
go test ./internal/io/adaptive/ -run 'TestLimiter|TestFairLane|TestBackend'
go test ./internal/sdk/localtestservice/ ./internal/io/...
go test -count=1 -run TestWorkerProcess_LocalTestService ./cmd/worker/
```
Test organization:
* `internal/io/core/core_test.go`: the request-path backbone — cache hits / cross-task sharing (`TestWorkerScope_CacheIsSharedAcrossTaskLifetimes`), singleflight (`TestTaskScope_ReadSingleflight`), task window acquire / release (`TestTaskScope_TaskWindowWaitsUntilCapacityIsReleased`, `TestTaskScope_IOBackendAttemptReleasesTaskWindow`), scope close and late results (`TestTaskScope_ActiveBackendContextLifecycle`, `TestTaskScope_ReadAfterClose`, `TestTaskScope_CloseIdempotent`, `TestWorkerScope_CloseCascades`), write rejection on read-only backends, and empty cacheKey bypass (`task_scope_bypass_test.go`).
* `internal/io/core/retry_test.go`: retry scope gating, a single leader owning the retry, resource release between attempts, backoff boundaries.
* `internal/io/core/task_scope_drain_test.go`: concurrent `Complete` and `DrainAll` do not double-close.
* `internal/io/adaptive/limiter_test.go`, `lanes_test.go`, `backend_test.go`, `backend_budget_test.go`: each AIMD rule, probing, fair lanes, admission queueing not charged to the budget, timeout starting only after admission.
* `internal/io/assembly/assembly_test.go`: stub fallback, duplicate kind, closing already-built adapters on failure, `OuterWrap`.
* `internal/io/capabilitywire/wire_test.go`: three-way envelope binding, shape errors failing closed.
* `internal/io/sflight/sflight_test.go`, `internal/io/cache/cache_test.go`: standalone unit tests for singleflight and the cache.
* Benchmarks: `internal/io/core/bench_test.go`, `adaptive/lanes_bench_test.go`.
* Process-level e2e tests that involve the Executor live in `cmd/worker/worker_process_*_test.go` (e.g. `worker_process_blockdb_bridge_test.go`, `worker_process_localtestservice_test.go`). See [Test organization and commands](/en/development/testing) for more.
## Related docs
Specs and design docs in the blockx repo:
* `docs/specs/io-subsystem.md`: the behavioral spec of this subsystem (Scoped Resource Context, adapter responsibilities, AIMD feedback classification, Close semantics, cache policy and fork risks, configuration items).
* `docs/io-backend-module-design.md`: design motivation and invariants of the backend module assembly layer.
* `docs/capability-backend-guide.md`: file-by-file guide and checklist for adding a capability backend.
* `docs/specs/architecture.md` §4.2.5 / §4.2.6: where the IO access subsystem and the shared state cache sit in the overall architecture.
* `docs/specs/noderpc-semantic-batching.md`: NodeRPC semantic batching (the user of `OuterWrap`).
* `docs/deploy.md`: deployment notes for the IO / AIMD-related environment variables.
Related pages on this site:
Host of WorkerIOScope; when task scopes are created and closed.
Concrete implementations and error classification of the BlockDB / NodeRPC / logical-types / Iceberg adapters.
How IO requests are intercepted on the Python side and relayed back over UDS.
The Orchestrator that hosts HandleIO, and the call result cache.