From 296098e9229ebf0bc44c6d1606b462162dd304b0 Mon Sep 17 00:00:00 2001 From: JustinLoye Date: Wed, 10 Jun 2026 12:50:28 +0900 Subject: [PATCH 1/7] resumable http to handle dropped connections --- src/client.rs | 10 +++++- src/lib.rs | 2 ++ src/resumable_http.rs | 81 +++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 92 insertions(+), 1 deletion(-) create mode 100644 src/resumable_http.rs diff --git a/src/client.rs b/src/client.rs index 67d5518..f438258 100644 --- a/src/client.rs +++ b/src/client.rs @@ -87,7 +87,15 @@ impl OneIo { let raw_reader: Box = match crate::get_protocol(path) { Some(protocol) => match protocol { #[cfg(feature = "http")] - "http" | "https" => Box::new(self.get_http_reader_raw(path)?), + #[cfg(feature = "http")] + "http" | "https" => { + let response = self.get_http_reader_raw(path)?; + Box::new(crate::resumable_http::ResumableHttpReader::new( + self.http_client().clone(), + path.to_string(), + response, + )) + } #[cfg(feature = "ftp")] "ftp" => remote::get_ftp_reader_raw(path)?, #[cfg(feature = "s3")] diff --git a/src/lib.rs b/src/lib.rs index b435658..d8326a2 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -236,6 +236,8 @@ mod client; mod compression; mod error; mod progress; +#[cfg(feature = "http")] +mod resumable_http; pub use builder::OneIoBuilder; pub use client::OneIo; diff --git a/src/resumable_http.rs b/src/resumable_http.rs new file mode 100644 index 0000000..aacba15 --- /dev/null +++ b/src/resumable_http.rs @@ -0,0 +1,81 @@ +//! A wrapper around an HTTP response that transparently retries using Range +//! requests when the connection is dropped (e.g., server send-timeout). +//! +//! This is critical for applications that read from multiple remote streams +//! concurrently: when one stream is paused while another is being consumed, +//! the server may close the idle connection. This reader detects the failure +//! and reconnects from where it left off. + +use reqwest::blocking::{Client, Response}; +use std::io::{self, Read}; + +/// Maximum number of consecutive retry attempts before giving up. +const MAX_RETRIES: u32 = 5; + +/// An HTTP reader that automatically resumes downloads using Range requests +/// when the underlying connection is dropped. +/// +/// The byte stream presented to the consumer is contiguous — reconnections +/// are invisible to layers above (e.g., decompressors). +pub(crate) struct ResumableHttpReader { + client: Client, + url: String, + response: Response, + /// Total raw (compressed) bytes successfully read so far. + offset: u64, +} + +impl ResumableHttpReader { + pub fn new(client: Client, url: String, response: Response) -> Self { + Self { + client, + url, + response, + offset: 0, + } + } +} + +impl Read for ResumableHttpReader { + fn read(&mut self, buf: &mut [u8]) -> io::Result { + match self.response.read(buf) { + Ok(0) => Ok(0), + Ok(n) => { + self.offset += n as u64; + Ok(n) + } + Err(original_err) => { + // Connection was reset/dropped — attempt to resume with Range + for attempt in 1..=MAX_RETRIES { + let backoff_ms = 200u64.saturating_mul(1u64 << attempt.min(4)); + std::thread::sleep(std::time::Duration::from_millis(backoff_ms)); + + let result = self + .client + .get(&self.url) + .header("Range", format!("bytes={}-", self.offset)) + .send(); + + match result { + Ok(resp) if resp.status() == reqwest::StatusCode::PARTIAL_CONTENT => { + // Successfully resumed + self.response = resp; + return self.read(buf); + } + Ok(resp) if resp.status() == reqwest::StatusCode::RANGE_NOT_SATISFIABLE => { + // Offset is at or past end of file — treat as EOF + return Ok(0); + } + Ok(_) | Err(_) => { + // Server doesn't support Range or other error — keep retrying + continue; + } + } + } + + // All retries exhausted — propagate original error + Err(original_err) + } + } + } +} From cc1d1b17c91972813f307046c91f9e8a09de2f57 Mon Sep 17 00:00:00 2001 From: JustinLoye Date: Wed, 15 Jul 2026 12:07:13 +0900 Subject: [PATCH 2/7] cover and test more edge cases --- CHANGELOG.md | 3 + src/client.rs | 1 - src/resumable_http.rs | 539 +++++++++++++++++++++++++++++++++++++++--- 3 files changed, 507 insertions(+), 36 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 7766a2d..a2455b6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,9 @@ All notable changes to this project will be documented in this file. ## [Unreleased] +### Added +- Remote HTTP(S) reads now resume automatically via Range requests: if the connection is dropped mid-download, the reader reconnects and continues from the last byte read instead of failing. Resumed responses are validated (`Content-Range` start offset, and `Last-Modified` when the original response provided it) to avoid splicing mismatched data. + ## v0.23.0 -- 2026-05-12 ### Added diff --git a/src/client.rs b/src/client.rs index f438258..07505b7 100644 --- a/src/client.rs +++ b/src/client.rs @@ -86,7 +86,6 @@ impl OneIo { pub fn get_reader_raw(&self, path: &str) -> Result, OneIoError> { let raw_reader: Box = match crate::get_protocol(path) { Some(protocol) => match protocol { - #[cfg(feature = "http")] #[cfg(feature = "http")] "http" | "https" => { let response = self.get_http_reader_raw(path)?; diff --git a/src/resumable_http.rs b/src/resumable_http.rs index aacba15..9e8ad25 100644 --- a/src/resumable_http.rs +++ b/src/resumable_http.rs @@ -12,6 +12,26 @@ use std::io::{self, Read}; /// Maximum number of consecutive retry attempts before giving up. const MAX_RETRIES: u32 = 5; +/// Normal delay +#[cfg(not(test))] +const BASE_RETRY_DELAY_MS: u64 = 200; + +/// Small delay for testing +#[cfg(test)] +const BASE_RETRY_DELAY_MS: u64 = 1; + +/// Parses the starting byte offset from a `Content-Range` header value. +/// +/// The header has the form `bytes -/` (RFC 9110 §14.4); +/// only `` is needed to confirm where the server resumed. Returns +/// `None` if the value is not in the expected form. +fn parse_content_range_start(value: &str) -> Option { + // "bytes 5-9/10" -> "5-9/10" -> "5" + let range = value.split_once(' ')?.1; + let start = range.split_once('-')?.0; + start.trim().parse().ok() +} + /// An HTTP reader that automatically resumes downloads using Range requests /// when the underlying connection is dropped. /// @@ -21,10 +41,22 @@ pub(crate) struct ResumableHttpReader { client: Client, url: String, response: Response, - /// Total raw (compressed) bytes successfully read so far. + /// Total raw bytes successfully read so far. offset: u64, } +/// Outcome of an attempt to resume the download from the current offset. +enum Resume { + /// The stream was replaced; read again from the new response. + Resumed, + /// The requested range is unsatisfiable; treat as a clean EOF. + Eof, + /// The server ignored the Range request; resuming is not possible. + Unsupported, + /// Every reconnection attempt failed to reach the server. + Failed, +} + impl ResumableHttpReader { pub fn new(client: Client, url: String, response: Response) -> Self { Self { @@ -34,48 +66,485 @@ impl ResumableHttpReader { offset: 0, } } + + /// Reconnects and resumes the download from `self.offset`. + /// + /// The request itself is retried up to `MAX_RETRIES` times with exponential + /// backoff to ride out transient connection failures. Returns `Err` only + /// when the server replies but its response is malformed (missing/invalid + /// `Content-Range`, or a start offset that does not match the request), + /// since continuing to read would corrupt the stream. + fn resume(&mut self) -> io::Result { + for attempt in 1..=MAX_RETRIES { + let backoff_ms = BASE_RETRY_DELAY_MS.saturating_mul(1u64 << attempt.min(4)); + std::thread::sleep(std::time::Duration::from_millis(backoff_ms)); + + let resp = match self + .client + .get(&self.url) + .header("Range", format!("bytes={}-", self.offset)) + .send() + { + Ok(resp) => resp, + // Couldn't reach the server — back off and try again. + Err(_) => continue, + }; + + return match resp.status() { + // Offset is at or past end of file. + reqwest::StatusCode::RANGE_NOT_SATISFIABLE => Ok(Resume::Eof), + // Server honored the Range request. + reqwest::StatusCode::PARTIAL_CONTENT => { + self.accept_resumed_response(resp)?; + Ok(Resume::Resumed) + } + // Anything else means the Range request was ignored. + _ => Ok(Resume::Unsupported), + }; + } + + Ok(Resume::Failed) + } + + /// Validates a `206 Partial Content` response and installs it as the current + /// stream. Fails if the server resumed from a wrong offset or a modified + /// resource, which would corrupt the stream. + fn accept_resumed_response(&mut self, resp: Response) -> io::Result<()> { + let content_range = resp + .headers() + .get(reqwest::header::CONTENT_RANGE) + .ok_or_else(|| { + io::Error::new( + io::ErrorKind::InvalidData, + "resumed response is missing the Content-Range header", + ) + })? + .to_str() + .map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e))?; + + let start = parse_content_range_start(content_range).ok_or_else(|| { + io::Error::new( + io::ErrorKind::InvalidData, + format!("malformed Content-Range header: {content_range}"), + ) + })?; + + if start != self.offset { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + format!("server resumed at byte {start}, expected {}", self.offset), + )); + } + + if let Some(last_modified) = self.response.headers().get(reqwest::header::LAST_MODIFIED) { + let resume_modified = resp.headers().get(reqwest::header::LAST_MODIFIED).ok_or_else(|| { + io::Error::new( + io::ErrorKind::InvalidData, + "resumed resource does not have a last-modified header, but original resource did" + ) + })?; + if resume_modified != last_modified { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + format!( + "resumed resource Last-Modified <{}> does not match the original <{}>", + String::from_utf8_lossy(resume_modified.as_bytes()), + String::from_utf8_lossy(last_modified.as_bytes()) + ), + )); + } + } + + self.response = resp; + Ok(()) + } } impl Read for ResumableHttpReader { fn read(&mut self, buf: &mut [u8]) -> io::Result { - match self.response.read(buf) { - Ok(0) => Ok(0), - Ok(n) => { - self.offset += n as u64; - Ok(n) - } - Err(original_err) => { - // Connection was reset/dropped — attempt to resume with Range - for attempt in 1..=MAX_RETRIES { - let backoff_ms = 200u64.saturating_mul(1u64 << attempt.min(4)); - std::thread::sleep(std::time::Duration::from_millis(backoff_ms)); - - let result = self - .client - .get(&self.url) - .header("Range", format!("bytes={}-", self.offset)) - .send(); - - match result { - Ok(resp) if resp.status() == reqwest::StatusCode::PARTIAL_CONTENT => { - // Successfully resumed - self.response = resp; - return self.read(buf); - } - Ok(resp) if resp.status() == reqwest::StatusCode::RANGE_NOT_SATISFIABLE => { - // Offset is at or past end of file — treat as EOF - return Ok(0); - } - Ok(_) | Err(_) => { - // Server doesn't support Range or other error — keep retrying - continue; - } + // Number of consecutive reconnections that delivered no new bytes. + // Reading real data returns immediately (so the next `read` call starts + // fresh), meaning this only fires when a server keeps reconnecting us + // without making progress — e.g. repeated empty `206` responses — which + // would otherwise loop forever. + let mut stalled_retries = 0u32; + + loop { + match self.response.read(buf) { + Ok(0) => return Ok(0), + Ok(n) => { + self.offset += n as u64; + return Ok(n); + } + Err(original_err) => { + // Connection was reset/dropped — attempt to resume with Range. + if stalled_retries >= MAX_RETRIES { + return Err(original_err); } + stalled_retries += 1; + + match self.resume()? { + // Read again from the freshly reconnected response. + Resume::Resumed => continue, + // Nothing more to read. + Resume::Eof => return Ok(0), + // Can't resume — surface the original failure. + Resume::Unsupported | Resume::Failed => return Err(original_err), + } + } + } + } + } +} + +#[cfg(test)] +mod test { + + use std::sync::{ + atomic::{AtomicUsize, Ordering}, + Arc, + }; + use std::{ + io::prelude::*, + net::{TcpListener, TcpStream}, + thread, + }; + + use crate::resumable_http::{ResumableHttpReader, MAX_RETRIES}; + + /// Reads a full HTTP request header block (up to and including the blank + /// CRLF line) from a stream. Reading byte-by-byte avoids consuming past the + /// header block and works regardless of how the request is split across TCP + /// reads. Returns whatever was read if the peer closes the connection early. + fn read_request(stream: &mut TcpStream) -> String { + let mut data = Vec::new(); + let mut byte = [0u8; 1]; + while let Ok(1) = stream.read(&mut byte) { + data.push(byte[0]); + if data.ends_with(b"\r\n\r\n") { + break; + } + } + String::from_utf8_lossy(&data).to_string() + } + + // Check reader works normally when there is no need to resume + #[test] + fn no_drop() { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let port = listener.local_addr().unwrap().port(); + let url = format!("http://127.0.0.1:{}/data.txt", port); + + let handle = thread::spawn(move || { + let (mut stream, _) = listener.accept().unwrap(); + + // Read the request to prevent reqwest from throwing an error + read_request(&mut stream); + + let response = "HTTP/1.1 200 OK\r\nContent-Length: 10\r\n\r\n1234567890"; + stream.write_all(response.as_bytes()).unwrap(); + }); + + let client = reqwest::blocking::Client::new(); + let resp = client.get(&url).send().unwrap(); + let mut reader = ResumableHttpReader::new(client, url, resp); + + let mut buf = String::new(); + reader.read_to_string(&mut buf).unwrap(); + + assert_eq!(buf.as_str(), "1234567890"); + handle.join().unwrap(); + } + + // Check reader resumes when server drops connection + #[test] + fn drop_resume() { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let port = listener.local_addr().unwrap().port(); + let url = format!("http://127.0.0.1:{}/data.txt", port); + + let handle = thread::spawn(move || { + let (mut stream1, _) = listener.accept().unwrap(); + read_request(&mut stream1); + + let response_part1 = "HTTP/1.1 200 OK\r\nContent-Length: 10\r\nLast-Modified: Tue, 15 Nov 1994 12:45:26 GMT\r\n\r\n12345"; + stream1.write_all(response_part1.as_bytes()).unwrap(); + drop(stream1); + + let (mut stream2, _) = listener.accept().unwrap(); + let req = read_request(&mut stream2); + assert!(req.contains("range: bytes=5-")); + + let response_part2 = "HTTP/1.1 206 Partial Content\r\nContent-Length: 5\r\nContent-Range: bytes 5-9/10\r\nLast-Modified: Tue, 15 Nov 1994 12:45:26 GMT\r\n\r\n67890"; + stream2.write_all(response_part2.as_bytes()).unwrap(); + }); + + let client = reqwest::blocking::Client::new(); + let resp = client.get(&url).send().unwrap(); + let mut reader = ResumableHttpReader::new(client, url, resp); + + let mut buf = String::new(); + reader.read_to_string(&mut buf).unwrap(); + + assert_eq!(buf.as_str(), "1234567890"); + handle.join().unwrap(); + } + + // Check reader returns an error when the server does not support Ranges + #[test] + fn range_not_supported_is_err() { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let port = listener.local_addr().unwrap().port(); + let url = format!("http://127.0.0.1:{}/data.txt", port); + + let handle = thread::spawn(move || { + let (mut stream1, _) = listener.accept().unwrap(); + read_request(&mut stream1); + + let response_part1 = "HTTP/1.1 200 OK\r\nContent-Length: 10\r\n\r\n12345"; + stream1.write_all(response_part1.as_bytes()).unwrap(); + drop(stream1); + + let (mut stream2, _) = listener.accept().unwrap(); + read_request(&mut stream2); + + // A server not supporting ranges respond with a 200 + let response_part2 = "HTTP/1.1 200 OK\r\nContent-Length: 10\r\n\r\n1234567890"; + stream2.write_all(response_part2.as_bytes()).unwrap(); + }); + + let client = reqwest::blocking::Client::new(); + let resp = client.get(&url).send().unwrap(); + let mut reader = ResumableHttpReader::new(client, url, resp); + + let mut buf = String::new(); + assert!(reader.read_to_string(&mut buf).is_err()); + handle.join().unwrap(); + } + + // Check out of bound range requests are treated as EOF + #[test] + fn range_oob_is_eof() { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let port = listener.local_addr().unwrap().port(); + let url = format!("http://127.0.0.1:{}/data.txt", port); + + let handle = thread::spawn(move || { + let (mut stream1, _) = listener.accept().unwrap(); + read_request(&mut stream1); + + // Declare more bytes than we actually send, then drop the connection + // so the client is forced to resume from offset 10. + let response_part1 = "HTTP/1.1 200 OK\r\nContent-Length: 20\r\n\r\n1234567890"; + stream1.write_all(response_part1.as_bytes()).unwrap(); + drop(stream1); + + // The resume request starts past the end of the real content, so the + // server reports 416 which the reader must treat as a clean EOF. + let (mut stream2, _) = listener.accept().unwrap(); + let req = read_request(&mut stream2); + assert!(req.contains("range: bytes=10-")); + + let response_part2 = "HTTP/1.1 416 Range Not Satisfiable\r\nContent-Length: 0\r\n\r\n"; + stream2.write_all(response_part2.as_bytes()).unwrap(); + }); + + let client = reqwest::blocking::Client::new(); + let resp = client.get(&url).send().unwrap(); + let mut reader = ResumableHttpReader::new(client, url, resp); + + let mut buf = String::new(); + reader.read_to_string(&mut buf).unwrap(); + assert_eq!(buf.as_str(), "1234567890"); + handle.join().unwrap(); + } + + // Check reader returns error when resuming reading a resource that was modified + #[test] + fn new_last_modified_is_err() { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let port = listener.local_addr().unwrap().port(); + let url = format!("http://127.0.0.1:{}/data.txt", port); + + let handle = thread::spawn(move || { + let (mut stream1, _) = listener.accept().unwrap(); + read_request(&mut stream1); + + let response_part1 = + "HTTP/1.1 200 OK\r\nContent-Length: 10\r\nLast-Modified: Tue, 15 Nov 1994 12:45:26 GMT\r\n\r\n12345"; + stream1.write_all(response_part1.as_bytes()).unwrap(); + drop(stream1); + + let (mut stream2, _) = listener.accept().unwrap(); + let req = read_request(&mut stream2); + assert!(req.contains("range: bytes=5-")); + + let response_part2 = "HTTP/1.1 206 Partial Content\r\nContent-Length: 5\r\nContent-Range: bytes 5-9/10\r\nLast-Modified: Tue, 15 Nov 1995 12:45:26 GMT\r\n\r\n67890"; + stream2.write_all(response_part2.as_bytes()).unwrap(); + }); + + let client = reqwest::blocking::Client::new(); + let resp = client.get(&url).send().unwrap(); + let mut reader = ResumableHttpReader::new(client, url, resp); + + let mut buf = String::new(); + assert!(reader.read_to_string(&mut buf).is_err()); + + handle.join().unwrap(); + } + + // Check reader returns error when the resumed response drops the + // Last-Modified header that the original response carried + #[test] + fn missing_last_modified_on_resume_is_err() { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let port = listener.local_addr().unwrap().port(); + let url = format!("http://127.0.0.1:{}/data.txt", port); + + let handle = thread::spawn(move || { + let (mut stream1, _) = listener.accept().unwrap(); + read_request(&mut stream1); + + let response_part1 = + "HTTP/1.1 200 OK\r\nContent-Length: 10\r\nLast-Modified: Tue, 15 Nov 1994 12:45:26 GMT\r\n\r\n12345"; + stream1.write_all(response_part1.as_bytes()).unwrap(); + drop(stream1); + + let (mut stream2, _) = listener.accept().unwrap(); + let req = read_request(&mut stream2); + assert!(req.contains("range: bytes=5-")); + + // The resumed response omits the Last-Modified header entirely. + let response_part2 = "HTTP/1.1 206 Partial Content\r\nContent-Length: 5\r\nContent-Range: bytes 5-9/10\r\n\r\n67890"; + stream2.write_all(response_part2.as_bytes()).unwrap(); + }); + + let client = reqwest::blocking::Client::new(); + let resp = client.get(&url).send().unwrap(); + let mut reader = ResumableHttpReader::new(client, url, resp); + + let mut buf = String::new(); + assert!(reader.read_to_string(&mut buf).is_err()); + + handle.join().unwrap(); + } + + // Check reader retries until MAX_RETRIES and then return an error + #[test] + fn max_retries_exhausted_is_err() { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let port = listener.local_addr().unwrap().port(); + let url = format!("http://127.0.0.1:{}/data.txt", port); + + // Set up a thread-safe counter to track how many times the client reconnects + let attempts_counter = Arc::new(AtomicUsize::new(0)); + let server_counter = attempts_counter.clone(); + + let handle = thread::spawn(move || { + // Handle the initial successful request + let (mut stream1, _) = listener.accept().unwrap(); + read_request(&mut stream1); + let response_part1 = "HTTP/1.1 200 OK\r\nContent-Length: 10\r\n\r\n12345"; + stream1.write_all(response_part1.as_bytes()).unwrap(); + drop(stream1); // Drop to trigger the client's retry logic + + // Loop indefinitely to prove read stop comes from the client + loop { + let (mut stream, _) = listener.accept().unwrap(); + let req = read_request(&mut stream); + + // Look for the "poison pill" to know when to gracefully shut down + if req.is_empty() || req.starts_with("STOP") { + break; } - // All retries exhausted — propagate original error - Err(original_err) + server_counter.fetch_add(1, Ordering::SeqCst); + drop(stream); } + }); + + // Setup the client and initial request + let client = reqwest::blocking::Client::new(); + let resp = client.get(&url).send().unwrap(); + let mut reader = ResumableHttpReader::new(client, url, resp); + let mut buf = String::new(); + + // Reader should exhaust its retries and fail. + assert!(reader.read_to_string(&mut buf).is_err()); + + // Reader has given up. Now, send a dummy "STOP" request to unblock + // the server's listener.accept() call so the thread can die cleanly. + if let Ok(mut wake_stream) = TcpStream::connect(format!("127.0.0.1:{}", port)) { + let _ = wake_stream.write_all(b"STOP"); } + + handle.join().unwrap(); + + // Assert that the client stopped exactly when it was supposed to + assert_eq!( + attempts_counter.load(Ordering::SeqCst), + MAX_RETRIES as usize + ); + } + + // Check reader does not loop forever when the server keeps returning a 206 + // whose body never arrives. + #[test] + fn no_data_206() { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let port = listener.local_addr().unwrap().port(); + let url = format!("http://127.0.0.1:{}/data.txt", port); + + // Set up a thread-safe counter to track how many times the client reconnects + let attempts_counter = Arc::new(AtomicUsize::new(0)); + let server_counter = attempts_counter.clone(); + + let handle = thread::spawn(move || { + // Handle the initial successful request + let (mut stream1, _) = listener.accept().unwrap(); + read_request(&mut stream1); + let response_part1 = "HTTP/1.1 200 OK\r\nContent-Length: 10\r\n\r\n12345"; + stream1.write_all(response_part1.as_bytes()).unwrap(); + drop(stream1); // Drop to trigger the client's retry logic + + // Every resume gets a well-formed 206 whose declared body never arrives. + let response_206 = "HTTP/1.1 206 Partial Content\r\nContent-Length: 5\r\nContent-Range: bytes 5-9/10\r\n\r\n"; + loop { + let (mut stream, _) = listener.accept().unwrap(); + let req = read_request(&mut stream); + + // Look for the "poison pill" to know when to gracefully shut down + if req.is_empty() || req.starts_with("STOP") { + break; + } + + server_counter.fetch_add(1, Ordering::SeqCst); + stream.write_all(response_206.as_bytes()).unwrap(); + drop(stream); + } + }); + + // Setup the client and initial request + let client = reqwest::blocking::Client::new(); + let resp = client.get(&url).send().unwrap(); + let mut reader = ResumableHttpReader::new(client, url, resp); + let mut buf = String::new(); + + // Reader should give up instead of spinning forever. + assert!(reader.read_to_string(&mut buf).is_err()); + + // Reader has given up. Now, send a dummy "STOP" request to unblock + // the server's listener.accept() call so the thread can die cleanly. + if let Ok(mut wake_stream) = TcpStream::connect(format!("127.0.0.1:{}", port)) { + let _ = wake_stream.write_all(b"STOP"); + } + + handle.join().unwrap(); + + // The reader stopped after exactly MAX_RETRIES no-progress attempts. + assert_eq!( + attempts_counter.load(Ordering::SeqCst), + MAX_RETRIES as usize + ); } } From 03ce583bcef6938327f58e7da49cbb7bffc8aa1b Mon Sep 17 00:00:00 2001 From: JustinLoye Date: Wed, 15 Jul 2026 14:35:12 +0900 Subject: [PATCH 3/7] make download() also use the resumable http reader --- CHANGELOG.md | 2 +- src/client.rs | 9 +++++++-- src/resumable_http.rs | 37 +++++++++++++++++++++++++++++++++++++ 3 files changed, 45 insertions(+), 3 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index a2455b6..93f54c6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,7 +5,7 @@ All notable changes to this project will be documented in this file. ## [Unreleased] ### Added -- Remote HTTP(S) reads now resume automatically via Range requests: if the connection is dropped mid-download, the reader reconnects and continues from the last byte read instead of failing. Resumed responses are validated (`Content-Range` start offset, and `Last-Modified` when the original response provided it) to avoid splicing mismatched data. +- Remote HTTP(S) reads and downloads now resume automatically via Range requests: if the connection is dropped mid-transfer, the reader reconnects and continues from the last byte read instead of failing. Applies to both streaming readers (`get_reader`/`get_reader_raw`) and `download`/`download_with_retry`. Resumed responses are validated (`Content-Range` start offset, and `Last-Modified` when the original response provided it) to avoid splicing mismatched data. ## v0.23.0 -- 2026-05-12 diff --git a/src/client.rs b/src/client.rs index 07505b7..4301051 100644 --- a/src/client.rs +++ b/src/client.rs @@ -319,8 +319,13 @@ impl OneIo { #[cfg(feature = "http")] Some("http" | "https") => { let mut writer = self.get_writer_raw(local_path)?; - let mut response = self.get_http_reader_raw(remote_path)?; - response.copy_to(&mut writer)?; + let response = self.get_http_reader_raw(remote_path)?; + let mut reader = crate::resumable_http::ResumableHttpReader::new( + self.http_client().clone(), + remote_path.to_string(), + response, + ); + std::io::copy(&mut reader, &mut writer)?; Ok(()) } #[cfg(feature = "ftp")] diff --git a/src/resumable_http.rs b/src/resumable_http.rs index 9e8ad25..5844ade 100644 --- a/src/resumable_http.rs +++ b/src/resumable_http.rs @@ -547,4 +547,41 @@ mod test { MAX_RETRIES as usize ); } + + // Check `download()` resumes when the server drops the connection mid-body, + // so the file written to disk contains the complete content. This exercises + // the resumable wiring through the download path (not just a direct reader). + #[test] + fn download_resumes_after_drop() { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let port = listener.local_addr().unwrap().port(); + let url = format!("http://127.0.0.1:{}/data.txt", port); + + let handle = thread::spawn(move || { + let (mut stream1, _) = listener.accept().unwrap(); + read_request(&mut stream1); + + let response_part1 = "HTTP/1.1 200 OK\r\nContent-Length: 10\r\nLast-Modified: Tue, 15 Nov 1994 12:45:26 GMT\r\n\r\n12345"; + stream1.write_all(response_part1.as_bytes()).unwrap(); + drop(stream1); + + let (mut stream2, _) = listener.accept().unwrap(); + let req = read_request(&mut stream2); + assert!(req.contains("range: bytes=5-")); + + let response_part2 = "HTTP/1.1 206 Partial Content\r\nContent-Length: 5\r\nContent-Range: bytes 5-9/10\r\nLast-Modified: Tue, 15 Nov 1994 12:45:26 GMT\r\n\r\n67890"; + stream2.write_all(response_part2.as_bytes()).unwrap(); + }); + + let local_path = std::env::temp_dir().join(format!("oneio_download_resume_{}.txt", port)); + let local = local_path.to_str().unwrap(); + + crate::OneIo::new().unwrap().download(&url, local).unwrap(); + + let content = std::fs::read_to_string(&local_path).unwrap(); + let _ = std::fs::remove_file(&local_path); + + assert_eq!(content.as_str(), "1234567890"); + handle.join().unwrap(); + } } From 3e3fdb67fca4c3507de3bfefa08dd56d03912a37 Mon Sep 17 00:00:00 2001 From: JustinLoye <32801681+JustinLoye@users.noreply.github.com> Date: Wed, 15 Jul 2026 15:01:48 +0900 Subject: [PATCH 4/7] validate bytes is in content range header Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com> --- src/resumable_http.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/resumable_http.rs b/src/resumable_http.rs index 5844ade..ebb5a11 100644 --- a/src/resumable_http.rs +++ b/src/resumable_http.rs @@ -27,7 +27,7 @@ const BASE_RETRY_DELAY_MS: u64 = 1; /// `None` if the value is not in the expected form. fn parse_content_range_start(value: &str) -> Option { // "bytes 5-9/10" -> "5-9/10" -> "5" - let range = value.split_once(' ')?.1; + let range = value.strip_prefix("bytes ")?; let start = range.split_once('-')?.0; start.trim().parse().ok() } From 4bbfbd2e18014b0c9dfb0dd998dfbcfcae5082bd Mon Sep 17 00:00:00 2001 From: JustinLoye Date: Wed, 15 Jul 2026 17:17:52 +0900 Subject: [PATCH 5/7] remove backoff for first resume attempt and use reqwest range header name --- src/resumable_http.rs | 13 +++++++------ 1 file changed, 7 insertions(+), 6 deletions(-) diff --git a/src/resumable_http.rs b/src/resumable_http.rs index ebb5a11..2391dbd 100644 --- a/src/resumable_http.rs +++ b/src/resumable_http.rs @@ -75,19 +75,20 @@ impl ResumableHttpReader { /// `Content-Range`, or a start offset that does not match the request), /// since continuing to read would corrupt the stream. fn resume(&mut self) -> io::Result { - for attempt in 1..=MAX_RETRIES { - let backoff_ms = BASE_RETRY_DELAY_MS.saturating_mul(1u64 << attempt.min(4)); - std::thread::sleep(std::time::Duration::from_millis(backoff_ms)); - + for attempt in 0..MAX_RETRIES { let resp = match self .client .get(&self.url) - .header("Range", format!("bytes={}-", self.offset)) + .header(reqwest::header::RANGE, format!("bytes={}-", self.offset)) .send() { Ok(resp) => resp, // Couldn't reach the server — back off and try again. - Err(_) => continue, + Err(_) => { + let backoff_ms = BASE_RETRY_DELAY_MS.saturating_mul(1u64 << attempt.min(4)); + std::thread::sleep(std::time::Duration::from_millis(backoff_ms)); + continue; + } }; return match resp.status() { From 1f9f908c548b4eece65e40a82ab2ed0bce99afbe Mon Sep 17 00:00:00 2001 From: JustinLoye Date: Wed, 15 Jul 2026 18:17:48 +0900 Subject: [PATCH 6/7] flush the writer before returning Ok in download() --- src/client.rs | 1 + 1 file changed, 1 insertion(+) diff --git a/src/client.rs b/src/client.rs index 4301051..9923ed0 100644 --- a/src/client.rs +++ b/src/client.rs @@ -326,6 +326,7 @@ impl OneIo { response, ); std::io::copy(&mut reader, &mut writer)?; + writer.flush()?; Ok(()) } #[cfg(feature = "ftp")] From c8e95572dfcea16c07651baa818eb8e75e09be29 Mon Sep 17 00:00:00 2001 From: JustinLoye Date: Wed, 15 Jul 2026 18:23:54 +0900 Subject: [PATCH 7/7] more robust range request header check in tests --- src/resumable_http.rs | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/src/resumable_http.rs b/src/resumable_http.rs index 2391dbd..94b0575 100644 --- a/src/resumable_http.rs +++ b/src/resumable_http.rs @@ -274,7 +274,7 @@ mod test { let (mut stream2, _) = listener.accept().unwrap(); let req = read_request(&mut stream2); - assert!(req.contains("range: bytes=5-")); + assert!(req.to_ascii_lowercase().contains("range: bytes=5-")); let response_part2 = "HTTP/1.1 206 Partial Content\r\nContent-Length: 5\r\nContent-Range: bytes 5-9/10\r\nLast-Modified: Tue, 15 Nov 1994 12:45:26 GMT\r\n\r\n67890"; stream2.write_all(response_part2.as_bytes()).unwrap(); @@ -344,7 +344,7 @@ mod test { // server reports 416 which the reader must treat as a clean EOF. let (mut stream2, _) = listener.accept().unwrap(); let req = read_request(&mut stream2); - assert!(req.contains("range: bytes=10-")); + assert!(req.to_ascii_lowercase().contains("range: bytes=10-")); let response_part2 = "HTTP/1.1 416 Range Not Satisfiable\r\nContent-Length: 0\r\n\r\n"; stream2.write_all(response_part2.as_bytes()).unwrap(); @@ -378,7 +378,7 @@ mod test { let (mut stream2, _) = listener.accept().unwrap(); let req = read_request(&mut stream2); - assert!(req.contains("range: bytes=5-")); + assert!(req.to_ascii_lowercase().contains("range: bytes=5-")); let response_part2 = "HTTP/1.1 206 Partial Content\r\nContent-Length: 5\r\nContent-Range: bytes 5-9/10\r\nLast-Modified: Tue, 15 Nov 1995 12:45:26 GMT\r\n\r\n67890"; stream2.write_all(response_part2.as_bytes()).unwrap(); @@ -413,7 +413,7 @@ mod test { let (mut stream2, _) = listener.accept().unwrap(); let req = read_request(&mut stream2); - assert!(req.contains("range: bytes=5-")); + assert!(req.to_ascii_lowercase().contains("range: bytes=5-")); // The resumed response omits the Last-Modified header entirely. let response_part2 = "HTTP/1.1 206 Partial Content\r\nContent-Length: 5\r\nContent-Range: bytes 5-9/10\r\n\r\n67890"; @@ -568,7 +568,7 @@ mod test { let (mut stream2, _) = listener.accept().unwrap(); let req = read_request(&mut stream2); - assert!(req.contains("range: bytes=5-")); + assert!(req.to_ascii_lowercase().contains("range: bytes=5-")); let response_part2 = "HTTP/1.1 206 Partial Content\r\nContent-Length: 5\r\nContent-Range: bytes 5-9/10\r\nLast-Modified: Tue, 15 Nov 1994 12:45:26 GMT\r\n\r\n67890"; stream2.write_all(response_part2.as_bytes()).unwrap();