diff --git a/backend/windmill-api/src/db.rs b/backend/windmill-api/src/db.rs index 57ec59eb73..8cacc5a6c1 100644 --- a/backend/windmill-api/src/db.rs +++ b/backend/windmill-api/src/db.rs @@ -249,14 +249,55 @@ async fn fix_flow_versioning_migration( } async fn fix_job_completed_index(db: &DB) -> Result<(), Error> { + // let has_done_migration = sqlx::query_scalar!( + // "SELECT EXISTS(SELECT name FROM windmill_migrations WHERE name = 'fix_job_completed_index')" + // ) + // .fetch_one(db) + // .await? + // .unwrap_or(false); + // if !has_done_migration { + // tracing::info!("Applying fix_job_completed_index migration"); + // let mut tx = db.begin().await?; + // let mut r = false; + // while !r { + // r = sqlx::query_scalar!("SELECT pg_try_advisory_lock(4242)") + // .fetch_one(&mut *tx) + // .await + // .map_err(|e| { + // tracing::error!("Error acquiring fix_job_completed_index lock: {e:#}"); + // sqlx::migrate::MigrateError::Execute(e) + // })? + // .unwrap_or(false); + // if !r { + // tracing::info!("PG fix_job_completed_index_migration lock already acquired by another server or worker, retrying in 5s. (look for the advisory lock in pg_lock with granted = true)"); + // tokio::time::sleep(std::time::Duration::from_secs(5)).await; + // } + // } + // // sqlx::query( + // // "CREATE INDEX CONCURRENTLY IF NOT EXISTS ix_completed_job_workspace_id_created_at_new ON completed_job (workspace_id, job_kind, is_skipped, is_flow_step, created_at DESC, started_at DESC)" + // // ).execute(db).await?; + + // sqlx::query("DROP INDEX CONCURRENTLY IF EXISTS ix_completed_job_workspace_id_created_at") + // .execute(db) + // .await?; + + // sqlx::query!("INSERT INTO windmill_migrations (name) VALUES ('fix_job_completed_index') ON CONFLICT DO NOTHING") + // .execute(&mut *tx) + // .await?; + // let _ = sqlx::query("SELECT pg_advisory_unlock(4242)") + // .execute(&mut *tx) + // .await?; + // tx.commit().await?; + // } + let has_done_migration = sqlx::query_scalar!( - "SELECT EXISTS(SELECT name FROM windmill_migrations WHERE name = 'fix_job_completed_index')" + "SELECT EXISTS(SELECT name FROM windmill_migrations WHERE name = 'fix_job_completed_index_2')" ) .fetch_one(db) .await? .unwrap_or(false); if !has_done_migration { - tracing::info!("Applying fix_job_completed_index migration"); + tracing::info!("Applying fix_job_completed_index_2 migration"); let mut tx = db.begin().await?; let mut r = false; while !r { @@ -264,30 +305,41 @@ async fn fix_job_completed_index(db: &DB) -> Result<(), Error> { .fetch_one(&mut *tx) .await .map_err(|e| { - tracing::error!("Error acquiring fix_job_completed_index lock: {e:#}"); + tracing::error!("Error acquiring fix_job_completed_index_2 lock: {e:#}"); sqlx::migrate::MigrateError::Execute(e) })? .unwrap_or(false); if !r { - tracing::info!("PG fix_job_completed_index_migration lock already acquired by another server or worker, retrying in 5s. (look for the advisory lock in pg_lock with granted = true)"); + tracing::info!("PG fix_job_completed_index_migration_2 lock already acquired by another server or worker, retrying in 5s. (look for the advisory lock in pg_lock with granted = true)"); tokio::time::sleep(std::time::Duration::from_secs(5)).await; } } sqlx::query( - "CREATE INDEX CONCURRENTLY IF NOT EXISTS ix_completed_job_workspace_id_created_at_new ON completed_job (workspace_id, job_kind, is_skipped, is_flow_step, created_at DESC, started_at DESC)" + "CREATE INDEX CONCURRENTLY IF NOT EXISTS ix_completed_job_workspace_id_created_at_new_2 ON completed_job (workspace_id, job_kind, success, is_skipped, is_flow_step, created_at DESC)" + ).execute(db).await?; + + sqlx::query( + "CREATE INDEX CONCURRENTLY IF NOT EXISTS ix_completed_job_workspace_id_started_at_new ON completed_job (workspace_id, job_kind, success, is_skipped, is_flow_step, started_at DESC)" ).execute(db).await?; sqlx::query("DROP INDEX CONCURRENTLY IF EXISTS ix_completed_job_workspace_id_created_at") .execute(db) .await?; - sqlx::query!("INSERT INTO windmill_migrations (name) VALUES ('fix_job_completed_index') ON CONFLICT DO NOTHING") + sqlx::query( + "DROP INDEX CONCURRENTLY IF EXISTS ix_completed_job_workspace_id_created_at_new", + ) + .execute(db) + .await?; + + sqlx::query!("INSERT INTO windmill_migrations (name) VALUES ('fix_job_completed_index_2') ON CONFLICT DO NOTHING") .execute(&mut *tx) .await?; let _ = sqlx::query("SELECT pg_advisory_unlock(4242)") .execute(&mut *tx) .await?; tx.commit().await?; + tracing::info!("Finished applying fix_job_completed_index_2 migration"); } Ok(()) diff --git a/frontend/src/lib/components/runs/JobLoader.svelte b/frontend/src/lib/components/runs/JobLoader.svelte index 80123e7da6..6cee6e0667 100644 --- a/frontend/src/lib/components/runs/JobLoader.svelte +++ b/frontend/src/lib/components/runs/JobLoader.svelte @@ -108,37 +108,49 @@ } } + let loadingFetch = false async function fetchJobs( startedBefore: string | undefined, startedAfter: string | undefined, startedAfterCompletedJobs: string | undefined ): Promise { - return JobService.listJobs({ - workspace: $workspaceStore!, - createdOrStartedBefore: startedBefore, - createdOrStartedAfter: startedAfter, - createdOrStartedAfterCompletedJobs: startedAfterCompletedJobs, - schedulePath, - scriptPathExact: path === null || path === '' ? undefined : path, - createdBy: user === null || user === '' ? undefined : user, - scriptPathStart: folder === null || folder === '' ? undefined : `f/${folder}/`, - jobKinds, - success: success == 'success' ? true : success == 'failure' ? false : undefined, - running: success == 'running' ? true : undefined, - isSkipped: isSkipped ? undefined : false, - isFlowStep: jobKindsCat != 'all' ? false : undefined, - label: label === null || label === '' ? undefined : label, - isNotSchedule: showSchedules == false ? true : undefined, - scheduledForBeforeNow: showFutureJobs == false ? true : undefined, - args: - argFilter && argFilter != '{}' && argFilter != '' && argError == '' ? argFilter : undefined, - result: - resultFilter && resultFilter != '{}' && resultFilter != '' && resultError == '' - ? resultFilter - : undefined, - allWorkspaces: allWorkspaces ? true : undefined, - perPage - }) + loadingFetch = true + try { + return JobService.listJobs({ + workspace: $workspaceStore!, + createdOrStartedBefore: startedBefore, + createdOrStartedAfter: startedAfter, + createdOrStartedAfterCompletedJobs: startedAfterCompletedJobs, + schedulePath, + scriptPathExact: path === null || path === '' ? undefined : path, + createdBy: user === null || user === '' ? undefined : user, + scriptPathStart: folder === null || folder === '' ? undefined : `f/${folder}/`, + jobKinds, + success: success == 'success' ? true : success == 'failure' ? false : undefined, + running: success == 'running' ? true : undefined, + isSkipped: isSkipped ? undefined : false, + isFlowStep: jobKindsCat != 'all' ? false : undefined, + label: label === null || label === '' ? undefined : label, + isNotSchedule: showSchedules == false ? true : undefined, + scheduledForBeforeNow: showFutureJobs == false ? true : undefined, + args: + argFilter && argFilter != '{}' && argFilter != '' && argError == '' + ? argFilter + : undefined, + result: + resultFilter && resultFilter != '{}' && resultFilter != '' && resultError == '' + ? resultFilter + : undefined, + allWorkspaces: allWorkspaces ? true : undefined, + perPage + }) + } catch (e) { + sendUserToast('There was an issue loading jobs, see browser console for more details', true) + console.error(e) + return [] + } finally { + loadingFetch = false + } } async function fetchExtendedJobs( @@ -147,34 +159,48 @@ startedAfter: string | undefined, startedAfterCompletedJobs: string | undefined ): Promise { - return ConcurrencyGroupsService.listExtendedJobs({ - rowLimit: 1000, - concurrencyKey: concurrencyKey == null || concurrencyKey == '' ? undefined : concurrencyKey, - workspace: $workspaceStore!, - createdOrStartedBefore: startedBefore, - createdOrStartedAfter: startedAfter, - createdOrStartedAfterCompletedJobs: startedAfterCompletedJobs, - schedulePath, - scriptPathExact: path === null || path === '' ? undefined : path, - createdBy: user === null || user === '' ? undefined : user, - scriptPathStart: folder === null || folder === '' ? undefined : `f/${folder}/`, - jobKinds, - success: success == 'success' ? true : success == 'failure' ? false : undefined, - running: success == 'running' ? true : undefined, - isSkipped: isSkipped ? undefined : false, - isFlowStep: jobKindsCat != 'all' ? false : undefined, - label: label === null || label === '' ? undefined : label, - isNotSchedule: showSchedules == false ? true : undefined, - scheduledForBeforeNow: showFutureJobs == false ? true : undefined, - args: - argFilter && argFilter != '{}' && argFilter != '' && argError == '' ? argFilter : undefined, - result: - resultFilter && resultFilter != '{}' && resultFilter != '' && resultError == '' - ? resultFilter - : undefined, - allWorkspaces: allWorkspaces ? true : undefined, - perPage - }) + loadingFetch = true + try { + return ConcurrencyGroupsService.listExtendedJobs({ + rowLimit: 1000, + concurrencyKey: concurrencyKey == null || concurrencyKey == '' ? undefined : concurrencyKey, + workspace: $workspaceStore!, + createdOrStartedBefore: startedBefore, + createdOrStartedAfter: startedAfter, + createdOrStartedAfterCompletedJobs: startedAfterCompletedJobs, + schedulePath, + scriptPathExact: path === null || path === '' ? undefined : path, + createdBy: user === null || user === '' ? undefined : user, + scriptPathStart: folder === null || folder === '' ? undefined : `f/${folder}/`, + jobKinds, + success: success == 'success' ? true : success == 'failure' ? false : undefined, + running: success == 'running' ? true : undefined, + isSkipped: isSkipped ? undefined : false, + isFlowStep: jobKindsCat != 'all' ? false : undefined, + label: label === null || label === '' ? undefined : label, + isNotSchedule: showSchedules == false ? true : undefined, + scheduledForBeforeNow: showFutureJobs == false ? true : undefined, + args: + argFilter && argFilter != '{}' && argFilter != '' && argError == '' + ? argFilter + : undefined, + result: + resultFilter && resultFilter != '{}' && resultFilter != '' && resultError == '' + ? resultFilter + : undefined, + allWorkspaces: allWorkspaces ? true : undefined, + perPage + }) + } catch (e) { + sendUserToast('There was an issue loading jobs, see browser console for more details', true) + console.error(e) + return { + jobs: [], + obscured_jobs: [] + } + } finally { + loadingFetch = false + } } export async function loadJobs( @@ -256,6 +282,9 @@ } async function syncer() { + if (loadingFetch) { + return + } if (sync) { if (syncQueuedRunsCount) { getCount()