diff --git a/backend/.sqlx/query-122090a0f89e5248a0a0f199ebd24582fdb302883aebd2da187ac0084e767ea3.json b/backend/.sqlx/query-122090a0f89e5248a0a0f199ebd24582fdb302883aebd2da187ac0084e767ea3.json deleted file mode 100644 index a329998c95..0000000000 --- a/backend/.sqlx/query-122090a0f89e5248a0a0f199ebd24582fdb302883aebd2da187ac0084e767ea3.json +++ /dev/null @@ -1,22 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "SELECT set_config('session.pgroups', $1, true)", - "describe": { - "columns": [ - { - "ordinal": 0, - "name": "set_config", - "type_info": "Text" - } - ], - "parameters": { - "Left": [ - "Text" - ] - }, - "nullable": [ - null - ] - }, - "hash": "122090a0f89e5248a0a0f199ebd24582fdb302883aebd2da187ac0084e767ea3" -} diff --git a/backend/.sqlx/query-18ddbb9eb4c0ec03e6170f8c70634546cddaf1029618ed14015fd7b0a7017441.json b/backend/.sqlx/query-18ddbb9eb4c0ec03e6170f8c70634546cddaf1029618ed14015fd7b0a7017441.json new file mode 100644 index 0000000000..c1258697c9 --- /dev/null +++ b/backend/.sqlx/query-18ddbb9eb4c0ec03e6170f8c70634546cddaf1029618ed14015fd7b0a7017441.json @@ -0,0 +1,27 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT set_session_context($1, $2, $3, $4, $5, $6)", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "set_session_context", + "type_info": "Void" + } + ], + "parameters": { + "Left": [ + "Bool", + "Text", + "Text", + "Text", + "Text", + "Text" + ] + }, + "nullable": [ + null + ] + }, + "hash": "18ddbb9eb4c0ec03e6170f8c70634546cddaf1029618ed14015fd7b0a7017441" +} diff --git a/backend/.sqlx/query-29fbc3a8c35845a997cd548ed417b9cc3c82d815d99b3d435adcfbb5a9246124.json b/backend/.sqlx/query-29fbc3a8c35845a997cd548ed417b9cc3c82d815d99b3d435adcfbb5a9246124.json deleted file mode 100644 index 72f3f1f469..0000000000 --- a/backend/.sqlx/query-29fbc3a8c35845a997cd548ed417b9cc3c82d815d99b3d435adcfbb5a9246124.json +++ /dev/null @@ -1,22 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "SELECT set_config('session.folders_read', $1, true)", - "describe": { - "columns": [ - { - "ordinal": 0, - "name": "set_config", - "type_info": "Text" - } - ], - "parameters": { - "Left": [ - "Text" - ] - }, - "nullable": [ - null - ] - }, - "hash": "29fbc3a8c35845a997cd548ed417b9cc3c82d815d99b3d435adcfbb5a9246124" -} diff --git a/backend/.sqlx/query-31df83e9eb6078e93ec5fe4168306caccb849db9e0f71d86da655b01c6a3e8d0.json b/backend/.sqlx/query-31df83e9eb6078e93ec5fe4168306caccb849db9e0f71d86da655b01c6a3e8d0.json deleted file mode 100644 index b96c05d674..0000000000 --- a/backend/.sqlx/query-31df83e9eb6078e93ec5fe4168306caccb849db9e0f71d86da655b01c6a3e8d0.json +++ /dev/null @@ -1,27 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "INSERT INTO v2_job_queue\n (workspace_id, id, running, scheduled_for, started_at, tag, priority)\n VALUES ($1, $2, $3, COALESCE($4, now()), CASE WHEN $3 THEN now() END, $5, $6) RETURNING id AS \"id!\"", - "describe": { - "columns": [ - { - "ordinal": 0, - "name": "id!", - "type_info": "Uuid" - } - ], - "parameters": { - "Left": [ - "Varchar", - "Uuid", - "Bool", - "Timestamptz", - "Varchar", - "Int2" - ] - }, - "nullable": [ - false - ] - }, - "hash": "31df83e9eb6078e93ec5fe4168306caccb849db9e0f71d86da655b01c6a3e8d0" -} diff --git a/backend/.sqlx/query-6199e8be5cb13db71108e555ea20f0b76dc38476670f9fc0667b057d2766d42e.json b/backend/.sqlx/query-6199e8be5cb13db71108e555ea20f0b76dc38476670f9fc0667b057d2766d42e.json deleted file mode 100644 index 2313dd087c..0000000000 --- a/backend/.sqlx/query-6199e8be5cb13db71108e555ea20f0b76dc38476670f9fc0667b057d2766d42e.json +++ /dev/null @@ -1,22 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "SELECT set_config('session.groups', $1, true)", - "describe": { - "columns": [ - { - "ordinal": 0, - "name": "set_config", - "type_info": "Text" - } - ], - "parameters": { - "Left": [ - "Text" - ] - }, - "nullable": [ - null - ] - }, - "hash": "6199e8be5cb13db71108e555ea20f0b76dc38476670f9fc0667b057d2766d42e" -} diff --git a/backend/.sqlx/query-6c90d5ea2a09b47b81fdab465062c25f8768b220e53bf469550a3a3697ab756a.json b/backend/.sqlx/query-6c90d5ea2a09b47b81fdab465062c25f8768b220e53bf469550a3a3697ab756a.json deleted file mode 100644 index fb5e174ced..0000000000 --- a/backend/.sqlx/query-6c90d5ea2a09b47b81fdab465062c25f8768b220e53bf469550a3a3697ab756a.json +++ /dev/null @@ -1,22 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "SELECT set_config('session.user', $1, true)", - "describe": { - "columns": [ - { - "ordinal": 0, - "name": "set_config", - "type_info": "Text" - } - ], - "parameters": { - "Left": [ - "Text" - ] - }, - "nullable": [ - null - ] - }, - "hash": "6c90d5ea2a09b47b81fdab465062c25f8768b220e53bf469550a3a3697ab756a" -} diff --git a/backend/.sqlx/query-7d78efab0a588f56a13a7b5251f0a72f5d341b053218e7aec83a834cf7ccc98f.json b/backend/.sqlx/query-7d78efab0a588f56a13a7b5251f0a72f5d341b053218e7aec83a834cf7ccc98f.json deleted file mode 100644 index 07c93d0d88..0000000000 --- a/backend/.sqlx/query-7d78efab0a588f56a13a7b5251f0a72f5d341b053218e7aec83a834cf7ccc98f.json +++ /dev/null @@ -1,21 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "INSERT INTO job_perms (job_id, email, username, is_admin, is_operator, folders, groups, workspace_id) \n values ($1, $2, $3, $4, $5, $6, $7, $8) \n ON CONFLICT (job_id) DO UPDATE SET email = $2, username = $3, is_admin = $4, is_operator = $5, folders = $6, groups = $7, workspace_id = $8", - "describe": { - "columns": [], - "parameters": { - "Left": [ - "Uuid", - "Varchar", - "Varchar", - "Bool", - "Bool", - "JsonbArray", - "TextArray", - "Varchar" - ] - }, - "nullable": [] - }, - "hash": "7d78efab0a588f56a13a7b5251f0a72f5d341b053218e7aec83a834cf7ccc98f" -} diff --git a/backend/.sqlx/query-9897705b7c265333559df757a725711a40924ec551a7589a64ee96f8aa7f6a21.json b/backend/.sqlx/query-9897705b7c265333559df757a725711a40924ec551a7589a64ee96f8aa7f6a21.json deleted file mode 100644 index abad579224..0000000000 --- a/backend/.sqlx/query-9897705b7c265333559df757a725711a40924ec551a7589a64ee96f8aa7f6a21.json +++ /dev/null @@ -1,22 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "SELECT set_config('session.folders_write', $1, true)", - "describe": { - "columns": [ - { - "ordinal": 0, - "name": "set_config", - "type_info": "Text" - } - ], - "parameters": { - "Left": [ - "Text" - ] - }, - "nullable": [ - null - ] - }, - "hash": "9897705b7c265333559df757a725711a40924ec551a7589a64ee96f8aa7f6a21" -} diff --git a/backend/.sqlx/query-ca8bf3dda133556ee7051d5436a5d76ae2223d59e9f321a8b6d0c27adc09f741.json b/backend/.sqlx/query-ca8bf3dda133556ee7051d5436a5d76ae2223d59e9f321a8b6d0c27adc09f741.json deleted file mode 100644 index b7a302213a..0000000000 --- a/backend/.sqlx/query-ca8bf3dda133556ee7051d5436a5d76ae2223d59e9f321a8b6d0c27adc09f741.json +++ /dev/null @@ -1,14 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "INSERT INTO v2_job_runtime (id, ping) VALUES ($1, null)", - "describe": { - "columns": [], - "parameters": { - "Left": [ - "Uuid" - ] - }, - "nullable": [] - }, - "hash": "ca8bf3dda133556ee7051d5436a5d76ae2223d59e9f321a8b6d0c27adc09f741" -} diff --git a/backend/.sqlx/query-29624682d687790dd199c4af759132d79fdb2982de111cb5fd43e3d9ecd0f15e.json b/backend/.sqlx/query-cccdcb7fe7968eadfc04d8957a8e98b2f2d92a6d7f687a9dd5a70edb3d5a63e6.json similarity index 51% rename from backend/.sqlx/query-29624682d687790dd199c4af759132d79fdb2982de111cb5fd43e3d9ecd0f15e.json rename to backend/.sqlx/query-cccdcb7fe7968eadfc04d8957a8e98b2f2d92a6d7f687a9dd5a70edb3d5a63e6.json index 3ea932c5ca..412b64c480 100644 --- a/backend/.sqlx/query-29624682d687790dd199c4af759132d79fdb2982de111cb5fd43e3d9ecd0f15e.json +++ b/backend/.sqlx/query-cccdcb7fe7968eadfc04d8957a8e98b2f2d92a6d7f687a9dd5a70edb3d5a63e6.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "INSERT INTO v2_job (id, workspace_id, raw_code, raw_lock, raw_flow, tag, parent_job,\n created_by, permissioned_as, runnable_id, runnable_path, args, kind, trigger,\n script_lang, same_worker, pre_run_error, permissioned_as_email, visible_to_owner,\n flow_innermost_root_job, concurrent_limit, concurrency_time_window_s, timeout, flow_step_id,\n cache_ttl, priority, trigger_kind, script_entrypoint_override, preprocessed)\n VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18,\n $19, $20, $21, $22, $23, $24, $25, $26,\n CASE WHEN $14::VARCHAR IS NOT NULL THEN 'schedule'::job_trigger_kind END,\n ($12::JSONB)->>'_ENTRYPOINT_OVERRIDE', $27)", + "query": "WITH inserted_job AS (\n INSERT INTO v2_job (id, workspace_id, raw_code, raw_lock, raw_flow, tag, parent_job,\n created_by, permissioned_as, runnable_id, runnable_path, args, kind, trigger,\n script_lang, same_worker, pre_run_error, permissioned_as_email, visible_to_owner,\n flow_innermost_root_job, concurrent_limit, concurrency_time_window_s, timeout, flow_step_id,\n cache_ttl, priority, trigger_kind, script_entrypoint_override, preprocessed)\n VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18,\n $19, $20, $21, $22, $23, $24, $25, $26,\n CASE WHEN $14::VARCHAR IS NOT NULL THEN 'schedule'::job_trigger_kind END,\n ($12::JSONB)->>'_ENTRYPOINT_OVERRIDE', $27)\n ),\n inserted_runtime AS (\n INSERT INTO v2_job_runtime (id, ping) VALUES ($1, null)\n ),\n inserted_job_perms AS (\n INSERT INTO job_perms (job_id, email, username, is_admin, is_operator, folders, groups, workspace_id) \n values ($1, $32, $33, $34, $35, $36, $37, $2) \n ON CONFLICT (job_id) DO UPDATE SET email = $32, username = $33, is_admin = $34, is_operator = $35, folders = $36, groups = $37, workspace_id = $2\n )\n INSERT INTO v2_job_queue\n (workspace_id, id, running, scheduled_for, started_at, tag, priority)\n VALUES ($2, $1, $28, COALESCE($29, now()), CASE WHEN $27 THEN now() END, $30, $31)", "describe": { "columns": [], "parameters": { @@ -86,10 +86,20 @@ "Varchar", "Int4", "Int2", - "Bool" + "Bool", + "Bool", + "Timestamptz", + "Varchar", + "Int2", + "Varchar", + "Varchar", + "Bool", + "Bool", + "JsonbArray", + "TextArray" ] }, "nullable": [] }, - "hash": "29624682d687790dd199c4af759132d79fdb2982de111cb5fd43e3d9ecd0f15e" + "hash": "cccdcb7fe7968eadfc04d8957a8e98b2f2d92a6d7f687a9dd5a70edb3d5a63e6" } diff --git a/backend/ee-repo-ref.txt b/backend/ee-repo-ref.txt index 6081d9fd35..91f337034b 100644 --- a/backend/ee-repo-ref.txt +++ b/backend/ee-repo-ref.txt @@ -1 +1 @@ -47869a5803b421a754173250fed79b63fe001ebd \ No newline at end of file +1475a504133766e005ec551004c2af2283663352 \ No newline at end of file diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs index 4fb7210048..9c7dd8a4fc 100644 --- a/backend/windmill-queue/src/jobs.rs +++ b/backend/windmill-queue/src/jobs.rs @@ -4190,93 +4190,6 @@ pub async fn push<'c, 'd>( _ => None, }); - sqlx::query!( - "INSERT INTO v2_job (id, workspace_id, raw_code, raw_lock, raw_flow, tag, parent_job, - created_by, permissioned_as, runnable_id, runnable_path, args, kind, trigger, - script_lang, same_worker, pre_run_error, permissioned_as_email, visible_to_owner, - flow_innermost_root_job, concurrent_limit, concurrency_time_window_s, timeout, flow_step_id, - cache_ttl, priority, trigger_kind, script_entrypoint_override, preprocessed) - VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, - $19, $20, $21, $22, $23, $24, $25, $26, - CASE WHEN $14::VARCHAR IS NOT NULL THEN 'schedule'::job_trigger_kind END, - ($12::JSONB)->>'_ENTRYPOINT_OVERRIDE', $27)", - job_id, - workspace_id, - raw_code, - raw_lock, - raw_flow as Option>, - tag, - parent_job, - user, - permissioned_as, - script_hash, - script_path.clone(), - Json(args) as Json, - job_kind.clone() as JobKind, - schedule_path, - language as Option, - same_worker, - pre_run_error.map(|e| e.to_string()), - email, - visible_to_owner, - root_job, - concurrent_limit, - if concurrent_limit.is_some() { - concurrency_time_window_s - } else { - None - }, - custom_timeout, - flow_step_id, - cache_ttl, - final_priority, - preprocessed, - ) - .execute(&mut *tx) - .warn_after_seconds(1) - .await?; - - tracing::debug!("Pushing job {job_id} with tag {tag}, schedule_path {schedule_path:?}, script_path: {script_path:?}, email {email}, workspace_id {workspace_id}"); - let uuid = sqlx::query_scalar!( - "INSERT INTO v2_job_queue - (workspace_id, id, running, scheduled_for, started_at, tag, priority) - VALUES ($1, $2, $3, COALESCE($4, now()), CASE WHEN $3 THEN now() END, $5, $6) \ - RETURNING id AS \"id!\"", - workspace_id, - job_id, - is_running, - scheduled_for_o, - tag, - final_priority, - ) - .fetch_one(&mut *tx) - .warn_after_seconds(1) - .await - .map_err(|e| Error::internal_err(format!("Could not insert into queue {job_id} with tag {tag}, schedule_path {schedule_path:?}, script_path: {script_path:?}, email {email}, workspace_id {workspace_id}: {e:#}")))?; - - sqlx::query!( - "INSERT INTO v2_job_runtime (id, ping) VALUES ($1, null)", - job_id - ) - .execute(&mut *tx) - .await?; - if let Some(flow_status) = flow_status { - sqlx::query!( - "INSERT INTO v2_job_status (id, flow_status) VALUES ($1, $2)", - job_id, - Json(flow_status) as Json, - ) - .execute(&mut *tx) - .await?; - } - - tracing::debug!("Pushed {job_id}"); - // TODO: technically the job isn't queued yet, as the transaction can be rolled back. Should be solved when moving these metrics to the queue abstraction. - #[cfg(feature = "prometheus")] - if METRICS_ENABLED.load(std::sync::atomic::Ordering::Relaxed) { - QUEUE_PUSH_COUNT.inc(); - } - let job_authed = match authed { Some(authed) if authed.email == email @@ -4309,21 +4222,130 @@ pub async fn push<'c, 'd>( .filter_map(|x| serde_json::to_value(x).ok()) .collect::>(); - if let Err(err) = sqlx::query!("INSERT INTO job_perms (job_id, email, username, is_admin, is_operator, folders, groups, workspace_id) - values ($1, $2, $3, $4, $5, $6, $7, $8) - ON CONFLICT (job_id) DO UPDATE SET email = $2, username = $3, is_admin = $4, is_operator = $5, folders = $6, groups = $7, workspace_id = $8", + // if let Err(err) = sqlx::query!("INSERT INTO job_perms (job_id, email, username, is_admin, is_operator, folders, groups, workspace_id) + // values ($1, $2, $3, $4, $5, $6, $7, $8) + // ON CONFLICT (job_id) DO UPDATE SET email = $2, username = $3, is_admin = $4, is_operator = $5, folders = $6, groups = $7, workspace_id = $8", + // job_id, + // job_authed.email, + // job_authed.username, + // job_authed.is_admin, + // job_authed.is_operator, + // folders.as_slice(), + // job_authed.groups.as_slice(), + // workspace_id, + // ).execute(&mut *tx).await { + // tracing::error!("Could not insert job_perms for job {job_id}: {err:#}"); + // } + + + sqlx::query!( + "WITH inserted_job AS ( + INSERT INTO v2_job (id, workspace_id, raw_code, raw_lock, raw_flow, tag, parent_job, + created_by, permissioned_as, runnable_id, runnable_path, args, kind, trigger, + script_lang, same_worker, pre_run_error, permissioned_as_email, visible_to_owner, + flow_innermost_root_job, concurrent_limit, concurrency_time_window_s, timeout, flow_step_id, + cache_ttl, priority, trigger_kind, script_entrypoint_override, preprocessed) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, + $19, $20, $21, $22, $23, $24, $25, $26, + CASE WHEN $14::VARCHAR IS NOT NULL THEN 'schedule'::job_trigger_kind END, + ($12::JSONB)->>'_ENTRYPOINT_OVERRIDE', $27) + ), + inserted_runtime AS ( + INSERT INTO v2_job_runtime (id, ping) VALUES ($1, null) + ), + inserted_job_perms AS ( + INSERT INTO job_perms (job_id, email, username, is_admin, is_operator, folders, groups, workspace_id) + values ($1, $32, $33, $34, $35, $36, $37, $2) + ON CONFLICT (job_id) DO UPDATE SET email = $32, username = $33, is_admin = $34, is_operator = $35, folders = $36, groups = $37, workspace_id = $2 + ) + INSERT INTO v2_job_queue + (workspace_id, id, running, scheduled_for, started_at, tag, priority) + VALUES ($2, $1, $28, COALESCE($29, now()), CASE WHEN $27 THEN now() END, $30, $31)", job_id, + workspace_id, + raw_code, + raw_lock, + raw_flow as Option>, + tag, + parent_job, + user, + permissioned_as, + script_hash, + script_path.clone(), + Json(args) as Json, + job_kind.clone() as JobKind, + schedule_path, + language as Option, + same_worker, + pre_run_error.map(|e| e.to_string()), + email, + visible_to_owner, + root_job, + concurrent_limit, + if concurrent_limit.is_some() { + concurrency_time_window_s + } else { + None + }, + custom_timeout, + flow_step_id, + cache_ttl, + final_priority, + preprocessed, + is_running, + scheduled_for_o, + tag, + final_priority, job_authed.email, job_authed.username, job_authed.is_admin, job_authed.is_operator, folders.as_slice(), job_authed.groups.as_slice(), - workspace_id, - ).execute(&mut *tx).await { - tracing::error!("Could not insert job_perms for job {job_id}: {err:#}"); + ) + .execute(&mut *tx) + .warn_after_seconds(1) + .await?; + +// tracing::debug!("Pushing job {job_id} with tag {tag}, schedule_path {schedule_path:?}, script_path: {script_path:?}, email {email}, workspace_id {workspace_id}"); +// let uuid = sqlx::query_scalar!( +// "INSERT INTO v2_job_queue +// (workspace_id, id, running, scheduled_for, started_at, tag, priority) +// VALUES ($1, $2, $3, COALESCE($4, now()), CASE WHEN $3 THEN now() END, $5, $6) \ +// RETURNING id AS \"id!\"", +// workspace_id, +// job_id, +// , +// ) +// .fetch_one(&mut *tx) +// .warn_after_seconds(1) +// .await +// .map_err(|e| Error::internal_err(format!("Could not insert into queue {job_id} with tag {tag}, schedule_path {schedule_path:?}, script_path: {script_path:?}, email {email}, workspace_id {workspace_id}: {e:#}")))?; + + // sqlx::query!( + // "INSERT INTO v2_job_runtime (id, ping) VALUES ($1, null)", + // job_id + // ) + // .execute(&mut *tx) + // .await?; + if let Some(flow_status) = flow_status { + sqlx::query!( + "INSERT INTO v2_job_status (id, flow_status) VALUES ($1, $2)", + job_id, + Json(flow_status) as Json, + ) + .execute(&mut *tx) + .await?; } - + + tracing::debug!("Pushed {job_id}"); + // TODO: technically the job isn't queued yet, as the transaction can be rolled back. Should be solved when moving these metrics to the queue abstraction. + #[cfg(feature = "prometheus")] + if METRICS_ENABLED.load(std::sync::atomic::Ordering::Relaxed) { + QUEUE_PUSH_COUNT.inc(); + } + + { let uuid_string = job_id.to_string(); @@ -4384,7 +4406,7 @@ pub async fn push<'c, 'd>( .await?; } - Ok((uuid, tx)) + Ok((job_id, tx)) } pub fn canceled_job_to_result(job: &MiniPulledJob) -> serde_json::Value {