make benchmarks even more stable

backend/.sqlx/query-c92cc71e6d10c41368f7aa75b0799c2e1e9ca0ed33077ade8d7560e7cc21fa06.json

@@ -1,6 +1,6 @@
 {
   "db_name": "PostgreSQL",
-  "query": "DELETE FROM v2_job_queue WHERE workspace_id = $1 AND id = $2 RETURNING 1",
+  "query": "DELETE FROM v2_job_queue WHERE id = $1 RETURNING 1",
   "describe": {
     "columns": [
       {
@@ -11,7 +11,6 @@
     ],
     "parameters": {
       "Left": [
-        "Text",
         "Uuid"
       ]
     },
@@ -19,5 +18,5 @@
       null
     ]
   },
-  "hash": "d25c58d2722ad3dcd91101ce6f66e1d802dd5d82e1cd5f5ed3a15cbc75eb6745"
+  "hash": "c92cc71e6d10c41368f7aa75b0799c2e1e9ca0ed33077ade8d7560e7cc21fa06"
 }

backend/.sqlx/query-ddf2eccb78a310ed00c7d8b9c3f05d394a7cbcf0038c72a78add5c7b02ef5927.json

@@ -15,7 +15,7 @@
       ]
     },
     "nullable": [
-      null
+      true
     ]
   },
   "hash": "ddf2eccb78a310ed00c7d8b9c3f05d394a7cbcf0038c72a78add5c7b02ef5927"

backend/windmill-common/src/lib.rs

@@ -281,7 +281,7 @@ pub async fn connect_db(
 pub async fn connect(
     database_url: &str,
     max_connections: u32,
-    _worker_mode: bool,
+    worker_mode: bool,
 ) -> Result<sqlx::Pool<sqlx::Postgres>, error::Error> {
     use std::time::Duration;
 
@@ -289,18 +289,18 @@ pub async fn connect(
         .min_connections((max_connections / 5).clamp(3, max_connections))
         .max_connections(max_connections)
         .max_lifetime(Duration::from_secs(30 * 60)) // 30 mins
-        // .after_connect(move |conn, _| {
-        //     if worker_mode {
-        //         Box::pin(async move {
-        //             // sqlx::query("SET enable_seqscan = OFF;")
-        //             //     .execute(conn)
-        //             //     .await?;
-        //             Ok(())
-        //         })
-        //     } else {
-        //         Box::pin(async move { Ok(()) })
-        //     }
-        // })
+        .after_connect(move |conn, _| {
+            if worker_mode {
+                Box::pin(async move {
+                    sqlx::query("SET enable_seqscan = OFF;")
+                        .execute(conn)
+                        .await?;
+                    Ok(())
+                })
+            } else {
+                Box::pin(async move { Ok(()) })
+            }
+        })
         .connect_with(
             sqlx::postgres::PgConnectOptions::from_str(database_url)?.statement_cache_capacity(400),
         )

backend/windmill-queue/src/jobs.rs

@@ -566,6 +566,9 @@ pub async fn add_completed_job<T: Serialize + Send + Sync + ValidableJson>(
     let result_columns = result_columns.as_ref();
     let _job_id = queued_job.id;
     let (opt_uuid, _duration, _skip_downstream_error_handlers) = (|| async {
+
+        // let start = std::time::Instant::now();
+
         let mut tx = db.begin().await?;
 
         let job_id = queued_job.id;
@@ -663,7 +666,7 @@ pub async fn add_completed_job<T: Serialize + Send + Sync + ValidableJson>(
         // tracing::error!("Added completed job {:#?}", queued_job);
 
         let mut _skip_downstream_error_handlers = false;
-        tx = delete_job(tx, &queued_job.workspace_id, job_id).await?;
+        tx = delete_job(tx, &job_id).await?;
         // tracing::error!("3 {:?}", start.elapsed());
 
         if queued_job.is_flow_step {
@@ -858,6 +861,7 @@ pub async fn add_completed_job<T: Serialize + Send + Sync + ValidableJson>(
             "inserted completed job: {} (success: {success})",
             queued_job.id
         );
+        // tracing::info!("completed job: {:?}", start.elapsed().as_micros());
         Ok((None, _duration, _skip_downstream_error_handlers)) as windmill_common::error::Result<(Option<Uuid>, i64, bool)>
     })
     .retry(
@@ -2597,21 +2601,17 @@ async fn extract_result_from_job_result(
 
 pub async fn delete_job<'c>(
     mut tx: Transaction<'c, Postgres>,
-    w_id: &str,
-    job_id: Uuid,
+    job_id: &Uuid,
 ) -> windmill_common::error::Result<Transaction<'c, Postgres>> {
     #[cfg(feature = "prometheus")]
     if METRICS_ENABLED.load(std::sync::atomic::Ordering::Relaxed) {
         QUEUE_DELETE_COUNT.inc();
     }
 
-    let job_removed = sqlx::query_scalar!(
-        "DELETE FROM v2_job_queue WHERE workspace_id = $1 AND id = $2 RETURNING 1",
-        w_id,
-        job_id
-    )
-    .fetch_optional(&mut *tx)
-    .await;
+    let job_removed =
+        sqlx::query_scalar!("DELETE FROM v2_job_queue WHERE id = $1 RETURNING 1", job_id,)
+            .fetch_optional(&mut *tx)
+            .await;
 
     if let Err(job_removed) = job_removed {
         tracing::error!(