diff --git a/Cargo.toml b/Cargo.toml index 5d1684a..1679a03 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -19,14 +19,47 @@ tracing = "0.1.44" tracing-subscriber = { version = "0.3.22", features = ["env-filter"] } [lints.clippy] -all = { level = "warn", priority = -1 } +# All lint groups except restriction and cargo nursery = { level = "warn", priority = -1 } pedantic = { level = "warn", priority = -1 } +# Explicit panics +expect_used = "forbid" +panic = "forbid" +todo = "forbid" +unimplemented = "forbid" +unreachable = "forbid" +unwrap_used = "forbid" + +# Implicit panics (and wrapping) +arithmetic_side_effects = "deny" # Desired for atomic counters meant to wrap +indexing_slicing = "forbid" +string_slice = "forbid" + +# Ignoring `#[must_use]` +let_underscore_must_use = "forbid" +unused_result_ok = "forbid" + +# Unsafe discipline (only relevant when `unsafe_code` is not set to "forbid") +multiple_unsafe_ops_per_block = "forbid" +undocumented_unsafe_blocks = "forbid" + +# Lint suppression allow_attributes = "forbid" allow_attributes_without_reason = "forbid" -expect_used = "forbid" -unwrap_used = "forbid" + +# Binding shadowing +shadow_reuse = "forbid" +shadow_same = "forbid" +shadow_unrelated = "forbid" + +# Redundant constructs +redundant_test_prefix = "deny" +unnecessary_self_imports = "warn" [lints.rust] unsafe_code = "forbid" + +closure_returning_async_block = "warn" +redundant_imports = "warn" +unused_qualifications = "warn" diff --git a/src/health.rs b/src/health.rs index de8fbeb..2297ef1 100644 --- a/src/health.rs +++ b/src/health.rs @@ -20,7 +20,7 @@ impl HealthChecker { Self { backends, path, interval } } - pub async fn run(self) { + pub async fn run(self) -> ! { loop { for backend in &self.backends { let is_healthy = self.check_backend(backend.addr()).await; @@ -42,14 +42,16 @@ impl HealthChecker { } async fn check_backend(&self, addr: SocketAddr) -> bool { - let check_fut = async move { + let check_fut = async { let Ok(stream) = TcpStream::connect(addr).await else { return false }; let Ok((mut sender, conn)) = client_http1::handshake(TokioIo::new(stream)).await else { return false; }; - tokio::spawn(async move { - let _ = conn.await; + tokio::spawn(async { + if let Err(e) = conn.await { + warn!("{e}"); + } }); let Ok(req) = Request::builder() @@ -78,10 +80,7 @@ mod tests { use crate::test_utils::tokio_test; use anyhow::Result; use http_body_util::Full; - use hyper::{ - Response, StatusCode, body::Bytes, server::conn::http1 as server_http1, service::service_fn, - }; - use hyper_util::rt::TokioIo; + use hyper::{Response, StatusCode, server::conn::http1 as server_http1, service::service_fn}; use std::convert::Infallible; use tokio::net::TcpListener; @@ -95,18 +94,21 @@ mod tests { let Ok((stream, _)) = listener.accept().await else { break }; tokio::spawn(async move { - let _ = server_http1::Builder::new() - .serve_connection( - TokioIo::new(stream), - service_fn(|_| async move { - let resp = Response::builder() - .status(status) - .body(Full::new(Bytes::new())) - .unwrap_or_else(|_| Response::new(Full::new(Bytes::new()))); - Ok::<_, Infallible>(resp) - }), - ) - .await; + assert!( + server_http1::Builder::new() + .serve_connection( + TokioIo::new(stream), + service_fn(async |_| { + let resp = Response::builder() + .status(status) + .body(Full::new(Bytes::new())) + .unwrap_or_else(|_| Response::new(Full::new(Bytes::new()))); + Ok::<_, Infallible>(resp) + }), + ) + .await + .is_ok() + ); }); } }); @@ -114,20 +116,20 @@ mod tests { Ok(addr) } - fn make_checker(addr: SocketAddr, starts_healthy: bool) -> (HealthChecker, Vec>) { - let backends = vec![Arc::new(if starts_healthy { + fn make_checker(addr: SocketAddr, starts_healthy: bool) -> (HealthChecker, Arc) { + let backend = Arc::new(if starts_healthy { Backend::healthy(addr) } else { Backend::unhealthy(addr) - })]; + }); let checker = HealthChecker::new( - backends.clone(), + vec![Arc::clone(&backend)], String::from("/"), Duration::from_millis(20), ); - (checker, backends) + (checker, backend) } #[test] @@ -137,11 +139,11 @@ mod tests { let unreachable_addr = listener.local_addr()?; drop(listener); - let (checker, backends) = make_checker(unreachable_addr, true); + let (checker, backend) = make_checker(unreachable_addr, true); tokio::spawn(checker.run()); tokio::time::sleep(Duration::from_millis(200)).await; - assert!(!backends[0].is_healthy()); + assert!(!backend.is_healthy()); Ok(()) }) } @@ -150,12 +152,12 @@ mod tests { fn marks_backend_healthy_when_responding_with_2xx() -> Result<()> { tokio_test(async { let backend_addr = spawn_test_backend(StatusCode::OK).await?; - let (checker, backends) = make_checker(backend_addr, false); + let (checker, backend) = make_checker(backend_addr, false); tokio::spawn(checker.run()); tokio::time::sleep(Duration::from_millis(200)).await; - assert!(backends[0].is_healthy()); + assert!(backend.is_healthy()); Ok(()) }) } @@ -164,12 +166,12 @@ mod tests { fn marks_backend_unhealthy_when_responding_with_non_2xx() -> Result<()> { tokio_test(async { let backend_addr = spawn_test_backend(StatusCode::INTERNAL_SERVER_ERROR).await?; - let (checker, backends) = make_checker(backend_addr, true); + let (checker, backend) = make_checker(backend_addr, true); tokio::spawn(checker.run()); tokio::time::sleep(Duration::from_millis(200)).await; - assert!(!backends[0].is_healthy()); + assert!(!backend.is_healthy()); Ok(()) }) } diff --git a/src/load_balancer/least_connections.rs b/src/load_balancer/least_connections.rs index 080b78f..d488a20 100644 --- a/src/load_balancer/least_connections.rs +++ b/src/load_balancer/least_connections.rs @@ -41,7 +41,7 @@ impl LoadBalancer for LeastConnections { let &idx = tied.choose(&mut rand::rng())?; - Some(Arc::clone(&self.backends[idx]).acquire()) + Some(Arc::clone(self.backends.get(idx).as_ref()?).acquire()) } } @@ -54,7 +54,7 @@ mod tests { #[test] fn single_backend_always_returns_same_addr() -> Result<()> { let addr = localhost_addr(8001); - let lc = LeastConnections::init(make_healthy_backends(&[8001]))?; + let lc = LeastConnections::init(make_healthy_backends([8001]).to_vec())?; for _ in 0..3 { assert_eq!(lc.next().context("expected backend")?.addr(), addr); @@ -65,7 +65,7 @@ mod tests { #[test] fn empty_backends_returns_err() { - assert!(LeastConnections::init(make_healthy_backends(&[])).is_err()); + assert!(LeastConnections::init(make_healthy_backends([]).to_vec()).is_err()); } #[test] @@ -98,8 +98,8 @@ mod tests { #[test] fn routes_to_backend_with_fewest_connections() -> Result<()> { - let backends = make_healthy_backends(&[8000, 8001]); - let lc = LeastConnections::init(backends.clone())?; + let backends = make_healthy_backends([8000, 8001]); + let lc = LeastConnections::init(backends.to_vec())?; // Give 8000 two active connections so 8001 always wins let _c1 = Arc::clone(&backends[0]).acquire(); @@ -119,8 +119,8 @@ mod tests { #[test] fn increments_count_on_next_and_decrements_on_drop() -> Result<()> { - let backends = make_healthy_backends(&[8001]); - let lc = LeastConnections::init(backends.clone())?; + let backends = make_healthy_backends([8001]); + let lc = LeastConnections::init(backends.to_vec())?; assert_eq!(backends[0].num_connections(), 0); @@ -136,7 +136,7 @@ mod tests { #[test] fn chooses_valid_backend_when_tied() -> Result<()> { let (a, b) = (localhost_addr(8001), localhost_addr(8002)); - let lc = LeastConnections::init(make_healthy_backends(&[8001, 8002]))?; + let lc = LeastConnections::init(make_healthy_backends([8001, 8002]).to_vec())?; for _ in 0..10 { let addr = lc.next().context("expected backend")?.addr(); diff --git a/src/load_balancer/round_robin.rs b/src/load_balancer/round_robin.rs index f9c9d16..8dc20d4 100644 --- a/src/load_balancer/round_robin.rs +++ b/src/load_balancer/round_robin.rs @@ -26,19 +26,23 @@ impl RoundRobin { } impl LoadBalancer for RoundRobin { + #[expect( + clippy::arithmetic_side_effects, + reason = "Wrapping desired for round robin counter" + )] fn next(&self) -> Option { let n = self.backends.len(); let start = self.counter.fetch_add(1, Ordering::Relaxed) % n; let mut i = start; - while !self.backends[i].is_healthy() { + while !self.backends.get(i)?.is_healthy() { i = self.counter.fetch_add(1, Ordering::Relaxed) % n; if i == start { return None; } } - Some(Arc::clone(&self.backends[i]).acquire()) + Some(Arc::clone(self.backends.get(i).as_ref()?).acquire()) } } @@ -55,7 +59,7 @@ mod tests { localhost_addr(8003), ); - let rr = RoundRobin::init(make_healthy_backends(&[8001, 8002, 8003]))?; + let rr = RoundRobin::init(make_healthy_backends([8001, 8002, 8003]).to_vec())?; assert_eq!(rr.next().map(|g| g.addr()), Some(a1)); assert_eq!(rr.next().map(|g| g.addr()), Some(a2)); @@ -68,7 +72,7 @@ mod tests { #[test] fn single_backend_always_returns_same_addr() -> Result<()> { let addr = localhost_addr(8001); - let rr = RoundRobin::init(make_healthy_backends(&[8001]))?; + let rr = RoundRobin::init(make_healthy_backends([8001]).to_vec())?; assert_eq!(rr.next().map(|g| g.addr()), Some(addr)); assert_eq!(rr.next().map(|g| g.addr()), Some(addr)); @@ -79,7 +83,7 @@ mod tests { #[test] fn empty_backends_returns_err() { - assert!(RoundRobin::init(make_healthy_backends(&[])).is_err()); + assert!(RoundRobin::init(make_healthy_backends([]).to_vec()).is_err()); } #[test] diff --git a/src/proxy.rs b/src/proxy.rs index ac70b61..76f3b1a 100644 --- a/src/proxy.rs +++ b/src/proxy.rs @@ -16,6 +16,10 @@ use tokio::net::TcpStream; use tracing::{error, warn}; /// A generic `Response` containing a boxed `Body` (may be `Incoming`, `Empty`, etc.). +#[expect( + unused_qualifications, + reason = "Absolute paths for clarity in type alias" +)] type BoxBodyResp = hyper::Response>; const X_FORWARDED_FOR: HeaderName = HeaderName::from_static("x-forwarded-for"); @@ -129,9 +133,10 @@ fn prepare_response(mut resp: Response) -> Response { /// Removes the standard hop-by-hop headers and any custom headers named in the Connection header. fn strip_hop_by_hop_headers(headers: &mut HeaderMap) { let extra = get_str_val(headers, &CONNECTION) - .map(|s| { - s.split(',') - .map(|s| s.trim().to_lowercase()) + .map(|connection_val| { + connection_val + .split(',') + .map(|header_name| header_name.trim().to_lowercase()) .collect::>() }) .unwrap_or_default(); diff --git a/src/server.rs b/src/server.rs index 3401259..079987f 100644 --- a/src/server.rs +++ b/src/server.rs @@ -72,7 +72,7 @@ pub async fn run( join_set.spawn(async move { let conn = server_http1::Builder::new().serve_connection( TokioIo::new(stream), - service_fn(move |req| { + service_fn( |req| { proxy::forward(req, client_addr, Arc::clone(&conn_routes)) }), ); @@ -142,15 +142,10 @@ mod tests { use http_body_util::{Empty, Full}; use hyper::{ Request, Response, StatusCode, body::Bytes, client::conn::http1 as client_http1, - server::conn::http1 as server_http1, service::service_fn, + server::conn::http1 as server_http1, }; - use hyper_util::rt::TokioIo; use std::{convert::Infallible, net::SocketAddr, time::Duration}; - use tokio::{ - net::{TcpListener, TcpStream}, - sync::oneshot, - time::Instant, - }; + use tokio::{net::TcpStream, sync::oneshot, time::Instant}; fn new_test_config(backend_addr: SocketAddr, shutdown_timeout: Duration) -> Config { Config { @@ -178,15 +173,18 @@ mod tests { let Ok((stream, _)) = listener.accept().await else { break }; tokio::spawn(async move { - let _ = server_http1::Builder::new() - .serve_connection( - TokioIo::new(stream), - service_fn(|_| async move { - tokio::time::sleep(response_delay).await; - Ok::<_, Infallible>(Response::new(Full::new(Bytes::new()))) - }), - ) - .await; + assert!( + server_http1::Builder::new() + .serve_connection( + TokioIo::new(stream), + service_fn(async |_| { + tokio::time::sleep(response_delay).await; + Ok::<_, Infallible>(Response::new(Full::new(Bytes::new()))) + }), + ) + .await + .is_ok() + ); }); } }); @@ -199,9 +197,7 @@ mod tests { let io = TokioIo::new(TcpStream::connect(proxy_addr).await?); let (mut sender, conn) = client_http1::handshake(io).await?; - tokio::spawn(async move { - let _ = conn.await; - }); + tokio::spawn(async { assert!(conn.await.is_ok()) }); let req = Request::builder() .uri(format!("http://{proxy_addr}/")) @@ -221,8 +217,8 @@ mod tests { let (shutdown_tx, shutdown_rx) = oneshot::channel::<()>(); let config = new_test_config(backend_addr, Duration::from_secs(2)); - let proxy = tokio::spawn(run(config, proxy_listener, async move { - let _ = shutdown_rx.await; + let proxy = tokio::spawn(run(config, proxy_listener, async { + assert!(shutdown_rx.await.is_ok()); })); let request = tokio::spawn(send_request(proxy_addr)); @@ -231,7 +227,7 @@ mod tests { tokio::time::sleep(Duration::from_millis(50)).await; // Trigger shutdown while the backend is still processing (it sleeps 100ms total) - let _ = shutdown_tx.send(()); + assert!(shutdown_tx.send(()).is_ok()); // The in-flight request should still complete successfully assert_eq!(request.await??, StatusCode::OK); @@ -251,8 +247,8 @@ mod tests { let (shutdown_tx, shutdown_rx) = oneshot::channel::<()>(); let config = new_test_config(backend_addr, Duration::from_millis(100)); - let proxy = tokio::spawn(run(config, proxy_listener, async move { - let _ = shutdown_rx.await; + let proxy = tokio::spawn(run(config, proxy_listener, async { + assert!(shutdown_rx.await.is_ok()); })); // Start a request that will be held at the slow backend @@ -262,7 +258,7 @@ mod tests { tokio::time::sleep(Duration::from_millis(50)).await; let start = Instant::now(); - let _ = shutdown_tx.send(()); + assert!(shutdown_tx.send(()).is_ok()); proxy.await??; let elapsed = start.elapsed(); @@ -289,8 +285,8 @@ mod tests { let (shutdown_tx, shutdown_rx) = oneshot::channel::<()>(); let config = new_test_config(backend_addr, Duration::from_secs(1)); - let proxy = tokio::spawn(run(config, proxy_listener, async move { - let _ = shutdown_rx.await; + let proxy = tokio::spawn(run(config, proxy_listener, async { + assert!(shutdown_rx.await.is_ok()); })); // Wait for the health checker to mark the backend unhealthy @@ -301,7 +297,7 @@ mod tests { StatusCode::SERVICE_UNAVAILABLE ); - let _ = shutdown_tx.send(()); + assert!(shutdown_tx.send(()).is_ok()); proxy.await??; Ok(()) diff --git a/src/test_utils.rs b/src/test_utils.rs index de21b78..086102a 100644 --- a/src/test_utils.rs +++ b/src/test_utils.rs @@ -23,9 +23,6 @@ pub const fn localhost_addr(port: u16) -> SocketAddr { } /// Creates healthy localhost backends from port numbers. -pub fn make_healthy_backends(ports: &[u16]) -> Vec> { - ports - .iter() - .map(|&p| Arc::new(Backend::healthy(localhost_addr(p)))) - .collect() +pub fn make_healthy_backends(ports: [u16; N]) -> [Arc; N] { + ports.map(|p| Arc::new(Backend::healthy(localhost_addr(p)))) }