fix(flows): fix relative imports cache invalidation (#6519)

* fix(flows): fix relative imports cache invalidation Signed-off-by: pyranota <pyra@duck.com> * update flow_versions in the end Signed-off-by: pyranota <pyra@duck.com> * comms + fallback Signed-off-by: pyranota <pyra@duck.com> * add warn if fallback is used Signed-off-by: pyranota <pyra@duck.com> * include referenced scripts' hashes in flow_node id Signed-off-by: pyranota <pyra@duck.com> * improve fallback + TODOs Signed-off-by: pyranota <pyra@duck.com> * perf: optimize trigger Signed-off-by: pyranota <pyra@duck.com> * reduce TODOs Signed-off-by: pyranota <pyra@duck.com> * remove more TODOs Signed-off-by: pyranota <pyra@duck.com> * test thing Signed-off-by: pyranota <pyra@duck.com> * remove TODO Signed-off-by: pyranota <pyra@duck.com> * drop old flow trigger Signed-off-by: pyranota <pyra@duck.com> * add comments to migration Signed-off-by: pyranota <pyra@duck.com> --------- Signed-off-by: pyranota <pyra@duck.com>
2026-08-18 16:02:10 +00:00 · 2025-09-05 18:06:51 +02:00
parent 27b365b58d
commit d0982432e2
12 changed files with 224 additions and 16 deletions
@@ -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"
+}
@@ -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"
+}
@@ -15,7 +15,7 @@
      ]
    },
    "nullable": [
-      null
+      true
    ]
  },
  "hash": "5a219a2532517869578c4504ff3153c43903f929ae5d62fbba12610f89c36d55"
@@ -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"
+}
@@ -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"
+}
@@ -0,0 +1,2 @@
+-- Add down migration script here
+DROP TRIGGER flow_versions_append_trigger ON flow;
@@ -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');
@@ -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');
+
@@ -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;
@@ -718,6 +718,7 @@ async fn update_flow(
 ) -> Result<String> {
    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
@@ -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:
@@ -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<String, Box<RawValue>> = 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<Box<RawValue>>>,
+    language: Option<ScriptLang>,
 ) -> 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,