From a3487f18c455179986d661615aaaae4f8ab063e0 Mon Sep 17 00:00:00 2001
From: Ruben Fiszel <ruben@rubenfiszel.com>
Date: Fri, 21 Apr 2023 22:20:57 +0200
Subject: [PATCH] fix(backend): global cache synco only start if all piptars
 have been downloaded

---
 backend/windmill-worker/src/global_cache.rs |  7 ++++---
 backend/windmill-worker/src/worker.rs       | 16 +++-------------
 2 files changed, 7 insertions(+), 16 deletions(-)

diff --git a/backend/windmill-worker/src/global_cache.rs b/backend/windmill-worker/src/global_cache.rs
index 2907109a3e..ef9f2afa02 100644
--- a/backend/windmill-worker/src/global_cache.rs
+++ b/backend/windmill-worker/src/global_cache.rs
@@ -161,7 +161,7 @@ pub async fn copy_cache_from_bucket(bucket: &str, tx: Sender<()>) -> error::Resu
     )
     .await
     {
-        tracing::info!("Failed to to copy cache from bucket. Error: {:?}", e);
+        tracing::info!("Failed to copy cache from bucket. Error: {:?}", e);
         return Err(e);
     }
 
@@ -201,7 +201,7 @@ pub async fn copy_cache_to_bucket(bucket: &str) -> error::Result<()> {
     )
     .await
     {
-        tracing::info!("Failed to to copy cache to bucket. Error: {:?}", e);
+        tracing::info!("Failed to copy cache to bucket. Error: {:?}", e);
         return Err(e);
     }
     tracing::info!(
@@ -451,9 +451,10 @@ pub async fn extract_pip_tar(tar: &str, folder: &str) -> error::Result<()> {
     use tokio::fs;
 
     let start: Instant = Instant::now();
-    fs::create_dir(&folder).await?;
+    fs::create_dir_all(&folder).await?;
     if let Err(e) = execute_command(&folder, "tar", vec!["-xpvf", tar]).await {
         tracing::info!("Failed to untar piptar. Error: {:?}", e);
+        fs::remove_dir_all(&folder).await?;
         return Err(e);
     }
     tracing::info!(
diff --git a/backend/windmill-worker/src/worker.rs b/backend/windmill-worker/src/worker.rs
index 022ebd3bb2..4ea3ee047b 100644
--- a/backend/windmill-worker/src/worker.rs
+++ b/backend/windmill-worker/src/worker.rs
@@ -38,7 +38,7 @@ use tokio::{
     sync::{
         mpsc::{self, Sender},  watch, broadcast, RwLock, Barrier
     },
-    time::{interval, sleep, Instant, MissedTickBehavior}
+    time::{interval, sleep, Instant, MissedTickBehavior}, join
 };
 
 use futures::{
@@ -428,20 +428,10 @@ pub async fn run_worker<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
             let worker_name2 = worker_name.clone();
 
             handles.push(tokio::task::spawn(async move {
-                tracing::info!(worker = %worker_name2, "Started initial denogo tar sync in background");
-                copy_denogo_cache_from_bucket_as_tar(&bucket).await;
+                tracing::info!(worker = %worker_name2, "Started initial sync in background");
+                join!(copy_denogo_cache_from_bucket_as_tar(&bucket), copy_all_piptars_from_bucket(&bucket));
                 let _ = copy_to_bucket_tx2.send(()).await;
             }));
-
-            let bucket = s.to_string();
-            let copy_to_bucket_tx = copy_to_bucket_tx.clone();
-            let worker_name = worker_name.clone();
-
-            handles.push(tokio::task::spawn(async move {
-                tracing::info!(worker = %worker_name, "Started initial piptars cache sync in background");
-                copy_all_piptars_from_bucket(&bucket).await;
-                let _ = copy_to_bucket_tx.send(()).await;
-            }));
         }
     }