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

feat(backend): only run fully deployed scripts

Browse Source
This commit is contained in:
Ruben Fiszel
2023-04-21 12:49:47 +02:00
parent c0ab27a9e7
commit 3d031c7017
9 changed files with 52 additions and 68 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/sqlx-data.json
+21 -21
View File
@@ -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": [],
backend/windmill-api/src/apps.rs
+1 -2
View File
@@ -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,
backend/windmill-api/src/jobs.rs
+1 -16
View File
@@ -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<JobPayload, Error> {
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<UserDB>,
backend/windmill-api/src/oauth2.rs
+1 -1
View File
@@ -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,
backend/windmill-common/src/jobs.rs
+17
View File
@@ -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<String>,
}
pub async fn script_path_to_payload<'c>(
script_path: &str,
db: &mut Transaction<'c, Postgres>,
w_id: &String,
) -> error::Result<JobPayload> {
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)
}
backend/windmill-common/src/lib.rs
+3 -4
View File
@@ -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<scripts::ScriptHash> {
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
)
backend/windmill-queue/src/schedule.rs
+2 -2
View File
@@ -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,
backend/windmill-worker/src/global_cache.rs
+5 -5
View File
@@ -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",
backend/windmill-worker/src/worker_flow.rs
+1 -17
View File
@@ -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<JobPayload, Error> {
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,