Bootstrap DD3 Rust port workspace with host-first compatibility tests

crates/dd3_sim/Cargo.toml

@@ -0,0 +1,9 @@
+[package]
+name = "dd3_sim"
+version = "0.1.0"
+edition = "2021"
+
+[dependencies]
+dd3_protocol = { path = "../dd3_protocol" }
+dd3_core = { path = "../dd3_core" }
+dd3_contracts = { path = "../dd3_contracts" }

crates/dd3_sim/src/fake_clock.rs

@@ -0,0 +1,47 @@
+use dd3_core::Clock;
+
+#[derive(Debug, Clone)]
+pub struct FakeClock {
+    now_ms: u64,
+    now_utc: u32,
+    synced: bool,
+}
+
+impl FakeClock {
+    pub fn new(now_ms: u64, now_utc: u32, synced: bool) -> Self {
+        Self {
+            now_ms,
+            now_utc,
+            synced,
+        }
+    }
+
+    pub fn advance_ms(&mut self, delta_ms: u64) {
+        let old_sec = self.now_ms / 1000;
+        self.now_ms = self.now_ms.saturating_add(delta_ms);
+        let new_sec = self.now_ms / 1000;
+        if self.synced && new_sec > old_sec {
+            let delta_sec = (new_sec - old_sec).min(u32::MAX as u64) as u32;
+            self.now_utc = self.now_utc.saturating_add(delta_sec);
+        }
+    }
+
+    pub fn set_time(&mut self, now_utc: u32, synced: bool) {
+        self.now_utc = now_utc;
+        self.synced = synced;
+    }
+}
+
+impl Clock for FakeClock {
+    fn now_ms(&self) -> u64 {
+        self.now_ms
+    }
+
+    fn now_utc(&self) -> u32 {
+        self.now_utc
+    }
+
+    fn is_time_synced(&self) -> bool {
+        self.synced
+    }
+}

crates/dd3_sim/src/fake_radio.rs

@@ -0,0 +1,161 @@
+use std::cell::RefCell;
+use std::rc::Rc;
+
+use dd3_core::Radio;
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum Endpoint {
+    Sender,
+    Receiver,
+}
+
+#[derive(Debug, Clone, Copy)]
+pub struct FakeRadioConfig {
+    pub loss_pct: u8,
+    pub duplicate_pct: u8,
+    pub max_delay_ms: u32,
+    pub seed: u64,
+}
+
+impl Default for FakeRadioConfig {
+    fn default() -> Self {
+        Self {
+            loss_pct: 0,
+            duplicate_pct: 0,
+            max_delay_ms: 0,
+            seed: 0xC0FFEE,
+        }
+    }
+}
+
+#[derive(Debug, Clone)]
+struct AirPacket {
+    to: Endpoint,
+    deliver_at_ms: u64,
+    payload: Vec<u8>,
+}
+
+#[derive(Debug)]
+pub struct FakeRadioBus {
+    cfg: FakeRadioConfig,
+    rng_state: u64,
+    packets: Vec<AirPacket>,
+}
+
+impl FakeRadioBus {
+    pub fn new(cfg: FakeRadioConfig) -> Self {
+        Self {
+            cfg,
+            rng_state: cfg.seed,
+            packets: Vec::new(),
+        }
+    }
+
+    fn next_rand(&mut self) -> u32 {
+        self.rng_state = self
+            .rng_state
+            .wrapping_mul(6364136223846793005)
+            .wrapping_add(1);
+        (self.rng_state >> 32) as u32
+    }
+
+    fn chance(&mut self, pct: u8) -> bool {
+        if pct == 0 {
+            return false;
+        }
+        (self.next_rand() % 100) < pct as u32
+    }
+
+    fn push_from(&mut self, from: Endpoint, now_ms: u64, payload: &[u8]) {
+        if self.chance(self.cfg.loss_pct) {
+            return;
+        }
+
+        let to = match from {
+            Endpoint::Sender => Endpoint::Receiver,
+            Endpoint::Receiver => Endpoint::Sender,
+        };
+
+        let delay = if self.cfg.max_delay_ms == 0 {
+            0
+        } else {
+            self.next_rand() % (self.cfg.max_delay_ms + 1)
+        } as u64;
+
+        self.packets.push(AirPacket {
+            to,
+            deliver_at_ms: now_ms.saturating_add(delay),
+            payload: payload.to_vec(),
+        });
+
+        if self.chance(self.cfg.duplicate_pct) {
+            let extra_delay = if self.cfg.max_delay_ms == 0 {
+                1
+            } else {
+                (self.next_rand() % (self.cfg.max_delay_ms + 1) + 1) as u64
+            };
+            self.packets.push(AirPacket {
+                to,
+                deliver_at_ms: now_ms.saturating_add(delay).saturating_add(extra_delay),
+                payload: payload.to_vec(),
+            });
+        }
+    }
+
+    fn pop_for(&mut self, to: Endpoint, now_ms: u64, window_ms: u32) -> Option<Vec<u8>> {
+        let deadline = now_ms.saturating_add(window_ms as u64);
+
+        let mut pick: Option<usize> = None;
+        let mut best_time = u64::MAX;
+
+        for (idx, pkt) in self.packets.iter().enumerate() {
+            if pkt.to != to {
+                continue;
+            }
+            if pkt.deliver_at_ms > deadline {
+                continue;
+            }
+            if pkt.deliver_at_ms < best_time {
+                best_time = pkt.deliver_at_ms;
+                pick = Some(idx);
+            }
+        }
+
+        pick.map(|idx| self.packets.swap_remove(idx).payload)
+    }
+}
+
+#[derive(Debug, Clone)]
+pub struct RadioEndpoint {
+    side: Endpoint,
+    bus: Rc<RefCell<FakeRadioBus>>,
+    now_ms: u64,
+}
+
+impl RadioEndpoint {
+    pub fn new(side: Endpoint, bus: Rc<RefCell<FakeRadioBus>>) -> Self {
+        Self {
+            side,
+            bus,
+            now_ms: 0,
+        }
+    }
+}
+
+impl Radio for RadioEndpoint {
+    fn send_frame(&mut self, frame: &[u8]) -> bool {
+        self.bus
+            .borrow_mut()
+            .push_from(self.side, self.now_ms, frame);
+        true
+    }
+
+    fn recv_frame(&mut self, window_ms: u32, now_ms: u64) -> Option<Vec<u8>> {
+        self.now_ms = now_ms;
+        self.bus.borrow_mut().pop_for(self.side, now_ms, window_ms)
+    }
+}
+
+pub fn shared_bus(cfg: FakeRadioConfig) -> Rc<RefCell<FakeRadioBus>> {
+    Rc::new(RefCell::new(FakeRadioBus::new(cfg)))
+}

crates/dd3_sim/src/lib.rs

@@ -0,0 +1,7 @@
+pub mod fake_clock;
+pub mod fake_radio;
+pub mod scenario;
+
+pub use fake_clock::FakeClock;
+pub use fake_radio::{Endpoint, FakeRadioBus, FakeRadioConfig, RadioEndpoint};
+pub use scenario::{MockPublisher, MockStatusSink, MockStorage, ScenarioRunner};

crates/dd3_sim/src/scenario.rs

@@ -0,0 +1,121 @@
+use std::cell::RefCell;
+use std::rc::Rc;
+
+use dd3_core::{
+    Publisher, ReceiverConfig, ReceiverPipeline, SenderConfig, SenderStateMachine, StatusSink, Storage,
+};
+
+use crate::fake_clock::FakeClock;
+use crate::fake_radio::{shared_bus, Endpoint, FakeRadioConfig, RadioEndpoint};
+
+#[derive(Debug, Default)]
+pub struct MockPublisher {
+    pub state_messages: Vec<(String, String)>,
+    pub fault_messages: Vec<(String, String)>,
+    pub discovery_messages: Vec<(String, String)>,
+}
+
+impl Publisher for MockPublisher {
+    fn publish_state(&mut self, device_id: &str, payload: &str) {
+        self.state_messages
+            .push((device_id.to_string(), payload.to_string()));
+    }
+
+    fn publish_faults(&mut self, device_id: &str, payload: &str) {
+        self.fault_messages
+            .push((device_id.to_string(), payload.to_string()));
+    }
+
+    fn publish_discovery(&mut self, device_id: &str, payload: &str) {
+        self.discovery_messages
+            .push((device_id.to_string(), payload.to_string()));
+    }
+}
+
+#[derive(Debug, Default)]
+pub struct MockStorage {
+    pub csv_lines: Vec<(String, String)>,
+}
+
+impl Storage for MockStorage {
+    fn append_csv(&mut self, device_id: &str, line: &str) {
+        self.csv_lines
+            .push((device_id.to_string(), line.to_string()));
+    }
+}
+
+#[derive(Debug, Default)]
+pub struct MockStatusSink {
+    pub last_sender_phase: String,
+    pub last_receiver_status: String,
+}
+
+impl StatusSink for MockStatusSink {
+    fn sender_phase(&mut self, phase: &str) {
+        self.last_sender_phase = phase.to_string();
+    }
+
+    fn receiver_status(&mut self, status: &str) {
+        self.last_receiver_status = status.to_string();
+    }
+}
+
+pub struct ScenarioRunner {
+    pub clock: FakeClock,
+    pub sender: SenderStateMachine,
+    pub receiver: ReceiverPipeline,
+    pub sender_radio: RadioEndpoint,
+    pub receiver_radio: RadioEndpoint,
+    pub publisher: MockPublisher,
+    pub storage: MockStorage,
+    pub status: MockStatusSink,
+    _bus: Rc<RefCell<crate::fake_radio::FakeRadioBus>>,
+}
+
+impl ScenarioRunner {
+    pub fn new(radio_cfg: FakeRadioConfig) -> Self {
+        let bus = shared_bus(radio_cfg);
+        let sender_radio = RadioEndpoint::new(Endpoint::Sender, bus.clone());
+        let receiver_radio = RadioEndpoint::new(Endpoint::Receiver, bus.clone());
+
+        Self {
+            clock: FakeClock::new(0, dd3_core::MIN_ACCEPTED_EPOCH_UTC, true),
+            sender: SenderStateMachine::new(
+                SenderConfig {
+                    short_id: 0xF19C,
+                    sender_id: 1,
+                    device_id: "dd3-F19C".to_string(),
+                },
+                0,
+            ),
+            receiver: ReceiverPipeline::new(ReceiverConfig {
+                short_id: 0xBEEF,
+                device_id: "dd3-BEEF".to_string(),
+                expected_sender_ids: vec![0xF19C],
+            }),
+            sender_radio,
+            receiver_radio,
+            publisher: MockPublisher::default(),
+            storage: MockStorage::default(),
+            status: MockStatusSink::default(),
+            _bus: bus,
+        }
+    }
+
+    pub fn tick(&mut self, ms: u64) {
+        for _ in 0..ms {
+            self.clock.advance_ms(1);
+            self.sender
+                .tick(&self.clock, &mut self.sender_radio, &mut self.status);
+            self.receiver.tick(
+                &self.clock,
+                &mut self.receiver_radio,
+                &mut self.publisher,
+                &mut self.storage,
+                &mut self.status,
+            );
+            self.sender
+                .tick(&self.clock, &mut self.sender_radio, &mut self.status);
+        }
+    }
+}

crates/dd3_sim/tests/state_machine_tests.rs

@@ -0,0 +1,214 @@
+use std::collections::HashSet;
+
+use dd3_core::{Clock, Radio, ReceiverConfig, ReceiverPipeline, SenderConfig, SenderStateMachine, StatusSink};
+use dd3_protocol::{encode_frame, MsgKind};
+use dd3_sim::{
+    Endpoint, FakeClock, FakeRadioConfig, MockPublisher, MockStatusSink, MockStorage, RadioEndpoint,
+    ScenarioRunner,
+};
+
+#[derive(Default)]
+struct NoopStatus;
+impl StatusSink for NoopStatus {}
+
+fn make_sync_ack(batch_id: u16, epoch: u32) -> Vec<u8> {
+    let payload = dd3_protocol::encode_ack_down_payload(dd3_protocol::AckDownPayload {
+        time_valid: true,
+        batch_id,
+        epoch_utc: epoch,
+    });
+    encode_frame(MsgKind::AckDown, 0xBEEF, &payload)
+}
+
+#[test]
+fn no_receiver_present_stays_unsynced_and_sync_only() {
+    let bus = dd3_sim::fake_radio::shared_bus(FakeRadioConfig::default());
+    let mut sender_radio = RadioEndpoint::new(Endpoint::Sender, bus.clone());
+    let mut sender = SenderStateMachine::new(
+        SenderConfig {
+            short_id: 0xF19C,
+            sender_id: 1,
+            device_id: "dd3-F19C".to_string(),
+        },
+        0,
+    );
+    let mut clock = FakeClock::new(0, dd3_core::MIN_ACCEPTED_EPOCH_UTC, true);
+    let mut status = NoopStatus::default();
+
+    for _ in 0..90_000u64 {
+        clock.advance_ms(1);
+        sender.tick(&clock, &mut sender_radio, &mut status);
+    }
+
+    let stats = sender.stats();
+    assert!(!sender.is_time_acquired());
+    assert_eq!(0, stats.build_count);
+    assert_eq!(0, stats.queue_depth);
+}
+
+#[test]
+fn receiver_bootstrap_unlocks_sampling_and_batches() {
+    let mut runner = ScenarioRunner::new(FakeRadioConfig::default());
+
+    // Initial unsynced period should result in sync request + ACK bootstrap.
+    runner.tick(20_000);
+    assert!(runner.sender.is_time_acquired());
+
+    // After unlock, sender should sample and send normal batches.
+    runner.tick(35_000);
+    assert!(!runner.publisher.state_messages.is_empty());
+    assert!(!runner.storage.csv_lines.is_empty());
+}
+
+#[test]
+fn packet_loss_eventually_progresses_without_duplicate_commit() {
+    let mut runner = ScenarioRunner::new(FakeRadioConfig {
+        loss_pct: 20,
+        duplicate_pct: 10,
+        max_delay_ms: 80,
+        seed: 0xBADC0DE,
+    });
+
+    runner.tick(180_000);
+
+    let sender_stats = runner.sender.stats();
+    assert!(sender_stats.last_acked_batch_id > 0);
+    assert!(!runner.publisher.state_messages.is_empty());
+
+    let mut uniq = HashSet::new();
+    for (topic, payload) in &runner.publisher.state_messages {
+        let key = format!("{topic}|{payload}");
+        assert!(uniq.insert(key), "duplicate committed state payload detected");
+    }
+}
+
+#[test]
+fn ack_mismatch_is_ignored_and_inflight_unchanged() {
+    let bus = dd3_sim::fake_radio::shared_bus(FakeRadioConfig::default());
+    let mut sender_radio = RadioEndpoint::new(Endpoint::Sender, bus.clone());
+    let mut inject_radio = RadioEndpoint::new(Endpoint::Receiver, bus.clone());
+
+    let mut sender = SenderStateMachine::new(
+        SenderConfig {
+            short_id: 0xF19C,
+            sender_id: 1,
+            device_id: "dd3-F19C".to_string(),
+        },
+        0,
+    );
+    let mut clock = FakeClock::new(0, dd3_core::MIN_ACCEPTED_EPOCH_UTC, true);
+    let mut status = MockStatusSink::default();
+
+    // Trigger sync-request send so we have inflight batch_id=1 and ack pending.
+    clock.advance_ms(15_000);
+    sender.tick(&clock, &mut sender_radio, &mut status);
+    assert!(sender.stats().ack_pending);
+
+    let bad_ack = make_sync_ack(999, dd3_core::MIN_ACCEPTED_EPOCH_UTC + 5);
+    assert!(inject_radio.send_frame(&bad_ack));
+
+    sender.tick(&clock, &mut sender_radio, &mut status);
+    let stats = sender.stats();
+    assert!(stats.ack_pending, "wrong ack must not clear inflight");
+    assert_eq!(0, stats.last_acked_batch_id);
+}
+
+#[test]
+fn backpressure_queue_depth_stays_bounded() {
+    let bus = dd3_sim::fake_radio::shared_bus(FakeRadioConfig::default());
+    let mut sender_radio = RadioEndpoint::new(Endpoint::Sender, bus.clone());
+    let mut inject_radio = RadioEndpoint::new(Endpoint::Receiver, bus.clone());
+
+    let mut sender = SenderStateMachine::new(
+        SenderConfig {
+            short_id: 0xF19C,
+            sender_id: 1,
+            device_id: "dd3-F19C".to_string(),
+        },
+        0,
+    );
+    let mut clock = FakeClock::new(0, dd3_core::MIN_ACCEPTED_EPOCH_UTC, true);
+    let mut status = MockStatusSink::default();
+
+    // Bootstrap one valid ACK so sender enters normal mode.
+    clock.advance_ms(15_000);
+    sender.tick(&clock, &mut sender_radio, &mut status);
+    let ack = make_sync_ack(1, dd3_core::MIN_ACCEPTED_EPOCH_UTC + 1);
+    assert!(inject_radio.send_frame(&ack));
+    sender.tick(&clock, &mut sender_radio, &mut status);
+    assert!(sender.is_time_acquired());
+
+    for _ in 0..600_000u64 {
+        clock.advance_ms(1);
+        sender.tick(&clock, &mut sender_radio, &mut status);
+    }
+
+    assert!(sender.stats().queue_depth as usize <= dd3_core::BATCH_QUEUE_DEPTH);
+}
+
+#[test]
+fn duplicate_batch_updates_counters_but_suppresses_publish_and_log() {
+    let bus = dd3_sim::fake_radio::shared_bus(FakeRadioConfig::default());
+    let mut tx_radio = RadioEndpoint::new(Endpoint::Sender, bus.clone());
+    let mut rx_radio = RadioEndpoint::new(Endpoint::Receiver, bus.clone());
+
+    let mut receiver = ReceiverPipeline::new(ReceiverConfig {
+        short_id: 0xBEEF,
+        device_id: "dd3-BEEF".to_string(),
+        expected_sender_ids: vec![0xF19C],
+    });
+
+    let mut clock = FakeClock::new(0, dd3_core::MIN_ACCEPTED_EPOCH_UTC + 100, true);
+    let mut pubsub = MockPublisher::default();
+    let mut storage = MockStorage::default();
+    let mut status = MockStatusSink::default();
+
+    let payload = include_bytes!("../../../fixtures/protocol/payload_v3/sparse_5.bin");
+    let batch_id = 42u16;
+    let total_len = payload.len() as u16;
+    let chunk_size = dd3_core::BATCH_CHUNK_PAYLOAD;
+    let chunks = ((payload.len() + chunk_size - 1) / chunk_size) as u8;
+
+    {
+        let mut offset = 0usize;
+        for idx in 0..chunks {
+            let part_len = (payload.len() - offset).min(chunk_size);
+            let mut pkt_payload = Vec::new();
+            pkt_payload.extend_from_slice(&batch_id.to_le_bytes());
+            pkt_payload.push(idx);
+            pkt_payload.push(chunks);
+            pkt_payload.extend_from_slice(&total_len.to_le_bytes());
+            pkt_payload.extend_from_slice(&payload[offset..offset + part_len]);
+            offset += part_len;
+            let frame = encode_frame(MsgKind::BatchUp, 0xF19C, &pkt_payload);
+            assert!(tx_radio.send_frame(&frame));
+        }
+        receiver.tick(&clock, &mut rx_radio, &mut pubsub, &mut storage, &mut status);
+    }
+    let first_state = pubsub.state_messages.len();
+    let first_csv = storage.csv_lines.len();
+
+    {
+        let mut offset = 0usize;
+        for idx in 0..chunks {
+            let part_len = (payload.len() - offset).min(chunk_size);
+            let mut pkt_payload = Vec::new();
+            pkt_payload.extend_from_slice(&batch_id.to_le_bytes());
+            pkt_payload.push(idx);
+            pkt_payload.push(chunks);
+            pkt_payload.extend_from_slice(&total_len.to_le_bytes());
+            pkt_payload.extend_from_slice(&payload[offset..offset + part_len]);
+            offset += part_len;
+            let frame = encode_frame(MsgKind::BatchUp, 0xF19C, &pkt_payload);
+            assert!(tx_radio.send_frame(&frame));
+        }
+        receiver.tick(&clock, &mut rx_radio, &mut pubsub, &mut storage, &mut status);
+    }
+
+    let statuses = receiver.sender_statuses();
+    assert_eq!(2, statuses[0].rx_batches_total);
+    assert_eq!(1, statuses[0].rx_batches_duplicate);
+
+    assert_eq!(first_state, pubsub.state_messages.len(), "duplicate should not republish state");
+    assert_eq!(first_csv, storage.csv_lines.len(), "duplicate should not duplicate csv log");
+}