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 <noreply@anthropic.com>
This commit is contained in:
Meta Vens
parent
4f5d0b1a7e
commit
272dc09800
6 changed files with 605 additions and 7 deletions
|
|
@ -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 `<state_root>/minds/<instance>/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.
|
||||
|
|
|
|||
9
Cargo.lock
generated
9
Cargo.lock
generated
|
|
@ -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"
|
||||
|
|
|
|||
11
README.md
11
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 `<state_root>/minds/<instance>/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.
|
||||
|
||||
|
|
|
|||
|
|
@ -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"
|
||||
|
|
|
|||
|
|
@ -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<HuginnMindResponse, ClientError> {
|
||||
self.exchange(&request)
|
||||
.map_err(|error| ClientError::Unavailable { endpoint: self.endpoint, detail: format!("{error:#}") })
|
||||
}
|
||||
|
||||
fn exchange(&self, request: &HuginnMindRequest) -> Result<HuginnMindResponse> {
|
||||
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::<HuginnMindResponse>(&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<CultNetRudpSocketTransportConnection> {
|
||||
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<CultNetMessage> {
|
||||
session.send_schema_message(message)?;
|
||||
let deadline = Instant::now() + self.timeout;
|
||||
loop {
|
||||
session.poll_resends()?;
|
||||
if let Some(reply) = session.receive_schema_message_once()? {
|
||||
return Ok(reply);
|
||||
}
|
||||
if Instant::now() >= deadline {
|
||||
bail!("no answer within {:?}", self.timeout);
|
||||
}
|
||||
std::thread::sleep(POLL);
|
||||
}
|
||||
}
|
||||
}
|
||||
439
crates/eureka-state/tests/client.rs
Normal file
439
crates/eureka-state/tests/client.rs
Normal file
|
|
@ -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<PipelineDocument>) -> 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::<Vec<_>>();
|
||||
let shorts = |count: usize| (0..count).map(short).collect::<Vec<_>>();
|
||||
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<S: MindStore> IndexSink<S> for Inert {
|
||||
fn committed(&mut self, _mind: &Mind<S>, _writes: &[PipelineRef]) -> anyhow::Result<()> {
|
||||
Ok(())
|
||||
}
|
||||
fn status(&self) -> IndexStatus {
|
||||
IndexStatus::Current
|
||||
}
|
||||
fn search(&mut self, _text: &str, _top_k: u32) -> Result<SearchTicket, String> {
|
||||
Err("this index cannot search".into())
|
||||
}
|
||||
fn search_deadline(&self) -> Duration {
|
||||
Duration::ZERO
|
||||
}
|
||||
fn searched(&mut self) -> Vec<(SearchTicket, Result<Hits, String>)> {
|
||||
Vec::new()
|
||||
}
|
||||
fn abandon(&mut self, _ticket: SearchTicket) {}
|
||||
}
|
||||
|
||||
type TestDaemon = Daemon<OwnedRedbMessagePackBackingStore, Inert>;
|
||||
|
||||
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<Vec<PipelineDocument>>) -> (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<AtomicBool>,
|
||||
thread: Option<JoinHandle<()>>,
|
||||
_root: Option<TempDir>,
|
||||
}
|
||||
|
||||
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<u8>) -> 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}");
|
||||
}
|
||||
Loading…
Reference in a new issue