From a4f3b4fc9bcdd562f4c706eb901e16367729ff42 Mon Sep 17 00:00:00 2001 From: dieriba Date: Fri, 14 Nov 2025 21:06:49 +0100 Subject: [PATCH] big --- ...200000_add_queue_mode_to_triggers.down.sql | 32 +- ...06200000_add_queue_mode_to_triggers.up.sql | 36 +- backend/src/monitor.rs | 2 +- backend/tests/common/mod.rs | 1 - backend/windmill-api/openapi.yaml | 55 ++- backend/windmill-api/src/apps.rs | 9 +- .../windmill-api/src/concurrency_groups.rs | 1 + backend/windmill-api/src/flows.rs | 6 +- backend/windmill-api/src/jobs.rs | 206 +++++------ backend/windmill-api/src/scripts.rs | 3 +- backend/windmill-api/src/settings.rs | 2 +- .../src/triggers/email/handler_oss.rs | 2 - backend/windmill-api/src/triggers/handler.rs | 44 +-- .../windmill-api/src/triggers/http/handler.rs | 54 ++- backend/windmill-api/src/triggers/http/mod.rs | 4 +- backend/windmill-api/src/triggers/listener.rs | 12 +- backend/windmill-api/src/triggers/mod.rs | 18 +- .../windmill-api/src/triggers/mqtt/handler.rs | 10 +- .../src/triggers/postgres/handler.rs | 10 +- .../src/triggers/trigger_helpers.rs | 350 +++--------------- .../src/triggers/websocket/handler.rs | 10 +- .../src/triggers/websocket/listener.rs | 16 +- .../src/triggers/websocket/mod.rs | 5 +- backend/windmill-common/src/triggers.rs | 13 + backend/windmill-queue/src/jobs.rs | 34 +- backend/windmill-queue/src/schedule.rs | 8 +- backend/windmill-worker/src/ai/tools.rs | 1 - backend/windmill-worker/src/worker_flow.rs | 1 - .../windmill-worker/src/worker_lockfiles.rs | 1 - .../triggers/TriggerStateToggle.svelte | 33 +- .../email/EmailTriggerEditorInner.svelte | 6 - .../triggers/gcp/GcpTriggerEditorInner.svelte | 6 +- .../triggers/http/RouteEditorInner.svelte | 6 +- .../kafka/KafkaTriggerEditorInner.svelte | 6 +- .../mqtt/MqttTriggerEditorInner.svelte | 6 +- .../nats/NatsTriggerEditorInner.svelte | 6 +- .../PostgresTriggerEditorInner.svelte | 6 +- .../triggers/sqs/SqsTriggerEditorInner.svelte | 6 +- .../WebsocketTriggerEditorInner.svelte | 6 +- 39 files changed, 371 insertions(+), 662 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 7f5272169a..eb44ce17ef 100644 --- a/backend/migrations/20251106200000_add_queue_mode_to_triggers.down.sql +++ b/backend/migrations/20251106200000_add_queue_mode_to_triggers.down.sql @@ -1,33 +1,33 @@ -- Add down migration script here -DROP INDEX IF EXISTS idx_websocket_trigger_suspend_number; -DROP INDEX IF EXISTS idx_sqs_trigger_suspend_number; -DROP INDEX IF EXISTS idx_postgres_trigger_suspend_number; -DROP INDEX IF EXISTS idx_nats_trigger_suspend_number; -DROP INDEX IF EXISTS idx_mqtt_trigger_suspend_number; -DROP INDEX IF EXISTS idx_kafka_trigger_suspend_number; -DROP INDEX IF EXISTS idx_http_trigger_suspend_number; -DROP INDEX IF EXISTS idx_gcp_trigger_suspend_number; +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; ALTER TABLE websocket_trigger -DROP COLUMN IF EXISTS suspend_number; +DROP COLUMN IF EXISTS active_mode; ALTER TABLE sqs_trigger -DROP COLUMN IF EXISTS suspend_number; +DROP COLUMN IF EXISTS active_mode; ALTER TABLE postgres_trigger -DROP COLUMN IF EXISTS suspend_number; +DROP COLUMN IF EXISTS active_mode; ALTER TABLE nats_trigger -DROP COLUMN IF EXISTS suspend_number; +DROP COLUMN IF EXISTS active_mode; ALTER TABLE mqtt_trigger -DROP COLUMN IF EXISTS suspend_number; +DROP COLUMN IF EXISTS active_mode; ALTER TABLE kafka_trigger -DROP COLUMN IF EXISTS suspend_number; +DROP COLUMN IF EXISTS active_mode; ALTER TABLE http_trigger -DROP COLUMN IF EXISTS suspend_number; +DROP COLUMN IF EXISTS active_mode; ALTER TABLE gcp_trigger -DROP COLUMN IF EXISTS suspend_number; \ No newline at end of file +DROP COLUMN IF EXISTS active_mode; \ No newline at end of file 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 99678f0c5a..9fe090b2dc 100644 --- a/backend/migrations/20251106200000_add_queue_mode_to_triggers.up.sql +++ b/backend/migrations/20251106200000_add_queue_mode_to_triggers.up.sql @@ -1,37 +1,37 @@ -- Add up migration script here ALTER TABLE gcp_trigger -ADD COLUMN suspend_number INTEGER NULL; +ADD COLUMN active_mode BOOLEAN NOT NULL DEFAULT TRUE; ALTER TABLE http_trigger -ADD COLUMN suspend_number INTEGER NULL; +ADD COLUMN active_mode BOOLEAN NOT NULL DEFAULT TRUE; ALTER TABLE kafka_trigger -ADD COLUMN suspend_number INTEGER NULL; +ADD COLUMN active_mode BOOLEAN NOT NULL DEFAULT TRUE; ALTER TABLE mqtt_trigger -ADD COLUMN suspend_number INTEGER NULL; +ADD COLUMN active_mode BOOLEAN NOT NULL DEFAULT TRUE; ALTER TABLE nats_trigger -ADD COLUMN suspend_number INTEGER NULL; +ADD COLUMN active_mode BOOLEAN NOT NULL DEFAULT TRUE; ALTER TABLE postgres_trigger -ADD COLUMN suspend_number INTEGER NULL; +ADD COLUMN active_mode BOOLEAN NOT NULL DEFAULT TRUE; ALTER TABLE sqs_trigger -ADD COLUMN suspend_number INTEGER NULL; +ADD COLUMN active_mode BOOLEAN NOT NULL DEFAULT TRUE; ALTER TABLE websocket_trigger -ADD COLUMN suspend_number INTEGER NULL; +ADD COLUMN active_mode BOOLEAN NOT NULL DEFAULT TRUE; ALTER TABLE email_trigger -ADD COLUMN suspend_number INTEGER NULL; +ADD COLUMN active_mode BOOLEAN NOT NULL DEFAULT TRUE; -CREATE INDEX idx_gcp_trigger_suspend_number ON gcp_trigger(workspace_id,path,suspend_number) WHERE suspend_number IS NOT NULL; -CREATE INDEX idx_http_trigger_suspend_number ON http_trigger(workspace_id,path,suspend_number) WHERE suspend_number IS NOT NULL; -CREATE INDEX idx_kafka_trigger_suspend_number ON kafka_trigger(workspace_id,path,suspend_number) WHERE suspend_number IS NOT NULL; -CREATE INDEX idx_mqtt_trigger_suspend_number ON mqtt_trigger(workspace_id,path,suspend_number) WHERE suspend_number IS NOT NULL; -CREATE INDEX idx_nats_trigger_suspend_number ON nats_trigger(workspace_id,path,suspend_number) WHERE suspend_number IS NOT NULL; -CREATE INDEX idx_postgres_trigger_suspend_number ON postgres_trigger(workspace_id,path,suspend_number) WHERE suspend_number IS NOT NULL; -CREATE INDEX idx_sqs_trigger_suspend_number ON sqs_trigger(workspace_id,path,suspend_number) WHERE suspend_number IS NOT NULL; -CREATE INDEX idx_websocket_trigger_suspend_number ON websocket_trigger(workspace_id,path,suspend_number) WHERE suspend_number IS NOT NULL; -CREATE INDEX idx_email_trigger_suspend_number ON email_trigger(workspace_id,path,suspend_number) WHERE suspend_number IS NOT NULL; \ No newline at end of file +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/src/monitor.rs b/backend/src/monitor.rs index b0818c96cd..978aba2faa 100644 --- a/backend/src/monitor.rs +++ b/backend/src/monitor.rs @@ -1271,7 +1271,7 @@ pub async fn reload_license_key(conn: &Connection) -> anyhow::Result<()> { tracing::error!("Could not parse LICENSE_KEY found: {:#?}", &q); } }; - set_license_key(value).await; + set_license_key(value, conn.as_sql()).await; Ok(()) } diff --git a/backend/tests/common/mod.rs b/backend/tests/common/mod.rs index c06a7470d5..cfdfe4124f 100644 --- a/backend/tests/common/mod.rs +++ b/backend/tests/common/mod.rs @@ -208,7 +208,6 @@ impl RunJob { false, None, debounce_job_id_o, - None, None ) .await diff --git a/backend/windmill-api/openapi.yaml b/backend/windmill-api/openapi.yaml index 37fc987f4d..a58ca3999e 100644 --- a/backend/windmill-api/openapi.yaml +++ b/backend/windmill-api/openapi.yaml @@ -8088,6 +8088,8 @@ paths: - $ref: "#/components/parameters/ScriptExactPath" - $ref: "#/components/parameters/ScriptStartPath" - $ref: "#/components/parameters/SchedulePath" + - $ref: "#/components/parameters/TriggerPath" + - $ref: "#/components/parameters/JobTriggerKind" - $ref: "#/components/parameters/ScriptExactHash" - $ref: "#/components/parameters/StartedBefore" - $ref: "#/components/parameters/StartedAfter" @@ -8374,13 +8376,7 @@ paths: content: application/json: schema: - type: object - properties: - suspend_number: - type: integer - description: the suspend number of jobs to resume - required: - - suspend_number + $ref: "#/components/schemas/SuspendedJobRequest" responses: "200": description: confirmation message @@ -8403,13 +8399,7 @@ paths: content: application/json: schema: - type: object - properties: - suspend_number: - type: integer - description: the suspend number of jobs to cancel - required: - - suspend_number + $ref: "#/components/schemas/SuspendedJobRequest" responses: "200": description: confirmation message @@ -15039,6 +15029,12 @@ components: in: query schema: type: string + TriggerPath: + name: trigger_path + description: mask to filter by trigger path + in: query + schema: + type: string ScriptExactPath: name: script_path_exact description: mask to filter exact matching path @@ -17038,6 +17034,18 @@ components: - timezone - args + SuspendedJobRequest: + type: object + properties: + trigger_path: + type: string + description: The path of the trigger + trigger_kind: + $ref: "#/components/parameters/JobTriggerKind" + required: + - runnable_path + - trigger_kind + TriggerExtraProperty: type: object properties: @@ -17060,8 +17068,8 @@ components: format: date-time is_flow: type: boolean - suspend_number: - type: number + active_mode: + type: boolean required: - path - script_path @@ -17071,6 +17079,7 @@ components: - edited_by - edited_at - is_flow + - active_mode AuthenticationMethod: type: string @@ -17307,7 +17316,7 @@ components: active_mode: type: boolean default: true - description: If true, queue jobs with suspend functionality instead of immediate execution + description: If set to false, each incomming event will be suspend job until ran manually or set it to true required: - path @@ -17372,7 +17381,6 @@ components: active_mode: type: boolean default: true - description: If true, queue jobs with suspend functionality instead of immediate execution required: - path - script_path @@ -17509,7 +17517,6 @@ components: active_mode: type: boolean default: true - description: If true, queue jobs with suspend functionality instead of immediate execution required: - path @@ -17561,7 +17568,6 @@ components: active_mode: type: boolean default: true - description: If true, queue jobs with suspend functionality instead of immediate execution required: - path @@ -17706,7 +17712,6 @@ components: active_mode: type: boolean default: true - description: If true, queue jobs with suspend functionality instead of immediate execution required: - path - script_path @@ -17748,7 +17753,6 @@ components: active_mode: type: boolean default: true - description: If true, queue jobs with suspend functionality instead of immediate execution required: - path - script_path @@ -18040,7 +18044,6 @@ components: active_mode: type: boolean default: true - description: If true, queue jobs with suspend functionality instead of immediate execution required: - queue_url - aws_resource_path @@ -18186,7 +18189,6 @@ components: active_mode: type: boolean default: true - description: If true, queue jobs with suspend functionality instead of immediate execution required: - path - script_path @@ -18222,7 +18224,6 @@ components: active_mode: type: boolean default: true - description: If true, queue jobs with suspend functionality instead of immediate execution required: - path - script_path @@ -18295,7 +18296,6 @@ components: active_mode: type: boolean default: true - description: If true, queue jobs with suspend functionality instead of immediate execution required: - path @@ -18329,7 +18329,6 @@ components: active_mode: type: boolean default: true - description: If true, queue jobs with suspend functionality instead of immediate execution retry: $ref: "../../openflow.openapi.yaml#/components/schemas/Retry" @@ -18412,7 +18411,6 @@ components: active_mode: type: boolean default: true - description: If true, queue jobs with suspend functionality instead of immediate execution required: - path @@ -18452,7 +18450,6 @@ components: active_mode: type: boolean default: true - description: If true, queue jobs with suspend functionality instead of immediate execution required: - path - script_path diff --git a/backend/windmill-api/src/apps.rs b/backend/windmill-api/src/apps.rs index 96858e3e3c..a8fe8ce82b 100644 --- a/backend/windmill-api/src/apps.rs +++ b/backend/windmill-api/src/apps.rs @@ -1241,8 +1241,7 @@ async fn create_app_internal<'a>( false, None, None, - None, - None, + None ) .await?; tracing::info!("Pushed app dependency job {}", dependency_job_uuid); @@ -1632,8 +1631,7 @@ async fn update_app_internal<'a>( false, None, None, - None, - None, + None ) .await?; tracing::info!("Pushed app dependency job {}", dependency_job_uuid); @@ -1953,8 +1951,7 @@ async fn execute_component( false, end_user_email, None, - None, - None, + None ) .await?; tx.commit().await?; diff --git a/backend/windmill-api/src/concurrency_groups.rs b/backend/windmill-api/src/concurrency_groups.rs index 4a96a16e0b..e9c06de4f1 100644 --- a/backend/windmill-api/src/concurrency_groups.rs +++ b/backend/windmill-api/src/concurrency_groups.rs @@ -203,6 +203,7 @@ async fn get_concurrent_intervals( has_null_parent: None, worker: None, label: None, + trigger_path: None, scheduled_for_before_now: _, is_not_schedule: _, started_before: _, diff --git a/backend/windmill-api/src/flows.rs b/backend/windmill-api/src/flows.rs index fad5d6bb99..ebc09952b5 100644 --- a/backend/windmill-api/src/flows.rs +++ b/backend/windmill-api/src/flows.rs @@ -562,8 +562,7 @@ async fn create_flow( false, None, None, - None, - None, + None ) .await?; @@ -1027,8 +1026,7 @@ async fn update_flow( false, None, None, - None, - None, + None ) .await?; diff --git a/backend/windmill-api/src/jobs.rs b/backend/windmill-api/src/jobs.rs index 24789f5f00..662b1b2349 100644 --- a/backend/windmill-api/src/jobs.rs +++ b/backend/windmill-api/src/jobs.rs @@ -38,6 +38,7 @@ use windmill_common::jobs::{ DynamicInput, JobTriggerKind, ENTRYPOINT_OVERRIDE, }; use windmill_common::s3_helpers::{upload_artifact_to_store, BundleFormat}; +use windmill_common::triggers::TriggerInfo; use windmill_common::utils::{RunnableKind, WarnAfterExt}; use windmill_common::worker::{Connection, CLOUD_HOSTED, TMP_DIR}; use windmill_common::DYNAMIC_INPUT_CACHE; @@ -47,6 +48,7 @@ use windmill_common::{email_oss::send_email_html, server::load_smtp_config}; use windmill_common::variables::get_workspace_key; use crate::triggers::trigger_helpers::ScriptId; +use crate::triggers::INACTIVE_TRIGGER_SCHEDULED_FOR_DATE; use crate::{ add_webhook_allowed_origin, args::{self, RawWebhookArgs}, @@ -1819,6 +1821,7 @@ pub struct ListQueueQuery { pub created_or_started_after: Option>, pub running: Option, pub schedule_path: Option, + pub trigger_path: Option, pub parent_job: Option, pub order_desc: Option, pub job_kinds: Option, @@ -1840,6 +1843,7 @@ pub struct ListQueueQuery { impl From for ListQueueQuery { fn from(lcq: ListCompletedQuery) -> Self { Self { + trigger_path: lcq.trigger_path, script_path_start: lcq.script_path_start, script_path_exact: lcq.script_path_exact, script_hash: lcq.script_hash, @@ -1994,6 +1998,9 @@ pub fn filter_list_queue_query( sqlb.and_where_eq("trigger_kind", "?".bind(&format!("{}", tk))); } + if let Some(p) = &lq.trigger_path { + sqlb.and_where_eq("trigger", "?".bind(p)); + } sqlb } @@ -2220,46 +2227,50 @@ async fn cancel_selection( } #[derive(Deserialize)] -pub struct ResumeSuspendedJobsRequest { - suspend_number: i32, +#[serde(untagged)] +enum SuspendJobsRequest { + Trigger { trigger_path: String, trigger_kind: JobTriggerKind }, } async fn resume_suspended_jobs( authed: ApiAuthed, Extension(user_db): Extension, Path(w_id): Path, - Json(request): Json, + Json(request): Json, ) -> error::JsonResult { require_admin(authed.is_admin, &authed.username)?; let mut tx = user_db.begin(&authed).await?; - sqlx::query!( - r#" - WITH jobs_to_resume AS ( - SELECT - jq.id - FROM - v2_job_queue jq - INNER JOIN v2_job j ON j.id = jq.id - WHERE - j.workspace_id = $1 AND - jq.suspend = $2 AND - jq.canceled_by IS NULL - ) - UPDATE - v2_job_queue - SET - suspend = 0, - scheduled_for = NOW() - WHERE - id IN (SELECT id FROM jobs_to_resume) - "#, - w_id, - request.suspend_number - ) - .fetch_all(&mut *tx) - .await?; + match request { + SuspendJobsRequest::Trigger { trigger_path, trigger_kind } => { + let scheduled_for = INACTIVE_TRIGGER_SCHEDULED_FOR_DATE.clone(); + sqlx::query!( + r#" + UPDATE + v2_job_queue + SET + scheduled_for = NOW() + FROM + v2_job + WHERE + v2_job_queue.running is FALSE AND + v2_job_queue.id = v2_job.id + AND v2_job.workspace_id = $1 + AND v2_job_queue.scheduled_for = $2 + AND v2_job_queue.canceled_by IS NULL + AND v2_job.trigger_kind = $3 + AND v2_job.trigger = $4 + "#, + w_id, + scheduled_for, + trigger_kind as JobTriggerKind, + trigger_path + ) + .fetch_all(&mut *tx) + .await?; + } + } audit_log( &mut *tx, @@ -2267,81 +2278,78 @@ async fn resume_suspended_jobs( "jobs.resume_suspended", ActionKind::Update, &w_id, - Some(&format!("suspend_number:{}", request.suspend_number)), + None, None, ) .await?; tx.commit().await?; - Ok(Json(format!( - "Resumed all suspended workspace jobs for suspend number: {}", - request.suspend_number - ))) -} - -#[derive(Deserialize)] -pub struct CancelSuspendedJobsRequest { - suspend_number: i32, + Ok(Json(format!("Resumed all suspended workspace jobs",))) } async fn cancel_suspended_jobs( authed: ApiAuthed, Extension(user_db): Extension, Path(w_id): Path, - Json(request): Json, + Json(request): Json, ) -> error::JsonResult { require_admin(authed.is_admin, &authed.username)?; let mut tx = user_db.begin(&authed).await?; - sqlx::query!( - r#" - WITH jobs_to_cancel AS ( - SELECT - v2_job_queue.id - FROM - v2_job_queue - INNER JOIN v2_job ON v2_job.id = v2_job_queue.id - WHERE - v2_job.workspace_id = $1 AND - v2_job_queue.suspend = $2 AND - v2_job_queue.canceled_by IS NULL - ) - UPDATE - v2_job_queue - SET - canceled_by = $3, - canceled_reason = 'Canceled all suspended jobs with suspend number', - suspend = 0, - scheduled_for = now() - WHERE - id IN (SELECT id FROM jobs_to_cancel) - "#, - w_id, - request.suspend_number, - authed.username - ) - .fetch_all(&mut *tx) - .await?; + match request { + SuspendJobsRequest::Trigger { trigger_path, trigger_kind } => { + let scheduled_for = INACTIVE_TRIGGER_SCHEDULED_FOR_DATE.clone(); + let reason = format!( + "Canceled all suspended jobs for {} with trigger at path: {}", + &trigger_kind, &trigger_path + ); + sqlx::query!( + r#" + UPDATE + v2_job_queue + SET + canceled_by = $1, + canceled_reason = $2, + scheduled_for = NOW() + FROM + v2_job + WHERE + v2_job_queue.id = v2_job.id + AND v2_job_queue.running is FALSE + AND v2_job_queue.scheduled_for = $3 + AND v2_job.workspace_id = $4 + AND v2_job_queue.canceled_by IS NULL + AND v2_job.trigger_kind = $5 + AND v2_job.trigger = $6 + "#, + authed.username, + reason, + scheduled_for, + w_id, + trigger_kind as JobTriggerKind, + trigger_path + ) + .fetch_all(&mut *tx) + .await?; + } + } audit_log( &mut *tx, &authed, "jobs.cancel_suspended", ActionKind::Delete, &w_id, - Some(&format!("suspend_number:{}", request.suspend_number)), + None, None, ) .await?; tx.commit().await?; - Ok(Json(format!( - "Canceled all suspended workspace jobs for suspend number: {}", - request.suspend_number - ))) + Ok(Json(format!("Canceled all suspended workspace jobs",))) } async fn list_filtered_job_uuids( @@ -4027,7 +4035,7 @@ async fn batch_rerun_handle_job( // Call appropriate function to push job to queue match job.kind { JobKind::Flow => { - let result = run_flow_by_path_inner( + let result = push_flow_job_by_path_into_queue( authed.clone(), db.clone(), user_db.clone(), @@ -4044,7 +4052,7 @@ async fn batch_rerun_handle_job( } JobKind::Script => { let result = if use_latest_version { - run_script_by_path_inner( + push_script_job_by_path_into_queue( authed.clone(), db.clone(), user_db.clone(), @@ -4172,13 +4180,15 @@ pub async fn run_flow_by_path( ) .await?; - let (uuid, _) = - run_flow_by_path_inner(authed, db, user_db, w_id, flow_path, run_query, args, None).await?; + let (uuid, _) = push_flow_job_by_path_into_queue( + authed, db, user_db, w_id, flow_path, run_query, args, None, + ) + .await?; Ok((StatusCode::CREATED, uuid.to_string())) } -pub async fn run_flow_by_path_inner( +pub async fn push_flow_job_by_path_into_queue( authed: ApiAuthed, db: DB, user_db: UserDB, @@ -4186,7 +4196,7 @@ pub async fn run_flow_by_path_inner( flow_path: StripPath, run_query: RunJobQuery, args: PushArgsOwned, - trigger_kind: Option, + trigger: Option, ) -> error::Result<(Uuid, Option)> { #[cfg(feature = "enterprise")] check_license_key_valid().await?; @@ -4265,8 +4275,7 @@ pub async fn run_flow_by_path_inner( false, None, None, - None, - trigger_kind, + trigger, ) .await?; @@ -4385,7 +4394,6 @@ pub async fn restart_flow( None, None, None, - None, ) .await?; tx.commit().await?; @@ -4410,7 +4418,7 @@ pub async fn run_script_by_path( ) .await?; - let (uuid, _) = run_script_by_path_inner( + let (uuid, _) = push_script_job_by_path_into_queue( authed, db, user_db, @@ -4425,7 +4433,7 @@ pub async fn run_script_by_path( Ok((StatusCode::CREATED, uuid.to_string())) } -pub async fn run_script_by_path_inner( +pub async fn push_script_job_by_path_into_queue( authed: ApiAuthed, db: DB, user_db: UserDB, @@ -4433,7 +4441,7 @@ pub async fn run_script_by_path_inner( script_path: StripPath, run_query: RunJobQuery, args: PushArgsOwned, - trigger_kind: Option, + trigger: Option, ) -> error::Result<(Uuid, Option)> { #[cfg(feature = "enterprise")] check_license_key_valid().await?; @@ -4505,8 +4513,7 @@ pub async fn run_script_by_path_inner( false, None, None, - None, - trigger_kind, + trigger, ) .await?; tx.commit().await?; @@ -4665,7 +4672,6 @@ pub async fn run_workflow_as_code( None, None, None, - None, ) .await?; @@ -5211,7 +5217,6 @@ pub async fn run_wait_result_job_by_path_get( None, None, None, - None, ) .await?; tx.commit().await?; @@ -5367,7 +5372,6 @@ pub async fn run_wait_result_script_by_path_internal( None, None, None, - None, ) .await?; tx.commit().await?; @@ -5491,7 +5495,6 @@ pub async fn run_wait_result_script_by_hash( None, None, None, - None, ) .await?; tx.commit().await?; @@ -5642,7 +5645,7 @@ pub async fn stream_job( let uuid = match runnable_id { RunnableId::ScriptId(ScriptId::ScriptPath(script_path)) | RunnableId::HubScript(script_path) => { - run_script_by_path_inner( + push_script_job_by_path_into_queue( authed.clone(), db.clone(), user_db, @@ -5669,7 +5672,7 @@ pub async fn stream_job( .0 } RunnableId::FlowPath(flow_path) => { - run_flow_by_path_inner( + push_flow_job_by_path_into_queue( authed.clone(), db.clone(), user_db, @@ -5808,7 +5811,6 @@ pub async fn run_wait_result_flow_by_path_internal( None, None, None, - None, ) .await?; @@ -5904,7 +5906,6 @@ async fn run_preview_script( None, None, None, - None, ) .await?; tx.commit().await?; @@ -6026,7 +6027,6 @@ async fn run_bundle_preview_script( None, None, None, - None, ) .await?; job_id = Some(uuid); @@ -6167,7 +6167,6 @@ async fn run_dependencies_job( None, None, None, - None, ) .await?; tx.commit().await?; @@ -6238,7 +6237,6 @@ async fn run_flow_dependencies_job( None, None, None, - None, ) .await?; tx.commit().await?; @@ -6593,7 +6591,6 @@ async fn run_preview_flow_job( None, None, None, - None, ) .await?; @@ -6678,7 +6675,7 @@ async fn run_dynamic_select( let push_args = PushArgsOwned { extra: None, args: script_args.clone() }; - let (uuid, _) = run_script_by_path_inner( + let (uuid, _) = push_script_job_by_path_into_queue( authed.clone(), db.clone(), user_db.clone(), @@ -6792,7 +6789,6 @@ async fn run_dynamic_select( None, None, None, - None, ) .await?; tx.commit().await?; @@ -6927,7 +6923,6 @@ pub async fn run_job_by_hash_inner( None, None, None, - None, ) .await?; tx.commit().await?; @@ -7789,6 +7784,10 @@ pub fn filter_list_completed_query( sqlb.and_where_eq("trigger_kind", "'schedule'"); } + if let Some(p) = &lq.trigger_path { + sqlb.and_where_eq("trigger", "?".bind(p)); + } + if let Some(ps) = &lq.script_path_start { sqlb.and_where_like_left("runnable_path", ps); } @@ -7963,6 +7962,7 @@ pub struct ListCompletedQuery { pub is_flow_step: Option, pub suspended: Option, pub schedule_path: Option, + pub trigger_path: Option, // filter by matching a subset of the args using base64 encoded json subset pub args: Option, // filter by matching a subset of the result using base64 encoded json subset diff --git a/backend/windmill-api/src/scripts.rs b/backend/windmill-api/src/scripts.rs index 0a15028f0b..b45f53c9ec 100644 --- a/backend/windmill-api/src/scripts.rs +++ b/backend/windmill-api/src/scripts.rs @@ -1022,8 +1022,7 @@ async fn create_script_internal<'c>( false, None, None, - None, - None, + None ) .await?; Ok((hash, new_tx, None)) diff --git a/backend/windmill-api/src/settings.rs b/backend/windmill-api/src/settings.rs index 4851130286..de00d3cbe6 100644 --- a/backend/windmill-api/src/settings.rs +++ b/backend/windmill-api/src/settings.rs @@ -188,7 +188,7 @@ pub async fn test_license_key( Json(TestKey { license_key }): Json, ) -> error::Result { require_super_admin(&db, &authed.email).await?; - let (_, expired) = validate_license_key(license_key).await?; + let (_, expired) = validate_license_key(license_key, Some(&db)).await?; if expired { Err(error::Error::BadRequest("Expired license key".to_string())) diff --git a/backend/windmill-api/src/triggers/email/handler_oss.rs b/backend/windmill-api/src/triggers/email/handler_oss.rs index 363f515f40..2f3a318bdd 100644 --- a/backend/windmill-api/src/triggers/email/handler_oss.rs +++ b/backend/windmill-api/src/triggers/email/handler_oss.rs @@ -46,7 +46,6 @@ impl TriggerCrud for EmailTrigger { _authed: &ApiAuthed, _w_id: &str, _trigger: TriggerData, - _suspend_number: Option, ) -> Result<()> { Err(Error::BadRequest( "Email triggers are not available in open source version".to_string(), @@ -61,7 +60,6 @@ impl TriggerCrud for EmailTrigger { _workspace_id: &str, _path: &str, _trigger: TriggerData, - _suspend_number: Option, ) -> Result<()> { Err(Error::BadRequest( "Email triggers are not available in open source version".to_string(), diff --git a/backend/windmill-api/src/triggers/handler.rs b/backend/windmill-api/src/triggers/handler.rs index d10a7848d2..d1d3d8c341 100644 --- a/backend/windmill-api/src/triggers/handler.rs +++ b/backend/windmill-api/src/triggers/handler.rs @@ -1,6 +1,5 @@ use crate::{ db::ApiAuthed, - jobs::generate_unique_suspend_number, triggers::{StandardTriggerQuery, TriggerData, BASE_TRIGGER_FIELDS}, }; use async_trait::async_trait; @@ -100,7 +99,6 @@ pub trait TriggerCrud: Send + Sync + 'static { authed: &ApiAuthed, w_id: &str, trigger: TriggerData, - suspend_number: Option, ) -> Result<()>; async fn update_trigger( @@ -111,7 +109,6 @@ pub trait TriggerCrud: Send + Sync + 'static { workspace_id: &str, path: &str, trigger: TriggerData, - suspend_number: Option, ) -> Result<()>; async fn test_connection( @@ -290,7 +287,7 @@ pub trait TriggerCrud: Send + Sync + 'static { "email", "edited_at", "extra_perms", - "suspend_number", + "active_mode", ]; if Self::SUPPORTS_SERVER_STATE { @@ -354,20 +351,6 @@ pub fn trigger_routes() -> Router { router } -pub async fn get_suspend_number_for_inactive_mode( - db: &DB, - workspace_id: &str, - active_mode: Option, -) -> Result> { - if let Some(false) = active_mode { - Ok(Some( - generate_unique_suspend_number(db, workspace_id).await?, - )) - } else { - Ok(None) - } -} - async fn create_trigger( Extension(handler): Extension>, authed: ApiAuthed, @@ -398,19 +381,9 @@ async fn create_trigger( let mut tx = user_db.begin(&authed).await?; let new_path = new_trigger.base.path.clone(); - let suspend_number = - get_suspend_number_for_inactive_mode(&db, &workspace_id, new_trigger.base.active_mode) - .await?; handler - .create_trigger( - &db, - &mut *tx, - &authed, - &workspace_id, - new_trigger, - suspend_number, - ) + .create_trigger(&db, &mut *tx, &authed, &workspace_id, new_trigger) .await?; audit_log( @@ -501,20 +474,9 @@ async fn update_trigger( let mut tx = user_db.begin(&authed).await?; let new_path = edit_trigger.base.path.to_string(); - let suspend_number = - get_suspend_number_for_inactive_mode(&db, &workspace_id, edit_trigger.base.active_mode) - .await?; handler - .update_trigger( - &db, - &mut *tx, - &authed, - &workspace_id, - path, - edit_trigger, - suspend_number, - ) + .update_trigger(&db, &mut *tx, &authed, &workspace_id, path, edit_trigger) .await?; audit_log( diff --git a/backend/windmill-api/src/triggers/http/handler.rs b/backend/windmill-api/src/triggers/http/handler.rs index c3b35f68c3..f98c8ee7ca 100644 --- a/backend/windmill-api/src/triggers/http/handler.rs +++ b/backend/windmill-api/src/triggers/http/handler.rs @@ -8,7 +8,6 @@ use crate::{ jobs::start_job_update_sse_stream, resources::try_get_resource_from_db_as, triggers::{ - handler::get_suspend_number_for_inactive_mode, http::{ refresh_routers, validate_authentication_method, HttpConfig, HttpConfigRequest, RouteExists, ROUTE_PATH_KEY_RE, VALID_ROUTE_PATH_RE, @@ -41,7 +40,8 @@ use windmill_audit::{audit_oss::audit_log, ActionKind}; use windmill_common::{ db::UserDB, error::{Error, Result}, - triggers::TriggerKind, + jobs::JobTriggerKind, + triggers::{TriggerInfo, TriggerKind}, utils::{not_found_if_none, require_admin, StripPath}, worker::CLOUD_HOSTED, }; @@ -177,7 +177,6 @@ pub async fn insert_new_trigger_into_db( w_id: &str, trigger: &TriggerData, route_path_key: &str, - suspend_number: Option, ) -> Result<()> { require_admin(authed.is_admin, &authed.username)?; @@ -209,7 +208,7 @@ pub async fn insert_new_trigger_into_db( error_handler_path, error_handler_args, retry, - suspend_number + active_mode ) VALUES ( $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, now(), $19, $20, $21, $22, $23 @@ -237,7 +236,7 @@ pub async fn insert_new_trigger_into_db( trigger.error_handling.error_handler_path, trigger.error_handling.error_handler_args as _, trigger.error_handling.retry as _, - suspend_number + trigger.base.active_mode.unwrap_or(true) ) .execute(&mut *tx) .await?; @@ -283,19 +282,9 @@ pub async fn create_many_http_triggers( let mut tx = user_db.begin(&authed).await?; for (new_http_trigger, route_path_key) in new_http_triggers.iter().zip(route_path_keys.iter()) { - let suspend_number = - get_suspend_number_for_inactive_mode(&db, &w_id, new_http_trigger.base.active_mode) - .await?; - insert_new_trigger_into_db( - &authed, - &mut tx, - &w_id, - new_http_trigger, - route_path_key, - suspend_number, - ) - .await - .map_err(|err| error_wrapper(&new_http_trigger.config.route_path, err))?; + insert_new_trigger_into_db(&authed, &mut tx, &w_id, new_http_trigger, route_path_key) + .await + .map_err(|err| error_wrapper(&new_http_trigger.config.route_path, err))?; audit_log( &mut *tx, @@ -448,12 +437,10 @@ impl TriggerCrud for HttpTrigger { authed: &ApiAuthed, w_id: &str, trigger: TriggerData, - suspend_number: Option, ) -> Result<()> { let route_path_key = check_if_route_exist(db, &trigger.config, &w_id, None).await?; - insert_new_trigger_into_db(authed, tx, w_id, &trigger, &route_path_key, suspend_number) - .await?; + insert_new_trigger_into_db(authed, tx, w_id, &trigger, &route_path_key).await?; increase_trigger_version(tx).await?; @@ -468,7 +455,6 @@ impl TriggerCrud for HttpTrigger { workspace_id: &str, path: &str, trigger: TriggerData, - suspend_number: Option, ) -> Result<()> { if authed.is_admin { if trigger.config.route_path.is_empty() { @@ -512,7 +498,7 @@ impl TriggerCrud for HttpTrigger { error_handler_path = $19, error_handler_args = $20, retry = $21, - suspend_number = $22 + active_mode = $22 WHERE workspace_id = $23 AND path = $24 @@ -538,7 +524,7 @@ impl TriggerCrud for HttpTrigger { trigger.error_handling.error_handler_path, trigger.error_handling.error_handler_args as _, trigger.error_handling.retry as _, - suspend_number, + trigger.base.active_mode.unwrap_or(true), workspace_id, path, ) @@ -571,7 +557,7 @@ impl TriggerCrud for HttpTrigger { error_handler_path = $16, error_handler_args = $17, retry = $18, - suspend_number = $19 + active_mode = $19 WHERE workspace_id = $20 AND path = $21 @@ -594,7 +580,7 @@ impl TriggerCrud for HttpTrigger { trigger.error_handling.error_handler_path, trigger.error_handling.error_handler_args as _, trigger.error_handling.retry as _, - suspend_number, + trigger.base.active_mode.unwrap_or(true), workspace_id, path, ) @@ -1054,7 +1040,8 @@ async fn route_job( ) .map_err(|e| e.into_response())?; - if let Some(suspend_number) = trigger.suspend_number { + let trigger_info = TriggerInfo::new(Some(trigger.path.clone()), JobTriggerKind::Http); + if !trigger.active_mode { let _ = trigger_runnable( &db, Some(user_db), @@ -1068,8 +1055,8 @@ async fn route_job( trigger.error_handler_args.as_ref(), format!("http_trigger/{}", trigger.path), None, - Some(suspend_number), - Some(windmill_common::jobs::JobTriggerKind::Http) + trigger.active_mode, + trigger_info, ) .await .map_err(|e| e.into_response())?; @@ -1101,7 +1088,8 @@ async fn route_job( trigger.error_handler_args.as_ref(), format!("http_trigger/{}", trigger.path), None, - Some(windmill_common::jobs::JobTriggerKind::Http), + trigger_info, + None, ) .await .map_err(|e| e.into_response())?; @@ -1160,8 +1148,8 @@ async fn route_job( trigger.error_handler_args.as_ref(), format!("http_trigger/{}", trigger.path), None, - None, - Some(windmill_common::jobs::JobTriggerKind::Http), + trigger.active_mode, + trigger_info, ) .await .map_err(|e| e.into_response()), @@ -1177,7 +1165,7 @@ async fn route_job( trigger.error_handler_path.as_deref(), trigger.error_handler_args.as_ref(), format!("http_trigger/{}", trigger.path), - Some(windmill_common::jobs::JobTriggerKind::Http), + trigger_info, ) .await .map_err(|e| e.into_response()), diff --git a/backend/windmill-api/src/triggers/http/mod.rs b/backend/windmill-api/src/triggers/http/mod.rs index 446dad7728..0025d92be7 100644 --- a/backend/windmill-api/src/triggers/http/mod.rs +++ b/backend/windmill-api/src/triggers/http/mod.rs @@ -48,7 +48,7 @@ pub struct TriggerRoute { error_handler_path: Option, error_handler_args: Option>>, retry: Option>, - suspend_number: Option, + active_mode: bool, } pub struct RoutersCache { @@ -259,7 +259,7 @@ pub async fn refresh_routers(db: &DB) -> Result<(bool, RwLockReadGuard<'_, Route error_handler_path, error_handler_args as "error_handler_args: _", retry as "retry: _", - suspend_number + active_mode FROM http_trigger WHERE diff --git a/backend/windmill-api/src/triggers/listener.rs b/backend/windmill-api/src/triggers/listener.rs index 421b3c4b09..f931d036d8 100644 --- a/backend/windmill-api/src/triggers/listener.rs +++ b/backend/windmill-api/src/triggers/listener.rs @@ -22,7 +22,7 @@ use tokio::sync::RwLock; use windmill_common::{ error::{Error, Result}, jobs::JobTriggerKind, - triggers::TriggerKind, + triggers::{TriggerInfo, TriggerKind}, utils::report_critical_error, DB, INSTANCE_NAME, }; @@ -95,7 +95,7 @@ pub trait Listener: TriggerCrud + TriggerJobArgs { trigger_config: trigger.config, error_handling: Some(trigger.error_handling), trigger_mode: true, - suspend_number: trigger.base.suspend_number, + active_mode: Some(trigger.base.active_mode), }) .collect_vec(); @@ -145,7 +145,7 @@ pub trait Listener: TriggerCrud + TriggerJobArgs { trigger_mode: false, is_flow: capture.is_flow, error_handling: None, - suspend_number: None, + active_mode: None, }) .collect_vec(); @@ -506,8 +506,8 @@ pub trait Listener: TriggerCrud + TriggerJobArgs { error_handler_args, format!("{}_trigger/{}", Self::TRIGGER_KIND, listening_trigger.path), None, - listening_trigger.suspend_number, - Some(Self::JOB_TRIGGER_KIND), + listening_trigger.active_mode.unwrap_or(false), + TriggerInfo::new(Some(listening_trigger.path.clone()), Self::JOB_TRIGGER_KIND), ) .await?; @@ -802,7 +802,7 @@ pub struct ListeningTrigger { pub script_path: String, pub trigger_mode: bool, pub error_handling: Option, - pub suspend_number: Option, + pub active_mode: Option, } impl ListeningTrigger { diff --git a/backend/windmill-api/src/triggers/mod.rs b/backend/windmill-api/src/triggers/mod.rs index 21c8fd5dc7..276e741171 100644 --- a/backend/windmill-api/src/triggers/mod.rs +++ b/backend/windmill-api/src/triggers/mod.rs @@ -1,7 +1,15 @@ -use chrono::{DateTime, Utc}; +use chrono::{DateTime, TimeZone, Utc}; use serde::{Deserialize, Serialize}; use sqlx::{types::Json as SqlxJson, FromRow}; use std::{collections::HashMap, fmt::Debug}; +use windmill_common::jobs::JobTriggerKind; + +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(tag = "type", rename_all = "snake_case")] +pub enum HandlerAction { + Trigger { path: String, trigger_kind: JobTriggerKind }, + // Future variants can be added here (e.g., Script, Flow, etc.) +} #[cfg(all(feature = "smtp", feature = "enterprise", feature = "private"))] pub mod email; @@ -52,7 +60,7 @@ pub struct BaseTrigger { pub email: String, pub edited_at: DateTime, pub extra_perms: Option, - pub suspend_number: Option, + pub active_mode: bool, } #[derive(Debug, FromRow, Clone, Serialize, Deserialize)] @@ -159,5 +167,9 @@ pub const BASE_TRIGGER_FIELDS: [&'static str; 9] = [ "email", "edited_at", "extra_perms", - "suspend_number", + "active_mode", ]; + +lazy_static::lazy_static! { + pub static ref INACTIVE_TRIGGER_SCHEDULED_FOR_DATE: DateTime = Utc.with_ymd_and_hms(9999, 12, 31, 23, 59, 59).unwrap(); +} diff --git a/backend/windmill-api/src/triggers/mqtt/handler.rs b/backend/windmill-api/src/triggers/mqtt/handler.rs index cdd2093c07..b4db9e2ed6 100644 --- a/backend/windmill-api/src/triggers/mqtt/handler.rs +++ b/backend/windmill-api/src/triggers/mqtt/handler.rs @@ -73,7 +73,6 @@ impl TriggerCrud for MqttTrigger { authed: &ApiAuthed, w_id: &str, trigger: TriggerData, - suspend_number: Option, ) -> Result<()> { let subscribe_topics = trigger .config @@ -103,7 +102,7 @@ impl TriggerCrud for MqttTrigger { error_handler_path, error_handler_args, retry, - suspend_number + active_mode ) VALUES ( $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17 @@ -124,7 +123,7 @@ impl TriggerCrud for MqttTrigger { trigger.error_handling.error_handler_path, trigger.error_handling.error_handler_args as _, trigger.error_handling.retry as _, - suspend_number + trigger.base.active_mode.unwrap_or(true) ) .execute(tx) .await?; @@ -140,7 +139,6 @@ impl TriggerCrud for MqttTrigger { workspace_id: &str, path: &str, trigger: TriggerData, - suspend_number: Option, ) -> Result<()> { let subscribe_topics = trigger .config @@ -174,7 +172,7 @@ impl TriggerCrud for MqttTrigger { error_handler_path = $14, error_handler_args = $15, retry = $16, - suspend_number = $17 + active_mode = $17 WHERE workspace_id = $12 AND path = $13 @@ -195,7 +193,7 @@ impl TriggerCrud for MqttTrigger { trigger.error_handling.error_handler_path, trigger.error_handling.error_handler_args as _, trigger.error_handling.retry as _, - suspend_number + trigger.base.active_mode.unwrap_or(true) ) .execute(tx) .await?; diff --git a/backend/windmill-api/src/triggers/postgres/handler.rs b/backend/windmill-api/src/triggers/postgres/handler.rs index f0156fdd72..0bdfe99595 100644 --- a/backend/windmill-api/src/triggers/postgres/handler.rs +++ b/backend/windmill-api/src/triggers/postgres/handler.rs @@ -71,7 +71,6 @@ impl TriggerCrud for PostgresTrigger { authed: &ApiAuthed, w_id: &str, trigger: TriggerData, - suspend_number: Option, ) -> Result<()> { let Self::TriggerConfigRequest { postgres_resource_path, @@ -128,7 +127,7 @@ impl TriggerCrud for PostgresTrigger { error_handler_path, error_handler_args, retry, - suspend_number + active_mode ) VALUES ( $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, now(), $11, $12, $13, $14 ) @@ -146,7 +145,7 @@ impl TriggerCrud for PostgresTrigger { trigger.error_handling.error_handler_path, trigger.error_handling.error_handler_args as _, trigger.error_handling.retry as _, - suspend_number + trigger.base.active_mode.unwrap_or(true) ) .execute(tx) .await?; @@ -161,7 +160,6 @@ impl TriggerCrud for PostgresTrigger { w_id: &str, path: &str, trigger: TriggerData, - suspend_number: Option, ) -> Result<()> { let Self::TriggerConfigRequest { replication_slot_name, @@ -232,7 +230,7 @@ impl TriggerCrud for PostgresTrigger { error_handler_path = $11, error_handler_args = $12, retry = $13, - suspend_number = $14 + active_mode = $14 WHERE workspace_id = $9 AND path = $10 "#, @@ -249,7 +247,7 @@ impl TriggerCrud for PostgresTrigger { trigger.error_handling.error_handler_path, trigger.error_handling.error_handler_args as _, trigger.error_handling.retry as _, - suspend_number + trigger.base.active_mode.unwrap_or(true) ) .execute(tx) .await?; diff --git a/backend/windmill-api/src/triggers/trigger_helpers.rs b/backend/windmill-api/src/triggers/trigger_helpers.rs index f938253092..e357b7b068 100644 --- a/backend/windmill-api/src/triggers/trigger_helpers.rs +++ b/backend/windmill-api/src/triggers/trigger_helpers.rs @@ -1,31 +1,17 @@ use anyhow::Context; use axum::response::IntoResponse; -use chrono::{TimeZone, Utc}; +use chrono::{DateTime, Utc}; use http::StatusCode; use serde::Deserialize; use serde_json::value::RawValue; use sqlx::types::Json; +use std::collections::HashMap; use std::future::Future; -use std::{collections::HashMap, i32}; use uuid::Uuid; use windmill_common::{ - db::{UserDB, UserDbWithAuthed}, - error::Result, - flows::{FlowModuleValue, Retry}, - get_latest_deployed_hash_for_path, get_latest_flow_version_info_for_path, - jobs::{ - get_has_preprocessor_from_content_and_lang, script_path_to_payload, JobPayload, - JobTriggerKind, - }, - scripts::{get_full_hub_script_by_path, ScriptHash, ScriptLang}, - triggers::{ - HubOrWorkspaceId, RunnableFormat, RunnableFormatVersion, TriggerKind, - RUNNABLE_FORMAT_VERSION_CACHE, - }, - users::username_to_permissioned_as, - utils::StripPath, - worker::to_raw_value, - FlowVersionInfo, + FlowVersionInfo, db::{UserDB, UserDbWithAuthed}, error::Result, flows::{FlowModuleValue, Retry}, get_latest_deployed_hash_for_path, get_latest_flow_version_info_for_path, jobs::{JobPayload, get_has_preprocessor_from_content_and_lang, script_path_to_payload}, scripts::{ScriptHash, ScriptLang, get_full_hub_script_by_path}, triggers::{ + HubOrWorkspaceId, RUNNABLE_FORMAT_VERSION_CACHE, RunnableFormat, RunnableFormatVersion, TriggerInfo, TriggerKind + }, users::username_to_permissioned_as, utils::StripPath, worker::to_raw_value }; use windmill_queue::{push, PushArgs, PushArgsOwned, PushIsolationLevel}; @@ -34,9 +20,11 @@ use crate::jobs::check_license_key_valid; use crate::{ db::{ApiAuthed, DB}, jobs::{ - check_tag_available_for_workspace, delete_job_metadata_after_use, result_to_response, - run_flow_by_path_inner, run_script_by_path_inner, run_wait_result_internal, RunJobQuery, + check_tag_available_for_workspace, delete_job_metadata_after_use, + push_flow_job_by_path_into_queue, push_script_job_by_path_into_queue, result_to_response, + run_wait_result_internal, RunJobQuery, }, + triggers::INACTIVE_TRIGGER_SCHEDULED_FOR_DATE, utils::check_scopes, HTTP_CLIENT, }; @@ -498,7 +486,8 @@ pub async fn trigger_runnable_inner( error_handler_args: Option<&sqlx::types::Json>>, trigger_path: String, job_id: Option, - trigger_kind: Option, + trigger: TriggerInfo, + active_mode: Option, ) -> Result<(Uuid, Option, Option)> { let error_handler_args = error_handler_args.map(|args| { let args = args @@ -510,10 +499,13 @@ 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 (uuid, delete_after_use, early_return) = if is_flow { - let run_query = RunJobQuery { job_id, ..Default::default() }; + let run_query = RunJobQuery { job_id, scheduled_for, ..Default::default() }; let path = StripPath(runnable_path.to_string()); - let (uuid, early_return) = run_flow_by_path_inner( + let (uuid, early_return) = push_flow_job_by_path_into_queue( authed, db.clone(), user_db, @@ -521,7 +513,7 @@ pub async fn trigger_runnable_inner( path, run_query, args, - trigger_kind, + Some(trigger), ) .await?; (uuid, None, early_return) @@ -538,7 +530,8 @@ pub async fn trigger_runnable_inner( error_handler_args.as_ref(), trigger_path, job_id, - trigger_kind, + trigger, + scheduled_for, ) .await?; (uuid, delete_after_use, None) @@ -561,49 +554,27 @@ pub async fn trigger_runnable( error_handler_args: Option<&sqlx::types::Json>>, trigger_path: String, job_id: Option, - suspend_number: Option, - trigger_kind: Option, + active_mode: bool, + trigger: TriggerInfo, ) -> Result { - let uuid = match suspend_number { - Some(suspend_number) => { - trigger_runnable_with_suspend( - db, - user_db, - authed, - workspace_id, - runnable_path, - is_flow, - args, - retry, - error_handler_path, - error_handler_args, - trigger_path, - job_id, - suspend_number, - trigger_kind, - ) - .await? - } - _ => { - trigger_runnable_inner( - db, - user_db, - authed, - workspace_id, - runnable_path, - is_flow, - args, - retry, - error_handler_path, - error_handler_args, - trigger_path, - job_id, - trigger_kind, - ) - .await? - .0 - } - }; + let uuid = trigger_runnable_inner( + db, + user_db, + authed, + workspace_id, + runnable_path, + is_flow, + args, + retry, + error_handler_path, + error_handler_args, + trigger_path, + job_id, + trigger, + Some(active_mode), + ) + .await? + .0; Ok((StatusCode::CREATED, uuid.to_string()).into_response()) } @@ -620,7 +591,7 @@ pub async fn trigger_runnable_and_wait_for_result( error_handler_path: Option<&str>, error_handler_args: Option<&sqlx::types::Json>>, trigger_path: String, - trigger_kind: Option, + trigger: TriggerInfo, ) -> Result { let username = authed.username.clone(); let (uuid, delete_after_use, early_return) = trigger_runnable_inner( @@ -636,7 +607,8 @@ pub async fn trigger_runnable_and_wait_for_result( error_handler_args, trigger_path, None, - trigger_kind, + trigger, + None, ) .await?; let (result, success) = @@ -663,7 +635,7 @@ pub async fn trigger_runnable_and_wait_for_raw_result( error_handler_path: Option<&str>, error_handler_args: Option<&sqlx::types::Json>>, trigger_path: String, - trigger_kind: Option, + trigger: TriggerInfo, ) -> Result<(Box, bool)> { let username = authed.username.clone(); let (uuid, delete_after_use, early_return) = trigger_runnable_inner( @@ -679,7 +651,8 @@ pub async fn trigger_runnable_and_wait_for_raw_result( error_handler_args, trigger_path, None, - trigger_kind, + trigger, + None, ) .await?; @@ -713,7 +686,7 @@ pub async fn trigger_runnable_and_wait_for_raw_result_with_error_ctx( error_handler_path: Option<&str>, error_handler_args: Option<&sqlx::types::Json>>, trigger_path: String, - trigger_kind: Option, + trigger: TriggerInfo, ) -> Result> { let (result, success) = trigger_runnable_and_wait_for_raw_result( db, @@ -727,7 +700,7 @@ pub async fn trigger_runnable_and_wait_for_raw_result_with_error_ctx( error_handler_path, error_handler_args, trigger_path, - trigger_kind, + trigger, ) .await?; @@ -754,12 +727,13 @@ async fn trigger_script_internal( error_handler_args: Option<&sqlx::types::Json>>>, trigger_path: String, job_id: Option, - trigger_kind: Option, + trigger: TriggerInfo, + scheduled_for: Option>, ) -> Result<(Uuid, Option)> { if retry.is_none() && error_handler_path.is_none() { - let run_query = RunJobQuery { job_id, ..Default::default() }; + let run_query = RunJobQuery { job_id, scheduled_for, ..Default::default() }; let path = StripPath(script_path.to_string()); - run_script_by_path_inner( + push_script_job_by_path_into_queue( authed, db.clone(), user_db, @@ -767,7 +741,7 @@ async fn trigger_script_internal( path, run_query, args, - trigger_kind, + Some(trigger), ) .await } else { @@ -783,7 +757,8 @@ async fn trigger_script_internal( error_handler_args, trigger_path, job_id, - trigger_kind, + trigger, + scheduled_for, ) .await } @@ -801,7 +776,8 @@ async fn trigger_script_with_retry_and_error_handler( error_handler_args: Option<&sqlx::types::Json>>>, trigger_path: String, job_id: Option, - trigger_kind: Option, + trigger: TriggerInfo, + scheduled_for: Option>, ) -> Result<(Uuid, Option)> { #[cfg(feature = "enterprise")] check_license_key_valid().await?; @@ -874,7 +850,7 @@ async fn trigger_script_with_retry_and_error_handler( priority, tag_override: tag.clone(), apply_preprocessor, - trigger_path: Some(trigger_path), + trigger_path: Some(trigger_path.clone()), custom_debounce_key, debounce_delay_s, }, @@ -885,7 +861,6 @@ async fn trigger_script_with_retry_and_error_handler( ))) } }; - let (uuid, tx) = push( &db, tx, @@ -896,7 +871,7 @@ async fn trigger_script_with_retry_and_error_handler( email, permissioned_as, authed.token_prefix.as_deref(), - None, + scheduled_for, None, None, None, @@ -914,217 +889,10 @@ async fn trigger_script_with_retry_and_error_handler( false, None, None, - None, - trigger_kind, + Some(trigger), ) .await?; tx.commit().await?; Ok((uuid, delete_after_use)) } - - -async fn trigger_runnable_with_suspend( - db: &DB, - user_db: Option, - authed: ApiAuthed, - workspace_id: &str, - runnable_path: &str, - is_flow: bool, - args: PushArgsOwned, - retry: Option<&sqlx::types::Json>, - error_handler_path: Option<&str>, - error_handler_args: Option<&sqlx::types::Json>>, - trigger_path: String, - job_id: Option, - suspend_number: i32, - trigger_kind: Option, -) -> Result { - let far_future_utc = Utc.with_ymd_and_hms(9999, 12, 31, 23, 59, 59).unwrap(); - if is_flow { - let user_db = user_db.unwrap_or_else(|| UserDB::new(db.clone())); - let db_authed = UserDbWithAuthed { db: user_db.clone(), authed: &authed.to_authed_ref() }; - - let FlowVersionInfo { version, .. } = get_latest_flow_version_info_for_path( - Some(db_authed), - &db, - workspace_id, - runnable_path, - false, - ) - .await?; - - let (email, permissioned_as, push_authed, tx) = ( - authed.email.as_str(), - username_to_permissioned_as(&authed.username), - Some(authed.clone().into()), - PushIsolationLevel::Isolated(user_db, authed.clone().into()), - ); - - let push_args = windmill_queue::PushArgs { args: &args.args, extra: args.extra }; - - let flow_payload = windmill_common::jobs::JobPayload::Flow { - path: runnable_path.to_string(), - version: version, - dedicated_worker: None, - apply_preprocessor: false, - }; - - let (uuid, tx) = push( - db, - tx, - workspace_id, - flow_payload, - push_args, - authed.display_username(), - email, - permissioned_as, - authed.token_prefix.as_deref(), - Some(far_future_utc), - None, - None, - None, - None, - job_id, - false, - false, - None, - true, - None, - None, - None, - None, - push_authed.as_ref(), - false, - None, - None, - Some(suspend_number), - trigger_kind - ) - .await?; - - tx.commit().await?; - Ok(uuid) - } else { - use crate::jobs::check_tag_available_for_workspace; - use windmill_common::db::UserDbWithAuthed; - use windmill_common::jobs::script_path_to_payload; - use windmill_common::users::username_to_permissioned_as; - use windmill_queue::{push, PushIsolationLevel}; - - let error_handler_args = error_handler_args.map(|args| { - args.0 - .iter() - .map(|(key, value)| (key.to_owned(), to_raw_value(&value))) - .collect::>>() - }); - - let user_db = user_db.unwrap_or_else(|| UserDB::new(db.clone())); - let db_authed = UserDbWithAuthed { db: user_db.clone(), authed: &authed.to_authed_ref() }; - let (job_payload, tag, _delete_after_use, timeout, on_behalf_of) = script_path_to_payload( - runnable_path, - Some(db_authed), - db.clone(), - workspace_id, - Some(false), - ) - .await?; - - check_tag_available_for_workspace(db, workspace_id, &tag, &authed).await?; - - let (email, permissioned_as, push_authed, tx) = - if let Some(on_behalf_of) = on_behalf_of.as_ref() { - ( - on_behalf_of.email.as_str(), - on_behalf_of.permissioned_as.clone(), - None, - PushIsolationLevel::IsolatedRoot(db.clone()), - ) - } else { - ( - authed.email.as_str(), - username_to_permissioned_as(&authed.username), - Some(authed.clone().into()), - PushIsolationLevel::Isolated(user_db, authed.clone().into()), - ) - }; - - let push_args = windmill_queue::PushArgs { args: &args.args, extra: args.extra }; - - let retryable_job_payload = match job_payload { - windmill_common::jobs::JobPayload::ScriptHash { - hash, - path, - custom_concurrency_key, - concurrent_limit, - concurrency_time_window_s, - custom_debounce_key, - debounce_delay_s, - cache_ttl, - priority, - apply_preprocessor, - .. - } => windmill_common::jobs::JobPayload::SingleStepFlow { - path, - hash: Some(hash), - flow_version: None, - args: HashMap::from(&push_args), - retry: retry.map(|r| r.0.clone()), - error_handler_path: error_handler_path.map(|s| s.to_string()), - error_handler_args, - skip_handler: None, - custom_concurrency_key, - concurrent_limit, - concurrency_time_window_s, - cache_ttl, - priority, - tag_override: tag.clone(), - apply_preprocessor, - trigger_path: Some(trigger_path), - custom_debounce_key, - debounce_delay_s, - }, - _ => { - return Err(windmill_common::error::Error::internal_err(format!( - "Unsupported job payload for suspended execution" - ))) - } - }; - - let (uuid, tx) = push( - db, - tx, - workspace_id, - retryable_job_payload, - push_args, - authed.display_username(), - email, - permissioned_as, - authed.token_prefix.as_deref(), - Some(far_future_utc), - None, - None, - None, - None, - job_id, - false, - false, - None, - true, - tag, - timeout, - None, - None, - push_authed.as_ref(), - false, - None, - None, - Some(suspend_number), - trigger_kind - ) - .await?; - - tx.commit().await?; - Ok(uuid) - } -} diff --git a/backend/windmill-api/src/triggers/websocket/handler.rs b/backend/windmill-api/src/triggers/websocket/handler.rs index f1c39b720f..1dcfd956e5 100644 --- a/backend/windmill-api/src/triggers/websocket/handler.rs +++ b/backend/windmill-api/src/triggers/websocket/handler.rs @@ -79,7 +79,6 @@ impl TriggerCrud for WebsocketTrigger { authed: &ApiAuthed, w_id: &str, trigger: TriggerData, - suspend_number: Option, ) -> Result<()> { let filters = trigger .config @@ -114,7 +113,7 @@ impl TriggerCrud for WebsocketTrigger { error_handler_path, error_handler_args, retry, - suspend_number + active_mode ) VALUES ( $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, now(), $14, $15, $16, $17 ) @@ -138,7 +137,7 @@ impl TriggerCrud for WebsocketTrigger { trigger.error_handling.error_handler_path, trigger.error_handling.error_handler_args as _, trigger.error_handling.retry as _, - suspend_number + trigger.base.active_mode.unwrap_or(true) ) .execute(&mut *tx) .await?; @@ -153,7 +152,6 @@ impl TriggerCrud for WebsocketTrigger { w_id: &str, path: &str, trigger: TriggerData, - suspend_number: Option, ) -> Result<()> { let filters = trigger .config @@ -192,7 +190,7 @@ impl TriggerCrud for WebsocketTrigger { error_handler_path = $14, error_handler_args = $15, retry = $16, - suspend_number = $17 + active_mode = $17 WHERE workspace_id = $12 AND path = $13 ", @@ -216,7 +214,7 @@ impl TriggerCrud for WebsocketTrigger { trigger.error_handling.error_handler_path, trigger.error_handling.error_handler_args as _, trigger.error_handling.retry as _, - suspend_number + trigger.base.active_mode.unwrap_or(true) ) .execute(&mut *tx) .await?; diff --git a/backend/windmill-api/src/triggers/websocket/listener.rs b/backend/windmill-api/src/triggers/websocket/listener.rs index ea0cebb0e5..b1fd2dc56d 100644 --- a/backend/windmill-api/src/triggers/websocket/listener.rs +++ b/backend/windmill-api/src/triggers/websocket/listener.rs @@ -24,6 +24,7 @@ use tokio_tungstenite::{connect_async, tungstenite::Message, MaybeTlsStream, Web use windmill_common::{ error::{to_anyhow, Error, Result}, jobs::JobTriggerKind, + triggers::TriggerInfo, utils::report_critical_error, worker::to_raw_value, DB, @@ -94,7 +95,7 @@ impl ListeningTrigger { None, None, "".to_string(), // doesn't matter as no retry/error handler - Some(windmill_common::jobs::JobTriggerKind::Websocket), + TriggerInfo::new(Some(self.path.to_owned()), JobTriggerKind::Websocket), ) .await .map(|r| r.get().to_owned())?; @@ -333,7 +334,7 @@ impl Listener for WebsocketTrigger { trigger_config, script_path, error_handling, - suspend_number, + active_mode, .. } = listening_trigger; @@ -366,8 +367,9 @@ impl Listener for WebsocketTrigger { ), None => (None, None, None), }; - - if suspend_number.is_some() || extra.is_none() { + let active_mode = active_mode.unwrap_or(false); + let trigger = TriggerInfo::new(Some(path.to_owned()), Self::JOB_TRIGGER_KIND); + if active_mode || extra.is_none() { trigger_runnable( db, None, @@ -381,8 +383,8 @@ impl Listener for WebsocketTrigger { error_handler_args, format!("websocket_trigger/{}", listening_trigger.path), None, - *suspend_number, - Some(windmill_common::jobs::JobTriggerKind::Websocket), + active_mode, + trigger, ) .await?; } else if let Some(ReturnMessageChannels { send_message_tx, mut killpill_rx }) = extra { @@ -413,7 +415,7 @@ impl Listener for WebsocketTrigger { error_handler_path.as_deref(), error_handler_args.as_ref(), format!("websocket_trigger/{}", trigger_path), - Some(windmill_common::jobs::JobTriggerKind::Websocket), + trigger, ) => { if let Ok((result, success)) = result { if !success && !can_return_error_result { diff --git a/backend/windmill-api/src/triggers/websocket/mod.rs b/backend/windmill-api/src/triggers/websocket/mod.rs index e19c958423..6d35c77e8e 100644 --- a/backend/windmill-api/src/triggers/websocket/mod.rs +++ b/backend/windmill-api/src/triggers/websocket/mod.rs @@ -11,7 +11,8 @@ use serde_json::value::RawValue; use sqlx::{types::Json as SqlxJson, FromRow}; use windmill_common::{ error::{Error, Result}, - triggers::TriggerKind, + jobs::JobTriggerKind, + triggers::{TriggerInfo, TriggerKind}, worker::to_raw_value, DB, }; @@ -113,7 +114,7 @@ pub async fn get_url_from_runnable_value( None, None, "".to_string(), // doesn't matter as no retry/error handler - Some(windmill_common::jobs::JobTriggerKind::Websocket), + TriggerInfo::new(Some(path.to_owned()), JobTriggerKind::Websocket), ) .await?; diff --git a/backend/windmill-common/src/triggers.rs b/backend/windmill-common/src/triggers.rs index 43c4ae50fc..3c28db449c 100644 --- a/backend/windmill-common/src/triggers.rs +++ b/backend/windmill-common/src/triggers.rs @@ -4,6 +4,8 @@ use serde::{Deserialize, Serialize}; use std::fmt; use strum_macros::EnumIter; +use crate::jobs::JobTriggerKind; + #[derive(sqlx::Type, Serialize, Deserialize, Debug, Clone, Eq, PartialEq, Hash, EnumIter)] #[sqlx(type_name = "TRIGGER_KIND", rename_all = "snake_case")] #[serde(rename_all = "snake_case")] @@ -82,3 +84,14 @@ lazy_static! { pub static ref RUNNABLE_FORMAT_VERSION_CACHE: Cache = Cache::new(1000); } + +pub struct TriggerInfo { + pub trigger_path: Option, + pub trigger_kind: JobTriggerKind, +} + +impl TriggerInfo { + pub fn new(trigger_path: Option, trigger_kind: JobTriggerKind) -> TriggerInfo { + TriggerInfo { trigger_path, trigger_kind } + } +} diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs index b3e6e85cab..5c3bca618c 100644 --- a/backend/windmill-queue/src/jobs.rs +++ b/backend/windmill-queue/src/jobs.rs @@ -37,6 +37,7 @@ use windmill_common::auth::JobPerms; use windmill_common::bench::BenchmarkIter; use windmill_common::lockfiles::is_generated_from_raw_requirements; use windmill_common::jobs::{JobTriggerKind, EMAIL_ERROR_HANDLER_USER_EMAIL}; +use windmill_common::triggers::TriggerInfo; use windmill_common::utils::{configure_client, now_from_db}; use windmill_common::worker::{Connection, MIN_VERSION_SUPPORTS_DEBOUNCING, SCRIPT_TOKEN_EXPIRY}; @@ -471,7 +472,6 @@ pub async fn push_init_job<'c>( None, None, None, - None, ) .await?; inner_tx.commit().await?; @@ -531,7 +531,6 @@ pub async fn push_periodic_bash_job<'c>( None, None, None, - None, ) .await?; inner_tx.commit().await?; @@ -1378,7 +1377,6 @@ async fn restart_job_if_perpetual_inner( None, None, None, - None, ) .await?; tx.commit().await?; @@ -1930,7 +1928,6 @@ pub async fn push_error_handler<'a, 'c, T: Serialize + Send + Sync>( None, None, None, - None, ) .await?; tx.commit().await?; @@ -3871,7 +3868,7 @@ pub async fn push<'c, 'd>( token_prefix: Option<&str>, #[allow(unused_mut)] mut scheduled_for_o: Option>, - schedule_path: Option, + schedule_path: Option, //should be removed in favor of the trigger param below parent_job: Option, root_job: Option, flow_innermost_root_job: Option, @@ -3890,8 +3887,7 @@ pub async fn push<'c, 'd>( // If we know there is already a debounce job, we can use this for debouncing. // NOTE: Only works with dependency jobs triggered by relative imports debounce_job_id_o: Option, - suspend_number: Option, // If provided, job will be created as suspended with this number - trigger_kind: Option, + trigger: Option, ) -> Result<(Uuid, Transaction<'c, Postgres>), Error> { #[cfg(feature = "cloud")] if *CLOUD_HOSTED { @@ -5211,6 +5207,16 @@ pub async fn push<'c, 'd>( } } + let (trigger_path, trigger_kind) = trigger.map_or_else( + || { + schedule_path.map(|path| (Some(path), JobTriggerKind::Schedule)) + }, + |trigger| { + Some((trigger.trigger_path, trigger.trigger_kind)) + }, + ) + .unzip(); + if concurrent_limit.is_some() { insert_concurrency_key( workspace_id, @@ -5286,13 +5292,6 @@ pub async fn push<'c, 'd>( // tracing::error!("Could not insert job_perms for job {job_id}: {err:#}"); // } - let trigger_kind = trigger_kind.or_else(|| { - if schedule_path.is_some() { - Some(JobTriggerKind::Schedule) - } else { - None - } - }); let root_job = if root_job.is_some() && (root_job == flow_innermost_root_job.or(parent_job).or(Some(job_id))) @@ -5324,8 +5323,8 @@ pub async fn push<'c, 'd>( ON CONFLICT (job_id) DO UPDATE SET email = EXCLUDED.email, username = EXCLUDED.username, is_admin = EXCLUDED.is_admin, is_operator = EXCLUDED.is_operator, folders = EXCLUDED.folders, groups = EXCLUDED.groups, workspace_id = EXCLUDED.workspace_id, end_user_email = EXCLUDED.end_user_email ) INSERT INTO v2_job_queue - (workspace_id, id, running, scheduled_for, started_at, tag, priority, suspend) - VALUES ($2, $1, $28, COALESCE($29, now()), CASE WHEN $27 OR $40 THEN now() END, $30, $31, $42)", + (workspace_id, id, running, scheduled_for, started_at, tag, priority) + VALUES ($2, $1, $28, COALESCE($29, now()), CASE WHEN $27 OR $40 THEN now() END, $30, $31)", job_id, workspace_id, raw_code, @@ -5339,7 +5338,7 @@ pub async fn push<'c, 'd>( script_path.clone(), Json(args) as Json, job_kind.clone() as JobKind, - schedule_path, + trigger_path.flatten(), language as Option, same_worker, pre_run_error.map(|e| e.to_string()), @@ -5371,7 +5370,6 @@ pub async fn push<'c, 'd>( trigger_kind as Option, running, end_user_email, - suspend_number.unwrap_or(0), ) .execute(&mut *tx) .warn_after_seconds(1) diff --git a/backend/windmill-queue/src/schedule.rs b/backend/windmill-queue/src/schedule.rs index 36f3d02aeb..343f005916 100644 --- a/backend/windmill-queue/src/schedule.rs +++ b/backend/windmill-queue/src/schedule.rs @@ -21,8 +21,10 @@ use windmill_common::get_latest_flow_version_id_for_path; use windmill_common::get_latest_flow_version_info_for_path_from_version; use windmill_common::jobs::check_tag_available_for_workspace_internal; use windmill_common::jobs::JobPayload; +use windmill_common::jobs::JobTriggerKind; use windmill_common::schedule::schedule_to_user; use windmill_common::scripts::ScriptHash; +use windmill_common::triggers::TriggerInfo; use windmill_common::utils::WarnAfterExt; use windmill_common::worker::to_raw_value; use windmill_common::FlowVersionInfo; @@ -503,8 +505,10 @@ pub async fn push_scheduled_job<'c>( false, None, None, - None, - Some(windmill_common::jobs::JobTriggerKind::Schedule), + Some(TriggerInfo::new( + Some(schedule.path.clone()), + JobTriggerKind::Schedule, + )), ) .warn_after_seconds_with_sql(1, "push in push_scheduled_job".to_string()) .await?; diff --git a/backend/windmill-worker/src/ai/tools.rs b/backend/windmill-worker/src/ai/tools.rs index 9504c82330..f132e91485 100644 --- a/backend/windmill-worker/src/ai/tools.rs +++ b/backend/windmill-worker/src/ai/tools.rs @@ -471,7 +471,6 @@ async fn execute_windmill_tool( true, None, None, - None, None ) .await?; diff --git a/backend/windmill-worker/src/worker_flow.rs b/backend/windmill-worker/src/worker_flow.rs index 252adb6df3..b28cf06d4b 100644 --- a/backend/windmill-worker/src/worker_flow.rs +++ b/backend/windmill-worker/src/worker_flow.rs @@ -3303,7 +3303,6 @@ async fn push_next_flow_job( false, None, None, - None, None ) .warn_after_seconds(2) diff --git a/backend/windmill-worker/src/worker_lockfiles.rs b/backend/windmill-worker/src/worker_lockfiles.rs index bbc49f725f..59b9e41b52 100644 --- a/backend/windmill-worker/src/worker_lockfiles.rs +++ b/backend/windmill-worker/src/worker_lockfiles.rs @@ -670,7 +670,6 @@ pub async fn trigger_dependents_to_recompute_dependencies( false, None, debounce_job_id_o, - None, None ) .await?; diff --git a/frontend/src/lib/components/triggers/TriggerStateToggle.svelte b/frontend/src/lib/components/triggers/TriggerStateToggle.svelte index a7ca694b6e..d57dc0f9e5 100644 --- a/frontend/src/lib/components/triggers/TriggerStateToggle.svelte +++ b/frontend/src/lib/components/triggers/TriggerStateToggle.svelte @@ -1,26 +1,27 @@