From a1b88f58bd3040c5c8685d0839eb0a0648e10d60 Mon Sep 17 00:00:00 2001 From: Michel de Graaf Date: Mon, 10 Nov 2025 20:04:06 +0100 Subject: [PATCH] chore: debug logging --- src/actor/mod.rs | 24 ++------ src/actor/peer_actor.rs | 68 ++++++++++++++--------- src/actor/peer_registry.rs | 1 - src/client.rs | 110 +++++++++++++++++++++++++++---------- src/error.rs | 1 - src/main.rs | 6 +- src/server.rs | 29 ++++++---- 7 files changed, 148 insertions(+), 91 deletions(-) diff --git a/src/actor/mod.rs b/src/actor/mod.rs index f56c2d8..0b619ad 100644 --- a/src/actor/mod.rs +++ b/src/actor/mod.rs @@ -101,44 +101,29 @@ impl ActorSystem { }; let handle_for_init = handle.clone(); - // Start the actor event loop on the thread pool self.thread_pool.execute(move || { - trace!("[actor_system] Actor task starting on thread pool"); init(&mut actor, handle_for_init); - trace!("[actor_system] Actor init callback completed"); actor.on_start(); - trace!( - "[actor_system] Actor on_start completed, starting event loop" - ); Self::run_actor_loop(&mut actor, receiver); - trace!("[actor_system] Actor event loop ended"); actor.on_stop(); - trace!("[actor_system] Actor on_stop completed"); }); handle } - /// Run the actor's message processing loop fn run_actor_loop( actor: &mut A, receiver: Receiver>, ) { - trace!("[actor_system] run_actor_loop STARTED"); let tick_interval = Duration::from_millis(100); let mut last_tick = Instant::now(); let mut message_count = 0; let mut tick_count = 0; loop { - // Try to receive messages with timeout for periodic ticks match receiver.recv_timeout(tick_interval) { Ok(ActorMessage::UserMessage(msg)) => { message_count += 1; - trace!( - "[actor_system] Processing message #{}", - message_count - ); actor.handle(msg); } Ok(ActorMessage::Stop) => { @@ -153,15 +138,15 @@ impl ActorSystem { "[actor_system] Received explicit Tick message #{}", tick_count ); + actor.tick(); } Err(std::sync::mpsc::RecvTimeoutError::Timeout) => { - // Periodic tick if last_tick.elapsed() >= tick_interval { tick_count += 1; - if tick_count % 10 == 0 { - trace!("[actor_system] Periodic tick #{} (every 1s log)", tick_count); - } + // if tick_count % 10 == 0 { + // trace!("[actor_system] Periodic tick #{} (every 1s log)", tick_count); + // } actor.tick(); last_tick = Instant::now(); } @@ -170,7 +155,6 @@ impl ActorSystem { trace!( "[actor_system] Channel disconnected, breaking loop" ); - // Channel closed, actor should stop break; } } diff --git a/src/actor/peer_actor.rs b/src/actor/peer_actor.rs index f555695..4be1294 100644 --- a/src/actor/peer_actor.rs +++ b/src/actor/peer_actor.rs @@ -11,35 +11,29 @@ use crate::peer::Peer; use crate::types::{Download, FileSearchResult, Transfer}; use crate::{debug, error, trace, warn}; -use std::io::{self, Write}; +use std::io::{self, Error, Write}; use std::net::TcpStream; use std::sync::mpsc::{Receiver, Sender}; use std::sync::{Arc, RwLock}; -/// Messages that can be sent to a PeerActor #[derive(Debug, Clone)] pub enum PeerMessage { - /// Send a raw message to the peer SendMessage(Message), - /// Handle a file search result FileSearchResult(FileSearchResult), - /// Handle a transfer request TransferRequest(Transfer), - /// Handle a transfer response + UploadFailed(String, String), TransferResponse { token: u32, allowed: bool, reason: Option, }, - /// Handle a place in queue response - PlaceInQueueResponse { filename: String, place: u32 }, - /// Set the peer's username + PlaceInQueueResponse { + filename: String, + place: u32, + }, SetUsername(String), - /// Queue an upload for a file QueueUpload(String), - /// Request a transfer for a download RequestTransfer(Download), - /// Internal: process read operations ProcessRead, } @@ -52,6 +46,7 @@ pub struct PeerActor { self_handle: Option>, dispatcher: Option>, dispatcher_receiver: Option>, + queued_filenames: Vec, } impl PeerActor { @@ -70,6 +65,7 @@ impl PeerActor { self_handle: None, dispatcher: None, dispatcher_receiver: None, + queued_filenames: Vec::new(), } } @@ -172,15 +168,15 @@ impl PeerActor { if !allowed { if let Some(reason_text) = reason { debug!( - "[peer_actor:{}] Transfer rejected: {} - token {}, waiting for TransferRequest...", - username, reason_text, token - ); + "[peer_actor:{}] Transfer rejected: {} - token {}, waiting for TransferRequest...", + username, reason_text, token + ); } } else { debug!( - "[peer_actor:{}] Transfer allowed, ready to connect with token {:}", - username, token - ); + "[peer_actor:{}] Transfer allowed, ready to connect with token {:}", + username, token + ); self.client_channel .send(ClientOperation::DownloadFromPeer( token, @@ -193,9 +189,9 @@ impl PeerActor { PeerMessage::PlaceInQueueResponse { filename, place } => { let username = self.peer.read().unwrap().username.clone(); debug!( - "[peer_actor:{}] Place in queue response - file: {}, place: {}", - username, filename, place - ); + "[peer_actor:{}] Place in queue response - file: {}, place: {}", + username, filename, place + ); } PeerMessage::SetUsername(username) => { trace!( @@ -206,6 +202,7 @@ impl PeerActor { self.peer.write().unwrap().username = username; } PeerMessage::QueueUpload(filename) => { + self.queued_filenames.push(filename.clone()); let message = MessageFactory::build_queue_upload_message(&filename); self.send_message(message); @@ -220,6 +217,11 @@ impl PeerActor { PeerMessage::ProcessRead => { self.process_read(); } + PeerMessage::UploadFailed(username, filename) => { + self.client_channel + .send(ClientOperation::UploadFailed(username, filename)) + .unwrap(); + } } } @@ -256,7 +258,7 @@ impl PeerActor { "[peer_actor:{}] Error reading from peer: {} (kind: {:?}). Disconnecting.", username, e, e.kind() ); - self.disconnect(); + self.disconnect_with_error(e); return; } } @@ -296,7 +298,7 @@ impl PeerActor { "[peer_actor:{}] Error extracting message: {}. Disconnecting peer.", username, e ); - self.disconnect(); + self.disconnect_with_error(e); return; } Ok(None) => { @@ -338,7 +340,7 @@ impl PeerActor { "[peer_actor:{}] Error writing message: {}. Disconnecting.", username, e ); - self.disconnect(); + self.disconnect_with_error(e); return; } @@ -347,10 +349,22 @@ impl PeerActor { "[peer_actor:{}] Error flushing stream: {}. Disconnecting.", username, e ); - self.disconnect(); + self.disconnect_with_error(e); } } + fn disconnect_with_error(&mut self, error: Error) { + let username = self.peer.read().unwrap().username.clone(); + debug!("[peer_actor:{}] disconnect", username); + + self.stream.take(); + + if let Err(e) = self.client_channel.send( + ClientOperation::PeerDisconnected(username, Some(error.into())), + ) { + error!("Failed to send disconnect notification: {}", e); + } + } /// Disconnect the peer and notify the client fn disconnect(&mut self) { let username = self.peer.read().unwrap().username.clone(); @@ -360,7 +374,7 @@ impl PeerActor { if let Err(e) = self .client_channel - .send(ClientOperation::PeerDisconnected(username)) + .send(ClientOperation::PeerDisconnected(username, None)) { error!("Failed to send disconnect notification: {}", e); } @@ -383,7 +397,7 @@ impl Actor for PeerActor { // Trigger initial read if let Some(ref handle) = self.self_handle { match handle.send(PeerMessage::ProcessRead) { - Ok(_) => {}, + Ok(_) => {} Err(e) => error!( "[peer_actor:{}] FAILED to send ProcessRead: {}", username, e diff --git a/src/actor/peer_registry.rs b/src/actor/peer_registry.rs index 5d99014..992396e 100644 --- a/src/actor/peer_registry.rs +++ b/src/actor/peer_registry.rs @@ -113,7 +113,6 @@ impl PeerRegistry { handle.send(message) } - /// Queue an upload for a peer pub fn queue_upload( &self, username: &str, diff --git a/src/client.rs b/src/client.rs index 794ed17..e5adc61 100644 --- a/src/client.rs +++ b/src/client.rs @@ -25,11 +25,13 @@ use std::{ use crate::{debug, error, info, trace}; const DEFALT_LISTEN_PORT: u32 = 2234; +#[derive(Debug)] + pub enum ClientOperation { NewPeer(NewPeer), ConnectToPeer(Peer), SearchResult(FileSearchResult), - PeerDisconnected(String), + PeerDisconnected(String, Option), PierceFireWall(Peer), DownloadFromPeer(u32, Peer, bool), UpdateDownloadTokens(Transfer, String), @@ -40,6 +42,7 @@ pub enum ClientOperation { obfuscation_type: u32, obfuscated_port: u16, }, + UploadFailed(String, String), } pub struct ClientContext { pub peer_registry: Option, @@ -228,6 +231,7 @@ impl Client { size: u64, download_directory: String, ) -> Result { + info!("[client] Downloading {} from {}", filename, username); let start_time = Instant::now(); let hash = md5::md5(&filename); @@ -280,6 +284,34 @@ impl Client { }) } + fn process_failed_uploads( + client_context: Arc>, + username: &str, + filename: Option<&str>, + ) { + let context = client_context.read().unwrap(); + let download_tokens = &context.download_tokens; + let failed_downloads: Vec<_> = download_tokens + .iter() + .filter(|(_, download)| { + download.username == username + && (filename.is_none_or(|f| download.filename == *f)) + }) + .map(|(token, download)| { + download.sender.send(DownloadStatus::Failed).unwrap(); + *token + }) + .collect(); + + failed_downloads.iter().for_each(|token| { + client_context + .write() + .unwrap() + .download_tokens + .remove(token); + }); + } + fn listen_to_client_operations( reader: Receiver, client_context: Arc>, @@ -289,6 +321,7 @@ impl Client { if let Ok(operation) = reader.recv() { match operation { ClientOperation::ConnectToPeer(peer) => { + trace!("[client] ConnectToPeer {:?}", peer); Self::connect_to_peer( peer, client_context.clone(), @@ -304,7 +337,7 @@ impl Client { .search_results .push(file_search.clone()); } - ClientOperation::PeerDisconnected(username) => { + ClientOperation::PeerDisconnected(username, error) => { let context = client_context.read().unwrap(); if let Some(ref registry) = context.peer_registry { if let Some(handle) = @@ -313,6 +346,17 @@ impl Client { let _ = handle.stop(); } } + if let Some(error) = error { + error!( + "[client] Peer {} disconnected with error: {:?}", + username, error + ); + Self::process_failed_uploads( + client_context.clone(), + &username, + None, + ); + } } ClientOperation::PierceFireWall(peer) => { Self::pierce_firewall( @@ -350,27 +394,27 @@ impl Client { .split('\\') .next_back(); match filename { - Some(filename) => { - match download_peer.download_file( - client_context_clone, - Some(download.clone()), - None - ) { - Ok((download, filename)) => { - download.sender.send(DownloadStatus::Completed).unwrap(); - info!("Successfully downloaded {} bytes to {}", download.size, filename); - } - Err(e) => { - download.sender.send(DownloadStatus::Failed).unwrap(); - error!( - "Failed to download file '{}' from {}:{} (token: {}) - Error: {}", - filename, peer.host, peer.port, download.token, e - ); - } - } - } - None => error!("Cant find filename to save download: {:?}", download.filename), - } + Some(filename) => { + match download_peer.download_file( + client_context_clone, + Some(download.clone()), + None + ) { + Ok((download, filename)) => { + download.sender.send(DownloadStatus::Completed).unwrap(); + info!("Successfully downloaded {} bytes to {}", download.size, filename); + } + Err(e) => { + download.sender.send(DownloadStatus::Failed).unwrap(); + error!( + "Failed to download file '{}' from {}:{} (token: {}) - Error: {}", + filename, peer.host, peer.port, download.token, e + ); + } + } + } + None => error!("Cant find filename to save download: {:?}", download.filename), + } }); } None => { @@ -435,9 +479,9 @@ impl Client { obfuscated_port, } => { debug!( - "Received peer address for {}: {}:{} (obf_type: {}, obf_port: {})", - username, host, port, obfuscation_type, obfuscated_port - ); + "Received peer address for {}: {}:{} (obf_type: {}, obf_port: {})", + username, host, port, obfuscation_type, obfuscated_port + ); let peer_exists = client_context .read() @@ -503,9 +547,9 @@ impl Client { if let Some((key, download)) = key_to_remove { trace!( - "[client] UpdateDownloadTokens found {key}, transfer: {:?}", - transfer - ); + "[client] UpdateDownloadTokens found {key}, transfer: {:?}", + transfer + ); context.download_tokens.insert( transfer.token, @@ -522,6 +566,13 @@ impl Client { context.download_tokens.remove(&key); } } + ClientOperation::UploadFailed(username, filename) => { + Self::process_failed_uploads( + client_context.clone(), + &username, + Some(&filename), + ); + } } } }); @@ -544,7 +595,6 @@ impl Client { ConnectionType::P => { let username = peer.username.clone(); - // Establish TCP connection or use provided stream let mut tcp_stream = match stream { Some(s) => s, None => { diff --git a/src/error.rs b/src/error.rs index 74fddab..81adadd 100644 --- a/src/error.rs +++ b/src/error.rs @@ -3,7 +3,6 @@ use std::{error::Error, fmt}; /// Custom error type for the Soulseek download library #[derive(Debug)] pub enum SoulseekRs { - /// Network-related errors (I/O, connection failures) NetworkError(std::io::Error), /// Authentication failed during login AuthenticationFailed, diff --git a/src/main.rs b/src/main.rs index 5e68ab7..55d15bd 100644 --- a/src/main.rs +++ b/src/main.rs @@ -6,14 +6,16 @@ fn main() { PeerAddress::new(String::from("server.slsknet.org"), 2242), String::from("xxxxxx"), String::from("xxxxxx"), - true, + false, None, ); client.connect(); match client.login() { Ok(_) => { - match client.search("michel test file", Duration::from_secs(10)) { + match client + .search("Eternity Alex Kassian", Duration::from_secs(10)) + { Ok(results) => { if let Some(result) = results.iter().find(|r| !r.files.is_empty()) diff --git a/src/server.rs b/src/server.rs index 4ad5082..d8cc7d9 100644 --- a/src/server.rs +++ b/src/server.rs @@ -341,21 +341,30 @@ impl Server { if let Ok(operation) = server_channel.recv() { match operation { ServerOperation::ConnectToPeer(peer) => { - debug!( - "[server] ConnectToPeer {} - ConnectionType {}", - peer.username, peer.connection_type + trace!( + "[server] → ConnectToPeer {}", + peer.username ); - if let Some(op) = match peer.connection_type { - ConnectionType::P => { - Some(ClientOperation::ConnectToPeer(peer)) - } - ConnectionType::F => { - Some(ClientOperation::ConnectToPeer(peer)) + ConnectionType::P | ConnectionType::F => { + Some(ClientOperation::ConnectToPeer( + peer.clone(), + )) } ConnectionType::D => None, } { - client_channel.send(op).unwrap(); + trace!( + "[server] → ConnectToPeer {} - op {:?}", + peer.username, + op + ); + + match client_channel.send(op) { + Ok(_) => {} + Err(e) => { + error!("[server] Failed to send: {}", e) + } + } } } ServerOperation::LoginStatus(message) => {