feat(core): Node::observe_to and belief excludes addressed-to-others

tines-core/src/node.rs

@@ -100,6 +100,43 @@ impl<S: Store, T: Transport> Node<S, T> {
         Ok(id)
     }
 
+    /// Records an observation addressed to `to` with a `lifetime_ms`, made
+    /// "now" (with `age_ms = 0`) under `topic`.
+    ///
+    /// The resulting observation has `to = Some(to)` and
+    /// `lifetime_ms = Some(lifetime_ms)`, so the exchange will deliver it only
+    /// to `to` (see [`Exchange`]). Returns the observation's content-derived
+    /// id. An oversized topic or payload is rejected with
+    /// [`ObserveError::TooLarge`] rather than truncated; a store rejection is
+    /// surfaced as [`ObserveError::Store`]. Duplicate ids are accepted (the id
+    /// is still returned).
+    pub fn observe_to(
+        &mut self,
+        to: NodeId,
+        lifetime_ms: u64,
+        topic: &[u8],
+        payload: &[u8],
+        observed_at: Option<u64>,
+        now: Tick,
+    ) -> Result<ObservationId, ObserveError> {
+        let topic = Topic::from_slice(topic).map_err(|_| ObserveError::TooLarge)?;
+        let payload = Payload::from_slice(payload).map_err(|_| ObserveError::TooLarge)?;
+        let obs = Observation {
+            observer: self.me,
+            seq: self.next_seq,
+            age_ms: 0,
+            observed_at,
+            topic,
+            payload,
+            to: Some(to),
+            lifetime_ms: Some(lifetime_ms),
+        };
+        self.next_seq += 1;
+        let id = obs.id();
+        self.store.insert(obs, now).map_err(ObserveError::Store)?;
+        Ok(id)
+    }
+
     /// Drains all pending transport events, storing any newly received
     /// observations, then emits at most one outbound frame.
     ///
@@ -140,6 +177,11 @@ impl<S: Store, T: Transport> Node<S, T> {
     ) -> Option<P::Belief> {
         let mut acc: Option<P::Belief> = None;
         self.store.for_each_topic(topic_prefix, now, &mut |obs| {
+            // App-facing view: broadcast plus addressed to `me` only.
+            // Drop anything addressed to a different node.
+            if matches!(obs.to, Some(other) if other != self.me) {
+                return;
+            }
             let prev = acc.take();
             acc = policy.step(self.me, prev, obs);
         });

tines-core/tests/pair.rs

@@ -220,3 +220,55 @@ fn pair_transport_send_rejects_unreachable() {
     // a's only peer is 1; sending to 2 is unreachable.
     assert_eq!(a.send(&2, b"hi"), Err(SendError::Unreachable));
 }
+
+#[test]
+fn observe_to_creates_addressed_message() {
+    let (a, _b) = PairTransport::pair(1500);
+    let mut node = Node::new([1u8; 16], 0, MemStore::new(), a);
+
+    let id = node.observe_to([2; 16], 5000, b"task", b"go", None, 0).unwrap();
+    assert_eq!(node.next_seq(), 1);
+
+    let obs = node.store().get(&id, 0).expect("observation stored");
+    assert_eq!(obs.observer, [1; 16]);
+    assert_eq!(obs.to, Some([2; 16]));
+    assert_eq!(obs.lifetime_ms, Some(5000));
+    assert_eq!(obs.age_ms, 0);
+}
+
+#[test]
+fn observe_to_rejects_oversized_topic() {
+    let (a, _b) = PairTransport::pair(1500);
+    let mut node = Node::new([1u8; 16], 0, MemStore::new(), a);
+
+    let oversized_topic = vec![0u8; consts::MAX_TOPIC + 1];
+    assert_eq!(
+        node.observe_to([2; 16], 5000, &oversized_topic, b"go", None, 0),
+        Err(ObserveError::TooLarge)
+    );
+    // Rejected whole: nothing stored, seq not advanced.
+    assert_eq!(node.store().len(), 0);
+    assert_eq!(node.next_seq(), 0);
+}
+
+#[test]
+fn belief_excludes_addressed_to_others() {
+    let (a, _b) = PairTransport::pair(1500);
+    let mut node = Node::new([1u8; 16], 0, MemStore::new(), a);
+
+    // A broadcast observation and an addressed-to-other observation on the
+    // same topic. `me` = [1;16]; the addressed one is for [2;16] (≠ me).
+    node.observe(b"door", b"broadcast", None, 0).unwrap();
+    node.observe_to([2; 16], 10_000, b"door", b"addressed", None, 10).unwrap();
+
+    let now = 100_000u64;
+    let belief = node
+        .belief(b"door", &LatestWins, now)
+        .expect("door belief exists");
+    assert_eq!(
+        belief.payload,
+        Payload::from_slice(b"broadcast").unwrap(),
+        "app view must show the broadcast, not the addressed-to-other"
+    );
+    assert_eq!(belief.observer, [1; 16]);
+}