test(driver): deterministic fault injection for the cancel-safety escape hatches (backlog#1103)

.github/workflows/ci.yml

@@ -40,7 +40,7 @@ jobs:
       - name: rustfmt
         run: cargo fmt --check
       - name: clippy
-        run: cargo clippy --all-targets -- -D warnings
+        run: cargo clippy --all-targets --all-features -- -D warnings
 
   test:
     name: test (real io_uring on runner)
@@ -50,11 +50,11 @@ jobs:
       - name: Install toolchain
         run: rustup toolchain install stable --profile minimal && rustup default stable
       - name: Build tests
-        run: cargo build --tests
+        run: cargo build --tests --all-features
       - name: Test and assert io_uring actually ran
         run: |
           set -o pipefail
-          cargo test -- --nocapture --test-threads=1 2>&1 | tee test.out
+          cargo test --all-features -- --nocapture --test-threads=1 2>&1 | tee test.out
           if grep -q "SKIP " test.out; then
             echo "::error::a test skipped — io_uring did not run on this runner (vacuous pass)"
             exit 1

Cargo.toml

@@ -27,6 +27,14 @@ 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
+# every published/production build; enabled only by the test harness
+# (run-docker.sh) to drive the panic-abort, bounded-drain-timeout, and
+# probe-drain leak paths deterministically. All gated code is `#[cfg(feature =
+# "fault-injection")]`, so a default build compiles none of it.
+fault-injection = []
+
 [dependencies]
 tokio = { version = "1.52.3", default-features = false, features = ["sync"] }
 

run-docker.sh

@@ -44,7 +44,7 @@ run_and_capture() {
         -e CARGO_TERM_COLOR=always \
         -w /spike \
         "$IMG" \
-        cargo test --release -- --nocapture --test-threads=1 2>&1
+        cargo test --release --features fault-injection -- --nocapture --test-threads=1 2>&1
 }
 
 echo "=================================================================="

src/driver.rs

@@ -290,6 +290,11 @@ enum Msg {
         id: u64,
     },
     Shutdown,
+    /// Test-only fault injection (rustfs/backlog#1103): unwind the driver thread
+    /// with ops in flight so the `DriverState::Drop` abort barrier (C2/#1054) is
+    /// exercised. Never present in a default build.
+    #[cfg(feature = "fault-injection")]
+    TestPanic,
 }
 
 /// One in-flight LOGICAL read. This struct — not the caller — owns everything
@@ -825,6 +830,17 @@ impl UringDriver {
         snap
     }
 
+    /// Test-only fault injection (rustfs/backlog#1103): poison one driver thread
+    /// so it panics with ops in flight, exercising the `DriverState::Drop` abort
+    /// barrier (C2/#1054). Compiled out entirely unless the `fault-injection`
+    /// feature is on — never in a default/production build.
+    #[cfg(feature = "fault-injection")]
+    pub fn test_inject_panic(&self) {
+        let shard = self.shard();
+        let _ = shard.tx.send(Msg::TestPanic);
+        shard.wake_efd.signal();
+    }
+
     /// Stop accepting work, cancel all in-flight ops, drain every ring to
     /// `in_flight == 0`, then join each driver thread. Only after that is a ring
     /// dropped/unmapped — the shutdown ordering P2 requires, per shard.
@@ -990,6 +1006,14 @@ fn drain_probe_cqe(ring: &mut IoUring) -> io::Result<i32> {
             Err(e) => return Err(e),
         }
         if let Some(cqe) = ring.completion().next() {
+            // fault-injection (backlog#1103 → C1/#1053): the real CQE has arrived,
+            // so the kernel is finished with the probe buffer. Forcing the error
+            // path here exercises probe_real_read's leak-over-UAF fallback with no
+            // live in-flight write to race.
+            #[cfg(feature = "fault-injection")]
+            if std::env::var_os("RUSTFS_URING_FAULT_PROBE_DRAIN").is_some() {
+                return Err(io::Error::other("fault-injection: forced probe drain failure"));
+            }
             return Ok(cqe.result());
         }
     }
@@ -1078,6 +1102,24 @@ fn drive(
     let mut shutting_down = false;
     let mut drain_deadline: Option<Instant> = None;
 
+    // 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
+    // leak-over-UAF escape hatch is testable without a 5 s wait (backlog#1103).
+    // Read once here (not per turn) so a `--test-threads=1` env toggle in one
+    // test never leaks into another's already-running driver thread.
+    #[cfg(not(feature = "fault-injection"))]
+    let drain_timeout = DRAIN_TIMEOUT;
+    #[cfg(feature = "fault-injection")]
+    let drain_timeout = std::env::var("RUSTFS_URING_FAULT_DRAIN_TIMEOUT_MS")
+        .ok()
+        .and_then(|ms| ms.parse().ok())
+        .map(Duration::from_millis)
+        .unwrap_or(DRAIN_TIMEOUT);
+    // When set, drop an op's real completion on the floor so it stays pending and
+    // the bounded drain is forced onto its timeout path (backlog#1103 → C4/#1055).
+    #[cfg(feature = "fault-injection")]
+    let fault_stuck_drain = std::env::var_os("RUSTFS_URING_FAULT_STUCK_DRAIN").is_some();
+
     loop {
         // Block until a CQE is ready (the ring's registered eventfd), a new
         // message arrives (the wakeup eventfd), or the heartbeat elapses —
@@ -1196,6 +1238,16 @@ fn drive(
                             .push_back(opcode::AsyncCancel::new(*id).build().user_data(*id | CANCEL_BIT));
                     }
                 }
+                #[cfg(feature = "fault-injection")]
+                Msg::TestPanic => {
+                    // Panic WITH buffers still in flight: the abort barrier in
+                    // `DriverState::Drop` must fire rather than let the unwind
+                    // free them under the kernel (rustfs/backlog#1103 → C2/#1054).
+                    panic!(
+                        "fault-injection: driver thread panic requested with {} ops in flight",
+                        state.pending.len()
+                    );
+                }
             }
         }
 
@@ -1244,6 +1296,15 @@ fn drive(
                 };
                 continue;
             }
+            // fault-injection (backlog#1103 → C4/#1055): drop this real completion
+            // so the op stays pending and the bounded drain must take its
+            // DRAIN_TIMEOUT leak path. The CQE has already arrived, so the kernel
+            // is done with the buffer — the eventual `forget` leaks a completed
+            // allocation, never live memory.
+            #[cfg(feature = "fault-injection")]
+            if fault_stuck_drain && state.pending.contains_key(&ud) {
+                continue;
+            }
             let res = cqe.result();
             if !state.pending.contains_key(&ud) {
                 continue;
@@ -1340,7 +1401,7 @@ fn drive(
                 sem.close();
                 return; // clean drain: DriverState drops normally, ring unmaps.
             }
-            let deadline = *drain_deadline.get_or_insert_with(|| Instant::now() + DRAIN_TIMEOUT);
+            let deadline = *drain_deadline.get_or_insert_with(|| Instant::now() + drain_timeout);
             if Instant::now() >= deadline {
                 // A CQE may never arrive (ASYNC_CANCEL cannot interrupt an
                 // in-execution regular-file read on a hung disk). We must NOT

tests/fault_injection.rs

@@ -0,0 +1,240 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! Deterministic fault-injection tests for the cancel-safety escape hatches
+//! (rustfs/backlog#1103). These exercise failure modes the normal API cannot
+//! reach — a driver-thread panic (#1054), a completion that never arrives
+//! (#1055), and a probe that cannot confirm its read op (#1053) — so each needs
+//! a production seam that exists only under the `fault-injection` feature. The
+//! whole file is gated on it.
+//!
+//! Skip contract is identical to `cancel.rs`: in a restricted environment
+//! (Docker default seccomp, gVisor) the probe fails with an expected-restriction
+//! errno and every test degrades to a `SKIP`. Run under
+//! `--security-opt seccomp=unconfined` (run-docker.sh leg 2) to exercise the
+//! real io_uring paths.
+#![cfg(all(target_os = "linux", feature = "fault-injection"))]
+
+use std::fs::File;
+use std::os::fd::FromRawFd;
+use std::os::unix::process::ExitStatusExt;
+use std::sync::Arc;
+use std::time::{Duration, Instant};
+
+use rustfs_uring::UringDriver;
+
+/// A pipe whose read side blocks until the write side is written or closed — the
+/// only portable way to hold an op provably in flight.
+fn os_pipe() -> (Arc<File>, File) {
+    let mut fds = [0i32; 2];
+    // SAFETY: `fds` is a valid two-int array; pipe(2) fills it with owned fds.
+    let rc = unsafe { libc::pipe(fds.as_mut_ptr()) };
+    assert_eq!(rc, 0, "pipe(2) failed");
+    // SAFETY: both fds are freshly owned by this process.
+    let read = unsafe { File::from_raw_fd(fds[0]) };
+    let write = unsafe { File::from_raw_fd(fds[1]) };
+    (Arc::new(read), write)
+}
+
+fn wait_until(deadline: Duration, mut cond: impl FnMut() -> bool) -> bool {
+    let start = Instant::now();
+    while start.elapsed() < deadline {
+        if cond() {
+            return true;
+        }
+        std::thread::sleep(Duration::from_millis(5));
+    }
+    cond()
+}
+
+const ABORT_CHILD_ENV: &str = "RUSTFS_URING_FAULT_ABORT_CHILD";
+
+/// #1054 (C2): a driver-thread panic must ABORT the process — never unwind and
+/// free the in-flight buffers the kernel can still write into. The abort itself
+/// takes down the process, so it runs in a child re-exec of this test binary and
+/// the parent asserts the child died from `SIGABRT`.
+#[test]
+fn driver_panic_aborts_instead_of_freeing_in_flight_buffers() {
+    if std::env::var_os(ABORT_CHILD_ENV).is_some() {
+        run_abort_child();
+        // Only reachable if the abort barrier failed to fire.
+        eprintln!("child: abort barrier did NOT fire within the wait window — safety regression");
+        std::process::exit(0);
+    }
+
+    // Parent. Honor the leg-1 skip contract before spawning anything.
+    match UringDriver::probe_and_start(64) {
+        Ok(d) => {
+            let _ = d.shutdown();
+        }
+        Err(e) => {
+            assert!(
+                e.is_expected_restriction(),
+                "probe failed OUTSIDE the expected restriction errno class: {e:?}"
+            );
+            eprintln!("SKIP driver_panic_aborts_instead_of_freeing_in_flight_buffers: restricted environment ({e:?})");
+            return;
+        }
+    }
+
+    let exe = std::env::current_exe().expect("locate test binary");
+    let status = std::process::Command::new(exe)
+        .args([
+            "--exact",
+            "driver_panic_aborts_instead_of_freeing_in_flight_buffers",
+            "--nocapture",
+            "--test-threads=1",
+        ])
+        .env(ABORT_CHILD_ENV, "1")
+        .status()
+        .expect("spawn abort child");
+
+    assert_eq!(
+        status.signal(),
+        Some(libc::SIGABRT),
+        "a driver-thread panic must abort the process (leak over UAF), got {status:?}"
+    );
+}
+
+fn run_abort_child() {
+    let driver = match UringDriver::probe_and_start(64) {
+        Ok(d) => d,
+        Err(e) => {
+            // leg-1 child: no io_uring. The parent skipped too and never inspects
+            // this exit; exit cleanly.
+            eprintln!("SKIP abort child: restricted environment ({e:?})");
+            std::process::exit(0);
+        }
+    };
+    let (pipe_read, pipe_write) = os_pipe();
+    // Never awaited, never cancelled: the read stays in flight so `pending` is
+    // non-empty when the driver thread panics.
+    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"
+    );
+    driver.test_inject_panic();
+    // The abort should fire on the driver thread within milliseconds; give it a
+    // generous window before declaring a regression.
+    std::thread::sleep(Duration::from_secs(3));
+    // Kept alive until here so the read never hit EOF before the panic.
+    drop(pipe_write);
+}
+
+/// #1055 (C4): when an in-flight op's completion never comes, `shutdown`'s
+/// bounded drain must bail out at `DRAIN_TIMEOUT` and LEAK the ring + buffers
+/// (memory-safe) rather than unmap under the kernel or block forever. The seam
+/// drops the op's (cancel-induced) completion so it stays pending; the drain
+/// timeout is shortened via env so the test is fast.
+#[test]
+fn bounded_drain_bails_out_and_leaks_on_a_stuck_op() {
+    // Set before the driver thread starts — it reads these once at spawn.
+    // SAFETY: `--test-threads=1` serializes tests, so no other thread reads the
+    // environment concurrently.
+    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 bounded_drain_bails_out_and_leaks_on_a_stuck_op: restricted environment ({e:?})");
+            return;
+        }
+    };
+
+    let (pipe_read, pipe_write) = os_pipe();
+    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"
+    );
+
+    let start = Instant::now();
+    let snap = driver.shutdown();
+    let elapsed = start.elapsed();
+    clear_stuck_env();
+
+    assert!(
+        elapsed >= Duration::from_millis(300),
+        "shutdown returned before the bounded-drain deadline ({elapsed:?}) — the timeout path was not taken"
+    );
+    assert!(
+        elapsed < Duration::from_secs(3),
+        "shutdown took {elapsed:?} — the shortened fault drain-timeout was not honored (real 5 s path?)"
+    );
+    assert!(
+        snap.in_flight >= 1,
+        "the stuck op should still count as in flight after the leak-over-UAF bailout: {snap:?}"
+    );
+    // Reaching here without a crash proves the ring + buffers were leaked (kept
+    // mapped/allocated), not unmapped under an in-flight op.
+    drop(pipe_write);
+}
+
+fn clear_stuck_env() {
+    // SAFETY: `--test-threads=1` teardown; no concurrent environment readers.
+    unsafe {
+        std::env::remove_var("RUSTFS_URING_FAULT_STUCK_DRAIN");
+        std::env::remove_var("RUSTFS_URING_FAULT_DRAIN_TIMEOUT_MS");
+    }
+}
+
+/// #1053 (C1): if the startup probe cannot confirm its read op terminated it
+/// must LEAK the probe buffer/file (the kernel may still write into them) and
+/// return a `ReadOp` failure — never drop them under an in-flight SQE and never
+/// crash. The seam forces the failure only AFTER the real CQE has arrived, so no
+/// live write races the leak; the point is to cover the fallback branch and
+/// prove a forced probe failure degrades cleanly without corrupting global state.
+#[test]
+fn probe_drain_failure_leaks_and_degrades() {
+    // SAFETY: `--test-threads=1` setup; no concurrent environment readers.
+    unsafe {
+        std::env::set_var("RUSTFS_URING_FAULT_PROBE_DRAIN", "1");
+    }
+    let result = UringDriver::probe_and_start(64);
+    // SAFETY: `--test-threads=1` teardown.
+    unsafe {
+        std::env::remove_var("RUSTFS_URING_FAULT_PROBE_DRAIN");
+    }
+
+    match result {
+        Ok(_) => panic!("probe must fail when the drain is forced to error"),
+        Err(e) if e.is_expected_restriction() => {
+            // leg 1: io_uring was blocked before the probe read op ran, so the
+            // forced-drain fault was never reached. Honor the skip contract.
+            eprintln!("SKIP probe_drain_failure_leaks_and_degrades: restricted environment ({e:?})");
+        }
+        Err(e) => {
+            // leg 2: the forced ReadOp failure ran the leak-over-UAF fallback and
+            // the process is still healthy. It must NOT be misclassified as an
+            // environment restriction (that would wrongly latch a healthy disk).
+            assert!(
+                !e.is_expected_restriction(),
+                "forced probe-drain failure must not look like an environment restriction: {e:?}"
+            );
+            // A subsequent NORMAL probe must still succeed — the forced fault did
+            // not corrupt any global state.
+            let driver = UringDriver::probe_and_start(64).expect("a normal probe after a forced-fault probe must still work");
+            let _ = driver.shutdown();
+        }
+    }
+}