diff --git a/crates/libtortillas/src/ARCHITECTURE.md b/crates/libtortillas/src/ARCHITECTURE.md index 6718098..f152b1e 100644 --- a/crates/libtortillas/src/ARCHITECTURE.md +++ b/crates/libtortillas/src/ARCHITECTURE.md @@ -9,3 +9,11 @@ The library is organized around a small actor hierarchy: Module facades should export stable public types while keeping actor internals private to the crate. Domain types such as torrent state, storage strategy, exported snapshots, tracker model types, and tracker stats live outside actor files so actors can focus on orchestration. + +## Choking + +`TorrentActor` owns the BEP 3 choking scheduler for its swarm. Active torrents run a rechoke round every 10 seconds, collect peer-local transfer stats from `PeerActor`, and keep at most four interested peers unchoked. + +While downloading, regular upload slots are assigned to interested peers with the highest recent download rate. While seeding, regular slots are assigned by recent upload rate. When more interested peers exist than upload slots, one slot is reserved for an optimistic unchoke and rotates every third rechoke round. + +`PeerActor` remains responsible for wire-level enforcement: it sends `Choke` and `Unchoke` messages when the torrent scheduler changes state, ignores piece requests from choked peers, and records uploaded bytes when serving piece data. diff --git a/crates/libtortillas/src/peer/actor.rs b/crates/libtortillas/src/peer/actor.rs index d543df1..5dc4d2d 100644 --- a/crates/libtortillas/src/peer/actor.rs +++ b/crates/libtortillas/src/peer/actor.rs @@ -22,7 +22,7 @@ use tracing::{Span, debug, info, instrument, trace, warn}; use crate::{ errors::PeerActorError, hashes::InfoHash, - peer::Peer, + peer::{Peer, PeerId}, protocol::{stream::PeerRecv, *}, torrent::{self, TorrentActor}, }; @@ -32,6 +32,34 @@ const MAX_PENDING_MESSAGES: usize = 8; const PEER_KEEPALIVE_TIMEOUT: u64 = 120; const PEER_DISCONNECT_TIMEOUT: u64 = 240; +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub(crate) struct PeerStats { + pub(crate) id: PeerId, + pub(crate) interested: bool, + pub(crate) choked: bool, + pub(crate) download_rate: usize, + pub(crate) upload_rate: usize, + pub(crate) bytes_downloaded: usize, + pub(crate) bytes_uploaded: usize, +} + +#[derive(Clone, Copy, Debug)] +struct RateSample { + at: Instant, + bytes_downloaded: usize, + bytes_uploaded: usize, +} + +impl RateSample { + fn new(peer: &Peer) -> Self { + Self { + at: Instant::now(), + bytes_downloaded: peer.bytes_downloaded(), + bytes_uploaded: peer.bytes_uploaded(), + } + } +} + /// The actor that handles all communications with a given peer. pub(crate) struct PeerActor { /// The peers state and statistics @@ -43,6 +71,7 @@ pub(crate) struct PeerActor { pending_block_requests: HashSet<(usize, usize, usize)>, pending_message_requests: VecDeque, + last_rate_sample: RateSample, } impl PeerActor { @@ -306,6 +335,41 @@ impl PeerActor { self.stream.send(msg).await } + + fn snapshot_stats(&mut self) -> Option { + let id = self.peer.id?; + let now = Instant::now(); + let bytes_downloaded = self.peer.bytes_downloaded(); + let bytes_uploaded = self.peer.bytes_uploaded(); + let elapsed_secs = now + .duration_since(self.last_rate_sample.at) + .as_secs() + .max(1) as usize; + + let download_rate = bytes_downloaded.saturating_sub(self.last_rate_sample.bytes_downloaded) + / 1024 + / elapsed_secs; + let upload_rate = + bytes_uploaded.saturating_sub(self.last_rate_sample.bytes_uploaded) / 1024 / elapsed_secs; + + self.peer.set_download_rate(download_rate); + self.peer.set_upload_rate(upload_rate); + self.last_rate_sample = RateSample { + at: now, + bytes_downloaded, + bytes_uploaded, + }; + + Some(PeerStats { + id, + interested: self.peer.interested(), + choked: self.peer.choked(), + download_rate, + upload_rate, + bytes_downloaded, + bytes_uploaded, + }) + } } impl Actor for PeerActor { @@ -340,6 +404,7 @@ impl Actor for PeerActor { .map_err(|e| PeerActorError::SupervisorCommunicationFailed(e.to_string()))?; Ok(Self { + last_rate_sample: RateSample::new(&peer), peer, stream, supervisor, @@ -431,17 +496,15 @@ impl Message for PeerActor { data_len = data.len(), "Received piece data" ); - self.peer.increment_bytes_downloaded(data.len()); let key = (index as usize, offset as usize, data.len()); // Only send the piece to the supervisor if they request it or it hasn't been // cancelled - if self.pending_block_requests.contains(&key) { - self.pending_block_requests.remove(&key); - + if self.pending_block_requests.remove(&key) { let Some(peer_id) = self.peer.id else { warn!("Received piece from peer without id; ignoring"); return; }; + self.peer.increment_bytes_downloaded(data.len()); let supervisor_msg = torrent::events::IncomingPiece { peer_id, index: index as usize, @@ -508,6 +571,14 @@ impl Message for PeerActor { } } PeerMessages::Request(index, offset, length) => { + if self.peer.choked() { + trace!( + piece_index = index, + offset, length, "Ignoring request from choked peer" + ); + return; + } + debug!( piece_index = index, offset, length, "Peer requested piece data" @@ -525,11 +596,13 @@ impl Message for PeerActor { match data { Some(data) => { + let uploaded_bytes = data.len(); self .stream .send(PeerMessages::Piece(index as u32, offset as u32, data)) .await .expect("Failed to send piece"); + self.peer.increment_bytes_uploaded(uploaded_bytes); } None => { warn!( @@ -682,6 +755,33 @@ pub(crate) mod commands { } } + #[message(derive(Clone, Debug))] + #[instrument(skip(self), fields(peer_addr = %self.stream, peer_id = %self.peer.id.unwrap()))] + pub(crate) async fn set_choked(&mut self, choked: bool) { + if self.peer.choked() == choked { + return; + } + + let message = if choked { + PeerMessages::Choke + } else { + PeerMessages::Unchoke + }; + + if let Err(err) = self.stream.send(message).await { + warn!(?err, choked, "Failed to update peer choke state"); + return; + } + + self.peer.set_choked(choked); + trace!(choked, "Updated peer choke state"); + } + + #[message] + pub(crate) fn stats(&mut self) -> Option { + self.snapshot_stats() + } + #[message(derive(Clone, Debug))] pub(crate) async fn have_info_dict(&mut self, bitfield: Arc>) { self diff --git a/crates/libtortillas/src/torrent/actor.rs b/crates/libtortillas/src/torrent/actor.rs index 8752eff..344489a 100644 --- a/crates/libtortillas/src/torrent/actor.rs +++ b/crates/libtortillas/src/torrent/actor.rs @@ -18,12 +18,15 @@ use kameo::{ mailbox::{MailboxReceiver, Signal}, supervision::{RestartPolicy, SupervisionStrategy}, }; -use kameo_actors::scheduler::Scheduler; +use kameo_actors::scheduler::{Scheduler, SetTimeout}; use librqbit_utp::UtpSocketUdp; -use tokio::sync::oneshot; +use tokio::{sync::oneshot, task::AbortHandle}; use tracing::{debug, error, info, instrument, trace, warn}; -use super::util; +use super::{ + choking::{ChokingScheduler, RECHOKE_INTERVAL}, + util, +}; use crate::{ errors::TorrentError, hashes::InfoHash, @@ -102,6 +105,9 @@ pub(crate) struct TorrentActor { pub state: TorrentState, /// Scheduler for managing piece and block requests pub(super) piece_scheduler: PieceScheduler, + /// Scheduler for BEP 3 upload choking decisions. + pub(super) choking_scheduler: ChokingScheduler, + pub(super) next_rechoke: Option, pub(super) start_time: Option, /// The number of peers we need to have before we start downloading, defaults @@ -235,6 +241,45 @@ impl TorrentActor { state = ?self.state, "Started torrenting process" ); + + self.rechoke_peers().await; + self.schedule_next_rechoke().await; + } + + pub(super) async fn schedule_next_rechoke(&mut self) { + if !matches!( + self.state, + TorrentState::Downloading | TorrentState::Seeding + ) { + return; + } + + if let Some(next_rechoke) = self.next_rechoke.take() { + next_rechoke.abort(); + } + + match self + .scheduler + .ask(SetTimeout::new( + self.actor_ref.downgrade(), + RECHOKE_INTERVAL, + super::commands::Rechoke, + )) + .await + { + Ok(next_rechoke) => self.next_rechoke = Some(next_rechoke), + Err(err) => { + warn!(?err, "Failed to schedule next rechoke; using local timeout"); + let actor_ref = self.actor_ref.clone(); + let fallback = tokio::spawn(async move { + tokio::time::sleep(RECHOKE_INTERVAL).await; + if let Err(err) = actor_ref.tell(super::commands::Rechoke).await { + warn!(?err, "Failed to run fallback rechoke"); + } + }); + self.next_rechoke = Some(fallback.abort_handle()); + } + } } pub(super) fn send_ready_hooks(&mut self) { @@ -482,6 +527,8 @@ impl Actor for TorrentActor { piece_store, state: TorrentState::default(), piece_scheduler: PieceScheduler::new(piece_count), + choking_scheduler: ChokingScheduler::default(), + next_rechoke: None, start_time: None, sufficient_peers: sufficient_peers.unwrap_or(6), autostart: autostart.unwrap_or(true), @@ -511,6 +558,9 @@ impl Actor for TorrentActor { for tracker in self.trackers.values() { tracker.kill(); } + if let Some(next_rechoke) = self.next_rechoke.take() { + next_rechoke.abort(); + } self.piece_store.kill(); self.scheduler.kill(); @@ -532,14 +582,20 @@ mod tests { use std::{path::PathBuf, time::Duration}; use librqbit_utp::UtpSocket; - use tokio::{fs, time::sleep}; + use tokio::{ + fs, + time::{sleep, timeout}, + }; use tracing::trace; use super::*; use crate::{ metainfo::MetaInfo, testing, - torrent::{BLOCK_SIZE, Torrent, TorrentExport, commands::HasInfoDict}, + torrent::{ + BLOCK_SIZE, Torrent, TorrentExport, + commands::{ExportState, HasInfoDict}, + }, }; #[tokio::test(flavor = "multi_thread")] @@ -672,27 +728,38 @@ mod tests { assert!(torrent.poll_ready().await.is_ok()); - loop { - let mut entries = fs::read_dir(&piece_path).await.unwrap(); - let mut found_piece = false; - while let Some(entry) = entries.next_entry().await.unwrap() { - let path = entry.path(); - if let Some(ext) = path.extension() - && ext == "piece" - { - let metadata = entry.metadata().await.unwrap(); - if metadata.len() == info_dict.piece_length { - found_piece = true; - break; + let wrote_piece_block = timeout(Duration::from_secs(60), async { + loop { + let export = actor.ask(ExportState).await.unwrap(); + let has_persisted_progress = export.bitfield.count_ones() > 0 + || export + .block_map + .iter() + .any(|entry| entry.value().count_ones() > 0); + + if has_persisted_progress { + let mut entries = fs::read_dir(&piece_path).await.unwrap(); + while let Some(entry) = entries.next_entry().await.unwrap() { + let path = entry.path(); + if let Some(ext) = path.extension() + && ext == "piece" + && entry.metadata().await.unwrap().len() > 0 + { + return true; + } } } + + sleep(Duration::from_millis(200)).await; } - if found_piece { - break; - } + }) + .await + .unwrap_or(false); - sleep(Duration::from_millis(200)).await; - } + assert!( + wrote_piece_block, + "timed out waiting for piece storage write" + ); actor.stop_gracefully().await.unwrap(); clear_piece_files(&piece_path).await; @@ -770,6 +837,8 @@ mod tests { )), state: TorrentState::Inactive, piece_scheduler, + choking_scheduler: ChokingScheduler::default(), + next_rechoke: None, start_time: None, sufficient_peers: 6, autostart: false, diff --git a/crates/libtortillas/src/torrent/choking.rs b/crates/libtortillas/src/torrent/choking.rs new file mode 100644 index 0000000..b5e351b --- /dev/null +++ b/crates/libtortillas/src/torrent/choking.rs @@ -0,0 +1,279 @@ +use std::time::Duration; + +use crate::{ + peer::{PeerId, PeerStats}, + torrent::TorrentState, +}; + +pub(crate) const DEFAULT_UPLOAD_SLOTS: usize = 4; +pub(crate) const RECHOKE_INTERVAL: Duration = Duration::from_secs(10); +pub(crate) const OPTIMISTIC_UNCHOKE_ROUNDS: usize = 3; + +#[derive(Clone, Debug)] +pub(crate) struct ChokingScheduler { + upload_slots: usize, + optimistic_unchoke_rounds: usize, + optimistic_index: usize, + rounds_since_optimistic_rotation: usize, +} + +impl Default for ChokingScheduler { + fn default() -> Self { + Self::new(DEFAULT_UPLOAD_SLOTS, OPTIMISTIC_UNCHOKE_ROUNDS) + } +} + +impl ChokingScheduler { + pub(crate) fn new(upload_slots: usize, optimistic_unchoke_rounds: usize) -> Self { + Self { + upload_slots, + optimistic_unchoke_rounds: optimistic_unchoke_rounds.max(1), + optimistic_index: 0, + rounds_since_optimistic_rotation: 0, + } + } + + pub(crate) fn decide( + &mut self, peers: &[PeerStats], torrent_state: TorrentState, + ) -> ChokingDecision { + let decision = select_unchoked_peers( + peers, + torrent_state, + self.upload_slots, + self.optimistic_index, + ); + + if decision.optimistic.is_some() { + self.rounds_since_optimistic_rotation += 1; + if self.rounds_since_optimistic_rotation >= self.optimistic_unchoke_rounds { + self.rounds_since_optimistic_rotation = 0; + self.optimistic_index = self.optimistic_index.wrapping_add(1); + } + } else { + self.rounds_since_optimistic_rotation = 0; + self.optimistic_index = 0; + } + + decision + } +} + +#[derive(Clone, Debug, Eq, PartialEq)] +pub(crate) struct ChokingDecision { + pub(crate) unchoked: Vec, + pub(crate) optimistic: Option, +} + +pub(crate) fn select_unchoked_peers( + peers: &[PeerStats], torrent_state: TorrentState, upload_slots: usize, optimistic_index: usize, +) -> ChokingDecision { + let mut candidates: Vec<_> = peers + .iter() + .filter(|peer| peer.interested) + .copied() + .collect(); + candidates.sort_by(|left, right| { + rate_for(right, torrent_state) + .cmp(&rate_for(left, torrent_state)) + .then_with(|| left.id.id().cmp(right.id.id())) + }); + + if upload_slots == 0 { + return ChokingDecision { + unchoked: Vec::new(), + optimistic: None, + }; + } + + let reserve_optimistic_slot = candidates.len() > upload_slots; + let regular_slots = if reserve_optimistic_slot { + upload_slots.saturating_sub(1) + } else { + upload_slots + }; + + let mut unchoked: Vec<_> = candidates + .iter() + .take(regular_slots) + .map(|peer| peer.id) + .collect(); + + let optimistic = if reserve_optimistic_slot { + let optimistic_candidates = &candidates[regular_slots..]; + let peer = optimistic_candidates[optimistic_index % optimistic_candidates.len()].id; + unchoked.push(peer); + Some(peer) + } else { + None + }; + + ChokingDecision { + unchoked, + optimistic, + } +} + +fn rate_for(peer: &PeerStats, torrent_state: TorrentState) -> usize { + match torrent_state { + TorrentState::Downloading => peer.download_rate, + TorrentState::Seeding => peer.upload_rate, + TorrentState::Inactive => 0, + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn peer_id(value: u8) -> PeerId { + PeerId::from([value; 20]) + } + + fn stats(id: u8) -> PeerStats { + PeerStats { + id: peer_id(id), + interested: true, + choked: true, + download_rate: 0, + upload_rate: 0, + bytes_downloaded: 0, + bytes_uploaded: 0, + } + } + + fn with_rates(id: u8, download_rate: usize, upload_rate: usize) -> PeerStats { + PeerStats { + download_rate, + upload_rate, + ..stats(id) + } + } + + #[test] + fn selector_only_includes_interested_peers() { + let mut not_interested = stats(2); + not_interested.interested = false; + let peers = [stats(1), not_interested, stats(3)]; + + let decision = + select_unchoked_peers(&peers, TorrentState::Downloading, DEFAULT_UPLOAD_SLOTS, 0); + + assert_eq!(decision.unchoked, vec![peer_id(1), peer_id(3)]); + assert_eq!(decision.optimistic, None); + } + + #[test] + fn selector_respects_upload_slot_limit() { + let peers = [stats(1), stats(2), stats(3), stats(4), stats(5)]; + + let decision = + select_unchoked_peers(&peers, TorrentState::Downloading, DEFAULT_UPLOAD_SLOTS, 0); + + assert_eq!(decision.unchoked.len(), DEFAULT_UPLOAD_SLOTS); + assert_eq!(decision.optimistic, Some(peer_id(4))); + assert_eq!( + decision.unchoked, + vec![peer_id(1), peer_id(2), peer_id(3), peer_id(4)] + ); + } + + #[test] + fn selector_chokes_everyone_without_upload_slots() { + let peers = [stats(1), stats(2), stats(3)]; + + let decision = select_unchoked_peers(&peers, TorrentState::Downloading, 0, 0); + + assert!(decision.unchoked.is_empty()); + assert_eq!(decision.optimistic, None); + } + + #[test] + fn selector_ranks_downloading_peers_by_download_rate() { + let peers = [ + with_rates(1, 30, 500), + with_rates(2, 90, 10), + with_rates(3, 60, 1_000), + ]; + + let decision = select_unchoked_peers(&peers, TorrentState::Downloading, 2, 0); + + assert_eq!(decision.unchoked, vec![peer_id(2), peer_id(3)]); + assert_eq!(decision.optimistic, Some(peer_id(3))); + } + + #[test] + fn selector_ranks_seeding_peers_by_upload_rate() { + let peers = [ + with_rates(1, 100, 20), + with_rates(2, 10, 80), + with_rates(3, 60, 40), + ]; + + let decision = select_unchoked_peers(&peers, TorrentState::Seeding, 2, 0); + + assert_eq!(decision.unchoked, vec![peer_id(2), peer_id(3)]); + assert_eq!(decision.optimistic, Some(peer_id(3))); + } + + #[test] + fn selector_rotates_optimistic_unchoke() { + let peers = [ + with_rates(1, 100, 0), + with_rates(2, 90, 0), + with_rates(3, 80, 0), + with_rates(4, 70, 0), + ]; + + let first = select_unchoked_peers(&peers, TorrentState::Downloading, 2, 0); + let second = select_unchoked_peers(&peers, TorrentState::Downloading, 2, 1); + + assert_eq!(first.unchoked, vec![peer_id(1), peer_id(2)]); + assert_eq!(first.optimistic, Some(peer_id(2))); + assert_eq!(second.unchoked, vec![peer_id(1), peer_id(3)]); + assert_eq!(second.optimistic, Some(peer_id(3))); + } + + #[test] + fn scheduler_rotates_optimistic_unchoke_after_configured_rounds() { + let peers = [ + with_rates(1, 100, 0), + with_rates(2, 90, 0), + with_rates(3, 80, 0), + with_rates(4, 70, 0), + ]; + let mut scheduler = ChokingScheduler::new(2, 3); + + let first = scheduler.decide(&peers, TorrentState::Downloading); + let second = scheduler.decide(&peers, TorrentState::Downloading); + let third = scheduler.decide(&peers, TorrentState::Downloading); + let fourth = scheduler.decide(&peers, TorrentState::Downloading); + + assert_eq!(first.optimistic, Some(peer_id(2))); + assert_eq!(second.optimistic, Some(peer_id(2))); + assert_eq!(third.optimistic, Some(peer_id(2))); + assert_eq!(fourth.optimistic, Some(peer_id(3))); + } + + #[test] + fn scheduler_resets_optimistic_rotation_without_extra_candidates() { + let peers = [with_rates(1, 100, 0), with_rates(2, 90, 0)]; + let mut scheduler = ChokingScheduler::new(4, 3); + + for _ in 0..4 { + let decision = scheduler.decide(&peers, TorrentState::Downloading); + assert_eq!(decision.unchoked, vec![peer_id(1), peer_id(2)]); + assert_eq!(decision.optimistic, None); + } + + let crowded_peers = [ + with_rates(1, 100, 0), + with_rates(2, 90, 0), + with_rates(3, 80, 0), + with_rates(4, 70, 0), + with_rates(5, 60, 0), + ]; + let decision = scheduler.decide(&crowded_peers, TorrentState::Downloading); + + assert_eq!(decision.optimistic, Some(peer_id(4))); + } +} diff --git a/crates/libtortillas/src/torrent/choking_flow.rs b/crates/libtortillas/src/torrent/choking_flow.rs new file mode 100644 index 0000000..3b8373a --- /dev/null +++ b/crates/libtortillas/src/torrent/choking_flow.rs @@ -0,0 +1,87 @@ +use std::{collections::HashSet, time::Duration}; + +use futures::{StreamExt, stream}; +use kameo::actor::ActorRef; +use tokio::time::timeout; +use tracing::{trace, warn}; + +use super::TorrentActor; +use crate::{ + peer::{ + PeerActor, PeerStats, + commands::{SetChoked, Stats}, + }, + torrent::TorrentState, +}; + +const PEER_STATS_CONCURRENCY: usize = 32; +const PEER_STATS_TIMEOUT: Duration = Duration::from_millis(250); + +impl TorrentActor { + pub(super) async fn rechoke_peers(&mut self) { + if !matches!( + self.state, + TorrentState::Downloading | TorrentState::Seeding + ) { + trace!(state = ?self.state, "Skipping rechoke while torrent is inactive"); + return; + } + + let peer_stats = self.peer_stats().await; + let decision = self.choking_scheduler.decide(&peer_stats, self.state); + let unchoked: HashSet<_> = decision.unchoked.iter().copied().collect(); + + trace!( + unchoked = decision.unchoked.len(), + optimistic = ?decision.optimistic, + "Applying choking decision" + ); + + for stats in peer_stats { + let choked = !unchoked.contains(&stats.id); + if stats.choked == choked { + continue; + } + + let Some(actor) = self.peers.get(&stats.id) else { + continue; + }; + + if let Err(err) = actor.tell(SetChoked { choked }).await { + warn!(?err, peer_id = %stats.id, choked, "Failed to update peer choke state"); + } + } + } + + async fn peer_stats(&self) -> Vec { + let actor_refs: Vec<(crate::peer::PeerId, ActorRef)> = self + .peers + .iter() + .filter(|(_, actor)| actor.is_alive()) + .map(|(peer_id, actor)| (*peer_id, actor.clone())) + .collect(); + + stream::iter(actor_refs) + .map(|(peer_id, actor)| async move { + match timeout(PEER_STATS_TIMEOUT, actor.ask(Stats)).await { + Ok(Ok(Some(stats))) => Some(stats), + Ok(Ok(None)) => { + trace!(%peer_id, "Peer stats unavailable"); + None + } + Ok(Err(err)) => { + warn!(?err, %peer_id, "Failed to collect peer stats"); + None + } + Err(_) => { + trace!(%peer_id, "Timed out collecting peer stats"); + None + } + } + }) + .buffer_unordered(PEER_STATS_CONCURRENCY) + .filter_map(std::future::ready) + .collect() + .await + } +} diff --git a/crates/libtortillas/src/torrent/messages.rs b/crates/libtortillas/src/torrent/messages.rs index ce67421..de8d1f1 100644 --- a/crates/libtortillas/src/torrent/messages.rs +++ b/crates/libtortillas/src/torrent/messages.rs @@ -235,6 +235,12 @@ pub(crate) mod commands { } } + #[message(derive(Debug, Clone, Copy))] + pub(crate) async fn rechoke(&mut self) { + self.rechoke_peers().await; + self.schedule_next_rechoke().await; + } + /// A hook that is called when the torrent is ready to start downloading. /// This is used to implement /// [`Torrent::poll_ready`](crate::torrent::Torrent::poll_ready). diff --git a/crates/libtortillas/src/torrent/mod.rs b/crates/libtortillas/src/torrent/mod.rs index 0840ea2..ca3bb3c 100644 --- a/crates/libtortillas/src/torrent/mod.rs +++ b/crates/libtortillas/src/torrent/mod.rs @@ -1,5 +1,7 @@ mod actor; mod block; +mod choking; +mod choking_flow; mod export; mod handle; mod messages; diff --git a/crates/libtortillas/src/torrent/piece_flow.rs b/crates/libtortillas/src/torrent/piece_flow.rs index cba8bd8..b760317 100644 --- a/crates/libtortillas/src/torrent/piece_flow.rs +++ b/crates/libtortillas/src/torrent/piece_flow.rs @@ -232,6 +232,8 @@ impl TorrentActor { .await; self.broadcast_to_trackers(Announce).await; info!("Torrenting process completed, switching to seeding mode"); + self.rechoke_peers().await; + self.schedule_next_rechoke().await; } }