feat(jstzd): spawn rollup in jstzd server

crates/jstzd/src/task/jstzd.rs

@@ -1,5 +1,11 @@
-use super::{octez_baker::OctezBaker, octez_node::OctezNode, utils::retry, Task};
-use anyhow::Result;
+use super::{
+    octez_baker::OctezBaker,
+    octez_node::OctezNode,
+    octez_rollup::OctezRollup,
+    utils::{get_block_level, retry},
+    Task,
+};
+use anyhow::{bail, Context, Result};
 use async_dropper_simple::{AsyncDrop, AsyncDropper};
 use async_trait::async_trait;
 use axum::{
@@ -11,8 +17,10 @@ use axum::{
 use octez::r#async::{
     baker::OctezBakerConfig,
     client::{OctezClient, OctezClientConfig},
+    endpoint::Endpoint,
     node_config::OctezNodeConfig,
     protocol::ProtocolParameter,
+    rollup::OctezRollupConfig,
 };
 use serde::Serialize;
 use std::sync::Arc;
@@ -21,6 +29,7 @@ use tokio::{net::TcpListener, sync::RwLock, task::JoinHandle};
 struct Jstzd {
     octez_node: Arc<RwLock<OctezNode>>,
     baker: Arc<RwLock<OctezBaker>>,
+    rollup: Arc<RwLock<OctezRollup>>,
 }
 
 #[derive(Clone, Serialize)]
@@ -32,6 +41,8 @@ pub struct JstzdConfig {
     #[serde(rename(serialize = "octez-client"))]
     octez_client_config: OctezClientConfig,
     #[serde(skip_serializing)]
+    octez_rollup_config: OctezRollupConfig,
+    #[serde(skip_serializing)]
     protocol_params: ProtocolParameter,
 }
 
@@ -40,12 +51,14 @@ impl JstzdConfig {
         octez_node_config: OctezNodeConfig,
         baker_config: OctezBakerConfig,
         octez_client_config: OctezClientConfig,
+        octez_rollup_config: OctezRollupConfig,
         protocol_params: ProtocolParameter,
     ) -> Self {
         Self {
             octez_node_config,
             baker_config,
             octez_client_config,
+            octez_rollup_config,
             protocol_params,
         }
     }
@@ -72,20 +85,24 @@ impl Task for Jstzd {
         let octez_client = OctezClient::new(config.octez_client_config.clone());
         Self::wait_for_node(&octez_node).await?;
 
-        Self::import_activator(&octez_client).await;
+        Self::import_activator(&octez_client).await?;
+        Self::import_rollup_operator(&octez_client).await?;
         Self::activate_protocol(&octez_client, &config.protocol_params).await?;
-
         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?;
         Ok(Self {
             octez_node: Arc::new(RwLock::new(octez_node)),
             baker: Arc::new(RwLock::new(baker)),
+            rollup: Arc::new(RwLock::new(rollup)),
         })
     }
 
     async fn kill(&mut self) -> Result<()> {
         let results = futures::future::join_all([
             self.octez_node.write().await.kill(),
             self.baker.write().await.kill(),
+            self.rollup.write().await.kill(),
         ])
         .await;
 
@@ -107,6 +124,7 @@ impl Task for Jstzd {
         let check_results = futures::future::join_all([
             self.octez_node.read().await.health_check(),
             self.baker.read().await.health_check(),
+            self.rollup.read().await.health_check(),
         ])
         .await;
 
@@ -131,12 +149,25 @@ impl Jstzd {
     const ACTIVATOR_ACCOUNT_SK: &'static str =
         "unencrypted:edsk31vznjHSSpGExDMHYASz45VZqXN4DPxvsa4hAyY8dHM28cZzp6";
     const ACTIVATOR_ACCOUNT_ALIAS: &'static str = "activator";
+    const ROLLUP_OPERATOR_ACCOUNT_SK: &'static str =
+        "unencrypted:edsk3gUfUPyBSfrS9CCgmCiQsTCHGkviBDusMxDJstFtojtc1zcpsh";
+    const ROLLUP_OPERATOR_ACCOUNT_ALIAS: &'static str = "bootstrap1";
 
-    async fn import_activator(octez_client: &OctezClient) {
+    async fn import_activator(octez_client: &OctezClient) -> Result<()> {
         octez_client
             .import_secret_key(Self::ACTIVATOR_ACCOUNT_ALIAS, Self::ACTIVATOR_ACCOUNT_SK)
             .await
-            .expect("Failed to import account 'activator'");
+            .context("Failed to import account 'activator'")
+    }
+
+    async fn import_rollup_operator(octez_client: &OctezClient) -> Result<()> {
+        octez_client
+            .import_secret_key(
+                Self::ROLLUP_OPERATOR_ACCOUNT_ALIAS,
+                Self::ROLLUP_OPERATOR_ACCOUNT_SK,
+            )
+            .await
+            .context("Failed to import account 'rollup_operator'")
     }
 
     async fn activate_protocol(
@@ -165,6 +196,20 @@ impl Jstzd {
         }
         Ok(())
     }
+
+    /// Wait for the baker to bake at least `level` blocks.
+    async fn wait_for_block_level(node_endpoint: &Endpoint, level: i64) -> Result<()> {
+        let ready = retry(10, 1000, || async {
+            get_block_level(&node_endpoint.to_string())
+                .await
+                .map(|l| l >= level)
+        })
+        .await;
+        if !ready {
+            bail!("baker is not ready after retries");
+        }
+        Ok(())
+    }
 }
 
 #[derive(Clone, Default)]

crates/jstzd/src/task/utils.rs

@@ -1,3 +1,6 @@
+use anyhow::{anyhow, Result};
+use serde_json::Value;
+
 pub async fn retry<'a, F>(retries: u16, interval_ms: u64, f: impl Fn() -> F) -> bool
 where
     F: std::future::Future<Output = anyhow::Result<bool>> + Send + 'a,
@@ -13,3 +16,16 @@ where
     }
     false
 }
+
+pub async fn get_block_level(rpc_endpoint: &str) -> Result<i64> {
+    let blocks_head_endpoint = format!("{}/chains/main/blocks/head", rpc_endpoint);
+    let response: Value = reqwest::get(&blocks_head_endpoint).await?.json().await?;
+
+    let level = response
+        .get("header")
+        .and_then(|header| header.get("level"))
+        .ok_or_else(|| anyhow!("Failed to extract level from head block"))?;
+    level
+        .as_i64()
+        .ok_or_else(|| anyhow!("Level is not a valid i64"))
+}