mab122
/
radicle-heartwood-lfs
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Make `track` functions return correctly 

Previously we always returned `true`. Now we return
what the node actually returns.

Signed-off-by: Alexis Sellier <alexis@radicle.xyz>
This commit is contained in:
Alexis Sellier 2022-11-09 13:12:48 +01:00
parent eb666265be
commit f05a040be6
No known key found for this signature in database
12 changed files with 133 additions and 50 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
Cargo.lock generated
View File

@ -1481,6 +1481,7 @@ dependencies = [
"radicle-cob",
"radicle-crypto",
"radicle-git-ext",
"radicle-node",
"radicle-ssh",
"serde",
"serde_json",

4
radicle-node/Cargo.toml
View File

@ -5,6 +5,9 @@ version = "0.2.0"
authors = ["Alexis Sellier <alexis@radicle.xyz>"]
edition = "2021"
[features]
test = ["radicle/test", "radicle-crypto/test", "quickcheck"]
[dependencies]
anyhow = { version = "1" }
bloomy = { version = "1.2" }
@ -19,6 +22,7 @@ log = { version = "0.4.17", features = ["std"] }
nakamoto-net = { version = "0.3.0" }
nakamoto-net-poll = { version = "0.3.0" }
nonempty = { version = "0.8.0", features = ["serialize"] }
quickcheck = { version = "1", default-features = false, optional = true }
sqlite = { version = "0.28.1" }
sqlite3-src = { version = "0.4.0", features = ["bundled"] } # Ensures static linking
scrypt = { version = "0.10.0", default-features = false }

16
radicle-node/src/client/handle.rs
View File

@ -60,25 +60,25 @@ impl<W: Waker> traits::Handle for Handle<W> {
self.listening.recv().map_err(Error::from)
}
fn fetch(&self, id: Id) -> Result<FetchLookup, Error> {
fn fetch(&mut self, id: Id) -> Result<FetchLookup, Error> {
let (sender, receiver) = chan::bounded(1);
self.commands.send(service::Command::Fetch(id, sender))?;
receiver.recv().map_err(Error::from)
}
fn track(&self, id: Id) -> Result<bool, Error> {
fn track(&mut self, id: Id) -> Result<bool, Error> {
let (sender, receiver) = chan::bounded(1);
self.commands.send(service::Command::Track(id, sender))?;
receiver.recv().map_err(Error::from)
}
fn untrack(&self, id: Id) -> Result<bool, Error> {
fn untrack(&mut self, id: Id) -> Result<bool, Error> {
let (sender, receiver) = chan::bounded(1);
self.commands.send(service::Command::Untrack(id, sender))?;
receiver.recv().map_err(Error::from)
}
fn announce_refs(&self, id: Id) -> Result<(), Error> {
fn announce_refs(&mut self, id: Id) -> Result<(), Error> {
self.command(service::Command::AnnounceRefs(id))
}
@ -143,14 +143,14 @@ pub mod traits {
/// Wait for the node's listening socket to be bound.
fn listening(&self) -> Result<net::SocketAddr, Error>;
/// Retrieve or update the project from network.
fn fetch(&self, id: Id) -> Result<FetchLookup, Error>;
fn fetch(&mut self, id: Id) -> Result<FetchLookup, Error>;
/// Start tracking the given project. Doesn't do anything if the project is already
/// tracked.
fn track(&self, id: Id) -> Result<bool, Error>;
fn track(&mut self, id: Id) -> Result<bool, Error>;
/// Untrack the given project and delete it from storage.
fn untrack(&self, id: Id) -> Result<bool, Error>;
fn untrack(&mut self, id: Id) -> Result<bool, Error>;
/// Notify the client that a project has been updated.
fn announce_refs(&self, id: Id) -> Result<(), Error>;
fn announce_refs(&mut self, id: Id) -> Result<(), Error>;
/// Send a command to the command channel, and wake up the event loop.
fn command(&self, cmd: service::Command) -> Result<(), Error>;
/// Ask the client to shutdown.

40
radicle-node/src/control.rs
View File

@ -10,6 +10,7 @@ use std::{fs, io, net};
use crate::client;
use crate::client::handle::traits::Handle;
use crate::identity::Id;
use crate::node;
use crate::service::FetchLookup;
use crate::service::FetchResult;
@ -22,7 +23,7 @@ pub enum Error {
}
/// Listen for commands on the control socket, and process them.
pub fn listen<P: AsRef<Path>, H: Handle>(path: P, handle: H) -> Result<(), Error> {
pub fn listen<P: AsRef<Path>, H: Handle>(path: P, mut handle: H) -> Result<(), Error> {
// Remove the socket file on startup before rebinding.
fs::remove_file(&path).ok();
fs::create_dir_all(
@ -38,7 +39,7 @@ pub fn listen<P: AsRef<Path>, H: Handle>(path: P, handle: H) -> Result<(), Error
for incoming in listener.incoming() {
match incoming {
Ok(mut stream) => {
if let Err(e) = drain(&stream, &handle) {
if let Err(e) = drain(&stream, &mut handle) {
log::error!("Received {} on control socket", e);
writeln!(stream, "error: {}", e).ok();
@ -68,8 +69,9 @@ enum DrainError {
Io(#[from] io::Error),
}
fn drain<H: Handle>(stream: &UnixStream, handle: &H) -> Result<(), DrainError> {
fn drain<H: Handle>(stream: &UnixStream, handle: &mut H) -> Result<(), DrainError> {
let mut reader = BufReader::new(stream);
let mut writer = LineWriter::new(stream);
// TODO: refactor to include helper
for line in reader.by_ref().lines().flatten() {
@ -83,8 +85,17 @@ fn drain<H: Handle>(stream: &UnixStream, handle: &H) -> Result<(), DrainError> {
}
Some(("track", arg)) => {
if let Ok(id) = arg.parse() {
if let Err(e) = handle.track(id) {
return Err(DrainError::Client(e));
match handle.track(id) {
Ok(updated) => {
if updated {
writeln!(writer, "{}", node::RESPONSE_OK)?;
} else {
writeln!(writer, "{}", node::RESPONSE_NOOP)?;
}
}
Err(e) => {
return Err(DrainError::Client(e));
}
}
} else {
return Err(DrainError::InvalidCommandArg(arg.to_owned()));
@ -92,8 +103,17 @@ fn drain<H: Handle>(stream: &UnixStream, handle: &H) -> Result<(), DrainError> {
}
Some(("untrack", arg)) => {
if let Ok(id) = arg.parse() {
if let Err(e) = handle.untrack(id) {
return Err(DrainError::Client(e));
match handle.untrack(id) {
Ok(updated) => {
if updated {
writeln!(writer, "{}", node::RESPONSE_OK)?;
} else {
writeln!(writer, "{}", node::RESPONSE_NOOP)?;
}
}
Err(e) => {
return Err(DrainError::Client(e));
}
}
} else {
return Err(DrainError::InvalidCommandArg(arg.to_owned()));
@ -114,8 +134,6 @@ fn drain<H: Handle>(stream: &UnixStream, handle: &H) -> Result<(), DrainError> {
None => match line.as_str() {
"routing" => match handle.routing() {
Ok(c) => {
let mut writer = LineWriter::new(stream);
for (id, seed) in c.iter() {
writeln!(writer, "{id} {seed}",)?;
}
@ -124,8 +142,6 @@ fn drain<H: Handle>(stream: &UnixStream, handle: &H) -> Result<(), DrainError> {
},
"inventory" => match handle.inventory() {
Ok(c) => {
let mut writer = LineWriter::new(stream);
for id in c.iter() {
writeln!(writer, "{id}")?;
}
@ -141,7 +157,7 @@ fn drain<H: Handle>(stream: &UnixStream, handle: &H) -> Result<(), DrainError> {
Ok(())
}
fn fetch<W: Write, H: Handle>(id: Id, mut writer: W, handle: &H) -> Result<(), DrainError> {
fn fetch<W: Write, H: Handle>(id: Id, mut writer: W, handle: &mut H) -> Result<(), DrainError> {
match handle.fetch(id) {
Err(e) => {
return Err(DrainError::Client(e));

2
radicle-node/src/lib.rs
View File

@ -6,7 +6,7 @@ pub mod decoder;
pub mod logger;
pub mod service;
pub mod sql;
#[cfg(test)]
#[cfg(any(test, feature = "test"))]
pub mod test;
pub mod transport;
pub mod wire;

2
radicle-node/src/service/reactor.rs
View File

@ -104,7 +104,7 @@ impl Reactor {
}
}
#[cfg(test)]
#[cfg(any(test, feature = "test"))]
pub(crate) fn outbox(&mut self) -> &mut VecDeque<Io> {
&mut self.io
}

15
radicle-node/src/test.rs
View File

@ -1,12 +1,9 @@
pub(crate) mod arbitrary;
pub(crate) mod gossip;
pub(crate) mod handle;
pub(crate) mod logger;
pub(crate) mod peer;
pub(crate) mod simulator;
pub(crate) mod tests;
pub mod arbitrary;
pub mod gossip;
pub mod handle;
pub mod logger;
pub mod peer;
pub mod simulator;
#[cfg(test)]
pub use radicle::assert_matches;
#[cfg(test)]
pub use radicle::test::*;

14
radicle-node/src/test/handle.rs
View File

@ -1,3 +1,4 @@
use std::collections::HashSet;
use std::sync::{Arc, Mutex};
use crossbeam_channel as chan;
@ -11,6 +12,7 @@ use crate::service::FetchLookup;
#[derive(Default, Clone)]
pub struct Handle {
pub updates: Arc<Mutex<Vec<Id>>>,
pub tracking: HashSet<Id>,
}
impl traits::Handle for Handle {
@ -18,19 +20,19 @@ impl traits::Handle for Handle {
unimplemented!()
}
fn fetch(&self, _id: Id) -> Result<FetchLookup, Error> {
fn fetch(&mut self, _id: Id) -> Result<FetchLookup, Error> {
Ok(FetchLookup::NotFound)
}
fn track(&self, _id: Id) -> Result<bool, Error> {
Ok(true)
fn track(&mut self, id: Id) -> Result<bool, Error> {
Ok(self.tracking.insert(id))
}
fn untrack(&self, _id: Id) -> Result<bool, Error> {
Ok(true)
fn untrack(&mut self, id: Id) -> Result<bool, Error> {
Ok(self.tracking.remove(&id))
}
fn announce_refs(&self, id: Id) -> Result<(), Error> {
fn announce_refs(&mut self, id: Id) -> Result<(), Error> {
self.updates.lock().unwrap().push(id);
Ok(())

3
radicle-node/src/test/simulator.rs
View File

@ -2,9 +2,6 @@
#![allow(clippy::collapsible_if)]
#![allow(dead_code)]
#[cfg(feature = "quickcheck")]
pub mod arbitrary;
use std::collections::{BTreeMap, BTreeSet, VecDeque};
use std::marker::PhantomData;
use std::ops::{Deref, DerefMut, Range};

0
radicle-node/src/test/tests.rs → radicle-node/src/tests.rs
View File

5
radicle/Cargo.toml
View File

@ -64,3 +64,8 @@ quickcheck = { version = "1", default-features = false }
path = "../radicle-crypto"
version = "0"
features = ["test"]
[dev-dependencies.radicle-node]
path = "../radicle-node"
version = "0"
features = ["test"]

81
radicle/src/node.rs
View File

@ -13,11 +13,19 @@ pub use features::Features;
/// Default name for control socket file.
pub const DEFAULT_SOCKET_NAME: &str = "radicle.sock";
/// 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";
#[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("received empty response for `{cmd}` command")]
EmptyResponse { cmd: &'static str },
}
pub trait Handle {
@ -67,31 +75,49 @@ impl Handle for Node {
fn fetch(&self, id: &Id) -> Result<(), Error> {
for line in self.call("fetch", id)? {
let line = line?;
log::info!("node: {}", line);
log::debug!("node: {}", line);
}
Ok(())
}
fn track(&self, id: &Id) -> Result<bool, Error> {
for line in self.call("track", id)? {
let line = line?;
log::info!("node: {}", line);
let mut line = self.call("track", id)?;
let line = line.next().ok_or(Error::EmptyResponse { cmd: "track" })??;
log::debug!("node: {}", line);
match line.as_str() {
RESPONSE_OK => Ok(true),
RESPONSE_NOOP => Ok(false),
_ => Err(Error::InvalidResponse {
cmd: "track",
response: line,
}),
}
Ok(true)
}
fn untrack(&self, id: &Id) -> Result<bool, Error> {
for line in self.call("untrack", id)? {
let line = line?;
log::info!("node: {}", line);
let mut line = self.call("untrack", id)?;
let line = line
.next()
.ok_or(Error::EmptyResponse { cmd: "untrack" })??;
log::debug!("node: {}", line);
match line.as_str() {
RESPONSE_OK => Ok(true),
RESPONSE_NOOP => Ok(false),
_ => Err(Error::InvalidResponse {
cmd: "untrack",
response: line,
}),
}
Ok(true)
}
fn announce_refs(&self, id: &Id) -> Result<(), Error> {
for line in self.call("announce-refs", id)? {
let line = line?;
log::info!("node: {}", line);
log::debug!("node: {}", line);
}
Ok(())
}
@ -105,3 +131,38 @@ impl Handle for Node {
pub fn connect<P: AsRef<Path>>(path: P) -> Result<Node, Error> {
Node::connect(path)
}
#[cfg(test)]
mod tests {
use std::thread;
use super::*;
use crate::test;
#[test]
fn test_track_untrack() {
let tmp = tempfile::tempdir().unwrap();
let socket = tmp.path().join("node.sock");
let proj = test::arbitrary::gen::<Id>(1);
thread::spawn({
use radicle_node as node;
let socket = socket.clone();
let handle = node::test::handle::Handle::default();
move || node::control::listen(socket, handle)
});
let handle = loop {
if let Ok(conn) = Node::connect(&socket) {
break conn;
}
};
assert!(handle.track(&proj).unwrap());
assert!(!handle.track(&proj).unwrap());
assert!(handle.untrack(&proj).unwrap());
assert!(!handle.untrack(&proj).unwrap());
}
}