feat(jstzd): spawn jstz node in jstzd server

crates/jstzd/src/config.rs

@@ -3,6 +3,7 @@ use std::path::{Path, PathBuf};
 use crate::task::jstzd::JstzdConfig;
 use crate::{EXCHANGER_ADDRESS, JSTZ_NATIVE_BRIDGE_ADDRESS, JSTZ_ROLLUP_ADDRESS};
 use anyhow::{Context, Result};
+use jstz_node::config::JstzNodeConfig;
 use octez::r#async::endpoint::Endpoint;
 use octez::r#async::protocol::{BootstrapContract, ProtocolParameter};
 use octez::r#async::rollup::{OctezRollupConfigBuilder, RollupDataDir};
@@ -83,6 +84,13 @@ pub(crate) async fn build_config(
     .build()
     .unwrap();
 
+    // TODO: https://linear.app/tezos/issue/JSTZ-240/add-jstz-node-config-builder
+    // Dummy jstz node config for now
+    let jstz_node_config = JstzNodeConfig::new(
+        &Endpoint::localhost(8000),
+        &Endpoint::localhost(8000),
+        &PathBuf::from("dummy-kernel-log-file"),
+    );
     let protocol_params = build_protocol_params(config.protocol).await?;
     let server_port = config.server_port.unwrap_or(unused_port());
     Ok((
@@ -92,6 +100,7 @@ pub(crate) async fn build_config(
             baker_config,
             octez_client_config,
             octez_rollup_config,
+            jstz_node_config,
             protocol_params,
         ),
     ))

crates/jstzd/src/task/child_wrapper.rs

@@ -4,7 +4,9 @@ use std::sync::Arc;
 use tokio::process::Child;
 use tokio::sync::RwLock;
 
-pub type SharedChildWrapper = Arc<RwLock<AsyncDropper<ChildWrapper>>>;
+pub type Shared<T> = Arc<RwLock<T>>;
+
+pub type SharedChildWrapper = Shared<AsyncDropper<ChildWrapper>>;
 
 #[derive(Default)]
 pub struct ChildWrapper {

crates/jstzd/src/task/jstzd.rs

@@ -1,4 +1,6 @@
 use super::{
+    child_wrapper::Shared,
+    jstz_node::JstzNode,
     octez_baker::OctezBaker,
     octez_node::OctezNode,
     octez_rollup::OctezRollup,
@@ -14,6 +16,7 @@ use axum::{
     routing::{get, put},
     Router,
 };
+use jstz_node::config::JstzNodeConfig;
 use octez::r#async::{
     baker::OctezBakerConfig,
     client::{OctezClient, OctezClientConfig},
@@ -30,10 +33,21 @@ use tokio::{
     task::JoinHandle,
 };
 
+trait IntoShared {
+    fn into_shared(self) -> Shared<Self>;
+}
+
+impl<T: Task> IntoShared for T {
+    fn into_shared(self) -> Shared<Self> {
+        Arc::new(RwLock::new(self))
+    }
+}
+
 struct Jstzd {
-    octez_node: Arc<RwLock<OctezNode>>,
-    baker: Arc<RwLock<OctezBaker>>,
-    rollup: Arc<RwLock<OctezRollup>>,
+    octez_node: Shared<OctezNode>,
+    baker: Shared<OctezBaker>,
+    rollup: Shared<OctezRollup>,
+    jstz_node: Shared<JstzNode>,
 }
 
 #[derive(Clone, Serialize)]
@@ -47,6 +61,8 @@ pub struct JstzdConfig {
     #[serde(skip_serializing)]
     octez_rollup_config: OctezRollupConfig,
     #[serde(skip_serializing)]
+    jstz_node_config: JstzNodeConfig,
+    #[serde(skip_serializing)]
     protocol_params: ProtocolParameter,
 }
 
@@ -56,13 +72,15 @@ impl JstzdConfig {
         baker_config: OctezBakerConfig,
         octez_client_config: OctezClientConfig,
         octez_rollup_config: OctezRollupConfig,
+        jstz_node_config: JstzNodeConfig,
         protocol_params: ProtocolParameter,
     ) -> Self {
         Self {
             octez_node_config,
             baker_config,
             octez_client_config,
             octez_rollup_config,
+            jstz_node_config,
             protocol_params,
         }
     }
@@ -99,10 +117,12 @@ impl Task for Jstzd {
         let baker = OctezBaker::spawn(config.baker_config.clone()).await?;
         Self::wait_for_block_level(&config.octez_node_config.rpc_endpoint, 3).await?;
         let rollup = OctezRollup::spawn(config.octez_rollup_config.clone()).await?;
+        let jstz_node = JstzNode::spawn(config.jstz_node_config.clone()).await?;
         Ok(Self {
-            octez_node: Arc::new(RwLock::new(octez_node)),
-            baker: Arc::new(RwLock::new(baker)),
-            rollup: Arc::new(RwLock::new(rollup)),
+            octez_node: octez_node.into_shared(),
+            baker: baker.into_shared(),
+            rollup: rollup.into_shared(),
+            jstz_node: jstz_node.into_shared(),
         })
     }
 
@@ -111,6 +131,7 @@ impl Task for Jstzd {
             self.octez_node.write().await.kill(),
             self.baker.write().await.kill(),
             self.rollup.write().await.kill(),
+            self.jstz_node.write().await.kill(),
         ])
         .await;
 
@@ -133,6 +154,7 @@ impl Task for Jstzd {
             self.octez_node.read().await.health_check(),
             self.baker.read().await.health_check(),
             self.rollup.read().await.health_check(),
+            self.jstz_node.read().await.health_check(),
         ])
         .await;
 
@@ -146,7 +168,7 @@ impl Task for Jstzd {
         }
 
         if !err.is_empty() {
-            Err(anyhow::anyhow!("failed to perform health check: {:?}", err))
+            bail!("failed to perform health check: {:?}", err)
         } else {
             Ok(healthy)
         }
@@ -228,7 +250,7 @@ impl Jstzd {
 
 #[derive(Clone, Default)]
 pub struct JstzdServerInner {
-    state: Arc<RwLock<ServerState>>,
+    state: Shared<ServerState>,
 }
 
 #[derive(Default)]
@@ -341,6 +363,19 @@ impl JstzdServer {
             None => false,
         }
     }
+
+    pub async fn jstz_node_healthy(&self) -> bool {
+        match &self.inner.state.read().await.jstzd {
+            Some(v) => v
+                .jstz_node
+                .read()
+                .await
+                .health_check()
+                .await
+                .unwrap_or(false),
+            None => false,
+        }
+    }
 }
 
 async fn health_check(state: &ServerState) -> bool {
@@ -375,31 +410,29 @@ async fn shutdown(state: &mut ServerState) -> Result<()> {
     Ok(())
 }
 
-async fn health_check_handler(
-    state: State<Arc<RwLock<ServerState>>>,
-) -> http::StatusCode {
+async fn health_check_handler(state: State<Shared<ServerState>>) -> http::StatusCode {
     let lock = state.read().await;
     match health_check(&lock).await {
         true => http::StatusCode::OK,
         _ => http::StatusCode::INTERNAL_SERVER_ERROR,
     }
 }
 
-async fn shutdown_handler(state: State<Arc<RwLock<ServerState>>>) -> http::StatusCode {
+async fn shutdown_handler(state: State<Shared<ServerState>>) -> http::StatusCode {
     let mut lock = state.write().await;
     if shutdown(&mut lock).await.is_err() {
         return http::StatusCode::INTERNAL_SERVER_ERROR;
     };
     http::StatusCode::NO_CONTENT
 }
 
-async fn all_config_handler(state: State<Arc<RwLock<ServerState>>>) -> impl IntoResponse {
+async fn all_config_handler(state: State<Shared<ServerState>>) -> impl IntoResponse {
     let config = &state.read().await.jstzd_config_json;
     serde_json::to_string(config).unwrap().into_response()
 }
 
 async fn config_handler(
-    state: State<Arc<RwLock<ServerState>>>,
+    state: State<Shared<ServerState>>,
     Path(config_type): Path<String>,
 ) -> impl IntoResponse {
     let config = &state.read().await.jstzd_config_json;

crates/jstzd/tests/jstzd_test.rs

@@ -4,6 +4,7 @@ use std::path::PathBuf;
 use std::str::FromStr;
 use std::time::Duration;
 
+use jstz_node::config::JstzNodeConfig;
 use jstzd::jstz_rollup_path::*;
 
 use http::Uri;
@@ -40,10 +41,15 @@ async fn jstzd_test() {
         Uri::from_str(&format!("http://127.0.0.1:{}", unused_port())).unwrap(),
     )
     .unwrap();
+    let jstz_node_rpc_endpoint = Endpoint::localhost(unused_port());
     let jstzd_port = unused_port();
-    let (mut jstzd, config, kernel_debug_file) =
-        create_jstzd_server(&octez_node_rpc_endpoint, &rollup_rpc_endpoint, jstzd_port)
-            .await;
+    let (mut jstzd, config, kernel_debug_file) = create_jstzd_server(
+        &octez_node_rpc_endpoint,
+        &rollup_rpc_endpoint,
+        &jstz_node_rpc_endpoint,
+        jstzd_port,
+    )
+    .await;
 
     jstzd.run().await.unwrap();
     ensure_jstzd_components_are_up(&jstzd, &octez_node_rpc_endpoint, jstzd_port).await;
@@ -80,6 +86,7 @@ async fn jstzd_test() {
 async fn create_jstzd_server(
     octez_node_rpc_endpoint: &Endpoint,
     rollup_rpc_endpoint: &Endpoint,
+    jstz_node_rpc_endpoint: &Endpoint,
     jstzd_port: u16,
 ) -> (JstzdServer, JstzdConfig, NamedTempFile) {
     let run_options = OctezNodeRunOptionsBuilder::new()
@@ -148,11 +155,17 @@ async fn create_jstzd_server(
     .set_kernel_debug_file(kernel_debug_file.path())
     .build()
     .expect("failed to build rollup config");
+    let jstz_node_config = JstzNodeConfig::new(
+        jstz_node_rpc_endpoint,
+        &rollup_config.rpc_endpoint,
+        kernel_debug_file.path(),
+    );
     let config = JstzdConfig::new(
         octez_node_config,
         baker_config,
         octez_client_config.clone(),
         rollup_config.clone(),
+        jstz_node_config,
         protocol_params,
     );
     (
@@ -186,6 +199,7 @@ async fn ensure_jstzd_components_are_up(
     let rollup_running =
         retry(10, 1000, || async { Ok(jstzd.rollup_healthy().await) }).await;
     assert!(rollup_running);
+    assert!(jstzd.jstz_node_healthy().await);
     assert!(jstzd.health_check().await);
 }
 
@@ -221,8 +235,8 @@ async fn ensure_jstzd_components_are_down(
     .await;
     assert!(node_destroyed);
     assert!(!jstzd.baker_healthy().await);
-
     assert!(!jstzd.rollup_healthy().await);
+    assert!(!jstzd.jstz_node_healthy().await);
     assert!(!jstzd.health_check().await);
 }