From 68f9a6450562ef74f739a60e4e9da9f5f290de96 Mon Sep 17 00:00:00 2001 From: Jarvis Date: Wed, 12 Aug 2026 17:01:06 +0800 Subject: [PATCH 1/3] fix(quota): match model conditions against the requested parent as well as the dispatched target MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A conditional rate-limit policy whose model condition names a routing group (or whose model_name names the group's alias) never fired: the request gate defers model-property policies to the per-target gate, where the condition input carried only the dispatched member's identity — the parent's id was compared nowhere. The per-target condition input now carries the {dispatched target, requested parent} pair for the model dimensions. A leaf is raw-true when either identity satisfies its operator, and negate flips the combined result — so a group-referencing condition selects exactly the requests addressed to that group, and a negated one excludes them instead of (absurdly) matching all of them. Reservation phases and bucket values are unchanged: group_by [model] still splits per concrete target, and classic scope:model rows (which already matched parent ids at the request gate) now agree with the conditional form. Wired through every group-capable dispatch loop (chat streaming + non-streaming, /v1/messages, /v1/responses, /v1/messages/count_tokens) and the ensemble panel/judge reservations; semantic routing shares the chat loop. On an ensemble, a parent-referencing condition reserves per sub-call (panel members + judge) — the parent-level per-request cap remains the entry's own inline rate_limit. Fixes api7/AISIX-Cloud#1267 --- .../src/models/policy_conditions.rs | 139 +++++- crates/aisix-proxy/src/chat.rs | 18 +- crates/aisix-proxy/src/count_tokens.rs | 5 +- crates/aisix-proxy/src/ensemble.rs | 4 + crates/aisix-proxy/src/lib.rs | 3 + crates/aisix-proxy/src/messages.rs | 5 +- crates/aisix-proxy/src/quota.rs | 157 ++++++- crates/aisix-proxy/src/responses.rs | 5 +- .../resources/rate_limit_policy.schema.json | 4 +- ...roup-model-condition-ratelimit-e2e.test.ts | 409 ++++++++++++++++++ 10 files changed, 711 insertions(+), 38 deletions(-) create mode 100644 tests/e2e/src/cases/group-model-condition-ratelimit-e2e.test.ts diff --git a/crates/aisix-core/src/models/policy_conditions.rs b/crates/aisix-core/src/models/policy_conditions.rs index b91efda0..576d5245 100644 --- a/crates/aisix-core/src/models/policy_conditions.rs +++ b/crates/aisix-core/src/models/policy_conditions.rs @@ -29,6 +29,13 @@ //! deliberately includes MCP/A2A traffic; //! - groups short-circuit (AND on the first false child, OR on the //! first true child); +//! - the model dimensions (`model` / `model_name`) evaluate against a +//! PAIR on a routing/ensemble/semantic dispatch: the dispatched +//! target and the caller-addressed parent entry (AISIX-Cloud#1267). +//! A leaf is raw-true when either identity satisfies its operator; +//! `negate` flips that combined result, so `!(model in [group])` +//! excludes every request addressed to the group instead of matching +//! all of them; //! - regexes are compiled once per distinct pattern into a process-wide //! cache. Load-time validation guarantees compilability, so a cache //! miss at evaluation time never fails in practice; a pattern that @@ -62,11 +69,15 @@ pub enum PolicyDimension { Member, /// Authenticated api_key entry id (UUID). ApiKey, - /// Dispatched model entry id (UUID); routing/model groups match per - /// selected target, never the group entry itself. + /// Model entry id (UUID). Matches the dispatched model; on a + /// Model-Group / semantic-router / ensemble dispatch it also + /// matches the requested parent entry, so a group's own id selects + /// every request addressed to that group. Model, - /// Dispatched model display name — the string dimension for - /// regex/prefix matching ("every gpt-4-family alias"). + /// Model display name — the string dimension for regex/prefix + /// matching ("every gpt-4-family alias"). Matches the dispatched + /// model's name, and on a virtual-parent dispatch also the + /// requested parent's name. ModelName, /// Dispatched model's `provider` (models.dev catalog id). Provider, @@ -386,6 +397,17 @@ pub struct ConditionInput<'a> { pub model: Option<&'a str>, pub model_name: Option<&'a str>, pub provider: Option<&'a str>, + /// Entry id of the caller-addressed virtual parent (routing group / + /// ensemble / semantic router) when `model` below is a dispatch + /// target that parent selected. A `model` leaf matches when EITHER + /// id satisfies it (AISIX-Cloud#1267) — the group a caller + /// addressed is as much "the model" as the member it dispatched to. + /// `None` on direct dispatch and at the request gate (where `model` + /// already IS the requested entry). + pub routing_parent_model: Option<&'a str>, + /// Display name of that parent, the `model_name` twin of + /// [`Self::routing_parent_model`]. + pub routing_parent_model_name: Option<&'a str>, } impl<'a> ConditionInput<'a> { @@ -400,6 +422,19 @@ impl<'a> ConditionInput<'a> { } } + /// The caller-addressed parent's value for a model-property + /// dimension, when the primary value describes a dispatch target + /// that parent selected. Identity/provider dimensions have no + /// parent variant (the parent shares the caller identity and + /// carries no provider). + fn routing_parent(&self, dimension: PolicyDimension) -> Option<&'a str> { + match dimension { + PolicyDimension::Model => self.routing_parent_model, + PolicyDimension::ModelName => self.routing_parent_model_name, + _ => None, + } + } + pub fn get_group_by(&self, dimension: GroupByDimension) -> Option<&'a str> { match dimension { GroupByDimension::Team => self.team, @@ -437,7 +472,19 @@ fn eval_leaf(leaf: &PolicyCondition, input: &ConditionInput<'_>) -> bool { let Some(var) = input.get(leaf.dimension) else { return false; }; - let raw = match (leaf.operator, &leaf.value) { + // Model dimensions carry a {dispatched target, requested parent} + // pair on a routing dispatch: raw-true when either satisfies the + // operator, and `negate` flips the combined result — so a negated + // leaf excluding the parent excludes every request addressed to it. + let raw = eval_operator(leaf, var) + || input + .routing_parent(leaf.dimension) + .is_some_and(|parent| eval_operator(leaf, parent)); + raw != leaf.negate +} + +fn eval_operator(leaf: &PolicyCondition, var: &str) -> bool { + match (leaf.operator, &leaf.value) { (ConditionOperator::Eq, ConditionValue::One(v)) => var == v, (ConditionOperator::Ne, ConditionValue::One(v)) => var != v, (ConditionOperator::In, ConditionValue::Many(items)) => { @@ -474,8 +521,7 @@ fn eval_leaf(leaf: &PolicyCondition, input: &ConditionInput<'_>) -> bool { (ConditionOperator::Has | ConditionOperator::IpMatch, _) => false, // Value shape mismatching the operator (validation rejects it). _ => false, - }; - raw != leaf.negate + } } /// Process-wide compiled-regex caches, one per case-sensitivity @@ -562,6 +608,19 @@ mod tests { model: Some("model-1"), model_name: Some("gpt-4.1-prod"), provider: Some("openai"), + routing_parent_model: None, + routing_parent_model_name: None, + } + } + + /// `input()` as the per-target gate of a routing dispatch sees it: + /// the target as the primary values, the caller-addressed group as + /// the parent pair. + fn routed_input<'a>() -> ConditionInput<'a> { + ConditionInput { + routing_parent_model: Some("group-1"), + routing_parent_model_name: Some("chat-group"), + ..input() } } @@ -656,6 +715,72 @@ mod tests { assert!(eval_condition_nodes(&nodes, &input())); } + #[test] + fn model_leaf_matches_routing_parent_id() { + // AISIX-Cloud#1267: `model in [group uuid]` must select requests + // dispatched THROUGH the group even though the per-target gate's + // primary value is the member id. + let nodes = vec![leaf( + PolicyDimension::Model, + ConditionOperator::In, + many(&["group-1"]), + )]; + assert!(eval_condition_nodes(&nodes, &routed_input())); + // Direct dispatch to the member (no parent): the group condition + // must NOT capture it. + assert!(!eval_condition_nodes(&nodes, &input())); + } + + #[test] + fn model_name_leaf_matches_routing_parent_name() { + let nodes = vec![leaf( + PolicyDimension::ModelName, + ConditionOperator::Regex, + one("^chat-"), + )]; + assert!(eval_condition_nodes(&nodes, &routed_input())); + assert!(!eval_condition_nodes(&nodes, &input())); + } + + #[test] + fn negated_model_leaf_excludes_parent_dispatch() { + // `!(model in [group-1])` means "everything except the group": + // negate flips the COMBINED pair result, so a request routed via + // the group is excluded — not (absurdly) matched because the + // member id alone misses the set. + let nodes = vec![neg_leaf( + PolicyDimension::Model, + ConditionOperator::In, + many(&["group-1"]), + )]; + assert!(!eval_condition_nodes(&nodes, &routed_input())); + // The same member reached directly stays matched. + assert!(eval_condition_nodes(&nodes, &input())); + } + + #[test] + fn member_leaf_still_matches_through_parent() { + // The 1087 principle survives the pair: a member-id condition + // keeps matching when the member is reached via a group. + let nodes = vec![leaf( + PolicyDimension::Model, + ConditionOperator::In, + many(&["model-1"]), + )]; + assert!(eval_condition_nodes(&nodes, &routed_input())); + } + + #[test] + fn parent_pair_never_leaks_into_identity_dimensions() { + // A team leaf must not consult the parent pair even when set. + let nodes = vec![leaf( + PolicyDimension::Team, + ConditionOperator::In, + many(&["group-1"]), + )]; + assert!(!eval_condition_nodes(&nodes, &routed_input())); + } + #[test] fn missing_dimension_is_false_even_negated() { let no_team = ConditionInput { diff --git a/crates/aisix-proxy/src/chat.rs b/crates/aisix-proxy/src/chat.rs index e016c22d..8d90f84e 100644 --- a/crates/aisix-proxy/src/chat.rs +++ b/crates/aisix-proxy/src/chat.rs @@ -1560,7 +1560,10 @@ async fn dispatch( state, snapshot, auth, - is_routing_request, + is_routing_request.then_some(crate::quota::RoutingParent { + name: &virtual_entry.value.display_name, + entry_id: &virtual_entry.id, + }), &model.display_name, &attempt.id, model, @@ -2639,7 +2642,10 @@ async fn dispatch( state, snapshot, auth, - is_routing_request, + is_routing_request.then_some(crate::quota::RoutingParent { + name: &virtual_entry.value.display_name, + entry_id: &virtual_entry.id, + }), &model.display_name, &attempt.id, model, @@ -3260,6 +3266,10 @@ async fn dispatch_ensemble( snapshot, request_id, client, + routing_parent: crate::quota::RoutingParent { + name: &virtual_entry.value.display_name, + entry_id: &virtual_entry.id, + }, }; // Streaming ensemble (OPTION A): the panel must be buffered to synthesize, @@ -3394,6 +3404,10 @@ async fn dispatch_ensemble( &ensemble_cfg.judge.model, &judge_entry.id, judge_model, + Some(crate::quota::RoutingParent { + name: &virtual_entry.value.display_name, + entry_id: &virtual_entry.id, + }), ) .await { diff --git a/crates/aisix-proxy/src/count_tokens.rs b/crates/aisix-proxy/src/count_tokens.rs index ac76e20c..43e9c80e 100644 --- a/crates/aisix-proxy/src/count_tokens.rs +++ b/crates/aisix-proxy/src/count_tokens.rs @@ -262,7 +262,10 @@ async fn dispatch( state, snapshot, auth, - is_routing_request, + is_routing_request.then_some(crate::quota::RoutingParent { + name: &model_entry.value.display_name, + entry_id: &model_entry.id, + }), &target.model.display_name, &target.id, &target.model, diff --git a/crates/aisix-proxy/src/ensemble.rs b/crates/aisix-proxy/src/ensemble.rs index 35b5db55..2b1e8576 100644 --- a/crates/aisix-proxy/src/ensemble.rs +++ b/crates/aisix-proxy/src/ensemble.rs @@ -93,6 +93,9 @@ pub(crate) struct ProxyModelCaller<'a> { /// the caller's behalf, so they carry the same caller identity and /// forwardable client headers as a single-upstream dispatch would. pub client: &'a crate::client_ip::ClientContext, + /// The ensemble entry the caller addressed, so per-member policy + /// matching sees the {member, parent} pair (AISIX-Cloud#1267). + pub routing_parent: crate::quota::RoutingParent<'a>, } #[async_trait] @@ -159,6 +162,7 @@ impl ModelCaller for ProxyModelCaller<'_> { target, &entry.id, model, + Some(self.routing_parent), ) .await .map_err(|e| { diff --git a/crates/aisix-proxy/src/lib.rs b/crates/aisix-proxy/src/lib.rs index cebbe3a3..d866b5f6 100644 --- a/crates/aisix-proxy/src/lib.rs +++ b/crates/aisix-proxy/src/lib.rs @@ -3783,6 +3783,7 @@ data: [DONE]\n\n" "mg-member", "model-id-1", &target, + None, ) .await; assert!(r.is_ok(), "suspended policy must reserve nothing"); @@ -3805,6 +3806,7 @@ data: [DONE]\n\n" "mg-member", "model-id-1", &target, + None, ) .await .is_ok()); @@ -3816,6 +3818,7 @@ data: [DONE]\n\n" "mg-member", "model-id-1", &target, + None, ) .await .is_err(), diff --git a/crates/aisix-proxy/src/messages.rs b/crates/aisix-proxy/src/messages.rs index 5853b2cd..5033e9b4 100644 --- a/crates/aisix-proxy/src/messages.rs +++ b/crates/aisix-proxy/src/messages.rs @@ -735,7 +735,10 @@ async fn dispatch( state, snapshot, auth, - is_routing_request, + is_routing_request.then_some(crate::quota::RoutingParent { + name: &model_entry.value.display_name, + entry_id: &model_entry.id, + }), &target.model.display_name, &target.id, &target.model, diff --git a/crates/aisix-proxy/src/quota.rs b/crates/aisix-proxy/src/quota.rs index 8504424a..4284a084 100644 --- a/crates/aisix-proxy/src/quota.rs +++ b/crates/aisix-proxy/src/quota.rs @@ -68,12 +68,28 @@ impl ModelRateLimit { } } +/// Identity of the caller-addressed virtual parent (routing group / +/// ensemble / semantic router), forwarded by the dispatch loops into the +/// per-target condition input so `model` / `model_name` leaves match the +/// parent as well as the concrete target (AISIX-Cloud#1267). +#[derive(Clone, Copy)] +pub(crate) struct RoutingParent<'a> { + /// The parent Model's `display_name` (the alias the caller sent). + pub name: &'a str, + /// The parent's resource entry id. + pub entry_id: &'a str, +} + /// The request's condition-dimension values at this gate point. Model /// dimensions are absent when no model is resolved (MCP, A2A) — leaves /// on them evaluate false while OR siblings can still match. +/// `routing_parent` is set only at the per-target gate of a routing +/// dispatch; the request gate passes `None` (there the primary values +/// already ARE the requested entry). fn condition_input<'a>( auth: &'a AuthenticatedKey, model_rl: Option<&'a ModelRateLimit>, + routing_parent: Option>, ) -> ConditionInput<'a> { ConditionInput { team: auth.key().team_id.as_deref(), @@ -82,6 +98,8 @@ fn condition_input<'a>( model: model_rl.map(|m| m.entry_id.as_str()), model_name: model_rl.map(|m| m.name.as_str()), provider: model_rl.and_then(|m| m.provider.as_deref()), + routing_parent_model: routing_parent.map(|p| p.entry_id), + routing_parent_model_name: routing_parent.map(|p| p.name), } } @@ -344,7 +362,7 @@ async fn reserve_layers( } // Layer 4+: Rate limit policies from snapshot. - let input = condition_input(auth, model_rl); + let input = condition_input(auth, model_rl, None); let phase = PolicyPhase::Request { defer_model_properties: model_rl.is_some_and(|m| m.routing_parent), }; @@ -521,6 +539,7 @@ pub(crate) async fn reserve_model_only( model_name: &str, model_entry_id: &str, model: &aisix_core::Model, + routing_parent: Option>, ) -> Result { let mut reservations = Vec::new(); @@ -536,8 +555,11 @@ pub(crate) async fn reserve_model_only( reservations.push(r); } - // Policies that follow the model to this target. - let input = condition_input(auth, Some(&mrl)); + // Policies that follow the model to this target. The condition + // input carries the {target, caller-addressed parent} pair so a + // policy pinning the parent's id or alias matches here too + // (AISIX-Cloud#1267). + let input = condition_input(auth, Some(&mrl), routing_parent); reserve_policy_layers( state, snapshot, @@ -565,17 +587,25 @@ pub(crate) async fn reserve_routing_target( state: &ProxyState, snapshot: &aisix_core::AisixSnapshot, auth: &AuthenticatedKey, - is_routing_request: bool, + routing_parent: Option>, target_name: &str, target_entry_id: &str, target: &aisix_core::Model, ) -> Result, ProxyError> { - if !is_routing_request { + let Some(parent) = routing_parent else { return Ok(None); - } - reserve_model_only(state, snapshot, auth, target_name, target_entry_id, target) - .await - .map(Some) + }; + reserve_model_only( + state, + snapshot, + auth, + target_name, + target_entry_id, + target, + Some(parent), + ) + .await + .map(Some) } /// Seconds until the offending window reopens, for a @@ -661,9 +691,14 @@ mod tests { entry_id: &str, auth: &AuthenticatedKey, ) -> String { - match_policy_layer(policy, entry_id, &condition_input(auth, None), REQUEST) - .expect("policy applies") - .bucket_key + match_policy_layer( + policy, + entry_id, + &condition_input(auth, None, None), + REQUEST, + ) + .expect("policy applies") + .bucket_key } #[test] @@ -805,8 +840,13 @@ mod tests { "limits": { "rpm": 100 }, })); let auth = make_auth(Some("team-1"), Some("user-a")); - let layer = match_policy_layer(&policy, "pol-1", &condition_input(&auth, None), REQUEST) - .expect("matches"); + let layer = match_policy_layer( + &policy, + "pol-1", + &condition_input(&auth, None, None), + REQUEST, + ) + .expect("matches"); // No group_by → one shared bucket for every matched request. assert_eq!(layer.bucket_key, "policy:v2:pol-1"); assert_eq!(layer.limits.rpm, Some(100)); @@ -827,7 +867,7 @@ mod tests { let layer = match_policy_layer( &policy, "pol-2", - &condition_input(&auth, Some(&mrl)), + &condition_input(&auth, Some(&mrl), None), REQUEST, ) .expect("matches"); @@ -850,9 +890,13 @@ mod tests { "limits": { "rpm": 20 }, })); let auth = make_auth(Some("team-1"), None); - assert!( - match_policy_layer(&policy, "pol-3", &condition_input(&auth, None), REQUEST).is_none() - ); + assert!(match_policy_layer( + &policy, + "pol-3", + &condition_input(&auth, None, None), + REQUEST + ) + .is_none()); } #[test] @@ -866,18 +910,78 @@ mod tests { })); let auth = make_auth(Some("team-1"), None); let parent = make_model_rl("gpt4-group", "group-1", None); - let input = condition_input(&auth, Some(&parent)); + let input = condition_input(&auth, Some(&parent), None); // Request gate of a routing dispatch: deferred even though the // parent's name would match — the concrete target decides. assert!(match_policy_layer(&policy, "pol-4", &input, REQUEST_DEFERRING).is_none()); // Per-target gate: matches the concrete target. let target = make_model_rl("gpt-4.1-prod", "model-1", Some("openai")); - let target_input = condition_input(&auth, Some(&target)); + let target_input = condition_input(&auth, Some(&target), None); let layer = match_policy_layer(&policy, "pol-4", &target_input, PolicyPhase::ModelTarget) .expect("target matches"); assert_eq!(layer.bucket_key, "policy:v2:pol-4"); } + #[test] + fn group_referencing_policy_matches_at_target_phase_via_parent() { + // AISIX-Cloud#1267: `model in [group uuid]` reserves at the + // per-target gate because the condition input carries the + // {target, parent} pair — previously the parent id was compared + // nowhere and the policy never fired. + let policy = make_conditional_policy(serde_json::json!({ + "name": "group-cap", + "conditions": [ + { "dimension": "model", "operator": "in", "value": ["group-1"] } + ], + "group_by": ["member"], + "limits": { "rph": 3 }, + })); + let auth = make_auth(Some("team-1"), Some("user-a")); + // Request gate of the routing dispatch: still deferred. + let gate = make_model_rl("chat-group", "group-1", None); + let gate_input = condition_input(&auth, Some(&gate), None); + assert!(match_policy_layer(&policy, "pol-9", &gate_input, REQUEST_DEFERRING).is_none()); + // Per-target gate: the parent pair makes it match, bucketed per + // member. + let target = make_model_rl("gpt-4.1-prod", "model-1", Some("openai")); + let parent = RoutingParent { + name: "chat-group", + entry_id: "group-1", + }; + let input = condition_input(&auth, Some(&target), Some(parent)); + let layer = match_policy_layer(&policy, "pol-9", &input, PolicyPhase::ModelTarget) + .expect("group-referencing policy matches via the parent"); + assert_eq!(layer.bucket_key, "policy:v2:pol-9:member=user-a"); + // Direct dispatch to the member (no parent): must NOT match. + let direct = condition_input(&auth, Some(&target), None); + assert!(match_policy_layer(&policy, "pol-9", &direct, REQUEST).is_none()); + } + + #[test] + fn group_by_model_buckets_on_target_not_parent() { + // The pair extends MATCHING only: a group-referencing policy + // splitting by model still buckets on the dispatched target id, + // so per-member counters stay per concrete model. + let policy = make_conditional_policy(serde_json::json!({ + "name": "group-per-model", + "conditions": [ + { "dimension": "model", "operator": "in", "value": ["group-1"] } + ], + "group_by": ["model"], + "limits": { "rpm": 1 }, + })); + let auth = make_auth(Some("team-1"), None); + let target = make_model_rl("gpt-4.1-prod", "model-1", Some("openai")); + let parent = RoutingParent { + name: "chat-group", + entry_id: "group-1", + }; + let input = condition_input(&auth, Some(&target), Some(parent)); + let layer = match_policy_layer(&policy, "pol-10", &input, PolicyPhase::ModelTarget) + .expect("matches via parent"); + assert_eq!(layer.bucket_key, "policy:v2:pol-10:model=model-1"); + } + #[test] fn non_model_policy_not_rereserved_at_target_phase() { let policy = make_conditional_policy(serde_json::json!({ @@ -889,7 +993,7 @@ mod tests { })); let auth = make_auth(Some("team-1"), None); let target = make_model_rl("gpt-4.1-prod", "model-1", Some("openai")); - let input = condition_input(&auth, Some(&target)); + let input = condition_input(&auth, Some(&target), None); // Reserved once at the request gate; the per-target scan must // not double-count it. assert!(match_policy_layer(&policy, "pol-5", &input, PolicyPhase::ModelTarget).is_none()); @@ -913,8 +1017,13 @@ mod tests { "limits": { "rpm": 50 }, })); let auth = make_auth(Some("team-1"), None); - let layer = match_policy_layer(&policy, "pol-6", &condition_input(&auth, None), REQUEST) - .expect("matches via team branch"); + let layer = match_policy_layer( + &policy, + "pol-6", + &condition_input(&auth, None, None), + REQUEST, + ) + .expect("matches via team branch"); assert_eq!(layer.bucket_key, "policy:v2:pol-6"); } @@ -923,7 +1032,7 @@ mod tests { let team_policy = make_scoped_policy("team", "team-1"); let auth = make_auth(Some("team-1"), Some("user-a")); let target = make_model_rl("gpt-4.1-prod", "model-1", Some("openai")); - let input = condition_input(&auth, Some(&target)); + let input = condition_input(&auth, Some(&target), None); assert!( match_policy_layer(&team_policy, "pol-7", &input, PolicyPhase::ModelTarget).is_none() ); diff --git a/crates/aisix-proxy/src/responses.rs b/crates/aisix-proxy/src/responses.rs index 23b79976..69b13591 100644 --- a/crates/aisix-proxy/src/responses.rs +++ b/crates/aisix-proxy/src/responses.rs @@ -704,7 +704,10 @@ async fn dispatch( state, snapshot, auth, - is_routing_request, + is_routing_request.then_some(crate::quota::RoutingParent { + name: &model_entry.value.display_name, + entry_id: &model_entry.id, + }), &target.model.display_name, &target.id, &target.model, diff --git a/schemas/resources/rate_limit_policy.schema.json b/schemas/resources/rate_limit_policy.schema.json index 57a114a8..0598db6a 100644 --- a/schemas/resources/rate_limit_policy.schema.json +++ b/schemas/resources/rate_limit_policy.schema.json @@ -142,14 +142,14 @@ "type": "string" }, { - "description": "Dispatched model entry id (UUID); routing/model groups match per selected target, never the group entry itself.", + "description": "Model entry id (UUID). Matches the dispatched model; on a Model-Group / semantic-router / ensemble dispatch it also matches the requested parent entry, so a group's own id selects every request addressed to that group.", "enum": [ "model" ], "type": "string" }, { - "description": "Dispatched model display name — the string dimension for regex/prefix matching (\"every gpt-4-family alias\").", + "description": "Model display name — the string dimension for regex/prefix matching (\"every gpt-4-family alias\"). Matches the dispatched model's name, and on a virtual-parent dispatch also the requested parent's name.", "enum": [ "model_name" ], diff --git a/tests/e2e/src/cases/group-model-condition-ratelimit-e2e.test.ts b/tests/e2e/src/cases/group-model-condition-ratelimit-e2e.test.ts new file mode 100644 index 00000000..265bfaee --- /dev/null +++ b/tests/e2e/src/cases/group-model-condition-ratelimit-e2e.test.ts @@ -0,0 +1,409 @@ +import { createHash } from "node:crypto"; +import { afterAll, beforeAll, describe, expect, test } from "vitest"; +import { + EtcdClient, + SeedClient, + ProxyClient, + spawnApp, + startOpenAiUpstream, + awaitWindowHeadroom, + waitConfigPropagation, + type OpenAiUpstream, + type SpawnedApp, +} from "../harness/index.js"; + +// E2E for AISIX-Cloud#1267: conditional rate-limit policies whose +// `model` / `model_name` conditions reference a ROUTING GROUP. The +// per-target gate evaluates the {dispatched target, requested parent} +// pair, so a group's own id/alias selects every request addressed to +// the group. Covers: +// +// 1. The reported scenario: `team ∈ {T} AND model ∈ {group uuid}` +// with `group_by: [member]` — throttles per member THROUGH the +// group, with policy attribution; a direct call to the member is +// NOT captured by the group condition even with the bucket hot. +// 2. `model_name == ` matches through the group and not +// on the member's own alias. +// 3. Leaf negate: `model !in [group uuid]` EXCLUDES requests routed +// via the group (previously the member id missed the set, so the +// negated leaf absurdly matched them) while direct dispatch to +// the member stays matched. +// 4. The AISIX-Cloud#1087 principle survives: a MEMBER-id condition +// keeps matching when the member is reached via the group, an +// over-limit member fails over, and the same bucket throttles the +// member's direct alias. +// 5. `/v1/messages` drives the same per-target gate (handler-family +// coverage beyond chat). +// +// Policies reference model UUIDs, so models are seeded FIRST, then the +// policies, then a canary model: etcd watch events arrive in revision +// order, so once the canary lists, every earlier policy row is applied. + +const sha256 = (s: string) => createHash("sha256").update(s).digest("hex"); + +const TEAM_GROUP = "team-1267-group"; +const TEAM_NAME = "team-1267-name"; +const TEAM_NEG = "team-1267-neg"; +const TEAM_MEMBER = "team-1267-member"; +const TEAM_SIB = "team-1267-sib"; + +const POLICY_GROUP = "12670000-0000-0000-0000-00000000000a"; +const POLICY_NAME = "12670000-0000-0000-0000-00000000000b"; +const POLICY_NEG = "12670000-0000-0000-0000-00000000000c"; +const POLICY_MEMBER = "12670000-0000-0000-0000-00000000000d"; +const POLICY_SIB = "12670000-0000-0000-0000-00000000000e"; + +const KEY_G1 = "sk-1267-g1"; +const KEY_G2 = "sk-1267-g2"; +const KEY_NAME = "sk-1267-name"; +const KEY_NEG = "sk-1267-neg"; +const KEY_MEMBER = "sk-1267-member"; +const KEY_SIB = "sk-1267-sib"; +const KEY_FREE = "sk-1267-free"; + +function chatBody(content: string) { + return { + id: "cmpl-1267", + object: "chat.completion", + created: 0, + model: "gpt-4o-mini", + choices: [ + { + index: 0, + message: { role: "assistant", content }, + finish_reason: "stop", + }, + ], + usage: { prompt_tokens: 5, completion_tokens: 3, total_tokens: 8 }, + }; +} + +type ChatResult = { + status: number; + body: { + choices?: Array<{ message?: { content?: string } }>; + error?: { message?: string; type?: string; policy?: { id?: string; name?: string } }; + }; +}; + +describe("group-referencing model conditions e2e (AISIX-Cloud#1267)", () => { + let app: SpawnedApp | undefined; + let etcd: EtcdClient | undefined; + let seed: SeedClient | undefined; + let etcdReachable = false; + const upstreams: OpenAiUpstream[] = []; + + // Seeded model ids the policies pin. + let memberAId = ""; + let groupMainId = ""; + let groupNegId = ""; + + async function newUpstream(body: string): Promise { + const u = await startOpenAiUpstream({ nonStreamBody: chatBody(body) }); + upstreams.push(u); + return u; + } + + async function seedOpenAiModel( + displayName: string, + upstream: OpenAiUpstream, + ): Promise { + if (!seed) throw new Error("seed client not initialized"); + const pk = await seed.createProviderKey({ + display_name: `${displayName}-pk`, + secret: "sk-openai-mock", + api_base: `${upstream.baseUrl}/v1`, + }); + const m = await seed.createModel({ + display_name: displayName, + provider: "openai", + model_name: "gpt-4o-mini", + provider_key_id: pk.id, + }); + return m.id; + } + + beforeAll(async () => { + etcd = new EtcdClient(); + etcdReachable = await etcd.ping(); + if (!etcdReachable) return; + + app = await spawnApp(); + seed = new SeedClient(etcd, app.etcdPrefix); + + // Members + groups first: the policies below embed their ids. + const a = await newUpstream("served-a"); + const b = await newUpstream("served-b"); + const c = await newUpstream("served-c"); + memberAId = await seedOpenAiModel("m-1267-a", a); + await seedOpenAiModel("m-1267-b", b); + await seedOpenAiModel("m-1267-c", c); + groupMainId = ( + await seed.createModel({ + display_name: "grp-1267-main", + routing: { + strategy: "failover", + targets: [{ model: "m-1267-a" }, { model: "m-1267-b" }], + }, + }) + ).id; + groupNegId = ( + await seed.createModel({ + display_name: "grp-1267-neg", + routing: { strategy: "failover", targets: [{ model: "m-1267-c" }] }, + }) + ).id; + + const putPolicy = (id: string, policy: Record) => + etcd!.put( + `${app!.etcdPrefix}/rate_limit_policies/${id}`, + JSON.stringify(policy), + ); + await putPolicy(POLICY_GROUP, { + name: "group-cap-1267", + conditions: [ + { dimension: "team", operator: "in", value: [TEAM_GROUP] }, + { dimension: "model", operator: "in", value: [groupMainId] }, + ], + group_by: ["member"], + limits: { rpm: 1 }, + }); + await putPolicy(POLICY_NAME, { + name: "group-alias-1267", + conditions: [ + { dimension: "team", operator: "in", value: [TEAM_NAME] }, + { dimension: "model_name", operator: "==", value: "grp-1267-main" }, + ], + limits: { rpm: 1 }, + }); + await putPolicy(POLICY_NEG, { + name: "all-but-group-1267", + conditions: [ + { dimension: "team", operator: "in", value: [TEAM_NEG] }, + { dimension: "model", operator: "in", negate: true, value: [groupNegId] }, + ], + limits: { rpm: 1 }, + }); + await putPolicy(POLICY_MEMBER, { + name: "member-cap-1267", + conditions: [ + { dimension: "team", operator: "in", value: [TEAM_MEMBER] }, + { dimension: "model", operator: "in", value: [memberAId] }, + ], + limits: { rpm: 1 }, + }); + await putPolicy(POLICY_SIB, { + name: "sibling-cap-1267", + conditions: [ + { dimension: "team", operator: "in", value: [TEAM_SIB] }, + { dimension: "model", operator: "in", value: [groupMainId] }, + ], + limits: { rpm: 1 }, + }); + + // Caller keys (raw etcd: team_id/user_id are CP-written fields the + // standalone Admin API omits). + const seedKey = ( + id: string, + plaintext: string, + extra: Record = {}, + ) => + etcd!.put( + `${app!.etcdPrefix}/api_keys/${id}`, + JSON.stringify({ + key_hash: sha256(plaintext), + allowed_models: ["*"], + ...extra, + }), + ); + await seedKey("12670001-0000-0000-0000-000000000001", KEY_G1, { + team_id: TEAM_GROUP, + user_id: "user-1267-g1", + }); + await seedKey("12670001-0000-0000-0000-000000000002", KEY_G2, { + team_id: TEAM_GROUP, + user_id: "user-1267-g2", + }); + await seedKey("12670001-0000-0000-0000-000000000003", KEY_NAME, { + team_id: TEAM_NAME, + }); + await seedKey("12670001-0000-0000-0000-000000000004", KEY_NEG, { + team_id: TEAM_NEG, + }); + await seedKey("12670001-0000-0000-0000-000000000005", KEY_MEMBER, { + team_id: TEAM_MEMBER, + }); + await seedKey("12670001-0000-0000-0000-000000000006", KEY_SIB, { + team_id: TEAM_SIB, + }); + await seedKey("12670001-0000-0000-0000-000000000007", KEY_FREE); + + // Canary AFTER every policy/key write: revision order means its + // visibility proves the rows above are applied. + const canary = await newUpstream("served-canary"); + await seedOpenAiModel("m-1267-canary", canary); + await waitModelsListed(KEY_FREE, [ + "m-1267-a", + "m-1267-b", + "m-1267-c", + "grp-1267-main", + "grp-1267-neg", + "m-1267-canary", + ]); + }); + + afterAll(async () => { + await app?.exit(); + await Promise.all(upstreams.map((u) => u.close())); + }); + + // Listing consumes no rpm slot, so probing never burns the buckets + // under test. + async function waitModelsListed(apiKey: string, names: string[]): Promise { + if (!app) throw new Error("app not initialized"); + const probe = new ProxyClient(app.proxyUrl, apiKey); + await waitConfigPropagation(async () => { + const res = await probe.listModels(); + if (res.status !== 200) return false; + const data = (res.body as { data?: Array<{ id?: string }> }).data ?? []; + return names.every((n) => data.some((m) => m.id === n)); + }); + } + + async function chatRaw(apiKey: string, model: string): Promise { + if (!app) throw new Error("app not initialized"); + const res = await fetch(`${app.proxyUrl}/v1/chat/completions`, { + method: "POST", + headers: { + authorization: `Bearer ${apiKey}`, + "content-type": "application/json", + }, + body: JSON.stringify({ + model, + messages: [{ role: "user", content: "hello" }], + }), + }); + return { status: res.status, body: (await res.json()) as ChatResult["body"] }; + } + + function servedContent(r: ChatResult): string { + expect(r.status).toBe(200); + return r.body.choices?.[0]?.message?.content ?? ""; + } + + test("group-id condition throttles per member through the group; direct member calls escape it", async (ctx) => { + if (!etcdReachable || !app) { + ctx.skip(); + return; + } + await awaitWindowHeadroom(5); + + // Member g1 burns their slot through the group; the 2nd call 429s + // with attribution to the group policy. + expect(servedContent(await chatRaw(KEY_G1, "grp-1267-main"))).toBe("served-a"); + const throttled = await chatRaw(KEY_G1, "grp-1267-main"); + expect(throttled.status).toBe(429); + expect(throttled.body.error?.type).toBe("rate_limit_exceeded"); + expect(throttled.body.error?.policy).toEqual({ + id: POLICY_GROUP, + name: "group-cap-1267", + }); + + // Same team, different member: independent bucket. + expect(servedContent(await chatRaw(KEY_G2, "grp-1267-main"))).toBe("served-a"); + + // Direct dispatch to the member is NOT addressed to the group, so + // the group condition must not capture it — even with g1's group + // bucket exhausted. + expect(servedContent(await chatRaw(KEY_G1, "m-1267-a"))).toBe("served-a"); + }); + + test("model_name == group alias matches through the group only", async (ctx) => { + if (!etcdReachable || !app) { + ctx.skip(); + return; + } + await awaitWindowHeadroom(5); + + expect(servedContent(await chatRaw(KEY_NAME, "grp-1267-main"))).toBe("served-a"); + const throttled = await chatRaw(KEY_NAME, "grp-1267-main"); + expect(throttled.status).toBe(429); + expect(throttled.body.error?.policy).toEqual({ + id: POLICY_NAME, + name: "group-alias-1267", + }); + + // The member's own alias is not the group alias: passes. + expect(servedContent(await chatRaw(KEY_NAME, "m-1267-a"))).toBe("served-a"); + }); + + test("negated group condition excludes via-group requests and keeps direct dispatch matched", async (ctx) => { + if (!etcdReachable || !app) { + ctx.skip(); + return; + } + await awaitWindowHeadroom(5); + + // Direct dispatch to the member matches `model !in [group]` and + // burns the shared bucket. + expect(servedContent(await chatRaw(KEY_NEG, "m-1267-c"))).toBe("served-c"); + expect((await chatRaw(KEY_NEG, "m-1267-c")).status).toBe(429); + + // Via the excluded group the SAME member escapes the policy even + // with the bucket hot — negate flips the pair result, so "everything + // except this group" no longer (absurdly) matches the group's own + // traffic. + expect(servedContent(await chatRaw(KEY_NEG, "grp-1267-neg"))).toBe("served-c"); + }); + + test("member-id condition still matches via the group, fails over, and shares its bucket with the direct alias", async (ctx) => { + if (!etcdReachable || !app) { + ctx.skip(); + return; + } + await awaitWindowHeadroom(5); + + // 1st via group lands on member a (failover order) and burns the + // member policy's bucket. + expect(servedContent(await chatRaw(KEY_MEMBER, "grp-1267-main"))).toBe("served-a"); + // 2nd via group: a is over the member policy → failed attempt → + // fails over to b, which the member condition does not match. + expect(servedContent(await chatRaw(KEY_MEMBER, "grp-1267-main"))).toBe("served-b"); + // Direct dispatch to a hits the same exhausted bucket at the + // request gate. + const direct = await chatRaw(KEY_MEMBER, "m-1267-a"); + expect(direct.status).toBe(429); + expect(direct.body.error?.policy).toEqual({ + id: POLICY_MEMBER, + name: "member-cap-1267", + }); + }); + + test("/v1/messages drives the same per-target gate for group-id conditions", async (ctx) => { + if (!etcdReachable || !app) { + ctx.skip(); + return; + } + await awaitWindowHeadroom(5); + + const messagesRaw = async (): Promise => { + const res = await fetch(`${app!.proxyUrl}/v1/messages`, { + method: "POST", + headers: { + authorization: `Bearer ${KEY_SIB}`, + "content-type": "application/json", + }, + body: JSON.stringify({ + model: "grp-1267-main", + max_tokens: 32, + messages: [{ role: "user", content: "hello" }], + }), + }); + await res.text(); + return res.status; + }; + + expect(await messagesRaw()).toBe(200); + expect(await messagesRaw()).toBe(429); + }); +}); From a1ea94fb6e8a11b273b8cdee4373e7c0624b7168 Mon Sep 17 00:00:00 2001 From: Jarvis Date: Wed, 12 Aug 2026 17:13:19 +0800 Subject: [PATCH 2/3] fix(core): evaluate ~= conjunctively over the identity pair The two identities in the {target, parent} pair are distinct strings, so combining ~= disjunctively made it vacuously true on every routed request and broke the a ~= b == !(a == b) equivalence the operator vocabulary (and the dashboard's operator normalization) relies on. Positive operators keep the exists-reading; ~= now requires every identity to differ. Also extends the e2e to the streaming-chat and /v1/responses loops per review. --- .../src/models/policy_conditions.rs | 62 +++++++++++--- ...roup-model-condition-ratelimit-e2e.test.ts | 81 +++++++++++++++++++ 2 files changed, 133 insertions(+), 10 deletions(-) diff --git a/crates/aisix-core/src/models/policy_conditions.rs b/crates/aisix-core/src/models/policy_conditions.rs index 576d5245..9a3ae496 100644 --- a/crates/aisix-core/src/models/policy_conditions.rs +++ b/crates/aisix-core/src/models/policy_conditions.rs @@ -32,10 +32,13 @@ //! - the model dimensions (`model` / `model_name`) evaluate against a //! PAIR on a routing/ensemble/semantic dispatch: the dispatched //! target and the caller-addressed parent entry (AISIX-Cloud#1267). -//! A leaf is raw-true when either identity satisfies its operator; -//! `negate` flips that combined result, so `!(model in [group])` -//! excludes every request addressed to the group instead of matching -//! all of them; +//! Positive operators are raw-true when EITHER identity satisfies +//! them (∃); `~=` is raw-true only when BOTH differ (∀ — the two +//! identities are distinct strings, so an ∃ reading would be +//! vacuously true on every routed request), preserving `a ~= b` ≡ +//! `!(a == b)`. `negate` flips the combined result, so `!(model in +//! [group])` excludes every request addressed to the group instead +//! of matching all of them; //! - regexes are compiled once per distinct pattern into a process-wide //! cache. Load-time validation guarantees compilability, so a cache //! miss at evaluation time never fails in practice; a pattern that @@ -473,13 +476,21 @@ fn eval_leaf(leaf: &PolicyCondition, input: &ConditionInput<'_>) -> bool { return false; }; // Model dimensions carry a {dispatched target, requested parent} - // pair on a routing dispatch: raw-true when either satisfies the - // operator, and `negate` flips the combined result — so a negated + // pair on a routing dispatch — the request's model-identity SET. + // Positive operators (==/in/regex) ask "does ANY identity satisfy" + // (∃); the negative operator `~=` asks "do ALL identities differ" + // (∀) — the two identities are distinct strings, so an ∃ reading + // of `~=` would be vacuously true on every routed request. This + // keeps `a ~= b` ≡ `!(a == b)` over the pair, the equivalence the + // operator vocabulary (and the dashboard's normalization) is + // built on; `negate` then flips the combined result, so a negated // leaf excluding the parent excludes every request addressed to it. - let raw = eval_operator(leaf, var) - || input - .routing_parent(leaf.dimension) - .is_some_and(|parent| eval_operator(leaf, parent)); + let parent = input.routing_parent(leaf.dimension); + let raw = if leaf.operator == ConditionOperator::Ne { + eval_operator(leaf, var) && parent.is_none_or(|p| eval_operator(leaf, p)) + } else { + eval_operator(leaf, var) || parent.is_some_and(|p| eval_operator(leaf, p)) + }; raw != leaf.negate } @@ -758,6 +769,37 @@ mod tests { assert!(eval_condition_nodes(&nodes, &input())); } + #[test] + fn ne_leaf_requires_both_identities_to_differ() { + // `model ~= group-1` on a request ADDRESSED to group-1: the + // dispatched member differs from the value, but ∃-combining + // would make `~=` vacuously true on every routed request (the + // two identities are distinct strings). ∀-combining keeps the + // exclusion meaningful… + let ne = |v: &str| vec![leaf(PolicyDimension::Model, ConditionOperator::Ne, one(v))]; + assert!(!eval_condition_nodes(&ne("group-1"), &routed_input())); + assert!(!eval_condition_nodes(&ne("model-1"), &routed_input())); + assert!(eval_condition_nodes(&ne("other"), &routed_input())); + // …and preserves `a ~= b` ≡ `!(a == b)` over the pair. + let neg_eq = |v: &str| { + vec![neg_leaf( + PolicyDimension::Model, + ConditionOperator::Eq, + one(v), + )] + }; + for v in ["group-1", "model-1", "other"] { + assert_eq!( + eval_condition_nodes(&ne(v), &routed_input()), + eval_condition_nodes(&neg_eq(v), &routed_input()), + "~= and !(==) diverged for {v}" + ); + } + // Direct dispatch (no parent): plain not-equal, unchanged. + assert!(eval_condition_nodes(&ne("group-1"), &input())); + assert!(!eval_condition_nodes(&ne("model-1"), &input())); + } + #[test] fn member_leaf_still_matches_through_parent() { // The 1087 principle survives the pair: a member-id condition diff --git a/tests/e2e/src/cases/group-model-condition-ratelimit-e2e.test.ts b/tests/e2e/src/cases/group-model-condition-ratelimit-e2e.test.ts index 265bfaee..277e8390 100644 --- a/tests/e2e/src/cases/group-model-condition-ratelimit-e2e.test.ts +++ b/tests/e2e/src/cases/group-model-condition-ratelimit-e2e.test.ts @@ -46,12 +46,16 @@ const TEAM_NAME = "team-1267-name"; const TEAM_NEG = "team-1267-neg"; const TEAM_MEMBER = "team-1267-member"; const TEAM_SIB = "team-1267-sib"; +const TEAM_STREAM = "team-1267-stream"; +const TEAM_RESP = "team-1267-resp"; const POLICY_GROUP = "12670000-0000-0000-0000-00000000000a"; const POLICY_NAME = "12670000-0000-0000-0000-00000000000b"; const POLICY_NEG = "12670000-0000-0000-0000-00000000000c"; const POLICY_MEMBER = "12670000-0000-0000-0000-00000000000d"; const POLICY_SIB = "12670000-0000-0000-0000-00000000000e"; +const POLICY_STREAM = "12670000-0000-0000-0000-00000000000f"; +const POLICY_RESP = "12670000-0000-0000-0000-000000000010"; const KEY_G1 = "sk-1267-g1"; const KEY_G2 = "sk-1267-g2"; @@ -59,6 +63,8 @@ const KEY_NAME = "sk-1267-name"; const KEY_NEG = "sk-1267-neg"; const KEY_MEMBER = "sk-1267-member"; const KEY_SIB = "sk-1267-sib"; +const KEY_STREAM = "sk-1267-stream"; +const KEY_RESP = "sk-1267-resp"; const KEY_FREE = "sk-1267-free"; function chatBody(content: string) { @@ -200,6 +206,22 @@ describe("group-referencing model conditions e2e (AISIX-Cloud#1267)", () => { ], limits: { rpm: 1 }, }); + await putPolicy(POLICY_STREAM, { + name: "stream-cap-1267", + conditions: [ + { dimension: "team", operator: "in", value: [TEAM_STREAM] }, + { dimension: "model", operator: "in", value: [groupMainId] }, + ], + limits: { rpm: 1 }, + }); + await putPolicy(POLICY_RESP, { + name: "responses-cap-1267", + conditions: [ + { dimension: "team", operator: "in", value: [TEAM_RESP] }, + { dimension: "model", operator: "in", value: [groupMainId] }, + ], + limits: { rpm: 1 }, + }); // Caller keys (raw etcd: team_id/user_id are CP-written fields the // standalone Admin API omits). @@ -237,6 +259,12 @@ describe("group-referencing model conditions e2e (AISIX-Cloud#1267)", () => { team_id: TEAM_SIB, }); await seedKey("12670001-0000-0000-0000-000000000007", KEY_FREE); + await seedKey("12670001-0000-0000-0000-000000000008", KEY_STREAM, { + team_id: TEAM_STREAM, + }); + await seedKey("12670001-0000-0000-0000-000000000009", KEY_RESP, { + team_id: TEAM_RESP, + }); // Canary AFTER every policy/key write: revision order means its // visibility proves the rows above are applied. @@ -406,4 +434,57 @@ describe("group-referencing model conditions e2e (AISIX-Cloud#1267)", () => { expect(await messagesRaw()).toBe(200); expect(await messagesRaw()).toBe(429); }); + + test("streaming chat drives the same per-target gate for group-id conditions", async (ctx) => { + if (!etcdReachable || !app) { + ctx.skip(); + return; + } + await awaitWindowHeadroom(5); + + const streamRaw = async (): Promise => { + const res = await fetch(`${app!.proxyUrl}/v1/chat/completions`, { + method: "POST", + headers: { + authorization: `Bearer ${KEY_STREAM}`, + "content-type": "application/json", + }, + body: JSON.stringify({ + model: "grp-1267-main", + messages: [{ role: "user", content: "hello" }], + stream: true, + }), + }); + // Drain so the connection finalises before the next call. + await res.text(); + return res.status; + }; + + expect(await streamRaw()).toBe(200); + expect(await streamRaw()).toBe(429); + }); + + test("/v1/responses drives the same per-target gate for group-id conditions", async (ctx) => { + if (!etcdReachable || !app) { + ctx.skip(); + return; + } + await awaitWindowHeadroom(5); + + const responsesRaw = async (): Promise => { + const res = await fetch(`${app!.proxyUrl}/v1/responses`, { + method: "POST", + headers: { + authorization: `Bearer ${KEY_RESP}`, + "content-type": "application/json", + }, + body: JSON.stringify({ model: "grp-1267-main", input: "hello" }), + }); + await res.text(); + return res.status; + }; + + expect(await responsesRaw()).toBe(200); + expect(await responsesRaw()).toBe(429); + }); }); From 3c21bf96c723e16e55a017db2a7c2155d4618196 Mon Sep 17 00:00:00 2001 From: Jarvis Date: Wed, 12 Aug 2026 17:27:33 +0800 Subject: [PATCH 3/3] test(e2e): serve the streaming case from a real SSE fixture MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The mock answers SSE or JSON per fixture, so the streaming case gets its own member + group and asserts the event-stream content type — the reservation now provably rides the streaming loop, not a JSON fallback. --- ...roup-model-condition-ratelimit-e2e.test.ts | 58 +++++++++++++++++-- 1 file changed, 52 insertions(+), 6 deletions(-) diff --git a/tests/e2e/src/cases/group-model-condition-ratelimit-e2e.test.ts b/tests/e2e/src/cases/group-model-condition-ratelimit-e2e.test.ts index 277e8390..783cf05e 100644 --- a/tests/e2e/src/cases/group-model-condition-ratelimit-e2e.test.ts +++ b/tests/e2e/src/cases/group-model-condition-ratelimit-e2e.test.ts @@ -103,6 +103,7 @@ describe("group-referencing model conditions e2e (AISIX-Cloud#1267)", () => { let memberAId = ""; let groupMainId = ""; let groupNegId = ""; + let groupStreamId = ""; async function newUpstream(body: string): Promise { const u = await startOpenAiUpstream({ nonStreamBody: chatBody(body) }); @@ -159,6 +160,45 @@ describe("group-referencing model conditions e2e (AISIX-Cloud#1267)", () => { routing: { strategy: "failover", targets: [{ model: "m-1267-c" }] }, }) ).id; + // The streaming case gets its own member + group: the mock serves + // SSE or JSON per FIXTURE (not per request), so the shared members + // must stay non-streaming while this one answers real SSE. + const sse = await startOpenAiUpstream({ + streamEvents: [ + JSON.stringify({ + id: "mock-1267-s", + object: "chat.completion.chunk", + model: "gpt-4o-mini", + choices: [{ index: 0, delta: { role: "assistant" } }], + }), + JSON.stringify({ + id: "mock-1267-s", + object: "chat.completion.chunk", + model: "gpt-4o-mini", + choices: [{ index: 0, delta: { content: "served-s" }, finish_reason: "stop" }], + usage: { prompt_tokens: 5, completion_tokens: 3, total_tokens: 8 }, + }), + "[DONE]", + ], + }); + upstreams.push(sse); + const sPk = await seed.createProviderKey({ + display_name: "m-1267-s-pk", + secret: "sk-openai-mock", + api_base: `${sse.baseUrl}/v1`, + }); + await seed.createModel({ + display_name: "m-1267-s", + provider: "openai", + model_name: "gpt-4o-mini", + provider_key_id: sPk.id, + }); + groupStreamId = ( + await seed.createModel({ + display_name: "grp-1267-stream", + routing: { strategy: "failover", targets: [{ model: "m-1267-s" }] }, + }) + ).id; const putPolicy = (id: string, policy: Record) => etcd!.put( @@ -210,7 +250,7 @@ describe("group-referencing model conditions e2e (AISIX-Cloud#1267)", () => { name: "stream-cap-1267", conditions: [ { dimension: "team", operator: "in", value: [TEAM_STREAM] }, - { dimension: "model", operator: "in", value: [groupMainId] }, + { dimension: "model", operator: "in", value: [groupStreamId] }, ], limits: { rpm: 1 }, }); @@ -274,8 +314,10 @@ describe("group-referencing model conditions e2e (AISIX-Cloud#1267)", () => { "m-1267-a", "m-1267-b", "m-1267-c", + "m-1267-s", "grp-1267-main", "grp-1267-neg", + "grp-1267-stream", "m-1267-canary", ]); }); @@ -442,7 +484,7 @@ describe("group-referencing model conditions e2e (AISIX-Cloud#1267)", () => { } await awaitWindowHeadroom(5); - const streamRaw = async (): Promise => { + const streamRaw = async (): Promise<{ status: number; contentType: string }> => { const res = await fetch(`${app!.proxyUrl}/v1/chat/completions`, { method: "POST", headers: { @@ -450,18 +492,22 @@ describe("group-referencing model conditions e2e (AISIX-Cloud#1267)", () => { "content-type": "application/json", }, body: JSON.stringify({ - model: "grp-1267-main", + model: "grp-1267-stream", messages: [{ role: "user", content: "hello" }], stream: true, }), }); // Drain so the connection finalises before the next call. await res.text(); - return res.status; + return { status: res.status, contentType: res.headers.get("content-type") ?? "" }; }; - expect(await streamRaw()).toBe(200); - expect(await streamRaw()).toBe(429); + const first = await streamRaw(); + expect(first.status).toBe(200); + // Real SSE, so the streaming loop (not a JSON fallback) held the + // per-target reservation. + expect(first.contentType).toContain("text/event-stream"); + expect((await streamRaw()).status).toBe(429); }); test("/v1/responses drives the same per-target gate for group-id conditions", async (ctx) => {