diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 5feaad8..bfb2907 100644 --- a/.github/workflows/ci.yml +++ b/.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 diff --git a/Cargo.toml b/Cargo.toml index f5c95eb..0fd4dc3 100644 --- a/Cargo.toml +++ b/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"] } diff --git a/run-docker.sh b/run-docker.sh index 41030ba..2625eb8 100755 --- a/run-docker.sh +++ b/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 "==================================================================" diff --git a/src/driver.rs b/src/driver.rs index 8cb1dca..e4df6eb 100644 --- a/src/driver.rs +++ b/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 { 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 = 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 diff --git a/tests/fault_injection.rs b/tests/fault_injection.rs new file mode 100644 index 0000000..9dfad04 --- /dev/null +++ b/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) { + 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(); + } + } +}