From 7ba851a3c7cd115791e8c63d191f39ed58cccb18 Mon Sep 17 00:00:00 2001 From: AlbertoV Date: Fri, 4 Sep 2026 18:38:34 +0200 Subject: [PATCH 1/2] fix(api): raise the body limit on the push routes, and only those MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit axum applies a 2 MiB `DefaultBodyLimit` to every route unless a layer says otherwise, and nothing here ever did. That default was therefore the real ceiling on everything a caller can hand this service, and it was invisible: the refusal reads `Failed to buffer the request body: length limit exceeded`, which names neither the limit nor the fact that it belongs to us and not to the vendor behind the graph. A caller spent a week failing against it. Measured from their side on 2026-09-04, from both sides of the wall: 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 — their payloads run about 130 to 175 bytes a record depending on the vendor. Worse than the refusal was what it cost them: a size refusal is deterministic, so their retry budget was spent re-sending identical bytes and the work was then parked, silently, for 2,683 coordinates. 8 MiB, as GATE_MAX_PUSH_BODY_BYTES. The ceiling on the ceiling is memory rather than taste: a body limit is a per-request buffer and this service runs with a 512 MiB limit, so 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. The env override is floored at axum's own default so that a typo cannot make the service refuse bodies it accepted before anybody set the variable. PUSH ROUTES ONLY, applied per route rather than as one layer on the Router. A push is a batch — one request stands for as many items as the caller grouped — and the honest bound on it is memory. A graph declaration, a breaker poke and a console read are documents, none has ever come near 2 MiB, and raising their ceiling would buy nothing while letting anybody hand a 512 MiB pod a body per request. The tests go through the ROUTER rather than reading the knob back, because the knob was never the thing that was wrong: a test asserting `knobs().max_push_body` would have passed just as happily on the day a caller was being refused. Mutation-checked: dropping the layer from a push route fails the case that a 3 MiB push is accepted, and moving it to the whole Router fails the case that a document route still refuses one. No broker is needed and none is reached — the extractor rejects before the handler runs. Claude-Session: https://claude.ai/code/session_01RJHziF1EPc5fH7Kd7WRdgk --- README.md | 1 + crates/server/src/api/mod.rs | 25 +++++++-- crates/server/src/knobs.rs | 36 ++++++++++++ crates/server/tests/units.rs | 104 +++++++++++++++++++++++++++++++++++ 4 files changed, 161 insertions(+), 5 deletions(-) diff --git a/README.md b/README.md index 5bfeba5..5beaf5a 100644 --- a/README.md +++ b/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. Floored at axum's 2 MiB default, which is what applied to everything until 2026-09-04 because nothing set one. Document routes keep the default: a push is a batch, a declaration is not | **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 diff --git a/crates/server/src/api/mod.rs b/crates/server/src/api/mod.rs index b9934c4..f51cc62 100644 --- a/crates/server/src/api/mod.rs +++ b/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 { Router::new() // ---- graphs: the one document type. @@ -154,11 +169,11 @@ fn routes() -> Router { ) .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 { ) .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)) diff --git a/crates/server/src/knobs.rs b/crates/server/src/knobs.rs index 106fdf8..79e50e6 100644 --- a/crates/server/src/knobs.rs +++ b/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,16 @@ 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; + fn env_u32(name: &str) -> Option { std::env::var(name).ok().and_then(|v| v.parse().ok()) } @@ -182,6 +212,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).max(AXUM_DEFAULT_BODY_LIMIT)) + .unwrap_or(d.max_push_body), } }) } diff --git a/crates/server/tests/units.rs b/crates/server/tests/units.rs index f4cfab6..e5697ef 100644 --- a/crates/server/tests/units.rs +++ b/crates/server/tests/units.rs @@ -334,6 +334,12 @@ 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_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 +384,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" + ); + } +} From d8e56457de89c24493b6e5d72349ab5f7ea5cf01 Mon Sep 17 00:00:00 2001 From: AlbertoV Date: Sat, 5 Sep 2026 18:02:49 +0200 Subject: [PATCH 2/2] fix(api): cap the push body knob MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `env_u32` accepts anything up to `u32::MAX`, so `GATE_MAX_PUSH_BODY_BYTES` could ask each push to buffer about 4 GiB. The value is floored at axum's default and now capped as well. The floor was reasoned about and the ceiling was not, and the ceiling is the side that matters: the limit is a per-request memory RESERVATION, and the wire body is only one of four copies alive at once — the buffered bytes, the `serde_json::Value` they parse into, the copy the envelope is built on and the body sent to the broker. Nothing in front of the router limits concurrency, so the multiplier is the number of callers, not a constant. Claude-Session: https://claude.ai/code/session_012K8u7BEJyd6nDNMCQAgH3z --- README.md | 2 +- crates/server/src/knobs.rs | 14 +++++++++++++- crates/server/tests/units.rs | 5 +++++ 3 files changed, 19 insertions(+), 2 deletions(-) diff --git a/README.md b/README.md index 5beaf5a..436cdbb 100644 --- a/README.md +++ b/README.md @@ -209,7 +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. Floored at axum's 2 MiB default, which is what applied to everything until 2026-09-04 because nothing set one. Document routes keep the default: a push is a batch, a declaration is not | +| `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 diff --git a/crates/server/src/knobs.rs b/crates/server/src/knobs.rs index 79e50e6..aa405a1 100644 --- a/crates/server/src/knobs.rs +++ b/crates/server/src/knobs.rs @@ -174,6 +174,18 @@ impl Default for Knobs { /// 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 { std::env::var(name).ok().and_then(|v| v.parse().ok()) } @@ -216,7 +228,7 @@ pub fn knobs() -> &'static Knobs { // 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).max(AXUM_DEFAULT_BODY_LIMIT)) + .map(|n| (n as usize).clamp(AXUM_DEFAULT_BODY_LIMIT, MAX_PUSH_BODY_CEILING)) .unwrap_or(d.max_push_body), } }) diff --git a/crates/server/tests/units.rs b/crates/server/tests/units.rs index e5697ef..842e766 100644 --- a/crates/server/tests/units.rs +++ b/crates/server/tests/units.rs @@ -334,6 +334,11 @@ 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,