Desk door: the second peer-checked unix-socket door (Answer and catalog; reads interim) #36
11 changed files with 1829 additions and 25 deletions
1
Cargo.lock
generated
1
Cargo.lock
generated
|
|
@ -935,6 +935,7 @@ dependencies = [
|
|||
"cultnet-rs 0.1.0",
|
||||
"huginn-mind",
|
||||
"rmp-serde",
|
||||
"rustix",
|
||||
"serde",
|
||||
"serde_json",
|
||||
"sha2",
|
||||
|
|
|
|||
|
|
@ -164,7 +164,7 @@ pub fn serve<I: IndexSink<OwnedRedbMessagePackBackingStore> + Send + 'static>(
|
|||
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();
|
||||
run(&mut daemon, &mut hub, ®istry, None, &loop_stopping, &ServeOptions::default()).unwrap();
|
||||
});
|
||||
Server { addr, stopping, thread: Some(thread), ended: Arc::default(), _root: Some(root) }
|
||||
}
|
||||
|
|
|
|||
|
|
@ -30,6 +30,8 @@ chrono = "0.4.44"
|
|||
cultnet-rs.workspace = true
|
||||
huginn-mind = { path = "../huginn-mind" }
|
||||
rmp-serde = "1"
|
||||
# The desk door asks the kernel who is on the other end of a unix socket; mind-body's line.
|
||||
rustix = { version = "1", default-features = false, features = ["std", "net"] }
|
||||
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
|
||||
|
|
|
|||
|
|
@ -5,6 +5,7 @@
|
|||
use chrono::{DateTime, Utc};
|
||||
use cultnet_rs::Selection;
|
||||
use crate::sources::forge::HttpForge;
|
||||
use huginn_mind::desk_wire::{DeskRequest, DeskResponse};
|
||||
use huginn_mind::eureka_pipeline::{PipelineRef, Slug};
|
||||
use huginn_mind::wire::{ACTIVITY_REPOS_MAX, HuginnMindRequest, HuginnMindResponse, IndexStatus};
|
||||
use huginn_mind::{Members, Mind, MindRefusal, MindStore, PipelineAdmissionOutcome, SemanticQuery, SourceHits, SourceQuery};
|
||||
|
|
@ -118,6 +119,9 @@ pub fn runtime_id(instance: &Slug) -> String {
|
|||
format!("huginn-{}", instance.0)
|
||||
}
|
||||
|
||||
/// What a desk read is told until follow-up desk-door-reads builds the reads and deletes this.
|
||||
const DESK_READS_UNBUILT: &str = "desk reads not built";
|
||||
|
||||
/// One mind and whatever indexes what lands in it.
|
||||
pub struct Daemon<S: MindStore, I: IndexSink<S>> {
|
||||
mind: Mind<S>,
|
||||
|
|
@ -231,6 +235,22 @@ impl<S: MindStore, I: IndexSink<S>> Daemon<S, I> {
|
|||
Handled::Answered(answer)
|
||||
}
|
||||
|
||||
/// One desk request, answered by the Mind method it names. The Mind is the instance's own
|
||||
/// (`mind` is checked against it); `deciders` is this call's list, handed to the Mind and
|
||||
/// dropped. A desk answer is not indexed (`index_text` gives it no text), so nothing here tells the index.
|
||||
/// The reads wait on follow-up desk-door-reads and are refused until they exist.
|
||||
pub fn handle_desk(&mut self, request: DeskRequest, now: DateTime<Utc>) -> DeskResponse {
|
||||
if let Err(refusal) = self.mind.require_instance(request.mind()) {
|
||||
return DeskResponse::Refused(refusal);
|
||||
}
|
||||
match request {
|
||||
DeskRequest::Answer { subject, authority, deciders, request, .. } => {
|
||||
self.mind.desk_answer(&deciders, subject, authority, request, now).0
|
||||
}
|
||||
DeskRequest::Forks { .. } | DeskRequest::Fork { .. } => DeskResponse::Refused(MindRefusal::Unavailable { detail: DESK_READS_UNBUILT.into() }),
|
||||
}
|
||||
}
|
||||
|
||||
/// A semantic query starts only if the mind accepts it, and the mind
|
||||
/// decides its members before the index is asked anything. A selection
|
||||
/// with no members is answered at once: nothing is embedded and the index
|
||||
|
|
@ -665,11 +685,11 @@ pub(crate) mod tests {
|
|||
Ok(Daemon::new(Mind::open(state_root, instance)?, NoIndex))
|
||||
}
|
||||
|
||||
type Seen = Vec<(Vec<String>, Vec<bool>)>;
|
||||
pub(crate) type Seen = Vec<(Vec<String>, Vec<bool>)>;
|
||||
|
||||
#[derive(Clone, Default)]
|
||||
struct RecordingIndex {
|
||||
seen: Arc<Mutex<Seen>>,
|
||||
pub(crate) struct RecordingIndex {
|
||||
pub(crate) seen: Arc<Mutex<Seen>>,
|
||||
}
|
||||
|
||||
impl<S: MindStore> IndexSink<S> for RecordingIndex {
|
||||
|
|
|
|||
328
crates/huginn-daemon/src/desk_door.rs
Normal file
328
crates/huginn-daemon/src/desk_door.rs
Normal file
|
|
@ -0,0 +1,328 @@
|
|||
//! The desk door: the second door onto the one mind, a unix socket one peer may speak on.
|
||||
//!
|
||||
//! This module owns three things. Who may speak: the kernel's report of the connecting
|
||||
//! process's uid must equal the configured one, and it is asked before a byte is read, so a
|
||||
//! refused peer has nothing read and nothing written. The socket's life: bind, replace a stale
|
||||
//! one, refuse a live one or a file that is not a socket, mode 0660, unlink on drop. And the
|
||||
//! bounds: one request per connection, one reply, a header over `DESK_MAX_REQUEST_BYTES` refused
|
||||
//! before anything is allocated, one `DESK_CONNECTION_DEADLINE` for the whole connection (every
|
||||
//! read and write spends what is left of it, so a peer that trickles bytes is cut off at the
|
||||
//! deadline like one that is silent), a reply past `DESK_MAX_RESPONSE_BYTES` refused by name.
|
||||
//!
|
||||
//! It owns no rule. Which Mind method answers is `Daemon::handle_desk`'s; when the door is
|
||||
//! served is `serve::run`'s, once per pass in the thread that owns the Mind. A peer that is
|
||||
//! slow costs the loop at most `DESK_ACCEPTS_PER_PASS` connections of `DESK_CONNECTION_DEADLINE`
|
||||
//! each, four seconds, and only the configured peer can reach that.
|
||||
//!
|
||||
//! The door keeps no request id: a retry after a lost reply is a new connection with the same
|
||||
//! request, and the stored answer is what replays it, across a restart as well.
|
||||
|
||||
use std::fmt;
|
||||
use std::fs;
|
||||
use std::io::{self, Read, Write};
|
||||
use std::os::fd::AsFd;
|
||||
use std::os::unix::fs::{FileTypeExt, PermissionsExt};
|
||||
use std::os::unix::net::{UnixListener, UnixStream};
|
||||
use std::path::{Path, PathBuf};
|
||||
use std::time::{Duration, Instant};
|
||||
|
||||
use anyhow::{Context, Result, bail, ensure};
|
||||
use chrono::{DateTime, Utc};
|
||||
use cultnet_rs::{
|
||||
CultNetMessage, CultNetSchemaKind, CultNetSchemaRegistration, CultNetSchemaRegistry, CultNetTransportProfile, CultNetWireContract,
|
||||
TcpFramedTransportConnection, TcpFramedTransportProfileOptions, create_tcp_framed_transport_profile, decode_cultnet_message_from_slice,
|
||||
encode_cultnet_message_to_vec,
|
||||
};
|
||||
use huginn_mind::desk_wire::{
|
||||
DESK_MAX_REQUEST_BYTES, DESK_MAX_RESPONSE_BYTES, DESK_REQUEST_SCHEMA, DESK_REQUEST_SCHEMA_JSON, DESK_RESPONSE_SCHEMA,
|
||||
DESK_RESPONSE_SCHEMA_JSON, DESK_SERVICE_ID, DeskResponse,
|
||||
};
|
||||
use huginn_mind::envelope::{OperationFailure, decode_desk_request, encode_desk_failure, encode_desk_response};
|
||||
use huginn_mind::{MindRefusal, MindStore};
|
||||
use rustix::net::sockopt::{Timeout, set_socket_timeout};
|
||||
|
||||
use crate::daemon::{Daemon, IndexSink};
|
||||
use crate::serve::{MAX_ECHOED_FIELD_BYTES, overlong_echo};
|
||||
|
||||
/// How long one connection of the configured peer may hold the loop, reading and writing together.
|
||||
pub const DESK_CONNECTION_DEADLINE: Duration = Duration::from_secs(1);
|
||||
|
||||
/// Connections one loop pass serves, so a peer that connects in a stream cannot hold the RUDP door off.
|
||||
pub const DESK_ACCEPTS_PER_PASS: u32 = 4;
|
||||
|
||||
/// The listening socket, and the one uid that may speak on it.
|
||||
pub struct DeskDoor {
|
||||
listener: UnixListener,
|
||||
path: PathBuf,
|
||||
peer_uid: u32,
|
||||
registry: CultNetSchemaRegistry,
|
||||
}
|
||||
|
||||
/// What became of one connection. `Display` is the log line for the ones that are not answered.
|
||||
#[derive(Debug, PartialEq, Eq)]
|
||||
pub enum Served {
|
||||
Answered,
|
||||
/// The peer's uid is not the configured one: nothing was read or written.
|
||||
Refused { uid: u32, expected: u32 },
|
||||
/// The connection ended without a reply, and why.
|
||||
Dropped(String),
|
||||
}
|
||||
|
||||
impl fmt::Display for Served {
|
||||
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
|
||||
match self {
|
||||
Self::Answered => formatter.write_str("desk door answered a peer"),
|
||||
Self::Refused { uid, expected } => write!(formatter, "desk door refused a peer with uid {uid} (expects {expected})"),
|
||||
Self::Dropped(why) => write!(formatter, "desk door dropped a connection: {why}"),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// The two desk contracts, as the door's catalog advertises them and nothing else.
|
||||
pub fn desk_registry() -> Result<CultNetSchemaRegistry> {
|
||||
let mut registry = CultNetSchemaRegistry::new();
|
||||
for (schema_id, title, schema_json) in [
|
||||
(DESK_REQUEST_SCHEMA, "Huginn desk request", DESK_REQUEST_SCHEMA_JSON),
|
||||
(DESK_RESPONSE_SCHEMA, "Huginn desk response", DESK_RESPONSE_SCHEMA_JSON),
|
||||
] {
|
||||
registry.register(CultNetSchemaRegistration {
|
||||
schema_id: schema_id.to_string(),
|
||||
kind: CultNetSchemaKind::WireMessage,
|
||||
wire_contracts: vec![CultNetWireContract::CultNetSchemaV0],
|
||||
schema_version: Some(schema_id.to_string()),
|
||||
document_type: None,
|
||||
title: Some(title.to_string()),
|
||||
schema_json: Some(schema_json.to_string()),
|
||||
})?;
|
||||
}
|
||||
Ok(registry)
|
||||
}
|
||||
|
||||
/// Binds the desk socket. A path that exists and is not a socket is refused; a socket that
|
||||
/// accepts a connection is a live server and is refused; a socket that refuses the connection is
|
||||
/// stale and is replaced (`Mind::open`'s exclusive lock has already shown no live daemon of this
|
||||
/// instance). The socket is then mode 0660 and non-blocking.
|
||||
pub fn bind_desk(path: &Path, peer_uid: u32) -> Result<DeskDoor> {
|
||||
ensure!(path.is_absolute(), "the desk socket path must be absolute, got {}", path.display());
|
||||
ensure!(peer_uid != 0, "the desk peer uid must not be 0");
|
||||
match fs::symlink_metadata(path) {
|
||||
Ok(found) => {
|
||||
ensure!(found.file_type().is_socket(), "{} exists and is not a socket", path.display());
|
||||
match UnixStream::connect(path) {
|
||||
Ok(_) => bail!("{} is a live socket; another server holds it", path.display()),
|
||||
Err(error) if error.kind() == io::ErrorKind::ConnectionRefused => {
|
||||
fs::remove_file(path).with_context(|| format!("removing the stale socket {}", path.display()))?;
|
||||
}
|
||||
Err(error) => return Err(error).with_context(|| format!("probing the socket {}", path.display())),
|
||||
}
|
||||
}
|
||||
Err(error) if error.kind() == io::ErrorKind::NotFound => {}
|
||||
Err(error) => return Err(error).with_context(|| format!("reading {}", path.display())),
|
||||
}
|
||||
let listener = UnixListener::bind(path).with_context(|| format!("binding {}", path.display()))?;
|
||||
// The door owns the path from here, so a failure below still unlinks it on drop.
|
||||
let door = DeskDoor { listener, path: path.to_path_buf(), peer_uid, registry: desk_registry()? };
|
||||
fs::set_permissions(path, fs::Permissions::from_mode(0o660)).with_context(|| format!("setting the mode of {}", path.display()))?;
|
||||
door.listener.set_nonblocking(true)?;
|
||||
Ok(door)
|
||||
}
|
||||
|
||||
impl DeskDoor {
|
||||
pub fn path(&self) -> &Path {
|
||||
&self.path
|
||||
}
|
||||
|
||||
/// Serves at most `DESK_ACCEPTS_PER_PASS` waiting connections, each to completion, and says
|
||||
/// how many it took. A refusal or a dropped connection is logged and counted: the loop was
|
||||
/// busy with it.
|
||||
pub fn serve_ready<S: MindStore, I: IndexSink<S>>(&mut self, daemon: &mut Daemon<S, I>, now: DateTime<Utc>) -> u32 {
|
||||
let mut taken = 0;
|
||||
while taken < DESK_ACCEPTS_PER_PASS {
|
||||
let stream = match self.listener.accept() {
|
||||
Ok((stream, _)) => stream,
|
||||
Err(error) if error.kind() == io::ErrorKind::WouldBlock => break,
|
||||
Err(error) => {
|
||||
eprintln!("huginn: the desk socket could not accept: {error}");
|
||||
break;
|
||||
}
|
||||
};
|
||||
taken += 1;
|
||||
let served = serve_connection(stream, self.peer_uid, daemon, &self.registry, now, |stream| stream.set_nonblocking(false));
|
||||
if served != Served::Answered {
|
||||
eprintln!("huginn: {served}");
|
||||
}
|
||||
}
|
||||
taken
|
||||
}
|
||||
}
|
||||
|
||||
impl Drop for DeskDoor {
|
||||
fn drop(&mut self) {
|
||||
let _ = fs::remove_file(&self.path);
|
||||
}
|
||||
}
|
||||
|
||||
fn profile() -> CultNetTransportProfile {
|
||||
create_tcp_framed_transport_profile(
|
||||
"huginn-desk",
|
||||
TcpFramedTransportProfileOptions { max_payload_bytes: Some(DESK_MAX_REQUEST_BYTES as u32), ..Default::default() },
|
||||
)
|
||||
}
|
||||
|
||||
/// A stream whose every read and write is bounded by what is left of one deadline: the socket's
|
||||
/// timeout is reset to the remainder before each call, so bytes arriving do not extend it.
|
||||
pub(crate) struct Deadlined<St> {
|
||||
inner: St,
|
||||
deadline: Instant,
|
||||
}
|
||||
|
||||
impl<St: AsFd> Deadlined<St> {
|
||||
pub(crate) fn new(inner: St, within: Duration) -> Self {
|
||||
Self { inner, deadline: Instant::now() + within }
|
||||
}
|
||||
|
||||
fn remaining(&self, which: Timeout) -> io::Result<()> {
|
||||
// Whole milliseconds: rustix rounds a sub-microsecond tail up into an out-of-range tv_usec.
|
||||
let left = Duration::from_millis(self.deadline.saturating_duration_since(Instant::now()).as_millis() as u64);
|
||||
if left.is_zero() {
|
||||
return Err(io::ErrorKind::TimedOut.into());
|
||||
}
|
||||
set_socket_timeout(&self.inner, which, Some(left)).map_err(io::Error::from)
|
||||
}
|
||||
}
|
||||
|
||||
impl<St: Read + AsFd> Read for Deadlined<St> {
|
||||
fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
|
||||
self.remaining(Timeout::Recv)?;
|
||||
self.inner.read(buf)
|
||||
}
|
||||
}
|
||||
|
||||
impl<St: Write + AsFd> Write for Deadlined<St> {
|
||||
fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
|
||||
self.remaining(Timeout::Send)?;
|
||||
self.inner.write(buf)
|
||||
}
|
||||
|
||||
fn flush(&mut self) -> io::Result<()> {
|
||||
self.inner.flush()
|
||||
}
|
||||
}
|
||||
|
||||
/// One connection: the peer first, before anything is read, then one frame in, one frame out,
|
||||
/// all inside `DESK_CONNECTION_DEADLINE`. `ready` readies the stream for blocking I/O and runs
|
||||
/// only after the peer is known to be the configured one.
|
||||
pub fn serve_connection<St, S, I>(
|
||||
stream: St,
|
||||
peer_uid: u32,
|
||||
daemon: &mut Daemon<S, I>,
|
||||
registry: &CultNetSchemaRegistry,
|
||||
now: DateTime<Utc>,
|
||||
ready: impl FnOnce(&St) -> io::Result<()>,
|
||||
) -> Served
|
||||
where
|
||||
St: Read + Write + AsFd,
|
||||
S: MindStore,
|
||||
I: IndexSink<S>,
|
||||
{
|
||||
let uid = match rustix::net::sockopt::socket_peercred(&stream) {
|
||||
Ok(credentials) => credentials.uid.as_raw(),
|
||||
Err(error) => return Served::Dropped(format!("the peer's credentials could not be read: {error}")),
|
||||
};
|
||||
if uid != peer_uid {
|
||||
return Served::Refused { uid, expected: peer_uid };
|
||||
}
|
||||
if let Err(error) = ready(&stream) {
|
||||
return Served::Dropped(format!("the stream could not be readied: {error}"));
|
||||
}
|
||||
let mut transport = TcpFramedTransportConnection::new(Deadlined::new(stream, DESK_CONNECTION_DEADLINE), profile());
|
||||
let frame = match transport.receive() {
|
||||
Ok(frame) => frame,
|
||||
Err(error) => return Served::Dropped(format!("no whole frame arrived: {error:#}")),
|
||||
};
|
||||
let message = match decode_cultnet_message_from_slice(&frame.payload, CultNetWireContract::CultNetSchemaV0) {
|
||||
Ok(message) => message,
|
||||
Err(error) => return Served::Dropped(format!("a frame was not a CultNet message: {error:#}")),
|
||||
};
|
||||
let reply = answer_desk_message(daemon, registry, message, now);
|
||||
let bytes = match encode_cultnet_message_to_vec(&reply, CultNetWireContract::CultNetSchemaV0) {
|
||||
Ok(bytes) => bytes,
|
||||
Err(error) => return Served::Dropped(format!("the reply could not be encoded: {error:#}")),
|
||||
};
|
||||
match transport.send("schema", &bytes) {
|
||||
Ok(()) => Served::Answered,
|
||||
Err(error) => Served::Dropped(format!("the reply was not delivered: {error:#}")),
|
||||
}
|
||||
}
|
||||
|
||||
/// One message in, one message out. Nothing here reaches a mind except through
|
||||
/// `Daemon::handle_desk`, and an envelope that does not decode never gets that far.
|
||||
pub fn answer_desk_message<S: MindStore, I: IndexSink<S>>(
|
||||
daemon: &mut Daemon<S, I>,
|
||||
registry: &CultNetSchemaRegistry,
|
||||
message: CultNetMessage,
|
||||
now: DateTime<Utc>,
|
||||
) -> CultNetMessage {
|
||||
let runtime_id = daemon.runtime_id();
|
||||
if let Some(field) = overlong_echo(&message) {
|
||||
return CultNetMessage::Error {
|
||||
error: format!("{DESK_SERVICE_ID} refuses a {field} over {MAX_ECHOED_FIELD_BYTES} bytes"),
|
||||
code: None,
|
||||
details: None,
|
||||
};
|
||||
}
|
||||
match &message {
|
||||
CultNetMessage::OperationRequest { message_id, operation, .. } => match decode_desk_request(&message) {
|
||||
Ok((message_id, request)) => {
|
||||
let operation = request.operation();
|
||||
let response = daemon.handle_desk(request, now);
|
||||
respond(&response, &message_id, operation, &runtime_id, DESK_MAX_RESPONSE_BYTES)
|
||||
}
|
||||
Err(failure) => encode_desk_failure(message_id, operation, &failure, &runtime_id),
|
||||
},
|
||||
CultNetMessage::SchemaCatalogRequest { .. } => match registry.create_catalog_response(&message) {
|
||||
Ok(reply) => bounded(reply, DESK_MAX_RESPONSE_BYTES, |size| CultNetMessage::Error {
|
||||
error: match size {
|
||||
Ok(bytes) => format!("{DESK_SERVICE_ID} refuses a catalog reply of {bytes} bytes over {DESK_MAX_RESPONSE_BYTES}"),
|
||||
Err(detail) => detail,
|
||||
},
|
||||
code: None,
|
||||
details: None,
|
||||
}),
|
||||
Err(error) => CultNetMessage::Error { error: format!("{error:#}"), code: None, details: None },
|
||||
},
|
||||
_ => CultNetMessage::Error {
|
||||
error: format!("{DESK_SERVICE_ID} answers cultnet.operation_request.v0 and cultnet.schema_catalog_request.v0"),
|
||||
code: None,
|
||||
details: None,
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
/// One answer as its message, or, past `limit` encoded, the refusal that names the size.
|
||||
pub(crate) fn respond(response: &DeskResponse, message_id: &str, operation: &str, runtime_id: &str, limit: usize) -> CultNetMessage {
|
||||
let not_encodable = |detail: String| {
|
||||
encode_desk_failure(message_id, operation, &OperationFailure { code: "response-not-encodable".into(), message: detail }, runtime_id)
|
||||
};
|
||||
let reply = match encode_desk_response(message_id, operation, response, runtime_id) {
|
||||
Ok(reply) => reply,
|
||||
Err(error) => return not_encodable(format!("{error:#}")),
|
||||
};
|
||||
bounded(reply, limit, |size| match size {
|
||||
Ok(bytes) => {
|
||||
let refusal = DeskResponse::Refused(MindRefusal::ResponseTooLarge { bytes: bytes as u64, limit: limit as u64 });
|
||||
encode_desk_response(message_id, operation, &refusal, runtime_id).unwrap_or_else(|error| not_encodable(format!("{error:#}")))
|
||||
}
|
||||
Err(detail) => not_encodable(detail),
|
||||
})
|
||||
}
|
||||
|
||||
/// `reply` itself when it encodes to at most `limit` bytes. Otherwise what `refuse` makes of the
|
||||
/// encoded size, or of the encoding error. Every reply the door sends passes here.
|
||||
fn bounded(reply: CultNetMessage, limit: usize, refuse: impl FnOnce(Result<usize, String>) -> CultNetMessage) -> CultNetMessage {
|
||||
match encode_cultnet_message_to_vec(&reply, CultNetWireContract::CultNetSchemaV0) {
|
||||
Ok(bytes) if bytes.len() <= limit => reply,
|
||||
Ok(bytes) => refuse(Ok(bytes.len())),
|
||||
Err(error) => refuse(Err(format!("{error:#}"))),
|
||||
}
|
||||
}
|
||||
887
crates/huginn-daemon/src/desk_door_tests.rs
Normal file
887
crates/huginn-daemon/src/desk_door_tests.rs
Normal file
|
|
@ -0,0 +1,887 @@
|
|||
//! The desk door's proofs: the peer check and the framing at the unit seam
|
||||
//! (a counting stream), the door and the loop over a real unix socket in a
|
||||
//! tempdir (pipeline smokes), and the startup gates. The Mind's own rules are
|
||||
//! `huginn-mind`'s tests; what is pinned here is who is read, what is bounded
|
||||
//! and which Mind method answers.
|
||||
|
||||
use std::io::{Read, Write};
|
||||
use std::os::fd::{AsFd, BorrowedFd};
|
||||
use std::os::unix::fs::{MetadataExt, PermissionsExt};
|
||||
use std::os::unix::net::{UnixListener, UnixStream};
|
||||
use std::path::{Path, PathBuf};
|
||||
use std::sync::Arc;
|
||||
use std::sync::atomic::{AtomicBool, Ordering};
|
||||
use std::time::{Duration, Instant};
|
||||
|
||||
use base64::Engine;
|
||||
use base64::engine::general_purpose::STANDARD;
|
||||
use cultnet_rs::{
|
||||
CULTNET_OPERATION_CONNECTION_ID, CultMesh, CultMeshRudpSocketOptions, CultNetMessage, CultNetWireContract,
|
||||
decode_cultnet_message_from_slice, encode_cultnet_message_to_vec, encode_frame,
|
||||
};
|
||||
use huginn_mind::desk_wire::{
|
||||
DESK_REQUEST_SCHEMA, DESK_SERVICE_ID, DeskRequest, DeskResponse, ForkAnswer, ForkAnswerCode, ForkAnswerRequest, RequestSchema,
|
||||
};
|
||||
use huginn_mind::envelope::{OperationFailure, decode_desk_response, decode_response, encode_desk_request, encode_request};
|
||||
use huginn_mind::eureka_pipeline::{DeskAuthority, PipelineDocument, Short};
|
||||
use huginn_mind::wire::{HuginnMindRequest, HuginnMindResponse, MIND_REQUEST_SCHEMA, MIND_SERVICE_ID, MindStatus};
|
||||
use huginn_mind::{MindRefusal, OwnedRedbMessagePackBackingStore, options_hash};
|
||||
use tempfile::TempDir;
|
||||
|
||||
use crate::daemon::tests::{INSTANCE, OTHER, RecordingIndex, batch, campaign_seed, now, question_and_ruling, slug};
|
||||
use crate::daemon::Daemon;
|
||||
use crate::desk_door::{DESK_CONNECTION_DEADLINE, DeskDoor, Served, answer_desk_message, bind_desk, desk_registry, serve_connection};
|
||||
use crate::index::Backoff;
|
||||
use crate::serve::{Options, ServeOptions, bind, parse_options, run, schema_registry, startup};
|
||||
|
||||
type Store = OwnedRedbMessagePackBackingStore;
|
||||
const FORK: &str = "eureka-state:question:Q1";
|
||||
|
||||
/// The uid this process runs as: the owner of a file it just made.
|
||||
fn my_uid() -> u32 {
|
||||
let dir = tempfile::tempdir().unwrap();
|
||||
std::fs::metadata(dir.path()).unwrap().uid()
|
||||
}
|
||||
|
||||
fn short(value: &str) -> Short {
|
||||
Short(value.into())
|
||||
}
|
||||
|
||||
/// A daemon over a fresh mind holding the campaign and one answerable question, and that
|
||||
/// question's current `options_hash`. The recording index forgets the seeding.
|
||||
fn seeded(root: &Path) -> (Daemon<Store, RecordingIndex>, String) {
|
||||
let mind = huginn_mind::Mind::open(root, &slug(INSTANCE)).unwrap();
|
||||
let index = RecordingIndex::default();
|
||||
let mut daemon = Daemon::new(mind, index.clone());
|
||||
let (question, _) = question_and_ruling();
|
||||
let PipelineDocument::Question(held) = &question else { unreachable!() };
|
||||
let hash = options_hash(held).0;
|
||||
for documents in [campaign_seed(), vec![question]] {
|
||||
let admitted = daemon.handle(HuginnMindRequest::Admit(batch(INSTANCE, documents)), now()).answered();
|
||||
assert!(matches!(admitted, HuginnMindResponse::Admit(huginn_mind::PipelineAdmissionOutcome::Committed { .. })), "{admitted:?}");
|
||||
}
|
||||
index.seen.lock().unwrap().clear();
|
||||
(daemon, hash)
|
||||
}
|
||||
|
||||
fn state(daemon: &mut Daemon<Store, RecordingIndex>) -> (u32, u32) {
|
||||
match daemon.handle(HuginnMindRequest::Whoami, now()).answered() {
|
||||
HuginnMindResponse::Whoami(MindStatus { documents, receipts, .. }) => (documents, receipts),
|
||||
other => panic!("expected a status, got {other:?}"),
|
||||
}
|
||||
}
|
||||
|
||||
fn wire(hash: &str, request_id: &str) -> ForkAnswerRequest {
|
||||
ForkAnswerRequest {
|
||||
schema_version: RequestSchema::V0,
|
||||
fork_id: FORK.into(),
|
||||
options_hash: hash.into(),
|
||||
answer: Some(ForkAnswer { choice: Some("A".into()), note: None }),
|
||||
retracted: false,
|
||||
retracts: None,
|
||||
answered_at: "2026-09-16T11:00:00Z".into(),
|
||||
request_id: request_id.into(),
|
||||
}
|
||||
}
|
||||
|
||||
const R1: &str = "00000000-0000-4000-8000-000000000001";
|
||||
const R2: &str = "00000000-0000-4000-8000-000000000002";
|
||||
const R3: &str = "00000000-0000-4000-8000-000000000003";
|
||||
|
||||
fn answer(who: &str, authority: DeskAuthority, deciders: &[&str], request: ForkAnswerRequest) -> DeskRequest {
|
||||
DeskRequest::Answer { mind: slug(INSTANCE), subject: short(who), authority, deciders: deciders.iter().map(|d| short(d)).collect(), request }
|
||||
}
|
||||
|
||||
fn message(id: &str, request: &DeskRequest) -> CultNetMessage {
|
||||
encode_desk_request(id, request, None).unwrap()
|
||||
}
|
||||
|
||||
fn framed(message: &CultNetMessage) -> Vec<u8> {
|
||||
encode_frame(&encode_cultnet_message_to_vec(message, CultNetWireContract::CultNetSchemaV0).unwrap()).unwrap()
|
||||
}
|
||||
|
||||
fn read_reply(stream: &mut impl Read) -> Option<CultNetMessage> {
|
||||
let mut header = [0_u8; 4];
|
||||
stream.read_exact(&mut header).ok()?;
|
||||
let mut payload = vec![0_u8; u32::from_be_bytes(header) as usize];
|
||||
stream.read_exact(&mut payload).ok()?;
|
||||
Some(decode_cultnet_message_from_slice(&payload, CultNetWireContract::CultNetSchemaV0).unwrap())
|
||||
}
|
||||
|
||||
fn connect(path: &Path) -> UnixStream {
|
||||
let stream = UnixStream::connect(path).unwrap();
|
||||
stream.set_read_timeout(Some(Duration::from_secs(10))).unwrap();
|
||||
stream.set_write_timeout(Some(Duration::from_secs(10))).unwrap();
|
||||
stream
|
||||
}
|
||||
|
||||
/// One connection: these bytes, then the reply if there is one.
|
||||
fn exchange(path: &Path, bytes: &[u8]) -> Option<CultNetMessage> {
|
||||
let mut stream = connect(path);
|
||||
let _ = stream.write_all(bytes);
|
||||
read_reply(&mut stream)
|
||||
}
|
||||
|
||||
fn reply_of(path: &Path, id: &str, request: &DeskRequest) -> DeskResponse {
|
||||
let reply = exchange(path, &framed(&message(id, request))).expect("the door replied");
|
||||
let (correlation, response) = decode_desk_response(&reply).unwrap();
|
||||
assert_eq!(correlation, id);
|
||||
response.unwrap_or_else(|failure| panic!("{failure:?}"))
|
||||
}
|
||||
|
||||
fn failure_of(path: &Path, bytes: &[u8]) -> OperationFailure {
|
||||
let reply = exchange(path, bytes).expect("the door replied");
|
||||
decode_desk_response(&reply).unwrap().1.unwrap_err()
|
||||
}
|
||||
|
||||
fn counted(response: DeskResponse) -> (String, Option<String>) {
|
||||
match response {
|
||||
DeskResponse::Answered { answer, counted_answer_id, .. } => {
|
||||
assert_eq!(answer.outcome, "Counted", "{answer:?}");
|
||||
(answer.answer_id, counted_answer_id)
|
||||
}
|
||||
other => panic!("expected an answer, got {other:?}"),
|
||||
}
|
||||
}
|
||||
|
||||
/// A door on a real socket in a tempdir, a seeded daemon, and the recording index.
|
||||
struct Rig {
|
||||
_dir: TempDir,
|
||||
_root: TempDir,
|
||||
path: PathBuf,
|
||||
daemon: Daemon<Store, RecordingIndex>,
|
||||
door: DeskDoor,
|
||||
hash: String,
|
||||
index: RecordingIndex,
|
||||
}
|
||||
|
||||
fn rig(peer_uid: u32) -> Rig {
|
||||
let (dir, root) = (tempfile::tempdir().unwrap(), tempfile::tempdir().unwrap());
|
||||
let (daemon, hash) = seeded(root.path());
|
||||
let path = dir.path().join("desk.sock");
|
||||
let door = bind_desk(&path, peer_uid).unwrap();
|
||||
let index = daemon.index().clone();
|
||||
Rig { _dir: dir, _root: root, path, daemon, door, hash, index }
|
||||
}
|
||||
|
||||
impl Rig {
|
||||
/// Runs `client` on its own thread while this one serves the door, then drains what is still waiting.
|
||||
fn serving<T: Send>(&mut self, client: impl FnOnce(&Path) -> T + Send) -> T {
|
||||
let path = self.path.clone();
|
||||
std::thread::scope(|scope| {
|
||||
let handle = scope.spawn(move || client(&path));
|
||||
while !handle.is_finished() {
|
||||
self.door.serve_ready(&mut self.daemon, now());
|
||||
std::thread::sleep(Duration::from_millis(1));
|
||||
}
|
||||
let out = handle.join().unwrap();
|
||||
while self.door.serve_ready(&mut self.daemon, now()) > 0 {}
|
||||
out
|
||||
})
|
||||
}
|
||||
|
||||
fn state(&mut self) -> (u32, u32) {
|
||||
state(&mut self.daemon)
|
||||
}
|
||||
}
|
||||
|
||||
// ---- flags ----
|
||||
|
||||
mod serve {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn desk_flags_parse_both_or_neither() {
|
||||
let root = "/state";
|
||||
let base = [
|
||||
"--state-root", root, "--instance", "yggdrasil", "--bind", "127.0.0.1:17872", "--qdrant-url", "http://127.0.0.1:6333",
|
||||
"--ollama-url", "http://10.77.0.4:11434", "--embedding-model", "m",
|
||||
];
|
||||
let parse = |extra: &[&str]| parse_options([base.to_vec(), extra.to_vec()].concat().into_iter().map(String::from));
|
||||
assert_eq!(parse(&[]).unwrap().desk, None);
|
||||
let desk = parse(&["--desk-socket", "/run/huginn/desk.sock", "--desk-peer-uid", "993"]).unwrap().desk.unwrap();
|
||||
assert_eq!((desk.socket, desk.peer_uid), (PathBuf::from("/run/huginn/desk.sock"), 993));
|
||||
for bad in [
|
||||
vec!["--desk-socket", "/run/huginn/desk.sock"],
|
||||
vec!["--desk-peer-uid", "993"],
|
||||
vec!["--desk-socket", "desk.sock", "--desk-peer-uid", "993"],
|
||||
vec!["--desk-socket", "/run/desk.sock", "--desk-peer-uid", "bifrost"],
|
||||
vec!["--desk-socket", "/run/desk.sock", "--desk-peer-uid", "-1"],
|
||||
vec!["--desk-socket", "/run/desk.sock", "--desk-peer-uid", "4294967296"],
|
||||
vec!["--desk-socket", "/run/desk.sock", "--desk-peer-uid", "0"],
|
||||
vec!["--desk-socket", "/run/desk.sock", "--desk-peer-uid", "993", "--desk-peer-uid", "994"],
|
||||
vec!["--desk-socket", "/run/desk.sock", "--desk-socket", "/run/other.sock", "--desk-peer-uid", "993"],
|
||||
] {
|
||||
assert!(parse(&bad).is_err(), "{bad:?}");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// ---- the unit seam: a counting stream ----
|
||||
|
||||
/// A stream that counts what the door reads and writes through it.
|
||||
struct Counting {
|
||||
inner: UnixStream,
|
||||
reads: Arc<std::sync::atomic::AtomicU32>,
|
||||
writes: Arc<std::sync::atomic::AtomicU32>,
|
||||
}
|
||||
|
||||
impl Read for Counting {
|
||||
fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
|
||||
self.reads.fetch_add(1, Ordering::Relaxed);
|
||||
self.inner.read(buf)
|
||||
}
|
||||
}
|
||||
impl Write for Counting {
|
||||
fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
|
||||
self.writes.fetch_add(1, Ordering::Relaxed);
|
||||
self.inner.write(buf)
|
||||
}
|
||||
fn flush(&mut self) -> std::io::Result<()> {
|
||||
self.inner.flush()
|
||||
}
|
||||
}
|
||||
impl AsFd for Counting {
|
||||
fn as_fd(&self) -> BorrowedFd<'_> {
|
||||
self.inner.as_fd()
|
||||
}
|
||||
}
|
||||
|
||||
mod desk_door {
|
||||
use super::*;
|
||||
|
||||
fn pair() -> (UnixStream, Counting, Arc<std::sync::atomic::AtomicU32>, Arc<std::sync::atomic::AtomicU32>) {
|
||||
let (client, server) = UnixStream::pair().unwrap();
|
||||
client.set_read_timeout(Some(Duration::from_secs(5))).unwrap();
|
||||
let (reads, writes) = (Arc::default(), Arc::default());
|
||||
(client, Counting { inner: server, reads: Arc::clone(&reads), writes: Arc::clone(&writes) }, reads, writes)
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_refused_peer_has_nothing_read() {
|
||||
let root = tempfile::tempdir().unwrap();
|
||||
let (mut daemon, hash) = seeded(root.path());
|
||||
let registry = desk_registry().unwrap();
|
||||
let before = state(&mut daemon);
|
||||
let mine = my_uid();
|
||||
let (mut client, server, reads, writes) = pair();
|
||||
client.write_all(&framed(&message("m-1", &answer("x", DeskAuthority::Operator, &["x"], wire(&hash, R1))))).unwrap();
|
||||
let served = serve_connection(server, mine + 1, &mut daemon, ®istry, now(), |_| Ok(()));
|
||||
assert_eq!(served, Served::Refused { uid: mine, expected: mine + 1 });
|
||||
assert_eq!(served.to_string(), format!("desk door refused a peer with uid {mine} (expects {})", mine + 1));
|
||||
assert_eq!((reads.load(Ordering::Relaxed), writes.load(Ordering::Relaxed)), (0, 0), "no byte read or written");
|
||||
assert!(read_reply(&mut client).is_none(), "the stream was dropped");
|
||||
assert_eq!(state(&mut daemon), before);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn the_configured_peer_is_answered() {
|
||||
let root = tempfile::tempdir().unwrap();
|
||||
let (mut daemon, hash) = seeded(root.path());
|
||||
let registry = desk_registry().unwrap();
|
||||
let (mut client, server, reads, _) = pair();
|
||||
client.write_all(&framed(&message("m-1", &answer("x", DeskAuthority::Operator, &["x"], wire(&hash, R1))))).unwrap();
|
||||
let served = serve_connection(server, my_uid(), &mut daemon, ®istry, now(), |_| Ok(()));
|
||||
assert_eq!(served, Served::Answered);
|
||||
assert!(reads.load(Ordering::Relaxed) > 0);
|
||||
let reply = read_reply(&mut client).expect("a reply");
|
||||
let (_, response) = decode_desk_response(&reply).unwrap();
|
||||
assert_eq!(counted(response.unwrap()), ("n1".to_string(), Some("n1".to_string())));
|
||||
}
|
||||
|
||||
/// A header over the request bound is refused on the header alone: the transport reads the four
|
||||
/// bytes, sees the claim exceed the profile's maximum and never reads for the payload.
|
||||
#[test]
|
||||
fn an_oversize_header_is_refused_before_the_payload_is_read() {
|
||||
let root = tempfile::tempdir().unwrap();
|
||||
let (mut daemon, _) = seeded(root.path());
|
||||
let registry = desk_registry().unwrap();
|
||||
for length in [65537_u32, u32::MAX] {
|
||||
let (mut client, server, reads, writes) = pair();
|
||||
client.write_all(&length.to_be_bytes()).unwrap();
|
||||
drop(client); // a payload read, if one were made, would end in EOF rather than wait
|
||||
let served = serve_connection(server, my_uid(), &mut daemon, ®istry, now(), |_| Ok(()));
|
||||
let Served::Dropped(why) = served else { panic!("{served:?}") };
|
||||
assert!(why.contains("exceeds"), "{why}");
|
||||
assert_eq!((reads.load(Ordering::Relaxed), writes.load(Ordering::Relaxed)), (1, 0), "the header was the only read");
|
||||
}
|
||||
}
|
||||
|
||||
/// The pure core answers the catalog with the two desk schemas and nothing else, and a
|
||||
/// reply past the response bound is a typed refusal that still names the request.
|
||||
#[test]
|
||||
fn the_catalog_is_the_desk_schemas_and_an_oversize_reply_is_refused() {
|
||||
let root = tempfile::tempdir().unwrap();
|
||||
let (mut daemon, _) = seeded(root.path());
|
||||
let registry = desk_registry().unwrap();
|
||||
let catalog = CultNetMessage::SchemaCatalogRequest {
|
||||
message_id: "c-1".into(),
|
||||
include_schema_json: Some(true),
|
||||
schema_ids: None,
|
||||
kinds: None,
|
||||
};
|
||||
let reply = answer_desk_message(&mut daemon, ®istry, catalog, now());
|
||||
let text = format!("{reply:?}");
|
||||
assert!(text.contains("huginn.desk_request.v1") && text.contains("huginn.desk_response.v1"), "{text}");
|
||||
assert!(!text.contains("huginn.mind_request.v1"), "{text}");
|
||||
let other = answer_desk_message(&mut daemon, ®istry, CultNetMessage::Error { error: "x".into(), code: None, details: None }, now());
|
||||
assert!(matches!(other, CultNetMessage::Error { .. }));
|
||||
let sized = crate::desk_door::respond(&DeskResponse::Refused(MindRefusal::Unavailable { detail: "x".repeat(2000) }), "m-9", "answer", "huginn-yggdrasil", 512);
|
||||
let (id, answered) = decode_desk_response(&sized).unwrap();
|
||||
assert_eq!(id, "m-9");
|
||||
assert!(matches!(answered.unwrap(), DeskResponse::Refused(MindRefusal::ResponseTooLarge { limit: 512, .. })));
|
||||
}
|
||||
}
|
||||
|
||||
// ---- the door over a real socket ----
|
||||
|
||||
mod door_smoke {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn a_counted_answer_over_the_socket_reads_back() {
|
||||
assert_ne!(my_uid(), 0, "the door refuses uid 0, so these tests cannot run as root");
|
||||
let mut rig = rig(my_uid());
|
||||
let (before, hash) = (rig.state(), rig.hash.clone());
|
||||
let response = rig.serving(|path| reply_of(path, "m-1", &answer("x", DeskAuthority::Operator, &["x", "y"], wire(&hash, R1))));
|
||||
let DeskResponse::Answered { answer, counted_answer_id, .. } = &response else { panic!("{response:?}") };
|
||||
assert_eq!((answer.outcome.as_str(), counted_answer_id.as_deref(), answer.authority), ("Counted", Some("n1"), DeskAuthority::Operator));
|
||||
assert_eq!(rig.state().0, before.0 + 1, "the answer is stored");
|
||||
assert!(rig.index.seen.lock().unwrap().is_empty(), "a desk answer is not indexed");
|
||||
}
|
||||
|
||||
/// A pass takes at most `DESK_ACCEPTS_PER_PASS` waiting connections and says how many it took.
|
||||
#[test]
|
||||
fn a_pass_takes_four_connections_and_the_rest_wait() {
|
||||
let mut rig = rig(my_uid());
|
||||
let hash = rig.hash.clone();
|
||||
let mut clients: Vec<UnixStream> = (0..6)
|
||||
.map(|n| {
|
||||
let mut stream = connect(&rig.path);
|
||||
let request = answer("x", DeskAuthority::Member, &["x"], wire(&hash, &format!("00000000-0000-4000-8000-0000000001{n:02}")));
|
||||
stream.write_all(&framed(&message(&format!("m-{n}"), &request))).unwrap();
|
||||
stream
|
||||
})
|
||||
.collect();
|
||||
assert_eq!(rig.door.serve_ready(&mut rig.daemon, now()), 4);
|
||||
assert_eq!(rig.door.serve_ready(&mut rig.daemon, now()), 2);
|
||||
assert_eq!(rig.door.serve_ready(&mut rig.daemon, now()), 0, "an empty queue ends the pass");
|
||||
for client in &mut clients {
|
||||
assert!(read_reply(client).is_some());
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn another_uid_is_refused_and_the_store_is_unchanged() {
|
||||
let mut rig = rig(my_uid() + 1);
|
||||
let (before, hash) = (rig.state(), rig.hash.clone());
|
||||
let reply = rig.serving(|path| exchange(path, &framed(&message("m-1", &answer("x", DeskAuthority::Operator, &["x"], wire(&hash, R1))))));
|
||||
assert!(reply.is_none(), "a refused peer reads EOF");
|
||||
assert_eq!(rig.state(), before);
|
||||
assert!(rig.index.seen.lock().unwrap().is_empty());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn an_oversize_header_is_dropped_and_the_next_peer_is_served() {
|
||||
let mut rig = rig(my_uid());
|
||||
let (before, hash) = (rig.state(), rig.hash.clone());
|
||||
let (dropped, answered) = rig.serving(|path| {
|
||||
let dropped: Vec<_> = [65537_u32, u32::MAX]
|
||||
.iter()
|
||||
.map(|length| {
|
||||
let mut stream = connect(path);
|
||||
stream.write_all(&length.to_be_bytes()).unwrap();
|
||||
read_reply(&mut stream).is_none()
|
||||
})
|
||||
.collect();
|
||||
(dropped, reply_of(path, "m-2", &answer("x", DeskAuthority::Operator, &["x"], wire(&hash, R1))))
|
||||
});
|
||||
assert_eq!(dropped, [true, true]);
|
||||
counted(answered);
|
||||
assert_eq!(rig.state().0, before.0 + 1);
|
||||
}
|
||||
|
||||
/// A peer that writes two header bytes and holds is cut off by the read timeout, the next
|
||||
/// connection is answered in time, and the RUDP door is served in the same loop.
|
||||
#[test]
|
||||
fn a_stalled_peer_is_cut_off() {
|
||||
let Rig { _dir, _root, path, daemon, door, hash, .. } = rig(my_uid());
|
||||
let mut daemon = daemon;
|
||||
let mut door = door;
|
||||
let mut hub = bind("127.0.0.1:0".parse().unwrap(), &daemon.runtime_id()).unwrap();
|
||||
let endpoint = format!("rudp://{}", hub.local_addr().unwrap());
|
||||
let registry = schema_registry().unwrap();
|
||||
let stopping = Arc::new(AtomicBool::new(false));
|
||||
let loop_stopping = Arc::clone(&stopping);
|
||||
let serving = std::thread::spawn(move || {
|
||||
run(&mut daemon, &mut hub, ®istry, Some(&mut door), &loop_stopping, &ServeOptions::default()).unwrap();
|
||||
});
|
||||
|
||||
let mut stalled = connect(&path);
|
||||
stalled.write_all(&[0, 0]).unwrap();
|
||||
let started = Instant::now();
|
||||
let mut rudp = CultMesh::create_rudp_client_for_endpoint(
|
||||
"desk-stall-test".to_string(),
|
||||
CULTNET_OPERATION_CONNECTION_ID,
|
||||
&endpoint,
|
||||
CultMeshRudpSocketOptions::default(),
|
||||
)
|
||||
.unwrap();
|
||||
rudp.connect(Vec::new()).unwrap();
|
||||
for _ in 0..2000 {
|
||||
let _ = rudp.receive_once();
|
||||
if rudp.connected() {
|
||||
break;
|
||||
}
|
||||
std::thread::sleep(Duration::from_millis(2));
|
||||
}
|
||||
assert!(rudp.connected(), "the RUDP door was served during the stall");
|
||||
rudp.send_schema_message(&encode_request("w-1", &HuginnMindRequest::Whoami, None).unwrap()).unwrap();
|
||||
let answered = reply_of(&path, "m-1", &answer("x", DeskAuthority::Operator, &["x"], wire(&hash, R1)));
|
||||
assert!(started.elapsed() < Duration::from_millis(4250), "{:?}", started.elapsed());
|
||||
counted(answered);
|
||||
let mut whoami = None;
|
||||
for _ in 0..2000 {
|
||||
rudp.poll_resends().unwrap();
|
||||
if let Some(reply) = rudp.receive_schema_message_once().unwrap() {
|
||||
whoami = Some(reply);
|
||||
break;
|
||||
}
|
||||
std::thread::sleep(Duration::from_millis(2));
|
||||
}
|
||||
let (_, whoami) = decode_response(&whoami.expect("whoami answered")).unwrap();
|
||||
assert!(matches!(whoami.unwrap(), HuginnMindResponse::Whoami(_)));
|
||||
assert!(read_reply(&mut stalled).is_none(), "the stalled connection was closed by the door");
|
||||
stopping.store(true, Ordering::Relaxed);
|
||||
serving.join().unwrap();
|
||||
}
|
||||
|
||||
fn raw(service: &str, schema: &str, operation: &str, id: &str, payload: Vec<u8>) -> CultNetMessage {
|
||||
CultNetMessage::OperationRequest {
|
||||
message_id: id.into(),
|
||||
service_id: service.into(),
|
||||
operation: operation.into(),
|
||||
payload_schema: schema.into(),
|
||||
payload_encoding: "messagepack-base64".into(),
|
||||
payload: STANDARD.encode(payload),
|
||||
source_runtime_id: None,
|
||||
target_runtime_id: None,
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn malformed_frames_get_typed_failures_or_are_dropped() {
|
||||
let mut rig = rig(my_uid());
|
||||
let (before, hash) = (rig.state(), rig.hash.clone());
|
||||
let good = answer("x", DeskAuthority::Operator, &["x"], wire(&hash, R1));
|
||||
let good_payload = rmp_serde::to_vec_named(&good).unwrap();
|
||||
let mut with_answerer = serde_json::to_value(&good).unwrap();
|
||||
with_answerer["Answer"]["answerer"] = serde_json::json!("someone-else");
|
||||
let long_id = "m".repeat(257);
|
||||
let outcomes = rig.serving(|path| {
|
||||
let garbage = {
|
||||
let mut payload = vec![0xc1, 0xc1, 0xc1];
|
||||
payload.extend_from_slice(b"not cultnet");
|
||||
let mut bytes = (payload.len() as u32).to_be_bytes().to_vec();
|
||||
bytes.extend(payload);
|
||||
exchange(path, &bytes).is_none()
|
||||
};
|
||||
let failure = |m: CultNetMessage| failure_of(path, &framed(&m));
|
||||
(
|
||||
garbage,
|
||||
failure(raw(MIND_SERVICE_ID, MIND_REQUEST_SCHEMA, "answer", "m-1", good_payload.clone())).code,
|
||||
failure(raw(DESK_SERVICE_ID, MIND_REQUEST_SCHEMA, "answer", "m-2", good_payload.clone())).code,
|
||||
failure(raw(DESK_SERVICE_ID, DESK_REQUEST_SCHEMA, "forks", "m-3", good_payload.clone())).code,
|
||||
failure(raw(DESK_SERVICE_ID, DESK_REQUEST_SCHEMA, "answer", "m-4", rmp_serde::to_vec_named(&with_answerer).unwrap())),
|
||||
exchange(path, &framed(&raw(DESK_SERVICE_ID, DESK_REQUEST_SCHEMA, "answer", &long_id, good_payload.clone()))),
|
||||
)
|
||||
});
|
||||
let (garbage, mind_service, mind_schema, operation, answerer, long) = outcomes;
|
||||
assert!(garbage, "bytes that are not CultNet are dropped");
|
||||
assert_eq!(mind_service, "wrong-service");
|
||||
assert_eq!(mind_schema, "wrong-payload-schema");
|
||||
assert_eq!(operation, "operation-mismatch");
|
||||
assert_eq!(answerer.code, "payload-not-a-request");
|
||||
assert!(answerer.message.contains("answerer") || answerer.message.contains("Answer"), "{}", answerer.message);
|
||||
assert!(matches!(long, Some(CultNetMessage::Error { .. })), "{long:?}");
|
||||
assert_eq!(rig.state(), before, "nothing stored in any case");
|
||||
}
|
||||
|
||||
/// The RUDP door has no desk method: a desk answer document in an Admit batch is refused,
|
||||
/// and a desk request payload is the wrong payload schema for the mind service.
|
||||
#[test]
|
||||
fn the_rudp_door_cannot_write_or_route_desk() {
|
||||
use huginn_mind::eureka_pipeline::{PipelineDeskAnswer, DeskOutcome, RequestId, Sha256Hex};
|
||||
let root = tempfile::tempdir().unwrap();
|
||||
let (mut daemon, hash) = seeded(root.path());
|
||||
let before = state(&mut daemon);
|
||||
let forged = PipelineDocument::DeskAnswer(PipelineDeskAnswer {
|
||||
campaign: slug("eureka-state"),
|
||||
fork: short(FORK),
|
||||
sequence: 1,
|
||||
options_hash: Sha256Hex(hash.clone()),
|
||||
choice: Some(short("A")),
|
||||
note: None,
|
||||
answered_at: short("2026-09-16T11:00:00Z"),
|
||||
received_at: short("2026-09-16T11:00:00Z"),
|
||||
claim_in_future: false,
|
||||
request_id: RequestId(R1.into()),
|
||||
body_sha256: Sha256Hex("0".repeat(64)),
|
||||
answerer: short("x"),
|
||||
authority: DeskAuthority::Operator,
|
||||
outcome: DeskOutcome::Counted,
|
||||
});
|
||||
let admitted = daemon.handle(HuginnMindRequest::Admit(batch(INSTANCE, vec![forged])), now()).answered();
|
||||
assert!(matches!(admitted, HuginnMindResponse::Admit(huginn_mind::PipelineAdmissionOutcome::Refused(_))), "{admitted:?}");
|
||||
assert_eq!(state(&mut daemon), before);
|
||||
|
||||
let registry = schema_registry().unwrap();
|
||||
let mut bodies = crate::bodies::DeferredBodies::new(1 << 20, Duration::from_secs(60));
|
||||
let desk = raw(MIND_SERVICE_ID, DESK_REQUEST_SCHEMA, "answer", "m-1", rmp_serde::to_vec_named(&answer("x", DeskAuthority::Operator, &["x"], wire(&hash, R1))).unwrap());
|
||||
let CultNetMessage::OperationResponse { payload_schema, diagnostics, .. } = crate::serve::answer(&mut daemon, ®istry, &mut bodies, desk, now()).reply() else {
|
||||
panic!("expected a response")
|
||||
};
|
||||
assert_eq!(payload_schema, huginn_mind::envelope::FAILURE_SCHEMA);
|
||||
assert_eq!(diagnostics, ["wrong-payload-schema"]);
|
||||
assert_eq!(state(&mut daemon), before);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_stale_answer_carries_no_card() {
|
||||
let mut rig = rig(my_uid());
|
||||
let response = rig.serving(|path| reply_of(path, "m-1", &answer("x", DeskAuthority::Operator, &["x"], wire(&"0".repeat(64), R1))));
|
||||
let DeskResponse::Answered { answer, counted_answer_id, .. } = response else { panic!("a stale answer is stored") };
|
||||
assert_eq!(answer.code, Some(ForkAnswerCode::StaleFork));
|
||||
assert_eq!(answer.outcome, "Refused");
|
||||
assert_eq!(counted_answer_id, None);
|
||||
// The reply type has no field a card could ride in; the encoded reply holds nothing but the answer.
|
||||
assert_eq!(answer.options_hash, "0".repeat(64));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn deciders_ride_each_request() {
|
||||
let mut rig = rig(my_uid());
|
||||
let (before, hash) = (rig.state(), rig.hash.clone());
|
||||
let out = rig.serving(|path| {
|
||||
let ask = |id: &str, who: &str, deciders: &[&str], request_id: &str| {
|
||||
reply_of(path, id, &answer(who, DeskAuthority::Operator, deciders, wire(&hash, request_id)))
|
||||
};
|
||||
let first = ask("m-1", "x", &["x"], R1);
|
||||
let other_list = ask("m-2", "x", &["y"], R2);
|
||||
let first_list_again = ask("m-3", "x", &["x"], R3);
|
||||
let mut empty = answer("x", DeskAuthority::Operator, &["x"], wire(&hash, R3));
|
||||
let DeskRequest::Answer { deciders, .. } = &mut empty else { unreachable!() };
|
||||
deciders.clear();
|
||||
let empty_bytes = framed(&message("m-4", &empty));
|
||||
(first, other_list, first_list_again, failure_of(path, &empty_bytes))
|
||||
});
|
||||
let (first, other_list, again, empty) = out;
|
||||
let DeskResponse::Answered { answer, .. } = &first else { panic!("{first:?}") };
|
||||
assert_eq!((answer.outcome.as_str(), answer.authority), ("Counted", DeskAuthority::Operator));
|
||||
let DeskResponse::Answered { answer, .. } = &other_list else { panic!("{other_list:?}") };
|
||||
assert_eq!(answer.code, Some(ForkAnswerCode::NotADecider), "the second list does not name x");
|
||||
let DeskResponse::Answered { answer, .. } = &again else { panic!("{again:?}") };
|
||||
assert_eq!(answer.outcome, "Revision", "nothing was kept between calls: the first list counts again");
|
||||
assert_eq!((empty.code.as_str(), empty.message.contains("deciders")), ("payload-not-a-request", true));
|
||||
assert_eq!(rig.state().0, before.0 + 3, "three answers landed: two counted, one refused");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn startup_replaces_a_stale_socket_and_refuses_a_live_one() {
|
||||
let dir = tempfile::tempdir().unwrap();
|
||||
let path = dir.path().join("desk.sock");
|
||||
// A stale socket file: bound once, listener gone.
|
||||
drop(UnixListener::bind(&path).unwrap());
|
||||
assert!(path.exists());
|
||||
let door = bind_desk(&path, my_uid()).expect("a stale socket is replaced");
|
||||
assert_eq!(std::fs::metadata(&path).unwrap().permissions().mode() & 0o777, 0o660);
|
||||
// A live listener refuses a second bind.
|
||||
let error = format!("{:#}", bind_desk(&path, my_uid()).err().expect("a live socket refuses"));
|
||||
assert!(error.contains("live"), "{error}");
|
||||
assert!(path.exists(), "the live server's socket was not unlinked");
|
||||
drop(door);
|
||||
assert!(!path.exists(), "drop unlinks the path");
|
||||
// A socket the door cannot probe is neither stale nor free: it is left alone.
|
||||
let guarded = dir.path().join("guarded.sock");
|
||||
let _held = UnixListener::bind(&guarded).unwrap();
|
||||
std::fs::set_permissions(&guarded, std::fs::Permissions::from_mode(0o000)).unwrap();
|
||||
let error = format!("{:#}", bind_desk(&guarded, my_uid()).err().expect("an unprobeable socket refuses"));
|
||||
assert!(error.contains("probing"), "{error}");
|
||||
assert!(guarded.exists());
|
||||
// A path the door cannot even look at is not a free path.
|
||||
let sealed = dir.path().join("sealed");
|
||||
std::fs::create_dir(&sealed).unwrap();
|
||||
std::fs::set_permissions(&sealed, std::fs::Permissions::from_mode(0o000)).unwrap();
|
||||
let error = format!("{:#}", bind_desk(&sealed.join("desk.sock"), my_uid()).err().expect("an unreadable path refuses"));
|
||||
assert!(error.contains("reading"), "{error}");
|
||||
std::fs::set_permissions(&sealed, std::fs::Permissions::from_mode(0o700)).unwrap();
|
||||
// A regular file is not a socket.
|
||||
std::fs::write(&path, b"x").unwrap();
|
||||
assert!(bind_desk(&path, my_uid()).is_err());
|
||||
assert!(path.is_file(), "the file was left alone");
|
||||
// A relative path and uid 0 are refused.
|
||||
assert!(bind_desk(Path::new("desk.sock"), my_uid()).is_err());
|
||||
std::fs::remove_file(&path).unwrap();
|
||||
assert!(bind_desk(&path, 0).is_err());
|
||||
assert!(!path.exists());
|
||||
|
||||
// A held mind binds no desk socket, and a free one binds it after the RUDP socket.
|
||||
let root = tempfile::tempdir().unwrap();
|
||||
let held = crate::daemon::tests::open_unindexed(root.path(), &slug(INSTANCE)).unwrap();
|
||||
let options = Options {
|
||||
state_root: root.path().to_path_buf(),
|
||||
instance: slug(INSTANCE),
|
||||
bind: "127.0.0.1:0".parse().unwrap(),
|
||||
qdrant_url: "http://127.0.0.1:1".into(),
|
||||
ollama_urls: vec!["http://127.0.0.1:1".into()],
|
||||
reserve_ollama_urls: Vec::new(),
|
||||
embedding_model: "a-model".into(),
|
||||
forge_url: None,
|
||||
desk: Some(crate::serve::DeskOptions { socket: path.clone(), peer_uid: my_uid() }),
|
||||
};
|
||||
let started = |options: &Options| {
|
||||
let (embedder, index) = crate::index::fakes::pair();
|
||||
startup(options, embedder, index, Backoff::default())
|
||||
};
|
||||
assert!(started(&options).is_err());
|
||||
assert!(!path.exists(), "a mind that will not open binds neither");
|
||||
drop(held);
|
||||
let up = started(&options).expect("both gates open");
|
||||
assert!(up.desk.is_some() && path.exists());
|
||||
drop(up);
|
||||
assert!(!path.exists());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_reply_lost_after_commit_replays_on_retry() {
|
||||
let mut rig = rig(my_uid());
|
||||
let hash = rig.hash.clone();
|
||||
let bytes = framed(&message("m-1", &answer("x", DeskAuthority::Operator, &["x"], wire(&hash, R1))));
|
||||
let (replayed, first_state) = {
|
||||
let sent = bytes.clone();
|
||||
rig.serving(move |path| {
|
||||
let mut stream = connect(path);
|
||||
stream.write_all(&sent).unwrap();
|
||||
drop(stream); // closes without reading the reply
|
||||
});
|
||||
let first_state = rig.state();
|
||||
let replayed = rig.serving(|path| exchange(path, &bytes).expect("the retry is answered"));
|
||||
(replayed, first_state)
|
||||
};
|
||||
let (_, response) = decode_desk_response(&replayed).unwrap();
|
||||
assert_eq!(counted(response.unwrap()).0, "n1", "the retry is answered with the stored answer");
|
||||
assert_eq!(rig.state(), first_state, "the retry stored nothing");
|
||||
assert!(rig.index.seen.lock().unwrap().is_empty(), "a desk answer is not indexed");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_frame_cut_off_before_the_door_has_it_stores_nothing() {
|
||||
let mut rig = rig(my_uid());
|
||||
let (before, hash) = (rig.state(), rig.hash.clone());
|
||||
let bytes = framed(&message("m-1", &answer("x", DeskAuthority::Operator, &["x"], wire(&hash, R1))));
|
||||
let cut = rig.serving(|path| {
|
||||
let mut stream = connect(path);
|
||||
stream.write_all(&bytes[..bytes.len() / 2]).unwrap();
|
||||
stream.shutdown(std::net::Shutdown::Write).unwrap();
|
||||
read_reply(&mut stream).is_none()
|
||||
});
|
||||
assert!(cut);
|
||||
assert_eq!(rig.state(), before);
|
||||
assert!(rig.index.seen.lock().unwrap().is_empty());
|
||||
let full = rig.serving(|path| exchange(path, &bytes).expect("answered"));
|
||||
let (_, response) = decode_desk_response(&full).unwrap();
|
||||
assert_eq!(counted(response.unwrap()).0, "n1");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn answer_deciders_64_accepted_65_refused() {
|
||||
let mut rig = rig(my_uid());
|
||||
let hash = rig.hash.clone();
|
||||
let list = |n: usize| (0..n).map(|d| format!("d{d}")).collect::<Vec<_>>();
|
||||
let (sixty_four, sixty_five_bytes) = {
|
||||
let (sixty_four, sixty_five) = (list(64), list(65));
|
||||
let sixty_four_refs: Vec<&str> = sixty_four.iter().map(String::as_str).collect();
|
||||
let sixty_five_refs: Vec<&str> = sixty_five.iter().map(String::as_str).collect();
|
||||
(
|
||||
answer("d0", DeskAuthority::Member, &sixty_four_refs, wire(&hash, R1)),
|
||||
framed(&message("m-2", &answer("d0", DeskAuthority::Member, &sixty_five_refs, wire(&hash, R2)))),
|
||||
)
|
||||
};
|
||||
let served = rig.serving(|path| reply_of(path, "m-1", &sixty_four));
|
||||
counted(served);
|
||||
let after_first = rig.state();
|
||||
let refused = rig.serving(|path| failure_of(path, &sixty_five_bytes));
|
||||
assert_eq!(refused.code, "payload-not-a-request");
|
||||
assert!(refused.message.contains("deciders"), "{}", refused.message);
|
||||
assert_eq!(rig.state(), after_first, "count and head unchanged");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn answer_deciders_empty_or_duplicate_refused() {
|
||||
let mut rig = rig(my_uid());
|
||||
let (before, hash) = (rig.state(), rig.hash.clone());
|
||||
let empty = framed(&message("m-1", &answer("x", DeskAuthority::Member, &[], wire(&hash, R1))));
|
||||
let twice = framed(&message("m-2", &answer("x", DeskAuthority::Member, &["x", "x"], wire(&hash, R2))));
|
||||
let (empty, twice, one) = rig.serving(|path| {
|
||||
(failure_of(path, &empty), failure_of(path, &twice), reply_of(path, "m-3", &answer("x", DeskAuthority::Member, &["x"], wire(&hash, R3))))
|
||||
});
|
||||
for refused in [empty, twice] {
|
||||
assert_eq!(refused.code, "payload-not-a-request");
|
||||
assert!(refused.message.contains("deciders"), "{}", refused.message);
|
||||
}
|
||||
counted(one);
|
||||
assert_eq!(rig.state().0, before.0 + 1, "only the one-id list stored an answer");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_mind_refusal_and_a_stored_refusal_are_typed_replies() {
|
||||
let mut rig = rig(my_uid());
|
||||
let (before, hash) = (rig.state(), rig.hash.clone());
|
||||
let mut foreign = answer("x", DeskAuthority::Member, &["x"], wire(&hash, R1));
|
||||
let DeskRequest::Answer { mind, .. } = &mut foreign else { unreachable!() };
|
||||
*mind = slug(OTHER);
|
||||
let (foreign, not_a_decider) = rig.serving(|path| {
|
||||
(reply_of(path, "m-1", &foreign), reply_of(path, "m-2", &answer("x", DeskAuthority::Member, &["y"], wire(&hash, R2))))
|
||||
});
|
||||
assert_eq!(foreign, DeskResponse::Refused(MindRefusal::ForeignInstance { declared: OTHER.into(), mind: INSTANCE.into() }));
|
||||
assert_eq!(rig.state().0, before.0 + 1, "only the stored refusal landed");
|
||||
let DeskResponse::Answered { answer, counted_answer_id, .. } = not_a_decider else { panic!("a stored refusal is an answer") };
|
||||
assert_eq!((answer.code, counted_answer_id), (Some(ForkAnswerCode::NotADecider), None));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn forks_and_fork_are_refused_unavailable_while_reads_are_unbuilt() {
|
||||
let mut rig = rig(my_uid());
|
||||
let before = rig.state();
|
||||
let forks = DeskRequest::Forks { mind: slug(INSTANCE), subject: short("x"), deciders: vec![short("x")] };
|
||||
let fork = DeskRequest::Fork { mind: slug(INSTANCE), subject: short("x"), deciders: vec![short("x")], fork_id: FORK.into() };
|
||||
let (forks, fork) = rig.serving(|path| (reply_of(path, "m-1", &forks), reply_of(path, "m-2", &fork)));
|
||||
for response in [forks, fork] {
|
||||
let DeskResponse::Refused(MindRefusal::Unavailable { detail }) = response else { panic!("{response:?}") };
|
||||
assert!(detail.contains("desk reads not built"), "{detail}");
|
||||
}
|
||||
assert_eq!(rig.state(), before);
|
||||
assert!(rig.index.seen.lock().unwrap().is_empty());
|
||||
}
|
||||
}
|
||||
|
||||
// ---- the bounds, pinned as the numbers the spec states ----
|
||||
|
||||
mod bounds {
|
||||
use crate::desk_door::Deadlined;
|
||||
use cultnet_rs::{CultNetSchemaKind, CultNetSchemaRegistration, CultNetSchemaRegistry};
|
||||
use huginn_mind::desk_wire::DESK_MAX_RESPONSE_BYTES;
|
||||
|
||||
use super::*;
|
||||
|
||||
/// One connection costs the loop one second and no more, whatever the peer does: the stated
|
||||
/// worst stall is `DESK_ACCEPTS_PER_PASS` of them, four seconds. The lower edge fails a
|
||||
/// shorter bound, the upper a longer one or one that restarts on every byte.
|
||||
fn served_within_one_second(client: impl FnOnce(UnixStream) + Send) {
|
||||
let mut rig = rig(my_uid());
|
||||
let peer = connect(&rig.path);
|
||||
std::thread::scope(|scope| {
|
||||
scope.spawn(move || client(peer));
|
||||
let started = Instant::now();
|
||||
assert_eq!(rig.door.serve_ready(&mut rig.daemon, now()), 1);
|
||||
let took = started.elapsed();
|
||||
assert!(took >= Duration::from_millis(950), "the connection was cut after {took:?}, short of the one-second deadline (the kernel timeout is jiffy-granular)");
|
||||
assert!(took < Duration::from_millis(1250), "the connection held the loop for {took:?}");
|
||||
});
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_silent_peer_is_cut_off_at_one_second() {
|
||||
served_within_one_second(|peer| {
|
||||
std::thread::sleep(Duration::from_millis(1500));
|
||||
drop(peer);
|
||||
});
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_dripping_peer_is_cut_off_at_one_second() {
|
||||
let bytes = framed(&message("m-1", &answer("x", DeskAuthority::Operator, &["x"], wire("0", R1))));
|
||||
served_within_one_second(move |mut peer| {
|
||||
for byte in bytes.iter().take(20) {
|
||||
if peer.write_all(&[*byte]).is_err() {
|
||||
break;
|
||||
}
|
||||
std::thread::sleep(Duration::from_millis(300));
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
/// The write side of the deadline: a peer that stops reading is let go of at one second, the
|
||||
/// socket's own send timeout reset to what is left before each write.
|
||||
#[test]
|
||||
fn a_peer_that_never_reads_is_cut_off_at_one_second() {
|
||||
let (_peer, server) = UnixStream::pair().unwrap();
|
||||
let mut stream = Deadlined::new(server, DESK_CONNECTION_DEADLINE);
|
||||
let (chunk, started) = (vec![0_u8; 64 * 1024], Instant::now());
|
||||
let error = loop {
|
||||
if let Err(error) = stream.write(&chunk) {
|
||||
break error;
|
||||
}
|
||||
};
|
||||
let took = started.elapsed();
|
||||
// The kernel's send timeout reads as WouldBlock; the deadline already spent reads as TimedOut.
|
||||
assert!(matches!(error.kind(), std::io::ErrorKind::WouldBlock | std::io::ErrorKind::TimedOut), "{error:?}");
|
||||
assert!(took >= Duration::from_millis(950) && took < Duration::from_millis(1250), "the write was cut after {took:?}");
|
||||
assert_eq!(stream.write(&chunk).unwrap_err().kind(), std::io::ErrorKind::TimedOut, "once spent, nothing more is written");
|
||||
}
|
||||
|
||||
/// A catalog is a reply like any other and is measured against the response bound: one that
|
||||
/// encodes to exactly the bound is sent, one byte more is refused naming the size.
|
||||
#[test]
|
||||
fn the_catalog_reply_is_bounded_at_exactly_the_response_limit() {
|
||||
assert_eq!(DESK_MAX_RESPONSE_BYTES, 65_536);
|
||||
let root = tempfile::tempdir().unwrap();
|
||||
let (mut daemon, _) = seeded(root.path());
|
||||
let request = CultNetMessage::SchemaCatalogRequest { message_id: "c-1".into(), include_schema_json: Some(true), schema_ids: None, kinds: None };
|
||||
let registry = |filler: usize| {
|
||||
let mut registry = desk_registry().unwrap();
|
||||
registry
|
||||
.register(CultNetSchemaRegistration {
|
||||
schema_id: "filler".into(),
|
||||
kind: CultNetSchemaKind::WireMessage,
|
||||
wire_contracts: vec![CultNetWireContract::CultNetSchemaV0],
|
||||
schema_version: Some("filler".into()),
|
||||
document_type: None,
|
||||
title: None,
|
||||
schema_json: Some(format!("\"{}\"", "x".repeat(filler))),
|
||||
})
|
||||
.unwrap();
|
||||
registry
|
||||
};
|
||||
let size = |registry: &CultNetSchemaRegistry| {
|
||||
encode_cultnet_message_to_vec(®istry.create_catalog_response(&request).unwrap(), CultNetWireContract::CultNetSchemaV0).unwrap().len()
|
||||
};
|
||||
let mut filler = DESK_MAX_RESPONSE_BYTES - size(®istry(0));
|
||||
while size(®istry(filler)) > DESK_MAX_RESPONSE_BYTES {
|
||||
filler -= 1;
|
||||
}
|
||||
while size(®istry(filler + 1)) <= DESK_MAX_RESPONSE_BYTES {
|
||||
filler += 1;
|
||||
}
|
||||
let at_limit = answer_desk_message(&mut daemon, ®istry(filler), request.clone(), now());
|
||||
assert!(matches!(at_limit, CultNetMessage::SchemaCatalogResponse { .. }), "{at_limit:?}");
|
||||
assert!(encode_cultnet_message_to_vec(&at_limit, CultNetWireContract::CultNetSchemaV0).unwrap().len() <= DESK_MAX_RESPONSE_BYTES);
|
||||
let over = answer_desk_message(&mut daemon, ®istry(filler + 1), request, now());
|
||||
let CultNetMessage::Error { error, .. } = over else { panic!("{over:?}") };
|
||||
assert!(error.contains("65536"), "{error}");
|
||||
}
|
||||
|
||||
/// A fresh answer stores one document; the retry that replays it stores none.
|
||||
#[test]
|
||||
fn a_fresh_answer_stores_one_document_and_its_replay_none() {
|
||||
let mut rig = rig(my_uid());
|
||||
let (before, hash) = (rig.state(), rig.hash.clone());
|
||||
let request = answer("x", DeskAuthority::Operator, &["x"], wire(&hash, R1));
|
||||
let first = rig.daemon.handle_desk(request.clone(), now());
|
||||
assert_eq!(counted(first.clone()).0, "n1");
|
||||
assert_eq!(rig.state().0, before.0 + 1);
|
||||
let replay = rig.daemon.handle_desk(request, now());
|
||||
assert_eq!(replay, first);
|
||||
assert_eq!(rig.state().0, before.0 + 1, "the replay stored nothing");
|
||||
}
|
||||
}
|
||||
|
|
@ -15,12 +15,17 @@
|
|||
|
||||
pub mod bodies;
|
||||
pub mod daemon;
|
||||
pub mod desk_door;
|
||||
pub mod index;
|
||||
pub mod serve;
|
||||
pub mod sources;
|
||||
|
||||
#[cfg(test)]
|
||||
mod desk_door_tests;
|
||||
|
||||
#[cfg(test)]
|
||||
mod semantic_tests;
|
||||
|
||||
pub use daemon::{Daemon, Found, Handled, Hits, IndexSink, Search, SearchTicket, runtime_id};
|
||||
pub use serve::{Options, ServeOptions, parse_options, run, startup};
|
||||
pub use desk_door::DeskDoor;
|
||||
pub use serve::{Options, ServeOptions, Started, parse_options, run, startup};
|
||||
|
|
|
|||
|
|
@ -9,13 +9,13 @@ use anyhow::{Context, Result};
|
|||
use huginn_daemon::index::Backoff;
|
||||
use huginn_daemon::index::ollama::pool;
|
||||
use huginn_daemon::index::qdrant::QdrantIndex;
|
||||
use huginn_daemon::serve::{ServeOptions, parse_options, run, startup};
|
||||
use huginn_daemon::serve::{ServeOptions, Started, parse_options, run, startup};
|
||||
|
||||
fn main() -> Result<()> {
|
||||
let options = parse_options(std::env::args().skip(1))?;
|
||||
let embedder = pool(&options.ollama_urls, &options.reserve_ollama_urls, &options.embedding_model);
|
||||
let index = QdrantIndex::new(&options.qdrant_url);
|
||||
let (mut daemon, mut hub, registry) = match startup(&options, embedder, index, Backoff::default()) {
|
||||
let Started { mut daemon, mut hub, registry, mut desk } = match startup(&options, embedder, index, Backoff::default()) {
|
||||
Ok(started) => started,
|
||||
Err(error) => {
|
||||
eprintln!("huginn-daemon refuses to serve {}: {error:#}", options.instance.0);
|
||||
|
|
@ -33,5 +33,8 @@ fn main() -> Result<()> {
|
|||
}
|
||||
|
||||
eprintln!("huginn-daemon serving {} on {}", options.instance.0, hub.local_addr()?);
|
||||
run(&mut daemon, &mut hub, ®istry, &stopping, &ServeOptions::default())
|
||||
if let Some(door) = &desk {
|
||||
eprintln!("huginn-daemon serving the desk door on {}", door.path().display());
|
||||
}
|
||||
run(&mut daemon, &mut hub, ®istry, desk.as_mut(), &stopping, &ServeOptions::default())
|
||||
}
|
||||
|
|
|
|||
|
|
@ -176,7 +176,7 @@ impl Harness {
|
|||
let stopping = Arc::new(AtomicBool::new(false));
|
||||
let loop_stopping = Arc::clone(&stopping);
|
||||
let options = ServeOptions::default();
|
||||
let serving = std::thread::spawn(move || run(&mut daemon, &mut hub, ®istry, &loop_stopping, &options));
|
||||
let serving = std::thread::spawn(move || run(&mut daemon, &mut hub, ®istry, None, &loop_stopping, &options));
|
||||
Self { endpoint, stopping, serving: Some(serving), embedder, index, _root: root }
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -79,6 +79,7 @@ use crate::bodies::{
|
|||
DEFERRED_BUDGET_BYTES, DEFERRED_CHUNK_BYTES, DeferredBodies,
|
||||
};
|
||||
use crate::daemon::{Daemon, Handled, IndexSink, Search, SearchTicket, runtime_id};
|
||||
use crate::desk_door::{DeskDoor, bind_desk};
|
||||
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};
|
||||
|
|
@ -118,12 +119,22 @@ pub struct Options {
|
|||
pub embedding_model: String,
|
||||
/// The forge a persona's activity read goes to; absent, that read is refused.
|
||||
pub forge_url: Option<String>,
|
||||
/// The desk door: the unix socket Bifrost's receiver answers on, and the one uid that may
|
||||
/// speak on it. Absent, the daemon serves the RUDP door alone.
|
||||
pub desk: Option<DeskOptions>,
|
||||
}
|
||||
|
||||
/// Where the desk door listens and who may speak there.
|
||||
#[derive(Clone, Debug, PartialEq, Eq)]
|
||||
pub struct DeskOptions {
|
||||
pub socket: PathBuf,
|
||||
pub peer_uid: u32,
|
||||
}
|
||||
|
||||
/// Exactly `--state-root`, `--instance`, `--bind`, `--qdrant-url`,
|
||||
/// `--ollama-url` and `--embedding-model`, each required, each once, except
|
||||
/// `--ollama-url`, which repeats for each GPU host, and the optional
|
||||
/// `--reserve-ollama-url`, which repeats for each CPU host, and the optional `--forge-url`, once. The URLs are plain
|
||||
/// `--reserve-ollama-url`, which repeats for each CPU host, and the optional `--forge-url`, once, and the optional pair `--desk-socket` (an absolute path) and `--desk-peer-uid` (a uid other than 0), both or neither, each once. The URLs are plain
|
||||
/// `http://`: this build carries no TLS.
|
||||
pub fn parse_options(args: impl Iterator<Item = String>) -> Result<Options> {
|
||||
let mut args = args;
|
||||
|
|
@ -132,7 +143,7 @@ pub fn parse_options(args: impl Iterator<Item = String>) -> Result<Options> {
|
|||
while let Some(name) = args.next() {
|
||||
let name = name.strip_prefix("--").with_context(|| format!("expected --option, got {name:?}"))?;
|
||||
ensure!(
|
||||
matches!(name, "state-root" | "instance" | "bind" | "qdrant-url" | "ollama-url" | "reserve-ollama-url" | "embedding-model" | "forge-url"),
|
||||
matches!(name, "state-root" | "instance" | "bind" | "qdrant-url" | "ollama-url" | "reserve-ollama-url" | "embedding-model" | "forge-url" | "desk-socket" | "desk-peer-uid"),
|
||||
"unsupported Huginn option --{name}"
|
||||
);
|
||||
let value = args.next().with_context(|| format!("missing value for --{name}"))?;
|
||||
|
|
@ -168,6 +179,17 @@ pub fn parse_options(args: impl Iterator<Item = String>) -> Result<Options> {
|
|||
reserve_ollama_urls,
|
||||
embedding_model,
|
||||
forge_url: values.contains_key("forge-url").then(|| url("forge-url")).transpose()?,
|
||||
desk: match (values.get("desk-socket"), values.get("desk-peer-uid")) {
|
||||
(None, None) => None,
|
||||
(Some(socket), Some(uid)) => {
|
||||
let socket = PathBuf::from(socket);
|
||||
ensure!(socket.is_absolute(), "--desk-socket must be an absolute path");
|
||||
let peer_uid: u32 = uid.parse().with_context(|| format!("--desk-peer-uid must be a uid, got {uid:?}"))?;
|
||||
ensure!(peer_uid != 0, "--desk-peer-uid must not be 0");
|
||||
Some(DeskOptions { socket, peer_uid })
|
||||
}
|
||||
_ => anyhow::bail!("--desk-socket and --desk-peer-uid are given together or not at all"),
|
||||
},
|
||||
})
|
||||
}
|
||||
|
||||
|
|
@ -177,18 +199,14 @@ pub fn parse_options(args: impl Iterator<Item = String>) -> Result<Options> {
|
|||
/// of the refusal and nothing reaches the vector store for a process that
|
||||
/// never listens. The worker's first act is to reconcile the collection
|
||||
/// against every indexable document the mind holds now.
|
||||
pub fn startup<E, V>(
|
||||
options: &Options,
|
||||
embedder: E,
|
||||
index: V,
|
||||
backoff: Backoff,
|
||||
) -> Result<(Daemon<OwnedRedbMessagePackBackingStore, WorkerSink>, CultNetRudpServerHub, CultNetSchemaRegistry)>
|
||||
pub fn startup<E, V>(options: &Options, embedder: E, index: V, backoff: Backoff) -> Result<Started>
|
||||
where
|
||||
E: Embedder + Send + 'static,
|
||||
V: VectorIndex + crate::sources::SourceIndex + Send + 'static,
|
||||
{
|
||||
let mind = Mind::open(&options.state_root, &options.instance)?;
|
||||
let hub = bind(options.bind, &runtime_id(&options.instance))?;
|
||||
let desk = options.desk.as_ref().map(|desk| bind_desk(&desk.socket, desk.peer_uid)).transpose()?;
|
||||
let startup_entries = mind.index_entries(None)?;
|
||||
let sink = WorkerSink::spawn(embedder, index, &options.instance, mind.genesis_receipt_id()?, startup_entries, backoff);
|
||||
let daemon = Daemon::new(mind, sink);
|
||||
|
|
@ -196,7 +214,16 @@ where
|
|||
Some(url) => daemon.with_forge(crate::sources::forge::HttpForge::new(url)),
|
||||
None => daemon,
|
||||
};
|
||||
Ok((daemon, hub, schema_registry()?))
|
||||
Ok(Started { daemon, hub, registry: schema_registry()?, desk })
|
||||
}
|
||||
|
||||
/// What `startup` hands the loop: the daemon over the opened mind, the bound RUDP hub, the
|
||||
/// catalog the RUDP door advertises, and the desk door when one was asked for.
|
||||
pub struct Started {
|
||||
pub daemon: Daemon<OwnedRedbMessagePackBackingStore, WorkerSink>,
|
||||
pub hub: CultNetRudpServerHub,
|
||||
pub registry: CultNetSchemaRegistry,
|
||||
pub desk: Option<DeskDoor>,
|
||||
}
|
||||
|
||||
/// The bytes one reliable packet carries, and the packets one session may hold
|
||||
|
|
@ -255,7 +282,7 @@ pub const MAX_ECHOED_FIELD_BYTES: usize = 256;
|
|||
|
||||
/// The first field of a request that a reply would echo and that is over
|
||||
/// `MAX_ECHOED_FIELD_BYTES`.
|
||||
fn overlong_echo(message: &CultNetMessage) -> Option<&'static str> {
|
||||
pub(crate) fn overlong_echo(message: &CultNetMessage) -> Option<&'static str> {
|
||||
let echoed: Vec<(&'static str, &str)> = match message {
|
||||
CultNetMessage::OperationRequest { message_id, service_id, operation, payload_schema, .. } => {
|
||||
vec![
|
||||
|
|
@ -479,6 +506,7 @@ pub fn run<S: MindStore, I: IndexSink<S>>(
|
|||
daemon: &mut Daemon<S, I>,
|
||||
hub: &mut CultNetRudpServerHub,
|
||||
registry: &CultNetSchemaRegistry,
|
||||
mut desk: Option<&mut DeskDoor>,
|
||||
stopping: &AtomicBool,
|
||||
options: &ServeOptions,
|
||||
) -> Result<()> {
|
||||
|
|
@ -523,6 +551,7 @@ pub fn run<S: MindStore, I: IndexSink<S>>(
|
|||
}
|
||||
served += 1;
|
||||
}
|
||||
let desk_served = desk.as_deref_mut().map_or(0, |door| door.serve_ready(daemon, Utc::now()));
|
||||
let finished = collect_searches(daemon, &waiting);
|
||||
let answered = !finished.is_empty();
|
||||
for (ticket, found) in finished {
|
||||
|
|
@ -531,7 +560,7 @@ pub fn run<S: MindStore, I: IndexSink<S>>(
|
|||
let reply = respond(&response, &asked.message_id, asked.operation, &daemon.runtime_id(), &mut bodies, Utc::now());
|
||||
send(hub, &asked.session, &reply);
|
||||
}
|
||||
if served == 0 && !answered {
|
||||
if served == 0 && desk_served == 0 && !answered {
|
||||
std::thread::sleep(options.idle_sleep);
|
||||
}
|
||||
}
|
||||
|
|
@ -756,7 +785,7 @@ mod tests {
|
|||
|
||||
let loop_stopping = Arc::clone(&stopping);
|
||||
let serving = std::thread::spawn(move || {
|
||||
let result = run(&mut daemon, &mut hub, ®istry, &loop_stopping, &ServeOptions::default());
|
||||
let result = run(&mut daemon, &mut hub, ®istry, None, &loop_stopping, &ServeOptions::default());
|
||||
(result, daemon)
|
||||
});
|
||||
|
||||
|
|
@ -863,7 +892,7 @@ mod tests {
|
|||
let stopping = Arc::new(AtomicBool::new(false));
|
||||
let loop_stopping = Arc::clone(&stopping);
|
||||
let serving = std::thread::spawn(move || {
|
||||
let result = run(&mut daemon, &mut hub, ®istry, &loop_stopping, &ServeOptions::default());
|
||||
let result = run(&mut daemon, &mut hub, ®istry, None, &loop_stopping, &ServeOptions::default());
|
||||
(result, daemon)
|
||||
});
|
||||
|
||||
|
|
@ -1104,7 +1133,7 @@ mod tests {
|
|||
let stopping = Arc::new(AtomicBool::new(false));
|
||||
let loop_stopping = Arc::clone(&stopping);
|
||||
let serving = std::thread::spawn(move || {
|
||||
let result = run(&mut daemon, &mut hub, ®istry, &loop_stopping, &ServeOptions::default());
|
||||
let result = run(&mut daemon, &mut hub, ®istry, None, &loop_stopping, &ServeOptions::default());
|
||||
(result, daemon)
|
||||
});
|
||||
|
||||
|
|
@ -1541,6 +1570,7 @@ mod tests {
|
|||
reserve_ollama_urls: Vec::new(),
|
||||
embedding_model: "a-model".into(),
|
||||
forge_url: None,
|
||||
desk: None,
|
||||
};
|
||||
let started = |options: &Options| {
|
||||
let (embedder, index) = crate::index::fakes::pair();
|
||||
|
|
@ -1556,8 +1586,8 @@ mod tests {
|
|||
let error = format!("{:#}", started(&options).err().expect("a held port refuses"));
|
||||
assert!(error.contains("binding"), "{error}");
|
||||
drop(occupied);
|
||||
let (_daemon, hub, _registry) = started(&options).expect("both gates open");
|
||||
assert_eq!(hub.local_addr().unwrap(), port);
|
||||
let started = started(&options).expect("both gates open");
|
||||
assert_eq!(started.hub.local_addr().unwrap(), port);
|
||||
}
|
||||
|
||||
#[test]
|
||||
|
|
@ -1579,6 +1609,7 @@ mod tests {
|
|||
reserve_ollama_urls: Vec::new(),
|
||||
embedding_model: "qwen3-embedding:0.6b".into(),
|
||||
forge_url: None,
|
||||
desk: None,
|
||||
}
|
||||
);
|
||||
// Each of the six is required: there is no index-less mode.
|
||||
|
|
|
|||
527
crates/huginn-daemon/tests/desk_door.rs
Normal file
527
crates/huginn-daemon/tests/desk_door.rs
Normal file
|
|
@ -0,0 +1,527 @@
|
|||
//! The desk door through the built binary: a real `huginn-daemon` process, its two sockets and its
|
||||
//! mind on disk. What the unit and pipeline smokes cannot show: the flags reach the door, the
|
||||
//! door answers beside the RUDP door in one process, a stopped daemon takes its socket with it,
|
||||
//! a leftover socket is replaced, a retry after a restart replays the stored answer, and a peer
|
||||
//! that is not the configured uid is refused (logged by name) with the mind untouched.
|
||||
//!
|
||||
//! Needs a non-root uid: the door refuses uid 0.
|
||||
|
||||
use std::io::{Read, Write};
|
||||
use std::net::UdpSocket;
|
||||
use std::os::unix::fs::MetadataExt;
|
||||
use std::os::unix::net::UnixStream;
|
||||
use std::path::{Path, PathBuf};
|
||||
use std::process::{Child, Command, Stdio};
|
||||
use std::time::{Duration, Instant};
|
||||
|
||||
use chrono::{TimeZone, Utc};
|
||||
use cultnet_rs::{
|
||||
CULTNET_OPERATION_CONNECTION_ID, CultMesh, CultMeshRudpSocketOptions, CultNetMessage, CultNetWireContract,
|
||||
decode_cultnet_message_from_slice, encode_cultnet_message_to_vec, encode_frame,
|
||||
};
|
||||
use huginn_mind::desk_wire::{DeskRequest, DeskResponse, ForkAnswer, ForkAnswerCode, ForkAnswerRequest, RequestSchema};
|
||||
use huginn_mind::envelope::{decode_desk_response, decode_response, encode_desk_request, encode_request};
|
||||
use huginn_mind::eureka_pipeline::{
|
||||
Date, DeskAuthority, DocRef, OrgRepo, PipelineCampaign, PipelineDocument, PipelineInstance, PipelineQuestion, PipelineStewardship,
|
||||
QuestionOption, Sha, Short, Title,
|
||||
};
|
||||
use huginn_mind::wire::{HuginnMindRequest, HuginnMindResponse};
|
||||
use huginn_mind::{Faculty, Mind, PipelineAdmissionBatch, PipelineAdmissionOutcome, PipelineProvenance, options_hash};
|
||||
|
||||
const INSTANCE: &str = "yggdrasil";
|
||||
const FORK: &str = "eureka-state:question:Q1";
|
||||
const R1: &str = "00000000-0000-4000-8000-000000000001";
|
||||
const R2: &str = "00000000-0000-4000-8000-000000000002";
|
||||
|
||||
/// These tests run one at a time: each forks daemons and times them, and a fork made while another
|
||||
/// test holds a mind open lends that child the mind's lock until it execs.
|
||||
fn serial() -> std::sync::MutexGuard<'static, ()> {
|
||||
static LOCK: std::sync::Mutex<()> = std::sync::Mutex::new(());
|
||||
LOCK.lock().unwrap_or_else(|poisoned| poisoned.into_inner())
|
||||
}
|
||||
|
||||
fn my_uid() -> u32 {
|
||||
let dir = tempfile::tempdir().unwrap();
|
||||
std::fs::metadata(dir.path()).unwrap().uid()
|
||||
}
|
||||
|
||||
fn short(value: &str) -> Short {
|
||||
Short(value.into())
|
||||
}
|
||||
|
||||
/// A mind holding the campaign and one answerable question; returns the question's `options_hash`.
|
||||
fn seed(root: &Path) -> String {
|
||||
let mut mind = Mind::open(root, &INSTANCE.into()).unwrap();
|
||||
let question = PipelineQuestion {
|
||||
campaign: "eureka-state".into(),
|
||||
label: "Q1".into(),
|
||||
title: Title("Who owns the state?".into()),
|
||||
question: "Who owns the state?".into(),
|
||||
options: vec![
|
||||
QuestionOption { label: "A".into(), text: "the organ".into() },
|
||||
QuestionOption { label: "B".into(), text: "the client".into() },
|
||||
],
|
||||
recommended: "A".into(),
|
||||
depends: vec![],
|
||||
raised_in: None,
|
||||
asked_on: Date("2026-09-16".into()),
|
||||
};
|
||||
let hash = options_hash(&question).0;
|
||||
let repo = OrgRepo("GameCult/Epiphany".into());
|
||||
let documents = vec![
|
||||
PipelineDocument::Instance(PipelineInstance {
|
||||
instance: INSTANCE.into(),
|
||||
display_name: short("yggdrasil mind"),
|
||||
created_at: Date("2026-09-16".into()),
|
||||
host: short(INSTANCE),
|
||||
}),
|
||||
PipelineDocument::Stewardship(PipelineStewardship {
|
||||
instance: INSTANCE.into(),
|
||||
repo: repo.clone(),
|
||||
sequence: 1,
|
||||
assigned_on: Date("2026-09-16".into()),
|
||||
note: "assigned".into(),
|
||||
names: vec![],
|
||||
}),
|
||||
PipelineDocument::Campaign(PipelineCampaign {
|
||||
slug: "eureka-state".into(),
|
||||
title: Title("Eureka pipeline state".into()),
|
||||
repos: vec![repo],
|
||||
working_branch: short("codex/eureka-pipeline-state"),
|
||||
target_doc: DocRef { path: short("notes/target.md"), start_line: 1, end_line: 9, commit: Sha("5f98228d".into()) },
|
||||
}),
|
||||
PipelineDocument::Question(question),
|
||||
];
|
||||
let provenance = PipelineProvenance {
|
||||
faculty: Faculty::Hands,
|
||||
agent: short("claude"),
|
||||
session: short("session-1"),
|
||||
tool: short("admit"),
|
||||
imported_from: None,
|
||||
};
|
||||
let batch = PipelineAdmissionBatch { instance: INSTANCE.into(), provenance, documents };
|
||||
let outcome = mind.admit(batch, Utc.with_ymd_and_hms(2026, 9, 16, 12, 0, 0).unwrap());
|
||||
assert!(matches!(outcome, PipelineAdmissionOutcome::Committed { .. }), "{outcome:?}");
|
||||
hash
|
||||
}
|
||||
|
||||
/// A daemon process over `root`, its RUDP door on a free local port, its desk door at `desk`.
|
||||
struct Daemon {
|
||||
child: Child,
|
||||
rudp: String,
|
||||
log: PathBuf,
|
||||
}
|
||||
|
||||
fn free_port() -> u16 {
|
||||
UdpSocket::bind("127.0.0.1:0").unwrap().local_addr().unwrap().port()
|
||||
}
|
||||
|
||||
fn spawn(root: &Path, desk: &Path, peer_uid: u32, log: &Path) -> Daemon {
|
||||
let port = free_port();
|
||||
let child = Command::new(env!("CARGO_BIN_EXE_huginn-daemon"))
|
||||
.args(["--state-root", root.to_str().unwrap(), "--instance", INSTANCE, "--bind", &format!("127.0.0.1:{port}")])
|
||||
.args(["--qdrant-url", "http://127.0.0.1:1", "--ollama-url", "http://127.0.0.1:1", "--embedding-model", "a-model"])
|
||||
.args(["--desk-socket", desk.to_str().unwrap(), "--desk-peer-uid", &peer_uid.to_string()])
|
||||
.stdout(Stdio::null())
|
||||
.stderr(std::fs::OpenOptions::new().create(true).append(true).open(log).unwrap())
|
||||
.spawn()
|
||||
.unwrap();
|
||||
let mut daemon = Daemon { child, rudp: format!("rudp://127.0.0.1:{port}"), log: log.to_path_buf() };
|
||||
let started = Instant::now();
|
||||
// Listening, not merely present: a killed predecessor leaves a socket nobody answers on.
|
||||
while UnixStream::connect(desk).is_err() {
|
||||
assert!(daemon.child.try_wait().unwrap().is_none(), "the daemon exited early:\n{}", daemon.log_text());
|
||||
assert!(started.elapsed() < Duration::from_secs(30), "the desk socket never appeared:\n{}", daemon.log_text());
|
||||
std::thread::sleep(Duration::from_millis(20));
|
||||
}
|
||||
daemon
|
||||
}
|
||||
|
||||
impl Daemon {
|
||||
fn log_text(&self) -> String {
|
||||
std::fs::read_to_string(&self.log).unwrap_or_default()
|
||||
}
|
||||
|
||||
fn signal(&mut self, name: &str) {
|
||||
let status = Command::new("kill").args([&format!("-{name}"), &self.child.id().to_string()]).status().unwrap();
|
||||
assert!(status.success());
|
||||
let started = Instant::now();
|
||||
while self.child.try_wait().unwrap().is_none() {
|
||||
assert!(started.elapsed() < Duration::from_secs(20), "the daemon did not stop on {name}");
|
||||
std::thread::sleep(Duration::from_millis(20));
|
||||
}
|
||||
}
|
||||
|
||||
fn documents(&self) -> u32 {
|
||||
let mut client = CultMesh::create_rudp_client_for_endpoint(
|
||||
"desk-e2e".to_string(),
|
||||
CULTNET_OPERATION_CONNECTION_ID,
|
||||
&self.rudp,
|
||||
CultMeshRudpSocketOptions::default(),
|
||||
)
|
||||
.unwrap();
|
||||
client.connect(Vec::new()).unwrap();
|
||||
for _ in 0..2000 {
|
||||
let _ = client.receive_once();
|
||||
if client.connected() {
|
||||
break;
|
||||
}
|
||||
std::thread::sleep(Duration::from_millis(2));
|
||||
}
|
||||
assert!(client.connected(), "the RUDP door never accepted:\n{}", self.log_text());
|
||||
client.send_schema_message(&encode_request("w-1", &HuginnMindRequest::Whoami, None).unwrap()).unwrap();
|
||||
for _ in 0..2000 {
|
||||
client.poll_resends().unwrap();
|
||||
if let Some(reply) = client.receive_schema_message_once().unwrap() {
|
||||
return match decode_response(&reply).unwrap().1.unwrap() {
|
||||
HuginnMindResponse::Whoami(status) => status.documents,
|
||||
other => panic!("{other:?}"),
|
||||
};
|
||||
}
|
||||
std::thread::sleep(Duration::from_millis(2));
|
||||
}
|
||||
panic!("whoami was not answered");
|
||||
}
|
||||
}
|
||||
|
||||
impl Drop for Daemon {
|
||||
fn drop(&mut self) {
|
||||
let _ = self.child.kill();
|
||||
let _ = self.child.wait();
|
||||
}
|
||||
}
|
||||
|
||||
fn answer(who: &str, hash: &str, request_id: &str) -> DeskRequest {
|
||||
DeskRequest::Answer {
|
||||
mind: INSTANCE.into(),
|
||||
subject: short(who),
|
||||
authority: DeskAuthority::Operator,
|
||||
deciders: vec![short(who)],
|
||||
request: ForkAnswerRequest {
|
||||
schema_version: RequestSchema::V0,
|
||||
fork_id: FORK.into(),
|
||||
options_hash: hash.into(),
|
||||
answer: Some(ForkAnswer { choice: Some("A".into()), note: None }),
|
||||
retracted: false,
|
||||
retracts: None,
|
||||
answered_at: "2026-09-16T11:00:00Z".into(),
|
||||
request_id: request_id.into(),
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
fn framed(message: &CultNetMessage) -> Vec<u8> {
|
||||
encode_frame(&encode_cultnet_message_to_vec(message, CultNetWireContract::CultNetSchemaV0).unwrap()).unwrap()
|
||||
}
|
||||
|
||||
fn desk_bytes(id: &str, request: &DeskRequest) -> Vec<u8> {
|
||||
framed(&encode_desk_request(id, request, None).unwrap())
|
||||
}
|
||||
|
||||
fn read_reply(stream: &mut impl Read) -> Option<CultNetMessage> {
|
||||
let mut header = [0_u8; 4];
|
||||
stream.read_exact(&mut header).ok()?;
|
||||
let mut payload = vec![0_u8; u32::from_be_bytes(header) as usize];
|
||||
stream.read_exact(&mut payload).ok()?;
|
||||
Some(decode_cultnet_message_from_slice(&payload, CultNetWireContract::CultNetSchemaV0).unwrap())
|
||||
}
|
||||
|
||||
fn exchange(path: &Path, bytes: &[u8]) -> Option<CultNetMessage> {
|
||||
let mut stream = UnixStream::connect(path).unwrap();
|
||||
stream.set_read_timeout(Some(Duration::from_secs(10))).unwrap();
|
||||
let _ = stream.write_all(bytes);
|
||||
read_reply(&mut stream)
|
||||
}
|
||||
|
||||
fn ask(path: &Path, id: &str, request: &DeskRequest) -> DeskResponse {
|
||||
let reply = exchange(path, &desk_bytes(id, request)).expect("the door replied");
|
||||
decode_desk_response(&reply).unwrap().1.unwrap_or_else(|failure| panic!("{failure:?}"))
|
||||
}
|
||||
|
||||
fn counted_id(response: DeskResponse) -> String {
|
||||
match response {
|
||||
DeskResponse::Answered { answer, .. } => {
|
||||
assert_eq!(answer.outcome, "Counted", "{answer:?}");
|
||||
answer.answer_id
|
||||
}
|
||||
other => panic!("{other:?}"),
|
||||
}
|
||||
}
|
||||
|
||||
fn catalog_request() -> CultNetMessage {
|
||||
CultNetMessage::SchemaCatalogRequest { message_id: "c-1".into(), include_schema_json: Some(false), schema_ids: None, kinds: None }
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn the_binary_serves_both_doors_and_survives_restart() {
|
||||
let _serial = serial();
|
||||
assert_ne!(my_uid(), 0, "the door refuses uid 0, so this test cannot run as root");
|
||||
let (root, run) = (tempfile::tempdir().unwrap(), tempfile::tempdir().unwrap());
|
||||
let hash = seed(root.path());
|
||||
let (desk, log) = (run.path().join("desk.sock"), run.path().join("daemon.log"));
|
||||
|
||||
let mut daemon = spawn(root.path(), &desk, my_uid(), &log);
|
||||
let seeded = daemon.documents();
|
||||
// The catalog is the two desk schemas.
|
||||
let catalog = format!("{:?}", exchange(&desk, &framed(&catalog_request())).expect("a catalog"));
|
||||
assert!(catalog.contains("huginn.desk_request.v1") && catalog.contains("huginn.desk_response.v1"), "{catalog}");
|
||||
assert!(!catalog.contains("huginn.mind_request.v1"), "{catalog}");
|
||||
// Reads are not built: Unavailable, nothing stored.
|
||||
let forks = DeskRequest::Forks { mind: INSTANCE.into(), subject: short("x"), deciders: vec![short("x")] };
|
||||
let fork = DeskRequest::Fork { mind: INSTANCE.into(), subject: short("x"), deciders: vec![short("x")], fork_id: FORK.into() };
|
||||
for request in [forks, fork] {
|
||||
let DeskResponse::Refused(huginn_mind::MindRefusal::Unavailable { detail }) = ask(&desk, "m-r", &request) else {
|
||||
panic!("a read is refused unavailable")
|
||||
};
|
||||
assert!(detail.contains("desk reads not built"), "{detail}");
|
||||
}
|
||||
assert_eq!(daemon.documents(), seeded);
|
||||
// An answer, then a stale one.
|
||||
let first = answer("x", &hash, R1);
|
||||
assert_eq!(counted_id(ask(&desk, "m-1", &first)), "n1");
|
||||
let DeskResponse::Answered { answer: stale, counted_answer_id, .. } = ask(&desk, "m-2", &answer("x", &"0".repeat(64), R2)) else {
|
||||
panic!("a stale answer is stored")
|
||||
};
|
||||
assert_eq!((stale.code, counted_answer_id.as_deref()), (Some(ForkAnswerCode::StaleFork), Some("n1")));
|
||||
let stored = daemon.documents();
|
||||
assert_eq!(stored, seeded + 2);
|
||||
|
||||
// A stop takes the socket with it; the stored answers are the retry's memory.
|
||||
daemon.signal("TERM");
|
||||
assert!(!desk.exists(), "a stopped daemon unlinks its socket");
|
||||
let mut daemon = spawn(root.path(), &desk, my_uid(), &log);
|
||||
assert_eq!(daemon.documents(), stored);
|
||||
assert_eq!(counted_id(ask(&desk, "m-1", &first)), "n1", "the same request id returns the stored answer");
|
||||
assert_eq!(daemon.documents(), stored, "the replay stored nothing");
|
||||
|
||||
// A killed daemon leaves its socket; the next one replaces it.
|
||||
daemon.signal("KILL");
|
||||
assert!(desk.exists(), "a killed daemon leaves its socket behind");
|
||||
let daemon = spawn(root.path(), &desk, my_uid(), &log);
|
||||
assert_eq!(counted_id(ask(&desk, "m-1", &first)), "n1");
|
||||
assert_eq!(daemon.documents(), stored);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn the_binary_refuses_another_uid() {
|
||||
let _serial = serial();
|
||||
let mine = my_uid();
|
||||
assert_ne!(mine, 0);
|
||||
let (root, run) = (tempfile::tempdir().unwrap(), tempfile::tempdir().unwrap());
|
||||
let hash = seed(root.path());
|
||||
let (desk, log) = (run.path().join("desk.sock"), run.path().join("daemon.log"));
|
||||
let mut daemon = spawn(root.path(), &desk, mine + 1, &log);
|
||||
let seeded = daemon.documents();
|
||||
assert!(exchange(&desk, &desk_bytes("m-1", &answer("x", &hash, R1))).is_none(), "a refused peer reads EOF");
|
||||
assert!(exchange(&desk, &framed(&catalog_request())).is_none(), "not even the catalog");
|
||||
assert_eq!(daemon.documents(), seeded);
|
||||
daemon.signal("TERM");
|
||||
let wanted = format!("desk door refused a peer with uid {mine} (expects {})", mine + 1);
|
||||
assert!(daemon.log_text().contains(&wanted), "the log names both uids:\n{}", daemon.log_text());
|
||||
// Read-only reopen: the mind is as it was.
|
||||
let reopened = Mind::open(root.path(), &INSTANCE.into()).unwrap();
|
||||
assert_eq!(reopened.status(huginn_mind::wire::IndexStatus::Current).documents, seeded);
|
||||
}
|
||||
|
||||
/// Soul's probe, from the built binary: a client that is uid 0 inside its own user namespace is
|
||||
/// the real uid to the kernel, so the door sees the real uid and namespace root cannot pass as root.
|
||||
#[test]
|
||||
fn namespace_root_is_seen_as_the_real_uid() {
|
||||
let _serial = serial();
|
||||
let probe = Command::new("unshare").args(["-r", "true"]).status();
|
||||
if !probe.is_ok_and(|status| status.success()) || Command::new("socat").arg("-V").stdout(Stdio::null()).status().is_err() {
|
||||
eprintln!("skipped: needs `unshare -r` (user namespaces) and socat");
|
||||
return;
|
||||
}
|
||||
let mine = my_uid();
|
||||
assert_ne!(mine, 0);
|
||||
let (root, run) = (tempfile::tempdir().unwrap(), tempfile::tempdir().unwrap());
|
||||
seed(root.path());
|
||||
let (desk, log) = (run.path().join("desk.sock"), run.path().join("daemon.log"));
|
||||
let frame = run.path().join("catalog.frame");
|
||||
std::fs::write(&frame, framed(&catalog_request())).unwrap();
|
||||
let through_namespace = || {
|
||||
Command::new("unshare")
|
||||
.args(["-r", "socat", "-t", "2", "-", &format!("UNIX-CONNECT:{}", desk.display())])
|
||||
.stdin(std::fs::File::open(&frame).unwrap())
|
||||
.output()
|
||||
.unwrap()
|
||||
.stdout
|
||||
};
|
||||
|
||||
let mut allowed = spawn(root.path(), &desk, mine, &log);
|
||||
let reply = through_namespace();
|
||||
assert!(!reply.is_empty(), "the real uid is the configured one, so namespace root is answered:\n{}", allowed.log_text());
|
||||
allowed.signal("TERM");
|
||||
|
||||
let mut other = spawn(root.path(), &desk, mine + 1, &log);
|
||||
assert!(through_namespace().is_empty(), "a door configured for another uid answers nobody");
|
||||
other.signal("TERM");
|
||||
let text = other.log_text();
|
||||
assert!(text.contains(&format!("desk door refused a peer with uid {mine} (expects {})", mine + 1)), "{text}");
|
||||
assert!(!text.contains("refused a peer with uid 0"), "the door never sees namespace root as 0:\n{text}");
|
||||
}
|
||||
|
||||
// ---- the loop's bounds, from the built binary and from run() in process ----
|
||||
|
||||
fn setup() -> (tempfile::TempDir, tempfile::TempDir, String, PathBuf, PathBuf) {
|
||||
let (root, run) = (tempfile::tempdir().unwrap(), tempfile::tempdir().unwrap());
|
||||
let hash = seed(root.path());
|
||||
let (desk, log) = (run.path().join("desk.sock"), run.path().join("daemon.log"));
|
||||
(root, run, hash, desk, log)
|
||||
}
|
||||
|
||||
/// A peer that sends a valid frame one byte per 300 ms: inside any per-read timeout, so only a
|
||||
/// per-connection deadline cuts it off. Returns when the door has closed on it, with the bytes sent.
|
||||
fn drip(desk: &Path, hash: &str) -> std::thread::JoinHandle<(usize, Duration)> {
|
||||
let (desk, bytes) = (desk.to_path_buf(), desk_bytes("m-1", &answer("x", hash, R1)));
|
||||
std::thread::spawn(move || {
|
||||
let mut stream = UnixStream::connect(&desk).unwrap();
|
||||
let started = Instant::now();
|
||||
let mut sent = 0;
|
||||
for byte in bytes.iter().take(60) {
|
||||
if stream.write_all(&[*byte]).is_err() {
|
||||
break;
|
||||
}
|
||||
sent += 1;
|
||||
std::thread::sleep(Duration::from_millis(300));
|
||||
}
|
||||
(sent, started.elapsed())
|
||||
})
|
||||
}
|
||||
|
||||
/// The bound the door states for one connection is one second, so a dripping peer neither delays
|
||||
/// another door's answer nor a stop beyond it (the sum is checked with the margin a loaded host needs).
|
||||
#[test]
|
||||
fn a_dripping_peer_does_not_delay_the_rudp_door() {
|
||||
let _serial = serial();
|
||||
let (root, _run, hash, desk, log) = setup();
|
||||
let daemon = spawn(root.path(), &desk, my_uid(), &log);
|
||||
let dripper = drip(&desk, &hash);
|
||||
std::thread::sleep(Duration::from_millis(400));
|
||||
let asked = Instant::now();
|
||||
assert!(daemon.documents() > 0);
|
||||
assert!(asked.elapsed() < Duration::from_millis(2500), "the RUDP whoami waited {:?} behind a drip", asked.elapsed());
|
||||
let (sent, lasted) = dripper.join().unwrap();
|
||||
assert!(lasted < Duration::from_millis(1800), "the drip was held for {lasted:?} ({sent} bytes), not cut at one second");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_dripping_peer_does_not_delay_a_stop() {
|
||||
let _serial = serial();
|
||||
let (root, _run, hash, desk, log) = setup();
|
||||
let mut daemon = spawn(root.path(), &desk, my_uid(), &log);
|
||||
let dripper = drip(&desk, &hash);
|
||||
std::thread::sleep(Duration::from_millis(300));
|
||||
let signalled = Instant::now();
|
||||
let status = Command::new("kill").args(["-TERM", &daemon.child.id().to_string()]).status().unwrap();
|
||||
assert!(status.success());
|
||||
while daemon.child.try_wait().unwrap().is_none() {
|
||||
assert!(signalled.elapsed() < Duration::from_millis(2500), "the daemon was still held by a drip {:?} after SIGTERM", signalled.elapsed());
|
||||
std::thread::sleep(Duration::from_millis(20));
|
||||
}
|
||||
dripper.join().unwrap();
|
||||
}
|
||||
|
||||
fn ticks(pid: u32) -> u64 {
|
||||
let stat = std::fs::read_to_string(format!("/proc/{pid}/stat")).unwrap();
|
||||
let rest = &stat[stat.rfind(')').unwrap() + 2..];
|
||||
let fields: Vec<&str> = rest.split(' ').collect();
|
||||
fields[11].parse::<u64>().unwrap() + fields[12].parse::<u64>().unwrap()
|
||||
}
|
||||
|
||||
/// An idle daemon sleeps between passes: a loop that spins when nothing waits burns the core.
|
||||
#[test]
|
||||
fn an_idle_daemon_does_not_spin() {
|
||||
let _serial = serial();
|
||||
let (root, _run, _hash, desk, log) = setup();
|
||||
let mut daemon = spawn(root.path(), &desk, my_uid(), &log);
|
||||
std::thread::sleep(Duration::from_millis(1500));
|
||||
let before = ticks(daemon.child.id());
|
||||
std::thread::sleep(Duration::from_secs(2));
|
||||
let burned = ticks(daemon.child.id()) - before;
|
||||
// 100 ticks a second: a spinning loop burns about 200 in two seconds, an idle one a handful.
|
||||
assert!(burned < 30, "an idle daemon burned {burned} ticks in 2 s");
|
||||
daemon.signal("TERM");
|
||||
}
|
||||
|
||||
/// WouldBlock on an idle listener is not an error and is not logged; a real accept error is.
|
||||
#[test]
|
||||
fn an_idle_door_logs_no_accept_error_and_a_real_one_is_logged() {
|
||||
let _serial = serial();
|
||||
let (root, _run, _hash, desk, log) = setup();
|
||||
let mut daemon = spawn(root.path(), &desk, my_uid(), &log);
|
||||
std::thread::sleep(Duration::from_millis(500));
|
||||
assert!(!daemon.log_text().contains("could not accept"), "an idle door logged:\n{}", daemon.log_text());
|
||||
// An accept error: the daemon's soft fd limit is 3, stdin, stdout and stderr, so no
|
||||
// descriptor number is free whatever else it holds open; then a peer connects.
|
||||
let pid = daemon.child.id();
|
||||
let limit = |soft: usize| {
|
||||
let status = Command::new("prlimit").args([&format!("--pid={pid}"), &format!("--nofile={soft}:4096")]).status();
|
||||
status.is_ok_and(|status| status.success())
|
||||
};
|
||||
if !limit(3) {
|
||||
eprintln!("skipped: needs prlimit");
|
||||
return;
|
||||
}
|
||||
let _peer = UnixStream::connect(&desk).unwrap();
|
||||
let started = Instant::now();
|
||||
while !daemon.log_text().contains("could not accept") && started.elapsed() < Duration::from_secs(5) {
|
||||
std::thread::sleep(Duration::from_millis(20));
|
||||
}
|
||||
limit(1024);
|
||||
assert!(daemon.log_text().contains("could not accept"), "an accept error went unlogged:\n{}", daemon.log_text());
|
||||
daemon.signal("TERM");
|
||||
}
|
||||
|
||||
/// A pass that served desk connections is followed at once by the next pass, not by the idle
|
||||
/// sleep: eight queued connections are answered in two passes of four, with no sleep between.
|
||||
#[test]
|
||||
fn a_busy_pass_is_not_followed_by_the_idle_sleep() {
|
||||
let _serial = serial();
|
||||
use huginn_daemon::index::Backoff;
|
||||
use huginn_daemon::index::ollama::pool;
|
||||
use huginn_daemon::index::qdrant::QdrantIndex;
|
||||
use huginn_daemon::serve::{ServeOptions, Started, parse_options, run, startup};
|
||||
use std::sync::Arc;
|
||||
use std::sync::atomic::{AtomicBool, Ordering};
|
||||
|
||||
let (root, _run, _hash, desk, _log) = setup();
|
||||
let args = [
|
||||
"--state-root", root.path().to_str().unwrap(), "--instance", INSTANCE, "--bind", &format!("127.0.0.1:{}", free_port()),
|
||||
"--qdrant-url", "http://127.0.0.1:1", "--ollama-url", "http://127.0.0.1:1", "--embedding-model", "a-model",
|
||||
"--desk-socket", desk.to_str().unwrap(), "--desk-peer-uid", &my_uid().to_string(),
|
||||
];
|
||||
let options = parse_options(args.iter().map(|arg| arg.to_string())).unwrap();
|
||||
let embedder = pool(&options.ollama_urls, &options.reserve_ollama_urls, &options.embedding_model);
|
||||
let Started { mut daemon, mut hub, registry, desk: mut door } =
|
||||
startup(&options, embedder, QdrantIndex::new(&options.qdrant_url), Backoff::default()).unwrap();
|
||||
let stopping = Arc::new(AtomicBool::new(false));
|
||||
let loop_stopping = Arc::clone(&stopping);
|
||||
let serving = std::thread::spawn(move || {
|
||||
let serve = ServeOptions { idle_sleep: Duration::from_millis(600), ..ServeOptions::default() };
|
||||
run(&mut daemon, &mut hub, ®istry, door.as_mut(), &loop_stopping, &serve)
|
||||
});
|
||||
std::thread::sleep(Duration::from_millis(100));
|
||||
let frame = framed(&catalog_request());
|
||||
let mut peers: Vec<UnixStream> = (0..8)
|
||||
.map(|_| {
|
||||
let mut peer = UnixStream::connect(&desk).unwrap();
|
||||
peer.write_all(&frame).unwrap();
|
||||
peer.set_read_timeout(Some(Duration::from_secs(10))).unwrap();
|
||||
peer
|
||||
})
|
||||
.collect();
|
||||
let started = Instant::now();
|
||||
let answered: Vec<Duration> = peers
|
||||
.iter_mut()
|
||||
.map(|peer| {
|
||||
assert!(read_reply(peer).is_some());
|
||||
started.elapsed()
|
||||
})
|
||||
.collect();
|
||||
stopping.store(true, Ordering::SeqCst);
|
||||
serving.join().unwrap().unwrap();
|
||||
let gap = answered[4] - answered[3];
|
||||
assert!(gap < Duration::from_millis(300), "the second batch waited {gap:?}: the loop slept after a busy pass ({answered:?})");
|
||||
}
|
||||
Loading…
Reference in a new issue