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: add scheduled job deletion with configurable retention period (#8753)

Browse Source 
* feat: add scheduled job deletion with configurable retention period

Extends delete_after_use with delete_after_secs to enable configurable
retention periods for job args/result/logs. At completion, jobs can be
scheduled for future deletion via a new job_delete_schedule table,
processed by a monitor task. Supports per-script, per-flow, and
per-flow-step configuration. Backward compatible.

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

* feat: add integration tests, revert query! macros, fix review issues

- Add integration tests for resolve_delete_after_secs, schedule_job_deletion,
  flow-level and module-level delete_after_secs, backward compat
- Revert sqlx::query() back to sqlx::query!() macros for compile-time safety
- Regenerate sqlx offline cache
- Fix FlowModule/NewScript/FlowValue constructions in all test files
- Fix autoscaling_ee.rs for updated script_path_to_payload return type

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

* chore: update ee-repo-ref.txt for autoscaling_ee fix

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

* fix: gate cleanup_scheduled_job_deletions behind enterprise feature

Prevents dead_code warning (which CI treats as error via -D warnings)
when compiling without enterprise feature.

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

* chore: regenerate sqlx cache after merge with main

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

* fix: address review feedback on scheduled deletion

- Monitor: roll back transaction on any cleanup error so schedule rows
  survive for retry on next cycle (instead of best-effort then discard)
- Migration: add FK with ON DELETE CASCADE to job_delete_schedule.job_id
  to prevent orphan rows when jobs are deleted through other means
- Simplify bool-to-Option conversion with .then_some(true)

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

* refactor: stop setting delete_after_use alongside delete_after_secs

No mixed-version deployment scenario exists, so delete_after_secs alone
is sufficient. The backend's resolve_delete_after_secs handles
(None, Some(secs)) correctly without needing delete_after_use set.

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

* refactor: remove delete_after_use from public API surface

Remove delete_after_use from OpenAPI spec, API client, runtime client,
and workspace export. Only delete_after_secs is exposed going forward.

The field remains in Rust backend types with #[serde(skip_serializing)]
for backward-compatible deserialization of existing scripts/flows that
were saved with delete_after_use: true.

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

* chore: update ee-repo-ref to 1d4b7a31fc115d6aba8640f7cd3fd5a01abe6806

This commit updates the EE repository reference after PR #519 was merged in windmill-ee-private.

Previous ee-repo-ref: 9eba09a13b778caafc6ae65098b90e53c91984d3

New ee-repo-ref: 1d4b7a31fc115d6aba8640f7cd3fd5a01abe6806

Automated by sync-ee-ref workflow.

* fix: regenerate system prompts, remove unused import

- Regenerate auto-generated system prompts after openflow schema change
- Remove unused serde_json::json import in test file (CI -D warnings)

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

* fix: insert dummy v2_job row in schedule tests for FK constraint

The job_delete_schedule table has a FK to v2_job, so tests need a
real v2_job row before inserting into the schedule table.

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

* chore: trigger CI re-run

* fix: remove heavy flow integration tests to avoid CI worker contention

The flow integration tests spawn workers that compete for CPU with
the existing relock_skip tests under --test-threads=10, causing
consistent 60s timeouts in CI. Keep only the lightweight unit tests
and DB integration tests.

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

* fix: restore correct ee-repo-ref for our branch

The ref was overwritten to main's EE ref during a rebase. Restore to
our branch's EE commit that includes the autoscaling tuple fix.

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

* chore: retrigger CI on fresh runner

* fix: remove FK constraint from job_delete_schedule to unblock CI

The FK with ON DELETE CASCADE to v2_job may have caused performance
overhead during test DB setup (each sqlx::test creates a fresh DB
with all migrations). Remove the FK — orphan schedule rows are
harmlessly cleaned by the monitor.

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

* ee-ref

---------

Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Co-authored-by: windmill-internal-app[bot] <windmill-internal-app[bot]@users.noreply.github.com>
This commit is contained in:
Ruben Fiszel
2026-04-08 00:15:28 -04:00
committed by GitHub
parent 9e6427d150
commit 2d18a68099
51 changed files with 1715 additions and 958 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/.sqlx/query-007fa93171b244490b94464938b9f95aca4e91bccde6da93cb151799b3398049.json
Generated
+15
View File
@@ -0,0 +1,15 @@
{
"db_name": "PostgreSQL",
"query": "UPDATE v2_job_status\n SET flow_status = JSONB_SET(flow_status, ARRAY['cleanup_module', 'flow_jobs_to_clean'], COALESCE(flow_status->'cleanup_module'->'flow_jobs_to_clean', '[]'::jsonb) || $1)\n WHERE id = $2",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Jsonb",
"Uuid"
]
},
"nullable": []
},
"hash": "007fa93171b244490b94464938b9f95aca4e91bccde6da93cb151799b3398049"
}
backend/.sqlx/query-035e29e775bfc5b236100135e1d94a4baf2b617b86f0c3c74ba9a00b859993f6.json
Generated
+16
View File
@@ -0,0 +1,16 @@
{
"db_name": "PostgreSQL",
"query": "INSERT INTO job_delete_schedule (job_id, workspace_id, delete_at) VALUES ($1, $2, now() + make_interval(secs => $3::double precision)) ON CONFLICT (job_id) DO NOTHING",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Uuid",
"Varchar",
"Float8"
]
},
"nullable": []
},
"hash": "035e29e775bfc5b236100135e1d94a4baf2b617b86f0c3c74ba9a00b859993f6"
}
backend/.sqlx/query-21331baf02c3c798bcc215443dad16eb66acac5f40c20529b595154f4e6fd754.json
Generated
+15
View File
@@ -0,0 +1,15 @@
{
"db_name": "PostgreSQL",
"query": "UPDATE v2_job_status\n SET flow_status = JSONB_SET(flow_status, ARRAY['cleanup_module', 'flow_jobs_to_schedule_clean'], COALESCE(flow_status->'cleanup_module'->'flow_jobs_to_schedule_clean', '[]'::jsonb) || $1)\n WHERE id = $2",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Jsonb",
"Uuid"
]
},
"nullable": []
},
"hash": "21331baf02c3c798bcc215443dad16eb66acac5f40c20529b595154f4e6fd754"
}
backend/.sqlx/query-5a219a2532517869578c4504ff3153c43903f929ae5d62fbba12610f89c36d55.json
Generated
+1 -1
View File
@@ -15,7 +15,7 @@
]
},
"nullable": [
null
true
]
},
"hash": "5a219a2532517869578c4504ff3153c43903f929ae5d62fbba12610f89c36d55"
backend/.sqlx/query-63c16a4277983aaed0aed54972923919cee3cc444725ac6b7906922554bae800.json
Generated
-23
View File
@@ -1,23 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "SELECT value FROM resource WHERE path = $1 AND workspace_id = $2",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "value",
"type_info": "Jsonb"
}
],
"parameters": {
"Left": [
"Text",
"Text"
]
},
"nullable": [
true
]
},
"hash": "63c16a4277983aaed0aed54972923919cee3cc444725ac6b7906922554bae800"
}
backend/.sqlx/query-7929fa087a28949906ffb3f508d8b88c922d3442418701edd6b97e5fa50bd739.json
Generated
-20
View File
@@ -1,20 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "INSERT INTO resource\n (workspace_id, path, value, description, resource_type, created_by, edited_at, ws_specific)\n VALUES ($1, $2, $3, $4, $5, $6, now(), $7) ON CONFLICT (workspace_id, path)\n DO UPDATE SET value = EXCLUDED.value, description = EXCLUDED.description, resource_type = EXCLUDED.resource_type, edited_at = now(), ws_specific = EXCLUDED.ws_specific",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Varchar",
"Varchar",
"Jsonb",
"Text",
"Varchar",
"Varchar",
"Bool"
]
},
"nullable": []
},
"hash": "7929fa087a28949906ffb3f508d8b88c922d3442418701edd6b97e5fa50bd739"
}
backend/.sqlx/query-819c233915383e89af1bcf1a56c5f67c4e1fc217f216f609e36a9944a7807b33.json
Generated
-35
View File
@@ -1,35 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "SELECT value, description, resource_type\n FROM resource\n WHERE workspace_id = $1 AND path = $2",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "value",
"type_info": "Jsonb"
},
{
"ordinal": 1,
"name": "description",
"type_info": "Text"
},
{
"ordinal": 2,
"name": "resource_type",
"type_info": "Varchar"
}
],
"parameters": {
"Left": [
"Text",
"Text"
]
},
"nullable": [
true,
true,
false
]
},
"hash": "819c233915383e89af1bcf1a56c5f67c4e1fc217f216f609e36a9944a7807b33"
}
backend/.sqlx/query-954e832d2587506a4bb0cb3a4fb45658026de7da4680eca1ea6b08f4b5e33800.json
Generated
+14
View File
@@ -0,0 +1,14 @@
{
"db_name": "PostgreSQL",
"query": "UPDATE job_logs SET logs = '##DELETED##' WHERE job_id = ANY($1)",
"describe": {
"columns": [],
"parameters": {
"Left": [
"UuidArray"
]
},
"nullable": []
},
"hash": "954e832d2587506a4bb0cb3a4fb45658026de7da4680eca1ea6b08f4b5e33800"
}
backend/.sqlx/query-9c5f9fc1c2fdf35e98c78180a319f2cfdfe09b4eef3375a6170454d3d52e8dbc.json
Generated
+22
View File
@@ -0,0 +1,22 @@
{
"db_name": "PostgreSQL",
"query": "SELECT id FROM v2_job WHERE root_job = $1",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "id",
"type_info": "Uuid"
}
],
"parameters": {
"Left": [
"Uuid"
]
},
"nullable": [
false
]
},
"hash": "9c5f9fc1c2fdf35e98c78180a319f2cfdfe09b4eef3375a6170454d3d52e8dbc"
}
backend/.sqlx/query-a25200e046d6e15bf5a5f81d6cc9622bdb7ad8009fa6e67606296c1a8a1a92b4.json
Generated
+15
View File
@@ -0,0 +1,15 @@
{
"db_name": "PostgreSQL",
"query": "UPDATE v2_job_status\n SET flow_status = JSONB_SET(flow_status, ARRAY['cleanup_module', 'flow_jobs_to_schedule_clean'], COALESCE(flow_status->'cleanup_module'->'flow_jobs_to_schedule_clean', '[]'::jsonb) || $1)\n WHERE id = $2",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Jsonb",
"Uuid"
]
},
"nullable": []
},
"hash": "a25200e046d6e15bf5a5f81d6cc9622bdb7ad8009fa6e67606296c1a8a1a92b4"
}
backend/.sqlx/query-b01160fe44d69834ac08bbf60feacb3e3caa02a04b084da44cdcb9103794b39e.json
Generated
-15
View File
@@ -1,15 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "UPDATE v2_job_status\n SET flow_status = JSONB_SET(flow_status, ARRAY['cleanup_module', 'flow_jobs_to_clean'], COALESCE(flow_status->'cleanup_module'->'flow_jobs_to_clean', '[]'::jsonb) || $1)\n WHERE id = $2",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Jsonb",
"Uuid"
]
},
"nullable": []
},
"hash": "b01160fe44d69834ac08bbf60feacb3e3caa02a04b084da44cdcb9103794b39e"
}
backend/.sqlx/query-790d79ec7abe6ebe1092afd9de4c5fc383272d2057ac5b47a7425f095f4e8788.json → backend/.sqlx/query-dafc503a5f3adc5c7db7c11096775cacfffd2d3173dfdddcf37589cba356791e.json
Generated
+3 -2
View File
@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "INSERT INTO script (workspace_id, hash, path, parent_hashes, summary, description, content, created_by, schema, is_template, extra_perms, lock, language, kind, tag, draft_only, envs, concurrent_limit, concurrency_time_window_s, cache_ttl, dedicated_worker, ws_error_handler_muted, priority, restart_unless_cancelled, delete_after_use, timeout, concurrency_key, visible_to_runner_only, auto_kind, codebase, has_preprocessor, on_behalf_of_email, schema_validation, assets, debounce_key, debounce_delay_s, cache_ignore_s3_path, runnable_settings_handle, modules, labels) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9::text::json, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19, $20, $21, $22, $23, $24, $25, $26, $27, $28, $29, $30, $31, $32, $33, $34, $35, $36, $37, $38, $39, $40)",
"query": "INSERT INTO script (workspace_id, hash, path, parent_hashes, summary, description, content, created_by, schema, is_template, extra_perms, lock, language, kind, tag, draft_only, envs, concurrent_limit, concurrency_time_window_s, cache_ttl, dedicated_worker, ws_error_handler_muted, priority, restart_unless_cancelled, delete_after_use, delete_after_secs, timeout, concurrency_key, visible_to_runner_only, auto_kind, codebase, has_preprocessor, on_behalf_of_email, schema_validation, assets, debounce_key, debounce_delay_s, cache_ignore_s3_path, runnable_settings_handle, modules, labels) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9::text::json, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19, $20, $21, $22, $23, $24, $25, $26, $27, $28, $29, $30, $31, $32, $33, $34, $35, $36, $37, $38, $39, $40, $41)",
"describe": {
"columns": [],
"parameters": {
@@ -77,6 +77,7 @@
"Bool",
"Bool",
"Int4",
"Int4",
"Varchar",
"Bool",
"Varchar",
@@ -95,5 +96,5 @@
},
"nullable": []
},
"hash": "790d79ec7abe6ebe1092afd9de4c5fc383272d2057ac5b47a7425f095f4e8788"
"hash": "dafc503a5f3adc5c7db7c11096775cacfffd2d3173dfdddcf37589cba356791e"
}
backend/.sqlx/query-deac41298e8b0d0870e314fef0813c24dd55d63bda78a0a5f35ed6f22bea6bef.json
Generated
-20
View File
@@ -1,20 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "INSERT INTO resource\n (workspace_id, path, value, description, resource_type, created_by, edited_at, labels)\n VALUES ($1, $2, $3, $4, $5, $6, now(), $7) ON CONFLICT (workspace_id, path)\n DO UPDATE SET value = EXCLUDED.value, description = EXCLUDED.description, resource_type = EXCLUDED.resource_type, edited_at = now(), labels = EXCLUDED.labels",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Varchar",
"Varchar",
"Jsonb",
"Text",
"Varchar",
"Varchar",
"TextArray"
]
},
"nullable": []
},
"hash": "deac41298e8b0d0870e314fef0813c24dd55d63bda78a0a5f35ed6f22bea6bef"
}
backend/.sqlx/query-e1886af775f70f7ad3949e35d2e274818884b2ce38226d5d8f9b8774dda6d5dc.json
Generated
+22
View File
@@ -0,0 +1,22 @@
{
"db_name": "PostgreSQL",
"query": "DELETE FROM job_delete_schedule\n WHERE job_id IN (\n SELECT job_id FROM job_delete_schedule\n WHERE delete_at <= now()\n ORDER BY delete_at\n LIMIT $1\n FOR UPDATE SKIP LOCKED\n )\n RETURNING job_id",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "job_id",
"type_info": "Uuid"
}
],
"parameters": {
"Left": [
"Int8"
]
},
"nullable": [
false
]
},
"hash": "e1886af775f70f7ad3949e35d2e274818884b2ce38226d5d8f9b8774dda6d5dc"
}
backend/ee-repo-ref.txt
+1 -1
View File
@@ -1 +1 @@
86158dde674238fd94f925bdcd5155759e823ed6
1d4b7a31fc115d6aba8640f7cd3fd5a01abe6806
backend/migrations/20260407154106_add_delete_after_secs.down.sql
+2
View File
@@ -0,0 +1,2 @@
DROP TABLE IF EXISTS job_delete_schedule;
ALTER TABLE script DROP COLUMN IF EXISTS delete_after_secs;
backend/migrations/20260407154106_add_delete_after_secs.up.sql
+9
View File
@@ -0,0 +1,9 @@
ALTER TABLE script ADD COLUMN delete_after_secs INTEGER;
CREATE TABLE job_delete_schedule (
job_id UUID PRIMARY KEY,
workspace_id VARCHAR(50) NOT NULL,
delete_at TIMESTAMPTZ NOT NULL
);
CREATE INDEX idx_job_delete_schedule_delete_at ON job_delete_schedule (delete_at);
backend/src/monitor.rs
+96
View File
@@ -1205,6 +1205,92 @@ pub async fn delete_expired_items(db: &DB) -> () {
}
}
#[cfg(feature = "enterprise")]
async fn cleanup_scheduled_job_deletions(db: &Pool<Postgres>) {
const BATCH_SIZE: i64 = 1000;
const MAX_BATCHES: i32 = 10;
let mut total_deleted = 0u64;
for batch_num in 0..MAX_BATCHES {
let mut tx = match db.begin().await {
Ok(tx) => tx,
Err(e) => {
tracing::error!("Error starting transaction for scheduled job deletion: {e:?}");
break;
}
};
let rows = match sqlx::query_scalar!(
"DELETE FROM job_delete_schedule
WHERE job_id IN (
SELECT job_id FROM job_delete_schedule
WHERE delete_at <= now()
ORDER BY delete_at
LIMIT $1
FOR UPDATE SKIP LOCKED
)
RETURNING job_id",
BATCH_SIZE,
)
.fetch_all(&mut *tx)
.await
{
Ok(rows) => rows,
Err(e) => {
tracing::error!("Error in scheduled job deletion batch {batch_num}: {e:?}");
break;
}
};
if rows.is_empty() {
break;
}
let job_ids = rows;
let count = job_ids.len() as u64;
let cleanup_result: Result<(), sqlx::Error> = async {
sqlx::query!(
"UPDATE v2_job SET args = '{}'::jsonb WHERE id = ANY($1)",
&job_ids,
)
.execute(&mut *tx)
.await?;
sqlx::query!(
"UPDATE v2_job_completed SET result = '{}'::jsonb WHERE id = ANY($1)",
&job_ids,
)
.execute(&mut *tx)
.await?;
sqlx::query!(
"UPDATE job_logs SET logs = '##DELETED##' WHERE job_id = ANY($1)",
&job_ids,
)
.execute(&mut *tx)
.await?;
Ok(())
}
.await;
if let Err(e) = cleanup_result {
// Roll back so schedule rows survive for retry on next cycle
tracing::error!("Error cleaning job data in batch {batch_num}, rolling back: {e:?}");
break;
}
if let Err(e) = tx.commit().await {
tracing::error!("Error committing scheduled job deletion batch {batch_num}: {e:?}");
break;
}
total_deleted += count;
}
if total_deleted > 0 {
tracing::info!("Scheduled job deletion: cleaned {total_deleted} jobs");
}
}
pub async fn check_expiring_tokens(db: &DB) {
// Find tokens expiring within 7 days that still have a pending notification row.
// The notification table stores token_hash (not plaintext) so the join works
@@ -2328,6 +2414,15 @@ pub async fn monitor_db(
}
};
let cleanup_scheduled_job_deletions_f = async {
#[cfg(feature = "enterprise")]
if server_mode && !initial_load {
if let Some(db) = conn.as_sql() {
cleanup_scheduled_job_deletions(&db).await;
}
}
};
join!(
expired_items_f,
zombie_jobs_f,
@@ -2351,6 +2446,7 @@ pub async fn monitor_db(
cleanup_notify_events_f,
check_expiring_tokens_f,
manage_audit_partitions_f,
cleanup_scheduled_job_deletions_f,
);
}
backend/tests/delete_after_secs.rs
+92
View File
@@ -0,0 +1,92 @@
use sqlx::{Pool, Postgres};
use windmill_common::jobs::{resolve_delete_after_secs, schedule_job_deletion};
use windmill_test_utils::*;
// ---------------------------------------------------------------------------
// Unit tests for resolve_delete_after_secs
// ---------------------------------------------------------------------------
#[test]
fn test_resolve_no_deletion() {
assert_eq!(resolve_delete_after_secs(None, None), None);
assert_eq!(resolve_delete_after_secs(Some(false), None), None);
}
#[test]
fn test_resolve_immediate_backward_compat() {
// delete_after_use=true with no secs → immediate (0)
assert_eq!(resolve_delete_after_secs(Some(true), None), Some(0));
}
#[test]
fn test_resolve_explicit_secs() {
assert_eq!(resolve_delete_after_secs(None, Some(0)), Some(0));
assert_eq!(resolve_delete_after_secs(None, Some(3600)), Some(3600));
assert_eq!(resolve_delete_after_secs(Some(true), Some(60)), Some(60));
assert_eq!(resolve_delete_after_secs(Some(false), Some(120)), Some(120));
}
#[test]
fn test_resolve_rejects_negative() {
assert_eq!(resolve_delete_after_secs(None, Some(-1)), None);
assert_eq!(resolve_delete_after_secs(Some(true), Some(-100)), None);
}
// ---------------------------------------------------------------------------
// Integration: schedule_job_deletion inserts into job_delete_schedule
// ---------------------------------------------------------------------------
#[sqlx::test(fixtures("base"))]
async fn test_schedule_job_deletion_inserts_row(db: Pool<Postgres>) -> anyhow::Result<()> {
initialize_tracing().await;
let job_id = uuid::Uuid::new_v4();
schedule_job_deletion(&db, job_id, "test-workspace", 3600).await?;
let row = sqlx::query_as::<_, (uuid::Uuid, String)>(
"SELECT job_id, workspace_id FROM job_delete_schedule WHERE job_id = $1",
)
.bind(job_id)
.fetch_one(&db)
.await?;
assert_eq!(row.0, job_id);
assert_eq!(row.1, "test-workspace");
// Verify delete_at is approximately now + 3600s
let delete_at: chrono::DateTime<chrono::Utc> =
sqlx::query_scalar("SELECT delete_at FROM job_delete_schedule WHERE job_id = $1")
.bind(job_id)
.fetch_one(&db)
.await?;
let expected_min = chrono::Utc::now() + chrono::Duration::seconds(3500);
let expected_max = chrono::Utc::now() + chrono::Duration::seconds(3700);
assert!(
delete_at > expected_min && delete_at < expected_max,
"delete_at should be ~1 hour from now, got {delete_at}"
);
Ok(())
}
#[sqlx::test(fixtures("base"))]
async fn test_schedule_job_deletion_is_idempotent(db: Pool<Postgres>) -> anyhow::Result<()> {
initialize_tracing().await;
let job_id = uuid::Uuid::new_v4();
schedule_job_deletion(&db, job_id, "test-workspace", 60).await?;
// Second call should not error (ON CONFLICT DO NOTHING)
schedule_job_deletion(&db, job_id, "test-workspace", 120).await?;
let count: i64 =
sqlx::query_scalar("SELECT COUNT(*) FROM job_delete_schedule WHERE job_id = $1")
.bind(job_id)
.fetch_one(&db)
.await?;
assert_eq!(count, 1, "should have exactly one row (idempotent)");
Ok(())
}
backend/tests/dependency_map.rs
+1 -1
View File
@@ -35,7 +35,7 @@ mod dependency_map {
schema: std::collections::HashMap::new(),
ws_error_handler_muted: Some(false),
priority: None,
delete_after_use: None,
delete_after_secs: None,
timeout: None,
restart_unless_cancelled: None,
deployment_message: None,
backend/tests/flow_engine_parity.rs
+734 -537
View File
File diff suppressed because it is too large Load Diff
backend/tests/relock_skip.rs
+1 -1
View File
@@ -33,7 +33,7 @@ mod relock_skip {
schema: std::collections::HashMap::new(),
ws_error_handler_muted: Some(false),
priority: None,
delete_after_use: None,
delete_after_secs: None,
timeout: None,
restart_unless_cancelled: None,
deployment_message: None,
backend/tests/worker.rs
+8
View File
@@ -206,6 +206,7 @@ async fn test_deno_flow(db: Pool<Postgres>) -> anyhow::Result<()> {
timeout: None,
priority: None,
delete_after_use: None,
delete_after_secs: None,
continue_on_error: None,
skip_if: None,
apply_preprocessor: None,
@@ -253,6 +254,7 @@ async fn test_deno_flow(db: Pool<Postgres>) -> anyhow::Result<()> {
timeout: None,
priority: None,
delete_after_use: None,
delete_after_secs: None,
continue_on_error: None,
skip_if: None,
apply_preprocessor: None,
@@ -274,6 +276,7 @@ async fn test_deno_flow(db: Pool<Postgres>) -> anyhow::Result<()> {
timeout: None,
priority: None,
delete_after_use: None,
delete_after_secs: None,
continue_on_error: None,
skip_if: None,
apply_preprocessor: None,
@@ -389,6 +392,7 @@ async fn test_deno_flow_same_worker(db: Pool<Postgres>) -> anyhow::Result<()> {
timeout: None,
priority: None,
delete_after_use: None,
delete_after_secs: None,
continue_on_error: None,
skip_if: None,
apply_preprocessor: None,
@@ -445,6 +449,7 @@ async fn test_deno_flow_same_worker(db: Pool<Postgres>) -> anyhow::Result<()> {
timeout: None,
priority: None,
delete_after_use: None,
delete_after_secs: None,
continue_on_error: None,
skip_if: None,
apply_preprocessor: None,
@@ -487,6 +492,7 @@ async fn test_deno_flow_same_worker(db: Pool<Postgres>) -> anyhow::Result<()> {
timeout: None,
priority: None,
delete_after_use: None,
delete_after_secs: None,
continue_on_error: None,
skip_if: None,
apply_preprocessor: None,
@@ -508,6 +514,7 @@ async fn test_deno_flow_same_worker(db: Pool<Postgres>) -> anyhow::Result<()> {
timeout: None,
priority: None,
delete_after_use: None,
delete_after_secs: None,
continue_on_error: None,
skip_if: None,
apply_preprocessor: None,
@@ -556,6 +563,7 @@ async fn test_deno_flow_same_worker(db: Pool<Postgres>) -> anyhow::Result<()> {
timeout: None,
priority: None,
delete_after_use: None,
delete_after_secs: None,
continue_on_error: None,
skip_if: None,
apply_preprocessor: None,
backend/windmill-api-client/src/lib.rs
+2 -2
View File
@@ -364,7 +364,7 @@ pub mod types {
#[serde(default, skip_serializing_if = "Option::is_none")]
pub dedicated_worker: Option<bool>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub delete_after_use: Option<bool>,
pub delete_after_secs: Option<i32>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub deployment_message: Option<String>,
pub description: String,
@@ -540,7 +540,7 @@ pub mod types {
#[serde(default, skip_serializing_if = "Option::is_none")]
pub continue_on_error: Option<bool>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub delete_after_use: Option<bool>,
pub delete_after_secs: Option<i32>,
pub id: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub mock: Option<serde_json::Value>,
backend/windmill-api-flows/src/flows.rs
+6
View File
@@ -1856,6 +1856,7 @@ mod tests {
timeout: None,
priority: None,
delete_after_use: None,
delete_after_secs: None,
continue_on_error: None,
skip_if: None,
apply_preprocessor: None,
@@ -1890,6 +1891,7 @@ mod tests {
timeout: None,
priority: None,
delete_after_use: None,
delete_after_secs: None,
continue_on_error: None,
skip_if: None,
apply_preprocessor: None,
@@ -1924,6 +1926,7 @@ mod tests {
timeout: None,
priority: None,
delete_after_use: None,
delete_after_secs: None,
continue_on_error: None,
skip_if: None,
apply_preprocessor: None,
@@ -1957,6 +1960,7 @@ mod tests {
timeout: None,
priority: None,
delete_after_use: None,
delete_after_secs: None,
continue_on_error: None,
skip_if: None,
apply_preprocessor: None,
@@ -1972,6 +1976,8 @@ mod tests {
early_return: None,
chat_input_enabled: None,
flow_env: None,
delete_after_use: None,
delete_after_secs: None,
concurrency_settings: ConcurrencySettings::default(),
debouncing_settings: DebouncingSettings::default(),
};
backend/windmill-api-jobs/src/execution.rs
+32 -11
View File
@@ -476,6 +476,25 @@ pub async fn delete_job_metadata_after_use(db: &DB, job_uuid: Uuid) -> Result<()
Ok(())
}
pub use windmill_common::jobs::resolve_delete_after_secs;
pub use windmill_common::jobs::schedule_job_deletion;
/// Handle deletion or scheduling for a completed job.
pub async fn handle_delete_after_completion(
db: &DB,
job_uuid: Uuid,
w_id: &str,
delete_after_use: Option<bool>,
delete_after_secs: Option<i32>,
) -> Result<(), Error> {
match resolve_delete_after_secs(delete_after_use, delete_after_secs) {
Some(0) => delete_job_metadata_after_use(db, job_uuid).await?,
Some(secs) => schedule_job_deletion(db, job_uuid, w_id, secs).await?,
None => {}
}
Ok(())
}
// ---------------------------------------------------------------------------
// Queue limit check
// ---------------------------------------------------------------------------
@@ -799,7 +818,7 @@ pub async fn push_script_job_by_path_into_queue<'c>(
trigger: Option<TriggerMetadata>,
) -> error::Result<(
Uuid,
Option<bool>,
Option<i32>,
Option<sqlx::Transaction<'c, sqlx::Postgres>>,
)> {
#[cfg(feature = "enterprise")]
@@ -809,14 +828,16 @@ pub async fn push_script_job_by_path_into_queue<'c>(
check_scopes(&authed, || format!("jobs:run:scripts:{script_path}"))?;
let userdb_authed = UserDbWithAuthed { db: user_db.clone(), authed: &authed.to_authed_ref() };
let (job_payload, tag, delete_after_use, timeout, on_behalf_of) = script_path_to_payload(
script_path,
Some(userdb_authed),
db.clone(),
&w_id,
run_query.skip_preprocessor,
)
.await?;
let (job_payload, tag, delete_after_use, delete_after_secs, timeout, on_behalf_of) =
script_path_to_payload(
script_path,
Some(userdb_authed),
db.clone(),
&w_id,
run_query.skip_preprocessor,
)
.await?;
let resolved_delete_secs = resolve_delete_after_secs(delete_after_use, delete_after_secs);
let scheduled_for = run_query.get_scheduled_for(&db).await?;
let tag = run_query.tag.clone().or(tag);
@@ -886,9 +907,9 @@ pub async fn push_script_job_by_path_into_queue<'c>(
// If we were given a transaction, return it; otherwise commit it
if return_tx {
Ok((uuid, delete_after_use, Some(tx)))
Ok((uuid, resolved_delete_secs, Some(tx)))
} else {
tx.commit().await?;
Ok((uuid, delete_after_use, None))
Ok((uuid, resolved_delete_secs, None))
}
}
backend/windmill-api-scripts/src/scripts.rs
+7 -3
View File
@@ -113,6 +113,8 @@ pub struct ScriptWDraft<SR> {
#[serde(skip_serializing_if = "Option::is_none")]
pub delete_after_use: Option<bool>,
#[serde(skip_serializing_if = "Option::is_none")]
pub delete_after_secs: Option<i32>,
#[serde(skip_serializing_if = "Option::is_none")]
pub timeout: Option<i32>,
#[serde(skip_serializing_if = "Option::is_none")]
pub visible_to_runner_only: Option<bool>,
@@ -178,6 +180,7 @@ impl ScriptWDraft<ScriptRunnableSettingsHandle> {
priority: self.priority,
restart_unless_cancelled: self.restart_unless_cancelled,
delete_after_use: self.delete_after_use,
delete_after_secs: self.delete_after_secs,
timeout: self.timeout,
visible_to_runner_only: self.visible_to_runner_only,
auto_kind: self.auto_kind,
@@ -943,8 +946,8 @@ async fn create_script_internal<'c>(
content, created_by, schema, is_template, extra_perms, lock, language, kind, tag, \
draft_only, envs, concurrent_limit, concurrency_time_window_s, cache_ttl, \
dedicated_worker, ws_error_handler_muted, priority, restart_unless_cancelled, \
delete_after_use, timeout, concurrency_key, visible_to_runner_only, auto_kind, codebase, has_preprocessor, on_behalf_of_email, schema_validation, assets, debounce_key, debounce_delay_s, cache_ignore_s3_path, runnable_settings_handle, modules, labels) \
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9::text::json, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19, $20, $21, $22, $23, $24, $25, $26, $27, $28, $29, $30, $31, $32, $33, $34, $35, $36, $37, $38, $39, $40)",
delete_after_use, delete_after_secs, timeout, concurrency_key, visible_to_runner_only, auto_kind, codebase, has_preprocessor, on_behalf_of_email, schema_validation, assets, debounce_key, debounce_delay_s, cache_ignore_s3_path, runnable_settings_handle, modules, labels) \
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9::text::json, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19, $20, $21, $22, $23, $24, $25, $26, $27, $28, $29, $30, $31, $32, $33, $34, $35, $36, $37, $38, $39, $40, $41)",
&w_id,
&hash.0,
ns.path,
@@ -970,12 +973,13 @@ async fn create_script_internal<'c>(
ns.priority,
ns.restart_unless_cancelled,
ns.delete_after_use,
ns.delete_after_secs,
ns.timeout,
guarded_concurrency_key,
ns.visible_to_runner_only,
auto_kind.as_deref(),
codebase,
has_preprocessor.filter(|x: &bool| *x), // should be Some(true) or None
has_preprocessor.filter(|x: &bool| *x),
windmill_common::resolve_on_behalf_of_email(
ns.on_behalf_of_email.as_deref(),
ns.preserve_on_behalf_of.unwrap_or(false),
backend/windmill-api/openapi.yaml
+6 -4
View File
@@ -20165,8 +20165,9 @@ components:
type: boolean
timeout:
type: integer
delete_after_use:
type: boolean
delete_after_secs:
type: integer
description: If set, delete the job's args, result and logs after this many seconds following job completion
visible_to_runner_only:
type: boolean
auto_kind:
@@ -20257,8 +20258,9 @@ components:
type: boolean
timeout:
type: integer
delete_after_use:
type: boolean
delete_after_secs:
type: integer
description: If set, delete the job's args, result and logs after this many seconds following job completion
deployment_message:
type: string
concurrency_key:
backend/windmill-api/src/jobs.rs
+72 -72
View File
@@ -4281,49 +4281,51 @@ pub async fn run_workflow_as_code(
)
.await?;
let (job_payload, tag, _delete_after_use, timeout, on_behalf_of) = match job.job_kind {
JobKind::Preview => (
JobPayload::Code(RawCode {
hash: None,
content: raw_code.unwrap_or_default(),
path: job.script_path,
language: job.language.unwrap_or_else(|| ScriptLang::Deno),
lock: raw_lock,
concurrency_settings: concurrency_settings
.maybe_fallback(
windmill_queue::custom_concurrency_key(&db, &job.id)
.await
.map_err(to_anyhow)?,
job.concurrent_limit,
job.concurrency_time_window_s,
)
.into(),
cache_ttl: job.cache_ttl,
cache_ignore_s3_path: job.cache_ignore_s3_path,
dedicated_worker: None,
// TODO(debouncing): enable for this mode
debouncing_settings: DebouncingSettings::default(),
modules: None,
}),
Some(job.tag.clone()),
None,
run_query.timeout,
None,
),
JobKind::Script => {
let userdb_authed =
UserDbWithAuthed { db: user_db.clone(), authed: &authed.to_authed_ref() };
script_path_to_payload(
job.script_path(),
Some(userdb_authed),
db.clone(),
&w_id,
run_query.skip_preprocessor,
)
.await?
}
_ => return Err(anyhow::anyhow!("Not supported").into()),
};
let (job_payload, tag, _delete_after_use, _delete_after_secs, timeout, on_behalf_of) =
match job.job_kind {
JobKind::Preview => (
JobPayload::Code(RawCode {
hash: None,
content: raw_code.unwrap_or_default(),
path: job.script_path,
language: job.language.unwrap_or_else(|| ScriptLang::Deno),
lock: raw_lock,
concurrency_settings: concurrency_settings
.maybe_fallback(
windmill_queue::custom_concurrency_key(&db, &job.id)
.await
.map_err(to_anyhow)?,
job.concurrent_limit,
job.concurrency_time_window_s,
)
.into(),
cache_ttl: job.cache_ttl,
cache_ignore_s3_path: job.cache_ignore_s3_path,
dedicated_worker: None,
// TODO(debouncing): enable for this mode
debouncing_settings: DebouncingSettings::default(),
modules: None,
}),
Some(job.tag.clone()),
None,
None,
run_query.timeout,
None,
),
JobKind::Script => {
let userdb_authed =
UserDbWithAuthed { db: user_db.clone(), authed: &authed.to_authed_ref() };
script_path_to_payload(
job.script_path(),
Some(userdb_authed),
db.clone(),
&w_id,
run_query.skip_preprocessor,
)
.await?
}
_ => return Err(anyhow::anyhow!("Not supported").into()),
};
if *CLOUD_HOSTED {
tracing::info!("workflow_as_code_tracing id {i} ");
@@ -4554,14 +4556,15 @@ pub async fn run_wait_result_job_by_path_get(
let user_db_with_authed =
UserDbWithAuthed { db: user_db.clone(), authed: &authed.to_authed_ref() };
let (job_payload, tag, delete_after_use, timeout, on_behalf_authed) = script_path_to_payload(
script_path,
Some(user_db_with_authed),
db.clone(),
&w_id,
run_query.skip_preprocessor,
)
.await?;
let (job_payload, tag, delete_after_use, delete_after_secs, timeout, on_behalf_authed) =
script_path_to_payload(
script_path,
Some(user_db_with_authed),
db.clone(),
&w_id,
run_query.skip_preprocessor,
)
.await?;
let tag = run_query.tag.clone().or(tag);
check_tag_available_for_workspace(&db, &w_id, &tag, &authed).await?;
@@ -4617,9 +4620,7 @@ pub async fn run_wait_result_job_by_path_get(
tx.commit().await?;
let wait_result = run_wait_result(&db, uuid, &w_id, None, &authed.username).await;
if delete_after_use.unwrap_or(false) {
delete_job_metadata_after_use(&db, uuid).await?;
}
handle_delete_after_completion(&db, uuid, &w_id, delete_after_use, delete_after_secs).await?;
return wait_result;
}
@@ -4699,14 +4700,15 @@ pub async fn run_wait_result_script_by_path_internal(
check_queue_too_long(&db, QUEUE_LIMIT_WAIT_RESULT.or(run_query.queue_limit)).await?;
let db_authed = UserDbWithAuthed { db: user_db.clone(), authed: &authed.to_authed_ref() };
let (job_payload, tag, delete_after_use, timeout, on_behalf_of) = script_path_to_payload(
script_path.to_path(),
Some(db_authed),
db.clone(),
&w_id,
run_query.skip_preprocessor,
)
.await?;
let (job_payload, tag, delete_after_use, delete_after_secs, timeout, on_behalf_of) =
script_path_to_payload(
script_path.to_path(),
Some(db_authed),
db.clone(),
&w_id,
run_query.skip_preprocessor,
)
.await?;
let tag = run_query.tag.clone().or(tag);
check_tag_available_for_workspace(&db, &w_id, &tag, &authed).await?;
@@ -4762,9 +4764,7 @@ pub async fn run_wait_result_script_by_path_internal(
tx.commit().await?;
let wait_result = run_wait_result(&db, uuid, &w_id, None, &authed.username).await;
if delete_after_use.unwrap_or(false) {
delete_job_metadata_after_use(&db, uuid).await?;
}
handle_delete_after_completion(&db, uuid, &w_id, delete_after_use, delete_after_secs).await?;
return wait_result;
}
@@ -4802,6 +4802,7 @@ pub async fn run_wait_result_script_by_hash(
dedicated_worker,
priority,
delete_after_use,
delete_after_secs,
timeout,
has_preprocessor,
on_behalf_of_email,
@@ -4888,9 +4889,7 @@ pub async fn run_wait_result_script_by_hash(
tx.commit().await?;
let wait_result = run_wait_result(&db, uuid, &w_id, None, &authed.username).await;
if delete_after_use.unwrap_or(false) {
delete_job_metadata_after_use(&db, uuid).await?;
}
handle_delete_after_completion(&db, uuid, &w_id, delete_after_use, delete_after_secs).await?;
return wait_result;
}
@@ -5060,7 +5059,7 @@ pub async fn stream_job(
(uuid, None)
}
RunnableId::ScriptId(ScriptId::ScriptHash(script_hash)) => {
let (uuid, _) = run_job_by_hash_inner(
let (uuid, _, _) = run_job_by_hash_inner(
authed.clone(),
db.clone(),
user_db,
@@ -6543,7 +6542,7 @@ pub async fn run_job_by_hash(
)
.await?;
let (uuid, _) = run_job_by_hash_inner(
let (uuid, _, _) = run_job_by_hash_inner(
authed,
db,
user_db,
@@ -6567,7 +6566,7 @@ pub async fn run_job_by_hash_inner(
run_query: RunJobQuery,
args: PushArgsOwned,
trigger: Option<TriggerMetadata>,
) -> error::Result<(Uuid, Option<bool>)> {
) -> error::Result<(Uuid, Option<bool>, Option<i32>)> {
#[cfg(feature = "enterprise")]
check_license_key_valid().await?;
@@ -6588,6 +6587,7 @@ pub async fn run_job_by_hash_inner(
on_behalf_of_email,
created_by,
delete_after_use,
delete_after_secs,
labels,
..
} = get_script_info_for_hash(Some(userdb_authed), &db, &w_id, hash)
@@ -6668,7 +6668,7 @@ pub async fn run_job_by_hash_inner(
.await?;
tx.commit().await?;
Ok((uuid, delete_after_use))
Ok((uuid, delete_after_use, delete_after_secs))
}
async fn get_log_file(Path((_w_id, file_p)): Path<(String, String)>) -> error::Result<Response> {
backend/windmill-api/src/workspaces_export.rs
+2 -2
View File
@@ -84,7 +84,7 @@ struct ScriptMetadata {
#[serde(skip_serializing_if = "Option::is_none")]
pub timeout: Option<i32>,
#[serde(skip_serializing_if = "Option::is_none")]
pub delete_after_use: Option<bool>,
pub delete_after_secs: Option<i32>,
#[serde(skip_serializing_if = "Option::is_none")]
pub restart_unless_cancelled: Option<bool>,
#[serde(skip_serializing_if = "Option::is_none")]
@@ -508,7 +508,7 @@ pub(crate) async fn tarball_workspace(
priority: script.priority,
tag: script.tag,
timeout: script.timeout,
delete_after_use: script.delete_after_use,
delete_after_secs: script.delete_after_secs,
restart_unless_cancelled: script.restart_unless_cancelled,
visible_to_runner_only: script.visible_to_runner_only,
auto_kind: script.auto_kind,
backend/windmill-common/src/jobs.rs
+108 -67
View File
@@ -43,6 +43,39 @@ pub fn get_has_preprocessor_from_content_and_lang(
Ok(has_preprocessor)
}
pub async fn schedule_job_deletion(
db: &DB,
job_id: uuid::Uuid,
w_id: &str,
delete_after_secs: i32,
) -> crate::error::Result<()> {
sqlx::query!(
"INSERT INTO job_delete_schedule (job_id, workspace_id, delete_at) \
VALUES ($1, $2, now() + make_interval(secs => $3::double precision)) \
ON CONFLICT (job_id) DO NOTHING",
job_id,
w_id,
delete_after_secs as f64,
)
.execute(db)
.await?;
Ok(())
}
/// Resolve effective delete behavior from delete_after_use (bool) and delete_after_secs.
/// Returns Some(secs) if deletion should happen, None otherwise.
pub fn resolve_delete_after_secs(
delete_after_use: Option<bool>,
delete_after_secs: Option<i32>,
) -> Option<i32> {
match (delete_after_use, delete_after_secs) {
(_, Some(secs)) if secs >= 0 => Some(secs),
(_, Some(_)) => None, // reject negative values
(Some(true), None) => Some(0), // backward compat: immediate
_ => None,
}
}
pub async fn script_path_to_payload<'e>(
script_path: &str,
db_authed: Option<UserDbWithAuthed<'e, AuthedRef<'e>>>,
@@ -54,88 +87,94 @@ pub async fn script_path_to_payload<'e>(
Option<Tag>,
Option<bool>,
Option<i32>,
Option<i32>,
Option<OnBehalfOf>,
)> {
let (job_payload, tag, delete_after_use, script_timeout, on_behalf_of) = if script_path
.starts_with("hub/")
{
let hub_script =
get_full_hub_script_by_path(StripPath(script_path.to_string()), &HTTP_CLIENT, None)
.await?;
let (job_payload, tag, delete_after_use, delete_after_secs, script_timeout, on_behalf_of) =
if script_path.starts_with("hub/") {
let hub_script =
get_full_hub_script_by_path(StripPath(script_path.to_string()), &HTTP_CLIENT, None)
.await?;
let has_preprocessor =
get_has_preprocessor_from_content_and_lang(&hub_script.content, &hub_script.language)?;
let has_preprocessor = get_has_preprocessor_from_content_and_lang(
&hub_script.content,
&hub_script.language,
)?;
(
JobPayload::ScriptHub {
path: script_path.to_owned(),
apply_preprocessor: has_preprocessor && !skip_preprocessor.unwrap_or(false),
},
None,
None,
None,
None,
)
} else {
let ScriptHashInfo {
hash,
tag,
runnable_settings:
super::scripts::ScriptRunnableSettingsInline {
concurrency_settings,
debouncing_settings,
(
JobPayload::ScriptHub {
path: script_path.to_owned(),
apply_preprocessor: has_preprocessor && !skip_preprocessor.unwrap_or(false),
},
cache_ttl,
cache_ignore_s3_path,
language,
dedicated_worker,
priority,
delete_after_use,
timeout,
has_preprocessor,
on_behalf_of_email,
created_by,
labels,
..
} = get_latest_deployed_hash_for_path(db_authed, db.clone(), w_id, script_path)
.await?
.prefetch_cached(&db)
.await?;
let on_behalf_of = if let Some(email) = on_behalf_of_email {
Some(OnBehalfOf {
email,
permissioned_as: username_to_permissioned_as(created_by.as_str()),
})
None,
None,
None,
None,
None,
)
} else {
None
};
(
JobPayload::ScriptHash {
hash: ScriptHash(hash),
path: script_path.to_owned(),
let ScriptHashInfo {
hash,
tag,
runnable_settings:
super::scripts::ScriptRunnableSettingsInline {
concurrency_settings,
debouncing_settings,
},
cache_ttl,
cache_ignore_s3_path,
language,
dedicated_worker,
priority,
apply_preprocessor: !skip_preprocessor.unwrap_or(false)
&& has_preprocessor.unwrap_or(false),
debouncing_settings,
concurrency_settings,
delete_after_use,
delete_after_secs,
timeout,
has_preprocessor,
on_behalf_of_email,
created_by,
labels,
},
tag,
delete_after_use,
timeout,
on_behalf_of,
)
};
..
} = get_latest_deployed_hash_for_path(db_authed, db.clone(), w_id, script_path)
.await?
.prefetch_cached(&db)
.await?;
let on_behalf_of = if let Some(email) = on_behalf_of_email {
Some(OnBehalfOf {
email,
permissioned_as: username_to_permissioned_as(created_by.as_str()),
})
} else {
None
};
(
JobPayload::ScriptHash {
hash: ScriptHash(hash),
path: script_path.to_owned(),
cache_ttl,
cache_ignore_s3_path,
language,
dedicated_worker,
priority,
apply_preprocessor: !skip_preprocessor.unwrap_or(false)
&& has_preprocessor.unwrap_or(false),
debouncing_settings,
concurrency_settings,
labels,
},
tag,
delete_after_use,
delete_after_secs,
timeout,
on_behalf_of,
)
};
Ok((
job_payload,
tag,
delete_after_use,
delete_after_secs,
script_timeout,
on_behalf_of,
))
@@ -146,7 +185,7 @@ pub async fn get_payload_tag_from_prefixed_path(
db: &DB,
w_id: &str,
) -> Result<(JobPayload, Option<String>, Option<OnBehalfOf>), Error> {
let (payload, tag, _, _, on_behalf_of) = if path.starts_with("script/") {
let (payload, tag, _, _, _, on_behalf_of) = if path.starts_with("script/") {
script_path_to_payload(
path.strip_prefix("script/").unwrap(),
None,
@@ -170,6 +209,7 @@ pub async fn get_payload_tag_from_prefixed_path(
None,
None,
None,
None,
)
} else {
let FlowVersionInfo { dedicated_worker, tag, version, labels, .. } =
@@ -186,6 +226,7 @@ pub async fn get_payload_tag_from_prefixed_path(
None,
None,
None,
None,
)
}
} else {
backend/windmill-common/src/lib.rs
+3
View File
@@ -1037,6 +1037,7 @@ pub struct ScriptHashInfo<SR> {
pub dedicated_worker: Option<bool>,
pub priority: Option<i16>,
pub delete_after_use: Option<bool>,
pub delete_after_secs: Option<i32>,
pub timeout: Option<i32>,
pub has_preprocessor: Option<bool>,
pub on_behalf_of_email: Option<String>,
@@ -1067,6 +1068,7 @@ impl ScriptHashInfo<ScriptRunnableSettingsHandle> {
dedicated_worker: self.dedicated_worker,
priority: self.priority,
delete_after_use: self.delete_after_use,
delete_after_secs: self.delete_after_secs,
timeout: self.timeout,
has_preprocessor: self.has_preprocessor,
on_behalf_of_email: self.on_behalf_of_email,
@@ -1236,6 +1238,7 @@ async fn get_script_info_for_hash_inner<'e, E: sqlx::PgExecutor<'e>>(
dedicated_worker,
priority,
delete_after_use,
delete_after_secs,
timeout,
has_preprocessor,
on_behalf_of_email,
backend/windmill-common/src/scripts.rs
+2
View File
@@ -106,6 +106,7 @@ pub async fn prefetch_cached_script(
cache_ignore_s3_path: script.cache_ignore_s3_path,
timeout: script.timeout,
delete_after_use: script.delete_after_use,
delete_after_secs: script.delete_after_secs,
restart_unless_cancelled: script.restart_unless_cancelled,
visible_to_runner_only: script.visible_to_runner_only,
auto_kind: script.auto_kind,
@@ -420,6 +421,7 @@ pub async fn clone_script<'c>(
priority: s.priority,
timeout: s.timeout,
delete_after_use: s.delete_after_use,
delete_after_secs: s.delete_after_secs,
restart_unless_cancelled: s.restart_unless_cancelled,
deployment_message,
visible_to_runner_only: s.visible_to_runner_only,
backend/windmill-queue/src/jobs.rs
+2
View File
@@ -5217,6 +5217,8 @@ async fn push_inner<'c, 'd>(
preprocessor_module: None,
chat_input_enabled: None,
flow_env: None,
delete_after_use: None,
delete_after_secs: None,
};
// this is a new flow being pushed, flow_status is set to flow_value:
let flow_status: FlowStatus = FlowStatus::new(&flow_value);
backend/windmill-runtime-nativets/src/windmill-client.js
+6 -6
View File
@@ -253,8 +253,8 @@ var $Script = {
timeout: {
type: "integer",
},
delete_after_use: {
type: "boolean",
delete_after_secs: {
type: "integer",
},
visible_to_runner_only: {
type: "boolean",
@@ -369,8 +369,8 @@ var $NewScript = {
timeout: {
type: "integer",
},
delete_after_use: {
type: "boolean",
delete_after_secs: {
type: "integer",
},
deployment_message: {
type: "string",
@@ -2830,8 +2830,8 @@ var $FlowModule = {
timeout: {
type: "number",
},
delete_after_use: {
type: "boolean",
delete_after_secs: {
type: "integer",
},
summary: {
type: "string",
backend/windmill-test-utils/src/lib.rs
+2 -2
View File
@@ -753,7 +753,7 @@ pub async fn assert_lockfile(
schema: std::collections::HashMap::new(),
ws_error_handler_muted: Some(false),
priority: None,
delete_after_use: None,
delete_after_secs: None,
timeout: None,
restart_unless_cancelled: None,
deployment_message: None,
@@ -851,7 +851,7 @@ pub async fn run_deployed_relative_imports(
schema: std::collections::HashMap::new(),
ws_error_handler_muted: Some(false),
priority: None,
delete_after_use: None,
delete_after_secs: None,
timeout: None,
restart_unless_cancelled: None,
deployment_message: None,
backend/windmill-trigger/src/trigger_helpers.rs
+29 -21
View File
@@ -31,7 +31,7 @@ use windmill_api_jobs::{
execution::{
check_tag_available_for_workspace, delete_job_metadata_after_use,
push_flow_job_by_path_into_queue, push_script_job_by_path_into_queue, result_to_response,
run_wait_result_internal,
run_wait_result_internal, schedule_job_deletion,
},
types::RunJobQuery,
};
@@ -522,7 +522,7 @@ pub async fn trigger_runnable_inner<'c>(
suspended_mode: Option<bool>,
) -> Result<(
Uuid,
Option<bool>,
Option<i32>,
Option<String>,
Option<sqlx::Transaction<'c, sqlx::Postgres>>,
)> {
@@ -536,7 +536,7 @@ pub async fn trigger_runnable_inner<'c>(
});
let user_db = user_db.unwrap_or_else(|| UserDB::new(db.clone()));
let (uuid, delete_after_use, early_return, tx_out) = if is_flow {
let (uuid, resolved_delete_secs, early_return, tx_out) = if is_flow {
let run_query = RunJobQuery { job_id, suspended_mode, ..Default::default() };
let path = StripPath(runnable_path.to_string());
let (uuid, early_return, tx_out) = push_flow_job_by_path_into_queue(
@@ -553,7 +553,7 @@ pub async fn trigger_runnable_inner<'c>(
.await?;
(uuid, None, early_return, tx_out)
} else {
let (uuid, delete_after_use, tx_out) = trigger_script_internal(
let (uuid, resolved_delete_secs, tx_out) = trigger_script_internal(
db,
tx_o,
user_db,
@@ -570,10 +570,10 @@ pub async fn trigger_runnable_inner<'c>(
suspended_mode,
)
.await?;
(uuid, delete_after_use, None, tx_out)
(uuid, resolved_delete_secs, None, tx_out)
};
Ok((uuid, delete_after_use, early_return, tx_out))
Ok((uuid, resolved_delete_secs, early_return, tx_out))
}
#[allow(dead_code)]
@@ -631,7 +631,7 @@ pub async fn trigger_runnable_and_wait_for_result(
trigger: TriggerMetadata,
) -> Result<axum::response::Response> {
let username = authed.username.clone();
let (uuid, delete_after_use, early_return, _) = trigger_runnable_inner(
let (uuid, resolved_delete_secs, early_return, _) = trigger_runnable_inner(
db,
None,
user_db,
@@ -652,8 +652,10 @@ pub async fn trigger_runnable_and_wait_for_result(
let (result, success) =
run_wait_result_internal(db, uuid, &workspace_id, early_return, &username).await?;
if delete_after_use.unwrap_or(false) {
delete_job_metadata_after_use(&db, uuid).await?;
match resolved_delete_secs {
Some(0) => delete_job_metadata_after_use(&db, uuid).await?,
Some(secs) => schedule_job_deletion(&db, uuid, &workspace_id, secs).await?,
None => {}
}
result_to_response(result, success)
@@ -675,7 +677,7 @@ pub async fn trigger_runnable_and_wait_for_raw_result(
trigger: TriggerMetadata,
) -> Result<(Box<RawValue>, bool)> {
let username = authed.username.clone();
let (uuid, delete_after_use, early_return, _) = trigger_runnable_inner(
let (uuid, resolved_delete_secs, early_return, _) = trigger_runnable_inner(
db,
None,
user_db,
@@ -705,8 +707,10 @@ pub async fn trigger_runnable_and_wait_for_raw_result(
)
})?;
if delete_after_use.unwrap_or(false) {
delete_job_metadata_after_use(&db, uuid).await?;
match resolved_delete_secs {
Some(0) => delete_job_metadata_after_use(&db, uuid).await?,
Some(secs) => schedule_job_deletion(&db, uuid, &workspace_id, secs).await?,
None => {}
}
Ok((result, success))
@@ -770,13 +774,13 @@ async fn trigger_script_internal<'c>(
suspended_mode: Option<bool>,
) -> Result<(
Uuid,
Option<bool>,
Option<i32>,
Option<sqlx::Transaction<'c, sqlx::Postgres>>,
)> {
if retry.is_none() && error_handler_path.is_none() {
let run_query = RunJobQuery { job_id, suspended_mode, ..Default::default() };
let path = StripPath(script_path.to_string());
let (uuid, delete_after_use, tx_out) = push_script_job_by_path_into_queue(
let (uuid, resolved_delete_secs, tx_out) = push_script_job_by_path_into_queue(
authed,
db.clone(),
tx_o,
@@ -788,9 +792,9 @@ async fn trigger_script_internal<'c>(
Some(trigger),
)
.await?;
Ok((uuid, delete_after_use, tx_out))
Ok((uuid, resolved_delete_secs, tx_out))
} else {
let (uuid, delete_after_use, tx_out) = trigger_script_with_retry_and_error_handler(
let (uuid, resolved_delete_secs, tx_out) = trigger_script_with_retry_and_error_handler(
db,
tx_o,
user_db,
@@ -807,7 +811,7 @@ async fn trigger_script_internal<'c>(
suspended_mode,
)
.await?;
Ok((uuid, delete_after_use, tx_out))
Ok((uuid, resolved_delete_secs, tx_out))
}
}
@@ -828,7 +832,7 @@ async fn trigger_script_with_retry_and_error_handler<'c>(
suspended_mode: Option<bool>,
) -> Result<(
Uuid,
Option<bool>,
Option<i32>,
Option<sqlx::Transaction<'c, sqlx::Postgres>>,
)> {
#[cfg(feature = "enterprise")]
@@ -840,7 +844,7 @@ async fn trigger_script_with_retry_and_error_handler<'c>(
let error_handler_path = error_handler_path.map(|p| p.to_string());
let error_handler_args = error_handler_args.map(|args| args.0.clone());
let (job_payload, tag, delete_after_use, timeout, on_behalf_of) = {
let (job_payload, tag, delete_after_use, delete_after_secs, timeout, on_behalf_of) = {
let db_authed = UserDbWithAuthed { db: user_db.clone(), authed: &authed.to_authed_ref() };
script_path_to_payload(
script_path,
@@ -851,6 +855,10 @@ async fn trigger_script_with_retry_and_error_handler<'c>(
)
.await?
};
let resolved_delete_secs = windmill_api_jobs::execution::resolve_delete_after_secs(
delete_after_use,
delete_after_secs,
);
check_tag_available_for_workspace(&db, &workspace_id, &tag, &authed).await?;
@@ -951,9 +959,9 @@ async fn trigger_script_with_retry_and_error_handler<'c>(
// If we were given a transaction, return it; otherwise commit it
if return_tx {
Ok((uuid, delete_after_use, Some(tx)))
Ok((uuid, resolved_delete_secs, Some(tx)))
} else {
tx.commit().await?;
Ok((uuid, delete_after_use, None))
Ok((uuid, resolved_delete_secs, None))
}
}
backend/windmill-types/src/flow_status.rs
+13 -1
View File
@@ -111,6 +111,15 @@ pub struct FlowCleanupModule {
#[serde(default)]
#[serde(skip_serializing_if = "Vec::is_empty")]
pub flow_jobs_to_clean: Vec<Uuid>,
#[serde(default)]
#[serde(skip_serializing_if = "Vec::is_empty")]
pub flow_jobs_to_schedule_clean: Vec<FlowJobScheduledClean>,
}
#[derive(Serialize, Deserialize, Debug, Clone)]
pub struct FlowJobScheduledClean {
pub id: Uuid,
pub delete_after_secs: i32,
}
#[derive(Deserialize, Serialize, Debug, Clone)]
@@ -513,7 +522,10 @@ impl FlowStatus {
} else {
None
},
cleanup_module: FlowCleanupModule { flow_jobs_to_clean: vec![] },
cleanup_module: FlowCleanupModule {
flow_jobs_to_clean: vec![],
flow_jobs_to_schedule_clean: vec![],
},
retry: RetryStatus { fail_count: 0, failed_jobs: vec![] },
restarted_from: None,
user_states: HashMap::new(),
backend/windmill-types/src/flows.rs
+8 -1
View File
@@ -184,6 +184,10 @@ pub struct FlowValue {
pub chat_input_enabled: Option<bool>,
#[serde(skip_serializing_if = "Option::is_none")]
pub flow_env: Option<HashMap<String, Box<RawValue>>>,
#[serde(skip_serializing, default)]
pub delete_after_use: Option<bool>,
#[serde(skip_serializing_if = "Option::is_none")]
pub delete_after_secs: Option<i32>,
}
impl FlowValue {
@@ -441,9 +445,11 @@ pub struct FlowModule {
pub timeout: Option<InputTransform>,
#[serde(skip_serializing_if = "Option::is_none")]
pub priority: Option<i16>,
#[serde(skip_serializing_if = "Option::is_none")]
#[serde(skip_serializing, default)]
pub delete_after_use: Option<bool>,
#[serde(skip_serializing_if = "Option::is_none")]
pub delete_after_secs: Option<i32>,
#[serde(skip_serializing_if = "Option::is_none")]
pub continue_on_error: Option<bool>,
#[serde(skip_serializing_if = "Option::is_none")]
pub skip_if: Option<SkipIf>,
@@ -1122,6 +1128,7 @@ pub fn add_virtual_items_if_necessary(modules: &mut Vec<FlowModule>) {
timeout: None,
priority: None,
delete_after_use: None,
delete_after_secs: None,
continue_on_error: None,
skip_if: None,
apply_preprocessor: None,
backend/windmill-types/src/scripts.rs
+5 -1
View File
@@ -357,9 +357,11 @@ pub struct Script<SR> {
pub cache_ignore_s3_path: Option<bool>,
#[serde(skip_serializing_if = "Option::is_none")]
pub timeout: Option<i32>,
#[serde(skip_serializing_if = "Option::is_none")]
#[serde(skip_serializing, default)]
pub delete_after_use: Option<bool>,
#[serde(skip_serializing_if = "Option::is_none")]
pub delete_after_secs: Option<i32>,
#[serde(skip_serializing_if = "Option::is_none")]
pub restart_unless_cancelled: Option<bool>,
#[serde(skip_serializing_if = "Option::is_none")]
pub visible_to_runner_only: Option<bool>,
@@ -505,7 +507,9 @@ pub struct NewScript {
pub ws_error_handler_muted: Option<bool>,
pub priority: Option<i16>,
pub timeout: Option<i32>,
#[serde(skip_serializing, default)]
pub delete_after_use: Option<bool>,
pub delete_after_secs: Option<i32>,
pub restart_unless_cancelled: Option<bool>,
pub deployment_message: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
backend/windmill-worker/src/ai/tools.rs
+3
View File
@@ -391,6 +391,7 @@ async fn execute_windmill_tool(
tool_module,
tag,
tool_module.delete_after_use.unwrap_or(false),
None,
)
}
FlowModuleValue::FlowScript { id, language, concurrency_settings, tag, .. } => {
@@ -407,6 +408,7 @@ async fn execute_windmill_tool(
},
tag: tag.clone(),
delete_after_use: tool_module.delete_after_use.unwrap_or(false),
delete_after_secs: None,
timeout: None,
on_behalf_of: None,
}
@@ -429,6 +431,7 @@ async fn execute_windmill_tool(
payload: JobPayload::AIAgent { path },
tag: None,
delete_after_use: tool_module.delete_after_use.unwrap_or(false),
delete_after_secs: None,
timeout: None,
on_behalf_of: None,
}
backend/windmill-worker/src/bun_executor.rs
+1 -1
View File
@@ -2590,7 +2590,7 @@ pub async fn handle_wac_v2_output(
}
"script" => {
// Resolve script path to job payload (handles hash, lang, etc.)
let (payload, _, _, _, _) = script_path_to_payload(
let (payload, _, _, _, _, _) = script_path_to_payload(
&step.script,
None, // no authed db for background workers
db.clone(),
backend/windmill-worker/src/worker_flow.rs
+270 -87
View File
@@ -1651,13 +1651,33 @@ pub async fn update_flow_status_after_job_completion_internal(
.warn_after_seconds(3)
.await?;
}
if !_cleanup_module.flow_jobs_to_schedule_clean.is_empty() {
let entries_json = serde_json::to_value(
&_cleanup_module.flow_jobs_to_schedule_clean,
)
.map_err(|e| {
error::Error::internal_err(format!(
"Unable to serialize scheduled clean entries: {e:#}"
))
})?;
sqlx::query!(
"UPDATE v2_job_status
SET flow_status = JSONB_SET(flow_status, ARRAY['cleanup_module', 'flow_jobs_to_schedule_clean'], COALESCE(flow_status->'cleanup_module'->'flow_jobs_to_schedule_clean', '[]'::jsonb) || $1)
WHERE id = $2",
entries_json,
parent_job
)
.execute(db)
.warn_after_seconds(3)
.await?;
}
} else {
// run the cleanup step only when the root job is complete
if !_cleanup_module.flow_jobs_to_clean.is_empty() {
tracing::debug!(
"Cleaning up jobs arguments, result and logs as they were marked as delete_after_use {:?}",
_cleanup_module.flow_jobs_to_clean
);
"Cleaning up jobs arguments, result and logs as they were marked as delete_after_use {:?}",
_cleanup_module.flow_jobs_to_clean
);
sqlx::query!(
"UPDATE v2_job SET args = '{}'::jsonb WHERE id = ANY($1)",
&_cleanup_module.flow_jobs_to_clean,
@@ -1677,6 +1697,92 @@ pub async fn update_flow_status_after_job_completion_internal(
Error::internal_err(format!("error while cleaning up completed job: {e:#}"))
})?;
}
// Process scheduled deletions — insert into job_delete_schedule
if !_cleanup_module.flow_jobs_to_schedule_clean.is_empty() {
let w_id = &flow_job.workspace_id;
for entry in &_cleanup_module.flow_jobs_to_schedule_clean {
windmill_common::jobs::schedule_job_deletion(
db,
entry.id,
w_id,
entry.delete_after_secs,
)
.await
.map_err(|e| {
Error::internal_err(format!(
"error scheduling job deletion for {}: {e:#}",
entry.id
))
})?;
}
}
// Flow-level delete_after_secs: apply to ALL child jobs + the parent flow job
let flow_value = flow_data.value();
let flow_delete_secs = windmill_common::jobs::resolve_delete_after_secs(
flow_value.delete_after_use,
flow_value.delete_after_secs,
);
if let Some(secs) = flow_delete_secs {
let w_id = &flow_job.workspace_id;
let mut all_ids: Vec<Uuid> =
sqlx::query_scalar!("SELECT id FROM v2_job WHERE root_job = $1", flow_job.id)
.fetch_all(db)
.await
.map_err(|e| {
Error::internal_err(format!(
"error fetching child jobs for flow-level deletion: {e:#}"
))
})?;
all_ids.push(flow_job.id);
if secs == 0 {
sqlx::query!(
"UPDATE v2_job SET args = '{}'::jsonb WHERE id = ANY($1)",
&all_ids,
)
.execute(db)
.await
.map_err(|e| {
Error::internal_err(format!(
"error while cleaning up completed job args: {e:#}"
))
})?;
sqlx::query!(
"UPDATE v2_job_completed SET result = '{}'::jsonb WHERE id = ANY($1)",
&all_ids,
)
.execute(db)
.await
.map_err(|e| {
Error::internal_err(format!(
"error while cleaning up completed job results: {e:#}"
))
})?;
sqlx::query!(
"UPDATE job_logs SET logs = '##DELETED##' WHERE job_id = ANY($1)",
&all_ids,
)
.execute(db)
.await
.map_err(|e| {
Error::internal_err(format!(
"error while cleaning up completed job logs: {e:#}"
))
})?;
} else {
for id in &all_ids {
windmill_common::jobs::schedule_job_deletion(db, *id, w_id, secs)
.await
.map_err(|e| {
Error::internal_err(format!(
"error scheduling job deletion for {id}: {e:#}"
))
})?;
}
}
}
}
if flow_job.is_canceled() {
@@ -3753,21 +3859,54 @@ async fn push_next_flow_job(
}
}
if payload_tag.delete_after_use {
let uuid_singleton_json = serde_json::to_value(&[uuid]).map_err(|e| {
error::Error::internal_err(format!("Unable to serialize uuid: {e:#}"))
})?;
sqlx::query!(
"UPDATE v2_job_status
SET flow_status = JSONB_SET(flow_status, ARRAY['cleanup_module', 'flow_jobs_to_clean'], COALESCE(flow_status->'cleanup_module'->'flow_jobs_to_clean', '[]'::jsonb) || $1)
WHERE id = $2",
uuid_singleton_json,
flow_innermost_root_job.unwrap_or(flow_job.id)
)
.execute(&mut *inner_tx)
.warn_after_seconds(3)
.await?;
{
use windmill_common::jobs::resolve_delete_after_secs;
let resolved = resolve_delete_after_secs(
payload_tag.delete_after_use.then_some(true),
payload_tag.delete_after_secs,
);
let root_id = flow_innermost_root_job.unwrap_or(flow_job.id);
match resolved {
Some(0) => {
// Immediate deletion: track in flow_jobs_to_clean (existing behavior)
let uuid_singleton_json = serde_json::to_value(&[uuid]).map_err(|e| {
error::Error::internal_err(format!("Unable to serialize uuid: {e:#}"))
})?;
sqlx::query!(
"UPDATE v2_job_status
SET flow_status = JSONB_SET(flow_status, ARRAY['cleanup_module', 'flow_jobs_to_clean'], COALESCE(flow_status->'cleanup_module'->'flow_jobs_to_clean', '[]'::jsonb) || $1)
WHERE id = $2",
uuid_singleton_json,
root_id
)
.execute(&mut *inner_tx)
.warn_after_seconds(3)
.await?;
}
Some(secs) => {
// Scheduled deletion: track in flow_jobs_to_schedule_clean
let entry = windmill_types::flow_status::FlowJobScheduledClean {
id: uuid,
delete_after_secs: secs,
};
let entry_json = serde_json::to_value(&[entry]).map_err(|e| {
error::Error::internal_err(format!(
"Unable to serialize scheduled clean entry: {e:#}"
))
})?;
sqlx::query!(
"UPDATE v2_job_status
SET flow_status = JSONB_SET(flow_status, ARRAY['cleanup_module', 'flow_jobs_to_schedule_clean'], COALESCE(flow_status->'cleanup_module'->'flow_jobs_to_schedule_clean', '[]'::jsonb) || $1)
WHERE id = $2",
entry_json,
root_id
)
.execute(&mut *inner_tx)
.warn_after_seconds(3)
.await?;
}
None => {}
}
}
tx = inner_tx;
@@ -4162,6 +4301,7 @@ pub struct JobPayloadWithTag {
pub payload: JobPayload,
pub tag: Option<String>,
pub delete_after_use: bool,
pub delete_after_secs: Option<i32>,
pub timeout: Option<i32>,
pub on_behalf_of: Option<OnBehalfOf>,
}
@@ -4263,6 +4403,7 @@ async fn compute_next_flow_transform(
payload: JobPayload::Identity,
tag: None,
delete_after_use: false,
delete_after_secs: None,
timeout: None,
on_behalf_of: None,
}),
@@ -4275,6 +4416,7 @@ async fn compute_next_flow_transform(
payload,
tag: None,
delete_after_use: false,
delete_after_secs: None,
timeout: None,
on_behalf_of: None,
}),
@@ -4282,6 +4424,7 @@ async fn compute_next_flow_transform(
))
};
let delete_after_use = module.delete_after_use.unwrap_or(false);
let delete_after_secs = module.delete_after_secs;
tracing::debug!(id = %flow_job.id, "computing next flow transform for {:?}", &module.value);
if is_skipped {
@@ -4291,8 +4434,14 @@ async fn compute_next_flow_transform(
match module.get_value()? {
FlowModuleValue::Identity => trivial_next_job(JobPayload::Identity),
FlowModuleValue::Flow { path, .. } => {
let payload =
flow_to_payload(path, delete_after_use, &flow_job.workspace_id, db).await?;
let payload = flow_to_payload(
path,
delete_after_use,
delete_after_secs,
&flow_job.workspace_id,
db,
)
.await?;
Ok(NextFlowTransform::Continue(
ContinuePayload::SingleJob(payload),
NextStatus::NextStep,
@@ -4306,6 +4455,7 @@ async fn compute_next_flow_transform(
payload,
tag: None,
delete_after_use,
delete_after_secs,
timeout: None,
on_behalf_of: None,
}),
@@ -4348,6 +4498,7 @@ async fn compute_next_flow_transform(
module,
tag,
delete_after_use,
delete_after_secs,
);
Ok(NextFlowTransform::Continue(
ContinuePayload::SingleJob(payload),
@@ -4375,6 +4526,7 @@ async fn compute_next_flow_transform(
},
tag: tag.clone(),
delete_after_use,
delete_after_secs,
timeout: None,
on_behalf_of: None,
};
@@ -4518,6 +4670,7 @@ async fn compute_next_flow_transform(
payload,
tag: None,
delete_after_use,
delete_after_secs,
timeout: None,
on_behalf_of: None,
})
@@ -4615,6 +4768,7 @@ async fn compute_next_flow_transform(
payload,
tag: None,
delete_after_use,
delete_after_secs,
timeout: None,
on_behalf_of: None,
}),
@@ -4650,6 +4804,7 @@ async fn compute_next_flow_transform(
payload,
tag: None,
delete_after_use,
delete_after_secs,
timeout: None,
on_behalf_of: None,
})
@@ -4728,6 +4883,7 @@ async fn compute_next_flow_transform(
payload,
tag: None,
delete_after_use,
delete_after_secs,
timeout: None,
on_behalf_of: None,
}),
@@ -4792,6 +4948,7 @@ async fn next_loop_iteration(
payload,
tag: None,
delete_after_use,
delete_after_secs: None,
timeout: None,
on_behalf_of: None,
}),
@@ -4986,9 +5143,17 @@ async fn payload_from_simple_module(
inner_path: String,
) -> Result<JobPayloadWithTag, Error> {
let delete_after_use = module.delete_after_use.unwrap_or(false);
let delete_after_secs = module.delete_after_secs;
Ok(match value {
FlowModuleValue::Flow { path, .. } => {
flow_to_payload(path, delete_after_use, &flow_job.workspace_id, db).await?
flow_to_payload(
path,
delete_after_use,
delete_after_secs,
&flow_job.workspace_id,
db,
)
.await?
}
FlowModuleValue::Script { path: script_path, hash: script_hash, tag_override, .. } => {
script_to_payload(
@@ -5019,6 +5184,7 @@ async fn payload_from_simple_module(
module,
tag,
delete_after_use,
delete_after_secs,
),
FlowModuleValue::FlowScript {
id, // flow_node(id).
@@ -5038,6 +5204,7 @@ async fn payload_from_simple_module(
},
tag,
delete_after_use,
delete_after_secs,
timeout: None, // timeout evaluation handled at higher level
on_behalf_of: None,
},
@@ -5054,6 +5221,7 @@ pub fn raw_script_to_payload(
module: &FlowModule,
tag: Option<String>,
delete_after_use: bool,
delete_after_secs: Option<i32>,
) -> JobPayloadWithTag {
JobPayloadWithTag {
payload: JobPayload::Code(RawCode {
@@ -5072,6 +5240,7 @@ pub fn raw_script_to_payload(
}),
tag,
delete_after_use,
delete_after_secs,
timeout: None, // timeout evaluation handled at higher level
on_behalf_of: None,
}
@@ -5080,6 +5249,7 @@ pub fn raw_script_to_payload(
async fn flow_to_payload(
path: String,
delete_after_use: bool,
delete_after_secs: Option<i32>,
w_id: &str,
db: &DB,
) -> Result<JobPayloadWithTag, Error> {
@@ -5097,7 +5267,14 @@ async fn flow_to_payload(
version,
labels: None,
};
Ok(JobPayloadWithTag { payload, tag, delete_after_use, timeout: None, on_behalf_of })
Ok(JobPayloadWithTag {
payload,
tag,
delete_after_use,
delete_after_secs,
timeout: None,
on_behalf_of,
})
}
pub async fn script_to_payload(
@@ -5114,75 +5291,80 @@ pub async fn script_to_payload(
} else {
tag_override
};
let (payload, tag, delete_after_use, script_timeout, on_behalf_of) = if script_hash.is_none() {
let (jp, tag, delete_after_use, script_timeout, on_behalf_of) = script_path_to_payload(
&script_path,
None,
db.clone(),
&flow_job.workspace_id,
Some(true),
)
.await?;
(
jp,
tag_override.to_owned().or(tag),
delete_after_use,
script_timeout,
on_behalf_of,
)
} else {
let hash = script_hash.unwrap();
let ScriptHashInfo {
tag,
cache_ttl,
language,
dedicated_worker,
priority,
delete_after_use,
timeout,
on_behalf_of_email,
created_by,
runnable_settings:
ScriptRunnableSettingsInline { concurrency_settings, debouncing_settings },
..
} = get_script_info_for_hash(None, db, &flow_job.workspace_id, hash.0)
.await?
.prefetch_cached(&db)
.await?;
let on_behalf_of = if let Some(email) = on_behalf_of_email {
Some(OnBehalfOf { email, permissioned_as: username_to_permissioned_as(&created_by) })
let (payload, tag, delete_after_use, delete_after_secs, script_timeout, on_behalf_of) =
if script_hash.is_none() {
let (jp, tag, delete_after_use, delete_after_secs, script_timeout, on_behalf_of) =
script_path_to_payload(
&script_path,
None,
db.clone(),
&flow_job.workspace_id,
Some(true),
)
.await?;
(
jp,
tag_override.to_owned().or(tag),
delete_after_use,
delete_after_secs,
script_timeout,
on_behalf_of,
)
} else {
None
};
(
// We only apply the preprocessor if it's explicitly set to true in the module,
// which can only happen if the the flow is a SingleStepFlow triggered by a trigger with retries or error handling.
// In that case, apply_preprocessor is still only set to true if the script has a preprocesor.
// We only check for script hash because SingleStepFlow triggers specifies the script hash
JobPayload::ScriptHash {
hash,
path: script_path,
concurrency_settings,
debouncing_settings,
cache_ttl: module.cache_ttl.map(|x| x as i32).ok_or(cache_ttl).ok(),
cache_ignore_s3_path: module.cache_ignore_s3_path,
let hash = script_hash.unwrap();
let ScriptHashInfo {
tag,
cache_ttl,
language,
dedicated_worker,
priority,
apply_preprocessor: apply_preprocessor.unwrap_or(false),
labels: None,
},
tag_override.to_owned().or(tag),
delete_after_use,
timeout,
on_behalf_of,
)
};
// the module value overrides the value set at the script level. Defaults to false if both are unset.
let final_delete_after_user =
delete_after_use,
delete_after_secs,
timeout,
on_behalf_of_email,
created_by,
runnable_settings:
ScriptRunnableSettingsInline { concurrency_settings, debouncing_settings },
..
} = get_script_info_for_hash(None, db, &flow_job.workspace_id, hash.0)
.await?
.prefetch_cached(&db)
.await?;
let on_behalf_of = if let Some(email) = on_behalf_of_email {
Some(OnBehalfOf {
email,
permissioned_as: username_to_permissioned_as(&created_by),
})
} else {
None
};
(
JobPayload::ScriptHash {
hash,
path: script_path,
concurrency_settings,
debouncing_settings,
cache_ttl: module.cache_ttl.map(|x| x as i32).ok_or(cache_ttl).ok(),
cache_ignore_s3_path: module.cache_ignore_s3_path,
language,
dedicated_worker,
priority,
apply_preprocessor: apply_preprocessor.unwrap_or(false),
labels: None,
},
tag_override.to_owned().or(tag),
delete_after_use,
delete_after_secs,
timeout,
on_behalf_of,
)
};
// Module-level delete_after_secs takes precedence over script-level, then fall back to booleans
let final_delete_after_use =
module.delete_after_use.unwrap_or(false) || delete_after_use.unwrap_or(false);
let final_delete_after_secs = module.delete_after_secs.or(delete_after_secs);
let flow_step_timeout = if module.timeout.is_some() {
None
@@ -5192,7 +5374,8 @@ pub async fn script_to_payload(
Ok(JobPayloadWithTag {
payload,
tag,
delete_after_use: final_delete_after_user,
delete_after_use: final_delete_after_use,
delete_after_secs: final_delete_after_secs,
timeout: flow_step_timeout,
on_behalf_of,
})
cli/src/guidance/skills.ts
+1 -1
View File
File diff suppressed because one or more lines are too long
frontend/src/lib/components/ScriptBuilder.svelte
+14 -7
View File
@@ -1644,7 +1644,7 @@
>
{/snippet}
</Section>
<Section label="Delete after use">
<Section label="Delete after completion">
{#snippet header()}
<Tooltip
documentationLink="https://www.windmill.dev/docs/script_editor/settings#delete-after-use"
@@ -1655,7 +1655,8 @@
<br />
<br />
The logs, arguments and results of the job will be completely deleted from Windmill
once it is complete and the result has been returned.
after the specified delay once it is complete and the result has been returned.
Set to 0 for immediate deletion.
<br />
<br />
The deletion is irreversible.
@@ -1670,18 +1671,24 @@
<Toggle
disabled={!$enterpriseLicense}
size="sm"
checked={Boolean(script.delete_after_use)}
checked={script.delete_after_secs != null}
on:change={() => {
if (script.delete_after_use) {
script.delete_after_use = undefined
if (script.delete_after_secs != null) {
script.delete_after_secs = undefined
} else {
script.delete_after_use = true
script.delete_after_secs = 0
}
}}
options={{
right: 'Delete logs, arguments and results after use'
right: 'Delete logs, arguments and results after completion'
}}
/>
{#if script.delete_after_secs != null}
<SecondsInput
bind:seconds={script.delete_after_secs}
disabled={!$enterpriseLicense}
/>
{/if}
</div>
</Section>
{#if !isCloudHosted()}
frontend/src/lib/components/flows/content/FlowModuleDeleteAfterUse.svelte
+16 -7
View File
@@ -2,6 +2,7 @@
import Toggle from '$lib/components/Toggle.svelte'
import Tooltip from '$lib/components/Tooltip.svelte'
import type { FlowModule } from '$lib/gen'
import { SecondsInput } from '$lib/components/common'
import Section from '$lib/components/Section.svelte'
@@ -11,18 +12,21 @@
}
let { flowModule = $bindable(), disabled = false }: Props = $props()
let enabled = $derived(flowModule.delete_after_secs != null)
</script>
<Section label="Delete after use">
<Section label="Delete after completion">
{#snippet header()}
<Tooltip>
The logs, arguments and results of this flow step will be completely deleted from Windmill
once the flow is complete. They might be temporarily visible in UI while the flow is running.
after the specified delay once the flow is complete. They might be temporarily visible in UI
while the flow is running.
<br />
This also applies to a flow step that has failed: the error will not be accessible.
<br />
<br />
The deletion is irreversible.
The deletion is irreversible. Set to 0 for immediate deletion.
{#if disabled}
<br />
<br />
@@ -34,16 +38,21 @@
<Toggle
{disabled}
size="sm"
checked={Boolean(flowModule.delete_after_use)}
checked={enabled}
on:change={() => {
if (flowModule.delete_after_use) {
flowModule.delete_after_use = undefined
if (enabled) {
flowModule.delete_after_secs = undefined
} else {
flowModule.delete_after_use = true
flowModule.delete_after_secs = 0
}
}}
options={{
right: 'Delete logs, arguments and results after the flow is complete'
}}
/>
{#if enabled}
<div class="mt-2">
<SecondsInput bind:seconds={flowModule.delete_after_secs} {disabled} size="sm" />
</div>
{/if}
</Section>
frontend/src/lib/components/flows/content/FlowSettings.svelte
+29
View File
@@ -125,6 +125,7 @@
}}
/>
</Label>
<!-- prettier-ignore -->
<LabelsInput bind:labels={(flowStore.val as any).labels} class="-mt-4" />
{#if !noEditor}
@@ -589,6 +590,34 @@
{/snippet}
</Toggle>
<Toggle
textClass="font-medium"
size="xs"
disabled={!$enterpriseLicense}
checked={flowStore.val.value.delete_after_secs != null}
on:change={() => {
if (flowStore.val.value.delete_after_secs != null) {
flowStore.val.value.delete_after_secs = undefined
} else {
flowStore.val.value.delete_after_secs = 0
}
}}
options={{
right: 'Delete all step results after completion',
rightTooltip: `When enabled, the logs, arguments and results of all flow steps will be deleted after the specified delay once the flow completes. Set to 0 for immediate deletion. The deletion is irreversible. ${!$enterpriseLicense ? 'This is a feature only available on enterprise edition.' : ''}`
}}
eeOnly={true}
/>
{#if flowStore.val.value.delete_after_secs != null}
<div class="ml-6 mt-1">
<SecondsInput
bind:seconds={flowStore.val.value.delete_after_secs}
disabled={!$enterpriseLicense}
size="sm"
/>
</div>
{/if}
<div>
<Toggle
textClass="font-medium"
openflow.openapi.yaml
+6 -3
View File
@@ -94,6 +94,9 @@ components:
description: Cache duration in seconds for flow results
cache_ignore_s3_path:
type: boolean
delete_after_secs:
type: integer
description: If set, delete the flow job's args, result and logs after this many seconds following job completion
flow_env:
type: object
description: "Environment variables available to all steps. Values can be strings, JSON values, or special references: '$var:path' (workspace variable) or '$res:path' (resource)."
@@ -303,9 +306,9 @@ components:
timeout:
description: Maximum execution time in seconds (static value or expression)
$ref: '#/components/schemas/InputTransform'
delete_after_use:
type: boolean
description: If true, this step's result is deleted after use to save memory
delete_after_secs:
type: integer
description: If set, delete the step's args, result and logs after this many seconds following job completion
summary:
type: string
description: Short description of what this step does
system_prompts/auto-generated/flow.md
+1 -1
View File
File diff suppressed because one or more lines are too long
system_prompts/auto-generated/prompts.ts
+1 -1
View File
File diff suppressed because one or more lines are too long
system_prompts/auto-generated/skills/write-flow/SKILL.md
+1 -1
View File
File diff suppressed because one or more lines are too long