From e85b3704e0b1df98cea316f9704129274a39ddb0 Mon Sep 17 00:00:00 2001 From: "K. Isom" Date: Tue, 20 Oct 2026 09:07:39 -0700 Subject: [PATCH] feat(core): Dest type replaces to + lifetime_ms --- tines-core/src/belief.rs | 3 +- tines-core/src/exchange.rs | 184 ++++++++++----------- tines-core/src/frag.rs | 3 +- tines-core/src/lib.rs | 2 +- tines-core/src/mem_store.rs | 19 ++- tines-core/src/node.rs | 16 +- tines-core/src/observation.rs | 264 +++++++++++++++++++++++++------ tines-core/src/store.rs | 6 +- tines-core/src/wire.rs | 6 +- tines-core/tests/pair.rs | 4 +- tines-store/src/log.rs | 29 ++-- tines-store/tests/persistence.rs | 30 +++- 12 files changed, 368 insertions(+), 198 deletions(-) diff --git a/tines-core/src/belief.rs b/tines-core/src/belief.rs index 97e3a028..7bc4c948 100644 --- a/tines-core/src/belief.rs +++ b/tines-core/src/belief.rs @@ -107,8 +107,7 @@ 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, + dest: None, } } diff --git a/tines-core/src/exchange.rs b/tines-core/src/exchange.rs index 6a83daf6..3187c704 100644 --- a/tines-core/src/exchange.rs +++ b/tines-core/src/exchange.rs @@ -703,9 +703,9 @@ impl Exchange { for deliver in [true, false] { let found = select_addressed(store, now, &self.bounce, &addr, cursor, |id, obs| { let wanted = if deliver { - obs.to == Some(recipient) + obs.dest.as_ref().map(|d| d.to) == Some(recipient) } else { - obs.to.is_some() && obs.to != Some(me) + obs.dest.is_some() && obs.dest.as_ref().map(|d| d.to) != Some(me) }; // `out` is scratch here; the chosen frame is re-encoded below. wanted && encode_frame(me, *id, obs, None, mtu, out).is_some() @@ -753,7 +753,7 @@ impl Exchange { peer.send_queue.remove(pos); undo.requeue = Some((pos, id)); } - if obs.to.is_some() + if obs.dest.is_some() && record_bounce(&mut self.bounce, id, addr.clone(), BounceKind::Sent) { undo.bounce = Some(id); @@ -908,7 +908,7 @@ impl Exchange { // recipient alike. The custody chain tracks that a copy left us, // so we stop forwarding this id; final delivery to the recipient // is best-effort, not guaranteed. - if store.get(id, now).is_some_and(|obs| obs.to.is_some()) { + if store.get(id, now).is_some_and(|obs| obs.dest.is_some()) { store.release(id); } } else if bcast_sent_contains(&self.bcast_sent, *id, &addr) { @@ -938,7 +938,7 @@ impl Exchange { return None; } let id = obs.id(); - let addressed = obs.to.is_some(); + let addressed = obs.dest.is_some(); let returned = obs.clone(); let res = store.insert(obs, now); // Enqueue a pending ack only for what the store actually kept: a newly @@ -1003,20 +1003,17 @@ fn summary_bytes(mtu: usize) -> Option { } /// Returns `true` when a received [`Observation`] must be dropped before it is -/// stored: either its addressing is malformed, or it is an addressed message -/// whose age has passed its lifetime. +/// stored: it is an addressed message (`dest = Some`) whose age has passed its +/// lifetime. /// -/// An addressed message (`to = Some`) must carry a `lifetime_ms`; a mismatched -/// pair — `to = Some` with no lifetime, or a lifetime with no recipient — -/// violates the addressing invariant and is dropped in either direction. A -/// well-formed addressed message is rejected once its age exceeds its -/// lifetime. Broadcasts (`to = None`, `lifetime_ms = None`) are never rejected. +/// An addressed message always carries a lifetime, so the only way to reject +/// is age exceeding that lifetime. Broadcasts (`dest = None`) are never +/// rejected. The `Dest` type now enforces the addressed/broadcast pairing, so +/// there is no separate "malformed addressing" case to drop. fn reject(obs: &Observation) -> bool { - match (obs.to, obs.lifetime_ms) { - (None, None) => false, - (Some(_), Some(lifetime)) => obs.age_ms > lifetime, - _ => true, - } + obs.dest + .as_ref() + .is_some_and(|d| obs.age_ms > d.lifetime_ms) } /// Finds the non-expired, well-formed addressed observation matching @@ -1054,7 +1051,7 @@ fn select_addressed( let Some(obs) = store.get(id, now) else { return; }; - if obs.to.is_none() || reject(&obs) || store.is_released(id) || !predicate(id, &obs) { + if obs.dest.is_none() || reject(&obs) || store.is_released(id) || !predicate(id, &obs) { return; } if is_ahead { @@ -1272,7 +1269,9 @@ fn insert_pos(heap: &Vec<(ObservationId, u64), { consts::SEND_QUEUE_CAP }>, age: #[cfg(all(test, feature = "alloc"))] mod tests { use super::*; - use crate::{MemStore, Observation, Payload, SendError, Store, StoreError, Topic, TrimPolicy}; + use crate::{ + Dest, MemStore, Observation, Payload, SendError, Store, StoreError, Topic, TrimPolicy, + }; use std::vec::Vec; const ME: NodeId = [42; 16]; @@ -1292,8 +1291,7 @@ mod tests { observed_at: at, topic: Topic::from_slice(topic).unwrap(), payload: Payload::from_slice(payload).unwrap(), - to: None, - lifetime_ms: None, + dest: None, } } @@ -1378,8 +1376,8 @@ mod tests { #[test] fn rebuild_send_queue_excludes_addressed() { - // A store with one broadcast (`to == None`) and one addressed - // (`to = Some([3; 16])`) observation. After a peer contacts us and + // A store with one broadcast (`dest = None`) and one addressed + // (`dest = Some([3; 16])`) observation. After a peer contacts us and // sends an empty Summary (lacking everything), the broadcast diff must // queue only the broadcast id and never leak in the addressed id. let mut ex = Exchange::new(ME); @@ -1387,7 +1385,10 @@ mod tests { let b = obs([2; 16], 1, 0, None, b"t", b"b"); let a_id = { let mut a = obs([2; 16], 2, 0, None, b"t", b"a"); - a.to = Some([3; 16]); + a.dest = Some(Dest { + to: [3; 16], + lifetime_ms: consts::SUMMARY_INTERVAL_MS * 10, + }); a.id() }; let b_id = b.id(); @@ -1396,7 +1397,10 @@ mod tests { .insert( { let mut a = obs([2; 16], 2, 0, None, b"t", b"a"); - a.to = Some([3; 16]); + a.dest = Some(Dest { + to: [3; 16], + lifetime_ms: consts::SUMMARY_INTERVAL_MS * 10, + }); a }, 0, @@ -1454,9 +1458,10 @@ mod tests { payload: &[u8], ) -> Observation { let mut o = obs(observer, seq, 0, None, topic, payload); - o.to = Some(to); - // An addressed message must carry a lifetime or `reject` drops it. - o.lifetime_ms = Some(consts::SUMMARY_INTERVAL_MS * 10); + o.dest = Some(Dest { + to, + lifetime_ms: consts::SUMMARY_INTERVAL_MS * 10, + }); o } @@ -2035,8 +2040,10 @@ mod tests { ex.on_event(Event::Contact(1), &mut store, 0); // Addressed to us, lifetime 100ms, but age 200ms -> expired. let mut o = obs([2; 16], 7, 0, None, b"t", b"hi"); - o.to = Some(ME); - o.lifetime_ms = Some(100); + o.dest = Some(Dest { + to: ME, + lifetime_ms: 100, + }); o.age_ms = 200; let frame = Frame::Obs { from: [2; 16], @@ -2060,8 +2067,10 @@ mod tests { let mut store = MemStore::new(); // A large addressed obs that must be fragmented; already expired. let mut big = obs([2; 16], 7, 0, None, b"t", &[0u8; 1024]); - big.to = Some(ME); - big.lifetime_ms = Some(100); + big.dest = Some(Dest { + to: ME, + lifetime_ms: 100, + }); big.age_ms = 200; let mut frames: Vec> = Vec::new(); crate::frag::fragment([2; 16], &big, 200, &mut |b| { @@ -2078,30 +2087,6 @@ mod tests { ); } - #[test] - fn addressing_invariant_violation_is_dropped() { - let mut ex = Exchange::new(ME); - let mut store = MemStore::new(); - ex.on_event(Event::Contact(1), &mut store, 0); - // to = Some but lifetime_ms = None: the lifetime==to invariant is - // violated, so this frame is malformed and must be dropped, not stored. - let mut o = obs([2; 16], 7, 0, None, b"t", b"hi"); - o.to = Some(ME); - o.lifetime_ms = None; - let frame = Frame::Obs { - from: [2; 16], - obs: o.clone(), - }; - assert!(ex - .on_event(Event::Frame(1, frame_bytes(&frame)), &mut store, 0) - .is_none()); - assert_eq!( - store.len(), - 0, - "an invariant-violating frame must not be stored" - ); - } - #[test] fn valid_addressed_obs_is_accepted() { let mut ex = Exchange::new(ME); @@ -2109,8 +2094,10 @@ mod tests { ex.on_event(Event::Contact(1), &mut store, 0); // A well-formed, non-expired addressed obs addressed to us IS stored. let mut o = obs([2; 16], 7, 0, None, b"t", b"hi"); - o.to = Some(ME); - o.lifetime_ms = Some(1000); + o.dest = Some(Dest { + to: ME, + lifetime_ms: 1000, + }); let id = o.id(); let frame = Frame::Obs { from: [2; 16], @@ -2125,30 +2112,6 @@ mod tests { assert_eq!(store.len(), 1); } - #[test] - fn addressing_invariant_violation_reverse_is_dropped() { - let mut ex = Exchange::new(ME); - let mut store = MemStore::new(); - ex.on_event(Event::Contact(1), &mut store, 0); - // to = None but lifetime_ms = Some: the invariant is violated in the - // reverse direction, so this frame is malformed and must be dropped. - let mut o = obs([2; 16], 7, 0, None, b"t", b"hi"); - o.to = None; - o.lifetime_ms = Some(100); - let frame = Frame::Obs { - from: [2; 16], - obs: o.clone(), - }; - assert!(ex - .on_event(Event::Frame(1, frame_bytes(&frame)), &mut store, 0) - .is_none()); - assert_eq!( - store.len(), - 0, - "an invariant-violating frame must not be stored" - ); - } - #[test] fn expired_addressed_obs_is_not_forwarded() { let mut ex = Exchange::new(ME); @@ -2156,8 +2119,10 @@ mod tests { // An addressed obs already past its lifetime, inserted directly so it // bypasses the receive-side reject (this tests the FORWARD path). let mut o = obs([2; 16], 7, 0, None, b"t", b"hi"); - o.to = Some([3; 16]); - o.lifetime_ms = Some(100); + o.dest = Some(Dest { + to: [3; 16], + lifetime_ms: 100, + }); o.age_ms = 200; let id = o.id(); store.insert(o, 0).unwrap(); @@ -2189,7 +2154,10 @@ mod tests { let b = obs([2; 16], 1, 0, None, b"t", b"b"); let b_id = b.id(); let mut a = obs([2; 16], 2, 0, None, b"t", b"a"); - a.to = Some([2; 16]); + a.dest = Some(Dest { + to: [2; 16], + lifetime_ms: consts::SUMMARY_INTERVAL_MS * 10, + }); let a_id = a.id(); store.insert(b, 0).unwrap(); store.insert(a, 0).unwrap(); @@ -2214,7 +2182,7 @@ mod tests { } } - /// Drains `next_frame` and returns the first `Obs` frame whose `to` matches + /// Drains `next_frame` and returns the first `Obs` frame whose destination matches /// `want`, along with its destination address — skipping summaries and /// non-Obs frames. Breaks on the first match or when `next_frame` yields /// `None`. @@ -2228,7 +2196,7 @@ mod tests { let mut buf = [0u8; consts::MAX_FRAME]; match ex.next_frame(store, consts::MAX_FRAME, 0, &mut buf) { Some((addr, n)) => match Frame::decode(&buf[..n]).unwrap() { - Frame::Obs { obs, .. } if obs.to == want => { + Frame::Obs { obs, .. } if obs.dest.as_ref().map(|d| d.to) == want => { result = Some((addr, obs.id())); break; } @@ -2246,8 +2214,10 @@ mod tests { let mut store = MemStore::new(); // An addressed obs to [2;16] already held in the store. let mut o = obs([2; 16], 5, 0, None, b"t", b"hi"); - o.to = Some([2; 16]); - o.lifetime_ms = Some(consts::SUMMARY_INTERVAL_MS * 10); + o.dest = Some(Dest { + to: [2; 16], + lifetime_ms: consts::SUMMARY_INTERVAL_MS * 10, + }); let id = o.id(); store.insert(o, 0).unwrap(); @@ -2275,8 +2245,10 @@ mod tests { let mut store = MemStore::new(); // Addressed to an unknown node [3;16]; the only peer's id is [2;16]. let mut o = obs([2; 16], 5, 0, None, b"t", b"hi"); - o.to = Some([3; 16]); - o.lifetime_ms = Some(consts::SUMMARY_INTERVAL_MS * 10); + o.dest = Some(Dest { + to: [3; 16], + lifetime_ms: consts::SUMMARY_INTERVAL_MS * 10, + }); let id = o.id(); store.insert(o, 0).unwrap(); @@ -2302,8 +2274,10 @@ mod tests { let mut store = MemStore::new(); // A single addressed obs, received from addr 1, addressed to [3;16]. let mut o = obs([2; 16], 7, 0, None, b"t", b"hi"); - o.to = Some([3; 16]); - o.lifetime_ms = Some(consts::SUMMARY_INTERVAL_MS * 10); + o.dest = Some(Dest { + to: [3; 16], + lifetime_ms: consts::SUMMARY_INTERVAL_MS * 10, + }); let frame = Frame::Obs { from: [2; 16], obs: o, @@ -2317,7 +2291,9 @@ mod tests { let mut buf = [0u8; consts::MAX_FRAME]; match ex.next_frame(&store, consts::MAX_FRAME, 0, &mut buf) { Some((addr, n)) => match Frame::decode(&buf[..n]).unwrap() { - Frame::Obs { obs, .. } if obs.to == Some([3; 16]) && addr == 1 => { + Frame::Obs { obs, .. } + if obs.dest.as_ref().map(|d| d.to) == Some([3; 16]) && addr == 1 => + { bounced = true; break; } @@ -2340,8 +2316,10 @@ mod tests { fn addressed(observer: [u8; 16], seq: u64, to: [u8; 16], payload: &[u8]) -> Observation { let mut o = obs(observer, seq, 0, None, b"t", payload); - o.to = Some(to); - o.lifetime_ms = Some(1_000_000); + o.dest = Some(Dest { + to, + lifetime_ms: 1_000_000, + }); o } @@ -2498,7 +2476,7 @@ mod tests { let mut saw_broadcast = false; for _ in 0..700 { match poll_frame(&mut ex, &store, consts::MAX_FRAME, 0) { - Some((_, Frame::Obs { obs, .. }, _)) if obs.to.is_some() => { + Some((_, Frame::Obs { obs, .. }, _)) if obs.dest.is_some() => { got.insert(obs.id()); } Some((_, Frame::Obs { obs, .. }, _)) if obs.id() == bid => saw_broadcast = true, @@ -2924,7 +2902,7 @@ mod tests { let frame = Frame::Obs { from: peer, obs: x }; ex.on_event(Event::Frame(1, frame_bytes(&frame)), &mut store, 0); assert!(store.contains(&x_id)); - assert!(store.get(&x_id, 0).unwrap().to.is_none()); + assert!(store.get(&x_id, 0).unwrap().dest.is_none()); let ids = heapless::Vec::from_slice(&[x_id]).unwrap(); let ack = Frame::Ack { from: peer, ids }; @@ -2953,12 +2931,12 @@ mod tests { // offered. store.release(&b_id); let empty: &[(ObservationId, u8, BounceKind)] = &[]; - let found = select_addressed(&store, 0, empty, &1u8, None, |_id, obs| obs.to.is_some()); + let found = select_addressed(&store, 0, empty, &1u8, None, |_id, obs| obs.dest.is_some()); assert_eq!(found.map(|(id, _)| id), Some(a_id)); // Release the addressed id too: nothing remains selectable. store.release(&a_id); - let found = select_addressed(&store, 0, empty, &1u8, None, |_id, obs| obs.to.is_some()); + let found = select_addressed(&store, 0, empty, &1u8, None, |_id, obs| obs.dest.is_some()); assert!(found.is_none(), "an id released by an Ack is not selected"); } @@ -3189,8 +3167,10 @@ mod tests { let mut ex = Exchange::::new(ME); // Addressed to us, lifetime 100ms, but age 200ms -> expired. let mut o = obs([2; 16], 7, 200, None, b"t", b"hi"); - o.to = Some(ME); - o.lifetime_ms = Some(100); + o.dest = Some(Dest { + to: ME, + lifetime_ms: 100, + }); let frame = Frame::Obs { from: [2; 16], obs: o, diff --git a/tines-core/src/frag.rs b/tines-core/src/frag.rs index 5887f6e5..8a8dde70 100644 --- a/tines-core/src/frag.rs +++ b/tines-core/src/frag.rs @@ -217,8 +217,7 @@ 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, + dest: None, } } diff --git a/tines-core/src/lib.rs b/tines-core/src/lib.rs index 20d77b52..39ef5f30 100644 --- a/tines-core/src/lib.rs +++ b/tines-core/src/lib.rs @@ -22,7 +22,7 @@ pub use bloom::Bloom; pub use exchange::{DropReason, Exchange, NodeEvent, Peer}; pub use frag::{fragment, Reassembler}; pub use node::{Node, ObserveError}; -pub use observation::{NodeId, Observation, ObservationId, Payload, Tick, Topic}; +pub use observation::{Dest, NodeId, Observation, ObservationId, Payload, Tick, Topic}; pub use store::{Store, StoreError, TrimPolicy}; pub use transport::{Event, FrameBuf, SendError, Transport}; pub use wire::{Frame, WireError}; diff --git a/tines-core/src/mem_store.rs b/tines-core/src/mem_store.rs index a73d786d..39d9b0d0 100644 --- a/tines-core/src/mem_store.rs +++ b/tines-core/src/mem_store.rs @@ -106,7 +106,7 @@ impl Store for MemStore { fn for_each_broadcast_id(&self, f: &mut dyn FnMut(&ObservationId)) { for (id, held) in self.map.iter() { - if held.obs.to.is_none() { + if held.obs.dest.is_none() { f(id); } } @@ -116,7 +116,7 @@ impl Store for MemStore { for (id, held) in self.map.iter() { // Skipping released ids is an optimization; the exchange predicate // is the load-bearing check. - if held.obs.to.is_some() && !held.released { + if held.obs.dest.is_some() && !held.released { f(id); } } @@ -203,7 +203,7 @@ impl Store for MemStore { #[cfg(all(test, feature = "alloc"))] mod tests { use super::MemStore; - use crate::observation::{Observation, Payload, Topic}; + use crate::observation::{Dest, Observation, Payload, Topic}; use crate::store::{Store, TrimPolicy}; fn obs(observer: [u8; 16], seq: u64, age: u64, at: Option, topic: &[u8]) -> Observation { @@ -214,8 +214,7 @@ mod tests { observed_at: at, topic: Topic::from_slice(topic).unwrap(), payload: Payload::from_slice(b"").unwrap(), - to: None, - lifetime_ms: None, + dest: None, } } @@ -406,7 +405,10 @@ mod tests { let broadcast = obs([1; 16], 1, 0, None, b"door/1"); let broadcast_id = broadcast.id(); let mut addressed = obs([1; 16], 2, 0, None, b"door/1"); - addressed.to = Some([2; 16]); + addressed.dest = Some(Dest { + to: [2; 16], + lifetime_ms: 1_000_000, + }); s.insert(broadcast, 0).unwrap(); s.insert(addressed, 0).unwrap(); @@ -423,7 +425,10 @@ mod tests { // 10 addressed observations with increasing current age. for i in 0..10u64 { let mut addressed = obs([1; 16], i, (i + 1) * 10, None, b"t"); - addressed.to = Some([2; 16]); + addressed.dest = Some(Dest { + to: [2; 16], + lifetime_ms: 1_000_000, + }); s.insert(addressed, 0).unwrap(); } // One broadcast, youngest of all, so the count trim keeps it. diff --git a/tines-core/src/node.rs b/tines-core/src/node.rs index f6b30cce..7646b259 100644 --- a/tines-core/src/node.rs +++ b/tines-core/src/node.rs @@ -11,8 +11,8 @@ //! transport frames are dropped by the [`Exchange`]. use crate::{ - consts, BeliefPolicy, Exchange, NodeEvent, NodeId, Observation, ObservationId, Payload, Store, - StoreError, Tick, Topic, Transport, TrimPolicy, + consts, BeliefPolicy, Dest, Exchange, NodeEvent, NodeId, Observation, ObservationId, Payload, + Store, StoreError, Tick, Topic, Transport, TrimPolicy, }; /// Errors returned by [`Node::observe`]. @@ -91,8 +91,7 @@ impl Node { observed_at, topic, payload, - to: None, - lifetime_ms: None, + dest: None, }; self.next_seq += 1; let id = obs.id(); @@ -103,8 +102,8 @@ impl Node { /// 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 + /// The resulting observation has `dest = Some(Dest { to, 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 @@ -128,8 +127,7 @@ impl Node { observed_at, topic, payload, - to: Some(to), - lifetime_ms: Some(lifetime_ms), + dest: Some(Dest { to, lifetime_ms }), }; self.next_seq += 1; let id = obs.id(); @@ -198,7 +196,7 @@ impl Node { 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) { + if matches!(obs.dest.as_ref().map(|d| d.to), Some(other) if other != self.me) { return; } let prev = acc.take(); diff --git a/tines-core/src/observation.rs b/tines-core/src/observation.rs index 180ae802..b6f7b6cd 100644 --- a/tines-core/src/observation.rs +++ b/tines-core/src/observation.rs @@ -16,26 +16,53 @@ pub type Topic = Vec; /// Heapless, fixed-capacity payload. pub type Payload = Vec; +/// A destination for an addressed [`Observation`]: the single recipient plus +/// the deadline after which the exchange rejects it. +/// +/// Its presence is information in itself: `None` is a broadcast, `Some` is +/// addressed to exactly one node and carries a lifetime. +/// +/// [`Observation`]: struct@Observation +#[non_exhaustive] +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct Dest { + /// The one node this observation is addressed to. + pub to: NodeId, + /// The deadline in ms: the exchange rejects the observation once its + /// [`Observation::age_ms`] exceeds this value. + pub lifetime_ms: u64, +} + /// An observation published by a node, with content-derived identity. #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub struct Observation { pub observer: NodeId, pub seq: u64, + /// The observer's elapsed time for this observation, in ms. + /// + /// This is a *lower bound* on the observation's true age: it counts time + /// only while the observer's clock runs. Time the host spent powered off is + /// recovered when the store is opened (see [`Self::observed_at`]); the time + /// in an un-flushed crash window is bounded by the host's flush cadence, + /// and in-flight time is negligible. Relays never advance it. pub age_ms: u64, + /// The observer's wall-clock time, in ms since the Unix epoch (UTC), when + /// this observation was published, or `None` if the observer has no + /// trusted clock. Set once at construction and never changed by a relay. pub observed_at: Option, pub topic: Topic, pub payload: Payload, - /// Addressing recipient; `None` means broadcast. - pub to: Option, - /// Optional absolute or relative lifetime in ms; `None` means never expires. - pub lifetime_ms: Option, + /// Addressing: `None` is a broadcast, `Some(dest)` is addressed to + /// `dest.to` and expires per `dest.lifetime_ms`. + pub dest: Option, } impl Observation { /// Content-derived identity: the first 16 bytes of Blake2s-256 over /// `observer ‖ seq_le ‖ topic_len_le(u16) ‖ topic ‖ payload_len_le(u16) ‖ /// payload ‖ to_marker`, where `to_marker` is `0x00` for broadcast - /// (`to = None`) and `0x01 ‖ node_id` for `to = Some(node_id)`. + /// (`dest = None`) and `0x01 ‖ node_id` for an addressed observation + /// (`dest.map(|d| d.to) = Some(node_id)`). /// /// The hash ignores `age_ms`, `observed_at` and `lifetime_ms`, so two /// observations that differ only in age, time or lifetime are still @@ -50,10 +77,10 @@ impl Observation { 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 { + // different nodes are treated as distinct keys. `dest = None` + // (broadcast) is encoded as a single `0x00` byte, distinct from a real + // zero node id (encoded `0x01` + sixteen zero bytes). + match self.dest.as_ref().map(|d| d.to) { None => hasher.update([0x00]), Some(node_id) => { hasher.update([0x01]); @@ -75,23 +102,61 @@ impl Observation { observed_at: self.observed_at, topic: self.topic.clone(), payload: self.payload.clone(), - to: self.to, - lifetime_ms: self.lifetime_ms, + dest: self.dest.clone(), } } + + /// A broadcast observation (`dest = None`) made "now" (`age_ms = 0`) under + /// `topic`. An oversized topic or payload panics: callers that must not + /// panic (e.g. [`crate::node::Node::observe`]) validate sizes first. + pub fn broadcast(observer: NodeId, seq: u64, topic: &[u8], payload: &[u8]) -> Observation { + Observation { + observer, + seq, + age_ms: 0, + observed_at: None, + topic: Topic::from_slice(topic).expect("topic fits MAX_TOPIC"), + payload: Payload::from_slice(payload).expect("payload fits MAX_PAYLOAD"), + dest: None, + } + } + + /// An addressed observation (`dest = Some(Dest { to, lifetime_ms })`) made + /// "now" (`age_ms = 0`) under `topic`, addressed to `to` and expiring after + /// `lifetime_ms`. An oversized topic or payload panics: callers that must + /// not panic (e.g. [`crate::node::Node::observe_to`]) validate sizes first. + pub fn addressed( + observer: NodeId, + seq: u64, + topic: &[u8], + payload: &[u8], + to: NodeId, + lifetime_ms: u64, + ) -> Observation { + Observation { + observer, + seq, + age_ms: 0, + observed_at: None, + topic: Topic::from_slice(topic).expect("topic fits MAX_TOPIC"), + payload: Payload::from_slice(payload).expect("payload fits MAX_PAYLOAD"), + dest: Some(Dest { to, lifetime_ms }), + } + } + + /// Sets `observed_at` (the observer's wall-clock publish time) and returns + /// the observation, to chain from [`Self::broadcast`] / [`Self::addressed`]. + pub fn with_observed_at(mut self, at: u64) -> Observation { + self.observed_at = Some(at); + self + } } #[cfg(test)] mod tests { use super::*; - fn obs( - seq: u64, - age: u64, - at: Option, - to: Option, - lifetime_ms: Option, - ) -> Observation { + fn obs(seq: u64, age: u64, at: Option, dest: Option) -> Observation { Observation { observer: [1; 16], seq, @@ -99,41 +164,34 @@ mod tests { observed_at: at, topic: Topic::from_slice(b"door/3").unwrap(), payload: Payload::from_slice(b"open").unwrap(), - to, - lifetime_ms, + dest, } } #[test] fn id_ignores_age_and_observed_at() { - assert_eq!( - obs(1, 0, None, None, None).id(), - obs(1, 999, Some(5), None, None).id() - ); + assert_eq!(obs(1, 0, None, None).id(), obs(1, 999, Some(5), None).id()); } #[test] fn id_changes_with_seq() { - assert_ne!( - obs(1, 0, None, None, None).id(), - obs(2, 0, None, None, None).id() - ); + assert_ne!(obs(1, 0, None, None).id(), obs(2, 0, None, None).id()); } #[test] fn id_changes_with_observer() { - let mut b = obs(1, 0, None, None, None); + let mut b = obs(1, 0, None, None); b.observer = [2; 16]; - assert_ne!(obs(1, 0, None, None, None).id(), b.id()); + assert_ne!(obs(1, 0, 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, None, None); + let mut x = obs(1, 0, 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, None, None); + let mut y = obs(1, 0, None, None); y.topic = Topic::from_slice(b"a").unwrap(); y.payload = Payload::from_slice(b"bc").unwrap(); assert_ne!(x.id(), y.id()); @@ -141,51 +199,157 @@ mod tests { #[test] fn aged_saturates() { - assert_eq!( - obs(1, u64::MAX - 1, None, None, None).aged(10).age_ms, - u64::MAX - ); + assert_eq!(obs(1, u64::MAX - 1, 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 + // same observation, broadcast vs addressed to Some([2;16]) — must differ assert_ne!( - obs(7, 0, None, None, None).id(), - obs(7, 0, None, Some([2; 16]), None).id() + obs(7, 0, None, None).id(), + obs( + 7, + 0, + None, + Some(Dest { + to: [2; 16], + lifetime_ms: 0 + }) + ) + .id() ); } #[test] fn lifetime_is_not_part_of_identity() { - // same payload, both to = Some([2;16]), lifetime 100 vs 200 — must be equal + // same payload, both addressed 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() + obs( + 7, + 0, + None, + Some(Dest { + to: [2; 16], + lifetime_ms: 100 + }) + ) + .id(), + obs( + 7, + 0, + None, + Some(Dest { + to: [2; 16], + lifetime_ms: 200 + }) + ) + .id() ); } #[test] fn broadcast_marker_distinct_from_zero_id() { - // broadcast marker (to = None) must not collide with a real zero node id + // broadcast marker (dest = 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() + obs(7, 0, None, None).id(), + obs( + 7, + 0, + None, + Some(Dest { + to: [0; 16], + lifetime_ms: 0 + }) + ) + .id() ); } #[test] fn aged_preserves_addressing() { - let aged = obs(1, 0, None, Some([2; 16]), Some(500)).aged(10); + let aged = obs( + 1, + 0, + None, + Some(Dest { + to: [2; 16], + lifetime_ms: 500, + }), + ) + .aged(10); assert_eq!(aged.age_ms, 10); - assert_eq!(aged.to, Some([2; 16])); - assert_eq!(aged.lifetime_ms, Some(500)); + assert_eq!(aged.dest.as_ref().map(|d| d.to), Some([2; 16])); + assert_eq!(aged.dest.as_ref().map(|d| d.lifetime_ms), Some(500)); + } + + #[test] + fn dest_invariant_by_type() { + // A broadcast observation never carries addressing... + let b = Observation::broadcast([1; 16], 1, b"t", b"p"); + assert_eq!(b.dest, None); + // ...and an addressed observation *always* carries a `dest` (this is + // the invariant the type now enforces — no more to/xor-lifetime pairs). + let o = Observation::addressed([1; 16], 1, b"t", b"p", [2; 16], 5000); + assert_eq!( + o.dest, + Some(Dest { + to: [2; 16], + lifetime_ms: 5000 + }), + "addressed must always set dest" + ); + } + + #[test] + fn id_reads_dest_to() { + // id() reads the recipient out of `dest`; the lifetime is still not + // part of identity. + let x = Observation::addressed([1; 16], 7, b"t", b"p", [2; 16], 100); + let y = Observation::addressed([1; 16], 7, b"t", b"p", [2; 16], 200); + assert_eq!(x.id(), y.id()); + } + + #[test] + fn constructors_set_age_zero() { + let b = Observation::broadcast([1; 16], 1, b"t", b"p"); + assert_eq!(b.age_ms, 0); + assert_eq!(b.observed_at, None); + assert_eq!(b.dest, None); + + let a = Observation::addressed([1; 16], 1, b"t", b"p", [2; 16], 5000); + assert_eq!(a.age_ms, 0); + assert_eq!(a.observed_at, None); + assert_eq!( + a.dest, + Some(Dest { + to: [2; 16], + lifetime_ms: 5000 + }) + ); + } + + #[test] + fn with_observed_at_sets_field() { + let b = Observation::broadcast([1; 16], 1, b"t", b"p").with_observed_at(1234); + assert_eq!(b.observed_at, Some(1234)); + assert_eq!(b.dest, None); + + let a = + Observation::addressed([1; 16], 1, b"t", b"p", [2; 16], 5000).with_observed_at(1234); + assert_eq!(a.observed_at, Some(1234)); + assert_eq!( + a.dest, + Some(Dest { + to: [2; 16], + lifetime_ms: 5000 + }) + ); } #[test] fn id_known_vector() { - // pin the hash layout: observer=[1;16], seq=7, topic="t", payload="p", to=None - let mut o = obs(7, 0, None, None, None); + // pin the hash layout: observer=[1;16], seq=7, topic="t", payload="p", dest=None (broadcast) + let mut o = obs(7, 0, None, None); o.topic = Topic::from_slice(b"t").unwrap(); o.payload = Payload::from_slice(b"p").unwrap(); assert_eq!( diff --git a/tines-core/src/store.rs b/tines-core/src/store.rs index abc25b11..40aff969 100644 --- a/tines-core/src/store.rs +++ b/tines-core/src/store.rs @@ -61,14 +61,14 @@ pub trait Store { fn for_each_id(&self, f: &mut dyn FnMut(&ObservationId)); /// Visit the ids of observations that are not addressed to anyone - /// (`to == None`), i.e. broadcasts. Order is unspecified. + /// (`dest` is `None`), i.e. broadcasts. Order is unspecified. fn for_each_broadcast_id(&self, f: &mut dyn FnMut(&ObservationId)); - /// Visit the ids of observations addressed to someone (`to == Some`). + /// Visit the ids of observations addressed to someone (`dest` is `Some`). /// Order is unspecified. /// /// The default visits every id, which is correct since callers still - /// check `to`; backends that can filter cheaply should override it so the + /// check `dest`; backends that can filter cheaply should override it so the /// exchange does not read every broadcast on each poll. fn for_each_addressed_id(&self, f: &mut dyn FnMut(&ObservationId)) { self.for_each_id(f) diff --git a/tines-core/src/wire.rs b/tines-core/src/wire.rs index 5d6a80bc..93b9b173 100644 --- a/tines-core/src/wire.rs +++ b/tines-core/src/wire.rs @@ -139,8 +139,7 @@ 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, + dest: None, }, } } @@ -155,8 +154,7 @@ mod tests { observed_at: None, topic: Topic::from_slice(b"t").unwrap(), payload: Payload::from_slice(&[0u8; 1024]).unwrap(), - to: None, - lifetime_ms: None, + dest: None, }, } } diff --git a/tines-core/tests/pair.rs b/tines-core/tests/pair.rs index 927ab0f6..3f94763d 100644 --- a/tines-core/tests/pair.rs +++ b/tines-core/tests/pair.rs @@ -236,8 +236,8 @@ fn observe_to_creates_addressed_message() { 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.dest.as_ref().map(|d| d.to), Some([2; 16])); + assert_eq!(obs.dest.as_ref().map(|d| d.lifetime_ms), Some(5000)); assert_eq!(obs.age_ms, 0); } diff --git a/tines-store/src/log.rs b/tines-store/src/log.rs index fc87d842..44cdf203 100644 --- a/tines-store/src/log.rs +++ b/tines-store/src/log.rs @@ -145,7 +145,7 @@ mod crc32c { /// Magic header: the first 8 bytes of every v1 log. const MAGIC: [u8; 8] = *b"TINESLOG"; /// The `format_version` field of the header. Bump for an incompatible encoding. -const FORMAT_VERSION: u16 = 1; +const FORMAT_VERSION: u16 = 2; /// 4-byte record sync marker (`b"TING"`), lets a torn/corrupt record be /// located and skipped by scanning forward. Made `pub` so hosts/tests can emit /// (and corrupt) framing directly. @@ -502,7 +502,7 @@ impl LogStore { } } -/// Write the v1 header: `magic ‖ format_version ‖ reserved`. +/// Write the log header: `magic ‖ format_version ‖ reserved`. fn write_header(file: &mut File) -> Result<(), StoreError> { file.write_all(&MAGIC).map_err(|_| StoreError::Backend)?; file.write_all(&FORMAT_VERSION.to_le_bytes()) @@ -516,9 +516,10 @@ fn write_header(file: &mut File) -> Result<(), StoreError> { /// naming the fix (delete the log) in the message. fn reject_bad_header(path: &Path, reason: &str) -> Result<(), StoreError> { eprintln!( - "tines-store: log header invalid ({reason}): {} is not a v1 \ + "tines-store: log header invalid ({reason}): {} is not a v{} \ 'TINESLOG' log — delete it and start fresh (fleet-fixes F5).", - path.display() + path.display(), + FORMAT_VERSION, ); Err(StoreError::Backend) } @@ -533,7 +534,14 @@ fn validate_header(data: &[u8], path: &Path) -> Result<(), StoreError> { } let version = u16::from_le_bytes([data[8], data[9]]); if version != FORMAT_VERSION { - return reject_bad_header(path, &format!("version {version} != {FORMAT_VERSION}")); + return reject_bad_header( + path, + &format!( + "on-disk format_version {version} != {FORMAT_VERSION}: the \ + Observation encoding changed in 1.0, so this log can no longer \ + be decoded" + ), + ); } Ok(()) } @@ -735,22 +743,22 @@ impl Store for LogStore { fn for_each_broadcast_id(&self, f: &mut dyn FnMut(&ObservationId)) { for (id, entry) in self.index.iter() { - if entry.obs.to.is_none() { + if entry.obs.dest.is_none() { f(id); } } } - /// Visit the ids of observations addressed to someone (`to == Some`), + /// Visit the ids of observations addressed to someone (`dest` is `Some`), /// excluding those already released. /// /// This is stricter than the [`Store`] trait's documented contract - /// (`addressed == to == Some`): a released addressed observation has + /// (`addressed` == `dest` is `Some`): a released addressed observation has /// already been sent to its recipient, so it is intentionally skipped and /// not offered for exchange again until (and if) it is re-inserted. fn for_each_addressed_id(&self, f: &mut dyn FnMut(&ObservationId)) { for (id, entry) in self.index.iter() { - if entry.obs.to.is_some() && !entry.released { + if entry.obs.dest.is_some() && !entry.released { f(id); } } @@ -1015,8 +1023,7 @@ mod tests { observed_at: at, topic: tines_core::Topic::from_slice(b"door/3").unwrap(), payload: tines_core::Payload::from_slice(b"open").unwrap(), - to: None, - lifetime_ms: None, + dest: None, } } diff --git a/tines-store/tests/persistence.rs b/tines-store/tests/persistence.rs index ac37fdbc..4a7ebf52 100644 --- a/tines-store/tests/persistence.rs +++ b/tines-store/tests/persistence.rs @@ -20,8 +20,7 @@ fn obs(observer: [u8; 16], seq: u64, age: u64, at: Option) -> Observation { observed_at: at, topic: Topic::from_slice(b"door/3").unwrap(), payload: Payload::from_slice(b"open").unwrap(), - to: None, - lifetime_ms: None, + dest: None, } } @@ -42,6 +41,28 @@ fn store_path(tag: &str) -> PathBuf { p } +/// A log whose header declares an older `format_version` must be refused on +/// open. The 1.0 work changed the Observation encoding (the `dest` field +/// replaces the old `to`/`lifetime_ms` pair), so a v1 log's records can no +/// longer be decoded: `open` fails and names the fix (delete the log). +#[test] +fn old_format_is_refused() { + let path = store_path("oldfmt"); + { + let mut f = std::fs::File::create(&path).unwrap(); + // magic ‖ format_version (v1, little-endian) ‖ reserved: a genuine v1 + // header that this build (v2) no longer accepts. + f.write_all(b"TINESLOG").unwrap(); + f.write_all(&1u16.to_le_bytes()).unwrap(); + f.write_all(&0u16.to_le_bytes()).unwrap(); + f.flush().unwrap(); + } + assert!( + LogStore::open(&path, 0).is_err(), + "open must refuse a log whose format_version is older than this build" + ); +} + #[test] fn reopen_preserves_store() { let path = store_path("reopen"); @@ -324,8 +345,7 @@ mod fleet_fixes_f5 { observed_at: Some(10), topic: Topic::from_slice(b"door/3").unwrap(), payload: Payload::from_slice(b"open").unwrap(), - to: None, - lifetime_ms: None, + dest: None, } } @@ -333,7 +353,7 @@ mod fleet_fixes_f5 { /// mirroring on-disk corruption of a complete record. fn corrupt_record(path: &Path, n: usize) { let mut data = std::fs::read(path).expect("read log"); - let mut pos = 12usize; // past the v1 header + let mut pos = 12usize; // past the header let mut i = 0; while i < n { assert!(pos + 12 <= data.len(), "record {i} out of range");