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

Revert "refactor(tunnel): 移除 proxy 连接 idle timeout 机制"

Browse Source 
This reverts commit b004551fe5.
This commit is contained in:
fawney19
2026-04-16 00:53:40 +08:00
parent 5f0fba1807
commit e4be1d6b56
4 changed files with 23 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
apps/aether-gateway/src/tunnel/embedded/hub.rs
View File

@@ -27,6 +27,7 @@ pub enum SendStatus {
#[derive(Debug, Clone, Copy)]
pub struct ConnConfig {
pub ping_interval: Duration,
pub idle_timeout: Duration,
pub outbound_queue_capacity: usize,
}

1
apps/aether-gateway/src/tunnel/embedded/local_relay.rs
View File

@@ -490,6 +490,7 @@ mod tests {
ControlPlaneClient::disabled(),
ConnConfig {
ping_interval: Duration::from_secs(15),
idle_timeout: Duration::from_secs(0),
outbound_queue_capacity: 128,
},
128,

21
apps/aether-gateway/src/tunnel/embedded/proxy_conn.rs
View File

@@ -3,6 +3,7 @@
/// Handles the lifecycle of a single aether-proxy connection:
/// accept -> authenticate (headers) -> read loop -> cleanup
use std::sync::Arc;
use std::time::Duration;
use aether_runtime::bounded_queue;
use axum::extract::ws::{Message, WebSocket};
@@ -107,7 +108,7 @@ pub async fn handle_proxy_connection(
let reader_hub = hub.clone();
let reader_conn = conn.clone();
let reader = tokio::spawn(async move {
run_proxy_reader(ws_rx, reader_hub, reader_conn).await;
run_proxy_reader(ws_rx, reader_hub, reader_conn, cfg.idle_timeout).await;
});
let _ = reader.await;
@@ -115,7 +116,7 @@ pub async fn handle_proxy_connection(
conn.request_close();
hub.unregister_proxy(conn_id, &node_id);
drop(conn);
tokio::time::sleep(std::time::Duration::from_millis(100)).await;
tokio::time::sleep(Duration::from_millis(100)).await;
writer.abort();
let _ = writer.await;
}
@@ -124,11 +125,25 @@ async fn run_proxy_reader(
mut ws_rx: futures_util::stream::SplitStream<WebSocket>,
hub: Arc<HubRouter>,
conn: Arc<ProxyConn>,
idle_timeout: Duration,
) {
let idle_enabled = !idle_timeout.is_zero();
let mut oversized_count = 0u32;
let mut frames_received: u64 = 0;
loop {
let msg = ws_rx.next().await;
let msg = if idle_enabled {
tokio::select! {
msg = ws_rx.next() => msg,
_ = tokio::time::sleep(idle_timeout) => {
warn!(conn_id = conn.id, node_id = %conn.node_id, "proxy idle timeout");
let _ = conn.send(Message::Binary(protocol::encode_goaway().into()));
conn.request_close();
break;
}
}
} else {
ws_rx.next().await
};
match msg {
Some(Ok(Message::Binary(data))) => {

3
apps/aether-gateway/src/tunnel/mod.rs
View File

@@ -47,6 +47,7 @@ pub(crate) const TUNNEL_NODE_STATUS_PATH: &str = "/api/internal/tunnel/node-stat
pub(crate) const TUNNEL_RELAY_PATH_PATTERN: &str = "/api/internal/tunnel/relay/{node_id}";
pub(crate) const TUNNEL_ROUTE_FAMILY: &str = "tunnel_manage";
const DEFAULT_PROXY_IDLE_TIMEOUT_MS: u64 = 2_000;
const DEFAULT_PING_INTERVAL_MS: u64 = 500;
const DEFAULT_MAX_STREAMS: usize = 2048;
const DEFAULT_OUTBOUND_QUEUE_CAPACITY: usize = 128;
@@ -416,6 +417,7 @@ impl EmbeddedTunnelState {
build_embedded_control_plane(Arc::clone(&data), attachment_directory.clone()),
ConnConfig {
ping_interval: Duration::from_millis(DEFAULT_PING_INTERVAL_MS),
idle_timeout: Duration::from_millis(DEFAULT_PROXY_IDLE_TIMEOUT_MS),
outbound_queue_capacity: DEFAULT_OUTBOUND_QUEUE_CAPACITY,
},
DEFAULT_MAX_STREAMS,
@@ -564,6 +566,7 @@ impl Default for EmbeddedTunnelState {
impl fmt::Debug for EmbeddedTunnelState {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.debug_struct("EmbeddedTunnelState")
.field("proxy_idle_timeout_ms", &DEFAULT_PROXY_IDLE_TIMEOUT_MS)
.field("ping_interval_ms", &DEFAULT_PING_INTERVAL_MS)
.field("max_streams", &DEFAULT_MAX_STREAMS)
.field("outbound_queue_capacity", &DEFAULT_OUTBOUND_QUEUE_CAPACITY)