Serve one mind over CultNet, and no rule beside it
`huginn-daemon` becomes a library and a binary. `Daemon` holds one mind and dispatches a `HuginnMindRequest` to the `Mind` method that answers it, with no socket and no clock: `now` is the caller's. Every read that names an instance is put to `Mind::require_instance` before the method runs, so ruling 14 is checked once, by its owner, whichever side of the wire the request came from. `IndexSink` is where Cut 11 attaches; its error is logged and the outcome is what the mind decided. `envelope` carries the vocabulary in CultNet's `cultnet.operation_request.v0` and `cultnet.operation_response.v0` as base64 of named MessagePack, which is what the C# reference and Eve's plugin ABI speak. The envelope's operation string is checked against the decoded payload rather than trusted. A message that does not decode is answered with a typed failure naming its code, and reaches no mind. `serve` binds one non-blocking socket on CultNet's operation connection id and answers each frame on the session it arrived on. A refusal is an answer with a `rejected` status, never a transport error. The two wire schemas go out on a schema catalog request. A hostile datagram and a departed peer are logged and served past. `startup` is the one place the mind opens and the socket binds, in that order, so a mind that will not open leaves the port free. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
Meta Vens
parent
c4b2a0fa2e
commit
6b619d6bbd
7 changed files with 1600 additions and 4 deletions
463
Cargo.lock
generated
463
Cargo.lock
generated
|
|
@ -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"
|
||||
|
|
|
|||
|
|
@ -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"
|
||||
|
|
|
|||
377
crates/huginn-daemon/src/daemon.rs
Normal file
377
crates/huginn-daemon/src/daemon.rs
Normal file
|
|
@ -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<S: MindStore> {
|
||||
fn committed(&mut self, mind: &Mind<S>, writes: &[PipelineRef]) -> anyhow::Result<()>;
|
||||
}
|
||||
|
||||
/// This cut's index: there is none.
|
||||
pub struct NoIndex;
|
||||
|
||||
impl<S: MindStore> IndexSink<S> for NoIndex {
|
||||
fn committed(&mut self, _mind: &Mind<S>, _writes: &[PipelineRef]) -> anyhow::Result<()> {
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
/// One mind and whatever indexes what lands in it.
|
||||
pub struct Daemon<S: MindStore, I: IndexSink<S>> {
|
||||
mind: Mind<S>,
|
||||
index: I,
|
||||
}
|
||||
|
||||
impl Daemon<OwnedRedbMessagePackBackingStore, NoIndex> {
|
||||
/// 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<Self, MindRefusal> {
|
||||
Ok(Self { mind: Mind::open(state_root, instance)?, index: NoIndex })
|
||||
}
|
||||
}
|
||||
|
||||
impl<S: MindStore, I: IndexSink<S>> Daemon<S, I> {
|
||||
pub fn with_index<J: IndexSink<S>>(self, index: J) -> Daemon<S, J> {
|
||||
Daemon { mind: self.mind, index }
|
||||
}
|
||||
|
||||
pub fn mind(&self) -> &Mind<S> {
|
||||
&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<Utc>) -> 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::<Vec<_>>().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> {
|
||||
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<PipelineDocument>) -> 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<OwnedRedbMessagePackBackingStore, NoIndex>) {
|
||||
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<Mutex<Vec<(Vec<String>, Vec<bool>)>>>,
|
||||
}
|
||||
|
||||
impl<S: MindStore> IndexSink<S> for RecordingIndex {
|
||||
fn committed(&mut self, mind: &Mind<S>, 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<S: MindStore> IndexSink<S> for FailingIndex {
|
||||
fn committed(&mut self, _mind: &Mind<S>, _writes: &[PipelineRef]) -> anyhow::Result<()> {
|
||||
Err(anyhow::anyhow!("the index is not there"))
|
||||
}
|
||||
}
|
||||
|
||||
fn status(daemon: &mut Daemon<OwnedRedbMessagePackBackingStore, impl IndexSink<OwnedRedbMessagePackBackingStore>>) -> 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<String> = 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::<OwnedRedbMessagePackBackingStore>::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::<OwnedRedbMessagePackBackingStore>::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() })
|
||||
);
|
||||
}
|
||||
}
|
||||
223
crates/huginn-daemon/src/envelope.rs
Normal file
223
crates/huginn-daemon/src/envelope.rs
Normal file
|
|
@ -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<String>) -> Self {
|
||||
Self { code: code.into(), message: message.into() }
|
||||
}
|
||||
}
|
||||
|
||||
fn encode_payload<T: Serialize>(value: &T) -> Result<String> {
|
||||
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<String>,
|
||||
) -> Result<CultNetMessage> {
|
||||
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<CultNetMessage> {
|
||||
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<HuginnMindResponse, OperationFailure>)> {
|
||||
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::<HuginnMindResponse>(&bytes)?),
|
||||
FAILURE_SCHEMA => Err(rmp_serde::from_slice::<OperationFailure>(&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");
|
||||
}
|
||||
}
|
||||
21
crates/huginn-daemon/src/lib.rs
Normal file
21
crates/huginn-daemon/src/lib.rs
Normal file
|
|
@ -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};
|
||||
|
|
@ -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())
|
||||
}
|
||||
|
|
|
|||
463
crates/huginn-daemon/src/serve.rs
Normal file
463
crates/huginn-daemon/src/serve.rs
Normal file
|
|
@ -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<Item = String>) -> Result<Options> {
|
||||
let mut args = args;
|
||||
let mut values: BTreeMap<String, String> = 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<String> {
|
||||
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<OwnedRedbMessagePackBackingStore, NoIndex>, 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<CultNetRudpServerHub> {
|
||||
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<CultNetSchemaRegistry> {
|
||||
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<S: MindStore, I: IndexSink<S>>(
|
||||
daemon: &mut Daemon<S, I>,
|
||||
registry: &CultNetSchemaRegistry,
|
||||
message: CultNetMessage,
|
||||
now: DateTime<Utc>,
|
||||
) -> 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<S: MindStore, I: IndexSink<S>>(
|
||||
daemon: &mut Daemon<S, I>,
|
||||
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<OwnedRedbMessagePackBackingStore, NoIndex>) -> 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::<Vec<_>>().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:?}");
|
||||
}
|
||||
}
|
||||
}
|
||||
Loading…
Reference in a new issue