diff --git a/README.md b/README.md index 5bfeba5..436cdbb 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, 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/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..aa405a1 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,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 { 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), } }) } diff --git a/crates/server/tests/units.rs b/crates/server/tests/units.rs index f4cfab6..842e766 100644 --- a/crates/server/tests/units.rs +++ b/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" + ); + } +}