WIP cleanup: split branch stash into 8 focused commits #2

Merged
osobh merged 8 commits from chore/wip-followups into main 2026-07-06 01:52:14 +00:00
Showing only changes of commit 51d501e36e - Show all commits
+38 -4
View File
@@ -134,13 +134,38 @@ async fn run(ws_url: &str) -> Result<(), Box<dyn std::error::Error>> {
std::thread::sleep(Duration::from_secs(900)); std::thread::sleep(Duration::from_secs(900));
}); });
// Liveness: the server pings every 15s. If nothing inbound arrives for 40s // Liveness has two independent failure modes to catch, both of which we
// the socket is dead — return so main() reconnects. // hit on architect during the Jul 5 2026 outage:
// (a) READ-side stall: server never sends anything (or the socket goes
// half-open on read). Caught by last_rx / 40s idle window below.
// (b) WRITE-side stall: peer's TCP stack has died but our OS buffer is
// still soaking heartbeat writes. write.send().await blocks
// indefinitely inside the select! branch — tokio::select doesn't
// preempt a running future, so the whole loop freezes; idle_tick
// never gets to fire. Wrapping the send in a timeout is the fix.
//
// WRITE_DEADLINE is short enough (10s) that a stuck send is caught before
// it can outlast the 40s read-idle threshold and leave the daemon spinning
// silently for hours (which is what happened pre-patch).
const WRITE_DEADLINE: Duration = Duration::from_secs(10);
let mut idle_tick = tokio::time::interval(Duration::from_secs(5)); let mut idle_tick = tokio::time::interval(Duration::from_secs(5));
let mut last_rx = std::time::Instant::now(); let mut last_rx = std::time::Instant::now();
loop { loop {
tokio::select! { tokio::select! {
Some(frame) = out_rx.recv() => { write.send(Message::Text(frame.into())).await?; } Some(frame) = out_rx.recv() => {
match tokio::time::timeout(
WRITE_DEADLINE,
write.send(Message::Text(frame.into())),
).await {
Ok(Ok(())) => {}
Ok(Err(e)) => return Err(e.into()),
Err(_) => {
eprintln!("write.send timeout > {}s — socket dead, reconnecting",
WRITE_DEADLINE.as_secs());
return Ok(());
}
}
}
_ = idle_tick.tick() => { _ = idle_tick.tick() => {
if last_rx.elapsed() > Duration::from_secs(40) { if last_rx.elapsed() > Duration::from_secs(40) {
return Ok(()); return Ok(());
@@ -158,7 +183,16 @@ async fn run(ws_url: &str) -> Result<(), Box<dyn std::error::Error>> {
let text = t.to_string(); let text = t.to_string();
tokio::spawn(async move { handle_frame(&text, &out, &ptys, &peers).await; }); tokio::spawn(async move { handle_frame(&text, &out, &ptys, &peers).await; });
} }
Some(Ok(Message::Ping(p))) => write.send(Message::Pong(p)).await?, Some(Ok(Message::Ping(p))) => {
match tokio::time::timeout(WRITE_DEADLINE, write.send(Message::Pong(p))).await {
Ok(Ok(())) => {}
Ok(Err(e)) => return Err(e.into()),
Err(_) => {
eprintln!("pong write timeout — socket dead, reconnecting");
return Ok(());
}
}
}
Some(Ok(Message::Close(_))) | None => return Ok(()), Some(Ok(Message::Close(_))) | None => return Ok(()),
Some(Err(e)) => return Err(e.into()), Some(Err(e)) => return Err(e.into()),
_ => {} _ => {}