Orchestrator, Executor and Executor Spawners

Cargo.toml

@@ -19,6 +19,7 @@ csv = "1.3.0"
 rand = "0.8.5"
 futures = "0.3.29"
 bollard = "0.15.0"
+tokio-util = "0.7.10"
 
 [[bin]]
 name = "mini-tree-server"

README.md

@@ -127,15 +127,15 @@ Before starting the tests, you need to build the `mini-tree-generator` image and
 Build the image with the following command:
 
 ```bash
-$ docker build . -t summa-aggregation
+docker build . -t summa-aggregation
 ```
 
 Ensure that the `summa-aggregation:latest` image exists in your local registry.
 
 Then, you can run the tests using this command:
 
 ```bash
-$ cargo test
+cargo test
 ```
 
 ## Mini Tree Generator

src/executor/container_spawner.rs

@@ -1,7 +1,5 @@
 use bollard::{
-    container::{
-        Config, CreateContainerOptions, RemoveContainerOptions, StartContainerOptions,
-    },
+    container::{Config, CreateContainerOptions, RemoveContainerOptions, StartContainerOptions},
     service::ContainerInspectResponse,
     Docker,
 };

src/executor/executor.rs

@@ -51,16 +51,20 @@ impl Executor {
     pub async fn generate_tree<const N_ASSETS: usize, const N_BYTES: usize>(
         &self,
         json_entries: Vec<JsonEntry>,
-    ) -> Result<MerkleSumTree<N_ASSETS, N_BYTES>, Box<dyn Error>> {
+    ) -> Result<MerkleSumTree<N_ASSETS, N_BYTES>, Box<dyn Error + Send>> {
         // Parse the response body into a MerkleSumTree
         let json_tree = self
             .client
             .post(&self.url)
             .json(&json_entries)
             .send()
-            .await?
+            .await
+            .map_err(|err| Box::new(err) as Box<dyn Error + Send>)
+            .unwrap()
             .json::<JsonMerkleSumTree>()
-            .await?;
+            .await
+            .map_err(|err| Box::new(err) as Box<dyn Error + Send>)
+            .unwrap();
 
         let entries = json_entries
             .iter()
@@ -100,25 +104,22 @@ impl Executor {
 #[cfg(test)]
 mod test {
     use futures::future;
-    use std::{error::Error, sync::atomic::AtomicUsize};
+    use std::error::Error;
 
-    use bollard::Docker;
-    use super::Executor;
-    use crate::orchestrator::entry_parser;
-    use crate::executor::ContainerSpawner;
     use crate::executor::spawner::ExecutorSpawner;
+    use crate::executor::ContainerSpawner;
+    use crate::orchestrator::entry_parser;
+    use bollard::Docker;
 
     #[tokio::test]
     async fn test_executor() -> Result<(), Box<dyn Error>> {
-        let spawner = ContainerSpawner::new(
-            "summa-aggregation".to_string(),
-            "excutor_test".to_string(),
-        );
+        let spawner =
+            ContainerSpawner::new("summa-aggregation".to_string(), "excutor_test".to_string());
 
         let executor = spawner.spawn_executor().await;
 
         let entries = entry_parser::<_, 2, 14>("./src/orchestrator/csv/entry_16.csv").unwrap();
-        let merkle_tree = executor.generate_tree::<2, 14>(entries).await?;
+        let merkle_tree = executor.generate_tree::<2, 14>(entries).await.unwrap();
 
         spawner.terminate_executors().await;
 

src/orchestrator/entry_csv_parser.rs

@@ -13,15 +13,23 @@ struct CSVEntry {
 /// Parses a CSV file stored at path into a vector of Entries
 pub fn entry_parser<P: AsRef<Path>, const N_ASSETS: usize, const N_BYTES: usize>(
     path: P,
-) -> Result<Vec<JsonEntry>, Box<dyn Error>> {
+) -> Result<Vec<JsonEntry>, Box<dyn Error + Send>> {
     let mut json_entries = Vec::<JsonEntry>::new();
-    let file = File::open(path)?;
+    // let file = File::open(path)?;
+    let file = match File::open(path) {
+        Ok(f) => f,
+        Err(e) => return Err(Box::new(e) as Box<dyn Error + Send>),
+    };
+
     let mut rdr = csv::ReaderBuilder::new()
         .delimiter(b';') // The fields are separated by a semicolon
         .from_reader(file);
 
     for result in rdr.deserialize() {
-        let record: CSVEntry = result?;
+        let record: CSVEntry = match result {
+            Ok(r) => r,
+            Err(e) => return Err(Box::new(e) as Box<dyn Error + Send>),
+        };
 
         let entry = JsonEntry {
             username: record.username,