starred/windmill
1
0
Fork 0
mirror of https://github.com/windmill-labs/windmill.git synced 2026-09-05 16:03:47 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix: hold the script row across a dbt state publication, and let a rename move it

Browse Source 
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) <noreply@anthropic.com>
This commit is contained in:
Ruben Fiszel
2026-09-05 03:38:05 +02:00
co-authored by Claude Opus 5
parent 6f2ccbd535
commit 6f211bb6f3
12 changed files with 309 additions and 221 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/.sqlx/query-3fc12582cbae2ebc77ecfcaeed4bce43c749a44cbc4e7b8f719bf2a598ce57f3.json
Generated
+21
View File
@@ -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"
}
backend/.sqlx/query-58ec340c78af046a40296b15543beeedc63b4a941e5b17ab7bb5d3c259f05147.json
Generated
+24
View File
@@ -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"
}
backend/.sqlx/query-7c2f07ec5e3586d865af1a764f23121eea8b1a37f925ce31f941e030c00eaf5c.json
Generated
-15
View File
@@ -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"
}
backend/.sqlx/query-7f421bdf3dc4d47b36457af703ce69ef1e0784e9f25ada27e1b0f5cc0944e9ba.json
Generated
+16
View File
@@ -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"
}
backend/.sqlx/query-cc324ab8c28139f720d094ae7d46b9c703d687a8a7f3ca94733e5f12dd9f984c.json
Generated
-22
View File
@@ -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"
}
backend/.sqlx/query-f963ea31d1744ff0d76ff86140f01cca0bcbc832f01eb3340050504b8b3875a2.json
Generated
+30
View File
@@ -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"
}
backend/migrations/20260904135713_dbt_environment_state.up.sql
+6 -4
View File
@@ -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,
backend/windmill-common/src/dbt_manifest.rs
+24 -20
View File
@@ -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(
backend/windmill-common/tests/dbt_graph_storage.rs
+12 -13
View File
@@ -557,13 +557,11 @@ async fn editor_markers(db: &Pool<Postgres>) -> 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<Postgres>) {
async fn environment_state_follows_the_script(db: Pool<Postgres>) {
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<Postgr
.unwrap();
tx.commit().await.unwrap();
assert_eq!(environment_states(&db, PATH).await, 0);
assert_eq!(environment_states(&db, MOVED).await, 0);
assert_eq!(environment_states(&db, MOVED).await, 1);
// Still live at this path as far as `script` is concerned, so a clear
// Still live at the old 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, PATH).await, 1);
assert_eq!(environment_states(&db, MOVED).await, 1);
let mut tx = db.begin().await.unwrap();
clear_dbt_script_state(&mut tx, WS, PATH).await.unwrap();
clear_dbt_script_state(&mut tx, WS, MOVED).await.unwrap();
tx.commit().await.unwrap();
assert_eq!(environment_states(&db, PATH).await, 0);
assert_eq!(environment_states(&db, MOVED).await, 0);
}
/// The worker publishes under a guard naming the version that ran, and the whole
@@ -624,7 +621,9 @@ async fn a_late_job_cannot_publish_for_a_path_it_no_longer_owns(db: Pool<Postgre
);
}
/// `dbt_state::publish`'s guarded insert, reduced to what it decides.
/// The predicate `dbt_state::publish` locks the script row on, reduced to what it
/// decides. Keep the two in step — this file cannot call `publish` itself, which
/// is `pub(crate)` in `windmill-worker`.
async fn guarded_publish(db: &Pool<Postgres>, path: &str, ran: i64) -> u64 {
sqlx::query!(
"INSERT INTO dbt_environment_state (workspace_id, script_path, environment, job_id,
backend/windmill-worker/src/dbt_executor.rs
+10 -4
View File
@@ -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
backend/windmill-worker/src/dbt_state.rs
+131 -127
View File
@@ -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<Home>) -> (Option<String>, Option<String>) {
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<Home> {
) -> error::Result<(Option<String>, Option<String>)> {
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 \
docs/dbt-runtime.md
+35 -16
View File
@@ -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