feat(core): Node::drain_events for confidence events
This commit is contained in:
+14
-2
@@ -11,8 +11,8 @@
|
|||||||
//! transport frames are dropped by the [`Exchange`].
|
//! transport frames are dropped by the [`Exchange`].
|
||||||
|
|
||||||
use crate::{
|
use crate::{
|
||||||
consts, BeliefPolicy, Exchange, NodeId, Observation, ObservationId, Payload, Store, StoreError,
|
consts, BeliefPolicy, ConfidenceEvent, Exchange, NodeId, Observation, ObservationId, Payload,
|
||||||
Tick, Topic, Transport, TrimPolicy,
|
Store, StoreError, Tick, Topic, Transport, TrimPolicy,
|
||||||
};
|
};
|
||||||
|
|
||||||
/// Errors returned by [`Node::observe`].
|
/// Errors returned by [`Node::observe`].
|
||||||
@@ -165,6 +165,18 @@ impl<S: Store, T: Transport> Node<S, T> {
|
|||||||
stored
|
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<ConfidenceEvent, { consts::EVENT_QUEUE_CAP }> {
|
||||||
|
self.exchange.drain_events()
|
||||||
|
}
|
||||||
|
|
||||||
/// The current belief for `topic_prefix` under `policy` (an empty prefix
|
/// The current belief for `topic_prefix` under `policy` (an empty prefix
|
||||||
/// matches all topics).
|
/// matches all topics).
|
||||||
///
|
///
|
||||||
|
|||||||
@@ -357,3 +357,37 @@ fn busy_sends_are_retried() {
|
|||||||
assert_eq!(na.store().len(), 4);
|
assert_eq!(na.store().len(), 4);
|
||||||
assert_eq!(nb.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"
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user