From d0982432e2a12e272b19ed7b0be6bd9a5c387abb Mon Sep 17 00:00:00 2001 From: pyranota <92104930+pyranota@users.noreply.github.com> Date: Fri, 5 Sep 2025 18:06:51 +0200 Subject: [PATCH] fix(flows): fix relative imports cache invalidation (#6519) * fix(flows): fix relative imports cache invalidation Signed-off-by: pyranota * update flow_versions in the end Signed-off-by: pyranota * comms + fallback Signed-off-by: pyranota * add warn if fallback is used Signed-off-by: pyranota * include referenced scripts' hashes in flow_node id Signed-off-by: pyranota * improve fallback + TODOs Signed-off-by: pyranota * perf: optimize trigger Signed-off-by: pyranota * reduce TODOs Signed-off-by: pyranota * remove more TODOs Signed-off-by: pyranota * test thing Signed-off-by: pyranota * remove TODO Signed-off-by: pyranota * drop old flow trigger Signed-off-by: pyranota * add comments to migration Signed-off-by: pyranota --------- Signed-off-by: pyranota --- ...f0402b908050b06cbc8d8987469217f59dbcf.json | 23 ++++ ...192b7373de62395626a18ad3672b604e23788.json | 16 +++ ...153c43903f929ae5d62fbba12610f89c36d55.json | 2 +- ...095fa1a373c2f9fdda11ce30fec6dda362d3d.json | 25 ++++ ...98a5e50bc98a8360387bd4158f9bc1289319f.json | 24 ++++ ...65240_flow_versions_update_notify.down.sql | 2 + ...2165240_flow_versions_update_notify.up.sql | 8 ++ ...05155321_drop_flow_update_trigger.down.sql | 6 + ...0905155321_drop_flow_update_trigger.up.sql | 2 + backend/windmill-api/src/flows.rs | 8 +- backend/windmill-worker/loader.py | 2 +- .../windmill-worker/src/worker_lockfiles.rs | 122 ++++++++++++++++-- 12 files changed, 224 insertions(+), 16 deletions(-) create mode 100644 backend/.sqlx/query-0a026af1b84c3724cadcd162422f0402b908050b06cbc8d8987469217f59dbcf.json create mode 100644 backend/.sqlx/query-15174732d4b52510d1ea73ce5d6192b7373de62395626a18ad3672b604e23788.json create mode 100644 backend/.sqlx/query-9ee36b224e931f24b4d48b22de0095fa1a373c2f9fdda11ce30fec6dda362d3d.json create mode 100644 backend/.sqlx/query-bf252ea52aeb57664ad5054b39998a5e50bc98a8360387bd4158f9bc1289319f.json create mode 100644 backend/migrations/20250902165240_flow_versions_update_notify.down.sql create mode 100644 backend/migrations/20250902165240_flow_versions_update_notify.up.sql create mode 100644 backend/migrations/20250905155321_drop_flow_update_trigger.down.sql create mode 100644 backend/migrations/20250905155321_drop_flow_update_trigger.up.sql diff --git a/backend/.sqlx/query-0a026af1b84c3724cadcd162422f0402b908050b06cbc8d8987469217f59dbcf.json b/backend/.sqlx/query-0a026af1b84c3724cadcd162422f0402b908050b06cbc8d8987469217f59dbcf.json new file mode 100644 index 0000000000..08429a6b6f --- /dev/null +++ b/backend/.sqlx/query-0a026af1b84c3724cadcd162422f0402b908050b06cbc8d8987469217f59dbcf.json @@ -0,0 +1,23 @@ +{ + "db_name": "PostgreSQL", + "query": "INSERT INTO flow_version\n (workspace_id, path, value, schema, created_by)\n\n SELECT workspace_id, path, value, schema, created_by\n FROM flow_version WHERE path = $1 AND workspace_id = $2\n\n RETURNING id", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "id", + "type_info": "Int8" + } + ], + "parameters": { + "Left": [ + "Text", + "Text" + ] + }, + "nullable": [ + false + ] + }, + "hash": "0a026af1b84c3724cadcd162422f0402b908050b06cbc8d8987469217f59dbcf" +} diff --git a/backend/.sqlx/query-15174732d4b52510d1ea73ce5d6192b7373de62395626a18ad3672b604e23788.json b/backend/.sqlx/query-15174732d4b52510d1ea73ce5d6192b7373de62395626a18ad3672b604e23788.json new file mode 100644 index 0000000000..82c7470a16 --- /dev/null +++ b/backend/.sqlx/query-15174732d4b52510d1ea73ce5d6192b7373de62395626a18ad3672b604e23788.json @@ -0,0 +1,16 @@ +{ + "db_name": "PostgreSQL", + "query": "UPDATE flow SET versions = array_append(versions, $1) WHERE path = $2 AND workspace_id = $3 AND NOT (versions @> ARRAY[$1::bigint])", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Int8", + "Text", + "Text" + ] + }, + "nullable": [] + }, + "hash": "15174732d4b52510d1ea73ce5d6192b7373de62395626a18ad3672b604e23788" +} diff --git a/backend/.sqlx/query-5a219a2532517869578c4504ff3153c43903f929ae5d62fbba12610f89c36d55.json b/backend/.sqlx/query-5a219a2532517869578c4504ff3153c43903f929ae5d62fbba12610f89c36d55.json index 713ccb9dd3..36ddb8ab9f 100644 --- a/backend/.sqlx/query-5a219a2532517869578c4504ff3153c43903f929ae5d62fbba12610f89c36d55.json +++ b/backend/.sqlx/query-5a219a2532517869578c4504ff3153c43903f929ae5d62fbba12610f89c36d55.json @@ -15,7 +15,7 @@ ] }, "nullable": [ - null + true ] }, "hash": "5a219a2532517869578c4504ff3153c43903f929ae5d62fbba12610f89c36d55" diff --git a/backend/.sqlx/query-9ee36b224e931f24b4d48b22de0095fa1a373c2f9fdda11ce30fec6dda362d3d.json b/backend/.sqlx/query-9ee36b224e931f24b4d48b22de0095fa1a373c2f9fdda11ce30fec6dda362d3d.json new file mode 100644 index 0000000000..1743998e54 --- /dev/null +++ b/backend/.sqlx/query-9ee36b224e931f24b4d48b22de0095fa1a373c2f9fdda11ce30fec6dda362d3d.json @@ -0,0 +1,25 @@ +{ + "db_name": "PostgreSQL", + "query": "\n UPDATE \n flow \n SET\n path = $1,\n summary = $2,\n description = $3,\n dependency_job = NULL,\n lock_error_logs = '',\n draft_only = NULL,\n tag = $4,\n dedicated_worker = $5,\n visible_to_runner_only = $6,\n on_behalf_of_email = $7,\n value = $8,\n schema = $9::text::json,\n edited_by = $10,\n edited_at = now()\n WHERE \n path = $11 AND workspace_id = $12", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Varchar", + "Text", + "Text", + "Varchar", + "Bool", + "Bool", + "Text", + "Jsonb", + "Text", + "Varchar", + "Text", + "Text" + ] + }, + "nullable": [] + }, + "hash": "9ee36b224e931f24b4d48b22de0095fa1a373c2f9fdda11ce30fec6dda362d3d" +} diff --git a/backend/.sqlx/query-bf252ea52aeb57664ad5054b39998a5e50bc98a8360387bd4158f9bc1289319f.json b/backend/.sqlx/query-bf252ea52aeb57664ad5054b39998a5e50bc98a8360387bd4158f9bc1289319f.json new file mode 100644 index 0000000000..97929c71e3 --- /dev/null +++ b/backend/.sqlx/query-bf252ea52aeb57664ad5054b39998a5e50bc98a8360387bd4158f9bc1289319f.json @@ -0,0 +1,24 @@ +{ + "db_name": "PostgreSQL", + "query": "INSERT INTO flow_version\n (workspace_id, path, value, schema, created_by)\n\n SELECT workspace_id, path, value, schema, created_by\n FROM flow_version WHERE path = $1 AND workspace_id = $2 AND id = $3\n\n RETURNING id", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "id", + "type_info": "Int8" + } + ], + "parameters": { + "Left": [ + "Text", + "Text", + "Int8" + ] + }, + "nullable": [ + false + ] + }, + "hash": "bf252ea52aeb57664ad5054b39998a5e50bc98a8360387bd4158f9bc1289319f" +} diff --git a/backend/migrations/20250902165240_flow_versions_update_notify.down.sql b/backend/migrations/20250902165240_flow_versions_update_notify.down.sql new file mode 100644 index 0000000000..c872c966ff --- /dev/null +++ b/backend/migrations/20250902165240_flow_versions_update_notify.down.sql @@ -0,0 +1,2 @@ +-- Add down migration script here +DROP TRIGGER flow_versions_append_trigger ON flow; diff --git a/backend/migrations/20250902165240_flow_versions_update_notify.up.sql b/backend/migrations/20250902165240_flow_versions_update_notify.up.sql new file mode 100644 index 0000000000..cec0b7065c --- /dev/null +++ b/backend/migrations/20250902165240_flow_versions_update_notify.up.sql @@ -0,0 +1,8 @@ +-- Trigger cache invalidation whenever the .versions are updated in `flow` +-- This is ultimate way of instant cache invalidation. If the notification is being lost, lazy expiration timer will work as a fallback +CREATE TRIGGER flow_versions_append_trigger +AFTER UPDATE ON flow +FOR EACH ROW +-- Trigger only when versions are different +WHEN (NEW.versions[array_upper(NEW.versions,1)] IS DISTINCT FROM OLD.versions[array_upper(OLD.versions,1)]) +EXECUTE FUNCTION notify_runnable_version_change('flow'); diff --git a/backend/migrations/20250905155321_drop_flow_update_trigger.down.sql b/backend/migrations/20250905155321_drop_flow_update_trigger.down.sql new file mode 100644 index 0000000000..a9abb8f081 --- /dev/null +++ b/backend/migrations/20250905155321_drop_flow_update_trigger.down.sql @@ -0,0 +1,6 @@ +-- Copied from runnable_version_notify.up.sql +CREATE TRIGGER flow_update_trigger +AFTER INSERT ON flow_version +FOR EACH ROW +EXECUTE FUNCTION notify_runnable_version_change('flow'); + diff --git a/backend/migrations/20250905155321_drop_flow_update_trigger.up.sql b/backend/migrations/20250905155321_drop_flow_update_trigger.up.sql new file mode 100644 index 0000000000..07a3604545 --- /dev/null +++ b/backend/migrations/20250905155321_drop_flow_update_trigger.up.sql @@ -0,0 +1,2 @@ +-- drop flow_update_trigger. We introduce new and more precise trigger in previous migration +DROP TRIGGER flow_update_trigger ON flow_version; diff --git a/backend/windmill-api/src/flows.rs b/backend/windmill-api/src/flows.rs index c8f861b561..ebc4e680e2 100644 --- a/backend/windmill-api/src/flows.rs +++ b/backend/windmill-api/src/flows.rs @@ -718,6 +718,7 @@ async fn update_flow( ) -> Result { let flow_path = flow_path.to_path(); check_scopes(&authed, || format!("flows:write:{}", flow_path))?; + #[cfg(not(feature = "enterprise"))] if nf .value @@ -732,7 +733,6 @@ async fn update_flow( } let authed = maybe_refresh_folders(&flow_path, &w_id, authed, &db).await; - let mut tx = user_db.clone().begin(&authed).await?; check_schedule_conflict(&mut tx, &w_id, flow_path).await?; @@ -745,10 +745,9 @@ async fn update_flow( ) .fetch_optional(&mut *tx) .await?; + let old_dep_job = not_found_if_none(old_dep_job, "Flow", flow_path)?; - let is_new_path = nf.path != flow_path; - let schema_str = schema.and_then(|x| serde_json::to_string(&x).ok()); sqlx::query!( @@ -965,7 +964,7 @@ async fn update_flow( JobPayload::FlowDependencies { path: nf.path.clone(), dedicated_worker: nf.dedicated_worker, - version: version, + version, }, windmill_queue::PushArgs { args: &args, extra: None }, &authed.username, @@ -1003,6 +1002,7 @@ async fn update_flow( "Error updating flow due to updating dependency job field: {e:#}" )) })?; + if let Some(old_dep_job) = old_dep_job { sqlx::query!( "UPDATE v2_job_queue SET diff --git a/backend/windmill-worker/loader.py b/backend/windmill-worker/loader.py index 646ec64029..74c378e6ec 100644 --- a/backend/windmill-worker/loader.py +++ b/backend/windmill-worker/loader.py @@ -24,7 +24,7 @@ class WindmillFinder(MetaPathFinder): if splitted[0] != "f" and splitted[0] != "u": return None - l = len(splitted) + l = len(splitted) # noqa: E741 if l <= 2: return ModuleSpec(name, WindmillLoader(name)) elif l > 2: diff --git a/backend/windmill-worker/src/worker_lockfiles.rs b/backend/windmill-worker/src/worker_lockfiles.rs index ed417dcfec..f9a882b2fc 100644 --- a/backend/windmill-worker/src/worker_lockfiles.rs +++ b/backend/windmill-worker/src/worker_lockfiles.rs @@ -41,6 +41,7 @@ use windmill_queue::{append_logs, CanceledBy, MiniPulledJob, PushIsolationLevel} lazy_static::lazy_static! { // TODO: To be removed in future versions static ref WMDEBUG_NO_HASH_CHANGE_ON_DJ: bool = std::env::var("WMDEBUG_NO_HASH_CHANGE_ON_DJ").is_ok(); + static ref WMDEBUG_NO_NEW_FLOW_VERSION_ON_DJ: bool = std::env::var("WMDEBUG_NO_NEW_FLOW_VERSION_ON_DJ").is_ok(); } use crate::common::OccupancyMetrics; @@ -645,6 +646,7 @@ pub async fn trigger_dependents_to_recompute_dependencies( if already_visited.contains(&s.importer_path) { continue; } + let tx = PushIsolationLevel::IsolatedRoot(db.clone()); let mut args: HashMap> = HashMap::new(); if let Some(ref dm) = deployment_message { @@ -669,6 +671,8 @@ pub async fn trigger_dependents_to_recompute_dependencies( get_latest_deployed_hash_for_path(None, db.clone(), w_id, s.importer_path.as_str()) .await; match r { + // We will create Dependency job as is. But the Dep Job Handler will detect that the job originates + // from [[trigger_dependents_to_recompute_dependencies]] and will create new script with new hash instead Ok(r) => JobPayload::Dependencies { path: s.importer_path.clone(), hash: ScriptHash(r.hash), @@ -685,28 +689,81 @@ pub async fn trigger_dependents_to_recompute_dependencies( } } } else if kind == "flow" { + // Unlike 'script', 'flow' will not delegate redeployment of new flow to the Dep Job Handler. + // We will create new flow in-place. + // It would be harder to do otherwise. + + // Create transaction to make operation atomic. + let mut flow_tx = db.begin().await?; args.insert( "nodes_to_relock".to_string(), to_raw_value(&s.importer_node_ids), ); + let r = sqlx::query_scalar!( "SELECT versions[array_upper(versions, 1)] FROM flow WHERE path = $1 AND workspace_id = $2", s.importer_path, w_id, - ).fetch_one(db) + ).fetch_one(&mut *flow_tx) .await .map_err(to_anyhow); + match r { - Ok(Some(version)) => JobPayload::FlowDependencies { - path: s.importer_path.clone(), - dedicated_worker: None, - version, - }, + // TODO: Fallback - remove eventually. + Ok(Some(version)) if *WMDEBUG_NO_NEW_FLOW_VERSION_ON_DJ => { + tracing::warn!("WMDEBUG_NO_NEW_FLOW_VERSION_ON_DJ usually should not be used. Behavior might be unstable. Please contact Windmill Team for support."); + JobPayload::FlowDependencies { + path: s.importer_path.clone(), + dedicated_worker: None, + version, + } + } + // Get current version of current flow. + Ok(Some(cur_version)) => { + // Find out what would be the next version. + // Also clone current flow_version to get new_version (which is usually c_v + 1). + // NOTE: It is fine if something goes wrong downstream and `flow` is not being appended with this new version. + // This version will just remain in db and cause no trouble. + let new_version = sqlx::query_scalar!( + "INSERT INTO flow_version + (workspace_id, path, value, schema, created_by) + + SELECT workspace_id, path, value, schema, created_by + FROM flow_version WHERE path = $1 AND workspace_id = $2 AND id = $3 + + RETURNING id", + &s.importer_path, + w_id, + cur_version + ) + .fetch_one(&mut *flow_tx) + .await + .map_err(|e| { + error::Error::internal_err(format!( + "Error updating flow due to flow history insert: {e:#}" + )) + })?; + + // Commit the transaction. + // NOTE: + // We do not append flow.versions with new version. + // We will do this in the end of the dependency job handler. + // Otherwise it might become a source of race-conditions. + flow_tx.commit().await?; + JobPayload::FlowDependencies { + path: s.importer_path.clone(), + dedicated_worker: None, + // Point Dep Job to the new version. + // We do this since we want to assume old ones are immutable. + version: new_version, + } + } Ok(None) => { tracing::error!( "no flow version found for path {path}", path = s.importer_path ); + // Do not commit the transaction. It will be dropped and rollbacked continue; } Err(err) => { @@ -714,6 +771,7 @@ pub async fn trigger_dependents_to_recompute_dependencies( "error getting latest deployed flow version for path {path}: {err}", path = s.importer_path, ); + // Do not commit the transaction. It will be dropped and rollbacked continue; } } @@ -845,9 +903,9 @@ pub async fn handle_flow_dependency_job( .clone(); let mut tx = db.begin().await?; - tx = clear_dependency_parent_path(&parent_path, &job_path, &job.workspace_id, "flow", tx) .await?; + if !skip_flow_update { sqlx::query!( "DELETE FROM workspace_runnable_dependencies WHERE flow_path = $1 AND workspace_id = $2", @@ -857,6 +915,7 @@ pub async fn handle_flow_dependency_job( .execute(&mut *tx) .await?; } + clear_asset_usage(&mut *tx, &job.workspace_id, &job_path, AssetUsageKind::Flow).await?; let modified_ids; @@ -880,6 +939,7 @@ pub async fn handle_flow_dependency_job( raw_deps, ) .await?; + if !errors.is_empty() { let error_message = errors .iter() @@ -895,6 +955,7 @@ pub async fn handle_flow_dependency_job( .await? .flatten() .unwrap_or_else(|| "no logs".to_string()); + sqlx::query!( "UPDATE flow SET lock_error_logs = $1 WHERE path = $2 AND workspace_id = $3", &format!("{logs2}\n{error_message}"), @@ -903,6 +964,7 @@ pub async fn handle_flow_dependency_job( ) .execute(db) .await?; + return Err(Error::ExecutionErr(format!( "Error locking flow modules:\n{}\n\nlogs:\n{}", error_message, @@ -917,6 +979,7 @@ pub async fn handle_flow_dependency_job( .execute(db) .await?; } + let new_flow_value = Json(serde_json::value::to_raw_value(&flow).map_err(to_anyhow)?); // Re-check cancellation to ensure we don't accidentally override a flow. @@ -968,6 +1031,7 @@ pub async fn handle_flow_dependency_job( flow.same_worker, ) .await?; + sqlx::query!( "INSERT INTO flow_version_lite (id, value) VALUES ($1, $2) ON CONFLICT (id) DO UPDATE SET value = EXCLUDED.value", @@ -977,6 +1041,15 @@ pub async fn handle_flow_dependency_job( .execute(&mut *tx) .await?; + // Making new version viewable as the current one. + // This will also trigger `flow_versions_append_trigger` (check _flow_versions_update_notify.up.sql) + // which will invalidate cache for the latest flow versions for all workers. + sqlx::query!("UPDATE flow SET versions = array_append(versions, $1) WHERE path = $2 AND workspace_id = $3", + version, + &job_path, + &job.workspace_id, + ).execute(&mut *tx).await?; + tx.commit().await?; if let Err(e) = handle_deployment_metadata( @@ -1408,12 +1481,33 @@ async fn insert_flow_node<'c>( code: Option<&String>, lock: Option<&String>, flow: Option<&Json>>, + language: Option, ) -> Result<(sqlx::Transaction<'c, sqlx::Postgres>, FlowNodeId)> { let hash = { let mut hasher = sha2::Sha256::new(); hasher.update(code.unwrap_or(&Default::default())); hasher.update(lock.unwrap_or(&Default::default())); hasher.update(flow.unwrap_or(&Default::default()).get()); + + if !*WMDEBUG_NO_NEW_FLOW_VERSION_ON_DJ { + if let Some(imports) = extract_relative_imports( + code.map(|s| s.as_str()).unwrap_or_default(), + path, + &language, + ) { + // We also want to take into account hashes of relative imports. + // TODO: May be use bytemuck or cast it in different, more proper way. + hasher.update(&format!( + "{:?}", + sqlx::query_scalar::<_, i64>( + "SELECT hash FROM script WHERE path = ANY($1) AND archived = false" + ) + .bind(imports) + .fetch_all(&mut *tx) + .await? + )); + } + } format!("{:x}", hasher.finalize()) }; @@ -1506,6 +1600,7 @@ async fn insert_flow_modules<'c>( same_worker, ..Default::default() }))), + None, ) .await?; *modules_node = Some(id); @@ -1550,9 +1645,16 @@ async fn reduce_flow<'c>( unreachable!() }; let id; - (tx, id) = - insert_flow_node(tx, path, workspace_id, Some(&content), lock.as_ref(), None) - .await?; + (tx, id) = insert_flow_node( + tx, + path, + workspace_id, + Some(&content), + lock.as_ref(), + None, + Some(language), + ) + .await?; val = FlowScript { input_transforms, id,