diff --git a/backend/.sqlx/query-225d08844cda9d185821e7e6c6956d2ab6037eee12a72a3e3f1f70d6c70a992f.json b/backend/.sqlx/query-225d08844cda9d185821e7e6c6956d2ab6037eee12a72a3e3f1f70d6c70a992f.json new file mode 100644 index 0000000000..f14c69ab10 --- /dev/null +++ b/backend/.sqlx/query-225d08844cda9d185821e7e6c6956d2ab6037eee12a72a3e3f1f70d6c70a992f.json @@ -0,0 +1,15 @@ +{ + "db_name": "PostgreSQL", + "query": "INSERT INTO variable (workspace_id, path, value, is_secret, description, extra_perms, account, is_oauth, expires_at, value_expires_at)\n SELECT $2, path, value, is_secret, description, extra_perms, account, is_oauth, expires_at, value_expires_at\n FROM variable\n WHERE workspace_id = $1", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Text", + "Varchar" + ] + }, + "nullable": [] + }, + "hash": "225d08844cda9d185821e7e6c6956d2ab6037eee12a72a3e3f1f70d6c70a992f" +} diff --git a/backend/.sqlx/query-50661c7f9bec1f902aebf0fde103f02780806874b48ce53eb91d1579f2af3d74.json b/backend/.sqlx/query-50661c7f9bec1f902aebf0fde103f02780806874b48ce53eb91d1579f2af3d74.json new file mode 100644 index 0000000000..a65726fe12 --- /dev/null +++ b/backend/.sqlx/query-50661c7f9bec1f902aebf0fde103f02780806874b48ce53eb91d1579f2af3d74.json @@ -0,0 +1,41 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT value, is_secret, description, value_expires_at\n FROM variable\n WHERE workspace_id = $1 AND path = $2", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "value", + "type_info": "Varchar" + }, + { + "ordinal": 1, + "name": "is_secret", + "type_info": "Bool" + }, + { + "ordinal": 2, + "name": "description", + "type_info": "Varchar" + }, + { + "ordinal": 3, + "name": "value_expires_at", + "type_info": "Timestamptz" + } + ], + "parameters": { + "Left": [ + "Text", + "Text" + ] + }, + "nullable": [ + false, + false, + false, + true + ] + }, + "hash": "50661c7f9bec1f902aebf0fde103f02780806874b48ce53eb91d1579f2af3d74" +} diff --git a/backend/.sqlx/query-63c3eaefa9c0bd20addacfb4eb2b33f94e1bf1dad987184492ca816d2a53348e.json b/backend/.sqlx/query-63c3eaefa9c0bd20addacfb4eb2b33f94e1bf1dad987184492ca816d2a53348e.json new file mode 100644 index 0000000000..00c9b506fb --- /dev/null +++ b/backend/.sqlx/query-63c3eaefa9c0bd20addacfb4eb2b33f94e1bf1dad987184492ca816d2a53348e.json @@ -0,0 +1,12 @@ +{ + "db_name": "PostgreSQL", + "query": "INSERT INTO variable (workspace_id, path, value, is_secret, description, value_expires_at)\n VALUES ('test-workspace', 'u/admin/expiring', 'v', false, 'same', '2027-01-01T00:00:00Z')", + "describe": { + "columns": [], + "parameters": { + "Left": [] + }, + "nullable": [] + }, + "hash": "63c3eaefa9c0bd20addacfb4eb2b33f94e1bf1dad987184492ca816d2a53348e" +} diff --git a/backend/.sqlx/query-0c5b02b6b70fb8fd2ab3e6c57897038750a44a67360d342b6ef705ef2e4d3007.json b/backend/.sqlx/query-6b40726bdc0d9be7eac7680eeabb0f92341d7ba86acaf43cf8209b45849c43da.json similarity index 50% rename from backend/.sqlx/query-0c5b02b6b70fb8fd2ab3e6c57897038750a44a67360d342b6ef705ef2e4d3007.json rename to backend/.sqlx/query-6b40726bdc0d9be7eac7680eeabb0f92341d7ba86acaf43cf8209b45849c43da.json index e74bd5de0e..996390d8ca 100644 --- a/backend/.sqlx/query-0c5b02b6b70fb8fd2ab3e6c57897038750a44a67360d342b6ef705ef2e4d3007.json +++ b/backend/.sqlx/query-6b40726bdc0d9be7eac7680eeabb0f92341d7ba86acaf43cf8209b45849c43da.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "INSERT INTO workspace_settings (workspace_id, slack_team_id, slack_name, slack_command_script, slack_email, customer_id, plan, webhook, ai_config, large_file_storage, git_sync, default_app, default_scripts, deploy_ui, mute_critical_alerts, color, operator_settings, teams_command_script, teams_team_id, teams_team_name, git_app_installations, ducklake, dbt_warehouses, slack_oauth_client_id, slack_oauth_client_secret, datatable, teams_team_guid, auto_invite, error_handler, success_handler, public_app_execution_limit_per_minute, error_handler_fallback_to_instance_alerts) SELECT $1, slack_team_id, slack_name, slack_command_script, slack_email, customer_id, plan, webhook, ai_config, large_file_storage, git_sync, default_app, default_scripts, deploy_ui, mute_critical_alerts, color, operator_settings, teams_command_script, teams_team_id, teams_team_name, git_app_installations, ducklake, dbt_warehouses, slack_oauth_client_id, slack_oauth_client_secret, datatable, teams_team_guid, auto_invite, error_handler, success_handler, public_app_execution_limit_per_minute, error_handler_fallback_to_instance_alerts FROM workspace_settings WHERE workspace_id = $2", + "query": "INSERT INTO workspace_settings (workspace_id, slack_team_id, slack_name, slack_command_script, slack_email, customer_id, plan, webhook, ai_config, large_file_storage, git_sync, default_app, default_scripts, deploy_ui, mute_critical_alerts, color, operator_settings, teams_command_script, teams_team_id, teams_team_name, git_app_installations, ducklake, dbt_warehouses, slack_oauth_client_id, slack_oauth_client_secret, datatable, teams_team_guid, auto_invite, error_handler, success_handler, public_app_execution_limit_per_minute, error_handler_fallback_to_instance_alerts, variable_expiration_handler) SELECT $1, slack_team_id, slack_name, slack_command_script, slack_email, customer_id, plan, webhook, ai_config, large_file_storage, git_sync, default_app, default_scripts, deploy_ui, mute_critical_alerts, color, operator_settings, teams_command_script, teams_team_id, teams_team_name, git_app_installations, ducklake, dbt_warehouses, slack_oauth_client_id, slack_oauth_client_secret, datatable, teams_team_guid, auto_invite, error_handler, success_handler, public_app_execution_limit_per_minute, error_handler_fallback_to_instance_alerts, variable_expiration_handler FROM workspace_settings WHERE workspace_id = $2", "describe": { "columns": [], "parameters": { @@ -11,5 +11,5 @@ }, "nullable": [] }, - "hash": "0c5b02b6b70fb8fd2ab3e6c57897038750a44a67360d342b6ef705ef2e4d3007" + "hash": "6b40726bdc0d9be7eac7680eeabb0f92341d7ba86acaf43cf8209b45849c43da" } diff --git a/backend/.sqlx/query-6e64c0eaa4b804424f5890468eac6b1cf4b1d811bcf6a3e2e7b0d39b321eaeef.json b/backend/.sqlx/query-6e64c0eaa4b804424f5890468eac6b1cf4b1d811bcf6a3e2e7b0d39b321eaeef.json new file mode 100644 index 0000000000..b85baecc09 --- /dev/null +++ b/backend/.sqlx/query-6e64c0eaa4b804424f5890468eac6b1cf4b1d811bcf6a3e2e7b0d39b321eaeef.json @@ -0,0 +1,60 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT v.workspace_id AS \"workspace_id!\", v.path AS \"path!\", v.description AS \"description!\",\n v.value_expires_at AS \"value_expires_at!\", v.is_secret AS \"is_secret!\",\n ws.variable_expiration_handler->>'path' AS \"handler_path!\",\n (ws.variable_expiration_handler->'extra_args')::text::json AS \"extra_args: Json>\"\n FROM variable v\n JOIN workspace_settings ws ON ws.workspace_id = v.workspace_id\n JOIN workspace w ON w.id = v.workspace_id\n WHERE NOT w.deleted\n AND v.workspace_id <> ALL($3)\n AND v.value_expires_at IS NOT NULL\n AND v.value_expires_at IS DISTINCT FROM v.expiration_dispatched_for\n AND v.value_expires_at <= now() + ($1::text)::interval\n AND ws.variable_expiration_handler->>'path' IS NOT NULL\n AND (NOT COALESCE((ws.variable_expiration_handler->>'muted_on_user_path')::boolean, false)\n OR v.path NOT LIKE 'u/%')\n ORDER BY v.value_expires_at\n LIMIT $2", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "workspace_id!", + "type_info": "Varchar" + }, + { + "ordinal": 1, + "name": "path!", + "type_info": "Varchar" + }, + { + "ordinal": 2, + "name": "description!", + "type_info": "Varchar" + }, + { + "ordinal": 3, + "name": "value_expires_at!", + "type_info": "Timestamptz" + }, + { + "ordinal": 4, + "name": "is_secret!", + "type_info": "Bool" + }, + { + "ordinal": 5, + "name": "handler_path!", + "type_info": "Text" + }, + { + "ordinal": 6, + "name": "extra_args: Json>", + "type_info": "Json" + } + ], + "parameters": { + "Left": [ + "Text", + "Int8", + "TextArray" + ] + }, + "nullable": [ + false, + false, + false, + true, + false, + null, + null + ] + }, + "hash": "6e64c0eaa4b804424f5890468eac6b1cf4b1d811bcf6a3e2e7b0d39b321eaeef" +} diff --git a/backend/.sqlx/query-74742f785888f280bf1d7e28793150c3ddc2fd4be3b8c682f6faa7b917387593.json b/backend/.sqlx/query-74742f785888f280bf1d7e28793150c3ddc2fd4be3b8c682f6faa7b917387593.json new file mode 100644 index 0000000000..5bf83d1b33 --- /dev/null +++ b/backend/.sqlx/query-74742f785888f280bf1d7e28793150c3ddc2fd4be3b8c682f6faa7b917387593.json @@ -0,0 +1,214 @@ +{ + "db_name": "PostgreSQL", + "query": "\n SELECT\n workspace_id,\n slack_team_id,\n teams_team_id,\n teams_team_name,\n teams_team_guid,\n slack_name,\n slack_command_script,\n teams_command_script,\n slack_email,\n slack_oauth_client_id,\n slack_oauth_client_secret,\n customer_id,\n plan,\n webhook,\n ai_config,\n dbt_warehouses,\n large_file_storage,\n datatable,\n ducklake,\n git_sync,\n deploy_ui,\n default_app,\n default_scripts,\n mute_critical_alerts,\n color,\n operator_settings,\n git_app_installations,\n auto_invite,\n error_handler,\n success_handler,\n variable_expiration_handler,\n public_app_execution_limit_per_minute,\n error_handler_fallback_to_instance_alerts\n FROM\n workspace_settings\n WHERE\n workspace_id = $1\n ", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "workspace_id", + "type_info": "Varchar" + }, + { + "ordinal": 1, + "name": "slack_team_id", + "type_info": "Varchar" + }, + { + "ordinal": 2, + "name": "teams_team_id", + "type_info": "Text" + }, + { + "ordinal": 3, + "name": "teams_team_name", + "type_info": "Text" + }, + { + "ordinal": 4, + "name": "teams_team_guid", + "type_info": "Text" + }, + { + "ordinal": 5, + "name": "slack_name", + "type_info": "Varchar" + }, + { + "ordinal": 6, + "name": "slack_command_script", + "type_info": "Varchar" + }, + { + "ordinal": 7, + "name": "teams_command_script", + "type_info": "Text" + }, + { + "ordinal": 8, + "name": "slack_email", + "type_info": "Varchar" + }, + { + "ordinal": 9, + "name": "slack_oauth_client_id", + "type_info": "Varchar" + }, + { + "ordinal": 10, + "name": "slack_oauth_client_secret", + "type_info": "Varchar" + }, + { + "ordinal": 11, + "name": "customer_id", + "type_info": "Varchar" + }, + { + "ordinal": 12, + "name": "plan", + "type_info": "Varchar" + }, + { + "ordinal": 13, + "name": "webhook", + "type_info": "Text" + }, + { + "ordinal": 14, + "name": "ai_config", + "type_info": "Jsonb" + }, + { + "ordinal": 15, + "name": "dbt_warehouses", + "type_info": "Jsonb" + }, + { + "ordinal": 16, + "name": "large_file_storage", + "type_info": "Jsonb" + }, + { + "ordinal": 17, + "name": "datatable", + "type_info": "Jsonb" + }, + { + "ordinal": 18, + "name": "ducklake", + "type_info": "Jsonb" + }, + { + "ordinal": 19, + "name": "git_sync", + "type_info": "Jsonb" + }, + { + "ordinal": 20, + "name": "deploy_ui", + "type_info": "Jsonb" + }, + { + "ordinal": 21, + "name": "default_app", + "type_info": "Varchar" + }, + { + "ordinal": 22, + "name": "default_scripts", + "type_info": "Jsonb" + }, + { + "ordinal": 23, + "name": "mute_critical_alerts", + "type_info": "Bool" + }, + { + "ordinal": 24, + "name": "color", + "type_info": "Varchar" + }, + { + "ordinal": 25, + "name": "operator_settings", + "type_info": "Jsonb" + }, + { + "ordinal": 26, + "name": "git_app_installations", + "type_info": "Jsonb" + }, + { + "ordinal": 27, + "name": "auto_invite", + "type_info": "Jsonb" + }, + { + "ordinal": 28, + "name": "error_handler", + "type_info": "Jsonb" + }, + { + "ordinal": 29, + "name": "success_handler", + "type_info": "Jsonb" + }, + { + "ordinal": 30, + "name": "variable_expiration_handler", + "type_info": "Jsonb" + }, + { + "ordinal": 31, + "name": "public_app_execution_limit_per_minute", + "type_info": "Int4" + }, + { + "ordinal": 32, + "name": "error_handler_fallback_to_instance_alerts", + "type_info": "Bool" + } + ], + "parameters": { + "Left": [ + "Text" + ] + }, + "nullable": [ + false, + true, + true, + true, + true, + true, + true, + true, + false, + true, + true, + true, + true, + true, + true, + true, + true, + true, + true, + true, + true, + true, + true, + true, + true, + true, + false, + true, + true, + true, + true, + true, + false + ] + }, + "hash": "74742f785888f280bf1d7e28793150c3ddc2fd4be3b8c682f6faa7b917387593" +} diff --git a/backend/.sqlx/query-7cbf347e27200745ba7dd75726d2c9bf5ebc5c9080274b848911beadeb016bd4.json b/backend/.sqlx/query-7cbf347e27200745ba7dd75726d2c9bf5ebc5c9080274b848911beadeb016bd4.json new file mode 100644 index 0000000000..2a973f1c20 --- /dev/null +++ b/backend/.sqlx/query-7cbf347e27200745ba7dd75726d2c9bf5ebc5c9080274b848911beadeb016bd4.json @@ -0,0 +1,24 @@ +{ + "db_name": "PostgreSQL", + "query": "UPDATE variable SET expiration_dispatched_for = $3\n WHERE workspace_id = $1 AND path = $2 AND value_expires_at = $3\n AND expiration_dispatched_for IS DISTINCT FROM $3\n RETURNING 1", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "?column?", + "type_info": "Int4" + } + ], + "parameters": { + "Left": [ + "Text", + "Text", + "Timestamptz" + ] + }, + "nullable": [ + null + ] + }, + "hash": "7cbf347e27200745ba7dd75726d2c9bf5ebc5c9080274b848911beadeb016bd4" +} diff --git a/backend/.sqlx/query-8896b664bbd101675c3d6e0f279e17585a9bd50b07724a7191be98bea5632d68.json b/backend/.sqlx/query-8896b664bbd101675c3d6e0f279e17585a9bd50b07724a7191be98bea5632d68.json new file mode 100644 index 0000000000..5334e8876b --- /dev/null +++ b/backend/.sqlx/query-8896b664bbd101675c3d6e0f279e17585a9bd50b07724a7191be98bea5632d68.json @@ -0,0 +1,24 @@ +{ + "db_name": "PostgreSQL", + "query": "INSERT INTO variable\n (workspace_id, path, value, is_secret, description, account, is_oauth, expires_at, value_expires_at, labels, edited_by)\n VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11)", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Varchar", + "Varchar", + "Varchar", + "Bool", + "Varchar", + "Int4", + "Bool", + "Timestamptz", + "Timestamptz", + "TextArray", + "Varchar" + ] + }, + "nullable": [] + }, + "hash": "8896b664bbd101675c3d6e0f279e17585a9bd50b07724a7191be98bea5632d68" +} diff --git a/backend/.sqlx/query-9c2b997126090a5bc191f8d7460e14e0f95b81c61241ffd0c9d633e562c4c4f6.json b/backend/.sqlx/query-9c2b997126090a5bc191f8d7460e14e0f95b81c61241ffd0c9d633e562c4c4f6.json new file mode 100644 index 0000000000..ecb5eb2c4b --- /dev/null +++ b/backend/.sqlx/query-9c2b997126090a5bc191f8d7460e14e0f95b81c61241ffd0c9d633e562c4c4f6.json @@ -0,0 +1,14 @@ +{ + "db_name": "PostgreSQL", + "query": "UPDATE workspace_settings SET variable_expiration_handler = NULL WHERE workspace_id = $1", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Text" + ] + }, + "nullable": [] + }, + "hash": "9c2b997126090a5bc191f8d7460e14e0f95b81c61241ffd0c9d633e562c4c4f6" +} diff --git a/backend/.sqlx/query-b05ebf6f739c1f78b469161a460f5494da01209632b483c58a250f55f035d271.json b/backend/.sqlx/query-b05ebf6f739c1f78b469161a460f5494da01209632b483c58a250f55f035d271.json new file mode 100644 index 0000000000..9c9777cf83 --- /dev/null +++ b/backend/.sqlx/query-b05ebf6f739c1f78b469161a460f5494da01209632b483c58a250f55f035d271.json @@ -0,0 +1,28 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT DISTINCT v.workspace_id AS \"workspace_id!\",\n ws.variable_expiration_handler->>'path' AS \"handler_path!\"\n FROM variable v\n JOIN workspace_settings ws ON ws.workspace_id = v.workspace_id\n JOIN workspace w ON w.id = v.workspace_id\n WHERE NOT w.deleted\n AND v.value_expires_at IS NOT NULL\n AND v.value_expires_at IS DISTINCT FROM v.expiration_dispatched_for\n AND v.value_expires_at <= now() + ($1::text)::interval\n AND ws.variable_expiration_handler->>'path' IS NOT NULL", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "workspace_id!", + "type_info": "Varchar" + }, + { + "ordinal": 1, + "name": "handler_path!", + "type_info": "Text" + } + ], + "parameters": { + "Left": [ + "Text" + ] + }, + "nullable": [ + false, + null + ] + }, + "hash": "b05ebf6f739c1f78b469161a460f5494da01209632b483c58a250f55f035d271" +} diff --git a/backend/.sqlx/query-bc6ebef9d41aba232f115f95404922c6054df01fa7f38f5d15d6d4af6c726a3c.json b/backend/.sqlx/query-de0d78bf28a5b6457e5568dce9823a4c216b4be67b68241461b42326d9d65256.json similarity index 74% rename from backend/.sqlx/query-bc6ebef9d41aba232f115f95404922c6054df01fa7f38f5d15d6d4af6c726a3c.json rename to backend/.sqlx/query-de0d78bf28a5b6457e5568dce9823a4c216b4be67b68241461b42326d9d65256.json index d77682a4ea..e2145fc10a 100644 --- a/backend/.sqlx/query-bc6ebef9d41aba232f115f95404922c6054df01fa7f38f5d15d6d4af6c726a3c.json +++ b/backend/.sqlx/query-de0d78bf28a5b6457e5568dce9823a4c216b4be67b68241461b42326d9d65256.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "SELECT COUNT(*) FROM group_ WHERE name != 'all' AND name != 'error_handler' AND name != 'slack' AND name != 'wm_deployers'", + "query": "SELECT COUNT(*) FROM group_ WHERE name != 'all' AND name != 'error_handler' AND name != 'slack' AND name != 'wm_deployers' AND name != 'variable_expiration_handler'", "describe": { "columns": [ { @@ -16,5 +16,5 @@ null ] }, - "hash": "bc6ebef9d41aba232f115f95404922c6054df01fa7f38f5d15d6d4af6c726a3c" + "hash": "de0d78bf28a5b6457e5568dce9823a4c216b4be67b68241461b42326d9d65256" } diff --git a/backend/.sqlx/query-e8eaa80631362dee4d696e41abf220cb6f8346d2bfcd901371b480a16a70109c.json b/backend/.sqlx/query-e8eaa80631362dee4d696e41abf220cb6f8346d2bfcd901371b480a16a70109c.json new file mode 100644 index 0000000000..b3dd32073d --- /dev/null +++ b/backend/.sqlx/query-e8eaa80631362dee4d696e41abf220cb6f8346d2bfcd901371b480a16a70109c.json @@ -0,0 +1,15 @@ +{ + "db_name": "PostgreSQL", + "query": "UPDATE workspace_settings SET variable_expiration_handler = $1 WHERE workspace_id = $2", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Jsonb", + "Text" + ] + }, + "nullable": [] + }, + "hash": "e8eaa80631362dee4d696e41abf220cb6f8346d2bfcd901371b480a16a70109c" +} diff --git a/backend/.sqlx/query-eb3875d80f559fa45df82010bd2bff830a4f95f58372fd63ebae26297e644c40.json b/backend/.sqlx/query-eb3875d80f559fa45df82010bd2bff830a4f95f58372fd63ebae26297e644c40.json new file mode 100644 index 0000000000..8638192347 --- /dev/null +++ b/backend/.sqlx/query-eb3875d80f559fa45df82010bd2bff830a4f95f58372fd63ebae26297e644c40.json @@ -0,0 +1,12 @@ +{ + "db_name": "PostgreSQL", + "query": "UPDATE variable SET value_expires_at = '2029-12-31T00:00:00Z'\n WHERE workspace_id = 'wm-fork-expiry' AND path = 'u/admin/expiring'", + "describe": { + "columns": [], + "parameters": { + "Left": [] + }, + "nullable": [] + }, + "hash": "eb3875d80f559fa45df82010bd2bff830a4f95f58372fd63ebae26297e644c40" +} diff --git a/backend/.sqlx/query-ee8b92d51f220cac5a32f0edcb3c4effa302380d4b434d8c74677bfbf895333c.json b/backend/.sqlx/query-ee8b92d51f220cac5a32f0edcb3c4effa302380d4b434d8c74677bfbf895333c.json new file mode 100644 index 0000000000..d101f5ec0a --- /dev/null +++ b/backend/.sqlx/query-ee8b92d51f220cac5a32f0edcb3c4effa302380d4b434d8c74677bfbf895333c.json @@ -0,0 +1,12 @@ +{ + "db_name": "PostgreSQL", + "query": "INSERT INTO workspace_diff\n (source_workspace_id, fork_workspace_id, path, kind, ahead, behind, has_changes)\n VALUES ('test-workspace', 'wm-fork-expiry', 'u/admin/expiring', 'variable', 0, 1, NULL)", + "describe": { + "columns": [], + "parameters": { + "Left": [] + }, + "nullable": [] + }, + "hash": "ee8b92d51f220cac5a32f0edcb3c4effa302380d4b434d8c74677bfbf895333c" +} diff --git a/backend/ee-repo-ref.txt b/backend/ee-repo-ref.txt index 4767454f25..f552cae039 100644 --- a/backend/ee-repo-ref.txt +++ b/backend/ee-repo-ref.txt @@ -1 +1 @@ -b58ad414b098d3d7787001a352bfbb13e43a335f +a4934a3213b598f77a35cdb3d774c45ad125e453 diff --git a/backend/migrations/20260824180615_variable_value_expires_at.down.sql b/backend/migrations/20260824180615_variable_value_expires_at.down.sql new file mode 100644 index 0000000000..fe2810cb44 --- /dev/null +++ b/backend/migrations/20260824180615_variable_value_expires_at.down.sql @@ -0,0 +1,7 @@ +DROP INDEX IF EXISTS idx_variable_expiration_due; + +COMMENT ON COLUMN variable.expires_at IS NULL; + +ALTER TABLE variable + DROP COLUMN IF EXISTS value_expires_at, + DROP COLUMN IF EXISTS expiration_dispatched_for; diff --git a/backend/migrations/20260824180615_variable_value_expires_at.up.sql b/backend/migrations/20260824180615_variable_value_expires_at.up.sql new file mode 100644 index 0000000000..078cea0608 --- /dev/null +++ b/backend/migrations/20260824180615_variable_value_expires_at.up.sql @@ -0,0 +1,20 @@ +-- `value_expires_at` is when the value stored in the variable stops working; the row itself +-- lives on. `expires_at` keeps its unrelated meaning — delete the row — so no existing writer +-- changes meaning under a rolling deploy and no data is rewritten here. +ALTER TABLE variable ADD COLUMN IF NOT EXISTS value_expires_at TIMESTAMP WITH TIME ZONE; + +COMMENT ON COLUMN variable.expires_at IS + 'Garbage collection deadline: the row is deleted once passed. Set by the ephemeral secret_arg mint. Not the value''s expiry — see value_expires_at.'; +COMMENT ON COLUMN variable.value_expires_at IS + 'When the value stops working. Drives the workspace variable expiration handler; never deletes the row.'; + +-- Holds the `value_expires_at` that was dispatched, rather than a plain "was dispatched" flag, +-- so that moving the date re-arms the handler by construction — no writer has to remember to +-- clear anything. +ALTER TABLE variable + ADD COLUMN IF NOT EXISTS expiration_dispatched_for TIMESTAMP WITH TIME ZONE; + +-- Serves the sweep's due predicate. Partial so it stays the size of the opted-in set rather +-- than the whole variable table. +CREATE INDEX IF NOT EXISTS idx_variable_expiration_due ON variable (value_expires_at) + WHERE value_expires_at IS NOT NULL AND value_expires_at IS DISTINCT FROM expiration_dispatched_for; diff --git a/backend/migrations/20260824180616_workspace_variable_expiration_handler.down.sql b/backend/migrations/20260824180616_workspace_variable_expiration_handler.down.sql new file mode 100644 index 0000000000..6b2b27ff0a --- /dev/null +++ b/backend/migrations/20260824180616_workspace_variable_expiration_handler.down.sql @@ -0,0 +1 @@ +ALTER TABLE workspace_settings DROP COLUMN IF EXISTS variable_expiration_handler; diff --git a/backend/migrations/20260824180616_workspace_variable_expiration_handler.up.sql b/backend/migrations/20260824180616_workspace_variable_expiration_handler.up.sql new file mode 100644 index 0000000000..80f7552651 --- /dev/null +++ b/backend/migrations/20260824180616_workspace_variable_expiration_handler.up.sql @@ -0,0 +1,3 @@ +-- Shape mirrors `error_handler` / `success_handler` as grouped by 20260124172000: +-- {"path": ..., "extra_args": ..., "muted_on_user_path": ...} +ALTER TABLE workspace_settings ADD COLUMN IF NOT EXISTS variable_expiration_handler JSONB; diff --git a/backend/src/monitor.rs b/backend/src/monitor.rs index 430f3659c3..00bf26a558 100644 --- a/backend/src/monitor.rs +++ b/backend/src/monitor.rs @@ -115,6 +115,7 @@ use windmill_object_store::reload_object_store_setting; use windmill_queue::{ cancel_job, get_queued_job_v2, schedule::{find_unarmed_schedules, rearm_schedule, RearmOutcome}, + variable_expiration::dispatch_expiring_variables, SameWorkerPayload, }; use windmill_store::resources::MAX_RESOURCE_VERSIONS; @@ -4007,6 +4008,16 @@ pub async fn monitor_db( } }; + // Run every 5 minutes (10 iterations * 30s). The lead time is an hour, so a variable is + // picked up comfortably inside its window without polling hot. + let dispatch_expiring_variables_f = async { + if server_mode && iteration.is_some() && iteration.as_ref().unwrap().should_run(10) { + if let Some(db) = conn.as_sql() { + dispatch_expiring_variables(&db).await; + } + } + }; + // run every hour (120 iterations * 30s = 3600s) let cleanup_stale_server_heartbeats_f = async { if server_mode && iteration.is_some() && iteration.as_ref().unwrap().should_run(120) { @@ -4131,6 +4142,7 @@ pub async fn monitor_db( native_triggers_sync_f, cleanup_notify_events_f, check_expiring_tokens_f, + dispatch_expiring_variables_f, cleanup_stale_server_heartbeats_f, manage_audit_partitions_f, export_audit_logs_to_object_store_f, diff --git a/backend/summarized_schema.txt b/backend/summarized_schema.txt index a9865cebd3..7a2996a9b1 100644 --- a/backend/summarized_schema.txt +++ b/backend/summarized_schema.txt @@ -196,7 +196,7 @@ v2_job_runtime: id(uuid), ping(ts), memory_peak(int) FK: (id) -> v2_job_queue(id) v2_job_status: id(uuid), flow_status(jsonb), flow_leaf_jobs(jsonb), workflow_as_code_status(jsonb) FK: (id) -> v2_job_queue(id) -variable: workspace_id(char), path(char), value(char), is_secret(bool), description(char), extra_perms(jsonb), account(int), is_oauth(bool), expires_at(ts), labels(text[]) +variable: workspace_id(char), path(char), value(char), is_secret(bool), description(char), extra_perms(jsonb), account(int), is_oauth(bool), expires_at(ts), labels(text[]), value_expires_at(timestamptz), expiration_dispatched_for(timestamptz) FK: (workspace_id) -> workspace(id) websocket_trigger: path(char), url(char), script_path(char), is_flow(bool), workspace_id(char), edited_by(char), email(char), edited_at(ts), extra_perms(jsonb), server_id(char), last_server_ping(ts), error(text), filters(jsonb[]), initial_messages(jsonb[]), url_runnable_args(jsonb), can_return_message(bool), error_handler_path(char), error_handler_args(jsonb), retry(jsonb), can_return_error_result(bool), mode(trigger_mode), labels(text[]) windmill_migrations: name(text), created_at(ts) @@ -220,7 +220,7 @@ workspace_protection_rule: workspace_id(char), name(char), rules(int), bypass_gr FK: (workspace_id) -> workspace(id) workspace_runnable_dependencies: flow_path(char), runnable_path(char), script_hash(bigint), runnable_is_flow(bool), workspace_id(char), app_path(char) FK: (app_path, workspace_id) -> app(path, workspace_id) | (flow_path, workspace_id) -> flow(path, workspace_id) -workspace_settings: workspace_id(char), slack_team_id(char), slack_name(char), slack_command_script(char), slack_email(char), customer_id(char), plan(char), webhook(text), ai_config(jsonb), large_file_storage(jsonb), git_sync(jsonb), default_app(char), default_scripts(jsonb), deploy_ui(jsonb), mute_critical_alerts(bool), color(char), operator_settings(jsonb), teams_command_script(text), teams_team_id(text), teams_team_name(text), git_app_installations(jsonb), ducklake(jsonb), slack_oauth_client_id(char), slack_oauth_client_secret(char), datatable(jsonb), teams_team_guid(text), auto_invite(jsonb), error_handler(jsonb), success_handler(jsonb), public_app_execution_limit_per_minute(int), dbt_warehouses(jsonb) +workspace_settings: workspace_id(char), slack_team_id(char), slack_name(char), slack_command_script(char), slack_email(char), customer_id(char), plan(char), webhook(text), ai_config(jsonb), large_file_storage(jsonb), git_sync(jsonb), default_app(char), default_scripts(jsonb), deploy_ui(jsonb), mute_critical_alerts(bool), color(char), operator_settings(jsonb), teams_command_script(text), teams_team_id(text), teams_team_name(text), git_app_installations(jsonb), ducklake(jsonb), slack_oauth_client_id(char), slack_oauth_client_secret(char), datatable(jsonb), teams_team_guid(text), auto_invite(jsonb), error_handler(jsonb), success_handler(jsonb), variable_expiration_handler(jsonb), public_app_execution_limit_per_minute(int), dbt_warehouses(jsonb) FK: (workspace_id) -> workspace(id) zombie_job_counter: job_id(uuid), counter(int) FK: (job_id) -> v2_job(id) diff --git a/backend/tests/variable_expiration.rs b/backend/tests/variable_expiration.rs new file mode 100644 index 0000000000..aa58f71085 --- /dev/null +++ b/backend/tests/variable_expiration.rs @@ -0,0 +1,366 @@ +//! Tests for the variable expiration sweep (`windmill_queue::variable_expiration`). +//! +//! `dispatch_expiring_variables` is called directly against seeded `variable` / +//! `workspace_settings` rows — no worker is needed, since the sweep's job ends at the push +//! and the queued job is itself part of the assertions. + +use sqlx::{Pool, Postgres}; +use windmill_queue::variable_expiration::dispatch_expiring_variables; +use windmill_test_utils::{initialize_tracing, ApiServer}; + +const WS: &str = "test-workspace"; + +async fn configure_handler(db: &Pool, handler: &str) -> anyhow::Result<()> { + sqlx::query("UPDATE workspace_settings SET variable_expiration_handler = $2::jsonb WHERE workspace_id = $1") + .bind(WS) + .bind(handler) + .execute(db) + .await?; + Ok(()) +} + +/// Seed a deployed script the handler setting can point at. +async fn seed_handler_script(db: &Pool, path: &str) -> anyhow::Result<()> { + sqlx::query( + r#"INSERT INTO script (workspace_id, hash, path, summary, description, content, created_by, language, lock) + VALUES ($1, $2, $3, '', '', 'export function main() {}', 'test-user', 'bun', '')"#, + ) + .bind(WS) + .bind(path.bytes().fold(0i64, |h, b| h.wrapping_mul(31).wrapping_add(b as i64))) + .bind(path) + .execute(db) + .await?; + Ok(()) +} + +async fn seed_variable(db: &Pool, path: &str, expires_in: &str) -> anyhow::Result<()> { + sqlx::query(&format!( + r#"INSERT INTO variable (workspace_id, path, value, is_secret, description, value_expires_at) + VALUES ($1, $2, 'MUST_NOT_LEAK', true, 'a credential', now() + interval '{expires_in}')"# + )) + .bind(WS) + .bind(path) + .execute(db) + .await?; + Ok(()) +} + +async fn queued_handler_jobs(db: &Pool, path: &str) -> anyhow::Result { + Ok( + sqlx::query_scalar::<_, i64>("SELECT count(*) FROM v2_job WHERE runnable_path = $1") + .bind(path) + .fetch_one(db) + .await?, + ) +} + +/// The sweep dispatches inside the lead window, leaves everything outside it alone, and +/// does not fire twice for the same expiry. +#[sqlx::test(fixtures("base"))] +async fn test_dispatch_within_lead_time_once(db: Pool) -> anyhow::Result<()> { + initialize_tracing().await; + seed_handler_script(&db, "u/test-user/expiry_handler").await?; + configure_handler(&db, r#"{"path": "script/u/test-user/expiry_handler"}"#).await?; + + seed_variable(&db, "u/test-user/due", "30 minutes").await?; + seed_variable(&db, "u/test-user/far", "10 days").await?; + + dispatch_expiring_variables(&db).await; + + assert_eq!( + queued_handler_jobs(&db, "u/test-user/expiry_handler").await?, + 1, + "exactly the variable inside the lead window should dispatch" + ); + + // The identity is load-bearing, not incidental: the settings tab tells admins to grant + // `g/variable_expiration_handler` access to the secrets it rotates, so the job must run as + // that group and under its own email — which is also what keeps cloud metering off the + // configuring user's quota. + let (path, is_secret, permissioned_as, permissioned_as_email): (String, bool, String, String) = + sqlx::query_as( + "SELECT args->>'variable_path', (args->>'is_secret')::boolean, permissioned_as, + permissioned_as_email FROM v2_job WHERE runnable_path = $1", + ) + .bind("u/test-user/expiry_handler") + .fetch_one(&db) + .await?; + assert_eq!(path, "u/test-user/due"); + assert!(is_secret); + assert_eq!(permissioned_as, "g/variable_expiration_handler"); + assert_eq!( + permissioned_as_email, + "variable_expiration_handler@windmill.dev" + ); + + // Job args are stored in cleartext and shown in Runs, so the value must never be in them. + let args: String = sqlx::query_scalar("SELECT args::text FROM v2_job WHERE runnable_path = $1") + .bind("u/test-user/expiry_handler") + .fetch_one(&db) + .await?; + assert!( + !args.contains("MUST_NOT_LEAK"), + "handler args leaked the value: {args}" + ); + + // A second pass must not re-dispatch: the claim is what makes replicas ticking + // concurrently safe. + dispatch_expiring_variables(&db).await; + assert_eq!( + queued_handler_jobs(&db, "u/test-user/expiry_handler").await?, + 1, + "a dispatched variable must not fire again" + ); + + Ok(()) +} + +/// Moving `value_expires_at` re-arms the variable; re-writing the same date does not. +#[sqlx::test(fixtures("base"))] +async fn test_rearm_on_date_change(db: Pool) -> anyhow::Result<()> { + initialize_tracing().await; + let server = ApiServer::start(db.clone()).await?; + let base_url = format!("http://localhost:{}", server.addr.port()); + seed_handler_script(&db, "u/test-user/expiry_handler").await?; + configure_handler(&db, r#"{"path": "script/u/test-user/expiry_handler"}"#).await?; + seed_variable(&db, "u/test-user/due", "30 minutes").await?; + + // Driven through the real endpoint: what re-arms is the stored date differing from + // `expiration_dispatched_for`, so a test writing that column itself would assert nothing. + let update = async |body: serde_json::Value| -> anyhow::Result { + Ok(reqwest::Client::new() + .post(format!( + "{base_url}/api/w/{WS}/variables/update/u/test-user/due" + )) + .bearer_auth("SECRET_TOKEN") + .json(&body) + .send() + .await? + .status() + .as_u16()) + }; + + dispatch_expiring_variables(&db).await; + assert_eq!( + queued_handler_jobs(&db, "u/test-user/expiry_handler").await?, + 1 + ); + + let stored: chrono::DateTime = sqlx::query_scalar( + "SELECT value_expires_at FROM variable WHERE workspace_id = $1 AND path = $2", + ) + .bind(WS) + .bind("u/test-user/due") + .fetch_one(&db) + .await?; + + // Without this, a `wmill sync push` of an unchanged spec re-fires the handler every push. + assert_eq!( + update(serde_json::json!({ "value_expires_at": stored })).await?, + 200 + ); + dispatch_expiring_variables(&db).await; + assert_eq!( + queued_handler_jobs(&db, "u/test-user/expiry_handler").await?, + 1, + "an unchanged expiry must not re-arm" + ); + + assert_eq!( + update(serde_json::json!({ "description": "touched" })).await?, + 200 + ); + dispatch_expiring_variables(&db).await; + assert_eq!( + queued_handler_jobs(&db, "u/test-user/expiry_handler").await?, + 1, + "an update that omits value_expires_at must not re-arm" + ); + + assert_eq!( + update(serde_json::json!({ "value_expires_at": stored + chrono::Duration::minutes(5) })) + .await?, + 200 + ); + dispatch_expiring_variables(&db).await; + assert_eq!( + queued_handler_jobs(&db, "u/test-user/expiry_handler").await?, + 2, + "a moved expiry must re-arm the handler" + ); + + Ok(()) +} + +/// With no handler configured nothing is dispatched, and `muted_on_user_path` keeps +/// personal-path variables out — mirroring `error_handler_muted_on_user_path`. +#[sqlx::test(fixtures("base"))] +async fn test_skips_unconfigured_and_muted_user_paths(db: Pool) -> anyhow::Result<()> { + initialize_tracing().await; + seed_handler_script(&db, "u/test-user/expiry_handler").await?; + seed_variable(&db, "u/test-user/personal", "30 minutes").await?; + seed_variable(&db, "f/test_folder/shared", "30 minutes").await?; + + // No handler configured yet. + dispatch_expiring_variables(&db).await; + assert_eq!( + queued_handler_jobs(&db, "u/test-user/expiry_handler").await?, + 0 + ); + + configure_handler( + &db, + r#"{"path": "script/u/test-user/expiry_handler", "muted_on_user_path": true}"#, + ) + .await?; + dispatch_expiring_variables(&db).await; + + let dispatched: Vec = + sqlx::query_scalar("SELECT args->>'variable_path' FROM v2_job WHERE runnable_path = $1") + .bind("u/test-user/expiry_handler") + .fetch_all(&db) + .await?; + assert_eq!( + dispatched, + vec!["f/test_folder/shared".to_string()], + "a muted personal path must not dispatch" + ); + + Ok(()) +} + +/// A workspace whose handler cannot be resolved must not consume the per-pass cap. Its +/// variables are the most overdue on the instance, so without the back-off they refill the +/// limit on every pass and no other workspace ever dispatches. +#[sqlx::test(fixtures("base"))] +async fn test_broken_handler_does_not_starve_other_workspaces( + db: Pool, +) -> anyhow::Result<()> { + initialize_tracing().await; + seed_handler_script(&db, "u/test-user/expiry_handler").await?; + configure_handler(&db, r#"{"path": "script/u/test-user/expiry_handler"}"#).await?; + seed_variable(&db, "u/test-user/due", "30 minutes").await?; + + sqlx::query( + "INSERT INTO workspace (id, name, owner) VALUES ('poison-ws', 'poison-ws', 'test-user')", + ) + .execute(&db) + .await?; + sqlx::query( + "INSERT INTO workspace_key (workspace_id, kind, key) VALUES ('poison-ws', 'cloud', 'k')", + ) + .execute(&db) + .await?; + sqlx::query( + r#"INSERT INTO workspace_settings (workspace_id, variable_expiration_handler) + VALUES ('poison-ws', '{"path": "script/u/test-user/does_not_exist"}'::jsonb)"#, + ) + .execute(&db) + .await?; + // More overdue rows than one pass can carry, so they alone would fill the cap. + sqlx::query( + r#"INSERT INTO variable (workspace_id, path, value, is_secret, description, value_expires_at) + SELECT 'poison-ws', 'f/creds/v' || i, 'v', false, '', now() - (i || ' minutes')::interval + FROM generate_series(1, 120) i"#, + ) + .execute(&db) + .await?; + + // First pass reaches only the poison workspace; the second finds it backing off. + dispatch_expiring_variables(&db).await; + dispatch_expiring_variables(&db).await; + + assert_eq!( + queued_handler_jobs(&db, "u/test-user/expiry_handler").await?, + 1, + "a workspace with an unresolvable handler must not starve the rest of the instance" + ); + assert_eq!( + queued_handler_jobs(&db, "u/test-user/does_not_exist").await?, + 0 + ); + + Ok(()) +} + +/// Pointing the workspace at another handler ends the back-off the broken one earned. +#[sqlx::test(fixtures("base"))] +async fn test_repointing_the_handler_ends_the_backoff(db: Pool) -> anyhow::Result<()> { + initialize_tracing().await; + seed_handler_script(&db, "u/test-user/expiry_handler").await?; + configure_handler(&db, r#"{"path": "script/u/test-user/does_not_exist"}"#).await?; + seed_variable(&db, "u/test-user/due", "30 minutes").await?; + + // Earns a cooldown, which the next pass would otherwise spend skipping the workspace. + dispatch_expiring_variables(&db).await; + configure_handler(&db, r#"{"path": "script/u/test-user/expiry_handler"}"#).await?; + dispatch_expiring_variables(&db).await; + + assert_eq!( + queued_handler_jobs(&db, "u/test-user/expiry_handler").await?, + 1, + "a repointed handler must be tried on the very next pass" + ); + + Ok(()) +} + +/// The endpoint that configures the handler: admin-only, prefix-checked, and clearing by +/// omission — which is what `wmill sync push` relies on to propagate an absent handler. +#[sqlx::test(fixtures("base"))] +async fn test_edit_handler_endpoint(db: Pool) -> anyhow::Result<()> { + initialize_tracing().await; + let server = ApiServer::start(db.clone()).await?; + let base_url = format!("http://localhost:{}", server.addr.port()); + + let edit = async |token: &str, body: serde_json::Value| -> anyhow::Result { + Ok(reqwest::Client::new() + .post(format!( + "{base_url}/api/w/{WS}/workspaces/edit_variable_expiration_handler" + )) + .bearer_auth(token) + .json(&body) + .send() + .await? + .status() + .as_u16()) + }; + let stored = async || -> anyhow::Result> { + Ok(sqlx::query_scalar( + "SELECT variable_expiration_handler FROM workspace_settings WHERE workspace_id = $1", + ) + .bind(WS) + .fetch_one(&db) + .await?) + }; + + let handler = serde_json::json!({ "path": "script/u/test-user/expiry_handler" }); + assert_eq!(edit("SECRET_TOKEN_2", handler.clone()).await?, 403); + assert!(stored().await?.is_none()); + + // A path the sweep could never resolve to a runnable is rejected up front. + assert_eq!( + edit( + "SECRET_TOKEN", + serde_json::json!({ "path": "u/test-user/expiry_handler" }) + ) + .await?, + 400 + ); + + assert_eq!(edit("SECRET_TOKEN", handler).await?, 200); + assert_eq!( + stored() + .await? + .and_then(|h| h["path"].as_str().map(String::from)), + Some("script/u/test-user/expiry_handler".to_string()) + ); + + assert_eq!(edit("SECRET_TOKEN", serde_json::json!({})).await?, 200); + assert!( + stored().await?.is_none(), + "an omitted path clears the handler" + ); + + Ok(()) +} diff --git a/backend/tests/workspace_export.rs b/backend/tests/workspace_export.rs index 5894ca8394..e405690522 100644 --- a/backend/tests/workspace_export.rs +++ b/backend/tests/workspace_export.rs @@ -288,6 +288,49 @@ async fn test_tarball_export_gates_values_on_item_scopes(db: Pool) -> Ok(()) } +/// The tarball filter keys on `expires_at` (garbage collection) and must ignore +/// `value_expires_at`, so an expiry set on a variable does not drop it from `wmill sync +/// pull`. The exported entry has to carry the date back, or a pull/push round-trip clears it. +#[sqlx::test(fixtures("base"))] +async fn test_tarball_export_excludes_only_gc_variables(db: Pool) -> anyhow::Result<()> { + initialize_tracing().await; + let server = ApiServer::start(db.clone()).await?; + let base_url = format!("http://localhost:{}", server.addr.port()); + + sqlx::query( + r#"INSERT INTO variable (workspace_id, path, value, is_secret, description, expires_at, value_expires_at) + VALUES ('test-workspace', 'u/test-user/expiring', 'v', false, '', NULL, now() + interval '30 days'), + ('test-workspace', 'u/test-user/ephemeral', 'v', false, '', now() + interval '7 days', NULL)"#, + ) + .execute(&db) + .await?; + + let resp = reqwest::Client::new() + .get(format!( + "{base_url}/api/w/test-workspace/workspaces/tarball" + )) + .bearer_auth("SECRET_TOKEN") + .send() + .await?; + assert_eq!(resp.status().as_u16(), 200); + let body = String::from_utf8_lossy(&resp.bytes().await?).into_owned(); + + assert!( + body.contains("u/test-user/expiring.variable.json"), + "a variable with an expiry must still reach git sync" + ); + assert!( + body.contains("value_expires_at"), + "the exported entry must carry the expiry back" + ); + assert!( + !body.contains("u/test-user/ephemeral.variable.json"), + "a variable awaiting garbage collection must stay out of the export" + ); + + Ok(()) +} + /// `settings.json` carries the admin-managed integration config that `get_settings` /// is admin-only for (the webhook URL, ai_config, git_sync, handler extra_args), so /// `include_settings` takes the same admin check as `get_settings` rather than diff --git a/backend/windmill-api-groups/src/groups.rs b/backend/windmill-api-groups/src/groups.rs index a15b9823d3..c3719cd608 100644 --- a/backend/windmill-api-groups/src/groups.rs +++ b/backend/windmill-api-groups/src/groups.rs @@ -215,12 +215,15 @@ pub async fn require_is_owner( } async fn _check_nb_of_groups(db: &DB) -> Result<()> { - let nb_groups = sqlx::query_scalar!("SELECT COUNT(*) FROM group_ WHERE name != 'all' AND name != 'error_handler' AND name != 'slack' AND name != 'wm_deployers'",) + // The exempt names are enumerated, not derived from how the group was created: these back + // workspace settings rather than the user's own org structure, and the limit prices the + // latter. `success_handler` is provisioned the same way but is not in the list. + let nb_groups = sqlx::query_scalar!("SELECT COUNT(*) FROM group_ WHERE name != 'all' AND name != 'error_handler' AND name != 'slack' AND name != 'wm_deployers' AND name != 'variable_expiration_handler'",) .fetch_one(db) .await?; if nb_groups.unwrap_or(0) >= 3 { return Err(Error::BadRequest( - "You have reached the maximum number of groups (3 outside of native groups 'all', 'slack', 'error_handler' and 'wm_deployers') without an enterprise license" + "You have reached the maximum number of groups (3 outside of native groups 'all', 'slack', 'error_handler', 'wm_deployers' and 'variable_expiration_handler') without an enterprise license" .to_string(), )); } diff --git a/backend/windmill-api-integration-tests/tests/variables.rs b/backend/windmill-api-integration-tests/tests/variables.rs index e5018f4f97..a307db961c 100644 --- a/backend/windmill-api-integration-tests/tests/variables.rs +++ b/backend/windmill-api-integration-tests/tests/variables.rs @@ -338,3 +338,43 @@ async fn test_variables_are_cached(db: Pool) -> anyhow::Result<()> { Ok(()) } + +/// A draft-only row is synthesized field by field from the editor's `VariableState`, +/// so a field the editor stores but the synthesis forgets is dropped silently. +#[sqlx::test(migrations = "../migrations", fixtures("base"))] +async fn test_draft_only_row_carries_value_expires_at(db: Pool) -> anyhow::Result<()> { + initialize_tracing().await; + let server = ApiServer::start(db.clone()).await?; + let port = server.addr.port(); + + authed(client().post(format!( + "http://localhost:{port}/api/w/test-workspace/drafts/update/variable/u/test-user/draft_var" + ))) + .json(&json!({ + "value": { + "path": "u/test-user/draft_var", + "variable": { "value": "x", "is_secret": true, "description": "" }, + "wsSpecific": false, + "value_expires_at": "2027-03-15T08:00:00Z" + } + })) + .send() + .await?; + + let resp = authed(client().get(format!( + "http://localhost:{port}/api/w/test-workspace/variables/list?include_draft_only=true" + ))) + .send() + .await?; + assert_eq!(resp.status(), 200); + let rows = resp.json::>().await?; + + let row = rows + .iter() + .find(|v| v["path"] == "u/test-user/draft_var") + .expect("draft-only row is listed"); + assert_eq!(row["draft_only"], true); + assert_eq!(row["value_expires_at"], "2027-03-15T08:00:00Z"); + + Ok(()) +} diff --git a/backend/windmill-api-integration-tests/tests/workspace_comparison.rs b/backend/windmill-api-integration-tests/tests/workspace_comparison.rs index 11c731d75c..716873f9cd 100644 --- a/backend/windmill-api-integration-tests/tests/workspace_comparison.rs +++ b/backend/windmill-api-integration-tests/tests/workspace_comparison.rs @@ -2537,7 +2537,10 @@ async fn test_rename_records_the_vacated_path(db: Pool) -> anyhow::Res .await?; let status = resp.status(); let hash = resp.text().await?; - assert!(status.is_success(), "script deploy failed: {status} — {hash}"); + assert!( + status.is_success(), + "script deploy failed: {status} — {hash}" + ); parent_hash = Some(hash.trim().trim_matches('"').to_string()); } let mut vacated = None; @@ -2560,3 +2563,88 @@ async fn test_rename_records_the_vacated_path(db: Pool) -> anyhow::Res Ok(()) } + +/// `compare_two_variables` hand-lists the columns it reads, so a column the variable +/// carries but the comparison forgets makes that difference invisible in the fork diff +/// — and an invisible difference can never be promoted. +#[sqlx::test(migrations = "../migrations", fixtures("base"))] +async fn test_compare_variables_detects_value_expires_at(db: Pool) -> anyhow::Result<()> { + initialize_tracing().await; + + let server = ApiServer::start(db.clone()).await?; + let port = server.addr.port(); + let base_url = format!("http://localhost:{port}/api"); + let admin = windmill_api_client::create_client( + &format!("http://localhost:{port}"), + "SECRET_TOKEN".to_string(), + ); + + sqlx::query!( + "INSERT INTO variable (workspace_id, path, value, is_secret, description, value_expires_at) + VALUES ('test-workspace', 'u/admin/expiring', 'v', false, 'same', '2027-01-01T00:00:00Z')" + ) + .execute(&db) + .await?; + + let resp = admin + .client() + .post(&format!( + "{base_url}/w/test-workspace/workspaces/create_fork" + )) + .json(&json!({ "id": "wm-fork-expiry", "name": "Expiry Fork" })) + .send() + .await?; + assert!( + resp.status().is_success(), + "fork creation failed: {}", + resp.status() + ); + + // Asserted before the UPDATE below overwrites it: the fork has to inherit the date from + // the source. Without this, a `clone_variables` that dropped the column would leave NULL + // here, the UPDATE would set 2029 anyway, and the comparison would still differ. + let inherited: Option = sqlx::query_scalar( + "SELECT to_char(value_expires_at, 'YYYY-MM-DD') FROM variable + WHERE workspace_id = 'wm-fork-expiry' AND path = 'u/admin/expiring'", + ) + .fetch_one(&db) + .await?; + assert_eq!(inherited.as_deref(), Some("2027-01-01")); + + // Only the expiry differs — value, is_secret and description stay identical, so a + // comparison that skips the column sees two equal variables. + sqlx::query!( + "UPDATE variable SET value_expires_at = '2029-12-31T00:00:00Z' + WHERE workspace_id = 'wm-fork-expiry' AND path = 'u/admin/expiring'" + ) + .execute(&db) + .await?; + + // `has_changes = NULL` makes the endpoint re-evaluate the pair instead of trusting + // a cached verdict. + sqlx::query!( + "INSERT INTO workspace_diff + (source_workspace_id, fork_workspace_id, path, kind, ahead, behind, has_changes) + VALUES ('test-workspace', 'wm-fork-expiry', 'u/admin/expiring', 'variable', 0, 1, NULL)" + ) + .execute(&db) + .await?; + + let comparison = admin + .client() + .get(&format!( + "{base_url}/w/test-workspace/workspaces/compare/wm-fork-expiry" + )) + .send() + .await? + .json::() + .await?; + + assert_eq!( + comparison["summary"]["variables_changed"].as_u64(), + Some(1), + "an expiry-only change must show as deployable: {comparison}" + ); + + Ok(()) +} diff --git a/backend/windmill-api-workspaces/src/workspaces.rs b/backend/windmill-api-workspaces/src/workspaces.rs index 3b3ea2713c..544a0c11b9 100644 --- a/backend/windmill-api-workspaces/src/workspaces.rs +++ b/backend/windmill-api-workspaces/src/workspaces.rs @@ -119,6 +119,10 @@ pub fn workspaced_service() -> Router { .route("/is_premium", get(is_premium)) .route("/edit_error_handler", post(edit_error_handler)) .route("/edit_success_handler", post(edit_success_handler)) + .route( + "/edit_variable_expiration_handler", + post(edit_variable_expiration_handler), + ) .route( "/edit_large_file_storage_config", post(edit_large_file_storage_config), @@ -313,6 +317,8 @@ pub struct WorkspaceSettings { #[serde(skip_serializing_if = "Option::is_none")] pub success_handler: Option, #[serde(skip_serializing_if = "Option::is_none")] + pub variable_expiration_handler: Option, + #[serde(skip_serializing_if = "Option::is_none")] pub public_app_execution_limit_per_minute: Option, pub error_handler_fallback_to_instance_alerts: bool, } @@ -1028,6 +1034,7 @@ async fn get_settings( auto_invite, error_handler, success_handler, + variable_expiration_handler, public_app_execution_limit_per_minute, error_handler_fallback_to_instance_alerts FROM @@ -4614,6 +4621,122 @@ async fn edit_error_handler( Ok(format!("Edit error_handler for workspace {}", &w_id)) } +#[derive(Deserialize)] +#[serde(deny_unknown_fields)] +pub struct EditVariableExpirationHandler { + pub path: Option, + pub extra_args: Option, + #[serde(default)] + pub muted_on_user_path: bool, +} + +async fn edit_variable_expiration_handler( + authed: ApiAuthed, + Extension(db): Extension, + Path(w_id): Path, + ApiAuthed { is_admin, username, .. }: ApiAuthed, + Json(es): Json, +) -> Result { + require_admin(is_admin, &username)?; + + let mut tx = db.begin().await?; + + // The group starts empty of permissions beyond its creator: rotating a variable requires + // an admin to grant it write access to that variable explicitly. + sqlx::query_as!( + Group, + "INSERT INTO group_ (workspace_id, name, summary, extra_perms) VALUES ($1, $2, $3, $4) ON CONFLICT DO NOTHING", + w_id, + "variable_expiration_handler", + "The group the variable expiration handler acts on behalf of", + serde_json::json!({username_to_permissioned_as(&authed.username): true}) + ) + .execute(&mut *tx) + .await?; + + if let Some(path) = &es.path { + if es + .extra_args + .as_ref() + .is_some_and(|extra_args| !extra_args.is_object()) + { + return Err(Error::BadRequest( + "Field `extra_args` expected to be JSON object".to_string(), + )); + } + + // Prefix only — whether the runnable exists is left to the sweep, which reports an + // unresolvable handler as a critical alert. + if !path.starts_with("script/") && !path.starts_with("flow/") { + return Err(Error::BadRequest(format!( + "Field `path` must start with script/ or flow/ (got {path})" + ))); + } + + // Persisted even when false, matching `error_handler`, so the stored shape is the same + // whether or not the caller sent the field. + let mut variable_expiration_handler = serde_json::json!({ + "path": path, + "muted_on_user_path": es.muted_on_user_path, + }); + if let Some(extra_args) = &es.extra_args { + variable_expiration_handler["extra_args"] = extra_args.clone(); + } + + sqlx::query!( + "UPDATE workspace_settings SET variable_expiration_handler = $1 WHERE workspace_id = $2", + variable_expiration_handler, + &w_id + ) + .execute(&mut *tx) + .await?; + } else { + sqlx::query!( + "UPDATE workspace_settings SET variable_expiration_handler = NULL WHERE workspace_id = $1", + &w_id + ) + .execute(&mut *tx) + .await?; + } + + audit_log( + &mut *tx, + &authed, + "workspaces.edit_variable_expiration_handler", + ActionKind::Update, + &w_id, + Some(&authed.email), + Some([("variable_expiration_handler", &format!("{:?}", es.path)[..])].into()), + ) + .await?; + tx.commit().await?; + + handle_deployment_metadata( + &authed.email, + &authed.username, + &db, + &w_id, + windmill_git_sync::DeployedObject::Settings { + setting_type: "variable_expiration_handler".to_string(), + }, + Some("Variable expiration handler configuration updated".to_string()), + false, + None, + ) + .await?; + + windmill_common::feature_usage::log_feature_usage( + "variable_expiration", + "handler_configured", + if es.path.is_some() { "set" } else { "cleared" }, + ); + + Ok(format!( + "Edit variable_expiration_handler for workspace {}", + &w_id + )) +} + async fn edit_success_handler( authed: ApiAuthed, Extension(db): Extension, @@ -6060,9 +6183,11 @@ async fn clone_variables( source_workspace_id: &str, target_workspace_id: &str, ) -> Result<()> { + // Dispatch bookkeeping is deliberately not copied: the fork re-arms its own expiration + // handler rather than inheriting the source's already-dispatched state. sqlx::query!( - "INSERT INTO variable (workspace_id, path, value, is_secret, description, extra_perms, account, is_oauth, expires_at) - SELECT $2, path, value, is_secret, description, extra_perms, account, is_oauth, expires_at + "INSERT INTO variable (workspace_id, path, value, is_secret, description, extra_perms, account, is_oauth, expires_at, value_expires_at) + SELECT $2, path, value, is_secret, description, extra_perms, account, is_oauth, expires_at, value_expires_at FROM variable WHERE workspace_id = $1", source_workspace_id, @@ -11315,7 +11440,7 @@ async fn compare_two_variables( // Get variable from each workspace let source_variable = sqlx::query!( - "SELECT value, is_secret, description + "SELECT value, is_secret, description, value_expires_at FROM variable WHERE workspace_id = $1 AND path = $2", source_workspace_id, @@ -11325,7 +11450,7 @@ async fn compare_two_variables( .await?; let target_variable = sqlx::query!( - "SELECT value, is_secret, description + "SELECT value, is_secret, description, value_expires_at FROM variable WHERE workspace_id = $1 AND path = $2", fork_workspace_id, @@ -11341,6 +11466,7 @@ async fn compare_two_variables( if source.is_secret != target.is_secret || source.value != target.value || source.description != target.description + || source.value_expires_at != target.value_expires_at { has_changes = true; } diff --git a/backend/windmill-api-workspaces/src/workspaces_extra.rs b/backend/windmill-api-workspaces/src/workspaces_extra.rs index 1968437e40..a8850a1144 100644 --- a/backend/windmill-api-workspaces/src/workspaces_extra.rs +++ b/backend/windmill-api-workspaces/src/workspaces_extra.rs @@ -113,7 +113,7 @@ pub(crate) async fn change_workspace_id( // Duplicate workspace settings (keep copy in old workspace for reference) info!("Duplicating workspace_settings table"); sqlx::query!( - "INSERT INTO workspace_settings (workspace_id, slack_team_id, slack_name, slack_command_script, slack_email, customer_id, plan, webhook, ai_config, large_file_storage, git_sync, default_app, default_scripts, deploy_ui, mute_critical_alerts, color, operator_settings, teams_command_script, teams_team_id, teams_team_name, git_app_installations, ducklake, dbt_warehouses, slack_oauth_client_id, slack_oauth_client_secret, datatable, teams_team_guid, auto_invite, error_handler, success_handler, public_app_execution_limit_per_minute, error_handler_fallback_to_instance_alerts) SELECT $1, slack_team_id, slack_name, slack_command_script, slack_email, customer_id, plan, webhook, ai_config, large_file_storage, git_sync, default_app, default_scripts, deploy_ui, mute_critical_alerts, color, operator_settings, teams_command_script, teams_team_id, teams_team_name, git_app_installations, ducklake, dbt_warehouses, slack_oauth_client_id, slack_oauth_client_secret, datatable, teams_team_guid, auto_invite, error_handler, success_handler, public_app_execution_limit_per_minute, error_handler_fallback_to_instance_alerts FROM workspace_settings WHERE workspace_id = $2", + "INSERT INTO workspace_settings (workspace_id, slack_team_id, slack_name, slack_command_script, slack_email, customer_id, plan, webhook, ai_config, large_file_storage, git_sync, default_app, default_scripts, deploy_ui, mute_critical_alerts, color, operator_settings, teams_command_script, teams_team_id, teams_team_name, git_app_installations, ducklake, dbt_warehouses, slack_oauth_client_id, slack_oauth_client_secret, datatable, teams_team_guid, auto_invite, error_handler, success_handler, public_app_execution_limit_per_minute, error_handler_fallback_to_instance_alerts, variable_expiration_handler) SELECT $1, slack_team_id, slack_name, slack_command_script, slack_email, customer_id, plan, webhook, ai_config, large_file_storage, git_sync, default_app, default_scripts, deploy_ui, mute_critical_alerts, color, operator_settings, teams_command_script, teams_team_id, teams_team_name, git_app_installations, ducklake, dbt_warehouses, slack_oauth_client_id, slack_oauth_client_secret, datatable, teams_team_guid, auto_invite, error_handler, success_handler, public_app_execution_limit_per_minute, error_handler_fallback_to_instance_alerts, variable_expiration_handler FROM workspace_settings WHERE workspace_id = $2", &rw.new_id, &old_id ) diff --git a/backend/windmill-api/openapi.yaml b/backend/windmill-api/openapi.yaml index 707c6fadfb..044f7b7cc3 100644 --- a/backend/windmill-api/openapi.yaml +++ b/backend/windmill-api/openapi.yaml @@ -3722,6 +3722,8 @@ paths: $ref: "#/components/schemas/ErrorHandlerConfig" success_handler: $ref: "#/components/schemas/SuccessHandlerConfig" + variable_expiration_handler: + $ref: "#/components/schemas/VariableExpirationHandlerConfig" large_file_storage: $ref: "#/components/schemas/LargeFileStorage" ducklake: @@ -4506,6 +4508,29 @@ paths: schema: type: string + /w/{workspace}/workspaces/edit_variable_expiration_handler: + post: + summary: edit variable expiration handler + operationId: editVariableExpirationHandler + tags: + - workspace + parameters: + - $ref: "#/components/parameters/WorkspaceId" + requestBody: + description: WorkspaceVariableExpirationHandler + required: true + content: + application/json: + schema: + $ref: "#/components/schemas/EditVariableExpirationHandler" + responses: + "200": + description: status + content: + text/plain: + schema: + type: string + /w/{workspace}/workspaces/edit_success_handler: post: summary: edit success handler @@ -26534,6 +26559,33 @@ components: type: boolean default: false + EditVariableExpirationHandler: + type: object + description: > + Request body for editing the workspace variable expiration handler. Omit + `path` (or send null) to unset the handler. + properties: + path: + type: string + nullable: true + description: Path to the variable expiration handler script or flow + extra_args: + $ref: "#/components/schemas/ScriptArgs" + muted_on_user_path: + type: boolean + description: Do not run the handler for variables on a personal (u/) path + + VariableExpirationHandlerConfig: + type: object + description: Workspace variable expiration handler configuration + properties: + path: + type: string + extra_args: + $ref: "#/components/schemas/ScriptArgs" + muted_on_user_path: + type: boolean + EditSuccessHandler: description: Request body for editing the workspace success handler. Accepts both new grouped format and legacy flat format for backward compatibility. oneOf: @@ -28862,6 +28914,10 @@ components: expires_at: type: string format: date-time + value_expires_at: + type: string + nullable: true + format: date-time labels: type: array items: @@ -28937,7 +28993,17 @@ components: description: Whether the variable is an OAuth variable expires_at: type: string - description: The expiration date of the variable + description: > + When to garbage-collect the variable. Used for ephemeral secrets; + the variable is deleted once this is reached. Not the value's expiry + — see value_expires_at. + format: date-time + value_expires_at: + type: string + description: > + When the value stored in this variable stops working. Triggers the + workspace's variable expiration handler one hour before it is + reached; never causes the variable to be deleted. format: date-time labels: type: array @@ -28966,6 +29032,14 @@ components: description: type: string description: The new description of the variable + value_expires_at: + type: string + nullable: true + description: > + When the value stored in this variable stops working. Omit to leave + it unchanged; send null to clear it. Changing it re-arms the + workspace's variable expiration handler. + format: date-time labels: type: array items: @@ -29069,6 +29143,7 @@ components: - "workspaces.edit_webhook" - "workspaces.edit_copilot_config" - "workspaces.edit_error_handler" + - "workspaces.edit_variable_expiration_handler" - "workspaces.create" - "workspaces.update" - "workspaces.archive" diff --git a/backend/windmill-api/src/mcp/auto_generated_endpoints.rs b/backend/windmill-api/src/mcp/auto_generated_endpoints.rs index 568ad854f9..3f89660fc4 100644 --- a/backend/windmill-api/src/mcp/auto_generated_endpoints.rs +++ b/backend/windmill-api/src/mcp/auto_generated_endpoints.rs @@ -144,7 +144,12 @@ pub fn all_tools() -> Vec { }, "expires_at": { "type": "string", - "description": "The expiration date of the variable", + "description": "When to garbage-collect the variable. Used for ephemeral secrets; the variable is deleted once this is reached. Not the value's expiry — see value_expires_at.\n", + "format": "date-time" + }, + "value_expires_at": { + "type": "string", + "description": "When the value stored in this variable stops working. Triggers the workspace's variable expiration handler one hour before it is reached; never causes the variable to be deleted.\n", "format": "date-time" }, "labels": { @@ -232,6 +237,12 @@ pub fn all_tools() -> Vec { "type": "string", "description": "The new description of the variable" }, + "value_expires_at": { + "type": "string", + "nullable": true, + "description": "When the value stored in this variable stops working. Omit to leave it unchanged; send null to clear it. Changing it re-arms the workspace's variable expiration handler.\n", + "format": "date-time" + }, "labels": { "type": "array", "items": { diff --git a/backend/windmill-api/src/workspaces_export.rs b/backend/windmill-api/src/workspaces_export.rs index 05cd354cd4..70910c885e 100644 --- a/backend/windmill-api/src/workspaces_export.rs +++ b/backend/windmill-api/src/workspaces_export.rs @@ -505,6 +505,7 @@ struct SimplifiedSettings { // mirroring every other workspace setting. error_handler: Option, success_handler: Option, + variable_expiration_handler: Option, #[serde(skip_serializing_if = "Option::is_none")] ai_config: Option, #[serde(skip_serializing_if = "Option::is_none")] @@ -582,6 +583,7 @@ struct SettingsRow { webhook: Option, error_handler: Option, success_handler: Option, + variable_expiration_handler: Option, ai_config: Option, large_file_storage: Option, git_sync: Option, @@ -983,9 +985,9 @@ pub(crate) async fn tarball_workspace( if !skip_variables.unwrap_or(false) { let variables = sqlx::query_as::<_, ExportableListableVariable>(if !skip_secrets.unwrap_or(false) { - "SELECT workspace_id, path, value, is_secret, description, extra_perms, account, is_oauth, expires_at, labels FROM variable WHERE workspace_id = $1 AND expires_at IS NULL" + "SELECT workspace_id, path, value, is_secret, description, extra_perms, account, is_oauth, expires_at, value_expires_at, labels FROM variable WHERE workspace_id = $1 AND expires_at IS NULL" } else { - "SELECT workspace_id, path, value, is_secret, description, extra_perms, account, is_oauth, expires_at, labels FROM variable WHERE workspace_id = $1 AND is_secret = false AND expires_at IS NULL" + "SELECT workspace_id, path, value, is_secret, description, extra_perms, account, is_oauth, expires_at, value_expires_at, labels FROM variable WHERE workspace_id = $1 AND is_secret = false AND expires_at IS NULL" }) .bind(&w_id) .fetch_all(&mut *tx) @@ -1563,6 +1565,7 @@ pub(crate) async fn tarball_workspace( webhook, error_handler, success_handler, + variable_expiration_handler, ai_config, large_file_storage, git_sync, @@ -1622,6 +1625,7 @@ pub(crate) async fn tarball_workspace( webhook: row.webhook, error_handler: row.error_handler, success_handler: row.success_handler, + variable_expiration_handler: row.variable_expiration_handler, ai_config: row.ai_config, large_file_storage: row.large_file_storage, git_sync: redact_git_sync_for_export(row.git_sync), diff --git a/backend/windmill-common/src/variables.rs b/backend/windmill-common/src/variables.rs index 045ddcd95b..2d9e1d2504 100644 --- a/backend/windmill-common/src/variables.rs +++ b/backend/windmill-common/src/variables.rs @@ -147,6 +147,7 @@ pub struct ListableVariable { pub refresh_error: Option, pub is_linked: Option, pub expires_at: Option>, + pub value_expires_at: Option>, #[serde(skip_serializing_if = "Option::is_none")] pub labels: Option>, /// Labels inherited from the parent folder, computed at read time. @@ -189,6 +190,8 @@ pub struct ExportableListableVariable { #[serde(skip_serializing_if = "Option::is_none")] pub expires_at: Option>, #[serde(skip_serializing_if = "Option::is_none")] + pub value_expires_at: Option>, + #[serde(skip_serializing_if = "Option::is_none")] pub labels: Option>, } @@ -204,7 +207,11 @@ pub struct CreateVariable { pub description: String, pub account: Option, pub is_oauth: Option, + /// When the row itself is garbage-collected. pub expires_at: Option>, + /// When the value stops working. Drives the workspace's variable expiration handler; + /// never deletes the row. + pub value_expires_at: Option>, #[serde(default)] pub labels: Option>, #[serde(default)] diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs index a1bd8034f0..5c433ef0cb 100644 --- a/backend/windmill-queue/src/jobs.rs +++ b/backend/windmill-queue/src/jobs.rs @@ -166,12 +166,15 @@ const ERROR_HANDLER_USERNAME: &str = "error_handler"; const SCHEDULE_ERROR_HANDLER_USERNAME: &str = "schedule_error_handler"; const GLOBAL_ERROR_HANDLER_USERNAME: &str = "global"; const SUCCESS_HANDLER_USERNAME: &str = "success_handler"; +const VARIABLE_EXPIRATION_HANDLER_USERNAME: &str = "variable_expiration_handler"; pub const ERROR_HANDLER_USER_GROUP: &str = "g/error_handler"; pub const ERROR_HANDLER_USER_EMAIL: &str = "error_handler@windmill.dev"; pub const SCHEDULE_ERROR_HANDLER_USER_EMAIL: &str = "schedule_error_handler@windmill.dev"; pub const SUCCESS_HANDLER_USER_GROUP: &str = "g/success_handler"; pub const SUCCESS_HANDLER_USER_EMAIL: &str = "success_handler@windmill.dev"; +pub const VARIABLE_EXPIRATION_HANDLER_USER_GROUP: &str = "g/variable_expiration_handler"; +pub const VARIABLE_EXPIRATION_HANDLER_USER_EMAIL: &str = "variable_expiration_handler@windmill.dev"; #[derive(Clone, Debug, Serialize, Deserialize)] pub struct CanceledBy { @@ -2924,6 +2927,102 @@ pub async fn push_error_handler<'a, 'c, T: Serialize + Send + Sync>( return Ok(uuid); } +/// Metadata handed to a variable expiration handler. These field names are wire format for +/// handler scripts: `variable_path` rather than `path`, which already means the failing +/// runnable in the error handler contract. +/// +/// The variable's value is deliberately absent: job results are stored in cleartext and shown +/// in Runs, so a handler must never receive or return one. +#[derive(Serialize)] +pub struct ExpiringVariable { + pub workspace_id: String, + pub variable_path: String, + pub description: String, + pub value_expires_at: DateTime, + pub is_secret: bool, +} + +/// Enqueue the workspace's variable expiration handler for one expiring variable. +/// +/// Takes and returns the caller's transaction so the claim on the variable and the push +/// commit together. +pub(crate) async fn push_variable_expiration_handler<'c>( + db: &Pool, + tx: Transaction<'c, Postgres>, + w_id: &str, + handler_path: &str, + expiring: &ExpiringVariable, + extra_args: Option>>, +) -> windmill_common::error::Result<(Uuid, Transaction<'c, Postgres>)> { + let (payload, tag, on_behalf_of) = + get_payload_tag_from_prefixed_path(handler_path, db, w_id).await?; + + // Built by serializing the struct rather than field by field, so a field added to + // `ExpiringVariable` reaches handlers instead of being silently dropped. + let mut extra: HashMap> = + serde_json::from_value(serde_json::to_value(expiring)?)?; + + if let Some(args_v) = extra_args { + if let Ok(args_m) = serde_json::from_str::>>(args_v.get()) { + extra.extend(args_m); + } else { + return Err(error::Error::ExecutionErr( + "args of scripts needs to be dict".to_string(), + )); + } + } + + let (email, permissioned_as) = if let Some(on_behalf_of) = on_behalf_of.as_ref() { + ( + on_behalf_of.email.as_str(), + on_behalf_of.permissioned_as.clone(), + ) + } else { + ( + VARIABLE_EXPIRATION_HANDLER_USER_EMAIL, + VARIABLE_EXPIRATION_HANDLER_USER_GROUP.to_string(), + ) + }; + + let empty: HashMap> = HashMap::new(); + let args = PushArgs { extra: Some(extra), args: &empty }; + push( + db, + PushIsolationLevel::Transaction(tx), + w_id, + payload, + args, + VARIABLE_EXPIRATION_HANDLER_USERNAME, + email, + permissioned_as, + Some(&format!( + "variable.expiration.{}", + expiring.variable_path.replace('/', ".") + )), + None, + None, + None, + None, + None, + None, + None, + false, + false, + None, + true, + tag, + None, + None, + None, + None, + false, + None, + None, + None, + ) + .await +} + pub async fn push_success_handler<'a, 'c, T: Serialize + Send + Sync>( db: &Pool, job_id: Uuid, @@ -5536,6 +5635,7 @@ async fn push_inner<'c, 'd>( if !team_plan_status.premium && email != ERROR_HANDLER_USER_EMAIL && email != SCHEDULE_ERROR_HANDLER_USER_EMAIL + && email != VARIABLE_EXPIRATION_HANDLER_USER_EMAIL && email != recovery_email && email != "worker@windmill.dev" && email != SUPERADMIN_SECRET_EMAIL diff --git a/backend/windmill-queue/src/lib.rs b/backend/windmill-queue/src/lib.rs index ff93ea2665..5358960635 100644 --- a/backend/windmill-queue/src/lib.rs +++ b/backend/windmill-queue/src/lib.rs @@ -34,6 +34,7 @@ pub mod jobs; pub mod jobs_ee; pub mod jobs_oss; pub mod schedule; +pub mod variable_expiration; pub use jobs::*; pub mod flow_status; pub mod tags; diff --git a/backend/windmill-queue/src/variable_expiration.rs b/backend/windmill-queue/src/variable_expiration.rs new file mode 100644 index 0000000000..c8c2599319 --- /dev/null +++ b/backend/windmill-queue/src/variable_expiration.rs @@ -0,0 +1,357 @@ +/* + * Author: Ruben Fiszel + * Copyright: Windmill Labs, Inc 2022 + * This file and its contents are licensed under the AGPLv3 License. + * Please see the included NOTICE for copyright information and + * LICENSE-AGPL for a copy of the license. + */ + +//! Detection pass for variables whose value is about to expire. +//! +//! Owns the transaction that claims a variable and pushes its handler job, so it lives in +//! this crate beside `push_variable_expiration_handler` rather than in the server's `monitor` +//! module, which only drives it on a timer. + +use std::collections::{HashMap, HashSet}; +use std::sync::Mutex; + +use serde_json::value::RawValue; +use sqlx::types::Json; +use windmill_common::utils::report_critical_error; +use windmill_common::DB; + +use crate::jobs::{push_variable_expiration_handler, ExpiringVariable}; + +/// How far ahead of a variable's `value_expires_at` its handler runs. +const VARIABLE_EXPIRATION_LEAD_TIME: &str = "1 hour"; + +/// Cap on variables dispatched per pass, so a large first-pass backlog is drained over +/// several passes rather than enqueuing every handler job at once. +const VARIABLE_EXPIRATION_MAX_PER_PASS: i64 = 100; + +/// Advisory lock id ensuring only one server replica sweeps at a time (adjacent to +/// `SCHEDULE_RECONCILE_LOCK_ID`). +const VARIABLE_EXPIRATION_LOCK_ID: i64 = 737_483_923; + +/// Consecutive passes a workspace's handler must fail before the failure is surfaced once as +/// a critical alert. Most push failures are a transient blip; one that keeps failing has a +/// real cause — the handler script is gone, or its stored path does not resolve — and would +/// otherwise leave every expiry in that workspace silently undispatched. +const VARIABLE_EXPIRATION_ALERT_THRESHOLD: u32 = 3; + +/// Cap on the exponential back-off (in passes) between retries of a workspace whose handler +/// keeps failing. The delay grows 2, 4, 8 and holds here, and resets as soon as one of its +/// pushes succeeds. +const VARIABLE_EXPIRATION_MAX_BACKOFF_PASSES: u32 = 8; + +/// State for a workspace whose handler keeps failing to push: paces retries and drives the +/// one-time visibility so the loop is neither hot nor silent. +#[derive(Default)] +struct HandlerFailureState { + /// The handler these failures were counted against. A workspace that points at a + /// different one has not failed yet, so the state is discarded rather than aged out. + handler_path: String, + consecutive_failures: u32, + /// Passes still to skip before the next attempt (exponential back-off). + cooldown_passes: u32, + /// Whether the persistent failure has already been surfaced. + surfaced: bool, +} + +lazy_static::lazy_static! { + /// `workspace_id` -> back-off/visibility state. Keyed by workspace rather than by + /// variable because the handler is a workspace setting: when it is unresolvable every + /// variable in that workspace fails identically, and one alert names the cause where a + /// hundred per-row errors would only repeat it. + /// + /// Per-replica, like `SCHEDULE_REARM_FAILURES`: the advisory lock serializes concurrent + /// sweeps but does not pin successive ones to one process, so a broken handler is retried + /// and alerted once per replica before all have backed off. Pacing only — no double + /// dispatch rides on it, as the claim is persisted in `expiration_dispatched_for`. + static ref VARIABLE_EXPIRATION_FAILURES: Mutex> = + Mutex::new(HashMap::new()); +} + +/// Enqueue the workspace variable expiration handler for every variable coming due. +/// +/// Not an authorization boundary: it dispatches handlers across every workspace, so this is +/// a system caller (the monitor loop) only. +/// +/// Detection rather than pre-enqueue: pushing a future-dated job would pin the handler's +/// script version months ahead and turn this into a reconciliation problem (see +/// `reconcile_unarmed_schedules`) instead of a detection one. +/// +/// Dispatch-once, not run-to-success: the claim is committed with the push, so a handler job +/// that then fails is not re-dispatched for that date. Retrying is the handler's own job (a +/// retry policy, or the workspace error handler) — re-running it here would re-run whatever +/// side effects the failed attempt already had against the third party it rotates. +pub async fn dispatch_expiring_variables(db: &DB) { + // Transaction-scoped advisory lock, not session-scoped: a session lock taken on a pooled + // connection would ride that connection back into the pool still held if this future is + // dropped mid-pass, wedging the sweep on every replica until restart. The tx is held open + // only to own the lock. + let mut lock_tx = match db.begin().await { + Ok(tx) => tx, + Err(e) => { + tracing::error!("variable expiration: failed to begin lock tx: {e:#}"); + return; + } + }; + let locked: bool = match sqlx::query_scalar("SELECT pg_try_advisory_xact_lock($1)") + .bind(VARIABLE_EXPIRATION_LOCK_ID) + .fetch_one(&mut *lock_tx) + .await + { + Ok(v) => v, + Err(e) => { + tracing::error!("variable expiration: advisory lock failed: {e:#}"); + return; + } + }; + if !locked { + // Another replica is already sweeping this tick. + return; + } + + dispatch_expiring_variables_inner(db).await; + + // Ends the transaction and releases the xact lock; a plain drop would too. + if let Err(e) = lock_tx.rollback().await { + tracing::error!("variable expiration: releasing lock failed: {e:#}"); + } +} + +async fn dispatch_expiring_variables_inner(db: &DB) { + // Every workspace with a variable coming due, before any back-off is applied. A + // deliberate superset of the dispatch predicate below — it skips the mute and the cap, + // which can only make a workspace spend its back-off more eagerly, never less. + let with_due = sqlx::query!( + r#"SELECT DISTINCT v.workspace_id AS "workspace_id!", + ws.variable_expiration_handler->>'path' AS "handler_path!" + FROM variable v + JOIN workspace_settings ws ON ws.workspace_id = v.workspace_id + JOIN workspace w ON w.id = v.workspace_id + WHERE NOT w.deleted + AND v.value_expires_at IS NOT NULL + AND v.value_expires_at IS DISTINCT FROM v.expiration_dispatched_for + AND v.value_expires_at <= now() + ($1::text)::interval + AND ws.variable_expiration_handler->>'path' IS NOT NULL"#, + VARIABLE_EXPIRATION_LEAD_TIME, + ) + .fetch_all(db) + .await; + + let with_due: HashMap = match with_due { + Ok(rows) => rows + .into_iter() + .map(|r| (r.workspace_id, r.handler_path)) + .collect(), + Err(e) => { + tracing::error!("Error listing workspaces with expiring variables: {e:#}"); + return; + } + }; + if with_due.is_empty() { + return; + } + + // Excluded in the query rather than filtered out of its result: the per-pass cap is + // applied by `LIMIT`, so a workspace whose every push fails would otherwise fill those + // slots on every pass and starve every other workspace on the instance. Filtering before + // the cap is the ordering `reconcile_unarmed_schedules_inner` relies on for the same + // reason. + // + // Only a workspace that has work this pass spends a pass of its back-off, so one that + // goes quiet does not silently serve out its cooldown while nothing is expiring. + let cooling_down: Vec = { + let mut failures = VARIABLE_EXPIRATION_FAILURES.lock().unwrap(); + // Back-off paces retries of one handler, so pointing the workspace at another one + // ends it. Without this a repaired workspace stays out of the sweep for up to + // `VARIABLE_EXPIRATION_MAX_BACKOFF_PASSES` passes — most of the lead time it has. + failures.retain(|w_id, state| match with_due.get(w_id) { + Some(current) => *current == state.handler_path, + None => true, + }); + failures + .iter_mut() + .filter_map(|(w_id, state)| { + if !with_due.contains_key(w_id) || state.cooldown_passes == 0 { + return None; + } + state.cooldown_passes -= 1; + Some(w_id.clone()) + }) + .collect() + }; + + // The due predicate has no lower bound on `value_expires_at`: an instance that was down + // catches up rather than silently skipping the variables whose window elapsed meanwhile. + let due = sqlx::query!( + r#"SELECT v.workspace_id AS "workspace_id!", v.path AS "path!", v.description AS "description!", + v.value_expires_at AS "value_expires_at!", v.is_secret AS "is_secret!", + ws.variable_expiration_handler->>'path' AS "handler_path!", + (ws.variable_expiration_handler->'extra_args')::text::json AS "extra_args: Json>" + FROM variable v + JOIN workspace_settings ws ON ws.workspace_id = v.workspace_id + JOIN workspace w ON w.id = v.workspace_id + WHERE NOT w.deleted + AND v.workspace_id <> ALL($3) + AND v.value_expires_at IS NOT NULL + AND v.value_expires_at IS DISTINCT FROM v.expiration_dispatched_for + AND v.value_expires_at <= now() + ($1::text)::interval + AND ws.variable_expiration_handler->>'path' IS NOT NULL + AND (NOT COALESCE((ws.variable_expiration_handler->>'muted_on_user_path')::boolean, false) + OR v.path NOT LIKE 'u/%') + ORDER BY v.value_expires_at + LIMIT $2"#, + VARIABLE_EXPIRATION_LEAD_TIME, + VARIABLE_EXPIRATION_MAX_PER_PASS, + &cooling_down, + ) + .fetch_all(db) + .await; + + let due = match due { + Ok(due) => due, + Err(e) => { + tracing::error!("Error listing expiring variables: {e:#}"); + return; + } + }; + + let mut dispatched = 0usize; + // A workspace's handler resolves the same way for every one of its variables, so once it + // has failed this pass the rest of its rows would only repeat the same error. + let mut failed_this_pass: HashSet = HashSet::new(); + + for row in due { + if failed_this_pass.contains(&row.workspace_id) { + continue; + } + + // Claim and push in one transaction. Storing the date that was dispatched makes the + // claim lose both to a concurrent re-arm (the `value_expires_at` guard) and to + // whichever replica got there first. + let mut tx = match db.begin().await { + Ok(tx) => tx, + Err(e) => { + tracing::error!("Error opening tx for expiring variable {}: {e:#}", row.path); + continue; + } + }; + + let claimed = sqlx::query_scalar!( + "UPDATE variable SET expiration_dispatched_for = $3 + WHERE workspace_id = $1 AND path = $2 AND value_expires_at = $3 + AND expiration_dispatched_for IS DISTINCT FROM $3 + RETURNING 1", + row.workspace_id, + row.path, + row.value_expires_at, + ) + .fetch_optional(&mut *tx) + .await; + + match claimed { + Ok(Some(_)) => {} + Ok(None) => continue, + Err(e) => { + tracing::error!("Error claiming expiring variable {}: {e:#}", row.path); + continue; + } + } + + let expiring = ExpiringVariable { + workspace_id: row.workspace_id.clone(), + variable_path: row.path.clone(), + description: row.description, + value_expires_at: row.value_expires_at, + is_secret: row.is_secret, + }; + + let pushed = push_variable_expiration_handler( + db, + tx, + &row.workspace_id, + &row.handler_path, + &expiring, + row.extra_args, + ) + .await; + + let tx = match pushed { + Ok((_job_id, tx)) => tx, + Err(e) => { + // The claim rides in the same transaction, so dropping it un-claims the + // variable and a later pass retries it once the back-off elapses. + tracing::error!( + "Error pushing variable expiration handler for {}: {e:#}", + row.path + ); + failed_this_pass.insert(row.workspace_id.clone()); + record_handler_failure(db, &row.workspace_id, &row.handler_path, e).await; + continue; + } + }; + + if let Err(e) = tx.commit().await { + tracing::error!("Error committing dispatch for {}: {e:#}", row.path); + continue; + } + clear_handler_failure(&row.workspace_id); + windmill_common::feature_usage::log_feature_usage( + "variable_expiration", + "dispatched", + if row.is_secret { "secret" } else { "plain" }, + ); + dispatched += 1; + } + + if dispatched > 0 { + tracing::info!("Dispatched {dispatched} variable expiration handler job(s)"); + } +} + +/// Back off this workspace's handler, and surface the cause once it is clearly not transient. +async fn record_handler_failure( + db: &DB, + w_id: &str, + handler_path: &str, + e: windmill_common::error::Error, +) { + let should_surface = { + let mut failures = VARIABLE_EXPIRATION_FAILURES.lock().unwrap(); + let state = failures.entry(w_id.to_string()).or_default(); + if state.handler_path != handler_path { + *state = HandlerFailureState::default(); + state.handler_path = handler_path.to_string(); + } + state.consecutive_failures += 1; + state.cooldown_passes = + (1u32 << state.consecutive_failures.min(5)).min(VARIABLE_EXPIRATION_MAX_BACKOFF_PASSES); + let surface = + !state.surfaced && state.consecutive_failures >= VARIABLE_EXPIRATION_ALERT_THRESHOLD; + state.surfaced |= surface; + surface + }; + + if should_surface { + report_critical_error( + format!( + "The variable expiration handler {handler_path} of workspace {w_id} has repeatedly \ + failed to start. Variables in this workspace will not run their expiration \ + handler until the cause is fixed: {e:#}" + ), + db.clone(), + Some(w_id), + None, + ) + .await; + } +} + +/// Drop a workspace's back-off as soon as one of its pushes succeeds, so a handler fixed out +/// of band recovers on the next pass. +fn clear_handler_failure(w_id: &str) { + VARIABLE_EXPIRATION_FAILURES.lock().unwrap().remove(w_id); +} diff --git a/backend/windmill-store/src/variables.rs b/backend/windmill-store/src/variables.rs index 8f288dc85e..35f46e37d1 100644 --- a/backend/windmill-store/src/variables.rs +++ b/backend/windmill-store/src/variables.rs @@ -149,6 +149,7 @@ async fn list_variables( "resource.path IS NOT NULL as is_linked", "account.refresh_token != '' as is_refreshed", "variable.expires_at", + "variable.value_expires_at", "variable.labels", "folder_labels(variable.workspace_id, variable.path) as inherited_labels", "ws_specific.path IS NOT NULL as ws_specific", @@ -245,7 +246,7 @@ async fn list_variables( for row in draft_only_rows { let v: serde_json::Value = serde_json::from_str(row.value.0.get()).unwrap_or(serde_json::Value::Null); - // VariableEditor's `VariableState`: { path, variable: { value, is_secret, description }, labels?, wsSpecific } + // VariableEditor's `VariableState`: { path, variable: { value, is_secret, description }, labels?, wsSpecific, value_expires_at? } let path = v .get("path") .and_then(|s| s.as_str()) @@ -284,6 +285,11 @@ async fn list_variables( }) }); let ws_specific = v.get("wsSpecific").and_then(|x| x.as_bool()); + let value_expires_at = v + .get("value_expires_at") + .and_then(|x| x.as_str()) + .and_then(|s| chrono::DateTime::parse_from_rfc3339(s).ok()) + .map(|dt| dt.with_timezone(&chrono::Utc)); rows.push(ListableVariable { workspace_id: w_id.clone(), @@ -299,6 +305,7 @@ async fn list_variables( refresh_error: None, is_linked: None, expires_at: None, + value_expires_at, labels, // No deployed row to inherit folder labels from. inherited_labels: None, @@ -339,7 +346,7 @@ async fn get_variable( let variable_o = sqlx::query_as::<_, ListableVariable>( "SELECT variable.workspace_id, variable.path, variable.value, variable.is_secret, variable.description, variable.extra_perms, variable.account, variable.is_oauth, - variable.expires_at, variable.labels, + variable.expires_at, variable.value_expires_at, variable.labels, folder_labels(variable.workspace_id, variable.path) as inherited_labels, variable.edited_at, variable.edited_by, (now() > account.expires_at) as is_expired, account.refresh_error, @@ -648,8 +655,8 @@ async fn create_variable( sqlx::query!( "INSERT INTO variable - (workspace_id, path, value, is_secret, description, account, is_oauth, expires_at, labels, edited_by) - VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10)", + (workspace_id, path, value, is_secret, description, account, is_oauth, expires_at, value_expires_at, labels, edited_by) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11)", &w_id, variable.path, value, @@ -658,6 +665,7 @@ async fn create_variable( variable.account, variable.is_oauth.unwrap_or(false), variable.expires_at, + variable.value_expires_at, variable.labels.as_deref() as Option<&[String]>, &authed.username ) @@ -1053,6 +1061,16 @@ async fn delete_variables_bulk( Ok(Json(deleted_paths)) } +/// Distinguishes an absent JSON field from an explicit `null`. A plain `Option>` +/// cannot: the outer `Option`'s own impl already maps `null` to `None`. +fn double_option<'de, T, D>(deserializer: D) -> std::result::Result>, D::Error> +where + T: Deserialize<'de>, + D: serde::Deserializer<'de>, +{ + Deserialize::deserialize(deserializer).map(Some) +} + #[derive(Deserialize)] struct EditVariable { path: Option, @@ -1062,6 +1080,10 @@ struct EditVariable { account: Option, labels: Option>, ws_specific: Option, + /// Absent leaves the stored expiry alone; an explicit `null` clears it — a caller + /// patching one field must not have its omissions read as "clear". + #[serde(default, deserialize_with = "double_option")] + value_expires_at: Option>>, } #[derive(Deserialize)] @@ -1161,6 +1183,17 @@ async fn update_variable( has_sql_updates = true; } + if let Some(nvalue_expires_at) = ns.value_expires_at { + // No bookkeeping to reset: the sweep compares this against + // `expiration_dispatched_for`, so moving the date re-arms the handler on its own and + // rewriting the same date (an unchanged CLI push) leaves it disarmed. + match nvalue_expires_at { + Some(ts) => sqlb.set("value_expires_at", "?".bind(&ts.to_rfc3339())), + None => sqlb.set("value_expires_at", "NULL"), + }; + has_sql_updates = true; + } + if let Some(nbool) = ns.is_secret { let old_secret = sqlx::query_scalar!( "SELECT is_secret from variable WHERE path = $1 AND workspace_id = $2",