feat(uring): bound admission and harden read-driver progress

.github/workflows/ci.yml

@@ -64,6 +64,8 @@ jobs:
             exit 1
           fi
           grep -q "DIRECT_OK direct_read_returns_exact_unaligned_ranges" test.out
+      - name: Ordered-prefetch contract smoke (real io_uring)
+        run: bash scripts/test-ordered-prefetch-cli.sh
       - name: Benchmark schema and correctness smoke
         run: bash scripts/test-benchmark-cli.sh
       - name: Instrumented benchmark schema and correctness smoke

CHANGELOG.md

@@ -25,6 +25,23 @@ aims to follow [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
 ### Added
 
+- Opt-in `SharedReadBudget` whole-driver quota reservations, retained through
+  deferred and kernel-owned reads, including bounded-drain leaks. Independent
+  drivers keep independent admission and shutdown; default constructors are unchanged.
+- Non-joining `request_shutdown`, advisory `is_finished`, and a default-off
+  `tokio-runtime` feature for eager consuming `shutdown_async` ownership handoff.
+  Synchronous `shutdown`/`Drop` and bounded-drain leak guarantees remain explicit.
+- Explicit same-binary A/A calibration mode for the evidence runner, including
+  individual middle-leg drift gates and no candidate attribution in control runs.
+- Example-only bounded ordered prefetch with deterministic backpressure,
+  cancellation and deferred-admission tests, plus a native CLI correctness gate.
+  This does not add a production streaming API or claim end-to-end speedups.
+- Opt-in `ReadLimits` for logical read size and driver-wide in-flight read-buffer
+  bytes, with aligned allocation accounting and terminal-CQE ownership.
+- Opt-in capacity-aware shard selection for positioned reads; round-robin remains
+  the default and stream reads retain their routing semantics.
+- `ReadRequest`, `MAX_BATCH_READS` and `read_at_batch` for bounded buffered groups
+  sharing eager notifications per final owning shard.
 - Opt-in `diagnostics` feature with per-shard sampled driver-stage histograms,
   aggregate snapshots, and measurement-interval deltas. Default builds compile
   out the timing fields and sampling work.
@@ -34,6 +51,24 @@ aims to follow [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
 ### Changed
 
+- Bound driver intake, allocations and completion work per turn, preserving
+  progress after notification draining and partial successful submissions.
+- Stop explicitly canceled positioned-read continuations after read CQEs and
+  avoid materializing orphaned results without releasing kernel-owned resources.
+
+### Fixed
+
+- Progress empty-SQ CQ overflow/taskrun work and propagate metadata failures
+  instead of treating an unconfirmed direct-read short prefix as EOF.
+- Close count-stage and byte-stage waiters together when shared byte admission
+  shuts down; retain charged resources for bounded-drain bailout leaks.
+- Retry interrupted eventfd operations and suppress repeated unchanged overflow
+  warnings without changing the cumulative snapshot counter.
+- Reap the whole benchmark process group after wrapper failure, including when
+  the group leader exits before a child process.
+
+### Benchmarking
+
 - Benchmark CSV schema v2 obtains headers from the executable and reports
   independent setup, workload, and teardown timings, configurable ring depth,
   workers, warmup, and instrumentation status. Positional CSV consumers must

Cargo.toml

@@ -36,6 +36,9 @@ fault-injection = []
 # Opt-in sampled timing. No timestamps, histogram storage, or tracing allocations
 # are compiled into the default driver.
 diagnostics = []
+# Optional Tokio blocking-pool shutdown adapter. Default builds only require
+# Tokio's synchronization primitives and can drive reads from other executors.
+tokio-runtime = ["tokio/rt"]
 
 [dependencies]
 # tracing for driver diagnostics: unifies with the RustFS tracing pipeline for

README.md

@@ -47,6 +47,67 @@ assert_eq!(snapshot.delivered + snapshot.orphan_reclaimed, snapshot.submitted);
 - `read_at_direct(file, offset, len, align)` — the same for an `O_DIRECT` fd; `offset`/`len` need not be aligned (the driver reads a block-aligned superset and returns exactly the requested range).
 - `read_current(file, len)` — `read(2)` semantics from the current position, for pipes and other non-seekable fds (a short read is a valid final result).
 - `probe_and_start_sharded(entries, shards)` — several independent rings per disk (each ring caps at one core's memory bandwidth for cache-hit reads); `probe_and_start(entries)` equals `..._sharded(entries, 1)`.
+- `probe_and_start_with_limits(entries, shards, ReadLimits { max_read_len, max_in_flight_bytes })` — optional logical read-size and driver-wide read-buffer limits. Both fields default to `None`, preserving existing constructor behavior.
+- `probe_and_start_with_shared_budget(entries, shards, limits, &pool)` — reserve the driver's whole configured byte quota from a cloneable `SharedReadBudget` before startup. See [shared reservation ownership](docs/shared-read-budget.md); this is not dynamic per-read sharing or an RSS limit.
+- `with_shard_policy(ShardPolicy::CapacityAware)` — opt-in capacity-aware routing for positioned reads. Constructors keep `ShardPolicy::RoundRobin` by default.
+- `request_shutdown()` — close admission and request cancellation/drain without joining; `is_finished()` reports advisory thread completion, not a clean drain.
+- `shutdown_async()` — with the default-off `tokio-runtime` feature, transfer consuming cleanup to Tokio's blocking pool at method call time. See [shutdown ownership and runtime boundaries](docs/shutdown.md).
+
+### Shard selection
+
+Round-robin binds each read to the next shard, even if that shard is busy or
+closed. Capacity-aware selection starts at the same cursor and tries each shard's
+count permit at most once, skipping closed count semaphores. It uses actual
+permit acquisition, not a free-capacity snapshot. If all healthy shards are busy,
+the handle waits on the first healthy candidate using Tokio's fair semaphore.
+The wait is local to that shard; it does not rebalance when another shard frees.
+
+A shared byte-budget shortage waits on the first shard whose count permit was
+available, returning that temporary count reservation before constructing the
+waiter. A closed shared byte budget rejects admission globally. Once a read is
+accepted or deferred, its read, wakeup, retry, and cancel retain the same owning
+shard. `read_current` always keeps the original round-robin behavior; concurrent
+stream reads still require caller serialization when ordering matters.
+
+Enable the policy explicitly on the constructed driver before sharing it:
+
+```rust,ignore
+let driver = UringDriver::probe_and_start_sharded(128, 4)?
+    .with_shard_policy(rustfs_uring::ShardPolicy::CapacityAware);
+```
+
+This policy has additional admission work under contention. Throughput, CPU cost,
+and tail-latency acceptance remain pending target-hardware measurements; it is
+not enabled by default.
+
+### Read allocation admission
+
+With `max_in_flight_bytes: Some(budget)`, all shards share one byte budget.
+Buffered reads reserve `len` bytes; direct reads reserve the block-aligned
+superset length plus `align - 1` bytes of allocation padding, including for
+zero-length direct reads. A request whose allocation exceeds the entire budget,
+or whose logical length exceeds `max_read_len`, returns `InvalidInput` before
+allocation. A byte budget of zero or above `tokio::sync::Semaphore::MAX_PERMITS`
+is rejected at construction. `max_read_len: Some(0)` allows only zero-length reads.
+
+Admission acquires the shard's count permit before its byte permits. Saturated
+handles wait asynchronously, holding no read buffer; a byte waiter may hold a
+count permit, and Tokio's fair byte semaphore can put small reads behind a large
+waiter. Dropping a waiting handle returns all partial reservations. After enqueue,
+both permits travel with the read until its terminal CQE, even if its caller is
+canceled. Short-read retries retain the same reservation. A leaked read retains
+its charge. Shutdown or any shard-thread exit closes both the shared byte
+semaphore and every shard's count semaphore when byte limits are enabled.
+This rejects further admission and wakes waiters at either acquisition stage,
+even when another shard has a hung read or takes a bounded-drain escape.
+Registration and terminal closure are synchronized at startup/shutdown; ordinary
+read admission does not take a registry lock.
+
+This limits reserved driver read-buffer allocation bytes, **not process RSS**.
+It excludes queued handle/FD metadata, allocator overhead, result copies and
+completed `Vec` results retained in channels or by callers. The caller must bound
+its task fan-out and result queue separately. Completion releases admission even
+when the returned result remains alive.
 
 ## API contract
 
@@ -87,6 +148,11 @@ stage overlap, cancellation, and instrumentation-overhead boundaries.
 
 Benchmark configuration, CSV schema, timing boundaries, and performance gates
 are documented in [the benchmarking guide](docs/benchmarking.md).
+See [implementation and acceptance status](docs/optimization-status.md) for
+completed correctness work and the still-open performance/integration gates.
+Application wiring has separate [RustFS integration prerequisites](docs/rustfs-integration.md).
+The [ordered-prefetch example](docs/ordered-prefetch.md) is a bounded consumer
+contract experiment, not a production streaming API or performance result.
 
 Linux only; on other hosts `cargo check` builds the empty stub.
 
@@ -95,14 +161,18 @@ Linux only; on other hosts `cargo check` builds the empty stub.
 cargo test -- --nocapture --test-threads=1
 
 # Two legs in Docker (also on macOS via Docker Desktop / OrbStack):
-#   leg 1 — io_uring blocked by an explicit seccomp profile → every test MUST
-#           degrade to a graceful skip;
+#   leg 1 — io_uring blocked by an explicit seccomp profile → ring-dependent
+#           tests gracefully skip; kernel-independent unit tests still run;
 #   leg 2 — seccomp=unconfined → real io_uring, and NO test may skip.
 ./run-docker.sh
 ```
 
 The harness fails on a non-degrading leg 1 or a vacuous-pass leg 2, so a skipped suite can never masquerade as coverage. The cancel-safety contract is pinned by the acceptance tests in `tests/cancel.rs`; the `fault-injection` feature (test-only) drives the panic-abort, bounded-drain-leak, and probe-failure escape hatches in `tests/fault_injection.rs`.
 
+For bounded buffered read groups, see [explicit batch reads](docs/batch-reads.md).
+`read_at_batch` shares eager notifications per owning shard while keeping each
+read's admission, result and cancellation independent.
+
 ## License
 
 Apache-2.0. See [LICENSE](LICENSE).

docs/batch-reads.md

@@ -0,0 +1,34 @@
+# Explicit buffered read batches
+
+`UringDriver::read_at_batch(Vec<ReadRequest>)` accepts up to `MAX_BATCH_READS`
+(64) buffered positioned reads and returns one `ReadHandle` per input, in input
+order. More than 64 requests fails before any submission; an empty batch sends
+no wakeup. Invalid individual offsets or lengths become normal asynchronous
+handle errors and do not reject other requests.
+
+Eagerly admitted reads share one eventfd notification per distinct final owning
+shard after handle construction. The capacity-aware policy, when selected, can
+route several inputs to the same owner, which still receives one notification.
+Requests awaiting count or byte admission signal their owner individually when
+polled and admitted. No tasks are spawned and existing single-read, direct-read
+and stream APIs retain their notification behavior.
+
+This is notification batching, not an atomic multi-read operation or snapshot.
+Kernel submission/completion order may differ from input order; each handle
+keeps independent cancellation and result ownership. Construction performs at
+most 64 submissions and at most 64-by-64 pointer identity comparisons for wake
+deduplication. It does not await capacity. If construction unwinds, already built
+handles are dropped, queue their normal cancels, and wake the owning shards.
+Driver-owned buffers, FDs and admission permits still survive until final CQE.
+
+Unit tests count real eventfd notifications using threadless driver queues, and
+cover rejection, mixed valid/invalid requests, final-owner routing, deferred byte
+admission, dropping handles and interrupted construction. Native integration
+tests verify byte-exact results and cancellation/drain conservation. Threadless
+tests model CQE resource release without submitting to a kernel. Notification
+reduction is verified deterministically; throughput and latency are unmeasured.
+
+```sh
+cargo test --all-features --lib batch_read_tests -- --nocapture
+cargo test --all-features --test cancel batch_ -- --nocapture
+```

docs/benchmarking.md

@@ -85,6 +85,15 @@ sample toward shard zero. Invalid requests can consume a sampling position
 without recording stages. Deterministic sampling is diagnostic, not an unbiased
 estimate for every possible periodic workload.
 
+With opt-in `ShardPolicy::CapacityAware`, positioned reads choose their owner
+before consuming that shard's sample sequence. Rejected geometry and unsuccessful
+candidate attempts do not consume sample positions; closed admission records no
+sample. For valid capacity-aware positioned reads, feature-on builds take one
+timestamp before selection even for unsampled reads, so sampled admission covers
+selection and permit acquisition (but excludes preceding geometry validation).
+Round-robin and stream sampling retain their original behavior. Measure this
+additional diagnostics cost with the selected routing policy.
+
 `UringDriver::diagnostics()` aggregates shards;
 `UringDriver::shard_diagnostics()` preserves shard identity. Each stage has a
 count, total nanoseconds, and 64 log2 nanosecond buckets. Bucket zero covers
@@ -117,7 +126,7 @@ even on std strategies (which do not use the instrumented driver).
 
 The default build compiles out timing fields, clock reads, histogram storage and
 sample allocations. Enabled builds add a per-shard atomic sampling counter per
-handle and an Arc/timestamps/histogram updates for sampled operations. Measure
+eligible handle and an Arc/timestamps/histogram updates for sampled operations. Measure
 that overhead on target hardware with separate feature-off/feature-on artifacts;
 do not assume it is free. Runtime schedule-latency and blocking-pool metrics must
 still be correlated in the application, which owns the Tokio runtime.
@@ -162,9 +171,40 @@ service can introduce load, supply `--require-inactive-unit UNIT`; the runner
 refuses to start or continue unless that unit is inactive. Any service stop or
 restore is an explicit operator action outside this tool. A reservation note and
 process checks are evidence aids, not a substitute for exclusive resources.
-Perform A/A calibration first by passing the baseline binary in both positions
-and `--candidate-interval 0`; then use the diagnostics candidate and the default
-interval of 64. Keep the workload and thresholds fixed between experiments.
+Perform explicit A/A calibration first with `--calibration --candidate-interval 0`
+and the same feature-off binary in both positions; then use the diagnostics
+candidate without `--calibration` and with the default interval of 64. Keep the
+workload and thresholds fixed between experiments.
+
+For example, use the preceding command with both executable arguments pointing
+to the baseline artifact and add:
+
+```sh
+--calibration --candidate-interval 0
+```
+
+Calibration requires matching executable SHA-256 values before running either
+artifact (identical copies at different paths are allowed), and requires interval
+zero for every measurement row. A mismatched hash or nonzero candidate interval
+is rejected, including in `--dry-run`. Each leg retains the normal binary-identity,
+geometry, resource and duration checks. Dry-run, provenance and summary output
+identify `mode` as `calibration` or `comparison`; dry-run does not execute workloads
+or establish that runtime gates will pass.
+
+Calibration preserves at least three A1/B1/B2/A2 rounds and the existing default
+thresholds: 3% for IOPS and 5% for p99. Every round checks A2 against A1, then checks
+**B1 and B2 separately** against the arithmetic mean of A1/A2, using those same
+thresholds in either direction. It never averages B1/B2 before gating: opposite
+noise must not cancel out. The first failed round stops the experiment. Only
+after all requested rounds pass does the summary say `valid-calibration`.
+Calibration records `baseline_drift` and per-leg `middle_drift`, never
+`candidate_change_pct`, whether it succeeds or fails. This validates the specified
+within-round noise gates, not an optimization benefit or stability under another
+workload. Cross-round trends, resources and application SLOs still require review.
+
+Omitting `--calibration` retains comparison mode and its original endpoint drift
+gate. Passing the same binary with `--candidate-interval 0` alone does not enable
+the additional middle-leg gates and must not be reported as validated calibration.
 
 Every leg must run at least five measured seconds by default. If it is too
 short, increase operations in a **new** experiment. The tool validates geometry,
@@ -181,5 +221,9 @@ and resource reports against the application SLO separately. Duration histograms
 with too few sampled operations are not reliable tail estimates.
 
 Runner gate tests: `python3 -m unittest discover -s scripts -p 'test_bench_abba.py'`.
+Calibration regressions use synthetic CSV/resource reports and mocked workload
+execution to cover mode validation, matching hashes, endpoint and middle drift,
+opposite-noise rejection, three-round success and early failure. They verify the
+runner's decisions, not native Linux performance or isolation.
 
 Tracking and implementation status: [rustfs/backlog#2647](https://github.com/rustfs/backlog/issues/2647).