diff --git a/CHANGELOG.md b/CHANGELOG.md index 61f1f01c..b03e7c45 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -19,7 +19,19 @@ still 0.3.0. - An ack for an addressed observation moves custody: the sender releases its copy (`Store::release`) and never offers it again. - An ack for a broadcast is confidence only. It surfaces as a - `ConfidenceEvent { peer, id }` from the new `Node::drain_events`. + `NodeEvent::Delivered { peer, id }` from the new `Node::drain_events`. + - Broadcast confirmations and counted drops both come through + `Node::drain_events`; a confirmed broadcast arrives as + `NodeEvent::Delivered`, and a frame the peer dropped arrives as + `NodeEvent::Dropped { reason, count }` (§3). +- **Counted drop reporting** (tines-core). A frame the peer could not use is no + dropped silently: the exchange counts it under a `DropReason` (`VersionMismatch` + or `Malformed`) keyed by the peer's transport address and reports the count + through `Node::drain_events` as `NodeEvent::Dropped`. Counters are held in a + bounded table (`consts::DROP_TABLE_CAP`) with a single overflow slot for the + case where the table is full, so an adversarial peer can't spin forever — it + just reports `addr: None`. Expired addressed observations dropped by policy are + *not* counted. - New constants: `ACK_IDS_CAP`, `EVENT_QUEUE_CAP`, `BCAST_SENT_CAP`, `ACK_BUFFER_CAP`. - `wire::frame_sender(bytes) -> Option`: a frame's claimed sender, diff --git a/README.md b/README.md index 85992f75..68f73e8a 100644 --- a/README.md +++ b/README.md @@ -294,9 +294,13 @@ example, because its store is full), and never acks an addressed duplicate it has already released, since that would hand custody back to a robot that no longer carries the message. -Broadcast confirmations surface as `ConfidenceEvent { peer, id }` from -`Node::drain_events`. The event queue is bounded (64): events are telemetry, -and losing one never loses data. +Broadcast confirmations surface as `NodeEvent::Delivered { peer, id }` from +`Node::drain_events`. Frames the peer could not use are counted under a +`DropReason` (`VersionMismatch` or `Malformed`) and surface as +`NodeEvent::Dropped { reason, count }` (see the `drop` reporting in the `tines` +service and the ROS2 node). The event queue is bounded (64) plus a small +counted-drop table, so drops are always reported even under a chatty peer — +events are telemetry, and losing one never loses data. ## Usage @@ -333,9 +337,17 @@ let stored = node.poll(500); // (e.g. the youngest observation wins). let belief = node.belief(b"door", &LatestWins, 500); -// Which peers have confirmed holding which of your broadcasts. +// Which peers have confirmed holding which of your broadcasts, and which +// frames the peer dropped and counted. for event in node.drain_events() { - println!("peer {:?} has {:?}", event.peer, event.id); + match event { + tines_core::NodeEvent::Delivered { peer, id } => { + println!("peer {peer:?} has {id:?}"); + } + tines_core::NodeEvent::Dropped { addr, reason, count } => { + println!("dropped {count} frame(s) from {addr:?}: {reason:?}"); + } + } } // Keep the store bounded. @@ -521,7 +533,7 @@ A robot runs either the service or the ROS2 node, not both: each owns its - It's configured by a JSON file: `node_id`, `peer`, `log_path`, `socket_path` and `poll_interval_ms`. - **`tines-ros2`**: a ROS2 node. - - It publishes each tines topic under `/tines/` and confidence events on + - It publishes each tines topic under `/tines/` and confirmed deliveries on `/tines/receipts`. - It turns messages on those topics into observations. - rclrs is behind the `ros2` feature, so the crate builds without a ROS2 @@ -569,7 +581,7 @@ Fixed limits, all compile-time constants in `tines-core/src/consts.rs`: | `SEND_QUEUE_CAP` | 256 | ids queued per peer | | `MAX_FRAME` | 1500 | largest frame (and MTU) supported | | `ACK_IDS_CAP` | 90 | ids in one `Ack` frame | -| `EVENT_QUEUE_CAP` | 64 | confidence events awaiting `drain_events`| +| `EVENT_QUEUE_CAP` | 64 | confirmed deliveries awaiting `drain_events` | | `BCAST_SENT_CAP` | 256 | broadcast sends remembered per peer, to validate acks | | `ACK_BUFFER_CAP` | 256 | received ids waiting to be acked, per kind | diff --git a/tines-core/src/consts.rs b/tines-core/src/consts.rs index 29012322..ccb71a76 100644 --- a/tines-core/src/consts.rs +++ b/tines-core/src/consts.rs @@ -42,10 +42,15 @@ pub const MAX_FRAME: usize = 1500; /// Maximum number of content-IDs an `Ack` receipt may carry. pub const ACK_IDS_CAP: usize = 90; -/// Cap on the `Exchange` `events` queue of `ConfidenceEvent`s (one per -/// broadcast id a peer has confirmed). +/// Cap on the `Exchange` `events` queue of `NodeEvent`s (one per broadcast id +/// a peer has confirmed). pub const EVENT_QUEUE_CAP: usize = 64; +/// Cap on the `Exchange` counted-drop table of distinct `(addr, DropReason)` +/// pairs. Once full, further counted drops fold into a single overflow counter +/// (reported with no address); see [`crate::exchange::NodeEvent::Dropped`]. +pub const DROP_TABLE_CAP: usize = 32; + /// Cap on the `Exchange` per-peer `bcast_sent` table of broadcast ids /// dispatched to a peer. pub const BCAST_SENT_CAP: usize = 256; diff --git a/tines-core/src/exchange.rs b/tines-core/src/exchange.rs index ee607339..6a83daf6 100644 --- a/tines-core/src/exchange.rs +++ b/tines-core/src/exchange.rs @@ -14,6 +14,7 @@ use crate::consts; use crate::{ Bloom, Event, Frame, NodeId, Observation, ObservationId, Reassembler, SendError, Store, Tick, + WireError, }; use heapless::Vec; @@ -44,15 +45,122 @@ pub struct InFlight { pub len: usize, } -/// Emitted for a broadcast observation a peer has confirmed holding via an -/// `Ack` receipt: [`ConfidenceEvent::peer`] is who received it and `id` is -/// the broadcast's content-ID. Produced through `Exchange::drain_events`. -#[derive(Clone, Copy, Debug, PartialEq, Eq)] -pub struct ConfidenceEvent { - /// The node id of the peer that acknowledged receiving the observation. - pub peer: NodeId, - /// The content-ID of the broadcast observation that was acknowledged. - pub id: ObservationId, +/// Why the [`Exchange`] dropped an inbound frame it could not use; reported +/// via [`NodeEvent::Dropped`]. A peer's [`Exchange`] rejects expired addressed +/// observations by policy — those are *not* counted as drops. +#[non_exhaustive] +#[derive(Clone, Copy, Debug, PartialEq)] +pub enum DropReason { + /// The frame's leading `WIRE_VERSION` byte did not match + /// [`consts::WIRE_VERSION`]. + VersionMismatch { + /// The version byte read off the wire. + version: u8, + }, + /// The frame was not a valid [`Frame`] (truncated, garbage, or empty). + Malformed, +} + +/// Emitted by [`crate::Node::drain_events`] for either a broadcast a peer +/// confirmed holding (a *confirmed delivery*) or an inbound frame the peer +/// dropped and counted. +#[non_exhaustive] +#[derive(Clone, Copy, Debug, PartialEq)] +pub enum NodeEvent { + /// A broadcast observation the peer at `peer` confirmed holding (via an + /// `Ack` receipt); `id` is the broadcast's content-ID. + Delivered { + /// The node id of the peer that acknowledged receiving the observation. + peer: NodeId, + /// The content-ID of the broadcast observation that was acknowledged. + id: ObservationId, + }, + /// The peer at `addr` sent a frame we dropped and counted under `reason`. + /// `count` is how many frames were counted for this `(addr, reason)` pair + /// since the last drain (always `>= 1`); `addr` is `None` when the + /// drop-counter table is full and the count lives in the single overflow + /// slot instead. + Dropped { + /// The wire address of the peer that sent the dropped frame. `None` + /// only for the overflow slot, which has no address to name. + addr: Option, + /// Why the frame was dropped. + reason: DropReason, + /// The number of frames folded into this counter since the last drain. + count: u32, + }, +} + +/// The [`Exchange`]'s counted-drop table: up to [`consts::DROP_TABLE_CAP`] +/// distinct `(addr, DropReason)` pairs, each with a saturating count, plus a +/// single overflow counter. Counters are drained — and reset — through +/// [`Exchange::drain_events`]. +#[derive(Clone, Debug)] +pub(crate) struct DropTable { + /// `(addr, reason, count)`, in insertion order; `count` saturates at `u32`. + entries: Vec<(A, DropReason, u32), { consts::DROP_TABLE_CAP }>, + /// A single slot for drops that arrive once the table is full: no address, + /// just the reason and a combined count. Reported with `addr: None`. + overflow: Option<(DropReason, u32)>, +} + +impl DropTable { + /// Creates an empty drop-counter table. + fn new() -> Self { + Self { + entries: Vec::new(), + overflow: None, + } + } + + /// Records one drop for `(addr, reason)`: bumps the matching entry's count + /// if present, otherwise adds a new entry, or folds into the single overflow + /// counter once the table is full. + fn record(&mut self, addr: A, reason: DropReason) { + for entry in self.entries.iter_mut() { + if entry.0 == addr && entry.1 == reason { + entry.2 = entry.2.saturating_add(1); + return; + } + } + if self.entries.is_full() { + match self.overflow { + Some((_, ref mut count)) => *count = count.saturating_add(1), + None => self.overflow = Some((reason, 1)), + } + } else { + let _ = self.entries.push((addr, reason, 1)); + } + } + + /// Emits one [`NodeEvent::Dropped`] per non-zero entry (resetting its count, + /// keeping the key) followed by the overflow counter, if any. + fn drain(&mut self) -> Vec, { consts::DROP_TABLE_CAP + 1 }> { + let mut out: Vec, { consts::DROP_TABLE_CAP + 1 }> = Vec::new(); + let mut i = 0; + while let Some(entry) = self.entries.get_mut(i) { + let (addr, reason, count) = (entry.0.clone(), entry.1, entry.2); + if count > 0 { + let _ = out.push(NodeEvent::Dropped { + addr: Some(addr), + reason, + count, + }); + entry.2 = 0; + } + i += 1; + } + if let Some((reason, count)) = self.overflow.take() { + if count > 0 { + let _ = out.push(NodeEvent::Dropped { + addr: None, + reason, + count, + }); + } + } + out + } } /// A single peer the [`Exchange`] is tracking. @@ -178,9 +286,13 @@ pub struct Exchange { /// broadcast emit path. Used to validate inbound `Ack` receipts against /// broadcasts we actually sent; overflow drops the oldest record. bcast_sent: Vec<(ObservationId, A), { consts::BCAST_SENT_CAP }>, - /// One [`ConfidenceEvent`] per broadcast id a peer has confirmed, pushed - /// drop-oldest when full. Drain through [`Exchange::drain_events`]. - events: Vec, + /// One [`NodeEvent::Delivered`] per broadcast id a peer has confirmed, + /// pushed drop-oldest when full. Drain through [`Exchange::drain_events`]. + events: Vec, { consts::EVENT_QUEUE_CAP }>, + /// Counted drops: distinct `(addr, DropReason)` pairs, each with a + /// saturating count, plus a single overflow counter; reported through + /// [`Exchange::drain_events`] and reset on drain. + drops: DropTable, /// Pending addressed acks: ids awaiting an `Ack` for a peer that sent an /// addressed observation. acks: AckBuffer, @@ -207,6 +319,7 @@ impl Exchange { bounce: Vec::new(), bcast_sent: Vec::new(), events: Vec::new(), + drops: DropTable::new(), undo: None, ack_undo: None, acks: AckBuffer::new(), @@ -655,12 +768,23 @@ impl Exchange { &self.peers } - /// [`Exchange`]s emit [`ConfidenceEvent`]s through [`Exchange::drain_events`]; - /// `tines` surfaces them to its transport, so the drain method is unused - /// inside `tines-core` itself. + /// [`Exchange`]s emit [`NodeEvent`]s — confirmed deliveries plus counted + /// drops — through `Exchange::drain_events`; `tines` surfaces them to its + /// transport, so the drain method is unused inside `tines-core` itself. + /// + /// Returns confirmed deliveries first, oldest first, then one + /// [`NodeEvent::Dropped`] per non-zero counter (resetting each counter). + /// Sized to hold the entire delivery queue plus the whole drop table and + /// its overflow slot, so a full delivery queue never evicts a counted drop. #[allow(dead_code)] - pub(crate) fn drain_events(&mut self) -> Vec { - let mut drained: Vec = Vec::new(); + pub(crate) fn drain_events( + &mut self, + ) -> Vec, { consts::EVENT_QUEUE_CAP + consts::DROP_TABLE_CAP + 1 }> { + let mut drained: Vec< + NodeEvent, + { consts::EVENT_QUEUE_CAP + consts::DROP_TABLE_CAP + 1 }, + > = Vec::new(); + // Confirmed deliveries, oldest first. while !self.events.is_empty() { if drained.is_full() { break; @@ -668,6 +792,10 @@ impl Exchange { let ev = self.events.remove(0); let _ = drained.push(ev); } + // Then one counted drop per non-zero counter, resetting each. + for dropped in self.drops.drain() { + let _ = drained.push(dropped); + } drained } @@ -716,7 +844,18 @@ impl Exchange { ) -> Option { let frame = match Frame::decode(bytes) { Ok(frame) => frame, - Err(_) => return None, + Err(WireError::VersionMismatch { version }) => { + // A wrong-version frame is a counted drop, not a silent drop. + self.drops + .record(addr, DropReason::VersionMismatch { version }); + return None; + } + // Any other decode failure (malformed, truncated, or the encode-only + // TooLarge which decode never yields) is a counted malformed drop. + Err(_) => { + self.drops.record(addr, DropReason::Malformed); + return None; + } }; let idx = self.peer_index(&addr)?; let peer = &mut self.peers[idx]; @@ -750,8 +889,8 @@ impl Exchange { /// For each id in the receipt: an id the local node `Sent` to `addr` (per /// the bounce table) is released from custody so it is never forwarded /// again; an id the node dispatched as a broadcast to `addr` (per - /// `bcast_sent`) fires a [`ConfidenceEvent`]. A receipt for an id the node - /// never dispatched to `addr` is silently ignored. No path panics on + /// `bcast_sent`) fires a [`NodeEvent::Delivered`]. A receipt for an id the + /// node never dispatched to `addr` is silently ignored. No path panics on /// malformed input — an oversized id list just drops the overflow. fn on_ack( &mut self, @@ -774,7 +913,7 @@ impl Exchange { } } else if bcast_sent_contains(&self.bcast_sent, *id, &addr) { // Broadcast: the peer confirms it holds our broadcast. - let _ = self.events.push(ConfidenceEvent { + let _ = self.events.push(NodeEvent::Delivered { peer: from, id: *id, }); @@ -2712,7 +2851,7 @@ mod tests { } /// An Ack for a broadcast we dispatched to the peer fires exactly one - /// ConfidenceEvent naming the peer and the broadcast id. + /// `NodeEvent::Delivered` naming the peer and the broadcast id. #[test] fn ack_broadcast_fires_confidence_event() { let peer: NodeId = [2; 16]; @@ -2734,8 +2873,16 @@ mod tests { 1, "one broadcast ack fires one confidence event" ); - assert_eq!(events[0].peer, peer); - assert_eq!(events[0].id, b_id); + match events[0] { + NodeEvent::Delivered { + peer: ev_peer, + id: ev_id, + } => { + assert_eq!(ev_peer, peer); + assert_eq!(ev_id, b_id); + } + other => panic!("expected a Delivered event, got {other:?}"), + } } /// A broadcast Ack does not release custody: there is no local addressed @@ -2847,8 +2994,8 @@ mod tests { } /// A broadcast Ack from a peer that never held the broadcast fires no - /// ConfidenceEvent: the receipt names a broadcast we did not dispatch to - /// that peer. + /// `NodeEvent::Delivered`: the receipt names a broadcast we did not dispatch + /// to that peer. #[test] fn ack_broadcast_forged_id_no_confidence() { let peer: NodeId = [2; 16]; @@ -2904,8 +3051,18 @@ mod tests { 2, "each broadcast ack fires a confidence event" ); - assert_eq!(events[0].id, a_id, "events come back oldest-first"); - assert_eq!(events[1].id, b_id); + match events[0] { + NodeEvent::Delivered { id, .. } => { + assert_eq!(id, a_id, "events come back oldest-first"); + } + other => panic!("expected a Delivered event, got {other:?}"), + } + match events[1] { + NodeEvent::Delivered { id, .. } => { + assert_eq!(id, b_id); + } + other => panic!("expected a Delivered event, got {other:?}"), + } } /// drain_events empties the queue: a second drain yields nothing. @@ -2925,7 +3082,12 @@ mod tests { let first = ex.drain_events(); assert_eq!(first.len(), 1, "the queued confidence event is drained"); - assert_eq!(first[0].id, b_id); + match first[0] { + NodeEvent::Delivered { id, .. } => { + assert_eq!(id, b_id); + } + other => panic!("expected a Delivered event, got {other:?}"), + } let second = ex.drain_events(); assert!( @@ -2933,4 +3095,192 @@ mod tests { "drain_events empties the queue; a second drain yields nothing" ); } + + /// A frame whose leading byte isn't WIRE_VERSION is a counted + /// version-mismatch drop, reported with the sender, the version byte, and a + /// count of one. + #[test] + fn version_mismatch_is_counted() { + let mut store = MemStore::new(); + let mut ex = Exchange::::new(ME); + // The first byte is 0, which isn't WIRE_VERSION: a version mismatch. + let bytes = heapless::Vec::::from_slice(&[0u8; 10]).unwrap(); + assert_eq!(ex.on_event(Event::Frame(1, bytes), &mut store, 0), None); + + let drained = ex.drain_events(); + assert_eq!(drained.len(), 1, "exactly one counted drop"); + match drained.into_iter().next().unwrap() { + NodeEvent::Dropped { + addr, + reason, + count, + } => { + assert_eq!(addr, Some(1u8)); + assert_eq!(reason, DropReason::VersionMismatch { version: 0 }); + assert_eq!(count, 1); + } + other => panic!("expected a counted Dropped, got {other:?}"), + } + } + + /// A frame that passes the version gate (byte 0 is WIRE_VERSION) but whose + /// body isn't a valid postcard [`Frame`] is a counted malformed drop. + #[test] + fn malformed_is_counted() { + let mut store = MemStore::new(); + let mut ex = Exchange::::new(ME); + let mut bytes = heapless::Vec::::new(); + bytes.push(consts::WIRE_VERSION).unwrap(); + bytes.push(0xFF).unwrap(); + bytes.push(0xFF).unwrap(); + assert_eq!(ex.on_event(Event::Frame(1, bytes), &mut store, 0), None); + + let drained = ex.drain_events(); + assert_eq!(drained.len(), 1, "exactly one counted drop"); + match drained.into_iter().next().unwrap() { + NodeEvent::Dropped { + addr, + reason, + count, + } => { + assert_eq!(addr, Some(1u8)); + assert_eq!(reason, DropReason::Malformed); + assert_eq!(count, 1); + } + other => panic!("expected a counted Dropped, got {other:?}"), + } + } + + /// A repeat of the same (addr, reason) folds into one counter whose count + /// is two: the two frames are one drop, counted once with count 2. + #[test] + fn same_drop_increments_count() { + let mut store = MemStore::new(); + let mut ex = Exchange::::new(ME); + let bytes = heapless::Vec::::from_slice(&[0u8; 10]).unwrap(); + assert_eq!( + ex.on_event(Event::Frame(1, bytes.clone()), &mut store, 0), + None + ); + assert_eq!(ex.on_event(Event::Frame(1, bytes), &mut store, 0), None); + + let drained = ex.drain_events(); + assert_eq!(drained.len(), 1, "a repeat is folded into one counter"); + match drained.into_iter().next().unwrap() { + NodeEvent::Dropped { + addr, + reason, + count, + } => { + assert_eq!(addr, Some(1u8)); + assert_eq!(reason, DropReason::VersionMismatch { version: 0 }); + assert_eq!(count, 2, "both frames counted against one peer+reason"); + } + other => panic!("expected a single counted Dropped with count 2, got {other:?}"), + } + } + + /// An addressed observation that arrives already past its lifetime_ms is + /// rejected by policy; that rejection is NOT a counted drop (unlike a + /// decode error). + #[test] + fn expired_addressed_is_not_a_drop() { + let mut store = MemStore::new(); + let mut ex = Exchange::::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); + let frame = Frame::Obs { + from: [2; 16], + obs: o, + }; + assert_eq!( + ex.on_event(Event::Frame(1, frame_bytes(&frame)), &mut store, 0), + None, + "an expired addressed obs must not be delivered" + ); + // A policy rejection is not a counted drop. + assert!( + ex.drain_events().is_empty(), + "an expired addressed obs is rejected by policy, not counted as a drop" + ); + } + + /// A full delivery queue and a counted drop survive together: drops are + /// appended after the deliveries and none are evicted (the drop-oldest + /// eviction of the delivery queue is not in play here, and the drop-table + /// and delivery queue never evict each other). + #[test] + fn drops_do_not_evict_delivered() { + let mut store = MemStore::new(); + let mut ex = Exchange::::new(ME); + // Fill the delivery queue to the brim with confirmed broadcasts. + for _ in 0..consts::EVENT_QUEUE_CAP { + ex.events + .push(NodeEvent::Delivered { + peer: [7; 16], + id: [0u8; 16], + }) + .unwrap(); + } + // One counted drop lands alongside the full delivery queue. + let bytes = heapless::Vec::::from_slice(&[0u8; 10]).unwrap(); + assert_eq!(ex.on_event(Event::Frame(1, bytes), &mut store, 0), None); + + let drained = ex.drain_events(); + let mut delivered = 0usize; + let mut dropped = 0usize; + for ev in drained { + match ev { + NodeEvent::Delivered { .. } => delivered += 1, + NodeEvent::Dropped { .. } => dropped += 1, + } + } + assert_eq!( + delivered, + consts::EVENT_QUEUE_CAP, + "every queued delivery survives a full queue" + ); + assert_eq!(dropped, 1, "the counted drop is reported alongside them"); + } + + /// Once the drop-counter table is full, the next drop folds into a single + /// overflow counter reported with `addr: None`; the table's 32 entries are + /// all still reported with a real address. + #[test] + fn overflow_counter_has_no_addr() { + let mut store = MemStore::new(); + let mut ex = Exchange::::new(ME); + // Drive DROP_TABLE_CAP + 1 distinct (addr, reason) pairs: the table + // holds DROP_TABLE_CAP, so the one beyond the cap folds into the single + // overflow counter and is reported with no address. + for addr in 0..=(consts::DROP_TABLE_CAP as u8) { + let bytes = heapless::Vec::::from_slice(&[0u8; 10]).unwrap(); + assert_eq!( + ex.on_event(Event::Frame(addr, bytes.clone()), &mut store, 0), + None + ); + } + + let drained = ex.drain_events(); + let mut reported = 0usize; + let mut overflow = 0usize; + for ev in drained { + match ev { + NodeEvent::Dropped { addr: Some(..), .. } => reported += 1, + NodeEvent::Dropped { addr: None, .. } => overflow += 1, + NodeEvent::Delivered { .. } => {} + } + } + assert_eq!( + overflow, 1, + "the drop beyond the table cap must report with no address" + ); + assert_eq!( + reported, + consts::DROP_TABLE_CAP, + "all capped entries reported" + ); + } } diff --git a/tines-core/src/lib.rs b/tines-core/src/lib.rs index c5149775..20d77b52 100644 --- a/tines-core/src/lib.rs +++ b/tines-core/src/lib.rs @@ -8,7 +8,6 @@ pub mod belief; pub mod bloom; pub mod consts; pub mod exchange; -pub use exchange::ConfidenceEvent; pub mod frag; #[cfg(feature = "alloc")] pub mod mem_store; @@ -20,7 +19,7 @@ pub mod wire; pub use belief::{BeliefPolicy, LatestWins, SelfFirst}; pub use bloom::Bloom; -pub use exchange::{Exchange, Peer}; +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}; diff --git a/tines-core/src/node.rs b/tines-core/src/node.rs index 9347fc57..f6b30cce 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, ConfidenceEvent, Exchange, NodeId, Observation, ObservationId, Payload, - Store, StoreError, Tick, Topic, Transport, TrimPolicy, + consts, BeliefPolicy, Exchange, NodeEvent, NodeId, Observation, ObservationId, Payload, Store, + StoreError, Tick, Topic, Transport, TrimPolicy, }; /// Errors returned by [`Node::observe`]. @@ -165,15 +165,20 @@ impl Node { stored } - /// Drains the buffered [`ConfidenceEvent`]s the [`Exchange`] has collected - /// as a peer acknowledged holding one of our broadcasts, oldest-first. + /// Drains the [`NodeEvent`]s the [`Exchange`] has collected: confirmed + /// deliveries first (oldest-first), then one [`NodeEvent::Dropped`] per + /// counted drop. /// /// 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 { + /// [`consts::EVENT_QUEUE_CAP`] confirmed deliveries plus the entire + /// [`consts::DROP_TABLE_CAP`]-entry drop table and its overflow slot. Index + /// it directly — a [`NodeEvent::Delivered`] names the peer that held our + /// broadcast and the broadcast's content-ID; a [`NodeEvent::Dropped`] names + /// the peer and reason a dropped frame was rejected with. + pub fn drain_events( + &mut self, + ) -> heapless::Vec, { consts::EVENT_QUEUE_CAP + consts::DROP_TABLE_CAP + 1 }> + { self.exchange.drain_events() } diff --git a/tines-core/tests/pair.rs b/tines-core/tests/pair.rs index c317b942..927ab0f6 100644 --- a/tines-core/tests/pair.rs +++ b/tines-core/tests/pair.rs @@ -7,7 +7,7 @@ use common::PairTransport; use std::collections::BTreeSet; use tines_core::observation::Payload; use tines_core::transport::{Event, FrameBuf, SendError, Transport}; -use tines_core::{consts, LatestWins, MemStore, Node, ObserveError, Store}; +use tines_core::{consts, LatestWins, MemStore, Node, NodeEvent, ObserveError, Store}; #[test] fn pair_transport_connect_then_frame() { @@ -384,10 +384,13 @@ fn confidence_event_surfaces_on_node() { "b must receive a's broadcast before an Ack fires" ); - // a's node surfaces the confidence event: peer == b, id == the broadcast. + // a's node surfaces the confirmed delivery: 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" + events.iter().any(|ev| matches!( + ev, + NodeEvent::Delivered { peer, id: observed_id } if *peer == nb.id() && *observed_id == id + )), + "a's node must surface the confirmed delivery for b's Ack" ); } diff --git a/tines-ros2/src/lib.rs b/tines-ros2/src/lib.rs index 633d1eb9..23a2ed07 100644 --- a/tines-ros2/src/lib.rs +++ b/tines-ros2/src/lib.rs @@ -9,8 +9,8 @@ //! dependency. This is what builds and is tested without a ROS2 toolchain. //! //! * **rclrs-backed** (feature `ros2`): `TinesNode`, a self-contained ROS2 node -//! that publishes each tines topic as a ROS2 topic and -//! [`crate::RECEIPTS_TOPIC`] for [`tines_core::ConfidenceEvent`]s, and +//! that publishes each tines topic as a ROS2 topic and publishes confirmed +//! deliveries to [`crate::RECEIPTS_TOPIC`] for [`tines_core::NodeEvent`]s, and //! subscribes for new observations. Compiled only behind //! `#[cfg(feature = "ros2")]` so the default build never needs rclrs. //! @@ -20,7 +20,7 @@ pub mod topics; -/// ROS2 topic carrying serialized [`tines_core::ConfidenceEvent`]s (one per +/// ROS2 topic carrying confirmed [`tines_core::NodeEvent`]s (one per /// broadcast a peer confirmed holding). Always the hex-escaped literal below. pub const RECEIPTS_TOPIC: &[u8] = b"/tines/receipts"; diff --git a/tines-ros2/src/node.rs b/tines-ros2/src/node.rs index 7f0043c6..0e0c68da 100644 --- a/tines-ros2/src/node.rs +++ b/tines-ros2/src/node.rs @@ -7,7 +7,7 @@ //! * **tines → ROS2**: each distinct live tines topic is published to its //! hex-escaped ROS2 topic (see [`crate::topics::topic_name`]) carrying the //! raw observation payload as a `UInt8MultiArray` (`data: Vec`), and every -//! [`tines_core::ConfidenceEvent`] the exchange emits is published to +//! [`tines_core::NodeEvent`] the exchange emits is published to //! [`RECEIPTS_TOPIC`] as a 32-byte `UInt8MultiArray` (`peer[..16] ++ //! id[..16]`). //! * **ROS2 → tines**: inbound messages on a topic are observed back into the @@ -45,7 +45,9 @@ use rclrs::{Context, CreateBasicExecutor, Node as RclNode, Publisher, RclrsError use ros_env::std_msgs::msg::{MultiArrayLayout, UInt8MultiArray}; use tines_bt::{BtTransport, Radio}; -use tines_core::{Node as TinesNodeCore, NodeId, Observation, ObservationId, Store, Tick}; +use tines_core::{ + Node as TinesNodeCore, NodeEvent, NodeId, Observation, ObservationId, Store, Tick, +}; use tines_store::LogStore; use crate::topics::topic_name; @@ -54,7 +56,7 @@ use crate::topics::topic_name; /// rclrs dispatch thread to the node's single owner task. type InboundMsg = (Vec, Vec); -/// ROS2 topic carrying confidence events, the literal `/tines/receipts` +/// ROS2 topic carrying confirmed deliveries, the literal `/tines/receipts` /// (hex-escaping is a no-op here since the bytes are already valid ROS2 name /// characters). This is a `&str` because ROS2 publisher/subscription creation /// takes a `&str` topic name, unlike the arbitrary-byte [`crate::RECEIPTS_TOPIC`]. @@ -106,7 +108,7 @@ pub struct TinesNode { tines: TinesNodeCore, /// ROS2 topics currently managed (ROS2 topic name → bookkeeping). managed: HashMap, - /// Receipt publisher for confidence events. + /// Receipt publisher for confirmed deliveries. receipt_pub: Publisher, /// Channel from subscription callbacks back to this node. inbound_rx: std::sync::mpsc::Receiver, @@ -174,7 +176,8 @@ impl TinesNode { } /// One bridge tick: consume inbound ROS2 observations, drive the tines - /// exchange, publish live topics, and publish any confidence events. + /// exchange, publish live topics, and publish confirmed deliveries (logging + /// any counted drop to stderr). /// /// `now` is a monotonic millisecond [`Tick`]: milliseconds since the host /// process started (a single `Instant`-based clock, `0` at startup), never @@ -325,24 +328,40 @@ impl TinesNode { best } - /// Publish every pending confidence event to [`RECEIPTS_TOPIC`]. + /// Publish every confirmed delivery to [`RECEIPTS_TOPIC`]; log any counted + /// drop to stderr (counted drops are the point of D3 — the operator must + /// see them — so they are never published to the receipts topic). /// /// Each receipt is a 32-byte `UInt8MultiArray` = `peer[..16] ++ id[..16]`, /// the raw node ids (no hex encoding). fn publish_receipts(&mut self) { let events = self.tines.drain_events(); - for ev in events.iter() { - let mut data = [0u8; 32]; - data[..16].copy_from_slice(&ev.peer); - data[16..32].copy_from_slice(&ev.id); - let msg = UInt8MultiArray { - data: data.to_vec(), - layout: MultiArrayLayout { - dim: vec![], - data_offset: 0, - }, - }; - let _ = self.receipt_pub.publish(&msg); + for ev in events { + match ev { + NodeEvent::Delivered { peer, id } => { + let mut data = [0u8; 32]; + data[..16].copy_from_slice(&peer); + data[16..32].copy_from_slice(&id); + let msg = UInt8MultiArray { + data: data.to_vec(), + layout: MultiArrayLayout { + dim: vec![], + data_offset: 0, + }, + }; + let _ = self.receipt_pub.publish(&msg); + } + // Counted drops are surfaced to the operator via stderr. + NodeEvent::Dropped { + addr, + reason, + count, + } => { + eprintln!("tines: drop: {reason:?} (count={count}, addr={addr:?})"); + } + // `NodeEvent` is non-exhaustive: ignore any future variant. + _ => {} + } } } } diff --git a/tines/src/ipc.rs b/tines/src/ipc.rs index 76a98762..7c38847e 100644 --- a/tines/src/ipc.rs +++ b/tines/src/ipc.rs @@ -39,7 +39,7 @@ use std::os::unix::net::{UnixListener, UnixStream}; use std::path::Path; use std::time::Duration; -use tines_core::{Node, Store, Transport, TrimPolicy}; +use tines_core::{Node, NodeEvent, Store, Transport, TrimPolicy}; use tines_store::LogStore; use crate::HostClock; @@ -51,7 +51,7 @@ const MAX_FRAME: usize = 1 << 20; // 1 MiB cap on a single JSON frame /// of its accept loop. const POLL_CAP: u64 = 64; -/// A confidence event handed out by the IPC `events` request. +/// A confirmed delivery handed out by the IPC `events` request. #[derive(serde::Serialize, serde::Deserialize, Debug)] pub struct EventOut { /// Hex of the `ObservationId` this confidence concerns. @@ -87,7 +87,7 @@ pub enum Request { /// Hex of the topic bytes to query. topic: String, }, - /// Drain outstanding confidence events. + /// Drain outstanding confirmed deliveries and drop counters. Events, /// Report the node's identity and log depth. Status, @@ -107,9 +107,9 @@ pub enum Response { /// The stored belief, or `None` if the topic is unknown. obs: Option, }, - /// Confidence events drained from the exchange. + /// Confirmed deliveries drained from the exchange. Events { - /// The drained confidence events. + /// The drained confirmed deliveries. events: Vec, }, /// The node's `id`, next tines `next_seq`, and observations `stored`. @@ -176,9 +176,24 @@ where let events = node .drain_events() .into_iter() - .map(|ev| EventOut { - id: encode_hex(&ev.id), - peer: encode_hex(&ev.peer), + .filter_map(|ev| match ev { + NodeEvent::Delivered { peer, id } => Some(EventOut { + id: encode_hex(&id), + peer: encode_hex(&peer), + }), + // Dropped frames are surfaced to the operator via stderr + // (they are the whole point of D3), not the `events` + // response, so they filter out here. + NodeEvent::Dropped { + addr, + reason, + count, + } => { + eprintln!("tines: drop: {reason:?} (count={count}, addr={addr:?})"); + None + } + // `NodeEvent` is non-exhaustive: ignore any future variant. + _ => None, }) .collect(); Response::Events { events } diff --git a/tines/tests/ipc.rs b/tines/tests/ipc.rs index d842c54d..f4e320de 100644 --- a/tines/tests/ipc.rs +++ b/tines/tests/ipc.rs @@ -16,7 +16,7 @@ use heapless::Vec as HeaplessVec; use std::fmt::Write as _; use tines::{handle, serve, HostClock, Request, Response}; use tines_core::{ - consts, Event, Frame, FrameBuf, Node, NodeId, ObservationId, SendError, Transport, + consts, Event, Frame, FrameBuf, Node, NodeEvent, NodeId, ObservationId, SendError, Transport, }; use tines_store::LogStore; @@ -67,7 +67,7 @@ impl Transport for MockTransport { Ok(Frame::Summary { salt, .. }) => { // Echo back an empty, full-size Summary so the exchange learns // the peer holds nothing and broadcasts its diff (which we then - // ack, firing a ConfidenceEvent). + // ack, firing a NodeEvent::Delivered). let raw: [u8; consts::SUMMARY_BYTES] = [0u8; consts::SUMMARY_BYTES]; let bits: HeaplessVec = HeaplessVec::from_slice(&raw).expect("512 fits SUMMARY_BYTES"); @@ -276,7 +276,7 @@ fn events_over_socket() { rpc(&mut stream, &Request::Events); match read_response(&mut stream) { Response::Events { events } => { - assert_eq!(events.len(), 1, "expected one confidence event"); + assert_eq!(events.len(), 1, "expected one confirmed delivery"); assert_eq!(events[0].id, hex_encode(&x), "event id matches observed id"); assert_eq!(events[0].peer, hex_encode(&peer), "event came from peer"); } @@ -303,7 +303,7 @@ fn handle_bad_hex_is_error() { } #[test] -fn confidence_event_direct_probe() { +fn confirmed_delivery_direct_probe() { let dir = unique_dir("ev"); let mut node = build_node([7; 16], &dir); let x = node.observe(&topic(), &payload(), None, 0).unwrap(); @@ -311,9 +311,14 @@ fn confidence_event_direct_probe() { node.poll(0); node.poll(0); let evs = node.drain_events(); - assert_eq!(evs.len(), 1, "one confidence event after observe+ack"); - assert_eq!(evs[0].id, x); - assert_eq!(evs[0].peer, [7; 16]); + assert_eq!(evs.len(), 1, "one confirmed delivery after observe+ack"); + match &evs[0] { + NodeEvent::Delivered { peer, id } => { + assert_eq!(*id, x); + assert_eq!(*peer, [7; 16]); + } + other => panic!("expected a Delivered event, got {other:?}"), + } } /// A malformed JSON frame must yield an `Error` response and leave the