diff --git a/backend/windmill-api/src/inputs.rs b/backend/windmill-api/src/inputs.rs index 03903fb9a1..75dfb1287e 100644 --- a/backend/windmill-api/src/inputs.rs +++ b/backend/windmill-api/src/inputs.rs @@ -114,7 +114,7 @@ async fn get_input_history( let sql = &format!( "select * from (select distinct on (args) * from completed_job \ where {} = $1 and job_kind = $2 and workspace_id = $3 \ - order by args, started_at desc limit $4 offset $5) t ORDER BY started_at desc", + order by args, started_at desc) t ORDER BY started_at desc limit $4 offset $5", r.runnable_type.column_name() ); diff --git a/backend/windmill-common/src/jobs.rs b/backend/windmill-common/src/jobs.rs index 7833aa7c8c..b4aaafd6a8 100644 --- a/backend/windmill-common/src/jobs.rs +++ b/backend/windmill-common/src/jobs.rs @@ -101,6 +101,46 @@ impl QueuedJob { } } +impl Default for QueuedJob { + fn default() -> Self { + Self { + workspace_id: "".to_string(), + id: Uuid::default(), + parent_job: None, + created_by: "".to_string(), + created_at: chrono::Utc::now(), + started_at: None, + scheduled_for: chrono::Utc::now(), + running: false, + script_hash: None, + script_path: None, + args: None, + logs: None, + raw_code: None, + raw_lock: None, + canceled: false, + canceled_by: None, + canceled_reason: None, + last_ping: None, + job_kind: JobKind::Identity, + schedule_path: None, + permissioned_as: "".to_string(), + flow_status: None, + raw_flow: None, + is_flow_step: false, + language: None, + same_worker: false, + pre_run_error: None, + email: "".to_string(), + visible_to_owner: false, + suspend: None, + mem_peak: None, + root_job: None, + leaf_jobs: None, + } + } +} + #[derive(Debug, Clone)] pub enum JobPayload { ScriptHub { path: String }, diff --git a/backend/windmill-worker/src/worker.rs b/backend/windmill-worker/src/worker.rs index 2e7c5106f3..0130026684 100644 --- a/backend/windmill-worker/src/worker.rs +++ b/backend/windmill-worker/src/worker.rs @@ -64,22 +64,25 @@ use crate::{ pub async fn create_token_for_owner_in_bg(db: &Pool, job: &QueuedJob) -> Arc> { let rw_lock = Arc::new(RwLock::new(String::new())); - let mut locked = rw_lock.clone().write_owned().await; - let db = db.clone(); - let job = job.clone(); - tokio::spawn(async move { + // skipping test runs + if job.workspace_id != "" { + let mut locked = rw_lock.clone().write_owned().await; + let db = db.clone(); let job = job.clone(); - let token = create_token_for_owner( - &db.clone(), - &job.workspace_id, - &job.permissioned_as, - "ephemeral-script", - *SESSION_TOKEN_EXPIRY, - &job.email, - ) - .await.expect("could not create job token"); - *locked = token; - }); + tokio::spawn(async move { + let job = job.clone(); + let token = create_token_for_owner( + &db.clone(), + &job.workspace_id, + &job.permissioned_as, + "ephemeral-script", + *SESSION_TOKEN_EXPIRY, + &job.email, + ) + .await.expect("could not create job token"); + *locked = token; + }); + }; return rw_lock; } @@ -422,9 +425,11 @@ pub async fn run_worker( let (same_worker_tx, mut same_worker_rx) = mpsc::channel::(5); + tracing::info!(worker = %worker_name, "listening for jobs"); - + let mut first_run = true; + loop { if *METRICS_ENABLED { worker_busy.set(0); @@ -464,44 +469,49 @@ pub async fn run_worker( last_sync = Instant::now(); } - let (do_break, next_job) = async { - tokio::select! { - biased; - _ = rx.recv() => { - if let Some(copy_cache_from_bucket_handle) = copy_cache_from_bucket_handle.as_ref() { - copy_cache_from_bucket_handle.abort(); - } - println!("received killpill for worker {}", i_worker); - (true, Ok(None)) - }, - _ = _copy_bucket_rx.recv() => { - if let Err(e) = move_tmp_cache_to_cache().await { - tracing::error!(worker = %worker_name, "failed to sync tmp cache to cache: {}", e); - } - copy_cache_from_bucket_handle = None; - initialized_cache = true; - (false, Ok(None)) - }, - Some(job_id) = same_worker_rx.recv() => { - (false, sqlx::query_as::<_, QueuedJob>("SELECT * FROM queue WHERE id = $1") - .bind(job_id) - .fetch_optional(db) - .await - .map_err(|_| Error::InternalErr("Impossible to fetch same_worker job".to_string()))) - }, - (job, timer) = { - let timer = if *METRICS_ENABLED { Some(worker_pull_duration.start_timer()) } else { None }; - pull(&db, WHITELIST_WORKSPACES.clone(), BLACKLIST_WORKSPACES.clone(), rsmq.clone()).map(|x| (x, timer)) - } => { - timer.map(|timer| { - let duration_pull_s = timer.stop_and_record(); - worker_pull_duration_counter.inc_by(duration_pull_s); - }); - (false, job) - }, - } - }.instrument(trace_span!("worker_get_next_job")).await; + let (do_break, next_job) = if first_run { + (false, Ok(Some(QueuedJob::default()))) + } else { + async { + tokio::select! { + biased; + _ = rx.recv() => { + if let Some(copy_cache_from_bucket_handle) = copy_cache_from_bucket_handle.as_ref() { + copy_cache_from_bucket_handle.abort(); + } + println!("received killpill for worker {}", i_worker); + (true, Ok(None)) + }, + _ = _copy_bucket_rx.recv() => { + if let Err(e) = move_tmp_cache_to_cache().await { + tracing::error!(worker = %worker_name, "failed to sync tmp cache to cache: {}", e); + } + copy_cache_from_bucket_handle = None; + initialized_cache = true; + (false, Ok(None)) + }, + Some(job_id) = same_worker_rx.recv() => { + (false, sqlx::query_as::<_, QueuedJob>("SELECT * FROM queue WHERE id = $1") + .bind(job_id) + .fetch_optional(db) + .await + .map_err(|_| Error::InternalErr("Impossible to fetch same_worker job".to_string()))) + }, + (job, timer) = { + let timer = if *METRICS_ENABLED { Some(worker_pull_duration.start_timer()) } else { None }; + pull(&db, WHITELIST_WORKSPACES.clone(), BLACKLIST_WORKSPACES.clone(), rsmq.clone()).map(|x| (x, timer)) + } => { + timer.map(|timer| { + let duration_pull_s = timer.stop_and_record(); + worker_pull_duration_counter.inc_by(duration_pull_s); + }); + (false, job) + }, + } + }.await + }; + first_run = false; if do_break { return true; } @@ -569,7 +579,7 @@ pub async fn run_worker( .expect("could not create shared dir"); } - let authed_client = AuthedClientBackgroundTask { base_internal_url: base_internal_url.to_string(), token: token, workspace: job.workspace_id.to_string(), client: OnceCell::new() }; + let authed_client = AuthedClientBackgroundTask { base_internal_url: base_internal_url.to_string(), token, workspace: job.workspace_id.to_string(), client: OnceCell::new() }; let is_flow = job.job_kind == JobKind::Flow || job.job_kind == JobKind::FlowPreview || job.job_kind == JobKind::FlowDependencies; if let Some(err) = handle_queued_job( @@ -834,6 +844,10 @@ async fn handle_queued_job( } }; + //it's a test job, no need to update the db + if job.workspace_id == "" { + return Ok(()); + } let client = &client.get_authed().await; match result { Ok(r) => {