From b6751855388454acf691f47e705f5032fa6ab15f Mon Sep 17 00:00:00 2001 From: cloudhead Date: Sat, 11 Nov 2023 21:00:13 +0100 Subject: [PATCH] node: Improve control socket API We improve the API so that error messages can be sent back and are always expected as potential responses to commands. This means that `CommandResult` is always the type that is sent for commands. This includes some breaking changes, for example the default "success" message is simply `{}` instead of `{"status":"ok"}`. --- radicle-node/src/control.rs | 36 ++++--- radicle/src/node.rs | 164 +++++++++++++++++++++--------- radicle/src/node/address/types.rs | 30 +++++- 3 files changed, 158 insertions(+), 72 deletions(-) diff --git a/radicle-node/src/control.rs b/radicle-node/src/control.rs index e0a6bae7..6a069061 100644 --- a/radicle-node/src/control.rs +++ b/radicle-node/src/control.rs @@ -108,45 +108,45 @@ where Command::Config => { let config = handle.config()?; - json::to_writer(writer, &config)?; + CommandResult::Okay(config).to_writer(writer)?; } Command::Seeds { rid } => { let seeds = handle.seeds(rid)?; - json::to_writer(writer, &seeds)?; + CommandResult::Okay(seeds).to_writer(writer)?; } Command::Sessions => { let sessions = handle.sessions()?; - json::to_writer(writer, &sessions)?; + CommandResult::Okay(sessions).to_writer(writer)?; } Command::TrackRepo { rid, scope } => match handle.track_repo(rid, scope) { - Ok(updated) => { - CommandResult::Okay { updated }.to_writer(writer)?; + Ok(result) => { + CommandResult::updated(result).to_writer(writer)?; } Err(e) => { return Err(CommandError::Runtime(e)); } }, Command::UntrackRepo { rid } => match handle.untrack_repo(rid) { - Ok(updated) => { - CommandResult::Okay { updated }.to_writer(writer)?; + Ok(result) => { + CommandResult::updated(result).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)?; + Ok(result) => { + CommandResult::updated(result).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)?; + Ok(result) => { + CommandResult::updated(result).to_writer(writer)?; } Err(e) => { return Err(CommandError::Runtime(e)); @@ -165,8 +165,8 @@ where CommandResult::ok().to_writer(writer).ok(); } Command::SyncInventory => match handle.sync_inventory() { - Ok(updated) => { - CommandResult::Okay { updated }.to_writer(writer)?; + Ok(result) => { + CommandResult::updated(result).to_writer(writer)?; } Err(e) => { return Err(CommandError::Runtime(e)); @@ -176,19 +176,17 @@ where Ok(events) => { for e in events { let event = e?; - let event = serde_json::to_string(&event)?; - - writeln!(&mut writer, "{event}")?; + CommandResult::Okay(event).to_writer(&mut writer)?; } } - Err(e) => log::error!(target: "control", "Error subscribing to events: {e}"), + Err(e) => return Err(CommandError::Runtime(e)), }, Command::Status => { CommandResult::ok().to_writer(writer).ok(); } Command::NodeId => match handle.nid() { Ok(nid) => { - writeln!(writer, "{nid}")?; + CommandResult::Okay(nid).to_writer(writer)?; } Err(e) => return Err(CommandError::Runtime(e)), }, @@ -267,7 +265,7 @@ mod tests { let stream = BufReader::new(stream); let line = stream.lines().next().unwrap().unwrap(); - assert_eq!(line, json::json!({ "status": "ok" }).to_string()); + assert_eq!(line, json::json!({}).to_string()); } for rid in &rids { diff --git a/radicle/src/node.rs b/radicle/src/node.rs index 83097476..e224e3a1 100644 --- a/radicle/src/node.rs +++ b/radicle/src/node.rs @@ -69,14 +69,17 @@ pub enum PingState { #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] #[allow(clippy::large_enum_variant)] +#[serde(rename_all = "camelCase")] pub enum State { /// Initial state for outgoing connections. Initial, /// Connection attempted successfully. Attempted, /// Initial state after handshake protocol hand-off. + #[serde(rename_all = "camelCase")] Connected { /// Connected since this time. + #[serde(with = "crate::serde_ext::localtime::time")] since: LocalTime, /// Ping state. #[serde(skip)] @@ -85,10 +88,13 @@ pub enum State { fetching: HashSet, }, /// When a peer is disconnected. + #[serde(rename_all = "camelCase")] Disconnected { /// Since when has this peer been disconnected. + #[serde(with = "crate::serde_ext::localtime::time")] since: LocalTime, /// When to retry the connection. + #[serde(with = "crate::serde_ext::localtime::time")] retry_at: LocalTime, }, } @@ -214,40 +220,48 @@ pub struct ConnectOptions { /// Result of a command, on the node control socket. #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] -#[serde(tag = "status")] -pub enum CommandResult { +#[serde(untagged)] +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, - }, + Okay(T), /// Response on node socket indicating that an error occured. Error { /// The reason for the error. + #[serde(rename = "error")] reason: String, }, } -impl CommandResult { +/// A success response. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct Success { + /// Whether something was updated. + #[serde(default, skip_serializing_if = "crate::serde_ext::is_default")] + updated: bool, +} + +impl CommandResult { /// Create an "updated" response. - pub fn updated() -> Self { - Self::Okay { updated: true } + pub fn updated(updated: bool) -> Self { + Self::Okay(Success { updated }) } /// Create an "ok" response. pub fn ok() -> Self { - Self::Okay { updated: false } + Self::Okay(Success { updated: false }) } +} +impl CommandResult<()> { /// Create an error result. pub fn error(err: impl std::error::Error) -> Self { Self::Error { reason: err.to_string(), } } +} +impl CommandResult { /// 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)?; @@ -255,15 +269,6 @@ impl CommandResult { } } -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, Hash, From, Serialize, Deserialize)] #[wrapper(Deref, Display, FromStr)] @@ -438,6 +443,7 @@ impl Seed { #[derive(Clone, Debug, serde::Serialize, serde::Deserialize)] /// Represents a set of seeds with associated metadata. Uses an RNG /// underneath, so every iteration returns a different ordering. +#[serde(into = "Vec", from = "Vec")] pub struct Seeds(address::AddressBook); impl Seeds { @@ -480,6 +486,20 @@ impl Seeds { } } +impl From for Vec { + fn from(seeds: Seeds) -> Vec { + seeds.0.into_shuffled().map(|(_, v)| v).collect() + } +} + +impl From> for Seeds { + fn from(other: Vec) -> Seeds { + Seeds(address::AddressBook::from_iter( + other.into_iter().map(|s| (s.nid, s)), + )) + } +} + /// Announcement result returned by [`Node::announce`]. #[derive(Debug)] pub struct AnnounceResult { @@ -631,7 +651,9 @@ impl Error { pub enum CallError { #[error("i/o: {0}")] Io(#[from] io::Error), - #[error("received invalid json in response to command: '{response}': {error}")] + #[error("command error: {reason}")] + Command { reason: String }, + #[error("received invalid json `{response}` in response to command: {error}")] InvalidJson { response: String, error: json::Error, @@ -739,12 +761,16 @@ impl Node { e } })?; - let v = json::from_str(&l).map_err(|e| CallError::InvalidJson { - response: l, - error: e, - })?; + let result: CommandResult = + json::from_str(&l).map_err(|e| CallError::InvalidJson { + response: l.clone(), + error: e, + })?; - Ok(v) + match result { + CommandResult::Okay(result) => Ok(result), + CommandResult::Error { reason } => Err(CallError::Command { reason }), + } })) } @@ -806,13 +832,13 @@ impl Handle for Node { } fn is_running(&self) -> bool { - let Ok(mut lines) = self.call::(Command::Status, DEFAULT_TIMEOUT) else { + let Ok(mut lines) = self.call::(Command::Status, DEFAULT_TIMEOUT) else { return false; }; - let Some(Ok(result)) = lines.next() else { + let Some(Ok(_)) = lines.next() else { return false; }; - matches!(result, CommandResult::Okay { .. }) + true } fn config(&self) -> Result { @@ -844,8 +870,8 @@ impl Handle for Node { } fn seeds(&mut self, rid: Id) -> Result { - let seeds: Seeds = self - .call(Command::Seeds { rid }, DEFAULT_TIMEOUT)? + let seeds = self + .call::(Command::Seeds { rid }, DEFAULT_TIMEOUT)? .next() .ok_or(Error::EmptyResponse)??; @@ -874,52 +900,52 @@ impl Handle for Node { } 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)??; + let mut line = self.call::(Command::TrackNode { nid, alias }, DEFAULT_TIMEOUT)?; + let response = line.next().ok_or(Error::EmptyResponse)??; - response.into() + Ok(response.updated) } 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)??; + let mut line = self.call::(Command::TrackRepo { rid, scope }, DEFAULT_TIMEOUT)?; + let response = line.next().ok_or(Error::EmptyResponse)??; - response.into() + Ok(response.updated) } 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)??; + let mut line = self.call::(Command::UntrackNode { nid }, DEFAULT_TIMEOUT)?; + let response = line.next().ok_or(Error::EmptyResponse)??; - response.into() + Ok(response.updated) } 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 {})??; + let mut line = self.call::(Command::UntrackRepo { rid }, DEFAULT_TIMEOUT)?; + let response = line.next().ok_or(Error::EmptyResponse {})??; - response.into() + Ok(response.updated) } fn announce_refs(&mut self, rid: Id) -> Result<(), Error> { - for line in self.call::(Command::AnnounceRefs { rid }, DEFAULT_TIMEOUT)? { + for line in self.call::(Command::AnnounceRefs { rid }, DEFAULT_TIMEOUT)? { line?; } Ok(()) } fn announce_inventory(&mut self) -> Result<(), Error> { - for line in self.call::(Command::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(Command::SyncInventory, DEFAULT_TIMEOUT)?; - let response: CommandResult = line.next().ok_or(Error::EmptyResponse {})??; + let mut line = self.call::(Command::SyncInventory, DEFAULT_TIMEOUT)?; + let response = line.next().ok_or(Error::EmptyResponse {})??; - response.into() + Ok(response.updated) } fn subscribe( @@ -934,6 +960,7 @@ impl Handle for Node { CallError::InvalidJson { .. } => { io::Error::new(io::ErrorKind::InvalidInput, err.to_string()) } + CallError::Command { reason } => io::Error::new(io::ErrorKind::Other, reason), }) }))) } @@ -948,7 +975,7 @@ impl Handle for Node { } fn shutdown(self) -> Result<(), Error> { - for line in self.call::(Command::Shutdown, DEFAULT_TIMEOUT)? { + for line in self.call::(Command::Shutdown, DEFAULT_TIMEOUT)? { line?; } // Wait until the shutdown has completed. @@ -986,6 +1013,7 @@ impl AliasStore for HashMap { #[cfg(test)] mod test { use super::*; + use crate::assert_matches; #[test] fn test_alias() { @@ -1001,4 +1029,40 @@ mod test { assert!(Alias::from_str("cloud head").is_err()); assert!(Alias::from_str("cloudhead\n").is_err()); } + + #[test] + fn test_command_result() { + #[derive(Debug, PartialEq, Eq, Serialize, Deserialize)] + struct Test { + value: u32, + } + + assert_eq!(json::to_string(&CommandResult::Okay(true)).unwrap(), "true"); + assert_eq!( + json::to_string(&CommandResult::Okay(Test { value: 42 })).unwrap(), + "{\"value\":42}" + ); + assert_eq!( + json::from_str::>("{\"value\":42}").unwrap(), + CommandResult::Okay(Test { value: 42 }) + ); + assert_eq!(json::to_string(&CommandResult::ok()).unwrap(), "{}"); + assert_eq!( + json::to_string(&CommandResult::updated(true)).unwrap(), + "{\"updated\":true}" + ); + assert_eq!( + json::to_string(&CommandResult::error(io::Error::from( + io::ErrorKind::NotFound + ))) + .unwrap(), + "{\"error\":\"entity not found\"}" + ); + assert_matches!( + json::from_str::>( + r#"[{"nid":"z6Mkux1aUQD2voWWukVb5nNUR7thrHveQG4pDQua8nVhib7Z","addrs":[],"state":{"connected":{"since":1699636852107,"fetching":[]}}}]"# + ), + Ok(CommandResult::Okay(_)) + ); + } } diff --git a/radicle/src/node/address/types.rs b/radicle/src/node/address/types.rs index d4f1f564..5cbdf26a 100644 --- a/radicle/src/node/address/types.rs +++ b/radicle/src/node/address/types.rs @@ -6,9 +6,9 @@ use localtime::LocalTime; use nonempty::NonEmpty; use crate::collections::RandomMap; -use crate::node; use crate::node::{Address, Alias}; use crate::prelude::Timestamp; +use crate::{node, profile}; /// A map with the ability to randomly select values. #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)] @@ -59,8 +59,8 @@ impl AddressBook { } } -impl AddressBook { - /// Return a shuffled iterator over the keys. +impl AddressBook { + /// Return a shuffled iterator. pub fn shuffled(&self) -> std::vec::IntoIter<(&K, &V)> { let mut items = self.inner.iter().collect::>(); items.sort_by_key(|(k, _)| *k); @@ -69,12 +69,36 @@ impl AddressBook { items.into_iter() } + /// Turn this object into a shuffled iterator. + pub fn into_shuffled(self) -> impl Iterator { + let mut items = self.inner.into_iter().collect::>(); + items.sort_by_key(|(k, _)| *k); + self.rng.borrow_mut().shuffle(&mut items); + + items.into_iter() + } + /// Cycle through the keys at random. The random cycle repeats ad-infintum. pub fn cycle(&self) -> impl Iterator { self.shuffled().map(|(k, _)| k).cycle() } } +impl FromIterator<(K, V)> for AddressBook { + fn from_iter>(iter: T) -> Self { + let rng = profile::env::rng(); + let mut inner = RandomMap::with_hasher(rng.clone().into()); + + for (k, v) in iter { + inner.insert(k, v); + } + Self { + inner, + rng: RefCell::new(rng), + } + } +} + impl Deref for AddressBook { type Target = RandomMap;