From debe7ce9e7160c5c63832f88d67f9add6f97b1aa Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Tue, 29 Sep 2026 23:30:21 +0200 Subject: [PATCH 1/5] Move the operation envelope to huginn-mind: one spelling for client and daemon envelope.rs (encode/decode request and response, OperationFailure, FAILURE_SCHEMA) leaves huginn-daemon for huginn-mind, next to the wire types it wraps. The daemon imports it and no longer re-exports it. base64 moves with it; the daemon keeps base64 because serve.rs measures and decodes payloads. schemars leaves the daemon, whose only user was the envelope's OperationFailure derive. The two envelope tests run unchanged from their new home on huginn-mind fixtures. Co-Authored-By: Claude Opus 5.5 --- Cargo.lock | 2 +- crates/huginn-daemon/Cargo.toml | 1 - crates/huginn-daemon/src/lib.rs | 10 +++------- crates/huginn-daemon/src/semantic_tests.rs | 2 +- crates/huginn-daemon/src/serve.rs | 8 ++++---- crates/huginn-mind/Cargo.toml | 1 + .../src/envelope.rs | 16 ++++++++++------ crates/huginn-mind/src/lib.rs | 4 +++- 8 files changed, 23 insertions(+), 21 deletions(-) rename crates/{huginn-daemon => huginn-mind}/src/envelope.rs (93%) diff --git a/Cargo.lock b/Cargo.lock index 3ecdc74..1e30f52 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -477,7 +477,6 @@ dependencies = [ "cultnet-rs", "huginn-mind", "rmp-serde", - "schemars", "serde", "serde_json", "sha2", @@ -491,6 +490,7 @@ name = "huginn-mind" version = "0.1.0" dependencies = [ "anyhow", + "base64", "chrono", "cultcache-rs", "cultnet-rs", diff --git a/crates/huginn-daemon/Cargo.toml b/crates/huginn-daemon/Cargo.toml index 428588e..5890c67 100644 --- a/crates/huginn-daemon/Cargo.toml +++ b/crates/huginn-daemon/Cargo.toml @@ -24,7 +24,6 @@ chrono = "0.4.44" cultnet-rs = { git = "https://github.com/GameCult/CultLib.git", rev = "6251cbcd37c12ef3f0c1cc50063a03369225b7ac", package = "cultnet-rs" } huginn-mind = { path = "../huginn-mind" } rmp-serde = "1" -schemars = "1" serde = { version = "1", features = ["derive"] } # The vector store's and the embedder's HTTP APIs are the xenos boundary, and # JSON exists in `index/ollama.rs` and `index/qdrant.rs` only. `ureq` 3 with diff --git a/crates/huginn-daemon/src/lib.rs b/crates/huginn-daemon/src/lib.rs index a9fbb9f..90ae5e8 100644 --- a/crates/huginn-daemon/src/lib.rs +++ b/crates/huginn-daemon/src/lib.rs @@ -1,14 +1,12 @@ //! `huginn-daemon`: one process that serves one mind over CultNet. //! -//! The crate owns five things and no rule. `startup` owns "may this process +//! The crate owns four things and no rule. `startup` owns "may this process //! serve this mind", by delegating to `Mind::open` and refusing before //! anything listens. `index` owns when the mind's documents are embedded and //! written to the vector store, off the serving path. `Daemon::handle` owns "which `Mind` method answers this //! request": it dispatches, it asks `Mind::require_instance` for every read -//! that names an instance, and it compares nothing itself. `envelope` owns "is -//! this a request at all": a message it cannot decode is answered with a typed -//! failure and touches no mind. `serve` owns "which session gets which reply" -//! and whether an answer is delivered whole, deferred or refused; `bodies` +//! that names an instance, and it compares nothing itself. `serve` owns "which +//! session gets which reply" and whether an answer is delivered whole, deferred or refused; `bodies` //! owns how long and how much of a deferred answer is held. //! //! Every admission rule, the instance check, the receipt, the derived status @@ -17,7 +15,6 @@ pub mod bodies; pub mod daemon; -pub mod envelope; pub mod index; pub mod serve; @@ -25,5 +22,4 @@ pub mod serve; mod semantic_tests; pub use daemon::{Daemon, Handled, Hits, IndexSink, Search, SearchTicket, runtime_id}; -pub use envelope::{FAILURE_SCHEMA, OperationFailure}; pub use serve::{Options, ServeOptions, parse_options, run, startup}; diff --git a/crates/huginn-daemon/src/semantic_tests.rs b/crates/huginn-daemon/src/semantic_tests.rs index 513f0e9..f72c5b4 100644 --- a/crates/huginn-daemon/src/semantic_tests.rs +++ b/crates/huginn-daemon/src/semantic_tests.rs @@ -20,7 +20,7 @@ use tempfile::TempDir; use crate::daemon::Daemon; use crate::daemon::tests::{CAMPAIGN, INSTANCE, batch, campaign_seed, now, slug}; -use crate::envelope::{decode_response, encode_request}; +use huginn_mind::envelope::{decode_response, encode_request}; use crate::index::fakes::{FakeEmbedder, FakeIndex, Gate}; use crate::index::{Backoff, Described, Hit, WorkerSink, collection_name}; use crate::serve::{ServeOptions, bind, run, schema_registry}; diff --git a/crates/huginn-daemon/src/serve.rs b/crates/huginn-daemon/src/serve.rs index a5efc02..bfc54b8 100644 --- a/crates/huginn-daemon/src/serve.rs +++ b/crates/huginn-daemon/src/serve.rs @@ -79,7 +79,7 @@ use crate::bodies::{ DEFERRED_BUDGET_BYTES, DEFERRED_CHUNK_BYTES, DeferredBodies, MAX_DEFERRED_BODY_BYTES, }; use crate::daemon::{Daemon, Handled, IndexSink, Search, SearchTicket, runtime_id}; -use crate::envelope::{OperationFailure, decode_request, encode_failure, encode_response}; +use huginn_mind::envelope::{OperationFailure, decode_request, encode_failure, encode_response}; use crate::index::{Backoff, Embedder, VectorIndex, WorkerSink}; /// What the loop does when nothing is waiting. A deferred body untouched for @@ -345,7 +345,7 @@ fn encode_or_fail( Err(error) => encode_failure( message_id, operation, - &crate::envelope::OperationFailure { + &OperationFailure { code: "response-not-encodable".into(), message: format!("{error:#}"), }, @@ -548,7 +548,7 @@ mod tests { open_unindexed, seeded_wide, slug, }; use huginn_mind::epiphany_pipeline::{PipelineKind, PipelineRef}; - use crate::envelope::{FAILURE_SCHEMA, decode_response, encode_request}; + use huginn_mind::envelope::{FAILURE_SCHEMA, decode_response, encode_request}; use crate::index::Backoff; use base64::Engine; use base64::engine::general_purpose::STANDARD; @@ -632,7 +632,7 @@ mod tests { assert_eq!(rejected(&reply).1, code); } // The sixth code, `not-an-operation-request`, is the codec's and is - // pinned in `envelope::tests`: `answer` never produces it, because a + // pinned in `huginn_mind::envelope::tests`: `answer` never produces it, because a // message of another family is answered with `Error` instead. let reply = answer(&mut daemon, ®istry, &mut fresh_bodies(), CultNetMessage::Error { error: "hello".into(), code: None, details: None }, now()).reply(); let CultNetMessage::Error { error, .. } = &reply else { panic!("expected an error, got {reply:?}") }; diff --git a/crates/huginn-mind/Cargo.toml b/crates/huginn-mind/Cargo.toml index a2e60d4..ccf4e94 100644 --- a/crates/huginn-mind/Cargo.toml +++ b/crates/huginn-mind/Cargo.toml @@ -15,6 +15,7 @@ publish.workspace = true # daemon's (Cut 10, Cut 11). [dependencies] anyhow = "1" +base64 = "0.22" chrono = "0.4.44" cultcache-rs = { git = "https://github.com/GameCult/CultLib.git", rev = "6251cbcd37c12ef3f0c1cc50063a03369225b7ac", package = "cultcache-rs" } cultnet-rs = { git = "https://github.com/GameCult/CultLib.git", rev = "6251cbcd37c12ef3f0c1cc50063a03369225b7ac", package = "cultnet-rs" } diff --git a/crates/huginn-daemon/src/envelope.rs b/crates/huginn-mind/src/envelope.rs similarity index 93% rename from crates/huginn-daemon/src/envelope.rs rename to crates/huginn-mind/src/envelope.rs index fecbde2..8152a54 100644 --- a/crates/huginn-daemon/src/envelope.rs +++ b/crates/huginn-mind/src/envelope.rs @@ -14,7 +14,7 @@ use anyhow::Result; use base64::Engine; use base64::engine::general_purpose::STANDARD; use cultnet_rs::CultNetMessage; -use huginn_mind::wire::{HuginnMindRequest, HuginnMindResponse, MIND_REQUEST_SCHEMA, MIND_RESPONSE_SCHEMA, MIND_SERVICE_ID}; +use crate::wire::{HuginnMindRequest, HuginnMindResponse, MIND_REQUEST_SCHEMA, MIND_RESPONSE_SCHEMA, MIND_SERVICE_ID}; use schemars::JsonSchema; use serde::{Deserialize, Serialize}; @@ -151,9 +151,9 @@ pub fn decode_response(message: &CultNetMessage) -> Result<(String, Result MindRefusal { MindRefusal::ForeignInstance { declared: OTHER.into(), mind: INSTANCE.into() } @@ -183,7 +183,7 @@ mod tests { schema_epoch: "epiphany.pipeline.epoch.v2".into(), documents: 0, receipts: 0, - index: huginn_mind::wire::IndexStatus::Current, + index: crate::wire::IndexStatus::Current, }), "accepted", ), @@ -214,7 +214,11 @@ mod tests { /// `Error` instead. #[test] fn a_request_round_trips_and_a_foreign_message_is_not_a_request() { - let request = HuginnMindRequest::Admit(batch(INSTANCE, vec![identity(INSTANCE)])); + let request = HuginnMindRequest::Admit(PipelineAdmissionBatch { + instance: slug(INSTANCE), + provenance: provenance(Faculty::Hands), + documents: vec![instance(INSTANCE)], + }); let message = encode_request("m-3", &request, Some("eureka-state".into())).unwrap(); assert_eq!(decode_request(&message).unwrap(), ("m-3".into(), request)); diff --git a/crates/huginn-mind/src/lib.rs b/crates/huginn-mind/src/lib.rs index 0fd85e7..b791aac 100644 --- a/crates/huginn-mind/src/lib.rs +++ b/crates/huginn-mind/src/lib.rs @@ -14,7 +14,8 @@ //! selection (CultNet's, over rows this crate supplies) through the same //! `docs` the rules use; nothing is stored for it. //! -//! The wire vocabulary is here too, so the daemon and the client share one set +//! The wire vocabulary and its operation envelope (`envelope`) are here too, +//! so the daemon and the client share one set //! of types. //! //! Document shape, bounds, formats and keys are `epiphany-pipeline`'s; the @@ -26,6 +27,7 @@ pub mod admission; mod docs; +pub mod envelope; pub mod index; pub mod mind; mod rows; From 4f5d0b1a7e622e6d4cb47d6bf080a7c5d162a00b Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Tue, 29 Sep 2026 23:32:54 +0200 Subject: [PATCH 2/5] Move the deferred-body bound to the wire vocabulary both sides read MAX_DEFERRED_BODY_BYTES was the daemon's constant, but a client caps its content fetch at the same number, and the client crate cannot depend on the daemon. It moves to huginn-mind::wire next to the response schema it bounds; the daemon imports it, the pin test stays in bodies.rs. Co-Authored-By: Claude Opus 5.5 --- crates/huginn-daemon/src/bodies.rs | 4 +--- crates/huginn-daemon/src/serve.rs | 3 ++- crates/huginn-mind/src/lib.rs | 2 +- crates/huginn-mind/src/wire.rs | 5 +++++ 4 files changed, 9 insertions(+), 5 deletions(-) diff --git a/crates/huginn-daemon/src/bodies.rs b/crates/huginn-daemon/src/bodies.rs index be6de34..2f21fa2 100644 --- a/crates/huginn-daemon/src/bodies.rs +++ b/crates/huginn-daemon/src/bodies.rs @@ -31,9 +31,6 @@ use cultnet_rs::{CultMeshCdnArtifactManifest, CultMeshCdnChunk}; /// The size of one chunk of a deferred body. Four fit one send with room for a /// control-plane answer beside them on the same session. pub const DEFERRED_CHUNK_BYTES: usize = 256 * 1024; -/// The largest answer that is deferred rather than refused, measured on the -/// encoded payload. -pub const MAX_DEFERRED_BODY_BYTES: u64 = 64 * 1024 * 1024; /// What the daemon keeps across all deferred bodies. pub const DEFERRED_BUDGET_BYTES: u64 = 256 * 1024 * 1024; @@ -163,6 +160,7 @@ impl DeferredBodies { mod tests { use super::*; use cultnet_rs::pack_content; + use huginn_mind::MAX_DEFERRED_BODY_BYTES; /// The three numbers a client author is told, spelled out: a chunk of /// 256 KiB, a largest deferred body of 64 MiB, a budget of 256 MiB that diff --git a/crates/huginn-daemon/src/serve.rs b/crates/huginn-daemon/src/serve.rs index bfc54b8..bb9fb8e 100644 --- a/crates/huginn-daemon/src/serve.rs +++ b/crates/huginn-daemon/src/serve.rs @@ -76,10 +76,11 @@ use huginn_mind::wire::{ use huginn_mind::{Mind, MindRefusal, MindStore, OwnedRedbMessagePackBackingStore}; use crate::bodies::{ - DEFERRED_BUDGET_BYTES, DEFERRED_CHUNK_BYTES, DeferredBodies, MAX_DEFERRED_BODY_BYTES, + DEFERRED_BUDGET_BYTES, DEFERRED_CHUNK_BYTES, DeferredBodies, }; use crate::daemon::{Daemon, Handled, IndexSink, Search, SearchTicket, runtime_id}; use huginn_mind::envelope::{OperationFailure, decode_request, encode_failure, encode_response}; +use huginn_mind::MAX_DEFERRED_BODY_BYTES; use crate::index::{Backoff, Embedder, VectorIndex, WorkerSink}; /// What the loop does when nothing is waiting. A deferred body untouched for diff --git a/crates/huginn-mind/src/lib.rs b/crates/huginn-mind/src/lib.rs index b791aac..29da386 100644 --- a/crates/huginn-mind/src/lib.rs +++ b/crates/huginn-mind/src/lib.rs @@ -52,7 +52,7 @@ pub use receipt::{DocumentVersion, Faculty, HuginnCommitReceipt, PipelineProvena pub use refusal::MindRefusal; pub use store::{MindStore, OwnedRedbMessagePackBackingStore}; pub use wire::{ - DeferredAnswer, HuginnMindRequest, HuginnMindResponse, IndexStatus, MIND_REQUEST_SCHEMA, MIND_REQUEST_SCHEMA_JSON, + DeferredAnswer, HuginnMindRequest, HuginnMindResponse, IndexStatus, MAX_DEFERRED_BODY_BYTES, MIND_REQUEST_SCHEMA, MIND_REQUEST_SCHEMA_JSON, MIND_RESPONSE_SCHEMA, MIND_RESPONSE_SCHEMA_JSON, MIND_SERVICE_ID, MindStatus, }; diff --git a/crates/huginn-mind/src/wire.rs b/crates/huginn-mind/src/wire.rs index dcbd491..54bce78 100644 --- a/crates/huginn-mind/src/wire.rs +++ b/crates/huginn-mind/src/wire.rs @@ -23,6 +23,11 @@ pub const MIND_SERVICE_ID: &str = "huginn.mind"; pub const MIND_REQUEST_SCHEMA: &str = "huginn.mind_request.v1"; pub const MIND_RESPONSE_SCHEMA: &str = "huginn.mind_response.v1"; +/// The largest answer that is deferred rather than refused, measured on the +/// encoded payload. The daemon refuses past it and a client caps its fetch at +/// it, so both sides read one number. +pub const MAX_DEFERRED_BODY_BYTES: u64 = 64 * 1024 * 1024; + /// The published JSON schema of each wire type, as the catalog advertises it. /// The files are derived from the types below and pinned byte for byte by /// `published_wire_schemas_match_derivation`, so the catalog cannot advertise From 272dc098008d378a6c2cf71a1d1714aed97ec9e1 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Tue, 29 Sep 2026 23:41:25 +0200 Subject: [PATCH 3/5] Add the eureka-state client core: HuginnClient::call One call carries one HuginnMindRequest to a daemon over CultNet RUDP on the operation connection and returns the mind's own HuginnMindResponse. Each call is its own session and nothing survives it. A Deferred answer is fetched on that session through cultnet-rs's fetch_content, capped at huginn-mind's MAX_DEFERRED_BODY_BYTES, and decoded, so a caller never sees the deferral. Refusals are answers. ClientError::Unavailable is the only error and names the endpoint. The client opens no store and admits nothing. Tests run a real daemon in-process on a loopback port over a temporary mind, plus scripted servers for the failures a daemon cannot be made to produce. Co-Authored-By: Claude Opus 5.5 --- AGENTS.md | 6 +- Cargo.lock | 9 + README.md | 11 +- crates/eureka-state/Cargo.toml | 15 + crates/eureka-state/src/lib.rs | 132 +++++++++ crates/eureka-state/tests/client.rs | 439 ++++++++++++++++++++++++++++ 6 files changed, 605 insertions(+), 7 deletions(-) create mode 100644 crates/eureka-state/tests/client.rs diff --git a/AGENTS.md b/AGENTS.md index e5b0656..bfa7ee0 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -18,12 +18,14 @@ Studio's job, not Huginn's. status included, over `cultcache-rs` and `epiphany-pipeline` pinned by git rev. `huginn-daemon` serves one mind over CultNet RUDP through `cultnet-rs` at the same rev, and reaches the leaf and the store type through - `huginn-mind` so one crate pins each revision. `eureka-state` is a stub. + `huginn-mind` so one crate pins each revision. `eureka-state` is the client + core: one call to a daemon over CultNet, holding no state. - Owned: an instance's memory documents, their admission and the read side that derives their status, in `huginn-mind`, over a redb CultCache store at `/minds//mind.redb`; and their CultNet surface in `huginn-daemon`, which owns the socket, the - sessions, the process and the operation envelope, and no rule. The two wire + sessions and the process, and no rule; the operation envelope is + `huginn-mind`'s. The two wire schemas are published from `schemas/cultnet/`. There is no index yet. - To depend on: Qdrant, directly, once retrieval exists. Unreachable Qdrant is to be a loud refusal, never a fallback store. No crate connects to it yet. diff --git a/Cargo.lock b/Cargo.lock index 1e30f52..f278655 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -335,6 +335,15 @@ dependencies = [ [[package]] name = "eureka-state" version = "0.1.0" +dependencies = [ + "anyhow", + "chrono", + "cultnet-rs", + "huginn-daemon", + "huginn-mind", + "rmp-serde", + "tempfile", +] [[package]] name = "fastrand" diff --git a/README.md b/README.md index 0b7a07f..985f9d3 100644 --- a/README.md +++ b/README.md @@ -42,10 +42,11 @@ A Cargo workspace of three crates: - `crates/huginn-mind`: storage, identity, admission, and queries and derived status over memory documents. -- `crates/huginn-daemon`: the CultNet surface. The socket, the sessions, the - process and the operation envelope, and no rule. -- `crates/eureka-state`: a stub. It will carry typed state for the Eureka - pipeline; today it holds none. +- `crates/huginn-daemon`: the CultNet surface. The socket, the sessions and + the process, and no rule. The operation envelope both sides speak is + `huginn-mind`'s. +- `crates/eureka-state`: the client core, `HuginnClient::call`: one request to a + daemon over CultNet RUDP, a deferred answer resolved, and no state. `huginn-mind` is live: it opens one instance's store (an owned redb CultCache at `/minds//mind.redb`, locked for the mind's lifetime), @@ -71,7 +72,7 @@ failure and reaches no mind. The two schemas it publishes live in `schemas/cultnet/` and are pinned to their derivation by a test. There is no index yet. -`eureka-state` is a stub; the campaign's cut map in +`eureka-state` has no MCP surface yet; the campaign's cut map in `Epiphany/notes/eureka-pipeline-state-cut.md` owns what each crate must do next. diff --git a/crates/eureka-state/Cargo.toml b/crates/eureka-state/Cargo.toml index 5b49ff0..0de0909 100644 --- a/crates/eureka-state/Cargo.toml +++ b/crates/eureka-state/Cargo.toml @@ -5,4 +5,19 @@ edition.workspace = true license.workspace = true publish.workspace = true +# The client core: one call over CultNet to a Huginn daemon. It owns the +# transport and nothing of the mind. The operation envelope, the wire +# vocabulary and the deferred-body bound are `huginn-mind`'s, so the client and +# the daemon cannot spell them differently; the content fetch and its +# verification are `cultnet-rs`'s. [dependencies] +anyhow = "1" +cultnet-rs = { git = "https://github.com/GameCult/CultLib.git", rev = "6251cbcd37c12ef3f0c1cc50063a03369225b7ac", package = "cultnet-rs" } +huginn-mind = { path = "../huginn-mind" } +rmp-serde = "1" + +# Tests run a real daemon in-process against a temporary mind. +[dev-dependencies] +chrono = "0.4.44" +huginn-daemon = { path = "../huginn-daemon" } +tempfile = "3" diff --git a/crates/eureka-state/src/lib.rs b/crates/eureka-state/src/lib.rs index e69de29..15b943d 100644 --- a/crates/eureka-state/src/lib.rs +++ b/crates/eureka-state/src/lib.rs @@ -0,0 +1,132 @@ +//! `eureka-state`, the client core: one call to a Huginn daemon. +//! +//! `HuginnClient::call` sends one `HuginnMindRequest` in CultNet's operation +//! envelope over RUDP, waits for the answer, and returns it as the mind's own +//! `HuginnMindResponse`. An answer too large for one send arrives as a +//! `Deferred` manifest; the client fetches the body on the same session, +//! verifies it, and decodes it, so the caller sees the answer the mind gave +//! and never the deferral. +//! +//! The client owns the transport and no state. A refusal is an answer, so it +//! comes back as `Ok(HuginnMindResponse::Refused(..))`. `ClientError` is the +//! only error, and it means the daemon could not be reached or did not answer +//! 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. + +use std::fmt; +use std::net::SocketAddr; +use std::time::{Duration, Instant}; + +use anyhow::{Context, Result, anyhow, bail}; +use cultnet_rs::{ + CULTNET_OPERATION_CONNECTION_ID, CultMesh, CultMeshRudpSocketOptions, CultNetMessage, + CultNetRudpSocketTransportConnection, fetch_content, +}; +use huginn_mind::envelope::{decode_response, encode_request}; +use huginn_mind::epiphany_pipeline::Slug; +use huginn_mind::{HuginnMindRequest, HuginnMindResponse, MAX_DEFERRED_BODY_BYTES}; + +/// The correlation key of the one request a session carries. +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. +#[derive(Debug)] +pub enum ClientError { + Unavailable { endpoint: SocketAddr, 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}") + } +} + +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`. +#[derive(Clone, Debug)] +pub struct HuginnClient { + endpoint: SocketAddr, + instance: Slug, + timeout: Duration, +} + +impl HuginnClient { + pub fn new(endpoint: SocketAddr, instance: Slug, timeout: Duration) -> Self { + Self { endpoint, instance, timeout } + } + + /// The instance this client acts for. Requests name their own instance; + /// this is the one the caller's tools fill in. + pub fn instance(&self) -> &Slug { + &self.instance + } + + pub fn call(&self, request: HuginnMindRequest) -> Result { + self.exchange(&request) + .map_err(|error| ClientError::Unavailable { endpoint: self.endpoint, detail: format!("{error:#}") }) + } + + 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))? { + HuginnMindResponse::Deferred(deferred) => { + let body = fetch_content(&deferred.manifest, MAX_DEFERRED_BODY_BYTES, |chunk| { + self.ask(&mut session, &chunk) + })?; + match rmp_serde::from_slice::(&body).context("the deferred body is not a response")? { + HuginnMindResponse::Deferred(_) => bail!("a deferred body is itself deferred"), + answered => Ok(answered), + } + } + answered => Ok(answered), + } + } + + fn connect(&self) -> Result { + let mut session = CultMesh::create_rudp_client_for_endpoint( + format!("eureka-state-{}", self.instance.0), + CULTNET_OPERATION_CONNECTION_ID, + &format!("rudp://{}", self.endpoint), + CultMeshRudpSocketOptions::default(), + )?; + session.connect(Vec::new())?; + let deadline = Instant::now() + self.timeout; + loop { + session.poll_resends()?; + let _ = session.receive_once()?; + if session.connected() { + return Ok(session); + } + if Instant::now() >= deadline { + bail!("no accept within {:?}", 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 { + 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); + } + std::thread::sleep(POLL); + } + } +} diff --git a/crates/eureka-state/tests/client.rs b/crates/eureka-state/tests/client.rs new file mode 100644 index 0000000..cac6418 --- /dev/null +++ b/crates/eureka-state/tests/client.rs @@ -0,0 +1,439 @@ +//! The client against a real daemon, in-process on a loopback port over a +//! temporary mind, and against scripted servers where the daemon cannot be +//! made to misbehave. + +use std::net::{SocketAddr, UdpSocket}; +use std::sync::Arc; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::thread::JoinHandle; +use std::time::{Duration, Instant}; + +use chrono::Utc; +use cultnet_rs::{ + CultNetMessage, CultNetRudpServerEvent, CultNetWireContract, Selection, answer_content_chunk_request, + decode_cultnet_message_from_slice, encode_cultnet_message_to_vec, pack_content, +}; +use eureka_state::{ClientError, HuginnClient}; +use huginn_daemon::serve::{MAX_RESPONSE_BYTES, bind, run, schema_registry}; +use huginn_daemon::{Daemon, Handled, Hits, IndexSink, SearchTicket, ServeOptions}; +use huginn_mind::envelope::{OperationFailure, encode_failure, encode_response}; +use huginn_mind::epiphany_pipeline::{ + AuthorityMap, CodeLocation, CutDelete, CutVerification, Date, DocRef, FileChange, Line, NegativeCheck, OrgRepo, + PipelineCampaign, PipelineCutSpec, PipelineDocument, PipelineInstance, PipelineKind, PipelineRef, + PipelineStewardship, Sha, Short, Slug, StructuralDelta, Title, VerificationTest, +}; +use huginn_mind::{ + DeferredAnswer, Faculty, HuginnMindRequest, HuginnMindResponse, IndexStatus, Mind, MindRefusal, MindStore, + OwnedRedbMessagePackBackingStore, PipelineAdmissionBatch, PipelineAdmissionOutcome, PipelineProvenance, +}; +use tempfile::TempDir; + +const INSTANCE: &str = "eureka"; +const OTHER: &str = "thought-cage"; +const CAMPAIGN: &str = "eureka-state"; +const REPO: &str = "GameCult/Epiphany"; +const TIMEOUT: Duration = Duration::from_secs(20); + +fn slug(value: &str) -> Slug { + Slug(value.into()) +} + +fn sha() -> Sha { + Sha("5f98228d".into()) +} + +fn identity(name: &str) -> PipelineDocument { + PipelineDocument::Instance(PipelineInstance { + instance: slug(name), + display_name: Short(format!("{name} mind")), + created_at: Date("2026-09-29".into()), + host: Short(name.into()), + }) +} + +fn stewardship() -> PipelineDocument { + PipelineDocument::Stewardship(PipelineStewardship { + instance: slug(INSTANCE), + repo: OrgRepo(REPO.into()), + sequence: 1, + assigned_on: Date("2026-09-29".into()), + note: "assigned".into(), + }) +} + +fn campaign() -> PipelineDocument { + PipelineDocument::Campaign(PipelineCampaign { + slug: slug(CAMPAIGN), + title: Title("Eureka pipeline state".into()), + repos: vec![OrgRepo(REPO.into())], + working_branch: Short("hands/cut13a".into()), + target_doc: DocRef { path: Short("notes/target.md".into()), start_line: 1, end_line: 9, commit: sha() }, + }) +} + +fn batch(instance: &str, documents: Vec) -> PipelineAdmissionBatch { + PipelineAdmissionBatch { + instance: slug(instance), + provenance: PipelineProvenance { + faculty: Faculty::Hands, + agent: Short("claude".into()), + session: Short("session-1".into()), + tool: Short("admit".into()), + }, + documents, + } +} + +/// A `Line` and a `Short` at the leaf's own bound. +fn line(n: usize) -> Line { + Line(format!("{n:0>1000}")) +} + +fn short(n: usize) -> Short { + Short(format!("{n:0>200}")) +} + +/// A cut spec with every list the leaf bounds filled: at 256 file changes a +/// megabyte of field content, more than one send carries. +fn cut_spec(cut: &str, file_changes: usize) -> PipelineDocument { + let lines = |count: usize| (0..count).map(line).collect::>(); + let shorts = |count: usize| (0..count).map(short).collect::>(); + PipelineDocument::CutSpec(PipelineCutSpec { + campaign: slug(CAMPAIGN), + cut: cut.into(), + revision: 1, + title: Title(short(0).0), + repo: OrgRepo(REPO.into()), + branch: short(1), + base: sha(), + depends_on: vec![], + first: lines(16), + deletes: (0..64).map(|n| CutDelete { path: short(n), lines: 9, note: line(n) }).collect(), + keeps_moves: lines(64), + adds: lines(64), + file_changes: (0..file_changes) + .map(|n| FileChange { + location: CodeLocation { path: short(n), line: 1, end_line: Some(9) }, + change: line(n), + }) + .collect(), + authority_map: Some(AuthorityMap { + owner: line(0), + inputs: lines(16), + outputs: lines(16), + derived_state: lines(16), + forbidden_writers: lines(16), + shared_paths: lines(16), + deletion_line: line(1), + }), + verification: CutVerification { + builds: lines(64), + tests: (0..64).map(|n| VerificationTest { name: short(n), pins: line(n) }).collect(), + negative: (0..64).map(|n| NegativeCheck { pattern: short(n), scope: line(n) }).collect(), + operator: lines(64), + }, + estimate: StructuralDelta { + lines_added: 0, + lines_removed: 0, + dependencies_added: shorts(64), + dependencies_removed: shorts(64), + formats_added: shorts(64), + formats_removed: shorts(64), + targets_added: shorts(64), + targets_removed: shorts(64), + }, + rulings: vec![], + questions: vec![], + }) +} + +fn cut_ref(cut: &str) -> PipelineRef { + PipelineRef { kind: PipelineKind::CutSpec, id: Short(format!("{CAMPAIGN}:cut_spec:cut-{cut}.r1")) } +} + +/// No index: the client's subject is the transport, not the projection. +struct Inert; + +impl IndexSink for Inert { + fn committed(&mut self, _mind: &Mind, _writes: &[PipelineRef]) -> anyhow::Result<()> { + Ok(()) + } + fn status(&self) -> IndexStatus { + IndexStatus::Current + } + fn search(&mut self, _text: &str, _top_k: u32) -> Result { + Err("this index cannot search".into()) + } + fn search_deadline(&self) -> Duration { + Duration::ZERO + } + fn searched(&mut self) -> Vec<(SearchTicket, Result)> { + Vec::new() + } + fn abandon(&mut self, _ticket: SearchTicket) {} +} + +type TestDaemon = Daemon; + +fn answered(handled: Handled) -> HuginnMindResponse { + match handled { + Handled::Answered(response) => response, + Handled::Searching(_) => panic!("the inert index cannot search"), + } +} + +/// A daemon over a temporary mind whose identity is admitted, and whatever +/// else `seed` admits. +fn mind(seed: Vec>) -> (TempDir, TestDaemon) { + let root = tempfile::tempdir().unwrap(); + let mut daemon = Daemon::new(Mind::open(root.path(), &slug(INSTANCE)).unwrap(), Inert); + for documents in std::iter::once(vec![identity(INSTANCE)]).chain(seed) { + let outcome = answered(daemon.handle(HuginnMindRequest::Admit(batch(INSTANCE, documents)), Utc::now())); + assert!(matches!(outcome, HuginnMindResponse::Admit(PipelineAdmissionOutcome::Committed { .. })), "{outcome:?}"); + } + (root, daemon) +} + +/// A server on a loopback port that stops when dropped. +struct Server { + addr: SocketAddr, + stopping: Arc, + thread: Option>, + _root: Option, +} + +impl Drop for Server { + fn drop(&mut self) { + self.stopping.store(true, Ordering::Relaxed); + if let Some(thread) = self.thread.take() { + let _ = thread.join(); + } + } +} + +fn serve(root: TempDir, mut daemon: TestDaemon) -> Server { + let mut hub = bind("127.0.0.1:0".parse().unwrap(), &daemon.runtime_id()).unwrap(); + let addr = hub.local_addr().unwrap(); + let stopping = Arc::new(AtomicBool::new(false)); + let loop_stopping = Arc::clone(&stopping); + let registry = schema_registry().unwrap(); + 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) } +} + +/// A server that answers each schema message with whatever `script` says. +fn scripted(mut script: impl FnMut(&CultNetMessage) -> CultNetMessage + Send + 'static) -> Server { + let mut hub = bind("127.0.0.1:0".parse().unwrap(), "scripted").unwrap(); + let addr = hub.local_addr().unwrap(); + let stopping = Arc::new(AtomicBool::new(false)); + let loop_stopping = Arc::clone(&stopping); + 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() { + let CultNetRudpServerEvent::Frame { session, frame } = event else { continue }; + let message = + decode_cultnet_message_from_slice(&frame.payload, CultNetWireContract::CultNetSchemaV0).unwrap(); + hub.send_schema_message(&session, &script(&message)).unwrap(); + } + std::thread::sleep(Duration::from_millis(2)); + } + }); + Server { addr, stopping, thread: Some(thread), _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"); + let deferred = encode_response( + "eureka-state-call", + "whoami", + &HuginnMindResponse::Deferred(DeferredAnswer { manifest }), + "scripted", + ) + .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()) + }), + }) +} + +fn client(addr: SocketAddr) -> HuginnClient { + HuginnClient::new(addr, slug(INSTANCE), TIMEOUT) +} + +fn unavailable(error: ClientError) -> (SocketAddr, String) { + let ClientError::Unavailable { endpoint, detail } = error; + (endpoint, detail) +} + +fn document_query(instance: &str) -> HuginnMindRequest { + HuginnMindRequest::Query { + instance: slug(instance), + selection: Selection { projection: "document".into(), ..Selection::default() }, + semantic: None, + } +} + +/// Every operation the wire has, asked of a live daemon over the socket, and +/// each answered with the mind's own answer. +#[test] +fn every_operation_round_trips_against_a_live_daemon() { + let (root, daemon) = mind(vec![]); + let server = serve(root, daemon); + let client = client(server.addr); + + let HuginnMindResponse::Whoami(status) = client.call(HuginnMindRequest::Whoami).unwrap() else { + panic!("whoami is answered with a status"); + }; + assert_eq!((status.instance, status.documents), (slug(INSTANCE), 1)); + + let admitted = client.call(HuginnMindRequest::Admit(batch(INSTANCE, vec![stewardship(), campaign()]))).unwrap(); + let HuginnMindResponse::Admit(PipelineAdmissionOutcome::Committed { writes, .. }) = admitted else { + panic!("the batch is committed, got {admitted:?}"); + }; + assert_eq!(writes.len(), 2); + + let HuginnMindResponse::Query(page) = client.call(document_query(INSTANCE)).unwrap() else { + panic!("a query is answered with a page"); + }; + assert_eq!(page.matched, 3, "the identity and the two admitted documents"); + + let campaign_ref = PipelineRef { kind: PipelineKind::Campaign, id: Short(format!("{CAMPAIGN}:campaign:self")) }; + let viewed = client.call(HuginnMindRequest::View { instance: slug(INSTANCE), id: campaign_ref.clone() }).unwrap(); + let HuginnMindResponse::View(Some(view)) = viewed else { panic!("the campaign is viewable, got {viewed:?}") }; + assert_eq!(view.id, campaign_ref); +} + +/// A refusal is data. A read that names another mind, a write that declares +/// another mind, and a document the mind does not hold each come back as `Ok`. +#[test] +fn a_foreign_instance_is_an_answer_not_an_error() { + let (root, daemon) = mind(vec![]); + let server = serve(root, daemon); + let client = client(server.addr); + let foreign = MindRefusal::ForeignInstance { declared: OTHER.into(), mind: INSTANCE.into() }; + + assert_eq!(client.call(document_query(OTHER)).unwrap(), HuginnMindResponse::Refused(foreign.clone())); + assert_eq!( + client.call(HuginnMindRequest::Admit(batch(OTHER, vec![identity(OTHER)]))).unwrap(), + HuginnMindResponse::Admit(PipelineAdmissionOutcome::Refused(foreign)), + ); + let absent = PipelineRef { kind: PipelineKind::CutSpec, id: Short(format!("{CAMPAIGN}:cut_spec:cut-none.r1")) }; + assert_eq!( + client.call(HuginnMindRequest::View { instance: slug(INSTANCE), id: absent }).unwrap(), + HuginnMindResponse::View(None), + ); +} + +/// The wide cut spec's answer is over one send and arrives deferred; the +/// narrow one arrives whole. Both come back equal to what the mind gave. +#[test] +fn a_deferred_answer_resolves_to_the_same_page_as_a_direct_one() { + let (root, mut daemon) = + mind(vec![vec![stewardship(), campaign()], vec![cut_spec("wide", 256), cut_spec("fits", 180)]]); + let measured = |request: &HuginnMindRequest, daemon: &mut TestDaemon| { + let answer = answered(daemon.handle(request.clone(), Utc::now())); + let envelope = encode_response("m", request.operation(), &answer, "huginn-eureka").unwrap(); + let size = encode_cultnet_message_to_vec(&envelope, CultNetWireContract::CultNetSchemaV0).unwrap().len(); + (answer, size as u64) + }; + let wide = HuginnMindRequest::View { instance: slug(INSTANCE), id: cut_ref("wide") }; + let fits = HuginnMindRequest::View { instance: slug(INSTANCE), id: cut_ref("fits") }; + let (wide_answer, wide_size) = measured(&wide, &mut daemon); + let (fits_answer, fits_size) = measured(&fits, &mut daemon); + assert!(matches!(wide_answer, HuginnMindResponse::View(Some(_))), "{wide_answer:?}"); + assert!(wide_size > MAX_RESPONSE_BYTES, "the wide answer must be over one send, or nothing was deferred"); + assert!(fits_size <= MAX_RESPONSE_BYTES, "the narrow answer must fit one send"); + + let server = serve(root, daemon); + let client = client(server.addr); + assert_eq!(client.call(wide).unwrap(), wide_answer); + assert_eq!(client.call(fits).unwrap(), fits_answer); +} + +/// Nothing listens on the port. The call fails naming the endpoint, and it +/// fails when the configured timeout says: two timeouts, two waits. +#[test] +fn a_closed_port_is_unavailable_naming_the_endpoint_within_the_timeout() { + let closed = UdpSocket::bind("127.0.0.1:0").unwrap().local_addr().unwrap(); + let mut waited = Vec::new(); + for timeout in [Duration::from_millis(300), Duration::from_millis(1200)] { + let started = Instant::now(); + let error = HuginnClient::new(closed, slug(INSTANCE), timeout).call(HuginnMindRequest::Whoami).unwrap_err(); + let elapsed = started.elapsed(); + assert!(elapsed >= timeout, "gave up after {elapsed:?}, before its {timeout:?}"); + assert!(elapsed < timeout + Duration::from_millis(800), "gave up after {elapsed:?}, long past its {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}"); + waited.push(elapsed); + } + assert!(waited[1] > waited[0] + Duration::from_millis(500), "{waited:?}"); +} + +/// A daemon that accepts and then says nothing is a timeout on the answer, +/// not on the connection. +#[test] +fn a_daemon_that_never_answers_is_unavailable_within_the_timeout() { + let server = scripted(|_| { + std::thread::sleep(Duration::from_secs(3)); + CultNetMessage::Error { error: "late".into(), code: None, details: None } + }); + let started = Instant::now(); + let error = HuginnClient::new(server.addr, slug(INSTANCE), Duration::from_millis(400)) + .call(HuginnMindRequest::Whoami) + .unwrap_err(); + let elapsed = started.elapsed(); + assert!(elapsed >= Duration::from_millis(400) && elapsed < Duration::from_secs(2), "{elapsed:?}"); + let (endpoint, detail) = unavailable(error); + assert_eq!(endpoint, server.addr); + 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. +#[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", + ) + }); + let (_, detail) = unavailable(client(server.addr).call(HuginnMindRequest::Whoami).unwrap_err()); + assert!(detail.contains("wrong-service"), "{detail}"); +} + +/// A deferred body that is not a response, and one that is another deferral, +/// are protocol violations named as such. +#[test] +fn a_deferred_body_must_be_a_response_and_not_another_deferral() { + let garbage = deferring(vec![0xc1; 40_000]); + let (_, detail) = unavailable(client(garbage.addr).call(HuginnMindRequest::Whoami).unwrap_err()); + assert!(detail.contains("not a response"), "{detail}"); + + let (inner, _) = pack_content("inner", "package", "", "", "", b"x", 16).unwrap(); + let again = rmp_serde::to_vec_named(&HuginnMindResponse::Deferred(DeferredAnswer { manifest: inner })).unwrap(); + let nested = deferring(again); + let (_, detail) = unavailable(client(nested.addr).call(HuginnMindRequest::Whoami).unwrap_err()); + assert!(detail.contains("itself deferred"), "{detail}"); +} + +/// A body over the deferral cap is refused before its first chunk is asked +/// for, by the number the daemon defers under. +#[test] +fn a_body_over_the_deferral_cap_is_refused() { + let over = huginn_mind::MAX_DEFERRED_BODY_BYTES as usize + 1; + let server = deferring(vec![0_u8; over]); + let (_, detail) = unavailable(client(server.addr).call(HuginnMindRequest::Whoami).unwrap_err()); + assert!(detail.contains("exceeds"), "{detail}"); +} From 26f4137d55f86735ae934487740135a64042e79e Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Tue, 29 Sep 2026 23:46:33 +0200 Subject: [PATCH 4/5] Pin the client's configured instance against a mutation that replaces it Co-Authored-By: Claude Opus 5.5 --- crates/eureka-state/tests/client.rs | 1 + 1 file changed, 1 insertion(+) diff --git a/crates/eureka-state/tests/client.rs b/crates/eureka-state/tests/client.rs index cac6418..78519bb 100644 --- a/crates/eureka-state/tests/client.rs +++ b/crates/eureka-state/tests/client.rs @@ -287,6 +287,7 @@ fn every_operation_round_trips_against_a_live_daemon() { let (root, daemon) = mind(vec![]); let server = serve(root, daemon); let client = client(server.addr); + assert_eq!(client.instance(), &slug(INSTANCE), "the client reports the instance it was configured for"); let HuginnMindResponse::Whoami(status) = client.call(HuginnMindRequest::Whoami).unwrap() else { panic!("whoami is answered with a status"); From 3dc50eec8e23018827c31d90dbfd4fd675f9cf65 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Wed, 30 Sep 2026 00:16:33 +0200 Subject: [PATCH 5/5] 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 --- crates/eureka-state/src/lib.rs | 103 ++++++++++++++----- crates/eureka-state/tests/client.rs | 150 ++++++++++++++++++++++++---- 2 files changed, 206 insertions(+), 47 deletions(-) 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,