From cedfd183d9a0bb00fbaa01e138df84250aa18fdf Mon Sep 17 00:00:00 2001 From: Ramtin Mesgari <26694963+iamramtin@users.noreply.github.com> Date: Tue, 4 Nov 2025 20:08:02 +0700 Subject: [PATCH] refactor: remove legacy database views v2_as_queue and v2_as_completed_job (#6689) * refactor: remove legacy database views v2_as_queue and v2_as_completed_job Signed-off-by: Ramtin Mesgari <26694963+iamramtin@users.noreply.github.com> * fix tests * fix jobs.rs * end * fix * improvement * improvement --------- Signed-off-by: Ramtin Mesgari <26694963+iamramtin@users.noreply.github.com> Co-authored-by: Ruben Fiszel --- ...af8688a6d1643747be3ec4f784c3029a59e52.json | 22 ++ ...2849fe6bc668654ffcbfbc22a02027280739.json} | 4 +- ...1e9366d1f6d3d2aa485e0f50c6f2d85693dd.json} | 6 +- ...84846f74d468f0073f02f81122895e86c364.json} | 4 +- ...bbd75d6d5bb77e6a92bf187d978a059d7af4a.json | 20 ++ ...f0563e3f98fff1f100616c33e8dc95fbff99.json} | 6 +- ...7abdad3198d3340ec7c04ed671baff0a4d0b.json} | 6 +- ...52ff982821ed7b24574cdd09a06eac0d628b.json} | 4 +- ...a2611dab30faff2891fc3a7f00ee8c120950.json} | 6 +- ...44744910a17af23f8201d853f0a4f8404fc73.json | 29 +++ ...64c72fe4b3cf2497644d514c5c00d6d71bf3.json} | 8 +- ...d5d3309a31b195b3147fa32f2ca8a6c9c90e.json} | 6 +- ...92ef0ad30e39eec59c27bae8cb0622062c8fb.json | 20 -- ...84028ce8a077308979a5ba8ef252e19aa825.json} | 4 +- ...13776870612e8faa8d8fa577f8e7b7309f76.json} | 10 +- ...2a6d9a6393c6bb1783ab8903d87dd099e236b.json | 58 +++++ ...aa01a048fa4f9281327cf5b78111178424b43.json | 58 ----- ...81a4f2edad152581950fdd80d758a0d242c17.json | 28 -- ...e5a3d1a91ea02de7608676575e1c03023ed71.json | 29 --- ...506dce6eb4af3fb32e6a4c6e76084873a539a.json | 216 ++++++++++++++++ ...dbb17c739415850e7c67f8575fb983c295fe.json} | 4 +- ...84b12dd961ba732772d8f6c59d18fe05f285d.json | 24 ++ ...ca3516ac5c1d32a98946196b1a42e3f103efd.json | 239 ++++++++++++++++++ ...1f78d7d61fbf83dda9b39bb22e0f9584d221b.json | 24 -- ...4a8b793c4ae618a04220b3609c8c3c168f8fd.json | 28 ++ ...7fd4280466eec443aff4f1e8c1ab810cca7e.json} | 4 +- backend/src/monitor.rs | 75 +++--- backend/tests/common/mod.rs | 12 +- backend/tests/worker.rs | 8 +- backend/windmill-api/src/approvals.rs | 44 ++-- backend/windmill-api/src/apps.rs | 14 +- backend/windmill-api/src/jobs.rs | 156 +++--------- backend/windmill-api/src/users.rs | 15 +- backend/windmill-queue/src/jobs.rs | 205 ++++++++++----- .../windmill-worker/src/result_processor.rs | 8 +- 35 files changed, 933 insertions(+), 471 deletions(-) create mode 100644 backend/.sqlx/query-0186c1058f147e012b8120c342caf8688a6d1643747be3ec4f784c3029a59e52.json rename backend/.sqlx/{query-6ab112fa42a9ae332bfa30427b70fa742351c5c180ac3de106df54f7badb494c.json => query-1368ccd2c15a75690041a6c87d4a2849fe6bc668654ffcbfbc22a02027280739.json} (63%) rename backend/.sqlx/{query-acc0b67c8e768b524b5cfb309e4307daeb0c095e07063c57b26ab94211bf6359.json => query-24d302b8215d49a289bedd14a5791e9366d1f6d3d2aa485e0f50c6f2d85693dd.json} (52%) rename backend/.sqlx/{query-f3571e1d2b57011e5f6a38725eb42d909456d28a98563923cca43e760862e5e0.json => query-25cba74bec5959e6752265cd7b6f84846f74d468f0073f02f81122895e86c364.json} (50%) create mode 100644 backend/.sqlx/query-2729c73b53605908e2fa26b4e4bbbd75d6d5bb77e6a92bf187d978a059d7af4a.json rename backend/.sqlx/{query-2bfa1ffb3d5869fc3038049ba77890203332e398c865c47aaf019dd5721d59f9.json => query-29785f22ee7092e8cebeae3757caf0563e3f98fff1f100616c33e8dc95fbff99.json} (58%) rename backend/.sqlx/{query-ca5d9a9d8d18da970c7fd6eab41ecbb3a5c7c29803e4c38b8a0b2ca3790e52f9.json => query-35061719d01929a7146c80de4b637abdad3198d3340ec7c04ed671baff0a4d0b.json} (54%) rename backend/.sqlx/{query-280a361076d1c6317610765960f543252891c53351bdc98da66cc30ffc895866.json => query-371d652fbb1d34d56f645c75d12852ff982821ed7b24574cdd09a06eac0d628b.json} (64%) rename backend/.sqlx/{query-2ea447f9e644554d415367b91042687ee8690d475b8ed31c48e31180689a278f.json => query-37285436c16684449b33810d97d0a2611dab30faff2891fc3a7f00ee8c120950.json} (55%) create mode 100644 backend/.sqlx/query-4545e1e0953cde730272353e78044744910a17af23f8201d853f0a4f8404fc73.json rename backend/.sqlx/{query-7af1cf089022fc1c3597b270b69aa669a153ab0c0bb2807cd4f7fd405afa6f69.json => query-48a41e1ad0ad8fc51624af4de34964c72fe4b3cf2497644d514c5c00d6d71bf3.json} (73%) rename backend/.sqlx/{query-2456fc71fc7a0758a4c1fbe77d72fbac2fead0e1bff4e909fd7fb1a41bc35d8f.json => query-49f7e7481019d27b44d5c8a167f9d5d3309a31b195b3147fa32f2ca8a6c9c90e.json} (65%) delete mode 100644 backend/.sqlx/query-519f4f76649947f036a2129c11e92ef0ad30e39eec59c27bae8cb0622062c8fb.json rename backend/.sqlx/{query-0cb0e912bc942af2b1ef784455f3f073a79e300f3dd48f14122d1782eee663cd.json => query-57bd445bfdc667089f847acb9dc484028ce8a077308979a5ba8ef252e19aa825.json} (50%) rename backend/.sqlx/{query-d4d83d8177144c91aa489b5a42a45c83f8b069a52f681f14afb4931ac77baf45.json => query-6b9ea0059ad1037a77f67b70d5de13776870612e8faa8d8fa577f8e7b7309f76.json} (57%) create mode 100644 backend/.sqlx/query-70a6880960d17218bc5bf05287e2a6d9a6393c6bb1783ab8903d87dd099e236b.json delete mode 100644 backend/.sqlx/query-72956f508f66312807738399b57aa01a048fa4f9281327cf5b78111178424b43.json delete mode 100644 backend/.sqlx/query-89940a53f29b173b6a8717f057a81a4f2edad152581950fdd80d758a0d242c17.json delete mode 100644 backend/.sqlx/query-90fbb9430ab03ce3aadd95cc263e5a3d1a91ea02de7608676575e1c03023ed71.json create mode 100644 backend/.sqlx/query-a001b4254e0f1ba8a87776e32f9506dce6eb4af3fb32e6a4c6e76084873a539a.json rename backend/.sqlx/{query-11db65c493990f6935103033b2fbb0c08ae6d91b05b2f3f7c89a990d1d5a5f8a.json => query-a3d18ae5e5125940ae0d6af315e2dbb17c739415850e7c67f8575fb983c295fe.json} (55%) create mode 100644 backend/.sqlx/query-a72e7fb55d7268fe1ea40015a4a84b12dd961ba732772d8f6c59d18fe05f285d.json create mode 100644 backend/.sqlx/query-d48c9a748746080e9b6cf0366b2ca3516ac5c1d32a98946196b1a42e3f103efd.json delete mode 100644 backend/.sqlx/query-ec9980f80a8bfa4b09225035e8a1f78d7d61fbf83dda9b39bb22e0f9584d221b.json create mode 100644 backend/.sqlx/query-edd57b3d59ddc21b99212b5fabd4a8b793c4ae618a04220b3609c8c3c168f8fd.json rename backend/.sqlx/{query-3f08ffbb5c71b873a9e164ecb0b10fffb37599f2a703885ee723cb9290fed13e.json => query-fbd38c9c4f4ecba2d3e3d79433327fd4280466eec443aff4f1e8c1ab810cca7e.json} (52%) diff --git a/backend/.sqlx/query-0186c1058f147e012b8120c342caf8688a6d1643747be3ec4f784c3029a59e52.json b/backend/.sqlx/query-0186c1058f147e012b8120c342caf8688a6d1643747be3ec4f784c3029a59e52.json new file mode 100644 index 0000000000..1880104006 --- /dev/null +++ b/backend/.sqlx/query-0186c1058f147e012b8120c342caf8688a6d1643747be3ec4f784c3029a59e52.json @@ -0,0 +1,22 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT j.id\n FROM v2_job_queue q JOIN v2_job j USING (id) LEFT JOIN v2_job_runtime r USING (id) LEFT JOIN v2_job_status s USING (id)\n WHERE r.ping < now() - ($1 || ' seconds')::interval\n AND q.running = true AND j.kind NOT IN ('flow', 'flowpreview', 'flownode', 'singlestepflow') AND j.same_worker = false", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "id", + "type_info": "Uuid" + } + ], + "parameters": { + "Left": [ + "Text" + ] + }, + "nullable": [ + false + ] + }, + "hash": "0186c1058f147e012b8120c342caf8688a6d1643747be3ec4f784c3029a59e52" +} diff --git a/backend/.sqlx/query-6ab112fa42a9ae332bfa30427b70fa742351c5c180ac3de106df54f7badb494c.json b/backend/.sqlx/query-1368ccd2c15a75690041a6c87d4a2849fe6bc668654ffcbfbc22a02027280739.json similarity index 63% rename from backend/.sqlx/query-6ab112fa42a9ae332bfa30427b70fa742351c5c180ac3de106df54f7badb494c.json rename to backend/.sqlx/query-1368ccd2c15a75690041a6c87d4a2849fe6bc668654ffcbfbc22a02027280739.json index ef26a8453e..4dac49a78c 100644 --- a/backend/.sqlx/query-6ab112fa42a9ae332bfa30427b70fa742351c5c180ac3de106df54f7badb494c.json +++ b/backend/.sqlx/query-1368ccd2c15a75690041a6c87d4a2849fe6bc668654ffcbfbc22a02027280739.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "SELECT COUNT(id) FROM v2_as_queue WHERE email = $1", + "query": "SELECT COUNT(id) FROM v2_job WHERE permissioned_as_email = $1", "describe": { "columns": [ { @@ -18,5 +18,5 @@ null ] }, - "hash": "6ab112fa42a9ae332bfa30427b70fa742351c5c180ac3de106df54f7badb494c" + "hash": "1368ccd2c15a75690041a6c87d4a2849fe6bc668654ffcbfbc22a02027280739" } diff --git a/backend/.sqlx/query-acc0b67c8e768b524b5cfb309e4307daeb0c095e07063c57b26ab94211bf6359.json b/backend/.sqlx/query-24d302b8215d49a289bedd14a5791e9366d1f6d3d2aa485e0f50c6f2d85693dd.json similarity index 52% rename from backend/.sqlx/query-acc0b67c8e768b524b5cfb309e4307daeb0c095e07063c57b26ab94211bf6359.json rename to backend/.sqlx/query-24d302b8215d49a289bedd14a5791e9366d1f6d3d2aa485e0f50c6f2d85693dd.json index 58f693df33..7acbf9deec 100644 --- a/backend/.sqlx/query-acc0b67c8e768b524b5cfb309e4307daeb0c095e07063c57b26ab94211bf6359.json +++ b/backend/.sqlx/query-24d302b8215d49a289bedd14a5791e9366d1f6d3d2aa485e0f50c6f2d85693dd.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "SELECT id AS \"id!\" FROM v2_as_queue WHERE id = ANY($1) AND schedule_path IS NULL AND ($2::text[] IS NULL OR tag = ANY($2))", + "query": "SELECT j.id AS \"id!\" FROM v2_job j WHERE j.id = ANY($1) AND j.trigger_kind != 'schedule'::job_trigger_kind AND ($2::text[] IS NULL OR j.tag = ANY($2))", "describe": { "columns": [ { @@ -16,8 +16,8 @@ ] }, "nullable": [ - true + false ] }, - "hash": "acc0b67c8e768b524b5cfb309e4307daeb0c095e07063c57b26ab94211bf6359" + "hash": "24d302b8215d49a289bedd14a5791e9366d1f6d3d2aa485e0f50c6f2d85693dd" } diff --git a/backend/.sqlx/query-f3571e1d2b57011e5f6a38725eb42d909456d28a98563923cca43e760862e5e0.json b/backend/.sqlx/query-25cba74bec5959e6752265cd7b6f84846f74d468f0073f02f81122895e86c364.json similarity index 50% rename from backend/.sqlx/query-f3571e1d2b57011e5f6a38725eb42d909456d28a98563923cca43e760862e5e0.json rename to backend/.sqlx/query-25cba74bec5959e6752265cd7b6f84846f74d468f0073f02f81122895e86c364.json index 804f35777e..7c07d169b8 100644 --- a/backend/.sqlx/query-f3571e1d2b57011e5f6a38725eb42d909456d28a98563923cca43e760862e5e0.json +++ b/backend/.sqlx/query-25cba74bec5959e6752265cd7b6f84846f74d468f0073f02f81122895e86c364.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "\n SELECT flow_status->'user_states'->$1\n FROM v2_as_queue\n WHERE id = $2 AND workspace_id = $3\n ", + "query": "\n SELECT COALESCE(s.flow_status, s.workflow_as_code_status)->'user_states'->$1\n FROM v2_job_queue q LEFT JOIN v2_job_status s USING (id)\n WHERE q.id = $2 AND q.workspace_id = $3\n ", "describe": { "columns": [ { @@ -20,5 +20,5 @@ null ] }, - "hash": "f3571e1d2b57011e5f6a38725eb42d909456d28a98563923cca43e760862e5e0" + "hash": "25cba74bec5959e6752265cd7b6f84846f74d468f0073f02f81122895e86c364" } diff --git a/backend/.sqlx/query-2729c73b53605908e2fa26b4e4bbbd75d6d5bb77e6a92bf187d978a059d7af4a.json b/backend/.sqlx/query-2729c73b53605908e2fa26b4e4bbbd75d6d5bb77e6a92bf187d978a059d7af4a.json new file mode 100644 index 0000000000..afa445d679 --- /dev/null +++ b/backend/.sqlx/query-2729c73b53605908e2fa26b4e4bbbd75d6d5bb77e6a92bf187d978a059d7af4a.json @@ -0,0 +1,20 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT COUNT(*) FROM v2_job_queue q WHERE q.canceled_by IS NULL AND (q.scheduled_for <= now()\n OR (q.suspend_until IS NOT NULL\n AND (q.suspend <= 0 OR q.suspend_until <= now())))", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "count", + "type_info": "Int8" + } + ], + "parameters": { + "Left": [] + }, + "nullable": [ + null + ] + }, + "hash": "2729c73b53605908e2fa26b4e4bbbd75d6d5bb77e6a92bf187d978a059d7af4a" +} diff --git a/backend/.sqlx/query-2bfa1ffb3d5869fc3038049ba77890203332e398c865c47aaf019dd5721d59f9.json b/backend/.sqlx/query-29785f22ee7092e8cebeae3757caf0563e3f98fff1f100616c33e8dc95fbff99.json similarity index 58% rename from backend/.sqlx/query-2bfa1ffb3d5869fc3038049ba77890203332e398c865c47aaf019dd5721d59f9.json rename to backend/.sqlx/query-29785f22ee7092e8cebeae3757caf0563e3f98fff1f100616c33e8dc95fbff99.json index 08bb98cc06..11f612ccda 100644 --- a/backend/.sqlx/query-2bfa1ffb3d5869fc3038049ba77890203332e398c865c47aaf019dd5721d59f9.json +++ b/backend/.sqlx/query-29785f22ee7092e8cebeae3757caf0563e3f98fff1f100616c33e8dc95fbff99.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "SELECT created_by AS \"created_by!\", args as \"args: sqlx::types::Json>\"\n FROM v2_as_completed_job\n WHERE id = $1 AND workspace_id = $2 AND ($3::text[] IS NULL OR tag = ANY($3))", + "query": "SELECT j.created_by AS \"created_by!\", j.args as \"args: sqlx::types::Json>\"\n FROM v2_job j\n WHERE j.id = $1 AND j.workspace_id = $2 AND ($3::text[] IS NULL OR j.tag = ANY($3))", "describe": { "columns": [ { @@ -22,9 +22,9 @@ ] }, "nullable": [ - true, + false, true ] }, - "hash": "2bfa1ffb3d5869fc3038049ba77890203332e398c865c47aaf019dd5721d59f9" + "hash": "29785f22ee7092e8cebeae3757caf0563e3f98fff1f100616c33e8dc95fbff99" } diff --git a/backend/.sqlx/query-ca5d9a9d8d18da970c7fd6eab41ecbb3a5c7c29803e4c38b8a0b2ca3790e52f9.json b/backend/.sqlx/query-35061719d01929a7146c80de4b637abdad3198d3340ec7c04ed671baff0a4d0b.json similarity index 54% rename from backend/.sqlx/query-ca5d9a9d8d18da970c7fd6eab41ecbb3a5c7c29803e4c38b8a0b2ca3790e52f9.json rename to backend/.sqlx/query-35061719d01929a7146c80de4b637abdad3198d3340ec7c04ed671baff0a4d0b.json index a76515fd7e..e3d94ad1cf 100644 --- a/backend/.sqlx/query-ca5d9a9d8d18da970c7fd6eab41ecbb3a5c7c29803e4c38b8a0b2ca3790e52f9.json +++ b/backend/.sqlx/query-35061719d01929a7146c80de4b637abdad3198d3340ec7c04ed671baff0a4d0b.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "SELECT created_by AS \"created_by!\", CONCAT(coalesce(v2_as_completed_job.logs, ''), coalesce(job_logs.logs, '')) as logs, job_logs.log_offset, job_logs.log_file_index\n FROM v2_as_completed_job\n LEFT JOIN job_logs ON job_logs.job_id = v2_as_completed_job.id\n WHERE v2_as_completed_job.id = $1 AND v2_as_completed_job.workspace_id = $2 AND ($3::text[] IS NULL OR v2_as_completed_job.tag = ANY($3))", + "query": "SELECT j.created_by AS \"created_by!\", CONCAT(coalesce(job_logs.logs, '')) as logs, job_logs.log_offset, job_logs.log_file_index\n FROM v2_job j\n LEFT JOIN job_logs ON job_logs.job_id = j.id\n WHERE j.id = $1 AND j.workspace_id = $2 AND ($3::text[] IS NULL OR j.tag = ANY($3))", "describe": { "columns": [ { @@ -32,11 +32,11 @@ ] }, "nullable": [ - true, + false, null, false, true ] }, - "hash": "ca5d9a9d8d18da970c7fd6eab41ecbb3a5c7c29803e4c38b8a0b2ca3790e52f9" + "hash": "35061719d01929a7146c80de4b637abdad3198d3340ec7c04ed671baff0a4d0b" } diff --git a/backend/.sqlx/query-280a361076d1c6317610765960f543252891c53351bdc98da66cc30ffc895866.json b/backend/.sqlx/query-371d652fbb1d34d56f645c75d12852ff982821ed7b24574cdd09a06eac0d628b.json similarity index 64% rename from backend/.sqlx/query-280a361076d1c6317610765960f543252891c53351bdc98da66cc30ffc895866.json rename to backend/.sqlx/query-371d652fbb1d34d56f645c75d12852ff982821ed7b24574cdd09a06eac0d628b.json index e9705c23a6..30550d5011 100644 --- a/backend/.sqlx/query-280a361076d1c6317610765960f543252891c53351bdc98da66cc30ffc895866.json +++ b/backend/.sqlx/query-371d652fbb1d34d56f645c75d12852ff982821ed7b24574cdd09a06eac0d628b.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "SELECT script_path FROM v2_as_completed_job WHERE id = $1", + "query": "SELECT runnable_path as script_path FROM v2_job WHERE id = $1", "describe": { "columns": [ { @@ -18,5 +18,5 @@ true ] }, - "hash": "280a361076d1c6317610765960f543252891c53351bdc98da66cc30ffc895866" + "hash": "371d652fbb1d34d56f645c75d12852ff982821ed7b24574cdd09a06eac0d628b" } diff --git a/backend/.sqlx/query-2ea447f9e644554d415367b91042687ee8690d475b8ed31c48e31180689a278f.json b/backend/.sqlx/query-37285436c16684449b33810d97d0a2611dab30faff2891fc3a7f00ee8c120950.json similarity index 55% rename from backend/.sqlx/query-2ea447f9e644554d415367b91042687ee8690d475b8ed31c48e31180689a278f.json rename to backend/.sqlx/query-37285436c16684449b33810d97d0a2611dab30faff2891fc3a7f00ee8c120950.json index ee5e15118d..c26f747c98 100644 --- a/backend/.sqlx/query-2ea447f9e644554d415367b91042687ee8690d475b8ed31c48e31180689a278f.json +++ b/backend/.sqlx/query-37285436c16684449b33810d97d0a2611dab30faff2891fc3a7f00ee8c120950.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "SELECT created_by AS \"created_by!\", CONCAT(coalesce(v2_as_queue.logs, ''), coalesce(job_logs.logs, '')) as logs, coalesce(job_logs.log_offset, 0) as log_offset, job_logs.log_file_index\n FROM v2_as_queue\n LEFT JOIN job_logs ON job_logs.job_id = v2_as_queue.id\n WHERE v2_as_queue.id = $1 AND v2_as_queue.workspace_id = $2 AND ($3::text[] IS NULL OR v2_as_queue.tag = ANY($3))", + "query": "SELECT j.created_by AS \"created_by!\", CONCAT(coalesce(job_logs.logs, '')) as logs, coalesce(job_logs.log_offset, 0) as log_offset, job_logs.log_file_index\n FROM v2_job j\n LEFT JOIN job_logs ON job_logs.job_id = j.id\n WHERE j.id = $1 AND j.workspace_id = $2 AND ($3::text[] IS NULL OR j.tag = ANY($3))", "describe": { "columns": [ { @@ -32,11 +32,11 @@ ] }, "nullable": [ - true, + false, null, null, true ] }, - "hash": "2ea447f9e644554d415367b91042687ee8690d475b8ed31c48e31180689a278f" + "hash": "37285436c16684449b33810d97d0a2611dab30faff2891fc3a7f00ee8c120950" } diff --git a/backend/.sqlx/query-4545e1e0953cde730272353e78044744910a17af23f8201d853f0a4f8404fc73.json b/backend/.sqlx/query-4545e1e0953cde730272353e78044744910a17af23f8201d853f0a4f8404fc73.json new file mode 100644 index 0000000000..c1fd3462f7 --- /dev/null +++ b/backend/.sqlx/query-4545e1e0953cde730272353e78044744910a17af23f8201d853f0a4f8404fc73.json @@ -0,0 +1,29 @@ +{ + "db_name": "PostgreSQL", + "query": "\n SELECT j.tag as \"tag!\", COUNT(*) as \"count!\"\n FROM v2_job_completed c JOIN v2_job j USING (id)\n WHERE c.started_at > NOW() - make_interval(secs => $1) AND ($2::text IS NULL OR j.workspace_id = $2)\n GROUP BY j.tag\n ORDER BY \"count!\" DESC\n ", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "tag!", + "type_info": "Varchar" + }, + { + "ordinal": 1, + "name": "count!", + "type_info": "Int8" + } + ], + "parameters": { + "Left": [ + "Float8", + "Text" + ] + }, + "nullable": [ + false, + null + ] + }, + "hash": "4545e1e0953cde730272353e78044744910a17af23f8201d853f0a4f8404fc73" +} diff --git a/backend/.sqlx/query-7af1cf089022fc1c3597b270b69aa669a153ab0c0bb2807cd4f7fd405afa6f69.json b/backend/.sqlx/query-48a41e1ad0ad8fc51624af4de34964c72fe4b3cf2497644d514c5c00d6d71bf3.json similarity index 73% rename from backend/.sqlx/query-7af1cf089022fc1c3597b270b69aa669a153ab0c0bb2807cd4f7fd405afa6f69.json rename to backend/.sqlx/query-48a41e1ad0ad8fc51624af4de34964c72fe4b3cf2497644d514c5c00d6d71bf3.json index 45a36b4fe3..61c93e6573 100644 --- a/backend/.sqlx/query-7af1cf089022fc1c3597b270b69aa669a153ab0c0bb2807cd4f7fd405afa6f69.json +++ b/backend/.sqlx/query-48a41e1ad0ad8fc51624af4de34964c72fe4b3cf2497644d514c5c00d6d71bf3.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "SELECT\n script_path, script_hash AS \"script_hash: ScriptHash\",\n job_kind AS \"job_kind!: JobKind\",\n flow_status AS \"flow_status: Json>\",\n raw_flow AS \"raw_flow: Json>\"\n FROM v2_as_completed_job WHERE id = $1 and workspace_id = $2", + "query": "SELECT\n j.runnable_path as script_path, j.runnable_id AS \"script_hash: ScriptHash\",\n j.kind AS \"job_kind!: JobKind\",\n COALESCE(c.flow_status, c.workflow_as_code_status) AS \"flow_status: Json>\",\n j.raw_flow AS \"raw_flow: Json>\"\n FROM v2_job_completed c JOIN v2_job j USING (id) WHERE j.id = $1 and j.workspace_id = $2", "describe": { "columns": [ { @@ -65,10 +65,10 @@ "nullable": [ true, true, - true, - true, + false, + null, true ] }, - "hash": "7af1cf089022fc1c3597b270b69aa669a153ab0c0bb2807cd4f7fd405afa6f69" + "hash": "48a41e1ad0ad8fc51624af4de34964c72fe4b3cf2497644d514c5c00d6d71bf3" } diff --git a/backend/.sqlx/query-2456fc71fc7a0758a4c1fbe77d72fbac2fead0e1bff4e909fd7fb1a41bc35d8f.json b/backend/.sqlx/query-49f7e7481019d27b44d5c8a167f9d5d3309a31b195b3147fa32f2ca8a6c9c90e.json similarity index 65% rename from backend/.sqlx/query-2456fc71fc7a0758a4c1fbe77d72fbac2fead0e1bff4e909fd7fb1a41bc35d8f.json rename to backend/.sqlx/query-49f7e7481019d27b44d5c8a167f9d5d3309a31b195b3147fa32f2ca8a6c9c90e.json index 07ae8f17c5..d57628b1aa 100644 --- a/backend/.sqlx/query-2456fc71fc7a0758a4c1fbe77d72fbac2fead0e1bff4e909fd7fb1a41bc35d8f.json +++ b/backend/.sqlx/query-49f7e7481019d27b44d5c8a167f9d5d3309a31b195b3147fa32f2ca8a6c9c90e.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "SELECT\n script_path, args AS \"args: sqlx::types::Json>>\",\n tag AS \"tag!\", priority\n FROM v2_as_completed_job\n WHERE id = $1 and workspace_id = $2", + "query": "SELECT\n j.runnable_path as script_path, j.args AS \"args: sqlx::types::Json>>\",\n j.tag AS \"tag!\", j.priority\n FROM v2_job j\n WHERE j.id = $1 and j.workspace_id = $2", "describe": { "columns": [ { @@ -33,9 +33,9 @@ "nullable": [ true, true, - true, + false, true ] }, - "hash": "2456fc71fc7a0758a4c1fbe77d72fbac2fead0e1bff4e909fd7fb1a41bc35d8f" + "hash": "49f7e7481019d27b44d5c8a167f9d5d3309a31b195b3147fa32f2ca8a6c9c90e" } diff --git a/backend/.sqlx/query-519f4f76649947f036a2129c11e92ef0ad30e39eec59c27bae8cb0622062c8fb.json b/backend/.sqlx/query-519f4f76649947f036a2129c11e92ef0ad30e39eec59c27bae8cb0622062c8fb.json deleted file mode 100644 index b2c0b4702c..0000000000 --- a/backend/.sqlx/query-519f4f76649947f036a2129c11e92ef0ad30e39eec59c27bae8cb0622062c8fb.json +++ /dev/null @@ -1,20 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "SELECT COUNT(*) FROM v2_as_queue WHERE canceled = false AND (scheduled_for <= now()\n OR (suspend_until IS NOT NULL\n AND ( suspend <= 0\n OR suspend_until <= now())))", - "describe": { - "columns": [ - { - "ordinal": 0, - "name": "count", - "type_info": "Int8" - } - ], - "parameters": { - "Left": [] - }, - "nullable": [ - null - ] - }, - "hash": "519f4f76649947f036a2129c11e92ef0ad30e39eec59c27bae8cb0622062c8fb" -} diff --git a/backend/.sqlx/query-0cb0e912bc942af2b1ef784455f3f073a79e300f3dd48f14122d1782eee663cd.json b/backend/.sqlx/query-57bd445bfdc667089f847acb9dc484028ce8a077308979a5ba8ef252e19aa825.json similarity index 50% rename from backend/.sqlx/query-0cb0e912bc942af2b1ef784455f3f073a79e300f3dd48f14122d1782eee663cd.json rename to backend/.sqlx/query-57bd445bfdc667089f847acb9dc484028ce8a077308979a5ba8ef252e19aa825.json index 0a9e91b206..1bb262acbf 100644 --- a/backend/.sqlx/query-0cb0e912bc942af2b1ef784455f3f073a79e300f3dd48f14122d1782eee663cd.json +++ b/backend/.sqlx/query-57bd445bfdc667089f847acb9dc484028ce8a077308979a5ba8ef252e19aa825.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "SELECT coalesce(COUNT(*) FILTER(WHERE suspend = 0 AND running = false), 0) as \"database_length!\", coalesce(COUNT(*) FILTER(WHERE suspend > 0), 0) as \"suspended!\" FROM v2_as_queue WHERE (workspace_id = $1 OR $2) AND scheduled_for <= now() AND ($3::text[] IS NULL OR tag = ANY($3))", + "query": "SELECT coalesce(COUNT(*) FILTER(WHERE q.suspend = 0 AND q.running = false), 0) as \"database_length!\", coalesce(COUNT(*) FILTER(WHERE q.suspend > 0), 0) as \"suspended!\" FROM v2_job_queue q JOIN v2_job j USING (id) WHERE (j.workspace_id = $1 OR $2) AND q.scheduled_for <= now() AND ($3::text[] IS NULL OR j.tag = ANY($3))", "describe": { "columns": [ { @@ -26,5 +26,5 @@ null ] }, - "hash": "0cb0e912bc942af2b1ef784455f3f073a79e300f3dd48f14122d1782eee663cd" + "hash": "57bd445bfdc667089f847acb9dc484028ce8a077308979a5ba8ef252e19aa825" } diff --git a/backend/.sqlx/query-d4d83d8177144c91aa489b5a42a45c83f8b069a52f681f14afb4931ac77baf45.json b/backend/.sqlx/query-6b9ea0059ad1037a77f67b70d5de13776870612e8faa8d8fa577f8e7b7309f76.json similarity index 57% rename from backend/.sqlx/query-d4d83d8177144c91aa489b5a42a45c83f8b069a52f681f14afb4931ac77baf45.json rename to backend/.sqlx/query-6b9ea0059ad1037a77f67b70d5de13776870612e8faa8d8fa577f8e7b7309f76.json index 3422605ef1..b2e0c68ba5 100644 --- a/backend/.sqlx/query-d4d83d8177144c91aa489b5a42a45c83f8b069a52f681f14afb4931ac77baf45.json +++ b/backend/.sqlx/query-6b9ea0059ad1037a77f67b70d5de13776870612e8faa8d8fa577f8e7b7309f76.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "\n SELECT id AS \"id!\", flow_status, suspend AS \"suspend!\", script_path\n FROM v2_as_queue\n WHERE id = $1\n ", + "query": "\n SELECT j.id AS \"id!\", COALESCE(s.flow_status, s.workflow_as_code_status) as flow_status, q.suspend AS \"suspend!\", j.runnable_path as script_path\n FROM v2_job_queue q JOIN v2_job j USING (id) LEFT JOIN v2_job_status s USING (id)\n WHERE j.id = $1\n ", "describe": { "columns": [ { @@ -30,11 +30,11 @@ ] }, "nullable": [ - true, - true, - true, + false, + null, + false, true ] }, - "hash": "d4d83d8177144c91aa489b5a42a45c83f8b069a52f681f14afb4931ac77baf45" + "hash": "6b9ea0059ad1037a77f67b70d5de13776870612e8faa8d8fa577f8e7b7309f76" } diff --git a/backend/.sqlx/query-70a6880960d17218bc5bf05287e2a6d9a6393c6bb1783ab8903d87dd099e236b.json b/backend/.sqlx/query-70a6880960d17218bc5bf05287e2a6d9a6393c6bb1783ab8903d87dd099e236b.json new file mode 100644 index 0000000000..7cfd1e5b64 --- /dev/null +++ b/backend/.sqlx/query-70a6880960d17218bc5bf05287e2a6d9a6393c6bb1783ab8903d87dd099e236b.json @@ -0,0 +1,58 @@ +{ + "db_name": "PostgreSQL", + "query": "\n SELECT\n j.id AS \"id!\", j.workspace_id AS \"workspace_id!\", j.parent_job, j.flow_step_id IS NOT NULL AS \"is_flow_step?\",\n COALESCE(s.flow_status, s.workflow_as_code_status) AS \"flow_status: Box\", r.ping AS last_ping, j.same_worker AS \"same_worker?\"\n FROM v2_job_queue q JOIN v2_job j USING (id) LEFT JOIN v2_job_runtime r USING (id) LEFT JOIN v2_job_status s USING (id)\n WHERE q.running = true AND q.suspend = 0 AND q.suspend_until IS null AND q.scheduled_for <= now()\n AND (j.kind = 'flow' OR j.kind = 'flowpreview' OR j.kind = 'flownode')\n AND r.ping IS NOT NULL AND r.ping < NOW() - ($1 || ' seconds')::interval\n AND q.canceled_by IS NULL\n \n ", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "id!", + "type_info": "Uuid" + }, + { + "ordinal": 1, + "name": "workspace_id!", + "type_info": "Varchar" + }, + { + "ordinal": 2, + "name": "parent_job", + "type_info": "Uuid" + }, + { + "ordinal": 3, + "name": "is_flow_step?", + "type_info": "Bool" + }, + { + "ordinal": 4, + "name": "flow_status: Box", + "type_info": "Jsonb" + }, + { + "ordinal": 5, + "name": "last_ping", + "type_info": "Timestamptz" + }, + { + "ordinal": 6, + "name": "same_worker?", + "type_info": "Bool" + } + ], + "parameters": { + "Left": [ + "Text" + ] + }, + "nullable": [ + false, + false, + true, + null, + null, + true, + false + ] + }, + "hash": "70a6880960d17218bc5bf05287e2a6d9a6393c6bb1783ab8903d87dd099e236b" +} diff --git a/backend/.sqlx/query-72956f508f66312807738399b57aa01a048fa4f9281327cf5b78111178424b43.json b/backend/.sqlx/query-72956f508f66312807738399b57aa01a048fa4f9281327cf5b78111178424b43.json deleted file mode 100644 index 2ffdf141b1..0000000000 --- a/backend/.sqlx/query-72956f508f66312807738399b57aa01a048fa4f9281327cf5b78111178424b43.json +++ /dev/null @@ -1,58 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "\n SELECT\n id AS \"id!\", workspace_id AS \"workspace_id!\", parent_job, is_flow_step,\n flow_status AS \"flow_status: Box\", last_ping, same_worker\n FROM v2_as_queue\n WHERE running = true AND suspend = 0 AND suspend_until IS null AND scheduled_for <= now()\n AND (job_kind = 'flow' OR job_kind = 'flowpreview' OR job_kind = 'flownode')\n AND last_ping IS NOT NULL AND last_ping < NOW() - ($1 || ' seconds')::interval\n AND canceled = false\n \n ", - "describe": { - "columns": [ - { - "ordinal": 0, - "name": "id!", - "type_info": "Uuid" - }, - { - "ordinal": 1, - "name": "workspace_id!", - "type_info": "Varchar" - }, - { - "ordinal": 2, - "name": "parent_job", - "type_info": "Uuid" - }, - { - "ordinal": 3, - "name": "is_flow_step", - "type_info": "Bool" - }, - { - "ordinal": 4, - "name": "flow_status: Box", - "type_info": "Jsonb" - }, - { - "ordinal": 5, - "name": "last_ping", - "type_info": "Timestamptz" - }, - { - "ordinal": 6, - "name": "same_worker", - "type_info": "Bool" - } - ], - "parameters": { - "Left": [ - "Text" - ] - }, - "nullable": [ - true, - true, - true, - true, - true, - true, - true - ] - }, - "hash": "72956f508f66312807738399b57aa01a048fa4f9281327cf5b78111178424b43" -} diff --git a/backend/.sqlx/query-89940a53f29b173b6a8717f057a81a4f2edad152581950fdd80d758a0d242c17.json b/backend/.sqlx/query-89940a53f29b173b6a8717f057a81a4f2edad152581950fdd80d758a0d242c17.json deleted file mode 100644 index 25dd18003c..0000000000 --- a/backend/.sqlx/query-89940a53f29b173b6a8717f057a81a4f2edad152581950fdd80d758a0d242c17.json +++ /dev/null @@ -1,28 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "\n SELECT usr.email, usage.executions\n FROM usr\n , LATERAL (\n SELECT COALESCE(SUM(duration_ms + 1000)/1000 , 0)::BIGINT executions\n FROM v2_as_completed_job\n WHERE workspace_id = $1\n AND job_kind NOT IN ('flow', 'flowpreview', 'flownode')\n AND email = usr.email\n AND now() - '1 week'::interval < created_at\n ) usage\n WHERE workspace_id = $1\n ", - "describe": { - "columns": [ - { - "ordinal": 0, - "name": "email", - "type_info": "Varchar" - }, - { - "ordinal": 1, - "name": "executions", - "type_info": "Int8" - } - ], - "parameters": { - "Left": [ - "Text" - ] - }, - "nullable": [ - false, - null - ] - }, - "hash": "89940a53f29b173b6a8717f057a81a4f2edad152581950fdd80d758a0d242c17" -} diff --git a/backend/.sqlx/query-90fbb9430ab03ce3aadd95cc263e5a3d1a91ea02de7608676575e1c03023ed71.json b/backend/.sqlx/query-90fbb9430ab03ce3aadd95cc263e5a3d1a91ea02de7608676575e1c03023ed71.json deleted file mode 100644 index 61598c0b2d..0000000000 --- a/backend/.sqlx/query-90fbb9430ab03ce3aadd95cc263e5a3d1a91ea02de7608676575e1c03023ed71.json +++ /dev/null @@ -1,29 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "\n SELECT tag as \"tag!\", COUNT(*) as \"count!\"\n FROM v2_as_completed_job\n WHERE started_at > NOW() - make_interval(secs => $1) AND ($2::text IS NULL OR workspace_id = $2)\n GROUP BY tag\n ORDER BY \"count!\" DESC\n ", - "describe": { - "columns": [ - { - "ordinal": 0, - "name": "tag!", - "type_info": "Varchar" - }, - { - "ordinal": 1, - "name": "count!", - "type_info": "Int8" - } - ], - "parameters": { - "Left": [ - "Float8", - "Text" - ] - }, - "nullable": [ - true, - null - ] - }, - "hash": "90fbb9430ab03ce3aadd95cc263e5a3d1a91ea02de7608676575e1c03023ed71" -} diff --git a/backend/.sqlx/query-a001b4254e0f1ba8a87776e32f9506dce6eb4af3fb32e6a4c6e76084873a539a.json b/backend/.sqlx/query-a001b4254e0f1ba8a87776e32f9506dce6eb4af3fb32e6a4c6e76084873a539a.json new file mode 100644 index 0000000000..54ab7e320e --- /dev/null +++ b/backend/.sqlx/query-a001b4254e0f1ba8a87776e32f9506dce6eb4af3fb32e6a4c6e76084873a539a.json @@ -0,0 +1,216 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT \n j.id, j.workspace_id, j.runnable_id AS \"runnable_id!: ScriptHash\", q.scheduled_for, q.started_at, j.parent_job, j.flow_innermost_root_job, j.runnable_path, j.kind as \"kind!: JobKind\", j.permissioned_as, \n j.created_by, j.script_lang AS \"script_lang!: ScriptLang\", j.permissioned_as_email, j.flow_step_id, j.trigger_kind AS \"trigger_kind!: JobTriggerKind\", j.trigger, j.priority, j.concurrent_limit, j.tag, j.cache_ttl\n FROM v2_job j LEFT JOIN v2_job_queue q ON j.id = q.id\n WHERE j.id = $1 AND j.workspace_id = $2", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "id", + "type_info": "Uuid" + }, + { + "ordinal": 1, + "name": "workspace_id", + "type_info": "Varchar" + }, + { + "ordinal": 2, + "name": "runnable_id!: ScriptHash", + "type_info": "Int8" + }, + { + "ordinal": 3, + "name": "scheduled_for", + "type_info": "Timestamptz" + }, + { + "ordinal": 4, + "name": "started_at", + "type_info": "Timestamptz" + }, + { + "ordinal": 5, + "name": "parent_job", + "type_info": "Uuid" + }, + { + "ordinal": 6, + "name": "flow_innermost_root_job", + "type_info": "Uuid" + }, + { + "ordinal": 7, + "name": "runnable_path", + "type_info": "Varchar" + }, + { + "ordinal": 8, + "name": "kind!: JobKind", + "type_info": { + "Custom": { + "name": "job_kind", + "kind": { + "Enum": [ + "script", + "preview", + "flow", + "dependencies", + "flowpreview", + "script_hub", + "identity", + "flowdependencies", + "http", + "graphql", + "postgresql", + "noop", + "appdependencies", + "deploymentcallback", + "singlestepflow", + "flowscript", + "flownode", + "appscript", + "aiagent" + ] + } + } + } + }, + { + "ordinal": 9, + "name": "permissioned_as", + "type_info": "Varchar" + }, + { + "ordinal": 10, + "name": "created_by", + "type_info": "Varchar" + }, + { + "ordinal": 11, + "name": "script_lang!: ScriptLang", + "type_info": { + "Custom": { + "name": "script_lang", + "kind": { + "Enum": [ + "python3", + "deno", + "go", + "bash", + "postgresql", + "nativets", + "bun", + "mysql", + "bigquery", + "snowflake", + "graphql", + "powershell", + "mssql", + "php", + "bunnative", + "rust", + "ansible", + "csharp", + "oracledb", + "nu", + "java", + "duckdb", + "ruby" + ] + } + } + } + }, + { + "ordinal": 12, + "name": "permissioned_as_email", + "type_info": "Varchar" + }, + { + "ordinal": 13, + "name": "flow_step_id", + "type_info": "Varchar" + }, + { + "ordinal": 14, + "name": "trigger_kind!: JobTriggerKind", + "type_info": { + "Custom": { + "name": "job_trigger_kind", + "kind": { + "Enum": [ + "webhook", + "http", + "websocket", + "kafka", + "email", + "nats", + "schedule", + "app", + "ui", + "postgres", + "sqs", + "gcp", + "mqtt" + ] + } + } + } + }, + { + "ordinal": 15, + "name": "trigger", + "type_info": "Varchar" + }, + { + "ordinal": 16, + "name": "priority", + "type_info": "Int2" + }, + { + "ordinal": 17, + "name": "concurrent_limit", + "type_info": "Int4" + }, + { + "ordinal": 18, + "name": "tag", + "type_info": "Varchar" + }, + { + "ordinal": 19, + "name": "cache_ttl", + "type_info": "Int4" + } + ], + "parameters": { + "Left": [ + "Uuid", + "Text" + ] + }, + "nullable": [ + false, + false, + true, + false, + true, + true, + true, + true, + false, + false, + false, + true, + false, + true, + true, + true, + true, + true, + false, + true + ] + }, + "hash": "a001b4254e0f1ba8a87776e32f9506dce6eb4af3fb32e6a4c6e76084873a539a" +} diff --git a/backend/.sqlx/query-11db65c493990f6935103033b2fbb0c08ae6d91b05b2f3f7c89a990d1d5a5f8a.json b/backend/.sqlx/query-a3d18ae5e5125940ae0d6af315e2dbb17c739415850e7c67f8575fb983c295fe.json similarity index 55% rename from backend/.sqlx/query-11db65c493990f6935103033b2fbb0c08ae6d91b05b2f3f7c89a990d1d5a5f8a.json rename to backend/.sqlx/query-a3d18ae5e5125940ae0d6af315e2dbb17c739415850e7c67f8575fb983c295fe.json index 1afb036030..d66a3aa9f0 100644 --- a/backend/.sqlx/query-11db65c493990f6935103033b2fbb0c08ae6d91b05b2f3f7c89a990d1d5a5f8a.json +++ b/backend/.sqlx/query-a3d18ae5e5125940ae0d6af315e2dbb17c739415850e7c67f8575fb983c295fe.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "SELECT COUNT(id) FROM v2_as_queue WHERE running = true AND email = $1", + "query": "SELECT COUNT(j.id) FROM v2_job_queue q JOIN v2_job j USING (id) WHERE q.running = true AND j.permissioned_as_email = $1", "describe": { "columns": [ { @@ -18,5 +18,5 @@ null ] }, - "hash": "11db65c493990f6935103033b2fbb0c08ae6d91b05b2f3f7c89a990d1d5a5f8a" + "hash": "a3d18ae5e5125940ae0d6af315e2dbb17c739415850e7c67f8575fb983c295fe" } diff --git a/backend/.sqlx/query-a72e7fb55d7268fe1ea40015a4a84b12dd961ba732772d8f6c59d18fe05f285d.json b/backend/.sqlx/query-a72e7fb55d7268fe1ea40015a4a84b12dd961ba732772d8f6c59d18fe05f285d.json new file mode 100644 index 0000000000..af4ada67f2 --- /dev/null +++ b/backend/.sqlx/query-a72e7fb55d7268fe1ea40015a4a84b12dd961ba732772d8f6c59d18fe05f285d.json @@ -0,0 +1,24 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT EXISTS (\n SELECT 1 FROM v2_job_completed c JOIN v2_job j USING (id)\n WHERE j.workspace_id = $2\n AND (j.kind = 'appscript' OR j.kind = 'preview')\n AND j.created_by = 'anonymous'\n AND c.started_at > now() - interval '3 hours'\n AND j.runnable_path LIKE $3 || '/%'\n AND c.result @> ('{\"s3\":\"' || $1 || '\"}')::jsonb\n )", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "exists", + "type_info": "Bool" + } + ], + "parameters": { + "Left": [ + "Text", + "Text", + "Text" + ] + }, + "nullable": [ + null + ] + }, + "hash": "a72e7fb55d7268fe1ea40015a4a84b12dd961ba732772d8f6c59d18fe05f285d" +} diff --git a/backend/.sqlx/query-d48c9a748746080e9b6cf0366b2ca3516ac5c1d32a98946196b1a42e3f103efd.json b/backend/.sqlx/query-d48c9a748746080e9b6cf0366b2ca3516ac5c1d32a98946196b1a42e3f103efd.json new file mode 100644 index 0000000000..73fb2e98d8 --- /dev/null +++ b/backend/.sqlx/query-d48c9a748746080e9b6cf0366b2ca3516ac5c1d32a98946196b1a42e3f103efd.json @@ -0,0 +1,239 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT id, q.workspace_id, j.runnable_id as \"runnable_id: ScriptHash\", scheduled_for, parent_job, flow_innermost_root_job, runnable_path, kind as \"kind: JobKind\", started_at, permissioned_as, created_by, script_lang as \"script_lang: ScriptLang\", \n permissioned_as_email, flow_step_id, trigger_kind as \"trigger_kind: JobTriggerKind\", trigger, q.priority, concurrent_limit, q.tag, cache_ttl, r.ping as last_ping, worker, memory_peak, running\n FROM v2_job_queue q JOIN v2_job j USING (id) LEFT JOIN v2_job_runtime r USING (id) LEFT JOIN v2_job_status s USING (id)\n WHERE j.id = $1", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "id", + "type_info": "Uuid" + }, + { + "ordinal": 1, + "name": "workspace_id", + "type_info": "Varchar" + }, + { + "ordinal": 2, + "name": "runnable_id: ScriptHash", + "type_info": "Int8" + }, + { + "ordinal": 3, + "name": "scheduled_for", + "type_info": "Timestamptz" + }, + { + "ordinal": 4, + "name": "parent_job", + "type_info": "Uuid" + }, + { + "ordinal": 5, + "name": "flow_innermost_root_job", + "type_info": "Uuid" + }, + { + "ordinal": 6, + "name": "runnable_path", + "type_info": "Varchar" + }, + { + "ordinal": 7, + "name": "kind: JobKind", + "type_info": { + "Custom": { + "name": "job_kind", + "kind": { + "Enum": [ + "script", + "preview", + "flow", + "dependencies", + "flowpreview", + "script_hub", + "identity", + "flowdependencies", + "http", + "graphql", + "postgresql", + "noop", + "appdependencies", + "deploymentcallback", + "singlestepflow", + "flowscript", + "flownode", + "appscript", + "aiagent" + ] + } + } + } + }, + { + "ordinal": 8, + "name": "started_at", + "type_info": "Timestamptz" + }, + { + "ordinal": 9, + "name": "permissioned_as", + "type_info": "Varchar" + }, + { + "ordinal": 10, + "name": "created_by", + "type_info": "Varchar" + }, + { + "ordinal": 11, + "name": "script_lang: ScriptLang", + "type_info": { + "Custom": { + "name": "script_lang", + "kind": { + "Enum": [ + "python3", + "deno", + "go", + "bash", + "postgresql", + "nativets", + "bun", + "mysql", + "bigquery", + "snowflake", + "graphql", + "powershell", + "mssql", + "php", + "bunnative", + "rust", + "ansible", + "csharp", + "oracledb", + "nu", + "java", + "duckdb", + "ruby" + ] + } + } + } + }, + { + "ordinal": 12, + "name": "permissioned_as_email", + "type_info": "Varchar" + }, + { + "ordinal": 13, + "name": "flow_step_id", + "type_info": "Varchar" + }, + { + "ordinal": 14, + "name": "trigger_kind: JobTriggerKind", + "type_info": { + "Custom": { + "name": "job_trigger_kind", + "kind": { + "Enum": [ + "webhook", + "http", + "websocket", + "kafka", + "email", + "nats", + "schedule", + "app", + "ui", + "postgres", + "sqs", + "gcp", + "mqtt" + ] + } + } + } + }, + { + "ordinal": 15, + "name": "trigger", + "type_info": "Varchar" + }, + { + "ordinal": 16, + "name": "priority", + "type_info": "Int2" + }, + { + "ordinal": 17, + "name": "concurrent_limit", + "type_info": "Int4" + }, + { + "ordinal": 18, + "name": "tag", + "type_info": "Varchar" + }, + { + "ordinal": 19, + "name": "cache_ttl", + "type_info": "Int4" + }, + { + "ordinal": 20, + "name": "last_ping", + "type_info": "Timestamptz" + }, + { + "ordinal": 21, + "name": "worker", + "type_info": "Varchar" + }, + { + "ordinal": 22, + "name": "memory_peak", + "type_info": "Int4" + }, + { + "ordinal": 23, + "name": "running", + "type_info": "Bool" + } + ], + "parameters": { + "Left": [ + "Uuid" + ] + }, + "nullable": [ + false, + false, + true, + false, + true, + true, + true, + false, + true, + false, + false, + true, + false, + true, + true, + true, + true, + true, + false, + true, + true, + true, + true, + false + ] + }, + "hash": "d48c9a748746080e9b6cf0366b2ca3516ac5c1d32a98946196b1a42e3f103efd" +} diff --git a/backend/.sqlx/query-ec9980f80a8bfa4b09225035e8a1f78d7d61fbf83dda9b39bb22e0f9584d221b.json b/backend/.sqlx/query-ec9980f80a8bfa4b09225035e8a1f78d7d61fbf83dda9b39bb22e0f9584d221b.json deleted file mode 100644 index d87e680abe..0000000000 --- a/backend/.sqlx/query-ec9980f80a8bfa4b09225035e8a1f78d7d61fbf83dda9b39bb22e0f9584d221b.json +++ /dev/null @@ -1,24 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "SELECT EXISTS (\n SELECT 1 FROM v2_as_completed_job\n WHERE workspace_id = $2\n AND (job_kind = 'appscript' OR job_kind = 'preview')\n AND created_by = 'anonymous'\n AND started_at > now() - interval '3 hours'\n AND script_path LIKE $3 || '/%'\n AND result @> ('{\"s3\":\"' || $1 || '\"}')::jsonb\n )", - "describe": { - "columns": [ - { - "ordinal": 0, - "name": "exists", - "type_info": "Bool" - } - ], - "parameters": { - "Left": [ - "Text", - "Text", - "Text" - ] - }, - "nullable": [ - null - ] - }, - "hash": "ec9980f80a8bfa4b09225035e8a1f78d7d61fbf83dda9b39bb22e0f9584d221b" -} diff --git a/backend/.sqlx/query-edd57b3d59ddc21b99212b5fabd4a8b793c4ae618a04220b3609c8c3c168f8fd.json b/backend/.sqlx/query-edd57b3d59ddc21b99212b5fabd4a8b793c4ae618a04220b3609c8c3c168f8fd.json new file mode 100644 index 0000000000..f67e52ab6b --- /dev/null +++ b/backend/.sqlx/query-edd57b3d59ddc21b99212b5fabd4a8b793c4ae618a04220b3609c8c3c168f8fd.json @@ -0,0 +1,28 @@ +{ + "db_name": "PostgreSQL", + "query": "\n SELECT usr.email, usage.executions\n FROM usr, LATERAL (\n SELECT COALESCE(SUM(c.duration_ms + 1000)/1000 , 0)::BIGINT executions\n FROM v2_job_completed c JOIN v2_job j USING (id)\n WHERE j.workspace_id = $1\n AND j.kind NOT IN ('flow', 'flowpreview', 'flownode')\n AND j.permissioned_as_email = usr.email\n AND now() - '1 week'::interval < j.created_at\n ) usage\n WHERE workspace_id = $1\n ", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "email", + "type_info": "Varchar" + }, + { + "ordinal": 1, + "name": "executions", + "type_info": "Int8" + } + ], + "parameters": { + "Left": [ + "Text" + ] + }, + "nullable": [ + false, + null + ] + }, + "hash": "edd57b3d59ddc21b99212b5fabd4a8b793c4ae618a04220b3609c8c3c168f8fd" +} diff --git a/backend/.sqlx/query-3f08ffbb5c71b873a9e164ecb0b10fffb37599f2a703885ee723cb9290fed13e.json b/backend/.sqlx/query-fbd38c9c4f4ecba2d3e3d79433327fd4280466eec443aff4f1e8c1ab810cca7e.json similarity index 52% rename from backend/.sqlx/query-3f08ffbb5c71b873a9e164ecb0b10fffb37599f2a703885ee723cb9290fed13e.json rename to backend/.sqlx/query-fbd38c9c4f4ecba2d3e3d79433327fd4280466eec443aff4f1e8c1ab810cca7e.json index 9cda5cc93e..29cedfbc1c 100644 --- a/backend/.sqlx/query-3f08ffbb5c71b873a9e164ecb0b10fffb37599f2a703885ee723cb9290fed13e.json +++ b/backend/.sqlx/query-fbd38c9c4f4ecba2d3e3d79433327fd4280466eec443aff4f1e8c1ab810cca7e.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "WITH job_info AS (\n -- Query for Teams (running jobs)\n SELECT\n parent.job_kind AS \"job_kind!: JobKind\",\n parent.script_hash AS \"script_hash: ScriptHash\",\n parent.raw_flow AS \"raw_flow: sqlx::types::Json>\",\n child.parent_job AS \"parent_job: Uuid\",\n parent.created_at AS \"created_at!: chrono::NaiveDateTime\",\n parent.created_by AS \"created_by!\",\n parent.script_path,\n parent.args AS \"args: sqlx::types::Json>\"\n FROM v2_as_queue child\n JOIN v2_as_queue parent ON parent.id = child.parent_job\n WHERE child.id = $1 AND child.workspace_id = $2\n UNION ALL\n -- Query for Slack (completed jobs)\n SELECT\n v2_as_queue.job_kind AS \"job_kind!: JobKind\",\n v2_as_queue.script_hash AS \"script_hash: ScriptHash\",\n v2_as_queue.raw_flow AS \"raw_flow: sqlx::types::Json>\",\n v2_as_completed_job.parent_job AS \"parent_job: Uuid\",\n v2_as_completed_job.created_at AS \"created_at!: chrono::NaiveDateTime\",\n v2_as_completed_job.created_by AS \"created_by!\",\n v2_as_queue.script_path,\n v2_as_queue.args AS \"args: sqlx::types::Json>\"\n FROM v2_as_queue\n JOIN v2_as_completed_job ON v2_as_completed_job.parent_job = v2_as_queue.id\n WHERE v2_as_completed_job.id = $1 AND v2_as_completed_job.workspace_id = $2\n )\n SELECT * FROM job_info LIMIT 1", + "query": "WITH job_info AS (\n -- Query for Teams (running jobs)\n SELECT\n parent_j.kind AS \"job_kind!: JobKind\",\n parent_j.runnable_id AS \"script_hash: ScriptHash\",\n parent_j.raw_flow AS \"raw_flow: sqlx::types::Json>\",\n child_j.parent_job AS \"parent_job: Uuid\",\n parent_j.created_at AS \"created_at!: chrono::NaiveDateTime\",\n parent_j.created_by AS \"created_by!\",\n parent_j.runnable_path as script_path,\n parent_j.args AS \"args: sqlx::types::Json>\"\n FROM v2_job_queue child_q JOIN v2_job child_j USING (id)\n JOIN v2_job parent_j ON parent_j.id = child_j.parent_job\n WHERE child_j.id = $1 AND child_j.workspace_id = $2\n UNION ALL\n -- Query for Slack (completed jobs)\n SELECT\n parent_j.kind AS \"job_kind!: JobKind\",\n parent_j.runnable_id AS \"script_hash: ScriptHash\",\n parent_j.raw_flow AS \"raw_flow: sqlx::types::Json>\",\n completed_j.parent_job AS \"parent_job: Uuid\",\n completed_j.created_at AS \"created_at!: chrono::NaiveDateTime\",\n completed_j.created_by AS \"created_by!\",\n parent_j.runnable_path as script_path,\n parent_j.args AS \"args: sqlx::types::Json>\"\n FROM v2_job_completed completed_c JOIN v2_job completed_j USING (id)\n JOIN v2_job parent_j ON parent_j.id = completed_j.parent_job\n WHERE completed_j.id = $1 AND completed_j.workspace_id = $2\n )\n SELECT * FROM job_info LIMIT 1", "describe": { "columns": [ { @@ -88,5 +88,5 @@ null ] }, - "hash": "3f08ffbb5c71b873a9e164ecb0b10fffb37599f2a703885ee723cb9290fed13e" + "hash": "fbd38c9c4f4ecba2d3e3d79433327fd4280466eec443aff4f1e8c1ab810cca7e" } diff --git a/backend/src/monitor.rs b/backend/src/monitor.rs index e310f0e3bf..b0818c96cd 100644 --- a/backend/src/monitor.rs +++ b/backend/src/monitor.rs @@ -59,7 +59,6 @@ use windmill_common::{ SAML_METADATA_SETTING, SCIM_TOKEN_SETTING, TIMEOUT_WAIT_RESULT_SETTING, }, indexer::load_indexer_config, - jobs::QueuedJob, jwt::JWT_SECRET, oauth2::REQUIRE_PREEXISTING_USER_FOR_OAUTH, server::load_smtp_config, @@ -80,7 +79,7 @@ use windmill_common::{ OTEL_METRICS_ENABLED, OTEL_TRACING_ENABLED, SERVICE_LOG_RETENTION_SECS, }; use windmill_common::{client::AuthedClient, global_settings::APP_WORKSPACED_ROUTE_SETTING}; -use windmill_queue::{cancel_job, SameWorkerPayload}; +use windmill_queue::{SameWorkerPayload, cancel_job, get_queued_job_v2}; use windmill_worker::{ handle_job_error, JobCompletedSender, SameWorkerSender, BUNFIG_INSTALL_SCOPES, INSTANCE_PYTHON_VERSION, JOB_DEFAULT_TIMEOUT, KEEP_JOB_DIR, MAVEN_REPOS, NO_DEFAULT_MAVEN, @@ -2047,7 +2046,7 @@ async fn cancel_stale_job( const RESTART_LIMIT: i32 = 3; -async fn handle_zombie_jobs(db: &Pool, base_internal_url: &str, worker_name: &str) { +async fn handle_zombie_jobs(db: &Pool, base_internal_url: &str, node_name: &str) { let mut zombie_jobs_uuid_restart_limit_reached = vec![]; if *RESTART_ZOMBIE_JOBS { @@ -2220,24 +2219,18 @@ async fn handle_zombie_jobs(db: &Pool, base_internal_url: &str, worker ); } - let jobs = sqlx::query_as::<_, QueuedJob>( - "SELECT *, null as workflow_as_code_status FROM v2_as_queue WHERE id = ANY($1)", - ) - .bind(&timeouts[..]) - .fetch_all(db) - .await - .map_err(|e| tracing::error!("Error fetching same worker jobs: {:?}", e)) - .unwrap_or_default(); - jobs + timeouts }; let non_restartable_jobs = if *RESTART_ZOMBIE_JOBS { vec![] } else { - sqlx::query_as::<_, QueuedJob>("SELECT *, null as workflow_as_code_status FROM v2_as_queue WHERE last_ping < now() - ($1 || ' seconds')::interval - AND running = true AND job_kind NOT IN ('flow', 'flowpreview', 'flownode', 'singlestepflow') AND same_worker = false") - .bind(ZOMBIE_JOB_TIMEOUT.as_str()) + sqlx::query_scalar!("SELECT j.id + FROM v2_job_queue q JOIN v2_job j USING (id) LEFT JOIN v2_job_runtime r USING (id) LEFT JOIN v2_job_status s USING (id) + WHERE r.ping < now() - ($1 || ' seconds')::interval + AND q.running = true AND j.kind NOT IN ('flow', 'flowpreview', 'flownode', 'singlestepflow') AND j.same_worker = false", + ZOMBIE_JOB_TIMEOUT.as_str()) .fetch_all(db) .await .ok() @@ -2260,14 +2253,6 @@ async fn handle_zombie_jobs(db: &Pool, base_internal_url: &str, worker } } - let zombie_jobs_restart_limit_reached = sqlx::query_as::<_, QueuedJob>( - "SELECT *, null as workflow_as_code_status FROM v2_as_queue WHERE id = ANY($1)", - ) - .bind(&zombie_jobs_uuid_restart_limit_reached[..]) - .fetch_all(db) - .await - .ok() - .unwrap_or_else(|| vec![]); let timeouts = non_restartable_jobs .into_iter() @@ -2278,7 +2263,7 @@ async fn handle_zombie_jobs(db: &Pool, base_internal_url: &str, worker .map(|x| (x, ErrorMessage::SameWorker)), ) .chain( - zombie_jobs_restart_limit_reached + zombie_jobs_uuid_restart_limit_reached .into_iter() .map(|x| (x, ErrorMessage::RestartLimit)), ) @@ -2289,7 +2274,7 @@ async fn handle_zombie_jobs(db: &Pool, base_internal_url: &str, worker QUEUE_ZOMBIE_DELETE_COUNT.inc_by(timeouts.len() as _); } - for (job, error_kind) in timeouts { + for (job_id, error_kind) in timeouts { // since the job is unrecoverable, the same worker queue should never be sent anything let (same_worker_tx_never_used, _same_worker_rx_never_used) = mpsc::channel::(1); @@ -2298,6 +2283,12 @@ async fn handle_zombie_jobs(db: &Pool, base_internal_url: &str, worker let (send_result_never_used, _send_result_rx_never_used) = JobCompletedSender::new_never_used(); + let job = get_queued_job_v2(db, &job_id).await; + if let Err(e) = job { + tracing::error!("Error getting queued job: {:?}", e); + continue; + } + if let Some(job) = job.unwrap() { let label = if job.permissioned_as != format!("u/{}", job.created_by) && job.permissioned_as != job.created_by { @@ -2311,7 +2302,7 @@ async fn handle_zombie_jobs(db: &Pool, base_internal_url: &str, worker &job.permissioned_as, &label, *SCRIPT_TOKEN_EXPIRY, - &job.email, + &job.permissioned_as_email, &job.id, None, Some(format!("handle_zombie_jobs")), @@ -2326,32 +2317,32 @@ async fn handle_zombie_jobs(db: &Pool, base_internal_url: &str, worker None, ); - let last_ping = job.last_ping.clone(); let error_message = format!( - "Job timed out after no ping from job since {} (ZOMBIE_JOB_TIMEOUT: {}, reason: {:?})", - last_ping - .map(|x| x.to_string()) - .unwrap_or_else(|| "no ping".to_string()), + "Job timed out after no ping from job since {} (ZOMBIE_JOB_TIMEOUT: {}, reason: {:?}).\nThis likely means that the job died on worker {}, OOM are a common reason for worker crashes.\nCheck the workers around the time of the last ping and the exit code if any.", + job.last_ping.unwrap_or_default(), *ZOMBIE_JOB_TIMEOUT, - error_kind.to_string() + error_kind.to_string(), + job.worker.clone().unwrap_or_default(), ); + let memory_peak = job.memory_peak.unwrap_or(0); let _ = handle_job_error( db, &client, - &windmill_queue::MiniCompletedJob::from(windmill_queue::MiniPulledJob::from(&job)), - 0, + &windmill_queue::MiniCompletedJob::from(job), + memory_peak, None, error::Error::ExecutionErr(error_message), true, Some(&same_worker_tx_never_used), "", - worker_name, + node_name, send_result_never_used, #[cfg(feature = "benchmark")] &mut windmill_common::bench::BenchmarkIter::new(), ) .await; } + } } async fn cleanup_concurrency_counters_orphaned_keys(db: &DB) -> error::Result<()> { @@ -2457,13 +2448,13 @@ async fn handle_zombie_flows(db: &DB) -> error::Result<()> { let flows = sqlx::query!( r#" SELECT - id AS "id!", workspace_id AS "workspace_id!", parent_job, is_flow_step, - flow_status AS "flow_status: Box", last_ping, same_worker - FROM v2_as_queue - WHERE running = true AND suspend = 0 AND suspend_until IS null AND scheduled_for <= now() - AND (job_kind = 'flow' OR job_kind = 'flowpreview' OR job_kind = 'flownode') - AND last_ping IS NOT NULL AND last_ping < NOW() - ($1 || ' seconds')::interval - AND canceled = false + j.id AS "id!", j.workspace_id AS "workspace_id!", j.parent_job, j.flow_step_id IS NOT NULL AS "is_flow_step?", + COALESCE(s.flow_status, s.workflow_as_code_status) AS "flow_status: Box", r.ping AS last_ping, j.same_worker AS "same_worker?" + FROM v2_job_queue q JOIN v2_job j USING (id) LEFT JOIN v2_job_runtime r USING (id) LEFT JOIN v2_job_status s USING (id) + WHERE q.running = true AND q.suspend = 0 AND q.suspend_until IS null AND q.scheduled_for <= now() + AND (j.kind = 'flow' OR j.kind = 'flowpreview' OR j.kind = 'flownode') + AND r.ping IS NOT NULL AND r.ping < NOW() - ($1 || ' seconds')::interval + AND q.canceled_by IS NULL "#, FLOW_ZOMBIE_TRANSITION_TIMEOUT.as_str() diff --git a/backend/tests/common/mod.rs b/backend/tests/common/mod.rs index 475a8540b7..40a9980a96 100644 --- a/backend/tests/common/mod.rs +++ b/backend/tests/common/mod.rs @@ -394,7 +394,17 @@ pub async fn listen_for_uuid_on( pub async fn completed_job(uuid: Uuid, db: &Pool) -> CompletedJob { sqlx::query_as::<_, CompletedJob>( - "SELECT *, result->'wm_labels' as labels FROM v2_as_completed_job WHERE id = $1", + "SELECT j.id, j.workspace_id, j.parent_job, j.created_by, j.created_at, c.duration_ms, + c.status = 'success' OR c.status = 'skipped' AS success, j.runnable_id AS script_hash, j.runnable_path AS script_path, + j.args, c.result, FALSE AS deleted, j.raw_code, c.status = 'canceled' AS canceled, + c.canceled_by, c.canceled_reason, j.kind AS job_kind, + CASE WHEN j.trigger_kind = 'schedule'::job_trigger_kind THEN j.trigger END AS schedule_path, + j.permissioned_as, COALESCE(c.flow_status, c.workflow_as_code_status) AS flow_status, j.raw_flow, + j.flow_step_id IS NOT NULL AS is_flow_step, j.script_lang AS language, c.started_at, + c.status = 'skipped' AS is_skipped, j.raw_lock, j.permissioned_as_email AS email, j.visible_to_owner, + c.memory_peak AS mem_peak, j.tag, j.priority, NULL::TEXT AS logs, c.result_columns, + j.script_entrypoint_override, j.preprocessed, c.result->'wm_labels' as labels + FROM v2_job_completed c JOIN v2_job j USING (id) WHERE j.id = $1", ) .bind(uuid) .fetch_one(db) diff --git a/backend/tests/worker.rs b/backend/tests/worker.rs index 1129846bc9..72b6aeea52 100644 --- a/backend/tests/worker.rs +++ b/backend/tests/worker.rs @@ -2395,7 +2395,7 @@ async fn test_script_schedule_handlers(db: Pool) -> anyhow::Result<()> let uuid = uuid.unwrap().unwrap(); let completed_job = sqlx::query!( - "SELECT script_path FROM v2_as_completed_job WHERE id = $1", + "SELECT runnable_path as script_path FROM v2_job WHERE id = $1", uuid ) .fetch_one(&db2) @@ -2466,7 +2466,7 @@ async fn test_script_schedule_handlers(db: Pool) -> anyhow::Result<()> let uuid = uuid.unwrap().unwrap(); let completed_job = - sqlx::query!("SELECT script_path FROM v2_as_completed_job WHERE id = $1", uuid) + sqlx::query!("SELECT runnable_path as script_path FROM v2_job WHERE id = $1", uuid) .fetch_one(&db2) .await .unwrap(); @@ -2553,7 +2553,7 @@ async fn test_flow_schedule_handlers(db: Pool) -> anyhow::Result<()> { let uuid = uuid.unwrap().unwrap(); let completed_job = sqlx::query!( - "SELECT script_path FROM v2_as_completed_job WHERE id = $1", + "SELECT runnable_path as script_path FROM v2_job WHERE id = $1", uuid ) .fetch_one(&db2) @@ -2625,7 +2625,7 @@ async fn test_flow_schedule_handlers(db: Pool) -> anyhow::Result<()> { let uuid = uuid.unwrap().unwrap(); let completed_job = - sqlx::query!("SELECT script_path FROM v2_as_completed_job WHERE id = $1", uuid) + sqlx::query!("SELECT runnable_path as script_path FROM v2_job WHERE id = $1", uuid) .fetch_one(&db2) .await .unwrap(); diff --git a/backend/windmill-api/src/approvals.rs b/backend/windmill-api/src/approvals.rs index e03b587f10..1296000481 100644 --- a/backend/windmill-api/src/approvals.rs +++ b/backend/windmill-api/src/approvals.rs @@ -204,31 +204,31 @@ pub async fn get_approval_form_details( "WITH job_info AS ( -- Query for Teams (running jobs) SELECT - parent.job_kind AS \"job_kind!: JobKind\", - parent.script_hash AS \"script_hash: ScriptHash\", - parent.raw_flow AS \"raw_flow: sqlx::types::Json>\", - child.parent_job AS \"parent_job: Uuid\", - parent.created_at AS \"created_at!: chrono::NaiveDateTime\", - parent.created_by AS \"created_by!\", - parent.script_path, - parent.args AS \"args: sqlx::types::Json>\" - FROM v2_as_queue child - JOIN v2_as_queue parent ON parent.id = child.parent_job - WHERE child.id = $1 AND child.workspace_id = $2 + parent_j.kind AS \"job_kind!: JobKind\", + parent_j.runnable_id AS \"script_hash: ScriptHash\", + parent_j.raw_flow AS \"raw_flow: sqlx::types::Json>\", + child_j.parent_job AS \"parent_job: Uuid\", + parent_j.created_at AS \"created_at!: chrono::NaiveDateTime\", + parent_j.created_by AS \"created_by!\", + parent_j.runnable_path as script_path, + parent_j.args AS \"args: sqlx::types::Json>\" + FROM v2_job_queue child_q JOIN v2_job child_j USING (id) + JOIN v2_job parent_j ON parent_j.id = child_j.parent_job + WHERE child_j.id = $1 AND child_j.workspace_id = $2 UNION ALL -- Query for Slack (completed jobs) SELECT - v2_as_queue.job_kind AS \"job_kind!: JobKind\", - v2_as_queue.script_hash AS \"script_hash: ScriptHash\", - v2_as_queue.raw_flow AS \"raw_flow: sqlx::types::Json>\", - v2_as_completed_job.parent_job AS \"parent_job: Uuid\", - v2_as_completed_job.created_at AS \"created_at!: chrono::NaiveDateTime\", - v2_as_completed_job.created_by AS \"created_by!\", - v2_as_queue.script_path, - v2_as_queue.args AS \"args: sqlx::types::Json>\" - FROM v2_as_queue - JOIN v2_as_completed_job ON v2_as_completed_job.parent_job = v2_as_queue.id - WHERE v2_as_completed_job.id = $1 AND v2_as_completed_job.workspace_id = $2 + parent_j.kind AS \"job_kind!: JobKind\", + parent_j.runnable_id AS \"script_hash: ScriptHash\", + parent_j.raw_flow AS \"raw_flow: sqlx::types::Json>\", + completed_j.parent_job AS \"parent_job: Uuid\", + completed_j.created_at AS \"created_at!: chrono::NaiveDateTime\", + completed_j.created_by AS \"created_by!\", + parent_j.runnable_path as script_path, + parent_j.args AS \"args: sqlx::types::Json>\" + FROM v2_job_completed completed_c JOIN v2_job completed_j USING (id) + JOIN v2_job parent_j ON parent_j.id = completed_j.parent_job + WHERE completed_j.id = $1 AND completed_j.workspace_id = $2 ) SELECT * FROM job_info LIMIT 1", job_id, diff --git a/backend/windmill-api/src/apps.rs b/backend/windmill-api/src/apps.rs index 1cd0750919..b669f5e9a9 100644 --- a/backend/windmill-api/src/apps.rs +++ b/backend/windmill-api/src/apps.rs @@ -2530,13 +2530,13 @@ async fn check_if_allowed_to_access_s3_file_from_app( || { sqlx::query_scalar!( r#"SELECT EXISTS ( - SELECT 1 FROM v2_as_completed_job - WHERE workspace_id = $2 - AND (job_kind = 'appscript' OR job_kind = 'preview') - AND created_by = 'anonymous' - AND started_at > now() - interval '3 hours' - AND script_path LIKE $3 || '/%' - AND result @> ('{"s3":"' || $1 || '"}')::jsonb + SELECT 1 FROM v2_job_completed c JOIN v2_job j USING (id) + WHERE j.workspace_id = $2 + AND (j.kind = 'appscript' OR j.kind = 'preview') + AND j.created_by = 'anonymous' + AND c.started_at > now() - interval '3 hours' + AND j.runnable_path LIKE $3 || '/%' + AND c.result @> ('{"s3":"' || $1 || '"}')::jsonb )"#, file_query.s3, w_id, diff --git a/backend/windmill-api/src/jobs.rs b/backend/windmill-api/src/jobs.rs index 4c11515291..0293b4b625 100644 --- a/backend/windmill-api/src/jobs.rs +++ b/backend/windmill-api/src/jobs.rs @@ -841,101 +841,6 @@ macro_rules! get_job_query { } } -// CREATE OR REPLACE VIEW v2_as_queue AS -// SELECT -// j.id, -// j.workspace_id, -// j.parent_job, -// j.created_by, -// j.created_at, -// q.started_at, -// q.scheduled_for, -// q.running, -// j.runnable_id AS script_hash, -// j.runnable_path AS script_path, -// j.args, -// j.raw_code, -// q.canceled_by IS NOT NULL AS canceled, -// q.canceled_by, -// q.canceled_reason, -// r.ping AS last_ping, -// j.kind AS job_kind, -// CASE WHEN j.trigger_kind = 'schedule'::job_trigger_kind THEN j.trigger END -// AS schedule_path, -// j.permissioned_as, -// COALESCE(s.flow_status, s.workflow_as_code_status) AS flow_status, -// j.raw_flow, -// j.flow_step_id IS NOT NULL AS is_flow_step, -// j.script_lang AS language, -// q.suspend, -// q.suspend_until, -// j.same_worker, -// j.raw_lock, -// j.pre_run_error, -// j.permissioned_as_email AS email, -// j.visible_to_owner, -// r.memory_peak AS mem_peak, -// j.flow_innermost_root_job AS root_job, -// s.flow_leaf_jobs AS leaf_jobs, -// j.tag, -// j.concurrent_limit, -// j.concurrency_time_window_s, -// j.timeout, -// j.flow_step_id, -// j.cache_ttl, -// j.priority, -// NULL::TEXT AS logs, -// j.script_entrypoint_override, -// j.preprocessed -// FROM v2_job_queue q -// JOIN v2_job j USING (id) -// LEFT JOIN v2_job_runtime r USING (id) -// LEFT JOIN v2_job_status s USING (id) -// ; - -// -- Add up migration script here -// CREATE OR REPLACE VIEW v2_as_completed_job AS -// SELECT -// j.id, -// j.workspace_id, -// j.parent_job, -// j.created_by, -// j.created_at, -// c.duration_ms, -// c.status = 'success' OR c.status = 'skipped' AS success, -// j.runnable_id AS script_hash, -// j.runnable_path AS script_path, -// j.args, -// c.result, -// FALSE AS deleted, -// j.raw_code, -// c.status = 'canceled' AS canceled, -// c.canceled_by, -// c.canceled_reason, -// j.kind AS job_kind, -// CASE WHEN j.trigger_kind = 'schedule'::job_trigger_kind THEN j.trigger END -// AS schedule_path, -// j.permissioned_as, -// COALESCE(c.flow_status, c.workflow_as_code_status) AS flow_status, -// j.raw_flow, -// j.flow_step_id IS NOT NULL AS is_flow_step, -// j.script_lang AS language, -// c.started_at, -// c.status = 'skipped' AS is_skipped, -// j.raw_lock, -// j.permissioned_as_email AS email, -// j.visible_to_owner, -// c.memory_peak AS mem_peak, -// j.tag, -// j.priority, -// NULL::TEXT AS logs, -// c.result_columns, -// j.script_entrypoint_override, -// j.preprocessed -// FROM v2_job_completed c -// JOIN v2_job j USING (id) -// ; - #[derive(Copy, Clone)] struct GetQuery<'a> { with_logs: bool, @@ -1506,10 +1411,10 @@ async fn get_job_logs( .flatten(); let record = sqlx::query!( - "SELECT created_by AS \"created_by!\", CONCAT(coalesce(v2_as_completed_job.logs, ''), coalesce(job_logs.logs, '')) as logs, job_logs.log_offset, job_logs.log_file_index - FROM v2_as_completed_job - LEFT JOIN job_logs ON job_logs.job_id = v2_as_completed_job.id - WHERE v2_as_completed_job.id = $1 AND v2_as_completed_job.workspace_id = $2 AND ($3::text[] IS NULL OR v2_as_completed_job.tag = ANY($3))", + "SELECT j.created_by AS \"created_by!\", CONCAT(coalesce(job_logs.logs, '')) as logs, job_logs.log_offset, job_logs.log_file_index + FROM v2_job j + LEFT JOIN job_logs ON job_logs.job_id = j.id + WHERE j.id = $1 AND j.workspace_id = $2 AND ($3::text[] IS NULL OR j.tag = ANY($3))", id, w_id, tags.as_ref().map(|v| v.as_slice()) @@ -1554,10 +1459,10 @@ async fn get_job_logs( Ok(content_plain(Body::from(logs))) } else { let text = sqlx::query!( - "SELECT created_by AS \"created_by!\", CONCAT(coalesce(v2_as_queue.logs, ''), coalesce(job_logs.logs, '')) as logs, coalesce(job_logs.log_offset, 0) as log_offset, job_logs.log_file_index - FROM v2_as_queue - LEFT JOIN job_logs ON job_logs.job_id = v2_as_queue.id - WHERE v2_as_queue.id = $1 AND v2_as_queue.workspace_id = $2 AND ($3::text[] IS NULL OR v2_as_queue.tag = ANY($3))", + "SELECT j.created_by AS \"created_by!\", CONCAT(coalesce(job_logs.logs, '')) as logs, coalesce(job_logs.log_offset, 0) as log_offset, job_logs.log_file_index + FROM v2_job j + LEFT JOIN job_logs ON job_logs.job_id = j.id + WHERE j.id = $1 AND j.workspace_id = $2 AND ($3::text[] IS NULL OR j.tag = ANY($3))", id, w_id, tags.as_ref().map(|v| v.as_slice()) @@ -1613,9 +1518,9 @@ async fn get_args( .map(|authed| get_scope_tags(authed)) .flatten(); let record = sqlx::query!( - "SELECT created_by AS \"created_by!\", args as \"args: sqlx::types::Json>\" - FROM v2_as_completed_job - WHERE id = $1 AND workspace_id = $2 AND ($3::text[] IS NULL OR tag = ANY($3))", + "SELECT j.created_by AS \"created_by!\", j.args as \"args: sqlx::types::Json>\" + FROM v2_job j + WHERE j.id = $1 AND j.workspace_id = $2 AND ($3::text[] IS NULL OR j.tag = ANY($3))", id, &w_id, tags.as_ref().map(|v| v.as_slice()) as Option<&[&str]>, @@ -2170,7 +2075,7 @@ async fn cancel_selection( let mut tx = user_db.begin(&authed).await?; let tags = get_scope_tags(&authed).map(|v| v.iter().map(|s| s.to_string()).collect_vec()); let jobs_to_cancel = sqlx::query_scalar!( - "SELECT id AS \"id!\" FROM v2_as_queue WHERE id = ANY($1) AND schedule_path IS NULL AND ($2::text[] IS NULL OR tag = ANY($2))", + "SELECT j.id AS \"id!\" FROM v2_job j WHERE j.id = ANY($1) AND j.trigger_kind != 'schedule'::job_trigger_kind AND ($2::text[] IS NULL OR j.tag = ANY($2))", &jobs, tags.as_ref().map(|v| v.as_slice()) ) @@ -2274,7 +2179,7 @@ async fn count_queue_jobs( Ok(Json( sqlx::query_as!( QueueStats, - "SELECT coalesce(COUNT(*) FILTER(WHERE suspend = 0 AND running = false), 0) as \"database_length!\", coalesce(COUNT(*) FILTER(WHERE suspend > 0), 0) as \"suspended!\" FROM v2_as_queue WHERE (workspace_id = $1 OR $2) AND scheduled_for <= now() AND ($3::text[] IS NULL OR tag = ANY($3))", + "SELECT coalesce(COUNT(*) FILTER(WHERE q.suspend = 0 AND q.running = false), 0) as \"database_length!\", coalesce(COUNT(*) FILTER(WHERE q.suspend > 0), 0) as \"suspended!\" FROM v2_job_queue q JOIN v2_job j USING (id) WHERE (j.workspace_id = $1 OR $2) AND q.scheduled_for <= now() AND ($3::text[] IS NULL OR j.tag = ANY($3))", w_id, w_id == "admins" && cq.all_workspaces.unwrap_or(false), tags.as_ref().map(|v| v.as_slice()) @@ -2717,9 +2622,9 @@ async fn get_suspended_flow_info<'c>( let flow = sqlx::query_as!( FlowInfo, r#" - SELECT id AS "id!", flow_status, suspend AS "suspend!", script_path - FROM v2_as_queue - WHERE id = $1 + SELECT j.id AS "id!", COALESCE(s.flow_status, s.workflow_as_code_status) as flow_status, q.suspend AS "suspend!", j.runnable_path as script_path + FROM v2_job_queue q JOIN v2_job j USING (id) LEFT JOIN v2_job_status s USING (id) + WHERE j.id = $1 "#, job_id, ) @@ -2933,9 +2838,9 @@ pub async fn get_flow_user_state( let mut tx = user_db.begin(&authed).await?; let r = sqlx::query_scalar!( r#" - SELECT flow_status->'user_states'->$1 - FROM v2_as_queue - WHERE id = $2 AND workspace_id = $3 + SELECT COALESCE(s.flow_status, s.workflow_as_code_status)->'user_states'->$1 + FROM v2_job_queue q LEFT JOIN v2_job_status s USING (id) + WHERE q.id = $2 AND q.workspace_id = $3 "#, key, job_id, @@ -4166,10 +4071,10 @@ pub async fn restart_flow( let mut tx = user_db.clone().begin(&authed).await?; let completed_job = sqlx::query!( "SELECT - script_path, args AS \"args: sqlx::types::Json>>\", - tag AS \"tag!\", priority - FROM v2_as_completed_job - WHERE id = $1 and workspace_id = $2", + j.runnable_path as script_path, j.args AS \"args: sqlx::types::Json>>\", + j.tag AS \"tag!\", j.priority + FROM v2_job j + WHERE j.id = $1 and j.workspace_id = $2", job_id, &w_id, ) @@ -4819,10 +4724,9 @@ pub async fn delete_job_metadata_after_use(db: &DB, job_uuid: Uuid) -> Result<() pub async fn check_queue_too_long(db: &DB, queue_limit: Option) -> error::Result<()> { if let Some(limit) = queue_limit { let count = sqlx::query_scalar!( - "SELECT COUNT(*) FROM v2_as_queue WHERE canceled = false AND (scheduled_for <= now() - OR (suspend_until IS NOT NULL - AND ( suspend <= 0 - OR suspend_until <= now())))", + "SELECT COUNT(*) FROM v2_job_queue q WHERE q.canceled_by IS NULL AND (q.scheduled_for <= now() + OR (q.suspend_until IS NOT NULL + AND (q.suspend <= 0 OR q.suspend_until <= now())))", ) .fetch_one(db) .await? @@ -7963,10 +7867,10 @@ async fn count_by_tag( let counts = sqlx::query_as!( TagCount, r#" - SELECT tag as "tag!", COUNT(*) as "count!" - FROM v2_as_completed_job - WHERE started_at > NOW() - make_interval(secs => $1) AND ($2::text IS NULL OR workspace_id = $2) - GROUP BY tag + SELECT j.tag as "tag!", COUNT(*) as "count!" + FROM v2_job_completed c JOIN v2_job j USING (id) + WHERE c.started_at > NOW() - make_interval(secs => $1) AND ($2::text IS NULL OR j.workspace_id = $2) + GROUP BY j.tag ORDER BY "count!" DESC "#, horizon as f64, diff --git a/backend/windmill-api/src/users.rs b/backend/windmill-api/src/users.rs index 165316ba56..9021d6fe8e 100644 --- a/backend/windmill-api/src/users.rs +++ b/backend/windmill-api/src/users.rs @@ -494,14 +494,13 @@ async fn list_user_usage( UserWithUsage, " SELECT usr.email, usage.executions - FROM usr - , LATERAL ( - SELECT COALESCE(SUM(duration_ms + 1000)/1000 , 0)::BIGINT executions - FROM v2_as_completed_job - WHERE workspace_id = $1 - AND job_kind NOT IN ('flow', 'flowpreview', 'flownode') - AND email = usr.email - AND now() - '1 week'::interval < created_at + FROM usr, LATERAL ( + SELECT COALESCE(SUM(c.duration_ms + 1000)/1000 , 0)::BIGINT executions + FROM v2_job_completed c JOIN v2_job j USING (id) + WHERE j.workspace_id = $1 + AND j.kind NOT IN ('flow', 'flowpreview', 'flownode') + AND j.permissioned_as_email = usr.email + AND now() - '1 week'::interval < j.created_at ) usage WHERE workspace_id = $1 ", diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs index 4311972923..f179216d24 100644 --- a/backend/windmill-queue/src/jobs.rs +++ b/backend/windmill-queue/src/jobs.rs @@ -6,6 +6,7 @@ * LICENSE-AGPL for a copy of the license. */ +use std::future::Future; use std::{collections::HashMap, sync::Arc, vec}; use anyhow::Context; @@ -154,18 +155,19 @@ pub struct JobCompleted { pub async fn cancel_single_job<'c>( username: &str, reason: Option, - job_running: Arc, + job_running: QueuedJobV2, w_id: &str, mut tx: Transaction<'c, Postgres>, db: &Pool, force_cancel: bool, ) -> error::Result<(Transaction<'c, Postgres>, Option)> { + + let id = job_running.id; if force_cancel || (job_running.parent_job.is_none() && !job_running.running) { let username = username.to_string(); let w_id = w_id.to_string(); let db = db.clone(); tracing::info!("cancelling job {:?}", job_running.id); - let job_running = job_running.clone(); tokio::task::spawn(async move { let reason: String = reason .clone() @@ -179,10 +181,11 @@ pub async fn cancel_single_job<'c>( &Connection::from(db.clone()), ) .await; + let memory_peak = job_running.memory_peak.unwrap_or(0); let add_job = add_completed_job_error( &db, - &MiniCompletedJob::from(MiniPulledJob::from(&job_running)), - job_running.mem_peak.unwrap_or(0), + &MiniCompletedJob::from(job_running), + memory_peak, Some(CanceledBy { username: Some(username.to_string()), reason: Some(reason) }), e, "server", @@ -210,7 +213,7 @@ pub async fn cancel_single_job<'c>( } } - Ok((tx, Some(job_running.id))) + Ok((tx, Some(id))) } pub async fn cancel_job<'c>( @@ -224,26 +227,35 @@ pub async fn cancel_job<'c>( require_anonymous: bool, ) -> error::Result<(Transaction<'c, Postgres>, Option)> { //TODO fetch mini completed job instead of QueuedJob - let job = get_queued_job_tx(id, &w_id, &mut tx).await?; + let job = get_queued_job_v2(&mut *tx, &id).await?; + if job.is_none() { return Ok((tx, None)); } - if require_anonymous && job.as_ref().unwrap().created_by != "anonymous" { + let mut job = job.unwrap(); + + if require_anonymous && job.created_by != "anonymous" { return Err(Error::BadRequest( "You are not logged in and this job was not created by an anonymous user like you so you cannot cancel it".to_string(), )); } - let mut job = job.unwrap(); + + if job.workspace_id != w_id { + return Err(Error::BadRequest( + "You are not authorized to cancel this job belonging to another workspace".to_string(), + )); + } + if force_cancel { // if force canceling a flow step, make sure we force cancel from the highest parent loop { if job.parent_job.is_none() { break; } - match get_queued_job_tx(job.parent_job.unwrap(), &w_id, &mut tx).await? { + match get_queued_job_v2(&mut *tx, &job.parent_job.unwrap()).await? { Some(j) => { job = j; } @@ -253,7 +265,7 @@ pub async fn cancel_job<'c>( } // prevent cancelling a future tick of a schedule - if let Some(schedule_path) = job.schedule_path.as_ref() { + if let Some(schedule_path) = job.schedule_path().as_ref() { let now = now_from_db(&mut *tx).await?; if job.scheduled_for > now { return Err(Error::BadRequest( @@ -266,7 +278,7 @@ pub async fn cancel_job<'c>( } } - let job = Arc::new(job); + let job = job; // get all children using recursive CTE let mut jobs_to_cancel = sqlx::query!( @@ -308,7 +320,7 @@ ORDER BY depth, id let (ntx, _) = cancel_single_job( username, reason.clone(), - job.clone(), + job, w_id, tx, db, @@ -335,13 +347,13 @@ ORDER BY depth, id } } for job_id in jobs_to_cancel { - let job = get_queued_job_tx(job_id, &w_id, &mut tx).await?; + let job = get_queued_job_v2(&mut *tx, &job_id).await?; if let Some(job) = job { let (ntx, _) = cancel_single_job( username, reason.clone(), - Arc::new(job), + job, w_id, tx, db, @@ -554,7 +566,7 @@ async fn cancel_persistent_script_jobs_internal<'c>( // we could have retrieved the job IDs in the first query where we retrieve the hashes, but just in case a job was inserted in the queue right in-between the two above query, we re-do the fetch here let jobs_to_cancel = sqlx::query_scalar::<_, Uuid>( - "SELECT id FROM v2_as_queue WHERE workspace_id = $1 AND script_path = $2 AND canceled = false", + "SELECT j.id FROM v2_job_queue q JOIN v2_job j USING (id) WHERE j.workspace_id = $1 AND j.runnable_path = $2 AND q.canceled_by IS NULL", ) .bind(w_id) .bind(script_path) @@ -1948,6 +1960,34 @@ pub struct MiniCompletedJob { pub cache_ttl: Option, } +impl From for MiniCompletedJob { + fn from(job: QueuedJobV2) -> Self { + MiniCompletedJob { + id: job.id, + workspace_id: job.workspace_id, + runnable_id: job.runnable_id, + scheduled_for: job.scheduled_for, + parent_job: job.parent_job, + flow_innermost_root_job: job.flow_innermost_root_job, + runnable_path: job.runnable_path, + kind: job.kind, + started_at: job.started_at, + permissioned_as: job.permissioned_as, + created_by: job.created_by, + script_lang: job.script_lang, + permissioned_as_email: job.permissioned_as_email, + flow_step_id: job.flow_step_id, + trigger_kind: job.trigger_kind, + trigger: job.trigger, + priority: job.priority, + concurrent_limit: job.concurrent_limit, + tag: job.tag, + cache_ttl: job.cache_ttl, + + } + } +} + impl From for MiniCompletedJob { fn from(job: MiniPulledJob) -> Self { MiniCompletedJob { @@ -2008,11 +2048,7 @@ impl MiniCompletedJob { self.flow_step_id.is_some() } pub fn schedule_path(&self) -> Option { - if self.trigger_kind.as_ref().is_some_and(|t| matches!(t, JobTriggerKind::Schedule)) { - self.trigger.clone() - } else { - None - } + schedule_path(&self.trigger_kind, &self.trigger) } pub fn is_flow(&self) -> bool { @@ -2025,6 +2061,14 @@ impl MiniCompletedJob { } +fn schedule_path(trigger_kind: &Option, trigger: &Option) -> Option { + if trigger_kind.as_ref().is_some_and(|t| matches!(t, JobTriggerKind::Schedule)) { + trigger.clone() + } else { + None + } +} + #[derive(Serialize, Deserialize, Debug, Clone)] struct FlowStatusChatInputEnabled { chat_input_enabled: Option, @@ -2111,15 +2155,7 @@ impl MiniPulledJob { } pub fn schedule_path(&self) -> Option { - if self - .trigger_kind - .as_ref() - .is_some_and(|t| matches!(t, JobTriggerKind::Schedule)) - { - self.trigger.clone() - } else { - None - } + schedule_path(&self.trigger_kind, &self.trigger) } pub async fn mark_as_started_if_step(&self, db: &DB) -> Result<(), Error> { @@ -2316,6 +2352,57 @@ pub async fn get_mini_pulled_job<'c>( Ok(job) } + +pub struct QueuedJobV2 { + pub id: Uuid, + pub workspace_id: String, + pub runnable_id: Option, + pub scheduled_for: chrono::DateTime, + pub parent_job: Option, + // pub root_job: Option, + pub flow_innermost_root_job: Option, + pub runnable_path: Option, + pub kind: JobKind, + pub started_at: Option>, + pub permissioned_as: String, + pub created_by: String, + pub script_lang: Option, + pub permissioned_as_email: String, + pub flow_step_id: Option, + pub trigger_kind: Option, + pub trigger: Option, + pub priority: Option, + pub concurrent_limit: Option, + pub tag: String, + pub cache_ttl: Option, + pub last_ping: Option>, + pub worker: Option, + pub memory_peak: Option, + pub running: bool, +} + +impl QueuedJobV2 { + pub fn schedule_path(&self) -> Option { + schedule_path(&self.trigger_kind, &self.trigger) + } +} + +pub async fn get_queued_job_v2<'c>( + e: impl PgExecutor<'c>, job_id: &Uuid) -> error::Result> { + let job = sqlx::query_as!( + QueuedJobV2, + "SELECT id, q.workspace_id, j.runnable_id as \"runnable_id: ScriptHash\", scheduled_for, parent_job, flow_innermost_root_job, runnable_path, kind as \"kind: JobKind\", started_at, permissioned_as, created_by, script_lang as \"script_lang: ScriptLang\", + permissioned_as_email, flow_step_id, trigger_kind as \"trigger_kind: JobTriggerKind\", trigger, q.priority, concurrent_limit, q.tag, cache_ttl, r.ping as last_ping, worker, memory_peak, running + FROM v2_job_queue q JOIN v2_job j USING (id) LEFT JOIN v2_job_runtime r USING (id) LEFT JOIN v2_job_status s USING (id) + WHERE j.id = $1", + job_id, + + ) + .fetch_optional(e) + .await?; + Ok(job) +} + #[derive(Serialize, Deserialize, Debug)] pub struct PulledJobResult { pub job: Option, @@ -3170,33 +3257,29 @@ pub async fn job_is_complete(db: &DB, id: Uuid, w_id: &str) -> error::Result( - id: Uuid, - w_id: &str, - tx: &mut Transaction<'c, Postgres>, -) -> error::Result> { - sqlx::query_as::<_, QueuedJob>( - "SELECT *, null as workflow_as_code_status - FROM v2_as_queue WHERE id = $1 AND workspace_id = $2", - ) - .bind(id) - .bind(w_id) - .fetch_optional(&mut **tx) - .await - .map_err(Into::into) +pub fn get_mini_completed_job< +'a, +'e, +A: sqlx::Acquire<'e, Database = Postgres> + Send + 'a, +>(id: &'a Uuid, w_id: &'a str, db: A) -> impl Future>> + Send + 'a { + async move { + let mut conn = db.acquire().await?; + sqlx::query_as!( + MiniCompletedJob, + "SELECT + j.id, j.workspace_id, j.runnable_id AS \"runnable_id!: ScriptHash\", q.scheduled_for, q.started_at, j.parent_job, j.flow_innermost_root_job, j.runnable_path, j.kind as \"kind!: JobKind\", j.permissioned_as, + j.created_by, j.script_lang AS \"script_lang!: ScriptLang\", j.permissioned_as_email, j.flow_step_id, j.trigger_kind AS \"trigger_kind!: JobTriggerKind\", j.trigger, j.priority, j.concurrent_limit, j.tag, j.cache_ttl + FROM v2_job j LEFT JOIN v2_job_queue q ON j.id = q.id + WHERE j.id = $1 AND j.workspace_id = $2", + id, + w_id + ) + .fetch_optional(&mut *conn) + .await + .map_err(Into::into) + } } -pub async fn get_queued_job(id: &Uuid, w_id: &str, db: &DB) -> error::Result> { - sqlx::query_as::<_, QueuedJob>( - "SELECT *, null as workflow_as_code_status - FROM v2_as_queue WHERE id = $1 AND workspace_id = $2", - ) - .bind(id) - .bind(w_id) - .fetch_optional(db) - .await - .map_err(Into::into) -} pub enum PushIsolationLevel<'c> { IsolatedRoot(DB), @@ -3557,7 +3640,7 @@ pub async fn push<'c, 'd>( } let in_queue = sqlx::query_scalar!( - "SELECT COUNT(id) FROM v2_as_queue WHERE email = $1", + "SELECT COUNT(id) FROM v2_job WHERE permissioned_as_email = $1", email ) .fetch_one(_db) @@ -3571,7 +3654,7 @@ pub async fn push<'c, 'd>( } let concurrent_runs = sqlx::query_scalar!( - "SELECT COUNT(id) FROM v2_as_queue WHERE running = true AND email = $1", + "SELECT COUNT(j.id) FROM v2_job_queue q JOIN v2_job j USING (id) WHERE q.running = true AND j.permissioned_as_email = $1", email ) .fetch_one(_db) @@ -5155,11 +5238,11 @@ async fn restarted_flows_resolution( > { let row = sqlx::query!( "SELECT - script_path, script_hash AS \"script_hash: ScriptHash\", - job_kind AS \"job_kind!: JobKind\", - flow_status AS \"flow_status: Json>\", - raw_flow AS \"raw_flow: Json>\" - FROM v2_as_completed_job WHERE id = $1 and workspace_id = $2", + j.runnable_path as script_path, j.runnable_id AS \"script_hash: ScriptHash\", + j.kind AS \"job_kind!: JobKind\", + COALESCE(c.flow_status, c.workflow_as_code_status) AS \"flow_status: Json>\", + j.raw_flow AS \"raw_flow: Json>\" + FROM v2_job_completed c JOIN v2_job j USING (id) WHERE j.id = $1 and j.workspace_id = $2", completed_flow_id, workspace_id, ) diff --git a/backend/windmill-worker/src/result_processor.rs b/backend/windmill-worker/src/result_processor.rs index 505213bb68..7cfe63dc34 100644 --- a/backend/windmill-worker/src/result_processor.rs +++ b/backend/windmill-worker/src/result_processor.rs @@ -31,8 +31,7 @@ use windmill_common::{ use windmill_common::bench::{BenchmarkInfo, BenchmarkIter}; use windmill_queue::{ - append_logs, get_queued_job, CanceledBy, JobCompleted, MiniCompletedJob, MiniPulledJob, - ValidableJson, WrappedError, INIT_SCRIPT_TAG, + CanceledBy, INIT_SCRIPT_TAG, JobCompleted, MiniCompletedJob, MiniPulledJob, ValidableJson, WrappedError, append_logs, get_mini_completed_job }; use serde_json::{json, value::RawValue, Value}; @@ -816,9 +815,8 @@ pub async fn handle_job_error( if let Err(err) = updated_flow { if let Some(parent_job_id) = job.parent_job { - // TODO get minicompleted job directly if let Ok(Some(parent_job)) = - get_queued_job(&parent_job_id, &job.workspace_id, &db).await + get_mini_completed_job(&parent_job_id, &job.workspace_id, db).await { let e = json!({"message": err.to_string(), "name": "InternalErr"}); append_logs( @@ -830,7 +828,7 @@ pub async fn handle_job_error( .await; let _ = add_completed_job_error( db, - &MiniCompletedJob::from(MiniPulledJob::from(&parent_job)), + &parent_job, mem_peak, canceled_by.clone(), e,