From 3e7761740fa2a39c53e1593ed9a37f5984870920 Mon Sep 17 00:00:00 2001 From: Alexis Sellier Date: Sun, 12 Feb 2023 22:16:10 +0100 Subject: [PATCH] node: Rely on inventory announcement in `rad init` --- ...init-announce-refs.md => rad-init-sync.md} | 2 +- radicle-cli/src/commands/init.rs | 4 +- radicle-cli/tests/commands.rs | 10 ++++- radicle-node/src/control.rs | 2 + radicle-node/src/service.rs | 43 ++++++++++++++++--- radicle-node/src/test/peer.rs | 2 +- radicle-node/src/test/simulator.rs | 27 ++++++------ radicle-node/src/tests.rs | 8 ++-- 8 files changed, 71 insertions(+), 27 deletions(-) rename radicle-cli/examples/{rad-init-announce-refs.md => rad-init-sync.md} (95%) diff --git a/radicle-cli/examples/rad-init-announce-refs.md b/radicle-cli/examples/rad-init-sync.md similarity index 95% rename from radicle-cli/examples/rad-init-announce-refs.md rename to radicle-cli/examples/rad-init-sync.md index da387c24..1bf50fdf 100644 --- a/radicle-cli/examples/rad-init-announce-refs.md +++ b/radicle-cli/examples/rad-init-sync.md @@ -13,7 +13,7 @@ ok Project heartwood created "description": "Radicle Heartwood Protocol & Stack", "defaultBranch": "master" } -ok Announcing refs.. +ok Syncing inventory.. Your project id is rad:z42hL2jL4XNk6K8oHQaSWfMgCL7ji. You can show it any time by running: rad . diff --git a/radicle-cli/src/commands/init.rs b/radicle-cli/src/commands/init.rs index bb11cc92..2374cce8 100644 --- a/radicle-cli/src/commands/init.rs +++ b/radicle-cli/src/commands/init.rs @@ -242,8 +242,8 @@ pub fn init(options: Options, profile: &profile::Profile) -> anyhow::Result<()> self::setup_signing(profile.id(), &repo, interactive)?; } if options.sync { - let spinner = term::spinner("Announcing refs.."); - if let Err(e) = node.announce_refs(id) { + let spinner = term::spinner("Syncing inventory.."); + if let Err(e) = node.sync_inventory() { spinner.error(e); } else { spinner.finish(); diff --git a/radicle-cli/tests/commands.rs b/radicle-cli/tests/commands.rs index 24006a71..fda80ca6 100644 --- a/radicle-cli/tests/commands.rs +++ b/radicle-cli/tests/commands.rs @@ -1,5 +1,6 @@ use std::env; use std::path::Path; +use std::{thread, time}; use radicle::git; use radicle::profile::Home; @@ -226,7 +227,7 @@ fn rad_clone_unknown() { } #[test] -fn rad_init_announce_refs_and_clone() { +fn rad_init_sync_and_clone() { logger::init(log::Level::Debug); let mut environment = Environment::new(); @@ -241,9 +242,14 @@ fn rad_init_announce_refs_and_clone() { fixtures::repository(working.join("alice")); + // Necessary for now, if we don't want the new inventry announcement to be considered stale + // for Bob. + // TODO: Find a way to advance internal clocks instead. + thread::sleep(time::Duration::from_secs(1)); + // Alice initializes a repo after her node has started, and after bob has connected to it. test( - "examples/rad-init-announce-refs.md", + "examples/rad-init-sync.md", &working.join("alice"), Some(&alice.home), [], diff --git a/radicle-node/src/control.rs b/radicle-node/src/control.rs index 1c58f1ae..314b4041 100644 --- a/radicle-node/src/control.rs +++ b/radicle-node/src/control.rs @@ -42,6 +42,8 @@ pub fn listen handle.shutdown().ok(); break; } + log::error!(target: "control", "Command returned error: {e}"); + CommandResult::error(e).to_writer(&mut stream).ok(); stream.flush().ok(); diff --git a/radicle-node/src/service.rs b/radicle-node/src/service.rs index 21bdde1e..c39ac800 100644 --- a/radicle-node/src/service.rs +++ b/radicle-node/src/service.rs @@ -579,6 +579,15 @@ where } else { log::debug!(target: "service", "No fetch requests found for {rid}.."); } + + // Announce the newly fetched project to the network, if necessary. + // Since this fetch could be either a full clone or simply a ref update, we need to + // either announce new inventory, or new refs. + // + // TODO: Announce new refs? Would require the ability to announce other peer refs. + if let Err(e) = self.sync_and_announce_inventory() { + error!(target: "service", "Failed to announce new inventory: {e}"); + } } if let Some(session) = self.sessions.get_mut(&remote) { @@ -760,6 +769,35 @@ where return Ok(false); } } + + for id in message.inventory.as_slice() { + if let Some(sess) = self.sessions.get_mut(announcer) { + // If we are connected to the announcer of this inventory, update the peer's + // subscription filter to include all inventory items. This way, we'll + // relay messages relating to the peer's inventory. + if let Some(sub) = &mut sess.subscribe { + sub.filter.insert(id); + } + + // If we're tracking and connected to the announcer, and we don't have + // the inventory, fetch it from the announcer. + if self.tracking.is_repo_tracked(id).expect( + "Service::handle_announcement: error accessing tracking configuration", + ) { + // Only if we do not have the repository locally do we fetch here. + // If we do have it, only fetch after receiving a ref announcement. + if let Ok(true) = self.storage.contains(id) { + // Do nothing. + } else { + // We may hit this branch due to an error returned by storage. + // We attempt to fetch in case of error because it's likely + // the repository was corrupted, and fetching will fix it. + self.fetch(*id, announcer); + } + } + } + } + return Ok(relay); } // Process a peer inventory update announcement by (maybe) fetching. @@ -1068,11 +1106,6 @@ where let peers = self.sessions.negotiated().map(|(_, p)| p); let timestamp = self.clock.as_secs(); - // Make sure our local routing table is up to date with new repositories. - if let Err(e) = self.routing.insert(id, self.node_id(), timestamp) { - log::error!(target: "service", "Failed to update routing table with repository {id}: {e}"); - } - if remote.refs.len() > Refs::max() { error!( target: "service", diff --git a/radicle-node/src/test/peer.rs b/radicle-node/src/test/peer.rs index 245262b7..0bffb833 100644 --- a/radicle-node/src/test/peer.rs +++ b/radicle-node/src/test/peer.rs @@ -202,7 +202,7 @@ where pub fn inventory_announcement(&self) -> Message { Message::inventory( InventoryAnnouncement { - inventory: arbitrary::gen(3), + inventory: arbitrary::vec(3).try_into().unwrap(), timestamp: self.timestamp(), }, self.signer(), diff --git a/radicle-node/src/test/simulator.rs b/radicle-node/src/test/simulator.rs index 5a06fc38..91bff090 100644 --- a/radicle-node/src/test/simulator.rs +++ b/radicle-node/src/test/simulator.rs @@ -108,8 +108,8 @@ impl fmt::Display for Scheduled { Input::Fetched(fetch, _) => { write!( f, - "{} <~ {} ({}): Fetched", - self.node, fetch.remote, fetch.rid + "{} <<~ {} ({}): Fetched (initiated={})", + self.node, fetch.remote, fetch.rid, fetch.initiated ) } } @@ -410,10 +410,10 @@ impl Simulation { } Input::Fetched(f, result) => { let result = Rc::try_unwrap(result).unwrap(); - let mut repo = p.storage().repository(f.rid).unwrap(); - - fetch(&mut repo, &f.remote, f.namespaces.clone()).unwrap(); - + if f.initiated { + let mut repo = p.storage().repository(f.rid).unwrap(); + fetch(&mut repo, &f.remote, f.namespaces.clone()).unwrap(); + } p.fetched(f, result); } } @@ -608,17 +608,19 @@ impl Simulation { } } Io::Fetch(fetch) => { + let remote = fetch.remote; + if fetch.initiated { log::info!( target: "sim", - "{:05} {} ~> {} ({})", - self.elapsed().as_millis(), node, fetch.remote, fetch.rid + "{:05} {} ~> {} ({}): Fetch outgoing", + self.elapsed().as_millis(), node, remote, fetch.rid ); } else { log::info!( target: "sim", - "{:05} {} <~ {} ({})", - self.elapsed().as_millis(), node, fetch.remote, fetch.rid + "{:05} {} <~ {} ({}): Fetch incoming", + self.elapsed().as_millis(), node, remote, fetch.rid ); } @@ -627,7 +629,7 @@ impl Simulation { self.time + LocalDuration::from_secs(3), Scheduled { node, - remote: fetch.remote, + remote, input: Input::Fetched( fetch, Rc::new(Err(FetchError::Io(io::ErrorKind::Other.into()))), @@ -639,7 +641,7 @@ impl Simulation { self.time + LocalDuration::from_secs(3), Scheduled { node, - remote: fetch.remote, + remote, input: Input::Fetched(fetch, Rc::new(Ok(vec![]))), }, ); @@ -694,7 +696,6 @@ fn fetch( }); opts.remote_callbacks(callbacks); - // Fetch from the remote into the staging copy. let mut remote = repo.raw().remote_anonymous( radicle::storage::git::transport::remote::Url { node: *node, diff --git a/radicle-node/src/tests.rs b/radicle-node/src/tests.rs index fb0bd8ac..3d5ab0b2 100644 --- a/radicle-node/src/tests.rs +++ b/radicle-node/src/tests.rs @@ -548,8 +548,8 @@ fn test_refs_announcement_relay() { alice.connect_to(&bob); alice.connect_to(&eve); alice.receive(eve.id(), Message::Subscribe(Subscribe::all())); - alice.receive(bob.id(), bob.refs_announcement(bob_inv[0])); + assert_matches!( alice.messages(eve.id()).next(), Some(Message::Announcement(_)), @@ -896,11 +896,13 @@ fn test_push_and_pull() { assert!(eve.get(proj_id).unwrap().is_none()); assert!(bob.get(proj_id).unwrap().is_none()); - // Alice announces her refs. + let (send, _) = chan::bounded(1); + // Alice announces her inventory. // We now expect Eve to fetch Alice's project from Alice. // Then we expect Bob to fetch Alice's project from Eve. alice.elapse(LocalDuration::from_secs(1)); // Make sure our announcement is fresh. - alice.command(service::Command::AnnounceRefs(proj_id)); + alice.command(service::Command::SyncInventory(send)); + sim.run_while([&mut alice, &mut bob, &mut eve], |s| !s.is_settled()); // TODO: Refs should be compared between the two peers.