feat(core): Observation gains to/lifetime_ms; identity includes to

tines-core/src/belief.rs

@@ -107,6 +107,8 @@ mod tests {
             observed_at: None,
             topic: Topic::from_slice(b"door/3").unwrap(),
             payload: Payload::from_slice(b"open").unwrap(),
+            to: None,
+            lifetime_ms: None,
         }
     }
 

tines-core/src/exchange.rs

@@ -421,6 +421,8 @@ mod tests {
             observed_at: at,
             topic: Topic::from_slice(topic).unwrap(),
             payload: Payload::from_slice(payload).unwrap(),
+            to: None,
+            lifetime_ms: None,
         }
     }
 

tines-core/src/frag.rs

@@ -201,6 +201,8 @@ mod tests {
             observed_at: None,
             topic: Topic::from_slice(b"door/3").unwrap(),
             payload: Payload::from_slice(&std::vec![0u8; payload_len]).unwrap(),
+            to: None,
+            lifetime_ms: None,
         }
     }
 

tines-core/src/mem_store.rs

@@ -184,6 +184,8 @@ mod tests {
             observed_at: at,
             topic: Topic::from_slice(topic).unwrap(),
             payload: Payload::from_slice(b"").unwrap(),
+            to: None,
+            lifetime_ms: None,
         }
     }
 

tines-core/src/node.rs

@@ -91,6 +91,8 @@ impl<S: Store, T: Transport> Node<S, T> {
             observed_at,
             topic,
             payload,
+            to: None,
+            lifetime_ms: None,
         };
         self.next_seq += 1;
         let id = obs.id();

tines-core/src/observation.rs

@@ -25,6 +25,10 @@ pub struct Observation {
     pub observed_at: Option<u64>,
     pub topic: Topic,
     pub payload: Payload,
+    /// Addressing recipient; `None` means broadcast.
+    pub to: Option<NodeId>,
+    /// Optional absolute or relative lifetime in ms; `None` means never expires.
+    pub lifetime_ms: Option<u64>,
 }
 
 impl Observation {
@@ -41,6 +45,18 @@ impl Observation {
         hasher.update(&self.topic[..]);
         hasher.update((self.payload.len() as u16).to_le_bytes());
         hasher.update(&self.payload[..]);
+        // Fold the addressing recipient into the content id so that two
+        // observations with otherwise-identical content addressed to
+        // different nodes are treated as distinct keys. `to = None` (broadcast)
+        // is encoded as a single `0x00` byte, distinct from a real zero node id
+        // (encoded `0x01` + sixteen zero bytes).
+        match self.to {
+            None => hasher.update([0x00]),
+            Some(node_id) => {
+                hasher.update([0x01]);
+                hasher.update(node_id);
+            }
+        }
         let digest = hasher.finalize();
         let mut id = [0u8; 16];
         id.copy_from_slice(&digest[..16]);
@@ -56,6 +72,8 @@ impl Observation {
             observed_at: self.observed_at,
             topic: self.topic.clone(),
             payload: self.payload.clone(),
+            to: self.to,
+            lifetime_ms: self.lifetime_ms,
         }
     }
 }
@@ -64,7 +82,13 @@ impl Observation {
 mod tests {
     use super::*;
 
-    fn obs(seq: u64, age: u64, at: Option<u64>) -> Observation {
+    fn obs(
+        seq: u64,
+        age: u64,
+        at: Option<u64>,
+        to: Option<NodeId>,
+        lifetime_ms: Option<u64>,
+    ) -> Observation {
         Observation {
             observer: [1; 16],
             seq,
@@ -72,33 +96,41 @@ mod tests {
             observed_at: at,
             topic: Topic::from_slice(b"door/3").unwrap(),
             payload: Payload::from_slice(b"open").unwrap(),
+            to,
+            lifetime_ms,
         }
     }
 
     #[test]
     fn id_ignores_age_and_observed_at() {
-        assert_eq!(obs(1, 0, None).id(), obs(1, 999, Some(5)).id());
+        assert_eq!(
+            obs(1, 0, None, None, None).id(),
+            obs(1, 999, Some(5), None, None).id()
+        );
     }
 
     #[test]
     fn id_changes_with_seq() {
-        assert_ne!(obs(1, 0, None).id(), obs(2, 0, None).id());
+        assert_ne!(
+            obs(1, 0, None, None, None).id(),
+            obs(2, 0, None, None, None).id()
+        );
     }
 
     #[test]
     fn id_changes_with_observer() {
-        let mut b = obs(1, 0, None);
+        let mut b = obs(1, 0, None, None, None);
         b.observer = [2; 16];
-        assert_ne!(obs(1, 0, None).id(), b.id());
+        assert_ne!(obs(1, 0, None, None, None).id(), b.id());
     }
 
     #[test]
     fn id_distinguishes_topic_payload_boundary() {
         // topic "ab" payload "c" vs topic "a" payload "bc" must differ
-        let mut x = obs(1, 0, None);
+        let mut x = obs(1, 0, None, None, None);
         x.topic = Topic::from_slice(b"ab").unwrap();
         x.payload = Payload::from_slice(b"c").unwrap();
-        let mut y = obs(1, 0, None);
+        let mut y = obs(1, 0, None, None, None);
         y.topic = Topic::from_slice(b"a").unwrap();
         y.payload = Payload::from_slice(b"bc").unwrap();
         assert_ne!(x.id(), y.id());
@@ -106,20 +138,58 @@ mod tests {
 
     #[test]
     fn aged_saturates() {
-        assert_eq!(obs(1, u64::MAX - 1, None).aged(10).age_ms, u64::MAX);
+        assert_eq!(
+            obs(1, u64::MAX - 1, None, None, None).aged(10).age_ms,
+            u64::MAX
+        );
+    }
+
+    #[test]
+    fn to_is_part_of_identity() {
+        // same observation, to = None vs Some([2;16]) — must differ
+        assert_ne!(
+            obs(7, 0, None, None, None).id(),
+            obs(7, 0, None, Some([2; 16]), None).id()
+        );
+    }
+
+    #[test]
+    fn lifetime_is_not_part_of_identity() {
+        // same payload, both to = Some([2;16]), lifetime 100 vs 200 — must be equal
+        assert_eq!(
+            obs(7, 0, None, Some([2; 16]), Some(100)).id(),
+            obs(7, 0, None, Some([2; 16]), Some(200)).id()
+        );
+    }
+
+    #[test]
+    fn broadcast_marker_distinct_from_zero_id() {
+        // broadcast marker (to = None) must not collide with a real zero node id
+        assert_ne!(
+            obs(7, 0, None, None, None).id(),
+            obs(7, 0, None, Some([0; 16]), None).id()
+        );
+    }
+
+    #[test]
+    fn aged_preserves_addressing() {
+        let aged = obs(1, 0, None, Some([2; 16]), Some(500)).aged(10);
+        assert_eq!(aged.age_ms, 10);
+        assert_eq!(aged.to, Some([2; 16]));
+        assert_eq!(aged.lifetime_ms, Some(500));
     }
 
     #[test]
     fn id_known_vector() {
-        // pin the hash layout: observer=[1;16], seq=7, topic="t", payload="p"
+        // pin the hash layout: observer=[1;16], seq=7, topic="t", payload="p", to=None
-        let mut o = obs(7, 0, None);
+        let mut o = obs(7, 0, None, None, None);
         o.topic = Topic::from_slice(b"t").unwrap();
         o.payload = Payload::from_slice(b"p").unwrap();
         assert_eq!(
             o.id(),
             [
-                0x92, 0x37, 0x37, 0xa1, 0x06, 0x9b, 0xd9, 0x5e, 0x08, 0xc2, 0xe6, 0x39, 0x7d, 0x4a,
+                0x4a, 0x33, 0x09, 0x92, 0xf1, 0x61, 0xbe, 0xc9, 0x19, 0x7b, 0x6e, 0x96, 0x11, 0x6f,
-                0x8f, 0xd0
+                0xa4, 0xa2
             ]
         );
     }

tines-core/src/wire.rs

@@ -89,6 +89,8 @@ mod tests {
                 observed_at: Some(999),
                 topic: Topic::from_slice(b"door/3").unwrap(),
                 payload: Payload::from_slice(b"open").unwrap(),
+                to: None,
+                lifetime_ms: None,
             },
         }
     }
@@ -103,6 +105,8 @@ mod tests {
                 observed_at: None,
                 topic: Topic::from_slice(b"t").unwrap(),
                 payload: Payload::from_slice(&[0u8; 1024]).unwrap(),
+                to: None,
+                lifetime_ms: None,
             },
         }
     }