diff --git a/backend/.sqlx/query-341491d10649cf76c6d3bce963cf03d449a85e186e4c0a10303f3a043ecb6816.json b/backend/.sqlx/query-341491d10649cf76c6d3bce963cf03d449a85e186e4c0a10303f3a043ecb6816.json new file mode 100644 index 0000000000..35d01aa0b1 --- /dev/null +++ b/backend/.sqlx/query-341491d10649cf76c6d3bce963cf03d449a85e186e4c0a10303f3a043ecb6816.json @@ -0,0 +1,24 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT 1 as _e FROM dbt_environment_state\n WHERE workspace_id = $1 AND script_path = $2 AND environment = $3 FOR UPDATE", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "_e", + "type_info": "Int4" + } + ], + "parameters": { + "Left": [ + "Text", + "Text", + "Text" + ] + }, + "nullable": [ + null + ] + }, + "hash": "341491d10649cf76c6d3bce963cf03d449a85e186e4c0a10303f3a043ecb6816" +} diff --git a/backend/.sqlx/query-7c2f07ec5e3586d865af1a764f23121eea8b1a37f925ce31f941e030c00eaf5c.json b/backend/.sqlx/query-7c2f07ec5e3586d865af1a764f23121eea8b1a37f925ce31f941e030c00eaf5c.json new file mode 100644 index 0000000000..a784a7aa39 --- /dev/null +++ b/backend/.sqlx/query-7c2f07ec5e3586d865af1a764f23121eea8b1a37f925ce31f941e030c00eaf5c.json @@ -0,0 +1,15 @@ +{ + "db_name": "PostgreSQL", + "query": "DELETE FROM dbt_environment_state\n WHERE workspace_id = $1 AND script_path = ANY($2)", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Text", + "TextArray" + ] + }, + "nullable": [] + }, + "hash": "7c2f07ec5e3586d865af1a764f23121eea8b1a37f925ce31f941e030c00eaf5c" +} diff --git a/backend/.sqlx/query-7f421bdf3dc4d47b36457af703ce69ef1e0784e9f25ada27e1b0f5cc0944e9ba.json b/backend/.sqlx/query-7f421bdf3dc4d47b36457af703ce69ef1e0784e9f25ada27e1b0f5cc0944e9ba.json deleted file mode 100644 index c5740cfb88..0000000000 --- a/backend/.sqlx/query-7f421bdf3dc4d47b36457af703ce69ef1e0784e9f25ada27e1b0f5cc0944e9ba.json +++ /dev/null @@ -1,16 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "UPDATE dbt_environment_state SET script_path = $3\n WHERE workspace_id = $1 AND script_path = $2", - "describe": { - "columns": [], - "parameters": { - "Left": [ - "Text", - "Text", - "Varchar" - ] - }, - "nullable": [] - }, - "hash": "7f421bdf3dc4d47b36457af703ce69ef1e0784e9f25ada27e1b0f5cc0944e9ba" -} diff --git a/backend/.sqlx/query-9f5516e218edc641e113bb4d44b9e3f9d6c179d15be2e7dc1797fe823a354284.json b/backend/.sqlx/query-9f5516e218edc641e113bb4d44b9e3f9d6c179d15be2e7dc1797fe823a354284.json new file mode 100644 index 0000000000..34d9f9dd04 --- /dev/null +++ b/backend/.sqlx/query-9f5516e218edc641e113bb4d44b9e3f9d6c179d15be2e7dc1797fe823a354284.json @@ -0,0 +1,21 @@ +{ + "db_name": "PostgreSQL", + "query": "INSERT INTO dbt_environment_state (workspace_id, script_path, environment, job_id,\n manifest, manifest_key, run_results, run_results_key,\n updated_at)\n SELECT $1::varchar, $2::varchar, $3::text, $4::uuid, $5::text, $6::text, $7::text,\n $8::text, now()\n WHERE EXISTS (SELECT 1 FROM script\n WHERE workspace_id = $1 AND path = $2\n AND deleted = false AND archived = false\n AND language = 'dbt')\n ON CONFLICT (workspace_id, script_path, environment) DO UPDATE SET\n job_id = EXCLUDED.job_id, manifest = EXCLUDED.manifest,\n manifest_key = EXCLUDED.manifest_key, run_results = EXCLUDED.run_results,\n run_results_key = EXCLUDED.run_results_key, updated_at = now()", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Varchar", + "Varchar", + "Text", + "Uuid", + "Text", + "Text", + "Text", + "Text" + ] + }, + "nullable": [] + }, + "hash": "9f5516e218edc641e113bb4d44b9e3f9d6c179d15be2e7dc1797fe823a354284" +} diff --git a/backend/.sqlx/query-bb03dea80d5baab9bbc7e251c471f9f148cbba8e0fecd86203eb40dfe8cf66dd.json b/backend/.sqlx/query-bb03dea80d5baab9bbc7e251c471f9f148cbba8e0fecd86203eb40dfe8cf66dd.json deleted file mode 100644 index acfa995916..0000000000 --- a/backend/.sqlx/query-bb03dea80d5baab9bbc7e251c471f9f148cbba8e0fecd86203eb40dfe8cf66dd.json +++ /dev/null @@ -1,21 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "INSERT INTO dbt_environment_state (workspace_id, script_path, environment, job_id,\n manifest, manifest_key, run_results, run_results_key,\n updated_at)\n VALUES ($1, $2, $3, $4, $5, $6, $7, $8, now())\n ON CONFLICT (workspace_id, script_path, environment) DO UPDATE SET\n job_id = EXCLUDED.job_id, manifest = EXCLUDED.manifest,\n manifest_key = EXCLUDED.manifest_key, run_results = EXCLUDED.run_results,\n run_results_key = EXCLUDED.run_results_key, updated_at = now()", - "describe": { - "columns": [], - "parameters": { - "Left": [ - "Varchar", - "Varchar", - "Text", - "Uuid", - "Text", - "Text", - "Text", - "Text" - ] - }, - "nullable": [] - }, - "hash": "bb03dea80d5baab9bbc7e251c471f9f148cbba8e0fecd86203eb40dfe8cf66dd" -} diff --git a/backend/windmill-common/src/dbt_manifest.rs b/backend/windmill-common/src/dbt_manifest.rs index 241a7befd0..c457cb9c84 100644 --- a/backend/windmill-common/src/dbt_manifest.rs +++ b/backend/windmill-common/src/dbt_manifest.rs @@ -1118,14 +1118,20 @@ pub async fn clear_dbt_editor_graphs( Ok(()) } -/// Move a dbt script's saved state to its new path: the run `dbt retry` resumes, -/// and the state each environment's deferrals resolve through. +/// Follow a renamed dbt script: the run `dbt retry` resumes travels with it, the +/// state its deferrals resolve through does not. /// -/// Keyed by path like the sidecar, but unlike the sidecar neither is -/// regenerated by anything: the deploy re-ingests a manifest, while these are -/// the results of runs that already happened. Clearing on rename would throw -/// away a resumable failure, and every deferral until the next full run, for a -/// cosmetic change — so they travel instead. +/// The retry state travels because nothing regenerates it — the deploy +/// re-ingests a manifest, while that is the result of a run that already +/// happened — so clearing it would throw away a resumable failure for a cosmetic +/// change. +/// +/// The environment state is cleared instead, because an artifact of it too large +/// for its row lives in object storage under a key derived from the PATH. A moved +/// row would keep pointing at that key while a script created at the old path +/// publishes over it, and the renamed project would then defer through an +/// unrelated project's manifest. The next successful run republishes; one +/// deferral is the price of a rename. /// /// See the mutator contract above: this authorizes nothing. pub async fn move_dbt_script_state( @@ -1135,7 +1141,7 @@ pub async fn move_dbt_script_state( new_path: &str, ) -> Result<()> { // The destination may already hold state from a script that lived there - // before; the incoming rows are the newer truth for this project. + // before; the incoming row is the newer truth for this project. sqlx::query!( "DELETE FROM dbt_run_state WHERE workspace_id = $1 AND script_path = $2", workspace_id, @@ -1152,18 +1158,10 @@ pub async fn move_dbt_script_state( .execute(&mut **tx) .await?; sqlx::query!( - "DELETE FROM dbt_environment_state WHERE workspace_id = $1 AND script_path = $2", + "DELETE FROM dbt_environment_state + WHERE workspace_id = $1 AND script_path = ANY($2)", workspace_id, - new_path - ) - .execute(&mut **tx) - .await?; - sqlx::query!( - "UPDATE dbt_environment_state SET script_path = $3 - WHERE workspace_id = $1 AND script_path = $2", - workspace_id, - old_path, - new_path + &[old_path.to_string(), new_path.to_string()][..] ) .execute(&mut **tx) .await?; diff --git a/backend/windmill-common/tests/dbt_graph_storage.rs b/backend/windmill-common/tests/dbt_graph_storage.rs index be81a55221..ce109be9e8 100644 --- a/backend/windmill-common/tests/dbt_graph_storage.rs +++ b/backend/windmill-common/tests/dbt_graph_storage.rs @@ -236,7 +236,11 @@ async fn clearing_one_version_leaves_the_others(db: Pool) { // this is where two versions coexist: it pins the batched edge insert // against a real database as well as the version scoping. assert_eq!(edges_for(&db, 1).await, 0, "the cleared version's edges go"); - assert_eq!(edges_for(&db, 2).await, 1, "the other version keeps its own"); + assert_eq!( + edges_for(&db, 2).await, + 1, + "the other version keeps its own" + ); } /// The routes that hard-delete a path clear no graph rows: they delete the @@ -364,7 +368,11 @@ async fn only_the_newest_deploys_keep_their_graph(db: Pool) { // The newest is always among them: losing the live version's graph would // empty the page of every run of it. assert_eq!(nodes_for(&db, over, DEPLOYED_GRAPH).await, 1); - assert_eq!(nodes_for(&db, 1, DEPLOYED_GRAPH).await, 0, "the oldest is reclaimed"); + assert_eq!( + nodes_for(&db, 1, DEPLOYED_GRAPH).await, + 0, + "the oldest is reclaimed" + ); } /// The third provenance: a `parse` of the EDITOR's buffer, which names no @@ -481,17 +489,27 @@ async fn a_version_clear_spares_editor_graphs_and_a_path_clear_does_not(db: Pool replace_dbt_editor_graph(&mut tx, WS, PATH, job, ME, &manifest(&["a"]), "root") .await .unwrap(); - clear_dbt_manifest_version(&mut tx, WS, PATH, 1).await.unwrap(); + clear_dbt_manifest_version(&mut tx, WS, PATH, 1) + .await + .unwrap(); tx.commit().await.unwrap(); assert_eq!(nodes_for(&db, 1, DEPLOYED_GRAPH).await, 0); - assert_eq!(editor_nodes(&db, job).await, 1, "the buffer's graph survives"); + assert_eq!( + editor_nodes(&db, job).await, + 1, + "the buffer's graph survives" + ); let mut tx = db.begin().await.unwrap(); clear_dbt_editor_graphs(&mut tx, WS, PATH).await.unwrap(); tx.commit().await.unwrap(); - assert_eq!(editor_nodes(&db, job).await, 0, "retiring the path takes it"); + assert_eq!( + editor_nodes(&db, job).await, + 0, + "retiring the path takes it" + ); } /// A preview names its own PATH and needs only `jobs:run`, so a bound over the @@ -555,34 +573,39 @@ async fn editor_markers(db: &Pool) -> i64 { } /// A deferral resolves a `ref()` through the manifest of the last successful run -/// at this path, so that state has to follow the script the way the retry state -/// does: a rename must not strand it, and a path a dbt script has left must not -/// hand its manifest to whatever is created there next. +/// at this path, and an oversized one lives in object storage under a key derived +/// from that path. So a rename takes the state rather than moving it — a moved +/// row would point at a key a script created at the old path publishes over — +/// and a path no live dbt version occupies must not hand its manifest to whatever +/// is created there next. #[sqlx::test(migrations = "../migrations", fixtures("base"))] -async fn environment_state_follows_the_script(db: Pool) { +async fn a_rename_clears_environment_state_rather_than_moving_it(db: Pool) { const MOVED: &str = "f/test/renamed"; deploy_script(&db, 1).await; publish_environment_state(&db, PATH).await; let mut tx = db.begin().await.unwrap(); - move_dbt_script_state(&mut tx, WS, PATH, MOVED).await.unwrap(); + move_dbt_script_state(&mut tx, WS, PATH, MOVED) + .await + .unwrap(); tx.commit().await.unwrap(); assert_eq!(environment_states(&db, PATH).await, 0); - assert_eq!(environment_states(&db, MOVED).await, 1); + assert_eq!(environment_states(&db, MOVED).await, 0); - // Still live at the old path as far as `script` is concerned, so a clear + // Still live at this path as far as `script` is concerned, so a clear // conditioned on retirement leaves it be. + publish_environment_state(&db, PATH).await; let mut tx = db.begin().await.unwrap(); clear_dbt_script_state_if_path_retired(&mut tx, WS, PATH) .await .unwrap(); tx.commit().await.unwrap(); - assert_eq!(environment_states(&db, MOVED).await, 1); + assert_eq!(environment_states(&db, PATH).await, 1); let mut tx = db.begin().await.unwrap(); - clear_dbt_script_state(&mut tx, WS, MOVED).await.unwrap(); + clear_dbt_script_state(&mut tx, WS, PATH).await.unwrap(); tx.commit().await.unwrap(); - assert_eq!(environment_states(&db, MOVED).await, 0); + assert_eq!(environment_states(&db, PATH).await, 0); } async fn publish_environment_state(db: &Pool, path: &str) { diff --git a/backend/windmill-worker/src/dbt_executor.rs b/backend/windmill-worker/src/dbt_executor.rs index dd7d70af6a..60d816519a 100644 --- a/backend/windmill-worker/src/dbt_executor.rs +++ b/backend/windmill-worker/src/dbt_executor.rs @@ -398,8 +398,7 @@ pub(crate) async fn handle_dbt_job( prepared.engine.engine.as_str() ))); } - let deferral = - prepare_deferral(&prepared, &job.workspace_id, &job.id, job_dir, conn, client).await?; + let deferral = prepare_deferral(&prepared, &job.workspace_id, job_dir, conn).await?; append_logs( &job.id, &job.workspace_id, @@ -600,9 +599,17 @@ pub(crate) async fn handle_dbt_job( if run.is_ok() && command == "build" && prepared.graph_refresh.publishes_ownership() { // Losing it costs the next deferral, not the run that just finished — // but silently, so the one actionable case (an artifact too large for - // the database on a workspace with no object storage) says so. - if let Err(e) = - crate::dbt_state::publish(&prepared, &job.workspace_id, &job.id, conn, client).await + // the database on an instance with no object storage) says so. + if let Err(e) = crate::dbt_state::publish( + &prepared, + &job.workspace_id, + &job.id, + // An attempt was spent, so `run_results.json` on disk is the one + // `dbt retry` left: the nodes it redid, not the build. + node_retry.is_some_and(|p| retries_left < p.attempts()), + conn, + ) + .await { append_logs( &job.id, @@ -988,6 +995,11 @@ pub struct PreparedProject { /// The descriptor's `profile.target`, passed as `--target` so it applies to /// a project-owned `profiles.yml` as well as a rendered one. pub target: Option, + /// The target dbt actually runs, which is the above only when the descriptor + /// names one: otherwise it is the workspace warehouse's, or the project's own + /// `profiles.yml` default. Half of an environment's identity, since a + /// `target.name` macro decides where a model is built. + pub effective_target: Option, /// 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, @@ -1157,8 +1169,8 @@ pub(crate) async fn prepare_project( .chain(invocation_env.iter().map(|(k, v)| (k.clone(), v.clone()))) .collect(); - let (profiles_dir, warehouse, adapter, default_database, default_schema, profile_digest) = - write_profiles(descriptor, &project_dir, job_dir, client, &template_env).await?; + let profile = write_profiles(descriptor, &project_dir, job_dir, client, &template_env).await?; + let adapter = profile.adapter.clone(); // The lockfile's version, when it pinned one for this same engine — a // descriptor edited to another engine invalidates the pin. let pinned_version = locks @@ -1279,18 +1291,19 @@ pub(crate) async fn prepare_project( h.finish() }, sandbox_config, - profile_digest, + profile_digest: profile.digest, project_dir, - profiles_dir, + profiles_dir: profile.dir, engine, graph_refresh, - warehouse, + warehouse: profile.warehouse, target: descriptor.profile.target.clone(), + effective_target: profile.target, descriptor_content: descriptor_content.to_string(), descriptor_env, - default_database, - default_schema, + default_database: profile.database, + default_schema: profile.schema, script_path: script_path.to_string(), env, }; @@ -1609,20 +1622,29 @@ async fn strip_git_remote(dir: &Path) -> std::io::Result<()> { /// the project itself. Both paths are supported (decision 8): the workspace /// warehouse is the ergonomic one, the project's own file is what makes an /// existing repo run unchanged. +/// What resolving the run's connection settled, beyond the file itself. +struct ResolvedProfile { + dir: PathBuf, + /// The workspace warehouse's NAME, when this project belongs to one. + warehouse: Option, + adapter: DbtAdapter, + database: Option, + schema: Option, + /// The target dbt actually runs, which is not always the descriptor's: it + /// falls back to the workspace warehouse's, and to the project's own + /// `profiles.yml` default. Resolved because it is half of an environment's + /// identity and a `target.name` macro can move every relation. + target: Option, + digest: String, +} + async fn write_profiles( descriptor: &DbtDescriptor, project_dir: &Path, job_dir: &str, client: &AuthedClient, template_env: &HashMap, -) -> error::Result<( - PathBuf, - Option, - DbtAdapter, - Option, - Option, - String, -)> { +) -> error::Result { // The workspace's warehouse, always: a descriptor names one by NAME or takes // `main`, and cannot name a resource at all. The NAME is what asset identity // keys on, so every project on one warehouse shares its nodes while the @@ -1703,14 +1725,15 @@ async fn write_profiles( } None => None, }; - return Ok(( + return Ok(ResolvedProfile { dir, - identity, + warehouse: identity, adapter, - target.database, - target.schema, - profile_digest, - )); + database: target.database, + schema: target.schema, + target: Some(target.name), + digest: profile_digest, + }); } use windmill_common::workspaces::DBT_PROFILE_RESOURCE_TYPE; @@ -1805,14 +1828,15 @@ async fn write_profiles( rendered.root_certificate_pem.as_deref(), &client.token, ); - Ok(( + Ok(ResolvedProfile { dir, - Some(warehouse.to_string()), + warehouse: Some(warehouse.to_string()), adapter, - rendered.database, - rendered.schema, - profile_digest, - )) + database: rendered.database, + schema: rendered.schema, + target: Some(target.to_string()), + digest: profile_digest, + }) } /// Where a workspace warehouse name points: its resource path and, if the @@ -1959,7 +1983,20 @@ async fn adapter_from_profiles_yml( .map(|v| v.to_string()) .filter(|v| !v.is_empty() && !v.contains("{{")) }; - Ok(ProfileTarget { adapter, database: read(database_key), schema: read(schema_key) }) + Ok(ProfileTarget { + adapter, + database: read(database_key), + schema: read(schema_key), + // The output actually chosen, which for a templated `target:` is the sole + // one rather than the template text no output answers to. + name: match ( + templated_target, + outputs.as_mapping().and_then(|m| m.keys().next()), + ) { + (true, Some(only)) => only.as_str().unwrap_or(target).to_string(), + _ => target.to_string(), + }, + }) } /// What a project-owned `profiles.yml` target says, for the two things Windmill @@ -1970,6 +2007,8 @@ struct ProfileTarget { adapter: DbtAdapter, database: Option, schema: Option, + /// The output this resolved to, by name. + name: String, } lazy_static::lazy_static! { diff --git a/backend/windmill-worker/src/dbt_state.rs b/backend/windmill-worker/src/dbt_state.rs index a911d7eb17..d07a91a351 100644 --- a/backend/windmill-worker/src/dbt_state.rs +++ b/backend/windmill-worker/src/dbt_state.rs @@ -16,7 +16,6 @@ use std::path::{Path, PathBuf}; use uuid::Uuid; -use windmill_common::client::AuthedClient; use windmill_common::error::{self, Error}; use windmill_common::worker::Connection; @@ -51,7 +50,10 @@ pub(crate) const STATE_DIR: &str = "wm_dbt_state"; pub(crate) fn environment(p: &PreparedProject) -> String { environment_key( p.warehouse.as_deref(), - p.target.as_deref(), + // The target dbt RUNS, not the descriptor's: it falls back to the + // workspace warehouse's and to the project's own default, so reading the + // descriptor's would put two inherited targets under one empty name. + p.effective_target.as_deref(), &p.relation_root(), ) } @@ -86,12 +88,26 @@ pub(crate) struct StoredState { /// Called for a run that BUILT what the script's own descriptor selects and /// succeeded (see `handle_dbt_job`). Best-effort in the same sense as the retry /// state: losing it costs the next deferral, not the run that just finished. +/// +/// **What the artifacts may carry follows from that condition.** A publishing run +/// added nothing of its own — no `select` or `vars` override, and a descriptor +/// interpolating a `{{ }}` placeholder into `vars` never publishes at all — so +/// dbt's `run_results.json` records the descriptor's own arguments, which are the +/// script's content. That is why this is keyed by environment where +/// `dbt_run_state` is keyed by principal: the retry state holds whatever a caller +/// submitted, this holds what the script says. Widen the publish condition and +/// that stops being true. pub(crate) async fn publish( p: &PreparedProject, w_id: &str, job_id: &Uuid, + // A build recovered by the automatic in-job node retry has a + // `run_results.json` naming only the nodes that retry redid. The manifest is + // unaffected — it is a function of the project, not of what ran — so the + // state is published without results rather than with a set describing some + // other slice of the build. + results_are_partial: bool, conn: &Connection, - client: &AuthedClient, ) -> error::Result<()> { let Connection::Sql(db) = conn else { // An agent worker reaches the database only through the API, which does @@ -110,44 +126,60 @@ pub(crate) async fn publish( let Ok(manifest) = tokio::fs::read_to_string(artifacts.join("manifest.json")).await else { return Ok(()); }; - let run_results = tokio::fs::read_to_string(artifacts.join("run_results.json")) - .await - .ok(); + let run_results = match results_are_partial { + true => None, + false => tokio::fs::read_to_string(artifacts.join("run_results.json")) + .await + .ok(), + }; let environment = environment(p); + // One publisher per environment at a time. The object keys are derived, so + // two of them would otherwise interleave their uploads and leave a manifest + // from one run beside results from another; the lock is also what makes the + // row and the objects agree once this commits. A reader between an upload and + // this commit still sees the older row's `job_id` over the newer manifest, + // which describes the same project in the same environment — versioned keys + // would move that window to a pointer at an object a reader may already have + // been about to fetch, and buy a grace period to sweep. + let mut tx = db.begin().await?; + sqlx::query!( + "SELECT 1 as _e FROM dbt_environment_state + WHERE workspace_id = $1 AND script_path = $2 AND environment = $3 FOR UPDATE", + w_id, + &p.script_path, + environment + ) + .fetch_optional(&mut *tx) + .await?; let manifest = store( manifest, "manifest.json", &environment, &p.script_path, w_id, - job_id, - client, - conn, ) .await?; let run_results = match run_results { - Some(r) => Some( - store( - r, - "run_results.json", - &environment, - &p.script_path, - w_id, - job_id, - client, - conn, - ) - .await?, - ), + Some(r) => Some(store(r, "run_results.json", &environment, &p.script_path, w_id).await?), None => None, }; let (manifest, manifest_key) = split(Some(manifest)); let (run_results, run_results_key) = split(run_results); + // Only while a live dbt version stays at this path, exactly as the retry + // state is saved: a job finishing after its script was renamed, archived, + // deleted or converted to another language would otherwise recreate state at + // a path no dbt script occupies, for whatever is created there next to defer + // through. sqlx::query!( "INSERT INTO dbt_environment_state (workspace_id, script_path, environment, job_id, manifest, manifest_key, run_results, run_results_key, updated_at) - VALUES ($1, $2, $3, $4, $5, $6, $7, $8, now()) + SELECT $1::varchar, $2::varchar, $3::text, $4::uuid, $5::text, $6::text, $7::text, + $8::text, now() + WHERE EXISTS (SELECT 1 FROM script + WHERE workspace_id = $1 AND path = $2 + AND deleted = false AND archived = false + AND language = 'dbt') ON CONFLICT (workspace_id, script_path, environment) DO UPDATE SET job_id = EXCLUDED.job_id, manifest = EXCLUDED.manifest, manifest_key = EXCLUDED.manifest_key, run_results = EXCLUDED.run_results, @@ -161,8 +193,9 @@ pub(crate) async fn publish( run_results, run_results_key, ) - .execute(db) + .execute(&mut *tx) .await?; + tx.commit().await?; Ok(()) } @@ -170,9 +203,7 @@ pub(crate) async fn publish( pub(crate) async fn load( p: &PreparedProject, w_id: &str, - job_id: &Uuid, conn: &Connection, - client: &AuthedClient, ) -> error::Result> { let Connection::Sql(db) = conn else { return Err(Error::BadRequest( @@ -196,19 +227,10 @@ pub(crate) async fn load( else { return Ok(None); }; - let Some(manifest) = fetch(row.manifest, row.manifest_key, w_id, job_id, client, conn).await? - else { + let Some(manifest) = fetch(row.manifest, row.manifest_key).await? else { return Ok(None); }; - let run_results = fetch( - row.run_results, - row.run_results_key, - w_id, - job_id, - client, - conn, - ) - .await?; + let run_results = fetch(row.run_results, row.run_results_key).await?; Ok(Some(StoredState { manifest, run_results, @@ -282,9 +304,13 @@ fn split(home: Option) -> (Option, Option) { /// /// Derived from the row's own key rather than randomly, so a republish /// overwrites in place and the store holds one object per artifact per -/// environment however many times a project runs. Digested because a Windmill -/// path and a schema name may both carry characters an object key gives meaning -/// to. +/// environment however many times a project runs — no versions to sweep, and no +/// window where the row points at an object a reader is about to find gone. +/// Digested because a Windmill path and a schema name may both carry characters +/// an object key gives meaning to. +/// +/// Being derived from the PATH is why a rename clears the row rather than moving +/// it (`move_dbt_script_state`): the object cannot move with it. fn object_key(w_id: &str, script_path: &str, environment: &str, artifact: &str) -> String { format!( "wmill_dbt_state/{w_id}/{}/{artifact}", @@ -293,28 +319,24 @@ fn object_key(w_id: &str, script_path: &str, environment: &str, artifact: &str) } /// Put an artifact where its size says it belongs. -#[allow(clippy::too_many_arguments)] async fn store( value: String, artifact: &str, environment: &str, script_path: &str, w_id: &str, - job_id: &Uuid, - client: &AuthedClient, - conn: &Connection, ) -> error::Result { if value.len() <= *DBT_STATE_INLINE_MAX_BYTES { return Ok(Home::Inline(value)); } let key = object_key(w_id, script_path, environment, artifact); let size = value.len(); - if put_object(&key, value, w_id, job_id, client, conn).await? { + if put_object(&key, value).await? { return Ok(Home::Stored(key)); } Err(Error::BadRequest(format!( "this project's {artifact} is {}, past the {} this instance keeps in the database, and \ - the workspace has no object storage configured to hold it. Configure workspace object \ + this instance has no object storage configured to hold it. Configure instance object \ storage, or raise DBT_STATE_INLINE_MAX_BYTES", mib(size), mib(*DBT_STATE_INLINE_MAX_BYTES), @@ -326,59 +348,29 @@ fn mib(bytes: usize) -> String { } /// Read an artifact back from whichever home the row names. -async fn fetch( - inline: Option, - key: Option, - w_id: &str, - job_id: &Uuid, - client: &AuthedClient, - conn: &Connection, -) -> error::Result> { +async fn fetch(inline: Option, key: Option) -> error::Result> { match (inline, key) { (Some(inline), _) => Ok(Some(inline)), - (None, Some(key)) => get_object(&key, w_id, job_id, client, conn).await.map(Some), + (None, Some(key)) => get_object(&key).await.map(Some), (None, None) => Ok(None), } } -/// The workspace's object storage, or `None` where it is not configured or not -/// reachable. Failing to reach it is reported here and answered by the caller: -/// a store falls back to the size error, and a fetch to "the state is no longer -/// where the row says it is". -#[cfg(feature = "parquet")] -async fn workspace_store( - w_id: &str, - job_id: &Uuid, - client: &AuthedClient, - conn: &Connection, -) -> Option> { - let Connection::Sql(db) = conn else { - return None; - }; - let resource = crate::common::get_workspace_s3_resource_path(db, client, w_id, None, job_id) - .await - .inspect_err(|e| tracing::warn!("dbt: resolving the workspace object storage: {e:#}")) - .ok() - .flatten()?; - windmill_object_store::build_object_store_client(&resource) - .await - .inspect_err(|e| tracing::warn!("dbt: reaching the workspace object storage: {e:#}")) - .ok() -} - -/// Whether the artifact was stored. `false` means the workspace has no object -/// storage this worker can write to. -#[cfg(feature = "parquet")] -async fn put_object( - key: &str, - value: String, - w_id: &str, - job_id: &Uuid, - client: &AuthedClient, - conn: &Connection, -) -> error::Result { +/// Whether the artifact was stored. `false` means this instance has no object +/// storage to put it in. +/// +/// The INSTANCE store, where every other internal worker artifact lives — bun +/// bundles, python wheels, job logs, the global cache. Not the workspace's: +/// that bucket is the one workspace members read and write through +/// `job_helpers/*` and `wmill.write_s3_file`, so a manifest there is one any +/// member could replace, and the next deferring run would hand dbt an +/// attacker-chosen `defer_relation` for every unbuilt `ref()` while holding the +/// script's warehouse credentials. Its compiled SQL would be readable there too, +/// for a project the reader may have no access to. +#[cfg(all(feature = "enterprise", feature = "parquet"))] +async fn put_object(key: &str, value: String) -> error::Result { use windmill_object_store::object_store_reexports::Path as ObjectPath; - let Some(store) = workspace_store(w_id, job_id, client, conn).await else { + let Some(store) = windmill_object_store::get_object_store().await else { return Ok(false); }; store @@ -388,15 +380,9 @@ async fn put_object( Ok(true) } -#[cfg(feature = "parquet")] -async fn get_object( - key: &str, - w_id: &str, - job_id: &Uuid, - client: &AuthedClient, - conn: &Connection, -) -> error::Result { - let Some(store) = workspace_store(w_id, job_id, client, conn).await else { +#[cfg(all(feature = "enterprise", feature = "parquet"))] +async fn get_object(key: &str) -> error::Result { + let Some(store) = windmill_object_store::get_object_store().await else { return Err(missing_storage()); }; let bytes = windmill_object_store::attempt_fetch_bytes(store, key).await?; @@ -404,35 +390,22 @@ async fn get_object( .map_err(|e| Error::internal_err(format!("the stored dbt state is not valid UTF-8: {e}"))) } -/// A build without `parquet` carries no object-store client at all, so an -/// oversized artifact has nowhere but the row and a row naming a key was written -/// by a worker that did. -#[cfg(not(feature = "parquet"))] -async fn put_object( - _key: &str, - _value: String, - _w_id: &str, - _job_id: &Uuid, - _client: &AuthedClient, - _conn: &Connection, -) -> error::Result { +/// A build without the instance store carries no client at all, so an oversized +/// artifact has nowhere but the row, and a row naming a key was written by a +/// worker that did have one. +#[cfg(not(all(feature = "enterprise", feature = "parquet")))] +async fn put_object(_key: &str, _value: String) -> error::Result { Ok(false) } -#[cfg(not(feature = "parquet"))] -async fn get_object( - _key: &str, - _w_id: &str, - _job_id: &Uuid, - _client: &AuthedClient, - _conn: &Connection, -) -> error::Result { +#[cfg(not(all(feature = "enterprise", feature = "parquet")))] +async fn get_object(_key: &str) -> error::Result { Err(missing_storage()) } fn missing_storage() -> Error { Error::BadRequest( - "the dbt state for this environment is in the workspace's object storage, which this \ + "the dbt state for this environment is in the instance's object storage, which this \ worker cannot reach: it is no longer configured, or this worker was built without \ object-storage support" .to_string(), @@ -457,10 +430,8 @@ pub(crate) struct Deferral { pub(crate) async fn prepare_deferral( p: &PreparedProject, w_id: &str, - job_id: &Uuid, job_dir: &str, conn: &Connection, - client: &AuthedClient, ) -> error::Result { if p.script_path.is_empty() { return Err(Error::BadRequest( @@ -470,11 +441,14 @@ pub(crate) async fn prepare_deferral( .to_string(), )); } - let Some(state) = load(p, w_id, job_id, conn, client).await? else { + 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 \ - build has no relation to resolve to. Run this script once without `defer`: a \ - successful run of the descriptor's own selection publishes it", + build has no relation to resolve to. It is published by a successful run that adds \ + nothing of its own: one overriding `select` or `vars` does not publish, and neither \ + does any run of a descriptor that interpolates a `{{{{ }}}}` placeholder into `vars` \ + or a `$var:` into `env` — those describe a model set the caller's arguments decided. \ + Run this script once without `defer` and without overrides", environment(p) ))); }; diff --git a/docs/dbt-runtime.md b/docs/dbt-runtime.md index 20b7b9791c..f7eb8b0981 100644 --- a/docs/dbt-runtime.md +++ b/docs/dbt-runtime.md @@ -1164,8 +1164,14 @@ Success is half of the contract: a relation a later run defers to has to exist. ### The environment is the warehouse, the target and where they resolve to `|||` — the workspace warehouse's name and -the descriptor's `profile.target`, plus the database and schema that target -resolves to, which is the same `relation_root` the graph's drift check reads. +the target dbt actually runs, plus the database and schema that target resolves +to, which is the same `relation_root` the graph's drift check reads. + +The target is the EFFECTIVE one, not the descriptor's `profile.target`: a +descriptor naming none inherits the workspace warehouse's, or the default in the +project's own `profiles.yml`, so reading the descriptor's would file every +inherited target under one empty name — and a `target.name` macro decides where a +model is built. The last two are in the key because deferring is resolving a relation NAME. A warehouse repointed at another database, or a `profile.schema` moved by a @@ -1203,29 +1209,63 @@ environment's state is therefore the last full successful build, exactly as dbt Cloud's "last successful run" is, and a run recovered by a retry leaves it at the previous one. +The AUTOMATIC in-job node retry is the same artifact under a different name: a +build it recovers is a successful build, but the `run_results.json` on disk is +the retry's. Such a run publishes the manifest **without** results, rather than +with a set describing some other slice of the build — the manifest is a function +of the project rather than of what ran, so deferral is unaffected and only +`result:` selectors lose their input. + Under `test_behavior: after_all` the stored `run_results.json` is the test phase's, because that is what the second invocation leaves in the target directory — the same artifact a local `dbt run && dbt test` leaves behind. +**What that condition means for what the artifacts may carry**, and why this +table is keyed by environment where `dbt_run_state` is keyed by principal. dbt +records the invocation's flags into `run_results.json`, and Windmill resolves +`$var:` / `$res:` references before dbt sees them — which is exactly why the +retry state is per-principal, so one caller's resolved `select` and `vars` are +not restorable by the next. Here they cannot be one caller's: a publishing run +added nothing of its own, and a descriptor that interpolates a `{{ }}` +placeholder into `vars` never publishes at all, so what is recorded is the +descriptor's own arguments — the script's content, which anyone entitled to run +it may already read. Widen the publish condition and that stops being true. + ### Where the blob goes `run_results.json` is small; `manifest.json` is not, and grows with the project (535 KB on a two-model fixture). Each takes the same two homes: inline in the row -under `DBT_STATE_INLINE_MAX_BYTES` (8 MiB), and the workspace's object storage +under `DBT_STATE_INLINE_MAX_BYTES` (8 MiB), and the INSTANCE's object storage above it, with the row keeping the key. Inline is what makes the feature work on -a workspace that has configured no storage at all; the ceiling is what stops one +an instance that has configured no storage at all; the ceiling is what stops one project's manifest from becoming a multi-megabyte row rewritten by every run. A project past the ceiling with no storage configured is told so, in the job log, naming the setting and the variable — the run itself still succeeds, since losing the state costs the next deferral rather than the build that just ran. +**The instance store, not the workspace's**, which is where every other internal +worker artifact already lives (bun bundles, python wheels, job logs, the global +cache). The workspace bucket is the one members read and write through +`job_helpers/*` and `wmill.write_s3_file` with a caller-supplied key, and only +`volumes/` is reserved there — so a manifest under it is one any member could +replace, and the next deferring run would hand dbt an attacker-chosen +`defer_relation` for every unbuilt `ref()` while holding the script's warehouse +credentials. Its compiled SQL would be readable there too, for a project the +reader may have no access to. The consequence to know: a project past the ceiling +needs the instance store configured, which is an EE feature, so on CE the ceiling +is the limit and `DBT_STATE_INLINE_MAX_BYTES` is how it moves. + The object key is derived from the path and the environment (`wmill_dbt_state///`), so a republish overwrites in place and the store holds one object per artifact per environment however many -times the project runs. A blob whose row is gone — the script was deleted, or the -project grew past the ceiling and then shrank back under it — stays in the -bucket, as a script bundle does; the key is derived rather than random precisely -so the project coming back reuses it instead of accumulating a second one. +times the project runs — no versions to sweep, and no window where the row points +at an object a reader is about to find gone. Publishers of one environment +serialize on the row (`FOR UPDATE`) so two of them cannot interleave their +uploads and leave one run's manifest beside another's results. A reader between +an upload and its commit still sees the older row's `job_id` over the newer +manifest, which describes the same project in the same environment. A blob whose +row is gone — the script was deleted, or the project grew past the ceiling and +then shrank back under it — stays in the store, as a script bundle does. ### Retention @@ -1234,10 +1274,19 @@ door. Those are pruned by age by the dbt runs themselves because their reader is a transient run page. This one holds a single row per script per environment, replaced in place, so it does not grow with runs — and its reader is every later run of that script, so a project that runs monthly must still find last month's -state. It goes with the script instead: a rename moves it, and a path no live dbt -version occupies any more clears it, alongside `dbt_run_state` -(`move_dbt_script_state`, `clear_dbt_script_state`, -`clear_dbt_script_state_if_path_retired`). +state. It goes with the script instead: a path no live dbt version occupies any +more clears it, alongside `dbt_run_state` (`clear_dbt_script_state`, +`clear_dbt_script_state_if_path_retired`). The same guard is on the write: a job +finishing after its script was renamed, archived, deleted or converted to another +language publishes nothing, so it cannot recreate state at a path for whatever is +created there next to defer through. + +A RENAME is where the two halves part. The retry state travels, because nothing +regenerates it. The environment state is cleared, because an oversized artifact's +key is derived from the path: a moved row would keep pointing at a key a script +created at the old path publishes over, and the renamed project would then defer +through an unrelated project's manifest. The next successful run republishes, so +one deferral is the price of a rename. ### Asking for it