diff --git a/backend/.sqlx/query-87877da7342ee745514442387c707f75182ea2f25162fe6e3c9b99a8a8cfacdb.json b/backend/.sqlx/query-87877da7342ee745514442387c707f75182ea2f25162fe6e3c9b99a8a8cfacdb.json new file mode 100644 index 0000000000..8435668f14 --- /dev/null +++ b/backend/.sqlx/query-87877da7342ee745514442387c707f75182ea2f25162fe6e3c9b99a8a8cfacdb.json @@ -0,0 +1,20 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT hash FROM script WHERE workspace_id = 'test-workspace' AND path = 'u/test-user/ingest' AND archived = false", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "hash", + "type_info": "Int8" + } + ], + "parameters": { + "Left": [] + }, + "nullable": [ + false + ] + }, + "hash": "87877da7342ee745514442387c707f75182ea2f25162fe6e3c9b99a8a8cfacdb" +} diff --git a/backend/.sqlx/query-63c0b60d1042ef74e69b11119355c26d8ca6e5e94902450c11ab7816a385fd9b.json b/backend/.sqlx/query-b8161f6481460bed1c985bedc7c638a42f03599ce5eb872cf601f1da2e76b946.json similarity index 93% rename from backend/.sqlx/query-63c0b60d1042ef74e69b11119355c26d8ca6e5e94902450c11ab7816a385fd9b.json rename to backend/.sqlx/query-b8161f6481460bed1c985bedc7c638a42f03599ce5eb872cf601f1da2e76b946.json index 82af09d6a6..6b2ec01b23 100644 --- a/backend/.sqlx/query-63c0b60d1042ef74e69b11119355c26d8ca6e5e94902450c11ab7816a385fd9b.json +++ b/backend/.sqlx/query-b8161f6481460bed1c985bedc7c638a42f03599ce5eb872cf601f1da2e76b946.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "SELECT s.path AS \"path!\", s.language AS \"language!: ScriptLang\"\n FROM asset a\n JOIN script s ON s.workspace_id = a.workspace_id AND s.path = a.usage_path\n AND s.archived = false AND s.deleted = false\n WHERE a.workspace_id = $1 AND a.kind = 'dbt' AND a.path = $2\n AND a.usage_kind = 'script' AND a.usage_access_type IN ('w', 'rw')\n AND a.usage_path <> $3", + "query": "SELECT s.path AS \"path!\", s.language AS \"language!: ScriptLang\"\n FROM asset a\n JOIN script s ON s.workspace_id = a.workspace_id AND s.path = a.usage_path\n AND s.archived = false AND s.deleted = false\n WHERE a.workspace_id = $1 AND a.kind = 'dbt' AND a.path = $2\n AND a.usage_kind = 'script' AND a.usage_access_type IN ('w', 'rw')\n AND a.usage_path <> ALL($3)", "describe": { "columns": [ { @@ -51,7 +51,7 @@ "Left": [ "Text", "Text", - "Text" + "TextArray" ] }, "nullable": [ @@ -59,5 +59,5 @@ false ] }, - "hash": "63c0b60d1042ef74e69b11119355c26d8ca6e5e94902450c11ab7816a385fd9b" + "hash": "b8161f6481460bed1c985bedc7c638a42f03599ce5eb872cf601f1da2e76b946" } diff --git a/backend/.sqlx/query-e37b74f77cfa8769aee4d95155a3b3b6856b44d49c0712f08e307f950f210570.json b/backend/.sqlx/query-e37b74f77cfa8769aee4d95155a3b3b6856b44d49c0712f08e307f950f210570.json new file mode 100644 index 0000000000..46a9c8e505 --- /dev/null +++ b/backend/.sqlx/query-e37b74f77cfa8769aee4d95155a3b3b6856b44d49c0712f08e307f950f210570.json @@ -0,0 +1,12 @@ +{ + "db_name": "PostgreSQL", + "query": "INSERT INTO asset (workspace_id, path, kind, usage_access_type, usage_path, usage_kind)\n VALUES ('test-workspace', 'main/analytics/orders', 'dbt', 'w', 'u/test-user/project',\n 'script')", + "describe": { + "columns": [], + "parameters": { + "Left": [] + }, + "nullable": [] + }, + "hash": "e37b74f77cfa8769aee4d95155a3b3b6856b44d49c0712f08e307f950f210570" +} diff --git a/backend/windmill-api-integration-tests/tests/dbt_materialize_target.rs b/backend/windmill-api-integration-tests/tests/dbt_materialize_target.rs index 785a64ccb8..477ed1d680 100644 --- a/backend/windmill-api-integration-tests/tests/dbt_materialize_target.rs +++ b/backend/windmill-api-integration-tests/tests/dbt_materialize_target.rs @@ -146,6 +146,41 @@ async fn test_dbt_materialize_target_deploy_contract(db: Pool) -> anyh assert_eq!(resp.status(), 400); assert!(resp.text().await?.contains("u/test-user/project")); + // A rename is the other half of that: the producer's write still sits at the + // OLD path in the committed snapshot this deploy reads, while the same + // transaction removes it — so it must not count as the producer that would + // wake the subscription the rename adds. + let hash = sqlx::query_scalar!( + "SELECT hash FROM script WHERE workspace_id = 'test-workspace' \ + AND path = 'u/test-user/ingest' AND archived = false" + ) + .fetch_one(&db) + .await?; + sqlx::query!( + "INSERT INTO asset (workspace_id, path, kind, usage_access_type, usage_path, usage_kind) + VALUES ('test-workspace', 'main/analytics/orders', 'dbt', 'w', 'u/test-user/project', + 'script')" + ) + .execute(&db) + .await?; + let resp = authed(client().post(format!( + "http://localhost:{port}/api/w/test-workspace/scripts/create" + ))) + .json(&json!({ + "path": "u/test-user/ingest_renamed", + "parent_hash": format!("{:x}", hash), + "summary": "", + "description": "", + "content": "// on dbt://main/analytics/orders\nexport async function main() {}", + "language": "deno", + "schema": { "type": "object", "properties": {}, "required": [] } + })) + .send() + .await + .unwrap(); + assert_eq!(resp.status(), 400); + assert!(resp.text().await?.contains("u/test-user/project")); + // Neither annotation is accepted on a dbt script: the graph ingest // republishes that path's asset and trigger rows wholesale, so either would // deploy something the dependency job then silently removes. diff --git a/backend/windmill-api-scripts/src/scripts.rs b/backend/windmill-api-scripts/src/scripts.rs index 5914274518..fd4f986779 100644 --- a/backend/windmill-api-scripts/src/scripts.rs +++ b/backend/windmill-api-scripts/src/scripts.rs @@ -2468,8 +2468,15 @@ async fn create_script_internal<'c>( and a project is run on its schedule, not woken by an asset cascade." ))); } + // Both paths under a rename: the old one's committed write row is + // still there and this transaction is about to remove it. + let deploying_paths = match p_path_opt.as_deref().filter(|old| *old != ns.path) { + Some(old) => vec![ns.path.clone(), old.to_string()], + None => vec![ns.path.clone()], + }; if let Some(dbt_owner) = - windmill_common::assets::sole_dbt_producer(&db, &w_id, relation, &ns.path).await? + windmill_common::assets::sole_dbt_producer(&db, &w_id, relation, &deploying_paths) + .await? { return Err(Error::BadRequest(format!( "`{trigger_ref}` cannot be subscribed to: it is built by the dbt project at \ diff --git a/backend/windmill-common/src/assets.rs b/backend/windmill-common/src/assets.rs index 057565cb89..9a134a1f11 100644 --- a/backend/windmill-common/src/assets.rs +++ b/backend/windmill-common/src/assets.rs @@ -258,12 +258,14 @@ pub fn derive_pipeline_asset_trigger_refs( /// for every `dbt://` node anyway (the source that script wrote stays gated). /// Callers must therefore already be scoped to `workspace_id`. /// -/// `subscriber_path` is excluded from the producer set, and has to be: reading +/// `deploying_paths` is excluded from the producer set, and has to be: reading /// committed rows means the deploying script's own are the version being /// replaced, so one that just dropped its `// materialize` would still count as a -/// producer and let a now-dormant subscription through. Excluding it is free of -/// the opposite error, because a script never wakes its own subscription — the -/// dispatcher skips that as a self-loop. +/// producer and let a now-dormant subscription through. Pass every path this +/// deploy is rewriting — under a rename that is the old path as well as the new +/// one, whose committed write row the transaction is about to remove. Excluding +/// them is free of the opposite error, because a script never wakes its own +/// subscription — the dispatcher skips that as a self-loop. /// /// A producer another deploy is committing concurrently is still invisible, so /// that race resolves toward refusing with a message the user can retry past. @@ -271,7 +273,7 @@ pub async fn sole_dbt_producer<'e>( executor: impl PgExecutor<'e>, workspace_id: &str, asset_path: &str, - subscriber_path: &str, + deploying_paths: &[String], ) -> error::Result> { use crate::scripts::ScriptLang; let producers = sqlx::query!( @@ -281,10 +283,10 @@ pub async fn sole_dbt_producer<'e>( AND s.archived = false AND s.deleted = false WHERE a.workspace_id = $1 AND a.kind = 'dbt' AND a.path = $2 AND a.usage_kind = 'script' AND a.usage_access_type IN ('w', 'rw') - AND a.usage_path <> $3"#, + AND a.usage_path <> ALL($3)"#, workspace_id, asset_path, - subscriber_path + deploying_paths ) .fetch_all(executor) .await?; diff --git a/backend/windmill-common/tests/dbt_producer_rules.rs b/backend/windmill-common/tests/dbt_producer_rules.rs index 1498f7fd3d..f63e0879b5 100644 --- a/backend/windmill-common/tests/dbt_producer_rules.rs +++ b/backend/windmill-common/tests/dbt_producer_rules.rs @@ -59,7 +59,7 @@ async fn plant_subscriber(db: &Pool, path: &str) { #[sqlx::test(migrations = "../migrations", fixtures("base"))] async fn no_producer_is_not_dormant(db: Pool) { assert_eq!( - sole_dbt_producer(&db, WS, RELATION, SUBSCRIBER) + sole_dbt_producer(&db, WS, RELATION, &[SUBSCRIBER.to_string()]) .await .unwrap(), None, @@ -71,7 +71,7 @@ async fn no_producer_is_not_dormant(db: Pool) { async fn dbt_only_producer_is_dormant(db: Pool) { plant_producer(&db, "u/test-user/project", "dbt", 1).await; assert_eq!( - sole_dbt_producer(&db, WS, RELATION, SUBSCRIBER) + sole_dbt_producer(&db, WS, RELATION, &[SUBSCRIBER.to_string()]) .await .unwrap(), Some("u/test-user/project".to_string()) @@ -83,7 +83,7 @@ async fn a_native_producer_beside_dbt_is_not_dormant(db: Pool) { plant_producer(&db, "u/test-user/project", "dbt", 1).await; plant_producer(&db, "u/test-user/ingest", "postgresql", 2).await; assert_eq!( - sole_dbt_producer(&db, WS, RELATION, SUBSCRIBER) + sole_dbt_producer(&db, WS, RELATION, &[SUBSCRIBER.to_string()]) .await .unwrap(), None @@ -102,7 +102,7 @@ async fn a_superseded_native_version_does_not_count(db: Pool) { .expect("archive the old version"); plant_producer(&db, "u/test-user/project", "dbt", 2).await; assert_eq!( - sole_dbt_producer(&db, WS, RELATION, SUBSCRIBER) + sole_dbt_producer(&db, WS, RELATION, &[SUBSCRIBER.to_string()]) .await .unwrap(), Some("u/test-user/project".to_string()) @@ -113,18 +113,38 @@ async fn a_superseded_native_version_does_not_count(db: Pool) { /// script dropping its `// materialize` while adding a subscription would /// otherwise count itself as the producer that wakes it — and commit a dormant /// edge. It can never be that producer anyway: the dispatcher skips self-loops. +/// Under a rename that write sits at the OLD path, which the deploy is removing +/// in the same uncommitted transaction, so both paths have to be excluded. #[sqlx::test(migrations = "../migrations", fixtures("base"))] async fn the_subscriber_is_never_its_own_producer(db: Pool) { plant_producer(&db, "u/test-user/project", "dbt", 1).await; plant_producer(&db, SUBSCRIBER, "postgresql", 2).await; assert_eq!( - sole_dbt_producer(&db, WS, RELATION, SUBSCRIBER) + sole_dbt_producer(&db, WS, RELATION, &[SUBSCRIBER.to_string()]) .await .unwrap(), Some("u/test-user/project".to_string()) ); } +#[sqlx::test(migrations = "../migrations", fixtures("base"))] +async fn a_renamed_producer_is_excluded_too(db: Pool) { + plant_producer(&db, "u/test-user/project", "dbt", 1).await; + plant_producer(&db, "u/test-user/old_ingest", "postgresql", 2).await; + assert_eq!( + sole_dbt_producer( + &db, + WS, + RELATION, + &[SUBSCRIBER.to_string(), "u/test-user/old_ingest".to_string()] + ) + .await + .unwrap(), + Some("u/test-user/project".to_string()), + "the write this deploy is moving off the old path cannot wake the subscription" + ); +} + #[sqlx::test(migrations = "../migrations", fixtures("base"))] async fn the_set_form_agrees_with_the_singular_one(db: Pool) { let relations = vec![RELATION.to_string()];