From 2bf9cfcd59e1101695160524c31a4da1a41699dc Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Tue, 6 Oct 2026 10:57:54 +0200 Subject: [PATCH] fix: search slow runs filters by time windows instead of timing out (#11549) * fix: search slow runs filters by time windows instead of timing out * fix: cancel a pending runs page when its listing is reset * fix: keep runs scan windows disjoint at microsecond precision * fix: race the single runs request against the window scan * fix: list the cursor window when a runs scan starts at the oldest job * fix: show the runs history scan as an info alert * fix: show runs scan progress as a bar and keep the session button on the searchbar row * fix: stop a runs scan whose windows are no faster when narrower * fix: right-align the runs searchbar row when the header wraps * fix: keep the runs scan stop button inside a narrow list pane --- backend/windmill-api/openapi.yaml | 25 ++ backend/windmill-api/src/jobs.rs | 36 ++ .../lib/components/BatchLoadProgress.svelte | 10 +- frontend/src/lib/components/RunsPage.svelte | 82 +++-- .../components/runs/jobsWindowScan.test.ts | 140 ++++++++ .../src/lib/components/runs/jobsWindowScan.ts | 139 ++++++++ .../components/runs/useJobsLoader.svelte.ts | 317 +++++++++++++++--- 7 files changed, 667 insertions(+), 82 deletions(-) create mode 100644 frontend/src/lib/components/runs/jobsWindowScan.test.ts create mode 100644 frontend/src/lib/components/runs/jobsWindowScan.ts diff --git a/backend/windmill-api/openapi.yaml b/backend/windmill-api/openapi.yaml index 221a3fb886..62b0e88b50 100644 --- a/backend/windmill-api/openapi.yaml +++ b/backend/windmill-api/openapi.yaml @@ -16536,6 +16536,31 @@ paths: required: - database_length + /w/{workspace}/jobs/oldest: + get: + summary: get the creation date of the oldest job + operationId: getOldestJob + tags: + - job + parameters: + - $ref: "#/components/parameters/WorkspaceId" + - name: all_workspaces + description: consider jobs from all workspaces (only valid if request come from the `admins` workspace) + in: query + schema: + type: boolean + responses: + "200": + description: creation date of the oldest job, absent when there is none + content: + application/json: + schema: + type: object + properties: + created_at: + type: string + format: date-time + /w/{workspace}/jobs/completed/count: get: summary: get completed count diff --git a/backend/windmill-api/src/jobs.rs b/backend/windmill-api/src/jobs.rs index ac3f06ef81..6ce04243d3 100644 --- a/backend/windmill-api/src/jobs.rs +++ b/backend/windmill-api/src/jobs.rs @@ -292,6 +292,7 @@ pub fn workspaced_service() -> Router { ) .route("/run/dynamic_select", post(run_dynamic_select)) .route("/list", get(list_jobs)) + .route("/oldest", get(get_oldest_job)) .route("/asset_dispatch_edges", get(list_asset_dispatch_edges)) .route( "/list_selected_job_groups", @@ -4728,6 +4729,41 @@ async fn count_completed_jobs( )) } +#[derive(Deserialize)] +struct OldestJobQuery { + all_workspaces: Option, +} + +#[derive(Serialize)] +struct OldestJob { + #[serde(skip_serializing_if = "Option::is_none")] + created_at: Option>, +} + +/// Where the job history starts, whatever the caller may see of it: a client scanning the +/// runs backwards in time windows needs a bound that row-level visibility cannot make slow. +async fn get_oldest_job( + Extension(db): Extension, + Path(w_id): Path, + Query(q): Query, +) -> error::JsonResult { + let created_at = if w_id == "admins" && q.all_workspaces.unwrap_or(false) { + // One index probe per workspace: a plain min() over the table has no index to walk. + sqlx::query_scalar( + "SELECT min(o.created_at) FROM workspace w CROSS JOIN LATERAL \ + (SELECT min(created_at) AS created_at FROM v2_job WHERE workspace_id = w.id) o", + ) + .fetch_one(&db) + .await? + } else { + sqlx::query_scalar("SELECT min(created_at) FROM v2_job WHERE workspace_id = $1") + .bind(&w_id) + .fetch_one(&db) + .await? + }; + Ok(Json(OldestJob { created_at })) +} + lazy_static::lazy_static! { /// 0 keeps the connection-wide statement_timeout. static ref LIST_JOBS_STATEMENT_TIMEOUT_SECS: u64 = diff --git a/frontend/src/lib/components/BatchLoadProgress.svelte b/frontend/src/lib/components/BatchLoadProgress.svelte index 16cf2d7212..a41151b210 100644 --- a/frontend/src/lib/components/BatchLoadProgress.svelte +++ b/frontend/src/lib/components/BatchLoadProgress.svelte @@ -6,6 +6,8 @@ loaded: number total: number itemsLabel: string + /** Replaces the default "Loading … of …" text, for progress that is not a count of items. */ + label?: string batchSize?: number | null /** Largest batch the caller can fetch in one request, when that is below the page size. */ batchSizeCap?: number @@ -17,6 +19,7 @@ loaded, total, itemsLabel, + label, batchSize = null, batchSizeCap, onBatchSizeChange, @@ -30,7 +33,12 @@
- Loading {itemsLabel}: {loaded} of {total}... + {#if label} + + {label} + {:else} + Loading {itemsLabel}: {loaded} of {total}... + {/if}
- - - pageHref(RUNS_PATH) + window.location.search, - workspaceId: $workspaceStore ?? undefined - }} - btnProps={{ unifiedSize: 'md' }} - /> + +
+ + + +
+ pageHref(RUNS_PATH) + window.location.search, + workspaceId: $workspaceStore ?? undefined + }} + btnProps={{ unifiedSize: 'md' }} + /> +
+
@@ -939,7 +949,21 @@
-
+
+ {#if scanProgress} +
+ +
+ jobsLoader.stopBatchLoading()} + /> +
+
+ {/if} {#if batchProgress}
{ + windows.push(w) + const lo = Date.parse(w.after) + const hi = w.before ? Date.parse(w.before) : Infinity + if (hi - lo > slowOver) return new CancelablePromise(() => {}) + const rows = at + .filter((t) => t >= lo && t <= hi) + .sort((a, b) => b - a) + .slice(0, w.limit) + .map(job) + return new CancelablePromise((resolve) => resolve(rows)) + } + return { windows, fetchWindow } +} + +describe('scanJobWindows', () => { + it('lists every match down to the oldest job, in order, without a gap between windows', async () => { + const at = [NOW - 2 * HOUR, NOW - 30 * HOUR, NOW - 200 * HOUR] + const { windows, fetchWindow } = history(at) + const seen: Job[] = [] + const wentToEnd = await scanJobWindows({ + before: new Date(NOW).toISOString(), + after: null, + pageSize: 10, + oldest: async () => new Date(NOW - 240 * HOUR).toISOString(), + fetchWindow, + onWindow: (jobs) => seen.push(...jobs) + }) + expect(wentToEnd).toBe(true) + expect(seen.map((j) => j.id)).toEqual(at.map((t) => `j${t}`)) + expect(windows.at(-1)!.after).toBe(new Date(NOW - 240 * HOUR).toISOString()) + }) + + it('counts a job created exactly on a window bound once', async () => { + // The first window is an hour wide, so its lower bound is the first job's creation time. + const at = [NOW - HOUR, NOW - 2 * HOUR, NOW - 100 * HOUR] + const { fetchWindow } = history(at) + const seen: Job[] = [] + const wentToEnd = await scanJobWindows({ + before: new Date(NOW).toISOString(), + after: new Date(NOW - 240 * HOUR).toISOString(), + pageSize: 2, + oldest: async () => undefined, + fetchWindow, + onWindow: (jobs) => seen.push(...jobs) + }) + expect(wentToEnd).toBe(false) + expect(seen.map((j) => j.id)).toEqual([`j${NOW - HOUR}`, `j${NOW - 2 * HOUR}`]) + }) + + it('still lists the window of a cursor that sits on the oldest job', async () => { + const oldest = NOW - 240 * HOUR + const { windows, fetchWindow } = history([oldest]) + const seen: Job[] = [] + const before = new Date(oldest).toISOString().replace('Z', '500Z') + const wentToEnd = await scanJobWindows({ + before, + after: null, + pageSize: 10, + oldest: async () => new Date(oldest).toISOString(), + fetchWindow, + onWindow: (jobs) => seen.push(...jobs) + }) + expect(wentToEnd).toBe(true) + expect(windows).toEqual([{ before, after: new Date(oldest).toISOString(), limit: 10 }]) + expect(seen.map((j) => j.id)).toEqual([`j${oldest}`]) + }) + + it('stops as soon as the page is full, asking each window only for what is missing', async () => { + const at = [NOW - 0.5 * HOUR, NOW - 2 * HOUR, NOW - 3 * HOUR, NOW - 100 * HOUR] + const { windows, fetchWindow } = history(at) + const seen: Job[] = [] + const wentToEnd = await scanJobWindows({ + before: new Date(NOW).toISOString(), + after: new Date(NOW - 240 * HOUR).toISOString(), + pageSize: 3, + oldest: async () => undefined, + fetchWindow, + onWindow: (jobs) => seen.push(...jobs) + }) + expect(wentToEnd).toBe(false) + expect(seen).toHaveLength(3) + expect(windows.map((w) => w.limit)).toEqual([3, 2]) + }) + + it('gives up when narrower windows are no faster', async () => { + // Every window costs the same whatever its width, as with an index that ignores the bounds. + const windows: JobWindow[] = [] + const fetchWindow = (w: JobWindow) => { + windows.push(w) + return new CancelablePromise((resolve) => setTimeout(() => resolve([]), 30)) + } + await expect( + scanJobWindows({ + before: new Date(NOW).toISOString(), + after: new Date(NOW - 240 * HOUR).toISOString(), + pageSize: 10, + oldest: async () => undefined, + fetchWindow, + onWindow: () => {}, + targetMs: 10, + patienceMs: 1000 + }) + ).rejects.toThrow('too slow') + expect(windows.length).toBeLessThan(10) + }) + + it('retries a window that does not answer in time with a narrower one', async () => { + const at = [NOW - 0.2 * HOUR, NOW - 0.9 * HOUR] + const { windows, fetchWindow } = history(at, 0.5 * HOUR) + const seen: Job[] = [] + const wentToEnd = await scanJobWindows({ + before: new Date(NOW).toISOString(), + after: new Date(NOW - HOUR).toISOString(), + pageSize: 10, + oldest: async () => undefined, + fetchWindow, + onWindow: (jobs) => seen.push(...jobs), + patienceMs: 5 + }) + expect(wentToEnd).toBe(true) + expect(seen.map((j) => j.id)).toEqual(at.map((t) => `j${t}`)) + expect(windows.length).toBeGreaterThan(2) + }) +}) diff --git a/frontend/src/lib/components/runs/jobsWindowScan.ts b/frontend/src/lib/components/runs/jobsWindowScan.ts new file mode 100644 index 0000000000..c273edbceb --- /dev/null +++ b/frontend/src/lib/components/runs/jobsWindowScan.ts @@ -0,0 +1,139 @@ +import { CancelablePromise, type Job } from '$lib/gen' + +// A window is sized so that listing it takes about this long: short enough that the scan keeps +// reporting progress, long enough that a sparse history is not crossed one round trip at a time. +const TARGET_MS = 2000 +// Past this the window is dropped and retried narrower. The server keeps running a dropped +// query until its own statement timeout, so it has to be rare: a window grows by at most +// MAX_STEP per round, which keeps a well-sized one under PATIENCE_MS = MAX_STEP * TARGET_MS. +const MAX_STEP = 4 +const PATIENCE_MS = MAX_STEP * TARGET_MS +const INITIAL_WIDTH_MS = 60 * 60 * 1000 +const MIN_WIDTH_MS = 60 * 1000 + +function nextWindowWidth(widthMs: number, elapsedMs: number, targetMs: number): number { + const step = Math.min(MAX_STEP, Math.max(1 / MAX_STEP, targetMs / Math.max(elapsedMs, 1))) + return Math.max(MIN_WIDTH_MS, widthMs * step) +} + +// A window's cost is expected to follow its width. It does not when the database answers it +// from an index that ignores the time bounds (another filter's own index, or no index leading +// with the sort column): every window then costs a pass over the whole history, and scanning +// would multiply the load it exists to bound. +function tooSlow(): Error { + return new Error( + `Listing the jobs of a single ${MIN_WIDTH_MS / 1000}s window is too slow. Narrow the filters.` + ) +} + +export interface JobWindow { + /** Upper bound, absent on a window that starts at the newest job. */ + before?: string + after: string + limit: number +} + +export interface JobWindowScan { + /** Upper bound of the scan, sent as is on the first window. null starts at the newest job. */ + before: string | null + /** Lower bound of the scan. null ends it at the oldest job, as given by `oldest`. */ + after: string | null + pageSize: number + oldest: () => Promise + fetchWindow: (w: JobWindow) => CancelablePromise + /** `fraction` is the share of the scan's time range listed so far. */ + onWindow: (jobs: Job[], scannedTo: string, fraction: number) => void + targetMs?: number + patienceMs?: number +} + +const SLOW = Symbol('slow') + +// Both bounds of a window are inclusive on the server, whose timestamps have microsecond +// precision. A window therefore ends one microsecond before the previous one started: sharing +// the bound would list a job created exactly on it twice, and count it twice toward the page. +function justBefore(ms: number): string { + return new Date(ms - 1).toISOString().replace('Z', '999Z') +} + +/** + * Lists the completed jobs of a time range newest first, one bounded window after the other, + * which is the same page a single request over the whole range returns, except that no request + * has to scan more history than the server allows one to. Resolves to whether the range was + * scanned to its end, false meaning the page filled up first. + */ +export function scanJobWindows(scan: JobWindowScan): CancelablePromise { + const targetMs = scan.targetMs ?? TARGET_MS + const patienceMs = scan.patienceMs ?? PATIENCE_MS + return new CancelablePromise((resolve, reject, onCancel) => { + let current: CancelablePromise | undefined + onCancel(() => current?.cancel()) + + async function fetchWithPatience(w: JobWindow): Promise { + const request = scan.fetchWindow(w) + current = request + let timer: ReturnType | undefined + const slow = new Promise((r) => (timer = setTimeout(() => r(SLOW), patienceMs))) + try { + const res = await Promise.race([request, slow]) + if (res === SLOW) { + request.catch(() => {}) + request.cancel() + } + return res + } finally { + clearTimeout(timer) + } + } + + async function run(): Promise { + const floorTs = scan.after ?? (await scan.oldest()) + if (!floorTs) return true + const floor = new Date(floorTs).getTime() + let upper = scan.before ? new Date(scan.before).getTime() : Date.now() + const start = upper + let first = true + let width = INITIAL_WIDTH_MS + let listed = 0 + // The first window is listed even when the scan starts at its floor: the bounds are + // inclusive, and jobs sharing the cursor's timestamp may still be missing from the page. + while (first || upper > floor) { + if (onCancel.isCancelled) return false + const lower = Math.min(upper, Math.max(floor, upper - width)) + const w: JobWindow = { + before: first ? (scan.before ?? undefined) : justBefore(upper), + after: new Date(lower).toISOString(), + limit: scan.pageSize - listed + } + const startedAt = Date.now() + let res: Job[] | typeof SLOW + try { + res = await fetchWithPatience(w) + } catch (e) { + // A window the server refused to finish is retried narrower like a slow one. + if (onCancel.isCancelled || width <= MIN_WIDTH_MS) throw e + res = SLOW + } + if (res === SLOW) { + if (width <= MIN_WIDTH_MS) throw tooSlow() + // Well below what the elapsed time suggests: each dropped window is one more + // statement left running on the server, so the retry has to land. + width = Math.max(MIN_WIDTH_MS, width / (MAX_STEP * MAX_STEP)) + continue + } + const completed = res.filter((j) => j.type === 'CompletedJob').length + scan.onWindow(res, w.after, start > floor ? (start - lower) / (start - floor) : 1) + if (completed >= w.limit) return false + listed += completed + upper = lower + first = false + const elapsed = Date.now() - startedAt + if (width <= MIN_WIDTH_MS && elapsed > targetMs && upper > floor) throw tooSlow() + width = nextWindowWidth(width, elapsed, targetMs) + } + return true + } + + run().then(resolve, reject) + }) +} diff --git a/frontend/src/lib/components/runs/useJobsLoader.svelte.ts b/frontend/src/lib/components/runs/useJobsLoader.svelte.ts index 828f060972..8055278c77 100644 --- a/frontend/src/lib/components/runs/useJobsLoader.svelte.ts +++ b/frontend/src/lib/components/runs/useJobsLoader.svelte.ts @@ -17,10 +17,27 @@ import { subtractDaysFromDateString } from '$lib/utils' import { CancelablePromiseUtils } from '$lib/cancelable-promise-utils' import type { Timeframe } from './timeframes' import { allowWildcards as _allowWildcards, type RunsFilterInstance } from './runsFilter' +import { scanJobWindows } from './jobsWindowScan' // windmill_common::utils::MAX_PER_PAGE: the server silently caps per_page at this value const MAX_PER_PAGE = 10000 +// How long a listing may run before a scan by time windows starts alongside it. A filter +// matching fewer jobs than the page holds walks the whole history to find that out, which the +// server stops after its statement timeout, with nothing to show for it. +const WINDOW_SCAN_AFTER_MS = 4000 + +const EPOCH = new Date(0).toISOString() + +interface JobsQuery { + completedBefore?: string | null + completedAfter?: string | null + createdBefore?: string + createdAfter?: string + createdAfterQueue?: string + perPage?: number +} + export function computeJobKinds(jobKindsCat: string | null): string { if (jobKindsCat == 'all') { return '' @@ -103,6 +120,13 @@ export function useJobsLoader(args: () => UseJobLoaderArgs) { let loadingExtra = $state(false) let lastFetchWentToEnd = $state(true) let batchProgress = $state<{ loaded: number; total: number } | null>(null) + let scanProgress = $state<{ scannedTo: string; fraction: number } | null>(null) + let activeScan: CancelablePromise | undefined + // The page being listed, single request or scan. Only the filter-change load is otherwise + // tracked for cancellation, and a page left running writes into whatever listing replaces its own. + let activePage: CancelablePromise | undefined + // Queue-only views list no completed job, and concurrency-key views go through another endpoint. + let canScanWindows = $derived(!isQueueOnly && (concurrencyKey == null || concurrencyKey === '')) let completedJobs: CompletedJob[] | undefined = $state() let externalJobs: Job[] | undefined = $state() @@ -119,6 +143,7 @@ export function useJobsLoader(args: () => UseJobLoaderArgs) { function onParamChanges() { resetJobs() slowStreamIntervalId = setInterval(() => { + if (scanProgress) return sendUserToast( 'Loading is taking a long time...', 'warning', @@ -128,7 +153,7 @@ export function useJobsLoader(args: () => UseJobLoaderArgs) { ) }, 15000) let promise = loadJobsIntern(true) - if (perPage > 25) { + if (perPage > 25 && !canScanWindows) { promise = CancelablePromiseUtils.onTimeout(promise, 4000, () => { const noStartDate = timeframe?.computeMinMax().minTs == null sendUserToast( @@ -159,6 +184,7 @@ export function useJobsLoader(args: () => UseJobLoaderArgs) { paramChangePromise?.cancel() resetJobs() slowStreamIntervalId = setInterval(() => { + if (scanProgress) return sendUserToast( 'Loading is taking a long time...', 'warning', @@ -196,8 +222,13 @@ export function useJobsLoader(args: () => UseJobLoaderArgs) { loading = true loadingExtra = true const batchSize = Math.min(perPage, 1000) - await loadExtraJobsBatch(batchSize) - loadingExtra = false + try { + await loadExtraJobsBatch(batchSize) + } catch (e) { + if (!(e instanceof CancelError)) throw e + } finally { + loadingExtra = false + } } // Mirrors when list_completed_jobs_query sorts by completed_at. A created_at cursor does not @@ -231,37 +262,201 @@ export function useJobsLoader(args: () => UseJobLoaderArgs) { const cursor = stepOver ? new Date(new Date(cursorTs).getTime() - 1).toISOString() : cursorTs const pageSize = stepOver ? batchSize : Math.min(batchSize + tied, MAX_PER_PAGE) return CancelablePromiseUtils.map( - byCompletedAt - ? fetchJobs(cursor, minTs, undefined, undefined, pageSize) - : fetchJobs(null, minTs, undefined, cursor, pageSize), - (olderJobs) => { - jobs = updateWithNewJobs(olderJobs ?? [], jobs ?? []) - if (extendedJobs) { - extendedJobs.jobs = jobs ?? [] - extendedJobs = extendedJobs + fetchPage( + { + byCompletedAt, + before: cursor, + after: minTs, + pageSize, + query: byCompletedAt + ? { completedBefore: cursor, completedAfter: minTs, perPage: pageSize } + : { completedAfter: minTs, createdBefore: cursor, perPage: pageSize } + }, + (olderJobs) => { + jobs = updateWithNewJobs(olderJobs, jobs ?? []) + if (extendedJobs) { + extendedJobs.jobs = jobs ?? [] + extendedJobs = extendedJobs + } + computeCompletedJobs() } - computeCompletedJobs() - lastFetchWentToEnd = (olderJobs?.length ?? 0) < pageSize + ), + (wentToEnd) => { + lastFetchWentToEnd = wentToEnd loading = false } ) } - function fetchJobs( - completedBefore: string | null, - completedAfter: string | null, - createdAfterQueue: string | undefined, - createdBefore?: string, - perPageOverride?: number - ): CancelablePromise { + // Lists one page of jobs older than `before`, handing them to `onJobs` as they arrive, and + // resolves to whether the listing reached the end of the range. `query` is the whole page in + // one request. When that is slow, the page is also listed by time windows, and whichever + // completes it first wins: the server runs an abandoned request to its timeout regardless, so + // dropping it for the scan would only throw away an answer already being paid for. + function fetchPage( + page: { + byCompletedAt: boolean + before: string | null + after: string | null + pageSize: number + withQueue?: boolean + query: JobsQuery + }, + onJobs: (jobs: Job[]) => void + ): CancelablePromise { + if (!canScanWindows || _args.skip) { + return CancelablePromiseUtils.map(fetchJobs(page.query), (res) => { + onJobs(res) + return res.length < page.pageSize + }) + } + const fetched = new CancelablePromise((resolve, reject, onCancel) => { + const single = listJobs(page.query) + let scan: CancelablePromise | undefined + onCancel(() => { + clearTimeout(timer) + single.cancel() + scan?.cancel() + }) + // Either side failing leaves the page to the other: it is lost only once both have. + let scanError: any + let singleError: any + const giveUp = (e: any, wentToEnd: boolean) => { + sendUserToast(`Could not load jobs: ${e.body ?? e.message}`, true) + console.error(e) + onJobs([]) + resolve(wentToEnd) + } + const timer = setTimeout(() => { + scan = scanWindows(page, onJobs) + scan.then( + (wentToEnd) => { + resolve(wentToEnd) + single.cancel() + }, + (e) => { + if (e instanceof CancelError) return reject(e) + scanError = e + if (singleError) giveUp(scanError, false) + } + ) + }, WINDOW_SCAN_AFTER_MS) + single.then( + (res) => { + clearTimeout(timer) + onJobs(res) + resolve(res.length < page.pageSize) + scan?.cancel() + }, + (e: any) => { + if (e instanceof CancelError) return + if (!scan) { + clearTimeout(timer) + giveUp(e, true) + return + } + // Once the scan runs, the single request failing is the expected server timeout. + singleError = e + if (scanError) giveUp(scanError, false) + } + ) + }) + activePage?.cancel() + activePage = fetched + const done = () => { + if (activePage === fetched) activePage = undefined + } + fetched.then(done, done) + return fetched + } + + function scanWindows( + page: { + byCompletedAt: boolean + before: string | null + after: string | null + pageSize: number + withQueue?: boolean + }, + onJobs: (jobs: Job[]) => void + ): CancelablePromise { + const workspace = currentWorkspace + const scan = scanJobWindows({ + before: page.before, + after: page.after, + pageSize: page.pageSize, + oldest: async () => + (await JobService.getOldestJob({ workspace, allWorkspaces: allWorkspaces || undefined })) + .created_at, + fetchWindow: (w) => { + // Only the first window of a listing that started at the newest job carries the queue. + // A created_before excludes it from the others, and bounds no completed job out of a + // completed_at window since a job is created before it completes. + const withQueue = page.withQueue && w.before == page.before + return listJobs( + page.byCompletedAt + ? { + completedBefore: w.before, + completedAfter: w.after, + createdBefore: withQueue ? undefined : w.before, + createdAfterQueue: page.after ?? undefined, + perPage: w.limit + } + : { + createdBefore: w.before, + createdAfter: w.after, + // created_after would bound the queue too, hiding older jobs still in it + createdAfterQueue: withQueue ? EPOCH : undefined, + perPage: w.limit + } + ) + }, + onWindow: (res, scannedTo, fraction) => { + if (activeScan !== scan) return + scanProgress = { scannedTo, fraction } + onJobs(res) + } + }) + activeScan?.cancel() + activeScan = scan + scanProgress = { scannedTo: page.before ?? new Date().toISOString(), fraction: 0 } + onJobs([]) + const done = () => { + if (activeScan === scan) { + activeScan = undefined + scanProgress = null + } + } + scan.then(done, done) + return scan + } + + function fetchJobs(q: JobsQuery): CancelablePromise { if (_args.skip) return CancelablePromiseUtils.pure([]) + return CancelablePromiseUtils.catchErr(listJobs(q), (e) => { + if (e instanceof CancelError) return CancelablePromiseUtils.err(e) + sendUserToast(`Could not load jobs: ${e.body ?? e.message}`, true) + console.error(e) + return CancelablePromiseUtils.pure([]) + }) + } + + function listJobs({ + completedBefore, + completedAfter, + createdBefore, + createdAfter, + createdAfterQueue, + perPage: perPageOverride + }: JobsQuery): CancelablePromise { loadingFetch = true let scriptPathStart = folder == null || folder === '' ? undefined : `f/${folder}/` let scriptPathExact = path == null || path === '' ? undefined : path let isCompletedOnly = success == 'success' || success == 'failure' || success == 'canceled' let promise = JobService.listJobs({ workspace: currentWorkspace, - createdBefore: createdBefore ?? undefined, + createdBefore, + createdAfter, completedBefore: isQueueOnly ? undefined : (completedBefore ?? undefined), completedAfter: isQueueOnly ? undefined : (completedAfter ?? undefined), createdBeforeQueue: isQueueOnly ? (completedBefore ?? undefined) : undefined, @@ -314,15 +509,10 @@ export function useJobsLoader(args: () => UseJobLoaderArgs) { broadFilter, excludesEntrypointOverride: excludesEntrypointOverride ? true : undefined }) - promise = CancelablePromiseUtils.catchErr(promise, (e) => { - if (e instanceof CancelError) return CancelablePromiseUtils.err(e) - sendUserToast(`Could not load jobs: ${e.body ?? e.message}`, true) - console.error(e) - return CancelablePromiseUtils.pure([]) - }) - CancelablePromiseUtils.pipe(promise, () => { + const settled = () => { loadingFetch = false - }) + } + promise.then(settled, settled) return promise } @@ -387,11 +577,16 @@ export function useJobsLoader(args: () => UseJobLoaderArgs) { async function loadJobs(reset: boolean, shouldGetCount?: boolean): Promise { if (reset) resetJobs() - await loadJobsIntern(shouldGetCount) + try { + await loadJobsIntern(shouldGetCount) + } catch (e) { + if (!(e instanceof CancelError)) throw e + } } function stopBatchLoading(): void { paramChangePromise?.cancel() + activePage?.cancel() if (slowStreamIntervalId) { clearInterval(slowStreamIntervalId) slowStreamIntervalId = undefined @@ -402,6 +597,7 @@ export function useJobsLoader(args: () => UseJobLoaderArgs) { } function resetJobs() { + activePage?.cancel() jobs = undefined completedJobs = undefined externalJobs = undefined @@ -438,15 +634,35 @@ export function useJobsLoader(args: () => UseJobLoaderArgs) { batchProgress = { loaded: 0, total: perPage } } currentBatchSize = batchSize - let slowBatchToastShown = false - let firstFetchPromise = fetchJobs( - maxTs, - extendedMinTs ?? null, - extendedMinTs, - undefined, - batchSize + // The windowed scan streams a slow listing by itself, whatever the batch size. + let slowBatchToastShown = canScanWindows + let firstJobs: Job[] = [] + let firstFetchPromise = fetchPage( + { + byCompletedAt: sortsByCompletedAt(minTs, maxTs), + before: maxTs, + after: extendedMinTs ?? null, + pageSize: batchSize, + withQueue: true, + query: { + completedBefore: maxTs, + completedAfter: extendedMinTs ?? null, + createdAfterQueue: extendedMinTs, + perPage: batchSize + } + }, + (newJobs) => { + firstJobs = updateWithNewJobs(newJobs, firstJobs) + extendedJobs = { jobs: firstJobs, obscured_jobs: [] } as ExtendedJobs + + // Filter on minTs here and not in the backend + // to get enough data for the concurrency graph + jobs = sortMinDate(minTs, firstJobs) + externalJobs = [] + computeCompletedJobs() + } ) - if (isBatched && batchSize > 1) { + if (isBatched && batchSize > 1 && !slowBatchToastShown) { firstFetchPromise = CancelablePromiseUtils.onTimeout(firstFetchPromise, 4000, () => { if (!slowBatchToastShown) { slowBatchToastShown = true @@ -458,15 +674,8 @@ export function useJobsLoader(args: () => UseJobLoaderArgs) { } }) } - let p = CancelablePromiseUtils.map(firstFetchPromise, (newJobs) => { - extendedJobs = { jobs: newJobs, obscured_jobs: [] } as ExtendedJobs - - // Filter on minTs here and not in the backend - // to get enough data for the concurrency graph - jobs = sortMinDate(minTs, newJobs) - externalJobs = [] - computeCompletedJobs() - lastFetchWentToEnd = newJobs.length < batchSize + let p = CancelablePromiseUtils.map(firstFetchPromise, (wentToEnd) => { + lastFetchWentToEnd = wentToEnd loading = false if (isBatched) { batchProgress = { loaded: jobs?.length ?? 0, total: perPage } @@ -608,11 +817,11 @@ export function useJobsLoader(args: () => UseJobLoaderArgs) { // unbounded scan of completed jobs, possibly the one that just timed out. Not for // queue-only views: fetchJobs turns this into the queue's created_at bound, hiding // older jobs that suspend or come due. The margin absorbs browser/database skew. - newJobs = await fetchJobs( - maxTs, - minTs ?? completedTs ?? (isQueueOnly ? null : listLoadedAt), - queueTs - ) + newJobs = await fetchJobs({ + completedBefore: maxTs, + completedAfter: minTs ?? completedTs ?? (isQueueOnly ? null : listLoadedAt), + createdAfterQueue: queueTs + }) } else { // Obscured jobs have no ids, so we have to do the full request extendedJobs = await fetchExtendedJobs(concurrencyKey, maxTs, minTs ?? completedTs) @@ -721,6 +930,7 @@ export function useJobsLoader(args: () => UseJobLoaderArgs) { slowStreamIntervalId = undefined } paramChangePromise?.cancel() + activePage?.cancel() }) $effect(() => { Object.keys(filters ?? {}).map((k) => filters?.[k as keyof RunsFilterInstance]) @@ -772,6 +982,9 @@ export function useJobsLoader(args: () => UseJobLoaderArgs) { get currentBatchSize() { return currentBatchSize }, + get scanProgress() { + return scanProgress + }, get queue_count() { return queue_count },