Enforce global request limits

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"

crates/walgit-server/src/lib.rs

@@ -86,6 +86,8 @@ pub struct AppState {
     pub registry: Arc<walgit_wal::Registry>,
     pub auth: Arc<auth::Authenticator>,
     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<middleware::Inflight>,
     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<AppState>) -> 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(

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<Semaphore>,
+    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<Box<tokio::time::Sleep>>,
+}
+
+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<Option<Result<http_body::Frame<Self::Data>, 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<RequestLimits>,
+    req: Request<Body>,
+    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