[bugfix] Prevent typed unix socket client from looping forever on fai…

…lure to connect (#796)

The implementation had double retry loop bug, resulting in the timeout
being irrelevant. Remove the outer retry loop and actually time out
after 30 seconds.

While we're at it, tune the backoff to be more reasonable (less spammy
-- now retries quickly initially, then backs off significantly and slows
down to 1 rps).

plane/src/typed_unix_socket/client.rs

@@ -1,5 +1,5 @@
-use super::{get_quick_backoff, WrappedMessage};
-use crate::util::{random_token, GuardHandle};
+use super::WrappedMessage;
+use crate::util::{random_token, ExponentialBackoff, GuardHandle};
 use anyhow::{Error, Result};
 use dashmap::DashMap;
 use serde::{Deserialize, Serialize};
@@ -47,14 +47,11 @@ where
             let response_map = Arc::clone(&response_map);
             let event_tx = event_tx.clone();
             GuardHandle::new(async move {
-                let mut backoff = get_quick_backoff();
                 loop {
                     let Ok(stream) = timeout(CONNECT_TIMEOUT, connect(&socket_path)).await else {
-                        tracing::error!("Timeout connecting to server");
-                        backoff.wait().await;
-                        continue;
+                        tracing::error!("Timeout connecting to server; shutting down");
+                        break;
                     };
-                    backoff.reset();
                     if handle_connection(stream, rx, Arc::clone(&response_map), event_tx.clone())
                         .await
                         .is_ok()
@@ -109,7 +106,12 @@ where
 
 async fn connect<P: AsRef<Path>>(socket_path: P) -> UnixStream {
     let socket_path = socket_path.as_ref().to_path_buf();
-    let mut backoff = get_quick_backoff();
+    let mut backoff = ExponentialBackoff::new(
+        chrono::Duration::milliseconds(10),
+        chrono::Duration::seconds(1),
+        1.5,
+        chrono::Duration::seconds(1),
+    );
     loop {
         match UnixStream::connect(&socket_path).await {
             Ok(stream) => return stream,

plane/src/typed_unix_socket/mod.rs

@@ -1,5 +1,3 @@
-use crate::util::ExponentialBackoff;
-use chrono::Duration;
 use serde::{Deserialize, Serialize};
 use std::fs;
 use std::path::PathBuf;
@@ -15,15 +13,6 @@ pub struct WrappedMessage<T> {
     pub message: T,
 }
 
-fn get_quick_backoff() -> ExponentialBackoff {
-    ExponentialBackoff::new(
-        Duration::milliseconds(10),
-        Duration::milliseconds(100),
-        1.1,
-        Duration::milliseconds(100),
-    )
-}
-
 pub struct SocketPath(pub PathBuf);
 
 impl Drop for SocketPath {