fix: address review — preview stamping, stale producer set, public doc

Three findings from the local review round:

- Record the warehouse write only for a DEPLOYED script job. The annotation is
  a deploy-time contract (`manual`, three segments, a configured warehouse)
  checked where write access to the path is also required; honouring it in a
  preview, hub or inline-flow body let `jobs:run` alone restamp any relation's
  last writer from a script that never touched it.
- Exclude the deploying script's own rows from the producer set. Read
  committed, they describe the version being replaced, so a script dropping its
  `// materialize` while adding a subscription counted itself as the producer
  that would wake it and committed a dormant edge. It could not be that
  producer anyway — the dispatcher skips self-loops.
- `AssetKind::Dbt`'s doc no longer claims dbt is the exclusive producer of a
  warehouse relation, on both the types and the parser enum.

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

backend/.sqlx/query-63c0b60d1042ef74e69b11119355c26d8ca6e5e94902450c11ab7816a385fd9b.json

@@ -1,6 +1,6 @@
 {
   "db_name": "PostgreSQL",
-  "query": "SELECT s.path AS \"path!\", s.language AS \"language!: ScriptLang\"\n             FROM asset a\n             JOIN script s ON s.workspace_id = a.workspace_id AND s.path = a.usage_path\n                          AND s.archived = false AND s.deleted = false\n            WHERE a.workspace_id = $1 AND a.kind = 'dbt' AND a.path = $2\n              AND a.usage_kind = 'script' AND a.usage_access_type IN ('w', 'rw')",
+  "query": "SELECT s.path AS \"path!\", s.language AS \"language!: ScriptLang\"\n             FROM asset a\n             JOIN script s ON s.workspace_id = a.workspace_id AND s.path = a.usage_path\n                          AND s.archived = false AND s.deleted = false\n            WHERE a.workspace_id = $1 AND a.kind = 'dbt' AND a.path = $2\n              AND a.usage_kind = 'script' AND a.usage_access_type IN ('w', 'rw')\n              AND a.usage_path <> $3",
   "describe": {
     "columns": [
       {
@@ -49,6 +49,7 @@
     ],
     "parameters": {
       "Left": [
+        "Text",
         "Text",
         "Text"
       ]
@@ -58,5 +59,5 @@
       false
     ]
   },
-  "hash": "6bdddcaa371df626c17b4d70d5e1b7a09c4c4ed7435fa2b23aa1a95acf0476e0"
+  "hash": "63c0b60d1042ef74e69b11119355c26d8ca6e5e94902450c11ab7816a385fd9b"
 }

backend/parsers/windmill-parser/src/asset_parser.rs

@@ -29,10 +29,10 @@ pub enum AssetKind {
     Ducklake,
     DataTable,
     Volume,
-    /// A warehouse relation a dbt project builds or reads,
-    /// `dbt://<warehouse>/<schema>/<name>`, the warehouse named as the
-    /// workspace configures it. The scheme names the producer, the path stays
-    /// the relation — see `windmill_types::AssetKind::Dbt`.
+    /// A warehouse relation, `dbt://<warehouse>/<schema>/<name>`, the warehouse
+    /// named as the workspace configures it. The scheme names the namespace dbt
+    /// made — a script of any language can declare a write to one — and the path
+    /// stays the relation. See `windmill_types::AssetKind::Dbt`.
     Dbt,
 }
 
@@ -1742,10 +1742,7 @@ mod pipeline_annotation_tests {
     // just stop being the same node and the cross-boundary cascade never fires.
     #[test]
     fn table_paths_from_every_spelling_canonicalize_to_one_key() {
-        let canonical = Some((
-            AssetKind::Dbt,
-            Cow::Owned("main/analytics/orders".into()),
-        ));
+        let canonical = Some((AssetKind::Dbt, Cow::Owned("main/analytics/orders".into())));
         for spelling in [
             // Hand-written annotation.
             "dbt://main/analytics/orders",
@@ -1793,10 +1790,7 @@ mod pipeline_annotation_tests {
         // database qualifier.
         assert_eq!(
             parse_asset_syntax("dbt://main/\"sales.v2\"/orders", false),
-            Some((
-                AssetKind::Dbt,
-                Cow::Owned("main/sales.v2/orders".into())
-            ))
+            Some((AssetKind::Dbt, Cow::Owned("main/sales.v2/orders".into())))
         );
     }
 
@@ -1815,14 +1809,8 @@ mod pipeline_annotation_tests {
                 "dbt://main/analytics/\"order\"\"s\"",
                 "main/analytics/order\"s",
             ),
-            (
-                "dbt://main/`da``ta`/`orders`",
-                "main/da`ta/orders",
-            ),
-            (
-                "dbt://main/[my]]schema]/[orders]",
-                "main/my]schema/orders",
-            ),
+            ("dbt://main/`da``ta`/`orders`", "main/da`ta/orders"),
+            ("dbt://main/[my]]schema]/[orders]", "main/my]schema/orders"),
             // And in one half of a database-qualified segment.
             (
                 "dbt://main/\"arch\"\"ive\".\"sales\"/orders",
@@ -1841,8 +1829,14 @@ mod pipeline_annotation_tests {
         // apart. A lone delimiter treated as opening a quote would be dropped —
         // `sa"les` filed as `sales` — and the two derivations would split.
         for (decoded, spelled) in [
-            ("dbt://main/sa\"les/orders", "dbt://main/\"sa\"\"les\"/orders"),
-            ("dbt://main/analytics/order\"s", "dbt://main/analytics/\"order\"\"s\""),
+            (
+                "dbt://main/sa\"les/orders",
+                "dbt://main/\"sa\"\"les\"/orders",
+            ),
+            (
+                "dbt://main/analytics/order\"s",
+                "dbt://main/analytics/\"order\"\"s\"",
+            ),
             (
                 "dbt://main/arch\"ive.sales/orders",
                 "dbt://main/\"arch\"\"ive\".\"sales\"/orders",

backend/windmill-api-scripts/src/scripts.rs

@@ -2441,7 +2441,8 @@ async fn create_script_internal<'c>(
                 )));
             }
             if let Some(dbt_owner) =
-                windmill_common::assets::sole_dbt_producer(&db, &w_id, relation).await?
+                windmill_common::assets::sole_dbt_producer(&db, &w_id, relation, &ns.path)
+                    .await?
             {
                 return Err(Error::BadRequest(format!(
                     "`{trigger_ref}` cannot be subscribed to: it is built by the dbt project at \

backend/windmill-common/src/assets.rs

@@ -258,13 +258,20 @@ pub fn derive_pipeline_asset_trigger_refs(
 /// for every `dbt://` node anyway (the source that script wrote stays gated).
 /// Callers must therefore already be scoped to `workspace_id`.
 ///
-/// Reads committed rows only, so a caller inside a deploy transaction does not
-/// see its own not-yet-committed writes: a script subscribing to a relation it
-/// also materializes is left to the dispatcher's self-loop skip.
+/// `subscriber_path` is excluded from the producer set, and has to be: reading
+/// committed rows means the deploying script's own are the version being
+/// replaced, so one that just dropped its `// materialize` would still count as a
+/// producer and let a now-dormant subscription through. Excluding it is free of
+/// the opposite error, because a script never wakes its own subscription — the
+/// dispatcher skips that as a self-loop.
+///
+/// A producer another deploy is committing concurrently is still invisible, so
+/// that race resolves toward refusing with a message the user can retry past.
 pub async fn sole_dbt_producer<'e>(
     executor: impl PgExecutor<'e>,
     workspace_id: &str,
     asset_path: &str,
+    subscriber_path: &str,
 ) -> error::Result<Option<String>> {
     use crate::scripts::ScriptLang;
     let producers = sqlx::query!(
@@ -273,9 +280,11 @@ pub async fn sole_dbt_producer<'e>(
              JOIN script s ON s.workspace_id = a.workspace_id AND s.path = a.usage_path
                           AND s.archived = false AND s.deleted = false
             WHERE a.workspace_id = $1 AND a.kind = 'dbt' AND a.path = $2
-              AND a.usage_kind = 'script' AND a.usage_access_type IN ('w', 'rw')"#,
+              AND a.usage_kind = 'script' AND a.usage_access_type IN ('w', 'rw')
+              AND a.usage_path <> $3"#,
         workspace_id,
-        asset_path
+        asset_path,
+        subscriber_path
     )
     .fetch_all(executor)
     .await?;

backend/windmill-common/tests/dbt_producer_rules.rs

@@ -15,6 +15,7 @@ use windmill_common::assets::{dormant_dbt_subscriptions, sole_dbt_producer};
 
 const WS: &str = "test-workspace";
 const RELATION: &str = "main/analytics/orders";
+const SUBSCRIBER: &str = "u/test-user/consumer";
 
 async fn plant_producer(db: &Pool<Postgres>, path: &str, language: &str, hash: i64) {
     sqlx::query(
@@ -58,7 +59,9 @@ async fn plant_subscriber(db: &Pool<Postgres>, path: &str) {
 #[sqlx::test(migrations = "../migrations", fixtures("base"))]
 async fn no_producer_is_not_dormant(db: Pool<Postgres>) {
     assert_eq!(
-        sole_dbt_producer(&db, WS, RELATION).await.unwrap(),
+        sole_dbt_producer(&db, WS, RELATION, SUBSCRIBER)
+            .await
+            .unwrap(),
         None,
         "a relation nothing produces yet must not refuse the subscription"
     );
@@ -68,7 +71,9 @@ async fn no_producer_is_not_dormant(db: Pool<Postgres>) {
 async fn dbt_only_producer_is_dormant(db: Pool<Postgres>) {
     plant_producer(&db, "u/test-user/project", "dbt", 1).await;
     assert_eq!(
-        sole_dbt_producer(&db, WS, RELATION).await.unwrap(),
+        sole_dbt_producer(&db, WS, RELATION, SUBSCRIBER)
+            .await
+            .unwrap(),
         Some("u/test-user/project".to_string())
     );
 }
@@ -77,7 +82,12 @@ async fn dbt_only_producer_is_dormant(db: Pool<Postgres>) {
 async fn a_native_producer_beside_dbt_is_not_dormant(db: Pool<Postgres>) {
     plant_producer(&db, "u/test-user/project", "dbt", 1).await;
     plant_producer(&db, "u/test-user/ingest", "postgresql", 2).await;
-    assert_eq!(sole_dbt_producer(&db, WS, RELATION).await.unwrap(), None);
+    assert_eq!(
+        sole_dbt_producer(&db, WS, RELATION, SUBSCRIBER)
+            .await
+            .unwrap(),
+        None
+    );
 }
 
 /// `asset` is keyed by path while `script` holds every version of it, so the
@@ -92,7 +102,25 @@ async fn a_superseded_native_version_does_not_count(db: Pool<Postgres>) {
         .expect("archive the old version");
     plant_producer(&db, "u/test-user/project", "dbt", 2).await;
     assert_eq!(
-        sole_dbt_producer(&db, WS, RELATION).await.unwrap(),
+        sole_dbt_producer(&db, WS, RELATION, SUBSCRIBER)
+            .await
+            .unwrap(),
+        Some("u/test-user/project".to_string())
+    );
+}
+
+/// The rows of the script being deployed describe the version it replaces, so a
+/// script dropping its `// materialize` while adding a subscription would
+/// otherwise count itself as the producer that wakes it — and commit a dormant
+/// edge. It can never be that producer anyway: the dispatcher skips self-loops.
+#[sqlx::test(migrations = "../migrations", fixtures("base"))]
+async fn the_subscriber_is_never_its_own_producer(db: Pool<Postgres>) {
+    plant_producer(&db, "u/test-user/project", "dbt", 1).await;
+    plant_producer(&db, SUBSCRIBER, "postgresql", 2).await;
+    assert_eq!(
+        sole_dbt_producer(&db, WS, RELATION, SUBSCRIBER)
+            .await
+            .unwrap(),
         Some("u/test-user/project".to_string())
     );
 }

backend/windmill-types/src/assets.rs

@@ -14,17 +14,19 @@ pub enum AssetKind {
     Ducklake,
     DataTable,
     Volume,
-    /// A warehouse relation a dbt project builds or reads,
-    /// `dbt://<warehouse>/<schema>/<name>`, where `<warehouse>` is the name the
-    /// workspace configures it under.
+    /// A warehouse relation, `dbt://<warehouse>/<schema>/<name>`, where
+    /// `<warehouse>` is the name the workspace configures it under.
     ///
-    /// The SCHEME names the producer — dbt is the only thing that creates one —
-    /// while the PATH stays the physical relation, because that is what two
-    /// projects agree on: a mart one builds is a `source` the next reads, and
-    /// their dbt `unique_id`s differ (`model.a.orders` vs
-    /// `source.b.analytics.orders`) where the relation does not
-    /// (docs/dbt-runtime.md, decision 11). A dbt run does not trigger that
-    /// reader — the shared node is lineage, not a cascade edge.
+    /// The SCHEME names the namespace dbt made rather than an exclusive
+    /// producer: dbt is what derives these relations from a project, and a
+    /// script of any language can DECLARE one it writes
+    /// (`// materialize manual dbt://…`). The PATH stays the physical relation,
+    /// because that is what two producers agree on: a mart one builds is a
+    /// `source` the next reads, and their dbt `unique_id`s differ
+    /// (`model.a.orders` vs `source.b.analytics.orders`) where the relation does
+    /// not (docs/dbt-runtime.md, decision 11). A dbt run does not trigger the
+    /// readers of what it built — that shared node is lineage, not a cascade
+    /// edge — while a declared write does (decision 25).
     Dbt,
 }
 

backend/windmill-worker/src/worker.rs

@@ -5583,6 +5583,15 @@ async fn record_declared_warehouse_write(
     use windmill_common::materialization::{
         MaterializationStatus, RecordMaterializationRequest, UNPARTITIONED,
     };
+    // A DEPLOYED script only. The annotation is a deploy-time contract — `manual`,
+    // a three-segment relation, a configured warehouse — checked in
+    // `create_script_internal`, which also required write access to the path. A
+    // preview, hub or inline-flow body reaches this function without any of that,
+    // so honouring it there would let `jobs:run` alone restamp any relation's last
+    // writer from a script that never touched it.
+    if job.kind != JobKind::Script {
+        return;
+    }
     // Cheap guard: the annotation scan is skipped for the overwhelming majority
     // of jobs, which carry no `materialize` line at all.
     if !code.contains("materialize") {

docs/dbt-runtime.md

@@ -715,11 +715,17 @@ deploy-order-independent syncs. A dbt script may not subscribe at all: its graph
 ingest clears its own `dbt://` trigger rows, so accepting one would deploy an edge
 the dependency job then silently removes.
 
-That leaves one ordering the deploy cannot catch: a subscriber accepted while the
-relation had no producer, and a dbt project deployed afterwards that claims it. So
-a dbt deploy names those edges in its own log rather than leaving them silently
-dormant — the same "an edge that can never fire is worse than saying so" the
-refusal is for, at the only other point where it is knowable.
+The producer set is read as it stands committed, minus the deploying script's own
+rows — those describe the version being replaced, so a script dropping its
+`// materialize` while adding a subscription would otherwise count itself as the
+producer that wakes it, which it could not be anyway (the dispatcher skips
+self-loops).
+
+What that leaves is a subscription accepted while it was live and later orphaned:
+a dbt project deployed afterwards that claims the relation, or a native producer
+that stops writing it. A dbt deploy names those edges in its own log rather than
+leaving them silently dormant — the same "an edge that can never fire is worse
+than saying so" the refusal is for, at the only other point where it is knowable.
 
 A plain READ still renders the consumer beside the model, which is what makes
 the lineage one graph — but it is written in the script's own code, not in a