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),