perf(serverless): download deployments in parallel (#756)

* feat(serverless): download deployments in parallel

* chore: add cronjob commented code

* fix(runtime): flaky test

.changeset/quiet-days-begin.md

@@ -0,0 +1,5 @@
+---
+'@lagon/serverless': patch
+---
+
+Download deployments in parallel

crates/runtime/tests/streams.rs

@@ -1,4 +1,4 @@
-use httptest::{bytes::Bytes, matchers::*, responders::*, Expectation, Server};
+use httptest::bytes::Bytes;
 use lagon_runtime_http::{Request, Response, RunResult, StreamResult};
 use lagon_runtime_isolate::options::IsolateOptions;
 use std::collections::HashMap;
@@ -167,34 +167,28 @@ async fn start_and_pull() {
 #[tokio::test]
 async fn response_before_write() {
     utils::setup();
-    let server = Server::run();
-    server.expect(
-        Expectation::matching(request::method_path("GET", "/"))
-            .respond_with(status_code(200).body("Hello")),
-    );
-    let url = server.url("/");
-
-    let (send, receiver) = utils::create_isolate(IsolateOptions::new(format!(
-        "export function handler() {{
-    const transformStream = new TransformStream({{
-        start(controller) {{
+    let (send, receiver) = utils::create_isolate(IsolateOptions::new(
+        "export function handler() {
+    const transformStream = new TransformStream({
+        start(controller) {
             controller.enqueue(new TextEncoder().encode('Loading...'));
-        }}
-    }})
+        }
+    })
 
     const writeableStream = transformStream.writable;
     const readableStream = transformStream.readable;
 
     const writer = writeableStream.getWriter();
 
-    fetch('{url}').then(res => res.text().then(text => {{
-        writer.write(new TextEncoder().encode(text));
+    setTimeout(() => {
+        writer.write(new TextEncoder().encode('Hello'));
         writer.close();
-    }}))
+    }, 100);
 
     return new Response(readableStream);
-}}"
-    )));
+}"
+        .into(),
+    ));
     send(Request::default());
 
     assert_eq!(

crates/serverless/src/deployments/mod.rs

@@ -169,31 +169,27 @@ OR
         error!("Failed to delete old deployments: {:?}", error);
     }
 
-    {
-        // let mut cronjob = cronjob.lock().await;
-
-        for deployment in deployments_list {
-            if !deployment.has_code() {
-                if let Err(error) = download_deployment(&deployment, Arc::clone(&downloader)).await
-                {
-                    error!("Failed to download deployment {}: {}", deployment.id, error);
-                    continue;
-                }
+    futures::future::join_all(deployments_list.into_iter().map(|deployment| async {
+        if !deployment.has_code() {
+            if let Err(error) = download_deployment(&deployment, Arc::clone(&downloader)).await {
+                error!("Failed to download deployment {}: {}", deployment.id, error);
+                return;
             }
+        }
 
-            let deployment = Arc::new(deployment);
-
-            for domain in deployment.get_domains() {
-                deployments.insert(domain, Arc::clone(&deployment));
-            }
+        let deployment = Arc::new(deployment);
 
-            // if deployment.should_run_cron() {
-            //     if let Err(error) = cronjob.add(deployment).await {
-            //         error!("Failed to register cron: {}", error);
-            //     }
-            // }
+        for domain in deployment.get_domains() {
+            deployments.insert(domain, Arc::clone(&deployment));
         }
-    }
+
+        // if deployment.should_run_cron() {
+        //     if let Err(error) = cronjob.add(deployment).await {
+        //         error!("Failed to register cron: {}", error);
+        //     }
+        // }
+    }))
+    .await;
 
     Ok(deployments)
 }