diff --git a/tines-core/src/node.rs b/tines-core/src/node.rs index 5f4c84c7..39183d70 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, NodeId, Observation, ObservationId, Payload, Store, StoreError, - Tick, Topic, Transport, TrimPolicy, + consts, BeliefPolicy, ConfidenceEvent, Exchange, NodeId, Observation, ObservationId, Payload, + Store, StoreError, Tick, Topic, Transport, TrimPolicy, }; /// Errors returned by [`Node::observe`]. @@ -165,6 +165,18 @@ impl Node { stored } + /// Drains the buffered [`ConfidenceEvent`]s the [`Exchange`] has collected + /// as a peer acknowledged holding one of our broadcasts, oldest-first. + /// + /// The returned [`heapless::Vec`] is capped at + /// [`consts::EVENT_QUEUE_CAP`] entries and is empty once drained (a second + /// drain yields nothing). Index it directly — [`ConfidenceEvent::peer`] + /// names who received our broadcast and [`ConfidenceEvent::id`] is the + /// broadcast's content-ID. + pub fn drain_events(&mut self) -> heapless::Vec { + self.exchange.drain_events() + } + /// The current belief for `topic_prefix` under `policy` (an empty prefix /// matches all topics). /// diff --git a/tines-core/tests/pair.rs b/tines-core/tests/pair.rs index 97f789b2..c317b942 100644 --- a/tines-core/tests/pair.rs +++ b/tines-core/tests/pair.rs @@ -357,3 +357,37 @@ fn busy_sends_are_retried() { assert_eq!(na.store().len(), 4); assert_eq!(nb.store().len(), 4); } + +#[test] +fn confidence_event_surfaces_on_node() { + 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 a broadcast; b receives it and its Ack fires a confidence + // event that must surface on a's node. + let id = na.observe(b"door/1", b"open", None, 0).unwrap(); + + na.transport_mut().connect(); + nb.transport_mut().connect(); + + // Poll until b holds the observation, then give b's Ack a few more rounds + // to round-trip back to a and fire the confidence event. + let mut t = 0u64; + for _ in 0..200 { + na.poll(t); + nb.poll(t); + t += 10; + } + assert!( + nb.store().contains(&id), + "b must receive a's broadcast before an Ack fires" + ); + + // a's node surfaces the confidence event: peer == b, id == the broadcast. + let events = na.drain_events(); + assert!( + events.iter().any(|ev| ev.peer == nb.id() && ev.id == id), + "a's node must surface the confidence event for b's Ack" + ); +}