fix: repair native range transfer progress

AnInsomniacy · AnInsomniacy · commit e1b342f61b0a · 2026-05-25T19:48:26.000+10:00
diff --git a/crates/raria-cli/src/daemon.rs b/crates/raria-cli/src/daemon.rs
@@ -251,6 +251,7 @@ pub(crate) async fn run_daemon_with_config(
                     let range_context = RangeExecutionContext {
                         task_id: task_id.clone(),
                     };
+                    let engine_for_failure = Arc::clone(&engine_ref);
                     tokio::spawn(async move {
                         if let Err(e) = run_job_download(
                             engine_ref,
@@ -261,6 +262,10 @@ pub(crate) async fn run_daemon_with_config(
                         .await
                         {
                             error!(%task_id, error = %e, "range download task failed");
+                            let _ = engine_for_failure.fail_native_task(
+                                &task_id,
+                                &classified_error_message(&e.to_string()),
+                            );
                         }
                     });
                 }
diff --git a/crates/raria-cli/tests/native_api_smoke.rs b/crates/raria-cli/tests/native_api_smoke.rs
@@ -985,6 +985,104 @@ async fn daemon_native_task_reports_effective_active_connections() {
     }
 }
 
+#[tokio::test]
+async fn daemon_native_http_range_task_progresses_with_bounded_segments() {
+    let payload = Arc::new(
+        (0..(512 * 1024))
+            .map(|i| (i % 251) as u8)
+            .collect::<Vec<_>>(),
+    );
+    let server = MockServer::start().await;
+
+    Mock::given(method("HEAD"))
+        .and(path("/bounded-range.bin"))
+        .respond_with(RangeResponder::new(Arc::clone(&payload)))
+        .mount(&server)
+        .await;
+    Mock::given(method("GET"))
+        .and(path("/bounded-range.bin"))
+        .respond_with(RangeResponder::new(Arc::clone(&payload)))
+        .mount(&server)
+        .await;
+
+    let temp = tempdir().expect("tempdir");
+    let session_file = temp.path().join("native-bounded-range.session.redb");
+    let port = allocate_port();
+    let mut child = spawn_native_daemon(temp.path(), &session_file, port);
+    wait_for_native_api_ready(port, &mut child)
+        .await
+        .expect("native API ready");
+
+    let client = reqwest::Client::new();
+    let download_dir = temp.path().join("nested").join("downloads");
+    let created: serde_json::Value = client
+        .post(format!("http://127.0.0.1:{port}/api/v1/tasks"))
+        .json(&serde_json::json!({
+            "sources": [format!("{}/bounded-range.bin", server.uri())],
+            "downloadDir": download_dir,
+            "filename": "bounded-range.bin",
+            "segments": 4
+        }))
+        .send()
+        .await
+        .expect("create bounded range task")
+        .json()
+        .await
+        .expect("bounded range task json");
+    let task_id = created["taskId"].as_str().expect("task id");
+
+    let deadline = Instant::now() + Duration::from_secs(30);
+    let completed = loop {
+        let task: serde_json::Value = client
+            .get(format!("http://127.0.0.1:{port}/api/v1/tasks/{task_id}"))
+            .send()
+            .await
+            .expect("bounded range task detail request")
+            .json()
+            .await
+            .expect("bounded range task detail json");
+        if task["completedBytes"].as_u64().unwrap_or(0) > 0 {
+            assert!(
+                task["downloadBytesPerSecond"].as_u64().unwrap_or(0) > 0,
+                "bounded range task should expose a non-zero speed after progress: {task}"
+            );
+        }
+        if task["lifecycle"] == "completed" {
+            break task;
+        }
+
+        assert!(
+            Instant::now() < deadline,
+            "bounded range task never completed: {task}"
+        );
+        tokio::time::sleep(Duration::from_millis(50)).await;
+    };
+    assert_eq!(
+        completed["completedBytes"].as_u64(),
+        Some(payload.len() as u64)
+    );
+    assert_eq!(
+        std::fs::read(temp.path().join("nested/downloads/bounded-range.bin"))
+            .expect("read bounded range output"),
+        payload.as_ref().clone()
+    );
+
+    let requests = server.received_requests().await.expect("received requests");
+    let bounded_range_seen = requests.iter().any(|request| {
+        request.method.as_str() == "GET"
+            && request.url.path() == "/bounded-range.bin"
+            && request
+                .headers
+                .get("range")
+                .and_then(|value| value.to_str().ok())
+                .is_some_and(|range| range.starts_with("bytes=") && !range.ends_with('-'))
+    });
+    assert!(
+        bounded_range_seen,
+        "native HTTP range task should issue bounded Range requests"
+    );
+}
+
 #[tokio::test]
 async fn daemon_native_task_stops_after_file_not_found_budget() {
     let server = MockServer::start().await;
diff --git a/crates/raria-cli/tests/single_download.rs b/crates/raria-cli/tests/single_download.rs
@@ -1592,7 +1592,7 @@ async fn single_download_continue_resumes_from_existing_file_length() {
 
     Mock::given(method("GET"))
         .and(path("/continue.bin"))
-        .and(header("range", "bytes=4-"))
+        .and(header("range", "bytes=4-7"))
         .respond_with(ResponseTemplate::new(206).set_body_bytes(b"5678"))
         .mount(&server)
         .await;
diff --git a/crates/raria-core/src/engine.rs b/crates/raria-core/src/engine.rs
@@ -33,7 +33,7 @@ use std::collections::{HashMap, HashSet};
 use std::path::PathBuf;
 use std::sync::Arc;
 use std::sync::atomic::{AtomicU64, Ordering};
-use std::time::Instant;
+use std::time::{Duration, Instant};
 use tokio::sync::Notify;
 use tokio_util::sync::CancellationToken;
 use tracing::{debug, error, info, warn};
@@ -42,6 +42,11 @@ fn native_task_log_fields(task_id: &TaskId) -> [(&'static str, String); 1] {
     [("task_id", task_id.to_string())]
 }
 
+fn speed_from_delta(bytes: u64, elapsed: Duration) -> u64 {
+    let nanos = elapsed.as_nanos().max(1);
+    ((bytes as u128).saturating_mul(1_000_000_000) / nanos) as u64
+}
+
 /// Input for planning a native segmented range download.
 #[derive(Debug, Clone, Copy)]
 pub struct NativeSegmentPlanningInput<'a> {
@@ -155,6 +160,8 @@ pub struct Engine {
     global_upload_limit: AtomicU64,
     /// Per-job limiter handles layered on top of the global limiter.
     job_rate_limiters: Mutex<HashMap<Gid, Arc<SharedRateLimiter>>>,
+    /// Per-job progress sampling state used to project native range speed.
+    job_speed_samples: Mutex<HashMap<Gid, SpeedSample>>,
     /// Unique session identifier (random hex, persisted for lifetime of process).
     pub session_id: String,
     store: Option<Arc<Store>>,
@@ -164,6 +171,12 @@ pub struct Engine {
     started_at: Instant,
 }
 
+#[derive(Debug, Clone, Copy)]
+struct SpeedSample {
+    downloaded: u64,
+    sampled_at: Instant,
+}
+
 impl Engine {
     /// Create a new Engine with the given configuration (no persistence).
     pub fn new(config: GlobalConfig) -> Self {
@@ -180,6 +193,7 @@ impl Engine {
             global_rate_limiter,
             global_upload_limit: AtomicU64::new(global_upload_limit),
             job_rate_limiters: Mutex::new(HashMap::new()),
+            job_speed_samples: Mutex::new(HashMap::new()),
             session_id: format!("{:016x}", rand::random::<u64>()),
             store: None,
             shutdown: CancellationToken::new(),
@@ -203,6 +217,7 @@ impl Engine {
             global_rate_limiter,
             global_upload_limit: AtomicU64::new(global_upload_limit),
             job_rate_limiters: Mutex::new(HashMap::new()),
+            job_speed_samples: Mutex::new(HashMap::new()),
             session_id: format!("{:016x}", rand::random::<u64>()),
             store: Some(store),
             shutdown: CancellationToken::new(),
@@ -778,6 +793,7 @@ impl Engine {
         self.cancel_registry.cancel(task_id);
         self.scheduler.dequeue_task(task_id);
         self.clear_job_rate_limiter(gid);
+        self.clear_job_speed_sample(gid);
         self.registry
             .update(gid, |job| {
                 job.status = Status::Removed;
@@ -805,6 +821,7 @@ impl Engine {
         let gid = self
             .gid_for_task_id(task_id)
             .context("native task not found")?;
+        self.clear_job_speed_sample(gid);
         self.registry
             .update(gid, |job| {
                 job.status = Status::Waiting;
@@ -1065,6 +1082,7 @@ impl Engine {
             [("gid", gid.to_string())],
         );
         self.clear_job_rate_limiter(gid);
+        self.clear_job_speed_sample(gid);
         Ok(())
     }
 
@@ -1155,6 +1173,7 @@ impl Engine {
             [("gid", gid.to_string())],
         );
         self.clear_job_rate_limiter(gid);
+        self.clear_job_speed_sample(gid);
         self.work_notify.notify_one();
         Ok(())
     }
@@ -1196,6 +1215,7 @@ impl Engine {
             [("gid", gid.to_string()), ("error", error_msg.to_string())],
         );
         self.clear_job_rate_limiter(gid);
+        self.clear_job_speed_sample(gid);
         self.work_notify.notify_one();
         Ok(())
     }
@@ -1374,10 +1394,45 @@ impl Engine {
         self.job_rate_limiters.lock().remove(&gid);
     }
 
+    fn clear_job_speed_sample(&self, gid: Gid) {
+        self.job_speed_samples.lock().remove(&gid);
+    }
+
     /// Update download progress for a job.
     pub fn update_progress(&self, gid: Gid, bytes: u64) {
         let _ = self.registry.update(gid, |job| {
-            job.downloaded += bytes;
+            job.downloaded = job.downloaded.saturating_add(bytes);
+            let downloaded = job.downloaded;
+            let now = Instant::now();
+            let mut samples = self.job_speed_samples.lock();
+            let speed = if let Some(sample) = samples.get(&gid).copied() {
+                let elapsed = now.duration_since(sample.sampled_at);
+                if !elapsed.is_zero() && downloaded >= sample.downloaded {
+                    let delta = downloaded - sample.downloaded;
+                    let bps = speed_from_delta(delta, elapsed);
+                    samples.insert(
+                        gid,
+                        SpeedSample {
+                            downloaded,
+                            sampled_at: now,
+                        },
+                    );
+                    bps
+                } else {
+                    job.download_speed
+                }
+            } else {
+                let speed = job.download_speed;
+                samples.insert(
+                    gid,
+                    SpeedSample {
+                        downloaded,
+                        sampled_at: now,
+                    },
+                );
+                speed
+            };
+            job.download_speed = speed;
         });
     }
 
@@ -1705,6 +1760,7 @@ impl Engine {
                 job.connections = 0;
             })
             .context("native task not found")?;
+        self.clear_job_speed_sample(gid);
         Ok(())
     }
 
@@ -1948,6 +2004,7 @@ impl Engine {
             self.scheduler.dequeue_task(&task_id);
         }
         self.clear_job_rate_limiter(gid);
+        self.clear_job_speed_sample(gid);
 
         // Force transition to Removed regardless of current state.
         self.registry
@@ -1975,6 +2032,7 @@ impl Engine {
             Status::Complete | Status::Error | Status::Removed => {
                 self.registry.remove(gid);
                 self.clear_job_rate_limiter(gid);
+                self.clear_job_speed_sample(gid);
                 debug!(%gid, "download result removed");
                 Ok(())
             }
@@ -1993,6 +2051,7 @@ impl Engine {
                 Status::Complete | Status::Error | Status::Removed => {
                     self.registry.remove(job.gid);
                     self.clear_job_rate_limiter(job.gid);
+                    self.clear_job_speed_sample(job.gid);
                     purged += 1;
                 }
                 _ => {}
@@ -3089,6 +3148,19 @@ mod tests {
         assert_eq!(job.downloaded, 3000);
     }
 
+    #[test]
+    fn update_progress_projects_speed_after_delta() {
+        let engine = Engine::new(default_config());
+        let handle = engine.add_uri(&default_spec()).unwrap();
+
+        engine.update_progress(handle.gid, 1000);
+        std::thread::sleep(std::time::Duration::from_millis(1));
+        engine.update_progress(handle.gid, 1000);
+
+        let job = engine.registry.get(handle.gid).unwrap();
+        assert!(job.download_speed > 0);
+    }
+
     #[test]
     fn completion_frees_slot() {
         let engine = Engine::new(GlobalConfig {
diff --git a/crates/raria-http/src/backend.rs b/crates/raria-http/src/backend.rs
@@ -399,8 +399,16 @@ impl ByteSourceBackend for HttpBackend {
                         request = request.header(name, value);
                     }
 
-                    if offset > 0 {
-                        request = request.header(RANGE, format!("bytes={offset}-"));
+                    if offset > 0 || ctx.length.is_some() {
+                        let range = match ctx.length {
+                            Some(0) => format!("bytes={offset}-{offset}"),
+                            Some(length) => {
+                                let end = offset.saturating_add(length).saturating_sub(1);
+                                format!("bytes={offset}-{end}")
+                            }
+                            None => format!("bytes={offset}-"),
+                        };
+                        request = request.header(RANGE, range);
                         if let Some(ref etag) = ctx.etag {
                             request = request.header("If-Range", etag.as_str());
                         }
diff --git a/crates/raria-range/src/backend.rs b/crates/raria-range/src/backend.rs
@@ -44,6 +44,9 @@ pub struct OpenContext {
     pub etag: Option<String>,
     /// Custom HTTP headers for stream requests.
     pub headers: Vec<(String, String)>,
+    /// Maximum number of bytes the backend should stream when the protocol
+    /// can express an upper bound. The executor still caps reads locally.
+    pub length: Option<u64>,
 }
 
 impl Default for OpenContext {
@@ -53,6 +56,7 @@ impl Default for OpenContext {
             timeout: Duration::from_secs(60),
             etag: None,
             headers: Vec::new(),
+            length: None,
         }
     }
 }
@@ -93,12 +97,12 @@ pub type ByteStream = Pin<Box<dyn AsyncRead + Send>>;
 /// Implementations exist for HTTP (`raria-http`), FTP (`raria-ftp`),
 /// and SFTP (`raria-sftp`).
 ///
-/// The key insight: `open_from(offset)` returns a forward-only stream
-/// with no upper bound. The caller (SegmentExecutor) is responsible for
-/// consuming only the bytes it needs and then dropping the stream.
+/// `open_from(offset)` returns a forward-only stream. Protocols that can
+/// express an upper bound should honor [`OpenContext::length`]. The caller
+/// still consumes only the bytes it needs and then drops the stream.
 ///
 /// This matches the natural semantics of:
-/// - HTTP: `Range: bytes=offset-`
+/// - HTTP: `Range: bytes=offset-end` when length is known
 /// - FTP: `REST offset` + `RETR`
 /// - SFTP: `read_from(offset)`
 #[async_trait]
@@ -109,8 +113,8 @@ pub trait ByteSourceBackend: Send + Sync + fmt::Debug {
     /// Open a byte stream starting from the given offset.
     ///
     /// The stream reads forward from `offset` until EOF or the caller
-    /// stops consuming. There is no upper-bound parameter because the
-    /// SegmentExecutor controls how many bytes to read.
+    /// stops consuming. When [`OpenContext::length`] is set, backends should
+    /// request that bounded span if the protocol supports it.
     async fn open_from(&self, uri: &Url, offset: u64, ctx: &OpenContext) -> Result<ByteStream>;
 
     /// Human-readable name for this backend (e.g., "http", "ftp", "sftp").
diff --git a/crates/raria-range/src/executor.rs b/crates/raria-range/src/executor.rs
diff --git a/docs/core-modernization/real-smoke-report.md b/docs/core-modernization/real-smoke-report.md

Original file line number	Diff line number	Diff line change
`@@ -251,6 +251,7 @@ pub(crate) async fn run_daemon_with_config(`
`251`	`251`	`let range_context = RangeExecutionContext {`
`252`	`252`	`task_id: task_id.clone(),`
`253`	`253`	`};`
	`254`	`+ let engine_for_failure = Arc::clone(&engine_ref);`
`254`	`255`	`tokio::spawn(async move {`
`255`	`256`	`if let Err(e) = run_job_download(`
`256`	`257`	`engine_ref,`
`@@ -261,6 +262,10 @@ pub(crate) async fn run_daemon_with_config(`
`261`	`262`	`.await`
`262`	`263`	`{`
`263`	`264`	`error!(%task_id, error = %e, "range download task failed");`
	`265`	`+ let _ = engine_for_failure.fail_native_task(`
	`266`	`+ &task_id,`
	`267`	`+ &classified_error_message(&e.to_string()),`
	`268`	`+ );`
`264`	`269`	`}`
`265`	`270`	`});`
`266`	`271`	`}`