i/Aether
1
0
Fork 0
mirror of https://github.com/fawney19/Aether.git synced 2026-09-01 17:00:21 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat(tunnel): 添加流帧分发超时机制和结构化请求日志

Browse Source 
- dispatcher: 为 stream handler 的帧接收添加超时保护, 防止单个
  handler 阻塞 WebSocket 读循环; 超时后发送 StreamError 并清理流
- stream_handler: 在所有请求完成和错误路径添加结构化日志, 记录
  method/host/path/status/duration 等关键信息
- 将常规隧道连接/断开日志从 info 降级为 debug, 减少日志噪音
This commit is contained in:
fawney19
2026-04-16 10:32:05 +08:00
parent 28a489acbe
commit 47a11ee0b5
4 changed files with 433 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
apps/aether-proxy/src/tunnel/client.rs
View File

@@ -8,7 +8,7 @@ use tokio::sync::watch;
use tokio_tungstenite::tungstenite::client::IntoClientRequest;
use tokio_tungstenite::tungstenite::http;
use tokio_tungstenite::tungstenite::protocol::WebSocketConfig;
use tracing::{debug, info, warn};
use tracing::{debug, warn};
use crate::state::{AppState, ServerContext};
@@ -34,7 +34,7 @@ pub async fn connect_and_run(
drain: watch::Receiver<bool>,
) -> Result<TunnelOutcome, anyhow::Error> {
let ws_url = build_tunnel_url(server);
info!(url = %ws_url, conn = conn_idx, "connecting tunnel");
debug!(url = %ws_url, conn = conn_idx, "connecting tunnel");
// Build WebSocket request with auth headers
let mut request = ws_url.clone().into_client_request()?;
@@ -124,7 +124,7 @@ pub async fn connect_and_run(
.config
.tunnel_ping_interval()
.expect("validated config should resolve tunnel ping interval");
info!(
debug!(
conn = conn_idx,
tcp_keepalive_secs = state.config.tunnel_tcp_keepalive_secs,
tcp_nodelay = state.config.tunnel_tcp_nodelay,
@@ -213,7 +213,7 @@ pub async fn connect_and_run(
let _ = tokio::time::timeout(Duration::from_secs(35), writer_handle).await;
}
info!("tunnel disconnected");
debug!("tunnel disconnected");
Ok(outcome)
}

135
apps/aether-proxy/src/tunnel/dispatcher.rs
View File

@@ -19,6 +19,13 @@ use super::protocol::{decompress_if_gzip, Frame, MsgType, RequestMeta};
use super::stream_handler;
use super::writer::FrameSender;
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum StreamDispatchStatus {
Delivered,
Closed,
TimedOut,
}
/// Run the dispatcher loop, reading from the WebSocket stream.
pub async fn run<S>(
state: Arc<AppState>,
@@ -97,7 +104,7 @@ where
Message::Ping(_) => continue,
Message::Pong(_) => continue,
Message::Close(_) => {
info!("received WebSocket close");
debug!("received WebSocket close");
break None;
}
_ => continue,
@@ -209,12 +216,19 @@ where
}
MsgType::RequestBody => {
if let Some(tx) = streams.get(&frame.stream_id) {
if let Some(tx) = streams.get(&frame.stream_id).cloned() {
let is_end = frame.is_end_stream();
let sid = frame.stream_id;
let _ = tx.send(frame).await;
if is_end {
let dispatch = dispatch_stream_frame(&tx, frame).await;
if is_end || dispatch != StreamDispatchStatus::Delivered {
streams.remove(&sid);
if dispatch == StreamDispatchStatus::TimedOut {
try_send_stream_error(
&frame_tx,
sid,
"proxy request body dispatch stalled",
);
}
if draining && streams.is_empty() {
info!("tunnel drained after request body completion");
break None;
@@ -226,7 +240,7 @@ where
MsgType::StreamEnd | MsgType::StreamError => {
// Client-side cancellation or end
if let Some(tx) = streams.remove(&frame.stream_id) {
let _ = tx.send(frame).await;
let _ = dispatch_stream_frame(&tx, frame).await;
if draining && streams.is_empty() {
info!("tunnel drained after stream termination");
break None;
@@ -280,6 +294,59 @@ where
}
}
async fn dispatch_stream_frame(tx: &mpsc::Sender<Frame>, frame: Frame) -> StreamDispatchStatus {
let stream_id = frame.stream_id;
match tokio::time::timeout(stream_frame_dispatch_timeout(), tx.send(frame)).await {
Ok(Ok(())) => StreamDispatchStatus::Delivered,
Ok(Err(_)) => {
warn!(
stream_id,
"stream handler channel closed while dispatching tunnel frame"
);
StreamDispatchStatus::Closed
}
Err(_) => {
warn!(
stream_id,
timeout_ms = stream_frame_dispatch_timeout().as_millis(),
"stream handler channel blocked while dispatching tunnel frame"
);
StreamDispatchStatus::TimedOut
}
}
}
/// Bound how long a single stream handler is allowed to block the shared
/// WebSocket read loop while receiving request-body frames.
fn stream_frame_dispatch_timeout() -> Duration {
#[cfg(test)]
{
Duration::from_millis(25)
}
#[cfg(not(test))]
{
Duration::from_secs(5)
}
}
fn try_send_stream_error(frame_tx: &FrameSender, stream_id: u32, message: &'static str) {
if frame_tx
.try_send(Frame::new(
stream_id,
MsgType::StreamError,
0,
Bytes::from(message),
))
.is_err()
{
warn!(
stream_id,
"writer channel full, StreamError dropped while aborting stalled stream"
);
}
}
/// Wait for all active stream handlers to finish (with a timeout).
async fn drain_handlers(handles: Vec<JoinHandle<()>>) {
if handles.is_empty() {
@@ -294,3 +361,61 @@ async fn drain_handlers(handles: Vec<JoinHandle<()>>) {
})
.await;
}
#[cfg(test)]
mod tests {
use super::*;
use aether_runtime::bounded_queue;
#[tokio::test]
async fn dispatch_stream_frame_times_out_when_handler_stops_draining() {
let (tx, mut rx) = mpsc::channel::<Frame>(1);
tx.send(Frame::new(
7,
MsgType::RequestBody,
0,
Bytes::from_static(b"first"),
))
.await
.expect("first frame should enqueue");
let stalled_send = tokio::spawn({
let tx = tx.clone();
async move {
dispatch_stream_frame(
&tx,
Frame::new(7, MsgType::RequestBody, 0, Bytes::from_static(b"second")),
)
.await
}
});
assert_eq!(
stalled_send.await.expect("dispatch task should join"),
StreamDispatchStatus::TimedOut
);
let retained = rx
.recv()
.await
.expect("queued frame should still be present");
assert_eq!(retained.payload, Bytes::from_static(b"first"));
}
#[tokio::test]
async fn try_send_stream_error_emits_stream_error_frame() {
let (frame_tx, mut frame_rx) = bounded_queue::<Frame>(4);
try_send_stream_error(&frame_tx, 9, "proxy request body dispatch stalled");
let frame = frame_rx
.recv()
.await
.expect("stream error frame should enqueue");
assert_eq!(frame.stream_id, 9);
assert_eq!(frame.msg_type, MsgType::StreamError);
assert_eq!(
frame.payload,
Bytes::from_static(b"proxy request body dispatch stalled")
);
}
}

28
apps/aether-proxy/src/tunnel/mod.rs
View File

@@ -9,7 +9,7 @@ use std::sync::Arc;
use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
use tokio::sync::watch;
use tracing::{error, info};
use tracing::{debug, error, info};
use crate::state::{AppState, ServerContext};
@@ -83,7 +83,7 @@ pub async fn run(
return;
}
Ok(client::TunnelOutcome::Disconnected) => {
info!(server = %server.server_label, conn = conn_idx, "tunnel disconnected, reconnecting");
debug!(server = %server.server_label, conn = conn_idx, "tunnel disconnected, reconnecting");
}
Err(e) => {
error!(server = %server.server_label, conn = conn_idx, error = %e, "tunnel connection error, reconnecting");
@@ -114,13 +114,23 @@ pub async fn run(
consecutive_failures,
reconnect_salt,
);
info!(
server = %server.server_label,
conn = conn_idx,
failures = consecutive_failures,
delay_ms = reconnect_delay.as_millis(),
"waiting before reconnect"
);
if reconnect_delay.is_zero() && consecutive_failures <= 1 {
debug!(
server = %server.server_label,
conn = conn_idx,
failures = consecutive_failures,
delay_ms = reconnect_delay.as_millis(),
"waiting before reconnect"
);
} else {
info!(
server = %server.server_label,
conn = conn_idx,
failures = consecutive_failures,
delay_ms = reconnect_delay.as_millis(),
"waiting before reconnect"
);
}
tokio::select! {
_ = tokio::time::sleep(reconnect_delay) => {}

302
apps/aether-proxy/src/tunnel/stream_handler.rs
View File

@@ -16,7 +16,7 @@ use futures_util::StreamExt;
use http_body_util::BodyExt;
use hyper::body::Frame as BodyFrame;
use tokio::sync::mpsc;
use tracing::{debug, warn};
use tracing::{debug, info, warn};
use crate::state::{AppState, ServerContext};
use crate::target_filter;
@@ -116,6 +116,110 @@ struct UpstreamResponseContext {
request_timing: upstream_client::RequestTiming,
}
#[derive(Clone, Copy)]
struct StreamLogContext<'a> {
server: &'a ServerContext,
stream_id: u32,
method: &'a hyper::Method,
url: Option<&'a url::Url>,
redirect_count: usize,
request_body_size: usize,
}
fn parse_request_method(method: &str) -> hyper::Method {
method.parse().unwrap_or(hyper::Method::GET)
}
fn request_log_host(url: &url::Url) -> &str {
url.host_str().unwrap_or("")
}
fn request_log_port(url: &url::Url) -> u16 {
url.port_or_known_default().unwrap_or(0)
}
fn request_log_path(url: &url::Url) -> &str {
let path = url.path();
if path.is_empty() {
"/"
} else {
path
}
}
fn stream_log_context<'a>(
server: &'a ServerContext,
stream_id: u32,
method: &'a hyper::Method,
url: Option<&'a url::Url>,
redirect_count: usize,
request_body_size: usize,
) -> StreamLogContext<'a> {
StreamLogContext {
server,
stream_id,
method,
url,
redirect_count,
request_body_size,
}
}
fn log_stream_success(ctx: StreamLogContext<'_>, status: u16, duration: Duration) {
let url = ctx
.url
.expect("successful requests should always have a URL");
info!(
server = %ctx.server.server_label,
stream_id = ctx.stream_id,
method = %ctx.method,
scheme = url.scheme(),
host = request_log_host(url),
port = request_log_port(url),
path = request_log_path(url),
query_present = url.query().is_some(),
status,
duration_ms = duration.as_millis() as u64,
redirect_count = ctx.redirect_count,
request_body_bytes = ctx.request_body_size,
"proxy request completed"
);
}
fn log_stream_failure(ctx: StreamLogContext<'_>, error: &str, duration: Duration) {
match ctx.url {
Some(url) => {
warn!(
server = %ctx.server.server_label,
stream_id = ctx.stream_id,
method = %ctx.method,
scheme = url.scheme(),
host = request_log_host(url),
port = request_log_port(url),
path = request_log_path(url),
query_present = url.query().is_some(),
error = %error,
duration_ms = duration.as_millis() as u64,
redirect_count = ctx.redirect_count,
request_body_bytes = ctx.request_body_size,
"proxy request failed"
);
}
None => {
warn!(
server = %ctx.server.server_label,
stream_id = ctx.stream_id,
method = %ctx.method,
error = %error,
duration_ms = duration.as_millis() as u64,
redirect_count = ctx.redirect_count,
request_body_bytes = ctx.request_body_size,
"proxy request failed"
);
}
}
}
impl PreparedRequestBody {
fn streaming(
body_rx: mpsc::Receiver<TunnelFrame>,
@@ -466,6 +570,8 @@ async fn execute_upstream_request(
async fn relay_upstream_response(
server: &ServerContext,
stream_id: u32,
method: &hyper::Method,
request_url: &url::Url,
frame_tx: &FrameSender,
response: hyper::Response<hyper::body::Incoming>,
total_dns_ms: u64,
@@ -516,6 +622,18 @@ async fn relay_upstream_response(
)
.await
{
log_stream_failure(
stream_log_context(
server,
stream_id,
method,
Some(request_url),
redirect_count,
request_body_size.load(Ordering::Relaxed),
),
"tunnel response headers relay failed",
total_elapsed,
);
return Some(total_elapsed);
}
@@ -531,6 +649,18 @@ async fn relay_upstream_response(
)
.await
{
log_stream_failure(
stream_log_context(
server,
stream_id,
method,
Some(request_url),
redirect_count,
request_body_size.load(Ordering::Relaxed),
),
"tunnel response body relay failed",
total_elapsed,
);
return Some(total_elapsed);
}
} else {
@@ -550,6 +680,18 @@ async fn relay_upstream_response(
)
.await
{
log_stream_failure(
stream_log_context(
server,
stream_id,
method,
Some(request_url),
redirect_count,
request_body_size.load(Ordering::Relaxed),
),
"tunnel response body relay failed",
total_elapsed,
);
return Some(total_elapsed);
}
offset = end;
@@ -559,13 +701,26 @@ async fn relay_upstream_response(
Err(error) => {
server.metrics.stream_errors.fetch_add(1, Ordering::Release);
warn!(stream_id, error = %error, "upstream body read error");
let error_message = format!("upstream body read error: {error}");
log_stream_failure(
stream_log_context(
server,
stream_id,
method,
Some(request_url),
redirect_count,
request_body_size.load(Ordering::Relaxed),
),
&error_message,
total_elapsed,
);
send_error(frame_tx, stream_id, &format!("body read error: {error}")).await;
return Some(total_elapsed);
}
}
}
let _ = send_frame(
if !send_frame(
frame_tx,
TunnelFrame::new(
stream_id,
@@ -574,7 +729,22 @@ async fn relay_upstream_response(
Bytes::new(),
),
)
.await;
.await
{
log_stream_failure(
stream_log_context(
server,
stream_id,
method,
Some(request_url),
redirect_count,
request_body_size.load(Ordering::Relaxed),
),
"tunnel stream end relay failed",
total_elapsed,
);
return Some(total_elapsed);
}
debug!(
stream_id,
@@ -582,6 +752,18 @@ async fn relay_upstream_response(
redirects = redirect_count,
"stream completed"
);
log_stream_success(
stream_log_context(
server,
stream_id,
method,
Some(request_url),
redirect_count,
request_body_size.load(Ordering::Relaxed),
),
status,
total_elapsed,
);
Some(total_elapsed)
}
@@ -606,6 +788,8 @@ pub async fn handle_stream(
body_rx: mpsc::Receiver<TunnelFrame>,
frame_tx: FrameSender,
) {
let request_method = parse_request_method(&meta.method);
let request_url = url::Url::parse(&meta.url).ok();
let permit = match state.try_acquire_stream_permit().await {
Ok(permit) => permit,
Err(err) => {
@@ -615,6 +799,18 @@ pub async fn handle_stream(
"proxy admission unavailable"
}
};
log_stream_failure(
stream_log_context(
&server,
stream_id,
&request_method,
request_url.as_ref(),
0,
0,
),
message,
Duration::ZERO,
);
send_error(&frame_tx, stream_id, message).await;
return;
}
@@ -660,9 +856,15 @@ async fn handle_stream_inner(
body_rx: mpsc::Receiver<TunnelFrame>,
frame_tx: &FrameSender,
) -> Option<Duration> {
let mut current_method: hyper::Method = parse_request_method(&meta.method);
let mut current_url = match url::Url::parse(&meta.url) {
Ok(u) => u,
Err(e) => {
log_stream_failure(
stream_log_context(server, stream_id, &current_method, None, 0, 0),
&format!("invalid URL: {e}"),
Duration::ZERO,
);
send_error(frame_tx, stream_id, &format!("invalid URL: {e}")).await;
return None;
}
@@ -672,12 +874,13 @@ async fn handle_stream_inner(
match current_url.scheme() {
"http" | "https" => {}
other => {
send_error(
frame_tx,
stream_id,
&format!("unsupported URL scheme: {other}"),
)
.await;
let error_message = format!("unsupported URL scheme: {other}");
log_stream_failure(
stream_log_context(server, stream_id, &current_method, Some(&current_url), 0, 0),
&error_message,
Duration::ZERO,
);
send_error(frame_tx, stream_id, &error_message).await;
return None;
}
}
@@ -685,7 +888,6 @@ async fn handle_stream_inner(
let deadline = Instant::now()
+ Duration::from_secs(meta.timeout.clamp(MIN_TIMEOUT_SECS, MAX_TIMEOUT_SECS));
let follow_redirects = follow_redirects_enabled(&meta);
let mut current_method: hyper::Method = meta.method.parse().unwrap_or(hyper::Method::GET);
let mut current_headers = sanitize_upstream_headers(&meta.headers);
let timeout = Duration::from_secs(meta.timeout.clamp(MIN_TIMEOUT_SECS, MAX_TIMEOUT_SECS));
let request_body_size = Arc::new(AtomicUsize::new(0));
@@ -700,6 +902,18 @@ async fn handle_stream_inner(
{
Ok(body) => body,
Err(message) => {
log_stream_failure(
stream_log_context(
server,
stream_id,
&current_method,
Some(&current_url),
0,
request_body_size.load(Ordering::Relaxed),
),
&message,
Duration::ZERO,
);
send_error(frame_tx, stream_id, &message).await;
return None;
}
@@ -715,6 +929,18 @@ async fn handle_stream_inner(
loop {
let Some(remaining) = remaining_timeout(deadline) else {
log_stream_failure(
stream_log_context(
server,
stream_id,
&current_method,
Some(&current_url),
redirects_followed,
request_body_size.load(Ordering::Relaxed),
),
"upstream timeout",
overall_start.elapsed(),
);
send_error(frame_tx, stream_id, "upstream timeout").await;
return None;
};
@@ -722,12 +948,20 @@ async fn handle_stream_inner(
Some(mode) => match prepared_body.build_redirect_request_body(mode) {
Some(body) => body,
None => {
send_error(
frame_tx,
stream_id,
"upstream redirect error: body not replayable",
)
.await;
let error_message = "upstream redirect error: body not replayable";
log_stream_failure(
stream_log_context(
server,
stream_id,
&current_method,
Some(&current_url),
redirects_followed,
request_body_size.load(Ordering::Relaxed),
),
error_message,
overall_start.elapsed(),
);
send_error(frame_tx, stream_id, error_message).await;
return None;
}
},
@@ -748,6 +982,18 @@ async fn handle_stream_inner(
{
Ok(context) => context,
Err(message) => {
log_stream_failure(
stream_log_context(
server,
stream_id,
&current_method,
Some(&current_url),
redirects_followed,
request_body_size.load(Ordering::Relaxed),
),
&message,
overall_start.elapsed(),
);
send_error(frame_tx, stream_id, &message).await;
return None;
}
@@ -767,6 +1013,8 @@ async fn handle_stream_inner(
return relay_upstream_response(
server,
stream_id,
&current_method,
&current_url,
frame_tx,
response_ctx.response,
total_dns_ms,
@@ -791,12 +1039,20 @@ async fn handle_stream_inner(
continue;
}
RedirectDecision::Error(message) => {
send_error(
frame_tx,
stream_id,
&format!("upstream redirect error: {message}"),
)
.await;
let error_message = format!("upstream redirect error: {message}");
log_stream_failure(
stream_log_context(
server,
stream_id,
&current_method,
Some(&current_url),
redirects_followed,
request_body_size.load(Ordering::Relaxed),
),
&error_message,
overall_start.elapsed(),
);
send_error(frame_tx, stream_id, &error_message).await;
return None;
}
}
@@ -805,6 +1061,8 @@ async fn handle_stream_inner(
return relay_upstream_response(
server,
stream_id,
&current_method,
&current_url,
frame_tx,
response_ctx.response,
total_dns_ms,