feat: report two-stage download progress to the *arr and logs

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,27 +70,51 @@ async fn download_target(app_data: &Data<AppData>, 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);
                         bail!(e)
                     }
                 };
             } 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);
             }
         }
     }
     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,16 +194,33 @@ 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?
     };
 
     let mut byte_stream = response.bytes_stream();
     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),

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),

src/download_system/transfer.rs

@@ -408,10 +408,13 @@ pub async fn produce_transfers(app_data: Data<AppData>, tx: Sender<TransferMessa
                     "Active transfers: {}",
                     list_transfer_response.transfers.len()
                 );
-                list_transfer_response
-                    .transfers
-                    .iter()
-                    .for_each(|t| info!("  {}", Transfer::from(app_data.clone(), t)));
+                for t in &list_transfer_response.transfers {
+                    let transfer = Transfer::from(app_data.clone(), t);
+                    // Report combined two-stage progress (cloud 0-50%, local pull
+                    // 50-100%) so a stalled download is visible in the log (#5).
+                    let pct = combined_progress_percent(&app_data, t).await;
+                    info!("  {} ({}%)", transfer, pct);
+                }
 
                 start = std::time::Instant::now();
             }
@@ -424,6 +427,45 @@ pub async fn produce_transfers(app_data: Data<AppData>, tx: Sender<TransferMessa
     }
 }
 
+/// Combined two-stage download progress (0-100) for a transfer, mirroring what
+/// [`crate::http::handlers`] reports to the *arr: put.io's cloud download is the
+/// first 50%, pulling the files to local disk is the second 50% (issue #5). Used
+/// for the periodic status log so a stuck download is distinguishable from one
+/// that's still making progress.
+async fn combined_progress_percent(app_data: &Data<AppData>, 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 {

src/http/handlers.rs

@@ -195,10 +195,16 @@ pub(crate) async fn handle_torrent_get(
     api_token: &str,
     app_data: &web::Data<AppData>,
 ) -> Option<serde_json::Value> {
+    // 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<i64> = 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<String> =
+        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<str> = 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();