IO 访问子系统

​

职责与边界

• 提供 TaskIOScope.Read / Write 这一个同步入口，供 Builder、Executor IO 回送、Plugin 共用。
• 请求路径固定为：Worker 级读缓存 → task 内 singleflight → task IO 窗口 → backend admission → adapter。
• 持有 Worker 级共享读缓存（IO cache）和独立的 system cache。
• 对带 Builder / Result retry scope 的请求执行有界重试。
• 把请求生命周期绑定到 task：scope 关闭后取消在途请求、拒绝新请求、丢弃迟到结果。
• 统一错误模型（model.IOError）与 per-task IO 统计。

• 决定 call 是否被调度（那是 Dispatcher 的事，见 Call 执行子系统）。
• 真实协议调用、连接池、错误分类的具体逻辑——这些在 internal/sdk/* 的 adapter 里，见 后端适配器。
• call result cache（归 Call 执行子系统）。
• 跨 task 的写幂等或事务恢复。

• 每个 backend kind 只有一个 admission controller，由 assembly.Build 强制包装（adaptive.WrapBackend）；TaskIOScope 前面不再叠第二层共享 quota。
• 写请求不进缓存、不做 singleflight，但同样经过 task 窗口和 backend admission。
• req.Timeout() 只约束获得 permit 之后的真实 adapter 调用；task 窗口和 admission 排队没有本地超时，只受请求 context 与 scope 生命周期约束。
• 空 CacheKey() 表示”不可缓存”，跳过缓存和 singleflight，直接走窗口 + adapter。
• TaskIOScope.Close 幂等；关闭后不得再回填缓存或 singleflight。
• IO cache key 由 (backend, operation) 命名空间 + 请求归一化 key 组成，不同后端 / 接口不会碰撞。

​

代码位置

​

核心类型与接口

• model.BackendAdapter（internal/io/core/model/adapter.go）：adapter 必须实现的最小接口。Write 返回 (*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（同文件）：请求接口。ReadReq 有 Backend() BackendKind、Operation() string、CacheKey() string、Timeout() time.Duration；WriteReq 没有 CacheKey。具体请求类型由各 SDK 包定义（如 internal/sdk/blockdb/types.go），Executor 回送的请求由 executor.IORequest 实现。
• model.ErrorClassifier：adapter 可选实现 ClassifyError(err) *IOError；没实现时未知错误一律归 retryable_io。
• model.CachedReadValidator：可缓存读的可选能力 AcceptCachedRead(data []byte) bool，返回 false 时旁路缓存值并进入同 key 的 singleflight 刷新（当前使用者是 internal/sdk/iceberg/io_adapter.go）。
• model.IOError / model.IOErrorKind：统一错误模型。Kind 取值 system_error、param_error、retryable_io、non_retryable、timeout、scope_closed；Code 是细粒度可观测码（如 io:no_adapter_registered、context:timeout）；StructuredError() 生成回给 Python SDK 的 errorKind / retryable / detailCode。
• model.BackendRegistration：Kind、Adapter、SystemCache bool（读走 system cache）、ReadOnly bool（Write 直接以 io:write_not_supported 拒绝）。
• core.WorkerIOScope（worker_scope.go）：NewWorkerIOScope(WorkerIOScopeParams)、NewTaskScope(*commontypes.TaskCtx)、Close()、Snapshot()。WorkerIOScopeParams.DisableSharedReadCache 供 Sync Invoker 关闭跨 call 结果复用。
• core.TaskIOScope（task_scope.go）：Read、Write、Close(reason)、Snapshot()、IOStats()、BackendHits()、AddCallIO / FinalizeCallIO、BackendIOStats()、DrainIOErrorClusters()。
• core.RetryScope / core.WithRetryScope（retry.go）：只有 RetryScopeBuilder、RetryScopeResult 两个框架标签能开启 IO-local retry；未知标签 fail-closed。
• adaptive.Backend / adaptive.WrapBackend(next, Config)（adaptive/backend.go）：给任意 adapter 外包一个独占 Limiter；同时实现 ErrorClassifier，把 admission 错误映射为 canonical 的 context:timeout / context:canceled / io:temporarily_unavailable。
• adaptive.Limiter / adaptive.Permit / adaptive.Outcome（adaptive/limiter.go）：Acquire(ctx) (*Permit, error)、Permit.Done(outcome)（幂等）；Outcome 三值 OutcomeSuccess / OutcomeOverloaded / OutcomeIgnore。
• adaptive.OutcomeClassifier：adapter 可选实现 ClassifyAdaptiveOutcome(err) Outcome；没实现时成功记 Success，错误记 Ignore。实现者见 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）：运维面的 AIMD 配置；ResolveAdmission(initialLimit) 在 Enabled=false 时退化为 FixedConfig。
• assembly.Module / assembly.Build（assembly/assembly.go）：见”扩展点”。
• capabilitywire.DecodeBinaryRequest(kind, req) (method, payload, error)：见”扩展点”。

​

数据流 / 执行流程

• 入口是 internal/worker/adapters/executor/handlers.go 的 processWaitRequest：按 CallWaitingPayload 的 Mode / Operation / Backend / CacheKey / TimeoutMs / Request 构造 executor.IORequest（带二进制 sidecar 时用 NewIORequestWithBinary，DeriveCacheKeyFromBinary=true 时以 sidecar 字节直接做 cache key），再调 IOHandler.HandleIO。
• Orchestrator.HandleIO（orchestrator_dispatch.go）按 taskID 找到 taskRuntime，把 instance id 写进 ctx（供公平 lane 用），Executor 来源的 blockdb 写在这里被拒绝，然后调 rt.io.Read / Write；成功后把 BackendLatencyMs 记进 per-call IO 统计。
• 回程：ResumeCall.BudgetUsedMs 取 ReadResult.BudgetChargedMs()——缓存 / singleflight 命中为 0，真实读为 adapter 测得的后端时长；admission 和窗口排队时间不计入 budget。请求经二进制 sidecar 到达的，结果 Data 也走 sidecar 回去（binary-in → binary-out）。
• Builder 与 Writer 不经 UDS，直接持有 TaskIOScope（Builder 拿的是 cbtypes.TaskIOReader，plugin 拿 types.TaskIO），ctx 分别带 RetryScopeBuilder / RetryScopeResult。
• 写路径与读路径的差别只在：跳过缓存与 singleflight、先检查 ReadOnly；其余（窗口、admission、重试、错误分类）相同。

​

缓存与命名空间

• IO cache：默认 4096 entries / 16 MiB，TTL 固定 1 分钟（workerIOCacheTTL），跨 task 共享，Worker 关闭时 Purge。
• system cache：默认 64 entries / 512 MiB / 20 分钟，只服务注册时声明 SystemCache: true 的 backend（当前是 iceberg 的 resolve_data_files）。

​

自适应准入

• 每个采样窗口（默认 1s）内首次 Overloaded 立即乘法下降（BackoffRatio 默认 0.5，下限 MinLimit）；同窗口再次下降要求新代际样本达到 RepeatBackoffMinSamples 且过载比例达到 RepeatBackoffOverloadRatio，且不超过 MaxDecreasesPerWindow。
• Ignore 比例达到 IgnoreRatioThreshold（0.10）时窗口失去增长资格，达到 IgnoreBackoffRatioThreshold（0.30）时窗口结算时收缩一次。
• 增长只在窗口结束时发生，要求无过载、样本够、饱和（最大 in-flight 达到 limit）、连续健康窗口达到 IncreaseAfterHealthyWindows；步长 IncreaseStep（0 表示 initial 的 1%，至少 1）。
• 超过已学习的 softLimit 时增长变成探测：探测代际过载则精确退回 probeBaseLimit，连续失败按窗口数指数退避（上限 ProbeBackoffMaxWindows）。
• limit 降到 in-flight 以下时不取消存量请求，用 shrink debt 偿还。
• 不起 goroutine，窗口由 Acquire / Done 事件惰性推进。

​

状态与生命周期

• 创建：task 激活时 Orchestrator 调 ioScope.NewTaskScope(ioTaskCtx)（orchestrator_phases.go），失败记为 task_activation_failed。scope 持有派生自 task ctx 的 scopeCtx。
• 请求：每次 Read / Write 派生 ioCtx，context.AfterFunc(scopeCtx, ioCancel) 使 scope 关闭能取消在途 adapter 调用；每个 attempt 单独 Acquire / Release task 窗口，退避期间不占窗口和 permit。
• 关闭：task 终态时 rt.io.Close("task_terminal")。Close 用 CompareAndSwap 保证幂等，然后依次：取消 scopeCtx、sfTable.DrainAll(scope_closed) 唤醒所有 waiter、从 WorkerIOScope.activeScopes 移除。
• 迟到结果：leader 在 runIORetry 成功返回后再检查一次 closing，已关闭则返回 scope_closed，不写缓存也不广播；sflight.Table.Complete 与 DrainAll 在同一把锁下做幂等 close，避免 double-close panic。
• Worker 关闭：WorkerIOScope.Close 先关所有活跃 task scope，Purge 两个缓存，再逐个 adapter.Close()（adaptive wrapper 先关 limiter 唤醒 waiter，再关 raw adapter）。

• 等待期错误由 classifyWaitContextError 归一：scope 已 closing → scope_closed / io:temporarily_unavailable；DeadlineExceeded → timeout / context:timeout；Canceled → scope_closed / context:canceled；其他 → system_error。没有独立的”排队超时”错误。
• backend 错误由 classifyBackendError 归一：*IOError 原样透传 → adapter 的 ClassifyError → 否则 retryable_io。
• shouldRetryIO 只在 ioErr.Retryable()、未用完 RetryMaxRetries、ctx 带已知 retry scope 且 ctx 未结束时放行。退避用 backoff/v5：默认三次窗口 250–500ms、500ms–1s、1–2s，单次上限 3s。
• 写重试是 at-least-once：IO core 只重放当前 typed write request。BlockDB 全部 API（含 BundleWrite 的 InitWriteJob / CommitWriteJob / CancelWriteJob）保证幂等，因此 Result scope 内的 BlockDB 写保持可重试；此前的 WithoutRetryScope 退出机制已在 commit aae6b77a 删除。接入不满足幂等契约的写 API 时必须先调整 retry scope 或 backend 契约。
• singleflight leader 拥有整个 retry 循环，waiter 只等最终结果；缓存只回填最终成功值。

​

配置

​

扩展点

​

新增一个 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)
}

​

其他常见改动

• 给热点 (backend, operation) 加编译期 cache 命名空间：改 worker_scope.go 的 knownIOCacheNamespace 和对应枚举，并更新 TestKnownIOCacheNamespacesAreDistinctEnums。
• 调整某个 backend 的 AIMD 反馈：改该 SDK 包的 adaptive.go（ClassifyAdaptiveOutcome），不要动 internal/io/adaptive；判断传输层错误用 adaptive.IsTransportFailure / IsTransportFailureMessage。
• 新增可缓存但需要请求级校验的读：让请求类型实现 model.CachedReadValidator。

​

测试

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：请求路径主干——缓存命中 / 跨 task 共享（TestWorkerScope_CacheIsSharedAcrossTaskLifetimes）、singleflight（TestTaskScope_ReadSingleflight）、task 窗口获取 / 释放（TestTaskScope_TaskWindowWaitsUntilCapacityIsReleased、TestTaskScope_IOBackendAttemptReleasesTaskWindow）、scope 关闭与迟到结果（TestTaskScope_ActiveBackendContextLifecycle、TestTaskScope_ReadAfterClose、TestTaskScope_CloseIdempotent、TestWorkerScope_CloseCascades）、只读后端拒写、空 cacheKey 旁路（task_scope_bypass_test.go）。
• internal/io/core/retry_test.go：retry scope 门控、单个 leader 拥有重试、attempt 间释放资源、退避边界。
• internal/io/core/task_scope_drain_test.go：Complete 与 DrainAll 并发不 double-close。
• internal/io/adaptive/limiter_test.go、lanes_test.go、backend_test.go、backend_budget_test.go：AIMD 各条规则、探测、公平 lane、admission 排队不计入 budget、超时在 admission 后才开始。
• internal/io/assembly/assembly_test.go：stub 回退、重复 kind、失败时关闭已建 adapter、OuterWrap。
• internal/io/capabilitywire/wire_test.go：信封三方绑定、shape 错误 fail-closed。
• internal/io/sflight/sflight_test.go、internal/io/cache/cache_test.go：singleflight 与缓存的独立单测。
• 基准：internal/io/core/bench_test.go、adaptive/lanes_bench_test.go。
• 与 Executor 联动的进程级 e2e 在 cmd/worker/worker_process_*_test.go（如 worker_process_blockdb_bridge_test.go、worker_process_localtestservice_test.go）。更多见 测试组织与命令。

​

相关文档

• docs/specs/io-subsystem.md：本子系统的行为规范（Scoped Resource Context、adapter 职责、AIMD 反馈分类、Close 语义、缓存策略与 fork 风险、配置项）。
• docs/io-backend-module-design.md：backend module 装配层的设计动机与不变量。
• docs/capability-backend-guide.md：新增能力后端的逐文件操作手册与检查单。
• docs/specs/architecture.md §4.2.5 / §4.2.6：IO 访问子系统与共享状态缓存在总体架构中的定位。
• docs/specs/noderpc-semantic-batching.md：NodeRPC 语义合批（OuterWrap 的使用者）。
• docs/deploy.md：IO / AIMD 相关环境变量的部署说明。