use std::fmt::Debug; use std::io::BufRead as _; use std::mem::ManuallyDrop; use std::net::Ipv4Addr; use std::path::Path; use std::{ collections::{BTreeMap, BTreeSet}, fs, io, iter, net, process, thread, time, time::Duration, }; use crossbeam_channel as chan; use radicle::cob; use radicle::cob::issue; use radicle::crypto::signature::Signer; use radicle::crypto::ssh::keystore::MemorySigner; use radicle::crypto::test::signer::MockSigner; use radicle::crypto::Signature; use radicle::git; use radicle::git::fmt::refname; use radicle::identity::{RepoId, Visibility}; use radicle::node::config::ConnectAddress; use radicle::node::policy::store as policy; use radicle::node::seed::Store as _; use radicle::node::Config; use radicle::node::Event; use radicle::node::{self, Alias}; use radicle::node::{ConnectOptions, Handle as _}; use radicle::node::{Database, POLICIES_DB_FILE}; use radicle::profile::{env, Home, Profile}; use radicle::rad; use radicle::storage::{ReadStorage as _, RemoteRepository as _, SignRepository as _}; use radicle::test::fixtures; use radicle::Storage; use crate::node::device::Device; use crate::node::NodeId; use crate::storage::git::transport; use crate::{runtime, runtime::Handle, service, Runtime}; /// A node that can be run. pub struct Node { pub id: NodeId, pub home: Home, pub signer: Device, pub storage: Storage, pub config: Config, pub db: service::Stores, pub policies: policy::Store, } impl Node { pub fn new(profile: Profile) -> Self { let signer = Device::from(MemorySigner::load(&profile.keystore, None).unwrap()); let id = *profile.id(); let policies_db = profile.home.node().join(POLICIES_DB_FILE); let policies = policy::Store::open(policies_db).unwrap(); let db = profile.database_mut().unwrap(); let db = service::Stores::from(db); Node { id, home: profile.home, config: profile.config.node, signer, db, policies, storage: profile.storage, } } } /// Handle to a running node. pub struct NodeHandle { pub id: NodeId, pub alias: Alias, pub storage: Storage, pub signer: Device, pub home: Home, pub addr: net::SocketAddr, pub thread: ManuallyDrop>>, pub handle: ManuallyDrop, } impl Drop for NodeHandle { fn drop(&mut self) { log::debug!(target: "test", "Node {} shutting down..", self.id); unsafe { ManuallyDrop::take(&mut self.handle) } .shutdown() .unwrap(); unsafe { ManuallyDrop::take(&mut self.thread) } .join() .unwrap() .unwrap(); } } impl + cyphernet::Ecdh> NodeHandle { /// Connect this node to another node, and wait for the connection to be established both ways. /// /// If the remote has blocked this node, then the remote event will be /// [`Event::PeerDisconnected`]. pub fn connect(&mut self, remote: &NodeHandle) -> &mut Self { let local_events = self.handle.events(); let remote_events = remote.handle.events(); self.handle .connect(remote.id, remote.addr.into(), ConnectOptions::default()) .ok(); local_events .iter() .find(|e| { matches!( e, Event::PeerConnected { nid } if nid == &remote.id ) }) .unwrap(); remote_events .iter() .find(|e| { matches!( e, Event::PeerConnected { nid } | Event::PeerDisconnected { nid, .. } if nid == &self.id ) }) .unwrap(); 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() } /// Get routing table entries. pub fn routing(&self) -> impl Iterator { use node::routing::Store as _; self.home .routing_mut(node::db::config::Config::default()) .unwrap() .entries() .unwrap() } pub fn inventory(&self) -> impl Iterator + '_ { self.routing() .filter(|(_, n)| *n == self.id) .map(|(r, _)| r) } /// Get sync status of a repo. pub fn synced_seeds(&self, rid: &RepoId) -> Vec { let db = Database::reader( self.home.node().join(node::NODE_DB_FILE), node::db::config::Config::default(), ) .unwrap(); let seeds = db.seeds_for(rid).unwrap(); seeds.into_iter().collect::, _>>().unwrap() } /// Wait until this node's routing table matches the remotes. pub fn converge<'a>( &'a self, remotes: impl IntoIterator>, ) -> BTreeSet<(RepoId, NodeId)> { converge(iter::once(self).chain(remotes)) } /// Wait until this node's routing table contains the given routes. #[track_caller] pub fn routes_to(&self, routes: &[(RepoId, NodeId)]) { log::debug!(target: "test", "Waiting for {} to route to {:?}", self.id, routes); let events = self.handle.events(); loop { let mut remaining: BTreeSet<_> = routes.iter().collect(); for (rid, nid) in self.routing() { if !remaining.remove(&(rid, nid)) { log::debug!(target: "test", "Found unexpected route for {}: ({rid}, {nid})", self.id); } } if remaining.is_empty() { break; } events .wait( |e| matches!(e, Event::SeedDiscovered { .. }).then_some(()), time::Duration::from_secs(6), ) .unwrap(); } } /// Wait until this node is synced with another node, for the given repository. #[track_caller] pub fn is_synced_with(&mut self, rid: &RepoId, nid: &NodeId) { log::debug!(target: "test", "Waiting for {} to be in sync with {nid} for {rid}", self.id); loop { let seeds = self.handle.seeds_for(*rid, [self.id]).unwrap(); if seeds.iter().any(|s| s.nid == *nid && s.is_synced()) { break; } thread::sleep(Duration::from_millis(100)); } } /// 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) { log::debug!(target: "test", "Waiting for {} to have {rid}/{nid}", self.id); let events = self.handle.events(); 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; } } events .wait( |e| matches!(e, Event::RefsFetched { .. }).then_some(()), time::Duration::from_secs(6), ) .unwrap(); } } /// Clone a repo into a directory. pub fn clone>(&self, rid: RepoId, cwd: P) -> io::Result<()> { self.rad("clone", &[rid.to_string().as_str()], cwd) } /// Fork a repo. pub fn fork>(&self, rid: RepoId, cwd: P) -> io::Result<()> { self.clone(rid, &cwd)?; self.rad("fork", &[rid.to_string().as_str()], &cwd)?; self.announce(rid, 1, &cwd)?; Ok(()) } /// Announce a repo. pub fn announce>(&self, rid: RepoId, replicas: usize, cwd: P) -> io::Result<()> { self.rad( "sync", &[ rid.to_string().as_str(), "--announce", "--replicas", replicas.to_string().as_str(), ], cwd, ) } /// Init a repo. pub fn init>(&self, name: &str, desc: &str, cwd: P) -> io::Result<()> { self.rad( "init", &[ "--name", name, "--description", desc, "--default-branch", "master", "--public", ], cwd, ) } /// Run a `rad` CLI command. pub fn rad>(&self, cmd: &str, args: &[&str], cwd: P) -> io::Result<()> { let cwd = cwd.as_ref(); log::debug!(target: "test", "Running `rad {cmd} {args:?}` in {}..", cwd.display()); fs::create_dir_all(cwd)?; let result = process::Command::new(snapbox::cmd::cargo_bin("rad")) .env_clear() .envs(env::vars().filter(|(k, _)| k == "PATH")) .env("GIT_AUTHOR_DATE", "1671125284") .env("GIT_AUTHOR_EMAIL", "radicle@localhost") .env("GIT_AUTHOR_NAME", "radicle") .env("GIT_COMMITTER_DATE", "1671125284") .env("GIT_COMMITTER_EMAIL", "radicle@localhost") .env("GIT_COMMITTER_NAME", "radicle") .env( env::RAD_HOME, self.home.path().to_string_lossy().to_string(), ) .env(env::RAD_PASSPHRASE, "radicle") .env(env::RAD_LOCAL_TIME, "1671125284") .env("TZ", "UTC") .env("LANG", "C") .envs(git::env::GIT_DEFAULT_CONFIG) .current_dir(cwd) .arg(cmd) .args(args) .output()?; for line in io::BufReader::new(io::Cursor::new(&result.stdout)) .lines() .map_while(Result::ok) { log::debug!(target: "test", "rad {cmd}: {line}"); } log::debug!( target: "test", "Ran command `rad {cmd}` (status={})", result.status.code().unwrap() ); if !result.status.success() { return Err(io::ErrorKind::Other.into()); } Ok(()) } /// Create an [`issue::Issue`] in the `NodeHandle`'s storage. pub fn issue(&self, rid: RepoId, title: cob::Title, desc: &str) -> cob::ObjectId { let repo = self.storage.repository(rid).unwrap(); let mut issues = issue::Cache::no_cache(&repo).unwrap(); *issues .create(title, desc, &[], &[], [], &self.signer) .unwrap() .id() } /// Perform a commit to `refname` by generating a blob of random data to a /// random path in a new tree. /// /// If the reference does not exist, a new one will be created with the new /// commit as its target. /// /// If the reference already exists, then its target is used as the parent /// of the new commit, and the reference will be updated. /// /// The `rad/sigrefs` are then updated to reflect the new change. pub fn commit_to(&self, rid: RepoId, refname: impl AsRef) { use radicle::test::arbitrary; let refname = refname.as_ref(); let repo = self.storage.repository(rid).unwrap(); let raw = &repo.backend; let info = self.storage.info(); let author = git::raw::Signature::now(&info.name(), &info.email()).unwrap(); let tree = { let mut tb = raw.treebuilder(None).unwrap(); let blob = raw.blob(&arbitrary::vec::(100)).unwrap(); tb.insert( arbitrary::alphanumeric(10), blob, git::raw::FileMode::Blob.into(), ) .unwrap(); let oid = tb.write().unwrap(); raw.find_tree(oid).unwrap() }; let parent = { let target = raw .find_reference(refname.as_str()) .ok() .and_then(|r| r.target()); target.and_then(|oid| raw.find_commit(oid).ok()) }; match parent { None => repo .backend .commit( Some(refname.as_str()), &author, &author, "New commit", &tree, &[], ) .unwrap(), Some(parent) => repo .backend .commit( Some(refname.as_str()), &author, &author, "New commit", &tree, &[&parent], ) .unwrap(), }; repo.sign_refs(&self.signer).unwrap(); } } impl Node { /// Create a new node. pub fn init(base: &Path, config: Config) -> Self { let home = base.join( iter::repeat_with(fastrand::alphanumeric) .take(8) .collect::(), ); let home = Home::new(home).unwrap(); let signer = Device::mock(); let storage = Storage::open( home.storage(), git::UserInfo { alias: config.alias.clone(), key: *signer.public_key(), }, ) .unwrap(); let policies = home.policies_mut().unwrap(); let db = home .database_mut(node::db::config::Config::default()) .unwrap(); let db = service::Stores::from(db); log::debug!(target: "test", "Node::init {}: {}", config.alias, signer.public_key()); Self { id: *signer.public_key(), home, signer, storage, config, db, policies, } } } impl + Signer + Clone + Debug> Node { /// Spawn a node in its own thread. pub fn spawn(self) -> NodeHandle { let alias = self.config.alias.clone(); let listen = vec![(Ipv4Addr::LOCALHOST, 0).into()]; let (_, signals) = chan::bounded(1); let rt = Runtime::init( self.home.clone(), self.config, listen, signals, self.signer.clone(), ) .unwrap(); let addr = *rt.local_addrs.first().unwrap(); let id = *self.signer.public_key(); let handle = ManuallyDrop::new(rt.handle.clone()); let thread = ManuallyDrop::new(runtime::thread::spawn(&id, "runtime", move || rt.run())); NodeHandle { id, alias, storage: self.storage, signer: self.signer, home: self.home, addr, handle, thread, } } /// Populate a storage instance with a project from the given repository. pub fn project_from( &mut self, name: &str, description: &str, repo: &git::raw::Repository, ) -> RepoId { transport::local::register(self.storage.clone()); let branch = refname!("master"); let id = rad::init( repo, name.try_into().unwrap(), description, branch.clone(), Visibility::default(), &self.signer, &self.storage, ) .map(|(id, _, _)| id) .unwrap(); assert!(self.policies.seed(&id, node::policy::Scope::All).unwrap()); log::debug!( target: "test", "Initialized project {id} for node {}", self.signer.public_key() ); // Push local branches to storage. let mut refs = Vec::<(git::fmt::Qualified, git::fmt::Qualified)>::new(); for branch in repo.branches(Some(git::raw::BranchType::Local)).unwrap() { let (branch, _) = branch.unwrap(); let name = git::fmt::RefString::try_from(branch.name().unwrap().unwrap()).unwrap(); refs.push(( git::fmt::lit::refs_heads(&name).into(), git::fmt::lit::refs_heads(&name).into(), )); } git::push(repo, "rad", refs.iter().map(|(a, b)| (a, b))).unwrap(); radicle::git::set_upstream( repo, &*radicle::rad::REMOTE_NAME, branch.clone(), radicle::git::refs::workdir::branch(&branch), ) .unwrap(); self.storage .repository(id) .unwrap() .sign_refs(&self.signer) .unwrap(); id } /// Populate a storage instance with a project. pub fn project(&mut self, name: &str, description: &str) -> RepoId { let tmp = tempfile::tempdir().unwrap(); let (repo, _) = fixtures::repository(tmp.path()); self.project_from(name, description, &repo) } } /// Checks whether the nodes have converged in their routing tables. #[track_caller] pub fn converge<'a, G: Signer + cyphernet::Ecdh + 'static>( nodes: impl IntoIterator>, ) -> BTreeSet<(RepoId, NodeId)> { let nodes = nodes.into_iter().collect::>(); let mut all_routes = BTreeSet::<(RepoId, NodeId)>::new(); let mut remaining = BTreeMap::from_iter(nodes.iter().map(|node| (node.id, node))); // First build the set of all routes. for node in &nodes { // Routes from the routing table. for (rid, seed_id) in node.routing() { all_routes.insert((rid, seed_id)); } // Routes from the local inventory. for rid in node.inventory() { all_routes.insert((rid, node.id)); } } // Then, while there are nodes remaining to converge, check each node to see if // its routing table has all routes. If so, remove it from the remaining nodes. while !remaining.is_empty() { remaining.retain(|_, node| { let routing = node.routing(); let routes = BTreeSet::from_iter(routing); if routes.is_superset(&all_routes) { log::debug!(target: "test", "Node {} has converged", node.id); return false; } else { let diff = all_routes.symmetric_difference(&routes).collect::>(); log::debug!(target: "test", "Node has missing routes: {diff:?}"); } true }); thread::sleep(Duration::from_millis(100)); } all_routes }