From 8d9bb67d23403fb27d18cb7c70f2b6bf88db6f36 Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Fri, 18 Jul 2025 19:57:58 +0000 Subject: [PATCH] nit fix --- backend/windmill-api/src/jobs.rs | 154 +++++++--- frontend/src/lib/components/JobLoader.svelte | 50 ++- .../src/lib/components/ResultJobLoader.svelte | 288 ------------------ .../src/lib/components/ScriptEditor.svelte | 4 + 4 files changed, 157 insertions(+), 339 deletions(-) delete mode 100644 frontend/src/lib/components/ResultJobLoader.svelte diff --git a/backend/windmill-api/src/jobs.rs b/backend/windmill-api/src/jobs.rs index aa74cc5591..90915922a9 100644 --- a/backend/windmill-api/src/jobs.rs +++ b/backend/windmill-api/src/jobs.rs @@ -20,6 +20,7 @@ use sqlx::Pool; use std::collections::HashMap; use std::ops::{Deref, DerefMut}; use std::str::FromStr; +use std::time::Instant; use tokio::io::AsyncReadExt; use tower::ServiceBuilder; use windmill_common::auth::TOKEN_PREFIX_LEN; @@ -947,7 +948,6 @@ impl<'a> GetQuery<'a> { with_code: self.with_code, with_flow: self.with_flow, ); - tracing::error!("query: {}", query); let query = sqlx::query_as::<_, JobExtended>(query) .bind(job_id) .bind(workspace_id) @@ -5674,6 +5674,7 @@ pub struct JobUpdateQuery { pub log_offset: Option, pub get_progress: Option, pub only_result: Option, + pub fast: Option, } #[derive(Serialize)] @@ -5689,6 +5690,28 @@ pub struct JobUpdate { pub only_result: Option>, } +#[derive(PartialEq)] +pub struct JobUpdateLastStatus { + pub running: Option, + pub completed: Option, + pub log_offset: Option, + pub mem_peak: Option, +} + +impl From<&JobUpdate> for JobUpdateLastStatus { + fn from(update: &JobUpdate) -> Self { + Self { + running: update.running, + completed: update.completed, + log_offset: update.log_offset, + mem_peak: update.mem_peak, + } + } +} + + + + async fn get_log_file(Path((_w_id, file_p)): Path<(String, String)>) -> error::Result { let local_file = format!("{TMP_DIR}/logs/{file_p}"); if tokio::fs::metadata(&local_file).await.is_ok() { @@ -5745,7 +5768,7 @@ async fn get_job_update( opt_tokened: OptTokened, Extension(db): Extension, Path((w_id, job_id)): Path<(String, Uuid)>, - Query(JobUpdateQuery { log_offset, get_progress, running, only_result }): Query, + Query(JobUpdateQuery { log_offset, get_progress, running, only_result, .. }): Query, ) -> JsonResult { Ok(Json( get_job_update_data( @@ -5770,8 +5793,9 @@ async fn get_job_update_sse( opt_tokened: OptTokened, Extension(db): Extension, Path((w_id, job_id)): Path<(String, Uuid)>, - Query(JobUpdateQuery { log_offset, get_progress, running, only_result }): Query, + Query(JobUpdateQuery { log_offset, get_progress, running, only_result, fast }): Query, ) -> Response { + let stream = get_job_update_sse_stream( opt_authed, opt_tokened, @@ -5782,7 +5806,11 @@ async fn get_job_update_sse( get_progress, running, only_result, - ); + fast, + ) + .map(|x| { + format!("data: {}\n\n", serde_json::to_string(&x).unwrap_or_default()) + }); let body = axum::body::Body::from_stream(stream.map(Result::<_, std::convert::Infallible>::Ok)); @@ -5795,6 +5823,16 @@ async fn get_job_update_sse( .unwrap() } +#[derive(Serialize)] +#[serde(tag = "type", rename_all = "lowercase")] +enum JobUpdateSSEStream { + Update(JobUpdate), + Error(String), + NotFound, + Timeout, + Ping, +} + fn get_job_update_sse_stream( opt_authed: Option, opt_tokened: OptTokened, @@ -5805,17 +5843,17 @@ fn get_job_update_sse_stream( get_progress: Option, running: Option, only_result: Option, -) -> impl futures::Stream { + fast: Option, +) -> impl futures::Stream { let (tx, rx) = tokio::sync::mpsc::channel(32); tokio::spawn(async move { let mut log_offset = initial_log_offset; - let mut last_update: Option = None; - let mut completion_sent = false; + let mut last_update: Option = None; // Send initial update immediately let mut running = running; - if let Ok(update) = get_job_update_data( + match get_job_update_data( &opt_authed, &opt_tokened, &db, @@ -5830,42 +5868,59 @@ fn get_job_update_sse_stream( ) .await { - if let Ok(serialized) = serde_json::to_string(&update) { - let event_data = format!("data: {}\n\n", serialized); - if tx.send(event_data.clone()).await.is_err() { - tracing::warn!("Failed to send initial job update for job {job_id}"); - return; - } - last_update = Some(serialized); - if let Some(new_offset) = update.log_offset { - log_offset = Some(new_offset); - } - completion_sent = update.completed.unwrap_or(false); - if running.is_some() { - running = Some(update.running.unwrap_or(false)); - } - } else { - tracing::warn!("Failed to serialize job update for job {job_id}"); + Ok(update) => { + last_update = Some((&update).into()); + let completion_sent = update.completed.unwrap_or(false); + if running.is_some() { + running = Some(update.running.unwrap_or(false)); + } + if let Some(new_offset) = update.log_offset { + log_offset = Some(new_offset); + } + if tx.send(JobUpdateSSEStream::Update(update)).await.is_err() { + tracing::warn!("Failed to send initial job update for job {job_id}"); + return; + } + if completion_sent { + return + } + } + Err(e) => { + if tx.send(JobUpdateSSEStream::Error(e.to_string())).await.is_err() { + tracing::warn!("Failed to send initial job update for job {job_id}"); return; } } + } - // If job is already completed, no need to poll - if completion_sent { - return; - } // Poll for updates every 1 second let mut i = 0; + let start = Instant::now(); + let mut last_ping = Instant::now(); loop { i += 1; - let ms_duration = if i > 10 { - 500 - } else if i > 100 { + let ms_duration = if i > 100 || !fast.unwrap_or(false) { 3000 + } else if i > 10 { + 500 } else { 100 }; + if last_ping.elapsed().as_secs() > 5 { + if tx.send(JobUpdateSSEStream::Ping).await.is_err() { + tracing::warn!("Failed to send job ping for job {job_id}"); + return; + } + last_ping = Instant::now(); + } + + if start.elapsed().as_secs() > 30 { + if tx.send(JobUpdateSSEStream::Timeout).await.is_err() { + tracing::warn!("Failed to send job timeout for job {job_id}"); + } + return; + } tokio::time::sleep(std::time::Duration::from_millis(ms_duration)).await; match get_job_update_data( @@ -5887,28 +5942,31 @@ fn get_job_update_sse_stream( if running.is_some() { running = Some(update.running.unwrap_or(false)); } - if let Ok(serialized) = serde_json::to_string(&update) { - // Only send if the update has changed - if last_update.as_ref() != Some(&serialized) { - let event_data = format!("data: {}\n\n", serialized); - if tx.send(event_data).await.is_err() { - break; - } - if update.completed.unwrap_or(false) { - break; - } - last_update = Some(serialized); + let update_last_status = (&update).into(); + // Only send if the update has changed + if last_update.as_ref() != Some(&update_last_status) { - // Update log offset if available - if let Some(new_offset) = update.log_offset { - log_offset = Some(new_offset); - } + // Update log offset if available + if let Some(new_offset) = update.log_offset { + log_offset = Some(new_offset); } + let completed = update.completed.unwrap_or(false); + if tx.send(JobUpdateSSEStream::Update(update)).await.is_err() { + break; + } + if completed { + break; + } + + last_update = Some(update_last_status); + } } Err(_) => { - // Job might have been deleted or access denied, break the loop - break; + if tx.send(JobUpdateSSEStream::NotFound).await.is_err() { + tracing::warn!("Failed to send job not found for job {job_id}"); + } + return; } } } diff --git a/frontend/src/lib/components/JobLoader.svelte b/frontend/src/lib/components/JobLoader.svelte index dc823b676e..e933d6ea87 100644 --- a/frontend/src/lib/components/JobLoader.svelte +++ b/frontend/src/lib/components/JobLoader.svelte @@ -78,6 +78,7 @@ let lastStartedAt: number = Date.now() let currentId: string | undefined = $state(undefined) + let noPingTimeout: NodeJS.Timeout | undefined = undefined $effect(() => { let newIsLoading = currentId !== undefined @@ -93,6 +94,7 @@ isLoading = true clearCurrentJob() lastCallbacks = callbacks + noPingTimeout = undefined const startedAt = Date.now() const testId = await fn() @@ -377,7 +379,9 @@ if (errorIteration == 5) { notfound = true job = undefined + currentId = undefined } + callbacks?.doneError?.({ error: err, id }) console.warn(err) } return isCompleted @@ -416,6 +420,21 @@ } } } + + function setNoPingTimeout(id: string, attempt: number, callbacks?: Callbacks) { + if (noPingTimeout) { + clearTimeout(noPingTimeout) + } + if (id === currentId || allowConcurentRequests) { + noPingTimeout = setTimeout(() => { + if (currentId === id || allowConcurentRequests) { + currentEventSource?.close() + currentEventSource = undefined + loadTestJobWithSSE(id, attempt + 1, callbacks) + } + }, 10000) + } + } async function loadTestJobWithSSE( id: string, attempt: number, @@ -467,10 +486,15 @@ params.set('only_result', 'true') } + if (lastStartedAt > Date.now() - 5000) { + params.set('fast', 'true') + } + const sseUrl = `/api/w/${workspace}/jobs_u/getupdate_sse/${id}?${params.toString()}` currentEventSource = new EventSource(sseUrl) + setNoPingTimeout(id, attempt, callbacks) currentEventSource.onmessage = async (event) => { if (currentId !== id) { currentEventSource?.close() @@ -480,6 +504,26 @@ try { const previewJobUpdates = JSON.parse(event.data) + let type = previewJobUpdates.type + if (type == 'timeout') { + currentEventSource?.close() + currentEventSource = undefined + loadTestJobWithSSE(id, 0, callbacks) + return + } else if (type == 'ping') { + setNoPingTimeout(id, attempt, callbacks) + return + } else if (type == 'error') { + currentEventSource?.close() + currentEventSource = undefined + console.error('SSE error:', previewJobUpdates) + throw new Error('SSE error: ' + previewJobUpdates) + } else if (type == 'not_found') { + currentEventSource?.close() + currentEventSource = undefined + console.error('Not found') + throw new Error('Not found') + } jobUpdateLastFetch = new Date() if (job) { @@ -492,6 +536,7 @@ if (previewJobUpdates.completed) { currentEventSource?.close() currentEventSource = undefined + noPingTimeout = undefined if (onlyResult) { callbacks?.doneResult?.({ id, @@ -516,9 +561,8 @@ currentEventSource?.close() currentEventSource = undefined if (attempt < 3) { - console.log(`SSE error (1), retrying ... attempt: ${attempt}/3`) - attempt++ - setTimeout(() => loadTestJobWithSSE(id, attempt, callbacks), 1000) + console.log(`SSE error (1), retrying ... attempt: ${attempt + 1}/3`) + setTimeout(() => loadTestJobWithSSE(id, attempt + 1, callbacks), 1000) } else { // Fall back to polling on error setTimeout(() => syncer(id), 1000) diff --git a/frontend/src/lib/components/ResultJobLoader.svelte b/frontend/src/lib/components/ResultJobLoader.svelte deleted file mode 100644 index f81e0aec25..0000000000 --- a/frontend/src/lib/components/ResultJobLoader.svelte +++ /dev/null @@ -1,288 +0,0 @@ - diff --git a/frontend/src/lib/components/ScriptEditor.svelte b/frontend/src/lib/components/ScriptEditor.svelte index 679d700b80..ffc367a0aa 100644 --- a/frontend/src/lib/components/ScriptEditor.svelte +++ b/frontend/src/lib/components/ScriptEditor.svelte @@ -212,6 +212,10 @@ { done(_x) { loadPastTests() + }, + doneError({ error }) { + console.error(error) + sendUserToast('Error running test', true) } } )