diff --git a/Cargo.lock b/Cargo.lock index 9efbed5..6b7bea5 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2,6 +2,41 @@ # It is not intended for manual editing. version = 4 +[[package]] +name = "aead" +version = "0.5.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d122413f284cf2d62fb1b7db97e02edb8cda96d769b16e443a4f6195e35662b0" +dependencies = [ + "crypto-common", + "generic-array", +] + +[[package]] +name = "aes" +version = "0.8.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b169f7a6d4742236a0a00c541b845991d0ac43e546831af1249753ab4c3aa3a0" +dependencies = [ + "cfg-if", + "cipher", + "cpufeatures", +] + +[[package]] +name = "aes-gcm" +version = "0.10.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "831010a0f742e1209b3bcea8fab6a8e149051ba6099432c8cb2cc117dec3ead1" +dependencies = [ + "aead", + "aes", + "cipher", + "ctr", + "ghash", + "subtle", +] + [[package]] name = "android_system_properties" version = "0.1.6" @@ -23,6 +58,18 @@ version = "1.5.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f2032f911046de80f0a198e0901378627c33f59ea0ac00e363d481118bd70a53" +[[package]] +name = "base64" +version = "0.22.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "72b3254f16251a8381aa12e40e3c4d2f0199f8c6508fbecb9d91f575e0fbb8c6" + +[[package]] +name = "base64ct" +version = "1.8.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2af50177e190e07a26ab74f8b1efbfe2ef87da2116221318cb1c2e82baf7de06" + [[package]] name = "bitflags" version = "2.13.2" @@ -74,6 +121,22 @@ dependencies = [ "windows-link", ] +[[package]] +name = "cipher" +version = "0.4.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "773f3b9af64447d2ce9850330c473515014aa235e6a783b02db81ff39e4a3dad" +dependencies = [ + "crypto-common", + "inout", +] + +[[package]] +name = "const-oid" +version = "0.9.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c2459377285ad874054d797f3ccebf984978aa39129f6eafde5cdc8315b612f8" + [[package]] name = "core-foundation-sys" version = "0.8.7" @@ -96,9 +159,19 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "78c8292055d1c1df0cce5d180393dc8cce0abec0a7102adb6c7b1eef6016d60a" dependencies = [ "generic-array", + "rand_core 0.6.4", "typenum", ] +[[package]] +name = "ctr" +version = "0.9.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0369ee1ad671834580515889b80f2ea915f23b8be8d0daa4bbaf2ac5c7590835" +dependencies = [ + "cipher", +] + [[package]] name = "cultcache-rs" version = "0.2.0" @@ -127,6 +200,68 @@ dependencies = [ "syn 2.0.119", ] +[[package]] +name = "cultnet-rs" +version = "0.1.0" +source = "git+https://github.com/GameCult/CultLib.git?rev=a0813c6eed24d30bf88073ef615b633c77ebfcd6#a0813c6eed24d30bf88073ef615b633c77ebfcd6" +dependencies = [ + "aes-gcm", + "anyhow", + "base64", + "chrono", + "cultcache-rs", + "ed25519-dalek", + "hmac", + "rand", + "rmp-serde", + "rmpv", + "serde", + "serde_bytes", + "serde_json", + "sha2", + "socket2", + "uuid", + "windows-sys 0.61.2", + "zeroize", +] + +[[package]] +name = "curve25519-dalek" +version = "4.1.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "97fb8b7c4503de7d6ae7b42ab72a5a59857b4c937ec27a3d4539dba95b5ab2be" +dependencies = [ + "cfg-if", + "cpufeatures", + "curve25519-dalek-derive", + "digest", + "fiat-crypto", + "rustc_version", + "subtle", + "zeroize", +] + +[[package]] +name = "curve25519-dalek-derive" +version = "0.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f46882e17999c6cc590af592290432be3bce0428cb0d5f8b6715e4dc7b383eb3" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.119", +] + +[[package]] +name = "der" +version = "0.7.10" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e7c1832837b905bbfb5101e07cc24c8deddf52f93225eee6ead5f4d63d53ddcb" +dependencies = [ + "const-oid", + "zeroize", +] + [[package]] name = "digest" version = "0.10.7" @@ -135,6 +270,7 @@ checksum = "9ed9a281f7bc9b7576e61468ba615a66a5c8cfdff42420a70aa82701a3b1e292" dependencies = [ "block-buffer", "crypto-common", + "subtle", ] [[package]] @@ -143,6 +279,30 @@ version = "1.0.20" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d0881ea181b1df73ff77ffaaf9c7544ecc11e82fba9b5f27b262a3c73a332555" +[[package]] +name = "ed25519" +version = "2.2.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "115531babc129696a58c64a4fef0a8bf9e9698629fb97e9e40767d235cfbcd53" +dependencies = [ + "pkcs8", + "signature", +] + +[[package]] +name = "ed25519-dalek" +version = "2.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "70e796c081cee67dc755e1a36a0a172b897fab85fc3f6bc48307991f64e4eca9" +dependencies = [ + "curve25519-dalek", + "ed25519", + "serde", + "sha2", + "subtle", + "zeroize", +] + [[package]] name = "epiphany-pipeline" version = "0.1.0" @@ -176,6 +336,12 @@ version = "2.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "da7c62ceae207dd37ea5b845da6a0696c799f85e97da1ab5b7910be3c1c80223" +[[package]] +name = "fiat-crypto" +version = "0.2.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "28dea519a9695b9977216879a3ebfddf92f1c08c05d984f8996aecd6ecdc811d" + [[package]] name = "find-msvc-tools" version = "0.1.12" @@ -226,6 +392,29 @@ dependencies = [ "version_check", ] +[[package]] +name = "getrandom" +version = "0.2.17" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ff2abc00be7fca6ebc474524697ae276ad847ad0a6b3faa4bcb027e9a4614ad0" +dependencies = [ + "cfg-if", + "libc", + "wasi", +] + +[[package]] +name = "getrandom" +version = "0.3.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "899def5c37c4fd7b2664648c28120ecec138e4d395b459e5ca34f9cce2dd77fd" +dependencies = [ + "cfg-if", + "libc", + "r-efi 5.3.0", + "wasip2", +] + [[package]] name = "getrandom" version = "0.4.3" @@ -234,12 +423,43 @@ checksum = "300e883d756b2e4ec94e02791f39b04b522276138852cfc41d9fb7e904106099" dependencies = [ "cfg-if", "libc", - "r-efi", + "r-efi 6.0.0", +] + +[[package]] +name = "ghash" +version = "0.5.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f0d8a4362ccb29cb0b265253fb0a2728f592895ee6854fd9bc13f2ffda266ff1" +dependencies = [ + "opaque-debug", + "polyval", +] + +[[package]] +name = "hmac" +version = "0.12.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6c49c37c09c17a53d937dfbb742eb3a961d65a994e6bcdcf37e7399d0cc8ab5e" +dependencies = [ + "digest", ] [[package]] name = "huginn-daemon" version = "0.1.0" +dependencies = [ + "anyhow", + "base64", + "chrono", + "cultnet-rs", + "huginn-mind", + "rmp-serde", + "schemars", + "serde", + "signal-hook", + "tempfile", +] [[package]] name = "huginn-mind" @@ -281,6 +501,15 @@ dependencies = [ "cc", ] +[[package]] +name = "inout" +version = "0.1.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "879f10e63c20629ecabbb64a8010319738c66a5cd0c29b02d63d272b03751d01" +dependencies = [ + "generic-array", +] + [[package]] name = "itoa" version = "1.0.18" @@ -337,12 +566,49 @@ version = "1.21.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9f7c3e4beb33f85d45ae3e3a1792185706c8e16d043238c593331cc7cd313b50" +[[package]] +name = "opaque-debug" +version = "0.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c08d65885ee38876c4f86fa503fb49d7b507c2b62552df7c70b2fce627e06381" + [[package]] name = "pin-project-lite" version = "0.2.17" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a89322df9ebe1c1578d689c92318e070967d1042b512afbe49518723f4e6d5cd" +[[package]] +name = "pkcs8" +version = "0.10.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f950b2377845cebe5cf8b5165cb3cc1a5e0fa5cfa3e1f7f55707d8fd82e0a7b7" +dependencies = [ + "der", + "spki", +] + +[[package]] +name = "polyval" +version = "0.6.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9d1fe60d06143b2430aa532c94cfe9e29783047f06c0d7fd359a9a51b729fa25" +dependencies = [ + "cfg-if", + "cpufeatures", + "opaque-debug", + "universal-hash", +] + +[[package]] +name = "ppv-lite86" +version = "0.2.21" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "85eae3c4ed2f50dcfe72643da4befc30deadb458a9b590d720cde2f2b1e97da9" +dependencies = [ + "zerocopy", +] + [[package]] name = "proc-macro2" version = "1.0.107" @@ -361,12 +627,56 @@ dependencies = [ "proc-macro2", ] +[[package]] +name = "r-efi" +version = "5.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "69cdb34c158ceb288df11e18b4bd39de994f6657d83847bdffdbd7f346754b0f" + [[package]] name = "r-efi" version = "6.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f8dcc9c7d52a811697d2151c701e0d08956f92b0e24136cf4cf27b57a6a0d9bf" +[[package]] +name = "rand" +version = "0.9.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b9ef1d0d795eb7d84685bca4f72f3649f064e6641543d3a8c415898726a57b41" +dependencies = [ + "rand_chacha", + "rand_core 0.9.5", +] + +[[package]] +name = "rand_chacha" +version = "0.9.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d3022b5f1df60f26e1ffddd6c66e8aa15de382ae63b3a0c1bfc0e4d3e3f325cb" +dependencies = [ + "ppv-lite86", + "rand_core 0.9.5", +] + +[[package]] +name = "rand_core" +version = "0.6.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ec0be4795e2f6a28069bec0b5ff3e2ac9bafc99e6a9a7dc3547996c5c816922c" +dependencies = [ + "getrandom 0.2.17", +] + +[[package]] +name = "rand_core" +version = "0.9.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "76afc826de14238e6e8c374ddcc1fa19e374fd8dd986b0d2af0d02377261d83c" +dependencies = [ + "getrandom 0.3.4", +] + [[package]] name = "redb" version = "4.3.0" @@ -415,6 +725,26 @@ dependencies = [ "serde", ] +[[package]] +name = "rmpv" +version = "1.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7a4e1d4b9b938a26d2996af33229f0ca0956c652c1375067f0b45291c1df8417" +dependencies = [ + "rmp", + "serde", + "serde_bytes", +] + +[[package]] +name = "rustc_version" +version = "0.4.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cfcb3a22ef46e85b45de6ee7e79d063319ebb6594faafcf1c225ea92ab6e9b92" +dependencies = [ + "semver", +] + [[package]] name = "rustix" version = "1.1.4" @@ -459,6 +789,12 @@ dependencies = [ "syn 3.0.5", ] +[[package]] +name = "semver" +version = "1.0.28" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8a7852d02fc848982e0c167ef163aaff9cd91dc640ba85e263cb1ce46fae51cd" + [[package]] name = "serde" version = "1.0.229" @@ -540,12 +876,67 @@ version = "2.0.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f8fadd59c855ef2080decdef8ff161eb6661b86933c9d82e5ba29dc602a55aba" +[[package]] +name = "signal-hook" +version = "0.3.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d881a16cf4426aa584979d30bd82cb33429027e42122b169753d6ef1085ed6e2" +dependencies = [ + "libc", + "signal-hook-registry", +] + +[[package]] +name = "signal-hook-registry" +version = "1.4.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c4db69cba1110affc0e9f7bcd48bbf87b3f4fc7c61fc9155afd4c469eb3d6c1b" +dependencies = [ + "errno", + "libc", +] + +[[package]] +name = "signature" +version = "2.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "77549399552de45a898a580c1b41d445bf730df867cc44e6c0233bbc4b8329de" +dependencies = [ + "rand_core 0.6.4", +] + [[package]] name = "slab" version = "0.4.12" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0c790de23124f9ab44544d7ac05d60440adc586479ce501c1d6d7da3cd8c9cf5" +[[package]] +name = "socket2" +version = "0.5.10" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e22376abed350d73dd1cd119b57ffccad95b4e585a7cda43e286245ce23c0678" +dependencies = [ + "libc", + "windows-sys 0.52.0", +] + +[[package]] +name = "spki" +version = "0.7.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d91ed6c858b01f942cd56b37a94b3e0a1798290327d1236e4d9cf4eaca44d29d" +dependencies = [ + "base64ct", + "der", +] + +[[package]] +name = "subtle" +version = "2.6.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "13c2bddecc57b384dee18652358fb23172facb8a2c51ccc10d74c157bdea3292" + [[package]] name = "syn" version = "2.0.119" @@ -575,7 +966,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "32497e9a4c7b38532efcdebeef879707aa9f794296a4f0244f6f69e9bc8574bd" dependencies = [ "fastrand", - "getrandom", + "getrandom 0.4.3", "once_cell", "rustix", "windows-sys 0.61.2", @@ -593,13 +984,23 @@ version = "1.0.24" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e6e4313cd5fcd3dad5cafa179702e2b244f760991f45397d14d4ebf38247da75" +[[package]] +name = "universal-hash" +version = "0.5.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fc1de2c688dc15305988b563c3854064043356019f97a4b46276fe734c4f07ea" +dependencies = [ + "crypto-common", + "subtle", +] + [[package]] name = "uuid" version = "1.26.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "2ef6dac1e96601b4fb3acccccff2139741fcb757cb9a36089bf5be91cfb285ce" dependencies = [ - "getrandom", + "getrandom 0.4.3", "js-sys", "wasm-bindgen", ] @@ -610,6 +1011,21 @@ version = "0.9.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0b928f33d975fc6ad9f86c8f283853ad26bdd5b10b7f1542aa2fa15e2289105a" +[[package]] +name = "wasi" +version = "0.11.1+wasi-snapshot-preview1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ccf3ec651a847eb01de73ccad15eb7d99f80485de043efb2f370cd654f4ea44b" + +[[package]] +name = "wasip2" +version = "1.0.4+wasi-0.2.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b67efb37e106e55ce722a510d6b5f9c17f083e5fc79afc2badeb12cc313d9487" +dependencies = [ + "wit-bindgen", +] + [[package]] name = "wasm-bindgen" version = "0.2.128" @@ -736,6 +1152,15 @@ dependencies = [ "windows-link", ] +[[package]] +name = "windows-sys" +version = "0.52.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "282be5f36a8ce781fad8c8ae18fa3f9beff57ec1b52cb3de0789201425d9a33d" +dependencies = [ + "windows-targets", +] + [[package]] name = "windows-sys" version = "0.59.0" @@ -818,6 +1243,38 @@ version = "0.52.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "589f6da84c646204747d1270a2a5661ea66ed1cced2631d546fdfb155959f9ec" +[[package]] +name = "wit-bindgen" +version = "0.57.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1ebf944e87a7c253233ad6766e082e3cd714b5d03812acc24c318f549614536e" + +[[package]] +name = "zerocopy" +version = "0.8.57" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d35102a9f36d089ccae9e4c6802bc118be4487b80aaffc0ab4e0cf5ce92d2873" +dependencies = [ + "zerocopy-derive", +] + +[[package]] +name = "zerocopy-derive" +version = "0.8.57" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "146c01f5ab44258da43cf276c74a2763db2ff3969c9c652c3f2de07041d0b2bc" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.119", +] + +[[package]] +name = "zeroize" +version = "1.9.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e13c156562582aa81c60cb29407084cdb54c4164760106ab78e6c5b0858cf64e" + [[package]] name = "zmij" version = "1.0.23" diff --git a/crates/huginn-daemon/Cargo.toml b/crates/huginn-daemon/Cargo.toml index d8071b2..6ce886d 100644 --- a/crates/huginn-daemon/Cargo.toml +++ b/crates/huginn-daemon/Cargo.toml @@ -5,4 +5,28 @@ edition.workspace = true license.workspace = true publish.workspace = true +# The organ's CultNet surface: the socket, the sessions, the process lifetime +# and the operation envelope, and no rule. Every rule, the instance check, the +# receipt, the status and the wire vocabulary are `huginn-mind`'s, and the leaf +# and the store type are reached through it so one crate pins one revision. +[lib] +name = "huginn_daemon" +path = "src/lib.rs" + +[[bin]] +name = "huginn-daemon" +path = "src/main.rs" + [dependencies] +anyhow = "1" +base64 = "0.22" +chrono = "0.4.44" +cultnet-rs = { git = "https://github.com/GameCult/CultLib.git", rev = "a0813c6eed24d30bf88073ef615b633c77ebfcd6", package = "cultnet-rs" } +huginn-mind = { path = "../huginn-mind" } +rmp-serde = "1" +schemars = "1" +serde = { version = "1", features = ["derive"] } +signal-hook = "0.3" + +[dev-dependencies] +tempfile = "3" diff --git a/crates/huginn-daemon/src/daemon.rs b/crates/huginn-daemon/src/daemon.rs new file mode 100644 index 0000000..c2a2722 --- /dev/null +++ b/crates/huginn-daemon/src/daemon.rs @@ -0,0 +1,377 @@ +//! The dispatch: one mind, one request, one response. No socket, no clock, no +//! rule. `now` is the caller's, so a test pins it and the loop is the only +//! thing that reads a wall clock. + +use std::path::Path; + +use chrono::{DateTime, Utc}; +use huginn_mind::epiphany_pipeline::{PipelineRef, Slug}; +use huginn_mind::wire::{HuginnMindRequest, HuginnMindResponse}; +use huginn_mind::{Mind, MindRefusal, MindStore, OwnedRedbMessagePackBackingStore, PipelineAdmissionOutcome}; + +/// Where Cut 11's index attaches. Called after a batch commits, with the refs +/// the batch landed, derived writes included; the sink may read each through +/// `mind.view`. Its error is logged and never reaches the caller: the index +/// does not decide whether a batch was admitted. +pub trait IndexSink { + fn committed(&mut self, mind: &Mind, writes: &[PipelineRef]) -> anyhow::Result<()>; +} + +/// This cut's index: there is none. +pub struct NoIndex; + +impl IndexSink for NoIndex { + fn committed(&mut self, _mind: &Mind, _writes: &[PipelineRef]) -> anyhow::Result<()> { + Ok(()) + } +} + +/// One mind and whatever indexes what lands in it. +pub struct Daemon> { + mind: Mind, + index: I, +} + +impl Daemon { + /// Opens the instance's mind, taking the store's exclusive lock for the + /// daemon's lifetime. Ruling 15: a mind that will not open is a refusal + /// here, before any socket exists; this function takes no address and + /// cannot bind one. + pub fn open(state_root: &Path, instance: &Slug) -> Result { + Ok(Self { mind: Mind::open(state_root, instance)?, index: NoIndex }) + } +} + +impl> Daemon { + pub fn with_index>(self, index: J) -> Daemon { + Daemon { mind: self.mind, index } + } + + pub fn mind(&self) -> &Mind { + &self.mind + } + + /// The envelope's `source_runtime_id` on every response this daemon sends. + pub fn runtime_id(&self) -> String { + format!("huginn-{}", self.mind.instance().0) + } + + /// One request, one answer. `Admit` goes to the mind whole, because the + /// batch declares its own instance and admission's A1 is the check. Every + /// other request names an instance, and the mind is asked whether that is + /// its own before the method runs. + pub fn handle(&mut self, request: HuginnMindRequest, now: DateTime) -> HuginnMindResponse { + if let Some(declared) = request.instance() + && !matches!(request, HuginnMindRequest::Admit(_)) + && let Err(refusal) = self.mind.require_instance(declared) + { + return HuginnMindResponse::Refused(refusal); + } + match request { + HuginnMindRequest::Whoami => HuginnMindResponse::Whoami(self.mind.status()), + HuginnMindRequest::Admit(batch) => { + let outcome = self.mind.admit(batch, now); + if let PipelineAdmissionOutcome::Committed { writes, .. } = &outcome + && let Err(error) = self.index.committed(&self.mind, writes) + { + let refs = writes.iter().map(|write| write.id.0.clone()).collect::>().join(", "); + eprintln!("huginn: the index refused the landed writes [{refs}]: {error:#}"); + } + HuginnMindResponse::Admit(outcome) + } + HuginnMindRequest::View { id, .. } => match self.mind.view(&id) { + Ok(view) => HuginnMindResponse::View(view), + Err(refusal) => HuginnMindResponse::Refused(refusal), + }, + HuginnMindRequest::Query { query, .. } => match self.mind.query(&query) { + Ok(page) => HuginnMindResponse::Query(page), + Err(refusal) => HuginnMindResponse::Refused(refusal), + }, + HuginnMindRequest::OpenItems { campaign, .. } => match self.mind.open_items(&campaign) { + Ok(items) => HuginnMindResponse::OpenItems(items), + Err(refusal) => HuginnMindResponse::Refused(refusal), + }, + HuginnMindRequest::History { scope, .. } => match self.mind.history(&scope) { + Ok(views) => HuginnMindResponse::History(views), + Err(refusal) => HuginnMindResponse::Refused(refusal), + }, + } + } +} + +#[cfg(test)] +pub(crate) mod tests { + use super::*; + use chrono::TimeZone; + use huginn_mind::epiphany_pipeline::{ + Date, OrgRepo, PipelineDocument, PipelineInstance, PipelineKind, PipelineQuestion, PipelineRuling, + QuestionOption, RulingAuthority, Short, + }; + use huginn_mind::wire::MindStatus; + use huginn_mind::{ + Faculty, HistoryScope, PipelineAdmissionBatch, PipelineProvenance, PipelineQuery, PipelineStatus, + }; + use std::sync::{Arc, Mutex}; + use tempfile::TempDir; + + pub(crate) const INSTANCE: &str = "yggdrasil"; + pub(crate) const OTHER: &str = "thought-cage"; + pub(crate) const CAMPAIGN: &str = "eureka-state"; + + pub(crate) fn slug(value: &str) -> Slug { + value.into() + } + + pub(crate) fn now() -> DateTime { + Utc.with_ymd_and_hms(2026, 9, 16, 12, 0, 0).unwrap() + } + + pub(crate) fn provenance() -> PipelineProvenance { + PipelineProvenance { + faculty: Faculty::Hands, + agent: Short("claude".into()), + session: Short("session-1".into()), + tool: Short("admit".into()), + } + } + + pub(crate) fn identity(name: &str) -> PipelineDocument { + PipelineDocument::Instance(PipelineInstance { + instance: slug(name), + display_name: Short(format!("{name} mind")), + created_at: Date("2026-09-16".into()), + host: Short(name.into()), + }) + } + + pub(crate) fn batch(instance: &str, documents: Vec) -> PipelineAdmissionBatch { + PipelineAdmissionBatch { instance: slug(instance), provenance: provenance(), documents } + } + + /// A question and the ruling that answers it: admitting the ruling derives + /// the question's resolution, so one batch lands two writes. + pub(crate) fn question_and_ruling() -> (PipelineDocument, PipelineDocument) { + let question = PipelineDocument::Question(PipelineQuestion { + campaign: slug(CAMPAIGN), + label: "Q1".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 ruling = PipelineDocument::Ruling(PipelineRuling { + campaign: slug(CAMPAIGN), + label: "R1".into(), + answers: Some(Short(format!("{CAMPAIGN}:question:Q1"))), + choice: Some("A".into()), + ruling: "An instance owns its mind.".into(), + operator_quote: None, + ruled_on: Date("2026-09-16".into()), + precedents: vec![], + authority: RulingAuthority::Operator, + }); + (question, ruling) + } + + /// A daemon over a real redb mind in a temporary directory, with its + /// identity admitted. The directory is returned because dropping it + /// removes the store. + pub(crate) fn seeded() -> (TempDir, Daemon) { + let root = tempfile::tempdir().unwrap(); + let mut daemon = Daemon::open(root.path(), &slug(INSTANCE)).unwrap(); + let outcome = daemon.handle(HuginnMindRequest::Admit(batch(INSTANCE, vec![identity(INSTANCE)])), now()); + assert!( + matches!(outcome, HuginnMindResponse::Admit(PipelineAdmissionOutcome::Committed { .. })), + "{outcome:?}" + ); + (root, daemon) + } + + #[derive(Clone, Default)] + struct RecordingIndex { + seen: Arc, Vec)>>>, + } + + impl IndexSink for RecordingIndex { + fn committed(&mut self, mind: &Mind, writes: &[PipelineRef]) -> anyhow::Result<()> { + let ids = writes.iter().map(|write| write.id.0.clone()).collect(); + let readable = writes.iter().map(|write| mind.view(write).unwrap().is_some()).collect(); + self.seen.lock().unwrap().push((ids, readable)); + Ok(()) + } + } + + struct FailingIndex; + + impl IndexSink for FailingIndex { + fn committed(&mut self, _mind: &Mind, _writes: &[PipelineRef]) -> anyhow::Result<()> { + Err(anyhow::anyhow!("the index is not there")) + } + } + + fn status(daemon: &mut Daemon>) -> MindStatus { + match daemon.handle(HuginnMindRequest::Whoami, now()) { + HuginnMindResponse::Whoami(status) => status, + other => panic!("expected a status, got {other:?}"), + } + } + + /// Typed hand-off between the organs without a socket: a batch enters and + /// every read answers over what it landed, at a clock the test pins. + #[test] + fn a_batch_round_trips_typed_through_handle_without_a_socket() { + let root = tempfile::tempdir().unwrap(); + let mut daemon = Daemon::open(root.path(), &slug(INSTANCE)).unwrap(); + let outcome = daemon.handle(HuginnMindRequest::Admit(batch(INSTANCE, vec![identity(INSTANCE)])), now()); + let HuginnMindResponse::Admit(PipelineAdmissionOutcome::Committed { writes, .. }) = outcome else { + panic!("expected a commit"); + }; + assert_eq!(writes.len(), 1); + assert_eq!(writes[0].kind, PipelineKind::Instance); + + assert_eq!( + status(&mut daemon), + MindStatus { + instance: slug(INSTANCE), + schema_epoch: "epiphany.pipeline.epoch.v1".into(), + documents: 1, + receipts: 1, + } + ); + assert_eq!(daemon.runtime_id(), "huginn-yggdrasil"); + + let query = PipelineQuery { kinds: vec![PipelineKind::Instance], ..PipelineQuery::default() }; + let HuginnMindResponse::Query(page) = + daemon.handle(HuginnMindRequest::Query { instance: slug(INSTANCE), query }, now()) + else { + panic!("expected a page"); + }; + assert_eq!(page.matched, 1); + assert_eq!(page.items[0].admission.provenance, provenance()); + assert_eq!(page.items[0].status, PipelineStatus::InForce); + + let id = writes[0].clone(); + let view = daemon.handle(HuginnMindRequest::View { instance: slug(INSTANCE), id: id.clone() }, now()); + assert_eq!(view, HuginnMindResponse::View(Some(page.items[0].clone()))); + + let HuginnMindResponse::OpenItems(items) = + daemon.handle(HuginnMindRequest::OpenItems { instance: slug(INSTANCE), campaign: slug(CAMPAIGN) }, now()) + else { + panic!("expected open items"); + }; + assert_eq!( + (items.questions.len(), items.findings.len(), items.follow_ups.len()), + (0, 0, 0), + "a mind holding only its identity has nothing open" + ); + assert!(items.specs_without_report.is_empty() && items.reports_without_verdict.is_empty()); + + let scope = HistoryScope::Repo(OrgRepo("GameCult/Huginn".into())); + assert_eq!( + daemon.handle(HuginnMindRequest::History { instance: slug(INSTANCE), scope }, now()), + HuginnMindResponse::History(vec![]) + ); + } + + /// Ruling 14 across the transport, on both sides: the mind refuses a read + /// and a write that name another instance, and nothing lands. + #[test] + fn a_read_or_a_write_naming_another_instance_is_refused_by_the_mind_and_writes_nothing() { + let root = tempfile::tempdir().unwrap(); + let mut daemon = Daemon::open(root.path(), &slug(INSTANCE)).unwrap(); + let landed = daemon.handle(HuginnMindRequest::Admit(batch(INSTANCE, vec![identity(INSTANCE)])), now()); + let HuginnMindResponse::Admit(PipelineAdmissionOutcome::Committed { writes, .. }) = landed else { + panic!("expected a commit"); + }; + let foreign = MindRefusal::ForeignInstance { declared: OTHER.into(), mind: INSTANCE.into() }; + + let query = HuginnMindRequest::Query { instance: slug(OTHER), query: PipelineQuery::default() }; + assert_eq!(daemon.handle(query, now()), HuginnMindResponse::Refused(foreign.clone())); + + let admit = HuginnMindRequest::Admit(batch(OTHER, vec![identity(OTHER)])); + assert_eq!( + daemon.handle(admit, now()), + HuginnMindResponse::Admit(PipelineAdmissionOutcome::Refused(foreign.clone())) + ); + + let view = HuginnMindRequest::View { instance: slug(OTHER), id: writes[0].clone() }; + assert_eq!(daemon.handle(view, now()), HuginnMindResponse::Refused(foreign.clone())); + + let items = HuginnMindRequest::OpenItems { instance: slug(OTHER), campaign: slug(CAMPAIGN) }; + assert_eq!(daemon.handle(items, now()), HuginnMindResponse::Refused(foreign.clone())); + + let scope = HistoryScope::Repo(OrgRepo("GameCult/Huginn".into())); + let history = HuginnMindRequest::History { instance: slug(OTHER), scope }; + assert_eq!(daemon.handle(history, now()), HuginnMindResponse::Refused(foreign)); + + assert_eq!(status(&mut daemon).documents, 1, "nothing landed"); + } + + /// Cut 11's seam: the index is handed every landed write, derived ones + /// included, and can read each; an index that fails changes nothing. + #[test] + fn the_index_is_handed_every_landed_write_after_commit_and_never_decides_the_outcome() { + let (_root, daemon) = seeded(); + let recording = RecordingIndex::default(); + let mut daemon = daemon.with_index(recording.clone()); + let (question, ruling) = question_and_ruling(); + + let outcome = daemon.handle(HuginnMindRequest::Admit(batch(INSTANCE, vec![question, ruling])), now()); + let HuginnMindResponse::Admit(PipelineAdmissionOutcome::Committed { writes, .. }) = outcome else { + panic!("expected a commit"); + }; + let landed: Vec = writes.iter().map(|write| write.id.0.clone()).collect(); + assert_eq!(landed.len(), 3, "the ruling answers the question, so a resolution is derived: {landed:?}"); + + let seen = recording.seen.lock().unwrap().clone(); + assert_eq!(seen.len(), 1); + assert_eq!(seen[0].0, landed, "the sink sees exactly the landed writes"); + assert_eq!(seen[0].1, vec![true, true, true], "and can read each one when it is called"); + + let (question, ruling) = question_and_ruling(); + let mut daemon = daemon.with_index(FailingIndex); + let root = tempfile::tempdir().unwrap(); + let mut fresh = Daemon::open(root.path(), &slug(INSTANCE)).unwrap().with_index(FailingIndex); + fresh.handle(HuginnMindRequest::Admit(batch(INSTANCE, vec![identity(INSTANCE)])), now()); + let outcome = fresh.handle(HuginnMindRequest::Admit(batch(INSTANCE, vec![question, ruling])), now()); + let HuginnMindResponse::Admit(PipelineAdmissionOutcome::Committed { writes, .. }) = outcome else { + panic!("a failing index does not refuse the batch"); + }; + for write in &writes { + let view = fresh.handle(HuginnMindRequest::View { instance: slug(INSTANCE), id: write.clone() }, now()); + assert!(matches!(view, HuginnMindResponse::View(Some(_))), "{write:?} is readable"); + } + assert_eq!(status(&mut daemon).documents, 4, "identity, question, ruling and the derived resolution"); + } + + /// Ruling 15: a mind that will not open is refused, and `Daemon::open` + /// takes no address, so no socket can precede it. `startup`'s order is + /// pinned in `serve::tests`. + #[test] + fn the_daemon_refuses_loudly_when_it_cannot_open_the_mind_and_binds_nothing() { + let root = tempfile::tempdir().unwrap(); + let held = Daemon::open(root.path(), &slug(INSTANCE)).unwrap(); + let path = Mind::::path_for(root.path(), &slug(INSTANCE)); + assert_eq!( + Daemon::open(root.path(), &slug(INSTANCE)).err(), + Some(MindRefusal::MindAlreadyOwned { path: path.display().to_string() }) + ); + drop(held); + + let mut planted = Daemon::open(root.path(), &slug(INSTANCE)).unwrap(); + planted.handle(HuginnMindRequest::Admit(batch(INSTANCE, vec![identity(INSTANCE)])), now()); + drop(planted); + let moved = Mind::::path_for(root.path(), &slug(OTHER)); + std::fs::create_dir_all(moved.parent().unwrap()).unwrap(); + std::fs::copy(&path, &moved).unwrap(); + assert_eq!( + Daemon::open(root.path(), &slug(OTHER)).err(), + Some(MindRefusal::ForeignInstance { declared: OTHER.into(), mind: INSTANCE.into() }) + ); + } +} diff --git a/crates/huginn-daemon/src/envelope.rs b/crates/huginn-daemon/src/envelope.rs new file mode 100644 index 0000000..66b0563 --- /dev/null +++ b/crates/huginn-daemon/src/envelope.rs @@ -0,0 +1,223 @@ +//! CultNet's operation envelope around the mind's vocabulary. Pure functions: +//! a request goes out as `cultnet.operation_request.v0` and an answer comes +//! back as `cultnet.operation_response.v0`, both carrying base64 of named +//! MessagePack, which is what the C# reference and Eve's plugin ABI speak. +//! +//! A message this module cannot decode is a typed failure with a code, never +//! a transport error and never a call into a mind. `payload_encoding` is +//! CultNet's own check in `validate_message` and is not re-checked here. +//! `source_runtime_id` and `target_runtime_id` on a request are read by +//! nothing: ruling 18 says an identity is declared in the payload, not +//! granted by the envelope. + +use anyhow::Result; +use base64::Engine; +use base64::engine::general_purpose::STANDARD; +use cultnet_rs::CultNetMessage; +use huginn_mind::wire::{HuginnMindRequest, HuginnMindResponse, MIND_REQUEST_SCHEMA, MIND_RESPONSE_SCHEMA, MIND_SERVICE_ID}; +use schemars::JsonSchema; +use serde::{Deserialize, Serialize}; + +/// The C# reference's `CultNetOperationServer.FailureSchemaId`. The Rust +/// `cultnet-rs` publishes no type for it, so this two-field mirror carries the +/// reference's keys; parity belongs in CultLib. +pub const FAILURE_SCHEMA: &str = "gamecult.cultnet.operation_failure.v1"; + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, JsonSchema)] +pub struct OperationFailure { + pub code: String, + pub message: String, +} + +impl OperationFailure { + fn new(code: &str, message: impl Into) -> Self { + Self { code: code.into(), message: message.into() } + } +} + +fn encode_payload(value: &T) -> Result { + Ok(STANDARD.encode(rmp_serde::to_vec_named(value)?)) +} + +/// One request, addressed to the mind service and naming its own operation. +pub fn encode_request( + message_id: &str, + request: &HuginnMindRequest, + source_runtime_id: Option, +) -> Result { + Ok(CultNetMessage::OperationRequest { + message_id: message_id.to_string(), + service_id: MIND_SERVICE_ID.to_string(), + operation: request.operation().to_string(), + payload_schema: MIND_REQUEST_SCHEMA.to_string(), + payload_encoding: "messagepack-base64".to_string(), + payload: encode_payload(request)?, + source_runtime_id, + target_runtime_id: None, + }) +} + +/// The message, or the failure that says why it is not a request of this +/// service. The envelope's `operation` string is checked against the decoded +/// request rather than trusted, so a payload cannot arrive under another +/// operation's name. +pub fn decode_request(message: &CultNetMessage) -> Result<(String, HuginnMindRequest), OperationFailure> { + let CultNetMessage::OperationRequest { message_id, service_id, operation, payload_schema, payload, .. } = message + else { + return Err(OperationFailure::new( + "not-an-operation-request", + "huginn.mind reads cultnet.operation_request.v0", + )); + }; + if service_id != MIND_SERVICE_ID { + return Err(OperationFailure::new("wrong-service", format!("{service_id} is not {MIND_SERVICE_ID}"))); + } + if payload_schema != MIND_REQUEST_SCHEMA { + return Err(OperationFailure::new( + "wrong-payload-schema", + format!("{payload_schema} is not {MIND_REQUEST_SCHEMA}"), + )); + } + let bytes = STANDARD + .decode(payload) + .map_err(|error| OperationFailure::new("payload-not-base64", error.to_string()))?; + let request: HuginnMindRequest = rmp_serde::from_slice(&bytes) + .map_err(|error| OperationFailure::new("payload-not-a-request", error.to_string()))?; + if operation != request.operation() { + return Err(OperationFailure::new( + "operation-mismatch", + format!("the envelope says {operation} and the payload is {}", request.operation()), + )); + } + Ok((message_id.clone(), request)) +} + +/// One answer, whose status the response derives. +pub fn encode_response( + message_id: &str, + operation: &str, + response: &HuginnMindResponse, + source_runtime_id: &str, +) -> Result { + Ok(CultNetMessage::OperationResponse { + message_id: message_id.to_string(), + service_id: MIND_SERVICE_ID.to_string(), + operation: operation.to_string(), + status: response.status().to_string(), + payload_schema: MIND_RESPONSE_SCHEMA.to_string(), + payload_encoding: "messagepack-base64".to_string(), + payload: encode_payload(response)?, + diagnostics: vec![], + source_runtime_id: Some(source_runtime_id.to_string()), + }) +} + +/// The answer to an envelope that never reached a mind. Infallible: the +/// failure is two strings and the daemon must always be able to say no. +pub fn encode_failure( + message_id: &str, + operation: &str, + failure: &OperationFailure, + source_runtime_id: &str, +) -> CultNetMessage { + CultNetMessage::OperationResponse { + message_id: message_id.to_string(), + service_id: MIND_SERVICE_ID.to_string(), + operation: operation.to_string(), + status: "rejected".to_string(), + payload_schema: FAILURE_SCHEMA.to_string(), + payload_encoding: "messagepack-base64".to_string(), + payload: STANDARD.encode(rmp_serde::to_vec_named(failure).expect("two strings encode")), + diagnostics: vec![failure.code.clone()], + source_runtime_id: Some(source_runtime_id.to_string()), + } +} + +/// The client's side: the correlation key and either the mind's answer or the +/// failure that says the envelope never reached one. +pub fn decode_response(message: &CultNetMessage) -> Result<(String, Result)> { + let CultNetMessage::OperationResponse { message_id, payload_schema, payload, .. } = message else { + anyhow::bail!("huginn.mind answers with cultnet.operation_response.v0"); + }; + let bytes = STANDARD.decode(payload)?; + let answer = match payload_schema.as_str() { + MIND_RESPONSE_SCHEMA => Ok(rmp_serde::from_slice::(&bytes)?), + FAILURE_SCHEMA => Err(rmp_serde::from_slice::(&bytes)?), + other => anyhow::bail!("{other} is neither {MIND_RESPONSE_SCHEMA} nor {FAILURE_SCHEMA}"), + }; + Ok((message_id.clone(), answer)) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::daemon::tests::{INSTANCE, OTHER, batch, identity, slug}; + use huginn_mind::wire::MindStatus; + use huginn_mind::{MindRefusal, PipelineAdmissionOutcome}; + + fn refusal() -> MindRefusal { + MindRefusal::ForeignInstance { declared: OTHER.into(), mind: INSTANCE.into() } + } + + /// A refusal is an answer: it rides the response schema with a `rejected` + /// status and decodes back to the same value. A conflict and a replay are + /// answers about the mind, not rejections of the request. + #[test] + fn a_refusal_is_a_typed_response_not_a_transport_failure() { + let cases = [ + (HuginnMindResponse::Refused(refusal()), "rejected"), + (HuginnMindResponse::Admit(PipelineAdmissionOutcome::Refused(refusal())), "rejected"), + ( + HuginnMindResponse::Admit(PipelineAdmissionOutcome::Committed { + receipt_id: "r1".into(), + committed_at: "2026-09-16T12:00:00Z".into(), + writes: vec![], + }), + "accepted", + ), + (HuginnMindResponse::Admit(PipelineAdmissionOutcome::AlreadyAdmitted { receipt_id: "r1".into() }), "accepted"), + (HuginnMindResponse::Admit(PipelineAdmissionOutcome::Conflict { identities: vec![] }), "accepted"), + ( + HuginnMindResponse::Whoami(MindStatus { + instance: slug(INSTANCE), + schema_epoch: "epiphany.pipeline.epoch.v1".into(), + documents: 0, + receipts: 0, + }), + "accepted", + ), + ]; + for (response, expected) in cases { + let message = encode_response("m-1", "admit", &response, "huginn-yggdrasil").unwrap(); + let CultNetMessage::OperationResponse { status, payload_schema, diagnostics, .. } = &message else { + panic!("not an operation response"); + }; + assert_eq!(status, expected, "{response:?}"); + assert_eq!(payload_schema, MIND_RESPONSE_SCHEMA); + assert!(diagnostics.is_empty()); + assert_eq!(decode_response(&message).unwrap(), ("m-1".into(), Ok(response))); + } + + let failure = OperationFailure::new("wrong-service", "odin.catalog is not huginn.mind"); + let message = encode_failure("m-2", "query", &failure, "huginn-yggdrasil"); + let CultNetMessage::OperationResponse { status, payload_schema, diagnostics, .. } = &message else { + panic!("not an operation response"); + }; + assert_eq!((status.as_str(), payload_schema.as_str()), ("rejected", FAILURE_SCHEMA)); + assert_eq!(diagnostics, &vec!["wrong-service".to_string()]); + assert_eq!(decode_response(&message).unwrap(), ("m-2".into(), Err(failure))); + } + + /// The round trip the client and the daemon share, and the one code that + /// `serve::answer` cannot reach because it answers a foreign message with + /// `Error` instead. + #[test] + fn a_request_round_trips_and_a_foreign_message_is_not_a_request() { + let request = HuginnMindRequest::Admit(batch(INSTANCE, vec![identity(INSTANCE)])); + let message = encode_request("m-3", &request, Some("eureka-state".into())).unwrap(); + assert_eq!(decode_request(&message).unwrap(), ("m-3".into(), request)); + + let foreign = CultNetMessage::Error { error: "not for you".into() }; + assert_eq!(decode_request(&foreign).unwrap_err().code, "not-an-operation-request"); + } +} diff --git a/crates/huginn-daemon/src/lib.rs b/crates/huginn-daemon/src/lib.rs new file mode 100644 index 0000000..ee36d7c --- /dev/null +++ b/crates/huginn-daemon/src/lib.rs @@ -0,0 +1,21 @@ +//! `huginn-daemon`: one process that serves one mind over CultNet. +//! +//! The crate owns four things and no rule. `Daemon::open` owns "may this +//! process serve this mind", by delegating to `Mind::open` and refusing before +//! anything listens. `Daemon::handle` owns "which `Mind` method answers this +//! request": it dispatches, it asks `Mind::require_instance` for every read +//! that names an instance, and it compares nothing itself. `envelope` owns "is +//! this a request at all": a message it cannot decode is answered with a typed +//! failure and touches no mind. `serve` owns "which session gets which reply". +//! +//! Every admission rule, the instance check, the receipt, the derived status +//! and the wire vocabulary are `huginn-mind`'s. The leaf and the store type +//! are reached through `huginn_mind`, so one crate pins one revision of each. + +pub mod daemon; +pub mod envelope; +pub mod serve; + +pub use daemon::{Daemon, IndexSink, NoIndex}; +pub use envelope::{FAILURE_SCHEMA, OperationFailure}; +pub use serve::{Options, ServeOptions, parse_options, run, startup}; diff --git a/crates/huginn-daemon/src/main.rs b/crates/huginn-daemon/src/main.rs index f328e4d..2c14b8d 100644 --- a/crates/huginn-daemon/src/main.rs +++ b/crates/huginn-daemon/src/main.rs @@ -1 +1,32 @@ -fn main() {} +//! The process: open one instance's mind, bind one socket, serve until a stop +//! signal. A mind that will not open ends the process with the refusal on +//! stderr and no socket ever bound. + +use std::sync::Arc; +use std::sync::atomic::AtomicBool; + +use anyhow::{Context, Result}; +use huginn_daemon::serve::{ServeOptions, parse_options, run, startup}; + +fn main() -> Result<()> { + let options = parse_options(std::env::args().skip(1))?; + let (mut daemon, mut hub, registry) = match startup(&options) { + Ok(started) => started, + Err(error) => { + eprintln!("huginn-daemon refuses to serve {}: {error:#}", options.instance.0); + std::process::exit(1); + } + }; + + // Idunn runs this process as PID 1 of its own PID namespace, and a + // namespace init ignores every signal it has not caught. Without this, + // a stop is a ninety-second wait for SIGKILL. + let stopping = Arc::new(AtomicBool::new(false)); + for signal in [signal_hook::consts::SIGTERM, signal_hook::consts::SIGINT] { + signal_hook::flag::register(signal, Arc::clone(&stopping)) + .context("registering Huginn's stop signal handler")?; + } + + eprintln!("huginn-daemon serving {} on {}", options.instance.0, hub.local_addr()?); + run(&mut daemon, &mut hub, ®istry, &stopping, &ServeOptions::default()) +} diff --git a/crates/huginn-daemon/src/serve.rs b/crates/huginn-daemon/src/serve.rs new file mode 100644 index 0000000..dd3bd42 --- /dev/null +++ b/crates/huginn-daemon/src/serve.rs @@ -0,0 +1,463 @@ +//! The loop: one UDP socket, one reliable session per peer, one reply per +//! frame on the frame's own session. The hub is CultLib's; what is here is the +//! routing, the schema catalog and the order the process starts in. +//! +//! Stated limits. A socket error and a hostile datagram are indistinguishable +//! at `receive_event_once`, so both are logged and served past: a dead socket +//! spins with logging rather than exiting, and Cut 14's health check is the +//! observer. The reliable window is `max_pending_reliable_packets` times +//! `max_fragment_bytes` per session, about 1.2 MB in flight; a page larger +//! than that fails to send and the client times out rather than receiving a +//! typed answer. + +use std::collections::BTreeMap; +use std::net::{SocketAddr, UdpSocket}; +use std::path::PathBuf; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::time::Duration; + +use anyhow::{Context, Result, ensure}; +use chrono::{DateTime, Utc}; +use cultnet_rs::{ + CULTNET_OPERATION_CONNECTION_ID, CultNetMessage, CultNetRudpServerEvent, CultNetRudpServerHub, + CultNetRudpServerHubOptions, CultNetSchemaKind, CultNetSchemaRegistration, CultNetSchemaRegistry, + CultNetWireContract, decode_cultnet_message_from_slice, +}; +use huginn_mind::epiphany_pipeline::Slug; +use huginn_mind::wire::{ + MIND_REQUEST_SCHEMA, MIND_REQUEST_SCHEMA_JSON, MIND_RESPONSE_SCHEMA, MIND_RESPONSE_SCHEMA_JSON, MIND_SERVICE_ID, +}; +use huginn_mind::{MindStore, OwnedRedbMessagePackBackingStore}; + +use crate::daemon::{Daemon, IndexSink, NoIndex}; +use crate::envelope::{decode_request, encode_failure, encode_response}; + +/// What the loop does when nothing is waiting. +pub struct ServeOptions { + pub session_timeout: Duration, + pub idle_sleep: Duration, +} + +impl Default for ServeOptions { + fn default() -> Self { + Self { session_timeout: Duration::from_secs(30), idle_sleep: Duration::from_millis(2) } + } +} + +/// The three things the operator supplies. Nothing here reads the environment: +/// how Idunn supplies `--bind` is Cut 14's. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct Options { + pub state_root: PathBuf, + pub instance: Slug, + pub bind: SocketAddr, +} + +/// Exactly `--state-root`, `--instance` and `--bind`, each required, each once. +pub fn parse_options(args: impl Iterator) -> Result { + let mut args = args; + let mut values: BTreeMap = BTreeMap::new(); + 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"), "unsupported Huginn option --{name}"); + let value = args.next().with_context(|| format!("missing value for --{name}"))?; + ensure!(values.insert(name.to_owned(), value).is_none(), "duplicate Huginn option --{name}"); + } + let take = |name: &str| -> Result { + values.get(name).cloned().with_context(|| format!("--{name} is required")) + }; + let state_root = PathBuf::from(take("state-root")?); + ensure!(state_root.is_absolute(), "--state-root must be an absolute path"); + let bind = take("bind")?; + Ok(Options { + state_root, + instance: Slug(take("instance")?), + bind: bind.parse().with_context(|| format!("--bind must be an ip:port, got {bind:?}"))?, + }) +} + +/// Ruling 15, as an order: the mind opens first, and only a mind that opened +/// gets a socket. This is the one place both happen, so nothing can bind ahead +/// of the refusal. +pub fn startup( + options: &Options, +) -> Result<(Daemon, CultNetRudpServerHub, CultNetSchemaRegistry)> { + let daemon = Daemon::open(&options.state_root, &options.instance)?; + let hub = bind(options.bind, &daemon.runtime_id())?; + Ok((daemon, hub, schema_registry()?)) +} + +/// One non-blocking socket serving the operation connection id. The hub drops +/// every datagram carrying another one. +pub fn bind(addr: SocketAddr, runtime_id: &str) -> Result { + let socket = UdpSocket::bind(addr).with_context(|| format!("binding {addr}"))?; + socket.set_nonblocking(true)?; + let mut options = CultNetRudpServerHubOptions::new(runtime_id, socket, CULTNET_OPERATION_CONNECTION_ID); + options.max_fragment_bytes = Some(1200); + options.max_pending_reliable_packets = Some(1024); + options.max_peers = 256; + CultNetRudpServerHub::new(options) +} + +/// The organ's two contracts, as the catalog advertises them. +pub fn schema_registry() -> Result { + let mut registry = CultNetSchemaRegistry::new(); + for (schema_id, title, schema_json) in [ + (MIND_REQUEST_SCHEMA, "Huginn mind request", MIND_REQUEST_SCHEMA_JSON), + (MIND_RESPONSE_SCHEMA, "Huginn mind response", MIND_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) +} + +/// One message in, one message out. Nothing here reaches a mind except through +/// `Daemon::handle`, and an envelope that does not decode never gets that far. +pub fn answer>( + daemon: &mut Daemon, + registry: &CultNetSchemaRegistry, + message: CultNetMessage, + now: DateTime, +) -> CultNetMessage { + let runtime_id = daemon.runtime_id(); + match &message { + CultNetMessage::OperationRequest { message_id, operation, .. } => match decode_request(&message) { + Ok((message_id, request)) => { + let operation = request.operation(); + let response = daemon.handle(request, now); + match encode_response(&message_id, operation, &response, &runtime_id) { + Ok(message) => message, + Err(error) => encode_failure( + &message_id, + operation, + &crate::envelope::OperationFailure { + code: "response-not-encodable".into(), + message: format!("{error:#}"), + }, + &runtime_id, + ), + } + } + Err(failure) => encode_failure(message_id, operation, &failure, &runtime_id), + }, + CultNetMessage::SchemaCatalogRequest { .. } => match registry.create_catalog_response(&message) { + Ok(response) => response, + Err(error) => CultNetMessage::Error { error: format!("{error:#}") }, + }, + _ => CultNetMessage::Error { + error: format!( + "{MIND_SERVICE_ID} answers cultnet.operation_request.v0 and cultnet.schema_catalog_request.v0" + ), + }, + } +} + +/// Until `stopping`: expire what timed out, resend what was not acknowledged, +/// then answer every frame waiting on the socket. A hostile datagram and a +/// departed session are logged and served past, never fatal. This is the +/// crate's only clock read, once per frame. +pub fn run>( + daemon: &mut Daemon, + hub: &mut CultNetRudpServerHub, + registry: &CultNetSchemaRegistry, + stopping: &AtomicBool, + options: &ServeOptions, +) -> Result<()> { + let timeout_ms = options.session_timeout.as_millis() as u64; + while !stopping.load(Ordering::Relaxed) { + hub.remove_timed_out_sessions(timeout_ms); + if let Err(error) = hub.poll_resends() { + eprintln!("huginn: resending to a peer failed: {error:#}"); + } + let mut served = 0_u32; + loop { + let event = match hub.receive_event_once() { + Ok(Some(event)) => event, + Ok(None) => break, + Err(error) => { + eprintln!("huginn: a datagram was not readable and was discarded: {error:#}"); + break; + } + }; + let CultNetRudpServerEvent::Frame { session, frame } = event else { + continue; + }; + if frame.channel_id != "schema" { + continue; + } + let message = match decode_cultnet_message_from_slice(&frame.payload, CultNetWireContract::CultNetSchemaV0) + { + Ok(message) => message, + Err(error) => { + eprintln!("huginn: a frame was not a CultNet message and was discarded: {error:#}"); + continue; + } + }; + let reply = answer(daemon, registry, message, Utc::now()); + if let Err(error) = hub.send_schema_message(&session, &reply) { + eprintln!("huginn: {} did not receive its reply: {error:#}", session.remote_addr); + } + served += 1; + } + if served == 0 { + std::thread::sleep(options.idle_sleep); + } + } + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::daemon::tests::{INSTANCE, batch, identity, now, slug}; + use crate::envelope::{FAILURE_SCHEMA, decode_response, encode_request}; + use base64::Engine; + use base64::engine::general_purpose::STANDARD; + use cultnet_rs::{CultMesh, CultMeshRudpSocketOptions}; + use huginn_mind::wire::{HuginnMindRequest, HuginnMindResponse, MindStatus}; + use huginn_mind::{PipelineAdmissionOutcome, PipelineQuery}; + use std::sync::Arc; + use std::sync::atomic::AtomicBool; + + fn documents(daemon: &mut Daemon) -> u32 { + match daemon.handle(HuginnMindRequest::Whoami, now()) { + HuginnMindResponse::Whoami(MindStatus { documents, .. }) => documents, + other => panic!("expected a status, got {other:?}"), + } + } + + fn request( + message_id: &str, + service_id: &str, + operation: &str, + payload_schema: &str, + payload: String, + ) -> CultNetMessage { + CultNetMessage::OperationRequest { + message_id: message_id.into(), + service_id: service_id.into(), + operation: operation.into(), + payload_schema: payload_schema.into(), + payload_encoding: "messagepack-base64".into(), + payload, + source_runtime_id: None, + target_runtime_id: None, + } + } + + fn rejected(message: &CultNetMessage) -> (String, String, String) { + let CultNetMessage::OperationResponse { message_id, status, payload_schema, payload, .. } = message else { + panic!("expected an operation response, got {message:?}"); + }; + assert_eq!(status, "rejected"); + assert_eq!(payload_schema, FAILURE_SCHEMA); + let failure = decode_response(message).unwrap().1.unwrap_err(); + (message_id.clone(), failure.code, payload.clone()) + } + + /// Six malformed envelopes, each with its own code, none of which reaches + /// a mind; a message that is not an operation at all is answered rather + /// than dropped; and the catalog answers with both schemas. + #[test] + fn a_malformed_envelope_is_answered_with_a_failure_and_touches_no_mind() { + let root = tempfile::tempdir().unwrap(); + let mut daemon = Daemon::open(root.path(), &slug(INSTANCE)).unwrap(); + let registry = schema_registry().unwrap(); + let admit = STANDARD.encode( + rmp_serde::to_vec_named(&HuginnMindRequest::Admit(batch(INSTANCE, vec![identity(INSTANCE)]))).unwrap(), + ); + + let cases = [ + ("m-1", request("m-1", "odin.catalog", "admit", MIND_REQUEST_SCHEMA, admit.clone()), "wrong-service"), + ("m-2", request("m-2", MIND_SERVICE_ID, "admit", "odin.topology.v1", admit.clone()), "wrong-payload-schema"), + ( + "m-3", + request("m-3", MIND_SERVICE_ID, "admit", MIND_REQUEST_SCHEMA, "not base64!!".into()), + "payload-not-base64", + ), + ( + "m-4", + request("m-4", MIND_SERVICE_ID, "admit", MIND_REQUEST_SCHEMA, STANDARD.encode([0xc1_u8, 0xc1])), + "payload-not-a-request", + ), + ("m-5", request("m-5", MIND_SERVICE_ID, "query", MIND_REQUEST_SCHEMA, admit), "operation-mismatch"), + ]; + for (id, message, code) in cases { + let reply = answer(&mut daemon, ®istry, message, now()); + assert_eq!(rejected(&reply).0, id, "the client's own correlation key is echoed"); + assert_eq!(rejected(&reply).1, code); + } + // The sixth code, `not-an-operation-request`, is the codec's and is + // pinned in `envelope::tests`: `answer` never produces it, because a + // message of another family is answered with `Error` instead. + let reply = answer(&mut daemon, ®istry, CultNetMessage::Error { error: "hello".into() }, now()); + let CultNetMessage::Error { error } = &reply else { panic!("expected an error, got {reply:?}") }; + assert!( + !error.is_empty() && error.contains("cultnet.operation_request.v0") && error.contains("cultnet.schema_catalog_request.v0"), + "{error}" + ); + + assert_eq!(documents(&mut daemon), 0, "no malformed envelope reached the mind"); + + let catalog = CultNetMessage::SchemaCatalogRequest { + message_id: "m-6".into(), + include_schema_json: Some(true), + schema_ids: None, + kinds: None, + }; + let reply = answer(&mut daemon, ®istry, catalog, now()); + let CultNetMessage::SchemaCatalogResponse { message_id, schemas } = reply else { + panic!("expected a catalog response"); + }; + assert_eq!(message_id, "m-6"); + let ids: Vec<&str> = schemas.iter().map(|schema| schema.schema_id.as_str()).collect(); + assert_eq!(ids, [MIND_REQUEST_SCHEMA, MIND_RESPONSE_SCHEMA]); + for schema in &schemas { + assert_eq!(schema.content_hash, registry.get(&schema.schema_id, true).unwrap().content_hash); + assert!(schema.schema_json.as_deref().unwrap().contains("HuginnMind")); + } + } + + /// Two clients over loopback, each answered on its own session; a client on + /// another connection id is not served at all; the stop flag ends the loop. + #[test] + fn two_clients_get_their_own_replies_over_loopback() { + let root = tempfile::tempdir().unwrap(); + let mut daemon = Daemon::open(root.path(), &slug(INSTANCE)).unwrap(); + 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 mut clients = Vec::new(); + for (name, connection_id) in + [("a", CULTNET_OPERATION_CONNECTION_ID), ("b", CULTNET_OPERATION_CONNECTION_ID), ("c", 0x4355_4c55)] + { + let mut client = CultMesh::create_rudp_client_for_endpoint( + format!("eureka-state-{name}"), + connection_id, + &endpoint, + CultMeshRudpSocketOptions::default(), + ) + .unwrap(); + client.connect(Vec::new()).unwrap(); + clients.push(client); + } + // The two admitted peers connect; the third's packets are dropped on + // their connection id and it never becomes a session. + for _ in 0..200 { + while hub.receive_event_once().unwrap().is_some() {} + for client in clients.iter_mut() { + let _ = client.receive_once(); + } + if clients[0].connected() && clients[1].connected() { + break; + } + std::thread::sleep(Duration::from_millis(2)); + } + assert!(clients[0].connected() && clients[1].connected()); + assert!(!clients[2].connected(), "another connection id is not served"); + assert_eq!(hub.sessions().len(), 2); + + let requests = [ + ("m-a", HuginnMindRequest::Admit(batch(INSTANCE, vec![identity(INSTANCE)]))), + ("m-b", HuginnMindRequest::Query { instance: slug(INSTANCE), query: PipelineQuery::default() }), + ]; + for (client, (message_id, request)) in clients.iter_mut().zip(requests.iter()) { + client.send_schema_message(&encode_request(message_id, request, None).unwrap()).unwrap(); + } + // The third client cannot even send: the hub dropped every packet + // carrying its connection id, so it never became a session. + let whoami = encode_request("m-c", &HuginnMindRequest::Whoami, None).unwrap(); + assert!(clients[2].send_schema_message(&whoami).is_err(), "an unserved connection id has no session"); + + let loop_stopping = Arc::clone(&stopping); + let serving = std::thread::spawn(move || { + let result = run(&mut daemon, &mut hub, ®istry, &loop_stopping, &ServeOptions::default()); + (result, daemon) + }); + + for (index, message_id) in ["m-a", "m-b"].iter().enumerate() { + let mut reply = None; + for _ in 0..500 { + clients[index].poll_resends().unwrap(); + if let Some(message) = clients[index].receive_schema_message_once().unwrap() { + reply = Some(message); + break; + } + std::thread::sleep(Duration::from_millis(2)); + } + let reply = reply.unwrap_or_else(|| panic!("client {index} got no reply")); + let (correlation, answer) = decode_response(&reply).unwrap(); + assert_eq!(&correlation, message_id, "each session gets its own reply"); + match (index, answer.unwrap()) { + (0, HuginnMindResponse::Admit(PipelineAdmissionOutcome::Committed { .. })) => {} + (1, HuginnMindResponse::Query(page)) => assert_eq!(page.matched, 1), + (index, other) => panic!("client {index} got {other:?}"), + } + } + // The third client, on its own connection id, is not answered. + for _ in 0..50 { + clients[2].poll_resends().unwrap(); + assert!(clients[2].receive_schema_message_once().unwrap().is_none()); + std::thread::sleep(Duration::from_millis(2)); + } + + stopping.store(true, Ordering::Relaxed); + let (result, mut daemon) = serving.join().unwrap(); + result.unwrap(); + assert_eq!(documents(&mut daemon), 1, "the admission that crossed the wire landed"); + } + + /// Ruling 15 as an order in the one place both happen: a mind that will not + /// open leaves the port free, so the socket cannot precede the refusal. + #[test] + fn startup_opens_the_mind_before_it_binds_anything() { + let root = tempfile::tempdir().unwrap(); + let held = Daemon::open(root.path(), &slug(INSTANCE)).unwrap(); + let probe = UdpSocket::bind("127.0.0.1:0").unwrap(); + let port = probe.local_addr().unwrap(); + drop(probe); + + let options = Options { state_root: root.path().to_path_buf(), instance: slug(INSTANCE), bind: port }; + let error = startup(&options).err().expect("a held mind refuses"); + assert!(format!("{error:#}").contains("MindAlreadyOwned"), "{error:#}"); + drop(held); + // Nothing bound the port, so the test can. + UdpSocket::bind(port).expect("startup bound nothing before the mind refused"); + } + + #[test] + fn options_are_exactly_state_root_instance_and_bind() { + let root = if cfg!(windows) { r"C:\state" } else { "/state" }; + let args = |values: &[&str]| values.iter().map(|value| value.to_string()).collect::>().into_iter(); + assert_eq!( + parse_options(args(&["--state-root", root, "--instance", "yggdrasil", "--bind", "127.0.0.1:17872"])) + .unwrap(), + Options { + state_root: PathBuf::from(root), + instance: slug("yggdrasil"), + bind: "127.0.0.1:17872".parse().unwrap(), + } + ); + for bad in [ + vec!["--state-root", root, "--instance", "yggdrasil"], + vec!["--state-root", root, "--bind", "127.0.0.1:1"], + vec!["--instance", "yggdrasil", "--bind", "127.0.0.1:1"], + vec!["--state-root", root, "--instance", "yggdrasil", "--bind", "127.0.0.1:1", "--bind", "127.0.0.1:2"], + vec!["--state-root", root, "--instance", "yggdrasil", "--bind", "not-an-address"], + vec!["--state-root", "relative", "--instance", "yggdrasil", "--bind", "127.0.0.1:1"], + vec!["--state-root", root, "--instance", "yggdrasil", "--bind", "127.0.0.1:1", "--idunn-anchor", "x"], + ] { + assert!(parse_options(args(&bad)).is_err(), "{bad:?}"); + } + } +}