diff --git a/backend/.sqlx/query-27ab808d7e05235180449df043ef57c6fbe7d3e261e5a52313d01782739c0e77.json b/backend/.sqlx/query-11ae5e1aade2bc9a78af2d956ece73766ae3330319619a68383b508a76f2d19c.json similarity index 85% rename from backend/.sqlx/query-27ab808d7e05235180449df043ef57c6fbe7d3e261e5a52313d01782739c0e77.json rename to backend/.sqlx/query-11ae5e1aade2bc9a78af2d956ece73766ae3330319619a68383b508a76f2d19c.json index 3e01927de4..67bdf6ea9d 100644 --- a/backend/.sqlx/query-27ab808d7e05235180449df043ef57c6fbe7d3e261e5a52313d01782739c0e77.json +++ b/backend/.sqlx/query-11ae5e1aade2bc9a78af2d956ece73766ae3330319619a68383b508a76f2d19c.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "select path, tag, concurrent_limit, concurrency_time_window_s, cache_ttl, language as \"language: ScriptLang\", dedicated_worker from script where hash = $1 AND workspace_id = $2", + "query": "select path, tag, concurrent_limit, concurrency_time_window_s, cache_ttl, language as \"language: ScriptLang\", dedicated_worker, priority from script where hash = $1 AND workspace_id = $2", "describe": { "columns": [ { @@ -42,7 +42,6 @@ "bash", "postgresql", "nativets", - "Nativets", "bun", "mysql", "bigquery", @@ -58,6 +57,11 @@ "ordinal": 6, "name": "dedicated_worker", "type_info": "Bool" + }, + { + "ordinal": 7, + "name": "priority", + "type_info": "Int2" } ], "parameters": { @@ -73,8 +77,9 @@ true, true, false, + true, true ] }, - "hash": "27ab808d7e05235180449df043ef57c6fbe7d3e261e5a52313d01782739c0e77" + "hash": "11ae5e1aade2bc9a78af2d956ece73766ae3330319619a68383b508a76f2d19c" } diff --git a/backend/.sqlx/query-25bef6a248f3ee0ea2cbcc376c217cbcf1013ae311c36b42d423bf6a02fa016c.json b/backend/.sqlx/query-254776959f60455a00a2c29a1783ee6b27c836197df6454d3bb5c59441c34f44.json similarity index 83% rename from backend/.sqlx/query-25bef6a248f3ee0ea2cbcc376c217cbcf1013ae311c36b42d423bf6a02fa016c.json rename to backend/.sqlx/query-254776959f60455a00a2c29a1783ee6b27c836197df6454d3bb5c59441c34f44.json index bf591ef11c..272aefc9d2 100644 --- a/backend/.sqlx/query-25bef6a248f3ee0ea2cbcc376c217cbcf1013ae311c36b42d423bf6a02fa016c.json +++ b/backend/.sqlx/query-254776959f60455a00a2c29a1783ee6b27c836197df6454d3bb5c59441c34f44.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "INSERT INTO completed_job AS cj\n ( workspace_id\n , id\n , parent_job\n , created_by\n , created_at\n , started_at\n , duration_ms\n , success\n , script_hash\n , script_path\n , args\n , result\n , logs\n , raw_code\n , raw_lock\n , canceled\n , canceled_by\n , canceled_reason\n , job_kind\n , schedule_path\n , permissioned_as\n , flow_status\n , raw_flow\n , is_flow_step\n , is_skipped\n , language\n , email\n , visible_to_owner\n , mem_peak\n , tag\n )\n VALUES ($1, $2, $3, $4, $5, COALESCE($6, now()), COALESCE($26, (EXTRACT('epoch' FROM (now())) - EXTRACT('epoch' FROM (COALESCE($6, now()))))*1000), $7, $8, $9,$10, $11, $12, $13, $14, $15, $16, $17, $18, $19, $20, $21, $22, $23, $24, $25, $27, $28, $29, $30)\n ON CONFLICT (id) DO UPDATE SET success = $7, result = $11, logs = concat(cj.logs, $12) RETURNING duration_ms", + "query": "INSERT INTO completed_job AS cj\n ( workspace_id\n , id\n , parent_job\n , created_by\n , created_at\n , started_at\n , duration_ms\n , success\n , script_hash\n , script_path\n , args\n , result\n , logs\n , raw_code\n , raw_lock\n , canceled\n , canceled_by\n , canceled_reason\n , job_kind\n , schedule_path\n , permissioned_as\n , flow_status\n , raw_flow\n , is_flow_step\n , is_skipped\n , language\n , email\n , visible_to_owner\n , mem_peak\n , tag\n , priority\n )\n VALUES ($1, $2, $3, $4, $5, COALESCE($6, now()), COALESCE($26, (EXTRACT('epoch' FROM (now())) - EXTRACT('epoch' FROM (COALESCE($6, now()))))*1000), $7, $8, $9,$10, $11, $12, $13, $14, $15, $16, $17, $18, $19, $20, $21, $22, $23, $24, $25, $27, $28, $29, $30, $31)\n ON CONFLICT (id) DO UPDATE SET success = $7, result = $11, logs = concat(cj.logs, $12) RETURNING duration_ms", "describe": { "columns": [ { @@ -67,7 +67,6 @@ "bash", "postgresql", "nativets", - "Nativets", "bun", "mysql", "bigquery", @@ -82,12 +81,13 @@ "Varchar", "Bool", "Int4", - "Varchar" + "Varchar", + "Int2" ] }, "nullable": [ false ] }, - "hash": "25bef6a248f3ee0ea2cbcc376c217cbcf1013ae311c36b42d423bf6a02fa016c" + "hash": "254776959f60455a00a2c29a1783ee6b27c836197df6454d3bb5c59441c34f44" } diff --git a/backend/.sqlx/query-b5646d44502b289e5d92f53f74bf0a52b8bf76775d87a1f698ed2a3c3446c86e.json b/backend/.sqlx/query-52ad1773a9f5a1b6be1bb16af408f051f27d3f21b615c468b85aa75b64d943db.json similarity index 84% rename from backend/.sqlx/query-b5646d44502b289e5d92f53f74bf0a52b8bf76775d87a1f698ed2a3c3446c86e.json rename to backend/.sqlx/query-52ad1773a9f5a1b6be1bb16af408f051f27d3f21b615c468b85aa75b64d943db.json index 7aaace0737..4df35d6cf7 100644 --- a/backend/.sqlx/query-b5646d44502b289e5d92f53f74bf0a52b8bf76775d87a1f698ed2a3c3446c86e.json +++ b/backend/.sqlx/query-52ad1773a9f5a1b6be1bb16af408f051f27d3f21b615c468b85aa75b64d943db.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "INSERT INTO script (workspace_id, hash, path, parent_hashes, summary, description, content, created_by, schema, is_template, extra_perms, lock, language, kind, tag, draft_only, envs, concurrent_limit, concurrency_time_window_s, cache_ttl, dedicated_worker, ws_error_handler_muted) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9::text::json, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19, $20, $21, $22)", + "query": "INSERT INTO script (workspace_id, hash, path, parent_hashes, summary, description, content, created_by, schema, is_template, extra_perms, lock, language, kind, tag, draft_only, envs, concurrent_limit, concurrency_time_window_s, cache_ttl, dedicated_worker, ws_error_handler_muted, priority) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9::text::json, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19, $20, $21, $22, $23)", "describe": { "columns": [], "parameters": { @@ -28,7 +28,6 @@ "bash", "postgresql", "nativets", - "Nativets", "bun", "mysql", "bigquery", @@ -60,10 +59,11 @@ "Int4", "Int4", "Bool", - "Bool" + "Bool", + "Int2" ] }, "nullable": [] }, - "hash": "b5646d44502b289e5d92f53f74bf0a52b8bf76775d87a1f698ed2a3c3446c86e" + "hash": "52ad1773a9f5a1b6be1bb16af408f051f27d3f21b615c468b85aa75b64d943db" } diff --git a/backend/.sqlx/query-9e8c3ff3d6b31e366e15beda1e96e03e870ccc3b353401439bc0ed8ff219249b.json b/backend/.sqlx/query-620ddf29c5e867079df4c2aa6e80bccb19beeb9ddfa308ca97f254cd5ba8157e.json similarity index 84% rename from backend/.sqlx/query-9e8c3ff3d6b31e366e15beda1e96e03e870ccc3b353401439bc0ed8ff219249b.json rename to backend/.sqlx/query-620ddf29c5e867079df4c2aa6e80bccb19beeb9ddfa308ca97f254cd5ba8157e.json index 6308bf3bb2..e6282c7e80 100644 --- a/backend/.sqlx/query-9e8c3ff3d6b31e366e15beda1e96e03e870ccc3b353401439bc0ed8ff219249b.json +++ b/backend/.sqlx/query-620ddf29c5e867079df4c2aa6e80bccb19beeb9ddfa308ca97f254cd5ba8157e.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "INSERT INTO queue\n (workspace_id, id, running, parent_job, created_by, permissioned_as, scheduled_for, \n script_hash, script_path, raw_code, raw_lock, args, job_kind, schedule_path, raw_flow, flow_status, is_flow_step, language, started_at, same_worker, pre_run_error, email, visible_to_owner, root_job, tag, concurrent_limit, concurrency_time_window_s, timeout, flow_step_id, cache_ttl)\n VALUES ($1, $2, $3, $4, $5, $6, COALESCE($7, now()), $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, CASE WHEN $3 THEN now() END, $19, $20, $21, $22, $23, $24, $25, $26, $27, $28, $29) RETURNING id", + "query": "INSERT INTO queue\n (workspace_id, id, running, parent_job, created_by, permissioned_as, scheduled_for, \n script_hash, script_path, raw_code, raw_lock, args, job_kind, schedule_path, raw_flow, flow_status, is_flow_step, language, started_at, same_worker, pre_run_error, email, visible_to_owner, root_job, tag, concurrent_limit, concurrency_time_window_s, timeout, flow_step_id, cache_ttl, priority)\n VALUES ($1, $2, $3, $4, $5, $6, COALESCE($7, now()), $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, CASE WHEN $3 THEN now() END, $19, $20, $21, $22, $23, $24, $25, $26, $27, $28, $29, $30) RETURNING id", "describe": { "columns": [ { @@ -60,7 +60,6 @@ "bash", "postgresql", "nativets", - "Nativets", "bun", "mysql", "bigquery", @@ -81,12 +80,13 @@ "Int4", "Int4", "Varchar", - "Int4" + "Int4", + "Int2" ] }, "nullable": [ false ] }, - "hash": "9e8c3ff3d6b31e366e15beda1e96e03e870ccc3b353401439bc0ed8ff219249b" + "hash": "620ddf29c5e867079df4c2aa6e80bccb19beeb9ddfa308ca97f254cd5ba8157e" } diff --git a/backend/.sqlx/query-6b9ff3fbca9e825c95d14705082a10de88172c0c748a45aba4d2d03c3b58f54d.json b/backend/.sqlx/query-6b9ff3fbca9e825c95d14705082a10de88172c0c748a45aba4d2d03c3b58f54d.json index dabac76bfe..73e6e17c76 100644 --- a/backend/.sqlx/query-6b9ff3fbca9e825c95d14705082a10de88172c0c748a45aba4d2d03c3b58f54d.json +++ b/backend/.sqlx/query-6b9ff3fbca9e825c95d14705082a10de88172c0c748a45aba4d2d03c3b58f54d.json @@ -46,7 +46,6 @@ "bash", "postgresql", "nativets", - "Nativets", "bun", "mysql", "bigquery", diff --git a/backend/.sqlx/query-9d3556319411a27a875bf6cf0e5eda837cc63e4d8be912c0b5bfeea4a0c8db2e.json b/backend/.sqlx/query-9d3556319411a27a875bf6cf0e5eda837cc63e4d8be912c0b5bfeea4a0c8db2e.json index 8b139967a6..61da7601af 100644 --- a/backend/.sqlx/query-9d3556319411a27a875bf6cf0e5eda837cc63e4d8be912c0b5bfeea4a0c8db2e.json +++ b/backend/.sqlx/query-9d3556319411a27a875bf6cf0e5eda837cc63e4d8be912c0b5bfeea4a0c8db2e.json @@ -40,7 +40,6 @@ "bash", "postgresql", "nativets", - "Nativets", "bun", "mysql", "bigquery", diff --git a/backend/.sqlx/query-b224cdd1221fc9e7227ef8e8c025eedc09bceb82bca349f3e31c8513ebbf0192.json b/backend/.sqlx/query-b6074eb9369de071f79165bf506fbc44f1688035e1928b4c3649d875187b1e1d.json similarity index 78% rename from backend/.sqlx/query-b224cdd1221fc9e7227ef8e8c025eedc09bceb82bca349f3e31c8513ebbf0192.json rename to backend/.sqlx/query-b6074eb9369de071f79165bf506fbc44f1688035e1928b4c3649d875187b1e1d.json index 38f81da395..40ab7503e5 100644 --- a/backend/.sqlx/query-b224cdd1221fc9e7227ef8e8c025eedc09bceb82bca349f3e31c8513ebbf0192.json +++ b/backend/.sqlx/query-b6074eb9369de071f79165bf506fbc44f1688035e1928b4c3649d875187b1e1d.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "select hash, tag, concurrent_limit, concurrency_time_window_s, cache_ttl, language as \"language: ScriptLang\", dedicated_worker from script where path = $1 AND workspace_id = $2 AND\n created_at = (SELECT max(created_at) FROM script WHERE path = $1 AND workspace_id = $2 AND\n deleted = false AND lock IS not NULL AND lock_error_logs IS NULL)", + "query": "select hash, tag, concurrent_limit, concurrency_time_window_s, cache_ttl, language as \"language: ScriptLang\", dedicated_worker, priority from script where path = $1 AND workspace_id = $2 AND\n created_at = (SELECT max(created_at) FROM script WHERE path = $1 AND workspace_id = $2 AND\n deleted = false AND lock IS not NULL AND lock_error_logs IS NULL)", "describe": { "columns": [ { @@ -42,7 +42,6 @@ "bash", "postgresql", "nativets", - "Nativets", "bun", "mysql", "bigquery", @@ -58,6 +57,11 @@ "ordinal": 6, "name": "dedicated_worker", "type_info": "Bool" + }, + { + "ordinal": 7, + "name": "priority", + "type_info": "Int2" } ], "parameters": { @@ -73,8 +77,9 @@ true, true, false, + true, true ] }, - "hash": "b224cdd1221fc9e7227ef8e8c025eedc09bceb82bca349f3e31c8513ebbf0192" + "hash": "b6074eb9369de071f79165bf506fbc44f1688035e1928b4c3649d875187b1e1d" } diff --git a/backend/.sqlx/query-251f18ccdd89255023258a33285757801cabe4e86385e2dea10b666d1c6f2b61.json b/backend/.sqlx/query-c7febee7146a28481ce36f3570bceb38751a6527827525a582d9f8804981f552.json similarity index 79% rename from backend/.sqlx/query-251f18ccdd89255023258a33285757801cabe4e86385e2dea10b666d1c6f2b61.json rename to backend/.sqlx/query-c7febee7146a28481ce36f3570bceb38751a6527827525a582d9f8804981f552.json index 26f33c8751..3b43ee29f5 100644 --- a/backend/.sqlx/query-251f18ccdd89255023258a33285757801cabe4e86385e2dea10b666d1c6f2b61.json +++ b/backend/.sqlx/query-c7febee7146a28481ce36f3570bceb38751a6527827525a582d9f8804981f552.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "select hash, tag, concurrent_limit, concurrency_time_window_s, cache_ttl, language as \"language: ScriptLang\", dedicated_worker from script where path = $1 AND workspace_id = $2 AND\n created_at = (SELECT max(created_at) FROM script WHERE path = $1 AND workspace_id = $2 AND\n deleted = false AND archived = false)", + "query": "select hash, tag, concurrent_limit, concurrency_time_window_s, cache_ttl, language as \"language: ScriptLang\", dedicated_worker, priority from script where path = $1 AND workspace_id = $2 AND\n created_at = (SELECT max(created_at) FROM script WHERE path = $1 AND workspace_id = $2 AND\n deleted = false AND archived = false)", "describe": { "columns": [ { @@ -42,7 +42,6 @@ "bash", "postgresql", "nativets", - "Nativets", "bun", "mysql", "bigquery", @@ -58,6 +57,11 @@ "ordinal": 6, "name": "dedicated_worker", "type_info": "Bool" + }, + { + "ordinal": 7, + "name": "priority", + "type_info": "Int2" } ], "parameters": { @@ -73,8 +77,9 @@ true, true, false, + true, true ] }, - "hash": "251f18ccdd89255023258a33285757801cabe4e86385e2dea10b666d1c6f2b61" + "hash": "c7febee7146a28481ce36f3570bceb38751a6527827525a582d9f8804981f552" } diff --git a/backend/.sqlx/query-901c430cba96e5cab1a186207ea708f3d34a04c8ab03b385f8f2e9c662d3e724.json b/backend/.sqlx/query-d48070ef889723f408ed6c1a97b5ab9f0e98340cd33c6e29824f6030f0089071.json similarity index 85% rename from backend/.sqlx/query-901c430cba96e5cab1a186207ea708f3d34a04c8ab03b385f8f2e9c662d3e724.json rename to backend/.sqlx/query-d48070ef889723f408ed6c1a97b5ab9f0e98340cd33c6e29824f6030f0089071.json index 8c75467990..c0c9ef873f 100644 --- a/backend/.sqlx/query-901c430cba96e5cab1a186207ea708f3d34a04c8ab03b385f8f2e9c662d3e724.json +++ b/backend/.sqlx/query-d48070ef889723f408ed6c1a97b5ab9f0e98340cd33c6e29824f6030f0089071.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "select tag, concurrent_limit, concurrency_time_window_s, cache_ttl, language as \"language: ScriptLang\", dedicated_worker from script where hash = $1 AND workspace_id = $2", + "query": "select tag, concurrent_limit, concurrency_time_window_s, cache_ttl, language as \"language: ScriptLang\", dedicated_worker, priority from script where hash = $1 AND workspace_id = $2", "describe": { "columns": [ { @@ -37,7 +37,6 @@ "bash", "postgresql", "nativets", - "Nativets", "bun", "mysql", "bigquery", @@ -53,6 +52,11 @@ "ordinal": 5, "name": "dedicated_worker", "type_info": "Bool" + }, + { + "ordinal": 6, + "name": "priority", + "type_info": "Int2" } ], "parameters": { @@ -67,8 +71,9 @@ true, true, false, + true, true ] }, - "hash": "901c430cba96e5cab1a186207ea708f3d34a04c8ab03b385f8f2e9c662d3e724" + "hash": "d48070ef889723f408ed6c1a97b5ab9f0e98340cd33c6e29824f6030f0089071" } diff --git a/backend/.sqlx/query-f056b5f3e66a764748925f1bfd3180923fde8c7fdf69088d0e4a5555cc049545.json b/backend/.sqlx/query-f056b5f3e66a764748925f1bfd3180923fde8c7fdf69088d0e4a5555cc049545.json deleted file mode 100644 index 37a0f80de4..0000000000 --- a/backend/.sqlx/query-f056b5f3e66a764748925f1bfd3180923fde8c7fdf69088d0e4a5555cc049545.json +++ /dev/null @@ -1,23 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "SELECT result FROM completed_job WHERE id = $1 AND workspace_id = $2", - "describe": { - "columns": [ - { - "ordinal": 0, - "name": "result", - "type_info": "Jsonb" - } - ], - "parameters": { - "Left": [ - "Uuid", - "Text" - ] - }, - "nullable": [ - true - ] - }, - "hash": "f056b5f3e66a764748925f1bfd3180923fde8c7fdf69088d0e4a5555cc049545" -} diff --git a/backend/migrations/20231020101153_queue_priority_column.down.sql b/backend/migrations/20231020101153_queue_priority_column.down.sql new file mode 100644 index 0000000000..d577c3ae27 --- /dev/null +++ b/backend/migrations/20231020101153_queue_priority_column.down.sql @@ -0,0 +1,9 @@ +-- Add up migration script here +ALTER TABLE queue DROP COLUMN priority; +DROP INDEX IF EXISTS queue_sort; +CREATE INDEX IF NOT EXISTS queue_sort ON queue (scheduled_for, created_at, tag) WHERE running = false; +DROP INDEX IF EXISTS queue_suspended; +CREATE INDEX IF NOT EXISTS queue_suspended ON queue (created_at, suspend_until, suspend, tag) WHERE suspend_until is not null; + +ALTER TABLE completed_job DROP COLUMN priority; +ALTER TABLE script DROP COLUMN priority; diff --git a/backend/migrations/20231020101153_queue_priority_column.up.sql b/backend/migrations/20231020101153_queue_priority_column.up.sql new file mode 100644 index 0000000000..81558f1607 --- /dev/null +++ b/backend/migrations/20231020101153_queue_priority_column.up.sql @@ -0,0 +1,9 @@ +-- Add up migration script here +ALTER TABLE queue ADD COLUMN priority SMALLINT; +DROP INDEX IF EXISTS queue_sort; +CREATE INDEX IF NOT EXISTS queue_sort ON queue (priority DESC NULLS LAST, scheduled_for, created_at, tag) WHERE running = false; +DROP INDEX IF EXISTS queue_suspended; +CREATE INDEX IF NOT EXISTS queue_suspended ON queue (priority DESC NULLS LAST, created_at, suspend_until, suspend, tag) WHERE suspend_until is not null; + +ALTER TABLE completed_job ADD COLUMN priority SMALLINT; +ALTER TABLE script ADD COLUMN priority SMALLINT; diff --git a/backend/tests/worker.rs b/backend/tests/worker.rs index 8be999dadb..0474a6a665 100644 --- a/backend/tests/worker.rs +++ b/backend/tests/worker.rs @@ -894,6 +894,7 @@ impl RunJob { None, None, None, + None, ) .await .expect("push has to succeed"); @@ -1080,6 +1081,7 @@ async fn test_deno_flow(db: Pool) { cache_ttl: None, mock: None, timeout: None, + priority: None, }, FlowModule { id: "b".to_string(), @@ -1114,6 +1116,7 @@ async fn test_deno_flow(db: Pool) { cache_ttl: None, mock: None, timeout: None, + priority: None, }], }, stop_after_if: Default::default(), @@ -1124,6 +1127,7 @@ async fn test_deno_flow(db: Pool) { cache_ttl: None, mock: None, timeout: None, + priority: None, }, ], same_worker: false, @@ -1223,7 +1227,7 @@ async fn test_deno_flow_same_worker(db: Pool) { cache_ttl: None, mock: None, timeout: None, - + priority: None, }, FlowModule { id: "b".to_string(), @@ -1269,6 +1273,7 @@ async fn test_deno_flow_same_worker(db: Pool) { cache_ttl: None, mock: None, timeout: None, + priority: None, }, FlowModule { id: "e".to_string(), @@ -1300,7 +1305,7 @@ async fn test_deno_flow_same_worker(db: Pool) { cache_ttl: None, mock: None, timeout: None, - + priority: None, }, ], }, @@ -1312,7 +1317,7 @@ async fn test_deno_flow_same_worker(db: Pool) { cache_ttl: None, mock: None, timeout: None, - + priority: None, }, FlowModule { id: "c".to_string(), @@ -1352,7 +1357,7 @@ async fn test_deno_flow_same_worker(db: Pool) { cache_ttl: None, mock: None, timeout: None, - + priority: None, }, ], same_worker: true, @@ -2565,6 +2570,7 @@ async fn test_flow_lock_all(db: Pool) { path: "g/all/flow_lock_all".to_owned(), tag: None, ws_error_handler_muted: None, + priority: None, }, draft_only: None, }, @@ -2919,6 +2925,7 @@ async fn run_deployed_relative_imports(db: &Pool, script_content: Stri tag: None, schema: std::collections::HashMap::new(), ws_error_handler_muted: Some(false), + priority: None, }, ).await.unwrap(); @@ -2941,6 +2948,7 @@ async fn run_deployed_relative_imports(db: &Pool, script_content: Stri cache_ttl: None, dedicated_worker: None, language, + priority: None, }).push(&db2).await; diff --git a/backend/windmill-api/openapi-deref.yaml b/backend/windmill-api/openapi-deref.yaml index b2be6994aa..ebd60169c6 100644 --- a/backend/windmill-api/openapi-deref.yaml +++ b/backend/windmill-api/openapi-deref.yaml @@ -3457,6 +3457,8 @@ paths: user_groups_required: oneOf: *ref_23 discriminator: *ref_24 + priority: + type: number retry: type: object properties: &ref_175 @@ -3491,6 +3493,8 @@ paths: type: string cache_ttl: type: number + priority: + type: number required: &ref_48 - modules schema: @@ -3990,6 +3994,8 @@ paths: type: boolean ws_error_handler_muted: type: boolean + priority: + type: integer required: &ref_30 - hash - path @@ -4176,6 +4182,8 @@ paths: type: boolean ws_error_handler_muted: type: boolean + priority: + type: integer required: &ref_33 - path - summary @@ -5050,6 +5058,8 @@ paths: type: string ws_error_handler_muted: type: boolean + priority: + type: integer required: &ref_143 - path - edited_by @@ -5195,6 +5205,8 @@ paths: type: string ws_error_handler_muted: type: boolean + priority: + type: integer required: - path - type: object @@ -6641,6 +6653,8 @@ paths: type: integer tag: type: string + priority: + type: integer required: &ref_68 - id - running @@ -6908,6 +6922,8 @@ paths: type: integer tag: type: string + priority: + type: integer required: &ref_66 - id - created_by diff --git a/backend/windmill-api/openapi.yaml b/backend/windmill-api/openapi.yaml index 22143f9859..bf31128566 100644 --- a/backend/windmill-api/openapi.yaml +++ b/backend/windmill-api/openapi.yaml @@ -6503,6 +6503,8 @@ components: type: boolean ws_error_handler_muted: type: boolean + priority: + type: integer required: - hash - path @@ -6582,6 +6584,8 @@ components: type: boolean ws_error_handler_muted: type: boolean + priority: + type: integer required: - path - summary @@ -6756,6 +6760,8 @@ components: type: integer tag: type: string + priority: + type: integer required: - id - running @@ -6863,6 +6869,8 @@ components: type: integer tag: type: string + priority: + type: integer required: - id - created_by @@ -7829,6 +7837,8 @@ components: type: string ws_error_handler_muted: type: boolean + priority: + type: integer required: - path - edited_by @@ -7847,6 +7857,8 @@ components: type: string ws_error_handler_muted: type: boolean + priority: + type: integer required: - path diff --git a/backend/windmill-api/src/apps.rs b/backend/windmill-api/src/apps.rs index 55945c5299..c9a548a9a0 100644 --- a/backend/windmill-api/src/apps.rs +++ b/backend/windmill-api/src/apps.rs @@ -534,6 +534,7 @@ async fn create_app( None, None, None, + None, ) .await?; tracing::info!("Pushed app dependency job {}", dependency_job_uuid); @@ -777,6 +778,7 @@ async fn update_app( None, None, None, + None, ) .await?; tracing::info!("Pushed app dependency job {}", dependency_job_uuid); @@ -938,6 +940,7 @@ async fn execute_component( tag, None, None, + None, ) .await?; tx.commit().await?; diff --git a/backend/windmill-api/src/flows.rs b/backend/windmill-api/src/flows.rs index dcbcda5541..8de121aec9 100644 --- a/backend/windmill-api/src/flows.rs +++ b/backend/windmill-api/src/flows.rs @@ -225,7 +225,7 @@ async fn toggle_workspace_error_handler( Json(req): Json, ) -> Result { #[cfg(not(feature = "enterprise"))] - if true { + { return Err(Error::BadRequest( "Muting the error handler for certain flow is only available in enterprise version" .to_string(), @@ -375,6 +375,7 @@ async fn create_flow( nf.tag, None, None, + None, ) .await?; @@ -580,6 +581,7 @@ async fn update_flow( None, None, None, + None, ) .await?; sqlx::query!( @@ -805,6 +807,7 @@ mod tests { cache_ttl: None, mock: None, timeout: None, + priority: None, }, FlowModule { id: "b".to_string(), @@ -829,6 +832,7 @@ mod tests { cache_ttl: None, mock: None, timeout: None, + priority: None, }, FlowModule { id: "c".to_string(), @@ -850,6 +854,7 @@ mod tests { cache_ttl: None, mock: None, timeout: None, + priority: None, }, ], failure_module: Some(FlowModule { @@ -870,6 +875,7 @@ mod tests { cache_ttl: None, mock: None, timeout: None, + priority: None, }), same_worker: false, concurrent_limit: None, @@ -877,6 +883,7 @@ mod tests { skip_expr: None, cache_ttl: None, ws_error_handler_muted: None, + priority: None, }; let expect = serde_json::json!({ "modules": [ diff --git a/backend/windmill-api/src/jobs.rs b/backend/windmill-api/src/jobs.rs index f2faf9d01a..4cfc4cb744 100644 --- a/backend/windmill-api/src/jobs.rs +++ b/backend/windmill-api/src/jobs.rs @@ -315,9 +315,10 @@ pub async fn get_path_tag_limits_cache_for_hash( Option, ScriptLang, Option, + Option, )> { let script = sqlx::query!( - "select path, tag, concurrent_limit, concurrency_time_window_s, cache_ttl, language as \"language: ScriptLang\", dedicated_worker from script where hash = $1 AND workspace_id = $2", + "select path, tag, concurrent_limit, concurrency_time_window_s, cache_ttl, language as \"language: ScriptLang\", dedicated_worker, priority from script where hash = $1 AND workspace_id = $2", hash, w_id ) @@ -336,6 +337,7 @@ pub async fn get_path_tag_limits_cache_for_hash( script.cache_ttl, script.language, script.dedicated_worker, + script.priority, )) } @@ -352,7 +354,7 @@ async fn get_job_internal(db: &DB, workspace_id: &str, job_id: Uuid) -> error::R id, workspace_id, parent_job, created_by, created_at, duration_ms, success, script_hash, script_path, CASE WHEN pg_column_size(args) < 2000000 THEN args ELSE '{\"reason\": \"WINDMILL_TOO_BIG\"}'::jsonb END as args, CASE WHEN pg_column_size(result) < 2000000 THEN result ELSE '\"WINDMILL_TOO_BIG\"'::jsonb END as result, logs, deleted, raw_code, canceled, canceled_by, canceled_reason, job_kind, env_id, schedule_path, permissioned_as, flow_status, raw_flow, is_flow_step, language, started_at, is_skipped, - raw_lock, email, visible_to_owner, mem_peak, tag + raw_lock, email, visible_to_owner, mem_peak, tag, priority FROM completed_job WHERE id = $1 AND workspace_id = $2") .bind(job_id) .bind(workspace_id) @@ -367,7 +369,7 @@ async fn get_job_internal(db: &DB, workspace_id: &str, job_id: Uuid) -> error::R script_hash, script_path, CASE WHEN pg_column_size(args) < 2000000 THEN args ELSE '{\"reason\": \"WINDMILL_TOO_BIG\"}'::jsonb END as args, logs, raw_code, canceled, canceled_by, canceled_reason, last_ping, job_kind, env_id, schedule_path, permissioned_as, flow_status, raw_flow, is_flow_step, language, suspend, suspend_until, same_worker, raw_lock, pre_run_error, email, visible_to_owner, mem_peak, - root_job, leaf_jobs, tag, concurrent_limit, concurrency_time_window_s, timeout, flow_step_id, cache_ttl + root_job, leaf_jobs, tag, concurrent_limit, concurrency_time_window_s, timeout, flow_step_id, cache_ttl, priority FROM queue WHERE id = $1 AND workspace_id = $2", ) .bind(job_id) @@ -441,6 +443,8 @@ pub struct CompletedJob { #[serde(skip_serializing_if = "Option::is_none")] pub mem_peak: Option, pub tag: String, + #[serde(skip_serializing_if = "Option::is_none")] + pub priority: Option, } impl CompletedJob { @@ -493,6 +497,8 @@ pub struct ListableCompletedJob { #[serde(skip_serializing_if = "Option::is_none")] pub mem_peak: Option, pub tag: String, + #[serde(skip_serializing_if = "Option::is_none")] + pub priority: Option, } impl<'a> IntoResponse for CompletedJob { @@ -646,6 +652,7 @@ struct ListableQueuedJob { pub email: String, pub suspend: Option, pub tag: String, + pub priority: Option, } async fn list_queue_jobs( @@ -789,6 +796,7 @@ async fn list_jobs( "tag", "null as concurrent_limit", "null as concurrency_time_window_s", + "priority", ], )) } else { @@ -849,6 +857,7 @@ async fn list_jobs( "tag", "concurrent_limit", "concurrency_time_window_s", + "priority", ], ); @@ -1247,7 +1256,7 @@ fn conditionally_require_authed_user( .to_string(), )); #[cfg(feature = "enterprise")] - if true { + { for required_group in approval_conditions.user_groups_required.iter() { if authed.as_ref().unwrap().groups.contains(&required_group) { return Ok(()); @@ -1407,6 +1416,7 @@ struct UnifiedJob { tag: String, concurrent_limit: Option, concurrency_time_window_s: Option, + priority: Option, } impl<'a> From for Job { @@ -1443,6 +1453,7 @@ impl<'a> From for Job { visible_to_owner: uj.visible_to_owner, mem_peak: uj.mem_peak, tag: uj.tag, + priority: uj.priority, }), "QueuedJob" => Job::QueuedJob(QueuedJob { workspace_id: uj.workspace_id, @@ -1484,6 +1495,7 @@ impl<'a> From for Job { timeout: None, flow_step_id: None, cache_ttl: None, + priority: uj.priority, }), t => panic!("job type {} not valid", t), } @@ -1666,6 +1678,7 @@ pub async fn run_flow_by_path( tag, None, None, + None, ) .await?; tx.commit().await?; @@ -1715,6 +1728,7 @@ pub async fn run_job_by_path( tag, None, None, + None, ) .await?; tx.commit().await?; @@ -1928,6 +1942,7 @@ pub async fn run_wait_result_job_by_path_get( tag, None, None, + None, ) .await?; tx.commit().await?; @@ -2036,6 +2051,7 @@ async fn run_wait_result_script_by_path_internal( tag, None, None, + None, ) .await?; tx.commit().await?; @@ -2066,6 +2082,7 @@ pub async fn run_wait_result_script_by_hash( cache_ttl, language, dedicated_worker, + priority, ) = get_path_tag_limits_cache_for_hash(&db, &w_id, hash).await?; check_scopes(&authed, || format!("run:script/{path}"))?; @@ -2084,6 +2101,7 @@ pub async fn run_wait_result_script_by_hash( cache_ttl, language, dedicated_worker, + priority, }, args, &authed.username, @@ -2101,6 +2119,7 @@ pub async fn run_wait_result_script_by_hash( tag, None, None, + None, ) .await?; tx.commit().await?; @@ -2175,6 +2194,7 @@ async fn run_wait_result_flow_by_path_internal( tag, None, None, + None, ) .await?; tx.commit().await?; @@ -2237,6 +2257,7 @@ async fn run_preview_job( preview.tag, None, None, + None, ) .await?; tx.commit().await?; @@ -2315,6 +2336,7 @@ async fn add_batch_jobs( None, None, None, + None, ) .await?; tx = PushIsolationLevel::Transaction(ntx); @@ -2415,6 +2437,7 @@ async fn run_preview_flow_job( raw_flow.tag, None, None, + None, ) .await?; tx.commit().await?; @@ -2444,6 +2467,7 @@ pub async fn run_job_by_hash( cache_ttl, language, dedicated_worker, + priority, ) = get_path_tag_limits_cache_for_hash(&db, &w_id, hash).await?; check_scopes(&authed, || format!("run:script/{path}"))?; @@ -2464,6 +2488,7 @@ pub async fn run_job_by_hash( cache_ttl, language, dedicated_worker, + priority, }, args, &authed.username, @@ -2481,6 +2506,7 @@ pub async fn run_job_by_hash( tag, None, None, + None, ) .await?; tx.commit().await?; @@ -2696,6 +2722,7 @@ async fn list_completed_jobs( "visible_to_owner", "mem_peak", "tag", + "priority", "'CompletedJob' as type", ], ) @@ -2713,7 +2740,7 @@ async fn get_completed_job<'a>( let job_o = sqlx::query("SELECT id, workspace_id, parent_job, created_by, created_at, duration_ms, success, script_hash, script_path, CASE WHEN pg_column_size(args) < 2000000 THEN args ELSE '\"WINDMILL_TOO_BIG\"'::jsonb END as args, CASE WHEN pg_column_size(result) < 2000000 THEN result ELSE '\"WINDMILL_TOO_BIG\"'::jsonb END as result, logs, deleted, raw_code, canceled, canceled_by, canceled_reason, job_kind, env_id, schedule_path, permissioned_as, flow_status, raw_flow, is_flow_step, language, started_at, is_skipped, - raw_lock, email, visible_to_owner, mem_peak, tag FROM completed_job WHERE id = $1 AND workspace_id = $2") + raw_lock, email, visible_to_owner, mem_peak, tag, priority FROM completed_job WHERE id = $1 AND workspace_id = $2") .bind(id) .bind(w_id) .fetch_optional(&db) diff --git a/backend/windmill-api/src/oauth2.rs b/backend/windmill-api/src/oauth2.rs index eceef64038..5162d70f7d 100644 --- a/backend/windmill-api/src/oauth2.rs +++ b/backend/windmill-api/src/oauth2.rs @@ -871,6 +871,7 @@ async fn slack_command( cache_ttl, language, dedicated_worker, + priority, ) = windmill_common::get_latest_deployed_hash_for_path( &db, &settings.workspace_id, @@ -886,6 +887,7 @@ async fn slack_command( cache_ttl, language, dedicated_worker, + priority, }, tag, ) @@ -919,6 +921,7 @@ async fn slack_command( tag, None, None, + None, ) .await?; let url = BASE_URL.read().await.clone(); diff --git a/backend/windmill-api/src/scripts.rs b/backend/windmill-api/src/scripts.rs index 91390eac87..eb45675804 100644 --- a/backend/windmill-api/src/scripts.rs +++ b/backend/windmill-api/src/scripts.rs @@ -77,7 +77,10 @@ pub struct ScriptWDraft { pub cache_ttl: Option, #[serde(skip_serializing_if = "Option::is_none")] pub dedicated_worker: Option, + #[serde(skip_serializing_if = "Option::is_none")] pub ws_error_handler_muted: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub priority: Option, } pub fn global_service() -> Router { @@ -459,8 +462,9 @@ async fn create_script( sqlx::query!( "INSERT INTO script (workspace_id, hash, path, parent_hashes, summary, description, \ content, created_by, schema, is_template, extra_perms, lock, language, kind, tag, \ - draft_only, envs, concurrent_limit, concurrency_time_window_s, cache_ttl, dedicated_worker, ws_error_handler_muted) \ - VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9::text::json, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19, $20, $21, $22)", + draft_only, envs, concurrent_limit, concurrency_time_window_s, cache_ttl, \ + dedicated_worker, ws_error_handler_muted, priority) \ + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9::text::json, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19, $20, $21, $22, $23)", &w_id, &hash.0, ns.path, @@ -483,6 +487,7 @@ async fn create_script( ns.cache_ttl, ns.dedicated_worker, ns.ws_error_handler_muted.unwrap_or(false), + ns.priority, ) .execute(&mut tx) .await?; @@ -614,6 +619,7 @@ async fn create_script( ns.tag, None, None, + None, ) .await?; tx = PushIsolationLevel::Transaction(new_tx); @@ -679,7 +685,7 @@ async fn get_script_by_path_w_draft( let mut tx = user_db.begin(&authed).await?; let script_o = sqlx::query_as::<_, ScriptWDraft>( - "SELECT hash, script.path, summary, description, content, language, kind, tag, schema, draft_only, envs, concurrent_limit, concurrency_time_window_s, cache_ttl, ws_error_handler_muted, draft.value as draft, dedicated_worker FROM script LEFT JOIN draft ON + "SELECT hash, script.path, summary, description, content, language, kind, tag, schema, draft_only, envs, concurrent_limit, concurrency_time_window_s, cache_ttl, ws_error_handler_muted, draft.value as draft, dedicated_worker, priority FROM script LEFT JOIN draft ON script.path = draft.path AND script.workspace_id = draft.workspace_id AND draft.typ = 'script' WHERE script.path = $1 AND script.workspace_id = $2 \ AND script.created_at = (SELECT max(created_at) FROM script WHERE path = $1 AND \ @@ -724,7 +730,7 @@ async fn toggle_workspace_error_handler( Json(req): Json, ) -> Result { #[cfg(not(feature = "enterprise"))] - if true { + { return Err(Error::BadRequest( "Muting the error handler for certain script is only available in enterprise version" .to_string(), diff --git a/backend/windmill-api/src/workspaces.rs b/backend/windmill-api/src/workspaces.rs index 5411caea61..a1422098d0 100644 --- a/backend/windmill-api/src/workspaces.rs +++ b/backend/windmill-api/src/workspaces.rs @@ -555,7 +555,8 @@ async fn run_slack_message_test_job( Some(json!(extra_args)), authed.username.as_str(), authed.email.as_str(), - username_to_permissioned_as(authed.username.as_str()) + username_to_permissioned_as(authed.username.as_str()), + None, // Note: we could mark it as high priority to return result quickly to the user ).await?; tx.commit().await?; diff --git a/backend/windmill-common/src/flows.rs b/backend/windmill-common/src/flows.rs index 5f6ba71445..62d8146fee 100644 --- a/backend/windmill-common/src/flows.rs +++ b/backend/windmill-common/src/flows.rs @@ -90,6 +90,9 @@ pub struct FlowValue { pub cache_ttl: Option, #[serde(skip_serializing_if = "Option::is_none")] pub ws_error_handler_muted: Option, + #[serde(skip_serializing_if = "Option::is_none")] + // Priority at the flow level + pub priority: Option, } #[derive(Deserialize, Serialize, Debug, Clone)] @@ -204,6 +207,9 @@ pub struct FlowModule { pub cache_ttl: Option, #[serde(skip_serializing_if = "Option::is_none")] pub timeout: Option, + #[serde(skip_serializing_if = "Option::is_none")] + // Priority at the flow step level + pub priority: Option, } impl FlowModule { diff --git a/backend/windmill-common/src/jobs.rs b/backend/windmill-common/src/jobs.rs index 22b48511cf..4257111f4d 100644 --- a/backend/windmill-common/src/jobs.rs +++ b/backend/windmill-common/src/jobs.rs @@ -95,6 +95,8 @@ pub struct QueuedJob { pub flow_step_id: Option, #[serde(skip_serializing_if = "Option::is_none")] pub cache_ttl: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub priority: Option, } impl QueuedJob { @@ -179,6 +181,7 @@ impl Default for QueuedJob { timeout: None, flow_step_id: None, cache_ttl: None, + priority: None, } } } @@ -196,6 +199,7 @@ pub enum JobPayload { cache_ttl: Option, dedicated_worker: Option, language: ScriptLang, + priority: Option, }, Code(RawCode), Dependencies { @@ -251,6 +255,7 @@ pub async fn script_path_to_payload( cache_ttl, language, dedicated_worker, + priority, ) = get_latest_deployed_hash_for_path(db, w_id, script_path).await?; ( JobPayload::ScriptHash { @@ -261,6 +266,7 @@ pub async fn script_path_to_payload( cache_ttl: cache_ttl, language, dedicated_worker, + priority, }, tag, ) @@ -279,9 +285,10 @@ pub async fn script_hash_to_tag_and_limits<'c>( Option, ScriptLang, Option, + Option, )> { let script = sqlx::query!( - "select tag, concurrent_limit, concurrency_time_window_s, cache_ttl, language as \"language: ScriptLang\", dedicated_worker from script where hash = $1 AND workspace_id = $2", + "select tag, concurrent_limit, concurrency_time_window_s, cache_ttl, language as \"language: ScriptLang\", dedicated_worker, priority from script where hash = $1 AND workspace_id = $2", script_hash.0, w_id ) @@ -299,6 +306,7 @@ pub async fn script_hash_to_tag_and_limits<'c>( script.cache_ttl, script.language, script.dedicated_worker, + script.priority, )) } diff --git a/backend/windmill-common/src/lib.rs b/backend/windmill-common/src/lib.rs index 3045ee932c..b2426a700f 100644 --- a/backend/windmill-common/src/lib.rs +++ b/backend/windmill-common/src/lib.rs @@ -198,9 +198,10 @@ pub async fn get_latest_deployed_hash_for_path( Option, ScriptLang, Option, + Option, )> { let r_o = sqlx::query!( - "select hash, tag, concurrent_limit, concurrency_time_window_s, cache_ttl, language as \"language: ScriptLang\", dedicated_worker from script where path = $1 AND workspace_id = $2 AND + "select hash, tag, concurrent_limit, concurrency_time_window_s, cache_ttl, language as \"language: ScriptLang\", dedicated_worker, priority from script where path = $1 AND workspace_id = $2 AND created_at = (SELECT max(created_at) FROM script WHERE path = $1 AND workspace_id = $2 AND deleted = false AND lock IS not NULL AND lock_error_logs IS NULL)", script_path, @@ -219,6 +220,7 @@ pub async fn get_latest_deployed_hash_for_path( script.cache_ttl, script.language, script.dedicated_worker, + script.priority, )) } @@ -234,9 +236,10 @@ pub async fn get_latest_hash_for_path<'c>( Option, ScriptLang, Option, + Option, )> { let r_o = sqlx::query!( - "select hash, tag, concurrent_limit, concurrency_time_window_s, cache_ttl, language as \"language: ScriptLang\", dedicated_worker from script where path = $1 AND workspace_id = $2 AND + "select hash, tag, concurrent_limit, concurrency_time_window_s, cache_ttl, language as \"language: ScriptLang\", dedicated_worker, priority from script where path = $1 AND workspace_id = $2 AND created_at = (SELECT max(created_at) FROM script WHERE path = $1 AND workspace_id = $2 AND deleted = false AND archived = false)", script_path, @@ -255,5 +258,6 @@ pub async fn get_latest_hash_for_path<'c>( script.cache_ttl, script.language, script.dedicated_worker, + script.priority, )) } diff --git a/backend/windmill-common/src/scripts.rs b/backend/windmill-common/src/scripts.rs index 5dad06b039..16f64818ef 100644 --- a/backend/windmill-common/src/scripts.rs +++ b/backend/windmill-common/src/scripts.rs @@ -162,7 +162,10 @@ pub struct Script { #[serde(skip_serializing_if = "Option::is_none")] pub concurrency_time_window_s: Option, pub dedicated_worker: Option, + #[serde(skip_serializing_if = "Option::is_none")] pub ws_error_handler_muted: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub priority: Option, } #[derive(Serialize)] @@ -220,6 +223,7 @@ pub struct NewScript { pub cache_ttl: Option, pub dedicated_worker: Option, pub ws_error_handler_muted: Option, + pub priority: Option, } #[derive(Deserialize)] diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs index 4eea26abd3..81b8c1225a 100644 --- a/backend/windmill-queue/src/jobs.rs +++ b/backend/windmill-queue/src/jobs.rs @@ -319,9 +319,10 @@ pub async fn add_completed_job< , visible_to_owner , mem_peak , tag + , priority ) VALUES ($1, $2, $3, $4, $5, COALESCE($6, now()), COALESCE($26, (EXTRACT('epoch' FROM (now())) - EXTRACT('epoch' FROM (COALESCE($6, now()))))*1000), $7, $8, $9,\ - $10, $11, $12, $13, $14, $15, $16, $17, $18, $19, $20, $21, $22, $23, $24, $25, $27, $28, $29, $30) + $10, $11, $12, $13, $14, $15, $16, $17, $18, $19, $20, $21, $22, $23, $24, $25, $27, $28, $29, $30, $31) ON CONFLICT (id) DO UPDATE SET success = $7, result = $11, logs = concat(cj.logs, $12) RETURNING duration_ms", queued_job.workspace_id, queued_job.id, @@ -353,6 +354,7 @@ pub async fn add_completed_job< queued_job.visible_to_owner, if mem_peak > 0 { Some(mem_peak) } else { None }, queued_job.tag, + queued_job.priority, ) .fetch_one(&mut tx) .await @@ -377,6 +379,7 @@ pub async fn add_completed_job< result, job_id, queued_job.started_at.unwrap_or(chrono::Utc::now()), + queued_job.priority, ) .await?; } @@ -531,6 +534,7 @@ pub async fn run_error_handler< tag, None, None, + None, ) .await?; tx.commit().await?; @@ -720,6 +724,7 @@ async fn apply_schedule_handlers< result: Json<&'a T>, job_id: Uuid, started_at: DateTime, + job_priority: Option, ) -> windmill_common::error::Result> { let schedule = get_schedule_opt(tx.transaction_mut(), w_id, schedule_path).await?; @@ -776,6 +781,7 @@ async fn apply_schedule_handlers< &schedule.email, &schedule_to_user(&schedule.path), username_to_permissioned_as(&schedule.edited_by), + job_priority, ) .await; @@ -894,6 +900,7 @@ pub async fn handle_on_failure< username: &str, email: &str, permissioned_as: String, + priority: Option, ) -> windmill_common::error::Result<(Uuid, QueueTransaction<'c, R>)> { let (payload, tag) = get_payload_tag_from_prefixed_path(on_failure_path, db, w_id).await?; @@ -938,6 +945,7 @@ pub async fn handle_on_failure< tag, None, None, + priority, ) .await?; tracing::info!( @@ -1031,6 +1039,7 @@ async fn handle_on_recovery< tag, None, None, + None, ) .await?; tracing::info!( @@ -1311,7 +1320,7 @@ async fn pull_single_job_and_mark_as_running_no_concurrency_limit< SELECT id FROM queue WHERE suspend_until IS NOT NULL AND (suspend <= 0 OR suspend_until <= now()) AND tag = ANY($1) - ORDER BY created_at + ORDER BY priority DESC NULLS LAST, created_at FOR UPDATE SKIP LOCKED LIMIT 1 ) @@ -1343,7 +1352,7 @@ async fn pull_single_job_and_mark_as_running_no_concurrency_limit< SELECT id FROM queue WHERE running = false AND scheduled_for <= now() AND tag = ANY($1) - ORDER BY scheduled_for, created_at + ORDER BY priority DESC NULLS LAST, scheduled_for, created_at FOR UPDATE SKIP LOCKED LIMIT 1 ) @@ -1700,6 +1709,7 @@ pub async fn push<'c, T: Serialize + Send + Sync, R: rsmq_async::RsmqConnection mut tag: Option, custom_timeout: Option, flow_step_id: Option, + priority_override: Option, ) -> Result<(Uuid, QueueTransaction<'c, R>), Error> { #[cfg(feature = "enterprise")] { @@ -1801,6 +1811,7 @@ pub async fn push<'c, T: Serialize + Send + Sync, R: rsmq_async::RsmqConnection concurrency_time_window_s, cache_ttl, dedicated_worker, + low_level_priority, ) = match job_payload { JobPayload::ScriptHash { hash, @@ -1810,6 +1821,7 @@ pub async fn push<'c, T: Serialize + Send + Sync, R: rsmq_async::RsmqConnection cache_ttl, language, dedicated_worker, + priority, } => ( Some(hash.0), Some(path), @@ -1821,6 +1833,7 @@ pub async fn push<'c, T: Serialize + Send + Sync, R: rsmq_async::RsmqConnection concurrency_time_window_s, cache_ttl, dedicated_worker, + priority, ), JobPayload::ScriptHub { path } => { ( @@ -1835,6 +1848,7 @@ pub async fn push<'c, T: Serialize + Send + Sync, R: rsmq_async::RsmqConnection None, None, None, + None, ) } JobPayload::Code(RawCode { @@ -1856,6 +1870,7 @@ pub async fn push<'c, T: Serialize + Send + Sync, R: rsmq_async::RsmqConnection concurrency_time_window_s, cache_ttl, None, + None, ), JobPayload::Dependencies { hash, dependencies, language, path } => ( Some(hash.0), @@ -1868,6 +1883,7 @@ pub async fn push<'c, T: Serialize + Send + Sync, R: rsmq_async::RsmqConnection None, None, None, + None, ), JobPayload::FlowDependencies { path } => { let value_json = fetch_scalar_isolated!( @@ -1895,6 +1911,7 @@ pub async fn push<'c, T: Serialize + Send + Sync, R: rsmq_async::RsmqConnection None, None, None, + None, ) } JobPayload::AppDependencies { path, version } => ( @@ -1908,6 +1925,7 @@ pub async fn push<'c, T: Serialize + Send + Sync, R: rsmq_async::RsmqConnection None, None, None, + None, ), JobPayload::RawFlow { value, path } => ( None, @@ -1920,6 +1938,7 @@ pub async fn push<'c, T: Serialize + Send + Sync, R: rsmq_async::RsmqConnection value.concurrency_time_window_s, value.cache_ttl.map(|x| x as i32), None, + value.priority, ), JobPayload::Flow(flow) => { let value_json = fetch_scalar_isolated!( @@ -1947,6 +1966,7 @@ pub async fn push<'c, T: Serialize + Send + Sync, R: rsmq_async::RsmqConnection value.concurrency_time_window_s, value.cache_ttl.map(|x| x as i32), None, + value.priority, ) } JobPayload::Identity => ( @@ -1960,6 +1980,7 @@ pub async fn push<'c, T: Serialize + Send + Sync, R: rsmq_async::RsmqConnection None, None, None, + None, ), JobPayload::Noop => ( None, @@ -1972,9 +1993,29 @@ pub async fn push<'c, T: Serialize + Send + Sync, R: rsmq_async::RsmqConnection None, None, None, + None, ), }; + let final_priority: Option; + #[cfg(not(feature = "enterprise"))] + { + // priority is only available on EE. Do not compute it on CE + final_priority = None; + } + #[cfg(feature = "enterprise")] + { + final_priority = if *CLOUD_HOSTED { + // for cloud hosted instance, priority queues is disabled + None + } else if priority_override.is_some() { + priority_override + } else { + // else it takes the priority defined at the script/flow level, if it's a script or flow + low_level_priority + }; // else it remains empty, i.e. no priority + } + let is_running = same_worker; if let Some(flow) = raw_flow.as_ref() { same_worker = same_worker || flow.same_worker; @@ -2015,6 +2056,7 @@ pub async fn push<'c, T: Serialize + Send + Sync, R: rsmq_async::RsmqConnection suspend: None, cache_ttl: None, timeout: None, + priority: None, }); raw_flow = Some(FlowValue { modules, ..flow.clone() }); } @@ -2092,8 +2134,8 @@ pub async fn push<'c, T: Serialize + Send + Sync, R: rsmq_async::RsmqConnection script_hash, script_path, raw_code, raw_lock, args, job_kind, schedule_path, raw_flow, \ flow_status, is_flow_step, language, started_at, same_worker, pre_run_error, email, \ visible_to_owner, root_job, tag, concurrent_limit, concurrency_time_window_s, timeout, \ - flow_step_id, cache_ttl) - VALUES ($1, $2, $3, $4, $5, $6, COALESCE($7, now()), $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, CASE WHEN $3 THEN now() END, $19, $20, $21, $22, $23, $24, $25, $26, $27, $28, $29) \ + flow_step_id, cache_ttl, priority) + VALUES ($1, $2, $3, $4, $5, $6, COALESCE($7, now()), $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, CASE WHEN $3 THEN now() END, $19, $20, $21, $22, $23, $24, $25, $26, $27, $28, $29, $30) \ RETURNING id", workspace_id, job_id, @@ -2124,10 +2166,12 @@ pub async fn push<'c, T: Serialize + Send + Sync, R: rsmq_async::RsmqConnection custom_timeout, flow_step_id, cache_ttl, + final_priority, ) .fetch_one(&mut tx) .await .map_err(|e| Error::InternalErr(format!("Could not insert into queue {job_id}: {e}")))?; + // 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. if *METRICS_ENABLED { QUEUE_PUSH_COUNT.inc(); diff --git a/backend/windmill-queue/src/schedule.rs b/backend/windmill-queue/src/schedule.rs index a2bb20468d..a32a74f4b8 100644 --- a/backend/windmill-queue/src/schedule.rs +++ b/backend/windmill-queue/src/schedule.rs @@ -82,6 +82,7 @@ pub async fn push_scheduled_job<'c, R: rsmq_async::RsmqConnection + Send + 'c>( cache_ttl, language, dedicated_worker, + priority, ) = windmill_common::get_latest_hash_for_path( tx.transaction_mut(), &schedule.workspace_id, @@ -97,6 +98,7 @@ pub async fn push_scheduled_job<'c, R: rsmq_async::RsmqConnection + Send + 'c>( cache_ttl: cache_ttl, dedicated_worker, language, + priority, }, tag, ) @@ -131,6 +133,7 @@ pub async fn push_scheduled_job<'c, R: rsmq_async::RsmqConnection + Send + 'c>( tag, None, None, + None, ) .await?; Ok(tx) // TODO: Bubble up pushed UUID from here diff --git a/backend/windmill-worker/src/worker.rs b/backend/windmill-worker/src/worker.rs index 81a66a2f84..ff0d1fc6c2 100644 --- a/backend/windmill-worker/src/worker.rs +++ b/backend/windmill-worker/src/worker.rs @@ -1414,6 +1414,7 @@ async fn queue_init_bash_maybe<'c, R: rsmq_async::RsmqConnection + Send + 'c>( None, None, None, + None, ) .await?; inner_tx.commit().await?; diff --git a/backend/windmill-worker/src/worker_flow.rs b/backend/windmill-worker/src/worker_flow.rs index c233b27305..26ab7186bc 100644 --- a/backend/windmill-worker/src/worker_flow.rs +++ b/backend/windmill-worker/src/worker_flow.rs @@ -1642,6 +1642,18 @@ async fn push_next_flow_job ContinuePayload::ForloopJobs { payload, .. } => payload.clone(), }; + // compute job-to-be-pushed priority + // The job definition itself might have its own priority, but as we're running + // it from a flow here, it inherits first the flow step priority and second the + // flow priority. + let new_job_priority_override = if module.priority.is_some() { + module.priority + } else if flow_job.priority.is_some() { + flow_job.priority + } else { + None + }; + let transform_inp; let args = match &next_status { NextStatus::AllFlowJobs { @@ -1742,6 +1754,7 @@ async fn push_next_flow_job }, module.timeout, Some(module.id.clone()), + new_job_priority_override, ) .await?; @@ -2176,6 +2189,7 @@ async fn compute_next_flow_transform( skip_expr: None, cache_ttl: None, ws_error_handler_muted: None, + priority: None, }, path: inner_path, }, @@ -2234,6 +2248,7 @@ async fn compute_next_flow_transform( skip_expr: None, cache_ttl: None, ws_error_handler_muted: None, + priority: None, }, path: Some(format!("{}/forloop", flow_job.script_path())), }, @@ -2326,6 +2341,7 @@ async fn compute_next_flow_transform( skip_expr: None, cache_ttl: None, ws_error_handler_muted: None, + priority: None, }, path: Some(format!( "{}/branchone-{}", @@ -2371,6 +2387,7 @@ async fn compute_next_flow_transform( skip_expr: None, cache_ttl: None, ws_error_handler_muted: None, + priority: None, }, path: Some(format!( "{}/branchall-{}", @@ -2433,6 +2450,7 @@ async fn compute_next_flow_transform( skip_expr: None, cache_ttl: None, ws_error_handler_muted: None, + priority: None, }, path: Some(format!( "{}/branchall-{}", @@ -2496,6 +2514,7 @@ async fn script_to_payload( cache_ttl, language, dedicated_worker, + priority, ) = script_hash_to_tag_and_limits(&hash, &mut tx, &flow_job.workspace_id).await?; ( JobPayload::ScriptHash { @@ -2506,6 +2525,7 @@ async fn script_to_payload( cache_ttl: module.cache_ttl.map(|x| x as i32).ok_or(cache_ttl).ok(), language, dedicated_worker, + priority, }, tag, ) diff --git a/frontend/src/lib/components/ScriptBuilder.svelte b/frontend/src/lib/components/ScriptBuilder.svelte index e72b9d0ad9..1077a8a66d 100644 --- a/frontend/src/lib/components/ScriptBuilder.svelte +++ b/frontend/src/lib/components/ScriptBuilder.svelte @@ -192,7 +192,8 @@ concurrent_limit: script.concurrent_limit, concurrency_time_window_s: script.concurrency_time_window_s, cache_ttl: script.cache_ttl, - ws_error_handler_muted: script.ws_error_handler_muted + ws_error_handler_muted: script.ws_error_handler_muted, + priority: script.priority } }) history.replaceState(history.state, '', `/scripts/edit/${script.path}`) @@ -236,7 +237,8 @@ concurrent_limit: script.concurrent_limit, concurrency_time_window_s: script.concurrency_time_window_s, cache_ttl: script.cache_ttl, - ws_error_handler_muted: script.ws_error_handler_muted + ws_error_handler_muted: script.ws_error_handler_muted, + priority: script.priority } }) } @@ -591,6 +593,51 @@ > + {#if !isCloudHosted()} +
+ 0} + on:change={() => { + if (script.priority) { + script.priority = undefined + } else { + script.priority = 100 + } + }} + options={{ + right: 'Label as high priority' + }} + > + + { + if (script.priority && script.priority > 100) { + script.priority = 100 + } else if (script.priority && script.priority < 0) { + script.priority = 0 + } + }} + /> + + + + + + Jobs from script labeled as high priority take precedence over the other + jobs when in the jobs queue. + {#if !$enterpriseLicense}This is a feature only available on enterprise + edition.{/if} + + +
+ {/if} {#if !isCloudHosted()}
diff --git a/frontend/src/lib/components/Toggle.svelte b/frontend/src/lib/components/Toggle.svelte index 46c5f37897..b3eb49bd34 100644 --- a/frontend/src/lib/components/Toggle.svelte +++ b/frontend/src/lib/components/Toggle.svelte @@ -84,5 +84,6 @@ {/if} {/if} + diff --git a/frontend/src/lib/components/flows/content/FlowModuleComponent.svelte b/frontend/src/lib/components/flows/content/FlowModuleComponent.svelte index eb78b2890c..6fc0386c84 100644 --- a/frontend/src/lib/components/flows/content/FlowModuleComponent.svelte +++ b/frontend/src/lib/components/flows/content/FlowModuleComponent.svelte @@ -5,6 +5,7 @@ import Editor from '$lib/components/Editor.svelte' import EditorBar from '$lib/components/EditorBar.svelte' import ModulePreview from '$lib/components/ModulePreview.svelte' + import Toggle from '$lib/components/Toggle.svelte' import { createScriptFromInlineScript, fork } from '$lib/components/flows/flowStateUtils' import { RawScript, type FlowModule, Script } from '$lib/gen' @@ -40,6 +41,8 @@ import s3Scripts from './s3Scripts/lib' import type { FlowCopilotContext } from '$lib/components/copilot/flow' import Label from '$lib/components/Label.svelte' + import { enterpriseLicense } from '$lib/stores' + import { isCloudHosted } from '$lib/cloud' import { loadSchemaFromModule } from '../flowInfers' const { selectedId, previewArgs, flowStateStore, flowStore, saveDraft } = @@ -331,6 +334,7 @@ Mock Shared Directory Timeout + Priority {#if flowModule.value['language'] === 'python3' || flowModule.value['language'] === 'deno'} S3 {/if} @@ -409,6 +413,46 @@ Set shared directory in the flow settings + {:else if advancedSelected === 'priority'} +
+ + 0} + on:change={() => { + if (flowModule.priority) { + flowModule.priority = undefined + } else { + flowModule.priority = 100 + } + }} + options={{ + right: 'High priority flow step', + rightTooltip: `Jobs scheduled from this step when the flow is executed are labeled as high priority and take precedence over the other jobs in the jobs queue. ${ + !$enterpriseLicense + ? 'This is a feature only available on enterprise edition.' + : '' + }` + }} + > + + { + if (flowModule.priority && flowModule.priority > 100) { + flowModule.priority = 100 + } else if (flowModule.priority && flowModule.priority < 0) { + flowModule.priority = 0 + } + }} + /> + + +
{:else if advancedSelected === 's3'}

diff --git a/frontend/src/lib/components/flows/content/FlowSettings.svelte b/frontend/src/lib/components/flows/content/FlowSettings.svelte index e9ca520507..ce1d8e550b 100644 --- a/frontend/src/lib/components/flows/content/FlowSettings.svelte +++ b/frontend/src/lib/components/flows/content/FlowSettings.svelte @@ -13,7 +13,8 @@ import type { FlowEditorContext } from '../types' import autosize from 'svelte-autosize' import Slider from '$lib/components/Slider.svelte' - import { workerTags, workspaceStore } from '$lib/stores' + import { enterpriseLicense, workerTags, workspaceStore } from '$lib/stores' + import { isCloudHosted } from '$lib/cloud' import { copyToClipboard } from '$lib/utils' import { Icon } from 'svelte-awesome' import { faClipboard } from '@fortawesome/free-solid-svg-icons' @@ -113,6 +114,44 @@ /> + + 0} + on:change={() => { + if ($flowStore.value.priority) { + $flowStore.value.priority = undefined + } else { + $flowStore.value.priority = 100 + } + }} + options={{ + right: `Label as high priority`, + rightTooltip: `All jobs scheduled by flows labeled as high priority take precedence over the other jobs in the jobs queue. ${ + !$enterpriseLicense + ? 'This is a feature only available on enterprise edition.' + : '' + }` + }} + > + + { + if ($flowStore.value.priority && $flowStore.value.priority > 100) { + $flowStore.value.priority = 100 + } else if ($flowStore.value.priority && $flowStore.value.priority < 0) { + $flowStore.value.priority = 0 + } + }} + /> + + +
- - {#if job && 'duration_ms' in job && job.duration_ms != undefined} + {#if job?.['priority']} + + priority: {job?.['priority']} + + {/if} + {#if job && 'duration_ms' in job && job.duration_ms != undefined} + Ran in ({msToSec(job.duration_ms)}s) - {/if} - + + {/if} {#if job?.['mem_peak']} Mem: {job?.['mem_peak'] ? `${(job['mem_peak'] / 1024).toPrecision(4)}MB` : 'N/A'} diff --git a/frontend/src/routes/(root)/(logged)/flows/get/[...path]/+page.svelte b/frontend/src/routes/(root)/(logged)/flows/get/[...path]/+page.svelte index 222e8e49dd..532cf48634 100644 --- a/frontend/src/routes/(root)/(logged)/flows/get/[...path]/+page.svelte +++ b/frontend/src/routes/(root)/(logged)/flows/get/[...path]/+page.svelte @@ -7,7 +7,7 @@ import DetailPageLayout from '$lib/components/details/DetailPageLayout.svelte' import { goto } from '$app/navigation' - import { Alert, Skeleton } from '$lib/components/common' + import { Alert, Badge as HeaderBadge, Skeleton } from '$lib/components/common' import MoveDrawer from '$lib/components/MoveDrawer.svelte' import RunForm from '$lib/components/RunForm.svelte' import ShareModal from '$lib/components/ShareModal.svelte' @@ -255,7 +255,22 @@ bind:errorHandlerMuted={flow.ws_error_handler_muted} scriptOrFlowPath={flow.path} errorHandlerKind="flow" - /> + > + {#if flow?.value?.priority != undefined} + + {/if} + {#if flow?.value?.concurrent_limit != undefined && flow?.value?.concurrency_time_window_s != undefined} + + {/if} + diff --git a/frontend/src/routes/(root)/(logged)/run/[...run]/+page.svelte b/frontend/src/routes/(root)/(logged)/run/[...run]/+page.svelte index 6c4dcda7c2..179a44ab88 100644 --- a/frontend/src/routes/(root)/(logged)/run/[...run]/+page.svelte +++ b/frontend/src/routes/(root)/(logged)/run/[...run]/+page.svelte @@ -291,6 +291,11 @@ {job.job_kind}
{/if} + {#if job && 'priority' in job} +
+ priority: {job.priority} +
+ {/if} {#if job.tag && !['deno', 'python3', 'flow', 'other', 'go', 'postgresql', 'mysql', 'bigquery', 'snowflake', 'graphql', 'nativets', 'bash', 'powershell', 'other', 'dependency'].includes(job.tag)}
Worker group: {job.tag} diff --git a/frontend/src/routes/(root)/(logged)/scripts/get/[...hash]/+page.svelte b/frontend/src/routes/(root)/(logged)/scripts/get/[...hash]/+page.svelte index aa25b134e0..b2025ec9f1 100644 --- a/frontend/src/routes/(root)/(logged)/scripts/get/[...hash]/+page.svelte +++ b/frontend/src/routes/(root)/(logged)/scripts/get/[...hash]/+page.svelte @@ -415,6 +415,13 @@ errorHandlerKind="script" scriptOrFlowPath={script.path} > + {#if script?.priority != undefined} + + {/if} {#if script?.concurrent_limit != undefined && script.concurrency_time_window_s != undefined}