From 7ba851a3c7cd115791e8c63d191f39ed58cccb18 Mon Sep 17 00:00:00 2001 From: AlbertoV Date: Fri, 4 Sep 2026 18:38:34 +0200 Subject: [PATCH 01/85] 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 a4faf852797d6895066ed90401b2f44fdb210035 Mon Sep 17 00:00:00 2001 From: AlbertoV Date: Fri, 4 Sep 2026 21:59:03 +0200 Subject: [PATCH 02/85] chore: standardize JavaScript runtime on Node 24 --- .github/workflows/test.yml | 2 +- .nvmrc | 1 + Dockerfile | 2 +- README.md | 3 ++ ui/package-lock.json | 75 +++++++++++++++++++++++++++++++++++--- ui/package.json | 3 ++ 6 files changed, 78 insertions(+), 8 deletions(-) create mode 100644 .nvmrc diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index 70ff2b8..e61c621 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -74,7 +74,7 @@ jobs: - uses: actions/setup-node@v4 with: - node-version: 22 + node-version: 24 cache: npm cache-dependency-path: ui/package-lock.json diff --git a/.nvmrc b/.nvmrc new file mode 100644 index 0000000..a45fd52 --- /dev/null +++ b/.nvmrc @@ -0,0 +1 @@ +24 diff --git a/Dockerfile b/Dockerfile index 6c03035..8849f79 100644 --- a/Dockerfile +++ b/Dockerfile @@ -9,7 +9,7 @@ # Run: docker run -p 8788:8788 -e QUEEN_URL=http://queen:6632 gate # ---------------------------------------------------------------- the console -FROM node:22-alpine AS ui-builder +FROM node:24-alpine AS ui-builder WORKDIR /app/ui COPY ui/package*.json ./ diff --git a/README.md b/README.md index 5bfeba5..914522d 100644 --- a/README.md +++ b/README.md @@ -183,7 +183,10 @@ reaches it — and `GATE_PUBLIC_BIND` requires a Google session on every route. the local sign-in bypass and Gate refuses to boot with it set on an `https` public URL; `GATE_ADMIN_EMAILS` is what makes that identity able to write rather than only read. +Building from source requires Node.js 24 for the embedded console; the root `.nvmrc` selects it. + ```bash +nvm use # Node.js 24, from the root .nvmrc cd ui && npm ci && npm run build && cd .. # the console is compiled into the binary cargo build --release --workspace cargo test --workspace # the live suite reports as ignored diff --git a/ui/package-lock.json b/ui/package-lock.json index 9a9b3cd..69748dc 100644 --- a/ui/package-lock.json +++ b/ui/package-lock.json @@ -18,6 +18,9 @@ "@vitejs/plugin-vue": "^5.2.1", "tailwindcss": "^4.3.3", "vite": "^6.0.7" + }, + "engines": { + "node": ">=24.0.0" } }, "node_modules/@babel/helper-string-parser": { @@ -1165,6 +1168,72 @@ "node": ">=14.0.0" } }, + "node_modules/@tailwindcss/oxide-wasm32-wasi/node_modules/@emnapi/core": { + "version": "1.11.1", + "dev": true, + "inBundle": true, + "license": "MIT", + "optional": true, + "dependencies": { + "@emnapi/wasi-threads": "1.2.2", + "tslib": "^2.4.0" + } + }, + "node_modules/@tailwindcss/oxide-wasm32-wasi/node_modules/@emnapi/runtime": { + "version": "1.11.1", + "dev": true, + "inBundle": true, + "license": "MIT", + "optional": true, + "dependencies": { + "tslib": "^2.4.0" + } + }, + "node_modules/@tailwindcss/oxide-wasm32-wasi/node_modules/@emnapi/wasi-threads": { + "version": "1.2.2", + "dev": true, + "inBundle": true, + "license": "MIT", + "optional": true, + "dependencies": { + "tslib": "^2.4.0" + } + }, + "node_modules/@tailwindcss/oxide-wasm32-wasi/node_modules/@napi-rs/wasm-runtime": { + "version": "1.1.4", + "dev": true, + "inBundle": true, + "license": "MIT", + "optional": true, + "dependencies": { + "@tybys/wasm-util": "^0.10.1" + }, + "funding": { + "type": "github", + "url": "https://github.com/sponsors/Brooooooklyn" + }, + "peerDependencies": { + "@emnapi/core": "^1.7.1", + "@emnapi/runtime": "^1.7.1" + } + }, + "node_modules/@tailwindcss/oxide-wasm32-wasi/node_modules/@tybys/wasm-util": { + "version": "0.10.2", + "dev": true, + "inBundle": true, + "license": "MIT", + "optional": true, + "dependencies": { + "tslib": "^2.4.0" + } + }, + "node_modules/@tailwindcss/oxide-wasm32-wasi/node_modules/tslib": { + "version": "2.8.1", + "dev": true, + "inBundle": true, + "license": "0BSD", + "optional": true + }, "node_modules/@tailwindcss/oxide-win32-arm64-msvc": { "version": "4.3.3", "resolved": "https://registry.npmjs.org/@tailwindcss/oxide-win32-arm64-msvc/-/oxide-win32-arm64-msvc-4.3.3.tgz", @@ -1457,7 +1526,6 @@ "integrity": "sha512-AC/7JofJvZGrrneWNaEnJeOLUx+JlGt7tNa0wZiRPT4MY1wmfKjt2+6O2p2uz2+skll8OZZmJMNqeke7kKbNgQ==", "dev": true, "license": "MIT", - "peer": true, "bin": { "jiti": "lib/jiti-cli.mjs" } @@ -1468,7 +1536,6 @@ "integrity": "sha512-NXYBzinNrblfraPGyrbPoD19C1h9lfI/1mzgWYvXUTe414Gz/X1FD2XBZSZM7rRTrMA8JL3OtAaGifrIKhQ5yQ==", "dev": true, "license": "MPL-2.0", - "peer": true, "dependencies": { "detect-libc": "^2.0.3" }, @@ -1902,7 +1969,6 @@ "integrity": "sha512-RvwwcruNjI1ncT5xRakeyS9Lf8lcItv34KD+aif+VH9kduAyfYBipGh12274xtenIPZ119/R9BdTBa8gAwSh0A==", "dev": true, "license": "MIT", - "peer": true, "engines": { "node": ">=12" }, @@ -1916,7 +1982,6 @@ "integrity": "sha512-NTKlcQjlAK7MlQoyb6LgaqHc8sso/pVyUJYWMws3jg21uTJw/LddqIFPcPqP6PzpgbIcZyKI85sFE4HBrQDA8A==", "dev": true, "license": "MIT", - "peer": true, "dependencies": { "esbuild": "^0.25.0", "fdir": "^6.4.4", @@ -2010,7 +2075,6 @@ "integrity": "sha512-RvwwcruNjI1ncT5xRakeyS9Lf8lcItv34KD+aif+VH9kduAyfYBipGh12274xtenIPZ119/R9BdTBa8gAwSh0A==", "dev": true, "license": "MIT", - "peer": true, "engines": { "node": ">=12" }, @@ -2023,7 +2087,6 @@ "resolved": "https://registry.npmjs.org/vue/-/vue-3.5.41.tgz", "integrity": "sha512-2laE0p+aK+/AOPG/XL/WepOs/GlK755LJ1XECi9kDUrz1FKNw8rb2Xzlw9JS1rqEV55nb0ttsKxVlTCcd+R5cg==", "license": "MIT", - "peer": true, "dependencies": { "@vue/compiler-dom": "3.5.41", "@vue/compiler-sfc": "3.5.41", diff --git a/ui/package.json b/ui/package.json index 1c84933..eda292d 100644 --- a/ui/package.json +++ b/ui/package.json @@ -3,6 +3,9 @@ "version": "0.1.0", "private": true, "type": "module", + "engines": { + "node": ">=24.0.0" + }, "scripts": { "dev": "vite", "build": "vite build", From 5aac094e3ede964b930ad000aea3384f0a27aeb4 Mon Sep 17 00:00:00 2001 From: AlbertoV Date: Fri, 4 Sep 2026 21:59:45 +0200 Subject: [PATCH 03/85] ci: run Docker builds for the default branch --- .github/workflows/docker-build.yml | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/.github/workflows/docker-build.yml b/.github/workflows/docker-build.yml index 193b06a..e96eae9 100644 --- a/.github/workflows/docker-build.yml +++ b/.github/workflows/docker-build.yml @@ -18,10 +18,10 @@ name: Docker Image on: push: - branches: [ "main" ] + branches: [ "master" ] tags: [ "v*" ] pull_request: - branches: [ "main" ] + branches: [ "master" ] workflow_dispatch: # One run per ref: a tag push landing on top of a branch push should supersede From 0e357bd80d190b90eba1a48dcac976b5ee0525c9 Mon Sep 17 00:00:00 2001 From: AlbertoV Date: Fri, 4 Sep 2026 22:53:00 +0200 Subject: [PATCH 04/85] fix(console): report node backlog in graph detail --- crates/server/src/api/declare.rs | 32 ++++++++++++++++++++++++++++++++ crates/server/tests/live.rs | 11 +++++++++++ ui/src/views/GraphDetail.vue | 2 ++ 3 files changed, 45 insertions(+) diff --git a/crates/server/src/api/declare.rs b/crates/server/src/api/declare.rs index b3c5204..8161835 100644 --- a/crates/server/src/api/declare.rs +++ b/crates/server/src/api/declare.rs @@ -317,6 +317,36 @@ async fn do_sync(st: &Shared, application: &str, bodies: Vec) -> ApiResul pub async fn view(st: &Shared, rt: &Arc) -> Value { let mut nodes = Vec::new(); for (name, np) in &rt.plan.nodes { + // Keep the two owners of backlog separate in the graph view, just as + // the metrics and ETA endpoints do. The first number is work each + // stage has not admitted yet; the second is work Gate has relayed to a + // terminal queue and the application's consumers have not picked up. + let mut waiting_for_budget = 0u64; + for s in rt.stages_of_node(name) { + waiting_for_budget += st + .depths + .pending_of_group(&st.queen, &s.stage.source, &s.stage.group) + .await + .values() + .sum::(); + } + + let waiting_for_workers = match (&np.egress_queue, &np.egress_group) { + (Some(queue), Some(group)) => st + .depths + .pending_of_group(&st.queen, queue, group) + .await + .values() + .sum::(), + (Some(queue), None) => st + .depths + .pending(&st.queen, queue) + .await + .values() + .sum::(), + (None, _) => 0, + }; + let keys: Vec = np.unscoped().map(|b| b.key.clone()).collect(); let states = st.budgets.read(&keys).await.unwrap_or_default(); let breaker = crate::breaker::held(&st.budgets, np).await; @@ -367,6 +397,8 @@ pub async fn view(st: &Shared, rt: &Arc) -> Value { "paths": gate_core::plan::paths_through(&rt.plan, name), "shares": np.shares, "budgets": budgets, + "waiting_for_budget": waiting_for_budget, + "waiting_for_workers": waiting_for_workers, "breaker": breaker.map(|b| json!({ "at": b.at, "retryAfterSeconds": b.retry_after_seconds, diff --git a/crates/server/tests/live.rs b/crates/server/tests/live.rs index 776e494..dbe04a4 100644 --- a/crates/server/tests/live.rs +++ b/crates/server/tests/live.rs @@ -2945,6 +2945,17 @@ async fn an_eta_tells_a_budget_backlog_from_a_worker_one() { assert_eq!(eta["waitingForBudget"], json!(0), "{eta}"); assert_eq!(eta["state"], "waiting-workers", "{eta}"); + // The graph detail is what the topology diagram reads. These fields used + // to be absent, which the Vue component silently rendered as two zeroes. + let (status, view) = h.get_graph("g").await; + assert_eq!(status, 200, "{view}"); + let node = &view["nodes"][0]; + assert_eq!(node["waiting_for_budget"], json!(0), "{view}"); + assert!( + node["waiting_for_workers"].as_u64().unwrap_or(0) as usize >= N, + "the graph must show the worker backlog instead of a fallback zero: {view}" + ); + h.cleanup("g").await; } diff --git a/ui/src/views/GraphDetail.vue b/ui/src/views/GraphDetail.vue index 5f05dc1..4c04efc 100644 --- a/ui/src/views/GraphDetail.vue +++ b/ui/src/views/GraphDetail.vue @@ -266,6 +266,8 @@ async function remove() { running: graph.running, paths: n.paths ?? [], budgets: n.budgets ?? [], + waiting_for_budget: n.waiting_for_budget, + waiting_for_workers: n.waiting_for_workers, }))" :edges="edges" /> From 77a3f52f034d33b917df2e055de700b3c795773d Mon Sep 17 00:00:00 2001 From: AlbertoV Date: Fri, 4 Sep 2026 22:54:22 +0200 Subject: [PATCH 05/85] fix(runtime): fail graph closed when a stage exits --- crates/server/src/registry.rs | 2 +- crates/server/src/relay.rs | 40 ++++++++++++++++++++++++++++++++- crates/server/src/supervisor.rs | 4 +++- 3 files changed, 43 insertions(+), 3 deletions(-) diff --git a/crates/server/src/registry.rs b/crates/server/src/registry.rs index 7fd8bc7..cfd38cf 100644 --- a/crates/server/src/registry.rs +++ b/crates/server/src/registry.rs @@ -42,7 +42,7 @@ pub struct GraphRuntime { /// is stop-then-start, so the state exists for as long as a swap takes and /// outlives it whenever a restore fails — this is what makes it visible /// instead of implied. - pub stopped: AtomicBool, + pub stopped: Arc, /// One token per graph, cloned into every stage. pub cancel: queen_mq::Cancel, } diff --git a/crates/server/src/relay.rs b/crates/server/src/relay.rs index 845e732..8dcd814 100644 --- a/crates/server/src/relay.rs +++ b/crates/server/src/relay.rs @@ -40,7 +40,7 @@ //! rotation cursor and `MAX_IN_FLIGHT` all existed to do, badly, what the broker //! does here for free. -use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; use std::sync::Arc; use std::time::{Duration, SystemTime}; @@ -228,6 +228,7 @@ pub fn spawn( budgets: Budgets, st: Arc, traces: Arc, + graph_stopped: Arc, ) -> tokio::task::JoinHandle<()> { let k = knobs(); let q = queen.clone(); @@ -276,10 +277,16 @@ pub fn spawn( } }) .await; + // A consumer normally returns only because its graph was deliberately + // cancelled. Any other return leaves this stage's source without a + // reader, so fail the whole runtime closed and stop its siblings. The + // reconcile loop can then see `is_running() == false` and replace it. + let unexpected = stop_graph_after_stage_exit(&graph_stopped, &st.cancel); match res { Ok(summary) => tracing::info!( stage = %st.key(), queue = %st.stage.source, processed = summary.processed, reason = ?summary.stopped_by, + unexpected, "stage stopped" ), // A stage that exits is a stopped graph, which is the failure v1's @@ -296,6 +303,17 @@ pub fn spawn( }) } +/// Mark a runtime unhealthy when a stage returns without a graph cancellation. +/// Returns whether this exit initiated the stop, for the terminal log line. +fn stop_graph_after_stage_exit(stopped: &AtomicBool, cancel: &queen_mq::Cancel) -> bool { + if cancel.is_cancelled() { + return false; + } + stopped.store(true, Ordering::Relaxed); + cancel.cancel(); + true +} + struct Ctx { queen: Queen, budgets: Budgets, @@ -1302,6 +1320,26 @@ fn jitter_ms(wait_ms: i64) -> i64 { mod tests { use super::*; + #[test] + fn an_unexpected_stage_exit_stops_the_graph_and_its_siblings() { + let stopped = AtomicBool::new(false); + let cancel = Cancel::new(); + + assert!(stop_graph_after_stage_exit(&stopped, &cancel)); + assert!(stopped.load(Ordering::Relaxed)); + assert!(cancel.is_cancelled()); + } + + #[test] + fn a_planned_stage_exit_does_not_reclassify_the_stop() { + let stopped = AtomicBool::new(false); + let cancel = Cancel::new(); + cancel.cancel(); + + assert!(!stop_graph_after_stage_exit(&stopped, &cancel)); + assert!(!stopped.load(Ordering::Relaxed)); + } + fn budget(id: &str, count_sub: i64) -> gate_core::CompiledBudget { gate_core::CompiledBudget { id: id.into(), diff --git a/crates/server/src/supervisor.rs b/crates/server/src/supervisor.rs index 4ad1afc..6e6a891 100644 --- a/crates/server/src/supervisor.rs +++ b/crates/server/src/supervisor.rs @@ -59,13 +59,14 @@ pub async fn start( })); } + let stopped = Arc::new(AtomicBool::new(false)); let rt = Arc::new(GraphRuntime { doc, plan, stages, handles: parking_lot::RwLock::new(Vec::new()), persisted: AtomicBool::new(false), - stopped: AtomicBool::new(false), + stopped: stopped.clone(), cancel, }); @@ -80,6 +81,7 @@ pub async fn start( budgets.clone(), st.clone(), traces.clone(), + stopped.clone(), )); } *rt.handles.write() = handles; From c25cbde086c1ec4486685b487701349b5167a047 Mon Sep 17 00:00:00 2001 From: AlbertoV Date: Fri, 4 Sep 2026 22:58:33 +0200 Subject: [PATCH 06/85] fix(api): reject payload shapes Gate would discard --- crates/server/src/api/data.rs | 7 ++---- crates/server/src/api/mod.rs | 37 +++++++++++++++++++++++++++++++- crates/server/src/api/reenter.rs | 9 +++----- 3 files changed, 41 insertions(+), 12 deletions(-) diff --git a/crates/server/src/api/data.rs b/crates/server/src/api/data.rs index 38b4baf..62666a8 100644 --- a/crates/server/src/api/data.rs +++ b/crates/server/src/api/data.rs @@ -23,7 +23,7 @@ use serde_json::{json, Value}; use gate_core::plan::NodePlan; use gate_core::GATE_META; -use crate::api::{find, ok, refuse_if_stopped, resolve, ApiResult, Fail, Shared}; +use crate::api::{find, object_payload, ok, refuse_if_stopped, resolve, ApiResult, Fail, Shared}; use crate::registry::GraphRuntime; #[derive(Debug, Deserialize)] @@ -187,10 +187,7 @@ async fn push_into( } // ---- the envelope. - let mut item = body.payload.clone(); - if !item.is_object() { - item = json!({}); - } + let mut item = object_payload(body.payload.clone())?; { let obj = item.as_object_mut().expect("object"); if !body.op.is_empty() { diff --git a/crates/server/src/api/mod.rs b/crates/server/src/api/mod.rs index b9934c4..1768b37 100644 --- a/crates/server/src/api/mod.rs +++ b/crates/server/src/api/mod.rs @@ -30,7 +30,7 @@ use axum::routing::{get, post, put}; use axum::{Json, Router}; use parking_lot::RwLock; use queen_mq::Queen; -use serde_json::json; +use serde_json::{json, Value}; use crate::budget::Budgets; use crate::obs::Traces; @@ -289,6 +289,23 @@ pub fn ok(v: serde_json::Value) -> ApiResult { Ok(Json(v).into_response()) } +/// Gate's HTTP doors add `_gate` metadata to the application payload. Refuse a +/// shape that cannot carry that metadata instead of silently replacing the +/// caller's data with an empty object. +pub(crate) fn object_payload(payload: Value) -> Result { + if payload.is_object() { + return Ok(payload); + } + Err(Fail( + StatusCode::UNPROCESSABLE_ENTITY, + format!( + "payload must be a JSON object: Gate must add `{}` metadata without changing the \ + application value", + gate_core::GATE_META + ), + )) +} + impl From for Fail { fn from(r: crate::graph::Refusal) -> Self { match r { @@ -351,3 +368,21 @@ pub fn refuse_if_stopped(rt: &Arc) -> Result<(), ), )) } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn an_http_payload_must_be_an_object_instead_of_being_discarded() { + let kept = object_payload(json!({ "kept": true })) + .ok() + .expect("object refused"); + assert_eq!(kept["kept"], true); + for value in [json!(null), json!(7), json!("lost"), json!([1, 2])] { + let err = object_payload(value).expect_err("non-object accepted"); + assert_eq!(err.0, StatusCode::UNPROCESSABLE_ENTITY); + assert!(err.1.contains("must be a JSON object")); + } + } +} diff --git a/crates/server/src/api/reenter.rs b/crates/server/src/api/reenter.rs index fd6d5a1..913d527 100644 --- a/crates/server/src/api/reenter.rs +++ b/crates/server/src/api/reenter.rs @@ -38,7 +38,7 @@ use serde_json::{json, Value}; use gate_core::GATE_META; -use crate::api::{find, ok, refuse_if_stopped, resolve, ApiResult, Fail, Shared}; +use crate::api::{find, object_payload, ok, refuse_if_stopped, resolve, ApiResult, Fail, Shared}; use crate::registry::GraphRuntime; #[derive(Debug, Deserialize)] @@ -85,7 +85,8 @@ pub async fn graph_reenter_default( async fn reenter(st: &Shared, rt: &std::sync::Arc, body: ReenterBody) -> ApiResult { refuse_if_stopped(rt)?; - let stamp = body.payload.get(GATE_META); + let mut item = object_payload(body.payload.clone())?; + let stamp = item.get(GATE_META); let path = body .path .clone() @@ -164,10 +165,6 @@ async fn reenter(st: &Shared, rt: &std::sync::Arc, body: ReenterBo // Restamped at hop 0 of its own path, with the attempt on it. The relay // carries `attempt` forward across every hop, so the next report of this // item counts from here rather than starting again at one. - let mut item = body.payload.clone(); - if !item.is_object() { - item = json!({}); - } { let obj = item.as_object_mut().expect("object"); obj.insert( From ae7c11d9638dde69d267c304e84fd8a3a9c11e0e Mon Sep 17 00:00:00 2001 From: AlbertoV Date: Fri, 4 Sep 2026 22:59:31 +0200 Subject: [PATCH 07/85] fix(relay): charge shared counters once per message --- crates/server/src/relay.rs | 27 ++++++++++++++++++++++++++- 1 file changed, 26 insertions(+), 1 deletion(-) diff --git a/crates/server/src/relay.rs b/crates/server/src/relay.rs index 845e732..bef9cf9 100644 --- a/crates/server/src/relay.rs +++ b/crates/server/src/relay.rs @@ -818,7 +818,13 @@ fn group(st: &StageRuntime, msgs: &[Message]) -> Grouped { keys.len() - 1 } }; - here.push((idx, cost)); + // A shared key is one counter even when more than one budget on + // this node names it. The validator permits identical declarations + // deliberately; charging the same key once per declaration would + // multiply this message's cost and enforce a smaller limit. + if !here.iter().any(|(seen, _)| *seen == idx) { + here.push((idx, cost)); + } } per_msg.push(here); } @@ -1393,6 +1399,25 @@ mod tests { assert_eq!(charges[0].max, 100); } + /// Two identical declarations may intentionally share one counter. They + /// remain one spend per message, not one spend per declaration. + #[test] + fn duplicate_shared_budgets_charge_their_counter_once() { + let mut first = budget("first", 100); + first.shared_key = Some("vendor".into()); + let mut second = budget("second", 100); + second.key = first.key.clone(); + second.shared_key = first.shared_key.clone(); + let st = runtime(vec![first, second], 1.0); + let msgs: Vec = (0..3) + .map(|i| msg(&format!("t{i}"), json!({ "w": 4 }))) + .collect(); + + let charges = group(&st, &msgs).charges(msgs.len()); + assert_eq!(charges.len(), 1, "one shared key must produce one incr"); + assert_eq!(charges[0].delta, 12, "the declarations doubled the cost"); + } + /// A path's share IS the ceiling it carries: `round(count_sub * share)`. #[test] fn the_share_is_the_max_on_the_incr() { From a311a9a648c7ef6680176c7e6ffaffbe4da3a76f Mon Sep 17 00:00:00 2001 From: AlbertoV Date: Fri, 4 Sep 2026 23:01:27 +0200 Subject: [PATCH 08/85] fix(api): shed on any applicable full budget --- crates/server/src/api/data.rs | 173 ++++++++++++++++++++++++++++++---- 1 file changed, 154 insertions(+), 19 deletions(-) diff --git a/crates/server/src/api/data.rs b/crates/server/src/api/data.rs index 38b4baf..b73a218 100644 --- a/crates/server/src/api/data.rs +++ b/crates/server/src/api/data.rs @@ -250,7 +250,7 @@ async fn push_into( // for work that has not moved. What it buys is a caller who can back off // instead of filling a queue — 429 with the deadline read off the counter's // own TTL. - if let Some(retry_after) = shed(st, np, cost).await { + if let Some(retry_after) = shed(st, np, &item, cost).await { return Err(Fail( StatusCode::TOO_MANY_REQUESTS, format!( @@ -342,36 +342,80 @@ fn spread(partitions: Option) -> Option { Some(format!("p{i}")) } -/// `Some(seconds)` when every unscoped counter of this node is already full. +/// `Some(seconds)` when any counter applicable to this item is already full. /// /// Read-only, and best-effort: a broker that will not answer means the push goes /// through and the relay decides, which is the right way round — the door must /// never be the thing that stops work when the limiter itself is fine. -async fn shed(st: &Shared, np: &NodePlan, cost: i64) -> Option { +async fn shed(st: &Shared, np: &NodePlan, item: &Value, cost: i64) -> Option { if !np.ingress_shed { return None; } - let keys: Vec = np.unscoped().map(|b| b.key.clone()).collect(); - if keys.is_empty() { + let applicable = shed_budgets(np, item); + if applicable.is_empty() { return None; } + let keys: Vec = applicable.iter().map(|b| b.key.clone()).collect(); let states = st.budgets.read(&keys).await.ok()?; - let now = crate::now_ms(); - let mut worst: Option = None; - for b in np.unscoped() { - let ceiling = b.max_for(np.widest_share()); - let s = states.iter().find(|s| s.key == b.key)?; - if s.value + cost <= ceiling { - return None; + shed_wait(&applicable, &states, cost, crate::now_ms()) +} + +#[derive(Debug, PartialEq, Eq)] +struct ShedBudget { + key: String, + ceiling: i64, +} + +/// Resolve the exact counters the relay would charge for this payload. +fn shed_budgets(np: &NodePlan, item: &Value) -> Vec { + let op = gate_core::op_of(item); + let mut out = Vec::new(); + for b in &np.budgets { + if b.when_op + .as_ref() + .is_some_and(|patterns| !gate_core::op_matches(patterns, op)) + { + continue; + } + let key = match &b.scope_by { + Some(path) => match gate_core::scope_value(item, path) { + Some(value) => b.key_for(Some(&value)), + None => continue, + }, + None => b.key.clone(), + }; + if out.iter().any(|seen: &ShedBudget| seen.key == key) { + continue; } - let wait = s - .expires_at_ms - .map(|e| ((e - now) as f64 / 1000.0).ceil() as i64) - .unwrap_or(1) - .max(1); - worst = Some(worst.map_or(wait, |w: i64| w.max(wait))); + out.push(ShedBudget { + key, + ceiling: b.max_for(np.widest_share()), + }); } - worst + out +} + +fn shed_wait( + budgets: &[ShedBudget], + states: &[crate::budget::State], + cost: i64, + now: i64, +) -> Option { + budgets + .iter() + .filter_map(|b| { + let s = states.iter().find(|s| s.key == b.key)?; + if s.value <= b.ceiling.saturating_sub(cost) { + return None; + } + let wait = s + .expires_at_ms + .map(|e| ((e - now) as f64 / 1000.0).ceil() as i64) + .unwrap_or(1) + .max(1); + Some(wait) + }) + .max() } // ------------------------------------------------------------------- gone @@ -466,3 +510,94 @@ fn egress_hint(st: &Shared, application: &str, graph: &str, node: &str) -> Strin ), } } + +#[cfg(test)] +mod tests { + use super::*; + + fn budget(id: &str) -> gate_core::CompiledBudget { + gate_core::CompiledBudget { + id: id.into(), + key: format!("key:{id}"), + scope_by: None, + shared_key: None, + when_op: None, + count: 10, + time_ms: 1000, + sub_windows: 1, + count_sub: 10, + window_sub_seconds: 1, + confidence: gate_core::Confidence::Inferred, + } + } + + fn node(budgets: Vec) -> NodePlan { + NodePlan { + name: "n".into(), + budgets, + cost: gate_core::Cost::Fixed(1), + ingress_queue: Some("in".into()), + ingress_owned: true, + ingress_http: true, + ingress_shed: true, + interior_queue: "interior".into(), + egress_queue: Some("out".into()), + egress_group: None, + breaker_key: "breaker".into(), + shares: Default::default(), + } + } + + #[test] + fn one_full_budget_is_enough_to_shed() { + let budgets = vec![ + ShedBudget { + key: "full".into(), + ceiling: 10, + }, + ShedBudget { + key: "room".into(), + ceiling: 10, + }, + ]; + let states = vec![ + crate::budget::State { + key: "full".into(), + value: 10, + expires_at_ms: Some(12_000), + }, + crate::budget::State { + key: "room".into(), + value: 0, + expires_at_ms: Some(20_000), + }, + ]; + + assert_eq!(shed_wait(&budgets, &states, 1, 10_000), Some(2)); + } + + #[test] + fn shed_resolves_when_op_and_scoped_keys_for_this_item() { + let global = budget("global"); + let mut writes = budget("writes"); + writes.when_op = Some(vec!["listing.write".into()]); + let mut customer = budget("customer"); + customer.scope_by = Some("payload.customerId".into()); + let np = node(vec![global, writes, customer]); + + let got = shed_budgets(&np, &json!({ "op": "listing.read", "customerId": "c-7" })); + assert_eq!( + got, + vec![ + ShedBudget { + key: "key:global".into(), + ceiling: 10, + }, + ShedBudget { + key: "key:customer:c-7".into(), + ceiling: 10, + }, + ] + ); + } +} From 86bd48199b07888311f02d75725f5988ecf3f4b5 Mon Sep 17 00:00:00 2001 From: AlbertoV Date: Fri, 4 Sep 2026 23:04:27 +0200 Subject: [PATCH 09/85] fix(console): derive pacing from current backlog --- crates/server/src/api/console.rs | 15 +++++++++++---- crates/server/tests/live.rs | 27 +++++++++++++++++++++++++++ ui/src/views/GraphDetail.vue | 4 ++-- 3 files changed, 40 insertions(+), 6 deletions(-) diff --git a/crates/server/src/api/console.rs b/crates/server/src/api/console.rs index 1451564..f10a878 100644 --- a/crates/server/src/api/console.rs +++ b/crates/server/src/api/console.rs @@ -140,10 +140,11 @@ pub async fn list_targets(State(app): State) -> ApiResult { let mut out = Vec::new(); for g in app.registry.all() { - // The backlog of everything that has not been admitted yet: every - // ingress queue this graph reads, under the group that reads it. + // The backlog of everything that has not been admitted yet: every stage + // source this graph reads, under the group that reads it. Interior + // stages matter too — a downstream budget can be the binding one. let mut backlog = 0u64; - for s in g.stages.iter().filter(|s| s.stage.first_hop) { + for s in &g.stages { backlog += app .depths .pending_of_group(&app.queen, &s.stage.source, &s.stage.group) @@ -228,7 +229,13 @@ pub async fn list_targets(State(app): State) -> ApiResult { "worst_assumed": worst.4, "admitted": adm, "denied": den, - "state": if den > 0 { "pacing" } else { "flowing" }, + "state": if !g.is_running() { + "down" + } else if backlog > 0 { + "pacing" + } else { + "flowing" + }, "backlog": backlog, "at": now, })); diff --git a/crates/server/tests/live.rs b/crates/server/tests/live.rs index dbe04a4..7cb2fb5 100644 --- a/crates/server/tests/live.rs +++ b/crates/server/tests/live.rs @@ -2895,6 +2895,20 @@ async fn an_eta_answers_from_the_declared_schedule_when_the_window_is_spent() { "the answer is a bound and must read as one: {eta}" ); + let (_, targets) = h.send(reqwest::Method::GET, "/api/targets", None).await; + let mine = targets + .as_array() + .and_then(|rows| { + rows.iter() + .find(|t| t["application"] == h.application && t["name"] == "g") + }) + .expect("graph missing from target index"); + assert_eq!( + mine["state"], "pacing", + "current backlog must drive state: {mine}" + ); + assert!(mine["backlog"].as_u64().unwrap_or(0) > 0, "{mine}"); + h.cleanup("g").await; } @@ -2956,6 +2970,19 @@ async fn an_eta_tells_a_budget_backlog_from_a_worker_one() { "the graph must show the worker backlog instead of a fallback zero: {view}" ); + let (_, targets) = h.send(reqwest::Method::GET, "/api/targets", None).await; + let mine = targets + .as_array() + .and_then(|rows| { + rows.iter() + .find(|t| t["application"] == h.application && t["name"] == "g") + }) + .expect("graph missing from target index"); + assert_eq!( + mine["state"], "flowing", + "worker backlog is not budget pacing: {mine}" + ); + h.cleanup("g").await; } diff --git a/ui/src/views/GraphDetail.vue b/ui/src/views/GraphDetail.vue index 4c04efc..e3efb78 100644 --- a/ui/src/views/GraphDetail.vue +++ b/ui/src/views/GraphDetail.vue @@ -131,7 +131,7 @@ function nodeState(n) { if (!graph.value?.running) return 'down' if (n.breaker) return 'breached' if ((n.budgets ?? []).some((b) => b.confidence === 'assumed')) return 'blind' - if (counters(n.node).deferred > 0) return 'pacing' + if ((n.waiting_for_budget ?? 0) > 0) return 'pacing' return 'flowing' } @@ -140,7 +140,7 @@ const state = computed(() => { if (nodes.value.some((n) => n.breaker)) return 'breached' if (nodes.value.some((n) => (n.budgets ?? []).some((b) => b.confidence === 'assumed'))) return 'blind' - return totals.value.deferred > 0 ? 'pacing' : 'flowing' + return nodes.value.some((n) => (n.waiting_for_budget ?? 0) > 0) ? 'pacing' : 'flowing' }) const CONFIDENCE_NOTE = { From eafd23ecd0e22653fa0c8ac0ac8f0ebbc32d9fb0 Mon Sep 17 00:00:00 2001 From: AlbertoV Date: Fri, 4 Sep 2026 23:06:47 +0200 Subject: [PATCH 10/85] fix(console): emit saturating for growing backlog --- crates/server/src/api/console.rs | 3 ++ crates/server/src/api/mod.rs | 2 + crates/server/src/lib.rs | 1 + crates/server/src/obs.rs | 74 +++++++++++++++++++++++++++++++- 4 files changed, 79 insertions(+), 1 deletion(-) diff --git a/crates/server/src/api/console.rs b/crates/server/src/api/console.rs index f10a878..5ea95a3 100644 --- a/crates/server/src/api/console.rs +++ b/crates/server/src/api/console.rs @@ -152,6 +152,7 @@ pub async fn list_targets(State(app): State) -> ApiResult { .values() .sum::(); } + let saturating = app.backlogs.sample(&g.key(), backlog); let (mut adm, mut den) = (0u64, 0u64); for s in &g.stages { @@ -231,6 +232,8 @@ pub async fn list_targets(State(app): State) -> ApiResult { "denied": den, "state": if !g.is_running() { "down" + } else if saturating { + "saturating" } else if backlog > 0 { "pacing" } else { diff --git a/crates/server/src/api/mod.rs b/crates/server/src/api/mod.rs index b9934c4..9695cbc 100644 --- a/crates/server/src/api/mod.rs +++ b/crates/server/src/api/mod.rs @@ -42,6 +42,7 @@ pub struct App { pub budgets: Budgets, pub registry: Registry, pub depths: Arc, + pub backlogs: crate::obs::BacklogTrends, pub traces: Arc, pub history: Option>, pub queen_url: String, @@ -75,6 +76,7 @@ impl App { queen, registry: Default::default(), depths: Arc::new(crate::depth::Depths::default()), + backlogs: Default::default(), traces: Arc::new(Traces::default()), history: None, queen_url, diff --git a/crates/server/src/lib.rs b/crates/server/src/lib.rs index bb42fa4..3c5473f 100644 --- a/crates/server/src/lib.rs +++ b/crates/server/src/lib.rs @@ -112,6 +112,7 @@ pub async fn run() -> Result<(), Box> { queen, registry: Default::default(), depths: Arc::new(depth::Depths::default()), + backlogs: Default::default(), traces: Arc::new(obs::Traces::default()), history: history.clone(), queen_url: queen_url.clone(), diff --git a/crates/server/src/obs.rs b/crates/server/src/obs.rs index 0e3ccb6..979d2b6 100644 --- a/crates/server/src/obs.rs +++ b/crates/server/src/obs.rs @@ -13,8 +13,9 @@ //! Nothing was broken; that is what the observability of the old design cost //! while idle, and idle is most of the time. -use std::collections::VecDeque; +use std::collections::{HashMap, VecDeque}; use std::sync::atomic::{AtomicU64, Ordering}; +use std::time::{Duration, Instant}; use parking_lot::RwLock; use serde_json::{json, Value}; @@ -112,6 +113,50 @@ impl StageCounters { } } +/// Consecutive live backlog samples for the overview's `saturating` state. +/// A growth observation is held across a few console polls so the state does +/// not flicker between the depth cache's refreshes. +const SATURATING_HOLD: Duration = Duration::from_secs(15); + +#[derive(Debug)] +struct BacklogSample { + depth: u64, + growing_until: Option, +} + +#[derive(Default)] +pub struct BacklogTrends { + samples: RwLock>, +} + +impl BacklogTrends { + pub fn sample(&self, graph: &str, depth: u64) -> bool { + self.sample_at(graph, depth, Instant::now()) + } + + fn sample_at(&self, graph: &str, depth: u64, now: Instant) -> bool { + let mut samples = self.samples.write(); + let Some(previous) = samples.get_mut(graph) else { + samples.insert( + graph.to_string(), + BacklogSample { + depth, + growing_until: None, + }, + ); + return false; + }; + + if depth > previous.depth { + previous.growing_until = Some(now + SATURATING_HOLD); + } else if depth < previous.depth || depth == 0 { + previous.growing_until = None; + } + previous.depth = depth; + previous.growing_until.is_some_and(|until| until > now) + } +} + /// One refusal, kept. /// /// **Denials only.** An admission is counted and never traced: it is the common @@ -191,3 +236,30 @@ impl Traces { self.len() == 0 } } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn backlog_growth_is_held_but_drain_clears_it_immediately() { + let trends = BacklogTrends::default(); + let now = Instant::now(); + + assert!(!trends.sample_at("app/g", 3, now)); + assert!(trends.sample_at("app/g", 5, now + Duration::from_secs(1))); + assert!(trends.sample_at("app/g", 5, now + Duration::from_secs(5))); + assert!(!trends.sample_at("app/g", 4, now + Duration::from_secs(6))); + assert!(!trends.sample_at("app/g", 0, now + Duration::from_secs(7))); + } + + #[test] + fn a_growth_latch_expires_without_another_increase() { + let trends = BacklogTrends::default(); + let now = Instant::now(); + + assert!(!trends.sample_at("app/g", 1, now)); + assert!(trends.sample_at("app/g", 2, now + Duration::from_secs(1))); + assert!(!trends.sample_at("app/g", 2, now + Duration::from_secs(17))); + } +} From 71c7d25af1b698353784cc91c8f3d2346f240fc9 Mon Sep 17 00:00:00 2001 From: AlbertoV Date: Fri, 4 Sep 2026 23:12:10 +0200 Subject: [PATCH 11/85] fix(history): checkpoint counters only after commit --- crates/server/src/history.rs | 44 ++++++++++++---- crates/server/src/lib.rs | 99 +++++++++++++++++++++++++++--------- crates/server/src/obs.rs | 43 ++++++++++++++++ 3 files changed, 151 insertions(+), 35 deletions(-) diff --git a/crates/server/src/history.rs b/crates/server/src/history.rs index ac84265..736d7f4 100644 --- a/crates/server/src/history.rs +++ b/crates/server/src/history.rs @@ -128,12 +128,21 @@ impl History { /// Add one minute's increments. Two replicas writing the same minute is the /// normal case, not a race: they saw different halves of the traffic, and /// the row is the sum. - pub async fn add(&self, app: &str, target: &str, minute: i64, lanes: &HashMap) { - let Ok(client) = self.pool.get().await else { - return; + pub async fn add( + &self, + app: &str, + target: &str, + minute: i64, + lanes: &HashMap, + ) -> bool { + let Ok(mut client) = self.pool.get().await else { + return false; + }; + let Ok(tx) = client.transaction().await else { + return false; }; for (lane, b) in lanes { - let _ = client + if tx .execute( "INSERT INTO gate.rollups (application, target, lane, minute, admitted, denied, calls, throttled, cost_est, cost_actual) @@ -152,8 +161,13 @@ impl History { &b.cost_estimated, &b.cost_actual, ], ) - .await; + .await + .is_err() + { + return false; + } } + tx.commit().await.is_ok() } pub async fn rollups(&self, app: &str, target: &str, minutes: i64) -> Vec { @@ -319,15 +333,18 @@ impl History { .collect() } - pub async fn add_traces(&self, rows: &[crate::obs::Trace]) { + pub async fn add_traces(&self, rows: &[crate::obs::Trace]) -> bool { if rows.is_empty() { - return; + return true; } - let Ok(client) = self.pool.get().await else { - return; + let Ok(mut client) = self.pool.get().await else { + return false; + }; + let Ok(tx) = client.transaction().await else { + return false; }; for t in rows { - let _ = client + if tx .execute( "INSERT INTO gate.traces (at, application, target, lane, op, outcome, budget_id, calls) VALUES ($1,$2,$3,$4,$5,$6,$7,$8)", @@ -336,8 +353,13 @@ impl History { &t.path, &t.op, &t.outcome, &t.budget_id, &0i64, ], ) - .await; + .await + .is_err() + { + return false; + } } + tx.commit().await.is_ok() } pub async fn traces(&self, outcome: Option<&str>, limit: i64) -> Vec { diff --git a/crates/server/src/lib.rs b/crates/server/src/lib.rs index bb42fa4..3ebeaa1 100644 --- a/crates/server/src/lib.rs +++ b/crates/server/src/lib.rs @@ -208,9 +208,12 @@ pub fn spawn_reconcile( /// It reads the stages' own `AtomicU64`s and writes the DELTA since the last /// pass, so two replicas writing the same minute is the normal case rather than /// a race: they saw different halves of the traffic and the row is the sum. +type CounterSnapshot = (u64, u64, u64); +type CounterCheckpoint = (String, CounterSnapshot); + pub fn spawn_counters(app: api::Shared) -> tokio::task::JoinHandle<()> { tokio::spawn(async move { - let mut last: HashMap = HashMap::new(); + let mut last: HashMap = HashMap::new(); loop { tokio::time::sleep(std::time::Duration::from_secs(60)).await; let now = now_ms(); @@ -223,53 +226,83 @@ pub fn spawn_counters(app: api::Shared) -> tokio::task::JoinHandle<()> { } let mut per_target: HashMap> = HashMap::new(); + let mut checkpoints: HashMap> = HashMap::new(); for s in &g.stages { let key = format!("{}/{}", g.key(), s.key()); + let target = format!("{}.{}", g.doc.graph, s.stage.node); let c = &s.counters; let o = std::sync::atomic::Ordering::Relaxed; let now3 = ( c.admitted.load(o), - c.deferred.load(o) + c.released.load(o), + c.deferred.load(o).saturating_add(c.released.load(o)), c.cost.load(o), ); - let was = last.insert(key, now3).unwrap_or((0, 0, 0)); - let d = ( - now3.0.saturating_sub(was.0), - now3.1.saturating_sub(was.1), - now3.2.saturating_sub(was.2), - ); + let d = counter_delta(now3, last.get(&key).copied()); + checkpoints + .entry(target.clone()) + .or_default() + .push((key, now3)); if d == (0, 0, 0) { continue; } - per_target - .entry(format!("{}.{}", g.doc.graph, s.stage.node)) - .or_default() - .insert( - s.stage.path.clone(), - history::Bucket { - admitted: d.0, - denied: d.1, - cost_estimated: d.2 as f64, - ..Default::default() - }, - ); + per_target.entry(target).or_default().insert( + s.stage.path.clone(), + history::Bucket { + admitted: d.0, + denied: d.1, + cost_estimated: d.2 as f64, + ..Default::default() + }, + ); } - for (target, paths) in per_target { - h.add(&g.doc.application, &target, minute, &paths).await; + for (target, samples) in checkpoints { + let written = match per_target.remove(&target) { + Some(paths) => h.add(&g.doc.application, &target, minute, &paths).await, + None => true, + }; + if written { + for (key, value) in samples { + last.insert(key, value); + } + } else { + tracing::warn!( + application = %g.doc.application, + %target, + "could not persist counter rollup; retaining the previous checkpoint" + ); + } } } // The refusal ring, on the same cadence. Bounded and // drop-oldest, so a flush that misses a pass loses the oldest // denials and never blocks the hot path. let traces = app.traces.drain(); - if !traces.is_empty() { - h.add_traces(&traces).await; + if !traces.is_empty() && !h.add_traces(&traces).await { + tracing::warn!( + count = traces.len(), + "could not persist traces; returning them to the ring" + ); + app.traces.restore(traces); } } } }) } +/// Delta of a lifetime counter tuple. A lower value means the stage runtime was +/// replaced and its atomics restarted at zero, so the new value is itself the +/// entire increment since that reset. +fn counter_delta(now: CounterSnapshot, previous: Option) -> CounterSnapshot { + let Some(was) = previous else { + return now; + }; + ( + now.0.checked_sub(was.0).unwrap_or(now.0), + now.1.checked_sub(was.1).unwrap_or(now.1), + now.2.checked_sub(was.2).unwrap_or(now.2), + ) +} + /// Bring back everything that was declared, at boot. pub async fn restore(app: &api::Shared) { let _guard = app.declare_lock.lock().await; @@ -364,3 +397,21 @@ pub async fn reconcile(app: &api::Shared) { } } } + +#[cfg(test)] +mod tests { + use super::counter_delta; + + #[test] + fn a_restarted_counter_counts_from_its_new_zero() { + assert_eq!(counter_delta((7, 3, 11), Some((100, 80, 900))), (7, 3, 11)); + } + + #[test] + fn a_running_counter_reports_only_its_increment() { + assert_eq!( + counter_delta((107, 83, 911), Some((100, 80, 900))), + (7, 3, 11) + ); + } +} diff --git a/crates/server/src/obs.rs b/crates/server/src/obs.rs index 0e3ccb6..6cea2b8 100644 --- a/crates/server/src/obs.rs +++ b/crates/server/src/obs.rs @@ -183,6 +183,18 @@ impl Traces { .collect() } + /// Put a failed durable flush back before anything recorded while the write + /// was in flight. The ring remains bounded and drops its oldest entries. + pub fn restore(&self, rows: Vec) { + let mut ring = self.ring.write(); + for row in rows.into_iter().rev() { + ring.push_front(row); + } + while ring.len() > TRACE_RING { + ring.pop_front(); + } + } + pub fn len(&self) -> usize { self.ring.read().len() } @@ -191,3 +203,34 @@ impl Traces { self.len() == 0 } } + +#[cfg(test)] +mod tests { + use super::*; + + fn trace(at: i64) -> Trace { + Trace { + at, + application: "a".into(), + graph: "g".into(), + node: "n".into(), + path: "p".into(), + op: String::new(), + outcome: "denied", + budget_id: Some("b".into()), + } + } + + #[test] + fn a_failed_trace_flush_returns_before_newer_rows() { + let traces = Traces::default(); + traces.push(trace(1)); + traces.push(trace(2)); + let failed = traces.drain(); + traces.push(trace(3)); + + traces.restore(failed); + let recent: Vec = traces.recent(None, 3).iter().map(|t| t.at).collect(); + assert_eq!(recent, vec![3, 2, 1]); + } +} From c87dee4685ef36c069fe3908882f20589e080e29 Mon Sep 17 00:00:00 2001 From: AlbertoV Date: Fri, 4 Sep 2026 23:15:41 +0200 Subject: [PATCH 12/85] fix(history): calculate utilisation in cost units --- crates/server/src/api/console.rs | 16 ++++++++++++---- crates/server/src/api/declare.rs | 1 + crates/server/src/history.rs | 10 +++++++--- ui/src/components/FlowChart.vue | 2 +- ui/src/lib/rollups.js | 12 +++++++++++- ui/src/views/BudgetHistory.vue | 12 ++++++++++++ 6 files changed, 44 insertions(+), 9 deletions(-) diff --git a/crates/server/src/api/console.rs b/crates/server/src/api/console.rs index 1451564..948a228 100644 --- a/crates/server/src/api/console.rs +++ b/crates/server/src/api/console.rs @@ -324,6 +324,9 @@ pub async fn flow(State(app): State, Query(q): Query) -> ApiR for (name, np) in &g.plan.nodes { let per_min = np .unscoped() + // A target-level rollup has no operation dimension, so it + // cannot be compared honestly with an operation-only budget. + .filter(|b| b.when_op.is_none()) .map(|b| b.count_sub as f64 * 60.0 / b.window_sub_seconds.max(1) as f64) .fold(f64::INFINITY, f64::min); if per_min.is_finite() && per_min > 0.0 { @@ -342,20 +345,22 @@ pub async fn flow(State(app): State, Query(q): Query) -> ApiR utilisation: f64, target: String, admitted: i64, + cost: f64, ceiling: f64, total: i64, } let mut cells: HashMap<(String, i64), Cell> = HashMap::new(); let mut minute_set: BTreeSet = BTreeSet::new(); - for (application, target, minute, admitted) in h.flow(minutes, now).await { + for (application, target, minute, admitted, cost) in h.flow(minutes, now).await { minute_set.insert(minute); let cap = ceiling.get(&(application.clone(), target.clone())).copied(); - let u = cap.map_or(0.0, |c| admitted as f64 / c); + let u = cap.map_or(0.0, |c| cost / c); let e = cells.entry((application.clone(), minute)).or_insert(Cell { utilisation: 0.0, target: target.clone(), admitted: 0, + cost: 0.0, ceiling: cap.unwrap_or(0.0), total: 0, }); @@ -364,6 +369,7 @@ pub async fn flow(State(app): State, Query(q): Query) -> ApiR e.utilisation = u; e.target = target; e.admitted = admitted; + e.cost = cost; e.ceiling = cap.unwrap_or(0.0); } } @@ -378,12 +384,14 @@ pub async fn flow(State(app): State, Query(q): Query) -> ApiR .map(|t| match cells.get(&(a.clone(), *t)) { Some(c) => json!({ "t": t, "utilisation": c.utilisation, "target": c.target, - "admitted": c.admitted, "ceiling": c.ceiling, "total_admitted": c.total, + "admitted": c.admitted, "cost": c.cost, + "ceiling": c.ceiling, "total_admitted": c.total, }), // A minute an application did not appear in is a minute it // admitted nothing, which is a real zero and not a gap. None => { - json!({ "t": t, "utilisation": 0.0, "admitted": 0, "total_admitted": 0 }) + json!({ "t": t, "utilisation": 0.0, "admitted": 0, + "cost": 0.0, "total_admitted": 0 }) } }) .collect(); diff --git a/crates/server/src/api/declare.rs b/crates/server/src/api/declare.rs index b3c5204..94819bc 100644 --- a/crates/server/src/api/declare.rs +++ b/crates/server/src/api/declare.rs @@ -333,6 +333,7 @@ pub async fn view(st: &Shared, rt: &Arc) -> Value { "key": b.key, "scopeBy": b.scope_by, "sharedKey": b.shared_key, + "whenOp": b.when_op, "count": b.count, "timeMs": b.time_ms, "subWindows": b.sub_windows, diff --git a/crates/server/src/history.rs b/crates/server/src/history.rs index ac84265..f754821 100644 --- a/crates/server/src/history.rs +++ b/crates/server/src/history.rs @@ -287,19 +287,22 @@ impl History { // counter now, so N ceilings cannot oversubscribe it, and the whole argument // evaporates with the feature. - /// Admissions per minute for every target, over the last `minutes`. + /// Admissions and admitted cost per minute for every target, over the last + /// `minutes`. /// /// One query for the whole deployment rather than one per target: the /// dashboard draws every application at once, and N round trips to draw one /// picture is how a console starts costing more than the thing it watches. - pub async fn flow(&self, minutes: i64, now_ms: i64) -> Vec<(String, String, i64, i64)> { + pub async fn flow(&self, minutes: i64, now_ms: i64) -> Vec<(String, String, i64, i64, f64)> { let Ok(client) = self.pool.get().await else { return vec![]; }; let since = now_ms / 60_000 * 60_000 - minutes * 60_000; let rows = client .query( - "SELECT application, target, minute, COALESCE(SUM(admitted), 0)::BIGINT + "SELECT application, target, minute, + COALESCE(SUM(admitted), 0)::BIGINT, + COALESCE(NULLIF(SUM(cost_est), 0), SUM(admitted)::DOUBLE PRECISION, 0) FROM gate.rollups WHERE minute >= $1 GROUP BY application, target, minute ORDER BY minute", @@ -314,6 +317,7 @@ impl History { r.try_get::<_, String>(1).ok()?, r.try_get::<_, i64>(2).ok()?, r.try_get::<_, i64>(3).ok()?, + r.try_get::<_, f64>(4).ok()?, )) }) .collect() diff --git a/ui/src/components/FlowChart.vue b/ui/src/components/FlowChart.vue index 280f3fc..8971259 100644 --- a/ui/src/components/FlowChart.vue +++ b/ui/src/components/FlowChart.vue @@ -221,7 +221,7 @@ function toneOf(u) { would read as a limit of zero, which is the opposite of idle. --> diff --git a/ui/src/lib/rollups.js b/ui/src/lib/rollups.js index 3a6e879..95d55de 100644 --- a/ui/src/lib/rollups.js +++ b/ui/src/lib/rollups.js @@ -90,6 +90,10 @@ export function perMinute(rows, path = null) { */ export function budgetSeries(minutes, budget) { if (!minutes) return null + // Roll-ups have node/path dimensions, but no scope value or operation. A + // percentage for either kind of selective budget would use unrelated work in + // its numerator, so leave it unknown instead of drawing a false zero/peak. + if (budget?.scopeBy || budget?.whenOp?.length) return null // The SUB-window and its count, because that is what is enforced: a budget // declared over ten seconds and subdivided into ten is a one-second window of // a tenth, and drawing the declared pair would draw a ceiling nothing meets. @@ -99,7 +103,13 @@ export function budgetSeries(minutes, budget) { const allowance = p > 0 ? cap * ((span * 60) / p) : 0 const out = minutes.map((w, i) => { let sum = 0 - for (let k = Math.max(0, i - span + 1); k <= i; k++) sum += minutes[k].admitted + for (let k = Math.max(0, i - span + 1); k <= i; k++) { + // Rows written before cost rollups existed legitimately contain zero in + // that column. Costs are positive, so admitted is a safe legacy fallback. + sum += (minutes[k].cost_estimated ?? 0) > 0 + ? minutes[k].cost_estimated + : (minutes[k].admitted ?? 0) + } return { ...w, utilisation: allowance > 0 ? sum / allowance : 0 } }) // A trailing sum that has not filled yet is not a low utilisation, it is an diff --git a/ui/src/views/BudgetHistory.vue b/ui/src/views/BudgetHistory.vue index ed3755e..5c97a0d 100644 --- a/ui/src/views/BudgetHistory.vue +++ b/ui/src/views/BudgetHistory.vue @@ -61,6 +61,9 @@ watch([() => props.app, () => props.name, () => props.node, () => props.budget, const node = computed(() => (target.value?.nodes ?? []).find((n) => n.node === props.node) ?? null) const spec = computed(() => (node.value?.budgets ?? []).find((b) => b.id === props.budget) ?? null) +const historicalComparable = computed( + () => !spec.value?.scopeBy && !(spec.value?.whenOp?.length) +) const points = computed(() => (spec.value ? budgetSeries(rows.value, spec.value) ?? [] : [])) const peak = computed(() => points.value.reduce((a, w) => Math.max(a, w.utilisation ?? 0), 0)) const totals = computed(() => @@ -223,6 +226,15 @@ const tone = computed(() => (peak.value > 1 ? 'text-bad' : peak.value >= 0.85 ?
+
+

Historical utilisation is not attributable.

+

+ This budget selects a scope value or operation, while roll-ups aggregate the whole node. + The live gauge remains authoritative; drawing a percentage from unrelated admissions would + be misleading. +

+
+

No window recorded yet.

From 07f1cc817aa0d5afc90a099765fea3af245004a0 Mon Sep 17 00:00:00 2001 From: AlbertoV Date: Fri, 4 Sep 2026 23:16:43 +0200 Subject: [PATCH 13/85] fix(auth): constrain OAuth next redirects to local paths --- crates/server/src/auth.rs | 48 +++++++++++++++++++++++++++++++++++++-- 1 file changed, 46 insertions(+), 2 deletions(-) diff --git a/crates/server/src/auth.rs b/crates/server/src/auth.rs index 4f511d7..b48841d 100644 --- a/crates/server/src/auth.rs +++ b/crates/server/src/auth.rs @@ -327,6 +327,24 @@ fn nonce(secret: &[u8]) -> String { format!("{h:016x}{:x}", t) } +/// A post-login destination is always local to this console. The state is +/// signed against tampering, but the caller chooses the value before it is +/// signed, so signature verification alone does not prevent an open redirect. +fn safe_next(candidate: Option<&str>) -> String { + let Some(path) = candidate else { + return "/".into(); + }; + if path.starts_with('/') + && !path.starts_with("//") + && !path.contains('\\') + && !path.chars().any(char::is_control) + { + path.to_string() + } else { + "/".into() + } +} + pub async fn login( State(app): State, Query(q): Query>, @@ -337,7 +355,7 @@ pub async fn login( let n = nonce(&auth.cfg.secret); let state = match auth.sign(&StateClaims { nonce: n.clone(), - next: q.get("next").cloned().unwrap_or_else(|| "/".into()), + next: safe_next(q.get("next").map(String::as_str)), exp: now() + STATE_TTL_S, }) { Ok(s) => s, @@ -448,6 +466,9 @@ pub async fn callback( }; let secure = auth.cfg.public_url.starts_with("https://"); + // Validate again after state verification so a state minted by an older + // version cannot retain an unsafe destination through a rolling deploy. + let next = safe_next(Some(&st.next)); let cookie = format!( "{COOKIE}={token}; Path=/; HttpOnly; SameSite=Lax; Max-Age={}{}", 8 * 3600, @@ -455,7 +476,7 @@ pub async fn callback( ); ( StatusCode::SEE_OTHER, - [(header::SET_COOKIE, cookie), (header::LOCATION, st.next)], + [(header::SET_COOKIE, cookie), (header::LOCATION, next)], ) .into_response() } @@ -607,3 +628,26 @@ fn urlencode(s: &str) -> String { }) .collect() } + +#[cfg(test)] +mod tests { + use super::safe_next; + + #[test] + fn oauth_next_accepts_only_local_absolute_paths() { + for path in ["/", "/graphs", "/#/apps/a/graphs/g?path=main"] { + assert_eq!(safe_next(Some(path)), path); + } + for path in [ + "https://evil.example", + "//evil.example/path", + "/\\evil.example/path", + "graphs", + "", + "/graphs\r\nLocation: https://evil.example", + ] { + assert_eq!(safe_next(Some(path)), "/", "accepted {path:?}"); + } + assert_eq!(safe_next(None), "/"); + } +} From 1615ff6545e6d26d20261187b9bb2f7df5592273 Mon Sep 17 00:00:00 2001 From: AlbertoV Date: Fri, 4 Sep 2026 23:17:50 +0200 Subject: [PATCH 14/85] fix(console): expose budget provenance consistently --- crates/server/src/api/declare.rs | 14 +++++++++++++- crates/server/tests/live.rs | 11 +++++++++++ ui/src/views/BudgetHistory.vue | 6 +++--- 3 files changed, 27 insertions(+), 4 deletions(-) diff --git a/crates/server/src/api/declare.rs b/crates/server/src/api/declare.rs index b3c5204..a285bd6 100644 --- a/crates/server/src/api/declare.rs +++ b/crates/server/src/api/declare.rs @@ -324,7 +324,17 @@ pub async fn view(st: &Shared, rt: &Arc) -> Value { let budgets: Vec = np .budgets .iter() - .map(|b| { + .enumerate() + .map(|(index, b)| { + // Enforcement uses the compiled budget, while provenance is + // intentionally documentation-only and remains on the source + // document. The compiler preserves budget order. + let declared = rt + .doc + .nodes + .get(name) + .and_then(|node| node.budgets.get(index)) + .filter(|source| source.id_or(index) == b.id); let s = states.iter().find(|s| s.key == b.key); let ceiling = b.max_for(np.widest_share()); let value = s.map(|s| s.value).unwrap_or(0); @@ -339,6 +349,8 @@ pub async fn view(st: &Shared, rt: &Arc) -> Value { "countSub": b.count_sub, "windowSubSeconds": b.window_sub_seconds, "confidence": b.confidence, + "source": declared.and_then(|source| source.source.as_ref()), + "asOf": declared.and_then(|source| source.as_of.as_ref()), // A per-key budget has no single counter to report: the // number that matters is the worst live key, and finding it // means enumerating a namespace. `null` says so rather than diff --git a/crates/server/tests/live.rs b/crates/server/tests/live.rs index 776e494..1375b92 100644 --- a/crates/server/tests/live.rs +++ b/crates/server/tests/live.rs @@ -2249,6 +2249,8 @@ async fn the_console_can_draw_what_is_running() { let out = egress_of("console", &h.application); let mut doc = chain_doc(); doc["nodes"]["ip"]["egress"] = json!(out); + doc["nodes"]["ip"]["budgets"][0]["source"] = json!("vendor limits page"); + doc["nodes"]["ip"]["budgets"][0]["asOf"] = json!("2026-08-20"); let (status, body) = h.put_graph("g", doc).await; assert_eq!(status, 200, "declare: {body}"); @@ -2264,6 +2266,15 @@ async fn the_console_can_draw_what_is_running() { assert_eq!(topo["edges"][0]["to"], "ip"); assert_eq!(topo["paths"][0]["name"], "main"); + let (status, detail) = h.get_graph("g").await; + assert_eq!(status, 200, "{detail}"); + let budget = &detail["nodes"] + .as_array() + .and_then(|nodes| nodes.iter().find(|n| n["node"] == "ip")) + .expect("ip node missing")["budgets"][0]; + assert_eq!(budget["source"], "vendor limits page", "{detail}"); + assert_eq!(budget["asOf"], "2026-08-20", "{detail}"); + let (status, graphs) = h.send(reqwest::Method::GET, "/api/graphs", None).await; assert_eq!(status, 200, "{graphs}"); assert!(graphs.as_array().is_some_and(|a| !a.is_empty())); diff --git a/ui/src/views/BudgetHistory.vue b/ui/src/views/BudgetHistory.vue index ed3755e..7081d48 100644 --- a/ui/src/views/BudgetHistory.vue +++ b/ui/src/views/BudgetHistory.vue @@ -328,9 +328,9 @@ const tone = computed(() => (peak.value > 1 ? 'text-bad' : peak.value >= 0.85 ?

{{ spec.confidence }}

-

as of {{ spec.as_of }}

-

- counted per {{ spec.scope.join(' + ') }} +

as of {{ spec.asOf }}

+

+ counted per {{ spec.scopeBy }}

From 2fbb4c4c2082bd7cf1851075f0a6a44ee4cd445c Mon Sep 17 00:00:00 2001 From: AlbertoV Date: Fri, 4 Sep 2026 23:18:41 +0200 Subject: [PATCH 15/85] fix(console): normalize trace budget field names --- crates/server/src/obs.rs | 3 +++ crates/server/tests/units.rs | 2 ++ ui/src/components/TraceList.vue | 4 +++- 3 files changed, 8 insertions(+), 1 deletion(-) diff --git a/crates/server/src/obs.rs b/crates/server/src/obs.rs index 0e3ccb6..38a7916 100644 --- a/crates/server/src/obs.rs +++ b/crates/server/src/obs.rs @@ -143,6 +143,9 @@ impl Trace { "path": self.path, "op": self.op, "outcome": self.outcome, + // Durable traces have always used the schema/API spelling. Keep the + // former live-only camelCase alias for one compatibility window. + "budget_id": self.budget_id, "budgetId": self.budget_id, }) } diff --git a/crates/server/tests/units.rs b/crates/server/tests/units.rs index f4cfab6..d4ac4d9 100644 --- a/crates/server/tests/units.rs +++ b/crates/server/tests/units.rs @@ -376,5 +376,7 @@ fn the_trace_ring_drops_the_oldest_and_never_grows() { (gate_server::obs::TRACE_RING + 49) as i64, "newest first" ); + assert_eq!(recent[0].view()["budget_id"], "b"); + assert_eq!(recent[0].view()["budgetId"], "b", "compatibility alias"); assert!(t.recent(Some("admitted"), 10).is_empty(), "denials only"); } diff --git a/ui/src/components/TraceList.vue b/ui/src/components/TraceList.vue index 41b6a13..700ff45 100644 --- a/ui/src/components/TraceList.vue +++ b/ui/src/components/TraceList.vue @@ -25,6 +25,8 @@ const WORD = { function tone(t) { return t.outcome === 'throttled' ? 'text-bad' : t.outcome === 'ok' ? 'text-fg-3' : 'text-fg-2' } + +const traceBudget = (t) => t.budget_id ?? t.budgetId