IO access subsystem

​

Responsibilities and boundaries

• 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.

• Deciding whether a call gets dispatched (that belongs to the Dispatcher; see Call execution subsystem).
• The concrete logic of real protocol calls, connection pools, and error classification — these live in the adapters under internal/sdk/*; see Backend Adapter.
• The call result cache (owned by the Call execution subsystem).
• Cross-task write idempotency or transaction recovery.

• 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

​

Core types and interfaces

• model.BackendAdapter (internal/io/core/model/adapter.go): the minimal interface an adapter must implement. Write returns (*WriteResult, error).

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

• 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

• 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).

​

Adaptive admission

• 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.

​

State and lifecycle

• 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).

• 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

​

Extension points

​

Adding a capability backend

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)
}

​

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

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/

• 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 for more.

​

Related docs

• 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.