fix(api): raise the body limit on the push routes, and only those

README.md

@@ -209,6 +209,7 @@ with no broker configured, which is green lines that verified nothing. CI sets
 | `GATE_MAX_PARK_MS` | 30000 | how long a handler holds its claim waiting for a window before releasing |
 | `GATE_INTERIOR_SEED_SKEW_SECONDS` | 120 | how far before a graph's start a new group on an **interior** queue is seeded; a margin for Gate's clock against the broker's, capped at 600 |
 | `GATE_RECONCILE_SECONDS` | 15 | how often a replica re-reads the store |
+| `GATE_MAX_PUSH_BODY_BYTES` | 8388608 | the largest body a **push** route buffers, clamped to 2 MiB–64 MiB. 2 MiB is axum's default, which is what applied to everything until 2026-09-04 because nothing set one; the ceiling is there because the limit is a per-request memory reservation and nothing bounds how many requests hold one at once. Document routes keep the default |
 
 **Where a new consumer group starts, and it is two rules.** On an **ingress** queue — yours, or
 Gate's own HTTP front door — a new group is seeded at the *head* of the retained log, because a

crates/server/src/api/mod.rs

@@ -23,7 +23,7 @@ pub mod reenter;
 use std::sync::Arc;
 use std::time::Instant;
 
-use axum::extract::State;
+use axum::extract::{DefaultBodyLimit, State};
 use axum::http::StatusCode;
 use axum::response::{IntoResponse, Response};
 use axum::routing::{get, post, put};
@@ -137,6 +137,21 @@ pub fn public_router(app: Shared) -> Router {
         .with_state(app)
 }
 
+/// The body limit the four push routes carry, and only they.
+///
+/// A push is a BATCH: one request stands for as many items as the caller managed
+/// to group, and the honest bound on it is memory. Everything else on this
+/// surface takes a document — a graph declaration, a breaker poke, a console
+/// read — and keeps axum's 2 MiB, which no document has ever come near.
+///
+/// Applied per route rather than as one `.layer()` on the Router for exactly
+/// that reason: a limit on the whole surface would raise the ceiling on
+/// endpoints that have no batch to justify it, and this service holds every
+/// buffered body in a 512 MiB pod.
+fn push_body_limit() -> DefaultBodyLimit {
+    DefaultBodyLimit::max(crate::knobs::knobs().max_push_body)
+}
+
 fn routes() -> Router<Shared> {
     Router::new()
         // ---- graphs: the one document type.
@@ -154,11 +169,11 @@ fn routes() -> Router<Shared> {
         )
         .route(
             "/v1/apps/:app/graphs/:graph/nodes/:node/push",
-            post(data::graph_push),
+            post(data::graph_push).layer(push_body_limit()),
         )
         .route(
             "/v1/graphs/:graph/nodes/:node/push",
-            post(data::graph_push_default),
+            post(data::graph_push_default).layer(push_body_limit()),
         )
         .route(
             "/v1/apps/:app/graphs/:graph/nodes/:node/eta",
@@ -206,11 +221,11 @@ fn routes() -> Router<Shared> {
         )
         .route(
             "/v1/apps/:app/targets/:name/lanes/:lane/push",
-            post(data::target_push),
+            post(data::target_push).layer(push_body_limit()),
         )
         .route(
             "/v1/targets/:name/lanes/:lane/push",
-            post(data::target_push_default),
+            post(data::target_push_default).layer(push_body_limit()),
         )
         .route("/v1/apps/:app/targets/:name/eta", get(eta::target_eta))
         .route("/v1/targets/:name/eta", get(eta::target_eta_default))

crates/server/src/knobs.rs

@@ -122,6 +122,30 @@ pub struct Knobs {
     /// (`004_log_pop.sql`), so an explicit failed ack is reserved for real
     /// poison and a retry budget means what it says.
     pub retry_limit: i32,
+    /// The largest body a PUSH route will buffer, in bytes.
+    ///
+    /// axum's own default is 2 MiB and nothing here ever overrode it, so that
+    /// number was the real ceiling on everything a caller can hand this service
+    /// — silently, because the refusal it produces says
+    /// `Failed to buffer the request body: length limit exceeded` and names
+    /// neither the limit nor the fact that it is ours.
+    ///
+    /// It was measured from prod on 2026-09-04, from both sides of the wall, by
+    /// a caller that had spent a week failing against it: pushes of 11,408 /
+    /// 10,387 / 8,976 records went through, and pushes of 12,000 and 16,096 did
+    /// not. Divide, and 2 MiB is exactly where those cross — the payloads run
+    /// about 130 to 175 bytes a record depending on the vendor.
+    ///
+    /// 8 MiB, and the ceiling on the ceiling is memory rather than taste: the
+    /// service runs with a 512 MiB limit, and a body limit is a per-request
+    /// buffer. Four times the old value keeps a large caller comfortably inside
+    /// it while a burst of ten concurrent pushes still costs under a sixth of
+    /// the pod.
+    ///
+    /// PUSH ROUTES ONLY. Declaring a graph or reading the console has no reason
+    /// to accept megabytes, and axum's default is the right answer everywhere
+    /// the body is a document rather than a batch.
+    pub max_push_body: usize,
 }
 
 impl Default for Knobs {
@@ -140,10 +164,28 @@ impl Default for Knobs {
             max_prefix_retries: 2,
             interior_seed_skew: crate::relay::INTERIOR_SEED_SKEW,
             retry_limit: 3,
+            max_push_body: 8 * 1024 * 1024,
         }
     }
 }
 
+/// axum's own `DefaultBodyLimit`, which applied to every route here until
+/// 2026-09-04 because nothing set one. Named so the floor below says why it is
+/// where it is.
+pub const AXUM_DEFAULT_BODY_LIMIT: usize = 2 * 1024 * 1024;
+
+/// The largest a push body may be configured to be.
+///
+/// The limit is a per-request memory reservation, not a per-request cost: the
+/// buffered bytes, the `serde_json::Value` they parse into, the copy the
+/// envelope is built on and the body sent to the broker are all live at once,
+/// and there is no concurrency limiter in front of any of it. A typo in a
+/// deployment manifest should not be able to ask one pod to hold gigabytes.
+///
+/// 64 MiB is eight times the default and far past any real push; a deployment
+/// that wants more than this wants a different shape, not a bigger number.
+pub const MAX_PUSH_BODY_CEILING: usize = 64 * 1024 * 1024;
+
 fn env_u32(name: &str) -> Option<u32> {
     std::env::var(name).ok().and_then(|v| v.parse().ok())
 }
@@ -182,6 +224,12 @@ pub fn knobs() -> &'static Knobs {
             retry_limit: env_u32("GATE_RETRY_LIMIT")
                 .map(|n| n as i32)
                 .unwrap_or(d.retry_limit),
+            // Floored at axum's own default rather than at zero: a typo in the
+            // environment must not be able to make this service refuse bodies it
+            // accepted before anybody set the variable.
+            max_push_body: env_u32("GATE_MAX_PUSH_BODY_BYTES")
+                .map(|n| (n as usize).clamp(AXUM_DEFAULT_BODY_LIMIT, MAX_PUSH_BODY_CEILING))
+                .unwrap_or(d.max_push_body),
         }
     })
 }

crates/server/tests/units.rs

@@ -334,6 +334,17 @@ fn the_knobs_default_to_what_the_design_says() {
         k.retry_limit, 3,
         "the DLQ is back: v1 had to disarm it because it paced by nacking"
     );
+    assert!(
+        k.max_push_body <= gate_server::knobs::MAX_PUSH_BODY_CEILING,
+        "the knob is floored at axum's default and capped: a per-request buffer is a memory \
+         reservation, and nothing limits how many requests hold one at once"
+    );
+    assert_eq!(
+        k.max_push_body,
+        8 * 1024 * 1024,
+        "four times axum's 2 MiB default, which was the real ceiling on every push \
+         until 2026-09-04 because nothing here ever set one"
+    );
 }
 
 /// `forwarded / commits` is THE number that explains a stage's throughput: the
@@ -378,3 +389,101 @@ fn the_trace_ring_drops_the_oldest_and_never_grows() {
     );
     assert!(t.recent(Some("admitted"), 10).is_empty(), "denials only");
 }
+
+// ----------------------------------------------------------- the body limit
+
+/// A push carries a BATCH and everything else carries a document, so only the
+/// push routes get the raised body limit.
+///
+/// The limit is asserted through the ROUTER rather than by reading the knob back,
+/// because the knob was never the thing that was wrong: axum applies a 2 MiB
+/// default to every route unless a layer says otherwise, and for the life of this
+/// service nothing did. A test that reads `knobs().max_push_body` would have
+/// passed just as happily on the day a caller was being refused.
+///
+/// No broker is needed and none is reached: a body over the limit is rejected by
+/// the extractor, so the handler never runs.
+mod body_limit {
+    use axum::body::Body;
+    use axum::http::{Request, StatusCode};
+    use std::sync::Arc;
+    use tower::ServiceExt;
+
+    fn app() -> gate_server::api::Shared {
+        // Deliberately unreachable: nothing in these cases gets far enough to
+        // speak to it, and a test that needed a broker would belong in live.rs.
+        let queen = queen_mq::Queen::connect(queen_mq::Config::new("http://127.0.0.1:1"))
+            .expect("the client is constructed, not connected");
+        Arc::new(gate_server::api::App::new(
+            queen,
+            "http://127.0.0.1:1".into(),
+        ))
+    }
+
+    fn body_of(bytes: usize) -> Body {
+        Body::from(vec![b'x'; bytes])
+    }
+
+    const MIB: usize = 1024 * 1024;
+
+    #[tokio::test]
+    async fn a_push_accepts_a_body_axums_default_would_have_refused() {
+        let res = gate_server::api::router(app())
+            .oneshot(
+                Request::post("/v1/apps/channel-go/graphs/google/nodes/hotel/push")
+                    .header("content-type", "application/json")
+                    .body(body_of(3 * MIB))
+                    .unwrap(),
+            )
+            .await
+            .unwrap();
+        assert_ne!(
+            res.status(),
+            StatusCode::PAYLOAD_TOO_LARGE,
+            "a 3 MiB push was refused for its size: this is the 2 MiB default that \
+             cost a caller a week of undelivered pushes, and raising it is the point"
+        );
+    }
+
+    #[tokio::test]
+    async fn a_push_over_the_ceiling_is_still_refused() {
+        let over = gate_server::knobs::knobs().max_push_body + MIB;
+        let res = gate_server::api::router(app())
+            .oneshot(
+                Request::post("/v1/apps/channel-go/graphs/google/nodes/hotel/push")
+                    .header("content-type", "application/json")
+                    .body(body_of(over))
+                    .unwrap(),
+            )
+            .await
+            .unwrap();
+        assert_eq!(
+            res.status(),
+            StatusCode::PAYLOAD_TOO_LARGE,
+            "the ceiling is a ceiling: a body past it must be refused, or the limit \
+             is memory nobody is bounding"
+        );
+    }
+
+    #[tokio::test]
+    async fn a_document_route_keeps_the_default() {
+        // Declaring a graph is a document, not a batch. Raising its ceiling would
+        // buy nothing and would let a caller hand a 512 MiB pod a body per
+        // request that no declaration has ever needed.
+        let res = gate_server::api::router(app())
+            .oneshot(
+                Request::put("/v1/apps/channel-go/graphs/google")
+                    .header("content-type", "application/json")
+                    .body(body_of(3 * MIB))
+                    .unwrap(),
+            )
+            .await
+            .unwrap();
+        assert_eq!(
+            res.status(),
+            StatusCode::PAYLOAD_TOO_LARGE,
+            "a document route accepted 3 MiB: the raise must be scoped to the push \
+             routes, not applied to the whole surface"
+        );
+    }
+}