diff --git a/src/iroh.rs b/src/iroh.rs index cef2988..86c30bd 100644 --- a/src/iroh.rs +++ b/src/iroh.rs @@ -35,6 +35,7 @@ const ANNOUNCEMENT_TIMEOUT: std::time::Duration = std::time::Duration::from_secs const CONTROL_OPERATION_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10); const BLOB_OPERATION_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(5 * 60); const ONLINE_WAIT_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(30); +const BLOB_ACTIVITY_EVENT_BUFFER: usize = 32; #[derive(Clone, Debug)] pub struct FolderSession { @@ -80,7 +81,7 @@ pub struct NodeHost { store: FsStore, router: Router, folders: Arc>>, - discovered_peers: Arc>>, + discovered_peers: Arc>>, announced_folders: Arc>>, lan_discovered_peers: Arc>>, lan_discovery_task: Option>, @@ -104,11 +105,17 @@ impl NodeHost { .await?; let store = FsStore::load(&blob_directory).await?; let folders = Arc::new(RwLock::new(BTreeMap::new())); - let discovered_peers = Arc::new(Mutex::new(Vec::new())); + let discovered_peers = Arc::new(Mutex::new(BTreeMap::new())); let announced_folders = Arc::new(Mutex::new(BTreeMap::new())); let lan_discovered_peers = Arc::new(Mutex::new(BTreeMap::new())); let lan_discovery_task = if lan_discovery_enabled { - Some(start_lan_discovery(&endpoint, Arc::clone(&lan_discovered_peers)).await?) + match start_lan_discovery(&endpoint, Arc::clone(&lan_discovered_peers)).await { + Ok(task) => Some(task), + Err(error) => { + endpoint.close().await; + return Err(error); + } + } } else { None }; @@ -389,6 +396,8 @@ impl NodeHost { .lock() .unwrap_or_else(std::sync::PoisonError::into_inner), ) + .into_values() + .collect() } async fn connect_to_peer(&self, peer_address: EndpointAddr) -> anyhow::Result { @@ -463,6 +472,13 @@ fn summarize_announcement_outcomes( } } } + if report.unavailable_count > 0 { + tracing::warn!( + folder = %folder_id, + unavailable_count = report.unavailable_count, + "Some peers did not receive the folder change announcement" + ); + } report } @@ -513,7 +529,7 @@ fn blob_activity_events() -> EventSender { get_many: RequestMode::Notify, ..EventMask::DEFAULT }; - let (sender, mut receiver) = EventSender::channel(32, event_mask); + let (sender, mut receiver) = EventSender::channel(BLOB_ACTIVITY_EVENT_BUFFER, event_mask); tokio::spawn(async move { while let Some(message) = receiver.recv().await { if matches!( diff --git a/src/iroh/handler.rs b/src/iroh/handler.rs index d0819ab..d90ac75 100644 --- a/src/iroh/handler.rs +++ b/src/iroh/handler.rs @@ -22,7 +22,7 @@ const CONTROL_REQUEST_TIMEOUT: Duration = Duration::from_secs(10); #[derive(Clone, Debug)] pub(super) struct AppaProtocol { pub(super) folders: Arc>>, - pub(super) discovered_peers: Arc>>, + pub(super) discovered_peers: Arc>>, pub(super) announced_folders: Arc>>, } @@ -75,10 +75,13 @@ impl AppaProtocol { self.discovered_peers .lock() .unwrap_or_else(std::sync::PoisonError::into_inner) - .push(DiscoveredPeer { - folder_id, - endpoint: requester_endpoint, - }); + .insert( + (folder_id, requester_endpoint.id), + DiscoveredPeer { + folder_id, + endpoint: requester_endpoint, + }, + ); response } ControlMessage::ManifestAnnouncement {