kyle/tines
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(core): Dest type replaces to + lifetime_ms

Browse Source
This commit is contained in:
kyle
2026-10-22 08:44:09 -07:00
parent 193c7bddb8
commit e85b3704e0
12 changed files with 368 additions and 198 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tines-core/src/belief.rs
+1 -2
View File
@@ -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,
}
}
tines-core/src/exchange.rs
+82 -102
View File
@@ -703,9 +703,9 @@ impl<A: Clone + PartialEq + core::fmt::Debug> Exchange<A> {
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<A: Clone + PartialEq + core::fmt::Debug> Exchange<A> {
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<A: Clone + PartialEq + core::fmt::Debug> Exchange<A> {
// 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<A: Clone + PartialEq + core::fmt::Debug> Exchange<A> {
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<usize> {
}
/// 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<S: Store, A: PartialEq>(
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<heapless::Vec<u8, { consts::MAX_FRAME }>> = 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::<u8>::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,
tines-core/src/frag.rs
+1 -2
View File
@@ -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,
}
}
tines-core/src/lib.rs
+1 -1
View File
@@ -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};
tines-core/src/mem_store.rs
+12 -7
View File
@@ -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<u64>, 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.
tines-core/src/node.rs
+7 -9
View File
@@ -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<S: Store, T: Transport> Node<S, T> {
observed_at,
topic,
payload,
to: None,
lifetime_ms: None,
dest: None,
};
self.next_seq += 1;
let id = obs.id();
@@ -103,8 +102,8 @@ impl<S: Store, T: Transport> Node<S, T> {
/// 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<S: Store, T: Transport> Node<S, T> {
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<S: Store, T: Transport> Node<S, T> {
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();
tines-core/src/observation.rs
+214 -50
View File
@@ -16,26 +16,53 @@ pub type Topic = Vec<u8, { consts::MAX_TOPIC }>;
/// Heapless, fixed-capacity payload.
pub type Payload = Vec<u8, { consts::MAX_PAYLOAD }>;
/// 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<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>,
/// Addressing: `None` is a broadcast, `Some(dest)` is addressed to
/// `dest.to` and expires per `dest.lifetime_ms`.
pub dest: Option<Dest>,
}
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<u64>,
to: Option<NodeId>,
lifetime_ms: Option<u64>,
) -> Observation {
fn obs(seq: u64, age: u64, at: Option<u64>, dest: Option<Dest>) -> 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!(
tines-core/src/store.rs
+3 -3
View File
@@ -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)
tines-core/src/wire.rs
+2 -4
View File
@@ -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,
},
}
}
tines-core/tests/pair.rs
+2 -2
View File
@@ -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);
}
tines-store/src/log.rs
+18 -11
View File
@@ -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,
}
}
tines-store/tests/persistence.rs
+25 -5
View File
@@ -20,8 +20,7 @@ fn obs(observer: [u8; 16], seq: u64, age: u64, at: Option<u64>) -> 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");