M5: implement real Iceoryx2Sink — iceoryx2 0.9.1 production plane sink

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 <noreply@commandcode.ai>

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<dyn PlaneSink> = 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<dyn PlaneSink> {
+    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<dyn PlaneSink> {
+    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<Arguments, String> {
+    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 <path> [--allow-null-sink]".to_owned()),
+    }
+}

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<Service=ipc_threadsafe::Service, Payload=[u8], UserHeader=()>
+type PublisherIpc = iceoryx2::port::publisher::Publisher<ipc_threadsafe::Service, [u8], ()>;
+
+// ── 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<String, StreamPublisher>,
+    /// Heartbeat publisher (best-effort; `None` if creation failed).
+    heartbeat: Option<StreamPublisher>,
+    /// The iceoryx2 node — held for the sink's lifetime.
+    _node: iceoryx2::node::Node<ipc_threadsafe::Service>,
+    /// Cached from config for use in `publish_and_notify`.
+    max_value_bytes: usize,
+}
+
+type NodeIpc = iceoryx2::node::Node<ipc_threadsafe::Service>;
+
+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<Self, SinkError> {
+        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::<ipc_threadsafe::Service>()
+            .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<StreamPublisher, SinkError> {
+        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<u8> {
+        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);
+    }
+}