starred/windmill
1
0
Fork 0
mirror of https://github.com/windmill-labs/windmill.git synced 2026-08-23 16:00:38 +00:00
Code Issues Packages Projects Releases Wiki Activity

big

Browse Source
This commit is contained in:
dieriba
2025-11-14 21:06:49 +01:00
parent bd3f3fd42d
commit a4f3b4fc9b
39 changed files with 371 additions and 662 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/migrations/20251106200000_add_queue_mode_to_triggers.down.sql
+16 -16
View File
@@ -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;
DROP COLUMN IF EXISTS active_mode;
backend/migrations/20251106200000_add_queue_mode_to_triggers.up.sql
+18 -18
View File
@@ -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;
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);
backend/src/monitor.rs
+1 -1
View File
@@ -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(())
}
backend/tests/common/mod.rs
-1
View File
@@ -208,7 +208,6 @@ impl RunJob {
false,
None,
debounce_job_id_o,
None,
None
)
.await
backend/windmill-api/openapi.yaml
+26 -29
View File
@@ -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
backend/windmill-api/src/apps.rs
+3 -6
View File
@@ -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?;
backend/windmill-api/src/concurrency_groups.rs
+1
View File
@@ -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: _,
backend/windmill-api/src/flows.rs
+2 -4
View File
@@ -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?;
backend/windmill-api/src/jobs.rs
+103 -103
View File
@@ -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<chrono::DateTime<chrono::Utc>>,
pub running: Option<bool>,
pub schedule_path: Option<String>,
pub trigger_path: Option<String>,
pub parent_job: Option<String>,
pub order_desc: Option<bool>,
pub job_kinds: Option<String>,
@@ -1840,6 +1843,7 @@ pub struct ListQueueQuery {
impl From<ListCompletedQuery> 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<UserDB>,
Path(w_id): Path<String>,
Json(request): Json<ResumeSuspendedJobsRequest>,
Json(request): Json<SuspendJobsRequest>,
) -> error::JsonResult<String> {
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<UserDB>,
Path(w_id): Path<String>,
Json(request): Json<CancelSuspendedJobsRequest>,
Json(request): Json<SuspendJobsRequest>,
) -> error::JsonResult<String> {
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<JobTriggerKind>,
trigger: Option<TriggerInfo>,
) -> error::Result<(Uuid, Option<String>)> {
#[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<JobTriggerKind>,
trigger: Option<TriggerInfo>,
) -> error::Result<(Uuid, Option<bool>)> {
#[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<bool>,
pub suspended: Option<bool>,
pub schedule_path: Option<String>,
pub trigger_path: Option<String>,
// filter by matching a subset of the args using base64 encoded json subset
pub args: Option<String>,
// filter by matching a subset of the result using base64 encoded json subset
backend/windmill-api/src/scripts.rs
+1 -2
View File
@@ -1022,8 +1022,7 @@ async fn create_script_internal<'c>(
false,
None,
None,
None,
None,
None
)
.await?;
Ok((hash, new_tx, None))
backend/windmill-api/src/settings.rs
+1 -1
View File
@@ -188,7 +188,7 @@ pub async fn test_license_key(
Json(TestKey { license_key }): Json<TestKey>,
) -> error::Result<String> {
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()))
backend/windmill-api/src/triggers/email/handler_oss.rs
-2
View File
@@ -46,7 +46,6 @@ impl TriggerCrud for EmailTrigger {
_authed: &ApiAuthed,
_w_id: &str,
_trigger: TriggerData<Self::TriggerConfigRequest>,
_suspend_number: Option<i32>,
) -> 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<Self::TriggerConfigRequest>,
_suspend_number: Option<i32>,
) -> Result<()> {
Err(Error::BadRequest(
"Email triggers are not available in open source version".to_string(),
backend/windmill-api/src/triggers/handler.rs
+3 -41
View File
@@ -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<Self::TriggerConfigRequest>,
suspend_number: Option<i32>,
) -> Result<()>;
async fn update_trigger(
@@ -111,7 +109,6 @@ pub trait TriggerCrud: Send + Sync + 'static {
workspace_id: &str,
path: &str,
trigger: TriggerData<Self::TriggerConfigRequest>,
suspend_number: Option<i32>,
) -> 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<T: TriggerCrud + 'static>() -> Router {
router
}
pub async fn get_suspend_number_for_inactive_mode(
db: &DB,
workspace_id: &str,
active_mode: Option<bool>,
) -> Result<Option<i32>> {
if let Some(false) = active_mode {
Ok(Some(
generate_unique_suspend_number(db, workspace_id).await?,
))
} else {
Ok(None)
}
}
async fn create_trigger<T: TriggerCrud>(
Extension(handler): Extension<Arc<T>>,
authed: ApiAuthed,
@@ -398,19 +381,9 @@ async fn create_trigger<T: TriggerCrud>(
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<T: TriggerCrud>(
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(
backend/windmill-api/src/triggers/http/handler.rs
+21 -33
View File
@@ -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<HttpConfigRequest>,
route_path_key: &str,
suspend_number: Option<i32>,
) -> 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<Self::TriggerConfigRequest>,
suspend_number: Option<i32>,
) -> 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<Self::TriggerConfigRequest>,
suspend_number: Option<i32>,
) -> 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()),
backend/windmill-api/src/triggers/http/mod.rs
+2 -2
View File
@@ -48,7 +48,7 @@ pub struct TriggerRoute {
error_handler_path: Option<String>,
error_handler_args: Option<sqlx::types::Json<HashMap<String, serde_json::Value>>>,
retry: Option<sqlx::types::Json<Retry>>,
suspend_number: Option<i32>,
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
backend/windmill-api/src/triggers/listener.rs
+6 -6
View File
@@ -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<T> {
pub script_path: String,
pub trigger_mode: bool,
pub error_handling: Option<TriggerErrorHandling>,
pub suspend_number: Option<i32>,
pub active_mode: Option<bool>,
}
impl<T> ListeningTrigger<T> {
backend/windmill-api/src/triggers/mod.rs
+15 -3
View File
@@ -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<Utc>,
pub extra_perms: Option<serde_json::Value>,
pub suspend_number: Option<i32>,
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> = Utc.with_ymd_and_hms(9999, 12, 31, 23, 59, 59).unwrap();
}
backend/windmill-api/src/triggers/mqtt/handler.rs
+4 -6
View File
@@ -73,7 +73,6 @@ impl TriggerCrud for MqttTrigger {
authed: &ApiAuthed,
w_id: &str,
trigger: TriggerData<Self::TriggerConfigRequest>,
suspend_number: Option<i32>,
) -> 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<Self::TriggerConfigRequest>,
suspend_number: Option<i32>,
) -> 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?;
backend/windmill-api/src/triggers/postgres/handler.rs
+4 -6
View File
@@ -71,7 +71,6 @@ impl TriggerCrud for PostgresTrigger {
authed: &ApiAuthed,
w_id: &str,
trigger: TriggerData<Self::TriggerConfigRequest>,
suspend_number: Option<i32>,
) -> 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<Self::TriggerConfigRequest>,
suspend_number: Option<i32>,
) -> 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?;
backend/windmill-api/src/triggers/trigger_helpers.rs
+59 -291
View File
@@ -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<HashMap<String, serde_json::Value>>>,
trigger_path: String,
job_id: Option<Uuid>,
trigger_kind: Option<JobTriggerKind>,
trigger: TriggerInfo,
active_mode: Option<bool>,
) -> Result<(Uuid, Option<bool>, Option<String>)> {
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<HashMap<String, serde_json::Value>>>,
trigger_path: String,
job_id: Option<Uuid>,
suspend_number: Option<i32>,
trigger_kind: Option<JobTriggerKind>,
active_mode: bool,
trigger: TriggerInfo,
) -> Result<axum::response::Response> {
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<HashMap<String, serde_json::Value>>>,
trigger_path: String,
trigger_kind: Option<JobTriggerKind>,
trigger: TriggerInfo,
) -> Result<axum::response::Response> {
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<HashMap<String, serde_json::Value>>>,
trigger_path: String,
trigger_kind: Option<JobTriggerKind>,
trigger: TriggerInfo,
) -> Result<(Box<RawValue>, 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<HashMap<String, serde_json::Value>>>,
trigger_path: String,
trigger_kind: Option<JobTriggerKind>,
trigger: TriggerInfo,
) -> Result<Box<RawValue>> {
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<HashMap<String, Box<RawValue>>>>,
trigger_path: String,
job_id: Option<Uuid>,
trigger_kind: Option<JobTriggerKind>,
trigger: TriggerInfo,
scheduled_for: Option<DateTime<Utc>>,
) -> Result<(Uuid, Option<bool>)> {
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<HashMap<String, Box<RawValue>>>>,
trigger_path: String,
job_id: Option<Uuid>,
trigger_kind: Option<JobTriggerKind>,
trigger: TriggerInfo,
scheduled_for: Option<DateTime<Utc>>,
) -> Result<(Uuid, Option<bool>)> {
#[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<UserDB>,
authed: ApiAuthed,
workspace_id: &str,
runnable_path: &str,
is_flow: bool,
args: PushArgsOwned,
retry: Option<&sqlx::types::Json<Retry>>,
error_handler_path: Option<&str>,
error_handler_args: Option<&sqlx::types::Json<HashMap<String, serde_json::Value>>>,
trigger_path: String,
job_id: Option<Uuid>,
suspend_number: i32,
trigger_kind: Option<JobTriggerKind>,
) -> Result<Uuid> {
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::<HashMap<String, Box<RawValue>>>()
});
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)
}
}
backend/windmill-api/src/triggers/websocket/handler.rs
+4 -6
View File
@@ -79,7 +79,6 @@ impl TriggerCrud for WebsocketTrigger {
authed: &ApiAuthed,
w_id: &str,
trigger: TriggerData<Self::TriggerConfigRequest>,
suspend_number: Option<i32>,
) -> 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<Self::TriggerConfigRequest>,
suspend_number: Option<i32>,
) -> 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?;
backend/windmill-api/src/triggers/websocket/listener.rs
+9 -7
View File
@@ -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<WebsocketConfig> {
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 {
backend/windmill-api/src/triggers/websocket/mod.rs
+3 -2
View File
@@ -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?;
backend/windmill-common/src/triggers.rs
+13
View File
@@ -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<RunnableFormatCacheKey, RunnableFormat> =
Cache::new(1000);
}
pub struct TriggerInfo {
pub trigger_path: Option<String>,
pub trigger_kind: JobTriggerKind,
}
impl TriggerInfo {
pub fn new(trigger_path: Option<String>, trigger_kind: JobTriggerKind) -> TriggerInfo {
TriggerInfo { trigger_path, trigger_kind }
}
}
backend/windmill-queue/src/jobs.rs
+16 -18
View File
@@ -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<chrono::DateTime<chrono::Utc>>,
schedule_path: Option<String>,
schedule_path: Option<String>, //should be removed in favor of the trigger param below
parent_job: Option<Uuid>,
root_job: Option<Uuid>,
flow_innermost_root_job: Option<Uuid>,
@@ -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<Uuid>,
suspend_number: Option<i32>, // If provided, job will be created as suspended with this number
trigger_kind: Option<JobTriggerKind>,
trigger: Option<TriggerInfo>,
) -> 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<PushArgs>,
job_kind.clone() as JobKind,
schedule_path,
trigger_path.flatten(),
language as Option<ScriptLang>,
same_worker,
pre_run_error.map(|e| e.to_string()),
@@ -5371,7 +5370,6 @@ pub async fn push<'c, 'd>(
trigger_kind as Option<JobTriggerKind>,
running,
end_user_email,
suspend_number.unwrap_or(0),
)
.execute(&mut *tx)
.warn_after_seconds(1)
backend/windmill-queue/src/schedule.rs
+6 -2
View File
@@ -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?;
backend/windmill-worker/src/ai/tools.rs
-1
View File
@@ -471,7 +471,6 @@ async fn execute_windmill_tool(
true,
None,
None,
None,
None
)
.await?;
backend/windmill-worker/src/worker_flow.rs
-1
View File
@@ -3303,7 +3303,6 @@ async fn push_next_flow_job(
false,
None,
None,
None,
None
)
.warn_after_seconds(2)
backend/windmill-worker/src/worker_lockfiles.rs
-1
View File
@@ -670,7 +670,6 @@ pub async fn trigger_dependents_to_recompute_dependencies(
false,
None,
debounce_job_id_o,
None,
None
)
.await?;
frontend/src/lib/components/triggers/TriggerStateToggle.svelte
+18 -15
View File
@@ -1,26 +1,27 @@
<script lang="ts">
import Modal2 from '../common/modal/Modal2.svelte'
import Toggle from '../Toggle.svelte'
import { JobService } from '$lib/gen/services.gen'
import type { QueuedJob } from '$lib/gen/types.gen'
import Button from '../common/button/Button.svelte'
import { workspaceStore } from '$lib/stores'
import RunRow from '../runs/RunRow.svelte'
import '../runs/runs-grid.css'
import { JobService } from '$lib/gen'
import { sendUserToast } from '$lib/toast'
import { type JobTriggerType } from './utils'
type Props = {
active_mode: boolean
suspendNumber?: number
triggerPath?: string
triggerPath: string
jobTriggerKind: JobTriggerType
}
let { active_mode = $bindable(), suspendNumber }: Props = $props()
let { active_mode = $bindable(), jobTriggerKind, triggerPath }: Props = $props()
let wasInInactiveMode = $state(!active_mode)
let shouldShowModal = $state(false)
$effect(() => {
if (active_mode && wasInInactiveMode && suspendNumber !== undefined) {
if (active_mode && wasInInactiveMode) {
shouldShowModal = true
} else if (!active_mode) {
wasInInactiveMode = true
@@ -34,25 +35,25 @@
let workspace = $workspaceStore!
let containerWidth = $state(1000)
$effect(() => {
if (shouldShowModal && suspendNumber !== undefined) {
if (shouldShowModal) {
fetchQueuedJobs()
}
})
async function fetchQueuedJobs() {
if (!suspendNumber) return
loading = true
error = null
try {
const allSuspendedJobs = await JobService.listQueue({
workspace,
suspended: true,
triggerKind: jobTriggerKind,
triggerPath,
running: false,
perPage: 100
})
queuedJobs = allSuspendedJobs.filter((job) => job.suspend === suspendNumber)
queuedJobs = allSuspendedJobs
} catch (e) {
error = `Failed to fetch queued jobs: ${e}`
console.error('Failed to fetch queued jobs:', e)
@@ -62,7 +63,7 @@
}
async function runAllJobs() {
if (queuedJobs.length === 0 || !suspendNumber) return
if (queuedJobs.length === 0) return
processingAction = true
error = null
@@ -71,7 +72,8 @@
const resumedJobs = await JobService.resumeSuspendedJobs({
workspace,
requestBody: {
suspend_number: suspendNumber
trigger_kind: jobTriggerKind,
trigger_path: triggerPath
}
})
sendUserToast(resumedJobs)
@@ -86,7 +88,7 @@
}
async function discardAllJobs() {
if (queuedJobs.length === 0 || !suspendNumber) return
if (queuedJobs.length === 0) return
processingAction = true
error = null
@@ -95,11 +97,12 @@
await JobService.cancelSuspendedJobs({
workspace,
requestBody: {
suspend_number: suspendNumber
trigger_kind: jobTriggerKind,
trigger_path: triggerPath
}
})
sendUserToast(`Successfully canceled all jobs with suspend number: ${suspendNumber}`)
sendUserToast(`Successfully canceled all jobs`)
} catch (e) {
error = `Failed to discard jobs: ${e}`
console.error('Failed to discard jobs:', e)
frontend/src/lib/components/triggers/email/EmailTriggerEditorInner.svelte
-6
View File
@@ -24,7 +24,6 @@
import Tabs from '$lib/components/common/tabs/Tabs.svelte'
import Tab from '$lib/components/common/tabs/Tab.svelte'
import TriggerRetriesAndErrorHandler from '../TriggerRetriesAndErrorHandler.svelte'
import TriggerStateToggle from '../TriggerStateToggle.svelte'
import { saveEmailTriggerFromCfg } from './utils'
let {
@@ -67,7 +66,6 @@
let error_handler_args: Record<string, any> = $state({})
let retry: Retry | undefined = $state()
let active_mode = $state(true)
let suspend_number: number | undefined = $state()
// Component references
let drawer = $state<Drawer | undefined>(undefined)
let initialConfig: NewEmailTrigger | undefined = undefined
@@ -164,8 +162,6 @@
error_handler_args = cfg?.error_handler_args ?? {}
retry = cfg?.retry
errorHandlerSelected = getHandlerType(error_handler_path ?? '')
active_mode = cfg?.suspend_number ? false : true
suspend_number = cfg?.suspend_number
}
async function loadTrigger(defaultConfig?: Partial<EmailTrigger>): Promise<void> {
@@ -302,8 +298,6 @@
</Section>
{/if}
<TriggerStateToggle suspendNumber={suspend_number} bind:active_mode />
<EmailTriggerEditorConfigSection
initialTriggerPath={initialPath}
bind:local_part
frontend/src/lib/components/triggers/gcp/GcpTriggerEditorInner.svelte
+2 -4
View File
@@ -62,7 +62,6 @@
let error_handler_args: Record<string, any> = $state({})
let retry: Retry | undefined = $state()
let active_mode = $state(true)
let suspend_number: number | undefined = $state(undefined)
let {
useDrawer = true,
description = undefined,
@@ -191,8 +190,7 @@
can_write = canWrite(cfg?.path, cfg?.extra_perms, $userStore)
error_handler_path = cfg?.error_handler_path
error_handler_args = cfg?.error_handler_args ?? {}
active_mode = cfg?.suspend_number ? false : true
suspend_number = cfg?.suspend_number
active_mode = cfg?.active_mode
retry = cfg?.retry
auto_acknowledge_msg = cfg?.auto_acknowledge_msg ?? true
ack_deadline = cfg?.ack_deadline
@@ -398,7 +396,7 @@
</div>
</Section>
<TriggerStateToggle suspendNumber={suspend_number} bind:active_mode />
<TriggerStateToggle triggerPath={path} jobTriggerKind={'gcp'} bind:active_mode />
{/if}
<GcpTriggerEditorConfigSection
frontend/src/lib/components/triggers/http/RouteEditorInner.svelte
+2 -4
View File
@@ -112,7 +112,6 @@
let optionTabSelected: 'request_options' | 'error_handler' | 'retries' = $state('request_options')
let errorHandlerSelected: ErrorHandler = $state('slack')
let active_mode = $state(true)
let suspend_number: number | undefined = $state(undefined)
const isAdmin = $derived($userStore?.is_admin || $userStore?.is_super_admin)
const routeConfig = $derived.by(getRouteConfig)
const captureConfig = $derived.by(isEditor ? getCaptureConfig : () => ({}))
@@ -291,8 +290,7 @@
error_handler_args = cfg?.error_handler_args ?? {}
retry = cfg?.retry
errorHandlerSelected = getHandlerType(error_handler_path ?? '')
active_mode = cfg?.suspend_number ? false : true
suspend_number = cfg?.suspend_number
active_mode = cfg?.active_mode ?? true
}
async function loadTrigger(defaultConfig?: Partial<HttpTrigger>): Promise<void> {
@@ -597,7 +595,7 @@
</div>
</Section>
<TriggerStateToggle suspendNumber={suspend_number} bind:active_mode />
<TriggerStateToggle triggerPath={path} jobTriggerKind={'http'} bind:active_mode />
{/if}
<RouteEditorConfigSection
frontend/src/lib/components/triggers/kafka/KafkaTriggerEditorInner.svelte
+2 -4
View File
@@ -84,7 +84,6 @@
let error_handler_args: Record<string, any> = $state({})
let retry: Retry | undefined = $state()
let active_mode = $state(true)
let suspend_number: number | undefined = $state(undefined)
const isValid = $derived(
!!kafkaResourcePath &&
@@ -191,8 +190,7 @@
error_handler_args = cfg?.error_handler_args ?? {}
retry = cfg?.retry
errorHandlerSelected = getHandlerType(error_handler_path ?? '')
active_mode = cfg?.suspend_number ? false : true
suspend_number = cfg?.suspend_number
active_mode = cfg?.active_mode
}
async function loadTrigger(defaultConfig?: Record<string, any>): Promise<void> {
@@ -389,7 +387,7 @@
</div>
</Section>
<TriggerStateToggle suspendNumber={suspend_number} bind:active_mode />
<TriggerStateToggle triggerPath={path} jobTriggerKind={'kafka'} bind:active_mode />
{/if}
<KafkaTriggersConfigSection
frontend/src/lib/components/triggers/mqtt/MqttTriggerEditorInner.svelte
+2 -4
View File
@@ -98,7 +98,6 @@
let error_handler_args: Record<string, any> = $state({})
let retry: Retry | undefined = $state()
let active_mode = $state(true)
let suspend_number: number | undefined = $state(undefined)
let optionTabSelected: 'connection_options' | 'error_handler' | 'retries' =
$state('connection_options')
@@ -205,8 +204,7 @@
error_handler_args = cfg?.error_handler_args ?? {}
retry = cfg?.retry
errorHandlerSelected = getHandlerType(error_handler_path ?? '')
active_mode = cfg?.suspend_number ? false : true
suspend_number = cfg?.suspend_number
active_mode = cfg?.active_mode
activateV5Options.topic_alias_maximum = Boolean(v5_config.topic_alias_maximum)
activateV5Options.session_expiry_interval = Boolean(v5_config.session_expiry_interval)
} catch (error) {
@@ -423,7 +421,7 @@
</Section>
{/if}
<TriggerStateToggle suspendNumber={suspend_number} bind:active_mode />
<TriggerStateToggle triggerPath={path} jobTriggerKind={'mqtt'} bind:active_mode />
<MqttEditorConfigSection
bind:mqtt_resource_path
frontend/src/lib/components/triggers/nats/NatsTriggerEditorInner.svelte
+2 -4
View File
@@ -93,7 +93,6 @@
let error_handler_args: Record<string, any> = $state({})
let retry: Retry | undefined = $state()
let active_mode = $state(true)
let suspend_number: number | undefined = $state(undefined)
const saveDisabled = $derived(
pathError != '' || emptyString(script_path) || !can_write || !isValid
@@ -193,8 +192,7 @@
error_handler_args = cfg?.error_handler_args ?? {}
retry = cfg?.retry
errorHandlerSelected = getHandlerType(error_handler_path ?? '')
active_mode = cfg?.suspend_number ? false : true
suspend_number = cfg?.suspend_number
active_mode = cfg?.active_mode
}
async function loadTrigger(defaultConfig?: Record<string, any>): Promise<void> {
@@ -404,7 +402,7 @@
</Section>
{/if}
<TriggerStateToggle suspendNumber={suspend_number} bind:active_mode />
<TriggerStateToggle triggerPath={path} jobTriggerKind={'nats'} bind:active_mode />
<NatsTriggersConfigSection
{path}
frontend/src/lib/components/triggers/postgres/PostgresTriggerEditorInner.svelte
+2 -4
View File
@@ -118,7 +118,6 @@
let error_handler_args: Record<string, any> = $state({})
let retry: Retry | undefined = $state()
let active_mode = $state(true)
let suspend_number: number | undefined = $state(undefined)
const errorMessage = $derived.by(() => {
if (relations && relations.length > 0) {
@@ -324,8 +323,7 @@
error_handler_args = cfg?.error_handler_args ?? {}
retry = cfg?.retry
errorHandlerSelected = getHandlerType(error_handler_path ?? '')
active_mode = cfg?.suspend_number ? false : true
suspend_number = cfg?.suspend_number
active_mode = cfg?.active_mode
}
async function loadTrigger(defaultConfig?: Record<string, any>): Promise<void> {
@@ -582,7 +580,7 @@
</Section>
{/if}
<TriggerStateToggle suspendNumber={suspend_number} bind:active_mode />
<TriggerStateToggle triggerPath={path} jobTriggerKind={'postgres'} bind:active_mode />
<Section label="Database">
{#snippet badge()}
{#if isEditor}
frontend/src/lib/components/triggers/sqs/SqsTriggerEditorInner.svelte
+2 -4
View File
@@ -90,7 +90,6 @@
let error_handler_args: Record<string, any> = $state({})
let retry: Retry | undefined = $state()
let active_mode = $state(true)
let suspend_number: number | undefined = $state(undefined)
const sqsConfig = $derived.by(getSaveCfg)
const captureConfig = $derived.by(getCaptureConfig)
@@ -179,8 +178,7 @@
error_handler_args = cfg?.error_handler_args ?? {}
retry = cfg?.retry
errorHandlerSelected = getHandlerType(error_handler_path ?? '')
active_mode = cfg?.suspend_number ? false : true
suspend_number = cfg?.suspend_number
active_mode = cfg?.active_mode
} catch (error) {
sendUserToast(`Could not load SQS trigger config: ${error.body}`, true)
}
@@ -390,7 +388,7 @@
</div>
</Section>
<TriggerStateToggle suspendNumber={suspend_number} bind:active_mode />
<TriggerStateToggle triggerPath={path} jobTriggerKind={'sqs'} bind:active_mode />
{/if}
<SqsTriggerEditorConfigSection
frontend/src/lib/components/triggers/websocket/WebsocketTriggerEditorInner.svelte
+2 -4
View File
@@ -107,7 +107,6 @@
let error_handler_args: Record<string, any> = $state({})
let retry: Retry | undefined = $state()
let active_mode = $state(true)
let suspend_number: number | undefined = $state(undefined)
const websocketCfg = $derived.by(getSaveCfg)
const captureConfig = $derived.by(isEditor ? getCaptureConfig : () => ({}))
@@ -222,8 +221,7 @@
error_handler_args = cfg?.error_handler_args ?? {}
retry = cfg?.retry
errorHandlerSelected = getHandlerType(error_handler_path ?? '')
active_mode = cfg?.suspend_number ? false : true
suspend_number = cfg?.suspend_number
active_mode = cfg?.active_mode
}
function getSaveCfg() {
@@ -496,7 +494,7 @@
/>
</Section>
<TriggerStateToggle suspendNumber={suspend_number} bind:active_mode />
<TriggerStateToggle triggerPath={path} jobTriggerKind={'websocket'} bind:active_mode />
<WebsocketEditorConfigSection
bind:url