feat(core): Node facade

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};

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<S: Store, T: Transport> {
+    me: NodeId,
+    next_seq: u64,
+    store: S,
+    transport: T,
+    exchange: Exchange<T::Addr>,
+    out: [u8; consts::MAX_FRAME],
+}
+
+impl<S: Store, T: Transport> Node<S, T> {
+    /// 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<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,
+        };
+        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<P: BeliefPolicy>(
+        &self,
+        topic_prefix: &[u8],
+        policy: &P,
+        now: Tick,
+    ) -> Option<P::Belief> {
+        let mut acc: Option<P::Belief> = 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
+    }
+}

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);
+}