Repository: algesten/ureq
Pull request #372 (2021-04-17 18:14:03)

Don't reuse conns with bytes pending from server
Related to #361 and #124.


Patch:
diff --git a/examples/cureq/main.rs b/examples/cureq/main.rs
index 183f8b0..0f49f09 100644
--- a/examples/cureq/main.rs
+++ b/examples/cureq/main.rs
@@ -1,6 +1,7 @@
 use std::error;
 use std::fmt;
 use std::io;
+use std::thread;
 use std::time::Duration;
 use std::{env, sync::Arc};
 
@@ -64,8 +65,19 @@ impl From<io::Error> for Error {
     }
 }
 
-fn get(agent: &ureq::Agent, url: &str, print_headers: bool) -> Result<(), Error> {
-    let response = agent.get(url).call()?;
+fn perform(
+    agent: &ureq::Agent,
+    method: &str,
+    url: &str,
+    data: &[u8],
+    print_headers: bool,
+) -> Result<(), Error> {
+    let req = agent.request(method, url);
+    let response = if method == "GET" && data.len() == 0 {
+        req.call()?
+    } else {
+        req.send_bytes(data)?
+    };
     if print_headers {
         println!(
             "{} {} {}",
@@ -108,29 +120,43 @@ fn main() {
 }
 
 fn main2() -> Result<(), Error> {
-    let mut args: Vec<String> = env::args().collect();
-    if args.len() == 1 {
+    let mut args = env::args();
+    if args.next().is_none() {
         println!(
             r##"Usage: {:#?} url [url ...]
-            
+
+-i            Include headers when printing response
+-X <method>   Use the given request method (GET, POST, etc)
+-d <string>   Use the given data as the request body (useful for POST)
+--wait <n>    Wait n seconds between requests
+-k            Ignore certificate errors
+
 Fetch url and copy it to stdout.
 "##,
             env::current_exe()?
         );
         return Ok(());
     }
-    args.remove(0);
     env_logger::init();
     let mut builder = ureq::builder()
         .timeout_connect(Duration::from_secs(30))
         .timeout(Duration::from_secs(300));
-    let flags: Vec<&String> = args.iter().filter(|s| s.starts_with("-")).collect();
-    let nonflags: Vec<&String> = args.iter().filter(|s| !s.starts_with("-")).collect();
+    let mut nonflags: Vec<String> = vec![];
 
     let mut print_headers: bool = false;
-    for flag in flags {
-        match flag.as_ref() {
+    let mut method: String = "GET".into();
+    let mut data: Vec<u8> = vec![];
+    let mut wait = Duration::new(0, 0);
+    while let Some(arg) = args.next() {
+        match arg.as_ref() {
             "-i" => print_headers = true,
+            "-X" => method = args.next().expect("flag -X requires a value"),
+            "-d" => data = args.next().expect("flag -d requires a value").into(),
+            "--wait" => {
+                let wait_string: String = args.next().expect("flag --wait requires a value").into();
+                let wait_seconds: u64 = wait_string.parse().expect("invalid --wait flag");
+                wait = Duration::from_secs(wait_seconds);
+            }
             "-k" => {
                 let mut client_config = ClientConfig::new();
                 client_config
@@ -138,14 +164,22 @@ Fetch url and copy it to stdout.
                     .set_certificate_verifier(Arc::new(AcceptAll {}));
                 builder = builder.tls_config(Arc::new(client_config));
             }
-            f => Err(StringError(format!("unrecognized flag '{}'", f)))?,
+            arg => {
+                if arg.starts_with("-") {
+                    Err(StringError(format!("unrecognized flag '{}'", arg)))?;
+                }
+                nonflags.push(arg.to_owned());
+            }
         }
     }
 
     let agent = builder.build();
 
-    for url in nonflags {
-        get(&agent, &url, print_headers)?;
+    for (i, url) in nonflags.iter().enumerate() {
+        perform(&agent, &method, &url, &data, print_headers)?;
+        if i != nonflags.len() - 1 {
+            thread::sleep(wait);
+        }
     }
     Ok(())
 }
diff --git a/src/stream.rs b/src/stream.rs
index 5ed48ca..7773fba 100644
--- a/src/stream.rs
+++ b/src/stream.rs
@@ -148,7 +148,15 @@ impl Stream {
 
     // Check if the server has closed a stream by performing a one-byte
     // non-blocking read. If this returns EOF, the server has closed the
-    // connection: return true. If this returns WouldBlock (aka EAGAIN),
+    // connection: return true. If this returns a successful read, there are
+    // some bytes on the connection even though there was no inflight request.
+    // For plain HTTP streams, that might mean an HTTP 408 was pushed; it
+    // could also mean a buggy server that sent more bytes than a response's
+    // Content-Length. For HTTPS streams, that might mean a close_notify alert,
+    // which is the proper way to shut down an idle stream.
+    // Either way, bytes available on the stream before we've made a request
+    // means the stream is not usable, so we should discard it.
+    // If this returns WouldBlock (aka EAGAIN),
     // that means the connection is still open: return false. Otherwise
     // return an error.
     fn serverclosed_stream(stream: &std::net::TcpStream) -> io::Result<bool> {
@@ -156,8 +164,13 @@ impl Stream {
         stream.set_nonblocking(true)?;
 
         let result = match stream.peek(&mut buf) {
-            Ok(0) => Ok(true),
-            Ok(_) => Ok(false), // TODO: Maybe this should produce an "unexpected response" error
+            Ok(n) => {
+                debug!(
+                    "peek on reused connection returned {}, not WouldBlock; discarding",
+                    n
+                );
+                Ok(true)
+            }
             Err(e) if e.kind() == io::ErrorKind::WouldBlock => Ok(false),
             Err(e) => Err(e),
         };


Test patch:
diff --git a/src/test/agent_test.rs b/src/test/agent_test.rs
index 36edeaa..b7d254c 100644
--- a/src/test/agent_test.rs
+++ b/src/test/agent_test.rs
@@ -4,6 +4,7 @@ use crate::error::Error;
 use crate::testserver::{read_request, TestServer};
 use std::io::{self, Read, Write};
 use std::net::TcpStream;
+use std::thread;
 use std::time::Duration;
 
 use super::super::*;
@@ -17,6 +18,18 @@ fn idle_timeout_handler(mut stream: TcpStream) -> io::Result<()> {
     Ok(())
 }
 
+// Handler that answers with a simple HTTP response, and times
+// out idle connections after 2 seconds, sending an HTTP 408 response
+fn idle_timeout_handler_408(mut stream: TcpStream) -> io::Result<()> {
+    read_request(&stream);
+    stream.write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 8\r\n\r\nresponse")?;
+    let twosec = Duration::from_secs(2);
+    stream.set_read_timeout(Some(twosec))?;
+    thread::sleep(twosec);
+    stream.write_all(b"HTTP/1.1 408 Request Timeout\r\nContent-Length: 7\r\n\r\ntimeout")?;
+    Ok(())
+}
+
 #[test]
 fn connection_reuse() {
     let testserver = TestServer::new(idle_timeout_handler);
@@ -44,6 +57,33 @@ fn connection_reuse() {
     assert_eq!(resp.status(), 200);
 }
 
+#[test]
+fn connection_reuse_with_408() {
+    let testserver = TestServer::new(idle_timeout_handler_408);
+    let url = format!("http://localhost:{}", testserver.port);
+    let agent = Agent::new();
+    let resp = agent.get(&url).call().unwrap();
+
+    // use up the connection so it gets returned to the pool
+    assert_eq!(resp.status(), 200);
+    resp.into_string().unwrap();
+
+    assert!(agent.state.pool.len() > 0);
+
+    // wait for the server to close the connection.
+    std::thread::sleep(Duration::from_secs(3));
+
+    // try and make a new request on the pool. this fails
+    // when we discover that the TLS connection is dead
+    // first when attempting to read from it.
+    // Note: This test assumes the second  .call() actually
+    // pulls from the pool. If for some reason the timed-out
+    // connection wasn't in the pool, we won't be testing what
+    // we thought we were testing.
+    let resp = agent.post(&url).send_string("hello".into()).unwrap();
+    assert_eq!(resp.status(), 200);
+}
+
 #[test]
 fn custom_resolver() {
     use std::io::Read;