diff --git a/backend/plot.py b/backend/plot.py new file mode 100644 index 0000000000..ab6e630916 --- /dev/null +++ b/backend/plot.py @@ -0,0 +1,59 @@ +import json +import matplotlib.pyplot as plt + +# Function to load JSON data from a file +def load_json_data(filepath): + with open(filepath, 'r') as file: + data = json.load(file) + return data + +# Function to plot two arrays of subarrays with tuples (step_name, duration) +def plot_two_arrays_of_subarrays(arrays1, arrays2): + # Function to calculate sum of durations for each step + def calculate_sums(arrays): + steps = [step for step, _ in arrays[0]] # Extract steps from the first iteration + sums = {step: 0 for step in steps} # Initialize sums dictionary with step names + + # Sum up the durations for each step across all subarrays + for subarray in arrays: + for step_name, duration in subarray: + sums[step_name] += duration + + # Convert the sums dictionary to two lists (for plotting) + step_names = list(sums.keys()) + durations = list(sums.values()) + return step_names, durations + + # Calculate sums for both arrays of subarrays + step_names1, sums1 = calculate_sums(arrays1) + step_names2, sums2 = calculate_sums(arrays2) + + # Create two subplots, one on top of the other + fig, (ax1, ax2) = plt.subplots(2, 1, figsize=(10, 8)) + + # First plot (top) for the first array of subarrays + ax1.plot(step_names1, sums1, marker='o', linestyle='-', color='b') + ax1.set_title('Total Duration per Step - Main Loop') + ax1.set_xlabel('Step Name') + ax1.set_ylabel('Total Duration') + ax1.grid(True) + + # Second plot (bottom) for the second array of subarrays + ax2.plot(step_names2, sums2, marker='o', linestyle='-', color='r') + ax2.set_title('Total Duration per Step - Result Processor') + ax2.set_xlabel('Step Name') + ax2.set_ylabel('Total Duration') + ax2.grid(True) + + # Adjust layout so the plots don't overlap + plt.tight_layout() + + # Display the plot + plt.show() + +# Load arrays from the JSON files +arrays1 = load_json_data('/tmp/windmill/profiling_main.json') +arrays2 = load_json_data('/tmp/windmill/profiling_result_processor.json') + +# Plot the data +plot_two_arrays_of_subarrays(arrays1, arrays2) \ No newline at end of file diff --git a/backend/src/monitor.rs b/backend/src/monitor.rs index 2c3220f237..0e58589813 100644 --- a/backend/src/monitor.rs +++ b/backend/src/monitor.rs @@ -525,7 +525,7 @@ fn read_log_counters(ts_str: String) -> (usize, usize) { ok_lines = counter.non_error_count; err_lines = counter.error_count; } else { - println!("no counter found for {ts_str}"); + // println!("no counter found for {ts_str}"); } } else { println!("Error reading log counters 2"); diff --git a/backend/windmill-api/src/flows.rs b/backend/windmill-api/src/flows.rs index 6e74866203..0b5082c6e0 100644 --- a/backend/windmill-api/src/flows.rs +++ b/backend/windmill-api/src/flows.rs @@ -242,6 +242,7 @@ pub async fn get_hub_flow_by_id( #[derive(Deserialize)] pub struct ToggleWorkspaceErrorHandler { + #[cfg(feature = "enterprise")] pub muted: Option, } diff --git a/backend/windmill-api/src/jobs.rs b/backend/windmill-api/src/jobs.rs index d187914abb..156be41b26 100644 --- a/backend/windmill-api/src/jobs.rs +++ b/backend/windmill-api/src/jobs.rs @@ -70,7 +70,7 @@ use windmill_common::{ oauth2::HmacSha256, scripts::{ScriptHash, ScriptLang}, users::username_to_permissioned_as, - utils::{not_found_if_none, now_from_db, paginate, require_admin, Pagination, StripPath}, + utils::{not_found_if_none, now_from_db, paginate, paginate_without_limits, require_admin, Pagination, StripPath}, }; #[cfg(all(feature = "enterprise", feature = "parquet"))] diff --git a/backend/windmill-api/src/scripts.rs b/backend/windmill-api/src/scripts.rs index aedb16bb15..bcede4f5e2 100644 --- a/backend/windmill-api/src/scripts.rs +++ b/backend/windmill-api/src/scripts.rs @@ -967,6 +967,7 @@ async fn list_paths( #[derive(Deserialize)] pub struct ToggleWorkspaceErrorHandler { + #[cfg(feature = "enterprise")] pub muted: Option, } diff --git a/backend/windmill-api/src/workspaces.rs b/backend/windmill-api/src/workspaces.rs index 685286d750..95cd97fad0 100644 --- a/backend/windmill-api/src/workspaces.rs +++ b/backend/windmill-api/src/workspaces.rs @@ -42,7 +42,10 @@ use windmill_common::schedule::Schedule; use windmill_common::users::username_to_permissioned_as; use windmill_common::variables::build_crypt; use windmill_common::worker::{to_raw_value, CLOUD_HOSTED}; -use windmill_common::workspaces::{WorkspaceDeploymentUISettings, WorkspaceGitSyncSettings}; +#[cfg(feature = "enterprise")] +use windmill_common::workspaces::WorkspaceDeploymentUISettings; +#[cfg(feature = "enterprise")] +use windmill_common::workspaces::WorkspaceGitSyncSettings; use windmill_common::{ error::{to_anyhow, Error, JsonResult, Result}, flows::Flow, @@ -991,6 +994,7 @@ async fn edit_large_file_storage_config( #[derive(Deserialize)] pub struct EditGitSyncConfig { + #[cfg(feature = "enterprise")] pub git_sync_settings: Option, } @@ -1056,6 +1060,7 @@ async fn edit_git_sync_config( #[derive(Deserialize)] struct EditDeployUIConfig { + #[cfg(feature = "enterprise")] deploy_ui_settings: Option, } @@ -1064,7 +1069,6 @@ async fn edit_deploy_ui_config( _authed: ApiAuthed, Extension(_db): Extension, Path(_w_id): Path, - Json(_new_config): Json, ) -> Result { return Err(Error::BadRequest( "Deployment UI is only available on Windmill Enterprise Edition".to_string(), @@ -1122,6 +1126,7 @@ async fn edit_deploy_ui_config( #[derive(Deserialize)] pub struct EditDefaultApp { + #[cfg(feature = "enterprise")] pub default_app_path: Option, } diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs index ec2e688871..8734dd9d24 100644 --- a/backend/windmill-queue/src/jobs.rs +++ b/backend/windmill-queue/src/jobs.rs @@ -37,8 +37,7 @@ use ulid::Ulid; use uuid::Uuid; use windmill_audit::audit_ee::{audit_log, AuditAuthor}; use windmill_audit::ActionKind; -#[cfg(not(feature = "enterprise"))] -use windmill_common::worker::PriorityTags; + use windmill_common::{ auth::{fetch_authed_from_permissioned_as, permissioned_as_to_username}, db::{Authed, UserDB}, @@ -62,9 +61,12 @@ use windmill_common::{ to_raw_value, DEFAULT_TAGS_PER_WORKSPACE, DEFAULT_TAGS_WORKSPACES, NO_LOGS, WORKER_CONFIG, WORKER_PULL_QUERIES, WORKER_SUSPENDED_PULL_QUERY, }, - BASE_URL, DB, METRICS_ENABLED, + DB, METRICS_ENABLED, }; +#[cfg(feature = "enterprise")] +use windmill_common::BASE_URL; + #[cfg(feature = "cloud")] use windmill_common::users::SUPERADMIN_SYNC_EMAIL; @@ -125,10 +127,12 @@ const MAX_FREE_CONCURRENT_RUNS: i32 = 30; const ERROR_HANDLER_USERNAME: &str = "error_handler"; const SCHEDULE_ERROR_HANDLER_USERNAME: &str = "schedule_error_handler"; +#[cfg(feature = "enterprise")] const SCHEDULE_RECOVERY_HANDLER_USERNAME: &str = "schedule_recovery_handler"; const ERROR_HANDLER_USER_GROUP: &str = "g/error_handler"; const ERROR_HANDLER_USER_EMAIL: &str = "error_handler@windmill.dev"; const SCHEDULE_ERROR_HANDLER_USER_EMAIL: &str = "schedule_error_handler@windmill.dev"; +#[cfg(feature = "enterprise")] const SCHEDULE_RECOVERY_HANDLER_USER_EMAIL: &str = "schedule_recovery_handler@windmill.dev"; #[derive(Clone, Debug)] @@ -477,6 +481,7 @@ where #[derive(Deserialize)] struct RawFlowFailureModule { + #[cfg(feature = "enterprise")] failure_module: Option>, } @@ -671,6 +676,7 @@ pub async fn add_completed_job< } } // tracing::error!("Added completed job {:#?}", queued_job); + #[cfg(feature = "enterprise")] let mut skip_downstream_error_handlers = false; tx = delete_job(tx, &queued_job.workspace_id, job_id).await?; // tracing::error!("3 {:?}", start.elapsed()); @@ -716,7 +722,10 @@ pub async fn add_completed_job< .await?; if let Some(schedule) = schedule { - skip_downstream_error_handlers = schedule.ws_error_handler_muted; + #[cfg(feature = "enterprise")] + { + skip_downstream_error_handlers = schedule.ws_error_handler_muted; + } // script or flow that failed on start and might not have been rescheduled let schedule_next_tick = !queued_job.is_flow() @@ -749,6 +758,7 @@ pub async fn add_completed_job< }; } + #[cfg(feature = "enterprise")] if let Err(err) = apply_schedule_handlers( rsmq.clone(), db, @@ -1324,6 +1334,8 @@ struct CompletedJobSubset { result: Option>>, started_at: chrono::DateTime, } + +#[cfg(feature = "enterprise")] async fn apply_schedule_handlers< 'a, 'c, @@ -1342,7 +1354,6 @@ async fn apply_schedule_handlers< job_priority: Option, ) -> windmill_common::error::Result<()> { if !success { - #[cfg(feature = "enterprise")] if let Some(on_failure_path) = schedule.on_failure.clone() { let times = schedule.on_failure_times.unwrap_or(1).max(1); let exact = schedule.on_failure_exact.unwrap_or(false); @@ -1392,7 +1403,6 @@ async fn apply_schedule_handlers< .await?; } } else { - #[cfg(feature = "enterprise")] if let Some(ref on_success_path) = schedule.on_success { handle_successful_schedule( db, @@ -1410,7 +1420,6 @@ async fn apply_schedule_handlers< .await?; } - #[cfg(feature = "enterprise")] if let Some(ref on_recovery_path) = schedule.on_recovery.clone() { let tx: QueueTransaction<'_, R> = (rsmq.clone(), db.begin().await?).into(); let times = schedule.on_recovery_times.unwrap_or(1).max(1); @@ -1579,6 +1588,7 @@ fn sanitize_result(result: Json<&T>) -> HashMap, +} + +impl Serialize for BenchmarkInfo { + fn serialize(&self, serializer: S) -> Result + where + S: serde::Serializer, + { + let timings: Vec> = self + .timings + .iter() + .map(|x| x.timings.clone()) + .collect::>>(); + //serialize timings as vec of vec of tuples + timings.serialize(serializer) + } +} + +impl BenchmarkInfo { + pub fn new() -> Self { + BenchmarkInfo { iters: 0, timings: vec![] } + } + + pub fn add_iter(&mut self, bench: BenchmarkIter) { + self.iters += 1; + self.timings.push(bench); + } + + pub fn write_to_file(&self, path: &str) -> anyhow::Result<()> { + println!("Writing benchmark {path}"); + write_file(TMP_DIR, path, &serde_json::to_string(&self).unwrap()).expect("write profiling"); + Ok(()) + } +} + +pub struct BenchmarkIter { + last_instant: Instant, + timings: Vec<(String, u32)>, +} + +impl BenchmarkIter { + pub fn new() -> Self { + BenchmarkIter { last_instant: Instant::now(), timings: vec![] } + } + + pub fn add_timing(&mut self, name: &str) { + let elapsed = self.last_instant.elapsed().as_nanos() as u32; + self.timings.push((name.to_string(), elapsed)); + self.last_instant = Instant::now(); + } +} + +pub async fn benchmark_init(is_dedicated_worker: bool, db: &DB) { + use windmill_common::{jobs::JobKind, scripts::ScriptLang}; + + let benchmark_jobs: i32 = std::env::var("BENCHMARK_JOBS_AT_INIT") + .unwrap_or("5000".to_string()) + .parse::() + .unwrap(); + if is_dedicated_worker { + // you need to create the script first, check https://github.com/windmill-labs/windmill/blob/b76a92cfe454c686f005c65f534e29e039f3c706/benchmarks/lib.ts#L47 + let hash = sqlx::query_scalar!( + "SELECT hash FROM script WHERE path = $1 AND workspace_id = $2", + "f/benchmarks/dedicated", + "admins" + ) + .fetch_one(db) + .await + .unwrap_or_else(|_e| panic!("failed to insert dedicated jobs")); + sqlx::query!("INSERT INTO queue (id, script_hash, script_path, job_kind, language, tag, created_by, permissioned_as, email, scheduled_for, workspace_id) (SELECT gen_random_uuid(), $1, $2, $3, $4, $5, $6, $7, $8, $9, $10 FROM generate_series(1, $11))", + hash, + "f/benchmarks/dedicated", + JobKind::Script as JobKind, + ScriptLang::Bun as ScriptLang, + "admins:f/benchmarks/dedicated", + "admin", + "u/admin", + "admin@windmill.dev", + chrono::Utc::now(), + "admins", + benchmark_jobs + ) + .execute(db) + .await.unwrap_or_else(|_e| panic!("failed to insert dedicated jobs")); + } else { + sqlx::query!("INSERT INTO queue (id, script_hash, script_path, job_kind, language, tag, created_by, permissioned_as, email, scheduled_for, workspace_id) (SELECT gen_random_uuid(), $1, $2, $3, $4, $5, $6, $7, $8, $9, $10 FROM generate_series(1, $11))", + None::, + None::, + JobKind::Noop as JobKind, + ScriptLang::Deno as ScriptLang, + "deno", + "admin", + "u/admin", + "admin@windmill.dev", + chrono::Utc::now(), + "admins", + benchmark_jobs + ) + .execute(db) + .await.unwrap_or_else(|_e| panic!("failed to insert noop jobs")); + } +} diff --git a/backend/windmill-worker/src/handle_child.rs b/backend/windmill-worker/src/handle_child.rs index e71629f3ca..d588ea8fc4 100644 --- a/backend/windmill-worker/src/handle_child.rs +++ b/backend/windmill-worker/src/handle_child.rs @@ -13,7 +13,6 @@ use windmill_common::error::{self, Error}; use windmill_common::worker::{get_windmill_memory_usage, get_worker_memory_usage, CLOUD_HOSTED}; -use anyhow::Result; use windmill_queue::{append_logs, CanceledBy}; #[cfg(any(target_os = "linux", target_os = "macos"))] diff --git a/backend/windmill-worker/src/lib.rs b/backend/windmill-worker/src/lib.rs index 2fc5d513dd..b30a2a5068 100644 --- a/backend/windmill-worker/src/lib.rs +++ b/backend/windmill-worker/src/lib.rs @@ -7,6 +7,8 @@ mod snowflake_executor; mod ansible_executor; mod bash_executor; +#[cfg(feature = "benchmark")] +mod bench; mod bun_executor; pub mod common; mod config; diff --git a/backend/windmill-worker/src/result_processor.rs b/backend/windmill-worker/src/result_processor.rs index 5203a65432..5b417c804f 100644 --- a/backend/windmill-worker/src/result_processor.rs +++ b/backend/windmill-worker/src/result_processor.rs @@ -26,9 +26,14 @@ use crate::{ bash_executor::ANSI_ESCAPE_RE, common::{read_result, save_in_cache}, worker_flow::update_flow_status_after_job_completion, - AuthedClient, Histo, JobCompleted, JobCompletedSender, SameWorkerSender, SendResult, + AuthedClient, JobCompleted, JobCompletedSender, SameWorkerSender, SendResult, }; +use crate::add_time; + +#[cfg(feature = "benchmark")] +use crate::bench::BenchmarkIter; + async fn send_job_completed( job_completed_tx: JobCompletedSender, job: Arc, @@ -168,9 +173,8 @@ pub async fn handle_receive_completed_job< same_worker_tx: &SameWorkerSender, rsmq: Option, worker_name: &str, - worker_save_completed_job_duration: Option, - worker_flow_transition_duration: Option, job_completed_tx: Sender, + #[cfg(feature = "benchmark")] bench: &mut BenchmarkIter, ) { let token = jc.token.clone(); let workspace = jc.job.workspace_id.clone(); @@ -191,9 +195,9 @@ pub async fn handle_receive_completed_job< same_worker_tx.clone(), rsmq.clone(), worker_name, - worker_save_completed_job_duration, - worker_flow_transition_duration, job_completed_tx.clone(), + #[cfg(feature = "benchmark")] + bench, ) .await { @@ -224,9 +228,8 @@ pub async fn process_completed_job, worker_name: &str, - _worker_save_completed_job_duration: Option, - _worker_flow_transition_duration: Option, job_completed_tx: Sender, + #[cfg(feature = "benchmark")] bench: &mut BenchmarkIter, ) -> windmill_common::error::Result<()> { if success { // println!("bef completed job{:?}", SystemTime::now()); @@ -238,10 +241,9 @@ pub async fn process_completed_job { + #[cfg(feature = "benchmark")] + { + $bench.add_timing($name); + // println!("{}: {:?}", $z, $y.elapsed()); + } + }; +} + pub async fn create_token_for_owner_in_bg( db: &Pool, job: &QueuedJob, @@ -541,37 +552,11 @@ impl AuthedClient { } } -#[cfg(feature = "benchmark")] -#[derive(Serialize)] -struct BenchmarkInfo { - iters: u64, - timings: Vec>, -} -#[macro_export] -macro_rules! add_time { - ($x:expr, $y:expr, $z:expr) => { - #[cfg(feature = "benchmark")] - { - $x.push($y.elapsed().as_nanos() as u32); - // println!("{}: {:?}", $z, $y.elapsed()); - } - }; -} - -#[cfg(feature = "prometheus")] -pub type Histo = Arc; -#[cfg(feature = "prometheus")] -type GGauge = Arc>; - -#[cfg(not(feature = "prometheus"))] -pub type Histo = (); -#[cfg(not(feature = "prometheus"))] -type GGauge = (); #[allow(dead_code)] #[derive(Clone)] -pub struct JobCompletedSender(Sender, Option, Option); +pub struct JobCompletedSender(Sender); #[derive(Clone)] @@ -588,16 +573,7 @@ impl JobCompletedSender { &self, jc: JobCompleted, ) -> Result<(), tokio::sync::mpsc::error::SendError> { - #[cfg(feature = "prometheus")] - if let Some(wj) = self.1.as_ref() { - wj.inc() - } - #[cfg(feature = "prometheus")] - let timer = self.2.as_ref().map(|x| x.start_timer()); - let r = self.0.send(SendResult::JobCompleted(jc)).await; - #[cfg(feature = "prometheus")] - timer.map(|x| x.stop_and_record()); - r + self.0.send(SendResult::JobCompleted(jc)).await } } @@ -804,44 +780,7 @@ pub async fn run_worker, - None::, - JobKind::Noop as JobKind, - ScriptLang::Deno as ScriptLang, - "deno", - "admin", - "u/admin", - "admin@windmill.dev", - chrono::Utc::now(), - "admins", - jobs - ) - .execute(db) - .await.unwrap_or_else(|_e| panic!("failed to insert noop jobs")); - } - } - - #[cfg(feature = "benchmark")] - let completed_jobs = Arc::new(AtomicUsize::new(0)); - #[cfg(feature = "benchmark")] - let start = Instant::now(); - #[cfg(feature = "benchmark")] - let main_duration = Arc::new(AtomicUsize::new(0)); - #[cfg(feature = "benchmark")] - let send_duration = Arc::new(AtomicUsize::new(0)); - #[cfg(feature = "benchmark")] - let process_duration = Arc::new(AtomicUsize::new(0)); - - #[cfg(feature = "benchmark")] - let main_duration2 = main_duration.clone(); - #[cfg(feature = "benchmark")] - let send_duration2 = send_duration.clone(); - - #[cfg(feature = "prometheus")] - let worker_job_completed_channel_queue2 = worker_job_completed_channel_queue.clone(); - #[cfg(feature = "prometheus")] - let worker_save_completed_job_duration2 = worker_save_completed_job_duration.clone(); - #[cfg(feature = "prometheus")] - let worker_flow_transition_duration2 = worker_flow_transition_duration.clone(); - - #[cfg(not(feature = "prometheus"))] - let worker_save_completed_job_duration2 = None; - #[cfg(not(feature = "prometheus"))] - let worker_flow_transition_duration2 = None; let worker_name2 = worker_name.clone(); let killpill_tx2 = killpill_tx.clone(); @@ -1151,14 +961,24 @@ pub async fn run_worker { - #[cfg(feature = "prometheus")] - if let Some(wj) = worker_job_completed_channel_queue2.as_ref() { - wj.dec(); - } let rsmq2 = rsmq2.clone(); let is_init_script_and_failure = !jc.success && jc.job.tag.as_str() == INIT_SCRIPT_TAG; @@ -1166,6 +986,8 @@ pub async fn run_worker>, ) = (HashMap::new(), false, vec![]); - #[cfg(feature = "benchmark")] - tracing::info!("pre loop time {}s", start.elapsed().as_secs_f64()); - if i_worker == 1 { if let Err(e) = queue_init_bash_maybe(db, same_worker_tx.clone(), &worker_name, rsmq.clone()).await @@ -1327,10 +1159,7 @@ pub async fn run_worker { @@ -1554,43 +1386,18 @@ pub async fn run_worker { @@ -1791,12 +1594,13 @@ pub async fn run_worker( } } -// async fn process_result( -// client: AuthedClient, -// job: QueuedJob, -// result: error::Result, -// cached_res_path: Option, -// db: &DB, -// worker_dir: &str, -// job_dir: &str, -// metrics: Option, -// same_worker_tx: Sender, -// base_internal_url: &str, -// rsmq: Option, -// job_completed_tx: Sender, -// logs: String, -// ) -> error::Result<()> { - -// fn build_language_metrics( -// worker_execution_failed: &HashMap< -// Option, -// prometheus::core::GenericCounter, -// >, -// language: &Option, -// ) -> Option { -// let metrics = if METRICS_ENABLED.load(std::sync::atomic::Ordering::Relaxed) { -// Some(Metrics { -// worker_execution_failed: worker_execution_failed -// .get(language) -// .expect("no timer found") -// .clone(), -// }) -// } else { -// None -// }; -// metrics -// } - -// pub async fn create_barrier_for_all_workers(num_workers: u32, sync_barrier: Arc>>) { -// tracing::debug!("acquiring write lock"); -// let mut barrier = sync_barrier.write().await; -// *barrier = Some(tokio::sync::Barrier::new(num_workers as usize)); -// drop(barrier); -// tracing::debug!("dropped write lock"); -// if let Some(b) = sync_barrier.read().await.as_ref() { -// tracing::debug!("leader worker waiting for barrier"); -// b.wait().await; -// tracing::debug!("leader worker done waiting for barrier"); -// }; -// let mut barrier = sync_barrier.write().await; -// *barrier = None; -// tracing::debug!("leader worker done waiting for"); -// } - pub enum SendResult { JobCompleted(JobCompleted), UpdateFlow { @@ -1994,19 +1741,6 @@ pub enum SendResult { Kill, } -// db: &DB, -// client: &AuthedClient, -// flow: uuid::Uuid, -// job_id_for_status: &Uuid, -// w_id: &str, -// success: bool, -// result: &'a RawValue, -// unrecoverable: bool, -// same_worker_tx: Sender, -// worker_dir: &str, -// stop_early_override: Option, -// rsmq: Option, -// worker_name: &str, #[derive(Debug, Clone)] pub struct JobCompleted { @@ -2076,8 +1810,6 @@ async fn handle_queued_job( rsmq: Option, job_completed_tx: JobCompletedSender, occupancy_metrics: &mut OccupancyMetrics, - _worker_flow_initial_transition_duration: Option, - _worker_code_execution_duration: Option, ) -> windmill_common::error::Result { if job.canceled { return Err(Error::JsonErr(canceled_job_to_result(&job))); @@ -2334,8 +2066,6 @@ async fn handle_queued_job( .map(|x| x.to_owned()) .unwrap_or_else(|| serde_json::from_str("{}").unwrap())), _ => { - #[cfg(feature = "prometheus")] - let timer = _worker_code_execution_duration.map(|x| x.start_timer()); let metric_timer = Instant::now(); let r = handle_code_execution_job( job.as_ref(), @@ -2353,8 +2083,6 @@ async fn handle_queued_job( ) .await; occupancy_metrics.total_duration_of_running_jobs += metric_timer.elapsed().as_secs_f32(); - #[cfg(feature = "prometheus")] - timer.map(|x| x.stop_and_record()); r } };