diff --git a/backend/tests/workspace_dependencies.rs b/backend/tests/workspace_dependencies.rs index 6875520125..6e0018f9cb 100644 --- a/backend/tests/workspace_dependencies.rs +++ b/backend/tests/workspace_dependencies.rs @@ -151,7 +151,7 @@ mod workspace_dependencies { name: None, // No name = default workspace dependencies description: None, } - .create("", "", "", db) + .create(("".to_owned(), "".to_owned(), "".to_owned()), db.clone()) .await .unwrap(); diff --git a/backend/windmill-api/src/workspace_dependencies.rs b/backend/windmill-api/src/workspace_dependencies.rs index fb043be071..2330de200a 100644 --- a/backend/windmill-api/src/workspace_dependencies.rs +++ b/backend/windmill-api/src/workspace_dependencies.rs @@ -6,7 +6,6 @@ use axum::{ use http::StatusCode; use serde::Deserialize; use windmill_common::{ - cache::workspace_dependencies::EXISTS_CACHE_TIMEOUT, error::{self, JsonResult}, scripts::ScriptLang, users::username_to_permissioned_as, @@ -14,9 +13,8 @@ use windmill_common::{ workspace_dependencies::WorkspaceDependencies, DB, }; -use windmill_worker::{ - scoped_dependency_map, trigger_dependents_to_recompute_dependencies, - workspace_dependencies::NewWorkspaceDependencies, +use windmill_worker::workspace_dependencies::{ + trigger_dependents_to_recompute_dependencies_in_the_background, NewWorkspaceDependencies, }; use crate::db::ApiAuthed; @@ -44,10 +42,12 @@ async fn create( format!( "{}", nwd.create( - &authed.email, - &authed.username, - &username_to_permissioned_as(&authed.username), - &db + ( + authed.email, + username_to_permissioned_as(&authed.username), + authed.username, + ), + db ) .await? ), @@ -94,33 +94,20 @@ async fn archive( let db = &db; WorkspaceDependencies::archive(params.name.clone(), language, &w_id, db).await?; - if params.name.is_none() { - tracing::debug!( - workspace_id = %w_id, - ?language, - "waiting for cache timeout after archiving unnamed workspace dependencies" - ); - // for context read [[NewWorkspaceDependencies::create]] - tokio::time::sleep(EXISTS_CACHE_TIMEOUT).await; - } - - trigger_dependents_to_recompute_dependencies( - &w_id, - scoped_dependency_map::ScopedDependencyMap::get_dependents( - WorkspaceDependencies::to_path(¶ms.name, language)?.as_str(), - &w_id, - db, - ) - .await?, - None, - None, - &authed.email, - &authed.username, - &username_to_permissioned_as(&authed.username), - db, - vec![], + trigger_dependents_to_recompute_dependencies_in_the_background( + params.name.is_none(), + w_id, + language, + ( + authed.email, + username_to_permissioned_as(&authed.username), + authed.username, + ), + WorkspaceDependencies::to_path(¶ms.name, language)?, + db.clone(), ) - .await + .await; + Ok(()) } #[axum::debug_handler] @@ -136,31 +123,18 @@ async fn delete( let db = &db; WorkspaceDependencies::delete(params.name.clone(), language, &w_id, db).await?; - if params.name.is_none() { - tracing::debug!( - workspace_id = %w_id, - ?language, - "waiting for cache timeout after deleting unnamed workspace dependencies" - ); - // for context read [[NewWorkspaceDependencies::create]] - tokio::time::sleep(EXISTS_CACHE_TIMEOUT).await; - } - - trigger_dependents_to_recompute_dependencies( - &w_id, - scoped_dependency_map::ScopedDependencyMap::get_dependents( - WorkspaceDependencies::to_path(¶ms.name, language)?.as_str(), - &w_id, - db, - ) - .await?, - None, - None, - &authed.email, - &authed.username, - &username_to_permissioned_as(&authed.username), - db, - vec![], + trigger_dependents_to_recompute_dependencies_in_the_background( + params.name.is_none(), + w_id, + language, + ( + authed.email, + username_to_permissioned_as(&authed.username), + authed.username, + ), + WorkspaceDependencies::to_path(¶ms.name, language)?, + db.clone(), ) - .await + .await; + Ok(()) } diff --git a/backend/windmill-worker/src/workspace_dependencies.rs b/backend/windmill-worker/src/workspace_dependencies.rs index 233b2368cd..31c1421748 100644 --- a/backend/windmill-worker/src/workspace_dependencies.rs +++ b/backend/windmill-worker/src/workspace_dependencies.rs @@ -28,10 +28,8 @@ impl NewWorkspaceDependencies { /// and rebuilds the dependency map if this is the first unnamed dependency for the workspace. pub async fn create<'c>( self, - email: &str, - created_by: &str, - permissioned_as: &str, - db: &sqlx::Pool, + metadata: (String, String, String), + db: sqlx::Pool, ) -> error::Result { // Check if all workers support workspace dependencies feature windmill_common::workspace_dependencies::min_version_supports_v0_workspace_dependencies() @@ -46,7 +44,7 @@ impl NewWorkspaceDependencies { let setting_name = format!("workspace_dependencies_map_rebuilt:{}", self.workspace_id); let already_rebuilt = windmill_common::global_settings::load_value_from_global_settings( - db, + &db, &setting_name, ) .await? @@ -57,11 +55,11 @@ impl NewWorkspaceDependencies { workspace_id = %self.workspace_id, "Rebuilding workspace dependencies map for first unnamed workspace dependencies" ); - ScopedDependencyMap::rebuild_map_unchecked(&self.workspace_id, db).await?; + ScopedDependencyMap::rebuild_map_unchecked(&self.workspace_id, &db).await?; // Mark as rebuilt by creating the setting windmill_common::global_settings::set_value_in_global_settings( - db, + &db, &setting_name, serde_json::json!({}), ) @@ -114,10 +112,33 @@ impl NewWorkspaceDependencies { .await?; tx.commit().await?; - if prev_description.is_none() && self.name.is_none() { + trigger_dependents_to_recompute_dependencies_in_the_background( + prev_description.is_none() && self.name.is_none(), + self.workspace_id, + self.language, + metadata, + path, + db, + ) + .await; + + Ok(new_id) + } +} + +pub async fn trigger_dependents_to_recompute_dependencies_in_the_background( + wait_for_cache_timeout: bool, + workspace_id: String, + language: ScriptLang, + (email, permissioned_as, created_by): (String, String, String), + path: String, + db: sqlx::Pool, +) { + tokio::spawn(async move { + if wait_for_cache_timeout { tracing::debug!( - workspace_id = %self.workspace_id, - language = ?self.language, + workspace_id = %workspace_id, + language = ?language, "waiting for cache timeout after creating first unnamed workspace dependencies" ); // Wait for cache timeout. @@ -127,27 +148,51 @@ impl NewWorkspaceDependencies { tokio::time::sleep(EXISTS_CACHE_TIMEOUT).await; } - // It's ok to fail, it will return an error and user will get notified that they should redeploy workspace dependencies - trigger_dependents_to_recompute_dependencies( - &self.workspace_id, - crate::scoped_dependency_map::ScopedDependencyMap::get_dependents( - path.as_str(), - &self.workspace_id, - db, - ) - .await?, - None, - None, - email, - created_by, - permissioned_as, - db, - vec![], - ) - .await?; + tracing::error!( + workspace_id = %workspace_id, + path = %path, + "CRITICAL: failed to get dependents for workspace dependencies - dependent runnables are not being redeployed. Please contact the Windmill team" + ); - Ok(new_id) - } + // It's ok to fail, it will return an error and user will get notified that they should redeploy workspace dependencies + if let Err(e) = trigger_dependents_to_recompute_dependencies( + &workspace_id, + match crate::scoped_dependency_map::ScopedDependencyMap::get_dependents( + path.as_str(), + &workspace_id, + &db, + ) + .await + { + Ok(importers) => importers, + Err(e) => { + tracing::error!( + workspace_id = %workspace_id, + path = %path, + error = %e, + "CRITICAL: failed to get dependents for workspace dependencies - dependent runnables are not being redeployed. Please contact the Windmill team" + ); + return; + } + }, + None, + None, + email.as_str(), + created_by.as_str(), + permissioned_as.as_str(), + &db, + vec![], + ) + .await + { + tracing::error!( + workspace_id = %workspace_id, + path = %path, + error = %e, + "CRITICAL: failed to trigger dependents to recompute dependencies - dependent runnables are not being redeployed. Please contact the Windmill team" + ); + } + }); } // Type aliases for backward compatibility