fix: dep jobs improvements (#7081)

* dep jobs improvements

* update
This commit is contained in:
Ruben Fiszel
2025-11-07 12:04:53 +01:00
committed by GitHub
parent 10e621cffe
commit ad43680bb9
7 changed files with 63 additions and 53 deletions
+1
View File
@@ -53,6 +53,7 @@ pub mod job_metrics;
pub mod job_s3_helpers_ee;
#[cfg(feature = "parquet")]
pub mod job_s3_helpers_oss;
pub mod lockfiles;
#[cfg(feature = "private")]
pub mod git_sync_ee;
+17
View File
@@ -0,0 +1,17 @@
use crate::scripts::ScriptLang;
pub const LOCKFILE_GENERATED_FROM_REQUIREMENTS_TXT: &str = "# from requirements.txt";
pub fn is_generated_from_raw_requirements(
lang: &Option<ScriptLang>,
lock: &Option<String>,
) -> bool {
(lang.is_some_and(|v| v == ScriptLang::Bun)
&& lock
.as_ref()
.is_some_and(|v| v.contains("generatedFromPackageJson")))
|| (lang.is_some_and(|v| v == ScriptLang::Python3)
&& lock
.as_ref()
.is_some_and(|v| v.starts_with(LOCKFILE_GENERATED_FROM_REQUIREMENTS_TXT)))
}
+22 -8
View File
@@ -732,18 +732,32 @@ pub fn hash_script(ns: &NewScript) -> i64 {
dh.finish() as i64
}
pub struct ClonedScript {
pub old_script: NewScript,
pub new_hash: i64,
}
pub async fn clone_script<'c>(
base_hash: ScriptHash,
w_id: &str,
deployment_message: Option<String>,
tx: &mut sqlx::Transaction<'c, sqlx::Postgres>,
) -> crate::error::Result<i64> {
let s =
sqlx::query_as::<_, Script>("SELECT * FROM script WHERE hash = $1 AND workspace_id = $2")
.bind(base_hash.0)
.bind(w_id)
.fetch_one(&mut **tx)
.await?;
) -> crate::error::Result<ClonedScript> {
let s = sqlx::query_as::<_, Script>(
"SELECT * FROM script WHERE hash = $1 AND workspace_id = $2 AND archived = false FOR UPDATE",
)
.bind(base_hash.0)
.bind(w_id)
.fetch_optional(&mut **tx)
.await?;
let s = if let Some(s) = s {
s
} else {
return Err(crate::error::Error::NotFound(format!(
"Non-archived script with hash {} not found",
base_hash.0
)));
};
let ns = NewScript {
path: s.path.clone(),
@@ -819,5 +833,5 @@ pub async fn clone_script<'c>(
.execute(&mut **tx)
.await?;
Ok(new_hash)
Ok(ClonedScript { old_script: ns, new_hash })
}
+9 -3
View File
@@ -35,6 +35,7 @@ use windmill_common::add_time;
use windmill_common::auth::JobPerms;
#[cfg(feature = "benchmark")]
use windmill_common::bench::BenchmarkIter;
use windmill_common::lockfiles::is_generated_from_raw_requirements;
use windmill_common::jobs::{JobTriggerKind, EMAIL_ERROR_HANDLER_USER_EMAIL};
use windmill_common::utils::{configure_client, now_from_db};
use windmill_common::worker::{Connection, MIN_VERSION_SUPPORTS_DEBOUNCING, SCRIPT_TOKEN_EXPIRY};
@@ -5660,15 +5661,20 @@ pub async fn preprocess_dependency_job(job: &mut PulledJob, db: &DB) -> error::R
args.insert("base_hash".to_owned(), to_raw_value(&*base_hash))
});
let new_hash = windmill_common::scripts::clone_script(
let cloned_script = windmill_common::scripts::clone_script(
base_hash,
&job.workspace_id,
deployment_message,
&mut tx,
)
.await?;
new_hash
if is_generated_from_raw_requirements(&Some(cloned_script.old_script.language), &cloned_script.old_script.lock.map(|v| v.to_string())) {
return Err(Error::BadRequest(format!(
"Script at path {} is generated from raw requirements, not overriding",
job.runnable_path()
)));
}
cloned_script.new_hash
}
JobKind::FlowDependencies => {
sqlx::query_scalar!(
@@ -12,6 +12,7 @@ use tokio::{fs::DirBuilder, process::Command, sync::RwLock};
use uuid::Uuid;
use windmill_common::{
error::{self, Error},
lockfiles::LOCKFILE_GENERATED_FROM_REQUIREMENTS_TXT,
worker::Connection,
};
@@ -22,7 +23,6 @@ use crate::{
common::{start_child_process, OccupancyMetrics},
handle_child::handle_child,
python_executor::{PYTHON_PATH, UV_PATH},
worker_lockfiles::LOCKFILE_GENERATED_FROM_REQUIREMENTS_TXT,
HOME_ENV, INSTANCE_PYTHON_VERSION, PATH_ENV, PROXY_ENVS, PY_INSTALL_DIR, WIN_ENVS,
};
@@ -5,15 +5,12 @@ use windmill_common::{
cache,
error::{Error, Result},
flows::{FlowModuleValue, FlowValue},
scripts::ScriptLang,
};
use std::collections::HashSet;
use crate::worker_lockfiles::{
extract_relative_imports, is_generated_from_raw_requirements,
LOCKFILE_GENERATED_FROM_REQUIREMENTS_TXT,
};
use crate::worker_lockfiles::extract_relative_imports;
use windmill_common::lockfiles::is_generated_from_raw_requirements;
// TODO: To be removed in future versions
lazy_static::lazy_static! {
@@ -295,16 +292,7 @@ SELECT importer_node_id, imported_path
let mut dmap = ScopedDependencyMap::fetch(w_id, &r.path, "script", db).await?;
let mut tx = db.begin().await?;
if (smd.language.is_some_and(|v| v == ScriptLang::Bun)
&& sd
.lock
.as_ref()
.is_some_and(|v| v.contains("generatedFromPackageJson")))
|| (smd.language.is_some_and(|v| v == ScriptLang::Python3)
&& sd.lock.as_ref().is_some_and(|v| {
v.starts_with(LOCKFILE_GENERATED_FROM_REQUIREMENTS_TXT)
}))
{
if is_generated_from_raw_requirements(&smd.language, &sd.lock) {
// if the lock file is generated from a package.json/requirements.txt, we need to clear the dependency map
// because we do not want to have dependencies be recomputed automatically. Empty relative imports passed
// to update_script_dependency_map will clear the dependency map.
@@ -350,7 +338,7 @@ SELECT importer_node_id, imported_path
match fmv {
// Since we fetched from flow_version it is safe to assume all inline scripts are in form of RawScript.
FlowModuleValue::RawScript { content, language, lock ,.. } => {
if !is_generated_from_raw_requirements(Some(*language), lock) {
if !is_generated_from_raw_requirements(&Some(*language), lock) {
to_process.push((
extract_relative_imports(
content,
@@ -20,6 +20,7 @@ use windmill_common::error::Error;
use windmill_common::error::Result;
use windmill_common::flows::{FlowModule, FlowModuleValue, FlowNodeId};
use windmill_common::jobs::JobPayload;
use windmill_common::lockfiles::is_generated_from_raw_requirements;
use windmill_common::scripts::ScriptHash;
use windmill_common::utils::WarnAfterExt;
#[cfg(feature = "python")]
@@ -373,15 +374,7 @@ pub async fn process_relative_imports(
db,
)
.await?;
if (script_lang.is_some_and(|v| v == ScriptLang::Bun)
&& lock
.as_ref()
.is_some_and(|v| v.contains("generatedFromPackageJson")))
|| (script_lang.is_some_and(|v| v == ScriptLang::Python3)
&& lock
.as_ref()
.is_some_and(|v| v.starts_with(LOCKFILE_GENERATED_FROM_REQUIREMENTS_TXT)))
{
if is_generated_from_raw_requirements(script_lang, &lock) {
// if the lock file is generated from a package.json/requirements.txt, we need to clear the dependency map
// because we do not want to have dependencies be recomputed automatically. Empty relative imports passed
// to update_script_dependency_map will clear the dependency map.
@@ -445,17 +438,6 @@ pub async fn process_relative_imports(
Ok(())
}
pub fn is_generated_from_raw_requirements(lang: Option<ScriptLang>, lock: &Option<String>) -> bool {
(lang.is_some_and(|v| v == ScriptLang::Bun)
&& lock
.as_ref()
.is_some_and(|v| v.contains("generatedFromPackageJson")))
|| (lang.is_some_and(|v| v == ScriptLang::Python3)
&& lock
.as_ref()
.is_some_and(|v| v.starts_with(LOCKFILE_GENERATED_FROM_REQUIREMENTS_TXT)))
}
pub async fn trigger_dependents_to_recompute_dependencies(
w_id: &str,
script_path: &str,
@@ -1450,7 +1432,7 @@ async fn lock_modules<'c>(
if let Some(locks_to_reload) = locks_to_reload {
if !locks_to_reload.contains(&e.id) {
if !is_generated_from_raw_requirements(Some(language), &lock) {
if !is_generated_from_raw_requirements(&Some(language), &lock) {
let relative_imports = get_imports();
tx = dependency_map
.patch(relative_imports.clone(), e.id.clone(), tx)
@@ -1463,7 +1445,7 @@ async fn lock_modules<'c>(
if lock.as_ref().is_some_and(|x| !x.trim().is_empty()) {
let skip_creating_new_lock = skip_creating_new_lock(&language, &content);
if skip_creating_new_lock {
if !is_generated_from_raw_requirements(Some(language), &lock) {
if !is_generated_from_raw_requirements(&Some(language), &lock) {
let relative_imports = get_imports();
tx = dependency_map
.patch(relative_imports.clone(), e.id.clone(), tx)
@@ -2582,8 +2564,6 @@ async fn ansible_dep(
serde_json::to_string(&ansible_lockfile).map_err(|e| e.into())
}
pub const LOCKFILE_GENERATED_FROM_REQUIREMENTS_TXT: &str = "# from requirements.txt";
async fn capture_dependency_job(
job_id: &Uuid,
job_language: &ScriptLang,
@@ -2672,7 +2652,11 @@ async fn capture_dependency_job(
.await
.map(|res| {
if raw_deps {
format!("{}\n{}", LOCKFILE_GENERATED_FROM_REQUIREMENTS_TXT, res)
format!(
"{}\n{}",
windmill_common::lockfiles::LOCKFILE_GENERATED_FROM_REQUIREMENTS_TXT,
res
)
} else {
res
}