fix(failover): patch P1-P3 reliability gaps surfaced by team review

- Forwarder buffers non-streaming bodies and primes streaming first
  chunk before signaling success, so body timeouts and SSE first-chunk
  failures route through the circuit breaker instead of being recorded
  as success on response-header arrival
- Atomic enable-failover: switch to P1 before persisting the flag, and
  roll back auto-added queue entries when the switch is rejected
  (e.g. official providers)
- Hot-reload circuit breaker config on per-app proxy config change
  instead of waiting for a proxy restart
- FailoverToggle / FailoverQueueManager / AutoFailoverConfigPanel
  require proxy takeover for the active app; the backend command also
  rejects enabling when takeover is off
- ProviderHealthBadge consumes the backend is_healthy flag instead of
  hardcoding the 5-failure threshold

Cleanup:
- impl From<&AppProxyConfig> for CircuitBreakerConfig and use it from
  the command layer
- Collapse three identical TabsContent blocks into a single map
This commit is contained in:
Jason
2026-05-14 21:35:02 +08:00
parent 940161fb0e
commit b642ef0633
14 changed files with 389 additions and 98 deletions
+47 -1
View File
@@ -6,7 +6,7 @@
use super::ProxyError;
use bytes::Bytes;
use futures::stream::Stream;
use futures::{stream::Stream, StreamExt};
use http_body_util::BodyExt;
use hyper_rustls::HttpsConnectorBuilder;
use hyper_util::{client::legacy::Client, rt::TokioExecutor};
@@ -79,13 +79,44 @@ fn global_hyper_client() -> &'static HyperClient {
pub enum ProxyResponse {
Hyper(hyper::Response<hyper::body::Incoming>),
Reqwest(reqwest::Response),
Buffered {
status: http::StatusCode,
headers: http::HeaderMap,
body: Bytes,
},
Streamed {
status: http::StatusCode,
headers: http::HeaderMap,
stream: std::pin::Pin<Box<dyn Stream<Item = Result<Bytes, std::io::Error>> + Send>>,
},
}
impl ProxyResponse {
pub fn buffered(status: http::StatusCode, headers: http::HeaderMap, body: Bytes) -> Self {
Self::Buffered {
status,
headers,
body,
}
}
pub fn streamed(
status: http::StatusCode,
headers: http::HeaderMap,
stream: impl Stream<Item = Result<Bytes, std::io::Error>> + Send + 'static,
) -> Self {
Self::Streamed {
status,
headers,
stream: Box::pin(stream),
}
}
pub fn status(&self) -> http::StatusCode {
match self {
Self::Hyper(r) => r.status(),
Self::Reqwest(r) => r.status(),
Self::Buffered { status, .. } | Self::Streamed { status, .. } => *status,
}
}
@@ -93,6 +124,7 @@ impl ProxyResponse {
match self {
Self::Hyper(r) => r.headers(),
Self::Reqwest(r) => r.headers(),
Self::Buffered { headers, .. } | Self::Streamed { headers, .. } => headers,
}
}
@@ -122,6 +154,17 @@ impl ProxyResponse {
Self::Reqwest(r) => r.bytes().await.map_err(|e| {
ProxyError::ForwardFailed(format!("Failed to read response body: {e}"))
}),
Self::Buffered { body, .. } => Ok(body),
Self::Streamed { mut stream, .. } => {
let mut body = bytes::BytesMut::new();
while let Some(chunk) = stream.next().await {
let chunk = chunk.map_err(|e| {
ProxyError::ForwardFailed(format!("Failed to read response body: {e}"))
})?;
body.extend_from_slice(&chunk);
}
Ok(body.freeze())
}
}
}
@@ -161,6 +204,9 @@ impl ProxyResponse {
.map(|r| r.map_err(|e| std::io::Error::other(e.to_string())));
Box::pin(stream)
}
Self::Buffered { body, .. } => Box::pin(futures::stream::once(async move { Ok(body) }))
as std::pin::Pin<Box<dyn Stream<Item = Result<Bytes, std::io::Error>> + Send>>,
Self::Streamed { stream, .. } => stream,
}
}
}