From c1cad1661ceaa8aadb5861203a2e76a5b5169181 Mon Sep 17 00:00:00 2001 From: "K. Isom" Date: Sat, 24 Oct 2026 22:28:08 -0700 Subject: [PATCH] feat(core): emit Contact/Lost NodeEvents (decision C) --- tines-core/src/exchange.rs | 189 +++++++++++++++++++++++++++++++++---- 1 file changed, 171 insertions(+), 18 deletions(-) diff --git a/tines-core/src/exchange.rs b/tines-core/src/exchange.rs index da2d9849..628865dc 100644 --- a/tines-core/src/exchange.rs +++ b/tines-core/src/exchange.rs @@ -99,6 +99,26 @@ pub enum NodeEvent { /// The number of frames folded into this counter since the last drain. count: u32, }, + /// A new contact started with the peer at `addr` (a first `Contact` + /// event for an addr we were not already tracking). A repeat `Contact` + /// for a known peer restarts the contact silently. Idle eviction (F8) + /// does not emit this. + /// + /// `addr` is the transport address; for both the BLE and bus transports + /// it IS the peer's [`NodeId`]. Best-effort telemetry: when the event + /// queue is full, an incoming contact is dropped rather than evicting + /// queued events. + Contact { + /// The wire address of the peer the contact started with. + addr: A, + }, + /// The peer at `addr` was lost: a transport `Lost` event for a peer we + /// were tracking. A `Lost` for an unknown addr, and F8 idle eviction, + /// emit nothing. + Lost { + /// The wire address of the peer that was lost. + addr: A, + }, } /// The `Exchange`'s counted-drop table: up to [`consts::DROP_TABLE_CAP`] @@ -323,7 +343,9 @@ pub(crate) struct Exchange { /// record. bcast_sent: Vec<(ObservationId, A), { consts::BCAST_SENT_CAP }>, /// One [`NodeEvent::Delivered`] per broadcast id a peer has confirmed, - /// pushed drop-oldest when full. Drain through [`Exchange::drain_events`]. + /// plus one [`NodeEvent::Contact`]/[`NodeEvent::Lost`] per contact + /// change, 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 @@ -375,11 +397,15 @@ impl Exchange { ) -> Option { match ev { Event::Contact(addr) => { - self.add_peer(addr, now); + if self.add_peer(addr.clone(), now) { + self.push_event(NodeEvent::Contact { addr }); + } None } Event::Lost(addr) => { - self.remove_peer(&addr); + if self.remove_peer(&addr) { + self.push_event(NodeEvent::Lost { addr }); + } None } Event::Frame(addr, bytes) => self.on_frame(addr, &bytes, store, now), @@ -829,13 +855,24 @@ impl Exchange { &self.peers } - /// Drains the [`NodeEvent`]s — confirmed deliveries plus counted drops — - /// for [`crate::Node::drain_events`]. + /// Pushes an event onto the delivery queue with drop-oldest overflow + /// (F13): when the queue is full the OLDEST event is evicted so the + /// freshest telemetry survives (a bare `push` would drop the newest). + fn push_event(&mut self, ev: NodeEvent) { + if self.events.is_full() { + let _ = self.events.remove(0); + } + let _ = self.events.push(ev); + } + + /// Drains the [`NodeEvent`]s — contacts and losses, confirmed deliveries, + /// plus counted drops — for [`crate::Node::drain_events`]. /// - /// Returns confirmed deliveries first, oldest first, then one + /// Returns queued events first, oldest first (contacts/losses and + /// deliveries share the queue in the order they happened), then one /// [`NodeEvent::Dropped`] per non-zero counter, emptying the drop table. - /// 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. + /// Sized to hold the entire event queue plus the whole drop table and + /// its overflow slot, so a full queue never evicts a counted drop. pub(crate) fn drain_events( &mut self, ) -> Vec, { consts::EVENT_QUEUE_CAP + consts::DROP_TABLE_CAP + 1 }> { @@ -862,18 +899,25 @@ impl Exchange { /// starts a new contact too: the F6 pin and the per-contact records /// (bounces, broadcast receipts) reset. /// A repeat `Contact` also restarts the peer's idle clock. - fn add_peer(&mut self, addr: A, now: Tick) { + /// + /// Returns `true` only when a new [`Peer`] was actually added to the + /// table (the contact is "new" for the console's event feed). A repeat + /// contact, and a contact that cannot be added because the table is + /// full ([`consts::MAX_PEERS`]), return `false`. + fn add_peer(&mut self, addr: A, now: Tick) -> bool { self.evict_idle(now); self.forget_contact(&addr); if let Some(peer) = self.peers.iter_mut().find(|p| p.addr == addr) { peer.id = None; peer.added_at = now; peer.last_inbound = None; - return; + return false; } if self.peers.len() < consts::MAX_PEERS { let _ = self.peers.push(Peer::new(addr, now)); + return true; } + false } /// F8: removes, as if Lost, every peer with no inbound frame (or, if it @@ -894,10 +938,13 @@ impl Exchange { } } - fn remove_peer(&mut self, addr: &A) { - if let Some(idx) = self.peers.iter().position(|p| p.addr == *addr) { + fn remove_peer(&mut self, addr: &A) -> bool { + let removed = if let Some(idx) = self.peers.iter().position(|p| p.addr == *addr) { self.peers.remove(idx); - } + true + } else { + false + }; // F13: purge the removed peer's pending acks (only its own) so their // ids are never acked again under this contact. self.acks.retain_other(addr); @@ -906,6 +953,7 @@ impl Exchange { self.ack_undo = None; } self.forget_contact(addr); + removed } /// Drops the per-contact records for `addr` — bounce suppression and @@ -1047,10 +1095,7 @@ impl Exchange { // delivery queue is full, drop the OLDEST Delivered rather than // the newest (a bare `push` drops the newest) so the freshest // confirmed deliveries survive overflow (F13). - if self.events.is_full() { - let _ = self.events.remove(0); - } - let _ = self.events.push(NodeEvent::Delivered { + self.push_event(NodeEvent::Delivered { peer: from, id: *id, }); @@ -2981,6 +3026,7 @@ mod tests { let b_id = b.id(); store.insert(b, 0).unwrap(); deliver_to(&mut ex, &mut store, 1, b_id); + let _ = ex.drain_events(); // clear the contact event assert!(!store.is_released(&b_id)); let ids = heapless::Vec::from_slice(&[b_id]).unwrap(); @@ -3126,6 +3172,7 @@ mod tests { store.insert(b, 0).unwrap(); // Dispatch the broadcast to a different peer (addr 2), not addr 1. deliver_to(&mut ex, &mut store, 2, b_id); + let _ = ex.drain_events(); // clear the contact event // Peer 1 (addr 1) claims to hold b_id though it never did. let ids = heapless::Vec::from_slice(&[b_id]).unwrap(); @@ -3154,6 +3201,7 @@ mod tests { store.insert(b, 0).unwrap(); deliver_to(&mut ex, &mut store, 1, a_id); deliver_to(&mut ex, &mut store, 1, b_id); + let _ = ex.drain_events(); // clear the contact event let ack_for = |ids: heapless::Vec| Frame::Ack { from: peer, @@ -3195,6 +3243,7 @@ mod tests { let b_id = b.id(); store.insert(b, 0).unwrap(); deliver_to(&mut ex, &mut store, 1, b_id); + let _ = ex.drain_events(); // clear the contact event let ids = heapless::Vec::from_slice(&[b_id]).unwrap(); let ack = Frame::Ack { from: peer, ids }; @@ -3357,6 +3406,7 @@ mod tests { match ev { NodeEvent::Delivered { .. } => delivered += 1, NodeEvent::Dropped { .. } => dropped += 1, + NodeEvent::Contact { .. } | NodeEvent::Lost { .. } => {} } } assert_eq!( @@ -3367,6 +3417,105 @@ mod tests { assert_eq!(dropped, 1, "the counted drop is reported alongside them"); } + /// A first `Contact` for an addr emits one [`NodeEvent::Contact`] naming + /// it; a repeat `Contact` for a known peer restarts the contact silently. + #[test] + fn contact_for_new_peer_emits_event() { + let mut ex = Exchange::new(ME); + let mut store = MemStore::new(); + ex.on_event(Event::Contact(1), &mut store, 0); + + let events = ex.drain_events(); + assert_eq!(events.len(), 1, "a new contact emits one event"); + assert_eq!(events[0], NodeEvent::Contact { addr: 1 }); + + ex.on_event(Event::Contact(1), &mut store, 0); + assert!( + ex.drain_events().is_empty(), + "a repeat contact for a known peer emits nothing" + ); + } + + /// `Lost` for a peer we were tracking emits [`NodeEvent::Lost`]; `Lost` + /// for an unknown addr emits nothing. + #[test] + fn lost_for_known_peer_emits_event() { + let mut ex = Exchange::new(ME); + let mut store = MemStore::new(); + ex.on_event(Event::Contact(1), &mut store, 0); + let _ = ex.drain_events(); // clear the contact event + + ex.on_event(Event::Lost(1), &mut store, 0); + let events = ex.drain_events(); + assert_eq!(events.len(), 1, "losing a known peer emits one event"); + assert_eq!(events[0], NodeEvent::Lost { addr: 1 }); + + ex.on_event(Event::Lost(2), &mut store, 0); + assert!( + ex.drain_events().is_empty(), + "losing an unknown peer emits nothing" + ); + } + + /// F8 idle eviction removes the peer but is not a transport contact + /// event: it must stay silent in the event feed. + #[test] + fn idle_eviction_emits_nothing() { + let mut ex = Exchange::new(ME); + let mut store = MemStore::new(); + ex.on_event(Event::Contact(1), &mut store, 0); + let _ = ex.drain_events(); + + let mut buf = [0u8; consts::MAX_FRAME]; + let _ = ex.next_frame( + &store, + consts::MAX_FRAME, + consts::PEER_IDLE_MS + 1, + &mut buf, + ); + assert!( + ex.drain_events().is_empty(), + "idle eviction is not a contact event" + ); + } + + /// A full event queue evicts the OLDEST event for an incoming contact + /// (F13): contact telemetry is best-effort, queued deliveries survive. + #[test] + fn contact_event_rides_the_drop_oldest_queue() { + let mut store = MemStore::new(); + let mut ex = Exchange::::new(ME); + for seq in 0..consts::EVENT_QUEUE_CAP as u64 { + ex.events + .push(NodeEvent::Delivered { + peer: [7; 16], + id: [seq as u8; 16], + }) + .unwrap(); + } + + ex.on_event(Event::Contact(1), &mut store, 0); + + let drained = ex.drain_events(); + assert_eq!( + drained.len(), + consts::EVENT_QUEUE_CAP, + "the queue stays at capacity" + ); + let first = match &drained[0] { + NodeEvent::Delivered { id, .. } => *id, + other => panic!("expected the first event to be a Delivered, got {other:?}"), + }; + assert_eq!( + first, [1u8; 16], + "the oldest Delivered (seq 0) was evicted for the contact" + ); + match drained.last().unwrap() { + NodeEvent::Contact { addr } => assert_eq!(*addr, 1), + other => panic!("expected the newest event to be the Contact, got {other:?}"), + } + } + /// 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. @@ -3392,7 +3541,9 @@ mod tests { match ev { NodeEvent::Dropped { addr: Some(..), .. } => reported += 1, NodeEvent::Dropped { addr: None, .. } => overflow += 1, - NodeEvent::Delivered { .. } => {} + NodeEvent::Delivered { .. } + | NodeEvent::Contact { .. } + | NodeEvent::Lost { .. } => {} } } assert_eq!( @@ -3466,6 +3617,7 @@ mod tests { let mut ex = Exchange::new(ME); let mut store = MemStore::new(); ex.on_event(Event::Contact(1), &mut store, 0); + let _ = ex.drain_events(); // clear the contact event ex.on_event(Event::Frame(1, hello(a, 1)), &mut store, 0); assert_eq!(ex.peers()[0].id, Some(a)); @@ -3552,6 +3704,7 @@ mod tests { let b_id = b.id(); store.insert(b, 0).unwrap(); deliver_to(&mut ex, &mut store, 1, b_id); + let _ = ex.drain_events(); // clear the contact event let ack = Frame::Ack { from: peer,