Skip to content

Commit f6f915d

Browse files
committed
fix(gateway): classify stream faults before recording health
A terminal SSE error can arrive over HTTP 200 and a clean EOF. Pulse correctly treats errors as progress for idle timing, but the relay used that signal to record success before inspecting the error. Its EOF path also only applied rate limits, so balance and quota failures stayed healthy. Inspect errors before recording an opening success, reuse the shared stream-fault classifier and known quota reset, and flush undelimited error frames before final health accounting. Apply the same EOF tolerance while buffering a stream opening so the original error is retained there too. Keep rate limits out of the health tally and retain the existing success after content has started. Cover the terminal health tally, pause and reset behavior, original error forwarding, subsequent routing, all four SSE dialects and no replay after content.
1 parent d863b16 commit f6f915d

5 files changed

Lines changed: 303 additions & 51 deletions

File tree

‎crates/tw-gateway/src/ending.rs‎

Lines changed: 18 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -76,7 +76,7 @@ pub struct Ending {
7676
/// 上游在流里报的错:哪一家、原话。**有它就不是成功** —— 响应头是 200,回答却断在了
7777
/// 半路
7878
upstream_error: Option<(String, String)>,
79-
/// 上游报错的那一帧的 `data`:回程照它认是不是限流(见 [`Ending::upstream_fault`])
79+
/// 上游报错的那一帧的 `data`:回程照它认失败原因(见 [`Ending::upstream_fault`])
8080
upstream_fault: Option<String>,
8181
/// 上游报的错是解不开对话里别家封存的内容(见 [`tw_dialect::sealed`]):失败的说明讲清楚
8282
/// 怎么回事,不只转述原话
@@ -495,12 +495,26 @@ impl Ending {
495495
}
496496

497497
/// 上游在流里报了错:那一帧的 `data`(上游的格式,[`Ending::streaming`] 时给的)。回程照它
498-
/// 认是不是限流 —— 答到一半才报的限流,回答照样以错误收尾,这一家照样暂停(见
499-
/// [`crate::throttle`])
498+
/// 认失败原因;流结束后先调 [`Self::flush_stream_error`],没有空行的尾帧也算。
500499
pub fn upstream_fault(&self) -> Option<&str> {
501500
self.upstream_fault.as_deref()
502501
}
503502

503+
/// 上游不再来字节了:看最后一帧有没有报错。只收错误嗅探器,不报结局、不结算用量,
504+
/// 回程先按同一个错误记这一家的成败,之后再交给 [`Self::finished`] 或 [`Self::failed`]。
505+
pub fn flush_stream_error(&mut self) {
506+
if let Some(mut w) = self.watch.take() {
507+
let frames = w.frames.flush();
508+
if self.upstream_error.is_none()
509+
&& let Some((found, data)) = w.error_in(frames)
510+
{
511+
self.sealed_refused = sealed_refusal(w.dialect, &data);
512+
self.upstream_error = Some(found);
513+
self.upstream_fault = Some(data);
514+
}
515+
}
516+
}
517+
504518
/// 上游 `provider` 回的不是 2xx,原样交给了客户端(4xx 是请求本身的问题,或者没有
505519
/// 下一家可换了;3xx 交还客户端,由它决定跟不跟)。
506520
///
@@ -749,17 +763,7 @@ impl Ending {
749763
self.fail(tw_api::FailureSource::Upstream, why);
750764
return;
751765
}
752-
// 最后一帧后面不带空行的上游:收尾时再看一眼
753-
if let Some(mut w) = self.watch.take() {
754-
let frames = w.frames.flush();
755-
if self.upstream_error.is_none()
756-
&& let Some((found, data)) = w.error_in(frames)
757-
{
758-
self.sealed_refused = sealed_refusal(w.dialect, &data);
759-
self.upstream_error = Some(found);
760-
self.upstream_fault = Some(data);
761-
}
762-
}
766+
self.flush_stream_error();
763767
if let Some((upstream, message)) = self.upstream_error.take() {
764768
let message: String = message.chars().take(500).collect();
765769
let why = if self.sealed_refused {

‎crates/tw-gateway/src/server/pipeline/hop.rs‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -2881,8 +2881,8 @@ fn known_reset(state: &AppState, provider: &str, cause: Cause) -> Cause {
28812881
/// 一个流式回答在第一段内容之前就收了尾,收尾的那个事件是 `data`:这算不算 `provider` 的错、
28822882
/// 按什么原因停用。**和这里开头报错的一跳同一个判据**([`super::opening`] 读出状态码,
28832883
/// [`crate::failure::classify`] 判断):5xx、限流、额度用完、凭据被拒是 `Some`;请求本身的
2884-
/// 问题、不是错误的是 `None` —— 那是这一家答上了。WebSocket 上 Responses 的一轮用它(见
2885-
/// `crate::ws::turn`)
2884+
/// 问题、不是错误的是 `None` —— 那是这一家答上了。HTTP 回程和 WebSocket 上 Responses 的一轮
2885+
/// 都用它(见 [`super::relay`]、`crate::ws::turn`)
28862886
pub(crate) fn stream_fault(
28872887
state: &AppState,
28882888
provider: &str,

‎crates/tw-gateway/src/server/pipeline/opening.rs‎

Lines changed: 16 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -91,8 +91,14 @@ pub(super) async fn watch(
9191
}
9292
};
9393
let chunk = match next {
94-
// 只有开头的例行事件就收了尾:一段内容都没有,和断了一样
95-
None => return Opening::Broken(ended()),
94+
None => {
95+
// 和回程一样容忍尾帧没有空行。只补给解析器,转发的仍是上游原字节;
96+
// 只有开头的例行事件就收了尾的,照旧算没有内容。
97+
let Some(judged) = sse.judge(dialect, b"\n\n") else {
98+
return Opening::Broken(ended());
99+
};
100+
break judged;
101+
}
96102
Some(Ok(c)) => c,
97103
Some(Err(e)) => {
98104
return Opening::Broken(crate::forward::map_reqwest_error(e));
@@ -112,12 +118,7 @@ pub(super) async fn watch(
112118
},
113119
};
114120
usage.feed(&text);
115-
let judged = sse
116-
.feed(&text)
117-
.into_iter()
118-
.map(|(event, data)| judge(dialect, event.as_deref(), &data))
119-
.find(|j| !matches!(j, Judge::Preamble));
120-
if let Some(j) = judged {
121+
if let Some(j) = sse.judge(dialect, &text) {
121122
break j;
122123
}
123124
if size >= HOLD_LIMIT {
@@ -422,6 +423,13 @@ struct Sse {
422423
}
423424

424425
impl Sse {
426+
fn judge(&mut self, dialect: Dialect, bytes: &[u8]) -> Option<Judge> {
427+
self.feed(bytes)
428+
.into_iter()
429+
.map(|(event, data)| judge(dialect, event.as_deref(), &data))
430+
.find(|j| !matches!(j, Judge::Preamble))
431+
}
432+
425433
fn feed(&mut self, bytes: &[u8]) -> Vec<(Option<String>, String)> {
426434
self.buf.push_str(&String::from_utf8_lossy(bytes));
427435
let mut out = Vec::new();

‎crates/tw-gateway/src/server/pipeline/relay.rs‎

Lines changed: 30 additions & 27 deletions
Original file line numberDiff line numberDiff line change
@@ -21,6 +21,7 @@ use futures::StreamExt;
2121
use super::Inbound;
2222
use super::hop::Served;
2323
use crate::error::GatewayError;
24+
use crate::failure::Cause;
2425
use crate::forward;
2526
use crate::server::{because, flagged};
2627
use crate::state::{AppState, Runtime};
@@ -197,6 +198,7 @@ pub(super) fn respond(
197198
let throttles = state.throttles.clone();
198199
// 这个请求结束时记一笔,到了时候重读这一家的余额(见 `crate::balances`)
199200
let passing = state.balance_passing(&upstream_name);
201+
let state = state.clone();
200202
let stream = async_stream::stream! {
201203
// **通行证跟着响应体走。**这个流被丢掉的时候它才还回去:正常
202204
// 发完是一种,客户端中途断开、hyper 丢掉响应体是另一种 —— 两种
@@ -327,11 +329,19 @@ pub(super) fn respond(
327329
match item {
328330
Ok(chunk) => {
329331
upstream_since = tokio::time::Instant::now();
332+
// **旁路嗅探和留档,不缓冲**:看的是上游原话(带占位符的那一版),
333+
// usage 和请求记录不受后面的还原、转换影响。先认错误,再记这一家的成败。
334+
ending.feed(&chunk);
330335
// 真内容才重新计时:心跳、开头的例行事件不算
331336
if pulse.feed(&chunk) {
332337
said = true;
333338
quiet_by = quiet.after(std::time::Instant::now());
334-
if std::mem::take(&mut unconfirmed) {
339+
// 错误帧也算上游有动静(不能报静默超时),却不是答上了。
340+
// 同一块里已经有内容再报错的,保留开头成功的那一笔。
341+
if unconfirmed
342+
&& (ending.has_content() || ending.upstream_fault().is_none())
343+
{
344+
unconfirmed = false;
335345
crate::server::note_health(
336346
&bus,
337347
&health,
@@ -340,13 +350,6 @@ pub(super) fn respond(
340350
);
341351
}
342352
}
343-
// **旁路嗅探和留档,不缓冲**:字节照常流向客户端,同时
344-
// 喂它一份。上游返回的 usage 是真相,而拿不到它就只能估。
345-
//
346-
// 看的是上游原话(带占位符的那一版):usage 数字不受
347-
// 影响,而请求详情里存的正是「我们发出去的和收回来的」,
348-
// 把还原后的存进去会让那一页说谎。
349-
ending.feed(&chunk);
350353
let (out, cut) = relay.chunk(&chunk).await;
351354
relay.sent(&out);
352355
if !out.is_empty() {
@@ -400,31 +403,31 @@ pub(super) fn respond(
400403
);
401404
ending.too_slow(rate);
402405
}
403-
// 上游在流里报的错是限流:回答照样以错误收尾(客户端已经收到了东西,换不了),这一家暂停
404-
if let Some((code, body)) = ending
406+
// 尾帧没带空行的也先认出来:健康记录和请求结局看的是同一个错误。
407+
ending.flush_stream_error();
408+
let fault = ending
405409
.upstream_fault()
406-
.and_then(|data| super::opening::stream_error(upstream_dialect, data))
407-
&& let crate::failure::Verdict::Failed(crate::failure::Cause::RateLimited {
408-
retry_after,
409-
}) = crate::failure::classify(
410-
code,
411-
&http::HeaderMap::new(),
412-
&body,
413-
crate::server::now_ms(),
414-
)
415-
{
410+
.and_then(|data| super::hop::stream_fault(&state, &upstream_name, upstream_dialect, data));
411+
// 答到一半才报的限流也暂停;客户端已经收到了东西,这一轮不重放。
412+
if let Some(Cause::RateLimited { retry_after }) = fault {
416413
throttles.limited(&upstream_name, sent_at, retry_after);
417414
}
418415
// 响应体留档和结束事件都在 `ending` 里:三种结局要交出去的是同一份
419416
// 东西,分开写就会有一种漏掉
420-
// 一段真内容都没有就好好收了尾:也是答上了
417+
// 还没有记过成败:正常 EOF 不代表成功,流里报的错和压着开头时按同一个原因记。
421418
if unconfirmed && broke.is_none() {
422-
crate::server::note_health(
423-
&bus,
424-
&health,
425-
&upstream_name,
426-
health.record_success(&upstream_name),
427-
);
419+
let changed = match fault {
420+
// 限流由上面暂停,不进健康的成功率和熔断。
421+
Some(Cause::RateLimited { .. }) => None,
422+
Some(cause) => {
423+
if matches!(cause, Cause::NoBalance | Cause::QuotaUsedUp { .. }) {
424+
throttles.released(&upstream_name, sent_at);
425+
}
426+
health.record_cause(&upstream_name, cause)
427+
}
428+
None => health.record_success(&upstream_name),
429+
};
430+
crate::server::note_health(&bus, &health, &upstream_name, changed);
428431
}
429432
match broke {
430433
None => {

0 commit comments

Comments
 (0)