From a16b11a8e0357553d3072ba4efbac597f56e92a5 Mon Sep 17 00:00:00 2001 From: fylorn <249551762+fylorn@users.noreply.github.com> Date: Thu, 22 Oct 2026 20:39:45 +0800 Subject: [PATCH] Record requests refused before reaching any upstream A request whose key was recognised but that was refused before routing finished left no trace: no RequestStarted, no RequestRouted, nothing in the request store or in-flight. A mistyped model, a model no upstream offers, a model outside the key's allow list, a request no rule matches, or a gateway with no upstreams configured could not be seen in the traffic view. These now take the same path as a rule denial (Routed::Refused): the request starts with an empty upstream, is routed with an empty attempt chain, and fails with the original error code and arguments. The model on the row is the one the client sent. When no rule decided the route (no match, evaluation error, no upstreams) the deciding rule is empty. The response the client receives is unchanged (same status and body). Requests rejected before the key is identified (missing or invalid key, source address not allowed, non-POST, oversized body) are still not recorded. WebSocket: - Each Responses turn that is not sent (model outside the key's allow list, alias the connected upstream cannot serve, no rule matched, rule evaluation error) records a row, like a rule-denied turn. - A Realtime upgrade refused for its model, and an upgrade whose routing fails, record a row. Route stats do not count an empty rule name as a rule; such rows count toward the route only. Co-Authored-By: Claude Opus 5.5 --- crates/tw-api/src/lib.rs | 7 +- crates/tw-control/tests/route_stats.rs | 146 ++++++++++++++++---- crates/tw-gateway/src/server.rs | 9 +- crates/tw-gateway/src/server/pipeline.rs | 77 ++++++----- crates/tw-gateway/src/server/upgrade.rs | 77 ++++++----- crates/tw-gateway/src/ws.rs | 69 +++++----- crates/tw-gateway/src/ws/turn.rs | 28 ++-- crates/tw-gateway/tests/alias_ws.rs | 82 ++++++++++-- crates/tw-gateway/tests/models.rs | 62 ++++++++- crates/tw-gateway/tests/routing_facts.rs | 164 ++++++++++++++++++++++- crates/tw-store/src/db.rs | 23 +++- 11 files changed, 588 insertions(+), 156 deletions(-) diff --git a/crates/tw-api/src/lib.rs b/crates/tw-api/src/lib.rs index 4e8a19d3..eb236734 100644 --- a/crates/tw-api/src/lib.rs +++ b/crates/tw-api/src/lib.rs @@ -1381,9 +1381,10 @@ pub enum Event { /// 握手有了结果就发；Responses 连接上的一轮，上游这一轮的第一帧到了就发（没等到就在 /// 结局之前），那一跳是这条连接连着的那一家。 /// - /// **规则做了决定、请求却一家上游都没到的也发**：规则拒绝了它（第一阶段）， - /// 或者规则选中的上游都服务不了这个模型 —— 那时尝试链是空的，紧跟着一条 - /// 失败。每条规则命中了多少请求要数得到它们（见 `GET /summary/routes`）。 + /// **认出了密钥、请求却一家上游都没到的也发**：规则拒绝了它（第一阶段），模型准入 + /// 没过（模型名没有上游提供、密钥不让用），规则选中的上游都服务不了这个模型，没有一条 + /// 规则命中，还没配上游 —— 那时尝试链是空的，紧跟着一条失败。每条规则命中了多少请求 + /// 要数得到它们（见 `GET /summary/routes`）；规则还没做出决定的，`rule` 是空的。 /// /// 带着的是这次请求**完整的路由记录**，落库的就是它（[`RoutingView`]）：开始 /// 事件里那几项（路由、规则、策略组、改写）在这里再给一遍，是这一份的终稿 —— diff --git a/crates/tw-control/tests/route_stats.rs b/crates/tw-control/tests/route_stats.rs index e373bb0f..be859251 100644 --- a/crates/tw-control/tests/route_stats.rs +++ b/crates/tw-control/tests/route_stats.rs @@ -79,13 +79,19 @@ async fn ask(gw: SocketAddr, model: &str) -> u16 { .as_u16() } -#[tokio::test] -async fn every_route_and_rule_counts_what_went_through_it_including_denials() { +/// 真网关、真存储层、控制面，和 `twcore` 一样接：网关的事件经总线落库 +struct World { + _dir: tempfile::TempDir, + gw: SocketAddr, + store: Arc>, + app: axum::Router, +} + +async fn world(text: &str) -> World { let d = tempfile::tempdir().unwrap(); - let text = yaml(upstream().await); let p = d.path().join("config.yaml"); - std::fs::write(&p, &text).unwrap(); - let cfg: tw_config::Config = serde_yaml_ng::from_str(&text).unwrap(); + std::fs::write(&p, text).unwrap(); + let cfg: tw_config::Config = serde_yaml_ng::from_str(text).unwrap(); let gw = tw_gateway::AppState::new(cfg).unwrap(); // 和 twcore 一样：存储层从总线给它的那条通道收事件，落库 let (_bodies, rx) = tokio::sync::mpsc::channel(8); @@ -113,33 +119,56 @@ async fn every_route_and_rule_counts_what_went_through_it_including_denials() { let addr = tw_gateway::serve(gw, ([127, 0, 0, 1], 0).into()) .await .unwrap(); + World { + _dir: d, + gw: addr, + store, + app, + } +} + +impl World { + /// 落库在另一个任务上：等 `n` 行都写下 + async fn landed(&self, n: i64) { + let deadline = std::time::Instant::now() + Duration::from_secs(5); + while self.store.lock().await.db().count().unwrap() < n { + assert!(std::time::Instant::now() < deadline, "{n} 个请求没有都落库"); + tokio::time::sleep(Duration::from_millis(20)).await; + } + } + + async fn get(&self, path: &str) -> serde_json::Value { + let resp = self + .app + .clone() + .oneshot(Request::get(path).body(Body::empty()).unwrap()) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::OK); + let body = axum::body::to_bytes(resp.into_body(), usize::MAX) + .await + .unwrap(); + serde_json::from_slice(&body).unwrap() + } +} + +#[tokio::test] +async fn every_route_and_rule_counts_what_went_through_it_including_denials() { + let w = world(&yaml(upstream().await)).await; + let addr = w.gw; let before = now_ms(); assert_eq!(ask(addr, "claude-sonnet-4-5").await, 200); assert_eq!(ask(addr, "claude-haiku-4-5").await, 200); assert_eq!(ask(addr, "claude-opus-4-1").await, 403, "该被规则拒绝"); - - // 落库在另一个任务上：等三行都写下 - let deadline = std::time::Instant::now() + Duration::from_secs(5); - while store.lock().await.db().count().unwrap() < 3 { - assert!(std::time::Instant::now() < deadline, "三个请求没有都落库"); - tokio::time::sleep(Duration::from_millis(20)).await; - } + w.landed(3).await; // 窗口的右端给到将来：缺省是「此刻」，而最后那个请求可能就开始在这一毫秒 - let resp = app - .oneshot( - Request::get("/summary/routes?from_ms=0&to_ms=99999999999999") - .body(Body::empty()) - .unwrap(), - ) - .await - .unwrap(); - assert_eq!(resp.status(), StatusCode::OK); - let body = axum::body::to_bytes(resp.into_body(), usize::MAX) - .await - .unwrap(); - let stats: tw_api::RouteStats = serde_json::from_slice(&body).unwrap(); + let stats: tw_api::RouteStats = serde_json::from_value( + w.get("/summary/routes?from_ms=0&to_ms=99999999999999") + .await, + ) + .unwrap(); // 库是这个测试里新建的：记录从第一个请求开始，不是问的起点（0） let covered = stats.covered_since_ms.expect("库里有请求，记录有起点"); assert!( @@ -171,6 +200,73 @@ async fn every_route_and_rule_counts_what_went_through_it_including_denials() { assert_eq!(rule("从没命中过"), None, "没命中过的规则不在里面"); } +/// 认出了密钥、却一家上游都没去的请求**落进请求库**：密钥不让用的模型（准入没过）、没有一条 +/// 规则命中的。流量里那一行的模型是客户端写的、错误是告诉客户端的那一句；路由的统计数得到 +/// 它们，规则没做出决定的只算进路由，不冒出一条没有名字的规则 +#[tokio::test] +async fn a_request_refused_before_any_upstream_is_a_row_with_the_model_the_client_asked_for() { + let up = upstream().await; + let w = world(&format!( + r#"version: 1 +listen: + control: + key: c0ffee00c0ffee00c0ffee00c0ffee00c0ffee00c0ffee00c0ffee00c0ffee00 +clients: + - name: 我 + key: tw-k + route: 工作 + allow: [claude-sonnet-*] +providers: + - name: up + base_url: http://{up} + key: sk-x + protocol: anthropic +routes: + - name: 工作 + rules: + - name: 只走 claude + when: {{ model: claude-* }} + to: up +"# + )) + .await; + + assert_eq!(ask(w.gw, "claude-opus-4-1").await, 400, "密钥不让用"); + assert_eq!(ask(w.gw, "gpt-5").await, 500, "没有一条规则命中"); + w.landed(2).await; + + let history = w.get("/history?limit=10").await; + let row = |model: &str| { + history + .as_array() + .unwrap() + .iter() + .find(|r| r["model"] == model) + .unwrap_or_else(|| panic!("流量里没有 {model}：{history}")) + .clone() + }; + let barred = row("claude-opus-4-1"); + assert_eq!(barred["error"]["code"], "gw.model.not_allowed", "{barred}"); + assert_eq!(barred["error"]["args"]["model"], "claude-opus-4-1"); + assert_eq!(barred["client"], "我"); + assert_eq!(barred["routing"]["rule"], "只走 claude", "{barred}"); + assert_eq!(barred["routing"]["attempts"], serde_json::json!([])); + let unrouted = row("gpt-5"); + assert_eq!(unrouted["error"]["code"], "gw.route.failed", "{unrouted}"); + assert_eq!(unrouted["routing"]["route"], "工作", "{unrouted}"); + assert_eq!(unrouted["routing"]["rule"], "", "{unrouted}"); + + let stats: tw_api::RouteStats = serde_json::from_value( + w.get("/summary/routes?from_ms=0&to_ms=99999999999999") + .await, + ) + .unwrap(); + let work = &stats.routes[0]; + assert_eq!((work.requests, work.failed), (2, 2), "{work:?}"); + let names: Vec<&str> = work.rules.iter().map(|h| h.rule.as_str()).collect(); + assert_eq!(names, ["只走 claude"], "{work:?}"); +} + /// 还没有一条记录的库：线上的起点是 null，界面不会把空表读成「一周都没命中」。 #[tokio::test] async fn a_store_with_no_history_yet_says_nothing_is_covered() { diff --git a/crates/tw-gateway/src/server.rs b/crates/tw-gateway/src/server.rs index 2ad1b029..640050fa 100644 --- a/crates/tw-gateway/src/server.rs +++ b/crates/tw-gateway/src/server.rs @@ -344,11 +344,12 @@ pub(crate) struct Choice { pub(crate) stayed_on: Option, } -/// 规则做了决定、这个请求却一家上游都不会去时的路由事件：尝试链是空的。 +/// 认出了密钥、这个请求却一家上游都不会去时的路由事件：尝试链是空的。 /// -/// **这样的请求照样留一行**（开始、这一条、然后一条失败）：被规则拒绝的、规则 -/// 选中的上游都服务不了它的，和别的失败一样要在流量里看得见，每条规则命中了 -/// 多少也要数得到它们。 +/// **这样的请求照样留一行**（开始、这一条、然后一条失败）：被规则拒绝的、模型准入没过的 +/// （模型名写错了、没有上游提供、密钥不让用）、规则选中的上游都服务不了它的、规则没做出 +/// 决定的、还没配上游的，和别的失败一样要在流量里看得见，每条规则命中了多少也要数得到 +/// 它们。规则没做出决定的，`choice` 里决定去向的规则是空的。 pub(crate) fn routed_nowhere(id: u64, choice: Choice) -> tw_api::Event { tw_api::Event::RequestRouted { id, diff --git a/crates/tw-gateway/src/server/pipeline.rs b/crates/tw-gateway/src/server/pipeline.rs index bd15a3fc..daaf3b5e 100644 --- a/crates/tw-gateway/src/server/pipeline.rs +++ b/crates/tw-gateway/src/server/pipeline.rs @@ -103,26 +103,16 @@ pub(super) async fn pipeline( Probe::Intent(intent) => intent, }; - // 首次运行还没配完是正常状态，不是配置错误。这条要在路由之前挡， - // 因为「一个 provider 都没有」时任何路由结果都是空的，而那条错误 - // 说不清下一步。 - if rt.config.providers.is_empty() { - return Err(GatewayError::config(msg!( - "gw.config.no_upstreams" => - "No upstream is configured yet. Add one in ThinkWatch Lite, or under `providers` in \ - config.yaml." - ))); - } - // 管线第 2 步：读出路由事实，路由。**看的是客户端的原话**：插件的请求钩子排在路由 // 之后（每发往一个上游跑一次，见 `plug`），左右不了请求去哪一家 let (reading, fp) = heavy(&req.body, || { read(&req, intent, parsed.as_ref(), (state.local_time)()) }); let conv = conversation(&rt, &req, &reading, fp.as_deref()); - let (choice, decision) = match route(&state, &rt, &req, &reading, conv.as_ref())? { + let (choice, decision) = match route(&state, &rt, &req, &reading, conv.as_ref()) { Routed::Go(choice, decision) => (choice, decision), - // 规则做了决定，请求却一家上游都不会去。**照样开始、照样报路由**，失败由 + // 请求一家上游都不会去：还没配上游、规则没做出决定、规则拒绝了、模型准入没过、规则 + // 选中的上游都服务不了。**照样开始、照样报路由**（尝试链是空的），失败由 // `passthrough` 按这个错误报 —— 不排队：它一个字节都不会发出去 Routed::Refused(choice, why) => { let to = ("", tw_api::Billing::PerToken); @@ -604,6 +594,9 @@ fn outgrown( /// 目录空着时不看清单：那说明探测还没回来或者上游都不给列表，这时候拦等于把整个网关 /// 关掉。**密钥的 `allow` 照样看**：它写在配置里，和清单问没问到无关 —— 不看的话， /// 下一步按密钥跳过候选时，请求会以「没有上游可用」被拒，说不清是密钥的事。 +/// +/// 没过的请求**照样留一行**（[`Routed::Refused`]）：模型名写错了、密钥不让用，要在流量里 +/// 看得见才排查得了。 fn admit( state: &AppState, rt: &Runtime, @@ -752,8 +745,10 @@ fn admit( enum Routed { /// 选出了候选，照常往下走 Go(Choice, tw_engine::Decision), - /// 规则做了决定，请求却一家上游都不会去：规则拒绝了它，或者规则选中的上游 - /// 都服务不了它。**这一行要留下**（见 [`super::routed_nowhere`]） + /// 请求一家上游都不会去：还没配上游，规则没做出决定（没有一条命中、求不了值），规则 + /// 拒绝了它，模型准入没过（[`admit`]），或者规则选中的上游都服务不了它。**这一行要 + /// 留下**（见 [`super::routed_nowhere`]）：密钥认出来了，被拒的请求和别的失败一样要在 + /// 流量里看得见。规则还没做出决定的，`Choice` 里决定去向的规则是空的 Refused(Choice, GatewayError), } @@ -761,20 +756,36 @@ enum Routed { /// provider」就是留给这一段的接缝。 /// /// 出来的候选已经去掉了服务不了的、按策略组排好了序；熔断过滤在并发 -/// 闸门之后（见 [`start`]）。 -/// -/// 返回的错误是**规则还没做出决定**的那些（没有一条规则命中、规则求不了值）： -/// 那时说不出这个请求算在哪条规则上，不留这一行，和鉴权没过一样。模型准入没过 -/// （[`admit`]）也从这里返回，同样不留这一行 —— 和准入放在路由之前时一样。 +/// 闸门之后（见 [`start`]）。去不了任何一家的一律是 [`Routed::Refused`]，留一行。 fn route( state: &AppState, rt: &Runtime, req: &Inbound, reading: &crate::client_api::Reading, conv: Option<&crate::affinity::Conversation>, -) -> Result { +) -> Routed { let facts = &reading.facts; let route = rt.engine.route_of(&req.client_name).to_string(); + // 规则还没做出决定：说不出这个请求算在哪条规则上，决定去向的规则空着 + let undecided = |why| { + Routed::Refused( + Choice { + route: route.clone(), + ..Default::default() + }, + why, + ) + }; + // 首次运行还没配完是正常状态，不是配置错误。这条要在规则求值之前挡， + // 因为「一个 provider 都没有」时任何路由结果都是空的，而那条错误 + // 说不清下一步。 + if rt.config.providers.is_empty() { + return undecided(GatewayError::config(msg!( + "gw.config.no_upstreams" => + "No upstream is configured yet. Add one in ThinkWatch Lite, or under `providers` in \ + config.yaml." + ))); + } let now = now_ms(); // 同一轮里沿用这一轮开头的决定：按输入大小、有没有图片分流的规则，不能让一轮 // 半路换家（见 `crate::affinity`）。输入超出了所选模型的上下文的，重新求值 @@ -784,16 +795,19 @@ fn route( let held_route = held.is_some(); let mut decision = match held { Some(d) => d, - None => match rt.engine.route(facts).map_err(|e| { - GatewayError::config(msg!("gw.route.failed", detail = e => "Routing failed: {detail}")) - })? { - tw_engine::Outcome::Route(d) => { + None => match rt.engine.route(facts) { + Err(e) => { + return undecided(GatewayError::config(msg!( + "gw.route.failed", detail = e => "Routing failed: {detail}" + ))); + } + Ok(tw_engine::Outcome::Route(d)) => { if let Some(c) = conv { state.affinity.decided(c, &rt.engine, &d, now); } d } - tw_engine::Outcome::Deny { rule, reason } => { + Ok(tw_engine::Outcome::Deny { rule, reason }) => { // **带理由的拒绝。**一个没有理由的拒绝，和一个 bug，在用户 // 眼里没有区别。 tracing::info!(%rule, "a rule denied the request"); @@ -810,7 +824,7 @@ fn route( affinity: None, stayed_on: None, }; - return Ok(Routed::Refused(choice, why)); + return Routed::Refused(choice, why); } }, }; @@ -832,7 +846,10 @@ fn route( facts, &decision, ); - admit(state, rt, req, reading, &decision, &asked)?; + // 准入没过：规则做了决定，模型却去不了（写错了、没有上游提供、密钥不让用）。照样留一行 + if let Err(why) = admit(state, rt, req, reading, &decision, &asked) { + return Routed::Refused(choice, why); + } // 每一家发出去的名字：别名对到各家自己的名称。跳过、比价看它（见 `crate::sent`） let catalog = state.catalog.load(); let allow = crate::models::key_allow(&rt.config, &req.client_name); @@ -847,7 +864,7 @@ fn route( let serving = crate::sent::serving(&rt.config, &catalog, &decision, &asked, allow); let model = decision.set.model.as_deref().unwrap_or(&facts.model); if serving.usable.is_empty() { - return Ok(Routed::Refused(choice, serving.explain(model))); + return Routed::Refused(choice, serving.explain(model)); } if !serving.skipped.is_empty() { tracing::debug!(skipped = ?serving.skipped, %model, "skipping the candidates that cannot serve this request"); @@ -874,7 +891,7 @@ fn route( if let Some(leader) = decision.candidates.first() { arranged.charge(&decision.candidates, leader); } - Ok(Routed::Go(choice, decision)) + Routed::Go(choice, decision) } /// 排好了的候选（[`arrange`]）：会话粘性留下了哪一家的理由，和 `load-balance` 这一次还没记的账。 diff --git a/crates/tw-gateway/src/server/upgrade.rs b/crates/tw-gateway/src/server/upgrade.rs index ffca81a7..2514dad4 100644 --- a/crates/tw-gateway/src/server/upgrade.rs +++ b/crates/tw-gateway/src/server/upgrade.rs @@ -79,11 +79,21 @@ pub(super) async fn ws_upgrade( at_ms: now_ms(), }) }; - let mut decision = match rt.engine.route(&facts).map_err(|e| { - GatewayError::config(msg!("gw.route.failed", detail = e => "Routing failed: {detail}")) - })? { - tw_engine::Outcome::Route(d) => d, - tw_engine::Outcome::Deny { rule, reason } => { + // 一家上游都不去的升级：开始、空尝试链的路由（阶段二拒绝的记上是哪条规则）、一条失败， + // 交回这个错误 + let refuse = |choice: Choice, denied_by: Option, why: GatewayError| { + let (id, ending) = open(&choice, "", tw_api::Billing::PerToken); + let mut routed = super::routed_nowhere(id, choice); + if let tw_api::Event::RequestRouted { denied_by: d, .. } = &mut routed { + *d = denied_by; + } + state.bus.emit(routed); + ending.failed(why.source.into(), why.detail.clone()); + why + }; + let mut decision = match rt.engine.route(&facts) { + Ok(tw_engine::Outcome::Route(d)) => d, + Ok(tw_engine::Outcome::Deny { rule, reason }) => { tracing::info!(%rule, "a rule denied the WebSocket upgrade"); let err = GatewayError::denied(msg!( "gw.route.denied", rule = rule.clone(), reason = reason => @@ -92,15 +102,20 @@ pub(super) async fn ws_upgrade( let choice = Choice { route, rule, - group: None, - rewritten_by: Vec::new(), - affinity: None, - stayed_on: None, + ..Default::default() }; - let (id, ending) = open(&choice, "", tw_api::Billing::PerToken); - state.bus.emit(super::routed_nowhere(id, choice)); - ending.failed(err.source.into(), err.detail.clone()); - return Err(err); + return Err(refuse(choice, None, err)); + } + // 规则没做出决定（没有一条命中、求不了值）：照样留一行，决定去向的规则空着 + Err(e) => { + let err = GatewayError::config(msg!( + "gw.route.failed", detail = e => "Routing failed: {detail}" + )); + let choice = Choice { + route, + ..Default::default() + }; + return Err(refuse(choice, None, err)); } }; // 候选的次序：**和 HTTP 那条路同一段**（见 `super::pipeline::arrange`）。先去掉服务不了的 @@ -137,6 +152,14 @@ pub(super) async fn ws_upgrade( local_time: state.local_time.clone(), }; let (alive, _) = state.health.filter(&decision.candidates); + // 规则做了决定、却一家都连不上时留的那一行记在哪儿 + let decided = || Choice { + route: route.clone(), + rule: decision.matched_rule.clone(), + group: decision.via_group.clone(), + rewritten_by: decision.rewritten_by.clone(), + ..Default::default() + }; // 升级请求带着模型的（Realtime）：挑头一家连得上的 —— 别名对得上、要发给它的名字密钥 // 让用，和 HTTP 那条路挑候选时一样。别的连接升级时不知道模型，连头一家活着的 let picked = if facts.model.is_empty() { @@ -164,35 +187,23 @@ pub(super) async fn ws_upgrade( unserved.get_or_insert(why); } // 阶段二的规则拒绝了：**和规则在阶段一拒绝一样留一行**，拒绝它的是哪条规则 - // 写进路由事件 + // 写进路由事件。阶段二求不了值的也留一行 crate::ws::Connect::Refused { rule, why } => { - let choice = Choice { - route, - rule: decision.matched_rule.clone(), - group: decision.via_group.clone(), - rewritten_by: decision.rewritten_by.clone(), - affinity: None, - stayed_on: None, - }; - let Some(rule) = rule else { return Err(why) }; - tracing::info!(%rule, provider = %c, "a phase-two rule denied the WebSocket upgrade"); - let (id, ending) = open(&choice, "", tw_api::Billing::PerToken); - let mut routed = super::routed_nowhere(id, choice); - if let tw_api::Event::RequestRouted { denied_by, .. } = &mut routed { - *denied_by = Some(rule); + if let Some(rule) = &rule { + tracing::info!(%rule, provider = %c, "a phase-two rule denied the WebSocket upgrade"); } - state.bus.emit(routed); - ending.failed(why.source.into(), why.detail.clone()); - return Err(why); + return Err(refuse(decided(), rule, why)); } } } // 一家都连不上。每一家要发的名字密钥都不让用的，说是密钥的事（和 HTTP 那条路的准入 - // 一样）；有让用的，说它为什么连不上（别名对不上）。**不留这一行**，和 HTTP 那条路 + // 一样）；有让用的，说它为什么连不上（别名对不上）。**照样留一行**，和 HTTP 那条路 // 准入没过一样 match (picked, unserved, barred) { (Some(p), _, _) => Some(p), - (None, Some(why), _) | (None, None, Some(why)) => return Err(why), + (None, Some(why), _) | (None, None, Some(why)) => { + return Err(refuse(decided(), None, why)); + } (None, None, None) => None, } }; diff --git a/crates/tw-gateway/src/ws.rs b/crates/tw-gateway/src/ws.rs index cec8401c..091b7c25 100644 --- a/crates/tw-gateway/src/ws.rs +++ b/crates/tw-gateway/src/ws.rs @@ -282,26 +282,34 @@ struct Outgoing { rewritten_by: Vec, } -/// 一帧为什么没发出去（[`Naming::frame`]）。 +/// 一帧为什么没发出去（[`Naming::frame`]）。**这一帧照样留一行**（见 [`turn::not_sent`]）， +/// 和 HTTP 那条路一家上游都不去的请求一样：规则拒绝了它、密钥不让用要发的模型、这一家服务 +/// 不了要的别名、规则没做出决定或者求不了值 struct NotSent { why: GatewayError, - /// 规则拒绝了它：**这一帧照样留一行**，和 HTTP 那条路被规则拒绝的请求一样（见 - /// [`turn::denied`]）。别的 —— 密钥不让用要发的模型、这一家服务不了要的别名、规则求不了 - /// 值 —— 不留，和 HTTP 那条路准入没过一样 - denied: Option, + /// 这一行记在哪条规则上 + by: Decider, } impl NotSent { + /// 去向照升级时的决定，只是这一帧发不出去 fn plain(why: GatewayError) -> Box { - Box::new(Self { why, denied: None }) + Box::new(Self { + why, + by: Decider::Upgrade, + }) } } -/// 拒绝了一帧的那条规则，在哪个阶段。 -pub(crate) enum Denied { - /// 阶段一：它就是决定这一帧去向的那条 +/// 没发出去的一帧记在哪条规则上。 +pub(crate) enum Decider { + /// 升级时决定去向的那条：密钥不让用要发的模型、这一家服务不了要的别名、阶段二求不了值 + Upgrade, + /// 这一帧没有规则做出决定（没有一条命中、求不了值）：空着 + Undecided, + /// 阶段一拒绝了它：它就是决定这一帧去向的那条 PhaseOne(String), - /// 阶段二：记在路由事件的 `denied_by` 上 + /// 阶段二拒绝了它：决定去向的还是升级时那条，拒绝它的记在路由事件的 `denied_by` 上 PhaseTwo(String), } @@ -447,7 +455,7 @@ impl Naming { tracing::info!(%rule, provider = %p.name, "a phase-two rule denied a WebSocket request"); return Err(Box::new(NotSent { why: denied(rule.clone(), reason), - denied: Some(Denied::PhaseTwo(rule)), + by: Decider::PhaseTwo(rule), })); } Err(e) => return Err(NotSent::plain(rule_failed(e))), @@ -494,12 +502,15 @@ impl Naming { tracing::info!(%rule, "a rule denied a WebSocket request"); Err(Box::new(NotSent { why: denied(rule.clone(), reason), - denied: Some(Denied::PhaseOne(rule)), + by: Decider::PhaseOne(rule), })) } - Err(e) => Err(NotSent::plain(GatewayError::config(msg!( - "gw.route.failed", detail = e => "Routing failed: {detail}" - )))), + Err(e) => Err(Box::new(NotSent { + why: GatewayError::config(msg!( + "gw.route.failed", detail = e => "Routing failed: {detail}" + )), + by: Decider::Undecided, + })), } } @@ -1339,7 +1350,7 @@ fn create_frame(text: &str) -> Option { /// 一轮的开头：一帧 `response.create`（`raw` 是客户端的原话，`frame` 是它解出来的样子）。 /// /// 位置和 HTTP 那条路一样：内容过滤先下结论（不报：结论挂在这一轮的号上报），定发给这一家的 -/// 模型名和参数改写（[`Naming`]：规则拒绝了的照样留一行），然后交去准入（[`turn::admit`]）。 +/// 模型名和参数改写（[`Naming`]：发不出去的照样留一行），然后交去准入（[`turn::admit`]）。 /// 插件的请求钩子、脱敏、发出在过了准入之后（[`admitted`]）。 async fn begin_turn( state: &AppState, @@ -1392,23 +1403,21 @@ async fn begin_turn( let catalog = state.catalog.load(); let out = match naming.frame(&catalog, &reading.facts) { Ok(out) => out, - // 只是这一帧不发：替它回一个 `response.failed`，连接照常 + // 只是这一帧不发：照样留一行，替它回一个 `response.failed`，连接照常 Err(no) => { tracing::info!(provider = %p.provider, why = %no.why.detail.text, "a WebSocket request was not sent"); - if let Some(by) = no.denied { - turn::denied( - state, - &line, - by, - &no.why, - requested, - fingerprint.as_deref(), - input_estimate, - arrived, - at_ms, - ); - } + turn::not_sent( + state, + &line, + no.by, + &no.why, + requested, + fingerprint.as_deref(), + input_estimate, + arrived, + at_ms, + ); return reply_failed(c_tx, no.why).await; } }; diff --git a/crates/tw-gateway/src/ws/turn.rs b/crates/tw-gateway/src/ws/turn.rs index 3345e6be..65542d8b 100644 --- a/crates/tw-gateway/src/ws/turn.rs +++ b/crates/tw-gateway/src/ws/turn.rs @@ -10,8 +10,9 @@ //! //! **连接本身不留行**：一条连接跑好几轮、中间可以闲着很久，流量里该看的是每一轮。会话照 //! HTTP 那条路按每一帧认（Codex 每段对话带着 `prompt_cache_key`），同一段对话的几轮归到同一 -//! 次会话里。没有哪一轮可挂的 —— 升级时就被规则拒绝的、连不上上游的 —— 那条连接自己留一行 -//! （见 `server::upgrade`）。 +//! 次会话里。没有哪一轮可挂的 —— 升级时就被规则拒绝的、规则没做出决定的、连不上上游的 —— +//! 那条连接自己留一行（见 `server::upgrade`）。一轮没发出去的（规则拒绝了、密钥不让用要发的 +//! 模型、这一家服务不了要的别名）也留一行（见 [`not_sent`]）。 //! //! **上限和并发按轮算，和 HTTP 那条路的一个请求同一套**（[`admit`]）：天、周、月用满了就拒， //! 密钥的并发上限等前面的结束，分钟、小时等得到就等；然后等这一家能发 —— 这条连接只连着这一家， @@ -851,14 +852,15 @@ fn refused(a: &Admit, choice: &Choice, why: GatewayError) -> NotAdmitted { NotAdmitted::Failed(why) } -/// 规则拒绝了这一帧：**和 HTTP 那条路被规则拒绝的请求一样留一行**（开始、空尝试链的路由、 -/// 一条 `denied` 的失败）。阶段二拒绝的，拒绝它的那条规则记在路由事件的 `denied_by` 上；阶段 -/// 一拒绝的，它就是决定去向的那条 +/// 这一帧没发出去（规则拒绝了、密钥不让用要发的模型、这一家服务不了要的别名、规则没做出 +/// 决定）：**和 HTTP 那条路一家上游都不去的请求一样留一行**（开始、空尝试链的路由、一条失败， +/// 错误照告诉客户端的那一句）。阶段二拒绝的，拒绝它的那条规则记在路由事件的 `denied_by` 上； +/// 阶段一拒绝的，它就是决定去向的那条；规则没做出决定的，决定去向的规则空着 #[allow(clippy::too_many_arguments)] -pub(crate) fn denied( +pub(crate) fn not_sent( state: &AppState, line: &Line, - by: super::Denied, + by: super::Decider, why: &GatewayError, requested: String, fingerprint: Option<&str>, @@ -868,11 +870,19 @@ pub(crate) fn denied( ) { let mut choice = line.choice.clone(); let denied_by = match by { - super::Denied::PhaseOne(rule) => { + super::Decider::Upgrade => None, + super::Decider::Undecided => { + choice = Choice { + route: choice.route, + ..Default::default() + }; + None + } + super::Decider::PhaseOne(rule) => { choice.rule = rule; None } - super::Denied::PhaseTwo(rule) => Some(rule), + super::Decider::PhaseTwo(rule) => Some(rule), }; let (id, ending) = line.opener.open(Opening { choice: &choice, diff --git a/crates/tw-gateway/tests/alias_ws.rs b/crates/tw-gateway/tests/alias_ws.rs index 8b9d5241..7a476802 100644 --- a/crates/tw-gateway/tests/alias_ws.rs +++ b/crates/tw-gateway/tests/alias_ws.rs @@ -179,6 +179,51 @@ fn answered(frames: &[Value]) -> Vec { } /// 这一次没发出去：替它回的 `response.failed` 里那句话 +/// 没发出去的那个请求留下的一行：开始（不发往哪一家、模型是客户端写的 `model`）、空尝试链 +/// 的路由、一条来源为 `request` 的失败。交回决定去向的规则和失败的错误 +async fn nowhere(events: &mut Receiver, model: &str) -> (String, tw_api::Msg) { + let mut rule_of = None; + loop { + let ev = tokio::time::timeout(Duration::from_secs(5), events.recv()) + .await + .expect("no ending within 5 s") + .expect("the event stream broke"); + match ev { + Event::RequestStarted { + provider, + model: m, + rule, + .. + } => { + assert!(provider.is_empty(), "{provider}"); + assert_eq!(m, model); + rule_of = Some(rule); + } + Event::RequestRouted { + rule, + attempts, + denied_by, + .. + } => { + assert!(attempts.is_empty() && denied_by.is_none(), "{attempts:?}"); + assert_eq!(Some(&rule), rule_of.as_ref()); + } + Event::RequestFailed { + model: m, + source, + message, + .. + } => { + assert_eq!(m, model); + assert_eq!(source.slug(), "request"); + return (rule_of.expect("started before it failed"), message); + } + Event::RequestFinished { .. } | Event::RequestCancelled { .. } => panic!("{ev:?}"), + _ => {} + } + } +} + fn refusal(frames: &[Value]) -> String { let last = frames.last().expect("no frame at all"); assert_eq!(last["type"], "response.failed", "{frames:?}"); @@ -373,16 +418,13 @@ async fn an_alias_the_upstream_cannot_serve_fails_that_request_and_the_connectio (claude-sonnet-5, us.anthropic.claude-sonnet-5-v1:0), so the request was not sent." ); assert!(seen.lock().unwrap().is_empty()); - // 这一家服务不了要的别名：**不留这一行**，和 HTTP 那条路准入没过一样 - while let Ok(Ok(ev)) = tokio::time::timeout(Duration::from_millis(200), events.recv()).await { - assert!( - !matches!( - ev, - Event::RequestStarted { .. } | Event::RequestRouted { .. } - ), - "{ev:?}" - ); - } + // 这一家服务不了要的别名：**这一轮照样留一行**，和 HTTP 那条路准入没过一样 —— 记在决定 + // 去向的规则上，模型是客户端写的，错误照回给客户端的那一句 + let (rule, why) = nowhere(&mut events, "gpt-5.1-codex").await; + assert_eq!(rule, "兜底"); + assert_eq!(why.code, "gw.ws.alias_unserved_rewritten"); + assert_eq!(why.args["from"], "gpt-5.1-codex"); + assert_eq!(format!("[ThinkWatch] {}", why.text), refusal(&frames)); } /// 插件把模型名换成别名：发给这一家的是它自己的名称，回答里写回客户端要的。插件的 @@ -454,11 +496,12 @@ fn scoped(p: Provider, aliases: &str, rules: &str, allow: &[&str]) -> Config { } /// 密钥的 `allow` 每一帧都看：范围外的模型这一帧不发，替它回一个说清楚的 `response.failed`， -/// 连接照常 —— 下一帧要的是范围里的就照发。和 HTTP 那条路同一句话 +/// 连接照常 —— 下一帧要的是范围里的就照发。和 HTTP 那条路同一句话，这一轮也和 HTTP 那条路 +/// 一样留一行 #[tokio::test] async fn a_frame_asking_for_a_model_outside_the_keys_allow_is_refused_and_the_connection_stays() { let (up, seen) = upstream().await; - let (gw, _) = gateway(scoped(provider(up, &[]), "", "", &["gpt-5*"]), vec![]).await; + let (gw, mut events) = gateway(scoped(provider(up, &[]), "", "", &["gpt-5*"]), vec![]).await; let mut c = connect(gw).await; c.send(create("other-model")).await.unwrap(); @@ -469,6 +512,10 @@ async fn a_frame_asking_for_a_model_outside_the_keys_allow_is_refused_and_the_co models that are available." ); assert!(seen.lock().unwrap().is_empty()); + let (rule, why) = nowhere(&mut events, "other-model").await; + assert_eq!(rule, tw_engine::CATCH_ALL_RULE); + assert_eq!(why.code, "gw.model.not_allowed"); + assert_eq!(why.args["key"], "codex"); c.send(create("gpt-5.1-codex")).await.unwrap(); let frames = one_answer(&mut c).await; @@ -795,7 +842,7 @@ async fn a_realtime_alias_in_the_query_connects_with_the_upstreams_own_name() { async fn a_realtime_model_outside_the_keys_allow_refuses_the_upgrade() { let (up, queries) = realtime_upstream().await; let aliases = format!("voice: [gpt-realtime-2, gpt-realtime]\n{SONNET}"); - let (gw, _) = gateway( + let (gw, mut events) = gateway( scoped( provider(up, &["gpt-realtime", "other-model"]), &aliases, @@ -816,12 +863,17 @@ async fn a_realtime_model_outside_the_keys_allow_refuses_the_upgrade() { "{body}" ); assert!(queries.lock().unwrap().is_empty()); + // 被拒的升级**照样留一行**，和 HTTP 那条路准入没过一样 + let (rule, why) = nowhere(&mut events, "other-model").await; + assert_eq!(rule, tw_engine::CATCH_ALL_RULE); + assert_eq!(why.code, "gw.model.not_allowed"); + assert_eq!(why.args["model"], "other-model"); let mut c = realtime(gw, "model=voice").await.unwrap(); c.next().await; assert_eq!(queries.lock().unwrap().as_slice(), ["model=gpt-realtime"]); - let (gw, _) = gateway( + let (gw, mut events) = gateway( config(provider(up, &["gpt-realtime"]), &aliases, ""), vec![], ) @@ -833,4 +885,6 @@ async fn a_realtime_model_outside_the_keys_allow_refuses_the_upgrade() { "{body}" ); assert_eq!(queries.lock().unwrap().len(), 1); + let (_, why) = nowhere(&mut events, "claude-sonnet").await; + assert_eq!(why.code, "gw.ws.alias_unserved"); } diff --git a/crates/tw-gateway/tests/models.rs b/crates/tw-gateway/tests/models.rs index 6bbabd29..fea31a02 100644 --- a/crates/tw-gateway/tests/models.rs +++ b/crates/tw-gateway/tests/models.rs @@ -283,6 +283,7 @@ async fn when_no_candidate_serves_the_model_the_error_names_each_one_and_why() { }])]; let state = tw_gateway::AppState::new(cfg).unwrap(); tw_gateway::models::refresh_all(&state).await; + let mut events = state.bus.subscribe(); let gw = serve(state).await; let (status, body) = ask(gw, "official-only").await; @@ -292,12 +293,71 @@ async fn when_no_candidate_serves_the_model_the_error_names_each_one_and_why() { msg.contains("official-only") && msg.contains("relay does not offer this model"), "{msg}" ); + let failed = refused(&mut events, "official-only").await; + assert_eq!(failed.code, "gw.model.no_upstream_available"); // 没有任何上游提供的模型：准入就拦下，而不是说客户端不允许 let (status, body) = ask(gw, "nobody-has-this").await; assert_eq!(status, 400, "{body}"); let msg = body["error"]["message"].as_str().unwrap(); - assert!(msg.contains("No upstream serves model"), "{msg}"); + assert_eq!( + msg, + "[ThinkWatch] No upstream serves model nobody-has-this. GET /v1/models lists the models \ + that are available." + ); + // 写错了、谁都不提供的模型名**照样留一行**：流量里看得见才排查得了。记在决定去向的那条 + // 规则上，模型是客户端写的那个，错误照告诉客户端的那一句 + let failed = refused(&mut events, "nobody-has-this").await; + assert_eq!(failed.code, "gw.model.no_upstream"); + assert_eq!(failed.args["model"], "nobody-has-this"); + assert_eq!(format!("[ThinkWatch] {}", failed.text), msg); +} + +/// 一家上游都没去的那个请求的几条事件：开始（不发往哪一家）、空尝试链的路由、一条来源为 +/// `request` 的失败，都记在规则「都给 relay」上、模型是客户端写的 `model`。交回失败的错误 +async fn refused( + events: &mut tokio::sync::broadcast::Receiver, + model: &str, +) -> tw_api::Msg { + use tw_api::Event; + let mut started = false; + let mut routed = false; + loop { + let ev = tokio::time::timeout(Duration::from_secs(5), events.recv()) + .await + .expect("5 秒内没等到结局") + .expect("事件流断了"); + match ev { + Event::RequestStarted { + provider, + model: m, + rule, + .. + } => { + assert!(provider.is_empty() && m == model && rule == "都给 relay"); + started = true; + } + Event::RequestRouted { rule, attempts, .. } => { + assert!(attempts.is_empty() && rule == "都给 relay", "{attempts:?}"); + routed = true; + } + Event::RequestFailed { + model: m, + source, + message, + .. + } => { + assert!(started && routed, "开始、路由都要在失败之前"); + assert_eq!(m, model); + assert_eq!(source.slug(), "request"); + return message; + } + Event::RequestFinished { .. } | Event::RequestCancelled { .. } => { + panic!("{ev:?}") + } + _ => {} + } + } } // ─────────────────────────────────────────────────────────── 手动添加的模型 diff --git a/crates/tw-gateway/tests/routing_facts.rs b/crates/tw-gateway/tests/routing_facts.rs index f300dc1a..475671ca 100644 --- a/crates/tw-gateway/tests/routing_facts.rs +++ b/crates/tw-gateway/tests/routing_facts.rs @@ -3,8 +3,9 @@ //! - 会话在开始时就定了，开始事件带着它（以前只有指纹，落库时才算出会话）； //! - 走的路由、决定去向的规则、策略组、改写了它的规则，开始时就有，路由事件 //! 给终稿（加上第二阶段的改写和拒绝）； -//! - 规则做了决定、请求却一家上游都没去的（被规则拒绝、选中的上游都服务不了）， -//! 照样有开始、路由和一条失败 —— 以前这些请求在事件里根本不存在。 +//! - 认出了密钥、请求却一家上游都没去的（被规则拒绝、选中的上游都服务不了、密钥不让用 +//! 那个模型、没有一条规则命中、还没配上游），照样有开始、路由和一条失败 —— 以前这些 +//! 请求在事件里根本不存在。 use std::net::SocketAddr; use std::time::Duration; @@ -327,6 +328,165 @@ async fn a_request_whose_chosen_upstreams_cannot_serve_it_is_recorded_under_its_ ); } +/// 发一个请求，交回状态码和响应体：被拒的请求留一行，**客户端收到的照旧** +async fn answer(gw: SocketAddr, body: &str) -> (u16, serde_json::Value) { + let r = reqwest::Client::builder() + .no_proxy() + .build() + .unwrap() + .post(format!("http://{gw}/v1/messages")) + .header("x-api-key", "tw-k") + .body(body.to_string()) + .send() + .await + .unwrap(); + let status = r.status().as_u16(); + (status, r.json().await.unwrap()) +} + +/// 一家上游都没去的那一行：开始（不发往哪一家，模型是客户端写的）、空尝试链的路由、一条 +/// 失败。交回三条事件里的路由、决定去向的规则，和失败的来源、错误 +fn nowhere(evs: &[Event], model: &str) -> (String, String, String, tw_api::Msg) { + assert_eq!(evs.len(), 3, "{evs:?}"); + let Event::RequestStarted { + provider, + model: started_model, + route, + rule, + .. + } = &evs[0] + else { + panic!("{evs:?}") + }; + assert!(provider.is_empty(), "{evs:?}"); + assert_eq!(started_model, model, "{evs:?}"); + let Event::RequestRouted { + route: routed_route, + rule: routed_rule, + attempts, + denied_by: None, + .. + } = &evs[1] + else { + panic!("{evs:?}") + }; + assert!(attempts.is_empty(), "{evs:?}"); + assert_eq!((routed_route, routed_rule), (route, rule), "{evs:?}"); + let Event::RequestFailed { + model: failed_model, + source, + message, + .. + } = &evs[2] + else { + panic!("{evs:?}") + }; + assert_eq!(failed_model, model, "{evs:?}"); + ( + route.clone(), + rule.clone(), + source.slug().to_string(), + message.clone(), + ) +} + +/// 密钥不让用的模型：准入没过。**照样留一行**，记在决定去向的那条规则上，模型是客户端写的 +/// 那个，错误照告诉客户端的那一句 —— 以前这个请求在流量里根本看不见，没法排查。客户端收到 +/// 的还是那个 400 +#[tokio::test] +async fn a_model_the_key_may_not_use_is_recorded_under_its_rule() { + let up = upstream().await; + let mut config = cfg( + vec![provider("up", up)], + vec![rule("走池子", "{}", Some("池"), None, None)], + ); + config.clients[0].allow = Some(vec!["claude-sonnet-*".into()]); + let (gw, mut events) = serve(config).await; + + let (status, body) = answer(gw, &chat("claude-opus-4-1", "hi")).await; + assert_eq!(status, 400, "{body}"); + assert_eq!( + body["error"]["message"], + "[ThinkWatch] Gateway key `我` may not use model claude-opus-4-1. GET /v1/models lists \ + the models that are available.", + "{body}" + ); + + let evs = one_request(&mut events).await; + let (route, rule, source, why) = nowhere(&evs, "claude-opus-4-1"); + assert_eq!((route.as_str(), rule.as_str()), ("工作", "走池子")); + assert_eq!(source, "request"); + assert_eq!(why.code, "gw.model.not_allowed"); + assert_eq!(why.args["model"], "claude-opus-4-1"); + assert_eq!(why.args["key"], "我"); + + // 让用的照常 + assert_eq!(ask(gw, &chat("claude-sonnet-4-5", "hi")).await, 200); +} + +/// 没有一条规则命中：规则还没做出决定。**也留一行**，记在密钥走的那条路由上，决定去向的规则 +/// 空着 +#[tokio::test] +async fn a_request_no_rule_matches_is_recorded_on_its_route_with_no_rule() { + let up = upstream().await; + let (gw, mut events) = serve(cfg( + vec![provider("up", up)], + vec![rule( + "只给 opus", + "{ model: claude-opus-* }", + Some("池"), + None, + None, + )], + )) + .await; + + let (status, body) = answer(gw, &chat("claude-sonnet-4-5", "hi")).await; + assert_eq!(status, 500, "{body}"); + let text = body["error"]["message"].as_str().unwrap(); + assert!(text.starts_with("[ThinkWatch] Routing failed: "), "{body}"); + + let evs = one_request(&mut events).await; + let (route, rule, source, why) = nowhere(&evs, "claude-sonnet-4-5"); + assert_eq!((route.as_str(), rule.as_str()), ("工作", "")); + assert_eq!(source, "config"); + assert_eq!(why.code, "gw.route.failed"); + assert_eq!(format!("[ThinkWatch] {}", why.text), text); +} + +/// 还没配上游：**也留一行**，记在密钥走的那条路由上，决定去向的规则空着。客户端收到的照旧 +/// 是「还没配上游」那一句 +#[tokio::test] +async fn a_request_with_no_upstream_configured_is_recorded() { + let config = Config { + clients: vec![Client { + name: "我".into(), + key: "tw-k".into(), + ..Default::default() + }], + ..Default::default() + }; + let (gw, mut events) = serve(config).await; + + let (status, body) = answer(gw, &chat("claude-sonnet-4-5", "hi")).await; + assert_eq!(status, 500, "{body}"); + assert_eq!( + body["error"]["message"], + "[ThinkWatch] No upstream is configured yet. Add one in ThinkWatch Lite, or under \ + `providers` in config.yaml.", + "{body}" + ); + + let evs = one_request(&mut events).await; + let (route, rule, source, why) = nowhere(&evs, "claude-sonnet-4-5"); + assert_eq!( + (route.as_str(), rule.as_str()), + (tw_engine::DEFAULT_ROUTE, "") + ); + assert_eq!(source, "config"); + assert_eq!(why.code, "gw.config.no_upstreams"); +} + /// 选定上游之后才判断的规则拒绝了它（第二阶段）：**路由事件照样发**，说是哪条规则、 /// 在哪一家停下的 —— 被拒的那一跳在尝试链上，没有发出去 #[tokio::test] diff --git a/crates/tw-store/src/db.rs b/crates/tw-store/src/db.rs index b059be96..e70539b6 100644 --- a/crates/tw-store/src/db.rs +++ b/crates/tw-store/src/db.rs @@ -1678,9 +1678,13 @@ impl Db { }; let (total, rules) = routes.get(route); total.add(at_ms, failed); - let decider = rules.get(rule); - decider.add(at_ms, failed); - decider.decided += 1; + // 规则还没做出决定就被拒的（没有一条命中、求不了值、还没配上游）：决定去向的规则 + // 是空的，只算进这条路由 + if !rule.is_empty() { + let decider = rules.get(rule); + decider.add(at_ms, failed); + decider.decided += 1; + } let mut counted = vec![rule]; for name in rewritten_by.iter().map(String::as_str).chain(denied_by) { if !counted.contains(&name) { @@ -3881,7 +3885,8 @@ mod route_stats_tests { /// 每条路由走了多少、每条规则命中了多少、多少失败了、最后一次是什么时候。 /// /// **一个请求算在它命中的每条规则上，每条只算一次**：决定去向的、附加了改写 - /// 的、选定上游之后拒绝了它的。被规则拒绝的也算命中 —— 它们是失败。 + /// 的、选定上游之后拒绝了它的。被规则拒绝的也算命中 —— 它们是失败。规则还没做出决定 + /// 就被拒的只算进路由。 #[test] fn each_route_and_rule_counts_the_requests_it_matched() { let db = Db::in_memory().unwrap(); @@ -3902,6 +3907,10 @@ mod route_stats_tests { let mut late = routed(5, 500, "默认", "兜底", &[], Some("中转不收密钥")); late.error = Some(upstream_failed("denied")); db.insert(&late).unwrap(); + // 规则还没做出决定就被拒了（没有一条命中）：算进这条路由，不算在哪条规则上 + let mut undecided = routed(6, 450, "默认", "", &[], None); + undecided.error = Some(upstream_failed("no rule")); + db.insert(&undecided).unwrap(); let got = db.route_stats(0, 1_000).unwrap().routes; assert_eq!( @@ -3936,8 +3945,12 @@ mod route_stats_tests { ); let default = &got[1]; - assert_eq!((default.requests, default.failed), (1, 1)); + assert_eq!((default.requests, default.failed), (2, 2)); assert_eq!(hits(default, "兜底").decided, 1); + assert!( + default.rules.iter().all(|h| !h.rule.is_empty()), + "{default:?}" + ); let late_deny = hits(default, "中转不收密钥"); assert_eq!( (late_deny.decided, late_deny.requests, late_deny.failed),