Repository: algesten/ureq (rust)

Issue:
Timeout is reset on stream when calling `into_reader()`
When calling `into_reader()` on a request, the timeout set with `set_timeout_read` (NOT the deadline `set_timeout`) on the request is reset to no timeout by `unit::connect()` calling `DeadlineStream::reset` before the socket is returned, and there is no way to set it again.

This makes `set_timeout_read` basically unused except for reading the headers.

Pull request:
Reinstate read timeouts on body.
This feature was broken in #67, which reset timeouts on the
stream before passing it to set_stream.

As part of this change, refactor the internal storage of
timeouts on the Request object to use Option<Duration>.

Remove the deadline field on Response. It wasn't used. The
deadline field on unit was used instead.

Add a unittest.

Fixes #196. /cc @birkenfeld

Patch:
diff --git a/src/request.rs b/src/request.rs
index daee345..7d469fb 100644
--- a/src/request.rs
+++ b/src/request.rs
@@ -1,8 +1,8 @@
-use std::fmt;
 use std::io::Read;
 use std::result::Result;
 use std::sync::{Arc, Mutex};
 use std::time;
+use std::{fmt, time::Duration};
 
 use qstring::QString;
 use url::{form_urlencoded, Url};
@@ -40,8 +40,8 @@ pub struct Request {
     pub(crate) headers: Vec<Header>,
     pub(crate) query: QString,
     pub(crate) timeout_connect: u64,
-    pub(crate) timeout_read: u64,
-    pub(crate) timeout_write: u64,
+    pub(crate) timeout_read: Option<time::Duration>,
+    pub(crate) timeout_write: Option<time::Duration>,
     pub(crate) timeout: Option<time::Duration>,
     pub(crate) redirects: u32,
     pub(crate) proxy: Option<Proxy>,
@@ -368,7 +368,10 @@ impl Request {
     /// println!("{:?}", r);
     /// ```
     pub fn timeout_read(&mut self, millis: u64) -> &mut Request {
-        self.timeout_read = millis;
+        match millis {
+            0 => self.timeout_read = None,
+            m => self.timeout_read = Some(Duration::from_millis(m)),
+        }
         self
     }
 
@@ -385,7 +388,10 @@ impl Request {
     /// println!("{:?}", r);
     /// ```
     pub fn timeout_write(&mut self, millis: u64) -> &mut Request {
-        self.timeout_write = millis;
+        match millis {
+            0 => self.timeout_write = None,
+            m => self.timeout_write = Some(Duration::from_millis(m)),
+        }
         self
     }
 
diff --git a/src/response.rs b/src/response.rs
index 38f44f1..75ba391 100644
--- a/src/response.rs
+++ b/src/response.rs
@@ -1,7 +1,6 @@
 use std::fmt;
 use std::io::{self, Cursor, ErrorKind, Read};
 use std::str::FromStr;
-use std::time::Instant;
 
 use chunked_transfer::Decoder as ChunkDecoder;
 
@@ -56,7 +55,6 @@ pub struct Response {
     headers: Vec<Header>,
     unit: Option<Unit>,
     stream: Option<Stream>,
-    deadline: Option<Instant>,
 }
 
 /// index into status_line where we split: HTTP/1.1 200 OK
@@ -221,6 +219,11 @@ impl Response {
         &self.error
     }
 
+    // Internal-only API, to allow unit::connect to return early on errors.
+    pub(crate) fn into_error(self) -> Option<Error> {
+        self.error
+    }
+
     /// The content type part of the "Content-Type" header without
     /// the charset.
     ///
@@ -322,12 +325,17 @@ impl Response {
 
         let stream = self.stream.expect("No reader in response?!");
         let unit = self.unit;
+        if let Some(unit) = &unit {
+            let result = stream.set_read_timeout(unit.req.timeout_read);
+            if let Err(e) = result {
+                return Box::new(ErrorReader(e)) as Box<dyn Read + Send>;
+            }
+        }
         let deadline = unit.as_ref().and_then(|u| u.deadline);
         let stream = DeadlineStream::new(stream, deadline);
 
         match (use_chunked, limit_bytes) {
-            (true, _) => Box::new(PoolReturnRead::new(unit, ChunkDecoder::new(stream)))
-                as Box<dyn Read + Send>,
+            (true, _) => Box::new(PoolReturnRead::new(unit, ChunkDecoder::new(stream))),
             (false, Some(len)) => {
                 Box::new(PoolReturnRead::new(unit, LimitedRead::new(stream, len)))
             }
@@ -500,7 +508,6 @@ impl Response {
             headers,
             unit: None,
             stream: None,
-            deadline: None,
         })
     }
 
@@ -580,9 +587,6 @@ impl Into<Response> for Error {
 /// *Internal API*
 pub(crate) fn set_stream(resp: &mut Response, url: String, unit: Option<Unit>, stream: Stream) {
     resp.url = Some(url);
-    if let Some(unit) = &unit {
-        resp.deadline = unit.deadline;
-    }
     resp.unit = unit;
     resp.stream = Some(stream);
 }
@@ -808,3 +812,14 @@ mod tests {
         assert_eq!(v, "Bad Status\n");
     }
 }
+
+// ErrorReader returns an error for every read.
+// The error is as close to a clone of the underlying
+// io::Error as we can get.
+struct ErrorReader(io::Error);
+
+impl Read for ErrorReader {
+    fn read(&mut self, _buf: &mut [u8]) -> io::Result<usize> {
+        Err(io::Error::new(self.0.kind(), self.0.to_string()))
+    }
+}
diff --git a/src/stream.rs b/src/stream.rs
index f14a3ae..d057e53 100644
--- a/src/stream.rs
+++ b/src/stream.rs
@@ -172,6 +172,14 @@ impl Stream {
         }
     }
 
+    pub(crate) fn set_read_timeout(&self, timeout: Option<Duration>) -> io::Result<()> {
+        if let Some(socket) = self.socket() {
+            socket.set_read_timeout(timeout)
+        } else {
+            Ok(())
+        }
+    }
+
     #[cfg(test)]
     pub fn to_write_vec(&self) -> Vec<u8> {
         match self {
@@ -447,30 +455,16 @@ pub(crate) fn connect_host(unit: &Unit, hostname: &str, port: u16) -> Result<Tcp
         return Err(err);
     };
 
-    // rust's absurd api returns Err if we set 0.
-    // Setting it to None will disable the native system timeout
     if let Some(deadline) = deadline {
-        stream
-            .set_read_timeout(Some(time_until_deadline(deadline)?))
-            .ok();
-    } else if unit.req.timeout_read > 0 {
-        stream
-            .set_read_timeout(Some(Duration::from_millis(unit.req.timeout_read as u64)))
-            .ok();
+        stream.set_read_timeout(Some(time_until_deadline(deadline)?))?;
     } else {
-        stream.set_read_timeout(None).ok();
+        stream.set_read_timeout(unit.req.timeout_read)?;
     }
 
     if let Some(deadline) = deadline {
-        stream
-            .set_write_timeout(Some(time_until_deadline(deadline)?))
-            .ok();
-    } else if unit.req.timeout_write > 0 {
-        stream
-            .set_write_timeout(Some(Duration::from_millis(unit.req.timeout_write as u64)))
-            .ok();
+        stream.set_write_timeout(Some(time_until_deadline(deadline)?))?;
     } else {
-        stream.set_write_timeout(None).ok();
+        stream.set_read_timeout(unit.req.timeout_read)?;
     }
 
     if proto == Some(Proto::HTTPConnect) {
diff --git a/src/unit.rs b/src/unit.rs
index 3ec4794..2957af1 100644
--- a/src/unit.rs
+++ b/src/unit.rs
@@ -198,6 +198,8 @@ pub(crate) fn connect(
             let empty = Payload::Empty.into_read();
             return connect(req, unit, false, redirect_count, empty, redir);
         }
+        // Non-retryable errors return early.
+        return Err(resp.into_error().unwrap());
     }
 
     // squirrel away cookies


Test patch:
diff --git a/src/test/timeout.rs b/src/test/timeout.rs
index cb69055..c9a0821 100644
--- a/src/test/timeout.rs
+++ b/src/test/timeout.rs
@@ -19,7 +19,7 @@ fn dribble_body_respond(mut stream: TcpStream, contents: &[u8]) -> io::Result<()
         stream.write_all(&contents[i..i + 1])?;
         stream.write_all(&[b'\n'; 1])?;
         stream.flush()?;
-        thread::sleep(Duration::from_millis(10));
+        thread::sleep(Duration::from_millis(100));
     }
     Ok(())
 }
@@ -47,10 +47,25 @@ fn overall_timeout_during_body() {
     get_and_expect_timeout(url);
 }
 
+#[test]
+fn read_timeout_during_body() {
+    let server = TestServer::new(|stream| dribble_body_respond(stream, &[b'a'; 300]));
+    let url = format!("http://localhost:{}/", server.port);
+    let agent = Agent::default().build();
+    let resp = agent.get(&url).timeout_read(5).call();
+    match resp.into_string() {
+        Err(io_error) => match io_error.kind() {
+            io::ErrorKind::TimedOut => Ok(()),
+            _ => Err(format!("{:?}", io_error)),
+        },
+        Ok(_) => Err("successful response".to_string()),
+    }
+    .expect("expected timeout but got something else");
+}
+
 // Send HTTP headers on the TcpStream at a rate of one header every 100
 // milliseconds, for a total of 30 headers.
 fn dribble_headers_respond(mut stream: TcpStream) -> io::Result<()> {
-    stream.write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 0\r\n")?;
     for _ in 0..30 {
         stream.write_all(b"a: b\n")?;
         stream.flush()?;
@@ -59,12 +74,35 @@ fn dribble_headers_respond(mut stream: TcpStream) -> io::Result<()> {
     Ok(())
 }
 
+#[test]
+fn read_timeout_during_headers() {
+    let server = TestServer::new(dribble_headers_respond);
+    let url = format!("http://localhost:{}/", server.port);
+    let resp = crate::get(&url).timeout_read(10).call();
+    assert!(!resp.ok());
+    assert_eq!(
+        resp.into_string().unwrap(),
+        "Network Error: timed out reading response\n"
+    );
+}
+
 #[test]
 fn overall_timeout_during_headers() {
     // Start a test server on an available port, that dribbles out a response at 1 write per 10ms.
     let server = TestServer::new(dribble_headers_respond);
     let url = format!("http://localhost:{}/", server.port);
-    get_and_expect_timeout(url);
+    let agent = Agent::default().build();
+    let timeout = Duration::from_millis(500);
+    let resp = agent.get(&url).timeout(timeout).call();
+    assert!(
+        matches!(resp.synthetic_error(), Some(Error::Io(_))),
+        "expected timeout error, got {:?}",
+        resp.synthetic_error()
+    );
+    assert_eq!(
+        resp.synthetic_error().as_ref().unwrap().body_text(),
+        "Network Error: timed out reading response"
+    );
 }
 
 #[test]