From 6f211bb6f3d1484a67a8a1bd3880592af820eb4e Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Sat, 5 Sep 2026 03:38:05 +0200 Subject: [PATCH] fix: hold the script row across a dbt state publication, and let a rename move it MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The version guard read `script` without a lock, so lifecycle cleanup could find no environment row to clear, finish, and leave this transaction to commit state at a path a new script goes on to occupy. It now holds that row (`FOR SHARE`) for the rest of the publication — taken before the sidecar, the order every other dbt writer takes — and the artifacts are uploaded before the transaction, so the lock covers the row work rather than a network round trip. A commit that reports an error may still have committed: what was lost can be the acknowledgement. Dropping this run's objects then leaves the committed row naming objects that are gone, so an orphan is the cheaper side to take. A failed second upload left the manifest it had already written behind; it is dropped now. Per-publication keys retired the reason a rename cleared the environment state rather than moving it: the path is only a prefix, and the row is what names an artifact, so a script created at the old path can no longer publish over a moved row. The rename moves both halves again. `dbt ls` gets the deferral flags too, without which a `result:` selector — which reads `run_results.json` out of the state directory, and which `select` passes to dbt verbatim — fails before the build that would have honoured it. Also: the migration was the last site describing the workspace's object storage rather than the instance's, `publication_lock` folded 32 bits where it claimed 64, and `ResolvedProfile` had taken `write_profiles`'s doc block. Co-Authored-By: Claude Opus 5 (1M context) --- ...bce43c749a44cbc4e7b8f719bf2a598ce57f3.json | 21 ++ ...beeedc63b4a941e5b17ab7bb5d3c259f05147.json | 24 ++ ...3121eea8b1a37f925ce31f941e030c00eaf5c.json | 15 - ...e69ef1e0784e9f25ada27e1b0f5cc0944e9ba.json | 16 ++ ...6b9c703d687a8a7f3ca94733e5f12dd9f984c.json | 22 -- ...01cca0bcbc832f01eb3340050504b8b3875a2.json | 30 ++ ...0260904135713_dbt_environment_state.up.sql | 10 +- backend/windmill-common/src/dbt_manifest.rs | 44 +-- .../tests/dbt_graph_storage.rs | 25 +- backend/windmill-worker/src/dbt_executor.rs | 14 +- backend/windmill-worker/src/dbt_state.rs | 258 +++++++++--------- docs/dbt-runtime.md | 51 ++-- 12 files changed, 309 insertions(+), 221 deletions(-) create mode 100644 backend/.sqlx/query-3fc12582cbae2ebc77ecfcaeed4bce43c749a44cbc4e7b8f719bf2a598ce57f3.json create mode 100644 backend/.sqlx/query-58ec340c78af046a40296b15543beeedc63b4a941e5b17ab7bb5d3c259f05147.json delete mode 100644 backend/.sqlx/query-7c2f07ec5e3586d865af1a764f23121eea8b1a37f925ce31f941e030c00eaf5c.json create mode 100644 backend/.sqlx/query-7f421bdf3dc4d47b36457af703ce69ef1e0784e9f25ada27e1b0f5cc0944e9ba.json delete mode 100644 backend/.sqlx/query-cc324ab8c28139f720d094ae7d46b9c703d687a8a7f3ca94733e5f12dd9f984c.json create mode 100644 backend/.sqlx/query-f963ea31d1744ff0d76ff86140f01cca0bcbc832f01eb3340050504b8b3875a2.json diff --git a/backend/.sqlx/query-3fc12582cbae2ebc77ecfcaeed4bce43c749a44cbc4e7b8f719bf2a598ce57f3.json b/backend/.sqlx/query-3fc12582cbae2ebc77ecfcaeed4bce43c749a44cbc4e7b8f719bf2a598ce57f3.json new file mode 100644 index 0000000000..1268656e34 --- /dev/null +++ b/backend/.sqlx/query-3fc12582cbae2ebc77ecfcaeed4bce43c749a44cbc4e7b8f719bf2a598ce57f3.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,\n run_results_key, 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": "3fc12582cbae2ebc77ecfcaeed4bce43c749a44cbc4e7b8f719bf2a598ce57f3" +} diff --git a/backend/.sqlx/query-58ec340c78af046a40296b15543beeedc63b4a941e5b17ab7bb5d3c259f05147.json b/backend/.sqlx/query-58ec340c78af046a40296b15543beeedc63b4a941e5b17ab7bb5d3c259f05147.json new file mode 100644 index 0000000000..b2a0d728cd --- /dev/null +++ b/backend/.sqlx/query-58ec340c78af046a40296b15543beeedc63b4a941e5b17ab7bb5d3c259f05147.json @@ -0,0 +1,24 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT 1 FROM script\n WHERE workspace_id = $1 AND path = $2\n AND deleted = false AND archived = false AND language = 'dbt'\n AND (hash = $3 OR $3 = ANY(parent_hashes))\n FOR SHARE", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "?column?", + "type_info": "Int4" + } + ], + "parameters": { + "Left": [ + "Text", + "Text", + "Int8" + ] + }, + "nullable": [ + null + ] + }, + "hash": "58ec340c78af046a40296b15543beeedc63b4a941e5b17ab7bb5d3c259f05147" +} diff --git a/backend/.sqlx/query-7c2f07ec5e3586d865af1a764f23121eea8b1a37f925ce31f941e030c00eaf5c.json b/backend/.sqlx/query-7c2f07ec5e3586d865af1a764f23121eea8b1a37f925ce31f941e030c00eaf5c.json deleted file mode 100644 index a784a7aa39..0000000000 --- a/backend/.sqlx/query-7c2f07ec5e3586d865af1a764f23121eea8b1a37f925ce31f941e030c00eaf5c.json +++ /dev/null @@ -1,15 +0,0 @@ -{ - "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 new file mode 100644 index 0000000000..c5740cfb88 --- /dev/null +++ b/backend/.sqlx/query-7f421bdf3dc4d47b36457af703ce69ef1e0784e9f25ada27e1b0f5cc0944e9ba.json @@ -0,0 +1,16 @@ +{ + "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-cc324ab8c28139f720d094ae7d46b9c703d687a8a7f3ca94733e5f12dd9f984c.json b/backend/.sqlx/query-cc324ab8c28139f720d094ae7d46b9c703d687a8a7f3ca94733e5f12dd9f984c.json deleted file mode 100644 index a83c23e169..0000000000 --- a/backend/.sqlx/query-cc324ab8c28139f720d094ae7d46b9c703d687a8a7f3ca94733e5f12dd9f984c.json +++ /dev/null @@ -1,22 +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 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 AND (hash = $9 OR $9 = ANY(parent_hashes)))\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", - "Int8" - ] - }, - "nullable": [] - }, - "hash": "cc324ab8c28139f720d094ae7d46b9c703d687a8a7f3ca94733e5f12dd9f984c" -} diff --git a/backend/.sqlx/query-f963ea31d1744ff0d76ff86140f01cca0bcbc832f01eb3340050504b8b3875a2.json b/backend/.sqlx/query-f963ea31d1744ff0d76ff86140f01cca0bcbc832f01eb3340050504b8b3875a2.json new file mode 100644 index 0000000000..540d2e0e08 --- /dev/null +++ b/backend/.sqlx/query-f963ea31d1744ff0d76ff86140f01cca0bcbc832f01eb3340050504b8b3875a2.json @@ -0,0 +1,30 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT manifest_key, run_results_key FROM dbt_environment_state\n WHERE workspace_id = $1 AND script_path = $2 AND environment = $3", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "manifest_key", + "type_info": "Text" + }, + { + "ordinal": 1, + "name": "run_results_key", + "type_info": "Text" + } + ], + "parameters": { + "Left": [ + "Text", + "Text", + "Text" + ] + }, + "nullable": [ + true, + true + ] + }, + "hash": "f963ea31d1744ff0d76ff86140f01cca0bcbc832f01eb3340050504b8b3875a2" +} diff --git a/backend/migrations/20260904135713_dbt_environment_state.up.sql b/backend/migrations/20260904135713_dbt_environment_state.up.sql index 4dda4f3d74..a49df5b3d1 100644 --- a/backend/migrations/20260904135713_dbt_environment_state.up.sql +++ b/backend/migrations/20260904135713_dbt_environment_state.up.sql @@ -24,11 +24,13 @@ CREATE TABLE IF NOT EXISTS dbt_environment_state ( -- The run that published it, so a deferring run can say what it deferred to. job_id UUID NOT NULL, -- Exactly one home each. A manifest grows with the project and passes a few - -- hundred KB on a handful of models, so a large one goes to the workspace's + -- hundred KB on a handful of models, so a large one goes to the INSTANCE's -- object storage and this row keeps the key; a small one stays here, where it - -- costs no round trip and works on a workspace that has configured no storage - -- at all. `run_results.json` is a tenth of the size and takes the same two - -- homes rather than a rule of its own. + -- costs no round trip and works on an instance that has configured no storage + -- at all. The instance's and not the workspace's, because a member can write + -- the workspace bucket under a key of their choosing, and a manifest is what a + -- later run resolves every unbuilt `ref()` through. `run_results.json` is a + -- tenth of the size and takes the same two homes rather than a rule of its own. manifest TEXT, manifest_key TEXT, run_results TEXT, diff --git a/backend/windmill-common/src/dbt_manifest.rs b/backend/windmill-common/src/dbt_manifest.rs index b1f64522f3..86ce15f7a8 100644 --- a/backend/windmill-common/src/dbt_manifest.rs +++ b/backend/windmill-common/src/dbt_manifest.rs @@ -1118,20 +1118,15 @@ pub async fn clear_dbt_editor_graphs( Ok(()) } -/// Follow a renamed dbt script: the run `dbt retry` resumes travels with it, the -/// state its deferrals resolve through does not. +/// 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. /// -/// 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. +/// 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. An artifact too large for its row is unaffected: its +/// key is that publication's own, and the moved row is what names it. /// /// See the mutator contract above: this authorizes nothing. pub async fn move_dbt_script_state( @@ -1158,10 +1153,18 @@ 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 = ANY($2)", + "DELETE FROM dbt_environment_state WHERE workspace_id = $1 AND script_path = $2", workspace_id, - &[old_path.to_string(), new_path.to_string()][..] + 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 ) .execute(&mut **tx) .await?; @@ -1215,10 +1218,11 @@ pub async fn clear_dbt_script_state_if_path_retired( /// otherwise inherit a stranger's resumable failure and defer to a project it /// has nothing to do with. /// -/// A manifest this moves out of the database and into the instance's object -/// storage is left in the bucket, as a script bundle is: the key is derived from -/// the path and the environment, so a project that comes back reuses it rather -/// than accumulating a second one. +/// An artifact too large for its row lives in the instance's object storage, and +/// this leaves it there — as a deleted script leaves its bundle. Reaching it from +/// here would mean an object-store client in this crate and a delete that has to +/// land after the caller's transaction commits, for one object per environment of +/// a script that is gone. /// /// See the mutator contract above: this authorizes nothing. pub async fn clear_dbt_script_state( diff --git a/backend/windmill-common/tests/dbt_graph_storage.rs b/backend/windmill-common/tests/dbt_graph_storage.rs index 94ae255968..792ee923df 100644 --- a/backend/windmill-common/tests/dbt_graph_storage.rs +++ b/backend/windmill-common/tests/dbt_graph_storage.rs @@ -557,13 +557,11 @@ async fn editor_markers(db: &Pool) -> i64 { } /// A deferral resolves a `ref()` through the manifest of the last successful run -/// 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. +/// 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 no live dbt version occupies +/// must not hand its manifest to whatever is created there next. #[sqlx::test(migrations = "../migrations", fixtures("base"))] -async fn a_rename_clears_environment_state_rather_than_moving_it(db: Pool) { +async fn environment_state_follows_the_script(db: Pool) { const MOVED: &str = "f/test/renamed"; deploy_script(&db, 1).await; publish_environment_state(&db, PATH).await; @@ -574,22 +572,21 @@ async fn a_rename_clears_environment_state_rather_than_moving_it(db: Pool, path: &str, ran: i64) -> u64 { sqlx::query!( "INSERT INTO dbt_environment_state (workspace_id, script_path, environment, job_id, diff --git a/backend/windmill-worker/src/dbt_executor.rs b/backend/windmill-worker/src/dbt_executor.rs index 219e7c0ff3..dd1aa9e54e 100644 --- a/backend/windmill-worker/src/dbt_executor.rs +++ b/backend/windmill-worker/src/dbt_executor.rs @@ -1622,10 +1622,6 @@ async fn strip_git_remote(dir: &Path) -> std::io::Result<()> { tokio::fs::write(&config, out).await } -/// Write `profiles.yml`, either rendered from a Windmill resource or taken from -/// 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, @@ -1642,6 +1638,10 @@ struct ResolvedProfile { digest: String, } +/// Write `profiles.yml`, either rendered from a Windmill resource or taken from +/// 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. async fn write_profiles( descriptor: &DbtDescriptor, project_dir: &Path, @@ -3579,6 +3579,12 @@ async fn resolve_selection( return Ok(None); } let mut cmd = dbt_command(p, &["ls"]); + // The same state the build resolves through, or a `result:` selector — which + // reads `run_results.json` out of it, and which `select` passes to dbt + // verbatim — fails here, before the build that would have honoured it. + if inv.deferral.is_some() { + cmd.args(defer_flags("ls", p.engine.engine)); + } // A project whose models call `var()` without a default fails to parse // without these, so the selection resolver needs them exactly as the run // does. Placeholders that only a run can fill are dropped rather than diff --git a/backend/windmill-worker/src/dbt_state.rs b/backend/windmill-worker/src/dbt_state.rs index fe72707601..f3154a6a11 100644 --- a/backend/windmill-worker/src/dbt_state.rs +++ b/backend/windmill-worker/src/dbt_state.rs @@ -135,38 +135,11 @@ pub(crate) async fn publish( .ok(), }; let environment = environment(p); - // One publisher per environment at a time, so two of them cannot interleave - // and leave a manifest from one run beside results from another. An advisory - // lock rather than the row's, because the first publish of an environment has - // no row to lock and is exactly when two runs of a newly deployed script are - // most likely to race. - let mut tx = db.begin().await?; - sqlx::query_scalar!( - "SELECT pg_advisory_xact_lock($1)", - publication_lock(w_id, &p.script_path, &environment) - ) - .execute(&mut *tx) - .await?; - // What the row points at NOW, so those objects can go once this one is - // committed in their place — never before, since a reader that has already - // read the row is about to fetch them. - let displaced = sqlx::query!( - "SELECT manifest_key, run_results_key FROM dbt_environment_state - WHERE workspace_id = $1 AND script_path = $2 AND environment = $3", - w_id, - &p.script_path, - environment - ) - .fetch_optional(&mut *tx) - .await? - .map(|r| [r.manifest_key, r.run_results_key]) - .unwrap_or_default(); - // Every publication writes its OWN keys and the row switches to them in one - // statement, so nothing overwrites an artifact another row still names: a - // failure anywhere below leaves the committed state pointing at the objects - // it was already paired with, rather than at this run's manifest beside the - // previous run's results. - let manifest = store( + // Uploaded BEFORE the transaction, and to this publication's own keys, so two + // 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 (manifest, manifest_key) = store( manifest, "manifest.json", &environment, @@ -175,74 +148,119 @@ pub(crate) async fn publish( job_id, ) .await?; - let run_results = match run_results { - Some(r) => Some( - store( - r, - "run_results.json", - &environment, - &p.script_path, - w_id, - job_id, - ) - .await?, - ), - None => None, - }; - let (manifest, manifest_key) = split(Some(manifest)); - let (run_results, run_results_key) = split(run_results); - // Only while the live dbt version at this path is the one this job ran, or a - // later version of it. The retry state settles for "some live dbt script is - // here", which a script created at a path this one was renamed away from also - // satisfies — 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. - let published = sqlx::query!( - "INSERT INTO dbt_environment_state (workspace_id, script_path, environment, job_id, - manifest, manifest_key, run_results, run_results_key, - updated_at) - 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' - AND (hash = $9 OR $9 = ANY(parent_hashes))) - 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, - run_results_key = EXCLUDED.run_results_key, updated_at = now()", - w_id, - &p.script_path, - environment, - job_id, - manifest, - manifest_key, - run_results, - run_results_key, - script_hash, - ) - .execute(&mut *tx) - .await; - let mine = [manifest_key, run_results_key]; - match published.and_then(|r| Ok(r.rows_affected())) { - // Committed: the row names this run's objects, so the ones it displaced - // have no reader left. - Ok(1) => match tx.commit().await { - Ok(()) => forget_objects(&displaced).await, + let (run_results, run_results_key) = match run_results { + Some(r) => match store( + r, + "run_results.json", + &environment, + &p.script_path, + w_id, + job_id, + ) + .await + { + Ok(stored) => stored, Err(e) => { - forget_objects(&mine).await; - return Err(e.into()); + forget_objects(&[manifest_key, None]).await; + return Err(e); } }, - // Refused by the guard, or the write failed: the committed state is - // untouched and what was uploaded above has no row. - Ok(_) => forget_objects(&mine).await, - Err(e) => { - forget_objects(&mine).await; - return Err(e.into()); + None => (None, None), + }; + let mine = [manifest_key.clone(), run_results_key.clone()]; + // One publisher per environment at a time, so the row and the objects it + // displaces are settled by one of them at a time. An advisory lock rather + // than the row's, because the first publish of an environment has no row to + // lock and is exactly when two runs of a newly deployed script are most + // likely to race. + let mut tx = db.begin().await?; + let staged = async { + sqlx::query_scalar!( + "SELECT pg_advisory_xact_lock($1)", + publication_lock(w_id, &p.script_path, &environment) + ) + .execute(&mut *tx) + .await?; + // The script row FIRST, and held, so a rename, archive or delete of this + // path either waits for this publication or is seen by it. Reading it + // unlocked leaves a window where lifecycle cleanup finds no row to clear, + // finishes, and this transaction then commits state at a path a new + // script goes on to occupy. Script row before sidecar is also the order + // every other dbt writer takes, which is what keeps the two off a + // deadlock. + // + // The version, not just the path: "some live dbt script is here" is also + // satisfied by a script created at a path this one was renamed away from. + // A preview names no version, so `script_hash` is NULL and nothing + // matches — right for a run of content that was never deployed. + let owns_path = sqlx::query_scalar!( + "SELECT 1 FROM script + WHERE workspace_id = $1 AND path = $2 + AND deleted = false AND archived = false AND language = 'dbt' + AND (hash = $3 OR $3 = ANY(parent_hashes)) + FOR SHARE", + w_id, + &p.script_path, + script_hash, + ) + .fetch_optional(&mut *tx) + .await? + .is_some(); + if !owns_path { + return error::Result::Ok(None); } + // What the row points at NOW, so those objects can go once this one is + // committed in their place — never before, since a reader that has + // already read the row is about to fetch them. + let displaced = sqlx::query!( + "SELECT manifest_key, run_results_key FROM dbt_environment_state + WHERE workspace_id = $1 AND script_path = $2 AND environment = $3", + w_id, + &p.script_path, + environment + ) + .fetch_optional(&mut *tx) + .await? + .map(|r| [r.manifest_key, r.run_results_key]) + .unwrap_or_default(); + 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()) + 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, + run_results_key = EXCLUDED.run_results_key, updated_at = now()", + w_id, + &p.script_path, + environment, + job_id, + manifest, + manifest_key, + run_results, + run_results_key, + ) + .execute(&mut *tx) + .await?; + error::Result::Ok(Some(displaced)) } + .await; + let displaced = match staged { + // Refused by the guard, or the write failed: nothing is committed and + // what was uploaded above has no row naming it. + Ok(None) | Err(_) => { + forget_objects(&mine).await; + return staged.map(|_| ()); + } + Ok(Some(displaced)) => displaced, + }; + // A commit that reports an error may still have committed — what was lost can + // be the acknowledgement. Dropping this run's objects would then leave the + // committed row naming objects that are gone, and every deferral would fail + // until the next publication; an orphan costs storage instead. + tx.commit().await?; + forget_objects(&displaced).await; Ok(()) } @@ -344,33 +362,19 @@ pub(crate) async fn write_state_dir( }) } -/// Where a stored artifact lives. -enum Home { - /// Small enough to sit in the row. - Inline(String), - /// In the instance's object storage, under this key. - Stored(String), -} - -fn split(home: Option) -> (Option, Option) { - match home { - Some(Home::Inline(v)) => (Some(v), None), - Some(Home::Stored(k)) => (None, Some(k)), - None => (None, None), - } -} - -/// The advisory lock one environment's publishers take, so that only one of them -/// is between its first upload and its row at a time. +/// The advisory lock one environment's publishers take, so only one of them +/// settles the row and the objects it displaces at a time. /// /// Derived from the same three components as the row's key. Two environments -/// whose digests collide in 64 bits wait for each other, which costs a moment and -/// nothing else. +/// whose digests collide wait for each other, which costs a moment and nothing +/// else. fn publication_lock(w_id: &str, script_path: &str, environment: &str) -> i64 { let d = digest(&format!("{w_id}|{script_path}|{environment}")); - let mut bytes = [0u8; 8]; - bytes.copy_from_slice(&d.as_bytes()[..8]); - i64::from_be_bytes(bytes) + i64::from_str_radix(&d[..16], 16).unwrap_or_else(|_| { + // `digest` is hex, so this cannot happen; a fixed key would only queue + // every publication behind one lock rather than lose one. + i64::MIN + }) } /// The object-storage key an artifact takes. @@ -379,11 +383,10 @@ fn publication_lock(w_id: &str, script_path: &str, environment: &str) -> i64 { /// committed row still names: a run that fails between its two uploads, or /// between them and its row, leaves the state pointing at the pair it already /// had. The row switches to these in one statement and the objects it displaced -/// are dropped afterwards. Digested because a Windmill path and a schema name may -/// both carry characters an object key gives meaning to. -/// -/// The environment prefix is derived from the PATH, which is why a rename clears -/// the row rather than moving it (`move_dbt_script_state`). +/// are dropped afterwards. The path and environment are only a prefix — the row +/// 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. fn object_key( w_id: &str, script_path: &str, @@ -397,7 +400,8 @@ fn object_key( ) } -/// Put an artifact where its size says it belongs. +/// Put an artifact where its size says it belongs: `(inline, key)`, exactly one +/// of which is set. #[allow(clippy::too_many_arguments)] async fn store( value: String, @@ -406,14 +410,14 @@ async fn store( script_path: &str, w_id: &str, job_id: &Uuid, -) -> error::Result { +) -> error::Result<(Option, Option)> { if value.len() <= *DBT_STATE_INLINE_MAX_BYTES { - return Ok(Home::Inline(value)); + return Ok((Some(value), None)); } let key = object_key(w_id, script_path, environment, job_id, artifact); let size = value.len(); if put_object(&key, value).await? { - return Ok(Home::Stored(key)); + return Ok((None, Some(key))); } Err(Error::BadRequest(format!( "this project's {artifact} is {}, past the {} this instance keeps in the database, and \ diff --git a/docs/dbt-runtime.md b/docs/dbt-runtime.md index d033ee578e..a189b4c37e 100644 --- a/docs/dbt-runtime.md +++ b/docs/dbt-runtime.md @@ -1264,13 +1264,19 @@ had. The objects the commit displaced are dropped afterwards, never before, sinc a reader that has already read the row is about to fetch them; a reader that loses that race re-reads the row once rather than reporting a state that is there. What a publication uploaded and then could not commit is dropped on the -way out. +way out — except after a commit that REPORTED an error, where what was lost may +be only the acknowledgement: dropping then would leave a committed row naming +objects that are gone, so an orphan is the cheaper side to take. -Publishers of one environment serialize on `pg_advisory_xact_lock`, so two of -them cannot interleave and leave one run's manifest beside another's results — -an advisory lock rather than the row's, because the first publish of an -environment has no row to lock and is exactly when two runs of a newly deployed -script are most likely to race. +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. + +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 +rather than the row's, because the first publish of an environment has no row to +lock and is exactly when two runs of a newly deployed script are most likely to +race. ### Retention @@ -1291,12 +1297,18 @@ 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. -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. +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, +finishes, and the publication then commits state at a path a new script goes on +to occupy. The script row is taken before the sidecar, which is the order every +other dbt writer takes and what keeps the two off a deadlock. + +An artifact too large for its row is left in the store when the row is cleared, +as a deleted script leaves its bundle: reaching it from the delete would mean an +object-store client in `windmill-common` and a delete that has to land after the +caller's transaction commits, for one object per environment of a script that is +gone. ### Asking for it @@ -1315,7 +1327,10 @@ which does not expose this table. A `show` defers too, and every engine takes the flags on it. It compiles the model it previews, so a model whose upstream this environment built and this run -did not is exactly the case a deferral exists for. +did not is exactly the case a deferral exists for. So does the `dbt ls` that +resolves what a run's selection owns, without which a `result:` selector — which +reads `run_results.json` out of the state directory, and which `select` passes to +dbt verbatim — would fail before the build that would have honoured it. The result carries `deferred_to`, the run whose state was used. Without it what a deferring run built against is unrecoverable, since the next successful run of @@ -1342,14 +1357,18 @@ later `dbt retry` restores them, so an absolute path would name the job director of the run being resumed, which is gone by then. Relative, it resolves against the project root — whichever job directory the retry landed in. -Two engine facts found while wiring this up, both worth knowing before filing a +Three engine facts found while wiring this up, all worth knowing before filing a bug against the feature. `dbt retry` on dbt-core 2.x restores **neither** the resumed invocation's `--vars` nor its deferral: it re-parses with the current (empty) ones, so a retry of a run that overrode `vars` rebuilds into the descriptor's schema rather than the run's. That is independent of deferral and predates it; the refusal above stops the deferring case from being the way it is -discovered. And neither Rust engine reached dbt's own service-backed State -(`--manage-state`) on any run measured here, so no flag is passed to disable it. +discovered. `dbt show` on either Rust engine prints a bare JSON array where +dbt-core frames it as `{"node": …, "show": […]}`, which `run_show` is written +against — so a preview there fails to parse whether or not it defers, and the +deferral itself resolves correctly under it. And neither Rust engine reached +dbt's own service-backed State (`--manage-state`) on any run measured here, so no +flag is passed to disable it. Because `select` reaches dbt verbatim, a deferring run also has a `--state` directory for `result:` selectors, which is why `run_results.json` is stored