diff --git a/CHANGELOG.md b/CHANGELOG.md index a125864d..71296aa3 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/crates/radicle-node/src/runtime/handle.rs b/crates/radicle-node/src/runtime/handle.rs index e695a3d1..d866bf07 100644 --- a/crates/radicle-node/src/runtime/handle.rs +++ b/crates/radicle-node/src/runtime/handle.rs @@ -350,20 +350,22 @@ impl radicle::node::Handle for Handle { fn debug(&self) -> Result { let (sender, receiver) = chan::bounded(1); let query: Arc = 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::>(), + "queue": fetcher_state.queued_fetches().iter().map(|(node, queue)| { json!({ - "rid": rid, - "from": state.from, - "refsAt": state.refs_at, - "subscribers": state.subscribers.len(), - }) - }).collect::>(), - "queue": state.sessions().values().map(|sess| { - json!({ - "nid": sess.id, - "queue": sess.queue.iter().map(|fetch| { + "nid": node, + "queue": queue.iter().map(|fetch| { json!({ "rid": fetch.rid, "from": fetch.from, diff --git a/crates/radicle-node/src/tests.rs b/crates/radicle-node/src/tests.rs index 3f0c7922..6f20ed44 100644 --- a/crates/radicle-node/src/tests.rs +++ b/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. diff --git a/crates/radicle-protocol/src/fetcher/state.rs b/crates/radicle-protocol/src/fetcher/state.rs index 6f64ff6d..38b1c357 100644 --- a/crates/radicle-protocol/src/fetcher/state.rs +++ b/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(super) refs_at: Vec, + pub from: NodeId, + pub refs_at: Vec, } impl ActiveFetch { diff --git a/crates/radicle-protocol/src/service.rs b/crates/radicle-protocol/src/service.rs index 713c07d3..2dce4eca 100644 --- a/crates/radicle-protocol/src/service.rs +++ b/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), -} - -impl From 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>, } -impl FetchState { - /// Add a subscriber to this fetch. - fn subscribe(&mut self, c: chan::Sender) { - if !self.subscribers.iter().any(|s| s.same_channel(&c)) { - self.subscribers.push(c); - } - } -} - /// Holds all node stores. #[derive(Debug)] pub struct Stores(D); @@ -441,8 +416,7 @@ pub struct Service { inventory: InventoryAnnouncement, /// Source of entropy. rng: Rng, - /// Ongoing fetches. - fetching: HashMap, + fetcher: FetcherService>, /// 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>, - ) -> bool { - self._fetch(rid, from, vec![], timeout, channel) - } - - fn _fetch( &mut self, rid: RepoId, from: NodeId, refs_at: Vec, timeout: time::Duration, channel: Option>, - ) -> bool { - match self.try_fetch(rid, &from, refs_at.clone(), timeout) { - Ok(fetching) => { + ) { + let session = { + 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; - } - Err(TryFetchError::AlreadyFetching(fetching)) => { - // 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) => { + return; + }; + if !session.is_connected() { if let Some(c) = channel { - c.send(FetchResult::Failed { - reason: e.to_string(), - }) - .ok(); + c.send(FetchResult::Failed { reason }).ok(); } + return; } - } - 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; + session }; - 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. - fn try_fetch( - &mut self, - rid: RepoId, - from: &NodeId, - refs_at: Vec, - 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 { + let cmd = fetcher::state::command::Fetch { 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, ) { - let Some(fetching) = self.fetching.remove(&rid) else { - debug!(target: "service", "Received unexpected fetch result for {rid}, from {remote}"); - return; - }; - debug_assert_eq!(fetching.from, remote); + let cmd = fetcher::state::command::Fetched { from, rid }; + let fetcher::service::FetchCompleted { event, subscribers } = self.fetcher.fetched(cmd); - if let Some(s) = self.sessions.get_mut(&remote) { - // 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. + // Dequeue next 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::>(); - // 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, - }) = sess.dequeue_fetch() - { - debug!(target: "service", "Dequeued fetch for {rid} from session {from}.."); + }) = self.fetcher.dequeue(&nid) + else { + continue; + }; - if let Some(refs) = NonEmpty::from_vec(refs_at) { - let repo_entry = self.policies.seed_policy(&rid).expect( - "Service::dequeue_fetch: error accessing repo seeding configuration", - ); - let SeedingPolicy::Allow { scope } = repo_entry.policy else { - debug!(target: "service", "Repository {rid} is no longer seeded, skipping.."); - continue; - }; - self.fetch_refs_at(rid, from, refs, scope, timeout, channel); - } else { - // If no refs are specified, always do a full fetch. - self.fetch(rid, from, timeout, channel); - } + // Check seeding policy + let repo_entry = self + .policies + .seed_policy(&rid) + .expect("error accessing repo seeding configuration"); + + let SeedingPolicy::Allow { scope } = repo_entry.policy else { + debug!(target: "service", "Repository {} no longer seeded, skipping", rid); + continue; + }; + + 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| { - if fetching.from != remote { - return true; + let cmd = fetcher::state::command::Cancel { from: remote }; + let fetcher::service::FetchesCancelled { event, orphaned } = self.fetcher.cancel(cmd); + + 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. - for resp in &fetching.subscribers { - resp.send(FetchResult::Failed { - reason: format!("disconnected: {reason}"), + fetcher::state::event::Cancel::Canceled { + from, + active, + 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( - nid, - addr.clone(), - persistent, - self.rng.clone(), - self.config.limits.clone(), - ), + Session::outbound(nid, addr.clone(), persistent, self.rng.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; + fn fetching(&self) -> &FetcherState; /// Get outbox. fn outbox(&self) -> &Outbox; /// Get rate limiter. @@ -2752,8 +2676,8 @@ where &self.sessions } - fn fetching(&self) -> &HashMap { - &self.fetching + fn fetching(&self) -> &FetcherState { + self.fetcher.state() } fn outbox(&self) -> &Outbox { diff --git a/crates/radicle-protocol/src/service/io.rs b/crates/radicle-protocol/src/service/io.rs index 1862c11d..92b7b0e7 100644 --- a/crates/radicle-protocol/src/service/io.rs +++ b/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 { diff --git a/crates/radicle-protocol/src/service/session.rs b/crates/radicle-protocol/src/service/session.rs index 66557656..121805ca 100644 --- a/crates/radicle-protocol/src/service/session.rs +++ b/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, /// Last time a message was received from the peer. pub last_active: LocalTime, - /// Fetch queue. - pub queue: VecDeque, /// 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( - id: NodeId, - addr: Address, - persistent: bool, - rng: Rng, - time: LocalTime, - limits: Limits, - ) -> Self { + pub fn inbound(id: NodeId, addr: Address, persistent: bool, rng: Rng, time: LocalTime) -> 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 { - 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, }; diff --git a/crates/radicle-schemars/src/main.rs b/crates/radicle-schemars/src/main.rs index cb0e4c87..511e3c52 100644 --- a/crates/radicle-schemars/src/main.rs +++ b/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), ListenAddrs(ListenAddrs), ConnectResult(radicle::node::ConnectResult), Success(radicle::node::Success), diff --git a/crates/radicle/CHANGELOG.md b/crates/radicle/CHANGELOG.md index cf7590e6..856b1c9e 100644 --- a/crates/radicle/CHANGELOG.md +++ b/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 diff --git a/crates/radicle/src/node.rs b/crates/radicle/src/node.rs index 8f761fb5..a4291f6d 100644 --- a/crates/radicle/src/node.rs +++ b/crates/radicle/src/node.rs @@ -107,8 +107,6 @@ pub enum State { /// Ping state. #[serde(skip)] ping: PingState, - /// Ongoing fetches. - fetching: HashSet, /// Measured latencies for this peer. #[serde(skip)] latencies: VecDeque, @@ -696,7 +694,7 @@ impl From> 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, })) diff --git a/crates/radicle/src/node/command.rs b/crates/radicle/src/node/command.rs index 7a32c75c..1999626b 100644 --- a/crates/radicle/src/node/command.rs +++ b/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::>( - 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(_)) );