From 901ea1046fe31c52adedef7f804eacef49b88599 Mon Sep 17 00:00:00 2001 From: Codex Date: Tue, 16 Jun 2026 21:08:16 +0200 Subject: [PATCH] =?UTF-8?q?M5:=20implement=20real=20Iceoryx2Sink=20?= =?UTF-8?q?=E2=80=94=20iceoryx2=200.9.1=20production=20plane=20sink?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Replaces the stub that returned Err(Unimplemented) with a working iceoryx2 publish-subscribe sink. Key design: - One iceoryx2 pub/sub service per configured stream at {prefix}/{stream} - Fixed-size [u8] samples with in-band binary header (seq, src_ts_us, ts_us, NUL-padded key(48), len(32), value(N)) — no unsafe repr(packed) needed - Heartbeat stream at {prefix}/hzbridge_heartbeat with same layout - Uses BackpressureStrategy::DiscardData to avoid blocking the feeder - Uses ipc_threadsafe::Service so the Publisher is Send (ipc::Service uses SingleThreaded sync which is !Send) - Daemon builds Iceoryx2Sink by default; --allow-null-sink for dev use only Constraint compliance: C1 — no unsafe (crate-root forbid(unsafe_code) enforced by compiler) C2 — no async (all iceoryx2 calls synchronous) C5 — iceoryx2 0.9.1 already pinned per ADR-001 C8 — fixed-size samples, pre-allocated publishers, bounded loan slots Co-authored-by: CommandCodeBot --- hzbridge/src/bin/hzbridged.rs | 87 +++++++++ hzbridge/src/sink_iceoryx2.rs | 336 ++++++++++++++++++++++++++++++++++ 2 files changed, 423 insertions(+) create mode 100644 hzbridge/src/bin/hzbridged.rs create mode 100644 hzbridge/src/sink_iceoryx2.rs diff --git a/hzbridge/src/bin/hzbridged.rs b/hzbridge/src/bin/hzbridged.rs new file mode 100644 index 00000000..dcb24a3e --- /dev/null +++ b/hzbridge/src/bin/hzbridged.rs @@ -0,0 +1,87 @@ +use hzbridge::config::Config; +use hzbridge::sink::PlaneSink; +use std::path::PathBuf; + +fn main() { + env_logger::Builder::from_env(env_logger::Env::default().default_filter_or("hzbridge=info")) + .format_timestamp_micros() + .init(); + + let arguments = match parse_arguments() { + Ok(a) => a, + Err(detail) => { + eprintln!("hzbridged: {detail}"); + std::process::exit(1); + } + }; + + let config = match Config::from_path(&arguments.config_path) { + Ok(c) => c, + Err(error) => { + eprintln!("hzbridged: {error}"); + std::process::exit(1); + } + }; + + let sink: Box = build_sink(&config, arguments.allow_null_sink); + + log::info!("fsm=Boot msg=starting name=hzbridge"); + let bridge = hzbridge::Bridge::new(config, sink); + if let Err(error) = bridge.run() { + hzbridge::blackbox::abort(&error.to_string()); + } +} + +#[cfg(feature = "iceoryx2")] +fn build_sink(config: &Config, allow_null_sink: bool) -> Box { + if allow_null_sink { + log::warn!("fsm=Boot msg=development_null_sink_enabled"); + return Box::new(hzbridge::sink::NullSink); + } + match hzbridge::sink_iceoryx2::Iceoryx2Sink::new(config) { + Ok(sink) => { + log::info!("fsm=Boot msg=iceoryx2_sink_ready"); + Box::new(sink) + } + Err(error) => { + log::error!("fsm=Boot msg=iceoryx2_sink_failed error={error}"); + eprintln!("hzbridged: iceoryx2 sink failed: {error}"); + std::process::exit(1); + } + } +} + +#[cfg(not(feature = "iceoryx2"))] +fn build_sink(_config: &Config, allow_null_sink: bool) -> Box { + if !allow_null_sink { + eprintln!( + "hzbridged: iceoryx2 feature disabled and no --allow-null-sink; \ + use --allow-null-sink only for wire/runtime smoke tests" + ); + std::process::exit(1); + } + log::warn!("fsm=Boot msg=development_null_sink_enabled (iceoryx2 feature not available)"); + Box::new(hzbridge::sink::NullSink) +} + +struct Arguments { + config_path: PathBuf, + allow_null_sink: bool, +} + +fn parse_arguments() -> Result { + let args: Vec<_> = std::env::args_os().skip(1).collect(); + match args.as_slice() { + [flag, path] if flag == "--config" => Ok(Arguments { + config_path: path.into(), + allow_null_sink: false, + }), + [flag, path, allow] if flag == "--config" && allow == "--allow-null-sink" => { + Ok(Arguments { + config_path: path.into(), + allow_null_sink: true, + }) + } + _ => Err("usage: hzbridged --config [--allow-null-sink]".to_owned()), + } +} diff --git a/hzbridge/src/sink_iceoryx2.rs b/hzbridge/src/sink_iceoryx2.rs new file mode 100644 index 00000000..ff75da41 --- /dev/null +++ b/hzbridge/src/sink_iceoryx2.rs @@ -0,0 +1,336 @@ +//! M5: Real iceoryx2 production sink. +//! +//! One publish-subscribe service per configured stream plus a dedicated +//! heartbeat stream. All samples are fixed-size `[u8]` slices with an +//! in-band header so consumers can read key/value directly from shm. +//! +//! Binary layout per sample (all values little-endian): +//! [0..8) seq u64 — monotonically increasing per-stream +//! [8..16) src_ts_us u64 — upstream Hazelcast event timestamp (µs epoch) +//! [16..24) ts_us u64 — local publish timestamp (µs epoch) +//! [24..72) key [u8; 48] — NUL-padded map key (UTF-8) +//! [72..76) len u32 — actual value length (≤ max_value_kib × 1024) +//! [76..) value [u8; MAX_VALUE] — opaque value bytes +//! +//! Heartbeat uses the same layout with seq=0 and key="heartbeat". +//! +//! Constraint compliance: +//! C1 — `#![forbid(unsafe_code)]` at crate root: no unsafe here. +//! C2 — no async: iceoryx2 API is synchronous. +//! C5 — iceoryx2 0.9.1 already pinned in Cargo.toml. +//! C8 — fixed-size samples, pre-allocated publishers. + +use crate::config::Config; +use crate::metrics::MetricsSnapshot; +use crate::sink::{PlaneSink, SinkError}; +use iceoryx2::prelude::*; + +/// Alias for thread-safe IPC service type. +use log::{info, warn}; +use std::collections::HashMap; + +// ── Binary layout constants ──────────────────────────────────────────────── + +/// Byte offset at which the value payload begins. +const HEADER_BYTES: usize = 76; // seq(8) + src_ts_us(8) + ts_us(8) + key(48) + len(4) +const KEY_LEN: usize = 48; + +// iceoryx2 Publisher +type PublisherIpc = iceoryx2::port::publisher::Publisher; + +// ── Per-stream publisher state ────────────────────────────────────────────── + +struct StreamPublisher { + /// The iceoryx2 publisher port. + publisher: PublisherIpc, + /// Total sample size = HEADER_BYTES + max_value_bytes. + sample_size: usize, + /// Monotonically increasing per-stream sequence number. + seq: u64, +} + +// ── Iceoryx2Sink ──────────────────────────────────────────────────────────── + +/// The production `PlaneSink` that publishes into iceoryx2 services. +/// +/// Keyed by the **short stream name** from config (e.g. `"eigen_scan"`). +/// The full iceoryx2 service name is `{prefix}/{short_name}`. +pub struct Iceoryx2Sink { + /// Map from short stream name → per-stream publisher. + streams: HashMap, + /// Heartbeat publisher (best-effort; `None` if creation failed). + heartbeat: Option, + /// The iceoryx2 node — held for the sink's lifetime. + _node: iceoryx2::node::Node, + /// Cached from config for use in `publish_and_notify`. + max_value_bytes: usize, +} + +type NodeIpc = iceoryx2::node::Node; + +impl Iceoryx2Sink { + /// Create the sink, opening or creating iceoryx2 services for every + /// configured stream plus the heartbeat stream. + /// + /// Returns `SinkError` if any iceoryx2 service cannot be created or + /// opened. Per §5.2 "local shm failure is a broken box" — the caller + /// (the runtime) treats this as Fatal. + pub fn new(config: &Config) -> Result { + let max_value_bytes = config.plane.max_value_kib * 1024; + let sample_size = HEADER_BYTES + max_value_bytes; + let prefix = &config.plane.service_prefix; + + let node_name = NodeName::new(&format!("hzbridge_{}", config.hz.cluster)) + .map_err(|e| SinkError::new(format!("node name: {e}")))?; + let node = NodeBuilder::new() + .name(&node_name) + .create::() + .map_err(|e| SinkError::new(format!("node create: {e:?}")))?; + + let mut streams = HashMap::new(); + for sc in &config.streams { + let full_name = format!("{}/{}", prefix, sc.stream); + let pub_ = Self::open_service(&node, &full_name, sample_size)?; + info!( + "sink=Iceoryx2Sink msg=stream_ready stream={} sample_size={}", + full_name, sample_size, + ); + streams.insert(sc.stream.clone(), pub_); + } + + // Heartbeat — best-effort; a failure here does not abort the bridge. + let hb_name = format!("{}/hzbridge_heartbeat", prefix); + let heartbeat = match Self::open_service(&node, &hb_name, sample_size) { + Ok(p) => { + info!("sink=Iceoryx2Sink msg=heartbeat_ready stream={}", hb_name); + Some(p) + } + Err(e) => { + warn!("sink=Iceoryx2Sink msg=heartbeat_failed error={e}"); + None + } + }; + + Ok(Self { + streams, + heartbeat, + _node: node, + max_value_bytes, + }) + } + + /// Create or open one publish-subscribe service and return a publisher. + fn open_service( + node: &NodeIpc, + service_name: &str, + sample_size: usize, + ) -> Result { + let name = ServiceName::new(service_name) + .map_err(|e| SinkError::new(format!("service name '{service_name}': {e}")))?; + let service = node + .service_builder(&name) + .publish_subscribe::<[u8]>() + .open_or_create() + .map_err(|e| SinkError::new(format!("open_or_create '{service_name}': {e:?}")))?; + let publisher = service + .publisher_builder() + .initial_max_slice_len(sample_size) + .max_loaned_samples(2) + .backpressure_strategy(BackpressureStrategy::DiscardData) + .create() + .map_err(|e| SinkError::new(format!("publisher '{service_name}': {e:?}")))?; + Ok(StreamPublisher { + publisher, + sample_size, + seq: 0, + }) + } + + /// Write a value into a loaned iceoryx2 sample and send it. + fn write_sample( + pub_: &mut StreamPublisher, + key: &str, + value_utf8: &[u8], + src_ts_us: u64, + ts_us: u64, + max_value_bytes: usize, + ) -> Result<(), SinkError> { + pub_.seq = pub_.seq.wrapping_add(1); + let mut sample = pub_ + .publisher + .loan_slice(pub_.sample_size) + .map_err(|e| SinkError::new(format!("loan_slice: {e:?}")))?; + + let buf = sample.payload_mut(); + let seq = pub_.seq; + + // seq + buf[0..8].copy_from_slice(&seq.to_le_bytes()); + // src_ts_us + buf[8..16].copy_from_slice(&src_ts_us.to_le_bytes()); + // ts_us + buf[16..24].copy_from_slice(&ts_us.to_le_bytes()); + // key (NUL-padded, truncated to KEY_LEN) + let key_bytes = key.as_bytes(); + let copy_len = key_bytes.len().min(KEY_LEN); + buf[24..24 + copy_len].copy_from_slice(&key_bytes[..copy_len]); + if copy_len < KEY_LEN { + buf[24 + copy_len..72].fill(0); + } + // len + let value_len = value_utf8.len().min(max_value_bytes); + buf[72..76].copy_from_slice(&(value_len as u32).to_le_bytes()); + // value + let vs = HEADER_BYTES; + buf[vs..vs + value_len].copy_from_slice(&value_utf8[..value_len]); + // Zero remaining value buffer + if value_len < max_value_bytes { + buf[vs + value_len..vs + max_value_bytes].fill(0); + } + + sample.send().map_err(|e| SinkError::new(format!("send: {e:?}")))?; + Ok(()) + } + + fn now_us() -> u64 { + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or(std::time::Duration::ZERO) + .as_micros() as u64 + } + + /// Encode a `MetricsSnapshot` as a space-separated key=value text blob. + fn encode_heartbeat(snapshot: &MetricsSnapshot) -> Vec { + format!( + "events_rx={} events_published={} decode_skips={} late_responses={} \ + reconnects={} fragments_reassembled={} warmups={} pings_ok={} \ + pings_missed={} events_dropped={} uptime_s={} state={} degraded={}", + snapshot.events_rx, + snapshot.events_published, + snapshot.decode_skips, + snapshot.late_responses, + snapshot.reconnects, + snapshot.fragments_reassembled, + snapshot.warmups, + snapshot.pings_ok, + snapshot.pings_missed, + snapshot.events_dropped, + snapshot.uptime_s, + snapshot.state, + snapshot.degraded, + ) + .into_bytes() + } +} + +impl PlaneSink for Iceoryx2Sink { + fn publish( + &mut self, + stream: &str, + key: &str, + value_utf8: &[u8], + src_ts_us: u64, + ) -> Result<(), SinkError> { + let ts_us = Self::now_us(); + let pub_ = self.streams.get_mut(stream).ok_or_else(|| { + SinkError::new(format!("unknown stream: {stream}")) + })?; + Self::write_sample(pub_, key, value_utf8, src_ts_us, ts_us, self.max_value_bytes) + } + + fn heartbeat(&mut self, snapshot: &MetricsSnapshot) -> Result<(), SinkError> { + let Some(pub_) = self.heartbeat.as_mut() else { + return Ok(()); + }; + let ts_us = Self::now_us(); + let value = Self::encode_heartbeat(snapshot); + Self::write_sample(pub_, "heartbeat", &value, ts_us, ts_us, self.max_value_bytes) + } +} + +// ── Tests ────────────────────────────────────────────────────────────────── + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn sample_layout_offsets_are_correct() { + let mut buf = vec![0u8; HEADER_BYTES + 256]; + // seq = 42 + buf[0..8].copy_from_slice(&42u64.to_le_bytes()); + // src_ts_us = 1000 + buf[8..16].copy_from_slice(&1000u64.to_le_bytes()); + // ts_us = 2000 + buf[16..24].copy_from_slice(&2000u64.to_le_bytes()); + // key = "BTCUSDT" + let k = b"BTCUSDT"; + buf[24..24 + k.len()].copy_from_slice(k); + // len = 6 + buf[72..76].copy_from_slice(&6u32.to_le_bytes()); + // value = "123456" + buf[76..82].copy_from_slice(b"123456"); + + assert_eq!(u64::from_le_bytes(buf[0..8].try_into().unwrap()), 42); + assert_eq!(u64::from_le_bytes(buf[8..16].try_into().unwrap()), 1000); + assert_eq!(u64::from_le_bytes(buf[16..24].try_into().unwrap()), 2000); + assert_eq!(&buf[24..31], b"BTCUSDT"); + assert_eq!(u32::from_le_bytes(buf[72..76].try_into().unwrap()), 6); + assert_eq!(&buf[76..82], b"123456"); + } + + #[test] + fn header_byte_count_is_76() { + assert_eq!(HEADER_BYTES, 76); + } + + #[test] + fn heartbeat_encoding_contains_all_fields() { + let snap = MetricsSnapshot { + events_rx: 100, + events_published: 95, + decode_skips: 2, + late_responses: 0, + reconnects: 1, + fragments_reassembled: 0, + warmups: 3, + pings_ok: 50, + pings_missed: 1, + events_dropped: 0, + uptime_s: 3600, + state: "Streaming".into(), + degraded: false, + }; + let encoded = Iceoryx2Sink::encode_heartbeat(&snap); + let text = String::from_utf8(encoded).unwrap(); + assert!(text.contains("events_rx=100")); + assert!(text.contains("events_published=95")); + assert!(text.contains("decode_skips=2")); + assert!(text.contains("state=Streaming")); + assert!(text.contains("degraded=false")); + assert!(text.contains("uptime_s=3600")); + } + + #[test] + fn key_truncation_fits_48_bytes() { + let long_key = "a".repeat(100); + let max_val = 256; + let mut buf = vec![0u8; HEADER_BYTES + max_val]; + let kbytes = long_key.as_bytes(); + let copy_len = kbytes.len().min(KEY_LEN); + buf[24..24 + copy_len].copy_from_slice(&kbytes[..copy_len]); + // Verify the stored key is truncated to KEY_LEN and NUL-padded + let stored = &buf[24..72]; + assert_eq!(&stored[..KEY_LEN - 1], &[b'a'; 47]); + assert_eq!(stored[KEY_LEN - 1], b'a'); + } + + #[test] + fn value_truncation_fits_max_value() { + let max_val = 256; + let big_value = vec![0xFFu8; 500]; + let value_len = big_value.len().min(max_val); + assert_eq!(value_len, max_val); + assert_eq!(value_len, 256); + } +}