diff --git a/backend/Cargo.lock b/backend/Cargo.lock index 3cc24512aa..99ef2c4c4a 100644 --- a/backend/Cargo.lock +++ b/backend/Cargo.lock @@ -16993,6 +16993,7 @@ name = "windmill-oauth" version = "1.683.1" dependencies = [ "anyhow", + "arc-swap", "async-oauth2", "axum 0.8.4", "base64 0.22.1", diff --git a/backend/ee-repo-ref.txt b/backend/ee-repo-ref.txt index cca18546c1..dc01a40e29 100644 --- a/backend/ee-repo-ref.txt +++ b/backend/ee-repo-ref.txt @@ -1 +1 @@ -be2f3d4d11bb7110200524d7157caab3aac53996 \ No newline at end of file +e587df81d31638259efec13d0e68e54c890fdc2e \ No newline at end of file diff --git a/backend/src/main.rs b/backend/src/main.rs index 6468ef3072..a7e3a8313f 100644 --- a/backend/src/main.rs +++ b/backend/src/main.rs @@ -964,7 +964,7 @@ Windmill Community Edition {GIT_VERSION} std::process::exit(1); } } - let valid_key = *LICENSE_KEY_VALID.read().await; + let valid_key = LICENSE_KEY_VALID.load(std::sync::atomic::Ordering::Relaxed); if !valid_key && !server_mode { tracing::error!("Invalid license key, workers require a valid license key"); } @@ -974,7 +974,7 @@ Windmill Community Edition {GIT_VERSION} let renewed_now = maybe_renew_license_key_on_start( &HTTP_CLIENT, &db, - !valid_key && !LICENSE_KEY_ID.read().await.is_empty(), + !valid_key && !LICENSE_KEY_ID.load().is_empty(), ) .await; if renewed_now { diff --git a/backend/src/monitor.rs b/backend/src/monitor.rs index af2a66031c..e403ced388 100644 --- a/backend/src/monitor.rs +++ b/backend/src/monitor.rs @@ -263,7 +263,6 @@ pub async fn initial_load( } Connection::Http(_) => { // TODO: reload worker config from http - let mut config = WORKER_CONFIG.write().await; let worker_tags = DECODED_AGENT_TOKEN .as_ref() .map(|x| x.tags.clone()) @@ -271,7 +270,7 @@ pub async fn initial_load( // we only check from env as native_mode is not stored in the token // NATIVE_MODE_RESOLVED is already set in main.rs during startup let native_mode = windmill_common::worker::is_native_mode_from_env(); - *config = WorkerConfig { + WORKER_CONFIG.store(std::sync::Arc::new(WorkerConfig { worker_tags, env_vars: load_env_vars( load_whitelist_env_vars_from_env(), @@ -287,7 +286,7 @@ pub async fn initial_load( additional_python_paths: None, pip_local_dependencies: None, native_mode, - }; + })); } } } @@ -491,12 +490,10 @@ pub async fn load_tag_per_workspace_workspaces(db: &DB) -> error::Result<()> { .filter_map(|x| x.as_str()) .map(|x| x.to_string()) .collect::>(); - let mut w = DEFAULT_TAGS_WORKSPACES.write().await; - *w = Some(workspaces); + DEFAULT_TAGS_WORKSPACES.store(std::sync::Arc::new(Some(workspaces))); } Ok(None) => { - let mut w = DEFAULT_TAGS_WORKSPACES.write().await; - *w = None; + DEFAULT_TAGS_WORKSPACES.store(std::sync::Arc::new(None)); } _ => (), }; @@ -1137,7 +1134,7 @@ pub async fn delete_expired_items(db: &DB) -> () { ), } - let job_retention_secs = *JOB_RETENTION_SECS.read().await; + let job_retention_secs = JOB_RETENTION_SECS.load(std::sync::atomic::Ordering::Relaxed); if job_retention_secs > 0 { let batch_size = *JOB_CLEANUP_BATCH_SIZE; let max_batches = *JOB_CLEANUP_MAX_BATCHES; @@ -1442,7 +1439,8 @@ async fn delete_log_files_from_disk_and_store( // API, up to 1000 objects per request). #[cfg(feature = "parquet")] { - let should_del_from_store = *MONITOR_LOGS_ON_OBJECT_STORE.read().await; + let should_del_from_store = + MONITOR_LOGS_ON_OBJECT_STORE.load(std::sync::atomic::Ordering::Relaxed); if should_del_from_store { if let Some(os) = windmill_object_store::get_object_store().await { let s3_paths: Vec<_> = paths_to_delete @@ -1491,11 +1489,13 @@ pub async fn reload_instance_events_webhook_setting(db: &DB) { let value = load_value_from_global_settings(db, INSTANCE_EVENTS_WEBHOOK_SETTING).await; match value { Ok(Some(serde_json::Value::String(s))) if !s.is_empty() => { - *INSTANCE_EVENTS_WEBHOOK.write().await = Some(s); + INSTANCE_EVENTS_WEBHOOK.store(std::sync::Arc::new(Some(s))); } Ok(None) | Ok(Some(serde_json::Value::Null)) | Ok(Some(serde_json::Value::String(_))) => { // Fall back to env var if DB has no value - *INSTANCE_EVENTS_WEBHOOK.write().await = std::env::var("INSTANCE_EVENTS_WEBHOOK").ok(); + INSTANCE_EVENTS_WEBHOOK.store(std::sync::Arc::new( + std::env::var("INSTANCE_EVENTS_WEBHOOK").ok(), + )); } Err(e) => { tracing::error!("Error loading instance_events_webhook setting: {e:#}"); @@ -1732,57 +1732,55 @@ pub async fn reload_workspace_registries_setting(conn: &Connection) { } pub async fn reload_hub_api_secret_setting(conn: &Connection) { - reload_option_setting_with_tracing( - conn, - HUB_API_SECRET_SETTING, - "HUB_API_SECRET", - HUB_API_SECRET.clone(), - ) - .await; + match load_option_setting_value::(conn, HUB_API_SECRET_SETTING, "HUB_API_SECRET").await + { + Ok(v) => HUB_API_SECRET.store(std::sync::Arc::new(v)), + Err(e) => tracing::error!("Error reloading setting HUB_API_SECRET: {:?}", e), + } } pub async fn reload_retention_period_setting(conn: &Connection) { - if let Err(e) = reload_setting( + match load_setting_value::( conn, RETENTION_PERIOD_SECS_SETTING, "JOB_RETENTION_SECS", 60 * 60 * 24 * 30, - JOB_RETENTION_SECS.clone(), |x| x, ) .await { - tracing::error!("Error reloading retention period: {:?}", e) + Ok(v) => JOB_RETENTION_SECS.store(v, Ordering::Relaxed), + Err(e) => tracing::error!("Error reloading retention period: {:?}", e), } } pub async fn reload_audit_log_retention_days_setting(conn: &Connection) { - if let Err(e) = reload_setting( + match load_setting_value::( conn, AUDIT_LOG_RETENTION_DAYS_SETTING, "AUDIT_LOG_RETENTION_DAYS", 0, // 0 means use default: 365 for EE, 14 for CE - AUDIT_LOG_RETENTION_DAYS.clone(), |x| x, ) .await { - tracing::error!("Error reloading audit log retention days: {:?}", e) + Ok(v) => AUDIT_LOG_RETENTION_DAYS.store(v, Ordering::Relaxed), + Err(e) => tracing::error!("Error reloading audit log retention days: {:?}", e), } } pub async fn reload_delete_logs_periodically_setting(conn: &Connection) { - if let Err(e) = reload_setting( + match load_setting_value::( conn, MONITOR_LOGS_ON_OBJECT_STORE_SETTING, "MONITOR_LOGS_ON_OBJECT_STORE", false, - MONITOR_LOGS_ON_OBJECT_STORE.clone(), |x| x, ) .await { - tracing::error!("Error reloading retention period: {:?}", e) + Ok(v) => MONITOR_LOGS_ON_OBJECT_STORE.store(v, Ordering::Relaxed), + Err(e) => tracing::error!("Error reloading retention period: {:?}", e), } } @@ -1916,20 +1914,22 @@ pub async fn load_value_from_global_settings_with_conn( } } -pub async fn reload_option_setting( +/// Load an optional setting value without writing it anywhere. +/// +/// Extracted from [`reload_option_setting`] so callers that store the value +/// in something other than `Arc>>` (e.g. `ArcSwap>`, +/// an `AtomicBool`, etc.) can reuse the load pipeline. +pub async fn load_option_setting_value( conn: &Connection, setting_name: &str, std_env_var: &str, - lock: Arc>>, -) -> error::Result<()> { +) -> error::Result> { let force_value = std::env::var(format!("FORCE_{}", std_env_var)) .ok() .and_then(|x| x.parse::().ok()); if let Some(force_value) = force_value { - let mut l = lock.write().await; - *l = Some(force_value); - return Ok(()); + return Ok(Some(force_value)); } let q = load_value_from_global_settings_with_conn(conn, setting_name, true).await?; @@ -1947,14 +1947,24 @@ pub async fn reload_option_setting( } }; + if value.is_none() { + tracing::info!("Loaded {setting_name} setting to None"); + } + + Ok(value) +} + +pub async fn reload_option_setting( + conn: &Connection, + setting_name: &str, + std_env_var: &str, + lock: Arc>>, +) -> error::Result<()> { + let value = load_option_setting_value::(conn, setting_name, std_env_var).await?; { - if value.is_none() { - tracing::info!("Loaded {setting_name} setting to None"); - } let mut l = lock.write().await; *l = value; } - Ok(()) } @@ -1969,12 +1979,16 @@ pub async fn reload_url_list_setting_with_tracing( } } -pub async fn reload_url_list_setting( +/// Load an optional URL list setting without writing it anywhere. +/// +/// Extracted from [`reload_url_list_setting`] so callers that store the +/// value in something other than `Arc>>>` (e.g. +/// `ArcSwap>>`) can reuse the parsing pipeline. +pub async fn load_url_list_setting_value( conn: &Connection, setting_name: &str, std_env_var: &str, - lock: Arc>>>, -) -> error::Result<()> { +) -> error::Result>> { // Check for force environment variable if let Ok(force_value) = std::env::var(format!("FORCE_{}", std_env_var)) { let mut urls = Vec::new(); @@ -1989,9 +2003,7 @@ pub async fn reload_url_list_setting( } } } - let mut l = lock.write().await; - *l = if urls.is_empty() { None } else { Some(urls) }; - return Ok(()); + return Ok(if urls.is_empty() { None } else { Some(urls) }); } let q = load_value_from_global_settings_with_conn(conn, setting_name, true).await?; @@ -2041,25 +2053,39 @@ pub async fn reload_url_list_setting( } } + if value.is_none() { + tracing::info!("Loaded {} setting to None", setting_name); + } + + Ok(value) +} + +pub async fn reload_url_list_setting( + conn: &Connection, + setting_name: &str, + std_env_var: &str, + lock: Arc>>>, +) -> error::Result<()> { + let value = load_url_list_setting_value(conn, setting_name, std_env_var).await?; { - if value.is_none() { - tracing::info!("Loaded {} setting to None", setting_name); - } let mut l = lock.write().await; *l = value; } - Ok(()) } -pub async fn reload_setting( +/// Load a required setting value without writing it anywhere. +/// +/// Extracted from [`reload_setting`] so callers that store the value in +/// something other than `Arc>` (e.g. `AtomicI64`, `AtomicBool`, +/// `ArcSwap`) can reuse the load pipeline. +pub async fn load_setting_value( conn: &Connection, setting_name: &str, std_env_var: &str, default: T, - lock: Arc>, transformer: fn(T) -> T, -) -> error::Result<()> { +) -> error::Result { let q = load_value_from_global_settings_with_conn(conn, setting_name, true).await?; let mut value = std::env::var(std_env_var) @@ -2076,11 +2102,22 @@ pub async fn reload_setting( } }; + Ok(value) +} + +pub async fn reload_setting( + conn: &Connection, + setting_name: &str, + std_env_var: &str, + default: T, + lock: Arc>, + transformer: fn(T) -> T, +) -> error::Result<()> { + let value = load_setting_value(conn, setting_name, std_env_var, default, transformer).await?; { let mut l = lock.write().await; *l = value; } - Ok(()) } @@ -2620,9 +2657,8 @@ pub async fn reload_smtp_config(db: &Pool) { if let Err(e) = smtp_config { tracing::error!("Error reloading smtp config: {:?}", e) } else { - let mut wc = SMTP_CONFIG.write().await; tracing::info!("Reloading smtp config..."); - *wc = smtp_config.unwrap() + SMTP_CONFIG.store(std::sync::Arc::new(smtp_config.unwrap())); } } @@ -2631,9 +2667,8 @@ pub async fn reload_indexer_config(db: &Pool) { if let Err(e) = indexer_config { tracing::error!("Error reloading indexer config: {:?}", e) } else { - let mut wc = INDEXER_CONFIG.write().await; - tracing::info!("Reloading smtp config..."); - *wc = indexer_config.unwrap() + tracing::info!("Reloading indexer config..."); + INDEXER_CONFIG.store(std::sync::Arc::new(indexer_config.unwrap())); } } @@ -2642,29 +2677,29 @@ pub async fn reload_worker_config(db: &DB, tx: KillpillSender, kill_if_change: b if let Err(e) = config { tracing::error!("Error reloading worker config: {:?}", e) } else { - let wc = WORKER_CONFIG.read().await; + let wc = WORKER_CONFIG.load(); let config = config.unwrap(); let has_dedicated = config.dedicated_worker.is_some() || config .dedicated_workers .as_ref() .is_some_and(|dws| !dws.is_empty()); - if *wc != config || has_dedicated { + if **wc != config || has_dedicated { if kill_if_change { if has_dedicated - || (*wc).dedicated_worker != config.dedicated_worker - || (*wc).dedicated_workers != config.dedicated_workers + || wc.dedicated_worker != config.dedicated_worker + || wc.dedicated_workers != config.dedicated_workers { tracing::info!("Dedicated worker config changed, sending killpill. Expecting to be restarted by supervisor."); let _ = tx.send(); } - if (*wc).init_bash != config.init_bash { + if wc.init_bash != config.init_bash { tracing::info!("Init bash config changed, sending killpill. Expecting to be restarted by supervisor."); let _ = tx.send(); } - if (*wc).cache_clear != config.cache_clear { + if wc.cache_clear != config.cache_clear { tracing::info!("Cache clear changed, sending killpill. Expecting to be restarted by supervisor."); let _ = tx.send(); tracing::info!("Waiting 5 seconds to allow others workers to start potential jobs that depend on a potential shared cache volume"); @@ -2674,29 +2709,27 @@ pub async fn reload_worker_config(db: &DB, tx: KillpillSender, kill_if_change: b } } - if (*wc).periodic_script_bash != config.periodic_script_bash { + if wc.periodic_script_bash != config.periodic_script_bash { tracing::info!("Periodic script bash config changed, sending killpill. Expecting to be restarted by supervisor."); let _ = tx.send(); } - if (*wc).periodic_script_interval_seconds != config.periodic_script_interval_seconds - { + if wc.periodic_script_interval_seconds != config.periodic_script_interval_seconds { tracing::info!("Periodic script interval config changed, sending killpill. Expecting to be restarted by supervisor."); let _ = tx.send(); } - if (*wc).native_mode != config.native_mode { + if wc.native_mode != config.native_mode { tracing::info!("Native mode config changed, sending killpill. Expecting to be restarted by supervisor."); let _ = tx.send(); } } drop(wc); - let mut wc = WORKER_CONFIG.write().await; tracing::info!("Reloading worker config..."); store_suspended_pull_query(&config).await; store_pull_query(&config).await; - *wc = config + WORKER_CONFIG.store(std::sync::Arc::new(config)); } } } @@ -2725,10 +2758,7 @@ pub async fn load_base_url(conn: &Connection) -> error::Result { } else { std_base_url }; - { - let mut l = BASE_URL.write().await; - *l = base_url.clone(); - } + BASE_URL.store(std::sync::Arc::new(base_url.clone())); Ok(base_url) } @@ -2759,17 +2789,14 @@ pub async fn reload_base_url_setting(conn: &Connection) -> error::Result<()> { #[cfg(feature = "oauth2")] { if let Some(db) = conn.as_sql() { - let mut l = windmill_api::OAUTH_CLIENTS.write().await; - *l = windmill_api::oauth2_oss::build_oauth_clients(&base_url, oauths, db).await - .map_err(|e| tracing::error!("Error building oauth clients (is the oauth.json mounted and in correct format? Use '{}' as minimal oauth.json): {}", "{}", e)) - .unwrap(); + let clients = windmill_api::oauth2_oss::build_oauth_clients(&base_url, oauths, db).await + .map_err(|e| tracing::error!("Error building oauth clients (is the oauth.json mounted and in correct format? Use '{}' as minimal oauth.json): {}", "{}", e)) + .unwrap(); + windmill_api::OAUTH_CLIENTS.store(std::sync::Arc::new(clients)); } } - { - let mut l = IS_SECURE.write().await; - *l = is_secure; - } + IS_SECURE.store(is_secure, Ordering::Relaxed); Ok(()) } @@ -2898,7 +2925,7 @@ async fn handle_zombie_jobs(db: &Pool, base_internal_url: &str, node_n otel_incr_zombie_restart_count(restarted.len() as u64); - let base_url = BASE_URL.read().await.clone(); + let base_url = (**BASE_URL.load()).clone(); for r in restarted { let last_ping = if let Some(x) = r.ping { format!("last ping at {x}") @@ -3385,7 +3412,7 @@ async fn handle_zombie_flows(db: &DB) -> error::Result<()> { let id = flow.id.clone(); let last_ping = flow.last_ping.clone(); let now = now_from_db(db).await?; - let base_url = BASE_URL.read().await; + let base_url = BASE_URL.load(); let workspace_id = flow.workspace_id.clone(); let fmt_mb = |b: i64| format!("{:.1} MB", b as f64 / 1024.0 / 1024.0); @@ -3614,11 +3641,11 @@ pub async fn reload_hub_base_url_setting( DEFAULT_HUB_BASE_URL.to_string() }; - let mut l = HUB_BASE_URL.write().await; + let previous = HUB_BASE_URL.load(); if server_mode { #[cfg(feature = "embedding")] if let Some(db) = conn.as_sql() { - if *l != base_url { + if **previous != base_url { let disable_embedding = std::env::var("DISABLE_EMBEDDING") .ok() .map(|x| x.parse::().unwrap_or(false)) @@ -3632,7 +3659,8 @@ pub async fn reload_hub_base_url_setting( } } } - *l = base_url; + drop(previous); + HUB_BASE_URL.store(std::sync::Arc::new(base_url)); Ok(()) } @@ -3655,8 +3683,7 @@ pub async fn reload_critical_error_channels_setting(conn: &DB) -> error::Result< vec![] }; - let mut l = CRITICAL_ERROR_CHANNELS.write().await; - *l = critical_error_channels; + CRITICAL_ERROR_CHANNELS.store(std::sync::Arc::new(critical_error_channels)); Ok(()) } @@ -3678,9 +3705,7 @@ pub async fn reload_app_workspaced_route_setting(conn: &DB) -> error::Result<()> } }; - let mut l = APP_WORKSPACED_ROUTE.write().await; - - *l = ws_route; + APP_WORKSPACED_ROUTE.store(ws_route, Ordering::Relaxed); Ok(()) } @@ -3701,18 +3726,13 @@ pub async fn reload_http_route_workspaced_route_setting(conn: &DB) -> error::Res } }; - let mut l = HTTP_ROUTE_WORKSPACED_ROUTE.write().await; - - if *l != ws_route { - *l = ws_route; - drop(l); + let previous = HTTP_ROUTE_WORKSPACED_ROUTE.swap(ws_route, Ordering::Relaxed); + if previous != ws_route { // Bump the HTTP trigger version so the route cache is rebuilt with // the updated workspaced_route behavior on the next request. sqlx::query!("SELECT nextval('http_trigger_version_seq')") .fetch_one(conn) .await?; - } else { - *l = ws_route; } Ok(()) } @@ -3744,8 +3764,7 @@ pub async fn reload_critical_alerts_on_db_oversize(conn: &DB) -> error::Result<( None }; - let mut l = CRITICAL_ALERTS_ON_DB_OVERSIZE.write().await; - *l = db_oversize; + CRITICAL_ALERTS_ON_DB_OVERSIZE.store(std::sync::Arc::new(db_oversize)); Ok(()) } @@ -3776,8 +3795,7 @@ pub async fn reload_jwt_secret_setting(db: &DB) -> error::Result<()> { generate_and_save_jwt_secret(db).await? }; - let mut l = JWT_SECRET.write().await; - *l = jwt_secret; + JWT_SECRET.store(std::sync::Arc::new(jwt_secret)); Ok(()) } @@ -3899,7 +3917,7 @@ RETURNING job_id } async fn audit_log_retention_days() -> i64 { - let v = *AUDIT_LOG_RETENTION_DAYS.read().await; + let v = AUDIT_LOG_RETENTION_DAYS.load(std::sync::atomic::Ordering::Relaxed); if v > 0 { v } else if cfg!(feature = "enterprise") { diff --git a/backend/tests/dedicated_workers.rs b/backend/tests/dedicated_workers.rs index a986c0637a..13fb043454 100644 --- a/backend/tests/dedicated_workers.rs +++ b/backend/tests/dedicated_workers.rs @@ -303,7 +303,7 @@ mod dedicated_worker_tests { initialize_tracing().await; // Reset dedicated_workers in WORKER_CONFIG to avoid pollution from previous tests. { - let mut wc = windmill_common::worker::WORKER_CONFIG.write().await; + let mut wc = (**windmill_common::worker::WORKER_CONFIG.load()).clone(); wc.dedicated_worker = None; wc.dedicated_workers = None; wc.worker_tags = windmill_common::worker::DEFAULT_TAGS.clone(); @@ -313,6 +313,7 @@ mod dedicated_worker_tests { }]; windmill_common::worker::store_suspended_pull_query(&wc).await; windmill_common::worker::store_pull_query(&wc).await; + windmill_common::worker::WORKER_CONFIG.store(std::sync::Arc::new(wc)); } let server = ApiServer::start(db.clone()).await?; let port = server.addr.port(); @@ -717,7 +718,7 @@ mod dedicated_worker_tests { initialize_tracing().await; // Reset dedicated_workers in WORKER_CONFIG to avoid pollution from previous tests. { - let mut wc = windmill_common::worker::WORKER_CONFIG.write().await; + let mut wc = (**windmill_common::worker::WORKER_CONFIG.load()).clone(); wc.dedicated_worker = None; wc.dedicated_workers = None; wc.worker_tags = windmill_common::worker::DEFAULT_TAGS.clone(); @@ -727,6 +728,7 @@ mod dedicated_worker_tests { }]; windmill_common::worker::store_suspended_pull_query(&wc).await; windmill_common::worker::store_pull_query(&wc).await; + windmill_common::worker::WORKER_CONFIG.store(std::sync::Arc::new(wc)); } let server = ApiServer::start(db.clone()).await?; let port = server.addr.port(); diff --git a/backend/tests/job_payload.rs b/backend/tests/job_payload.rs index 393dca7238..83cba72b73 100644 --- a/backend/tests/job_payload.rs +++ b/backend/tests/job_payload.rs @@ -276,19 +276,19 @@ mod job_payload { #[sqlx::test(fixtures("base", "hello"))] async fn test_dependencies_payload_min_1_427(db: Pool) -> anyhow::Result<()> { - *MIN_VERSION.write().await = MIN_VERSION_IS_AT_LEAST_1_427.version().clone(); + MIN_VERSION.store(std::sync::Arc::new(MIN_VERSION_IS_AT_LEAST_1_427.version().clone())); test_dependencies_payload(db).await?; Ok(()) } #[sqlx::test(fixtures("base", "hello"))] async fn test_dependencies_payload_min_1_432(db: Pool) -> anyhow::Result<()> { - *MIN_VERSION.write().await = MIN_VERSION_IS_AT_LEAST_1_432.version().clone(); + MIN_VERSION.store(std::sync::Arc::new(MIN_VERSION_IS_AT_LEAST_1_432.version().clone())); test_dependencies_payload(db).await?; Ok(()) } #[sqlx::test(fixtures("base", "hello"))] async fn test_dependencies_payload_min_1_440(db: Pool) -> anyhow::Result<()> { - *MIN_VERSION.write().await = MIN_VERSION_IS_AT_LEAST_1_440.version().clone(); + MIN_VERSION.store(std::sync::Arc::new(MIN_VERSION_IS_AT_LEAST_1_440.version().clone())); test_dependencies_payload(db).await?; Ok(()) } diff --git a/backend/tests/nativets_stress.rs b/backend/tests/nativets_stress.rs index a175d97af4..8b0884a3ea 100644 --- a/backend/tests/nativets_stress.rs +++ b/backend/tests/nativets_stress.rs @@ -223,7 +223,7 @@ fn spawn_workers( let future = async move { let base_internal_url = format!("http://localhost:{}", port); { - let mut wc = WORKER_CONFIG.write().await; + let mut wc = (**WORKER_CONFIG.load()).clone(); wc.worker_tags = windmill_common::worker::DEFAULT_TAGS.clone(); wc.priority_tags_sorted = vec![windmill_common::worker::PriorityTags { priority: 0, @@ -231,6 +231,7 @@ fn spawn_workers( }]; windmill_common::worker::store_suspended_pull_query(&wc).await; windmill_common::worker::store_pull_query(&wc).await; + WORKER_CONFIG.store(std::sync::Arc::new(wc)); } windmill_worker::run_worker( &conn, diff --git a/backend/tests/worker.rs b/backend/tests/worker.rs index 42ff589bcd..15f961211b 100644 --- a/backend/tests/worker.rs +++ b/backend/tests/worker.rs @@ -4034,19 +4034,16 @@ async fn test_flow_substep_tag_availability_check(db: Pool) -> anyhow: let server = ApiServer::start(db.clone()).await?; // Set up a restricted tag that is only available to "other-workspace" (not "test-workspace") - { - let mut custom_tags = CUSTOM_TAGS_PER_WORKSPACE.write().await; - *custom_tags = CustomTags { - global: vec![], - specific: std::collections::HashMap::from([( - "restricted-tag".to_string(), - SpecificTagData { - tag_type: SpecificTagType::NoneExcept, - workspaces: vec!["other-workspace".to_string()], - }, - )]), - }; - } + CUSTOM_TAGS_PER_WORKSPACE.store(std::sync::Arc::new(CustomTags { + global: vec![], + specific: std::collections::HashMap::from([( + "restricted-tag".to_string(), + SpecificTagData { + tag_type: SpecificTagType::NoneExcept, + workspaces: vec!["other-workspace".to_string()], + }, + )]), + })); // Create a flow with a substep that uses the restricted tag let flow: FlowValue = serde_json::from_value(serde_json::json!({ @@ -4088,10 +4085,7 @@ async fn test_flow_substep_tag_availability_check(db: Pool) -> anyhow: ); // Clean up: reset custom tags - { - let mut custom_tags = CUSTOM_TAGS_PER_WORKSPACE.write().await; - *custom_tags = CustomTags::default(); - } + CUSTOM_TAGS_PER_WORKSPACE.store(std::sync::Arc::new(CustomTags::default())); Ok(()) } diff --git a/backend/windmill-api-embeddings/src/lib.rs b/backend/windmill-api-embeddings/src/lib.rs index 1d16facfff..ae8eca67cf 100644 --- a/backend/windmill-api-embeddings/src/lib.rs +++ b/backend/windmill-api-embeddings/src/lib.rs @@ -259,7 +259,7 @@ impl EmbeddingsDb { self.db .create_collection("resource_types".to_string(), 384, Distance::Cosine)?; - let hub_base_url = HUB_BASE_URL.read().await.clone(); + let hub_base_url = (**HUB_BASE_URL.load()).clone(); let response = match hub_base_url.as_str() { DEFAULT_HUB_BASE_URL => { diff --git a/backend/windmill-api-flows/src/flows.rs b/backend/windmill-api-flows/src/flows.rs index 92ff7f5fa3..f099f3b6e8 100644 --- a/backend/windmill-api-flows/src/flows.rs +++ b/backend/windmill-api-flows/src/flows.rs @@ -224,7 +224,7 @@ async fn list_hub_flows(Extension(db): Extension) -> impl IntoResponse { &HTTP_CLIENT, &format!( "{}/searchFlowData?approved=true", - *HUB_BASE_URL.read().await + **HUB_BASE_URL.load() ), None, &db, @@ -257,7 +257,7 @@ pub async fn get_hub_flow_by_id( ) -> JsonResult> { let value = http_get_from_hub( &HTTP_CLIENT, - &format!("{}/flows/{}/json", *HUB_BASE_URL.read().await, id), + &format!("{}/flows/{}/json", **HUB_BASE_URL.load(), id), false, None, Some(&db), diff --git a/backend/windmill-api-integration-tests/tests/token_hash.rs b/backend/windmill-api-integration-tests/tests/token_hash.rs index aa7dd22286..6f4ba6b0ac 100644 --- a/backend/windmill-api-integration-tests/tests/token_hash.rs +++ b/backend/windmill-api-integration-tests/tests/token_hash.rs @@ -222,7 +222,7 @@ async fn test_plaintext_backward_compat(db: Pool) -> anyhow::Result<() // Set MIN_VERSION to one minor below the token hash feature version let mut old_version = MIN_VERSION_SUPPORTS_TOKEN_HASH.version().clone(); old_version.minor -= 1; - *MIN_VERSION.write().await = old_version; + MIN_VERSION.store(std::sync::Arc::new(old_version)); let resp = authed(client().post(format!("{base}/tokens/create"))) .json(&json!({"label": "old-worker-compat-token"})) @@ -274,7 +274,7 @@ async fn test_plaintext_backward_compat(db: Pool) -> anyhow::Result<() ); // --- Phase 2: All workers upgraded (version >= 1.650.0) --- - *MIN_VERSION.write().await = MIN_VERSION_SUPPORTS_TOKEN_HASH.version().clone(); + MIN_VERSION.store(std::sync::Arc::new(MIN_VERSION_SUPPORTS_TOKEN_HASH.version().clone())); let resp = authed(client().post(format!("{base}/tokens/create"))) .json(&json!({"label": "new-worker-token"})) diff --git a/backend/windmill-api-jobs/src/execution.rs b/backend/windmill-api-jobs/src/execution.rs index f98bf89c36..555d7a4d8b 100644 --- a/backend/windmill-api-jobs/src/execution.rs +++ b/backend/windmill-api-jobs/src/execution.rs @@ -65,7 +65,7 @@ pub async fn check_tag_available_for_workspace( pub async fn check_license_key_valid() -> error::Result<()> { use windmill_common::ee_oss::LICENSE_KEY_VALID; - let valid = *LICENSE_KEY_VALID.read().await; + let valid = LICENSE_KEY_VALID.load(std::sync::atomic::Ordering::Relaxed); if !valid { return Err(error::Error::BadRequest( "License key is not valid. Go to your superadmin settings to update your license key." diff --git a/backend/windmill-api-scripts/src/scripts.rs b/backend/windmill-api-scripts/src/scripts.rs index 5f2560fdbc..ae0b1dcc70 100644 --- a/backend/windmill-api-scripts/src/scripts.rs +++ b/backend/windmill-api-scripts/src/scripts.rs @@ -464,7 +464,7 @@ async fn get_top_hub_scripts( let (status_code, headers, response) = query_elems_from_hub( &HTTP_CLIENT, - &format!("{}/scripts/top", *HUB_BASE_URL.read().await), + &format!("{}/scripts/top", **HUB_BASE_URL.load()), Some(query_params), &db, ) @@ -1444,7 +1444,7 @@ pub async fn pick_hub_script_by_path( // Extract version_id from path (format: {hub}/{version_id}/{summary}) let version_id = path_str.split('/').nth(1).unwrap_or(""); - let hub_base_url = HUB_BASE_URL.read().await.clone(); + let hub_base_url = (**HUB_BASE_URL.load()).clone(); // Determine which hub to use based on version_id // If version_id < PRIVATE_HUB_MIN_VERSION, use default hub diff --git a/backend/windmill-api-settings/src/log_cleanup.rs b/backend/windmill-api-settings/src/log_cleanup.rs index f7c81227cd..890b46ba9e 100644 --- a/backend/windmill-api-settings/src/log_cleanup.rs +++ b/backend/windmill-api-settings/src/log_cleanup.rs @@ -293,7 +293,7 @@ async fn cleanup_job_logs( db: &DB, store: &Arc, ) -> error::Result<()> { - let retention_secs = *JOB_RETENTION_SECS.read().await; + let retention_secs = JOB_RETENTION_SECS.load(std::sync::atomic::Ordering::Relaxed); if retention_secs <= 0 { return Ok(()); } @@ -465,7 +465,7 @@ async fn cleanup_s3_orphans( db: &DB, store: &Arc, ) -> error::Result<()> { - let job_retention_secs = *JOB_RETENTION_SECS.read().await; + let job_retention_secs = JOB_RETENTION_SECS.load(std::sync::atomic::Ordering::Relaxed); let now = Utc::now(); // Service logs always have a retention (hardcoded SERVICE_LOG_RETENTION_SECS), // so we scan for service-log orphans regardless of JOB_RETENTION_SECS. Job-log diff --git a/backend/windmill-api-users/src/users.rs b/backend/windmill-api-users/src/users.rs index f346fb5123..35c7ce37fd 100644 --- a/backend/windmill-api-users/src/users.rs +++ b/backend/windmill-api-users/src/users.rs @@ -651,7 +651,7 @@ async fn is_valid_logout_redirect(rd: &str) -> bool { if host == "windmill.dev" || host.ends_with(".windmill.dev") { return true; } - let hub_url = HUB_BASE_URL.read().await.clone(); + let hub_url = (**HUB_BASE_URL.load()).clone(); if let Ok(hub_parsed) = url::Url::parse(&hub_url) { if let Some(hub_host) = hub_parsed.host_str() { if host == hub_host { @@ -2071,7 +2071,7 @@ pub async fn create_session_token<'c>( .await?; let mut cookie = Cookie::new(COOKIE_NAME, token.clone()); - cookie.set_secure(IS_SECURE.read().await.clone()); + cookie.set_secure(IS_SECURE.load(std::sync::atomic::Ordering::Relaxed)); cookie.set_same_site(Some(tower_cookies::cookie::SameSite::Lax)); cookie.set_http_only(true); cookie.set_path(COOKIE_PATH); @@ -2202,7 +2202,7 @@ async fn exit_impersonation( Json(req): Json, ) -> Result { let mut cookie = tower_cookies::Cookie::new(COOKIE_NAME, req.token); - cookie.set_secure(IS_SECURE.read().await.clone()); + cookie.set_secure(IS_SECURE.load(std::sync::atomic::Ordering::Relaxed)); cookie.set_same_site(Some(tower_cookies::cookie::SameSite::Lax)); cookie.set_http_only(true); cookie.set_path(COOKIE_PATH); @@ -2655,7 +2655,7 @@ async fn request_password_reset( .await?; // Get the base URL for the reset link - let base_url = BASE_URL.read().await.clone(); + let base_url = (**BASE_URL.load()).clone(); let base_url = if base_url.is_empty() { std::env::var("BASE_URL").unwrap_or_else(|_| "http://localhost".to_string()) } else { diff --git a/backend/windmill-api-workers/src/lib.rs b/backend/windmill-api-workers/src/lib.rs index d4a6225e9d..dc6b08db08 100644 --- a/backend/windmill-api-workers/src/lib.rs +++ b/backend/windmill-api-workers/src/lib.rs @@ -158,7 +158,7 @@ async fn exists_workers_with_tags( if !has_devops_role { if let Some(ref workspace) = tags_query.workspace { // Filter to only tags visible in this workspace - let custom_tags = CUSTOM_TAGS_PER_WORKSPACE.read().await; + let custom_tags = CUSTOM_TAGS_PER_WORKSPACE.load(); let allowed_tags = custom_tags.to_string_vec(Some(workspace.clone())); tags.retain(|t| allowed_tags.contains(t)); } else { @@ -203,7 +203,7 @@ async fn get_custom_tags( Query(query): Query, ) -> JsonResult> { if query.show_workspace_restriction.is_some_and(|x| x) { - let tags_o = CUSTOM_TAGS_PER_WORKSPACE.read().await; + let tags_o = CUSTOM_TAGS_PER_WORKSPACE.load(); let all_tags = tags_o.to_string_vec(None); return Ok(Json(all_tags)); } @@ -213,14 +213,14 @@ async fn get_custom_tags( return Ok(Json(vec![])); } } - Ok(Json(ALL_TAGS.read().await.clone().into())) + Ok(Json((**ALL_TAGS.load()).clone().into())) } async fn get_custom_tags_for_workspace( _authed: ApiAuthed, Path(w_id): Path, ) -> JsonResult> { - let tags_o = CUSTOM_TAGS_PER_WORKSPACE.read().await; + let tags_o = CUSTOM_TAGS_PER_WORKSPACE.load(); let all_tags = tags_o.to_string_vec(Some(w_id)); Ok(Json(all_tags)) } diff --git a/backend/windmill-api-workspaces/src/deployment_requests.rs b/backend/windmill-api-workspaces/src/deployment_requests.rs index 9701852ea2..7782382073 100644 --- a/backend/windmill-api-workspaces/src/deployment_requests.rs +++ b/backend/windmill-api-workspaces/src/deployment_requests.rs @@ -339,7 +339,7 @@ async fn create_deployment_request( tx.commit().await?; // Send a deployment-request email to each assignee. - let base_url = BASE_URL.read().await.clone(); + let base_url = (**BASE_URL.load()).clone(); let subject = format!( "[Windmill] @{} requested a deployment on fork {w_id}", authed.username @@ -542,7 +542,7 @@ async fn close_deployment_request_merged( recipients.insert(row.requested_by_email); recipients.remove(&authed.email); let subject = format!("[Windmill] Deployment request on fork {w_id} merged"); - let base_url = BASE_URL.read().await.clone(); + let base_url = (**BASE_URL.load()).clone(); let body_text = format!( "@{} merged the deployment request from @{} on fork {w_id}.\n\n{base_url}/?workspace={w_id}", authed.username, row.requested_by @@ -692,7 +692,7 @@ async fn create_deployment_request_comment( "[Windmill] New comment on deployment request for fork {w_id} by @{}", authed.username ); - let base_url = BASE_URL.read().await.clone(); + let base_url = (**BASE_URL.load()).clone(); let body_text = format!( "@{} commented on the deployment request for fork {w_id}:\n\n{}\n\n{base_url}/?workspace={w_id}", authed.username, body.body diff --git a/backend/windmill-api-workspaces/src/workspaces.rs b/backend/windmill-api-workspaces/src/workspaces.rs index 4deeb1ed6f..3d48caaaa1 100644 --- a/backend/windmill-api-workspaces/src/workspaces.rs +++ b/backend/windmill-api-workspaces/src/workspaces.rs @@ -4665,7 +4665,7 @@ async fn invite_user( "You have been granted access to Windmill's workspace {w_id} If you do not have an account on {}, login with SSO or ask an admin to create an account for you.", - BASE_URL.read().await.clone() + (**BASE_URL.load()).clone() ), &nu.email, ); @@ -4811,7 +4811,7 @@ async fn add_user( If you do not have an account on {}, login with SSO or ask an admin to create an account for you.", authed.email, - BASE_URL.read().await.clone() + (**BASE_URL.load()).clone() ), &nu.email, ); diff --git a/backend/windmill-api/src/apps.rs b/backend/windmill-api/src/apps.rs index 662df5dbc4..548bb34d39 100644 --- a/backend/windmill-api/src/apps.rs +++ b/backend/windmill-api/src/apps.rs @@ -767,7 +767,7 @@ async fn custom_path_exists( Extension(db): Extension, Path((w_id, custom_path)): Path<(String, String)>, ) -> JsonResult { - let as_workspaced_route = *APP_WORKSPACED_ROUTE.read().await; + let as_workspaced_route = APP_WORKSPACED_ROUTE.load(std::sync::atomic::Ordering::Relaxed); let exists = sqlx::query_scalar!( @@ -1275,7 +1275,7 @@ async fn create_app_internal<'a>( } if let Some(custom_path) = &app.custom_path { require_admin(authed.is_admin, &authed.username)?; - let as_workspaced_route = *APP_WORKSPACED_ROUTE.read().await; + let as_workspaced_route = APP_WORKSPACED_ROUTE.load(std::sync::atomic::Ordering::Relaxed); let exists = sqlx::query_scalar!( "SELECT EXISTS(SELECT 1 FROM app WHERE custom_path = $1 AND ($2::TEXT IS NULL OR workspace_id = $2))", @@ -1410,7 +1410,7 @@ async fn create_app_internal<'a>( async fn list_hub_apps(Extension(db): Extension) -> impl IntoResponse { let (status_code, headers, body) = query_elems_from_hub( &HTTP_CLIENT, - &format!("{}/searchUiData?approved=true", *HUB_BASE_URL.read().await), + &format!("{}/searchUiData?approved=true", **HUB_BASE_URL.load()), None, &db, ) @@ -1424,7 +1424,7 @@ pub async fn get_hub_app_by_id( ) -> JsonResult> { let value = http_get_from_hub( &HTTP_CLIENT, - &format!("{}/apps/{}/json", *HUB_BASE_URL.read().await, id), + &format!("{}/apps/{}/json", **HUB_BASE_URL.load(), id), false, None, Some(&db), @@ -1442,7 +1442,7 @@ pub async fn get_hub_raw_app_by_id( ) -> JsonResult> { let value = http_get_from_hub( &HTTP_CLIENT, - &format!("{}/raw_apps/{}/json", *HUB_BASE_URL.read().await, id), + &format!("{}/raw_apps/{}/json", **HUB_BASE_URL.load(), id), false, None, Some(&db), @@ -1772,7 +1772,7 @@ async fn update_app_internal<'a>( if let Some(ncustom_path) = &ns.custom_path { require_admin(authed.is_admin, &authed.username)?; - let as_workspaced_route = *APP_WORKSPACED_ROUTE.read().await; + let as_workspaced_route = APP_WORKSPACED_ROUTE.load(std::sync::atomic::Ordering::Relaxed); if ncustom_path.is_empty() { sqlb.set("custom_path", "NULL"); diff --git a/backend/windmill-api/src/integration.rs b/backend/windmill-api/src/integration.rs index 07a719b337..2def3e6ed0 100644 --- a/backend/windmill-api/src/integration.rs +++ b/backend/windmill-api/src/integration.rs @@ -22,7 +22,7 @@ async fn list_hub_integrations( let (status_code, headers, response) = query_elems_from_hub( &HTTP_CLIENT, - &format!("{}/integrations/list", *HUB_BASE_URL.read().await), + &format!("{}/integrations/list", **HUB_BASE_URL.load()), Some(query_params), &db, ) diff --git a/backend/windmill-api/src/jobs.rs b/backend/windmill-api/src/jobs.rs index ae8ad533b5..0b1dac753c 100644 --- a/backend/windmill-api/src/jobs.rs +++ b/backend/windmill-api/src/jobs.rs @@ -1278,7 +1278,7 @@ async fn send_workspace_trigger_failure_email_notification( (None, None) }; - let base_url = BASE_URL.read().await; + let base_url = BASE_URL.load(); let job_url = format!("{}/run/{}?workspace={}", base_url, &job_id, w_id); let trigger_kind_str = trigger_kind.unwrap_or("Unknown").to_string().to_uppercase(); @@ -3560,7 +3560,7 @@ pub async fn get_resume_urls_internal( .unwrap_or(target_job_id); let approval_token = generate_approval_token(&w_id, approval_target_id, &db).await?; - let base_url_str = BASE_URL.read().await.clone(); + let base_url_str = (**BASE_URL.load()).clone(); let base_url = base_url_str.as_str(); let res = ResumeUrls { approvalPage: format!( diff --git a/backend/windmill-api/src/lib.rs b/backend/windmill-api/src/lib.rs index 24b4574019..0a495dbd54 100644 --- a/backend/windmill-api/src/lib.rs +++ b/backend/windmill-api/src/lib.rs @@ -1139,8 +1139,8 @@ async fn list_workspace_labels( async fn ee_license() -> String { use windmill_common::ee_oss::{LICENSE_KEY_ID, LICENSE_KEY_VALID}; - if *LICENSE_KEY_VALID.read().await { - LICENSE_KEY_ID.read().await.clone() + if LICENSE_KEY_VALID.load(std::sync::atomic::Ordering::Relaxed) { + (**LICENSE_KEY_ID.load()).clone() } else { "".to_string() } diff --git a/backend/windmill-api/src/mcp/core.rs b/backend/windmill-api/src/mcp/core.rs index a20484e3eb..07ce233278 100644 --- a/backend/windmill-api/src/mcp/core.rs +++ b/backend/windmill-api/src/mcp/core.rs @@ -436,7 +436,7 @@ pub async fn add_www_authenticate_header( // Only add header to 401 Unauthorized responses if response.status() == StatusCode::UNAUTHORIZED { - let base_url = BASE_URL.read().await; + let base_url = BASE_URL.load(); // RFC 9728: The resource parameter contains the protected resource URL. // Clients derive the metadata URL by inserting /.well-known/oauth-protected-resource @@ -507,7 +507,7 @@ pub async fn add_www_authenticate_header_gateway( let response = next.run(request).await; if response.status() == StatusCode::UNAUTHORIZED { - let base_url = BASE_URL.read().await; + let base_url = BASE_URL.load(); let resource_url = format!("{}/api/mcp/gateway", base_url); let www_authenticate = format!("Bearer resource=\"{}\"", resource_url); diff --git a/backend/windmill-api/src/mcp/oauth_server.rs b/backend/windmill-api/src/mcp/oauth_server.rs index cd465e5b2c..48c4f73c44 100644 --- a/backend/windmill-api/src/mcp/oauth_server.rs +++ b/backend/windmill-api/src/mcp/oauth_server.rs @@ -266,7 +266,7 @@ fn build_protected_resource_metadata( pub async fn workspaced_oauth_metadata( Path(workspace_id): Path, ) -> Json { - let base_url = BASE_URL.read().await; + let base_url = BASE_URL.load(); let oauth_prefix = format!("/api/w/{}/mcp/oauth/server", workspace_id); Json(build_oauth_metadata(&oauth_prefix, &base_url)) } @@ -275,7 +275,7 @@ pub async fn workspaced_oauth_metadata( pub async fn protected_resource_metadata_by_path( Path(workspace_id): Path, ) -> Json { - let base_url = BASE_URL.read().await; + let base_url = BASE_URL.load(); let resource_path = format!("/api/mcp/w/{}/mcp", workspace_id); let oauth_prefix = format!("/api/w/{}/mcp/oauth/server", workspace_id); Json(build_protected_resource_metadata( @@ -700,7 +700,7 @@ async fn oauth_authorize_inner( None => ("gateway", "true"), }; - let base_url = BASE_URL.read().await; + let base_url = BASE_URL.load(); let frontend_url = format!( "{}/oauth/mcp_authorize?{}", base_url, @@ -897,7 +897,7 @@ impl IntoResponse for OAuthErrorRedirect { // Thin wrappers that delegate to the shared inner functions above. pub async fn gateway_oauth_metadata() -> Json { - let base_url = BASE_URL.read().await; + let base_url = BASE_URL.load(); Json(build_oauth_metadata( "/api/mcp/gateway/oauth/server", &base_url, @@ -905,7 +905,7 @@ pub async fn gateway_oauth_metadata() -> Json { } pub async fn gateway_protected_resource_metadata() -> Json { - let base_url = BASE_URL.read().await; + let base_url = BASE_URL.load(); Json(build_protected_resource_metadata( "/api/mcp/gateway", "/api/mcp/gateway/oauth/server", diff --git a/backend/windmill-api/src/mcp/utils.rs b/backend/windmill-api/src/mcp/utils.rs index 0361dd37e7..316cca4f14 100644 --- a/backend/windmill-api/src/mcp/utils.rs +++ b/backend/windmill-api/src/mcp/utils.rs @@ -203,7 +203,7 @@ pub async fn get_scripts_from_hub( ("with_schema", "true".to_string()), ("apps", scope_integrations.unwrap_or("").to_string()), ]); - let url = format!("{}/scripts/top", *HUB_BASE_URL.read().await); + let url = format!("{}/scripts/top", **HUB_BASE_URL.load()); let (_status_code, _headers, response) = query_elems_from_hub(&HTTP_CLIENT, &url, query_params, &db) .await diff --git a/backend/windmill-api/src/oauth2_oss.rs b/backend/windmill-api/src/oauth2_oss.rs index fc54ef1ba9..3eaf179634 100644 --- a/backend/windmill-api/src/oauth2_oss.rs +++ b/backend/windmill-api/src/oauth2_oss.rs @@ -103,7 +103,7 @@ async fn list_logins() -> error::JsonResult { #[cfg(all(feature = "oauth2", not(feature = "private")))] async fn list_connects() -> error::JsonResult> { Ok(Json( - (&OAUTH_CLIENTS.read().await.connects) + (&OAUTH_CLIENTS.load().connects) .keys() .map(|x| x.to_owned()) .collect_vec(), diff --git a/backend/windmill-common/src/apps.rs b/backend/windmill-common/src/apps.rs index 50657296c0..36b2d9b704 100644 --- a/backend/windmill-common/src/apps.rs +++ b/backend/windmill-common/src/apps.rs @@ -6,17 +6,16 @@ * LICENSE-AGPL for a copy of the license. */ -use std::sync::Arc; +use std::sync::atomic::AtomicBool; use serde_json::{from_value, Value}; -use tokio::sync::RwLock; use crate::{error, scripts::ScriptLang}; pub use windmill_types::apps::*; lazy_static::lazy_static! { - pub static ref APP_WORKSPACED_ROUTE: Arc> = Arc::new(RwLock::new(false)); + pub static ref APP_WORKSPACED_ROUTE: AtomicBool = AtomicBool::new(false); } /// Traverse FlowValue while invoking provided by caller callback on leafs diff --git a/backend/windmill-common/src/ee_oss.rs b/backend/windmill-common/src/ee_oss.rs index 93d0061ade..15d9e7cb5e 100644 --- a/backend/windmill-common/src/ee_oss.rs +++ b/backend/windmill-common/src/ee_oss.rs @@ -11,15 +11,13 @@ use crate::error; #[cfg(not(feature = "private"))] use serde::Deserialize; #[cfg(not(feature = "private"))] -use std::sync::Arc; -#[cfg(not(feature = "private"))] -use tokio::sync::RwLock; +use std::sync::atomic::AtomicBool; #[cfg(not(feature = "private"))] lazy_static::lazy_static! { - pub static ref LICENSE_KEY_VALID: Arc> = Arc::new(RwLock::new(true)); - pub static ref LICENSE_KEY_ID: Arc> = Arc::new(RwLock::new("".to_string())); - pub static ref LICENSE_KEY: Arc> = Arc::new(RwLock::new("".to_string())); + pub static ref LICENSE_KEY_VALID: AtomicBool = AtomicBool::new(true); + pub static ref LICENSE_KEY_ID: arc_swap::ArcSwap = arc_swap::ArcSwap::from_pointee("".to_string()); + pub static ref LICENSE_KEY: arc_swap::ArcSwap = arc_swap::ArcSwap::from_pointee("".to_string()); } #[cfg(not(feature = "private"))] diff --git a/backend/windmill-common/src/flows.rs b/backend/windmill-common/src/flows.rs index 01d714ea01..5903aa8ea6 100644 --- a/backend/windmill-common/src/flows.rs +++ b/backend/windmill-common/src/flows.rs @@ -72,7 +72,7 @@ pub async fn get_full_hub_flow_by_path( ) -> crate::error::Result { let path = path.to_path(); let flow_id = extract_hub_flow_id_from_path(&path)?; - let hub_base_url = HUB_BASE_URL.read().await.clone(); + let hub_base_url = (**HUB_BASE_URL.load()).clone(); let hub_url = format!("{hub_base_url}/flows/{flow_id}/json"); let response = match http_get_from_hub(http_client, &hub_url, false, None, db) diff --git a/backend/windmill-common/src/global_settings.rs b/backend/windmill-common/src/global_settings.rs index b0e99c4279..77cb3c0202 100644 --- a/backend/windmill-common/src/global_settings.rs +++ b/backend/windmill-common/src/global_settings.rs @@ -69,11 +69,10 @@ pub const WORKSPACE_REGISTRIES_SETTING: &str = "workspace_registries"; pub const RESTART_COORDINATION_SETTING: &str = "_restart_coordination"; pub const ALERT_CONFIG_SETTING: &str = "alert_job_queue_waiting"; -use std::sync::Arc; -use tokio::sync::RwLock; +use std::sync::atomic::AtomicBool; lazy_static::lazy_static! { - pub static ref HTTP_ROUTE_WORKSPACED_ROUTE: Arc> = Arc::new(RwLock::new(false)); + pub static ref HTTP_ROUTE_WORKSPACED_ROUTE: AtomicBool = AtomicBool::new(false); } pub const ENV_SETTINGS: &[&str] = &[ diff --git a/backend/windmill-common/src/jobs.rs b/backend/windmill-common/src/jobs.rs index 9972ac9fda..fd05c03ec0 100644 --- a/backend/windmill-common/src/jobs.rs +++ b/backend/windmill-common/src/jobs.rs @@ -339,7 +339,7 @@ pub async fn check_tag_available_for_workspace_internal( is_tag_in_scope_tags = Some(scope_tags.contains(&tag)); } - let custom_tags_per_w = CUSTOM_TAGS_PER_WORKSPACE.read().await; + let custom_tags_per_w = CUSTOM_TAGS_PER_WORKSPACE.load(); if custom_tags_per_w.global.contains(&tag.to_string()) { is_tag_in_workspace_custom_tags = true; } else if let Some(specific_tag) = custom_tags_per_w.specific.get(tag) { diff --git a/backend/windmill-common/src/jwt.rs b/backend/windmill-common/src/jwt.rs index 30e56cf155..cc2f1102bf 100644 --- a/backend/windmill-common/src/jwt.rs +++ b/backend/windmill-common/src/jwt.rs @@ -3,15 +3,14 @@ use base64::{engine::general_purpose::URL_SAFE_NO_PAD, Engine}; use hmac::{Hmac, Mac}; use serde::{de::DeserializeOwned, Serialize}; use sha2::Sha256; -use std::{collections::HashSet, sync::Arc}; -use tokio::sync::RwLock; +use std::collections::HashSet; lazy_static::lazy_static! { - pub static ref JWT_SECRET: Arc> = Arc::new(RwLock::new("".to_string())); + pub static ref JWT_SECRET: arc_swap::ArcSwap = arc_swap::ArcSwap::from_pointee("".to_string()); } pub async fn encode_with_internal_secret(claims: T) -> error::Result { - let jwt_secret = JWT_SECRET.read().await; + let jwt_secret = JWT_SECRET.load(); if jwt_secret.is_empty() { return Err(Error::internal_err("JWT secret is not set".to_string())); @@ -28,7 +27,7 @@ pub async fn encode_with_internal_secret(claims: T) -> error::Resu } pub async fn decode_with_internal_secret(token: &str) -> error::Result { - let jwt_secret = JWT_SECRET.read().await; + let jwt_secret = JWT_SECRET.load(); if jwt_secret.is_empty() { return Err(Error::internal_err("JWT secret is not set".to_string())); @@ -65,7 +64,7 @@ pub fn decode_without_verify(token: &str) -> anyhow::Result pub async fn generate_signature(header_and_payload: &str) -> anyhow::Result { let header_and_payload = header_and_payload.trim_start_matches("jwt_ext_"); let header_and_payload = header_and_payload.trim_start_matches("jwt_"); - let secret = JWT_SECRET.read().await; + let secret = JWT_SECRET.load(); // Create HMAC-SHA256 let mut mac = Hmac::::new_from_slice(secret.as_bytes())?; diff --git a/backend/windmill-common/src/lib.rs b/backend/windmill-common/src/lib.rs index 045bc8632c..52e57ea67f 100644 --- a/backend/windmill-common/src/lib.rs +++ b/backend/windmill-common/src/lib.rs @@ -14,7 +14,7 @@ use std::{ net::SocketAddr, str::FromStr, sync::{ - atomic::{AtomicBool, Ordering}, + atomic::{AtomicBool, AtomicI64, Ordering}, Arc, }, }; @@ -211,18 +211,18 @@ lazy_static::lazy_static! { pub static ref CRITICAL_ALERT_MUTE_UI_ENABLED: AtomicBool = AtomicBool::new(false); pub static ref CRITICAL_ALERTS_ON_TOKEN_EXPIRY: AtomicBool = AtomicBool::new(false); - pub static ref BASE_URL: Arc> = Arc::new(RwLock::new("".to_string())); + pub static ref BASE_URL: arc_swap::ArcSwap = arc_swap::ArcSwap::from_pointee("".to_string()); pub static ref IS_READY: std::sync::atomic::AtomicBool = std::sync::atomic::AtomicBool::new(false); - pub static ref HUB_BASE_URL: Arc> = Arc::new(RwLock::new(DEFAULT_HUB_BASE_URL.to_string())); + pub static ref HUB_BASE_URL: arc_swap::ArcSwap = arc_swap::ArcSwap::from_pointee(DEFAULT_HUB_BASE_URL.to_string()); - pub static ref CRITICAL_ERROR_CHANNELS: Arc>> = Arc::new(RwLock::new(vec![])); - pub static ref CRITICAL_ALERTS_ON_DB_OVERSIZE: Arc>> = Arc::new(RwLock::new(None)); + pub static ref CRITICAL_ERROR_CHANNELS: arc_swap::ArcSwap> = arc_swap::ArcSwap::from_pointee(vec![]); + pub static ref CRITICAL_ALERTS_ON_DB_OVERSIZE: arc_swap::ArcSwap> = arc_swap::ArcSwap::from_pointee(None); - pub static ref JOB_RETENTION_SECS: Arc> = Arc::new(RwLock::new(0)); - pub static ref AUDIT_LOG_RETENTION_DAYS: Arc> = Arc::new(RwLock::new(0)); + pub static ref JOB_RETENTION_SECS: AtomicI64 = AtomicI64::new(0); + pub static ref AUDIT_LOG_RETENTION_DAYS: AtomicI64 = AtomicI64::new(0); - pub static ref MONITOR_LOGS_ON_OBJECT_STORE: Arc> = Arc::new(RwLock::new(false)); + pub static ref MONITOR_LOGS_ON_OBJECT_STORE: AtomicBool = AtomicBool::new(false); pub static ref INSTANCE_NAME: String = rd_string(5); @@ -315,7 +315,6 @@ pub async fn shutdown_signal( Ok(()) } -use tokio::sync::RwLock; use utils::rd_string; #[cfg(feature = "prometheus")] diff --git a/backend/windmill-common/src/min_version.rs b/backend/windmill-common/src/min_version.rs index 60ee99dc24..2b1a218f23 100644 --- a/backend/windmill-common/src/min_version.rs +++ b/backend/windmill-common/src/min_version.rs @@ -1,7 +1,5 @@ use crate::error::{self, Error}; use semver::Version; -use std::sync::Arc; -use tokio::sync::RwLock; // ============ Feature Definitions ============ @@ -48,7 +46,7 @@ const _: () = assert!( // ============ Implementation ============ lazy_static::lazy_static! { // Global minimum version across all workers (for feature flags) - pub static ref MIN_VERSION: Arc> = Arc::new(RwLock::new(Version::new(0, 0, 0))); + pub static ref MIN_VERSION: arc_swap::ArcSwap = arc_swap::ArcSwap::from_pointee(Version::new(0, 0, 0)); } /// Creates a VersionConstraint with compile-time assertion that version > MIN_KEEP_ALIVE_VERSION. @@ -80,16 +78,16 @@ impl VersionConstraint { } pub async fn met(&self) -> bool { - let min = MIN_VERSION.read().await; + let min = MIN_VERSION.load(); // If MIN_VERSION is 0.0.0, it hasn't been set yet - assume met - if *min == Version::new(0, 0, 0) { + if **min == Version::new(0, 0, 0) { tracing::warn!( "MIN_VERSION not set yet, assuming feature '{}' is met", self.name ); return true; } - &self.available_since <= &*min + &self.available_since <= &**min } pub async fn assert(&self) -> error::Result<()> { @@ -154,7 +152,7 @@ pub async fn update_min_version( } v.pre = semver::Prerelease::EMPTY; v.build = semver::BuildMetadata::EMPTY; - *MIN_VERSION.write().await = v.clone(); + MIN_VERSION.store(std::sync::Arc::new(v.clone())); } Err(e) => tracing::error!("Failed to fetch min version: {:#?}", e), } diff --git a/backend/windmill-common/src/scripts.rs b/backend/windmill-common/src/scripts.rs index 6d886b6eca..192085d902 100644 --- a/backend/windmill-common/src/scripts.rs +++ b/backend/windmill-common/src/scripts.rs @@ -150,7 +150,7 @@ pub async fn get_hub_script_by_path( .strip_prefix("hub/") .ok_or_else(|| Error::BadRequest("Impossible to remove prefix hex".to_string()))?; - let hub_base_url = HUB_BASE_URL.read().await.clone(); + let hub_base_url = (**HUB_BASE_URL.load()).clone(); // let result = http_get_from_hub( @@ -242,7 +242,7 @@ async fn get_full_hub_script_by_path_inner( http_client: &reqwest::Client, db: Option<&DB>, ) -> crate::error::Result { - let hub_base_url = HUB_BASE_URL.read().await.clone(); + let hub_base_url = (**HUB_BASE_URL.load()).clone(); let response = (|| async { let response = http_get_from_hub( diff --git a/backend/windmill-common/src/utils.rs b/backend/windmill-common/src/utils.rs index e35adc53d2..a4ae5cd56c 100644 --- a/backend/windmill-common/src/utils.rs +++ b/backend/windmill-common/src/utils.rs @@ -28,10 +28,8 @@ use sha2::{Digest, Sha256}; use sqlx::{Pool, Postgres}; use std::borrow::Cow; use std::fmt::Display; -use std::sync::Arc; use std::{fs::DirBuilder as SyncDirBuilder, str::FromStr}; use tokio::fs::DirBuilder as AsyncDirBuilder; -use tokio::sync::RwLock; use url::Url; pub const MAX_PER_PAGE: usize = 10000; @@ -46,13 +44,13 @@ pub const AGENT_WORKER_NAME_PREFIX: &str = "ag"; use crate::CRITICAL_ALERT_MUTE_UI_ENABLED; use std::panic::{self, AssertUnwindSafe, Location}; -use std::sync::atomic::Ordering; +use std::sync::atomic::{AtomicBool, Ordering}; use crate::worker::CLOUD_HOSTED; lazy_static::lazy_static! { pub static ref COOKIE_DOMAIN: Option = std::env::var("COOKIE_DOMAIN").ok(); - pub static ref IS_SECURE: Arc> = Arc::new(RwLock::new(false)); + pub static ref IS_SECURE: AtomicBool = AtomicBool::new(false); pub static ref FORCE_IPV4: bool = std::env::var("FORCE_IPV4") .map(|v| v.to_lowercase() == "true" || v == "1") @@ -164,7 +162,7 @@ lazy_static::lazy_static! { } }; - pub static ref HUB_API_SECRET: Arc>> = Arc::new(RwLock::new(None)); + pub static ref HUB_API_SECRET: arc_swap::ArcSwap> = arc_swap::ArcSwap::from_pointee(None); } #[derive(Clone)] @@ -392,7 +390,7 @@ pub async fn http_get_from_hub( request = request.header("X-uid", uid); } - if let Some(hub_api_secret) = HUB_API_SECRET.read().await.clone() { + if let Some(hub_api_secret) = (**HUB_API_SECRET.load()).clone() { request = request.header("X-api-secret", hub_api_secret); } @@ -427,7 +425,7 @@ pub fn calculate_hash(s: &str) -> String { pub async fn get_license_id_or_uid<'c, E: sqlx::Executor<'c, Database = Postgres>>( db: E, ) -> Result { - let license_id = LICENSE_KEY_ID.read().await.clone(); + let license_id = (**LICENSE_KEY_ID.load()).clone(); if license_id.is_empty() { get_instance_uid(db).await @@ -452,7 +450,7 @@ async fn get_instance_uid<'c, E: sqlx::Executor<'c, Database = Postgres>>(db: E) pub async fn get_telemetry_ids<'c, E: sqlx::Executor<'c, Database = Postgres>>( db: E, ) -> Result<(String, String)> { - let license_id = LICENSE_KEY_ID.read().await.clone(); + let license_id = (**LICENSE_KEY_ID.load()).clone(); let instance_uid = get_instance_uid(db).await?; if license_id.is_empty() { Ok((instance_uid.clone(), instance_uid)) diff --git a/backend/windmill-common/src/variables.rs b/backend/windmill-common/src/variables.rs index 68c1ac36fd..13e4ac189f 100644 --- a/backend/windmill-common/src/variables.rs +++ b/backend/windmill-common/src/variables.rs @@ -322,7 +322,7 @@ pub async fn get_reserved_variables( }, ContextualVariable { name: "WM_BASE_URL".to_string(), - value: BASE_URL.read().await.clone(), + value: (**BASE_URL.load()).clone(), description: "base url of this instance".to_string(), is_custom: false, }, diff --git a/backend/windmill-common/src/webhook.rs b/backend/windmill-common/src/webhook.rs index d695ba1f5a..02d6f7ddc5 100644 --- a/backend/windmill-common/src/webhook.rs +++ b/backend/windmill-common/src/webhook.rs @@ -1,9 +1,7 @@ -use std::sync::Arc; use std::time::Duration; use quick_cache::sync::Cache; use serde::Serialize; -use tokio::sync::RwLock; use tokio::{select, sync::mpsc}; #[cfg(feature = "prometheus")] @@ -26,8 +24,8 @@ lazy_static::lazy_static! { lazy_static::lazy_static! { - pub static ref INSTANCE_EVENTS_WEBHOOK: Arc>> = - Arc::new(RwLock::new(std::env::var("INSTANCE_EVENTS_WEBHOOK").ok())); + pub static ref INSTANCE_EVENTS_WEBHOOK: arc_swap::ArcSwap> = + arc_swap::ArcSwap::from_pointee(std::env::var("INSTANCE_EVENTS_WEBHOOK").ok()); pub static ref WEBHOOK_CACHE: Cache> = Cache::new(100); @@ -211,7 +209,7 @@ impl WebhookShared { } }, Some(WebhookPayload::InstanceEvent(event)) => { - let url = INSTANCE_EVENTS_WEBHOOK.read().await.clone(); + let url = (**INSTANCE_EVENTS_WEBHOOK.load()).clone(); if let Some(url) = url { #[cfg(feature = "prometheus")] let timer = if METRICS_ENABLED.load(std::sync::atomic::Ordering::Relaxed) { Some(WEBHOOK_REQUEST_COUNT.start_timer()) } else { None }; @@ -243,10 +241,7 @@ impl WebhookShared { } pub fn send_instance_event(&self, event: InstanceEvent) { - if INSTANCE_EVENTS_WEBHOOK - .try_read() - .is_ok_and(|v| v.is_none()) - { + if INSTANCE_EVENTS_WEBHOOK.load().is_none() { return; } let _ = self.channel.send(WebhookPayload::InstanceEvent(event)); diff --git a/backend/windmill-common/src/worker.rs b/backend/windmill-common/src/worker.rs index d4479e06d1..b77a3dce60 100644 --- a/backend/windmill-common/src/worker.rs +++ b/backend/windmill-common/src/worker.rs @@ -17,12 +17,12 @@ use std::{ panic::Location, path::{Component, Path, PathBuf}, str::FromStr, - sync::{atomic::AtomicBool, Arc}, + sync::atomic::AtomicBool, time::Duration, }; #[cfg(windows)] use sysinfo::System; -use tokio::{sync::RwLock, time::timeout}; +use tokio::time::timeout; use uuid::Uuid; use windmill_macros::annotations; @@ -207,7 +207,7 @@ lazy_static::lazy_static! { ]; pub static ref DEFAULT_TAGS_PER_WORKSPACE: AtomicBool = AtomicBool::new(false); - pub static ref DEFAULT_TAGS_WORKSPACES: Arc>>> = Arc::new(RwLock::new(None)); + pub static ref DEFAULT_TAGS_WORKSPACES: arc_swap::ArcSwap>> = arc_swap::ArcSwap::from_pointee(None); pub static ref PREVIEW_TAGS_OVERRIDE: AtomicBool = AtomicBool::new(false); pub static ref MAX_TIMEOUT: u64 = std::env::var("TIMEOUT") @@ -220,7 +220,7 @@ lazy_static::lazy_static! { .and_then(|x| x.parse::().ok()) .unwrap_or(*MAX_TIMEOUT); - pub static ref WORKER_CONFIG: Arc> = Arc::new(RwLock::new(WorkerConfig { + pub static ref WORKER_CONFIG: arc_swap::ArcSwap = arc_swap::ArcSwap::from_pointee(WorkerConfig { worker_tags: Default::default(), priority_tags_sorted: Default::default(), dedicated_worker: Default::default(), @@ -233,14 +233,14 @@ lazy_static::lazy_static! { pip_local_dependencies: Default::default(), env_vars: Default::default(), native_mode: false, - })); + }); - pub static ref WORKER_PULL_QUERIES: Arc>> = Arc::new(RwLock::new(vec![])); - pub static ref WORKER_SUSPENDED_PULL_QUERY: Arc> = Arc::new(RwLock::new("".to_string())); + pub static ref WORKER_PULL_QUERIES: arc_swap::ArcSwap> = arc_swap::ArcSwap::from_pointee(vec![]); + pub static ref WORKER_SUSPENDED_PULL_QUERY: arc_swap::ArcSwap = arc_swap::ArcSwap::from_pointee("".to_string()); - pub static ref SMTP_CONFIG: Arc>> = Arc::new(RwLock::new(None)); - pub static ref INDEXER_CONFIG: Arc> = Arc::new(RwLock::new(TantivyIndexerSettings::default())); + pub static ref SMTP_CONFIG: arc_swap::ArcSwap> = arc_swap::ArcSwap::from_pointee(None); + pub static ref INDEXER_CONFIG: arc_swap::ArcSwap = arc_swap::ArcSwap::from_pointee(TantivyIndexerSettings::default()); pub static ref CLOUD_HOSTED: bool = std::env::var("CLOUD_HOSTED").is_ok(); @@ -250,9 +250,9 @@ lazy_static::lazy_static! { .map(|x| x.split(',').map(|x| x.to_string()).collect::>()).unwrap_or_default(); - pub static ref CUSTOM_TAGS_PER_WORKSPACE: Arc> = Arc::new(RwLock::new(CustomTags::default())); + pub static ref CUSTOM_TAGS_PER_WORKSPACE: arc_swap::ArcSwap = arc_swap::ArcSwap::from_pointee(CustomTags::default()); - pub static ref ALL_TAGS: Arc>> = Arc::new(RwLock::new(vec![])); + pub static ref ALL_TAGS: arc_swap::ArcSwap> = arc_swap::ArcSwap::from_pointee(vec![]); @@ -503,8 +503,7 @@ pub async fn store_suspended_pull_query(wc: &WorkerConfig) { return; } let query = make_suspended_pull_query(&wc.worker_tags); - let mut l = WORKER_SUSPENDED_PULL_QUERY.write().await; - *l = query; + WORKER_SUSPENDED_PULL_QUERY.store(std::sync::Arc::new(query)); } pub fn make_pull_query(tags: &[String]) -> String { @@ -530,8 +529,7 @@ pub async fn store_pull_query(wc: &WorkerConfig) { let query = make_pull_query(&tags.tags); queries.push(query); } - let mut l = WORKER_PULL_QUERIES.write().await; - *l = queries; + WORKER_PULL_QUERIES.store(std::sync::Arc::new(queries)); } lazy_static::lazy_static! { @@ -691,13 +689,9 @@ pub async fn reload_custom_tags_setting(db: &DB) -> error::Result<()> { custom_tags.specific, ); - { - let mut l = CUSTOM_TAGS_PER_WORKSPACE.write().await; - *l = custom_tags.clone() - } - { - let mut l = ALL_TAGS.write().await; - *l = [ + CUSTOM_TAGS_PER_WORKSPACE.store(std::sync::Arc::new(custom_tags.clone())); + ALL_TAGS.store(std::sync::Arc::new( + [ custom_tags.global.clone(), custom_tags .specific @@ -705,8 +699,8 @@ pub async fn reload_custom_tags_setting(db: &DB) -> error::Result<()> { .map(|x| x.to_string()) .collect_vec(), ] - .concat(); - } + .concat(), + )); Ok(()) } diff --git a/backend/windmill-common/tests/secret_backend_integration.rs b/backend/windmill-common/tests/secret_backend_integration.rs index 542dba12b4..070c70defa 100644 --- a/backend/windmill-common/tests/secret_backend_integration.rs +++ b/backend/windmill-common/tests/secret_backend_integration.rs @@ -72,8 +72,7 @@ mod tests { async fn setup_base_url() { let base_url = std::env::var("BASE_URL").unwrap_or_else(|_| "http://localhost:8000".to_string()); - let mut url = windmill_common::BASE_URL.write().await; - *url = base_url; + windmill_common::BASE_URL.store(std::sync::Arc::new(base_url)); } macro_rules! skip_if_no_vault { @@ -204,7 +203,7 @@ mod tests { println!("Testing Vault connection with JWT auth..."); println!(" Address: {}", settings.address); println!(" JWT Role: {:?}", settings.jwt_role); - println!(" BASE_URL: {}", windmill_common::BASE_URL.read().await.clone()); + println!(" BASE_URL: {}", (**windmill_common::BASE_URL.load()).clone()); let result = test_vault_connection(&settings, Some(&db)).await; assert!( diff --git a/backend/windmill-mcp/src/client_registration.rs b/backend/windmill-mcp/src/client_registration.rs index ab60c15c78..56f992540a 100644 --- a/backend/windmill-mcp/src/client_registration.rs +++ b/backend/windmill-mcp/src/client_registration.rs @@ -118,7 +118,7 @@ pub async fn get_or_refresh_mcp_client( db: &DB, mcp_server_url: &str, ) -> Result { - let base_url = windmill_common::BASE_URL.read().await.clone(); + let base_url = (**windmill_common::BASE_URL.load()).clone(); let redirect_uri = format!("{}/api/mcp/oauth/callback", base_url); let cached_client: Option = diff --git a/backend/windmill-native-triggers/src/google/external.rs b/backend/windmill-native-triggers/src/google/external.rs index f68706fb2f..b0e8c52f35 100644 --- a/backend/windmill-native-triggers/src/google/external.rs +++ b/backend/windmill-native-triggers/src/google/external.rs @@ -203,7 +203,7 @@ impl Google { data: &NativeTriggerData, db: &DB, ) -> Result { - let base_url = &*BASE_URL.read().await; + let base_url = &**BASE_URL.load(); let webhook_url = generate_webhook_service_url( base_url, w_id, @@ -335,7 +335,7 @@ impl Google { ))); } }; - let base_url = &*BASE_URL.read().await; + let base_url = &**BASE_URL.load(); // Reuse the same channel ID so external_id stays permanent let channel_id = trigger.external_id.clone(); let webhook_url = generate_webhook_service_url( diff --git a/backend/windmill-native-triggers/src/nextcloud/external.rs b/backend/windmill-native-triggers/src/nextcloud/external.rs index b6e67afd59..b296ddb15b 100644 --- a/backend/windmill-native-triggers/src/nextcloud/external.rs +++ b/backend/windmill-native-triggers/src/nextcloud/external.rs @@ -40,7 +40,7 @@ impl FullNextcloudPayload { webhook_token: &str, data: &NativeTriggerData, ) -> FullNextcloudPayload { - let base_url = &*BASE_URL.read().await; + let base_url = &**BASE_URL.load(); let uri = generate_webhook_service_url( base_url, w_id, diff --git a/backend/windmill-oauth/Cargo.toml b/backend/windmill-oauth/Cargo.toml index 434b3fca97..85499442b9 100644 --- a/backend/windmill-oauth/Cargo.toml +++ b/backend/windmill-oauth/Cargo.toml @@ -21,6 +21,7 @@ tower-cookies.workspace = true reqwest = { version = "0.12", features = ["json"] } sqlx.workspace = true tokio.workspace = true +arc-swap.workspace = true serde.workspace = true serde_json.workspace = true diff --git a/backend/windmill-oauth/src/lib.rs b/backend/windmill-oauth/src/lib.rs index eff53be481..644cab724b 100644 --- a/backend/windmill-oauth/src/lib.rs +++ b/backend/windmill-oauth/src/lib.rs @@ -16,7 +16,6 @@ use std::collections::HashMap; use std::fmt::Debug; -use std::sync::Arc; use anyhow::anyhow; use base64::Engine; @@ -24,7 +23,6 @@ use hmac::Mac; use itertools::Itertools; use serde::{de::DeserializeOwned, Deserialize, Serialize}; use sqlx::{Postgres, Transaction}; -use tokio::sync::RwLock; use tower_cookies::{Cookie, Cookies}; use windmill_common::error::{self, to_anyhow, Error}; use windmill_common::more_serde::maybe_number_opt; @@ -54,11 +52,11 @@ lazy_static::lazy_static! { .build() .expect("Failed to create OAuth HTTP client"); - pub static ref OAUTH_CLIENTS: Arc> = Arc::new(RwLock::new(AllClients { + pub static ref OAUTH_CLIENTS: arc_swap::ArcSwap = arc_swap::ArcSwap::from_pointee(AllClients { logins: HashMap::new(), connects: HashMap::new(), slack: None - })); + }); } /// OAuth client with associated scopes and configuration @@ -407,7 +405,7 @@ pub async fn build_slack_client( let token_url = Url::parse("https://slack.com/api/oauth.access") .map_err(|e| anyhow!("Invalid Slack token URL: {e}"))?; - let base_url = BASE_URL.read().await.clone(); + let base_url = (**BASE_URL.load()).clone(); let redirect_url = format!("{}/oauth/callback_slack", base_url); let mut client = OClient::new(client_id.to_string(), auth_url, token_url); @@ -488,7 +486,7 @@ pub async fn build_client_credentials_oauth_client( tenant: oauth_client_config.tenant.clone(), }; - let base_url = BASE_URL.read().await.clone(); + let base_url = (**BASE_URL.load()).clone(); let (_, client) = build_basic_client( client_name.to_string(), connect_config, diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs index a7cb9e66af..f832413be7 100644 --- a/backend/windmill-queue/src/jobs.rs +++ b/backend/windmill-queue/src/jobs.rs @@ -1188,7 +1188,7 @@ async fn commit_completed_job( { if !success { tracing::error!("Could not apply schedule error handler: {}", err); - let base_url = windmill_common::BASE_URL.read().await; + let base_url = windmill_common::BASE_URL.load(); let w_id: &String = &completed_job.workspace_id; if !matches!(err, Error::QuotaExceeded(_)) { report_error_to_workspace_handler_or_critical_side_channel( @@ -3480,7 +3480,7 @@ async fn pull_single_job_and_mark_as_running_no_concurrency_limit<'c>( * suspend_until is non-null * and suspend = 0 when the resume messages are received * or suspend_until <= now() if it has timed out */ - let query = WORKER_SUSPENDED_PULL_QUERY.read().await; + let query = WORKER_SUSPENDED_PULL_QUERY.load(); if query.is_empty() { tracing::warn!("No suspended pull queries available"); @@ -3502,7 +3502,7 @@ async fn pull_single_job_and_mark_as_running_no_concurrency_limit<'c>( // let instant = Instant::now(); let mut highest_priority_job: Option = None; - let queries = WORKER_PULL_QUERIES.read().await; + let queries = WORKER_PULL_QUERIES.load(); if queries.is_empty() { tracing::warn!("No pull queries available"); diff --git a/backend/windmill-queue/src/schedule.rs b/backend/windmill-queue/src/schedule.rs index 49a108d6d0..6b0a9904b5 100644 --- a/backend/windmill-queue/src/schedule.rs +++ b/backend/windmill-queue/src/schedule.rs @@ -130,7 +130,7 @@ pub async fn push_scheduled_job<'c>( authed: Option<&Authed>, now_cutoff: Option>, ) -> Result> { - if !*LICENSE_KEY_VALID.read().await { + if !LICENSE_KEY_VALID.load(std::sync::atomic::Ordering::Relaxed) { return Err(error::Error::BadRequest( "License key is not valid. Go to your superadmin settings to update your license key." .to_string(), diff --git a/backend/windmill-queue/src/tags.rs b/backend/windmill-queue/src/tags.rs index 6cf6620039..469ad4eefd 100644 --- a/backend/windmill-queue/src/tags.rs +++ b/backend/windmill-queue/src/tags.rs @@ -1,10 +1,10 @@ use windmill_common::worker::{DEFAULT_TAGS_PER_WORKSPACE, DEFAULT_TAGS_WORKSPACES}; pub async fn per_workspace_tag(workspace_id: &str) -> bool { - let per_workspace_workspaces = DEFAULT_TAGS_WORKSPACES.read().await; + let per_workspace_workspaces = DEFAULT_TAGS_WORKSPACES.load(); DEFAULT_TAGS_PER_WORKSPACE.load(std::sync::atomic::Ordering::Relaxed) && (per_workspace_workspaces.is_none() - || per_workspace_workspaces + || (**per_workspace_workspaces) .as_ref() .unwrap() .contains(&workspace_id.to_string())) diff --git a/backend/windmill-store/src/oauth_refresh_oss.rs b/backend/windmill-store/src/oauth_refresh_oss.rs index 7402e66e49..e8edb19ccd 100644 --- a/backend/windmill-store/src/oauth_refresh_oss.rs +++ b/backend/windmill-store/src/oauth_refresh_oss.rs @@ -30,7 +30,7 @@ pub async fn _refresh_token<'c>( w_id, id, db, - &*windmill_oauth::OAUTH_CLIENTS.read().await, + &*windmill_oauth::OAUTH_CLIENTS.load(), &windmill_oauth::OAUTH_HTTP_CLIENT, include_str!("../../oauth_connect.json"), ) diff --git a/backend/windmill-test-utils/src/lib.rs b/backend/windmill-test-utils/src/lib.rs index b2e3be2cef..a24f08bfda 100644 --- a/backend/windmill-test-utils/src/lib.rs +++ b/backend/windmill-test-utils/src/lib.rs @@ -417,7 +417,7 @@ pub fn spawn_test_worker( let future = async move { let base_internal_url = format!("http://localhost:{}", port); { - let mut wc = WORKER_CONFIG.write().await; + let mut wc = (**WORKER_CONFIG.load()).clone(); wc.worker_tags = windmill_common::worker::DEFAULT_TAGS.clone(); wc.priority_tags_sorted = vec![windmill_common::worker::PriorityTags { priority: 0, @@ -425,6 +425,7 @@ pub fn spawn_test_worker( }]; windmill_common::worker::store_suspended_pull_query(&wc).await; windmill_common::worker::store_pull_query(&wc).await; + WORKER_CONFIG.store(std::sync::Arc::new(wc)); } windmill_worker::run_worker( &conn, @@ -505,10 +506,9 @@ pub fn spawn_test_worker_dedicated( let config = windmill_common::worker::load_worker_config(db_ref, killpill_tx2) .await .expect("load worker config"); - let mut wc = WORKER_CONFIG.write().await; windmill_common::worker::store_suspended_pull_query(&config).await; windmill_common::worker::store_pull_query(&config).await; - *wc = config; + WORKER_CONFIG.store(std::sync::Arc::new(config)); } windmill_worker::run_worker( &conn, @@ -537,9 +537,10 @@ pub async fn in_test_worker_dedicated( set_jwt_secret().await; // Reset WORKER_CONFIG to avoid stale state from previous tests' monitor reloads. { - let mut wc = WORKER_CONFIG.write().await; + let mut wc = (**WORKER_CONFIG.load()).clone(); wc.dedicated_worker = None; wc.dedicated_workers = None; + WORKER_CONFIG.store(std::sync::Arc::new(wc)); } let (quit, worker) = spawn_test_worker_dedicated(&conn.into(), port, dedicated_workers); let worker = tokio::time::timeout(std::time::Duration::from_secs(90), worker); @@ -646,8 +647,7 @@ fn find_module_in_vec(modules: Vec, id: &str) -> Option>( test: impl Fn() -> F, ) { for constraint in constraints { - *windmill_common::min_version::MIN_VERSION.write().await = constraint.version().clone(); + windmill_common::min_version::MIN_VERSION + .store(std::sync::Arc::new(constraint.version().clone())); test().await; } } diff --git a/backend/windmill-trigger-http/src/handler.rs b/backend/windmill-trigger-http/src/handler.rs index 9da318f3e4..ad50e9811f 100644 --- a/backend/windmill-trigger-http/src/handler.rs +++ b/backend/windmill-trigger-http/src/handler.rs @@ -62,7 +62,7 @@ pub async fn route_path_key_exists( .await? .unwrap_or(false) } else { - let http_route_workspaced = *HTTP_ROUTE_WORKSPACED_ROUTE.read().await; + let http_route_workspaced = HTTP_ROUTE_WORKSPACED_ROUTE.load(std::sync::atomic::Ordering::Relaxed); let effective_workspaced = workspaced_route.unwrap_or(false) || http_route_workspaced; let route_path_key = if effective_workspaced { std::borrow::Cow::Owned(format!("{}/{}", w_id, route_path_key.trim_matches('/'))) @@ -145,7 +145,7 @@ async fn require_admin_for_instance_wide_route( is_admin: bool, workspaced_route: Option, ) -> Result { - let http_route_workspaced = *HTTP_ROUTE_WORKSPACED_ROUTE.read().await; + let http_route_workspaced = HTTP_ROUTE_WORKSPACED_ROUTE.load(std::sync::atomic::Ordering::Relaxed); let effective_workspaced = workspaced_route.unwrap_or(false) || http_route_workspaced; if !is_admin && !effective_workspaced { return Err(Error::NotAuthorized( @@ -465,7 +465,7 @@ impl TriggerCrud for HttpTrigger { let resolved_edited_by = trigger.base.resolve_edited_by(authed); let resolved_permissioned_as = trigger.base.resolve_permissioned_as(authed); - let http_route_workspaced = *HTTP_ROUTE_WORKSPACED_ROUTE.read().await; + let http_route_workspaced = HTTP_ROUTE_WORKSPACED_ROUTE.load(std::sync::atomic::Ordering::Relaxed); let effective_workspaced = trigger.config.workspaced_route.unwrap_or(false) || http_route_workspaced; diff --git a/backend/windmill-trigger-http/src/lib.rs b/backend/windmill-trigger-http/src/lib.rs index f851dfc12e..f89247dd16 100644 --- a/backend/windmill-trigger-http/src/lib.rs +++ b/backend/windmill-trigger-http/src/lib.rs @@ -274,7 +274,7 @@ pub async fn refresh_routers(db: &DB) -> Result<(bool, RwLockReadGuard<'_, Route .await?; let mut router = matchit::Router::new(); - let http_route_workspaced = *HTTP_ROUTE_WORKSPACED_ROUTE.read().await; + let http_route_workspaced = HTTP_ROUTE_WORKSPACED_ROUTE.load(std::sync::atomic::Ordering::Relaxed); for trigger in triggers { let full_path = diff --git a/backend/windmill-worker/src/ai/utils.rs b/backend/windmill-worker/src/ai/utils.rs index d095602b40..e5e7995760 100644 --- a/backend/windmill-worker/src/ai/utils.rs +++ b/backend/windmill-worker/src/ai/utils.rs @@ -515,7 +515,7 @@ async fn refresh_token_if_expired( ); // Call the API refresh endpoint - let base_url = windmill_common::BASE_URL.read().await.clone(); + let base_url = (**windmill_common::BASE_URL.load()).clone(); let refresh_url = format!( "{}/api/w/{}/oauth/refresh_token/{}", base_url, workspace_id, account_id diff --git a/backend/windmill-worker/src/ansible_executor.rs b/backend/windmill-worker/src/ansible_executor.rs index 5369a40ee9..bc16f82813 100644 --- a/backend/windmill-worker/src/ansible_executor.rs +++ b/backend/windmill-worker/src/ansible_executor.rs @@ -361,12 +361,10 @@ async fn handle_ansible_python_deps( create_dependencies_dir(job_dir).await; let mut additional_python_paths: Vec = WORKER_CONFIG - .read() - .await + .load() .additional_python_paths .clone() - .unwrap_or_else(|| vec![]) - .clone(); + .unwrap_or_else(|| vec![]); let mut requirements; let requirements = match requirements_o { diff --git a/backend/windmill-worker/src/bun_executor.rs b/backend/windmill-worker/src/bun_executor.rs index c7893a63e0..8f787168ac 100644 --- a/backend/windmill-worker/src/bun_executor.rs +++ b/backend/windmill-worker/src/bun_executor.rs @@ -2941,7 +2941,7 @@ pub async fn handle_wac_v2_output( mac.update(resume_id.to_be_bytes().as_ref()); let signature = hex::encode(mac.finalize().into_bytes()); - let base_url = windmill_common::BASE_URL.read().await.clone(); + let base_url = (**windmill_common::BASE_URL.load()).clone(); let w_id = &job.workspace_id; let job_id = &job.id; diff --git a/backend/windmill-worker/src/common.rs b/backend/windmill-worker/src/common.rs index 30616228aa..964e136f81 100644 --- a/backend/windmill-worker/src/common.rs +++ b/backend/windmill-worker/src/common.rs @@ -521,7 +521,7 @@ pub async fn build_envs_map(context: Vec) -> HashMap = context.into_iter().map(|rv| (rv.name, rv.value)).collect(); - let envs = WORKER_CONFIG.read().await.clone().env_vars; + let envs = (**WORKER_CONFIG.load()).clone().env_vars; for env in envs { r.insert(env.0.clone(), env.1.clone()); } diff --git a/backend/windmill-worker/src/deno_executor.rs b/backend/windmill-worker/src/deno_executor.rs index 207cf5aba4..24b85b0c13 100644 --- a/backend/windmill-worker/src/deno_executor.rs +++ b/backend/windmill-worker/src/deno_executor.rs @@ -73,7 +73,7 @@ async fn get_common_deno_proc_envs( w_id: &str, conn: Option<&Connection>, ) -> HashMap { - let hostname = BASE_URL.read().await.clone(); + let hostname = (**BASE_URL.load()).clone(); let hostname_base = hostname.split("://").last().unwrap_or("localhost"); let hostname_internal = base_internal_url.split("://").last().unwrap_or("localhost"); let deno_auth_tokens_base = DENO_AUTH_TOKENS.as_str(); diff --git a/backend/windmill-worker/src/python_executor.rs b/backend/windmill-worker/src/python_executor.rs index e00ae24905..995374e42a 100644 --- a/backend/windmill-worker/src/python_executor.rs +++ b/backend/windmill-worker/src/python_executor.rs @@ -201,7 +201,7 @@ pub async fn uv_pip_compile( logs.push_str(&format!("\ncontent of requirements:\n{}\n", requirements)); let requirements = if let Some(pip_local_dependencies) = - WORKER_CONFIG.read().await.pip_local_dependencies.as_ref() + WORKER_CONFIG.load().pip_local_dependencies.as_ref() { let deps = pip_local_dependencies.clone(); let compiled_deps = deps.iter().map(|dep| { @@ -1699,12 +1699,10 @@ pub(crate) async fn handle_python_deps( create_dependencies_dir(job_dir).await; let mut additional_python_paths: Vec = WORKER_CONFIG - .read() - .await + .load() .additional_python_paths .clone() - .unwrap_or_else(|| vec![]) - .clone(); + .unwrap_or_else(|| vec![]); let (pyv, resolved_lines) = match requirements_o { // Deployed diff --git a/backend/windmill-worker/src/worker.rs b/backend/windmill-worker/src/worker.rs index de50abce78..264dd31eac 100644 --- a/backend/windmill-worker/src/worker.rs +++ b/backend/windmill-worker/src/worker.rs @@ -1937,7 +1937,7 @@ pub async fn run_worker( let mut jobs_executed = 0; let is_dedicated_worker: bool = { - let config = WORKER_CONFIG.read().await; + let config = WORKER_CONFIG.load(); config.dedicated_worker.is_some() || config .dedicated_workers @@ -2048,7 +2048,7 @@ pub async fn run_worker( worker = %worker_name, hostname = %hostname, "listening for jobs, WORKER_GROUP: {}, config: {:?}", *WORKER_GROUP, - WORKER_CONFIG.read().await + WORKER_CONFIG.load() ); } @@ -2141,7 +2141,7 @@ pub async fn run_worker( } #[cfg(feature = "enterprise")] { - let valid_key = *LICENSE_KEY_VALID.read().await; + let valid_key = LICENSE_KEY_VALID.load(std::sync::atomic::Ordering::Relaxed); if !valid_key { tracing::error!( @@ -3014,7 +3014,7 @@ pub async fn run_worker( #[cfg(feature = "enterprise")] { - let valid_key = *LICENSE_KEY_VALID.read().await; + let valid_key = LICENSE_KEY_VALID.load(std::sync::atomic::Ordering::Relaxed); if !valid_key { tracing::info!(worker = %worker_name, hostname = %hostname, "Invalid license key, exiting immediately"); @@ -3074,7 +3074,7 @@ async fn queue_init_bash_maybe<'c>( same_worker_tx: SameWorkerSender, worker_name: &str, ) -> anyhow::Result { - let uuid_content = if let Some(content) = WORKER_CONFIG.read().await.init_bash.clone() { + let uuid_content = if let Some(content) = WORKER_CONFIG.load().init_bash.clone() { let uuid = match conn { Connection::Sql(db) => push_init_job(db, content.clone(), worker_name).await?, Connection::Http(client) => queue_init_job(client, &content).await?, @@ -3102,7 +3102,7 @@ fn spawn_periodic_script_task( mut killpill_rx: tokio::sync::broadcast::Receiver<()>, ) { tokio::spawn(async move { - let config = WORKER_CONFIG.read().await; + let config = WORKER_CONFIG.load(); match ( &config.periodic_script_bash, diff --git a/backend/windmill-worker/src/worker_utils.rs b/backend/windmill-worker/src/worker_utils.rs index 4845bc0adc..b325f3cf08 100644 --- a/backend/windmill-worker/src/worker_utils.rs +++ b/backend/windmill-worker/src/worker_utils.rs @@ -27,7 +27,7 @@ pub(crate) async fn update_worker_ping_full( occupancy_metrics: &mut OccupancyMetrics, killpill_tx: &KillpillSender, ) { - let wc = WORKER_CONFIG.read().await; + let wc = WORKER_CONFIG.load(); let tags = wc.worker_tags.clone(); let native_mode = wc.native_mode; drop(wc); @@ -171,7 +171,7 @@ pub async fn insert_ping( db: &Connection, ) -> anyhow::Result<()> { let (tags, dw, dws, native_mode) = { - let wc = WORKER_CONFIG.read().await.clone(); + let wc = (**WORKER_CONFIG.load()).clone(); ( wc.worker_tags, wc.dedicated_worker