From f88103a510d6b6edd68901b0eabcdbea6c542664 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Iwo=20Strzebo=C5=84ski?= Date: Sun, 18 Oct 2026 20:03:49 +0000 Subject: [PATCH] Back the uplink reconnect off from 5s to 60s A gateway cannot tell a cloud that is briefly restarting from a token the cloud will never accept: with the edge token on the upgrade request, both are just a failed handshake here. So a flat 5s retry meant a permanently misconfigured gateway knocking twelve times a minute, forever - pointless, and the shape of the thing the other end would eventually have to rate-limit. Doubles to a 60s ceiling, which is still short enough that an outage resolves without anyone touching the gateway. That matters when the gateway is in someone else's workshop. The reset is on a completed handshake, which is why `connect` is split out of what is now `serve`: a handshake is the only unambiguous evidence that the cloud is reachable and this gateway is accepted. Resetting after the connection ended instead could not tell a connection that lived an hour from one that never opened. Re-verified live against the backend and both printers with continuum-backend's scripts/e2e-edge.sh. --- src/uplink/mod.rs | 59 +++++++++++++++++++++++++++++++++++++++-------- 1 file changed, 50 insertions(+), 9 deletions(-) diff --git a/src/uplink/mod.rs b/src/uplink/mod.rs index e036518..20a786e 100644 --- a/src/uplink/mod.rs +++ b/src/uplink/mod.rs @@ -43,7 +43,25 @@ use tracing::{info, warn}; use crate::config::Config; use crate::printer::{GenericPrinter, PrinterHandle}; +/// The websocket, once connected. Named because `connect` has to hand it to +/// `serve` and the full type is three layers of generics. +type WsStream = tokio_tungstenite::WebSocketStream< + tokio_tungstenite::MaybeTlsStream, +>; + +/// First retry delay after a failed connection, doubling up to +/// `RETRY_DELAY_MAX`. const RETRY_DELAY: Duration = Duration::from_secs(5); +/// Ceiling for the backoff. +/// +/// A gateway cannot tell a cloud that is briefly restarting from a token the +/// cloud will never accept - both are a failed connection here - so a flat +/// 5s retry meant a permanently misconfigured gateway knocking twelve times a +/// minute, forever, which is both pointless and the shape of something worth +/// rate-limiting at the other end. A minute is short enough that an outage +/// resolves without anyone touching the gateway, which matters when the +/// gateway is in someone else's workshop. +const RETRY_DELAY_MAX: Duration = Duration::from_secs(60); /// Depth of the out-of-band outbound queue. /// /// Bounded, because an unbounded queue facing a cloud that may be down for @@ -107,24 +125,36 @@ pub async fn run(config: Config, fleet: SharedFleet) { // sender is kept for the whole program's life, which is also what stops // `recv()` below from ever returning `None`. let (outbound, mut inbox) = tokio::sync::mpsc::channel::(OUTBOUND_CAPACITY); + let mut delay = RETRY_DELAY; loop { info!(url = %config.uplink_url, "connecting to cloud uplink"); - if let Err(err) = connect_and_serve(&config, &fleet, &outbound, &mut inbox).await { - warn!(?err, "uplink connection failed, retrying in {RETRY_DELAY:?}"); + match connect(&config).await { + Ok(ws_stream) => { + // Reset on a completed handshake, which is the only + // unambiguous evidence that the cloud is reachable and this + // gateway's token is accepted - and the reason `connect` is + // split out at all. Resetting after `serve` returns instead + // would not distinguish a connection that lived for an hour + // from one that failed to open. + delay = RETRY_DELAY; + if let Err(err) = serve(ws_stream, &config, &fleet, &outbound, &mut inbox).await { + warn!(?err, "uplink connection dropped, reconnecting in {delay:?}"); + } + } + Err(err) => warn!(?err, "uplink connection failed, retrying in {delay:?}"), } - tokio::time::sleep(RETRY_DELAY).await; + tokio::time::sleep(delay).await; + delay = (delay * 2).min(RETRY_DELAY_MAX); } } -async fn connect_and_serve( - config: &Config, - fleet: &SharedFleet, - outbound: &tokio::sync::mpsc::Sender, - inbox: &mut tokio::sync::mpsc::Receiver, -) -> anyhow::Result<()> { +/// Opens the websocket. Separate from `serve` so `run` can tell a failed +/// handshake from a connection that opened and later dropped; they want +/// different retry behaviour. +async fn connect(config: &Config) -> anyhow::Result { // The edge token goes on the upgrade request, not in the first message. // The backend verifies it before the socket exists, so a bad token is an // HTTP 401 here rather than a connection that opens and is then closed - @@ -145,6 +175,17 @@ async fn connect_and_serve( .body(())?; let (ws_stream, _response) = tokio_tungstenite::connect_async(request).await?; + Ok(ws_stream) +} + +/// Says hello, then pumps the connection until it drops. +async fn serve( + ws_stream: WsStream, + config: &Config, + fleet: &SharedFleet, + outbound: &tokio::sync::mpsc::Sender, + inbox: &mut tokio::sync::mpsc::Receiver, +) -> anyhow::Result<()> { let (mut write, mut read) = ws_stream.split(); let hello = ClientMessage::Hello {