//! Torrent enclosures via librqbit. Replaces iPXDownloader.getTorrent and the vendored //! BitTorrent 4.2.1 + khashmir tree. use anyhow::{Context, Result, bail}; use librqbit::{AddTorrent, AddTorrentOptions, AddTorrentResponse, Session, SessionOptions}; use std::net::SocketAddr; use std::path::{Path, PathBuf}; use std::sync::Arc; use std::time::{Duration, Instant}; use crate::config::Config; pub struct Torrents { session: Arc, } /// How long a torrent may make no progress before we give up on it. The original called /// this torrentMaxBeatTime and counted display ticks; a wall clock is easier to reason about. fn stall_limit(cfg: &Config) -> Duration { Duration::from_secs(cfg.torrent.stall_mins.max(1) * 60) } impl Torrents { pub async fn new(cfg: &Config) -> Result { let (lo, _hi) = cfg.torrent.ports(); let listen_addr: SocketAddr = format!("0.0.0.0:{lo}").parse()?; let opts = SessionOptions { listen: Some(librqbit::ListenerOptions { listen_addr, ..Default::default() }), client_name_and_version: Some(concat!("ipx ", env!("CARGO_PKG_VERSION")).into()), ..Default::default() }; // The session root is only a fallback; every torrent names its own output folder. let session = Session::new_with_opts(cfg.general.download_dir.clone(), opts) .await .context("starting the torrent session")?; Ok(Self { session }) } /// Downloads a torrent (a .torrent URL or a magnet link) straight into `dest_dir`, /// seeds it up to the configured limit, then stops. /// /// Downloading in place rather than into a staging dir is deliberate: seeding serves /// the files it downloaded, so moving them first would break it. pub async fn fetch( &self, cfg: &Config, url: &str, dest_dir: &Path, mut on_progress: impl FnMut(u64, u64), ) -> Result<(PathBuf, u64)> { tokio::fs::create_dir_all(dest_dir).await?; let limit = stall_limit(cfg); // Resolving a magnet's metadata happens inside add_torrent, and against a dead // swarm it never returns. The stall budget has to cover this phase too, or a // torrent nobody is seeding wedges the scan forever. let added = tokio::time::timeout( limit, self.session.add_torrent( AddTorrent::from_url(url), Some(AddTorrentOptions { output_folder: Some(dest_dir.to_string_lossy().into_owned()), overwrite: true, ..Default::default() }), ), ) .await .map_err(|_| { anyhow::anyhow!( "no metadata after {} minutes, gave up", limit.as_secs() / 60 ) })? .context("adding the torrent")?; let (id, handle) = match added { AddTorrentResponse::Added(id, h) | AddTorrentResponse::AlreadyManaged(id, h) => (id, h), AddTorrentResponse::ListOnly(_) => bail!("torrent added in list-only mode"), }; let mut last_progress = 0u64; let mut last_move = Instant::now(); loop { tokio::time::sleep(Duration::from_secs(1)).await; let stats = handle.stats(); if let Some(err) = &stats.error { let _ = self.session.delete(id.into(), false).await; bail!("torrent failed: {err}"); } on_progress(stats.progress_bytes, stats.total_bytes); if stats.finished { break; } if stats.progress_bytes > last_progress { last_progress = stats.progress_bytes; last_move = Instant::now(); } else if last_move.elapsed() > limit { // Leave the partial files behind; the row records the failure. let _ = self.session.delete(id.into(), false).await; bail!("no progress for {} minutes, gave up", limit.as_secs() / 60); } } let name = handle.name().unwrap_or_else(|| "torrent".to_owned()); let path = handle.output_folder().join(&name); let size = handle.stats().total_bytes; self.seed(cfg, &handle).await; self.session .delete(id.into(), false) .await .context("releasing the torrent")?; Ok((path, size)) } /// Seeds until the ratio or the time limit is reached, whichever comes first. async fn seed(&self, cfg: &Config, handle: &Arc) { if cfg.torrent.seed_ratio <= 0.0 || cfg.torrent.seed_time_mins == 0 { return; } let deadline = Instant::now() + Duration::from_secs(cfg.torrent.seed_time_mins * 60); loop { let stats = handle.stats(); if ratio(stats.uploaded_bytes, stats.total_bytes) >= cfg.torrent.seed_ratio { tracing::info!(ratio = cfg.torrent.seed_ratio, "seed ratio reached"); return; } if Instant::now() >= deadline { tracing::info!(mins = cfg.torrent.seed_time_mins, "seed time reached"); return; } tokio::time::sleep(Duration::from_secs(5)).await; } } } /// Uploaded over total. A zero-byte torrent counts as fully seeded rather than dividing by zero. pub fn ratio(uploaded: u64, total: u64) -> f64 { if total == 0 { return f64::INFINITY; } uploaded as f64 / total as f64 } #[cfg(test)] mod tests { use super::*; #[test] fn ratio_handles_the_empty_torrent() { assert_eq!(ratio(0, 100), 0.0); assert_eq!(ratio(50, 100), 0.5); assert_eq!(ratio(200, 100), 2.0); assert!(ratio(0, 0).is_infinite(), "must not divide by zero or seed forever"); } #[test] fn stall_limit_is_never_zero() { let mut cfg = Config::default(); cfg.torrent.stall_mins = 0; assert_eq!(stall_limit(&cfg), Duration::from_secs(60), "0 would abort instantly"); cfg.torrent.stall_mins = 30; assert_eq!(stall_limit(&cfg), Duration::from_secs(1800)); } }