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
This commit is contained in:
Ruben Fiszel
2026-10-06 08:57:54 +00:00
committed by GitHub
parent 18e44d3174
commit 2bf9cfcd59
7 changed files with 667 additions and 82 deletions
+25
View File
@@ -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
+36
View File
@@ -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<bool>,
}
#[derive(Serialize)]
struct OldestJob {
#[serde(skip_serializing_if = "Option::is_none")]
created_at: Option<chrono::DateTime<chrono::Utc>>,
}
/// 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<DB>,
Path(w_id): Path<String>,
Query(q): Query<OldestJobQuery>,
) -> error::JsonResult<OldestJob> {
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 =
@@ -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 @@
</script>
<div class="flex items-center gap-3 text-xs text-secondary">
<span class="whitespace-nowrap shrink-0">Loading {itemsLabel}: {loaded} of {total}...</span>
{#if label}
<!-- A custom label has no bounded length: it gives way before the bar and Stop do. -->
<span class="min-w-0 truncate" title={label}>{label}</span>
{:else}
<span class="whitespace-nowrap shrink-0">Loading {itemsLabel}: {loaded} of {total}...</span>
{/if}
<div class="flex-1 min-w-8 bg-surface-hover rounded-full h-1.5">
<div
class="bg-blue-500 h-1.5 rounded-full transition-all duration-300"
+53 -29
View File
@@ -10,6 +10,7 @@
} from '$lib/gen'
import { sendUserToast } from '$lib/toast'
import { displayDate } from '$lib/utils'
import {
userStore,
workspaceStore,
@@ -62,6 +63,7 @@
CirclePlay,
Clock,
Hourglass,
Loader2,
TriangleAlertIcon
} from 'lucide-svelte'
import DropdownV2 from './DropdownV2.svelte'
@@ -189,6 +191,7 @@
}))
let batchProgress = $derived(jobsLoader.batchProgress)
let currentBatchSize = $derived(jobsLoader.currentBatchSize)
let scanProgress = $derived(jobsLoader.scanProgress)
let lastFetchWentToEnd = $derived(jobsLoader.lastFetchWentToEnd)
let queue_count = $derived(jobsLoader.queue_count)
let suspended_count = $derived(jobsLoader.suspended_count)
@@ -837,34 +840,41 @@
items={runsTimeframes}
bind:value={_timeframe.val}
/>
<FilterSearchbar
class={twMerge(
'flex-1 relative min-w-[18rem]',
Object.keys(filters.val).length <= 3
? 'max-w-[20rem] 2xl:max-w-[28rem]'
: 'max-w-[34rem]',
ButtonType.UnifiedMinHeightClasses.md
)}
schema={runsFilterSearchbarSchema}
presets={buildRunsFilterPresets({
isSuperAdminOrDevops: !!$superadmin || !!$devopsRole,
isAdminsWorkspace: $workspaceStore === 'admins'
})}
bind:value={filters.val}
placeholder="Filter runs..."
autofocus
/>
<!-- The filters are shallow-routed, so the search has to come off
`window.location` at click time — `page.url` never sees them. Always the
canonical `/runs`: only that is a recognized preview page, and the
`/runs/<path>` route mirrors its path into `?path=` anyway. -->
<OpenInSessionButton
source={{
page: () => pageHref(RUNS_PATH) + window.location.search,
workspaceId: $workspaceStore ?? undefined
}}
btnProps={{ unifiedSize: 'md' }}
/>
<!-- One flex item, so the session button never wraps to a row of its own. -->
<div class="flex flex-1 items-start justify-end gap-3">
<FilterSearchbar
class={twMerge(
'flex-1 relative min-w-[14rem]',
Object.keys(filters.val).length <= 3
? 'max-w-[20rem] 2xl:max-w-[28rem]'
: 'max-w-[34rem]',
ButtonType.UnifiedMinHeightClasses.md
)}
schema={runsFilterSearchbarSchema}
presets={buildRunsFilterPresets({
isSuperAdminOrDevops: !!$superadmin || !!$devopsRole,
isAdminsWorkspace: $workspaceStore === 'admins'
})}
bind:value={filters.val}
placeholder="Filter runs..."
autofocus
/>
<!-- The filters are shallow-routed, so the search has to come off
`window.location` at click time — `page.url` never sees them. Always the
canonical `/runs`: only that is a recognized preview page, and the
`/runs/<path>` route mirrors its path into `?path=` anyway. -->
<!-- The button stretches to its parent's height, which is the whole header once
this row wraps: pin it to one row's height. -->
<div class={twMerge('flex empty:hidden', ButtonType.UnifiedHeightClasses.md)}>
<OpenInSessionButton
source={{
page: () => pageHref(RUNS_PATH) + window.location.search,
workspaceId: $workspaceStore ?? undefined
}}
btnProps={{ unifiedSize: 'md' }}
/>
</div>
</div>
</div>
<!-- Graph -->
@@ -939,7 +949,21 @@
<Splitpanes>
<Pane minSize={40}>
<div class="h-full flex">
<div class="flex flex-col flex-1 m-4 mt-2 mr-2">
<div class="flex flex-col flex-1 min-w-0 m-4 mt-2 mr-2">
{#if scanProgress}
<div class="px-1 pb-2 flex items-center gap-2">
<Loader2 size={14} class="animate-spin shrink-0 text-accent" />
<div class="flex-1 min-w-0">
<BatchLoadProgress
loaded={Math.round(scanProgress.fraction * 100)}
total={100}
itemsLabel="runs"
label={`Slow search: runs since ${displayDate(scanProgress.scannedTo)} searched, ${jobs?.length ?? 0} found`}
onStop={() => jobsLoader.stopBatchLoading()}
/>
</div>
</div>
{/if}
{#if batchProgress}
<div class="px-1 pb-2">
<BatchLoadProgress
@@ -0,0 +1,140 @@
import { describe, it, expect } from 'vitest'
import { CancelablePromise, type Job } from '$lib/gen'
import { scanJobWindows, type JobWindow } from './jobsWindowScan'
const HOUR = 60 * 60 * 1000
const NOW = Date.parse('2026-01-10T00:00:00.000Z')
function job(at: number): Job {
return { type: 'CompletedJob', id: `j${at}`, created_at: new Date(at).toISOString() } as Job
}
// A history the scan can only read through its windows: `at` are the creation times of the
// matching jobs, and a window wider than `slowOver` never answers.
function history(at: number[], slowOver = Infinity) {
const windows: JobWindow[] = []
const fetchWindow = (w: JobWindow) => {
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<Job[]>(() => {})
const rows = at
.filter((t) => t >= lo && t <= hi)
.sort((a, b) => b - a)
.slice(0, w.limit)
.map(job)
return new CancelablePromise<Job[]>((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<Job[]>((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)
})
})
@@ -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<string | undefined>
fetchWindow: (w: JobWindow) => CancelablePromise<Job[]>
/** `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<boolean> {
const targetMs = scan.targetMs ?? TARGET_MS
const patienceMs = scan.patienceMs ?? PATIENCE_MS
return new CancelablePromise<boolean>((resolve, reject, onCancel) => {
let current: CancelablePromise<Job[]> | undefined
onCancel(() => current?.cancel())
async function fetchWithPatience(w: JobWindow): Promise<Job[] | typeof SLOW> {
const request = scan.fetchWindow(w)
current = request
let timer: ReturnType<typeof setTimeout> | undefined
const slow = new Promise<typeof SLOW>((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<boolean> {
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)
})
}
@@ -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<boolean> | 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<boolean> | 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<Job[]> {
// 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<boolean> {
if (!canScanWindows || _args.skip) {
return CancelablePromiseUtils.map(fetchJobs(page.query), (res) => {
onJobs(res)
return res.length < page.pageSize
})
}
const fetched = new CancelablePromise<boolean>((resolve, reject, onCancel) => {
const single = listJobs(page.query)
let scan: CancelablePromise<boolean> | 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<boolean> {
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<Job[]> {
if (_args.skip) return CancelablePromiseUtils.pure<Job[]>([])
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<Job[]>([])
})
}
function listJobs({
completedBefore,
completedAfter,
createdBefore,
createdAfter,
createdAfterQueue,
perPage: perPageOverride
}: JobsQuery): CancelablePromise<Job[]> {
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<Job[]>([])
})
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<void> {
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
},