remove unecessary postgres transactions and improve performance

backend/tests/worker.rs

@@ -13,7 +13,7 @@ use windmill_common::{
     jobs::{JobPayload, RawCode},
     scripts::ScriptLang,
 };
-use windmill_queue::get_queued_job;
+use windmill_queue::{get_queued_job, PushIsolationLevel};
 
 async fn initialize_tracing() {
     use std::sync::Once;
@@ -839,8 +839,10 @@ impl RunJob {
 
     async fn push(self, db: &Pool<Postgres>) -> Uuid {
         let RunJob { payload, args } = self;
+        let tx = PushIsolationLevel::IsolatedRoot(db.clone(), None);
         let (uuid, tx) = windmill_queue::push::<rsmq_async::MultiplexedRsmq>(
-            (None, db.begin().await.unwrap()).into(),
+            &db,
+            tx,
             "test-workspace",
             payload,
             args,

backend/windmill-api/src/apps.rs

@@ -8,8 +8,8 @@ use std::collections::HashMap;
  * LICENSE-AGPL for a copy of the license.
  */
 use crate::{
-    db::{UserDB, DB},
-    users::{require_owner_of_path, Authed, OptAuthed},
+    db::{ApiAuthed, DB},
+    users::{require_owner_of_path, OptAuthed},
     variables::build_crypt,
     webhook_util::{WebhookMessage, WebhookShared},
     HTTP_CLIENT,
@@ -30,6 +30,7 @@ use std::str;
 use windmill_audit::{audit_log, ActionKind};
 use windmill_common::{
     apps::ListAppQuery,
+    db::UserDB,
     error::{to_anyhow, Error, JsonResult, Result},
     jobs::{get_payload_tag_from_prefixed_path, JobPayload, RawCode},
     users::username_to_permissioned_as,
@@ -37,7 +38,7 @@ use windmill_common::{
         http_get_from_hub, list_elems_from_hub, not_found_if_none, paginate, Pagination, StripPath,
     },
 };
-use windmill_queue::{push, QueueTransaction};
+use windmill_queue::push;
 
 pub fn workspaced_service() -> Router {
     Router::new()
@@ -159,7 +160,7 @@ pub struct EditApp {
 }
 
 async fn list_apps(
-    authed: Authed,
+    authed: ApiAuthed,
     Extension(user_db): Extension<UserDB>,
     Path(w_id): Path<String>,
     Query(pagination): Query<Pagination>,
@@ -220,7 +221,7 @@ async fn list_apps(
 }
 
 async fn get_app(
-    authed: Authed,
+    authed: ApiAuthed,
     Extension(user_db): Extension<UserDB>,
     Path((w_id, path)): Path<(String, StripPath)>,
 ) -> JsonResult<AppWithLastVersion> {
@@ -245,7 +246,7 @@ async fn get_app(
 }
 
 async fn get_app_w_draft(
-    authed: Authed,
+    authed: ApiAuthed,
     Extension(user_db): Extension<UserDB>,
     Path((w_id, path)): Path<(String, StripPath)>,
 ) -> JsonResult<AppWithLastVersionAndDraft> {
@@ -276,7 +277,7 @@ async fn get_app_w_draft(
 }
 
 async fn get_app_by_id(
-    authed: Authed,
+    authed: ApiAuthed,
     Extension(user_db): Extension<UserDB>,
     Path((w_id, id)): Path<(String, i64)>,
 ) -> JsonResult<AppWithLastVersion> {
@@ -341,7 +342,7 @@ async fn get_public_app_by_secret(
 }
 
 async fn get_secret_id(
-    authed: Authed,
+    authed: ApiAuthed,
     Extension(user_db): Extension<UserDB>,
     Path((w_id, path)): Path<(String, StripPath)>,
 ) -> Result<String> {
@@ -369,7 +370,7 @@ async fn get_secret_id(
 }
 
 async fn create_app(
-    authed: Authed,
+    authed: ApiAuthed,
     Extension(user_db): Extension<UserDB>,
     Extension(webhook): Extension<WebhookShared>,
     Path(w_id): Path<String>,
@@ -461,7 +462,7 @@ async fn create_app(
     Ok((StatusCode::CREATED, app.path))
 }
 
-async fn list_hub_apps(Authed { email, .. }: Authed) -> JsonResult<serde_json::Value> {
+async fn list_hub_apps(ApiAuthed { email, .. }: ApiAuthed) -> JsonResult<serde_json::Value> {
     let flows = list_elems_from_hub(
         &HTTP_CLIENT,
         "https://hub.windmill.dev/searchUiData?approved=true",
@@ -472,7 +473,7 @@ async fn list_hub_apps(Authed { email, .. }: Authed) -> JsonResult<serde_json::V
 }
 
 pub async fn get_hub_app_by_id(
-    Authed { email, .. }: Authed,
+    ApiAuthed { email, .. }: ApiAuthed,
     Path(id): Path<i32>,
 ) -> JsonResult<serde_json::Value> {
     let value = http_get_from_hub(
@@ -489,7 +490,7 @@ pub async fn get_hub_app_by_id(
 }
 
 async fn delete_app(
-    authed: Authed,
+    authed: ApiAuthed,
     Extension(user_db): Extension<UserDB>,
     Extension(webhook): Extension<WebhookShared>,
     Path((w_id, path)): Path<(String, StripPath)>,
@@ -539,7 +540,7 @@ async fn delete_app(
 }
 
 async fn update_app(
-    authed: Authed,
+    authed: ApiAuthed,
     Extension(user_db): Extension<UserDB>,
     Extension(webhook): Extension<WebhookShared>,
     Path((w_id, path)): Path<(String, StripPath)>,
@@ -709,7 +710,6 @@ async fn execute_component(
     };
 
     let path = path.to_path();
-    let mut tx: QueueTransaction<'_, _> = (rsmq, db.begin().await?).into();
 
     let policy = if let Some(static_fields) = payload.clone().force_viewer_static_fields {
         let mut hm = HashMap::new();
@@ -738,7 +738,7 @@ async fn execute_component(
             path,
             &w_id
         )
-        .fetch_optional(&mut tx)
+        .fetch_optional(&db)
         .await?;
 
         let policy = not_found_if_none(policy_o, "App", path)?;
@@ -782,15 +782,16 @@ async fn execute_component(
             (payload, args, None)
         }
         ExecuteApp { args, component, raw_code: None, path: Some(path), .. } => {
-            let (payload, tag) =
-                get_payload_tag_from_prefixed_path(path, tx.transaction_mut(), &w_id).await?;
+            let (payload, tag) = get_payload_tag_from_prefixed_path(path, &db, &w_id).await?;
             let args = build_args(policy, component, path.to_string(), args)?;
             (payload, args, tag)
         }
         _ => unreachable!(),
     };
+    let tx = windmill_queue::PushIsolationLevel::IsolatedRoot(db.clone(), rsmq);
 
     let (uuid, tx) = push(
+        &db,
         tx,
         &w_id,
         job_payload,
@@ -841,7 +842,7 @@ fn get_on_behalf_of(policy: &Policy) -> Result<(String, String)> {
     Ok((permissioned_as, email))
 }
 
-pub async fn require_is_writer(authed: &Authed, path: &str, w_id: &str, db: DB) -> Result<()> {
+pub async fn require_is_writer(authed: &ApiAuthed, path: &str, w_id: &str, db: DB) -> Result<()> {
     return crate::users::require_is_writer(
         authed,
         path,

backend/windmill-api/src/audit.rs

@@ -12,9 +12,9 @@ use axum::{
     Extension, Json, Router,
 };
 use windmill_audit::{AuditLog, ListAuditLogQuery};
-use windmill_common::{error::JsonResult, utils::Pagination};
+use windmill_common::{db::UserDB, error::JsonResult, utils::Pagination};
 
-use crate::{db::UserDB, users::Authed};
+use crate::db::ApiAuthed;
 
 pub fn workspaced_service() -> Router {
     Router::new()
@@ -23,7 +23,7 @@ pub fn workspaced_service() -> Router {
 }
 
 async fn get_audit(
-    authed: Authed,
+    authed: ApiAuthed,
     Extension(user_db): Extension<UserDB>,
     Path(id): Path<i32>,
 ) -> JsonResult<AuditLog> {
@@ -32,7 +32,7 @@ async fn get_audit(
     Ok(Json(audit))
 }
 async fn list_audit(
-    authed: Authed,
+    authed: ApiAuthed,
     Extension(user_db): Extension<UserDB>,
     Path(w_id): Path<String>,
     Query(pagination): Query<Pagination>,

backend/windmill-api/src/capture.rs

@@ -14,14 +14,14 @@ use axum::{
 use hyper::{HeaderMap, StatusCode};
 use serde::Deserialize;
 use windmill_common::{
+    db::UserDB,
     error::{JsonResult, Result},
     utils::{not_found_if_none, StripPath},
 };
 
 use crate::{
-    db::{UserDB, DB},
+    db::{ApiAuthed, DB},
     jobs::add_include_headers,
-    users::Authed,
 };
 
 const KEEP_LAST: i64 = 8;
@@ -37,7 +37,7 @@ pub fn global_service() -> Router {
 }
 
 pub async fn new_payload(
-    authed: Authed,
+    authed: ApiAuthed,
     Extension(user_db): Extension<UserDB>,
     Path((w_id, path)): Path<(String, StripPath)>,
 ) -> Result<StatusCode> {
@@ -120,7 +120,7 @@ pub async fn update_payload(
 }
 
 pub async fn get_payload(
-    authed: Authed,
+    authed: ApiAuthed,
     Extension(user_db): Extension<UserDB>,
     Path((w_id, path)): Path<(String, StripPath)>,
 ) -> JsonResult<serde_json::Value> {

backend/windmill-api/src/db.rs

@@ -6,10 +6,11 @@
  * LICENSE-AGPL for a copy of the license.
  */
 
-use sqlx::{Pool, Postgres, Transaction};
-use windmill_common::error::Error;
-
-use crate::users::Authed;
+use sqlx::{Pool, Postgres};
+use windmill_common::{
+    db::{Authable, Authed},
+    error::Error,
+};
 
 pub type DB = Pool<Postgres>;
 
@@ -28,84 +29,58 @@ pub async fn migrate(db: &DB) -> Result<(), Error> {
     Ok(())
 }
 
-#[derive(Clone)]
-pub struct UserDB {
-    db: DB,
+#[derive(Clone, Debug)]
+pub struct ApiAuthed {
+    pub email: String,
+    pub username: String,
+    pub is_admin: bool,
+    pub is_operator: bool,
+    pub groups: Vec<String>,
+    // (folder name, can write, is owner)
+    pub folders: Vec<(String, bool, bool)>,
+    pub scopes: Option<Vec<String>>,
 }
 
-impl UserDB {
-    pub fn new(db: DB) -> Self {
-        Self { db }
-    }
-
-    pub async fn begin(
-        self,
-        authed: &Authed,
-    ) -> Result<Transaction<'static, Postgres>, sqlx::Error> {
-        let mut tx = self.db.begin().await?;
-        let user = if authed.is_admin {
-            "windmill_admin"
-        } else {
-            "windmill_user"
-        };
-
-        sqlx::query(&format!("SET LOCAL ROLE {}", user))
-            .execute(&mut *tx)
-            .await?;
-
-        sqlx::query!(
-            "SELECT set_config('session.user', $1, true)",
-            authed.username
-        )
-        .fetch_optional(&mut *tx)
-        .await?;
-
-        sqlx::query!(
-            "SELECT set_config('session.groups', $1, true)",
-            &authed.groups.join(",")
-        )
-        .fetch_optional(&mut *tx)
-        .await?;
-
-        sqlx::query!(
-            "SELECT set_config('session.pgroups', $1, true)",
-            &authed
-                .groups
-                .iter()
-                .map(|x| format!("g/{}", x))
-                .collect::<Vec<_>>()
-                .join(",")
-        )
-        .fetch_optional(&mut *tx)
-        .await?;
-
-        let (folders_write, folders_read): &(Vec<_>, Vec<_>) =
-            &authed.folders.clone().into_iter().partition(|x| x.1);
-
-        let mut folders_read = folders_read.clone();
-        folders_read.extend(folders_write.clone());
-        sqlx::query!(
-            "SELECT set_config('session.folders_read', $1, true)",
-            folders_read
-                .iter()
-                .map(|x| x.0.clone())
-                .collect::<Vec<_>>()
-                .join(",")
-        )
-        .fetch_optional(&mut *tx)
-        .await?;
-
-        sqlx::query!(
-            "SELECT set_config('session.folders_write', $1, true)",
-            folders_write
-                .iter()
-                .map(|x| x.0.clone())
-                .collect::<Vec<_>>()
-                .join(",")
-        )
-        .fetch_optional(&mut *tx)
-        .await?;
-
-        Ok(tx)
+impl From<ApiAuthed> for Authed {
+    fn from(value: ApiAuthed) -> Self {
+        Self {
+            email: value.email,
+            username: value.username,
+            is_admin: value.is_admin,
+            is_operator: value.is_operator,
+            groups: value.groups,
+            folders: value.folders,
+            scopes: value.scopes,
+        }
+    }
+}
+
+impl Authable for ApiAuthed {
+    fn is_admin(&self) -> bool {
+        self.is_admin
+    }
+
+    fn is_operator(&self) -> bool {
+        self.is_operator
+    }
+
+    fn groups(&self) -> &[String] {
+        &self.groups
+    }
+
+    fn folders(&self) -> &[(String, bool, bool)] {
+        &self.folders
+    }
+
+    fn scopes(&self) -> Option<&[std::string::String]> {
+        self.scopes.as_ref().map(|x| x.as_slice())
+    }
+
+    fn email(&self) -> &str {
+        &self.email
+    }
+
+    fn username(&self) -> &str {
+        &self.username
     }
 }

backend/windmill-api/src/drafts.rs

@@ -7,8 +7,8 @@
  */
 
 use crate::{
-    db::{UserDB, DB},
-    users::{maybe_refresh_folders, require_owner_of_path, Authed},
+    db::{ApiAuthed, DB},
+    users::{maybe_refresh_folders, require_owner_of_path},
 };
 
 use axum::{
@@ -18,7 +18,7 @@ use axum::{
 };
 use hyper::StatusCode;
 use serde::{Deserialize, Serialize};
-use windmill_common::{error::Result, utils::StripPath};
+use windmill_common::{db::UserDB, error::Result, utils::StripPath};
 
 pub fn workspaced_service() -> Router {
     Router::new()
@@ -43,7 +43,7 @@ pub struct Draft {
 }
 
 pub async fn require_writer_of_path(
-    authed: &Authed,
+    authed: &ApiAuthed,
     path: &str,
     w_id: &str,
     db: DB,
@@ -63,7 +63,7 @@ pub async fn require_writer_of_path(
 }
 
 async fn create_draft(
-    authed: Authed,
+    authed: ApiAuthed,
     Extension(db): Extension<DB>,
     Extension(user_db): Extension<UserDB>,
     Path(w_id): Path<String>,
@@ -95,7 +95,7 @@ async fn create_draft(
 }
 
 async fn delete_draft(
-    authed: Authed,
+    authed: ApiAuthed,
     Extension(user_db): Extension<UserDB>,
     Path((w_id, kind, path)): Path<(String, DraftType, StripPath)>,
 ) -> Result<String> {
@@ -115,7 +115,7 @@ async fn delete_draft(
 }
 
 // async fn get_draft(
-//     authed: Authed,
+//     authed: ApiAuthed,
 //     Extension(user_db): Extension<UserDB>,
 //     Path((w_id, path)): Path<(String, StripPath)>,
 // ) -> JsonResult<Draft> {

backend/windmill-api/src/favorite.rs

@@ -6,7 +6,7 @@
  * LICENSE-AGPL for a copy of the license.
  */
 
-use crate::{db::DB, users::Authed};
+use crate::db::{ApiAuthed, DB};
 use axum::{
     extract::{Extension, Path},
     routing::post,
@@ -39,7 +39,7 @@ pub struct Favorite {
 }
 
 async fn star(
-    authed: Authed,
+    authed: ApiAuthed,
     Extension(db): Extension<DB>,
     Path(w_id): Path<String>,
     Json(Favorite { favorite_kind, path }): Json<Favorite>,
@@ -58,7 +58,7 @@ async fn star(
 }
 
 async fn unstar(
-    authed: Authed,
+    authed: ApiAuthed,
     Extension(db): Extension<DB>,
     Path(w_id): Path<String>,
     Json(Favorite { favorite_kind, path }): Json<Favorite>,

backend/windmill-api/src/flows.rs

@@ -6,10 +6,11 @@
  * LICENSE-AGPL for a copy of the license.
  */
 
+use crate::db::ApiAuthed;
 use crate::{
-    db::{UserDB, DB},
+    db::DB,
     schedule::clear_schedule,
-    users::{maybe_refresh_folders, require_owner_of_path, Authed},
+    users::{maybe_refresh_folders, require_owner_of_path},
     webhook_util::{WebhookMessage, WebhookShared},
     HTTP_CLIENT,
 };
@@ -18,6 +19,7 @@ use axum::{
     routing::{delete, get, post},
     Json, Router,
 };
+
 use hyper::StatusCode;
 use serde::{Deserialize, Serialize};
 use sql_builder::prelude::*;
@@ -25,6 +27,7 @@ use sql_builder::SqlBuilder;
 use sqlx::{Postgres, Transaction};
 use windmill_audit::{audit_log, ActionKind};
 use windmill_common::{
+    db::UserDB,
     error::{self, to_anyhow, Error, JsonResult, Result},
     flows::{Flow, ListFlowQuery, ListableFlow, NewFlow},
     jobs::JobPayload,
@@ -34,7 +37,7 @@ use windmill_common::{
         http_get_from_hub, list_elems_from_hub, not_found_if_none, paginate, Pagination, StripPath,
     },
 };
-use windmill_queue::{push, schedule::push_scheduled_job, QueueTransaction};
+use windmill_queue::{push, schedule::push_scheduled_job, PushIsolationLevel, QueueTransaction};
 
 pub fn workspaced_service() -> Router {
     Router::new()
@@ -56,7 +59,7 @@ pub fn global_service() -> Router {
 }
 
 async fn list_flows(
-    authed: Authed,
+    authed: ApiAuthed,
     Extension(user_db): Extension<UserDB>,
     Path(w_id): Path<String>,
     Query(pagination): Query<Pagination>,
@@ -120,7 +123,7 @@ async fn list_flows(
     Ok(Json(rows))
 }
 
-async fn list_hub_flows(Authed { email, .. }: Authed) -> JsonResult<serde_json::Value> {
+async fn list_hub_flows(ApiAuthed { email, .. }: ApiAuthed) -> JsonResult<serde_json::Value> {
     let flows = list_elems_from_hub(
         &HTTP_CLIENT,
         "https://hub.windmill.dev/searchFlowData?approved=true",
@@ -131,7 +134,7 @@ async fn list_hub_flows(Authed { email, .. }: Authed) -> JsonResult<serde_json::
 }
 
 async fn list_paths(
-    authed: Authed,
+    authed: ApiAuthed,
     Extension(user_db): Extension<UserDB>,
     Path(w_id): Path<String>,
 ) -> JsonResult<Vec<String>> {
@@ -149,7 +152,7 @@ async fn list_paths(
 }
 
 pub async fn get_hub_flow_by_id(
-    Authed { email, .. }: Authed,
+    ApiAuthed { email, .. }: ApiAuthed,
     Path(id): Path<i32>,
 ) -> JsonResult<serde_json::Value> {
     let value = http_get_from_hub(
@@ -185,7 +188,7 @@ async fn check_path_conflict<'c>(
 }
 
 async fn create_flow(
-    authed: Authed,
+    authed: ApiAuthed,
     Extension(db): Extension<DB>,
     Extension(user_db): Extension<UserDB>,
     Extension(rsmq): Extension<Option<rsmq_async::MultiplexedRsmq>>,
@@ -246,7 +249,9 @@ async fn create_flow(
         WebhookMessage::CreateFlow { workspace: w_id.clone(), path: nf.path.clone() },
     );
 
+    let tx = PushIsolationLevel::Transaction(tx);
     let (dependency_job_uuid, mut tx) = push(
+        &db,
         tx,
         &w_id,
         JobPayload::FlowDependencies { path: nf.path.clone() },
@@ -305,7 +310,7 @@ async fn check_schedule_conflict<'c>(
     Ok(())
 }
 
-pub async fn require_is_writer(authed: &Authed, path: &str, w_id: &str, db: DB) -> Result<()> {
+pub async fn require_is_writer(authed: &ApiAuthed, path: &str, w_id: &str, db: DB) -> Result<()> {
     return crate::users::require_is_writer(
         authed,
         path,
@@ -318,7 +323,7 @@ pub async fn require_is_writer(authed: &Authed, path: &str, w_id: &str, db: DB)
 }
 
 async fn update_flow(
-    authed: Authed,
+    authed: ApiAuthed,
     Extension(user_db): Extension<UserDB>,
     Extension(rsmq): Extension<Option<rsmq_async::MultiplexedRsmq>>,
     Extension(db): Extension<DB>,
@@ -393,7 +398,7 @@ async fn update_flow(
         clear_schedule(tx.transaction_mut(), &schedule.path, true, &w_id).await?;
 
         if schedule.enabled {
-            tx = push_scheduled_job(tx, schedule).await?;
+            tx = push_scheduled_job(&db, tx, schedule).await?;
         }
     }
 
@@ -430,7 +435,10 @@ async fn update_flow(
         },
     );
 
+    let tx = PushIsolationLevel::Transaction(tx);
+
     let (dependency_job_uuid, mut tx) = push(
+        &db,
         tx,
         &w_id,
         JobPayload::FlowDependencies { path: nf.path.clone() },
@@ -473,7 +481,7 @@ async fn update_flow(
 }
 
 async fn get_flow_by_path(
-    authed: Authed,
+    authed: ApiAuthed,
     Extension(user_db): Extension<UserDB>,
     Path((w_id, path)): Path<(String, StripPath)>,
 ) -> JsonResult<Flow> {
@@ -508,7 +516,7 @@ pub struct FlowWDraft {
 }
 
 async fn get_flow_by_path_w_draft(
-    authed: Authed,
+    authed: ApiAuthed,
     Extension(user_db): Extension<UserDB>,
     Path((w_id, path)): Path<(String, StripPath)>,
 ) -> JsonResult<FlowWDraft> {
@@ -555,7 +563,7 @@ struct Archived {
 }
 
 async fn archive_flow_by_path(
-    authed: Authed,
+    authed: ApiAuthed,
     Extension(user_db): Extension<UserDB>,
     Extension(webhook): Extension<WebhookShared>,
     Path((w_id, path)): Path<(String, StripPath)>,
@@ -593,7 +601,7 @@ async fn archive_flow_by_path(
 }
 
 async fn delete_flow_by_path(
-    authed: Authed,
+    authed: ApiAuthed,
     Extension(user_db): Extension<UserDB>,
     Extension(webhook): Extension<WebhookShared>,
     Path((w_id, path)): Path<(String, StripPath)>,

backend/windmill-api/src/folders.rs

@@ -8,9 +8,11 @@
 
 use std::sync::Arc;
 
+use crate::db::ApiAuthed;
+
 use crate::{
-    db::{UserDB, DB},
-    users::{AuthCache, Authed, Tokened},
+    db::DB,
+    users::{AuthCache, Tokened},
     webhook_util::{WebhookMessage, WebhookShared},
 };
 use axum::{
@@ -22,6 +24,7 @@ use lazy_static::lazy_static;
 use regex::Regex;
 use windmill_audit::{audit_log, ActionKind};
 use windmill_common::{
+    db::UserDB,
     error::{self, to_anyhow, JsonResult, Result},
     users::username_to_permissioned_as,
     utils::{not_found_if_none, paginate, Pagination},
@@ -74,7 +77,7 @@ pub struct Owner {
 }
 
 async fn list_folders(
-    authed: Authed,
+    authed: ApiAuthed,
     Extension(user_db): Extension<UserDB>,
     Path(w_id): Path<String>,
     Query(pagination): Query<Pagination>,
@@ -96,7 +99,7 @@ async fn list_folders(
     Ok(Json(rows))
 }
 async fn list_foldernames(
-    authed: Authed,
+    authed: ApiAuthed,
     Extension(user_db): Extension<UserDB>,
     Path(w_id): Path<String>,
     Query(pagination): Query<Pagination>,
@@ -145,7 +148,7 @@ lazy_static! {
 }
 
 async fn create_folder(
-    authed: Authed,
+    authed: ApiAuthed,
     Tokened { token }: Tokened,
     Extension(user_db): Extension<UserDB>,
     Extension(webhook): Extension<WebhookShared>,
@@ -220,13 +223,13 @@ async fn create_folder(
 }
 
 pub async fn is_owner_api(
-    authed: Authed,
+    authed: ApiAuthed,
     Path((_w_id, name)): Path<(String, String)>,
 ) -> JsonResult<bool> {
     Ok(Json(is_owner(&authed, &name)))
 }
 
-pub fn is_owner(Authed { is_admin, folders, .. }: &Authed, name: &str) -> bool {
+pub fn is_owner(ApiAuthed { is_admin, folders, .. }: &ApiAuthed, name: &str) -> bool {
     if *is_admin {
         true
     } else {
@@ -234,7 +237,7 @@ pub fn is_owner(Authed { is_admin, folders, .. }: &Authed, name: &str) -> bool {
     }
 }
 
-pub fn require_is_owner(authed: &Authed, name: &str) -> Result<()> {
+pub fn require_is_owner(authed: &ApiAuthed, name: &str) -> Result<()> {
     if is_owner(authed, name) {
         Ok(())
     } else {
@@ -246,7 +249,7 @@ pub fn require_is_owner(authed: &Authed, name: &str) -> Result<()> {
 }
 
 async fn update_folder(
-    authed: Authed,
+    authed: ApiAuthed,
     Extension(user_db): Extension<UserDB>,
     Extension(webhook): Extension<WebhookShared>,
     Path((w_id, name)): Path<(String, String)>,
@@ -343,7 +346,7 @@ pub async fn get_folderopt<'c>(
 }
 
 async fn get_folder(
-    authed: Authed,
+    authed: ApiAuthed,
     Extension(user_db): Extension<UserDB>,
     Path((w_id, name)): Path<(String, String)>,
 ) -> JsonResult<Folder> {
@@ -365,7 +368,7 @@ struct FolderUsage {
     pub variables: i64,
 }
 async fn get_folder_usage(
-    authed: Authed,
+    authed: ApiAuthed,
     Extension(user_db): Extension<UserDB>,
     Path((w_id, name)): Path<(String, String)>,
 ) -> JsonResult<FolderUsage> {
@@ -446,7 +449,7 @@ async fn get_folder_usage(
 }
 
 async fn delete_folder(
-    authed: Authed,
+    authed: ApiAuthed,
     Extension(user_db): Extension<UserDB>,
     Extension(webhook): Extension<WebhookShared>,
     Path((w_id, name)): Path<(String, String)>,
@@ -483,7 +486,7 @@ async fn delete_folder(
 }
 
 async fn add_owner(
-    authed: Authed,
+    authed: ApiAuthed,
     Extension(user_db): Extension<UserDB>,
     Extension(webhook): Extension<WebhookShared>,
     Path((w_id, name)): Path<(String, String)>,
@@ -550,7 +553,7 @@ pub async fn get_folders_for_user(
 }
 
 async fn remove_owner(
-    authed: Authed,
+    authed: ApiAuthed,
     Extension(user_db): Extension<UserDB>,
     Extension(webhook): Extension<WebhookShared>,
     Path((w_id, name)): Path<(String, String)>,

backend/windmill-api/src/granular_acls.rs

@@ -6,18 +6,18 @@
  * LICENSE-AGPL for a copy of the license.
  */
 
-use crate::{
-    db::{UserDB, DB},
-    users::{require_owner_of_path, Authed},
-};
+use crate::{db::DB, users::require_owner_of_path};
 use axum::{
     extract::{Extension, Path},
     routing::{get, post},
     Json, Router,
 };
 
+use crate::db::ApiAuthed;
+
 use serde::{Deserialize, Serialize};
 use windmill_common::{
+    db::UserDB,
     error::{Error, JsonResult, Result},
     utils::{not_found_if_none, StripPath},
 };
@@ -36,7 +36,7 @@ pub struct GranularAcl {
 }
 
 async fn add_granular_acl(
-    authed: Authed,
+    authed: ApiAuthed,
     Extension(db): Extension<DB>,
     Extension(user_db): Extension<UserDB>,
     Path((w_id, path)): Path<(String, StripPath)>,
@@ -83,7 +83,7 @@ async fn add_granular_acl(
 }
 
 async fn remove_granular_acl(
-    authed: Authed,
+    authed: ApiAuthed,
     Extension(db): Extension<DB>,
     Extension(user_db): Extension<UserDB>,
     Path((w_id, path)): Path<(String, StripPath)>,
@@ -135,7 +135,7 @@ async fn remove_granular_acl(
 }
 
 async fn get_granular_acls(
-    authed: Authed,
+    authed: ApiAuthed,
     Extension(user_db): Extension<UserDB>,
     Path((w_id, path)): Path<(String, StripPath)>,
 ) -> JsonResult<serde_json::Value> {

backend/windmill-api/src/groups.rs

@@ -6,18 +6,16 @@
  * LICENSE-AGPL for a copy of the license.
  */
 
-use crate::{
-    db::{UserDB, DB},
-    users::{get_groups_for_user, Authed},
-    utils::require_super_admin,
-};
+use crate::db::ApiAuthed;
+use crate::{db::DB, users::get_groups_for_user, utils::require_super_admin};
+
 use axum::{
     extract::{Extension, Path, Query},
     routing::{delete, get, post},
     Json, Router,
 };
 use windmill_audit::{audit_log, ActionKind};
-use windmill_common::users::username_to_permissioned_as;
+use windmill_common::{db::UserDB, users::username_to_permissioned_as};
 use windmill_common::{
     error::{Error, JsonResult, Result},
     utils::{not_found_if_none, paginate, Pagination},
@@ -113,7 +111,7 @@ struct QueryListGroup {
     pub only_member_of: Option<bool>,
 }
 async fn list_group_names(
-    Authed { username, email, .. }: Authed,
+    ApiAuthed { username, email, .. }: ApiAuthed,
     Extension(db): Extension<DB>,
     Query(QueryListGroup { only_member_of }): Query<QueryListGroup>,
     Path(w_id): Path<String>,
@@ -158,7 +156,7 @@ async fn check_name_conflict<'c>(
 }
 
 pub async fn is_owner(
-    Authed { username, is_admin, groups, .. }: Authed,
+    ApiAuthed { username, is_admin, groups, .. }: ApiAuthed,
     Extension(db): Extension<DB>,
     Path((w_id, name)): Path<(String, String)>,
 ) -> JsonResult<bool> {
@@ -203,7 +201,7 @@ pub async fn require_is_owner(
 }
 
 async fn create_group(
-    authed: Authed,
+    authed: ApiAuthed,
     Extension(user_db): Extension<UserDB>,
     Path(w_id): Path<String>,
     Json(ng): Json<NewGroup>,
@@ -247,12 +245,13 @@ async fn create_group(
 }
 
 async fn create_igroup(
-    authed: Authed,
+    authed: ApiAuthed,
     Extension(db): Extension<DB>,
     Json(ng): Json<NewGroup>,
 ) -> Result<String> {
+    require_super_admin(&db, &authed.email).await?;
     let mut tx = db.begin().await?;
-    require_super_admin(&mut tx, &authed.email).await?;
+
     sqlx::query!(
         "INSERT INTO instance_group (name, summary) VALUES ($1, $2) ON CONFLICT DO NOTHING",
         ng.name,
@@ -277,12 +276,12 @@ async fn create_igroup(
 }
 
 async fn delete_igroup(
-    authed: Authed,
+    authed: ApiAuthed,
     Extension(db): Extension<DB>,
     Json(ng): Json<NewGroup>,
 ) -> Result<String> {
-    let mut tx = db.begin().await?;
-    require_super_admin(&mut tx, &authed.email).await?;
+    require_super_admin(&db, &authed.email).await?;
+    let mut tx: Transaction<'_, Postgres> = db.begin().await?;
     sqlx::query!("DELETE FROM instance_group WHERE name = $1", ng.name,)
         .execute(&mut *tx)
         .await?;
@@ -319,7 +318,7 @@ pub async fn get_group_opt<'c>(
 }
 
 async fn get_group(
-    authed: Authed,
+    authed: ApiAuthed,
     Extension(user_db): Extension<UserDB>,
     Path((w_id, name)): Path<(String, String)>,
 ) -> JsonResult<GroupInfo> {
@@ -358,7 +357,7 @@ async fn get_group(
 }
 
 async fn delete_group(
-    authed: Authed,
+    authed: ApiAuthed,
     Extension(db): Extension<DB>,
     Extension(user_db): Extension<UserDB>,
     Path((w_id, name)): Path<(String, String)>,
@@ -406,7 +405,7 @@ async fn delete_group(
 }
 
 async fn update_group(
-    authed: Authed,
+    authed: ApiAuthed,
     Extension(db): Extension<DB>,
     Extension(user_db): Extension<UserDB>,
     Path((w_id, name)): Path<(String, String)>,
@@ -442,7 +441,7 @@ async fn update_group(
 }
 
 async fn add_user(
-    authed: Authed,
+    authed: ApiAuthed,
     Extension(db): Extension<DB>,
     Extension(user_db): Extension<UserDB>,
     Path((w_id, name)): Path<(String, String)>,
@@ -479,14 +478,14 @@ async fn add_user(
 }
 
 async fn add_user_igroup(
-    authed: Authed,
+    authed: ApiAuthed,
     Extension(db): Extension<DB>,
     Path(name): Path<String>,
     Json(Email { email }): Json<Email>,
 ) -> Result<String> {
-    let mut tx = db.begin().await?;
+    require_super_admin(&db, &authed.email).await?;
 
-    require_super_admin(&mut tx, &authed.email).await?;
+    let mut tx: Transaction<'_, Postgres> = db.begin().await?;
 
     let group_opt = sqlx::query_scalar!("SELECT name FROM instance_group WHERE name = $1", name)
         .fetch_optional(&mut *tx)
@@ -521,10 +520,10 @@ struct IGroup {
     name: String,
     emails: Option<Vec<String>>,
 }
-async fn list_igroups(authed: Authed, Extension(db): Extension<DB>) -> JsonResult<Vec<IGroup>> {
-    let mut tx = db.begin().await?;
+async fn list_igroups(authed: ApiAuthed, Extension(db): Extension<DB>) -> JsonResult<Vec<IGroup>> {
+    require_super_admin(&db, &authed.email).await?;
+    let mut tx: Transaction<'_, Postgres> = db.begin().await?;
 
-    require_super_admin(&mut tx, &authed.email).await?;
     let groups = sqlx::query_as!(
         IGroup,
         "SELECT name, array_remove(array_agg(email_to_igroup.email), null) as emails FROM email_to_igroup RIGHT JOIN instance_group ON instance_group.name = email_to_igroup.igroup GROUP BY name"
@@ -549,15 +548,14 @@ async fn get_igroup(Path(name): Path<String>, Extension(db): Extension<DB>) -> J
 }
 
 async fn remove_user_igroup(
-    authed: Authed,
+    authed: ApiAuthed,
     Extension(db): Extension<DB>,
     Path(name): Path<String>,
     Json(Email { email }): Json<Email>,
 ) -> Result<String> {
+    require_super_admin(&db, &authed.email).await?;
     let mut tx = db.begin().await?;
 
-    require_super_admin(&mut tx, &authed.email).await?;
-
     let group_opt = sqlx::query_scalar!("SELECT name FROM instance_group WHERE name = $1", name,)
         .fetch_optional(&mut *tx)
         .await?;
@@ -587,7 +585,7 @@ async fn remove_user_igroup(
 }
 
 async fn remove_user(
-    authed: Authed,
+    authed: ApiAuthed,
     Extension(db): Extension<DB>,
     Extension(user_db): Extension<UserDB>,
     Path((w_id, name)): Path<(String, String)>,

backend/windmill-api/src/inputs.rs

@@ -6,7 +6,7 @@
  * LICENSE-AGPL for a copy of the license.
  */
 
-use crate::{db::UserDB, jobs::CompletedJob, users::Authed};
+use crate::{db::ApiAuthed, jobs::CompletedJob};
 use axum::{
     extract::{Path, Query},
     routing::{get, post},
@@ -21,6 +21,7 @@ use std::{
     vec,
 };
 use windmill_common::{
+    db::UserDB,
     error::JsonResult,
     jobs::JobKind,
     scripts::to_i64,
@@ -102,7 +103,7 @@ pub struct Input {
 }
 
 async fn get_input_history(
-    authed: Authed,
+    authed: ApiAuthed,
     Extension(user_db): Extension<UserDB>,
     Path(w_id): Path<String>,
     Query(pagination): Query<Pagination>,
@@ -158,7 +159,7 @@ async fn get_input_history(
 }
 
 async fn list_saved_inputs(
-    authed: Authed,
+    authed: ApiAuthed,
     Extension(user_db): Extension<UserDB>,
     Path(w_id): Path<String>,
     Query(pagination): Query<Pagination>,
@@ -209,7 +210,7 @@ pub struct CreateInput {
 }
 
 async fn create_input(
-    authed: Authed,
+    authed: ApiAuthed,
     Extension(user_db): Extension<UserDB>,
     Path(w_id): Path<String>,
     Query(r): Query<RunnableParams>,
@@ -245,7 +246,7 @@ pub struct UpdateInput {
 }
 
 async fn update_input(
-    authed: Authed,
+    authed: ApiAuthed,
     Extension(user_db): Extension<UserDB>,
     Path(w_id): Path<String>,
     Json(input): Json<UpdateInput>,
@@ -266,7 +267,7 @@ async fn update_input(
 }
 
 async fn delete_input(
-    authed: Authed,
+    authed: ApiAuthed,
     Extension(user_db): Extension<UserDB>,
     Path((w_id, i_id)): Path<(String, Uuid)>,
 ) -> JsonResult<String> {

backend/windmill-api/src/jobs.rs

@@ -6,9 +6,11 @@
  * LICENSE-AGPL for a copy of the license.
  */
 
+use crate::db::ApiAuthed;
+
 use crate::{
-    db::{UserDB, DB},
-    users::{check_scopes, require_owner_of_path, Authed, OptAuthed},
+    db::DB,
+    users::{check_scopes, require_owner_of_path, OptAuthed},
     utils::require_super_admin,
     variables::get_workspace_key,
     workers::{CUSTOM_TAGS, CUSTOM_TAGS_PER_WORKSPACE},
@@ -32,6 +34,7 @@ use tower_http::cors::{Any, CorsLayer};
 use urlencoding::encode;
 use windmill_audit::{audit_log, ActionKind};
 use windmill_common::{
+    db::UserDB,
     error::{self, to_anyhow, Error},
     flow_status::{Approval, FlowStatus, FlowStatusModule},
     flows::FlowValue,
@@ -41,7 +44,7 @@ use windmill_common::{
     users::username_to_permissioned_as,
     utils::{not_found_if_none, now_from_db, paginate, require_admin, Pagination, StripPath},
 };
-use windmill_queue::{get_queued_job, push, QueueTransaction};
+use windmill_queue::{get_queued_job, push, PushIsolationLevel};
 
 pub fn workspaced_service() -> Router {
     let cors = CorsLayer::new()
@@ -302,8 +305,8 @@ pub async fn get_path_for_hash<'c>(
     Ok(path)
 }
 
-pub async fn get_path_tag_and_limits_for_hash<'c>(
-    db: &mut Transaction<'c, Postgres>,
+pub async fn get_path_tag_and_limits_for_hash(
+    db: &DB,
     w_id: &str,
     hash: i64,
 ) -> error::Result<(String, Option<String>, Option<i32>, Option<i32>)> {
@@ -312,7 +315,7 @@ pub async fn get_path_tag_and_limits_for_hash<'c>(
         hash,
         w_id
     )
-    .fetch_one(&mut **db)
+    .fetch_one(db)
     .await
     .map_err(|e| {
         Error::InternalErr(format!(
@@ -455,12 +458,12 @@ lazy_static::lazy_static! {
 impl RunJobQuery {
     async fn get_scheduled_for<'c>(
         &self,
-        db: &mut Transaction<'c, Postgres>,
+        db: &DB,
     ) -> error::Result<Option<chrono::DateTime<chrono::Utc>>> {
         if let Some(scheduled_for) = self.scheduled_for {
             Ok(Some(scheduled_for))
         } else if let Some(scheduled_in_secs) = self.scheduled_in_secs {
-            let now = now_from_db(&mut **db).await?;
+            let now = now_from_db(db).await?;
             Ok(Some(now + chrono::Duration::seconds(scheduled_in_secs)))
         } else {
             Ok(None)
@@ -654,7 +657,7 @@ async fn list_queue_jobs(
 }
 
 async fn cancel_all(
-    authed: Authed,
+    authed: ApiAuthed,
     Extension(db): Extension<DB>,
     Path(w_id): Path<String>,
 ) -> error::JsonResult<Vec<Uuid>> {
@@ -690,7 +693,7 @@ async fn count_queue_jobs(
 }
 
 async fn list_jobs(
-    authed: Authed,
+    authed: ApiAuthed,
     Extension(user_db): Extension<UserDB>,
     Path(w_id): Path<String>,
     Query(pagination): Query<Pagination>,
@@ -807,7 +810,7 @@ async fn list_jobs(
 }
 
 pub async fn resume_suspended_flow_as_owner(
-    authed: Authed,
+    authed: ApiAuthed,
     Extension(db): Extension<DB>,
     Path((_w_id, flow_id)): Path<(String, Uuid)>,
     QueryOrBody(value): QueryOrBody<serde_json::Value>,
@@ -1130,7 +1133,7 @@ pub async fn get_suspended_job_flow(
 }
 
 pub async fn create_job_signature(
-    authed: Authed,
+    authed: ApiAuthed,
     Extension(user_db): Extension<UserDB>,
     Path((w_id, job_id, resume_id)): Path<(String, Uuid, u32)>,
     Query(approver): Query<QueryApprover>,
@@ -1175,7 +1178,7 @@ fn build_resume_url(
 }
 
 pub async fn get_resume_urls(
-    authed: Authed,
+    authed: ApiAuthed,
     Extension(user_db): Extension<UserDB>,
     Path((w_id, job_id, resume_id)): Path<(String, Uuid, u32)>,
     Query(approver): Query<QueryApprover>,
@@ -1528,7 +1531,8 @@ fn check_tag_available_for_workspace(w_id: &str, tag: &Option<String>) -> error:
 }
 
 pub async fn run_flow_by_path(
-    authed: Authed,
+    authed: ApiAuthed,
+    Extension(db): Extension<DB>,
     Extension(user_db): Extension<UserDB>,
     Extension(rsmq): Extension<Option<rsmq_async::MultiplexedRsmq>>,
     Path((w_id, flow_path)): Path<(String, StripPath)>,
@@ -1539,20 +1543,21 @@ pub async fn run_flow_by_path(
     let flow_path = flow_path.to_path();
     check_scopes(&authed, || format!("run:flow/{flow_path}"))?;
 
-    let mut tx: QueueTransaction<'_, _> = (rsmq, user_db.begin(&authed).await?).into();
     let tag = sqlx::query_scalar!(
         "SELECT tag from flow WHERE path = $1 and workspace_id = $2",
         flow_path,
         w_id
     )
-    .fetch_optional(&mut tx)
+    .fetch_optional(&db)
     .await?
     .flatten();
     check_tag_available_for_workspace(&w_id, &tag)?;
-    let scheduled_for = run_query.get_scheduled_for(tx.transaction_mut()).await?;
+    let scheduled_for = run_query.get_scheduled_for(&db).await?;
     let args = run_query.add_include_headers(headers, args.unwrap_or_default());
     let args = add_raw_string(raw_string, args);
+    let tx = PushIsolationLevel::Isolated(user_db.clone(), authed.clone().into(), rsmq);
     let (uuid, tx) = push(
+        &db,
         tx,
         &w_id,
         JobPayload::Flow(flow_path.to_string()),
@@ -1579,7 +1584,8 @@ pub async fn run_flow_by_path(
 }
 
 pub async fn run_job_by_path(
-    authed: Authed,
+    authed: ApiAuthed,
+    Extension(db): Extension<DB>,
     Extension(user_db): Extension<UserDB>,
     Extension(rsmq): Extension<Option<rsmq_async::MultiplexedRsmq>>,
     Path((w_id, script_path)): Path<(String, StripPath)>,
@@ -1590,15 +1596,16 @@ pub async fn run_job_by_path(
     let script_path = script_path.to_path();
     check_scopes(&authed, || format!("run:script/{script_path}"))?;
 
-    let mut tx: QueueTransaction<'_, _> = (rsmq, user_db.begin(&authed).await?).into();
-    let (job_payload, tag) =
-        script_path_to_payload(script_path, tx.transaction_mut(), &w_id).await?;
-    let scheduled_for = run_query.get_scheduled_for(tx.transaction_mut()).await?;
+    let (job_payload, tag) = script_path_to_payload(script_path, &db, &w_id).await?;
+    let scheduled_for = run_query.get_scheduled_for(&db).await?;
     let args = run_query.add_include_headers(headers, args.unwrap_or_default());
     let args = add_raw_string(raw_string, args);
+
     check_tag_available_for_workspace(&w_id, &tag)?;
+    let tx = PushIsolationLevel::Isolated(user_db.clone(), authed.clone().into(), rsmq);
 
     let (uuid, tx) = push(
+        &db,
         tx,
         &w_id,
         job_payload,
@@ -1629,7 +1636,7 @@ struct Guard {
     id: Uuid,
     w_id: String,
     db: UserDB,
-    authed: Authed,
+    authed: ApiAuthed,
 }
 
 impl Drop for Guard {
@@ -1661,7 +1668,7 @@ impl Drop for Guard {
 }
 
 async fn run_wait_result<T>(
-    authed: Authed,
+    authed: ApiAuthed,
     Extension(user_db): Extension<UserDB>,
     timeout: i32,
     uuid: Uuid,
@@ -1726,7 +1733,7 @@ async fn run_wait_result<T>(
     }
 }
 
-pub async fn check_queue_too_long(db: DB, queue_limit: Option<i64>) -> error::Result<()> {
+pub async fn check_queue_too_long(db: &DB, queue_limit: Option<i64>) -> error::Result<()> {
     if let Some(limit) = queue_limit {
         let count = sqlx::query_scalar!(
             "SELECT COUNT(*) FROM queue WHERE  canceled = false AND (scheduled_for <= now()
@@ -1734,7 +1741,7 @@ pub async fn check_queue_too_long(db: DB, queue_limit: Option<i64>) -> error::Re
             AND (   suspend <= 0
                  OR suspend_until <= now())))",
         )
-        .fetch_one(&db)
+        .fetch_one(db)
         .await?
         .unwrap_or(0);
 
@@ -1771,7 +1778,7 @@ lazy_static::lazy_static! {
 
 pub async fn run_wait_result_job_by_path_get(
     method: hyper::http::Method,
-    authed: Authed,
+    authed: ApiAuthed,
     Extension(rsmq): Extension<Option<rsmq_async::MultiplexedRsmq>>,
     Extension(user_db): Extension<UserDB>,
     Extension(db): Extension<DB>,
@@ -1792,16 +1799,16 @@ pub async fn run_wait_result_job_by_path_get(
         serde_json::Map::new()
     };
 
-    check_queue_too_long(db, QUEUE_LIMIT_WAIT_RESULT.or(run_query.queue_limit)).await?;
+    check_queue_too_long(&db, QUEUE_LIMIT_WAIT_RESULT.or(run_query.queue_limit)).await?;
     let script_path = script_path.to_path();
     check_scopes(&authed, || format!("run:script/{script_path}"))?;
 
-    let mut tx: QueueTransaction<'_, _> = (rsmq, user_db.clone().begin(&authed).await?).into();
-    let (job_payload, tag) =
-        script_path_to_payload(script_path, tx.transaction_mut(), &w_id).await?;
+    let (job_payload, tag) = script_path_to_payload(script_path, &db, &w_id).await?;
     check_tag_available_for_workspace(&w_id, &tag)?;
+    let tx = PushIsolationLevel::Isolated(user_db.clone(), authed.clone().into(), rsmq);
 
     let (uuid, tx) = push(
+        &db,
         tx,
         &w_id,
         job_payload,
@@ -1837,7 +1844,7 @@ pub async fn run_wait_result_job_by_path_get(
 
 pub async fn run_wait_result_flow_by_path_get(
     method: hyper::http::Method,
-    authed: Authed,
+    authed: ApiAuthed,
     Extension(rsmq): Extension<Option<rsmq_async::MultiplexedRsmq>>,
     Extension(user_db): Extension<UserDB>,
     Extension(db): Extension<DB>,
@@ -1875,7 +1882,7 @@ pub async fn run_wait_result_flow_by_path_get(
 }
 
 pub async fn run_wait_result_script_by_path(
-    authed: Authed,
+    authed: ApiAuthed,
     Extension(user_db): Extension<UserDB>,
     Extension(rsmq): Extension<Option<rsmq_async::MultiplexedRsmq>>,
     Extension(db): Extension<DB>,
@@ -1924,7 +1931,7 @@ fn convert_from_openai_json(
 }
 
 pub async fn openai_sync_script_by_path(
-    authed: Authed,
+    authed: ApiAuthed,
     Extension(user_db): Extension<UserDB>,
     Extension(rsmq): Extension<Option<rsmq_async::MultiplexedRsmq>>,
     Extension(db): Extension<DB>,
@@ -1952,7 +1959,7 @@ async fn run_wait_result_script_by_path_internal(
     db: sqlx::Pool<Postgres>,
     run_query: RunJobQuery,
     script_path: StripPath,
-    authed: Authed,
+    authed: ApiAuthed,
     rsmq: Option<rsmq_async::MultiplexedRsmq>,
     user_db: UserDB,
     w_id: String,
@@ -1960,19 +1967,19 @@ async fn run_wait_result_script_by_path_internal(
     args: Option<serde_json::Map<String, serde_json::Value>>,
     raw_string: Option<String>,
 ) -> Result<Json<serde_json::Value>, Error> {
-    check_queue_too_long(db, QUEUE_LIMIT_WAIT_RESULT.or(run_query.queue_limit)).await?;
+    check_queue_too_long(&db, QUEUE_LIMIT_WAIT_RESULT.or(run_query.queue_limit)).await?;
     let script_path = script_path.to_path();
     check_scopes(&authed, || format!("run:script/{script_path}"))?;
 
-    let mut tx: QueueTransaction<'_, _> = (rsmq, user_db.clone().begin(&authed).await?).into();
-    let (job_payload, tag) =
-        script_path_to_payload(script_path, tx.transaction_mut(), &w_id).await?;
+    let (job_payload, tag) = script_path_to_payload(script_path, &db, &w_id).await?;
 
     let args = run_query.add_include_headers(headers, args.unwrap_or_default());
     let args = add_raw_string(raw_string, args);
     check_tag_available_for_workspace(&w_id, &tag)?;
+    let tx = PushIsolationLevel::Isolated(user_db.clone(), authed.clone().into(), rsmq);
 
     let (uuid, tx) = push(
+        &db,
         tx,
         &w_id,
         job_payload,
@@ -2007,7 +2014,7 @@ async fn run_wait_result_script_by_path_internal(
 }
 
 pub async fn run_wait_result_script_by_hash(
-    authed: Authed,
+    authed: ApiAuthed,
     Extension(user_db): Extension<UserDB>,
     Extension(rsmq): Extension<Option<rsmq_async::MultiplexedRsmq>>,
     Extension(db): Extension<DB>,
@@ -2016,19 +2023,20 @@ pub async fn run_wait_result_script_by_hash(
     headers: HeaderMap,
     JsonOrForm(args, raw_string): JsonOrForm,
 ) -> error::JsonResult<serde_json::Value> {
-    check_queue_too_long(db, run_query.queue_limit).await?;
+    check_queue_too_long(&db, run_query.queue_limit).await?;
 
     let hash = script_hash.0;
-    let mut tx: QueueTransaction<'_, _> = (rsmq, user_db.clone().begin(&authed).await?).into();
     let (path, tag, concurrent_limit, concurrency_time_window_s) =
-        get_path_tag_and_limits_for_hash(tx.transaction_mut(), &w_id, hash).await?;
+        get_path_tag_and_limits_for_hash(&db, &w_id, hash).await?;
     check_scopes(&authed, || format!("run:script/{path}"))?;
 
     let args = run_query.add_include_headers(headers, args.unwrap_or_default());
     let args = add_raw_string(raw_string, args);
     check_tag_available_for_workspace(&w_id, &tag)?;
+    let tx = PushIsolationLevel::Isolated(user_db.clone(), authed.clone().into(), rsmq);
 
     let (uuid, tx) = push(
+        &db,
         tx,
         &w_id,
         JobPayload::ScriptHash {
@@ -2068,7 +2076,7 @@ pub async fn run_wait_result_script_by_hash(
 }
 
 pub async fn openai_sync_flow_by_path(
-    authed: Authed,
+    authed: ApiAuthed,
     Extension(user_db): Extension<UserDB>,
     Extension(rsmq): Extension<Option<rsmq_async::MultiplexedRsmq>>,
     Extension(db): Extension<DB>,
@@ -2093,7 +2101,7 @@ pub async fn openai_sync_flow_by_path(
 }
 
 pub async fn run_wait_result_flow_by_path(
-    authed: Authed,
+    authed: ApiAuthed,
     Extension(user_db): Extension<UserDB>,
     Extension(rsmq): Extension<Option<rsmq_async::MultiplexedRsmq>>,
     Extension(db): Extension<DB>,
@@ -2112,7 +2120,7 @@ async fn run_wait_result_flow_by_path_internal(
     db: sqlx::Pool<Postgres>,
     run_query: RunJobQuery,
     flow_path: StripPath,
-    authed: Authed,
+    authed: ApiAuthed,
     rsmq: Option<rsmq_async::MultiplexedRsmq>,
     user_db: UserDB,
     headers: HeaderMap,
@@ -2120,13 +2128,12 @@ async fn run_wait_result_flow_by_path_internal(
     raw_string: Option<String>,
     w_id: String,
 ) -> Result<Json<serde_json::Value>, Error> {
-    check_queue_too_long(db, run_query.queue_limit).await?;
+    check_queue_too_long(&db, run_query.queue_limit).await?;
 
     let flow_path = flow_path.to_path();
     check_scopes(&authed, || format!("run:flow/{flow_path}"))?;
 
-    let mut tx: QueueTransaction<'_, _> = (rsmq, user_db.clone().begin(&authed).await?).into();
-    let scheduled_for = run_query.get_scheduled_for(tx.transaction_mut()).await?;
+    let scheduled_for = run_query.get_scheduled_for(&db).await?;
     let args = run_query.add_include_headers(headers, args.unwrap_or_default());
     let args = add_raw_string(raw_string, args);
     let tag = sqlx::query_scalar!(
@@ -2134,12 +2141,14 @@ async fn run_wait_result_flow_by_path_internal(
         flow_path,
         w_id
     )
-    .fetch_optional(&mut tx)
+    .fetch_optional(&db)
     .await?
     .flatten();
     check_tag_available_for_workspace(&w_id, &tag)?;
+    let tx = PushIsolationLevel::Isolated(user_db.clone(), authed.clone().into(), rsmq);
 
     let (uuid, tx) = push(
+        &db,
         tx,
         &w_id,
         JobPayload::Flow(flow_path.to_string()),
@@ -2174,7 +2183,8 @@ async fn run_wait_result_flow_by_path_internal(
 }
 
 async fn run_preview_job(
-    authed: Authed,
+    authed: ApiAuthed,
+    Extension(db): Extension<DB>,
     Extension(user_db): Extension<UserDB>,
     Extension(rsmq): Extension<Option<rsmq_async::MultiplexedRsmq>>,
     Path(w_id): Path<String>,
@@ -2188,12 +2198,13 @@ async fn run_preview_job(
             "Operators cannot run preview jobs for security reasons".to_string(),
         ));
     }
-    let mut tx: QueueTransaction<'_, _> = (rsmq, user_db.begin(&authed).await?).into();
-    let scheduled_for = run_query.get_scheduled_for(tx.transaction_mut()).await?;
+    let scheduled_for = run_query.get_scheduled_for(&db).await?;
     let args = run_query.add_include_headers(headers, preview.args.unwrap_or_default());
     check_tag_available_for_workspace(&w_id, &preview.tag)?;
+    let tx = PushIsolationLevel::Isolated(user_db.clone(), authed.clone().into(), rsmq);
 
     let (uuid, tx) = push(
+        &db,
         tx,
         &w_id,
         match preview.kind {
@@ -2232,17 +2243,18 @@ async fn run_preview_job(
 }
 
 async fn add_noop_jobs(
-    authed: Authed,
+    authed: ApiAuthed,
     Extension(db): Extension<DB>,
     Extension(rsmq): Extension<Option<rsmq_async::MultiplexedRsmq>>,
     Path((w_id, n)): Path<(String, i32)>,
 ) -> error::JsonResult<Vec<String>> {
-    require_super_admin(&mut db.begin().await?, &authed.email).await?;
-    let mut tx: QueueTransaction<'_, _> = (rsmq, db.begin().await?).into();
+    require_super_admin(&db, &authed.email).await?;
+    let mut tx = PushIsolationLevel::IsolatedRoot(db.clone(), rsmq);
 
     let mut uuids: Vec<String> = Vec::new();
     for _ in 0..n {
         let (uuid, ntx) = push(
+            &db,
             tx,
             &w_id,
             JobPayload::Noop,
@@ -2264,15 +2276,21 @@ async fn add_noop_jobs(
             None,
         )
         .await?;
-        tx = ntx;
+        tx = PushIsolationLevel::Transaction(ntx);
         uuids.push(uuid.to_string());
     }
-    tx.commit().await?;
+    match tx {
+        PushIsolationLevel::Transaction(tx) => {
+            tx.commit().await?;
+        }
+        _ => (),
+    }
 
     Ok(Json(uuids))
 }
 async fn run_preview_flow_job(
-    authed: Authed,
+    authed: ApiAuthed,
+    Extension(db): Extension<DB>,
     Extension(user_db): Extension<UserDB>,
     Extension(rsmq): Extension<Option<rsmq_async::MultiplexedRsmq>>,
     Path(w_id): Path<String>,
@@ -2286,12 +2304,13 @@ async fn run_preview_flow_job(
             "Operators cannot run preview jobs for security reasons".to_string(),
         ));
     }
-    let mut tx: QueueTransaction<'_, _> = (rsmq, user_db.begin(&authed).await?).into();
-    let scheduled_for = run_query.get_scheduled_for(tx.transaction_mut()).await?;
+    let scheduled_for = run_query.get_scheduled_for(&db).await?;
     let args = run_query.add_include_headers(headers, raw_flow.args.unwrap_or_default());
     check_tag_available_for_workspace(&w_id, &raw_flow.tag)?;
+    let tx = PushIsolationLevel::Isolated(user_db.clone(), authed.clone().into(), rsmq);
 
     let (uuid, tx) = push(
+        &db,
         tx,
         &w_id,
         JobPayload::RawFlow { value: raw_flow.value, path: raw_flow.path },
@@ -2319,7 +2338,9 @@ async fn run_preview_flow_job(
 }
 
 pub async fn run_job_by_hash(
-    authed: Authed,
+    authed: ApiAuthed,
+    Extension(db): Extension<DB>,
+
     Extension(user_db): Extension<UserDB>,
     Extension(rsmq): Extension<Option<rsmq_async::MultiplexedRsmq>>,
     Path((w_id, script_hash)): Path<(String, ScriptHash)>,
@@ -2328,17 +2349,18 @@ pub async fn run_job_by_hash(
     JsonOrForm(args, raw_string): JsonOrForm,
 ) -> error::Result<(StatusCode, String)> {
     let hash = script_hash.0;
-    let mut tx: QueueTransaction<'_, _> = (rsmq, user_db.begin(&authed).await?).into();
     let (path, tag, concurrent_limit, concurrency_time_window_s) =
-        get_path_tag_and_limits_for_hash(tx.transaction_mut(), &w_id, hash).await?;
+        get_path_tag_and_limits_for_hash(&db, &w_id, hash).await?;
     check_scopes(&authed, || format!("run:script/{path}"))?;
 
-    let scheduled_for = run_query.get_scheduled_for(tx.transaction_mut()).await?;
+    let scheduled_for = run_query.get_scheduled_for(&db).await?;
     let args = run_query.add_include_headers(headers, args.unwrap_or_default());
     let args = add_raw_string(raw_string, args);
     check_tag_available_for_workspace(&w_id, &tag)?;
+    let tx = PushIsolationLevel::Isolated(user_db.clone(), authed.clone().into(), rsmq);
 
     let (uuid, tx) = push(
+        &db,
         tx,
         &w_id,
         JobPayload::ScriptHash {
@@ -2534,7 +2556,7 @@ pub struct ListCompletedQuery {
 }
 
 async fn list_completed_jobs(
-    authed: Authed,
+    authed: ApiAuthed,
     Extension(db): Extension<DB>,
     Path(w_id): Path<String>,
     Query(pagination): Query<Pagination>,
@@ -2647,7 +2669,7 @@ async fn get_completed_job_result_maybe(
 }
 
 async fn delete_completed_job(
-    authed: Authed,
+    authed: ApiAuthed,
     Extension(user_db): Extension<UserDB>,
     Path((w_id, id)): Path<(String, Uuid)>,
 ) -> error::JsonResult<CompletedJob> {

backend/windmill-api/src/lib.rs

@@ -6,16 +6,16 @@
  * LICENSE-AGPL for a copy of the license.
  */
 
+use crate::db::ApiAuthed;
 use crate::oauth2::AllClients;
 use crate::saml::{SamlSsoLogin, ServiceProviderExt};
 use crate::scim::has_scim_token;
 use crate::tracing_init::MyOnFailure;
 use crate::workers::ALL_TAGS;
 use crate::{
-    db::UserDB,
     oauth2::{build_oauth_clients, SlackVerifier},
     tracing_init::{MyMakeSpan, MyOnResponse},
-    users::{Authed, OptAuthed},
+    users::OptAuthed,
     webhook_util::WebhookShared,
 };
 use anyhow::Context;
@@ -34,6 +34,7 @@ use tower_http::{
     cors::{Any, CorsLayer},
     trace::TraceLayer,
 };
+use windmill_common::db::UserDB;
 use windmill_common::utils::rd_string;
 
 use windmill_common::error::AppError;
@@ -239,7 +240,7 @@ pub async fn run_server(
                 .nest("/flows", flows::global_service())
                 .nest("/apps", apps::global_service().layer(cors.clone()))
                 .nest("/schedules", schedule::global_service())
-                .route_layer(from_extractor::<Authed>())
+                .route_layer(from_extractor::<ApiAuthed>())
                 .route_layer(from_extractor::<users::Tokened>())
                 .nest(
                     "/saml",

backend/windmill-api/src/oauth2.rs

@@ -33,15 +33,17 @@ use serde::{de::DeserializeOwned, Deserialize, Serialize};
 use sqlx::{Postgres, Transaction};
 use tower_cookies::{Cookie, Cookies};
 use windmill_audit::{audit_log, ActionKind};
+use windmill_common::db::UserDB;
 use windmill_common::jobs::JobPayload;
 use windmill_common::users::username_to_permissioned_as;
 use windmill_common::utils::{not_found_if_none, now_from_db};
 
+use crate::db::ApiAuthed;
 use crate::saml::SamlSsoLogin;
-use crate::users::{login_externally, Authed, LoginUserInfo};
+use crate::users::{login_externally, LoginUserInfo};
 use crate::webhook_util::{InstanceEvent, WebhookShared};
 use crate::{
-    db::{UserDB, DB},
+    db::DB,
     variables::{build_crypt, encrypt},
     workspaces::WorkspaceSettings,
 };
@@ -49,7 +51,7 @@ use crate::{BASE_URL, HTTP_CLIENT, IS_SECURE, OAUTH_CLIENTS, SLACK_SIGNING_SECRE
 use windmill_common::error::{self, to_anyhow, Error};
 use windmill_common::oauth2::*;
 
-use windmill_queue::QueueTransaction;
+use windmill_queue::PushIsolationLevel;
 
 use std::{fs, str};
 
@@ -340,7 +342,7 @@ struct CreateAccount {
     expires_in: i64,
 }
 async fn create_account(
-    authed: Authed,
+    authed: ApiAuthed,
     Extension(user_db): Extension<UserDB>,
     Path(w_id): Path<String>,
     Json(payload): Json<CreateAccount>,
@@ -364,7 +366,7 @@ async fn create_account(
 }
 
 async fn delete_account(
-    authed: Authed,
+    authed: ApiAuthed,
     Extension(user_db): Extension<UserDB>,
     Query((w_id, id)): Query<(String, i32)>,
 ) -> error::Result<String> {
@@ -455,7 +457,7 @@ async fn connect_slack(cookies: Cookies) -> error::Result<Redirect> {
 }
 
 async fn disconnect(
-    authed: Authed,
+    authed: ApiAuthed,
     Path((w_id, id)): Path<(String, i32)>,
     Extension(user_db): Extension<UserDB>,
 ) -> error::Result<String> {
@@ -474,7 +476,7 @@ async fn disconnect(
 }
 
 async fn disconnect_slack(
-    authed: Authed,
+    authed: ApiAuthed,
     Path(w_id): Path<String>,
     Extension(user_db): Extension<UserDB>,
 ) -> error::Result<String> {
@@ -502,7 +504,7 @@ struct VariablePath {
     path: String,
 }
 async fn refresh_token(
-    authed: Authed,
+    authed: ApiAuthed,
     Path((w_id, id)): Path<(String, i32)>,
     Extension(user_db): Extension<UserDB>,
     Json(VariablePath { path }): Json<VariablePath>,
@@ -635,7 +637,7 @@ async fn connect_callback(
 
 async fn connect_slack_callback(
     Path(w_id): Path<String>,
-    authed: Authed,
+    authed: ApiAuthed,
     cookies: Cookies,
     Extension(user_db): Extension<UserDB>,
     Json(callback): Json<OAuthCallback>,
@@ -783,13 +785,12 @@ async fn slack_command(
         }
     }
 
-    let mut tx: QueueTransaction<'_, _> = (rsmq, db.begin().await?).into();
     let settings = sqlx::query_as!(
         WorkspaceSettings,
         "SELECT * FROM workspace_settings WHERE slack_team_id = $1",
         form.team_id,
     )
-    .fetch_optional(&mut tx)
+    .fetch_optional(&db)
     .await?;
 
     if let Some(settings) = settings {
@@ -800,7 +801,7 @@ async fn slack_command(
                 let path = path.strip_prefix("script/").unwrap_or_else(|| path);
                 let (script_hash, tag, concurrent_limit, concurrency_time_window_s) =
                     windmill_common::get_latest_deployed_hash_for_path(
-                        tx.transaction_mut(),
+                        &db,
                         &settings.workspace_id,
                         path,
                     )
@@ -821,8 +822,10 @@ async fn slack_command(
                 "response_url".to_string(),
                 serde_json::Value::String(form.response_url),
             );
+            let tx = PushIsolationLevel::IsolatedRoot(db.clone(), rsmq);
 
             let (uuid, tx) = windmill_queue::push(
+                &db,
                 tx,
                 &settings.workspace_id,
                 payload,
@@ -852,7 +855,6 @@ async fn slack_command(
             ));
         }
     }
-    tx.commit().await?;
 
     return Ok(format!(
         "workspace not properly configured (did you set the script to trigger in the settings?)"

backend/windmill-api/src/openai.rs

@@ -1,4 +1,8 @@
-use crate::{db::DB, users::Authed, variables::build_crypt, HTTP_CLIENT};
+use crate::{
+    db::{ApiAuthed, DB},
+    variables::build_crypt,
+    HTTP_CLIENT,
+};
 
 use axum::{
     body::{Bytes, StreamBody},
@@ -69,7 +73,7 @@ lazy_static::lazy_static! {
 }
 
 async fn proxy(
-    authed: Authed,
+    authed: ApiAuthed,
     Extension(db): Extension<DB>,
     Path((w_id, openai_path)): Path<(String, String)>,
     body: Bytes,

backend/windmill-api/src/raw_apps.rs

@@ -6,8 +6,8 @@
  * LICENSE-AGPL for a copy of the license.
  */
 use crate::{
-    db::{UserDB, DB},
-    users::{require_owner_of_path, Authed},
+    db::{ApiAuthed, DB},
+    users::require_owner_of_path,
     webhook_util::{WebhookMessage, WebhookShared},
 };
 use axum::{
@@ -26,6 +26,7 @@ use std::str;
 use windmill_audit::{audit_log, ActionKind};
 use windmill_common::{
     apps::ListAppQuery,
+    db::UserDB,
     error::{Error, JsonResult, Result},
     utils::{not_found_if_none, paginate, Pagination, StripPath},
 };
@@ -66,7 +67,7 @@ pub struct EditApp {
 }
 
 async fn list_apps(
-    authed: Authed,
+    authed: ApiAuthed,
     Extension(user_db): Extension<UserDB>,
     Path(w_id): Path<String>,
     Query(pagination): Query<Pagination>,
@@ -113,7 +114,7 @@ async fn list_apps(
 }
 
 async fn get_data(
-    authed: Authed,
+    authed: ApiAuthed,
     Extension(user_db): Extension<UserDB>,
     Path((w_id, _version, path)): Path<(String, u16, StripPath)>,
 ) -> Result<Response> {
@@ -139,7 +140,7 @@ async fn get_data(
 }
 
 async fn create_app(
-    authed: Authed,
+    authed: ApiAuthed,
     Extension(user_db): Extension<UserDB>,
     Extension(webhook): Extension<WebhookShared>,
     Path(w_id): Path<String>,
@@ -199,7 +200,7 @@ async fn create_app(
 }
 
 async fn delete_app(
-    authed: Authed,
+    authed: ApiAuthed,
     Extension(user_db): Extension<UserDB>,
     Extension(webhook): Extension<WebhookShared>,
     Path((w_id, path)): Path<(String, StripPath)>,
@@ -234,7 +235,7 @@ async fn delete_app(
 }
 
 async fn update_app(
-    authed: Authed,
+    authed: ApiAuthed,
     Extension(user_db): Extension<UserDB>,
     Extension(webhook): Extension<WebhookShared>,
     Path((w_id, path)): Path<(String, StripPath)>,

backend/windmill-api/src/resources.rs

@@ -7,8 +7,8 @@
  */
 
 use crate::{
-    db::{UserDB, DB},
-    users::{maybe_refresh_folders, require_owner_of_path, Authed},
+    db::{ApiAuthed, DB},
+    users::{maybe_refresh_folders, require_owner_of_path},
     webhook_util::{WebhookMessage, WebhookShared},
 };
 use axum::{
@@ -23,6 +23,7 @@ use sql_builder::{bind::Bind, SqlBuilder};
 use sqlx::{FromRow, Postgres, Transaction};
 use windmill_audit::{audit_log, ActionKind};
 use windmill_common::{
+    db::UserDB,
     error::{Error, JsonResult, Result},
     utils::{not_found_if_none, paginate, require_admin, Pagination, StripPath},
 };
@@ -117,7 +118,7 @@ pub struct ListResourceQuery {
     resource_type_exclude: Option<String>,
 }
 async fn list_resources(
-    authed: Authed,
+    authed: ApiAuthed,
     Query(lq): Query<ListResourceQuery>,
     Query(pagination): Query<Pagination>,
     Extension(user_db): Extension<UserDB>,
@@ -175,7 +176,7 @@ async fn list_resources(
 }
 
 async fn get_resource(
-    authed: Authed,
+    authed: ApiAuthed,
     Extension(user_db): Extension<UserDB>,
     Path((w_id, path)): Path<(String, StripPath)>,
 ) -> JsonResult<ListableResource> {
@@ -223,7 +224,7 @@ async fn exists_resource(
 }
 
 async fn get_resource_value(
-    authed: Authed,
+    authed: ApiAuthed,
     Extension(user_db): Extension<UserDB>,
     Path((w_id, path)): Path<(String, StripPath)>,
 ) -> JsonResult<Option<serde_json::Value>> {
@@ -244,7 +245,7 @@ async fn get_resource_value(
 }
 
 async fn get_resource_value_interpolated(
-    authed: Authed,
+    authed: ApiAuthed,
     Extension(user_db): Extension<UserDB>,
     Path((w_id, path)): Path<(String, StripPath)>,
 ) -> JsonResult<Option<serde_json::Value>> {
@@ -274,7 +275,7 @@ use async_recursion::async_recursion;
 
 #[async_recursion]
 pub async fn transform_json_value<'c>(
-    authed: &Authed,
+    authed: &ApiAuthed,
     user_db: &UserDB,
     workspace: &str,
     v: Value,
@@ -282,7 +283,7 @@ pub async fn transform_json_value<'c>(
     match v {
         Value::String(y) if y.starts_with("$var:") => {
             let path = y.strip_prefix("$var:").unwrap();
-            let tx: Transaction<'_, Postgres> = user_db.clone().begin(&authed).await?;
+            let tx: Transaction<'_, Postgres> = user_db.clone().begin(authed).await?;
             let v =
                 crate::variables::get_value_internal(tx, workspace, path, &authed.username).await?;
             Ok(Value::String(v))
@@ -292,7 +293,7 @@ pub async fn transform_json_value<'c>(
             if path.split("/").count() < 2 {
                 return Err(Error::InternalErr(format!("Invalid resource path: {path}")));
             }
-            let mut tx: Transaction<'_, Postgres> = user_db.clone().begin(&authed).await?;
+            let mut tx: Transaction<'_, Postgres> = user_db.clone().begin(authed).await?;
             let v = sqlx::query_scalar!(
                 "SELECT value from resource WHERE path = $1 AND workspace_id = $2",
                 path,
@@ -348,7 +349,7 @@ struct CreateResourceQuery {
     update_if_exists: Option<bool>,
 }
 async fn create_resource(
-    authed: Authed,
+    authed: ApiAuthed,
     Extension(user_db): Extension<UserDB>,
     Extension(webhook): Extension<WebhookShared>,
     Extension(db): Extension<DB>,
@@ -402,7 +403,7 @@ async fn create_resource(
 }
 
 async fn delete_resource(
-    authed: Authed,
+    authed: ApiAuthed,
     Extension(user_db): Extension<UserDB>,
     Extension(webhook): Extension<WebhookShared>,
     Path((w_id, path)): Path<(String, StripPath)>,
@@ -445,7 +446,7 @@ async fn delete_resource(
 }
 
 async fn update_resource(
-    authed: Authed,
+    authed: ApiAuthed,
     Extension(user_db): Extension<UserDB>,
     Extension(webhook): Extension<WebhookShared>,
     Extension(db): Extension<DB>,
@@ -527,7 +528,7 @@ struct UpdateResource {
 }
 
 async fn update_resource_value(
-    authed: Authed,
+    authed: ApiAuthed,
     Extension(user_db): Extension<UserDB>,
     Extension(webhook): Extension<WebhookShared>,
     Path((w_id, path)): Path<(String, StripPath)>,
@@ -599,7 +600,7 @@ async fn list_resource_types_names(
 }
 
 async fn get_resource_type(
-    authed: Authed,
+    authed: ApiAuthed,
     Extension(user_db): Extension<UserDB>,
     Path((w_id, name)): Path<(String, String)>,
 ) -> JsonResult<ResourceType> {
@@ -636,7 +637,7 @@ async fn exists_resource_type(
 }
 
 async fn create_resource_type(
-    authed: Authed,
+    authed: ApiAuthed,
     Extension(user_db): Extension<UserDB>,
     Extension(webhook): Extension<WebhookShared>,
     Path(w_id): Path<String>,
@@ -703,7 +704,7 @@ async fn check_rt_path_conflict<'c>(
 }
 
 async fn delete_resource_type(
-    authed: Authed,
+    authed: ApiAuthed,
     Extension(user_db): Extension<UserDB>,
     Extension(webhook): Extension<WebhookShared>,
     Path((w_id, name)): Path<(String, String)>,
@@ -739,7 +740,7 @@ async fn delete_resource_type(
 }
 
 async fn update_resource_type(
-    authed: Authed,
+    authed: ApiAuthed,
     Extension(user_db): Extension<UserDB>,
     Extension(webhook): Extension<WebhookShared>,
     Path((w_id, name)): Path<(String, String)>,

backend/windmill-api/src/schedule.rs

@@ -7,8 +7,8 @@
  */
 
 use crate::{
-    db::{UserDB, DB},
-    users::{maybe_refresh_folders, Authed},
+    db::{ApiAuthed, DB},
+    users::maybe_refresh_folders,
 };
 use axum::{
     extract::{Extension, Path, Query},
@@ -22,6 +22,7 @@ use sqlx::{Postgres, Transaction};
 use std::str::FromStr;
 use windmill_audit::{audit_log, ActionKind};
 use windmill_common::{
+    db::UserDB,
     error::{Error, JsonResult, Result},
     jobs::JobKind,
     schedule::Schedule,
@@ -80,7 +81,7 @@ async fn check_path_conflict<'c>(
 }
 
 async fn create_schedule(
-    authed: Authed,
+    authed: ApiAuthed,
     Extension(db): Extension<DB>,
     Extension(user_db): Extension<UserDB>,
     Extension(rsmq): Extension<Option<rsmq_async::MultiplexedRsmq>>,
@@ -141,7 +142,7 @@ async fn create_schedule(
     .await?;
 
     if ns.enabled.unwrap_or(true) {
-        tx = push_scheduled_job(tx, schedule).await?
+        tx = push_scheduled_job(&db, tx, schedule).await?
     }
     tx.commit().await?;
 
@@ -149,7 +150,7 @@ async fn create_schedule(
 }
 
 async fn edit_schedule(
-    authed: Authed,
+    authed: ApiAuthed,
     Extension(db): Extension<DB>,
     Extension(user_db): Extension<UserDB>,
     Extension(rsmq): Extension<Option<rsmq_async::MultiplexedRsmq>>,
@@ -207,7 +208,7 @@ async fn edit_schedule(
     .await?;
 
     if schedule.enabled {
-        tx = push_scheduled_job(tx, schedule).await?;
+        tx = push_scheduled_job(&db, tx, schedule).await?;
     }
     tx.commit().await?;
 
@@ -223,7 +224,7 @@ pub struct ListScheduleQuery {
 }
 
 async fn list_schedule(
-    authed: Authed,
+    authed: ApiAuthed,
     Extension(user_db): Extension<UserDB>,
     Path(w_id): Path<String>,
     Query(lsq): Query<ListScheduleQuery>,
@@ -271,7 +272,7 @@ pub struct ScheduleWJobs {
 }
 
 async fn list_schedule_with_jobs(
-    authed: Authed,
+    authed: ApiAuthed,
     Extension(user_db): Extension<UserDB>,
     Path(w_id): Path<String>,
     Query(pagination): Query<Pagination>,
@@ -303,7 +304,7 @@ async fn list_schedule_with_jobs(
 //    ) t;
 
 async fn get_schedule(
-    authed: Authed,
+    authed: ApiAuthed,
     Extension(user_db): Extension<UserDB>,
     Path((w_id, path)): Path<(String, StripPath)>,
 ) -> JsonResult<Schedule> {
@@ -352,7 +353,8 @@ pub async fn preview_schedule(
 }
 
 pub async fn set_enabled(
-    authed: Authed,
+    authed: ApiAuthed,
+    Extension(db): Extension<DB>,
     Extension(user_db): Extension<UserDB>,
     Extension(rsmq): Extension<Option<rsmq_async::MultiplexedRsmq>>,
     Path((w_id, path)): Path<(String, StripPath)>,
@@ -388,7 +390,7 @@ pub async fn set_enabled(
     .await?;
 
     if payload.enabled {
-        tx = push_scheduled_job(tx, schedule).await?;
+        tx = push_scheduled_job(&db, tx, schedule).await?;
     }
     tx.commit().await?;
 
@@ -399,7 +401,7 @@ pub async fn set_enabled(
 }
 
 async fn delete_schedule(
-    authed: Authed,
+    authed: ApiAuthed,
     Extension(user_db): Extension<UserDB>,
     Path((w_id, path)): Path<(String, StripPath)>,
 ) -> Result<String> {

backend/windmill-api/src/scripts.rs

@@ -7,9 +7,9 @@
  */
 
 use crate::{
-    db::{UserDB, DB},
+    db::{ApiAuthed, DB},
     schedule::clear_schedule,
-    users::{maybe_refresh_folders, require_owner_of_path, AuthCache, Authed},
+    users::{maybe_refresh_folders, require_owner_of_path, AuthCache},
     webhook_util::{WebhookMessage, WebhookShared},
     HTTP_CLIENT,
 };
@@ -31,6 +31,7 @@ use std::{
 };
 use windmill_audit::{audit_log, ActionKind};
 use windmill_common::{
+    db::UserDB,
     error::{Error, JsonResult, Result},
     jobs::JobPayload,
     schedule::Schedule,
@@ -43,7 +44,7 @@ use windmill_common::{
         list_elems_from_hub, not_found_if_none, paginate, require_admin, Pagination, StripPath,
     },
 };
-use windmill_queue::{self, schedule::push_scheduled_job, QueueTransaction};
+use windmill_queue::{self, schedule::push_scheduled_job, PushIsolationLevel, QueueTransaction};
 
 const MAX_HASH_HISTORY_LENGTH_STORED: usize = 20;
 
@@ -103,7 +104,7 @@ pub fn workspaced_service() -> Router {
 }
 
 async fn list_scripts(
-    authed: Authed,
+    authed: ApiAuthed,
     Extension(user_db): Extension<UserDB>,
     Path(w_id): Path<String>,
     Query(pagination): Query<Pagination>,
@@ -192,7 +193,7 @@ async fn list_scripts(
     Ok(Json(rows))
 }
 
-async fn list_hub_scripts(Authed { email, .. }: Authed) -> JsonResult<serde_json::Value> {
+async fn list_hub_scripts(ApiAuthed { email, .. }: ApiAuthed) -> JsonResult<serde_json::Value> {
     let asks = list_elems_from_hub(
         &HTTP_CLIENT,
         "https://hub.windmill.dev/searchData?approved=true",
@@ -209,7 +210,7 @@ fn hash_script(ns: &NewScript) -> i64 {
 }
 
 async fn create_script(
-    authed: Authed,
+    authed: ApiAuthed,
     Extension(user_db): Extension<UserDB>,
     Extension(rsmq): Extension<Option<rsmq_async::MultiplexedRsmq>>,
     Extension(webhook): Extension<WebhookShared>,
@@ -429,7 +430,7 @@ async fn create_script(
             clear_schedule(tx.transaction_mut(), &schedule.path, false, &w_id).await?;
 
             if schedule.enabled {
-                tx = push_scheduled_job(tx, schedule).await?;
+                tx = push_scheduled_job(&db, tx, schedule).await?;
             }
         }
     } else {
@@ -488,6 +489,8 @@ async fn create_script(
         );
     }
 
+    let mut tx = PushIsolationLevel::Transaction(tx);
+
     if needs_lock_gen {
         let dependencies = match ns.language {
             ScriptLang::Python3 => {
@@ -498,6 +501,7 @@ async fn create_script(
             _ => ns.content,
         };
         let (_, new_tx) = windmill_queue::push(
+            &db,
             tx,
             &w_id,
             JobPayload::Dependencies { hash, dependencies, language: ns.language, path: ns.path },
@@ -519,20 +523,30 @@ async fn create_script(
             None,
         )
         .await?;
-        tx = new_tx;
+        tx = PushIsolationLevel::Transaction(new_tx);
     }
 
-    tx.commit().await?;
+    match tx {
+        PushIsolationLevel::Transaction(tx) => tx.commit().await?,
+        _ => {
+            return Err(Error::InternalErr(
+                "Expected a transaction here".to_string(),
+            ));
+        }
+    }
 
     Ok((StatusCode::CREATED, format!("{}", hash)))
 }
 
-pub async fn get_hub_script_by_path(authed: Authed, Path(path): Path<StripPath>) -> Result<String> {
+pub async fn get_hub_script_by_path(
+    authed: ApiAuthed,
+    Path(path): Path<StripPath>,
+) -> Result<String> {
     windmill_common::scripts::get_hub_script_by_path(&authed.email, path, &HTTP_CLIENT).await
 }
 
 pub async fn get_full_hub_script_by_path(
-    Authed { email, .. }: Authed,
+    ApiAuthed { email, .. }: ApiAuthed,
     Path(path): Path<StripPath>,
 ) -> JsonResult<HubScript> {
     Ok(Json(
@@ -541,7 +555,7 @@ pub async fn get_full_hub_script_by_path(
 }
 
 async fn get_script_by_path(
-    authed: Authed,
+    authed: ApiAuthed,
     Extension(user_db): Extension<UserDB>,
     Path((w_id, path)): Path<(String, StripPath)>,
 ) -> JsonResult<Script> {
@@ -564,7 +578,7 @@ async fn get_script_by_path(
 }
 
 async fn get_script_by_path_w_draft(
-    authed: Authed,
+    authed: ApiAuthed,
     Extension(user_db): Extension<UserDB>,
     Path((w_id, path)): Path<(String, StripPath)>,
 ) -> JsonResult<ScriptWDraft> {
@@ -589,7 +603,7 @@ async fn get_script_by_path_w_draft(
 }
 
 async fn list_paths(
-    authed: Authed,
+    authed: ApiAuthed,
     Extension(user_db): Extension<UserDB>,
     Path(w_id): Path<String>,
 ) -> JsonResult<Vec<String>> {
@@ -619,7 +633,7 @@ async fn get_tokened_raw_script_by_path(
 }
 
 async fn raw_script_by_path(
-    authed: Authed,
+    authed: ApiAuthed,
     Extension(user_db): Extension<UserDB>,
     Path((w_id, path)): Path<(String, StripPath)>,
 ) -> Result<String> {
@@ -742,7 +756,7 @@ async fn get_deployment_status(
     Ok(Json(status))
 }
 
-pub async fn require_is_writer(authed: &Authed, path: &str, w_id: &str, db: DB) -> Result<()> {
+pub async fn require_is_writer(authed: &ApiAuthed, path: &str, w_id: &str, db: DB) -> Result<()> {
     return crate::users::require_is_writer(
         authed,
         path,
@@ -757,7 +771,7 @@ pub async fn require_is_writer(authed: &Authed, path: &str, w_id: &str, db: DB)
 }
 
 async fn archive_script_by_path(
-    authed: Authed,
+    authed: ApiAuthed,
     Extension(webhook): Extension<WebhookShared>,
     Extension(user_db): Extension<UserDB>,
     Extension(db): Extension<DB>,
@@ -796,7 +810,7 @@ async fn archive_script_by_path(
 }
 
 async fn archive_script_by_hash(
-    authed: Authed,
+    authed: ApiAuthed,
     Extension(user_db): Extension<UserDB>,
     Extension(webhook): Extension<WebhookShared>,
     Path((w_id, hash)): Path<(String, ScriptHash)>,
@@ -832,7 +846,7 @@ async fn archive_script_by_hash(
 }
 
 async fn delete_script_by_hash(
-    authed: Authed,
+    authed: ApiAuthed,
     Extension(user_db): Extension<UserDB>,
     Extension(webhook): Extension<WebhookShared>,
     Extension(db): Extension<DB>,
@@ -872,7 +886,7 @@ async fn delete_script_by_hash(
 }
 
 async fn delete_script_by_path(
-    authed: Authed,
+    authed: ApiAuthed,
     Extension(user_db): Extension<UserDB>,
     Extension(webhook): Extension<WebhookShared>,
     Extension(db): Extension<DB>,

backend/windmill-api/src/utils.rs

@@ -6,21 +6,18 @@
  * LICENSE-AGPL for a copy of the license.
  */
 
-use sqlx::{Postgres, Transaction};
 use windmill_common::{
     error::{self, Error},
     users::SUPERADMIN_SECRET_EMAIL,
+    DB,
 };
 
-pub async fn require_super_admin<'c>(
-    db: &mut Transaction<'c, Postgres>,
-    email: &str,
-) -> error::Result<()> {
+pub async fn require_super_admin(db: &DB, email: &str) -> error::Result<()> {
     if email == SUPERADMIN_SECRET_EMAIL {
         return Ok(());
     }
     let is_admin = sqlx::query_scalar!("SELECT super_admin FROM password WHERE email = $1", email)
-        .fetch_optional(&mut **db)
+        .fetch_optional(db)
         .await
         .map_err(|e| Error::InternalErr(format!("fetching super admin: {e}")))?
         .unwrap_or(false);

backend/windmill-api/src/variables.rs

@@ -7,9 +7,9 @@
  */
 
 use crate::{
-    db::{UserDB, DB},
+    db::{ApiAuthed, DB},
     oauth2::_refresh_token,
-    users::{maybe_refresh_folders, require_owner_of_path, Authed},
+    users::{maybe_refresh_folders, require_owner_of_path},
     webhook_util::{WebhookMessage, WebhookShared},
 };
 
@@ -22,6 +22,7 @@ use hyper::StatusCode;
 use serde_json::Value;
 use windmill_audit::{audit_log, ActionKind};
 use windmill_common::{
+    db::UserDB,
     error::{Error, JsonResult, Result},
     utils::{not_found_if_none, StripPath},
     variables::{get_reserved_variables, ContextualVariable, CreateVariable, ListableVariable},
@@ -51,7 +52,7 @@ pub fn workspaced_service() -> Router {
 
 async fn list_contextual_variables(
     Path(w_id): Path<String>,
-    Authed { username, email, .. }: Authed,
+    ApiAuthed { username, email, .. }: ApiAuthed,
 ) -> JsonResult<Vec<ContextualVariable>> {
     Ok(Json(
         get_reserved_variables(
@@ -72,7 +73,7 @@ async fn list_contextual_variables(
 }
 
 async fn list_variables(
-    authed: Authed,
+    authed: ApiAuthed,
     Extension(user_db): Extension<UserDB>,
     Path(w_id): Path<String>,
 ) -> JsonResult<Vec<ListableVariable>> {
@@ -103,7 +104,7 @@ struct GetVariableQuery {
 }
 
 async fn get_variable(
-    authed: Authed,
+    authed: ApiAuthed,
     Extension(user_db): Extension<UserDB>,
     Query(q): Query<GetVariableQuery>,
     Path((w_id, path)): Path<(String, StripPath)>,
@@ -166,7 +167,7 @@ async fn get_variable(
 }
 
 async fn get_value(
-    authed: Authed,
+    authed: ApiAuthed,
     Extension(user_db): Extension<UserDB>,
     Path((w_id, path)): Path<(String, StripPath)>,
 ) -> JsonResult<String> {
@@ -263,7 +264,7 @@ async fn check_path_conflict<'c>(
 }
 
 async fn create_variable(
-    authed: Authed,
+    authed: ApiAuthed,
     Extension(db): Extension<DB>,
     Extension(user_db): Extension<UserDB>,
     Extension(webhook): Extension<WebhookShared>,
@@ -338,7 +339,7 @@ async fn encrypt_value(
 }
 
 async fn delete_variable(
-    authed: Authed,
+    authed: ApiAuthed,
     Extension(user_db): Extension<UserDB>,
     Extension(webhook): Extension<WebhookShared>,
     Path((w_id, path)): Path<(String, StripPath)>,
@@ -395,7 +396,7 @@ struct AlreadyEncrypted {
 }
 
 async fn update_variable(
-    authed: Authed,
+    authed: ApiAuthed,
     Extension(user_db): Extension<UserDB>,
     Extension(webhook): Extension<WebhookShared>,
     Extension(db): Extension<DB>,

backend/windmill-api/src/workers.rs

@@ -6,7 +6,6 @@
  * LICENSE-AGPL for a copy of the license.
  */
 
-use crate::{db::UserDB, users::Authed};
 use axum::{
     extract::{Extension, Query},
     routing::get,
@@ -18,6 +17,7 @@ use regex::Regex;
 use serde::{Deserialize, Serialize};
 use sqlx::FromRow;
 use windmill_common::{
+    db::UserDB,
     error::JsonResult,
     utils::{paginate, Pagination},
 };
@@ -28,6 +28,8 @@ use std::sync::atomic::Ordering;
 #[cfg(feature = "benchmark")]
 use windmill_queue::IDLE_WORKERS;
 
+use crate::db::ApiAuthed;
+
 #[cfg(not(feature = "benchmark"))]
 pub fn global_service() -> Router {
     Router::new()
@@ -92,7 +94,7 @@ struct EnableWorkerQuery {
 }
 
 async fn list_worker_pings(
-    authed: Authed,
+    authed: ApiAuthed,
     Extension(user_db): Extension<UserDB>,
     Query(pagination): Query<Pagination>,
 ) -> JsonResult<Vec<WorkerPing>> {

backend/windmill-api/src/workspaces.rs

@@ -10,12 +10,13 @@
 use std::str::FromStr;
 
 use crate::BASE_URL;
+use crate::db::ApiAuthed;
 use crate::{
     apps::AppWithLastVersion,
-    db::{UserDB, DB},
+    db::DB,
     folders::Folder,
     resources::{Resource, ResourceType},
-    users::{Authed, WorkspaceInvite, VALID_USERNAME, send_email_if_possible},
+    users::{WorkspaceInvite, VALID_USERNAME, send_email_if_possible},
     utils::require_super_admin,
     variables::build_crypt,
     webhook_util::{InstanceEvent, WebhookShared}
@@ -34,6 +35,7 @@ use magic_crypt::MagicCryptTrait;
 #[cfg(feature = "enterprise")]
 use stripe::CustomerId;
 use windmill_audit::{audit_log, ActionKind};
+use windmill_common::db::UserDB;
 use windmill_common::schedule::Schedule;
 use windmill_common::users::username_to_permissioned_as;
 use windmill_common::{
@@ -220,7 +222,7 @@ pub struct EditErrorHandler {
 }
 
 async fn list_pending_invites(
-    authed: Authed,
+    authed: ApiAuthed,
     Extension(user_db): Extension<UserDB>,
     Path(w_id): Path<String>,
 ) -> JsonResult<Vec<WorkspaceInvite>> {
@@ -243,7 +245,7 @@ pub struct PremiumWorkspaceInfo {
     pub usage: Option<i32>,
 }
 async fn premium_info(
-    authed: Authed,
+    authed: ApiAuthed,
     Extension(db): Extension<DB>,
     Path(w_id): Path<String>,
 ) -> JsonResult<PremiumWorkspaceInfo> {
@@ -267,7 +269,7 @@ struct PlanQuery {
 
 #[cfg(feature = "enterprise")]
 async fn stripe_checkout(
-    authed: Authed,
+    authed: ApiAuthed,
     Path(w_id): Path<String>,
     Query(plan): Query<PlanQuery>,
 ) -> Result<Redirect> {
@@ -311,7 +313,7 @@ async fn stripe_checkout(
 
 #[cfg(feature = "enterprise")]
 async fn stripe_portal(
-    authed: Authed,
+    authed: ApiAuthed,
     Path(w_id): Path<String>,
     Extension(db): Extension<DB>,
 ) -> Result<Redirect> {
@@ -337,7 +339,7 @@ async fn stripe_portal(
 }
 
 // async fn stripe_usage(
-//     authed: Authed,
+//     authed: ApiAuthed,
 //     Path(w_id): Path<String>,
 //     Extension(db): Extension<DB>,
 //     Extension(base_url): Extension<Arc<BaseUrl>>,
@@ -381,7 +383,7 @@ async fn stripe_portal(
 // }
 
 async fn exists_workspace(
-    authed: Authed,
+    authed: ApiAuthed,
     Extension(user_db): Extension<UserDB>,
     Json(WorkspaceId { id }): Json<WorkspaceId>,
 ) -> JsonResult<bool> {
@@ -398,7 +400,7 @@ async fn exists_workspace(
 }
 
 async fn list_workspaces(
-    authed: Authed,
+    authed: ApiAuthed,
     Extension(user_db): Extension<UserDB>,
 ) -> JsonResult<Vec<Workspace>> {
     let mut tx = user_db.begin(&authed).await?;
@@ -415,7 +417,7 @@ async fn list_workspaces(
 }
 
 async fn get_settings(
-    authed: Authed,
+    authed: ApiAuthed,
     Path(w_id): Path<String>,
     Extension(user_db): Extension<UserDB>,
 ) -> JsonResult<WorkspaceSettings> {
@@ -438,7 +440,7 @@ struct DeployTo {
     deploy_to: Option<String>,
 }
 async fn get_deploy_to(
-    authed: Authed,
+    authed: ApiAuthed,
     Path(w_id): Path<String>,
     Extension(user_db): Extension<UserDB>,
 ) -> JsonResult<DeployTo> {
@@ -457,10 +459,10 @@ async fn get_deploy_to(
 }
 
 async fn edit_slack_command(
-    authed: Authed,
+    authed: ApiAuthed,
     Extension(db): Extension<DB>,
     Path(w_id): Path<String>,
-    Authed { is_admin, username, .. }: Authed,
+    ApiAuthed { is_admin, username, .. }: ApiAuthed,
     Json(es): Json<EditCommandScript>,
 ) -> Result<String> {
     require_admin(is_admin, &username)?;
@@ -499,10 +501,10 @@ async fn edit_slack_command(
 
 #[cfg(feature = "enterprise")]
 async fn edit_deploy_to(
-    authed: Authed,
+    authed: ApiAuthed,
     Extension(db): Extension<DB>,
     Path(w_id): Path<String>,
-    Authed { is_admin, username, .. }: Authed,
+    ApiAuthed { is_admin, username, .. }: ApiAuthed,
     Json(es): Json<EditDeployTo>,
 ) -> Result<String> {
     require_admin(is_admin, &username)?;
@@ -546,16 +548,16 @@ async fn edit_deploy_to() -> Result<String> {
 
 const BANNED_DOMAINS: &str = include_str!("../banned_domains.txt");
 
-async fn is_allowed_auto_domain(Authed { email, .. }: Authed) -> JsonResult<bool> {
+async fn is_allowed_auto_domain(ApiAuthed { email, .. }: ApiAuthed) -> JsonResult<bool> {
     let domain = email.split('@').last().unwrap();
     return Ok(Json(!BANNED_DOMAINS.contains(domain)));
 }
 
 async fn edit_auto_invite(
-    authed: Authed,
+    authed: ApiAuthed,
     Extension(db): Extension<DB>,
     Path(w_id): Path<String>,
-    Authed { is_admin, email, username, .. }: Authed,
+    ApiAuthed { is_admin, email, username, .. }: ApiAuthed,
     Json(ea): Json<EditAutoInvite>,
 ) -> Result<String> {
     require_admin(is_admin, &username)?;
@@ -620,10 +622,10 @@ async fn edit_auto_invite(
 }
 
 async fn edit_webhook(
-    authed: Authed,
+    authed: ApiAuthed,
     Extension(db): Extension<DB>,
     Path(w_id): Path<String>,
-    Authed { is_admin, username, .. }: Authed,
+    ApiAuthed { is_admin, username, .. }: ApiAuthed,
     Json(ew): Json<EditWebhook>,
 ) -> Result<String> {
     require_admin(is_admin, &username)?;
@@ -662,10 +664,10 @@ async fn edit_webhook(
 }
 
 async fn edit_openai_resource_path(
-    authed: Authed,
+    authed: ApiAuthed,
     Extension(db): Extension<DB>,
     Path(w_id): Path<String>,
-    Authed { is_admin, username, .. }: Authed,
+    ApiAuthed { is_admin, username, .. }: ApiAuthed,
     Json(eo): Json<EditOpenaiResourcePath>,
 ) -> Result<String> {
     require_admin(is_admin, &username)?;
@@ -726,10 +728,10 @@ async fn exists_openai_resource_path(
 
 
 async fn edit_error_handler(
-    authed: Authed,
+    authed: ApiAuthed,
     Extension(db): Extension<DB>,
     Path(w_id): Path<String>,
-    Authed { is_admin, username, .. }: Authed,
+    ApiAuthed { is_admin, username, .. }: ApiAuthed,
     Json(ee): Json<EditErrorHandler>,
 ) -> Result<String> {
     require_admin(is_admin, &username)?;
@@ -781,15 +783,16 @@ async fn edit_error_handler(
 }
 
 async fn list_workspaces_as_super_admin(
-    authed: Authed,
+    authed: ApiAuthed,
+    Extension(db): Extension<DB>,
     Extension(user_db): Extension<UserDB>,
     Query(pagination): Query<Pagination>,
-    Authed { email, .. }: Authed,
+    ApiAuthed { email, .. }: ApiAuthed,
 ) -> JsonResult<Vec<Workspace>> {
-    let mut tx = user_db.begin(&authed).await?;
-    require_super_admin(&mut tx, &email).await?;
+    require_super_admin(&db, &email).await?;
     let (per_page, offset) = paginate(pagination);
 
+    let mut tx = user_db.begin(&authed).await?;
     let workspaces = sqlx::query_as!(
         Workspace,
         "SELECT * FROM workspace LIMIT $1 OFFSET $2",
@@ -804,7 +807,7 @@ async fn list_workspaces_as_super_admin(
 
 async fn user_workspaces(
     Extension(db): Extension<DB>,
-    Authed { email, .. }: Authed,
+    ApiAuthed { email, .. }: ApiAuthed,
 ) -> JsonResult<WorkspaceList> {
     let mut tx = db.begin().await?;
     let workspaces = sqlx::query_as!(
@@ -841,15 +844,16 @@ lazy_static::lazy_static! {
 }
 
 async fn create_workspace(
-    authed: Authed,
+    authed: ApiAuthed,
     Extension(db): Extension<DB>,
     Json(nw): Json<CreateWorkspace>,
 ) -> Result<String> {
 
-    let mut tx: Transaction<'_, Postgres> = db.begin().await?;
     if *CREATE_WORKSPACE_REQUIRE_SUPERADMIN {
-        require_super_admin(&mut tx, &authed.email).await?;
+        require_super_admin(&db, &authed.email).await?;
     }
+    let mut tx: Transaction<'_, Postgres> = db.begin().await?;
+
     check_name_conflict(&mut tx, &nw.id).await?;
     sqlx::query!(
         "INSERT INTO workspace
@@ -937,10 +941,10 @@ async fn create_workspace(
 }
 
 async fn edit_workspace(
-    authed: Authed,
+    authed: ApiAuthed,
     Extension(db): Extension<DB>,
     Path(w_id): Path<String>,
-    Authed { is_admin, username, .. }: Authed,
+    ApiAuthed { is_admin, username, .. }: ApiAuthed,
     Json(ew): Json<EditWorkspace>,
 ) -> Result<String> {
     require_admin(is_admin, &username)?;
@@ -972,7 +976,7 @@ async fn edit_workspace(
 async fn archive_workspace(
     Extension(db): Extension<DB>,
     Path(w_id): Path<String>,
-    Authed { is_admin, username, email, .. }: Authed,
+    ApiAuthed { is_admin, username, email, .. }: ApiAuthed,
 ) -> Result<String> {
     require_admin(is_admin, &username)?;
     let mut tx = db.begin().await?;
@@ -998,7 +1002,7 @@ async fn archive_workspace(
 async fn unarchive_workspace(
     Extension(db): Extension<DB>,
     Path(w_id): Path<String>,
-    Authed { is_admin, username, email, .. }: Authed,
+    ApiAuthed { is_admin, username, email, .. }: ApiAuthed,
 ) -> Result<String> {
     require_admin(is_admin, &username)?;
     let mut tx = db.begin().await?;
@@ -1024,7 +1028,7 @@ async fn unarchive_workspace(
 async fn delete_workspace(
     Extension(db): Extension<DB>,
     Path(w_id): Path<String>,
-    Authed { username, email, .. }: Authed,
+    ApiAuthed { username, email, .. }: ApiAuthed,
 ) -> Result<String> {
     let w_id = match w_id.as_str() {
         "starter" => Err(Error::BadRequest(
@@ -1036,7 +1040,7 @@ async fn delete_workspace(
         _ => Ok(w_id),
     }?;
     let mut tx = db.begin().await?;
-    require_super_admin(&mut tx, &email).await?;
+    require_super_admin(&db, &email).await?;
 
     sqlx::query!("DELETE FROM script WHERE workspace_id = $1", &w_id)
         .execute(&mut *tx)
@@ -1142,7 +1146,7 @@ pub async fn invite_user_to_all_auto_invite_worspaces(db: &DB, email: &str) -> R
 }
 
 async fn invite_user(
-    Authed { username, is_admin, .. }: Authed,
+    ApiAuthed { username, is_admin, .. }: ApiAuthed,
     Extension(db): Extension<DB>,
     Extension(webhook): Extension<WebhookShared>,
     Path(w_id): Path<String>,
@@ -1191,7 +1195,7 @@ If you do not have an account on {}, login with SSO or ask an admin to create an
 }
 
 async fn add_user(
-    Authed { username, email, is_admin, .. }: Authed,
+    ApiAuthed { username, email, is_admin, .. }: ApiAuthed,
     Extension(db): Extension<DB>,
     Extension(webhook): Extension<WebhookShared>,
     Path(w_id): Path<String>,
@@ -1255,7 +1259,7 @@ If you do not have an account on {}, login with SSO or ask an admin to create an
 }
 
 async fn delete_invite(
-    Authed { username, is_admin, .. }: Authed,
+    ApiAuthed { username, is_admin, .. }: ApiAuthed,
     Extension(db): Extension<DB>,
     Path(w_id): Path<String>,
     Json(nu): Json<NewWorkspaceInvite>,
@@ -1413,7 +1417,7 @@ where
 }
 
 async fn tarball_workspace(
-    authed: Authed,
+    authed: ApiAuthed,
     Extension(db): Extension<DB>,
     Path(w_id): Path<String>,
     Query(ArchiveQueryParams {

backend/windmill-common/src/db.rs

@@ -0,0 +1,137 @@
+use sqlx::{Pool, Postgres, Transaction};
+
+pub type DB = Pool<Postgres>;
+
+#[derive(Clone, Debug)]
+pub struct Authed {
+    pub email: String,
+    pub username: String,
+    pub is_admin: bool,
+    pub is_operator: bool,
+    pub groups: Vec<String>,
+    // (folder name, can write, is owner)
+    pub folders: Vec<(String, bool, bool)>,
+    pub scopes: Option<Vec<String>>,
+}
+
+#[derive(Clone)]
+pub struct UserDB {
+    db: DB,
+}
+
+pub trait Authable {
+    fn email(&self) -> &str;
+    fn username(&self) -> &str;
+    fn is_admin(&self) -> bool;
+    fn is_operator(&self) -> bool;
+    fn groups(&self) -> &[String];
+    fn folders(&self) -> &[(String, bool, bool)];
+    fn scopes(&self) -> Option<&[String]>;
+}
+
+impl Authable for Authed {
+    fn is_admin(&self) -> bool {
+        self.is_admin
+    }
+
+    fn is_operator(&self) -> bool {
+        self.is_operator
+    }
+
+    fn groups(&self) -> &[String] {
+        &self.groups
+    }
+
+    fn folders(&self) -> &[(String, bool, bool)] {
+        &self.folders
+    }
+
+    fn scopes(&self) -> Option<&[std::string::String]> {
+        self.scopes.as_ref().map(|x| x.as_slice())
+    }
+
+    fn email(&self) -> &str {
+        &self.email
+    }
+
+    fn username(&self) -> &str {
+        &self.username
+    }
+}
+
+impl UserDB {
+    pub fn new(db: DB) -> Self {
+        Self { db }
+    }
+
+    pub async fn begin<T>(self, authed: &T) -> Result<Transaction<'static, Postgres>, sqlx::Error>
+    where
+        T: Authable,
+    {
+        let mut tx = self.db.begin().await?;
+        let user = if authed.is_admin() {
+            "windmill_admin"
+        } else {
+            "windmill_user"
+        };
+
+        sqlx::query(&format!("SET LOCAL ROLE {}", user))
+            .execute(&mut *tx)
+            .await?;
+
+        sqlx::query!(
+            "SELECT set_config('session.user', $1, true)",
+            authed.username()
+        )
+        .fetch_optional(&mut *tx)
+        .await?;
+
+        sqlx::query!(
+            "SELECT set_config('session.groups', $1, true)",
+            &authed.groups().join(",")
+        )
+        .fetch_optional(&mut *tx)
+        .await?;
+
+        sqlx::query!(
+            "SELECT set_config('session.pgroups', $1, true)",
+            &authed
+                .groups()
+                .iter()
+                .map(|x| format!("g/{}", x))
+                .collect::<Vec<_>>()
+                .join(",")
+        )
+        .fetch_optional(&mut *tx)
+        .await?;
+
+        let (folders_write, folders_read): &(Vec<_>, Vec<_>) =
+            &authed.folders().clone().into_iter().partition(|x| x.1);
+
+        let mut folders_read = folders_read.clone();
+        folders_read.extend(folders_write.clone());
+        sqlx::query!(
+            "SELECT set_config('session.folders_read', $1, true)",
+            folders_read
+                .iter()
+                .map(|x| x.0.clone())
+                .collect::<Vec<_>>()
+                .join(",")
+        )
+        .fetch_optional(&mut *tx)
+        .await?;
+
+        sqlx::query!(
+            "SELECT set_config('session.folders_write', $1, true)",
+            folders_write
+                .iter()
+                .map(|x| x.0.clone())
+                .collect::<Vec<_>>()
+                .join(",")
+        )
+        .fetch_optional(&mut *tx)
+        .await?;
+
+        Ok(tx)
+    }
+}

backend/windmill-common/src/jobs.rs

@@ -1,5 +1,5 @@
 use serde::{Deserialize, Serialize};
-use sqlx::{Postgres, Transaction};
+use sqlx::{Pool, Postgres, Transaction};
 use uuid::Uuid;
 
 use crate::{
@@ -201,9 +201,11 @@ pub struct RawCode {
 
 type Tag = String;
 
-pub async fn script_path_to_payload<'c>(
+pub type DB = Pool<Postgres>;
+
+pub async fn script_path_to_payload(
     script_path: &str,
-    db: &mut Transaction<'c, Postgres>,
+    db: &DB,
     w_id: &str,
 ) -> error::Result<(JobPayload, Option<Tag>)> {
     let (job_payload, tag) = if script_path.starts_with("hub/") {
@@ -248,13 +250,13 @@ pub async fn script_hash_to_tag_and_limits<'c>(
     ))
 }
 
-pub async fn get_payload_tag_from_prefixed_path<'c>(
+pub async fn get_payload_tag_from_prefixed_path(
     path: &str,
-    db: &mut Transaction<'c, Postgres>,
+    db: &DB,
     w_id: &str,
 ) -> Result<(JobPayload, Option<String>), Error> {
     let (payload, tag) = if path.starts_with("script/") {
-        script_path_to_payload(path.strip_prefix("script/").unwrap(), db, w_id).await?
+        script_path_to_payload(path.strip_prefix("script/").unwrap(), &db, w_id).await?
     } else if path.starts_with("flow/") {
         (
             JobPayload::Flow(path.strip_prefix("flow/").unwrap().to_string()),

backend/windmill-common/src/lib.rs

@@ -9,8 +9,10 @@
 use std::net::SocketAddr;
 
 use error::Error;
+use sqlx::{Pool, Postgres};
 
 pub mod apps;
+pub mod db;
 pub mod error;
 pub mod external_ip;
 pub mod flow_status;
@@ -155,8 +157,10 @@ pub async fn connect(
 
 type Tag = String;
 
-pub async fn get_latest_deployed_hash_for_path<'c>(
-    db: &mut sqlx::Transaction<'c, sqlx::Postgres>,
+pub type DB = Pool<Postgres>;
+
+pub async fn get_latest_deployed_hash_for_path(
+    db: &DB,
     w_id: &str,
     script_path: &str,
 ) -> error::Result<(scripts::ScriptHash, Option<Tag>, Option<i32>, Option<i32>)> {
@@ -167,7 +171,7 @@ pub async fn get_latest_deployed_hash_for_path<'c>(
         script_path,
         w_id
     )
-    .fetch_optional(&mut **db)
+    .fetch_optional(db)
     .await?;
 
     let script = utils::not_found_if_none(r_o, "script", script_path)?;

backend/windmill-queue/src/jobs.rs

@@ -21,6 +21,7 @@ use ulid::Ulid;
 use uuid::Uuid;
 use windmill_audit::{audit_log, ActionKind};
 use windmill_common::{
+    db::{Authed, UserDB},
     error::{self, Error},
     flow_status::{
         FlowStatus, FlowStatusModule, JobResult, MAX_RETRY_ATTEMPTS, MAX_RETRY_INTERVAL,
@@ -33,7 +34,7 @@ use windmill_common::{
     schedule::{schedule_to_user, Schedule},
     scripts::{ScriptHash, ScriptLang},
     users::{username_to_permissioned_as, SUPERADMIN_SECRET_EMAIL},
-    METRICS_ENABLED,
+    DB, METRICS_ENABLED,
 };
 
 use crate::{
@@ -436,17 +437,17 @@ pub async fn run_error_handler<R: rsmq_async::RsmqConnection + Clone + Send>(
     let w_id = &queued_job.workspace_id;
     let script_w_id = if is_global { "admins" } else { w_id }; // script workspace id
     let job_id = queued_job.id;
-    let mut tx: QueueTransaction<'_, _> = (rsmq, db.begin().await?).into();
-    let (job_payload, tag) =
-        script_path_to_payload(&error_handler_path, tx.transaction_mut(), script_w_id).await?;
+    let (job_payload, tag) = script_path_to_payload(&error_handler_path, db, script_w_id).await?;
     let mut args = result.as_object().unwrap().clone();
     args.insert("workspace_id".to_string(), json!(w_id));
     args.insert("job_id".to_string(), json!(job_id));
     args.insert("path".to_string(), json!(queued_job.script_path));
     args.insert("is_flow".to_string(), json!(queued_job.raw_flow.is_some()));
     args.insert("email".to_string(), json!(queued_job.email));
+    let tx = PushIsolationLevel::IsolatedRoot(db.clone(), rsmq);
 
     let (uuid, tx) = push(
+        &db,
         tx,
         script_w_id,
         job_payload,
@@ -556,6 +557,7 @@ pub async fn handle_maybe_scheduled_job<'c, R: rsmq_async::RsmqConnection + Clon
         if !success {
             if let Some(on_failure_path) = schedule.on_failure.clone() {
                 let on_failure_result = handle_on_failure(
+                    db,
                     tx,
                     schedule_path,
                     script_path,
@@ -593,6 +595,7 @@ pub async fn handle_maybe_scheduled_job<'c, R: rsmq_async::RsmqConnection + Clon
         }
 
         let res = push_scheduled_job(
+            db,
             tx,
             Schedule {
                 workspace_id: w_id.to_owned(),
@@ -635,7 +638,8 @@ pub async fn handle_maybe_scheduled_job<'c, R: rsmq_async::RsmqConnection + Clon
 }
 
 async fn handle_on_failure<'c, R: rsmq_async::RsmqConnection + Clone + Send + 'c>(
-    mut tx: QueueTransaction<'c, R>,
+    db: &Pool<Postgres>,
+    tx: QueueTransaction<'c, R>,
     schedule_path: &str,
     script_path: &str,
     w_id: &str,
@@ -645,8 +649,7 @@ async fn handle_on_failure<'c, R: rsmq_async::RsmqConnection + Clone + Send + 'c
     email: &str,
     permissioned_as: String,
 ) -> windmill_common::error::Result<QueueTransaction<'c, R>> {
-    let (payload, tag) =
-        get_payload_tag_from_prefixed_path(on_failure_path, tx.transaction_mut(), w_id).await?;
+    let (payload, tag) = get_payload_tag_from_prefixed_path(on_failure_path, db, w_id).await?;
 
     let mut args = result
         .map(|x| x.clone())
@@ -656,7 +659,9 @@ async fn handle_on_failure<'c, R: rsmq_async::RsmqConnection + Clone + Send + 'c
         .clone();
     args.insert("schedule_path".to_string(), json!(schedule_path));
     args.insert("path".to_string(), json!(script_path));
+    let tx = PushIsolationLevel::Transaction(tx);
     let (uuid, tx) = push(
+        &db,
         tx,
         w_id,
         payload,
@@ -1002,9 +1007,16 @@ pub async fn get_queued_job<'c>(
     Ok(r)
 }
 
+pub enum PushIsolationLevel<'c, R: rsmq_async::RsmqConnection + Send + 'c> {
+    IsolatedRoot(DB, Option<R>),
+    Isolated(UserDB, Authed, Option<R>),
+    Transaction(QueueTransaction<'c, R>),
+}
+
 // #[instrument(level = "trace", skip_all)]
 pub async fn push<'c, R: rsmq_async::RsmqConnection + Send + 'c>(
-    mut tx: QueueTransaction<'c, R>,
+    db: &Pool<Postgres>,
+    tx: PushIsolationLevel<'c, R>,
     workspace_id: &str,
     job_payload: JobPayload,
     args: serde_json::Map<String, serde_json::Value>,
@@ -1030,7 +1042,7 @@ pub async fn push<'c, R: rsmq_async::RsmqConnection + Send + 'c>(
             "SELECT 1 FROM queue WHERE id = $1 UNION ALL select 1 FROM completed_job WHERE id = $1",
             job_id
         )
-        .fetch_optional(&mut tx)
+        .fetch_optional(db)
         .await?;
 
         if conflicting_id.is_some() {
@@ -1048,7 +1060,7 @@ pub async fn push<'c, R: rsmq_async::RsmqConnection + Send + 'c>(
     {
         let premium_workspace = *CLOUD_HOSTED
             && sqlx::query_scalar!("SELECT premium FROM workspace WHERE id = $1", workspace_id)
-                .fetch_one(&mut tx)
+                .fetch_one(db)
                 .await
                 .map_err(|e| {
                     Error::InternalErr(format!("fetching if {workspace_id} is premium: {e}"))
@@ -1067,7 +1079,7 @@ pub async fn push<'c, R: rsmq_async::RsmqConnection + Send + 'c>(
                     if premium_workspace { workspace_id } else { email },
                     premium_workspace
                 )
-                .fetch_one(&mut tx)
+                .fetch_one(db)
                 .await
                 .map_err(|e| Error::InternalErr(format!("updating usage: {e}")))?
         } else if *CLOUD_HOSTED && !premium_workspace {
@@ -1079,7 +1091,7 @@ pub async fn push<'c, R: rsmq_async::RsmqConnection + Send + 'c>(
      AND id = $1",
                 email
             )
-            .fetch_optional(&mut tx)
+            .fetch_optional(db)
             .await?
             .flatten()
             .unwrap_or(0)
@@ -1090,7 +1102,7 @@ pub async fn push<'c, R: rsmq_async::RsmqConnection + Send + 'c>(
         if *CLOUD_HOSTED && !premium_workspace {
             let is_super_admin =
                 sqlx::query_scalar!("SELECT super_admin FROM password WHERE email = $1", email)
-                    .fetch_optional(&mut tx)
+                    .fetch_optional(db)
                     .await?
                     .unwrap_or(false);
 
@@ -1105,7 +1117,7 @@ pub async fn push<'c, R: rsmq_async::RsmqConnection + Send + 'c>(
                 }
                 let in_queue =
                     sqlx::query_scalar!("SELECT COUNT(id) FROM queue WHERE email = $1", email)
-                        .fetch_one(&mut tx)
+                        .fetch_one(db)
                         .await?
                         .unwrap_or(0);
 
@@ -1119,7 +1131,7 @@ pub async fn push<'c, R: rsmq_async::RsmqConnection + Send + 'c>(
                     "SELECT COUNT(id) FROM queue WHERE running = true AND email = $1",
                     email
                 )
-                .fetch_one(&mut tx)
+                .fetch_one(db)
                 .await?
                 .unwrap_or(0);
 
@@ -1148,7 +1160,7 @@ pub async fn push<'c, R: rsmq_async::RsmqConnection + Send + 'c>(
                     hash.0,
                     workspace_id
                 )
-                .fetch_one(&mut tx)
+                .fetch_one(db)
                 .await
                 .map_err(|e| {
                     Error::InternalErr(format!(
@@ -1212,7 +1224,7 @@ pub async fn push<'c, R: rsmq_async::RsmqConnection + Send + 'c>(
                 path,
                 workspace_id
             )
-            .fetch_optional(&mut tx)
+            .fetch_optional(db)
             .await?
             .ok_or_else(|| Error::InternalErr(format!("not found flow at path {:?}", path)))?;
             let value = serde_json::from_value::<FlowValue>(value_json).map_err(|err| {
@@ -1247,7 +1259,7 @@ pub async fn push<'c, R: rsmq_async::RsmqConnection + Send + 'c>(
                 flow,
                 workspace_id
             )
-            .fetch_optional(&mut tx)
+            .fetch_optional(db)
             .await?
             .ok_or_else(|| Error::InternalErr(format!("not found flow at path {:?}", flow)))?;
             let value = serde_json::from_value::<FlowValue>(value_json).map_err(|err| {
@@ -1347,6 +1359,13 @@ pub async fn push<'c, R: rsmq_async::RsmqConnection + Send + 'c>(
         })
     };
 
+    let mut tx = match tx {
+        PushIsolationLevel::Isolated(user_db, authed, rsmq) => {
+            (rsmq, user_db.begin(&authed).await?).into()
+        }
+        PushIsolationLevel::IsolatedRoot(db, rsmq) => (rsmq, db.begin().await?).into(),
+        PushIsolationLevel::Transaction(tx) => tx,
+    };
     let uuid = sqlx::query_scalar!(
         "INSERT INTO queue
             (workspace_id, id, running, parent_job, created_by, permissioned_as, scheduled_for, 

backend/windmill-queue/src/schedule.rs

@@ -7,11 +7,13 @@
  */
 
 use crate::push;
+use crate::PushIsolationLevel;
 use crate::QueueTransaction;
 use sqlx::{query_scalar, Postgres, Transaction};
 use std::str::FromStr;
 use windmill_common::jobs::JobPayload;
 use windmill_common::schedule::schedule_to_user;
+use windmill_common::DB;
 use windmill_common::{
     error::{self, Result},
     schedule::Schedule,
@@ -20,6 +22,7 @@ use windmill_common::{
 };
 
 pub async fn push_scheduled_job<'c, R: rsmq_async::RsmqConnection + Send + 'c>(
+    db: &DB,
     mut tx: QueueTransaction<'c, R>,
     schedule: Schedule,
 ) -> Result<QueueTransaction<'c, R>> {
@@ -48,7 +51,7 @@ pub async fn push_scheduled_job<'c, R: rsmq_async::RsmqConnection + Send + 'c>(
         &schedule.path,
         next
     )
-    .fetch_one(&mut tx)
+    .fetch_one(db)
     .await?
     .unwrap_or(false);
 
@@ -96,8 +99,9 @@ pub async fn push_scheduled_job<'c, R: rsmq_async::RsmqConnection + Send + 'c>(
     )
     .execute(&mut tx)
     .await?;
-
+    let tx = PushIsolationLevel::Transaction(tx);
     let (_, tx) = push(
+        &db,
         tx,
         &schedule.workspace_id,
         payload,

backend/windmill-worker/src/worker_flow.rs

@@ -29,7 +29,9 @@ use windmill_common::{
     },
     flows::{FlowModule, FlowModuleValue, FlowValue, InputTransform, Retry, Suspend},
 };
-use windmill_queue::{add_completed_job, add_completed_job_error, handle_maybe_scheduled_job};
+use windmill_queue::{
+    add_completed_job, add_completed_job_error, handle_maybe_scheduled_job, PushIsolationLevel,
+};
 
 type DB = sqlx::Pool<sqlx::Postgres>;
 
@@ -1410,8 +1412,10 @@ async fn push_next_flow_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
             flow_job.root_job.or_else(|| Some(flow_job.id))
         };
 
+        let tx2 = PushIsolationLevel::Transaction(tx);
         let (uuid, inner_tx) = push(
-            tx,
+            &db,
+            tx2,
             &flow_job.workspace_id,
             payload_tag.payload,
             ok.unwrap_or_else(|| Map::new()),
@@ -1708,8 +1712,7 @@ async fn compute_next_flow_transform(
         }
         FlowModuleValue::Script { path: script_path, hash: script_hash, .. } => {
             let (payload, tag) = if script_hash.is_none() {
-                let mut tx: sqlx::Transaction<'_, sqlx::Postgres> = db.begin().await?;
-                script_path_to_payload(script_path, &mut tx, &flow_job.workspace_id).await?
+                script_path_to_payload(script_path, &db, &flow_job.workspace_id).await?
             } else {
                 let hash = script_hash.clone().unwrap();
                 let mut tx: sqlx::Transaction<'_, sqlx::Postgres> = db.begin().await?;