diff --git a/tines-core/src/lib.rs b/tines-core/src/lib.rs index 25a5d59..a2e41e7 100644 --- a/tines-core/src/lib.rs +++ b/tines-core/src/lib.rs @@ -13,6 +13,7 @@ pub mod observation; pub mod store; #[cfg(feature = "alloc")] pub mod mem_store; +pub mod node; pub mod transport; pub mod wire; @@ -20,6 +21,7 @@ pub use belief::{BeliefPolicy, LatestWins, SelfFirst}; pub use bloom::Bloom; pub use exchange::{Exchange, Peer}; pub use frag::{fragment, Reassembler}; +pub use node::{Node, ObserveError}; pub use observation::{Observation, NodeId, ObservationId, Payload, Tick, Topic}; pub use store::{Store, TrimPolicy, StoreError}; pub use transport::{Event, FrameBuf, SendError, Transport}; diff --git a/tines-core/src/node.rs b/tines-core/src/node.rs new file mode 100644 index 0000000..7495ee3 --- /dev/null +++ b/tines-core/src/node.rs @@ -0,0 +1,160 @@ +//! Node facade for tines-core. +//! +//! [`Node`] is the single entry point that ties the [`Store`], the +//! [`Exchange`] state machine, and a [`Transport`] together behind the two +//! operations an application needs: [`Node::observe`] to record an observation, +//! and [`Node::poll`] to drive the exchange. +//! +//! The facade never touches the heap: its only owned buffer, `out`, is a fixed +//! `[u8; MAX_FRAME]`. It also panics on nothing — oversized topics/payloads +//! become [`ObserveError::TooLarge`] rather than truncating, and malformed +//! transport frames are dropped by the [`Exchange`]. + +use crate::{ + BeliefPolicy, consts, Exchange, NodeId, Observation, ObservationId, Payload, Store, StoreError, + TrimPolicy, Tick, Topic, Transport, +}; + +/// Errors returned by [`Node::observe`]. +#[derive(Debug, PartialEq, Eq)] +pub enum ObserveError { + /// The topic or payload exceeded `MAX_TOPIC`/`MAX_PAYLOAD`. The + /// observation is rejected whole and never truncated. + TooLarge, + /// The underlying [`Store`] rejected the observation (e.g. it was full). + Store(StoreError), +} + +/// The application-facing facade over the store, exchange, and transport. +/// +/// `Node` is generic over any [`Store`] and [`Transport`], so a concrete +/// backend (heap-backed, `no_std`, ...) can be plugged in without changing the +/// facade. The exchange tracks the peer this node meets on `transport`. +pub struct Node { + me: NodeId, + next_seq: u64, + store: S, + transport: T, + exchange: Exchange, + out: [u8; consts::MAX_FRAME], +} + +impl Node { + /// Creates a node for local id `me`, starting observation sequencing at + /// `first_seq`. + /// + /// Persist [`Node::next_seq`] (e.g. `boot_count << 32 | counter`) across + /// reboots and pass it back as `first_seq`; a repeated seq after reboot + /// collides in the content id hash and dedupes silently. + pub fn new(me: NodeId, first_seq: u64, store: S, transport: T) -> Self { + Node { + me, + next_seq: first_seq, + store, + transport, + exchange: Exchange::new(me), + out: [0u8; consts::MAX_FRAME], + } + } + + /// The local node id. + pub fn id(&self) -> NodeId { + self.me + } + + /// The seq the next [`Node::observe`] will use. + /// + /// The application must persist this across reboots (see [`Node::new`]). + pub fn next_seq(&self) -> u64 { + self.next_seq + } + + /// Records an observation made "now" (with `age_ms = 0`) under `topic`. + /// + /// 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( + &mut self, + topic: &[u8], + payload: &[u8], + observed_at: Option, + now: Tick, + ) -> Result { + 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, + }; + 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. + /// + /// Returns the number of *new* observations stored this round. A + /// [`tines_core::transport::SendError`] is ignored: the observation stays + /// queued only when the transport was `Busy`, otherwise it is dropped and + /// re-queued by the next summary round. + pub fn poll(&mut self, now: Tick) -> usize { + let mut stored = 0usize; + while let Some(ev) = self.transport.poll() { + if self.exchange.on_event(ev, &mut self.store, now).is_some() { + stored += 1; + } + } + // Hoist the MTU, then borrow `exchange`/`store`, then borrow + // `transport` for the send — disjoint field borrows, never overlapping. + let mtu = self.transport.mtu(); + if let Some((addr, n)) = self.exchange.next_frame(&self.store, mtu, now, &mut self.out) { + let _ = self.transport.send(&addr, &self.out[..n]); + } + stored + } + + /// The current belief for `topic_prefix` under `policy` (an empty prefix + /// matches all topics). + /// + /// Streams through `Store::for_each_topic`, threading an accumulator + /// through [`BeliefPolicy::step`] — no buffering of the topic's + /// observations. + pub fn belief( + &self, + topic_prefix: &[u8], + policy: &P, + now: Tick, + ) -> Option { + let mut acc: Option = None; + self.store + .for_each_topic(topic_prefix, now, &mut |obs| { + let prev = acc.take(); + acc = policy.step(self.me, prev, obs); + }); + acc + } + + /// Applies `policy` to the store and returns the number of observations + /// removed. + pub fn trim(&mut self, policy: &TrimPolicy, now: Tick) -> usize { + self.store.trim(policy, now) + } + + /// Access the backing store. + pub fn store(&self) -> &S { + &self.store + } + + /// Access the transport (e.g. to `connect`/`disconnect` a peer in tests). + pub fn transport_mut(&mut self) -> &mut T { + &mut self.transport + } +} diff --git a/tines-core/tests/pair.rs b/tines-core/tests/pair.rs index ad9d93a..6e564ae 100644 --- a/tines-core/tests/pair.rs +++ b/tines-core/tests/pair.rs @@ -1,7 +1,10 @@ mod common; use common::PairTransport; +use tines_core::observation::Payload; use tines_core::transport::{Event, FrameBuf, Transport}; +use tines_core::{consts, LatestWins, MemStore, Node, ObserveError, Store}; +use std::collections::BTreeSet; #[test] fn pair_transport_connect_then_frame() { @@ -14,3 +17,181 @@ fn pair_transport_connect_then_frame() { Some(Event::Frame(0, FrameBuf::from_slice(b"hi").unwrap())) ); } + +#[test] +fn observe_assigns_increasing_seq() { + let (a, _b) = PairTransport::pair(1500); + let mut node = Node::new([1u8; 16], 10, MemStore::new(), a); + node.observe(b"door/1", b"open", None, 0).unwrap(); + node.observe(b"door/2", b"closed", None, 0).unwrap(); + // next_seq advances by one per observe(): first_seq + 2. + assert_eq!(node.next_seq(), 12); +} + +#[test] +fn observe_rejects_oversized_topic_and_payload() { + let (a, _b) = PairTransport::pair(1500); + let mut node = Node::new([1u8; 16], 0, MemStore::new(), a); + + // Review Focus 2: oversized topic and payload are rejected whole. + let oversized_topic = vec![0u8; consts::MAX_TOPIC + 1]; + assert_eq!( + node.observe(&oversized_topic, b"ok", None, 0), + Err(ObserveError::TooLarge) + ); + + let oversized_payload = vec![0u8; consts::MAX_PAYLOAD + 1]; + assert_eq!( + node.observe(b"ok", &oversized_payload, None, 0), + Err(ObserveError::TooLarge) + ); + + // Neither was stored; the store is unchanged, nothing truncated. + assert_eq!(node.store().len(), 0); +} + +#[test] +fn two_nodes_converge_over_pair_transport() { + let (a, b) = PairTransport::pair(1500); + let mut na = Node::new([1u8; 16], 0, MemStore::new(), a); + let mut nb = Node::new([2u8; 16], 0, MemStore::new(), b); + + na.observe(b"door/1", b"open", None, 0).unwrap(); + na.observe(b"door/2", b"closed", None, 0).unwrap(); + na.observe(b"bat/1", b"up", None, 0).unwrap(); + nb.observe(b"door/3", b"open", None, 0).unwrap(); + nb.observe(b"bat/2", b"down", None, 0).unwrap(); + + na.transport_mut().connect(); + nb.transport_mut().connect(); + + let mut t = 0u64; + for _ in 0..1000 { + na.poll(t); + nb.poll(t); + t += 10; + } + + assert_eq!(na.store().len(), 5); + assert_eq!(nb.store().len(), 5); + + let mut ids_a = BTreeSet::new(); + na.store().for_each_id(&mut |id: &[u8; 16]| { + ids_a.insert(*id); + }); + let mut ids_b = BTreeSet::new(); + nb.store().for_each_id(&mut |id: &[u8; 16]| { + ids_b.insert(*id); + }); + assert_eq!(ids_a, ids_b, "both stores must hold identical id sets"); +} + +#[test] +fn belief_latest_wins_across_nodes() { + let (a, b) = PairTransport::pair(1500); + let mut na = Node::new([1u8; 16], 0, MemStore::new(), a); + let mut nb = Node::new([2u8; 16], 0, MemStore::new(), b); + + // Same topic "door", different observers. a is older, b is younger. + na.observe(b"door", b"early", Some(0), 0).unwrap(); + nb.observe(b"door", b"late", Some(500), 500).unwrap(); + + na.transport_mut().connect(); + nb.transport_mut().connect(); + + let mut t = 0u64; + for _ in 0..1000 { + na.poll(t); + nb.poll(t); + t += 10; + } + + let now = 100_000u64; + // LatestWins must pick b's younger observation on both nodes. Local + // observe stores at `now` (b's stored_at=500, a's stored_at=0), so b's + // is younger everywhere it is held; a re-received copy has a slightly + // different age, so compare by identity, not byte-equality. + let belief = nb.belief(b"door", &LatestWins, now).expect("door belief exists"); + assert_eq!(belief.observer, [2; 16], "nb belief should be b's observation"); + assert_eq!(belief.seq, 0); + assert_eq!(belief.payload, Payload::from_slice(b"late").unwrap()); + assert_eq!(belief.observed_at, Some(500)); + + // a's node, looking at the same topic, agrees on the same observation. + let belief_a = na.belief(b"door", &LatestWins, now).unwrap(); + assert_eq!(belief_a.observer, [2; 16], "na belief should be b's observation"); + assert_eq!(belief_a.payload, Payload::from_slice(b"late").unwrap()); + assert_eq!(belief_a.observed_at, Some(500)); +} + +#[test] +fn cut_off_mid_exchange_leaves_both_valid_and_resumes() { + let (a, b) = PairTransport::pair(1500); + let mut na = Node::new([1u8; 16], 0, MemStore::new(), a); + let mut nb = Node::new([2u8; 16], 0, MemStore::new(), b); + + // a publishes 200 observations with 1 KiB payloads each. + let big = vec![0u8; 1024]; + for _ in 0..200 { + na.observe(b"big", &big, None, 0).unwrap(); + } + + na.transport_mut().connect(); + nb.transport_mut().connect(); + + let mut t = 0u64; + for _ in 0..20 { + na.poll(t); + nb.poll(t); + t += 10; + } + + // Cut the link mid-exchange, then reconnect. + na.transport_mut().disconnect(); + nb.transport_mut().disconnect(); + na.transport_mut().connect(); + nb.transport_mut().connect(); + + let mut done = false; + for _ in 0..20000 { + na.poll(t); + nb.poll(t); + t += 10; + if nb.store().len() == 200 { + done = true; + break; + } + } + assert!(done, "exchange must resume after reconnect"); + assert_eq!(nb.store().len(), 200); +} + +#[test] +fn third_party_observation_propagates_within_contact() { + let (a, b) = PairTransport::pair(1500); + let mut na = Node::new([1u8; 16], 0, MemStore::new(), a); + let mut nb = Node::new([2u8; 16], 0, MemStore::new(), b); + + na.transport_mut().connect(); + nb.transport_mut().connect(); + + let mut t = 0u64; + for _ in 0..1000 { + na.poll(t); + nb.poll(t); + t += 10; + } + // Converged (both empty), still in contact. + + // Inject a fresh observation into a. + let id = na.observe(b"fresh", b"hi", None, t).unwrap(); + + // Within ~SUMMARY_INTERVAL_MS*2 of polling, b must have received it. + for _ in 0..400 { + na.poll(t); + nb.poll(t); + t += 10; + } + assert!(nb.store().contains(&id), "fresh observation must propagate to b"); + assert_eq!(nb.store().len(), 1); +}