Integrate all 62 open PRs

.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

.github/workflows/test.yml

@@ -17,7 +17,10 @@ on:
   push:
     branches: [ "master" ]
   pull_request:
-    branches: [ "master" ]
+    # Every base, not only master. A PR stacked on another feature branch is
+    # otherwise never checked at all: its first build is the merge into master,
+    # which is the one moment nobody wants to learn it does not compile.
+    branches: [ '**' ]
   workflow_dispatch:
 
 concurrency:
@@ -74,7 +77,7 @@ jobs:
 
     - uses: actions/setup-node@v4
       with:
-        node-version: 22
+        node-version: 24
         cache: npm
         cache-dependency-path: ui/package-lock.json
 

.nvmrc

@@ -0,0 +1 @@
+24

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 ./

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
@@ -209,6 +212,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
@@ -231,8 +235,9 @@ there are counters (`popped`, `admitted`, `deferred`, `parked`, `released`, `for
 explains a stage's throughput. `wedged` is the one to alert on: it counts a stage whose ack the
 broker keeps refusing at a claim head that never moves, which is a stuck cursor and not a budget
 backlog — the stage says so once at `ERROR` with the `seek` that fixes it. Denials are kept in a bounded in-process ring; admissions are counted, never
-traced. Rollups are opt-in per graph (`"counters": { "windowSeconds": 60 }`), because observability
-is a thing you switch on, not a thing that runs whether or not anyone is looking.
+traced. Rollups are opt-in per graph (`"counters": { "windowSeconds": 60 }`); the current storage
+and API contract is a fixed one-minute window, so `60` is the only accepted value. Observability is
+a thing you switch on, not a thing that runs whether or not anyone is looking.
 
 **One thing to say out loud.** The declaration names your egress queue, and:
 

crates/core/src/cost.rs

@@ -7,7 +7,8 @@
 
 use serde_json::Value;
 
-use crate::doc::{Cost, PAYLOAD_ROOT};
+use crate::doc::{Cost, GATE_META, PAYLOAD_ROOT};
+use crate::plan::CompiledBudget;
 
 /// Walk a dotted payload path. The first segment must be `payload`, which names
 /// the message's own `data`; `payload.a.b` is `data["a"]["b"]`.
@@ -21,22 +22,30 @@ pub fn resolve<'a>(data: &'a Value, path: &str) -> Option<&'a Value> {
     if segs.next()? != PAYLOAD_ROOT {
         return None;
     }
-    let mut cur = data;
+    let first = segs.next()?;
+    if first.is_empty() || first == GATE_META {
+        return None;
+    }
+    let mut cur = data.get(first)?;
     for s in segs {
+        if s.is_empty() {
+            return None;
+        }
         cur = cur.get(s)?;
     }
     Some(cur)
 }
 
 /// Whether a string is a usable payload path: `payload` plus at least one
-/// segment, each of them non-empty.
+/// non-empty segment. Gate's root `_gate` envelope is deliberately outside the
+/// declaration language: costs and scopes may only come from producer data.
 pub fn ok_payload_path(path: &str) -> bool {
     let mut segs = path.split('.');
     if segs.next() != Some(PAYLOAD_ROOT) {
         return false;
     }
     let rest: Vec<&str> = segs.collect();
-    !rest.is_empty() && rest.iter().all(|s| !s.is_empty())
+    !rest.is_empty() && rest.first() != Some(&GATE_META) && rest.iter().all(|s| !s.is_empty())
 }
 
 /// The scope value a budget keys on, as it reaches the kv key.
@@ -54,6 +63,32 @@ pub fn scope_value(data: &Value, path: &str) -> Option<String> {
     }
 }
 
+/// The first applicable scoped budget whose key cannot be resolved.
+///
+/// Applicability comes first: a `photo.delete` per-listing budget has no reason
+/// to require `listingId` from a `photo.upload`. Both the HTTP door and the
+/// relay use this one answer so direct queue ingress cannot enforce a different
+/// contract from HTTP ingress.
+pub fn missing_scope<'a>(
+    budgets: &'a [CompiledBudget],
+    data: &Value,
+) -> Option<(&'a str, &'a str)> {
+    let op = op_of(data);
+    budgets.iter().find_map(|budget| {
+        if budget
+            .when_op
+            .as_ref()
+            .is_some_and(|patterns| !op_matches(patterns, op))
+        {
+            return None;
+        }
+        let path = budget.scope_by.as_deref()?;
+        scope_value(data, path)
+            .is_none()
+            .then_some((budget.id.as_str(), path))
+    })
+}
+
 /// What this item costs, or the reason it can never be admitted.
 ///
 /// Integers, because `kv.incr`'s delta is `i64` on this wire. A resolved cost
@@ -69,10 +104,22 @@ pub fn cost_of(cost: &Cost, data: &Value) -> Result<i64, TooExpensive> {
         Cost::Fixed(n) => (*n, *n),
         Cost::Path(c) => {
             let max = c.max.unwrap_or(c.default);
-            let v = resolve(data, &c.path)
-                .and_then(integral)
-                .filter(|n| *n >= 1)
-                .unwrap_or(c.default);
+            let v = match resolve(data, &c.path) {
+                Some(value) => match integral(value) {
+                    Ok(value) => value.filter(|n| *n >= 1).unwrap_or(c.default),
+                    // `i64::MAX` is the largest lower bound the public error
+                    // type can carry. It is enough to refuse every ordinary
+                    // maximum; equality is the sentinel for an out-of-range
+                    // positive number and gets its own truthful message below.
+                    Err(()) => {
+                        return Err(TooExpensive {
+                            cost: i64::MAX,
+                            max,
+                        })
+                    }
+                },
+                None => c.default,
+            };
             (v, max)
         }
     };
@@ -94,6 +141,13 @@ pub struct TooExpensive {
 
 impl std::fmt::Display for TooExpensive {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        if self.cost == i64::MAX && self.max == i64::MAX {
+            return write!(
+                f,
+                "this item declares a cost outside the signed 64-bit range the broker can charge: \
+                 refusing it is safer than silently charging i64::MAX"
+            );
+        }
         write!(
             f,
             "this item declares a cost of {} and the node admits at most {}: an item that cannot \
@@ -105,13 +159,38 @@ impl std::fmt::Display for TooExpensive {
 }
 
 /// A JSON number that is a whole number. `3.0` is three; `3.5` is not a cost.
-fn integral(v: &Value) -> Option<i64> {
+///
+/// A whole number outside `i64` is an error rather than a missing value. Rust's
+/// float-to-integer cast saturates, so treating it as an ordinary conversion
+/// would collapse every larger JSON number to `i64::MAX` and undercharge it.
+fn integral(v: &Value) -> Result<Option<i64>, ()> {
     match v {
         Value::Number(n) => match n.as_i64() {
-            Some(i) => Some(i),
-            None => n.as_f64().filter(|f| f.fract() == 0.0).map(|f| f as i64),
+            Some(i) => Ok(Some(i)),
+            None => {
+                let Some(f) = n.as_f64() else {
+                    return Ok(None);
+                };
+                if f.fract() != 0.0 {
+                    return Ok(None);
+                }
+                // `i64::MAX as f64` rounds to 2^63, one past the largest i64,
+                // so the upper bound is deliberately exclusive. The lower one
+                // is inclusive because -2^63 is representable.
+                const I64_BOUND: f64 = 9_223_372_036_854_775_808.0;
+                if f >= I64_BOUND {
+                    return Err(());
+                }
+                // A negative value already means "use the default". Keep that
+                // tolerance even when its magnitude is outside i64; unlike an
+                // oversized positive cost, it cannot make Gate undercharge.
+                if f < -I64_BOUND {
+                    return Ok(None);
+                }
+                Ok(Some(f as i64))
+            }
         },
-        _ => None,
+        _ => Ok(None),
     }
 }
 

crates/core/src/doc.rs

@@ -100,11 +100,16 @@ pub struct Counters {
     pub window_seconds: u32,
 }
 
+/// Roll-ups are stored in minute-keyed rows and every history endpoint reads
+/// those rows as minutes. Keep the one supported value named in one place so a
+/// declaration cannot promise a window the runtime does not actually emit.
+pub const COUNTERS_WINDOW_SECONDS: u32 = 60;
+
 #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Default)]
 #[serde(deny_unknown_fields)]
 pub struct Node {
-    /// At least one, and at least one of them unscoped — see `node-budget` and
-    /// `node-unscoped-budget`.
+    /// At least one, and at least one of them unconditional and unscoped — see
+    /// `node-budget` and `node-unscoped-budget`.
     #[serde(default)]
     pub budgets: Vec<Budget>,
 
@@ -129,9 +134,9 @@ pub struct Node {
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub batch: Option<u32>,
 
-    /// How many workers drain this node's stages. Defaults to
-    /// `max(4, source partitions)`. More workers than partitions is harmless
-    /// (the extras find nothing and park); fewer is a throughput ceiling.
+    /// How many workers drain this node's stages. By default this is derived
+    /// from the tightest unconditional, unscoped rate and capped at the source
+    /// partition count; an explicit value overrides that derivation.
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub concurrency: Option<u32>,
 }

crates/core/src/lib.rs

@@ -20,11 +20,13 @@ pub mod plan;
 pub mod v1;
 pub mod validate;
 
-pub use cost::{cost_of, ok_payload_path, op_matches, op_of, resolve, scope_value, TooExpensive};
+pub use cost::{
+    cost_of, missing_scope, ok_payload_path, op_matches, op_of, resolve, scope_value, TooExpensive,
+};
 pub use doc::{
     default_application, ok_name, ok_target_name, Budget, Confidence, Cost, CostPath, Counters,
-    Egress, EgressSpec, GraphDoc, Ingress, IngressSpec, Node, Path, PathElem, GATE_META,
-    PAYLOAD_ROOT,
+    Egress, EgressSpec, GraphDoc, Ingress, IngressSpec, Node, Path, PathElem,
+    COUNTERS_WINDOW_SECONDS, GATE_META, PAYLOAD_ROOT,
 };
 pub use ids::derive;
 pub use plan::{
@@ -34,6 +36,7 @@ pub use plan::{
     ASSUMED_FACTOR, DEFAULT_BATCH, DEFAULT_INGRESS_PARTITIONS,
 };
 pub use validate::{
-    needs_version_bump, validate, validate_with, warnings, warnings_with, ExternalFacts, Problem,
-    QueueFacts,
+    needs_version_bump, refuses_stored_document, validate, validate_plan_with, validate_with,
+    warnings, warnings_with, ExternalFacts, Problem, QueueFacts, MAX_BREAKER_COUNTERS,
+    MAX_GRAPH_WORKERS,
 };

crates/core/src/migrate.rs

@@ -103,7 +103,26 @@ fn rolling_sub_windows(time_ms: i64, count: i64, cost_max: i64) -> u32 {
 }
 
 fn budget(b: &v1::Budget, cost_max: i64, out: &mut Vec<Problem>, node: &str) -> Budget {
-    let time_ms = b.period_seconds.max(1) * 1000;
+    let seconds = b.period_seconds.max(1);
+    // v1 carries seconds in an i64 while v2 carries milliseconds in one. The
+    // multiplication can therefore overflow for a document that is perfectly
+    // valid on the old wire. Keep the migration total and say explicitly when
+    // the destination type cannot represent the original duration.
+    let time_ms = seconds.saturating_mul(1000);
+    if seconds > i64::MAX / 1000 {
+        out.push(w(
+            "period-clamped",
+            format!(
+                "budget `{}` of node `{node}` declares periodSeconds {}. The v2 `timeMs` field \
+                 cannot represent that many milliseconds, so it was capped at {}ms instead of \
+                 overflowing the migration. Lower the period and redeclare if this budget is \
+                 intended to rotate on an operational timescale.",
+                b.id,
+                b.period_seconds,
+                i64::MAX
+            ),
+        ));
+    }
     let count = (b.cap.floor() as i64).max(1);
 
     let sub_windows = match b.alignment {
@@ -125,8 +144,8 @@ fn budget(b: &v1::Budget, cost_max: i64, out: &mut Vec<Problem>, node: &str) ->
                      admitted.",
                     b.id,
                     (time_ms / n.max(1) as i64) / 1000,
-                    2 * (count / n.max(1) as i64).max(1),
-                    2 * count
+                    (count / n.max(1) as i64).max(1).saturating_mul(2),
+                    count.saturating_mul(2)
                 ),
             ));
             Some(n)
@@ -227,9 +246,9 @@ fn budget(b: &v1::Budget, cost_max: i64, out: &mut Vec<Problem>, node: &str) ->
 /// Two v1 shapes land here. A **class node** with an out-edge was allowed to
 /// declare no budget at all: it existed to isolate a traffic class and carry a
 /// priority, and the limit it was checked against lived downstream. And a node
-/// with only SCOPED budgets was legal too — v1's ETA read the worst key and its
-/// breach ring was per-replica, so neither needed a node-level denominator; v2's
-/// ETA and its breaker both do.
+/// with only SCOPED or CONDITIONAL budgets was legal too — v1's ETA read the
+/// worst key and its breach ring was per-replica, so neither needed a counter
+/// every item meets; v2's ETA and its breaker both do.
 ///
 /// Either way the mapping declares a pass-through — which limits nothing,
 /// exactly as before — and says so loudly, rather than inventing a ceiling
@@ -238,11 +257,12 @@ fn passthrough_budget(node: &str, out: &mut Vec<Problem>) -> Budget {
     out.push(w(
         "node-budget",
         format!(
-            "node `{node}` declared no budget on the node itself (v1 allowed that for a class \
-             node, and for a node carrying only per-key budgets). v2 requires one — it is what \
-             the ETA measures a rate against and what the breaker spends when a vendor says 429 \
-             — so a pass-through of 1000000 per second has been declared for it: it limits \
-             nothing, exactly as before. Replace it with the real limit."
+            "node `{node}` declared no unconditional budget on the node itself (v1 allowed that \
+             for a class node, and for a node carrying only per-key or conditional budgets). v2 \
+             requires one — it is what the ETA measures a rate against and what the breaker \
+             spends when a vendor says 429 — so a pass-through of 1000000 per second has been \
+             declared for it: it limits nothing, exactly as before. Replace it with the real \
+             limit."
         ),
     ));
     Budget {
@@ -392,7 +412,11 @@ pub fn from_v1_target(spec: &v1::TargetSpec) -> Result<Migrated, Refused> {
         // NOT mapped — see `lane_concurrency_warning`.
         concurrency: None,
     };
-    if node.budgets.iter().all(|b| b.scope_by.is_some()) {
+    if node
+        .budgets
+        .iter()
+        .all(|b| b.scope_by.is_some() || b.when_op.is_some())
+    {
         node.budgets.push(passthrough_budget(&node_name, &mut out));
     }
     lane_concurrency_warning(&lanes, &node_name, &mut out);
@@ -480,7 +504,11 @@ pub fn from_v1_graph(spec: &v1::GraphSpec) -> Result<Migrated, Refused> {
             // NOT mapped — see `lane_concurrency_warning`.
             concurrency: None,
         };
-        if node.budgets.iter().all(|b| b.scope_by.is_some()) {
+        if node
+            .budgets
+            .iter()
+            .all(|b| b.scope_by.is_some() || b.when_op.is_some())
+        {
             node.budgets.push(passthrough_budget(name, &mut out));
         }
         if node.egress.is_some() {