diff --git a/CHANGELOG.md b/CHANGELOG.md
index e7c65ed..ea6bd9a 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -23,7 +23,72 @@ aims to follow [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
 ## [Unreleased]
 
-No unreleased changes.
+### Docs
+
+- Rewrote the README from the current public API — all three read entry points
+  (`read_at`, `read_at_direct`, `read_current`) and the degradation contract —
+  and removed the `docs/DESIGN.md` design notes. The per-invariant rationale now
+  lives in the module and function docs and in the README, so there is a single
+  source of truth.
+
+## [0.2.0] - 2026-07-11
+
+Hardening pass from the rustfs/backlog#1160 audit. All changes are on the read
+path and preserve the cancel-safety ownership model; both `run-docker.sh` legs
+(seccomp-blocked degradation and real io_uring) pass.
+
+### Added
+
+- **Test-only fault-injection seams** for the cancel-safety escape hatches
+  (driver-thread panic, stuck bounded drain, forced probe-drain failure), gated
+  behind the `fault-injection` feature and never present in a default build.
+  ([#11], backlog#1103)
+- `StatsSnapshot::submit_errors` — count of non-transient `ring.submit()`
+  failures, so a persistently failing `io_uring_enter` is observable.
+  (backlog#1162)
+
+### Fixed
+
+- **Bounded-drain bailout no longer hangs awaited handles.** It now fails every
+  stranded caller with a driver-gone error before leaking the pending table, and
+  leaks the ring-registered eventfd alongside the ring so the "cq_efd outlives
+  the ring" invariant holds on that exit too. (backlog#1161, #1167)
+- **Persistent `ring.submit()` errors are classified** instead of retried
+  forever in silence: EINTR/EBUSY stay transient; any other errno is counted and,
+  after a bounded run, shuts the shard down so callers fall back. (backlog#1162)
+- **Wakeup gaps closed:** short-read resubmits are flushed the same turn, and a
+  drop-cancel now signals the wake eventfd (after closing the receiver so the
+  reclaim is still counted as an orphan), removing up-to-50 ms stalls.
+  (backlog#1163)
+- **Driver-thread spawn failure degrades** to a `ProbeFailure` instead of
+  panicking out of disk init. (backlog#1164)
+- **The startup probe is time-bounded** (EXT_ARG timeout with a pre-5.11
+  fallback) and runs on the first shard only. (backlog#1165)
+- **EINTR/EAGAIN completions are retried** (bounded) rather than surfaced as the
+  read's final error, and the submit offset guard also rejects an aligned end
+  crossing `i64::MAX`. (backlog#1166)
+- **CQ overflow is reported as a NODROP backpressure warning**, not fatal loss,
+  and AsyncCancel SQEs are deduplicated per op. (backlog#1167)
+- **O_DIRECT tail short reads are disambiguated with `fstat`** instead of assuming
+  any non-block-multiple read is EOF, so a stacked filesystem cannot cause a
+  silent truncation. (backlog#1168)
+- **Idle churn cut**: the loop skips `io_uring_enter` on an empty SQ and uses a
+  longer heartbeat when idle; each ring caps its io-wq bounded workers.
+  (backlog#1169)
+
+### Changed
+
+- **Internal driver-loop refactor** folding the duplicated submit,
+  cancel-enqueue, and resubmit-SQE paths into single helpers (`submit_ring`,
+  `queue_cancel`, `Pending::resubmit_sqe`, `flush_backlog`). No behavior change;
+  the perf-sensitive submit → reap → conditional re-flush order is kept so a
+  cache-hit read is still reaped the same turn it completes inline.
+  (backlog#1160)
+
+### Tests
+
+- Deterministic sharded cancel-routing test proving a drop-cancel reaches the
+  ring that accepted the op. (backlog#1180)
 
 ## [0.1.0] - 2026-07-11
 
@@ -116,7 +181,7 @@ polished (`39018c0`).
 The cancel-safety invariants that survived that audit — the driver's pending table owning the
 buffer and fd from SQE submission until the CQE, drop-abandons-result-only, bounded shutdown
 drain, abort-before-free on a driver panic — are the ones every change above is measured
-against. See the [design notes](https://github.com/rustfs/uring/blob/v0.1.0/docs/DESIGN.md).
+against. See the [design notes](https://github.com/rustfs/uring/blob/0.1.0/docs/DESIGN.md).
 
 ## Decisions recorded, not implemented
 
@@ -145,8 +210,10 @@ built. They are listed so nobody re-opens them without new evidence.
 [#4]: https://github.com/rustfs/uring/pull/4
 [#5]: https://github.com/rustfs/uring/pull/5
 [#6]: https://github.com/rustfs/uring/pull/6
+[#11]: https://github.com/rustfs/uring/pull/11
 [rustfs/backlog#1051]: https://github.com/rustfs/backlog/issues/1051
 [rustfs/backlog#1144]: https://github.com/rustfs/backlog/issues/1144
 [rustfs/backlog#1159]: https://github.com/rustfs/backlog/issues/1159
-[Unreleased]: https://github.com/rustfs/uring/compare/v0.1.0...HEAD
-[0.1.0]: https://github.com/rustfs/uring/releases/tag/v0.1.0
+[Unreleased]: https://github.com/rustfs/uring/compare/0.2.0...HEAD
+[0.2.0]: https://github.com/rustfs/uring/compare/0.1.0...0.2.0
+[0.1.0]: https://github.com/rustfs/uring/releases/tag/0.1.0
diff --git a/Cargo.lock b/Cargo.lock
index 3dc98f5..90402c2 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -57,7 +57,7 @@ dependencies = [
 
 [[package]]
 name = "rustfs-uring"
-version = "0.1.0"
+version = "0.2.0"
 dependencies = [
  "io-uring",
  "libc",
diff --git a/Cargo.toml b/Cargo.toml
index 0fd4dc3..e93422a 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -14,7 +14,7 @@
 
 [package]
 name = "rustfs-uring"
-version = "0.1.0"
+version = "0.2.0"
 edition = "2024"
 rust-version = "1.96"
 license = "Apache-2.0"
@@ -25,7 +25,6 @@ readme = "README.md"
 homepage = "https://rustfs.com"
 keywords = ["io_uring", "async", "storage", "rustfs", "linux"]
 categories = ["asynchronous", "filesystem"]
-exclude = ["docs/**"]
 
 [features]
 # Test-only fault-injection seams (rustfs/backlog#1103). OFF by default and in
diff --git a/README.md b/README.md
index 00d8a04..1bdf85b 100644
--- a/README.md
+++ b/README.md
@@ -7,40 +7,49 @@
 
 Cancel-safe async `io_uring` read backend for [RustFS](https://github.com/rustfs/rustfs) storage.
 
-This crate is the io_uring integration that RustFS's read path is built on. It started as the Spike 0 cancel-safety
-prototype ([rustfs/backlog#894](https://github.com/rustfs/backlog/issues/894)) and was hardened per
-the [#1048/#1051 audit](https://github.com/rustfs/backlog/issues/1051). It lives in its own repository so it can be
+This crate is the io_uring integration that RustFS's read path is built on. It lives in its own repository so it can be
 verified in isolation — with a real io_uring CI leg that the main `rustfs/rustfs` workspace cannot run — before being
-wired into the storage layer.
+wired into the storage layer behind a runtime probe.
 
-> **Status:** read path only, Linux only. The read path is wired into `rustfs/rustfs` behind a runtime probe and is
-> **off by default** (`RUSTFS_IO_URING_READ_ENABLE`). See [`CHANGELOG.md`](CHANGELOG.md) for what has landed and the
-> [design notes](https://github.com/rustfs/uring/blob/v0.1.0/docs/DESIGN.md) for the invariants.
+> **Status:** read path only, Linux only. On any other target the crate compiles to an empty stub. The read path is
+> wired into `rustfs/rustfs` behind a runtime probe and is **off by default** (`RUSTFS_IO_URING_READ_ENABLE`). What has
+> landed is in [`CHANGELOG.md`](CHANGELOG.md); the per-invariant rationale lives in the module and function docs on
+> [docs.rs](https://docs.rs/rustfs-uring/) and inline in `src/`.
 
 ```toml
 [target.'cfg(target_os = "linux")'.dependencies]
-rustfs-uring = "0.1.0"
+rustfs-uring = "0.2.0"
 ```
 
 ## The ownership model it enforces
 
 When a caller drops the future of an in-flight read (EC quorum reached, timeout, disconnect), the kernel may still write
 into the read buffer at any point until the CQE. Freeing the buffer at future-drop is a use-after-free. This crate
-proves and enforces the invariants any production io_uring integration must follow:
+enforces the invariants any production io_uring integration must follow:
 
 - **The buffer and the file handle are owned by the driver's pending (orphan) table** from SQE submission until the
   CQE — never by the caller's future.
-- **Dropping the future abandons only the result**; reclamation always happens at the CQE (optionally accelerated by
-  `IORING_OP_ASYNC_CANCEL`).
-- **Shutdown drains in-flight ops to zero** (with a bounded escape hatch for hung disks) before the ring is unmapped.
-- A driver-thread panic **aborts before freeing in-flight buffers** (leak over UAF); backpressure permits are released
-  at the CQE, not at future drop; short reads on positioned reads are resubmitted to satisfy the whole-range contract;
+- **Dropping the future abandons only the result**; reclamation always happens at the CQE, optionally accelerated by an
+  `IORING_OP_ASYNC_CANCEL` sent on drop.
+- **Shutdown drains in-flight ops to zero** before the ring is unmapped, with a bounded escape hatch for a hung disk:
+  on timeout it fails the stranded callers with an error and *leaks* the ring and its buffers rather than free memory
+  the kernel may still touch (leak over UAF).
+- A driver-thread panic **aborts the process before freeing in-flight buffers**; backpressure permits are released at
+  the CQE, not at future drop; a short read on a positioned read is resubmitted to satisfy the whole-range contract;
   the probe file is opened via `O_TMPFILE`.
 
-Each invariant holds **per shard** (see below), because a shard is an independent instance of the same driver.
+It also has a **degradation contract** — the reason it can ship default-on-probe safely:
 
-The full invariant list, the corrected fd-reuse mechanism, and the design constraints are in
-the [design notes](https://github.com/rustfs/uring/blob/v0.1.0/docs/DESIGN.md).
+- The startup probe runs a real `IORING_OP_READ` under a wall-clock timeout. A restricted environment
+  (seccomp / gVisor / old kernel) fails with a `ProbeFailure` whose `is_expected_restriction()` tells the caller to
+  degrade to the std backend quietly; an unexpected failure is surfaced, not hidden.
+- A persistently failing `io_uring_enter` is classified (not retried forever in silence): the shard shuts down so
+  callers fall back, and the failure count is visible in `StatsSnapshot`.
+- Transient CQE errnos (`EINTR`/`EAGAIN`) are retried; the whole-range and EOF contracts are honoured even on stacked
+  filesystems by disambiguating a short read with `fstat` rather than assuming EOF.
+
+Each invariant holds **per shard** (see below), because a shard is an independent instance of the same driver. Every
+one is pinned by an acceptance test (see [Testing](#testing)).
 
 ## Usage
 
@@ -50,9 +59,8 @@ use std::sync::Arc;
 use rustfs_uring::UringDriver;
 
 # async fn demo() -> std::io::Result<()> {
-    // Probe a real IORING_OP_READ before accepting work. A restricted environment
-    // (seccomp/gVisor/old kernel) returns a ProbeFailure whose
-    // `is_expected_restriction()` tells you to degrade to the std backend quietly.
+    // Probe a real IORING_OP_READ before accepting work. On a restricted host the
+    // ProbeFailure's `is_expected_restriction()` tells you to degrade quietly.
     let driver = UringDriver::probe_and_start(64).expect("io_uring available");
 
     let file = Arc::new(File::open("/data/object")?);
@@ -70,6 +78,17 @@ use rustfs_uring::UringDriver;
 }
 ```
 
+Three read entry points, all returning an awaitable `ReadHandle`:
+
+- `read_at(file, offset, len)` — positioned (pread) read, whole-range: short reads are resubmitted until the range is
+  satisfied or a real EOF.
+- `read_at_direct(file, offset, len, align)` — the same, for an fd opened with `O_DIRECT` (see below).
+- `read_current(file, len)` — reads from the fd's current position with `read(2)` semantics: a short read is a valid
+  final result and is *not* resubmitted. Use it for pipes and other non-seekable fds.
+
+`ReadHandle::without_cancel_on_drop()` opts a handle out of the drop-time `ASYNC_CANCEL` when you know the op will
+complete on its own; and `driver.stats()` returns a `StatsSnapshot` at any time.
+
 ### Sharded rings
 
 A buffered read that hits the page cache completes *inline* inside `io_uring_enter`, so the thread driving a ring
@@ -135,15 +154,19 @@ cargo test -- --nocapture --test-threads=1
 
 # Two legs in Docker (also runs on macOS via Docker Desktop / OrbStack):
 #   leg 1 — io_uring blocked by an explicit seccomp profile → the suite MUST
-#           degrade to a graceful skip (reproduces the #4313 restricted env);
+#           degrade to a graceful skip (reproduces a restricted environment);
 #   leg 2 — seccomp=unconfined → real io_uring, and NO test may skip.
 ./run-docker.sh
 ```
 
 The harness fails on either a non-degrading leg 1 or a vacuous-pass leg 2, so a skipped suite can never masquerade as
-real coverage. The 15 acceptance tests are the cancel-safety contract: buffer conservation under a mixed
-drop/keep stress across shards, an orphaned op reclaimed only at its CQE, bounded shutdown drain, `O_DIRECT` returning
-exact unaligned ranges, and backpressure deferring rather than blocking a runtime worker.
+real coverage. The cancel-safety contract is pinned by 16 acceptance tests in `tests/cancel.rs` — buffer conservation
+under a mixed drop/keep stress across shards, an orphaned op reclaimed only at its CQE, sharded cancel routed to the
+ring that owns the op, bounded shutdown drain, `O_DIRECT` returning exact unaligned ranges, and backpressure deferring
+rather than blocking a runtime worker. Five deterministic fault-injection tests in `tests/fault_injection.rs` (behind
+the test-only `fault-injection` feature, never compiled into a release build) drive the escape hatches: the
+panic-abort barrier, the bounded-drain leak-and-error path, a forced probe-drain failure, and a driver-thread spawn
+failure that must degrade rather than panic.
 
 ## Benchmarks
 
@@ -168,10 +191,9 @@ alone cannot tell a correct strategy from one reading the wrong offsets.
 - **`SQPOLL`.** Eliminates `io_uring_enter` under sustained load, at the cost of a kernel polling thread per ring —
   which multiplies by shards and by disks. Only for high-end deployments.
 
-Closed by measurement, not built — see [`CHANGELOG.md`](CHANGELOG.md#decisions-recorded-not-implemented): streaming
-reads through io_uring (NO-GO), `AsyncFd` reaping without a driver thread (would break the public API), a process-wide
-singleton ring (conflicts with per-disk isolation), and registered buffers (conflicts with the `Vec<u8>` ownership
-model, and the bottleneck is elsewhere).
+Closed by measurement, not built (see the CHANGELOG): streaming reads through io_uring (NO-GO), `AsyncFd` reaping
+without a driver thread (would break the public API), a process-wide singleton ring (conflicts with per-disk
+isolation), and registered buffers (conflicts with the `Vec<u8>` ownership model, and the bottleneck is elsewhere).
 
 ## License
 
diff --git a/docs/DESIGN.md b/docs/DESIGN.md
deleted file mode 100644
index dd26588..0000000
--- a/docs/DESIGN.md
+++ /dev/null
@@ -1,114 +0,0 @@
-# rustfs-uring 设计:io_uring 取消安全读后端(backlog#894 / #1048 / #1051)
-
-> 本文档源自 Spike 0 取消安全原型(原 `SPIKE.md`),经 rustfs/backlog#1051 审计整改后作为 `rustfs-uring` 库的设计与不变量说明保留。逐 issue 修复的历史见本仓库 git log。
-
-## 这是什么
-
-rustfs/backlog#897 路线图中 P2(io_uring 读后端)被 P1.5 基准判 NO-GO 而 defer。本 spike 是 #894 明确要求先行的**取消安全原型**——P2 中风险最高、最容易随时间流失的知识,按"只实现原型、不进主干、不启用"的方案 B 存档。
-
-> **现状更新(2026-07):** P2 主体已在此库实现并接入 `rustfs/rustfs`,但**默认灰度关**(`RUSTFS_IO_URING_READ_ENABLE`)。端到端 A/B(rustfs/backlog#1159)显示 io_uring 对 S3 GET 大致中性(−7%~+4%),瓶颈在用户态拷贝而非磁盘读,因此 #1048 转为 **[Watch] 看护 issue**——真实直连 NVMe 证据满足前不默认启用。下方"对 P2 主体实现的遗留项"一节记录了每项的落地/决策现状。本文其余部分(所有权模型、不变量、测试矩阵)是这些实现共同遵守的契约,持续有效。
-
-**本 crate 是独立 workspace**(Cargo.toml 内含空 `[workspace]` 表),io-uring 依赖不进入 rustfs 主 Cargo.lock、不参与主工程构建与 CI。这与守卫脚本 `scripts/check_no_tokio_io_uring.sh` 的约束一致:禁的是 tokio 的 io-uring runtime feature,应用层显式 io-uring 集成必须走运行时探测的独立后端(即本原型验证的模型)。
-
-## 要证明的问题
-
-EC 读会 drop 在途的分片读 future;若该 future 已向内核提交了 read SQE,内核在 CQE 之前始终可能向目标 buffer 写入。**future 的 drop 不能回收 buffer,否则是 use-after-free。**
-
-> **该场景在生产 GET 上被真实行使(2026-07 核对 main):** 主触发点是**读者建立阶段**——`crates/ecstore/src/set_disk/core/io_primitives.rs` 的 `create_bitrot_readers_until_quorum_all_shards`(`FuturesUnordered` 于 :1362 建立、setup quorum break 于 :1403、多余任务于 :1428 `drop(reader_tasks)`)。数据分片在此阶段经 `read_file_mmap_copy → UringBackend::pread_bytes → driver.read_at(...).await` **急切**读入,因此一个仍停在该 `.await` 的任务被 drop 时,其 `ReadHandle` 在 `Submitted` 态被 drop、触发 `ASYNC_CANCEL`(`ReadHandle::drop`,`src/driver.rs`)。这在**每个响应盘多于 setup quorum 的 GET(常态)**都发生,与后续是否重建无关。decode 阶段(`ParallelReader`,`crates/ecstore/src/erasure/coding/decode.rs`)的 `FuturesUnordered` 也在 quorum 处 drop 落败者,但对 io_uring 只在**非 lockstep 路径的延迟 parity reader 边打开边被 drop**时命中(数据分片此时已是内存 `Bytes`,lockstep 路径则全量 drain、从不中途 drop)。
-
-## 验证的所有权模型
-
-```
-caller                    driver thread                     kernel
-------                    -------------                     ------
-read_at() ──Msg::Read──▶  分配 buf,登记 pending 表
-                          (buf + Arc<File> + oneshot tx)
-                          push SQE(user_data=id) ─submit──▶ 开始随时可能写 buf
-await ◀───oneshot────────                                    │
-                                                             │
-future drop(任意时刻)                                        │
-  └─(可选)Msg::Cancel ──▶ push ASYNC_CANCEL ────────────────▶│ 加速 CQE
-  └─绝不触碰 buf                                             │
-                          CQE 到达 ◀─────────────────────────┘
-                          pending.remove(id)  ← 全程唯一的 buf 回收点
-                          send 结果:成功=delivered
-                                    失败(接收方已 drop)=orphan_reclaimed
-```
-
-关键不变量:
-
-1. **buffer 与 fd 归 pending 表所有,不归 future。** SQE 里的裸指针指向表项 `Vec` 的堆块;`Vec` 结构体可随 HashMap 移动(堆块地址不变),但在 CQE 前绝不 resize/drop。
-2. **fd 也必须由表项持有**(`Arc<File>`)。真正的危险窗口是 **SQE 构造(`as_raw_fd`)→ `io_uring_enter` 内核消费**:此窗口内 SQE 携带裸 fd 号在 backlog 中滞留、内核尚未 `fget`;若 fd 被 drop 关闭并被新 `open` 复用,提交时内核解析到错误文件——对 READ op 意味着从**错误文件读出数据**(跨对象数据错读/泄露),而非"内核的写落到别人文件"。表项持有 `Arc<File>` 到 CQE 是该窗口的安全超集。(机理更正见 rustfs/backlog#1063:原文把危险窗口误标为"提交→CQE"、后果误标为"写别人文件"——已提交的 op 因内核已持 struct file 引用而对 fd close/复用免疫;若未来用 SQPOLL,消费点还会与 enter 脱钩。)
-3. **future drop 只放弃结果领取**,默认附带提交 `IORING_OP_ASYNC_CANCEL`(best-effort 加速),也可以不提交(裸 drop)——两种情况下回收都只发生在 CQE。
-4. **shutdown 顺序**:停收新 SQE → 对所有在途 op 提交 cancel → drain 到 `in_flight == 0` → 线程退出 → ring drop(unmap)。ring 决不能在内核仍持有 buffer 引用时 unmap。
-5. **探测必须提交真实 read op**:`io_uring_setup` 成功不代表 op 可用(gVisor/seccomp 可以建 ring 但 op ENOSYS/EINVAL);探测失败按 EACCES/EPERM/ENOSYS/EINVAL/EOPNOTSUPP 分类,命中即优雅降级(测试中表现为 skip),其余 errno 视为真 bug 直接断言失败。
-
-以下不变量是本次审计整改(rustfs/backlog#1051)新增/固化,P2 必须一并沿用:
-
-6. **驱动线程 unwind 安全**(rustfs/backlog#1054):驱动线程绝不允许在栈展开中释放 pending 表或 unmap ring——否则内核仍可能向在途 buffer 写入即 UAF。实现为 `DriverState::Drop` 检测 `thread::panicking()` 时在字段析构前 `process::abort()`(leak over UAF)。所有 caller 可控的 panic 面(如超大 `len`)在 `submit` 入口拒止。`catch_unwind` 不够——析构在展开时、catch 边界之前就已发生。
-7. **背压 permit 在 CQE 点释放**(rustfs/backlog#1060;异步化见 #1102):in-flight 上界 ≤ CQ 容量(取 SQ 深度 `entries` < `2*entries`,使 CQ overflow 结构性不可达),permit 随 pending 表项移除(CQE)释放,**绝不随 future drop 释放**——否则 quorum 大量 drop future 会让 permit 计数与驻留内存脱钩,重开内存 DoS 面。
-   - **已落地**:`tokio::sync::Semaphore`;`OwnedSemaphorePermit` 随 `Msg::Read` 存进 `Pending` 表项,表项在最终 CQE 被移除时 permit 自动 drop ——"CQE 点释放"由**类型系统强制**,不再依赖手写 `release()`(短读 resubmit 保留表项,故也保留 permit)。
-   - **获取从不阻塞线程**:未饱和走 `try_acquire_owned()` 快路径(无分配、无 await、提交仍是即时的);饱和时把 acquire future 交给返回的 `ReadHandle`,首次 poll 时 await 到 permit 再提交。因此**一个在首次 poll 前就被 drop 的 handle 从未提交、从未分配 buffer**(比阻塞实现更省内存),`delivered + orphan_reclaimed == submitted` 的守恒式仍恒成立。
-8. **复用缓冲内容卫生**(rustfs/backlog#1062,P3 前置):当前 spike 每 op 新分配零页 + `truncate(res)`,**无泄露**。P3 改用驱动自有对齐 slab(registered buffer)后,缓冲跨请求复用即脏内存——任何路径忘记按 `cqe.res` 截断/掩蔽(O_DIRECT 整块读再由上层切片、或错误路径把整块缓冲交还)就把上一租户请求的对象字节泄给当前请求(CWE-226)。不变量:**复用缓冲对调用方可见的字节严格 ⊆ `[0, cqe.res)`**,越界部分零化或由类型系统(返回带长度上限的 view 而非整块 slice)保证不可达。docs/DESIGN.md 与 #1048 原约束只讲 slab 生命周期(防 UAF),不讲内容卫生;需配套"脏缓冲 + 短读"回归测试。
-
-补充契约:
-
-- **errno 三分类**(rustfs/backlog#1059):`is_expected_restriction` **仅用于 probe 期**;运行期 errno 必须分——probe 受限 → 该盘永久降级;运行期参数错误(offset>i64::MAX、O_DIRECT 未对齐等 EINVAL)→ 返回错误、绝不闩锁;瞬态(EINTR/EAGAIN)→ 重试。`submit` 已在入口拒止 offset>i64::MAX 与 len>MAX_RW_COUNT。
-- **shutdown 有界 drain**(rustfs/backlog#1055):drain-to-zero 可能因坏盘上 cancel 不可中断(EALREADY)而不终止;超时(`DRAIN_TIMEOUT`)后泄漏 ring+buffer 退出(leak over UAF),绝不提前 unmap。cancel CQE 三态(succeeded/not_found/already)已纳入统计,EALREADY 上升即坏盘信号。
-- **短读 resubmit**(rustfs/backlog#1058):io_uring 对常规文件可合法短读;驱动 resubmit 剩余到 `buf[nread..]`,回收点移到逻辑读的最后一个 CQE。P2 须明确短读归属(后端循环 vs 调用方 `read_exact`)。
-
-## 测试矩阵
-
-| 测试 | 验证点 |
-|---|---|
-| `read_matches_std` | 完成路径正确性:64 次变长/变偏移读与文件内容逐字节一致 |
-| `dropped_future_buffer_lives_until_cqe` | **核心断言**:阻塞的 pipe 读上裸 drop future(不提交 cancel),300ms 后 op 仍 in-flight、buffer 未回收;向 pipe 写入触发 CQE 后才回收(orphan_reclaimed=1) |
-| `async_cancel_accelerates_reclaim` | 默认 drop 路径:ASYNC_CANCEL 使孤儿 op 在无数据到达的情况下经 ECANCELED CQE 及时回收 |
-| `cancel_stress_accounts_for_every_buffer` | 压力:256 并发读、一半立即 drop;`delivered + orphan_reclaimed == submitted`,幸存读逐字节正确 |
-| `shutdown_drains_in_flight_ops` | 关停:两个阻塞在途 op 被 cancel + drain 到 0 后线程才退出,持有的 future 解析为 ECANCELED |
-
-> 上表是 spike 原始 5 项核心断言。接入期实现随之新增覆盖,`tests/cancel.rs` 现共 **15 项**,补充:`saturated_submit_defers_instead_of_blocking`(异步背压不阻塞 runtime worker)、`direct_read_returns_exact_unaligned_ranges`(O_DIRECT 非对齐区间精确交付、填充不外泄)、`sharded_driver_conserves_buffers_across_shards` 与 `sharded_driver_with_one_shard_matches_single_ring`(分片下守恒 + 单片等价)、以及 boundary/pipe/EOF 等边界。所有项在 `run-docker.sh` 两腿下运行(leg 1 全优雅降级、leg 2 真实 io_uring)。
-
-需要 Docker(Linux 内核)。macOS 宿主上 `cargo check` 只验证非 Linux 桩编译。
-
-```bash
-./run-docker.sh
-```
-
-- **leg 1(默认 seccomp)**:多数 Docker 版本默认禁 io_uring(即 #4313 事故环境),探测失败 → 全部测试走优雅降级 skip,套件仍绿。若宿主 Docker 放行 io_uring,则此腿等同 leg 2。
-- **leg 2(seccomp=unconfined)**:真实 io_uring,完整跑取消安全套件。
-
-## 运行结果
-
-两腿一次通过,详见"实测记录"。
-
-## 对 P2 主体实现的遗留项 — 落地/决策现状
-
-> 本节原为"本 spike 不覆盖"的清单;下列各项经 rustfs/backlog#1102/#1144/#1145/#1159 处理后现状如下。三种收尾:**✅ 已实现**、**⛔ 经度量/设计决策关闭(不做)**、**⬜ 仍未做(正确 defer)**。
-
-- **✅ eventfd 唤醒收割(rustfs/backlog#1102)。** 200µs 忙轮询已由 eventfd 替换:一个 eventfd 注册到 ring(内核每 CQE 信号)、一个由 `submit`/shutdown 信号,驱动线程 `poll` 两者阻塞等待,`submit()` 每轮仍冲刷 NODROP overflow list。
-  - **⛔ tokio `AsyncFd` 去驱动线程 — 不做。** `Drop` 不能 `await`,shutdown 的有界 drain 排空会被逼成公开 API 破坏。现"专用驱动线程 + eventfd 阻塞收割"已消除忙轮询,是无破坏的等价收益。
-- **✅→⛔ ring 生命周期 → 改为 per-disk 分片。** "进程级单例 ring"与坏盘隔离诉求(一块坏盘不得拖垮其它盘)相冲,已**重定义**为 per-disk ring 集,并用 `probe_and_start_sharded(entries, shards)` 在盘内横向扩展(每分片独立线程/pending 表/背压/eventfd)。Drop/shutdown 先让所有分片停再逐片 join,`DRAIN_TIMEOUT` 上界防坏盘无界阻塞。
-- **✅ O_DIRECT 对齐读:`read_at_direct(file, offset, len, align)`(rustfs/backlog#1102)。** 驱动读**块对齐超范围**到**块对齐 buffer**(超额分配 `align-1` 字节,在分配内部取第一个对齐字节作为读区起点),完成后只把逻辑区间 `[offset, offset+len)` 切出——**对齐填充、区间前缀、块对齐尾部一律不外泄**(否则 `BitrotReader` 会把补齐字节当损坏)。短读 resubmit 保持块对齐;内核返回非块倍数即文件尾。缓冲读是 `align == 1` 的退化情形。
-  - **✅ ecstore 已原生接线**(rustfs/rustfs#4649):`pread_uring_direct` 以 O_DIRECT 打开 fd 并调 `read_at_direct`,分层兜底 + per-disk 闩锁。**已取代 #4645 的临时分流**(那版把 O_DIRECT 合格读交回 StdBackend)。
-- **✅→⛔ 三读形态接入 `LocalIoBackend` — 部分做、部分不做。** **定位读 `pread_bytes` 已接** io_uring(缓冲 + 原生 O_DIRECT)。**流式读 `open_read_stream`/`open_full_read` 判 NO-GO**(rustfs/backlog#1144):io_uring 对单条顺序流无杠杆(内核 readahead 已胜),冷读设备瓶颈、暖读仅 11–41% 于 buffered——保持委托 StdBackend。
-- **✅ per-disk 探测缓存 + 运行期 errno 降级闩锁(rustfs/backlog#1101)。** `URING_UNSUPPORTED_DISKS` 负探测缓存 + `is_io_uring_unsupported` 运行期 errno 三分类(仅限制类 `EPERM/EACCES/ENOSYS/EOPNOTSUPP` 触发整盘闩死,数据/参数错误不闩),与 probe 期分类分离。
-- **⬜ registered buffers(P3)/ 写路径(P4)— 仍未做(正确 defer)。** `register_buffers` 与本文档"buffer 归 pending 表、`Vec<u8>` 所有权"的取消安全模型冲突,需重设计;且 #1159 端到端 profiling 显示读路径已非瓶颈,优先级低。写路径(PUT)完全未涉及,profiling 提示其收益可能大于读路径。
-
-**已在本 spike 内整改**(rustfs/backlog#1051):SQ 深度背压(不变量 7)、驱动线程 unwind 安全(不变量 6)、shutdown 有界 drain、CQ overflow/NODROP/EBUSY 处理、probe UAF、probe 文件安全创建、errno 三分类、len/offset 校验、短读 resubmit。
-
-**接入后新增的行为契约**(实现期发现,非 spike 原文):`UringBackend` 读完须遵守 StdBackend 的页缓存回收策略(≥4 MiB 读后 `fadvise(DONTNEED)`),否则开启 io_uring 即静默污染页缓存(rustfs/rustfs#4662 修复的回归)。
-
-## 实测记录
-
-2026-07-07,宿主 macOS + OrbStack Docker(Linux arm64,内核 7.0.11-orbstack),镜像 `rust:1-bookworm`,`cargo test --release`:
-
-- **leg 1(默认 seccomp)**:`io_uring_setup` 失败 `EPERM (Operation not permitted)`——与 #4313 事故环境同类。`ProbeFailure::is_expected_restriction()` 命中,5 个测试全部优雅降级 skip,套件绿。证明探测 + errno 分类降级契约按设计工作。
-- **leg 2(seccomp=unconfined)**:5 个测试全部通过(0.45s):
-  - `read_matches_std` ok — 64 次读逐字节正确;
-  - `dropped_future_buffer_lives_until_cqe` ok — 裸 drop 后 op 保持 in-flight 300ms、buffer 未回收,写 pipe 触发 CQE 后 `orphan_reclaimed=1`;
-  - `async_cancel_accelerates_reclaim` ok — ECANCELED CQE 路径回收;
-  - `cancel_stress_accounts_for_every_buffer` ok — 256 op、128 drop,`delivered(128) + orphan_reclaimed(128) == submitted(256)`;
-  - `shutdown_drains_in_flight_ops` ok — drain 到 0 后退出,持有 future 解析为 ECANCELED。
-
-**结论:GO(模型可行)。** buffer/fd 归驱动 pending 表、CQE 唯一回收点、ASYNC_CANCEL 加速、shutdown drain 的组合在真实内核上成立,且降级契约在受限环境下按设计生效。**P2 主体已据此模型实现**(见上"落地/决策现状"),接入 `rustfs/rustfs` 后默认灰度关,15 项取消安全验收测试是这些实现共同遵守的回归门禁。
diff --git a/src/driver.rs b/src/driver.rs
index e4df6eb..83e4644 100644
--- a/src/driver.rs
+++ b/src/driver.rs
@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use std::collections::{HashMap, VecDeque};
+use std::collections::{HashMap, HashSet, VecDeque};
 use std::fs::File;
 use std::io;
 use std::io::Write as _;
@@ -82,6 +82,33 @@ fn aligned_geometry(offset: u64, len: usize, align: usize) -> Option<(u64, usize
 /// deadline is still checked and any queued cancel is picked up promptly.
 const LOOP_HEARTBEAT: Duration = Duration::from_millis(50);
 
+/// Heartbeat used when the shard is fully idle — no in-flight ops and not
+/// shutting down (rustfs/backlog#1169). New work still wakes the loop instantly
+/// via `wake_efd` and completions via the registered `cq_efd`; this only bounds
+/// the fallback wait, so a much longer value cuts idle timer/​syscall churn
+/// across many per-disk shards without affecting latency.
+const IDLE_HEARTBEAT: Duration = Duration::from_secs(1);
+
+/// Per-ring cap on io-wq BOUNDED workers (rustfs/backlog#1169). Cold buffered
+/// and O_DIRECT reads punted to io-wq each spawn a bounded worker, and the
+/// kernel default is min(sq_entries, 4*nCPU) PER ring — one ring per shard per
+/// disk can otherwise materialize thousands of PF_IO_WORKER threads under a
+/// cold-read burst. Best-effort (needs kernel >= 5.15); older kernels keep the
+/// default.
+const IOWQ_MAX_BOUNDED_WORKERS: u32 = 16;
+
+/// Consecutive non-transient `ring.submit()` failures the driver tolerates
+/// before it stops retrying silently and shuts the shard down, so callers get a
+/// driver-gone error and fall back to the std backend instead of stalling
+/// forever on ops the kernel will never accept (rustfs/backlog#1162). With the
+/// 50 ms heartbeat this bounds the silent-retry window to a few seconds.
+const MAX_CONSECUTIVE_SUBMIT_ERRORS: u32 = 128;
+
+/// How many times a single logical read retries a transient CQE errno
+/// (EINTR/EAGAIN) without making progress before it surfaces the error, so a
+/// pathological storm cannot spin the driver thread (rustfs/backlog#1166).
+const MAX_TRANSIENT_RETRIES: u32 = 16;
+
 /// Owned `eventfd(2)` used to wake the driver loop (backlog#1102): one is
 /// registered with the ring so the kernel signals it on every CQE, the other is
 /// signaled by `submit`/shutdown so a new message wakes the loop immediately —
@@ -237,6 +264,7 @@ struct DriverStats {
     cancel_not_found: AtomicU64,
     cancel_already: AtomicU64,
     cq_overflow: AtomicU64,
+    submit_errors: AtomicU64,
 }
 
 /// Point-in-time copy of the driver counters.
@@ -262,10 +290,19 @@ pub struct StatsSnapshot {
     /// signal that makes drain-to-zero non-terminating (C4,
     /// rustfs/backlog#1055).
     pub cancel_already: u64,
-    /// Kernel CQ-ring overflow counter. MUST stay 0: a non-zero value means
-    /// CQEs were lost, so their pending entries are never reclaimed and drain
-    /// never completes. Treated as fatal (C5, rustfs/backlog#1056).
+    /// Kernel CQ-ring overflow counter. With NODROP (asserted at probe) overflow
+    /// CQEs are buffered in the kernel overflow list and flushed on the next
+    /// enter, NOT lost — so a non-zero value is a backpressure warning, not fatal
+    /// loss (C5, rustfs/backlog#1056, #1167). In-flight is capped at `entries`
+    /// and cancels are deduped, keeping completions <= 2*entries, so it should
+    /// stay 0 in practice.
     pub cq_overflow: u64,
+    /// `ring.submit()` calls that returned a non-transient error. A rising count
+    /// means `io_uring_enter` is persistently failing (e.g. a seccomp/LSM policy
+    /// applied after startup); the driver shuts the shard down after a bounded
+    /// run of consecutive failures so callers fall back instead of stalling
+    /// (rustfs/backlog#1162).
+    pub submit_errors: u64,
 }
 
 enum Msg {
@@ -347,6 +384,33 @@ struct Pending {
     region_len: usize,
     /// `1` for buffered, the block size for `O_DIRECT`.
     align: usize,
+    /// Consecutive transient-errno (EINTR/EAGAIN) retries since the last byte of
+    /// progress, bounded by `MAX_TRANSIENT_RETRIES` so a storm cannot spin the
+    /// driver thread (rustfs/backlog#1166). Reset whenever a read makes progress.
+    transient_retries: u32,
+}
+
+impl Pending {
+    /// Build the read SQE for the not-yet-read remainder
+    /// `[pad + nread, pad + region_len)` at file offset `offset + nread`. This is
+    /// the single place a read SQE is constructed: the initial submit calls it
+    /// with `nread == 0` (the whole region), and a short-read or transient-errno
+    /// resubmit calls it after `nread` has advanced (rustfs/backlog#1058/#1166).
+    /// For an `O_DIRECT` read `pad + nread`, `offset + nread`, and the remaining
+    /// length are all block-aligned.
+    fn read_sqe(&self, ud: u64) -> io_uring::squeue::Entry {
+        let remaining = self.region_len - self.nread;
+        // SAFETY: `pad + nread < pad + region_len <= buf.len()`, and the buffer
+        // lives in the pending table until the CQE, so the kernel may write here.
+        // The read region is exclusively owned by this entry (no live aliases),
+        // so deriving a `*mut` from the shared `as_ptr` is sound.
+        let ptr = unsafe { self.buf.as_ptr().add(self.pad + self.nread).cast_mut() };
+        let next_off = self.offset + self.nread as u64;
+        opcode::Read::new(types::Fd(self.file.as_raw_fd()), ptr, remaining as u32)
+            .offset(next_off)
+            .build()
+            .user_data(ud)
+    }
 }
 
 /// Where a [`ReadHandle`] is in its lifecycle (rustfs/backlog#1102).
@@ -369,7 +433,12 @@ enum HandleState {
     },
     /// The op is with the driver: its buffer lives in the pending table and is
     /// reclaimed only at the CQE.
-    Submitted,
+    Submitted {
+        /// The accepting shard's wakeup eventfd, so a cancel sent on drop wakes
+        /// the driver loop now instead of after the heartbeat
+        /// (rustfs/backlog#1163).
+        wake: Arc<EventFd>,
+    },
 }
 
 /// Handle to a read. Await it for the result.
@@ -422,6 +491,13 @@ impl Future for ReadHandle {
                 this.finished = true;
                 return Poll::Ready(Err(io::Error::other("uring driver shut down")));
             };
+            // Clone the wake before moving the WaitingPermit out, so the new
+            // Submitted state carries it for the drop-cancel path
+            // (rustfs/backlog#1163).
+            let submitted_wake = match &this.state {
+                HandleState::WaitingPermit { wake, .. } => Arc::clone(wake),
+                _ => unreachable!("state was WaitingPermit"),
+            };
             let HandleState::WaitingPermit {
                 file,
                 offset,
@@ -430,7 +506,7 @@ impl Future for ReadHandle {
                 done,
                 wake,
                 ..
-            } = std::mem::replace(&mut this.state, HandleState::Submitted)
+            } = std::mem::replace(&mut this.state, HandleState::Submitted { wake: submitted_wake })
             else {
                 unreachable!("state was WaitingPermit")
             };
@@ -474,8 +550,23 @@ impl Drop for ReadHandle {
         // until the CQE. All we may do is ask the kernel to hurry up. A handle
         // dropped before it was submitted (Inert / WaitingPermit) has no buffer,
         // no permit and no SQE, so there is nothing to cancel.
-        if matches!(self.state, HandleState::Submitted) && !self.finished && self.cancel_on_drop {
+        if let HandleState::Submitted { wake } = &self.state
+            && !self.finished
+            && self.cancel_on_drop
+        {
+            // Close the receiver BEFORE waking the driver. The wake below
+            // makes the driver process the cancel immediately, possibly while
+            // this drop is still running — before the `rx` field is
+            // destroyed. Closing it first guarantees the cancel-induced
+            // completion the driver reaps is counted as an orphan reclaim, not
+            // delivered to a receiver that is about to drop anyway
+            // (rustfs/backlog#1163).
+            self.rx.close();
             let _ = self.tx.send(Msg::Cancel { id: self.id });
+            // Wake the loop so the cancel is queued now, not after the
+            // heartbeat. On an idle ring (the hung-disk case cancel-on-drop
+            // exists for) this keeps orphan reclamation prompt.
+            wake.signal();
         }
     }
 }
@@ -564,9 +655,14 @@ impl UringDriver {
     /// joined before the error is returned.
     pub fn probe_and_start_sharded(entries: u32, shards: usize) -> Result<Self, ProbeFailure> {
         let mut started = Vec::with_capacity(shards.max(1));
-        for _ in 0..shards.max(1) {
+        for i in 0..shards.max(1) {
+            // Probe only the first shard (rustfs/backlog#1165): the probe read
+            // exercises io_uring against the environment-global temp_dir, so one
+            // confirmation is representative. Shards 2..n only create a ring and
+            // verify NODROP — this avoids `shards - 1` extra O_TMPFILE
+            // create+write+read round-trips per disk on every start and renew.
             // `?` drops `started`, whose `Shard::drop` joins each running thread.
-            started.push(Self::start_shard(entries)?);
+            started.push(Self::start_shard(entries, i == 0)?);
         }
         Ok(Self {
             shards: started,
@@ -583,7 +679,7 @@ impl UringDriver {
         &self.shards[self.rr.fetch_add(1, Ordering::Relaxed) % n]
     }
 
-    fn start_shard(entries: u32) -> Result<Shard, ProbeFailure> {
+    fn start_shard(entries: u32, probe: bool) -> Result<Shard, ProbeFailure> {
         let mut ring = IoUring::new(entries).map_err(ProbeFailure::Setup)?;
         // Require the NODROP feature (kernel >= 5.5). Without it, CQ overflow
         // silently drops CQEs, stranding pending entries forever and hanging
@@ -592,7 +688,12 @@ impl UringDriver {
         if !ring.params().is_feature_nodrop() {
             return Err(ProbeFailure::Setup(io::Error::from_raw_os_error(libc::ENOSYS)));
         }
-        probe_real_read(&mut ring).map_err(ProbeFailure::ReadOp)?;
+        // Only the first shard runs the real-read probe (rustfs/backlog#1165); the
+        // rest still create a ring and check NODROP above, which is what makes
+        // io_uring usable, but skip the redundant temp_dir round-trip.
+        if probe {
+            probe_real_read(&mut ring).map_err(ProbeFailure::ReadOp)?;
+        }
 
         // Wake the driver loop on CQEs (kernel-signaled via a registered
         // eventfd) and on new messages (submit-signaled), replacing the 200 µs
@@ -603,6 +704,15 @@ impl UringDriver {
         ring.submitter()
             .register_eventfd(cq_efd.as_raw())
             .map_err(ProbeFailure::Setup)?;
+
+        // Cap the ring's io-wq bounded worker pool so a cold-read burst cannot
+        // materialize thousands of PF_IO_WORKER threads against the process's
+        // TasksMax/RLIMIT_NPROC (rustfs/backlog#1169). Best-effort: 0 leaves the
+        // unbounded pool unchanged, and a kernel without this op (< 5.15) keeps
+        // the default — neither is fatal to a working ring.
+        let mut iowq_max = [IOWQ_MAX_BOUNDED_WORKERS, 0u32];
+        let _ = ring.submitter().register_iowq_max_workers(&mut iowq_max);
+
         let wake_efd = Arc::new(EventFd::new().map_err(ProbeFailure::Setup)?);
         let thread_wake = Arc::clone(&wake_efd);
 
@@ -613,10 +723,24 @@ impl UringDriver {
         // (2*entries), so CQ overflow is structurally unreachable (C5/C10).
         let sem = Arc::new(Semaphore::new(entries as usize));
         let thread_sem = Arc::clone(&sem);
+        // Deterministic spawn-failure seam (rustfs/backlog#1164): exercise the
+        // degrade-not-panic path without a real cgroup pids-limit. Never present
+        // in a default build.
+        #[cfg(feature = "fault-injection")]
+        if std::env::var_os("RUSTFS_URING_FAULT_SPAWN").is_some() {
+            return Err(ProbeFailure::Setup(io::Error::from_raw_os_error(libc::EAGAIN)));
+        }
+
+        // Thread creation fails with EAGAIN under a cgroup pids-limit or
+        // RLIMIT_NPROC — exactly the constrained environments the probe/degrade
+        // design exists for. Degrade to the std backend instead of panicking out
+        // of async disk init/reconnect (rustfs/backlog#1164). The spawn happens
+        // after the probe read already drained, so on failure `ring`/`cq_efd`
+        // (moved into the closure) drop cleanly with no SQE in flight.
         let handle = std::thread::Builder::new()
             .name("uring-spike-driver".into())
             .spawn(move || drive(ring, rx, thread_stats, thread_sem, cq_efd, thread_wake))
-            .expect("spawn driver thread");
+            .map_err(ProbeFailure::Setup)?;
 
         Ok(Shard {
             tx,
@@ -709,15 +833,29 @@ impl UringDriver {
             };
         }
 
-        // Reject a bad O_DIRECT alignment, and a request whose block-aligned
-        // superset range would exceed the kernel's single-read cap
-        // (rustfs/backlog#1102). `align == 1` (buffered) always passes.
+        // Reject a bad O_DIRECT alignment, a request whose block-aligned superset
+        // range would exceed the kernel's single-read cap, and one whose aligned
+        // END crosses i64::MAX — the kernel reads pos as a signed loff_t, so
+        // `kernel_offset + region_len > i64::MAX` fails at runtime with
+        // EINVAL/EOVERFLOW, exactly the errno class the C7 guard must pre-empt at
+        // submit (rustfs/backlog#1102, #1166). Pre-empting it here also makes
+        // every resubmit's `next_off < kernel_offset + region_len` provably
+        // <= i64::MAX. `align == 1` (buffered) always passes the alignment part.
         match aligned_geometry(offset, len, align) {
-            Some((_, _, region_len)) if region_len <= MAX_READ_LEN => {}
+            // CURRENT_POSITION (stream) reads use no positional offset — the
+            // kernel reads from the current file position — so the i64::MAX end
+            // check does not apply to them (their sentinel offset would overflow
+            // it). Exempt them exactly as the offset guard above does.
+            Some((kernel_offset, _, region_len))
+                if region_len <= MAX_READ_LEN
+                    && (offset == CURRENT_POSITION
+                        || kernel_offset
+                            .checked_add(region_len as u64)
+                            .is_some_and(|end| end <= i64::MAX as u64)) => {}
             _ => {
                 let _ = done.send(Err(io::Error::new(
                     io::ErrorKind::InvalidInput,
-                    "alignment must be a power of two and the block-aligned range must fit MAX_RW_COUNT",
+                    "alignment must be a power of two, and the block-aligned range must fit MAX_RW_COUNT and end within i64::MAX",
                 )));
                 return ReadHandle {
                     id,
@@ -773,7 +911,9 @@ impl UringDriver {
                     tx: shard.tx.clone(),
                     finished: false,
                     cancel_on_drop: true,
-                    state: HandleState::Submitted,
+                    state: HandleState::Submitted {
+                        wake: Arc::clone(&shard.wake_efd),
+                    },
                 }
             }
             // Saturated: `entries` ops are already in flight. Do NOT block the
@@ -826,6 +966,7 @@ impl UringDriver {
             snap.cancel_not_found += s.cancel_not_found.load(Ordering::SeqCst);
             snap.cancel_already += s.cancel_already.load(Ordering::SeqCst);
             snap.cq_overflow += s.cq_overflow.load(Ordering::SeqCst);
+            snap.submit_errors += s.submit_errors.load(Ordering::SeqCst);
         }
         snap
     }
@@ -996,13 +1137,38 @@ fn open_probe_file_exclusive(dir: &std::path::Path, pattern: &[u8]) -> io::Resul
 /// device from blocking forever; exhausting it returns an error that drives
 /// the caller's leak-over-UAF fallback.
 fn drain_probe_cqe(ring: &mut IoUring) -> io::Result<i32> {
-    const MAX_WAIT_ATTEMPTS: u32 = 4096;
-    for _ in 0..MAX_WAIT_ATTEMPTS {
-        match ring.submit_and_wait(1) {
+    // Bound the wait by WALL-CLOCK, not by an attempt count. `submit_and_wait(1)`
+    // parks in the kernel's io_cqring_wait until a CQE or a signal, so a single
+    // call can block forever when the probe read never completes — e.g. a
+    // temp_dir backed by a hung/D-state or NFS device. Since this runs on the
+    // caller's (async disk-init) thread, an unbounded block hangs startup. On
+    // kernels with EXT_ARG (>= 5.11) pass a timeout to the enter; on older
+    // kernels fall back to the blocking wait, whose only real risk is a hung
+    // temp_dir (rare) and which the deadline still re-checks between returns
+    // (rustfs/backlog#1165). On expiry, error out so the caller's leak-over-UAF
+    // fallback degrades the disk to the std backend instead of hanging.
+    const PROBE_TIMEOUT: Duration = Duration::from_secs(2);
+    let deadline = Instant::now() + PROBE_TIMEOUT;
+    let ext_arg = ring.params().is_feature_ext_arg();
+    loop {
+        let remaining = deadline.saturating_duration_since(Instant::now());
+        if remaining.is_zero() {
+            return Err(io::Error::other("probe: no CQE within the bounded wait"));
+        }
+        let waited = if ext_arg {
+            let ts = types::Timespec::new().sec(remaining.as_secs()).nsec(remaining.subsec_nanos());
+            let args = types::SubmitArgs::new().timespec(&ts);
+            ring.submitter().submit_with_args(1, &args)
+        } else {
+            ring.submit_and_wait(1)
+        };
+        match waited {
             Ok(_) => {}
-            // Signal interrupted the wait; the SQE is already in flight, so
-            // just wait again (do NOT re-push).
+            // Signal interrupted the wait; the SQE is already in flight, so wait
+            // again (do NOT re-push). The deadline still bounds the total time.
             Err(e) if e.raw_os_error() == Some(libc::EINTR) => {}
+            // EXT_ARG timeout elapsed with no CQE: loop to re-check the deadline.
+            Err(e) if e.raw_os_error() == Some(libc::ETIME) => {}
             Err(e) => return Err(e),
         }
         if let Some(cqe) = ring.completion().next() {
@@ -1017,7 +1183,6 @@ fn drain_probe_cqe(ring: &mut IoUring) -> io::Result<i32> {
             return Ok(cqe.result());
         }
     }
-    Err(io::Error::other("probe: no CQE after bounded wait"))
 }
 
 /// Owns everything the kernel can still be writing into: the ring, the
@@ -1056,6 +1221,15 @@ impl Drop for DriverState {
     }
 }
 
+/// Best-effort file length via `fstat` on the driver thread, used to tell a
+/// genuine O_DIRECT tail short read from a non-block-multiple short read that
+/// happened mid-file on a stacked filesystem (rustfs/backlog#1168). `None` when
+/// the stat fails, in which case the caller keeps the conservative EOF
+/// assumption rather than risk a wrong error or an unbounded resubmit loop.
+fn file_len(file: &File) -> Option<u64> {
+    file.metadata().ok().map(|m| m.len())
+}
+
 /// Hand the caller exactly the logical range `[head, head + want)` of the read
 /// region, truncated to what was actually read (rustfs/backlog#1102).
 ///
@@ -1086,6 +1260,79 @@ enum ReapStep {
     Resubmit(io_uring::squeue::Entry),
 }
 
+/// Queue at most one `AsyncCancel` per op (rustfs/backlog#1167): a drop-cancel
+/// followed by a shutdown, or the submit-error shutdown, must not enqueue a
+/// second cancel for the same id. The set is bounded by the pending table
+/// because ids are monotonic and an entry is removed when its op is reaped.
+fn queue_cancel(backlog: &mut VecDeque<io_uring::squeue::Entry>, queued_cancels: &mut HashSet<u64>, id: u64) {
+    if queued_cancels.insert(id) {
+        backlog.push_back(opcode::AsyncCancel::new(id).build().user_data(id | CANCEL_BIT));
+    }
+}
+
+/// Push as much of the backlog into the SQ as fits, stopping when the ring is
+/// full (the remainder retries next turn).
+fn flush_backlog(ring: &mut IoUring, backlog: &mut VecDeque<io_uring::squeue::Entry>) {
+    let mut sq = ring.submission();
+    while let Some(sqe) = backlog.pop_front() {
+        // SAFETY: read SQEs point into `pending`-owned buffers that live until
+        // their CQE; cancel SQEs carry no pointers.
+        if unsafe { sq.push(&sqe) }.is_err() {
+            backlog.push_front(sqe);
+            break;
+        }
+    }
+}
+
+/// Flush the backlog into the SQ and submit it, with submit-error classification
+/// (rustfs/backlog#1162). The single submit path for the whole loop: called once
+/// after intake and once more after reap when resubmits were queued. Skips the
+/// `io_uring_enter` syscall on an empty SQ (rustfs/backlog#1169). EINTR/EBUSY are
+/// transient; any other errno is counted and, after a bounded run, transitions
+/// the shard to shutdown so callers fall back to the std backend.
+fn submit_ring(
+    state: &mut DriverState,
+    stats: &DriverStats,
+    consecutive_submit_errors: &mut u32,
+    submit_error_logged: &mut bool,
+    shutting_down: &mut bool,
+    queued_cancels: &mut HashSet<u64>,
+) {
+    flush_backlog(&mut state.ring, &mut state.backlog);
+    if state.ring.submission().is_empty() {
+        return;
+    }
+    match state.ring.submit() {
+        Ok(_) => *consecutive_submit_errors = 0,
+        // CQ-overflow backpressure (EBUSY) and signal interruption (EINTR) are
+        // transient — retry next turn without counting them (C5, backlog#1056).
+        Err(e) if matches!(e.raw_os_error(), Some(libc::EBUSY) | Some(libc::EINTR)) => *consecutive_submit_errors = 0,
+        Err(e) => {
+            // The queued SQEs were not accepted, so their CQEs never arrive. A
+            // brief run may be transient (EAGAIN); a persistent one (e.g. EPERM
+            // from a seccomp/LSM policy applied after startup) must not be retried
+            // forever in silence.
+            stats.submit_errors.fetch_add(1, Ordering::SeqCst);
+            *consecutive_submit_errors += 1;
+            if !*submit_error_logged {
+                *submit_error_logged = true;
+                eprintln!("uring-spike driver: ring.submit() failed ({e}); retrying, will shut down if persistent");
+            }
+            if !*shutting_down && *consecutive_submit_errors >= MAX_CONSECUTIVE_SUBMIT_ERRORS {
+                eprintln!(
+                    "uring-spike driver: {} consecutive submit failures; shutting down so callers fall back to the std backend",
+                    *consecutive_submit_errors
+                );
+                *shutting_down = true;
+                let ids: Vec<u64> = state.pending.keys().copied().collect();
+                for id in ids {
+                    queue_cancel(&mut state.backlog, queued_cancels, id);
+                }
+            }
+        }
+    }
+}
+
 fn drive(
     ring: IoUring,
     rx: mpsc::Receiver<Msg>,
@@ -1101,6 +1348,16 @@ fn drive(
     };
     let mut shutting_down = false;
     let mut drain_deadline: Option<Instant> = None;
+    // Consecutive non-transient submit failures, and a once-only log latch, for
+    // the persistent-submit-failure escape hatch (rustfs/backlog#1162).
+    let mut consecutive_submit_errors: u32 = 0;
+    let mut submit_error_logged = false;
+    // Ids with an AsyncCancel already queued, so a drop-cancel followed by a
+    // shutdown (or vice versa) does not enqueue a second cancel for the same op —
+    // keeping total completions <= 2*entries and CQ overflow unreachable
+    // (rustfs/backlog#1167). Ids are monotonic, so an entry is removed only when
+    // its pending op is reaped; the set stays bounded by the pending table.
+    let mut queued_cancels: HashSet<u64> = HashSet::new();
 
     // Bounded-drain deadline (C4, rustfs/backlog#1055). Production always uses the
     // fixed DRAIN_TIMEOUT; a fault-injection build may shorten it via env so the
@@ -1127,7 +1384,17 @@ fn drive(
         // both eventfds after waking keeps them from staying spuriously
         // readable; a missed edge is harmless because the CQ/mpsc are re-checked
         // unconditionally below.
-        wait_for_events(&cq_efd, &wake_efd, LOOP_HEARTBEAT);
+        // Adaptive heartbeat (rustfs/backlog#1169): poll at 50 ms only while
+        // there is in-flight work to reap or a drain deadline to honor; when the
+        // shard is fully idle, wait up to IDLE_HEARTBEAT. New work still wakes us
+        // immediately via wake_efd and completions via cq_efd, so the longer idle
+        // wait only cuts timer/syscall churn.
+        let heartbeat = if shutting_down || !state.pending.is_empty() {
+            LOOP_HEARTBEAT
+        } else {
+            IDLE_HEARTBEAT
+        };
+        wait_for_events(&cq_efd, &wake_efd, heartbeat);
         cq_efd.drain();
         wake_efd.drain();
 
@@ -1178,7 +1445,7 @@ fn drive(
                             continue;
                         }
                     };
-                    let mut buf = vec![0u8; cap];
+                    let buf = vec![0u8; cap];
                     let pad = buf.as_ptr().align_offset(align);
                     // Runtime guard (not a debug-only assert): if the allocator
                     // ever returned a block `align_offset` cannot satisfy, refuse
@@ -1189,18 +1456,12 @@ fn drive(
                         continue;
                     }
 
-                    // The raw pointer is captured before `buf` moves into the
-                    // table; moving the Vec never relocates its heap block, and
-                    // the entry is only removed at the CQE. `region_len as u32`
-                    // is lossless: `submit` rejected anything > MAX_READ_LEN.
-                    //
-                    // SAFETY: `pad <= align - 1` and `pad + region_len <=
-                    // buf.len()`, so the pointer stays inside the allocation.
-                    let region_ptr = unsafe { buf.as_mut_ptr().add(pad) };
-                    let sqe = opcode::Read::new(types::Fd(file.as_raw_fd()), region_ptr, region_len as u32)
-                        .offset(kernel_offset)
-                        .build()
-                        .user_data(id);
+                    // Move the buffer into the pending table (which owns it until
+                    // the CQE), THEN build the SQE from the entry: the initial read
+                    // is `read_sqe` with `nread == 0`, so the read-region pointer
+                    // math and the `Read` builder live in exactly one place.
+                    // Moving the Vec never relocates its heap block, so the pointer
+                    // the SQE captures stays valid.
                     state.pending.insert(
                         id,
                         Pending {
@@ -1217,25 +1478,24 @@ fn drive(
                             want: len,
                             region_len,
                             align,
+                            transient_retries: 0,
                         },
                     );
+                    let sqe = state.pending.get(&id).expect("just inserted").read_sqe(id);
                     stats.submitted.fetch_add(1, Ordering::SeqCst);
                     stats.in_flight.fetch_add(1, Ordering::SeqCst);
                     state.backlog.push_back(sqe);
                 }
                 Msg::Cancel { id } => {
                     if state.pending.contains_key(&id) {
-                        state
-                            .backlog
-                            .push_back(opcode::AsyncCancel::new(id).build().user_data(id | CANCEL_BIT));
+                        queue_cancel(&mut state.backlog, &mut queued_cancels, id);
                     }
                 }
                 Msg::Shutdown => {
                     shutting_down = true;
-                    for id in state.pending.keys() {
-                        state
-                            .backlog
-                            .push_back(opcode::AsyncCancel::new(*id).build().user_data(*id | CANCEL_BIT));
+                    let ids: Vec<u64> = state.pending.keys().copied().collect();
+                    for id in ids {
+                        queue_cancel(&mut state.backlog, &mut queued_cancels, id);
                     }
                 }
                 #[cfg(feature = "fault-injection")]
@@ -1251,30 +1511,16 @@ fn drive(
             }
         }
 
-        // 2. Push backlog into the SQ (stop when full; retry next turn).
-        {
-            let mut sq = state.ring.submission();
-            while let Some(sqe) = state.backlog.pop_front() {
-                // SAFETY: read SQEs point into `pending`-owned buffers that
-                // live until their CQE; cancel SQEs carry no pointers.
-                if unsafe { sq.push(&sqe) }.is_err() {
-                    state.backlog.push_front(sqe);
-                    break;
-                }
-            }
-        }
-        match state.ring.submit() {
-            Ok(_) => {}
-            Err(e) if e.raw_os_error() == Some(libc::EBUSY) => {
-                // CQ-overflow backpressure on pre-5.19 NODROP kernels: the
-                // kernel refuses new submissions until we reap. Keep the
-                // backlog and reap this turn instead of spinning (C5,
-                // rustfs/backlog#1056).
-            }
-            Err(_) => {
-                // EINTR and friends: retry on the next loop turn.
-            }
-        }
+        // 2. Flush the backlog into the SQ and submit it (the single submit path;
+        //    see `submit_ring`).
+        submit_ring(
+            &mut state,
+            &stats,
+            &mut consecutive_submit_errors,
+            &mut submit_error_logged,
+            &mut shutting_down,
+            &mut queued_cancels,
+        );
 
         // 3. Reap. A Pending entry (and thus its buffer) is dropped ONLY when
         //    the logical read finishes; a short read is resubmitted for the
@@ -1315,13 +1561,34 @@ fn drive(
             let step = {
                 let p = state.pending.get_mut(&ud).expect("checked above");
                 if res < 0 {
-                    // Error (incl. ECANCELED) terminates the logical read.
-                    ReapStep::Finish(Err(io::Error::from_raw_os_error(-res)))
+                    let err = -res;
+                    // C7 three-class contract (rustfs/backlog#1166): a transient
+                    // errno (EINTR/EAGAIN) must be retried, not surfaced as the
+                    // read's final result — surfacing it would also discard the
+                    // already-read prefix of a resubmit. Bounded per logical read
+                    // so a storm cannot spin the driver thread. Streams
+                    // (CURRENT_POSITION) cannot resubmit positionally; ECANCELED
+                    // and every other errno terminate the logical read.
+                    let transient = err == libc::EINTR || err == libc::EAGAIN;
+                    if transient
+                        && p.offset != CURRENT_POSITION
+                        && p.nread < p.region_len
+                        && p.transient_retries < MAX_TRANSIENT_RETRIES
+                    {
+                        p.transient_retries += 1;
+                        ReapStep::Resubmit(p.read_sqe(ud))
+                    } else {
+                        // Error (incl. ECANCELED, or a transient errno past its
+                        // retry budget) terminates the logical read.
+                        ReapStep::Finish(Err(io::Error::from_raw_os_error(err)))
+                    }
                 } else if res == 0 {
                     // Real EOF: deliver whatever of the logical range was read.
                     ReapStep::Finish(Ok(deliver(p)))
                 } else {
                     p.nread += res as usize;
+                    // Progress resets the transient-retry budget (rustfs/backlog#1166).
+                    p.transient_retries = 0;
                     // Only POSITIONED reads (read_at / read_at_direct, whole-range
                     // pread contract) resubmit a short read. CURRENT_POSITION
                     // reads (read_current on pipes/streams) follow read(2)
@@ -1330,28 +1597,36 @@ fn drive(
                     // for stream data that may never come.
                     let is_stream = p.offset == CURRENT_POSITION;
                     let covered = p.nread >= p.head + p.want;
-                    // An O_DIRECT resubmit must stay block-aligned. The kernel
-                    // returns block multiples except at the file tail, so a
-                    // non-multiple means we reached EOF: stop and deliver.
-                    let unaligned_tail = p.align > 1 && !p.nread.is_multiple_of(p.align);
-                    if is_stream || covered || unaligned_tail || p.nread >= p.region_len {
+                    if is_stream || covered || p.nread >= p.region_len {
                         ReapStep::Finish(Ok(deliver(p)))
+                    } else if p.align > 1 && !p.nread.is_multiple_of(p.align) {
+                        // O_DIRECT non-block-multiple short read below the covered
+                        // range. The kernel returns block multiples EXCEPT at the
+                        // file tail — but a stacked filesystem (NFS/FUSE, or a
+                        // signal-split direct I/O) can legally return a non-multiple
+                        // mid-file, and assuming EOF there would silently truncate
+                        // the delivered range. Disambiguate with the actual file
+                        // length instead of inferring it (rustfs/backlog#1168).
+                        match file_len(&p.file) {
+                            // Genuine tail: at or past EOF — deliver what we read.
+                            Some(len) if p.offset + p.nread as u64 >= len => ReapStep::Finish(Ok(deliver(p))),
+                            // Mid-file non-multiple: an O_DIRECT read cannot resubmit
+                            // from a non-block-aligned offset, so surface an error
+                            // rather than truncate. The integration falls back to
+                            // the std backend for this read, preserving correctness.
+                            Some(_) => ReapStep::Finish(Err(io::Error::other(
+                                "io_uring O_DIRECT: non-block-aligned short read before EOF",
+                            ))),
+                            // fstat failed: keep the conservative EOF assumption
+                            // rather than risk a wrong error or an infinite loop.
+                            None => ReapStep::Finish(Ok(deliver(p))),
+                        }
                     } else {
-                        // Positioned short read, not EOF: resubmit the remainder
-                        // into the read region. The buffer stays owned by the
-                        // driver and in_flight is unchanged — one logical op.
-                        // For a direct read, `pad + nread` and `offset + nread`
-                        // are both block-aligned, as is `remaining`.
-                        let remaining = p.region_len - p.nread;
-                        // SAFETY: `pad + nread < pad + region_len <= buf.len()`,
-                        // and the buffer lives in the pending table until the CQE.
-                        let ptr = unsafe { p.buf.as_mut_ptr().add(p.pad + p.nread) };
-                        let next_off = p.offset + p.nread as u64;
-                        let sqe = opcode::Read::new(types::Fd(p.file.as_raw_fd()), ptr, remaining as u32)
-                            .offset(next_off)
-                            .build()
-                            .user_data(ud);
-                        ReapStep::Resubmit(sqe)
+                        // Positioned short read, not EOF, block-aligned: resubmit
+                        // the remainder into the read region. The buffer stays
+                        // owned by the driver and in_flight is unchanged — one
+                        // logical op.
+                        ReapStep::Resubmit(p.read_sqe(ud))
                     }
                 }
             };
@@ -1371,6 +1646,10 @@ fn drive(
                         Err(_) => stats.orphan_reclaimed.fetch_add(1, Ordering::SeqCst),
                     };
                     stats.in_flight.fetch_sub(1, Ordering::SeqCst);
+                    // Drop any queued-cancel bookkeeping for this now-gone op so
+                    // the dedup set stays bounded by the pending table
+                    // (rustfs/backlog#1167).
+                    queued_cancels.remove(&ud);
                     // `p` (and with it `_permit`) is dropped here, at the CQE
                     // and pending-table removal — never at future drop (C10,
                     // rustfs/backlog#1060). No manual release to forget.
@@ -1379,15 +1658,31 @@ fn drive(
             }
         }
 
-        // Monitor CQ overflow. With NODROP (asserted at probe) the crate's
-        // submit() auto-flushes the kernel overflow list, so this should stay
-        // 0; any non-zero value means CQEs were lost — pending entries would
-        // never be reclaimed. Record it as a fatal signal (C5,
-        // rustfs/backlog#1056).
+        // A short-read resubmit queued during reap must reach the kernel in THIS
+        // turn, not wait out the next heartbeat (rustfs/backlog#1163). Reap runs
+        // after the submit above, so re-run the single submit path when reap left
+        // work in the backlog; an idle turn leaves it empty and skips the call.
+        if !state.backlog.is_empty() {
+            submit_ring(
+                &mut state,
+                &stats,
+                &mut consecutive_submit_errors,
+                &mut submit_error_logged,
+                &mut shutting_down,
+                &mut queued_cancels,
+            );
+        }
+
+        // Monitor CQ overflow. With NODROP (asserted at probe) overflowed CQEs
+        // are BUFFERED in the kernel overflow list and flushed on the next enter,
+        // never lost — so a non-zero value is a backpressure warning, not fatal
+        // loss (rustfs/backlog#1056, #1167). In-flight reads are capped at
+        // `entries` and cancels are deduped (at most one per op), keeping total
+        // completions <= 2*entries, so this should stay 0 in practice.
         let overflow = state.ring.completion().overflow();
         if overflow != 0 {
             stats.cq_overflow.store(overflow as u64, Ordering::SeqCst);
-            eprintln!("uring-spike driver: CQ overflow = {overflow}; CQEs lost — treat as fatal in P2");
+            eprintln!("uring-spike driver: CQ overflow = {overflow}; CQEs buffered (NODROP), not lost — backpressure warning");
         }
 
         // 4. Exit when drained: the kernel no longer references any buffer, so
@@ -1412,10 +1707,30 @@ fn drive(
                      leaking ring + buffers to stay memory-safe",
                     state.pending.len()
                 );
-                // Close the semaphore so any handle awaiting a permit resolves
-                // with a driver-gone error. The leaked pending entries keep
-                // their permits, which is fine: nothing will wait on them.
+                // Fail every stranded caller BEFORE leaking the pending table.
+                // `oneshot::Sender::send` consumes the sender and never touches
+                // `p.buf`, so the kernel-owned buffer stays allocated (leak over
+                // UAF preserved) while an awaited `ReadHandle` resolves with an
+                // error instead of pending forever — every other driver-gone path
+                // already delivers an error, and this one must too
+                // (rustfs/backlog#1161).
+                for p in state.pending.values_mut() {
+                    if let Some(tx) = p.done.take() {
+                        let _ = tx.send(Err(io::Error::other("uring driver leaked op on bounded-drain timeout")));
+                    }
+                }
+                // Close the semaphore so any handle still awaiting a permit
+                // resolves with a driver-gone error too. The leaked pending
+                // entries keep their permits, which is fine: nothing waits on
+                // them any more.
                 sem.close();
+                // The leaked ring still has `cq_efd` registered via
+                // IORING_REGISTER_EVENTFD and in-flight ops that may post CQEs, so
+                // the eventfd must outlive it. Leak it alongside the ring instead
+                // of letting the returning `drive` drop (close) it out from under
+                // the still-mapped ring, honoring start_shard's documented "cq_efd
+                // outlives the ring" invariant on this exit too (rustfs/backlog#1167).
+                std::mem::forget(cq_efd);
                 std::mem::forget(state);
                 return;
             }
diff --git a/src/lib.rs b/src/lib.rs
index aada8cf..e6ae45e 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -30,12 +30,11 @@
 //!   unmapped.
 //!
 //! Status: read path only, Linux only. The driver supports positioned buffered
-//! reads, `O_DIRECT` reads with internal alignment, sharded rings, async
-//! backpressure, eventfd-driven reaping, graceful restricted-environment
-//! detection, and bounded shutdown drain. The write path is intentionally out
-//! of scope for `0.1.0`; see the
-//! [design notes](https://github.com/rustfs/uring/blob/v0.1.0/docs/DESIGN.md)
-//! for the invariant details.
+//! reads, `O_DIRECT` reads with internal alignment, stream (`read(2)`) reads,
+//! sharded rings, async backpressure, eventfd-driven reaping, graceful
+//! restricted-environment detection, and bounded shutdown drain. The write path
+//! is intentionally out of scope. See the crate README and the per-item docs
+//! below for the invariant details.
 
 #[cfg(target_os = "linux")]
 mod driver;
diff --git a/tests/cancel.rs b/tests/cancel.rs
index 5ff104b..3ccebd4 100644
--- a/tests/cancel.rs
+++ b/tests/cancel.rs
@@ -712,3 +712,52 @@ async fn sharded_driver_with_one_shard_matches_single_ring() {
     assert_eq!(snap.delivered + snap.orphan_reclaimed, snap.submitted, "{snap:?}");
     let _ = std::fs::remove_file(path);
 }
+
+/// Deterministic sharded cancel routing (backlog#1180). A `ReadHandle` carries
+/// the tx/wake of the shard that accepted its op, so a drop-cancel must reach
+/// THAT shard's ring. The existing sharded stress test cannot prove this: its
+/// dropped regular-file reads complete on their own, so a mis-routed cancel
+/// (answered -ENOENT) still conserves. Here every op is a blocked-pipe read that
+/// can ONLY finish by being canceled, and there is more than one op per shard, so
+/// a cancel that landed on the wrong ring would leave its read stuck and
+/// in_flight would never reach 0.
+#[tokio::test(flavor = "multi_thread")]
+async fn sharded_cancel_routes_to_the_owning_ring() {
+    const SHARDS: usize = 4;
+    // Comfortably more than SHARDS so round-robin gives every shard at least one.
+    const OPS: usize = 16;
+    let Some(driver) = sharded_driver_or_skip("sharded_cancel_routes_to_the_owning_ring", SHARDS) else {
+        return;
+    };
+    let (pipe_read, pipe_write) = os_pipe();
+
+    let mut handles = Vec::new();
+    for _ in 0..OPS {
+        handles.push(driver.read_current(Arc::clone(&pipe_read), 4096));
+    }
+    assert!(
+        wait_until(Duration::from_secs(2), || driver.stats().in_flight == OPS as u64).await,
+        "not all reads reached in-flight: {:?}",
+        driver.stats()
+    );
+
+    // Drop every handle: each sends AsyncCancel to the shard that accepted it.
+    drop(handles);
+
+    assert!(
+        wait_until(Duration::from_secs(3), || {
+            let s = driver.stats();
+            s.in_flight == 0 && s.orphan_reclaimed == OPS as u64
+        })
+        .await,
+        "sharded cancels did not reclaim every orphan — a cancel routed to the wrong ring? {:?}",
+        driver.stats()
+    );
+    // Blocked-pipe reads cannot complete on their own, so reclaiming every one
+    // proves the cancels — not a timeout — did it.
+    let snap = driver.stats();
+    assert!(snap.cancel_succeeded > 0, "orphans not reclaimed by cancel: {snap:?}");
+
+    drop(pipe_write);
+    driver.shutdown();
+}
diff --git a/tests/fault_injection.rs b/tests/fault_injection.rs
index 9dfad04..3500d2c 100644
--- a/tests/fault_injection.rs
+++ b/tests/fault_injection.rs
@@ -190,6 +190,63 @@ fn bounded_drain_bails_out_and_leaks_on_a_stuck_op() {
     drop(pipe_write);
 }
 
+/// #1161: the bounded-drain bailout must FAIL every stranded caller before it
+/// leaks the pending table, so a `ReadHandle` still awaited across the timeout
+/// resolves with an error instead of hanging forever. Same stuck-op seam as
+/// above, but this time the handle is kept and awaited after shutdown.
+#[test]
+fn stranded_handle_errors_after_bounded_drain_bailout() {
+    // SAFETY: `--test-threads=1` serializes tests; no concurrent env readers.
+    unsafe {
+        std::env::set_var("RUSTFS_URING_FAULT_STUCK_DRAIN", "1");
+        std::env::set_var("RUSTFS_URING_FAULT_DRAIN_TIMEOUT_MS", "400");
+    }
+
+    let driver = match UringDriver::probe_and_start(64) {
+        Ok(d) => d,
+        Err(e) => {
+            clear_stuck_env();
+            assert!(
+                e.is_expected_restriction(),
+                "probe failed OUTSIDE the expected restriction errno class: {e:?}"
+            );
+            eprintln!("SKIP stranded_handle_errors_after_bounded_drain_bailout: restricted environment ({e:?})");
+            return;
+        }
+    };
+
+    let (pipe_read, pipe_write) = os_pipe();
+    // Keep the handle (do not drop it): its oneshot receiver must still be live
+    // when the bailout runs so we can prove the bailout delivered an error.
+    let handle = driver.read_current(Arc::clone(&pipe_read), 4096).without_cancel_on_drop();
+    assert!(
+        wait_until(Duration::from_secs(2), || driver.stats().in_flight == 1),
+        "read never reached in-flight state"
+    );
+
+    // Drive the stuck op to the bounded-drain bailout.
+    let snap = driver.shutdown();
+    clear_stuck_env();
+    assert!(
+        snap.in_flight >= 1,
+        "the stuck op should still count as in flight after the leak-over-UAF bailout: {snap:?}"
+    );
+
+    // The still-awaited handle MUST resolve with an error within a bounded
+    // window — never hang (rustfs/backlog#1161). Before the fix the forgotten
+    // oneshot sender left this future pending forever and the timeout would fire.
+    let rt = tokio::runtime::Builder::new_current_thread()
+        .enable_time()
+        .build()
+        .expect("build current-thread runtime");
+    match rt.block_on(async { tokio::time::timeout(Duration::from_secs(2), handle).await }) {
+        Ok(Err(_)) => {} // driver-gone error delivered — correct.
+        Ok(Ok(_)) => panic!("stranded handle unexpectedly succeeded on a stuck op"),
+        Err(_) => panic!("stranded handle HUNG after bounded-drain bailout (rustfs/backlog#1161 regression)"),
+    }
+    drop(pipe_write);
+}
+
 fn clear_stuck_env() {
     // SAFETY: `--test-threads=1` teardown; no concurrent environment readers.
     unsafe {
@@ -238,3 +295,43 @@ fn probe_drain_failure_leaks_and_degrades() {
         }
     }
 }
+
+/// #1164: driver-thread spawn failure (EAGAIN under a cgroup pids-limit /
+/// RLIMIT_NPROC) must degrade to a `ProbeFailure` so the caller selects the std
+/// backend — never panic out of async disk init. The seam forces the spawn step
+/// to fail; the probe must return an error, not unwind, and leave global state
+/// intact for a later normal probe.
+#[test]
+fn spawn_failure_degrades_instead_of_panicking() {
+    // SAFETY: `--test-threads=1` serializes tests; no concurrent env readers.
+    unsafe {
+        std::env::set_var("RUSTFS_URING_FAULT_SPAWN", "1");
+    }
+    let result = UringDriver::probe_and_start_sharded(64, 3);
+    // SAFETY: `--test-threads=1` teardown.
+    unsafe {
+        std::env::remove_var("RUSTFS_URING_FAULT_SPAWN");
+    }
+
+    match result {
+        Ok(_) => panic!("driver started despite the forced spawn failure"),
+        Err(e) if e.is_expected_restriction() => {
+            // leg 1: io_uring was blocked before the probe reached the spawn seam.
+            eprintln!("SKIP spawn_failure_degrades_instead_of_panicking: restricted environment ({e:?})");
+        }
+        Err(e) => {
+            // leg 2: the forced EAGAIN propagated as a ProbeFailure (degrade, not
+            // panic). EAGAIN is not a restriction errno, so the caller logs an
+            // unexpected-probe warning and still falls back to the std backend.
+            assert!(
+                !e.is_expected_restriction(),
+                "forced spawn EAGAIN must not look like an environment restriction: {e:?}"
+            );
+            // A subsequent NORMAL probe must still succeed — the aborted start
+            // cleaned up any partially-started shard without corrupting state.
+            let driver =
+                UringDriver::probe_and_start_sharded(64, 2).expect("a normal probe after a forced spawn failure must still work");
+            let _ = driver.shutdown();
+        }
+    }
+}