starred/windmill
1
0
Fork 0
mirror of https://github.com/windmill-labs/windmill.git synced 2026-08-18 16:02:10 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(pipelines): record upstream snapshot ids on cascade-dispatched jobs (#9910)

Browse Source 
* feat(pipelines): record upstream snapshot ids on cascade-dispatched jobs

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

* refactor: batch upstream-snapshot lookup and memoize per subscriber

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

---------

Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Ruben Fiszel
2026-07-03 22:10:18 +00:00
committed by GitHub
parent df6e511763
commit af36498432
7 changed files with 452 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/.sqlx/query-0ee4a4dc8c6f09e7027bd99f98d8d1d346f9d67056ef272314dd9b80bee9f285.json
Generated
+77
View File
@@ -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"
}
backend/.sqlx/query-3703847d0cef9df3a480af8b6a5c21b6af282c68f06ce037a11f703284d8980d.json
Generated
+23
View File
@@ -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"
}
backend/tests/asset_trigger_dispatch.rs
+129
View File
@@ -1044,3 +1044,132 @@ async fn reaper_clears_only_stale_join_slots(db: Pool<Postgres>) -> 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<Postgres>,
kind: &str,
asset_path: &str,
partition: &str,
status: &str,
snapshot_id: Option<i64>,
) -> 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<Postgres>) -> 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(())
}
backend/windmill-queue/src/asset_dispatch.rs
+131 -3
View File
@@ -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<DispatchResult>
// subscriber × asset write). The mid-pass join-slot writes
// (record_and_check_join_slot) are a separate table and unaffected.
let mut events: Vec<EventRow> = 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<String, Arc<Vec<UpstreamSnapshot>>> = 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<DispatchResult>
}
}
}
// 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<DispatchResult>
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<String>,
}
/// 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<Postgres>,
workspace_id: &str,
subscriber_path: &str,
) -> Result<Vec<UpstreamSnapshot>> {
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<AssetKind> = parsed.iter().map(|(_, k, _)| *k).collect();
let paths: Vec<String> = 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<AssetKind>,
&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<i32>,
retry_count: Option<i16>,
retry_delay_s: Option<i32>,
upstream_snapshots: &[UpstreamSnapshot],
) -> Result<Uuid> {
// 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<String, Box<RawValue>> = 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
docs/ducklake-materialization.md
+8
View File
@@ -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"
frontend/src/lib/components/runs/UpstreamSnapshotsPanel.svelte
+82
View File
@@ -0,0 +1,82 @@
<script lang="ts">
import { copyToClipboard } from '$lib/utils'
import { Button } from '$lib/components/common'
import Tooltip from '$lib/components/Tooltip.svelte'
import { ClipboardCopy } from 'lucide-svelte'
// Shape produced by asset-dispatch (backend asset_dispatch.rs
// `UpstreamSnapshot`): the substrate version each direct upstream asset was
// at when this job was dispatched. Forensic record only — the run's reads
// are not pinned to these versions.
type UpstreamSnapshot = { asset: string; snapshot_id: number; partition?: string }
type Props = { args: any }
let { args }: Props = $props()
const snapshots: UpstreamSnapshot[] = $derived.by(() => {
const trigger = args?.trigger
if (trigger?.kind !== 'asset' || !Array.isArray(trigger?.upstream_snapshots)) return []
// args are caller-supplied JSON, so dedupe by asset — a crafted
// duplicate would otherwise crash the keyed each below.
const seen = new Set<string>()
return trigger.upstream_snapshots.filter((s: any) => {
if (typeof s?.asset !== 'string' || typeof s?.snapshot_id !== 'number') return false
if (seen.has(s.asset)) return false
seen.add(s.asset)
return true
})
})
</script>
{#if snapshots.length > 0}
<div class="mr-2 sm:mr-0 mt-12 mb-6">
<h3 class="text-xs font-semibold text-emphasis mb-1">
Upstream snapshots
<Tooltip>
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 <span class="font-mono">AT (VERSION =&gt; n)</span> clause. For partitioned
assets, the partition shown is the slice whose write produced that snapshot — the snapshot itself
covers the whole table.
</Tooltip>
</h3>
<div class="border rounded-md overflow-hidden">
<table class="w-full text-xs">
<thead class="bg-surface-secondary text-secondary">
<tr>
<th class="text-left font-normal px-3 py-2">Asset</th>
<th class="text-left font-normal px-3 py-2">Snapshot</th>
<th class="text-right font-normal px-3 py-2">Time travel</th>
</tr>
</thead>
<tbody>
{#each snapshots as s (s.asset)}
<tr class="border-t">
<td class="px-3 py-2 align-top">
<span class="font-mono">{s.asset}</span>
</td>
<td class="px-3 py-2 align-top whitespace-nowrap">
<span class="font-mono">@ {s.snapshot_id}</span>
{#if s.partition}
<span class="text-secondary">
· partition <span class="font-mono">{s.partition}</span></span
>
{/if}
</td>
<td class="px-3 py-2 align-top text-right whitespace-nowrap">
<Button
variant="subtle"
unifiedSize="sm"
startIcon={{ icon: ClipboardCopy }}
onclick={() => copyToClipboard(`AT (VERSION => ${s.snapshot_id})`)}
>
AT (VERSION =&gt; {s.snapshot_id})
</Button>
</td>
</tr>
{/each}
</tbody>
</table>
</div>
</div>
{/if}
frontend/src/routes/(root)/(logged)/run/[...run]/+page.svelte
+2
View File
@@ -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 @@
</div>
</div>
{#if job.id && job.workspace_id}
<UpstreamSnapshotsPanel args={job.args} />
<DispatchEventsPanel workspace={job.workspace_id} jobId={job.id} />
{/if}
{/if}