fix(peer): serve full browse listings + harden the listener; TUI transfer/UX work

Peer layer:
- Stream peer messages with a partial-write queue instead of write_all, so a
  large SharedFileListResponse over the non-blocking socket is no longer
  truncated (third-party clients like SoulseekQt can now browse our shares).
- Bound the peer-init handshake with a read timeout and handle each accepted
  connection on its own thread, so one stalled peer can't wedge the listener.
- e2e: third-party direct/brokered/large-share browse + stalled-peer coverage.

TUI:
- Uploads share the downloads table columns (real Speed) and the info pane now
  renders details for a selected upload.
- Download progress bar counts only in-progress transfers and hides when idle.
- Always-visible unread badge (private messages + rooms) in the shortcuts bar,
  self-drawn blink so it works under tmux.
- Persist the unread private-message count across restarts.

soulseek-rs-lib/src/actor/peer_actor.rs

@@ -80,8 +80,17 @@ pub struct PeerActor {
     /// Transfer tokens for uploads we are serving to this peer. A TransferResponse
     /// for one of these is our upload being accepted, not a download offer.
     serving_tokens: std::collections::HashSet<u32>,
+    /// Bytes queued for the peer that the socket would not take yet. The stream
+    /// is non-blocking (so socket reads never starve the mailbox), so a message
+    /// larger than the send buffer — a full share listing is easily megabytes —
+    /// only partly fits; the rest drains on later ticks.
+    pending_write: Vec<u8>,
 }
 
+/// Cap on queued outbound bytes. A peer that stops reading must not be able to
+/// grow this without bound; dropping that connection is the right answer.
+const MAX_PENDING_WRITE: usize = 32 * 1024 * 1024;
+
 impl PeerActor {
     #[must_use]
     pub fn new(
@@ -115,6 +124,7 @@ impl PeerActor {
             disconnect_reported: false,
             id,
             serving_tokens: std::collections::HashSet::new(),
+            pending_write: Vec::new(),
         }
     }
 
@@ -512,10 +522,10 @@ impl PeerActor {
 
     fn send_message(&mut self, message: Message) {
         let username = self.peer_username();
-        let Some(stream) = self.stream.as_mut() else {
+        if self.stream.is_none() {
             error!("Cannot send message: stream is None");
             return;
-        };
+        }
 
         trace!(
             "[peer:{}] ➡ {:?}",
@@ -530,7 +540,51 @@ impl PeerActor {
                 .map_err(|e| e.to_string())
         );
 
-        if let Err(e) = stream.write_all(&message.get_buffer()) {
+        // Queue then drain, so messages always leave in order even when an
+        // earlier one is still partly buffered.
+        self.pending_write.extend_from_slice(&message.get_buffer());
+        self.flush_pending_write();
+    }
+
+    /// Push as much of `pending_write` into the socket as it will take. A
+    /// non-blocking socket refuses the rest with `WouldBlock`; that is normal
+    /// backpressure, not a failure, so the remainder stays queued for the next
+    /// tick. Only a real I/O error disconnects.
+    fn flush_pending_write(&mut self) {
+        if self.pending_write.is_empty() {
+            return;
+        }
+
+        let username = self.peer_username();
+        let Some(stream) = self.stream.as_mut() else {
+            return;
+        };
+
+        let mut written = 0;
+        let outcome = loop {
+            match stream.write(&self.pending_write[written..]) {
+                Ok(0) => {
+                    break Err(Error::new(
+                        io::ErrorKind::WriteZero,
+                        "peer stopped accepting data",
+                    ));
+                }
+                Ok(n) => {
+                    written += n;
+                    if written == self.pending_write.len() {
+                        break Ok(());
+                    }
+                }
+                Err(ref e) if e.kind() == io::ErrorKind::Interrupted => {}
+                Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => {
+                    break Ok(());
+                }
+                Err(e) => break Err(e),
+            }
+        };
+        self.pending_write.drain(..written);
+
+        if let Err(e) = outcome {
             error!(
                 "[peer:{}] Error writing message: {}. Disconnecting.",
                 username, e
@@ -539,12 +593,17 @@ impl PeerActor {
             return;
         }
 
-        if let Err(e) = stream.flush() {
+        // Backpressure is normal, but what a peer has still not taken after we
+        // pushed everything we could is a peer that is not reading at all.
+        if self.pending_write.len() > MAX_PENDING_WRITE {
             error!(
-                "[peer:{}] Error flushing stream: {}. Disconnecting.",
-                username, e
+                "[peer:{}] Outbound queue stuck above {} bytes. Disconnecting.",
+                username, MAX_PENDING_WRITE
             );
-            self.disconnect_with_error(e);
+            self.disconnect_with_error(Error::new(
+                io::ErrorKind::WriteZero,
+                "peer is not draining our outbound queue",
+            ));
         }
     }
 
@@ -785,7 +844,13 @@ impl Actor for PeerActor {
             }
             ConnectionState::Connected => {
                 if self.stream.is_some() {
+                    // a backlogged message drains one socket buffer
+                    // per 100ms tick. Enough for a share listing; if a future
+                    // message stream needs more throughput, wake the actor as
+                    // soon as the socket is writable instead of on the tick.
+                    self.flush_pending_write();
                     self.process_read();
+                    self.flush_pending_write();
                 }
             }
             ConnectionState::Disconnected => {}

soulseek-rs-lib/src/client/mod.rs

@@ -71,6 +71,30 @@ struct ActiveUpload {
     bytes_sent: Arc<std::sync::atomic::AtomicU64>,
     cancel: Arc<std::sync::atomic::AtomicBool>,
     status: crate::types::UploadStatus,
+    /// When streaming began, so a snapshot can report a transfer rate.
+    started: Instant,
+}
+
+/// Transfer rate of an in-progress upload, in bytes per second. Finished,
+/// cancelled and failed uploads report zero, matching how a download reports
+/// its speed only while running.
+///
+/// an average over the whole transfer rather than the rolling window
+/// the download path samples. Enough to fill the Speed column; sample a window
+/// if the figure ever needs to track sudden stalls.
+fn upload_speed(
+    status: &crate::types::UploadStatus,
+    bytes_sent: u64,
+    started: Instant,
+) -> f64 {
+    if !matches!(status, crate::types::UploadStatus::InProgress) {
+        return 0.0;
+    }
+    let elapsed = started.elapsed().as_secs_f64();
+    if elapsed <= 0.0 {
+        return 0.0;
+    }
+    bytes_sent as f64 / elapsed
 }
 
 /// Build a `FileSearchResponse` for `query` against `shares`, or `None` if
@@ -662,6 +686,12 @@ impl Client {
         }
     }
 
+    /// The username we log in as, for attributing our own messages.
+    #[must_use]
+    pub fn username(&self) -> &str {
+        &self.username
+    }
+
     /// The directories whose files are currently shared with other peers.
     #[must_use]
     pub fn shared_directories(&self) -> Vec<String> {
@@ -692,13 +722,19 @@ impl Client {
                     .into_iter()
                     .map(|token| {
                         let upload = &ctx.active_uploads[token];
+                        let bytes_sent = upload
+                            .bytes_sent
+                            .load(std::sync::atomic::Ordering::Relaxed);
                         crate::types::UploadInfo {
                             username: upload.username.clone(),
                             filename: upload.filename.clone(),
                             size: upload.size,
-                            bytes_sent: upload
-                                .bytes_sent
-                                .load(std::sync::atomic::Ordering::Relaxed),
+                            bytes_sent,
+                            speed_bytes_per_sec: upload_speed(
+                                &upload.status,
+                                bytes_sent,
+                                upload.started,
+                            ),
                             status: upload.status.clone(),
                         }
                     })
@@ -773,3 +809,40 @@ mod operations;
 mod rooms;
 mod search;
 mod uploads;
+
+#[cfg(test)]
+mod tests {
+    use super::upload_speed;
+    use crate::types::UploadStatus;
+    use std::time::{Duration, Instant};
+
+    #[test]
+    fn upload_speed_is_reported_only_while_running() {
+        let two_seconds_ago = Instant::now()
+            .checked_sub(Duration::from_secs(2))
+            .expect("clock supports a 2s offset");
+
+        // 1 KiB over ~2s is ~512 B/s.
+        let rate =
+            upload_speed(&UploadStatus::InProgress, 1024, two_seconds_ago);
+        assert!((rate - 512.0).abs() < 50.0, "unexpected rate {rate}");
+
+        // A finished upload reports no rate, exactly as a finished download
+        // does, so the Speed column renders "-" rather than a stale figure.
+        for status in [
+            UploadStatus::Completed,
+            UploadStatus::Cancelled,
+            UploadStatus::Failed("nope".to_string()),
+        ] {
+            let rate = upload_speed(&status, 1024, two_seconds_ago);
+            assert!(rate.abs() < f64::EPSILON, "unexpected rate {rate}");
+        }
+
+        // A just-started upload must not divide by a zero elapsed time.
+        let rate = upload_speed(&UploadStatus::InProgress, 0, Instant::now());
+        assert!(
+            rate.is_finite() && rate.abs() < f64::EPSILON,
+            "unexpected rate {rate}"
+        );
+    }
+}

soulseek-rs-lib/src/client/uploads.rs

@@ -32,6 +32,7 @@ impl Client {
                 bytes_sent: bytes_sent.clone(),
                 cancel: cancel.clone(),
                 status: UploadStatus::InProgress,
+                started: std::time::Instant::now(),
             },
         );
         drop(ctx);

soulseek-rs-lib/src/peer/listen.rs

@@ -3,6 +3,7 @@ use std::net::{TcpListener, TcpStream};
 use std::sync::mpsc::Sender;
 use std::sync::{Arc, RwLock};
 use std::thread;
+use std::time::Duration;
 
 use crate::client::{ClientContext, ClientOperation};
 
@@ -14,6 +15,9 @@ use crate::{DownloadStatus, debug, error, info, trace};
 
 const PEER_INIT_MESSAGE_CODE: u8 = 1;
 
+/// How long an accepted peer gets to send its peer-init handshake.
+const PEER_INIT_TIMEOUT: Duration = Duration::from_secs(10);
+
 #[derive(Clone)]
 struct ConnectionContext {
     client_sender: Sender<ClientOperation>,
@@ -31,13 +35,29 @@ fn read_peer_init_message(
     stream: &mut TcpStream,
     reader: &mut MessageReader,
 ) -> io::Result<Message> {
-    loop {
+    // An untrusted peer gets a bounded handshake. Without this a peer that
+    // connects and stays silent parks this read forever, and one that hangs up
+    // spins it (a zero-byte read is EOF, which `read_from_socket` reports as
+    // success) — either way pinning a thread that owes us a peer init.
+    stream.set_read_timeout(Some(PEER_INIT_TIMEOUT))?;
+    let message = loop {
+        let buffered = reader.buffer_len();
         reader.read_from_socket(stream)?;
+        if reader.buffer_len() == buffered {
+            return Err(io::Error::new(
+                io::ErrorKind::UnexpectedEof,
+                "peer closed before sending a peer init",
+            ));
+        }
 
         if let Ok(Some(msg)) = reader.extract_message() {
-            return Ok(msg);
+            break msg;
         }
-    }
+    };
+
+    // Whatever this connection turns into reads on its own terms from here.
+    stream.set_read_timeout(None)?;
+    Ok(message)
 }
 
 fn parse_peer_init_message(mut message: Message) -> Option<PeerInitData> {
@@ -285,11 +305,15 @@ fn handle_incoming_connection(stream: TcpStream, context: ConnectionContext) {
     let mut stream = stream;
     let mut reader = MessageReader::new();
 
-    let Ok(message) = read_peer_init_message(&mut stream, &mut reader) else {
-        error!(
-            "[listener:{peer_ip}:{peer_port}] Failed to read peer init message"
-        );
-        return;
+    // A peer that dials and then goes away is routine, not an error.
+    let message = match read_peer_init_message(&mut stream, &mut reader) {
+        Ok(message) => message,
+        Err(e) => {
+            debug!(
+                "[listener:{peer_ip}:{peer_port}] no peer init message: {e}"
+            );
+            return;
+        }
     };
 
     // A firewalled peer brokered through the server connects back with a
@@ -330,22 +354,15 @@ fn handle_incoming_connection(stream: TcpStream, context: ConnectionContext) {
             peer, stream, reader, &context, &peer_ip, peer_port,
         ),
 
-        ConnectionType::F => {
-            thread::spawn(move || {
-                trace!(
-                    "[listener:{peer_ip}:{peer_port}] handling file connection in thread"
-                );
-                handle_file_connection(
-                    peer,
-                    stream,
-                    reader,
-                    init_data.token,
-                    &context,
-                    &peer_ip,
-                    peer_port,
-                );
-            });
-        }
+        ConnectionType::F => handle_file_connection(
+            peer,
+            stream,
+            reader,
+            init_data.token,
+            &context,
+            &peer_ip,
+            peer_port,
+        ),
         ConnectionType::D => {
             debug!(
                 "[listener:{peer_ip}:{peer_port}] connection type is D, not supported yet, closing connection. "
@@ -384,7 +401,10 @@ impl Listen {
             };
 
             let context = context.clone();
-            handle_incoming_connection(stream, context);
+            // One thread per connection: the peer-init handshake blocks, and a
+            // peer that is slow to send one must not stop us accepting anybody
+            // else — a wedged accept loop makes us unreachable to every peer.
+            thread::spawn(move || handle_incoming_connection(stream, context));
         }
     }
 }

soulseek-rs-lib/src/types.rs

@@ -271,4 +271,7 @@ pub struct UploadInfo {
     pub size: u64,
     pub bytes_sent: u64,
     pub status: UploadStatus,
+    /// Transfer rate, so an upload row can fill the same Speed column a
+    /// download does. Zero unless the upload is in progress.
+    pub speed_bytes_per_sec: f64,
 }

soulseek-rs-lib/tests/e2e.rs

@@ -1433,3 +1433,276 @@ fn browse_a_firewalled_peer_via_broker() {
 
     let _ = std::fs::remove_dir_all(share_dir);
 }
+
+/// Browse `sharer_addr` exactly the way a third-party client (SoulseekQt,
+/// Nicotine+) does: dial the peer's listener directly, announce with a
+/// `PeerInit(P)`, ask for the share list (peer code 4) and read the
+/// `SharedFileListResponse` (peer code 5) back off the same socket.
+fn third_party_browse(
+    sharer_addr: &str,
+    username: &str,
+    read_delay: Duration,
+    timeout: Duration,
+) -> std::io::Result<Vec<soulseek_rs::SharedDirectory>> {
+    let mut p = connect_retry(sharer_addr, Duration::from_secs(5))?;
+    p.set_read_timeout(Some(timeout))?;
+    p.write_all(&peer_init_bytes(username, "P", 0))?;
+    p.write_all(&MessageFactory::build_get_share_file_list().get_buffer())?;
+    p.flush()?;
+
+    // A real peer is not always ready to drain the socket the instant the
+    // response starts arriving; a slow reader must not cost us the listing.
+    std::thread::sleep(read_delay);
+
+    let mut response = expect_code(&mut p, 5, timeout)?;
+    response.set_pointer(8);
+    Ok(soulseek_rs::message::peer::parse_shared_file_list(
+        &mut response,
+    ))
+}
+
+// A third-party client browsing us over a direct connection — the scenario of
+// running soulseek-rs and SoulseekQt side by side. The other two browse tests
+// only prove soulseek-rs can talk to itself.
+#[test]
+fn a_third_party_client_browses_our_shares_directly() {
+    let server = server_or_skip!();
+
+    let share_dir = unique_download_dir();
+    std::fs::create_dir_all(share_dir.join("album")).unwrap();
+    std::fs::write(share_dir.join("album").join("track.flac"), b"xxxx")
+        .unwrap();
+
+    let sharer_port = free_port().expect("sharer port");
+    let mut sharer = Client::with_settings(ClientSettings {
+        shared_directories: vec![share_dir.display().to_string()],
+        ..server.listening_settings("e2e_qt_sharer", "pw", sharer_port)
+    });
+    sharer.connect().expect("sharer connect");
+    assert!(sharer.login().expect("sharer login"));
+
+    // The browsing client is a real logged-in user, as SoulseekQt would be.
+    let server_addr = format!("{}:{}", server.host, server.port);
+    let _qt = login_raw(&server_addr, "e2e_qt_browser", "pw")
+        .expect("third-party client logs in");
+
+    let directories = third_party_browse(
+        &format!("127.0.0.1:{sharer_port}"),
+        "e2e_qt_browser",
+        Duration::ZERO,
+        Duration::from_secs(15),
+    )
+    .expect("third-party client should receive a SharedFileListResponse");
+
+    assert!(
+        directories
+            .iter()
+            .any(|d| d.files.iter().any(|(name, _)| name == "track.flac")),
+        "the listing should include the shared file, got {directories:?}"
+    );
+
+    let _ = std::fs::remove_dir_all(share_dir);
+}
+
+// A third-party client that cannot dial us directly browses us through the
+// server broker: it asks the server to broker, we connect back with a
+// PierceFirewall, and it browses over that connection. This is the path a
+// second client on the same machine takes whenever the direct dial to our
+// advertised (public) address does not come back to us.
+#[test]
+fn a_third_party_client_browses_our_shares_via_the_server_broker() {
+    let server = server_or_skip!();
+
+    let share_dir = unique_download_dir();
+    std::fs::create_dir_all(share_dir.join("album")).unwrap();
+    std::fs::write(share_dir.join("album").join("brokered.flac"), b"xxxx")
+        .unwrap();
+
+    // The sharer does not listen at all, so the browse can only be brokered.
+    let mut sharer = Client::with_settings(ClientSettings {
+        shared_directories: vec![share_dir.display().to_string()],
+        ..server.settings("e2e_qt_brok_sharer", "pw")
+    });
+    sharer.connect().expect("sharer connect");
+    assert!(sharer.login().expect("sharer login"));
+
+    // The third-party client listens and advertises its port, then asks the
+    // server to broker a P connection from the sharer.
+    let listener =
+        std::net::TcpListener::bind("127.0.0.1:0").expect("browser listener");
+    let browser_port = listener.local_addr().unwrap().port();
+    let server_addr = format!("{}:{}", server.host, server.port);
+    let mut srv = login_raw(&server_addr, "e2e_qt_brok_browser", "pw")
+        .expect("third-party client logs in");
+    srv.write_all(
+        &MessageFactory::build_set_wait_port_message(browser_port).get_buffer(),
+    )
+    .expect("set wait port");
+    srv.flush().expect("flush wait port");
+    std::thread::sleep(Duration::from_secs(1));
+
+    let token = 424_242_u32;
+    srv.write_all(
+        &MessageFactory::build_connect_to_peer(
+            token,
+            "e2e_qt_brok_sharer",
+            ConnectionType::P,
+        )
+        .get_buffer(),
+    )
+    .expect("ask the server to broker");
+    srv.flush().expect("flush ConnectToPeer");
+
+    // The sharer must dial us back, quoting our correlation token.
+    listener.set_nonblocking(true).expect("non-blocking accept");
+    let deadline = Instant::now() + Duration::from_secs(20);
+    let mut accepted = None;
+    while Instant::now() < deadline {
+        match listener.accept() {
+            Ok((stream, _)) => {
+                accepted = Some(stream);
+                break;
+            }
+            Err(ref e) if e.kind() == std::io::ErrorKind::WouldBlock => {
+                std::thread::sleep(Duration::from_millis(50));
+            }
+            Err(e) => panic!("accept failed: {e}"),
+        }
+    }
+    let mut p = accepted.expect("the sharer should connect back to us");
+    p.set_nonblocking(false).expect("blocking peer socket");
+    p.set_read_timeout(Some(Duration::from_secs(20)))
+        .expect("read timeout");
+
+    let mut pierce = read_framed(&mut p).expect("a PierceFirewall frame");
+    assert_eq!(
+        pierce.get_message_code(),
+        0,
+        "the brokered connect-back must start with a PierceFirewall"
+    );
+    pierce.set_pointer(5); // length prefix (4) + int8 code (1)
+    assert_eq!(
+        pierce.read_int32(),
+        token,
+        "the PierceFirewall must quote the token we brokered with"
+    );
+
+    p.write_all(&MessageFactory::build_get_share_file_list().get_buffer())
+        .expect("request the share list");
+    p.flush().expect("flush share list request");
+
+    let mut response = expect_code(&mut p, 5, Duration::from_secs(20))
+        .expect("a SharedFileListResponse over the brokered connection");
+    response.set_pointer(8);
+    let directories =
+        soulseek_rs::message::peer::parse_shared_file_list(&mut response);
+
+    assert!(
+        directories
+            .iter()
+            .any(|d| d.files.iter().any(|(name, _)| name == "brokered.flac")),
+        "the brokered listing should include the shared file, got {directories:?}"
+    );
+
+    let _ = std::fs::remove_dir_all(share_dir);
+}
+
+// Real peers connect to our listener and go away again (a dropped dial, a port
+// scan, a client that gave up). None of that may cost us the listener: the next
+// client to come along must still be able to browse.
+#[test]
+fn a_stalled_peer_connection_does_not_wedge_the_listener() {
+    let server = server_or_skip!();
+
+    let share_dir = unique_download_dir();
+    std::fs::create_dir_all(share_dir.join("album")).unwrap();
+    std::fs::write(share_dir.join("album").join("still.flac"), b"xxxx")
+        .unwrap();
+
+    let sharer_port = free_port().expect("sharer port");
+    let mut sharer = Client::with_settings(ClientSettings {
+        shared_directories: vec![share_dir.display().to_string()],
+        ..server.listening_settings("e2e_wedge_sharer", "pw", sharer_port)
+    });
+    sharer.connect().expect("sharer connect");
+    assert!(sharer.login().expect("sharer login"));
+
+    let sharer_addr = format!("127.0.0.1:{sharer_port}");
+
+    // A peer that connects and hangs up without ever sending a PeerInit.
+    drop(
+        connect_retry(&sharer_addr, Duration::from_secs(5))
+            .expect("dial the listener"),
+    );
+    // A peer that connects and then just sits there, saying nothing.
+    let _silent = connect_retry(&sharer_addr, Duration::from_secs(5))
+        .expect("dial the listener");
+
+    let server_addr = format!("{}:{}", server.host, server.port);
+    let _qt = login_raw(&server_addr, "e2e_wedge_browser", "pw")
+        .expect("third-party client logs in");
+
+    let directories = third_party_browse(
+        &sharer_addr,
+        "e2e_wedge_browser",
+        Duration::ZERO,
+        Duration::from_secs(15),
+    )
+    .expect("the listener must still serve browse requests");
+
+    assert!(
+        directories
+            .iter()
+            .any(|d| d.files.iter().any(|(name, _)| name == "still.flac")),
+        "the listing should include the shared file, got {directories:?}"
+    );
+
+    let _ = std::fs::remove_dir_all(share_dir);
+}
+
+// The same direct browse against a share big enough that the response cannot
+// fit in one socket buffer. This is the realistic case: a shared music library
+// serialises to hundreds of KB, and the payload is zlib-STORED (not actually
+// compressed), so the listing goes out at full size.
+#[test]
+fn a_third_party_client_browses_a_large_share() {
+    let server = server_or_skip!();
+
+    let share_dir = unique_download_dir().join("big");
+    let _ = std::fs::remove_dir_all(&share_dir);
+    std::fs::create_dir_all(&share_dir).unwrap();
+    // ~4000 entries with long names: well over any socket buffer once framed.
+    let padding = "x".repeat(150);
+    for i in 0..4000 {
+        std::fs::write(share_dir.join(format!("{padding}-{i:05}.flac")), b"z")
+            .unwrap();
+    }
+
+    let sharer_port = free_port().expect("sharer port");
+    let mut sharer = Client::with_settings(ClientSettings {
+        shared_directories: vec![share_dir.display().to_string()],
+        ..server.listening_settings("e2e_qt_big_sharer", "pw", sharer_port)
+    });
+    sharer.connect().expect("sharer connect");
+    assert!(sharer.login().expect("sharer login"));
+
+    let server_addr = format!("{}:{}", server.host, server.port);
+    let _qt = login_raw(&server_addr, "e2e_qt_big_browser", "pw")
+        .expect("third-party client logs in");
+
+    let directories = third_party_browse(
+        &format!("127.0.0.1:{sharer_port}"),
+        "e2e_qt_big_browser",
+        Duration::from_secs(1),
+        Duration::from_secs(20),
+    )
+    .expect("third-party client should receive the full large listing");
+
+    let file_count: usize = directories.iter().map(|d| d.files.len()).sum();
+    assert_eq!(
+        file_count, 4000,
+        "the whole listing must arrive, not just what fit in one socket buffer"
+    );
+
+    let _ = std::fs::remove_dir_all(share_dir);
+}