Retry the last upstream after a temporary failure before any content

README.md

@@ -62,7 +62,9 @@ Documentation: [configuration reference](docs/config.md) ·
   images, thinking and more, and send requests to an upstream or a group (in
   order, manual, rotation, lowest latency, lowest cost). Until the first byte
   of the response reaches the client, a failing upstream is replaced by the
-  next candidate and set aside for a time that depends on the reason it gives.
+  next candidate and set aside for a time that depends on the reason it gives;
+  when the last one is only overloaded or briefly unreachable, it is retried
+  after a pause.
 - **Many kinds of upstream.** Provider API keys, any compatible endpoint, relays
   such as OpenRouter, local models, Amazon Bedrock, and ChatGPT or Z.ai
   accounts. Health checks and warm-ups are answered locally by default, at no cost.

README.zh-CN.md

@@ -28,7 +28,7 @@ ThinkWatch Core 是 ThinkWatch 的网关引擎，由一组 Rust crate 及其构
 - **切断恶意工具调用**。中转站可以改写回答，塞入让客户端执行的工具调用。回答中的工具调用若是下载即执行、外发环境变量或凭据文件、读取私钥、读写 ThinkWatch 自己的配置、写入开机启动项或定时任务，工具调用审查可以在客户端收到完整调用之前切断回答。
 - **清除隐藏指令**。屏幕上看不见的字符可以夹带模型会读取的指令，内容过滤可以在请求发出前把它们从用户消息和工具结果中删除，也可以拒绝要求模型忽略自身指令的请求。出站脱敏、工具调用审查和内容过滤出厂均为观察档，只记录检出的内容，不改变任何请求。
 - **每个请求都可追溯**。每个请求连同决定其去向的规则、每次尝试、格式转换、用量、费用及价格来源、首 token 时间和生成速度一并保存。试算可以在不发出请求的情况下说明请求会被送往何处；已保存的请求可以对另一个上游重放，以便对比。
-- **路由与故障转移**。规则可按模型、密钥、格式、请求大小、工具、图片、思考等条件匹配，把请求交给一个上游或策略组（按顺序、手动指定、轮流、最低延迟、最低价格）。响应的首字节到达客户端之前，失败的上游由下一个候选替换，并按其给出的失败原因暂停相应的时间。
+- **路由与故障转移**。规则可按模型、密钥、格式、请求大小、工具、图片、思考等条件匹配，把请求交给一个上游或策略组（按顺序、手动指定、轮流、最低延迟、最低价格）。响应的首字节到达客户端之前，失败的上游由下一个候选替换，并按其给出的失败原因暂停相应的时间；最后一个候选只是过载或暂时连不上的，等待片刻后重试。
 - **多种上游**。服务商的 API 密钥、任意兼容接口、OpenRouter 等中转站、本地模型、Amazon Bedrock，以及 ChatGPT 和 Z.ai 账号。连通性检查和预热请求默认在本地应答，不产生费用。
 - **如实计费**。用量（含缓存读写）按每日刷新的公开价目表或自定义价目表计价。估算的费用会标明，无法计价的用量单独统计，不计作零。
 - **单一配置，即改即生效**。全部设置保存在 `config.yaml` 中；无论改动来自编辑器、命令行还是控制面，通过校验后一秒内生效，最近五十个版本均可恢复。

crates/tw-api/msg-codes.txt

@@ -450,6 +450,7 @@ gw.upstream.sealed_elsewhere
 gw.upstream.sign_failed
 gw.upstream.status
 gw.upstream.status_message
+gw.upstream.stream_ended_early
 gw.upstream.stream_error
 gw.upstream.stream_exception
 gw.upstream.stream_opening_error

crates/tw-api/src/lib.rs

@@ -1066,7 +1066,15 @@ pub const MSG_CODES: &str = include_str!("../msg-codes.txt");
 /// 父进程、开交接端点、等接替者；[`Status`] 多了 `handover`（[`HandoverState`]）和 `draining`
 /// （排空中的旧 core 手上还有几个请求，只有接管了它的新 core 报，平时是 0）。
 /// 新码：`control.handover.*`
-pub const CONTROL_API_VERSION: u32 = 52;
+///
+/// **53 起最后一家暂时出错时退避重试**：候选都试过了、最后那一跳在第一段内容之前过载、5xx、
+/// 连不上、流断了，请求在等待期限（`wait_secs`，等上游和重试共用）之内退避一阵，把暂时出错的
+/// 那几家再试一轮。新事件 [`Event::RequestRetrying`]（开始退避：再试哪几家、第几次、什么时候
+/// 发出去），退避结束发 [`Event::RequestDequeued`]，两条都在 `InFlightRequest::events` 里。
+/// [`AttemptView`] 多了 `retry`（这一跳是第几次重试）和 `backoff_ms`（发出之前退避了多久）。
+/// 新码：`gw.upstream.stream_ended_early`（流只发了开头的例行事件就收了尾，算在
+/// [`FAILED_AFTER_SENDING`] 里）
+pub const CONTROL_API_VERSION: u32 = 53;
 
 /// `/live` 的速度看最近多久（见 [`LiveView::tokens_per_sec`]），毫秒
 pub const LIVE_RATE_WINDOW_MS: u64 = 3_000;
@@ -1423,10 +1431,28 @@ pub enum Event {
     },
     /// 这个请求不排队了：等到了一家、要发出去了，被手动中止了，或者等不到了。和它之前的那条
     /// [`Event::RequestQueued`] 成对；路由事件要等整条尝试链走完才发，界面靠这一条知道它不在
-    /// 排队了。也在 [`InFlightRequest::events`] 里。
+    /// 排队了。也在 [`InFlightRequest::events`] 里。退避重试结束时也发，和之前的那条
+    /// [`Event::RequestRetrying`] 成对。
     ///
     /// 客户端在排队时走了的不发：紧跟着的结局（取消）就是它的结束
     RequestDequeued { id: u64, at_ms: u64 },
+    /// 这个请求在退避，之后重试：候选都试过了，最后那一跳在第一段内容之前出了暂时的错（过载、
+    /// 5xx、连不上、流断了），暂时出错的那几家在等待期限之内再试一轮。
+    ///
+    /// **一次退避发一条**，退避结束发 [`Event::RequestDequeued`]（重试发出去了、被手动中止了、
+    /// 等不到了）。在 [`InFlightRequest::events`] 里
+    RequestRetrying {
+        id: u64,
+        /// 要再试哪几家，按候选的顺序
+        upstreams: Vec<String>,
+        /// 接下来是第几次重试，从 1 数
+        attempt: u32,
+        /// 下一次什么时候发出去（Unix 毫秒）
+        retry_at_ms: u64,
+        /// 这个请求的等待期限（Unix 毫秒），同 [`Event::RequestQueued`] 的 `until_ms`
+        until_ms: u64,
+        at_ms: u64,
+    },
     /// 路由决定完了，尝试链也走完了。
     ///
     /// **单独一个事件，因为成功和失败两条路都要发它。**挂在
@@ -1975,6 +2001,13 @@ pub struct AttemptView {
     /// 不算进这家的用量上限（[`Self::counts_for_upstream`]）。别的结果没有
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub pause_ms: Option<u64>,
+    /// 这一跳是第几次重试，从 1 数（见 [`Event::RequestRetrying`]）：一轮重试里的每一跳都带着。
+    /// 第一轮的没有
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub retry: Option<u32>,
+    /// 这一跳发出去之前退避了多少毫秒：一轮重试的头一跳带着。不算在 `ms` 里。别的没有
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub backoff_ms: Option<u64>,
     /// 这一跳走的代理：这家上游配置的代理名（跟随系统的是 `system`），直连是 None。没发出去
     /// 的一跳（满着、被拒、转换不了……）是它要走的那个 —— 连不上的那一跳，问题可能就出在
     /// 这个代理上。WebSocket 只连直连的上游，一律是 None
@@ -1993,6 +2026,7 @@ pub const FAILED_AFTER_SENDING: &[&str] = &[
     "gw.upstream.forward_failed",
     "gw.upstream.body_broken",
     "gw.upstream.stream_opening_error",
+    "gw.upstream.stream_ended_early",
 ];
 
 /// 手动中止的请求报的码（[`FailureSource::Aborted`] 的 `message`，尝试链上 `aborted` 那一跳的
@@ -2215,6 +2249,7 @@ impl Event {
             | Event::ThrottleChanged { id, .. }
             | Event::RequestQueued { id, .. }
             | Event::RequestDequeued { id, .. }
+            | Event::RequestRetrying { id, .. }
             | Event::ModelsChanged { id, .. }
             | Event::ProxyChanged { id, .. }
             | Event::AuthChanged { id, .. }
@@ -6892,6 +6927,8 @@ mod tests {
             queued_ms: None,
             skipped: None,
             pause_ms: None,
+            retry: None,
+            backoff_ms: None,
             proxy: None,
         }
     }

crates/tw-config/src/failover.rs

@@ -45,8 +45,9 @@ pub struct Failover {
     pub idle_timeout_secs: u64,
     /// 一个请求最多等多少秒，**整个请求合起来算**：准入时等密钥的分钟、小时上限空出名额，
     /// 之后等上游 —— 并发数满了（`providers[].max_concurrent`）、被限流暂停着、分钟或小时的
-    /// 用量上限满了 —— 共用这一段。留在那一家的对话等它，候选都等不上时等先能发的那一家；
-    /// 等不到的换下一家，或者回 429。0 是不等
+    /// 用量上限满了 —— 和最后一家在内容之前暂时出错（过载、5xx、连不上）之后退避重试，共用这
+    /// 一段。留在那一家的对话等它，候选都等不上时等先能发的那一家；等不到的换下一家，或者回
+    /// 429；重试到时还没成的交出最后那个错误。0 是不等，也不重试
     #[serde(default = "d_wait_secs")]
     pub wait_secs: u64,
 }

crates/tw-config/tests/manual/schema.rs

@@ -1579,8 +1579,8 @@ pub fn sections() -> Vec<Section> {
                     Kind::Int,
                     Def::Is("120"),
                     t(
-                        "Seconds a request waits in all, counted once the key's own `max_concurrent` lets it in: for a key's `minute` or `hour` limit to free up, and for an upstream that cannot take it yet — at its `max_concurrent`, paused after answering 429, or at one of its `minute` or `hour` usage limits. A key limit that does not free up in time refuses the request. An upstream that cannot take it is skipped for the next one, and when none is left the request waits for the first to become available; a conversation that stays on an upstream waits for it. When nothing became available in time, the client gets 429 with `Retry-After`. `0`: never wait. From 0 to 600.",
-                        "一个请求合计最多等的秒数，从过了密钥自己的 `max_concurrent` 时算起：等密钥的 `minute`、`hour` 上限空出名额，和等暂时不能发的上游 —— 并发数满了（`max_concurrent`）、回了 429 正在暂停、到了这家 `minute` 或 `hour` 的用量上限 —— 都算在里面。密钥的上限到时空不出来就拒绝。上游暂时不能发的先换下一家，没有下一家了就等最先恢复的那一家；留在某一家的对话等这一家。到时都没有恢复，客户端收到带 `Retry-After` 的 429。`0`：不等。取值 0 到 600。",
+                        "Seconds a request waits in all, counted once the key's own `max_concurrent` lets it in: for a key's `minute` or `hour` limit to free up, for an upstream that cannot take it yet — at its `max_concurrent`, paused after answering 429, or at one of its `minute` or `hour` usage limits — and before retrying upstreams that failed temporarily before any content (overloaded, a server error, unreachable). A key limit that does not free up in time refuses the request. An upstream that cannot take it is skipped for the next one, and when none is left the request waits for the first to become available; a conversation that stays on an upstream waits for it. When nothing became available in time, the client gets 429 with `Retry-After`; when retries did not succeed in time, it gets the last upstream error. `0`: never wait or retry. From 0 to 600.",
+                        "一个请求合计最多等的秒数，从过了密钥自己的 `max_concurrent` 时算起：等密钥的 `minute`、`hour` 上限空出名额，等暂时不能发的上游 —— 并发数满了（`max_concurrent`）、回了 429 正在暂停、到了这家 `minute` 或 `hour` 的用量上限 —— 和上游在内容之前暂时出错（过载、服务端出错、连不上）后重试之前的等待，都算在里面。密钥的上限到时空不出来就拒绝。上游暂时不能发的先换下一家，没有下一家了就等最先恢复的那一家；留在某一家的对话等这一家。到时都没有恢复，客户端收到带 `Retry-After` 的 429；到时重试仍未成功，客户端收到最后一次的上游错误。`0`：不等，也不重试。取值 0 到 600。",
                     ),
                 ),
             ],

crates/tw-control/tests/error_answers.rs

@@ -89,10 +89,12 @@ struct World {
 async fn world() -> World {
     let up = upstream().await;
     let d = tempfile::tempdir().unwrap();
+    // 不等、不重试（`wait_secs: 0`）：唯一的上游不压开头，第三轮吐的那一帧当场交给客户端
     let yaml = format!(
         "version: 1\nlisten:\n  control:\n    key: c0ffee00c0ffee00c0ffee00c0ffee00c0ffee00c0ffee00c0ffee00c0ffee00\n\
          clients:\n  - name: 我\n    key: tw-k\n\
-         providers:\n  - name: 中转\n    base_url: http://{up}\n    key: sk-upstream\n    protocol: anthropic\n"
+         providers:\n  - name: 中转\n    base_url: http://{up}\n    key: sk-upstream\n    protocol: anthropic\n\
+         failover:\n  wait_secs: 0\n"
     );
     let p = d.path().join("config.yaml");
     std::fs::write(&p, &yaml).unwrap();

crates/tw-control/tests/handover.rs

@@ -146,7 +146,7 @@ async fn status_reports_the_handover_phase_and_the_draining_count() {
     let st = status(&s).await;
     assert_eq!(st.handover, tw_api::HandoverState::Draining);
     assert_eq!(st.draining, 3);
-    assert_eq!(st.api_version, 52);
+    assert_eq!(st.api_version, 53);
 }
 
 /// arm 之后（以及接替者正在接管时）不开始新的账号登录：回调端口不交接。（平常、排空中的

crates/tw-control/tests/in_flight.rs

@@ -102,6 +102,8 @@ async fn it_gives_every_running_request_with_what_has_happened_to_it_so_far() {
             queued_ms: None,
             skipped: None,
             pause_ms: None,
+            retry: None,
+            backoff_ms: None,
             proxy: None,
         }],
         billing: tw_api::Billing::PerToken,

crates/tw-control/tests/key_limits.rs

@@ -145,6 +145,8 @@ fn request(id: u64, key: &str, at: i64, input: u64) -> Vec<Event> {
                 queued_ms: None,
                 skipped: None,
                 pause_ms: None,
+                retry: None,
+                backoff_ms: None,
                 proxy: None,
             }],
             billing: tw_api::Billing::PerToken,
@@ -456,6 +458,8 @@ fn hop(outcome: tw_api::AttemptOutcome, code: &str, busy: bool) -> tw_api::Attem
         queued_ms: None,
         skipped: busy.then_some(tw_api::ServeSkip::Busy),
         pause_ms: None,
+        retry: None,
+        backoff_ms: None,
         proxy: None,
     }
 }

crates/tw-control/tests/live_state.rs

@@ -159,6 +159,8 @@ async fn a_request_still_running_can_be_opened_and_becomes_whole_when_it_ends()
                 queued_ms: None,
                 skipped: None,
                 pause_ms: None,
+                retry: None,
+                backoff_ms: None,
                 proxy: None,
             }],
             billing: tw_api::Billing::PerToken,
@@ -249,7 +251,7 @@ async fn the_overview_says_which_proxy_is_down_and_why() {
     let dead = dead_addr();
     let d = tempfile::tempdir().unwrap();
     let yaml = format!(
-        "version: 1\nlisten:\n  control:\n    key: c0ffee00c0ffee00c0ffee00c0ffee00c0ffee00c0ffee00c0ffee00c0ffee00\nclients:\n  - name: 我\n    key: tw-k\nproxies:\n  - name: 代理一\n    type: http\n    addr: {dead}\nproviders:\n  - name: 中转\n    base_url: http://127.0.0.1:9\n    protocol: anthropic\n    key: sk-x\n    proxy: 代理一\n"
+        "version: 1\nlisten:\n  control:\n    key: c0ffee00c0ffee00c0ffee00c0ffee00c0ffee00c0ffee00c0ffee00c0ffee00\nclients:\n  - name: 我\n    key: tw-k\nproxies:\n  - name: 代理一\n    type: http\n    addr: {dead}\nproviders:\n  - name: 中转\n    base_url: http://127.0.0.1:9\n    protocol: anthropic\n    key: sk-x\n    proxy: 代理一\nfailover:\n  wait_secs: 0\n"
     );
     let (gw, app) = control(&d, &yaml, None);
     let mut rx = gw.bus.subscribe();

crates/tw-control/tests/replay.rs

@@ -392,6 +392,8 @@ fn routed(hops: &[(&str, Option<&str>)]) -> Option<String> {
             queued_ms: None,
             skipped: None,
             pause_ms: None,
+            retry: None,
+            backoff_ms: None,
             proxy: None,
         })
         .collect();

crates/tw-control/tests/upstream_limits.rs

@@ -108,6 +108,8 @@ fn attempt(provider: &str, status: u16) -> tw_api::AttemptView {
         queued_ms: None,
         skipped: None,
         pause_ms: None,
+        retry: None,
+        backoff_ms: None,
         proxy: None,
     }
 }

crates/tw-gateway/src/guard.rs

@@ -182,6 +182,35 @@ pub fn more_found(before: &[Finding], after: Vec<Finding>) -> Vec<Finding> {
         .collect()
 }
 
+/// 一个请求已经报过的命中。插件改过的请求每一跳都再查一遍内容过滤、再找一遍密钥（见
+/// `server::pipeline::plug`）：重试、故障转移时插件写进来的还是那几处，**一个请求只报一次** ——
+/// 安全日志里一处一行，不因为多试了几跳多出几行
+#[derive(Default)]
+pub struct Reported {
+    /// 出站脱敏报过的值：规则和打了码的那个值
+    secrets: HashSet<(String, String)>,
+    /// 内容过滤报过的命中：规则和摘录
+    content: HashSet<(String, String)>,
+}
+
+impl Reported {
+    /// `found` 里这个请求还没报过的，记下来
+    pub fn fresh_secrets(&mut self, found: Vec<Finding>) -> Vec<Finding> {
+        found
+            .into_iter()
+            .filter(|f| {
+                self.secrets
+                    .insert((f.rule.id().to_string(), f.masked.clone()))
+            })
+            .collect()
+    }
+
+    /// 先前几跳已经报过几个值。一个请求加起来最多报 [`REPORTED_MAX`] 个
+    pub fn secrets(&self) -> usize {
+        self.secrets.len()
+    }
+}
+
 /// 一个请求报出去的值最多几个。
 ///
 /// 安全日志一个值一行（[`tw_api::Event::SecretsFound`] 的一项就是一行）。一个请求里贴进来
@@ -319,6 +348,44 @@ pub fn report_with(
     sc.refusal().map(|r| refusal(&r.hit))
 }
 
+/// [`report_with`]，这个请求先前几跳报过的命中不再报（见 [`Reported`]）。拒绝照旧按这一次查下来
+/// 的结论：拒绝的那一条先前报过的话，请求那时就被拒了，到不了这一跳
+pub fn report_once(
+    bus: &tw_observe::EventBus,
+    id: u64,
+    provider: &str,
+    sc: &Screening,
+    details: Vec<tw_api::SecurityHitDetail>,
+    reported: &mut Reported,
+) -> Option<tw_types::Msg> {
+    if sc.hits.is_empty() {
+        return None;
+    }
+    let refused = sc.refusal().map(|r| refusal(&r.hit));
+    let refusing = sc.refused;
+    let mut details = details.into_iter();
+    let mut fresh = Screening {
+        body: sc.body.clone(),
+        ..Default::default()
+    };
+    let mut kept = Vec::new();
+    for (i, h) in sc.hits.iter().enumerate() {
+        let detail = details.next().unwrap_or_default();
+        if reported
+            .content
+            .insert((h.hit.rule.clone(), h.hit.snippet.clone()))
+        {
+            if refusing == Some(i) {
+                fresh.refused = Some(fresh.hits.len());
+            }
+            fresh.hits.push(h.clone());
+            kept.push(detail);
+        }
+    }
+    report_with(bus, id, provider, &fresh, kept);
+    refused
+}
+
 /// 回答里的命中报出去：每条规则一条 [`tw_api::Event::ContentMatched`]（方向是回答），挂在请求
 /// `id` 上。片段和细节按这个请求的脱敏规则和账本打码（[`detail::answer`]）。
 pub fn report_answer(

crates/tw-gateway/src/lib.rs

@@ -50,6 +50,7 @@ pub mod probe;
 mod pulse;
 pub mod quota;
 pub mod quote;
+pub mod retry;
 pub mod seal;
 pub mod sent;
 pub mod server;