From 3d031c701705459f418b11d2ca83e71943e4079b Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Fri, 21 Apr 2023 12:49:47 +0200 Subject: [PATCH] feat(backend): only run fully deployed scripts --- backend/sqlx-data.json | 42 ++++++++++----------- backend/windmill-api/src/apps.rs | 3 +- backend/windmill-api/src/jobs.rs | 17 +-------- backend/windmill-api/src/oauth2.rs | 2 +- backend/windmill-common/src/jobs.rs | 17 +++++++++ backend/windmill-common/src/lib.rs | 7 ++-- backend/windmill-queue/src/schedule.rs | 4 +- backend/windmill-worker/src/global_cache.rs | 10 ++--- backend/windmill-worker/src/worker_flow.rs | 18 +-------- 9 files changed, 52 insertions(+), 68 deletions(-) diff --git a/backend/sqlx-data.json b/backend/sqlx-data.json index 14ca54a791..9eeb1e1ecb 100644 --- a/backend/sqlx-data.json +++ b/backend/sqlx-data.json @@ -3160,6 +3160,27 @@ }, "query": "DELETE FROM usr_to_group WHERE workspace_id = $1" }, + "897523b3096f22d1fdc877cb2501ead81589ad3eb851459b8b3764833261f7fa": { + "describe": { + "columns": [ + { + "name": "hash", + "ordinal": 0, + "type_info": "Int8" + } + ], + "nullable": [ + false + ], + "parameters": { + "Left": [ + "Text", + "Text" + ] + } + }, + "query": "select hash from script where path = $1 AND workspace_id = $2 AND\n created_at = (SELECT max(created_at) FROM script WHERE path = $1 AND workspace_id = $2 AND\n deleted = false AND archived = false AND lock IS not NULL AND lock_error_logs IS NULL)" + }, "8a80333c2fbf7b50fed305882de6e4ffda985d5c648cd617add6c9e6a9c03f34": { "describe": { "columns": [], @@ -5141,27 +5162,6 @@ }, "query": "INSERT INTO group_\n VALUES ($1, 'all', 'The group that always contains all users of this workspace')" }, - "cb9b85cba9feec1ed8543718fba2856cf1aed6239ed9a2173ebb66f4a2697df2": { - "describe": { - "columns": [ - { - "name": "hash", - "ordinal": 0, - "type_info": "Int8" - } - ], - "nullable": [ - false - ], - "parameters": { - "Left": [ - "Text", - "Text" - ] - } - }, - "query": "select hash from script where path = $1 AND workspace_id = $2 AND\n created_at = (SELECT max(created_at) FROM script WHERE path = $1 AND workspace_id = $2) AND\n deleted = false" - }, "d0308abac80575038203b60bb66d3b39b586939da0421a595e47c7a759616431": { "describe": { "columns": [], diff --git a/backend/windmill-api/src/apps.rs b/backend/windmill-api/src/apps.rs index e59d1fde09..316f4024f6 100644 --- a/backend/windmill-api/src/apps.rs +++ b/backend/windmill-api/src/apps.rs @@ -9,7 +9,6 @@ use std::collections::HashMap; */ use crate::{ db::{UserDB, DB}, - jobs::script_path_to_payload, users::{require_owner_of_path, Authed, OptAuthed}, variables::build_crypt, webhook_util::{WebhookMessage, WebhookShared}, @@ -32,7 +31,7 @@ use windmill_audit::{audit_log, ActionKind}; use windmill_common::{ apps::ListAppQuery, error::{to_anyhow, Error, JsonResult, Result}, - jobs::{JobPayload, RawCode}, + jobs::{JobPayload, RawCode, script_path_to_payload}, users::username_to_permissioned_as, utils::{ http_get_from_hub, list_elems_from_hub, not_found_if_none, paginate, Pagination, StripPath, diff --git a/backend/windmill-api/src/jobs.rs b/backend/windmill-api/src/jobs.rs index e99d5447f7..f5ee72ee43 100644 --- a/backend/windmill-api/src/jobs.rs +++ b/backend/windmill-api/src/jobs.rs @@ -31,7 +31,7 @@ use windmill_common::{ error::{self, to_anyhow, Error}, flow_status::{Approval, FlowStatus, FlowStatusModule}, flows::FlowValue, - jobs::{JobKind, JobPayload, QueuedJob, RawCode}, + jobs::{script_path_to_payload, JobKind, JobPayload, QueuedJob, RawCode}, oauth2::HmacSha256, scripts::{ScriptHash, ScriptLang}, users::username_to_permissioned_as, @@ -1609,21 +1609,6 @@ pub async fn run_wait_result_flow_by_path( .await } -// a similar function exists on the worker -pub async fn script_path_to_payload<'c>( - script_path: &str, - db: &mut Transaction<'c, Postgres>, - w_id: &String, -) -> std::result::Result { - let job_payload = if script_path.starts_with("hub/") { - JobPayload::ScriptHub { path: script_path.to_owned() } - } else { - let script_hash = windmill_common::get_latest_hash_for_path(db, w_id, script_path).await?; - JobPayload::ScriptHash { hash: script_hash, path: script_path.to_owned() } - }; - Ok(job_payload) -} - async fn run_preview_job( authed: Authed, Extension(user_db): Extension, diff --git a/backend/windmill-api/src/oauth2.rs b/backend/windmill-api/src/oauth2.rs index bb30d74868..f93f0c4536 100644 --- a/backend/windmill-api/src/oauth2.rs +++ b/backend/windmill-api/src/oauth2.rs @@ -758,7 +758,7 @@ async fn slack_command( JobPayload::Flow(path.to_string()) } else { let path = path.strip_prefix("script/").unwrap_or_else(|| path); - let script_hash = windmill_common::get_latest_hash_for_path( + let script_hash = windmill_common::get_latest_deployed_hash_for_path( tx.transaction_mut(), &settings.workspace_id, path, diff --git a/backend/windmill-common/src/jobs.rs b/backend/windmill-common/src/jobs.rs index b4aaafd6a8..722bdf0629 100644 --- a/backend/windmill-common/src/jobs.rs +++ b/backend/windmill-common/src/jobs.rs @@ -1,9 +1,12 @@ use serde::{Deserialize, Serialize}; +use sqlx::{Postgres, Transaction}; use uuid::Uuid; use crate::{ + error, flow_status::FlowStatus, flows::FlowValue, + get_latest_deployed_hash_for_path, scripts::{ScriptHash, ScriptLang}, }; @@ -160,3 +163,17 @@ pub struct RawCode { pub language: ScriptLang, pub lock: Option, } + +pub async fn script_path_to_payload<'c>( + script_path: &str, + db: &mut Transaction<'c, Postgres>, + w_id: &String, +) -> error::Result { + let job_payload = if script_path.starts_with("hub/") { + JobPayload::ScriptHub { path: script_path.to_owned() } + } else { + let script_hash = get_latest_deployed_hash_for_path(db, w_id, script_path).await?; + JobPayload::ScriptHash { hash: script_hash, path: script_path.to_owned() } + }; + Ok(job_payload) +} diff --git a/backend/windmill-common/src/lib.rs b/backend/windmill-common/src/lib.rs index 55885e76fb..af98b6645a 100644 --- a/backend/windmill-common/src/lib.rs +++ b/backend/windmill-common/src/lib.rs @@ -147,16 +147,15 @@ pub async fn connect( .map_err(|err| Error::ConnectingToDatabase(err.to_string())) } -// TODO: Move this elsewhere -pub async fn get_latest_hash_for_path<'c>( +pub async fn get_latest_deployed_hash_for_path<'c>( db: &mut sqlx::Transaction<'c, sqlx::Postgres>, w_id: &str, script_path: &str, ) -> error::Result { let script_hash_o = sqlx::query_scalar!( "select hash from script where path = $1 AND workspace_id = $2 AND - created_at = (SELECT max(created_at) FROM script WHERE path = $1 AND workspace_id = $2) AND - deleted = false", + created_at = (SELECT max(created_at) FROM script WHERE path = $1 AND workspace_id = $2 AND + deleted = false AND archived = false AND lock IS not NULL AND lock_error_logs IS NULL)", script_path, w_id ) diff --git a/backend/windmill-queue/src/schedule.rs b/backend/windmill-queue/src/schedule.rs index e58c217ac4..47ea575f26 100644 --- a/backend/windmill-queue/src/schedule.rs +++ b/backend/windmill-queue/src/schedule.rs @@ -9,8 +9,8 @@ use crate::push; use crate::QueueTransaction; use sqlx::{query_scalar, Postgres, Transaction}; -use windmill_common::jobs::JobPayload; use std::str::FromStr; +use windmill_common::jobs::JobPayload; use windmill_common::{ error::{self, Result}, schedule::Schedule, @@ -71,7 +71,7 @@ pub async fn push_scheduled_job<'c, R: rsmq_async::RsmqConnection + Send + 'c>( JobPayload::Flow(schedule.script_path) } else { JobPayload::ScriptHash { - hash: windmill_common::get_latest_hash_for_path( + hash: windmill_common::get_latest_deployed_hash_for_path( tx.transaction_mut(), &schedule.workspace_id, &schedule.script_path, diff --git a/backend/windmill-worker/src/global_cache.rs b/backend/windmill-worker/src/global_cache.rs index 704fd0cb8a..2907109a3e 100644 --- a/backend/windmill-worker/src/global_cache.rs +++ b/backend/windmill-worker/src/global_cache.rs @@ -148,7 +148,7 @@ pub async fn copy_cache_from_bucket(bucket: &str, tx: Sender<()>) -> error::Resu "--size-only", "--fast-list", "--filter", - "- deno/gen/file/tmp/windmill/**", + "- deno/gen/file/**", "--filter", "+ deno/**", "--filter", @@ -190,7 +190,7 @@ pub async fn copy_cache_to_bucket(bucket: &str) -> error::Result<()> { "--size-only", "--fast-list", "--filter", - "- deno/gen/file/tmp/windmill/**", + "- deno/gen/file/**", "--filter", "+ deno/**", "--filter", @@ -312,7 +312,7 @@ pub async fn copy_denogo_cache_from_bucket_as_tar(bucket: &str) { return; } - let denogen = format!("{ROOT_TMP_CACHE_DIR}deno/gen/file/tmp/windmill"); + let denogen = format!("{ROOT_TMP_CACHE_DIR}deno/gen/file"); if metadata(&denogen).await.is_ok() { let _ = tokio::fs::remove_dir_all(denogen).await; } @@ -385,7 +385,7 @@ pub async fn copy_tmp_cache_to_cache() -> error::Result<()> { ROOT_TMP_CACHE_DIR, ROOT_CACHE_DIR, "--filter", - "- deno/gen/file/tmp/windmill/**", + "- deno/gen/file/**", "--filter", "+ deno/**", "--filter", @@ -474,7 +474,7 @@ pub async fn copy_cache_to_tmp_cache() -> error::Result<()> { ROOT_CACHE_DIR, ROOT_TMP_CACHE_DIR, "--filter", - "- deno/gen/file/tmp/windmill/**", + "- deno/gen/file/**", "--filter", "+ deno/**", "--filter", diff --git a/backend/windmill-worker/src/worker_flow.rs b/backend/windmill-worker/src/worker_flow.rs index d1ddc2a4df..3feb75669b 100644 --- a/backend/windmill-worker/src/worker_flow.rs +++ b/backend/windmill-worker/src/worker_flow.rs @@ -19,7 +19,7 @@ use tokio::sync::mpsc::Sender; use tracing::instrument; use uuid::Uuid; use windmill_common::flow_status::{FlowStatusModuleWParent, Iterator, JobResult}; -use windmill_common::jobs::{QueuedJob, JobPayload, RawCode}; +use windmill_common::jobs::{script_path_to_payload, JobPayload, QueuedJob, RawCode}; use windmill_common::{ error::{self, to_anyhow, Error}, flow_status::{ @@ -1614,22 +1614,6 @@ enum NextFlowTransform { Continue(ContinuePayload, NextStatus), } -// a similar function exists on the backend -// TODO: rewrite this to use an endpoint in the backend directly, instead of checking for hub itself, and then using the API -async fn script_path_to_payload<'c>( - script_path: &str, - db: &mut sqlx::Transaction<'c, sqlx::Postgres>, - w_id: &String, -) -> Result { - let job_payload = if script_path.starts_with("hub/") { - JobPayload::ScriptHub { path: script_path.to_owned() } - } else { - let script_hash = windmill_common::get_latest_hash_for_path(db, w_id, script_path).await?; - JobPayload::ScriptHash { hash: script_hash, path: script_path.to_owned() } - }; - Ok(job_payload) -} - async fn compute_next_flow_transform<'c>( flow_job: &QueuedJob, flow: &FlowValue,