fix: correct BEP 3 tracker announce state

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<TrackerAnnounceProgress> {
+      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)
@@ -606,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;
@@ -617,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<String>) {
+      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() {
@@ -889,6 +1022,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"),

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;

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<usize>,
    pub(crate) supervisor: ActorRef<TorrentActor>,
    pub(crate) scheduler: ActorRef<Scheduler>,
    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(),

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,21 +456,58 @@ 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,
       peer::PeerId,
       testing::{
          KNOPPIX_TORRENT_FILE, init_tracing, random_port, read_torrent_fixture, udp_server,
       },
-      tracker::TrackerBase,
+      tracker::{TrackerBase, TrackerUpdate},
    };
 
+   #[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();
@@ -526,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();
@@ -560,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();