diff --git a/Cargo.toml b/Cargo.toml index 90ad051..1539ebf 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "prosa-hyper" -version = "0.3.1" +version = "0.4.0" authors = ["Jérémy HERGAULT ", "Anthony THOMAS ", "Julien TERUEL ", "Rene-Louis EYMARD "] description = "ProSA Hyper processor for HTTP client/server" homepage = "https://worldline.com/" @@ -39,20 +39,17 @@ bytes = ">=1.11.1, < 2" thiserror = "2" serde = { version = "1", features = ["derive"] } tokio = { version = ">=1.48, < 2", features = ["macros", "net", "rt", "rt-multi-thread"] } -tracing = "0.1" -prosa = "0.4" -aquamarine = "0.6" +prosa = { version = "0.5", git = "https://github.com/worldline/ProSA.git", branch = "reload_improve" } +simple-mermaid = "0.2" url = { version = "2", features = ["serde"] } -opentelemetry = { version = "0.31", features = ["metrics", "trace", "logs"] } - hyper = { version = "1", features = ["full"] } http = "1" http-body-util = "0.1" hyper-util = { version = "0.1", features = ["full"] } [dev-dependencies] -prosa-utils = "0.4" +prosa-utils = { version = "0.5", git = "https://github.com/worldline/ProSA.git", branch = "reload_improve" } openssl = ">=0.10.75, < 0.11" reqwest = { version = "0.13" } config = { version = "0.15", default-features = false, features = ["toml", "json", "yaml", "json5", "convert-case", "async"] } diff --git a/README.md b/README.md index 3de689a..3bb1ab9 100644 --- a/README.md +++ b/README.md @@ -25,7 +25,7 @@ The server configuration is straightforward. You only need to set a [ListenerSetting](https://docs.rs/prosa/latest/prosa/io/listener/struct.ListenerSetting.html) to configure. ```yaml -http_server: +hyper_server: listener: url: https://0.0.0.0:443 ssl: @@ -38,6 +38,9 @@ http_server: If you have some slow services, you can set the `service_timeout` parameter (800 ms by default). +When ProSA stops, the processor releases its listening port right away so a new client is refused +instead of waiting, then answers the requests it still has in flight before shutting down. + ### Client The client exposes a service if it is available. @@ -45,16 +48,23 @@ For backends, you need to use [TargetSetting](https://docs.rs/prosa/latest/prosa All backends will be load-balanced with ProSA's internal service load balancing. ```yaml -http_client: +hyper_client: service_name: "service_name" - min_socket: 1 - max_socket: 20 + nb_socket: 1 backends: - url: http://backend01:8080 ``` +`nb_socket` is how many connections the processor keeps open to each backend (one by default). +An HTTP/1.1 socket serves one request at a time, so it is worth raising against a plain backend; +an HTTP/2 one multiplexes them. + If you have a slow backend response, you can set the `http_timeout` parameter (5 seconds by default). +A socket that can't reach its backend is retried instead of being dropped. It waits `reconnect_delay` +(500 ms by default), doubling that delay on every consecutive failure, up to `max_reconnect_delay` +(30 seconds by default). + ## Examples ### Server @@ -70,8 +80,8 @@ cargo run --example server ``` The server provides the following targets: - - [/](http://localhost:8080/) returns the ProSA name - - [/test](http://localhost:8080/test) contacts an internal service named SRV_TEST (requires starting the stub processor) + - [/](https://localhost:8443/) returns the ProSA name + - [/test](https://localhost:8443/test) contacts an internal service named SRV_TEST (requires starting the stub processor) - [metrics](http://localhost:9090/metrics) exposes Prometheus metrics as configured ### Client diff --git a/deny.toml b/deny.toml index 1e7295a..94d8294 100644 --- a/deny.toml +++ b/deny.toml @@ -11,6 +11,7 @@ allow = [ "MIT", "BSD-2-Clause", "BSD-3-Clause", + "CC0-1.0", "CDLA-Permissive-2.0", "ISC", "Zlib", diff --git a/examples/client.rs b/examples/client.rs index a622426..3096862 100644 --- a/examples/client.rs +++ b/examples/client.rs @@ -1,28 +1,34 @@ -use std::convert::Infallible; -use std::env; +use std::{convert::Infallible, env}; use bytes::Bytes; use clap::{ArgAction, Command, arg}; use config::Config; -use http_body_util::Empty; -use http_body_util::combinators::BoxBody; +use http_body_util::{Empty, combinators::BoxBody}; use hyper::{Request, Response}; -use prosa::core::adaptor::Adaptor; -use prosa::core::error::ProcError; -use prosa::core::main::MainProc; -use prosa::core::main::MainRunnable as _; -use prosa::core::proc::{Proc, ProcBusParam, ProcConfig}; -use prosa::core::service::ServiceError; -use prosa::core::settings::settings; -use prosa::inj::adaptor::InjAdaptor; -use prosa::inj::proc::{InjProc, InjSettings}; -use prosa_hyper::PRODUCT_VERSION_HEADER; -use prosa_hyper::client::adaptor::HyperClientAdaptor; -use prosa_hyper::client::proc::{HyperClientProc, HyperClientSettings}; -use prosa_utils::config::tracing::TelemetryFilter; -use prosa_utils::msg::simple_string_tvf::SimpleStringTvf; +use prosa::{ + core::{ + adaptor::Adaptor, + error::ProcError, + main::{MainProc, MainRunnable as _}, + proc::{Proc, ProcBusParam, ProcConfig}, + service::ServiceError, + settings::settings, + }, + inj::{ + adaptor::InjAdaptor, + proc::{InjProc, InjSettings}, + }, + tracing::debug, +}; +use prosa_hyper::{ + PRODUCT_VERSION_HEADER, + client::{ + adaptor::HyperClientAdaptor, + proc::{HyperClientProc, HyperClientSettings}, + }, +}; +use prosa_utils::{config::tracing::TelemetryFilter, msg::simple_string_tvf::SimpleStringTvf}; use serde::{Deserialize, Serialize}; -use tracing::debug; use url::Url; /// Demo Hyper processor adaptor diff --git a/examples/config.yml b/examples/config.yml index d191579..278ba3b 100644 --- a/examples/config.yml +++ b/examples/config.yml @@ -13,8 +13,7 @@ hyper_client: ssl: store: path: "examples/" - min_socket: 2 - max_socket: 20 + nb_socket: 2 observability: level: DEBUG diff --git a/examples/server.rs b/examples/server.rs index cf3ab1d..393786b 100644 --- a/examples/server.rs +++ b/examples/server.rs @@ -1,28 +1,34 @@ -use std::borrow::Cow; - -use std::env; +use std::{borrow::Cow, env}; use bytes::Bytes; use clap::{ArgAction, Command, arg}; use config::Config; -use http_body_util::Full; -use http_body_util::combinators::BoxBody; +use http_body_util::{Full, combinators::BoxBody}; use hyper::{Request, Response, StatusCode}; -use prosa::core::adaptor::Adaptor; -use prosa::core::error::ProcError; -use prosa::core::main::MainRunnable as _; -use prosa::core::proc::{Proc, ProcBusParam as _, ProcConfig}; -use prosa::core::settings::settings; -use prosa::stub::adaptor::StubParotAdaptor; -use prosa::stub::proc::StubSettings; -use prosa::{core::main::MainProc, stub::proc::StubProc}; -use prosa_hyper::server::adaptor::{HyperServerAdaptor, default_srv_error_response}; -use prosa_hyper::server::proc::{HyperServerProc, HyperServerSettings}; -use prosa_hyper::{HyperResp, PRODUCT_VERSION_HEADER}; -use prosa_utils::config::tracing::TelemetryFilter; -use prosa_utils::msg::simple_string_tvf::SimpleStringTvf; +use prosa::{ + core::{ + adaptor::Adaptor, + error::ProcError, + main::{MainProc, MainRunnable as _}, + proc::{Proc, ProcBusParam as _, ProcConfig}, + settings::settings, + }, + io::SocketAddr, + stub::{ + adaptor::StubParotAdaptor, + proc::{StubProc, StubSettings}, + }, + tracing::debug, +}; +use prosa_hyper::{ + HyperResp, PRODUCT_VERSION_HEADER, + server::{ + adaptor::{HyperServerAdaptor, default_srv_error_response}, + proc::{HyperServerProc, HyperServerSettings}, + }, +}; +use prosa_utils::{config::tracing::TelemetryFilter, msg::simple_string_tvf::SimpleStringTvf}; use serde::{Deserialize, Serialize}; -use tracing::debug; /// Demo Hyper processor adaptor #[derive(Debug, Adaptor, Clone)] @@ -41,7 +47,10 @@ where + prosa_utils::msg::tvf::Tvf + std::default::Default, { - fn new(proc: &HyperServerProc) -> Result> { + fn new( + proc: &HyperServerProc, + _addr: SocketAddr, + ) -> Result> { Ok(HyperDemoAdaptor { prosa_name: proc.name().to_string(), }) diff --git a/src/client/adaptor.rs b/src/client/adaptor.rs index 887f857..1616438 100644 --- a/src/client/adaptor.rs +++ b/src/client/adaptor.rs @@ -8,17 +8,9 @@ use url::Url; use crate::client::proc::HyperClientProc; -#[cfg_attr(doc, aquamarine::aquamarine)] /// Trait to define the Hyper adaptor structure /// -/// ```mermaid -/// graph LR -/// OUT1[Output HTTP server] -/// ProSA[ProSA Hyper Procesor] -/// -/// ProSA-- HTTP request (process_client_request) -->OUT -/// OUT-- HTTP response (process_client_response) -->ProSA -/// ``` +#[doc = simple_mermaid::mermaid!("diagrams/adaptor.mmd")] pub trait HyperClientAdaptor where M: 'static diff --git a/src/client/diagrams/adaptor.mmd b/src/client/diagrams/adaptor.mmd new file mode 100644 index 0000000..33539f0 --- /dev/null +++ b/src/client/diagrams/adaptor.mmd @@ -0,0 +1,6 @@ +graph LR + OUT[Output HTTP server] + ProSA[ProSA Hyper Processor] + + ProSA-- HTTP request (process_srv_request) -->OUT + OUT-- HTTP response (process_http_response) -->ProSA diff --git a/src/client/proc.rs b/src/client/proc.rs index 70b94d2..1cac07c 100644 --- a/src/client/proc.rs +++ b/src/client/proc.rs @@ -1,22 +1,27 @@ -use std::sync::Arc; +use std::{sync::Arc, time::Duration}; -use opentelemetry::KeyValue; use prosa::{ core::{ adaptor::Adaptor, error::ProcError, msg::InternalMsg, - proc::{Proc, ProcBusParam as _, ProcConfig as _, proc, proc_settings}, + proc::{Proc, ProcBusParam as _, ProcConfig as _, ProcParam, proc, proc_settings}, }, io::stream::TargetSetting, + tracing::{debug, info, warn}, }; use serde::{Deserialize, Serialize}; -use tokio::task::JoinSet; -use tracing::{info, warn}; +use tokio::{ + sync::{mpsc, watch}, + task::{self, JoinError, JoinSet}, +}; use crate::{ HyperProcError, - client::{adaptor::HyperClientAdaptor, socket::HyperClientSocket}, + client::{ + adaptor::HyperClientAdaptor, + socket::{HyperClientSocket, SocketControl, SocketMeters}, + }, }; /// Hyper client processor settings @@ -27,30 +32,38 @@ pub struct HyperClientSettings { pub service_name: String, /// List of backend services pub backends: Vec, - /// Minimum number of socket connections per target - #[serde(default = "HyperClientSettings::default_min_socket")] - min_socket: u32, - /// Maximum number of socket connections per target - #[serde(default = "HyperClientSettings::default_max_socket")] - max_socket: u32, + /// Number of socket connections per target, at least one + #[serde(default = "HyperClientSettings::default_nb_socket")] + nb_socket: u32, /// Timeout for HTTP messages in milliseconds #[serde(default = "HyperClientSettings::default_http_timeout")] http_timeout: u64, + /// Delay before reconnecting a socket that just failed, in milliseconds. + /// It doubles on every consecutive failure, up to `max_reconnect_delay` + #[serde(default = "HyperClientSettings::default_reconnect_delay")] + reconnect_delay: u64, + /// Maximum delay between two reconnection attempts of a socket, in milliseconds + #[serde(default = "HyperClientSettings::default_max_reconnect_delay")] + max_reconnect_delay: u64, } impl HyperClientSettings { - fn default_min_socket() -> u32 { + fn default_nb_socket() -> u32 { 1 } - fn default_max_socket() -> u32 { - 20 - } - fn default_http_timeout() -> u64 { 5000 } + fn default_reconnect_delay() -> u64 { + 500 + } + + fn default_max_reconnect_delay() -> u64 { + 30000 + } + /// Create a new Hyper client settings listenning to a service pub fn new(service_name: String) -> Self { HyperClientSettings { @@ -63,6 +76,62 @@ impl HyperClientSettings { pub fn add_backend(&mut self, target: TargetSetting) { self.backends.push(target); } + + /// Number of socket connections the processor keeps open per backend. + /// + /// Never zero, a client without socket can't serve anything, so a configuration asking for none + /// is read as asking for one. Every reader goes through here, which is what keeps the processor + /// and its sockets from disagreeing on how many sockets should exist + pub(crate) fn nb_socket(&self) -> u32 { + self.nb_socket.max(1) + } + + /// Timeout applied to HTTP messages + pub(crate) fn http_timeout(&self) -> Duration { + Duration::from_millis(self.http_timeout) + } + + /// Maximum delay between two reconnection attempts of a socket, never below the floor + fn max_reconnect_delay(&self) -> Duration { + Duration::from_millis(self.max_reconnect_delay).max(MIN_RECONNECT_DELAY) + } + + /// Delay of the first reconnection attempt of a socket. + /// + /// Doubles as how long a connection that served nothing must have lasted to count as a working + /// one, so a socket never reconnects faster than it would after a refused connection. + /// + /// Floored, because a zero would do both at once: no wait between two attempts, and every + /// attempt counting as a working connection, so the retry count would never grow and a backend + /// that refuses everything would be dialled at the speed of the machine + pub(crate) fn base_reconnect_delay(&self) -> Duration { + Duration::from_millis(self.reconnect_delay) + .min(self.max_reconnect_delay()) + .max(MIN_RECONNECT_DELAY) + } + + /// Delay to wait before the `retry`-th consecutive reconnection attempt of a socket. + /// + /// `retry` is zero for a socket that never failed, which connects without waiting + pub(crate) fn reconnect_delay(&self, retry: u32) -> Duration { + if retry == 0 { + return Duration::ZERO; + } + + self.base_reconnect_delay() + .saturating_mul(1u32.checked_shl(retry - 1).unwrap_or(u32::MAX)) + .min(self.max_reconnect_delay()) + } + + /// Negotiate HTTP/2 first on every SSL backend, so a backend compares equal to the target of a + /// socket that is already running it. + /// + /// [`TargetSetting::set_alpn`] is idempotent and does nothing on a plain backend + pub(crate) fn normalize_backends(&mut self) { + for backend in &mut self.backends { + backend.set_alpn(vec!["h2".into(), "http/1.1".into()]); + } + } } #[proc_settings] @@ -71,14 +140,285 @@ impl Default for HyperClientSettings { HyperClientSettings { service_name: "hyper".to_string(), backends: Vec::new(), - min_socket: Self::default_min_socket(), - max_socket: Self::default_max_socket(), + nb_socket: Self::default_nb_socket(), http_timeout: Self::default_http_timeout(), + reconnect_delay: Self::default_reconnect_delay(), + max_reconnect_delay: Self::default_max_reconnect_delay(), } } } -/// Hyper server processor +/// Depth of the bus queue of a socket, as many messages as the processor's own queue +const SOCKET_QUEUE_SIZE: usize = 2048; + +/// Shortest a socket ever waits between two connection attempts. +/// +/// Short enough that a configured delay is used as written, and that the doubling still reaches +/// `max_reconnect_delay` in a dozen attempts, but not zero +const MIN_RECONNECT_DELAY: Duration = Duration::from_millis(10); + +/// One socket the processor wants to keep open to a backend +#[derive(Debug)] +struct SocketHandle { + /// Bus queue id of the socket, stable across its reconnections + id: u32, + /// Backend the socket connects to, and its position in the pool of that backend + target: TargetSetting, + index: u32, + /// Service the socket advertises, so renaming it replaces the whole pool + service_name: String, + /// What the processor decides for the socket, out of band from the requests it serves + control: watch::Sender, + /// Task currently running the socket, which is how a slot is found again from a join error + task: task::Id, +} + +impl SocketHandle { + /// Method to know if two targets open the same connection. + /// + /// [`TargetSetting`] compares its `connect_timeout` too, which applies to the next attempt like + /// the reconnection delays do. Comparing it here would retire every socket of a backend on a + /// reload that only shortened it, which is the opposite of what it is for + fn same_connection(target: &TargetSetting, other: &TargetSetting) -> bool { + target.url == other.url && target.ssl() == other.ssl() && target.proxy == other.proxy + } + + /// Give the backend of `settings` this slot connects to, if the settings still describe it. + /// + /// The one place that decides which sockets exist. The reconnection delays, the message timeout + /// and the connection timeout aren't part of it: they apply to the next attempt and to the next + /// request, so they reach the socket through its control channel instead + fn matching_backend<'s>(&self, settings: &'s HyperClientSettings) -> Option<&'s TargetSetting> { + if self.index >= settings.nb_socket() || self.service_name != settings.service_name { + return None; + } + + settings + .backends + .iter() + .find(|backend| Self::same_connection(backend, &self.target)) + } +} + +/// The sockets the processor keeps open to its backends, and the tasks running them +struct SocketPool +where + M: Sized + Clone + prosa::core::msg::Tvf, +{ + /// Running socket tasks. A task ends when its connection does, and the socket it gives back is + /// started again for as long as its slot is still in [`Self::handles`] + tasks: JoinSet<(HyperClientSocket, Option)>, + /// One per socket that should exist, which is the only thing saying a socket must come back + handles: Vec, + /// Slot ids are never reused, so a queue the bus hasn't removed yet can't be confused with the + /// one of the socket that takes its place + next_id: u32, + /// Instruments shared by every socket + meters: SocketMeters, +} + +impl SocketPool +where + M: 'static + + std::marker::Send + + std::marker::Sync + + std::marker::Sized + + std::clone::Clone + + std::fmt::Debug + + prosa::core::msg::Tvf + + std::default::Default, +{ + /// Create an empty pool, which [`Self::align`] then fills from the settings + fn new(meters: SocketMeters) -> Self { + SocketPool { + tasks: JoinSet::new(), + handles: Vec::new(), + next_id: 1, + meters, + } + } + + /// Answer `true` while the pool still runs a socket task + fn is_running(&self) -> bool { + !self.tasks.is_empty() + } + + /// Wait for the next socket task to end and hand its socket back + #[allow(clippy::type_complexity)] + async fn join_next( + &mut self, + ) -> Option, Option), JoinError>> { + self.tasks.join_next().await + } + + /// Bring the pool in line with `settings`. + /// + /// [`Self::handles`] holds exactly the sockets that should exist, so retiring one is dropping + /// its handle after telling it to stop, and spawning one is opening its channels. Idempotent, + /// which is what lets the same call serve the startup and the configuration reload + fn align( + &mut self, + settings: &HyperClientSettings, + proc: &Arc>, + adaptor: &Arc, + ) where + A: 'static + Adaptor + HyperClientAdaptor + std::marker::Send + std::marker::Sync, + { + // Retire the sockets the settings don't describe anymore. They answer what they have in + // flight and come back through the task set, where the processor finds no handle for them + self.handles.retain(|handle| { + if let Some(backend) = handle.matching_backend(settings) { + handle.control.send_modify(|control| { + control.connect_timeout = backend.connect_timeout; + control.http_timeout = settings.http_timeout(); + }); + true + } else { + debug!( + "Retire the Hyper client socket {} on {}", + handle.id, + handle.target.get_safe_url() + ); + handle.control.send_modify(|control| control.stopped = true); + false + } + }); + + // And spawn the ones that are missing. Matched the same way the sockets were kept, so a + // backend that only changed its connection timeout doesn't get a second pool + for target in &settings.backends { + for index in 0..settings.nb_socket() { + if !self.handles.iter().any(|handle| { + handle.index == index && SocketHandle::same_connection(&handle.target, target) + }) { + self.spawn(target.clone(), index, settings, proc, adaptor); + } + } + } + } + + /// Open a slot on `target` and start the socket that serves it + fn spawn( + &mut self, + target: TargetSetting, + index: u32, + settings: &HyperClientSettings, + proc: &Arc>, + adaptor: &Arc, + ) where + A: 'static + Adaptor + HyperClientAdaptor + std::marker::Send + std::marker::Sync, + { + let (tx_queue, rx_queue) = mpsc::channel(SOCKET_QUEUE_SIZE); + let (control_tx, control_rx) = + SocketControl::channel(target.connect_timeout, settings.http_timeout()); + let id = self.next_id; + self.next_id += 1; + + let task = HyperClientSocket::new( + id, + target.clone(), + settings.service_name.clone(), + control_rx, + rx_queue, + tx_queue, + ) + .spawn( + &mut self.tasks, + proc.clone(), + adaptor.clone(), + settings, + self.meters.clone(), + ); + + self.handles.push(SocketHandle { + id, + target, + index, + service_name: settings.service_name.clone(), + control: control_tx, + task, + }); + } + + /// Take back a socket task that just ended, and start its slot again if it is still open + fn take_back( + &mut self, + ended: Result<(HyperClientSocket, Option), JoinError>, + settings: &HyperClientSettings, + proc: &Arc>, + adaptor: &Arc, + ) where + A: 'static + Adaptor + HyperClientAdaptor + std::marker::Send + std::marker::Sync, + { + let socket = match ended { + Ok((socket, None)) => socket, + Ok((socket, Some(error))) => { + warn!("A Hyper client socket task ended with error: {error}"); + socket + } + // A task that panicked can't hand its socket back, and the queue it held went with it. + // Taking the processor down over it would drop the queues of every other socket too, + // so the slot is closed here and opened again below, with a socket of its own + Err(error) => { + warn!("A Hyper client socket task panicked, closing its slot: {error}"); + self.handles.retain(|handle| handle.task != error.id()); + // Opened again unless ProSA is on its way out, where `handles` is already empty and + // realigning would connect the whole pool back to the backends + if !proc.is_stopping() { + self.align(settings, proc, adaptor); + } + return; + } + }; + + // The bus tells the sockets and the processor to stop at the same time, so a socket can end + // before the processor has read its own `Shutdown`. Restarting it then would open a + // connection to the backend in the middle of the shutdown, so the stop flag is asked first: + // ProSA raises it before it sends any of those messages + let slot = self + .handles + .iter() + .position(|handle| handle.id == socket.id()); + if let Some(slot) = slot + && !proc.is_stopping() + { + debug!("A Hyper client socket has ended, restarting a new one: {socket:?}"); + self.handles[slot].task = socket.spawn( + &mut self.tasks, + proc.clone(), + adaptor.clone(), + settings, + self.meters.clone(), + ); + } else { + debug!("A Hyper client socket has been retired: {socket:?}"); + socket.retire(); + } + } + + /// Tell every socket to stop and wait for them to answer what they have in flight. + /// + /// The main task only reaches the sockets that are connected. Telling them here reaches the + /// ones that are only waiting to reconnect too, and waiting rather than dropping the task set + /// is what keeps a socket from being aborted with a request it never answered. Everything a + /// retired socket still does is bounded by the message timeout, so this cannot outlast one, and + /// bounding it again here would only cut a drain short of the budget it was given + async fn shutdown(&mut self) { + for handle in &self.handles { + handle.control.send_modify(|control| control.stopped = true); + } + self.handles.clear(); + + while let Some(socket) = self.tasks.join_next().await { + match socket { + Ok((socket, _)) => socket.retire(), + Err(error) => warn!("A Hyper client socket task panicked while stopping: {error}"), + } + } + } +} + +/// Hyper client processor #[proc(settings = HyperClientSettings)] pub struct HyperClientProc {} @@ -103,49 +443,20 @@ where // Add proc main queue (id: 0) self.proc.add_proc().await?; - // List of client sockets tasks - let mut client_sockets = JoinSet::new(); - - // Meter to log HTTP requests - let meter = self.get_proc_param().meter("hyper_client"); - let observable_http_histogram = meter - .u64_histogram("prosa_hyper_cli_duration") - .with_description("Hyper HTTP client request duration histogram") - .build(); - let observable_http_socket = meter - .u64_gauge("prosa_hyper_cli_socket") - .with_description("Hyper HTTP client socket counter") - .build(); + // The sockets the processor keeps open, with the meters they all report to + let mut pool = SocketPool::new(SocketMeters::new( + &self.get_proc_param().meter("hyper_client"), + )); // Create client sockets - if !self.settings.backends.is_empty() { - for backend in &self.settings.backends { - for _ in 0..self.settings.min_socket { - let client_socket = - HyperClientSocket::new(backend.clone(), self.settings.http_timeout); - client_socket.spawn( - &mut client_sockets, - self.proc.clone(), - adaptor.clone(), - self.settings.service_name.clone(), - observable_http_histogram.clone(), - ); - } - } - } else { + if self.settings.backends.is_empty() { return Err(Box::new(HyperProcError::Other( "No backend configured for the Hyper client processor".to_string(), ))); } - // Update socket number after all creations - observable_http_socket.record( - client_sockets.len() as u64, - &[ - KeyValue::new("proc", self.name().to_string()), - KeyValue::new("service", self.settings.service_name.clone()), - ], - ); + self.settings.normalize_backends(); + pool.align(&self.settings, &self.proc, &adaptor); loop { tokio::select! { @@ -166,10 +477,31 @@ where self.get_proc_id(), err_msg ), - InternalMsg::Command(_) => todo!(), - InternalMsg::Config => todo!(), InternalMsg::Service(table) => self.service = table, + InternalMsg::Config(config) => { + // Read here and only here, so `Adaptor::reload_config` runs once. What + // came out of it reaches the sockets through their control channel, + // which gets to one that is busy serving where its bus queue wouldn't + if let Some(mut settings) = config.reload_proc::(self.proc.as_ref(), adaptor.as_ref()) { + // Keep the current configuration, a client without backend can't + // serve anything + if settings.backends.is_empty() { + warn!("Ignoring the configuration reload of {}: no backend configured", self.name()); + continue; + } + + info!("Reload the configuration of the Hyper client processor {}", self.name()); + + settings.normalize_backends(); + self.settings = settings; + pool.align(&self.settings, &self.proc, &adaptor); + } + } InternalMsg::Shutdown => { + // Wait for poll shutdown + pool.shutdown().await; + + // Terminated once the sockets are done, the requests they were finishing use it adaptor.terminate(); self.proc.remove_proc(None).await?; warn!("The Hyper client processor will shut down"); @@ -177,32 +509,124 @@ where } } }, - Some(socket) = client_sockets.join_next(), if !client_sockets.is_empty() => { - match socket.map_err(|e| HyperProcError::Other(format!("Hyper client socket task join error: {}", e)))? { - Ok(s) => { - info!("A Hyper client socket has ended, restarting a new one: {s:?}"); - s.spawn( - &mut client_sockets, - self.proc.clone(), - adaptor.clone(), - self.settings.service_name.clone(), - observable_http_histogram.clone() - ); - }, - Err(e) => { - warn!("A Hyper client socket task ended with error: {}", e); - } - } - - observable_http_socket.record( - client_sockets.len() as u64, - &[ - KeyValue::new("proc", self.name().to_string()), - KeyValue::new("service", self.settings.service_name.clone()), - ], - ); + Some(socket) = pool.join_next(), if pool.is_running() => { + // The socket is given back whatever ended it, so a failed connection is retried + // instead of shrinking the pool until the next configuration reload + pool.take_back(socket, &self.settings, &self.proc, &adaptor); }, } } } } + +#[cfg(test)] +mod tests { + use super::*; + + fn backend(url: &str) -> TargetSetting { + TargetSetting::new( + url.parse().expect("Backend URL should be valid"), + None, + None, + ) + } + + #[test] + fn only_the_connection_decides_which_sockets_exist() { + let target = backend("http://backend01:8080"); + + // A reload that only retimes the connection keeps the socket, where comparing the whole + // target would retire it + let mut retimed = target.clone(); + retimed.connect_timeout += 1000; + assert_ne!(target, retimed); + assert!(SocketHandle::same_connection(&target, &retimed)); + + // Anything that changes where or how the socket connects is a different socket + assert!(!SocketHandle::same_connection( + &target, + &backend("http://backend02:8080") + )); + + let mut proxied = target.clone(); + proxied.proxy = Some( + "http://proxy:3128" + .parse() + .expect("Proxy URL should be valid"), + ); + assert!(!SocketHandle::same_connection(&target, &proxied)); + + let mut secured = target.clone(); + secured.set_alpn(vec!["h2".into()]); + secured.set_ssl(Some(Default::default())); + assert!(!SocketHandle::same_connection(&target, &secured)); + } + + #[test] + fn reconnect_delay_backoff() { + let settings = HyperClientSettings::new("hyper".into()); + + // A socket that never failed reconnects right away + assert_eq!(Duration::ZERO, settings.reconnect_delay(0)); + + // Then the delay doubles on every consecutive failure, up to the maximum + assert_eq!(Duration::from_millis(500), settings.reconnect_delay(1)); + assert_eq!(Duration::from_millis(1000), settings.reconnect_delay(2)); + assert_eq!(Duration::from_millis(2000), settings.reconnect_delay(3)); + assert_eq!(Duration::from_secs(30), settings.reconnect_delay(7)); + + // A backend that stays down for a long time must not overflow the delay + assert_eq!(Duration::from_secs(30), settings.reconnect_delay(u32::MAX)); + } + + #[test] + fn reconnect_delay_below_first_attempt() { + // A maximum lower than the first delay caps it, instead of reconnecting slower than the + // maximum on the first attempt + let settings: HyperClientSettings = config::Config::builder() + .set_override("reconnect_delay", 5000) + .expect("Reconnect delay override should be valid") + .set_override("max_reconnect_delay", 1000) + .expect("Maximum reconnect delay override should be valid") + .set_override("service_name", "hyper") + .expect("Service name override should be valid") + .set_override("backends", Vec::::new()) + .expect("Backends override should be valid") + .build() + .expect("Configuration should be valid") + .try_deserialize() + .expect("Hyper client settings should be valid"); + + // Asserted on the base delay too: `reconnect_delay` caps against the maximum on its way + // out, so it would read the same whether the base was clamped or not + assert_eq!(Duration::from_secs(1), settings.base_reconnect_delay()); + assert_eq!(Duration::from_secs(1), settings.reconnect_delay(1)); + assert_eq!(Duration::from_secs(1), settings.reconnect_delay(9)); + } + + #[test] + fn reconnect_delay_is_never_zero() { + // A socket that waited nothing and counted every attempt as a working connection would dial + // a backend that refuses everything at the speed of the machine + let settings: HyperClientSettings = config::Config::builder() + .set_override("reconnect_delay", 0) + .expect("Reconnect delay override should be valid") + .set_override("max_reconnect_delay", 0) + .expect("Maximum reconnect delay override should be valid") + .set_override("service_name", "hyper") + .expect("Service name override should be valid") + .set_override("backends", Vec::::new()) + .expect("Backends override should be valid") + .build() + .expect("Configuration should be valid") + .try_deserialize() + .expect("Hyper client settings should be valid"); + + assert!(!settings.base_reconnect_delay().is_zero()); + assert!(!settings.reconnect_delay(1).is_zero()); + assert!(!settings.reconnect_delay(u32::MAX).is_zero()); + + // The first attempt of a socket that never failed still doesn't wait + assert_eq!(Duration::ZERO, settings.reconnect_delay(0)); + } +} diff --git a/src/client/socket.rs b/src/client/socket.rs index 753e00e..22c2fe1 100644 --- a/src/client/socket.rs +++ b/src/client/socket.rs @@ -1,19 +1,34 @@ +//! Hyper client sockets +//! +//! A socket decides nothing. The processor tells it which backend to connect to and when to stop, +//! it serves what its bus queue brings until the connection ends, then hands itself back. +//! +//! Two channels reach a socket, and they carry different things. The bus queue carries the +//! requests, and it is as deep as the traffic the socket is behind on, so anything put there is +//! only read once the socket caught up. What the processor decides — the message timeout, and +//! whether the socket should stop — goes through a [`watch`] channel instead, which is always +//! current, reaches a socket that is busy serving, and never blocks the processor's loop. + use std::{ convert::Infallible, io, + ops::ControlFlow, os::fd::AsRawFd, - sync::Arc, + sync::{ + Arc, + atomic::{AtomicBool, Ordering}, + }, time::{Duration, Instant}, }; use bytes::Bytes; use http_body_util::combinators::BoxBody; use hyper::{ - Request, + Request, Response, + body::Incoming, client::conn::{http1, http2}, }; use hyper_util::rt::{TokioExecutor, TokioIo}; -use opentelemetry::{KeyValue, metrics::Histogram}; use prosa::{ core::{ adaptor::Adaptor, @@ -21,114 +36,263 @@ use prosa::{ proc::{ProcBusParam as _, ProcParam}, service::ServiceError, }, - io::{ - SslConfig, - stream::{Stream, TargetSetting}, - url_is_ssl, + io::stream::{Stream, TargetSetting}, + otel::{ + KeyValue, + metrics::{Histogram, UpDownCounter}, }, + tracing::{debug, info, warn}, }; use tokio::{ + sync::{mpsc, watch}, task::JoinSet, time::{self, timeout}, }; -use tracing::{debug, info, warn}; -use crate::{H2, HyperProcError, client::adaptor::HyperClientAdaptor, hyper_version_str}; +use crate::{ + H2, HyperProcError, + client::{adaptor::HyperClientAdaptor, proc::HyperClientSettings}, + hyper_version_str, +}; /// Type alias for HTTP request pair to reduce type complexity type HttpRequestPair = (RequestMsg, Request>); -/// Hyper client socket +/// Instruments reported by the Hyper client sockets #[derive(Debug, Clone)] -pub struct HyperClientSocket { - /// Target of the socket - target: TargetSetting, - /// Whether the socket is using HTTP/2 - is_http2: bool, - /// HTTP message timeout duration - http_timeout: Duration, +pub(super) struct SocketMeters { + /// Duration of the HTTP messages + message_histogram: Histogram, + /// Number of connected sockets + socket_counter: UpDownCounter, } -macro_rules! close_socket { - ($self:ident, $proc:ident, $socket_id:ident, $msg_queue:ident, $service_name:ident, $return:expr) => { - $proc.remove_proc_queue($socket_id as u32).await?; - while let Ok(msg) = $msg_queue.try_recv() { - if let InternalMsg::Request(req_msg) = msg { - let _ = req_msg.return_error_to_sender( - None, - ServiceError::UnableToReachService($service_name.clone()), - ); - } +impl SocketMeters { + /// Create the instruments of the Hyper client sockets from the processor meter + pub(super) fn new(meter: &prosa::otel::metrics::Meter) -> Self { + SocketMeters { + message_histogram: meter + .u64_histogram("prosa_hyper_cli_duration") + .with_description("Hyper HTTP client request duration histogram") + .build(), + socket_counter: meter + .i64_up_down_counter("prosa_hyper_cli_socket") + .with_description("Hyper HTTP client connected socket counter") + .build(), } - return $return; - }; + } } -impl HyperClientSocket { - pub fn new(mut target: TargetSetting, http_timeout: u64) -> Self { - // Set default protocol to HTTP2 if target enabled SSL - if let Some(ssl) = target.ssl.as_mut() { - info!( - "Target {} enable SSL, set ALPN to support HTTP/2 and HTTP/1.1", - target.url - ); - ssl.set_alpn(vec!["h2".into(), "http/1.1".into()]); - } else if url_is_ssl(&target.url) { - let mut ssl = SslConfig::default(); - ssl.set_alpn(vec!["h2".into(), "http/1.1".into()]); - target.ssl = Some(ssl); - } +/// What the processor decides for one of its sockets, out of band from the requests it serves. +/// +/// Everything that applies to the next attempt or the next message, so changing it doesn't retire a +/// healthy socket. What defines the connection itself isn't here: a socket that has to reach +/// somewhere else is a different socket +#[derive(Debug)] +pub(super) struct SocketControl { + /// Timeout applied to the next connection attempt, in milliseconds + pub(super) connect_timeout: u64, + /// Timeout applied to the next HTTP message + pub(super) http_timeout: Duration, + /// Set once the processor retires the socket + pub(super) stopped: bool, +} + +impl SocketControl { + /// Open the control channel of a socket the processor is about to spawn + pub(super) fn channel( + connect_timeout: u64, + http_timeout: Duration, + ) -> (watch::Sender, watch::Receiver) { + watch::channel(SocketControl { + connect_timeout, + http_timeout, + stopped: false, + }) + } +} + +/// Report the duration of a message under the backend it was sent to. +/// +/// The target is formatted rather than taken from the URL, so a backend configured with credentials +/// doesn't put them in a metric attribute, and both protocols report under the same value +fn record_message( + message_histogram: &Histogram, + target_addr: &str, + response: &Result, hyper::Error>, + started: Instant, + default_version: &'static str, +) { + let (code, version) = response.as_ref().map_or((500, default_version), |r| { + (r.status().as_u16() as i64, hyper_version_str(r.version())) + }); + + message_histogram.record( + started.elapsed().as_millis() as u64, + &[ + KeyValue::new("target", target_addr.to_string()), + KeyValue::new("code", code), + KeyValue::new("version", version), + ], + ); +} + +/// Hyper client socket +/// +/// A socket connects to its backend, serves what its bus queue brings until the connection ends or +/// it is asked to stop, then hands itself back. Which sockets must exist is the processor's call, +/// and the way it retires one is [`SocketControl::stopped`]. +#[derive(Debug)] +pub(crate) struct HyperClientSocket +where + M: Sized + Clone + prosa::core::msg::Tvf, +{ + /// Bus queue id of the socket, stable across its reconnections + id: u32, + /// Target of the socket + target: TargetSetting, + /// Service the socket advertises on its bus queue + service_name: String, + /// Number of consecutive failed connection attempts, delaying the next reconnection + retry: u32, + /// What the processor decides for the socket, always current + control: watch::Receiver, + /// Queue the bus brings the requests to. + /// + /// Kept across reconnections, so a request that arrives while the socket is down is served by + /// its next connection instead of landing in a receiver nobody holds anymore + rx_queue: mpsc::Receiver>, + /// Sending end of [`Self::rx_queue`], declared to the bus while the socket is connected + tx_queue: mpsc::Sender>, +} +impl HyperClientSocket +where + M: 'static + + std::marker::Send + + std::marker::Sync + + std::marker::Sized + + std::clone::Clone + + std::fmt::Debug + + prosa::core::msg::Tvf + + std::default::Default, +{ + /// Create a socket on `target`, which must have been normalized by + /// [`HyperClientSettings::normalize_backends`] so it compares equal to the configured backend + pub(super) fn new( + id: u32, + target: TargetSetting, + service_name: String, + control: watch::Receiver, + rx_queue: mpsc::Receiver>, + tx_queue: mpsc::Sender>, + ) -> Self { HyperClientSocket { + id, target, - is_http2: false, - http_timeout: Duration::from_millis(http_timeout), + service_name, + retry: 0, + control, + rx_queue, + tx_queue, } } - /// Helper to setup socket queue and service registration - async fn setup_socket_queue( - proc: &Arc>, - socket_id: u32, - service_name: &str, - ) -> Result>, HyperProcError> - where - M: 'static - + std::marker::Send - + std::marker::Sync - + std::marker::Sized - + std::clone::Clone - + std::fmt::Debug - + prosa::core::msg::Tvf - + std::default::Default, - { - let (tx_queue, rx_queue) = tokio::sync::mpsc::channel(2048); - proc.add_proc_queue(tx_queue, socket_id).await?; - proc.add_service(vec![service_name.to_string()], socket_id) + /// Bus queue id of the socket, which is the slot it occupies in the pool of the processor + pub(super) fn id(&self) -> u32 { + self.id + } + + /// Timeout the processor currently applies to an HTTP message + fn http_timeout(&self) -> Duration { + self.control.borrow().http_timeout + } + + /// Answer `true` once the processor retired the socket. + /// + /// Reads without marking the value seen, so it never consumes the wake-up [`Self::stopped`] is + /// waiting for + fn is_stopped(&self) -> bool { + self.control.borrow().stopped + } + + /// Resolve once the processor retires the socket, and never otherwise. + /// + /// A dropped sender counts as retired: the processor only ever lets a handle go after it set + /// [`SocketControl::stopped`], and a closed channel makes `changed` return instantly forever, + /// which as a `select!` arm would spin the socket rather than stop it + async fn stopped(control: &mut watch::Receiver) { + while control.changed().await.is_ok() { + if control.borrow().stopped { + return; + } + } + } + + /// Answer whatever still reaches a socket that won't come back, until nobody can reach it. + /// + /// The bus is told to remove the queue, but a processor that hasn't received the new service + /// table yet still holds it and still sends to it. Closing it there turns a request that should + /// have come back `UnableToReachService` into a send error for its sender, and a processor that + /// treats that as fatal restarts on it. So the queue outlives the socket, answering rather than + /// closing, and only goes away once the last sender did. + pub(super) fn retire(self) { + let HyperClientSocket { + service_name, + mut rx_queue, + tx_queue, + .. + } = self; + + // The socket holds a sending end of its own queue, which would keep it open forever + drop(tx_queue); + + tokio::spawn(async move { + while let Some(msg) = rx_queue.recv().await { + if let InternalMsg::Request(req_msg) = msg { + let _ = req_msg.return_error_to_sender( + None, + ServiceError::UnableToReachService(service_name.clone()), + ); + } + } + }); + } + + /// Declare the socket queue and the service it serves to the bus + async fn declare_socket_queue(&self, proc: &ProcParam) -> Result<(), HyperProcError> { + proc.add_proc_queue(self.tx_queue.clone(), self.id).await?; + proc.add_service(vec![self.service_name.clone()], self.id) .await?; - Ok(rx_queue) + Ok(()) + } + + /// Take the socket queue back off the bus, which also takes the service it advertised. + /// + /// Reported rather than propagated: a bus that can't be told is no reason to abandon the + /// requests the socket still has to answer, and it is the ordinary case while ProSA stops, + /// where the main task is already gone + async fn withdraw_socket_queue(&self, proc: &ProcParam) { + if let Err(e) = proc.remove_proc_queue(self.id).await { + debug!( + socket_id = self.id, + addr = %self.target, + "Can't remove the socket queue from the bus: {e}" + ); + } } /// Helper to process a service request into an HTTP request - fn process_request( + fn process_request( + &self, adaptor: &Arc, mut msg: RequestMsg, - target_url: &url::Url, - service_name: &str, ) -> Option> where - M: 'static - + std::marker::Send - + std::marker::Sync - + std::marker::Sized - + std::clone::Clone - + std::fmt::Debug - + prosa::core::msg::Tvf - + std::default::Default, A: 'static + HyperClientAdaptor + std::marker::Send + std::marker::Sync, { if let Some(data) = msg.take_data() { - match adaptor.process_srv_request(data, target_url) { + match adaptor.process_srv_request(data, &self.target.url) { Ok(http_request) => Some((msg, http_request)), Err(e) => { let _ = msg.return_error_to_sender(None, e); @@ -138,345 +302,535 @@ impl HyperClientSocket { } else { let _ = msg.return_error_to_sender( None, - ServiceError::UnableToReachService(service_name.to_string()), + ServiceError::UnableToReachService(self.service_name.clone()), ); None } } - /// Helper to handle handshake timeout errors - fn handle_handshake_timeout( - socket_id: i32, - target_addr: &str, - timeout_ms: u64, + /// Turn a handshake attempt into the connected pair it yields, reporting what went wrong + fn handshake_result( + &self, + result: Result, time::error::Elapsed>, protocol: &str, - ) -> HyperProcError { - warn!( - socket_id = socket_id, - addr = target_addr, - "{protocol} handshake timeout after {timeout_ms} ms" - ); - HyperProcError::Io(io::Error::new( - io::ErrorKind::TimedOut, - format!("{protocol} handshake timeout after {timeout_ms} ms"), - )) - } - - /// Helper to handle handshake errors - fn handle_handshake_error( - socket_id: i32, - target_addr: &str, - error: hyper::Error, - protocol: &str, - ) -> HyperProcError { - warn!( - socket_id = socket_id, - addr = target_addr, - "{protocol} handshake error: {error}" - ); - HyperProcError::Hyper(error, target_addr.to_string()) + ) -> Result { + match result { + Ok(Ok(connected)) => Ok(connected), + Ok(Err(e)) => { + warn!( + socket_id = self.id, + addr = %self.target, + "{protocol} handshake error: {e}" + ); + Err(HyperProcError::Hyper(e, self.target.to_string())) + } + Err(_) => { + let connect_timeout = self.target.connect_timeout; + warn!( + socket_id = self.id, + addr = %self.target, + "{protocol} handshake timeout after {connect_timeout} ms" + ); + Err(HyperProcError::Io(io::Error::new( + io::ErrorKind::TimedOut, + format!("{protocol} handshake timeout after {connect_timeout} ms"), + ))) + } + } } - /// Method to spawn a task that handle the Hyper client socket with HTTP/1.1 - async fn spawn_http1( - self, - io: TokioIo, - proc: Arc>, - adaptor: Arc, - service_name: String, - message_histogram: Histogram, - ) -> Result + /// Send one HTTP/1.1 request and answer its sender, telling whether the socket can carry on and + /// whether the backend answered at all. + /// + /// The message timeout bounds the whole exchange, body included, because a response that is + /// only half read leaves the connection where the next one can't be correlated. That is also + /// why a timeout ends the socket instead of only failing the request + async fn exchange( + &self, + sender: &mut http1::SendRequest>, + msg: RequestMsg, + request: Request>, + adaptor: &Arc, + message_histogram: &Histogram, + ) -> ControlFlow<(), bool> where - M: 'static - + std::marker::Send - + std::marker::Sync - + std::marker::Sized - + std::clone::Clone - + std::fmt::Debug - + prosa::core::msg::Tvf - + std::default::Default, A: 'static + Adaptor + HyperClientAdaptor + std::marker::Send + std::marker::Sync, { - let socket_id = io.inner().as_raw_fd(); + let http_timeout = self.http_timeout(); let target_addr = self.target.to_string(); - match time::timeout( - Duration::from_millis(self.target.connect_timeout), - http1::handshake(io), - ) - .await - { - Ok(Ok((mut sender, mut connection))) => { - debug!( - socket_id = socket_id, + let http_log = request.uri().to_string(); + let started = Instant::now(); + + let exchanged = timeout(http_timeout, async { + // Hyper takes the request the moment it is handed over and refuses it until the + // connection reported the previous one done, so readiness is asked for, not assumed + let response = match sender.ready().await { + Ok(()) => sender.send_request(request).await, + Err(e) => Err(e), + }; + + // Whether the backend answered, which is not the same as the adaptor accepting what it + // answered: a rejected payload still came over a connection that works + let answered = response.is_ok(); + + record_message( + message_histogram, + &target_addr, + &response, + started, + "HTTP/1.1", + ); + + (answered, adaptor.process_http_response(response).await) + }) + .await; + + match exchanged { + Ok((answered, Ok(response))) => { + let _ = msg.return_to_sender(response); + ControlFlow::Continue(answered) + } + Ok((answered, Err(e))) => { + let _ = msg.return_error_to_sender(None, e); + ControlFlow::Continue(answered) + } + Err(_) => { + info!( + socket_id = self.id, addr = target_addr, - "Connected to HTTP1 remote" + "Message timeout after {} ms: {:?} - {}", + http_timeout.as_millis(), + msg, + http_log ); - let mut rx_queue = - Self::setup_socket_queue(&proc, socket_id as u32, &service_name).await?; - debug!( - socket_id = socket_id, - addr = target_addr, - "HTTP client expose service name: {}", - service_name + let _ = msg.return_error_to_sender( + None, + ServiceError::Timeout( + self.service_name.clone(), + http_timeout.as_millis() as u64, + ), ); - let mut msg_to_send: Option> = None; - let mut req_instant = Instant::now(); - - loop { - if let Some((msg, http_request)) = msg_to_send.take() { - let http_log = http_request.uri().to_string(); - tokio::select! { - // Closed the socket - Err(_) = &mut connection => { - debug!(socket_id = socket_id, addr = target_addr, "Remote HTTP1 close the socket"); - close_socket!(self, proc, socket_id, rx_queue, service_name, Ok(self)); - } - // Send an HTTP request - response_sent = timeout(self.http_timeout, sender.send_request(http_request)) => { - match response_sent { - Ok(response) => { - let (code, version) = response.as_ref().map_or((500, "HTTP/1"), |r| (r.status().as_u16() as i64, hyper_version_str(r.version()))); - message_histogram.record( - req_instant.elapsed().as_millis() as u64, - &[ - KeyValue::new("target", target_addr.clone()), - KeyValue::new("code", code), - KeyValue::new("version", version), - ], - ); - - match adaptor.process_http_response(response).await { - Ok(r) => { - let _ = msg.return_to_sender(r); - }, - Err(e) => { - let _ = msg.return_error_to_sender(None, e); - } - } - }, - Err(elapsed) => { - info!(socket_id = socket_id, addr = target_addr, "Message timeout after {} ms: {:?} - {}", elapsed, msg, http_log); - let _ = msg.return_error_to_sender(None, ServiceError::Timeout(service_name.clone(), self.http_timeout.as_millis() as u64)); - // Need to drop the connection because it's HTTP1 - close_socket!(self, proc, socket_id, rx_queue, service_name, Ok(self)); - }, - }; - } - } - } else { - tokio::select! { - // Closed the socket - Err(_) = &mut connection => { - debug!(socket_id = socket_id, addr = target_addr, "Remote close the socket"); - close_socket!(self, proc, socket_id, rx_queue, service_name, Ok(self)); - } - // Receive a message to send from the queue - Some(msg) = rx_queue.recv() => { - debug!(socket_id = socket_id, addr = target_addr, "HTTP client receive a message to send: {:?}", msg); - match msg { - InternalMsg::Request(req_msg) => { - if let Some(result) = Self::process_request(&adaptor, req_msg, &self.target.url, &service_name) { - msg_to_send = Some(result); - req_instant = Instant::now(); - } - }, - InternalMsg::Response(msg) => panic!( - "The HTTP1 hyper client socket {}/{socket_id} receive a response {:?}", - proc.get_proc_id(), - msg - ), - InternalMsg::Error(err_msg) => panic!( - "The HTTP1 hyper client socket {}/{socket_id} receive an error {:?}", - proc.get_proc_id(), - err_msg - ), - InternalMsg::Command(_) | InternalMsg::Config => { - // TODO: Implement Command/Config handling or document as unsupported - }, - InternalMsg::Service(_table) => {/* Will not use service table */}, - InternalMsg::Shutdown => { - // Remove the socket queue and wait message to finish - close_socket!(self, proc, socket_id, rx_queue, service_name, Ok(self)); - } - } + ControlFlow::Break(()) + } + } + } + + /// Serve the Hyper client socket with HTTP/1.1 until its connection ends, answering whether it + /// managed to serve anything on it + async fn serve_http1( + &mut self, + io: TokioIo, + proc: &ProcParam, + adaptor: &Arc, + message_histogram: &Histogram, + ) -> Result + where + A: 'static + Adaptor + HyperClientAdaptor + std::marker::Send + std::marker::Sync, + { + let fd = io.inner().as_raw_fd(); + let connect_timeout = Duration::from_millis(self.target.connect_timeout); + + // Raced against the shutdown like the connect that produced the stream: it has a budget of + // its own, and a socket that has not finished shaking hands holds nothing anyone waits on + let handshake = tokio::select! { + handshake = time::timeout(connect_timeout, http1::handshake(io)) => handshake, + _ = Self::stopped(&mut self.control) => return Ok(true), + }; + let (mut sender, connection) = self.handshake_result(handshake, "HTTP1")?; + + // The connection runs in its own task, which is what lets the loop below await a response + // body: Hyper only hands over the bytes it has read, and it only reads while the connection + // is polled. Driving it from the loop would stall any body that doesn't arrive with the head + let mut connection = tokio::task::spawn(connection); + + debug!( + socket_id = self.id, + fd = fd, + addr = %self.target, + "Connected to HTTP1 remote, expose the service {}", + self.service_name + ); + self.declare_socket_queue(proc).await?; + + // Whether the backend answered anything on this connection, which is what tells one that is + // merely slow from one that accepts and closes without ever serving + let mut served = false; + + loop { + tokio::select! { + // Closed the socket + closed = &mut connection => { + debug!(socket_id = self.id, addr = %self.target, "Remote HTTP1 close the socket: {closed:?}"); + break; + } + // The processor retired the socket + _ = Self::stopped(&mut self.control) => break, + // Receive a message to send from the queue. A retired socket stops taking them, so + // the exchange it is in the middle of is the last one it serves: the arms above + // are only reached between two of them, never during one + Some(msg) = self.rx_queue.recv(), if !self.is_stopped() => { + debug!(socket_id = self.id, addr = %self.target, "HTTP client receive a message to send: {msg:?}"); + match msg { + InternalMsg::Request(req_msg) => { + let Some((msg, request)) = self.process_request(adaptor, req_msg) else { + continue; + }; + + // HTTP/1.1 correlates by order, so the exchange is awaited here and + // nothing else goes out until it is answered + match self.exchange(&mut sender, msg, request, adaptor, message_histogram).await { + ControlFlow::Continue(answered) => served |= answered, + ControlFlow::Break(()) => break, } - } + }, + InternalMsg::Response(msg) => panic!( + "The HTTP1 hyper client socket {}/{} receive a response {:?}", + proc.get_proc_id(), + self.id, + msg + ), + InternalMsg::Error(err_msg) => panic!( + "The HTTP1 hyper client socket {}/{} receive an error {:?}", + proc.get_proc_id(), + self.id, + err_msg + ), + // The processor is the one that reads the service table and the + // configuration, and tells its sockets what came out of it + InternalMsg::Service(_) | InternalMsg::Config(_) => {}, + // The main task shutting ProSA down + InternalMsg::Shutdown => break, } } } - Ok(Err(e)) => Err(Self::handle_handshake_error( - socket_id, - &target_addr, - e, - "HTTP1", - )), - Err(_) => Err(Self::handle_handshake_timeout( - socket_id, - &target_addr, - self.target.connect_timeout, - "HTTP1", - )), } + + // What is left in the queue is served by the next connection of the socket, or answered by + // the processor if it doesn't restart it + self.withdraw_socket_queue(proc).await; + + Ok(served) } - /// Method to spawn a task that handle the Hyper client socket with HTTP/2 - async fn spawn_h2( - mut self, + /// Let the requests already sent on an HTTP/2 socket finish before the socket goes away. + /// + /// They run in their own task and hold a sender of the connection, which keeps its task alive + /// for as long as they need it, so waiting is all there is to do. Bounded by the message + /// timeout, past which they are detached rather than aborted, because an aborted task never + /// answers the sender of the request it carries + async fn drain_requests(&self, requests: &mut JoinSet<()>) { + if requests.is_empty() { + return; + } + + debug!( + socket_id = self.id, + addr = %self.target, + "Wait for {} in flight request(s) before closing the socket", + requests.len() + ); + + let drained = timeout(self.http_timeout(), async { + while let Some(request) = requests.join_next().await { + if let Err(e) = request { + warn!( + socket_id = self.id, + addr = %self.target, + "An HTTP2 request task panicked, its sender won't be answered: {e}" + ); + } + } + }) + .await; + + if drained.is_err() { + warn!( + socket_id = self.id, + addr = %self.target, + "Close the socket with {} request(s) still in flight", + requests.len() + ); + + requests.detach_all(); + } + } + + /// Serve the Hyper client socket with HTTP/2 until its connection ends, answering whether it + /// managed to serve anything on it + async fn serve_h2( + &mut self, io: TokioIo, - proc: Arc>, - adaptor: Arc, - service_name: String, - message_histogram: Histogram, - ) -> Result + proc: &ProcParam, + adaptor: &Arc, + message_histogram: &Histogram, + ) -> Result where - M: 'static - + std::marker::Send - + std::marker::Sync - + std::marker::Sized - + std::clone::Clone - + std::fmt::Debug - + prosa::core::msg::Tvf - + std::default::Default, A: 'static + Adaptor + HyperClientAdaptor + std::marker::Send + std::marker::Sync, { - let socket_id = io.inner().as_raw_fd(); - let target_addr = self.target.to_string(); - self.is_http2 = true; - match time::timeout( - Duration::from_millis(self.target.connect_timeout), - http2::handshake(TokioExecutor::new(), io), - ) - .await - { - Ok(Ok((sender, mut connection))) => { - debug!( - socket_id = socket_id, - addr = target_addr, - "Connected to HTTP2 remote" - ); - let mut rx_queue = - Self::setup_socket_queue(&proc, socket_id as u32, &service_name).await?; - - loop { - tokio::select! { - // Closed the socket - Err(_) = &mut connection => { - debug!(socket_id = socket_id, addr = target_addr, "Remote HTTP2 close the socket"); - close_socket!(self, proc, socket_id, rx_queue, service_name, Ok(self)); - } - // Receive a message to send from the queue - Some(msg) = rx_queue.recv() => { - match msg { - InternalMsg::Request(mut req_msg) => { - if let Some(data) = req_msg.take_data() { - let req_instant = Instant::now(); - let mut sender = sender.clone(); - let adaptor = adaptor.clone(); - let target_url = self.target.url.clone(); - let message_histogram = message_histogram.clone(); - let http_timeout = self.http_timeout; - let service_name_clone = service_name.clone(); - - tokio::spawn(async move { - match adaptor.process_srv_request(data, &target_url) { - Ok(http_request) => { - match timeout(http_timeout, sender.send_request(http_request)).await { - Ok(http_response) => { - let (code, version) = http_response.as_ref().map_or((500, "HTTP/2"), |r| (r.status().as_u16() as i64, hyper_version_str(r.version()))); - message_histogram.record( - req_instant.elapsed().as_millis() as u64, - &[ - KeyValue::new("target", target_url.to_string()), - KeyValue::new("code", code), - KeyValue::new("version", version), - ], - ); - - match adaptor.process_http_response(http_response).await { - Ok(response) => { - let _ = req_msg.return_to_sender(response); - }, - Err(e) => { let _ = req_msg.return_error_to_sender(None, e); }, - } - }, - Err(_) => { let _ = req_msg.return_error_to_sender(None, ServiceError::Timeout(service_name_clone, http_timeout.as_millis() as u64)); }, - }; - }, - Err(e) => { - let _ = req_msg.return_error_to_sender(None, e); - }, - } - }); - } else { - let _ = req_msg.return_error_to_sender(None, ServiceError::UnableToReachService(service_name.clone())); + let fd = io.inner().as_raw_fd(); + let connect_timeout = Duration::from_millis(self.target.connect_timeout); + + // Raced against the shutdown like the connect that produced the stream: it has a budget of + // its own, and a socket that has not finished shaking hands holds nothing anyone waits on + let handshake = tokio::select! { + handshake = time::timeout(connect_timeout, http2::handshake(TokioExecutor::new(), io)) => handshake, + _ = Self::stopped(&mut self.control) => return Ok(true), + }; + let (sender, connection) = self.handshake_result(handshake, "HTTP2")?; + + // The connection runs in its own task, so it keeps serving the requests below, which read + // their response body long after the loop moved on, and outlives this method for the ones + // that are still draining + let mut connection = tokio::task::spawn(connection); + + debug!( + socket_id = self.id, + fd = fd, + addr = %self.target, + "Connected to HTTP2 remote, expose the service {}", + self.service_name + ); + self.declare_socket_queue(proc).await?; + + // Requests are multiplexed on the connection, each one running in its own task. Tracked so + // they can be drained when the socket closes + let mut requests = JoinSet::new(); + + // Whether the backend answered anything on this connection, which is what tells one that is + // merely slow from one that accepts and closes without ever serving. Shared with the + // request tasks, so an answer counts as long as its task was joined: one detached by + // `drain_requests` stores it after this has been read, and is lost. That only happens after + // a whole message timeout of waiting, by which point the connection lasted long enough to + // count on its own + let served = Arc::new(AtomicBool::new(false)); + + loop { + tokio::select! { + // Closed the socket + closed = &mut connection => { + debug!(socket_id = self.id, addr = %self.target, "Remote HTTP2 close the socket: {closed:?}"); + break; + } + // The processor retired the socket + _ = Self::stopped(&mut self.control) => break, + // Reap the requests that are done, so the set doesn't grow with the socket. A task + // that panicked took the request it held with it, which nothing can answer anymore, + // so the least it can do is not be silent about it + Some(request) = requests.join_next(), if !requests.is_empty() => { + if let Err(e) = request { + warn!(socket_id = self.id, addr = %self.target, "An HTTP2 request task panicked, its sender won't be answered: {e}"); + } + }, + // Receive a message to send from the queue. A retired socket stops taking them and + // drains the ones it multiplexed rather than starting another + Some(msg) = self.rx_queue.recv(), if !self.is_stopped() => { + match msg { + InternalMsg::Request(mut req_msg) => { + let Some(data) = req_msg.take_data() else { + let _ = req_msg.return_error_to_sender(None, ServiceError::UnableToReachService(self.service_name.clone())); + continue; + }; + + let started = Instant::now(); + let mut sender = sender.clone(); + let adaptor = adaptor.clone(); + let target_url = self.target.url.clone(); + let target_addr = self.target.to_string(); + let message_histogram = message_histogram.clone(); + let http_timeout = self.http_timeout(); + let service_name = self.service_name.clone(); + let served = served.clone(); + + requests.spawn(async move { + let request = match adaptor.process_srv_request(data, &target_url) { + Ok(request) => request, + Err(e) => { + let _ = req_msg.return_error_to_sender(None, e); + return; + } + }; + + let answered = timeout(http_timeout, async { + let response = sender.send_request(request).await; + if response.is_ok() { + served.store(true, Ordering::Relaxed); } - }, - InternalMsg::Response(msg) => panic!( - "The H2 hyper client socket {}/{socket_id} receive a response {:?}", - proc.get_proc_id(), - msg - ), - InternalMsg::Error(err_msg) => panic!( - "The H2 hyper client socket {}/{socket_id} receive an error {:?}", - proc.get_proc_id(), - err_msg - ), - InternalMsg::Command(_) | InternalMsg::Config => { - // TODO: Implement Command/Config handling or document as unsupported - }, - InternalMsg::Service(_table) => {/* Will not use service table */}, - InternalMsg::Shutdown => { - // Remove the socket queue and wait message to finish - close_socket!(self, proc, socket_id, rx_queue, service_name, Ok(self)); + record_message(&message_histogram, &target_addr, &response, started, "HTTP/2"); + adaptor.process_http_response(response).await + }) + .await; + + match answered { + Ok(Ok(response)) => { let _ = req_msg.return_to_sender(response); }, + Ok(Err(e)) => { let _ = req_msg.return_error_to_sender(None, e); }, + Err(_) => { let _ = req_msg.return_error_to_sender(None, ServiceError::Timeout(service_name, http_timeout.as_millis() as u64)); }, } - } + }); + }, + InternalMsg::Response(msg) => panic!( + "The H2 hyper client socket {}/{} receive a response {:?}", + proc.get_proc_id(), + self.id, + msg + ), + InternalMsg::Error(err_msg) => panic!( + "The H2 hyper client socket {}/{} receive an error {:?}", + proc.get_proc_id(), + self.id, + err_msg + ), + // The processor is the one that reads the service table and the + // configuration, and tells its sockets what came out of it + InternalMsg::Service(_) | InternalMsg::Config(_) => {}, + // The main task shutting ProSA down + InternalMsg::Shutdown => break, + } + } + } + } + + // Stop taking new requests before draining the ones already sent, which must happen even if + // the bus couldn't be told: dropping the set here would abort them unanswered + self.withdraw_socket_queue(proc).await; + self.drain_requests(&mut requests).await; + + Ok(served.load(Ordering::Relaxed)) + } + + /// Connect the socket and serve it until it closes, answering whether the connection worked. + /// + /// A connection that served something did its job, and so did one that merely stayed open long + /// enough: an idle keep-alive close is healthy, and a socket that backed off from those would + /// leave a quiet pool disconnected. What is left is a backend that accepts and closes right + /// away, which is indistinguishable from one that refuses and has to be backed off the same way + async fn connect( + &mut self, + proc: &ProcParam, + adaptor: &Arc, + meters: &SocketMeters, + healthy_connection: Duration, + ) -> Result + where + A: 'static + Adaptor + HyperClientAdaptor + std::marker::Send + std::marker::Sync, + { + // Picked up here rather than at spawn, so a reload that only changes it applies to the next + // attempt instead of retiring a socket that is connecting perfectly well + self.target.connect_timeout = self.control.borrow().connect_timeout; + + // Connecting is the one thing a socket does that its bus queue can't interrupt, and it can + // take two `connect_timeout` with the handshake below, so the shutdown is raced against it + let stream = tokio::select! { + stream = self.target.connect() => stream?, + _ = Self::stopped(&mut self.control) => return Ok(true), + }; + + let io = TokioIo::new(stream); + let is_http2 = io.inner().selected_alpn_check(|alpn| alpn == H2); + + // Counted here rather than in the processor, which only knows the sockets it wants to + // exist. This reports the ones that are really connected, so a backend that is down or + // flapping shows up as its counter dropping + let socket_attributes = [ + KeyValue::new("target", self.target.to_string()), + KeyValue::new("version", if is_http2 { "HTTP/2" } else { "HTTP/1.1" }), + ]; + meters.socket_counter.add(1, &socket_attributes); + + let connected_at = Instant::now(); + let result = if is_http2 { + self.serve_h2(io, proc, adaptor, &meters.message_histogram) + .await + } else { + self.serve_http1(io, proc, adaptor, &meters.message_histogram) + .await + }; + + meters.socket_counter.add(-1, &socket_attributes); + result.map(|served| served || connected_at.elapsed() >= healthy_connection) + } + + /// Wait `reconnect_delay` out before connecting again, serving the queue in the meantime. + /// + /// The socket has no bus queue while it isn't connected, so nothing should reach it here, but a + /// processor holding an older service table still can. Answering rather than letting a request + /// sit in the queue is what keeps its sender from waiting on a backend that is down. + /// + /// Break when the socket was asked to stop, which the processor does here too: a socket that is + /// only waiting to reconnect is unknown to the main task + async fn wait_reconnect(&mut self, reconnect_delay: Duration) -> ControlFlow<()> { + if reconnect_delay.is_zero() { + return ControlFlow::Continue(()); + } + + debug!( + socket_id = self.id, + addr = %self.target, + "Reconnect the socket in {reconnect_delay:?}" + ); + + let sleep = time::sleep(reconnect_delay); + tokio::pin!(sleep); + loop { + tokio::select! { + _ = &mut sleep => return ControlFlow::Continue(()), + _ = Self::stopped(&mut self.control) => return ControlFlow::Break(()), + Some(msg) = self.rx_queue.recv() => { + match msg { + InternalMsg::Request(req_msg) => { + let _ = req_msg.return_error_to_sender( + None, + ServiceError::UnableToReachService(self.service_name.clone()), + ); } + InternalMsg::Shutdown => return ControlFlow::Break(()), + _ => {} } } } - Ok(Err(e)) => Err(Self::handle_handshake_error( - socket_id, - &target_addr, - e, - "HTTP2", - )), - Err(_) => Err(Self::handle_handshake_timeout( - socket_id, - &target_addr, - self.target.connect_timeout, - "HTTP2", - )), } } - /// Method to spawn a task to handle the Hyper client socket - pub fn spawn( - self, - join_set: &mut JoinSet>, + /// Method to spawn a task to handle the Hyper client socket, answering the task it runs in. + /// + /// The task always gives the socket back to the processor, even when it never managed to + /// connect, so a backend that is down doesn't silently shrink the pool. The processor is the + /// only one that decides whether to restart it, and the task id is how it finds the slot again + /// if the task panicked instead of returning + pub(super) fn spawn( + mut self, + join_set: &mut JoinSet<(Self, Option)>, proc: Arc>, adaptor: Arc, - service_name: String, - message_histogram: Histogram, - ) where - M: 'static - + std::marker::Send - + std::marker::Sync - + std::marker::Sized - + std::clone::Clone - + std::fmt::Debug - + prosa::core::msg::Tvf - + std::default::Default, + settings: &HyperClientSettings, + meters: SocketMeters, + ) -> tokio::task::Id + where A: 'static + Adaptor + HyperClientAdaptor + std::marker::Send + std::marker::Sync, { - join_set.spawn(async move { - let io = TokioIo::new(self.target.connect().await?); - if io.inner().selected_alpn_check(|alpn| alpn == H2) { - self.spawn_h2::(io, proc, adaptor, service_name, message_histogram) - .await - } else { - self.spawn_http1::(io, proc, adaptor, service_name, message_histogram) - .await - } - }); + let reconnect_delay = settings.reconnect_delay(self.retry); + let healthy_connection = settings.base_reconnect_delay(); + + join_set + .spawn(async move { + if self.is_stopped() || self.wait_reconnect(reconnect_delay).await.is_break() { + return (self, None); + } + + let connected = self + .connect(&proc, &adaptor, &meters, healthy_connection) + .await; + + // Back off from a backend the socket couldn't get a working connection out of, whether + // it refused one or gave one it closed straight away. Both look the same to a caller, + // and reconnecting either without waiting is a loop at the speed of the machine + self.retry = if matches!(connected, Ok(true)) { + 0 + } else { + self.retry.saturating_add(1) + }; + + (self, connected.err()) + }) + .id() } } diff --git a/src/server.rs b/src/server.rs index c8189cc..b3c3885 100644 --- a/src/server.rs +++ b/src/server.rs @@ -10,13 +10,15 @@ pub(crate) mod service; #[cfg(test)] mod tests { use bytes::Bytes; - use http_body_util::{Full, combinators::BoxBody}; + use http_body_util::{Empty, Full, combinators::BoxBody}; use hyper::{Request, StatusCode}; + use hyper_util::rt::TokioIo; use prosa::core::{ adaptor::Adaptor, error::ProcError, main::{MainProc, MainRunnable as _}, - proc::{Proc, ProcConfig as _}, + proc::{Proc, ProcBusParam as _, ProcConfig as _}, + settings::ProsaConfig, }; use prosa_utils::{ config::ssl::{SslConfig, Store}, @@ -27,6 +29,11 @@ mod tests { env, fs::{self, File}, io::{self, Read as _}, + path::Path, + sync::{ + Arc, + atomic::{AtomicBool, AtomicUsize, Ordering}, + }, time::Duration, }; use tokio::time; @@ -35,7 +42,7 @@ mod tests { use crate::{ HyperResp, server::{adaptor::HyperServerAdaptor, proc::HyperServerProc}, - tests::HttpTestSettings, + tests::{HttpTestSettings, TEST_TIMEOUT, bound_url, set_bound_port, wait_for}, }; const WAIT_TIME: time::Duration = time::Duration::from_secs(5); @@ -57,11 +64,15 @@ mod tests { + std::default::Default, { fn new( - _proc: &crate::server::proc::HyperServerProc, + proc: &crate::server::proc::HyperServerProc, + addr: prosa::io::SocketAddr, ) -> Result> where Self: Sized, { + // The listener is configured on the port 0, this is where the test learns where it landed + set_bound_port(proc.name(), addr.port()); + Ok(ServerTestAdaptor {}) } @@ -80,10 +91,72 @@ mod tests { } } + /// Time [`SlowServerTestAdaptor`] takes to answer its first request, long enough to still be + /// serving when ProSA stops. Every further request takes a multiple of it, so the requests in + /// flight don't all end at the same moment + const SLOW_RESPONSE_TIME: time::Duration = time::Duration::from_millis(300); + + /// Number of requests [`SlowServerTestAdaptor`] started answering, so a test knows how many are + /// in flight, and so each one can be given a different response time + static SLOW_REQUESTS_STARTED: AtomicUsize = AtomicUsize::new(0); + + /// Set when the Hyper server processor reaches the end of its loop, which it only does once + /// every connection has been drained + static SLOW_PROC_TERMINATED: AtomicBool = AtomicBool::new(false); + + /// Adaptor that takes its time to answer, so a request is still being served when ProSA stops + #[derive(Clone)] + struct SlowServerTestAdaptor { + // Nothing + } + + impl Adaptor for SlowServerTestAdaptor { + fn terminate(&self) { + SLOW_PROC_TERMINATED.store(true, Ordering::Relaxed); + } + } + + impl HyperServerAdaptor for SlowServerTestAdaptor + where + M: 'static + + std::marker::Send + + std::marker::Sync + + std::marker::Sized + + std::clone::Clone + + std::fmt::Debug + + prosa_utils::msg::tvf::Tvf + + std::default::Default, + { + fn new( + proc: &crate::server::proc::HyperServerProc, + addr: prosa::io::SocketAddr, + ) -> Result> + where + Self: Sized, + { + set_bound_port(proc.name(), addr.port()); + + Ok(SlowServerTestAdaptor {}) + } + + async fn process_http_request( + &self, + _req: Request, + ) -> HyperResp { + let request_index = SLOW_REQUESTS_STARTED.fetch_add(1, Ordering::Relaxed); + time::sleep(SLOW_RESPONSE_TIME * (request_index as u32 + 1)).await; + + >::response_builder(self, StatusCode::OK) + .body(BoxBody::new(Full::new(Bytes::from("Hello, slow world")))) + .into() + } + } + async fn run_test( settings: HttpTestSettings, certificate: Option, http2: bool, + proc_name: &str, ) -> io::Result<()> { let url = settings.server.listener.url.clone(); @@ -96,14 +169,14 @@ mod tests { // Launch an HTTP server processor let http_server_proc = HyperServerProc::::create( 1, - String::from("HTTP_SERVER_PROC"), + String::from(proc_name), bus.clone(), settings.server, ); Proc::::run(http_server_proc)?; - // Wait for processor to start - std::thread::sleep(Duration::from_secs(1)); + // The listener is on the port 0, the processor publishes where it bound + let url = bound_url(proc_name, &url).await; // Send request to the server with reqwest let mut client_builder = reqwest::ClientBuilder::new() @@ -148,13 +221,17 @@ mod tests { #[tokio::test] async fn http_client_server() { let test_settings = HttpTestSettings::new( - Url::parse("http://localhost:48180").expect("HTTP client/server URL should be valid"), + Url::parse("http://localhost:0").expect("HTTP client/server URL should be valid"), None, None, ); // Run a ProSA to test - assert!(run_test(test_settings, None, false).await.is_ok()); + assert!( + run_test(test_settings, None, false, "SRV_HTTP_PROC") + .await + .is_ok() + ); } #[tokio::test] @@ -203,14 +280,14 @@ mod tests { client_ssl_config.set_store(client_ssl_store); let test_settings = HttpTestSettings::new( - Url::parse("https://localhost:48543").expect("HTTPS client/server URL should be valid"), + Url::parse("https://localhost:0").expect("HTTPS client/server URL should be valid"), Some(server_ssl_config), Some(client_ssl_config), ); // Run a ProSA to test assert!( - run_test(test_settings, Some(client_cert), false) + run_test(test_settings, Some(client_cert), false, "SRV_HTTPS_PROC") .await .is_ok() ); @@ -230,7 +307,7 @@ mod tests { .as_os_str() .to_str() .expect("Cert path should be a valid String"); - let mut server_ssl_config = HttpTestSettings::create_server_cert( + let server_ssl_config = HttpTestSettings::create_server_cert( key_path .as_os_str() .to_str() @@ -239,8 +316,6 @@ mod tests { cert_path_str.into(), ) .expect("Server certificate should be created"); - // Need to set the ALPN for server because of inline configuration @see TargetSetting::new - server_ssl_config.set_alpn(vec!["h2".into()]); let mut buf = Vec::new(); File::open(cert_path_str) @@ -264,16 +339,364 @@ mod tests { client_ssl_config.set_alpn(vec!["h2".into()]); let test_settings = HttpTestSettings::new( - Url::parse("https://localhost:49543").expect("HTTP2 client/server URL should be valid"), + Url::parse("https://localhost:0").expect("HTTP2 client/server URL should be valid"), Some(server_ssl_config), Some(client_ssl_config), ); // Run a ProSA to test assert!( - run_test(test_settings, Some(client_cert), true) + run_test(test_settings, Some(client_cert), true, "SRV_H2_PROC") .await .is_ok() ); } + + /// Send a GET request through a UNIX socket, and return the status the server answered. + /// + /// A listener that binds without ever serving accepts connections all the same, so only a + /// complete round trip tells that the processor rebound + async fn unix_get(socket_path: &Path) -> Option { + let stream = tokio::net::UnixStream::connect(socket_path).await.ok()?; + let (mut sender, connection) = hyper::client::conn::http1::handshake(TokioIo::new(stream)) + .await + .ok()?; + tokio::spawn(connection); + + let request = Request::builder() + .uri("/") + .header(hyper::header::HOST, "localhost") + .body(Empty::::new()) + .ok()?; + + sender + .send_request(request) + .await + .ok() + .map(|resp| resp.status()) + } + + #[tokio::test] + async fn server_config_reload() { + const PROC_NAME: &str = "SRV_RELOAD_PROC"; + const PROSA_RELOAD_TEST_DIR_NAME: &str = "ProSA_server_reload"; + + let settings = HttpTestSettings::new( + Url::parse("http://127.0.0.1:0").expect("Initial server URL should be valid"), + None, + None, + ); + let initial_url = settings.server.listener.url.clone(); + + // A reload has to say where to go, and a port named before anything binds it is a port + // another test can be given in the meantime. A UNIX socket is an address the test owns + // outright, so the rebind lands where it is told + let prosa_temp_dir = env::temp_dir().join(PROSA_RELOAD_TEST_DIR_NAME); + let _ = fs::remove_dir_all(&prosa_temp_dir); + fs::create_dir_all(&prosa_temp_dir) + .expect("Can't create ProSA temporary directory for the configuration reload"); + let socket_path = prosa_temp_dir.join("prosa_server_reload.sock"); + let reloaded_url = Url::parse(&format!( + "unix://{}", + socket_path + .to_str() + .expect("Socket path should be a valid String") + )) + .expect("Reloaded server URL should be valid"); + + // Create bus and main processor + let (bus, main) = MainProc::::create(&settings, Some(1)); + + // The main task must run to broadcast the configuration to the processors + let main_task = tokio::spawn(main.run()); + + // Launch an HTTP server processor + let http_server_proc = HyperServerProc::::create( + 1, + String::from(PROC_NAME), + bus.clone(), + settings.server, + ); + Proc::::run(http_server_proc) + .expect("Hyper server processor should run"); + + // The listener is on the port 0, the processor publishes where it bound + let initial_url = bound_url(PROC_NAME, &initial_url).await; + let initial_addr = format!( + "127.0.0.1:{}", + initial_url.port().expect("Initial URL should have a port") + ); + + // The connection is kept alive on purpose: it outlives the rebind, and must not keep the + // retired listener bound + let client = reqwest::ClientBuilder::new() + .timeout(WAIT_TIME) + .build() + .expect("reqwest client should be valid"); + let resp = client + .get(initial_url) + .send() + .await + .expect("Failed to send request to the initial URL"); + assert_eq!(resp.status(), StatusCode::OK); + + // Move the listener to another address + let config = ProsaConfig::from_config( + config::Config::builder() + .set_override(format!("{PROC_NAME}.listener.url"), reloaded_url.as_str()) + .expect("Reloaded listener URL should be a valid config override") + .build() + .expect("Reloaded configuration should be valid"), + ) + .expect("Reloaded ProSA configuration should be valid"); + bus.update_config(Arc::new(config)) + .await + .expect("ProSA configuration should be updated"); + + assert!( + wait_for(TEST_TIMEOUT, async || unix_get(&socket_path).await + == Some(StatusCode::OK)) + .await, + "The Hyper server processor should serve on {reloaded_url}" + ); + + assert!( + wait_for(TEST_TIMEOUT, async || tokio::net::TcpListener::bind( + initial_addr.as_str() + ) + .await + .is_ok()) + .await, + "The initial listener should have been freed by the reload" + ); + + bus.stop("ProSA HTTP server configuration reload unit test end".into()) + .await + .expect("ProSA should stop"); + + // Wait on main task to end + let _ = main_task.await; + } + + /// Open a TLS connection and give back the certificate the server presented. + /// + /// Done with openssl rather than `reqwest`, which verifies the certificate and then keeps it to + /// itself. Blocking, so it runs off the test runtime + async fn served_certificate(addr: String) -> Option> { + tokio::task::spawn_blocking(move || { + let mut connector = openssl::ssl::SslConnector::builder(openssl::ssl::SslMethod::tls()) + .expect("The test TLS connector should build"); + // Which certificate is served is the whole point, whether it is trusted isn't + connector.set_verify(openssl::ssl::SslVerifyMode::NONE); + + let stream = std::net::TcpStream::connect(&addr).ok()?; + let stream = connector.build().connect("localhost", stream).ok()?; + let certificate = stream.ssl().peer_certificate()?; + certificate.to_pem().ok() + }) + .await + .ok() + .flatten() + } + + #[tokio::test] + async fn server_certificate_reload() { + const PROC_NAME: &str = "SRV_CERT_RELOAD_PROC"; + const PROSA_CERT_RELOAD_TEST_DIR_NAME: &str = "ProSA_server_cert_reload"; + + let prosa_temp_dir = env::temp_dir().join(PROSA_CERT_RELOAD_TEST_DIR_NAME); + let _ = fs::remove_dir_all(&prosa_temp_dir); + fs::create_dir_all(&prosa_temp_dir) + .expect("Can't create ProSA temporary directory for the certificate reload"); + + let key_path = prosa_temp_dir.join("prosa_server_cert_reload.key"); + let cert_path = prosa_temp_dir.join("prosa_server_cert_reload.pem"); + let key_path = key_path + .to_str() + .expect("Key path should be a valid String") + .to_string(); + let cert_path = cert_path + .to_str() + .expect("Cert path should be a valid String") + .to_string(); + + let server_ssl_config = + HttpTestSettings::create_server_cert(key_path.clone(), cert_path.clone()) + .expect("Server certificate should be created"); + + let settings = HttpTestSettings::new( + Url::parse("https://localhost:0").expect("Certificate reload URL should be valid"), + Some(server_ssl_config), + None, + ); + let initial_url = settings.server.listener.url.clone(); + + // Create bus and main processor + let (bus, main) = MainProc::::create(&settings, Some(1)); + + // The main task must run to broadcast the configuration to the processors + let main_task = tokio::spawn(main.run()); + + // Launch an HTTP server processor + let http_server_proc = HyperServerProc::::create( + 1, + String::from(PROC_NAME), + bus.clone(), + settings.server, + ); + Proc::::run(http_server_proc) + .expect("Hyper server processor should run"); + + // The listener is on the port 0, the processor publishes where it bound + let bound = bound_url(PROC_NAME, &initial_url).await; + let addr = format!( + "localhost:{}", + bound.port().expect("Bound URL should have a port") + ); + + let first_certificate = served_certificate(addr.clone()) + .await + .expect("The server should present a certificate"); + + // Renew it under the very same paths, which is what a certificate manager does. The + // configuration is left describing exactly what it described before + HttpTestSettings::create_server_cert(key_path.clone(), cert_path.clone()) + .expect("Server certificate should be renewed"); + + // Reload with settings equal to the running ones, down to the URL: the port is still the 0 + // the processor was given, so a rebind would land on another port and the address below + // would stop answering altogether + let reloaded = format!( + "{PROC_NAME}:\n listener:\n url: {}\n ssl:\n cert: {cert_path}\n key: {key_path}\n passphrase: {}\n", + initial_url.as_str(), + HttpTestSettings::PASSPHRASE, + ); + let config = ProsaConfig::from_config( + config::Config::builder() + .add_source(config::File::from_str(&reloaded, config::FileFormat::Yaml)) + .build() + .expect("Reloaded configuration should be valid"), + ) + .expect("Reloaded ProSA configuration should be valid"); + bus.update_config(Arc::new(config)) + .await + .expect("ProSA configuration should be updated"); + + // The new certificate is served, on the socket that was never rebound + assert!( + wait_for(TEST_TIMEOUT, async || { + served_certificate(addr.clone()) + .await + .is_some_and(|certificate| certificate != first_certificate) + }) + .await, + "The Hyper server processor should serve the renewed certificate on {addr}" + ); + + bus.stop("ProSA HTTP server certificate reload unit test end".into()) + .await + .expect("ProSA should stop"); + + // Wait on main task to end + let _ = main_task.await; + } + + #[tokio::test] + async fn server_graceful_shutdown() { + const PROC_NAME: &str = "SRV_SHUTDOWN_PROC"; + + let settings = HttpTestSettings::new( + Url::parse("http://127.0.0.1:0").expect("Graceful shutdown server URL should be valid"), + None, + None, + ); + let url = settings.server.listener.url.clone(); + + // Create bus and main processor + let (bus, main) = MainProc::::create(&settings, Some(1)); + let main_task = tokio::spawn(main.run()); + + // Launch an HTTP server processor + let http_server_proc = HyperServerProc::::create( + 1, + String::from(PROC_NAME), + bus.clone(), + settings.server, + ); + Proc::::run(http_server_proc) + .expect("Hyper server processor should run"); + + // The listener is on the port 0, the processor publishes where it bound + let url = bound_url(PROC_NAME, &url).await; + let addr = format!( + "127.0.0.1:{}", + url.port().expect("Bound URL should have a port") + ); + + // Fire a request on each of several connections and leave them all in flight. They are + // answered at different moments, so draining until the first one is done is not enough + const IN_FLIGHT_CONNECTIONS: usize = 3; + let mut in_flight = Vec::with_capacity(IN_FLIGHT_CONNECTIONS); + for _ in 0..IN_FLIGHT_CONNECTIONS { + let stream = tokio::net::TcpStream::connect(&addr) + .await + .expect("The Hyper server processor should accept a connection"); + let (mut sender, connection) = + hyper::client::conn::http1::handshake(TokioIo::new(stream)) + .await + .expect("The HTTP/1.1 handshake should succeed"); + tokio::spawn(connection); + let request = Request::builder() + .uri("/") + .header(hyper::header::HOST, "localhost") + .body(Empty::::new()) + .expect("The request should be valid"); + in_flight.push(tokio::spawn( + async move { sender.send_request(request).await }, + )); + } + + assert!( + wait_for(TEST_TIMEOUT, async || SLOW_REQUESTS_STARTED + .load(Ordering::Relaxed) + >= IN_FLIGHT_CONNECTIONS) + .await, + "The Hyper server processor should be serving every request" + ); + + bus.stop("ProSA HTTP server graceful shutdown unit test end".into()) + .await + .expect("ProSA should stop"); + + // The listener is released as the drain starts, so a new client is told right away instead + // of waiting in the accept queue of a server that will never take it + assert!( + wait_for(TEST_TIMEOUT, async || tokio::net::TcpStream::connect(&addr) + .await + .is_err()) + .await, + "A new connection should be refused once the processor stops accepting" + ); + + // Every connection is answered, not just the one that finished first + for request in in_flight { + let resp = request + .await + .expect("The in flight request should not be dropped") + .expect("The Hyper server processor should answer the request it was serving"); + assert_eq!(resp.status(), StatusCode::OK); + } + + // The drain completing is what ends the loop, so the processor only terminates once it has + // nothing left to serve + assert!( + wait_for(TEST_TIMEOUT, async || SLOW_PROC_TERMINATED + .load(Ordering::Relaxed)) + .await, + "The Hyper server processor should terminate once its connections are drained" + ); + + // Wait on main task to end + let _ = main_task.await; + } } diff --git a/src/server/adaptor.rs b/src/server/adaptor.rs index b9e5d0d..13dd5ee 100644 --- a/src/server/adaptor.rs +++ b/src/server/adaptor.rs @@ -5,23 +5,16 @@ use bytes::Bytes; use http::response; use http_body_util::{Empty, Full, combinators::BoxBody}; use hyper::{Request, Response, StatusCode}; -use prosa::core::{adaptor::Adaptor, error::ProcError, msg::ErrorMsg, proc::ProcBusParam as _}; +use prosa::{ + core::{adaptor::Adaptor, error::ProcError, msg::ErrorMsg, proc::ProcBusParam as _}, + io::SocketAddr, +}; -use crate::{HttpError, HyperResp, PRODUCT_VERSION_HEADER}; +use crate::{HttpError, HyperResp, PRODUCT_VERSION_HEADER, server::proc::HyperServerProc}; -use super::proc::HyperServerProc; - -#[cfg_attr(doc, aquamarine::aquamarine)] /// Trait to define the Hyper server adaptor structure /// -/// ```mermaid -/// graph LR -/// IN[Input HTTP server] -/// ProSA[ProSA Hyper Procesor] -/// -/// IN-- HTTP request (process_server_request) -->ProSA -/// ProSA-- HTTP response (process_server_response) -->IN -/// ``` +#[doc = simple_mermaid::mermaid!("diagrams/adaptor.mmd")] pub trait HyperServerAdaptor where M: 'static @@ -33,8 +26,15 @@ where + prosa::core::msg::Tvf + std::default::Default, { - /// Create a new adaptor - fn new(proc: &HyperServerProc) -> Result> + /// Create a new adaptor. + /// + /// `addr` is the address the processor bound at startup. It is not always the one configured: + /// a listener asking for the port 0 only knows where it landed once bound. A configuration + /// reload that moves the listener reuses the same adaptor, so it doesn't refresh `addr` + fn new( + proc: &HyperServerProc, + addr: SocketAddr, + ) -> Result> where Self: Sized; @@ -130,7 +130,10 @@ where + prosa::core::msg::Tvf + std::default::Default, { - fn new(proc: &HyperServerProc) -> Result> { + fn new( + proc: &HyperServerProc, + _addr: SocketAddr, + ) -> Result> { Ok(HelloHyperServerAdaptor { hello_msg: format!("Hello from {}", proc.name()), }) diff --git a/src/server/diagrams/adaptor.mmd b/src/server/diagrams/adaptor.mmd new file mode 100644 index 0000000..6a47a03 --- /dev/null +++ b/src/server/diagrams/adaptor.mmd @@ -0,0 +1,6 @@ +graph LR + IN[Input HTTP server] + ProSA[ProSA Hyper Processor] + + IN-- HTTP request (process_http_request) -->ProSA + ProSA-- HTTP response -->IN diff --git a/src/server/proc.rs b/src/server/proc.rs index 18d3c0e..419b1f6 100644 --- a/src/server/proc.rs +++ b/src/server/proc.rs @@ -1,8 +1,10 @@ use std::{env, sync::Arc, time::Duration}; use hyper::server::conn::{http1, http2}; -use hyper_util::rt::{TokioExecutor, TokioIo}; -use opentelemetry::KeyValue; +use hyper_util::{ + rt::{TokioExecutor, TokioIo}, + server::graceful::GracefulShutdown, +}; use prosa::{ core::{ adaptor::Adaptor, @@ -12,17 +14,19 @@ use prosa::{ service::ServiceError, }, event::pending::PendingMsgs, - io::{SslConfig, listener::ListenerSetting, url_is_ssl}, + io::listener::ListenerSetting, + otel::KeyValue, + tracing::{debug, info, warn}, }; use serde::{Deserialize, Serialize}; -use tokio::sync::mpsc; -use tracing::{debug, info, warn}; +use tokio::{sync::mpsc, task::JoinHandle}; use url::Url; -use crate::{H2, server::service::HyperService}; - -use super::adaptor::HyperServerAdaptor; +use crate::{ + H2, + server::{adaptor::HyperServerAdaptor, service::HyperService}, +}; /// Hyper server processor settings #[proc_settings] @@ -87,12 +91,28 @@ where + std::fmt::Debug + prosa::core::msg::Tvf + std::default::Default, - A: 'static + Adaptor + HyperServerAdaptor + Clone + std::marker::Send + std::marker::Sync, + A: 'static + Adaptor + HyperServerAdaptor + std::marker::Send + std::marker::Sync, { /// Main loop of the processor async fn internal_run(&mut self) -> Result<(), Box> { - // Initiate an adaptor for the hyper server processor - let adaptor = A::new(self)?; + // Force default protocol to HTTP2 for SSL + self.settings + .listener + .set_alpn(vec!["h2".into(), "http/1.1".into()]); + + // The listener is shared with every task that handshakes a client, so it can't be replaced + // to serve a new certificate. `bind_raw` leaves the SSL parameters out of it and hands them + // over instead, which makes the handshaker below the only copy and a rotation a plain + // assignment. `None` when the processor listens without SSL + let (bound_listener, mut handshaker) = self.settings.listener.bind_raw().await?; + let local_addr = bound_listener.local_addr()?; + let mut listener = Some(Arc::new(bound_listener)); + info!("Listening on {local_addr}"); + + // Initiate an adaptor for the hyper server processor. + // The very same instance is shared with every `HyperService`, so a configuration reload + // through `Adaptor::reload_config` is seen by the requests being served + let adaptor = Arc::new(A::new(self, local_addr)?); // Add proc main queue (id: 0) self.proc.add_proc().await?; @@ -103,17 +123,6 @@ where // Declare a list for pending HTTP request let mut pending_req = PendingMsgs::, M>::default(); - // Set default protocol to HTTP2 - if url_is_ssl(&self.settings.listener.url) { - if let Some(ssl) = self.settings.listener.ssl.as_mut() { - ssl.set_alpn(vec!["h2".into(), "http/1.1".into()]); - } else { - let mut ssl = SslConfig::default(); - ssl.set_alpn(vec!["h2".into(), "http/1.1".into()]); - self.settings.listener.ssl = Some(ssl); - } - } - // Meter to log HTTP reponses let meter = self.get_proc_param().meter("hyper_server"); let observable_http_counter = meter @@ -125,10 +134,16 @@ where .with_description("Hyper HTTP server socket counter") .build(); - let listener = Arc::new(self.settings.listener.bind().await?); - let service_adaptor = Arc::new(adaptor.clone()); - info!("Listening on {:?}", listener.local_addr()); + // `Some` while the processor serves. Taken when it is asked to stop, to signal every open + // connection to answer what it has in flight and then close + let mut graceful = Some(GracefulShutdown::new()); + + // `None` until the processor is asked to stop, then holds the task that drains the connections + let mut draining: Option> = None; + loop { + // Clone the listener so the configuration reload can swap it while an accept is pending + let accept_listener = listener.clone(); tokio::select! { Some(msg) = self.internal_rx_queue.recv() => { match msg { @@ -149,14 +164,72 @@ where let _ = hyper_err_msg.return_error_to_sender(err_msg.take_data(), err_msg.into_err()); } } - InternalMsg::Command(_) => todo!(), - InternalMsg::Config => todo!(), InternalMsg::Service(table) => self.service = table, + InternalMsg::Config(config) => { + // A reload landing while the processor drains would bind a listener it will never accept from + if let Some(mut settings) = graceful.is_some() + .then(|| config.reload_proc::(self.proc.as_ref(), adaptor.as_ref())) + .flatten() + { + // Normalize the ALPN as done at startup before comparing + settings.listener.set_alpn(vec!["h2".into(), "http/1.1".into()]); + + // Only listening somewhere else needs a new socket. Turning SSL on, + // off, or rotating a certificate is served on the socket that is + // already bound, by the handshaker rebuilt below + if self.settings.listener.needs_rebind(&settings.listener) { + match settings.listener.bind_raw().await { + Ok((new_listener, new_handshaker)) => { + let local_addr = new_listener.local_addr(); + handshaker = new_handshaker; + listener = Some(Arc::new(new_listener)); + match local_addr { + Ok(addr) => info!("Reload the Hyper server processor configuration, listening on {addr}"), + Err(e) => info!("Reload the Hyper server processor configuration, can't read the address it bound: {e}"), + } + } + // An address the processor can't bind is no reason to lose the + // one it serves on, so keep the listener and the settings that describe it + Err(e) => { + warn!("Can't listen on {}, keep the previous address: {e}", settings.listener.get_safe_url()); + settings.listener = self.settings.listener.clone(); + } + } + } else { + // Built again whatever the configuration says, because it holds + // the path of the certificate and not the certificate: a renewal + // that rewrites the file in place leaves the two configurations + // equal, so comparing them would skip exactly the reload this is + // for. One file read, and neither the socket nor an established + // connection is touched + match settings.listener.build_handshaker().await { + Ok(new_handshaker) => { + handshaker = new_handshaker; + info!("Reload the Hyper server processor configuration, serving {}", settings.listener.get_safe_url()); + } + // The processor keeps serving the certificate it has, so the + // settings have to keep describing it + Err(e) => { + warn!("Can't serve the certificate of {}, keep the previous one: {e}", settings.listener.get_safe_url()); + settings.listener = self.settings.listener.clone(); + } + } + } + + // The service timeout is picked up by the next request + self.settings = settings; + } + } InternalMsg::Shutdown => { - adaptor.terminate(); - self.proc.remove_proc(None).await?; - warn!("The Hyper server processor will shut down"); - return Ok(()); + // Release the port right away. A listener left bound but never accepted from + // would have the kernel complete handshakes into the accept queue, so a new + // client would wait out the whole drain only to be reset + listener = None; + + if let Some(graceful) = graceful.take() { + warn!("The Hyper server processor stops accepting and drains its connections"); + draining = Some(tokio::task::spawn(graceful.shutdown())); + } } } }, @@ -166,8 +239,17 @@ where { let request = RequestMsg::new(http_msg.get_service().clone(), http_msg_data, self.proc.get_service_queue().clone()); let request_id = request.get_id(); - service.proc_queue.send(InternalMsg::Request(request)).await?; - pending_req.push_with_id(request_id, http_msg, self.settings.service_timeout); + + // A processor that stopped since the service table was received only concerns + // this request. Answering it keeps the ones already in flight alive + if let Err(e) = service.proc_queue.send(InternalMsg::Request(request)).await { + warn!(parent: http_msg.get_span(), code = "503", "hyper::server::Msg"); + debug!("Can't reach the service {}: {e}", http_msg.get_service()); + let service_name = http_msg.get_service().clone(); + let _ = http_msg.return_error_to_sender(None, ServiceError::UnableToReachService(service_name)); + } else { + pending_req.push_with_id(request_id, http_msg, self.settings.service_timeout); + } } else { warn!( parent: http_msg.get_span(), @@ -179,16 +261,34 @@ where let _ = http_msg.return_error_to_sender(data, ServiceError::UnableToReachService(service_name)); } }, - accept_result = listener.accept_raw() => { + Some(accept_result) = async { + match &accept_listener { + Some(listener) => Some(listener.accept_raw().await), + None => None, + } + }, if accept_listener.is_some() => { let (stream, addr) = accept_result?; - let listener = listener.clone(); - let service_adaptor = service_adaptor.clone(); + // The watcher is taken before the connection is spawned, so a shutdown asked + // in between is not missed + let Some(watcher) = graceful.as_ref().map(GracefulShutdown::watcher) else { + continue; + }; + + // Owned snapshot of the SSL parameters, so a rotation doesn't wait for the + // handshake and the client is served the certificate of its accept + let handshaker = handshaker.clone(); + let service_adaptor = adaptor.clone(); let http_tx = http_tx.clone(); let http_counter = observable_http_counter.clone(); let http_socket = observable_http_socket.clone(); tokio::task::spawn(async move { - match listener.handshake(stream).await { + let handshake = match handshaker { + Some(handshaker) => handshaker.handshake(stream).await, + None => Ok(stream), + }; + + match handshake { Ok(stream) => { let is_http2 = stream.selected_alpn_check(|alpn| alpn == H2); @@ -197,21 +297,21 @@ where let io = TokioIo::new(stream); let service = HyperService::new(service_adaptor, http_tx, http_counter); if is_http2 { - if let Err(err) = http2::Builder::new(TokioExecutor::new()) - .serve_connection( + if let Err(err) = watcher.watch( + http2::Builder::new(TokioExecutor::new()).serve_connection( io, service, ) - .await + ).await { warn!("Failed to serve http/2 connection[{addr}]: {err:?}"); } - } else if let Err(err) = http1::Builder::new() - .serve_connection( + } else if let Err(err) = watcher.watch( + http1::Builder::new().serve_connection( io, service, ) - .await + ).await { warn!("Failed to serve http/1 connection[{addr}]: {err:?}"); } @@ -224,6 +324,13 @@ where debug!("Connection closed {addr}"); }); }, + // Every connection has been answered and closed, nothing is left to serve + Some(_) = async { + match draining.as_mut() { + Some(drain) => Some(drain.await), + None => None, + } + }, if draining.is_some() => break, Some(mut msg) = pending_req.pull(), if !pending_req.is_empty() => { warn!(parent: msg.get_span(), "Timeout message {:?}", msg); let data = msg.take_data(); @@ -238,5 +345,11 @@ where }, } } + + adaptor.terminate(); + self.proc.remove_proc(None).await?; + warn!("The Hyper server processor is shut down"); + + Ok(()) } } diff --git a/src/server/service.rs b/src/server/service.rs index ff0e30f..48ff7d2 100644 --- a/src/server/service.rs +++ b/src/server/service.rs @@ -1,26 +1,20 @@ //! Hyper service definition -use std::convert::Infallible; -use std::future::Future; -use std::pin::Pin; -use std::sync::Arc; +use std::{convert::Infallible, future::Future, pin::Pin, sync::Arc}; use bytes::Bytes; use http::StatusCode; -use http_body_util::combinators::BoxBody; -use http_body_util::{Empty, Full}; -use hyper::service::Service; -use hyper::{Request, Response}; -use opentelemetry::KeyValue; -use opentelemetry::metrics::Counter; -use prosa::core::msg::{InternalMsg, Msg, RequestMsg}; +use http_body_util::{Empty, Full, combinators::BoxBody}; +use hyper::{Request, Response, service::Service}; +use prosa::{ + core::msg::{InternalMsg, Msg, RequestMsg}, + otel::{KeyValue, metrics::Counter}, +}; use tokio::sync::{mpsc, oneshot}; -use crate::{HttpError, hyper_version_str}; +use crate::{HttpError, hyper_version_str, server::adaptor::HyperServerAdaptor}; -use super::adaptor::HyperServerAdaptor; - -#[derive(Debug, Clone)] +#[derive(Debug)] /// Struct to define parameters for a service (HTTP server) pub(crate) struct HyperService where @@ -41,7 +35,7 @@ where impl HyperService where - A: 'static + HyperServerAdaptor + Clone + std::marker::Sync + std::marker::Send, + A: 'static + HyperServerAdaptor + std::marker::Sync + std::marker::Send, M: 'static + std::marker::Send + std::marker::Sync @@ -144,7 +138,7 @@ where impl Service> for HyperService where - A: 'static + HyperServerAdaptor + Clone + std::marker::Sync + std::marker::Send, + A: 'static + HyperServerAdaptor + std::marker::Sync + std::marker::Send, M: 'static + std::marker::Send + std::marker::Sized diff --git a/src/tests.rs b/src/tests.rs index 28382c8..840bcf9 100644 --- a/src/tests.rs +++ b/src/tests.rs @@ -30,12 +30,17 @@ pub(crate) struct HttpTestSettings { } impl HttpTestSettings { + /// Passphrase protecting the private keys [`HttpTestSettings::create_server_cert`] writes. + /// + /// A test restating the SSL configuration in a reload has to say it again + pub(crate) const PASSPHRASE: &str = "prosa_test"; + /// Method to create private key and certificate for a server pub(crate) fn create_server_cert( key_path: String, cert_path: String, ) -> Result { - const PASSPHRASE: &str = "prosa_test"; + const PASSPHRASE: &str = HttpTestSettings::PASSPHRASE; let mut group = EcGroup::from_curve_name(Nid::X9_62_PRIME256V1)?; group.set_asn1_flag(Asn1Flag::NAMED_CURVE); @@ -133,6 +138,101 @@ impl HttpTestSettings { } } +/// Ports the Hyper server processors of the tests bound, by processor name. +/// +/// The tests share a process, so a name must belong to a single processor of a single test: +/// [`bound_url`] reads the first port published under a name, and a processor publishes once +static BOUND_PORTS: std::sync::Mutex> = std::sync::Mutex::new(Vec::new()); + +/// Publish where a Hyper server processor bound, called by the test adaptors. +/// +/// The tests listen on the port 0 so that the operating system picks a port nothing else holds, +/// which means the port is only known once bound. The processor hands its address to the adaptor, +/// and the adaptor leaves it here for the test to pick up +pub(crate) fn set_bound_port(proc_name: &str, port: u16) { + BOUND_PORTS + .lock() + .expect("Bound test ports should be writable") + .push((proc_name.to_string(), port)); +} + +/// Wait for the Hyper server processor `proc_name` to bind, and return `url` pointing at it. +/// +/// The processor publishes its address before serving anything, so this also stands for waiting on +/// the listener +pub(crate) async fn bound_url(proc_name: &str, url: &url::Url) -> url::Url { + let deadline = tokio::time::Instant::now() + TEST_TIMEOUT; + loop { + let bound_port = BOUND_PORTS + .lock() + .expect("Bound test ports should be readable") + .iter() + .find(|(name, _)| name == proc_name) + .map(|(_, port)| *port); + + if let Some(port) = bound_port { + let mut url = url.clone(); + url.set_port(Some(port)) + .expect("Test URL should accept a port"); + return url; + } + + assert!( + tokio::time::Instant::now() < deadline, + "The Hyper server processor {proc_name} should have bound" + ); + tokio::time::sleep(POLL_INTERVAL).await; + } +} + +/// Interval between two polls of [`wait_for`] +const POLL_INTERVAL: std::time::Duration = std::time::Duration::from_millis(10); + +/// Time a test waits for something it expects to happen. +/// +/// Generous on purpose: it is only ever reached by a test that is going to fail anyway +pub(crate) const TEST_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(30); + +/// Poll `condition` until it holds, returning `false` if it still doesn't after `timeout`. +/// +/// Tests wait on the state they expect instead of sleeping a fixed budget, so a loaded machine +/// makes them slower rather than making them fail +pub(crate) async fn wait_for(timeout: std::time::Duration, mut condition: F) -> bool +where + F: AsyncFnMut() -> bool, +{ + let deadline = tokio::time::Instant::now() + timeout; + loop { + if condition().await { + return true; + } else if tokio::time::Instant::now() >= deadline { + return false; + } + + tokio::time::sleep(POLL_INTERVAL).await; + } +} + +/// Wait for `counter` to stop moving for a whole `quiet` window, and return the value it settled on. +/// +/// Traffic doesn't stop the instant a socket is retired, it still answers what it had in flight +pub(crate) async fn wait_for_quiescence( + counter: &std::sync::atomic::AtomicU32, + quiet: std::time::Duration, + timeout: std::time::Duration, +) -> u32 { + use std::sync::atomic::Ordering; + + let deadline = tokio::time::Instant::now() + timeout; + loop { + let count = counter.load(Ordering::SeqCst); + tokio::time::sleep(quiet).await; + if count == counter.load(Ordering::SeqCst) || tokio::time::Instant::now() >= deadline { + return counter.load(Ordering::SeqCst); + } + } +} + #[allow(clippy::module_inception)] #[cfg(all(feature = "server", feature = "client"))] mod tests { @@ -145,8 +245,9 @@ mod tests { error::ProcError, main::{MainProc, MainRunnable as _}, msg::Tvf, - proc::{Proc, ProcConfig as _}, + proc::{Proc, ProcBusParam as _, ProcConfig as _}, service::ServiceError, + settings::ProsaConfig, }, inj::{adaptor::InjAdaptor, proc::InjProc}, stub::{adaptor::StubAdaptor, proc::StubProc}, @@ -160,7 +261,10 @@ mod tests { }; use std::{ env, fs, io, - sync::atomic::{AtomicU32, Ordering}, + sync::{ + Arc, + atomic::{AtomicBool, AtomicU32, Ordering}, + }, }; use tokio::{runtime, time}; use url::Url; @@ -172,17 +276,82 @@ mod tests { adaptor::{HyperServerAdaptor, default_srv_error_response}, proc::HyperServerProc, }, - tests::HttpTestSettings, + tests::{ + HttpTestSettings, TEST_TIMEOUT, bound_url, set_bound_port, wait_for, + wait_for_quiescence, + }, }; const WAIT_TIME: time::Duration = time::Duration::from_secs(1); - static COUNTER: [AtomicU32; 3] = [ + /// Prefix of the user agent the Hyper client sets on every request + const CLIENT_USER_AGENT: &str = concat!(env!("CARGO_PKG_NAME"), "/"); + + /// What the stub answers, and therefore what a complete round trip brings back + const STUB_RESPONSE: &str = "Hello from the stub!"; + + /// Server processor of each test. + /// + /// The tests share a process, so they name their processors apart to read back the port of + /// their own listener and not of somebody else's + const SERVER_PROCS: [&str; 9] = [ + "HTTP_SERVER_PROC", + "HTTPS_SERVER_PROC", + "H2_SERVER_PROC", + "RELOAD_SERVER_PROC", + "RECONNECT_SERVER_PROC", + // The next three tests answer from a raw socket, they run no server processor + "SPLIT_BODY_SERVER_PROC", + "GRACEFUL_SERVER_PROC", + "BACKOFF_SERVER_PROC", + "PANIC_SERVER_PROC", + ]; + + /// Client processor of each test, at the index of the counter that test uses + const CLIENT_PROCS: [&str; 9] = [ + "HTTP_CLIENT_PROC", + "HTTPS_CLIENT_PROC", + "H2_CLIENT_PROC", + "RELOAD_CLIENT_PROC", + "RECONNECT_CLIENT_PROC", + "SPLIT_BODY_CLIENT_PROC", + "GRACEFUL_CLIENT_PROC", + "BACKOFF_CLIENT_PROC", + "PANIC_CLIENT_PROC", + ]; + + static COUNTER: [AtomicU32; 9] = [ AtomicU32::new(0), // HTTP AtomicU32::new(0), // HTTPS AtomicU32::new(0), // HTTP/2 + AtomicU32::new(0), // HTTP, configuration reload + AtomicU32::new(0), // HTTP, socket reconnection + AtomicU32::new(0), // HTTP, response body split across two writes + AtomicU32::new(0), // HTTP, graceful shutdown of the client socket + AtomicU32::new(0), // HTTP, backoff on a backend that closes right away + AtomicU32::new(0), // HTTP, socket task panicking in the adaptor ]; + /// Test whose client socket is stopped in the middle of a transaction + const GRACEFUL_TEST_TYPE: usize = 6; + + /// Requests the socket of [`GRACEFUL_TEST_TYPE`] handed to Hyper, and the ones Hyper answered. + /// + /// A transaction the socket started must come back, ProSA stopping in the middle of it included + static GRACEFUL_STARTED: AtomicU32 = AtomicU32::new(0); + static GRACEFUL_ANSWERED: AtomicU32 = AtomicU32::new(0); + + /// Test whose client adaptor panics once, in the middle of a socket task + const PANIC_TEST_TYPE: usize = 8; + + /// Set to make the client adaptor of [`PANIC_TEST_TYPE`] panic on the next response it reads + static PANIC_ARMED: AtomicBool = AtomicBool::new(false); + + /// Number of adaptors the client processor of [`PANIC_TEST_TYPE`] built. + /// + /// One per run of its main loop, so it counts the times the processor was restarted + static PANIC_CLIENT_ADAPTORS: AtomicU32 = AtomicU32::new(0); + #[derive(Adaptor, Default, Clone, Copy)] struct TestAdaptor { test_type: u64, @@ -213,10 +382,10 @@ mod tests { .and_then(|b| request.get_string(2).map(|ua| (b, ua))) { Ok((content, user_agent)) => { - if !content.starts_with("Hello") || !user_agent.starts_with("ProSA-Hyper/") { - return Err(ServiceError::ProtocolError( - "Invalid request content".into(), - )) + if !content.starts_with("Hello") || !user_agent.starts_with(CLIENT_USER_AGENT) { + return Err(ServiceError::ProtocolError(format!( + "Invalid request content: {content:?} from {user_agent:?}" + ))) .into(); } } @@ -224,7 +393,7 @@ mod tests { } let mut srv_req = M::default(); - srv_req.put_string(1, "Hello from the stub!"); + srv_req.put_string(1, STUB_RESPONSE); Ok(srv_req).into() } } @@ -242,10 +411,14 @@ mod tests { { fn new( proc: &crate::server::proc::HyperServerProc, + addr: prosa::io::SocketAddr, ) -> Result> where Self: Sized, { + // The listener is configured on the port 0, this is where the test learns where it landed + set_bound_port(proc.name(), addr.port()); + let test_type = match proc.settings.listener.url.scheme() { "http" => 0, "https" => 1, @@ -337,6 +510,17 @@ mod tests { response: M, _service_name: &str, ) -> Result<(), Box> { + // The counters must only count a complete round trip. The client adaptor turns any HTTP + // response into a message, so without this an error status reads as a success + let content = response + .get_string(1) + .map_err(|e| ServiceError::ProtocolError(e.to_string()))?; + if !content.starts_with(STUB_RESPONSE) { + return Err(Box::new(ServiceError::ProtocolError(format!( + "Unexpected response content: {content:?}" + )))); + } + match response .get_unsigned(10) .map_err(|e| ServiceError::ProtocolError(e.to_string()))? @@ -353,6 +537,30 @@ mod tests { // HTTP/2 COUNTER[2].fetch_add(1, Ordering::SeqCst); } + 3 => { + // HTTP, configuration reload + COUNTER[3].fetch_add(1, Ordering::SeqCst); + } + 4 => { + // HTTP, socket reconnection + COUNTER[4].fetch_add(1, Ordering::SeqCst); + } + 5 => { + // HTTP, response body split across two writes + COUNTER[5].fetch_add(1, Ordering::SeqCst); + } + 6 => { + // HTTP, graceful shutdown of the client socket + COUNTER[6].fetch_add(1, Ordering::SeqCst); + } + 7 => { + // HTTP, backoff on a backend that closes right away + COUNTER[7].fetch_add(1, Ordering::SeqCst); + } + 8 => { + // HTTP, socket task panicking in the adaptor + COUNTER[8].fetch_add(1, Ordering::SeqCst); + } _ => { return Err(Box::new(ServiceError::ProtocolError( "Invalid response type".into(), @@ -379,19 +587,22 @@ mod tests { where Self: Sized, { - let test_type = match proc.settings.backends.first().map(|b| b.url.scheme()) { - Some("http") => 0, - Some("https") => 1, - Some("h2") => 2, - _ => { - return Err(Box::new(ConfigError::WrongValue( - "HyperClientSettings::scheme".into(), - "Unsupported scheme".into(), - ))); - } + // The backend ports come from the operating system, so the test a client belongs to is + // read from its name + let Some(test_type) = CLIENT_PROCS.iter().position(|name| *name == proc.name()) else { + return Err(Box::new(ConfigError::WrongValue( + "HyperClientProc::name".into(), + proc.name().into(), + ))); }; - Ok(TestAdaptor { test_type }) + if test_type == PANIC_TEST_TYPE { + PANIC_CLIENT_ADAPTORS.fetch_add(1, Ordering::SeqCst); + } + + Ok(TestAdaptor { + test_type: test_type as u64, + }) } fn process_srv_request( @@ -408,6 +619,13 @@ mod tests { .uri(socket_url.as_str()) .header(hyper::header::USER_AGENT, PRODUCT_VERSION_HEADER) .body(BoxBody::new(Full::new(Bytes::from(body.into_owned())))) + .inspect(|_| { + // The socket sends the request right after this, so the transaction is in + // flight from here until the response comes back + if self.test_type == GRACEFUL_TEST_TYPE as u64 { + GRACEFUL_STARTED.fetch_add(1, Ordering::SeqCst); + } + }) .map_err(|e| { ServiceError::ProtocolError(format!("Failed to build request: {}", e)) }), @@ -421,11 +639,22 @@ mod tests { &self, resp: Result, hyper::Error>, ) -> Result { + // Adaptors are user code and this one is asked to fail the way user code does. It runs + // inside the socket task, so the panic takes that task down with the request it holds + if self.test_type == PANIC_TEST_TYPE as u64 && PANIC_ARMED.swap(false, Ordering::SeqCst) + { + panic!("The client adaptor of the panic test was armed to panic"); + } + let http_body = resp.map_err(|e| ServiceError::ProtocolError(format!("HTTP error: {}", e)))?; if let Ok(body) = http_body.into_body().collect().await && let Ok(body_str) = String::from_utf8(body.to_bytes().to_vec()) { + if self.test_type == GRACEFUL_TEST_TYPE as u64 { + GRACEFUL_ANSWERED.fetch_add(1, Ordering::SeqCst); + } + let mut srv_req = M::default(); srv_req.put_string(1, body_str); srv_req.put_unsigned(10, self.test_type); @@ -438,7 +667,9 @@ mod tests { } } - async fn run_test(settings: HttpTestSettings, test_type: u64) -> Result<(), io::Error> { + async fn run_test(mut settings: HttpTestSettings, test_type: usize) -> Result<(), io::Error> { + let server_url = settings.server.listener.url.clone(); + // Create bus and main processor let (bus, main) = MainProc::::create(&settings, Some(4)); @@ -470,19 +701,23 @@ mod tests { // Launch an HTTP server processor let http_server_proc = HyperServerProc::::create( 2, - String::from("HTTP_SERVER_PROC"), + String::from(SERVER_PROCS[test_type]), bus.clone(), settings.server, ); Proc::::run(http_server_proc)?; - // Wait for processor to start - std::thread::sleep(WAIT_TIME); + // The listener is on the port 0, so the client can only be pointed at the server once the + // processor has published where it bound + let server_url = bound_url(SERVER_PROCS[test_type], &server_url).await; + for backend in &mut settings.client.backends { + backend.url = server_url.clone(); + } // Launch an HTTP client processor let http_client_proc = HyperClientProc::::create( 3, - String::from("HTTP_CLIENT_PROC"), + String::from(CLIENT_PROCS[test_type]), bus.clone(), settings.client, ); @@ -497,15 +732,18 @@ mod tests { ); Proc::::run(http_inj_proc)?; - // Wait for processor to finish processing - std::thread::sleep(WAIT_TIME); + // Wait for a full loop: injector, client, HTTP, server and stub + let responded = wait_for(TEST_TIMEOUT, async || { + COUNTER[test_type].load(Ordering::SeqCst) > 0 + }) + .await; bus.stop("ProSA HTTP client server unit test end".into()) .await .map_err(io::Error::other)?; assert!( - COUNTER[test_type as usize].load(Ordering::SeqCst) > 0, + responded, "No response received for test type {}", test_type ); @@ -518,7 +756,7 @@ mod tests { #[tokio::test] async fn http_client_server() { let test_settings = HttpTestSettings::new( - Url::parse("http://localhost:48080").expect("HTTP client/server URL should be valid"), + Url::parse("http://localhost:0").expect("HTTP client/server URL should be valid"), None, None, ); @@ -563,7 +801,7 @@ mod tests { client_ssl_config.set_store(client_ssl_store); let test_settings = HttpTestSettings::new( - Url::parse("https://localhost:48443").expect("HTTPS client/server URL should be valid"), + Url::parse("https://localhost:0").expect("HTTPS client/server URL should be valid"), Some(server_ssl_config), Some(client_ssl_config), ); @@ -582,7 +820,7 @@ mod tests { let key_path = prosa_temp_dir.join("prosa_h2.key"); let cert_path = prosa_temp_dir.join("prosa_h2.pem"); - let mut server_ssl_config = HttpTestSettings::create_server_cert( + let server_ssl_config = HttpTestSettings::create_server_cert( key_path .as_os_str() .to_str() @@ -595,8 +833,6 @@ mod tests { .into(), ) .expect("Server certificate should be created"); - // Need to set the ALPN for server because of inline configuration @see TargetSetting::new - server_ssl_config.set_alpn(vec!["h2".into()]); let client_ssl_store = Store::File { path: format!( @@ -612,7 +848,7 @@ mod tests { client_ssl_config.set_alpn(vec!["h2".into()]); let test_settings = HttpTestSettings::new( - Url::parse("h2://localhost:49443").expect("HTTP2 client/server URL should be valid"), + Url::parse("h2://localhost:0").expect("HTTP2 client/server URL should be valid"), Some(server_ssl_config), Some(client_ssl_config), ); @@ -620,4 +856,704 @@ mod tests { // Run a ProSA to test assert!(run_test(test_settings, 2).await.is_ok()); } + + /// Build a ProSA configuration that points the Hyper client processor at the given backend + fn client_backend_config(proc_name: &str, backend_url: &Url) -> ProsaConfig { + ProsaConfig::from_config( + config::Config::builder() + .add_source(config::File::from_str( + &format!( + "[{proc_name}]\nservice_name = \"HTTP_CLIENT_SRV\"\nbackends = [{{ url = \"{backend_url}\" }}]\n" + ), + config::FileFormat::Toml, + )) + .build() + .expect("Hyper client configuration should be valid"), + ) + .expect("Reloaded ProSA configuration should be valid") + } + + #[tokio::test] + async fn client_config_reload() { + const TEST_TYPE: usize = 3; + // The port 1 is privileged, so nothing can be listening there and the client can't reach + // any backend once reloaded on it + let dead_url = Url::parse("http://localhost:1").expect("Dead backend URL should be valid"); + + let mut settings = HttpTestSettings::new( + Url::parse("http://localhost:0").expect("Backend URL should be valid"), + None, + None, + ); + let server_url = settings.server.listener.url.clone(); + + // Create bus and main processor + let (bus, main) = MainProc::::create(&settings, Some(4)); + + // The main task must run to broadcast the configuration to the processors + let main_handle = std::thread::Builder::new() + .name("main".to_string()) + .spawn(move || { + runtime::Builder::new_multi_thread() + .worker_threads(1) + .enable_all() + .thread_name("main") + .build() + .expect("Runtime should be valid") + .block_on(async { + main.run().await; + }) + }) + .expect("Main thread should be spawned"); + + // Launch stub to respond to the HTTP server + let http_server_stub = StubProc::::create( + 1, + String::from("HTTP_SERVER_STUB"), + bus.clone(), + settings.stub, + ); + Proc::::run(http_server_stub).expect("Stub processor should run"); + + // Launch an HTTP server processor + let http_server_proc = HyperServerProc::::create( + 2, + String::from(SERVER_PROCS[TEST_TYPE]), + bus.clone(), + settings.server, + ); + Proc::::run(http_server_proc).expect("Hyper server processor should run"); + + // The listener is on the port 0, so the backend is only known once the processor bound + let backend_url = bound_url(SERVER_PROCS[TEST_TYPE], &server_url).await; + for backend in &mut settings.client.backends { + backend.url = backend_url.clone(); + } + + // Launch an HTTP client processor + let http_client_proc = HyperClientProc::::create( + 3, + String::from(CLIENT_PROCS[TEST_TYPE]), + bus.clone(), + settings.client, + ); + Proc::::run(http_client_proc).expect("Hyper client processor should run"); + + // Launch an HTTP injector processor + let http_inj_proc = InjProc::::create( + 4, + String::from("HTTP_INJ_PROC"), + bus.clone(), + settings.inj, + ); + Proc::::run(http_inj_proc).expect("Inj processor should run"); + + assert!( + wait_for(TEST_TIMEOUT, async || COUNTER[TEST_TYPE] + .load(Ordering::SeqCst) + > 0) + .await, + "No response received before the configuration reload" + ); + + // Reload the client on a backend that can't be reached, retiring the current sockets + bus.update_config(Arc::new(client_backend_config( + CLIENT_PROCS[TEST_TYPE], + &dead_url, + ))) + .await + .expect("ProSA configuration should be updated"); + + // The retired sockets still answer what they had in flight, so wait for the traffic to stop + // before taking the reference count + let retired_count = wait_for_quiescence(&COUNTER[TEST_TYPE], WAIT_TIME, TEST_TIMEOUT).await; + time::sleep(WAIT_TIME).await; + assert_eq!( + retired_count, + COUNTER[TEST_TYPE].load(Ordering::SeqCst), + "The sockets of the previous backend should have been retired" + ); + + // Reload the client back on the reachable backend + bus.update_config(Arc::new(client_backend_config( + CLIENT_PROCS[TEST_TYPE], + &backend_url, + ))) + .await + .expect("ProSA configuration should be updated"); + + assert!( + wait_for(TEST_TIMEOUT, async || COUNTER[TEST_TYPE] + .load(Ordering::SeqCst) + > retired_count) + .await, + "No response received after the configuration reload" + ); + + bus.stop("ProSA HTTP client configuration reload unit test end".into()) + .await + .expect("ProSA should stop"); + + // Wait on main task to end + let _ = main_handle.join(); + } + + /// A client socket that can't connect must keep retrying instead of leaving the pool empty + #[tokio::test] + async fn client_socket_reconnect() { + const TEST_TYPE: usize = 4; + + // The client has to start while the backend is down, so this is the one address that can't + // come from the processor. Binding and closing right away is only a way to have the + // operating system name a port, the server processor takes it further down + let backend_url = { + let listener = + std::net::TcpListener::bind("127.0.0.1:0").expect("A port should be free"); + let addr = listener + .local_addr() + .expect("A bound listener should have an address"); + Url::parse(&format!("http://{addr}")).expect("Backend URL should be valid") + }; + let settings = HttpTestSettings::new(backend_url, None, None); + + // Create bus and main processor + let (bus, main) = MainProc::::create(&settings, Some(4)); + let main_handle = std::thread::Builder::new() + .name("main".to_string()) + .spawn(move || { + runtime::Builder::new_multi_thread() + .worker_threads(1) + .enable_all() + .thread_name("main") + .build() + .expect("Runtime should be valid") + .block_on(async { + main.run().await; + }) + }) + .expect("Main thread should be spawned"); + + // Launch stub to respond to the HTTP server + let http_server_stub = StubProc::::create( + 1, + String::from("HTTP_SERVER_STUB"), + bus.clone(), + settings.stub, + ); + Proc::::run(http_server_stub).expect("Stub processor should run"); + + // Launch the client and the injector while nothing listens on the backend port + let http_client_proc = HyperClientProc::::create( + 2, + String::from(CLIENT_PROCS[TEST_TYPE]), + bus.clone(), + settings.client, + ); + Proc::::run(http_client_proc).expect("Hyper client processor should run"); + + let http_inj_proc = InjProc::::create( + 3, + String::from("HTTP_INJ_PROC"), + bus.clone(), + settings.inj, + ); + Proc::::run(http_inj_proc).expect("Inj processor should run"); + + // The client sockets fail to connect and back off, none of them can serve anything + time::sleep(WAIT_TIME).await; + assert_eq!( + 0, + COUNTER[TEST_TYPE].load(Ordering::SeqCst), + "No response can be received while the backend is down" + ); + + // Bring the backend up, the sockets must find it on their next attempt + let http_server_proc = HyperServerProc::::create( + 4, + String::from(SERVER_PROCS[TEST_TYPE]), + bus.clone(), + settings.server, + ); + Proc::::run(http_server_proc).expect("Hyper server processor should run"); + + assert!( + wait_for(TEST_TIMEOUT, async || COUNTER[TEST_TYPE] + .load(Ordering::SeqCst) + > 0) + .await, + "The client sockets should have reconnected once the backend came up" + ); + + bus.stop("ProSA HTTP client socket reconnection unit test end".into()) + .await + .expect("ProSA should stop"); + + // Wait on main task to end + let _ = main_handle.join(); + } + + /// Answer requests with a body that cannot reach the client with the head. + /// + /// A real HTTP server splits a response whenever it is bigger than one write, which a stub + /// answering a short string never does. Hyper only hands over the bytes it has read, so the + /// socket has to keep driving its connection while the adaptor collects the body + async fn split_body_origin(listener: tokio::net::TcpListener) { + use tokio::io::AsyncWriteExt as _; + + raw_origin(listener, async |sock| { + // The head and the first half go out, then a pause long enough that the client cannot + // have them in the same read, then the rest + let (first, second) = STUB_RESPONSE.split_at(STUB_RESPONSE.len() / 2); + let head = format!( + "HTTP/1.1 200 OK\r\nContent-Length: {}\r\n\r\n{first}", + STUB_RESPONSE.len() + ); + if sock.write_all(head.as_bytes()).await.is_err() { + return false; + } + time::sleep(time::Duration::from_millis(50)).await; + sock.write_all(second.as_bytes()).await.is_ok() + }) + .await; + } + + /// Answer requests slowly enough that ProSA can be stopped while one is in flight + async fn slow_origin(listener: tokio::net::TcpListener) { + use tokio::io::AsyncWriteExt as _; + + raw_origin(listener, async |sock| { + time::sleep(SLOW_RESPONSE_TIME).await; + let response = format!( + "HTTP/1.1 200 OK\r\nContent-Length: {}\r\n\r\n{STUB_RESPONSE}", + STUB_RESPONSE.len() + ); + sock.write_all(response.as_bytes()).await.is_ok() + }) + .await; + } + + /// Serve `answer` on every request the client sends, until it goes away. + /// + /// A raw socket rather than a Hyper server, because these tests are about what the client does + /// with a response Hyper would never produce on its own. One connection at a time is enough, + /// they all configure a single socket + async fn raw_origin(listener: tokio::net::TcpListener, answer: F) + where + F: AsyncFn(&mut tokio::net::TcpStream) -> bool, + { + while let Ok((mut sock, _)) = listener.accept().await { + let mut buf = Vec::new(); + let mut chunk = [0u8; 1024]; + + // Keep the connection alive across requests, which is also what makes the client send + // a second request on a connection it already used + while read_request(&mut sock, &mut chunk, &mut buf).await && answer(&mut sock).await {} + } + } + + /// Consume one request from `sock`, answering `false` once the peer is gone + async fn read_request( + sock: &mut tokio::net::TcpStream, + chunk: &mut [u8], + buf: &mut Vec, + ) -> bool { + // The head first, then as many body bytes as it announces + let head_end = loop { + if let Some(end) = buf.windows(4).position(|w| w == b"\r\n\r\n") { + break end + 4; + } else if !read_more(sock, chunk, buf).await { + return false; + } + }; + let body_len = String::from_utf8_lossy(&buf[..head_end]) + .to_lowercase() + .split("content-length:") + .nth(1) + .and_then(|value| value.split("\r\n").next()) + .and_then(|value| value.trim().parse::().ok()) + .unwrap_or(0); + while buf.len() < head_end + body_len { + if !read_more(sock, chunk, buf).await { + return false; + } + } + buf.drain(..head_end + body_len); + + true + } + + /// Read the next bytes of a request into `buf`, answering `false` once the peer is gone + async fn read_more( + sock: &mut tokio::net::TcpStream, + chunk: &mut [u8], + buf: &mut Vec, + ) -> bool { + use tokio::io::AsyncReadExt as _; + + match sock.read(chunk).await { + Ok(0) | Err(_) => false, + Ok(read) => { + buf.extend_from_slice(&chunk[..read]); + true + } + } + } + + /// A response body that doesn't arrive with the head must still be read + #[tokio::test] + async fn client_split_body_response() { + const TEST_TYPE: usize = 5; + + let listener = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("A port should be free"); + let backend_url = Url::parse(&format!( + "http://{}", + listener + .local_addr() + .expect("A bound listener should have an address") + )) + .expect("Backend URL should be valid"); + tokio::spawn(split_body_origin(listener)); + + let settings = HttpTestSettings::new(backend_url, None, None); + + // Create bus and main processor + let (bus, main) = MainProc::::create(&settings, Some(4)); + let main_handle = std::thread::Builder::new() + .name("main".to_string()) + .spawn(move || { + runtime::Builder::new_multi_thread() + .worker_threads(1) + .enable_all() + .thread_name("main") + .build() + .expect("Runtime should be valid") + .block_on(async { + main.run().await; + }) + }) + .expect("Main thread should be spawned"); + + // Launch an HTTP client processor on the raw backend, and the injector that drives it + let http_client_proc = HyperClientProc::::create( + 1, + String::from(CLIENT_PROCS[TEST_TYPE]), + bus.clone(), + settings.client, + ); + Proc::::run(http_client_proc).expect("Hyper client processor should run"); + + let http_inj_proc = InjProc::::create( + 2, + String::from("HTTP_INJ_PROC"), + bus.clone(), + settings.inj, + ); + Proc::::run(http_inj_proc).expect("Inj processor should run"); + + // More than one, so a second request also goes out on a connection already used once + assert!( + wait_for(TEST_TIMEOUT, async || COUNTER[TEST_TYPE] + .load(Ordering::SeqCst) + > 1) + .await, + "The client should have read the split response bodies" + ); + + bus.stop("ProSA HTTP client split body unit test end".into()) + .await + .expect("ProSA should stop"); + + // Wait on main task to end + let _ = main_handle.join(); + } + + /// Long enough that ProSA can be stopped while the origin still owes a response + const SLOW_RESPONSE_TIME: time::Duration = time::Duration::from_millis(300); + + /// A transaction the socket started must be answered, even if ProSA stops in the middle of it + #[tokio::test] + async fn client_graceful_shutdown() { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("A port should be free"); + let backend_url = Url::parse(&format!( + "http://{}", + listener + .local_addr() + .expect("A bound listener should have an address") + )) + .expect("Backend URL should be valid"); + tokio::spawn(slow_origin(listener)); + + let settings = HttpTestSettings::new(backend_url, None, None); + + // Create bus and main processor + let (bus, main) = MainProc::::create(&settings, Some(4)); + let main_handle = std::thread::Builder::new() + .name("main".to_string()) + .spawn(move || { + runtime::Builder::new_multi_thread() + .worker_threads(1) + .enable_all() + .thread_name("main") + .build() + .expect("Runtime should be valid") + .block_on(async { + main.run().await; + }) + }) + .expect("Main thread should be spawned"); + + // Launch an HTTP client processor on the slow backend, and the injector that drives it + let http_client_proc = HyperClientProc::::create( + 1, + String::from(CLIENT_PROCS[GRACEFUL_TEST_TYPE]), + bus.clone(), + settings.client, + ); + Proc::::run(http_client_proc).expect("Hyper client processor should run"); + + let http_inj_proc = InjProc::::create( + 2, + String::from("HTTP_INJ_PROC"), + bus.clone(), + settings.inj, + ); + Proc::::run(http_inj_proc).expect("Inj processor should run"); + + // Let a full loop go through first, so what is asserted below is a socket that was working + assert!( + wait_for(TEST_TIMEOUT, async || COUNTER[GRACEFUL_TEST_TYPE] + .load(Ordering::SeqCst) + > 0) + .await, + "No response received before the shutdown" + ); + + // Stop ProSA while the origin owes a response, which is what the socket has to see through. + // The injector saturates the socket, so it is in flight for all but an instant of the time + assert!( + wait_for(TEST_TIMEOUT, async || { + GRACEFUL_STARTED.load(Ordering::SeqCst) > GRACEFUL_ANSWERED.load(Ordering::SeqCst) + }) + .await, + "A transaction should have been in flight" + ); + + bus.stop("ProSA HTTP client graceful shutdown unit test end".into()) + .await + .expect("ProSA should stop"); + + // Every transaction the socket handed to Hyper came back, the one it was in the middle of + // included. A retired socket starts no other, so a dropped one leaves the counts apart for + // good. Whether the answer still finds its requester is up to ProSA, which tells the + // injector to stop at the very same time + assert!( + wait_for(TEST_TIMEOUT, async || { + GRACEFUL_STARTED.load(Ordering::SeqCst) == GRACEFUL_ANSWERED.load(Ordering::SeqCst) + }) + .await, + "The socket dropped a transaction it had started: {} started, {} answered", + GRACEFUL_STARTED.load(Ordering::SeqCst), + GRACEFUL_ANSWERED.load(Ordering::SeqCst) + ); + + // Wait on main task to end + let _ = main_handle.join(); + } + + /// A backend that accepts and closes right away must be backed off from, like one that refuses. + /// + /// Nothing distinguishes the two for a caller, and a socket that treats the first as a healthy + /// connection that simply ended reconnects with no delay at all, at the speed of the machine + #[tokio::test] + async fn client_dead_backend_backoff() { + const TEST_TYPE: usize = 7; + /// Long enough for the backoff to have doubled a few times if it engages at all + const OBSERVE: time::Duration = time::Duration::from_secs(2); + /// Delays of 0, then 500, 1000 and 2000 ms by default, so attempts land at 0, 0.5, 1.5 and + /// 3.5 s and only three of them fall inside `OBSERVE`. One of slack for a loaded machine, + /// and low enough that a delay that stopped doubling would show up here + const MAX_ATTEMPTS: u32 = 4; + + static ACCEPTED: AtomicU32 = AtomicU32::new(0); + + let listener = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("A port should be free"); + let backend_url = Url::parse(&format!( + "http://{}", + listener + .local_addr() + .expect("A bound listener should have an address") + )) + .expect("Backend URL should be valid"); + + // Accept the connection and drop it, which is what an overloaded backend or a load balancer + // with no healthy upstream does. The TCP connect succeeds, so the socket has to notice on + // its own that it got nothing out of it + tokio::spawn(async move { + while let Ok((sock, _)) = listener.accept().await { + ACCEPTED.fetch_add(1, Ordering::SeqCst); + drop(sock); + } + }); + + let settings = HttpTestSettings::new(backend_url, None, None); + + // Create bus and main processor + let (bus, main) = MainProc::::create(&settings, Some(4)); + let main_handle = std::thread::Builder::new() + .name("main".to_string()) + .spawn(move || { + runtime::Builder::new_multi_thread() + .worker_threads(1) + .enable_all() + .thread_name("main") + .build() + .expect("Runtime should be valid") + .block_on(async { + main.run().await; + }) + }) + .expect("Main thread should be spawned"); + + let http_client_proc = HyperClientProc::::create( + 1, + String::from(CLIENT_PROCS[TEST_TYPE]), + bus.clone(), + settings.client, + ); + Proc::::run(http_client_proc).expect("Hyper client processor should run"); + + time::sleep(OBSERVE).await; + let accepted = ACCEPTED.load(Ordering::SeqCst); + + bus.stop("ProSA HTTP client backoff unit test end".into()) + .await + .expect("ProSA should stop"); + + assert!( + accepted > 0, + "The client socket should have tried to connect" + ); + assert!( + accepted <= MAX_ATTEMPTS, + "The client socket reconnected {accepted} times in {OBSERVE:?}, it is not backing off" + ); + + // Wait on main task to end + let _ = main_handle.join(); + } + + /// A socket task that panics must not take the other sockets down with it. + /// + /// The adaptor runs inside the socket task, so any panic in user code ends that task. The + /// processor used to propagate the join error, which dropped the whole task set and aborted + /// every other socket with the requests its queue was holding + #[tokio::test] + async fn client_socket_panic_is_contained() { + let mut settings = HttpTestSettings::new( + Url::parse("http://localhost:0").expect("Panic test URL should be valid"), + None, + None, + ); + let server_url = settings.server.listener.url.clone(); + + let (bus, main) = MainProc::::create(&settings, Some(4)); + let main_handle = std::thread::Builder::new() + .name("main".to_string()) + .spawn(move || { + runtime::Builder::new_multi_thread() + .worker_threads(1) + .enable_all() + .thread_name("main") + .build() + .expect("Runtime should be valid") + .block_on(async { + main.run().await; + }) + }) + .expect("Main thread should be spawned"); + + let http_server_stub = StubProc::::create( + 1, + String::from("HTTP_SERVER_STUB"), + bus.clone(), + settings.stub, + ); + Proc::::run(http_server_stub).expect("Stub processor should run"); + + let http_server_proc = HyperServerProc::::create( + 2, + String::from(SERVER_PROCS[PANIC_TEST_TYPE]), + bus.clone(), + settings.server, + ); + Proc::::run(http_server_proc).expect("Hyper server processor should run"); + + let server_url = bound_url(SERVER_PROCS[PANIC_TEST_TYPE], &server_url).await; + for backend in &mut settings.client.backends { + backend.url = server_url.clone(); + } + + let http_client_proc = HyperClientProc::::create( + 3, + String::from(CLIENT_PROCS[PANIC_TEST_TYPE]), + bus.clone(), + settings.client, + ); + Proc::::run(http_client_proc).expect("Hyper client processor should run"); + + // The request the socket was holding dies with it, and the injector has no timeout of its + // own, so with a single transaction in flight it would wait on that one forever and the + // test would measure the injector rather than the pool + settings.inj.max_concurrents_send = 4; + let http_inj_proc = InjProc::::create( + 4, + String::from("HTTP_INJ_PROC"), + bus.clone(), + settings.inj, + ); + Proc::::run(http_inj_proc).expect("Inj processor should run"); + + // Let the loop settle before breaking it, so the panic lands on a socket that was serving + assert!( + wait_for(TEST_TIMEOUT, async || COUNTER[PANIC_TEST_TYPE] + .load(Ordering::SeqCst) + > 0) + .await, + "The panic test client should have completed a round trip" + ); + + let before_panic = COUNTER[PANIC_TEST_TYPE].load(Ordering::SeqCst); + PANIC_ARMED.store(true, Ordering::SeqCst); + + // The socket that panicked is gone, and with it the request it was holding. What must come + // back is the pool: the processor reopens the slot and the next requests are served again + assert!( + wait_for(TEST_TIMEOUT, async || COUNTER[PANIC_TEST_TYPE] + .load(Ordering::SeqCst) + > before_panic + 2) + .await, + "The client should keep serving after one of its socket tasks panicked" + ); + + // A restarted processor would serve again too, so this is what tells the two apart. The + // adaptor is built once per run of the main loop, and a restart is what abandons the queues + // of every other socket + assert_eq!( + 1, + PANIC_CLIENT_ADAPTORS.load(Ordering::SeqCst), + "A panicking socket task should not restart the Hyper client processor" + ); + + bus.stop("ProSA HTTP client socket panic unit test end".into()) + .await + .expect("ProSA should stop"); + + let _ = main_handle.join(); + } }