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

fix(proxy-node): 监控 writer task 退出以避免 tunnel 连接半关闭时 dispatcher 阻塞

Browse Source 
当对端关闭连接导致 write half 退出但 read half 仍打开时，
dispatcher 会在 ws_stream.next() 上永久阻塞。通过在 tokio::select!
中监控 writer_handle，检测到 writer 退出后立即触发重连。
This commit is contained in:
fawney19
2026-02-26 13:07:42 +08:00
parent 67b8eb5c2f
commit 54332d31bc
1 changed files with 14 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

17
aether-proxy/src/tunnel/client.rs
View File

@@ -55,7 +55,7 @@ pub async fn connect_and_run(
let (ws_sink, ws_read) = futures_util::StreamExt::split(ws_stream); let (ws_sink, ws_read) = futures_util::StreamExt::split(ws_stream);
// Spawn writer task // Spawn writer task
let (frame_tx, writer_handle) = writer::spawn_writer(ws_sink); let (frame_tx, mut writer_handle) = writer::spawn_writer(ws_sink);
// Spawn heartbeat task // Spawn heartbeat task
let hb_handle = heartbeat::spawn( let hb_handle = heartbeat::spawn(
@@ -65,7 +65,11 @@ pub async fn connect_and_run(
shutdown.clone(), shutdown.clone(),
); );
// Run dispatcher (blocks until disconnect or shutdown) // Run dispatcher (blocks until disconnect or shutdown).
// Also watch for writer exit — if the write half dies (e.g. the peer
// closed the connection) but the read half stays open, dispatcher would
// block forever on `ws_stream.next()`. Monitoring `writer_handle`
// ensures we detect this and trigger a reconnect promptly.
let state_clone = Arc::clone(state); let state_clone = Arc::clone(state);
let server_clone = Arc::clone(server); let server_clone = Arc::clone(server);
let outcome = tokio::select! { let outcome = tokio::select! {
@@ -75,6 +79,10 @@ pub async fn connect_and_run(
Err(e) => return Err(e), Err(e) => return Err(e),
} }
} }
_ = &mut writer_handle => {
debug!("writer task exited, triggering reconnect");
TunnelOutcome::Disconnected
}
_ = shutdown.changed() => { _ = shutdown.changed() => {
debug!("shutdown during tunnel dispatch"); debug!("shutdown during tunnel dispatch");
TunnelOutcome::Shutdown TunnelOutcome::Shutdown
@@ -88,7 +96,10 @@ pub async fn connect_and_run(
// Wait for the writer task to finish with a generous timeout — the // Wait for the writer task to finish with a generous timeout — the
// dispatcher already waits up to 30s for stream handlers, so 35s here // dispatcher already waits up to 30s for stream handlers, so 35s here
// covers that plus a small margin. // covers that plus a small margin.
let _ = tokio::time::timeout(Duration::from_secs(35), writer_handle).await; // Skip if the writer already exited (the select branch that fired).
if !writer_handle.is_finished() {
let _ = tokio::time::timeout(Duration::from_secs(35), writer_handle).await;
}
info!("tunnel disconnected"); info!("tunnel disconnected");
Ok(outcome) Ok(outcome)