From beb9e291647e4b3674e1ce58bed4230737949ff5 Mon Sep 17 00:00:00 2001 From: Preetam Dwivedi Date: Wed, 30 Sep 2026 11:08:09 -0700 Subject: [PATCH] Enforce global request limits MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Summary ### Why? The server accepted global concurrency and timeout settings without applying them, allowing unbounded work and requests that could outlive their configured deadline. ### What? Add shared request-limit middleware that bounds queueing and active work, retains permits for streaming bodies, and applies one deadline across queue time, handler execution, and response delivery. Reject zero-valued global limits during configuration validation. ## Test Plan ✅ `cargo test -p walgit-config` ✅ `cargo fmt --all -- --check` ✅ `git diff --check` ⚠️ `cargo check -p walgit-server` reached the protobuf build script but could not run because `protoc` is unavailable in this environment. ## Issue Closes #78 --- crates/walgit-config/src/lib.rs | 8 +++ crates/walgit-server/src/lib.rs | 12 ++++ crates/walgit-server/src/middleware.rs | 92 ++++++++++++++++++++++++-- 3 files changed, 107 insertions(+), 5 deletions(-) diff --git a/crates/walgit-config/src/lib.rs b/crates/walgit-config/src/lib.rs index 80f0ec86..b5aca125 100644 --- a/crates/walgit-config/src/lib.rs +++ b/crates/walgit-config/src/lib.rs @@ -1077,6 +1077,14 @@ impl Config { pub fn validate(&self) -> Result<()> { self.refs.validate()?; self.packs.validate()?; + anyhow::ensure!( + self.server.max_concurrent_requests > 0, + "server.max_concurrent_requests must be positive" + ); + anyhow::ensure!( + !self.server.request_timeout.is_zero(), + "server.request_timeout must be positive" + ); anyhow::ensure!( self.packfile_uri.uri_min_bytes.as_u64() > 0, "packfile_uri.uri_min_bytes must be positive" diff --git a/crates/walgit-server/src/lib.rs b/crates/walgit-server/src/lib.rs index 5f6c6154..905211ba 100644 --- a/crates/walgit-server/src/lib.rs +++ b/crates/walgit-server/src/lib.rs @@ -86,6 +86,8 @@ pub struct AppState { pub registry: Arc, pub auth: Arc, pub semaphores: middleware::RepoSemaphores, + /// Global request concurrency and deadline enforcement. + pub request_limits: middleware::RequestLimits, /// HTTP requests in flight (counted until the response body is done); on the watchdog line. pub inflight: Arc, pub caches: cache::ServerCaches, @@ -123,6 +125,10 @@ impl AppState { registry, auth: auth::Authenticator::new(&cfg), semaphores: middleware::RepoSemaphores::new(cfg.server.max_concurrent_per_repo), + request_limits: middleware::RequestLimits::new( + cfg.server.max_concurrent_requests, + cfg.server.request_timeout, + ), inflight: Arc::new(middleware::Inflight::default()), caches: cache::ServerCaches::new(&cfg), metrics_handle, @@ -241,6 +247,12 @@ pub fn router(state: Arc) -> Router { state.clone(), web::canonical_browser_host, )) + // Bound queued + running requests. The permit and deadline remain attached + // to streaming response bodies, not merely to the handler future. + .layer(axum::middleware::from_fn_with_state( + state.request_limits.clone(), + middleware::request_limits, + )) // Outermost: `http.request` span (request_id, trace, principal, repo, // status, elapsed) — every log line inside inherits its fields. .layer(axum::middleware::from_fn_with_state( diff --git a/crates/walgit-server/src/middleware.rs b/crates/walgit-server/src/middleware.rs index 78de0d76..5c6f21b3 100644 --- a/crates/walgit-server/src/middleware.rs +++ b/crates/walgit-server/src/middleware.rs @@ -1,16 +1,22 @@ -//! HTTP middleware: request id + tracing span. Request timeout, body limit and -//! tracing layers are applied in [`crate::router`] via `tower-http`. Per-repo -//! concurrency limiting lives in the handlers (they hold a repo-keyed semaphore -//! for the duration of the git operation). +//! HTTP middleware: request id + tracing span, global request limits, and +//! per-repository concurrency limits. +use std::future::Future; +use std::pin::Pin; use std::sync::Arc; use std::sync::atomic::{AtomicU64, Ordering}; +use std::task::{Context, Poll}; +use std::time::Duration; +use axum::body::Body; +use axum::extract::State; +use axum::http::StatusCode; use axum::http::{HeaderValue, Request}; use axum::middleware::Next; +use axum::response::IntoResponse; use axum::response::Response; use dashmap::DashMap; -use tokio::sync::Semaphore; +use tokio::sync::{OwnedSemaphorePermit, Semaphore}; use tracing::Instrument; use uuid::Uuid; @@ -70,6 +76,82 @@ impl http_body::Body for CountedBody { } } +/// Global request limits shared by every clone of the router. +#[derive(Clone)] +pub struct RequestLimits { + permits: Arc, + timeout: Duration, +} + +impl RequestLimits { + pub fn new(max_concurrent: usize, timeout: Duration) -> Self { + Self { + permits: Arc::new(Semaphore::new(max_concurrent)), + timeout, + } + } +} + +/// Response body that holds its concurrency permit and enforces the request's +/// original deadline for the full lifetime of a streaming response. +struct LimitedBody { + inner: Body, + _permit: OwnedSemaphorePermit, + deadline: Pin>, +} + +impl http_body::Body for LimitedBody { + type Data = bytes::Bytes; + type Error = axum::Error; + + fn poll_frame( + mut self: Pin<&mut Self>, + cx: &mut Context<'_>, + ) -> Poll, Self::Error>>> { + if self.deadline.as_mut().poll(cx).is_ready() { + return Poll::Ready(Some(Err(axum::Error::new(std::io::Error::new( + std::io::ErrorKind::TimedOut, + "request deadline exceeded", + ))))); + } + Pin::new(&mut self.inner).poll_frame(cx) + } + + fn is_end_stream(&self) -> bool { + self.inner.is_end_stream() + } + + fn size_hint(&self) -> http_body::SizeHint { + self.inner.size_hint() + } +} + +/// Apply the configured global concurrency cap and one deadline spanning queue +/// time, handler execution, and streaming response delivery. +pub async fn request_limits( + State(limits): State, + req: Request, + next: Next, +) -> Response { + let deadline = tokio::time::Instant::now() + limits.timeout; + let permit = match tokio::time::timeout_at(deadline, limits.permits.acquire_owned()).await { + Ok(Ok(permit)) => permit, + Ok(Err(_)) => unreachable!("global request semaphore is never closed"), + Err(_) => return StatusCode::REQUEST_TIMEOUT.into_response(), + }; + let response = match tokio::time::timeout_at(deadline, next.run(req)).await { + Ok(response) => response, + Err(_) => return StatusCode::REQUEST_TIMEOUT.into_response(), + }; + response.map(|body| { + Body::new(LimitedBody { + inner: body, + _permit: permit, + deadline: Box::pin(tokio::time::sleep_until(deadline)), + }) + }) +} + /// Generate a request id, attach it to the `http.request` span + response /// header. The span captures method, path, and trace context from /// `X-Cloud-Trace-Context` / `traceparent` headers for Cloud Logging