fix(sync): fix PeerData encoding, neighbor events, better & predictable tests (#1513)

## Description

* **fix(gossip): properly encode peer data.** #1506 introduced a bug:
The `PeerData` was encoded from the `PeerAddr` (including the peer id)
but decoded to `AddrInfo` (without the peer id). So it failed, and
dialing peers failed. It only did not matter much because most tests use
tickets separately, so do not rely on the `PeerData` gossip.
* **feat: expose neighbor events** through document subscriptions. for
now used to write better and less flakey tests. also useful for stats
like usecases, and potentially others.
* **tests: improve sync tests** and make `sync_full_basic` not flakey
anymore (hopefully). the main change is that for some events, we don't
care about the exact order in tests anymore, because the exact order is
too unpredictable timing-wise for things that happen concurrently.
instead they are matched in chunks.

## Change checklist

- [x] Self-review.
- [x] Documentation updates if relevant.
- [x] Tests if relevant.

Replaces #1512

Author: Frando

iroh-gossip/src/net.rs

@@ -15,7 +15,7 @@ use tokio::{
 use tracing::{debug, trace, warn};
 
 use self::util::{read_message, write_message, Dialer, Timers};
-use crate::proto::{self, Scope, TopicId};
+use crate::proto::{self, PeerData, Scope, TopicId};
 
 pub mod util;
 
@@ -347,10 +347,9 @@ impl Actor {
                     }
                 },
                 _ = self.on_endpoints_rx.changed() => {
-                    let info = self.endpoint.my_addr().await?;
-                    let peer_data = Bytes::from(postcard::to_stdvec(&info)?);
-
-                    self.handle_in_event(InEvent::UpdatePeerData(peer_data.into()), Instant::now()).await?;
+                    let addr = self.endpoint.my_addr().await?;
+                    let peer_data = encode_peer_data(&addr.info)?;
+                    self.handle_in_event(InEvent::UpdatePeerData(peer_data), Instant::now()).await?;
                 }
                 (peer_id, res) = self.dialer.next_conn() => {
                     match res {
@@ -385,7 +384,7 @@ impl Actor {
 
     async fn handle_to_actor_msg(&mut self, msg: ToActor, now: Instant) -> anyhow::Result<()> {
         let me = *self.state.me();
-        debug!(?me, "handle to_actor  {msg:?}");
+        trace!(?me, "handle to_actor  {msg:?}");
         match msg {
             ToActor::ConnIncoming(peer_id, origin, conn) => {
                 self.conns.insert(peer_id, conn.clone());
@@ -396,7 +395,7 @@ impl Actor {
                 // Spawn a task for this connection
                 let in_event_tx = self.in_event_tx.clone();
                 tokio::spawn(async move {
-                    debug!(?me, peer = ?peer_id, "connection established, start loop");
+                    debug!(?me, peer = ?peer_id, "connection established");
                     match connection_loop(peer_id, conn, origin, send_rx, &in_event_tx).await {
                         Ok(()) => {
                             debug!(?me, peer = ?peer_id, "connection closed without error")
@@ -514,8 +513,8 @@ impl Actor {
                     self.pending_sends.remove(&peer);
                     self.dialer.abort_dial(&peer);
                 }
-                OutEvent::PeerData(peer, data) => match postcard::from_bytes::<AddrInfo>(&data) {
-                    Err(err) => warn!("Failed to decode PeerData from {peer}: {err}"),
+                OutEvent::PeerData(peer, data) => match decode_peer_data(&data) {
+                    Err(err) => warn!("Failed to decode {data:?} from {peer}: {err}"),
                     Ok(info) => {
                         debug!(me = ?self.endpoint.peer_id(), peer = ?peer, "add known addrs: {info:?}");
                         let peer_addr = PeerAddr {
@@ -600,6 +599,15 @@ async fn connection_loop(
     Ok(())
 }
 
+fn encode_peer_data(info: &AddrInfo) -> anyhow::Result<PeerData> {
+    Ok(PeerData::new(postcard::to_stdvec(info)?))
+}
+
+fn decode_peer_data(peer_data: &PeerData) -> anyhow::Result<AddrInfo> {
+    let info = postcard::from_bytes(peer_data.as_bytes())?;
+    Ok(info)
+}
+
 #[cfg(test)]
 mod test {
     use std::time::Duration;

iroh-gossip/src/proto.rs

@@ -47,6 +47,7 @@
 
 use std::{fmt, hash::Hash};
 
+use bytes::Bytes;
 use serde::{de::DeserializeOwned, Deserialize, Serialize};
 
 mod hyparview;
@@ -80,20 +81,26 @@ impl<T> PeerIdentity for T where T: Hash + Eq + Copy + fmt::Debug + Serialize +
 ///
 /// Implementations may use these bytes to supply addresses or other information needed to connect
 /// to a peer that is not included in the peer's [`PeerIdentity`].
-#[derive(
-    derive_more::Debug,
-    Serialize,
-    Deserialize,
-    Clone,
-    PartialEq,
-    Eq,
-    derive_more::From,
-    derive_more::Into,
-    derive_more::Deref,
-    Default,
-)]
+#[derive(derive_more::Debug, Serialize, Deserialize, Clone, PartialEq, Eq, Default)]
 #[debug("PeerData({}b)", self.0.len())]
-pub struct PeerData(bytes::Bytes);
+pub struct PeerData(Bytes);
+
+impl PeerData {
+    /// Create a new [`PeerData`] from a byte buffer.
+    pub fn new(data: impl Into<Bytes>) -> Self {
+        Self(data.into())
+    }
+
+    /// Get a reference to the contained [`bytes::Bytes`].
+    pub fn inner(&self) -> &bytes::Bytes {
+        &self.0
+    }
+
+    /// Get the peer data as a byte slice.
+    pub fn as_bytes(&self) -> &[u8] {
+        &self.0
+    }
+}
 
 /// PeerInfo contains a peer's identifier and the opaque peer data as provided by the implementer.
 #[derive(Debug, Serialize, Deserialize, Clone, PartialEq, Eq)]

iroh/examples/sync.rs

@@ -291,6 +291,7 @@ async fn run(args: Args) -> anyhow::Result<()> {
                             println!("change: {}", fmt_entry(&entry));
                         }
                     }
+                    _ => {}
                 }
             }
         }

iroh/src/commands/sync.rs

@@ -265,6 +265,12 @@ impl DocCommands {
                                 ),
                             }
                         }
+                        LiveEvent::NeighborUp(peer) => {
+                            println!("neighbor peer up: {peer:?}");
+                        }
+                        LiveEvent::NeighborDown(peer) => {
+                            println!("neighbor peer down: {peer:?}");
+                        }
                     }
                 }
             }

iroh/src/sync_engine/live.rs

@@ -124,7 +124,7 @@ pub type OnLiveEventCallback =
     Box<dyn Fn(LiveEvent) -> BoxFuture<'static, KeepCallback> + Send + Sync + 'static>;
 
 /// Events informing about actions of the live sync progres.
-#[derive(Serialize, Deserialize, Debug, Clone)]
+#[derive(Serialize, Deserialize, Debug, Clone, Eq, PartialEq)]
 #[allow(clippy::large_enum_variant)]
 pub enum LiveEvent {
     /// A local insertion.
@@ -146,6 +146,10 @@ pub enum LiveEvent {
         /// The content hash of the newly available entry content
         hash: Hash,
     },
+    /// We have a new neighbor in the swarm.
+    NeighborUp(PublicKey),
+    /// We lost a neighbor in the swarm.
+    NeighborDown(PublicKey),
     /// A set-reconciliation sync finished.
     SyncFinished(SyncEvent),
 }
@@ -795,14 +799,17 @@ impl<S: store::Store, B: baomap::Store> Actor<S, B> {
                 let op: Op = postcard::from_bytes(&msg.content)?;
                 match op {
                     Op::Put(entry) => {
-                        debug!(peer = ?msg.delivered_from, topic = ?topic, "received entry via gossip");
-                        // If the distance is 0, we received the message from its original author.
-                        // In this case, assume that the peer can provide the content to us.
+                        debug!(peer = ?msg.delivered_from, ?namespace, "received entry via gossip");
+                        // Insert the entry into our replica.
+                        // If the message was broadcast with neighbor scope, or is received
+                        // directly from the author, we assume that the content is available at
+                        // that peer. Otherwise we don't.
+                        // The download is not triggered here, but in the `on_replica_event`
+                        // handler for the `InsertRemote` event.
                         let content_status = match msg.scope.is_direct() {
                             true => ContentStatus::Complete,
                             false => ContentStatus::Missing,
                         };
-                        // At this point, we do not know if the peer has the content.
                         replica.insert_remote_entry(
                             entry,
                             *msg.delivered_from.as_bytes(),
@@ -822,9 +829,18 @@ impl<S: store::Store, B: baomap::Store> Actor<S, B> {
             // [Self::sync_with_peer] will check to not resync with peers synced previously in the
             // same session. TODO: Maybe this is too broad and leads to too many sync requests.
             Event::NeighborUp(peer) => {
+                debug!(?peer, ?namespace, "neighbor up");
                 self.sync_with_peer(namespace, peer, SyncReason::NewNeighbor);
+                if let Some(subs) = self.event_subscriptions.get_mut(&namespace) {
+                    notify_all(subs, LiveEvent::NeighborUp(peer)).await;
+                }
+            }
+            Event::NeighborDown(peer) => {
+                debug!(?peer, ?namespace, "neighbor down");
+                if let Some(subs) = self.event_subscriptions.get_mut(&namespace) {
+                    notify_all(subs, LiveEvent::NeighborDown(peer)).await;
+                }
             }
-            _ => {}
         }
         Ok(())
     }
@@ -844,7 +860,7 @@ impl<S: store::Store, B: baomap::Store> Actor<S, B> {
                 // A new entry was inserted locally. Broadcast a gossip message.
                 let op = Op::Put(signed_entry);
                 let message = postcard::to_stdvec(&op)?.into();
-                debug!(topic = ?topic, "broadcast new entry");
+                debug!(?namespace, "broadcast new entry");
                 self.gossip.broadcast(topic, message).await?;
 
                 // Notify subscribers about the event