reuse existing transaction in push to reduce pool pressure (#7858)

* fix: reuse existing transaction in push instead of acquiring new connection

In push_inner, fetch_authed_from_permissioned_as was acquiring a new
connection from the pool to fetch job permissions, even though a
transaction was already open. Use fetch_authed_from_permissioned_as_conn
with the existing transaction instead, reducing pool pressure when many
jobs are pushed concurrently.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* improve contention

* improve contention

---------

Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>

backend/windmill-api-auth/src/lib.rs

@@ -447,8 +447,9 @@ pub async fn fetch_api_authed_from_permissioned_as(
         }
         _ => {
             tracing::debug!("API authed cache miss for user {}", email);
+
             let authed =
-                fetch_authed_from_permissioned_as(permissioned_as, email.clone(), w_id, db).await?;
+                fetch_authed_from_permissioned_as(&permissioned_as, &email, w_id, db).await?;
 
             let api_authed = ApiAuthed {
                 username: authed.username,

backend/windmill-common/src/auth.rs

@@ -244,20 +244,26 @@ pub fn permissioned_as_to_username(permissioned_as: &str) -> String {
     }
 }
 
-pub async fn fetch_authed_from_permissioned_as(
-    permissioned_as: String,
-    email: String,
-    w_id: &str,
-    db: &DB,
-) -> Result<Authed> {
-    let mut conn = db
-        .acquire()
-        .await
-        .map_err(|e| Error::internal_err(format!("acquiring connection: {e:#}")))?;
-    fetch_authed_from_permissioned_as_conn(&permissioned_as, &email, w_id, &mut conn).await
+pub fn fetch_authed_from_permissioned_as<'a, A>(
+    permissioned_as: &'a str,
+    email: &'a str,
+    w_id: &'a str,
+    db: A,
+) -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<Authed>> + Send + 'a>>
+where
+    A: sqlx::Acquire<'a, Database = sqlx::Postgres> + Send + 'a,
+{
+    Box::pin(async move {
+        let mut conn = db
+            .acquire()
+            .await
+            .map_err(|e| Error::internal_err(format!("acquiring connection: {e:#}")))?;
+
+        fetch_authed_from_permissioned_as_inner(permissioned_as, email, w_id, &mut *conn).await
+    })
 }
 
-pub async fn fetch_authed_from_permissioned_as_conn(
+async fn fetch_authed_from_permissioned_as_inner(
     permissioned_as: &str,
     email: &str,
     w_id: &str,
@@ -418,7 +424,7 @@ pub async fn create_token_for_owner(
         Ok(Some(jp)) => jp.into(),
         _ => {
             tracing::warn!("Could not get permissions for job {job_id} from job_perms table, getting permissions directly...");
-            fetch_authed_from_permissioned_as(owner.to_string(), email.to_string(), w_id, db)
+            fetch_authed_from_permissioned_as(owner, email, w_id, db)
                 .await
                 .map_err(|e| {
                     Error::internal_err(format!(

backend/windmill-queue/src/jobs.rs

@@ -53,7 +53,7 @@ use windmill_common::utils::{calculate_hash, configure_client, now_from_db};
 use windmill_common::worker::{Connection, SCRIPT_TOKEN_EXPIRY};
 
 use windmill_common::{
-    auth::{fetch_authed_from_permissioned_as, permissioned_as_to_username},
+    auth::permissioned_as_to_username,
     cache::{self, FlowData},
     db::{Authed, UserDB},
     error::{self, Error},
@@ -1827,7 +1827,7 @@ pub async fn try_schedule_next_job<'c>(
         &job.workspace_id
     );
 
-    let schedule_authed = windmill_common::auth::fetch_authed_from_permissioned_as_conn(
+    let schedule_authed = windmill_common::auth::fetch_authed_from_permissioned_as(
         &windmill_common::users::username_to_permissioned_as(&schedule.edited_by),
         &schedule.email,
         &job.workspace_id,
@@ -5418,11 +5418,11 @@ async fn push_inner<'c, 'd>(
             if authed.is_some() {
                 tracing::warn!("Authed passed to push is not the same as permissioned_as, refetching direclty permissions for job {job_id}...")
             }
-            fetch_authed_from_permissioned_as(
-                permissioned_as.clone(),
-                email.to_string(),
+            windmill_common::auth::fetch_authed_from_permissioned_as(
+                &permissioned_as,
+                email,
                 workspace_id,
-                _db,
+                &mut *tx,
             )
             .await
             .map_err(|e| {

backend/windmill-queue/src/schedule.rs

@@ -468,7 +468,7 @@ pub async fn push_scheduled_job<'c>(
     let push_authed = match push_authed {
         Some(a) => Some(a),
         None => {
-            obo_authed = windmill_common::auth::fetch_authed_from_permissioned_as_conn(
+            obo_authed = windmill_common::auth::fetch_authed_from_permissioned_as(
                 &permissioned_as,
                 email,
                 &schedule.workspace_id,