From 341cdcf66efdfd504492be32d9b4f5cb9db2df2a Mon Sep 17 00:00:00 2001
From: Ruben Fiszel <ruben@windmill.dev>
Date: Fri, 26 Sep 2025 17:19:37 +0000
Subject: [PATCH] fix: improve behavior for already completed jobs when doing
 immediate cancels

---
 backend/windmill-queue/src/jobs.rs            | 39 ++++++++++++++++---
 .../runs/NoWorkerWithTagWarning.svelte        |  8 ++--
 2 files changed, 37 insertions(+), 10 deletions(-)

diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs
index d395abd280..c951f06307 100644
--- a/backend/windmill-queue/src/jobs.rs
+++ b/backend/windmill-queue/src/jobs.rs
@@ -794,7 +794,11 @@ pub async fn add_completed_job<T: Serialize + Send + Sync + ValidableJson>(
             .with_max_times(5)
             .build(),
     )
-    .when(|err| !matches!(err, Error::QuotaExceeded(_)) && !matches!(err, Error::ResultTooLarge(_)))
+    .when(|err| {
+        !matches!(err, Error::QuotaExceeded(_))
+            && !matches!(err, Error::ResultTooLarge(_))
+            && !matches!(err, Error::AlreadyCompleted(_))
+    })
     .notify(|err, dur| {
         tracing::error!("Could not insert completed job, retrying in {dur:#?}, err: {err:#?}");
     })
@@ -860,7 +864,7 @@ async fn commit_completed_job<T: Serialize + Send + Sync + ValidableJson>(
         return value;
     }
 
-    let _duration =  sqlx::query_scalar!(
+    let duration =  sqlx::query_scalar!(
             "INSERT INTO v2_job_completed AS cj
                     ( workspace_id
                     , id
@@ -896,10 +900,33 @@ async fn commit_completed_job<T: Serialize + Send + Sync + ValidableJson>(
             /* $9 */ duration,
             /* $10 */ result_columns as Option<&Vec<String>>,
         )
-        .fetch_one(&mut *tx)
+        .fetch_optional(&mut *tx)
         .await
         .map_err(|e| Error::internal_err(format!("Could not add completed job {job_id}: {e:#}")))?;
 
+    let duration = if let Some(duration) = duration {
+        duration
+    } else {
+        let already_inserted = sqlx::query_scalar!(
+            "SELECT EXISTS(SELECT 1 FROM v2_job_completed WHERE id = $1)",
+            job_id
+        )
+        .fetch_one(&mut *tx)
+        .await
+        .map_err(|e| Error::internal_err(format!("Could not add completed job {job_id}: {e:#}")))?
+        .unwrap_or(false);
+
+        if already_inserted {
+            return Err(Error::AlreadyCompleted(format!(
+                "The queued job {job_id} is already completed."
+            )));
+        } else {
+            return Err(Error::AlreadyCompleted(format!(
+                "There is no queued job anymore for {job_id} but there is no completed job either."
+            )));
+        }
+    };
+
     if let Some(labels) = result.wm_labels() {
         sqlx::query!(
             "UPDATE v2_job SET labels = (
@@ -929,7 +956,7 @@ async fn commit_completed_job<T: Serialize + Send + Sync + ValidableJson>(
                         )
                     WHERE id = $3",
                 &queued_job.id.to_string(),
-                _duration,
+                duration,
                 parent_job
             )
             .execute(&mut *tx)
@@ -1135,7 +1162,7 @@ async fn commit_completed_job<T: Serialize + Send + Sync + ValidableJson>(
         path = &queued_job.runnable_path(),
         job_kind = ?queued_job.kind,
         started_at = ?queued_job.started_at.map(|x| x.to_string()).unwrap_or_else(|| String::new()),
-        duration = ?_duration,
+        duration = ?duration,
         permissioned_as = ?queued_job.permissioned_as,
         email = ?queued_job.permissioned_as_email,
         created_by = queued_job.created_by,
@@ -1148,7 +1175,7 @@ async fn commit_completed_job<T: Serialize + Send + Sync + ValidableJson>(
         queued_job.id
     );
     // tracing::info!("completed job: {:?}", start.elapsed().as_micros());
-    Ok((None, _duration, _skip_downstream_error_handlers))
+    Ok((None, duration, _skip_downstream_error_handlers))
 }
 
 async fn check_result_size<T: ValidableJson>(
diff --git a/frontend/src/lib/components/runs/NoWorkerWithTagWarning.svelte b/frontend/src/lib/components/runs/NoWorkerWithTagWarning.svelte
index a44dbc763f..061cd84bf9 100644
--- a/frontend/src/lib/components/runs/NoWorkerWithTagWarning.svelte
+++ b/frontend/src/lib/components/runs/NoWorkerWithTagWarning.svelte
@@ -4,7 +4,7 @@
 	import Popover from '../Popover.svelte'
 	import { onDestroy, untrack } from 'svelte'
 	interface Props {
-		tag: string
+		tag: string | undefined
 		tagLabel?: string
 	}
 
@@ -17,9 +17,9 @@
 	let visible = true
 
 	let customTag = $derived.by(() => {
-		if (tag.includes('$workspace') || tag.includes('$args')) return
+		if (tag?.includes('$workspace') || tag?.includes('$args')) return
 
-		if (tag.includes('(')) {
+		if (tag?.includes('(')) {
 			return tag.split('(')[0]
 		}
 		return tag
@@ -44,7 +44,7 @@
 
 	$effect(() => {
 		customTag
-		untrack(() => lookForTag())
+		untrack(() => timeout && setTimeout(() => lookForTag(), 2500))
 	})
 
 	onDestroy(() => {