diff --git a/radicle-node/src/control.rs b/radicle-node/src/control.rs index 4df4bc8b..f2aafe69 100644 --- a/radicle-node/src/control.rs +++ b/radicle-node/src/control.rs @@ -5,7 +5,6 @@ use std::io::LineWriter; use std::os::unix::net::UnixListener; use std::os::unix::net::UnixStream; use std::path::PathBuf; -use std::str::FromStr; use std::{io, net, time}; use radicle::node::Handle; @@ -13,7 +12,7 @@ use serde_json as json; use crate::identity::Id; use crate::node::NodeId; -use crate::node::{Alias, Command, CommandName, CommandResult}; +use crate::node::{Command, CommandResult}; use crate::runtime; use crate::runtime::thread; @@ -67,11 +66,7 @@ where #[derive(thiserror::Error, Debug)] enum CommandError { - #[error("invalid command argument `{0}`, {1}")] - InvalidCommandArg(String, Box), - #[error("invalid command arguments `{0:?}`")] - InvalidCommandArgs(Vec), - #[error("serialization failed: {0}")] + #[error("(de)serialization failed: {0}")] Serialization(#[from] json::Error), #[error("runtime error: {0}")] Runtime(#[from] runtime::HandleError), @@ -96,106 +91,29 @@ where log::debug!(target: "control", "Received `{input}` on control socket"); let cmd: Command = json::from_str(input)?; - match cmd.name { - CommandName::Connect => { - let (nid, addr) = parse::args(cmd)?; + match cmd { + Command::Connect { addr } => { + let (nid, addr) = addr.into(); if let Err(e) = handle.connect(nid, addr) { return Err(CommandError::Runtime(e)); } else { CommandResult::Okay { updated: true }.to_writer(writer)?; } } - CommandName::Fetch => { - let (rid, nid): (Id, NodeId) = parse::args(cmd)?; + Command::Fetch { rid, nid } => { fetch(rid, nid, writer, &mut handle)?; } - CommandName::Seeds => { - let rid: Id = parse::arg(cmd)?; + Command::Seeds { rid } => { let seeds = handle.seeds(rid)?; json::to_writer(writer, &seeds)?; } - CommandName::Sessions => { + Command::Sessions => { let sessions = handle.sessions()?; json::to_writer(writer, &sessions)?; } - CommandName::TrackRepo => { - let (rid, scope) = parse::args(cmd)?; - - match handle.track_repo(rid, scope) { - Ok(updated) => { - CommandResult::Okay { updated }.to_writer(writer)?; - } - Err(e) => { - return Err(CommandError::Runtime(e)); - } - } - } - CommandName::UntrackRepo => { - let rid: Id = parse::arg(cmd)?; - - match handle.untrack_repo(rid) { - Ok(updated) => { - CommandResult::Okay { updated }.to_writer(writer)?; - } - Err(e) => { - return Err(CommandError::Runtime(e)); - } - } - } - CommandName::TrackNode => { - let (node, alias) = match cmd.args.as_slice() { - [node] => (node.as_str(), None), - [node, alias] => (node.as_str(), Some(alias.to_owned())), - _ => return Err(CommandError::InvalidCommandArgs(cmd.args)), - }; - let alias = alias - .map(|a| { - Alias::from_str(&a) - .map_err(|e| CommandError::InvalidCommandArg(node.to_owned(), Box::new(e))) - }) - .transpose()?; - let nid = node - .parse() - .map_err(|e| CommandError::InvalidCommandArg(node.to_owned(), Box::new(e)))?; - - match handle.track_node(nid, alias) { - Ok(updated) => { - CommandResult::Okay { updated }.to_writer(writer)?; - } - Err(e) => { - return Err(CommandError::Runtime(e)); - } - } - } - CommandName::UntrackNode => { - let nid: NodeId = parse::arg(cmd)?; - - match handle.untrack_node(nid) { - Ok(updated) => { - CommandResult::Okay { updated }.to_writer(writer)?; - } - Err(e) => { - return Err(CommandError::Runtime(e)); - } - } - } - CommandName::AnnounceRefs => { - let rid: Id = parse::arg(cmd)?; - - if let Err(e) = handle.announce_refs(rid) { - return Err(CommandError::Runtime(e)); - } - CommandResult::ok().to_writer(writer).ok(); - } - CommandName::AnnounceInventory => { - if let Err(e) = handle.announce_inventory() { - return Err(CommandError::Runtime(e)); - } - CommandResult::ok().to_writer(writer).ok(); - } - CommandName::SyncInventory => match handle.sync_inventory() { + Command::TrackRepo { rid, scope } => match handle.track_repo(rid, scope) { Ok(updated) => { CommandResult::Okay { updated }.to_writer(writer)?; } @@ -203,7 +121,51 @@ where return Err(CommandError::Runtime(e)); } }, - CommandName::Subscribe => match handle.subscribe(MAX_TIMEOUT) { + Command::UntrackRepo { rid } => match handle.untrack_repo(rid) { + Ok(updated) => { + CommandResult::Okay { updated }.to_writer(writer)?; + } + Err(e) => { + return Err(CommandError::Runtime(e)); + } + }, + Command::TrackNode { nid, alias } => match handle.track_node(nid, alias) { + Ok(updated) => { + CommandResult::Okay { updated }.to_writer(writer)?; + } + Err(e) => { + return Err(CommandError::Runtime(e)); + } + }, + Command::UntrackNode { nid } => match handle.untrack_node(nid) { + Ok(updated) => { + CommandResult::Okay { updated }.to_writer(writer)?; + } + Err(e) => { + return Err(CommandError::Runtime(e)); + } + }, + Command::AnnounceRefs { rid } => { + if let Err(e) = handle.announce_refs(rid) { + return Err(CommandError::Runtime(e)); + } + CommandResult::ok().to_writer(writer).ok(); + } + Command::AnnounceInventory => { + if let Err(e) = handle.announce_inventory() { + return Err(CommandError::Runtime(e)); + } + CommandResult::ok().to_writer(writer).ok(); + } + Command::SyncInventory => match handle.sync_inventory() { + Ok(updated) => { + CommandResult::Okay { updated }.to_writer(writer)?; + } + Err(e) => { + return Err(CommandError::Runtime(e)); + } + }, + Command::Subscribe => match handle.subscribe(MAX_TIMEOUT) { Ok(events) => { for e in events { let event = e?; @@ -214,16 +176,16 @@ where } Err(e) => log::error!(target: "control", "Error subscribing to events: {e}"), }, - CommandName::Status => { + Command::Status => { CommandResult::ok().to_writer(writer).ok(); } - CommandName::NodeId => match handle.nid() { + Command::NodeId => match handle.nid() { Ok(nid) => { writeln!(writer, "{nid}")?; } Err(e) => return Err(CommandError::Runtime(e)), }, - CommandName::Shutdown => { + Command::Shutdown => { log::debug!(target: "control", "Shutdown requested.."); // Channel might already be disconnected if shutdown // came from somewhere else. Ignore errors. @@ -251,45 +213,6 @@ fn fetch>( Ok(()) } -mod parse { - use super::*; - - pub(super) fn arg(cmd: Command) -> Result - where - ::Err: std::error::Error + 'static, - { - let [arg]: [String; 1] = cmd - .args - .clone() - .try_into() - .map_err(|_| CommandError::InvalidCommandArgs(cmd.args))?; - - arg.parse() - .map_err(|e| CommandError::InvalidCommandArg(arg, Box::new(e))) - } - - pub(super) fn args(cmd: Command) -> Result<(S, T), CommandError> - where - ::Err: std::error::Error + 'static, - ::Err: std::error::Error + 'static, - { - let [arg1, arg2]: [String; 2] = cmd - .args - .clone() - .try_into() - .map_err(|_| CommandError::InvalidCommandArgs(cmd.args))?; - - let arg1 = arg1 - .parse() - .map_err(|e| CommandError::InvalidCommandArg(arg1, Box::new(e)))?; - let arg2 = arg2 - .parse() - .map_err(|e| CommandError::InvalidCommandArg(arg2, Box::new(e)))?; - - Ok((arg1, arg2)) - } -} - #[cfg(test)] mod tests { use std::io::prelude::*; @@ -299,7 +222,7 @@ mod tests { use super::*; use crate::identity::Id; use crate::node::Handle; - use crate::node::{Node, NodeId}; + use crate::node::{Alias, Node, NodeId}; use crate::service::tracking::Scope; use crate::test; @@ -308,7 +231,7 @@ mod tests { let tmp = tempfile::tempdir().unwrap(); let handle = test::handle::Handle::default(); let socket = tmp.path().join("alice.sock"); - let projs = test::arbitrary::set::(1..3); + let rids = test::arbitrary::set::(1..3); let listener = UnixListener::bind(&socket).unwrap(); thread::spawn({ @@ -317,7 +240,7 @@ mod tests { move || listen(listener, handle) }); - for proj in &projs { + for rid in &rids { let stream = loop { if let Ok(stream) = UnixStream::connect(&socket) { break stream; @@ -326,7 +249,10 @@ mod tests { writeln!( &stream, "{}", - json::to_string(&Command::new(CommandName::AnnounceRefs, [proj])).unwrap() + json::to_string(&Command::AnnounceRefs { + rid: rid.to_owned() + }) + .unwrap() ) .unwrap(); @@ -336,8 +262,8 @@ mod tests { assert_eq!(line, json::json!({ "status": "ok" }).to_string()); } - for proj in &projs { - assert!(handle.updates.lock().unwrap().contains(proj)); + for rid in &rids { + assert!(handle.updates.lock().unwrap().contains(rid)); } } diff --git a/radicle-node/src/runtime/handle.rs b/radicle-node/src/runtime/handle.rs index 830417a1..a3d22e93 100644 --- a/radicle-node/src/runtime/handle.rs +++ b/radicle-node/src/runtime/handle.rs @@ -245,7 +245,7 @@ impl radicle::node::Handle for Handle { // control thread gracefully. Since the control thread may have called this function, // the control socket may already be disconnected. Ignore errors. UnixStream::connect(self.home.socket()) - .and_then(|sock| Command::SHUTDOWN.to_writer(sock)) + .and_then(|sock| Command::Shutdown.to_writer(sock)) .ok(); self.controller diff --git a/radicle/src/node.rs b/radicle/src/node.rs index d91636ba..df7a8af0 100644 --- a/radicle/src/node.rs +++ b/radicle/src/node.rs @@ -276,80 +276,65 @@ impl From for Address { } /// Command name. -#[derive(Debug, Copy, Clone, Serialize, Deserialize)] -#[serde(rename_all = "kebab-case")] -pub enum CommandName { +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(rename_all = "camelCase", tag = "type")] +pub enum Command { /// Announce repository references for given repository to peers. - AnnounceRefs, + #[serde(rename_all = "camelCase")] + AnnounceRefs { rid: Id }, + /// Announce local repositories to peers. + #[serde(rename_all = "camelCase")] AnnounceInventory, + /// Sync local inventory with node. SyncInventory, + /// Connect to node with the given address. - Connect, + #[serde(rename_all = "camelCase")] + Connect { addr: config::ConnectAddress }, + /// Lookup seeds for the given repository in the routing table. - Seeds, + #[serde(rename_all = "camelCase")] + Seeds { rid: Id }, + /// Get the current peer sessions. Sessions, + /// Fetch the given repository from the network. - Fetch, + #[serde(rename_all = "camelCase")] + Fetch { rid: Id, nid: NodeId }, + /// Track the given repository. - TrackRepo, + #[serde(rename_all = "camelCase")] + TrackRepo { rid: Id, scope: tracking::Scope }, + /// Untrack the given repository. - UntrackRepo, + #[serde(rename_all = "camelCase")] + UntrackRepo { rid: Id }, + /// Track the given node. - TrackNode, + #[serde(rename_all = "camelCase")] + TrackNode { nid: NodeId, alias: Option }, + /// Untrack the given node. - UntrackNode, + #[serde(rename_all = "camelCase")] + UntrackNode { nid: NodeId }, + /// Get the node's status. Status, + /// Get the node's NID. NodeId, + /// Shutdown the node. Shutdown, + /// Subscribe to events. Subscribe, } -impl fmt::Display for CommandName { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - // SAFETY: The enum can always be converted to a value. - #[allow(clippy::unwrap_used)] - let val = json::to_value(self).unwrap(); - // SAFETY: The value is always a string. - #[allow(clippy::unwrap_used)] - let s = val.as_str().unwrap(); - - write!(f, "{s}") - } -} - -/// Commands sent to the node via the control socket. -#[derive(Debug, Serialize, Deserialize)] -pub struct Command { - /// Command name. - #[serde(rename = "cmd")] - pub name: CommandName, - /// Command arguments. - #[serde(rename = "args")] - pub args: Vec, -} - impl Command { - /// Shutdown command. - pub const SHUTDOWN: Self = Self { - name: CommandName::Shutdown, - args: vec![], - }; - - /// Create a new command. - pub fn new(name: CommandName, args: impl IntoIterator) -> Self { - Self { - name, - args: args.into_iter().map(|a| a.to_string()).collect(), - } - } - /// Write this command to a stream, including a terminating LF character. pub fn to_writer(&self, mut w: impl io::Write) -> io::Result<()> { json::to_writer(&mut w, self).map_err(|_| io::ErrorKind::InvalidInput)?; @@ -543,8 +528,8 @@ pub enum Error { Call(#[from] CallError), #[error("node: {0}")] Node(String), - #[error("received empty response for `{cmd}` command")] - EmptyResponse { cmd: CommandName }, + #[error("received empty response for command")] + EmptyResponse, } impl Error { @@ -559,9 +544,8 @@ impl Error { pub enum CallError { #[error("i/o: {0}")] Io(#[from] io::Error), - #[error("received invalid json in response for `{cmd}` command: '{response}': {error}")] + #[error("received invalid json in response to command: '{response}': {error}")] InvalidJson { - cmd: CommandName, response: String, error: json::Error, }, @@ -628,21 +612,19 @@ impl Node { } /// Call a command on the node. - pub fn call( + pub fn call( &self, - name: CommandName, - args: impl IntoIterator, + cmd: Command, timeout: time::Duration, ) -> Result>, io::Error> { let stream = UnixStream::connect(&self.socket)?; - Command::new(name, args).to_writer(&stream)?; + cmd.to_writer(&stream)?; stream.set_read_timeout(Some(timeout))?; Ok(BufReader::new(stream).lines().map(move |l| { let l = l?; let v = json::from_str(&l).map_err(|e| CallError::InvalidJson { - cmd: name, response: l, error: e, })?; @@ -702,16 +684,14 @@ impl Handle for Node { type Error = Error; fn nid(&self) -> Result { - self.call::<&str, NodeId>(CommandName::NodeId, [], DEFAULT_TIMEOUT)? + self.call::(Command::NodeId, DEFAULT_TIMEOUT)? .next() - .ok_or(Error::EmptyResponse { - cmd: CommandName::NodeId, - })? + .ok_or(Error::EmptyResponse)? .map_err(Error::from) } fn is_running(&self) -> bool { - let Ok(mut lines) = self.call::<&str, CommandResult>(CommandName::Status, [], DEFAULT_TIMEOUT) else { + let Ok(mut lines) = self.call::(Command::Status, DEFAULT_TIMEOUT) else { return false; }; let Some(Ok(result)) = lines.next() else { @@ -721,115 +701,81 @@ impl Handle for Node { } fn connect(&mut self, nid: NodeId, addr: Address) -> Result<(), Error> { - self.call::<_, CommandResult>( - CommandName::Connect, - [nid.to_human(), addr.to_string()], + self.call::( + Command::Connect { + addr: (nid, addr).into(), + }, DEFAULT_TIMEOUT, )? .next() - .ok_or(Error::EmptyResponse { - cmd: CommandName::Connect, - })??; + .ok_or(Error::EmptyResponse)??; Ok(()) } - fn seeds(&mut self, id: Id) -> Result { + fn seeds(&mut self, rid: Id) -> Result { let seeds: Seeds = self - .call(CommandName::Seeds, [id.urn()], DEFAULT_TIMEOUT)? + .call(Command::Seeds { rid }, DEFAULT_TIMEOUT)? .next() - .ok_or(Error::EmptyResponse { - cmd: CommandName::Seeds, - })??; + .ok_or(Error::EmptyResponse)??; Ok(seeds) } - fn fetch(&mut self, id: Id, from: NodeId) -> Result { + fn fetch(&mut self, rid: Id, from: NodeId) -> Result { let result = self - .call( - CommandName::Fetch, - [id.urn(), from.to_human()], - DEFAULT_TIMEOUT, - )? + .call(Command::Fetch { rid, nid: from }, DEFAULT_TIMEOUT)? .next() - .ok_or(Error::EmptyResponse { - cmd: CommandName::Fetch, - })??; + .ok_or(Error::EmptyResponse)??; Ok(result) } - fn track_node(&mut self, id: NodeId, alias: Option) -> Result { - let id = id.to_human(); - let args = if let Some(alias) = alias.as_deref() { - vec![id.as_str(), alias] - } else { - vec![id.as_str()] - }; - - let mut line = self.call(CommandName::TrackNode, args, DEFAULT_TIMEOUT)?; - let response: CommandResult = line.next().ok_or(Error::EmptyResponse { - cmd: CommandName::TrackNode, - })??; + fn track_node(&mut self, nid: NodeId, alias: Option) -> Result { + let mut line = self.call(Command::TrackNode { nid, alias }, DEFAULT_TIMEOUT)?; + let response: CommandResult = line.next().ok_or(Error::EmptyResponse)??; response.into() } - fn track_repo(&mut self, id: Id, scope: tracking::Scope) -> Result { - let mut line = self.call( - CommandName::TrackRepo, - [id.urn(), scope.to_string()], - DEFAULT_TIMEOUT, - )?; - let response: CommandResult = line.next().ok_or(Error::EmptyResponse { - cmd: CommandName::TrackRepo, - })??; + fn track_repo(&mut self, rid: Id, scope: tracking::Scope) -> Result { + let mut line = self.call(Command::TrackRepo { rid, scope }, DEFAULT_TIMEOUT)?; + let response: CommandResult = line.next().ok_or(Error::EmptyResponse)??; response.into() } - fn untrack_node(&mut self, id: NodeId) -> Result { - let mut line = self.call(CommandName::UntrackNode, [id], DEFAULT_TIMEOUT)?; - let response: CommandResult = line.next().ok_or(Error::EmptyResponse { - cmd: CommandName::UntrackNode, - })??; + fn untrack_node(&mut self, nid: NodeId) -> Result { + let mut line = self.call(Command::UntrackNode { nid }, DEFAULT_TIMEOUT)?; + let response: CommandResult = line.next().ok_or(Error::EmptyResponse)??; response.into() } - fn untrack_repo(&mut self, id: Id) -> Result { - let mut line = self.call(CommandName::UntrackRepo, [id.urn()], DEFAULT_TIMEOUT)?; - let response: CommandResult = line.next().ok_or(Error::EmptyResponse { - cmd: CommandName::UntrackRepo, - })??; + fn untrack_repo(&mut self, rid: Id) -> Result { + let mut line = self.call(Command::UntrackRepo { rid }, DEFAULT_TIMEOUT)?; + let response: CommandResult = line.next().ok_or(Error::EmptyResponse {})??; response.into() } - fn announce_refs(&mut self, id: Id) -> Result<(), Error> { - for line in - self.call::<_, CommandResult>(CommandName::AnnounceRefs, [id.urn()], DEFAULT_TIMEOUT)? - { + fn announce_refs(&mut self, rid: Id) -> Result<(), Error> { + for line in self.call::(Command::AnnounceRefs { rid }, DEFAULT_TIMEOUT)? { line?; } Ok(()) } fn announce_inventory(&mut self) -> Result<(), Error> { - for line in - self.call::<&str, CommandResult>(CommandName::AnnounceInventory, [], DEFAULT_TIMEOUT)? - { + for line in self.call::(Command::AnnounceInventory, DEFAULT_TIMEOUT)? { line?; } Ok(()) } fn sync_inventory(&mut self) -> Result { - let mut line = self.call::<&str, _>(CommandName::SyncInventory, [], DEFAULT_TIMEOUT)?; - let response: CommandResult = line.next().ok_or(Error::EmptyResponse { - cmd: CommandName::SyncInventory, - })??; + let mut line = self.call(Command::SyncInventory, DEFAULT_TIMEOUT)?; + let response: CommandResult = line.next().ok_or(Error::EmptyResponse {})??; response.into() } @@ -838,7 +784,7 @@ impl Handle for Node { &self, timeout: time::Duration, ) -> Result>>, Error> { - let events = self.call::<&str, _>(CommandName::Subscribe, [], timeout)?; + let events = self.call(Command::Subscribe, timeout)?; Ok(Box::new(events.map(|e| { e.map_err(|err| match err { @@ -852,17 +798,15 @@ impl Handle for Node { fn sessions(&self) -> Result { let sessions = self - .call::<&str, Vec>(CommandName::Sessions, [], DEFAULT_TIMEOUT)? + .call::>(Command::Sessions, DEFAULT_TIMEOUT)? .next() - .ok_or(Error::EmptyResponse { - cmd: CommandName::Sessions, - })??; + .ok_or(Error::EmptyResponse {})??; Ok(sessions) } fn shutdown(self) -> Result<(), Error> { - for line in self.call::<&str, CommandResult>(CommandName::Shutdown, [], DEFAULT_TIMEOUT)? { + for line in self.call::(Command::Shutdown, DEFAULT_TIMEOUT)? { line?; } // Wait until the shutdown has completed. @@ -901,11 +845,6 @@ impl AliasStore for HashMap { mod test { use super::*; - #[test] - fn test_command_name_display() { - assert_eq!(CommandName::TrackNode.to_string(), "track-node"); - } - #[test] fn test_alias() { assert!(Alias::from_str("cloudhead").is_ok());