diff --git a/backend/.sqlx/query-d48070ef889723f408ed6c1a97b5ab9f0e98340cd33c6e29824f6030f0089071.json b/backend/.sqlx/query-0234079d04372b82ec8f2990916433bd8771c21d22a1db02eb1e388a8bfb28ff.json similarity index 85% rename from backend/.sqlx/query-d48070ef889723f408ed6c1a97b5ab9f0e98340cd33c6e29824f6030f0089071.json rename to backend/.sqlx/query-0234079d04372b82ec8f2990916433bd8771c21d22a1db02eb1e388a8bfb28ff.json index 53974ad40e..3f2301cd28 100644 --- a/backend/.sqlx/query-d48070ef889723f408ed6c1a97b5ab9f0e98340cd33c6e29824f6030f0089071.json +++ b/backend/.sqlx/query-0234079d04372b82ec8f2990916433bd8771c21d22a1db02eb1e388a8bfb28ff.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "select tag, concurrent_limit, concurrency_time_window_s, cache_ttl, language as \"language: ScriptLang\", dedicated_worker, priority from script where hash = $1 AND workspace_id = $2", + "query": "select tag, concurrent_limit, concurrency_time_window_s, cache_ttl, language as \"language: ScriptLang\", dedicated_worker, priority, delete_after_use from script where hash = $1 AND workspace_id = $2", "describe": { "columns": [ { @@ -58,6 +58,11 @@ "ordinal": 6, "name": "priority", "type_info": "Int2" + }, + { + "ordinal": 7, + "name": "delete_after_use", + "type_info": "Bool" } ], "parameters": { @@ -73,8 +78,9 @@ true, false, true, + true, true ] }, - "hash": "d48070ef889723f408ed6c1a97b5ab9f0e98340cd33c6e29824f6030f0089071" + "hash": "0234079d04372b82ec8f2990916433bd8771c21d22a1db02eb1e388a8bfb28ff" } diff --git a/backend/.sqlx/query-0aff8b0f2cddc0dd696759f48368c7315ff39060be5430c59582e1e05a4b3d90.json b/backend/.sqlx/query-0aff8b0f2cddc0dd696759f48368c7315ff39060be5430c59582e1e05a4b3d90.json new file mode 100644 index 0000000000..4f6b896ad4 --- /dev/null +++ b/backend/.sqlx/query-0aff8b0f2cddc0dd696759f48368c7315ff39060be5430c59582e1e05a4b3d90.json @@ -0,0 +1,16 @@ +{ + "db_name": "PostgreSQL", + "query": "UPDATE queue\n SET flow_status = JSONB_SET(\n JSONB_SET(flow_status, ARRAY['failure_module'], $1), ARRAY['step'], $2)\n WHERE id = $3", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Jsonb", + "Jsonb", + "Uuid" + ] + }, + "nullable": [] + }, + "hash": "0aff8b0f2cddc0dd696759f48368c7315ff39060be5430c59582e1e05a4b3d90" +} diff --git a/backend/.sqlx/query-0c0b5d5d1e6ab2fed7532f94b50be3210e3845b61551691bbef81c2b6fb01121.json b/backend/.sqlx/query-0c0b5d5d1e6ab2fed7532f94b50be3210e3845b61551691bbef81c2b6fb01121.json new file mode 100644 index 0000000000..9b37241475 --- /dev/null +++ b/backend/.sqlx/query-0c0b5d5d1e6ab2fed7532f94b50be3210e3845b61551691bbef81c2b6fb01121.json @@ -0,0 +1,15 @@ +{ + "db_name": "PostgreSQL", + "query": "UPDATE queue\n SET flow_status = JSONB_SET(flow_status, ARRAY['failure_module'], $1)\n WHERE id = $2", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Jsonb", + "Uuid" + ] + }, + "nullable": [] + }, + "hash": "0c0b5d5d1e6ab2fed7532f94b50be3210e3845b61551691bbef81c2b6fb01121" +} diff --git a/backend/.sqlx/query-26106be4d94c159cc8d9eb37a3b94927c5d3d43c18cdb5e98e979e761ed6ed0e.json b/backend/.sqlx/query-26106be4d94c159cc8d9eb37a3b94927c5d3d43c18cdb5e98e979e761ed6ed0e.json new file mode 100644 index 0000000000..fa5e4bff4c --- /dev/null +++ b/backend/.sqlx/query-26106be4d94c159cc8d9eb37a3b94927c5d3d43c18cdb5e98e979e761ed6ed0e.json @@ -0,0 +1,15 @@ +{ + "db_name": "PostgreSQL", + "query": "UPDATE queue\n SET flow_status = JSONB_SET(flow_status, ARRAY['step'], $1)\n WHERE id = $2", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Jsonb", + "Uuid" + ] + }, + "nullable": [] + }, + "hash": "26106be4d94c159cc8d9eb37a3b94927c5d3d43c18cdb5e98e979e761ed6ed0e" +} diff --git a/backend/.sqlx/query-2a3ebe1b0eae5b2164894321e138cc4dc0293788aeb98d05d95d18dfc708d6a6.json b/backend/.sqlx/query-2a3ebe1b0eae5b2164894321e138cc4dc0293788aeb98d05d95d18dfc708d6a6.json deleted file mode 100644 index ccb49b632a..0000000000 --- a/backend/.sqlx/query-2a3ebe1b0eae5b2164894321e138cc4dc0293788aeb98d05d95d18dfc708d6a6.json +++ /dev/null @@ -1,16 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "\n UPDATE queue\n SET leaf_jobs = JSONB_SET(coalesce(leaf_jobs, '{}'::jsonb), ARRAY[$1::TEXT], $2)\n WHERE COALESCE((SELECT root_job FROM queue WHERE id = $3), $3) = id\n ", - "describe": { - "columns": [], - "parameters": { - "Left": [ - "Text", - "Jsonb", - "Uuid" - ] - }, - "nullable": [] - }, - "hash": "2a3ebe1b0eae5b2164894321e138cc4dc0293788aeb98d05d95d18dfc708d6a6" -} diff --git a/backend/.sqlx/query-322038aae939d738f47b5bddaec2a6342e25e815dcd8b9fd620ddb81063c5d76.json b/backend/.sqlx/query-322038aae939d738f47b5bddaec2a6342e25e815dcd8b9fd620ddb81063c5d76.json new file mode 100644 index 0000000000..0e672f46a2 --- /dev/null +++ b/backend/.sqlx/query-322038aae939d738f47b5bddaec2a6342e25e815dcd8b9fd620ddb81063c5d76.json @@ -0,0 +1,22 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT flow_status->'cleanup_module' as cleanup_module\n FROM queue\n WHERE id = $1", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "cleanup_module", + "type_info": "Jsonb" + } + ], + "parameters": { + "Left": [ + "Uuid" + ] + }, + "nullable": [ + null + ] + }, + "hash": "322038aae939d738f47b5bddaec2a6342e25e815dcd8b9fd620ddb81063c5d76" +} diff --git a/backend/.sqlx/query-3c83e6adf21b7a8572529f9175cf6d2d30ea8c921ae2bf751b6825a3e453a012.json b/backend/.sqlx/query-3c83e6adf21b7a8572529f9175cf6d2d30ea8c921ae2bf751b6825a3e453a012.json deleted file mode 100644 index b2da6e359a..0000000000 --- a/backend/.sqlx/query-3c83e6adf21b7a8572529f9175cf6d2d30ea8c921ae2bf751b6825a3e453a012.json +++ /dev/null @@ -1,15 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "\n UPDATE queue\n SET suspend = $1, suspend_until = now() + interval '14 day', running = true\n WHERE id = $2\n ", - "describe": { - "columns": [], - "parameters": { - "Left": [ - "Int4", - "Uuid" - ] - }, - "nullable": [] - }, - "hash": "3c83e6adf21b7a8572529f9175cf6d2d30ea8c921ae2bf751b6825a3e453a012" -} diff --git a/backend/.sqlx/query-42543450a5c1b988b258fb1e0c00cd0a53a0395bc60e35817ff149c4f27a19fb.json b/backend/.sqlx/query-42543450a5c1b988b258fb1e0c00cd0a53a0395bc60e35817ff149c4f27a19fb.json new file mode 100644 index 0000000000..89287ed48d --- /dev/null +++ b/backend/.sqlx/query-42543450a5c1b988b258fb1e0c00cd0a53a0395bc60e35817ff149c4f27a19fb.json @@ -0,0 +1,14 @@ +{ + "db_name": "PostgreSQL", + "query": "UPDATE completed_job\n SET logs = '##DELETED##', args = '{}'::jsonb, result = '{}'::jsonb\n WHERE id = ANY($1)", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "UuidArray" + ] + }, + "nullable": [] + }, + "hash": "42543450a5c1b988b258fb1e0c00cd0a53a0395bc60e35817ff149c4f27a19fb" +} diff --git a/backend/.sqlx/query-47a74719faba8a6377ba157d25929a6f1e74afdffd642d5a1c7db57e745e0c01.json b/backend/.sqlx/query-47a74719faba8a6377ba157d25929a6f1e74afdffd642d5a1c7db57e745e0c01.json deleted file mode 100644 index 7f2ad2cc69..0000000000 --- a/backend/.sqlx/query-47a74719faba8a6377ba157d25929a6f1e74afdffd642d5a1c7db57e745e0c01.json +++ /dev/null @@ -1,16 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "\n UPDATE queue\n SET flow_status = JSONB_SET(flow_status, ARRAY['modules', $1::TEXT], $2)\n WHERE id = $3\n ", - "describe": { - "columns": [], - "parameters": { - "Left": [ - "Text", - "Jsonb", - "Uuid" - ] - }, - "nullable": [] - }, - "hash": "47a74719faba8a6377ba157d25929a6f1e74afdffd642d5a1c7db57e745e0c01" -} diff --git a/backend/.sqlx/query-4de3ba6930b91c639f9f3939dab7b3ab35aa792484e4ccb8084c336a2ec60f45.json b/backend/.sqlx/query-4de3ba6930b91c639f9f3939dab7b3ab35aa792484e4ccb8084c336a2ec60f45.json deleted file mode 100644 index 50e9827aae..0000000000 --- a/backend/.sqlx/query-4de3ba6930b91c639f9f3939dab7b3ab35aa792484e4ccb8084c336a2ec60f45.json +++ /dev/null @@ -1,15 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "\n UPDATE queue\n SET flow_status = JSONB_SET(flow_status, ARRAY['step'], $1)\n WHERE id = $2\n ", - "describe": { - "columns": [], - "parameters": { - "Left": [ - "Jsonb", - "Uuid" - ] - }, - "nullable": [] - }, - "hash": "4de3ba6930b91c639f9f3939dab7b3ab35aa792484e4ccb8084c336a2ec60f45" -} diff --git a/backend/.sqlx/query-53ec897ac019abf3fc4499f593adadf85af465d66c1a23b0549d176edbabad79.json b/backend/.sqlx/query-53ec897ac019abf3fc4499f593adadf85af465d66c1a23b0549d176edbabad79.json deleted file mode 100644 index b8b45177cb..0000000000 --- a/backend/.sqlx/query-53ec897ac019abf3fc4499f593adadf85af465d66c1a23b0549d176edbabad79.json +++ /dev/null @@ -1,17 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "\n UPDATE queue\n SET flow_status = JSONB_SET(\n JSONB_SET(flow_status, ARRAY['modules', $1::TEXT], $2),\n ARRAY['step'], $3)\n WHERE id = $4\n ", - "describe": { - "columns": [], - "parameters": { - "Left": [ - "Text", - "Jsonb", - "Jsonb", - "Uuid" - ] - }, - "nullable": [] - }, - "hash": "53ec897ac019abf3fc4499f593adadf85af465d66c1a23b0549d176edbabad79" -} diff --git a/backend/.sqlx/query-53ff0e14c35a3e84585a699e55093546db5c5f0ad0c5f92f34aaf2cdd125d130.json b/backend/.sqlx/query-53ff0e14c35a3e84585a699e55093546db5c5f0ad0c5f92f34aaf2cdd125d130.json new file mode 100644 index 0000000000..ba39c47062 --- /dev/null +++ b/backend/.sqlx/query-53ff0e14c35a3e84585a699e55093546db5c5f0ad0c5f92f34aaf2cdd125d130.json @@ -0,0 +1,15 @@ +{ + "db_name": "PostgreSQL", + "query": "UPDATE queue\n SET suspend = $1, suspend_until = now() + interval '14 day', running = true\n WHERE id = $2", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Int4", + "Uuid" + ] + }, + "nullable": [] + }, + "hash": "53ff0e14c35a3e84585a699e55093546db5c5f0ad0c5f92f34aaf2cdd125d130" +} diff --git a/backend/.sqlx/query-61656548991bf6d44c839373cb3e29d3ca1170a2ac0d3dce0b5df0e8677a4874.json b/backend/.sqlx/query-61656548991bf6d44c839373cb3e29d3ca1170a2ac0d3dce0b5df0e8677a4874.json new file mode 100644 index 0000000000..1cb035f456 --- /dev/null +++ b/backend/.sqlx/query-61656548991bf6d44c839373cb3e29d3ca1170a2ac0d3dce0b5df0e8677a4874.json @@ -0,0 +1,16 @@ +{ + "db_name": "PostgreSQL", + "query": "UPDATE queue\n SET flow_status = JSONB_SET(flow_status, ARRAY['modules', $1::TEXT], $2)\n WHERE id = $3", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Text", + "Jsonb", + "Uuid" + ] + }, + "nullable": [] + }, + "hash": "61656548991bf6d44c839373cb3e29d3ca1170a2ac0d3dce0b5df0e8677a4874" +} diff --git a/backend/.sqlx/query-cc916f88edb1ac791cdfb24c37691db9048e7d921a628a1ff5b71fead1de5e3a.json b/backend/.sqlx/query-7db59e88a10150ab6a3e81f164343e6bb2be6149fb690e63b8a0c376f39e9924.json similarity index 86% rename from backend/.sqlx/query-cc916f88edb1ac791cdfb24c37691db9048e7d921a628a1ff5b71fead1de5e3a.json rename to backend/.sqlx/query-7db59e88a10150ab6a3e81f164343e6bb2be6149fb690e63b8a0c376f39e9924.json index 779c29a8d4..28167accf8 100644 --- a/backend/.sqlx/query-cc916f88edb1ac791cdfb24c37691db9048e7d921a628a1ff5b71fead1de5e3a.json +++ b/backend/.sqlx/query-7db59e88a10150ab6a3e81f164343e6bb2be6149fb690e63b8a0c376f39e9924.json @@ -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) 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)", + "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) 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)", "describe": { "columns": [], "parameters": { @@ -62,10 +62,11 @@ "Bool", "Bool", "Int2", + "Bool", "Bool" ] }, "nullable": [] }, - "hash": "cc916f88edb1ac791cdfb24c37691db9048e7d921a628a1ff5b71fead1de5e3a" + "hash": "7db59e88a10150ab6a3e81f164343e6bb2be6149fb690e63b8a0c376f39e9924" } diff --git a/backend/.sqlx/query-8929150cb9262623eb12c908cb96c4d7e8cac594900ab0956c098e1cfa75e2f0.json b/backend/.sqlx/query-8929150cb9262623eb12c908cb96c4d7e8cac594900ab0956c098e1cfa75e2f0.json new file mode 100644 index 0000000000..9e66b2ae2c --- /dev/null +++ b/backend/.sqlx/query-8929150cb9262623eb12c908cb96c4d7e8cac594900ab0956c098e1cfa75e2f0.json @@ -0,0 +1,16 @@ +{ + "db_name": "PostgreSQL", + "query": "UPDATE queue\n SET leaf_jobs = JSONB_SET(coalesce(leaf_jobs, '{}'::jsonb), ARRAY[$1::TEXT], $2)\n WHERE COALESCE((SELECT root_job FROM queue WHERE id = $3), $3) = id", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Text", + "Jsonb", + "Uuid" + ] + }, + "nullable": [] + }, + "hash": "8929150cb9262623eb12c908cb96c4d7e8cac594900ab0956c098e1cfa75e2f0" +} diff --git a/backend/.sqlx/query-8a44ca0cfe1e154138cbbf5ebddebbba2e1149d2e329d1d89eecc1c008c93a31.json b/backend/.sqlx/query-8a44ca0cfe1e154138cbbf5ebddebbba2e1149d2e329d1d89eecc1c008c93a31.json new file mode 100644 index 0000000000..fdfaa0c029 --- /dev/null +++ b/backend/.sqlx/query-8a44ca0cfe1e154138cbbf5ebddebbba2e1149d2e329d1d89eecc1c008c93a31.json @@ -0,0 +1,14 @@ +{ + "db_name": "PostgreSQL", + "query": "UPDATE completed_job\n SET logs = '##DELETED##', args = '{}'::jsonb, result = '{}'::jsonb\n WHERE id = $1", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Uuid" + ] + }, + "nullable": [] + }, + "hash": "8a44ca0cfe1e154138cbbf5ebddebbba2e1149d2e329d1d89eecc1c008c93a31" +} diff --git a/backend/.sqlx/query-9681c348429f5e647ab55a4a54933faa64018f8539a143641f10d14ddd3a09c4.json b/backend/.sqlx/query-9681c348429f5e647ab55a4a54933faa64018f8539a143641f10d14ddd3a09c4.json deleted file mode 100644 index 51e344913e..0000000000 --- a/backend/.sqlx/query-9681c348429f5e647ab55a4a54933faa64018f8539a143641f10d14ddd3a09c4.json +++ /dev/null @@ -1,23 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "\n UPDATE queue\n SET flow_status = JSONB_SET(flow_status, ARRAY['modules', $1::TEXT, 'branchall', 'branch'], ((flow_status->'modules'->$1::int->'branchall'->>'branch')::int + 1)::text::jsonb)\n WHERE id = $2\n RETURNING (flow_status->'modules'->$1::int->'branchall'->>'branch')::int\n ", - "describe": { - "columns": [ - { - "ordinal": 0, - "name": "int4", - "type_info": "Int4" - } - ], - "parameters": { - "Left": [ - "Int4", - "Uuid" - ] - }, - "nullable": [ - null - ] - }, - "hash": "9681c348429f5e647ab55a4a54933faa64018f8539a143641f10d14ddd3a09c4" -} diff --git a/backend/.sqlx/query-a6145b0482c9e5da245059a80b1563cad20318fd2dd8aef33f9ca97de1826b8b.json b/backend/.sqlx/query-a6145b0482c9e5da245059a80b1563cad20318fd2dd8aef33f9ca97de1826b8b.json deleted file mode 100644 index 72d85a2a9a..0000000000 --- a/backend/.sqlx/query-a6145b0482c9e5da245059a80b1563cad20318fd2dd8aef33f9ca97de1826b8b.json +++ /dev/null @@ -1,15 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "\n UPDATE queue\n SET flow_status = JSONB_SET(flow_status, ARRAY['failure_module'], $1)\n WHERE id = $2\n ", - "describe": { - "columns": [], - "parameters": { - "Left": [ - "Jsonb", - "Uuid" - ] - }, - "nullable": [] - }, - "hash": "a6145b0482c9e5da245059a80b1563cad20318fd2dd8aef33f9ca97de1826b8b" -} diff --git a/backend/.sqlx/query-a94dbd1f7aab20682548471c5cc06c7a496edac36e3de537b00b76ad93c7556c.json b/backend/.sqlx/query-a94dbd1f7aab20682548471c5cc06c7a496edac36e3de537b00b76ad93c7556c.json new file mode 100644 index 0000000000..5906375f53 --- /dev/null +++ b/backend/.sqlx/query-a94dbd1f7aab20682548471c5cc06c7a496edac36e3de537b00b76ad93c7556c.json @@ -0,0 +1,23 @@ +{ + "db_name": "PostgreSQL", + "query": "UPDATE queue\n SET flow_status = JSONB_SET(flow_status, ARRAY['modules', $1::TEXT, 'branchall', 'branch'], ((flow_status->'modules'->$1::int->'branchall'->>'branch')::int + 1)::text::jsonb)\n WHERE id = $2\n RETURNING (flow_status->'modules'->$1::int->'branchall'->>'branch')::int", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "int4", + "type_info": "Int4" + } + ], + "parameters": { + "Left": [ + "Int4", + "Uuid" + ] + }, + "nullable": [ + null + ] + }, + "hash": "a94dbd1f7aab20682548471c5cc06c7a496edac36e3de537b00b76ad93c7556c" +} diff --git a/backend/.sqlx/query-b64c7f3b483c74c51aecda65664c36d9686112182dbf35651e494da739b922f9.json b/backend/.sqlx/query-b64c7f3b483c74c51aecda65664c36d9686112182dbf35651e494da739b922f9.json new file mode 100644 index 0000000000..948e713ef6 --- /dev/null +++ b/backend/.sqlx/query-b64c7f3b483c74c51aecda65664c36d9686112182dbf35651e494da739b922f9.json @@ -0,0 +1,17 @@ +{ + "db_name": "PostgreSQL", + "query": "UPDATE queue\n SET flow_status = JSONB_SET(\n JSONB_SET(flow_status, ARRAY['modules', $1::TEXT], $2), ARRAY['step'], $3)\n WHERE id = $4", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Text", + "Jsonb", + "Jsonb", + "Uuid" + ] + }, + "nullable": [] + }, + "hash": "b64c7f3b483c74c51aecda65664c36d9686112182dbf35651e494da739b922f9" +} diff --git a/backend/.sqlx/query-c461bdf083d7aaacafb6c981556b218342626524ee0c92f083dbff2930b7de9e.json b/backend/.sqlx/query-c461bdf083d7aaacafb6c981556b218342626524ee0c92f083dbff2930b7de9e.json deleted file mode 100644 index d99f1a64d5..0000000000 --- a/backend/.sqlx/query-c461bdf083d7aaacafb6c981556b218342626524ee0c92f083dbff2930b7de9e.json +++ /dev/null @@ -1,16 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "\n UPDATE queue\n SET flow_status = JSONB_SET(\n JSONB_SET(flow_status, ARRAY['failure_module'], $1),\n ARRAY['step'], $2)\n WHERE id = $3\n ", - "describe": { - "columns": [], - "parameters": { - "Left": [ - "Jsonb", - "Jsonb", - "Uuid" - ] - }, - "nullable": [] - }, - "hash": "c461bdf083d7aaacafb6c981556b218342626524ee0c92f083dbff2930b7de9e" -} diff --git a/backend/.sqlx/query-d6c25421bb6513908697ebe74c159da0ef78b5252b6ac669a3c9e545d05c0d43.json b/backend/.sqlx/query-d6c25421bb6513908697ebe74c159da0ef78b5252b6ac669a3c9e545d05c0d43.json new file mode 100644 index 0000000000..7d7274d774 --- /dev/null +++ b/backend/.sqlx/query-d6c25421bb6513908697ebe74c159da0ef78b5252b6ac669a3c9e545d05c0d43.json @@ -0,0 +1,23 @@ +{ + "db_name": "PostgreSQL", + "query": "UPDATE queue\n SET flow_status = JSONB_SET(flow_status, ARRAY['modules', $1::TEXT, 'iterator', 'index'], ((flow_status->'modules'->$1::int->'iterator'->>'index')::int + 1)::text::jsonb)\n WHERE id = $2\n RETURNING (flow_status->'modules'->$1::int->'iterator'->>'index')::int", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "int4", + "type_info": "Int4" + } + ], + "parameters": { + "Left": [ + "Int4", + "Uuid" + ] + }, + "nullable": [ + null + ] + }, + "hash": "d6c25421bb6513908697ebe74c159da0ef78b5252b6ac669a3c9e545d05c0d43" +} diff --git a/backend/.sqlx/query-11ae5e1aade2bc9a78af2d956ece73766ae3330319619a68383b508a76f2d19c.json b/backend/.sqlx/query-e541faa52c530c787d45c5ee70930687587c139606056bdc06cd860abeb0805b.json similarity index 85% rename from backend/.sqlx/query-11ae5e1aade2bc9a78af2d956ece73766ae3330319619a68383b508a76f2d19c.json rename to backend/.sqlx/query-e541faa52c530c787d45c5ee70930687587c139606056bdc06cd860abeb0805b.json index 2cb08ba2d4..b3297c8172 100644 --- a/backend/.sqlx/query-11ae5e1aade2bc9a78af2d956ece73766ae3330319619a68383b508a76f2d19c.json +++ b/backend/.sqlx/query-e541faa52c530c787d45c5ee70930687587c139606056bdc06cd860abeb0805b.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "select path, tag, concurrent_limit, concurrency_time_window_s, cache_ttl, language as \"language: ScriptLang\", dedicated_worker, priority from script where hash = $1 AND workspace_id = $2", + "query": "select path, tag, concurrent_limit, concurrency_time_window_s, cache_ttl, language as \"language: ScriptLang\", dedicated_worker, priority, delete_after_use from script where hash = $1 AND workspace_id = $2", "describe": { "columns": [ { @@ -63,6 +63,11 @@ "ordinal": 7, "name": "priority", "type_info": "Int2" + }, + { + "ordinal": 8, + "name": "delete_after_use", + "type_info": "Bool" } ], "parameters": { @@ -79,8 +84,9 @@ true, false, true, + true, true ] }, - "hash": "11ae5e1aade2bc9a78af2d956ece73766ae3330319619a68383b508a76f2d19c" + "hash": "e541faa52c530c787d45c5ee70930687587c139606056bdc06cd860abeb0805b" } diff --git a/backend/.sqlx/query-ed400dbf4a1fa597bc984240cfe9069369a8a265b127116187d78c0ba93ae076.json b/backend/.sqlx/query-e58cf2e3deb9aa1e9f37313a33e5d44fcfdb40a4764e1dc34896bd85b3d007f9.json similarity index 53% rename from backend/.sqlx/query-ed400dbf4a1fa597bc984240cfe9069369a8a265b127116187d78c0ba93ae076.json rename to backend/.sqlx/query-e58cf2e3deb9aa1e9f37313a33e5d44fcfdb40a4764e1dc34896bd85b3d007f9.json index 65733e1555..ab1bd399ea 100644 --- a/backend/.sqlx/query-ed400dbf4a1fa597bc984240cfe9069369a8a265b127116187d78c0ba93ae076.json +++ b/backend/.sqlx/query-e58cf2e3deb9aa1e9f37313a33e5d44fcfdb40a4764e1dc34896bd85b3d007f9.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "\n SELECT success\n FROM completed_job\n WHERE id = ANY($1)\n ", + "query": "SELECT success FROM completed_job WHERE id = ANY($1)", "describe": { "columns": [ { @@ -18,5 +18,5 @@ false ] }, - "hash": "ed400dbf4a1fa597bc984240cfe9069369a8a265b127116187d78c0ba93ae076" + "hash": "e58cf2e3deb9aa1e9f37313a33e5d44fcfdb40a4764e1dc34896bd85b3d007f9" } diff --git a/backend/.sqlx/query-e5a130478c8a581412f786b430c2c166c489faccbd0d9b448f8e7ad57559e605.json b/backend/.sqlx/query-e5a130478c8a581412f786b430c2c166c489faccbd0d9b448f8e7ad57559e605.json deleted file mode 100644 index c180575ff5..0000000000 --- a/backend/.sqlx/query-e5a130478c8a581412f786b430c2c166c489faccbd0d9b448f8e7ad57559e605.json +++ /dev/null @@ -1,23 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "\n UPDATE queue\n SET flow_status = JSONB_SET(flow_status, ARRAY['modules', $1::TEXT, 'iterator', 'index'], ((flow_status->'modules'->$1::int->'iterator'->>'index')::int + 1)::text::jsonb)\n WHERE id = $2\n RETURNING (flow_status->'modules'->$1::int->'iterator'->>'index')::int\n ", - "describe": { - "columns": [ - { - "ordinal": 0, - "name": "int4", - "type_info": "Int4" - } - ], - "parameters": { - "Left": [ - "Int4", - "Uuid" - ] - }, - "nullable": [ - null - ] - }, - "hash": "e5a130478c8a581412f786b430c2c166c489faccbd0d9b448f8e7ad57559e605" -} diff --git a/backend/.sqlx/query-b6074eb9369de071f79165bf506fbc44f1688035e1928b4c3649d875187b1e1d.json b/backend/.sqlx/query-f760667ffd6727443f023480627f62728bf9b6284d92e5fa9b97dc4e0b1c3a93.json similarity index 78% rename from backend/.sqlx/query-b6074eb9369de071f79165bf506fbc44f1688035e1928b4c3649d875187b1e1d.json rename to backend/.sqlx/query-f760667ffd6727443f023480627f62728bf9b6284d92e5fa9b97dc4e0b1c3a93.json index f5122a7f3c..f32ede17a7 100644 --- a/backend/.sqlx/query-b6074eb9369de071f79165bf506fbc44f1688035e1928b4c3649d875187b1e1d.json +++ b/backend/.sqlx/query-f760667ffd6727443f023480627f62728bf9b6284d92e5fa9b97dc4e0b1c3a93.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "select hash, tag, concurrent_limit, concurrency_time_window_s, cache_ttl, language as \"language: ScriptLang\", dedicated_worker, priority from script where path = $1 AND workspace_id = $2 AND\n created_at = (SELECT max(created_at) FROM script WHERE path = $1 AND workspace_id = $2 AND\n deleted = false AND lock IS not NULL AND lock_error_logs IS NULL)", + "query": "select hash, tag, concurrent_limit, concurrency_time_window_s, cache_ttl, language as \"language: ScriptLang\", dedicated_worker, priority, delete_after_use from script where path = $1 AND workspace_id = $2 AND\n created_at = (SELECT max(created_at) FROM script WHERE path = $1 AND workspace_id = $2 AND\n deleted = false AND lock IS not NULL AND lock_error_logs IS NULL)", "describe": { "columns": [ { @@ -63,6 +63,11 @@ "ordinal": 7, "name": "priority", "type_info": "Int2" + }, + { + "ordinal": 8, + "name": "delete_after_use", + "type_info": "Bool" } ], "parameters": { @@ -79,8 +84,9 @@ true, false, true, + true, true ] }, - "hash": "b6074eb9369de071f79165bf506fbc44f1688035e1928b4c3649d875187b1e1d" + "hash": "f760667ffd6727443f023480627f62728bf9b6284d92e5fa9b97dc4e0b1c3a93" } diff --git a/backend/tests/worker.rs b/backend/tests/worker.rs index 0266f50ef0..75354e188b 100644 --- a/backend/tests/worker.rs +++ b/backend/tests/worker.rs @@ -1084,6 +1084,7 @@ async fn test_deno_flow(db: Pool) { mock: None, timeout: None, priority: None, + delete_after_use: None }, FlowModule { id: "b".to_string(), @@ -1119,6 +1120,7 @@ async fn test_deno_flow(db: Pool) { mock: None, timeout: None, priority: None, + delete_after_use: None, }], }, stop_after_if: Default::default(), @@ -1130,6 +1132,7 @@ async fn test_deno_flow(db: Pool) { mock: None, timeout: None, priority: None, + delete_after_use: None, }, ], same_worker: false, @@ -1230,6 +1233,7 @@ async fn test_deno_flow_same_worker(db: Pool) { mock: None, timeout: None, priority: None, + delete_after_use: None, }, FlowModule { id: "b".to_string(), @@ -1276,6 +1280,7 @@ async fn test_deno_flow_same_worker(db: Pool) { mock: None, timeout: None, priority: None, + delete_after_use: None, }, FlowModule { id: "e".to_string(), @@ -1308,6 +1313,7 @@ async fn test_deno_flow_same_worker(db: Pool) { mock: None, timeout: None, priority: None, + delete_after_use: None }, ], }, @@ -1320,26 +1326,26 @@ async fn test_deno_flow_same_worker(db: Pool) { mock: None, timeout: None, priority: None, + delete_after_use: None }, FlowModule { id: "c".to_string(), value: FlowModuleValue::RawScript { input_transforms: [ - - ( - "loops".to_string(), - InputTransform::Javascript { expr: "results.b".to_string() }, - ), - ( - "path".to_string(), - InputTransform::Static { value: json!("outer.txt") }, - ), - ( - "path2".to_string(), - InputTransform::Static { value: json!("inner.txt") }, - ), - ] - .into(), + ( + "loops".to_string(), + InputTransform::Javascript { expr: "results.b".to_string() }, + ), + ( + "path".to_string(), + InputTransform::Static { value: json!("outer.txt") }, + ), + ( + "path2".to_string(), + InputTransform::Static { value: json!("inner.txt") }, + ), + ] + .into(), language: ScriptLang::Deno, content: r#"export async function main(path: string, loops: string[], path2: string) { return await Deno.readTextFile(`./shared/${path}`) + "," + loops + "," + await Deno.readTextFile(`./shared/${path2}`); @@ -1360,6 +1366,7 @@ async fn test_deno_flow_same_worker(db: Pool) { mock: None, timeout: None, priority: None, + delete_after_use: None, }, ], same_worker: true, diff --git a/backend/windmill-api/openapi-deref.yaml b/backend/windmill-api/openapi-deref.yaml index dee9535a01..263a682b8d 100644 --- a/backend/windmill-api/openapi-deref.yaml +++ b/backend/windmill-api/openapi-deref.yaml @@ -1,6 +1,6 @@ openapi: 3.0.3 info: - version: 1.219.1 + version: 1.221.0 title: Windmill API contact: name: Windmill Team @@ -3410,6 +3410,8 @@ paths: type: number concurrency_time_window_s: type: number + delete_after_use: + type: null required: &ref_176 - type - content diff --git a/backend/windmill-api/src/flows.rs b/backend/windmill-api/src/flows.rs index 325ca1df17..1daac2c31f 100644 --- a/backend/windmill-api/src/flows.rs +++ b/backend/windmill-api/src/flows.rs @@ -863,6 +863,7 @@ mod tests { mock: None, timeout: None, priority: None, + delete_after_use: None, }, FlowModule { id: "b".to_string(), @@ -888,6 +889,7 @@ mod tests { mock: None, timeout: None, priority: None, + delete_after_use: None, }, FlowModule { id: "c".to_string(), @@ -910,6 +912,7 @@ mod tests { mock: None, timeout: None, priority: None, + delete_after_use: None, }, ], failure_module: Some(FlowModule { @@ -931,6 +934,7 @@ mod tests { mock: None, timeout: None, priority: None, + delete_after_use: None, }), same_worker: false, concurrent_limit: None, diff --git a/backend/windmill-api/src/jobs.rs b/backend/windmill-api/src/jobs.rs index 1decad3f64..933121d47d 100644 --- a/backend/windmill-api/src/jobs.rs +++ b/backend/windmill-api/src/jobs.rs @@ -329,9 +329,10 @@ pub async fn get_path_tag_limits_cache_for_hash( ScriptLang, Option, Option, + Option, )> { let script = sqlx::query!( - "select path, tag, concurrent_limit, concurrency_time_window_s, cache_ttl, language as \"language: ScriptLang\", dedicated_worker, priority from script where hash = $1 AND workspace_id = $2", + "select path, tag, concurrent_limit, concurrency_time_window_s, cache_ttl, language as \"language: ScriptLang\", dedicated_worker, priority, delete_after_use from script where hash = $1 AND workspace_id = $2", hash, w_id ) @@ -351,6 +352,7 @@ pub async fn get_path_tag_limits_cache_for_hash( script.language, script.dedicated_worker, script.priority, + script.delete_after_use, )) } @@ -1812,7 +1814,8 @@ pub async fn run_job_by_path( check_scopes(&authed, || format!("run:script/{script_path}"))?; - let (job_payload, tag) = script_path_to_payload(script_path, &db, &w_id).await?; + let (job_payload, tag, _delete_after_use) = + script_path_to_payload(script_path, &db, &w_id).await?; let scheduled_for = run_query.get_scheduled_for(&db).await?; check_tag_available_for_workspace(&w_id, &tag).await?; @@ -1997,6 +2000,18 @@ async fn run_wait_result( } } +async fn delete_job_metadata_after_use(db: &DB, job_uuid: Uuid) -> Result<(), Error> { + sqlx::query!( + "UPDATE completed_job + SET logs = '##DELETED##', args = '{}'::jsonb, result = '{}'::jsonb + WHERE id = $1", + job_uuid, + ) + .execute(db) + .await?; + Ok(()) +} + pub async fn check_queue_too_long(db: &DB, queue_limit: Option) -> error::Result<()> { if let Some(limit) = queue_limit { let count = sqlx::query_scalar!( @@ -2073,7 +2088,8 @@ pub async fn run_wait_result_job_by_path_get( let script_path = script_path.to_path(); check_scopes(&authed, || format!("run:script/{script_path}"))?; - let (job_payload, tag) = script_path_to_payload(script_path, &db, &w_id).await?; + let (job_payload, tag, delete_after_use) = + script_path_to_payload(script_path, &db, &w_id).await?; check_tag_available_for_workspace(&w_id, &tag).await?; let tx = PushIsolationLevel::Isolated(user_db.clone(), authed.clone().into(), rsmq); @@ -2103,7 +2119,11 @@ pub async fn run_wait_result_job_by_path_get( .await?; tx.commit().await?; - run_wait_result(&db, uuid, Path((w_id, script_path)), None).await + let wait_result = run_wait_result(&db, uuid, Path((w_id, script_path)), None).await; + if delete_after_use.unwrap_or(false) { + delete_job_metadata_after_use(&db, uuid).await?; + } + return wait_result; } pub async fn run_wait_result_flow_by_path_get( @@ -2185,7 +2205,8 @@ async fn run_wait_result_script_by_path_internal( let script_path = script_path.to_path(); check_scopes(&authed, || format!("run:script/{script_path}"))?; - let (job_payload, tag) = script_path_to_payload(script_path, &db, &w_id).await?; + let (job_payload, tag, delete_after_use) = + script_path_to_payload(script_path, &db, &w_id).await?; check_tag_available_for_workspace(&w_id, &tag).await?; let tx = PushIsolationLevel::Isolated(user_db.clone(), authed.clone().into(), rsmq); @@ -2216,7 +2237,11 @@ async fn run_wait_result_script_by_path_internal( .await?; tx.commit().await?; - run_wait_result(&db, uuid, Path((w_id, script_path)), None).await + let wait_result = run_wait_result(&db, uuid, Path((w_id, script_path)), None).await; + if delete_after_use.unwrap_or(false) { + delete_job_metadata_after_use(&db, uuid).await?; + } + return wait_result; } pub async fn run_wait_result_script_by_hash( @@ -2243,6 +2268,7 @@ pub async fn run_wait_result_script_by_hash( language, dedicated_worker, priority, + delete_after_use, ) = get_path_tag_limits_cache_for_hash(&db, &w_id, hash).await?; check_scopes(&authed, || format!("run:script/{path}"))?; @@ -2284,7 +2310,11 @@ pub async fn run_wait_result_script_by_hash( .await?; tx.commit().await?; - run_wait_result(&db, uuid, Path((w_id, script_hash)), None).await + let wait_result = run_wait_result(&db, uuid, Path((w_id, script_hash)), None).await; + if delete_after_use.unwrap_or(false) { + delete_job_metadata_after_use(&db, uuid).await?; + } + return wait_result; } pub async fn run_wait_result_flow_by_path( @@ -2465,6 +2495,7 @@ async fn add_batch_jobs( language, dedicated_worker, _priority, + _delete_after_use, ) = get_latest_deployed_hash_for_path(&db, &w_id, &path).await?; ( Some(script_hash), @@ -2656,6 +2687,7 @@ pub async fn run_job_by_hash( language, dedicated_worker, priority, + _delete_after_use, // not taken into account in async endpoints ) = get_path_tag_limits_cache_for_hash(&db, &w_id, hash).await?; check_scopes(&authed, || format!("run:script/{path}"))?; diff --git a/backend/windmill-api/src/oauth2.rs b/backend/windmill-api/src/oauth2.rs index aa35177153..60d14c8ae4 100644 --- a/backend/windmill-api/src/oauth2.rs +++ b/backend/windmill-api/src/oauth2.rs @@ -873,6 +873,7 @@ async fn slack_command( language, dedicated_worker, priority, + _delete_after_use, ) = windmill_common::get_latest_deployed_hash_for_path( &db, &settings.workspace_id, diff --git a/backend/windmill-api/src/scripts.rs b/backend/windmill-api/src/scripts.rs index 3a4ffbd62a..1775777c7c 100644 --- a/backend/windmill-api/src/scripts.rs +++ b/backend/windmill-api/src/scripts.rs @@ -84,6 +84,8 @@ pub struct ScriptWDraft { pub priority: Option, #[serde(skip_serializing_if = "Option::is_none")] pub restart_unless_cancelled: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub delete_after_use: Option, } pub fn global_service() -> Router { @@ -472,8 +474,8 @@ async fn create_script( "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) \ - 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)", + dedicated_worker, ws_error_handler_muted, priority, restart_unless_cancelled, delete_after_use) \ + 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)", &w_id, &hash.0, ns.path, @@ -497,7 +499,8 @@ async fn create_script( ns.dedicated_worker, ns.ws_error_handler_muted.unwrap_or(false), ns.priority, - ns.restart_unless_cancelled + ns.restart_unless_cancelled, + ns.delete_after_use, ) .execute(&mut tx) .await?; @@ -716,7 +719,7 @@ async fn get_script_by_path_w_draft( let mut tx = user_db.begin(&authed).await?; let script_o = sqlx::query_as::<_, ScriptWDraft>( - "SELECT hash, script.path, summary, description, content, language, kind, tag, schema, draft_only, envs, concurrent_limit, concurrency_time_window_s, cache_ttl, ws_error_handler_muted, draft.value as draft, dedicated_worker, priority, restart_unless_cancelled FROM script LEFT JOIN draft ON + "SELECT hash, script.path, summary, description, content, language, kind, tag, schema, draft_only, envs, concurrent_limit, concurrency_time_window_s, cache_ttl, ws_error_handler_muted, draft.value as draft, dedicated_worker, priority, restart_unless_cancelled, delete_after_use FROM script LEFT JOIN draft ON script.path = draft.path AND script.workspace_id = draft.workspace_id AND draft.typ = 'script' WHERE script.path = $1 AND script.workspace_id = $2 \ AND script.created_at = (SELECT max(created_at) FROM script WHERE path = $1 AND \ diff --git a/backend/windmill-common/src/flow_status.rs b/backend/windmill-common/src/flow_status.rs index d57352ea3a..ea293c7c06 100644 --- a/backend/windmill-common/src/flow_status.rs +++ b/backend/windmill-common/src/flow_status.rs @@ -29,6 +29,7 @@ pub struct FlowStatus { pub step: i32, pub modules: Vec, pub failure_module: FlowStatusModuleWParent, + pub cleanup_module: FlowCleanupModule, #[serde(default)] #[serde(skip_serializing_if = "is_retry_default")] pub retry: RetryStatus, @@ -96,6 +97,15 @@ pub struct FlowStatusModuleWParent { pub module_status: FlowStatusModule, } +#[derive(Serialize, Deserialize, Debug, Clone)] +pub struct FlowCleanupModule { + #[serde(default)] + #[serde(skip_serializing_if = "Vec::is_empty")] + pub flow_jobs_to_clean: Vec, + #[serde(flatten)] + pub module_status: FlowStatusModule, +} + #[derive(Serialize, Deserialize, Debug, Clone)] #[serde(tag = "type")] pub enum FlowStatusModule { @@ -218,6 +228,16 @@ impl FlowStatus { .unwrap_or_else(|| "failure".to_string()), }, }, + cleanup_module: FlowCleanupModule { + flow_jobs_to_clean: vec![], + module_status: FlowStatusModule::WaitingForPriorSteps { + id: f + .failure_module + .as_ref() + .map(|x| x.id.clone()) + .unwrap_or_else(|| "cleanup".to_string()), + }, + }, retry: RetryStatus { fail_count: 0, failed_jobs: vec![] }, restarted_from: None, } diff --git a/backend/windmill-common/src/flows.rs b/backend/windmill-common/src/flows.rs index 5eb8070dfe..f5e9b3d71c 100644 --- a/backend/windmill-common/src/flows.rs +++ b/backend/windmill-common/src/flows.rs @@ -220,6 +220,8 @@ pub struct FlowModule { #[serde(skip_serializing_if = "Option::is_none")] // Priority at the flow step level pub priority: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub delete_after_use: Option, } impl FlowModule { @@ -369,6 +371,7 @@ pub fn add_virtual_items_if_necessary(modules: &mut Vec) { cache_ttl: None, timeout: None, priority: None, + delete_after_use: None, }); } } diff --git a/backend/windmill-common/src/jobs.rs b/backend/windmill-common/src/jobs.rs index 202b81122f..0e1b47fb68 100644 --- a/backend/windmill-common/src/jobs.rs +++ b/backend/windmill-common/src/jobs.rs @@ -335,9 +335,13 @@ pub async fn script_path_to_payload( script_path: &str, db: &DB, w_id: &str, -) -> error::Result<(JobPayload, Option)> { - let (job_payload, tag) = if script_path.starts_with("hub/") { - (JobPayload::ScriptHub { path: script_path.to_owned() }, None) +) -> error::Result<(JobPayload, Option, Option)> { + let (job_payload, tag, delete_after_use) = if script_path.starts_with("hub/") { + ( + JobPayload::ScriptHub { path: script_path.to_owned() }, + None, + None, + ) } else { let ( script_hash, @@ -348,6 +352,7 @@ pub async fn script_path_to_payload( language, dedicated_worker, priority, + delete_after_use, ) = get_latest_deployed_hash_for_path(db, w_id, script_path).await?; ( JobPayload::ScriptHash { @@ -361,9 +366,10 @@ pub async fn script_path_to_payload( priority, }, tag, + delete_after_use, ) }; - Ok((job_payload, tag)) + Ok((job_payload, tag, delete_after_use)) } pub async fn script_hash_to_tag_and_limits<'c>( @@ -378,9 +384,10 @@ pub async fn script_hash_to_tag_and_limits<'c>( ScriptLang, Option, Option, + Option, )> { let script = sqlx::query!( - "select tag, concurrent_limit, concurrency_time_window_s, cache_ttl, language as \"language: ScriptLang\", dedicated_worker, priority from script where hash = $1 AND workspace_id = $2", + "select tag, concurrent_limit, concurrency_time_window_s, cache_ttl, language as \"language: ScriptLang\", dedicated_worker, priority, delete_after_use from script where hash = $1 AND workspace_id = $2", script_hash.0, w_id ) @@ -399,6 +406,7 @@ pub async fn script_hash_to_tag_and_limits<'c>( script.language, script.dedicated_worker, script.priority, + script.delete_after_use, )) } @@ -407,7 +415,7 @@ pub async fn get_payload_tag_from_prefixed_path( db: &DB, w_id: &str, ) -> Result<(JobPayload, Option), Error> { - let (payload, tag) = if path.starts_with("script/") { + let (payload, tag, _) = if path.starts_with("script/") { script_path_to_payload(path.strip_prefix("script/").unwrap(), &db, w_id).await? } else if path.starts_with("flow/") { let path = path.strip_prefix("flow/").unwrap().to_string(); @@ -421,7 +429,7 @@ pub async fn get_payload_tag_from_prefixed_path( let (tag, dedicated_worker) = r .map(|x| (x.tag, x.dedicated_worker)) .unwrap_or_else(|| (None, None)); - (JobPayload::Flow { path, dedicated_worker }, tag) + (JobPayload::Flow { path, dedicated_worker }, tag, None) } else { return Err(Error::BadRequest(format!( "path must start with script/ or flow/ (got {})", diff --git a/backend/windmill-common/src/lib.rs b/backend/windmill-common/src/lib.rs index d0db6170aa..5eed4573eb 100644 --- a/backend/windmill-common/src/lib.rs +++ b/backend/windmill-common/src/lib.rs @@ -228,9 +228,10 @@ pub async fn get_latest_deployed_hash_for_path( ScriptLang, Option, Option, + Option, )> { let r_o = sqlx::query!( - "select hash, tag, concurrent_limit, concurrency_time_window_s, cache_ttl, language as \"language: ScriptLang\", dedicated_worker, priority from script where path = $1 AND workspace_id = $2 AND + "select hash, tag, concurrent_limit, concurrency_time_window_s, cache_ttl, language as \"language: ScriptLang\", dedicated_worker, priority, delete_after_use from script where path = $1 AND workspace_id = $2 AND created_at = (SELECT max(created_at) FROM script WHERE path = $1 AND workspace_id = $2 AND deleted = false AND lock IS not NULL AND lock_error_logs IS NULL)", script_path, @@ -250,6 +251,7 @@ pub async fn get_latest_deployed_hash_for_path( script.language, script.dedicated_worker, script.priority, + script.delete_after_use, )) } diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs index 7cbd652bdb..d1a94cbe00 100644 --- a/backend/windmill-queue/src/jobs.rs +++ b/backend/windmill-queue/src/jobs.rs @@ -45,8 +45,8 @@ use windmill_common::{ db::{Authed, UserDB}, error::{self, Error}, flow_status::{ - BranchAllStatus, FlowStatus, FlowStatusModule, FlowStatusModuleWParent, Iterator, - JobResult, RestartedFrom, RetryStatus, MAX_RETRY_ATTEMPTS, MAX_RETRY_INTERVAL, + BranchAllStatus, FlowCleanupModule, FlowStatus, FlowStatusModule, FlowStatusModuleWParent, + Iterator, JobResult, RestartedFrom, RetryStatus, MAX_RETRY_ATTEMPTS, MAX_RETRY_INTERVAL, }, flows::{add_virtual_items_if_necessary, FlowModuleValue, FlowValue}, jobs::{ @@ -2449,6 +2449,12 @@ pub async fn push<'c, T: Serialize + Send + Sync, R: rsmq_async::RsmqConnection id: "failure".to_string(), }, }, + cleanup_module: FlowCleanupModule { + module_status: FlowStatusModule::WaitingForPriorSteps { + id: "cleanup".to_string(), + }, + flow_jobs_to_clean: vec![], + }, // retry status is reset retry: RetryStatus { fail_count: 0, failed_jobs: vec![] }, // TODO: for now, flows with approval conditions aren't supported for restart @@ -2534,6 +2540,12 @@ pub async fn push<'c, T: Serialize + Send + Sync, R: rsmq_async::RsmqConnection id: "failure".to_string(), }, }, + cleanup_module: FlowCleanupModule { + module_status: FlowStatusModule::WaitingForPriorSteps { + id: "cleanup".to_string(), + }, + flow_jobs_to_clean: vec![], + }, // retry status is reset retry: RetryStatus { fail_count: 0, failed_jobs: vec![] }, // TODO: for now, flows with approval conditions aren't supported for restart diff --git a/backend/windmill-worker/src/worker_flow.rs b/backend/windmill-worker/src/worker_flow.rs index 3e777e5bd7..f5c340afdb 100644 --- a/backend/windmill-worker/src/worker_flow.rs +++ b/backend/windmill-worker/src/worker_flow.rs @@ -27,7 +27,7 @@ use tokio::sync::mpsc::Sender; use tracing::instrument; use uuid::Uuid; use windmill_common::flow_status::{ - ApprovalConditions, FlowStatusModuleWParent, Iterator, JobResult, + ApprovalConditions, FlowCleanupModule, FlowStatusModuleWParent, Iterator, JobResult, }; use windmill_common::flows::add_virtual_items_if_necessary; use windmill_common::jobs::{ @@ -233,20 +233,18 @@ pub async fn update_flow_status_after_job_completion_internal< (false, false) } else { let row = sqlx::query( - " - SELECT raw_flow->'modules'->$1::int->'stop_after_if'->>'expr' as stop_early_expr, - (raw_flow->'modules'->$1::int->'stop_after_if'->>'skip_if_stopped')::bool as skip_if_stopped, - args - FROM queue - WHERE id = $2 - ").bind( - old_status.step) - .bind( - flow - ) - .fetch_one(db) - .await - .map_err(|e| Error::InternalErr(format!("retrieval of stop_early_expr from state: {e}")))?; + "SELECT + raw_flow->'modules'->$1::int->'stop_after_if'->>'expr' as stop_early_expr, + (raw_flow->'modules'->$1::int->'stop_after_if'->>'skip_if_stopped')::bool as skip_if_stopped, + args + FROM queue + WHERE id = $2" + ) + .bind(old_status.step) + .bind(flow) + .fetch_one(db) + .await + .map_err(|e| Error::InternalErr(format!("retrieval of stop_early_expr from state: {e}")))?; let r = SkipIfStopped::from_row(&row)?; let stop_early = success @@ -294,34 +292,30 @@ pub async fn update_flow_status_after_job_completion_internal< let (nindex, len) = match (iterator, branchall) { (Some(Iterator { itered, .. }), _) => { let nindex = sqlx::query_scalar!( - " - UPDATE queue - SET flow_status = JSONB_SET(flow_status, ARRAY['modules', $1::TEXT, 'iterator', 'index'], ((flow_status->'modules'->$1::int->'iterator'->>'index')::int + 1)::text::jsonb) - WHERE id = $2 - RETURNING (flow_status->'modules'->$1::int->'iterator'->>'index')::int - ", - old_status.step, - flow - ) - .fetch_one(&mut tx) - .await? - .ok_or_else(|| Error::InternalErr(format!("requiring an index in InProgress")))?; + "UPDATE queue + SET flow_status = JSONB_SET(flow_status, ARRAY['modules', $1::TEXT, 'iterator', 'index'], ((flow_status->'modules'->$1::int->'iterator'->>'index')::int + 1)::text::jsonb) + WHERE id = $2 + RETURNING (flow_status->'modules'->$1::int->'iterator'->>'index')::int", + old_status.step, + flow + ) + .fetch_one(&mut tx) + .await? + .ok_or_else(|| Error::InternalErr(format!("requiring an index in InProgress")))?; (nindex, itered.len() as i32) } (_, Some(BranchAllStatus { len, .. })) => { let nindex = sqlx::query_scalar!( - " - UPDATE queue - SET flow_status = JSONB_SET(flow_status, ARRAY['modules', $1::TEXT, 'branchall', 'branch'], ((flow_status->'modules'->$1::int->'branchall'->>'branch')::int + 1)::text::jsonb) - WHERE id = $2 - RETURNING (flow_status->'modules'->$1::int->'branchall'->>'branch')::int - ", - old_status.step, - flow - ) - .fetch_one(&mut tx) - .await? - .ok_or_else(|| Error::InternalErr(format!("requiring an index in InProgress")))?; + "UPDATE queue + SET flow_status = JSONB_SET(flow_status, ARRAY['modules', $1::TEXT, 'branchall', 'branch'], ((flow_status->'modules'->$1::int->'branchall'->>'branch')::int + 1)::text::jsonb) + WHERE id = $2 + RETURNING (flow_status->'modules'->$1::int->'branchall'->>'branch')::int", + old_status.step, + flow + ) + .fetch_one(&mut tx) + .await? + .ok_or_else(|| Error::InternalErr(format!("requiring an index in InProgress")))?; (nindex, *len as i32) } _ => Err(Error::InternalErr(format!( @@ -331,12 +325,8 @@ pub async fn update_flow_status_after_job_completion_internal< if nindex == len { let new_status = if skip_loop_failures || sqlx::query_scalar!( - " - SELECT success - FROM completed_job - WHERE id = ANY($1) - ", - jobs.as_slice(), + "SELECT success FROM completed_job WHERE id = ANY($1)", + jobs.as_slice() ) .fetch_all(&mut tx) .await? @@ -353,7 +343,6 @@ pub async fn update_flow_status_after_job_completion_internal< } } else { success = false; - FlowStatusModule::Failure { id: module_status.id(), job: job_id_for_status.clone(), @@ -435,11 +424,9 @@ pub async fn update_flow_status_after_job_completion_internal< let step_counter = if inc_step_counter { sqlx::query!( - " - UPDATE queue - SET flow_status = JSONB_SET(flow_status, ARRAY['step'], $1) - WHERE id = $2 - ", + "UPDATE queue + SET flow_status = JSONB_SET(flow_status, ARRAY['step'], $1) + WHERE id = $2", json!(old_status.step + 1), flow ) @@ -458,18 +445,16 @@ pub async fn update_flow_status_after_job_completion_internal< if let Some(new_status) = new_status.as_ref() { if is_failure_step { let parent_module = sqlx::query_scalar!( - "SELECT flow_status->'failure_module'->>'parent_module' FROM queue WHERE id = $1", - flow - ) + "SELECT flow_status->'failure_module'->>'parent_module' FROM queue WHERE id = $1", + flow + ) .fetch_one(&mut tx) .await?; sqlx::query!( - " - UPDATE queue - SET flow_status = JSONB_SET(flow_status, ARRAY['failure_module'], $1) - WHERE id = $2 - ", + "UPDATE queue + SET flow_status = JSONB_SET(flow_status, ARRAY['failure_module'], $1) + WHERE id = $2", json!(FlowStatusModuleWParent { parent_module, module_status: new_status.clone() @@ -480,11 +465,9 @@ pub async fn update_flow_status_after_job_completion_internal< .await?; } else { sqlx::query!( - " - UPDATE queue - SET flow_status = JSONB_SET(flow_status, ARRAY['modules', $1::TEXT], $2) - WHERE id = $3 - ", + "UPDATE queue + SET flow_status = JSONB_SET(flow_status, ARRAY['modules', $1::TEXT], $2) + WHERE id = $3", old_status.step.to_string(), json!(new_status), flow @@ -494,11 +477,9 @@ pub async fn update_flow_status_after_job_completion_internal< if let Some(job_result) = new_status.job_result() { sqlx::query!( - " - UPDATE queue - SET leaf_jobs = JSONB_SET(coalesce(leaf_jobs, '{}'::jsonb), ARRAY[$1::TEXT], $2) - WHERE COALESCE((SELECT root_job FROM queue WHERE id = $3), $3) = id - ", + "UPDATE queue + SET leaf_jobs = JSONB_SET(coalesce(leaf_jobs, '{}'::jsonb), ARRAY[$1::TEXT], $2) + WHERE COALESCE((SELECT root_job FROM queue WHERE id = $3), $3) = id", new_status.id(), json!(job_result), flow @@ -519,12 +500,10 @@ pub async fn update_flow_status_after_job_completion_internal< if matches!(&new_status, Some(FlowStatusModule::Success { .. })) { sqlx::query( - " - UPDATE queue - SET flow_status = flow_status - 'retry' - WHERE id = $1 - RETURNING flow_status - ", + "UPDATE queue + SET flow_status = flow_status - 'retry' + WHERE id = $1 + RETURNING flow_status", ) .bind(flow) .execute(&mut tx) @@ -608,6 +587,26 @@ pub async fn update_flow_status_after_job_completion_internal< } else { "Flow job completed with error".to_string() }; + + #[cfg(feature = "enterprise")] + if flow_job.parent_job.is_none() { + // run the cleanup step only when the root job is complete + let cleanup_module = retrieve_cleanup_module(flow, db).await?; + if cleanup_module.flow_jobs_to_clean.len() > 0 { + tracing::debug!( + "Cleaning up jobs arguments, result and logs as they were marked as delete_after_use {:?}", + cleanup_module.flow_jobs_to_clean + ); + sqlx::query!( + "UPDATE completed_job + SET logs = '##DELETED##', args = '{}'::jsonb, result = '{}'::jsonb + WHERE id = ANY($1)", + &cleanup_module.flow_jobs_to_clean, + ) + .execute(db) + .await?; + } + } if flow_job.canceled { add_completed_job_error( db, @@ -740,12 +739,9 @@ async fn retrieve_flow_jobs_results( job_uuids: &Vec, ) -> error::Result> { let results = sqlx::query( - " - SELECT result, id + "SELECT result, id FROM completed_job - WHERE id = ANY($1) - AND workspace_id = $2 - ", + WHERE id = ANY($1) AND workspace_id = $2", ) .bind(job_uuids.as_slice()) .bind(w_id) @@ -793,11 +789,9 @@ async fn compute_skip_loop_failures_and_parallelism( db: &DB, ) -> Result<(Option, Option), Error> { sqlx::query_as( - " - SELECT (raw_flow->'modules'->$1->'value'->>'skip_failures')::bool, (raw_flow->'modules'->$1->'value'->>'parallelism')::int - FROM queue - WHERE id = $2 - ", + "SELECT (raw_flow->'modules'->$1->'value'->>'skip_failures')::bool, (raw_flow->'modules'->$1->'value'->>'parallelism')::int + FROM queue + WHERE id = $2", ) .bind(step) .bind(flow) @@ -814,11 +808,9 @@ async fn compute_skip_branchall_failure<'c>( db: &DB, ) -> Result, Error> { sqlx::query_as( - " - SELECT (raw_flow->'modules'->$1->'value'->'branches'->$2->>'skip_failure')::bool - FROM queue - WHERE id = $3 - ", + "SELECT (raw_flow->'modules'->$1->'value'->'branches'->$2->>'skip_failure')::bool + FROM queue + WHERE id = $3", ) .bind(step) .bind(branch as i32) @@ -834,11 +826,9 @@ async fn has_failure_module<'c>( tx: &mut sqlx::Transaction<'c, sqlx::Postgres>, ) -> Result { sqlx::query_scalar::<_, Option>( - " - SELECT raw_flow->'failure_module' != 'null'::jsonb - FROM queue - WHERE id = $1 - ", + "SELECT raw_flow->'failure_module' != 'null'::jsonb + FROM queue + WHERE id = $1", ) .bind(flow) .fetch_one(&mut **tx) @@ -847,6 +837,27 @@ async fn has_failure_module<'c>( .map(|v| v.unwrap_or(false)) } +async fn retrieve_cleanup_module<'c>(flow_uuid: Uuid, db: &DB) -> Result { + tracing::warn!("Retrieving cleanup module of flow {}", flow_uuid); + let raw_value = sqlx::query_scalar!( + "SELECT flow_status->'cleanup_module' as cleanup_module + FROM queue + WHERE id = $1", + flow_uuid, + ) + .fetch_one(db) + .await + .map_err(|e| Error::InternalErr(format!("error during retrieval of cleanup module: {e}")))?; + + raw_value + .clone() + .and_then(|rv| serde_json::from_value::(rv).ok()) + .ok_or(Error::InternalErr(format!( + "Unable to parse flow cleanup module {:?}", + raw_value + ))) +} + fn next_retry(retry: &Retry, status: &RetryStatus) -> Option<(u16, Duration)> { (status.fail_count <= MAX_RETRY_ATTEMPTS) .then(|| &retry) @@ -1100,7 +1111,7 @@ async fn push_next_flow_job let flow_job_args = flow_job.get_args(); - // if this is an empty module of if the module has aleady been completed, successfully, update the parent flow + // if this is an empty module of if the module has already been completed, successfully, update the parent flow if flow.modules.is_empty() || matches!(status_module, FlowStatusModule::Success { .. }) { let r; return update_flow_status_after_job_completion( @@ -1281,12 +1292,9 @@ async fn push_next_flow_job user_groups_required: user_groups_required, }; sqlx::query( - " - UPDATE queue - SET flow_status = - JSONB_SET(flow_status, ARRAY['approval_conditions'], $1) - WHERE id = $2 - ", + "UPDATE queue + SET flow_status = JSONB_SET(flow_status, ARRAY['approval_conditions'], $1) + WHERE id = $2", ) .bind(json!(approval_conditions)) .bind(flow_job.id) @@ -1296,12 +1304,9 @@ async fn push_next_flow_job if resume_messages.len() >= required_events as usize { sqlx::query( - " - UPDATE queue - SET flow_status = - JSONB_SET(flow_status, ARRAY['modules', $1::TEXT, 'approvers'], $2) - WHERE id = $3 - ", + "UPDATE queue + SET flow_status = JSONB_SET(flow_status, ARRAY['modules', $1::TEXT, 'approvers'], $2) + WHERE id = $3", ) .bind(status.step - 1) .bind(json!(resumes @@ -1321,11 +1326,9 @@ async fn push_next_flow_job // Remove the approval conditions from the flow status sqlx::query( - " - UPDATE queue - SET flow_status = flow_status - 'approval_conditions' - WHERE id = $1 - ", + "UPDATE queue + SET flow_status = flow_status - 'approval_conditions' + WHERE id = $1", ) .bind(flow_job.id) .execute(&mut *tx) @@ -1340,13 +1343,11 @@ async fn push_next_flow_job FlowStatusModule::WaitingForPriorSteps { .. } ) { sqlx::query( - " - UPDATE queue - SET flow_status = JSONB_SET(flow_status, ARRAY['modules', flow_status->>'step'::text], $1) - , suspend = $2 - , suspend_until = now() + $3 - WHERE id = $4 - ", + "UPDATE queue SET + flow_status = JSONB_SET(flow_status, ARRAY['modules', flow_status->>'step'::text], $1), + suspend = $2, + suspend_until = now() + $3 + WHERE id = $4", ) .bind(json!(FlowStatusModule::WaitingForEvents { id: status_module.id(), count: required_events, job: last })) .bind((required_events - resume_messages.len() as u16) as i32) @@ -1498,11 +1499,9 @@ async fn push_next_flow_job scheduled_for_o = Some(from_now(retry_in)); status.retry.failed_jobs.push(job.clone()); sqlx::query( - " - UPDATE queue - SET flow_status = JSONB_SET(flow_status, ARRAY['retry'], $1) - WHERE id = $2 - ", + "UPDATE queue + SET flow_status = JSONB_SET(flow_status, ARRAY['retry'], $1) + WHERE id = $2", ) .bind(json!(RetryStatus { fail_count, ..status.retry.clone() })) .bind(flow_job.id) @@ -1536,11 +1535,9 @@ async fn push_next_flow_job let retry = &module.retry.clone().unwrap_or_default(); if retry.has_attempts() { sqlx::query( - " - UPDATE queue - SET flow_status = JSONB_SET(flow_status, ARRAY['retry'], $1) - WHERE id = $2 - ", + "UPDATE queue + SET flow_status = JSONB_SET(flow_status, ARRAY['retry'], $1) + WHERE id = $2", ) .bind(json!(RetryStatus { fail_count: 0, failed_jobs: vec![] })) .bind(flow_job.id) @@ -1562,11 +1559,9 @@ async fn push_next_flow_job && status.retry.fail_count == 0 => { sqlx::query( - " - UPDATE queue - SET flow_status = JSONB_SET(flow_status, ARRAY['retry'], $1) - WHERE id = $2 - ", + "UPDATE queue + SET flow_status = JSONB_SET(flow_status, ARRAY['retry'], $1) + WHERE id = $2", ) .bind(json!(RetryStatus { fail_count: 0, failed_jobs: vec![] })) .bind(flow_job.id) @@ -1671,18 +1666,22 @@ async fn push_next_flow_job NextFlowTransform::Continue(job_payload, next_state) => (job_payload, next_state), NextFlowTransform::EmptyInnerFlows => { sqlx::query( - r#" - UPDATE queue - SET flow_status = JSONB_SET(flow_status, ARRAY['modules', $1::TEXT], $2) - WHERE id = $3 - "#, - ) - .bind(status.step) - .bind(json!(FlowStatusModule::Success { id: status_module.id(), job: Uuid::nil(), flow_jobs: Some(vec![]), branch_chosen: None, approvers: vec![] })) - .bind(flow_job.id) - .execute(db) - .await?; - // flow is reprocessed by the worker in a state where the module has completed succesfully. + "UPDATE queue + SET flow_status = JSONB_SET(flow_status, ARRAY['modules', $1::TEXT], $2) + WHERE id = $3", + ) + .bind(status.step) + .bind(json!(FlowStatusModule::Success { + id: status_module.id(), + job: Uuid::nil(), + flow_jobs: Some(vec![]), + branch_chosen: None, + approvers: vec![] + })) + .bind(flow_job.id) + .execute(db) + .await?; + // flow is reprocessed by the worker in a state where the module has completed successfully. // The next steps are pull -> handle flow -> push next flow job -> update flow status since module status is success same_worker_tx .send(flow_job.id) @@ -1846,11 +1845,9 @@ async fn push_next_flow_job if let FlowModuleValue::ForloopFlow { parallelism: Some(p), .. } = &module.value { if i as u16 >= *p { sqlx::query!( - " - UPDATE queue - SET suspend = $1, suspend_until = now() + interval '14 day', running = true - WHERE id = $2 - ", + "UPDATE queue + SET suspend = $1, suspend_until = now() + interval '14 day', running = true + WHERE id = $2", (i as u16 - p + 1) as i32, uuid, ) @@ -1858,6 +1855,22 @@ async fn push_next_flow_job .await?; } } + + if payload_tag.delete_after_use { + let uuid_singleton_json = serde_json::to_value(&[uuid]) + .map_err(|e| error::Error::InternalErr(format!("Unable to serialize uuid: {e}")))?; + + sqlx::query( + "UPDATE queue + 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", + ) + .bind(uuid_singleton_json) + .bind(root_job.unwrap_or(flow_job.id)) + .execute(&mut inner_tx) + .await?; + } + tx = inner_tx; uuids.push(uuid); } @@ -1931,13 +1944,10 @@ async fn push_next_flow_job if i >= flow.modules.len() { sqlx::query!( - " - UPDATE queue - SET flow_status = JSONB_SET( - JSONB_SET(flow_status, ARRAY['failure_module'], $1), - ARRAY['step'], $2) - WHERE id = $3 - ", + "UPDATE queue + SET flow_status = JSONB_SET( + JSONB_SET(flow_status, ARRAY['failure_module'], $1), ARRAY['step'], $2) + WHERE id = $3", json!(FlowStatusModuleWParent { parent_module: Some(current_id.clone()), module_status: new_status.clone() @@ -1945,23 +1955,20 @@ async fn push_next_flow_job json!(i), flow_job.id ) - .execute(db) + .execute(&mut tx) .await?; } else { sqlx::query!( - " - UPDATE queue - SET flow_status = JSONB_SET( - JSONB_SET(flow_status, ARRAY['modules', $1::TEXT], $2), - ARRAY['step'], $3) - WHERE id = $4 - ", + "UPDATE queue + SET flow_status = JSONB_SET( + JSONB_SET(flow_status, ARRAY['modules', $1::TEXT], $2), ARRAY['step'], $3) + WHERE id = $4", i as i32, json!(new_status), json!(i), flow_job.id ) - .execute(db) + .execute(&mut tx) .await?; }; @@ -2085,6 +2092,7 @@ enum NextStatus { struct JobPayloadWithTag { payload: JobPayload, tag: Option, + delete_after_use: bool, } enum ContinuePayload { SingleJob(JobPayloadWithTag), @@ -2137,20 +2145,26 @@ async fn compute_next_flow_transform( ContinuePayload::SingleJob(JobPayloadWithTag { payload: JobPayload::Identity, tag: None, + delete_after_use: false, }), NextStatus::NextStep, )); } let trivial_next_job = |payload| { Ok(NextFlowTransform::Continue( - ContinuePayload::SingleJob(JobPayloadWithTag { payload, tag: None }), + ContinuePayload::SingleJob(JobPayloadWithTag { + payload, + tag: None, + delete_after_use: false, + }), NextStatus::NextStep, )) }; + let delete_after_use = module.delete_after_use.unwrap_or(false); match &module.value { FlowModuleValue::Identity => trivial_next_job(JobPayload::Identity), FlowModuleValue::Flow { path, .. } => { - let payload = flow_to_payload(path); + let payload = flow_to_payload(path, &delete_after_use); Ok(NextFlowTransform::Continue( ContinuePayload::SingleJob(payload), NextStatus::NextStep, @@ -2185,6 +2199,7 @@ async fn compute_next_flow_transform( concurrency_time_window_s, module, tag, + &delete_after_use, ); Ok(NextFlowTransform::Continue( ContinuePayload::SingleJob(payload), @@ -2374,6 +2389,7 @@ async fn compute_next_flow_transform( restarted_from: None, }, tag: None, + delete_after_use: delete_after_use, }), NextStatus::NextLoopIteration { next: ns, @@ -2413,6 +2429,7 @@ async fn compute_next_flow_transform( restarted_from: None, }, tag: None, + delete_after_use: delete_after_use, } }; ContinuePayload::ForloopJobs { n: itered.len(), payload } @@ -2513,6 +2530,7 @@ async fn compute_next_flow_transform( restarted_from: None, }, tag: None, + delete_after_use: delete_after_use, }), NextStatus::BranchChosen(branch), )) @@ -2562,6 +2580,7 @@ async fn compute_next_flow_transform( restarted_from: None, }, tag: None, + delete_after_use: delete_after_use, } }) .collect(), @@ -2627,6 +2646,7 @@ async fn compute_next_flow_transform( restarted_from: None, }, tag: None, + delete_after_use: delete_after_use, }), NextStatus::NextBranchStep(NextBranch { status: branch_status, flow_jobs }), )) @@ -2641,8 +2661,9 @@ async fn payload_from_simple_module( module: &FlowModule, inner_path: Option, ) -> Result { + let delete_after_use = module.delete_after_use.unwrap_or(false); Ok(match value { - FlowModuleValue::Flow { path, .. } => flow_to_payload(path), + FlowModuleValue::Flow { path, .. } => flow_to_payload(path, &delete_after_use), FlowModuleValue::Script { path: script_path, hash: script_hash, .. } => { script_to_payload(script_hash, script_path, db, flow_job, module).await? } @@ -2664,6 +2685,7 @@ async fn payload_from_simple_module( concurrency_time_window_s, module, tag, + &delete_after_use, ), _ => unreachable!("is simple flow"), }) @@ -2678,6 +2700,7 @@ fn raw_script_to_payload( concurrency_time_window_s: &Option, module: &FlowModule, tag: &Option, + delete_after_use: &bool, ) -> JobPayloadWithTag { JobPayloadWithTag { payload: JobPayload::Code(RawCode { @@ -2691,12 +2714,13 @@ fn raw_script_to_payload( dedicated_worker: None, }), tag: tag.clone(), + delete_after_use: *delete_after_use, } } -fn flow_to_payload(path: &str) -> JobPayloadWithTag { +fn flow_to_payload(path: &str, delete_after_use: &bool) -> JobPayloadWithTag { let payload = JobPayload::Flow { path: path.to_string(), dedicated_worker: None }; - JobPayloadWithTag { payload, tag: None } + JobPayloadWithTag { payload, tag: None, delete_after_use: *delete_after_use } } async fn script_to_payload( @@ -2706,7 +2730,7 @@ async fn script_to_payload( flow_job: &QueuedJob, module: &FlowModule, ) -> Result { - let (payload, tag) = if script_hash.is_none() { + let (payload, tag, delete_after_use) = if script_hash.is_none() { script_path_to_payload(script_path, &db, &flow_job.workspace_id).await? } else { let hash = script_hash.clone().unwrap(); @@ -2719,6 +2743,7 @@ async fn script_to_payload( language, dedicated_worker, priority, + delete_after_use, ) = script_hash_to_tag_and_limits(&hash, &mut tx, &flow_job.workspace_id).await?; ( JobPayload::ScriptHash { @@ -2732,9 +2757,13 @@ async fn script_to_payload( priority, }, tag, + delete_after_use, ) }; - Ok(JobPayloadWithTag { payload, tag }) + // the module value overrides the value set at the script level. Defaults to false if both are unset. + let final_delete_after_user = + module.delete_after_use.unwrap_or(false) || delete_after_use.unwrap_or(false); + Ok(JobPayloadWithTag { payload, tag, delete_after_use: final_delete_after_user }) } async fn get_transform_context( diff --git a/frontend/src/lib/components/ScriptBuilder.svelte b/frontend/src/lib/components/ScriptBuilder.svelte index dbef4643a4..afdbeceae7 100644 --- a/frontend/src/lib/components/ScriptBuilder.svelte +++ b/frontend/src/lib/components/ScriptBuilder.svelte @@ -214,7 +214,8 @@ cache_ttl: script.cache_ttl, ws_error_handler_muted: script.ws_error_handler_muted, priority: script.priority, - restart_unless_cancelled: script.restart_unless_cancelled + restart_unless_cancelled: script.restart_unless_cancelled, + delete_after_use: script.delete_after_use } }) savedScript = cloneDeep(script) as NewScriptWithDraft @@ -289,7 +290,8 @@ cache_ttl: script.cache_ttl, ws_error_handler_muted: script.ws_error_handler_muted, priority: script.priority, - restart_unless_cancelled: script.restart_unless_cancelled + restart_unless_cancelled: script.restart_unless_cancelled, + delete_after_use: script.delete_after_use } }) } @@ -707,6 +709,44 @@ > +
+ + + WARNING: This settings ONLY applies to synchronous webhooks or when the script + is used within a flow. If used individually, this script must be triggered + using a synchronous endpoint to have the desired effect. +
+
+ The logs, arguments and results of the job will be completely deleted from Windmill + once it is complete and the result has been returned. +
+
+ The deletion is irreversible. + {#if !$enterpriseLicense} +
+
+ This option is only available on Windmill Enterprise Edition. + {/if} + + +
+ { + if (script.delete_after_use) { + script.delete_after_use = undefined + } else { + script.delete_after_use = true + } + }} + options={{ + right: 'Delete logs, arguments and results after use' + }} + /> +
+
{#if !isCloudHosted()}
Suspend/Approval Sleep Mock + Lifetime
@@ -92,6 +95,11 @@
+ +
+ +
+ diff --git a/frontend/src/lib/components/flows/content/FlowBranchesOneWrapper.svelte b/frontend/src/lib/components/flows/content/FlowBranchesOneWrapper.svelte index fd2fc22290..f33c5a3fb4 100644 --- a/frontend/src/lib/components/flows/content/FlowBranchesOneWrapper.svelte +++ b/frontend/src/lib/components/flows/content/FlowBranchesOneWrapper.svelte @@ -8,10 +8,12 @@ import FlowCard from '../common/FlowCard.svelte' import BranchPredicateEditor from './BranchPredicateEditor.svelte' import FlowModuleEarlyStop from './FlowModuleEarlyStop.svelte' + import FlowModuleDeleteAfterUse from './FlowModuleDeleteAfterUse.svelte' import FlowModuleSleep from './FlowModuleSleep.svelte' import FlowModuleSuspend from './FlowModuleSuspend.svelte' import SplitPanesWrapper from '../../splitPanes/SplitPanesWrapper.svelte' import FlowModuleMock from './FlowModuleMock.svelte' + import { enterpriseLicense } from '$lib/stores' // import FlowRetries from './FlowRetries.svelte' export let flowModule: FlowModule @@ -70,6 +72,7 @@ Suspend/Approval Sleep Mock + Lifetime
@@ -92,6 +95,11 @@
+ +
+ +
+ diff --git a/frontend/src/lib/components/flows/content/FlowLoop.svelte b/frontend/src/lib/components/flows/content/FlowLoop.svelte index 5c0b7e956b..881435e637 100644 --- a/frontend/src/lib/components/flows/content/FlowLoop.svelte +++ b/frontend/src/lib/components/flows/content/FlowLoop.svelte @@ -13,12 +13,14 @@ import type { FlowModule } from '$lib/gen/models/FlowModule' import { Pane, Splitpanes } from 'svelte-splitpanes' import { getStepPropPicker } from '../previousResults' + import { enterpriseLicense } from '$lib/stores' import FlowModuleSleep from './FlowModuleSleep.svelte' import FlowModuleMock from './FlowModuleMock.svelte' import { Play } from 'lucide-svelte' import type { Job } from '$lib/gen' import FlowLoopIterationPreview from '$lib/components/FlowLoopIterationPreview.svelte' + import FlowModuleDeleteAfterUse from './FlowModuleDeleteAfterUse.svelte' const { previewArgs, flowStateStore, flowStore } = getContext('FlowEditorContext') @@ -151,6 +153,7 @@ Suspend/Approval Sleep Mock + Lifetime
@@ -181,6 +184,11 @@
+ +
+ +
+ diff --git a/frontend/src/lib/components/flows/content/FlowModuleComponent.svelte b/frontend/src/lib/components/flows/content/FlowModuleComponent.svelte index f7edd45d24..a5a7cf5150 100644 --- a/frontend/src/lib/components/flows/content/FlowModuleComponent.svelte +++ b/frontend/src/lib/components/flows/content/FlowModuleComponent.svelte @@ -19,6 +19,7 @@ import FlowModuleEarlyStop from './FlowModuleEarlyStop.svelte' import FlowModuleSuspend from './FlowModuleSuspend.svelte' import FlowModuleCache from './FlowModuleCache.svelte' + import FlowModuleDeleteAfterUse from './FlowModuleDeleteAfterUse.svelte' import FlowRetries from './FlowRetries.svelte' import { getStepPropPicker } from '../previousResults' import { deepEqual } from 'fast-equals' @@ -70,6 +71,7 @@ } let selected = 'inputs' let advancedSelected = 'retries' + let advancedRuntimeSelected = 'concurrency' let s3Kind = 'push' let wrapper: HTMLDivElement let panes: HTMLElement @@ -347,46 +349,30 @@ Retries {#if !$selectedId.includes('failure')} + Runtime Cache - Concurrency Early Stop Suspend Sleep Mock Shared Directory - Timeout - Priority {#if flowModule.value['language'] === 'python3' || flowModule.value['language'] === 'deno'} S3 {/if} {/if} + {#if advancedSelected === 'runtime'} + + Concurrency + Timeout + Priority + Lifetime + + {/if}
{#if advancedSelected === 'retries'} - {:else if advancedSelected === 'early-stop'} - - {:else if advancedSelected === 'suspend'} -
- -
- {:else if advancedSelected === 'sleep'} -
- -
- {:else if advancedSelected === 'cache'} -
- -
- {:else if advancedSelected === 'mock'} -
- -
- {:else if advancedSelected === 'timeout'} -
- -
- {:else if advancedSelected === 'concurrency'} + {:else if advancedSelected === 'runtime' && advancedRuntimeSelected === 'concurrency'}
Allowed concurrency within a given timeframe @@ -426,22 +412,11 @@ {/if}
- {:else if advancedSelected === 'same_worker'} + {:else if advancedSelected === 'runtime' && advancedRuntimeSelected === 'timeout'}
- - If shared directory is set, will share a folder that will be mounted on - `./shared` for each of them to pass data between each other. - - +
- {:else if advancedSelected === 'priority'} + {:else if advancedSelected === 'runtime' && advancedRuntimeSelected === 'priority'}
+ {:else if advancedSelected === 'runtime' && advancedRuntimeSelected === 'lifetime'} +
+ +
+ {:else if advancedSelected === 'cache'} +
+ +
+ {:else if advancedSelected === 'early-stop'} + + {:else if advancedSelected === 'suspend'} +
+ +
+ {:else if advancedSelected === 'sleep'} +
+ +
+ {:else if advancedSelected === 'mock'} +
+ +
+ {:else if advancedSelected === 'same_worker'} +
+ + If shared directory is set, will share a folder that will be mounted on + `./shared` for each of them to pass data between each other. + + +
{:else if advancedSelected === 's3'}

diff --git a/frontend/src/lib/components/flows/content/FlowModuleDeleteAfterUse.svelte b/frontend/src/lib/components/flows/content/FlowModuleDeleteAfterUse.svelte new file mode 100644 index 0000000000..162121fb99 --- /dev/null +++ b/frontend/src/lib/components/flows/content/FlowModuleDeleteAfterUse.svelte @@ -0,0 +1,45 @@ + + +
+ + + 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. +
+ This also applies to a flow step that has failed: the error will not be accessible. +
+
+ The deletion is irreversible. + {#if disabled} +
+
+ This option is only available on Windmill Enterprise Edition. + {/if} + + + + { + if (flowModule.delete_after_use) { + flowModule.delete_after_use = undefined + } else { + flowModule.delete_after_use = true + } + }} + options={{ + right: 'Delete logs, arguments and results after the flow is complete' + }} + /> +