diff --git a/backend/windmill-worker/src/dbt_executor.rs b/backend/windmill-worker/src/dbt_executor.rs index 47b840c6e9..f3c258c189 100644 --- a/backend/windmill-worker/src/dbt_executor.rs +++ b/backend/windmill-worker/src/dbt_executor.rs @@ -19,6 +19,7 @@ use tokio::process::Command; use uuid::Uuid; use windmill_common::client::AuthedClient; use windmill_common::error::{self, Error}; +use windmill_common::jobs::JobKind; use windmill_common::materialization::{ record_materialization, MaterializationStatus, RecordMaterializationRequest, }; @@ -406,7 +407,8 @@ pub(crate) async fn handle_dbt_job( &job.id, &job.workspace_id, format!( - "\nDeferring unbuilt refs to the dbt state published by run {}\n", + "\nDeferring unbuilt refs to the dbt state published by run {}; this run \ + publishes none of its own\n", deferral.published_by ), conn, @@ -610,6 +612,11 @@ pub(crate) async fn handle_dbt_job( if run.is_ok() && command == "build" && inv.deferral.is_none() + // A run of the DEPLOYED version, by kind. A preview carries a + // caller-supplied `script_hash` into `runnable_id` + // (`run_preview_script`), so the version guard alone would let anyone who + // may run a job publish arbitrary content as a deployed script's state. + && job.kind == JobKind::Script && prepared.graph_refresh.publishes_ownership() { // Losing it costs the next deferral, not the run that just finished — @@ -1016,6 +1023,11 @@ pub struct PreparedProject { /// `profiles.yml` default. Half of an environment's identity, since a /// `target.name` macro decides where a model is built. pub effective_target: Option, + /// Whether a project-owned `profiles.yml` templates where its relations go, + /// in which case two renderings share one `relation_root` and an environment + /// cannot be told apart — so a deferral is refused rather than resolved + /// through another rendering's manifest. + pub templated_location: bool, /// The profile target's database. Nodes that override it qualify their /// `dbt://` schema segment so two databases cannot collapse onto one node. pub default_database: Option, @@ -1315,6 +1327,7 @@ pub(crate) async fn prepare_project( warehouse: profile.warehouse, target: descriptor.profile.target.clone(), effective_target: profile.target, + templated_location: profile.templated_location, descriptor_content: descriptor_content.to_string(), descriptor_env, @@ -1647,6 +1660,8 @@ struct ResolvedProfile { /// `profiles.yml` default. Resolved because it is half of an environment's /// identity and a `target.name` macro can move every relation. target: Option, + /// Whether a project-owned `profiles.yml` templates where its relations go. + templated_location: bool, digest: String, } @@ -1748,6 +1763,7 @@ async fn write_profiles( database: target.database, schema: target.schema, target: Some(target.name), + templated_location: target.templated_location, digest: profile_digest, }); } @@ -1851,6 +1867,9 @@ async fn write_profiles( database: rendered.database, schema: rendered.schema, target: Some(target.to_string()), + // Rendered from a resource, so its location is whatever that resource + // says rather than something the run's environment decides. + templated_location: false, digest: profile_digest, }) } @@ -1993,16 +2012,23 @@ async fn adapter_from_profiles_yml( // identically to one on a workspace warehouse, which is what lets the two // meet on the same node when they are on the same relation. let (database_key, schema_key) = adapter.target_identity_keys(); - let read = |k: &str| { + let raw = |k: &str| { out.get(k) .and_then(|v| v.as_str()) - .map(|v| v.to_string()) - .filter(|v| !v.is_empty() && !v.contains("{{")) + .filter(|v| !v.is_empty()) }; + let read = |k: &str| raw(k).filter(|v| !v.contains("{{")).map(|v| v.to_string()); Ok(ProfileTarget { adapter, database: read(database_key), schema: read(schema_key), + // A TEMPLATED location reads as absent above, so two renderings of this + // file resolve to one `relation_root` and would share one environment. + // Distinguished from plainly absent, which is the adapter's default and + // does not move: only the templated case has to refuse a deferral. + templated_location: [database_key, schema_key] + .iter() + .any(|k| raw(k).is_some_and(|v| v.contains("{{"))), // The output actually chosen, which for a templated `target:` is the sole // one rather than the template text no output answers to. name: match ( @@ -2025,6 +2051,11 @@ struct ProfileTarget { schema: Option, /// The output this resolved to, by name. name: String, + /// Whether its database or schema is a template rather than a literal. Both + /// read as absent, so this is the only thing that separates "the adapter's + /// default, which does not move" from "wherever this run's environment + /// renders it to". + templated_location: bool, } lazy_static::lazy_static! { diff --git a/backend/windmill-worker/src/dbt_state.rs b/backend/windmill-worker/src/dbt_state.rs index 8551f6c20a..7e111d90bd 100644 --- a/backend/windmill-worker/src/dbt_state.rs +++ b/backend/windmill-worker/src/dbt_state.rs @@ -139,6 +139,7 @@ pub(crate) async fn publish( // publishers cannot collide on them and nothing here can overwrite an // artifact a committed row still names. A failure below has only its own // objects to drop. + let nonce = Uuid::new_v4(); let (manifest, manifest_key) = store( manifest, "manifest.json", @@ -146,6 +147,7 @@ pub(crate) async fn publish( &p.script_path, w_id, job_id, + &nonce, ) .await?; let (run_results, run_results_key) = match run_results { @@ -156,6 +158,7 @@ pub(crate) async fn publish( &p.script_path, w_id, job_id, + &nonce, ) .await { @@ -228,7 +231,11 @@ pub(crate) async fn publish( .fetch_optional(&mut *tx) .await? .map(|r| [r.manifest_key, r.run_results_key]) - .unwrap_or_default(); + .unwrap_or_default() + // Never a key this publication is about to commit. The keys carry a + // per-execution nonce so the two cannot coincide, and this is what says + // so rather than leaving it to be re-derived. + .map(|k| k.filter(|k| !mine.iter().flatten().any(|m| m == k))); sqlx::query!( "INSERT INTO dbt_environment_state (workspace_id, script_path, environment, job_id, manifest, manifest_key, run_results, @@ -391,15 +398,21 @@ fn publication_lock(w_id: &str, script_path: &str, environment: &str) -> i64 { /// is what says where an artifact is, so state that moves with a renamed script /// keeps naming objects under the old one. Digested because a Windmill path and a /// schema name may both carry characters an object key gives meaning to. +/// +/// The `nonce` is per EXECUTION rather than per job, because zombie recovery +/// re-runs a job under its own id: keyed on that alone, the second attempt would +/// overwrite the objects the first attempt's committed row still names, and then +/// read those same keys back as displaced and drop them. fn object_key( w_id: &str, script_path: &str, environment: &str, job_id: &Uuid, + nonce: &Uuid, artifact: &str, ) -> String { format!( - "wmill_dbt_state/{w_id}/{}/{job_id}/{artifact}", + "wmill_dbt_state/{w_id}/{}/{job_id}.{nonce}/{artifact}", digest(&format!("{script_path}|{environment}")) ) } @@ -414,11 +427,12 @@ async fn store( script_path: &str, w_id: &str, job_id: &Uuid, + nonce: &Uuid, ) -> error::Result<(Option, Option)> { if value.len() <= *DBT_STATE_INLINE_MAX_BYTES { return Ok((Some(value), None)); } - let key = object_key(w_id, script_path, environment, job_id, artifact); + let key = object_key(w_id, script_path, environment, job_id, nonce, artifact); let size = value.len(); if put_object(&key, value).await? { return Ok((None, Some(key))); @@ -553,6 +567,20 @@ pub(crate) async fn prepare_deferral( .to_string(), )); } + // An environment is the warehouse, the target and where they RESOLVE to, and + // a `profiles.yml` that templates its schema or database resolves somewhere + // this runtime does not render. Two renderings would then share one + // environment, and a deferral after the value changed would resolve every + // unbuilt `ref()` through the previous location's manifest. + if p.templated_location { + return Err(Error::BadRequest( + "this project's `profiles.yml` selects its schema or database with a template, which \ + dbt renders and Windmill does not — so two environments cannot be told apart and a \ + deferral could resolve through the wrong one's manifest. Spell the target's schema \ + and database literally, or set `profile.schema` in the descriptor" + .to_string(), + )); + } let Some(state) = load(p, w_id, conn).await? else { return Err(Error::BadRequest(format!( "no dbt state is stored for this environment ({}), so a `ref()` this run does not \ diff --git a/docs/dbt-runtime.md b/docs/dbt-runtime.md index dff084b760..0afce68ce9 100644 --- a/docs/dbt-runtime.md +++ b/docs/dbt-runtime.md @@ -1276,7 +1276,10 @@ objects that are gone, so an orphan is the cheaper side to take. The path and the environment are only a prefix of that key. The row is what says where an artifact is, which is why state can travel with a renamed script and go -on naming objects under the old path's digest. +on naming objects under the old path's digest. The rest of the key is the job and +a per-EXECUTION nonce — zombie recovery re-runs a job under its own id, so keyed +on that alone a second attempt would overwrite the objects the first attempt's +committed row still names, then read those keys back as displaced and drop them. Publishers of one environment serialize on `pg_advisory_xact_lock`, so only one of them settles the row and the objects it displaces at a time — an advisory lock @@ -1303,6 +1306,11 @@ at a path this one was renamed away from, and this job's manifest would then become that project's deferral state. A preview names no version and so publishes nothing, which is right for a run of content that was never deployed. +The job's KIND is checked beside it, because a preview carries a caller-supplied +`script_hash` into `runnable_id` (`run_preview_script`): the version alone would +let anyone who may run a job publish arbitrary content as a deployed script's +state. + That guard HOLDS the script row (`FOR SHARE`) for the rest of the publication, so a rename, archive or delete of the path either waits for it or is seen by it. Read unlocked, it leaves a window where the lifecycle clear finds no row to take, @@ -1324,6 +1332,13 @@ run that publishes an environment's state and the run that defers to it are two invocations of ONE script (decision 6: N scripts means N projects): a project that could only defer by descriptor could never populate the state it reads. +A project whose own `profiles.yml` selects its schema or database with a +template is refused a deferral outright: dbt renders those and Windmill does not, +so two renderings resolve to one `relation_root`, and a deferral after the value +changed would resolve every unbuilt `ref()` through the previous location's +manifest. Plainly absent is different — that is the adapter's default, which does +not move. + A run that asks to defer with nothing published is refused, naming the environment and the runs that cannot publish one. The alternative — running without deferral — fails deep inside dbt with a relation-not-found the caller has