starred/windmill
1
0
Fork 0
mirror of https://github.com/windmill-labs/windmill.git synced 2026-09-05 08:02:18 +00:00
Code Issues Packages Projects Releases Wiki Activity

use push isolation level to isolated in more push cases

Browse Source
This commit is contained in:
Ruben Fiszel
2024-10-12 17:00:54 +02:00
parent 7a92a152d1
commit 242a251ab1
1 changed files with 18 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/windmill-api/src/jobs.rs
+18 -28
View File
@@ -81,7 +81,7 @@ use windmill_common::{METRICS_DEBUG_ENABLED, METRICS_ENABLED};
use windmill_common::{get_latest_deployed_hash_for_path, BASE_URL};
use windmill_queue::{
cancel_job, get_queued_job, get_result_by_id_from_running_flow, job_is_complete, push,
DecodeQueries, PushArgs, PushArgsOwned, PushIsolationLevel, QueueTransaction,
DecodeQueries, PushArgs, PushArgsOwned, PushIsolationLevel,
};
#[cfg(feature = "prometheus")]
@@ -544,8 +544,8 @@ pub async fn get_path_for_hash<'c>(
Ok(path)
}
pub async fn get_path_tag_limits_cache_for_hash<'c, R: rsmq_async::RsmqConnection + Send>(
tx: &mut QueueTransaction<'c, R>,
pub async fn get_path_tag_limits_cache_for_hash(
tx: &DB,
w_id: &str,
hash: i64,
) -> error::Result<(
@@ -2810,7 +2810,6 @@ pub async fn run_flow_by_path_inner(
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, dedicated_worker, has_preprocessor) = sqlx::query!(
"SELECT tag, dedicated_worker, flow_version.value->>'preprocessor_module' IS NOT NULL as has_preprocessor
@@ -2821,7 +2820,7 @@ pub async fn run_flow_by_path_inner(
flow_path,
w_id
)
.fetch_optional(&mut tx)
.fetch_optional(&db)
.await?
.map(|x| (x.tag, x.dedicated_worker, x.has_preprocessor))
.ok_or_else(|| {
@@ -2834,7 +2833,7 @@ pub async fn run_flow_by_path_inner(
check_tag_available_for_workspace(&w_id, &tag).await?;
let scheduled_for = run_query.get_scheduled_for(&db).await?;
let tx = PushIsolationLevel::Transaction(tx);
let tx = PushIsolationLevel::Isolated(user_db, authed.clone().into(), rsmq);
let (uuid, tx) = push(
&db,
tx,
@@ -2906,14 +2905,13 @@ pub async fn restart_flow(
) -> error::Result<(StatusCode, String)> {
check_license_key_valid().await?;
let mut tx: QueueTransaction<'_, _> = (rsmq, user_db.begin(&authed).await?).into();
let completed_job = sqlx::query_as::<_, CompletedJob>(
"SELECT *, result->'wm_labels' as labels from completed_job WHERE id = $1 and workspace_id = $2",
)
.bind(job_id)
.bind(&w_id)
.fetch_optional(&mut tx)
.fetch_optional(&db)
.await?
.with_context(|| "Unable to find completed job with the given job UUID")?;
@@ -2931,7 +2929,7 @@ pub async fn restart_flow(
let scheduled_for = run_query.get_scheduled_for(&db).await?;
let tx = PushIsolationLevel::Transaction(tx);
let tx = PushIsolationLevel::Isolated(user_db, authed.clone().into(), rsmq);
let (uuid, tx) = push(
&db,
@@ -3577,15 +3575,13 @@ pub async fn run_wait_result_job_by_path_get(
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, delete_after_use, timeout) =
script_path_to_payload(script_path, &mut tx, &w_id, run_query.skip_preprocessor).await?;
script_path_to_payload(script_path, &db, &w_id, run_query.skip_preprocessor).await?;
let tag = run_query.tag.clone().or(tag);
check_tag_available_for_workspace(&w_id, &tag).await?;
let tx = PushIsolationLevel::Transaction(tx);
let tx = PushIsolationLevel::Isolated(user_db, authed.clone().into(), rsmq);
let (uuid, tx) = push(
&db,
@@ -3702,15 +3698,13 @@ pub async fn run_wait_result_script_by_path_internal(
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, delete_after_use, timeout) =
script_path_to_payload(script_path, &mut tx, &w_id, run_query.skip_preprocessor).await?;
script_path_to_payload(script_path, &db, &w_id, run_query.skip_preprocessor).await?;
let tag = run_query.tag.clone().or(tag);
check_tag_available_for_workspace(&w_id, &tag).await?;
let tx = PushIsolationLevel::Transaction(tx);
let tx = PushIsolationLevel::Isolated(user_db, authed.clone().into(), rsmq);
let (uuid, tx) = push(
&db,
@@ -3762,8 +3756,6 @@ pub async fn run_wait_result_script_by_hash(
check_queue_too_long(&db, run_query.queue_limit).await?;
let mut tx: QueueTransaction<'_, _> = (rsmq, user_db.begin(&authed).await?).into();
let hash = script_hash.0;
let (
path,
@@ -3778,7 +3770,7 @@ pub async fn run_wait_result_script_by_hash(
delete_after_use,
timeout,
has_preprocessor,
) = get_path_tag_limits_cache_for_hash(&mut tx, &w_id, hash).await?;
) = get_path_tag_limits_cache_for_hash(&db, &w_id, hash).await?;
if let Some(run_query_cache_ttl) = run_query.cache_ttl {
cache_ttl = Some(run_query_cache_ttl);
}
@@ -3787,7 +3779,7 @@ pub async fn run_wait_result_script_by_hash(
let tag = run_query.tag.clone().or(tag);
check_tag_available_for_workspace(&w_id, &tag).await?;
let tx = PushIsolationLevel::Transaction(tx);
let tx = PushIsolationLevel::Isolated(user_db, authed.clone().into(), rsmq);
let (uuid, tx) = push(
&db,
@@ -3869,7 +3861,6 @@ pub async fn run_wait_result_flow_by_path_internal(
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 scheduled_for = run_query.get_scheduled_for(&db).await?;
@@ -3882,7 +3873,7 @@ pub async fn run_wait_result_flow_by_path_internal(
flow_path,
w_id
)
.fetch_optional(&mut tx)
.fetch_optional(&db)
.await?
.map(|x| (x.tag, x.dedicated_worker, x.early_return, x.has_preprocessor))
.ok_or_else(|| {
@@ -3894,7 +3885,7 @@ pub async fn run_wait_result_flow_by_path_internal(
let tag = run_query.tag.clone().or(tag);
check_tag_available_for_workspace(&w_id, &tag).await?;
let tx = PushIsolationLevel::Transaction(tx);
let tx = PushIsolationLevel::Isolated(user_db, authed.clone().into(), rsmq);
let (uuid, tx) = push(
&db,
@@ -4366,7 +4357,6 @@ async fn add_batch_jobs(
}
}
"flow" => {
let mut tx = PushIsolationLevel::IsolatedRoot(db.clone(), rsmq);
let mut uuids: Vec<Uuid> = Vec::new();
let payload = if let Some(ref fv) = batch_info.flow_value {
@@ -4384,6 +4374,7 @@ async fn add_batch_jobs(
))?
}
};
let mut tx = PushIsolationLevel::IsolatedRoot(db.clone(), rsmq);
for _ in 0..n {
let ehm = HashMap::new();
let (uuid, ntx) = push(
@@ -4584,7 +4575,6 @@ pub async fn run_job_by_hash_inner(
#[cfg(feature = "enterprise")]
check_license_key_valid().await?;
let mut tx: QueueTransaction<'_, _> = (rsmq, user_db.begin(&authed).await?).into();
let hash = script_hash.0;
let (
@@ -4600,7 +4590,7 @@ pub async fn run_job_by_hash_inner(
_delete_after_use, // not taken into account in async endpoints
timeout,
has_preprocessor,
) = get_path_tag_limits_cache_for_hash(&mut tx, &w_id, hash).await?;
) = get_path_tag_limits_cache_for_hash(&db, &w_id, hash).await?;
check_scopes(&authed, || format!("run:script/{path}"))?;
if let Some(run_query_cache_ttl) = run_query.cache_ttl {
cache_ttl = Some(run_query_cache_ttl);
@@ -4609,7 +4599,7 @@ pub async fn run_job_by_hash_inner(
let tag = run_query.tag.clone().or(tag);
check_tag_available_for_workspace(&w_id, &tag).await?;
let tx = PushIsolationLevel::Transaction(tx);
let tx = PushIsolationLevel::Isolated(user_db, authed.clone().into(), rsmq);
let (uuid, tx) = push(
&db,