diff --git a/backend/.sqlx/query-0ee4a4dc8c6f09e7027bd99f98d8d1d346f9d67056ef272314dd9b80bee9f285.json b/backend/.sqlx/query-0ee4a4dc8c6f09e7027bd99f98d8d1d346f9d67056ef272314dd9b80bee9f285.json new file mode 100644 index 0000000000..238522a3ff --- /dev/null +++ b/backend/.sqlx/query-0ee4a4dc8c6f09e7027bd99f98d8d1d346f9d67056ef272314dd9b80bee9f285.json @@ -0,0 +1,77 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT DISTINCT ON (mp.asset_kind, mp.asset_path)\n mp.asset_kind AS \"asset_kind: AssetKind\", mp.asset_path,\n mp.snapshot_id AS \"snapshot_id!\", mp.partition\n FROM materialized_partition mp\n JOIN unnest($2::ASSET_KIND[], $3::text[]) AS u(kind, path)\n ON mp.asset_kind = u.kind AND mp.asset_path = u.path\n WHERE mp.workspace_id = $1\n AND mp.status = 'materialized' AND mp.snapshot_id IS NOT NULL\n ORDER BY mp.asset_kind, mp.asset_path, mp.snapshot_id DESC", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "asset_kind: AssetKind", + "type_info": { + "Custom": { + "name": "asset_kind", + "kind": { + "Enum": [ + "s3object", + "resource", + "variable", + "ducklake", + "datatable", + "volume" + ] + } + } + } + }, + { + "ordinal": 1, + "name": "asset_path", + "type_info": "Varchar" + }, + { + "ordinal": 2, + "name": "snapshot_id!", + "type_info": "Int8" + }, + { + "ordinal": 3, + "name": "partition", + "type_info": "Text" + } + ], + "parameters": { + "Left": [ + "Text", + { + "Custom": { + "name": "asset_kind[]", + "kind": { + "Array": { + "Custom": { + "name": "asset_kind", + "kind": { + "Enum": [ + "s3object", + "resource", + "variable", + "ducklake", + "datatable", + "volume" + ] + } + } + } + } + } + }, + "TextArray" + ] + }, + "nullable": [ + false, + false, + true, + false + ] + }, + "hash": "0ee4a4dc8c6f09e7027bd99f98d8d1d346f9d67056ef272314dd9b80bee9f285" +} diff --git a/backend/.sqlx/query-3703847d0cef9df3a480af8b6a5c21b6af282c68f06ce037a11f703284d8980d.json b/backend/.sqlx/query-3703847d0cef9df3a480af8b6a5c21b6af282c68f06ce037a11f703284d8980d.json new file mode 100644 index 0000000000..3a784490ef --- /dev/null +++ b/backend/.sqlx/query-3703847d0cef9df3a480af8b6a5c21b6af282c68f06ce037a11f703284d8980d.json @@ -0,0 +1,23 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT DISTINCT trigger_ref AS \"trigger_ref!\"\n FROM script_trigger\n WHERE workspace_id = $1\n AND runnable_path = $2\n AND trigger_kind = 'asset'\n AND runnable_kind = 'script'\n ORDER BY trigger_ref", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "trigger_ref!", + "type_info": "Text" + } + ], + "parameters": { + "Left": [ + "Text", + "Text" + ] + }, + "nullable": [ + false + ] + }, + "hash": "3703847d0cef9df3a480af8b6a5c21b6af282c68f06ce037a11f703284d8980d" +} diff --git a/backend/tests/asset_trigger_dispatch.rs b/backend/tests/asset_trigger_dispatch.rs index 198472d404..7184d4810d 100644 --- a/backend/tests/asset_trigger_dispatch.rs +++ b/backend/tests/asset_trigger_dispatch.rs @@ -1044,3 +1044,132 @@ async fn reaper_clears_only_stale_join_slots(db: Pool) -> anyhow::Resu Ok(()) } + +/// Seed one `materialized_partition` row (the state a managed `// materialize` +/// write records) so dispatch has a snapshot to look up. +async fn seed_materialization( + db: &Pool, + kind: &str, + asset_path: &str, + partition: &str, + status: &str, + snapshot_id: Option, +) -> anyhow::Result<()> { + sqlx::query( + r#"INSERT INTO materialized_partition + (workspace_id, asset_kind, asset_path, partition, status, snapshot_id) + VALUES ($1, $2::asset_kind, $3, $4, $5::materialization_status, $6)"#, + ) + .bind(WS) + .bind(kind) + .bind(asset_path) + .bind(partition) + .bind(status) + .bind(snapshot_id) + .execute(db) + .await?; + Ok(()) +} + +/// Dispatch records, on the consumer's `trigger` arg, the latest captured +/// materialization snapshot of each of its direct upstream assets +/// (`upstream_snapshots`) — the forensic "what did this run see" record. +/// Covered here: +/// - latest = highest `snapshot_id` with status `materialized` (a stale +/// partition and a failed/no-snapshot row are both passed over), +/// - whole-table (sentinel '') upstreams omit `partition`, +/// - upstreams with no captured snapshot produce no entry, +/// - a consumer with no materialized upstream at all gets no +/// `upstream_snapshots` key. +#[sqlx::test(fixtures("base"))] +async fn upstream_snapshots_recorded_on_dispatch(db: Pool) -> anyhow::Result<()> { + initialize_tracing().await; + + seed_script(&db, SUB_S3, "echo lake consumer", "bash").await?; + seed_script(&db, SUB_RES, "echo raw consumer", "bash").await?; + seed_asset_write(&db, PRODUCER, "ducklake", "analytics/orders").await?; + seed_asset_write(&db, PRODUCER, "s3object", "f/raw").await?; + + // SUB_S3's direct upstream set: the firing ducklake asset, a second + // materialized ducklake dimension (written by some other producer), and a + // plain s3 object that is never materialized. + seed_subscription(&db, SUB_S3, "script", "ducklake://analytics/orders").await?; + seed_subscription(&db, SUB_S3, "script", "ducklake://analytics/customers").await?; + seed_subscription(&db, SUB_S3, "script", "s3://f/raw").await?; + // SUB_RES subscribes only to the raw (non-materialized) asset. + seed_subscription(&db, SUB_RES, "script", "s3://f/raw").await?; + + // orders: an older partition, the latest one, and a failed slice with no + // snapshot — only 2026-06-19 @ 42 must be recorded. + seed_materialization( + &db, + "ducklake", + "analytics/orders", + "2026-06-18", + "materialized", + Some(41), + ) + .await?; + seed_materialization( + &db, + "ducklake", + "analytics/orders", + "2026-06-19", + "materialized", + Some(42), + ) + .await?; + seed_materialization( + &db, + "ducklake", + "analytics/orders", + "2026-06-20", + "failed", + None, + ) + .await?; + // customers: unpartitioned (sentinel '') → entry without `partition`. + seed_materialization( + &db, + "ducklake", + "analytics/customers", + "", + "materialized", + Some(7), + ) + .await?; + + let id = seed_producer_job(&db, json!({})).await?; + let r = dispatch_asset_triggers(&db, &make_mini(id, PRODUCER)).await; + assert_eq!( + r.dispatched.len(), + 3, + "SUB_S3 fired for both written assets, SUB_RES for the raw one" + ); + + let expected_snaps = json!([ + { "asset": "ducklake://analytics/customers", "snapshot_id": 7 }, + { "asset": "ducklake://analytics/orders", "snapshot_id": 42, "partition": "2026-06-19" }, + ]); + for (path, _, args) in fetch_dispatched(&db).await? { + let trigger = args + .as_ref() + .and_then(|a| a.get("trigger")) + .cloned() + .expect("dispatched job carries a trigger arg"); + match path.as_str() { + SUB_S3 => assert_eq!( + trigger.get("upstream_snapshots"), + Some(&expected_snaps), + "latest materialized snapshot per upstream, sorted by ref, raw asset absent" + ), + SUB_RES => assert!( + trigger.get("upstream_snapshots").is_none(), + "no materialized upstream → no upstream_snapshots key" + ), + other => panic!("unexpected dispatched path {other}"), + } + } + + Ok(()) +} diff --git a/backend/windmill-queue/src/asset_dispatch.rs b/backend/windmill-queue/src/asset_dispatch.rs index 753614d07b..29f44b153d 100644 --- a/backend/windmill-queue/src/asset_dispatch.rs +++ b/backend/windmill-queue/src/asset_dispatch.rs @@ -37,21 +37,31 @@ //! "asset_path": "...", //! "producer_path": "...", //! "producer_job_id": "...", -//! "chain": ["f/a/producer0", "f/a/producer1"] +//! "chain": ["f/a/producer0", "f/a/producer1"], +//! "upstream_snapshots": [ +//! { "asset": "ducklake://analytics/orders", "snapshot_id": 42 } +//! ] //! } //! } //! ``` //! +//! `upstream_snapshots` is a forensic record only (present when at least one +//! direct upstream has a captured materialization snapshot): it says which +//! substrate version each of the subscriber's `// on` assets was at when the +//! job was dispatched, so a failing run can be replayed against DuckLake +//! time-travel (`AT (VERSION => n)`). Nothing pins the consumer's reads to it. +//! //! Errors are logged but never bubble up to fail the producer's job. use crate::{push, MiniCompletedJob, PushArgs, PushIsolationLevel}; +use serde::Serialize; use serde_json::value::RawValue; use sqlx::types::Json; use sqlx::{Pool, Postgres}; use std::collections::HashMap; use std::sync::Arc; use uuid::Uuid; -use windmill_common::assets::AssetKind; +use windmill_common::assets::{parse_asset_trigger_ref, AssetKind}; use windmill_common::error::{self, Result}; use windmill_common::get_latest_deployed_hash_for_path; use windmill_common::jobs::{JobKind, JobPayload, JobTriggerKind}; @@ -286,6 +296,10 @@ async fn try_dispatch(db: &DB, job: &MiniCompletedJob) -> Result // subscriber × asset write). The mid-pass join-slot writes // (record_and_check_join_slot) are a separate table and unaffected. let mut events: Vec = Vec::new(); + // A subscriber listening to several of this producer's writes is pushed + // once per edge; its upstream-snapshot record is identical across those + // pushes (same instant, same trigger set), so resolve it once per pass. + let mut snapshot_memo: HashMap>> = HashMap::new(); for (asset_kind, asset_path) in writes { let Some(prefix) = asset_kind.canonical_prefix() else { continue; @@ -361,6 +375,29 @@ async fn try_dispatch(db: &DB, job: &MiniCompletedJob) -> Result } } } + // Forensic upstream-state capture, resolved at dispatch time (a + // debounced job that gets superseded is re-pushed by the later + // arrival, which re-resolves — the surviving job records what its + // own dispatch saw). Best-effort: a lookup failure must not stop + // the cascade. + let snapshots = match snapshot_memo.get(&sub_path) { + Some(s) => s.clone(), + None => { + let s = Arc::new( + upstream_snapshots(db, &job.workspace_id, &sub_path) + .await + .unwrap_or_else(|e| { + tracing::error!( + "upstream-snapshot lookup failed for {}: {e:#}", + sub_path + ); + Vec::new() + }), + ); + snapshot_memo.insert(sub_path.clone(), s.clone()); + s + } + }; match push_subscriber( db, job, @@ -373,6 +410,7 @@ async fn try_dispatch(db: &DB, job: &MiniCompletedJob) -> Result debounce_s, retry_count, retry_delay_s, + &snapshots, ) .await { @@ -552,6 +590,91 @@ async fn workspace_producer_writes( Ok(map) } +/// Forensic record of one direct upstream's state at dispatch time: the +/// latest captured materialization snapshot of an asset in the subscriber's +/// `// on` trigger set. Serialized into the dispatched job's `trigger` arg +/// (`upstream_snapshots`) so a failing consumer run stays debuggable against +/// DuckLake time-travel. Record-only — the consumer's reads are not pinned. +#[derive(Debug, Serialize)] +struct UpstreamSnapshot { + /// Canonical asset uri, e.g. `ducklake://analytics/orders_daily`. + asset: String, + snapshot_id: i64, + /// Partition whose write produced this snapshot — i.e. the latest slice + /// written, not necessarily the slice this consumer processes. The + /// snapshot itself is table-global. Omitted for whole-table + /// materializations. + #[serde(skip_serializing_if = "Option::is_none")] + partition: Option, +} + +/// Latest captured snapshot per direct upstream of `subscriber_path`: its +/// asset trigger set joined against `materialized_partition`, keeping the +/// highest `snapshot_id` per asset (the newest substrate version the consumer +/// could read). Assets with no captured snapshot (non-materialized upstreams) +/// simply produce no entry. Two queries total regardless of upstream count. +async fn upstream_snapshots( + db: &Pool, + workspace_id: &str, + subscriber_path: &str, +) -> Result> { + let refs = sqlx::query_scalar!( + r#"SELECT DISTINCT trigger_ref AS "trigger_ref!" + FROM script_trigger + WHERE workspace_id = $1 + AND runnable_path = $2 + AND trigger_kind = 'asset' + AND runnable_kind = 'script' + ORDER BY trigger_ref"#, + workspace_id, + subscriber_path, + ) + .fetch_all(db) + .await?; + // Keep only refs with a recognized asset prefix, preserving ref order so + // the recorded list is deterministic. + let parsed: Vec<(String, AssetKind, String)> = refs + .into_iter() + .filter_map(|r| parse_asset_trigger_ref(&r).map(|(k, p)| (r, k, p))) + .collect(); + if parsed.is_empty() { + return Ok(Vec::new()); + } + let kinds: Vec = parsed.iter().map(|(_, k, _)| *k).collect(); + let paths: Vec = parsed.iter().map(|(_, _, p)| p.clone()).collect(); + let rows = sqlx::query!( + r#"SELECT DISTINCT ON (mp.asset_kind, mp.asset_path) + mp.asset_kind AS "asset_kind: AssetKind", mp.asset_path, + mp.snapshot_id AS "snapshot_id!", mp.partition + FROM materialized_partition mp + JOIN unnest($2::ASSET_KIND[], $3::text[]) AS u(kind, path) + ON mp.asset_kind = u.kind AND mp.asset_path = u.path + WHERE mp.workspace_id = $1 + AND mp.status = 'materialized' AND mp.snapshot_id IS NOT NULL + ORDER BY mp.asset_kind, mp.asset_path, mp.snapshot_id DESC"#, + workspace_id, + kinds as Vec, + &paths, + ) + .fetch_all(db) + .await?; + let mut latest: HashMap<(AssetKind, String), (i64, String)> = rows + .into_iter() + .map(|r| ((r.asset_kind, r.asset_path), (r.snapshot_id, r.partition))) + .collect(); + Ok(parsed + .into_iter() + .filter_map(|(trigger_ref, kind, path)| { + let (snapshot_id, partition) = latest.remove(&(kind, path))?; + Some(UpstreamSnapshot { + asset: trigger_ref, + snapshot_id, + partition: (!partition.is_empty()).then_some(partition), + }) + }) + .collect()) +} + /// A subscriber row resolved from `script_trigger`. Bundles the per-edge /// options (debounce) and the script-level policy fields (`join_all`, /// retry) that travel together to dispatch. @@ -614,6 +737,7 @@ async fn push_subscriber( debounce_s: Option, retry_count: Option, retry_delay_s: Option, + upstream_snapshots: &[UpstreamSnapshot], ) -> Result { // Same resolution as every other trigger path (`script_path_to_payload`): // latest deployed hash plus the script's own runnable settings @@ -700,7 +824,7 @@ async fn push_subscriber( }; let mut args: HashMap> = HashMap::new(); - let trigger_payload = serde_json::json!({ + let mut trigger_payload = serde_json::json!({ "kind": "asset", "asset_kind": serde_json::to_value(&asset_kind).expect("AssetKind serializes"), "asset_path": asset_path, @@ -709,6 +833,10 @@ async fn push_subscriber( CHAIN_KEY: chain, PARTITION_ARG: partition, }); + if !upstream_snapshots.is_empty() { + trigger_payload["upstream_snapshots"] = + serde_json::to_value(upstream_snapshots).expect("UpstreamSnapshot serializes"); + } args.insert(TRIGGER_ARG.to_string(), to_raw_value(&trigger_payload)); // Carry the producer's resolved partition forward as a top-level arg so // the subscriber's body can read it and the next cascade hop's diff --git a/docs/ducklake-materialization.md b/docs/ducklake-materialization.md index e7a52dbf80..4481d7555b 100644 --- a/docs/ducklake-materialization.md +++ b/docs/ducklake-materialization.md @@ -268,6 +268,14 @@ without anyone asking. This is deliberately **not** built, for three reasons: If a workload ever shows the consistency race in practice, pinning can be layered on top — the capture and the snapshot surfacing built here are its foundation. +What **is** built is the forensic slice of that sketch: when the cascade +dispatches a consumer, it records the latest captured snapshot of each of the +consumer's direct upstream assets into the job's `trigger` arg +(`upstream_snapshots`, rendered read-only on the run detail page with a +copyable `AT (VERSION => n)` clause). "What did the failing run actually see" +stays answerable after the fact, with zero read-path changes — reads are +*not* pinned to the recorded versions. + This is distinct from **SCD2 history** (`// materialize … key=… history`), which *is* built: DuckLake time-travel answers "what did the whole table look like at snapshot N?" but not "give me each entity's version history as queryable rows" diff --git a/frontend/src/lib/components/runs/UpstreamSnapshotsPanel.svelte b/frontend/src/lib/components/runs/UpstreamSnapshotsPanel.svelte new file mode 100644 index 0000000000..ee265d5874 --- /dev/null +++ b/frontend/src/lib/components/runs/UpstreamSnapshotsPanel.svelte @@ -0,0 +1,82 @@ + + +{#if snapshots.length > 0} +
+

+ Upstream snapshots + + Version of each upstream asset when this run was dispatched. Recorded for debugging only — + the run reads the latest data, not these versions. To inspect what this run saw, query the + asset with the copied AT (VERSION => n) clause. For partitioned + assets, the partition shown is the slice whose write produced that snapshot — the snapshot itself + covers the whole table. + +

+
+ + + + + + + + + + {#each snapshots as s (s.asset)} + + + + + + {/each} + +
AssetSnapshotTime travel
+ {s.asset} + + @ {s.snapshot_id} + {#if s.partition} + + · partition {s.partition} + {/if} + + +
+
+
+{/if} diff --git a/frontend/src/routes/(root)/(logged)/run/[...run]/+page.svelte b/frontend/src/routes/(root)/(logged)/run/[...run]/+page.svelte index 6ca81a5005..14acd3c89c 100644 --- a/frontend/src/routes/(root)/(logged)/run/[...run]/+page.svelte +++ b/frontend/src/routes/(root)/(logged)/run/[...run]/+page.svelte @@ -42,6 +42,7 @@ import DisplayResult from '$lib/components/DisplayResult.svelte' import DispatchEventsPanel from '$lib/components/runs/DispatchEventsPanel.svelte' + import UpstreamSnapshotsPanel from '$lib/components/runs/UpstreamSnapshotsPanel.svelte' import { enterpriseLicense, initialArgsStore, @@ -918,6 +919,7 @@ {#if job.id && job.workspace_id} + {/if} {/if}