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.
This commit is contained in:
+50
-9
@@ -43,7 +43,25 @@ use tracing::{info, warn};
|
|||||||
use crate::config::Config;
|
use crate::config::Config;
|
||||||
use crate::printer::{GenericPrinter, PrinterHandle};
|
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<tokio::net::TcpStream>,
|
||||||
|
>;
|
||||||
|
|
||||||
|
/// First retry delay after a failed connection, doubling up to
|
||||||
|
/// `RETRY_DELAY_MAX`.
|
||||||
const RETRY_DELAY: Duration = Duration::from_secs(5);
|
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.
|
/// Depth of the out-of-band outbound queue.
|
||||||
///
|
///
|
||||||
/// Bounded, because an unbounded queue facing a cloud that may be down for
|
/// 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
|
// sender is kept for the whole program's life, which is also what stops
|
||||||
// `recv()` below from ever returning `None`.
|
// `recv()` below from ever returning `None`.
|
||||||
let (outbound, mut inbox) = tokio::sync::mpsc::channel::<ClientMessage>(OUTBOUND_CAPACITY);
|
let (outbound, mut inbox) = tokio::sync::mpsc::channel::<ClientMessage>(OUTBOUND_CAPACITY);
|
||||||
|
let mut delay = RETRY_DELAY;
|
||||||
|
|
||||||
loop {
|
loop {
|
||||||
info!(url = %config.uplink_url, "connecting to cloud uplink");
|
info!(url = %config.uplink_url, "connecting to cloud uplink");
|
||||||
|
|
||||||
if let Err(err) = connect_and_serve(&config, &fleet, &outbound, &mut inbox).await {
|
match connect(&config).await {
|
||||||
warn!(?err, "uplink connection failed, retrying in {RETRY_DELAY:?}");
|
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(
|
/// Opens the websocket. Separate from `serve` so `run` can tell a failed
|
||||||
config: &Config,
|
/// handshake from a connection that opened and later dropped; they want
|
||||||
fleet: &SharedFleet,
|
/// different retry behaviour.
|
||||||
outbound: &tokio::sync::mpsc::Sender<ClientMessage>,
|
async fn connect(config: &Config) -> anyhow::Result<WsStream> {
|
||||||
inbox: &mut tokio::sync::mpsc::Receiver<ClientMessage>,
|
|
||||||
) -> anyhow::Result<()> {
|
|
||||||
// The edge token goes on the upgrade request, not in the first message.
|
// 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
|
// 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 -
|
// HTTP 401 here rather than a connection that opens and is then closed -
|
||||||
@@ -145,6 +175,17 @@ async fn connect_and_serve(
|
|||||||
.body(())?;
|
.body(())?;
|
||||||
|
|
||||||
let (ws_stream, _response) = tokio_tungstenite::connect_async(request).await?;
|
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<ClientMessage>,
|
||||||
|
inbox: &mut tokio::sync::mpsc::Receiver<ClientMessage>,
|
||||||
|
) -> anyhow::Result<()> {
|
||||||
let (mut write, mut read) = ws_stream.split();
|
let (mut write, mut read) = ws_stream.split();
|
||||||
|
|
||||||
let hello = ClientMessage::Hello {
|
let hello = ClientMessage::Hello {
|
||||||
|
|||||||
Reference in New Issue
Block a user