From 865abd35e225116f4bec440ba9dfd266d67a9bce Mon Sep 17 00:00:00 2001 From: Lorenz Leutgeb Date: Fri, 13 Feb 2026 12:41:19 +0100 Subject: [PATCH] protocol/fetcher: Intro `RefsToFetch` Using empty `Vec`s to mean "all refs" can be confusing. Avoid the chance of confusion by introduction of a new type that clearly describes which refs should be fetched. --- crates/radicle-node/src/runtime/handle.rs | 4 +- crates/radicle-protocol/src/fetcher.rs | 55 +++++++++++ .../radicle-protocol/src/fetcher/service.rs | 40 ++++---- crates/radicle-protocol/src/fetcher/state.rs | 60 +++++------- .../src/fetcher/state/command.rs | 9 +- .../src/fetcher/state/event.rs | 9 +- .../src/fetcher/test/queue.rs | 2 +- .../src/fetcher/test/queue/helpers.rs | 4 +- .../fetcher/test/queue/properties/merge.rs | 32 ++++--- .../src/fetcher/test/queue/unit.rs | 19 ++-- .../src/fetcher/test/state/command/cancel.rs | 16 ++-- .../src/fetcher/test/state/command/fetch.rs | 93 ++++++++++--------- .../src/fetcher/test/state/command/fetched.rs | 22 ++--- .../src/fetcher/test/state/concurrent.rs | 10 +- .../src/fetcher/test/state/config.rs | 8 +- .../src/fetcher/test/state/dequeue.rs | 16 ++-- .../src/fetcher/test/state/helpers.rs | 9 +- .../src/fetcher/test/state/invariant.rs | 10 +- .../src/fetcher/test/state/multinode.rs | 10 +- crates/radicle-protocol/src/service.rs | 30 +++--- 20 files changed, 263 insertions(+), 195 deletions(-) diff --git a/crates/radicle-node/src/runtime/handle.rs b/crates/radicle-node/src/runtime/handle.rs index ff5f6b1b..95166895 100644 --- a/crates/radicle-node/src/runtime/handle.rs +++ b/crates/radicle-node/src/runtime/handle.rs @@ -359,7 +359,7 @@ impl radicle::node::Handle for Handle { json!({ "rid": rid, "from": active.from(), - "refsAt": active.refs_at(), + "refsAt": active.refs(), }) }).collect::>(), "queue": fetcher_state.queued_fetches().iter().map(|(node, queue)| { @@ -368,7 +368,7 @@ impl radicle::node::Handle for Handle { "queue": queue.iter().map(|fetch| { json!({ "rid": fetch.rid, - "refsAt": fetch.refs_at, + "refsAt": fetch.refs, }) }).collect::>() }) diff --git a/crates/radicle-protocol/src/fetcher.rs b/crates/radicle-protocol/src/fetcher.rs index 00cbe20b..cf22498f 100644 --- a/crates/radicle-protocol/src/fetcher.rs +++ b/crates/radicle-protocol/src/fetcher.rs @@ -1,3 +1,7 @@ +use nonempty::NonEmpty; +use radicle::storage::refs::RefsAt; +use serde::{Deserialize, Serialize}; + pub mod service; pub use service::FetcherService; @@ -10,3 +14,54 @@ mod test; // TODO(finto): `Service::fetch_refs_at` and the use of `refs_status_of` is a // layer above the `Fetcher` where it would perform I/O, mocked out by a trait, // to check if there are wants and add a fetch to the Fetcher. + +/// Represents references to fetch, in the context of a repository. +#[derive(Clone, Debug, PartialEq, Eq, Hash, Serialize, Deserialize)] +pub enum RefsToFetch { + /// Indicates that all references should be fetched. + All, + /// Contains a non-empty collection of specific references to fetch. + Refs(NonEmpty), +} + +impl RefsToFetch { + /// Merges another `RefsToFetch` into this one, resulting in a new + /// `RefsToFetch` that represents the combined set of references to fetch. + /// If either `RefsToFetch` is `All`, the result will be `All`. If both are + /// `Refs`, their contents will be combined into a single `Refs` variant. + pub(super) fn merge(self, other: RefsToFetch) -> Self { + match (self, other) { + (RefsToFetch::All, _) | (_, RefsToFetch::All) => RefsToFetch::All, + (RefsToFetch::Refs(mut ours), RefsToFetch::Refs(theirs)) => { + ours.extend(theirs); + RefsToFetch::Refs(ours) + } + } + } + + #[cfg(test)] + pub fn len(&self) -> Option { + match self { + RefsToFetch::All => None, + RefsToFetch::Refs(refs) => std::num::NonZeroUsize::new(refs.len()), + } + } +} + +impl From for Vec { + fn from(val: RefsToFetch) -> Self { + match val { + RefsToFetch::All => Vec::new(), + RefsToFetch::Refs(refs) => refs.into(), + } + } +} + +impl From> for RefsToFetch { + fn from(refs_at: Vec) -> Self { + match NonEmpty::from_vec(refs_at) { + Some(refs) => RefsToFetch::Refs(refs), + None => RefsToFetch::All, + } + } +} diff --git a/crates/radicle-protocol/src/fetcher/service.rs b/crates/radicle-protocol/src/fetcher/service.rs index 734a5e1d..51a2549a 100644 --- a/crates/radicle-protocol/src/fetcher/service.rs +++ b/crates/radicle-protocol/src/fetcher/service.rs @@ -1,9 +1,14 @@ use std::collections::HashMap; -use radicle::storage::refs::RefsAt; use radicle_core::{NodeId, RepoId}; -use crate::fetcher::state::{command, event, Config, FetcherState, QueuedFetch}; +use crate::fetcher::{ + state::{ + command::{self}, + event, Config, FetcherState, QueuedFetch, + }, + RefsToFetch, +}; /// Service layer that wraps [`FetcherState`] and manages subscriber coalescing. /// @@ -38,12 +43,12 @@ impl FetcherService { struct FetchKey { rid: RepoId, node: NodeId, - refs_at: Vec, + refs: RefsToFetch, } impl FetchKey { - fn new(rid: RepoId, node: NodeId, refs_at: Vec) -> Self { - Self { rid, node, refs_at } + fn new(rid: RepoId, node: NodeId, refs: RefsToFetch) -> Self { + Self { rid, node, refs } } } @@ -88,7 +93,7 @@ impl FetcherService { /// /// See [`FetcherState::fetch`]. pub fn fetch(&mut self, cmd: command::Fetch, subscriber: Option) -> FetchInitiated { - let key = FetchKey::new(cmd.rid, cmd.from, cmd.refs_at.clone()); + let key = FetchKey::new(cmd.rid, cmd.from, cmd.refs.clone()); let event = self.state.fetch(cmd); let rejected = match &event { @@ -115,8 +120,8 @@ impl FetcherService { event: e, subscribers: vec![], }, - ref e @ event::Fetched::Completed { ref refs_at, .. } => { - let key = FetchKey::new(cmd.rid, cmd.from, refs_at.clone()); + ref e @ event::Fetched::Completed { ref refs, .. } => { + let key = FetchKey::new(cmd.rid, cmd.from, refs.clone()); let subscribers = self.subscribers.remove(&key).unwrap_or_default(); FetchCompleted { event: e.clone(), @@ -157,6 +162,7 @@ impl FetcherService { #[cfg(test)] mod tests { + use radicle::storage::refs::RefsAt; use radicle::test::arbitrary; use std::num::NonZeroUsize; use std::time::Duration; @@ -182,7 +188,7 @@ mod tests { command::Fetch { from: node, rid: repo, - refs_at: refs_specific.clone(), + refs: refs_specific.clone().into(), timeout, }, Some(1), @@ -195,7 +201,7 @@ mod tests { command::Fetch { from: node, rid: repo, - refs_at: refs_all.clone(), + refs: refs_all.clone().into(), timeout, }, Some(2), @@ -213,8 +219,8 @@ mod tests { }); match completed.event { - event::Fetched::Completed { ref refs_at, .. } => { - assert_eq!(refs_at, &refs_specific); + event::Fetched::Completed { ref refs, .. } => { + assert_eq!(refs, &refs_specific.into()); } _ => panic!("Expected Completed event"), } @@ -223,11 +229,13 @@ mod tests { assert_eq!(completed.subscribers, vec![1]); // subscriber 2 should still be waiting - assert!(service - .subscribers - .contains_key(&FetchKey::new(repo, node, refs_all.clone()))); + assert!(service.subscribers.contains_key(&FetchKey::new( + repo, + node, + refs_all.clone().into() + ))); - let remaining = &service.subscribers[&FetchKey::new(repo, node, refs_all)]; + let remaining = &service.subscribers[&FetchKey::new(repo, node, refs_all.into())]; assert_eq!(remaining.len(), 1); assert_eq!(remaining[0], 2); } diff --git a/crates/radicle-protocol/src/fetcher/state.rs b/crates/radicle-protocol/src/fetcher/state.rs index 69165375..cde707a6 100644 --- a/crates/radicle-protocol/src/fetcher/state.rs +++ b/crates/radicle-protocol/src/fetcher/state.rs @@ -15,9 +15,10 @@ use std::collections::{BTreeMap, VecDeque}; use std::num::NonZeroUsize; use std::time; -use radicle::storage::refs::RefsAt; use radicle_core::{NodeId, RepoId}; +use crate::fetcher::RefsToFetch; + /// Default for the maximum items per fetch queue. pub const MAX_FETCH_QUEUE_SIZE: usize = 128; /// Default for maximum concurrency per node. @@ -96,32 +97,32 @@ impl FetcherState { command::Fetch { from, rid, - refs_at, + refs, timeout, }: command::Fetch, ) -> event::Fetch { if let Some(active) = self.active.get(&rid) { - if active.refs_at == refs_at && active.from == from { + if active.refs == refs && active.from == from { return event::Fetch::AlreadyFetching { rid, from }; } else { - return self.enqueue(rid, from, refs_at, timeout); + return self.enqueue(rid, from, refs, timeout); } } if self.is_at_node_capacity(&from) { - self.enqueue(rid, from, refs_at, timeout) + self.enqueue(rid, from, refs, timeout) } else { self.active.insert( rid, ActiveFetch { from, - refs_at: refs_at.clone(), + refs: refs.clone(), }, ); event::Fetch::Started { rid, from, - refs_at, + refs, timeout, } } @@ -136,7 +137,7 @@ impl FetcherState { pub fn fetched(&mut self, command::Fetched { from, rid }: command::Fetched) -> event::Fetched { match self.active.remove(&rid) { None => event::Fetched::NotFound { from, rid }, - Some(ActiveFetch { from, refs_at }) => event::Fetched::Completed { from, rid, refs_at }, + Some(ActiveFetch { from, refs }) => event::Fetched::Completed { from, rid, refs }, } } @@ -182,29 +183,23 @@ impl FetcherState { &mut self, rid: RepoId, from: NodeId, - refs_at: Vec, + refs: RefsToFetch, timeout: time::Duration, ) -> event::Fetch { let queue = self .queues .entry(from) .or_insert(Queue::new(self.config.maximum_queue_size)); - match queue.enqueue(QueuedFetch { - rid, - refs_at, - timeout, - }) { - Enqueue::CapacityReached(QueuedFetch { - rid, - refs_at, - timeout, - }) => event::Fetch::QueueAtCapacity { - rid, - from, - refs_at, - timeout, - capacity: queue.len(), - }, + match queue.enqueue(QueuedFetch { rid, refs, timeout }) { + Enqueue::CapacityReached(QueuedFetch { rid, refs, timeout }) => { + event::Fetch::QueueAtCapacity { + rid, + from, + refs, + timeout, + capacity: queue.len(), + } + } Enqueue::Queued => event::Fetch::Queued { rid, from }, Enqueue::Merged => event::Fetch::Queued { rid, from }, } @@ -279,7 +274,7 @@ impl Default for Config { #[derive(Clone, Debug, PartialEq, Eq)] pub struct ActiveFetch { pub from: NodeId, - pub refs_at: Vec, + pub refs: RefsToFetch, } impl ActiveFetch { @@ -289,8 +284,8 @@ impl ActiveFetch { } /// The set of references that fetch is being performed for. - pub fn refs_at(&self) -> &[RefsAt] { - &self.refs_at + pub fn refs(&self) -> &RefsToFetch { + &self.refs } } @@ -300,7 +295,7 @@ pub struct QueuedFetch { /// The repository that will be fetched. pub rid: RepoId, /// The references that the fetch is being performed for. - pub refs_at: Vec, + pub refs: RefsToFetch, /// The timeout given for the fetch request. pub timeout: time::Duration, } @@ -379,12 +374,7 @@ impl Queue { /// the queue has not reached capacity and if the item is unique. pub(super) fn enqueue(&mut self, fetch: QueuedFetch) -> Enqueue { if let Some(existing) = self.queue.iter_mut().find(|qf| qf.rid == fetch.rid) { - if existing.refs_at.is_empty() || fetch.refs_at.is_empty() { - // We fetch everything - existing.refs_at = vec![] - } else { - existing.refs_at.extend(fetch.refs_at); - } + existing.refs = existing.refs.clone().merge(fetch.refs); // Take the longer timeout (more generous) existing.timeout = existing.timeout.max(fetch.timeout); return Enqueue::Merged; diff --git a/crates/radicle-protocol/src/fetcher/state/command.rs b/crates/radicle-protocol/src/fetcher/state/command.rs index ef289aca..ae4f04ed 100644 --- a/crates/radicle-protocol/src/fetcher/state/command.rs +++ b/crates/radicle-protocol/src/fetcher/state/command.rs @@ -1,8 +1,9 @@ use std::time; -use radicle::storage::refs::RefsAt; use radicle_core::{NodeId, RepoId}; +use crate::fetcher::RefsToFetch; + /// Commands for transitioning the [`FetcherState`]. /// /// [`FetcherState`]: super::FetcherState @@ -32,11 +33,11 @@ impl From for Command { } impl Command { - pub fn fetch(from: NodeId, rid: RepoId, refs_at: Vec, timeout: time::Duration) -> Self { + pub fn fetch(from: NodeId, rid: RepoId, refs: RefsToFetch, timeout: time::Duration) -> Self { Self::from(Fetch { from, rid, - refs_at, + refs, timeout, }) } @@ -58,7 +59,7 @@ pub struct Fetch { /// The repository to fetch. pub rid: RepoId, /// The references to fetch. - pub refs_at: Vec, + pub refs: RefsToFetch, /// The timeout for the fetch process. pub timeout: time::Duration, } diff --git a/crates/radicle-protocol/src/fetcher/state/event.rs b/crates/radicle-protocol/src/fetcher/state/event.rs index 0a2c6084..5167cda6 100644 --- a/crates/radicle-protocol/src/fetcher/state/event.rs +++ b/crates/radicle-protocol/src/fetcher/state/event.rs @@ -1,9 +1,10 @@ use std::collections::{BTreeMap, VecDeque}; use std::time; -use radicle::storage::refs::RefsAt; use radicle_core::{NodeId, RepoId}; +use crate::fetcher::RefsToFetch; + use super::{ActiveFetch, QueuedFetch}; /// Event returned from [`FetchState::handle`]. @@ -44,7 +45,7 @@ pub enum Fetch { /// The node to fetch from. from: NodeId, /// The references to be fetched. - refs_at: Vec, + refs: RefsToFetch, /// The timeout for the fetch process. timeout: time::Duration, }, @@ -63,7 +64,7 @@ pub enum Fetch { /// The node who's queue is at capacity. from: NodeId, /// The references expected to be fetched. - refs_at: Vec, + refs: RefsToFetch, /// The timeout for the fetch process. timeout: time::Duration, /// The capacity of the queue. @@ -91,7 +92,7 @@ pub enum Fetched { /// The repository that was fetched. rid: RepoId, /// The references that were fetched. - refs_at: Vec, + refs: RefsToFetch, }, } diff --git a/crates/radicle-protocol/src/fetcher/test/queue.rs b/crates/radicle-protocol/src/fetcher/test/queue.rs index 4bc1aaf2..a906b927 100644 --- a/crates/radicle-protocol/src/fetcher/test/queue.rs +++ b/crates/radicle-protocol/src/fetcher/test/queue.rs @@ -20,7 +20,7 @@ impl Arbitrary for QueuedFetch { QueuedFetch { rid: RepoId::arbitrary(g), - refs_at, + refs: refs_at.into(), timeout: Duration::from_secs(u64::arbitrary(g) % 3600), } } diff --git a/crates/radicle-protocol/src/fetcher/test/queue/helpers.rs b/crates/radicle-protocol/src/fetcher/test/queue/helpers.rs index e4e0d0a6..243518ab 100644 --- a/crates/radicle-protocol/src/fetcher/test/queue/helpers.rs +++ b/crates/radicle-protocol/src/fetcher/test/queue/helpers.rs @@ -2,7 +2,7 @@ use std::{num::NonZeroUsize, time::Duration}; use radicle::test::arbitrary; -use crate::fetcher::{MaxQueueSize, Queue, QueuedFetch}; +use crate::fetcher::{MaxQueueSize, Queue, QueuedFetch, RefsToFetch}; pub fn create_queue(capacity: usize) -> Queue { Queue::new(MaxQueueSize::new( @@ -13,7 +13,7 @@ pub fn create_queue(capacity: usize) -> Queue { pub fn create_fetch() -> QueuedFetch { QueuedFetch { rid: arbitrary::gen(1), - refs_at: vec![], + refs: RefsToFetch::All, timeout: Duration::from_secs(30), } } diff --git a/crates/radicle-protocol/src/fetcher/test/queue/properties/merge.rs b/crates/radicle-protocol/src/fetcher/test/queue/properties/merge.rs index c75d03a6..c22314b3 100644 --- a/crates/radicle-protocol/src/fetcher/test/queue/properties/merge.rs +++ b/crates/radicle-protocol/src/fetcher/test/queue/properties/merge.rs @@ -1,3 +1,4 @@ +use std::num::NonZeroUsize; use std::time::Duration; use qcheck_macros::quickcheck; @@ -7,6 +8,7 @@ use radicle_core::RepoId; use crate::fetcher::state::Enqueue; use crate::fetcher::test::queue::helpers::*; +use crate::fetcher::RefsToFetch; use crate::fetcher::{MaxQueueSize, Queue, QueuedFetch}; #[quickcheck] @@ -30,7 +32,7 @@ fn same_rid_merges_anywhere_in_queue(max_size: MaxQueueSize, merge_index: usize) let target_index = merge_index % items.len(); let same_rid_item = QueuedFetch { rid: items[target_index].rid, - refs_at: vec![arbitrary::gen(1)], + refs: vec![arbitrary::gen(1)].into(), timeout: Duration::from_secs(60), }; @@ -50,13 +52,13 @@ fn combines_refs(base_refs_count: u8, merge_refs_count: u8) -> bool { let base_item = QueuedFetch { rid, - refs_at: base_refs.clone(), + refs: base_refs.clone().into(), timeout: Duration::from_secs(30), }; let merge_item = QueuedFetch { rid, - refs_at: merge_refs.clone(), + refs: merge_refs.clone().into(), timeout: Duration::from_secs(30), }; @@ -71,10 +73,10 @@ fn combines_refs(base_refs_count: u8, merge_refs_count: u8) -> bool { // If either was empty, result should be empty (fetch everything) if base_refs.is_empty() || merge_refs.is_empty() { - dequeued.refs_at.is_empty() + dequeued.refs == RefsToFetch::All } else { // Otherwise refs should be combined - dequeued.refs_at.len() == base_refs_count + merge_refs_count + dequeued.refs.len() == Some(NonZeroUsize::new(base_refs_count + merge_refs_count).unwrap()) } } @@ -86,14 +88,14 @@ fn empty_refs_fetches_all() -> bool { // First enqueue with specific refs let item_with_refs = QueuedFetch { rid, - refs_at: vec![arbitrary::gen(1), arbitrary::gen(1)], + refs: vec![arbitrary::gen(1), arbitrary::gen(1)].into(), timeout: Duration::from_secs(30), }; // Second enqueue with empty refs (fetch everything) let item_empty_refs = QueuedFetch { rid, - refs_at: vec![], + refs: RefsToFetch::All, timeout: Duration::from_secs(30), }; @@ -101,7 +103,7 @@ fn empty_refs_fetches_all() -> bool { let _ = queue.enqueue(item_empty_refs); let dequeued = queue.dequeue().unwrap(); - dequeued.refs_at.is_empty() // Should fetch everything + dequeued.refs == RefsToFetch::All // Should fetch everything } #[quickcheck] @@ -114,13 +116,13 @@ fn longer_timeout_preserved(short_secs: u16, long_secs: u16) -> bool { let item_short = QueuedFetch { rid, - refs_at: vec![], + refs: RefsToFetch::All, timeout: short, }; let item_long = QueuedFetch { rid, - refs_at: vec![], + refs: RefsToFetch::All, timeout: long, }; @@ -144,13 +146,13 @@ fn does_not_increase_queue_length() -> bool { let item1 = QueuedFetch { rid, - refs_at: vec![arbitrary::gen(1)], + refs: vec![arbitrary::gen(1)].into(), timeout: Duration::from_secs(30), }; let item2 = QueuedFetch { rid, - refs_at: vec![arbitrary::gen(1)], + refs: vec![arbitrary::gen(1)].into(), timeout: Duration::from_secs(60), }; @@ -185,19 +187,19 @@ fn succeed_when_at_capacity() -> bool { let item1 = QueuedFetch { rid, - refs_at: vec![], + refs: RefsToFetch::All, timeout: Duration::from_secs(30), }; let item2 = QueuedFetch { rid: arbitrary::gen(1), // Different rid - refs_at: vec![], + refs: RefsToFetch::All, timeout: Duration::from_secs(30), }; let merge_item = QueuedFetch { rid, // Same as item1 - refs_at: vec![arbitrary::gen(1)], + refs: vec![arbitrary::gen(1)].into(), timeout: Duration::from_secs(60), }; diff --git a/crates/radicle-protocol/src/fetcher/test/queue/unit.rs b/crates/radicle-protocol/src/fetcher/test/queue/unit.rs index 42533a0b..38efd480 100644 --- a/crates/radicle-protocol/src/fetcher/test/queue/unit.rs +++ b/crates/radicle-protocol/src/fetcher/test/queue/unit.rs @@ -6,13 +6,14 @@ use radicle_core::RepoId; use crate::fetcher::state::Enqueue; use crate::fetcher::test::queue::helpers::*; use crate::fetcher::QueuedFetch; +use crate::fetcher::RefsToFetch; #[test] fn zero_timeout_accepted() { let mut queue = create_queue(10); let item = QueuedFetch { rid: arbitrary::gen(1), - refs_at: vec![], + refs: RefsToFetch::All, timeout: Duration::ZERO, }; assert_eq!(queue.enqueue(item), Enqueue::Queued); @@ -23,25 +24,25 @@ fn max_timeout_accepted() { let mut queue = create_queue(10); let item = QueuedFetch { rid: arbitrary::gen(1), - refs_at: vec![], + refs: RefsToFetch::All, timeout: Duration::MAX, }; assert_eq!(queue.enqueue(item), Enqueue::Queued); } #[test] -fn empty_refs_at_items_can_be_equal() { +fn empty_refs_items_can_be_equal() { let rid: RepoId = arbitrary::gen(1); let timeout = Duration::from_secs(30); let item1 = QueuedFetch { rid, - refs_at: vec![], + refs: RefsToFetch::All, timeout, }; let item2 = QueuedFetch { rid, - refs_at: vec![], + refs: RefsToFetch::All, timeout, }; @@ -59,24 +60,24 @@ fn merge_preserves_position_in_queue() { // Enqueue three items let _ = queue.enqueue(QueuedFetch { rid: rid_first, - refs_at: vec![], + refs: RefsToFetch::All, timeout: Duration::from_secs(30), }); let _ = queue.enqueue(QueuedFetch { rid: rid_second, - refs_at: vec![], + refs: RefsToFetch::All, timeout: Duration::from_secs(30), }); let _ = queue.enqueue(QueuedFetch { rid: rid_third, - refs_at: vec![], + refs: RefsToFetch::All, timeout: Duration::from_secs(30), }); // Merge into the second item let result = queue.enqueue(QueuedFetch { rid: rid_second, - refs_at: vec![arbitrary::gen(1)], + refs: vec![arbitrary::gen(1)].into(), timeout: Duration::from_secs(60), }); assert_eq!(result, Enqueue::Merged); diff --git a/crates/radicle-protocol/src/fetcher/test/state/command/cancel.rs b/crates/radicle-protocol/src/fetcher/test/state/command/cancel.rs index 7fafba6b..98038c5a 100644 --- a/crates/radicle-protocol/src/fetcher/test/state/command/cancel.rs +++ b/crates/radicle-protocol/src/fetcher/test/state/command/cancel.rs @@ -12,13 +12,13 @@ fn single_ongoing() { let mut state = FetcherState::new(helpers::config(1, 10)); let node_a: NodeId = arbitrary::gen(1); let repo_1: RepoId = arbitrary::gen(1); - let refs_at_1 = helpers::gen_refs_at(1); + let refs_1 = helpers::gen_refs(1); let timeout = Duration::from_secs(30); state.fetch(command::Fetch { from: node_a, rid: repo_1, - refs_at: refs_at_1.clone(), + refs: refs_1.clone(), timeout, }); @@ -36,7 +36,7 @@ fn single_ongoing() { ongoing.get(&repo_1), Some(&ActiveFetch { from: node_a, - refs_at: refs_at_1, + refs: refs_1, }) ); assert!(queued.is_empty()); @@ -58,19 +58,19 @@ fn ongoing_and_queued() { state.fetch(command::Fetch { from: node_a, rid: repo_1, - refs_at: helpers::gen_refs_at(1), + refs: helpers::gen_refs(1), timeout, }); state.fetch(command::Fetch { from: node_a, rid: repo_2, - refs_at: helpers::gen_refs_at(1), + refs: helpers::gen_refs(1), timeout, }); state.fetch(command::Fetch { from: node_a, rid: repo_3, - refs_at: helpers::gen_refs_at(1), + refs: helpers::gen_refs(1), timeout, }); @@ -112,13 +112,13 @@ fn cancellation_is_isolated() { state.fetch(command::Fetch { from: node_a, rid: repo_1, - refs_at: helpers::gen_refs_at(1), + refs: helpers::gen_refs(1), timeout, }); state.fetch(command::Fetch { from: node_b, rid: repo_2, - refs_at: helpers::gen_refs_at(1), + refs: helpers::gen_refs(1), timeout, }); diff --git a/crates/radicle-protocol/src/fetcher/test/state/command/fetch.rs b/crates/radicle-protocol/src/fetcher/test/state/command/fetch.rs index 303a3ec7..16849057 100644 --- a/crates/radicle-protocol/src/fetcher/test/state/command/fetch.rs +++ b/crates/radicle-protocol/src/fetcher/test/state/command/fetch.rs @@ -5,6 +5,7 @@ use radicle_core::{NodeId, RepoId}; use crate::fetcher::state::{command, event}; use crate::fetcher::test::state::helpers; +use crate::fetcher::RefsToFetch; use crate::fetcher::{ActiveFetch, FetcherState}; #[test] @@ -12,13 +13,13 @@ fn fetch_start_first_fetch_for_node() { let mut state = FetcherState::new(helpers::config(1, 10)); let node_a: NodeId = arbitrary::gen(1); let repo_1: RepoId = arbitrary::gen(1); - let refs_at_1 = helpers::gen_refs_at(2); + let refs_1 = helpers::gen_refs(2); let timeout = Duration::from_secs(30); let event = state.fetch(command::Fetch { from: node_a, rid: repo_1, - refs_at: refs_at_1.clone(), + refs: refs_1.clone(), timeout, }); @@ -27,7 +28,7 @@ fn fetch_start_first_fetch_for_node() { event::Fetch::Started { rid: repo_1, from: node_a, - refs_at: refs_at_1.clone(), + refs: refs_1.clone(), timeout, } ); @@ -35,7 +36,7 @@ fn fetch_start_first_fetch_for_node() { state.get_active_fetch(&repo_1), Some(&ActiveFetch { from: node_a, - refs_at: refs_at_1, + refs: refs_1, }) ); } @@ -51,7 +52,7 @@ fn fetch_different_repo_same_node_within_capacity() { let event1 = state.fetch(command::Fetch { from: node_a, rid: repo_1, - refs_at: helpers::gen_refs_at(1), + refs: helpers::gen_refs(1), timeout, }); assert!(matches!(event1, event::Fetch::Started { .. })); @@ -59,7 +60,7 @@ fn fetch_different_repo_same_node_within_capacity() { let event2 = state.fetch(command::Fetch { from: node_a, rid: repo_2, - refs_at: helpers::gen_refs_at(1), + refs: helpers::gen_refs(1), timeout, }); @@ -74,13 +75,13 @@ fn fetch_same_repo_different_nodes_queues_second() { let node_a: NodeId = arbitrary::gen(1); let node_b: NodeId = arbitrary::gen(1); let repo_1: RepoId = arbitrary::gen(1); - let refs_at_1 = helpers::gen_refs_at(1); + let refs_1 = helpers::gen_refs(1); let timeout = Duration::from_secs(30); let event1 = state.fetch(command::Fetch { from: node_a, rid: repo_1, - refs_at: refs_at_1.clone(), + refs: refs_1.clone(), timeout, }); assert!(matches!(event1, event::Fetch::Started { .. })); @@ -89,7 +90,7 @@ fn fetch_same_repo_different_nodes_queues_second() { let event2 = state.fetch(command::Fetch { from: node_b, rid: repo_1, - refs_at: refs_at_1.clone(), + refs: refs_1.clone(), timeout, }); @@ -107,20 +108,20 @@ fn fetch_duplicate_returns_already_fetching() { let mut state = FetcherState::new(helpers::config(1, 10)); let node_a: NodeId = arbitrary::gen(1); let repo_1: RepoId = arbitrary::gen(1); - let refs_at_1 = helpers::gen_refs_at(2); + let refs_1 = helpers::gen_refs(2); let timeout = Duration::from_secs(30); state.fetch(command::Fetch { from: node_a, rid: repo_1, - refs_at: refs_at_1.clone(), + refs: refs_1.clone(), timeout, }); let event = state.fetch(command::Fetch { from: node_a, rid: repo_1, - refs_at: refs_at_1.clone(), + refs: refs_1.clone(), timeout, }); @@ -138,21 +139,21 @@ fn fetch_same_repo_different_refs_enqueues() { let mut state = FetcherState::new(helpers::config(1, 10)); let node_a: NodeId = arbitrary::gen(1); let repo_1: RepoId = arbitrary::gen(1); - let refs_at_1 = helpers::gen_refs_at(1); - let refs_at_2 = helpers::gen_refs_at(2); + let refs_1 = helpers::gen_refs(1); + let refs_2 = helpers::gen_refs(2); let timeout = Duration::from_secs(30); state.fetch(command::Fetch { from: node_a, rid: repo_1, - refs_at: refs_at_1.clone(), + refs: refs_1.clone(), timeout, }); let event = state.fetch(command::Fetch { from: node_a, rid: repo_1, - refs_at: refs_at_2.clone(), + refs: refs_2.clone(), timeout, }); @@ -176,14 +177,14 @@ fn fetch_at_capacity_enqueues() { state.fetch(command::Fetch { from: node_a, rid: repo_1, - refs_at: helpers::gen_refs_at(1), + refs: helpers::gen_refs(1), timeout, }); let event = state.fetch(command::Fetch { from: node_a, rid: repo_2, - refs_at: helpers::gen_refs_at(1), + refs: helpers::gen_refs(1), timeout, }); @@ -212,7 +213,7 @@ fn fetch_queue_rejected_capacity_reached() { state.fetch(command::Fetch { from: node_a, rid: repo_1, - refs_at: helpers::gen_refs_at(1), + refs: helpers::gen_refs(1), timeout, }); @@ -220,22 +221,22 @@ fn fetch_queue_rejected_capacity_reached() { state.fetch(command::Fetch { from: node_a, rid: repo_2, - refs_at: helpers::gen_refs_at(1), + refs: helpers::gen_refs(1), timeout, }); state.fetch(command::Fetch { from: node_a, rid: repo_3, - refs_at: helpers::gen_refs_at(1), + refs: helpers::gen_refs(1), timeout, }); // Exceed queue capacity - let refs_at_4 = helpers::gen_refs_at(1); + let refs_4 = helpers::gen_refs(1); let event = state.fetch(command::Fetch { from: node_a, rid: repo_4, - refs_at: refs_at_4.clone(), + refs: refs_4.clone(), timeout, }); @@ -244,7 +245,7 @@ fn fetch_queue_rejected_capacity_reached() { event::Fetch::QueueAtCapacity { rid: repo_4, from: node_a, - refs_at: refs_at_4, + refs: refs_4, timeout, capacity: 2, } @@ -257,21 +258,21 @@ fn fetch_queue_merges_already_queued() { let node_a: NodeId = arbitrary::gen(1); let repo_1: RepoId = arbitrary::gen(1); let repo_2: RepoId = arbitrary::gen(1); - let refs_at_2a = helpers::gen_refs_at(1); - let refs_at_2b = helpers::gen_refs_at(1); + let refs_2a = helpers::gen_refs(1); + let refs_2b = helpers::gen_refs(1); let timeout = Duration::from_secs(30); state.fetch(command::Fetch { from: node_a, rid: repo_1, - refs_at: helpers::gen_refs_at(1), + refs: helpers::gen_refs(1), timeout, }); state.fetch(command::Fetch { from: node_a, rid: repo_2, - refs_at: refs_at_2a.clone(), + refs: refs_2a.clone(), timeout, }); @@ -279,7 +280,7 @@ fn fetch_queue_merges_already_queued() { let event = state.fetch(command::Fetch { from: node_a, rid: repo_2, - refs_at: refs_at_2b.clone(), + refs: refs_2b.clone(), timeout, }); @@ -299,8 +300,16 @@ fn fetch_queue_merges_already_queued() { }); let queued = state.dequeue(&node_a).unwrap(); assert_eq!(queued.rid, repo_2); - // refs_at should contain both sets of refs - assert_eq!(queued.refs_at.len(), refs_at_2a.len() + refs_at_2b.len()); + // `queued.refs` should be the union of both sets of refs. + assert_eq!( + queued.refs.len(), + Some( + refs_2a + .len() + .unwrap() + .saturating_add(refs_2b.len().unwrap().into()) + ) + ); } #[test] @@ -309,13 +318,13 @@ fn fetch_queue_merge_empty_refs_fetches_all() { let node_a: NodeId = arbitrary::gen(1); let repo_1: RepoId = arbitrary::gen(1); let repo_2: RepoId = arbitrary::gen(1); - let refs_at_2 = helpers::gen_refs_at(2); + let refs_2 = helpers::gen_refs(2); let timeout = Duration::from_secs(30); state.fetch(command::Fetch { from: node_a, rid: repo_1, - refs_at: helpers::gen_refs_at(1), + refs: helpers::gen_refs(1), timeout, }); @@ -323,7 +332,7 @@ fn fetch_queue_merge_empty_refs_fetches_all() { state.fetch(command::Fetch { from: node_a, rid: repo_2, - refs_at: refs_at_2.clone(), + refs: refs_2.clone(), timeout, }); @@ -331,7 +340,7 @@ fn fetch_queue_merge_empty_refs_fetches_all() { state.fetch(command::Fetch { from: node_a, rid: repo_2, - refs_at: vec![], + refs: RefsToFetch::All, timeout, }); @@ -342,7 +351,7 @@ fn fetch_queue_merge_empty_refs_fetches_all() { }); let queued = state.dequeue(&node_a).unwrap(); assert_eq!(queued.rid, repo_2); - assert!(queued.refs_at.is_empty()); + assert_eq!(queued.refs, RefsToFetch::All); } #[test] @@ -357,7 +366,7 @@ fn fetch_queue_merge_takes_longer_timeout() { state.fetch(command::Fetch { from: node_a, rid: repo_1, - refs_at: helpers::gen_refs_at(1), + refs: helpers::gen_refs(1), timeout: short_timeout, }); @@ -365,7 +374,7 @@ fn fetch_queue_merge_takes_longer_timeout() { state.fetch(command::Fetch { from: node_a, rid: repo_2, - refs_at: helpers::gen_refs_at(1), + refs: helpers::gen_refs(1), timeout: short_timeout, }); @@ -373,7 +382,7 @@ fn fetch_queue_merge_takes_longer_timeout() { state.fetch(command::Fetch { from: node_a, rid: repo_2, - refs_at: helpers::gen_refs_at(1), + refs: helpers::gen_refs(1), timeout: long_timeout, }); @@ -391,13 +400,13 @@ fn fetch_after_previous_completed() { let mut state = FetcherState::new(helpers::config(1, 10)); let node_a: NodeId = arbitrary::gen(1); let repo_1: RepoId = arbitrary::gen(1); - let refs_at_1 = helpers::gen_refs_at(1); + let refs_1 = helpers::gen_refs(1); let timeout = Duration::from_secs(30); state.fetch(command::Fetch { from: node_a, rid: repo_1, - refs_at: refs_at_1.clone(), + refs: refs_1.clone(), timeout, }); state.fetched(command::Fetched { @@ -408,7 +417,7 @@ fn fetch_after_previous_completed() { let event = state.fetch(command::Fetch { from: node_a, rid: repo_1, - refs_at: refs_at_1.clone(), + refs: refs_1.clone(), timeout, }); diff --git a/crates/radicle-protocol/src/fetcher/test/state/command/fetched.rs b/crates/radicle-protocol/src/fetcher/test/state/command/fetched.rs index b286ea60..574ab874 100644 --- a/crates/radicle-protocol/src/fetcher/test/state/command/fetched.rs +++ b/crates/radicle-protocol/src/fetcher/test/state/command/fetched.rs @@ -12,13 +12,13 @@ fn complete_single_ongoing() { let mut state = FetcherState::new(helpers::config(1, 10)); let node_a: NodeId = arbitrary::gen(1); let repo_1: RepoId = arbitrary::gen(1); - let refs_at_1 = helpers::gen_refs_at(2); + let refs_1 = helpers::gen_refs(2); let timeout = Duration::from_secs(30); state.fetch(command::Fetch { from: node_a, rid: repo_1, - refs_at: refs_at_1.clone(), + refs: refs_1.clone(), timeout, }); @@ -32,7 +32,7 @@ fn complete_single_ongoing() { event::Fetched::Completed { from: node_a, rid: repo_1, - refs_at: refs_at_1, + refs: refs_1, } ); assert!(state.get_active_fetch(&repo_1).is_none()); @@ -45,13 +45,13 @@ fn complete_then_dequeue_fifo() { let repo_1: RepoId = arbitrary::gen(1); let repo_2: RepoId = arbitrary::gen(1); let repo_3: RepoId = arbitrary::gen(1); - let refs_at_2 = helpers::gen_refs_at(1); + let refs_2 = helpers::gen_refs(1); let timeout = Duration::from_secs(30); state.fetch(command::Fetch { from: node_a, rid: repo_1, - refs_at: helpers::gen_refs_at(1), + refs: helpers::gen_refs(1), timeout, }); @@ -59,13 +59,13 @@ fn complete_then_dequeue_fifo() { state.fetch(command::Fetch { from: node_a, rid: repo_2, - refs_at: refs_at_2.clone(), + refs: refs_2.clone(), timeout, }); state.fetch(command::Fetch { from: node_a, rid: repo_3, - refs_at: helpers::gen_refs_at(1), + refs: helpers::gen_refs(1), timeout, }); @@ -81,7 +81,7 @@ fn complete_then_dequeue_fifo() { assert!(queued.is_some()); let queued = queued.unwrap(); assert_eq!(queued.rid, repo_2); - assert_eq!(queued.refs_at, refs_at_2); + assert_eq!(queued.refs, refs_2); } #[test] @@ -96,19 +96,19 @@ fn complete_one_of_multiple() { state.fetch(command::Fetch { from: node_a, rid: repo_1, - refs_at: helpers::gen_refs_at(1), + refs: helpers::gen_refs(1), timeout, }); state.fetch(command::Fetch { from: node_a, rid: repo_2, - refs_at: helpers::gen_refs_at(1), + refs: helpers::gen_refs(1), timeout, }); state.fetch(command::Fetch { from: node_a, rid: repo_3, - refs_at: helpers::gen_refs_at(1), + refs: helpers::gen_refs(1), timeout, }); diff --git a/crates/radicle-protocol/src/fetcher/test/state/concurrent.rs b/crates/radicle-protocol/src/fetcher/test/state/concurrent.rs index d4b59bde..e5c86d01 100644 --- a/crates/radicle-protocol/src/fetcher/test/state/concurrent.rs +++ b/crates/radicle-protocol/src/fetcher/test/state/concurrent.rs @@ -21,7 +21,7 @@ fn interleaved_operations() { let e1 = state.fetch(command::Fetch { from: node_a, rid: repo_1, - refs_at: helpers::gen_refs_at(1), + refs: helpers::gen_refs(1), timeout, }); assert!(matches!(e1, event::Fetch::Started { .. })); @@ -30,7 +30,7 @@ fn interleaved_operations() { let e2 = state.fetch(command::Fetch { from: node_b, rid: repo_2, - refs_at: helpers::gen_refs_at(1), + refs: helpers::gen_refs(1), timeout, }); assert!(matches!(e2, event::Fetch::Started { .. })); @@ -46,7 +46,7 @@ fn interleaved_operations() { let e4 = state.fetch(command::Fetch { from: node_a, rid: repo_3, - refs_at: helpers::gen_refs_at(1), + refs: helpers::gen_refs(1), timeout, }); assert!(matches!(e4, event::Fetch::Started { .. })); @@ -75,13 +75,13 @@ fn fetched_then_cancel() { state.fetch(command::Fetch { from: node_a, rid: repo_1, - refs_at: helpers::gen_refs_at(1), + refs: helpers::gen_refs(1), timeout, }); state.fetch(command::Fetch { from: node_a, rid: repo_2, - refs_at: helpers::gen_refs_at(1), + refs: helpers::gen_refs(1), timeout, }); diff --git a/crates/radicle-protocol/src/fetcher/test/state/config.rs b/crates/radicle-protocol/src/fetcher/test/state/config.rs index ce148eb2..a68c434e 100644 --- a/crates/radicle-protocol/src/fetcher/test/state/config.rs +++ b/crates/radicle-protocol/src/fetcher/test/state/config.rs @@ -18,7 +18,7 @@ fn high_concurrency() { let event = state.fetch(command::Fetch { from: node_a, rid: repo, - refs_at: helpers::gen_refs_at(1), + refs: helpers::gen_refs(1), timeout, }); assert!( @@ -50,14 +50,14 @@ fn min_queue_size() { state.fetch(command::Fetch { from: node_a, rid: repo_1, - refs_at: helpers::gen_refs_at(1), + refs: helpers::gen_refs(1), timeout, }); let event1 = state.fetch(command::Fetch { from: node_a, rid: repo_2, - refs_at: helpers::gen_refs_at(1), + refs: helpers::gen_refs(1), timeout, }); assert!(matches!(event1, event::Fetch::Queued { .. })); @@ -65,7 +65,7 @@ fn min_queue_size() { let event2 = state.fetch(command::Fetch { from: node_a, rid: repo_3, - refs_at: helpers::gen_refs_at(1), + refs: helpers::gen_refs(1), timeout, }); assert!(matches!(event2, event::Fetch::QueueAtCapacity { .. })); diff --git a/crates/radicle-protocol/src/fetcher/test/state/dequeue.rs b/crates/radicle-protocol/src/fetcher/test/state/dequeue.rs index 56182712..550fece6 100644 --- a/crates/radicle-protocol/src/fetcher/test/state/dequeue.rs +++ b/crates/radicle-protocol/src/fetcher/test/state/dequeue.rs @@ -13,20 +13,20 @@ fn cannot_dequeue_while_node_at_capacity() { let node_a: NodeId = arbitrary::gen(1); let repo_1: RepoId = arbitrary::gen(1); let repo_2: RepoId = arbitrary::gen(1); - let refs_at_2 = helpers::gen_refs_at(3); + let refs_2 = helpers::gen_refs(3); let timeout_2 = Duration::from_secs(42); state.fetch(command::Fetch { from: node_a, rid: repo_1, - refs_at: helpers::gen_refs_at(1), + refs: helpers::gen_refs(1), timeout: Duration::from_secs(10), }); state.fetch(command::Fetch { from: node_a, rid: repo_2, - refs_at: refs_at_2.clone(), + refs: refs_2.clone(), timeout: timeout_2, }); @@ -41,7 +41,7 @@ fn cannot_dequeue_while_node_at_capacity() { let result = state.dequeue(&node_a); let queued = result.unwrap(); assert_eq!(queued.rid, repo_2); - assert_eq!(queued.refs_at, refs_at_2); + assert_eq!(queued.refs, refs_2); assert_eq!(queued.timeout, timeout_2); } @@ -58,7 +58,7 @@ fn maintains_fifo_order() { state.fetch(command::Fetch { from: node_a, rid: repo_1, - refs_at: helpers::gen_refs_at(1), + refs: helpers::gen_refs(1), timeout, }); @@ -66,19 +66,19 @@ fn maintains_fifo_order() { state.fetch(command::Fetch { from: node_a, rid: repo_2, - refs_at: helpers::gen_refs_at(1), + refs: helpers::gen_refs(1), timeout, }); state.fetch(command::Fetch { from: node_a, rid: repo_3, - refs_at: helpers::gen_refs_at(1), + refs: helpers::gen_refs(1), timeout, }); state.fetch(command::Fetch { from: node_a, rid: repo_4, - refs_at: helpers::gen_refs_at(1), + refs: helpers::gen_refs(1), timeout, }); diff --git a/crates/radicle-protocol/src/fetcher/test/state/helpers.rs b/crates/radicle-protocol/src/fetcher/test/state/helpers.rs index 2200316d..1f801ab9 100644 --- a/crates/radicle-protocol/src/fetcher/test/state/helpers.rs +++ b/crates/radicle-protocol/src/fetcher/test/state/helpers.rs @@ -1,8 +1,8 @@ use std::num::NonZeroUsize; -use radicle::{storage::refs::RefsAt, test::arbitrary}; +use radicle::test::arbitrary; -use crate::fetcher::{Config, MaxQueueSize}; +use crate::fetcher::{Config, MaxQueueSize, RefsToFetch}; pub fn config(max_concurrency: usize, max_queue_size: usize) -> Config { Config::new() @@ -12,6 +12,7 @@ pub fn config(max_concurrency: usize, max_queue_size: usize) -> Config { )) } -pub fn gen_refs_at(count: usize) -> Vec { - (0..count).map(|_| arbitrary::gen(1)).collect() +pub fn gen_refs(count: usize) -> RefsToFetch { + let refs: Vec<_> = (0..count).map(|_| arbitrary::gen(1)).collect(); + refs.into() } diff --git a/crates/radicle-protocol/src/fetcher/test/state/invariant.rs b/crates/radicle-protocol/src/fetcher/test/state/invariant.rs index f587d612..9505b8cf 100644 --- a/crates/radicle-protocol/src/fetcher/test/state/invariant.rs +++ b/crates/radicle-protocol/src/fetcher/test/state/invariant.rs @@ -13,21 +13,21 @@ fn queue_integrity_after_merge() { let node_a: NodeId = arbitrary::gen(1); let repo_1: RepoId = arbitrary::gen(1); let repo_2: RepoId = arbitrary::gen(1); - let refs_at_2a = helpers::gen_refs_at(1); - let refs_at_2b = helpers::gen_refs_at(1); + let refs_2a = helpers::gen_refs(1); + let refs_2b = helpers::gen_refs(1); let timeout = Duration::from_secs(30); state.fetch(command::Fetch { from: node_a, rid: repo_1, - refs_at: helpers::gen_refs_at(1), + refs: helpers::gen_refs(1), timeout, }); state.fetch(command::Fetch { from: node_a, rid: repo_2, - refs_at: refs_at_2a.clone(), + refs: refs_2a.clone(), timeout, }); @@ -35,7 +35,7 @@ fn queue_integrity_after_merge() { state.fetch(command::Fetch { from: node_a, rid: repo_2, - refs_at: refs_at_2b.clone(), + refs: refs_2b.clone(), timeout, }); diff --git a/crates/radicle-protocol/src/fetcher/test/state/multinode.rs b/crates/radicle-protocol/src/fetcher/test/state/multinode.rs index c8d2a2d2..6467c32e 100644 --- a/crates/radicle-protocol/src/fetcher/test/state/multinode.rs +++ b/crates/radicle-protocol/src/fetcher/test/state/multinode.rs @@ -22,13 +22,13 @@ fn independent_queues() { state.fetch(command::Fetch { from: node_a, rid: repo_a_active, - refs_at: helpers::gen_refs_at(1), + refs: helpers::gen_refs(1), timeout, }); state.fetch(command::Fetch { from: node_b, rid: repo_b_active, - refs_at: helpers::gen_refs_at(1), + refs: helpers::gen_refs(1), timeout, }); @@ -36,13 +36,13 @@ fn independent_queues() { state.fetch(command::Fetch { from: node_a, rid: repo_a_queued, - refs_at: helpers::gen_refs_at(1), + refs: helpers::gen_refs(1), timeout, }); state.fetch(command::Fetch { from: node_b, rid: repo_b_queued, - refs_at: helpers::gen_refs_at(1), + refs: helpers::gen_refs(1), timeout, }); @@ -73,7 +73,7 @@ fn high_count() { let event = state.fetch(command::Fetch { from: node, rid: repo, - refs_at: helpers::gen_refs_at(1), + refs: helpers::gen_refs(1), timeout, }); assert!(matches!(event, event::Fetch::Started { .. })); diff --git a/crates/radicle-protocol/src/service.rs b/crates/radicle-protocol/src/service.rs index 89c22d22..63bece7c 100644 --- a/crates/radicle-protocol/src/service.rs +++ b/crates/radicle-protocol/src/service.rs @@ -41,6 +41,7 @@ use radicle_fetch::policy::SeedingPolicy; use crate::fetcher; use crate::fetcher::service::FetcherService; use crate::fetcher::FetcherState; +use crate::fetcher::RefsToFetch; use crate::service::gossip::Store as _; use crate::service::message::{ Announcement, AnnouncementMessage, Info, NodeAnnouncement, Ping, RefsAnnouncement, RefsStatus, @@ -1017,7 +1018,7 @@ where let cmd = fetcher::state::command::Fetch { from, rid, - refs_at, + refs: refs_at.into(), timeout, }; let fetcher::service::FetchInitiated { event, rejected } = self.fetcher.fetch(cmd, channel); @@ -1033,14 +1034,14 @@ where fetcher::state::event::Fetch::Started { rid, from, - refs_at, + refs: refs_at, timeout, } => { debug!(target: "service", "Starting fetch for {rid} from {from}"); self.outbox.fetch( session, rid, - refs_at, + refs_at.into(), timeout, self.config.limits.fetch_pack_receive, ); @@ -1073,11 +1074,7 @@ where 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: _, - } => { + fetcher::state::event::Fetched::Completed { from, rid, refs: _ } => { // Notify responders let fetch_result = match &result { Ok(success) => FetchResult::Success { @@ -1184,7 +1181,7 @@ where let Some(fetcher::QueuedFetch { rid, - refs_at, + refs: refs_at, timeout, }) = self.fetcher.dequeue(&nid) else { @@ -1204,12 +1201,15 @@ where debug!(target: "service", "Dequeued fetch for {} from {}", rid, nid); - if let Some(refs) = NonEmpty::from_vec(refs_at.clone()) { - self.fetch_refs_at(rid, nid, refs, scope, timeout); - } else { - // Channel is `None` since they will already be - // registered with the fetcher service. - self.fetch(rid, nid, refs_at, timeout, None); + match refs_at { + RefsToFetch::Refs(refs) => { + self.fetch_refs_at(rid, nid, refs, scope, timeout); + } + RefsToFetch::All => { + // Channel is `None` since they will already be + // registered with the fetcher service. + self.fetch(rid, nid, vec![], timeout, None); + } } } }