diff --git a/backend/.sqlx/query-089d7bc7acdbb97cf477159e111bc7e9ee85289ff5c52af43166928337c257e7.json b/backend/.sqlx/query-089d7bc7acdbb97cf477159e111bc7e9ee85289ff5c52af43166928337c257e7.json index a779aa0e95..6efb66005d 100644 --- a/backend/.sqlx/query-089d7bc7acdbb97cf477159e111bc7e9ee85289ff5c52af43166928337c257e7.json +++ b/backend/.sqlx/query-089d7bc7acdbb97cf477159e111bc7e9ee85289ff5c52af43166928337c257e7.json @@ -35,7 +35,8 @@ "ci_test", "github", "azure", - "asset" + "asset", + "freshness" ] } } diff --git a/backend/.sqlx/query-0e22295e40bfc173d85f0a962d491f3afd87eeea5e4e8d935b1b86eb02f061e7.json b/backend/.sqlx/query-0e22295e40bfc173d85f0a962d491f3afd87eeea5e4e8d935b1b86eb02f061e7.json new file mode 100644 index 0000000000..906c710fdf --- /dev/null +++ b/backend/.sqlx/query-0e22295e40bfc173d85f0a962d491f3afd87eeea5e4e8d935b1b86eb02f061e7.json @@ -0,0 +1,15 @@ +{ + "db_name": "PostgreSQL", + "query": "DELETE FROM pipeline_freshness_state WHERE workspace_id = $1 AND script_path = $2", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Text", + "Text" + ] + }, + "nullable": [] + }, + "hash": "0e22295e40bfc173d85f0a962d491f3afd87eeea5e4e8d935b1b86eb02f061e7" +} diff --git a/backend/.sqlx/query-212553c83e4dcdc6d045eb2fe2dadbb2860ce52d37a56b2861de1215260ecff8.json b/backend/.sqlx/query-212553c83e4dcdc6d045eb2fe2dadbb2860ce52d37a56b2861de1215260ecff8.json index f9bc97d29d..704883d4f1 100644 --- a/backend/.sqlx/query-212553c83e4dcdc6d045eb2fe2dadbb2860ce52d37a56b2861de1215260ecff8.json +++ b/backend/.sqlx/query-212553c83e4dcdc6d045eb2fe2dadbb2860ce52d37a56b2861de1215260ecff8.json @@ -39,7 +39,8 @@ "ci_test", "github", "azure", - "asset" + "asset", + "freshness" ] } } @@ -77,7 +78,8 @@ "ci_test", "github", "azure", - "asset" + "asset", + "freshness" ] } } diff --git a/backend/.sqlx/query-3c84781704b84b8a927ecce5a3fcb3adcf0175d0a71013f2497397c1c8ccc619.json b/backend/.sqlx/query-3c84781704b84b8a927ecce5a3fcb3adcf0175d0a71013f2497397c1c8ccc619.json index 3568d1723e..f5ee767768 100644 --- a/backend/.sqlx/query-3c84781704b84b8a927ecce5a3fcb3adcf0175d0a71013f2497397c1c8ccc619.json +++ b/backend/.sqlx/query-3c84781704b84b8a927ecce5a3fcb3adcf0175d0a71013f2497397c1c8ccc619.json @@ -128,7 +128,8 @@ "ci_test", "github", "azure", - "asset" + "asset", + "freshness" ] } } diff --git a/backend/.sqlx/query-3ea82f16050318a0b3b87c12beb66b49254036ca93f7a6415ceb4e316ed64eed.json b/backend/.sqlx/query-3ea82f16050318a0b3b87c12beb66b49254036ca93f7a6415ceb4e316ed64eed.json new file mode 100644 index 0000000000..60dbee76ab --- /dev/null +++ b/backend/.sqlx/query-3ea82f16050318a0b3b87c12beb66b49254036ca93f7a6415ceb4e316ed64eed.json @@ -0,0 +1,29 @@ +{ + "db_name": "PostgreSQL", + "query": "\n SELECT p.path AS \"path!\",\n (SELECT c.completed_at\n FROM v2_job j\n JOIN v2_job_completed c ON c.id = j.id\n WHERE j.workspace_id = $1\n AND j.runnable_path = p.path\n AND j.parent_job IS NULL\n -- No 'singlestepflow': flows may share a script's path, and\n -- a same-path flow run must not read as the script being\n -- fresh (false-fresh). Script retries land as native\n -- 'script' jobs; only the rare flow-wrapper fallback is\n -- missed, which errs stale. Kept in lockstep with the\n -- freshness watchdog's queries (freshness_watchdog_ee).\n AND j.kind IN ('script', 'preview')\n AND c.status = 'success'\n ORDER BY j.created_at DESC\n LIMIT 1) AS last_success_at\n FROM unnest($2::text[]) AS p(path)\n ", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "path!", + "type_info": "Text" + }, + { + "ordinal": 1, + "name": "last_success_at", + "type_info": "Timestamptz" + } + ], + "parameters": { + "Left": [ + "Text", + "TextArray" + ] + }, + "nullable": [ + null, + null + ] + }, + "hash": "3ea82f16050318a0b3b87c12beb66b49254036ca93f7a6415ceb4e316ed64eed" +} diff --git a/backend/.sqlx/query-45a539e61a3c35098fb6d59c3d5de8b6fcebe83002c74647666e3a6f5b8490dd.json b/backend/.sqlx/query-45a539e61a3c35098fb6d59c3d5de8b6fcebe83002c74647666e3a6f5b8490dd.json new file mode 100644 index 0000000000..5b3e70acbf --- /dev/null +++ b/backend/.sqlx/query-45a539e61a3c35098fb6d59c3d5de8b6fcebe83002c74647666e3a6f5b8490dd.json @@ -0,0 +1,23 @@ +{ + "db_name": "PostgreSQL", + "query": "\n SELECT EXISTS(\n SELECT 1 FROM v2_job_queue q\n JOIN v2_job j ON j.id = q.id\n WHERE q.workspace_id = $1\n AND j.runnable_path = $2\n AND j.parent_job IS NULL\n AND j.kind IN ('script', 'preview')\n AND (q.running = true OR q.scheduled_for <= now())\n ) AS \"in_flight!\"\n ", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "in_flight!", + "type_info": "Bool" + } + ], + "parameters": { + "Left": [ + "Text", + "Text" + ] + }, + "nullable": [ + null + ] + }, + "hash": "45a539e61a3c35098fb6d59c3d5de8b6fcebe83002c74647666e3a6f5b8490dd" +} diff --git a/backend/.sqlx/query-45c73bd8742601753ad947c88b65ff3ae182ff7c0631d4fa44b67b11a133498f.json b/backend/.sqlx/query-45c73bd8742601753ad947c88b65ff3ae182ff7c0631d4fa44b67b11a133498f.json new file mode 100644 index 0000000000..6dc5fdb9a6 --- /dev/null +++ b/backend/.sqlx/query-45c73bd8742601753ad947c88b65ff3ae182ff7c0631d4fa44b67b11a133498f.json @@ -0,0 +1,17 @@ +{ + "db_name": "PostgreSQL", + "query": "\n UPDATE pipeline_freshness_state\n SET attempts = attempts + 1,\n last_push_at = now(),\n next_attempt_at = now()\n + (LEAST($3::bigint, $4::bigint * (1::bigint << LEAST(attempts + 1, 20)))::text\n || ' seconds')::interval\n WHERE workspace_id = $1 AND script_path = $2 AND next_attempt_at <= now()\n ", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Text", + "Text", + "Int8", + "Int8" + ] + }, + "nullable": [] + }, + "hash": "45c73bd8742601753ad947c88b65ff3ae182ff7c0631d4fa44b67b11a133498f" +} diff --git a/backend/.sqlx/query-4a43d4df6c5b2e8dda4308dcb88c23caf312ec377dd91e5307f00d3fb8ec325d.json b/backend/.sqlx/query-4a43d4df6c5b2e8dda4308dcb88c23caf312ec377dd91e5307f00d3fb8ec325d.json index 7d950f6d8f..7474818bf3 100644 --- a/backend/.sqlx/query-4a43d4df6c5b2e8dda4308dcb88c23caf312ec377dd91e5307f00d3fb8ec325d.json +++ b/backend/.sqlx/query-4a43d4df6c5b2e8dda4308dcb88c23caf312ec377dd91e5307f00d3fb8ec325d.json @@ -80,7 +80,8 @@ "ci_test", "github", "azure", - "asset" + "asset", + "freshness" ] } } diff --git a/backend/.sqlx/query-572b80f7b16fd5d1a05f8fe9bbb07667fc1b5e802dc2465ffd13178a042e4f89.json b/backend/.sqlx/query-572b80f7b16fd5d1a05f8fe9bbb07667fc1b5e802dc2465ffd13178a042e4f89.json new file mode 100644 index 0000000000..580666f07d --- /dev/null +++ b/backend/.sqlx/query-572b80f7b16fd5d1a05f8fe9bbb07667fc1b5e802dc2465ffd13178a042e4f89.json @@ -0,0 +1,24 @@ +{ + "db_name": "PostgreSQL", + "query": "\n SELECT EXISTS(\n SELECT 1 FROM v2_job j\n JOIN v2_job_completed c ON c.id = j.id\n WHERE j.workspace_id = $1\n AND j.runnable_path = $2\n AND j.parent_job IS NULL\n AND j.kind IN ('script', 'preview')\n AND c.status = 'success'\n AND c.completed_at > now() - ($3::bigint::text || ' seconds')::interval\n ) AS \"fresh!\"\n ", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "fresh!", + "type_info": "Bool" + } + ], + "parameters": { + "Left": [ + "Text", + "Text", + "Int8" + ] + }, + "nullable": [ + null + ] + }, + "hash": "572b80f7b16fd5d1a05f8fe9bbb07667fc1b5e802dc2465ffd13178a042e4f89" +} diff --git a/backend/.sqlx/query-5e0569092c559a1b19f636fae68f6d740a1720f7746c01f9ea693c70271ff55c.json b/backend/.sqlx/query-5e0569092c559a1b19f636fae68f6d740a1720f7746c01f9ea693c70271ff55c.json new file mode 100644 index 0000000000..18e0936d3a --- /dev/null +++ b/backend/.sqlx/query-5e0569092c559a1b19f636fae68f6d740a1720f7746c01f9ea693c70271ff55c.json @@ -0,0 +1,17 @@ +{ + "db_name": "PostgreSQL", + "query": "\n INSERT INTO pipeline_freshness_state\n (workspace_id, script_path, attempts, last_push_at, next_attempt_at)\n VALUES ($1, $2, 1, now(),\n now() + (LEAST($3::bigint, $4::bigint * 2)::text || ' seconds')::interval)\n ON CONFLICT (workspace_id, script_path) DO NOTHING\n ", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Varchar", + "Varchar", + "Int8", + "Int8" + ] + }, + "nullable": [] + }, + "hash": "5e0569092c559a1b19f636fae68f6d740a1720f7746c01f9ea693c70271ff55c" +} diff --git a/backend/.sqlx/query-67e25a7c19ea0ffaf7ea5303fcd04af5a7eb488c76f783e690af0c2153b1d6a8.json b/backend/.sqlx/query-67e25a7c19ea0ffaf7ea5303fcd04af5a7eb488c76f783e690af0c2153b1d6a8.json index 08ebe6bba5..fcc16e9a7c 100644 --- a/backend/.sqlx/query-67e25a7c19ea0ffaf7ea5303fcd04af5a7eb488c76f783e690af0c2153b1d6a8.json +++ b/backend/.sqlx/query-67e25a7c19ea0ffaf7ea5303fcd04af5a7eb488c76f783e690af0c2153b1d6a8.json @@ -161,7 +161,8 @@ "ci_test", "github", "azure", - "asset" + "asset", + "freshness" ] } } diff --git a/backend/.sqlx/query-6d81570849980f316b08487a03f6017386cf2767c953e0d152e7983a00218c6f.json b/backend/.sqlx/query-6d81570849980f316b08487a03f6017386cf2767c953e0d152e7983a00218c6f.json new file mode 100644 index 0000000000..4e8c7acee2 --- /dev/null +++ b/backend/.sqlx/query-6d81570849980f316b08487a03f6017386cf2767c953e0d152e7983a00218c6f.json @@ -0,0 +1,15 @@ +{ + "db_name": "PostgreSQL", + "query": "UPDATE pipeline_freshness_state SET next_attempt_at = now() - interval '1 second'\n WHERE workspace_id = $1 AND script_path = $2", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Text", + "Text" + ] + }, + "nullable": [] + }, + "hash": "6d81570849980f316b08487a03f6017386cf2767c953e0d152e7983a00218c6f" +} diff --git a/backend/.sqlx/query-7cd0b27f4b4a20b05ac876861b934128c3a6ade024d6662379857905a7447a2a.json b/backend/.sqlx/query-7cd0b27f4b4a20b05ac876861b934128c3a6ade024d6662379857905a7447a2a.json new file mode 100644 index 0000000000..5c8ab6fb5b --- /dev/null +++ b/backend/.sqlx/query-7cd0b27f4b4a20b05ac876861b934128c3a6ade024d6662379857905a7447a2a.json @@ -0,0 +1,32 @@ +{ + "db_name": "PostgreSQL", + "query": "\n SELECT DISTINCT ON (workspace_id, path)\n workspace_id AS \"workspace_id!\", path AS \"path!\", content AS \"content!\"\n FROM script\n WHERE auto_kind = 'pipeline'\n AND archived = false\n AND deleted = false\n AND content ILIKE '%freshness%'\n -- Workspace archival stops all execution but leaves script rows\n -- intact for unarchival; without this the watchdog would keep\n -- resurrecting runs in a workspace the admin shut down.\n AND EXISTS (SELECT 1 FROM workspace w\n WHERE w.id = script.workspace_id AND w.deleted = false)\n ORDER BY workspace_id, path, created_at DESC\n ", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "workspace_id!", + "type_info": "Varchar" + }, + { + "ordinal": 1, + "name": "path!", + "type_info": "Varchar" + }, + { + "ordinal": 2, + "name": "content!", + "type_info": "Text" + } + ], + "parameters": { + "Left": [] + }, + "nullable": [ + false, + false, + false + ] + }, + "hash": "7cd0b27f4b4a20b05ac876861b934128c3a6ade024d6662379857905a7447a2a" +} diff --git a/backend/.sqlx/query-81d1c56c369c5328d947f1ad21da995a8a555e4a9c2d1387dd8cee9c81efd40b.json b/backend/.sqlx/query-81d1c56c369c5328d947f1ad21da995a8a555e4a9c2d1387dd8cee9c81efd40b.json new file mode 100644 index 0000000000..3f6a4a8136 --- /dev/null +++ b/backend/.sqlx/query-81d1c56c369c5328d947f1ad21da995a8a555e4a9c2d1387dd8cee9c81efd40b.json @@ -0,0 +1,15 @@ +{ + "db_name": "PostgreSQL", + "query": "\n DELETE FROM pipeline_freshness_state s\n WHERE NOT EXISTS (\n SELECT 1 FROM unnest($1::text[], $2::text[]) AS w(workspace_id, script_path)\n WHERE w.workspace_id = s.workspace_id AND w.script_path = s.script_path\n )\n ", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "TextArray", + "TextArray" + ] + }, + "nullable": [] + }, + "hash": "81d1c56c369c5328d947f1ad21da995a8a555e4a9c2d1387dd8cee9c81efd40b" +} diff --git a/backend/.sqlx/query-9ecb404e46a4eac55f977f05a3afbafe5dc3cdecc17a3d5a7476b160c1b6e7e1.json b/backend/.sqlx/query-9ecb404e46a4eac55f977f05a3afbafe5dc3cdecc17a3d5a7476b160c1b6e7e1.json index ad9e57801e..ab01730c02 100644 --- a/backend/.sqlx/query-9ecb404e46a4eac55f977f05a3afbafe5dc3cdecc17a3d5a7476b160c1b6e7e1.json +++ b/backend/.sqlx/query-9ecb404e46a4eac55f977f05a3afbafe5dc3cdecc17a3d5a7476b160c1b6e7e1.json @@ -35,7 +35,8 @@ "ci_test", "github", "azure", - "asset" + "asset", + "freshness" ] } } diff --git a/backend/.sqlx/query-a4b6371d33206010b2f3ffd2b09e33244fe8ab9a803248fc23f334034d24aad4.json b/backend/.sqlx/query-a4b6371d33206010b2f3ffd2b09e33244fe8ab9a803248fc23f334034d24aad4.json index 9a21f228ea..405904604a 100644 --- a/backend/.sqlx/query-a4b6371d33206010b2f3ffd2b09e33244fe8ab9a803248fc23f334034d24aad4.json +++ b/backend/.sqlx/query-a4b6371d33206010b2f3ffd2b09e33244fe8ab9a803248fc23f334034d24aad4.json @@ -191,7 +191,8 @@ "ci_test", "github", "azure", - "asset" + "asset", + "freshness" ] } } diff --git a/backend/.sqlx/query-b10e21ec9edf2d22ff956fdcd085bd601b7dcabdf70e85543455cb4fc3986d74.json b/backend/.sqlx/query-b10e21ec9edf2d22ff956fdcd085bd601b7dcabdf70e85543455cb4fc3986d74.json new file mode 100644 index 0000000000..d5b10b13b9 --- /dev/null +++ b/backend/.sqlx/query-b10e21ec9edf2d22ff956fdcd085bd601b7dcabdf70e85543455cb4fc3986d74.json @@ -0,0 +1,23 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT attempts FROM pipeline_freshness_state WHERE workspace_id = $1 AND script_path = $2", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "attempts", + "type_info": "Int4" + } + ], + "parameters": { + "Left": [ + "Text", + "Text" + ] + }, + "nullable": [ + false + ] + }, + "hash": "b10e21ec9edf2d22ff956fdcd085bd601b7dcabdf70e85543455cb4fc3986d74" +} diff --git a/backend/.sqlx/query-b3771b690c5966272b1f42c9965bb6a8f961c119516e4c33dc928cd3b4f4edbc.json b/backend/.sqlx/query-b3771b690c5966272b1f42c9965bb6a8f961c119516e4c33dc928cd3b4f4edbc.json index 3efa843923..092d15e592 100644 --- a/backend/.sqlx/query-b3771b690c5966272b1f42c9965bb6a8f961c119516e4c33dc928cd3b4f4edbc.json +++ b/backend/.sqlx/query-b3771b690c5966272b1f42c9965bb6a8f961c119516e4c33dc928cd3b4f4edbc.json @@ -166,7 +166,8 @@ "ci_test", "github", "azure", - "asset" + "asset", + "freshness" ] } } diff --git a/backend/.sqlx/query-bc6f566bc958923045dc3e1ff1639a26f649c960a4ee8d0a584884a236129d43.json b/backend/.sqlx/query-bc6f566bc958923045dc3e1ff1639a26f649c960a4ee8d0a584884a236129d43.json new file mode 100644 index 0000000000..ce77275faf --- /dev/null +++ b/backend/.sqlx/query-bc6f566bc958923045dc3e1ff1639a26f649c960a4ee8d0a584884a236129d43.json @@ -0,0 +1,15 @@ +{ + "db_name": "PostgreSQL", + "query": "INSERT INTO pipeline_freshness_state (workspace_id, script_path, attempts) VALUES ($1, $2, 3)", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Varchar", + "Varchar" + ] + }, + "nullable": [] + }, + "hash": "bc6f566bc958923045dc3e1ff1639a26f649c960a4ee8d0a584884a236129d43" +} diff --git a/backend/.sqlx/query-bcfa34cf80abea05f0c24883b9e77429c51e6166c414bcc5ce2e97fac25bcd77.json b/backend/.sqlx/query-bcfa34cf80abea05f0c24883b9e77429c51e6166c414bcc5ce2e97fac25bcd77.json index f6ff25a4bf..419ab26383 100644 --- a/backend/.sqlx/query-bcfa34cf80abea05f0c24883b9e77429c51e6166c414bcc5ce2e97fac25bcd77.json +++ b/backend/.sqlx/query-bcfa34cf80abea05f0c24883b9e77429c51e6166c414bcc5ce2e97fac25bcd77.json @@ -80,7 +80,8 @@ "ci_test", "github", "azure", - "asset" + "asset", + "freshness" ] } } diff --git a/backend/.sqlx/query-be6d2c92a62b7b284651c45af809746147aa9b8d0a81642a7b7cb4738a0cad66.json b/backend/.sqlx/query-be6d2c92a62b7b284651c45af809746147aa9b8d0a81642a7b7cb4738a0cad66.json index 8dc66064dc..b9d33a6b5f 100644 --- a/backend/.sqlx/query-be6d2c92a62b7b284651c45af809746147aa9b8d0a81642a7b7cb4738a0cad66.json +++ b/backend/.sqlx/query-be6d2c92a62b7b284651c45af809746147aa9b8d0a81642a7b7cb4738a0cad66.json @@ -111,7 +111,8 @@ "ci_test", "github", "azure", - "asset" + "asset", + "freshness" ] } } diff --git a/backend/.sqlx/query-d1a1f64cb6b14c561c53c44910cad8bc7a2d226a9f22418782bcc66e46d0f706.json b/backend/.sqlx/query-d1a1f64cb6b14c561c53c44910cad8bc7a2d226a9f22418782bcc66e46d0f706.json new file mode 100644 index 0000000000..cbf96dad52 --- /dev/null +++ b/backend/.sqlx/query-d1a1f64cb6b14c561c53c44910cad8bc7a2d226a9f22418782bcc66e46d0f706.json @@ -0,0 +1,15 @@ +{ + "db_name": "PostgreSQL", + "query": "INSERT INTO pipeline_freshness_state (workspace_id, script_path) VALUES ($1, $2)", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Varchar", + "Varchar" + ] + }, + "nullable": [] + }, + "hash": "d1a1f64cb6b14c561c53c44910cad8bc7a2d226a9f22418782bcc66e46d0f706" +} diff --git a/backend/.sqlx/query-d41ea93fd58381b89e151c965eae1ea2fe96a1b94f5a92953fb1c1642d15c016.json b/backend/.sqlx/query-d41ea93fd58381b89e151c965eae1ea2fe96a1b94f5a92953fb1c1642d15c016.json index d5365ffe94..d97c02d26b 100644 --- a/backend/.sqlx/query-d41ea93fd58381b89e151c965eae1ea2fe96a1b94f5a92953fb1c1642d15c016.json +++ b/backend/.sqlx/query-d41ea93fd58381b89e151c965eae1ea2fe96a1b94f5a92953fb1c1642d15c016.json @@ -111,7 +111,8 @@ "ci_test", "github", "azure", - "asset" + "asset", + "freshness" ] } } diff --git a/backend/.sqlx/query-d4211392e174a0e8f89c7fcebdf120e5b0f629f9f04e08a2982df33ff23ac7a9.json b/backend/.sqlx/query-d4211392e174a0e8f89c7fcebdf120e5b0f629f9f04e08a2982df33ff23ac7a9.json index 33a5534b42..052d83fcd9 100644 --- a/backend/.sqlx/query-d4211392e174a0e8f89c7fcebdf120e5b0f629f9f04e08a2982df33ff23ac7a9.json +++ b/backend/.sqlx/query-d4211392e174a0e8f89c7fcebdf120e5b0f629f9f04e08a2982df33ff23ac7a9.json @@ -251,7 +251,8 @@ "ci_test", "github", "azure", - "asset" + "asset", + "freshness" ] } } diff --git a/backend/.sqlx/query-e4d71278fb80126a7a9da73f1889352d4d1e3cb3a8a08f1c9c03055a1cab1235.json b/backend/.sqlx/query-e4d71278fb80126a7a9da73f1889352d4d1e3cb3a8a08f1c9c03055a1cab1235.json index 470c651020..a35373a959 100644 --- a/backend/.sqlx/query-e4d71278fb80126a7a9da73f1889352d4d1e3cb3a8a08f1c9c03055a1cab1235.json +++ b/backend/.sqlx/query-e4d71278fb80126a7a9da73f1889352d4d1e3cb3a8a08f1c9c03055a1cab1235.json @@ -191,7 +191,8 @@ "ci_test", "github", "azure", - "asset" + "asset", + "freshness" ] } } diff --git a/backend/.sqlx/query-e50bce4bba01c90171973862bcfa088b5c0c1ad601146ff1c57d4a3ddc30556a.json b/backend/.sqlx/query-e50bce4bba01c90171973862bcfa088b5c0c1ad601146ff1c57d4a3ddc30556a.json new file mode 100644 index 0000000000..8c80c14ea4 --- /dev/null +++ b/backend/.sqlx/query-e50bce4bba01c90171973862bcfa088b5c0c1ad601146ff1c57d4a3ddc30556a.json @@ -0,0 +1,34 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT runnable_path AS \"runnable_path!\", created_by AS \"created_by!\",\n args AS \"args: sqlx::types::Json\"\n FROM v2_job\n WHERE workspace_id = $1 AND trigger_kind = 'freshness'\n ORDER BY created_at", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "runnable_path!", + "type_info": "Varchar" + }, + { + "ordinal": 1, + "name": "created_by!", + "type_info": "Varchar" + }, + { + "ordinal": 2, + "name": "args: sqlx::types::Json", + "type_info": "Jsonb" + } + ], + "parameters": { + "Left": [ + "Text" + ] + }, + "nullable": [ + true, + false, + true + ] + }, + "hash": "e50bce4bba01c90171973862bcfa088b5c0c1ad601146ff1c57d4a3ddc30556a" +} diff --git a/backend/.sqlx/query-f26cde14890cc754e04341264c4dba01f2f3a02bb36a8d7c20e1990dc81e0472.json b/backend/.sqlx/query-f26cde14890cc754e04341264c4dba01f2f3a02bb36a8d7c20e1990dc81e0472.json new file mode 100644 index 0000000000..8257722c5c --- /dev/null +++ b/backend/.sqlx/query-f26cde14890cc754e04341264c4dba01f2f3a02bb36a8d7c20e1990dc81e0472.json @@ -0,0 +1,22 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT COUNT(*) AS \"count!\" FROM pipeline_freshness_state WHERE workspace_id = $1", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "count!", + "type_info": "Int8" + } + ], + "parameters": { + "Left": [ + "Text" + ] + }, + "nullable": [ + null + ] + }, + "hash": "f26cde14890cc754e04341264c4dba01f2f3a02bb36a8d7c20e1990dc81e0472" +} diff --git a/backend/ee-repo-ref.txt b/backend/ee-repo-ref.txt index 857e48ba09..0cd164d1ce 100644 --- a/backend/ee-repo-ref.txt +++ b/backend/ee-repo-ref.txt @@ -1 +1 @@ -81d8a449effdc540a6e8810668dd5d4aea0c485a +6f5fe0f7f56696fbef5a8349da38496c32e71666 diff --git a/backend/migrations/20260703165613_pipeline_freshness_watchdog.down.sql b/backend/migrations/20260703165613_pipeline_freshness_watchdog.down.sql new file mode 100644 index 0000000000..aed2f9d32a --- /dev/null +++ b/backend/migrations/20260703165613_pipeline_freshness_watchdog.down.sql @@ -0,0 +1,5 @@ +-- Postgres has no ALTER TYPE ... DROP VALUE for enums. The 'freshness' value +-- stays even on rollback, consistent with prior job_trigger_kind additions +-- (see 20260510174213_asset_trigger_dispatch). +DROP INDEX IF EXISTS idx_script_pipeline_freshness_scan; +DROP TABLE IF EXISTS pipeline_freshness_state; diff --git a/backend/migrations/20260703165613_pipeline_freshness_watchdog.up.sql b/backend/migrations/20260703165613_pipeline_freshness_watchdog.up.sql new file mode 100644 index 0000000000..acf9545c96 --- /dev/null +++ b/backend/migrations/20260703165613_pipeline_freshness_watchdog.up.sql @@ -0,0 +1,35 @@ +-- Attribution for runs pushed by the pipeline freshness watchdog (the EE +-- background loop that re-runs a `// freshness`-annotated producer whose +-- output aged past its window). +ALTER TYPE job_trigger_kind ADD VALUE IF NOT EXISTS 'freshness'; + +-- Per-(workspace, script) watchdog state: exponential-backoff bookkeeping so +-- a persistently failing producer isn't re-pushed on every scan tick, and an +-- atomic claim so concurrent servers can't double-push in the same tick +-- (claim = the UPDATE/INSERT that advances next_attempt_at; only the winner +-- pushes). Rows exist only while a script is stale — observing it fresh (or +-- its annotation gone) deletes the row, resetting the backoff. +CREATE TABLE pipeline_freshness_state ( + workspace_id VARCHAR(50) NOT NULL REFERENCES workspace (id) ON DELETE CASCADE, + script_path VARCHAR(510) NOT NULL, + attempts INTEGER NOT NULL DEFAULT 1, + last_push_at TIMESTAMPTZ NOT NULL DEFAULT now(), + next_attempt_at TIMESTAMPTZ NOT NULL DEFAULT now(), + PRIMARY KEY (workspace_id, script_path) +); + +-- Written only by the server monitor loop on the raw (non-RLS) pool, but +-- granted like every other app table so a future user-transaction reader +-- doesn't hit the recurring missing-GRANT class of bug. +GRANT ALL ON pipeline_freshness_state TO windmill_user; +GRANT ALL ON pipeline_freshness_state TO windmill_admin; + +-- The watchdog's ~60s candidate scan (latest deployed pipeline members) +-- filters on this exact predicate and orders by (workspace_id, path, +-- created_at DESC); without a matching partial index it seq-scans the whole +-- script-version heap on every tick, on instances that mostly have zero +-- pipeline scripts. (idx_script_pipeline_path is text_pattern_ops for +-- prefix LIKE — it can't serve this ordering.) +CREATE INDEX idx_script_pipeline_freshness_scan + ON script (workspace_id, path, created_at DESC) + WHERE auto_kind = 'pipeline' AND archived = false AND deleted = false; diff --git a/backend/parsers/windmill-parser/src/asset_parser.rs b/backend/parsers/windmill-parser/src/asset_parser.rs index c22359f6b0..299e3a70f9 100644 --- a/backend/parsers/windmill-parser/src/asset_parser.rs +++ b/backend/parsers/windmill-parser/src/asset_parser.rs @@ -77,10 +77,12 @@ pub struct ParseAssetsOutput { #[serde(skip_serializing_if = "Option::is_none", default)] pub partition: Option, // `// freshness ` — SLA stating outputs must be at most - // `duration` old. Active backstop: when no other trigger has fired the - // script within the window, a watchdog re-runs it. Distinct from - // schedule (which is producer cadence); freshness is consumer SLA and - // applies regardless of which trigger last fired. + // `duration` old. Drives passive monitoring in CE (the asset graph + // colors the node's badge fresh/stale against its last successful run) + // and the enterprise watchdog (windmill-queue `freshness_watchdog`), + // which re-runs a stale unpartitioned producer. Distinct from schedule + // (which is producer cadence); freshness is consumer SLA and applies + // regardless of which trigger last fired. #[serde(skip_serializing_if = "Option::is_none", default)] pub freshness: Option, // `// trigger all` → AND join barrier; default (`any`) = OR (current @@ -637,7 +639,7 @@ fn parse_kv_opts(s: &str) -> BTreeMap { // - `on ` → asset / native trigger edge (including // the marker-only `on schedule` form) // - `partitioned [opts]` → partition declaration -// - `freshness ` → SLA / active backstop +// - `freshness ` → SLA window (badge + EE watchdog) // - `tag ` → worker-tag override (annotation wins // over UI-set value at deploy) // - `retry []` → cascade-only retry policy diff --git a/backend/src/monitor.rs b/backend/src/monitor.rs index 817887eaad..60b4024c0a 100644 --- a/backend/src/monitor.rs +++ b/backend/src/monitor.rs @@ -178,6 +178,14 @@ lazy_static::lazy_static! { .and_then(|x| x.parse::().ok()) .unwrap_or(false); + // Ops kill switch for the pipeline freshness watchdog (a background + // pusher — being able to stop it without a redeploy matters more than + // for read-only monitors). + pub static ref DISABLE_FRESHNESS_WATCHDOG: bool = std::env::var("DISABLE_FRESHNESS_WATCHDOG") + .ok() + .and_then(|x| x.parse::().ok()) + .unwrap_or(false); + pub static ref WORKERS_NAMES: Arc>> = Arc::new(RwLock::new(Vec::new())); static ref QUEUE_COUNT_TAGS: Arc>> = Arc::new(RwLock::new(Vec::new())); @@ -2994,6 +3002,28 @@ pub async fn monitor_db( } }; + // run every ~60s (2 iterations * 30s). Enterprise feature: the active + // `// freshness` backstop lives in windmill-queue's `freshness_watchdog` + // (`private`); OSS gets a no-op stub. Runtime-gated on an Enterprise + // license like the audit export above. Safe on concurrent servers — the + // watchdog claims per-script state rows atomically before pushing. + let pipeline_freshness_watchdog_f = async { + if server_mode + && !*DISABLE_FRESHNESS_WATCHDOG + && iteration.is_some() + && iteration.as_ref().unwrap().should_run(2) + { + if let Some(db) = conn.as_sql() { + if matches!( + windmill_common::ee_oss::get_license_plan().await, + windmill_common::ee_oss::LicensePlan::Enterprise + ) { + windmill_queue::freshness_watchdog::tick(db).await; + } + } + } + }; + join!( expired_items_f, zombie_jobs_f, @@ -3021,6 +3051,7 @@ pub async fn monitor_db( manage_audit_partitions_f, export_audit_logs_to_object_store_f, cleanup_scheduled_job_deletions_f, + pipeline_freshness_watchdog_f, ); } diff --git a/backend/tests/freshness_watchdog.rs b/backend/tests/freshness_watchdog.rs new file mode 100644 index 0000000000..17851d0a1e --- /dev/null +++ b/backend/tests/freshness_watchdog.rs @@ -0,0 +1,294 @@ +//! End-to-end tests for the pipeline freshness watchdog (Enterprise). +//! +//! `windmill_queue::freshness_watchdog::tick` is called directly against +//! seeded `script` / `v2_job(_completed)` rows — no worker or API server is +//! needed, since the watchdog's job ends at the push (the pushed job sitting +//! in `v2_job_queue` is itself part of the assertions). Covers: staleness on +//! never-ran and aged-out members, the fresh short-circuit + state reset, +//! the in-flight suppression, the backoff claim, and the skip rules +//! (partitioned, malformed window, non-pipeline scripts). + +#![cfg(feature = "private")] + +use sqlx::{Pool, Postgres}; +use windmill_queue::freshness_watchdog::tick; +use windmill_test_utils::initialize_tracing; + +const WS: &str = "test-workspace"; +const PATH: &str = "u/test-user/freshness_producer"; + +/// Seed a deployed pipeline-member script. Mirrors the deploy path's output: +/// `auto_kind = 'pipeline'`, empty (non-NULL) lock so run-by-path resolution +/// treats it as deployed, hash derived from path+content for uniqueness. +async fn seed_pipeline_script( + db: &Pool, + path: &str, + content: &str, +) -> anyhow::Result<()> { + let mut h = 0i64; + for b in path.bytes().chain(content.bytes()) { + h = h.wrapping_mul(31).wrapping_add(b as i64); + } + sqlx::query( + r#"INSERT INTO script (workspace_id, hash, path, summary, description, content, + created_by, language, tag, lock, auto_kind) + VALUES ($1, $2, $3, '', '', $4, 'test-user', 'bash'::script_lang, 'bash', '', 'pipeline') + ON CONFLICT DO NOTHING"#, + ) + .bind(WS) + .bind(h) + .bind(path) + .bind(content) + .execute(db) + .await?; + // Process-global deployed-script caches are keyed by (workspace, path) / + // (workspace, hash) and would leak between #[sqlx::test] isolated DBs + // that reuse both — resolve everything from this test's own DB. + windmill_common::DEPLOYED_SCRIPT_CACHE_DISABLED + .store(true, std::sync::atomic::Ordering::Relaxed); + Ok(()) +} + +/// Seed a completed root run of `path` that finished `age_s` seconds ago. +async fn seed_completed_run( + db: &Pool, + path: &str, + age_s: i64, + success: bool, +) -> anyhow::Result<()> { + let id = uuid::Uuid::new_v4(); + sqlx::query( + r#"INSERT INTO v2_job (id, workspace_id, runnable_path, kind, created_at, + created_by, permissioned_as, permissioned_as_email, tag) + VALUES ($1, $2, $3, 'script'::job_kind, + now() - ($4::bigint::text || ' seconds')::interval, + 'test-user', 'u/test-user', 'test@windmill.dev', 'bash')"#, + ) + .bind(id) + .bind(WS) + .bind(path) + .bind(age_s) + .execute(db) + .await?; + sqlx::query( + r#"INSERT INTO v2_job_completed (id, workspace_id, duration_ms, status, started_at, completed_at) + VALUES ($1, $2, 0, CASE WHEN $3 THEN 'success'::job_status ELSE 'failure'::job_status END, + now() - ($4::bigint::text || ' seconds')::interval, + now() - ($4::bigint::text || ' seconds')::interval)"#, + ) + .bind(id) + .bind(WS) + .bind(success) + .bind(age_s) + .execute(db) + .await?; + Ok(()) +} + +/// Jobs the watchdog pushed: (path, created_by, args) rows attributed to +/// `trigger_kind = 'freshness'`. +async fn fetch_pushed( + db: &Pool, +) -> anyhow::Result)>> { + let rows = sqlx::query!( + r#"SELECT runnable_path AS "runnable_path!", created_by AS "created_by!", + args AS "args: sqlx::types::Json" + FROM v2_job + WHERE workspace_id = $1 AND trigger_kind = 'freshness' + ORDER BY created_at"#, + WS, + ) + .fetch_all(db) + .await?; + Ok(rows + .into_iter() + .map(|r| (r.runnable_path, r.created_by, r.args.map(|a| a.0))) + .collect()) +} + +async fn state_row(db: &Pool, path: &str) -> anyhow::Result> { + Ok(sqlx::query_scalar!( + "SELECT attempts FROM pipeline_freshness_state WHERE workspace_id = $1 AND script_path = $2", + WS, + path, + ) + .fetch_optional(db) + .await?) +} + +#[sqlx::test(fixtures("base"))] +async fn never_ran_member_is_pushed_once(db: Pool) -> anyhow::Result<()> { + initialize_tracing().await; + seed_pipeline_script(&db, PATH, "# pipeline\n# freshness 30s\necho hi\n").await?; + + tick(&db).await; + + let pushed = fetch_pushed(&db).await?; + assert_eq!(pushed.len(), 1, "one watchdog push expected"); + let (path, created_by, args) = &pushed[0]; + assert_eq!(path, PATH); + assert_eq!(created_by, &format!("freshness-{PATH}")); + let args = args.as_ref().expect("args recorded"); + assert_eq!( + args.get("_wmill_skip_asset_dispatch"), + Some(&serde_json::json!(true)), + "watchdog runs must not re-fire the cascade" + ); + assert_eq!( + args.pointer("/trigger/kind"), + Some(&serde_json::json!("freshness")) + ); + assert_eq!(state_row(&db, PATH).await?, Some(1), "claim row recorded"); + + // Second tick: the pushed job is queued-and-due, so the in-flight guard + // suppresses a duplicate regardless of backoff. + tick(&db).await; + assert_eq!( + fetch_pushed(&db).await?.len(), + 1, + "no duplicate while queued" + ); + + // Simulate the queued job vanishing without a completion: the backoff + // claim (next_attempt_at in the future) now carries the suppression. + sqlx::query!("DELETE FROM v2_job_queue WHERE workspace_id = $1", WS) + .execute(&db) + .await?; + tick(&db).await; + assert_eq!(fetch_pushed(&db).await?.len(), 1, "backoff holds the retry"); + + // Force the backoff window open: the watchdog retries and escalates. + sqlx::query!( + "UPDATE pipeline_freshness_state SET next_attempt_at = now() - interval '1 second' + WHERE workspace_id = $1 AND script_path = $2", + WS, + PATH, + ) + .execute(&db) + .await?; + tick(&db).await; + assert_eq!(fetch_pushed(&db).await?.len(), 2, "due retry pushed"); + assert_eq!(state_row(&db, PATH).await?, Some(2), "attempts escalated"); + Ok(()) +} + +#[sqlx::test(fixtures("base"))] +async fn fresh_member_is_skipped_and_state_reset(db: Pool) -> anyhow::Result<()> { + initialize_tracing().await; + seed_pipeline_script(&db, PATH, "# pipeline\n# freshness 1h\necho hi\n").await?; + seed_completed_run(&db, PATH, 10, true).await?; + // Leftover backoff row from an earlier staleness episode. + sqlx::query!( + "INSERT INTO pipeline_freshness_state (workspace_id, script_path, attempts) VALUES ($1, $2, 3)", + WS, + PATH, + ) + .execute(&db) + .await?; + + tick(&db).await; + + assert!( + fetch_pushed(&db).await?.is_empty(), + "fresh member not pushed" + ); + assert_eq!(state_row(&db, PATH).await?, None, "backoff reset on fresh"); + Ok(()) +} + +#[sqlx::test(fixtures("base"))] +async fn aged_out_member_is_pushed(db: Pool) -> anyhow::Result<()> { + initialize_tracing().await; + seed_pipeline_script(&db, PATH, "# pipeline\n# freshness 1h\necho hi\n").await?; + // Old success outside the window + a recent failure: still stale. + seed_completed_run(&db, PATH, 7200, true).await?; + seed_completed_run(&db, PATH, 60, false).await?; + + tick(&db).await; + + assert_eq!(fetch_pushed(&db).await?.len(), 1, "aged-out member pushed"); + Ok(()) +} + +#[sqlx::test(fixtures("base"))] +async fn partitioned_malformed_and_plain_members_are_skipped( + db: Pool, +) -> anyhow::Result<()> { + initialize_tracing().await; + // Partitioned: freshness means partition-gap detection, out of scope. + seed_pipeline_script( + &db, + "u/test-user/partitioned", + "# pipeline\n# partitioned daily\n# freshness 1h\necho hi\n", + ) + .await?; + // Malformed window: fails safe to unwatched. + seed_pipeline_script( + &db, + "u/test-user/malformed", + "# pipeline\n# freshness soonish\necho hi\n", + ) + .await?; + // Freshness only in prose (parser must reject; ILIKE prefilter passes). + seed_pipeline_script( + &db, + "u/test-user/prose", + "# pipeline\n# ensure freshness of data below\necho hi\n", + ) + .await?; + + tick(&db).await; + + assert!(fetch_pushed(&db).await?.is_empty(), "no member is watched"); + let rows = sqlx::query_scalar!( + r#"SELECT COUNT(*) AS "count!" FROM pipeline_freshness_state WHERE workspace_id = $1"#, + WS, + ) + .fetch_one(&db) + .await?; + assert_eq!(rows, 0, "no state rows for unwatched members"); + Ok(()) +} + +#[sqlx::test(fixtures("base"))] +async fn archived_workspace_is_not_resurrected(db: Pool) -> anyhow::Result<()> { + initialize_tracing().await; + // Workspace archival stops all execution but keeps script rows for + // unarchival — the watchdog must not keep pushing runs there. + seed_pipeline_script(&db, PATH, "# pipeline\n# freshness 30s\necho hi\n").await?; + sqlx::query!("UPDATE workspace SET deleted = true WHERE id = $1", WS) + .execute(&db) + .await?; + + tick(&db).await; + + assert!( + fetch_pushed(&db).await?.is_empty(), + "no pushes into an archived workspace" + ); + assert_eq!( + state_row(&db, PATH).await?, + None, + "no state bookkeeping either" + ); + Ok(()) +} + +#[sqlx::test(fixtures("base"))] +async fn state_of_unwatched_member_is_cleaned_up(db: Pool) -> anyhow::Result<()> { + initialize_tracing().await; + // A stale-bookkeeping row whose script no longer declares freshness + // (e.g. annotation removed and redeployed) must not survive the sweep. + sqlx::query!( + "INSERT INTO pipeline_freshness_state (workspace_id, script_path) VALUES ($1, $2)", + WS, + "u/test-user/gone", + ) + .execute(&db) + .await?; + + tick(&db).await; + + assert_eq!(state_row(&db, "u/test-user/gone").await?, None); + Ok(()) +} diff --git a/backend/windmill-api-assets/src/lib.rs b/backend/windmill-api-assets/src/lib.rs index b4b0583863..2eb24cee6c 100644 --- a/backend/windmill-api-assets/src/lib.rs +++ b/backend/windmill-api-assets/src/lib.rs @@ -632,6 +632,14 @@ struct GraphRunnableNode { partition_kind: Option, #[serde(skip_serializing_if = "Option::is_none", default)] freshness: Option, + // Completion time of the most recently started successful run of this + // pipeline member. The canvas checks it against the `// freshness` window + // to color the badge fresh/stale. The badge itself is passive; on EE the + // freshness watchdog (windmill-queue) separately re-runs stale + // unpartitioned producers. Absent when no successful run is visible to + // the caller (job RLS applies). + #[serde(skip_serializing_if = "Option::is_none", default)] + last_success_at: Option>, #[serde(skip_serializing_if = "Option::is_none", default)] tag: Option, #[serde(skip_serializing_if = "Option::is_none", default)] @@ -920,6 +928,46 @@ async fn asset_graph( .fetch_all(&mut *tx) .await?; + // Newest successful completed run per pipeline member, for the passive + // freshness status on the canvas. Correlated per-path lookup walks + // ix_job_root_job_index_by_path_2 newest-first until the first success, + // so cost is bounded by the member count, not run history. "Newest" is + // by created_at (the index order), not completed_at: with overlapping + // runs of one path this can pick an earlier completion, erring toward + // stale — never toward false-fresh. Inside the user tx so job-visibility + // RLS applies — a caller who can't see the runs gets no timestamp rather + // than leaked completion times. + let member_paths: Vec = pipeline_member_paths + .iter() + .map(|r| r.path.clone()) + .collect(); + let last_success_rows = sqlx::query!( + r#" + SELECT p.path AS "path!", + (SELECT c.completed_at + FROM v2_job j + JOIN v2_job_completed c ON c.id = j.id + WHERE j.workspace_id = $1 + AND j.runnable_path = p.path + AND j.parent_job IS NULL + -- No 'singlestepflow': flows may share a script's path, and + -- a same-path flow run must not read as the script being + -- fresh (false-fresh). Script retries land as native + -- 'script' jobs; only the rare flow-wrapper fallback is + -- missed, which errs stale. Kept in lockstep with the + -- freshness watchdog's queries (freshness_watchdog_ee). + AND j.kind IN ('script', 'preview') + AND c.status = 'success' + ORDER BY j.created_at DESC + LIMIT 1) AS last_success_at + FROM unnest($2::text[]) AS p(path) + "#, + &w_id, + &member_paths, + ) + .fetch_all(&mut *tx) + .await?; + // Existing scripts / flows in the workspace. Used to filter out // orphan trigger rows whose `script_path` no longer resolves — those // would otherwise be added to `runnable_set` below and surface as @@ -1010,6 +1058,11 @@ async fn asset_graph( (r.path.clone(), lineage) }) .collect(); + let last_success_by_path: std::collections::HashMap> = + last_success_rows + .into_iter() + .filter_map(|r| r.last_success_at.map(|t| (r.path, t))) + .collect(); let pipeline_member_script_paths: std::collections::HashSet = pipeline_member_paths.into_iter().map(|r| r.path).collect(); let existing_script_paths: std::collections::HashSet = @@ -1215,6 +1268,10 @@ async fn asset_graph( freshness: ann .and_then(|a| a.freshness.as_ref()) .map(|f| f.duration.clone()), + last_success_at: (usage_kind == AssetUsageKind::Script) + .then(|| last_success_by_path.get(&path)) + .flatten() + .copied(), tag: ann.and_then(|a| a.tag.clone()), retry: ann.and_then(|a| a.retry.clone()), data_tests: ann.map(|a| a.data_tests.clone()).unwrap_or_default(), diff --git a/backend/windmill-api-scripts/src/scripts.rs b/backend/windmill-api-scripts/src/scripts.rs index 4f9de4f416..49a4091f07 100644 --- a/backend/windmill-api-scripts/src/scripts.rs +++ b/backend/windmill-api-scripts/src/scripts.rs @@ -1252,16 +1252,6 @@ async fn create_script_internal<'c>( // membership; parsed writes tell us what is produced (we don't record // them in auto_kind itself). let pipeline_annotations = parse_pipeline_annotations(&ns.content); - // `// freshness` is parsed but enforcement is a not-yet-implemented - // enterprise feature (skeleton in windmill_common::pipeline_advanced). - // Surface a clear TODO at deploy rather than silently accepting an - // annotation that does nothing. - if pipeline_annotations.freshness.is_some() { - tracing::warn!( - "{}", - windmill_common::pipeline_advanced::freshness_enforcement_todo() - ); - } // `// materialize` materializes a `ducklake:///` target from a // DuckDB script. These two constraints hold for *both* modes: a non-DuckLake // target would otherwise deploy, register a producer in the asset graph, then diff --git a/backend/windmill-api/openapi.yaml b/backend/windmill-api/openapi.yaml index 01ae7229e6..fbad5fffd4 100644 --- a/backend/windmill-api/openapi.yaml +++ b/backend/windmill-api/openapi.yaml @@ -25592,6 +25592,7 @@ components: - google - github - asset + - freshness TriggerMode: description: job trigger mode diff --git a/backend/windmill-common/src/assets.rs b/backend/windmill-common/src/assets.rs index 0757ef6894..722ad29747 100644 --- a/backend/windmill-common/src/assets.rs +++ b/backend/windmill-common/src/assets.rs @@ -352,6 +352,10 @@ mod debounce_duration_tests { assert_eq!(parse_duration_secs("5m"), Some(300)); assert_eq!(parse_duration_secs("2h"), Some(7200)); assert_eq!(parse_duration_secs(" 1d "), Some(86400)); + // Explicit plus sign comes free with i64 parsing; the TS mirror + // (parseDurationSecs) matches it — keep the two in lockstep. + assert_eq!(parse_duration_secs("+5m"), Some(300)); + assert_eq!(parse_duration_secs("+45"), Some(45)); } #[test] diff --git a/backend/windmill-common/src/pipeline_advanced_oss.rs b/backend/windmill-common/src/pipeline_advanced_oss.rs index 009e783bd5..15bb4b127a 100644 --- a/backend/windmill-common/src/pipeline_advanced_oss.rs +++ b/backend/windmill-common/src/pipeline_advanced_oss.rs @@ -1,16 +1,12 @@ -//! OSS fallback: pipeline freshness/SLA enforcement and partition backfills -//! are enterprise features; their implementations live in windmill-ee-private -//! (see `pipeline_advanced_ee`). In the public build the entry points report -//! that the enterprise edition is required. +//! OSS fallback: pipeline partition backfills are an enterprise feature; +//! their implementations live in windmill-ee-private (see +//! `pipeline_advanced_ee`). In the public build the entry points report that +//! the enterprise edition is required. (Freshness lives elsewhere: the +//! fresh/stale badge is CE in the assets API, the active watchdog is +//! windmill-queue's `freshness_watchdog`.) use crate::error::Error; -pub fn freshness_enforcement_todo() -> Error { - Error::internal_err( - "Pipeline freshness/SLA enforcement requires the enterprise edition".to_string(), - ) -} - pub fn backfill_todo() -> Error { Error::internal_err("Pipeline partition backfill requires the enterprise edition".to_string()) } diff --git a/backend/windmill-queue/src/freshness_watchdog_oss.rs b/backend/windmill-queue/src/freshness_watchdog_oss.rs new file mode 100644 index 0000000000..319fb25f62 --- /dev/null +++ b/backend/windmill-queue/src/freshness_watchdog_oss.rs @@ -0,0 +1,9 @@ +//! OSS fallback for the pipeline freshness watchdog. The active backstop — +//! re-running a `// freshness`-annotated producer whose output aged past its +//! window — is an enterprise feature (see `freshness_watchdog_ee`). In the +//! public build the tick is a no-op; CE keeps the passive fresh/stale badge +//! on the asset graph. + +use windmill_common::DB; + +pub async fn tick(_db: &DB) {} diff --git a/backend/windmill-queue/src/lib.rs b/backend/windmill-queue/src/lib.rs index 6f689c7026..59ac7f1fa8 100644 --- a/backend/windmill-queue/src/lib.rs +++ b/backend/windmill-queue/src/lib.rs @@ -14,6 +14,13 @@ pub mod cascade_oss; pub use cascade_ee as cascade; #[cfg(not(feature = "private"))] pub use cascade_oss as cascade; +#[cfg(feature = "private")] +pub mod freshness_watchdog_ee; +pub mod freshness_watchdog_oss; +#[cfg(feature = "private")] +pub use freshness_watchdog_ee as freshness_watchdog; +#[cfg(not(feature = "private"))] +pub use freshness_watchdog_oss as freshness_watchdog; pub mod jobs; #[cfg(feature = "private")] pub mod jobs_ee; diff --git a/backend/windmill-types/src/jobs.rs b/backend/windmill-types/src/jobs.rs index 448e6f4d01..f9372b949d 100644 --- a/backend/windmill-types/src/jobs.rs +++ b/backend/windmill-types/src/jobs.rs @@ -46,6 +46,9 @@ pub enum JobTriggerKind { // A run dispatched because an upstream pipeline script wrote an asset // this runnable subscribes to via `// on s3://...` annotations. Asset, + // A run pushed by the pipeline freshness watchdog (EE) because the + // script's `// freshness` window elapsed without a successful run. + Freshness, } impl std::fmt::Display for JobTriggerKind { @@ -68,6 +71,7 @@ impl std::fmt::Display for JobTriggerKind { JobTriggerKind::Github => "github", JobTriggerKind::CiTest => "ci_test", JobTriggerKind::Asset => "asset", + JobTriggerKind::Freshness => "freshness", }; write!(f, "{}", kind) } diff --git a/frontend/src/lib/components/assets/AssetGraph/AssetGraphCanvas.svelte b/frontend/src/lib/components/assets/AssetGraph/AssetGraphCanvas.svelte index 1c5cf2dde6..6352e6adc9 100644 --- a/frontend/src/lib/components/assets/AssetGraph/AssetGraphCanvas.svelte +++ b/frontend/src/lib/components/assets/AssetGraph/AssetGraphCanvas.svelte @@ -401,6 +401,7 @@ in_pipeline: r.in_pipeline ?? false, partition_kind: r.partition_kind, freshness: r.freshness, + last_success_at: r.last_success_at, tag: r.tag, retry: r.retry, macros: r.macros, diff --git a/frontend/src/lib/components/assets/AssetGraph/RunnableNode.svelte b/frontend/src/lib/components/assets/AssetGraph/RunnableNode.svelte index 8b5e90170c..6bbf8f5656 100644 --- a/frontend/src/lib/components/assets/AssetGraph/RunnableNode.svelte +++ b/frontend/src/lib/components/assets/AssetGraph/RunnableNode.svelte @@ -22,12 +22,13 @@ import { preventDefault, stopPropagation } from 'svelte/legacy' import type { GraphUsageKind } from './types' import type { RunnableRunState } from './activeRunnables.svelte' + import { parseDurationSecs } from './parsePipelineAnnotations' import { NODE } from '$lib/components/graph/util' import DropdownV2 from '$lib/components/DropdownV2.svelte' import Popover from '$lib/components/meltComponents/Popover.svelte' import type { Item } from '$lib/utils' import { workspaceStore } from '$lib/stores' - import { sendUserToast } from '$lib/utils' + import { sendUserToast, msToReadableTimeShort } from '$lib/utils' interface Props { data: { @@ -36,6 +37,9 @@ in_pipeline?: boolean partition_kind?: 'daily' | 'hourly' | 'weekly' | 'monthly' | 'dynamic' freshness?: string + // Completion time (ISO) of the newest successful run visible to + // the caller. With `freshness`, drives the fresh/stale chip state. + last_success_at?: string tag?: string retry?: { count: number; delay?: string } // Macros this script provides (deployed/drafted `// macros` library). @@ -122,6 +126,44 @@ } } + // Freshness verdict: newest successful run (server `last_success_at`, + // or a newer one the session poll observed) vs the `// freshness` + // window. No verdict (undefined) for drafts — no run history — and for + // unparseable windows; the chip then stays neutral like the other + // annotation chips. + let freshnessWindowS = $derived(data.freshness ? parseDurationSecs(data.freshness) : undefined) + // Ticks so a node crosses fresh→stale while the canvas stays open (the + // graph payload is static between refetches). Armed only when a verdict + // is rendered. + let nowMs = $state(Date.now()) + $effect(() => { + if (freshnessWindowS === undefined || data.unsaved) return + const id = setInterval(() => (nowMs = Date.now()), 30_000) + return () => clearInterval(id) + }) + let lastSuccessMs = $derived.by(() => { + const server = data.last_success_at ? new Date(data.last_success_at).getTime() : undefined + const polled = data.runState?.lastSuccessAt + ? new Date(data.runState.lastSuccessAt).getTime() + : undefined + if (server === undefined) return polled + return polled === undefined ? server : Math.max(server, polled) + }) + let freshnessState = $derived.by((): 'fresh' | 'stale' | undefined => { + if (freshnessWindowS === undefined || data.unsaved) return undefined + if (lastSuccessMs === undefined) return 'stale' + return nowMs - lastSuccessMs <= freshnessWindowS * 1000 ? 'fresh' : 'stale' + }) + let freshnessTooltip = $derived.by(() => { + const base = `// freshness ${data.freshness}` + if (freshnessState === undefined) return base + if (lastSuccessMs === undefined) return `${base} — stale: no successful run yet` + const ago = msToReadableTimeShort(Math.max(0, nowMs - lastSuccessMs)) + return freshnessState === 'fresh' + ? `${base} — fresh: last successful run ${ago} ago` + : `${base} — stale: last successful run ${ago} ago` + }) + // Cascade + bounded-run options live on the Run button's caret popover // (whenever there's a cascade OR a bounded-run start — see `hasCaret` // below), so the kebab menu stays focused on lifecycle actions only. @@ -164,8 +206,8 @@ + guidelines). Only the freshness chip (when it has a verdict) + and the run-state chip below use semantic colors. --> {#if data.partition_kind}
{data.partition_kind}
{/if} + {#if data.freshness}
{data.freshness} diff --git a/frontend/src/lib/components/assets/AssetGraph/activeRunnables.svelte.ts b/frontend/src/lib/components/assets/AssetGraph/activeRunnables.svelte.ts index 917e21766a..f6682804c7 100644 --- a/frontend/src/lib/components/assets/AssetGraph/activeRunnables.svelte.ts +++ b/frontend/src/lib/components/assets/AssetGraph/activeRunnables.svelte.ts @@ -1,8 +1,15 @@ import { JobService } from '$lib/gen' export type RunStatus = 'running' | 'success' | 'failure' -/** Per-runnable badge state: latest run status + runs observed this session. */ -export type RunnableRunState = { status: RunStatus; runs: number } +/** + * Per-runnable badge state: latest run status + runs observed this session. + * `lastSuccessAt` is the completion time (start + duration when the listing + * carries it, else start as a conservative lower bound) of the newest + * successful run seen by the poll — lets the freshness chip go green right + * after an in-session run, ahead of the next graph refetch (whose + * `last_success_at` would carry it). + */ +export type RunnableRunState = { status: RunStatus; runs: number; lastSuccessAt?: string } export type EventStatus = 'queued' | 'running' | 'success' | 'failure' /** One folder activity-log row (a job observed by the poll). */ @@ -14,6 +21,12 @@ export type PipelineEvent = { /** What started it, as far as the job listing reveals. */ source: 'schedule' | 'run' at: string + /** + * Completion time (start + duration) for completed rows. The freshness + * chip compares against completion — `at` is the start time and would + * read a long run as older than its output actually is. + */ + completedAt?: string /** * Queued jobs: when the job is due to start. A future value means a * scheduled run waiting for its cron tick, not pipeline activity. @@ -45,7 +58,8 @@ function statesEq(a: Map, b: Map() const countedJobIds = new Set() // Job ids we've observed in-flight at least once. The catch-up pulse is @@ -241,10 +255,25 @@ export function useActiveRunnableIds( const prev = completedHistory.get(id) const status: RunStatus = (j as any).success === true ? 'success' : 'failure' const ts = startedTs ?? new Date(pollStartedMs).toISOString() + // Freshness compares against COMPLETION time (that's + // when the output materialized — the server-side + // last_success_at is completed_at too). The listing + // only carries started_at, so add duration_ms; when + // absent, the start is a conservative lower bound + // (errs stale, never false-fresh). + const durationMs = (j as any).duration_ms + const doneTs = + typeof durationMs === 'number' && startedTs + ? new Date(new Date(startedTs).getTime() + durationMs).toISOString() + : ts completedHistory.set(id, { runs: (prev?.runs ?? 0) + 1, lastStatus: !prev || ts >= prev.lastTs ? status : prev.lastStatus, - lastTs: !prev || ts >= prev.lastTs ? ts : prev.lastTs + lastTs: !prev || ts >= prev.lastTs ? ts : prev.lastTs, + lastSuccessTs: + status === 'success' && (!prev?.lastSuccessTs || doneTs >= prev.lastSuccessTs) + ? doneTs + : prev?.lastSuccessTs }) } } @@ -267,6 +296,10 @@ export function useActiveRunnableIds( : 'failure', source: (j as any).schedule_path ? 'schedule' : 'run', at: startedTs ?? new Date(pollStartedMs).toISOString(), + completedAt: + !isQueued && typeof (j as any).duration_ms === 'number' && startedTs + ? new Date(new Date(startedTs).getTime() + (j as any).duration_ms).toISOString() + : undefined, scheduledFor: isQueued ? ((j as any).scheduled_for as string | undefined) : undefined }) } @@ -289,7 +322,11 @@ export function useActiveRunnableIds( // previous badge state until a worker picks the job up. const snap = new Map() for (const [id, h] of completedHistory) { - snap.set(id, { status: runningThisTick.has(id) ? 'running' : h.lastStatus, runs: h.runs }) + snap.set(id, { + status: runningThisTick.has(id) ? 'running' : h.lastStatus, + runs: h.runs, + lastSuccessAt: h.lastSuccessTs + }) } for (const id of runningThisTick) { if (!snap.has(id)) snap.set(id, { status: 'running', runs: 0 }) diff --git a/frontend/src/lib/components/assets/AssetGraph/parsePipelineAnnotations.test.ts b/frontend/src/lib/components/assets/AssetGraph/parsePipelineAnnotations.test.ts index 315c7526f3..e9e23525ec 100644 --- a/frontend/src/lib/components/assets/AssetGraph/parsePipelineAnnotations.test.ts +++ b/frontend/src/lib/components/assets/AssetGraph/parsePipelineAnnotations.test.ts @@ -1,6 +1,7 @@ import { describe, expect, it } from 'vitest' import { mergeColumnLineage, + parseDurationSecs, parsePipelineAnnotations, type ColumnLineage } from './parsePipelineAnnotations' @@ -188,3 +189,40 @@ describe('mergeColumnLineage', () => { expect(mergeColumnLineage([], annotated)).toEqual(annotated) }) }) + +// Mirror of the Rust `parse_duration_secs` tests (windmill-common assets.rs) +// — the freshness chip's staleness verdict depends on identical parsing. +describe('parseDurationSecs', () => { + it('parses suffixed durations', () => { + expect(parseDurationSecs('30s')).toBe(30) + expect(parseDurationSecs('5m')).toBe(300) + expect(parseDurationSecs('2h')).toBe(7200) + expect(parseDurationSecs('1d')).toBe(86400) + }) + + it('bare integer means seconds', () => { + expect(parseDurationSecs('45')).toBe(45) + }) + + it('tolerates surrounding whitespace', () => { + expect(parseDurationSecs(' 5 m ')).toBe(300) + }) + + it('accepts an explicit plus sign (Rust i64 parsing does)', () => { + expect(parseDurationSecs('+5m')).toBe(300) + expect(parseDurationSecs('+45')).toBe(45) + }) + + it('rejects malformed / non-positive input', () => { + expect(parseDurationSecs('')).toBeUndefined() + expect(parseDurationSecs('h')).toBeUndefined() + expect(parseDurationSecs('1.5h')).toBeUndefined() + expect(parseDurationSecs('-5m')).toBeUndefined() + expect(parseDurationSecs('0')).toBeUndefined() + expect(parseDurationSecs('fast')).toBeUndefined() + }) + + it('rejects values beyond i32 seconds (mirrors backend cap)', () => { + expect(parseDurationSecs('999999999d')).toBeUndefined() + }) +}) diff --git a/frontend/src/lib/components/assets/AssetGraph/parsePipelineAnnotations.ts b/frontend/src/lib/components/assets/AssetGraph/parsePipelineAnnotations.ts index f33a03fff5..964e48570a 100644 --- a/frontend/src/lib/components/assets/AssetGraph/parsePipelineAnnotations.ts +++ b/frontend/src/lib/components/assets/AssetGraph/parsePipelineAnnotations.ts @@ -67,6 +67,28 @@ export type FreshnessSpec = { duration: string } +// Mirrors backend `parse_duration_secs` (windmill-common assets.rs): a bare +// integer means seconds, otherwise `` with an `s`/`m`/`h`/`d` suffix +// (e.g. `30s`, `5m`, `2h`, `1d`). Returns undefined for malformed or +// non-positive input so a typo'd `// freshness` window fails safe (the chip +// stays neutral instead of guessing a staleness verdict). +export function parseDurationSecs(s: string): number | undefined { + const t = s.trim() + if (!t) return undefined + const last = t[t.length - 1] + const mult = + last === 's' ? 1 : last === 'm' ? 60 : last === 'h' ? 3600 : last === 'd' ? 86400 : undefined + const num = (mult !== undefined ? t.slice(0, -1) : t).trim() + // `+?`: Rust's i64 parsing accepts an explicit plus sign (`+5m`), so the + // mirror must too — divergence here would leave the chip neutral for a + // window the deploy path and watchdog honor. + if (mult === undefined && !/^\+?\d+$/.test(t)) return undefined + if (!/^\+?\d+$/.test(num)) return undefined + const secs = Number(num) * (mult ?? 1) + if (!Number.isSafeInteger(secs) || secs <= 0 || secs > 2147483647) return undefined + return secs +} + // `// retry []` — see backend RetrySpec. Delay is kept as the // raw duration string and resolved to seconds at deploy. export type RetrySpec = { diff --git a/frontend/src/lib/components/assets/AssetGraph/pipelineHistory.svelte.ts b/frontend/src/lib/components/assets/AssetGraph/pipelineHistory.svelte.ts index 50ad3e1d6e..d654085179 100644 --- a/frontend/src/lib/components/assets/AssetGraph/pipelineHistory.svelte.ts +++ b/frontend/src/lib/components/assets/AssetGraph/pipelineHistory.svelte.ts @@ -80,7 +80,15 @@ export function usePipelineHistory( kind: j.job_kind.startsWith('flow') ? 'flow' : 'script', status: j.success ? 'success' : 'failure', source: j.schedule_path ? 'schedule' : 'run', - at: j.started_at ?? j.created_at + at: j.started_at ?? j.created_at, + // Same completion-time derivation as the live poll — + // the freshness chip compares against completion, and + // `at` (start) would read a long run as older than its + // output actually is. + completedAt: + j.started_at != undefined + ? new Date(new Date(j.started_at).getTime() + j.duration_ms).toISOString() + : undefined }) } sawFullPage = rows.length === PER_PAGE diff --git a/frontend/src/lib/components/assets/AssetGraph/types.ts b/frontend/src/lib/components/assets/AssetGraph/types.ts index 0db259a389..fb7b5b1de7 100644 --- a/frontend/src/lib/components/assets/AssetGraph/types.ts +++ b/frontend/src/lib/components/assets/AssetGraph/types.ts @@ -22,6 +22,11 @@ export interface AssetGraphRunnableNode { // Raw `// freshness ` value, e.g. "1h", "30m". Surfaced for // the badge; the runtime parses it as needed. freshness?: string + // Completion time (ISO) of the newest successful run of this pipeline + // member visible to the caller. The freshness chip compares it against + // the `// freshness` window to render fresh/stale. Absent = no + // successful run found (or none visible under job RLS). + last_success_at?: string // `// tag ` worker-tag override. Surfaced for the badge so users // can see which worker pool will pick this script up at a glance. tag?: string diff --git a/frontend/src/lib/components/triggers/utils.ts b/frontend/src/lib/components/triggers/utils.ts index 231de5cea4..3d17491791 100644 --- a/frontend/src/lib/components/triggers/utils.ts +++ b/frontend/src/lib/components/triggers/utils.ts @@ -1,4 +1,14 @@ -import { Webhook, Mail, Calendar, Route, Unplug, Database, Terminal } from 'lucide-svelte' +import { + Webhook, + Mail, + Calendar, + Route, + Unplug, + Database, + Terminal, + Timer, + Zap +} from 'lucide-svelte' import KafkaIcon from '$lib/components/icons/KafkaIcon.svelte' import NatsIcon from '$lib/components/icons/NatsIcon.svelte' import MqttIcon from '$lib/components/icons/MqttIcon.svelte' @@ -82,7 +92,8 @@ export const jobTriggerKinds: JobTriggerKind[] = [ 'azure', 'google', 'github', - 'asset' + 'asset', + 'freshness' ] export type Trigger = { @@ -118,7 +129,12 @@ export const triggerIconMap = { cli: Terminal, nextcloud: NextcloudIcon, google: GoogleIcon, - github: GithubIcon + github: GithubIcon, + // Job-attribution-only kinds (no trigger CRUD page): the pipeline asset + // cascade and the freshness watchdog. Needed so the Runs filter and job + // detail render these trigger kinds instead of a blank label / no icon. + asset: Zap, + freshness: Timer } export const triggerDisplayNamesMap = { @@ -139,8 +155,12 @@ export const triggerDisplayNamesMap = { cli: 'CLI', nextcloud: 'Nextcloud', google: 'Google', - github: 'GitHub' -} as const satisfies Record + github: 'GitHub', + asset: 'Asset cascade', + freshness: 'Freshness' + // `asset` / `freshness` are job-attribution-only (JobTriggerKind, not + // TriggerType) — hence the union in the satisfies below. +} as const satisfies Record /** * Converts a TriggerType to a CaptureTriggerKind when a mapping exists diff --git a/frontend/src/routes/(root)/(logged)/pipeline/[folder]/+page.svelte b/frontend/src/routes/(root)/(logged)/pipeline/[folder]/+page.svelte index 4d41f7ff05..845a93505b 100644 --- a/frontend/src/routes/(root)/(logged)/pipeline/[folder]/+page.svelte +++ b/frontend/src/routes/(root)/(logged)/pipeline/[folder]/+page.svelte @@ -1337,8 +1337,17 @@ ? 'failure' : 'success' const cur = m.get(id) - if (cur) cur.runs += 1 - else m.set(id, { status, runs: 1 }) + // Freshness compares against completion; `at` (start) is the + // fallback lower bound for rows without a duration. + const successAt = e.status === 'success' ? (e.completedAt ?? e.at) : undefined + if (cur) { + cur.runs += 1 + // Newest-first, so the first success per id is the latest one — + // it feeds the freshness chip between graph refetches. + if (successAt && !cur.lastSuccessAt) cur.lastSuccessAt = successAt + } else { + m.set(id, { status, runs: 1, lastSuccessAt: successAt }) + } } return m })