feat(sync): track incoming sync requests, allow subscriptions without sync, close inactive replicas (#1491)

## Description

This PR adds a few missing parts for sync in iroh:

* Allow to subscribe to replicas if they're not in the set of syncing
replicas (this was not possible before).
* Properly track open replicas in the `LiveSync` and close them once
they're not needed anymore. Therefore adds
`iroh_sync::store::close_replica` to remove a replica instance from the
store-internal cache of open replicas. This is needed because otherwise
it grows unbounded (for anchor nodes especially).
* Pass incoming connections to the `LiveSync` actor and handle them
there. Before, incoming connections for sync were *always* processed if
a replica with the requested `NamespaceId` is found in the local replica
store. This PR changes this mechanism so that incoming sync connections
are passed into the sync engine first. This leads to the changes that
motivate this PR:
* Incoming sync requests are only answered if the document (replica) in
question was opened for sync before (via a call to `start_sync`). Note
that this is *not* persisted at the moment, but should be (separate
change though).
* We can track the sync status for outgoing *and incoming* sync flows.
This will reduce the number of sync connections, because currently we
will oftenly double-sync in both directions (unneeded, because the sync
is always bidirectional). Now, we abort if a sync is already in
progress.
* Expose events about successfull and failed sync runs to the upper
layer via the document subscriptions with a new
`LiveEvent:.SyncFinished` that reports all available information

## Notes & open questions

Do we want to track history of syncs internally or leave it up to the
app via the events? Not sure myself.

The diff of `sync_engine/live.rs` is a little hard to read, because
there's a change from `TopicId` to `NamespaceId` in a few places that is
mostly unrelated. Sorry for the mixup, but it makes sense to track
everything by `NamespaceId` and it ended up in a commit here.

Logging is much improved in this PR. 

Maybe we can improve the code flow a little to make things more obvious.

<!-- Any notes, remarks or open questions you have to make about the PR.
-->

## Change checklist

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

Author: Frando

iroh-sync/src/keys.rs

@@ -201,6 +201,18 @@ impl fmt::Debug for Namespace {
     }
 }
 
+impl fmt::Debug for NamespaceId {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(f, "NamespaceId({})", base32::fmt_short(self.0))
+    }
+}
+
+impl fmt::Debug for AuthorId {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(f, "AuthorId({})", base32::fmt_short(self.0))
+    }
+}
+
 impl fmt::Debug for Author {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         write!(f, "Author({})", self)
@@ -321,6 +333,14 @@ pub(super) mod base32 {
         text.make_ascii_lowercase();
         text
     }
+    /// Convert to a base32 string limited to the first 10 bytes
+    pub fn fmt_short(bytes: impl AsRef<[u8]>) -> String {
+        let len = bytes.as_ref().len().min(10);
+        let mut text = data_encoding::BASE32_NOPAD.encode(&bytes.as_ref()[..len]);
+        text.make_ascii_lowercase();
+        text.push('…');
+        text
+    }
     /// Parse from a base32 string into a byte array
     pub fn parse_array<const N: usize>(input: &str) -> anyhow::Result<[u8; N]> {
         data_encoding::BASE32_NOPAD
@@ -332,7 +352,6 @@ pub(super) mod base32 {
 
 /// [`NamespacePublicKey`] in bytes
 #[derive(
-    Debug,
     Default,
     Clone,
     Copy,
@@ -351,7 +370,6 @@ pub struct NamespaceId([u8; 32]);
 
 /// [`AuthorPublicKey`] in bytes
 #[derive(
-    Debug,
     Default,
     Clone,
     Copy,

iroh-sync/src/metrics.rs

@@ -13,8 +13,10 @@ pub struct Metrics {
     pub new_entries_remote: Counter,
     pub new_entries_local_size: Counter,
     pub new_entries_remote_size: Counter,
-    pub initial_sync_success: Counter,
-    pub initial_sync_failed: Counter,
+    pub sync_via_connect_success: Counter,
+    pub sync_via_connect_failure: Counter,
+    pub sync_via_accept_success: Counter,
+    pub sync_via_accept_failure: Counter,
 }
 
 impl Default for Metrics {
@@ -24,8 +26,10 @@ impl Default for Metrics {
             new_entries_remote: Counter::new("Number of document entries added by peers"),
             new_entries_local_size: Counter::new("Total size of entry contents added locally"),
             new_entries_remote_size: Counter::new("Total size of entry contents added by peers"),
-            initial_sync_success: Counter::new("Number of successfull initial syncs "),
-            initial_sync_failed: Counter::new("Number of failed initial syncs"),
+            sync_via_accept_success: Counter::new("Number of successfull syncs (via accept)"),
+            sync_via_accept_failure: Counter::new("Number of failed syncs (via accept)"),
+            sync_via_connect_success: Counter::new("Number of successfull syncs (via connect)"),
+            sync_via_connect_failure: Counter::new("Number of failed syncs (via connect)"),
         }
     }
 }

iroh-sync/src/net.rs

@@ -1,6 +1,6 @@
 //! Network implementation of the iroh-sync protocol
 
-use std::net::SocketAddr;
+use std::{future::Future, net::SocketAddr};
 
 use anyhow::{Context, Result};
 use iroh_net::{key::PublicKey, magic_endpoint::get_peer_id, MagicEndpoint};
@@ -10,6 +10,7 @@ use crate::{
     net::codec::{run_alice, run_bob},
     store,
     sync::Replica,
+    NamespaceId,
 };
 
 #[cfg(feature = "metrics")]
@@ -26,52 +27,176 @@ mod codec;
 pub async fn connect_and_sync<S: store::Store>(
     endpoint: &MagicEndpoint,
     doc: &Replica<S::Instance>,
-    peer_id: PublicKey,
+    peer: PublicKey,
     derp_region: Option<u16>,
     addrs: &[SocketAddr],
-) -> anyhow::Result<()> {
-    debug!("sync with peer {}: start", peer_id);
+) -> Result<()> {
+    debug!(peer = ?peer, "sync (via connect): start");
+    let namespace = doc.namespace();
     let connection = endpoint
-        .connect(peer_id, SYNC_ALPN, derp_region, addrs)
+        .connect(peer, SYNC_ALPN, derp_region, addrs)
         .await
-        .context("dial_and_sync")?;
+        .context("failed to establish connection")?;
+    debug!(?peer, ?namespace, "sync (via connect): connected");
     let (mut send_stream, mut recv_stream) = connection.open_bi().await?;
-    let res = run_alice::<S, _, _>(&mut send_stream, &mut recv_stream, doc, peer_id).await;
+    let res = run_alice::<S, _, _>(&mut send_stream, &mut recv_stream, doc, peer).await;
+    send_stream.finish().await?;
+    recv_stream.read_to_end(0).await?;
 
     #[cfg(feature = "metrics")]
     if res.is_ok() {
-        inc!(Metrics, initial_sync_success);
+        inc!(Metrics, sync_via_connect_success);
     } else {
-        inc!(Metrics, initial_sync_failed);
+        inc!(Metrics, sync_via_connect_failure);
     }
 
-    debug!("sync with peer {}: finish {:?}", peer_id, res);
+    debug!(peer = ?peer, ?res, "sync (via connect): done");
     res
 }
 
+/// What to do with incoming sync requests
+#[derive(Debug)]
+pub enum AcceptOutcome<S: store::Store> {
+    /// This namespace is not available for sync.
+    NotAvailable,
+    /// This namespace is already syncing, therefore abort.
+    AlreadySyncing,
+    /// Accept the sync request.
+    Accept(Replica<S::Instance>),
+}
+
+impl<S: store::Store> From<Option<Replica<S::Instance>>> for AcceptOutcome<S> {
+    fn from(replica: Option<Replica<S::Instance>>) -> Self {
+        match replica {
+            Some(replica) => AcceptOutcome::Accept(replica),
+            None => AcceptOutcome::NotAvailable,
+        }
+    }
+}
+
 /// Handle an iroh-sync connection and sync all shared documents in the replica store.
-pub async fn handle_connection<S: store::Store>(
+pub async fn handle_connection<S, F, Fut>(
     connecting: quinn::Connecting,
-    replica_store: S,
-) -> Result<()> {
-    let connection = connecting.await?;
-    let peer_id = get_peer_id(&connection).await?;
-    let (mut send_stream, mut recv_stream) = connection.accept_bi().await?;
-    debug!(peer = ?peer_id, "incoming sync: start");
+    accept_cb: F,
+) -> std::result::Result<(NamespaceId, PublicKey), SyncError>
+where
+    S: store::Store,
+    F: Fn(NamespaceId, PublicKey) -> Fut,
+    Fut: Future<Output = anyhow::Result<AcceptOutcome<S>>>,
+{
+    let connection = connecting.await.map_err(SyncError::connect)?;
+    let peer = get_peer_id(&connection).await.map_err(SyncError::connect)?;
+    let (mut send_stream, mut recv_stream) = connection
+        .accept_bi()
+        .await
+        .map_err(|error| SyncError::open(peer, error))?;
+    debug!(peer = ?peer, "sync (via accept): start");
 
-    let res = run_bob(&mut send_stream, &mut recv_stream, replica_store, peer_id).await;
+    let res = run_bob::<S, _, _, _, _>(&mut send_stream, &mut recv_stream, accept_cb, peer).await;
 
     #[cfg(feature = "metrics")]
     if res.is_ok() {
-        inc!(Metrics, initial_sync_success);
+        inc!(Metrics, sync_via_accept_success);
     } else {
-        inc!(Metrics, initial_sync_failed);
+        inc!(Metrics, sync_via_accept_failure);
     }
 
-    res?;
-    send_stream.finish().await?;
+    debug!(peer = ?peer, ?res, "sync (via accept): done");
+
+    let namespace = res?;
+    send_stream
+        .finish()
+        .await
+        .map_err(|error| SyncError::close(peer, namespace, error))?;
+    recv_stream
+        .read_to_end(0)
+        .await
+        .map_err(|error| SyncError::close(peer, namespace, error))?;
+    Ok((namespace, peer))
+}
 
-    debug!(peer = ?peer_id, "incoming sync: done");
+/// Failure reasons for sync.
+#[derive(thiserror::Error, Debug)]
+#[allow(missing_docs)]
+pub enum SyncError {
+    /// Failed to establish connection
+    #[error("Failed to establish connection")]
+    Connect {
+        #[source]
+        error: anyhow::Error,
+    },
+    /// Failed to open replica
+    #[error("Failed to open replica with {peer:?}")]
+    Open {
+        peer: PublicKey,
+        #[source]
+        error: anyhow::Error,
+    },
+    /// Failed to run sync
+    #[error("Failed to sync {namespace:?} with {peer:?}")]
+    Sync {
+        peer: PublicKey,
+        namespace: Option<NamespaceId>,
+        #[source]
+        error: anyhow::Error,
+    },
+    /// Failed to close
+    #[error("Failed to close {namespace:?} with {peer:?}")]
+    Close {
+        peer: PublicKey,
+        namespace: NamespaceId,
+        #[source]
+        error: anyhow::Error,
+    },
+}
 
-    Ok(())
+impl SyncError {
+    fn connect(error: impl Into<anyhow::Error>) -> Self {
+        Self::Connect {
+            error: error.into(),
+        }
+    }
+    fn open(peer: PublicKey, error: impl Into<anyhow::Error>) -> Self {
+        Self::Open {
+            peer,
+            error: error.into(),
+        }
+    }
+    pub(crate) fn sync(
+        peer: PublicKey,
+        namespace: Option<NamespaceId>,
+        error: impl Into<anyhow::Error>,
+    ) -> Self {
+        Self::Sync {
+            peer,
+            namespace,
+            error: error.into(),
+        }
+    }
+    fn close(peer: PublicKey, namespace: NamespaceId, error: impl Into<anyhow::Error>) -> Self {
+        Self::Close {
+            peer,
+            namespace,
+            error: error.into(),
+        }
+    }
+    /// Get the peer's node ID (if available)
+    pub fn peer(&self) -> Option<PublicKey> {
+        match self {
+            SyncError::Connect { .. } => None,
+            SyncError::Open { peer, .. } => Some(*peer),
+            SyncError::Sync { peer, .. } => Some(*peer),
+            SyncError::Close { peer, .. } => Some(*peer),
+        }
+    }
+
+    /// Get the namespace (if available)
+    pub fn namespace(&self) -> Option<NamespaceId> {
+        match self {
+            SyncError::Connect { .. } => None,
+            SyncError::Open { .. } => None,
+            SyncError::Sync { namespace, .. } => namespace.to_owned(),
+            SyncError::Close { namespace, .. } => Some(*namespace),
+        }
+    }
 }