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}"); +}