From 7bdcb866cbef5a0c2740b9e9ff83c48abd326242 Mon Sep 17 00:00:00 2001 From: Randy Treit Date: Sun, 26 Jul 2026 11:28:57 -0700 Subject: [PATCH] Publish dynamic WinKeyer runtime endpoint --- Cargo.lock | 2 +- README.md | 12 +++ crates/cathub/Cargo.toml | 2 +- crates/cathub/src/hamlib_net.rs | 11 ++- crates/cathub/src/lib.rs | 54 ++++++++++++-- crates/cathub/src/runtime_info.rs | 116 +++++++++++++++++++++++++++++ crates/cathub/src/winkeyer/grpc.rs | 41 +++++----- crates/cathub/tests/cli.rs | 6 ++ docs/integration/setup.md | 8 +- 9 files changed, 217 insertions(+), 35 deletions(-) create mode 100644 crates/cathub/src/runtime_info.rs diff --git a/Cargo.lock b/Cargo.lock index d8b812f..83b2d80 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -179,7 +179,7 @@ checksum = "fc652a48c352aef3ea3aed32080501cf3ef6ed5da78602a020c991775b0aff04" [[package]] name = "cathub" -version = "0.1.1" +version = "0.1.2" dependencies = [ "async-trait", "bytes", diff --git a/README.md b/README.md index b1bafda..fe3fc87 100644 --- a/README.md +++ b/README.md @@ -114,6 +114,16 @@ cathub config validate cathub ``` +An orchestrator can request an available typed API port: + +```powershell +cathub --winkeyer-api-bind 127.0.0.1:0 --runtime-info .\cathub-runtime.json +``` + +CatHub publishes the selected endpoint after all configured listeners bind. +The runtime file includes the CatHub process ID. +Clients must validate that process ID before they use the endpoint. + ## Client interfaces - Hamlib-aware clients connect to a configured `[[hamlib_net]]` TCP listener. @@ -121,6 +131,8 @@ cathub - Legacy WinKeyer clients connect to their own virtual serial pair. - Typed WinKeyer clients connect to the loopback gRPC address in `[winkeyer].api_bind`. +A launcher-managed client uses the endpoint in CatHub's runtime file. + Each endpoint has its own permissions. Applications never open the physical radio or keyer port directly. diff --git a/crates/cathub/Cargo.toml b/crates/cathub/Cargo.toml index 7eb506f..a18b9d2 100644 --- a/crates/cathub/Cargo.toml +++ b/crates/cathub/Cargo.toml @@ -1,7 +1,7 @@ [package] name = "cathub" description = "Multi-client CAT and WinKeyer hub daemon" -version.workspace = true +version = "0.1.2" edition.workspace = true rust-version.workspace = true license.workspace = true diff --git a/crates/cathub/src/hamlib_net.rs b/crates/cathub/src/hamlib_net.rs index 0efd0dc..e95803d 100644 --- a/crates/cathub/src/hamlib_net.rs +++ b/crates/cathub/src/hamlib_net.rs @@ -659,16 +659,19 @@ where /// closed. Real rigctld commands are short; this only bounds a misbehaving client. const MAX_LINE_LEN: usize = 4096; -/// Bind a Hamlib net endpoint and serve connections. Each connection gets a fresh +/// Bind a Hamlib net endpoint before the daemon reports that it is ready. +pub(crate) async fn bind_listener(bind: &str) -> std::io::Result { + TcpListener::bind(bind).await +} + +/// Serve connections from a bound Hamlib net endpoint. Each connection gets a fresh /// [`ClientSessionContext`] sharing the same state/radio/ptt but its own session id and the /// endpoint's permissions. pub(crate) async fn run_listener( - bind: &str, + listener: TcpListener, next_session_id: Arc, template: ClientSessionContext, ) -> std::io::Result<()> { - let listener = TcpListener::bind(bind).await?; - tracing::info!(bind, "hamlib_net endpoint listening"); loop { let (stream, peer) = listener.accept().await?; let session_id = next_session_id.fetch_add(1, std::sync::atomic::Ordering::SeqCst); diff --git a/crates/cathub/src/lib.rs b/crates/cathub/src/lib.rs index 2f441b3..84bc5e3 100644 --- a/crates/cathub/src/lib.rs +++ b/crates/cathub/src/lib.rs @@ -29,6 +29,7 @@ mod model; mod permissions; mod ptt; mod radio; +mod runtime_info; mod serial_endpoint; mod state; mod winkeyer; @@ -36,6 +37,7 @@ mod winkeyer; #[cfg(test)] mod integration; +use std::net::SocketAddr; use std::path::PathBuf; use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::Arc; @@ -55,13 +57,14 @@ use crate::dialect::kenwood::ts2000::Ts2000Dialect; use crate::dialect::kenwood::ts590::Ts590Dialect; use crate::dialect::{ClientDialect, ClientSessionContext}; use crate::events::{spawn_poller, POLLER_SESSION}; -use crate::hamlib_net::run_listener; +use crate::hamlib_net::{bind_listener as bind_hamlib_listener, run_listener}; use crate::model::StateMutation; use crate::ptt::PttManager; use crate::radio::{ link_channel, run_transport_supervised, spawn_scheduler, OpKind, Priority, RECONNECT_INITIAL, RECONNECT_MAX, }; +use crate::runtime_info::{HamlibEndpoint, RuntimeInfo}; use crate::serial_endpoint::{open_serial, run_endpoint_session}; use crate::state::StateHandle; use crate::winkeyer::{ @@ -101,6 +104,12 @@ pub struct Cli { /// Explicit top-level section to read from a managed configuration file. #[arg(long, global = true)] pub section: Option, + /// Runtime override for the typed WinKeyer API bind address. + #[arg(long, global = true, value_name = "HOST:PORT")] + pub winkeyer_api_bind: Option, + /// Write effective endpoints here after all configured listeners bind. + #[arg(long, global = true, value_name = "FILE")] + pub runtime_info: Option, /// Optional explicit log file path (informational; logging also writes a rolling file). #[arg(long)] pub log: Option, @@ -250,7 +259,16 @@ pub async fn run(cli: Cli) -> Result<(), CatHubError> { if let Some(log) = &cli.log { tracing::debug!(log = %log.display(), "log path override requested"); } - let cfg = Config::load_selected(&path, cli.section.as_deref())?; + let mut cfg = Config::load_selected(&path, cli.section.as_deref())?; + + if let Some(bind) = cli.winkeyer_api_bind { + let Some(winkeyer) = cfg.winkeyer.as_mut() else { + return Err(CatHubError::Backend( + "--winkeyer-api-bind requires a configured [winkeyer] section".to_string(), + )); + }; + winkeyer.api_bind = bind.to_string(); + } if cli.dry_run { println!("{}", cfg.describe()); @@ -264,6 +282,7 @@ pub async fn run(cli: Cli) -> Result<(), CatHubError> { let state = StateHandle::new(); let ptt = PttManager::new(cfg.ptt_max_tx()); + let mut winkeyer_endpoint = None; let winkeyer: Option = if let Some(keyer) = &cfg.winkeyer { let port = open_winkeyer_serial(&keyer.port, keyer.baud)?; let port_name = keyer.port.clone(); @@ -287,9 +306,11 @@ pub async fn run(cli: Cli) -> Result<(), CatHubError> { let bind = keyer.api_bind.parse().map_err(|error| { CatHubError::Backend(format!("invalid WinKeyer API bind address: {error}")) })?; - let server = bind_winkeyer_server(bind, handle.clone()) + let (address, server) = bind_winkeyer_server(bind, handle.clone()) .await .map_err(|error| CatHubError::Backend(format!("cannot bind WinKeyer API: {error}")))?; + winkeyer_endpoint = Some(format!("http://{address}")); + tracing::info!(bind = %address, "WinKeyer broker API listening"); tokio::spawn(async move { match server.await { Ok(Err(error)) => tracing::error!(%error, "WinKeyer broker gRPC server stopped"), @@ -443,6 +464,7 @@ pub async fn run(cli: Cli) -> Result<(), CatHubError> { ); } + let mut hamlib_endpoints = Vec::with_capacity(cfg.hamlib_net.len()); for ep in &cfg.hamlib_net { let id = next_id.fetch_add(1, Ordering::SeqCst); let template = ClientSessionContext::new( @@ -454,17 +476,31 @@ pub async fn run(cli: Cli) -> Result<(), CatHubError> { caps.clone(), ) .with_single_vfo(ep.single_vfo); - let bind = ep.bind.clone(); + let listener = bind_hamlib_listener(&ep.bind).await?; + let address = listener.local_addr()?; let name = ep.name.clone(); let ids = next_id.clone(); tokio::spawn(async move { - if let Err(e) = run_listener(&bind, ids, template).await { + if let Err(e) = run_listener(listener, ids, template).await { tracing::error!(endpoint = %name, error = %e, "hamlib_net listener stopped"); } }); - tracing::info!(endpoint = %ep.name, bind = %ep.bind, "hamlib_net endpoint listening"); + tracing::info!(endpoint = %ep.name, bind = %address, "hamlib_net endpoint listening"); + hamlib_endpoints.push(HamlibEndpoint { + name: ep.name.clone(), + endpoint: address.to_string(), + }); } + let _runtime_info_lease = if let Some(path) = cli.runtime_info.as_deref() { + Some(runtime_info::publish( + path, + &RuntimeInfo::new(winkeyer_endpoint, hamlib_endpoints), + )?) + } else { + None + }; + // PTT safety watchdog: a transmitter that exceeds the configured ceiling is unkeyed at // the radio first, then released, so the ceiling is a real stuck-transmitter backstop // and the lease is never freed while the radio is still keyed. @@ -659,6 +695,8 @@ mod tests { let cli = Cli { config: Some(path.clone()), section: None, + winkeyer_api_bind: None, + runtime_info: None, log: None, dry_run: true, command: None, @@ -672,6 +710,8 @@ mod tests { let cli = Cli { config: Some(PathBuf::from("does-not-exist-cathub.toml")), section: None, + winkeyer_api_bind: None, + runtime_info: None, log: None, dry_run: true, command: None, @@ -706,6 +746,8 @@ mod tests { let cli = Cli { config: Some(path.clone()), section: None, + winkeyer_api_bind: None, + runtime_info: None, log: None, dry_run: false, command: None, diff --git a/crates/cathub/src/runtime_info.rs b/crates/cathub/src/runtime_info.rs new file mode 100644 index 0000000..6ef6e23 --- /dev/null +++ b/crates/cathub/src/runtime_info.rs @@ -0,0 +1,116 @@ +//! Runtime endpoint publication for launcher-managed CatHub processes. + +use std::fs; +use std::path::{Path, PathBuf}; + +use serde::{Deserialize, Serialize}; + +use crate::CatHubError; + +const SCHEMA_VERSION: u32 = 1; + +/// One effective Hamlib network endpoint. +#[derive(Debug, Clone, Serialize)] +pub(crate) struct HamlibEndpoint { + pub(crate) name: String, + pub(crate) endpoint: String, +} + +/// Effective endpoints for one ready CatHub process. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub(crate) struct RuntimeInfo { + pub(crate) schema_version: u32, + pub(crate) pid: u32, + pub(crate) winkeyer_endpoint: Option, + #[serde(default, skip_deserializing)] + pub(crate) hamlib_endpoints: Vec, +} + +impl RuntimeInfo { + pub(crate) fn new( + winkeyer_endpoint: Option, + hamlib_endpoints: Vec, + ) -> Self { + Self { + schema_version: SCHEMA_VERSION, + pid: std::process::id(), + winkeyer_endpoint, + hamlib_endpoints, + } + } +} + +/// Removes a runtime file when the publishing process exits normally. +pub(crate) struct RuntimeInfoLease { + path: PathBuf, + pid: u32, +} + +impl Drop for RuntimeInfoLease { + fn drop(&mut self) { + let belongs_to_process = fs::read(&self.path) + .ok() + .and_then(|bytes| serde_json::from_slice::(&bytes).ok()) + .is_some_and(|info| info.pid == self.pid); + if belongs_to_process { + let _ = fs::remove_file(&self.path); + } + } +} + +/// Publish effective endpoints after all configured listeners bind. +pub(crate) fn publish(path: &Path, info: &RuntimeInfo) -> Result { + if let Some(parent) = path.parent() { + fs::create_dir_all(parent)?; + } + let bytes = serde_json::to_vec_pretty(info) + .map_err(|error| CatHubError::Backend(format!("cannot serialize runtime info: {error}")))?; + let temporary = path.with_extension(format!("tmp-{}", info.pid)); + fs::write(&temporary, bytes)?; + if path.exists() { + fs::remove_file(path)?; + } + fs::rename(&temporary, path)?; + tracing::info!(path = %path.display(), "published runtime endpoint information"); + Ok(RuntimeInfoLease { + path: path.to_path_buf(), + pid: info.pid, + }) +} + +#[cfg(test)] +#[allow(clippy::expect_used, clippy::indexing_slicing, clippy::unwrap_used)] +mod tests { + use super::*; + + #[test] + fn publishes_and_removes_runtime_info() { + let directory = + std::env::temp_dir().join(format!("cathub-runtime-info-test-{}", std::process::id())); + fs::create_dir_all(&directory).expect("create test directory"); + let path = directory.join("runtime.json"); + let info = RuntimeInfo::new( + Some("http://127.0.0.1:54321".to_string()), + vec![HamlibEndpoint { + name: "engine".to_string(), + endpoint: "127.0.0.1:4532".to_string(), + }], + ); + + let lease = publish(&path, &info).expect("publish runtime info"); + let published: serde_json::Value = + serde_json::from_slice(&fs::read(&path).expect("read runtime info")) + .expect("parse runtime info"); + assert_eq!(published["schema_version"], 1); + assert_eq!(published["pid"], std::process::id()); + assert_eq!(published["winkeyer_endpoint"], "http://127.0.0.1:54321"); + assert_eq!( + published["hamlib_endpoints"][0]["endpoint"], + "127.0.0.1:4532" + ); + + drop(lease); + assert!(!path.exists()); + let _ = fs::remove_dir_all(directory); + } +} diff --git a/crates/cathub/src/winkeyer/grpc.rs b/crates/cathub/src/winkeyer/grpc.rs index 89756ac..b85c1dc 100644 --- a/crates/cathub/src/winkeyer/grpc.rs +++ b/crates/cathub/src/winkeyer/grpc.rs @@ -187,29 +187,25 @@ impl WinkeyerBrokerService for Service { } } -#[cfg(test)] -pub(crate) async fn run_server( - bind: SocketAddr, - broker: BrokerHandle, -) -> Result<(), tonic::transport::Error> { - tonic::transport::Server::builder() - .add_service(WinkeyerBrokerServiceServer::new(Service::new(broker))) - .serve(bind) - .await -} - /// Bind the loopback endpoint before returning so daemon startup fails loudly on conflicts. pub(crate) async fn bind_server( bind: SocketAddr, broker: BrokerHandle, -) -> std::io::Result>> { +) -> std::io::Result<( + SocketAddr, + tokio::task::JoinHandle>, +)> { let listener = tokio::net::TcpListener::bind(bind).await?; - Ok(tokio::spawn(async move { - tonic::transport::Server::builder() - .add_service(WinkeyerBrokerServiceServer::new(Service::new(broker))) - .serve_with_incoming(TcpListenerStream::new(listener)) - .await - })) + let address = listener.local_addr()?; + Ok(( + address, + tokio::spawn(async move { + tonic::transport::Server::builder() + .add_service(WinkeyerBrokerServiceServer::new(Service::new(broker))) + .serve_with_incoming(TcpListenerStream::new(listener)) + .await + }), + )) } #[allow(clippy::result_large_err)] @@ -399,10 +395,11 @@ mod tests { .await .expect("initialization"); - let reserved = std::net::TcpListener::bind("127.0.0.1:0").expect("reserve port"); - let address = reserved.local_addr().expect("address"); - drop(reserved); - let server = tokio::spawn(run_server(address, broker)); + let (address, server) = + bind_server("127.0.0.1:0".parse().expect("dynamic bind address"), broker) + .await + .expect("bind server"); + assert_ne!(address.port(), 0); let endpoint = format!("http://{address}"); let mut client = loop { match crate::broker_proto::winkeyer_broker_service_client::WinkeyerBrokerServiceClient::connect( diff --git a/crates/cathub/tests/cli.rs b/crates/cathub/tests/cli.rs index 625bc3e..dcbf30f 100644 --- a/crates/cathub/tests/cli.rs +++ b/crates/cathub/tests/cli.rs @@ -23,6 +23,8 @@ async fn dry_run_accepts_a_valid_loopback_config() { let cli = Cli { config: Some(path.clone()), section: None, + winkeyer_api_bind: None, + runtime_info: None, log: None, dry_run: true, command: None, @@ -41,6 +43,8 @@ async fn dry_run_rejects_an_invalid_backend() { let cli = Cli { config: Some(path.clone()), section: None, + winkeyer_api_bind: None, + runtime_info: None, log: None, dry_run: true, command: None, @@ -54,6 +58,8 @@ async fn missing_config_is_an_error() { let cli = Cli { config: Some(PathBuf::from("definitely-missing-cathub-config.toml")), section: None, + winkeyer_api_bind: None, + runtime_info: None, log: None, dry_run: true, command: None, diff --git a/docs/integration/setup.md b/docs/integration/setup.md index 58393aa..eefb2a6 100644 --- a/docs/integration/setup.md +++ b/docs/integration/setup.md @@ -215,6 +215,12 @@ Use the loopback gRPC address in `[winkeyer].api_bind`, normally `http://127.0.0.1:50071`. Supply a stable `client_name` so cancellation and telemetry remain scoped to that client. +An orchestrator can use `--winkeyer-api-bind 127.0.0.1:0`. +Windows or Linux then selects an available loopback port. +Use `--runtime-info ` to receive the effective endpoint and CatHub process ID. +Read the runtime file only after CatHub publishes it. +Validate the process ID before a client uses the endpoint. + ## 8. Safety acceptance Perform transmit tests while attended and into a suitable load: @@ -238,7 +244,7 @@ Perform transmit tests while attended and into a suitable load: | Writes return not supported | Check endpoint permissions and whether the command is modeled for that dialect. | | Digital mode stops on VFO B | Enable `single_vfo` for that client and use fake split. | | PTT is busy | Find the current lease owner in the log and confirm the previous client unkeyed. | -| Typed keyer API is unavailable | Verify `[winkeyer].api_bind` is loopback and not already in use. | +| Typed keyer API is unavailable | Verify `[winkeyer].api_bind` is loopback. An orchestrator can request port `0` and read `--runtime-info`. | | Maintenance is rejected | Stop active and queued sends, then connect through the `config_write` endpoint. | Set `CATHUB_LOG=debug` for verbose tracing. Narrow it to a module, such as