From 404d7b8a4f572d07ff4e9f661e014413973e8432 Mon Sep 17 00:00:00 2001 From: "claude[bot]" <209825114+claude[bot]@users.noreply.github.com> Date: Thu, 11 Sep 2025 00:13:52 +0000 Subject: [PATCH] feat: show position of job in queue when waiting for executor (#6554) * feat: show position of job in queue when waiting for executor - Added new API endpoint /queue/position/:id to get job's position in queue - Modified DisplayResult.svelte to fetch and display queue position - Shows 'Waiting for executor (position X in queue)' when job is queued - Refreshes position every 2 seconds while waiting Fixes #6553 Co-authored-by: windmill-internal-app[bot] * iterate * iterate * all * all --------- Co-authored-by: claude[bot] <209825114+claude[bot]@users.noreply.github.com> Co-authored-by: windmill-internal-app[bot] Co-authored-by: Ruben Fiszel --- ...9896b2820efb8cd0d1325334e18bacde50eb5.json | 23 +++++ ...ac5fb13680b0e574d6c7c16e3bc420dd90727.json | 22 +++++ backend/windmill-api/openapi.yaml | 59 ++++++++++++- backend/windmill-api/src/jobs.rs | 47 ++++++++++ .../src/lib/components/DisplayResult.svelte | 1 + frontend/src/lib/components/JobStatus.svelte | 2 + frontend/src/lib/components/LogViewer.svelte | 4 + .../src/lib/components/QueuePosition.svelte | 88 +++++++++++++++++++ 8 files changed, 244 insertions(+), 2 deletions(-) create mode 100644 backend/.sqlx/query-31834667a72b0d25a9b82e9b63e9896b2820efb8cd0d1325334e18bacde50eb5.json create mode 100644 backend/.sqlx/query-b2f6327828acc4ba1872c20a138ac5fb13680b0e574d6c7c16e3bc420dd90727.json create mode 100644 frontend/src/lib/components/QueuePosition.svelte diff --git a/backend/.sqlx/query-31834667a72b0d25a9b82e9b63e9896b2820efb8cd0d1325334e18bacde50eb5.json b/backend/.sqlx/query-31834667a72b0d25a9b82e9b63e9896b2820efb8cd0d1325334e18bacde50eb5.json new file mode 100644 index 0000000000..596dd9c190 --- /dev/null +++ b/backend/.sqlx/query-31834667a72b0d25a9b82e9b63e9896b2820efb8cd0d1325334e18bacde50eb5.json @@ -0,0 +1,23 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT scheduled_for FROM v2_job_queue WHERE id = $1 AND workspace_id = $2", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "scheduled_for", + "type_info": "Timestamptz" + } + ], + "parameters": { + "Left": [ + "Uuid", + "Text" + ] + }, + "nullable": [ + false + ] + }, + "hash": "31834667a72b0d25a9b82e9b63e9896b2820efb8cd0d1325334e18bacde50eb5" +} diff --git a/backend/.sqlx/query-b2f6327828acc4ba1872c20a138ac5fb13680b0e574d6c7c16e3bc420dd90727.json b/backend/.sqlx/query-b2f6327828acc4ba1872c20a138ac5fb13680b0e574d6c7c16e3bc420dd90727.json new file mode 100644 index 0000000000..9695c3e1c3 --- /dev/null +++ b/backend/.sqlx/query-b2f6327828acc4ba1872c20a138ac5fb13680b0e574d6c7c16e3bc420dd90727.json @@ -0,0 +1,22 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT COUNT(*) as count\n FROM v2_job_queue\n WHERE scheduled_for < to_timestamp($1::bigint / 1000.0)\n AND running = false\n AND suspend_until IS NULL", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "count", + "type_info": "Int8" + } + ], + "parameters": { + "Left": [ + "Int8" + ] + }, + "nullable": [ + null + ] + }, + "hash": "b2f6327828acc4ba1872c20a138ac5fb13680b0e574d6c7c16e3bc420dd90727" +} diff --git a/backend/windmill-api/openapi.yaml b/backend/windmill-api/openapi.yaml index 9f16dfd8e7..42901d3732 100644 --- a/backend/windmill-api/openapi.yaml +++ b/backend/windmill-api/openapi.yaml @@ -3210,7 +3210,7 @@ paths: text/plain: schema: type: string - + /w/{workspace}/variables/delete_bulk: delete: summary: delete variables in bulk @@ -8383,6 +8383,49 @@ paths: schema: type: string + /w/{workspace}/jobs/queue/position/{scheduled_for}: + get: + summary: get queue position for a job + operationId: getQueuePosition + tags: + - job + parameters: + - $ref: "#/components/parameters/WorkspaceId" + - name: scheduled_for + in: path + required: true + schema: + type: integer + description: The scheduled for timestamp in milliseconds + responses: + "200": + description: queue position information + content: + application/json: + schema: + type: object + properties: + position: + type: integer + description: The position in queue (1-based), null if not in queue or already running + + /w/{workspace}/jobs/queue/scheduled_for/{id}: + get: + summary: get scheduled for timestamp for a job + operationId: getScheduledFor + tags: + - job + parameters: + - $ref: "#/components/parameters/WorkspaceId" + - $ref: "#/components/parameters/JobId" + responses: + "200": + description: scheduled for timestamp + content: + application/json: + schema: + type: integer + /w/{workspace}/jobs/job_signature/{id}/{resume_id}: get: summary: create an HMac signature given a job id and a resume id @@ -18438,7 +18481,19 @@ components: CaptureTriggerKind: type: string enum: - [webhook, http, websocket, kafka, default_email, nats, postgres, sqs, mqtt, gcp, email] + [ + webhook, + http, + websocket, + kafka, + default_email, + nats, + postgres, + sqs, + mqtt, + gcp, + email, + ] Capture: type: object diff --git a/backend/windmill-api/src/jobs.rs b/backend/windmill-api/src/jobs.rs index 48386582d4..5071730b51 100644 --- a/backend/windmill-api/src/jobs.rs +++ b/backend/windmill-api/src/jobs.rs @@ -203,6 +203,8 @@ pub fn workspaced_service() -> Router { .route("/queue/list", get(list_queue_jobs)) .route("/queue/count", get(count_queue_jobs)) .route("/queue/list_filtered_uuids", get(list_filtered_uuids)) + .route("/queue/position/:timestamp", get(get_queue_position)) + .route("/queue/scheduled_for/:id", get(get_scheduled_for)) .route("/queue/cancel_selection", post(cancel_selection)) .route("/completed/count", get(count_completed_jobs)) .route("/completed/count_jobs", get(count_completed_jobs_detail)) @@ -537,6 +539,51 @@ async fn force_cancel( } } +#[derive(Serialize)] +struct QueuePosition { + position: Option, +} + +async fn get_queue_position( + _authed: ApiAuthed, + Extension(db): Extension, + Path((_w_id, scheduled_for)): Path<(String, i64)>, +) -> error::Result> { + // First check if the job exists and is in queue + + // Count jobs that are scheduled before this job and are not suspended + let count: i64 = sqlx::query_scalar!( + "SELECT COUNT(*) as count + FROM v2_job_queue + WHERE scheduled_for < to_timestamp($1::bigint / 1000.0) + AND running = false + AND suspend_until IS NULL", + scheduled_for, + ) + .fetch_one(&db) + .await? + .unwrap_or(0); + + Ok(Json(QueuePosition { position: Some(count + 1) })) +} + +async fn get_scheduled_for( + _authed: ApiAuthed, + Extension(db): Extension, + Path((w_id, id)): Path<(String, Uuid)>, +) -> error::Result> { + let scheduled_for = sqlx::query_scalar!( + "SELECT scheduled_for FROM v2_job_queue WHERE id = $1 AND workspace_id = $2", + id, + w_id, + ) + .fetch_optional(&db) + .await?; + + let scheduled_for = not_found_if_none(scheduled_for, "QueuedJob", &id.to_string())?; + Ok(Json(scheduled_for.timestamp_millis())) +} + async fn get_flow_job_debug_info( OptAuthed(opt_authed): OptAuthed, tokened_o: OptTokened, diff --git a/frontend/src/lib/components/DisplayResult.svelte b/frontend/src/lib/components/DisplayResult.svelte index a0d5784552..1c03241e00 100644 --- a/frontend/src/lib/components/DisplayResult.svelte +++ b/frontend/src/lib/components/DisplayResult.svelte @@ -476,6 +476,7 @@ onDestroy(() => { dispatch('toolbar-location-changed', undefined) }) + $effect(() => { ;[result] resultKind = inferResultKind(result) diff --git a/frontend/src/lib/components/JobStatus.svelte b/frontend/src/lib/components/JobStatus.svelte index c071c1568d..e97fea156b 100644 --- a/frontend/src/lib/components/JobStatus.svelte +++ b/frontend/src/lib/components/JobStatus.svelte @@ -6,6 +6,7 @@ import DurationMs from './DurationMs.svelte' import { Calendar, CheckCircle2, Circle, Clock, Hourglass, Play, XCircle } from 'lucide-svelte' import NoWorkerWithTagWarning from './runs/NoWorkerWithTagWarning.svelte' + import QueuePosition from './QueuePosition.svelte' const SMALL_ICON_SIZE = 12 @@ -60,6 +61,7 @@
Queued +
{:else} diff --git a/frontend/src/lib/components/LogViewer.svelte b/frontend/src/lib/components/LogViewer.svelte index a07596856b..d9a9203bc5 100644 --- a/frontend/src/lib/components/LogViewer.svelte +++ b/frontend/src/lib/components/LogViewer.svelte @@ -17,6 +17,7 @@ import { JobService } from '$lib/gen' import Tooltip from './Tooltip.svelte' import { twMerge } from 'tailwind-merge' + import QueuePosition from './QueuePosition.svelte' interface Props { content: string | undefined @@ -263,6 +264,9 @@ {/if} + {#if jobId} + + {/if} {:else if duration} + import { JobService } from '$lib/gen' + import { workspaceStore } from '$lib/stores' + + let { + jobId, + workspaceId, + minimal = false + }: { jobId: string; workspaceId?: string | undefined; minimal?: boolean } = $props() + + let queuePositionInterval: NodeJS.Timeout | undefined + let queueState = $state(undefined) as undefined | { position?: number } + + let fetchingQueuePosition = false + + let workspace = $derived(workspaceId ?? $workspaceStore) + + let scheduledFor = $state(undefined) as undefined | number + + let scheduledForTimeout: NodeJS.Timeout | undefined + $effect(() => { + if (jobId && workspace) { + clearTimeout(scheduledForTimeout) + queueState = undefined + scheduledForTimeout = setTimeout(() => { + JobService.getScheduledFor({ + workspace: workspace, + id: jobId + }).then((response) => { + scheduledFor = response + }) + }, 2000) + } + }) + + $effect(() => { + // Fetch queue position when loading and we have jobId + if (scheduledFor) { + // Initial fetch + fetchQueuePosition() + + // Set up interval to refresh every 2 seconds + queuePositionInterval = setInterval(() => { + fetchQueuePosition() + }, 5000) + } else { + // Clear interval when not loading + if (queuePositionInterval) { + clearInterval(queuePositionInterval) + queuePositionInterval = undefined + } + queueState = undefined + } + + return () => { + if (queuePositionInterval) { + scheduledForTimeout && clearTimeout(scheduledForTimeout) + clearInterval(queuePositionInterval) + } + } + }) + + async function fetchQueuePosition() { + if (!workspace || !scheduledFor || fetchingQueuePosition) return + + try { + fetchingQueuePosition = true + queueState = await JobService.getQueuePosition({ + workspace: workspace, + scheduledFor: scheduledFor + }) + } catch (error) { + console.error('Failed to fetch queue position:', error) + queueState = undefined + } finally { + fetchingQueuePosition = false + } + } + + +{#if queueState} +
+ Queue position: {queueState.position} + {#if !minimal} + (Waiting for an available worker) + {/if} +
+{/if}