diff --git a/backend/Cargo.lock b/backend/Cargo.lock index 52cf1f564f..c95e0dac69 100644 --- a/backend/Cargo.lock +++ b/backend/Cargo.lock @@ -9683,7 +9683,6 @@ dependencies = [ "itertools 0.12.1", "lazy_static", "magic-crypt", - "mail-send", "mime_guess", "object_store", "openidconnect", @@ -9777,6 +9776,7 @@ dependencies = [ "itertools 0.12.1", "lazy_static", "magic-crypt", + "mail-send", "object_store", "prometheus", "rand 0.8.5", diff --git a/backend/ee-repo-ref.txt b/backend/ee-repo-ref.txt index 7dd78b4749..4160837664 100644 --- a/backend/ee-repo-ref.txt +++ b/backend/ee-repo-ref.txt @@ -1 +1 @@ -66d9cbb158ab9a5869a45ba253bf57f2cbb5ecb6 \ No newline at end of file +f59f9c1e55e5e93f9eb6c081847cc566611029d2 \ No newline at end of file diff --git a/backend/src/main.rs b/backend/src/main.rs index 9f2ef0d530..b3c5dc3cac 100644 --- a/backend/src/main.rs +++ b/backend/src/main.rs @@ -18,11 +18,11 @@ use tokio::fs::DirBuilder; use windmill_api::HTTP_CLIENT; use windmill_common::{ global_settings::{ - BASE_URL_SETTING, BUNFIG_INSTALL_SCOPES_SETTING, CUSTOM_TAGS_SETTING, - DEFAULT_TAGS_PER_WORKSPACE_SETTING, ENV_SETTINGS, EXPOSE_DEBUG_METRICS_SETTING, - EXPOSE_METRICS_SETTING, EXTRA_PIP_INDEX_URL_SETTING, HUB_BASE_URL_SETTING, - JOB_DEFAULT_TIMEOUT_SECS_SETTING, KEEP_JOB_DIR_SETTING, LICENSE_KEY_SETTING, - NPM_CONFIG_REGISTRY_SETTING, OAUTH_SETTING, PIP_INDEX_URL_SETTING, + BASE_URL_SETTING, BUNFIG_INSTALL_SCOPES_SETTING, CRITICAL_ERROR_CHANNELS_SETTING, + CUSTOM_TAGS_SETTING, DEFAULT_TAGS_PER_WORKSPACE_SETTING, ENV_SETTINGS, + EXPOSE_DEBUG_METRICS_SETTING, EXPOSE_METRICS_SETTING, EXTRA_PIP_INDEX_URL_SETTING, + HUB_BASE_URL_SETTING, JOB_DEFAULT_TIMEOUT_SECS_SETTING, KEEP_JOB_DIR_SETTING, + LICENSE_KEY_SETTING, NPM_CONFIG_REGISTRY_SETTING, OAUTH_SETTING, PIP_INDEX_URL_SETTING, REQUEST_SIZE_LIMIT_SETTING, REQUIRE_PREEXISTING_USER_FOR_OAUTH_SETTING, RETENTION_PERIOD_SECS_SETTING, SAML_METADATA_SETTING, SCIM_TOKEN_SETTING, }, @@ -47,10 +47,11 @@ use windmill_worker::{ use crate::monitor::{ initial_load, load_keep_job_dir, load_require_preexisting_user, load_tag_per_workspace_enabled, monitor_db, monitor_pool, reload_base_url_setting, reload_bunfig_install_scopes_setting, - reload_extra_pip_index_url_setting, reload_hub_base_url_setting, - reload_job_default_timeout_setting, reload_license_key, reload_npm_config_registry_setting, - reload_pip_index_url_setting, reload_retention_period_setting, reload_scim_token_setting, - reload_server_config, reload_worker_config, + reload_critical_error_channels_setting, reload_extra_pip_index_url_setting, + reload_hub_base_url_setting, reload_job_default_timeout_setting, reload_license_key, + reload_npm_config_registry_setting, reload_pip_index_url_setting, + reload_retention_period_setting, reload_scim_token_setting, reload_server_config, + reload_worker_config, }; #[cfg(feature = "parquet")] @@ -503,6 +504,11 @@ Windmill Community Edition {GIT_VERSION} tracing::error!(error = %e, "Could not reload hub base url setting"); } }, + CRITICAL_ERROR_CHANNELS_SETTING => { + if let Err(e) = reload_critical_error_channels_setting(&db).await { + tracing::error!(error = %e, "Could not reload critical error emails setting"); + } + }, a @_ => { tracing::info!("Unrecognized Global Setting Change Payload: {:?}", a); } diff --git a/backend/src/monitor.rs b/backend/src/monitor.rs index 08a778182c..0fbe4375b6 100644 --- a/backend/src/monitor.rs +++ b/backend/src/monitor.rs @@ -22,15 +22,17 @@ use windmill_api::{ DEFAULT_BODY_LIMIT, IS_SECURE, OAUTH_CLIENTS, REQUEST_SIZE_LIMIT, SAML_METADATA, SCIM_TOKEN, }; use windmill_common::{ + ee::CriticalErrorChannel, error, flow_status::FlowStatusModule, global_settings::{ - BASE_URL_SETTING, BUNFIG_INSTALL_SCOPES_SETTING, DEFAULT_TAGS_PER_WORKSPACE_SETTING, - EXPOSE_DEBUG_METRICS_SETTING, EXPOSE_METRICS_SETTING, EXTRA_PIP_INDEX_URL_SETTING, - HUB_BASE_URL_SETTING, JOB_DEFAULT_TIMEOUT_SECS_SETTING, KEEP_JOB_DIR_SETTING, - LICENSE_KEY_SETTING, NPM_CONFIG_REGISTRY_SETTING, OAUTH_SETTING, PIP_INDEX_URL_SETTING, - REQUEST_SIZE_LIMIT_SETTING, REQUIRE_PREEXISTING_USER_FOR_OAUTH_SETTING, - RETENTION_PERIOD_SECS_SETTING, SAML_METADATA_SETTING, SCIM_TOKEN_SETTING, + BASE_URL_SETTING, BUNFIG_INSTALL_SCOPES_SETTING, CRITICAL_ERROR_CHANNELS_SETTING, + DEFAULT_TAGS_PER_WORKSPACE_SETTING, EXPOSE_DEBUG_METRICS_SETTING, EXPOSE_METRICS_SETTING, + EXTRA_PIP_INDEX_URL_SETTING, HUB_BASE_URL_SETTING, JOB_DEFAULT_TIMEOUT_SECS_SETTING, + KEEP_JOB_DIR_SETTING, LICENSE_KEY_SETTING, NPM_CONFIG_REGISTRY_SETTING, OAUTH_SETTING, + PIP_INDEX_URL_SETTING, REQUEST_SIZE_LIMIT_SETTING, + REQUIRE_PREEXISTING_USER_FOR_OAUTH_SETTING, RETENTION_PERIOD_SECS_SETTING, + SAML_METADATA_SETTING, SCIM_TOKEN_SETTING, }, jobs::QueuedJob, oauth2::REQUIRE_PREEXISTING_USER_FOR_OAUTH, @@ -40,7 +42,8 @@ use windmill_common::{ load_worker_config, reload_custom_tags_setting, DEFAULT_TAGS_PER_WORKSPACE, SERVER_CONFIG, WORKER_CONFIG, }, - BASE_URL, DB, DEFAULT_HUB_BASE_URL, HUB_BASE_URL, METRICS_DEBUG_ENABLED, METRICS_ENABLED, + BASE_URL, CRITICAL_ERROR_CHANNELS, DB, DEFAULT_HUB_BASE_URL, HUB_BASE_URL, + METRICS_DEBUG_ENABLED, METRICS_ENABLED, }; use windmill_queue::cancel_job; use windmill_worker::{ @@ -143,6 +146,10 @@ pub async fn initial_load( tracing::error!("Error reloading hub base url: {:?}", e) } + if let Err(e) = reload_critical_error_channels_setting(&db).await { + tracing::error!("Could not reload critical error emails setting: {:?}", e); + } + #[cfg(feature = "parquet")] if !_is_agent { reload_s3_cache_setting(&db).await; @@ -1074,3 +1081,27 @@ pub async fn reload_hub_base_url_setting(db: &DB, server_mode: bool) -> error::R Ok(()) } + +pub async fn reload_critical_error_channels_setting(db: &DB) -> error::Result<()> { + let critical_error_channels = + load_value_from_global_settings(db, CRITICAL_ERROR_CHANNELS_SETTING).await?; + + let critical_error_channels = if let Some(q) = critical_error_channels { + if let Ok(v) = serde_json::from_value::>(q.clone()) { + v + } else { + tracing::error!( + "Could not parse critical_error_emails setting as an array of channels, found: {:#?}", + &q + ); + vec![] + } + } else { + vec![] + }; + + let mut l = CRITICAL_ERROR_CHANNELS.write().await; + *l = critical_error_channels; + + Ok(()) +} diff --git a/backend/windmill-api/Cargo.toml b/backend/windmill-api/Cargo.toml index 00e965f292..0769d3db34 100644 --- a/backend/windmill-api/Cargo.toml +++ b/backend/windmill-api/Cargo.toml @@ -72,7 +72,6 @@ async_zip.workspace = true rsmq_async.workspace = true regex.workspace = true bytes.workspace = true -mail-send.workspace = true samael = { workspace = true, optional = true } async-recursion.workspace = true rsa.workspace = true diff --git a/backend/windmill-api/src/flows.rs b/backend/windmill-api/src/flows.rs index cc0980e4a3..7da1c6582a 100644 --- a/backend/windmill-api/src/flows.rs +++ b/backend/windmill-api/src/flows.rs @@ -543,7 +543,7 @@ async fn update_flow( clear_schedule(tx.transaction_mut(), &schedule.path, &w_id).await?; if schedule.enabled { - tx = push_scheduled_job(&db, tx, schedule).await?; + tx = push_scheduled_job(&db, tx, &schedule).await?; } } diff --git a/backend/windmill-api/src/schedule.rs b/backend/windmill-api/src/schedule.rs index be98471c7a..7cda4a1157 100644 --- a/backend/windmill-api/src/schedule.rs +++ b/backend/windmill-api/src/schedule.rs @@ -221,7 +221,7 @@ async fn create_schedule( .await?; if ns.enabled.unwrap_or(true) { - tx = push_scheduled_job(&db, tx, schedule).await? + tx = push_scheduled_job(&db, tx, &schedule).await? } tx.commit().await?; @@ -303,7 +303,7 @@ async fn edit_schedule( .await?; if schedule.enabled { - tx = push_scheduled_job(&db, tx, schedule).await?; + tx = push_scheduled_job(&db, tx, &schedule).await?; } tx.commit().await?; @@ -512,7 +512,7 @@ pub async fn set_enabled( .await?; if payload.enabled { - tx = push_scheduled_job(&db, tx, schedule).await?; + tx = push_scheduled_job(&db, tx, &schedule).await?; } tx.commit().await?; @@ -560,7 +560,7 @@ pub async fn set_enabled( // .await?; // if payload.enabled { -// tx = push_scheduled_job(&db, tx, schedule).await?; +// tx = push_scheduled_job(&db, tx, &schedule).await?; // } // tx.commit().await?; diff --git a/backend/windmill-api/src/scripts.rs b/backend/windmill-api/src/scripts.rs index 4cc923a732..b19481e3cf 100644 --- a/backend/windmill-api/src/scripts.rs +++ b/backend/windmill-api/src/scripts.rs @@ -554,7 +554,7 @@ async fn create_script( clear_schedule(tx.transaction_mut(), &schedule.path, &w_id).await?; if schedule.enabled { - tx = push_scheduled_job(&db, tx, schedule).await?; + tx = push_scheduled_job(&db, tx, &schedule).await?; } } } else { diff --git a/backend/windmill-api/src/settings.rs b/backend/windmill-api/src/settings.rs index 7bd13ccfab..b825286b43 100644 --- a/backend/windmill-api/src/settings.rs +++ b/backend/windmill-api/src/settings.rs @@ -21,15 +21,17 @@ use axum::{ Json, Router, }; -use mail_send::{mail_builder::MessageBuilder, SmtpClientBuilder}; use serde::Deserialize; -use tokio::time::timeout; use windmill_common::{ - error::{self, to_anyhow, JsonResult, Result}, + error::{self, JsonResult, Result}, global_settings::{AUTOMATE_USERNAME_CREATION_SETTING, ENV_SETTINGS, HUB_BASE_URL_SETTING}, server::Smtp, + utils::send_email, }; +#[cfg(feature = "parquet")] +use windmill_common::error::to_anyhow; + pub fn global_service() -> Router { #[warn(unused_mut)] let r = Router::new() @@ -67,34 +69,17 @@ pub async fn test_email( require_super_admin(&db, &authed.email).await?; let smtp = test_email.smtp; let to = test_email.to; - let mut client = SmtpClientBuilder::new(smtp.host, smtp.port) - .implicit_tls(smtp.tls_implicit.unwrap_or(false)); - if std::env::var("ACCEPT_INVALID_CERTS").is_ok() { - client = client.allow_invalid_certs(); - } - let client = if let (Some(username), Some(password)) = (smtp.username, smtp.password) { - if !username.is_empty() { - client.credentials((username, password)) - } else { - client - } - } else { - client - }; - let message = MessageBuilder::new() - .from(("Windmill", smtp.from.as_str())) - .to(to.clone()) - .subject("Test email from Windmill") - .text_body("Test email content"); - let dur = Duration::from_secs(3); - timeout(dur, client.connect()) - .await - .map_err(to_anyhow)? - .map_err(to_anyhow)? - .send(message) - .await - .map_err(to_anyhow)?; - tracing::info!("Sent test email to {to}"); + + let client_timeout = Duration::from_secs(3); + send_email( + "Test email from Windmill", + "Test email content", + vec![to], + smtp, + Some(client_timeout), + ) + .await?; + Ok("Sent test email".to_string()) } diff --git a/backend/windmill-api/src/workspaces.rs b/backend/windmill-api/src/workspaces.rs index ee542d9907..b85302d236 100644 --- a/backend/windmill-api/src/workspaces.rs +++ b/backend/windmill-api/src/workspaces.rs @@ -53,7 +53,6 @@ use windmill_common::{ variables::ExportableListableVariable, }; use windmill_git_sync::handle_deployment_metadata; -use windmill_queue::QueueTransaction; use crate::oauth2_ee::InstanceEvent; use crate::variables::{decrypt, encrypt}; @@ -488,25 +487,25 @@ async fn run_slack_message_test_job( json!(format!("$res:{WORKSPACE_SLACK_BOT_TOKEN_PATH}")), ); - let tx: QueueTransaction<'_, _> = (rsmq.clone(), db.begin().await?).into(); - let (uuid, tx) = windmill_queue::handle_on_failure( + let uuid = windmill_queue::push_error_handler( &db, - tx, + rsmq, Uuid::parse_str("00000000-0000-0000-0000-000000000000")?, - "slack_message_test", - "slack_message_test", + None, + Some("slack_message_test".to_string()), false, w_id.as_str(), &format!("script/{}", req.hub_script_path.as_str()), sqlx::types::Json(&fake_result), - 0, - Utc::now(), + None, + Some(Utc::now()), Some(json!(extra_args)), authed.email.as_str(), + false, + false, None, // Note: we could mark it as high priority to return result quickly to the user ) .await?; - tx.commit().await?; Ok(Json(RunSlackMessageTestJobResponse { job_uuid: uuid.to_string(), diff --git a/backend/windmill-common/Cargo.toml b/backend/windmill-common/Cargo.toml index 5e5c8f4673..9ac9ea67a7 100644 --- a/backend/windmill-common/Cargo.toml +++ b/backend/windmill-common/Cargo.toml @@ -46,4 +46,5 @@ object_store = { workspace = true, optional = true } prometheus = { workspace = true, optional = true } aws-config = { workspace = true, optional = true } aws-sdk-sts = { workspace = true, optional = true } -indexmap.workspace = true \ No newline at end of file +indexmap.workspace = true +mail-send.workspace = true \ No newline at end of file diff --git a/backend/windmill-common/src/ee.rs b/backend/windmill-common/src/ee.rs index c942447bbc..d3ffd98584 100644 --- a/backend/windmill-common/src/ee.rs +++ b/backend/windmill-common/src/ee.rs @@ -1,4 +1,5 @@ use crate::ee::LicensePlan::Community; +use serde::Deserialize; use std::sync::Arc; use tokio::sync::RwLock; @@ -18,3 +19,9 @@ pub async fn get_license_plan() -> LicensePlan { // Implementation is not open source return Community; } + +#[derive(Deserialize)] +#[serde(untagged)] +pub enum CriticalErrorChannel {} + +pub async fn trigger_critical_error_channels(_error_message: String) {} diff --git a/backend/windmill-common/src/error.rs b/backend/windmill-common/src/error.rs index e3c1a7f95d..32e9740d78 100644 --- a/backend/windmill-common/src/error.rs +++ b/backend/windmill-common/src/error.rs @@ -8,10 +8,7 @@ use axum::body::Body; use axum::response::Response; -use axum::{ - response::IntoResponse, - response::Json, -}; +use axum::{response::IntoResponse, response::Json}; use hyper::StatusCode; use sqlx::migrate::MigrateError; @@ -45,6 +42,8 @@ pub enum Error { SqlErr(#[from] sqlx::Error), #[error("Bad request: {0}")] BadRequest(String), + #[error("Quota exceeded: {0}")] + QuotaExceeded(String), #[error("Internal: {0}")] InternalErr(String), #[error("Hexadecimal decoding error: {0}")] @@ -81,9 +80,10 @@ impl IntoResponse for Error { Self::NotFound(_) => axum::http::StatusCode::NOT_FOUND, Self::NotAuthorized(_) => axum::http::StatusCode::UNAUTHORIZED, Self::RequireAdmin(_) => axum::http::StatusCode::FORBIDDEN, - Self::SqlErr(_) | Self::BadRequest(_) | Self::OpenAIError(_) => { - axum::http::StatusCode::BAD_REQUEST - } + Self::SqlErr(_) + | Self::BadRequest(_) + | Self::OpenAIError(_) + | Self::QuotaExceeded(_) => axum::http::StatusCode::BAD_REQUEST, _ => axum::http::StatusCode::INTERNAL_SERVER_ERROR, }; diff --git a/backend/windmill-common/src/global_settings.rs b/backend/windmill-common/src/global_settings.rs index d24b6b7070..335bb80813 100644 --- a/backend/windmill-common/src/global_settings.rs +++ b/backend/windmill-common/src/global_settings.rs @@ -24,6 +24,7 @@ pub const OBJECT_STORE_CACHE_CONFIG_SETTING: &str = "object_store_cache_config"; pub const AUTOMATE_USERNAME_CREATION_SETTING: &str = "automate_username_creation"; pub const HUB_BASE_URL_SETTING: &str = "hub_base_url"; +pub const CRITICAL_ERROR_CHANNELS_SETTING: &str = "critical_error_channels"; pub const ENV_SETTINGS: [&str; 50] = [ "DISABLE_NSJAIL", diff --git a/backend/windmill-common/src/lib.rs b/backend/windmill-common/src/lib.rs index c90f0e702c..7e9ce4ca84 100644 --- a/backend/windmill-common/src/lib.rs +++ b/backend/windmill-common/src/lib.rs @@ -11,6 +11,7 @@ use std::{ sync::{atomic::AtomicBool, Arc}, }; +use ee::CriticalErrorChannel; use error::Error; use scripts::ScriptLang; use sqlx::{Pool, Postgres}; @@ -73,6 +74,9 @@ lazy_static::lazy_static! { pub static ref IS_READY: std::sync::atomic::AtomicBool = std::sync::atomic::AtomicBool::new(false); pub static ref HUB_BASE_URL: Arc> = Arc::new(RwLock::new(DEFAULT_HUB_BASE_URL.to_string())); + + + pub static ref CRITICAL_ERROR_CHANNELS: Arc>> = Arc::new(RwLock::new(vec![])); } pub async fn shutdown_signal( diff --git a/backend/windmill-common/src/utils.rs b/backend/windmill-common/src/utils.rs index 40dcc6531e..fc7157308e 100644 --- a/backend/windmill-common/src/utils.rs +++ b/backend/windmill-common/src/utils.rs @@ -6,11 +6,14 @@ * LICENSE-AGPL for a copy of the license. */ -use crate::ee::LICENSE_KEY_ID; +use crate::ee::{trigger_critical_error_channels, LICENSE_KEY_ID}; use crate::error::{to_anyhow, Error, Result}; use crate::global_settings::UNIQUE_ID_SETTING; +use crate::server::Smtp; use crate::DB; use git_version::git_version; +use mail_send::mail_builder::MessageBuilder; +use mail_send::SmtpClientBuilder; use rand::{distributions::Alphanumeric, thread_rng, Rng}; use serde::{Deserialize, Serialize}; use sha2::{Digest, Sha256}; @@ -85,12 +88,20 @@ pub async fn query_elems_from_hub( url: &str, query_params: Option>, db: &DB, -) -> Result<(reqwest::StatusCode, reqwest::header::HeaderMap, axum::body::Body)> { +) -> Result<( + reqwest::StatusCode, + reqwest::header::HeaderMap, + axum::body::Body, +)> { let response = http_get_from_hub(http_client, url, false, query_params, db).await?; let status = response.status(); - Ok((status, response.headers().clone(), axum::body::Body::from_stream(response.bytes_stream()))) + Ok(( + status, + response.headers().clone(), + axum::body::Body::from_stream(response.bytes_stream()), + )) } pub async fn http_get_from_hub( @@ -167,3 +178,60 @@ pub enum Mode { Server, Standalone, } + +pub async fn send_email( + subject: &str, + content: &str, + to: Vec, + smtp: Smtp, + client_timeout: Option, +) -> Result<()> { + let mut client = SmtpClientBuilder::new(smtp.host, smtp.port) + .implicit_tls(smtp.tls_implicit.unwrap_or(false)); + if std::env::var("ACCEPT_INVALID_CERTS").is_ok() { + client = client.allow_invalid_certs(); + } + let client = if let (Some(username), Some(password)) = (smtp.username, smtp.password) { + if !username.is_empty() { + client.credentials((username, password)) + } else { + client + } + } else { + client + }; + let message = MessageBuilder::new() + .from(("Windmill", smtp.from.as_str())) + .to(to.clone()) + .subject(subject) + .text_body(content); + + match client_timeout { + Some(timeout) => { + tokio::time::timeout(timeout, client.connect()) + .await + .map_err(to_anyhow)? + .map_err(to_anyhow)? + .send(message) + .await + .map_err(to_anyhow)?; + } + None => { + client + .connect() + .await + .map_err(to_anyhow)? + .send(message) + .await + .map_err(to_anyhow)?; + } + } + tracing::info!("Sent email to {:#?}: {subject}", to); + + return Ok(()); +} + +pub async fn report_critical_error(error_message: String) -> () { + tracing::error!("CRITICAL ERROR: {error_message}"); + trigger_critical_error_channels(error_message).await; +} diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs index c978b08bd6..0b3bd7a791 100644 --- a/backend/windmill-queue/src/jobs.rs +++ b/backend/windmill-queue/src/jobs.rs @@ -46,7 +46,7 @@ use windmill_audit::ActionKind; use windmill_common::worker::PriorityTags; use windmill_common::{ db::{Authed, UserDB}, - error::{self, Error}, + error::{self, to_anyhow, Error}, flow_status::{ BranchAllStatus, FlowCleanupModule, FlowStatus, FlowStatusModule, FlowStatusModuleWParent, Iterator, JobResult, RestartedFrom, RetryStatus, MAX_RETRY_ATTEMPTS, MAX_RETRY_INTERVAL, @@ -61,8 +61,9 @@ use windmill_common::{ schedule::Schedule, scripts::{ScriptHash, ScriptLang}, users::{SUPERADMIN_NOTIFICATION_EMAIL, SUPERADMIN_SECRET_EMAIL}, + utils::report_critical_error, worker::{to_raw_value, DEFAULT_TAGS_PER_WORKSPACE, NO_LOGS, WORKER_CONFIG}, - DB, METRICS_ENABLED, + BASE_URL, DB, METRICS_ENABLED, }; #[cfg(feature = "cloud")] @@ -119,8 +120,12 @@ const MAX_FREE_EXECS: i32 = 1000; const MAX_FREE_CONCURRENT_RUNS: i32 = 30; const ERROR_HANDLER_USERNAME: &str = "error_handler"; +const SCHEDULE_ERROR_HANDLER_USERNAME: &str = "schedule_error_handler"; +const SCHEDULE_RECOVERY_HANDLER_USERNAME: &str = "schedule_recovery_handler"; const ERROR_HANDLER_USER_GROUP: &str = "g/error_handler"; const ERROR_HANDLER_USER_EMAIL: &str = "error_handler@windmill.dev"; +const SCHEDULE_ERROR_HANDLER_USER_EMAIL: &str = "schedule_error_handler@windmill.dev"; +const SCHEDULE_RECOVERY_HANDLER_USER_EMAIL: &str = "schedule_recovery_handler@windmill.dev"; #[derive(Clone, Debug)] pub struct CanceledBy { @@ -615,44 +620,80 @@ pub async fn add_completed_job< } } else { if queued_job.schedule_path.is_some() && queued_job.script_path.is_some() { - let schedule_handlers_tx: QueueTransaction<'_, R> = - (rsmq.clone(), db.begin().await?).into(); - match apply_schedule_handlers( - schedule_handlers_tx, - db, - queued_job.schedule_path.as_ref().unwrap(), - queued_job.script_path.as_ref().unwrap(), + let schedule_path = queued_job.schedule_path.as_ref().unwrap(); + let script_path = queued_job.script_path.as_ref().unwrap(); + + let schedule = get_schedule_opt( + tx.transaction_mut(), &queued_job.workspace_id, - success, - result, - job_id, - queued_job.started_at.unwrap_or(chrono::Utc::now()), - queued_job.priority, + schedule_path, ) - .await - { - Ok((skip, mut schedule_handlers_tx)) => { - skip_downstream_error_handlers = skip; + .await?; - if !queued_job.is_flow() { - // script only - schedule_handlers_tx = handle_maybe_scheduled_job( - schedule_handlers_tx, - db, - queued_job.schedule_path.as_ref().unwrap(), - queued_job.script_path.as_ref().unwrap(), - &queued_job.workspace_id, - ) - .await?; + if let Some(schedule) = schedule { + skip_downstream_error_handlers = schedule.ws_error_handler_muted; + + if !queued_job.is_flow() { + // script only + if let Err(err) = handle_maybe_scheduled_job( + rsmq.clone(), + db, + queued_job, + &schedule, + script_path, + &queued_job.workspace_id, + ) + .await + { + match err { + Error::QuotaExceeded(_) => return Err(err.into()), + // scheduling next job failed and could not disable schedule => make zombie job to retry + _ => return Ok(job_id), + } + }; + } + + if let Err(err) = apply_schedule_handlers( + rsmq.clone(), + db, + &schedule, + script_path, + &queued_job.workspace_id, + success, + result, + job_id, + queued_job.started_at.unwrap_or(chrono::Utc::now()), + queued_job.priority, + ) + .await + { + if !success { + tracing::error!("Could not apply schedule error handler: {}", err); + let base_url = BASE_URL.read().await; + let w_id: &String = &queued_job.workspace_id; + if !matches!(err, Error::QuotaExceeded(_)) { + report_error_to_workspace_handler_or_critical_side_channel( + rsmq.clone(), + &queued_job, + db, + format!( + "Failed to push schedule error handler job to handle failed job ({base_url}/run/{}?workspace={w_id}): {}", + queued_job.id, + err + ), + ) + .await; + } + } else { + tracing::error!("Could not apply schedule recovery handler: {}", err); } - - schedule_handlers_tx.commit().await?; - } - Err(err) => { - skip_downstream_error_handlers = true; - tracing::error!("Could not apply schedule handlers with error: {}", err); - } - }; + }; + } else { + tracing::error!( + "Schedule {schedule_path} in {} not found. Impossible to schedule again and apply schedule handlers", + &queued_job.workspace_id + ); + } } } if queued_job.concurrent_limit.is_some() { @@ -735,37 +776,90 @@ pub async fn add_completed_job< } #[cfg(feature = "enterprise")] - if !skip_downstream_error_handlers - && matches!(queued_job.job_kind, JobKind::Flow | JobKind::Script) - && queued_job.parent_job.is_none() - && !success - { - tracing::info!( - "Sending error of job {} to error handlers (if any)", - queued_job.id - ); - if let Err(e) = send_error_to_global_handler(rsmq.clone(), &queued_job, db, result).await { - tracing::error!( - "Could not run global error handler for job {}: {}", - &queued_job.id, - e - ); - } - - if let Err(e) = send_error_to_workspace_handler( - rsmq.clone(), - &queued_job, - canceled_by.is_some(), - db, - result, - ) - .await + if !success { + if queued_job.email == ERROR_HANDLER_USER_EMAIL { + let base_url = BASE_URL.read().await; + let w_id = &queued_job.workspace_id; + report_critical_error(format!( + "Workspace error handler job failed ({base_url}/run/{}?workspace={w_id}){}", + queued_job.id, + queued_job + .parent_job + .map(|id| format!( + " trying to handle failed job ({base_url}/run/{id}?workspace={w_id})" + )) + .unwrap_or("".to_string()), + )) + .await; + } else if queued_job.email == SCHEDULE_ERROR_HANDLER_USER_EMAIL { + let base_url = BASE_URL.read().await; + let w_id = &queued_job.workspace_id; + report_error_to_workspace_handler_or_critical_side_channel( + rsmq.clone(), + &queued_job, + db, + format!( + "Schedule error handler job failed ({base_url}/run/{}?workspace={w_id}){}", + queued_job.id, + queued_job + .parent_job + .map(|id| format!( + " trying to handle failed job: {base_url}/run/{id}?workspace={w_id}" + )) + .unwrap_or("".to_string()), + ), + ) + .await; + } else if !skip_downstream_error_handlers + && matches!(queued_job.job_kind, JobKind::Flow | JobKind::Script) + && queued_job.parent_job.is_none() { - tracing::error!( - "Could not run workspace error handler for job {}: {}", - &queued_job.id, - e + let result = serde_json::from_str( + &serde_json::to_string(result.0).unwrap_or_else(|_| "{}".to_string()), + ) + .unwrap_or_else(|_| json!({})); + let result = if result.is_object() || result.is_null() { + result + } else { + json!({ "error": result }) + }; + tracing::info!( + "Sending error of job {} to error handlers (if any)", + queued_job.id ); + if let Err(e) = + send_error_to_global_handler(rsmq.clone(), &queued_job, db, Json(&result)).await + { + tracing::error!( + "Could not run global error handler for job {}: {}", + &queued_job.id, + e + ); + } + + if let Err(err) = send_error_to_workspace_handler( + rsmq.clone(), + &queued_job, + canceled_by.is_some(), + db, + Json(&result), + ) + .await + { + match err { + Error::QuotaExceeded(_) => {} + _ => { + let base_url = BASE_URL.read().await; + let w_id: &String = &queued_job.workspace_id; + report_critical_error(format!( + "Could not push workspace error handler for failed job ({base_url}/run/{}?workspace={w_id}): {}", + queued_job.id, + err + )) + .await; + } + } + } } } @@ -846,113 +940,6 @@ pub async fn add_completed_job< Ok(queued_job.id) } -pub async fn run_error_handler< - 'a, - T: Serialize + Send + Sync, - R: rsmq_async::RsmqConnection + Clone + Send, ->( - rsmq: Option, - queued_job: &QueuedJob, - db: &Pool, - result: Json<&'a T>, - error_handler_path: &str, - error_handler_extra_args: Option, - is_global: bool, -) -> Result<(), Error> { - let w_id = &queued_job.workspace_id; - let handler_w_id = if is_global { "admins" } else { w_id }; // script workspace id - let job_id = queued_job.id; - let (job_payload, tag) = - get_payload_tag_from_prefixed_path(&error_handler_path, db, handler_w_id).await?; - - let mut extra = HashMap::new(); - extra.insert("workspace_id".to_string(), to_raw_value(&handler_w_id)); - extra.insert("job_id".to_string(), to_raw_value(&job_id)); - extra.insert("path".to_string(), to_raw_value(&queued_job.script_path)); - extra.insert( - "is_flow".to_string(), - to_raw_value(&queued_job.raw_flow.is_some()), - ); - extra.insert( - "started_at".to_string(), - to_raw_value(&queued_job.started_at), - ); - extra.insert("email".to_string(), to_raw_value(&queued_job.email)); - - if let Some(schedule_path) = &queued_job.schedule_path { - extra.insert("schedule_path".to_string(), to_raw_value(schedule_path)); - } - - // TODO(gbouv): REMOVE THIS after December 1st 2023 and ping users to re-save their error handlers - if error_handler_path - .to_string() - .eq("script/hub/5792/workspace-or-schedule-error-handler-slack") - { - // default slack error handler being used -> we need to inject the slack token - let slack_resource = format!("$res:{WORKSPACE_SLACK_BOT_TOKEN_PATH}"); - extra.insert("slack".to_string(), to_raw_value(&slack_resource)); - } - - if let Some(extra_args) = error_handler_extra_args { - if let serde_json::Value::Object(args_m) = extra_args { - for (k, v) in args_m { - extra.insert(k, to_raw_value(&v)); - } - } else { - return Err(error::Error::ExecutionErr( - "args of scripts needs to be dict".to_string(), - )); - } - } - - let tx = PushIsolationLevel::IsolatedRoot(db.clone(), rsmq); - - let (uuid, tx) = push( - &db, - tx, - handler_w_id, - job_payload, - PushArgs { extra, args: result.to_owned() }, - if is_global { - "global" - } else { - ERROR_HANDLER_USERNAME - }, - if is_global { - SUPERADMIN_SECRET_EMAIL - } else { - ERROR_HANDLER_USER_EMAIL - }, - if is_global { - SUPERADMIN_SECRET_EMAIL.to_string() - } else { - ERROR_HANDLER_USER_GROUP.to_string() - }, - None, - None, - Some(job_id), - Some(job_id), - None, - false, - false, - None, - true, - tag, - None, - None, - None, - ) - .await?; - tx.commit().await?; - - let error_handler_type = if is_global { "global" } else { "workspace" }; - tracing::info!( - "Sent error of job {job_id} to {error_handler_type} error handler under uuid {uuid}" - ); - - Ok(()) -} - pub async fn send_error_to_global_handler< 'a, T: Serialize + Send + Sync, @@ -971,22 +958,85 @@ pub async fn send_error_to_global_handler< } else { format!("script/{}", global_error_handler) }; - let result = sanitize_result(result); - run_error_handler( - rsmq, - queued_job, + push_error_handler( db, - Json(&result), + rsmq, + queued_job.id, + queued_job.schedule_path.clone(), + queued_job.script_path.clone(), + queued_job.is_flow(), + &queued_job.workspace_id, &prefixed_global_error_handler_path, + result, None, + queued_job.started_at, + None, + &queued_job.email, + false, true, + None, ) - .await? + .await?; } Ok(()) } +pub async fn report_error_to_workspace_handler_or_critical_side_channel< + R: rsmq_async::RsmqConnection + Clone + Send, +>( + rsmq: Option, + queued_job: &QueuedJob, + db: &Pool, + error_message: String, +) -> () { + let w_id = &queued_job.workspace_id; + let (error_handler, error_handler_extra_args) = sqlx::query_as::<_, (Option, Option)>( + "SELECT error_handler, error_handler_extra_args FROM workspace_settings WHERE workspace_id = $1", + ).bind(&w_id) + .fetch_optional(db) + .await + .ok() + .flatten() + .unwrap_or((None, None)); + + if let Some(error_handler) = error_handler { + if let Err(err) = push_error_handler( + db, + rsmq, + queued_job.id, + queued_job.schedule_path.clone(), + queued_job.script_path.clone(), + queued_job.is_flow(), + w_id, + &error_handler, + Json(&json!({ + "error": { + "message": error_message + } + })), + None, + queued_job.started_at, + error_handler_extra_args, + &queued_job.email, + false, + false, + None, + ) + .await + { + tracing::error!( + "Could not push workspace error handler trying to handle critical error ({error_message}) of failed job ({}): {}", + queued_job.id, + err + ); + report_critical_error(error_message).await; + } + } else { + report_critical_error(error_message).await; + } +} + pub async fn send_error_to_workspace_handler< 'a, 'c, @@ -1000,11 +1050,10 @@ pub async fn send_error_to_workspace_handler< result: Json<&'a T>, ) -> Result<(), Error> { let w_id = &queued_job.workspace_id; - let mut tx = db.begin().await?; let (error_handler, error_handler_extra_args, error_handler_muted_on_cancel) = sqlx::query_as::<_, (Option, Option, bool)>( "SELECT error_handler, error_handler_extra_args, error_handler_muted_on_cancel FROM workspace_settings WHERE workspace_id = $1", ).bind(&w_id) - .fetch_optional(&mut *tx) + .fetch_optional(db) .await .context("fetching error handler info from workspace_settings")? .ok_or_else(|| Error::InternalErr(format!("no workspace settings for id {w_id}")))?; @@ -1038,18 +1087,27 @@ pub async fn send_error_to_workspace_handler< let muted = ws_error_handler_muted.unwrap_or(false); if !muted { - let result = sanitize_result(result); tracing::info!("workspace error handled for job {}", &queued_job.id); - run_error_handler( - rsmq, - queued_job, + + push_error_handler( db, - Json(&result), + rsmq, + queued_job.id, + queued_job.schedule_path.clone(), + queued_job.script_path.clone(), + queued_job.is_flow(), + &queued_job.workspace_id, &error_handler, + result, + None, + queued_job.started_at, error_handler_extra_args, + &queued_job.email, false, + false, + None, ) - .await? + .await?; } } @@ -1058,83 +1116,71 @@ pub async fn send_error_to_workspace_handler< #[instrument(level = "trace", skip_all)] pub async fn handle_maybe_scheduled_job<'c, R: rsmq_async::RsmqConnection + Clone + Send + 'c>( - mut tx: QueueTransaction<'c, R>, + rsmq: Option, db: &Pool, - schedule_path: &str, + job: &QueuedJob, + schedule: &Schedule, script_path: &str, w_id: &str, -) -> windmill_common::error::Result> { - tracing::info!("Schedule {schedule_path} scheduling next job for {script_path} in {w_id}",); - let schedule = get_schedule_opt(tx.transaction_mut(), w_id, schedule_path).await?; - - if schedule.is_none() { - tracing::error!( - "Schedule {schedule_path} in {w_id} not found. Impossible to schedule again" - ); - return Ok(tx); - } - - let schedule = schedule.unwrap(); +) -> windmill_common::error::Result<()> { + tracing::info!( + "Schedule {} scheduling next job for {} in {w_id}", + schedule.path, + schedule.script_path + ); if schedule.enabled && script_path == schedule.script_path { - let res = push_scheduled_job( - db, - tx, - Schedule { - workspace_id: w_id.to_owned(), - path: schedule.path.clone(), - edited_by: schedule.edited_by, - edited_at: schedule.edited_at, - schedule: schedule.schedule, - timezone: schedule.timezone, - enabled: schedule.enabled, - script_path: schedule.script_path, - is_flow: schedule.is_flow, - args: schedule - .args - .and_then(|e| serde_json::to_value(e).map_or(None, |v| Some(v))), - extra_perms: serde_json::to_value(schedule.extra_perms).expect("hashmap -> json"), - email: schedule.email, - error: None, - on_failure: schedule.on_failure, - on_failure_times: schedule.on_failure_times, - on_failure_exact: schedule.on_failure_exact, - on_failure_extra_args: schedule.on_failure_extra_args, - on_recovery: schedule.on_recovery, - on_recovery_times: schedule.on_recovery_times, - on_recovery_extra_args: schedule.on_recovery_extra_args, - ws_error_handler_muted: schedule.ws_error_handler_muted, - retry: schedule.retry, - summary: schedule.summary, - no_flow_overlap: schedule.no_flow_overlap, - tag: schedule.tag, - }, - ) - .await; - match res { - Ok(tx) => Ok(tx), + let push_next_job_future = async { + let mut tx: QueueTransaction<'_, _> = (rsmq.clone(), db.begin().await?).into(); + tx = push_scheduled_job(db, tx, &schedule).await?; + tx.commit().await?; + Ok::<(), Error>(()) + }; + match push_next_job_future.await { + Ok(_) => Ok(()), Err(err) => { - sqlx::query!( + match sqlx::query!( "UPDATE schedule SET enabled = false, error = $1 WHERE workspace_id = $2 AND path = $3", err.to_string(), &schedule.workspace_id, &schedule.path ) - .execute(db) - .await?; - tracing::warn!("Could not schedule job for {}: {}", schedule_path, err); - Err(err) + .execute(db).await { + Ok(_) => { + match err { + Error::QuotaExceeded(_) => {} + _ => { + report_error_to_workspace_handler_or_critical_side_channel(rsmq, job, db, + format!("Could not schedule next job for {} with err {}. Schedule disabled", schedule.path, err) + ).await; + } + } + Ok(()) + } + Err(disable_err) => match err { + Error::QuotaExceeded(_) => Err(err), + _ => { + report_error_to_workspace_handler_or_critical_side_channel(rsmq, job, db, + format!("Could not schedule next job for {} and could not disable schedule with err {}. Will retry", schedule.path, disable_err) + ).await; + Err(to_anyhow(disable_err).into()) + } + }, + } } } } else { if script_path != schedule.script_path { tracing::warn!( - "Schedule {schedule_path} in {w_id} has a different script path than the job. Not scheduling again" + "Schedule {} in {w_id} has a different script path than the job. Not scheduling again", schedule.path ); } else { - tracing::info!("Schedule {schedule_path} in {w_id} is disabled. Not scheduling again."); + tracing::info!( + "Schedule {} in {w_id} is disabled. Not scheduling again.", + schedule.path + ); } - Ok(tx) + Ok(()) } } @@ -1150,9 +1196,9 @@ async fn apply_schedule_handlers< T: Serialize + Send + Sync, R: rsmq_async::RsmqConnection + Clone + Send + 'c, >( - mut tx: QueueTransaction<'c, R>, + rsmq: Option, db: &Pool, - schedule_path: &str, + schedule: &Schedule, script_path: &str, w_id: &str, success: bool, @@ -1160,19 +1206,7 @@ async fn apply_schedule_handlers< job_id: Uuid, started_at: DateTime, job_priority: Option, -) -> windmill_common::error::Result<(bool, QueueTransaction<'c, R>)> { - let schedule = get_schedule_opt(tx.transaction_mut(), w_id, schedule_path).await?; - - if schedule.is_none() { - tracing::error!( - "Schedule {schedule_path} in {w_id} not found. Impossible to apply schedule handlers" - ); - return Ok((false, tx)); - } - - let schedule = schedule.unwrap(); - let skip_downstream_error_handlers = schedule.ws_error_handler_muted; - +) -> windmill_common::error::Result<()> { if !success { #[cfg(feature = "enterprise")] if let Some(on_failure_path) = schedule.on_failure.clone() { @@ -1184,10 +1218,10 @@ async fn apply_schedule_handlers< "SELECT success, result, started_at FROM completed_job WHERE workspace_id = $1 AND schedule_path = $2 AND script_path = $3 AND id != $4 ORDER BY created_at DESC LIMIT $5", &schedule.workspace_id, &schedule.path, - &schedule.script_path, + script_path, job_id, if exact { times } else { times - 1 } as i64, - ).fetch_all(&mut tx).await?; + ).fetch_all(db).await?; let match_times = if exact { past_jobs.len() == times as usize @@ -1199,82 +1233,63 @@ async fn apply_schedule_handlers< }; if !match_times { - return Ok((skip_downstream_error_handlers, tx)); + return Ok(()); } } - let on_failure_result = handle_on_failure( + push_error_handler( db, - tx, + rsmq, job_id, - schedule_path, - script_path, + Some(schedule.path.to_string()), + Some(script_path.to_string()), schedule.is_flow, w_id, &on_failure_path, result, - times, - started_at, - schedule.on_failure_extra_args, + Some(times), + Some(started_at), + schedule.on_failure_extra_args.clone(), &schedule.email, + true, + false, job_priority, ) - .await; - - match on_failure_result { - Ok((_, ntx)) => { - tx = ntx; - } - Err(err) => { - sqlx::query!( - "UPDATE schedule SET enabled = false, error = $1 WHERE workspace_id = $2 AND path = $3", - format!("Could not trigger error handler: {err}"), - &schedule.workspace_id, - &schedule.path - ) - .execute(db) - .await?; - tracing::warn!( - "Could not trigger error handler for {}: {}", - schedule_path, - err - ); - return Err(err); - } - } + .await?; } } else { #[cfg(feature = "enterprise")] if let Some(on_recovery_path) = schedule.on_recovery.clone() { + let mut tx: QueueTransaction<'_, R> = (rsmq.clone(), db.begin().await?).into(); let times = schedule.on_recovery_times.unwrap_or(1).max(1); let past_jobs = sqlx::query_as!( CompletedJobSubset, "SELECT success, result, started_at FROM completed_job WHERE workspace_id = $1 AND schedule_path = $2 AND script_path = $3 AND id != $4 ORDER BY created_at DESC LIMIT $5", &schedule.workspace_id, &schedule.path, - &schedule.script_path, + script_path, job_id, times as i64, - ).fetch_all(&mut tx).await?; + ).fetch_all(db).await?; if past_jobs.len() < times as usize { - return Ok((skip_downstream_error_handlers, tx)); + return Ok(()); } let n_times_successful = past_jobs[..(times - 1) as usize].iter().all(|j| j.success); if !n_times_successful { - return Ok((skip_downstream_error_handlers, tx)); + return Ok(()); } let failed_job = past_jobs[past_jobs.len() - 1].clone(); if !failed_job.success { - let on_recovery_result = handle_on_recovery( + tx = handle_recovered_schedule( db, tx, job_id, - schedule_path, + &schedule.path, script_path, schedule.is_flow, w_id, @@ -1283,70 +1298,62 @@ async fn apply_schedule_handlers< result, times, started_at, - schedule.on_recovery_extra_args, + schedule.on_recovery_extra_args.clone(), ) - .await; - - match on_recovery_result { - Ok(ntx) => { - tx = ntx; - } - Err(err) => { - sqlx::query!( - "UPDATE schedule SET enabled = false, error = $1 WHERE workspace_id = $2 AND path = $3", - format!("Could not trigger recovery handler: {err}"), - &schedule.workspace_id, - &schedule.path - ) - .execute(db) - .await?; - tracing::warn!( - "Could not trigger recovery handler for {}: {}", - schedule_path, - err - ); - return Err(err); - } - } + .await?; } + + tx.commit().await?; } } - Ok((skip_downstream_error_handlers, tx)) + Ok(()) } -pub async fn handle_on_failure< +pub async fn push_error_handler< 'a, 'c, T: Serialize + Send + Sync, R: rsmq_async::RsmqConnection + Clone + Send + 'c, >( db: &Pool, - tx: QueueTransaction<'c, R>, + rsmq: Option, job_id: Uuid, - schedule_path: &str, - script_path: &str, + schedule_path: Option, + script_path: Option, is_flow: bool, w_id: &str, on_failure_path: &str, result: Json<&'a T>, - failed_times: i32, - started_at: DateTime, + failed_times: Option, + started_at: Option>, extra_args: Option, email: &str, + is_schedule_error_handler: bool, + is_global_error_handler: bool, priority: Option, -) -> windmill_common::error::Result<(Uuid, QueueTransaction<'c, R>)> { - let (payload, tag) = get_payload_tag_from_prefixed_path(on_failure_path, db, w_id).await?; +) -> windmill_common::error::Result { + let handler_w_id = if is_global_error_handler { + "admins" + } else { + w_id + }; + let (payload, tag) = + get_payload_tag_from_prefixed_path(on_failure_path, db, handler_w_id).await?; let mut extra = HashMap::new(); - extra.insert("schedule_path".to_string(), to_raw_value(&schedule_path)); + if let Some(schedule_path) = schedule_path { + extra.insert("schedule_path".to_string(), to_raw_value(&schedule_path)); + } extra.insert("workspace_id".to_string(), to_raw_value(&w_id)); extra.insert("job_id".to_string(), to_raw_value(&job_id)); extra.insert("path".to_string(), to_raw_value(&script_path)); extra.insert("is_flow".to_string(), to_raw_value(&is_flow)); extra.insert("started_at".to_string(), to_raw_value(&started_at)); extra.insert("email".to_string(), to_raw_value(&email)); - extra.insert("failed_times".to_string(), to_raw_value(&failed_times)); + if let Some(failed_times) = failed_times { + extra.insert("failed_times".to_string(), to_raw_value(&failed_times)); + } if let Some(args_v) = extra_args { if let serde_json::Value::Object(args_m) = args_v { @@ -1372,16 +1379,32 @@ pub async fn handle_on_failure< let result = sanitize_result(result); - let tx = PushIsolationLevel::Transaction(tx); + let tx = PushIsolationLevel::IsolatedRoot(db.clone(), rsmq); let (uuid, tx) = push( &db, tx, - w_id, + handler_w_id, payload, PushArgs { extra, args: Json(&result) }, - ERROR_HANDLER_USERNAME, - ERROR_HANDLER_USER_EMAIL, - ERROR_HANDLER_USER_GROUP.to_string(), + if is_global_error_handler { + "global" + } else if is_schedule_error_handler { + SCHEDULE_ERROR_HANDLER_USERNAME + } else { + ERROR_HANDLER_USERNAME + }, + if is_global_error_handler { + SUPERADMIN_SECRET_EMAIL + } else if is_schedule_error_handler { + SCHEDULE_ERROR_HANDLER_USER_EMAIL + } else { + ERROR_HANDLER_USER_EMAIL + }, + if is_global_error_handler { + SUPERADMIN_SECRET_EMAIL.to_string() + } else { + ERROR_HANDLER_USER_GROUP.to_string() + }, None, None, Some(job_id), @@ -1397,12 +1420,8 @@ pub async fn handle_on_failure< priority, ) .await?; - tracing::info!( - "Pushed on_failure job {} for {} to queue", - uuid, - schedule_path - ); - return Ok((uuid, tx)); + tx.commit().await?; + return Ok(uuid); } fn sanitize_result(result: Json<&T>) -> serde_json::Value { @@ -1425,7 +1444,7 @@ fn sanitize_result(result: Json<&T>) -> serde_json:: // is_flow: boolean, // extra_args: serde_json::Value // } -async fn handle_on_recovery< +async fn handle_recovered_schedule< 'a, 'c, T: Serialize + Send + Sync, @@ -1492,8 +1511,8 @@ async fn handle_on_recovery< w_id, payload, args, - ERROR_HANDLER_USERNAME, - ERROR_HANDLER_USER_EMAIL, + SCHEDULE_RECOVERY_HANDLER_USERNAME, + SCHEDULE_RECOVERY_HANDLER_USER_EMAIL, ERROR_HANDLER_USER_GROUP.to_string(), None, None, @@ -2683,6 +2702,8 @@ pub async fn push<'c, T: Serialize + Send + Sync, R: rsmq_async::RsmqConnection if !is_super_admin { if email != ERROR_HANDLER_USER_EMAIL + && email != SCHEDULE_ERROR_HANDLER_USER_EMAIL + && email != SCHEDULE_RECOVERY_HANDLER_USER_EMAIL && email != "worker@windmill.dev" && email != SUPERADMIN_SECRET_EMAIL && email != SUPERADMIN_SYNC_EMAIL @@ -2709,7 +2730,7 @@ pub async fn push<'c, T: Serialize + Send + Sync, R: rsmq_async::RsmqConnection && !matches!(job_payload, JobPayload::FlowDependencies { .. }) && !matches!(job_payload, JobPayload::AppDependencies { .. }) { - return Err(error::Error::BadRequest(format!( + return Err(error::Error::QuotaExceeded(format!( "User {email} has exceeded the free usage limit of {MAX_FREE_EXECS} that applies outside of premium workspaces." ))); } @@ -2721,7 +2742,7 @@ pub async fn push<'c, T: Serialize + Send + Sync, R: rsmq_async::RsmqConnection .unwrap_or(0); if in_queue > MAX_FREE_EXECS.into() { - return Err(error::Error::BadRequest(format!( + return Err(error::Error::QuotaExceeded(format!( "User {email} has exceeded the jobs in queue limit of {MAX_FREE_EXECS} that applies outside of premium workspaces." ))); } @@ -2735,7 +2756,7 @@ pub async fn push<'c, T: Serialize + Send + Sync, R: rsmq_async::RsmqConnection .unwrap_or(0); if concurrent_runs > MAX_FREE_CONCURRENT_RUNS.into() { - return Err(error::Error::BadRequest(format!( + return Err(error::Error::QuotaExceeded(format!( "User {email} has exceeded the concurrent runs limit of {MAX_FREE_CONCURRENT_RUNS} that applies outside of premium workspaces." ))); } @@ -2763,7 +2784,7 @@ pub async fn push<'c, T: Serialize + Send + Sync, R: rsmq_async::RsmqConnection && !matches!(job_payload, JobPayload::FlowDependencies { .. }) && !matches!(job_payload, JobPayload::AppDependencies { .. }) { - return Err(error::Error::BadRequest(format!( + return Err(error::Error::QuotaExceeded(format!( "Workspace {workspace_id} has exceeded the free usage limit of {MAX_FREE_EXECS} that applies outside of premium workspaces." ))); } @@ -2777,7 +2798,7 @@ pub async fn push<'c, T: Serialize + Send + Sync, R: rsmq_async::RsmqConnection .unwrap_or(0); if in_queue_workspace > MAX_FREE_EXECS.into() { - return Err(error::Error::BadRequest(format!( + return Err(error::Error::QuotaExceeded(format!( "Workspace {workspace_id} has exceeded the jobs in queue limit of {MAX_FREE_EXECS} that applies outside of premium workspaces." ))); } @@ -2791,7 +2812,7 @@ pub async fn push<'c, T: Serialize + Send + Sync, R: rsmq_async::RsmqConnection .unwrap_or(0); if concurrent_runs_workspace > MAX_FREE_CONCURRENT_RUNS.into() { - return Err(error::Error::BadRequest(format!( + return Err(error::Error::QuotaExceeded(format!( "Workspace {workspace_id} has exceeded the concurrent runs limit of {MAX_FREE_CONCURRENT_RUNS} that applies outside of premium workspaces." ))); } diff --git a/backend/windmill-queue/src/schedule.rs b/backend/windmill-queue/src/schedule.rs index c9970b0644..b582b0f7fa 100644 --- a/backend/windmill-queue/src/schedule.rs +++ b/backend/windmill-queue/src/schedule.rs @@ -26,9 +26,9 @@ use windmill_common::{ pub async fn push_scheduled_job<'c, R: rsmq_async::RsmqConnection + Send + 'c>( db: &DB, mut tx: QueueTransaction<'c, R>, - schedule: Schedule, + schedule: &Schedule, ) -> Result> { - let sched = cron::Schedule::from_str(&schedule.schedule) + let sched = cron::Schedule::from_str(schedule.schedule.as_ref()) .map_err(|e| error::Error::BadRequest(e.to_string()))?; let tz = chrono_tz::Tz::from_str(&schedule.timezone) @@ -68,9 +68,9 @@ pub async fn push_scheduled_job<'c, R: rsmq_async::RsmqConnection + Send + 'c>( let mut args: serde_json::Map = serde_json::Map::new(); - if let Some(args_v) = schedule.args { + if let Some(args_v) = &schedule.args { if let serde_json::Value::Object(args_m) = args_v { - args = args_m + args = args_m.clone() } else { return Err(error::Error::ExecutionErr( "args of scripts needs to be dict".to_string(), @@ -90,7 +90,7 @@ pub async fn push_scheduled_job<'c, R: rsmq_async::RsmqConnection + Send + 'c>( .map(|x| (x.tag, x.dedicated_worker)) .unwrap_or_else(|| (None, None)); ( - JobPayload::Flow { path: schedule.script_path, dedicated_worker }, + JobPayload::Flow { path: schedule.script_path.clone(), dedicated_worker }, tag, None, ) @@ -113,8 +113,8 @@ pub async fn push_scheduled_job<'c, R: rsmq_async::RsmqConnection + Send + 'c>( .await?; if schedule.retry.is_some() { - let parsed_retry = - serde_json::from_value::(schedule.retry.unwrap()).map_err(|err| { + let parsed_retry = serde_json::from_value::(schedule.retry.clone().unwrap()) + .map_err(|err| { error::Error::InternalErr(format!( "Unable to parse retry information from schedule: {}", err.to_string(), @@ -127,7 +127,7 @@ pub async fn push_scheduled_job<'c, R: rsmq_async::RsmqConnection + Send + 'c>( // if retry is set, we wrap the script into a one step flow with a retry on the module ( JobPayload::SingleScriptFlow { - path: schedule.script_path, + path: schedule.script_path.clone(), hash: hash, retry: parsed_retry, args: static_args, @@ -144,7 +144,7 @@ pub async fn push_scheduled_job<'c, R: rsmq_async::RsmqConnection + Send + 'c>( ( JobPayload::ScriptHash { hash, - path: schedule.script_path, + path: schedule.script_path.clone(), concurrent_limit: concurrent_limit, concurrency_time_window_s: concurrency_time_window_s, cache_ttl: cache_ttl, @@ -153,7 +153,7 @@ pub async fn push_scheduled_job<'c, R: rsmq_async::RsmqConnection + Send + 'c>( priority, }, if schedule.tag.as_ref().is_some_and(|x| x != "") { - schedule.tag + schedule.tag.clone() } else { tag }, diff --git a/backend/windmill-worker/src/worker_flow.rs b/backend/windmill-worker/src/worker_flow.rs index bcbed40a39..a0b2789774 100644 --- a/backend/windmill-worker/src/worker_flow.rs +++ b/backend/windmill-worker/src/worker_flow.rs @@ -42,6 +42,7 @@ use windmill_common::{ }, flows::{FlowModule, FlowModuleValue, FlowValue, InputTransform, Retry, Suspend}, }; +use windmill_queue::schedule::get_schedule_opt; use windmill_queue::{ add_completed_job, add_completed_job_error, append_logs, get_queued_job, handle_maybe_scheduled_job, CanceledBy, PushIsolationLevel, WrappedError, @@ -1201,23 +1202,37 @@ pub async fn handle_flow( && flow_job.script_path.is_some() && status.step == 0 { - let tx: QueueTransaction<'_, R> = (rsmq.clone(), db.begin().await?).into(); + let mut tx: QueueTransaction<'_, R> = (rsmq.clone(), db.begin().await?).into(); - match handle_maybe_scheduled_job( - tx, - db, - flow_job.schedule_path.as_ref().unwrap(), - flow_job.script_path.as_ref().unwrap(), - &flow_job.workspace_id, - ) - .await - { - Ok(tx) => { - tx.commit().await?; - } - Err(e) => { - tracing::error!("Error during handle_maybe_scheduled_job: {e}"); - } + let schedule_path = flow_job.schedule_path.as_ref().unwrap(); + + let schedule = + get_schedule_opt(tx.transaction_mut(), &flow_job.workspace_id, schedule_path).await?; + + tx.commit().await?; + + if let Some(schedule) = schedule { + if let Err(err) = handle_maybe_scheduled_job( + rsmq.clone(), + db, + flow_job, + &schedule, + flow_job.script_path.as_ref().unwrap(), + &flow_job.workspace_id, + ) + .await + { + match err { + Error::QuotaExceeded(_) => return Err(err.into()), + // scheduling next job failed and could not disable schedule => make zombie job to retry + _ => return Ok(()), + } + }; + } else { + tracing::error!( + "Schedule {schedule_path} in {} not found. Impossible to schedule again", + &flow_job.workspace_id + ); } } diff --git a/frontend/src/lib/components/InstanceSettings.svelte b/frontend/src/lib/components/InstanceSettings.svelte index f1df1e0926..dfda5f4b04 100644 --- a/frontend/src/lib/components/InstanceSettings.svelte +++ b/frontend/src/lib/components/InstanceSettings.svelte @@ -16,7 +16,7 @@ import { capitalize } from '$lib/utils' import { enterpriseLicense } from '$lib/stores' import CustomOauth from './CustomOauth.svelte' - import { AlertTriangle, Plus } from 'lucide-svelte' + import { AlertTriangle, Plus, X } from 'lucide-svelte' import CustomSso from './CustomSso.svelte' import AuthentikSetting from '$lib/components/AuthentikSetting.svelte' import AutheliaSetting from '$lib/components/AutheliaSetting.svelte' @@ -24,6 +24,7 @@ import ZitadelSetting from '$lib/components/ZitadelSetting.svelte' import Password from './Password.svelte' import ObjectStoreConfigSettings from './ObjectStoreConfigSettings.svelte' + import { fade } from 'svelte/transition' export let tab: string = 'Core' export let hideTabs: boolean = false @@ -498,6 +499,78 @@ placeholder={setting.placeholder} bind:value={values[setting.key]} /> + {:else if setting.fieldType == 'critical_error_channels'} +
+
+ + + +
+ {#if $enterpriseLicense && Array.isArray(values[setting.key])} + {#each values[setting.key] ?? [] as v, i} +
+ + { + if (e.target?.['value']) { + values[setting.key][i] = { + email: e.target['value'] + } + } + }} + value={v?.email ?? ''} + /> + +
+ {/each} + {/if} +
+
+ +
{:else if setting.fieldType == 'object_store_config'} {:else if setting.fieldType == 'number'} diff --git a/frontend/src/lib/components/instanceSettings.ts b/frontend/src/lib/components/instanceSettings.ts index c905c32961..1dd5cdd007 100644 --- a/frontend/src/lib/components/instanceSettings.ts +++ b/frontend/src/lib/components/instanceSettings.ts @@ -17,6 +17,7 @@ export interface Setting { | 'email' | 'license_key' | 'object_store_config' + | 'critical_error_channels' storage: SettingStorage isValid?: (value: any) => boolean error?: string @@ -102,6 +103,15 @@ export const settings: Record = { storage: 'setting', ee_only: '' }, + { + label: 'Critical Error Channels', + description: + 'Channels to send critical errors to. SMTP must be configured for the email channel.', + key: 'critical_error_channels', + fieldType: 'critical_error_channels', + storage: 'setting', + ee_only: 'Channels other than tracing are only available in the EE version' + }, { label: 'Azure OpenAI base path', description: