eureka-state: whole-call deadline, correlated answers, Rejected, disconnect

The timeout now bounds the whole call: one deadline at call start, every wait
gets what remains, so a chunk-by-chunk deferred body cannot outlive it. The
response's message id and operation are checked against the request. An
envelope-level OperationFailure is ClientError::Rejected, distinct from the
transport's Unavailable, so a retry on Unavailable cannot loop on a permanent
protocol mismatch. Every call disconnects its session, success or failure.

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>

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
+/// Why a call produced no answer. `Unavailable` is the transport's failure and
-/// protocol says it must. `detail` says which.
+/// 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;
+        match self {
-        write!(f, "the Huginn daemon at rudp://{endpoint} is unavailable: {detail}")
+            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
+/// What `exchange` can fail with, before the endpoint is attached.
-/// wait may last: the connection, the answer, and each chunk of a deferred
+enum Failure {
-/// body each get `timeout`.
+    Transport(anyhow::Error),
+    Rejected(OperationFailure),
+}
+
+impl From<anyhow::Error> 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<HuginnMindResponse, ClientError> {
-        self.exchange(&request)
+        let deadline = Instant::now() + self.timeout;
-            .map_err(|error| ClientError::Unavailable { endpoint: self.endpoint, detail: format!("{error:#}") })
+        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<HuginnMindResponse> {
+    fn exchange(
-        let mut session = self.connect()?;
+        &self,
-        let reply = self.ask(&mut session, &encode_request(MESSAGE_ID, request, None)?)?;
+        session: &mut CultNetRudpSocketTransportConnection,
-        let (_, answer) = decode_response(&reply)?;
+        request: &HuginnMindRequest,
-        match answer.map_err(|failure| anyhow!("the daemon rejected the envelope: {}: {}", failure.code, failure.message))? {
+        deadline: Instant,
+    ) -> Result<HuginnMindResponse, Failure> {
+        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::<HuginnMindResponse>(&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<CultNetRudpSocketTransportConnection> {
+    fn connect(&self, deadline: Instant) -> Result<CultNetRudpSocketTransportConnection> {
         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`.
+    /// One message on the session and the next message back, before the call's
-    fn ask(&self, session: &mut CultNetRudpSocketTransportConnection, message: &CultNetMessage) -> Result<CultNetMessage> {
+    /// `deadline`.
+    fn ask(
+        &self,
+        session: &mut CultNetRudpSocketTransportConnection,
+        message: &CultNetMessage,
+        deadline: Instant,
+    ) -> Result<CultNetMessage> {
         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);
         }

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<AtomicBool>,
     thread: Option<JoinHandle<()>>,
+    /// The sessions a scripted server has seen end: what the client's
+    /// disconnect looks like from the daemon's side.
+    ended: Arc<AtomicUsize>,
     _root: Option<TempDir>,
 }
 
@@ -220,7 +223,7 @@ fn serve(root: TempDir, mut daemon: TestDaemon) -> Server {
     let thread = std::thread::spawn(move || {
         run(&mut daemon, &mut hub, &registry, &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<u8>) -> 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<u8>, 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<u8>) -> Server {
     .unwrap();
     scripted(move |message| match message {
         CultNetMessage::OperationRequest { .. } => deferred.clone(),
-        chunk => answer_content_chunk_request(chunk, |hash| {
+        chunk => {
-            chunks.iter().find(|c| c.chunk_hash == hash).map(|c| c.payload.as_slice())
+            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;
+    match error {
-    (endpoint, detail)
+        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
+/// An envelope failure means the request never reached a mind. It is the
-/// answer, so the caller sees the code.
+/// 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() {
+fn an_envelope_failure_is_rejected_and_carries_its_code() {
-    let server = scripted(|_| {
+    let server = answering(encode_failure(
-        encode_failure(
+        "eureka-state-call",
-            "eureka-state-call",
+        "whoami",
-            "whoami",
+        &OperationFailure { code: "wrong-service".into(), message: "not huginn.mind".into() },
-            &OperationFailure { code: "wrong-service".into(), message: "not huginn.mind".into() },
+        "scripted",
-            "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());
+    let error = HuginnClient::new(silent.addr, slug(INSTANCE), Duration::from_millis(300))
-    assert!(detail.contains("wrong-service"), "{detail}");
+        .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,