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 3ecdc74..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" @@ -477,7 +486,6 @@ dependencies = [ "cultnet-rs", "huginn-mind", "rmp-serde", - "schemars", "serde", "serde_json", "sha2", @@ -491,6 +499,7 @@ name = "huginn-mind" version = "0.1.0" dependencies = [ "anyhow", + "base64", "chrono", "cultcache-rs", "cultnet-rs", 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..94dbfec 100644 --- a/crates/eureka-state/src/lib.rs +++ b/crates/eureka-state/src/lib.rs @@ -0,0 +1,185 @@ +//! `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 disconnects, whether the call succeeded or failed. 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::{OperationFailure, 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); + +/// 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 { + 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 {} + +/// 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, + 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 { + 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, + 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(session, &chunk, deadline) + })?; + match rmp_serde::from_slice::(&body).context("the deferred body is not a response")? { + HuginnMindResponse::Deferred(_) => Err(anyhow!("a deferred body is itself deferred").into()), + answered => Ok(answered), + } + } + answered => Ok(answered), + } + } + + 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, + &format!("rudp://{}", self.endpoint), + CultMeshRudpSocketOptions::default(), + )?; + session.connect(Vec::new())?; + loop { + session.poll_resends()?; + let _ = session.receive_once()?; + if session.connected() { + return Ok(session); + } + if Instant::now() >= deadline { + bail!("no accept before the call's {:?} deadline", self.timeout); + } + std::thread::sleep(POLL); + } + } + + /// 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)?; + loop { + session.poll_resends()?; + if let Some(reply) = session.receive_schema_message_once()? { + return Ok(reply); + } + if Instant::now() >= deadline { + 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 new file mode 100644 index 0000000..07f458a --- /dev/null +++ b/crates/eureka-state/tests/client.rs @@ -0,0 +1,546 @@ +//! 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, AtomicUsize, 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>, + /// The sessions a scripted server has seen end: what the client's + /// disconnect looks like from the daemon's side. + ended: Arc, + _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), ended: Arc::default(), _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 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(); + hub.send_schema_message(&session, &script(&message)).unwrap(); + } + std::thread::sleep(Duration::from_millis(2)); + } + }); + 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 { + 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", + &HuginnMindResponse::Deferred(DeferredAnswer { manifest }), + "scripted", + ) + .unwrap(); + scripted(move |message| match message { + CultNetMessage::OperationRequest { .. } => deferred.clone(), + 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()) + }) + } + }) +} + +fn client(addr: SocketAddr) -> HuginnClient { + HuginnClient::new(addr, slug(INSTANCE), TIMEOUT) +} + +fn unavailable(error: ClientError) -> (SocketAddr, String) { + 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 { + 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); + 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"); + }; + 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.contains("deadline"), "{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 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_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 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, +/// 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}"); +} 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/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/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..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 crate::envelope::{OperationFailure, decode_request, encode_failure, encode_response}; +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 @@ -345,7 +346,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 +549,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 +633,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..29da386 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; @@ -50,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