nit fix

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<QueuedJob>>(query)
             .bind(job_id)
             .bind(workspace_id)
@@ -5674,6 +5674,7 @@ pub struct JobUpdateQuery {
     pub log_offset: Option<i32>,
     pub get_progress: Option<bool>,
     pub only_result: Option<bool>,
+    pub fast: Option<bool>,
 }
 
 #[derive(Serialize)]
@@ -5689,6 +5690,28 @@ pub struct JobUpdate {
     pub only_result: Option<Box<serde_json::value::RawValue>>,
 }
 
+#[derive(PartialEq)]
+pub struct JobUpdateLastStatus {
+    pub running: Option<bool>,
+    pub completed: Option<bool>,
+    pub log_offset: Option<i32>,
+    pub mem_peak: Option<i32>,
+}
+
+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<Response> {
     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<DB>,
     Path((w_id, job_id)): Path<(String, Uuid)>,
-    Query(JobUpdateQuery { log_offset, get_progress, running, only_result }): Query<JobUpdateQuery>,
+    Query(JobUpdateQuery { log_offset, get_progress, running, only_result, .. }): Query<JobUpdateQuery>,
 ) -> JsonResult<JobUpdate> {
     Ok(Json(
         get_job_update_data(
@@ -5770,8 +5793,9 @@ async fn get_job_update_sse(
     opt_tokened: OptTokened,
     Extension(db): Extension<DB>,
     Path((w_id, job_id)): Path<(String, Uuid)>,
-    Query(JobUpdateQuery { log_offset, get_progress, running, only_result }): Query<JobUpdateQuery>,
+    Query(JobUpdateQuery { log_offset, get_progress, running, only_result, fast }): Query<JobUpdateQuery>,
 ) -> 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<ApiAuthed>,
     opt_tokened: OptTokened,
@@ -5805,17 +5843,17 @@ fn get_job_update_sse_stream(
     get_progress: Option<bool>,
     running: Option<bool>,
     only_result: Option<bool>,
-) -> impl futures::Stream<Item = String> {
+    fast: Option<bool>,
+) -> impl futures::Stream<Item = JobUpdateSSEStream> {
     let (tx, rx) = tokio::sync::mpsc::channel(32);
 
     tokio::spawn(async move {
         let mut log_offset = initial_log_offset;
-        let mut last_update: Option<String> = None;
-        let mut completion_sent = false;
+        let mut last_update: Option<JobUpdateLastStatus> = 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;
                 }
             }
         }

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)

frontend/src/lib/components/ResultJobLoader.svelte

@@ -1,288 +0,0 @@
-<script lang="ts">
-	import { JobService, type Preview } from '$lib/gen'
-	import { workspaceStore } from '$lib/stores'
-	import { onDestroy, tick, untrack } from 'svelte'
-	import { createEventDispatcher } from 'svelte'
-	import type { SupportedLanguage } from '$lib/common'
-
-	interface Props {
-		isLoading?: boolean
-		job?: { completed: boolean; result: any; id: string; success?: boolean } | undefined
-		workspaceOverride?: string | undefined
-		notfound?: boolean
-		isEditor?: boolean
-		allowConcurentRequests?: boolean
-	}
-
-	let {
-		isLoading = $bindable(false),
-		job = $bindable(undefined),
-		workspaceOverride = undefined,
-		notfound = $bindable(false),
-		isEditor = false,
-		allowConcurentRequests = false
-	}: Props = $props()
-
-	const dispatch = createEventDispatcher()
-
-	let workspace = $derived(workspaceOverride ?? $workspaceStore!)
-
-	let syncIteration: number = 0
-	let errorIteration = 0
-
-	let ITERATIONS_BEFORE_SLOW_REFRESH = 10
-	let ITERATIONS_BEFORE_SUPER_SLOW_REFRESH = 100
-
-	let lastStartedAt: number = Date.now()
-	let currentId: string | undefined = $state(undefined)
-
-	$effect(() => {
-		let newIsLoading = currentId !== undefined
-		untrack(() => {
-			if (isLoading !== newIsLoading) {
-				isLoading = newIsLoading
-			}
-		})
-	})
-
-	type Callbacks = { done: (x: any) => void; cancel: () => void; error: (err: Error) => void }
-
-	let running = false
-	let lastCallbacks: Callbacks | undefined = undefined
-
-	let finished: string[] = []
-	export async function abstractRun(fn: () => Promise<string>, callbacks?: Callbacks) {
-		try {
-			running = false
-			isLoading = true
-			clearCurrentJob()
-			const startedAt = Date.now()
-			const testId = await fn()
-			lastCallbacks = callbacks
-			if (lastStartedAt < startedAt || allowConcurentRequests) {
-				lastStartedAt = startedAt
-				if (testId) {
-					dispatch('started', testId)
-					try {
-						await watchJob(testId, callbacks)
-					} catch (e) {
-						callbacks?.cancel()
-						dispatch('cancel', testId)
-						if (currentId === testId) {
-							currentId = undefined
-						}
-					}
-				}
-			}
-			return testId
-		} catch (err) {
-			callbacks?.error(err)
-			// if error happens on submitting the job, reset UI state so the user can try again
-			isLoading = false
-			currentId = undefined
-			job = undefined
-			throw err
-		}
-	}
-
-	export async function runScriptByPath(
-		path: string,
-		args: Record<string, any>,
-		callbacks?: Callbacks
-	): Promise<string> {
-		return abstractRun(
-			() =>
-				JobService.runScriptByPath({
-					workspace: workspace,
-					path: path,
-					requestBody: args,
-					skipPreprocessor: true
-				}),
-			callbacks
-		)
-	}
-
-	export async function runScriptByHash(
-		hash: string,
-		args: Record<string, any>,
-		callbacks?: Callbacks
-	): Promise<string> {
-		return abstractRun(
-			() =>
-				JobService.runScriptByHash({
-					workspace: workspace,
-					hash: hash,
-					requestBody: args,
-					skipPreprocessor: true
-				}),
-			callbacks
-		)
-	}
-
-	export async function runFlowByPath(
-		path: string | undefined,
-		args: Record<string, any>,
-		callbacks?: Callbacks
-	): Promise<string> {
-		return abstractRun(
-			() =>
-				JobService.runFlowByPath({
-					workspace: workspace,
-					path: path ?? '',
-					requestBody: args,
-					skipPreprocessor: true
-				}),
-			callbacks
-		)
-	}
-
-	export async function runPreview(
-		path: string | undefined,
-		code: string,
-		lang: SupportedLanguage,
-		args: Record<string, any>,
-		tag: string | undefined,
-		callbacks?: Callbacks
-	): Promise<string> {
-		return abstractRun(
-			() =>
-				JobService.runScriptPreview({
-					workspace: workspace,
-					requestBody: {
-						path,
-						content: code,
-						args,
-						language: lang as Preview['language'],
-						tag
-					}
-				}),
-			callbacks
-		)
-	}
-
-	export async function cancelJob() {
-		const id = currentId
-		if (id) {
-			lastCallbacks?.cancel()
-			lastCallbacks = undefined
-
-			dispatch('cancel', id)
-
-			currentId = undefined
-			try {
-				await JobService.cancelQueuedJob({
-					workspace: workspace ?? '',
-					id,
-					requestBody: {}
-				})
-			} catch (err) {
-				console.error(err)
-			}
-		}
-	}
-
-	export async function clearCurrentJob() {
-		if (currentId && !allowConcurentRequests) {
-			lastCallbacks?.cancel()
-			dispatch('cancel', currentId)
-			lastCallbacks = undefined
-			job = undefined
-			await cancelJob()
-		}
-	}
-
-	export async function watchJob(testId: string, callbacks?: Callbacks) {
-		syncIteration = 0
-		errorIteration = 0
-		currentId = testId
-		job = undefined
-
-		const isCompleted = await loadTestJob(testId, callbacks)
-		if (!isCompleted) {
-			setTimeout(() => {
-				syncer(testId, callbacks)
-			}, 50)
-		}
-	}
-
-	async function loadTestJob(id: string, callbacks?: Callbacks): Promise<boolean> {
-		let isCompleted = false
-		if (currentId === id || allowConcurentRequests) {
-			try {
-				let maybe_job = await JobService.getCompletedJobResultMaybe({
-					workspace: workspace ?? '',
-					id,
-					getStarted: isEditor
-				})
-				if (maybe_job.started && !running) {
-					running = true
-					dispatch('running', id)
-				}
-				if (maybe_job.completed) {
-					isCompleted = true
-					if (currentId === id || allowConcurentRequests) {
-						job = { ...maybe_job, id }
-						await tick()
-						if (!job?.success && typeof job?.result == 'object' && 'error' in (job?.result ?? {})) {
-							callbacks?.error(job.result.error)
-							dispatch('doneError', {
-								id,
-								error: job.result.error
-							})
-						} else {
-							callbacks?.done(job.result)
-							dispatch('done', job)
-						}
-						finished.push(id)
-						if (!allowConcurentRequests) {
-							currentId = undefined
-						}
-					} else {
-						callbacks?.cancel()
-						dispatch('cancel', id)
-					}
-				}
-				notfound = false
-			} catch (err) {
-				errorIteration += 1
-				if (errorIteration == 5) {
-					notfound = true
-					await clearCurrentJob()
-					dispatch('doneError', err)
-				}
-				console.warn(err)
-			}
-			return isCompleted
-		} else {
-			callbacks?.cancel()
-			dispatch('cancel', id)
-			return true
-		}
-	}
-
-	async function syncer(id: string, callbacks?: Callbacks): Promise<void> {
-		if ((currentId != id && !allowConcurentRequests) || finished.includes(id)) {
-			callbacks?.cancel()
-			dispatch('cancel', id)
-			return
-		}
-		syncIteration++
-		let r = await loadTestJob(id, callbacks)
-		if (r) {
-			return
-		}
-		let nextIteration = 50
-		if (syncIteration > ITERATIONS_BEFORE_SLOW_REFRESH) {
-			nextIteration = 500
-		} else if (syncIteration > ITERATIONS_BEFORE_SUPER_SLOW_REFRESH) {
-			nextIteration = 2000
-		}
-		setTimeout(() => {
-			syncer(id, callbacks)
-		}, nextIteration)
-	}
-
-	onDestroy(async () => {
-		currentId = undefined
-	})
-</script>

frontend/src/lib/components/ScriptEditor.svelte

@@ -212,6 +212,10 @@
 			{
 				done(_x) {
 					loadPastTests()
+				},
+				doneError({ error }) {
+					console.error(error)
+					sendUserToast('Error running test', true)
 				}
 			}
 		)