Compare commits

...

5 Commits

Author SHA1 Message Date
windmill-internal-app[bot]
64b6ac8ee8 chore: update ee-repo-ref to a0d34a319a0dcdcb3ea3b32a7654b0f42d280ac3
This commit updates the EE repository reference after PR #531 was merged in windmill-ee-private.

Previous ee-repo-ref: e37b699b89ae51601e5b86d0f96ff9e7d9227829

New ee-repo-ref: a0d34a319a0dcdcb3ea3b32a7654b0f42d280ac3

Automated by sync-ee-ref workflow.
2026-04-13 22:59:02 +00:00
Ruben Fiszel
79475e70d2 chore: pin ee-repo-ref to arcswap-refactor EE branch commit 2026-04-13 22:22:50 +00:00
Ruben Fiszel
e931be5192 refactor: convert CRITICAL_*/HUB_API_SECRET/INSTANCE_EVENTS_WEBHOOK/JWT_SECRET to ArcSwap
Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
2026-04-13 22:22:24 +00:00
Ruben Fiszel
d498f53dab refactor: convert atomic primitive globals to AtomicBool/AtomicI64
Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
2026-04-13 22:12:09 +00:00
Ruben Fiszel
7f619c2ee1 refactor: extract load helpers from reload_setting family
Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
2026-04-13 22:05:09 +00:00
18 changed files with 131 additions and 106 deletions

View File

@@ -1 +1 @@
be2f3d4d11bb7110200524d7157caab3aac53996
a0d34a319a0dcdcb3ea3b32a7654b0f42d280ac3

View File

@@ -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");
}

View File

@@ -1137,7 +1137,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 +1442,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 +1492,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 +1735,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 +1917,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 +1950,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?;
{
let mut l = lock.write().await;
*l = value;
}
Ok(())
}
@@ -1969,12 +1982,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 +2006,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 +2056,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?;
{
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 +2105,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(())
}
@@ -2766,10 +2806,7 @@ pub async fn reload_base_url_setting(conn: &Connection) -> error::Result<()> {
}
}
{
let mut l = IS_SECURE.write().await;
*l = is_secure;
}
IS_SECURE.store(is_secure, Ordering::Relaxed);
Ok(())
}
@@ -3655,8 +3692,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 +3737,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 +3775,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 +3806,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 +3928,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") {

View File

@@ -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."

View File

@@ -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

View File

@@ -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);

View File

@@ -1139,7 +1139,7 @@ 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 {
if LICENSE_KEY_VALID.load(std::sync::atomic::Ordering::Relaxed) {
LICENSE_KEY_ID.read().await.clone()
} else {
"".to_string()

View File

@@ -11,13 +11,15 @@ use crate::error;
#[cfg(not(feature = "private"))]
use serde::Deserialize;
#[cfg(not(feature = "private"))]
use std::sync::atomic::AtomicBool;
#[cfg(not(feature = "private"))]
use std::sync::Arc;
#[cfg(not(feature = "private"))]
use tokio::sync::RwLock;
#[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_VALID: AtomicBool = AtomicBool::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()));
}

View File

@@ -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] = &[

View File

@@ -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())?;

View File

@@ -14,7 +14,7 @@ use std::{
net::SocketAddr,
str::FromStr,
sync::{
atomic::{AtomicBool, Ordering},
atomic::{AtomicBool, AtomicI64, Ordering},
Arc,
},
};
@@ -216,13 +216,13 @@ lazy_static::lazy_static! {
pub static ref HUB_BASE_URL: Arc<RwLock<String>> = Arc::new(RwLock::new(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);

View File

@@ -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);
}

View File

@@ -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));

View File

@@ -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(),

View File

@@ -646,8 +646,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)]

View File

@@ -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;

View File

@@ -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 =

View File

@@ -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");