From 93ebc65c3338e329ccf17a98f9db4b2a97a19a40 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Mon, 15 Jun 2026 22:50:43 +0100 Subject: [PATCH] Add Idunn RUDP health ingress --- Cargo.toml | 1 + src/main.rs | 204 ++++++++++++++++++++++++++++++++++++++++++++++++++-- 2 files changed, 199 insertions(+), 6 deletions(-) diff --git a/Cargo.toml b/Cargo.toml index 7c7550c..2613edd 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -14,3 +14,4 @@ anyhow = "1" cultmesh-rs = { path = "../../vendor/cultmesh-rs" } cultnet-rs = { path = "../../vendor/cultnet-rs" } odin-core = { path = "../odin-core" } +rmp-serde = "1.3" diff --git a/src/main.rs b/src/main.rs index bc4d4ca..61c99eb 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1,17 +1,17 @@ use anyhow::{Context, Result, anyhow}; use cultmesh_rs::{CultMesh, CultMeshNode, CultMeshNodeOptions}; use cultnet_rs::{ - CULTNET_RUDP_PROTOCOL_ID, CultNetMessage, CultNetRudpOptions, recv_cultnet_message_rudp, - send_cultnet_message_rudp, + CULTNET_RUDP_PROTOCOL_ID, CultNetMessage, CultNetRawPayloadEncoding, CultNetRudpOptions, + recv_cultnet_message_rudp, send_cultnet_message_rudp, }; use odin_core::{ IdunnCommandBoundaryRecord, IdunnDaemonHealthRecord, IdunnDaemonSurgeryPlanRecord, IdunnDaemonTransportProfileRecord, IdunnDeploymentResultRecord, IdunnDesiredDaemonRecord, - IdunnOperatorAlarmRecord, IdunnRestartResultRecord, IdunnRuntimeTransportCheckRecord, - OdinDocuments, plan_keepalive, + IdunnOperatorAlarmRecord, IdunnRestartResultRecord, IdunnRudpHealthIngressRecord, + IdunnRuntimeTransportCheckRecord, OdinDocuments, plan_keepalive, }; use std::env; -use std::net::UdpSocket; +use std::net::{SocketAddr, UdpSocket}; use std::path::PathBuf; use std::process::{Child, Command, Stdio}; use std::sync::{Arc, Mutex}; @@ -49,6 +49,7 @@ fn health_contract(id: &str, default_failure_state: &str) -> HealthContract { struct CommonOptions { store_path: PathBuf, operator_alarm_command: Option, + rudp_health_bind: Option, execute: bool, command_timeout_seconds: u64, } @@ -110,6 +111,7 @@ fn run_swarm(options: &SwarmOptions, common: &CommonOptions) -> Result<()> { let store_lock = Arc::new(Mutex::new(())); let now = timestamp()?; publish_runtime_transport_check(common, &store_lock, &now)?; + start_rudp_health_ingress(common, &store_lock, &now)?; publish_daemon_surgery_plans(&targets, common, &store_lock, &now)?; let mut workers = Vec::with_capacity(targets.len()); @@ -399,6 +401,154 @@ fn verify_rudp_loopback() -> Result { Ok(transfer_id) } +fn start_rudp_health_ingress( + options: &CommonOptions, + store_lock: &Arc>, + observed_at: &str, +) -> Result<()> { + let Some(bind_address) = options.rudp_health_bind else { + return Ok(()); + }; + + let ingress_id = "idunn-rudp-health-ingress".to_string(); + let socket = match UdpSocket::bind(bind_address) { + Ok(socket) => socket, + Err(error) => { + let failed = rudp_health_ingress_record( + &ingress_id, + bind_address, + "failed", + &format!("failed to bind RUDP health ingress: {error}"), + observed_at, + ); + with_store_node(options, store_lock, |node| { + node.put(&failed.ingress_id, &failed)?; + Ok(()) + })?; + return Err(error.into()); + } + }; + let local_addr = socket.local_addr()?; + socket.set_read_timeout(Some(Duration::from_millis(500)))?; + + let active = rudp_health_ingress_record( + &ingress_id, + local_addr, + "active", + "listening for raw idunn.daemon_health documents over CultNet RUDP", + observed_at, + ); + with_store_node(options, store_lock, |node| { + node.put(&active.ingress_id, &active)?; + Ok(()) + })?; + + let worker_options = options.clone(); + let worker_store_lock = Arc::clone(store_lock); + thread::spawn(move || { + if let Err(error) = run_rudp_health_ingress_loop(socket, worker_options, worker_store_lock) + { + eprintln!("Idunn RUDP health ingress stopped: {error}"); + } + }); + Ok(()) +} + +fn rudp_health_ingress_record( + ingress_id: &str, + bind_address: SocketAddr, + state: &str, + detail: &str, + observed_at: &str, +) -> IdunnRudpHealthIngressRecord { + IdunnRudpHealthIngressRecord { + ingress_id: ingress_id.to_string(), + bind_address: bind_address.to_string(), + transport: CULTNET_RUDP_PROTOCOL_ID.to_string(), + accepted_schema: "idunn.daemon_health".to_string(), + state: state.to_string(), + detail: detail.to_string(), + observed_at: observed_at.to_string(), + } +} + +fn run_rudp_health_ingress_loop( + socket: UdpSocket, + options: CommonOptions, + store_lock: Arc>, +) -> Result<()> { + let rudp_options = CultNetRudpOptions { + ack_timeout: Duration::from_millis(100), + retries: 3, + ..CultNetRudpOptions::default() + }; + + loop { + match recv_cultnet_message_rudp(&socket, &rudp_options) { + Ok((message, source, transfer_id)) => { + let observed_at = timestamp()?; + match health_from_rudp_message(&message) { + Ok(health) => { + with_store_node(&options, &store_lock, |node| { + node.put(&health.daemon_id, &health)?; + Ok(()) + })?; + println!( + "Idunn accepted RUDP health for {} from {} transfer {}.", + health.daemon_id, source, transfer_id + ); + } + Err(error) => { + let failed = rudp_health_ingress_record( + "idunn-rudp-health-ingress", + socket.local_addr()?, + "degraded", + &format!("rejected RUDP health message from {source}: {error}"), + &observed_at, + ); + with_store_node(&options, &store_lock, |node| { + node.put(&failed.ingress_id, &failed)?; + Ok(()) + })?; + } + } + } + Err(error) + if error.downcast_ref::().is_some_and(|io| { + matches!( + io.kind(), + std::io::ErrorKind::WouldBlock | std::io::ErrorKind::TimedOut + ) + }) => {} + Err(error) => return Err(error), + } + } +} + +fn health_from_rudp_message(message: &CultNetMessage) -> Result { + let CultNetMessage::DocumentPutRaw { document, .. } = message else { + return Err(anyhow!("expected cultnet.document_put_raw.v0")); + }; + if document.schema_id != "idunn.daemon_health" { + return Err(anyhow!( + "expected idunn.daemon_health schema, received {}", + document.schema_id + )); + } + if document.payload_encoding != CultNetRawPayloadEncoding::Messagepack { + return Err(anyhow!("expected MessagePack raw payload encoding")); + } + let health: IdunnDaemonHealthRecord = rmp_serde::from_slice(&document.payload)?; + if document.record_key != health.daemon_id { + return Err(anyhow!( + "record key {} does not match health daemon_id {}", + document.record_key, + health.daemon_id + )); + } + Ok(health) +} + fn publish_daemon_surgery_plans( targets: &[DaemonTarget], options: &CommonOptions, @@ -917,6 +1067,7 @@ impl Options { fn parse(args: impl Iterator) -> Result { let mut store_path = PathBuf::from("scratch/idunn/idunn.keepalive.cc"); let mut operator_alarm_command = None; + let mut rudp_health_bind = Some("127.0.0.1:17870".parse::()?); let mut execute = false; let mut command_timeout_seconds = 30; let mut daemon_id = None; @@ -950,6 +1101,18 @@ impl Options { operator_alarm_command = Some(take_value(&mut args, "--operator-alarm-command")?) } + "--rudp-health-bind" => { + let value = take_value(&mut args, "--rudp-health-bind")?; + rudp_health_bind = if value.eq_ignore_ascii_case("none") { + None + } else { + Some( + value + .parse() + .with_context(|| "--rudp-health-bind must be a socket address")?, + ) + }; + } "--disabled" => enabled = false, "--execute" => execute = true, "--interval-seconds" => { @@ -987,6 +1150,7 @@ impl Options { let common = CommonOptions { store_path, operator_alarm_command, + rudp_health_bind, execute, command_timeout_seconds, }; @@ -1349,12 +1513,13 @@ fn timestamp() -> Result { } fn help_text() -> &'static str { - "Usage: idunn --daemon [--name ] [--verse ] [--store ] [--health-command ] [--deploy-command ] [--restart-command ] [--operator-alarm-command ] [--execute] [--interval-seconds ] [--command-timeout-seconds ] [--repo-root ] [--swarm-profile ]\n\nIdunn can run one manual daemon probe lane with --daemon, or one built-in swarm supervisor with --swarm-profile starfire-local." + "Usage: idunn --daemon [--name ] [--verse ] [--store ] [--health-command ] [--deploy-command ] [--restart-command ] [--operator-alarm-command ] [--rudp-health-bind ] [--execute] [--interval-seconds ] [--command-timeout-seconds ] [--repo-root ] [--swarm-profile ]\n\nIdunn can run one manual daemon probe lane with --daemon, or one built-in swarm supervisor with --swarm-profile starfire-local." } #[cfg(test)] mod tests { use super::*; + use cultnet_rs::{CultNetRawDocumentRecord, CultNetRawPayloadEncoding}; fn target(default_failure_state: &str, deploy_command: Option<&str>) -> DaemonTarget { DaemonTarget { @@ -1407,4 +1572,31 @@ mod tests { assert_eq!(check.transport, CULTNET_RUDP_PROTOCOL_ID); assert_eq!(check.state, "available"); } + + #[test] + fn rudp_health_ingress_accepts_raw_daemon_health_document() { + let health = IdunnDaemonHealthRecord { + daemon_id: "test-daemon".to_string(), + state: "active".to_string(), + detail: "published by daemon".to_string(), + observed_at: "2026-06-15T00:00:00Z".to_string(), + health_contract: "test.cultnet-rudp-health".to_string(), + }; + let message = CultNetMessage::DocumentPutRaw { + message_id: "test-health".to_string(), + document: CultNetRawDocumentRecord { + schema_id: "idunn.daemon_health".to_string(), + record_key: health.daemon_id.clone(), + stored_at: "2026-06-15T00:00:00Z".to_string(), + payload_encoding: CultNetRawPayloadEncoding::Messagepack, + payload: rmp_serde::to_vec(&health).unwrap(), + source_runtime_id: Some("test-daemon".to_string()), + source_agent_id: None, + source_role: Some("daemon-health-publisher".to_string()), + tags: Some(vec!["cultnet.transport.rudp.v0".to_string()]), + }, + }; + + assert_eq!(health_from_rudp_message(&message).unwrap(), health); + } }