make things work

Author: Frando

src/actor.rs

@@ -649,7 +649,7 @@ impl Actor {
                     break reply;
                 }
                 action => {
-                    if self.on_action(action).is_err() {
+                    if self.on_action(action).await.is_err() {
                         warn!("failed to send reply: receiver dropped");
                     }
                 }
@@ -667,7 +667,7 @@ impl Actor {
         }
     }
 
-    fn on_action(&mut self, action: Action) -> Result<(), SendReplyError> {
+    async fn on_action(&mut self, action: Action) -> Result<(), SendReplyError> {
         match action {
             Action::Shutdown { .. } => {
                 unreachable!("Shutdown is handled in run()")
@@ -711,11 +711,11 @@ impl Actor {
                 send_reply_with(reply, self, |this| this.store.content_hashes())
             }
             Action::FlushStore { reply } => send_reply(reply, self.store.flush()),
-            Action::Replica(namespace, action) => self.on_replica_action(namespace, action),
+            Action::Replica(namespace, action) => self.on_replica_action(namespace, action).await,
         }
     }
 
-    fn on_replica_action(
+    async fn on_replica_action(
         &mut self,
         namespace: NamespaceId,
         action: ReplicaAction,
@@ -801,13 +801,19 @@ impl Actor {
                 from,
                 mut state,
                 reply,
-            } => send_reply_with(reply, self, move |this| {
-                let mut replica = this
-                    .states
-                    .replica_if_syncing(&namespace, &mut this.store)?;
-                let res = replica.sync_process_message(message, from, &mut state)?;
-                Ok((res, state))
-            }),
+            } => {
+                let res = async {
+                    let mut replica = self
+                        .states
+                        .replica_if_syncing(&namespace, &mut self.store)?;
+                    let res = replica
+                        .sync_process_message(message, from, &mut state)
+                        .await?;
+                    Ok((res, state))
+                }
+                .await;
+                reply.send(res).map_err(send_reply_error)
+            }
             ReplicaAction::GetSyncPeers { reply } => send_reply_with(reply, self, move |this| {
                 this.states.ensure_open(&namespace)?;
                 let peers = this.store.get_sync_peers(&namespace)?;

src/engine.rs

@@ -3,7 +3,6 @@
 //! [`crate::Replica`] is also called documents here.
 
 use std::{
-    io,
     path::PathBuf,
     str::FromStr,
     sync::{Arc, RwLock},
@@ -76,13 +75,13 @@ impl Engine {
         let me = endpoint.node_id().fmt_short();
 
         let content_status_cb: ContentStatusCallback = {
-            let bao_store = bao_store.clone();
+            let blobs = bao_store.blobs().clone();
             Arc::new(move |hash: iroh_blobs::Hash| {
-                let fut = Box::pin(async move {
-                    let blob_status = bao_store.blobs().status(hash).await;
+                let blobs = blobs.clone();
+                Box::pin(async move {
+                    let blob_status = blobs.status(hash).await;
                     entry_to_content_status(blob_status)
-                });
-                fut
+                })
             })
         };
         let sync = SyncHandle::spawn(replica_store, Some(content_status_cb.clone()), me.clone());
@@ -205,25 +204,26 @@ impl Engine {
         &self,
         namespace: NamespaceId,
     ) -> Result<impl Stream<Item = Result<LiveEvent>> + Unpin + 'static> {
-        let content_status_cb = self.content_status_cb.clone();
-
         // Create a future that sends channel senders to the respective actors.
         // We clone `self` so that the future does not capture any lifetimes.
-        let this = self;
+        let content_status_cb = self.content_status_cb.clone();
 
         // Subscribe to insert events from the replica.
         let a = {
             let (s, r) = async_channel::bounded(SUBSCRIBE_CHANNEL_CAP);
-            this.sync.subscribe(namespace, s).await?;
-            Box::pin(r).map(move |ev| LiveEvent::from_replica_event(ev, &content_status_cb))
+            self.sync.subscribe(namespace, s).await?;
+            Box::pin(r).then(move |ev| {
+                let content_status_cb = content_status_cb.clone();
+                Box::pin(async move { LiveEvent::from_replica_event(ev, &content_status_cb).await })
+            })
         };
 
         // Subscribe to events from the [`live::Actor`].
         let b = {
             let (s, r) = async_channel::bounded(SUBSCRIBE_CHANNEL_CAP);
             let r = Box::pin(r);
             let (reply, reply_rx) = oneshot::channel();
-            this.to_live_actor
+            self.to_live_actor
                 .send(ToLiveActor::Subscribe {
                     namespace,
                     sender: s,
@@ -259,8 +259,8 @@ impl Engine {
 /// Converts an [`BlobStatus`] into a ['ContentStatus'].
 fn entry_to_content_status(entry: irpc::Result<BlobStatus>) -> ContentStatus {
     match entry {
-        Ok(BlobStatus::Complete) => ContentStatus::Complete,
-        Ok(BlobStatus::Partial) => ContentStatus::Incomplete,
+        Ok(BlobStatus::Complete { .. }) => ContentStatus::Complete,
+        Ok(BlobStatus::Partial { .. }) => ContentStatus::Incomplete,
         Ok(BlobStatus::NotFound) => ContentStatus::Missing,
         Err(cause) => {
             tracing::warn!("Error while checking entry status: {cause:?}");
@@ -322,7 +322,7 @@ impl From<live::Event> for LiveEvent {
 }
 
 impl LiveEvent {
-    fn from_replica_event(
+    async fn from_replica_event(
         ev: crate::Event,
         content_status_cb: &ContentStatusCallback,
     ) -> Result<Self> {
@@ -331,7 +331,7 @@ impl LiveEvent {
                 entry: entry.into(),
             },
             crate::Event::RemoteInsert { entry, from, .. } => Self::InsertRemote {
-                content_status: content_status_cb(entry.content_hash()),
+                content_status: content_status_cb(entry.content_hash()).await,
                 entry: entry.into(),
                 from: PublicKey::from_bytes(&from)?,
             },

src/engine/live.rs

@@ -12,10 +12,9 @@ use iroh::{Endpoint, NodeAddr, NodeId, PublicKey};
 use iroh_blobs::{
     api::{
         blobs::BlobStatus,
-        downloader::{DownloadRequest, Downloader},
+        downloader::{ContentDiscovery, DownloadRequest, Downloader, SplitStrategy},
         Store,
     },
-    get::Stats,
     Hash, HashAndFormat,
 };
 use iroh_gossip::net::Gossip;
@@ -148,7 +147,7 @@ type SyncConnectRes = (
     Result<SyncFinished, ConnectError>,
 );
 type SyncAcceptRes = Result<SyncFinished, AcceptError>;
-type DownloadRes = (NamespaceId, Hash, Result<Stats, DownloadError>);
+type DownloadRes = (NamespaceId, Hash, Result<(), anyhow::Error>);
 
 // Currently peers might double-sync in both directions.
 pub struct LiveActor {
@@ -177,6 +176,8 @@ pub struct LiveActor {
     missing_hashes: HashSet<Hash>,
     /// Content hashes queued in downloader.
     queued_hashes: QueuedHashes,
+    /// Nodes known to have a hash
+    hash_providers: ProviderNodes,
 
     /// Subscribers to actor events
     subscribers: SubscribersMap,
@@ -217,6 +218,7 @@ impl LiveActor {
             state: Default::default(),
             missing_hashes: Default::default(),
             queued_hashes: Default::default(),
+            hash_providers: Default::default(),
             metrics,
         }
     }
@@ -629,7 +631,7 @@ impl LiveActor {
         }
     }
 
-    async fn broadcast_neighbors(&self, namespace: NamespaceId, op: &Op) {
+    async fn broadcast_neighbors(&mut self, namespace: NamespaceId, op: &Op) {
         if !self.state.is_syncing(&namespace) {
             return;
         }
@@ -651,7 +653,7 @@ impl LiveActor {
         &mut self,
         namespace: NamespaceId,
         hash: Hash,
-        res: Result<Stats, DownloadError>,
+        res: Result<(), anyhow::Error>,
     ) {
         let completed_namespaces = self.queued_hashes.remove_hash(&hash);
         debug!(namespace=%namespace.fmt_short(), success=res.is_ok(), completed_namespaces=completed_namespaces.len(), "download ready");
@@ -758,12 +760,22 @@ impl LiveActor {
             self.missing_hashes.remove(&hash);
             return;
         }
+        self.hash_providers
+            .0
+            .lock()
+            .expect("poisoned")
+            .entry(hash)
+            .or_default()
+            .insert(node);
         if self.queued_hashes.contains_hash(&hash) {
             self.queued_hashes.insert(hash, namespace);
-            self.downloader.nodes_have(hash, vec![node]).await;
         } else if !only_if_missing || self.missing_hashes.contains(&hash) {
-            let req = DownloadRequest::new(HashAndFormat::raw(hash), vec![node]);
-            let handle = self.downloader.queue(req).await;
+            let req = DownloadRequest::new(
+                HashAndFormat::raw(hash),
+                self.hash_providers.clone(),
+                SplitStrategy::None,
+            );
+            let handle = self.downloader.download_with_opts(req);
 
             self.queued_hashes.insert(hash, namespace);
             self.missing_hashes.remove(&hash);
@@ -885,6 +897,24 @@ struct QueuedHashes {
     by_namespace: HashMap<NamespaceId, HashSet<Hash>>,
 }
 
+#[derive(Debug, Clone, Default)]
+struct ProviderNodes(Arc<std::sync::Mutex<HashMap<Hash, HashSet<NodeId>>>>);
+
+impl ContentDiscovery for ProviderNodes {
+    fn find_providers(&self, hash: HashAndFormat) -> n0_future::stream::Boxed<NodeId> {
+        let nodes = self
+            .0
+            .lock()
+            .expect("poisoned")
+            .get(&hash.hash)
+            .into_iter()
+            .flatten()
+            .cloned()
+            .collect::<Vec<_>>();
+        Box::pin(n0_future::stream::iter(nodes))
+    }
+}
+
 impl QueuedHashes {
     fn insert(&mut self, hash: Hash, namespace: NamespaceId) {
         self.by_hash.entry(hash).or_default().insert(namespace);