fix(gateway): 改进流式传输稳定性

- stream_pump 读取错误时记录完整错误链并输出 warn 日志
- hub body 转发从 try_send 改为 async send,支持背压避免丢帧
- tunnel_stale_timeout 默认值从 900ms 提升到 10s,减少误判过期
This commit is contained in:
fawney19
2026-04-15 14:26:57 +08:00
parent 707d9ac274
commit fbb8249c0d
3 changed files with 34 additions and 7 deletions

View File

@@ -8,6 +8,7 @@ use async_stream::stream;
use axum::body::Bytes;
use base64::Engine as _;
use futures_util::{Stream, StreamExt};
use tracing::warn;
use crate::execution_runtime::ndjson::encode_stream_frame_ndjson;
use crate::execution_runtime::DirectUpstreamStreamExecution;
@@ -87,13 +88,22 @@ pub(crate) fn build_direct_execution_frame_stream(
}
}
Err(err) => {
let message = format_error_chain(&err);
warn!(
event_name = "stream_pump_body_read_error",
log_type = "ops",
status_code,
upstream_bytes,
error = %message,
"upstream body stream read error"
);
let frame = StreamFrame {
frame_type: StreamFrameType::Error,
payload: StreamFramePayload::Error {
error: ExecutionError {
kind: ExecutionErrorKind::Internal,
phase: ExecutionPhase::StreamRead,
message: err.to_string(),
message,
upstream_status: Some(status_code),
retryable: false,
failover_recommended: false,
@@ -136,6 +146,17 @@ pub(crate) fn build_direct_execution_frame_stream(
}
}
fn format_error_chain(err: &(dyn std::error::Error + 'static)) -> String {
let mut message = err.to_string();
let mut source = err.source();
while let Some(cause) = source {
message.push_str(": ");
message.push_str(&cause.to_string());
source = cause.source();
}
message
}
#[cfg(test)]
mod tests {
use std::collections::BTreeMap;

View File

@@ -278,12 +278,13 @@ impl LocalStream {
}
}
fn push_body_chunk(&self, payload: Bytes) -> bool {
async fn push_body_chunk(&self, payload: Bytes) -> bool {
if self.terminal.load(Ordering::Acquire) {
return false;
}
self.body_tx
.try_send(LocalBodyEvent::Chunk(payload))
.send(LocalBodyEvent::Chunk(payload))
.await
.is_ok()
}
@@ -644,7 +645,7 @@ impl HubRouter {
self.route_response_headers(proxy_conn_id, header, data);
}
protocol::RESPONSE_BODY => {
self.route_response_body(proxy_conn_id, header, data);
self.route_response_body(proxy_conn_id, header, data).await;
}
protocol::STREAM_END => {
self.finish_proxy_stream(proxy_conn_id, header.stream_id);
@@ -720,7 +721,12 @@ impl HubRouter {
}
}
fn route_response_body(&self, proxy_conn_id: u64, header: protocol::FrameHeader, data: &[u8]) {
async fn route_response_body(
&self,
proxy_conn_id: u64,
header: protocol::FrameHeader,
data: &[u8],
) {
let Some(local_id) = self.lookup_local_stream(proxy_conn_id, header.stream_id) else {
return;
};
@@ -738,7 +744,7 @@ impl HubRouter {
None => return,
};
if !stream.push_body_chunk(Bytes::from(payload)) {
if !stream.push_body_chunk(Bytes::from(payload)).await {
self.cancel_local_stream(local_id, "local relay response congested");
}
}

View File

@@ -55,7 +55,7 @@ pub const DEFAULT_LOG_RETENTION_DAYS: u64 = 7;
pub const DEFAULT_LOG_MAX_FILES: usize = 30;
pub const DEFAULT_TUNNEL_PING_INTERVAL_MS: u64 = 250;
pub const DEFAULT_TUNNEL_CONNECT_TIMEOUT_MS: u64 = 800;
pub const DEFAULT_TUNNEL_STALE_TIMEOUT_MS: u64 = 900;
pub const DEFAULT_TUNNEL_STALE_TIMEOUT_MS: u64 = 10_000;
pub const DEFAULT_TUNNEL_SCALE_CHECK_INTERVAL_MS: u64 = 1_000;
pub const DEFAULT_TUNNEL_SCALE_UP_THRESHOLD_PERCENT: u32 = 70;
pub const DEFAULT_TUNNEL_SCALE_DOWN_THRESHOLD_PERCENT: u32 = 35;