perf: allow BATCH_PULL_SIZE in any mode for benchmarking

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

backend/ee-repo-ref.txt

@@ -1 +1 @@
-1d4b7a31fc115d6aba8640f7cd3fd5a01abe6806
+a30e828a8742b8b644a25a603f0ee380a374607f

backend/src/main.rs

@@ -754,7 +754,7 @@ async fn windmill_main() -> anyhow::Result<()> {
         .and_then(|x| x.parse().ok())
         .unwrap_or(IpAddr::from(default_bind_addr));
 
-    let (conn, first_suffix, agent_config) = if mode == Mode::Agent {
+    let (conn, first_suffix, agent_config) = if matches!(mode, Mode::Agent | Mode::AgentBatch) {
         let agent_config = match AgentConfig::from_env() {
             Ok(config) => config,
             Err(e) => {
@@ -829,7 +829,7 @@ async fn windmill_main() -> anyhow::Result<()> {
 
     let _guard = windmill_common::tracing_init::initialize_tracing(&hostname, &mode, &environment);
 
-    let is_agent = mode == Mode::Agent;
+    let is_agent = matches!(mode, Mode::Agent | Mode::AgentBatch);
 
     let mut migration_handle: Option<JoinHandle<()>> = None;
     #[cfg(feature = "parquet")]
@@ -953,7 +953,7 @@ async fn windmill_main() -> anyhow::Result<()> {
         }
     }
 
-    let conn = if mode == Mode::Agent {
+    let conn = if matches!(mode, Mode::Agent | Mode::AgentBatch) {
         conn
     } else {
         // Drop the initial connection pool before creating the main one.
@@ -1289,7 +1289,7 @@ Windmill Community Edition {GIT_VERSION}
                                 )
                             },
                             worker_name: worker_name_with_suffix(
-                                mode == Mode::Agent,
+                                matches!(mode, Mode::Agent | Mode::AgentBatch),
                                 WORKER_GROUP.as_str(),
                                 &suffix,
                             ),

backend/windmill-common/src/utils.rs

@@ -122,6 +122,21 @@ lazy_static::lazy_static! {
                 }
                 #[cfg(feature = "enterprise")]
                 Mode::Agent
+            } else if &x == "agent-batch" {
+                println!("Binary is in 'agent-batch' mode with BASE_INTERNAL_URL={}", std::env::var("BASE_INTERNAL_URL").unwrap_or_default());
+                if std::env::var("BASE_INTERNAL_URL").is_err() {
+                    panic!("BASE_INTERNAL_URL is required in agent-batch mode")
+                }
+                if std::env::var("AGENT_TOKEN").is_err() {
+                    println!("AGENT_TOKEN is not passed. This is required for the agent to work and contains the JWT to authenticate with the server.")
+                }
+
+                #[cfg(not(feature = "enterprise"))]
+                {
+                    panic!("Agent-batch mode is only available in the EE, ignoring...");
+                }
+                #[cfg(feature = "enterprise")]
+                Mode::AgentBatch
             } else if &x == "indexer" {
                 tracing::info!("Binary is in 'indexer' mode");
                 #[cfg(not(feature = "tantivy"))]
@@ -474,6 +489,7 @@ pub fn map_string_to_number(s: &str, max_number: u64) -> u64 {
 pub enum Mode {
     Worker,
     Agent,
+    AgentBatch,
     Server,
     Standalone,
     Indexer,
@@ -485,6 +501,7 @@ impl std::fmt::Display for Mode {
         match self {
             Mode::Worker => write!(f, "worker"),
             Mode::Agent => write!(f, "agent"),
+            Mode::AgentBatch => write!(f, "agent-batch"),
             Mode::Server => write!(f, "server"),
             Mode::Standalone => write!(f, "standalone"),
             Mode::Indexer => write!(f, "indexer"),

backend/windmill-common/src/worker.rs

@@ -237,6 +237,12 @@ lazy_static::lazy_static! {
 
     pub static ref WORKER_PULL_QUERIES: Arc<RwLock<Vec<String>>> = Arc::new(RwLock::new(vec![]));
     pub static ref WORKER_SUSPENDED_PULL_QUERY: Arc<RwLock<String>> = Arc::new(RwLock::new("".to_string()));
+    pub static ref WORKER_BATCH_PULL_QUERIES: Arc<RwLock<Vec<String>>> = Arc::new(RwLock::new(vec![]));
+
+    pub static ref BATCH_PULL_SIZE: i32 = std::env::var("BATCH_PULL_SIZE")
+        .ok()
+        .and_then(|v| v.parse().ok())
+        .unwrap_or(0);
 
 
     pub static ref SMTP_CONFIG: Arc<RwLock<Option<Smtp>>> = Arc::new(RwLock::new(None));
@@ -522,16 +528,84 @@ pub fn make_pull_query(tags: &[String]) -> String {
 
 pub async fn store_pull_query(wc: &WorkerConfig) {
     let mut queries = vec![];
+    let mut batch_queries = vec![];
     for tags in wc.priority_tags_sorted.iter() {
         if tags.tags.len() == 0 {
             tracing::error!("Empty tags in priority tags, skipping");
             continue;
         }
-        let query = make_pull_query(&tags.tags);
-        queries.push(query);
+        queries.push(make_pull_query(&tags.tags));
+        batch_queries.push(make_batch_pull_query(&tags.tags));
     }
     let mut l = WORKER_PULL_QUERIES.write().await;
     *l = queries;
+    drop(l);
+    let mut l = WORKER_BATCH_PULL_QUERIES.write().await;
+    *l = batch_queries;
+}
+
+/// Build a batch pull query that claims up to $2 jobs at once.
+/// Uses $1 for worker_name and $2 for batch_size (i32).
+pub fn make_batch_pull_query(tags: &[String]) -> String {
+    format_batch_pull_query(format!(
+        "SELECT id
+        FROM v2_job_queue
+        WHERE running = false
+            AND tag IN ({}) AND scheduled_for <= now()
+            AND id NOT IN (SELECT id FROM v2_job WHERE same_worker = true)
+        ORDER BY priority DESC NULLS LAST, scheduled_for
+        FOR UPDATE SKIP LOCKED
+        LIMIT $2",
+        tags.iter().map(|x| format!("'{x}'")).join(", ")
+    ))
+}
+
+fn format_batch_pull_query(peek: String) -> String {
+    format!(
+        "WITH peek AS (
+            {}
+        ), q AS NOT MATERIALIZED (
+            UPDATE v2_job_queue SET
+                running = true,
+                started_at = coalesce(started_at, now()),
+                suspend_until = null,
+                worker = $1
+            WHERE id IN (SELECT id FROM peek)
+            RETURNING
+                id, started_at, scheduled_for,
+                canceled_by, canceled_reason, worker, cache_ignore_s3_path, runnable_settings_handle
+        ), r AS NOT MATERIALIZED (
+            UPDATE v2_job_runtime SET
+                ping = now()
+            WHERE id IN (SELECT id FROM peek)
+        ), j AS NOT MATERIALIZED (
+            SELECT
+                id, workspace_id, parent_job, created_by, created_at, runnable_id,
+                runnable_path, args, kind, trigger, trigger_kind,
+                permissioned_as, permissioned_as_email, script_lang,
+                flow_innermost_root_job, root_job, flow_step_id,
+                same_worker, pre_run_error, visible_to_owner, tag, concurrent_limit,
+                concurrency_time_window_s, timeout, cache_ttl, priority, raw_code, raw_lock,
+                raw_flow, script_entrypoint_override, preprocessed
+            FROM v2_job
+            WHERE id IN (SELECT id FROM peek)
+        ) SELECT j.id, j.workspace_id, j.parent_job, j.created_by, q.started_at, q.scheduled_for,
+            j.runnable_id, j.runnable_path, j.args, q.canceled_by,
+            q.canceled_reason, j.kind, j.trigger, j.trigger_kind, j.permissioned_as,
+            f.flow_status, j.script_lang,
+            j.same_worker, j.pre_run_error, j.visible_to_owner,
+            j.tag, j.concurrent_limit, j.concurrency_time_window_s, j.flow_innermost_root_job, j.root_job,
+            j.timeout, j.flow_step_id, j.cache_ttl, q.cache_ignore_s3_path, q.runnable_settings_handle, j.priority, j.raw_code, j.raw_lock, j.raw_flow,
+            j.script_entrypoint_override, j.preprocessed, COALESCE(pj.runnable_path, j.args->>'_FLOW_PATH') as parent_runnable_path,
+            COALESCE(p.email, j.permissioned_as_email) as permissioned_as_email, p.username as permissioned_as_username, p.is_admin as permissioned_as_is_admin,
+            p.is_operator as permissioned_as_is_operator, p.groups as permissioned_as_groups, p.folders as permissioned_as_folders, p.end_user_email as permissioned_as_end_user_email
+        FROM q
+            INNER JOIN j ON q.id = j.id
+            LEFT JOIN v2_job_status f ON f.id = j.id
+            LEFT JOIN job_perms p ON p.job_id = j.id
+            LEFT JOIN v2_job pj ON j.parent_job = pj.id",
+        peek
+    )
 }
 
 lazy_static::lazy_static! {

backend/windmill-queue/src/jobs.rs

@@ -443,6 +443,92 @@ pub async fn append_logs(
     }
 }
 
+/// Pull up to `batch_size` jobs at once using the given batch pull query.
+/// The query must use $1 for worker_name and $2 for batch_size (i32).
+pub async fn batch_pull(
+    db: &Pool<Postgres>,
+    worker_name: &str,
+    batch_query: &str,
+    batch_size: i32,
+) -> windmill_common::error::Result<Vec<PulledJob>> {
+    let jobs = sqlx::query_as::<_, PulledJob>(batch_query)
+        .bind(worker_name)
+        .bind(batch_size)
+        .fetch_all(db)
+        .await
+        .map_err(|e| {
+            windmill_common::error::Error::InternalErr(format!("batch pull error: {e:#}"))
+        })?;
+    Ok(jobs)
+}
+
+/// Batch-commit multiple completed jobs in a single transaction.
+/// Only for simple jobs (no flows, no schedules, no concurrency limits).
+/// Returns (job_id, duration_ms) for each committed job.
+pub async fn batch_commit_completed_jobs(
+    db: &Pool<Postgres>,
+    jobs: &[(Uuid, bool, &serde_json::value::RawValue, i32, Option<i64>)], // (id, success, result, mem_peak, duration)
+) -> windmill_common::error::Result<Vec<(Uuid, i64)>> {
+    if jobs.is_empty() {
+        return Ok(vec![]);
+    }
+    let mut tx = db.begin().await.map_err(|e| {
+        windmill_common::error::Error::InternalErr(format!("batch commit begin: {e:#}"))
+    })?;
+
+    let mut results = Vec::with_capacity(jobs.len());
+
+    for &(job_id, success, result, mem_peak, duration) in jobs {
+        let status = if success { "success" } else { "failure" };
+        let duration_ms: Option<i64> = sqlx::query_scalar(
+            "INSERT INTO v2_job_completed AS cj
+                (workspace_id, id, started_at, duration_ms, result,
+                 flow_status, workflow_as_code_status,
+                 memory_peak, status, worker)
+            SELECT q.workspace_id, q.id, started_at,
+                COALESCE($3::bigint, (EXTRACT('epoch' FROM (now())) - EXTRACT('epoch' FROM (COALESCE(started_at, now()))))*1000),
+                $2::jsonb, flow_status, workflow_as_code_status,
+                $4, $5::job_status, q.worker
+            FROM v2_job_queue q LEFT JOIN v2_job_status USING (id)
+            WHERE q.id = $1
+            ON CONFLICT (id) DO UPDATE SET status = EXCLUDED.status, result = $2::jsonb
+            RETURNING duration_ms",
+        )
+        .bind(job_id)
+        .bind(result.get())
+        .bind(duration)
+        .bind(if mem_peak > 0 { Some(mem_peak) } else { None })
+        .bind(status)
+        .fetch_optional(&mut *tx)
+        .await
+        .map_err(|e| {
+            windmill_common::error::Error::InternalErr(format!(
+                "batch commit insert job {job_id}: {e:#}"
+            ))
+        })?;
+
+        if let Some(dur) = duration_ms {
+            sqlx::query!("DELETE FROM v2_job_queue WHERE id = $1", job_id)
+                .execute(&mut *tx)
+                .await
+                .map_err(|e| {
+                    windmill_common::error::Error::InternalErr(format!(
+                        "batch commit delete job {job_id}: {e:#}"
+                    ))
+                })?;
+            results.push((job_id, dur));
+        } else {
+            tracing::warn!("batch commit: job {job_id} not found in queue, skipping");
+        }
+    }
+
+    tx.commit()
+        .await
+        .map_err(|e| windmill_common::error::Error::InternalErr(format!("batch commit: {e:#}")))?;
+
+    Ok(results)
+}
+
 pub const PERIODIC_SCRIPT_TAG: &str = "periodic_bash_script";
 pub const INIT_SCRIPT_TAG: &str = "init_script";
 pub const INIT_SCRIPT_PATH_PREFIX: &str = "init_script_";
@@ -3638,8 +3724,11 @@ pub fn resolve_debounce_key<'b>(
                 .join(":"),
         ));
 
-    tracing::debug!("Original debounce key (len={}): {}", original_debounce_key.len(), original_debounce_key);
-
+    tracing::debug!(
+        "Original debounce key (len={}): {}",
+        original_debounce_key.len(),
+        original_debounce_key
+    );
 
     // If debounce_key is not too long (< 255 chars), keep it as is, otherwise hash it.
     // On cloud, we prepend "{workspace_id}:" so we must reserve space for that prefix

backend/windmill-worker/src/agent_workers.rs

@@ -37,6 +37,15 @@ pub async fn pull_job(
         .await
 }
 
+pub async fn batch_pull_jobs(
+    client: &HttpClient,
+    batch_size: i32,
+) -> anyhow::Result<Vec<JobAndPerms>> {
+    client
+        .post("/api/agent_workers/batch_pull", None, &batch_size)
+        .await
+}
+
 pub async fn send_result(client: &HttpClient, jc: JobCompleted) -> anyhow::Result<String> {
     client
         .post(

backend/windmill-worker/src/result_processor.rs

@@ -218,6 +218,22 @@ enum JobCompletedRx {
     WakeUp,
 }
 
+fn is_batchable(jc: &JobCompleted) -> bool {
+    jc.success
+        && !jc.job.is_flow_step()
+        && jc.job.flow_step_id.as_deref() != Some("preprocessor")
+        && jc.preprocessed_args.is_none()
+        && jc.job.tag.as_str() != INIT_SCRIPT_TAG
+        && !matches!(
+            jc.job.kind,
+            JobKind::Dependencies | JobKind::FlowDependencies
+        )
+        && jc.canceled_by.is_none()
+        && jc.cached_res_path.is_none()
+        && jc.job.concurrent_limit.is_none()
+        && jc.job.schedule_path().is_none()
+}
+
 pub fn start_background_processor(
     job_completed_rx: JobCompletedReceiver,
     job_completed_sender: JobCompletedSender,
@@ -264,6 +280,8 @@ pub fn start_background_processor(
             }
         });
 
+        let mut batch_result_buffer: Vec<windmill_queue::JobCompleted> = Vec::new();
+
         //if we have been killed, we want to drain the queue of jobs
         while let Some(sr) = {
             if has_been_killed {
@@ -303,6 +321,97 @@ pub fn start_background_processor(
                     result: SendResultPayload::JobCompleted(jc),
                     time,
                 }) => {
+                    let batch_mode = *windmill_common::worker::BATCH_PULL_SIZE > 0
+                        || windmill_common::utils::MODE_AND_ADDONS.mode
+                            == windmill_common::utils::Mode::AgentBatch;
+
+                    if batch_mode && is_batchable(&jc) {
+                        // Accumulate for batch commit
+                        batch_result_buffer.push(jc);
+
+                        // Drain any additional ready results
+                        while let Ok(SendResult {
+                            result: SendResultPayload::JobCompleted(jc2),
+                            ..
+                        }) = bounded_rx.try_recv()
+                        {
+                            if is_batchable(&jc2) {
+                                batch_result_buffer.push(jc2);
+                            } else {
+                                process_jc(
+                                    jc2,
+                                    &worker_name,
+                                    &base_internal_url,
+                                    &db,
+                                    &worker_dir,
+                                    Some(&same_worker_tx),
+                                    &job_completed_sender,
+                                    &stats_map,
+                                    &killpill_rx,
+                                    #[cfg(feature = "benchmark")]
+                                    &mut bench,
+                                    #[cfg(feature = "benchmark")]
+                                    &mut infos,
+                                )
+                                .warn_after_seconds(10)
+                                .await;
+                            }
+                            if batch_result_buffer.len() >= 50 {
+                                break;
+                            }
+                        }
+
+                        // Flush batch
+                        if !batch_result_buffer.is_empty() {
+                            let batch_items: Vec<_> = batch_result_buffer
+                                .iter()
+                                .map(|jc| {
+                                    (
+                                        jc.job.id,
+                                        jc.success,
+                                        jc.result.as_ref() as &serde_json::value::RawValue,
+                                        jc.mem_peak,
+                                        jc.duration,
+                                    )
+                                })
+                                .collect();
+                            match windmill_queue::batch_commit_completed_jobs(&db, &batch_items)
+                                .await
+                            {
+                                Ok(committed) => {
+                                    tracing::debug!("batch committed {} jobs", committed.len());
+                                }
+                                Err(e) => {
+                                    tracing::error!(
+                                        "batch commit failed, falling back to individual: {e:#}"
+                                    );
+                                    for jc in batch_result_buffer.drain(..) {
+                                        process_jc(
+                                            jc,
+                                            &worker_name,
+                                            &base_internal_url,
+                                            &db,
+                                            &worker_dir,
+                                            Some(&same_worker_tx),
+                                            &job_completed_sender,
+                                            &stats_map,
+                                            &killpill_rx,
+                                            #[cfg(feature = "benchmark")]
+                                            &mut bench,
+                                            #[cfg(feature = "benchmark")]
+                                            &mut infos,
+                                        )
+                                        .warn_after_seconds(10)
+                                        .await;
+                                    }
+                                }
+                            }
+                            batch_result_buffer.clear();
+                        }
+
+                        last_processing_duration
+                            .store(time.elapsed().as_secs() as u16, Ordering::SeqCst);
+                    } else {
                         let is_init_script_and_failure =
                             !jc.success && jc.job.tag.as_str() == INIT_SCRIPT_TAG;
                         let is_dependency_job = matches!(
@@ -359,6 +468,7 @@ pub fn start_background_processor(
                         last_processing_duration
                             .store(time.elapsed().as_secs() as u16, Ordering::SeqCst);
                     }
+                }
                 JobCompletedRx::JobCompleted(SendResult {
                     result:
                         SendResultPayload::UpdateFlow(UpdateFlow {

backend/windmill-worker/src/worker.rs

@@ -2075,6 +2075,23 @@ pub async fn run_worker(
     let mut last_suspend_first = Instant::now();
     let mut killed_but_draining_same_worker_jobs = false;
 
+    let is_agent_batch =
+        windmill_common::utils::MODE_AND_ADDONS.mode == windmill_common::utils::Mode::AgentBatch;
+    let batch_pull_size = {
+        let s = *windmill_common::worker::BATCH_PULL_SIZE;
+        if s > 0 {
+            s
+        } else if is_agent_batch {
+            100
+        } else {
+            0
+        }
+    };
+    let mut batch_pull_buffer: std::collections::VecDeque<windmill_queue::PulledJob> =
+        std::collections::VecDeque::new();
+    let mut agent_batch_buffer: std::collections::VecDeque<windmill_queue::JobAndPerms> =
+        std::collections::VecDeque::new();
+
     let mut killpill_rx2 = killpill_rx.resubscribe();
 
     loop {
@@ -2318,6 +2335,53 @@ pub async fn run_worker(
                     tokio::time::sleep(Duration::from_millis(200)).await;
                     continue;
                 }
+            } else if batch_pull_size > 0 && !batch_pull_buffer.is_empty() {
+                // Serve from batch pull buffer
+                Ok(batch_pull_buffer
+                    .pop_front()
+                    .map(|job| NextJob::Sql { flow_runners: None, job }))
+            } else if batch_pull_size > 0 && matches!(&conn, Connection::Sql(_)) {
+                // Batch pull: try to refill buffer from DB
+                let db = conn.as_sql().unwrap();
+                let queries = windmill_common::worker::WORKER_BATCH_PULL_QUERIES
+                    .read()
+                    .await;
+                if queries.is_empty() {
+                    drop(queries);
+                    // Queries not populated yet, fall through to normal pull below
+                    None
+                } else {
+                    let mut pulled = Vec::new();
+                    for query in queries.iter() {
+                        match windmill_queue::batch_pull(db, &worker_name, query, batch_pull_size)
+                            .await
+                        {
+                            Ok(jobs) if !jobs.is_empty() => {
+                                pulled = jobs;
+                                break;
+                            }
+                            Ok(_) => {}
+                            Err(e) => {
+                                tracing::error!(worker = %worker_name, "batch pull error: {e:#}");
+                            }
+                        }
+                    }
+                    drop(queries);
+                    if pulled.is_empty() {
+                        Some(Ok(None))
+                    } else {
+                        let mut iter = pulled.into_iter();
+                        let first = iter.next();
+                        for job in iter {
+                            batch_pull_buffer.push_back(job);
+                        }
+                        Some(Ok(first.map(|job| NextJob::Sql { flow_runners: None, job })))
+                    }
+                }
+                .unwrap_or_else(|| {
+                    // Fall through: queries not ready yet
+                    Ok(None)
+                })
             } else {
                 match &conn {
                     Connection::Sql(db) => {
@@ -2453,10 +2517,36 @@ pub async fn run_worker(
                         }
                     }
 
-                    Connection::Http(client) => crate::agent_workers::pull_job(&client, None, None)
+                    Connection::Http(client) => {
+                        if batch_pull_size > 0 {
+                            // Agent-batch mode: serve from buffer, refill via batch pull
+                            if let Some(job) = agent_batch_buffer.pop_front() {
+                                Ok(Some(NextJob::Http(job)))
+                            } else {
+                                match crate::agent_workers::batch_pull_jobs(
+                                    &client,
+                                    batch_pull_size,
+                                )
+                                .await
+                                {
+                                    Ok(mut jobs) if !jobs.is_empty() => {
+                                        let first = jobs.remove(0);
+                                        for job in jobs {
+                                            agent_batch_buffer.push_back(job);
+                                        }
+                                        Ok(Some(NextJob::Http(first)))
+                                    }
+                                    Ok(_) => Ok(None),
+                                    Err(e) => Err(error::Error::InternalErr(e.to_string())),
+                                }
+                            }
+                        } else {
+                            crate::agent_workers::pull_job(&client, None, None)
                                 .await
                                 .map_err(|e| error::Error::InternalErr(e.to_string()))
-                        .map(|x| x.map(|y| NextJob::Http(y))),
+                                .map(|x| x.map(|y| NextJob::Http(y)))
+                        }
+                    }
                 }
             }
         };