Skip to content
Merged
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
24 changes: 23 additions & 1 deletion sandd/src/main.rs
Original file line number Diff line number Diff line change
Expand Up @@ -324,17 +324,30 @@ where
.context("Failed to initialize snapshot manager")?,
);

// Spawn heartbeat task
// Spawn heartbeat task. A failed heartbeat send is our ONLY reliable signal
// that the connection is dead: over a DERP-relayed mesh, a controller that
// vanishes (e.g. pod restart) often produces no TCP FIN/RST on the daemon side,
// so `ws_rx.next()` in the loop below blocks forever and never surfaces the
// drop. The heartbeat write, by contrast, fails. So on send failure we trip
// `dead_tx`, which the select! polls to break the serve loop and let main()
// reconnect — without this the daemon wedges half-open until the pod is deleted.
let ws_tx_clone = Arc::new(tokio::sync::Mutex::new(ws_tx));
let ws_tx_heartbeat = ws_tx_clone.clone();
let (dead_tx, dead_rx) = tokio::sync::oneshot::channel::<()>();
let heartbeat_handle = tokio::spawn(async move {
let mut interval = tokio::time::interval(Duration::from_secs(heartbeat_interval));
let mut dead_tx = Some(dead_tx);
loop {
interval.tick().await;
let heartbeat = Message::Heartbeat;
if let Ok(json) = serde_json::to_string(&heartbeat) {
let mut tx = ws_tx_heartbeat.lock().await;
if tx.send(WsMessage::Text(json)).await.is_err() {
// Signal the serve loop that the connection is dead so it
// reconnects instead of blocking forever on a half-open read.
if let Some(d) = dead_tx.take() {
let _ = d.send(());
}
break;
}
}
Expand All @@ -350,6 +363,7 @@ where
// Pin the shutdown future once so it can be polled across loop iterations
// without being moved (it may be `!Unpin`).
tokio::pin!(shutdown);
tokio::pin!(dead_rx);
let outcome = loop {
tokio::select! {
// Poll shutdown FIRST. With `biased`, tokio checks branches top to
Expand All @@ -371,6 +385,14 @@ where
break ServeOutcome::Shutdown;
}

// Heartbeat send failed => connection is dead. Reconnect. (The Err
// arm — heartbeat task gone without signalling — is treated the same:
// no live heartbeat means no live connection.)
_ = &mut dead_rx => {
warn!("Heartbeat send failed, connection is dead; reconnecting");
break ServeOutcome::Disconnected;
}

msg = ws_rx.next() => {
let msg = match msg {
Some(Ok(WsMessage::Text(text))) => text,
Expand Down
Loading