diff --git a/frontend/src/lib/components/DBManagerContent.svelte b/frontend/src/lib/components/DBManagerContent.svelte index 9332d7ad26..f84c5dfce4 100644 --- a/frontend/src/lib/components/DBManagerContent.svelte +++ b/frontend/src/lib/components/DBManagerContent.svelte @@ -1,9 +1,15 @@ + +{#if served.current === false} +
+ + + {subject} run as Windmill jobs tagged {tag}, and no worker is currently listening to + that tag, so they stay queued until one is. If no worker group is meant to serve it, add + {tag} + to a group's worker tags on the workers page. + +
+{/if} diff --git a/frontend/src/lib/components/jobs/missingWorker.test.ts b/frontend/src/lib/components/jobs/missingWorker.test.ts new file mode 100644 index 0000000000..01f591aece --- /dev/null +++ b/frontend/src/lib/components/jobs/missingWorker.test.ts @@ -0,0 +1,121 @@ +import { describe, expect, it, vi, beforeEach, afterEach } from 'vitest' + +const getCompletedJobResultMaybe = vi.fn() +const getJob = vi.fn() +const cancelQueuedJob = vi.fn() +const existsWorkersWithTags = vi.fn() + +vi.mock('$lib/gen', () => ({ + JobService: { + getCompletedJobResultMaybe: (...a: unknown[]) => getCompletedJobResultMaybe(...(a as [])), + getJob: (...a: unknown[]) => getJob(...(a as [])), + cancelQueuedJob: (...a: unknown[]) => cancelQueuedJob(...(a as [])) + }, + WorkerService: { + existsWorkersWithTags: (...a: unknown[]) => existsWorkersWithTags(...(a as [])) + } +})) + +import { pollJobResult } from './utils' +import { hasWorkerForTag, NoWorkerForTagError } from './missingWorker' + +function settlementTracker(promise: Promise) { + const state = { settled: false } + promise.then( + () => (state.settled = true), + () => (state.settled = true) + ) + return state +} + +beforeEach(() => { + getCompletedJobResultMaybe.mockReset() + getJob.mockReset() + cancelQueuedJob.mockReset() + existsWorkersWithTags.mockReset() + getCompletedJobResultMaybe.mockResolvedValue({ completed: false }) + getJob.mockResolvedValue({ type: 'QueuedJob', running: false, tag: 'postgresql' }) + cancelQueuedJob.mockResolvedValue(undefined) + vi.useFakeTimers() +}) + +afterEach(() => { + vi.useRealTimers() +}) + +// Long enough for the whole confirmation window (first probe + 2 intervals). +const PAST_CONFIRMATION_WINDOW_MS = 120_000 + +describe('pollJobResult', () => { + it('reports a queued read whose tag stays unserved without cancelling it', async () => { + existsWorkersWithTags.mockResolvedValue({ postgresql: false }) + + const promise = pollJobResult('job-1', 'ws') + const rejects = expect(promise).rejects.toBeInstanceOf(NoWorkerForTagError) + await vi.advanceTimersByTimeAsync(PAST_CONFIRMATION_WINDOW_MS) + await rejects + // The backlog is what the autoscaler scales up on: cancelling would stop a + // group coming back from zero from ever recovering. + expect(cancelQueuedJob).not.toHaveBeenCalled() + }) + + it('never abandons a write, and reports once why it is waiting', async () => { + existsWorkersWithTags.mockResolvedValue({ postgresql: false }) + const onNoWorkerForTag = vi.fn() + + const promise = pollJobResult('job-1', 'ws', { sideEffecting: true, onNoWorkerForTag }) + const tracker = settlementTracker(promise) + + await vi.advanceTimersByTimeAsync(PAST_CONFIRMATION_WINDOW_MS * 3) + // Reporting failure while the write stays executable would let it apply after + // the caller gave up and duplicate on retry; cancelling it first cannot be + // done atomically from the client. + expect(tracker.settled).toBe(false) + expect(cancelQueuedJob).not.toHaveBeenCalled() + expect(onNoWorkerForTag).toHaveBeenCalledTimes(1) + expect(onNoWorkerForTag).toHaveBeenCalledWith('postgresql') + + getCompletedJobResultMaybe.mockResolvedValue({ completed: true, success: true, result: 7 }) + await vi.advanceTimersByTimeAsync(3_000) + await expect(promise).resolves.toBe(7) + }) + + it('does not give up while a worker group could still be coming up', async () => { + // A worker group booting is absent from worker_ping exactly like an unserved + // tag; only a run of empty lookups distinguishes them. + existsWorkersWithTags.mockResolvedValueOnce({ postgresql: false }) + existsWorkersWithTags.mockResolvedValueOnce({ postgresql: false }) + existsWorkersWithTags.mockResolvedValue({ postgresql: true }) + + const promise = pollJobResult('job-1', 'ws') + const tracker = settlementTracker(promise) + + await vi.advanceTimersByTimeAsync(PAST_CONFIRMATION_WINDOW_MS) + expect(tracker.settled).toBe(false) + expect(cancelQueuedJob).not.toHaveBeenCalled() + }) + + it('keeps waiting on a job queued behind a busy worker that serves its tag', async () => { + existsWorkersWithTags.mockResolvedValue({ postgresql: true }) + + const promise = pollJobResult('job-1', 'ws') + const tracker = settlementTracker(promise) + + await vi.advanceTimersByTimeAsync(PAST_CONFIRMATION_WINDOW_MS) + expect(tracker.settled).toBe(false) + + getCompletedJobResultMaybe.mockResolvedValue({ completed: true, success: true, result: 42 }) + await vi.advanceTimersByTimeAsync(3_000) + await expect(promise).resolves.toBe(42) + }) +}) + +describe('hasWorkerForTag', () => { + it('treats an answer it did not get as a worker being there', async () => { + // `existsWorkersWithTags` returns an empty map when TAGS_ARE_SENSITIVE hides + // the tag from the caller. Reading that as "unserved" would diagnose a + // perfectly healthy instance. + existsWorkersWithTags.mockResolvedValue({}) + await expect(hasWorkerForTag('ws', 'postgresql')).resolves.toBe(true) + }) +}) diff --git a/frontend/src/lib/components/jobs/missingWorker.ts b/frontend/src/lib/components/jobs/missingWorker.ts new file mode 100644 index 0000000000..e392e06d3e --- /dev/null +++ b/frontend/src/lib/components/jobs/missingWorker.ts @@ -0,0 +1,76 @@ +import { JobService, WorkerService } from '$lib/gen' + +/** + * A queued job whose tag no running worker serves is never picked up: without + * this the UI polls until the server-side `run_wait_result` timeout (10min by + * default) or, for the client-side pollers, forever. + * + * The most common cause is a language that defaults to a native tag + * (`postgresql`, `mysql`, `bigquery`, …) on an instance whose worker groups + * only declare the default tags. + */ +export class NoWorkerForTagError extends Error { + tag: string + + constructor(tag: string) { + super( + `No worker has been listening to the tag "${tag}" while this job waited, so it was never ` + + `picked up. It stays queued and will run once a worker with that tag comes online. ` + + `Add "${tag}" to the worker tags of one of your worker groups (Workers page), or run a ` + + `worker that serves it.` + ) + this.name = 'NoWorkerForTagError' + this.tag = tag + } +} + +/** Shown while a write waits, which is never abandoned (see `sideEffecting`). */ +export function queuedWithoutWorkerMessage(tag: string): string { + return ( + `No worker is listening to the tag "${tag}", so this operation is queued and will only run ` + + `once one is. Add "${tag}" to the worker tags of one of your worker groups (Workers page), ` + + `or run a worker that serves it.` + ) +} + +/** How long a job may sit un-started before the first lookup for a worker serving its tag. */ +export const NO_WORKER_FIRST_PROBE_MS = 10_000 +/** How long to wait between lookups while the job stays queued. */ +export const NO_WORKER_PROBE_INTERVAL_MS = 40_000 +/** + * How many consecutive lookups must come back empty before the caller stops + * waiting. A worker group scaling from zero, or every worker down for a rollout, + * is indistinguishable from an unserved tag in any single lookup, so no single + * empty reading is acted on. + */ +export const NO_WORKER_CONFIRMATIONS = 3 + +/** + * Whether any worker pinged in the last minute declares `tag`. Unknown answers + * (the endpoint returns an empty map when tags are sensitive and the caller may + * not see them) count as "yes", so an opaque instance never gets a wrong + * diagnosis. + */ +export async function hasWorkerForTag(workspace: string, tag: string): Promise { + const existing = await WorkerService.existsWorkersWithTags({ workspace, tags: tag }) + return existing[tag] !== false +} + +/** + * The tag of `jobId` when it is still queued and no worker serves it, else + * undefined. Never throws: a failed lookup means "can't tell", and the caller + * keeps waiting rather than reporting a cause it did not establish. + */ +export async function missingWorkerTagOfQueuedJob( + workspace: string, + jobId: string +): Promise { + try { + const job = await JobService.getJob({ workspace, id: jobId, noCode: true, noLogs: true }) + if (job.type !== 'QueuedJob' || job.running || !job.tag) return undefined + return (await hasWorkerForTag(workspace, job.tag)) ? undefined : job.tag + } catch (err) { + console.warn('Could not determine whether a worker serves the job tag', err) + return undefined + } +} diff --git a/frontend/src/lib/components/jobs/utils.ts b/frontend/src/lib/components/jobs/utils.ts index 294df7c3ff..735484c1f2 100644 --- a/frontend/src/lib/components/jobs/utils.ts +++ b/frontend/src/lib/components/jobs/utils.ts @@ -1,4 +1,11 @@ import { JobService, type RunScriptByPathData, type RunScriptPreviewData } from '$lib/gen' +import { + missingWorkerTagOfQueuedJob, + NoWorkerForTagError, + NO_WORKER_CONFIRMATIONS, + NO_WORKER_FIRST_PROBE_MS, + NO_WORKER_PROBE_INTERVAL_MS +} from './missingWorker' function isRunScriptByPathData( arg: RunScriptPreviewData | RunScriptByPathData @@ -9,13 +16,26 @@ function isRunScriptByPathData( type RunScriptOptions = { maxRetries?: number withJobData?: boolean + /** Set to false to keep polling a job that no worker can pick up. */ + failIfNoWorkerForTag?: boolean + /** + * The job writes (insert/update/delete, DDL, arbitrary SQL). Such a job is + * never given up on: reporting it as failed while it stays executable invites + * a duplicate retry, and cancelling it first is not something the client can + * do atomically: a worker can claim it between the tag probe and the cancel, + * and a soft-cancelled statement may already have committed. `onNoWorkerForTag` + * is what tells the user why it is waiting. + */ + sideEffecting?: boolean + /** Called once when the job has stayed queued on a tag no worker serves. */ + onNoWorkerForTag?: (tag: string) => void } /** * @function runScript * @param {RunScriptPreviewData | RunScriptByPathData} data - Data for running the script. * @returns {Promise} A UUID representing the running script. - * + * * @example * const uuid = await runScript(data) */ @@ -29,6 +49,15 @@ export async function runScript(data: RunScriptPreviewData | RunScriptByPathData return uuid } +/** Tight at first so a quick job feels instant, then slower: a schema + * introspection or a DDL migration can run for minutes, and a fixed sub-second + * tick would cost hundreds of round-trips for it. */ +function pollDelayMs(poll: number): number { + if (poll < 4) return 375 + if (poll < 12) return 750 + return 2000 +} + /** * @function pollJobResult * @description Polls a job result by UUID until success, failure, or max retries reached. @@ -36,19 +65,35 @@ export async function runScript(data: RunScriptPreviewData | RunScriptByPathData * @param {string} workspace - Workspace identifier. * @param {RunScriptOptions} [options] - Optional settings like retries and job data inclusion. * @returns {Promise} Final job result or throws error if it fails. - * + * * @example * const result = await pollJobResult(uuid, 'my-workspace', { maxRetries: 5, withJobData: true }); */ export async function pollJobResult( uuid: string, workspace: string, - { maxRetries = 7, withJobData }: RunScriptOptions = {} + { + maxRetries = 7, + withJobData, + failIfNoWorkerForTag = true, + sideEffecting = false, + onNoWorkerForTag + }: RunScriptOptions = {} ): Promise { let attempts = 0 + let polls = 0 + // `attempts` only advances on errors, so a queued job would poll forever. The + // one case that never resolves on its own is a tag no worker serves, which + // takes NO_WORKER_CONFIRMATIONS consecutive empty lookups to establish, since + // a worker group booting reads like an unserved tag in any single one. + let noWorkerProbeAt = Date.now() + NO_WORKER_FIRST_PROBE_MS + let unservedProbes = 0 + let reportedNoWorker = false while (attempts < maxRetries) { try { - await new Promise((resolve) => setTimeout(resolve, 500 * (attempts || 0.75))) + await new Promise((resolve) => + setTimeout(resolve, attempts ? 500 * attempts : pollDelayMs(polls++)) + ) const job = await JobService.getCompletedJobResultMaybe({ id: uuid, workspace @@ -65,8 +110,26 @@ export async function pollJobResult( if (typeof errorMsg !== 'string') errorMsg = undefined console.error('JOB FAILED', job.result) throw new Error(errorMsg ?? 'Job failed') + } else if (failIfNoWorkerForTag && Date.now() >= noWorkerProbeAt) { + const tag = await missingWorkerTagOfQueuedJob(workspace, uuid) + noWorkerProbeAt = Date.now() + NO_WORKER_PROBE_INTERVAL_MS + unservedProbes = tag ? unservedProbes + 1 : 0 + if (tag && unservedProbes >= NO_WORKER_CONFIRMATIONS) { + if (!reportedNoWorker) { + reportedNoWorker = true + onNoWorkerForTag?.(tag) + } + // Reads give up the wait but leave the job queued: cancelling one would + // remove the very backlog the autoscaler scales up on, so a group coming + // back from zero (300s cooldown) would never recover. Writes keep + // waiting instead (see `sideEffecting`). + if (!sideEffecting) throw new NoWorkerForTagError(tag) + } } } catch (e) { + if (e instanceof NoWorkerForTagError) { + throw e + } if (attempts == maxRetries) { throw e } diff --git a/frontend/src/lib/components/jobs/writingJob.ts b/frontend/src/lib/components/jobs/writingJob.ts new file mode 100644 index 0000000000..d1c32ba826 --- /dev/null +++ b/frontend/src/lib/components/jobs/writingJob.ts @@ -0,0 +1,12 @@ +import { sendUserToast } from '$lib/toast' +import { queuedWithoutWorkerMessage } from './missingWorker' + +/** + * Poll options for a job that writes (row edits, DDL, arbitrary SQL). Such a job + * is never abandoned (see `sideEffecting` in `pollJobResult`), so this is what + * explains the wait when it sits on a tag no worker serves. + */ +export const writingJobOptions = { + sideEffecting: true, + onNoWorkerForTag: (tag: string) => sendUserToast(queuedWithoutWorkerMessage(tag), true) +} diff --git a/frontend/src/lib/components/raw_apps/RawAppEditor.svelte b/frontend/src/lib/components/raw_apps/RawAppEditor.svelte index 31e99dbe89..7dc2b99d7e 100644 --- a/frontend/src/lib/components/raw_apps/RawAppEditor.svelte +++ b/frontend/src/lib/components/raw_apps/RawAppEditor.svelte @@ -54,6 +54,7 @@ } from 'lucide-svelte' import DraggableTabs, { type TabItem } from '$lib/components/common/tabs/DraggableTabs.svelte' import { runScriptAndPollResult } from '../jobs/utils' + import { writingJobOptions } from '../jobs/writingJob' import { RawAppHistoryManager } from './RawAppHistoryManager.svelte' import { sendUserToast } from '$lib/utils' import { UserDraftDbSyncer } from '$lib/userDraftDbSyncer.svelte' @@ -1020,14 +1021,17 @@ } try { - const result = await runScriptAndPollResult({ - workspace: opWorkspace, - requestBody: { - language: 'postgresql', - content: sql, - args: { database: `datatable://${datatableName}` } - } - }) + const result = await runScriptAndPollResult( + { + workspace: opWorkspace, + requestBody: { + language: 'postgresql', + content: sql, + args: { database: `datatable://${datatableName}` } + } + }, + writingJobOptions + ) // If newTable was specified and the query succeeded, add it to data.tables if (newTable) { diff --git a/frontend/src/lib/components/workspaceSettings/DataTableMigrationsButton.svelte b/frontend/src/lib/components/workspaceSettings/DataTableMigrationsButton.svelte index 307c3eecd7..bb908a433c 100644 --- a/frontend/src/lib/components/workspaceSettings/DataTableMigrationsButton.svelte +++ b/frontend/src/lib/components/workspaceSettings/DataTableMigrationsButton.svelte @@ -25,6 +25,7 @@ import Tooltip from '../Tooltip.svelte' import Portal from '$lib/components/Portal.svelte' import DropdownV2 from '../DropdownV2.svelte' + import MissingWorkerTagAlert from '../jobs/MissingWorkerTagAlert.svelte' import { superadmin, userStore } from '$lib/stores' let { @@ -422,6 +423,7 @@ {/if} {:else} + {#if loadError}
Could not read applied status from the data table: {loadError} diff --git a/frontend/src/lib/components/workspaceSettings/DataTableSettings.svelte b/frontend/src/lib/components/workspaceSettings/DataTableSettings.svelte index 069ef6093d..254a20fe80 100644 --- a/frontend/src/lib/components/workspaceSettings/DataTableSettings.svelte +++ b/frontend/src/lib/components/workspaceSettings/DataTableSettings.svelte @@ -87,6 +87,7 @@ import { clone } from '$lib/utils' import SettingsFooter from './SettingsFooter.svelte' import Alert from '../common/alert/Alert.svelte' + import MissingWorkerTagAlert from '../jobs/MissingWorkerTagAlert.svelte' import { isCloudHosted } from '$lib/cloud' type Props = { @@ -252,6 +253,8 @@ {/if} + + diff --git a/frontend/src/lib/components/workspaceSettings/projectInstall.ts b/frontend/src/lib/components/workspaceSettings/projectInstall.ts index 387dc82ade..abc48a3ad6 100644 --- a/frontend/src/lib/components/workspaceSettings/projectInstall.ts +++ b/frontend/src/lib/components/workspaceSettings/projectInstall.ts @@ -23,6 +23,7 @@ import { updateRawAppPolicy } from '$lib/sharedUtils' import { apiErrorMessage as errorMessage } from '$lib/utils' import type { App } from '$lib/components/apps/types' import { runScriptAndPollResult } from '$lib/components/jobs/utils' +import { writingJobOptions } from '$lib/components/jobs/writingJob' import { classifyPath, collectExportVarPaths, @@ -233,14 +234,17 @@ async function applyOneMigration( only: created.timestamp }) } else { - await runScriptAndPollResult({ - workspace, - requestBody: { - language: 'postgresql', - content: m.sql, - args: { database: `datatable://${m.datatable_name}` } - } - }) + await runScriptAndPollResult( + { + workspace, + requestBody: { + language: 'postgresql', + content: m.sql, + args: { database: `datatable://${m.datatable_name}` } + } + }, + writingJobOptions + ) } }