From d747efc0ff7f247f4cda3d610e3053b0f2ebae29 Mon Sep 17 00:00:00 2001 From: artrixdotdev Date: Wed, 29 Jul 2026 00:17:15 -0700 Subject: [PATCH 1/3] feat: add peer and tracker management --- crates/libtortillas/src/errors.rs | 32 ++ crates/libtortillas/src/facade.rs | 1 + crates/libtortillas/src/lib.rs | 2 +- crates/libtortillas/src/live/hub.rs | 31 +- crates/libtortillas/src/torrent/actor.rs | 105 ++---- crates/libtortillas/src/torrent/handle.rs | 298 +++++++++++++++++- crates/libtortillas/src/torrent/messages.rs | 128 ++++++-- crates/libtortillas/src/torrent/mod.rs | 12 + crates/libtortillas/src/torrent/piece_flow.rs | 1 + crates/libtortillas/src/torrent/swarm.rs | 208 ++++++++---- 10 files changed, 652 insertions(+), 166 deletions(-) diff --git a/crates/libtortillas/src/errors.rs b/crates/libtortillas/src/errors.rs index 2d09b0e0..ac77c9c1 100644 --- a/crates/libtortillas/src/errors.rs +++ b/crates/libtortillas/src/errors.rs @@ -312,6 +312,26 @@ pub enum TorrentError { reason: String, }, + /// Tried to connect a peer that is already part of this torrent's swarm. + #[error("Peer is already connected: {peer_id}")] + PeerAlreadyConnected { peer_id: PeerId }, + + /// Tried to disconnect a peer that is not part of this torrent's swarm. + #[error("Peer not found: {peer_id}")] + PeerNotFound { peer_id: PeerId }, + + /// Tried to add a tracker that is already configured for this torrent. + #[error("Tracker already exists: {endpoint}")] + TrackerAlreadyExists { endpoint: String }, + + /// Tried to operate on a tracker that is not configured for this torrent. + #[error("Tracker not found: {endpoint}")] + TrackerNotFound { endpoint: String }, + + /// Tried to add a tracker protocol that the runtime cannot operate. + #[error("Unsupported tracker protocol: {protocol}")] + UnsupportedTrackerProtocol { protocol: &'static str }, + /// Bitfield operation failed #[error("Bitfield operation failed: {reason}")] BitfieldError { reason: String }, @@ -385,6 +405,18 @@ pub(crate) fn map_torrent_send_error( }, } } + +pub(crate) fn map_torrent_communication_error( + operation: &'static str, error: SendError, +) -> TorrentError +where + E: std::fmt::Debug + std::fmt::Display, +{ + TorrentError::ActorCommunicationFailed { + operation, + reason: error.to_string(), + } +} // Conversion implementations for backward compatibility during transition impl From> for TrackerActorError { fn from(err: num_enum::TryFromPrimitiveError) -> Self { diff --git a/crates/libtortillas/src/facade.rs b/crates/libtortillas/src/facade.rs index 1c104d8d..599ff235 100644 --- a/crates/libtortillas/src/facade.rs +++ b/crates/libtortillas/src/facade.rs @@ -5,6 +5,7 @@ pub use crate::{ engine::{Engine, EngineSnapshot, EngineStatus, TorrentSource}, torrent::{RestoreVerification, Torrent, TorrentSnapshot}, + tracker::Tracker, }; #[cfg(feature = "live")] pub use crate::{ diff --git a/crates/libtortillas/src/lib.rs b/crates/libtortillas/src/lib.rs index fd2e871d..8f64e081 100644 --- a/crates/libtortillas/src/lib.rs +++ b/crates/libtortillas/src/lib.rs @@ -571,7 +571,7 @@ pub(crate) mod testing { stream.write_all(&message.to_bytes()?).await?; } - Ok(()) + std::future::pending().await } pub(crate) fn test_info_hash() -> InfoHash { diff --git a/crates/libtortillas/src/live/hub.rs b/crates/libtortillas/src/live/hub.rs index 64b1f308..5d2e5fbe 100644 --- a/crates/libtortillas/src/live/hub.rs +++ b/crates/libtortillas/src/live/hub.rs @@ -76,11 +76,7 @@ where } fn remove_value(&self, value: &Arc) -> bool { - let key = self - .values - .iter() - .find(|entry| Arc::ptr_eq(entry.value(), value)) - .map(|entry| entry.key().clone()); + let key = self.key_for_value(value); key.is_some_and(|key| { self .values @@ -89,6 +85,14 @@ where }) } + fn key_for_value(&self, value: &Arc) -> Option { + self + .values + .iter() + .find(|entry| Arc::ptr_eq(entry.value(), value)) + .map(|entry| entry.key().clone()) + } + fn values(&self) -> Vec> { self .values @@ -607,6 +611,23 @@ impl Hub { }) } + pub(crate) fn tracker_handle( + &self, torrent: InfoHash, source: &Tracker, + ) -> Option { + let inner = self.inner()?; + let scope = inner.torrents.get(&torrent)?; + scope + .trackers + .get(source) + .map(|inner| TrackerHandle { inner }) + } + + pub(crate) fn tracker_source(&self, tracker: &TrackerHandle) -> Option { + let inner = self.inner()?; + let scope = inner.torrents.get(&tracker.torrent())?; + scope.trackers.key_for_value(&tracker.inner) + } + pub(crate) fn register_tracker_scope( &self, torrent: InfoHash, source: &Tracker, view: TrackerView, ) -> Option { diff --git a/crates/libtortillas/src/torrent/actor.rs b/crates/libtortillas/src/torrent/actor.rs index baa1ab05..00cce9d0 100644 --- a/crates/libtortillas/src/torrent/actor.rs +++ b/crates/libtortillas/src/torrent/actor.rs @@ -35,17 +35,12 @@ use crate::{ BLOCK_SIZE, PieceBlockSnapshot, PieceStorageStrategy, TORRENT_SNAPSHOT_VERSION, TorrentSnapshot, TorrentState, }, - tracker::{ - Announce, Event, Tracker, TrackerActor, TrackerActorArgs, TrackerUpdate, udp::UdpServer, - }, + tracker::{Announce, Event, Tracker, TrackerActor, TrackerUpdate, udp::UdpServer}, }; #[cfg(feature = "live")] use crate::{ - live::{Hub, LiveHealthLevel, TorrentEventKind, TorrentView, TrackerStatus, TrackerView}, - metrics::{ - ByteCount, ContentProgress, HasTransferMetrics, TorrentMetrics, TrackerMetrics, - TransferMetrics, - }, + live::{Hub, LiveHealthLevel, TorrentEventKind, TorrentView}, + metrics::{ByteCount, ContentProgress, HasTransferMetrics, TorrentMetrics, TransferMetrics}, }; /// A hook that is called when the torrent is ready to start downloading. @@ -117,6 +112,7 @@ pub(crate) struct TorrentActor { pub(super) metainfo: MetaInfo, #[allow(dead_code)] pub(super) tracker_server: UdpServer, + pub(super) primary_addr: SocketAddr, pub(super) scheduler: ActorRef, /// Should only be used to create new connections pub(super) utp_server: Arc, @@ -477,10 +473,13 @@ impl TorrentActor { ); } - pub(super) fn remove_peer(&mut self, id: PeerId) { + pub(super) fn remove_peer(&mut self, id: PeerId) -> bool { self.piece_scheduler.peer_disconnected(id); if let Some(actor) = self.peers.remove(&id) { actor.kill(); + true + } else { + false } } @@ -757,61 +756,7 @@ impl Actor for TorrentActor { TorrentState::ResolvingMetadata }; let bitfield = BitVec::repeat(false, piece_count); - let initial_left = info.map(Info::total_length); - - // Create tracker actors let tracker_list = metainfo.announce_list(); - let mut trackers = HashMap::new(); - for tracker in tracker_list { - if matches!(tracker, Tracker::Websocket(_)) { - warn!( - tracker_uri = %tracker.redacted_endpoint(), - "Skipping unsupported websocket tracker" - ); - continue; - } - #[cfg(feature = "live")] - let endpoint = tracker.redacted_endpoint(); - #[cfg(feature = "live")] - let Some(tracker_handle) = hub.register_tracker_scope( - torrent_id, - &tracker, - TrackerView { - endpoint, - status: TrackerStatus::Pending, - metrics: TrackerMetrics::default(), - }, - ) else { - return Err(TorrentError::ActorCommunicationFailed { - operation: "register tracker live scope", - reason: "live hub is no longer available".to_string(), - }); - }; - let actor = TrackerActor::supervise( - &us, - TrackerActorArgs { - tracker: tracker.clone(), - peer_id, - server: tracker_server.clone(), - socket_addr: primary_addr, - initial_left, - supervisor: us.clone(), - scheduler: scheduler.clone(), - settings: settings.tracker.clone(), - #[cfg(feature = "live")] - live_handle: tracker_handle, - }, - ) - .restart_policy(RestartPolicy::Transient) - .restart_limit( - settings.torrent.tracker_restart.limit, - settings.torrent.tracker_restart.period, - ) - .spawn() - .await; - - trackers.insert(tracker, actor); - } let default_manager = FilePieceManager(base_path, info.cloned()); let piece_store = PieceStoreActor::supervise(&us, ()) .restart_policy(RestartPolicy::Permanent) @@ -822,15 +767,16 @@ impl Actor for TorrentActor { .spawn() .await; - let actor = Self { + let mut actor = Self { #[cfg(feature = "live")] hub, peers: HashMap::new(), bitfield, tracker_server, + primary_addr, scheduler, utp_server, - trackers, + trackers: HashMap::new(), id: peer_id, info_hash: torrent_id, metainfo, @@ -853,6 +799,15 @@ impl Actor for TorrentActor { piece_manager: PieceManagerProxy::Default(default_manager), settings, }; + for tracker in tracker_list { + if let Err(error) = actor.register_tracker_actor(tracker.clone()).await { + warn!( + %error, + tracker_uri = %tracker.redacted_endpoint(), + "Skipping unavailable tracker" + ); + } + } crate::live_only!(actor.hub.initialize_torrent_projection(actor.live_view())); Ok(actor) @@ -938,9 +893,11 @@ mod tests { use super::*; use crate::{ hashes::HashVec, - live::{PeerIdentity, PeerView}, + live::{PeerIdentity, PeerView, TrackerStatus, TrackerView}, metainfo::{InfoKeys, MetaInfo, TorrentFile}, - metrics::{BytesPerSecond, PeerMetrics, TrafficTotals, TransferRates, TransferSample}, + metrics::{ + BytesPerSecond, PeerMetrics, TrackerMetrics, TrafficTotals, TransferRates, TransferSample, + }, protocol::{ messages::{Handshake, PeerMessages}, stream::{PeerRecv, PeerSend, PeerStream}, @@ -949,8 +906,8 @@ mod tests { testing, torrent::{ BLOCK_SIZE, Torrent, TorrentSnapshot, - commands::{GetState, HasInfoDict, SetState, SnapshotState}, - events::{AddPeer, IncomingPiece}, + commands::{AddPeer, GetState, HasInfoDict, SetState, SnapshotState}, + events::IncomingPiece, }, tracker::Tracker, }; @@ -1346,7 +1303,12 @@ mod tests { hub: Hub::default(), }); let torrent = Torrent::new(info_hash, actor.clone()); - actor.tell(AddPeer { peer: seed.peer() }).await.unwrap(); + let connection = actor.ask(AddPeer { peer: seed.peer() }).await.unwrap(); + timeout(Duration::from_secs(5), connection) + .await + .expect("local seed connection should complete") + .unwrap() + .unwrap(); timeout(Duration::from_secs(5), torrent.poll_ready()) .await @@ -1701,6 +1663,7 @@ mod tests { resolved_magnet_info: None, metainfo: metainfo.clone(), tracker_server: udp_server.clone(), + primary_addr: utp_server.bind_addr(), scheduler: Scheduler::spawn(Scheduler::new()), utp_server, actor_ref: actor_ref.clone(), @@ -1861,6 +1824,7 @@ mod tests { resolved_magnet_info: None, metainfo: metainfo.clone(), tracker_server: udp_server, + primary_addr: utp_server.bind_addr(), scheduler: Scheduler::spawn(Scheduler::new()), utp_server, actor_ref: actor_ref.clone(), @@ -2083,6 +2047,7 @@ mod tests { resolved_magnet_info: None, metainfo, tracker_server: udp_server, + primary_addr: utp_server.bind_addr(), scheduler: Scheduler::spawn(Scheduler::new()), utp_server, actor_ref: actor_ref.clone(), diff --git a/crates/libtortillas/src/torrent/handle.rs b/crates/libtortillas/src/torrent/handle.rs index 34d442ee..78ca281e 100644 --- a/crates/libtortillas/src/torrent/handle.rs +++ b/crates/libtortillas/src/torrent/handle.rs @@ -1,4 +1,6 @@ #[cfg(feature = "live")] +use std::net::SocketAddr; +#[cfg(feature = "live")] use std::sync::Weak; use std::{fmt, path::PathBuf, sync::Arc}; @@ -9,8 +11,8 @@ use tracing::error; use super::{ PieceStorageStrategy, TorrentActor, TorrentSnapshot, TorrentState, commands::{ - GetState, ReadyHook, SetAutoStart, SetOutputPath, SetPieceManager, SetPieceStorage, SetState, - SetSufficientPeers, SnapshotState, + ForceReannounce, GetState, ReadyHook, SetAutoStart, SetOutputPath, SetPieceManager, + SetPieceStorage, SetState, SetSufficientPeers, SnapshotState, }, }; #[cfg(feature = "live")] @@ -18,6 +20,13 @@ use crate::live::{ EventSubscription, Hub, HubInner, LivePublisher, PeerHandle, TorrentEventKind, TorrentListener, TorrentView, TrackerHandle, }; +#[cfg(feature = "live")] +use crate::{ + errors::map_torrent_communication_error, + peer::WirePeer, + torrent::commands::{AddPeer, AddTracker, DisconnectPeer, ReannounceTracker, RemoveTracker}, + tracker::Tracker, +}; use crate::{ errors::{TorrentError, map_torrent_send_error}, hashes::InfoHash, @@ -189,6 +198,183 @@ impl Torrent { })?; Ok(()) } + + /// Queues an immediate announce on every configured tracker. + /// + /// Every tracker is attempted even when another command delivery fails. + /// `Ok` returns the number of trackers that accepted the command. Tracker + /// responses remain asynchronous and are reported through their live views + /// and event streams. + pub async fn force_reannounce(&self) -> Result { + self + .actor() + .ask(ForceReannounce) + .await + .map_err(|error| map_torrent_send_error("force tracker reannounce", error)) + } +} + +#[cfg(all(test, feature = "live"))] +mod tests { + use std::time::Duration; + + use tokio::time::timeout; + + use crate::{ + engine::{Engine, TorrentSource}, + errors::TorrentError, + live::{PeerEventKind, TrackerEventKind, TrackerStatus}, + peer::PeerId, + settings::Settings, + testing::{self, LocalHttpTracker, LocalPeer}, + tracker::Tracker, + }; + + fn deterministic_engine() -> Engine { + let mut settings = Settings::default(); + settings.dht.enabled = false; + Engine::builder() + .settings(settings) + .autostart(false) + .build() + } + + async fn trackerless_source() -> TorrentSource { + let mut metainfo = testing::read_torrent_fixture(testing::BIG_BUCK_BUNNY_TORRENT_FILE).await; + metainfo.clear_announce_list(); + TorrentSource::torrent_file_bytes(serde_bencode::to_bytes(&metainfo).unwrap()) + } + + #[tokio::test] + async fn torrent_handle_manages_manual_peer_lifecycle() { + let engine = deterministic_engine(); + let torrent = engine + .add_torrent(trackerless_source().await) + .await + .unwrap(); + let remote_id = PeerId::Unknown([42; 20]); + let local_peer = LocalPeer::start(remote_id, Vec::new()).await.unwrap(); + + let peer = torrent + .add_peer(local_peer.peer().socket_addr()) + .await + .unwrap(); + let mut listener = peer.listener(); + + assert_eq!(peer.id(), remote_id); + assert!(peer.view().connected); + assert_eq!(torrent.peers(), vec![peer.clone()]); + + torrent.disconnect_peer(&peer).await.unwrap(); + + assert!(!listener.view().connected); + assert_eq!( + listener.recv().await.unwrap().kind, + PeerEventKind::Disconnected + ); + assert!(torrent.peers().is_empty()); + assert!(matches!( + torrent.disconnect_peer(&peer).await, + Err(TorrentError::PeerNotFound { peer_id }) if peer_id == remote_id + )); + + engine.shutdown().await.unwrap(); + } + + #[tokio::test] + async fn torrent_handle_manages_tracker_and_reannounce_lifecycle() { + let engine = deterministic_engine(); + let torrent = engine + .add_torrent(trackerless_source().await) + .await + .unwrap(); + let local_tracker = LocalHttpTracker::start([]).await.unwrap(); + let source = Tracker::Http(local_tracker.uri()); + + assert!(matches!( + torrent + .add_tracker(Tracker::Websocket( + "wss://tracker.example/announce".to_string() + )) + .await, + Err(TorrentError::UnsupportedTrackerProtocol { + protocol: "websocket" + }) + )); + let tracker = timeout(Duration::from_secs(2), torrent.add_tracker(source.clone())) + .await + .expect("tracker initialization timed out") + .unwrap(); + let mut listener = tracker.listener(); + + assert_eq!(tracker.view().status, TrackerStatus::Pending); + assert_eq!(torrent.trackers(), vec![tracker.clone()]); + assert!(matches!( + timeout(Duration::from_secs(2), torrent.add_tracker(source)) + .await + .expect("duplicate tracker check timed out"), + Err(TorrentError::TrackerAlreadyExists { .. }) + )); + + timeout(Duration::from_secs(2), torrent.reannounce_tracker(&tracker)) + .await + .expect("tracker reannounce timed out") + .unwrap(); + let announce = timeout(Duration::from_secs(2), listener.recv()) + .await + .unwrap() + .unwrap(); + assert!(matches!( + announce.kind, + TrackerEventKind::AnnounceSucceeded { peers_returned: 0 } + )); + assert_eq!( + timeout(Duration::from_secs(2), torrent.force_reannounce()) + .await + .expect("torrent reannounce timed out") + .unwrap(), + 1 + ); + let announce = timeout(Duration::from_secs(2), listener.recv()) + .await + .unwrap() + .unwrap(); + assert!(matches!( + announce.kind, + TrackerEventKind::AnnounceSucceeded { peers_returned: 0 } + )); + assert_eq!(listener.view().status, TrackerStatus::Healthy); + assert!(local_tracker.requests().await.len() >= 2); + + drop(local_tracker); + timeout(Duration::from_secs(2), torrent.reannounce_tracker(&tracker)) + .await + .expect("failed tracker reannounce timed out") + .unwrap(); + assert!(matches!( + timeout(Duration::from_secs(2), listener.recv()) + .await + .unwrap() + .unwrap() + .kind, + TrackerEventKind::AnnounceFailed + )); + assert_eq!(listener.view().status, TrackerStatus::Degraded); + + timeout(Duration::from_secs(2), torrent.remove_tracker(&tracker)) + .await + .expect("tracker removal timed out") + .unwrap(); + + assert_eq!(listener.view().status, TrackerStatus::Stopped); + assert!(torrent.trackers().is_empty()); + assert!(matches!( + torrent.reannounce_tracker(&tracker).await, + Err(TorrentError::TrackerNotFound { .. }) + )); + + engine.shutdown().await.unwrap(); + } } #[cfg(not(feature = "live"))] @@ -202,6 +388,17 @@ impl Torrent { #[cfg(feature = "live")] impl Torrent { + async fn await_delegated( + result: oneshot::Receiver>, operation: &'static str, + ) -> Result { + result + .await + .map_err(|error| TorrentError::ActorCommunicationFailed { + operation, + reason: error.to_string(), + })? + } + #[cfg(test)] pub(crate) fn new(info_hash: InfoHash, actor_ref: ActorRef) -> Self { Self::new_with_hub(info_hash, actor_ref, &Hub::default(), None) @@ -226,6 +423,79 @@ impl Torrent { Self { inner } } + /// Connects and handshakes with a manually supplied peer. + /// + /// A successful result guarantees that the handshake completed and the + /// peer was registered with this torrent. Peer actor initialization then + /// follows the same asynchronous path as discovered peers. + pub async fn add_peer(&self, address: SocketAddr) -> Result { + let result = self + .actor() + .ask(AddPeer { + peer: WirePeer::from_socket_addr(address), + }) + .await + .map_err(|error| map_torrent_communication_error("add peer", error))?; + Self::await_delegated(result, "add peer").await + } + + /// Disconnects a peer. + /// + /// `Ok` guarantees that the peer is no longer registered in the active + /// swarm, its actor has been told to stop, and its terminal live event has + /// been published. + pub async fn disconnect_peer(&self, peer: &PeerHandle) -> Result<(), TorrentError> { + self.ensure_torrent_identity(peer.torrent(), "disconnect peer")?; + self + .actor() + .ask(DisconnectPeer { + id: peer.id(), + handle: peer.clone(), + }) + .await + .map_err(|error| map_torrent_send_error("disconnect peer", error)) + } + + /// Adds a tracker and starts its actor. + /// + /// `Ok` guarantees that the tracker was accepted and registered. Actor + /// initialization and announces are asynchronous; use its listener or + /// [`TrackerHandle::view`] to observe their result. + pub async fn add_tracker(&self, tracker: Tracker) -> Result { + self + .actor() + .ask(AddTracker { tracker }) + .await + .map_err(|error| map_torrent_send_error("add tracker", error)) + } + + /// Removes a tracker and publishes its terminal event. + /// + /// `Ok` guarantees that the tracker is no longer configured and its live + /// scope is terminal. Its final stopped announce is best-effort and may + /// finish after this method returns. + pub async fn remove_tracker(&self, tracker: &TrackerHandle) -> Result<(), TorrentError> { + let source = self.tracker_source(tracker, "remove tracker")?; + self + .actor() + .ask(RemoveTracker { tracker: source }) + .await + .map_err(|error| map_torrent_send_error("remove tracker", error)) + } + + /// Queues an immediate announce on one tracker. + /// + /// `Ok` guarantees command delivery. The tracker response is asynchronous + /// and updates the handle's view and event stream. + pub async fn reannounce_tracker(&self, tracker: &TrackerHandle) -> Result<(), TorrentError> { + let source = self.tracker_source(tracker, "reannounce tracker")?; + self + .actor() + .ask(ReannounceTracker { tracker: source }) + .await + .map_err(|error| map_torrent_send_error("reannounce tracker", error)) + } + /// Subscribes to live events for this torrent only. #[must_use] pub fn subscribe(&self) -> EventSubscription { @@ -265,4 +535,28 @@ impl Torrent { fn hub(&self) -> Option { self.inner.hub.upgrade().map(Hub::from_inner) } + + fn ensure_torrent_identity( + &self, torrent: InfoHash, operation: &'static str, + ) -> Result<(), TorrentError> { + if torrent == self.info_hash() { + return Ok(()); + } + Err(TorrentError::InvalidOperation { + operation, + reason: "the handle belongs to a different torrent".to_string(), + }) + } + + fn tracker_source( + &self, tracker: &TrackerHandle, operation: &'static str, + ) -> Result { + self.ensure_torrent_identity(tracker.torrent(), operation)?; + self + .hub() + .and_then(|hub| hub.tracker_source(tracker)) + .ok_or_else(|| TorrentError::TrackerNotFound { + endpoint: tracker.endpoint(), + }) + } } diff --git a/crates/libtortillas/src/torrent/messages.rs b/crates/libtortillas/src/torrent/messages.rs index d2d91a69..c02c1882 100644 --- a/crates/libtortillas/src/torrent/messages.rs +++ b/crates/libtortillas/src/torrent/messages.rs @@ -7,24 +7,25 @@ use bitvec::vec::BitVec; use bytes::Bytes; use kameo::{Reply, messages}; use sha1::{Digest, Sha1}; +use tokio::sync::oneshot; use tracing::{info, instrument, trace, warn}; use super::{ - AnnounceFrom, BLOCK_SIZE, PieceStorageStrategy, TorrentActor, TorrentSnapshot, TorrentState, - ValidatedTorrentState, + AnnounceFrom, BLOCK_SIZE, ConfiguredTracker, ConnectedPeer, PeerDisconnect, + PieceStorageStrategy, TorrentActor, TorrentSnapshot, TorrentState, ValidatedTorrentState, actor::{PieceManagerProxy, ReadyHookSender}, util, }; #[cfg(feature = "live")] use crate::live::TorrentView; use crate::{ - errors::TorrentError, + errors::{TorrentError, map_torrent_communication_error}, hashes::InfoHash, metainfo::Info, peer::{PeerId, WirePeer, commands::HaveInfoDict}, pieces::{PieceManager, PieceScheduler}, protocol::stream::PeerStream, - tracker::Tracker, + tracker::{Announce, Tracker}, }; #[derive(Debug, Reply)] @@ -58,22 +59,13 @@ pub(crate) mod events { self.append_peer(peer, Some((stream, reserved))); } - /// Used to manually add a peer. This is primarily used for testing but - /// can be used to initiate a peer connection without it having to - /// come from an announce. - #[message(derive(Debug))] - #[instrument(skip(self), fields(torrent_id = %self.info_hash()))] - pub(crate) fn add_peer(&mut self, peer: WirePeer) { - self.append_peer(peer, None); - } - /// Sent by a connection task after peer handshaking completes. #[message] #[instrument(skip(self, stream), fields(torrent_id = %self.info_hash()))] pub(crate) fn peer_connected( &mut self, peer: WirePeer, reserved: [u8; 8], stream: PeerStream, - ) { - self.insert_peer(peer, reserved, stream); + ) -> Result { + self.insert_peer(peer, reserved, stream) } /// Index, offset, and data for a received peer `Piece` message. @@ -177,16 +169,106 @@ pub(crate) mod commands { #[messages] impl TorrentActor { + #[message(derive(Debug))] + pub(crate) fn add_peer( + &mut self, peer: WirePeer, + ) -> oneshot::Receiver> { + let (result_tx, result_rx) = oneshot::channel(); + self.spawn_peer_connection(peer, None, Some(result_tx)); + result_rx + } + #[message] - pub(crate) fn kill_tracker(&mut self, tracker: Tracker) { - // Kill the actor quietly. - if let Some(actor) = self.trackers.get(&tracker) { - actor.kill(); - self.trackers.remove(&tracker); - self.publish_updated(); - } else { - warn!("Received kill tracker message for unknown tracker"); + pub(crate) fn disconnect_peer( + &mut self, id: PeerId, handle: PeerDisconnect, + ) -> Result<(), TorrentError> { + #[cfg(not(feature = "live"))] + let () = handle; + if !self.remove_peer(id) { + return Err(TorrentError::PeerNotFound { peer_id: id }); + } + crate::live_only!(handle.disconnected()); + self.publish_updated(); + self.fill_all_peer_request_windows(); + Ok(()) + } + + #[message(derive(Debug))] + pub(crate) async fn add_tracker( + &mut self, tracker: Tracker, + ) -> Result { + self.register_tracker_actor(tracker).await + } + + #[message(derive(Debug))] + pub(crate) async fn remove_tracker(&mut self, tracker: Tracker) -> Result<(), TorrentError> { + let endpoint = tracker.redacted_endpoint(); + let actor = self + .trackers + .remove(&tracker) + .ok_or(TorrentError::TrackerNotFound { endpoint })?; + #[cfg(feature = "live")] + let tracker_handle = self.hub.tracker_handle(self.info_hash(), &tracker); + + self.actor_ref.unlink(&actor).await; + // Tracker shutdown reports final metrics through this actor's + // mailbox. Remove it now and let that best-effort cleanup finish + // asynchronously instead of waiting on our own mailbox. + if actor.is_alive() { + actor.stop_gracefully().await.map_err(|error| { + TorrentError::ActorCommunicationFailed { + operation: "remove tracker", + reason: error.to_string(), + } + })?; + } + crate::live_only! { + if let Some(tracker_handle) = tracker_handle { + tracker_handle.stopped(); + } + } + self.publish_updated(); + Ok(()) + } + + #[message(derive(Debug))] + pub(crate) async fn reannounce_tracker( + &mut self, tracker: Tracker, + ) -> Result<(), TorrentError> { + let endpoint = tracker.redacted_endpoint(); + let actor = self + .trackers + .get(&tracker) + .ok_or(TorrentError::TrackerNotFound { endpoint })?; + actor + .tell(Announce) + .await + .map_err(|error| map_torrent_communication_error("reannounce tracker", error)) + } + + #[message(derive(Debug, Clone, Copy))] + pub(crate) async fn force_reannounce(&mut self) -> Result { + if self.trackers.is_empty() { + return Err(TorrentError::InvalidOperation { + operation: "force reannounce", + reason: "torrent has no configured trackers".to_string(), + }); + } + + let mut queued = 0usize; + let mut first_error = None; + for (tracker, actor) in &self.trackers { + match actor.tell(Announce).await { + Ok(()) => queued = queued.saturating_add(1), + Err(error) => { + first_error.get_or_insert_with(|| TorrentError::ActorCommunicationFailed { + operation: "force tracker reannounce", + reason: format!("{}: {error}", tracker.redacted_endpoint()), + }); + } + } } + first_error.map_or(Ok(queued), Err) } #[message] diff --git a/crates/libtortillas/src/torrent/mod.rs b/crates/libtortillas/src/torrent/mod.rs index 14d49aac..5e32a68b 100644 --- a/crates/libtortillas/src/torrent/mod.rs +++ b/crates/libtortillas/src/torrent/mod.rs @@ -47,6 +47,18 @@ mod storage; mod swarm; pub(crate) use actor::{TorrentActor, TorrentActorArgs}; +#[cfg(feature = "live")] +pub(crate) type ConnectedPeer = crate::live::PeerHandle; +#[cfg(not(feature = "live"))] +pub(crate) type ConnectedPeer = crate::peer::PeerId; +#[cfg(feature = "live")] +pub(crate) type PeerDisconnect = crate::live::PeerHandle; +#[cfg(not(feature = "live"))] +pub(crate) type PeerDisconnect = (); +#[cfg(feature = "live")] +pub(crate) type ConfiguredTracker = crate::live::TrackerHandle; +#[cfg(not(feature = "live"))] +pub(crate) type ConfiguredTracker = (); pub use block::{BLOCK_SIZE, BlockMap}; pub use discovery::AnnounceFrom; pub use handle::Torrent; diff --git a/crates/libtortillas/src/torrent/piece_flow.rs b/crates/libtortillas/src/torrent/piece_flow.rs index 0c8e6f5f..c258a9f1 100644 --- a/crates/libtortillas/src/torrent/piece_flow.rs +++ b/crates/libtortillas/src/torrent/piece_flow.rs @@ -511,6 +511,7 @@ mod tests { resolved_magnet_info: None, metainfo, tracker_server, + primary_addr: utp_server.bind_addr(), scheduler: Scheduler::spawn(Scheduler::new()), utp_server, actor_ref, diff --git a/crates/libtortillas/src/torrent/swarm.rs b/crates/libtortillas/src/torrent/swarm.rs index 8c63d92f..7869746e 100644 --- a/crates/libtortillas/src/torrent/swarm.rs +++ b/crates/libtortillas/src/torrent/swarm.rs @@ -3,102 +3,109 @@ use kameo::{ actor::{ActorRef, Spawn}, mailbox, prelude::Message, + supervision::RestartPolicy, }; +use tokio::sync::oneshot; use tracing::{debug, instrument, trace, warn}; -use super::TorrentActor; -#[cfg(feature = "live")] -use crate::live::{PeerIdentity, PeerView}; +use super::{ConfiguredTracker, ConnectedPeer, TorrentActor}; use crate::{ + errors::{TorrentError, map_torrent_send_error}, peer::{PeerActor, PeerActorArgs, PeerId, WirePeer}, protocol::{ messages::{Handshake, PeerMessages}, stream::{PeerSend, PeerStream, validate_handshake}, }, torrent::events::PeerConnected, - tracker::{Tracker, TrackerActor, TrackerUpdate}, + tracker::{Tracker, TrackerActor, TrackerActorArgs, TrackerUpdate}, +}; +#[cfg(feature = "live")] +use crate::{ + live::{PeerIdentity, PeerView, TrackerStatus, TrackerView}, + metrics::TrackerMetrics, }; impl TorrentActor { #[instrument(skip(self, peer, stream), fields(%self, peer_addr = ?peer.socket_addr(), torrent_id = %self.info_hash()))] - pub(super) fn append_peer(&self, mut peer: WirePeer, stream: Option<(PeerStream, [u8; 8])>) { + pub(super) fn append_peer(&self, peer: WirePeer, stream: Option<(PeerStream, [u8; 8])>) { + self.spawn_peer_connection(peer, stream, None); + } + + pub(super) fn spawn_peer_connection( + &self, mut peer: WirePeer, stream: Option<(PeerStream, [u8; 8])>, + result: Option>>, + ) { let info_hash = self.info_hash(); let actor_ref = self.actor_ref.clone(); let our_id = self.id; let utp_server = self.utp_server.clone(); + // Handshakes may involve network timeouts, so keep them outside the + // torrent actor's mailbox. Explicit additions receive the result through + // `result`; discovery paths remain fire-and-report. tokio::spawn(async move { let mut id = peer.id; - let (stream, reserved) = match stream { - Some((mut stream, reserved)) => { - let handshake = Handshake::new(info_hash, our_id); - if let Err(err) = stream.send(PeerMessages::Handshake(handshake)).await { - debug!(error = %err, peer_addr = %peer.socket_addr(), "Failed to send handshake to peer"); - return; + let connection = async { + let (stream, reserved) = match stream { + Some((mut stream, reserved)) => { + let handshake = Handshake::new(info_hash, our_id); + stream.send(PeerMessages::Handshake(handshake)).await?; + (stream, reserved) } - (stream, reserved) - } - None => { - let stream = PeerStream::connect(peer.socket_addr(), Some(utp_server)).await; - match stream { - Ok(mut stream) => match stream.send_handshake(our_id, info_hash).await { - Ok(_) => match stream.recv_handshake_message().await { - Ok(handshake) => { - if let Err(err) = - validate_handshake(&handshake, peer.socket_addr(), info_hash) - { - trace!(error = %err, peer_addr = %peer.socket_addr(), "Failed to validate peer handshake; exiting"); - return; - } - id = Some(handshake.peer_id); - (stream, handshake.reserved) - } - Err(err) => { - trace!(error = %err, peer_addr = %peer.socket_addr(), "Failed to receive handshake from peer; exiting"); - return; - } - }, - Err(err) => { - trace!(error = %err, peer_addr = %peer.socket_addr(), "Failed to send handshake to peer; exiting"); - return; - } - }, - Err(err) => { - trace!(error = %err, peer_addr = %peer.socket_addr(), "Failed to connect to peer; exiting"); - return; - } + None => { + let mut stream = + PeerStream::connect(peer.socket_addr(), Some(utp_server)).await?; + stream.send_handshake(our_id, info_hash).await?; + let handshake = stream.recv_handshake_message().await?; + validate_handshake(&handshake, peer.socket_addr(), info_hash)?; + id = Some(handshake.peer_id); + (stream, handshake.reserved) } + }; + + let id = id.ok_or_else(|| TorrentError::InvalidOperation { + operation: "add peer", + reason: "peer connection completed without a peer id".to_string(), + })?; + + if id == our_id { + return Err(TorrentError::InvalidOperation { + operation: "add peer", + reason: "a torrent cannot connect to its own peer id".to_string(), + }); } - }; - let Some(id) = id else { - trace!(peer_addr = %peer.socket_addr(), "Peer connection completed without a peer id; exiting"); - return; - }; + peer.id = Some(id); - if id == our_id { - return; + // Registration is the boundary where the connection becomes part + // of the swarm, so a manual add does not succeed before this ask. + actor_ref + .ask(PeerConnected { + peer, + reserved, + stream, + }) + .await + .map_err(|error| map_torrent_send_error("register connected peer", error)) } + .await; - peer.id = Some(id); - - if let Err(err) = actor_ref - .tell(PeerConnected { - peer, - reserved, - stream, - }) - .await - { - warn!(?err, peer_id = %id, "Failed to route connected peer back to torrent actor"); + if let Some(result_tx) = result { + let _ = result_tx.send(connection); + } else if let Err(error) = connection { + debug!(%error, "Failed to add discovered peer"); } }); } - pub(super) fn insert_peer(&mut self, peer: WirePeer, reserved: [u8; 8], stream: PeerStream) { + pub(super) fn insert_peer( + &mut self, peer: WirePeer, reserved: [u8; 8], stream: PeerStream, + ) -> Result { let Some(id) = peer.id else { - trace!(peer_addr = %peer.socket_addr(), "Connected peer missing peer id; ignoring"); - return; + return Err(TorrentError::InvalidOperation { + operation: "register connected peer", + reason: "connected peer is missing its peer id".to_string(), + }); }; let actor_ref = self.actor_ref.clone(); @@ -106,7 +113,7 @@ impl TorrentActor { let peer_settings = self.settings.peer.clone(); let peer_mailbox_size = self.settings.torrent.peer_mailbox_size; if self.peers.contains_key(&id) { - return; + return Err(TorrentError::PeerAlreadyConnected { peer_id: id }); } #[cfg(feature = "live")] @@ -117,7 +124,10 @@ impl TorrentActor { }, PeerView::connected(peer.socket_addr(), id), ) else { - return; + return Err(TorrentError::ActorCommunicationFailed { + operation: "register peer live scope", + reason: "live hub is no longer available".to_string(), + }); }; let peer_args = PeerActorArgs { @@ -140,6 +150,74 @@ impl TorrentActor { self.peers.insert(id, peer_actor); self.publish_updated(); crate::live_only!(self.hub.emit_peer_connected(&peer_handle)); + #[cfg(feature = "live")] + return Ok(peer_handle); + #[cfg(not(feature = "live"))] + Ok(id) + } + + pub(super) async fn register_tracker_actor( + &mut self, tracker: Tracker, + ) -> Result { + // Startup trackers and runtime additions share this path so the actor + // registry and live projection cannot drift apart. + let endpoint = tracker.redacted_endpoint(); + if self.trackers.contains_key(&tracker) { + return Err(TorrentError::TrackerAlreadyExists { endpoint }); + } + if matches!(tracker, Tracker::Websocket(_)) { + return Err(TorrentError::UnsupportedTrackerProtocol { + protocol: "websocket", + }); + } + + #[cfg(feature = "live")] + let Some(tracker_handle) = self.hub.register_tracker_scope( + self.info_hash(), + &tracker, + TrackerView { + endpoint: endpoint.clone(), + status: TrackerStatus::Pending, + metrics: TrackerMetrics::default(), + }, + ) else { + return Err(TorrentError::ActorCommunicationFailed { + operation: "register tracker live scope", + reason: "live hub is no longer available".to_string(), + }); + }; + + let actor = TrackerActor::supervise( + &self.actor_ref, + TrackerActorArgs { + tracker: tracker.clone(), + peer_id: self.id, + server: self.tracker_server.clone(), + socket_addr: self.primary_addr, + initial_left: self + .tracker_announce_progress() + .map(|progress| progress.left), + supervisor: self.actor_ref.clone(), + scheduler: self.scheduler.clone(), + settings: self.settings.tracker.clone(), + #[cfg(feature = "live")] + live_handle: tracker_handle.clone(), + }, + ) + .restart_policy(RestartPolicy::Transient) + .restart_limit( + self.settings.torrent.tracker_restart.limit, + self.settings.torrent.tracker_restart.period, + ) + .spawn() + .await; + + self.trackers.insert(tracker, actor); + self.publish_updated(); + #[cfg(feature = "live")] + return Ok(tracker_handle); + #[cfg(not(feature = "live"))] + Ok(()) } #[instrument(skip(self, tell), fields(torrent_id = %self.info_hash(), msg = ?tell))] From c325c651ea531c45c9e50c9338a7952275ca9e7b Mon Sep 17 00:00:00 2001 From: artrixdotdev Date: Wed, 29 Jul 2026 00:21:16 -0700 Subject: [PATCH 2/3] ci: use portable build target --- .github/workflows/checks.yml | 1 - 1 file changed, 1 deletion(-) diff --git a/.github/workflows/checks.yml b/.github/workflows/checks.yml index 43000bfa..f6a0996a 100644 --- a/.github/workflows/checks.yml +++ b/.github/workflows/checks.yml @@ -9,7 +9,6 @@ on: env: CARGO_TERM_COLOR: always - RUSTFLAGS: "-C target-cpu=native" CARGO_INCREMENTAL: 0 RUST_BACKTRACE: 1 From 7e629c1b671803208577cd1476eb012a104839a7 Mon Sep 17 00:00:00 2001 From: artrixdotdev Date: Sun, 2 Aug 2026 14:54:54 -0700 Subject: [PATCH 3/3] fix: bound peer handshakes and tracker announcements - Prevent tracker announcements from blocking the torrent actor - Enforce incoming peer handshake timeouts - Stabilize peer disconnect test timing --- crates/libtortillas/src/lib.rs | 2 + crates/libtortillas/src/torrent/handle.rs | 6 ++- crates/libtortillas/src/torrent/messages.rs | 2 +- crates/libtortillas/src/torrent/swarm.rs | 41 +++++++++++++-------- 4 files changed, 33 insertions(+), 18 deletions(-) diff --git a/crates/libtortillas/src/lib.rs b/crates/libtortillas/src/lib.rs index 8f64e081..1b838ea2 100644 --- a/crates/libtortillas/src/lib.rs +++ b/crates/libtortillas/src/lib.rs @@ -571,6 +571,8 @@ pub(crate) mod testing { stream.write_all(&message.to_bytes()?).await?; } + // Keep the connection open so the peer does not observe EOF until its + // accept task and JoinSet are aborted by LocalPeer::drop. std::future::pending().await } diff --git a/crates/libtortillas/src/torrent/handle.rs b/crates/libtortillas/src/torrent/handle.rs index 78ca281e..88e8f035 100644 --- a/crates/libtortillas/src/torrent/handle.rs +++ b/crates/libtortillas/src/torrent/handle.rs @@ -269,7 +269,11 @@ mod tests { assert!(!listener.view().connected); assert_eq!( - listener.recv().await.unwrap().kind, + timeout(Duration::from_secs(2), listener.recv()) + .await + .expect("peer disconnected event timed out") + .unwrap() + .kind, PeerEventKind::Disconnected ); assert!(torrent.peers().is_empty()); diff --git a/crates/libtortillas/src/torrent/messages.rs b/crates/libtortillas/src/torrent/messages.rs index c02c1882..2242c42a 100644 --- a/crates/libtortillas/src/torrent/messages.rs +++ b/crates/libtortillas/src/torrent/messages.rs @@ -258,7 +258,7 @@ pub(crate) mod commands { let mut queued = 0usize; let mut first_error = None; for (tracker, actor) in &self.trackers { - match actor.tell(Announce).await { + match actor.tell(Announce).try_send() { Ok(()) => queued = queued.saturating_add(1), Err(error) => { first_error.get_or_insert_with(|| TorrentError::ActorCommunicationFailed { diff --git a/crates/libtortillas/src/torrent/swarm.rs b/crates/libtortillas/src/torrent/swarm.rs index 7869746e..4abf8680 100644 --- a/crates/libtortillas/src/torrent/swarm.rs +++ b/crates/libtortillas/src/torrent/swarm.rs @@ -5,7 +5,7 @@ use kameo::{ prelude::Message, supervision::RestartPolicy, }; -use tokio::sync::oneshot; +use tokio::{sync::oneshot, time::timeout}; use tracing::{debug, instrument, trace, warn}; use super::{ConfiguredTracker, ConnectedPeer, TorrentActor}; @@ -39,6 +39,7 @@ impl TorrentActor { let actor_ref = self.actor_ref.clone(); let our_id = self.id; let utp_server = self.utp_server.clone(); + let handshake_timeout = self.settings.engine.incoming_peer_handshake_timeout; // Handshakes may involve network timeouts, so keep them outside the // torrent actor's mailbox. Explicit additions receive the result through @@ -46,22 +47,30 @@ impl TorrentActor { tokio::spawn(async move { let mut id = peer.id; let connection = async { - let (stream, reserved) = match stream { - Some((mut stream, reserved)) => { - let handshake = Handshake::new(info_hash, our_id); - stream.send(PeerMessages::Handshake(handshake)).await?; - (stream, reserved) - } - None => { - let mut stream = - PeerStream::connect(peer.socket_addr(), Some(utp_server)).await?; - stream.send_handshake(our_id, info_hash).await?; - let handshake = stream.recv_handshake_message().await?; - validate_handshake(&handshake, peer.socket_addr(), info_hash)?; - id = Some(handshake.peer_id); - (stream, handshake.reserved) + let (stream, reserved, handshake_id) = timeout(handshake_timeout, async { + match stream { + Some((mut stream, reserved)) => { + let handshake = Handshake::new(info_hash, our_id); + stream.send(PeerMessages::Handshake(handshake)).await?; + Ok::<_, TorrentError>((stream, reserved, None)) + } + None => { + let mut stream = + PeerStream::connect(peer.socket_addr(), Some(utp_server)).await?; + stream.send_handshake(our_id, info_hash).await?; + let handshake = stream.recv_handshake_message().await?; + validate_handshake(&handshake, peer.socket_addr(), info_hash)?; + Ok::<_, TorrentError>((stream, handshake.reserved, Some(handshake.peer_id))) + } } - }; + }) + .await + .map_err(|_| { + TorrentError::PeerActor(crate::errors::PeerActorError::PeerTimeout { + seconds: handshake_timeout.as_secs().max(1), + }) + })??; + id = id.or(handshake_id); let id = id.ok_or_else(|| TorrentError::InvalidOperation { operation: "add peer",