fix(retry): handle hyper::Error(IncompleteMessage) as Transient

try-to-manage-hyper-incomplete-message
Alessandro Caprarelli 2021-10-13 17:01:25 +02:00
parent 6a3c266645
commit 7b1fad447d
4 changed files with 95 additions and 51 deletions

View File

@ -28,3 +28,4 @@ task-local-extensions = "0.1.1"
wiremock = "0.5" wiremock = "0.5"
tokio = { version = "1", features = ["macros"] } tokio = { version = "1", features = ["macros"] }
paste = "1" paste = "1"
async-std = { version = "1.10"}

View File

@ -45,19 +45,20 @@ impl Retryable {
Some(Retryable::Transient) Some(Retryable::Transient)
} else if error.is_body() } else if error.is_body()
|| error.is_decode() || error.is_decode()
|| error.is_request()
|| error.is_builder() || error.is_builder()
|| error.is_redirect() || error.is_redirect()
{ {
Some(Retryable::Fatal) Some(Retryable::Fatal)
} else if let Some(hyper_error) = get_source_error_type::<hyper::Error>(&error) } else if error.is_request() {
{ if let Some(hyper_error) = get_source_error_type::<hyper::Error>(&error) {
if hyper_error.is_incomplete_message() { if hyper_error.is_incomplete_message() {
Some(Retryable::Fatal)
} else {
Some(Retryable::Transient) Some(Retryable::Transient)
} else {
Some(Retryable::Fatal)
}
} else {
Some(Retryable::Fatal)
} }
// TODO: map all the hyper_error types
} else { } else {
// We omit checking if error.is_status() since we check that already. // We omit checking if error.is_status() since we check that already.
// However, if Response::error_for_status is used the status will still // However, if Response::error_for_status is used the status will still

View File

@ -1,19 +1,17 @@
use async_std::io::ReadExt;
use async_std::io::WriteExt;
use async_std::net::{TcpListener, TcpStream};
use futures::stream::StreamExt;
use std::error::Error; use std::error::Error;
use std::io::{Read, Write}; use std::fmt;
use std::net::TcpStream;
use std::path::Path;
use std::{
error::Error,
fmt,
net::{TcpListener, TcpStream},
};
use std::{fmt, fs};
/// This is a simple server that returns the responses given at creation time: [`self.raw_http_responses`] following a round-robin mechanism.
pub struct SimpleServer { pub struct SimpleServer {
listener: TcpListener, listener: TcpListener,
port: u16, port: u16,
host: String, host: String,
raw_http_response: String, raw_http_responses: Vec<String>,
calls_counter: usize,
} }
/// Request-Line = Method SP Request-URI SP HTTP-Version CRLF /// Request-Line = Method SP Request-URI SP HTTP-Version CRLF
@ -29,14 +27,16 @@ impl<'a> fmt::Display for Request<'a> {
} }
} }
/// This is a simple server that
impl SimpleServer { impl SimpleServer {
pub fn new( /// Creates an instance of a [`SimpleServer`]
/// If [`port`] is None os Some(0), it gets randomly chosen between the available ones.
pub async fn new(
host: &str, host: &str,
port: Option<u16>, port: Option<u16>,
raw_http_response: &str, raw_http_responses: Vec<String>,
) -> Result<Self, anyhow::Error> { ) -> Result<Self, anyhow::Error> {
let listener = TcpListener::bind(format!("{}:{}", host, port.ok_or(0).unwrap()))?; let port = port.unwrap_or(0);
let listener = TcpListener::bind(format!("{}:{}", host, port)).await?;
let port = listener.local_addr()?.port(); let port = listener.local_addr()?.port();
@ -44,57 +44,70 @@ impl SimpleServer {
listener, listener,
port, port,
host: host.to_string(), host: host.to_string(),
raw_http_response: raw_http_response.to_string(), raw_http_responses,
calls_counter: 0,
}) })
} }
/// Returns the uri in which the server is listening to.
pub fn uri(&self) -> String { pub fn uri(&self) -> String {
format!("http://{}:{}", self.host, self.port) format!("http://{}:{}", self.host, self.port)
} }
pub fn start(&self) { /// Starts the TcpListener and handles the requests.
for stream in self.listener.incoming() { pub async fn start(mut self) -> () {
while let Some(stream) = self.listener.incoming().next().await {
match stream { match stream {
Ok(stream) => { Ok(stream) => {
match self.handle_connection(stream, self.raw_http_response.clone()) { match self.handle_connection(stream).await {
Ok(_) => (), Ok(_) => (),
Err(e) => println!("Error handling connection: {}", e), Err(e) => {
println!("Error handling connection: {}", e);
()
} }
} }
Err(e) => println!("Connection failed: {}", e), self.calls_counter = self.calls_counter + 1;
}
Err(e) => {
println!("Connection failed: {}", e);
()
}
} }
} }
} }
fn handle_connection( /// Asyncrounously reads from the buffer and handle the request.
&self, /// It first checks that the format is correct, then returns the response.
mut stream: TcpStream, ///
raw_http_response: String, /// Returns a 400 if the request if formatted badly.
) -> Result<(), Box<dyn Error>> { async fn handle_connection(&self, mut stream: TcpStream) -> Result<(), Box<dyn Error>> {
// 512 bytes is enough for a toy HTTP server let mut buffer = vec![0; 1024];
let mut buffer = [0; 512];
// writes stream into buffer stream.read(&mut buffer).await.unwrap();
stream.read(&mut buffer).unwrap();
let request = String::from_utf8_lossy(&buffer[..]); let request = String::from_utf8_lossy(&buffer[..]);
let request_line = request.lines().next().unwrap(); let request_line = request.lines().next().unwrap();
match Self::parse_request_line(&request_line) { let response = match Self::parse_request_line(&request_line) {
Ok(request) => { Ok(request) => {
println!("Request: {}", &request); println!("== Request == \n{}\n=============", request);
self.get_response().clone()
let response = format!("{}", raw_http_response.clone());
stream.write(response.as_bytes()).unwrap();
stream.flush().unwrap();
} }
Err(e) => print!("Bad request: {}", e), Err(e) => {
println!("++ Bad request: {} ++++++", e);
self.get_bad_request_response()
} }
};
println!("-- Response --\n{}\n--------------", response.clone());
stream.write(response.as_bytes()).await.unwrap();
stream.flush().await.unwrap();
Ok(()) Ok(())
} }
/// Parses the request line and checks that it contains the method, uri and http_version parts.
/// It does not check if the content of the checked parts is correct. It just checks the format (it contains enough parts) of the request.
fn parse_request_line(request: &str) -> Result<Request, Box<dyn Error>> { fn parse_request_line(request: &str) -> Result<Request, Box<dyn Error>> {
let mut parts = request.split_whitespace(); let mut parts = request.split_whitespace();
@ -110,4 +123,20 @@ impl SimpleServer {
http_version, http_version,
}) })
} }
/// Returns the response to use based on the calls counter.
/// It uses a round-robin mechanism.
fn get_response(&self) -> String {
let index = if self.calls_counter >= self.raw_http_responses.len() {
self.raw_http_responses.len() % self.calls_counter
} else {
self.calls_counter
};
self.raw_http_responses[index].clone()
}
/// Returns the raw HTTP response in case of a 400 Bad Request.
fn get_bad_request_response(&self) -> String {
"HTTP/1.1 400 Bad Request\r\n\r\n".to_string()
}
} }

View File

@ -11,7 +11,6 @@ use wiremock::matchers::{method, path};
use wiremock::{Mock, MockServer, Respond, ResponseTemplate}; use wiremock::{Mock, MockServer, Respond, ResponseTemplate};
use crate::helpers::SimpleServer; use crate::helpers::SimpleServer;
pub struct RetryResponder(Arc<AtomicU32>, u32, u16); pub struct RetryResponder(Arc<AtomicU32>, u32, u16);
impl RetryResponder { impl RetryResponder {
@ -218,10 +217,24 @@ async fn assert_retry_on_request_timeout() {
#[tokio::test] #[tokio::test]
async fn assert_retry_on_incomplete_message() { async fn assert_retry_on_incomplete_message() {
let raw_response = "HTTP/1.1 200"; // the full working response is: "HTTP/1.1 200 OK\r\n\r\n" let incomplete_message = "HTTP/1.1";
let complete_message = "HTTP/1.1 200 OK\r\n\r\n";
let simple_server = SimpleServer::new("127.0.0.1", None, raw_response) let simple_server = SimpleServer::new(
"127.0.0.1",
None,
vec![
incomplete_message.to_string(),
incomplete_message.to_string(),
incomplete_message.to_string(),
complete_message.to_string(),
],
)
.await
.expect("Error when creating a simple server"); .expect("Error when creating a simple server");
let uri = simple_server.uri();
tokio::spawn(simple_server.start()); tokio::spawn(simple_server.start());
let reqwest_client = Client::builder().build().unwrap(); let reqwest_client = Client::builder().build().unwrap();
@ -237,8 +250,8 @@ async fn assert_retry_on_incomplete_message() {
.build(); .build();
let resp = client let resp = client
.get(&format!("{}/foo", simple_server.uri())) .get(&format!("{}/foo", uri))
.timeout(std::time::Duration::from_millis(10)) .timeout(std::time::Duration::from_millis(100))
.send() .send()
.await .await
.expect("call failed"); .expect("call failed");