protocol/service: Wire up `FetcherService`

Wire up the new `FetcherService`. This reduces the `Service` fetch
methods down to:
- `fetch`
- `fetched`
- `dequeue_fetches`
- `fetch_refs_at`

This simplifies the code by off-loading the logic to the `fetcher`
family of types, rather than handling the intricacies in the `Service`
itself.

The `Service` now just performs the necessary I/O based on the
returned events from the `fetcher`.

This also removes the fetching state from the node sessions. This
follows the single responsibility, and sessions now only care about
their connectivity.

Breaking changes:
- The `Connected` state of a peer no longer contains fetching
  information, which was being returned as part of the JSON payload
  result.
- The `rad debug` information for ongoing fetches contained the number
  listeners awaiting for results, this was removed.

CHANGELOG.md

@@ -27,6 +27,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   pushed the default branch to the local user's namespace. The command is now
   deprecated, and the user should use `git push` instead.
 
+## Breaking Changes
+
+- The `Connected` state of a peer no longer contains fetching information. This
+  information was returned when requesting for `Seeds` on the control socket.
+  Callers should no longer expect the `fetching` inside that JSON result.
+- The `rad debug` information for ongoing fetches contained the number of
+  subscribers awaiting for results, this was removed.
+
 ## 1.6.1
 
 ## Fixed Bugs

crates/radicle-node/src/runtime/handle.rs

@@ -350,20 +350,22 @@ impl radicle::node::Handle for Handle {
     fn debug(&self) -> Result<serde_json::Value, Self::Error> {
         let (sender, receiver) = chan::bounded(1);
         let query: Arc<QueryState> = Arc::new(move |state| {
+            let fetcher_state = state.fetching();
             let debug = serde_json::json!({
                 "outboxSize": state.outbox().len(),
-                "fetching": state.fetching().iter().map(|(rid, state)| {
+                "fetching": fetcher_state.active_fetches()
+                    .iter()
+                    .map(|(rid, active)| {
+                        json!({
+                            "rid": rid,
+                            "from": active.from(),
+                            "refsAt": active.refs_at(),
+                        })
+                    }).collect::<Vec<_>>(),
+                "queue": fetcher_state.queued_fetches().iter().map(|(node, queue)| {
                     json!({
-                        "rid": rid,
+                        "nid": node,
-                        "from": state.from,
+                        "queue": queue.iter().map(|fetch| {
-                        "refsAt": state.refs_at,
-                        "subscribers": state.subscribers.len(),
-                    })
-                }).collect::<Vec<_>>(),
-                "queue": state.sessions().values().map(|sess| {
-                    json!({
-                        "nid": sess.id,
-                        "queue": sess.queue.iter().map(|fetch| {
                             json!({
                                 "rid": fetch.rid,
                                 "from": fetch.from,

crates/radicle-node/src/tests.rs

@@ -1512,6 +1512,7 @@ fn test_queued_fetch_max_capacity() {
 
     // Finish the 1st fetch.
     alice.fetched(rid1, bob.id, Ok(fetch::FetchResult::new(doc.clone())));
+
     // Now the 1st fetch is done, the 2nd fetch is dequeued.
     assert_matches!(alice.fetches().next(), Some((rid, _)) if rid == rid2);
     // ... but not the third.

crates/radicle-protocol/src/fetcher/state.rs

@@ -280,8 +280,8 @@ impl Default for Config {
 /// An active fetch represents a repository being fetched by a particular node.
 #[derive(Clone, Debug, PartialEq, Eq)]
 pub struct ActiveFetch {
-    pub(super) from: NodeId,
+    pub from: NodeId,
-    pub(super) refs_at: Vec<RefsAt>,
+    pub refs_at: Vec<RefsAt>,
 }
 
 impl ActiveFetch {

crates/radicle-protocol/src/service.rs

@@ -38,6 +38,9 @@ use radicle::storage::refs::SIGREFS_BRANCH;
 use radicle::storage::RepositoryError;
 use radicle_fetch::policy::SeedingPolicy;
 
+use crate::fetcher;
+use crate::fetcher::service::FetcherService;
+use crate::fetcher::FetcherState;
 use crate::service::gossip::Store as _;
 use crate::service::message::{
     Announcement, AnnouncementMessage, Info, NodeAnnouncement, Ping, RefsAnnouncement, RefsStatus,
@@ -303,25 +306,6 @@ pub enum CommandError {
     Policy(#[from] policy::Error),
 }
 
-/// Error returned by [`Service::try_fetch`].
-#[derive(thiserror::Error, Debug)]
-enum TryFetchError<'a> {
-    #[error("ongoing fetch for repository exists")]
-    AlreadyFetching(&'a mut FetchState),
-    #[error("peer is not connected; cannot initiate fetch")]
-    SessionNotConnected,
-    #[error("peer fetch capacity reached; cannot initiate fetch")]
-    SessionCapacityReached,
-    #[error(transparent)]
-    Namespaces(Box<NamespacesError>),
-}
-
-impl From<NamespacesError> for TryFetchError<'_> {
-    fn from(e: NamespacesError) -> Self {
-        Self::Namespaces(Box::new(e))
-    }
-}
-
 /// Fetch state for an ongoing fetch.
 #[derive(Debug)]
 pub struct FetchState {
@@ -333,15 +317,6 @@ pub struct FetchState {
     pub subscribers: Vec<chan::Sender<FetchResult>>,
 }
 
-impl FetchState {
-    /// Add a subscriber to this fetch.
-    fn subscribe(&mut self, c: chan::Sender<FetchResult>) {
-        if !self.subscribers.iter().any(|s| s.same_channel(&c)) {
-            self.subscribers.push(c);
-        }
-    }
-}
-
 /// Holds all node stores.
 #[derive(Debug)]
 pub struct Stores<D>(D);
@@ -441,8 +416,7 @@ pub struct Service<D, S, G> {
     inventory: InventoryAnnouncement,
     /// Source of entropy.
     rng: Rng,
-    /// Ongoing fetches.
+    fetcher: FetcherService<chan::Sender<FetchResult>>,
-    fetching: HashMap<RepoId, FetchState>,
     /// Request/connection rate limiter.
     limiter: RateLimiter,
     /// Current seeded repositories bloom filter.
@@ -510,7 +484,15 @@ where
         let last_timestamp = node.timestamp;
         let clock = LocalTime::default(); // Updated on initialize.
         let inventory = gossip::inventory(clock.into(), []); // Updated on initialize.
-
+        let fetcher = {
+            let config = fetcher::Config::new()
+                .with_max_concurrency(
+                    std::num::NonZeroUsize::new(config.limits.fetch_concurrency.into())
+                        .expect("fetch concurrency was zero, must be at least 1"),
+                )
+                .with_max_capacity(fetcher::MaxQueueSize::default());
+            FetcherService::new(config)
+        };
         Self {
             config,
             storage,
@@ -524,7 +506,7 @@ where
             outbox: Outbox::default(),
             limiter,
             sessions,
-            fetching: HashMap::new(),
+            fetcher,
             filter: Filter::empty(),
             relayed_by: HashMap::default(),
             last_idle: LocalTime::default(),
@@ -625,6 +607,10 @@ where
         Events::from(self.emitter.subscribe())
     }
 
+    pub fn fetcher(&self) -> &FetcherState {
+        self.fetcher.state()
+    }
+
     /// Get I/O outbox.
     pub fn outbox(&mut self) -> &mut Outbox {
         &mut self.outbox
@@ -900,7 +886,7 @@ where
                 }
             },
             Command::Fetch(rid, seed, timeout, resp) => {
-                self.fetch(rid, seed, timeout, Some(resp));
+                self.fetch(rid, seed, vec![], timeout, Some(resp));
             }
             Command::Seed(rid, scope, resp) => {
                 // Update our seeding policy.
@@ -992,7 +978,8 @@ where
                 if status.want.is_empty() {
                     debug!(target: "service", "Skipping fetch for {rid}, all refs are already in storage");
                 } else {
-                    return self._fetch(rid, from, status.want, timeout, channel);
+                    self.fetch(rid, from, status.want, timeout, channel);
+                    return true;
                 }
             }
             Err(e) => {
@@ -1003,247 +990,176 @@ where
         false
     }
 
-    /// Initiate an outgoing fetch for some repository.
     fn fetch(
-        &mut self,
-        rid: RepoId,
-        from: NodeId,
-        timeout: time::Duration,
-        channel: Option<chan::Sender<FetchResult>>,
-    ) -> bool {
-        self._fetch(rid, from, vec![], timeout, channel)
-    }
-
-    fn _fetch(
         &mut self,
         rid: RepoId,
         from: NodeId,
         refs_at: Vec<RefsAt>,
         timeout: time::Duration,
         channel: Option<chan::Sender<FetchResult>>,
-    ) -> bool {
+    ) {
-        match self.try_fetch(rid, &from, refs_at.clone(), timeout) {
+        let session = {
-            Ok(fetching) => {
+            let reason = format!("peer {from} is not connected; cannot initiate fetch");
+            let Some(session) = self.sessions.get_mut(&from) else {
                 if let Some(c) = channel {
-                    fetching.subscribe(c);
+                    c.send(FetchResult::Failed { reason }).ok();
                 }
-                return true;
+                return;
-            }
+            };
-            Err(TryFetchError::AlreadyFetching(fetching)) => {
+            if !session.is_connected() {
-                // If we're already fetching the same refs from the requested peer, there's nothing
-                // to do, we simply add the supplied channel to the list of subscribers so that it
-                // is notified on completion. Otherwise, we queue a fetch with the requested peer.
-                if fetching.from == from && fetching.refs_at == refs_at {
-                    debug!(target: "service", "Ignoring redundant fetch of {rid} from {from}");
-
-                    if let Some(c) = channel {
-                        fetching.subscribe(c);
-                    }
-                } else {
-                    let fetch = QueuedFetch {
-                        rid,
-                        refs_at,
-                        from,
-                        timeout,
-                        channel,
-                    };
-                    debug!(target: "service", "Queueing fetch for {rid} with {from} (already fetching)..");
-
-                    self.queue_fetch(fetch);
-                }
-            }
-            Err(TryFetchError::SessionCapacityReached) => {
-                debug!(target: "service", "Fetch capacity reached for {from}, queueing {rid}..");
-                self.queue_fetch(QueuedFetch {
-                    rid,
-                    refs_at,
-                    from,
-                    timeout,
-                    channel,
-                });
-            }
-            Err(e) => {
                 if let Some(c) = channel {
-                    c.send(FetchResult::Failed {
+                    c.send(FetchResult::Failed { reason }).ok();
-                        reason: e.to_string(),
-                    })
-                    .ok();
                 }
+                return;
             }
-        }
+            session
-        false
-    }
-
-    fn queue_fetch(&mut self, fetch: QueuedFetch) {
-        let Some(s) = self.sessions.get_mut(&fetch.from) else {
-            log::debug!(target: "service", "Cannot queue fetch for unknown session {}", fetch.from);
-            return;
         };
-        if let Err(e) = s.queue_fetch(fetch) {
-            let fetch = e.inner();
-            log::debug!(target: "service", "Unable to queue fetch for {} with {}: {e}", &fetch.rid, &fetch.from);
-        }
-    }
 
-    // TODO: Buffer/throttle fetches.
+        let cmd = fetcher::state::command::Fetch {
-    fn try_fetch(
-        &mut self,
-        rid: RepoId,
-        from: &NodeId,
-        refs_at: Vec<RefsAt>,
-        timeout: time::Duration,
-    ) -> Result<&mut FetchState, TryFetchError<'_>> {
-        let from = *from;
-        let Some(session) = self.sessions.get_mut(&from) else {
-            return Err(TryFetchError::SessionNotConnected);
-        };
-        let fetching = self.fetching.entry(rid);
-
-        trace!(target: "service", "Trying to fetch {refs_at:?} for {rid}..");
-
-        let fetching = match fetching {
-            Entry::Vacant(fetching) => fetching,
-            Entry::Occupied(fetching) => {
-                // We're already fetching this repo from some peer.
-                return Err(TryFetchError::AlreadyFetching(fetching.into_mut()));
-            }
-        };
-        // Sanity check: We shouldn't be fetching from this session, since we return above if we're
-        // fetching from any session.
-        debug_assert!(!session.is_fetching(&rid));
-
-        if !session.is_connected() {
-            // This can happen if a session disconnects in the time between asking for seeds to
-            // fetch from, and initiating the fetch from one of those seeds.
-            return Err(TryFetchError::SessionNotConnected);
-        }
-        if session.is_at_capacity() {
-            // If we're already fetching multiple repos from this peer.
-            return Err(TryFetchError::SessionCapacityReached);
-        }
-
-        let fetching = fetching.insert(FetchState {
             from,
-            refs_at: refs_at.clone(),
-            subscribers: vec![],
-        });
-        self.outbox.fetch(
-            session,
             rid,
             refs_at,
             timeout,
-            self.config.limits.fetch_pack_receive,
+        };
-        );
+        let fetcher::service::FetchInitiated { event, rejected } = self.fetcher.fetch(cmd, channel);
 
-        Ok(fetching)
+        if let Some(c) = rejected {
+            c.send(FetchResult::Failed {
+                reason: "fetch queue at capacity".to_string(),
+            })
+            .ok();
+        }
+
+        match event {
+            fetcher::state::event::Fetch::Started {
+                rid,
+                from,
+                refs_at,
+                timeout,
+            } => {
+                debug!(target: "service", "Starting fetch for {rid} from {from}");
+                self.outbox.fetch(
+                    session,
+                    rid,
+                    refs_at,
+                    timeout,
+                    self.config.limits.fetch_pack_receive,
+                );
+            }
+            fetcher::state::event::Fetch::Queued { rid, from } => {
+                debug!(target: "service", "Queued fetch for {rid} from {from}");
+            }
+            fetcher::state::event::Fetch::AlreadyFetching { rid, from } => {
+                debug!(target: "service", "Already fetching {rid} from {from}");
+            }
+            fetcher::state::event::Fetch::QueueAtCapacity { rid, from, .. } => {
+                debug!(target: "service", "Queue at capacity for {from}, rejected {rid}");
+            }
+        }
     }
 
     pub fn fetched(
         &mut self,
         rid: RepoId,
-        remote: NodeId,
+        from: NodeId,
         result: Result<crate::worker::fetch::FetchResult, crate::worker::FetchError>,
     ) {
-        let Some(fetching) = self.fetching.remove(&rid) else {
+        let cmd = fetcher::state::command::Fetched { from, rid };
-            debug!(target: "service", "Received unexpected fetch result for {rid}, from {remote}");
+        let fetcher::service::FetchCompleted { event, subscribers } = self.fetcher.fetched(cmd);
-            return;
-        };
-        debug_assert_eq!(fetching.from, remote);
 
-        if let Some(s) = self.sessions.get_mut(&remote) {
+        // Dequeue next fetches
-            // Mark this RID as fetched for this session.
-            s.fetched(rid);
-        }
-
-        // Notify all fetch subscribers of the fetch result. This is used when the user requests
-        // a fetch via the CLI, for example.
-        for sub in &fetching.subscribers {
-            debug!(target: "service", "Found existing fetch request from {remote}, sending result..");
-
-            let result = match &result {
-                Ok(success) => FetchResult::Success {
-                    updated: success.updated.clone(),
-                    namespaces: success.namespaces.clone(),
-                    clone: success.clone,
-                },
-                Err(e) => FetchResult::Failed {
-                    reason: e.to_string(),
-                },
-            };
-            if sub.send(result).is_err() {
-                debug!(target: "service", "Failed to send fetch result for {rid} from {remote}..");
-            } else {
-                debug!(target: "service", "Sent fetch result for {rid} from {remote}..");
-            }
-        }
-
-        match result {
-            Ok(crate::worker::fetch::FetchResult {
-                updated,
-                canonical,
-                namespaces,
-                clone,
-                doc,
-            }) => {
-                info!(target: "service", "Fetched {rid} from {remote} successfully");
-                // Update our routing table in case this fetch was user-initiated and doesn't
-                // come from an announcement.
-                self.seed_discovered(rid, remote, self.clock.into());
-
-                for update in &updated {
-                    if update.is_skipped() {
-                        trace!(target: "service", "Ref skipped: {update} for {rid}");
-                    } else {
-                        debug!(target: "service", "Ref updated: {update} for {rid}");
-                    }
-                }
-                self.emitter.emit(Event::RefsFetched {
-                    remote,
-                    rid,
-                    updated: updated.clone(),
-                });
-                self.emitter
-                    .emit_all(canonical.into_iter().map(|(refname, target)| {
-                        Event::CanonicalRefUpdated {
-                            rid,
-                            refname,
-                            target,
-                        }
-                    }));
-
-                // Announce our new inventory if this fetch was a full clone.
-                // Only update and announce inventory for public repositories.
-                if clone && doc.is_public() {
-                    debug!(target: "service", "Updating and announcing inventory for cloned repository {rid}..");
-
-                    if let Err(e) = self.add_inventory(rid) {
-                        warn!(target: "service", "Failed to announce inventory for {rid}: {e}");
-                    }
-                }
-
-                // It's possible for a fetch to succeed but nothing was updated.
-                if updated.is_empty() || updated.iter().all(|u| u.is_skipped()) {
-                    debug!(target: "service", "Nothing to announce, no refs were updated..");
-                } else {
-                    // Finally, announce the refs. This is useful for nodes to know what we've synced,
-                    // beyond just knowing that we have added an item to our inventory.
-                    if let Err(e) = self.announce_refs(rid, doc.into(), namespaces, false) {
-                        warn!(target: "service", "Failed to announce new refs: {e}");
-                    }
-                }
-            }
-            Err(err) => {
-                warn!(target: "service", "Fetch failed for {rid} from {remote}: {err}");
-
-                // For now, we only disconnect the remote in case of timeout. In the future,
-                // there may be other reasons to disconnect.
-                if err.is_timeout() {
-                    self.outbox.disconnect(remote, DisconnectReason::Fetch(err));
-                }
-            }
-        }
-        // We can now try to dequeue more fetches.
         self.dequeue_fetches();
+
+        match event {
+            fetcher::state::event::Fetched::NotFound { from, rid } => {
+                debug!(target: "service", "Unexpected fetch result for {rid} from {from}");
+            }
+            fetcher::state::event::Fetched::Completed {
+                from,
+                rid,
+                refs_at: _,
+            } => {
+                // Notify responders
+                let fetch_result = match &result {
+                    Ok(success) => FetchResult::Success {
+                        updated: success.updated.clone(),
+                        namespaces: success.namespaces.clone(),
+                        clone: success.clone,
+                    },
+                    Err(e) => FetchResult::Failed {
+                        reason: e.to_string(),
+                    },
+                };
+                for responder in subscribers {
+                    responder.send(fetch_result.clone()).ok();
+                }
+                match result {
+                    Ok(crate::worker::fetch::FetchResult {
+                        updated,
+                        canonical,
+                        namespaces,
+                        clone,
+                        doc,
+                    }) => {
+                        info!(target: "service", "Fetched {rid} from {from} successfully");
+                        // Update our routing table in case this fetch was user-initiated and doesn't
+                        // come from an announcement.
+                        self.seed_discovered(rid, from, self.clock.into());
+
+                        for update in &updated {
+                            if update.is_skipped() {
+                                trace!(target: "service", "Ref skipped: {update} for {rid}");
+                            } else {
+                                debug!(target: "service", "Ref updated: {update} for {rid}");
+                            }
+                        }
+                        self.emitter.emit(Event::RefsFetched {
+                            remote: from,
+                            rid,
+                            updated: updated.clone(),
+                        });
+                        self.emitter
+                            .emit_all(canonical.into_iter().map(|(refname, target)| {
+                                Event::CanonicalRefUpdated {
+                                    rid,
+                                    refname,
+                                    target,
+                                }
+                            }));
+
+                        // Announce our new inventory if this fetch was a full clone.
+                        // Only update and announce inventory for public repositories.
+                        if clone && doc.is_public() {
+                            debug!(target: "service", "Updating and announcing inventory for cloned repository {rid}..");
+
+                            if let Err(e) = self.add_inventory(rid) {
+                                warn!(target: "service", "Failed to announce inventory for {rid}: {e}");
+                            }
+                        }
+
+                        // It's possible for a fetch to succeed but nothing was updated.
+                        if updated.is_empty() || updated.iter().all(|u| u.is_skipped()) {
+                            debug!(target: "service", "Nothing to announce, no refs were updated..");
+                        } else {
+                            // Finally, announce the refs. This is useful for nodes to know what we've synced,
+                            // beyond just knowing that we have added an item to our inventory.
+                            if let Err(e) = self.announce_refs(rid, doc.into(), namespaces, false) {
+                                warn!(target: "service", "Failed to announce new refs: {e}");
+                            }
+                        }
+                    }
+                    Err(err) => {
+                        warn!(target: "service", "Fetch failed for {rid} from {from}: {err}");
+
+                        // For now, we only disconnect the from in case of timeout. In the future,
+                        // there may be other reasons to disconnect.
+                        if err.is_timeout() {
+                            self.outbox.disconnect(from, DisconnectReason::Fetch(err));
+                        }
+                    }
+                }
+            }
+        }
     }
 
     /// Attempt to dequeue fetches from all peers.
@@ -1260,38 +1176,42 @@ where
             .map(|(k, _)| *k)
             .collect::<Vec<_>>();
 
-        // Try to dequeue once per session.
         for nid in sessions {
-            // SAFETY: All the keys we are iterating on exist.
             #[allow(clippy::unwrap_used)]
             let sess = self.sessions.get_mut(&nid).unwrap();
-            if !sess.is_connected() || sess.is_at_capacity() {
+            if !sess.is_connected() {
                 continue;
             }
 
-            if let Some(QueuedFetch {
+            let Some(fetcher::QueuedFetch {
                 rid,
                 from,
                 refs_at,
                 timeout,
-                channel,
+            }) = self.fetcher.dequeue(&nid)
-            }) = sess.dequeue_fetch()
+            else {
-            {
+                continue;
-                debug!(target: "service", "Dequeued fetch for {rid} from session {from}..");
+            };
 
-                if let Some(refs) = NonEmpty::from_vec(refs_at) {
+            // Check seeding policy
-                    let repo_entry = self.policies.seed_policy(&rid).expect(
+            let repo_entry = self
-                        "Service::dequeue_fetch: error accessing repo seeding configuration",
+                .policies
-                    );
+                .seed_policy(&rid)
-                    let SeedingPolicy::Allow { scope } = repo_entry.policy else {
+                .expect("error accessing repo seeding configuration");
-                        debug!(target: "service", "Repository {rid} is no longer seeded, skipping..");
+
-                        continue;
+            let SeedingPolicy::Allow { scope } = repo_entry.policy else {
-                    };
+                debug!(target: "service", "Repository {} no longer seeded, skipping", rid);
-                    self.fetch_refs_at(rid, from, refs, scope, timeout, channel);
+                continue;
-                } else {
+            };
-                    // If no refs are specified, always do a full fetch.
+
-                    self.fetch(rid, from, timeout, channel);
+            debug!(target: "service", "Dequeued fetch for {} from {}", rid, from);
-                }
+
+            // Channel is `None` in both cases since they will already be
+            // registered with the fetcher service.
+            if let Some(refs) = NonEmpty::from_vec(refs_at.clone()) {
+                self.fetch_refs_at(rid, from, refs, scope, timeout, None);
+            } else {
+                self.fetch(rid, from, refs_at, timeout, None);
             }
         }
     }
@@ -1393,7 +1313,6 @@ where
                         self.config.is_persistent(&remote),
                         self.rng.clone(),
                         self.clock,
-                        self.config.limits.clone(),
                     ));
                     self.outbox.write_all(peer, msgs);
                 }
@@ -1424,19 +1343,30 @@ where
         let link = session.link;
         let addr = session.addr.clone();
 
-        self.fetching.retain(|_, fetching| {
+        let cmd = fetcher::state::command::Cancel { from: remote };
-            if fetching.from != remote {
+        let fetcher::service::FetchesCancelled { event, orphaned } = self.fetcher.cancel(cmd);
-                return true;
+
+        match event {
+            fetcher::state::event::Cancel::Unexpected { from } => {
+                debug!(target: "service", "No fetches to cancel for {from}");
             }
-            // Remove and fail any pending fetches from this remote node.
+            fetcher::state::event::Cancel::Canceled {
-            for resp in &fetching.subscribers {
+                from,
-                resp.send(FetchResult::Failed {
+                active,
-                    reason: format!("disconnected: {reason}"),
+                queued,
+            } => {
+                debug!(target: "service", "Cancelled {} ongoing, {} queued for {from}", active.len(), queued.len());
+            }
+        }
+
+        // Notify orphaned responders
+        for (rid, responder) in orphaned {
+            responder
+                .send(FetchResult::Failed {
+                    reason: format!("failed fetch to {rid}, peer disconnected: {reason}"),
                 })
                 .ok();
-            }
+        }
-            false
-        });
 
         // Attempt to re-connect to persistent peers.
         if self.config.is_persistent(&remote) {
@@ -1653,7 +1583,7 @@ where
 
                 for rid in missing {
                     debug!(target: "service", "Missing seeded inventory {rid}; initiating fetch..");
-                    self.fetch(rid, *announcer, FETCH_TIMEOUT, None);
+                    self.fetch(rid, *announcer, vec![], FETCH_TIMEOUT, None);
                 }
                 return Ok(relay);
             }
@@ -2286,13 +2216,7 @@ where
         }
         self.sessions.insert(
             nid,
-            Session::outbound(
+            Session::outbound(nid, addr.clone(), persistent, self.rng.clone()),
-                nid,
-                addr.clone(),
-                persistent,
-                self.rng.clone(),
-                self.config.limits.clone(),
-            ),
         );
         self.outbox.connect(nid, addr);
 
@@ -2565,7 +2489,7 @@ where
                 Ok(seeds) => {
                     if let Some(connected) = NonEmpty::from_vec(seeds.connected().collect()) {
                         for seed in connected {
-                            self.fetch(rid, seed.nid, FETCH_TIMEOUT, None);
+                            self.fetch(rid, seed.nid, vec![], FETCH_TIMEOUT, None);
                         }
                     } else {
                         // TODO: We should make sure that this fetch is retried later, either
@@ -2719,7 +2643,7 @@ pub trait ServiceState {
     /// Get the existing sessions.
     fn sessions(&self) -> &Sessions;
     /// Get fetch state.
-    fn fetching(&self) -> &HashMap<RepoId, FetchState>;
+    fn fetching(&self) -> &FetcherState;
     /// Get outbox.
     fn outbox(&self) -> &Outbox;
     /// Get rate limiter.
@@ -2752,8 +2676,8 @@ where
         &self.sessions
     }
 
-    fn fetching(&self) -> &HashMap<RepoId, FetchState> {
+    fn fetching(&self) -> &FetcherState {
-        &self.fetching
+        self.fetcher.state()
     }
 
     fn outbox(&self) -> &Outbox {

crates/radicle-protocol/src/service/io.rs

@@ -138,8 +138,6 @@ impl Outbox {
         timeout: time::Duration,
         reader_limit: FetchPackSizeLimit,
     ) {
-        peer.fetching(rid);
-
         let refs_at = (!refs_at.is_empty()).then_some(refs_at);
 
         if let Some(refs_at) = &refs_at {

crates/radicle-protocol/src/service/session.rs

@@ -1,8 +1,7 @@
-use std::collections::{HashSet, VecDeque};
+use std::collections::VecDeque;
 use std::{fmt, time};
 
 use crossbeam_channel as chan;
-use radicle::node::config::Limits;
 use radicle::node::{FetchResult, Severity};
 use radicle::node::{Link, Timestamp};
 pub use radicle::node::{PingState, State};
@@ -111,8 +110,6 @@ pub struct Session {
     pub subscribe: Option<message::Subscribe>,
     /// Last time a message was received from the peer.
     pub last_active: LocalTime,
-    /// Fetch queue.
-    pub queue: VecDeque<QueuedFetch>,
 
     /// Connection attempts. For persistent peers, Tracks
     /// how many times we've attempted to connect. We reset this to zero
@@ -120,8 +117,6 @@ pub struct Session {
     attempts: usize,
     /// Source of entropy.
     rng: Rng,
-    /// Protocol limits.
-    limits: Limits,
 }
 
 impl fmt::Display for Session {
@@ -159,7 +154,7 @@ impl From<&Session> for radicle::node::Session {
 }
 
 impl Session {
-    pub fn outbound(id: NodeId, addr: Address, persistent: bool, rng: Rng, limits: Limits) -> Self {
+    pub fn outbound(id: NodeId, addr: Address, persistent: bool, rng: Rng) -> Self {
         Self {
             id,
             addr,
@@ -168,28 +163,18 @@ impl Session {
             subscribe: None,
             persistent,
             last_active: LocalTime::default(),
-            queue: VecDeque::with_capacity(MAX_FETCH_QUEUE_SIZE),
             attempts: 1,
             rng,
-            limits,
         }
     }
 
-    pub fn inbound(
+    pub fn inbound(id: NodeId, addr: Address, persistent: bool, rng: Rng, time: LocalTime) -> Self {
-        id: NodeId,
-        addr: Address,
-        persistent: bool,
-        rng: Rng,
-        time: LocalTime,
-        limits: Limits,
-    ) -> Self {
         Self {
             id,
             addr,
             state: State::Connected {
                 since: time,
                 ping: PingState::default(),
-                fetching: HashSet::default(),
                 latencies: VecDeque::default(),
                 stable: false,
             },
@@ -197,10 +182,8 @@ impl Session {
             subscribe: None,
             persistent,
             last_active: time,
-            queue: VecDeque::new(),
             attempts: 0,
             rng,
-            limits,
         }
     }
 
@@ -224,41 +207,6 @@ impl Session {
         matches!(self.state, State::Initial)
     }
 
-    pub fn is_at_capacity(&self) -> bool {
-        if let State::Connected { fetching, .. } = &self.state {
-            if fetching.len() >= self.limits.fetch_concurrency.into() {
-                return true;
-            }
-        }
-        false
-    }
-
-    pub fn is_fetching(&self, rid: &RepoId) -> bool {
-        if let State::Connected { fetching, .. } = &self.state {
-            return fetching.contains(rid);
-        }
-        false
-    }
-
-    /// Queue a fetch. Returns `true` if it was added to the queue, and `false` if
-    /// it already was present in the queue.
-    pub fn queue_fetch(&mut self, fetch: QueuedFetch) -> Result<(), QueueError> {
-        assert_eq!(fetch.from, self.id);
-
-        if self.queue.len() >= MAX_FETCH_QUEUE_SIZE {
-            return Err(QueueError::CapacityReached(fetch));
-        } else if self.queue.contains(&fetch) {
-            return Err(QueueError::Duplicate(fetch));
-        }
-        self.queue.push_back(fetch);
-
-        Ok(())
-    }
-
-    pub fn dequeue_fetch(&mut self) -> Option<QueuedFetch> {
-        self.queue.pop_front()
-    }
-
     pub fn attempts(&self) -> usize {
         self.attempts
     }
@@ -279,33 +227,6 @@ impl Session {
         }
     }
 
-    /// Mark this session as fetching the given RID.
-    ///
-    /// # Panics
-    ///
-    /// If it is already fetching that RID, or the session is disconnected.
-    pub fn fetching(&mut self, rid: RepoId) {
-        if let State::Connected { fetching, .. } = &mut self.state {
-            assert!(
-                fetching.insert(rid),
-                "Session must not already be fetching {rid}"
-            );
-        } else {
-            panic!(
-                "Attempting to fetch {rid} from disconnected session {}",
-                self.id
-            );
-        }
-    }
-
-    pub fn fetched(&mut self, rid: RepoId) {
-        if let State::Connected { fetching, .. } = &mut self.state {
-            if !fetching.remove(&rid) {
-                log::debug!(target: "service", "Fetched unknown repository {rid}");
-            }
-        }
-    }
-
     pub fn to_attempted(&mut self) {
         assert!(
             self.is_initial(),
@@ -324,7 +245,6 @@ impl Session {
         self.state = State::Connected {
             since,
             ping: PingState::default(),
-            fetching: HashSet::default(),
             latencies: VecDeque::default(),
             stable: false,
         };

crates/radicle-schemars/src/main.rs

@@ -87,7 +87,7 @@ fn print_schema() -> io::Result<()> {
                     #[schemars(with = "radicle::schemars_ext::crypto::PublicKey")]
                     radicle::node::NodeId,
                 ),
-                Config(radicle::node::Config),
+                Config(Box<radicle::node::Config>),
                 ListenAddrs(ListenAddrs),
                 ConnectResult(radicle::node::ConnectResult),
                 Success(radicle::node::Success),

crates/radicle/CHANGELOG.md

@@ -46,6 +46,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Removed
 
+- The data returned by `Seeds` contains `state`, which in turn contained the
+  field `fetching` for ongoing fetches of that node, if in the `Connected`
+  state. `Connected` no longer contains that field.
+
 ### Security
 
 ## 0.20.0

crates/radicle/src/node.rs

@@ -107,8 +107,6 @@ pub enum State {
         /// Ping state.
         #[serde(skip)]
         ping: PingState,
-        /// Ongoing fetches.
-        fetching: HashSet<RepoId>,
         /// Measured latencies for this peer.
         #[serde(skip)]
         latencies: VecDeque<LocalDuration>,
@@ -696,7 +694,7 @@ impl From<Vec<Seed>> for Seeds {
     }
 }
 
-#[derive(Clone, Debug, Serialize, Deserialize)]
+#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)]
 #[serde(tag = "status", rename_all = "camelCase")]
 #[cfg_attr(feature = "schemars", derive(schemars::JsonSchema))]
 pub enum FetchResult {
@@ -1486,7 +1484,6 @@ mod test {
             &serde_json::to_string(&CommandResult::Okay(State::Connected {
                 since: LocalTime::now(),
                 ping: Default::default(),
-                fetching: Default::default(),
                 latencies: VecDeque::default(),
                 stable: false,
             }))

crates/radicle/src/node/command.rs

@@ -313,7 +313,6 @@ mod test {
             &serde_json::to_string(&CommandResult::Okay(State::Connected {
                 since: LocalTime::now(),
                 ping: Default::default(),
-                fetching: Default::default(),
                 latencies: VecDeque::default(),
                 stable: false,
             }))
@@ -329,7 +328,7 @@ mod test {
         );
         assert_matches!(
             json::from_str::<CommandResult<Seeds>>(
-                r#"[{"nid":"z6MksmpU5b1dS7oaqF2bHXhQi1DWy2hB7Mh9CuN7y1DN6QSz","addrs":[{"addr":"seed.radicle.example.com:8776","source":"peer","lastSuccess":1699983994234,"lastAttempt":1699983994000,"banned":false}],"state":{"connected":{"since":1699983994,"fetching":[]}}}]"#
+                r#"[{"nid":"z6MksmpU5b1dS7oaqF2bHXhQi1DWy2hB7Mh9CuN7y1DN6QSz","addrs":[{"addr":"seed.radicle.example.com:8776","source":"peer","lastSuccess":1699983994234,"lastAttempt":1699983994000,"banned":false}],"state":{"connected":{"since":1699983994}}}]"#
             ),
             Ok(CommandResult::Okay(_))
         );