fix(daemon): reconnect after the server session drops

A long-running daemon that lost its TCP connection to the server reported
session_loss "disconnected" and answered every command from the dead session
until a manual restart. Two causes:

- the daemon owned one session for its lifetime with no reconnect at all; and
- the lib's ServerActor queued a Login sent while disconnected instead of
  dialing, so even asking it to log in again never re-established a connection.

Reconnect on a dedicated thread with 1s-60s capped backoff, only for a dropped
connection to a session this process owns (never a displaced or borrowed one,
which would start a login war or fight another daemon). Let a Login through the
disconnected guard so a parked actor dials again, and bound that dial with
connect_timeout so a blackholed route cannot park the actor thread. Reset the
message reader, the queued backlog and the login verdict on disconnect so a
partial frame or stale Some(false) cannot wedge the reconnected session.
Announce the drop once per outage and re-join rooms and re-watch users on
recovery; the daemon main loop is back to save-and-stop so `daemon stop` stays
responsive during an outage. Update the session_loss event summary, which no
longer means the session is gone for good.

soulseek-rs/docs/openrpc.json

@@ -2193,7 +2193,7 @@
           "type": "null"
         }
       },
-      "summary": "The server session ended; nothing seen after this means anything.",
+      "summary": "The server session was lost. A disconnected session is reconnected on its own and events resume; a displaced one is terminal. Read daemon.status for the live state.",
       "x-notification": true
     }
   ],

soulseek-rs/src/api.rs

@@ -60,6 +60,9 @@ pub trait SessionApi: Send + Sync {
     }
     fn listen_port(&self) -> Option<u16>;
     fn session_loss(&self) -> Option<SessionLoss>;
+    fn relogin(&self) -> Result<bool> {
+        Err(soulseek_rs::SoulseekRs::NotConnected)
+    }
 
     /// Search and wait out the window. Every implementation is
     /// `search_with_cancel` with nothing to cancel on.
@@ -191,6 +194,10 @@ impl SessionApi for Client {
         Self::session_loss(self)
     }
 
+    fn relogin(&self) -> Result<bool> {
+        Self::login(self)
+    }
+
     fn search_with_cancel(
         &self,
         query: &str,

soulseek-rs/src/daemon/hub.rs

@@ -15,7 +15,7 @@ use super::proto::{
     SessionLossEvent, UploadInfoDto, UploadStatusDto, UserMessageDto,
 };
 use crate::api::SessionApi;
-use soulseek_rs::DownloadStatus;
+use soulseek_rs::{DownloadStatus, SessionLoss};
 use std::collections::HashMap;
 use std::sync::atomic::{AtomicU64, Ordering};
 use std::sync::mpsc::{Receiver, SyncSender, TrySendError, sync_channel};
@@ -175,9 +175,7 @@ struct DrainState {
     /// Last upload state reported per (user, file), so an upload that has not
     /// moved is not re-announced every tick.
     uploads: HashMap<(String, String), UploadInfoDto>,
-    /// Whether the session-loss event has already gone out. It is terminal, so
-    /// it is worth saying exactly once.
-    reported_loss: bool,
+    last_loss: Option<SessionLoss>,
 }
 
 /// How long to keep asking whether a peer has answered a browse request.
@@ -273,10 +271,12 @@ fn tick(
     publish_browses(session, hub, browses);
     publish_downloads(hub, downloads);
 
-    if !state.reported_loss
-        && let Some(loss) = session.session_loss()
-    {
-        state.reported_loss = true;
+    let loss = session.session_loss();
+    if loss == state.last_loss {
+        return;
+    }
+    state.last_loss = loss;
+    if let Some(loss) = loss {
         hub.publish(
             Event::SessionLoss,
             serde_json::to_value(SessionLossEvent { loss: loss.into() })

soulseek-rs/src/daemon/mod.rs

@@ -24,11 +24,12 @@ pub mod server;
 
 use crate::api::SessionApi;
 use crate::commands::{Ctx, Session};
-use crate::output::{CliError, CliResult};
+use crate::output::{CliError, CliResult, Out};
 use crate::persist::state::{PersistedDownload, PersistedMessage, StateStore};
 use hub::{Hub, PendingBrowses, PendingDownload};
 use proto::ChatMessageDto;
 use server::Daemon;
+use soulseek_rs::SessionLoss;
 use std::path::{Path, PathBuf};
 use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::{Arc, Mutex};
@@ -44,6 +45,8 @@ pub const TOKEN_NAME: &str = "daemon.token";
 /// loses little, rare enough not to churn the disk on a busy queue.
 const SAVE_INTERVAL: Duration = Duration::from_secs(5);
 
+const RECONNECT_BACKOFF_CAP: Duration = Duration::from_mins(1);
+
 #[must_use]
 pub fn socket_path() -> Option<PathBuf> {
     crate::persist::paths::state_dir().map(|dir| dir.join(SOCKET_NAME))
@@ -216,6 +219,17 @@ pub fn run(
         })
     };
 
+    {
+        let session = Arc::clone(&session.client);
+        let daemon = Arc::clone(&daemon);
+        let stop = Arc::clone(&stop);
+        let out = ctx.out.clone();
+        let address = ctx.settings.server_address.to_string();
+        std::thread::spawn(move || {
+            reconnect_loop(&session, &daemon, &out, &address, &stop);
+        });
+    }
+
     for listener in listeners {
         let daemon = Arc::clone(&daemon);
         std::thread::spawn(move || listener.serve(daemon));
@@ -302,6 +316,89 @@ fn cleanup() {
     }
 }
 
+const fn should_reconnect(loss: Option<SessionLoss>, owned: bool) -> bool {
+    owned && matches!(loss, Some(SessionLoss::Disconnected))
+}
+
+fn reconnect_loop(
+    session: &Arc<dyn SessionApi>,
+    daemon: &Daemon,
+    out: &Out,
+    address: &str,
+    stop: &AtomicBool,
+) {
+    let owned = session.daemon_endpoint().is_none();
+    let mut failures = 0u32;
+    let mut announced = false;
+    while !stop.load(Ordering::Relaxed) {
+        std::thread::sleep(hub::POLL);
+        if !should_reconnect(session.session_loss(), owned) {
+            failures = 0;
+            announced = false;
+            continue;
+        }
+        if !announced {
+            out.status(&format!(
+                "the connection to {address} dropped; logging back in…"
+            ));
+            announced = true;
+        }
+        match session.relogin() {
+            Ok(true) => {
+                out.status("logged back in");
+                rejoin(out, session.as_ref(), daemon);
+                failures = 0;
+                announced = false;
+            }
+            outcome => {
+                let wait = backoff(failures);
+                failures = failures.saturating_add(1);
+                out.warn(&format!(
+                    "cannot log back in: {}; trying again in {}s",
+                    relogin_error(outcome),
+                    wait.as_secs()
+                ));
+                wait_or_stop(wait, stop);
+            }
+        }
+    }
+}
+
+fn relogin_error(outcome: soulseek_rs::Result<bool>) -> String {
+    match outcome {
+        Ok(_) => "login rejected by the server".to_string(),
+        Err(e) => format!("login failed: {e}"),
+    }
+}
+
+fn backoff(failures: u32) -> Duration {
+    Duration::from_secs(1 << failures.min(6)).min(RECONNECT_BACKOFF_CAP)
+}
+
+fn wait_or_stop(total: Duration, stop: &AtomicBool) {
+    let deadline = Instant::now() + total;
+    while Instant::now() < deadline && !stop.load(Ordering::Relaxed) {
+        std::thread::sleep(hub::POLL);
+    }
+}
+
+fn rejoin(out: &Out, session: &dyn SessionApi, daemon: &Daemon) {
+    join_rooms(out, session, &daemon.rooms());
+    for user in session.watched_users() {
+        if let Err(e) = session.watch_user(&user) {
+            out.warn(&format!("cannot watch {user} again: {e}"));
+        }
+    }
+}
+
+fn join_rooms(out: &Out, session: &dyn SessionApi, rooms: &[String]) {
+    for room in rooms {
+        if let Err(e) = session.join_room(room) {
+            out.warn(&format!("cannot rejoin {room}: {e}"));
+        }
+    }
+}
+
 /// Bring back what the last run left: re-queue unfinished transfers, rejoin
 /// rooms, and reload the conversation.
 fn restore(
@@ -325,11 +422,7 @@ fn restore(
     );
 
     let rooms = store.load_rooms();
-    for room in &rooms {
-        if let Err(e) = session.join_room(room) {
-            ctx.out.warn(&format!("cannot rejoin {room}: {e}"));
-        }
-    }
+    join_rooms(&ctx.out, session, &rooms);
 
     let unfinished: Vec<PersistedDownload> = store
         .load_downloads()
@@ -425,6 +518,35 @@ mod tests {
         assert!(first.chars().all(|c| c.is_ascii_hexdigit()));
     }
 
+    #[test]
+    fn only_an_owned_dropped_session_reconnects() {
+        assert!(should_reconnect(Some(SessionLoss::Disconnected), true));
+        assert!(
+            !should_reconnect(Some(SessionLoss::Displaced), true),
+            "reconnecting a displaced name starts a login war with whoever \
+             took it"
+        );
+        assert!(
+            !should_reconnect(None, true),
+            "a live session is not retried"
+        );
+        assert!(
+            !should_reconnect(Some(SessionLoss::Disconnected), false),
+            "a borrowed session is the daemon it came from to mend"
+        );
+    }
+
+    #[test]
+    fn the_backoff_doubles_and_then_holds_at_the_cap() {
+        let seconds: Vec<u64> = (0..8).map(|n| backoff(n).as_secs()).collect();
+        assert_eq!(seconds, [1, 2, 4, 8, 16, 32, 60, 60]);
+        assert_eq!(
+            backoff(u32::MAX),
+            RECONNECT_BACKOFF_CAP,
+            "a very long outage must not overflow into a tiny wait"
+        );
+    }
+
     #[test]
     fn two_tokens_are_not_the_same() {
         let dir = tempfile::tempdir().expect("a temp dir");

soulseek-rs/src/daemon/openrpc.rs

@@ -226,7 +226,9 @@ fn event_summary(event: Event) -> &'static str {
         Event::DownloadStatus => "A transfer moved.",
         Event::Browse => "A peer answered a browse request.",
         Event::SessionLoss => {
-            "The server session ended; nothing seen after this means anything."
+            "The server session was lost. A disconnected session is reconnected \
+             on its own and events resume; a displaced one is terminal. Read \
+             daemon.status for the live state."
         }
     }
 }

soulseek-rs/tests/daemon_protocol_e2e.rs

@@ -29,7 +29,7 @@ struct Daemon {
     child: Child,
     state: TempDir,
     port: u16,
-    _server: Soulfind,
+    server: Soulfind,
 }
 
 /// The binary with a scrubbed environment: no test may see the developer's
@@ -78,7 +78,7 @@ impl Daemon {
             child,
             state,
             port,
-            _server: server,
+            server,
         };
         daemon.wait_until_ready()?;
         Some(daemon)
@@ -130,6 +130,17 @@ impl Daemon {
         let read = stream.try_clone().expect("a second handle");
         Some(Connection::from_parts(Box::new(read), Box::new(stream)))
     }
+
+    fn status(&self) -> Option<serde_json::Value> {
+        let mut connection = self.connect_tcp()?;
+        connection.authenticate(Some(&self.token()))?;
+        let reply = connection
+            .call(r#"{"jsonrpc":"2.0","id":1,"method":"daemon.status"}"#)?;
+        serde_json::from_str::<serde_json::Value>(&reply)
+            .ok()?
+            .get_mut("result")
+            .map(serde_json::Value::take)
+    }
 }
 
 impl Drop for Daemon {
@@ -428,10 +439,15 @@ struct Soulfind {
 
 impl Soulfind {
     fn start() -> Option<Self> {
-        let bin = soulfind_binary()?;
         let port = free_port()?;
         let db = std::env::temp_dir().join(format!("soulfind-proto-{port}.db"));
         let _ = std::fs::remove_file(&db);
+        let child = Self::spawn(port, &db)?;
+        Some(Self { child, db, port })
+    }
+
+    fn spawn(port: u16, db: &Path) -> Option<Child> {
+        let bin = soulfind_binary()?;
 
         let mut child = Command::new(bin)
             .args(["-d", db.to_str()?, "-p", &port.to_string()])
@@ -443,7 +459,7 @@ impl Soulfind {
         let deadline = Instant::now() + Duration::from_secs(10);
         while Instant::now() < deadline {
             if TcpStream::connect(("127.0.0.1", port)).is_ok() {
-                return Some(Self { child, db, port });
+                return Some(child);
             }
             std::thread::sleep(Duration::from_millis(100));
         }
@@ -455,16 +471,65 @@ impl Soulfind {
     fn address(&self) -> String {
         format!("127.0.0.1:{}", self.port)
     }
+
+    fn stop(&mut self) {
+        let _ = self.child.kill();
+        let _ = self.child.wait();
+    }
+
+    fn restart(&mut self) -> Option<()> {
+        self.stop();
+        self.child = Self::spawn(self.port, &self.db)?;
+        Some(())
+    }
 }
 
 impl Drop for Soulfind {
     fn drop(&mut self) {
-        let _ = self.child.kill();
-        let _ = self.child.wait();
+        self.stop();
         let _ = std::fs::remove_file(&self.db);
     }
 }
 
+fn wait_until(timeout: Duration, mut f: impl FnMut() -> bool) -> bool {
+    let deadline = Instant::now() + timeout;
+    while Instant::now() < deadline {
+        if f() {
+            return true;
+        }
+        std::thread::sleep(Duration::from_millis(250));
+    }
+    false
+}
+
+#[test]
+fn the_daemon_logs_back_in_after_the_server_drops_it() {
+    let mut daemon = daemon_or_skip!();
+    assert!(
+        daemon.status().expect("a status")["session_loss"].is_null(),
+        "the daemon starts with a live session"
+    );
+
+    daemon.server.stop();
+    assert!(
+        wait_until(Duration::from_secs(15), || daemon
+            .status()
+            .is_some_and(|status| status["session_loss"] == "disconnected")),
+        "the daemon never noticed the server going away"
+    );
+
+    daemon
+        .server
+        .restart()
+        .expect("soulfind comes back on the same port");
+    assert!(
+        wait_until(Duration::from_secs(45), || daemon
+            .status()
+            .is_some_and(|status| status["session_loss"].is_null())),
+        "the daemon never logged back in once the server was reachable again"
+    );
+}
+
 /// A client configured for a remote daemon reports the *daemon's* world.
 ///
 /// This is the download-box workflow: the token lives in the client's config