diff --git a/src/download_system/download.rs b/src/download_system/download.rs index 69ba5df..d401368 100644 --- a/src/download_system/download.rs +++ b/src/download_system/download.rs @@ -8,6 +8,7 @@ use file_owner::PathExt; use futures::StreamExt; use log::{error, info, warn}; use nix::unistd::Uid; +use std::sync::atomic::{AtomicU64, Ordering}; use std::time::Duration; use std::{fs, path::Path}; @@ -69,10 +70,16 @@ async fn download_target(app_data: &Data, target: &DownloadTarget) -> R } } TargetType::File => { + // Shared per-transfer counter of bytes pulled to local disk, so the + // *arr can show second-stage progress (issue #5). + let counter = app_data + .state + .local_byte_counter(&target.transfer_hash) + .await; // Delete file if already exists if !Path::new(&target.to).exists() { info!("{}: download {}", &target, "started".yellow()); - match fetch(target, app_data.config.uid, &app_data.http).await { + match fetch(target, app_data.config.uid, &app_data.http, &counter).await { Ok(_) => info!("{}: download {}", &target, "succeeded".green()), Err(e) => { error!("{}: download {}: {}", &target, "failed".red(), e); @@ -80,6 +87,12 @@ async fn download_target(app_data: &Data, target: &DownloadTarget) -> R } }; } else { + // Already fully downloaded (e.g. a resumed run after the files + // were pulled but not yet imported): still count its bytes so the + // aggregate progress across the transfer's files stays accurate. + if let Ok(m) = fs::metadata(&target.to) { + counter.fetch_add(m.len(), Ordering::Relaxed); + } info!("{}: already exists", &target); } } @@ -87,9 +100,21 @@ async fn download_target(app_data: &Data, target: &DownloadTarget) -> R Ok(()) } -async fn fetch(target: &DownloadTarget, uid: u32, client: &reqwest::Client) -> Result<()> { +async fn fetch( + target: &DownloadTarget, + uid: u32, + client: &reqwest::Client, + counter: &AtomicU64, +) -> Result<()> { let tmp_path = format!("{}.downloading", &target.to); + // Bytes of *this file* currently reflected in the shared transfer counter. + // Tracked across attempts so a resume can account for bytes already on disk + // exactly once, and a from-scratch restart (a server that ignores our Range + // and returns 200) can undo this file's contribution before re-counting the + // re-downloaded bytes (issue #5). + let mut counted: u64 = 0; + // Make sure the destination directory exists. A File target can be processed // before its parent Directory target, and external cleanup may have removed // an emptied folder, so create it here rather than failing with "No such @@ -106,7 +131,7 @@ async fn fetch(target: &DownloadTarget, uid: u32, client: &reqwest::Client) -> R let mut attempt = 0; loop { attempt += 1; - match fetch_attempt(target, &tmp_path, client).await { + match fetch_attempt(target, &tmp_path, client, counter, &mut counted).await { Ok(()) => break, Err(e) if attempt < MAX_ATTEMPTS => { warn!("{}: download attempt {} failed ({}), resuming", target, attempt, e); @@ -133,6 +158,8 @@ async fn fetch_attempt( target: &DownloadTarget, tmp_path: &str, client: &reqwest::Client, + counter: &AtomicU64, + counted: &mut u64, ) -> Result<()> { let existing = tokio::fs::metadata(tmp_path) .await @@ -167,8 +194,20 @@ async fn fetch_attempt( // otherwise it returned the whole file (200), so start it over. let resumed = status == reqwest::StatusCode::PARTIAL_CONTENT && existing > 0; let mut tmp_file = if resumed { + // Count the bytes already on disk exactly once (they aren't re-streamed). + if *counted < existing { + counter.fetch_add(existing - *counted, Ordering::Relaxed); + *counted = existing; + } tokio::fs::OpenOptions::new().append(true).open(tmp_path).await? } else { + // Starting over from scratch: the file is about to be truncated, so drop + // whatever this file previously added to the counter before re-counting + // the re-downloaded bytes below. + if *counted > 0 { + counter.fetch_sub(*counted, Ordering::Relaxed); + *counted = 0; + } tokio::fs::File::create(tmp_path).await? }; @@ -176,7 +215,12 @@ async fn fetch_attempt( loop { match tokio::time::timeout(STREAM_IDLE_TIMEOUT, byte_stream.next()).await { Ok(Some(item)) => { - tokio::io::copy(&mut item?.as_ref(), &mut tmp_file).await?; + let chunk = item?; + tokio::io::copy(&mut chunk.as_ref(), &mut tmp_file).await?; + // Lock-free running total for progress reporting (issue #5). + let n = chunk.len() as u64; + counter.fetch_add(n, Ordering::Relaxed); + *counted += n; } Ok(None) => break, Err(_) => bail!("stalled: no data received for {:?}", STREAM_IDLE_TIMEOUT), diff --git a/src/download_system/orchestration.rs b/src/download_system/orchestration.rs index 1968976..994b084 100644 --- a/src/download_system/orchestration.rs +++ b/src/download_system/orchestration.rs @@ -120,6 +120,12 @@ impl Worker { // The files now exist locally, so it's safe to report this transfer // as complete to the *arr (see issue #16). self.app_data.state.mark_local_complete(t.transfer_id).await; + // Local download finished; the progress counter is no longer read + // (torrent-get reports completion via local_complete now), so drop + // it to keep the map bounded (issue #5). + if let Some(hash) = &t.hash { + self.app_data.state.clear_local_bytes(hash).await; + } self.tx .send(TransferMessage::Downloaded(Transfer { targets: Some(targets), diff --git a/src/download_system/transfer.rs b/src/download_system/transfer.rs index c27dbc3..9cf5356 100644 --- a/src/download_system/transfer.rs +++ b/src/download_system/transfer.rs @@ -408,10 +408,13 @@ pub async fn produce_transfers(app_data: Data, tx: Sender, tx: Sender, t: &PutIOTransfer) -> u64 { + if app_data.state.is_local_complete(t.id).await { + return 100; + } + let size = t.size.unwrap_or(0).max(0) as u64; + if size == 0 { + return 0; + } + let putio_done = t.finished_at.is_some() + || matches!( + t.status.to_uppercase().as_str(), + "SEEDING" | "COMPLETED" | "STOPPED" + ); + if putio_done { + let local = match &t.hash { + Some(hash) => app_data + .state + .local_bytes_downloaded(hash) + .await + .unwrap_or(0) + .min(size), + None => 0, + }; + // Cap at 99 until local_complete flips (handled by the early return + // above): the counter can reach `size` a moment before the transfer is + // marked complete, and logging 100% then would misreport a still-active + // pull as finished. + (50 + 50 * local / size).min(99) + } else { + let downloaded = (t.downloaded.unwrap_or(0).max(0) as u64).min(size); + 50 * downloaded / size + } +} + /// Heuristic: does this name look like a TV episode (SxxExx, or "Season")? /// Used to route an orphaned file to the Sonarr vs Radarr category folder. fn looks_like_episode(name: &str) -> bool { diff --git a/src/http/handlers.rs b/src/http/handlers.rs index 66f4b00..8d7a2b8 100644 --- a/src/http/handlers.rs +++ b/src/http/handlers.rs @@ -195,10 +195,16 @@ pub(crate) async fn handle_torrent_get( api_token: &str, app_data: &web::Data, ) -> Option { + // Whether the account listing actually succeeded. On failure `transfers` is + // empty, which must not be mistaken for "nothing is on the account" when + // pruning per-transfer progress counters below (issue #5) — that would wipe + // progress for downloads still in flight. + let mut listing_ok = true; let transfers = match putio::list_transfers(api_token).await { Ok(r) => r.transfers, Err(e) => { error!("Failed to list put.io transfers: {}", e); + listing_ok = false; Vec::new() } }; @@ -209,6 +215,12 @@ pub(crate) async fn handle_torrent_get( let active_file_ids: HashSet = transfers.iter().filter_map(|t| t.file_id).collect(); app_data.state.retain_file_names(&active_file_ids).await; + // Track the hashes of everything still present (put.io transfers plus the + // orphans reported below) so the local-progress counters can be pruned to + // just those at the end, keeping that map bounded (issue #5). + let mut active_hashes: HashSet = + transfers.iter().filter_map(|t| t.hash.clone()).collect(); + // Allocate the token once and share it cheaply (refcount bump) with each // per-transfer task, rather than allocating a new String per transfer. let api_token: Arc = Arc::from(api_token); @@ -253,20 +265,54 @@ pub(crate) async fn handle_torrent_get( tt.name = name; } } - // put.io marks a transfer complete as soon as *its own* (cloud) - // download finishes, but the files don't exist on local disk until - // putioarr has pulled them down. Reporting completion to the *arr - // too early makes it try to import missing files ("No files found - // eligible for import"). Keep the torrent in a downloading state - // until putioarr has actually finished the local download (#16). + // A transfer has two download stages the *arr can't see separately: + // put.io's own (cloud) download, then putioarr pulling the files to + // local disk. put.io reports 100% as soon as the cloud download + // finishes, but the files don't exist locally until the pull is done + // — reporting completion then makes the *arr try to import missing + // files ("No files found eligible for import", #16). Instead, map the + // cloud download to 0-50% and the local pull to 50-100% so the *arr + // shows a single bar that keeps moving through both stages and only + // reaches 100% once the files are actually on disk (issue #5). Both + // stages move the same number of bytes, so the 50/50 split is + // byte-accurate. let putio_done = tt.is_finished || matches!( tt.status, TransmissionTorrentStatus::Seeding | TransmissionTorrentStatus::Stopped ); - if putio_done && !app_data.state.is_local_complete(t.id).await { + let size = tt.total_size.max(0); + if !app_data.state.is_local_complete(t.id).await { + if size > 0 { + let done = if putio_done { + // Cloud finished (first 50%); report the local pull so + // far as the second 50%. + let local = match &t.hash { + Some(hash) => app_data + .state + .local_bytes_downloaded(hash) + .await + .unwrap_or(0) + .min(size as u64) as i64, + None => 0, + }; + size / 2 + local / 2 + } else { + // Still on put.io; scale its progress into the first 50%. + tt.downloaded_ever.clamp(0, size) / 2 + }; + tt.downloaded_ever = done; + // Keep >= 1 so 0/0 or a near-complete pull isn't read as + // "done" before local_complete flips. + tt.left_until_done = std::cmp::max(size - done, 1); + } else { + // Size unknown (put.io omitted it): we can't show a + // percentage, but the local pull still isn't done, so leave a + // non-zero amount remaining rather than let 0/0 read as + // complete and trigger an early import (#16). + tt.left_until_done = std::cmp::max(tt.left_until_done, 1); + } tt.is_finished = false; - tt.left_until_done = std::cmp::max(tt.total_size, 1); tt.status = TransmissionTorrentStatus::Downloading; } tt @@ -285,18 +331,27 @@ pub(crate) async fn handle_torrent_get( Ok(id) => id, Err(_) => continue, }; + active_hashes.insert(orphan.hash.clone()); let complete = app_data.state.is_local_complete(id).await; - // Report consistent size/progress. Keep left_until_done <= total_size, + // An orphan's file already exists on put.io (100% cloud), so its only + // stage is the local pull — report that directly as 0-100% from the + // bytes downloaded so far (issue #5). Keep left_until_done <= total_size, // and when incomplete report a non-zero amount remaining even if the // size is unknown (put.io omitted it) so a client can't read 0/0 as // "done" while it's still downloading. let size = orphan.size.max(0); - let (total_size, left_until_done) = if complete { - (size, 0) + let local = app_data + .state + .local_bytes_downloaded(&orphan.hash) + .await + .unwrap_or(0) + .min(size.max(0) as u64) as i64; + let (total_size, left_until_done, downloaded_ever) = if complete { + (size, 0, size) } else if size > 0 { - (size, size) + (size, std::cmp::max(size - local, 1), local) } else { - (1, 1) + (1, 1, 0) }; transmission_transfers.push(TransmissionTorrent { id, @@ -314,7 +369,7 @@ pub(crate) async fn handle_torrent_get( }, seconds_downloading: 0, error_string: None, - downloaded_ever: if complete { size } else { 0 }, + downloaded_ever, seed_ratio_limit: 0.0, seed_ratio_mode: 0, seed_idle_limit: 0, @@ -323,6 +378,13 @@ pub(crate) async fn handle_torrent_get( }); } + // Prune progress counters for transfers that are no longer present — but + // only when the account listing succeeded, so a transient list failure + // (empty `active_hashes`) can't wipe progress for in-flight downloads (#5). + if listing_ok { + app_data.state.retain_local_bytes(&active_hashes).await; + } + let torrents = json!(transmission_transfers); let mut arguments = serde_json::Map::new(); diff --git a/src/state.rs b/src/state.rs index eb05a82..136ede5 100644 --- a/src/state.rs +++ b/src/state.rs @@ -3,6 +3,7 @@ use anyhow::Result; use log::{debug, error, info, warn}; use serde::{Deserialize, Serialize}; use std::collections::{HashMap, HashSet}; +use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::Arc; use std::time::{Duration, Instant}; use tokio::sync::RwLock; @@ -62,6 +63,14 @@ pub struct StateManager { /// the log. A misconfigured Sonarr/Radarr fails on every poll for every /// transfer, and logging each one filled users' disks over time (issue #21). arr_error_logged: Arc>>, + /// Bytes pulled to local disk so far, keyed by transfer hash. put.io's own + /// percentage only covers the cloud download; this tracks the second stage + /// (put.io -> local disk) so the *arr can show real progress instead of a + /// bar frozen at 0% while the files are being pulled (issue #5). Written + /// lock-free per chunk by the download workers via an AtomicU64, and read + /// only when the *arr polls or the status line is logged, so it adds no + /// measurable cost to the download hot path. + local_bytes: Arc>>>, } impl StateManager { @@ -74,9 +83,52 @@ impl StateManager { failed_names: Arc::new(RwLock::new(HashMap::new())), orphans: Arc::new(RwLock::new(HashMap::new())), arr_error_logged: Arc::new(RwLock::new(HashMap::new())), + local_bytes: Arc::new(RwLock::new(HashMap::new())), } } + /// Returns the shared byte counter for a transfer's local download, creating + /// it on first use. Download workers hold this and add to it per chunk, so + /// progress accumulates lock-free across the (possibly several) files of a + /// transfer (issue #5). + pub async fn local_byte_counter(&self, hash: &str) -> Arc { + // Fast path: the counter almost always already exists (a transfer's + // several files all resolve the same one), so take only a read lock and + // avoid contending the write lock across download workers. + if let Some(counter) = self.local_bytes.read().await.get(hash) { + return counter.clone(); + } + self.local_bytes + .write() + .await + .entry(hash.to_string()) + .or_insert_with(|| Arc::new(AtomicU64::new(0))) + .clone() + } + + /// Bytes pulled to local disk so far for a transfer, or None if it isn't + /// being downloaded (no counter registered yet). + pub async fn local_bytes_downloaded(&self, hash: &str) -> Option { + self.local_bytes + .read() + .await + .get(hash) + .map(|c| c.load(Ordering::Relaxed)) + } + + /// Forgets a transfer's local byte counter once it's done or removed, so the + /// map doesn't grow without bound over the process lifetime. + pub async fn clear_local_bytes(&self, hash: &str) { + self.local_bytes.write().await.remove(hash); + } + + /// Drops local byte counters for transfers no longer present, keeping the + /// map bounded regardless of which cleanup path a transfer took (mirrors + /// [`Self::retain_file_names`]). + pub async fn retain_local_bytes(&self, keep: &HashSet) { + self.local_bytes.write().await.retain(|h, _| keep.contains(h)); + } + /// Minimum time between logging the same *arr's connection error. pub const ARR_ERROR_LOG_INTERVAL: Duration = Duration::from_secs(300);