From fcbed2aeab8eb1cf5f7ae4dcbae034d94cb553aa Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Tue, 23 Jun 2026 22:45:39 +0100 Subject: [PATCH] Publish Muninn startup respect to Odin --- crates/muninn-daemon/src/main.rs | 63 ++++++++- scripts/restart-muninn.ps1 | 4 + scripts/restart-nightwing-muninn.ps1 | 8 +- scripts/restart-starfire-muninn.ps1 | 6 +- vendor/cultmesh-rs/src/lib.rs | 204 +++++++++++++++++++++++++++ 5 files changed, 281 insertions(+), 4 deletions(-) diff --git a/crates/muninn-daemon/src/main.rs b/crates/muninn-daemon/src/main.rs index 1323885..e0d30d1 100644 --- a/crates/muninn-daemon/src/main.rs +++ b/crates/muninn-daemon/src/main.rs @@ -8,7 +8,8 @@ use crate::media_packetizer::{ use anyhow::{Context, Result, anyhow}; use cultmesh_rs::{ CultMesh, CultMeshNodeOptions, CultMeshSharedMemoryFrameRing, CultMeshStreamBodyTransport, - CultMeshStreamCatalog, CultMeshStreamClock, CultMeshStreamDescriptor, CultMeshStreamKind, + CultMeshRudpDocumentPublishOptions, CultMeshStreamCatalog, CultMeshStreamClock, + CultMeshStreamDescriptor, CultMeshStreamKind, }; use cultnet_rs::{ CultNetMessage, CultNetRawDocumentRecord, CultNetRawPayloadEncoding, @@ -156,6 +157,7 @@ struct Options { quest_pose_stream_id: Option, quest_video_input_stream_id: Option, idunn_rudp_health: Option, + odin_cultmesh_rudp: Option, capture_command_rudp_bind: Option, capture_command_rudp_target: Option, obs_catalog_rudp_target: Option, @@ -327,6 +329,7 @@ fn serve(options: Options) -> Result<()> { let mut active_move_lights = Vec::new(); let mut last_default_move_light_write_at = None; let mut last_idunn_health_publish_attempt_at = None; + let mut odin_respects_paid = false; let mut last_move_host_claim_attempt_at = None; let mut last_move_bluetooth_pickup_attempt_at = None; let move_runtime_enabled = serve_should_manage_move_runtime(&options); @@ -372,6 +375,14 @@ fn serve(options: Options) -> Result<()> { }; publish_surface(&mut node, &options, state, &active_stream_ids)?; publish_runtime_boundary_records(&mut node, &options, state, &active_stream_ids)?; + if !odin_respects_paid { + odin_respects_paid = true; + if let Err(error) = publish_odin_startup_respect(&node, &options) { + eprintln!( + "Muninn could not publish startup respect to Odin: {error:#}" + ); + } + } publish_obs_catalog_idle(&mut node, &options)?; } claim_move_host_if_due(&options, &mut last_move_host_claim_attempt_at); @@ -2857,6 +2868,31 @@ fn publish_runtime_boundary_records( Ok(()) } +fn publish_odin_startup_respect( + node: &cultmesh_rs::CultMeshNode, + options: &Options, +) -> Result<()> { + let Some(target) = options.odin_cultmesh_rudp else { + return Ok(()); + }; + let provider_id = muninn_provider_id(options); + let provider = node.get_required::(&provider_id)?; + node.publish_document_to_rudp_catalog( + &provider_id, + &provider, + CultMeshRudpDocumentPublishOptions { + target, + runtime_id: muninn_daemon_id(options), + source_role: Some("muninn.telemetry-provider".to_string()), + tags: vec![ + "startup-respect".to_string(), + "odin-verse-discovery".to_string(), + ], + ..CultMeshRudpDocumentPublishOptions::default() + }, + ) +} + fn muninn_provider_id(options: &Options) -> String { format!("muninn.telemetry.{}", options.host_id) } @@ -6592,6 +6628,7 @@ impl Options { quest_pose_stream_id: None, quest_video_input_stream_id: None, idunn_rudp_health: None, + odin_cultmesh_rudp: None, capture_command_rudp_bind: None, capture_command_rudp_target: None, obs_catalog_rudp_target: None, @@ -6773,6 +6810,13 @@ impl Options { "--idunn-health-contract" => { idunn_health_contract = Some(take_value(&mut args, "--idunn-health-contract")?) } + "--odin-cultmesh-rudp" => { + options.odin_cultmesh_rudp = Some( + take_value(&mut args, "--odin-cultmesh-rudp")? + .parse() + .context("--odin-cultmesh-rudp must be a socket address")?, + ) + } "--capture-command-rudp-bind" => { options.capture_command_rudp_bind = Some( take_value(&mut args, "--capture-command-rudp-bind")? @@ -6962,7 +7006,7 @@ fn parse_catalog_source(value: &str) -> Result { } fn help_text() -> &'static str { - "Usage: muninn [serve|activate|request-stream|capture-stream-status|request-move-light|move-light-status|move-identity-status|move-source-status|move-state-status|claim-move-host|quest-access-status] [--store ] [--activate-store ] [--stream-action ] [--target-host ] [--port ] [--obs-target-host ] [--obs-port ] [--media-transport ] [--media-packet-bytes ] [--rudp-video-bitrate-kbps ] [--rudp-latency-budget-ms ] [--video-source ] [--audio-source ] [--audio-source-id ] [--no-video] [--no-audio] [--loopback-script ] [--ffmpeg ] [--capture-command-rudp-bind ] [--capture-command-rudp-target ] [--obs-catalog-rudp-target ] [--move-state =] [--move-host ] [--move-evidence-stream ] [--move-evidence-verse ] [--quest-adb] [--quest-serial ] [--quest-input-stream ] [--quest-pose-stream ] [--quest-video-input-stream ] [--idunn-rudp-health ] [--idunn-daemon ] [--idunn-health-contract ] [--dry-run] [--health]\n\nMuninn is Odin's portable telemetry Verse assembler. serve publishes cheap typed telemetry affordances, optional Quest USB access surfaces, and the explicitly configured Move runtime; when serve receives --move-state, --move-host, or --move-evidence-stream it may publish source-local Move controller state, typed Move identity records, a CultMesh Move evidence stream, and keep USB-attached PS Moves claimed to that explicit Bluetooth host; serve also consumes typed capture stream commands from --activate-store or --capture-command-rudp-bind and owns the local ffmpeg/loopback activation child lifecycle, and may project the OBS stream catalog to a local OBS plugin over --obs-catalog-rudp-target; activate starts an explicitly requested local stream over SRT or CultNet RUDP as a daemon child; request-stream publishes a typed capture stream command for Muninn serve to execute, either into the activation store or over --capture-command-rudp-target; capture-stream-status reads typed capture stream command receipts; use --no-video or --no-audio to request one leg without the other when the transport is CultNet RUDP; request-move-light publishes a typed Move light command for Muninn serve to execute; move-light-status reads typed command receipts; move-identity-status reads typed Move identity records; move-source-status prints live Move source discovery; move-state-status reads typed controller-state records; claim-move-host assigns USB-attached PS Moves to a Bluetooth host; quest-access-status reads typed Quest access state. In --health mode, the Idunn RUDP flags publish typed daemon health to Idunn while preserving command-probe exit semantics." + "Usage: muninn [serve|activate|request-stream|capture-stream-status|request-move-light|move-light-status|move-identity-status|move-source-status|move-state-status|claim-move-host|quest-access-status] [--store ] [--activate-store ] [--stream-action ] [--target-host ] [--port ] [--obs-target-host ] [--obs-port ] [--media-transport ] [--media-packet-bytes ] [--rudp-video-bitrate-kbps ] [--rudp-latency-budget-ms ] [--video-source ] [--audio-source ] [--audio-source-id ] [--no-video] [--no-audio] [--loopback-script ] [--ffmpeg ] [--capture-command-rudp-bind ] [--capture-command-rudp-target ] [--obs-catalog-rudp-target ] [--odin-cultmesh-rudp ] [--move-state =] [--move-host ] [--move-evidence-stream ] [--move-evidence-verse ] [--quest-adb] [--quest-serial ] [--quest-input-stream ] [--quest-pose-stream ] [--quest-video-input-stream ] [--idunn-rudp-health ] [--idunn-daemon ] [--idunn-health-contract ] [--dry-run] [--health]\n\nMuninn is Odin's portable telemetry Verse assembler. serve publishes cheap typed telemetry affordances, optional Quest USB access surfaces, and the explicitly configured Move runtime; when serve receives --move-state, --move-host, or --move-evidence-stream it may publish source-local Move controller state, typed Move identity records, a CultMesh Move evidence stream, and keep USB-attached PS Moves claimed to that explicit Bluetooth host; serve also consumes typed capture stream commands from --activate-store or --capture-command-rudp-bind and owns the local ffmpeg/loopback activation child lifecycle, may project the OBS stream catalog to a local OBS plugin over --obs-catalog-rudp-target, and pays startup respect to Odin over --odin-cultmesh-rudp by publishing its provider advertisement once; activate starts an explicitly requested local stream over SRT or CultNet RUDP as a daemon child; request-stream publishes a typed capture stream command for Muninn serve to execute, either into the activation store or over --capture-command-rudp-target; capture-stream-status reads typed capture stream command receipts; use --no-video or --no-audio to request one leg without the other when the transport is CultNet RUDP; request-move-light publishes a typed Move light command for Muninn serve to execute; move-light-status reads typed command receipts; move-identity-status reads typed Move identity records; move-source-status prints live Move source discovery; move-state-status reads typed controller-state records; claim-move-host assigns USB-attached PS Moves to a Bluetooth host; quest-access-status reads typed Quest access state. In --health mode, the Idunn RUDP flags publish typed daemon health to Idunn while preserving command-probe exit semantics." } fn parse_move_state_source(value: &str) -> Result { @@ -7293,6 +7337,21 @@ mod tests { assert!(error.contains("must be provided together")); } + #[test] + fn serve_accepts_odin_cultmesh_rudp_startup_target() { + let options = Options::parse( + ["serve", "--odin-cultmesh-rudp", "10.77.0.2:17871"] + .into_iter() + .map(String::from), + ) + .unwrap(); + + assert_eq!( + options.odin_cultmesh_rudp, + Some("10.77.0.2:17871".parse().unwrap()) + ); + } + #[test] fn idunn_rudp_health_publisher_sends_raw_daemon_health_document() { let receiver = UdpSocket::bind("127.0.0.1:0").unwrap(); diff --git a/scripts/restart-muninn.ps1 b/scripts/restart-muninn.ps1 index 79ba4eb..e1f28cc 100644 --- a/scripts/restart-muninn.ps1 +++ b/scripts/restart-muninn.ps1 @@ -18,6 +18,7 @@ param( [string] $IdunnRudpHealth = "10.77.0.2:17870", [string] $IdunnDaemon = "muninn", [string] $IdunnHealthContract = "muninn.cultnet-rudp-remote-telemetry-health", + [string] $OdinCultMeshRudp = "10.77.0.2:17871", [string] $CaptureCommandRudpBind = "0.0.0.0:17883", [string] $CaptureCommandRudpTarget = "127.0.0.1:17883", [string] $ObsCatalogRudpTarget = "10.77.0.2:17874", @@ -373,6 +374,9 @@ $serveArguments = @( "--port", $Port.ToString(), "--media-transport", $MediaTransport ) +if (-not [string]::IsNullOrWhiteSpace($OdinCultMeshRudp)) { + $serveArguments += @("--odin-cultmesh-rudp", $OdinCultMeshRudp) +} foreach ($videoSource in $VideoSources) { $serveArguments += @("--video-source", $videoSource) } diff --git a/scripts/restart-nightwing-muninn.ps1 b/scripts/restart-nightwing-muninn.ps1 index dc2d687..5105225 100644 --- a/scripts/restart-nightwing-muninn.ps1 +++ b/scripts/restart-nightwing-muninn.ps1 @@ -10,7 +10,8 @@ param( [int] $IntervalSeconds = 15, [string] $IdunnRudpHealth = "10.77.0.2:17870", [string] $IdunnDaemon = "nightwing-muninn", - [string] $IdunnHealthContract = "muninn.cultnet-rudp-remote-telemetry-and-move-hid" + [string] $IdunnHealthContract = "muninn.cultnet-rudp-remote-telemetry-and-move-hid", + [string] $OdinCultMeshRudp = "10.77.0.2:17871" ) $ErrorActionPreference = "Stop" @@ -84,6 +85,10 @@ if ($moveStateSpecs.Count -gt 0) { $moveRuntimeSetLines += "set -- ""`$@"" --move-evidence-stream $(Quote-ShSingle $MoveEvidenceStream)" } $moveRuntimeSetBlock = ($moveRuntimeSetLines -join "`n") +$odinCultMeshRudpSetBlock = "" +if (-not [string]::IsNullOrWhiteSpace($OdinCultMeshRudp)) { + $odinCultMeshRudpSetBlock = "set -- ""`$@"" --odin-cultmesh-rudp $(Quote-ShSingle $OdinCultMeshRudp)" +} $remoteScript = @" set -eu @@ -109,6 +114,7 @@ set -- "`$@" \ --idunn-rudp-health '$IdunnRudpHealth' \ --idunn-daemon '$IdunnDaemon' \ --idunn-health-contract '$IdunnHealthContract' +$odinCultMeshRudpSetBlock nohup '$MuninnExe' "`$@" \ > '$LogRoot/muninn-serve.out.log' \ 2> '$LogRoot/muninn-serve.err.log' \ diff --git a/scripts/restart-starfire-muninn.ps1 b/scripts/restart-starfire-muninn.ps1 index 30eb0fb..ce80332 100644 --- a/scripts/restart-starfire-muninn.ps1 +++ b/scripts/restart-starfire-muninn.ps1 @@ -8,7 +8,8 @@ param( [switch] $EnableUsbMoveState, [string] $IdunnRudpHealth = "127.0.0.1:17870", [string] $IdunnDaemon = "starfire-muninn", - [string] $IdunnHealthContract = "muninn.cultnet-rudp-local-telemetry-and-quest-access" + [string] $IdunnHealthContract = "muninn.cultnet-rudp-local-telemetry-and-quest-access", + [string] $OdinCultMeshRudp = "127.0.0.1:17871" ) $ErrorActionPreference = "Stop" @@ -49,6 +50,9 @@ $arguments = @( "--idunn-daemon", $IdunnDaemon, "--idunn-health-contract", $IdunnHealthContract ) +if (-not [string]::IsNullOrWhiteSpace($OdinCultMeshRudp)) { + $arguments += @("--odin-cultmesh-rudp", $OdinCultMeshRudp) +} if (-not [string]::IsNullOrWhiteSpace($MoveBluetoothHost)) { $arguments += @("--move-host", $MoveBluetoothHost) } diff --git a/vendor/cultmesh-rs/src/lib.rs b/vendor/cultmesh-rs/src/lib.rs index 4e37442..a4979ef 100644 --- a/vendor/cultmesh-rs/src/lib.rs +++ b/vendor/cultmesh-rs/src/lib.rs @@ -2,9 +2,22 @@ use anyhow::Result; use cultcache_rs::CultCache; use cultcache_rs::DatabaseEntry; use cultcache_rs::SingleFileMessagePackBackingStore; +use cultnet_rs::CultNetDocumentPutOptions; use cultnet_rs::CultNetDocumentRegistry; +use cultnet_rs::CultNetRudpSocketTransportConnection; +use cultnet_rs::CultNetRudpSocketTransportOptions; +use cultnet_rs::CultNetWireContract; +use cultnet_rs::encode_cultnet_message_to_vec; +use serde::Serialize; use std::collections::BTreeMap; +use std::net::SocketAddr; +use std::net::UdpSocket; use std::path::Path; +use std::thread; +use std::time::Duration; +use std::time::Instant; + +pub const CULTMESH_RUDP_DOCUMENT_CATALOG_CONNECTION_ID: u32 = 0x0d1d_0002; pub trait CultMeshDocumentSet: Clone + Send + Sync + 'static { fn register_cache(&self, cache: &mut CultCache) -> Result<()>; @@ -52,6 +65,47 @@ pub struct CultMeshNodeOptions { pub pull_on_start: bool, } +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct CultMeshRudpDocumentPublishOptions { + pub target: SocketAddr, + pub runtime_id: String, + pub connection_id: u32, + pub connect_timeout: Duration, + pub flush_timeout: Duration, + pub poll_interval: Duration, + pub resend_delay_ms: u64, + pub source_agent_id: Option, + pub source_role: Option, + pub tags: Vec, +} + +impl CultMeshRudpDocumentPublishOptions { + pub fn odin(target: SocketAddr, runtime_id: impl Into) -> Self { + Self { + target, + runtime_id: runtime_id.into(), + ..Self::default() + } + } +} + +impl Default for CultMeshRudpDocumentPublishOptions { + fn default() -> Self { + Self { + target: SocketAddr::from(([127, 0, 0, 1], 17871)), + runtime_id: "cultmesh-rudp-document-publisher".to_string(), + connection_id: CULTMESH_RUDP_DOCUMENT_CATALOG_CONNECTION_ID, + connect_timeout: Duration::from_secs(1), + flush_timeout: Duration::from_millis(150), + poll_interval: Duration::from_millis(5), + resend_delay_ms: 100, + source_agent_id: None, + source_role: None, + tags: Vec::new(), + } + } +} + impl Default for CultMeshNodeOptions { fn default() -> Self { Self { @@ -532,6 +586,91 @@ impl CultMeshNode { pub fn flush(&mut self) -> Result<()> { Ok(()) } + + pub fn publish_document_to_rudp_catalog( + &self, + key: impl Into, + value: &T, + options: CultMeshRudpDocumentPublishOptions, + ) -> Result<()> + where + T: DatabaseEntry + Serialize, + { + let key = key.into(); + let message = self.documents.create_raw_document_put_message( + format!("{}:{}:{}", options.runtime_id, T::TYPE, key), + key, + value, + CultNetDocumentPutOptions { + source_runtime_id: Some(options.runtime_id.clone()), + source_agent_id: options.source_agent_id.clone(), + source_role: options.source_role.clone(), + tags: if options.tags.is_empty() { + None + } else { + Some(options.tags.clone()) + }, + ..CultNetDocumentPutOptions::default() + }, + )?; + publish_cultnet_message_to_rudp_catalog(&message, options) + } +} + +fn publish_cultnet_message_to_rudp_catalog( + message: &cultnet_rs::CultNetMessage, + options: CultMeshRudpDocumentPublishOptions, +) -> Result<()> { + if options.runtime_id.trim().is_empty() { + anyhow::bail!("runtime_id must be non-empty"); + } + let bind_addr = if options.target.is_ipv4() { + SocketAddr::from(([0, 0, 0, 0], 0)) + } else { + SocketAddr::from(([0, 0, 0, 0, 0, 0, 0, 0], 0)) + }; + let socket = UdpSocket::bind(bind_addr)?; + socket.set_read_timeout(Some(options.poll_interval))?; + let mut client = CultNetRudpSocketTransportConnection::new( + CultNetRudpSocketTransportOptions { + runtime_id: options.runtime_id.clone(), + socket, + mode: cultnet_rs::CultNetRudpSocketMode::Client, + remote_addr: Some(options.target), + connection_id: options.connection_id, + initial_sequence: 1, + resend_delay_ms: options.resend_delay_ms, + transport_id: Some("cultmesh-rudp-document-publisher".to_string()), + max_payload_bytes: None, + max_fragment_bytes: Some(1200), + max_pending_reliable_packets: Some(64), + media_reliable_expire_after_ms: None, + }, + )?; + client.connect(Vec::new())?; + let connect_deadline = Instant::now() + options.connect_timeout; + while !client.connected() && Instant::now() < connect_deadline { + let _ = client.receive_once()?; + client.poll_resends()?; + thread::sleep(options.poll_interval); + } + if !client.connected() { + anyhow::bail!( + "Timed out connecting CultMesh RUDP document publisher {} to {}", + options.runtime_id, + options.target + ); + } + + let payload = encode_cultnet_message_to_vec(message, CultNetWireContract::CultNetSchemaV0)?; + client.send("schema", payload)?; + let flush_deadline = Instant::now() + options.flush_timeout; + while Instant::now() < flush_deadline { + let _ = client.receive_once()?; + client.poll_resends()?; + thread::sleep(options.poll_interval); + } + Ok(()) } #[derive(Clone, Copy, Debug, Default)] @@ -653,6 +792,71 @@ mod tests { Ok(()) } + #[test] + fn node_publishes_registered_document_to_rudp_catalog() -> Result<()> { + let temp = tempfile::tempdir()?; + let store_path = temp.path().join("cultmesh.cc"); + let note = Note { + body: "respect odin once".to_string(), + }; + let node = CultMesh::create_node( + &store_path, + TestDocuments, + CultMeshNodeOptions { + runtime_id: "muninn-test".to_string(), + ..CultMeshNodeOptions::default() + }, + )?; + + let socket = UdpSocket::bind("127.0.0.1:0")?; + socket.set_read_timeout(Some(Duration::from_millis(10)))?; + let target = socket.local_addr()?; + let (sender, receiver) = std::sync::mpsc::channel(); + let server = thread::spawn(move || -> Result<()> { + let mut server = CultNetRudpSocketTransportConnection::new( + CultNetRudpSocketTransportOptions::server( + "odin-test-catalog", + socket, + CULTMESH_RUDP_DOCUMENT_CATALOG_CONNECTION_ID, + ), + )?; + let deadline = Instant::now() + Duration::from_secs(2); + while Instant::now() < deadline { + if let Some(frame) = server.receive_once()? { + let message = cultnet_rs::decode_cultnet_message_from_slice( + &frame.payload, + CultNetWireContract::CultNetSchemaV0, + )?; + if let cultnet_rs::CultNetMessage::DocumentPutRaw { document, .. } = message { + sender + .send(( + document.schema_id, + document.record_key, + document.source_runtime_id, + )) + .ok(); + return Ok(()); + } + } + server.poll_resends()?; + } + anyhow::bail!("timed out waiting for RUDP document put") + }); + + node.publish_document_to_rudp_catalog( + "note", + ¬e, + CultMeshRudpDocumentPublishOptions::odin(target, "muninn-test"), + )?; + let received = receiver.recv_timeout(Duration::from_secs(2))?; + server.join().expect("server thread should not panic")?; + + assert_eq!(received.0, "cultmesh.test.note.v0"); + assert_eq!(received.1, "note"); + assert_eq!(received.2.as_deref(), Some("muninn-test")); + Ok(()) + } + #[test] fn stream_catalog_negotiates_and_publishes_latest_shared_memory_frame() -> Result<()> { let mut catalog = CultMesh::create_stream_catalog();