diff --git a/crates/eureka-state/src/lib.rs b/crates/eureka-state/src/lib.rs index 15b943d..94dbfec 100644 --- a/crates/eureka-state/src/lib.rs +++ b/crates/eureka-state/src/lib.rs @@ -13,7 +13,8 @@ //! in the shape the protocol promises. //! //! Every call is its own session: it connects, asks, fetches what was -//! deferred, and drops the session. Nothing is kept between calls. +//! deferred, and disconnects, whether the call succeeded or failed. Nothing is +//! kept between calls. use std::fmt; use std::net::SocketAddr; @@ -24,7 +25,7 @@ use cultnet_rs::{ CULTNET_OPERATION_CONNECTION_ID, CultMesh, CultMeshRudpSocketOptions, CultNetMessage, CultNetRudpSocketTransportConnection, fetch_content, }; -use huginn_mind::envelope::{decode_response, encode_request}; +use huginn_mind::envelope::{OperationFailure, decode_response, encode_request}; use huginn_mind::epiphany_pipeline::Slug; use huginn_mind::{HuginnMindRequest, HuginnMindResponse, MAX_DEFERRED_BODY_BYTES}; @@ -32,25 +33,47 @@ use huginn_mind::{HuginnMindRequest, HuginnMindResponse, MAX_DEFERRED_BODY_BYTES const MESSAGE_ID: &str = "eureka-state-call"; const POLL: Duration = Duration::from_millis(2); -/// The daemon could not be reached, went quiet, or did not answer as the -/// protocol says it must. `detail` says which. +/// Why a call produced no answer. `Unavailable` is the transport's failure and +/// the only one worth retrying: the daemon could not be reached, went quiet +/// past the call's deadline, or answered out of protocol. `Rejected` is the +/// daemon refusing the envelope itself, which a retry of the same call cannot +/// change. #[derive(Debug)] pub enum ClientError { Unavailable { endpoint: SocketAddr, detail: String }, + Rejected { endpoint: SocketAddr, code: String, detail: String }, } impl fmt::Display for ClientError { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - let Self::Unavailable { endpoint, detail } = self; - write!(f, "the Huginn daemon at rudp://{endpoint} is unavailable: {detail}") + match self { + Self::Unavailable { endpoint, detail } => { + write!(f, "the Huginn daemon at rudp://{endpoint} is unavailable: {detail}") + } + Self::Rejected { endpoint, code, detail } => { + write!(f, "the Huginn daemon at rudp://{endpoint} rejected the request: {code}: {detail}") + } + } } } impl std::error::Error for ClientError {} -/// A daemon's address, the instance the caller acts for, and how long any one -/// wait may last: the connection, the answer, and each chunk of a deferred -/// body each get `timeout`. +/// What `exchange` can fail with, before the endpoint is attached. +enum Failure { + Transport(anyhow::Error), + Rejected(OperationFailure), +} + +impl From for Failure { + fn from(error: anyhow::Error) -> Self { + Self::Transport(error) + } +} + +/// A daemon's address, the instance the caller acts for, and the time one whole +/// call may take: the connection, the answer, and every chunk of a deferred +/// body together must finish within `timeout` of the call starting. #[derive(Clone, Debug)] pub struct HuginnClient { endpoint: SocketAddr, @@ -70,21 +93,47 @@ impl HuginnClient { } pub fn call(&self, request: HuginnMindRequest) -> Result { - self.exchange(&request) - .map_err(|error| ClientError::Unavailable { endpoint: self.endpoint, detail: format!("{error:#}") }) + let deadline = Instant::now() + self.timeout; + let unavailable = + |error: anyhow::Error| ClientError::Unavailable { endpoint: self.endpoint, detail: format!("{error:#}") }; + let mut session = self.connect(deadline).map_err(unavailable)?; + let outcome = self.exchange(&mut session, &request, deadline); + // Best effort: the daemon drops a session it never hears from, only later. + let _ = session.disconnect(Vec::new()); + outcome.map_err(|failure| match failure { + Failure::Transport(error) => unavailable(error), + Failure::Rejected(rejection) => ClientError::Rejected { + endpoint: self.endpoint, + code: rejection.code, + detail: rejection.message, + }, + }) } - fn exchange(&self, request: &HuginnMindRequest) -> Result { - let mut session = self.connect()?; - let reply = self.ask(&mut session, &encode_request(MESSAGE_ID, request, None)?)?; - let (_, answer) = decode_response(&reply)?; - match answer.map_err(|failure| anyhow!("the daemon rejected the envelope: {}: {}", failure.code, failure.message))? { + fn exchange( + &self, + session: &mut CultNetRudpSocketTransportConnection, + request: &HuginnMindRequest, + deadline: Instant, + ) -> Result { + let reply = self.ask(session, &encode_request(MESSAGE_ID, request, None)?, deadline)?; + let CultNetMessage::OperationResponse { operation, .. } = &reply else { + return Err(anyhow!("the daemon answered with something other than an operation response").into()); + }; + if operation != request.operation() { + return Err(anyhow!("the daemon answered {operation}, not the {} that was asked", request.operation()).into()); + } + let (message_id, answer) = decode_response(&reply)?; + if message_id != MESSAGE_ID { + return Err(anyhow!("the daemon answered request {message_id}, not {MESSAGE_ID}").into()); + } + match answer.map_err(Failure::Rejected)? { HuginnMindResponse::Deferred(deferred) => { let body = fetch_content(&deferred.manifest, MAX_DEFERRED_BODY_BYTES, |chunk| { - self.ask(&mut session, &chunk) + self.ask(session, &chunk, deadline) })?; match rmp_serde::from_slice::(&body).context("the deferred body is not a response")? { - HuginnMindResponse::Deferred(_) => bail!("a deferred body is itself deferred"), + HuginnMindResponse::Deferred(_) => Err(anyhow!("a deferred body is itself deferred").into()), answered => Ok(answered), } } @@ -92,7 +141,7 @@ impl HuginnClient { } } - fn connect(&self) -> Result { + fn connect(&self, deadline: Instant) -> Result { let mut session = CultMesh::create_rudp_client_for_endpoint( format!("eureka-state-{}", self.instance.0), CULTNET_OPERATION_CONNECTION_ID, @@ -100,7 +149,6 @@ impl HuginnClient { CultMeshRudpSocketOptions::default(), )?; session.connect(Vec::new())?; - let deadline = Instant::now() + self.timeout; loop { session.poll_resends()?; let _ = session.receive_once()?; @@ -108,23 +156,28 @@ impl HuginnClient { return Ok(session); } if Instant::now() >= deadline { - bail!("no accept within {:?}", self.timeout); + bail!("no accept before the call's {:?} deadline", self.timeout); } std::thread::sleep(POLL); } } - /// One message on the session and the next message back, within `timeout`. - fn ask(&self, session: &mut CultNetRudpSocketTransportConnection, message: &CultNetMessage) -> Result { + /// One message on the session and the next message back, before the call's + /// `deadline`. + fn ask( + &self, + session: &mut CultNetRudpSocketTransportConnection, + message: &CultNetMessage, + deadline: Instant, + ) -> Result { session.send_schema_message(message)?; - let deadline = Instant::now() + self.timeout; loop { session.poll_resends()?; if let Some(reply) = session.receive_schema_message_once()? { return Ok(reply); } if Instant::now() >= deadline { - bail!("no answer within {:?}", self.timeout); + bail!("no answer before the call's {:?} deadline", self.timeout); } std::thread::sleep(POLL); } diff --git a/crates/eureka-state/tests/client.rs b/crates/eureka-state/tests/client.rs index 78519bb..07f458a 100644 --- a/crates/eureka-state/tests/client.rs +++ b/crates/eureka-state/tests/client.rs @@ -4,7 +4,7 @@ use std::net::{SocketAddr, UdpSocket}; use std::sync::Arc; -use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; use std::thread::JoinHandle; use std::time::{Duration, Instant}; @@ -199,6 +199,9 @@ struct Server { addr: SocketAddr, stopping: Arc, thread: Option>, + /// The sessions a scripted server has seen end: what the client's + /// disconnect looks like from the daemon's side. + ended: Arc, _root: Option, } @@ -220,7 +223,7 @@ fn serve(root: TempDir, mut daemon: TestDaemon) -> Server { let thread = std::thread::spawn(move || { run(&mut daemon, &mut hub, ®istry, &loop_stopping, &ServeOptions::default()).unwrap(); }); - Server { addr, stopping, thread: Some(thread), _root: Some(root) } + Server { addr, stopping, thread: Some(thread), ended: Arc::default(), _root: Some(root) } } /// A server that answers each schema message with whatever `script` says. @@ -229,10 +232,15 @@ fn scripted(mut script: impl FnMut(&CultNetMessage) -> CultNetMessage + Send + ' let addr = hub.local_addr().unwrap(); let stopping = Arc::new(AtomicBool::new(false)); let loop_stopping = Arc::clone(&stopping); + let ended = Arc::new(AtomicUsize::new(0)); + let loop_ended = Arc::clone(&ended); let thread = std::thread::spawn(move || { while !loop_stopping.load(Ordering::Relaxed) { hub.poll_resends().unwrap(); while let Some(event) = hub.receive_event_once().unwrap() { + if matches!(event, CultNetRudpServerEvent::Disconnected { .. }) { + loop_ended.fetch_add(1, Ordering::Relaxed); + } let CultNetRudpServerEvent::Frame { session, frame } = event else { continue }; let message = decode_cultnet_message_from_slice(&frame.payload, CultNetWireContract::CultNetSchemaV0).unwrap(); @@ -241,13 +249,19 @@ fn scripted(mut script: impl FnMut(&CultNetMessage) -> CultNetMessage + Send + ' std::thread::sleep(Duration::from_millis(2)); } }); - Server { addr, stopping, thread: Some(thread), _root: None } + Server { addr, stopping, thread: Some(thread), ended, _root: None } } /// A scripted server that answers any operation with a deferral of `body` and /// serves its chunks faithfully. fn deferring(body: Vec) -> Server { - let (manifest, chunks) = pack_content("art", "package", "", "", "", &body, 256 * 1024).expect("a body packs"); + deferring_in_chunks(body, 256 * 1024, Duration::ZERO) +} + +/// `deferring`, with `chunk_bytes` per chunk and each chunk answered only after +/// `drip`. +fn deferring_in_chunks(body: Vec, chunk_bytes: usize, drip: Duration) -> Server { + let (manifest, chunks) = pack_content("art", "package", "", "", "", &body, chunk_bytes).expect("a body packs"); let deferred = encode_response( "eureka-state-call", "whoami", @@ -257,9 +271,12 @@ fn deferring(body: Vec) -> Server { .unwrap(); scripted(move |message| match message { CultNetMessage::OperationRequest { .. } => deferred.clone(), - chunk => answer_content_chunk_request(chunk, |hash| { - chunks.iter().find(|c| c.chunk_hash == hash).map(|c| c.payload.as_slice()) - }), + chunk => { + std::thread::sleep(drip); + answer_content_chunk_request(chunk, |hash| { + chunks.iter().find(|c| c.chunk_hash == hash).map(|c| c.payload.as_slice()) + }) + } }) } @@ -268,8 +285,38 @@ fn client(addr: SocketAddr) -> HuginnClient { } fn unavailable(error: ClientError) -> (SocketAddr, String) { - let ClientError::Unavailable { endpoint, detail } = error; - (endpoint, detail) + match error { + ClientError::Unavailable { endpoint, detail } => (endpoint, detail), + rejected @ ClientError::Rejected { .. } => panic!("expected Unavailable, got {rejected}"), + } +} + +fn rejected(error: ClientError) -> (SocketAddr, String, String) { + match error { + ClientError::Rejected { endpoint, code, detail } => (endpoint, code, detail), + unavailable @ ClientError::Unavailable { .. } => panic!("expected Rejected, got {unavailable}"), + } +} + +/// A scripted server whose every answer is `reply`. +fn answering(reply: CultNetMessage) -> Server { + scripted(move |_| reply.clone()) +} + +fn nothing_found(message_id: &str, operation: &str) -> CultNetMessage { + encode_response(message_id, operation, &HuginnMindResponse::View(None), "scripted").unwrap() +} + +/// Wait for a scripted server to have seen `count` sessions end. +fn ended_within(server: &Server, count: usize, wait: Duration) -> bool { + let until = Instant::now() + wait; + while server.ended.load(Ordering::Relaxed) < count { + if Instant::now() >= until { + return false; + } + std::thread::sleep(Duration::from_millis(10)); + } + true } fn document_query(instance: &str) -> HuginnMindRequest { @@ -373,7 +420,7 @@ fn a_closed_port_is_unavailable_naming_the_endpoint_within_the_timeout() { assert!(error.to_string().contains(&closed.to_string()), "{error}"); let (endpoint, detail) = unavailable(error); assert_eq!(endpoint, closed); - assert!(detail.contains("no accept"), "{detail}"); + assert!(detail.contains("no accept") && detail.contains("deadline"), "{detail}"); waited.push(elapsed); } assert!(waited[1] > waited[0] + Duration::from_millis(500), "{waited:?}"); @@ -398,20 +445,79 @@ fn a_daemon_that_never_answers_is_unavailable_within_the_timeout() { assert!(detail.contains("no answer"), "{detail}"); } -/// An envelope failure means the request never reached a mind; it is not an -/// answer, so the caller sees the code. +/// An envelope failure means the request never reached a mind. It is the +/// daemon refusing, not a transport failure, so a caller that retries on +/// `Unavailable` does not loop on it; the code and the message arrive intact. #[test] -fn an_envelope_failure_is_unavailable_and_carries_its_code() { - let server = scripted(|_| { - encode_failure( - "eureka-state-call", - "whoami", - &OperationFailure { code: "wrong-service".into(), message: "not huginn.mind".into() }, - "scripted", - ) +fn an_envelope_failure_is_rejected_and_carries_its_code() { + let server = answering(encode_failure( + "eureka-state-call", + "whoami", + &OperationFailure { code: "wrong-service".into(), message: "not huginn.mind".into() }, + "scripted", + )); + let error = client(server.addr).call(HuginnMindRequest::Whoami).unwrap_err(); + assert!(error.to_string().contains("wrong-service"), "{error}"); + let (endpoint, code, detail) = rejected(error); + assert_eq!((endpoint, code.as_str(), detail.as_str()), (server.addr, "wrong-service", "not huginn.mind")); +} + +/// An answer to another request, or to another operation, is not this call's +/// answer, whatever it says. +#[test] +fn an_answer_that_is_not_to_this_request_is_refused() { + let foreign = answering(nothing_found("someone-elses-call", "whoami")); + let (_, detail) = unavailable(client(foreign.addr).call(HuginnMindRequest::Whoami).unwrap_err()); + assert!(detail.contains("someone-elses-call"), "{detail}"); + + let other_operation = answering(nothing_found("eureka-state-call", "query")); + let (_, detail) = unavailable(client(other_operation.addr).call(HuginnMindRequest::Whoami).unwrap_err()); + assert!(detail.contains("query") && detail.contains("whoami"), "{detail}"); + + let matching = answering(nothing_found("eureka-state-call", "whoami")); + assert_eq!(client(matching.addr).call(HuginnMindRequest::Whoami).unwrap(), HuginnMindResponse::View(None)); +} + +/// The timeout bounds the whole call. Seven chunks each answered after 300 ms +/// take two seconds; no single wait exceeds the 500 ms timeout, and the call +/// still gives up at it. +#[test] +fn the_timeout_bounds_the_whole_call_not_each_wait() { + let server = deferring_in_chunks(vec![7_u8; 7 * 1024], 1024, Duration::from_millis(300)); + let timeout = Duration::from_millis(500); + let started = Instant::now(); + let error = HuginnClient::new(server.addr, slug(INSTANCE), timeout).call(HuginnMindRequest::Whoami).unwrap_err(); + let elapsed = started.elapsed(); + assert!(elapsed >= timeout && elapsed < Duration::from_millis(1100), "{elapsed:?}"); + let (_, detail) = unavailable(error); + assert!(detail.contains("deadline"), "{detail}"); +} + +/// Every call ends its session, whether it answered, was rejected, or gave up. +#[test] +fn every_call_disconnects_its_session() { + let answered = answering(nothing_found("eureka-state-call", "whoami")); + client(answered.addr).call(HuginnMindRequest::Whoami).unwrap(); + assert!(ended_within(&answered, 1, Duration::from_secs(5)), "an answered call left its session open"); + + let refused = answering(encode_failure( + "eureka-state-call", + "whoami", + &OperationFailure { code: "wrong-service".into(), message: "no".into() }, + "scripted", + )); + rejected(client(refused.addr).call(HuginnMindRequest::Whoami).unwrap_err()); + assert!(ended_within(&refused, 1, Duration::from_secs(5)), "a rejected call left its session open"); + + let silent = scripted(|_| { + std::thread::sleep(Duration::from_millis(800)); + CultNetMessage::Error { error: "late".into(), code: None, details: None } }); - let (_, detail) = unavailable(client(server.addr).call(HuginnMindRequest::Whoami).unwrap_err()); - assert!(detail.contains("wrong-service"), "{detail}"); + let error = HuginnClient::new(silent.addr, slug(INSTANCE), Duration::from_millis(300)) + .call(HuginnMindRequest::Whoami) + .unwrap_err(); + unavailable(error); + assert!(ended_within(&silent, 1, Duration::from_secs(5)), "a timed-out call left its session open"); } /// A deferred body that is not a response, and one that is another deferral,