diff --git a/crates/loopd/src/broker_port.rs b/crates/loopd/src/broker_port.rs index c16d616..1f68f2b 100644 --- a/crates/loopd/src/broker_port.rs +++ b/crates/loopd/src/broker_port.rs @@ -19,6 +19,9 @@ pub const UNAVAILABLE: &str = "the tool broker is unavailable"; pub const NOT_CONFIGURED: &str = "no tool broker is configured"; /// The failure a request too big for one frame reaches. pub const TOO_LARGE: &str = "the request is too large for the tool broker"; +/// The longest `BrokerPort` waits for an answer after a pending frame, whatever its `expires` +/// says, so a turn cannot be parked for ever by a far expiry. +pub const MAX_PENDING_WAIT: Duration = Duration::from_secs(24 * 60 * 60); /// The runbook entry every outage line ends with. pub const POINTER: &str = "see docs/runbook.md#broker-unavailable"; @@ -55,6 +58,7 @@ impl std::io::Read for Deadline<'_> { pub struct BrokerPort { socket: PathBuf, timeout: Duration, + pending_cap: Duration, log: Box, } @@ -63,10 +67,19 @@ impl BrokerPort { BrokerPort { socket, timeout, + pending_cap: MAX_PENDING_WAIT, log: Box::new(|l| eprintln!("{l}")), } } + /// The same port with another cap on the wait after a pending frame (`MAX_PENDING_WAIT`). + pub fn with_pending_cap(self, pending_cap: Duration) -> BrokerPort { + BrokerPort { + pending_cap, + ..self + } + } + pub fn with_log( socket: PathBuf, timeout: Duration, @@ -75,6 +88,7 @@ impl BrokerPort { BrokerPort { socket, timeout, + pending_cap: MAX_PENDING_WAIT, log, } } @@ -172,7 +186,8 @@ impl ToolPort for BrokerPort { let wait = expires .unix_millis() .saturating_sub(Timestamp::now().unix_millis()); - until = match Instant::now().checked_add(Duration::from_millis(wait)) { + let wait = Duration::from_millis(wait).min(self.pending_cap); + until = match Instant::now().checked_add(wait) { Some(until) => until, None => return self.unavailable("an expiry too far away"), }; diff --git a/crates/loopd/tests/broker_port_cap.rs b/crates/loopd/tests/broker_port_cap.rs new file mode 100644 index 0000000..9e96751 --- /dev/null +++ b/crates/loopd/tests/broker_port_cap.rs @@ -0,0 +1,60 @@ +//! After a pending frame, `BrokerPort` waits for `expires` plus its timeout, but never longer than +//! its cap, whatever `expires` says (M3a review finding 8). `brokerd` falls back to the largest +//! timestamp when `now + ttl_ms` does not fit, so a far `expires` is reachable by configuration. + +#[path = "support/broker.rs"] +mod fake; + +use std::sync::{Arc, Mutex}; +use std::thread; +use std::time::{Duration, Instant}; + +use fake::{broker, request, send}; +use loopd::broker_port::{BrokerPort, MAX_PENDING_WAIT, UNAVAILABLE}; +use loopd::tools::ToolPort; +use proto::{Timestamp, ToolResponse}; + +#[test] +fn the_default_cap_is_a_day() { + assert_eq!(MAX_PENDING_WAIT, Duration::from_secs(24 * 60 * 60)); +} + +#[test] +fn a_pending_frame_with_the_largest_expiry_still_ends_at_the_cap() { + let (socket, broker) = broker(|stream, request| { + let pending = ToolResponse::PendingApproval { + approval: 7, + expires: Timestamp::MAX, + }; + send(stream, request.id, false, pending); + thread::sleep(Duration::from_millis(1_500)); + }); + let lines = Arc::new(Mutex::new(Vec::new())); + let sink = lines.clone(); + let port = BrokerPort::with_log( + socket, + Duration::from_millis(100), + Box::new(move |line| sink.lock().unwrap().push(line.to_string())), + ) + .with_pending_cap(Duration::from_millis(200)); + + let started = Instant::now(); + let answer = port.call(&request(), &mut |_| {}); + let took = started.elapsed(); + assert_eq!( + answer, + ToolResponse::Failed { + message: UNAVAILABLE.to_string() + } + ); + assert!( + took >= Duration::from_millis(250), + "cap plus timeout: {took:?}" + ); + assert!( + took < Duration::from_millis(1_200), + "gave up late: {took:?}" + ); + assert_eq!(lines.lock().unwrap().len(), 1); + broker.join().unwrap(); +}