dumbgoy/soulseek-rs-modded
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

refactor: replace lock unwraps with poison-aware helpers

Browse Source 
Add SoulseekRs::LockPoisoned variant and a small RwLockExt/MutexExt
helper module exposing read_safe/write_safe/lock_safe that map
PoisonError to that variant. Walk every non-test lock acquisition
across client, peer/server actors, peer registry, listener and the
thread pool: propagate via ? where the function returns Result, log
and degrade to a sensible default (return early, false, None, empty)
where it does not. Client::connect now returns Result so the CLI can
surface the failure instead of panicking.
This commit is contained in:
michel
2026-04-26 12:32:46 +02:00
parent e1ee3ac493
commit 2722552e35
10 changed files with 473 additions and 164 deletions
Download Patch File Download Diff File Expand all files Collapse all files
soulseek-rs-lib/src/actor/peer_actor.rs
+100 -39
View File
@@ -9,6 +9,7 @@ use crate::message::server::MessageFactory;
use crate::message::{Handlers, Message, MessageReader, MessageType};
use crate::peer::Peer;
use crate::types::{Download, SearchResult, Transfer};
use crate::utils::lock::RwLockExt;
use crate::{debug, error, trace, warn};
use std::io::{self, Error, Write};
@@ -80,6 +81,26 @@ impl PeerActor {
self.self_handle = Some(handle);
}
fn peer_username(&self) -> String {
match self.peer.read_safe() {
Ok(p) => p.username.clone(),
Err(e) => {
error!("[peer_actor] peer lock poisoned: {}", e);
"<unknown>".to_string()
}
}
}
fn peer_snapshot(&self) -> Option<Peer> {
match self.peer.read_safe() {
Ok(p) => Some(p.clone()),
Err(e) => {
error!("[peer_actor] peer lock poisoned: {}", e);
None
}
}
}
fn initialize_dispatcher(&mut self) {
let (dispatcher_sender, dispatcher_receiver) =
std::sync::mpsc::channel::<PeerMessage>();
@@ -135,23 +156,35 @@ impl PeerActor {
self.send_message(message);
}
PeerMessage::FileSearchResult(file_search) => {
self.client_channel
if let Err(e) = self
.client_channel
.send(ClientOperation::SearchResult(file_search))
.unwrap();
{
error!(
"[peer:{}] failed to forward search result: {}",
self.peer_username(),
e
);
}
}
PeerMessage::TransferRequest(transfer) => {
let username = self.peer.read().unwrap().username.clone();
let username = self.peer_username();
debug!(
"[peer:{}] TransferRequest for {}",
username, transfer.token
);
self.client_channel
.send(ClientOperation::UpdateDownloadTokens(
if let Err(e) = self.client_channel.send(
ClientOperation::UpdateDownloadTokens(
transfer.clone(),
username.clone(),
))
.unwrap();
),
) {
error!(
"[peer:{}] failed to send UpdateDownloadTokens: {}",
username, e
);
}
let transfer_response =
MessageFactory::build_transfer_response_message(transfer);
@@ -171,7 +204,7 @@ impl PeerActor {
allowed,
reason,
} => {
let username = self.peer.read().unwrap().username.clone();
let username = self.peer_username();
debug!(
"[peer:{}] transfer response token: {} allowed: {}",
username, token, allowed
@@ -189,17 +222,25 @@ impl PeerActor {
"[peer:{}] Transfer allowed, ready to connect with token {:}",
username, token
);
self.client_channel
.send(ClientOperation::DownloadFromPeer(
let Some(peer_snapshot) = self.peer_snapshot() else {
return;
};
if let Err(e) = self.client_channel.send(
ClientOperation::DownloadFromPeer(
token,
self.peer.read().unwrap().clone(),
peer_snapshot,
allowed,
))
.unwrap();
),
) {
error!(
"[peer:{}] failed to send DownloadFromPeer: {}",
username, e
);
}
}
}
PeerMessage::PlaceInQueueResponse { filename, place } => {
let username = self.peer.read().unwrap().username.clone();
let username = self.peer_username();
debug!(
"[peer:{}] Place in queue response - file: {}, place: {}",
username, filename, place
@@ -208,10 +249,15 @@ impl PeerActor {
PeerMessage::SetUsername(username) => {
trace!(
"[peer:{}] SetUsername: {}",
self.peer.read().unwrap().username,
self.peer_username(),
username
);
self.peer.write().unwrap().username = username;
match self.peer.write_safe() {
Ok(mut p) => p.username = username,
Err(e) => {
error!("[peer_actor] SetUsername write: {}", e)
}
}
}
PeerMessage::QueueUpload(filename) => {
let message =
@@ -229,9 +275,15 @@ impl PeerActor {
self.process_read();
}
PeerMessage::UploadFailed(username, filename) => {
self.client_channel
if let Err(e) = self
.client_channel
.send(ClientOperation::UploadFailed(username, filename))
.unwrap();
{
error!(
"[peer_actor] failed to forward UploadFailed: {}",
e
);
}
}
}
}
@@ -251,14 +303,15 @@ impl PeerActor {
Ok(()) => {}
Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => {}
Err(ref e) if e.kind() == io::ErrorKind::TimedOut => {
let peer_lock = self.peer.read().unwrap();
debug!(
"Read operation timed out for peer actor {:}:{:}",
peer_lock.host, peer_lock.port
);
if let Ok(peer_lock) = self.peer.read_safe() {
debug!(
"Read operation timed out for peer actor {:}:{:}",
peer_lock.host, peer_lock.port
);
}
}
Err(e) => {
let username = self.peer.read().unwrap().username.clone();
let username = self.peer_username();
error!(
"[peer:{}] Error reading from peer: {} (kind: {:?}). Disconnecting.",
username,
@@ -274,7 +327,7 @@ impl PeerActor {
}
fn extract_and_process_messages(&mut self) {
let username = self.peer.read().unwrap().username.clone();
let username = self.peer_username();
let mut extracted_count = 0;
loop {
match self.reader.extract_message() {
@@ -315,6 +368,7 @@ impl PeerActor {
}
fn send_message(&mut self, message: Message) {
let username = self.peer_username();
let stream = match self.stream.as_mut() {
Some(s) => s,
None => {
@@ -323,7 +377,6 @@ impl PeerActor {
}
};
let username = self.peer.read().unwrap().username.clone();
trace!(
"[peer:{}] ➡ {:?}",
username,
@@ -356,7 +409,7 @@ impl PeerActor {
}
fn disconnect_with_error(&mut self, error: Error) {
let username = self.peer.read().unwrap().username.clone();
let username = self.peer_username();
debug!("[peer:{}] disconnect", username);
self.stream.take();
@@ -368,7 +421,7 @@ impl PeerActor {
}
}
fn disconnect(&mut self) {
let username = self.peer.read().unwrap().username.clone();
let username = self.peer_username();
debug!("[peer:{}] disconnect", username);
self.stream.take();
@@ -382,11 +435,13 @@ impl PeerActor {
}
fn initiate_connection(&mut self) -> bool {
let peer = self.peer.read().unwrap();
let username = peer.username.clone();
let host = peer.host.clone();
let port = peer.port;
drop(peer);
let (username, host, port) = match self.peer.read_safe() {
Ok(peer) => (peer.username.clone(), peer.host.clone(), peer.port),
Err(e) => {
error!("[peer_actor] initiate_connection peer lock: {}", e);
return false;
}
};
let socket_addr =
format!("{}:{}", host, port).parse::<std::net::SocketAddr>();
@@ -438,7 +493,7 @@ impl PeerActor {
return;
};
let username = self.peer.read().unwrap().username.clone();
let username = self.peer_username();
if since.elapsed() > Duration::from_secs(20) {
error!("[peer:{}] Connection timeout after 20 seconds", username);
@@ -467,10 +522,16 @@ impl PeerActor {
}
fn on_connection_established(&mut self) {
let peer = self.peer.read().unwrap();
let username = peer.username.clone();
let token = peer.token;
drop(peer);
let (username, token) = match self.peer.read_safe() {
Ok(peer) => (peer.username.clone(), peer.token),
Err(e) => {
error!(
"[peer_actor] on_connection_established peer lock: {}",
e
);
return;
}
};
let Some(ref mut stream) = self.stream else {
return;
@@ -520,7 +581,7 @@ impl Actor for PeerActor {
}
fn on_stop(&mut self) {
let username = self.peer.read().unwrap().username.clone();
let username = self.peer_username();
trace!("[peer:{}] actor stopping", username);
self.disconnect();
}
soulseek-rs-lib/src/actor/peer_registry.rs
+41 -11
View File
@@ -1,9 +1,10 @@
use crate::actor::peer_actor::{PeerActor, PeerMessage};
use crate::actor::{ActorHandle, ActorSystem};
use crate::client::ClientOperation;
use crate::debug;
use crate::message::MessageReader;
use crate::peer::Peer;
use crate::utils::lock::MutexExt;
use crate::{debug, error};
use std::collections::HashMap;
use std::net::TcpStream;
@@ -44,22 +45,36 @@ impl PeerRegistry {
actor.set_self_handle(handle);
});
let mut peers = self.peers.lock().unwrap();
let mut peers = self
.peers
.lock_safe()
.map_err(|e| format!("peer registry lock poisoned: {}", e))?;
peers.insert(username.clone(), handle.clone());
Ok(handle)
}
pub fn get_peer(&self, username: &str) -> Option<ActorHandle<PeerMessage>> {
let peers = self.peers.lock().unwrap();
peers.get(username).cloned()
match self.peers.lock_safe() {
Ok(peers) => peers.get(username).cloned(),
Err(e) => {
error!("[peer_registry] get_peer: {}", e);
None
}
}
}
pub fn remove_peer(
&self,
username: &str,
) -> Option<ActorHandle<PeerMessage>> {
let mut peers = self.peers.lock().unwrap();
let mut peers = match self.peers.lock_safe() {
Ok(p) => p,
Err(e) => {
error!("[peer_registry] remove_peer: {}", e);
return None;
}
};
let handle = peers.remove(username);
if handle.is_some() {
@@ -70,18 +85,33 @@ impl PeerRegistry {
}
pub fn get_all_usernames(&self) -> Vec<String> {
let peers = self.peers.lock().unwrap();
peers.keys().cloned().collect()
match self.peers.lock_safe() {
Ok(peers) => peers.keys().cloned().collect(),
Err(e) => {
error!("[peer_registry] get_all_usernames: {}", e);
Vec::new()
}
}
}
pub fn count(&self) -> usize {
let peers = self.peers.lock().unwrap();
peers.len()
match self.peers.lock_safe() {
Ok(peers) => peers.len(),
Err(e) => {
error!("[peer_registry] count: {}", e);
0
}
}
}
pub fn contains(&self, username: &str) -> bool {
let peers = self.peers.lock().unwrap();
peers.contains_key(username)
match self.peers.lock_safe() {
Ok(peers) => peers.contains_key(username),
Err(e) => {
error!("[peer_registry] contains: {}", e);
false
}
}
}
pub fn send_to_peer(
soulseek-rs-lib/src/actor/server_actor.rs
+27 -11
View File
@@ -17,6 +17,7 @@ use crate::message::{Handlers, MessageType};
use crate::message::{Message, MessageReader};
use crate::peer::ConnectionType;
use crate::peer::Peer;
use crate::utils::lock::RwLockExt;
use std::io::{self, Error, Write};
use std::net::{TcpStream, ToSocketAddrs};
@@ -338,9 +339,12 @@ impl ServerActor {
self.dispatcher_receiver = Some(dispatcher_receiver);
self.dispatcher_sender = Some(dispatcher_sender.clone());
self.client_channel
if let Err(e) = self
.client_channel
.send(ClientOperation::SetServerSender(dispatcher_sender.clone()))
.unwrap();
{
error!("[server] failed to send SetServerSender: {}", e);
}
let mut handlers = Handlers::new();
@@ -401,7 +405,7 @@ impl ServerActor {
}
{
logged_in = context.read().unwrap().logged_in
logged_in = context.read_safe()?.logged_in
}
if logged_in.is_some() {
@@ -409,7 +413,8 @@ impl ServerActor {
}
}
if logged_in.unwrap() {
let logged_in = logged_in.ok_or(SoulseekRs::Timeout)?;
if logged_in {
info!("Logged in as {}", username);
self.queue_message(MessageFactory::build_shared_folders_message(
1, 499,
@@ -425,7 +430,7 @@ impl ServerActor {
}
}
Ok(logged_in.unwrap())
Ok(logged_in)
}
pub fn file_search(&mut self, token: u32, query: &str) {
@@ -455,12 +460,18 @@ impl ServerActor {
Some(ClientOperation::ConnectToPeer(peer.clone()))
}
ConnectionType::D => None,
} {
self.client_channel.send(op).unwrap();
} && let Err(e) = self.client_channel.send(op)
{
error!("[server] failed to send ConnectToPeer: {}", e);
}
}
ServerMessage::LoginStatus(message) => {
self.context.write().unwrap().logged_in = Some(message);
match self.context.write_safe() {
Ok(mut ctx) => ctx.logged_in = Some(message),
Err(e) => {
error!("[server] LoginStatus write: {}", e)
}
}
}
ServerMessage::PierceFirewall(token) => {
self.send_message(
@@ -525,9 +536,14 @@ impl ServerActor {
break;
}
if let Some(logged_in) =
context.read().unwrap().logged_in
{
let logged_in = match context.read_safe() {
Ok(ctx) => ctx.logged_in,
Err(e) => {
let _ = response.send(Err(e));
break;
}
};
if let Some(logged_in) = logged_in {
let result = if logged_in {
Ok(true)
} else {