From d77b5d5abe1032d48bd7aa0c1cb70272813674ac Mon Sep 17 00:00:00 2001 From: cloudhead Date: Tue, 23 Apr 2024 09:52:25 +0200 Subject: [PATCH] node: Add 'disconnect' command to handle --- radicle-node/src/control.rs | 6 +++++ radicle-node/src/runtime/handle.rs | 14 ++++++++++ radicle-node/src/test/environment.rs | 30 +++++++++++++++++++++ radicle-node/src/test/handle.rs | 4 +++ radicle-node/src/tests/e2e.rs | 40 ++++++++++++++++++++++++++++ radicle/src/node.rs | 14 ++++++++++ 6 files changed, 108 insertions(+) diff --git a/radicle-node/src/control.rs b/radicle-node/src/control.rs index b306793f..e491e519 100644 --- a/radicle-node/src/control.rs +++ b/radicle-node/src/control.rs @@ -102,6 +102,12 @@ where } } } + Command::Disconnect { nid } => match handle.disconnect(nid) { + Err(e) => return Err(CommandError::Runtime(e)), + Ok(()) => { + CommandResult::ok().to_writer(writer).ok(); + } + }, Command::Fetch { rid, nid, timeout } => { fetch(rid, nid, timeout, writer, &mut handle)?; } diff --git a/radicle-node/src/runtime/handle.rs b/radicle-node/src/runtime/handle.rs index a96c2456..16054129 100644 --- a/radicle-node/src/runtime/handle.rs +++ b/radicle-node/src/runtime/handle.rs @@ -179,6 +179,20 @@ impl radicle::node::Handle for Handle { .map_err(Error::from) } + fn disconnect(&mut self, node: NodeId) -> Result<(), Self::Error> { + let events = self.events(); + self.command(service::Command::Disconnect(node))?; + events + .wait( + |e| match e { + Event::PeerDisconnected { nid, .. } if nid == &node => Some(()), + _ => None, + }, + time::Duration::MAX, + ) + .map_err(Error::from) + } + fn seeds(&mut self, id: RepoId) -> Result { let (sender, receiver) = chan::bounded(1); self.command(service::Command::Seeds(id, sender))?; diff --git a/radicle-node/src/test/environment.rs b/radicle-node/src/test/environment.rs index c124c44c..205c9b36 100644 --- a/radicle-node/src/test/environment.rs +++ b/radicle-node/src/test/environment.rs @@ -232,6 +232,15 @@ impl NodeHandle { self } + pub fn disconnect(&mut self, remote: &NodeHandle) { + self.handle.disconnect(remote.id).unwrap(); + } + + /// Shutdown node. + pub fn shutdown(self) { + drop(self) + } + /// Get the full address of this node. pub fn address(&self) -> ConnectAddress { (self.id, node::Address::from(self.addr)).into() @@ -301,6 +310,26 @@ impl NodeHandle { } } + /// Wait until this node has a repository. + #[track_caller] + pub fn has_repository(&self, rid: &RepoId) { + log::debug!(target: "test", "Waiting for {} to have {rid}", self.id); + let events = self.handle.events(); + + loop { + if self.storage.repository(*rid).is_ok() { + log::debug!(target: "test", "Node {} has {rid}", self.id); + break; + } + events + .wait( + |e| matches!(e, Event::RefsFetched { .. }).then_some(()), + time::Duration::from_secs(6), + ) + .unwrap(); + } + } + /// Wait until this node has the inventory of another node. #[track_caller] pub fn has_remote_of(&self, rid: &RepoId, nid: &NodeId) { @@ -310,6 +339,7 @@ impl NodeHandle { loop { if let Ok(repo) = self.storage.repository(*rid) { if repo.remote(nid).is_ok() { + log::debug!(target: "test", "Node {} has {rid}/{nid}", self.id); break; } } diff --git a/radicle-node/src/test/handle.rs b/radicle-node/src/test/handle.rs index 231a7e43..87de45ee 100644 --- a/radicle-node/src/test/handle.rs +++ b/radicle-node/src/test/handle.rs @@ -48,6 +48,10 @@ impl radicle::node::Handle for Handle { unimplemented!(); } + fn disconnect(&mut self, _node: NodeId) -> Result<(), Self::Error> { + unimplemented!(); + } + fn seeds(&mut self, _id: RepoId) -> Result { unimplemented!(); } diff --git a/radicle-node/src/tests/e2e.rs b/radicle-node/src/tests/e2e.rs index eeaec58c..0dec00ad 100644 --- a/radicle-node/src/tests/e2e.rs +++ b/radicle-node/src/tests/e2e.rs @@ -1265,3 +1265,43 @@ fn test_background_foreground_fetch() { assert_ne!(eves_refs_expected, old_refs); assert_eq!(eves_refs_expected, eves_refs); } + +#[test] +/// Alice is offline while Bob pushes some changes to the repo. When Alice reconnects, +/// she is made aware of the changes via the `subscribe` message, and fetches from the seed. +fn test_catchup_on_refs_announcements() { + logger::init(log::Level::Debug); + + let tmp = tempfile::tempdir().unwrap(); + let mut alice = Node::init(tmp.path(), Config::test(Alias::new("alice"))); + let bob = Node::init(tmp.path(), Config::test(Alias::new("bob"))); + let bob_id = bob.id; + let seed = Node::init(tmp.path(), Config::test(Alias::new("seed"))); + let acme = alice.project("acme", ""); + + let mut alice = alice.spawn(); + let mut bob = bob.spawn(); + let mut seed = seed.spawn(); + + bob.handle.seed(acme, Scope::All).unwrap(); + seed.handle.seed(acme, Scope::All).unwrap(); + + alice.connect(&seed); + seed.has_repository(&acme); + alice.disconnect(&seed); + bob.connect(&seed); + bob.has_repository(&acme); + + log::debug!(target: "test", "Bob creating his issue.."); + bob.issue(acme, "Bob's issue", "[..]"); + bob.handle.announce_refs(acme).unwrap(); + + log::debug!(target: "test", "Waiting for seed to fetch Bob's refs from Bob.."); + seed.has_remote_of(&acme, &bob.id); // Seed fetches Bob's refs. + bob.disconnect(&seed); + bob.shutdown(); + + log::debug!(target: "test", "Alice re-connects to the seed.."); + alice.connect(&seed); + alice.has_remote_of(&acme, &bob_id); +} diff --git a/radicle/src/node.rs b/radicle/src/node.rs index d2bfd793..ed2d308d 100644 --- a/radicle/src/node.rs +++ b/radicle/src/node.rs @@ -455,6 +455,10 @@ pub enum Command { opts: ConnectOptions, }, + /// Disconnect from a node. + #[serde(rename_all = "camelCase")] + Disconnect { nid: NodeId }, + /// Lookup seeds for the given repository in the routing table. #[serde(rename_all = "camelCase")] Seeds { rid: RepoId }, @@ -862,6 +866,8 @@ pub trait Handle: Clone + Sync + Send { addr: Address, opts: ConnectOptions, ) -> Result; + /// Disconnect from a peer. + fn disconnect(&mut self, node: NodeId) -> Result<(), Self::Error>; /// Lookup the seeds of a given repository in the routing table. fn seeds(&mut self, id: RepoId) -> Result; /// Fetch a repository from the network. @@ -1069,6 +1075,14 @@ impl Handle for Node { Ok(result) } + fn disconnect(&mut self, nid: NodeId) -> Result<(), Self::Error> { + self.call::(Command::Disconnect { nid }, DEFAULT_TIMEOUT)? + .next() + .ok_or(Error::EmptyResponse)??; + + Ok(()) + } + fn seeds(&mut self, rid: RepoId) -> Result { let seeds = self .call::(Command::Seeds { rid }, DEFAULT_TIMEOUT)?