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

feat: improve cancel all for non started jobs (#4065)

Browse Source 
* update version

* improve cancelAll

* improve error

* update sqlx
This commit is contained in:
Ruben Fiszel
2024-07-11 17:11:29 +02:00
committed by GitHub
parent 3b4a376208
commit 2747e1b087
11 changed files with 130 additions and 91 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/.sqlx/query-170f620fbd99269d194d14d56f6a3863d9db5fe736a0a34325b824d9cec9b1a0.json
Generated
+25
View File
@@ -0,0 +1,25 @@
{
"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 , 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 SELECT workspace_id\n , id\n , parent_job\n , created_by\n , created_at\n , now()\n , 0\n , false\n , script_hash\n , script_path\n , args\n , $4\n , raw_code\n , raw_lock\n , true\n , $1\n , canceled_reason\n , job_kind\n , schedule_path\n , permissioned_as\n , flow_status\n , raw_flow\n , is_flow_step\n , false\n , language\n , email\n , visible_to_owner\n , mem_peak\n , tag\n , priority FROM queue \n WHERE id = any($2) AND running = false AND parent_job IS NULL AND workspace_id = $3 AND schedule_path IS NULL FOR UPDATE SKIP LOCKED\n ON CONFLICT (id) DO NOTHING RETURNING id",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "id",
"type_info": "Uuid"
}
],
"parameters": {
"Left": [
"Varchar",
"UuidArray",
"Text",
"Jsonb"
]
},
"nullable": [
false
]
},
"hash": "170f620fbd99269d194d14d56f6a3863d9db5fe736a0a34325b824d9cec9b1a0"
}
backend/.sqlx/query-50d6f8b1a80c122c89abfcb10a5bf13421d1d7da818123db5b140552138c17ad.json
Generated
-16
View File
@@ -1,16 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "UPDATE queue SET canceled = true, canceled_by = $1, canceled_reason = 'cancelled all by user' WHERE id IN (SELECT id FROM queue where id = any($2) AND workspace_id = $3 AND schedule_path IS NULL FOR UPDATE SKIP LOCKED)",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Varchar",
"UuidArray",
"Text"
]
},
"nullable": []
},
"hash": "50d6f8b1a80c122c89abfcb10a5bf13421d1d7da818123db5b140552138c17ad"
}
backend/.sqlx/query-661f472ff3860983322162420457f5033b9c9afc344d9c3e385ba20a3ad2197a.json
Generated
+1 -1
View File
@@ -5,7 +5,7 @@
"columns": [
{
"ordinal": 0,
"name": "bool",
"name": "?column?",
"type_info": "Bool"
}
],
backend/.sqlx/query-8455e77a6a87bd0b8dbe55c854abfb3938cb9a440e9d53323a93ec9473417890.json
Generated
-22
View File
@@ -1,22 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "SELECT id FROM queue WHERE created_at <= $1",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "id",
"type_info": "Uuid"
}
],
"parameters": {
"Left": [
"Timestamptz"
]
},
"nullable": [
false
]
},
"hash": "8455e77a6a87bd0b8dbe55c854abfb3938cb9a440e9d53323a93ec9473417890"
}
backend/.sqlx/query-bd7311e7a93b55714b25992af6378f2e4b00fea6493eeafccd38ad81337e82d2.json → backend/.sqlx/query-8481d1bd91b7d23f946b4cf0d312acfc0ed4973309b4860e93882a37a21a0bd0.json
Generated
+2 -2
View File
@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "SELECT coalesce(COUNT(*), 0) as \"database_length!\" FROM queue WHERE (workspace_id = $1 OR $2) AND scheduled_for <= now() AND running = false AND canceled = false",
"query": "SELECT coalesce(COUNT(*), 0) as \"database_length!\" FROM queue WHERE (workspace_id = $1 OR $2) AND scheduled_for <= now() AND running = false",
"describe": {
"columns": [
{
@@ -19,5 +19,5 @@
null
]
},
"hash": "bd7311e7a93b55714b25992af6378f2e4b00fea6493eeafccd38ad81337e82d2"
"hash": "8481d1bd91b7d23f946b4cf0d312acfc0ed4973309b4860e93882a37a21a0bd0"
}
backend/.sqlx/query-8d655c34a00510699d2ad7044f7e526ba5082e5d1945c76a98404fe5d92e32ee.json
Generated
+15
View File
@@ -0,0 +1,15 @@
{
"db_name": "PostgreSQL",
"query": "DELETE FROM queue WHERE id = any($1) AND workspace_id = $2",
"describe": {
"columns": [],
"parameters": {
"Left": [
"UuidArray",
"Text"
]
},
"nullable": []
},
"hash": "8d655c34a00510699d2ad7044f7e526ba5082e5d1945c76a98404fe5d92e32ee"
}
backend/.sqlx/query-919c9432f36a6212ba79b50fbf887589bf7c37d3557cd831787b9b16b3665e47.json
Generated
-20
View File
@@ -1,20 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "SELECT MAX(created_at) FROM completed_job",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "max",
"type_info": "Timestamptz"
}
],
"parameters": {
"Left": []
},
"nullable": [
null
]
},
"hash": "919c9432f36a6212ba79b50fbf887589bf7c37d3557cd831787b9b16b3665e47"
}
backend/.sqlx/query-b3dbdfb50ee8118bdaed3164b210cb549a34b96554ae1872355b90304f5dcb76.json
Generated
+2 -2
View File
@@ -18,8 +18,8 @@
"Left": []
},
"nullable": [
false,
true
true,
false
]
},
"hash": "b3dbdfb50ee8118bdaed3164b210cb549a34b96554ae1872355b90304f5dcb76"
backend/.sqlx/query-f036e930beff03e8a3f1f91c32267910c7d050f514dfba76991b644b86b914b9.json
Generated
-23
View File
@@ -1,23 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "SELECT CONCAT(coalesce(completed_job.logs, ''), coalesce(job_logs.logs, '')) as logs FROM completed_job\n LEFT JOIN job_logs ON job_logs.job_id = completed_job.id\n WHERE completed_job.id = $1 AND completed_job.workspace_id = $2",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "logs",
"type_info": "Text"
}
],
"parameters": {
"Left": [
"Uuid",
"Text"
]
},
"nullable": [
null
]
},
"hash": "f036e930beff03e8a3f1f91c32267910c7d050f514dfba76991b644b86b914b9"
}
backend/Cargo.lock
Generated
+1 -1
View File
@@ -10531,7 +10531,7 @@ dependencies = [
[[package]]
name = "windmill-indexer"
version = "1.360.1"
version = "1.361.1"
dependencies = [
"anyhow",
"chrono",
backend/windmill-api/src/jobs.rs
+84 -4
View File
@@ -1298,13 +1298,93 @@ async fn cancel_jobs(
rsmq: Option<rsmq_async::MultiplexedRsmq>,
) -> error::JsonResult<Vec<Uuid>> {
let mut uuids = vec![];
let mut tx = db.begin().await?;
let trivial_jobs = sqlx::query!("INSERT INTO completed_job AS cj
( workspace_id
, id
, parent_job
, created_by
, created_at
, started_at
, duration_ms
, success
, script_hash
, script_path
, args
, result
, raw_code
, raw_lock
, canceled
, canceled_by
, canceled_reason
, job_kind
, schedule_path
, permissioned_as
, flow_status
, raw_flow
, is_flow_step
, is_skipped
, language
, email
, visible_to_owner
, mem_peak
, tag
, priority
)
SELECT workspace_id
, id
, parent_job
, created_by
, created_at
, now()
, 0
, false
, script_hash
, script_path
, args
, $4
, raw_code
, raw_lock
, true
, $1
, canceled_reason
, job_kind
, schedule_path
, permissioned_as
, flow_status
, raw_flow
, is_flow_step
, false
, language
, email
, visible_to_owner
, mem_peak
, tag
, priority FROM queue
WHERE id = any($2) AND running = false AND parent_job IS NULL AND workspace_id = $3 AND schedule_path IS NULL FOR UPDATE SKIP LOCKED
ON CONFLICT (id) DO NOTHING RETURNING id", username, &jobs, w_id, serde_json::json!({"error": { "message": format!("Job canceled: cancel all by {username}"), "name": "Canceled", "reason": "cancel all", "canceler": username}}))
.fetch_all(&mut *tx)
.await?.into_iter().map(|x| x.id).collect::<Vec<Uuid>>();
sqlx::query!(
"UPDATE queue SET canceled = true, canceled_by = $1, canceled_reason = 'cancelled all by user' WHERE id IN (SELECT id FROM queue where id = any($2) AND workspace_id = $3 AND schedule_path IS NULL FOR UPDATE SKIP LOCKED)",
username,
"DELETE FROM queue WHERE id = any($1) AND workspace_id = $2",
&jobs,
w_id
).execute(db).await?;
)
.execute(&mut *tx)
.await?;
tx.commit().await?;
// sqlx::query!(
// "UPDATE queue SET canceled = true, canceled_by = $1, canceled_reason = 'cancelled all by user' WHERE id IN (SELECT id FROM queue where id = any($2) AND workspace_id = $3 AND schedule_path IS NULL FOR UPDATE SKIP LOCKED) RETURNING id",
// username,
// &jobs,
// w_id
// ).execute(db).await?;
for job_id in jobs.into_iter() {
if trivial_jobs.contains(&job_id) {
continue;
}
let rsmq = rsmq.clone();
match tokio::time::timeout(tokio::time::Duration::from_secs(5), async move {
let tx = db.begin().await?;
@@ -1416,7 +1496,7 @@ async fn count_queue_jobs(
Ok(Json(
sqlx::query_as!(
QueueStats,
"SELECT coalesce(COUNT(*), 0) as \"database_length!\" FROM queue WHERE (workspace_id = $1 OR $2) AND scheduled_for <= now() AND running = false AND canceled = false",
"SELECT coalesce(COUNT(*), 0) as \"database_length!\" FROM queue WHERE (workspace_id = $1 OR $2) AND scheduled_for <= now() AND running = false",
w_id,
w_id == "admins" && cq.all_workspaces.unwrap_or(false),
)