wip: sync improvments

iroh-sync/src/sync.rs

@@ -422,6 +422,7 @@ impl<S: ranger::Store<SignedEntry> + PublicKeyStore + store::DownloadPolicyStore
         validate_entry(system_time_now(), store, namespace, &entry, &origin)?;
 
         let outcome = self.peer.put(entry.clone()).map_err(InsertError::Store)?;
+        tracing::debug!(?origin, hash = %entry.content_hash(), ?outcome, "insert");
 
         let removed_count = match outcome {
             InsertOutcome::Inserted { removed } => removed,
@@ -450,7 +451,7 @@ impl<S: ranger::Store<SignedEntry> + PublicKeyStore + store::DownloadPolicyStore
                 let download_policy = self
                     .peer
                     .store
-                    .get_download_policy(&self.capability().id())
+                    .get_download_policy(&self.id())
                     .unwrap_or_default();
                 let should_download = download_policy.matches(entry.entry());
                 Event::RemoteInsert {

iroh/src/sync_engine/gossip.rs

@@ -112,6 +112,7 @@ impl GossipActor {
                 return Ok(false);
             }
             ToGossipActor::Join { namespace, peers } => {
+                debug!(?namespace, peers = peers.len(), "join gossip");
                 // join gossip for the topic to receive and send message
                 let fut = self
                     .gossip

iroh/src/sync_engine/live.rs

@@ -342,6 +342,7 @@ impl<B: iroh_bytes::store::Store> LiveActor<B> {
     }
 
     async fn start_sync(&mut self, namespace: NamespaceId, mut peers: Vec<NodeAddr>) -> Result<()> {
+        debug!(?namespace, peers = peers.len(), "start sync");
         // update state to allow sync
         if !self.state.is_syncing(&namespace) {
             let opts = OpenOpts::default()
@@ -531,7 +532,7 @@ impl<B: iroh_bytes::store::Store> LiveActor<B> {
         };
 
         let result_for_event = match &result {
-            Ok(_) => Ok(()),
+            Ok(details) => Ok(details.into()),
             Err(err) => Err(err.to_string()),
         };
 
@@ -686,8 +687,10 @@ impl<B: iroh_bytes::store::Store> LiveActor<B> {
         };
         debug!("incoming connection");
         let sync = self.sync.clone();
-        self.running_sync_accept
-            .spawn(async move { handle_connection(sync, conn, accept_request_cb).await });
+        self.running_sync_accept.spawn(
+            async move { handle_connection(sync, conn, accept_request_cb).await }
+                .instrument(Span::current()),
+        );
     }
 
     pub fn accept_sync_request(
@@ -712,7 +715,24 @@ pub struct SyncEvent {
     /// Timestamp when the sync finished
     pub started: SystemTime,
     /// Result of the sync operation
-    pub result: std::result::Result<(), String>,
+    pub result: std::result::Result<SyncDetails, String>,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize, Eq, PartialEq)]
+pub struct SyncDetails {
+    /// Number of entries received
+    pub entries_received: usize,
+    /// Number of entries sent
+    pub entries_sent: usize,
+}
+
+impl From<&SyncFinished> for SyncDetails {
+    fn from(value: &SyncFinished) -> Self {
+        Self {
+            entries_received: value.outcome.num_recv,
+            entries_sent: value.outcome.num_sent,
+        }
+    }
 }
 
 #[derive(Debug, Default)]

iroh/src/sync_engine/state.rs

@@ -81,7 +81,6 @@ impl NamespaceStates {
         if state.start_connect(reason) {
             true
         } else {
-            debug!("abort connect: already syncing");
             false
         }
     }
@@ -170,6 +169,7 @@ impl PeerState {
     }
 
     fn start_connect(&mut self, reason: SyncReason) -> bool {
+        debug!(?reason, "start connect");
         match self.state {
             // never run two syncs at the same time
             SyncState::Running { .. } => {