diff --git a/crates/kerykeion/src/collector.rs b/crates/kerykeion/src/collector.rs index 58a8fd3..74db40e 100644 --- a/crates/kerykeion/src/collector.rs +++ b/crates/kerykeion/src/collector.rs @@ -211,6 +211,48 @@ impl MeshCollector { drop(guard); } + /// Dispatches a decoded `FromRadio` message to the `PacketProcessor`. + /// + /// Mesh packets are routed to [`PacketProcessor::process_mesh_packet`] for + /// topology + `GeoSignal` emission. Top-level `NodeInfo` frames never pass + /// through that decode path (`process_mesh_packet` only sees `NODEINFO_APP` + /// payloads carried inside a `MeshPacket`), so they are mirrored directly + /// into the processor's `NodeDb` here - otherwise a node learned only via + /// a runtime `NodeInfo` frame stays invisible to the processor (#198). + async fn dispatch_to_processor( + &self, + from_radio: &FromRadio, + processor: &Arc>, + ) { + match &from_radio.payload_variant { + Some(from_radio::PayloadVariant::Packet(pkt)) => { + processor.lock().await.process_mesh_packet(pkt); + // Inbound ACK/NAK: update the delivery tracker + outbound queue. + self.dispatch_routing(pkt).await; + } + Some(from_radio::PayloadVariant::NodeInfo(node_info)) => { + let mesh_node = crate::handshake::node_info_to_mesh_node(node_info); + processor.lock().await.node_db_mut().insert(mesh_node); + } + _ => {} // WHY: every other variant is already logged by `process_packet`, which runs unconditionally before this call — nothing else concerns the processor. + } + } + + /// Builds the `PacketProcessor`, seeded with every node already known to + /// the collector (typically handshake-discovered nodes) so the processor + /// does not start blind to nodes learned before its construction (#198). + async fn make_processor( + &self, + tx: broadcast::Sender, + ) -> Arc> { + let seeded = self.node_db.lock().await.clone(); + Arc::new(Mutex::new(PacketProcessor::new( + seeded, + MeshTopology::new(), + tx, + ))) + } + /// Connects to all configured transports and performs handshakes. /// /// # Errors @@ -407,12 +449,10 @@ impl Collector for MeshCollector { // kanon:ignore ARCHITECTURE/trait-impl-coloc return Ok(()); } - // 3. Create packet processor (owns topology graph, emits GeoSignals). - let processor = Arc::new(Mutex::new(PacketProcessor::new( - NodeDb::new(), - MeshTopology::new(), - tx.clone(), - ))); + // 3. Create packet processor (owns topology graph, emits GeoSignals), + // seeded with every node the handshake already discovered so it does + // not start blind to nodes learned before this point (#198). + let processor = self.make_processor(tx.clone()).await; // 4–7. Start heartbeat, gateway health, discovery, and router tasks. let mut tasks: JoinSet> = JoinSet::new(); @@ -443,13 +483,7 @@ impl Collector for MeshCollector { // kanon:ignore ARCHITECTURE/trait-impl-coloc // Update node_db for CLI display. self.process_packet(&from_radio).await; // Dispatch to PacketProcessor for topology + GeoSignal emission. - if let Some(from_radio::PayloadVariant::Packet(pkt)) = - &from_radio.payload_variant - { - processor.lock().await.process_mesh_packet(pkt); - // Inbound ACK/NAK: update the delivery tracker + outbound queue. - self.dispatch_routing(pkt).await; - } + self.dispatch_to_processor(&from_radio, &processor).await; } Err(e) => { tracing::warn!(error = %e, "receive error"); diff --git a/crates/kerykeion/src/collector_tests.rs b/crates/kerykeion/src/collector_tests.rs index e9a7480..741c911 100644 --- a/crates/kerykeion/src/collector_tests.rs +++ b/crates/kerykeion/src/collector_tests.rs @@ -140,6 +140,66 @@ async fn process_empty_payload_is_noop() { ); } +// WHY (#198): `PacketProcessor` owns topology + GeoSignal emission and must +// see every node the collector knows about, including ones learned only via +// a runtime `NodeInfo` frame with no subsequent mesh packet. + +#[tokio::test] +async fn dispatch_to_processor_routes_runtime_nodeinfo_to_processor_node_db() { + let c = MeshCollector::new(make_config(vec![])); + let processor = c.make_processor(make_tx()).await; + + let from_radio = FromRadio { + id: 9, + payload_variant: Some(from_radio::PayloadVariant::NodeInfo( + crate::proto::NodeInfo { + num: 0x00C0_FFEE, + ..Default::default() + }, + )), + }; + + c.dispatch_to_processor(&from_radio, &processor).await; + + let guard = processor.lock().await; + let found = guard + .node_db() + .get(crate::types::NodeNum(0x00C0_FFEE)) + .is_some(); + drop(guard); + assert!( + found, + "node learned only via a runtime NodeInfo frame must reach the processor's NodeDb" + ); +} + +#[tokio::test] +async fn make_processor_is_seeded_with_nodes_already_known_to_the_collector() { + let c = MeshCollector::new(make_config(vec![])); + + // Simulate a handshake-discovered node landing in the collector's + // NodeDb before the processor is constructed. + c.node_db().lock().await.insert(crate::node_db::MeshNode { + num: crate::types::NodeNum(0xFEED), + user: None, + position: None, + metrics: None, + last_heard: None, + snr: None, + hop_count: None, + }); + + let processor = c.make_processor(make_tx()).await; + + let guard = processor.lock().await; + let found = guard.node_db().get(crate::types::NodeNum(0xFEED)).is_some(); + drop(guard); + assert!( + found, + "processor must be seeded with nodes the collector already knew about" + ); +} + #[test] fn compute_hop_count_valid() { assert_eq!(MeshCollector::compute_hop_count(3, 1), Some(2)); diff --git a/crates/kerykeion/src/node_db.rs b/crates/kerykeion/src/node_db.rs index 153a427..173525b 100644 --- a/crates/kerykeion/src/node_db.rs +++ b/crates/kerykeion/src/node_db.rs @@ -9,7 +9,7 @@ use serde::{Deserialize, Serialize}; use crate::types::{NodeIdStr, NodeNum}; /// In-memory store of all mesh nodes seen during a session. -#[derive(Debug, Default)] +#[derive(Debug, Default, Clone)] pub struct NodeDb { nodes: HashMap, my_node: Option,