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

feat(hub,stability): bounded outbound queue、worker liveness 检测、事件循环 watchdog 及 DB 操作异步化

Browse Source 
- aether-hub: unbounded channel 改为 bounded channel (BoundedOutbound)，队列满时标记拥塞并主动关闭连接，防止内存无限增长
- aether-hub: worker idle timeout 从命令行参数改为基于心跳的 liveness 检测，默认 60 秒
- aether-hub: 新增 ConnConfig 统一管理连接配置，新增 outbound_queue_capacity 参数
- hub_transport: 新增事件循环 watchdog，检测 lag 超过阈值时临时降级暂停新流
- gunicorn_conf: 启用 faulthandler，worker abort 时自动 dump 全部线程栈用于诊断
- health/endpoint_checker/recording: 同步 DB 操作移至 asyncio.to_thread，避免阻塞事件循环
- Dockerfile: 移除 --worker-idle-timeout 0 命令行参数，改由环境变量和默认值控制
This commit is contained in:
fawney19
2026-03-12 12:19:04 +08:00
parent 4d338ebd3d
commit 71ae1a2307
14 changed files with 676 additions and 280 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
aether-hub/Cargo.lock generated
View File

@@ -10,7 +10,7 @@ checksum = "320119579fcad9c21884f5c4861d16174d0e06250625266f50fe6898340abefa"
[[package]]
name = "aether-hub"
version = "0.1.4"
version = "0.1.5"
dependencies = [
"axum",
"clap",

5
aether-hub/README.md
View File

@@ -28,6 +28,11 @@ cd /path/to/Aether
- `--load`: 加载到本地 Docker（单平台）
- `--latest`: 额外打 `latest` tag
## 运行时参数
- `TUNNEL_HUB_WORKER_IDLE_TIMEOUT`：worker 心跳空闲超时，默认 `60` 秒
- `TUNNEL_HUB_OUTBOUND_QUEUE_CAPACITY`：单连接出站队列容量，默认 `128`；队列打满时会把连接视为拥塞并主动关闭，避免 Hub 内存无限增长
## 与部署脚本关系
- `./deploy.sh`: 本地构建部署（会本地构建 app/base，并在构建 app 时从 GitHub Release 下载 Hub，可用 `--hub-tag` 固定版本）。

162
aether-hub/src/hub.rs
View File

@@ -2,17 +2,89 @@
///
/// Manages proxy connections (node_id -> [ProxyConn]) and worker connections (conn_id -> WorkerConn).
/// Routes frames between workers and proxies with stream_id remapping.
use std::sync::atomic::{AtomicU32, AtomicU64, AtomicUsize, Ordering};
use std::sync::atomic::{AtomicBool, AtomicU32, AtomicU64, AtomicUsize, Ordering};
use std::sync::Arc;
use std::time::Duration;
use axum::extract::ws::Message;
use dashmap::DashMap;
use parking_lot::RwLock;
use tokio::sync::mpsc;
use tokio::sync::mpsc::error::TrySendError;
use tokio::sync::watch;
use tracing::{debug, info, warn};
use crate::protocol;
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum SendStatus {
Queued,
Closed,
Congested,
}
// ---------------------------------------------------------------------------
// Connection configuration (shared by proxy and worker handlers)
// ---------------------------------------------------------------------------
#[derive(Debug, Clone, Copy)]
pub struct ConnConfig {
pub ping_interval: Duration,
pub idle_timeout: Duration,
pub outbound_queue_capacity: usize,
}
// ---------------------------------------------------------------------------
// Bounded outbound channel with congestion-aware close
// ---------------------------------------------------------------------------
pub struct BoundedOutbound {
tx: mpsc::Sender<Message>,
close_tx: watch::Sender<bool>,
closing: AtomicBool,
}
impl BoundedOutbound {
pub fn new(tx: mpsc::Sender<Message>, close_tx: watch::Sender<bool>) -> Self {
Self {
tx,
close_tx,
closing: AtomicBool::new(false),
}
}
pub fn send(&self, msg: Message) -> SendStatus {
if self.is_closing() {
return SendStatus::Closed;
}
match self.tx.try_send(msg) {
Ok(()) => SendStatus::Queued,
Err(TrySendError::Closed(_)) => {
self.mark_closing();
SendStatus::Closed
}
Err(TrySendError::Full(_)) => {
self.mark_closing();
SendStatus::Congested
}
}
}
pub fn is_closing(&self) -> bool {
self.closing.load(Ordering::Acquire)
}
/// Mark as closing. Returns `true` if this call was the first to flip the flag.
pub fn mark_closing(&self) -> bool {
if self.closing.swap(true, Ordering::AcqRel) {
return false;
}
let _ = self.close_tx.send(true);
true
}
}
// ---------------------------------------------------------------------------
// Proxy connection
// ---------------------------------------------------------------------------
@@ -21,7 +93,7 @@ pub struct ProxyConn {
pub id: u64,
pub node_id: String,
pub node_name: String,
pub tx: mpsc::UnboundedSender<Message>,
pub outbound: BoundedOutbound,
next_stream_id: AtomicU32,
pub stream_count: AtomicUsize,
pub max_streams: usize,
@@ -32,14 +104,15 @@ impl ProxyConn {
id: u64,
node_id: String,
node_name: String,
tx: mpsc::UnboundedSender<Message>,
tx: mpsc::Sender<Message>,
close_tx: watch::Sender<bool>,
max_streams: usize,
) -> Self {
Self {
id,
node_id,
node_name,
tx,
outbound: BoundedOutbound::new(tx, close_tx),
next_stream_id: AtomicU32::new(2), // even IDs, start at 2
stream_count: AtomicUsize::new(0),
max_streams,
@@ -51,7 +124,7 @@ impl ProxyConn {
// Reserve one stream slot first (CAS to honor max_streams under contention).
let mut current = self.stream_count.load(Ordering::Relaxed);
loop {
if current >= self.max_streams {
if current >= self.max_streams || !self.is_available() {
return None;
}
match self.stream_count.compare_exchange_weak(
@@ -99,8 +172,27 @@ impl ProxyConn {
}
}
pub fn send(&self, msg: Message) -> bool {
self.tx.send(msg).is_ok()
pub fn is_available(&self) -> bool {
!self.outbound.is_closing()
}
pub fn request_close(&self) {
self.outbound.mark_closing();
}
pub fn send(&self, msg: Message) -> SendStatus {
let was_closing = self.outbound.is_closing();
let status = self.outbound.send(msg);
if status == SendStatus::Congested && !was_closing {
warn!(
conn_id = self.id,
node_id = %self.node_id,
node_name = %self.node_name,
queued_streams = self.stream_count.load(Ordering::Relaxed),
"proxy outbound queue full, closing congested connection"
);
}
status
}
}
@@ -110,16 +202,35 @@ impl ProxyConn {
pub struct WorkerConn {
pub id: u64,
pub tx: mpsc::UnboundedSender<Message>,
pub outbound: BoundedOutbound,
}
impl WorkerConn {
pub fn new(id: u64, tx: mpsc::UnboundedSender<Message>) -> Self {
Self { id, tx }
pub fn new(id: u64, tx: mpsc::Sender<Message>, close_tx: watch::Sender<bool>) -> Self {
Self {
id,
outbound: BoundedOutbound::new(tx, close_tx),
}
}
pub fn send(&self, msg: Message) -> bool {
self.tx.send(msg).is_ok()
pub fn is_available(&self) -> bool {
!self.outbound.is_closing()
}
pub fn request_close(&self) {
self.outbound.mark_closing();
}
pub fn send(&self, msg: Message) -> SendStatus {
let was_closing = self.outbound.is_closing();
let status = self.outbound.send(msg);
if status == SendStatus::Congested && !was_closing {
warn!(
worker_id = self.id,
"worker outbound queue full, closing congested connection"
);
}
status
}
}
@@ -242,6 +353,7 @@ impl HubRouter {
let conns = map.get(node_id)?;
conns
.iter()
.filter(|c| c.is_available())
.min_by_key(|c| c.stream_count.load(Ordering::Relaxed))
.cloned()
}
@@ -404,14 +516,17 @@ impl HubRouter {
},
);
if !proxy_conn.send(Message::Binary(rebuilt_frame.into())) {
// Send failed, clean up mapping
self.worker_to_proxy
.remove(&(worker_conn_id, worker_stream_id));
self.proxy_to_worker
.remove(&(proxy_conn.id, proxy_stream_id));
proxy_conn.release_stream();
return Some("proxy connection send failed".to_string());
match proxy_conn.send(Message::Binary(rebuilt_frame.into())) {
SendStatus::Queued => {}
SendStatus::Closed | SendStatus::Congested => {
// Send failed, clean up mapping
self.worker_to_proxy
.remove(&(worker_conn_id, worker_stream_id));
self.proxy_to_worker
.remove(&(proxy_conn.id, proxy_stream_id));
proxy_conn.release_stream();
return Some("proxy connection congested".to_string());
}
}
None
@@ -558,7 +673,10 @@ impl HubRouter {
let workers: Vec<Arc<WorkerConn>> = self
.worker_conns
.iter()
.map(|e| e.value().clone())
.filter_map(|e| {
let worker = e.value().clone();
worker.is_available().then_some(worker)
})
.collect();
if workers.is_empty() {
debug!("no workers to forward heartbeat to");
@@ -654,7 +772,7 @@ impl HubRouter {
let mut sent = 0usize;
for entry in self.worker_conns.iter() {
if entry.value().send(msg.clone()) {
if matches!(entry.value().send(msg.clone()), SendStatus::Queued) {
sent += 1;
}
}

42
aether-hub/src/main.rs
View File

@@ -14,7 +14,7 @@ use axum::Router;
use clap::Parser;
use tracing::{info, warn};
use crate::hub::HubRouter;
use crate::hub::{ConnConfig, HubRouter};
#[derive(Parser, Debug)]
#[command(name = "aether-hub", about = "Tunnel Hub for Aether")]
@@ -28,7 +28,7 @@ struct Args {
proxy_idle_timeout: u64,
/// Worker-side idle timeout in seconds (0 to disable)
#[arg(long, default_value_t = 0, env = "TUNNEL_HUB_WORKER_IDLE_TIMEOUT")]
#[arg(long, default_value_t = 60, env = "TUNNEL_HUB_WORKER_IDLE_TIMEOUT")]
worker_idle_timeout: u64,
/// Ping interval in seconds (for both sides)
@@ -38,14 +38,21 @@ struct Args {
/// Max concurrent streams per proxy connection
#[arg(long, default_value_t = 2048, env = "TUNNEL_HUB_MAX_STREAMS")]
max_streams: usize,
/// Per-connection outbound queue capacity before treating the socket as congested
#[arg(
long,
default_value_t = 128,
env = "TUNNEL_HUB_OUTBOUND_QUEUE_CAPACITY"
)]
outbound_queue_capacity: usize,
}
#[derive(Clone)]
struct AppState {
hub: Arc<HubRouter>,
proxy_idle_timeout: Duration,
worker_idle_timeout: Duration,
ping_interval: Duration,
proxy_conn_cfg: ConnConfig,
worker_conn_cfg: ConnConfig,
max_streams: usize,
}
@@ -62,11 +69,20 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
let args = Args::parse();
let hub = HubRouter::new();
let outbound_queue_capacity = args.outbound_queue_capacity.clamp(8, 4096);
let ping_interval = Duration::from_secs(args.ping_interval);
let state = AppState {
hub,
proxy_idle_timeout: Duration::from_secs(args.proxy_idle_timeout),
worker_idle_timeout: Duration::from_secs(args.worker_idle_timeout),
ping_interval: Duration::from_secs(args.ping_interval),
proxy_conn_cfg: ConnConfig {
ping_interval,
idle_timeout: Duration::from_secs(args.proxy_idle_timeout),
outbound_queue_capacity,
},
worker_conn_cfg: ConnConfig {
ping_interval,
idle_timeout: Duration::from_secs(args.worker_idle_timeout),
outbound_queue_capacity,
},
max_streams: args.max_streams,
};
@@ -139,8 +155,7 @@ async fn ws_proxy(
node_id,
node_name,
max_streams,
state.ping_interval,
state.proxy_idle_timeout,
state.proxy_conn_cfg,
)
})
.into_response()
@@ -149,11 +164,6 @@ async fn ws_proxy(
async fn ws_worker(ws: WebSocketUpgrade, State(state): State<AppState>) -> impl IntoResponse {
ws.max_frame_size(64 * 1024 * 1024)
.on_upgrade(move |socket| {
worker_conn::handle_worker_connection(
socket,
state.hub,
state.ping_interval,
state.worker_idle_timeout,
)
worker_conn::handle_worker_connection(socket, state.hub, state.worker_conn_cfg)
})
}

82
aether-hub/src/proxy_conn.rs
View File

@@ -7,10 +7,10 @@ use std::time::Duration;
use axum::extract::ws::{Message, WebSocket};
use futures_util::{SinkExt, StreamExt};
use tokio::sync::mpsc;
use tokio::sync::{mpsc, watch};
use tracing::{debug, info, warn};
use crate::hub::{HubRouter, ProxyConn};
use crate::hub::{ConnConfig, HubRouter, ProxyConn, SendStatus};
use crate::protocol;
/// Maximum single frame size: 64 MB
@@ -22,69 +22,71 @@ pub async fn handle_proxy_connection(
node_id: String,
node_name: String,
max_streams: usize,
ping_interval: Duration,
idle_timeout: Duration,
cfg: ConnConfig,
) {
let conn_id = hub.alloc_conn_id();
let (mut ws_tx, ws_rx) = ws.split();
// Create channel for outbound messages
let (tx, mut rx) = mpsc::unbounded_channel::<Message>();
let (tx, mut rx) = mpsc::channel::<Message>(cfg.outbound_queue_capacity);
let (close_tx, mut close_rx) = watch::channel(false);
let conn = Arc::new(ProxyConn::new(
conn_id,
node_id.clone(),
node_name.clone(),
tx,
close_tx,
max_streams,
));
hub.register_proxy(conn.clone());
// Spawn writer task: drains channel -> WebSocket
let writer = tokio::spawn(async move {
while let Some(msg) = rx.recv().await {
if ws_tx.send(msg).await.is_err() {
break;
loop {
tokio::select! {
msg = rx.recv() => match msg {
Some(msg) => {
if ws_tx.send(msg).await.is_err() {
break;
}
}
None => break,
},
changed = close_rx.changed() => {
if changed.is_err() || *close_rx.borrow() {
break;
}
}
}
}
let _ = ws_tx.close().await;
});
// Spawn ping task
let ping_tx = conn.tx.clone();
let ping_conn = conn.clone();
let ping_interval = cfg.ping_interval;
let ping_task = tokio::spawn(async move {
loop {
tokio::time::sleep(ping_interval).await;
let ping = protocol::encode_ping();
if ping_tx.send(Message::Binary(ping.into())).is_err() {
if !matches!(
ping_conn.send(Message::Binary(ping.into())),
SendStatus::Queued
) {
break;
}
}
});
// Spawn reader task
let reader_hub = hub.clone();
let reader_node_id = node_id.clone();
let reader_tx = conn.tx.clone();
let reader_conn = conn.clone();
let reader = tokio::spawn(async move {
run_proxy_reader(
ws_rx,
reader_hub,
conn_id,
reader_node_id,
reader_tx,
idle_timeout,
)
.await;
run_proxy_reader(ws_rx, reader_hub, reader_conn, cfg.idle_timeout).await;
});
// Wait for reader to end, then cleanup.
let _ = reader.await;
ping_task.abort();
conn.request_close();
hub.unregister_proxy(conn_id, &node_id);
// conn still holds an Arc<ProxyConn> with a channel sender clone.
// Drop it so the writer can drain and exit.
drop(conn);
tokio::time::sleep(Duration::from_millis(100)).await;
writer.abort();
@@ -94,9 +96,7 @@ pub async fn handle_proxy_connection(
async fn run_proxy_reader(
mut ws_rx: futures_util::stream::SplitStream<WebSocket>,
hub: Arc<HubRouter>,
conn_id: u64,
node_id: String,
tx: mpsc::UnboundedSender<Message>,
conn: Arc<ProxyConn>,
idle_timeout: Duration,
) {
let idle_enabled = !idle_timeout.is_zero();
@@ -106,8 +106,9 @@ async fn run_proxy_reader(
tokio::select! {
msg = ws_rx.next() => msg,
_ = tokio::time::sleep(idle_timeout) => {
warn!(conn_id = conn_id, node_id = %node_id, "proxy idle timeout");
let _ = tx.send(Message::Binary(protocol::encode_goaway().into()));
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;
}
}
@@ -121,12 +122,13 @@ async fn run_proxy_reader(
if data.len() > MAX_FRAME_SIZE {
oversized_count += 1;
warn!(
conn_id = conn_id,
conn_id = conn.id,
size = data.len(),
"oversized frame from proxy"
);
if oversized_count >= 5 {
warn!(conn_id = conn_id, "too many oversized frames, closing");
warn!(conn_id = conn.id, "too many oversized frames, closing");
conn.request_close();
break;
}
continue;
@@ -134,21 +136,21 @@ async fn run_proxy_reader(
oversized_count = 0;
if data.len() < protocol::HEADER_SIZE {
debug!(conn_id = conn_id, "frame too small, skipping");
debug!(conn_id = conn.id, "frame too small, skipping");
continue;
}
hub.handle_proxy_frame(conn_id, &mut data);
hub.handle_proxy_frame(conn.id, &mut data);
}
Some(Ok(Message::Close(_))) | None => {
info!(conn_id = conn_id, node_id = %node_id, "proxy WebSocket closed");
info!(conn_id = conn.id, node_id = %conn.node_id, "proxy WebSocket closed");
break;
}
Some(Err(e)) => {
warn!(conn_id = conn_id, error = %e, "proxy WebSocket error");
warn!(conn_id = conn.id, error = %e, "proxy WebSocket error");
break;
}
_ => {} // Ignore text/ping/pong at WS level
_ => {}
}
}
}

169
aether-hub/src/worker_conn.rs
View File

@@ -2,76 +2,92 @@
///
/// Handles the lifecycle of a single Gunicorn worker connection:
/// accept -> read loop (route frames via Hub) -> cleanup
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::Arc;
use std::time::Duration;
use std::time::{Duration, Instant};
use axum::extract::ws::{Message, WebSocket};
use futures_util::{SinkExt, StreamExt};
use tokio::sync::mpsc;
use tokio::sync::{mpsc, watch};
use tracing::{debug, info, warn};
use crate::hub::{HubRouter, WorkerConn};
use crate::hub::{ConnConfig, HubRouter, SendStatus, WorkerConn};
use crate::protocol;
pub async fn handle_worker_connection(
ws: WebSocket,
hub: Arc<HubRouter>,
ping_interval: Duration,
idle_timeout: Duration,
) {
pub async fn handle_worker_connection(ws: WebSocket, hub: Arc<HubRouter>, cfg: ConnConfig) {
let conn_id = hub.alloc_conn_id();
let (mut ws_tx, ws_rx) = ws.split();
// Create channel for outbound messages
let (tx, mut rx) = mpsc::unbounded_channel::<Message>();
let (tx, mut rx) = mpsc::channel::<Message>(cfg.outbound_queue_capacity);
let (close_tx, mut close_rx) = watch::channel(false);
let conn = Arc::new(WorkerConn::new(conn_id, tx));
let conn = Arc::new(WorkerConn::new(conn_id, tx, close_tx));
hub.register_worker(conn.clone());
// Spawn writer task
let writer = tokio::spawn(async move {
while let Some(msg) = rx.recv().await {
if ws_tx.send(msg).await.is_err() {
break;
loop {
tokio::select! {
msg = rx.recv() => match msg {
Some(msg) => {
if ws_tx.send(msg).await.is_err() {
break;
}
}
None => break,
},
changed = close_rx.changed() => {
if changed.is_err() || *close_rx.borrow() {
break;
}
}
}
}
let _ = ws_tx.close().await;
});
// Spawn ping task
let ping_tx = conn.tx.clone();
let ping_task = tokio::spawn(async move {
loop {
tokio::time::sleep(ping_interval).await;
let ping = protocol::encode_ping();
if ping_tx.send(Message::Binary(ping.into())).is_err() {
break;
}
}
});
let liveness_clock = Instant::now();
let last_seen_ms = Arc::new(AtomicU64::new(0));
// Spawn reader task
let reader_hub = hub.clone();
let reader_tx = conn.tx.clone();
let reader = tokio::spawn(async move {
let reader_conn = conn.clone();
let reader_last_seen_ms = last_seen_ms.clone();
let liveness_conn = conn.clone();
let mut reader = tokio::spawn(async move {
run_worker_reader(
ws_rx,
reader_hub,
conn_id,
conn.clone(),
reader_tx,
idle_timeout,
reader_conn,
reader_last_seen_ms,
liveness_clock,
)
.await;
});
// Wait for reader to end, then cleanup.
let liveness_last_seen_ms = last_seen_ms.clone();
let mut liveness = tokio::spawn(async move {
run_worker_liveness(
conn_id,
liveness_conn,
cfg.ping_interval,
cfg.idle_timeout,
liveness_last_seen_ms,
liveness_clock,
)
.await;
});
tokio::select! {
_ = &mut reader => {}
_ = &mut liveness => {}
}
conn.request_close();
reader.abort();
liveness.abort();
let _ = reader.await;
ping_task.abort();
// Unregister first so all channel senders are dropped (reader_tx dropped
// when reader completes, ping_tx dropped by abort, conn.tx dropped when
// the last Arc<WorkerConn> is removed from hub). This lets the writer
// drain buffered messages (e.g. GOAWAY) before we force-abort it.
let _ = liveness.await;
hub.unregister_worker(conn_id);
tokio::time::sleep(Duration::from_millis(100)).await;
writer.abort();
@@ -83,26 +99,14 @@ async fn run_worker_reader(
hub: Arc<HubRouter>,
conn_id: u64,
conn: Arc<WorkerConn>,
tx: mpsc::UnboundedSender<Message>,
idle_timeout: Duration,
last_seen_ms: Arc<AtomicU64>,
liveness_clock: Instant,
) {
let idle_enabled = !idle_timeout.is_zero();
loop {
let msg = if idle_enabled {
tokio::select! {
msg = ws_rx.next() => msg,
_ = tokio::time::sleep(idle_timeout) => {
warn!(worker_id = conn_id, "worker idle timeout");
let _ = tx.send(Message::Binary(protocol::encode_goaway().into()));
break;
}
}
} else {
ws_rx.next().await
};
match msg {
match ws_rx.next().await {
Some(Ok(Message::Binary(data))) => {
last_seen_ms.store(elapsed_millis(liveness_clock), Ordering::Relaxed);
let mut data = data.to_vec();
if data.len() < protocol::HEADER_SIZE {
debug!(worker_id = conn_id, "frame too small, skipping");
@@ -114,15 +118,12 @@ async fn run_worker_reader(
None => continue,
};
// HEARTBEAT_ACK from worker -> route back to proxy
if header.msg_type == protocol::HEARTBEAT_ACK {
hub.handle_worker_heartbeat_ack(&mut data);
continue;
}
// Regular frames: route via hub
if let Some(err_msg) = hub.handle_worker_frame(conn_id, &mut data) {
// Send STREAM_ERROR back to worker
let err_frame = protocol::encode_stream_error(header.stream_id, &err_msg);
let _ = conn.send(Message::Binary(err_frame.into()));
}
@@ -135,7 +136,55 @@ async fn run_worker_reader(
warn!(worker_id = conn_id, error = %e, "worker WebSocket error");
break;
}
_ => {} // Ignore text/ping/pong at WS level
_ => {}
}
}
}
async fn run_worker_liveness(
conn_id: u64,
conn: Arc<WorkerConn>,
ping_interval: Duration,
idle_timeout: Duration,
last_seen_ms: Arc<AtomicU64>,
liveness_clock: Instant,
) {
let ping_interval_ms = ping_interval.as_millis().max(1) as u64;
let idle_timeout_ms = idle_timeout.as_millis() as u64;
loop {
tokio::time::sleep(ping_interval).await;
let ping = protocol::encode_ping();
if !matches!(conn.send(Message::Binary(ping.into())), SendStatus::Queued) {
break;
}
if idle_timeout.is_zero() {
continue;
}
let now_ms = elapsed_millis(liveness_clock);
let last_seen = last_seen_ms.load(Ordering::Relaxed);
let silent_for_ms = now_ms.saturating_sub(last_seen);
if silent_for_ms < idle_timeout_ms {
continue;
}
let missed_heartbeats = (silent_for_ms / ping_interval_ms).max(1);
warn!(
worker_id = conn_id,
idle_timeout_secs = idle_timeout.as_secs(),
silent_for_ms = silent_for_ms,
missed_heartbeats = missed_heartbeats,
"worker heartbeat timeout"
);
let _ = conn.send(Message::Binary(protocol::encode_goaway().into()));
conn.request_close();
break;
}
}
fn elapsed_millis(started_at: Instant) -> u64 {
started_at.elapsed().as_millis().min(u64::MAX as u128) as u64
}