Compare commits

..
10 Commits
Author SHA1 Message Date
dumbgoy 1f0b541bdb Add containing-folder download shortcut to search results 2026-10-27 09:55:24 -04:00
Michel de Graaf 25fa5884ed Merge pull request #104 from michel/michel/chore/clippy-redundant-else
chore: satisfy clippy 1.99
2026-10-26 17:17:22 +01:00
Michel de Graaf 727e2bc8db chore: satisfy clippy 1.99
CI's Lint job runs current stable, which gained two lints that failed
develop: `redundant_else` in `Client::login`, now written with early
returns, and `assert_is_empty`, which asks every `assert!(x.is_empty())`
in the tests to become `assert_eq!(x, [] as [T; 0])`. The second trades a
clear check for a type-cast chore, so it joins the curated allow-list.
2026-10-26 17:16:18 +01:00
Michel de Graaf ae2cc57c30 Merge pull request #103 from michel/michel/fix/brokered-upload-connection
fix(lib): serve uploads to downloaders we cannot dial
2026-10-26 17:08:28 +01:00
Michel de Graaf 3205611609 fix(lib): serve uploads to downloaders we cannot dial
An accepted upload only ever dialled the downloader at the address the
server reported. Most downloaders sit behind a NAT, so the dial timed out
after 20s ("serve …: connection timed out") or, with port 0, the upload
sat in pending_serves forever. There was no fallback: an inbound
PierceFirewall was always treated as a P control connection.

Do what the reference clients do. When an upload starts, ask the server to
have the downloader connect back (ConnectToPeer type F) and dial them at
the same time; the first socket to arrive claims the transfer. Over the
pierced socket only the bare 4-byte transfer token precedes the offset
exchange. CantConnectToPeer and the brokered-connect timeout fail the
upload and free its slot; port 0 skips the dial instead of parking the job.

Covered by an e2e test where the leecher runs without a listener.

Closes #102
2026-10-26 17:03:23 +01:00
Michel de Graaf 68bc3067c3 Merge pull request #101 from MaikSteiger/develop
feat: save downloads in per-user folders (use_username_dirs). Thanks @MaikSteiger.
2026-10-22 14:19:42 +02:00
Michel de Graaf ea884535a8 fix(tui): apply username folders everywhere and resume in place
Move the sanitizer next to the directory helpers so the CLI download
command and the daemon honour use_username_dirs too, not only the TUI.
Resumed downloads keep the folder they were saved with instead of
nesting the username a second time, and toggling the setting applies
to the next download without a restart. Document the config key.
2026-10-22 14:19:18 +02:00
Maik Steiger 9805b807e5 fix(tui): removed a ton of use_username_dirs references
feat(tui): username is now appended to download_dir before passed as parameter
fix(tui): username is sanitized before appended to path
2026-10-09 13:33:22 +02:00
Maik Steiger 5e979a7425 test(tui): cover username directories setting and regenerate openrpc doc 2026-10-08 22:07:13 +02:00
Maik Steiger e567fa75ed feat(tui): added option for storing downloaded files in username directories 2026-10-08 21:59:23 +02:00
30 changed files with 1344 additions and 126 deletions
+3
View File
@@ -53,6 +53,9 @@ significant_drop_tightening = "allow"
# Subjective nursery lint: the suggested map_or/map_or_else rewrites read worse
# than the explicit if let/match they replace.
option_if_let_else = "allow"
# `assert!(x.is_empty())` in tests is fine; the suggested
# `assert_eq!(x, [] as [T; 0])` only trades a clear check for a type-cast chore.
assert_is_empty = "allow"
# The MD5 round functions use the conventional a/b/c/d register names from the
# RFC 1321 reference implementation.
many_single_char_names = "allow"
+4 -3
View File
@@ -606,6 +606,7 @@ file, in that order of precedence. Flags work before or after the subcommand.
| `--listener-port` | `SOULSEEK_LISTENER_PORT` | `listener_port` | `2234` |
| `--no-listener` / `--listener` | `SOULSEEK_NO_LISTENER` | `disable_listener` | listener on |
| — | — | `accept_children` | off |
| — | — | `use_username_dirs` | off |
| `--max-concurrent-downloads` | `SOULSEEK_MAX_CONCURRENT_DOWNLOADS` | `max_concurrent_downloads` | `20` |
| `--search-timeout` | `SOULSEEK_SEARCH_TIMEOUT` | `search_timeout` | `10` |
| `--daemon ADDR` | `SOULSEEK_DAEMON` | `daemon` | local socket if one is up |
@@ -627,10 +628,10 @@ to them. It is off by default — each child costs a socket and a copy of the
network's whole search stream — and it needs the listener, since a child has to
be able to dial in.
`config get` and `config set` cover the twelve settings the file holds
`config get` and `config set` cover the thirteen settings the file holds
(`username`, `server`, `listener_port`, `disable_listener`, `accept_children`,
`download_dir`, `shared_dirs`, `max_concurrent_downloads`, `search_timeout`,
`password_cmd`, `daemon`, `daemon_token`), and list them back at you when you
`download_dir`, `use_username_dirs`, `shared_dirs`, `max_concurrent_downloads`,
`search_timeout`, `password_cmd`, `daemon`, `daemon_token`), and list them back at you when you
name something
else. An empty string clears a key, and `shared_dirs` takes a comma-separated
list. Waits expressed in seconds (`--search-timeout`, `--timeout`,
+24 -24
View File
@@ -209,33 +209,33 @@ impl Client {
/// control.
pub fn login(&self) -> Result<bool> {
info!("Logging in as {}", self.username);
if let Some(handle) = &self.server_handle
&& !self.cancel.is_cancelled()
{
let (tx, rx) = std::sync::mpsc::channel();
let _ = handle.send(ServerMessage::Login {
username: self.username.clone(),
password: self.password.clone(),
version: self.version,
response: tx,
});
let Some(handle) = &self.server_handle else {
return Err(SoulseekRs::NotConnected);
};
if self.cancel.is_cancelled() {
return Err(SoulseekRs::NotConnected);
}
let (tx, rx) = std::sync::mpsc::channel();
let _ = handle.send(ServerMessage::Login {
username: self.username.clone(),
password: self.password.clone(),
version: self.version,
response: tx,
});
let deadline = Instant::now() + LOGIN_RESPONSE_TIMEOUT;
loop {
match rx.recv_timeout(LOGIN_POLL_INTERVAL) {
Ok(result) => return result,
Err(RecvTimeoutError::Timeout)
if self.cancel.is_cancelled() =>
{
return Err(SoulseekRs::NotConnected);
}
Err(RecvTimeoutError::Timeout)
if Instant::now() < deadline => {}
Err(_) => return Err(SoulseekRs::Timeout),
let deadline = Instant::now() + LOGIN_RESPONSE_TIMEOUT;
loop {
match rx.recv_timeout(LOGIN_POLL_INTERVAL) {
Ok(result) => return result,
Err(RecvTimeoutError::Timeout)
if self.cancel.is_cancelled() =>
{
return Err(SoulseekRs::NotConnected);
}
Err(RecvTimeoutError::Timeout) if Instant::now() < deadline => {
}
Err(_) => return Err(SoulseekRs::Timeout),
}
} else {
Err(SoulseekRs::NotConnected)
}
}
+41 -1
View File
@@ -5,7 +5,7 @@
//! of them having to be widened for the move.
use super::{
ActorSystem, Arc, AtomicUsize, BROKER_CONNECT_TIMEOUT,
ActorSystem, Arc, AtomicBool, AtomicUsize, BROKER_CONNECT_TIMEOUT,
BROWSE_PROTECT_WINDOW, ClientContext, DEFAULT_MAX_PEERS,
DEFAULT_UPLOAD_SLOTS, DownloadStatus, DownloadStore, HashMap, HashSet,
Instant, PENDING_PEER_TTL, Recommendation, RoomEvent, RoomInfo, RoomTicker,
@@ -50,6 +50,7 @@ impl ClientContext {
uploads: HashMap::new(),
active_uploads: HashMap::new(),
pending_serves: HashMap::new(),
brokered_uploads: HashMap::new(),
browse_results: HashMap::new(),
folder_contents: HashMap::new(),
peer_infos: HashMap::new(),
@@ -707,6 +708,45 @@ impl ClientContext {
.map(|(username, _)| username)
}
/// Returns the claim flag the dialling route races against.
pub(crate) fn add_brokered_upload(
&mut self,
connect_token: u32,
token: u32,
real_path: std::path::PathBuf,
) -> Arc<AtomicBool> {
let claimed = Arc::new(AtomicBool::new(false));
self.brokered_uploads.insert(
connect_token,
super::BrokeredUpload {
token,
real_path,
deadline: Instant::now() + BROKER_CONNECT_TIMEOUT,
claimed: claimed.clone(),
},
);
claimed
}
pub(crate) fn take_brokered_upload(
&mut self,
connect_token: u32,
) -> Option<super::BrokeredUpload> {
self.brokered_uploads.remove(&connect_token)
}
/// Uploads whose downloader never connected back in time.
pub(crate) fn take_expired_brokered_uploads(
&mut self,
now: Instant,
) -> Vec<super::BrokeredUpload> {
let (expired, live) = std::mem::take(&mut self.brokered_uploads)
.into_iter()
.partition(|(_, upload)| now >= upload.deadline);
self.brokered_uploads = live;
expired.into_values().collect()
}
pub fn take_expired_connects(&mut self, now: Instant) -> Vec<String> {
let mut expired = Vec::new();
self.pending_connect_tokens
+22
View File
@@ -114,6 +114,24 @@ struct UploadJob {
offered: Option<Instant>,
}
/// An upload waiting for its downloader to connect back to our listener while
/// we also dial them. Whichever socket arrives first claims the transfer.
pub(crate) struct BrokeredUpload {
pub(crate) token: u32,
pub(crate) real_path: std::path::PathBuf,
deadline: Instant,
claimed: Arc<AtomicBool>,
}
impl BrokeredUpload {
/// Take the transfer for this route; false if another route already did.
pub(crate) fn claim(&self) -> bool {
self.claimed
.compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire)
.is_ok()
}
}
/// Live bookkeeping for an upload being served (or recently finished).
struct ActiveUpload {
username: String,
@@ -504,6 +522,10 @@ pub struct ClientContext {
active_uploads: HashMap<u32, ActiveUpload>,
/// Upload tokens waiting for the downloader's address to be resolved.
pending_serves: HashMap<String, Vec<u32>>,
/// Uploads whose downloader we could not dial, keyed by the correlation
/// token we sent in `ConnectToPeer(F)`; the peer quotes it back in a
/// `PierceFirewall` and we serve over that socket instead.
brokered_uploads: HashMap<u32, BrokeredUpload>,
/// Shared-file listings received from peers we browsed.
browse_results: HashMap<String, Vec<SharedDirectory>>,
/// One-folder listings received from peers, keyed by peer and folder.
+43 -19
View File
@@ -151,26 +151,26 @@ impl Client {
// its failure is exactly what makes an
// unreachable peer fall back to the
// server-brokered path.
let waiting_serves = if port == 0 {
if port == 0 {
warn!(
"[client] server reports no listening port for {}",
username
);
Vec::new()
} else {
match client_context.write_safe() {
Ok(mut ctx) => {
}
let waiting_serves = match client_context.write_safe() {
Ok(mut ctx) => {
if port != 0 {
ctx.cache_peer_address(
&username,
host.clone(),
port,
);
ctx.pending_serves
.remove(&username)
.unwrap_or_default()
}
Err(_) => Vec::new(),
ctx.pending_serves
.remove(&username)
.unwrap_or_default()
}
Err(_) => Vec::new(),
};
for token in waiting_serves {
Self::spawn_serve(
@@ -622,16 +622,29 @@ impl Client {
// token is the correlation we sent, so it names the
// peer: end the wait now instead of letting every
// queued download for them sit until the timeout.
let username = match client_context.write_safe() {
Ok(mut ctx) => ctx.take_pending_connect(token),
Err(e) => {
error!(
"[client] CantConnectToPeer write: {}",
e
);
None
}
};
let (username, upload) =
match client_context.write_safe() {
Ok(mut ctx) => (
ctx.take_pending_connect(token),
ctx.take_brokered_upload(token),
),
Err(e) => {
error!(
"[client] CantConnectToPeer write: {}",
e
);
(None, None)
}
};
if let Some(upload) = upload
&& upload.claim()
{
Self::fail_upload(
&client_context,
upload.token,
"peer could not connect back",
);
}
if let Some(username) = username {
debug!(
"[client] {} cannot connect back (token {})",
@@ -788,5 +801,16 @@ impl Client {
for username in expired {
Self::fail_queued_downloads(client_context, &username);
}
let expired = match client_context.write_safe() {
Ok(mut ctx) => ctx.take_expired_brokered_uploads(Instant::now()),
Err(_) => return,
};
for upload in expired.iter().filter(|u| u.claim()) {
Self::fail_upload(
client_context,
upload.token,
"peer never connected back",
);
}
}
}
+180 -42
View File
@@ -1,11 +1,14 @@
use super::{
ActiveUpload, Arc, Client, ClientContext, DownloadStatus, PeerMessage,
RwLock, RwLockExt, ServerMessage, collect_failed_tokens, error,
next_upload_token, thread,
ActiveUpload, Arc, BrokeredUpload, Client, ClientContext, DownloadStatus,
PeerMessage, RwLock, RwLockExt, ServerMessage, collect_failed_tokens,
debug, error, next_connect_token, next_upload_token, thread,
};
use crate::message::server::MessageFactory;
use crate::peer::{ConnectionType, upload_peer};
use crate::types::UploadStatus;
use std::sync::atomic::{AtomicBool, AtomicU64};
use std::net::TcpStream;
use std::path::Path;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
impl Client {
/// Whether the server listed `username` as privileged. Privileged peers are
@@ -136,55 +139,190 @@ impl Client {
drop(ctx);
let own = own_username.to_string();
let real_path = job.real_path;
let downloader = job.downloader;
let context = client_context.clone();
thread::spawn(move || {
let result = crate::peer::upload_peer::serve_file(
&host,
port,
// Most downloaders sit behind a NAT, so dialling them is a gamble
// that costs a 20s timeout when it fails. Like the reference
// clients, ask the server to have them connect back at the same
// time; the first socket to arrive claims the transfer, and port 0
// (the server not knowing where they listen) skips the dial.
let Some((connect_token, claimed)) =
Self::broker_upload(&context, token, &real_path, &downloader)
else {
return;
};
if port == 0 {
return;
}
let stream = match upload_peer::dial(&host, port) {
Ok(stream) => stream,
Err(e) => {
debug!(
"[client] cannot dial {downloader} at {host}:{port} \
for upload {token} ({e}); waiting for them to connect"
);
return;
}
};
if claimed
.compare_exchange(
false,
true,
Ordering::AcqRel,
Ordering::Acquire,
)
.is_err()
{
return;
}
if let Ok(mut ctx) = context.write_safe() {
ctx.take_brokered_upload(connect_token);
}
let result = upload_peer::serve_direct(
stream,
&own,
token,
&real_path,
&bytes_sent,
&cancel,
);
let streamed = result.as_ref().ok().copied();
let status = match &result {
Ok(_) => UploadStatus::Completed,
Err(e) if e.kind() == std::io::ErrorKind::Interrupted => {
UploadStatus::Cancelled
}
Err(e) => {
error!("[client] serve {}: {}", real_path.display(), e);
UploadStatus::Failed(e.to_string())
}
};
let mut report = None;
if let Ok(mut ctx) = context.write_safe()
&& let Some(upload) = ctx.active_uploads.get_mut(&token)
{
let secs = upload.started.elapsed().as_secs_f64();
upload.status = status;
if let Some(bytes) = streamed
&& secs > 0.0
{
let speed = (bytes as f64 / secs) as u32;
ctx.last_upload_speed = speed;
report = ctx.server_sender.clone().map(|s| (s, speed));
}
}
// The server folds each finished upload's rate into the average
// speed it shows other users for us.
if let Some((server, speed)) = report {
let _ = server.send(ServerMessage::SendMessage(
MessageFactory::build_send_upload_speed(speed),
));
}
// The slot this transfer held is free now, so whoever is next in
// line gets it without waiting for another request to arrive.
Self::pump_upload_queue(&context);
Self::finish_upload(&context, token, &real_path, result);
});
}
/// Park the upload under a correlation token and ask the server to have
/// `downloader` connect back to us with it. Returns the token and the claim
/// flag the dialling route races against; `None` fails the upload.
fn broker_upload(
context: &Arc<RwLock<ClientContext>>,
token: u32,
real_path: &Path,
downloader: &str,
) -> Option<(u32, Arc<AtomicBool>)> {
let connect_token = next_connect_token();
let (claimed, sender) = match context.write_safe() {
Ok(mut ctx) => (
ctx.add_brokered_upload(
connect_token,
token,
real_path.to_path_buf(),
),
ctx.server_sender.clone(),
),
Err(_) => return None,
};
let Some(sender) = sender else {
Self::fail_upload(context, token, "not connected to the server");
return None;
};
let _ = sender.send(ServerMessage::SendMessage(
MessageFactory::build_connect_to_peer(
connect_token,
downloader,
ConnectionType::F,
),
));
Some((connect_token, claimed))
}
/// The downloader of a brokered upload connected back: stream the file
/// over its socket on a background thread, unless the dial beat it.
pub(crate) fn serve_pierced(
client_context: &Arc<RwLock<ClientContext>>,
upload: BrokeredUpload,
stream: TcpStream,
) {
if !upload.claim() {
return;
}
let Some((bytes_sent, cancel)) =
client_context.write_safe().ok().and_then(|mut ctx| {
let active = ctx.active_uploads.get_mut(&upload.token)?;
active.started = std::time::Instant::now();
Some((active.bytes_sent.clone(), active.cancel.clone()))
})
else {
return;
};
let context = client_context.clone();
thread::spawn(move || {
let result = upload_peer::serve_pierced(
stream,
upload.token,
&upload.real_path,
&bytes_sent,
&cancel,
);
Self::finish_upload(
&context,
upload.token,
&upload.real_path,
result,
);
});
}
/// Mark upload `token` failed and hand its slot to the next in line.
pub(crate) fn fail_upload(
client_context: &Arc<RwLock<ClientContext>>,
token: u32,
reason: &str,
) {
if let Ok(mut ctx) = client_context.write_safe()
&& let Some(upload) = ctx.active_uploads.get_mut(&token)
{
error!("[client] upload {token} to {}: {reason}", upload.username);
upload.status = UploadStatus::Failed(reason.to_string());
}
Self::pump_upload_queue(client_context);
}
/// Record how a served upload ended, report its rate to the server and free
/// its slot.
fn finish_upload(
context: &Arc<RwLock<ClientContext>>,
token: u32,
real_path: &Path,
result: std::io::Result<u64>,
) {
let streamed = result.as_ref().ok().copied();
let status = match &result {
Ok(_) => UploadStatus::Completed,
Err(e) if e.kind() == std::io::ErrorKind::Interrupted => {
UploadStatus::Cancelled
}
Err(e) => {
error!("[client] serve {}: {}", real_path.display(), e);
UploadStatus::Failed(e.to_string())
}
};
let mut report = None;
if let Ok(mut ctx) = context.write_safe()
&& let Some(upload) = ctx.active_uploads.get_mut(&token)
{
let secs = upload.started.elapsed().as_secs_f64();
upload.status = status;
if let Some(bytes) = streamed
&& secs > 0.0
{
let speed = (bytes as f64 / secs) as u32;
ctx.last_upload_speed = speed;
report = ctx.server_sender.clone().map(|s| (s, speed));
}
}
// The server folds each finished upload's rate into the average
// speed it shows other users for us.
if let Some((server, speed)) = report {
let _ = server.send(ServerMessage::SendMessage(
MessageFactory::build_send_upload_speed(speed),
));
}
// The slot this transfer held is free now, so whoever is next in
// line gets it without waiting for another request to arrive.
Self::pump_upload_queue(context);
}
pub(crate) fn process_failed_uploads(
client_context: Arc<RwLock<ClientContext>>,
username: &str,
+12 -1
View File
@@ -261,7 +261,18 @@ fn handle_pierce_firewall(
let token = message.read_int32();
let username = match context.client_context.write_safe() {
Ok(mut ctx) => ctx.take_pending_connect(token),
Ok(mut ctx) => {
if let Some(upload) = ctx.take_brokered_upload(token) {
drop(ctx);
crate::client::Client::serve_pierced(
&context.client_context,
upload,
stream,
);
return;
}
ctx.take_pending_connect(token)
}
Err(e) => {
error!("[listener] pierce firewall lock: {}", e);
return;
+66 -4
View File
@@ -24,6 +24,7 @@ use crate::trace;
///
/// # Errors
/// Returns any I/O error opening the file or talking to the peer.
#[cfg(test)]
pub fn serve_file(
host: &str,
port: u32,
@@ -33,16 +34,28 @@ pub fn serve_file(
bytes_sent: &AtomicU64,
cancel: &AtomicBool,
) -> io::Result<u64> {
let mut file = File::open(path)?;
let stream = dial(host, port)?;
serve_direct(stream, own_username, token, path, bytes_sent, cancel)
}
/// Connect to `host:port` with the transfer deadlines applied.
///
/// # Errors
/// Returns the connect error, or `InvalidInput` for an unresolvable address.
pub fn dial(host: &str, port: u32) -> io::Result<TcpStream> {
let socket = format!("{host}:{port}")
.to_socket_addrs()?
.next()
.ok_or_else(|| {
io::Error::new(io::ErrorKind::InvalidInput, "no address")
})?;
let mut stream =
TcpStream::connect_timeout(&socket, Duration::from_secs(20))?;
with_deadlines(TcpStream::connect_timeout(
&socket,
Duration::from_secs(20),
)?)
}
fn with_deadlines(stream: TcpStream) -> io::Result<TcpStream> {
// A downloader that connects and then goes silent must cost seconds, not
// the rest of the session: without deadlines a wedged `read_exact` (or a
// zero-window `write_all`) holds this thread, the socket, the open file,
@@ -55,7 +68,23 @@ pub fn serve_file(
stream.set_read_timeout(Some(Duration::from_secs(30)))?;
stream.set_write_timeout(Some(Duration::from_secs(30)))?;
stream.set_nodelay(true).ok();
Ok(stream)
}
/// Stream `path` over a connection we dialled: announce with `PeerInit(F)`
/// and the token, then hand over to [`stream_file`].
///
/// # Errors
/// Returns any I/O error opening the file or talking to the peer.
pub fn serve_direct(
mut stream: TcpStream,
own_username: &str,
token: u32,
path: &Path,
bytes_sent: &AtomicU64,
cancel: &AtomicBool,
) -> io::Result<u64> {
let file = File::open(path)?;
// PeerInit(F) + the 4-byte token in a single write so they coalesce.
let mut init = MessageFactory::build_peer_init_message(
own_username,
@@ -66,7 +95,36 @@ pub fn serve_file(
init.extend_from_slice(&token.to_le_bytes());
stream.write_all(&init)?;
stream.flush()?;
stream_file(stream, file, path, bytes_sent, cancel)
}
/// Stream `path` over a connection the downloader opened to us after the
/// server brokered it: it has already sent `PierceFirewall`, so only the
/// 4-byte transfer token precedes the offset exchange.
///
/// # Errors
/// Returns any I/O error opening the file or talking to the peer.
pub fn serve_pierced(
stream: TcpStream,
token: u32,
path: &Path,
bytes_sent: &AtomicU64,
cancel: &AtomicBool,
) -> io::Result<u64> {
let file = File::open(path)?;
let mut stream = with_deadlines(stream)?;
stream.write_all(&token.to_le_bytes())?;
stream.flush()?;
stream_file(stream, file, path, bytes_sent, cancel)
}
fn stream_file(
mut stream: TcpStream,
mut file: File,
path: &Path,
bytes_sent: &AtomicU64,
cancel: &AtomicBool,
) -> io::Result<u64> {
// The downloader replies with an 8-byte START_DOWNLOAD offset first. A
// non-zero offset means it is resuming and already holds that prefix, so
// seek past it and count it as sent to keep progress absolute.
@@ -104,7 +162,11 @@ pub fn serve_file(
// and a peer that leaves the socket open, or whose FIN a NAT swallows,
// would hold it for the whole wait. Nicotine+ finishes the transfer at
// delivery for the same reason; the close below is ours to make.
trace!("[upload] served {} to {}:{}", path.display(), host, port);
trace!(
"[upload] served {} to {:?}",
path.display(),
stream.peer_addr()
);
Ok(streamed)
}
+64 -2
View File
@@ -482,8 +482,8 @@ fn registered_username_can_relogin_with_same_password() {
// ---------------------------------------------------------------------------
// Peer-to-peer download coverage.
//
// The client cannot serve files (the upload side is not implemented), so a real
// download is exercised with a minimal in-process "mock uploader" that speaks
// To exercise the download path on its own, the peer is a minimal in-process
// "mock uploader" rather than a second client. It speaks
// the peer protocol using the library's own public `Message` wire format. It
// drives the same path a real peer would after a search: a `P` control
// connection to our listener, a `QueueUpload` → `TransferRequest` →
@@ -5692,3 +5692,65 @@ fn a_cancel_ends_a_login_waiting_for_its_verdict_and_nothing_follows_it() {
0
);
}
#[test]
fn a_firewalled_downloader_still_receives_the_upload() {
// Issue #102: the leecher cannot be dialled (no listener, as behind a NAT),
// so the sharer must ask the server to broker the F connection and serve
// the file over the leecher's PierceFirewall connection.
let server = server_or_skip!();
let share_dir = unique_download_dir();
let content: Vec<u8> = (0..4096u32).map(|i| (i % 251) as u8).collect();
let filename = "e2e_probe_fwleech.bin";
std::fs::write(share_dir.join(filename), &content).unwrap();
let mut sharer = Client::with_settings(ClientSettings {
shared_directories: vec![share_dir.display().to_string()],
..server.listening_settings(
"e2e_fw_sharer",
"pw",
free_port().expect("sharer port"),
)
});
sharer.connect().expect("sharer connect");
assert!(sharer.login().expect("sharer login"));
let mut leecher =
Client::with_settings(server.settings("e2e_fw_leecher", "pw"));
leecher.connect().expect("leecher connect");
assert!(leecher.login().expect("leecher login"));
std::thread::sleep(Duration::from_secs(1));
let query = "fwleech";
let _ = leecher.search(query, Duration::from_secs(3));
let (result_path, size) = reply_from(&leecher, query, "e2e_fw_sharer")
.and_then(|reply| reply.files.into_iter().next())
.map(|file| (file.name, file.size))
.expect("leecher should find the sharer's file");
let download_dir = unique_download_dir();
let (_download, status_rx) = leecher
.download(
result_path.clone(),
"e2e_fw_sharer".to_string(),
size,
download_dir.display().to_string(),
)
.expect("start download");
let completed =
wait_for_completion(&leecher, &status_rx, Duration::from_secs(40));
assert!(
completed,
"the download to a firewalled leecher should complete"
);
let basename = result_path.rsplit(['\\', '/']).next().unwrap();
assert_eq!(
std::fs::read(download_dir.join(basename)).expect("downloaded file"),
content
);
let _ = std::fs::remove_dir_all(&download_dir);
let _ = std::fs::remove_dir_all(&share_dir);
}
+12
View File
@@ -34,6 +34,8 @@ pub struct Ctx {
pub out: Out,
pub settings: ClientSettings,
pub download_dir: String,
/// Save each peer's files under `download_dir/<username>`.
pub use_username_dirs: bool,
pub max_concurrent_downloads: usize,
pub search_timeout: Duration,
/// Where to find a daemon, when this run should use one rather than open
@@ -47,6 +49,16 @@ pub struct Ctx {
}
impl Ctx {
/// Where a download from `username` lands.
#[must_use]
pub fn dir_for(&self, username: &str) -> String {
crate::directories::download_dir_for(
&self.download_dir,
username,
self.use_username_dirs,
)
}
/// Re-read a "found nothing", "timed out" or "transfer failed" verdict
/// against the session that produced it.
///
+1 -1
View File
@@ -458,7 +458,7 @@ fn download_all(
client.as_ref(),
&ctx.out,
&request,
&ctx.download_dir,
&ctx.dir_for(&request.user),
timeout,
);
if sender.send((request, outcome)).is_err() {
+1
View File
@@ -202,6 +202,7 @@ pub fn run(
browses: Arc::clone(&browses),
downloads: Arc::clone(&downloads),
download_dir: Mutex::new(ctx.download_dir.clone()),
use_username_dirs: ctx.use_username_dirs,
server: ctx.settings.server_address.to_string(),
token: token.clone(),
started: Instant::now(),
+7 -1
View File
@@ -38,6 +38,7 @@ pub struct Daemon {
/// Where transfers land, on *this* machine. A caller may point it at a
/// different folder with `download.set_dir`, but never per download.
pub download_dir: Mutex<String>,
pub use_username_dirs: bool,
pub server: String,
pub token: String,
pub started: Instant,
@@ -428,7 +429,11 @@ impl Daemon {
// per download would scatter transfers in ways `daemon.status` cannot
// report; changing the folder is a deliberate session-wide act,
// `download.set_dir`.
let directory = self.download_dir();
let directory = crate::directories::download_dir_for(
&self.download_dir(),
&params.username,
self.use_username_dirs,
);
let (download, updates) = self
.session
.download_with_metadata(
@@ -800,6 +805,7 @@ mod tests {
browses: Arc::new(PendingBrowses::default()),
downloads: Arc::new(Mutex::new(Vec::new())),
download_dir: Mutex::new("/tmp".to_string()),
use_username_dirs: false,
server: "server.invalid:2416".to_string(),
token: token.to_string(),
started: Instant::now(),
+65 -1
View File
@@ -50,9 +50,54 @@ pub fn resolve_shared_directories(dirs: &[String]) -> Vec<String> {
resolved
}
/// The folder a download from `username` lands in: `download_dir` itself, or
/// a per-user subfolder of it when `per_user` is set, as SoulseekQt does.
#[must_use]
pub fn download_dir_for(
download_dir: &str,
username: &str,
per_user: bool,
) -> String {
if !per_user {
return download_dir.to_string();
}
std::path::Path::new(download_dir)
.join(sanitize_path_component(username))
.display()
.to_string()
}
/// A peer-chosen name made safe as one path component: separators and
/// characters Windows refuses become `_`, so `../..` cannot climb out.
fn sanitize_path_component(name: &str) -> String {
let mut out: String = name
.chars()
.map(|c| match c {
'/' | '\\' | ':' | '*' | '?' | '"' | '<' | '>' | '|' => '_',
c if c.is_control() => '_',
c => c,
})
.collect();
if out.is_empty() || out.ends_with(['.', ' ']) {
out.push('_');
}
if is_reserved(&out) {
out.insert(0, '_');
}
out
}
/// Windows device names (`CON`, `COM1`, …) that cannot be a folder name.
fn is_reserved(name: &str) -> bool {
let stem = name.split('.').next().unwrap_or(name).to_ascii_uppercase();
matches!(stem.as_str(), "CON" | "PRN" | "AUX" | "NUL")
|| (matches!(stem.get(..3), Some("COM" | "LPT"))
&& matches!(stem.as_bytes().get(3..), Some([b'1'..=b'9'])))
}
#[cfg(test)]
mod tests {
use super::resolve_shared_directory;
use super::{download_dir_for, resolve_shared_directory};
use std::sync::atomic::{AtomicU32, Ordering};
fn unique_temp_dir() -> std::path::PathBuf {
@@ -97,4 +142,23 @@ mod tests {
assert!(result.is_err());
let _ = std::fs::remove_dir_all(dir);
}
fn dl(name: &str) -> String {
std::path::Path::new("/dl").join(name).display().to_string()
}
#[test]
fn per_user_folder_is_a_child_of_the_download_dir() {
assert_eq!(download_dir_for("/dl", "alice", false), "/dl");
assert_eq!(download_dir_for("/dl", "alice", true), dl("alice"));
}
#[test]
fn a_hostile_username_cannot_leave_the_download_dir() {
assert_eq!(download_dir_for("/dl", "../..", true), dl(".._.._"));
assert_eq!(download_dir_for("/dl", "/tmp/x", true), dl("_tmp_x"));
assert_eq!(download_dir_for("/dl", "..", true), dl(".._"));
assert_eq!(download_dir_for("/dl", "", true), dl("_"));
assert_eq!(download_dir_for("/dl", "com1", true), dl("_com1"));
}
}
+14
View File
@@ -38,6 +38,7 @@ impl SettingsMode {
pub enum SettingsRow {
ChangePassword,
Logout,
UseUsernameDirs,
DownloadDir,
/// Share path at this index.
Share(usize),
@@ -79,6 +80,7 @@ pub struct SettingsState {
pub account: AccountInfo,
pub download_dir: String,
pub share_dirs: Vec<String>,
pub use_username_dirs: bool,
/// Index into [`Self::rows`].
pub selected: usize,
pub mode: SettingsMode,
@@ -94,11 +96,13 @@ impl SettingsState {
account: AccountInfo,
download_dir: String,
share_dirs: Vec<String>,
use_username_dirs: bool,
) -> Self {
Self {
account,
download_dir,
share_dirs,
use_username_dirs,
selected: 0,
mode: SettingsMode::Navigate,
input: String::new(),
@@ -119,6 +123,7 @@ impl SettingsState {
if self.can_log_out() {
rows.push(SettingsRow::Logout);
}
rows.push(SettingsRow::UseUsernameDirs);
rows.push(SettingsRow::DownloadDir);
rows.extend((0..self.share_dirs.len()).map(SettingsRow::Share));
rows
@@ -191,6 +196,10 @@ impl SettingsState {
self.input.clear();
}
SettingsRow::Logout => self.mode = SettingsMode::ConfirmingLogout,
SettingsRow::UseUsernameDirs => {
self.use_username_dirs = !self.use_username_dirs;
return SettingsAction::Apply;
}
SettingsRow::DownloadDir => {
self.mode = SettingsMode::EditingDownloadDir;
self.input.clone_from(&self.download_dir);
@@ -300,6 +309,7 @@ mod tests {
account(),
"/dl".into(),
vec!["/dl".into(), "/music".into()],
false,
)
}
@@ -493,11 +503,13 @@ mod tests {
},
"/dl".into(),
vec!["/music".into()],
false,
);
assert_eq!(
s.rows(),
vec![
SettingsRow::ChangePassword,
SettingsRow::UseUsernameDirs,
SettingsRow::DownloadDir,
SettingsRow::Share(0),
]
@@ -505,6 +517,8 @@ mod tests {
// The row below "change password" is the download folder, not a
// logout the window cannot perform.
s.handle_key(key(KeyCode::Down));
assert_eq!(s.selected_row(), SettingsRow::UseUsernameDirs);
s.handle_key(key(KeyCode::Down));
assert_eq!(s.selected_row(), SettingsRow::DownloadDir);
}
}
+19
View File
@@ -40,6 +40,7 @@ pub struct FileConfig {
/// A TOML array rather than a `config set` key, because a query may itself
/// contain a comma.
pub wishlist: Option<Vec<String>>,
pub use_username_dirs: Option<bool>,
}
impl FileConfig {
@@ -59,6 +60,7 @@ impl FileConfig {
"password_cmd",
"daemon",
"daemon_token",
"use_username_dirs",
];
/// The value of `key`, or `None` when it is unset or unknown. Lists are
@@ -83,6 +85,9 @@ impl FileConfig {
"password_cmd" => self.password_cmd.clone(),
"daemon" => self.daemon.clone(),
"daemon_token" => self.daemon_token.clone(),
"use_username_dirs" => {
self.use_username_dirs.map(|v| v.to_string())
}
_ => None,
}
}
@@ -162,6 +167,17 @@ impl FileConfig {
// The single-folder spelling would otherwise shadow the list.
self.shared_dir = None;
}
"use_username_dirs" => {
self.use_username_dirs = Some(match value {
"true" | "1" | "yes" => true,
"false" | "0" | "no" => false,
other => {
return Err(format!(
"use_username_dirs wants true or false, got '{other}'"
));
}
});
}
other => {
return Err(format!(
"unknown setting '{other}' — try one of: {}",
@@ -277,6 +293,7 @@ pub struct Resolved {
pub accept_children: bool,
pub download_dir: String,
pub shared_dirs: Vec<String>,
pub use_username_dirs: bool,
pub max_concurrent_downloads: usize,
pub search_timeout: u64,
pub password_cmd: Option<String>,
@@ -330,6 +347,7 @@ pub fn resolve(cli: &crate::cli::Cli, file: &FileConfig) -> Resolved {
&& !disable_listener,
download_dir: download_dir.clone(),
shared_dirs: resolve_shared_dirs(cli, file, &download_dir),
use_username_dirs: file.use_username_dirs.unwrap_or(false),
max_concurrent_downloads: cli
.max_concurrent_downloads
.or(file.max_concurrent_downloads)
@@ -573,6 +591,7 @@ mod tests {
daemon: Some("nas.local:5030".into()),
daemon_token: Some("deadbeef".into()),
wishlist: None,
use_username_dirs: Some(false),
};
let resolved = resolve(&bare_cli(), &file);
assert_eq!(resolved.username.as_deref(), Some("alice"));
+3
View File
@@ -276,6 +276,7 @@ fn context(
version: ClientVersion::REFERENCE_CLIENT,
},
download_dir,
use_username_dirs: resolved.use_username_dirs,
max_concurrent_downloads: resolved.max_concurrent_downloads,
search_timeout: Duration::from_secs(resolved.search_timeout),
daemon,
@@ -522,6 +523,7 @@ fn run_default_tui(
terminal,
Arc::new(outcome.client),
resolved.download_dir.clone(),
resolved.use_username_dirs,
Duration::from_secs(resolved.search_timeout),
store,
config_path.clone(),
@@ -598,6 +600,7 @@ fn run_attached_tui(
terminal,
Arc::new(session),
download_dir,
resolved.use_username_dirs,
Duration::from_secs(resolved.search_timeout),
None,
config_path,
+263 -8
View File
@@ -2,12 +2,96 @@ use super::MainTui;
use super::input::{FilterEdit, edit_filter, jumped, list_jump};
use crate::models::BrowseStatus;
use ratatui::crossterm::event::{KeyCode, KeyEvent, KeyModifiers};
use std::{thread, time::Duration};
use std::{
collections::HashSet,
sync::mpsc::{self, Receiver, TryRecvError},
thread,
time::{Duration, Instant},
};
/// How long to wait for a browse response before showing a timeout notice.
const BROWSE_TIMEOUT: Duration = Duration::from_secs(20);
/// One full peer listing can satisfy several containing-folder requests.
pub(super) struct FolderFetch {
username: String,
folders: Vec<String>,
pub(super) requested_at: Instant,
request: Option<Receiver<Result<(), String>>>,
}
/// Soulseek paths use remote separators, independently of the local OS.
pub(super) fn containing_folder(path: &str) -> Option<&str> {
let (parent, name) = path.rsplit_once('\\')?;
if parent.trim_matches('\\').is_empty()
|| name.is_empty()
|| path.contains('/')
|| path.chars().any(char::is_control)
|| path
.trim_start_matches('\\')
.split('\\')
.any(|part| matches!(part, "" | "." | ".."))
{
return None;
}
Some(parent)
}
impl MainTui {
pub(super) fn download_containing_folder(&mut self) {
let Some(file) = self.highlighted_result() else {
return;
};
let Some(folder) = containing_folder(&file.filename) else {
self.folder_download_status =
Some("Cannot download folder: invalid remote file path".into());
return;
};
let username = file.username.clone();
let folder = folder.to_string();
if username.trim().is_empty() {
self.folder_download_status =
Some("Cannot download folder: missing peer username".into());
return;
}
self.folder_download_status =
Some(format!("Fetching folder {folder} from {username}…"));
if let Some(fetch) = self
.folder_fetches
.iter_mut()
.find(|fetch| fetch.username == username)
{
if !fetch.folders.contains(&folder) {
fetch.folders.push(folder);
}
return;
}
let (sender, request) = mpsc::channel();
let client = self.client.clone();
let peer = username.clone();
// SessionApi's browse works for both local and daemon sessions. The
// request itself can involve IPC, so even that runs off the UI thread.
if self.state.browse.tabs.iter().any(|tab| {
!tab.own
&& tab.username == username
&& tab.status == BrowseStatus::Loading
}) {
let _ = sender.send(Ok(()));
} else {
thread::spawn(move || {
let _ = sender
.send(client.browse_user(&peer).map_err(|e| e.to_string()));
});
}
self.folder_fetches.push(FolderFetch {
username,
folders: vec![folder],
requested_at: Instant::now(),
request: Some(request),
});
}
pub(super) fn handle_browse_input(&mut self, key: KeyEvent) {
// Typing goes to the filter; the other keys still move around.
if self.state.browse.active_tab().is_some_and(|b| b.filtering)
@@ -218,10 +302,24 @@ impl MainTui {
if files.is_empty() {
return;
}
self.queue_peer_files(username, files, self.download_dir.clone());
}
/// Shared browse queue: keep remote paths intact and let the session's
/// download machinery choose local names and enforce transfer limits.
fn queue_peer_files(
&mut self,
username: String,
files: Vec<(String, u64)>,
download_dir: String,
) -> Receiver<String> {
let sender = self.downloads_sender();
let client = self.client.clone();
let download_dir = self.download_dir.clone();
let (status_sender, status_receiver) = mpsc::channel();
thread::spawn(move || {
let mut queued = 0;
let mut failed = 0;
let mut last_error = String::new();
for (path, size) in files {
match client.download(
path.clone(),
@@ -231,15 +329,27 @@ impl MainTui {
) {
Ok((download, rx)) => {
let _ = sender.send((download, rx));
queued += 1;
}
Err(e) => {
failed += 1;
last_error = e.to_string();
soulseek_rs::warn!(
"Failed to queue download {path}: {e}"
);
}
}
}
let status = if failed == 0 {
format!("Queued {queued} files from {username}")
} else {
format!(
"Queued {queued} files from {username}; {failed} failed: {last_error}"
)
};
let _ = status_sender.send(status);
});
status_receiver
}
/// Request a user's shared files and open (or focus) their browse tab.
@@ -250,7 +360,9 @@ impl MainTui {
}
// Open/focus the tab; only (re)issue the request when it's new or a
// previous attempt timed out.
if self.state.browse.open(&username) {
if self.state.browse.open(&username)
&& !self.folder_fetches.iter().any(|f| f.username == username)
{
let _ = self.client.browse_user(&username);
}
self.state.show_browse = true;
@@ -289,6 +401,7 @@ impl MainTui {
/// Drain browse responses into any loading tabs, or time them out.
pub(super) fn poll_browse_result(&mut self) {
self.poll_folder_requests();
// Which loading tabs are waiting, and for whom.
let loading: Vec<(usize, String, std::time::Instant)> = self
.state
@@ -300,20 +413,162 @@ impl MainTui {
.map(|(i, b)| (i, b.username.clone(), b.requested_at))
.collect();
for (idx, username, requested_at) in loading {
// take_browse_result is destructive. Consume once per peer, then
// deliver the listing to both browse tabs and folder downloads.
let users: HashSet<String> = loading
.iter()
.map(|(_, user, _)| user.clone())
.chain(self.folder_fetches.iter().map(|f| f.username.clone()))
.collect();
for username in users {
if let Some(directories) = self.client.take_browse_result(&username)
{
if let Some(browse) = self.state.browse.tabs.get_mut(idx) {
for (idx, _, _) in
loading.iter().filter(|(_, user, _)| *user == username)
{
let browse = &mut self.state.browse.tabs[*idx];
browse.load(&directories);
if *idx == self.state.browse.active {
self.state.browse_table_state.select(Some(0));
}
}
if idx == self.state.browse.active {
self.state.browse_table_state.select(Some(0));
if let Some(index) = self
.folder_fetches
.iter()
.position(|f| f.username == username)
{
let fetch = self.folder_fetches.remove(index);
for folder in fetch.folders {
self.queue_containing_folder(
&username,
&folder,
&directories,
);
}
}
} else if requested_at.elapsed() > BROWSE_TIMEOUT
}
}
for (idx, _, requested_at) in loading {
if requested_at.elapsed() > BROWSE_TIMEOUT
&& let Some(browse) = self.state.browse.tabs.get_mut(idx)
&& browse.status == BrowseStatus::Loading
{
browse.status = BrowseStatus::TimedOut;
}
}
}
fn queue_containing_folder(
&mut self,
username: &str,
folder: &str,
directories: &[soulseek_rs::SharedDirectory],
) {
let files = match files_in_folder(directories, folder) {
Ok(files) if !files.is_empty() => files,
Ok(_) => {
self.folder_download_status =
Some(format!("Folder {folder} from {username} is empty"));
return;
}
Err(error) => {
self.folder_download_status = Some(format!(
"Cannot download {folder} from {username}: {error}"
));
return;
}
};
self.folder_download_status = Some(format!(
"Queuing {} files from {username}: {folder}…",
files.len()
));
let directory = crate::directories::download_dir_for(
&self.download_dir,
username,
self.use_username_dirs,
);
let result =
self.queue_peer_files(username.to_string(), files, directory);
self.folder_queue_results.push(result);
}
fn poll_folder_requests(&mut self) {
self.folder_queue_results
.retain(|result| match result.try_recv() {
Ok(status) => {
self.folder_download_status = Some(status);
false
}
Err(TryRecvError::Empty) => true,
Err(TryRecvError::Disconnected) => {
self.folder_download_status =
Some("Folder download worker stopped".into());
false
}
});
self.folder_fetches.retain_mut(|fetch| {
if let Some(request) = &fetch.request {
let error = match request.try_recv() {
Ok(Ok(())) => {
fetch.request = None;
None
}
Ok(Err(error)) => Some(error),
Err(TryRecvError::Disconnected) => {
Some("request worker stopped".into())
}
Err(TryRecvError::Empty) => None,
};
if let Some(error) = error {
self.folder_download_status = Some(format!(
"Could not fetch folder from {}: {error}",
fetch.username
));
return false;
}
}
if fetch.requested_at.elapsed() > BROWSE_TIMEOUT {
self.folder_download_status = Some(format!(
"Folder request to {} timed out; press D to retry",
fetch.username
));
return false;
}
true
});
}
}
/// Only direct children of an exact directory match, never descendants or
/// prefix matches. Reject malformed entries rather than queue a partial folder.
fn files_in_folder(
directories: &[soulseek_rs::SharedDirectory],
folder: &str,
) -> Result<Vec<(String, u64)>, &'static str> {
let mut found = false;
let mut seen = HashSet::new();
let mut files = Vec::new();
for dir in directories
.iter()
.filter(|dir| dir.name.trim_end_matches('\\') == folder)
{
found = true;
for file in &dir.files {
if file.name.is_empty()
|| file.name.contains(['\\', '/'])
|| file.name.chars().any(char::is_control)
|| matches!(file.name.as_str(), "." | "..")
{
return Err("invalid file name in peer listing");
}
let path = format!("{folder}\\{}", file.name);
if seen.insert(path.clone()) {
files.push((path, file.size));
}
}
}
if !found {
return Err("folder not found in peer listing");
}
Ok(files)
}
+7 -1
View File
@@ -1,4 +1,5 @@
use super::MainTui;
use crate::directories::download_dir_for;
use crate::models::{DownloadEntry, FileDisplayData};
use crate::ui::panes::{InfoSubject, selected_transfer};
use soulseek_rs::types::UploadStatus;
@@ -197,6 +198,7 @@ impl MainTui {
let sender = self.downloads_sender();
let client = self.client.clone();
let download_dir = self.download_dir.clone();
let use_username_dirs = self.use_username_dirs;
thread::spawn(move || {
for file in selected_files {
@@ -210,7 +212,11 @@ impl MainTui {
file.filename.clone(),
file.username.clone(),
file.size,
download_dir.clone(),
download_dir_for(
&download_dir,
&file.username,
use_username_dirs,
),
metadata,
) {
Ok((download, rx)) => {
+1
View File
@@ -484,6 +484,7 @@ impl MainTui {
KeyCode::Enter => {
self.queue_selected_downloads();
}
KeyCode::Char('D') => self.download_containing_folder(),
_ => {}
}
}
+19 -2
View File
@@ -23,6 +23,7 @@ pub struct MainTui {
client: Arc<dyn SessionApi>,
state: AppState,
download_dir: String,
use_username_dirs: bool,
search_timeout: Duration,
spinner_state: usize,
store: Option<StateStore>,
@@ -34,12 +35,16 @@ pub struct MainTui {
/// Which divider the left mouse button is holding, if any.
resize_drag: Option<input::ResizeDrag>,
listener_fallback: Option<(u16, u16)>,
folder_fetches: Vec<browse::FolderFetch>,
folder_queue_results: Vec<std::sync::mpsc::Receiver<String>>,
folder_download_status: Option<String>,
}
impl MainTui {
pub fn new(
client: Arc<dyn SessionApi>,
download_dir: String,
use_username_dirs: bool,
search_timeout: Duration,
store: Option<StateStore>,
config_path: Option<std::path::PathBuf>,
@@ -49,12 +54,16 @@ impl MainTui {
state: AppState::new(),
download_dir,
search_timeout,
use_username_dirs,
spinner_state: 0,
store,
config_path,
saved_snapshot: Snapshot::default(),
resize_drag: None,
listener_fallback: None,
folder_fetches: Vec::new(),
folder_queue_results: Vec::new(),
folder_download_status: None,
};
tui.restore_persisted_state();
tui
@@ -252,17 +261,25 @@ impl MainTui {
}
}
#[expect(clippy::too_many_arguments, reason = "one call site per mode")]
pub fn launch_main_tui(
terminal: DefaultTerminal,
client: Arc<dyn SessionApi>,
download_dir: String,
use_username_dirs: bool,
search_timeout: Duration,
store: Option<StateStore>,
config_path: Option<std::path::PathBuf>,
listener_fallback: Option<(u16, u16)>,
) -> Result<TuiExit> {
let mut tui =
MainTui::new(client, download_dir, search_timeout, store, config_path);
let mut tui = MainTui::new(
client,
download_dir,
use_username_dirs,
search_timeout,
store,
config_path,
);
tui.listener_fallback = listener_fallback;
tui.run(terminal)
}
+10
View File
@@ -369,6 +369,16 @@ impl MainTui {
lines.push(Line::from(""));
lines.push(section("Folders"));
lines.push(entry(
selected == SettingsRow::UseUsernameDirs,
"Use username directory ",
if settings.use_username_dirs {
"[x]"
} else {
"[ ]"
}
.to_string(),
));
lines.push(if settings.mode == SettingsMode::EditingDownloadDir {
typing("Download folder: ", settings.input.clone())
} else {
@@ -79,6 +79,7 @@ const HELP_RIGHT: &[(&str, &[(&str, &str)])] = &[
("Space", "select / deselect"),
("a / A", "select all / none"),
("Enter", "download the selection"),
("D", "download highlighted file's containing folder"),
("/", "filter, Enter keeps it, Esc clears"),
("b", "browse the owner"),
("c", "chat rooms"),
@@ -166,6 +167,7 @@ fn settings_shortcuts(
SettingsRow::ChangePassword => ("Enter", "change password"),
SettingsRow::Logout => ("Enter", "log out"),
SettingsRow::DownloadDir => ("Enter", "edit folder"),
SettingsRow::UseUsernameDirs => ("Enter", "toggle option"),
SettingsRow::Share(_) => ("d", "remove share"),
},
("a", "add share"),
@@ -396,6 +398,7 @@ impl MainTui {
FocusedPane::Results => vec![
("Space", "select"),
("Enter", "download"),
("D", "download containing folder"),
("b", "browse owner"),
("c", "chat"),
("/", "filter"),
@@ -425,7 +428,12 @@ impl MainTui {
// Borders and padding, and the unread badge's column when it shows.
let badge = self.unread_badge().width() as u16;
let reserved = 4 + if badge > 0 { badge + 1 } else { 0 };
pack_shortcuts(&self.shortcuts(), width.saturating_sub(reserved))
let mut rows =
pack_shortcuts(&self.shortcuts(), width.saturating_sub(reserved));
if let Some(status) = &self.folder_download_status {
rows.push(Line::styled(status.clone(), info_style()));
}
rows
}
/// A prominent unread badge for the shortcuts bar. It rides in the
+4
View File
@@ -20,6 +20,7 @@ impl MainTui {
account,
self.download_dir.clone(),
self.client.shared_directories(),
self.use_username_dirs,
));
}
@@ -66,6 +67,7 @@ impl MainTui {
};
let download_dir = settings.download_dir.clone();
let share_dirs = settings.share_dirs.clone();
let use_username_dirs = settings.use_username_dirs;
let shares = if self.client.daemon_endpoint().is_some() {
share_dirs.clone()
@@ -107,6 +109,7 @@ impl MainTui {
if dir_applied {
self.download_dir.clone_from(&download_dir);
}
self.use_username_dirs = use_username_dirs;
// Persist to config.toml so the change survives a restart. What was
// refused live is not written either: the file must not promise a
@@ -121,6 +124,7 @@ impl MainTui {
// load, so leaving it would resurrect a removed folder.
config.shared_dir = None;
config.shared_dirs = Some(share_dirs);
config.use_username_dirs = Some(use_username_dirs);
config.save(&path)
});
if let Err(e) = result {
+360
View File
@@ -2,6 +2,366 @@
use super::*;
fn directory(
name: &str,
files: &[(&str, u64)],
) -> soulseek_rs::SharedDirectory {
soulseek_rs::SharedDirectory {
name: name.to_string(),
files: files
.iter()
.map(|(name, size)| soulseek_rs::SharedFileEntry {
name: (*name).to_string(),
size: *size,
attributes: Vec::new(),
})
.collect(),
}
}
fn folder_window(session: Arc<TalkativeSession>) -> MainTui {
let mut tui = attach(session);
tui.state.focused_pane = FocusedPane::Results;
tui.state.results_items = results(2);
tui.state.results_items[0].filename =
"Music\\Artist\\Album\\03 - Track.flac".into();
tui.state.results_items[1].filename = "Other\\track.mp3".into();
tui.state.results_items[1].username = "alice".into();
tui.state.results_table_state.select(Some(0));
tui
}
fn wait_for(mut ready: impl FnMut() -> bool) {
let deadline = std::time::Instant::now() + Duration::from_secs(5);
while !ready() {
assert!(
std::time::Instant::now() < deadline,
"worker did not finish"
);
std::thread::sleep(Duration::from_millis(5));
}
}
fn finish_folder(tui: &mut MainTui) {
wait_for(|| {
tui.poll_session();
tui.folder_fetches.is_empty() && tui.folder_queue_results.is_empty()
});
tui.update_downloads();
}
fn answer(
session: &TalkativeSession,
directories: Vec<soulseek_rs::SharedDirectory>,
) {
wait_for(|| !session.browse_requests.lock().unwrap().is_empty());
session
.browse_results
.lock()
.unwrap()
.insert("bob".into(), directories);
}
#[test]
fn containing_folder_uses_soulseek_paths_and_rejects_malformed_paths() {
use super::super::browse::containing_folder;
for (path, parent) in [
(
"Music\\Artist\\Album\\03 - Track.flac",
"Music\\Artist\\Album",
),
("@@share\\song.mp3", "@@share"),
("\\Music\\Björk\\曲.flac", "\\Music\\Björk"),
("C:\\Music\\song.mp3", "C:\\Music"),
] {
assert_eq!(containing_folder(path), Some(parent), "{path}");
}
for path in [
"",
"track.mp3",
"\\track.mp3",
"Music\\",
"Music\\..\\song.mp3",
"Music\\.\\song.mp3",
"Music\\\\song.mp3",
"Music\\bad\0.mp3",
"Music/track.mp3",
] {
assert_eq!(containing_folder(path), None, "{path:?}");
}
}
#[test]
fn shift_d_downloads_only_highlighted_folder_from_complete_peer_listing() {
for per_user in [false, true] {
let (release, gate) = std::sync::mpsc::channel();
let session = Arc::new(TalkativeSession {
accept_downloads: true,
browse_gate: Some(std::sync::Mutex::new(gate)),
..Default::default()
});
let mut tui = folder_window(session.clone());
tui.use_username_dirs = per_user;
tui.state.results_selected_indices.insert(1);
// The kept filter maps the highlighted row to the original result.
tui.state.results_items.swap(0, 1);
tui.state.results_filter_query = "Track.flac".into();
tui.apply_filter();
tui.state.results_selected_indices.clear();
tui.state.results_selected_indices.insert(0);
tui.handle_key_event(KeyEvent::new(
KeyCode::Char('D'),
KeyModifiers::SHIFT,
));
// The worker is held at the request: navigation and rendering still run.
press(&mut tui, KeyCode::Tab);
assert_ne!(tui.state.focused_pane, FocusedPane::Results);
assert!(screen_of(&mut tui).contains("Fetching folder"));
assert!(!tui.state.show_browse);
assert!(tui.state.browse.tabs.is_empty());
assert!(session.download_requests.lock().unwrap().is_empty());
release.send(()).unwrap();
answer(
&session,
vec![
directory("Music\\Artist", &[("parent.mp3", 1)]),
directory(
"Music\\Artist\\Album\\",
&[
("03 - Track.flac", 30),
("01 - Missing.flac", 10),
("cover.jpg", 20),
],
),
directory("Music\\Artist\\Album\\Disc 2", &[("child.flac", 2)]),
directory(
"Music\\Artist\\Album extras",
&[("sibling.flac", 3)],
),
directory("Other", &[("track.mp3", 4)]),
],
);
finish_folder(&mut tui);
let downloads = session.download_requests.lock().unwrap();
assert_eq!(downloads.len(), 3);
for (download, name, size) in [
(&downloads[0], "03 - Track.flac", 30),
(&downloads[1], "01 - Missing.flac", 10),
(&downloads[2], "cover.jpg", 20),
] {
assert_eq!(download.username, "bob");
assert_eq!(
download.filename,
format!("Music\\Artist\\Album\\{name}")
);
assert_eq!(download.size, size);
assert_eq!(
download.download_directory,
if per_user { "/tmp/bob" } else { "/tmp" }
);
}
assert_eq!(tui.state.downloads.len(), 3);
assert_eq!(tui.state.results_selected_indices, [0].into());
assert!(screen_of(&mut tui).contains("Queued 3 files from bob"));
}
}
#[test]
fn enter_still_downloads_selected_files_and_b_still_opens_the_owner() {
let session = Arc::new(TalkativeSession {
accept_downloads: true,
..Default::default()
});
let mut tui = folder_window(session.clone());
tui.state.results_selected_indices.insert(1);
press(&mut tui, KeyCode::Enter);
wait_for(|| !session.download_requests.lock().unwrap().is_empty());
let requests = session.download_requests.lock().unwrap();
assert_eq!(requests.len(), 1);
assert_eq!(requests[0].username, "alice");
assert_eq!(requests[0].filename, "Other\\track.mp3");
assert!(tui.state.results_selected_indices.is_empty());
assert!(session.browse_requests.lock().unwrap().is_empty());
drop(requests);
press(&mut tui, KeyCode::Char('b'));
assert!(tui.state.show_browse);
assert_eq!(tui.state.browse.active_tab().unwrap().username, "bob");
assert_eq!(*session.browse_requests.lock().unwrap(), ["bob"]);
assert!(tui.folder_fetches.is_empty());
}
#[test]
fn d_without_a_highlight_or_outside_results_does_nothing() {
let session = Arc::new(TalkativeSession::default());
let mut tui = folder_window(session.clone());
for selection in [None, Some(99)] {
tui.state.results_table_state.select(selection);
press(&mut tui, KeyCode::Char('D'));
assert!(tui.folder_fetches.is_empty());
assert!(tui.folder_download_status.is_none());
}
tui.state.results_items.clear();
tui.state.results_table_state.select(Some(0));
press(&mut tui, KeyCode::Char('D'));
assert!(tui.folder_fetches.is_empty());
tui.state.results_items = results(1);
for pane in [FocusedPane::Searches, FocusedPane::Downloads] {
tui.state.focused_pane = pane;
press(&mut tui, KeyCode::Char('D'));
assert!(tui.folder_fetches.is_empty());
}
assert!(session.browse_requests.lock().unwrap().is_empty());
}
#[test]
fn d_is_text_while_editing_a_filter_and_ctrl_d_still_pages() {
let mut tui = results_tui(40);
tui.state.results_table_state.select(Some(0));
ctrl(&mut tui, 'd');
assert!(selected(&tui).unwrap() > 0);
press(&mut tui, KeyCode::Char('/'));
press(&mut tui, KeyCode::Char('D'));
assert_eq!(tui.state.results_filter_query, "D");
assert!(tui.folder_fetches.is_empty());
}
#[test]
fn failed_requests_and_timeouts_report_errors_and_allow_retry() {
let session = Arc::new(TalkativeSession {
browse_fails: true,
..Default::default()
});
let mut tui = folder_window(session.clone());
press(&mut tui, KeyCode::Char('D'));
finish_folder(&mut tui);
assert!(screen_of(&mut tui).contains("Could not fetch folder from bob"));
assert!(session.download_requests.lock().unwrap().is_empty());
let session = Arc::new(TalkativeSession::default());
let mut tui = folder_window(session.clone());
press(&mut tui, KeyCode::Char('D'));
wait_for(|| !session.browse_requests.lock().unwrap().is_empty());
tui.folder_fetches[0].requested_at = std::time::Instant::now()
.checked_sub(Duration::from_secs(21))
.expect("a time before the browse timeout");
finish_folder(&mut tui);
assert!(screen_of(&mut tui).contains("timed out"));
// A late response must not queue anything after the timeout.
answer(
&session,
vec![directory("Music\\Artist\\Album", &[("late.flac", 1)])],
);
tui.poll_session();
assert!(session.download_requests.lock().unwrap().is_empty());
press(&mut tui, KeyCode::Char('D'));
wait_for(|| session.browse_requests.lock().unwrap().len() == 2);
assert_eq!(tui.folder_fetches.len(), 1);
}
#[test]
fn empty_missing_and_malformed_folders_do_not_queue_files() {
for (listing, message) in [
(vec![directory("Music\\Artist\\Album", &[])], "is empty"),
(vec![], "folder not found"),
(
vec![directory(
"Music\\Artist\\Album\\Child",
&[("child.flac", 1)],
)],
"folder not found",
),
(
vec![directory(
"Music\\Artist\\Album",
&[("valid.flac", 1), ("..\\escape.flac", 2)],
)],
"invalid file name",
),
] {
let session = Arc::new(TalkativeSession {
accept_downloads: true,
..Default::default()
});
let mut tui = folder_window(session.clone());
press(&mut tui, KeyCode::Char('D'));
answer(&session, listing);
finish_folder(&mut tui);
assert!(
tui.folder_download_status
.as_ref()
.unwrap()
.contains(message)
);
assert!(session.download_requests.lock().unwrap().is_empty());
}
let session = Arc::new(TalkativeSession::default());
let mut tui = folder_window(session.clone());
tui.state.results_items[0].filename = "track.flac".into();
press(&mut tui, KeyCode::Char('D'));
assert!(screen_of(&mut tui).contains("invalid remote file path"));
assert!(tui.folder_fetches.is_empty());
assert!(session.browse_requests.lock().unwrap().is_empty());
}
#[test]
fn download_queue_failures_are_reported() {
let session = Arc::new(TalkativeSession::default());
let mut tui = folder_window(session.clone());
press(&mut tui, KeyCode::Char('D'));
answer(
&session,
vec![directory(
"Music\\Artist\\Album",
&[("a.flac", 1), ("b.flac", 2)],
)],
);
finish_folder(&mut tui);
assert!(screen_of(&mut tui).contains("Queued 0 files from bob; 2 failed"));
assert!(tui.state.downloads.is_empty());
}
#[test]
fn a_listing_serves_both_browse_and_folder_download_without_duplicate_requests()
{
for browse_first in [false, true] {
let session = Arc::new(TalkativeSession {
accept_downloads: true,
..Default::default()
});
let mut tui = folder_window(session.clone());
if browse_first {
press(&mut tui, KeyCode::Char('b'));
press(&mut tui, KeyCode::Esc);
}
press(&mut tui, KeyCode::Char('D'));
press(&mut tui, KeyCode::Char('D'));
assert_eq!(tui.folder_fetches.len(), 1);
press(&mut tui, KeyCode::Char('b'));
answer(
&session,
vec![directory("Music\\Artist\\Album", &[("a.flac", 1)])],
);
finish_folder(&mut tui);
assert_eq!(*session.browse_requests.lock().unwrap(), ["bob"]);
assert_eq!(
tui.state.browse.active_tab().unwrap().status,
crate::models::BrowseStatus::Loaded
);
assert_eq!(session.download_requests.lock().unwrap().len(), 1);
}
}
#[test]
fn help_documents_the_containing_folder_key() {
let mut tui = results_tui(1);
press(&mut tui, KeyCode::Char('?'));
let screen = screen_sized(&mut tui, 180, 60);
let row =
row_with(&screen, "download highlighted file's containing folder");
assert!(row.contains('D'));
}
#[test]
fn shift_b_shows_what_this_session_shares_and_r_re_indexes_it() {
let session = Arc::new(TalkativeSession::default());
+57 -15
View File
@@ -51,6 +51,14 @@ struct TalkativeSession {
download_cancelled: std::sync::Mutex<Vec<(String, String)>>,
upload_cancelled: std::sync::Mutex<Vec<(String, String)>>,
download_removed: std::sync::Mutex<Vec<(String, String)>>,
browse_requests: std::sync::Mutex<Vec<String>>,
browse_results: std::sync::Mutex<
std::collections::HashMap<String, Vec<soulseek_rs::SharedDirectory>>,
>,
browse_fails: bool,
browse_gate: Option<std::sync::Mutex<std::sync::mpsc::Receiver<()>>>,
accept_downloads: bool,
download_requests: std::sync::Mutex<Vec<soulseek_rs::types::Download>>,
}
fn queued(username: &str, filename: &str) -> soulseek_rs::types::Download {
@@ -156,28 +164,44 @@ impl SessionApi for TalkativeSession {
}
fn download(
&self,
_filename: String,
_username: String,
_size: u64,
_directory: String,
filename: String,
username: String,
size: u64,
directory: String,
) -> soulseek_rs::Result<(
soulseek_rs::types::Download,
std::sync::mpsc::Receiver<soulseek_rs::DownloadStatus>,
)> {
Err(soulseek_rs::SoulseekRs::NotConnected)
let mut download = queued(&username, &filename);
download.size = size;
download.download_directory = directory;
self.download_requests
.lock()
.expect("not poisoned")
.push(download.clone());
if self.accept_downloads {
Ok((download, std::sync::mpsc::channel().1))
} else {
Err(soulseek_rs::SoulseekRs::NotConnected)
}
}
fn download_with_metadata(
&self,
_filename: String,
_username: String,
_size: u64,
_directory: String,
_metadata: soulseek_rs::types::DownloadMetadata,
filename: String,
username: String,
size: u64,
directory: String,
metadata: soulseek_rs::types::DownloadMetadata,
) -> soulseek_rs::Result<(
soulseek_rs::types::Download,
std::sync::mpsc::Receiver<soulseek_rs::DownloadStatus>,
)> {
Err(soulseek_rs::SoulseekRs::NotConnected)
self.download(filename, username, size, directory).map(
|(mut download, receiver)| {
download.metadata = metadata;
(download, receiver)
},
)
}
fn get_all_downloads(&self) -> Vec<soulseek_rs::types::Download> {
self.downloads.clone()
@@ -263,14 +287,31 @@ impl SessionApi for TalkativeSession {
fn take_private_messages(&self) -> Vec<soulseek_rs::UserMessage> {
Vec::new()
}
fn browse_user(&self, _username: &str) -> soulseek_rs::Result<()> {
Ok(())
fn browse_user(&self, username: &str) -> soulseek_rs::Result<()> {
self.browse_requests
.lock()
.expect("not poisoned")
.push(username.to_string());
if let Some(gate) = &self.browse_gate {
gate.lock()
.expect("not poisoned")
.recv_timeout(Duration::from_secs(5))
.expect("release browse request");
}
if self.browse_fails {
Err(soulseek_rs::SoulseekRs::NotConnected)
} else {
Ok(())
}
}
fn take_browse_result(
&self,
_username: &str,
username: &str,
) -> Option<Vec<soulseek_rs::SharedDirectory>> {
None
self.browse_results
.lock()
.expect("not poisoned")
.remove(username)
}
fn request_user_info(&self, _username: &str) -> soulseek_rs::Result<()> {
Ok(())
@@ -349,6 +390,7 @@ fn attach(session: Arc<TalkativeSession>) -> MainTui {
MainTui::new(
session,
"/tmp".to_string(),
false,
Duration::from_secs(1),
None,
None,
@@ -17,6 +17,7 @@ fn applying_settings_attached_pushes_both_folders_to_the_daemon() {
let mut tui = MainTui::new(
session.clone(),
"/daemon/old".to_string(),
false,
Duration::from_secs(1),
None,
Some(config_path.clone()),
@@ -73,6 +74,7 @@ fn applying_settings_locally_stays_on_this_machine() {
let mut tui = MainTui::new(
session.clone(),
"/tmp".to_string(),
false,
Duration::from_secs(1),
None,
Some(config_path),
@@ -106,6 +108,30 @@ fn applying_settings_locally_stays_on_this_machine() {
assert_eq!(tui.download_dir, downloads_dir);
}
#[test]
fn toggling_username_folders_applies_live_and_persists() {
let session = Arc::new(TalkativeSession::default());
let scratch = tempfile::tempdir().expect("a temp dir");
let config_path = scratch.path().join("config.toml");
let mut tui = MainTui::new(
session,
scratch.path().to_str().expect("utf-8").to_string(),
false,
Duration::from_secs(1),
None,
Some(config_path.clone()),
);
tui.open_settings();
tui.state.settings.as_mut().expect("open").use_username_dirs = true;
tui.apply_settings();
assert!(tui.use_username_dirs, "the next download uses it");
let saved = crate::persist::config::FileConfig::load(&config_path)
.expect("config saved");
assert_eq!(saved.use_username_dirs, Some(true));
}
/// The account block answers "who is this session, and what is it
/// offering" without a trip to another screen.
#[test]
@@ -190,6 +216,7 @@ fn an_attached_window_offers_no_logout() {
let mut tui = attach(session);
press(&mut tui, KeyCode::Char('o'));
press(&mut tui, KeyCode::Down);
press(&mut tui, KeyCode::Down);
press(&mut tui, KeyCode::Enter);
let settings = tui.state.settings.as_ref().expect("still open");
+1
View File
@@ -56,6 +56,7 @@ impl Window {
let tui = MainTui::new(
client,
downloads.display().to_string(),
false,
SEARCH_WINDOW,
Some(StateStore::new(state.to_path_buf())),
None,
+5
View File
@@ -129,6 +129,11 @@ const SETTINGS: readonly Setting[] = [
</>
),
},
{
key: 'use_username_dirs',
fallback: 'false',
body: <>Save each peer's files in a subfolder named after them.</>,
},
{
key: 'max_concurrent_downloads',
fallback: '20',