diff --git a/.github/workflows/helm.yaml b/.github/workflows/helm.yaml index f678baa..67ec089 100644 --- a/.github/workflows/helm.yaml +++ b/.github/workflows/helm.yaml @@ -22,6 +22,10 @@ env: HELM_VERSION: v3.17.3 KIND_VERSION: v0.32.0 KUBECONFORM_VERSION: 0.6.7 + # v1.x's plugin.yaml uses the platformCommand/platformHooks manifest + # schema, which the pinned Helm v3.17.3 cannot load. v0.8.2 is the last + # release using the command/hooks schema Helm 3.17 supports. + HELM_UNITTEST_VERSION: v0.8.2 jobs: # ---------------------------------------------------------------------------- @@ -49,6 +53,12 @@ jobs: rm /tmp/kubeconform.tar.gz kubeconform -v + - name: Install helm-unittest + run: helm plugin install https://github.com/helm-unittest/helm-unittest --version "${HELM_UNITTEST_VERSION}" + + - name: Chart template unit tests + run: helm unittest charts/grid-site + - name: Validate charts (static) run: ./scripts/verify-helm-chart.sh diff --git a/charts/grid-operator/crds/gridnetwork.yaml b/charts/grid-operator/crds/gridnetwork.yaml index cbc177a..6d62104 100644 --- a/charts/grid-operator/crds/gridnetwork.yaml +++ b/charts/grid-operator/crds/gridnetwork.yaml @@ -34,6 +34,42 @@ spec: Defines the grid's seed peers, gateway associations, SWIM tuning, and TLS secret references. properties: + budgetPolicy: + description: |- + Budget policy configuration for per-tenant spend tracking. + + Selects which tenants Grid tracks cumulative spend for. See + [`BudgetPolicyConfig`] for what this does and does not do. + + **Default (absent):** no tenants are tracked; `budgetStatus` is + always empty. + nullable: true + properties: + tenants: + default: [] + description: Per-tenant budget caps. + items: + description: Per-tenant budget cap declaration. + properties: + capUsd: + description: |- + Maximum cumulative spend in USD before this tenant is considered over budget. + + **Minimum value:** `0`. The generated CRD schema rejects negative caps; + [`validate_budget_policy`] additionally rejects `NaN`/infinite values + that the schema's numeric minimum does not catch. + format: double + minimum: 0.0 + type: number + tenantId: + description: Tenant identifier. Must be non-empty and unique within the policy. + type: string + required: + - capUsd + - tenantId + type: object + type: array + type: object gatewayRefs: default: [] description: References to Praxis Gateways that participate in this grid. @@ -416,6 +452,48 @@ spec: description: Observed status of a [`GridNetwork`]. nullable: true properties: + budgetStatus: + description: |- + Per-tenant budget status, derived from `spec.budgetPolicy` and merged + cross-site CRDT spend state. + + Empty when `budgetPolicy` is absent. This is a status signal only — + Grid does not enforce budget limits itself (see [`BudgetPolicyConfig`]). + items: + description: |- + Per-tenant budget status derived from policy + merged CRDT spend state. + + Populated in [`GridNetworkStatus::budget_status`] for every tenant + declared in `spec.budgetPolicy`, regardless of whether spend has been + recorded for that tenant yet. This is a status signal only — Grid does + not enforce budget limits (see [`BudgetPolicyConfig`] doc). + properties: + capUsd: + description: Budget cap for this tenant, in USD, copied from `spec.budgetPolicy`. + format: double + type: number + spendRatio: + description: '`spend_usd / cap_usd`, clamped to `0.0..=1.0`. See [`spend_ratio`].' + format: double + type: number + spendUsd: + description: |- + Cumulative spend observed for this tenant, in USD. + + Converged across all sites that have merged CRDT state for this + tenant; may lag briefly during a partition (see [`GCounter`]). + format: double + type: number + tenantId: + description: Tenant identifier, matching `spec.budgetPolicy.tenants[].tenantId`. + type: string + required: + - capUsd + - spendRatio + - spendUsd + - tenantId + type: object + type: array connectedSites: default: 0 description: Number of connected (Active) sites. diff --git a/charts/grid-site/templates/gridnetwork.yaml b/charts/grid-site/templates/gridnetwork.yaml index 909c44c..2ab23e8 100644 --- a/charts/grid-site/templates/gridnetwork.yaml +++ b/charts/grid-site/templates/gridnetwork.yaml @@ -6,9 +6,14 @@ metadata: labels: {{- include "grid-site.labels" . | nindent 4 }} spec: - {{- with .Values.gridNetwork.gridId }} - gridId: {{ . | quote }} - {{- end }} + {{- /* + gridId is always rendered (even when empty) so `spec` is never a null + YAML value. The CRD schema requires `spec` to be present, and a bare + `spec:` with no children renders as null when every other field below + is also left at its falsy chart default, which the API server rejects + with "spec: Required value" on both create and update. + */}} + gridId: {{ .Values.gridNetwork.gridId | quote }} {{- with .Values.gridNetwork.region }} region: {{ . | quote }} {{- end }} @@ -29,6 +34,10 @@ spec: metricsRefreshInterval: {{ . | quote }} {{- end }} {{- end }} + {{- with .Values.gridNetwork.budgetPolicy }} + budgetPolicy: + {{- toYaml . | nindent 4 }} + {{- end }} {{- with .Values.gridNetwork.swim }} swim: {{- toYaml . | nindent 4 }} diff --git a/charts/grid-site/tests/gridnetwork_test.yaml b/charts/grid-site/tests/gridnetwork_test.yaml new file mode 100644 index 0000000..fbe5df9 --- /dev/null +++ b/charts/grid-site/tests/gridnetwork_test.yaml @@ -0,0 +1,63 @@ +suite: GridNetwork template +templates: + - templates/gridnetwork.yaml +tests: + - it: renders as a GridNetwork resource with no budgetPolicy by default + set: + gridNetwork.name: prod-grid + gridSite.name: prod-site + asserts: + - hasDocuments: + count: 1 + - isKind: + of: GridNetwork + - equal: + path: metadata.name + value: prod-grid + - notExists: + path: spec.budgetPolicy + + - it: renders a non-null spec when only the required name fields are set + # Regression test: every optional field used to be gated behind `with`, + # so `spec:` rendered as YAML null when all of them were left at their + # falsy chart defaults. A real API server rejects that with + # "spec: Required value" on both create and update — helm-unittest + # can't reproduce server-side CRD validation, so this asserts the + # concrete symptom instead: gridId must always be present. + set: + gridNetwork.name: prod-grid + gridSite.name: prod-site + asserts: + - equal: + path: spec.gridId + value: "" + + - it: renders budgetPolicy.tenants verbatim when set + set: + gridNetwork.name: prod-grid + gridSite.name: prod-site + gridNetwork.budgetPolicy: + tenants: + - tenantId: tenant-a + capUsd: 100 + - tenantId: tenant-b + capUsd: 250 + asserts: + - equal: + path: spec.budgetPolicy.tenants + value: + - tenantId: tenant-a + capUsd: 100 + - tenantId: tenant-b + capUsd: 250 + + - it: renders budgetPolicy with an empty tenants list without erroring + set: + gridNetwork.name: prod-grid + gridSite.name: prod-site + gridNetwork.budgetPolicy: + tenants: [] + asserts: + - equal: + path: spec.budgetPolicy.tenants + value: [] diff --git a/charts/grid-site/values.schema.json b/charts/grid-site/values.schema.json index 5da18ba..1f81332 100644 --- a/charts/grid-site/values.schema.json +++ b/charts/grid-site/values.schema.json @@ -53,6 +53,24 @@ "type": "string", "pattern": "^([1-9][0-9]*s|[1-9][0-9]{3,}ms)$" }, + "budgetPolicy": { + "type": "object", + "additionalProperties": false, + "properties": { + "tenants": { + "type": "array", + "items": { + "type": "object", + "additionalProperties": false, + "required": ["tenantId", "capUsd"], + "properties": { + "tenantId": { "type": "string", "minLength": 1 }, + "capUsd": { "type": "number", "minimum": 0 } + } + } + } + } + }, "gatewayRefs": { "type": "array", "items": { diff --git a/crdt/src/gcounter.rs b/crdt/src/gcounter.rs index 96b895b..071ef1c 100644 --- a/crdt/src/gcounter.rs +++ b/crdt/src/gcounter.rs @@ -27,7 +27,7 @@ use serde::{Deserialize, Serialize}; /// c.increment(10); /// assert_eq!(c.total(), 10); /// ``` -#[derive(Clone, Debug, Deserialize, Serialize)] +#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)] pub struct GCounter { /// Site identifier for this replica. site_id: String, @@ -73,6 +73,30 @@ impl GCounter { *slot = (*slot).max(*count); } } + + /// Return a copy of this counter containing only the slot for `origin_site`. + /// + /// Used at trust boundaries (e.g. gossip wire-ingest) where a payload's + /// claimed origin should only ever be believed for its own contribution. + /// Any other slot present in `self` — legitimate or forged — is dropped, + /// mirroring how provider records are scoped to their claimed origin + /// before being accepted. + #[must_use] + pub fn retain_origin(&self, origin_site: &str) -> Self { + let mut retained = Self::new(origin_site.to_owned()); + if let Some(&value) = self.slots.get(origin_site) { + retained.slots.insert(origin_site.to_owned(), value); + } + retained + } + + /// Remove the slot belonging to `site`, if present. + /// + /// Used when evicting a dead site so its contribution doesn't linger in + /// other tenants' counters forever. A no-op if `site` never contributed. + pub fn remove_slot(&mut self, site: &str) { + self.slots.remove(site); + } } // --------------------------------------------------------------------------- @@ -178,6 +202,62 @@ mod tests { ); } + #[test] + fn retain_origin_keeps_only_the_named_slot() { + let mut c = GCounter::new("site-a".to_owned()); + c.increment(10); + c.slots.insert("site-b".to_owned(), 999); + c.slots.insert("site-c".to_owned(), 42); + + let retained = c.retain_origin("site-a"); + + assert_eq!(retained.total(), 10, "only site-a's slot must survive"); + assert_eq!( + retained.local(), + 10, + "the retained counter is keyed by site-a, so local() reflects its slot" + ); + } + + #[test] + fn retain_origin_for_absent_slot_is_zero() { + let mut c = GCounter::new("site-a".to_owned()); + c.increment(10); + + let retained = c.retain_origin("site-b"); + + assert_eq!( + retained.total(), + 0, + "a slot the origin never wrote must retain as zero, not forged" + ); + } + + #[test] + fn remove_slot_drops_only_the_named_site() { + let mut c = GCounter::new("site-a".to_owned()); + c.increment(10); + c.slots.insert("site-b".to_owned(), 20); + + c.remove_slot("site-a"); + + assert_eq!( + c.total(), + 20, + "removing site-a's slot must leave site-b's contribution intact" + ); + } + + #[test] + fn remove_slot_for_absent_site_is_a_no_op() { + let mut c = GCounter::new("site-a".to_owned()); + c.increment(10); + + c.remove_slot("site-never-contributed"); + + assert_eq!(c.total(), 10, "removing an absent slot must not change the total"); + } + #[test] fn gcounter_serde_round_trip() { let mut c = GCounter::new("site-x".to_owned()); diff --git a/crdt/src/grid_state.rs b/crdt/src/grid_state.rs index 2f32382..d6adec4 100644 --- a/crdt/src/grid_state.rs +++ b/crdt/src/grid_state.rs @@ -21,7 +21,16 @@ use std::collections::BTreeMap; use serde::{Deserialize, Serialize}; -use crate::OrSet; +use crate::{GCounter, OrSet}; + +/// Hard bound on the number of distinct `tenant_id`s tracked in +/// [`GridStateSnapshot::tenant_spend`]. +/// +/// Bounds memory growth from gossip: a malicious or buggy origin could +/// otherwise flood arbitrary `tenant_id` keys indefinitely. Mirrors the +/// `max_origins` bound already used for retained-origin maps in `swim`. +pub const MAX_TRACKED_TENANTS: usize = 1_024; + // --------------------------------------------------------------------------- // Access policy // --------------------------------------------------------------------------- @@ -163,6 +172,20 @@ pub struct GridStateSnapshot { /// Provider records keyed by a stable `network_id/site_id/provider_id` string. pub providers: BTreeMap, + + /// Per-tenant cumulative spend, keyed by tenant identifier. + /// + /// Each [`GCounter`] total is denominated in **cents** (`u64`) to keep + /// the CRDT free of float-merge precision concerns; consumers convert to + /// USD at the edge (see `operator::crd::grid_network::spend_ratio`). + /// This is a cross-site *visibility* signal only — Grid does not enforce + /// budget limits; that is a gateway-side policy-filter concern. + /// + /// `#[serde(default)]` so snapshots serialized before this field existed + /// (or peers running an older build) deserialize to an empty map instead + /// of failing. + #[serde(default)] + pub tenant_spend: BTreeMap, } impl GridStateSnapshot { @@ -172,6 +195,7 @@ impl GridStateSnapshot { Self { capabilities: OrSet::new(site_id.clone()), providers: BTreeMap::new(), + tenant_spend: BTreeMap::new(), site_id, } } @@ -198,6 +222,80 @@ impl GridStateSnapshot { for provider in other.providers.values() { self.upsert_provider(provider.clone()); } + self.merge_tenant_spend(&other.tenant_spend); + } + + /// Merge tenant spend counters from another snapshot's `tenant_spend` map. + /// + /// Mirrors the add-wins semantics of `capabilities.merge`: each tenant's + /// [`GCounter`] merge takes the max of every site's slot, so this is safe + /// to call repeatedly, out of order, or with disjoint tenant sets. + pub fn merge_tenant_spend(&mut self, other: &BTreeMap) { + for (tenant_id, counter) in other { + self.tenant_spend + .entry(tenant_id.clone()) + .or_insert_with(|| GCounter::new(self.site_id.clone())) + .merge(counter); + } + } + + /// Merge tenant spend from a single wire broadcast, trusting only the + /// slot attributable to its claimed `origin_site`. + /// + /// This is the trust boundary for gossip ingest (called from + /// `StateBroadcastHandler::receive_item`). Unlike [`merge_tenant_spend`] + /// — used for trusted full-snapshot-to-full-snapshot merges where every + /// slot is already locally attested — a single broadcast should only + /// ever carry its own origin's contribution. Any other slot present in + /// the payload is dropped rather than merged, so a compromised or buggy + /// peer cannot forge another site's recorded spend by embedding extra + /// slots in its own broadcast. + /// + /// Also enforces [`MAX_TRACKED_TENANTS`]: once that many distinct + /// `tenant_id`s are tracked, brand-new `tenant_id`s are silently refused + /// (already-tracked tenants keep accepting updates) to bound memory + /// growth from gossip carrying arbitrary attacker-supplied `tenant_id`s. + /// + /// [`merge_tenant_spend`]: Self::merge_tenant_spend + pub fn merge_tenant_spend_from_origin(&mut self, origin_site: &str, other: &BTreeMap) { + for (tenant_id, counter) in other { + let origin_only = counter.retain_origin(origin_site); + if origin_only.total() == 0 { + continue; + } + if !self.tenant_spend.contains_key(tenant_id) && self.tenant_spend.len() >= MAX_TRACKED_TENANTS { + continue; + } + self.tenant_spend + .entry(tenant_id.clone()) + .or_insert_with(|| GCounter::new(self.site_id.clone())) + .merge(&origin_only); + } + } + + /// Remove `origin_site`'s contribution from every tenant's spend counter. + /// + /// Used by the SWIM runtime's dead-member eviction sweep (mirrors + /// [`remove_origin_providers`](Self::remove_origin_providers)) so an + /// evicted site's slot doesn't linger forever. A tenant whose spend + /// counter becomes entirely empty as a result is pruned from the map, to + /// bound its long-term growth as origins churn. + pub fn remove_origin_tenant_spend(&mut self, origin_site: &str) { + self.tenant_spend.retain(|_, counter| { + counter.remove_slot(origin_site); + counter.total() > 0 + }); + } + + /// Increment this site's local slot of a tenant's spend counter. + /// + /// Creates the tenant's counter if this is the first spend recorded for + /// it. `amount_cents` is in cents (see [`GridStateSnapshot::tenant_spend`]). + pub fn increment_tenant_spend(&mut self, tenant_id: &str, amount_cents: u64) { + self.tenant_spend + .entry(tenant_id.to_owned()) + .or_insert_with(|| GCounter::new(self.site_id.clone())) + .increment(amount_cents); } /// Replace provider records owned by one authoritative origin snapshot. @@ -555,4 +653,380 @@ mod tests { "other origins' providers must be preserved" ); } + + // ----------------------------------------------------------------------- + // tenant_spend tests (C1-C8) + // ----------------------------------------------------------------------- + + #[test] + fn new_snapshot_has_empty_tenant_spend() { + let snap = GridStateSnapshot::new("site-p".to_owned()); + assert!( + snap.tenant_spend.is_empty(), + "new snapshot must start with no tenant spend" + ); + } + + #[test] + fn merge_adds_tenant_present_only_in_other() { + let mut a = GridStateSnapshot::new("site-a".to_owned()); + let mut b = GridStateSnapshot::new("site-b".to_owned()); + b.increment_tenant_spend("tenant-x", 500); + + a.merge(&b); + + assert_eq!( + a.tenant_spend + .get("tenant-x") + .unwrap_or_else(|| std::process::abort()) + .total(), + 500, + "tenant absent from self before merge must be added from other" + ); + } + + #[test] + fn merge_tenant_spend_takes_per_site_max() { + let mut a = GridStateSnapshot::new("site-a".to_owned()); + a.increment_tenant_spend("tenant-x", 100); + let mut b = GridStateSnapshot::new("site-b".to_owned()); + b.increment_tenant_spend("tenant-x", 200); + + a.merge(&b); + + assert_eq!( + a.tenant_spend + .get("tenant-x") + .unwrap_or_else(|| std::process::abort()) + .total(), + 300, + "per-site slots must sum via GCounter's max-per-slot merge rule" + ); + } + + #[test] + fn merge_tenant_spend_is_idempotent() { + let mut a = GridStateSnapshot::new("site-a".to_owned()); + a.increment_tenant_spend("tenant-x", 100); + let duplicate = a.clone(); + + a.merge(&duplicate); + + assert_eq!( + a.tenant_spend + .get("tenant-x") + .unwrap_or_else(|| std::process::abort()) + .total(), + 100, + "merging a duplicate snapshot must not double-count tenant spend" + ); + } + + #[test] + fn merge_tenant_spend_is_commutative() { + let mut a = GridStateSnapshot::new("site-a".to_owned()); + a.increment_tenant_spend("tenant-x", 100); + let mut b = GridStateSnapshot::new("site-b".to_owned()); + b.increment_tenant_spend("tenant-x", 200); + + let mut ab = a.clone(); + ab.merge(&b); + let mut ba = b; + ba.merge(&a); + + assert_eq!( + ab.tenant_spend + .get("tenant-x") + .unwrap_or_else(|| std::process::abort()) + .total(), + ba.tenant_spend + .get("tenant-x") + .unwrap_or_else(|| std::process::abort()) + .total(), + "tenant spend merge must be commutative" + ); + } + + #[test] + fn merge_tenant_spend_is_associative() { + let mut a = GridStateSnapshot::new("site-a".to_owned()); + a.increment_tenant_spend("tenant-x", 100); + let mut b = GridStateSnapshot::new("site-b".to_owned()); + b.increment_tenant_spend("tenant-x", 200); + let mut c = GridStateSnapshot::new("site-c".to_owned()); + c.increment_tenant_spend("tenant-x", 300); + + let mut ab_then_c = a.clone(); + ab_then_c.merge(&b); + ab_then_c.merge(&c); + + let mut bc = b; + bc.merge(&c); + let mut a_then_bc = a; + a_then_bc.merge(&bc); + + assert_eq!( + ab_then_c + .tenant_spend + .get("tenant-x") + .unwrap_or_else(|| std::process::abort()) + .total(), + a_then_bc + .tenant_spend + .get("tenant-x") + .unwrap_or_else(|| std::process::abort()) + .total(), + "tenant spend merge must be associative" + ); + } + + #[test] + fn tenant_spend_survives_bincode_round_trip() { + let mut snap = GridStateSnapshot::new("site-p".to_owned()); + snap.increment_tenant_spend("tenant-x", 4200); + + let bytes = + bincode::serde::encode_to_vec(&snap, bincode::config::standard()).unwrap_or_else(|_| std::process::abort()); + let (restored, _len): (GridStateSnapshot, usize) = + bincode::serde::decode_from_slice(&bytes, bincode::config::standard()) + .unwrap_or_else(|_| std::process::abort()); + + assert_eq!( + restored + .tenant_spend + .get("tenant-x") + .unwrap_or_else(|| std::process::abort()) + .total(), + 4200, + "tenant_spend must survive bincode round-trip" + ); + } + + #[test] + fn tenant_spend_json_decode_missing_field_defaults_to_empty() { + // Simulates a peer running an older build whose wire payload predates + // this field: start from a real serialized snapshot (avoiding any + // guesswork about OrSet's/ProviderState's internal JSON shape), strip + // the key, and confirm deserialization still succeeds and defaults. + let mut snap = GridStateSnapshot::new("site-p".to_owned()); + snap.increment_tenant_spend("tenant-x", 100); + let mut json = serde_json::to_value(&snap).unwrap_or_else(|_| std::process::abort()); + json.as_object_mut() + .unwrap_or_else(|| std::process::abort()) + .remove("tenant_spend"); + + let restored: GridStateSnapshot = serde_json::from_value(json).unwrap_or_else(|_| std::process::abort()); + + assert!( + restored.tenant_spend.is_empty(), + "missing tenant_spend field must default to an empty map, not fail to deserialize" + ); + } + + // ----------------------------------------------------------------------- + // merge_tenant_spend_from_origin: wire-ingest trust boundary (security) + // ----------------------------------------------------------------------- + + #[test] + fn merge_tenant_spend_from_origin_accepts_the_claimed_origins_own_slot() { + let mut local = GridStateSnapshot::new("site-local".to_owned()); + let mut incoming = BTreeMap::new(); + let mut counter = GCounter::new("site-p".to_owned()); + counter.increment(500); + incoming.insert("tenant-x".to_owned(), counter); + + local.merge_tenant_spend_from_origin("site-p", &incoming); + + assert_eq!( + local + .tenant_spend + .get("tenant-x") + .unwrap_or_else(|| std::process::abort()) + .total(), + 500, + "the claimed origin's own slot must be merged normally" + ); + } + + #[test] + fn merge_tenant_spend_from_origin_drops_forged_foreign_slots() { + // Security: a broadcast claiming origin "site-a" must not be able to + // smuggle in an inflated slot for "site-b" and have it accepted as + // site-b's real contribution — that would let one compromised/buggy + // peer forge another site's recorded spend mesh-wide. + let mut local = GridStateSnapshot::new("site-local".to_owned()); + let mut forged = BTreeMap::new(); + let mut counter = GCounter::new("site-a".to_owned()); + counter.increment(10); // site-a's genuine contribution + let mut fake_site_b = GCounter::new("site-b".to_owned()); + fake_site_b.increment(u64::MAX); + counter.merge(&fake_site_b); // origin locally folds in a forged foreign slot + forged.insert("tenant-x".to_owned(), counter); + + local.merge_tenant_spend_from_origin("site-a", &forged); + + assert_eq!( + local + .tenant_spend + .get("tenant-x") + .unwrap_or_else(|| std::process::abort()) + .total(), + 10, + "only the claimed origin's own slot must be accepted; the forged site-b slot must be dropped" + ); + } + + #[test] + fn merge_tenant_spend_from_origin_ignores_tenant_the_claimed_origin_never_contributed_to() { + // The incoming counter carries only a foreign slot (no slot at all + // for the claimed origin) — after stripping the forged foreign slot + // there is nothing genuine left to merge, so the tenant must not be + // created locally at all (not even as a zero entry). + let mut local = GridStateSnapshot::new("site-local".to_owned()); + let mut forged = BTreeMap::new(); + let mut foreign_only = GCounter::new("site-b".to_owned()); + foreign_only.increment(999); + forged.insert("tenant-x".to_owned(), foreign_only); + + local.merge_tenant_spend_from_origin("site-a", &forged); + + assert!( + !local.tenant_spend.contains_key("tenant-x"), + "a tenant with zero genuine contribution from the claimed origin must not be created" + ); + } + + #[test] + fn merge_tenant_spend_from_origin_repeated_calls_are_idempotent() { + let mut local = GridStateSnapshot::new("site-local".to_owned()); + let mut incoming = BTreeMap::new(); + let mut counter = GCounter::new("site-p".to_owned()); + counter.increment(500); + incoming.insert("tenant-x".to_owned(), counter); + + local.merge_tenant_spend_from_origin("site-p", &incoming); + local.merge_tenant_spend_from_origin("site-p", &incoming); + + assert_eq!( + local + .tenant_spend + .get("tenant-x") + .unwrap_or_else(|| std::process::abort()) + .total(), + 500, + "re-merging the same origin broadcast must not double-count (max-per-slot semantics)" + ); + } + + // ----------------------------------------------------------------------- + // remove_origin_tenant_spend + tenant-count bound (security: unbounded growth) + // ----------------------------------------------------------------------- + + #[test] + fn remove_origin_tenant_spend_strips_only_that_origins_slot() { + let mut snap = GridStateSnapshot::new("site-local".to_owned()); + let mut incoming_a = BTreeMap::new(); + let mut counter_a = GCounter::new("site-a".to_owned()); + counter_a.increment(100); + incoming_a.insert("tenant-x".to_owned(), counter_a); + snap.merge_tenant_spend_from_origin("site-a", &incoming_a); + + let mut incoming_b = BTreeMap::new(); + let mut counter_b = GCounter::new("site-b".to_owned()); + counter_b.increment(200); + incoming_b.insert("tenant-x".to_owned(), counter_b); + snap.merge_tenant_spend_from_origin("site-b", &incoming_b); + + snap.remove_origin_tenant_spend("site-a"); + + assert_eq!( + snap.tenant_spend + .get("tenant-x") + .unwrap_or_else(|| std::process::abort()) + .total(), + 200, + "evicting site-a must remove only its own contribution, leaving site-b's spend intact" + ); + } + + #[test] + fn remove_origin_tenant_spend_prunes_tenant_entry_once_fully_empty() { + let mut snap = GridStateSnapshot::new("site-local".to_owned()); + let mut incoming = BTreeMap::new(); + let mut counter = GCounter::new("site-a".to_owned()); + counter.increment(100); + incoming.insert("tenant-x".to_owned(), counter); + snap.merge_tenant_spend_from_origin("site-a", &incoming); + + snap.remove_origin_tenant_spend("site-a"); + + assert!( + !snap.tenant_spend.contains_key("tenant-x"), + "a tenant with no remaining site contributions must be pruned entirely, not left as a zero entry" + ); + } + + #[test] + fn merge_tenant_spend_from_origin_refuses_new_tenants_once_at_capacity() { + // Security: bound the number of distinct tenant_id keys accepted from + // gossip so a malicious/buggy origin flooding arbitrary tenant_ids + // cannot grow tenant_spend without bound. Already-tracked tenants may + // still accumulate; only brand-new tenant_ids are refused at capacity. + let mut snap = GridStateSnapshot::new("site-local".to_owned()); + for i in 0..MAX_TRACKED_TENANTS { + let mut incoming = BTreeMap::new(); + let mut counter = GCounter::new("site-a".to_owned()); + counter.increment(1); + incoming.insert(format!("tenant-{i}"), counter); + snap.merge_tenant_spend_from_origin("site-a", &incoming); + } + assert_eq!( + snap.tenant_spend.len(), + MAX_TRACKED_TENANTS, + "precondition: at capacity" + ); + + let mut overflow = BTreeMap::new(); + let mut counter = GCounter::new("site-a".to_owned()); + counter.increment(1); + overflow.insert("tenant-overflow".to_owned(), counter); + snap.merge_tenant_spend_from_origin("site-a", &overflow); + + assert_eq!( + snap.tenant_spend.len(), + MAX_TRACKED_TENANTS, + "a brand-new tenant_id must be refused once the map is at its hard bound" + ); + assert!( + !snap.tenant_spend.contains_key("tenant-overflow"), + "the refused tenant_id must not be present at all" + ); + } + + #[test] + fn merge_tenant_spend_from_origin_still_updates_already_tracked_tenant_at_capacity() { + let mut snap = GridStateSnapshot::new("site-local".to_owned()); + for i in 0..MAX_TRACKED_TENANTS { + let mut incoming = BTreeMap::new(); + let mut counter = GCounter::new("site-a".to_owned()); + counter.increment(1); + incoming.insert(format!("tenant-{i}"), counter); + snap.merge_tenant_spend_from_origin("site-a", &incoming); + } + + let mut more = BTreeMap::new(); + let mut counter = GCounter::new("site-a".to_owned()); + counter.increment(99); + more.insert("tenant-0".to_owned(), counter); + snap.merge_tenant_spend_from_origin("site-a", &more); + + assert_eq!( + snap.tenant_spend + .get("tenant-0") + .unwrap_or_else(|| std::process::abort()) + .total(), + 99, + "an already-tracked tenant must still accept updates while the map is at capacity" + ); + } } diff --git a/deploy/crds/gridnetwork.yaml b/deploy/crds/gridnetwork.yaml index cbc177a..6d62104 100644 --- a/deploy/crds/gridnetwork.yaml +++ b/deploy/crds/gridnetwork.yaml @@ -34,6 +34,42 @@ spec: Defines the grid's seed peers, gateway associations, SWIM tuning, and TLS secret references. properties: + budgetPolicy: + description: |- + Budget policy configuration for per-tenant spend tracking. + + Selects which tenants Grid tracks cumulative spend for. See + [`BudgetPolicyConfig`] for what this does and does not do. + + **Default (absent):** no tenants are tracked; `budgetStatus` is + always empty. + nullable: true + properties: + tenants: + default: [] + description: Per-tenant budget caps. + items: + description: Per-tenant budget cap declaration. + properties: + capUsd: + description: |- + Maximum cumulative spend in USD before this tenant is considered over budget. + + **Minimum value:** `0`. The generated CRD schema rejects negative caps; + [`validate_budget_policy`] additionally rejects `NaN`/infinite values + that the schema's numeric minimum does not catch. + format: double + minimum: 0.0 + type: number + tenantId: + description: Tenant identifier. Must be non-empty and unique within the policy. + type: string + required: + - capUsd + - tenantId + type: object + type: array + type: object gatewayRefs: default: [] description: References to Praxis Gateways that participate in this grid. @@ -416,6 +452,48 @@ spec: description: Observed status of a [`GridNetwork`]. nullable: true properties: + budgetStatus: + description: |- + Per-tenant budget status, derived from `spec.budgetPolicy` and merged + cross-site CRDT spend state. + + Empty when `budgetPolicy` is absent. This is a status signal only — + Grid does not enforce budget limits itself (see [`BudgetPolicyConfig`]). + items: + description: |- + Per-tenant budget status derived from policy + merged CRDT spend state. + + Populated in [`GridNetworkStatus::budget_status`] for every tenant + declared in `spec.budgetPolicy`, regardless of whether spend has been + recorded for that tenant yet. This is a status signal only — Grid does + not enforce budget limits (see [`BudgetPolicyConfig`] doc). + properties: + capUsd: + description: Budget cap for this tenant, in USD, copied from `spec.budgetPolicy`. + format: double + type: number + spendRatio: + description: '`spend_usd / cap_usd`, clamped to `0.0..=1.0`. See [`spend_ratio`].' + format: double + type: number + spendUsd: + description: |- + Cumulative spend observed for this tenant, in USD. + + Converged across all sites that have merged CRDT state for this + tenant; may lag briefly during a partition (see [`GCounter`]). + format: double + type: number + tenantId: + description: Tenant identifier, matching `spec.budgetPolicy.tenants[].tenantId`. + type: string + required: + - capUsd + - spendRatio + - spendUsd + - tenantId + type: object + type: array connectedSites: default: 0 description: Number of connected (Active) sites. diff --git a/docs/architecture/crds.md b/docs/architecture/crds.md index 4dbd5c4..3edf55f 100644 --- a/docs/architecture/crds.md +++ b/docs/architecture/crds.md @@ -54,12 +54,18 @@ spec: swimKeyRef: name: swim-key namespace: praxis-system + budgetPolicy: # optional; absent means no tenants are tracked + tenants: + - tenantId: tenant-a + capUsd: 100.0 + - tenantId: tenant-b + capUsd: 250.0 ``` **Phases**: Pending → Initializing → Active → Degraded **Status fields**: `gridId`, `connectedSites`, `distributedProviderCount`, -`observedGeneration`, `phase`, `consumerConfigStatus[]` +`observedGeneration`, `phase`, `consumerConfigStatus[]`, `budgetStatus[]` `distributedProviderCount` reflects the number of remote `InferenceProvider` records received from peer sites via CRDT broadcast. Local providers and records @@ -69,6 +75,35 @@ from other `GridNetwork`s are excluded from the count. `consumerConfig.enabled: true`, reporting the outcome of the most recent render/apply attempt. +### Tenant budget tracking + +`budgetPolicy.tenants[]` opts individual tenants into cumulative spend +tracking. Grid merges each site's locally recorded spend for a tenant into a +per-tenant CRDT counter (a `GCounter`, one slot per originating site) that is +gossiped over SWIM alongside provider state, so the reported total reflects +spend recorded anywhere in the grid, not just the local site. + +For every tenant declared in `budgetPolicy`, `budgetStatus[]` reports: + +- `tenantId` — matches `budgetPolicy.tenants[].tenantId` +- `capUsd` — copied from the policy, in USD +- `spendUsd` — the converged cross-site total, in USD +- `spendRatio` — `spendUsd / capUsd`, for at-a-glance dashboarding + +`budgetStatus[]` is a status **signal only**. Grid does not itself degrade or +reject traffic when a tenant's `spendRatio` reaches or exceeds `1.0` — that +enforcement decision is expected to live in a gateway-side policy filter +(cross-repo, `praxis-ai`), the same split used for `provider_route` +authorization. Real per-request tenant attribution also depends on +upstream work (`praxis-ai#130`/`praxis-ai#104`) and does not exist yet. + +Because `budgetStatus[]` is visible to any caller with read access to the +`GridNetwork` resource, and Kubernetes RBAC is not field-level, a reader +authorized to view one tenant's status can see every other tracked tenant's +spend on the same `GridNetwork`. See +[`grid#48`](https://github.com/praxis-proxy/grid/issues/48) for the +options under consideration if per-tenant confidentiality is required. + | Field | Type | Meaning | |---|---|---| | `gatewayName` | string | Name of the gateway reference | diff --git a/operator/src/controller/grid_network.rs b/operator/src/controller/grid_network.rs index 3528be5..418c452 100644 --- a/operator/src/controller/grid_network.rs +++ b/operator/src/controller/grid_network.rs @@ -27,7 +27,7 @@ use crate::{ crd::{ grid_network::{ ConsumerConfig, ConsumerConfigPhase, ConsumerConfigStatus, GatewayRef, GridNetwork, GridNetworkPhase, - GridNetworkStatus, OverlayPhase, OverlayRevisionStatus, TransportMode, + GridNetworkStatus, OverlayPhase, OverlayRevisionStatus, TenantBudgetStatus, TransportMode, }, grid_site::{GridSite, GridSitePhase, GridSiteStatus}, inference_provider::InferenceProvider, @@ -187,6 +187,24 @@ fn grid_network_name(network: &GridNetwork) -> Result<&str, OperatorError> { .ok_or_else(|| OperatorError::InvalidResource("GridNetwork missing metadata.name".into())) } +/// Reject a [`GridNetwork`] whose `budgetPolicy` fails validation, before any +/// other reconcile work begins. +/// +/// Pure and I/O-free (network fields only), so the reconcile-time wiring this +/// guards is exercised directly by unit tests without a live or mocked +/// Kubernetes client, per this repo's convention of preferring pure decision +/// functions for reconciliation logic (`docs/conventions.md`). The CRD +/// schema's numeric minimum on `capUsd` already rejects negative values at +/// admission time; this is the defensive second layer for `NaN`/infinite +/// caps and blank/duplicate `tenantId`s that the schema cannot express. +fn reject_invalid_budget_policy(network: &GridNetwork) -> Result<(), OperatorError> { + let Some(policy) = network.spec.budget_policy.as_ref() else { + return Ok(()); + }; + crate::crd::grid_network::validate_budget_policy(policy) + .map_err(|error| OperatorError::InvalidResource(format!("invalid budgetPolicy: {error}"))) +} + // --------------------------------------------------------------------------- // Reconcile // --------------------------------------------------------------------------- @@ -208,6 +226,7 @@ fn grid_network_name(network: &GridNetwork) -> Result<&str, OperatorError> { )] pub async fn reconcile(network: Arc, ctx: Arc) -> Result { let name = grid_network_name(&network)?; + reject_invalid_budget_policy(&network)?; info!(name, "reconciling GridNetwork"); @@ -303,6 +322,18 @@ pub async fn reconcile(network: Arc, ctx: Arc) -> Resu 0 }; + // Resolve per-tenant budget status from the merged CRDT spend state, if any. + // Empty tenant_spend (SWIM disabled, or no spend broadcast received yet) is + // indistinguishable here from "no spend recorded" — resolve_budget_statuses + // still emits a zero-spend entry for every policy-declared tenant. + let tenant_spend = ctx + .swim + .as_ref() + .map(|swim| swim.state_snapshot().tenant_spend) + .unwrap_or_default(); + let budget_statuses = + crate::crd::grid_network::resolve_budget_statuses(network.spec.budget_policy.as_ref(), &tenant_spend); + update_status( &network, client, @@ -312,6 +343,7 @@ pub async fn reconcile(network: Arc, ctx: Arc) -> Resu distributed_provider_count, consumer_config_statuses, overlay_statuses, + budget_statuses, ) .await?; @@ -1498,6 +1530,9 @@ pub(crate) fn apply_swim_staleness_override( /// CRDT state broadcasts. Both are `0` when SWIM is disabled. /// `consumer_config_statuses` holds per-gateway render/apply outcomes for /// gateways with `consumerConfig.enabled: true`; empty when no gateways opted in. +/// `budget_statuses` holds per-tenant spend status derived from +/// `spec.budgetPolicy` and merged CRDT spend state; empty when `budgetPolicy` +/// is absent. /// /// [`Alive`]: MemberStatus::Alive #[expect( @@ -1513,6 +1548,7 @@ async fn update_status( distributed_provider_count: u32, consumer_config_statuses: Vec, overlay_statuses: Vec, + budget_statuses: Vec, ) -> Result<(), OperatorError> { let name = grid_network_name(network)?; @@ -1527,6 +1563,7 @@ async fn update_status( phase: phase.clone(), consumer_config_status: consumer_config_statuses, overlay_status: overlay_statuses, + budget_status: budget_statuses, }; if !grid_network_status_needs_update(network.status.as_ref(), &status) { @@ -2194,7 +2231,10 @@ fn parse_metrics_refresh_interval(value: &str) -> Result InferenceProvider { serde_json::from_value(serde_json::json!({ @@ -2320,6 +2360,85 @@ mod tests { .unwrap_or_else(|_| std::process::abort()) } + // ----------------------------------------------------------------------- + // reject_invalid_budget_policy + // ----------------------------------------------------------------------- + + fn network_with_budget_policy(tenants: Vec) -> GridNetwork { + let mut network = base_network(); + network.spec.budget_policy = Some(BudgetPolicyConfig { tenants }); + network + } + + fn tenant(tenant_id: &str, cap_usd: f64) -> TenantBudgetConfig { + TenantBudgetConfig { + tenant_id: tenant_id.to_owned(), + cap_usd, + } + } + + #[test] + fn reject_invalid_budget_policy_accepts_absent_policy() { + let network = base_network(); + assert!( + reject_invalid_budget_policy(&network).is_ok(), + "a GridNetwork with no budgetPolicy at all must not be rejected" + ); + } + + #[test] + fn reject_invalid_budget_policy_accepts_valid_policy() { + let network = network_with_budget_policy(vec![tenant("tenant-a", 100.0), tenant("tenant-b", 250.0)]); + assert!( + reject_invalid_budget_policy(&network).is_ok(), + "distinct positive caps and non-empty tenant ids must be accepted" + ); + } + + #[test] + fn reject_invalid_budget_policy_rejects_blank_tenant_id() { + let network = network_with_budget_policy(vec![tenant("", 100.0)]); + let Err(error) = reject_invalid_budget_policy(&network) else { + std::process::abort() + }; + assert!( + error.to_string().contains("budgetPolicy"), + "error must identify the budgetPolicy as the invalid field, got: {error}" + ); + } + + #[test] + fn reject_invalid_budget_policy_rejects_duplicate_tenant_id() { + let network = network_with_budget_policy(vec![tenant("tenant-a", 100.0), tenant("tenant-a", 200.0)]); + let Err(error) = reject_invalid_budget_policy(&network) else { + std::process::abort() + }; + assert!( + error.to_string().contains("tenant-a"), + "error must name the offending tenant_id, got: {error}" + ); + } + + #[test] + fn reject_invalid_budget_policy_rejects_negative_cap() { + let network = network_with_budget_policy(vec![tenant("tenant-a", -5.0)]); + assert!( + reject_invalid_budget_policy(&network).is_err(), + "negative capUsd must be rejected even though the CRD schema minimum should already catch this before reconcile" + ); + } + + #[test] + fn reject_invalid_budget_policy_rejects_non_finite_cap() { + for bad_cap in [f64::NAN, f64::INFINITY, f64::NEG_INFINITY] { + let network = network_with_budget_policy(vec![tenant("tenant-a", bad_cap)]); + assert!( + reject_invalid_budget_policy(&network).is_err(), + "non-finite capUsd ({bad_cap}) must be rejected" + ); + } + } + fn alive_snapshot(count: usize) -> MembershipSnapshot { MembershipSnapshot { members: (0..count) @@ -3027,6 +3146,7 @@ mod tests { phase: GridNetworkPhase::Active, consumer_config_status: Vec::new(), overlay_status: Vec::new(), + budget_status: Vec::new(), }; assert!(!grid_network_status_needs_update(Some(&baseline), &baseline)); diff --git a/operator/src/crd/grid_network.rs b/operator/src/crd/grid_network.rs index 9b0da00..a0cd835 100644 --- a/operator/src/crd/grid_network.rs +++ b/operator/src/crd/grid_network.rs @@ -3,6 +3,9 @@ //! The top-level tenancy boundary for the AI Grid. A cluster //! can host multiple `GridNetworks` for multi-tenancy. +use std::collections::BTreeMap; + +use crdt::GCounter; use kube::CustomResource; use schemars::JsonSchema; use serde::{Deserialize, Serialize}; @@ -150,6 +153,214 @@ pub fn resolve_scoring_weights(policy: Option<&ScoringPolicyConfig>) -> scoring: .weights() } +// --------------------------------------------------------------------------- +// Budget policy +// --------------------------------------------------------------------------- + +/// Per-tenant budget cap declaration. +#[derive(Clone, Debug, Deserialize, JsonSchema, PartialEq, Serialize)] +#[serde(rename_all = "camelCase", deny_unknown_fields)] +#[schemars(deny_unknown_fields)] +pub struct TenantBudgetConfig { + /// Tenant identifier. Must be non-empty and unique within the policy. + pub tenant_id: String, + + /// Maximum cumulative spend in USD before this tenant is considered over budget. + /// + /// **Minimum value:** `0`. The generated CRD schema rejects negative caps; + /// [`validate_budget_policy`] additionally rejects `NaN`/infinite values + /// that the schema's numeric minimum does not catch. + #[schemars(range(min = 0.0))] + pub cap_usd: f64, +} + +/// Budget policy configuration for per-tenant spend tracking. +/// +/// Declares the tenants Grid should track cumulative spend for. Grid tracks +/// and cross-site-converges the spend signal (via G-Counter CRDT, see +/// [`crdt::GridStateSnapshot::tenant_spend`]) and exposes it in +/// [`GridNetworkStatus::budget_status`]; it does **not** enforce budget +/// limits itself — degrade/reject decisions are a gateway-side `praxis-ai` +/// policy-filter concern, not a Grid-side one. +/// +/// **Default (absent):** no tenants are tracked; `budgetStatus` is always empty. +#[derive(Clone, Debug, Default, Deserialize, JsonSchema, Serialize)] +#[serde(rename_all = "camelCase", deny_unknown_fields)] +#[schemars(deny_unknown_fields)] +pub struct BudgetPolicyConfig { + /// Per-tenant budget caps. + #[serde(default)] + pub tenants: Vec, +} + +/// Reason a [`BudgetPolicyConfig`] failed validation. +/// +/// The CRD schema's numeric minimum on `capUsd` (see [`TenantBudgetConfig`]) +/// already rejects negative values at admission time; [`validate_budget_policy`] +/// is a defensive second layer for callers that construct or deserialize a +/// [`BudgetPolicyConfig`] outside the Kubernetes API path. +#[derive(Clone, Debug, Eq, PartialEq, thiserror::Error)] +pub enum BudgetPolicyValidationError { + /// A tenant's `capUsd` is negative. + #[error("tenant {tenant_id:?} has a negative capUsd")] + NegativeCap { + /// Offending tenant identifier. + tenant_id: String, + }, + /// A tenant's `capUsd` is `NaN` or infinite. + #[error("tenant {tenant_id:?} has a non-finite capUsd")] + NonFiniteCap { + /// Offending tenant identifier. + tenant_id: String, + }, + /// The same `tenantId` appears more than once. + #[error("tenant id {tenant_id:?} appears more than once")] + DuplicateTenant { + /// The repeated tenant identifier. + tenant_id: String, + }, + /// A tenant entry has a blank (empty or whitespace-only) `tenantId`. + #[error("a tenant entry has a blank tenantId")] + BlankTenantId, +} + +/// Validate a [`BudgetPolicyConfig`]. +/// +/// # Errors +/// +/// Returns [`BudgetPolicyValidationError`] for the first invalid tenant entry +/// found, in declaration order: a blank `tenantId`, a duplicate `tenantId`, a +/// negative `capUsd`, or a non-finite `capUsd`. +pub fn validate_budget_policy(policy: &BudgetPolicyConfig) -> Result<(), BudgetPolicyValidationError> { + let mut seen_tenant_ids = std::collections::HashSet::new(); + for tenant in &policy.tenants { + if tenant.tenant_id.trim().is_empty() { + return Err(BudgetPolicyValidationError::BlankTenantId); + } + if !seen_tenant_ids.insert(tenant.tenant_id.as_str()) { + return Err(BudgetPolicyValidationError::DuplicateTenant { + tenant_id: tenant.tenant_id.clone(), + }); + } + if !tenant.cap_usd.is_finite() { + return Err(BudgetPolicyValidationError::NonFiniteCap { + tenant_id: tenant.tenant_id.clone(), + }); + } + if tenant.cap_usd < 0.0 { + return Err(BudgetPolicyValidationError::NegativeCap { + tenant_id: tenant.tenant_id.clone(), + }); + } + } + Ok(()) +} + +/// Convert a G-Counter total (cents, `u64`) into USD (`f64`). +#[expect( + clippy::cast_precision_loss, + reason = "budget ratio is inherently approximate under partition; see GCounter docs" +)] +pub(crate) fn cents_to_usd(cents: u64) -> f64 { + cents as f64 / 100.0 +} + +/// Convert a tenant's cumulative spend counter into a cap-relative ratio. +/// +/// `tenant_spend.total()` is denominated in cents (see +/// [`crdt::GridStateSnapshot::tenant_spend`]); `cap_usd` is dollars. The +/// result is always clamped to `0.0..=1.0`: +/// +/// - `cap_usd <= 0.0` (including non-finite) is treated defensively as "no budget available" and always returns `1.0`, +/// regardless of spend. The CRD schema and [`validate_budget_policy`] should already prevent this, but a caller +/// bypassing both must not panic or divide by zero. +/// - Spend above the cap clamps to `1.0` rather than exceeding it — a real possibility, not a bug: G-Counter is +/// monotonic and an individual site sees only a lower bound under partition, so local overspend is expected. +#[must_use] +pub fn spend_ratio(tenant_spend: &GCounter, cap_usd: f64) -> f64 { + if !cap_usd.is_finite() || cap_usd <= 0.0 { + return 1.0; + } + (cents_to_usd(tenant_spend.total()) / cap_usd).clamp(0.0, 1.0) +} + +/// Per-tenant budget status derived from policy + merged CRDT spend state. +/// +/// Populated in [`GridNetworkStatus::budget_status`] for every tenant +/// declared in `spec.budgetPolicy`, regardless of whether spend has been +/// recorded for that tenant yet. This is a status signal only — Grid does +/// not enforce budget limits (see [`BudgetPolicyConfig`] doc). +#[derive(Clone, Debug, Deserialize, JsonSchema, PartialEq, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct TenantBudgetStatus { + /// Tenant identifier, matching `spec.budgetPolicy.tenants[].tenantId`. + pub tenant_id: String, + + /// Cumulative spend observed for this tenant, in USD. + /// + /// Converged across all sites that have merged CRDT state for this + /// tenant; may lag briefly during a partition (see [`GCounter`]). + pub spend_usd: f64, + + /// Budget cap for this tenant, in USD, copied from `spec.budgetPolicy`. + pub cap_usd: f64, + + /// `spend_usd / cap_usd`, clamped to `0.0..=1.0`. See [`spend_ratio`]. + pub spend_ratio: f64, +} + +/// Build one tenant's status entry. +/// +/// `counter` is `None` when no spend has been recorded for this tenant yet; +/// in that case both `spend_usd` and `spend_ratio` are `0.0` rather than +/// delegating to [`spend_ratio`] (which would read a non-positive cap as +/// maxed — not the right answer for "no traffic yet"). +fn tenant_budget_status(tenant: &TenantBudgetConfig, counter: Option<&GCounter>) -> TenantBudgetStatus { + let (spend_usd, ratio) = counter.map_or((0.0, 0.0), |counter| { + (cents_to_usd(counter.total()), spend_ratio(counter, tenant.cap_usd)) + }); + TenantBudgetStatus { + tenant_id: tenant.tenant_id.clone(), + spend_usd, + cap_usd: tenant.cap_usd, + spend_ratio: ratio, + } +} + +/// Assemble per-tenant budget status from policy and merged CRDT spend state. +/// +/// Driven by `policy.tenants`, not by `tenant_spend`: a tenant declared in +/// the policy but with no recorded spend yet still gets an entry +/// (`spendUsd: 0.0`); CRDT spend recorded for a tenant no longer declared in +/// the policy is silently excluded — the policy is the source of truth for +/// which tenants are tracked. Output is sorted by `tenantId` for +/// deterministic status ordering. +#[must_use] +pub fn tenant_spend_status( + policy: &BudgetPolicyConfig, + tenant_spend: &BTreeMap, +) -> Vec { + let mut statuses: Vec = policy + .tenants + .iter() + .map(|tenant| tenant_budget_status(tenant, tenant_spend.get(&tenant.tenant_id))) + .collect(); + statuses.sort_by(|a, b| a.tenant_id.cmp(&b.tenant_id)); + statuses +} + +/// Resolve tenant budget statuses for [`GridNetworkStatus::budget_status`]. +/// +/// `policy` is `None` when `spec.budgetPolicy` is absent — no tenants are +/// tracked, so the result is always empty in that case. +#[must_use] +pub fn resolve_budget_statuses( + policy: Option<&BudgetPolicyConfig>, + tenant_spend: &BTreeMap, +) -> Vec { + policy.map_or_else(Vec::new, |policy| tenant_spend_status(policy, tenant_spend)) +} + // --------------------------------------------------------------------------- // Spec // --------------------------------------------------------------------------- @@ -238,6 +449,16 @@ pub struct GridNetworkSpec { #[serde(default, skip_serializing_if = "Option::is_none")] pub metrics_refresh_interval: Option, + /// Budget policy configuration for per-tenant spend tracking. + /// + /// Selects which tenants Grid tracks cumulative spend for. See + /// [`BudgetPolicyConfig`] for what this does and does not do. + /// + /// **Default (absent):** no tenants are tracked; `budgetStatus` is + /// always empty. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub budget_policy: Option, + /// Maximum age in seconds before a stale (`fresh=false`) remote routing /// candidate is removed from the overlay. /// @@ -547,7 +768,7 @@ pub struct SecretRef { // --------------------------------------------------------------------------- /// Observed status of a [`GridNetwork`]. -#[derive(Clone, Debug, Default, Deserialize, Eq, JsonSchema, PartialEq, Serialize)] +#[derive(Clone, Debug, Default, Deserialize, JsonSchema, PartialEq, Serialize)] #[serde(rename_all = "camelCase")] pub struct GridNetworkStatus { /// Number of connected (Active) sites. @@ -592,6 +813,14 @@ pub struct GridNetworkStatus { /// the last successfully distributed revision. #[serde(default, skip_serializing_if = "Vec::is_empty")] pub overlay_status: Vec, + + /// Per-tenant budget status, derived from `spec.budgetPolicy` and merged + /// cross-site CRDT spend state. + /// + /// Empty when `budgetPolicy` is absent. This is a status signal only — + /// Grid does not enforce budget limits itself (see [`BudgetPolicyConfig`]). + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub budget_status: Vec, } /// Phase of an operator-generated consumer Praxis `ConfigMap` for one gateway. @@ -788,6 +1017,9 @@ impl Default for SwimConfig { #[cfg(test)] mod tests { + use std::collections::BTreeMap; + + use crdt::GCounter; use kube::CustomResourceExt as _; use super::*; @@ -1748,4 +1980,352 @@ mod tests { assert_weight(first.breakdown.queue_depth, 0.0, "queue must not contribute"); assert_weight(second.breakdown.queue_depth, 0.0, "queue must not contribute"); } + + // ----------------------------------------------------------------------- + // validate_budget_policy tests (A1-A6) + // ----------------------------------------------------------------------- + + fn tenant(id: &str, cap_usd: f64) -> TenantBudgetConfig { + TenantBudgetConfig { + tenant_id: id.to_owned(), + cap_usd, + } + } + + #[test] + fn validate_budget_policy_accepts_valid_config() { + let policy = BudgetPolicyConfig { + tenants: vec![tenant("tenant-a", 100.0), tenant("tenant-b", 250.0)], + }; + assert!( + validate_budget_policy(&policy).is_ok(), + "distinct positive caps and non-empty tenant ids must be valid" + ); + } + + #[test] + fn validate_budget_policy_rejects_negative_cap() { + let policy = BudgetPolicyConfig { + tenants: vec![tenant("tenant-a", -5.0)], + }; + assert_eq!( + validate_budget_policy(&policy), + Err(BudgetPolicyValidationError::NegativeCap { + tenant_id: "tenant-a".to_owned() + }), + "negative capUsd must be rejected" + ); + } + + #[test] + fn validate_budget_policy_rejects_non_finite_cap() { + for bad_cap in [f64::NAN, f64::INFINITY, f64::NEG_INFINITY] { + let policy = BudgetPolicyConfig { + tenants: vec![tenant("tenant-a", bad_cap)], + }; + assert_eq!( + validate_budget_policy(&policy), + Err(BudgetPolicyValidationError::NonFiniteCap { + tenant_id: "tenant-a".to_owned() + }), + "non-finite capUsd ({bad_cap}) must be rejected" + ); + } + } + + #[test] + fn validate_budget_policy_rejects_duplicate_tenant_id() { + let policy = BudgetPolicyConfig { + tenants: vec![tenant("tenant-a", 100.0), tenant("tenant-a", 200.0)], + }; + assert_eq!( + validate_budget_policy(&policy), + Err(BudgetPolicyValidationError::DuplicateTenant { + tenant_id: "tenant-a".to_owned() + }), + "duplicate tenantId must be rejected" + ); + } + + #[test] + fn validate_budget_policy_rejects_blank_tenant_id() { + for blank in ["", " "] { + let policy = BudgetPolicyConfig { + tenants: vec![tenant(blank, 100.0)], + }; + assert_eq!( + validate_budget_policy(&policy), + Err(BudgetPolicyValidationError::BlankTenantId), + "blank tenantId ({blank:?}) must be rejected" + ); + } + } + + #[test] + fn validate_budget_policy_accepts_empty_tenant_list() { + let policy = BudgetPolicyConfig { tenants: Vec::new() }; + assert!( + validate_budget_policy(&policy).is_ok(), + "an empty tenants list is a valid no-op policy" + ); + } + + // ----------------------------------------------------------------------- + // spend_ratio tests (B1-B6) + // ----------------------------------------------------------------------- + + fn spend_of(cents: u64) -> GCounter { + let mut counter = GCounter::new("site-a".to_owned()); + counter.increment(cents); + counter + } + + #[test] + fn spend_ratio_below_cap() { + // 50.00 spent against a 100.00 cap. + assert_weight(spend_ratio(&spend_of(5000), 100.0), 0.5, "spend/cap"); + } + + #[test] + fn spend_ratio_at_cap() { + assert_weight(spend_ratio(&spend_of(10_000), 100.0), 1.0, "spend == cap"); + } + + #[test] + fn spend_ratio_overspend_clamps_to_one() { + // 150.00 spent against a 100.00 cap must not exceed 1.0. + assert_weight(spend_ratio(&spend_of(15_000), 100.0), 1.0, "overspend must clamp"); + } + + #[test] + fn spend_ratio_zero_spend_is_zero() { + assert_weight( + spend_ratio(&spend_of(0), 100.0), + 0.0, + "zero spend against a positive cap", + ); + } + + #[test] + fn spend_ratio_non_positive_cap_is_always_one() { + for bad_cap in [0.0, -1.0, f64::NAN, f64::NEG_INFINITY] { + assert_weight( + spend_ratio(&spend_of(0), bad_cap), + 1.0, + &format!("cap_usd={bad_cap} must defensively read as maxed regardless of spend"), + ); + } + } + + #[test] + fn spend_ratio_near_u64_max_does_not_panic_and_clamps() { + let ratio = spend_ratio(&spend_of(u64::MAX), 100.0); + assert!(ratio.is_finite(), "must not produce NaN/inf from a huge u64 conversion"); + assert_weight(ratio, 1.0, "huge spend against a small cap must clamp to 1.0"); + } + + // ----------------------------------------------------------------------- + // BudgetPolicy CRD spec wiring tests (D1-D6) + // ----------------------------------------------------------------------- + + #[test] + fn budget_policy_absent_defaults_to_none() { + let json = serde_json::json!({ "gridId": "", "seeds": [] }); + let spec: GridNetworkSpec = serde_json::from_value(json).unwrap_or_else(|_| std::process::abort()); + assert!(spec.budget_policy.is_none(), "absent budgetPolicy must default to None"); + } + + #[test] + fn budget_policy_absent_not_serialized() { + let json = serde_json::json!({ "gridId": "", "seeds": [] }); + let spec: GridNetworkSpec = serde_json::from_value(json).unwrap_or_else(|_| std::process::abort()); + let serialized = serde_json::to_value(&spec).unwrap_or_else(|_| std::process::abort()); + assert!( + serialized.get("budgetPolicy").is_none(), + "absent budgetPolicy must not appear in serialized output" + ); + } + + #[test] + fn budget_policy_with_tenants_round_trips() { + let json = serde_json::json!({ + "gridId": "", + "seeds": [], + "budgetPolicy": { + "tenants": [ + { "tenantId": "tenant-a", "capUsd": 100.0 }, + { "tenantId": "tenant-b", "capUsd": 50.0 } + ] + } + }); + let spec: GridNetworkSpec = serde_json::from_value(json).unwrap_or_else(|_| std::process::abort()); + let policy = spec.budget_policy.unwrap_or_else(|| std::process::abort()); + assert_eq!(policy.tenants.len(), 2, "both tenants must round-trip"); + let first = policy.tenants.first().unwrap_or_else(|| std::process::abort()); + let second = policy.tenants.get(1).unwrap_or_else(|| std::process::abort()); + assert_eq!(first.tenant_id, "tenant-a"); + assert_weight(first.cap_usd, 100.0, "tenant-a capUsd"); + assert_eq!(second.tenant_id, "tenant-b"); + assert_weight(second.cap_usd, 50.0, "tenant-b capUsd"); + } + + #[test] + fn budget_policy_appears_in_crd_schema() { + let crd = crd_json(); + let schema = crd + .pointer("/spec/versions/0/schema/openAPIV3Schema/properties/spec/properties/budgetPolicy") + .unwrap_or_else(|| std::process::abort()); + assert!(schema.is_object(), "budgetPolicy must appear in the CRD schema"); + } + + #[test] + fn budget_policy_cap_usd_schema_has_zero_minimum() { + let crd = crd_json(); + let cap_schema = crd + .pointer( + "/spec/versions/0/schema/openAPIV3Schema/properties/spec/properties\ + /budgetPolicy/properties/tenants/items/properties/capUsd", + ) + .unwrap_or_else(|| std::process::abort()); + assert_eq!( + cap_schema.pointer("/minimum").and_then(serde_json::Value::as_f64), + Some(0.0), + "capUsd schema must reject negative values" + ); + } + + #[test] + fn budget_policy_rejects_unknown_shape() { + let json = serde_json::json!({ + "gridId": "", + "seeds": [], + "budgetPolicy": { "caps": [] } + }); + let result = serde_json::from_value::(json); + assert!(result.is_err(), "unknown budgetPolicy shape must be rejected"); + } + + // ----------------------------------------------------------------------- + // tenant_spend_status tests (E1-E6) + // ----------------------------------------------------------------------- + + fn spend_map(entries: &[(&str, u64)]) -> BTreeMap { + entries + .iter() + .map(|(tenant_id, cents)| ((*tenant_id).to_owned(), spend_of(*cents))) + .collect() + } + + #[test] + fn tenant_spend_status_reports_recorded_spend() { + let policy = BudgetPolicyConfig { + tenants: vec![tenant("tenant-a", 100.0)], + }; + let spend = spend_map(&[("tenant-a", 2500)]); + let statuses = tenant_spend_status(&policy, &spend); + assert_eq!(statuses.len(), 1); + let entry = statuses.first().unwrap_or_else(|| std::process::abort()); + assert_eq!(entry.tenant_id, "tenant-a"); + assert_weight(entry.spend_usd, 25.0, "spend_usd"); + assert_weight(entry.cap_usd, 100.0, "cap_usd"); + assert_weight(entry.spend_ratio, 0.25, "spend_ratio"); + } + + #[test] + fn tenant_spend_status_includes_tenant_with_no_spend_yet() { + let policy = BudgetPolicyConfig { + tenants: vec![tenant("tenant-a", 100.0)], + }; + let statuses = tenant_spend_status(&policy, &BTreeMap::new()); + assert_eq!( + statuses.len(), + 1, + "declared tenant must be present even with zero recorded spend" + ); + let entry = statuses.first().unwrap_or_else(|| std::process::abort()); + assert_weight(entry.spend_usd, 0.0, "spend_usd with no traffic yet"); + assert_weight(entry.spend_ratio, 0.0, "spend_ratio with no traffic yet"); + } + + #[test] + fn tenant_spend_status_excludes_spend_for_undeclared_tenant() { + let policy = BudgetPolicyConfig { + tenants: vec![tenant("tenant-a", 100.0)], + }; + let spend = spend_map(&[("tenant-a", 1000), ("tenant-orphan", 9999)]); + let statuses = tenant_spend_status(&policy, &spend); + assert_eq!( + statuses.len(), + 1, + "CRDT spend for a tenant no longer declared in policy must be excluded" + ); + assert_eq!( + statuses.first().unwrap_or_else(|| std::process::abort()).tenant_id, + "tenant-a" + ); + } + + #[test] + fn tenant_spend_status_empty_policy_is_empty() { + let policy = BudgetPolicyConfig { tenants: Vec::new() }; + let statuses = tenant_spend_status(&policy, &spend_map(&[("tenant-a", 1000)])); + assert!(statuses.is_empty(), "empty policy must produce empty status"); + } + + #[test] + fn tenant_spend_status_is_sorted_by_tenant_id() { + let policy = BudgetPolicyConfig { + tenants: vec![tenant("tenant-z", 100.0), tenant("tenant-a", 100.0)], + }; + let statuses = tenant_spend_status(&policy, &BTreeMap::new()); + let ids: Vec<&str> = statuses.iter().map(|s| s.tenant_id.as_str()).collect(); + assert_eq!( + ids, + vec!["tenant-a", "tenant-z"], + "status must be deterministically sorted by tenant_id" + ); + } + + #[test] + fn tenant_spend_status_over_cap_clamps_ratio() { + let policy = BudgetPolicyConfig { + tenants: vec![tenant("tenant-a", 10.0)], + }; + // 20.00 spent against a 10.00 cap. + let spend = spend_map(&[("tenant-a", 2000)]); + let statuses = tenant_spend_status(&policy, &spend); + assert_weight( + statuses.first().unwrap_or_else(|| std::process::abort()).spend_ratio, + 1.0, + "over-cap spend_ratio must clamp to 1.0", + ); + } + + // ----------------------------------------------------------------------- + // resolve_budget_statuses tests (G1) + // ----------------------------------------------------------------------- + + #[test] + fn resolve_budget_statuses_none_policy_is_empty() { + let statuses = resolve_budget_statuses(None, &spend_map(&[("tenant-a", 1000)])); + assert!( + statuses.is_empty(), + "absent budgetPolicy must produce empty budget_status" + ); + } + + #[test] + fn resolve_budget_statuses_delegates_to_tenant_spend_status() { + let policy = BudgetPolicyConfig { + tenants: vec![tenant("tenant-a", 100.0)], + }; + let spend = spend_map(&[("tenant-a", 5000)]); + let statuses = resolve_budget_statuses(Some(&policy), &spend); + assert_eq!(statuses.len(), 1); + assert_weight( + statuses.first().unwrap_or_else(|| std::process::abort()).spend_ratio, + 0.5, + "spend_ratio via resolve_budget_statuses", + ); + } } diff --git a/operator/src/swim_runtime.rs b/operator/src/swim_runtime.rs index 1f84217..d2fe86c 100644 --- a/operator/src/swim_runtime.rs +++ b/operator/src/swim_runtime.rs @@ -2341,6 +2341,149 @@ mod tests { drop(handle2); } + /// Poll a handle's `state_snapshot()` until a tenant's converged spend + /// reaches `expected_cents`, or panic at the deadline. + /// + /// Exercises the real production merge path: [`GridStateSnapshot::merge_tenant_spend`] + /// as invoked by [`swim::state_broadcast::StateBroadcastHandler::receive_item`] on + /// every SWIM gossip round — no CRDT logic is duplicated here. + async fn wait_until_tenant_spend_converges(handle: &SwimHandle, tenant_id: &str, expected_cents: u64) { + let deadline = tokio::time::Instant::now() + Duration::from_secs(10); + loop { + let total = handle + .state_snapshot() + .tenant_spend + .get(tenant_id) + .map(crdt::GCounter::total) + .unwrap_or_default(); + if total == expected_cents { + return; + } + assert!( + tokio::time::Instant::now() < deadline, + "tenant '{tenant_id}' spend must converge to {expected_cents} cents via real SWIM gossip \ + (last observed: {total} cents)" + ); + tokio::time::sleep(Duration::from_millis(50)).await; + } + } + + #[tokio::test] + #[expect( + clippy::too_many_lines, + reason = "three-node real-UDP gossip proof: join, dual increment, convergence, late join, status tie-in" + )] + async fn tenant_spend_converges_at_late_joining_site_after_partition_heals() { + // grid#40 AC3, live network proof: sites A and B accumulate real per-request + // spend for the same tenant *before* site C ever joins the mesh (a stand-in + // for C being partitioned away while A and B kept serving traffic). C then + // joins ("the partition heals") and must converge to the true cross-site sum + // purely through the real UDP-bound SWIM runtime — no manual message shuttling, + // unlike the lower-tier unit-level proof in `swim::node::tests`. + let addr_a = reserve_local_addr().await; + let addr_b = reserve_local_addr().await; + let addr_c = reserve_local_addr().await; + + let handle_a = start(SwimConfig { + bind_addr: addr_a, + advertise_addr: Some(addr_a), + site_name: "site-a".to_owned(), + seeds: Vec::new(), + gateway_address: None, + swim_key: None, + revision_lease: test_revision_lease(80_000), + }) + .await + .unwrap_or_else(|_| std::process::abort()); + + let handle_b = start(SwimConfig { + bind_addr: addr_b, + advertise_addr: Some(addr_b), + site_name: "site-b".to_owned(), + seeds: vec![addr_a], + gateway_address: None, + swim_key: None, + revision_lease: test_revision_lease(90_000), + }) + .await + .unwrap_or_else(|_| std::process::abort()); + wait_until_member_alive(&handle_a, "site-b").await; + + // Real per-request cost values, matching `scoring::BackendConfig::cost_per_1k_input` + // shape rather than an arbitrary test constant. + let a_cents = cost_cents_for_tokens(0.03, 1_800); + let mut a_snap = GridStateSnapshot::new("site-a".to_owned()); + a_snap.increment_tenant_spend("tenant-acme", a_cents); + handle_a + .publish_state_broadcast(swim::StateBroadcast::new("site-a".to_owned(), 1, a_snap, None)) + .unwrap_or_else(|_| std::process::abort()); + + let b_cents = cost_cents_for_tokens(0.06, 2_500); + let mut b_snap = GridStateSnapshot::new("site-b".to_owned()); + b_snap.increment_tenant_spend("tenant-acme", b_cents); + handle_b + .publish_state_broadcast(swim::StateBroadcast::new("site-b".to_owned(), 1, b_snap, None)) + .unwrap_or_else(|_| std::process::abort()); + + let total_cents = a_cents + b_cents; + wait_until_tenant_spend_converges(&handle_a, "tenant-acme", total_cents).await; + wait_until_tenant_spend_converges(&handle_b, "tenant-acme", total_cents).await; + + // The partition "heals": site-c joins the already-converged A/B mesh for + // the first time, having missed every prior broadcast. + let handle_c = start(SwimConfig { + bind_addr: addr_c, + advertise_addr: Some(addr_c), + site_name: "site-c".to_owned(), + seeds: vec![addr_a], + gateway_address: None, + swim_key: None, + revision_lease: test_revision_lease(100_000), + }) + .await + .unwrap_or_else(|_| std::process::abort()); + wait_until_member_alive(&handle_a, "site-c").await; + + wait_until_tenant_spend_converges(&handle_c, "tenant-acme", total_cents).await; + + // End-to-end tie-in: the same status-derivation function the reconciler + // calls on every reconcile must report the correct spendRatio from C's + // independently-converged view, proving the full CRDT -> status pipeline. + let policy = crate::crd::grid_network::BudgetPolicyConfig { + tenants: vec![crate::crd::grid_network::TenantBudgetConfig { + tenant_id: "tenant-acme".to_owned(), + cap_usd: 1.00, + }], + }; + let statuses = + crate::crd::grid_network::resolve_budget_statuses(Some(&policy), &handle_c.state_snapshot().tenant_spend); + let status = statuses.first().unwrap_or_else(|| std::process::abort()); + assert_eq!( + status.tenant_id, "tenant-acme", + "status must be keyed by the policy's tenant_id" + ); + assert!( + (status.spend_usd - crate::crd::grid_network::cents_to_usd(total_cents)).abs() < f64::EPSILON, + "spend_usd must reflect the fully-converged cross-site total observed at the late-joining site" + ); + + drop((handle_b, handle_c)); + } + + /// Real per-request USD cost for `tokens` at `cost_per_1k`, in integer cents — + /// mirrors how a gateway-side policy filter would size a spend increment from + /// `scoring::BackendConfig::cost_per_1k_input` (AC5 non-goal: that filter does + /// not exist yet, so this helper stands in for it here). + #[expect( + clippy::cast_possible_truncation, + clippy::cast_sign_loss, + clippy::cast_precision_loss, + reason = "test-only cost simulation over small constant token counts, well within f64 exact-integer range" + )] + fn cost_cents_for_tokens(cost_per_1k: f64, tokens: u64) -> u64 { + (cost_per_1k * (tokens as f64 / 1000.0) * 100.0).round() as u64 + } + // ----------------------------------------------------------------------- // Dead-member eviction // ----------------------------------------------------------------------- diff --git a/swim/src/node.rs b/swim/src/node.rs index ffbe6a9..37d7543 100644 --- a/swim/src/node.rs +++ b/swim/src/node.rs @@ -706,6 +706,87 @@ mod tests { ); } + // ----------------------------------------------------------------------- + // Tenant budget spend convergence (grid#40 AC3) + // ----------------------------------------------------------------------- + + /// Compute a spend increment in cents from a real product cost field + /// ([`scoring::BackendConfig::cost_per_1k_input`], mirrored here without a + /// crate dependency to keep `swim` free of the `scoring` crate) and a + /// request's input token count — the same unit conversion + /// `operator::crd::grid_network::spend_ratio` expects on the read side. + fn cost_cents_for_request(cost_per_1k_input_usd: f64, input_tokens: u64) -> u64 { + #[expect( + clippy::cast_precision_loss, + reason = "test-only cost simulation, not the production conversion path" + )] + let tokens = input_tokens as f64; + let usd = cost_per_1k_input_usd * (tokens / 1000.0); + #[expect( + clippy::cast_sign_loss, + clippy::cast_possible_truncation, + reason = "usd is always non-negative in this test fixture" + )] + let cents = (usd * 100.0).round() as u64; + cents + } + + #[test] + #[expect( + clippy::too_many_lines, + reason = "two-origin gossip proof establishes membership, broadcasts twice, and verifies convergence" + )] + fn tenant_spend_from_two_origin_sites_converges_at_third_via_gossip() { + let id_a = local_id("site-a", 19_216); + let id_b = local_id("site-b", 19_217); + let id_c = local_id("site-c", 19_218); + let (mut node_a, _) = make_node("site-a", 19_216); + let (mut node_b, _) = make_node("site-b", 19_217); + let (mut node_c, _) = make_node("site-c", 19_218); + + establish_membership(&mut node_a, &mut node_c, &id_a, &id_c); + establish_membership(&mut node_b, &mut node_c, &id_b, &id_c); + + // Site A serves a request for tenant-acme against a $0.03/1k-input-token backend. + let a_cents = cost_cents_for_request(0.03, 4_000); + let mut a_snap = GridStateSnapshot::new("site-a".to_owned()); + a_snap.increment_tenant_spend("tenant-acme", a_cents); + node_a + .publish_state_broadcast(&StateBroadcast::new("site-a".to_owned(), 1, a_snap, None)) + .unwrap_or_else(|_| std::process::abort()); + for msg in &node_a.gossip().messages { + if msg.addr == id_c.socket_addr() { + drop(node_c.handle_data(&msg.data)); + } + } + + // Site B independently serves a request for the same tenant against a pricier backend. + let b_cents = cost_cents_for_request(0.06, 2_500); + let mut b_snap = GridStateSnapshot::new("site-b".to_owned()); + b_snap.increment_tenant_spend("tenant-acme", b_cents); + node_b + .publish_state_broadcast(&StateBroadcast::new("site-b".to_owned(), 1, b_snap, None)) + .unwrap_or_else(|_| std::process::abort()); + for msg in &node_b.gossip().messages { + if msg.addr == id_c.socket_addr() { + drop(node_c.handle_data(&msg.data)); + } + } + + let c_snap = node_c.state_snapshot(); + let converged_total = c_snap + .tenant_spend + .get("tenant-acme") + .unwrap_or_else(|| std::process::abort()) + .total(); + assert_eq!( + converged_total, + a_cents + b_cents, + "tenant spend from two independent origin sites must converge to the true sum \ + at a third site via real SWIM gossip broadcast, proving AC3 (cross-site convergence)" + ); + } + // ----------------------------------------------------------------------- // StateBroadcastError display // ----------------------------------------------------------------------- diff --git a/swim/src/state_broadcast.rs b/swim/src/state_broadcast.rs index a209748..88bd0a1 100644 --- a/swim/src/state_broadcast.rs +++ b/swim/src/state_broadcast.rs @@ -149,7 +149,23 @@ impl StateBroadcast { /// (gateway address and/or site cert PEM) with no CRDT state. #[must_use] fn is_metadata_only(&self) -> bool { - self.snapshot.providers.is_empty() && self.snapshot.capabilities.is_empty() + self.snapshot.providers.is_empty() + && self.snapshot.capabilities.is_empty() + && self.snapshot.tenant_spend.is_empty() + } + + /// Return true when this payload carries provider or capability records. + /// + /// Distinct from [`carries_grid_state`](Self::carries_grid_state): a + /// tenant-spend-only broadcast carries grid state (must not be treated as + /// metadata-only) but must **not** trigger `replace_origin_providers`, + /// which performs a destructive retain-then-replace of the origin's + /// provider set. Only a broadcast that actually carries provider or + /// capability data represents an authoritative provider-state sync for + /// its origin. + #[must_use] + fn carries_provider_state(&self) -> bool { + !self.snapshot.providers.is_empty() || !self.snapshot.capabilities.is_empty() } /// Return true when this payload only advertises a gateway address. @@ -382,11 +398,27 @@ impl OriginStateHandle { }) } - /// Remove all state associated with an origin and publish the result. + /// Remove provider and transport state for a departed origin, and + /// publish the result. + /// + /// Deliberately does **not** touch `tenant_spend`: this method fires on + /// ordinary SWIM membership churn (a site marked `Suspect`/`Dead` past + /// its suspect/dead TTL, e.g. a pod restart or a transient partition — + /// see `operator::swim_runtime::prune_tracked_members`), not on + /// permanent tenant-budget retirement. `tenant_spend` is a cumulative + /// (grow-only) ledger; wiping a site's slot here would let a tenant's + /// `spendRatio` drop on a restart or blip and reopen an + /// already-exhausted budget. If spend ever needs to expire, that must be + /// an explicit budget-epoch/window reset, not a side effect of + /// membership eviction — tracked in + /// [grid#52](https://github.com/praxis-proxy/grid/issues/52), which also + /// covers bounding per-tenant site-slot growth now that this path no + /// longer prunes it. pub(crate) fn remove_origin(&self, origin: &str) { self.lock().remove(origin); - self.state_tx - .send_modify(|snapshot| snapshot.remove_origin_providers(origin)); + self.state_tx.send_modify(|snapshot| { + snapshot.remove_origin_providers(origin); + }); self.gateway_addrs_tx.send_modify(|addresses| { addresses.remove(origin); }); @@ -671,9 +703,16 @@ impl foca::BroadcastHandler for StateBroadcastHandler { return Ok(None); } + let carries_provider_state = broadcast.carries_provider_state(); self.state_tx.send_modify(|snap| { snap.capabilities.merge(&broadcast.snapshot.capabilities); - snap.replace_origin_providers(&broadcast.origin_site, broadcast.revision, &broadcast.snapshot); + snap.merge_tenant_spend_from_origin(&broadcast.origin_site, &broadcast.snapshot.tenant_spend); + // A spend-only broadcast must not run the destructive + // origin-provider replace below — it doesn't carry an + // authoritative provider list for this cycle at all. + if carries_provider_state { + snap.replace_origin_providers(&broadcast.origin_site, broadcast.revision, &broadcast.snapshot); + } }); self.retained .lock() @@ -690,7 +729,7 @@ impl foca::BroadcastHandler for StateBroadcastHandler { #[cfg(test)] mod tests { - use crdt::{Capability, ProviderMetricsSnapshot, ProviderPhase, ProviderState}; + use crdt::{Capability, GCounter, ProviderMetricsSnapshot, ProviderPhase, ProviderState}; use foca::{BroadcastHandler as _, Invalidates as _}; use super::*; @@ -762,6 +801,121 @@ mod tests { ); } + #[test] + fn tenant_spend_only_snapshot_carries_grid_state() { + // A broadcast can carry only a tenant_spend increment (no provider or + // capability change this gossip cycle) — this must NOT be classified + // as metadata-only, or receive_item's merge path is skipped entirely + // (regression: `is_metadata_only` originally only checked + // providers/capabilities, silently dropping spend-only broadcasts). + let mut snap = GridStateSnapshot::new("site-p".to_owned()); + snap.increment_tenant_spend("tenant-x", 500); + let broadcast = StateBroadcast::new("site-p".to_owned(), 1, snap, None); + + assert!( + broadcast.carries_grid_state(), + "tenant_spend-only snapshot must be treated as carrying grid state, not metadata-only" + ); + } + + #[test] + fn receive_item_merges_tenant_spend_only_broadcast_with_no_providers_or_capabilities() { + let mut handler = StateBroadcastHandler::new("site-local".to_owned()); + let mut snap = GridStateSnapshot::new("site-p".to_owned()); + snap.increment_tenant_spend("tenant-x", 500); + let broadcast = StateBroadcast::new("site-p".to_owned(), 1, snap, None); + + receive(&mut handler, &broadcast); + + assert_eq!( + handler + .snapshot() + .tenant_spend + .get("tenant-x") + .unwrap_or_else(|| std::process::abort()) + .total(), + 500, + "a broadcast with only tenant_spend set (no providers/capabilities) must still be merged, \ + not dropped as metadata-only" + ); + } + + #[test] + fn receive_item_spend_only_broadcast_does_not_wipe_origins_existing_providers() { + // Bugbot regression: a later spend-only broadcast from an origin that + // already has real providers must not erase those providers. + // `replace_origin_providers` performs a destructive retain-then-replace + // for the origin's provider set; it must only run when the broadcast + // actually carries provider/capability data, not merely because + // `carries_grid_state()` is true (which tenant_spend alone satisfies). + let mut handler = StateBroadcastHandler::new("site-local".to_owned()); + let full = StateBroadcast::new("site-p".to_owned(), 1, snapshot("site-p", 1, 0.2), None); + receive(&mut handler, &full); + assert!( + handler.snapshot().provider("net", "site-p", "provider").is_some(), + "precondition: origin's provider must be present after the first broadcast" + ); + + let mut spend_only = GridStateSnapshot::new("site-p".to_owned()); + spend_only.increment_tenant_spend("tenant-x", 500); + let spend_broadcast = StateBroadcast::new("site-p".to_owned(), 2, spend_only, None); + receive(&mut handler, &spend_broadcast); + + assert!( + handler.snapshot().provider("net", "site-p", "provider").is_some(), + "a spend-only broadcast from the same origin must not wipe that origin's provider records" + ); + assert_eq!( + handler + .snapshot() + .tenant_spend + .get("tenant-x") + .unwrap_or_else(|| std::process::abort()) + .total(), + 500, + "the spend-only broadcast's tenant_spend must still be merged" + ); + } + + #[test] + fn origin_state_handle_remove_origin_preserves_that_origins_tenant_spend() { + // Correctness (grid#47 review): membership eviction fires on ordinary + // SWIM churn (a restart or a transient partition exceeding the + // suspect/dead TTL), not on permanent tenant-budget retirement. + // Wiping a site's cumulative spend contribution here would let + // `spendRatio` drop and reopen an already-exhausted budget on a mere + // restart, unlike provider records (which are membership-scoped and + // correctly pruned). + let (mut handler, control) = StateBroadcastHandler::with_capacity("site-local".to_owned(), 8); + let mut snap = GridStateSnapshot::new("site-p".to_owned()); + snap.increment_tenant_spend("tenant-x", 500); + let broadcast = StateBroadcast::new("site-p".to_owned(), 1, snap, None); + receive(&mut handler, &broadcast); + assert_eq!( + control + .state_tx + .borrow() + .tenant_spend + .get("tenant-x") + .map(GCounter::total), + Some(500), + "precondition: tenant spend merged before eviction" + ); + + control.remove_origin("site-p"); + + assert_eq!( + control + .state_tx + .borrow() + .tenant_spend + .get("tenant-x") + .map(GCounter::total), + Some(500), + "evicting an origin from membership must not erase its cumulative spend contribution" + ); + } + #[test] fn newer_key_invalidates_older_from_same_origin() { let old = StateBroadcastKey { @@ -1280,6 +1434,55 @@ mod tests { assert_eq!(handler.cert_pem_for_site("site-p").as_deref(), Some(cert)); } + // ----------------------------------------------------------------------- + // tenant_spend broadcast wiring tests (F1-F2) + // ----------------------------------------------------------------------- + + #[test] + fn receive_item_merges_tenant_spend_from_broadcast() { + let mut handler = StateBroadcastHandler::new("site-local".to_owned()); + let mut snap = snapshot("site-p", 1, 0.2); + snap.increment_tenant_spend("tenant-x", 500); + let broadcast = StateBroadcast::new("site-p".to_owned(), 1, snap, None); + + receive(&mut handler, &broadcast); + + let merged = handler.snapshot(); + assert_eq!( + merged + .tenant_spend + .get("tenant-x") + .unwrap_or_else(|| std::process::abort()) + .total(), + 500, + "receive_item must merge the broadcast's tenant_spend into the handler's snapshot" + ); + } + + #[test] + fn receive_item_sums_tenant_spend_across_origin_sites() { + let mut handler = StateBroadcastHandler::new("site-local".to_owned()); + + let mut snap_a = snapshot("site-a", 1, 0.2); + snap_a.increment_tenant_spend("tenant-x", 300); + receive(&mut handler, &StateBroadcast::new("site-a".to_owned(), 1, snap_a, None)); + + let mut snap_b = snapshot("site-b", 1, 0.2); + snap_b.increment_tenant_spend("tenant-x", 700); + receive(&mut handler, &StateBroadcast::new("site-b".to_owned(), 1, snap_b, None)); + + let merged = handler.snapshot(); + assert_eq!( + merged + .tenant_spend + .get("tenant-x") + .unwrap_or_else(|| std::process::abort()) + .total(), + 1000, + "tenant spend from two different origin sites must sum, proving cross-site convergence at the wiring layer" + ); + } + #[test] fn handler_accepts_v1_broadcast_without_gateway_address() { let mut handler = StateBroadcastHandler::new("site-local".to_owned());