diff --git a/backend/windmill-worker/src/worker.rs b/backend/windmill-worker/src/worker.rs index 97f67a4b46..c7bc935f9a 100644 --- a/backend/windmill-worker/src/worker.rs +++ b/backend/windmill-worker/src/worker.rs @@ -49,8 +49,8 @@ use crate::{ }; #[cfg(feature = "enterprise")] -async fn run_periodic_jobs(bucket: &str) { - tracing::info!("Running periodic jobs"); +async fn copy_cache_from_bucket(bucket: &str) { + tracing::info!("Copying cache from bucket {bucket}"); match Command::new("rclone") .arg("copy") @@ -68,6 +68,19 @@ async fn run_periodic_jobs(bucket: &str) { Err(e) => tracing::warn!("Failed to run periodic job pull. Error: {:?}", e), } + for x in [PIP_CACHE_DIR, DENO_CACHE_DIR, GO_CACHE_DIR] { + DirBuilder::new() + .recursive(true) + .create(x) + .await + .expect("could not create initial worker dir"); + } +} + +#[cfg(feature = "enterprise")] +async fn copy_cache_to_bucket(bucket: &str) { + tracing::info!("Copying cache to bucket {bucket}"); + match Command::new("rclone") .arg("copy") .arg(ROOT_CACHE_DIR) @@ -85,37 +98,6 @@ async fn run_periodic_jobs(bucket: &str) { } } -#[cfg(feature = "enterprise")] -async fn run_periodic_jobs_initial(bucket: &str) { - tracing::info!("Running periodic jobs"); - - match Command::new("rclone") - .arg("bisync") - .arg(format!(":s3,env_auth=true:{}", bucket)) - .arg(ROOT_CACHE_DIR) - .arg("--size-only") - .arg("--fast-list") - .arg("--force") - .arg("--resync") - .stdin(Stdio::null()) - .stdout(Stdio::null()) - .spawn() - { - Ok(mut h) => { - h.wait().await.unwrap(); - } - Err(e) => tracing::warn!("Failed to run periodic job. Error: {:?}", e), - } - - for x in [PIP_CACHE_DIR, DENO_CACHE_DIR, GO_CACHE_DIR] { - DirBuilder::new() - .recursive(true) - .create(x) - .await - .expect("could not create initial worker dir"); - } -} - #[tracing::instrument(level = "trace", skip_all)] pub async fn create_token_for_owner<'c>( mut tx: Transaction<'c, Postgres>, @@ -265,9 +247,6 @@ pub async fn run_worker( ) .await; - #[cfg(feature = "enterprise")] - let mut last_sync = Instant::now() - Duration::from_secs(NUM_SECS_ENV_CHECK + 1); - let mut last_ping = Instant::now() - Duration::from_secs(NUM_SECS_SYNC + 1); insert_initial_ping(worker_instance, &worker_name, ip, db).await; @@ -333,9 +312,12 @@ pub async fn run_worker( #[cfg(feature = "enterprise")] if let Some(ref s) = sync_bucket.clone() { - run_periodic_jobs_initial(&s).await; + copy_cache_from_bucket(&s).await; } + #[cfg(feature = "enterprise")] + let mut last_sync = Instant::now(); + let (same_worker_tx, mut same_worker_rx) = mpsc::channel::(5); loop { @@ -361,7 +343,8 @@ pub async fn run_worker( #[cfg(feature = "enterprise")] if last_sync.elapsed().as_secs() > NUM_SECS_SYNC { if let Some(ref s) = sync_bucket.clone() { - run_periodic_jobs(&s).await; + copy_cache_from_bucket(&s).await; + copy_cache_to_bucket(&s).await; } last_sync = Instant::now(); }