diff --git a/node/relay/src/event_feedback.rs b/node/relay/src/event_feedback.rs new file mode 100644 index 0000000..a9c176d --- /dev/null +++ b/node/relay/src/event_feedback.rs @@ -0,0 +1,721 @@ +use libp2p::{Multiaddr, PeerId}; +use serde::{Deserialize, Serialize}; +use std::collections::HashMap; +use std::sync::Arc; +use std::time::{Duration, Instant}; +use tokio::sync::{mpsc, oneshot, Mutex}; +use log::{debug, warn}; + +/// Unified result type for all relay events +#[derive(Debug, Clone, Serialize, Deserialize)] +pub enum EventResult { + // Circuit events + CircuitAccepted { + src_peer_id: String, + dst_peer_id: String, + established_at: u64, + }, + CircuitClosed { + src_peer_id: String, + dst_peer_id: String, + duration_ms: u64, + error: Option, + success: bool, + }, + CircuitDenied { + src_peer_id: String, + dst_peer_id: String, + status: String, + reason: String, + }, + + // Reservation events + ReservationAccepted { + peer_id: String, + renewed: bool, + expires_at: u64, + }, + ReservationClosed { + peer_id: String, + duration_ms: u64, + reason: String, + }, + ReservationDenied { + peer_id: String, + status: String, + reason: String, + }, + ReservationTimedOut { + peer_id: String, + duration_ms: u64, + }, + + // Connection events + ConnectionEstablished { + peer_id: String, + duration_ms: u64, + endpoint: String, + }, + ConnectionClosed { + peer_id: String, + duration_ms: u64, + cause: String, + }, + ConnectionDialing { + peer_id: String, + started_at: u64, + }, + ConnectionError { + peer_id: Option, + error: String, + }, + + // Listener events + ListenerStarted { + address: String, + }, + ListenerClosed { + listener_id: String, + addresses: Vec, + reason: String, + }, + ListenerExpired { + address: String, + }, + + // External address events + ExternalAddrCandidate { + address: String, + }, + ExternalAddrConfirmed { + address: String, + }, + ExternalAddrExpired { + address: String, + }, + ExternalAddrOfPeer { + peer_id: String, + address: String, + }, + + // Generic event for unhandled cases + UnknownEvent { + event_type: String, + details: String, + }, +} + +impl EventResult { + pub fn is_success(&self) -> bool { + match self { + EventResult::CircuitClosed { success, .. } => *success, + EventResult::CircuitDenied { .. } => false, + EventResult::ReservationDenied { .. } => false, + EventResult::ReservationTimedOut { .. } => false, + EventResult::ConnectionError { .. } => false, + _ => true, // Most events represent successful operations + } + } + + pub fn get_peer_id(&self) -> Option { + match self { + EventResult::CircuitAccepted { src_peer_id, .. } => Some(src_peer_id.clone()), + EventResult::CircuitClosed { src_peer_id, .. } => Some(src_peer_id.clone()), + EventResult::CircuitDenied { src_peer_id, .. } => Some(src_peer_id.clone()), + EventResult::ReservationAccepted { peer_id, .. } => Some(peer_id.clone()), + EventResult::ReservationClosed { peer_id, .. } => Some(peer_id.clone()), + EventResult::ReservationDenied { peer_id, .. } => Some(peer_id.clone()), + EventResult::ReservationTimedOut { peer_id, .. } => Some(peer_id.clone()), + EventResult::ConnectionEstablished { peer_id, .. } => Some(peer_id.clone()), + EventResult::ConnectionClosed { peer_id, .. } => Some(peer_id.clone()), + EventResult::ConnectionDialing { peer_id, .. } => Some(peer_id.clone()), + EventResult::ConnectionError { peer_id, .. } => peer_id.clone(), + EventResult::ExternalAddrOfPeer { peer_id, .. } => Some(peer_id.clone()), + _ => None, + } + } +} + +/// Event subscription identifier +#[derive(Debug, Clone, Copy, Hash, PartialEq, Eq)] +pub enum EventSubscription { + // Circuit subscriptions (by peer pair) + Circuit { src: PeerId, dst: PeerId }, + + // Reservation subscriptions (by peer) + Reservation { peer: PeerId }, + + // Connection subscriptions (by peer) + Connection { peer: PeerId }, + + // Listener subscriptions (by address) + Listener { address_hash: u64 }, + + // Global subscriptions (all events of a type) + AllCircuits, + AllReservations, + AllConnections, + AllListeners, + AllExternalAddrs, + AllEvents, +} + +impl EventSubscription { + pub fn circuit_key(src: PeerId, dst: PeerId) -> Self { + EventSubscription::Circuit { src, dst } + } + + pub fn reservation_key(peer: PeerId) -> Self { + EventSubscription::Reservation { peer } + } + + pub fn connection_key(peer: PeerId) -> Self { + EventSubscription::Connection { peer } + } +} + +/// Tracking info for active events +#[derive(Debug)] +struct EventTrackingInfo { + started_at: Instant, + metadata: HashMap, + feedback_txs: Vec>, +} + +/// Central event feedback manager +pub struct EventFeedbackManager { + // Active event tracking + active_circuits: Arc>>, + active_reservations: Arc>>, + active_connections: Arc>>, + + // Broadcast channels for global subscriptions + circuit_broadcast: mpsc::UnboundedSender, + reservation_broadcast: mpsc::UnboundedSender, + connection_broadcast: mpsc::UnboundedSender, + listener_broadcast: mpsc::UnboundedSender, + external_addr_broadcast: mpsc::UnboundedSender, + all_events_broadcast: mpsc::UnboundedSender, +} + +impl EventFeedbackManager { + pub fn new() -> Self { + let (circuit_tx, _) = mpsc::unbounded_channel(); + let (reservation_tx, _) = mpsc::unbounded_channel(); + let (connection_tx, _) = mpsc::unbounded_channel(); + let (listener_tx, _) = mpsc::unbounded_channel(); + let (external_addr_tx, _) = mpsc::unbounded_channel(); + let (all_events_tx, _) = mpsc::unbounded_channel(); + + Self { + active_circuits: Arc::new(Mutex::new(HashMap::new())), + active_reservations: Arc::new(Mutex::new(HashMap::new())), + active_connections: Arc::new(Mutex::new(HashMap::new())), + circuit_broadcast: circuit_tx, + reservation_broadcast: reservation_tx, + connection_broadcast: connection_tx, + listener_broadcast: listener_tx, + external_addr_broadcast: external_addr_tx, + all_events_broadcast: all_events_tx, + } + } + + /// Subscribe to a specific event or event type + pub async fn subscribe(&self, subscription: EventSubscription) -> mpsc::UnboundedReceiver { + let (tx, rx) = mpsc::unbounded_channel(); + + match subscription { + EventSubscription::AllCircuits => { + // Clone the sender and spawn a task to forward events + let mut broadcast_rx = self.subscribe_broadcast(&self.circuit_broadcast).await; + tokio::spawn(async move { + while let Some(event) = broadcast_rx.recv().await { + if tx.send(event).is_err() { + break; + } + } + }); + } + EventSubscription::AllReservations => { + let mut broadcast_rx = self.subscribe_broadcast(&self.reservation_broadcast).await; + tokio::spawn(async move { + while let Some(event) = broadcast_rx.recv().await { + if tx.send(event).is_err() { + break; + } + } + }); + } + EventSubscription::AllConnections => { + let mut broadcast_rx = self.subscribe_broadcast(&self.connection_broadcast).await; + tokio::spawn(async move { + while let Some(event) = broadcast_rx.recv().await { + if tx.send(event).is_err() { + break; + } + } + }); + } + EventSubscription::AllListeners => { + let mut broadcast_rx = self.subscribe_broadcast(&self.listener_broadcast).await; + tokio::spawn(async move { + while let Some(event) = broadcast_rx.recv().await { + if tx.send(event).is_err() { + break; + } + } + }); + } + EventSubscription::AllExternalAddrs => { + let mut broadcast_rx = self.subscribe_broadcast(&self.external_addr_broadcast).await; + tokio::spawn(async move { + while let Some(event) = broadcast_rx.recv().await { + if tx.send(event).is_err() { + break; + } + } + }); + } + EventSubscription::AllEvents => { + let mut broadcast_rx = self.subscribe_broadcast(&self.all_events_broadcast).await; + tokio::spawn(async move { + while let Some(event) = broadcast_rx.recv().await { + if tx.send(event).is_err() { + break; + } + } + }); + } + _ => { + // For specific event subscriptions, we'll handle in the event methods + warn!("Specific event subscription requires using subscribe_once methods"); + } + } + + rx + } + + async fn subscribe_broadcast(&self, tx: &mpsc::UnboundedSender) -> mpsc::UnboundedReceiver { + let (new_tx, rx) = mpsc::unbounded_channel(); + // In production, maintain a list of subscribers + rx + } + + /// Subscribe to a specific circuit (one-time feedback) + pub async fn subscribe_circuit_once( + &self, + src_peer: PeerId, + dst_peer: PeerId, + ) -> oneshot::Receiver { + let (tx, rx) = oneshot::channel(); + let key = (src_peer, dst_peer); + + let mut circuits = self.active_circuits.lock().await; + circuits.entry(key) + .or_insert_with(|| EventTrackingInfo { + started_at: Instant::now(), + metadata: HashMap::new(), + feedback_txs: Vec::new(), + }) + .feedback_txs + .push(tx); + + rx + } + + /// Subscribe to a specific reservation (one-time feedback) + pub async fn subscribe_reservation_once( + &self, + peer: PeerId, + ) -> oneshot::Receiver { + let (tx, rx) = oneshot::channel(); + + let mut reservations = self.active_reservations.lock().await; + reservations.entry(peer) + .or_insert_with(|| EventTrackingInfo { + started_at: Instant::now(), + metadata: HashMap::new(), + feedback_txs: Vec::new(), + }) + .feedback_txs + .push(tx); + + rx + } + + /// Subscribe to a specific connection (one-time feedback) + pub async fn subscribe_connection_once( + &self, + peer: PeerId, + ) -> oneshot::Receiver { + let (tx, rx) = oneshot::channel(); + + let mut connections = self.active_connections.lock().await; + connections.entry(peer) + .or_insert_with(|| EventTrackingInfo { + started_at: Instant::now(), + metadata: HashMap::new(), + feedback_txs: Vec::new(), + }) + .feedback_txs + .push(tx); + + rx + } + + // ============ Circuit Event Handlers ============ + + pub async fn notify_circuit_accepted(&self, src_peer: PeerId, dst_peer: PeerId) { + let key = (src_peer, dst_peer); + let now = Instant::now(); + + let mut circuits = self.active_circuits.lock().await; + circuits.insert(key, EventTrackingInfo { + started_at: now, + metadata: HashMap::new(), + feedback_txs: Vec::new(), + }); + + let result = EventResult::CircuitAccepted { + src_peer_id: src_peer.to_string(), + dst_peer_id: dst_peer.to_string(), + established_at: now.elapsed().as_millis() as u64, + }; + + self.broadcast_event(result).await; + } + + pub async fn notify_circuit_closed( + &self, + src_peer: PeerId, + dst_peer: PeerId, + error: Option, + ) { + let key = (src_peer, dst_peer); + let mut circuits = self.active_circuits.lock().await; + + if let Some(info) = circuits.remove(&key) { + let duration_ms = info.started_at.elapsed().as_millis() as u64; + let success = error.is_none(); + + let result = EventResult::CircuitClosed { + src_peer_id: src_peer.to_string(), + dst_peer_id: dst_peer.to_string(), + duration_ms, + error: error.clone(), + success, + }; + + // Notify specific subscribers + for tx in info.feedback_txs { + let _ = tx.send(result.clone()); + } + + // Broadcast to all subscribers + self.broadcast_event(result).await; + } + } + + pub async fn notify_circuit_denied( + &self, + src_peer: PeerId, + dst_peer: PeerId, + status: String, + ) { + let key = (src_peer, dst_peer); + let mut circuits = self.active_circuits.lock().await; + + if let Some(info) = circuits.remove(&key) { + let result = EventResult::CircuitDenied { + src_peer_id: src_peer.to_string(), + dst_peer_id: dst_peer.to_string(), + status: status.clone(), + reason: format!("Circuit denied: {}", status), + }; + + for tx in info.feedback_txs { + let _ = tx.send(result.clone()); + } + + self.broadcast_event(result).await; + } + } + + // ============ Reservation Event Handlers ============ + + pub async fn notify_reservation_accepted(&self, peer: PeerId, renewed: bool) { + let now = Instant::now(); + let expires_at = now.elapsed().as_secs() + 3600; // 1 hour + + let mut reservations = self.active_reservations.lock().await; + reservations.insert(peer, EventTrackingInfo { + started_at: now, + metadata: HashMap::new(), + feedback_txs: Vec::new(), + }); + + let result = EventResult::ReservationAccepted { + peer_id: peer.to_string(), + renewed, + expires_at, + }; + + self.broadcast_event(result).await; + } + + pub async fn notify_reservation_closed(&self, peer: PeerId, reason: String) { + let mut reservations = self.active_reservations.lock().await; + + if let Some(info) = reservations.remove(&peer) { + let duration_ms = info.started_at.elapsed().as_millis() as u64; + + let result = EventResult::ReservationClosed { + peer_id: peer.to_string(), + duration_ms, + reason: reason.clone(), + }; + + for tx in info.feedback_txs { + let _ = tx.send(result.clone()); + } + + self.broadcast_event(result).await; + } + } + + pub async fn notify_reservation_denied(&self, peer: PeerId, status: String) { + let mut reservations = self.active_reservations.lock().await; + + if let Some(info) = reservations.remove(&peer) { + let result = EventResult::ReservationDenied { + peer_id: peer.to_string(), + status: status.clone(), + reason: format!("Reservation denied: {}", status), + }; + + for tx in info.feedback_txs { + let _ = tx.send(result.clone()); + } + + self.broadcast_event(result).await; + } + } + + pub async fn notify_reservation_timed_out(&self, peer: PeerId) { + let mut reservations = self.active_reservations.lock().await; + + if let Some(info) = reservations.remove(&peer) { + let duration_ms = info.started_at.elapsed().as_millis() as u64; + + let result = EventResult::ReservationTimedOut { + peer_id: peer.to_string(), + duration_ms, + }; + + for tx in info.feedback_txs { + let _ = tx.send(result.clone()); + } + + self.broadcast_event(result).await; + } + } + + // ============ Connection Event Handlers ============ + + pub async fn notify_connection_established( + &self, + peer: PeerId, + duration_ms: u64, + endpoint: String, + ) { + let now = Instant::now(); + + let mut connections = self.active_connections.lock().await; + connections.insert(peer, EventTrackingInfo { + started_at: now, + metadata: HashMap::new(), + feedback_txs: Vec::new(), + }); + + let result = EventResult::ConnectionEstablished { + peer_id: peer.to_string(), + duration_ms, + endpoint, + }; + + self.broadcast_event(result).await; + } + + pub async fn notify_connection_closed(&self, peer: PeerId, cause: String) { + let mut connections = self.active_connections.lock().await; + + if let Some(info) = connections.remove(&peer) { + let duration_ms = info.started_at.elapsed().as_millis() as u64; + + let result = EventResult::ConnectionClosed { + peer_id: peer.to_string(), + duration_ms, + cause: cause.clone(), + }; + + for tx in info.feedback_txs { + let _ = tx.send(result.clone()); + } + + self.broadcast_event(result).await; + } + } + + pub async fn notify_connection_dialing(&self, peer: PeerId) { + let result = EventResult::ConnectionDialing { + peer_id: peer.to_string(), + started_at: Instant::now().elapsed().as_millis() as u64, + }; + + self.broadcast_event(result).await; + } + + pub async fn notify_connection_error(&self, peer: Option, error: String) { + let result = EventResult::ConnectionError { + peer_id: peer.map(|p| p.to_string()), + error, + }; + + self.broadcast_event(result).await; + } + + // ============ Listener Event Handlers ============ + + pub async fn notify_listener_started(&self, address: String) { + let result = EventResult::ListenerStarted { address }; + self.broadcast_event(result).await; + } + + pub async fn notify_listener_closed( + &self, + listener_id: String, + addresses: Vec, + reason: String, + ) { + let result = EventResult::ListenerClosed { + listener_id, + addresses, + reason, + }; + self.broadcast_event(result).await; + } + + pub async fn notify_listener_expired(&self, address: String) { + let result = EventResult::ListenerExpired { address }; + self.broadcast_event(result).await; + } + + // ============ External Address Event Handlers ============ + + pub async fn notify_external_addr_candidate(&self, address: String) { + let result = EventResult::ExternalAddrCandidate { address }; + self.broadcast_event(result).await; + } + + pub async fn notify_external_addr_confirmed(&self, address: String) { + let result = EventResult::ExternalAddrConfirmed { address }; + self.broadcast_event(result).await; + } + + pub async fn notify_external_addr_expired(&self, address: String) { + let result = EventResult::ExternalAddrExpired { address }; + self.broadcast_event(result).await; + } + + pub async fn notify_external_addr_of_peer(&self, peer: PeerId, address: String) { + let result = EventResult::ExternalAddrOfPeer { + peer_id: peer.to_string(), + address, + }; + self.broadcast_event(result).await; + } + + // ============ Generic Event Handler ============ + + pub async fn notify_unknown_event(&self, event_type: String, details: String) { + let result = EventResult::UnknownEvent { + event_type, + details, + }; + self.broadcast_event(result).await; + } + + // ============ Broadcast Helper ============ + + async fn broadcast_event(&self, event: EventResult) { + // Broadcast to category-specific channels + match &event { + EventResult::CircuitAccepted { .. } + | EventResult::CircuitClosed { .. } + | EventResult::CircuitDenied { .. } => { + let _ = self.circuit_broadcast.send(event.clone()); + } + EventResult::ReservationAccepted { .. } + | EventResult::ReservationClosed { .. } + | EventResult::ReservationDenied { .. } + | EventResult::ReservationTimedOut { .. } => { + let _ = self.reservation_broadcast.send(event.clone()); + } + EventResult::ConnectionEstablished { .. } + | EventResult::ConnectionClosed { .. } + | EventResult::ConnectionDialing { .. } + | EventResult::ConnectionError { .. } => { + let _ = self.connection_broadcast.send(event.clone()); + } + EventResult::ListenerStarted { .. } + | EventResult::ListenerClosed { .. } + | EventResult::ListenerExpired { .. } => { + let _ = self.listener_broadcast.send(event.clone()); + } + EventResult::ExternalAddrCandidate { .. } + | EventResult::ExternalAddrConfirmed { .. } + | EventResult::ExternalAddrExpired { .. } + | EventResult::ExternalAddrOfPeer { .. } => { + let _ = self.external_addr_broadcast.send(event.clone()); + } + _ => {} + } + + // Always broadcast to all-events channel + let _ = self.all_events_broadcast.send(event); + } + + // ============ Statistics & Monitoring ============ + + pub async fn get_active_circuits_count(&self) -> usize { + self.active_circuits.lock().await.len() + } + + pub async fn get_active_reservations_count(&self) -> usize { + self.active_reservations.lock().await.len() + } + + pub async fn get_active_connections_count(&self) -> usize { + self.active_connections.lock().await.len() + } + + pub async fn get_active_circuit_details(&self) -> Vec<(String, String, u64)> { + let circuits = self.active_circuits.lock().await; + circuits + .iter() + .map(|((src, dst), info)| { + ( + src.to_string(), + dst.to_string(), + info.started_at.elapsed().as_secs(), + ) + }) + .collect() + } +} + +impl Default for EventFeedbackManager { + fn default() -> Self { + Self::new() + } +} \ No newline at end of file diff --git a/node/relay/src/lib.rs b/node/relay/src/lib.rs index 5490205..49e7b95 100644 --- a/node/relay/src/lib.rs +++ b/node/relay/src/lib.rs @@ -1,8 +1,10 @@ mod metric_server; mod p2p; +mod event_feedback; use crate::metric_server::{metrics_server, MetricService}; use crate::p2p::RelayP2pWorker; +use crate::event_feedback::EventFeedbackManager; use anyhow::anyhow; use log::{error, info}; pub use std::net::SocketAddr; @@ -26,6 +28,9 @@ impl MainRelayServerService { // Create one shared metric service let metric_service = MetricService::new(); + // Create shared event feedback manager + let feedback_manager = Arc::new(EventFeedbackManager::new()); + // Start metrics server on a different port, using the same service let metrics_port = 9945; let metrics_service_clone = metric_service.clone(); @@ -38,9 +43,11 @@ impl MainRelayServerService { // Spawn P2P worker with pure Tokio instead of TaskManager let p2p_worker_clone = p2p_worker.clone(); let metrics_for_p2p = metric_service.clone(); + let feedback_for_p2p = feedback_manager.clone(); let p2p_handle = tokio::spawn(async move { let mut p2p_worker = p2p_worker_clone.lock().await; p2p_worker.metrics = std::sync::Arc::new(metrics_for_p2p); + p2p_worker.feedback_manager = feedback_for_p2p; let swarm_res = p2p_worker.start_swarm().await; if let Err(err) = swarm_res { error!("swarm start encountered error: caused by {err}"); diff --git a/node/relay/src/p2p.rs b/node/relay/src/p2p.rs index b77535e..4e34f82 100644 --- a/node/relay/src/p2p.rs +++ b/node/relay/src/p2p.rs @@ -12,6 +12,7 @@ use libp2p::swarm::{NetworkInfo, SwarmEvent}; use libp2p::{Multiaddr, PeerId, StreamProtocol, Swarm, SwarmBuilder}; use crate::metric_server::MetricService; +use crate::event_feedback::EventFeedbackManager; use anyhow::anyhow; use futures::future::Either; use libp2p::kad::Event as DhtEvent; @@ -46,6 +47,7 @@ pub struct RelayP2pWorker { pub external_multi_addr: Multiaddr, pub listening_addr: (Multiaddr, Multiaddr), pub metrics: Arc, + pub feedback_manager: Arc, } impl RelayP2pWorker { @@ -120,6 +122,7 @@ impl RelayP2pWorker { relay_swarm.add_external_address(external_multi_addr.clone()); let metrics = Arc::new(MetricService::new()); + let feedback_manager = Arc::new(EventFeedbackManager::new()); Ok(Self { peer_id, @@ -127,6 +130,7 @@ impl RelayP2pWorker { external_multi_addr, listening_addr: (ipv4_listening_addr, ipv6_listening_addr), metrics, + feedback_manager, }) } @@ -154,6 +158,7 @@ impl RelayP2pWorker { SwarmEvent::ConnectionEstablished { peer_id, established_in, + endpoint, .. } => { info!(target:"p2p","connection established {:?}: duration {:?}",peer_id,established_in); @@ -162,6 +167,13 @@ impl RelayP2pWorker { Some(&peer_id.to_string()), None, ); + + // Send feedback + self.feedback_manager.notify_connection_established( + peer_id, + established_in.as_millis() as u64, + format!("{:?}", endpoint), + ).await; } SwarmEvent::ConnectionClosed { peer_id, cause, .. } => { info!(target:"p2p","connection closed: {:?}, reason {:?}",peer_id,cause); @@ -170,6 +182,12 @@ impl RelayP2pWorker { Some(&peer_id.to_string()), None, ); + + // Send feedback + self.feedback_manager.notify_connection_closed( + peer_id, + format!("{:?}", cause), + ).await; } SwarmEvent::Dialing { peer_id, .. } => { info!(target:"p2p","dialing: {:?}",peer_id); @@ -179,10 +197,19 @@ impl RelayP2pWorker { Some(&pid.to_string()), None, ); + + // Send feedback + self.feedback_manager.notify_connection_dialing(pid).await; } } SwarmEvent::OutgoingConnectionError { peer_id, error, .. } => { info!(target:"p2p","outgoing connection error: {:?}, error {:?}",peer_id,error); + + // Send feedback + self.feedback_manager.notify_connection_error( + peer_id, + format!("{:?}", error), + ).await; } SwarmEvent::ListenerClosed { listener_id, @@ -191,29 +218,73 @@ impl RelayP2pWorker { .. } => { info!(target:"p2p","listener closed: {:?}, addresses {:?}, reason {:?}",listener_id,addresses,reason); + + // Send feedback + self.feedback_manager.notify_listener_closed( + format!("{:?}", listener_id), + addresses.iter().map(|a| a.to_string()).collect(), + format!("{:?}", reason), + ).await; } SwarmEvent::NewListenAddr { address, .. } => { debug!(target:"p2p","new listen addr: {:?}",address); + + // Send feedback + self.feedback_manager.notify_listener_started( + address.to_string(), + ).await; } SwarmEvent::ExpiredListenAddr { address, .. } => { debug!(target:"p2p","expired listen addr: {:?}",address); + + // Send feedback + self.feedback_manager.notify_listener_expired( + address.to_string(), + ).await; } SwarmEvent::NewExternalAddrCandidate { address, .. } => { debug!(target:"p2p","new external addr candidate: {:?}",address); + + // Send feedback + self.feedback_manager.notify_external_addr_candidate( + address.to_string(), + ).await; } SwarmEvent::ExternalAddrConfirmed { address, .. } => { debug!(target:"p2p","external addr confirmed: {:?}",address); + + // Send feedback + self.feedback_manager.notify_external_addr_confirmed( + address.to_string(), + ).await; } SwarmEvent::ExternalAddrExpired { address, .. } => { debug!(target:"p2p","external addr expired: {:?}",address); + + // Send feedback + self.feedback_manager.notify_external_addr_expired( + address.to_string(), + ).await; } SwarmEvent::NewExternalAddrOfPeer { peer_id, address, .. } => { info!(target:"p2p","new external addr of peer: {:?}, {:?}",peer_id,address); + + // Send feedback + self.feedback_manager.notify_external_addr_of_peer( + peer_id, + address.to_string(), + ).await; } _ => { info!(target:"p2p","swarm event: {:?}",swarm_event); + + // Send feedback for unknown events + self.feedback_manager.notify_unknown_event( + "unknown_swarm_event".to_string(), + format!("{:?}", swarm_event), + ).await; } } } @@ -234,6 +305,13 @@ impl RelayP2pWorker { &dst_peer_id.to_string(), error.as_ref().map(|e| format!("{:?}", e)).as_deref(), ); + + // Send feedback + self.feedback_manager.notify_circuit_closed( + src_peer_id, + dst_peer_id, + error.map(|e| format!("{:?}", e)), + ).await; } RelayServerEvent::CircuitReqAccepted { src_peer_id, @@ -247,6 +325,12 @@ impl RelayP2pWorker { &dst_peer_id.to_string(), None, ); + + // Send feedback + self.feedback_manager.notify_circuit_accepted( + src_peer_id, + dst_peer_id, + ).await; } RelayServerEvent::CircuitReqDenied { src_peer_id, @@ -256,6 +340,13 @@ impl RelayP2pWorker { info!(target:"p2p","relay message, circuit req denied status: {:?}, src_peer_id: {:?}, dst_peer_id: {:?}",status,src_peer_id,dst_peer_id); self.metrics .record_circuit_event("denied", Some(&src_peer_id.to_string()), None); + + // Send feedback + self.feedback_manager.notify_circuit_denied( + src_peer_id, + dst_peer_id, + format!("{:?}", status), + ).await; } RelayServerEvent::ReservationReqAccepted { src_peer_id, @@ -265,6 +356,12 @@ impl RelayP2pWorker { // Count only via peer-level recorder to avoid double counting self.metrics .record_reservation_event_peer("accepted", &src_peer_id.to_string()); + + // Send feedback + self.feedback_manager.notify_reservation_accepted( + src_peer_id, + renewed, + ).await; } RelayServerEvent::ReservationReqDenied { src_peer_id, @@ -276,6 +373,12 @@ impl RelayP2pWorker { Some(&src_peer_id.to_string()), None, ); + + // Send feedback + self.feedback_manager.notify_reservation_denied( + src_peer_id, + format!("{:?}", status), + ).await; } RelayServerEvent::ReservationTimedOut { src_peer_id } => { info!(target:"p2p","relay message, reservation timed out: {:?}",src_peer_id); @@ -284,16 +387,93 @@ impl RelayP2pWorker { Some(&src_peer_id.to_string()), None, ); + + // Send feedback + self.feedback_manager.notify_reservation_timed_out( + src_peer_id, + ).await; } RelayServerEvent::ReservationClosed { src_peer_id } => { info!(target:"p2p","relay message, reservation closed: {:?}",src_peer_id); // Count only via peer-level recorder to avoid double counting self.metrics .record_reservation_event_peer("closed", &src_peer_id.to_string()); + + // Send feedback + self.feedback_manager.notify_reservation_closed( + src_peer_id, + "graceful_close".to_string(), + ).await; } _ => { trace!(target:"p2p","relay message: {:?}",event); + + // Send feedback for unknown relay events + self.feedback_manager.notify_unknown_event( + "unknown_relay_event".to_string(), + format!("{:?}", event), + ).await; } } } + + // ========== PUBLIC FEEDBACK API ========== + + /// Subscribe to all circuit events + pub async fn subscribe_all_circuits(&self) -> tokio::sync::mpsc::UnboundedReceiver { + use crate::event_feedback::EventSubscription; + self.feedback_manager.subscribe(EventSubscription::AllCircuits).await + } + + /// Subscribe to all reservation events + pub async fn subscribe_all_reservations(&self) -> tokio::sync::mpsc::UnboundedReceiver { + use crate::event_feedback::EventSubscription; + self.feedback_manager.subscribe(EventSubscription::AllReservations).await + } + + /// Subscribe to all connection events + pub async fn subscribe_all_connections(&self) -> tokio::sync::mpsc::UnboundedReceiver { + use crate::event_feedback::EventSubscription; + self.feedback_manager.subscribe(EventSubscription::AllConnections).await + } + + /// Subscribe to all events + pub async fn subscribe_all_events(&self) -> tokio::sync::mpsc::UnboundedReceiver { + use crate::event_feedback::EventSubscription; + self.feedback_manager.subscribe(EventSubscription::AllEvents).await + } + + /// Subscribe to a specific circuit (one-time feedback when closed/denied) + pub async fn subscribe_circuit( + &self, + src_peer: PeerId, + dst_peer: PeerId, + ) -> tokio::sync::oneshot::Receiver { + self.feedback_manager.subscribe_circuit_once(src_peer, dst_peer).await + } + + /// Subscribe to a specific reservation (one-time feedback when closed/denied/timeout) + pub async fn subscribe_reservation( + &self, + peer: PeerId, + ) -> tokio::sync::oneshot::Receiver { + self.feedback_manager.subscribe_reservation_once(peer).await + } + + /// Subscribe to a specific connection (one-time feedback when closed/error) + pub async fn subscribe_connection( + &self, + peer: PeerId, + ) -> tokio::sync::oneshot::Receiver { + self.feedback_manager.subscribe_connection_once(peer).await + } + + /// Get statistics on active events + pub async fn get_feedback_stats(&self) -> (usize, usize, usize) { + ( + self.feedback_manager.get_active_circuits_count().await, + self.feedback_manager.get_active_reservations_count().await, + self.feedback_manager.get_active_connections_count().await, + ) + } }