mirror of
https://github.com/windmill-labs/windmill.git
synced 2026-10-23 16:03:33 +00:00
refactor: convert atomic primitive globals to AtomicBool/AtomicI64
Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude Opus 4.6
parent
7f619c2ee1
commit
d498f53dab
+1
-1
@@ -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");
|
||||
}
|
||||
|
||||
+16
-23
@@ -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
|
||||
@@ -1742,47 +1743,47 @@ pub async fn reload_hub_api_secret_setting(conn: &Connection) {
|
||||
}
|
||||
|
||||
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),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -2805,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(())
|
||||
}
|
||||
@@ -3740,18 +3738,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(())
|
||||
}
|
||||
@@ -3938,7 +3931,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") {
|
||||
|
||||
@@ -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."
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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);
|
||||
|
||||
@@ -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()
|
||||
|
||||
@@ -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()));
|
||||
}
|
||||
|
||||
@@ -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] = &[
|
||||
|
||||
@@ -14,7 +14,7 @@ use std::{
|
||||
net::SocketAddr,
|
||||
str::FromStr,
|
||||
sync::{
|
||||
atomic::{AtomicBool, Ordering},
|
||||
atomic::{AtomicBool, AtomicI64, Ordering},
|
||||
Arc,
|
||||
},
|
||||
};
|
||||
@@ -219,10 +219,10 @@ lazy_static::lazy_static! {
|
||||
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 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);
|
||||
|
||||
|
||||
@@ -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")
|
||||
|
||||
@@ -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(),
|
||||
|
||||
@@ -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 =
|
||||
|
||||
@@ -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");
|
||||
|
||||
Reference in New Issue
Block a user