From 3b1bed700995ae5a36cb6a501ec9021e678f1f51 Mon Sep 17 00:00:00 2001 From: Zhengchao An Date: Fri, 17 Jul 2026 09:04:32 +0800 Subject: [PATCH] test(e2e): socket-level network fault-injection proxy (backlog#1325) (#4938) test(e2e): add socket-level network fault-injection proxy (backlog#1325) Add `FaultProxy`, an in-process async TCP proxy for black-box cluster E2E tests, under `crates/e2e_test/src/fault_proxy.rs`. It binds a random local port, forwards every accepted connection to a fixed target, and lets a test flip the wire behaviour at runtime. Fault modes: `Pass` (transparent), `Latency(Duration)` (delay each forwarded chunk), `Blackhole` (accept but forward nothing either way and never respond), and `Partition(Direction)` (block exactly one direction while the other keeps flowing). Modes are switchable at runtime via `set_mode`, and `shutdown` stops the accept loop and drops the listener so the bound port is released. This is the black-box counterpart to the in-process white-box hooks (rename barrier, disk call counters), which cannot reach across the process boundary into a spawned RustFS server. It serves the lock-plane one-way partition and accept-then-blackhole-peer acceptance for #1312/#1319 and the cross-process replay/tamper seam for #1327. Wiring it into the cluster harness (expose a node port through a proxy) plus the 5GiB / nightly CI-lane budget are follow-ups, since the harness multi-drive / 2-pool work lands separately (rustfs#4937); this block stays independent and self-tested. Self-tests run against a loopback echo server that records received bytes, covering: pass round-trip identity, latency lower-bound delay, blackhole no-response-no-panic, one-way partition in both directions (request-path vs return-path), runtime mode switching on a live connection, and port release after shutdown. Timing assertions use loose lower bounds and bounded read timeouts only, never fixed-sleep absolute-value checks, to stay non-flaky. No product code changes; the module is `#[cfg(test)]`-gated. --- crates/e2e_test/src/fault_proxy.rs | 498 +++++++++++++++++++++++++++++ crates/e2e_test/src/lib.rs | 8 + 2 files changed, 506 insertions(+) create mode 100644 crates/e2e_test/src/fault_proxy.rs diff --git a/crates/e2e_test/src/fault_proxy.rs b/crates/e2e_test/src/fault_proxy.rs new file mode 100644 index 000000000..1879532ca --- /dev/null +++ b/crates/e2e_test/src/fault_proxy.rs @@ -0,0 +1,498 @@ +// 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. + +//! In-process, socket-level network fault-injection proxy for black-box +//! cluster E2E tests (backlog#1325 network fault-injection block). +//! +//! `FaultProxy` binds a TCP listener on a random local port, forwards every +//! accepted connection to a fixed target address, and lets a test flip the +//! forwarding behaviour at runtime. It is the black-box counterpart to the +//! in-process white-box hooks (rename barrier, disk call counters): those +//! `#[cfg(test)]` primitives cannot reach across the process boundary into a +//! spawned RustFS server, whereas this proxy sits on the wire between two +//! nodes and needs no cooperation from the peer. +//! +//! Service targets: +//! - **#1312 / #1319** lock-plane one-way partition and accept-then-blackhole +//! peer: run a node's lock/RPC port through the proxy, then +//! [`FaultMode::Partition`] one direction (data plane stays reachable via a +//! separate direct port) or [`FaultMode::Blackhole`] the whole connection. +//! - **#1327** cross-process replay / tamper harness: the proxy is the wire +//! seam a replay/tamper filter can later hook into. +//! +//! Supported fault modes: +//! - [`FaultMode::Pass`]: transparent byte forwarding (round-trip identical). +//! - [`FaultMode::Latency`]: delay every forwarded chunk by a fixed duration. +//! - [`FaultMode::Blackhole`]: accept the connection but forward nothing in +//! either direction and never respond (models an accepted-then-dead peer); +//! the client observes a read timeout, the proxy never panics. +//! - [`FaultMode::Partition`]: block exactly one direction while the other +//! keeps flowing (models a lock-RPC one-way partition with a live data +//! plane). +//! +//! Wiring this into the cluster harness ("expose node port N through a proxy") +//! is intentionally left as a follow-up: the harness multi-drive / 2-pool +//! changes land separately (rustfs#4937), so this block ships the proxy tool +//! plus its local self-tests against a loopback echo server and stays +//! independent of that harness work. +//! +//! # Example +//! +//! ```ignore +//! let proxy = FaultProxy::start(target_addr).await?; +//! let via = proxy.local_addr(); // point the client here instead of `target_addr` +//! proxy.set_mode(FaultMode::Latency(Duration::from_millis(50))); +//! // ... exercise the client ... +//! proxy.set_mode(FaultMode::Partition(Direction::ClientToServer)); +//! // ... assert the blocked direction is dead ... +//! proxy.shutdown().await; // releases the listener port +//! ``` + +use std::io; +use std::net::{Ipv4Addr, SocketAddr}; +use std::time::Duration; + +use tokio::io::{AsyncReadExt, AsyncWriteExt}; +use tokio::net::{TcpListener, TcpStream}; +use tokio::sync::watch; +use tokio::task::JoinHandle; + +/// A single logical direction of a proxied connection. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum Direction { + /// Bytes travelling from the connecting client towards the target server + /// (the forward / request path). + ClientToServer, + /// Bytes travelling from the target server back to the client (the return + /// / response path). + ServerToClient, +} + +/// Runtime-switchable forwarding behaviour of a [`FaultProxy`]. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum FaultMode { + /// Forward bytes transparently in both directions. + Pass, + /// Forward bytes in both directions, but delay every chunk by this + /// duration before it is written onward. + Latency(Duration), + /// Accept connections but forward nothing in either direction and never + /// respond. Bytes read from either side are drained and discarded so the + /// proxy never blocks or panics; the peer simply never hears back. + Blackhole, + /// Block exactly one direction (drop its bytes) while the other direction + /// keeps forwarding normally. + Partition(Direction), +} + +/// An async TCP proxy that forwards a random local port to a fixed target and +/// can inject network faults at runtime. +/// +/// The proxy owns a background accept loop; each accepted connection is +/// handled by its own task pair (one per direction). [`FaultProxy::shutdown`] +/// stops the accept loop and drops the listener, releasing the bound port. +pub struct FaultProxy { + listen_addr: SocketAddr, + target: SocketAddr, + mode_tx: watch::Sender, + shutdown_tx: watch::Sender, + accept_task: JoinHandle<()>, +} + +impl FaultProxy { + /// Bind a listener on `127.0.0.1:0` and start forwarding accepted + /// connections to `target`. Starts in [`FaultMode::Pass`]. + pub async fn start(target: SocketAddr) -> io::Result { + let listener = TcpListener::bind((Ipv4Addr::LOCALHOST, 0)).await?; + let listen_addr = listener.local_addr()?; + + let (mode_tx, mode_rx) = watch::channel(FaultMode::Pass); + let (shutdown_tx, shutdown_rx) = watch::channel(false); + + let accept_task = tokio::spawn(accept_loop(listener, target, mode_rx, shutdown_rx)); + + Ok(Self { + listen_addr, + target, + mode_tx, + shutdown_tx, + accept_task, + }) + } + + /// Local address the proxy is listening on. Point clients here instead of + /// the real target to route their traffic through the proxy. + pub fn local_addr(&self) -> SocketAddr { + self.listen_addr + } + + /// The fixed target address every connection is forwarded to. + pub fn target_addr(&self) -> SocketAddr { + self.target + } + + /// The mode currently in effect. + pub fn mode(&self) -> FaultMode { + *self.mode_tx.borrow() + } + + /// Switch the fault mode at runtime. Takes effect on the next chunk read by + /// each direction of every live and future connection. + pub fn set_mode(&self, mode: FaultMode) { + // A send only fails if every receiver has dropped, i.e. the accept loop + // and all connections have already ended; there is nothing to steer. + let _ = self.mode_tx.send(mode); + } + + /// Stop the accept loop, signal live connections to wind down, and drop the + /// listener so the bound port is released. Awaits the accept loop so the + /// port is free once this returns. + pub async fn shutdown(self) { + let _ = self.shutdown_tx.send(true); + let _ = self.accept_task.await; + } +} + +/// Accept loop: takes new connections until shutdown is signalled, then returns +/// (dropping `listener`, which releases the port). +async fn accept_loop( + listener: TcpListener, + target: SocketAddr, + mode_rx: watch::Receiver, + mut shutdown_rx: watch::Receiver, +) { + loop { + tokio::select! { + biased; + _ = shutdown_rx.changed() => { + if *shutdown_rx.borrow() { + break; + } + } + accepted = listener.accept() => { + match accepted { + Ok((client, _peer)) => { + tokio::spawn(handle_connection( + client, + target, + mode_rx.clone(), + shutdown_rx.clone(), + )); + } + // A transient accept error should not tear the proxy down. + Err(_) => continue, + } + } + } + } +} + +/// Bridge one accepted client connection to a fresh connection to the target, +/// running an independent pump per direction. +async fn handle_connection( + client: TcpStream, + target: SocketAddr, + mode_rx: watch::Receiver, + shutdown_rx: watch::Receiver, +) { + let server = match TcpStream::connect(target).await { + Ok(s) => s, + // Target unreachable: nothing to bridge; drop the client. + Err(_) => return, + }; + + let (client_rd, client_wr) = client.into_split(); + let (server_rd, server_wr) = server.into_split(); + + let up = tokio::spawn(pump( + Direction::ClientToServer, + client_rd, + server_wr, + mode_rx.clone(), + shutdown_rx.clone(), + )); + let down = tokio::spawn(pump(Direction::ServerToClient, server_rd, client_wr, mode_rx, shutdown_rx)); + + let _ = up.await; + let _ = down.await; +} + +/// Copy bytes from `from` to `to` for a single direction, applying the current +/// [`FaultMode`]. Returns when the source hits EOF/error, a write fails, or +/// shutdown is signalled. +async fn pump( + dir: Direction, + mut from: R, + mut to: W, + mut mode_rx: watch::Receiver, + mut shutdown_rx: watch::Receiver, +) where + R: AsyncReadExt + Unpin, + W: AsyncWriteExt + Unpin, +{ + let mut buf = vec![0u8; 16 * 1024]; + loop { + let n = tokio::select! { + biased; + _ = shutdown_rx.changed() => break, + read = from.read(&mut buf) => match read { + Ok(0) => break, // clean EOF + Ok(n) => n, + Err(_) => break, // reset / error: end this direction + }, + }; + + // Snapshot the mode without holding the borrow across an await. + let mode = *mode_rx.borrow_and_update(); + match mode { + // Whole connection is a black hole: drain and drop. + FaultMode::Blackhole => continue, + // This direction is partitioned off: drop its bytes; the peer keeps + // reading nothing while the other direction may still flow. + FaultMode::Partition(blocked) if blocked == dir => continue, + FaultMode::Latency(delay) => { + tokio::time::sleep(delay).await; + if to.write_all(&buf[..n]).await.is_err() { + break; + } + if to.flush().await.is_err() { + break; + } + } + FaultMode::Pass | FaultMode::Partition(_) => { + if to.write_all(&buf[..n]).await.is_err() { + break; + } + if to.flush().await.is_err() { + break; + } + } + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::sync::Arc; + use std::time::Instant; + use tokio::sync::Mutex; + + /// A loopback echo server that records every byte it receives, so tests can + /// distinguish "the server never got it" (client->server blocked) from "the + /// server got it but the reply never came back" (server->client blocked). + struct EchoServer { + addr: SocketAddr, + received: Arc>>, + _task: JoinHandle<()>, + } + + impl EchoServer { + async fn start() -> io::Result { + let listener = TcpListener::bind((Ipv4Addr::LOCALHOST, 0)).await?; + let addr = listener.local_addr()?; + let received = Arc::new(Mutex::new(Vec::new())); + let received_task = received.clone(); + let task = tokio::spawn(async move { + loop { + let Ok((mut sock, _)) = listener.accept().await else { + break; + }; + let received_conn = received_task.clone(); + tokio::spawn(async move { + let mut buf = vec![0u8; 16 * 1024]; + loop { + match sock.read(&mut buf).await { + Ok(0) | Err(_) => break, + Ok(n) => { + received_conn.lock().await.extend_from_slice(&buf[..n]); + if sock.write_all(&buf[..n]).await.is_err() { + break; + } + let _ = sock.flush().await; + } + } + } + }); + } + }); + Ok(Self { + addr, + received, + _task: task, + }) + } + + async fn received(&self) -> Vec { + self.received.lock().await.clone() + } + } + + /// Write `payload`, then try to read up to `want` bytes with a deadline. + /// Returns the bytes actually received before the timeout (possibly empty). + async fn write_then_read(stream: &mut TcpStream, payload: &[u8], want: usize, timeout: Duration) -> Vec { + stream.write_all(payload).await.expect("client write"); + stream.flush().await.expect("client flush"); + + let mut out = Vec::new(); + let mut buf = vec![0u8; want.max(1)]; + let deadline = tokio::time::Instant::now() + timeout; + while out.len() < want { + let remaining = deadline.saturating_duration_since(tokio::time::Instant::now()); + if remaining.is_zero() { + break; + } + match tokio::time::timeout(remaining, stream.read(&mut buf)).await { + Ok(Ok(0)) => break, // peer closed + Ok(Ok(n)) => out.extend_from_slice(&buf[..n]), + Ok(Err(_)) => break, // connection error + Err(_) => break, // read timed out + } + } + out + } + + #[tokio::test] + async fn pass_mode_round_trips_bytes() { + let echo = EchoServer::start().await.unwrap(); + let proxy = FaultProxy::start(echo.addr).await.unwrap(); + assert_eq!(proxy.mode(), FaultMode::Pass); + + let mut client = TcpStream::connect(proxy.local_addr()).await.unwrap(); + let payload = b"hello-fault-proxy"; + let got = write_then_read(&mut client, payload, payload.len(), Duration::from_secs(5)).await; + + assert_eq!(got, payload, "pass mode must forward bytes unchanged"); + assert_eq!(echo.received().await, payload, "server must have received the bytes"); + + proxy.shutdown().await; + } + + #[tokio::test] + async fn latency_mode_delays_forwarding() { + // Loose lower bound only: sleep() guarantees *at least* the delay, so + // the assertion cannot flake on a slow machine, and we never assert an + // upper bound. + let delay = Duration::from_millis(200); + let echo = EchoServer::start().await.unwrap(); + let proxy = FaultProxy::start(echo.addr).await.unwrap(); + proxy.set_mode(FaultMode::Latency(delay)); + + let mut client = TcpStream::connect(proxy.local_addr()).await.unwrap(); + let payload = b"delayed"; + let start = Instant::now(); + let got = write_then_read(&mut client, payload, payload.len(), Duration::from_secs(5)).await; + let elapsed = start.elapsed(); + + assert_eq!(got, payload, "latency mode must still deliver the bytes"); + // One delay on the request path plus one on the echo return path: the + // round trip must exceed at least a single configured delay. + assert!(elapsed >= delay, "round trip {elapsed:?} shorter than configured delay {delay:?}"); + + proxy.shutdown().await; + } + + #[tokio::test] + async fn blackhole_mode_yields_no_response_and_no_panic() { + let echo = EchoServer::start().await.unwrap(); + let proxy = FaultProxy::start(echo.addr).await.unwrap(); + proxy.set_mode(FaultMode::Blackhole); + + let mut client = TcpStream::connect(proxy.local_addr()).await.unwrap(); + let got = write_then_read(&mut client, b"into-the-void", 1, Duration::from_millis(300)).await; + + assert!(got.is_empty(), "blackhole mode must not return any bytes, got {got:?}"); + assert!(echo.received().await.is_empty(), "blackhole must not forward to the server"); + + // Proxy is still alive and steerable after a blackholed connection. + proxy.set_mode(FaultMode::Pass); + proxy.shutdown().await; + } + + #[tokio::test] + async fn partition_client_to_server_blocks_request_path() { + let echo = EchoServer::start().await.unwrap(); + let proxy = FaultProxy::start(echo.addr).await.unwrap(); + proxy.set_mode(FaultMode::Partition(Direction::ClientToServer)); + + let mut client = TcpStream::connect(proxy.local_addr()).await.unwrap(); + let got = write_then_read(&mut client, b"blocked-request", 1, Duration::from_millis(300)).await; + + assert!(got.is_empty(), "client->server partition must yield no echo"); + assert!( + echo.received().await.is_empty(), + "client->server partition must stop bytes from reaching the server" + ); + + proxy.shutdown().await; + } + + #[tokio::test] + async fn partition_server_to_client_blocks_only_return_path() { + let echo = EchoServer::start().await.unwrap(); + let proxy = FaultProxy::start(echo.addr).await.unwrap(); + proxy.set_mode(FaultMode::Partition(Direction::ServerToClient)); + + let mut client = TcpStream::connect(proxy.local_addr()).await.unwrap(); + let payload = b"one-way"; + let got = write_then_read(&mut client, payload, 1, Duration::from_millis(300)).await; + + // Client hears nothing back... + assert!(got.is_empty(), "server->client partition must block the reply"); + // ...but the request path is live, so the server did receive the bytes. + // Poll briefly to avoid racing the forward direction. + let mut server_saw = Vec::new(); + for _ in 0..30 { + server_saw = echo.received().await; + if !server_saw.is_empty() { + break; + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + assert_eq!(server_saw, payload, "server->client partition must leave the request path intact"); + + proxy.shutdown().await; + } + + #[tokio::test] + async fn set_mode_switches_behaviour_at_runtime() { + let echo = EchoServer::start().await.unwrap(); + let proxy = FaultProxy::start(echo.addr).await.unwrap(); + + // Start passing: first exchange round-trips. + let mut client = TcpStream::connect(proxy.local_addr()).await.unwrap(); + let first = write_then_read(&mut client, b"first", 5, Duration::from_secs(5)).await; + assert_eq!(first, b"first", "initial pass exchange must round-trip"); + + // Flip to blackhole on the same live connection: the next write gets no reply. + proxy.set_mode(FaultMode::Blackhole); + let second = write_then_read(&mut client, b"second", 1, Duration::from_millis(300)).await; + assert!(second.is_empty(), "after switching to blackhole the live connection must go silent"); + + proxy.shutdown().await; + } + + #[tokio::test] + async fn shutdown_releases_the_listener_port() { + let echo = EchoServer::start().await.unwrap(); + let proxy = FaultProxy::start(echo.addr).await.unwrap(); + let addr = proxy.local_addr(); + + proxy.shutdown().await; + + // The port must be free to bind again once shutdown returns. + let rebind = TcpListener::bind(addr).await; + assert!(rebind.is_ok(), "shutdown must release the listener port {addr}"); + } +} diff --git a/crates/e2e_test/src/lib.rs b/crates/e2e_test/src/lib.rs index fef5d6b91..aeb77892e 100644 --- a/crates/e2e_test/src/lib.rs +++ b/crates/e2e_test/src/lib.rs @@ -27,6 +27,14 @@ pub mod chaos; #[cfg(test)] pub mod fake_s3_target; +// Socket-level network fault-injection proxy for black-box cluster tests +// (backlog#1325 network fault-injection block): latency / blackhole / one-way +// partition on the wire between nodes. Serves #1312/#1319 (lock-plane one-way +// partition, accept-then-blackhole peer) and #1327 (cross-process replay/tamper +// seam); cluster-harness wiring is a follow-up (rustfs#4937). +#[cfg(test)] +pub mod fault_proxy; + // Reliability tests built on the fault-injection harness #[cfg(test)] mod reliability_disk_fault_test;