feat(jstzd): implement OctezNode

Cargo.toml

@@ -33,6 +33,8 @@ actix-web = "4.5.1"
 actix-web-lab = "0.20.0"
 ansi_term = "0.12.1"
 anyhow = "1.0.82"
+async-dropper-simple = { version = "0.2.6", features = ["tokio"] }
+async-trait = "0.1.82"
 base64 = "0.21.7"
 bincode = "1.3.3"
 boa_engine = { version = "0.17.0", features = ["fuzz"] }
@@ -100,8 +102,6 @@ tokio-util = "0.7.10"
 url = "2.4.1"
 urlpattern = "0.2.0"
 wasm-bindgen = "0.2.92"
-async-dropper = { version = "0.3.1", features = ["tokio", "simple"] }
-async-trait = "0.1.82"
 
 [workspace.dependencies.tezos-smart-rollup]
 version = "0.2.2"

crates/jstzd/Cargo.toml

@@ -7,12 +7,17 @@ repository.workspace = true
 
 [dependencies]
 anyhow.workspace = true
-async-dropper.workspace = true
+async-dropper-simple.workspace = true
 async-trait.workspace = true
 bollard.workspace = true
 futures-util.workspace = true
+octez = { path = "../octez" }
 tokio.workspace = true
 
+[dev-dependencies]
+rand.workspace = true
+reqwest.workspace = true
+
 [[bin]]
 name = "jstzd"
-path = "src/main.rs"
+path = "src/main.rs"

crates/jstzd/src/task/mod.rs

@@ -1,3 +1,5 @@
+pub mod octez_node;
+
 use anyhow::Result;
 use async_trait::async_trait;
 

crates/jstzd/src/task/octez_node.rs

@@ -0,0 +1,294 @@
+use super::Task;
+use anyhow::Result;
+use async_dropper_simple::{AsyncDrop, AsyncDropper};
+use async_trait::async_trait;
+use std::{fs::File, path::PathBuf, sync::Arc};
+use tokio::sync::RwLock;
+
+#[cfg(test)]
+use tests::{Child, InnerOctezNode};
+
+#[cfg(not(test))]
+use octez::OctezNode as InnerOctezNode;
+#[cfg(not(test))]
+use std::process::Child;
+
+#[derive(Clone)]
+pub struct OctezNodeConfig {
+    binary_path: PathBuf,
+    data_dir: PathBuf,
+    network: String,
+    rpc_endpoint: String,
+    log_file: PathBuf,
+    options: Vec<String>,
+}
+
+#[derive(Default)]
+pub struct OctezNodeConfigBuilder {
+    binary_path: Option<PathBuf>,
+    data_dir: Option<PathBuf>,
+    network: Option<String>,
+    rpc_endpoint: Option<String>,
+    log_file: Option<PathBuf>,
+    options: Option<Vec<String>>,
+}
+
+impl OctezNodeConfigBuilder {
+    pub fn new() -> Self {
+        OctezNodeConfigBuilder::default()
+    }
+
+    pub fn set_binary_path(&mut self, path: &str) -> &mut Self {
+        self.binary_path = Some(PathBuf::from(path));
+        self
+    }
+
+    pub fn set_data_dir(&mut self, path: &str) -> &mut Self {
+        self.data_dir = Some(PathBuf::from(path));
+        self
+    }
+
+    pub fn set_network(&mut self, network: &str) -> &mut Self {
+        self.network = Some(network.to_owned());
+        self
+    }
+
+    pub fn set_rpc_endpoint(&mut self, endpoint: &str) -> &mut Self {
+        self.rpc_endpoint = Some(endpoint.to_owned());
+        self
+    }
+
+    pub fn set_log_file(&mut self, path: &str) -> &mut Self {
+        self.log_file = Some(PathBuf::from(path));
+        self
+    }
+
+    pub fn set_run_options(&mut self, options: &[&str]) -> &mut Self {
+        self.options = Some(
+            options
+                .iter()
+                .map(|v| (*v).to_owned())
+                .collect::<Vec<String>>(),
+        );
+        self
+    }
+
+    pub fn build(&mut self) -> Result<OctezNodeConfig> {
+        Ok(OctezNodeConfig {
+            binary_path: self
+                .binary_path
+                .take()
+                .unwrap_or(PathBuf::from("octez-node")),
+            data_dir: self
+                .data_dir
+                .take()
+                .unwrap_or(PathBuf::from("/tmp/.octez-node")),
+            network: self.network.take().unwrap_or("sandbox".to_owned()),
+            rpc_endpoint: self
+                .rpc_endpoint
+                .take()
+                .unwrap_or("localhost:8732".to_owned()),
+            log_file: self
+                .log_file
+                .take()
+                .unwrap_or(PathBuf::from("/tmp/octez-node.log")),
+            options: self.options.take().unwrap_or_default(),
+        })
+    }
+}
+
+#[derive(Default)]
+struct ChildWrapper {
+    inner: Option<Child>,
+}
+
+impl ChildWrapper {
+    pub async fn kill(&mut self) -> anyhow::Result<()> {
+        if let Some(mut v) = self.inner.take() {
+            return Ok(v.kill()?);
+        }
+        Ok(())
+    }
+}
+
+#[async_trait]
+impl AsyncDrop for ChildWrapper {
+    async fn async_drop(&mut self) {
+        let _ = self.kill().await;
+    }
+}
+
+#[derive(Default, Clone)]
+pub struct OctezNode {
+    inner: Arc<RwLock<AsyncDropper<ChildWrapper>>>,
+}
+
+#[async_trait]
+impl Task for OctezNode {
+    type Config = OctezNodeConfig;
+
+    /// Spins up the task with the given config.
+    async fn spawn(config: Self::Config) -> Result<Self> {
+        let node = InnerOctezNode {
+            octez_node_bin: Some(config.binary_path),
+            octez_node_dir: config.data_dir,
+        };
+
+        node.config_init(&config.network, "localhost:8731", &config.rpc_endpoint, 0)?;
+        node.generate_identity()?;
+        Ok(OctezNode {
+            inner: Arc::new(RwLock::new(AsyncDropper::new(ChildWrapper {
+                inner: Some(
+                    node.run(
+                        &File::create(&config.log_file)?,
+                        config
+                            .options
+                            .iter()
+                            .map(|s| s as &str)
+                            .collect::<Vec<&str>>()
+                            .as_slice(),
+                    )?,
+                ),
+            }))),
+        })
+    }
+
+    /// Aborts the running task.
+    async fn kill(&mut self) -> Result<()> {
+        let mut inner = self.inner.write().await;
+        Ok(inner.inner_mut().kill().await?)
+    }
+
+    /// Conducts a health check on the running task.
+    async fn health_check(&self) -> Result<bool> {
+        todo!()
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::{OctezNode, OctezNodeConfigBuilder};
+    use crate::task::Task;
+    use anyhow::Result;
+    use rand::{distributions::Alphanumeric, Rng};
+    use std::{fs::File, path::PathBuf};
+
+    pub struct Child {
+        file_path: Option<String>,
+    }
+
+    impl Child {
+        pub fn kill(&mut self) -> std::io::Result<()> {
+            if self.file_path.is_none() {
+                return Err(std::io::Error::new(
+                    std::io::ErrorKind::Other,
+                    anyhow::anyhow!("failed"),
+                ));
+            }
+            File::create(self.file_path.as_ref().unwrap()).unwrap();
+            Ok(())
+        }
+    }
+
+    pub struct InnerOctezNode {
+        pub octez_node_bin: Option<PathBuf>,
+        pub octez_node_dir: PathBuf,
+    }
+
+    impl InnerOctezNode {
+        pub fn config_init(
+            &self,
+            _network: &str,
+            _http_endpoint: &str,
+            _rpc_endpoint: &str,
+            _num_connections: u32,
+        ) -> Result<()> {
+            Ok(())
+        }
+
+        pub fn run(&self, _log_file: &File, options: &[&str]) -> Result<Child> {
+            let file_path = if !options.is_empty() {
+                Some(options[0].to_owned())
+            } else {
+                None
+            };
+            Ok(Child { file_path })
+        }
+
+        pub fn generate_identity(&self) -> Result<()> {
+            Ok(())
+        }
+    }
+
+    fn get_random_string() -> String {
+        rand::thread_rng()
+            .sample_iter(&Alphanumeric)
+            .take(16)
+            .map(char::from)
+            .collect::<String>()
+    }
+
+    fn get_temp_file_path() -> String {
+        let dir = std::env::temp_dir();
+        let path = dir.join(format!("jstzd-test-{}.txt", get_random_string()));
+        path.into_os_string().into_string().unwrap()
+    }
+
+    #[tokio::test(flavor = "multi_thread")]
+    async fn kill_should_succeed() {
+        let file_path = get_temp_file_path();
+
+        // The mock child is made to create a file based on the first item in
+        // `options` when it is killed
+        let mut f = OctezNode::spawn(
+            OctezNodeConfigBuilder::new()
+                .set_binary_path("/tmp/binary")
+                .set_data_dir("/tmp/something")
+                .set_network("sandbox")
+                .set_rpc_endpoint("localhost:8732")
+                .set_log_file(&get_temp_file_path())
+                .set_run_options(&[&file_path])
+                .build()
+                .unwrap(),
+        )
+        .await
+        .unwrap();
+
+        let target_file = PathBuf::from(&file_path);
+        let op = f.inner.read().await;
+        assert!(op.inner.is_some());
+        assert!(!target_file.exists());
+        drop(op);
+
+        assert!(f.kill().await.is_ok());
+        let op = f.inner.read().await;
+        assert!(op.inner.is_none());
+        assert!(target_file.exists());
+    }
+
+    #[tokio::test(flavor = "multi_thread")]
+    async fn kill_should_fail() {
+        // The mock child is made to fail the `kill` call when it is given
+        // an empty list of options
+        let mut f = OctezNode::spawn(
+            OctezNodeConfigBuilder::new()
+                .set_binary_path("/tmp/binary")
+                .set_data_dir("/tmp/something")
+                .set_network("sandbox")
+                .set_rpc_endpoint("localhost:8732")
+                .set_log_file(&get_temp_file_path())
+                .set_run_options(&[])
+                .build()
+                .unwrap(),
+        )
+        .await
+        .unwrap();
+
+        let op = f.inner.read().await;
+        assert!(op.inner.is_some());
+        drop(op);
+
+        let res = f.kill().await;
+        assert_eq!(res.unwrap_err().to_string(), "failed");
+    }
+}