From e74e04b247ac97080c7f2678982b70976329e49a Mon Sep 17 00:00:00 2001 From: dieriba Date: Sat, 15 Nov 2025 00:48:45 +0100 Subject: [PATCH] better --- ...200000_add_queue_mode_to_triggers.down.sql | 11 --- ...06200000_add_queue_mode_to_triggers.up.sql | 9 --- backend/windmill-api/openapi.yaml | 6 +- .../src/triggers/global_handler.rs | 67 +++++++++++++++++-- backend/windmill-api/src/triggers/mod.rs | 1 - .../src/triggers/trigger_helpers.rs | 14 +++- 6 files changed, 75 insertions(+), 33 deletions(-) diff --git a/backend/migrations/20251106200000_add_queue_mode_to_triggers.down.sql b/backend/migrations/20251106200000_add_queue_mode_to_triggers.down.sql index 7fc8875c33..0a587265c2 100644 --- a/backend/migrations/20251106200000_add_queue_mode_to_triggers.down.sql +++ b/backend/migrations/20251106200000_add_queue_mode_to_triggers.down.sql @@ -1,15 +1,4 @@ -- Add down migration script here -DROP INDEX IF EXISTS idx_websocket_trigger_active_mode; -DROP INDEX IF EXISTS idx_sqs_trigger_active_mode; -DROP INDEX IF EXISTS idx_postgres_trigger_active_mode; -DROP INDEX IF EXISTS idx_nats_trigger_active_mode; -DROP INDEX IF EXISTS idx_mqtt_trigger_active_mode; -DROP INDEX IF EXISTS idx_kafka_trigger_active_mode; -DROP INDEX IF EXISTS idx_http_trigger_active_mode; -DROP INDEX IF EXISTS idx_gcp_trigger_active_mode; -DROP INDEX IF EXISTS idx_email_trigger_active_mode; - - ALTER TABLE websocket_trigger DROP COLUMN IF EXISTS active_mode; diff --git a/backend/migrations/20251106200000_add_queue_mode_to_triggers.up.sql b/backend/migrations/20251106200000_add_queue_mode_to_triggers.up.sql index 9fe090b2dc..e51716a178 100644 --- a/backend/migrations/20251106200000_add_queue_mode_to_triggers.up.sql +++ b/backend/migrations/20251106200000_add_queue_mode_to_triggers.up.sql @@ -26,12 +26,3 @@ ADD COLUMN active_mode BOOLEAN NOT NULL DEFAULT TRUE; ALTER TABLE email_trigger ADD COLUMN active_mode BOOLEAN NOT NULL DEFAULT TRUE; -CREATE INDEX idx_gcp_trigger_active_mode ON gcp_trigger(workspace_id,path,active_mode); -CREATE INDEX idx_http_trigger_active_mode ON http_trigger(workspace_id,path,active_mode); -CREATE INDEX idx_kafka_trigger_active_mode ON kafka_trigger(workspace_id,path,active_mode); -CREATE INDEX idx_mqtt_trigger_active_mode ON mqtt_trigger(workspace_id,path,active_mode); -CREATE INDEX idx_nats_trigger_active_mode ON nats_trigger(workspace_id,path,active_mode); -CREATE INDEX idx_postgres_trigger_active_mode ON postgres_trigger(workspace_id,path,active_mode); -CREATE INDEX idx_sqs_trigger_active_mode ON sqs_trigger(workspace_id,path,active_mode); -CREATE INDEX idx_websocket_trigger_active_mode ON websocket_trigger(workspace_id,path,active_mode); -CREATE INDEX idx_email_trigger_active_mode ON email_trigger(workspace_id,path,active_mode); \ No newline at end of file diff --git a/backend/windmill-api/openapi.yaml b/backend/windmill-api/openapi.yaml index 8ecde8f160..2e9d8520c2 100644 --- a/backend/windmill-api/openapi.yaml +++ b/backend/windmill-api/openapi.yaml @@ -17542,7 +17542,7 @@ components: active_mode: type: boolean default: true - description: If set to false, each incomming event will be suspend job until ran manually or set it to true + description: If set to false, each incoming event will be suspend job until ran manually or set it to true required: - path @@ -18725,7 +18725,7 @@ components: active_mode: type: boolean default: true - description: If set to false, each incomming event will be suspend job until ran manually or set it to true + description: If set to false, each incoming event will be suspend job until ran manually or set it to true required: - path @@ -18755,7 +18755,7 @@ components: active_mode: type: boolean default: true - description: If set to false, each incomming event will be suspend job until ran manually or set it to true + description: If set to false, each incoming event will be suspend job until ran manually or set it to true required: - path - script_path diff --git a/backend/windmill-api/src/triggers/global_handler.rs b/backend/windmill-api/src/triggers/global_handler.rs index 619e676580..ae76b1d740 100644 --- a/backend/windmill-api/src/triggers/global_handler.rs +++ b/backend/windmill-api/src/triggers/global_handler.rs @@ -1,16 +1,22 @@ -use crate::{db::{ApiAuthed, DB}, triggers::INACTIVE_TRIGGER_SCHEDULED_FOR_DATE}; +use crate::{db::ApiAuthed, triggers::INACTIVE_TRIGGER_SCHEDULED_FOR_DATE}; use axum::{ extract::{Extension, Path}, response::Json, }; -use windmill_common::{error, jobs::JobTriggerKind, utils::require_admin}; +use windmill_audit::{audit_oss::audit_log, ActionKind}; +use windmill_common::{db::UserDB, error, jobs::JobTriggerKind, utils::require_admin}; pub async fn resume_suspended_trigger_jobs( authed: ApiAuthed, - Extension(db): Extension, + Extension(user_db): Extension, Path((w_id, trigger_kind, trigger_path)): Path<(String, JobTriggerKind, String)>, ) -> error::Result> { require_admin(authed.is_admin, &authed.username)?; + + let mut tx = user_db.begin(&authed).await?; + + // Use the date constant to identify suspended jobs + // This date (9999-12-31 23:59:59) is used as a marker for suspended jobs let scheduled_for = INACTIVE_TRIGGER_SCHEDULED_FOR_DATE.clone(); let result = sqlx::query!( r#" @@ -33,11 +39,34 @@ pub async fn resume_suspended_trigger_jobs( trigger_kind as _, trigger_path ) - .execute(&db) + .execute(&mut *tx) .await?; let count = result.rows_affected(); + let trigger_kind_str = format!("{:?}", trigger_kind); + let count_str = count.to_string(); + audit_log( + &mut *tx, + &authed, + "triggers.bulk_resume", + ActionKind::Execute, + &w_id, + Some(&trigger_path), + Some( + [ + ("trigger_kind", trigger_kind_str.as_str()), + ("jobs_count", count_str.as_str()), + ] + .iter() + .cloned() + .collect(), + ), + ) + .await?; + + tx.commit().await?; + let message = format!( "Successfully resumed {} suspended job{} for trigger at path: {}", count, @@ -49,10 +78,13 @@ pub async fn resume_suspended_trigger_jobs( pub async fn cancel_suspended_trigger_jobs( authed: ApiAuthed, - Extension(db): Extension, + Extension(user_db): Extension, Path((w_id, trigger_kind, trigger_path)): Path<(String, JobTriggerKind, String)>, ) -> error::Result> { require_admin(authed.is_admin, &authed.username)?; + + let mut tx = user_db.begin(&authed).await?; + let scheduled_for = INACTIVE_TRIGGER_SCHEDULED_FOR_DATE.clone(); let result = sqlx::query!( @@ -79,11 +111,34 @@ pub async fn cancel_suspended_trigger_jobs( trigger_kind as _, trigger_path ) - .execute(&db) + .execute(&mut *tx) .await?; let count = result.rows_affected(); + let trigger_kind_str = format!("{:?}", trigger_kind); + let count_str = count.to_string(); + audit_log( + &mut *tx, + &authed, + "triggers.bulk_cancel", + ActionKind::Delete, + &w_id, + Some(&trigger_path), + Some( + [ + ("trigger_kind", trigger_kind_str.as_str()), + ("jobs_count", count_str.as_str()), + ] + .iter() + .cloned() + .collect(), + ), + ) + .await?; + + tx.commit().await?; + let message = format!( "Successfully cancelled {} suspended job{} for trigger at path: {}", count, diff --git a/backend/windmill-api/src/triggers/mod.rs b/backend/windmill-api/src/triggers/mod.rs index 2bd8237962..407ad1d8d8 100644 --- a/backend/windmill-api/src/triggers/mod.rs +++ b/backend/windmill-api/src/triggers/mod.rs @@ -126,7 +126,6 @@ pub struct BaseTriggerData { pub is_flow: bool, pub enabled: Option, pub active_mode: Option, - pub process_queued_jobs: Option, } #[derive(Debug, Clone, Serialize, Deserialize)] diff --git a/backend/windmill-api/src/triggers/trigger_helpers.rs b/backend/windmill-api/src/triggers/trigger_helpers.rs index 874a687e7a..dd997e06a9 100644 --- a/backend/windmill-api/src/triggers/trigger_helpers.rs +++ b/backend/windmill-api/src/triggers/trigger_helpers.rs @@ -25,6 +25,12 @@ use windmill_common::{ }; use windmill_queue::{push, PushArgs, PushArgsOwned, PushIsolationLevel}; +/// Helper function to check if triggers should use queue mode. +/// Queue mode suspends jobs by scheduling them for a far future date. +fn is_queue_mode(active_mode: Option) -> bool { + matches!(active_mode, Some(false)) +} + #[cfg(feature = "enterprise")] use crate::jobs::check_license_key_valid; use crate::{ @@ -532,9 +538,11 @@ pub async fn trigger_runnable_inner( }); let user_db = user_db.unwrap_or_else(|| UserDB::new(db.clone())); - let scheduled_for = active_mode - .filter(|active_mode| !*active_mode) - .map(|_| INACTIVE_TRIGGER_SCHEDULED_FOR_DATE.clone()); + let scheduled_for = if is_queue_mode(active_mode) { + Some(INACTIVE_TRIGGER_SCHEDULED_FOR_DATE.clone()) + } else { + None + }; let (uuid, delete_after_use, early_return) = if is_flow { let run_query = RunJobQuery { job_id, scheduled_for, ..Default::default() }; let path = StripPath(runnable_path.to_string());