feat: run a workspace handler when a secret variable expires

Secret variables gain an optional `value_expires_at`, and a workspace can point at a script or flow to run once that date passes, so a rotation is driven by the platform instead of tracked out of band. The secret never leaves the database. The sweep selects the path, description and expiry and nothing else, so the handler is given a reference to rotate rather than a value to handle. Job results are stored in cleartext and shown in Runs, which is what makes that a boundary rather than a preference; a test asserts the job's args never contain the value. Each row is claimed by writing `expiration_dispatched_for` in the same transaction as the push, so a failed push un-claims it and a lost race claims nothing. Moving the date re-arms the row; re-saving the same date does not. A pass takes at most 100 rows under a transaction-scoped advisory lock, and orders so that one broken workspace cannot starve the rest. Dispatch runs as `g/variable_expiration_handler`, exempt from the CE group limit and from non-premium exec metering the way the error and schedule handlers already are. A handler that keeps failing backs off and raises a critical alert instead of retrying every pass, and the back-off ends when the handler is repointed. The expiry travels with the variable through workspace clone, tarball export and workspace comparison. The dispatch marker deliberately does not, so a fork re-arms on its own copy. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
2026-10-14 00:02:55 +00:00 · 2026-08-27 12:15:23 +02:00
parent 90b40fffc3
commit 35bde36730
36 changed files with 1791 additions and 23 deletions
@@ -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"
+}
@@ -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"
+}
@@ -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"
+}
@@ -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"
 }
@@ -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<Box<RawValue>>\"\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<Box<RawValue>>",
+        "type_info": "Json"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Int8",
+        "TextArray"
+      ]
+    },
+    "nullable": [
+      false,
+      false,
+      false,
+      true,
+      false,
+      null,
+      null
+    ]
+  },
+  "hash": "6e64c0eaa4b804424f5890468eac6b1cf4b1d811bcf6a3e2e7b0d39b321eaeef"
+}
@@ -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"
+}
@@ -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"
+}
@@ -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"
+}
@@ -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"
+}
@@ -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"
+}
@@ -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"
 }
@@ -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"
+}
@@ -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"
+}
@@ -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"
+}
@@ -1 +1 @@
-b58ad414b098d3d7787001a352bfbb13e43a335f
+a4934a3213b598f77a35cdb3d774c45ad125e453
@@ -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;
@@ -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;
@@ -0,0 +1 @@
+ALTER TABLE workspace_settings DROP COLUMN IF EXISTS variable_expiration_handler;
@@ -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;
@@ -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,
@@ -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)
@@ -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<Postgres>, 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<Postgres>, 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<Postgres>, 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<Postgres>, path: &str) -> anyhow::Result<i64> {
+    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<Postgres>) -> 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<Postgres>) -> 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<u16> {
+        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<chrono::Utc> = 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<Postgres>) -> 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<String> =
+        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<Postgres>,
+) -> 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<Postgres>) -> 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<Postgres>) -> 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<u16> {
+        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<Option<serde_json::Value>> {
+        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(())
+}
@@ -288,6 +288,49 @@ async fn test_tarball_export_gates_values_on_item_scopes(db: Pool<Postgres>) ->
    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<Postgres>) -> 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
@@ -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(),
        ));
    }
@@ -338,3 +338,43 @@ async fn test_variables_are_cached(db: Pool<Postgres>) -> 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<Postgres>) -> 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::<Vec<serde_json::Value>>().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(())
+}
@@ -2537,7 +2537,10 @@ async fn test_rename_records_the_vacated_path(db: Pool<Postgres>) -> 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<Postgres>) -> 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<Postgres>) -> 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<String> = 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::<serde_json::Value>()
+        .await?;
+
+    assert_eq!(
+        comparison["summary"]["variables_changed"].as_u64(),
+        Some(1),
+        "an expiry-only change must show as deployable: {comparison}"
+    );
+
+    Ok(())
+}
@@ -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_json::Value>,
    #[serde(skip_serializing_if = "Option::is_none")]
+    pub variable_expiration_handler: Option<serde_json::Value>,
+    #[serde(skip_serializing_if = "Option::is_none")]
    pub public_app_execution_limit_per_minute: Option<i32>,
    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<String>,
+    pub extra_args: Option<serde_json::Value>,
+    #[serde(default)]
+    pub muted_on_user_path: bool,
+}
+
+async fn edit_variable_expiration_handler(
+    authed: ApiAuthed,
+    Extension(db): Extension<DB>,
+    Path(w_id): Path<String>,
+    ApiAuthed { is_admin, username, .. }: ApiAuthed,
+    Json(es): Json<EditVariableExpirationHandler>,
+) -> Result<String> {
+    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<DB>,
@@ -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;
        }
@@ -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
    )
@@ -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"
@@ -144,7 +144,12 @@ pub fn all_tools() -> Vec<EndpointTool> {
                },
                "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<EndpointTool> {
                        "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": {
@@ -505,6 +505,7 @@ struct SimplifiedSettings {
    // mirroring every other workspace setting.
    error_handler: Option<Value>,
    success_handler: Option<Value>,
+    variable_expiration_handler: Option<Value>,
    #[serde(skip_serializing_if = "Option::is_none")]
    ai_config: Option<serde_json::Value>,
    #[serde(skip_serializing_if = "Option::is_none")]
@@ -582,6 +583,7 @@ struct SettingsRow {
    webhook: Option<String>,
    error_handler: Option<Value>,
    success_handler: Option<Value>,
+    variable_expiration_handler: Option<Value>,
    ai_config: Option<serde_json::Value>,
    large_file_storage: Option<Value>,
    git_sync: Option<Value>,
@@ -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),
@@ -147,6 +147,7 @@ pub struct ListableVariable {
    pub refresh_error: Option<String>,
    pub is_linked: Option<bool>,
    pub expires_at: Option<chrono::DateTime<Utc>>,
+    pub value_expires_at: Option<chrono::DateTime<Utc>>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub labels: Option<Vec<String>>,
    /// 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<chrono::DateTime<Utc>>,
    #[serde(skip_serializing_if = "Option::is_none")]
+    pub value_expires_at: Option<chrono::DateTime<Utc>>,
+    #[serde(skip_serializing_if = "Option::is_none")]
    pub labels: Option<Vec<String>>,
 }

@@ -204,7 +207,11 @@ pub struct CreateVariable {
    pub description: String,
    pub account: Option<i32>,
    pub is_oauth: Option<bool>,
+    /// When the row itself is garbage-collected.
    pub expires_at: Option<chrono::DateTime<Utc>>,
+    /// When the value stops working. Drives the workspace's variable expiration handler;
+    /// never deletes the row.
+    pub value_expires_at: Option<chrono::DateTime<Utc>>,
    #[serde(default)]
    pub labels: Option<Vec<String>>,
    #[serde(default)]
@@ -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<Utc>,
+    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<Postgres>,
+    tx: Transaction<'c, Postgres>,
+    w_id: &str,
+    handler_path: &str,
+    expiring: &ExpiringVariable,
+    extra_args: Option<Json<Box<RawValue>>>,
+) -> 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<String, Box<RawValue>> =
+        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::<HashMap<String, Box<RawValue>>>(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<String, Box<RawValue>> = 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<Postgres>,
    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
@@ -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;
@@ -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<HashMap<String, HandlerFailureState>> =
+        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<String, String> = 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<String> = {
+        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<Box<RawValue>>"
+           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<String> = 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);
+}
@@ -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<Option<T>>`
+/// cannot: the outer `Option`'s own impl already maps `null` to `None`.
+fn double_option<'de, T, D>(deserializer: D) -> std::result::Result<Option<Option<T>>, D::Error>
+where
+    T: Deserialize<'de>,
+    D: serde::Deserializer<'de>,
+{
+    Deserialize::deserialize(deserializer).map(Some)
+}
+
 #[derive(Deserialize)]
 struct EditVariable {
    path: Option<String>,
@@ -1062,6 +1080,10 @@ struct EditVariable {
    account: Option<i32>,
    labels: Option<Vec<String>>,
    ws_specific: Option<bool>,
+    /// 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<Option<chrono::DateTime<chrono::Utc>>>,
 }

 #[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",
				`@@ -0,0 +1 @@`
				`ALTER TABLE workspace_settings DROP COLUMN IF EXISTS variable_expiration_handler;`