starred/windmill
1
0
Fork 0
mirror of https://github.com/windmill-labs/windmill.git synced 2026-08-21 16:02:28 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: show position of job in queue when waiting for executor (#6554)

Browse Source 
* 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] <windmill-internal-app[bot]@users.noreply.github.com>

* iterate

* iterate

* all

* all

---------

Co-authored-by: claude[bot] <209825114+claude[bot]@users.noreply.github.com>
Co-authored-by: windmill-internal-app[bot] <windmill-internal-app[bot]@users.noreply.github.com>
Co-authored-by: Ruben Fiszel <ruben@windmill.dev>
This commit is contained in:
claude[bot]
2025-09-11 00:13:52 +00:00
committed by GitHub
parent 94045bb6fb
commit 404d7b8a4f
8 changed files with 244 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/.sqlx/query-31834667a72b0d25a9b82e9b63e9896b2820efb8cd0d1325334e18bacde50eb5.json
Generated
+23
View File
@@ -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"
}
backend/.sqlx/query-b2f6327828acc4ba1872c20a138ac5fb13680b0e574d6c7c16e3bc420dd90727.json
Generated
+22
View File
@@ -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"
}
backend/windmill-api/openapi.yaml
+57 -2
View File
@@ -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
backend/windmill-api/src/jobs.rs
+47
View File
@@ -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<i64>,
}
async fn get_queue_position(
_authed: ApiAuthed,
Extension(db): Extension<DB>,
Path((_w_id, scheduled_for)): Path<(String, i64)>,
) -> error::Result<Json<QueuePosition>> {
// 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<DB>,
Path((w_id, id)): Path<(String, Uuid)>,
) -> error::Result<Json<i64>> {
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,
frontend/src/lib/components/DisplayResult.svelte
+1
View File
@@ -476,6 +476,7 @@
onDestroy(() => {
dispatch('toolbar-location-changed', undefined)
})
$effect(() => {
;[result]
resultKind = inferResultKind(result)
frontend/src/lib/components/JobStatus.svelte
+2
View File
@@ -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 @@
<div class="flex flex-row gap-1 items-center">
<Badge color="orange" icon={{ icon: Clock, position: 'left' }}>Queued</Badge>
<NoWorkerWithTagWarning tag={job.tag} />
<QueuePosition jobId={job.id} workspaceId={job.workspace_id} minimal />
</div>
{:else}
<Circle size={SMALL_ICON_SIZE} class="text-gray-200" />
frontend/src/lib/components/LogViewer.svelte
+4
View File
@@ -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 @@
<NoWorkerWithTagWarning {tagLabel} {tag} />
</div>
{/if}
{#if jobId}
<QueuePosition {jobId} />
{/if}
</div>
{:else if duration}
<span
frontend/src/lib/components/QueuePosition.svelte
+88
View File
@@ -0,0 +1,88 @@
<script lang="ts">
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
}
}
</script>
{#if queueState}
<div class="text-small ml-4">
<span class="text-orange-600">Queue position: <b>{queueState.position}</b></span>
{#if !minimal}
<span class="ml-2 text-tertiary">(Waiting for an available worker)</span>
{/if}
</div>
{/if}