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

all

Browse Source
This commit is contained in:
Ruben Fiszel
2026-05-11 14:49:13 +00:00
parent 99ab97215c
commit 8809b4ea83
19 changed files with 741 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/.sqlx/query-2a934656604281fd66aad72e08cf7f3268d5fc4144184ed8379ca0d3a638bc1c.json
Generated
+15
View File
@@ -0,0 +1,15 @@
{
"db_name": "PostgreSQL",
"query": "DELETE FROM schedule\n WHERE workspace_id = $1\n AND managed_by_runnable_path = $2",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Text",
"Text"
]
},
"nullable": []
},
"hash": "2a934656604281fd66aad72e08cf7f3268d5fc4144184ed8379ca0d3a638bc1c"
}
backend/.sqlx/query-3d66da60d4da1663bacce789d2b2bfc65f9786616e8f86570f7b91764e8b7c57.json
Generated
+23
View File
@@ -0,0 +1,23 @@
{
"db_name": "PostgreSQL",
"query": "\n SELECT runnable_path AS \"runnable_path!\"\n FROM script_trigger\n WHERE workspace_id = $1\n AND trigger_kind = 'asset'\n AND trigger_ref = $2\n AND runnable_kind = 'script'\n ",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "runnable_path!",
"type_info": "Varchar"
}
],
"parameters": {
"Left": [
"Text",
"Text"
]
},
"nullable": [
false
]
},
"hash": "3d66da60d4da1663bacce789d2b2bfc65f9786616e8f86570f7b91764e8b7c57"
}
backend/.sqlx/query-58e8e13acd9f7ff951f37d555beab84dcea21e0a38906de164889cf2dacf2e43.json
Generated
+23
View File
@@ -0,0 +1,23 @@
{
"db_name": "PostgreSQL",
"query": "SELECT args AS \"args!: Json<HashMap<String, Box<RawValue>>>\"\n FROM v2_job\n WHERE workspace_id = $1 AND id = $2",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "args!: Json<HashMap<String, Box<RawValue>>>",
"type_info": "Jsonb"
}
],
"parameters": {
"Left": [
"Text",
"Uuid"
]
},
"nullable": [
true
]
},
"hash": "58e8e13acd9f7ff951f37d555beab84dcea21e0a38906de164889cf2dacf2e43"
}
backend/.sqlx/query-7cee1d98c68d900d57ac5539fd10f054270fe998411ffb93c7372e2fab2acd4b.json
Generated
+47
View File
@@ -0,0 +1,47 @@
{
"db_name": "PostgreSQL",
"query": "INSERT INTO script_trigger\n (workspace_id, runnable_kind, runnable_path, trigger_kind, trigger_ref)\n VALUES ($1, $2, $3, $4, $5)",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Varchar",
{
"Custom": {
"name": "asset_usage_kind",
"kind": {
"Enum": [
"script",
"flow",
"job"
]
}
}
},
"Varchar",
{
"Custom": {
"name": "script_trigger_kind",
"kind": {
"Enum": [
"asset",
"schedule",
"webhook",
"email",
"kafka",
"mqtt",
"nats",
"postgres",
"sqs",
"gcp"
]
}
}
},
"Text"
]
},
"nullable": []
},
"hash": "7cee1d98c68d900d57ac5539fd10f054270fe998411ffb93c7372e2fab2acd4b"
}
backend/.sqlx/query-818188e13fd4da80070697027934503130d042aafe2229bf3e64b925d3e93c05.json
Generated
+15
View File
@@ -0,0 +1,15 @@
{
"db_name": "PostgreSQL",
"query": "DELETE FROM schedule\n WHERE workspace_id = $1\n AND managed_by_runnable_path = $2",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Text",
"Text"
]
},
"nullable": []
},
"hash": "818188e13fd4da80070697027934503130d042aafe2229bf3e64b925d3e93c05"
}
backend/.sqlx/query-8bce3f969b4bbbcf3dc1b8c671ab76e947e491da84fbaf5f130f58fbc851594e.json
Generated
+27
View File
@@ -0,0 +1,27 @@
{
"db_name": "PostgreSQL",
"query": "DELETE FROM script_trigger\n WHERE workspace_id = $1 AND runnable_kind = $2 AND runnable_path = $3",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Text",
{
"Custom": {
"name": "asset_usage_kind",
"kind": {
"Enum": [
"script",
"flow",
"job"
]
}
}
},
"Text"
]
},
"nullable": []
},
"hash": "8bce3f969b4bbbcf3dc1b8c671ab76e947e491da84fbaf5f130f58fbc851594e"
}
backend/.sqlx/query-a9a99880d870266f474878dd6ef541df988da527d30f663ef6f764f0c3d70d4b.json
Generated
+23
View File
@@ -0,0 +1,23 @@
{
"db_name": "PostgreSQL",
"query": "\n SELECT path AS \"path!\"\n FROM script\n WHERE workspace_id = $1\n AND auto_kind = 'pipeline'\n AND archived = false\n AND deleted = false\n AND ($2::text IS NULL OR path LIKE $2)\n ",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "path!",
"type_info": "Varchar"
}
],
"parameters": {
"Left": [
"Text",
"Text"
]
},
"nullable": [
false
]
},
"hash": "a9a99880d870266f474878dd6ef541df988da527d30f663ef6f764f0c3d70d4b"
}
backend/.sqlx/query-abb36bfddf707c7897b3936e982725f973eeb6ba5abbd61416060fb7d675593b.json
Generated
+28
View File
@@ -0,0 +1,28 @@
{
"db_name": "PostgreSQL",
"query": "\n SELECT\n substring(path from '^f/([^/]+)/') AS \"folder!\",\n COUNT(*) AS \"script_count!\"\n FROM script\n WHERE workspace_id = $1\n AND auto_kind = 'pipeline'\n AND archived = false\n AND deleted = false\n AND path LIKE 'f/%'\n GROUP BY substring(path from '^f/([^/]+)/')\n ORDER BY substring(path from '^f/([^/]+)/')\n ",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "folder!",
"type_info": "Text"
},
{
"ordinal": 1,
"name": "script_count!",
"type_info": "Int8"
}
],
"parameters": {
"Left": [
"Text"
]
},
"nullable": [
null,
null
]
},
"hash": "abb36bfddf707c7897b3936e982725f973eeb6ba5abbd61416060fb7d675593b"
}
backend/.sqlx/query-c0a3643369481c056214b61fb8d42f3e935111904a9d05f86c5b6d44704662df.json
Generated
+20
View File
@@ -0,0 +1,20 @@
{
"db_name": "PostgreSQL",
"query": "\n INSERT INTO schedule (\n workspace_id, path, schedule, timezone, edited_by, script_path,\n is_flow, enabled, email, permissioned_as,\n ws_error_handler_muted, no_flow_overlap, cron_version,\n managed_by_runnable_path\n )\n VALUES ($1, $2, $3, 'UTC', $4, $2, $5, true, $6, $7, false, false, 'v2', $2)\n ON CONFLICT (workspace_id, path) DO UPDATE\n SET schedule = EXCLUDED.schedule,\n edited_at = now(),\n edited_by = EXCLUDED.edited_by,\n managed_by_runnable_path = EXCLUDED.managed_by_runnable_path\n WHERE schedule.managed_by_runnable_path = EXCLUDED.managed_by_runnable_path\n OR schedule.managed_by_runnable_path IS NULL AND schedule.script_path = EXCLUDED.script_path\n ",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Varchar",
"Varchar",
"Varchar",
"Varchar",
"Bool",
"Varchar",
"Varchar"
]
},
"nullable": []
},
"hash": "c0a3643369481c056214b61fb8d42f3e935111904a9d05f86c5b6d44704662df"
}
backend/.sqlx/query-c17d1607191ccb033c48f9f620db980d2c02ed435b56b0ff0d11c49a63154e08.json
Generated
+52
View File
@@ -0,0 +1,52 @@
{
"db_name": "PostgreSQL",
"query": "\n SELECT\n runnable_kind AS \"runnable_kind!: AssetUsageKind\",\n runnable_path AS \"runnable_path!\",\n trigger_kind::text AS \"trigger_kind!\",\n trigger_ref AS \"trigger_ref!\"\n FROM script_trigger\n WHERE workspace_id = $1\n AND ($2::text IS NULL OR runnable_path LIKE $2)\n ",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "runnable_kind!: AssetUsageKind",
"type_info": {
"Custom": {
"name": "asset_usage_kind",
"kind": {
"Enum": [
"script",
"flow",
"job"
]
}
}
}
},
{
"ordinal": 1,
"name": "runnable_path!",
"type_info": "Varchar"
},
{
"ordinal": 2,
"name": "trigger_kind!",
"type_info": "Text"
},
{
"ordinal": 3,
"name": "trigger_ref!",
"type_info": "Text"
}
],
"parameters": {
"Left": [
"Text",
"Text"
]
},
"nullable": [
false,
false,
null,
false
]
},
"hash": "c17d1607191ccb033c48f9f620db980d2c02ed435b56b0ff0d11c49a63154e08"
}
backend/.sqlx/query-de06f44bad94710f14e9be4c0a6e6080e3c4faae5052500b93cc24b6fe556f2b.json
Generated
+43
View File
@@ -0,0 +1,43 @@
{
"db_name": "PostgreSQL",
"query": "\n SELECT\n kind AS \"kind!: AssetKind\",\n path AS \"path!\"\n FROM asset\n WHERE workspace_id = $1\n AND usage_kind = 'script'\n AND usage_path = $2\n AND usage_access_type IN ('w', 'rw')\n ",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "kind!: AssetKind",
"type_info": {
"Custom": {
"name": "asset_kind",
"kind": {
"Enum": [
"s3object",
"resource",
"variable",
"ducklake",
"datatable",
"volume"
]
}
}
}
},
{
"ordinal": 1,
"name": "path!",
"type_info": "Varchar"
}
],
"parameters": {
"Left": [
"Text",
"Text"
]
},
"nullable": [
false,
false
]
},
"hash": "de06f44bad94710f14e9be4c0a6e6080e3c4faae5052500b93cc24b6fe556f2b"
}
backend/migrations/20260510174213_asset_trigger_dispatch.down.sql
+3
View File
@@ -0,0 +1,3 @@
-- Postgres has no ALTER TYPE ... DROP VALUE for enums. The 'asset' value
-- stays even on rollback; this is consistent with how other job_trigger_kind
-- values were added (see 20250323162033_add-missing-trigger-kind-...).
backend/migrations/20260510174213_asset_trigger_dispatch.up.sql
+5
View File
@@ -0,0 +1,5 @@
-- Add 'asset' as a job_trigger_kind so jobs that get dispatched as a
-- consequence of an upstream pipeline script writing an asset can be
-- attributed via v2_job.trigger_kind = 'asset'. The producer's runnable
-- path goes into v2_job.trigger.
ALTER TYPE job_trigger_kind ADD VALUE IF NOT EXISTS 'asset';
backend/windmill-api/openapi.yaml
+1
View File
@@ -23677,6 +23677,7 @@ components:
- azure
- google
- github
- asset
TriggerMode:
description: job trigger mode
backend/windmill-queue/src/asset_dispatch.rs
+400
View File
@@ -0,0 +1,400 @@
/*
* Author: Ruben Fiszel
* Copyright: Windmill Labs, Inc 2022
* This file and its contents are licensed under the AGPLv3 License.
* Please see the included NOTICE for copyright information and
* LICENSE-AGPL for a copy of the license.
*/
//! Runtime fan-out for asset-triggered scripts.
//!
//! When a producer pipeline script (`// pipeline`) writes an asset and a
//! downstream script subscribes to that asset via `// on s3://...`, this
//! module pushes a job for each subscriber after the producer's job
//! completes successfully.
//!
//! Eligibility (V1, narrow on purpose):
//! - Producer kind is `Script` or `Preview`. Flows defer.
//! - Producer is top-level (no `parent_job`, no `flow_step_id`).
//! - Producer succeeded.
//! - The producer's args do not contain `_wmill_skip_asset_dispatch: true`.
//! - Cascade depth (carried in args under `trigger.depth`) is below
//! `MAX_CHAIN_DEPTH`.
//!
//! Subscribers (V1):
//! - Only `script` runnables. Flow subscribers defer.
//! - The subscriber must have at least one non-archived script row.
//! - A subscriber is skipped if its path equals the producer's path
//! (avoids trivial self-loops).
//!
//! Args sent to subscribers:
//! ```json
//! {
//! "trigger": {
//! "kind": "asset",
//! "asset_kind": "s3object",
//! "asset_path": "...",
//! "producer_path": "...",
//! "producer_job_id": "...",
//! "depth": 1
//! }
//! }
//! ```
//!
//! Errors are logged but never bubble up to fail the producer's job.
use crate::{push, MiniCompletedJob, PushArgs, PushIsolationLevel};
use serde_json::value::RawValue;
use sqlx::types::Json;
use sqlx::{Pool, Postgres};
use std::collections::HashMap;
use uuid::Uuid;
use windmill_common::assets::{parse_asset_trigger_ref, AssetKind};
use windmill_common::error::{self, Result};
use windmill_common::get_latest_hash_for_path;
use windmill_common::jobs::{JobKind, JobPayload, JobTriggerKind};
use windmill_common::runnable_settings::{ConcurrencySettings, DebouncingSettings};
use windmill_common::triggers::TriggerMetadata;
use windmill_common::users::username_to_permissioned_as;
use windmill_common::worker::to_raw_value;
use windmill_common::DB;
/// Reserved arg key that suppresses asset-trigger dispatch for a single run.
/// Set by the test panel when the user opts out of the cascade.
pub const SKIP_ASSET_DISPATCH_ARG: &str = "_wmill_skip_asset_dispatch";
/// Reserved arg key (under `trigger.depth`) that carries cascade depth.
const CHAIN_DEPTH_KEY: &str = "depth";
/// Cap cascade depth so a misconfigured ring doesn't fan out unbounded.
const MAX_CHAIN_DEPTH: i64 = 5;
/// Returned to the caller (the worker's completed-job hook) so logs can
/// reference the dispatched ids.
#[derive(Debug, Default)]
pub struct DispatchResult {
pub dispatched: Vec<Uuid>,
}
/// Top-level entry. Returns `Ok(default)` and logs on any internal failure
/// rather than propagating, because dispatch is best-effort and must not
/// retroactively fail the producer.
pub async fn dispatch_asset_triggers(db: &DB, job: &MiniCompletedJob) -> DispatchResult {
match try_dispatch(db, job).await {
Ok(r) => r,
Err(e) => {
tracing::error!("asset-trigger dispatch failed for job {}: {e:#}", job.id);
DispatchResult::default()
}
}
}
async fn try_dispatch(db: &DB, job: &MiniCompletedJob) -> Result<DispatchResult> {
if !is_eligible_kind(job) {
return Ok(DispatchResult::default());
}
let runnable_path = match job.runnable_path.as_deref() {
Some(p) if !p.is_empty() => p,
_ => return Ok(DispatchResult::default()),
};
// Args were moved from v2_job to v2_job_completed by add_completed_job
// before dispatch runs. Fetch from v2_job_completed.
let args = fetch_args(db, &job.workspace_id, job.id).await?;
if read_skip_arg(args.as_ref()) {
return Ok(DispatchResult::default());
}
let depth = read_chain_depth(args.as_ref());
if depth >= MAX_CHAIN_DEPTH {
tracing::warn!(
"asset-trigger dispatch skipped: chain depth {} >= cap {} (job {}, path {})",
depth,
MAX_CHAIN_DEPTH,
job.id,
runnable_path
);
return Ok(DispatchResult::default());
}
let writes = fetch_producer_writes(db, &job.workspace_id, runnable_path).await?;
if writes.is_empty() {
return Ok(DispatchResult::default());
}
let mut dispatched = Vec::new();
for (asset_kind, asset_path) in writes {
let Some(prefix) = prefix_for(asset_kind) else {
continue;
};
let trigger_ref = format!("{}{}", prefix, asset_path);
let subs = fetch_subscribers(db, &job.workspace_id, &trigger_ref).await?;
for sub_path in subs {
if sub_path == runnable_path {
continue;
}
match push_subscriber(
db,
job,
&sub_path,
asset_kind,
&asset_path,
runnable_path,
depth + 1,
)
.await
{
Ok(id) => dispatched.push(id),
Err(e) => {
tracing::error!("failed to push asset-triggered job for {}: {e:#}", sub_path)
}
}
}
}
if !dispatched.is_empty() {
tracing::info!(
"asset-trigger dispatch from job {} ({}): pushed {} downstream jobs",
job.id,
runnable_path,
dispatched.len()
);
}
Ok(DispatchResult { dispatched })
}
fn is_eligible_kind(job: &MiniCompletedJob) -> bool {
if !matches!(job.kind, JobKind::Script | JobKind::Preview) {
return false;
}
if job.parent_job.is_some() || job.flow_step_id.is_some() {
return false;
}
true
}
async fn fetch_args(
db: &Pool<Postgres>,
workspace_id: &str,
job_id: Uuid,
) -> Result<Option<HashMap<String, Box<RawValue>>>> {
// Read from v2_job because args live there permanently — v2_job_completed
// is the *result* row and doesn't carry args. The producer's v2_job row
// is still present at dispatch time (deletion happens later in the
// completion pipeline, after this hook).
let row = sqlx::query!(
r#"SELECT args AS "args!: Json<HashMap<String, Box<RawValue>>>"
FROM v2_job
WHERE workspace_id = $1 AND id = $2"#,
workspace_id,
job_id,
)
.fetch_optional(db)
.await?;
Ok(row.map(|r| r.args.0))
}
fn read_skip_arg(args: Option<&HashMap<String, Box<RawValue>>>) -> bool {
args.and_then(|a| a.get(SKIP_ASSET_DISPATCH_ARG))
.and_then(|v| serde_json::from_str::<bool>(v.get()).ok())
.unwrap_or(false)
}
fn read_chain_depth(args: Option<&HashMap<String, Box<RawValue>>>) -> i64 {
let Some(args) = args else {
return 0;
};
let Some(trigger) = args.get("trigger") else {
return 0;
};
let Ok(map) = serde_json::from_str::<HashMap<String, Box<RawValue>>>(trigger.get()) else {
return 0;
};
map.get(CHAIN_DEPTH_KEY)
.and_then(|v| serde_json::from_str::<i64>(v.get()).ok())
.unwrap_or(0)
}
fn prefix_for(kind: AssetKind) -> Option<&'static str> {
match kind {
AssetKind::S3Object => Some("s3://"),
AssetKind::Resource => Some("$res:"),
AssetKind::Ducklake => Some("ducklake://"),
AssetKind::DataTable => Some("datatable://"),
AssetKind::Volume => Some("volume://"),
// Deprecated kind from before the parser was unified — has no
// canonical trigger ref and never produced trigger rows.
AssetKind::Variable => None,
}
}
async fn fetch_producer_writes(
db: &Pool<Postgres>,
workspace_id: &str,
runnable_path: &str,
) -> Result<Vec<(AssetKind, String)>> {
let rows = sqlx::query!(
r#"
SELECT
kind AS "kind!: AssetKind",
path AS "path!"
FROM asset
WHERE workspace_id = $1
AND usage_kind = 'script'
AND usage_path = $2
AND usage_access_type IN ('w', 'rw')
"#,
workspace_id,
runnable_path,
)
.fetch_all(db)
.await?;
Ok(rows.into_iter().map(|r| (r.kind, r.path)).collect())
}
async fn fetch_subscribers(
db: &Pool<Postgres>,
workspace_id: &str,
trigger_ref: &str,
) -> Result<Vec<String>> {
// V1: script subscribers only. Flow subscribers (`runnable_kind = 'flow'`)
// are intentionally excluded — wiring them is straightforward but the
// payload shape and permissioning need their own pass.
let rows = sqlx::query!(
r#"
SELECT runnable_path AS "runnable_path!"
FROM script_trigger
WHERE workspace_id = $1
AND trigger_kind = 'asset'
AND trigger_ref = $2
AND runnable_kind = 'script'
"#,
workspace_id,
trigger_ref,
)
.fetch_all(db)
.await?;
// Sanity check: parsing the stored trigger_ref must succeed. If it
// doesn't, the row is corrupt; skip it loudly rather than silently.
if parse_asset_trigger_ref(trigger_ref).is_none() {
tracing::warn!(
"asset-trigger dispatch: trigger_ref {} did not round-trip through parse_asset_trigger_ref",
trigger_ref
);
}
Ok(rows.into_iter().map(|r| r.runnable_path).collect())
}
async fn push_subscriber(
db: &DB,
producer: &MiniCompletedJob,
subscriber_path: &str,
asset_kind: AssetKind,
asset_path: &str,
producer_path: &str,
depth: i64,
) -> Result<Uuid> {
let (
hash,
tag,
_concurrency_key,
_concurrent_limit,
_concurrency_time_window_s,
_debounce_key,
_debounce_delay_s,
cache_ttl,
cache_ignore_s3_path,
language,
dedicated_worker,
priority,
_timeout,
on_behalf_of_email,
created_by,
_runnable_settings_handle,
labels,
) = get_latest_hash_for_path(db, &producer.workspace_id, subscriber_path, false).await?;
let payload = JobPayload::ScriptHash {
hash,
path: subscriber_path.to_string(),
cache_ttl,
cache_ignore_s3_path,
dedicated_worker,
language,
priority,
apply_preprocessor: false,
// V1: skip debouncing/concurrency for asset-triggered runs. The
// trigger fan-out is the user's intent — we don't want a noisy
// upstream's writes to silently drop downstream runs because a
// debounce key collides. Revisit if we see lots of dups.
debouncing_settings: DebouncingSettings::default(),
concurrency_settings: ConcurrencySettings::default(),
labels,
};
// Subscriber's own on_behalf_of_email controls identity when set;
// otherwise we run as the producer. This keeps the asset cascade
// attributable to whoever originally wrote the asset, while still
// honoring scripts that explicitly opted into a service-account email.
let (permissioned_as, email) = if let Some(obo) = on_behalf_of_email {
(username_to_permissioned_as(&created_by), obo)
} else {
(
producer.permissioned_as.clone(),
producer.permissioned_as_email.clone(),
)
};
let mut args: HashMap<String, Box<RawValue>> = HashMap::new();
let trigger_payload = serde_json::json!({
"kind": "asset",
"asset_kind": serde_json::to_value(&asset_kind).unwrap_or(serde_json::Value::Null),
"asset_path": asset_path,
"producer_path": producer_path,
"producer_job_id": producer.id.to_string(),
CHAIN_DEPTH_KEY: depth,
});
args.insert("trigger".to_string(), to_raw_value(&trigger_payload));
// Attribute the dispatched run to a synthetic user so audit logs reflect
// it came from the asset cascade, not the original human runner.
let pseudo_user = format!("asset-{producer_path}");
let tx = PushIsolationLevel::IsolatedRoot(db.clone());
let (id, tx) = push(
db,
tx,
&producer.workspace_id,
payload,
PushArgs { args: &args, extra: None },
&pseudo_user,
&email,
permissioned_as,
Some(producer_path),
None,
Some(producer_path.to_string()),
None,
None,
None,
None,
false,
false,
None,
true,
tag,
None,
None,
None,
None,
false,
None,
Some(TriggerMetadata::new(
Some(producer_path.to_string()),
JobTriggerKind::Asset,
)),
None,
)
.await
.map_err(|e| error::Error::internal_err(format!("push asset-triggered job: {e:#}")))?;
tx.commit().await?;
Ok(id)
}
backend/windmill-queue/src/lib.rs
+1
View File
@@ -6,6 +6,7 @@
* LICENSE-AGPL for a copy of the license.
*/
pub mod asset_dispatch;
pub mod jobs;
#[cfg(feature = "private")]
pub mod jobs_ee;
backend/windmill-types/src/jobs.rs
+4
View File
@@ -43,6 +43,9 @@ pub enum JobTriggerKind {
#[serde(rename = "ci_test")]
#[sqlx(rename = "ci_test")]
CiTest,
// A run dispatched because an upstream pipeline script wrote an asset
// this runnable subscribes to via `// on s3://...` annotations.
Asset,
}
impl std::fmt::Display for JobTriggerKind {
@@ -64,6 +67,7 @@ impl std::fmt::Display for JobTriggerKind {
JobTriggerKind::Google => "google",
JobTriggerKind::Github => "github",
JobTriggerKind::CiTest => "ci_test",
JobTriggerKind::Asset => "asset",
};
write!(f, "{}", kind)
}
backend/windmill-worker/src/result_processor.rs
+9 -3
View File
@@ -35,9 +35,9 @@ use windmill_common::{
use windmill_common::bench::{BenchmarkInfo, BenchmarkIter};
use windmill_queue::{
append_logs, get_mini_completed_job, is_pre_shaped_wm_failure_result, CanceledBy, FlowRunners,
JobCompleted, MiniCompletedJob, MiniPulledJob, ValidableJson, WrappedError, INIT_SCRIPT_TAG,
MANUAL_FAILURE_ERROR_NAME,
append_logs, asset_dispatch, get_mini_completed_job, is_pre_shaped_wm_failure_result,
CanceledBy, FlowRunners, JobCompleted, MiniCompletedJob, MiniPulledJob, ValidableJson,
WrappedError, INIT_SCRIPT_TAG, MANUAL_FAILURE_ERROR_NAME,
};
use serde_json::{json, value::RawValue, Value};
@@ -789,6 +789,12 @@ pub async fn process_completed_job(
from_cache.unwrap_or(false),
)
.await?;
// Asset-trigger fan-out: best-effort, never propagates errors.
// Internal eligibility checks gate to top-level Script/Preview runs;
// see windmill_queue::asset_dispatch.
asset_dispatch::dispatch_asset_triggers(db, &job).await;
drop(job);
add_time!(bench, "add_completed_job END");
frontend/src/lib/components/triggers/utils.ts
+2 -1
View File
@@ -81,7 +81,8 @@ export const jobTriggerKinds: JobTriggerKind[] = [
'gcp',
'azure',
'google',
'github'
'github',
'asset'
]
export type Trigger = {