diff --git a/backend/windmill-api/openapi.yaml b/backend/windmill-api/openapi.yaml index 9dca91e94f..5064f25a0e 100644 --- a/backend/windmill-api/openapi.yaml +++ b/backend/windmill-api/openapi.yaml @@ -3099,7 +3099,7 @@ paths: type: array items: $ref: "#/components/schemas/ContextualVariable" - + /w/{workspace}/workspaces/get_secondary_storage_names: get: summary: get secondary storage names @@ -7488,6 +7488,10 @@ paths: in: query schema: type: boolean + - name: no_code + in: query + schema: + type: boolean responses: "200": description: job details diff --git a/backend/windmill-api/src/jobs.rs b/backend/windmill-api/src/jobs.rs index f3ba9e5ae9..7ce3fc27be 100644 --- a/backend/windmill-api/src/jobs.rs +++ b/backend/windmill-api/src/jobs.rs @@ -527,7 +527,7 @@ async fn get_flow_job_debug_info( Path((w_id, id)): Path<(String, Uuid)>, ) -> error::Result { let job = GetQuery::new() - .fetch_queued((&db).into(), id, &w_id) + .fetch_queued((&db).into(), &id, &w_id) .await?; if let Some(job) = job { let is_flow = job.is_flow(); @@ -567,7 +567,7 @@ async fn get_flow_job_debug_info( for job_id in job_ids { let job = GetQuery::new() .with_auth(&opt_authed) - .fetch(&db, job_id, &w_id) + .fetch(&db, &job_id, &w_id) .await; if let Ok(job) = job { jobs.insert(job.id().to_string(), job); @@ -637,6 +637,7 @@ async fn list_selected_job_groups( #[derive(Deserialize)] struct GetJobQuery { pub no_logs: Option, + pub no_code: Option, } async fn get_job( @@ -644,7 +645,7 @@ async fn get_job( opt_tokened: OptTokened, Extension(db): Extension, Path((w_id, id)): Path<(String, Uuid)>, - Query(GetJobQuery { no_logs }): Query, + Query(GetJobQuery { no_logs, no_code }): Query, ) -> error::Result { let tags = opt_authed .as_ref() @@ -655,10 +656,14 @@ async fn get_job( .with_auth(&opt_authed) .with_in_tags(tags.as_ref()); + if no_code.unwrap_or(false) { + get = get.without_code(); + } + if no_logs.unwrap_or(false) { get = get.without_logs(); } - let mut job = get.fetch(&db, id, &w_id).await?; + let mut job = get.fetch(&db, &id, &w_id).await?; job.fetch_outstanding_wait_time(&db).await?; log_job_view( @@ -938,7 +943,7 @@ impl<'a> GetQuery<'a> { async fn fetch_queued( self, db: &DB, - job_id: Uuid, + job_id: &Uuid, workspace_id: &str, ) -> error::Result>> { let query = get_job_query!("v2_job_queue", @@ -970,7 +975,7 @@ impl<'a> GetQuery<'a> { async fn fetch_completed( self, db: &DB, - job_id: Uuid, + job_id: &Uuid, workspace_id: &str, ) -> error::Result>> { let query = get_job_query!("v2_job_completed", @@ -1007,7 +1012,7 @@ impl<'a> GetQuery<'a> { Ok(cjob) } - async fn fetch(self, db: &DB, job_id: Uuid, workspace_id: &str) -> error::Result { + async fn fetch(self, db: &DB, job_id: &Uuid, workspace_id: &str) -> error::Result { let cjob = self .fetch_completed(db.into(), job_id, workspace_id) .await? @@ -2092,7 +2097,7 @@ async fn resume_suspended_job_internal( .without_logs() .without_code() .without_flow() - .fetch(&db, parent_flow_info.id, &w_id) + .fetch(&db, &parent_flow_info.id, &w_id) .await?; let flow_status = parent_flow .flow_status() @@ -2390,7 +2395,7 @@ pub async fn get_suspended_job_flow( let flow = GetQuery::new() .without_logs() .without_code() - .fetch(&db, flow_id, &w_id) + .fetch(&db, &flow_id, &w_id) .await?; let flow_status = flow @@ -3800,7 +3805,7 @@ pub async fn run_workflow_as_code( let job = GetQuery::new() .without_logs() - .fetch_queued(&db, job_id, &w_id) + .fetch_queued(&db, &job_id, &w_id) .await?; if *CLOUD_HOSTED { @@ -5682,6 +5687,7 @@ pub struct JobUpdate { pub mem_peak: Option, pub progress: Option, pub flow_status: Option>, + pub job: Option, } async fn get_log_file(Path((_w_id, file_p)): Path<(String, String)>) -> error::Result { @@ -5751,6 +5757,7 @@ async fn get_job_update( log_offset, get_progress, running, + false, ).await?)) } @@ -5810,10 +5817,15 @@ fn get_job_update_sse_stream( log_offset, get_progress, running, + true ).await { + if update.completed.unwrap_or(false) { + return; + } 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); @@ -5830,9 +5842,11 @@ fn get_job_update_sse_stream( } // Poll for updates every 1 second - let mut interval = tokio::time::interval(std::time::Duration::from_secs(1)); + let mut i = 0; loop { - interval.tick().await; + i += 1; + let ms_duration = if i > 10 { 1000 } else if i > 50 { 5000 } else { 100 }; + tokio::time::sleep(std::time::Duration::from_millis(ms_duration)).await; match get_job_update_data( &opt_authed, @@ -5843,6 +5857,7 @@ fn get_job_update_sse_stream( log_offset, get_progress, running, + true, ).await { Ok(update) => { if let Ok(serialized) = serde_json::to_string(&update) { @@ -5852,6 +5867,9 @@ fn get_job_update_sse_stream( if tx.send(event_data).await.is_err() { break; } + if update.completed.unwrap_or(false) { + break; + } last_update = Some(serialized); // Update log offset if available @@ -5859,21 +5877,6 @@ fn get_job_update_sse_stream( log_offset = new_offset; } - // Check if job is completed - if update.completed.unwrap_or(false) { - // let full_job = get_job_by_id(db, &job_id).await?; - - let mut get = GetQuery::new() - .with_auth(&opt_authed) - .without_logs(); - - let full_job = get.fetch(&db, job_id, &w_id).await?; - - tx.send( - format!("data: {{'completed': true, 'job': {}}}\n\n", - serde_json::to_string(&full_job).unwrap())).await.unwrap(); - break; - } } } } @@ -5897,6 +5900,7 @@ async fn get_job_update_data( log_offset: i32, get_progress: Option, running: bool, + get_full_job_on_completion: bool, ) -> error::Result { let record = sqlx::query!( "SELECT @@ -5960,6 +5964,15 @@ async fn get_job_update_data( ) .await?; + let job = if record.completed.unwrap_or(false) && get_full_job_on_completion { + let get = GetQuery::new() + .with_auth(&opt_authed) + .without_logs(); + Some(get.fetch(&db, job_id, &w_id).await?) + } else { + None + }; + Ok(JobUpdate { running: record.running, completed: record.completed, @@ -5967,6 +5980,7 @@ async fn get_job_update_data( new_logs: record.logs, mem_peak: record.mem_peak, progress: record.progress, + job, flow_status: record .flow_status .map(|x: sqlx::types::Json>| x.0), @@ -6259,7 +6273,7 @@ async fn get_completed_job<'a>( let job_o = GetQuery::new() .with_auth(&opt_authed) .with_in_tags(tags.as_ref()) - .fetch_completed(&db, id, &w_id) + .fetch_completed(&db, &id, &w_id) .await?; let cj = not_found_if_none(job_o, "Completed Job", id.to_string())?; diff --git a/frontend/src/lib/components/Dev.svelte b/frontend/src/lib/components/Dev.svelte index 2b10436f63..22c2928955 100644 --- a/frontend/src/lib/components/Dev.svelte +++ b/frontend/src/lib/components/Dev.svelte @@ -638,6 +638,7 @@ + @@ -181,10 +202,10 @@ >{#if content}{@const len = (content?.length ?? 0) + (loadedFromObjectStore?.length ?? 0)}{#if downloadStartUrl}
{:else if len > LOG_LIMIT}(truncated to the last {LOG_LIMIT} characters)...


{/if}{@html html}{:else if isLoading}Waiting for job to start...{:else}No logs are available yet{/if} @@ -209,7 +230,7 @@ {/if} - + {#if !noAutoScroll}
{#if content}{@const len = (content?.length ?? 0) + - (loadedFromObjectStore?.length ?? 0)}{#if downloadStartUrl}
{:else if len > LOG_LIMIT}(truncated to the last {LOG_LIMIT} characters)

{/if}
{/if}{@html html}{:else if !isLoading}{customEmptyMessage}{/if} diff --git a/frontend/src/lib/components/ModuleTest.svelte b/frontend/src/lib/components/ModuleTest.svelte index 02ec203898..80930bd471 100644 --- a/frontend/src/lib/components/ModuleTest.svelte +++ b/frontend/src/lib/components/ModuleTest.svelte @@ -117,6 +117,7 @@ jobDone()} bind:scriptProgress diff --git a/frontend/src/lib/components/ScriptEditor.svelte b/frontend/src/lib/components/ScriptEditor.svelte index 011e3592db..cf96ccbf44 100644 --- a/frontend/src/lib/components/ScriptEditor.svelte +++ b/frontend/src/lib/components/ScriptEditor.svelte @@ -445,6 +445,7 @@ - +