mirror of
https://github.com/windmill-labs/windmill.git
synced 2026-09-18 08:02:29 +00:00
Compare commits
8
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
c631091daf | ||
|
|
119d2836ac | ||
|
|
5e2bd56f75 | ||
|
|
49bb1863c9 | ||
|
|
79475e70d2 | ||
|
|
e931be5192 | ||
|
|
d498f53dab | ||
|
|
7f619c2ee1 |
Generated
+1
@@ -16994,6 +16994,7 @@ name = "windmill-oauth"
|
||||
version = "1.682.0"
|
||||
dependencies = [
|
||||
"anyhow",
|
||||
"arc-swap",
|
||||
"async-oauth2",
|
||||
"axum 0.8.4",
|
||||
"base64 0.22.1",
|
||||
|
||||
@@ -1 +1 @@
|
||||
be2f3d4d11bb7110200524d7157caab3aac53996
|
||||
4b377e3eaef39897fdb9fb753a782f2eb757c1da
|
||||
+2
-2
@@ -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 {
|
||||
|
||||
+119
-99
@@ -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::<Vec<String>>();
|
||||
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::<String>(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::<i64>(
|
||||
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::<i64>(
|
||||
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::<bool>(
|
||||
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<T: FromStr + DeserializeOwned>(
|
||||
/// 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<RwLock<Option<T>>>` (e.g. `ArcSwap<Option<T>>`,
|
||||
/// an `AtomicBool`, etc.) can reuse the load pipeline.
|
||||
pub async fn load_option_setting_value<T: FromStr + DeserializeOwned>(
|
||||
conn: &Connection,
|
||||
setting_name: &str,
|
||||
std_env_var: &str,
|
||||
lock: Arc<RwLock<Option<T>>>,
|
||||
) -> error::Result<()> {
|
||||
) -> error::Result<Option<T>> {
|
||||
let force_value = std::env::var(format!("FORCE_{}", std_env_var))
|
||||
.ok()
|
||||
.and_then(|x| x.parse::<T>().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<T: FromStr + DeserializeOwned>(
|
||||
}
|
||||
};
|
||||
|
||||
if value.is_none() {
|
||||
tracing::info!("Loaded {setting_name} setting to None");
|
||||
}
|
||||
|
||||
Ok(value)
|
||||
}
|
||||
|
||||
pub async fn reload_option_setting<T: FromStr + DeserializeOwned>(
|
||||
conn: &Connection,
|
||||
setting_name: &str,
|
||||
std_env_var: &str,
|
||||
lock: Arc<RwLock<Option<T>>>,
|
||||
) -> error::Result<()> {
|
||||
let value = load_option_setting_value::<T>(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<RwLock<Option<Vec<Url>>>>` (e.g.
|
||||
/// `ArcSwap<Option<Vec<Url>>>`) can reuse the parsing pipeline.
|
||||
pub async fn load_url_list_setting_value(
|
||||
conn: &Connection,
|
||||
setting_name: &str,
|
||||
std_env_var: &str,
|
||||
lock: Arc<RwLock<Option<Vec<url::Url>>>>,
|
||||
) -> error::Result<()> {
|
||||
) -> error::Result<Option<Vec<url::Url>>> {
|
||||
// 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<RwLock<Option<Vec<url::Url>>>>,
|
||||
) -> 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<T: FromStr + DeserializeOwned + Display>(
|
||||
/// Load a required setting value without writing it anywhere.
|
||||
///
|
||||
/// Extracted from [`reload_setting`] so callers that store the value in
|
||||
/// something other than `Arc<RwLock<T>>` (e.g. `AtomicI64`, `AtomicBool`,
|
||||
/// `ArcSwap<T>`) can reuse the load pipeline.
|
||||
pub async fn load_setting_value<T: FromStr + DeserializeOwned + Display>(
|
||||
conn: &Connection,
|
||||
setting_name: &str,
|
||||
std_env_var: &str,
|
||||
default: T,
|
||||
lock: Arc<RwLock<T>>,
|
||||
transformer: fn(T) -> T,
|
||||
) -> error::Result<()> {
|
||||
) -> error::Result<T> {
|
||||
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<T: FromStr + DeserializeOwned + Display>(
|
||||
}
|
||||
};
|
||||
|
||||
Ok(value)
|
||||
}
|
||||
|
||||
pub async fn reload_setting<T: FromStr + DeserializeOwned + Display>(
|
||||
conn: &Connection,
|
||||
setting_name: &str,
|
||||
std_env_var: &str,
|
||||
default: T,
|
||||
lock: Arc<RwLock<T>>,
|
||||
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<Postgres>) {
|
||||
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<Postgres>) {
|
||||
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<String> {
|
||||
} 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<Postgres>, 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::<bool>().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(())
|
||||
}
|
||||
@@ -3701,18 +3728,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 +3766,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 +3797,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 +3919,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") {
|
||||
|
||||
@@ -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();
|
||||
|
||||
@@ -276,19 +276,19 @@ mod job_payload {
|
||||
|
||||
#[sqlx::test(fixtures("base", "hello"))]
|
||||
async fn test_dependencies_payload_min_1_427(db: Pool<Postgres>) -> 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<Postgres>) -> 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<Postgres>) -> 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(())
|
||||
}
|
||||
|
||||
@@ -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,
|
||||
|
||||
+11
-17
@@ -4034,19 +4034,16 @@ async fn test_flow_substep_tag_availability_check(db: Pool<Postgres>) -> 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<Postgres>) -> 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(())
|
||||
}
|
||||
|
||||
@@ -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 => {
|
||||
|
||||
@@ -224,7 +224,7 @@ async fn list_hub_flows(Extension(db): Extension<DB>) -> 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<Box<serde_json::value::RawValue>> {
|
||||
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),
|
||||
|
||||
@@ -222,7 +222,7 @@ async fn test_plaintext_backward_compat(db: Pool<Postgres>) -> 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<Postgres>) -> 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"}))
|
||||
|
||||
@@ -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."
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -293,7 +293,7 @@ async fn cleanup_job_logs(
|
||||
db: &DB,
|
||||
store: &Arc<dyn ObjectStore>,
|
||||
) -> 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<dyn ObjectStore>,
|
||||
) -> 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
|
||||
|
||||
@@ -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<ExitImpersonationRequest>,
|
||||
) -> Result<String> {
|
||||
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 {
|
||||
|
||||
@@ -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<CustomTagQuery>,
|
||||
) -> JsonResult<Vec<String>> {
|
||||
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<String>,
|
||||
) -> JsonResult<Vec<String>> {
|
||||
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))
|
||||
}
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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,
|
||||
);
|
||||
|
||||
@@ -1410,7 +1410,7 @@ async fn create_app_internal<'a>(
|
||||
async fn list_hub_apps(Extension(db): Extension<DB>) -> 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<Box<serde_json::value::RawValue>> {
|
||||
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<Box<serde_json::value::RawValue>> {
|
||||
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),
|
||||
|
||||
@@ -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,
|
||||
)
|
||||
|
||||
@@ -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!(
|
||||
|
||||
@@ -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()
|
||||
}
|
||||
|
||||
@@ -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);
|
||||
|
||||
|
||||
@@ -266,7 +266,7 @@ fn build_protected_resource_metadata(
|
||||
pub async fn workspaced_oauth_metadata(
|
||||
Path(workspace_id): Path<String>,
|
||||
) -> Json<AuthorizationMetadata> {
|
||||
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<String>,
|
||||
) -> Json<ProtectedResourceMetadata> {
|
||||
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<AuthorizationMetadata> {
|
||||
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<AuthorizationMetadata> {
|
||||
}
|
||||
|
||||
pub async fn gateway_protected_resource_metadata() -> Json<ProtectedResourceMetadata> {
|
||||
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",
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -103,7 +103,7 @@ async fn list_logins() -> error::JsonResult<Logins> {
|
||||
#[cfg(all(feature = "oauth2", not(feature = "private")))]
|
||||
async fn list_connects() -> error::JsonResult<Vec<String>> {
|
||||
Ok(Json(
|
||||
(&OAUTH_CLIENTS.read().await.connects)
|
||||
(&OAUTH_CLIENTS.load().connects)
|
||||
.keys()
|
||||
.map(|x| x.to_owned())
|
||||
.collect_vec(),
|
||||
|
||||
@@ -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<RwLock<bool>> = Arc::new(RwLock::new(true));
|
||||
pub static ref LICENSE_KEY_ID: Arc<RwLock<String>> = Arc::new(RwLock::new("".to_string()));
|
||||
pub static ref LICENSE_KEY: Arc<RwLock<String>> = 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<String> = arc_swap::ArcSwap::from_pointee("".to_string());
|
||||
pub static ref LICENSE_KEY: arc_swap::ArcSwap<String> = arc_swap::ArcSwap::from_pointee("".to_string());
|
||||
}
|
||||
|
||||
#[cfg(not(feature = "private"))]
|
||||
|
||||
@@ -72,7 +72,7 @@ pub async fn get_full_hub_flow_by_path(
|
||||
) -> crate::error::Result<HubFlow> {
|
||||
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)
|
||||
|
||||
@@ -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<RwLock<bool>> = Arc::new(RwLock::new(false));
|
||||
pub static ref HTTP_ROUTE_WORKSPACED_ROUTE: AtomicBool = AtomicBool::new(false);
|
||||
}
|
||||
|
||||
pub const ENV_SETTINGS: &[&str] = &[
|
||||
|
||||
@@ -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) {
|
||||
|
||||
@@ -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<RwLock<String>> = Arc::new(RwLock::new("".to_string()));
|
||||
pub static ref JWT_SECRET: arc_swap::ArcSwap<String> = arc_swap::ArcSwap::from_pointee("".to_string());
|
||||
}
|
||||
|
||||
pub async fn encode_with_internal_secret<T: Serialize>(claims: T) -> error::Result<String> {
|
||||
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<T: Serialize>(claims: T) -> error::Resu
|
||||
}
|
||||
|
||||
pub async fn decode_with_internal_secret<T: DeserializeOwned>(token: &str) -> error::Result<T> {
|
||||
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<T: DeserializeOwned>(token: &str) -> anyhow::Result
|
||||
pub async fn generate_signature(header_and_payload: &str) -> anyhow::Result<String> {
|
||||
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::<Sha256>::new_from_slice(secret.as_bytes())?;
|
||||
|
||||
@@ -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<RwLock<String>> = Arc::new(RwLock::new("".to_string()));
|
||||
pub static ref BASE_URL: arc_swap::ArcSwap<String> = 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<RwLock<String>> = Arc::new(RwLock::new(DEFAULT_HUB_BASE_URL.to_string()));
|
||||
pub static ref HUB_BASE_URL: arc_swap::ArcSwap<String> = arc_swap::ArcSwap::from_pointee(DEFAULT_HUB_BASE_URL.to_string());
|
||||
|
||||
|
||||
pub static ref CRITICAL_ERROR_CHANNELS: Arc<RwLock<Vec<CriticalErrorChannel>>> = Arc::new(RwLock::new(vec![]));
|
||||
pub static ref CRITICAL_ALERTS_ON_DB_OVERSIZE: Arc<RwLock<Option<f32>>> = Arc::new(RwLock::new(None));
|
||||
pub static ref CRITICAL_ERROR_CHANNELS: arc_swap::ArcSwap<Vec<CriticalErrorChannel>> = arc_swap::ArcSwap::from_pointee(vec![]);
|
||||
pub static ref CRITICAL_ALERTS_ON_DB_OVERSIZE: arc_swap::ArcSwap<Option<f32>> = arc_swap::ArcSwap::from_pointee(None);
|
||||
|
||||
pub static ref JOB_RETENTION_SECS: Arc<RwLock<i64>> = Arc::new(RwLock::new(0));
|
||||
pub static ref AUDIT_LOG_RETENTION_DAYS: Arc<RwLock<i64>> = 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<RwLock<bool>> = 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);
|
||||
|
||||
|
||||
@@ -48,7 +48,7 @@ const _: () = assert!(
|
||||
// ============ Implementation ============
|
||||
lazy_static::lazy_static! {
|
||||
// Global minimum version across all workers (for feature flags)
|
||||
pub static ref MIN_VERSION: Arc<RwLock<Version>> = Arc::new(RwLock::new(Version::new(0, 0, 0)));
|
||||
pub static ref MIN_VERSION: arc_swap::ArcSwap<Version> = 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 +80,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 +154,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),
|
||||
}
|
||||
|
||||
@@ -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<HubScript> {
|
||||
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(
|
||||
|
||||
@@ -46,13 +46,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<String> = std::env::var("COOKIE_DOMAIN").ok();
|
||||
pub static ref IS_SECURE: Arc<RwLock<bool>> = 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 +164,7 @@ lazy_static::lazy_static! {
|
||||
}
|
||||
};
|
||||
|
||||
pub static ref HUB_API_SECRET: Arc<RwLock<Option<String>>> = Arc::new(RwLock::new(None));
|
||||
pub static ref HUB_API_SECRET: arc_swap::ArcSwap<Option<String>> = arc_swap::ArcSwap::from_pointee(None);
|
||||
}
|
||||
|
||||
#[derive(Clone)]
|
||||
@@ -392,7 +392,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 +427,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<String> {
|
||||
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 +452,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))
|
||||
|
||||
@@ -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,
|
||||
},
|
||||
|
||||
@@ -26,8 +26,8 @@ lazy_static::lazy_static! {
|
||||
|
||||
lazy_static::lazy_static! {
|
||||
|
||||
pub static ref INSTANCE_EVENTS_WEBHOOK: Arc<RwLock<Option<String>>> =
|
||||
Arc::new(RwLock::new(std::env::var("INSTANCE_EVENTS_WEBHOOK").ok()));
|
||||
pub static ref INSTANCE_EVENTS_WEBHOOK: arc_swap::ArcSwap<Option<String>> =
|
||||
arc_swap::ArcSwap::from_pointee(std::env::var("INSTANCE_EVENTS_WEBHOOK").ok());
|
||||
|
||||
pub static ref WEBHOOK_CACHE: Cache<String, Option<String>> = Cache::new(100);
|
||||
|
||||
@@ -211,7 +211,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 +243,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));
|
||||
|
||||
@@ -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<RwLock<Option<Vec<String>>>> = Arc::new(RwLock::new(None));
|
||||
pub static ref DEFAULT_TAGS_WORKSPACES: arc_swap::ArcSwap<Option<Vec<String>>> = 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::<u64>().ok())
|
||||
.unwrap_or(*MAX_TIMEOUT);
|
||||
|
||||
pub static ref WORKER_CONFIG: Arc<RwLock<WorkerConfig>> = Arc::new(RwLock::new(WorkerConfig {
|
||||
pub static ref WORKER_CONFIG: arc_swap::ArcSwap<WorkerConfig> = 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<RwLock<Vec<String>>> = Arc::new(RwLock::new(vec![]));
|
||||
pub static ref WORKER_SUSPENDED_PULL_QUERY: Arc<RwLock<String>> = Arc::new(RwLock::new("".to_string()));
|
||||
pub static ref WORKER_PULL_QUERIES: arc_swap::ArcSwap<Vec<String>> = arc_swap::ArcSwap::from_pointee(vec![]);
|
||||
pub static ref WORKER_SUSPENDED_PULL_QUERY: arc_swap::ArcSwap<String> = arc_swap::ArcSwap::from_pointee("".to_string());
|
||||
|
||||
|
||||
pub static ref SMTP_CONFIG: Arc<RwLock<Option<Smtp>>> = Arc::new(RwLock::new(None));
|
||||
pub static ref INDEXER_CONFIG: Arc<RwLock<TantivyIndexerSettings>> = Arc::new(RwLock::new(TantivyIndexerSettings::default()));
|
||||
pub static ref SMTP_CONFIG: arc_swap::ArcSwap<Option<Smtp>> = arc_swap::ArcSwap::from_pointee(None);
|
||||
pub static ref INDEXER_CONFIG: arc_swap::ArcSwap<TantivyIndexerSettings> = 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::<Vec<_>>()).unwrap_or_default();
|
||||
|
||||
|
||||
pub static ref CUSTOM_TAGS_PER_WORKSPACE: Arc<RwLock<CustomTags>> = Arc::new(RwLock::new(CustomTags::default()));
|
||||
pub static ref CUSTOM_TAGS_PER_WORKSPACE: arc_swap::ArcSwap<CustomTags> = arc_swap::ArcSwap::from_pointee(CustomTags::default());
|
||||
|
||||
pub static ref ALL_TAGS: Arc<RwLock<Vec<String>>> = Arc::new(RwLock::new(vec![]));
|
||||
pub static ref ALL_TAGS: arc_swap::ArcSwap<Vec<String>> = 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(())
|
||||
}
|
||||
|
||||
|
||||
@@ -204,7 +204,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!(
|
||||
|
||||
@@ -118,7 +118,7 @@ pub async fn get_or_refresh_mcp_client(
|
||||
db: &DB,
|
||||
mcp_server_url: &str,
|
||||
) -> Result<McpClientCredentials, error::Error> {
|
||||
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<McpOAuthClient> =
|
||||
|
||||
@@ -203,7 +203,7 @@ impl Google {
|
||||
data: &NativeTriggerData<GoogleServiceConfig>,
|
||||
db: &DB,
|
||||
) -> Result<CreateWatchResponse> {
|
||||
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(
|
||||
|
||||
@@ -40,7 +40,7 @@ impl FullNextcloudPayload {
|
||||
webhook_token: &str,
|
||||
data: &NativeTriggerData<NextcloudServiceConfig>,
|
||||
) -> FullNextcloudPayload {
|
||||
let base_url = &*BASE_URL.read().await;
|
||||
let base_url = &**BASE_URL.load();
|
||||
let uri = generate_webhook_service_url(
|
||||
base_url,
|
||||
w_id,
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -54,11 +54,11 @@ lazy_static::lazy_static! {
|
||||
.build()
|
||||
.expect("Failed to create OAuth HTTP client");
|
||||
|
||||
pub static ref OAUTH_CLIENTS: Arc<RwLock<AllClients>> = Arc::new(RwLock::new(AllClients {
|
||||
pub static ref OAUTH_CLIENTS: arc_swap::ArcSwap<AllClients> = arc_swap::ArcSwap::from_pointee(AllClients {
|
||||
logins: HashMap::new(),
|
||||
connects: HashMap::new(),
|
||||
slack: None
|
||||
}));
|
||||
});
|
||||
}
|
||||
|
||||
/// OAuth client with associated scopes and configuration
|
||||
@@ -407,7 +407,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 +488,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,
|
||||
|
||||
@@ -1188,7 +1188,7 @@ async fn commit_completed_job<T: Serialize + Send + Sync + ValidableJson>(
|
||||
{
|
||||
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<PulledJob> = None;
|
||||
|
||||
let queries = WORKER_PULL_QUERIES.read().await;
|
||||
let queries = WORKER_PULL_QUERIES.load();
|
||||
|
||||
if queries.is_empty() {
|
||||
tracing::warn!("No pull queries available");
|
||||
|
||||
@@ -130,7 +130,7 @@ pub async fn push_scheduled_job<'c>(
|
||||
authed: Option<&Authed>,
|
||||
now_cutoff: Option<DateTime<Utc>>,
|
||||
) -> Result<Transaction<'c, Postgres>> {
|
||||
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(),
|
||||
|
||||
@@ -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()))
|
||||
|
||||
@@ -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"),
|
||||
)
|
||||
|
||||
@@ -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<Fut: std::future::Future>(
|
||||
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<FlowStatusModule>, id: &str) -> Option<FlowSt
|
||||
|
||||
pub async fn set_jwt_secret() {
|
||||
let secret = "mytestsecret".to_string();
|
||||
let mut l = JWT_SECRET.write().await;
|
||||
*l = secret;
|
||||
JWT_SECRET.store(std::sync::Arc::new(secret));
|
||||
}
|
||||
|
||||
#[derive(Debug, sqlx::FromRow, Serialize)]
|
||||
@@ -708,7 +708,8 @@ pub async fn test_for_versions<F: Future<Output = ()>>(
|
||||
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;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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<bool>,
|
||||
) -> Result<bool> {
|
||||
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;
|
||||
|
||||
|
||||
@@ -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 =
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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;
|
||||
|
||||
|
||||
@@ -521,7 +521,7 @@ pub async fn build_envs_map(context: Vec<ContextualVariable>) -> HashMap<String,
|
||||
let mut r: HashMap<String, String> =
|
||||
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());
|
||||
}
|
||||
|
||||
@@ -73,7 +73,7 @@ async fn get_common_deno_proc_envs(
|
||||
w_id: &str,
|
||||
conn: Option<&Connection>,
|
||||
) -> HashMap<String, String> {
|
||||
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();
|
||||
|
||||
@@ -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| {
|
||||
|
||||
@@ -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<bool> {
|
||||
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,
|
||||
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user