Merge pull request #52 from michel/daemon-reconnect

fix(daemon): reconnect after the server session drops
This commit is contained in:
Michel de Graaf
2026-09-02 14:45:50 +02:00
committed by GitHub
8 changed files with 436 additions and 63 deletions
+213 -41
View File
@@ -34,7 +34,7 @@ use crate::types::{
use crate::utils::lock::RwLockExt; use crate::utils::lock::RwLockExt;
use std::io::{self, Write}; use std::io::{self, Write};
use std::net::TcpStream; use std::net::{TcpStream, ToSocketAddrs};
use std::sync::mpsc::{Receiver, Sender}; use std::sync::mpsc::{Receiver, Sender};
use std::sync::{Arc, RwLock}; use std::sync::{Arc, RwLock};
use std::time::{Duration, Instant}; use std::time::{Duration, Instant};
@@ -46,6 +46,8 @@ use crate::{SoulseekRs, debug, error, trace, warn};
/// rather than undercutting it. /// rather than undercutting it.
const LOGIN_VERDICT_TIMEOUT: Duration = Duration::from_secs(30); const LOGIN_VERDICT_TIMEOUT: Duration = Duration::from_secs(30);
const CONNECT_TIMEOUT: Duration = Duration::from_secs(10);
#[derive(Debug, Clone)] #[derive(Debug, Clone)]
pub struct PeerAddress { pub struct PeerAddress {
host: String, host: String,
@@ -327,23 +329,25 @@ impl ServerActor {
self.session = session; self.session = session;
} }
fn initiate_connection(&mut self) { fn initiate_connection(&mut self) -> bool {
let stream = match TcpStream::connect(( let stream = (self.address.host.as_str(), self.address.port)
self.address.host.as_str(), .to_socket_addrs()
self.address.port, .ok()
)) { .and_then(|mut addrs| {
Ok(s) => s, addrs.find_map(|addr| {
Err(e) => { TcpStream::connect_timeout(&addr, CONNECT_TIMEOUT).ok()
error!("[server] Failed to connect to {}: {}", self.address, e); })
self.disconnect_with_error(); });
return; let Some(stream) = stream else {
} error!("[server] Failed to connect to {}", self.address);
self.disconnect_with_error();
return false;
}; };
if let Err(e) = stream.set_nonblocking(true) { if let Err(e) = stream.set_nonblocking(true) {
error!("[server] Failed to set non-blocking: {}", e); error!("[server] Failed to set non-blocking: {}", e);
self.disconnect_with_error(); self.disconnect_with_error();
return; return false;
} }
stream.set_nodelay(true).ok(); stream.set_nodelay(true).ok();
@@ -351,6 +355,7 @@ impl ServerActor {
self.connection_state = ConnectionState::Connecting { self.connection_state = ConnectionState::Connecting {
since: Instant::now(), since: Instant::now(),
}; };
true
} }
pub fn set_self_handle(&mut self, handle: ActorHandle<ServerMessage>) { pub fn set_self_handle(&mut self, handle: ActorHandle<ServerMessage>) {
@@ -420,14 +425,14 @@ impl ServerActor {
} }
fn handle_message(&mut self, msg: ServerMessage) { fn handle_message(&mut self, msg: ServerMessage) {
if !matches!(self.connection_state, ConnectionState::Connected) { if !matches!(self.connection_state, ConnectionState::Connected)
if matches!(&msg, ServerMessage::ProcessRead) { && !matches!(
// Always process read operations &msg,
} else { ServerMessage::ProcessRead | ServerMessage::Login { .. }
// Queue all other messages when not connected )
self.queued_messages.push(msg); {
return; self.queued_messages.push(msg);
} return;
} }
match msg { match msg {
@@ -615,12 +620,6 @@ impl ServerActor {
} }
fn handle_login_status(&mut self, message: bool) { fn handle_login_status(&mut self, message: bool) {
match self.context.write_safe() {
Ok(mut ctx) => ctx.logged_in = Some(message),
Err(e) => {
error!("[server] LoginStatus write: {}", e);
}
}
// Send the post-login handshake exactly once, only on success, // Send the post-login handshake exactly once, only on success,
// on the live path (the old ServerActor::login did this but was // on the live path (the old ServerActor::login did this but was
// never called). Advertises real shared counts and, when // never called). Advertises real shared counts and, when
@@ -634,6 +633,13 @@ impl ServerActor {
) { ) {
self.send_message(msg); self.send_message(msg);
} }
self.session.clear();
}
match self.context.write_safe() {
Ok(mut ctx) => ctx.logged_in = Some(message),
Err(e) => {
error!("[server] LoginStatus write: {}", e);
}
} }
} }
@@ -695,6 +701,13 @@ impl ServerActor {
version: ClientVersion, version: ClientVersion,
response: std::sync::mpsc::Sender<Result<bool, SoulseekRs>>, response: std::sync::mpsc::Sender<Result<bool, SoulseekRs>>,
) { ) {
if self.stream.is_none() && !self.initiate_connection() {
let _ = response.send(Err(SoulseekRs::NotConnected));
return;
}
if let Ok(mut ctx) = self.context.write_safe() {
ctx.logged_in = None;
}
self.queue_message(MessageFactory::build_login_message( self.queue_message(MessageFactory::build_login_message(
&username, &password, version, &username, &password, version,
)); ));
@@ -882,13 +895,10 @@ impl ServerActor {
fn disconnect_with_error(&mut self) { fn disconnect_with_error(&mut self) {
debug!("[server] disconnect"); debug!("[server] disconnect");
// Losing an established connection ends the session: nothing reconnects
// it, so anything still waiting on the network is waiting for good.
if matches!(self.connection_state, ConnectionState::Connected) { if matches!(self.connection_state, ConnectionState::Connected) {
self.session.record(SessionLoss::Disconnected); self.session.record(SessionLoss::Disconnected);
} }
self.connection_state = ConnectionState::Disconnected; self.disconnect();
self.stream.take();
} }
fn disconnect(&mut self) { fn disconnect(&mut self) {
@@ -896,6 +906,14 @@ impl ServerActor {
self.stream.take(); self.stream.take();
self.connection_state = ConnectionState::Disconnected; self.connection_state = ConnectionState::Disconnected;
self.dispatcher = None;
self.dispatcher_sender = None;
self.dispatcher_receiver = None;
self.queued_messages.clear();
self.reader = MessageReader::new();
if let Ok(mut ctx) = self.context.write_safe() {
ctx.logged_in = None;
}
} }
fn check_connection_status(&mut self) { fn check_connection_status(&mut self) {
@@ -952,7 +970,7 @@ impl Actor for ServerActor {
fn on_start(&mut self) { fn on_start(&mut self) {
if self.stream.is_none() { if self.stream.is_none() {
self.initiate_connection(); let _ = self.initiate_connection();
} else { } else {
self.connection_state = ConnectionState::Connected; self.connection_state = ConnectionState::Connected;
self.on_connection_established(); self.on_connection_established();
@@ -989,15 +1007,7 @@ mod tests {
#[test] #[test]
fn a_timed_out_connect_parks_the_actor_in_disconnected() { fn a_timed_out_connect_parks_the_actor_in_disconnected() {
let (tx, _rx) = std::sync::mpsc::channel(); let mut actor = parked_actor(1);
let mut actor = ServerActor::new(
PeerAddress::new("127.0.0.1".to_string(), 1),
tx,
0,
false,
0,
0,
);
actor.connection_state = ConnectionState::Connecting { actor.connection_state = ConnectionState::Connecting {
since: Instant::now().checked_sub(Duration::from_secs(21)).unwrap(), since: Instant::now().checked_sub(Duration::from_secs(21)).unwrap(),
}; };
@@ -1010,6 +1020,168 @@ mod tests {
); );
} }
fn parked_actor(port: u16) -> ServerActor {
let (tx, _rx) = std::sync::mpsc::channel();
ServerActor::new(
PeerAddress::new("127.0.0.1".to_string(), port),
tx,
0,
false,
0,
0,
)
}
#[test]
fn dropping_the_connection_forgets_the_dispatcher_and_the_verdict() {
let mut actor = parked_actor(1);
actor.initialize_dispatcher();
actor.connection_state = ConnectionState::Connected;
actor.handle_login_status(true);
actor.disconnect_with_error();
assert_eq!(actor.session.loss(), Some(SessionLoss::Disconnected));
assert!(
actor.dispatcher_sender.is_none(),
"a login queued now must wait in the buffer, not in a channel \
the next connection replaces"
);
assert_eq!(
actor.context.read().unwrap().logged_in,
None,
"the old verdict must not answer the next login"
);
}
#[test]
fn disconnect_clears_a_partial_frame_so_the_next_session_reframes_clean() {
use std::io::Write;
let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
let addr = listener.local_addr().unwrap();
let client = TcpStream::connect(addr).unwrap();
let (mut server_side, _) = listener.accept().unwrap();
let partial_frame = [10u8, 0, 0, 0, 1, 2, 3];
server_side.write_all(&partial_frame).unwrap();
server_side.flush().unwrap();
client.set_nonblocking(true).unwrap();
let mut actor = parked_actor(addr.port());
actor.stream = Some(client);
actor.connection_state = ConnectionState::Connected;
for _ in 0..50 {
actor.process_read();
if actor.reader.buffer_len() > 0 {
break;
}
std::thread::sleep(Duration::from_millis(10));
}
assert!(
actor.reader.buffer_len() > 0,
"an incomplete frame should be waiting for its rest"
);
actor.disconnect();
assert_eq!(
actor.reader.buffer_len(),
0,
"a stale partial frame carried across a reconnect reads the next \
session's bytes as a length prefix and wedges recovery"
);
}
#[test]
fn a_login_on_a_parked_actor_dials_again_and_queues_the_login() {
let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
let mut actor = parked_actor(listener.local_addr().unwrap().port());
actor.session.record(SessionLoss::Disconnected);
let (response, _verdict) = std::sync::mpsc::channel();
actor.handle_login(
"u".into(),
"p".into(),
ClientVersion::default(),
response,
);
assert!(
matches!(
actor.connection_state,
ConnectionState::Connecting { .. }
),
"a parked actor must dial again on login"
);
assert_eq!(
actor.queued_messages.len(),
1,
"the login waits for the connection to come up"
);
}
#[test]
fn a_parked_actor_handles_login_rather_than_queueing_it() {
let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
let mut actor = parked_actor(listener.local_addr().unwrap().port());
actor.session.record(SessionLoss::Disconnected);
let (response, _verdict) = std::sync::mpsc::channel();
actor.handle_message(ServerMessage::Login {
username: "u".into(),
password: "p".into(),
version: ClientVersion::default(),
response,
});
assert!(
matches!(
actor.connection_state,
ConnectionState::Connecting { .. }
),
"a parked actor must act on a login, not park it in the queue"
);
assert!(
actor
.queued_messages
.iter()
.all(|m| !matches!(m, ServerMessage::Login { .. })),
"the login must not be sitting unhandled in the queue"
);
}
#[test]
fn a_login_with_nobody_listening_is_refused_at_once() {
let mut actor = parked_actor(1);
let (response, verdict) = std::sync::mpsc::channel();
actor.handle_login(
"u".into(),
"p".into(),
ClientVersion::default(),
response,
);
assert!(
matches!(
verdict.recv_timeout(Duration::from_secs(5)),
Ok(Err(SoulseekRs::NotConnected))
),
"a refused dial must not make the caller wait out the verdict \
timeout"
);
}
#[test]
fn a_successful_login_marks_the_session_live_again() {
let mut actor = parked_actor(1);
actor.session.record(SessionLoss::Disconnected);
actor.handle_login_status(true);
assert_eq!(actor.session.loss(), None);
}
#[test] #[test]
fn post_login_messages_carry_counts_and_conditional_wait_port() { fn post_login_messages_carry_counts_and_conditional_wait_port() {
let messages = post_login_messages(true, 4321, 3, 7); let messages = post_login_messages(true, 4321, 3, 7);
+5
View File
@@ -510,4 +510,9 @@ impl SessionWatch {
_ => None, _ => None,
} }
} }
pub fn clear(&self) {
self.0
.store(Self::LIVE, std::sync::atomic::Ordering::Relaxed);
}
} }
+1 -1
View File
@@ -2193,7 +2193,7 @@
"type": "null" "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 "x-notification": true
} }
], ],
+7
View File
@@ -60,6 +60,9 @@ pub trait SessionApi: Send + Sync {
} }
fn listen_port(&self) -> Option<u16>; fn listen_port(&self) -> Option<u16>;
fn session_loss(&self) -> Option<SessionLoss>; 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 and wait out the window. Every implementation is
/// `search_with_cancel` with nothing to cancel on. /// `search_with_cancel` with nothing to cancel on.
@@ -191,6 +194,10 @@ impl SessionApi for Client {
Self::session_loss(self) Self::session_loss(self)
} }
fn relogin(&self) -> Result<bool> {
Self::login(self)
}
fn search_with_cancel( fn search_with_cancel(
&self, &self,
query: &str, query: &str,
+8 -8
View File
@@ -15,7 +15,7 @@ use super::proto::{
SessionLossEvent, UploadInfoDto, UploadStatusDto, UserMessageDto, SessionLossEvent, UploadInfoDto, UploadStatusDto, UserMessageDto,
}; };
use crate::api::SessionApi; use crate::api::SessionApi;
use soulseek_rs::DownloadStatus; use soulseek_rs::{DownloadStatus, SessionLoss};
use std::collections::HashMap; use std::collections::HashMap;
use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::mpsc::{Receiver, SyncSender, TrySendError, sync_channel}; 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 /// Last upload state reported per (user, file), so an upload that has not
/// moved is not re-announced every tick. /// moved is not re-announced every tick.
uploads: HashMap<(String, String), UploadInfoDto>, uploads: HashMap<(String, String), UploadInfoDto>,
/// Whether the session-loss event has already gone out. It is terminal, so last_loss: Option<SessionLoss>,
/// it is worth saying exactly once.
reported_loss: bool,
} }
/// How long to keep asking whether a peer has answered a browse request. /// 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_browses(session, hub, browses);
publish_downloads(hub, downloads); publish_downloads(hub, downloads);
if !state.reported_loss let loss = session.session_loss();
&& let Some(loss) = session.session_loss() if loss == state.last_loss {
{ return;
state.reported_loss = true; }
state.last_loss = loss;
if let Some(loss) = loss {
hub.publish( hub.publish(
Event::SessionLoss, Event::SessionLoss,
serde_json::to_value(SessionLossEvent { loss: loss.into() }) serde_json::to_value(SessionLossEvent { loss: loss.into() })
+128 -6
View File
@@ -24,11 +24,12 @@ pub mod server;
use crate::api::SessionApi; use crate::api::SessionApi;
use crate::commands::{Ctx, Session}; use crate::commands::{Ctx, Session};
use crate::output::{CliError, CliResult}; use crate::output::{CliError, CliResult, Out};
use crate::persist::state::{PersistedDownload, PersistedMessage, StateStore}; use crate::persist::state::{PersistedDownload, PersistedMessage, StateStore};
use hub::{Hub, PendingBrowses, PendingDownload}; use hub::{Hub, PendingBrowses, PendingDownload};
use proto::ChatMessageDto; use proto::ChatMessageDto;
use server::Daemon; use server::Daemon;
use soulseek_rs::SessionLoss;
use std::path::{Path, PathBuf}; use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex}; 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. /// loses little, rare enough not to churn the disk on a busy queue.
const SAVE_INTERVAL: Duration = Duration::from_secs(5); const SAVE_INTERVAL: Duration = Duration::from_secs(5);
const RECONNECT_BACKOFF_CAP: Duration = Duration::from_mins(1);
#[must_use] #[must_use]
pub fn socket_path() -> Option<PathBuf> { pub fn socket_path() -> Option<PathBuf> {
crate::persist::paths::state_dir().map(|dir| dir.join(SOCKET_NAME)) 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 { for listener in listeners {
let daemon = Arc::clone(&daemon); let daemon = Arc::clone(&daemon);
std::thread::spawn(move || listener.serve(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 /// Bring back what the last run left: re-queue unfinished transfers, rejoin
/// rooms, and reload the conversation. /// rooms, and reload the conversation.
fn restore( fn restore(
@@ -325,11 +422,7 @@ fn restore(
); );
let rooms = store.load_rooms(); let rooms = store.load_rooms();
for room in &rooms { join_rooms(&ctx.out, session, &rooms);
if let Err(e) = session.join_room(room) {
ctx.out.warn(&format!("cannot rejoin {room}: {e}"));
}
}
let unfinished: Vec<PersistedDownload> = store let unfinished: Vec<PersistedDownload> = store
.load_downloads() .load_downloads()
@@ -425,6 +518,35 @@ mod tests {
assert!(first.chars().all(|c| c.is_ascii_hexdigit())); 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] #[test]
fn two_tokens_are_not_the_same() { fn two_tokens_are_not_the_same() {
let dir = tempfile::tempdir().expect("a temp dir"); let dir = tempfile::tempdir().expect("a temp dir");
+3 -1
View File
@@ -226,7 +226,9 @@ fn event_summary(event: Event) -> &'static str {
Event::DownloadStatus => "A transfer moved.", Event::DownloadStatus => "A transfer moved.",
Event::Browse => "A peer answered a browse request.", Event::Browse => "A peer answered a browse request.",
Event::SessionLoss => { 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."
} }
} }
} }
+71 -6
View File
@@ -29,7 +29,7 @@ struct Daemon {
child: Child, child: Child,
state: TempDir, state: TempDir,
port: u16, port: u16,
_server: Soulfind, server: Soulfind,
} }
/// The binary with a scrubbed environment: no test may see the developer's /// The binary with a scrubbed environment: no test may see the developer's
@@ -78,7 +78,7 @@ impl Daemon {
child, child,
state, state,
port, port,
_server: server, server,
}; };
daemon.wait_until_ready()?; daemon.wait_until_ready()?;
Some(daemon) Some(daemon)
@@ -130,6 +130,17 @@ impl Daemon {
let read = stream.try_clone().expect("a second handle"); let read = stream.try_clone().expect("a second handle");
Some(Connection::from_parts(Box::new(read), Box::new(stream))) 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 { impl Drop for Daemon {
@@ -428,10 +439,15 @@ struct Soulfind {
impl Soulfind { impl Soulfind {
fn start() -> Option<Self> { fn start() -> Option<Self> {
let bin = soulfind_binary()?;
let port = free_port()?; let port = free_port()?;
let db = std::env::temp_dir().join(format!("soulfind-proto-{port}.db")); let db = std::env::temp_dir().join(format!("soulfind-proto-{port}.db"));
let _ = std::fs::remove_file(&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) let mut child = Command::new(bin)
.args(["-d", db.to_str()?, "-p", &port.to_string()]) .args(["-d", db.to_str()?, "-p", &port.to_string()])
@@ -443,7 +459,7 @@ impl Soulfind {
let deadline = Instant::now() + Duration::from_secs(10); let deadline = Instant::now() + Duration::from_secs(10);
while Instant::now() < deadline { while Instant::now() < deadline {
if TcpStream::connect(("127.0.0.1", port)).is_ok() { 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)); std::thread::sleep(Duration::from_millis(100));
} }
@@ -455,16 +471,65 @@ impl Soulfind {
fn address(&self) -> String { fn address(&self) -> String {
format!("127.0.0.1:{}", self.port) 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 { impl Drop for Soulfind {
fn drop(&mut self) { fn drop(&mut self) {
let _ = self.child.kill(); self.stop();
let _ = self.child.wait();
let _ = std::fs::remove_file(&self.db); 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. /// 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 /// This is the download-box workflow: the token lives in the client's config