From ed88070a68438cfc936aa28ffccfd6bf9fc88dc2 Mon Sep 17 00:00:00 2001 From: artrixdotdev Date: Wed, 1 Jul 2026 21:41:24 -0700 Subject: [PATCH 1/4] fix(#200): initialize tracker announces without lifecycle events --- crates/libtortillas/src/tracker/http.rs | 45 ++++++++++++++++++++--- crates/libtortillas/src/tracker/udp.rs | 47 +++++++++++++++++++++++-- 2 files changed, 85 insertions(+), 7 deletions(-) diff --git a/crates/libtortillas/src/tracker/http.rs b/crates/libtortillas/src/tracker/http.rs index a19fa48..5e9762e 100644 --- a/crates/libtortillas/src/tracker/http.rs +++ b/crates/libtortillas/src/tracker/http.rs @@ -118,8 +118,8 @@ impl TrackerRequest { port, uploaded: 0, downloaded: 0, - left: Some(0), - event: Event::Stopped, + left: None, + event: Event::Empty, compact, } } @@ -456,11 +456,14 @@ where #[cfg(test)] mod tests { - use std::net::{IpAddr, Ipv4Addr}; + use std::{ + net::{IpAddr, Ipv4Addr, SocketAddr}, + str::FromStr, + }; use tracing_test::traced_test; - use super::{HttpTracker, TrackerResponse, parse_tracker_response}; + use super::{HttpTracker, TrackerRequest, TrackerResponse, parse_tracker_response}; use crate::{ errors::TrackerActorError, metainfo::MetaInfo, @@ -471,6 +474,40 @@ mod tests { tracker::TrackerBase, }; + #[test] + fn tracker_request_when_created_then_omits_unknown_left_and_lifecycle_event() { + let default_peer_addr = "0.0.0.0:6881".parse().unwrap(); + let tracker_request = TrackerRequest::new(None, default_peer_addr, true); + + let query = tracker_request.to_string(); + + assert!(query.contains("port=6881")); + assert!(query.contains("uploaded=0")); + assert!(query.contains("downloaded=0")); + assert!(!query.contains("left=")); + assert!(!query.contains("event=")); + } + + #[test] + fn tracker_request_when_updated_with_lifecycle_state_then_serializes_values() { + let peer_addr = SocketAddr::from_str("192.0.2.10:51413").unwrap(); + let mut tracker_request = TrackerRequest::new(Some(peer_addr), peer_addr, false); + tracker_request.uploaded = 21; + tracker_request.downloaded = 34; + tracker_request.left = Some(55); + tracker_request.event = crate::tracker::Event::Started; + + let query = tracker_request.to_string(); + + assert!(query.contains("ip=192.0.2.10")); + assert!(query.contains("port=51413")); + assert!(query.contains("uploaded=21")); + assert!(query.contains("downloaded=34")); + assert!(query.contains("left=55")); + assert!(query.contains("event=started")); + assert!(query.contains("compact=0")); + } + #[test] fn parse_tracker_response_when_failure_reason_is_present_then_returns_tracker_error() { let error = parse_tracker_response(b"d14:failure reason12:tracker downe").unwrap_err(); diff --git a/crates/libtortillas/src/tracker/udp.rs b/crates/libtortillas/src/tracker/udp.rs index d2f1407..5c17cc9 100644 --- a/crates/libtortillas/src/tracker/udp.rs +++ b/crates/libtortillas/src/tracker/udp.rs @@ -526,7 +526,7 @@ pub enum Action { #[derive(Debug, Default)] struct AnnounceParams { /// Bytes left to download. - left: u32, + left: u64, /// Total downloaded bytes. downloaded: u64, /// Total uploaded bytes. @@ -894,7 +894,7 @@ impl TrackerBase for UdpTracker { let params = self.announce_params.read().await; ( params.downloaded, - params.left as u64, + params.left, params.uploaded, params.event, ) @@ -999,7 +999,7 @@ impl TrackerBase for UdpTracker { params.downloaded = bytes as u64; } TrackerUpdate::Left(bytes) => { - params.left = bytes as u32; + params.left = bytes as u64; } TrackerUpdate::Event(event) => { params.event = event; @@ -1050,6 +1050,47 @@ mod tests { const EXTERNAL_TRACKER_ATTEMPT_TIMEOUT: Duration = Duration::from_secs(8); + #[tokio::test] + async fn udp_tracker_when_updated_then_preserves_full_announce_state() { + let tracker = UdpTracker { + addr: SocketAddr::from(([127, 0, 0, 1], 8080)), + uri: "udp://127.0.0.1:8080".to_string(), + server: udp_server().await, + connection_id: Arc::new(AtomicU64::new(0)), + ready_state: Arc::new(AtomicU8::new(ReadyState::Disconnected as u8)), + peer_id: PeerId::new(), + info_hash: InfoHash::new([0; 20]), + peer_addr: random_socket_addr(), + interval: Arc::new(AtomicU32::new(u32::MAX)), + stats: TrackerStats::default(), + announce_params: Arc::new(RwLock::new(AnnounceParams::default())), + settings: TrackerSettings::default(), + }; + + tracker + .update(TrackerUpdate::Uploaded(21)) + .await + .expect("uploaded update should succeed"); + tracker + .update(TrackerUpdate::Downloaded(34)) + .await + .expect("downloaded update should succeed"); + tracker + .update(TrackerUpdate::Left(u32::MAX as usize + 1)) + .await + .expect("left update should succeed"); + tracker + .update(TrackerUpdate::Event(Event::Started)) + .await + .expect("event update should succeed"); + + let params = tracker.announce_params.read().await; + assert_eq!(params.uploaded, 21); + assert_eq!(params.downloaded, 34); + assert_eq!(params.left, u32::MAX as u64 + 1); + assert_eq!(params.event, Event::Started); + } + #[tokio::test] #[ignore = "external-network test: reaches public UDP trackers"] async fn udp_tracker_when_public_magnet_tracker_is_available_then_returns_ipv4_peer() { From 3b73cb087be3f63ff27f21e4a50678b156439821 Mon Sep 17 00:00:00 2001 From: artrixdotdev Date: Wed, 1 Jul 2026 21:43:49 -0700 Subject: [PATCH 2/4] fix(#200): sync tracker announce progress before events --- crates/libtortillas/src/torrent/actor.rs | 76 +++++++++++++++---- crates/libtortillas/src/torrent/piece_flow.rs | 8 +- crates/libtortillas/src/tracker/actor.rs | 6 ++ 3 files changed, 68 insertions(+), 22 deletions(-) diff --git a/crates/libtortillas/src/torrent/actor.rs b/crates/libtortillas/src/torrent/actor.rs index 59f79bf..2485b7e 100644 --- a/crates/libtortillas/src/torrent/actor.rs +++ b/crates/libtortillas/src/torrent/actor.rs @@ -47,6 +47,12 @@ pub(super) enum PieceManagerProxy { Default(FilePieceManager), } +#[derive(Debug, Clone, Copy, Eq, PartialEq)] +pub(super) struct TrackerAnnounceProgress { + pub(super) downloaded: usize, + pub(super) left: usize, +} + impl Display for PieceManagerProxy { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { match self { @@ -204,12 +210,15 @@ impl TorrentActor { return; } + self.sync_tracker_announce_progress().await; + if self.is_full() { self.state = TorrentState::Seeding; info!(id = %self.info_hash(), "Torrent is now seeding"); self - .update_trackers(TrackerUpdate::Event(Event::Completed)) + .update_trackers(TrackerUpdate::Event(Event::Empty)) .await; + self.broadcast_to_trackers(Announce).await; } else { self.state = TorrentState::Downloading; info!(id = %self.info_hash(), "Torrent is now downloading"); @@ -345,6 +354,29 @@ impl TorrentActor { Some(total_bytes) } + pub(super) fn tracker_announce_progress(&self) -> Option { + let info = self.info_dict()?; + let total_length = info.total_length(); + let downloaded = self.total_bytes_downloaded()?.min(total_length); + Some(TrackerAnnounceProgress { + downloaded, + left: total_length.saturating_sub(downloaded), + }) + } + + pub(super) async fn sync_tracker_announce_progress(&mut self) { + let Some(progress) = self.tracker_announce_progress() else { + return; + }; + + self + .update_trackers(TrackerUpdate::Downloaded(progress.downloaded)) + .await; + self + .update_trackers(TrackerUpdate::Left(progress.left)) + .await; + } + pub fn export(&self) -> TorrentExport { TorrentExport { info_hash: self.info_hash(), @@ -480,6 +512,22 @@ impl Actor for TorrentActor { .spawn() .await; + let info = match &metainfo { + MetaInfo::Torrent(t) => Some(t.info.clone()), + _ => None, + }; + if info.is_none() { + debug!(torrent_id = %torrent_id, "No info dict found in metainfo, you're probably using a magnet uri"); + } + let piece_count = if let Some(info) = &info { + debug!(torrent_id = %torrent_id, "Using bitfield length {}", info.piece_count()); + info.piece_count() + } else { + 0 + }; + let bitfield = BitVec::repeat(false, piece_count); + let initial_left = info.as_ref().map(Info::total_length); + // Create tracker actors let tracker_list = metainfo.announce_list(); let mut trackers = HashMap::new(); @@ -491,6 +539,7 @@ impl Actor for TorrentActor { peer_id, server: tracker_server.clone(), socket_addr: primary_addr, + initial_left, supervisor: us.clone(), scheduler: scheduler.clone(), settings: settings.tracker.clone(), @@ -506,20 +555,6 @@ impl Actor for TorrentActor { trackers.insert(tracker, actor); } - let info = match &metainfo { - MetaInfo::Torrent(t) => Some(t.info.clone()), - _ => None, - }; - if info.is_none() { - debug!(torrent_id = %torrent_id, "No info dict found in metainfo, you're probably using a magnet uri"); - } - let piece_count = if let Some(info) = &info { - debug!(torrent_id = %torrent_id, "Using bitfield length {}", info.piece_count()); - info.piece_count() - } else { - 0 - }; - let bitfield = BitVec::repeat(false, piece_count); let default_manager = FilePieceManager(base_path, info.clone()); let piece_store = PieceStoreActor::supervise(&us, ()) .restart_policy(RestartPolicy::Permanent) @@ -889,6 +924,17 @@ mod tests { let partial_entry = export.block_map.get(&partial_piece_index).unwrap(); assert_eq!(partial_entry.count_ones(), partial_blocks_received); + let announce_progress = test_actor + .tracker_announce_progress() + .expect("tracker progress should be available with an info dict"); + let expected_downloaded = (fake_completed * info_dict.piece_length as usize) + + (partial_blocks_received * BLOCK_SIZE); + assert_eq!(announce_progress.downloaded, expected_downloaded); + assert_eq!( + announce_progress.left, + info_dict.total_length() - expected_downloaded + ); + match &export.piece_storage { PieceStorageStrategy::Disk(p) => assert_eq!(p, &piece_path), _ => panic!("Expected Disk storage strategy"), diff --git a/crates/libtortillas/src/torrent/piece_flow.rs b/crates/libtortillas/src/torrent/piece_flow.rs index e62f7f4..506a664 100644 --- a/crates/libtortillas/src/torrent/piece_flow.rs +++ b/crates/libtortillas/src/torrent/piece_flow.rs @@ -218,7 +218,6 @@ impl TorrentActor { .info_dict() .expect("Can't receive piece without info dict"); let piece_count = info_dict.piece_count(); - let total_length = info_dict.total_length(); if !self .validate_and_send_piece(peer_id, index, previous_blocks) @@ -238,12 +237,7 @@ impl TorrentActor { self.broadcast_to_peers(Have { piece: index }).await; - if let Some(total_downloaded) = self.total_bytes_downloaded() { - let total_bytes_left = total_length - total_downloaded; - self - .update_trackers(TrackerUpdate::Left(total_bytes_left)) - .await; - } + self.sync_tracker_announce_progress().await; if self.piece_scheduler.next_piece() >= piece_count { self.state = TorrentState::Seeding; diff --git a/crates/libtortillas/src/tracker/actor.rs b/crates/libtortillas/src/tracker/actor.rs index 1a77e86..81854db 100644 --- a/crates/libtortillas/src/tracker/actor.rs +++ b/crates/libtortillas/src/tracker/actor.rs @@ -40,6 +40,7 @@ pub(crate) struct TrackerActorArgs { pub(crate) peer_id: PeerId, pub(crate) server: UdpServer, pub(crate) socket_addr: SocketAddr, + pub(crate) initial_left: Option, pub(crate) supervisor: ActorRef, pub(crate) scheduler: ActorRef, pub(crate) settings: TrackerSettings, @@ -55,6 +56,7 @@ impl Actor for TrackerActor { peer_id, server, socket_addr, + initial_left, supervisor, scheduler, settings, @@ -108,6 +110,10 @@ impl Actor for TrackerActor { } }; + if let Some(left) = initial_left { + tracker.update(TrackerUpdate::Left(left)).await?; + } + let next_announce = scheduler .ask(SetTimeout::new( actor_ref.downgrade(), From b5e4def6988175ada4f18d8e6d503c35cf695800 Mon Sep 17 00:00:00 2001 From: artrixdotdev Date: Wed, 1 Jul 2026 21:46:28 -0700 Subject: [PATCH 3/4] test(#200): cover torrent start tracker announce state --- crates/libtortillas/src/torrent/actor.rs | 100 ++++++++++++++++++++++- 1 file changed, 99 insertions(+), 1 deletion(-) diff --git a/crates/libtortillas/src/torrent/actor.rs b/crates/libtortillas/src/torrent/actor.rs index 2485b7e..ae745e6 100644 --- a/crates/libtortillas/src/torrent/actor.rs +++ b/crates/libtortillas/src/torrent/actor.rs @@ -641,6 +641,9 @@ mod tests { use librqbit_utp::UtpSocket; use tokio::{ fs, + io::{AsyncReadExt, AsyncWriteExt}, + net::TcpListener, + sync::oneshot, time::{sleep, timeout}, }; use tracing::trace; @@ -652,10 +655,105 @@ mod tests { testing, torrent::{ BLOCK_SIZE, Torrent, TorrentExport, - commands::{ExportState, HasInfoDict}, + commands::{ExportState, HasInfoDict, SetState}, }, + tracker::Tracker, }; + async fn one_shot_http_tracker() -> (Tracker, oneshot::Receiver) { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + let (query_tx, query_rx) = oneshot::channel(); + + tokio::spawn(async move { + let (mut stream, _) = listener.accept().await.unwrap(); + let mut buf = [0; 4096]; + let read = stream.read(&mut buf).await.unwrap(); + let request = String::from_utf8_lossy(&buf[..read]); + let request_target = request + .lines() + .next() + .and_then(|line| line.split_whitespace().nth(1)) + .unwrap_or_default(); + let query = request_target + .split_once('?') + .map(|(_, query)| query.to_string()) + .unwrap_or_default(); + let _ = query_tx.send(query); + + let body = b"d8:intervali60e5:peers0:e"; + let response = format!( + "HTTP/1.1 200 OK\r\nContent-Length: {}\r\nConnection: close\r\n\r\n", + body.len() + ); + stream.write_all(response.as_bytes()).await.unwrap(); + stream.write_all(body).await.unwrap(); + }); + + (Tracker::Http(format!("http://{addr}/announce")), query_rx) + } + + #[tokio::test(flavor = "multi_thread")] + async fn torrent_actor_when_started_then_announces_started_with_current_progress() { + testing::init_tracing(); + let (tracker, query_rx) = one_shot_http_tracker().await; + let mut metainfo = testing::read_torrent_fixture(testing::BIG_BUCK_BUNNY_TORRENT_FILE).await; + let info = match &mut metainfo { + MetaInfo::Torrent(file) => { + file.announce = tracker; + file.announce_list = None; + file.info.clone() + } + _ => unreachable!(), + }; + let info_hash = info.hash().unwrap(); + let peer_id = testing::peer_id(); + let piece_path = testing::torrent_temp_path(); + let file_path = piece_path.join("files"); + let udp_server = testing::udp_server().await; + let utp_server = UtpSocket::new_udp(testing::ephemeral_socket_addr()) + .await + .unwrap(); + let mut settings = Settings::default(); + settings.tracker.initial_announce_delay = Duration::from_secs(60); + + let actor = TorrentActor::spawn(TorrentActorArgs { + peer_id, + metainfo, + utp_server, + tracker_server: udp_server, + primary_addr: None, + piece_storage: PieceStorageStrategy::default(), + autostart: Some(false), + sufficient_peers: Some(usize::MAX), + base_path: Some(file_path), + settings, + }); + actor + .tell(SetState { + state: TorrentState::Downloading, + }) + .await + .unwrap(); + + let query = timeout(Duration::from_secs(5), query_rx) + .await + .expect("tracker should receive a start announce") + .expect("tracker query should be captured"); + + assert!(query.contains("event=started")); + assert!(query.contains("uploaded=0")); + assert!(query.contains("downloaded=0")); + assert!(query.contains(&format!("left={}", info.total_length()))); + assert!(query.contains("compact=0")); + + let export = actor.ask(ExportState).await.unwrap(); + assert_eq!(export.info_hash, info_hash); + assert_eq!(export.state, TorrentState::Downloading); + + actor.stop_gracefully().await.unwrap(); + } + #[tokio::test(flavor = "multi_thread")] #[ignore = "external-network test: reaches public trackers and peers"] async fn torrent_actor_when_public_torrent_is_available_then_reaches_ready_state() { From b5d4bc3effee5d69a04cb493b27623a58e6dcb97 Mon Sep 17 00:00:00 2001 From: artrixdotdev Date: Wed, 1 Jul 2026 21:47:51 -0700 Subject: [PATCH 4/4] test(#200): send explicit left in HTTP tracker network tests --- crates/libtortillas/src/tracker/http.rs | 10 +++++++++- 1 file changed, 9 insertions(+), 1 deletion(-) diff --git a/crates/libtortillas/src/tracker/http.rs b/crates/libtortillas/src/tracker/http.rs index 5e9762e..9b9dc77 100644 --- a/crates/libtortillas/src/tracker/http.rs +++ b/crates/libtortillas/src/tracker/http.rs @@ -471,7 +471,7 @@ mod tests { testing::{ KNOPPIX_TORRENT_FILE, init_tracing, random_port, read_torrent_fixture, udp_server, }, - tracker::TrackerBase, + tracker::{TrackerBase, TrackerUpdate}, }; #[test] @@ -563,6 +563,10 @@ mod tests { // An HTTP tracker let announce_uri = announce_list[1].uri(); let http_tracker = HttpTracker::new(announce_uri, info_hash.unwrap(), None, None); + http_tracker + .update(TrackerUpdate::Left(file.info.total_length())) + .await + .unwrap(); // Spawn a task to re-fetch the latest list of peers at a given interval let peers = http_tracker.announce().await.unwrap(); @@ -597,6 +601,10 @@ mod tests { .to_instance(info_hash, peer_id, port, server) .await .unwrap(); + tracker + .update(TrackerUpdate::Left(torrent.info.total_length())) + .await + .unwrap(); // Use the announce_stream method from TrackerInstance trait let peers = tracker.announce().await.unwrap();