diff --git a/backend/.sqlx/query-3cc7ecab48c379cd845b22012ef1fe1573fc332c9f1c44960084f3784ddb3f54.json b/backend/.sqlx/query-3cc7ecab48c379cd845b22012ef1fe1573fc332c9f1c44960084f3784ddb3f54.json new file mode 100644 index 0000000000..6ba020e059 --- /dev/null +++ b/backend/.sqlx/query-3cc7ecab48c379cd845b22012ef1fe1573fc332c9f1c44960084f3784ddb3f54.json @@ -0,0 +1,22 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT logs FROM job_logs WHERE created_at > $1", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "logs", + "type_info": "Text" + } + ], + "parameters": { + "Left": [ + "Timestamptz" + ] + }, + "nullable": [ + true + ] + }, + "hash": "3cc7ecab48c379cd845b22012ef1fe1573fc332c9f1c44960084f3784ddb3f54" +} diff --git a/backend/.sqlx/query-55a7460901c8ccdb478a2b7a3d4bd9d367838fe80844b5a14d2ef66452f36297.json b/backend/.sqlx/query-55a7460901c8ccdb478a2b7a3d4bd9d367838fe80844b5a14d2ef66452f36297.json new file mode 100644 index 0000000000..a1f3a7d6ba --- /dev/null +++ b/backend/.sqlx/query-55a7460901c8ccdb478a2b7a3d4bd9d367838fe80844b5a14d2ef66452f36297.json @@ -0,0 +1,19 @@ +{ + "db_name": "PostgreSQL", + "query": "INSERT INTO dependency_map (workspace_id, importer_path, importer_kind, imported_path, importer_node_id, imported_lockfile_hash)\n VALUES ($1, $2, $3::text::IMPORTER_KIND, $4, $5, $6)\n ON CONFLICT (workspace_id, importer_node_id, importer_kind, importer_path, imported_path)\n DO UPDATE SET imported_lockfile_hash = EXCLUDED.imported_lockfile_hash", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Varchar", + "Varchar", + "Text", + "Varchar", + "Varchar", + "Int8" + ] + }, + "nullable": [] + }, + "hash": "55a7460901c8ccdb478a2b7a3d4bd9d367838fe80844b5a14d2ef66452f36297" +} 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-5f6d0d1b24693db22230ec8322b3174ecc195242a1205131f5d0d32148c0280a.json b/backend/.sqlx/query-5f6d0d1b24693db22230ec8322b3174ecc195242a1205131f5d0d32148c0280a.json new file mode 100644 index 0000000000..9ffceac7b9 --- /dev/null +++ b/backend/.sqlx/query-5f6d0d1b24693db22230ec8322b3174ecc195242a1205131f5d0d32148c0280a.json @@ -0,0 +1,20 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT hash FROM script WHERE path = 'f/rel/leaf_2' AND workspace_id = 'test-workspace' AND archived = false ORDER BY created_at DESC LIMIT 1", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "hash", + "type_info": "Int8" + } + ], + "parameters": { + "Left": [] + }, + "nullable": [ + false + ] + }, + "hash": "5f6d0d1b24693db22230ec8322b3174ecc195242a1205131f5d0d32148c0280a" +} diff --git a/backend/.sqlx/query-6c10c38a5d5e560d8d76159ce0ae118cf7191b06d72285e71256dd6d70c19451.json b/backend/.sqlx/query-8acc8c47123af84d2308042dcd34d04b941e928c7621e66d3bdc510c335f8c20.json similarity index 62% rename from backend/.sqlx/query-6c10c38a5d5e560d8d76159ce0ae118cf7191b06d72285e71256dd6d70c19451.json rename to backend/.sqlx/query-8acc8c47123af84d2308042dcd34d04b941e928c7621e66d3bdc510c335f8c20.json index 1e49b8405c..a450218ccd 100644 --- a/backend/.sqlx/query-6c10c38a5d5e560d8d76159ce0ae118cf7191b06d72285e71256dd6d70c19451.json +++ b/backend/.sqlx/query-8acc8c47123af84d2308042dcd34d04b941e928c7621e66d3bdc510c335f8c20.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "\n INSERT INTO workspace_dependencies(name, workspace_id, content, language, description)\n VALUES ($1, $2, $3, $4, $5)\n RETURNING id\n ", + "query": "\n WITH ins AS (\n INSERT INTO workspace_dependencies(name, workspace_id, content, language, description)\n VALUES ($1, $2, $3, $4, $5)\n RETURNING id\n ), lock_ins AS (\n INSERT INTO lock_hash (workspace_id, path, lockfile_hash)\n VALUES ($2, $6, $7)\n ON CONFLICT (workspace_id, path) DO UPDATE SET lockfile_hash = $7\n )\n SELECT id FROM ins\n ", "describe": { "columns": [ { @@ -46,12 +46,14 @@ } } }, - "Text" + "Text", + "Varchar", + "Int8" ] }, "nullable": [ false ] }, - "hash": "6c10c38a5d5e560d8d76159ce0ae118cf7191b06d72285e71256dd6d70c19451" + "hash": "8acc8c47123af84d2308042dcd34d04b941e928c7621e66d3bdc510c335f8c20" } diff --git a/backend/.sqlx/query-9ca267c5a88f0cf3a4821e7514fe585131a42136c522a0fdde47c63cc0c6bc5a.json b/backend/.sqlx/query-9ca267c5a88f0cf3a4821e7514fe585131a42136c522a0fdde47c63cc0c6bc5a.json new file mode 100644 index 0000000000..995bf49b52 --- /dev/null +++ b/backend/.sqlx/query-9ca267c5a88f0cf3a4821e7514fe585131a42136c522a0fdde47c63cc0c6bc5a.json @@ -0,0 +1,24 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT\n COUNT(*) > 0\n AND BOOL_AND(\n CASE\n -- For dependencies/: use IS NOT DISTINCT FROM (NULL = NULL is true for legacy)\n -- And the reason for this, is that for every script we also add dependency on default workspace dependencies by default\n -- that default/unnamed workspace dependencies may not exist, but we still do this.\n -- it is needed for windmill to know what to redeploy when default workspace dependencies are being added\n\n -- naturally for non-existant entries we have no hash of it\n -- so if we compared hash with '=' (instead of IS NOT DISTINCT FROM), it would give false on NULL = NULL,\n -- which would mean that this entire query returns false, which means relock skip cannot happen\n\n -- The solution is to say if both: referenced and current hash are NULLs we treat it as true, so it becomes no longer a blocker from skip.\n --\n -- It is backed up by the fact that server is responsible for deploying new workspace dependencies\n -- so if one was to deploy a new wdeps, server would assign it new hash, and this expression would be invalid and would not approve skip\n WHEN dm.imported_path LIKE 'dependencies/%'\n THEN dm.imported_lockfile_hash IS NOT DISTINCT FROM lh.lockfile_hash\n\n -- For scripts: use = with COALESCE (NULL = NULL becomes false)\n -- unlike w deps, we can't do IS NOT DISTINCT FROM here\n -- the reason is that scripts deployments are issued by other workers instead of the server\n -- which would mean that there is no guarantee that new d job will also write it's lock's hash to the `lock_hash`\n -- which could lead to false positives\n ELSE COALESCE(dm.imported_lockfile_hash = lh.lockfile_hash, false)\n END\n )\n FROM dependency_map dm\n LEFT JOIN lock_hash lh\n ON lh.workspace_id = dm.workspace_id\n AND lh.path = dm.imported_path\n WHERE dm.workspace_id = $1\n AND dm.importer_path = $2\n AND dm.importer_node_id = $3", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "?column?", + "type_info": "Bool" + } + ], + "parameters": { + "Left": [ + "Text", + "Text", + "Text" + ] + }, + "nullable": [ + null + ] + }, + "hash": "9ca267c5a88f0cf3a4821e7514fe585131a42136c522a0fdde47c63cc0c6bc5a" +} diff --git a/backend/.sqlx/query-ead84a63cb965e36155605434c9e3670e50344f39b0d61eef8e60c2cdb57ae1f.json b/backend/.sqlx/query-c8dee3bd1ce03d8ee3c05345a0bdd5bdbd75396fe5bc24fd2d42edc79261169c.json similarity index 64% rename from backend/.sqlx/query-ead84a63cb965e36155605434c9e3670e50344f39b0d61eef8e60c2cdb57ae1f.json rename to backend/.sqlx/query-c8dee3bd1ce03d8ee3c05345a0bdd5bdbd75396fe5bc24fd2d42edc79261169c.json index 499e2e6730..9227c072bd 100644 --- a/backend/.sqlx/query-ead84a63cb965e36155605434c9e3670e50344f39b0d61eef8e60c2cdb57ae1f.json +++ b/backend/.sqlx/query-c8dee3bd1ce03d8ee3c05345a0bdd5bdbd75396fe5bc24fd2d42edc79261169c.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "\n DELETE FROM dependency_map\n WHERE workspace_id = $1\n AND importer_path = $2\n AND importer_kind = $3::text::IMPORTER_KIND\n AND importer_node_id = $4\n AND imported_path = $5\n ", + "query": "\n DELETE FROM dependency_map\n WHERE workspace_id = $1\n AND importer_path = $2\n AND importer_kind = $3::text::IMPORTER_KIND\n AND importer_node_id = $4\n AND imported_path = $5\n AND imported_lockfile_hash IS NOT DISTINCT FROM $6 -- we don't want to delete other entries with other lockfile hash.\n ", "describe": { "columns": [], "parameters": { @@ -9,10 +9,11 @@ "Text", "Text", "Text", - "Text" + "Text", + "Int8" ] }, "nullable": [] }, - "hash": "ead84a63cb965e36155605434c9e3670e50344f39b0d61eef8e60c2cdb57ae1f" + "hash": "c8dee3bd1ce03d8ee3c05345a0bdd5bdbd75396fe5bc24fd2d42edc79261169c" } diff --git a/backend/.sqlx/query-cbbe20745c2b743acc814e8e77c0e4e154b511af676e085bbe88e72e7d14aee3.json b/backend/.sqlx/query-cbbe20745c2b743acc814e8e77c0e4e154b511af676e085bbe88e72e7d14aee3.json new file mode 100644 index 0000000000..b72346db9e --- /dev/null +++ b/backend/.sqlx/query-cbbe20745c2b743acc814e8e77c0e4e154b511af676e085bbe88e72e7d14aee3.json @@ -0,0 +1,23 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT lock FROM script WHERE path = $1 AND workspace_id = $2 AND lock IS NOT NULL\n AND deleted = false ORDER BY created_at DESC LIMIT 1", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "lock", + "type_info": "Text" + } + ], + "parameters": { + "Left": [ + "Text", + "Text" + ] + }, + "nullable": [ + true + ] + }, + "hash": "cbbe20745c2b743acc814e8e77c0e4e154b511af676e085bbe88e72e7d14aee3" +} diff --git a/backend/.sqlx/query-d697b7311430e7bd5375ec5494179f4071e3bfe123d9600249cfa80c1103edd8.json b/backend/.sqlx/query-d697b7311430e7bd5375ec5494179f4071e3bfe123d9600249cfa80c1103edd8.json deleted file mode 100644 index e35c4e2730..0000000000 --- a/backend/.sqlx/query-d697b7311430e7bd5375ec5494179f4071e3bfe123d9600249cfa80c1103edd8.json +++ /dev/null @@ -1,16 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "UPDATE script SET lock = $1 WHERE hash = $2 AND workspace_id = $3", - "describe": { - "columns": [], - "parameters": { - "Left": [ - "Text", - "Int8", - "Text" - ] - }, - "nullable": [] - }, - "hash": "d697b7311430e7bd5375ec5494179f4071e3bfe123d9600249cfa80c1103edd8" -} diff --git a/backend/.sqlx/query-e32d6c6ae4e0d824c4cf19128182d67f36d1fd87fe4cf4f002e18367088c497c.json b/backend/.sqlx/query-e32d6c6ae4e0d824c4cf19128182d67f36d1fd87fe4cf4f002e18367088c497c.json deleted file mode 100644 index 915cf7416b..0000000000 --- a/backend/.sqlx/query-e32d6c6ae4e0d824c4cf19128182d67f36d1fd87fe4cf4f002e18367088c497c.json +++ /dev/null @@ -1,18 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "INSERT INTO dependency_map (workspace_id, importer_path, importer_kind, imported_path, importer_node_id)\n VALUES ($1, $2, $3::text::IMPORTER_KIND, $4, $5) ON CONFLICT DO NOTHING", - "describe": { - "columns": [], - "parameters": { - "Left": [ - "Varchar", - "Varchar", - "Text", - "Varchar", - "Varchar" - ] - }, - "nullable": [] - }, - "hash": "e32d6c6ae4e0d824c4cf19128182d67f36d1fd87fe4cf4f002e18367088c497c" -} diff --git a/backend/.sqlx/query-f8c08997d7fd0158b4d3cc093fb0a2aadd4473e98873d5a7873f4fa944b99d58.json b/backend/.sqlx/query-f8c08997d7fd0158b4d3cc093fb0a2aadd4473e98873d5a7873f4fa944b99d58.json new file mode 100644 index 0000000000..45979be595 --- /dev/null +++ b/backend/.sqlx/query-f8c08997d7fd0158b4d3cc093fb0a2aadd4473e98873d5a7873f4fa944b99d58.json @@ -0,0 +1,18 @@ +{ + "db_name": "PostgreSQL", + "query": "WITH update_lock AS (\n UPDATE script SET lock = $1 WHERE hash = $2 AND workspace_id = $3\n )\n INSERT INTO lock_hash (workspace_id, path, lockfile_hash)\n VALUES ($3, $4, $5)\n ON CONFLICT (workspace_id, path) DO UPDATE SET lockfile_hash = $5", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Text", + "Int8", + "Text", + "Varchar", + "Int8" + ] + }, + "nullable": [] + }, + "hash": "f8c08997d7fd0158b4d3cc093fb0a2aadd4473e98873d5a7873f4fa944b99d58" +} diff --git a/backend/migrations/20260209000001_dependency_map_lockfile_hash.down.sql b/backend/migrations/20260209000001_dependency_map_lockfile_hash.down.sql new file mode 100644 index 0000000000..53c1828256 --- /dev/null +++ b/backend/migrations/20260209000001_dependency_map_lockfile_hash.down.sql @@ -0,0 +1,3 @@ +DROP TABLE IF EXISTS lock_hash; +DROP INDEX IF EXISTS dependency_map_importer_path_idx; +ALTER TABLE dependency_map DROP COLUMN IF EXISTS imported_lockfile_hash; diff --git a/backend/migrations/20260209000001_dependency_map_lockfile_hash.up.sql b/backend/migrations/20260209000001_dependency_map_lockfile_hash.up.sql new file mode 100644 index 0000000000..4c4203fab5 --- /dev/null +++ b/backend/migrations/20260209000001_dependency_map_lockfile_hash.up.sql @@ -0,0 +1,14 @@ +-- Add column to track lockfile hash of imported scripts +-- Used to skip re-locking when imports' lockfiles haven't changed +ALTER TABLE dependency_map ADD COLUMN imported_lockfile_hash BIGINT; + +-- Index for queries filtering by importer (skip-relock check, load(), clear_map_for_item) +CREATE INDEX IF NOT EXISTS dependency_map_importer_path_idx ON dependency_map (workspace_id, importer_path); + +-- Stores lockfile/content hashes to detect when imports' locks have changed +CREATE TABLE lock_hash ( + workspace_id VARCHAR(50) NOT NULL REFERENCES workspace(id) ON DELETE CASCADE, + path VARCHAR(255) NOT NULL, + lockfile_hash BIGINT NOT NULL, + PRIMARY KEY (workspace_id, path) +); diff --git a/backend/tests/relock_skip.rs b/backend/tests/relock_skip.rs new file mode 100644 index 0000000000..ce274796bc --- /dev/null +++ b/backend/tests/relock_skip.rs @@ -0,0 +1,378 @@ +use sqlx::{Pool, Postgres}; +use tokio_stream::StreamExt; +use windmill_api_client::types::NewScript; +use windmill_test_utils::*; + +mod relock_skip { + use super::*; + + fn quick_ns( + content: &str, + language: windmill_api_client::types::ScriptLang, + path: &str, + lock: Option, + parent_hash: Option, + ) -> NewScript { + NewScript { + content: content.into(), + language, + lock, + parent_hash, + path: path.into(), + concurrent_limit: None, + concurrency_time_window_s: None, + cache_ttl: None, + dedicated_worker: None, + description: "".to_string(), + draft_only: None, + envs: vec![], + is_template: None, + kind: None, + summary: "".to_string(), + tag: None, + schema: std::collections::HashMap::new(), + ws_error_handler_muted: Some(false), + priority: None, + delete_after_use: None, + timeout: None, + restart_unless_cancelled: None, + deployment_message: None, + concurrency_key: None, + visible_to_runner_only: None, + no_main_func: None, + codebase: None, + has_preprocessor: None, + on_behalf_of_email: None, + assets: vec![], + } + } + + async fn init(db: Pool) -> (windmill_api_client::Client, u16, ApiServer) { + init_client(db).await + } + + /// Counts occurrences of a pattern in job logs for all jobs created after a given time + async fn count_pattern_in_job_logs( + db: &Pool, + pattern: &str, + after: chrono::DateTime, + ) -> i64 { + let logs = sqlx::query_scalar!( + "SELECT logs FROM job_logs WHERE created_at > $1", + after + ) + .fetch_all(db) + .await + .unwrap(); + + logs.iter() + .filter_map(|l| l.as_ref()) + .map(|l| l.matches(pattern).count() as i64) + .sum() + } + + /// Waits for exactly N jobs to complete. Returns the timestamp before waiting. + async fn wait_for_jobs( + completed: &mut (impl futures::Stream + Unpin), + count: usize, + ) -> chrono::DateTime { + let before = chrono::Utc::now(); + for _ in 0..count { + completed.next().await; + } + before + } + + /// Waits for at least N jobs to complete, then drains any additional jobs + /// that complete within a short timeout. Returns the timestamp before waiting. + async fn wait_for_jobs_ge( + completed: &mut (impl futures::Stream + Unpin), + min_count: usize, + ) -> chrono::DateTime { + let before = chrono::Utc::now(); + for _ in 0..min_count { + completed.next().await; + } + // Drain any additional jobs that complete within 5 seconds + loop { + match tokio::time::timeout(std::time::Duration::from_secs(1), completed.next()).await { + Ok(Some(_)) => continue, + _ => break, + } + } + before + } + + #[cfg(feature = "python")] + #[sqlx::test(fixtures("base", "dependency_map"))] + async fn relock_skip_on_script_redeployment(db: Pool) -> anyhow::Result<()> { + std::env::set_var("DEPENDENCY_JOB_DEBOUNCE_DELAY", "0"); + let (client, port, _s) = init(db.clone()).await; + let mut completed = listen_for_completed_jobs(&db).await; + + in_test_worker(&db, async { + // Step 1: Redeploy leaf_1 - first time, no hashes exist, all should relock + let before = chrono::Utc::now(); + client + .create_script( + "test-workspace", + &quick_ns( + " +def main(): + return 'leaf1' + ", + windmill_api_client::types::ScriptLang::Python3, + "f/rel/leaf_1", + None, + Some("0000000000051658".into()), + ), + ) + .await + .unwrap(); + + // leaf_1(1) + branch(1) + root_script(2: leaf_1+branch) + root_flow(2) + root_app(2) = 8 jobs + wait_for_jobs(&mut completed, 8).await; + + let skipping_count = count_pattern_in_job_logs(&db, "Skipping relock", before).await; + let relocking_count = count_pattern_in_job_logs(&db, "Relocking", before).await; + assert_eq!(skipping_count, 0, "First deployment should not skip"); + assert!(relocking_count > 0, "First deployment should have relocking jobs"); + + // Step 2: Redeploy leaf_2 - first time for leaf_2, should relock + let before = chrono::Utc::now(); + client + .create_script( + "test-workspace", + &quick_ns( + " +def main(): + return 'leaf2' + ", + windmill_api_client::types::ScriptLang::Python3, + "f/rel/leaf_2", + None, + Some("0000000000051659".into()), + ), + ) + .await + .unwrap(); + + // leaf_2(1) + root_script(1) + root_flow(1) + root_app(1) = 4 jobs (no cascade, branch doesn't depend on leaf_2) + wait_for_jobs(&mut completed, 4).await; + + let skipping_count = count_pattern_in_job_logs(&db, "Skipping relock", before).await; + let relocking_count = count_pattern_in_job_logs(&db, "Relocking", before).await; + assert_eq!(skipping_count, 0, "leaf_2 first deployment should not skip"); + assert!(relocking_count > 0, "leaf_2 first deployment should have relocking jobs"); + + // Step 3: Redeploy leaf_2 with trivial change (comment) - lock stays same, should SKIP + // Get current parent hash for leaf_2 + let leaf2_hash = sqlx::query_scalar!( + "SELECT hash FROM script WHERE path = 'f/rel/leaf_2' AND workspace_id = 'test-workspace' AND archived = false ORDER BY created_at DESC LIMIT 1" + ) + .fetch_one(&db) + .await + .unwrap(); + + let before = chrono::Utc::now(); + client + .create_script( + "test-workspace", + &quick_ns( + " +# comment to change hash but not lock +def main(): + return 'leaf2' + ", + windmill_api_client::types::ScriptLang::Python3, + "f/rel/leaf_2", + None, + Some(format!("{:016X}", leaf2_hash)), + ), + ) + .await + .unwrap(); + + // Same as leaf_2 first deployment: 4 jobs + wait_for_jobs(&mut completed, 4).await; + + let skipping_count = count_pattern_in_job_logs(&db, "Skipping relock", before).await; + assert!(skipping_count > 0, "Trivial change (comment only) should skip - lock unchanged"); + + // Step 4: Redeploy leaf_2 with actual dependency change (add tiny via comment) - should NOT skip + let leaf2_hash = sqlx::query_scalar!( + "SELECT hash FROM script WHERE path = 'f/rel/leaf_2' AND workspace_id = 'test-workspace' AND archived = false ORDER BY created_at DESC LIMIT 1" + ) + .fetch_one(&db) + .await + .unwrap(); + + let before = chrono::Utc::now(); + client + .create_script( + "test-workspace", + &quick_ns( + " +# requirements: +# tiny + +def main(): + return 'leaf2 with tiny' + ", + windmill_api_client::types::ScriptLang::Python3, + "f/rel/leaf_2", + None, + Some(format!("{:016X}", leaf2_hash)), + ), + ) + .await + .unwrap(); + + // Same as leaf_2 first deployment: 4 jobs + wait_for_jobs(&mut completed, 4).await; + + let skipping_count = count_pattern_in_job_logs(&db, "Skipping relock", before).await; + assert_eq!(skipping_count, 0, "Changed dependencies should not skip"); + }, port).await; + + Ok(()) + } + + #[cfg(feature = "python")] + #[sqlx::test(fixtures("base", "dependency_map"))] + async fn relock_skip_on_workspace_deps_redeployment(db: Pool) -> anyhow::Result<()> { + use windmill_common::scripts::ScriptLang; + use windmill_dep_map::workspace_dependencies::NewWorkspaceDependencies; + + std::env::set_var("DEPENDENCY_JOB_DEBOUNCE_DELAY", "0"); + std::env::set_var("EXISTS_CACHE_TIMEOUT_MS", "0"); + + let (_client, port, _s) = init(db.clone()).await; + let mut completed = listen_for_completed_jobs(&db).await; + + // Step 1: Redeploy default (unnamed) workspace deps - first time, should relock + let before = chrono::Utc::now(); + NewWorkspaceDependencies { + workspace_id: "test-workspace".into(), + language: ScriptLang::Python3, + content: "".into(), + name: None, // Default/unnamed + description: None, + } + .create(("".to_owned(), "".to_owned(), "".to_owned()), db.clone()) + .await + .unwrap(); + + in_test_worker(&db, wait_for_jobs_ge(&mut completed, 10), port).await; + + // Note: within a cascade, the same script may be triggered multiple times. + // After the first trigger relocks and stores the hash, subsequent triggers skip. + // We allow up to 3 skips from cascade re-triggers. + let skipping_count = count_pattern_in_job_logs(&db, "Skipping relock", before).await; + let relocking_count = count_pattern_in_job_logs(&db, "Relocking", before).await; + assert!(skipping_count <= 3, "First deployment should have at most 3 skips from cascade"); + assert!(relocking_count >= 3, "First deployment should have at least 3 relocking jobs"); + + // Step 2: Redeploy default workspace deps again - should SKIP + let before = chrono::Utc::now(); + NewWorkspaceDependencies { + workspace_id: "test-workspace".into(), + language: ScriptLang::Python3, + content: "".into(), + name: None, + description: None, + } + .create(("".to_owned(), "".to_owned(), "".to_owned()), db.clone()) + .await + .unwrap(); + + in_test_worker(&db, wait_for_jobs_ge(&mut completed, 10), port).await; + + let skipping_count = count_pattern_in_job_logs(&db, "Skipping relock", before).await; + assert!(skipping_count >= 3, "Second deployment of same content should skip at least 3 times"); + + // Step 3: Redeploy default workspace deps with different content - should NOT skip + let before = chrono::Utc::now(); + NewWorkspaceDependencies { + workspace_id: "test-workspace".into(), + language: ScriptLang::Python3, + content: "tiny".into(), + name: None, + description: None, + } + .create(("".to_owned(), "".to_owned(), "".to_owned()), db.clone()) + .await + .unwrap(); + + in_test_worker(&db, wait_for_jobs_ge(&mut completed, 10), port).await; + + let skipping_count = count_pattern_in_job_logs(&db, "Skipping relock", before).await; + let relocking_count = count_pattern_in_job_logs(&db, "Relocking", before).await; + assert!(skipping_count <= 4, "Changed content should have at most 3 skips from cascade"); + assert!(relocking_count >= 3, "Changed content should trigger at least 3 relocking jobs"); + + // Step 4: Deploy named workspace deps first time - should relock (no hash exists yet) + // Named deps trigger exactly 3 independent objects with no cascade + let before = chrono::Utc::now(); + NewWorkspaceDependencies { + workspace_id: "test-workspace".into(), + language: ScriptLang::Python3, + content: "".into(), + name: Some("test".to_owned()), + description: None, + } + .create(("".to_owned(), "".to_owned(), "".to_owned()), db.clone()) + .await + .unwrap(); + + in_test_worker(&db, wait_for_jobs(&mut completed, 3), port).await; + + let skipping_count = count_pattern_in_job_logs(&db, "Skipping relock", before).await; + let relocking_count = count_pattern_in_job_logs(&db, "Relocking", before).await; + assert_eq!(skipping_count, 0, "Named workspace deps first deployment should not skip"); + assert!(relocking_count > 0, "Named workspace deps first deployment should relock"); + + // Step 5: Deploy named workspace deps again with no change - should SKIP + let before = chrono::Utc::now(); + NewWorkspaceDependencies { + workspace_id: "test-workspace".into(), + language: ScriptLang::Python3, + content: "".into(), + name: Some("test".to_owned()), + description: None, + } + .create(("".to_owned(), "".to_owned(), "".to_owned()), db.clone()) + .await + .unwrap(); + + in_test_worker(&db, wait_for_jobs(&mut completed, 3), port).await; + + let skipping_count = count_pattern_in_job_logs(&db, "Skipping relock", before).await; + let relocking_count = count_pattern_in_job_logs(&db, "Relocking", before).await; + + assert!(skipping_count > 0, "Named workspace deps second deployment should skip"); + assert_eq!(relocking_count, 0, "Named workspace deps second deployment should not relock"); + + // Step 6: Deploy named workspace deps with small change - should NOT skip + let before = chrono::Utc::now(); + NewWorkspaceDependencies { + workspace_id: "test-workspace".into(), + language: ScriptLang::Python3, + content: "tiny".into(), + name: Some("test".to_owned()), + description: None, + } + .create(("".to_owned(), "".to_owned(), "".to_owned()), db.clone()) + .await + .unwrap(); + + in_test_worker(&db, wait_for_jobs(&mut completed, 3), port).await; + + let skipping_count = count_pattern_in_job_logs(&db, "Skipping relock", before).await; + let relocking_count = count_pattern_in_job_logs(&db, "Relocking", before).await; + assert_eq!(skipping_count, 0, "Named workspace deps with change should not skip"); + assert!(relocking_count > 0, "Named workspace deps with change should relock"); + + Ok(()) + } +} diff --git a/backend/windmill-common/src/cache.rs b/backend/windmill-common/src/cache.rs index 11d6217fba..27af55ea01 100644 --- a/backend/windmill-common/src/cache.rs +++ b/backend/windmill-common/src/cache.rs @@ -909,9 +909,15 @@ pub mod workspace_dependencies { /// Cache key: (workspace_id, language) /// Cache value: (exists: bool, cached_at timestamp) static ref DEFAULT_WD_EXISTS_CACHE: quick_cache::sync::Cache<(String, String), (bool, Instant)> = quick_cache::sync::Cache::new(500); + + /// Cache timeout for existence checks (default 10 seconds, configurable via EXISTS_CACHE_TIMEOUT_MS env var) + pub static ref EXISTS_CACHE_TIMEOUT: Duration = Duration::from_millis( + std::env::var("EXISTS_CACHE_TIMEOUT_MS") + .ok() + .and_then(|v| v.parse().ok()) + .unwrap_or(10_000) + ); } - /// Cache timeout for existence checks (10 seconds) - pub const EXISTS_CACHE_TIMEOUT: Duration = Duration::from_secs(10); pub fn fetch_workspace_dependencies<'c>( id: i64, @@ -932,7 +938,7 @@ pub mod workspace_dependencies { let exists_key = (workspace_id.to_string(), dependencies_filename); if let Some((exists, cached_at)) = DEFAULT_WD_EXISTS_CACHE.get(&exists_key) { - if cached_at.elapsed() < EXISTS_CACHE_TIMEOUT { + if cached_at.elapsed() < *EXISTS_CACHE_TIMEOUT { tracing::debug!( workspace_id = %workspace_id, exists, diff --git a/backend/windmill-common/src/scripts.rs b/backend/windmill-common/src/scripts.rs index 2f87cb14b0..c6a4a787ea 100644 --- a/backend/windmill-common/src/scripts.rs +++ b/backend/windmill-common/src/scripts.rs @@ -395,8 +395,6 @@ pub async fn clone_script<'c>( content: s.content, schema: s.schema, is_template: s.is_template, - // TODO: Make it either None everywhere (particularly when raw reqs are calculated) - // Or handle this case and conditionally make Some (only with raw reqs) lock: None, language: s.language, kind: Some(s.kind), diff --git a/backend/windmill-dep-map/src/scoped_dependency_map.rs b/backend/windmill-dep-map/src/scoped_dependency_map.rs index 4770fd516d..987cc4d529 100644 --- a/backend/windmill-dep-map/src/scoped_dependency_map.rs +++ b/backend/windmill-dep-map/src/scoped_dependency_map.rs @@ -15,6 +15,9 @@ lazy_static::lazy_static! { pub static ref WMDEBUG_NO_DMAP_DISSOLVE: bool = std::env::var("WMDEBUG_NO_DMAP_DISSOLVE").is_ok(); } +type PathString = String; +type LockHash = i64; + #[derive(Serialize)] pub struct DependencyMap { pub workspace_id: Option, @@ -33,8 +36,8 @@ pub struct DependencyDependent { #[derive(Debug)] pub struct ScopedDependencyMap { - /// (importer_node_id, imported_path) - to_delete: HashSet<(String, String)>, + /// (importer_node_id, imported_path, imported_lockfile_hash) + to_delete: HashSet<(String, PathString, Option)>, w_id: String, importer_path: String, importer_kind: String, @@ -61,14 +64,14 @@ impl ScopedDependencyMap { parent_path.clone().unwrap_or_default(), ); - let dmap = sqlx::query_as::<_, (String, String)>( + let dmap = sqlx::query_as::<_, (String, String, Option)>( " UPDATE dependency_map SET importer_path = $1 WHERE importer_path = $2 AND importer_kind = $3::text::IMPORTER_KIND AND workspace_id = $4 -RETURNING importer_node_id, imported_path +RETURNING importer_node_id, imported_path, imported_lockfile_hash ", ) .bind(importer_path) @@ -95,9 +98,9 @@ RETURNING importer_node_id, imported_path importer_kind: &str, executor: impl sqlx::Executor<'a, Database = sqlx::Postgres>, ) -> Result { - let dmap = sqlx::query_as::<_, (String, String)>( + let dmap = sqlx::query_as::<_, (String, String, Option)>( " -SELECT importer_node_id, imported_path +SELECT importer_node_id, imported_path, imported_lockfile_hash FROM dependency_map WHERE workspace_id = $1 AND importer_path = $2 @@ -146,29 +149,45 @@ SELECT importer_node_id, imported_path return Ok(()); }; - referenced_paths.retain(|imported_path| { + // Fetch lock hashes for all referenced paths + let lock_hashes: std::collections::HashMap> = sqlx::query_as( + "SELECT path, lockfile_hash FROM lock_hash WHERE workspace_id = $1 AND path = ANY($2)", + ) + .bind(&self.w_id) + .bind(&referenced_paths) + .fetch_all(&mut **tx) + .await? + .into_iter() + .collect(); + + referenced_paths.retain(|path| { + let hash = lock_hashes.get(path).cloned().flatten(); !self .to_delete - .remove(&(node_id.to_owned(), imported_path.to_owned())) + .remove(&(node_id.to_owned(), path.clone(), hash)) }); if !referenced_paths.is_empty() { tracing::info!("adding missing entries to dependency_map: importer_node_id - {}, importer_kind - {}, new_imported_paths - {:?}", &node_id, &self.importer_kind, - &referenced_paths, + &lock_hashes, ); } for import in referenced_paths { + let lock_hash = lock_hashes.get(&import).copied().flatten(); sqlx::query!( - "INSERT INTO dependency_map (workspace_id, importer_path, importer_kind, imported_path, importer_node_id) - VALUES ($1, $2, $3::text::IMPORTER_KIND, $4, $5) ON CONFLICT DO NOTHING", + "INSERT INTO dependency_map (workspace_id, importer_path, importer_kind, imported_path, importer_node_id, imported_lockfile_hash) + VALUES ($1, $2, $3::text::IMPORTER_KIND, $4, $5, $6) + ON CONFLICT (workspace_id, importer_node_id, importer_kind, importer_path, imported_path) + DO UPDATE SET imported_lockfile_hash = EXCLUDED.imported_lockfile_hash", &self.w_id, &self.importer_path, &self.importer_kind, - import, - node_id + &import, + node_id, + lock_hash ) .execute(&mut **tx) .await?; @@ -192,11 +211,13 @@ SELECT importer_node_id, imported_path tracing::info!("dissolving dependency_map: {:?}", &self); - for (importer_node_id, imported_path) in self.to_delete.into_iter() { - tracing::info!("cleaning orphan entry from dependency_map: importer_kind - {}, imported_path - {}, importer_node_id - {}", + for (importer_node_id, imported_path, imported_lockfile_hash) in self.to_delete.into_iter() + { + tracing::info!("cleaning orphan entry from dependency_map: importer_kind - {}, imported_path - {}, importer_node_id - {}, imported_lockfile_hash - {:?}", &self.importer_kind, &imported_path, &importer_node_id, + &imported_lockfile_hash, ); if let Err(err) = sqlx::query!( @@ -207,12 +228,14 @@ SELECT importer_node_id, imported_path AND importer_kind = $3::text::IMPORTER_KIND AND importer_node_id = $4 AND imported_path = $5 + AND imported_lockfile_hash IS NOT DISTINCT FROM $6 -- we don't want to delete other entries with other lockfile hash. ", &self.w_id, &self.importer_path, &self.importer_kind, &importer_node_id, &imported_path, + imported_lockfile_hash ) .execute(&mut *tx) .await diff --git a/backend/windmill-dep-map/src/workspace_dependencies.rs b/backend/windmill-dep-map/src/workspace_dependencies.rs index 6186556a25..03db1405e5 100644 --- a/backend/windmill-dep-map/src/workspace_dependencies.rs +++ b/backend/windmill-dep-map/src/workspace_dependencies.rs @@ -89,11 +89,19 @@ impl NewWorkspaceDependencies { .fetch_optional(&mut *tx) .await?; + let content_hash = windmill_common::scripts::hash_script(&self.content); let new_id = sqlx::query_scalar!( " - INSERT INTO workspace_dependencies(name, workspace_id, content, language, description) - VALUES ($1, $2, $3, $4, $5) - RETURNING id + WITH ins AS ( + INSERT INTO workspace_dependencies(name, workspace_id, content, language, description) + VALUES ($1, $2, $3, $4, $5) + RETURNING id + ), lock_ins AS ( + INSERT INTO lock_hash (workspace_id, path, lockfile_hash) + VALUES ($2, $6, $7) + ON CONFLICT (workspace_id, path) DO UPDATE SET lockfile_hash = $7 + ) + SELECT id FROM ins ", self.name.clone(), self.workspace_id, @@ -101,7 +109,9 @@ impl NewWorkspaceDependencies { self.language as ScriptLang, self.description .or(prev_description.clone()) - .unwrap_or("Default Workspace Dependencies".to_owned()) + .unwrap_or("Default Workspace Dependencies".to_owned()), + path, + content_hash ) .fetch_one(&mut *tx) .await?; @@ -136,7 +146,7 @@ pub async fn trigger_dependents_to_recompute_dependencies_in_the_background( language = ?language, "waiting for cache timeout after creating first unnamed workspace dependencies" ); - tokio::time::sleep(EXISTS_CACHE_TIMEOUT).await; + tokio::time::sleep(*EXISTS_CACHE_TIMEOUT).await; } if let Err(e) = trigger_dependents_to_recompute_dependencies( diff --git a/backend/windmill-types/src/scripts.rs b/backend/windmill-types/src/scripts.rs index a925bd2031..a1538be04b 100644 --- a/backend/windmill-types/src/scripts.rs +++ b/backend/windmill-types/src/scripts.rs @@ -606,7 +606,7 @@ pub struct HubScript { pub summary: Option, } -pub fn hash_script(ns: &NewScript) -> i64 { +pub fn hash_script(ns: impl std::hash::Hash) -> i64 { let mut dh = std::hash::DefaultHasher::new(); ns.hash(&mut dh); dh.finish() as i64 diff --git a/backend/windmill-worker/src/worker_lockfiles.rs b/backend/windmill-worker/src/worker_lockfiles.rs index 5da446c825..8636105243 100644 --- a/backend/windmill-worker/src/worker_lockfiles.rs +++ b/backend/windmill-worker/src/worker_lockfiles.rs @@ -52,6 +52,7 @@ lazy_static::lazy_static! { static ref WMDEBUG_NO_NEW_FLOW_VERSION_ON_DJ: bool = std::env::var("WMDEBUG_NO_NEW_FLOW_VERSION_ON_DJ").is_ok(); static ref WMDEBUG_NO_NEW_APP_VERSION_ON_DJ: bool = std::env::var("WMDEBUG_NO_NEW_APP_VERSION_ON_DJ").is_ok(); static ref WMDEBUG_NO_COMPONENTS_TO_RELOCK: bool = std::env::var("WMDEBUG_NO_COMPONENTS_TO_RELOCK").is_ok(); + static ref WMDEBUG_NO_RELOCK_SKIP_OPTIMIZATION: bool = std::env::var("WMDEBUG_NO_RELOCK_SKIP_OPTIMIZATION").is_ok(); } use crate::common::{MaybeLock, OccupancyMetrics}; @@ -132,6 +133,12 @@ pub async fn handle_dependency_job( }, }; + let triggered_by_relative_import = job + .args + .as_ref() + .map(|x| x.get("triggered_by_relative_import").is_some()) + .unwrap_or_default(); + let content = capture_dependency_job( &job.id, job.script_lang.as_ref().map(|v| Ok(v)).unwrap_or_else(|| { @@ -152,6 +159,11 @@ pub async fn handle_dependency_job( script_path, occupancy_metrics, &raw_workspace_dependencies_o, + None, + triggered_by_relative_import, + script_path, + None, + "script", ) .await; @@ -172,11 +184,20 @@ pub async fn handle_dependency_job( // We do not create new row for this update // That means we can keep current hash and just update lock + // Also store lockfile hash for dependency change detection + let lockfile_hash = windmill_common::scripts::hash_script(&content); sqlx::query!( - "UPDATE script SET lock = $1 WHERE hash = $2 AND workspace_id = $3", + "WITH update_lock AS ( + UPDATE script SET lock = $1 WHERE hash = $2 AND workspace_id = $3 + ) + INSERT INTO lock_hash (workspace_id, path, lockfile_hash) + VALUES ($3, $4, $5) + ON CONFLICT (workspace_id, path) DO UPDATE SET lockfile_hash = $5", &content, ¤t_hash.0, - w_id + w_id, + script_path, + &lockfile_hash ) .execute(db) .await?; @@ -395,6 +416,7 @@ pub async fn handle_flow_dependency_job( &raw_deps, &mut dependency_map, &raw_workspace_dependencies_o, + triggered_by_relative_import, ) .await?; @@ -608,6 +630,7 @@ async fn lock_flow_value<'c>( raw_deps: &Option>, dependency_map: &mut ScopedDependencyMap, raw_workspace_dependencies_o: &Option, + triggered_by_relative_import: bool, ) -> Result<( FlowValue, sqlx::Transaction<'c, sqlx::Postgres>, @@ -637,6 +660,7 @@ async fn lock_flow_value<'c>( &raw_deps, dependency_map, &raw_workspace_dependencies_o, + triggered_by_relative_import, ) .await?; @@ -667,6 +691,7 @@ async fn lock_flow_value<'c>( &raw_deps, dependency_map, &raw_workspace_dependencies_o, + triggered_by_relative_import, ) .await?; @@ -703,6 +728,7 @@ async fn lock_flow_value<'c>( &raw_deps, dependency_map, raw_workspace_dependencies_o, + triggered_by_relative_import, ) .await?; @@ -740,6 +766,7 @@ async fn lock_modules<'c>( raw_deps: &Option>, dependency_map: &mut ScopedDependencyMap, // (modules to replace old seq (even unmmodified ones), new transaction, modified ids) ) raw_workspace_dependencies_o: &Option, + triggered_by_relative_import: bool, ) -> Result<( Vec, sqlx::Transaction<'c, sqlx::Postgres>, @@ -794,6 +821,7 @@ async fn lock_modules<'c>( &raw_deps, dependency_map, &raw_workspace_dependencies_o, + triggered_by_relative_import, )) .await?; e.value = FlowModuleValue::ForloopFlow { @@ -832,6 +860,7 @@ async fn lock_modules<'c>( &raw_deps, dependency_map, raw_workspace_dependencies_o, + triggered_by_relative_import, )) .await?; nmodified_ids.extend(inner_modified_ids); @@ -862,6 +891,7 @@ async fn lock_modules<'c>( &raw_deps, dependency_map, raw_workspace_dependencies_o, + triggered_by_relative_import, )) .await?; e.value = FlowModuleValue::WhileloopFlow { @@ -897,6 +927,7 @@ async fn lock_modules<'c>( &raw_deps, dependency_map, raw_workspace_dependencies_o, + triggered_by_relative_import, )) .await?; nmodified_ids.extend(inner_modified_ids); @@ -926,6 +957,7 @@ async fn lock_modules<'c>( &raw_deps, dependency_map, raw_workspace_dependencies_o, + triggered_by_relative_import, )) .await?; errors.extend(ninner_errors); @@ -995,6 +1027,7 @@ async fn lock_modules<'c>( &raw_deps, dependency_map, raw_workspace_dependencies_o, + triggered_by_relative_import, )) .await?; @@ -1091,6 +1124,11 @@ async fn lock_modules<'c>( ), occupancy_metrics, raw_workspace_dependencies_o, + lock.as_deref(), + triggered_by_relative_import, + job_path, + Some(&e.id), + "flow", ) .await; // @@ -1497,6 +1535,7 @@ async fn lock_modules_app( container_id: Option, dependency_map: &mut ScopedDependencyMap, raw_workspace_dependencies_o: &Option, + triggered_by_relative_import: bool, ) -> Result { match value { Value::Object(mut m) => { @@ -1585,7 +1624,7 @@ async fn lock_modules_app( return Ok(Value::Object(m.clone())); } } - logs.push_str("Found lockable inline script. Generating lock...\n"); + let existing_lock = v.get("lock").and_then(|x| x.as_str()); let new_lock = capture_dependency_job( &job.id, &language, @@ -1601,8 +1640,12 @@ async fn lock_modules_app( token, &format!("{}/app", job.runnable_path()), occupancy_metrics, - // TODO: &None, + existing_lock, + triggered_by_relative_import, + &job.runnable_path(), + container_id.as_deref(), + "app", ) .await; match new_lock { @@ -1679,6 +1722,7 @@ async fn lock_modules_app( .or(container_id.clone()), dependency_map, raw_workspace_dependencies_o, + triggered_by_relative_import, ) .await?, ); @@ -1706,6 +1750,7 @@ async fn lock_modules_app( container_id.clone(), dependency_map, raw_workspace_dependencies_o, + triggered_by_relative_import, ) .await?, ); @@ -1803,6 +1848,7 @@ pub async fn handle_app_dependency_job( None, &mut dependency_map, &raw_workspace_dependencies_o, + triggered_by_relative_import, ) .await?; @@ -2174,6 +2220,110 @@ async fn ansible_dep( serde_json::to_string(&ansible_lockfile).map_err(|e| e.into()) } +/// Checks if we can skip relocking because imported lockfiles haven't changed. +/// Returns Ok(Some(lock)) if we can skip, Ok(None) if we should relock. +async fn try_skip_relock( + db: &sqlx::Pool, + w_id: &str, + base_path: &str, + step_id: Option<&str>, + runnable_type: &str, + existing_lock: Option<&str>, +) -> error::Result> { + tracing::debug!( + workspace_id = %w_id, + base_path = %base_path, + step_id = ?step_id, + runnable_type = %runnable_type, + has_existing_lock = existing_lock.is_some(), + "try_skip_relock: checking if we can skip" + ); + + // Check that ALL imports have matching hashes and at least one import exists + // Returns true only if: count > 0 AND all hashes match + // Returns false if: no imports OR any hash mismatch + let all_imports_unchanged = sqlx::query_scalar!( + "SELECT + COUNT(*) > 0 + AND BOOL_AND( + CASE + -- For dependencies/: use IS NOT DISTINCT FROM (NULL = NULL is true for legacy) + -- And the reason for this, is that for every script we also add dependency on default workspace dependencies by default + -- that default/unnamed workspace dependencies may not exist, but we still do this. + -- it is needed for windmill to know what to redeploy when default workspace dependencies are being added + + -- naturally for non-existant entries we have no hash of it + -- so if we compared hash with '=' (instead of IS NOT DISTINCT FROM), it would give false on NULL = NULL, + -- which would mean that this entire query returns false, which means relock skip cannot happen + + -- The solution is to say if both: referenced and current hash are NULLs we treat it as true, so it becomes no longer a blocker from skip. + -- + -- It is backed up by the fact that server is responsible for deploying new workspace dependencies + -- so if one was to deploy a new wdeps, server would assign it new hash, and this expression would be invalid and would not approve skip + WHEN dm.imported_path LIKE 'dependencies/%' + THEN dm.imported_lockfile_hash IS NOT DISTINCT FROM lh.lockfile_hash + + -- For scripts: use = with COALESCE (NULL = NULL becomes false) + -- unlike w deps, we can't do IS NOT DISTINCT FROM here + -- the reason is that scripts deployments are issued by other workers instead of the server + -- which would mean that there is no guarantee that new d job will also write it's lock's hash to the `lock_hash` + -- which could lead to false positives + ELSE COALESCE(dm.imported_lockfile_hash = lh.lockfile_hash, false) + END + ) + FROM dependency_map dm + LEFT JOIN lock_hash lh + ON lh.workspace_id = dm.workspace_id + AND lh.path = dm.imported_path + WHERE dm.workspace_id = $1 + AND dm.importer_path = $2 + AND dm.importer_node_id = $3", + w_id, + base_path, + step_id.unwrap_or("") + ) + .fetch_one(db) + .await? + .unwrap_or(false); + + tracing::debug!( + workspace_id = %w_id, + base_path = %base_path, + all_imports_unchanged = all_imports_unchanged, + "try_skip_relock: query result" + ); + + if !all_imports_unchanged { + return Ok(None); + } + + // Fetch existing lock based on runnable type + let lock = match runnable_type { + "script" => sqlx::query_scalar!( + "SELECT lock FROM script WHERE path = $1 AND workspace_id = $2 AND lock IS NOT NULL + AND deleted = false ORDER BY created_at DESC LIMIT 1", + base_path, + w_id + ) + .fetch_optional(db) + .await? + .flatten(), + + "flow" | "app" => existing_lock.map(|s| s.to_string()), + _ => None, + }; + + tracing::debug!( + workspace_id = %w_id, + base_path = %base_path, + lock_found = lock.is_some(), + "try_skip_relock: lock fetch result" + ); + + Ok(lock) +} + +/// Captures dependencies for a script and generates a lockfile. async fn capture_dependency_job( job_id: &Uuid, job_language: &ScriptLang, @@ -2190,7 +2340,48 @@ async fn capture_dependency_job( script_path: &str, occupancy_metrics: &mut OccupancyMetrics, raw_workspace_dependencies_o: &Option, + existing_lock: Option<&str>, + triggered_by_relative_import: bool, + // The base path of the runnable (script/flow/app) without suffixes. + // Used for dependency_map lookups. For flows/apps, this is the flow/app path, + // not the `script_path` which may have `/flow` or `/app` appended. + base_path: &str, + step_id: Option<&str>, + runnable_type: &str, // "script", "flow", or "app" ) -> error::Result { + // Check if we can skip relocking: + // - Must be triggered by relative import + // - Debug flag must not be set + if triggered_by_relative_import && !*WMDEBUG_NO_RELOCK_SKIP_OPTIMIZATION { + match try_skip_relock(db, w_id, base_path, step_id, runnable_type, existing_lock).await { + Ok(Some(lock)) => { + let log_msg = match step_id { + Some(id) => format!( + "\nSkipping relock for step '{}' - imported lockfiles unchanged", + id + ), + None => "\nSkipping relock - imported lockfiles unchanged".to_string(), + }; + + tracing::info!(workspace_id = %w_id, job_id = %job_id, "{log_msg}"); + append_logs(job_id, w_id, log_msg, &db.into()).await; + return Ok(lock); + } + Ok(None) => {} // Continue to relock + Err(e) => { + tracing::error!(workspace_id = %w_id, job_id = %job_id, "Failed to check skip relock: {e}"); + } + } + } + + let log_msg = match step_id { + Some(id) => format!("\nRelocking script for step '{}'", id), + None => "\nRelocking script".to_string(), + }; + + tracing::info!(workspace_id = %w_id, job_id = %job_id, "{log_msg}"); + append_logs(job_id, w_id, log_msg, &db.into()).await; + let workspace_dependencies = WorkspaceDependenciesPrefetched::extract( job_raw_code, *job_language, @@ -2466,15 +2657,14 @@ async fn capture_dependency_job( // for related places search: ADD_NEW_LANG _ => "".to_owned(), }; - { - let mut lines = vec![]; - add_lock_header(&mut lines, workspace_dependencies, *job_language, w_id, db).await?; - Ok(if lines.is_empty() { - lock - } else { - format!("{}\n{lock}", lines.join("\n")) - }) - } + + let mut lines = vec![]; + add_lock_header(&mut lines, workspace_dependencies, *job_language, w_id, db).await?; + Ok(if lines.is_empty() { + lock + } else { + format!("{}\n{lock}", lines.join("\n")) + }) } async fn add_lock_header(