test(sim): addressed relay scenario and property tests

tines-sim/src/sim.rs

@@ -13,7 +13,7 @@ use crate::world::{SimWorld, WorldConfig};
 use std::cell::RefCell;
 use std::collections::{BTreeMap, BTreeSet};
 use std::rc::Rc;
-use tines_core::{MemStore, Node, ObservationId, Store};
+use tines_core::{MemStore, Node, NodeId, ObservationId, Store};
 
 /// A single node as seen through the simulation transport.
 pub type SimNode = Node<MemStore, SimTransport>;
@@ -67,6 +67,19 @@ impl Sim {
             .expect("observe")
     }
 
+    pub fn observe_to(
+        &mut self,
+        node: usize,
+        to: NodeId,
+        lifetime_ms: u64,
+        topic: &[u8],
+        payload: &[u8],
+    ) -> ObservationId {
+        self.nodes[node]
+            .observe_to(to, lifetime_ms, topic, payload, None, self.t)
+            .expect("observe_to")
+    }
+
     pub fn id_set(&self, node: usize) -> BTreeSet<ObservationId> {
         let mut set = BTreeSet::new();
         self.nodes[node].store().for_each_id(&mut |id| {

tines-sim/tests/props.rs

@@ -11,6 +11,11 @@
 //!   advances (no trimming).
 //! * [`garbage_frames_never_panic`] — malformed wire frames are dropped, never
 //!   panic the node.
+//! * [`addressed_reaches_recipient_along_connected_path`] — an addressed task
+//!   handed off hop-by-hop along a connected contact path reaches the
+//!   recipient. Addressed messages diverge by design (the recipient holds a
+//!   tasking the carriers no longer need), so this asserts only that the
+//!   recipient eventually holds the id, never whole-graph convergence.
 
 use proptest::prelude::*;
 use tines_core::{FrameBuf, Store};
@@ -115,6 +120,40 @@ fn permute3(perm: u64) -> [usize; 3] {
     bases[(perm % 6) as usize]
 }
 
+/// The forward stride between consecutive edges of [`chain_trace`]: each edge
+/// `(i-1, i)` is shifted a full stride past the last, keeping the windows
+/// ordered so a task can hop forward along the path.
+const PATH_STRIDE: u64 = 4_000;
+
+/// The length of each edge's contact window in [`chain_trace`], leaving a
+/// 1_000 ms gap to the next edge for frames to land before that contact closes.
+const PATH_WINDOW: u64 = 3_000;
+
+/// A forward path over `nodes` nodes: edges `(0,1),(1,2),...,(n-2,n-1)` with
+/// ordered, non-overlapping contact windows. The contact graph is connected and
+/// a task observed before the first contact can be relayed hop-by-hop to the
+/// max-id node.
+fn chain_trace(nodes: usize) -> ContactTrace {
+    let mut trace = ContactTrace::default();
+    for j in 1..nodes {
+        trace.contacts.push(Contact {
+            start_ms: j as u64 * PATH_STRIDE,
+            end_ms: j as u64 * PATH_STRIDE + PATH_WINDOW,
+            a: j - 1,
+            b: j,
+        });
+    }
+    trace
+}
+
+/// A scenario with a connected path trace and a random tasker node:
+/// `(tasker, nodes, trace)`.
+fn addressed_scenario() -> impl Strategy<Value = (usize, usize, ContactTrace)> {
+    (2..=6usize).prop_flat_map(|nodes| {
+        (0usize..nodes).prop_map(move |tasker| (tasker, nodes, chain_trace(nodes)))
+    })
+}
+
 proptest! {
     #![proptest_config(ProptestConfig { cases: 64, .. ProptestConfig::default() })]
 
@@ -259,4 +298,38 @@ proptest! {
         prop_assert_eq!(stored, 0, "a garbage frame must not be stored");
         prop_assert_eq!(node.store().len(), 0, "store must be unchanged by a garbage frame");
     }
+
+    /// An addressed task handed off hop-by-hop along a connected contact path
+    /// reaches the recipient.
+    ///
+    /// The trace is a forward path (see [`chain_trace`]) over `2..=6` nodes, so
+    /// it is always connected; a random node issues the task and the max-id
+    /// node is the recipient. Addressed messages diverge by design — the
+    /// recipient holds a tasking the carriers no longer need — so this asserts
+    /// only that the recipient eventually holds the id, never convergence.
+    #[test]
+    fn addressed_reaches_recipient_along_connected_path(
+        (tasker, nodes, trace) in addressed_scenario(),
+    ) {
+        let end = trace.end();
+        let cfg = WorldConfig {
+            nodes,
+            mtu: 512,
+            loss: 0.0,
+            dup: 0.0,
+            seed: 1,
+        };
+        let mut sim = Sim::new(cfg, trace);
+        // The max-id node's id is its index replicated to 16 bytes, matching
+        // how `Sim` assigns node ids.
+        let recipient = [nodes.saturating_sub(1) as u8; 16];
+        let id = sim.observe_to(tasker, recipient, 600_000, b"task", b"investigate");
+        sim.run_until(end + 1_000);
+
+        let recipient_idx = nodes.saturating_sub(1);
+        prop_assert!(
+            sim.nodes[recipient_idx].store().contains(&id),
+            "addressed task from node {tasker} did not reach the max-id node {recipient_idx}"
+        );
+    }
 }

tines-sim/tests/relay.rs

@@ -0,0 +1,81 @@
+//! Addressed (relayed) delivery scenarios for the `Sim` harness.
+//!
+//! Unlike the broadcast convergence scenarios in `convergence.rs`, addressed
+//! observations diverge by design: the recipient holds a tasking that the
+//! carriers no longer need once they have forwarded it. These tests exercise
+//! the relay path — an [`Sim::observe_to`] observation handed off hop-by-hop
+//! along a contact chain — without asserting whole-graph convergence.
+
+use tines_core::{NodeId, Store};
+use tines_sim::{Contact, ContactTrace, Sim, WorldConfig};
+
+/// The recipient id. `Sim` assigns node `i` the id `[i as u8; 16]`, so node 2's
+/// id is `[2; 16]` and the task in both tests below is addressed to it.
+const RECIPIENT: NodeId = [2; 16];
+
+/// The relay contact trace shared by both tests: `(0,1)` during `[0, 5000)`,
+/// then `(1,2)` during `[10_000, 15_000)`. Node 0 and node 2 never share a
+/// contact, so a task must be relayed through node 1 to reach node 2.
+fn relay_trace() -> ContactTrace {
+    let mut trace = ContactTrace::default();
+    trace.contacts.push(Contact {
+        start_ms: 0,
+        end_ms: 5_000,
+        a: 0,
+        b: 1,
+    });
+    trace.contacts.push(Contact {
+        start_ms: 10_000,
+        end_ms: 15_000,
+        a: 1,
+        b: 2,
+    });
+    trace
+}
+
+/// The task observed at `t = 0` reaches the far end of the chain even though
+/// node 0 and node 2 never met; it is relayed through node 1.
+#[test]
+fn relayed_task_reaches_recipient() {
+    let trace = relay_trace();
+    let end = trace.end();
+    let cfg = WorldConfig {
+        nodes: 3,
+        mtu: 1500,
+        loss: 0.0,
+        dup: 0.0,
+        seed: 0,
+    };
+    let mut sim = Sim::new(cfg, trace);
+    let id = sim.observe_to(0, RECIPIENT, 60_000, b"task", b"investigate");
+    sim.run_until(end);
+
+    assert!(
+        sim.nodes[2].store().contains(&id),
+        "node 2 must hold the relayed task"
+    );
+}
+
+/// An addressed task with a short lifetime never reaches the recipient: the
+/// carrier drops it once its age passes the lifetime, so node 2 never stores
+/// it after running well past expiry.
+#[test]
+fn carrier_drops_expired() {
+    let trace = relay_trace();
+    let end = trace.end();
+    let cfg = WorldConfig {
+        nodes: 3,
+        mtu: 1500,
+        loss: 0.0,
+        dup: 0.0,
+        seed: 0,
+    };
+    let mut sim = Sim::new(cfg, trace);
+    let id = sim.observe_to(0, RECIPIENT, 10, b"task", b"investigate");
+    sim.run_until(end + 10_000);
+
+    assert!(
+        !sim.nodes[2].store().contains(&id),
+        "an addressed task past its lifetime must not survive at the recipient"
+    );
+}