diff --git a/radicle-node/src/control.rs b/radicle-node/src/control.rs index 44c69516..382c02e3 100644 --- a/radicle-node/src/control.rs +++ b/radicle-node/src/control.rs @@ -5,15 +5,15 @@ 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}; use radicle::node::Handle; use serde_json as json; use crate::identity::Id; -use crate::node; -use crate::node::FetchResult; use crate::node::NodeId; +use crate::node::{Command, CommandName, CommandResult, FetchResult}; use crate::runtime; #[derive(thiserror::Error, Debug)] @@ -44,7 +44,7 @@ pub fn listen handle.shutdown().ok(); break; } - writeln!(stream, "error: {e}").ok(); + CommandResult::error(e).to_writer(&mut stream).ok(); stream.flush().ok(); stream.shutdown(net::Shutdown::Both).ok(); @@ -62,6 +62,8 @@ pub fn listen enum CommandError { #[error("invalid command argument `{0}`, {1}")] InvalidCommandArg(String, Box), + #[error("invalid command arguments `{0:?}`")] + InvalidCommandArgs(Vec), #[error("unknown command `{0}`")] UnknownCommand(String), #[error("serialization failed: {0}")] @@ -83,165 +85,110 @@ fn command>( let mut line = String::new(); reader.read_line(&mut line)?; + let input = line.trim_end(); - let cmd = line.trim_end(); + log::debug!(target: "control", "Received `{input}` on control socket"); + let cmd: Command = json::from_str(input)?; - log::debug!(target: "control", "Received `{cmd}` on control socket"); - - // TODO: refactor to include helper - match cmd.split_once(' ') { - Some(("fetch", args)) => { - if let Some((rid, node)) = args.split_once(' ') { - let rid: Id = rid - .parse() - .map_err(|e| CommandError::InvalidCommandArg(rid.to_owned(), Box::new(e)))?; - let node: NodeId = node - .parse() - .map_err(|e| CommandError::InvalidCommandArg(node.to_owned(), Box::new(e)))?; - - fetch(rid, node, LineWriter::new(stream), handle)?; - } + match cmd.name { + CommandName::Fetch => { + let (rid, nid): (Id, NodeId) = parse::args(cmd)?; + fetch(rid, nid, LineWriter::new(stream), handle)?; } - Some(("seeds", arg)) => { - let rid: Id = arg - .parse() - .map_err(|e| CommandError::InvalidCommandArg(arg.to_owned(), Box::new(e)))?; + CommandName::Seeds => { + let rid: Id = parse::arg(cmd)?; + let seeds = handle.seeds(rid)?; - for seed in handle.seeds(rid)? { - writeln!(writer, "{seed}")?; - } + json::to_writer(writer, &seeds)?; } - Some(("track-repo", arg)) => match arg.parse() { - Ok(id) => match handle.track_repo(id) { + CommandName::TrackRepo => { + let rid: Id = parse::arg(cmd)?; + + match handle.track_repo(rid) { Ok(updated) => { - if updated { - writeln!(writer, "{}", node::RESPONSE_OK)?; - } else { - writeln!(writer, "{}", node::RESPONSE_NOOP)?; - } + CommandResult::Okay { updated }.to_writer(writer)?; } Err(e) => { return Err(CommandError::Runtime(e)); } - }, - Err(err) => { - return Err(CommandError::InvalidCommandArg( - arg.to_owned(), - Box::new(err), - )); } - }, - Some(("untrack-repo", arg)) => match arg.parse() { - Ok(id) => match handle.untrack_repo(id) { + } + CommandName::UntrackRepo => { + let rid: Id = parse::arg(cmd)?; + + match handle.untrack_repo(rid) { Ok(updated) => { - if updated { - writeln!(writer, "{}", node::RESPONSE_OK)?; - } else { - writeln!(writer, "{}", node::RESPONSE_NOOP)?; - } + CommandResult::Okay { updated }.to_writer(writer)?; } Err(e) => { return Err(CommandError::Runtime(e)); } - }, - Err(err) => { - return Err(CommandError::InvalidCommandArg( - arg.to_owned(), - Box::new(err), - )); } - }, - Some(("track-node", args)) => { - let (peer, alias) = if let Some((peer, alias)) = args.split_once(' ') { - (peer, Some(alias.to_owned())) - } else { - (args, None) + } + 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)), }; - match peer.parse() { - Ok(id) => match handle.track_node(id, alias) { - Ok(updated) => { - if updated { - writeln!(writer, "{}", node::RESPONSE_OK)?; - } else { - writeln!(writer, "{}", node::RESPONSE_NOOP)?; - } - } - Err(e) => { - return Err(CommandError::Runtime(e)); - } - }, - Err(err) => { - return Err(CommandError::InvalidCommandArg( - args.to_owned(), - Box::new(err), - )); - } - } - } - Some(("untrack-node", arg)) => match arg.parse() { - Ok(id) => match handle.untrack_node(id) { + let nid = node + .parse() + .map_err(|e| CommandError::InvalidCommandArg(node.to_owned(), Box::new(e)))?; + + match handle.track_node(nid, alias) { Ok(updated) => { - if updated { - writeln!(writer, "{}", node::RESPONSE_OK)?; - } else { - writeln!(writer, "{}", node::RESPONSE_NOOP)?; - } + CommandResult::Okay { updated }.to_writer(writer)?; } Err(e) => { return Err(CommandError::Runtime(e)); } - }, - Err(err) => { - return Err(CommandError::InvalidCommandArg( - arg.to_owned(), - Box::new(err), - )); } - }, - Some(("announce-refs", arg)) => match arg.parse() { - Ok(id) => { - if let Err(e) = handle.announce_refs(id) { + } + 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)); } - writeln!(writer, "{}", node::RESPONSE_OK)?; } - Err(err) => { - return Err(CommandError::InvalidCommandArg( - arg.to_owned(), - Box::new(err), - )); - } - }, - Some((cmd, _)) => return Err(CommandError::UnknownCommand(cmd.to_owned())), + } + CommandName::AnnounceRefs => { + let rid: Id = parse::arg(cmd)?; - // Commands with no arguments. - None => match cmd { - "status" => { - writeln!(writer, "{}", node::RESPONSE_OK).ok(); + if let Err(e) = handle.announce_refs(rid) { + return Err(CommandError::Runtime(e)); } - "routing" => match handle.routing() { - Ok(c) => { - for (id, seed) in c.iter() { - writeln!(writer, "{id} {seed}",)?; - } + CommandResult::ok().to_writer(writer).ok(); + } + CommandName::Status => { + CommandResult::ok().to_writer(writer).ok(); + } + CommandName::Routing => match handle.routing() { + Ok(c) => { + for (id, seed) in c.iter() { + writeln!(writer, "{id} {seed}")?; } - Err(e) => return Err(CommandError::Runtime(e)), - }, - "inventory" => match handle.inventory() { - Ok(c) => { - for id in c.iter() { - writeln!(writer, "{id}")?; - } - } - Err(e) => return Err(CommandError::Runtime(e)), - }, - "shutdown" => { - return Err(CommandError::Shutdown); - } - _ => { - return Err(CommandError::UnknownCommand(line)); } + Err(e) => return Err(CommandError::Runtime(e)), }, + CommandName::Inventory => match handle.inventory() { + Ok(c) => { + for id in c.iter() { + writeln!(writer, "{id}")?; + } + } + Err(e) => return Err(CommandError::Runtime(e)), + }, + CommandName::Shutdown => { + return Err(CommandError::Shutdown); + } + _ => { + return Err(CommandError::UnknownCommand(line)); + } } Ok(()) } @@ -263,6 +210,45 @@ fn fetch(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::*; @@ -290,15 +276,22 @@ mod tests { }); for proj in &projs { - let mut buf = [0; 2]; - let mut stream = loop { + let stream = loop { if let Ok(stream) = UnixStream::connect(&socket) { break stream; } }; - writeln!(&stream, "announce-refs {proj}").unwrap(); - stream.read_exact(&mut buf).unwrap(); - assert_eq!(&buf, &[b'o', b'k']); + writeln!( + &stream, + "{}", + json::to_string(&Command::new(CommandName::AnnounceRefs, [proj])).unwrap() + ) + .unwrap(); + + let stream = BufReader::new(stream); + let line = stream.lines().next().unwrap().unwrap(); + + assert_eq!(line, json::json!({ "status": "ok" }).to_string()); } for proj in &projs { diff --git a/radicle-node/src/runtime/handle.rs b/radicle-node/src/runtime/handle.rs index 1d0a79b9..1013bd5a 100644 --- a/radicle-node/src/runtime/handle.rs +++ b/radicle-node/src/runtime/handle.rs @@ -1,5 +1,5 @@ use std::fmt; -use std::io::{self, Write}; +use std::io; use std::os::unix::net::UnixStream; use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::Arc; @@ -10,7 +10,7 @@ use thiserror::Error; use crate::crypto::Signer; use crate::identity::Id; -use crate::node::FetchResult; +use crate::node::{Command, FetchResult}; use crate::profile::Home; use crate::service; use crate::service::{CommandError, QueryState}; @@ -221,7 +221,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(|mut sock| sock.write_all(b"shutdown")) + .and_then(|sock| Command::SHUTDOWN.to_writer(sock)) .ok(); self.controller.shutdown().map_err(|_| Error::NotConnected) diff --git a/radicle/src/node.rs b/radicle/src/node.rs index da69778e..41e23bdc 100644 --- a/radicle/src/node.rs +++ b/radicle/src/node.rs @@ -1,14 +1,14 @@ mod features; -use std::io::{BufRead, BufReader, Write}; +use std::io::{BufRead, BufReader}; use std::os::unix::net::UnixStream; use std::path::{Path, PathBuf}; -use std::str::FromStr; -use std::{io, net}; +use std::{fmt, io, net}; use amplify::WrapperMut; use crossbeam_channel as chan; use cyphernet::addr::{HostName, NetAddr}; +use serde::de::DeserializeOwned; use serde::{Deserialize, Serialize}; use serde_json as json; @@ -22,10 +22,58 @@ pub use features::Features; pub const DEFAULT_SOCKET_NAME: &str = "radicle.sock"; /// Default radicle protocol port. pub const DEFAULT_PORT: u16 = 8776; -/// Response on node socket indicating that a command was carried out successfully. -pub const RESPONSE_OK: &str = "ok"; -/// Response on node socket indicating that a command had no effect. -pub const RESPONSE_NOOP: &str = "noop"; + +/// Result of a command, on the node control socket. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(tag = "status")] +pub enum CommandResult { + /// Response on node socket indicating that a command was carried out successfully. + #[serde(rename = "ok")] + Okay { + /// Whether the command had any effect. + #[serde(default, skip_serializing_if = "crate::serde_ext::is_default")] + updated: bool, + }, + /// Response on node socket indicating that an error occured. + Error { + /// The reason for the error. + reason: String, + }, +} + +impl CommandResult { + /// Create an "updated" response. + pub fn updated() -> Self { + Self::Okay { updated: true } + } + + /// Create an "ok" response. + pub fn ok() -> Self { + Self::Okay { updated: false } + } + + /// Create an error result. + pub fn error(err: impl std::error::Error) -> Self { + Self::Error { + reason: err.to_string(), + } + } + + /// Write this command result 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)?; + w.write_all(b"\n") + } +} + +impl From for Result { + fn from(value: CommandResult) -> Self { + match value { + CommandResult::Okay { updated } => Ok(updated), + CommandResult::Error { reason } => Err(Error::Node(reason)), + } + } +} /// Peer public protocol address. #[derive(Wrapper, WrapperMut, Clone, Eq, PartialEq, Debug, From)] @@ -54,6 +102,82 @@ impl From for Address { } } +/// Command name. +#[derive(Debug, Copy, Clone, Serialize, Deserialize)] +#[serde(rename_all = "kebab-case")] +pub enum CommandName { + /// Announce repository references for given repository to peers. + AnnounceRefs, + /// Connect to node with the given address. + Connect, + /// Lookup seeds for the given repository in the routing table. + Seeds, + /// Fetch the given repository from the network. + Fetch, + /// Track the given repository. + TrackRepo, + /// Untrack the given repository. + UntrackRepo, + /// Track the given node. + TrackNode, + /// Untrack the given node. + UntrackNode, + /// Get the node's inventory. + Inventory, + /// Get the node's routing table. + Routing, + /// Get the node's status. + Status, + /// Shutdown the node. + Shutdown, +} + +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)?; + w.write_all(b"\n") + } +} + #[derive(Debug, Serialize, Deserialize)] #[serde(tag = "status", rename_all = "kebab-case")] pub enum FetchResult { @@ -85,20 +209,30 @@ impl From, S>> for FetchResult { } } +/// Error returned by [`Handle`] functions. #[derive(thiserror::Error, Debug)] pub enum Error { #[error("failed to connect to node: {0}")] Connect(#[from] io::Error), - #[error("received invalid response for `{cmd}` command: '{response}'")] - InvalidResponse { cmd: &'static str, response: String }, + #[error("failed to call node: {0}")] + Call(#[from] CallError), + #[error("node: {0}")] + Node(String), + #[error("received empty response for `{cmd}` command")] + EmptyResponse { cmd: CommandName }, +} + +/// Error returned by [`Node::call`] iterator. +#[derive(thiserror::Error, Debug)] +pub enum CallError { + #[error("i/o: {0}")] + Io(#[from] io::Error), #[error("received invalid json in response for `{cmd}` command: '{response}': {error}")] InvalidJson { - cmd: &'static str, + cmd: CommandName, response: String, error: json::Error, }, - #[error("received empty response for `{cmd}` command")] - EmptyResponse { cmd: &'static str }, } /// A handle to send commands to the node or request information. @@ -157,24 +291,24 @@ impl Node { } /// Call a command on the node. - pub fn call( + pub fn call( &self, - cmd: &str, - args: &[A], - ) -> Result>, io::Error> { + name: CommandName, + args: impl IntoIterator, + ) -> Result>, io::Error> { let stream = UnixStream::connect(&self.socket)?; - let args = args - .iter() - .map(ToString::to_string) - .collect::>() - .join(" "); + Command::new(name, args).to_writer(&stream)?; - if args.is_empty() { - writeln!(&stream, "{cmd}")?; - } else { - writeln!(&stream, "{cmd} {args}")?; - } - Ok(BufReader::new(stream).lines()) + 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, + })?; + + Ok(v) + })) } } @@ -184,13 +318,13 @@ impl Handle for Node { type FetchResult = FetchResult; fn is_running(&self) -> bool { - let Ok(mut lines) = self.call::<&str>("status", &[]) else { + let Ok(mut lines) = self.call::<&str, CommandResult>(CommandName::Status, []) else { return false; }; - let Some(Ok(line)) = lines.next() else { + let Some(Ok(result)) = lines.next() else { return false; }; - line == RESPONSE_OK + matches!(result, CommandResult::Okay { .. }) } fn connect(&mut self, _node: NodeId, _addr: Address) -> Result<(), Error> { @@ -198,113 +332,73 @@ impl Handle for Node { } fn seeds(&mut self, id: Id) -> Result, Error> { - self.call("seeds", &[id.urn()])? - .map(|line| { - let line = line?; - let node = NodeId::from_str(&line).map_err(|_| Error::InvalidResponse { - cmd: "seeds", - response: line, - })?; - Ok(node) - }) - .collect() + let seeds: Vec = + self.call(CommandName::Seeds, [id.urn()])? + .next() + .ok_or(Error::EmptyResponse { + cmd: CommandName::Seeds, + })??; + + Ok(seeds) } fn fetch(&mut self, id: Id, from: NodeId) -> Result { let result = self - .call("fetch", &[id.urn(), from.to_human()])? + .call(CommandName::Fetch, [id.urn(), from.to_human()])? .next() - .ok_or(Error::EmptyResponse { cmd: "fetch" })??; - let lookup = json::from_str(&result).map_err(|e| Error::InvalidJson { - cmd: "fetch", - response: result, - error: e, - })?; + .ok_or(Error::EmptyResponse { + cmd: CommandName::Fetch, + })??; - Ok(lookup) + Ok(result) } fn track_node(&mut self, id: NodeId, alias: Option) -> Result { let id = id.to_human(); - let mut line = if let Some(alias) = alias.as_deref() { - self.call("track-node", &[id.as_str(), alias]) + let args = if let Some(alias) = alias.as_deref() { + vec![id.as_str(), alias] } else { - self.call("track-node", &[id.as_str()]) - }?; - let line = line - .next() - .ok_or(Error::EmptyResponse { cmd: "track-node" })??; + vec![id.as_str()] + }; - log::debug!("node: {}", line); + let mut line = self.call(CommandName::TrackNode, args)?; + let response: CommandResult = line.next().ok_or(Error::EmptyResponse { + cmd: CommandName::TrackNode, + })??; - match line.as_str() { - RESPONSE_OK => Ok(true), - RESPONSE_NOOP => Ok(false), - _ => Err(Error::InvalidResponse { - cmd: "track-node", - response: line, - }), - } + response.into() } fn track_repo(&mut self, id: Id) -> Result { - let mut line = self.call("track-repo", &[id.urn()])?; - let line = line - .next() - .ok_or(Error::EmptyResponse { cmd: "track-repo" })??; + let mut line = self.call(CommandName::TrackRepo, [id.urn()])?; + let response: CommandResult = line.next().ok_or(Error::EmptyResponse { + cmd: CommandName::TrackRepo, + })??; - log::debug!("node: {}", line); - - match line.as_str() { - RESPONSE_OK => Ok(true), - RESPONSE_NOOP => Ok(false), - _ => Err(Error::InvalidResponse { - cmd: "track-repo", - response: line, - }), - } + response.into() } fn untrack_node(&mut self, id: NodeId) -> Result { - let mut line = self.call("untrack-node", &[id])?; - let line = line.next().ok_or(Error::EmptyResponse { - cmd: "untrack-node", + let mut line = self.call(CommandName::UntrackNode, [id])?; + let response: CommandResult = line.next().ok_or(Error::EmptyResponse { + cmd: CommandName::UntrackNode, })??; - log::debug!("node: {}", line); - - match line.as_str() { - RESPONSE_OK => Ok(true), - RESPONSE_NOOP => Ok(false), - _ => Err(Error::InvalidResponse { - cmd: "untrack-node", - response: line, - }), - } + response.into() } fn untrack_repo(&mut self, id: Id) -> Result { - let mut line = self.call("untrack-repo", &[id.urn()])?; - let line = line.next().ok_or(Error::EmptyResponse { - cmd: "untrack-repo", + let mut line = self.call(CommandName::UntrackRepo, [id.urn()])?; + let response: CommandResult = line.next().ok_or(Error::EmptyResponse { + cmd: CommandName::UntrackRepo, })??; - log::debug!("node: {}", line); - - match line.as_str() { - RESPONSE_OK => Ok(true), - RESPONSE_NOOP => Ok(false), - _ => Err(Error::InvalidResponse { - cmd: "untrack-repo", - response: line, - }), - } + response.into() } fn announce_refs(&mut self, id: Id) -> Result<(), Error> { - for line in self.call("announce-refs", &[id.urn()])? { - let line = line?; - log::debug!("node: {}", line); + for line in self.call(CommandName::AnnounceRefs, [id.urn()])? { + line?; } Ok(()) } @@ -325,3 +419,13 @@ impl Handle for Node { todo!(); } } + +#[cfg(test)] +mod test { + use super::*; + + #[test] + fn test_command_name_display() { + assert_eq!(CommandName::TrackNode.to_string(), "track-node"); + } +} diff --git a/radicle/src/serde_ext.rs b/radicle/src/serde_ext.rs index 16326eff..62c85f64 100644 --- a/radicle/src/serde_ext.rs +++ b/radicle/src/serde_ext.rs @@ -23,3 +23,8 @@ pub mod string { .map_err(de::Error::custom) } } + +/// Return true if the given value is the default for that type. +pub fn is_default(t: &T) -> bool { + t == &T::default() +}