fix: address the local review of the dbt environment state

The oversized-artifact home moves from the workspace's object storage to the
instance's, where every other internal worker artifact already lives. The
workspace bucket is the one members read and write through `job_helpers/*` 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.

The environment key takes the target dbt actually runs rather than the
descriptor's `profile.target`, which is absent whenever the target is inherited
from the workspace warehouse or the project's own `profiles.yml` — filing every
inherited target under one empty name, while a `target.name` macro decides where
a model is built. `write_profiles` returns a named struct now that it resolves
one more thing.

Publishing takes the row's lock before uploading, so two publishers of one
environment cannot interleave their uploads and leave one run's manifest beside
another's results, and carries the live-dbt-script guard the retry state already
had, so a job finishing after its script was renamed, archived or deleted cannot
recreate state at a path for whatever is created there next.

A rename now clears the environment state instead of moving it: an oversized
artifact's key is derived from the path, so a moved row would keep pointing at a
key a script created at the old path publishes over.

A build recovered by the automatic in-job node retry publishes its manifest
without results — `run_results.json` is then the retry's, naming only the nodes
it redid — and the refusal for an environment with nothing published names the
runs that cannot publish rather than suggesting a run that would not help.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

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"
+}

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"
+}

backend/.sqlx/query-7f421bdf3dc4d47b36457af703ce69ef1e0784e9f25ada27e1b0f5cc0944e9ba.json

@@ -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"
-}

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"
+}

backend/.sqlx/query-bb03dea80d5baab9bbc7e251c471f9f148cbba8e0fecd86203eb40dfe8cf66dd.json

@@ -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"
-}

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?;

backend/windmill-common/tests/dbt_graph_storage.rs

@@ -236,7 +236,11 @@ async fn clearing_one_version_leaves_the_others(db: Pool<Postgres>) {
     // 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<Postgres>) {
     // 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<Postgres>) -> 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<Postgres>) {
+async fn a_rename_clears_environment_state_rather_than_moving_it(db: Pool<Postgres>) {
     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<Postgres>, path: &str) {

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<String>,
+    /// 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<String>,
     /// 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<String>,
@@ -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<String>,
+    adapter: DbtAdapter,
+    database: Option<String>,
+    schema: Option<String>,
+    /// 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<String>,
+    digest: String,
+}
+
 async fn write_profiles(
     descriptor: &DbtDescriptor,
     project_dir: &Path,
     job_dir: &str,
     client: &AuthedClient,
     template_env: &HashMap<String, String>,
-) -> error::Result<(
-    PathBuf,
-    Option<String>,
-    DbtAdapter,
-    Option<String>,
-    Option<String>,
-    String,
-)> {
+) -> error::Result<ResolvedProfile> {
     // 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<String>,
     schema: Option<String>,
+    /// The output this resolved to, by name.
+    name: String,
 }
 
 lazy_static::lazy_static! {

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<Option<StoredState>> {
     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<Home>) -> (Option<String>, Option<String>) {
 ///
 /// 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<Home> {
     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<String>,
-    key: Option<String>,
-    w_id: &str,
-    job_id: &Uuid,
-    client: &AuthedClient,
-    conn: &Connection,
-) -> error::Result<Option<String>> {
+async fn fetch(inline: Option<String>, key: Option<String>) -> error::Result<Option<String>> {
     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<std::sync::Arc<dyn windmill_object_store::object_store_reexports::ObjectStore>> {
-    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<bool> {
+/// 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<bool> {
     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<String> {
-    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<String> {
+    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<bool> {
+/// 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<bool> {
     Ok(false)
 }
 
-#[cfg(not(feature = "parquet"))]
-async fn get_object(
-    _key: &str,
-    _w_id: &str,
-    _job_id: &Uuid,
-    _client: &AuthedClient,
-    _conn: &Connection,
-) -> error::Result<String> {
+#[cfg(not(all(feature = "enterprise", feature = "parquet")))]
+async fn get_object(_key: &str) -> error::Result<String> {
     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<Deferral> {
     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)
         )));
     };

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
 
 `<warehouse>|<target>|<schema>|<database>` — 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/<workspace>/<digest>/<artifact>`), 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