From c77d38563d8d770a381b8ba7d1e8efbdc03fbecc Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Tue, 22 Aug 2023 00:34:35 +0200 Subject: [PATCH] remove unecessary postgres transactions and improve performance --- backend/tests/worker.rs | 6 +- backend/windmill-api/src/apps.rs | 37 ++--- backend/windmill-api/src/audit.rs | 8 +- backend/windmill-api/src/capture.rs | 8 +- backend/windmill-api/src/db.rs | 137 ++++++++---------- backend/windmill-api/src/drafts.rs | 14 +- backend/windmill-api/src/favorite.rs | 6 +- backend/windmill-api/src/flows.rs | 38 +++-- backend/windmill-api/src/folders.rs | 31 ++-- backend/windmill-api/src/granular_acls.rs | 14 +- backend/windmill-api/src/groups.rs | 54 ++++--- backend/windmill-api/src/inputs.rs | 13 +- backend/windmill-api/src/jobs.rs | 156 ++++++++++++--------- backend/windmill-api/src/lib.rs | 7 +- backend/windmill-api/src/oauth2.rs | 28 ++-- backend/windmill-api/src/openai.rs | 8 +- backend/windmill-api/src/raw_apps.rs | 15 +- backend/windmill-api/src/resources.rs | 35 ++--- backend/windmill-api/src/schedule.rs | 26 ++-- backend/windmill-api/src/scripts.rs | 54 ++++--- backend/windmill-api/src/utils.rs | 9 +- backend/windmill-api/src/variables.rs | 19 +-- backend/windmill-api/src/workers.rs | 6 +- backend/windmill-api/src/workspaces.rs | 88 ++++++------ backend/windmill-common/src/db.rs | 137 ++++++++++++++++++ backend/windmill-common/src/jobs.rs | 14 +- backend/windmill-common/src/lib.rs | 10 +- backend/windmill-queue/src/jobs.rs | 55 +++++--- backend/windmill-queue/src/schedule.rs | 8 +- backend/windmill-worker/src/worker_flow.rs | 11 +- 30 files changed, 630 insertions(+), 422 deletions(-) create mode 100644 backend/windmill-common/src/db.rs diff --git a/backend/tests/worker.rs b/backend/tests/worker.rs index a89a27e8c5..a60c1a8d6b 100644 --- a/backend/tests/worker.rs +++ b/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) -> Uuid { let RunJob { payload, args } = self; + let tx = PushIsolationLevel::IsolatedRoot(db.clone(), None); let (uuid, tx) = windmill_queue::push::( - (None, db.begin().await.unwrap()).into(), + &db, + tx, "test-workspace", payload, args, diff --git a/backend/windmill-api/src/apps.rs b/backend/windmill-api/src/apps.rs index 77e75f4acc..75811f039f 100644 --- a/backend/windmill-api/src/apps.rs +++ b/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, Path(w_id): Path, Query(pagination): Query, @@ -220,7 +221,7 @@ async fn list_apps( } async fn get_app( - authed: Authed, + authed: ApiAuthed, Extension(user_db): Extension, Path((w_id, path)): Path<(String, StripPath)>, ) -> JsonResult { @@ -245,7 +246,7 @@ async fn get_app( } async fn get_app_w_draft( - authed: Authed, + authed: ApiAuthed, Extension(user_db): Extension, Path((w_id, path)): Path<(String, StripPath)>, ) -> JsonResult { @@ -276,7 +277,7 @@ async fn get_app_w_draft( } async fn get_app_by_id( - authed: Authed, + authed: ApiAuthed, Extension(user_db): Extension, Path((w_id, id)): Path<(String, i64)>, ) -> JsonResult { @@ -341,7 +342,7 @@ async fn get_public_app_by_secret( } async fn get_secret_id( - authed: Authed, + authed: ApiAuthed, Extension(user_db): Extension, Path((w_id, path)): Path<(String, StripPath)>, ) -> Result { @@ -369,7 +370,7 @@ async fn get_secret_id( } async fn create_app( - authed: Authed, + authed: ApiAuthed, Extension(user_db): Extension, Extension(webhook): Extension, Path(w_id): Path, @@ -461,7 +462,7 @@ async fn create_app( Ok((StatusCode::CREATED, app.path)) } -async fn list_hub_apps(Authed { email, .. }: Authed) -> JsonResult { +async fn list_hub_apps(ApiAuthed { email, .. }: ApiAuthed) -> JsonResult { 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, ) -> JsonResult { 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, Extension(webhook): Extension, 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, Extension(webhook): Extension, 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, diff --git a/backend/windmill-api/src/audit.rs b/backend/windmill-api/src/audit.rs index 57fd8cff49..1efd398fd6 100644 --- a/backend/windmill-api/src/audit.rs +++ b/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, Path(id): Path, ) -> JsonResult { @@ -32,7 +32,7 @@ async fn get_audit( Ok(Json(audit)) } async fn list_audit( - authed: Authed, + authed: ApiAuthed, Extension(user_db): Extension, Path(w_id): Path, Query(pagination): Query, diff --git a/backend/windmill-api/src/capture.rs b/backend/windmill-api/src/capture.rs index ddfca959e7..2a7669156f 100644 --- a/backend/windmill-api/src/capture.rs +++ b/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, Path((w_id, path)): Path<(String, StripPath)>, ) -> Result { @@ -120,7 +120,7 @@ pub async fn update_payload( } pub async fn get_payload( - authed: Authed, + authed: ApiAuthed, Extension(user_db): Extension, Path((w_id, path)): Path<(String, StripPath)>, ) -> JsonResult { diff --git a/backend/windmill-api/src/db.rs b/backend/windmill-api/src/db.rs index 8df5dd9ccb..cc1b652ee6 100644 --- a/backend/windmill-api/src/db.rs +++ b/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; @@ -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, + // (folder name, can write, is owner) + pub folders: Vec<(String, bool, bool)>, + pub scopes: Option>, } -impl UserDB { - pub fn new(db: DB) -> Self { - Self { db } - } - - pub async fn begin( - self, - authed: &Authed, - ) -> Result, 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::>() - .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::>() - .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::>() - .join(",") - ) - .fetch_optional(&mut *tx) - .await?; - - Ok(tx) +impl From 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 } } diff --git a/backend/windmill-api/src/drafts.rs b/backend/windmill-api/src/drafts.rs index 0220cf939d..a56aba9e3d 100644 --- a/backend/windmill-api/src/drafts.rs +++ b/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, Extension(user_db): Extension, Path(w_id): Path, @@ -95,7 +95,7 @@ async fn create_draft( } async fn delete_draft( - authed: Authed, + authed: ApiAuthed, Extension(user_db): Extension, Path((w_id, kind, path)): Path<(String, DraftType, StripPath)>, ) -> Result { @@ -115,7 +115,7 @@ async fn delete_draft( } // async fn get_draft( -// authed: Authed, +// authed: ApiAuthed, // Extension(user_db): Extension, // Path((w_id, path)): Path<(String, StripPath)>, // ) -> JsonResult { diff --git a/backend/windmill-api/src/favorite.rs b/backend/windmill-api/src/favorite.rs index 542f74711a..75c7b19bb2 100644 --- a/backend/windmill-api/src/favorite.rs +++ b/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, Path(w_id): Path, Json(Favorite { favorite_kind, path }): Json, @@ -58,7 +58,7 @@ async fn star( } async fn unstar( - authed: Authed, + authed: ApiAuthed, Extension(db): Extension, Path(w_id): Path, Json(Favorite { favorite_kind, path }): Json, diff --git a/backend/windmill-api/src/flows.rs b/backend/windmill-api/src/flows.rs index 797cc9d3b7..af71698a44 100644 --- a/backend/windmill-api/src/flows.rs +++ b/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, Path(w_id): Path, Query(pagination): Query, @@ -120,7 +123,7 @@ async fn list_flows( Ok(Json(rows)) } -async fn list_hub_flows(Authed { email, .. }: Authed) -> JsonResult { +async fn list_hub_flows(ApiAuthed { email, .. }: ApiAuthed) -> JsonResult { 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, Path(w_id): Path, ) -> JsonResult> { @@ -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, ) -> JsonResult { 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, Extension(user_db): Extension, Extension(rsmq): Extension>, @@ -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, Extension(rsmq): Extension>, Extension(db): Extension, @@ -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, Path((w_id, path)): Path<(String, StripPath)>, ) -> JsonResult { @@ -508,7 +516,7 @@ pub struct FlowWDraft { } async fn get_flow_by_path_w_draft( - authed: Authed, + authed: ApiAuthed, Extension(user_db): Extension, Path((w_id, path)): Path<(String, StripPath)>, ) -> JsonResult { @@ -555,7 +563,7 @@ struct Archived { } async fn archive_flow_by_path( - authed: Authed, + authed: ApiAuthed, Extension(user_db): Extension, Extension(webhook): Extension, 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, Extension(webhook): Extension, Path((w_id, path)): Path<(String, StripPath)>, diff --git a/backend/windmill-api/src/folders.rs b/backend/windmill-api/src/folders.rs index 4022cb4a0a..37df06e6e7 100644 --- a/backend/windmill-api/src/folders.rs +++ b/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, Path(w_id): Path, Query(pagination): Query, @@ -96,7 +99,7 @@ async fn list_folders( Ok(Json(rows)) } async fn list_foldernames( - authed: Authed, + authed: ApiAuthed, Extension(user_db): Extension, Path(w_id): Path, Query(pagination): Query, @@ -145,7 +148,7 @@ lazy_static! { } async fn create_folder( - authed: Authed, + authed: ApiAuthed, Tokened { token }: Tokened, Extension(user_db): Extension, Extension(webhook): Extension, @@ -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 { 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, Extension(webhook): Extension, 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, Path((w_id, name)): Path<(String, String)>, ) -> JsonResult { @@ -365,7 +368,7 @@ struct FolderUsage { pub variables: i64, } async fn get_folder_usage( - authed: Authed, + authed: ApiAuthed, Extension(user_db): Extension, Path((w_id, name)): Path<(String, String)>, ) -> JsonResult { @@ -446,7 +449,7 @@ async fn get_folder_usage( } async fn delete_folder( - authed: Authed, + authed: ApiAuthed, Extension(user_db): Extension, Extension(webhook): Extension, 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, Extension(webhook): Extension, 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, Extension(webhook): Extension, Path((w_id, name)): Path<(String, String)>, diff --git a/backend/windmill-api/src/granular_acls.rs b/backend/windmill-api/src/granular_acls.rs index c392df331c..1eca5fba2a 100644 --- a/backend/windmill-api/src/granular_acls.rs +++ b/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, Extension(user_db): Extension, 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, Extension(user_db): Extension, 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, Path((w_id, path)): Path<(String, StripPath)>, ) -> JsonResult { diff --git a/backend/windmill-api/src/groups.rs b/backend/windmill-api/src/groups.rs index 8924c6837b..55904ebf48 100644 --- a/backend/windmill-api/src/groups.rs +++ b/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, } async fn list_group_names( - Authed { username, email, .. }: Authed, + ApiAuthed { username, email, .. }: ApiAuthed, Extension(db): Extension, Query(QueryListGroup { only_member_of }): Query, Path(w_id): Path, @@ -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, Path((w_id, name)): Path<(String, String)>, ) -> JsonResult { @@ -203,7 +201,7 @@ pub async fn require_is_owner( } async fn create_group( - authed: Authed, + authed: ApiAuthed, Extension(user_db): Extension, Path(w_id): Path, Json(ng): Json, @@ -247,12 +245,13 @@ async fn create_group( } async fn create_igroup( - authed: Authed, + authed: ApiAuthed, Extension(db): Extension, Json(ng): Json, ) -> Result { + 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, Json(ng): Json, ) -> Result { - 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, Path((w_id, name)): Path<(String, String)>, ) -> JsonResult { @@ -358,7 +357,7 @@ async fn get_group( } async fn delete_group( - authed: Authed, + authed: ApiAuthed, Extension(db): Extension, Extension(user_db): Extension, 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, Extension(user_db): Extension, 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, Extension(user_db): Extension, 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, Path(name): Path, Json(Email { email }): Json, ) -> Result { - 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>, } -async fn list_igroups(authed: Authed, Extension(db): Extension) -> JsonResult> { - let mut tx = db.begin().await?; +async fn list_igroups(authed: ApiAuthed, Extension(db): Extension) -> JsonResult> { + 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, Extension(db): Extension) -> J } async fn remove_user_igroup( - authed: Authed, + authed: ApiAuthed, Extension(db): Extension, Path(name): Path, Json(Email { email }): Json, ) -> Result { + 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, Extension(user_db): Extension, Path((w_id, name)): Path<(String, String)>, diff --git a/backend/windmill-api/src/inputs.rs b/backend/windmill-api/src/inputs.rs index 1654bc7398..8284b5dd9a 100644 --- a/backend/windmill-api/src/inputs.rs +++ b/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, Path(w_id): Path, Query(pagination): Query, @@ -158,7 +159,7 @@ async fn get_input_history( } async fn list_saved_inputs( - authed: Authed, + authed: ApiAuthed, Extension(user_db): Extension, Path(w_id): Path, Query(pagination): Query, @@ -209,7 +210,7 @@ pub struct CreateInput { } async fn create_input( - authed: Authed, + authed: ApiAuthed, Extension(user_db): Extension, Path(w_id): Path, Query(r): Query, @@ -245,7 +246,7 @@ pub struct UpdateInput { } async fn update_input( - authed: Authed, + authed: ApiAuthed, Extension(user_db): Extension, Path(w_id): Path, Json(input): Json, @@ -266,7 +267,7 @@ async fn update_input( } async fn delete_input( - authed: Authed, + authed: ApiAuthed, Extension(user_db): Extension, Path((w_id, i_id)): Path<(String, Uuid)>, ) -> JsonResult { diff --git a/backend/windmill-api/src/jobs.rs b/backend/windmill-api/src/jobs.rs index 8564b2e48f..729d48463d 100644 --- a/backend/windmill-api/src/jobs.rs +++ b/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, Option, Option)> { @@ -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>> { 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, Path(w_id): Path, ) -> error::JsonResult> { @@ -690,7 +693,7 @@ async fn count_queue_jobs( } async fn list_jobs( - authed: Authed, + authed: ApiAuthed, Extension(user_db): Extension, Path(w_id): Path, Query(pagination): Query, @@ -807,7 +810,7 @@ async fn list_jobs( } pub async fn resume_suspended_flow_as_owner( - authed: Authed, + authed: ApiAuthed, Extension(db): Extension, Path((_w_id, flow_id)): Path<(String, Uuid)>, QueryOrBody(value): QueryOrBody, @@ -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, Path((w_id, job_id, resume_id)): Path<(String, Uuid, u32)>, Query(approver): Query, @@ -1175,7 +1178,7 @@ fn build_resume_url( } pub async fn get_resume_urls( - authed: Authed, + authed: ApiAuthed, Extension(user_db): Extension, Path((w_id, job_id, resume_id)): Path<(String, Uuid, u32)>, Query(approver): Query, @@ -1528,7 +1531,8 @@ fn check_tag_available_for_workspace(w_id: &str, tag: &Option) -> error: } pub async fn run_flow_by_path( - authed: Authed, + authed: ApiAuthed, + Extension(db): Extension, Extension(user_db): Extension, Extension(rsmq): Extension>, 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, Extension(user_db): Extension, Extension(rsmq): Extension>, 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( - authed: Authed, + authed: ApiAuthed, Extension(user_db): Extension, timeout: i32, uuid: Uuid, @@ -1726,7 +1733,7 @@ async fn run_wait_result( } } -pub async fn check_queue_too_long(db: DB, queue_limit: Option) -> error::Result<()> { +pub async fn check_queue_too_long(db: &DB, queue_limit: Option) -> 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) -> 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>, Extension(user_db): Extension, Extension(db): Extension, @@ -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>, Extension(user_db): Extension, Extension(db): Extension, @@ -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, Extension(rsmq): Extension>, Extension(db): Extension, @@ -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, Extension(rsmq): Extension>, Extension(db): Extension, @@ -1952,7 +1959,7 @@ async fn run_wait_result_script_by_path_internal( db: sqlx::Pool, run_query: RunJobQuery, script_path: StripPath, - authed: Authed, + authed: ApiAuthed, rsmq: Option, user_db: UserDB, w_id: String, @@ -1960,19 +1967,19 @@ async fn run_wait_result_script_by_path_internal( args: Option>, raw_string: Option, ) -> Result, 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, Extension(rsmq): Extension>, Extension(db): Extension, @@ -2016,19 +2023,20 @@ pub async fn run_wait_result_script_by_hash( headers: HeaderMap, JsonOrForm(args, raw_string): JsonOrForm, ) -> error::JsonResult { - 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, Extension(rsmq): Extension>, Extension(db): Extension, @@ -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, Extension(rsmq): Extension>, Extension(db): Extension, @@ -2112,7 +2120,7 @@ async fn run_wait_result_flow_by_path_internal( db: sqlx::Pool, run_query: RunJobQuery, flow_path: StripPath, - authed: Authed, + authed: ApiAuthed, rsmq: Option, user_db: UserDB, headers: HeaderMap, @@ -2120,13 +2128,12 @@ async fn run_wait_result_flow_by_path_internal( raw_string: Option, w_id: String, ) -> Result, 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, Extension(user_db): Extension, Extension(rsmq): Extension>, Path(w_id): Path, @@ -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, Extension(rsmq): Extension>, Path((w_id, n)): Path<(String, i32)>, ) -> error::JsonResult> { - 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 = 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, Extension(user_db): Extension, Extension(rsmq): Extension>, Path(w_id): Path, @@ -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, + Extension(user_db): Extension, Extension(rsmq): Extension>, 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, Path(w_id): Path, Query(pagination): Query, @@ -2647,7 +2669,7 @@ async fn get_completed_job_result_maybe( } async fn delete_completed_job( - authed: Authed, + authed: ApiAuthed, Extension(user_db): Extension, Path((w_id, id)): Path<(String, Uuid)>, ) -> error::JsonResult { diff --git a/backend/windmill-api/src/lib.rs b/backend/windmill-api/src/lib.rs index 34a76ccda1..59d9bb9d01 100644 --- a/backend/windmill-api/src/lib.rs +++ b/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::()) + .route_layer(from_extractor::()) .route_layer(from_extractor::()) .nest( "/saml", diff --git a/backend/windmill-api/src/oauth2.rs b/backend/windmill-api/src/oauth2.rs index 4f8a34e09e..065603aebd 100644 --- a/backend/windmill-api/src/oauth2.rs +++ b/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, Path(w_id): Path, Json(payload): Json, @@ -364,7 +366,7 @@ async fn create_account( } async fn delete_account( - authed: Authed, + authed: ApiAuthed, Extension(user_db): Extension, Query((w_id, id)): Query<(String, i32)>, ) -> error::Result { @@ -455,7 +457,7 @@ async fn connect_slack(cookies: Cookies) -> error::Result { } async fn disconnect( - authed: Authed, + authed: ApiAuthed, Path((w_id, id)): Path<(String, i32)>, Extension(user_db): Extension, ) -> error::Result { @@ -474,7 +476,7 @@ async fn disconnect( } async fn disconnect_slack( - authed: Authed, + authed: ApiAuthed, Path(w_id): Path, Extension(user_db): Extension, ) -> error::Result { @@ -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, Json(VariablePath { path }): Json, @@ -635,7 +637,7 @@ async fn connect_callback( async fn connect_slack_callback( Path(w_id): Path, - authed: Authed, + authed: ApiAuthed, cookies: Cookies, Extension(user_db): Extension, Json(callback): Json, @@ -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?)" diff --git a/backend/windmill-api/src/openai.rs b/backend/windmill-api/src/openai.rs index 5cd5717ed8..9bfded69b1 100644 --- a/backend/windmill-api/src/openai.rs +++ b/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, Path((w_id, openai_path)): Path<(String, String)>, body: Bytes, diff --git a/backend/windmill-api/src/raw_apps.rs b/backend/windmill-api/src/raw_apps.rs index cd4f1256b6..af73cf5b7b 100644 --- a/backend/windmill-api/src/raw_apps.rs +++ b/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, Path(w_id): Path, Query(pagination): Query, @@ -113,7 +114,7 @@ async fn list_apps( } async fn get_data( - authed: Authed, + authed: ApiAuthed, Extension(user_db): Extension, Path((w_id, _version, path)): Path<(String, u16, StripPath)>, ) -> Result { @@ -139,7 +140,7 @@ async fn get_data( } async fn create_app( - authed: Authed, + authed: ApiAuthed, Extension(user_db): Extension, Extension(webhook): Extension, Path(w_id): Path, @@ -199,7 +200,7 @@ async fn create_app( } async fn delete_app( - authed: Authed, + authed: ApiAuthed, Extension(user_db): Extension, Extension(webhook): Extension, 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, Extension(webhook): Extension, Path((w_id, path)): Path<(String, StripPath)>, diff --git a/backend/windmill-api/src/resources.rs b/backend/windmill-api/src/resources.rs index faed168588..73e237e6ae 100644 --- a/backend/windmill-api/src/resources.rs +++ b/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, } async fn list_resources( - authed: Authed, + authed: ApiAuthed, Query(lq): Query, Query(pagination): Query, Extension(user_db): Extension, @@ -175,7 +176,7 @@ async fn list_resources( } async fn get_resource( - authed: Authed, + authed: ApiAuthed, Extension(user_db): Extension, Path((w_id, path)): Path<(String, StripPath)>, ) -> JsonResult { @@ -223,7 +224,7 @@ async fn exists_resource( } async fn get_resource_value( - authed: Authed, + authed: ApiAuthed, Extension(user_db): Extension, Path((w_id, path)): Path<(String, StripPath)>, ) -> JsonResult> { @@ -244,7 +245,7 @@ async fn get_resource_value( } async fn get_resource_value_interpolated( - authed: Authed, + authed: ApiAuthed, Extension(user_db): Extension, Path((w_id, path)): Path<(String, StripPath)>, ) -> JsonResult> { @@ -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, } async fn create_resource( - authed: Authed, + authed: ApiAuthed, Extension(user_db): Extension, Extension(webhook): Extension, Extension(db): Extension, @@ -402,7 +403,7 @@ async fn create_resource( } async fn delete_resource( - authed: Authed, + authed: ApiAuthed, Extension(user_db): Extension, Extension(webhook): Extension, 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, Extension(webhook): Extension, Extension(db): Extension, @@ -527,7 +528,7 @@ struct UpdateResource { } async fn update_resource_value( - authed: Authed, + authed: ApiAuthed, Extension(user_db): Extension, Extension(webhook): Extension, 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, Path((w_id, name)): Path<(String, String)>, ) -> JsonResult { @@ -636,7 +637,7 @@ async fn exists_resource_type( } async fn create_resource_type( - authed: Authed, + authed: ApiAuthed, Extension(user_db): Extension, Extension(webhook): Extension, Path(w_id): Path, @@ -703,7 +704,7 @@ async fn check_rt_path_conflict<'c>( } async fn delete_resource_type( - authed: Authed, + authed: ApiAuthed, Extension(user_db): Extension, Extension(webhook): Extension, 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, Extension(webhook): Extension, Path((w_id, name)): Path<(String, String)>, diff --git a/backend/windmill-api/src/schedule.rs b/backend/windmill-api/src/schedule.rs index f0835218a0..36f057281c 100644 --- a/backend/windmill-api/src/schedule.rs +++ b/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, Extension(user_db): Extension, Extension(rsmq): Extension>, @@ -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, Extension(user_db): Extension, Extension(rsmq): Extension>, @@ -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, Path(w_id): Path, Query(lsq): Query, @@ -271,7 +272,7 @@ pub struct ScheduleWJobs { } async fn list_schedule_with_jobs( - authed: Authed, + authed: ApiAuthed, Extension(user_db): Extension, Path(w_id): Path, Query(pagination): Query, @@ -303,7 +304,7 @@ async fn list_schedule_with_jobs( // ) t; async fn get_schedule( - authed: Authed, + authed: ApiAuthed, Extension(user_db): Extension, Path((w_id, path)): Path<(String, StripPath)>, ) -> JsonResult { @@ -352,7 +353,8 @@ pub async fn preview_schedule( } pub async fn set_enabled( - authed: Authed, + authed: ApiAuthed, + Extension(db): Extension, Extension(user_db): Extension, Extension(rsmq): Extension>, 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, Path((w_id, path)): Path<(String, StripPath)>, ) -> Result { diff --git a/backend/windmill-api/src/scripts.rs b/backend/windmill-api/src/scripts.rs index 3e59f5dd30..93bf5c7a91 100644 --- a/backend/windmill-api/src/scripts.rs +++ b/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, Path(w_id): Path, Query(pagination): Query, @@ -192,7 +193,7 @@ async fn list_scripts( Ok(Json(rows)) } -async fn list_hub_scripts(Authed { email, .. }: Authed) -> JsonResult { +async fn list_hub_scripts(ApiAuthed { email, .. }: ApiAuthed) -> JsonResult { 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, Extension(rsmq): Extension>, Extension(webhook): Extension, @@ -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) -> Result { +pub async fn get_hub_script_by_path( + authed: ApiAuthed, + Path(path): Path, +) -> Result { 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, ) -> JsonResult { 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, Path((w_id, path)): Path<(String, StripPath)>, ) -> JsonResult