feat: schedule recovery handler (#2126)

* feat: schedule recovery handler * fix: migration down * fix: rename * fix: adjust + add tests * fix: merge + sqlx prepare * fix: more complete recovery handler * feat: schedule error and recovery times * fix: schedule worker test * feat: slack schedule handlers * fix: update schedule handlers
2026-08-18 16:02:10 +00:00 · 2023-08-24 18:20:35 +02:00
parent b1c2ccc024
commit cbbfee8439
31 changed files with 1426 additions and 85632 deletions
@@ -59,7 +59,6 @@
                "bash",
                "postgresql",
                "nativets",
-                "Nativets",
                "bun",
                "mysql",
                "bigquery",
@@ -1,6 +1,6 @@
 {
  "db_name": "PostgreSQL",
-  "query": "UPDATE schedule SET schedule = $1, timezone = $2, args = $3, on_failure = $4 WHERE path = $5 AND workspace_id = $6 RETURNING *",
+  "query": "UPDATE schedule SET schedule = $1, timezone = $2, args = $3, on_failure = $4, on_failure_times = $5, on_failure_exact = $6, on_failure_extra_args = $7, on_recovery = $8, on_recovery_times = $9, on_recovery_extra_args = $10 WHERE path = $11 AND workspace_id = $12 RETURNING *",
  "describe": {
    "columns": [
      {
@@ -72,6 +72,36 @@
        "ordinal": 13,
        "name": "on_failure",
        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 14,
+        "name": "on_recovery",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 15,
+        "name": "on_failure_times",
+        "type_info": "Int4"
+      },
+      {
+        "ordinal": 16,
+        "name": "on_failure_exact",
+        "type_info": "Bool"
+      },
+      {
+        "ordinal": 17,
+        "name": "on_failure_extra_args",
+        "type_info": "Json"
+      },
+      {
+        "ordinal": 18,
+        "name": "on_recovery_times",
+        "type_info": "Int4"
+      },
+      {
+        "ordinal": 19,
+        "name": "on_recovery_extra_args",
+        "type_info": "Json"
      }
    ],
    "parameters": {
@@ -80,6 +110,12 @@
        "Varchar",
        "Jsonb",
        "Varchar",
+        "Int4",
+        "Bool",
+        "Json",
+        "Varchar",
+        "Int4",
+        "Json",
        "Text",
        "Text"
      ]
@@ -98,8 +134,14 @@
      false,
      true,
      false,
+      true,
+      true,
+      true,
+      true,
+      true,
+      true,
      true
    ]
  },
-  "hash": "1e0769c59ada78848e034250c11be0ca2e8c49b96a0066b9ab67f5a205ec456d"
+  "hash": "0be97088e8ca1e8be9bf4ea45f830cceedd598a3221c522a1f9026d2d48a0ffb"
 }
@@ -66,7 +66,6 @@
                "bash",
                "postgresql",
                "nativets",
-                "Nativets",
                "bun",
                "mysql",
                "bigquery",
@@ -0,0 +1,38 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT success, result, started_at FROM completed_job WHERE workspace_id = $1 AND schedule_path = $2 AND script_path = $3 AND id != $4 ORDER BY created_at DESC LIMIT $5",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "success",
+        "type_info": "Bool"
+      },
+      {
+        "ordinal": 1,
+        "name": "result",
+        "type_info": "Jsonb"
+      },
+      {
+        "ordinal": 2,
+        "name": "started_at",
+        "type_info": "Timestamptz"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Text",
+        "Text",
+        "Uuid",
+        "Int8"
+      ]
+    },
+    "nullable": [
+      false,
+      true,
+      false
+    ]
+  },
+  "hash": "4a3917318d967d96c4b6e70786a034302f04da9d1e9eec59b1eb6052c2d947be"
+}
@@ -17,7 +17,6 @@
                "bash",
                "postgresql",
                "nativets",
-                "Nativets",
                "bun",
                "mysql",
                "bigquery",
@@ -75,6 +75,36 @@
      },
      {
        "ordinal": 14,
+        "name": "on_recovery",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 15,
+        "name": "on_failure_times",
+        "type_info": "Int4"
+      },
+      {
+        "ordinal": 16,
+        "name": "on_failure_exact",
+        "type_info": "Bool"
+      },
+      {
+        "ordinal": 17,
+        "name": "on_failure_extra_args",
+        "type_info": "Json"
+      },
+      {
+        "ordinal": 18,
+        "name": "on_recovery_times",
+        "type_info": "Int4"
+      },
+      {
+        "ordinal": 19,
+        "name": "on_recovery_extra_args",
+        "type_info": "Json"
+      },
+      {
+        "ordinal": 20,
        "name": "jobs",
        "type_info": "JsonArray"
      }
@@ -101,6 +131,12 @@
      true,
      false,
      true,
+      true,
+      true,
+      true,
+      true,
+      true,
+      true,
      null
    ]
  },
@@ -72,6 +72,36 @@
        "ordinal": 13,
        "name": "on_failure",
        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 14,
+        "name": "on_recovery",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 15,
+        "name": "on_failure_times",
+        "type_info": "Int4"
+      },
+      {
+        "ordinal": 16,
+        "name": "on_failure_exact",
+        "type_info": "Bool"
+      },
+      {
+        "ordinal": 17,
+        "name": "on_failure_extra_args",
+        "type_info": "Json"
+      },
+      {
+        "ordinal": 18,
+        "name": "on_recovery_times",
+        "type_info": "Int4"
+      },
+      {
+        "ordinal": 19,
+        "name": "on_recovery_extra_args",
+        "type_info": "Json"
      }
    ],
    "parameters": {
@@ -95,6 +125,12 @@
      false,
      true,
      false,
+      true,
+      true,
+      true,
+      true,
+      true,
+      true,
      true
    ]
  },
@@ -72,6 +72,36 @@
        "ordinal": 13,
        "name": "on_failure",
        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 14,
+        "name": "on_recovery",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 15,
+        "name": "on_failure_times",
+        "type_info": "Int4"
+      },
+      {
+        "ordinal": 16,
+        "name": "on_failure_exact",
+        "type_info": "Bool"
+      },
+      {
+        "ordinal": 17,
+        "name": "on_failure_extra_args",
+        "type_info": "Json"
+      },
+      {
+        "ordinal": 18,
+        "name": "on_recovery_times",
+        "type_info": "Int4"
+      },
+      {
+        "ordinal": 19,
+        "name": "on_recovery_extra_args",
+        "type_info": "Json"
      }
    ],
    "parameters": {
@@ -96,6 +126,12 @@
      false,
      true,
      false,
+      true,
+      true,
+      true,
+      true,
+      true,
+      true,
      true
    ]
  },
@@ -1,6 +1,6 @@
 {
  "db_name": "PostgreSQL",
-  "query": "INSERT INTO schedule (workspace_id, path, schedule, timezone, edited_by, script_path, is_flow, args, enabled, email, on_failure) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11) RETURNING *",
+  "query": "INSERT INTO schedule (workspace_id, path, schedule, timezone, edited_by, script_path, is_flow, args, enabled, email, on_failure, on_failure_times, on_failure_exact, on_failure_extra_args, on_recovery, on_recovery_times, on_recovery_extra_args) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17) RETURNING *",
  "describe": {
    "columns": [
      {
@@ -72,6 +72,36 @@
        "ordinal": 13,
        "name": "on_failure",
        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 14,
+        "name": "on_recovery",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 15,
+        "name": "on_failure_times",
+        "type_info": "Int4"
+      },
+      {
+        "ordinal": 16,
+        "name": "on_failure_exact",
+        "type_info": "Bool"
+      },
+      {
+        "ordinal": 17,
+        "name": "on_failure_extra_args",
+        "type_info": "Json"
+      },
+      {
+        "ordinal": 18,
+        "name": "on_recovery_times",
+        "type_info": "Int4"
+      },
+      {
+        "ordinal": 19,
+        "name": "on_recovery_extra_args",
+        "type_info": "Json"
      }
    ],
    "parameters": {
@@ -86,7 +116,13 @@
        "Jsonb",
        "Bool",
        "Varchar",
-        "Varchar"
+        "Varchar",
+        "Int4",
+        "Bool",
+        "Json",
+        "Varchar",
+        "Int4",
+        "Json"
      ]
    },
    "nullable": [
@@ -103,8 +139,14 @@
      false,
      true,
      false,
+      true,
+      true,
+      true,
+      true,
+      true,
+      true,
      true
    ]
  },
-  "hash": "80e632cb6618cde2bac2c12d0afc796a4a1768f902d0540af33fe6f442cf583a"
+  "hash": "9b9afb9d5ef57b70de1207c4546b5547e579d978d899325d821a5aee1df1a9fc"
 }
@@ -28,7 +28,6 @@
                "bash",
                "postgresql",
                "nativets",
-                "Nativets",
                "bun",
                "mysql",
                "bigquery",
@@ -72,6 +72,36 @@
        "ordinal": 13,
        "name": "on_failure",
        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 14,
+        "name": "on_recovery",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 15,
+        "name": "on_failure_times",
+        "type_info": "Int4"
+      },
+      {
+        "ordinal": 16,
+        "name": "on_failure_exact",
+        "type_info": "Bool"
+      },
+      {
+        "ordinal": 17,
+        "name": "on_failure_extra_args",
+        "type_info": "Json"
+      },
+      {
+        "ordinal": 18,
+        "name": "on_recovery_times",
+        "type_info": "Int4"
+      },
+      {
+        "ordinal": 19,
+        "name": "on_recovery_extra_args",
+        "type_info": "Json"
      }
    ],
    "parameters": {
@@ -94,6 +124,12 @@
      false,
      true,
      false,
+      true,
+      true,
+      true,
+      true,
+      true,
+      true,
      true
    ]
  },
@@ -72,6 +72,36 @@
        "ordinal": 13,
        "name": "on_failure",
        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 14,
+        "name": "on_recovery",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 15,
+        "name": "on_failure_times",
+        "type_info": "Int4"
+      },
+      {
+        "ordinal": 16,
+        "name": "on_failure_exact",
+        "type_info": "Bool"
+      },
+      {
+        "ordinal": 17,
+        "name": "on_failure_extra_args",
+        "type_info": "Json"
+      },
+      {
+        "ordinal": 18,
+        "name": "on_recovery_times",
+        "type_info": "Int4"
+      },
+      {
+        "ordinal": 19,
+        "name": "on_recovery_extra_args",
+        "type_info": "Json"
      }
    ],
    "parameters": {
@@ -95,6 +125,12 @@
      false,
      true,
      false,
+      true,
+      true,
+      true,
+      true,
+      true,
+      true,
      true
    ]
  },
@@ -72,6 +72,36 @@
        "ordinal": 13,
        "name": "on_failure",
        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 14,
+        "name": "on_recovery",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 15,
+        "name": "on_failure_times",
+        "type_info": "Int4"
+      },
+      {
+        "ordinal": 16,
+        "name": "on_failure_exact",
+        "type_info": "Bool"
+      },
+      {
+        "ordinal": 17,
+        "name": "on_failure_extra_args",
+        "type_info": "Json"
+      },
+      {
+        "ordinal": 18,
+        "name": "on_recovery_times",
+        "type_info": "Int4"
+      },
+      {
+        "ordinal": 19,
+        "name": "on_recovery_extra_args",
+        "type_info": "Json"
      }
    ],
    "parameters": {
@@ -95,6 +125,12 @@
      false,
      true,
      false,
+      true,
+      true,
+      true,
+      true,
+      true,
+      true,
      true
    ]
  },
@@ -72,6 +72,36 @@
        "ordinal": 13,
        "name": "on_failure",
        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 14,
+        "name": "on_recovery",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 15,
+        "name": "on_failure_times",
+        "type_info": "Int4"
+      },
+      {
+        "ordinal": 16,
+        "name": "on_failure_exact",
+        "type_info": "Bool"
+      },
+      {
+        "ordinal": 17,
+        "name": "on_failure_extra_args",
+        "type_info": "Json"
+      },
+      {
+        "ordinal": 18,
+        "name": "on_recovery_times",
+        "type_info": "Int4"
+      },
+      {
+        "ordinal": 19,
+        "name": "on_recovery_extra_args",
+        "type_info": "Json"
      }
    ],
    "parameters": {
@@ -95,6 +125,12 @@
      false,
      true,
      false,
+      true,
+      true,
+      true,
+      true,
+      true,
+      true,
      true
    ]
  },
@@ -72,6 +72,36 @@
        "ordinal": 13,
        "name": "on_failure",
        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 14,
+        "name": "on_recovery",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 15,
+        "name": "on_failure_times",
+        "type_info": "Int4"
+      },
+      {
+        "ordinal": 16,
+        "name": "on_failure_exact",
+        "type_info": "Bool"
+      },
+      {
+        "ordinal": 17,
+        "name": "on_failure_extra_args",
+        "type_info": "Json"
+      },
+      {
+        "ordinal": 18,
+        "name": "on_recovery_times",
+        "type_info": "Int4"
+      },
+      {
+        "ordinal": 19,
+        "name": "on_recovery_extra_args",
+        "type_info": "Json"
      }
    ],
    "parameters": {
@@ -93,6 +123,12 @@
      false,
      true,
      false,
+      true,
+      true,
+      true,
+      true,
+      true,
+      true,
      true
    ]
  },
@@ -0,0 +1,2 @@
+-- Add down migration script here
+ALTER TABLE schedule DROP COLUMN on_recovery;
@@ -0,0 +1,2 @@
+-- Add up migration script here
+ALTER TABLE schedule ADD COLUMN on_recovery VARCHAR(1000);
@@ -0,0 +1,7 @@
+-- Add down migration script here
+ALTER TABLE schedule 
+DROP COLUMN on_failure_times,
+DROP COLUMN on_failure_exact,
+DROP COLUMN on_failure_extra_args,
+DROP COLUMN on_recovery_times,
+DROP COLUMN on_recovery_extra_args;
@@ -0,0 +1,7 @@
+-- Add up migration script here
+ALTER TABLE schedule
+ADD COLUMN on_failure_times INTEGER,
+ADD COLUMN on_failure_exact BOOLEAN,
+ADD COLUMN on_failure_extra_args json,
+ADD COLUMN on_recovery_times INTEGER,
+ADD COLUMN on_recovery_extra_args json;
@@ -13,6 +13,59 @@ INSERT INTO workspace_key(workspace_id, kind, key) VALUES

 insert INTO token(token, email, label, super_admin) VALUES ('SECRET_TOKEN', 'test@windmill.dev', 'test token', true);

+INSERT INTO public.script(workspace_id, created_by, content, schema, summary, description, path, hash, language, lock) VALUES (
+'test-workspace',
+'system',
+'
+export async function main(fail: boolean = true) {
+  if (fail) {
+    throw new Error("Failed")
+  }
+
+  return "OK"
+}
+',
+'{"$schema":"https://json-schema.org/draft/2020-12/schema","properties":{"fail":{"default":true,"description":"","type":"boolean"}},"required":[],"type":"object"}',
+'',
+'',
+'f/system/failing_script', -28028598712388162, 'deno', '');
+
+INSERT INTO public.script(workspace_id, created_by, content, schema, summary, description, path, hash, language, lock) VALUES (
+'test-workspace',
+'system',
+'
+export async function main() {
+  return "Error handler";
+}
+',
+'{"$schema":"https://json-schema.org/draft/2020-12/schema","properties":{"path":{"default":null,"description":"","type":"string"},"schedule_path":{"default":null,"description":"","type":"string"},"error":{"default":null,"description":"","properties":{},"type":"object"}},"required":["path","schedule_path","error"],"type":"object"}',
+'',
+'',
+'f/system/schedule_error_handler', -28028598712388161, 'deno', '');
+
+INSERT INTO public.script(workspace_id, created_by, content, schema, summary, description, path, hash, language, lock) VALUES (
+'test-workspace',
+'system',
+'
+export async function main() {
+  return "Recovery handler";
+}
+',
+'{"$schema":"https://json-schema.org/draft/2020-12/schema","properties":{"path":{"default":null,"description":"","type":"string"},"schedule_path":{"default":null,"description":"","type":"string"},"previous_job_error":{"default":null,"description":"","type":"string"},"result":{"default":null,"description":"","type":"string"}},"required":["path","schedule_path","previous_job_error","result"],"type":"object"}',
+'',
+'',
+'f/system/schedule_recovery_handler', -28028598712388160, 'deno', '');
+
+INSERT INTO public.flow(workspace_id, edited_by, value, schema, summary, description, path) VALUES (
+'test-workspace',
+'system',
+'{"modules": [{"id": "a", "value": {"path": "f/system/failing_script", "type": "script", "input_transforms": {"fail": {"expr": "flow_input.fail", "type": "javascript"}}}}]}',
+'{"$schema":"https://json-schema.org/draft/2020-12/schema","properties":{"fail":{"default":true,"description":"","type":"boolean","format":""}},"required":[],"type":"object"}',
+'',
+'',
+'f/system/failing_flow'
+);
+
 GRANT ALL PRIVILEGES ON TABLE workspace_key TO windmill_admin;
 GRANT ALL PRIVILEGES ON TABLE workspace_key TO windmill_user;

@@ -1,12 +1,19 @@
 use std::sync::Arc;

+use chrono::Timelike;
+use futures::StreamExt;
 use futures::{stream, Stream};
 use serde::Deserialize;
 use serde_json::json;
 use sqlx::{postgres::PgListener, types::Uuid, Pool, Postgres, Transaction};
-use tokio::sync::RwLock;
+use tokio::{
+    sync::RwLock,
+    time::{timeout, Duration},
+};
 use windmill_api::jobs::{CompletedJob, Job};
-use windmill_api_client::types::{CreateFlowBody, RawScript};
+use windmill_api_client::types::{
+    CreateFlowBody, EditSchedule, NewSchedule, RawScript, ScriptArgs,
+};
 use windmill_common::{
    flow_status::{FlowStatus, FlowStatusModule},
    flows::{FlowModule, FlowModuleValue, FlowValue, InputTransform},
@@ -2571,3 +2578,274 @@ async fn test_rust_client(db: Pool<Postgres>) {
    .await
    .unwrap();
 }
+
+#[sqlx::test(fixtures("base"))]
+async fn test_script_schedule_handlers(db: Pool<Postgres>) {
+    initialize_tracing().await;
+    let server = ApiServer::start(db.clone()).await;
+    let port = server.addr.port();
+
+    let client = windmill_api_client::create_client(
+        &format!("http://localhost:{port}"),
+        "SECRET_TOKEN".to_string(),
+    );
+
+    let mut args = std::collections::HashMap::new();
+    args.insert("fail".to_string(), json!(true));
+
+    let now = chrono::Utc::now();
+    // add 5 seconds to now
+    let then = now
+        .checked_add_signed(chrono::Duration::seconds(5))
+        .unwrap();
+
+    let schedule = NewSchedule {
+        args: ScriptArgs::from(args),
+        enabled: Some(true),
+        is_flow: false,
+        on_failure: Some("script/f/system/schedule_error_handler".to_string()),
+        on_failure_times: None,
+        on_failure_exact: None,
+        on_failure_extra_args: None,
+        on_recovery: Some("script/f/system/schedule_recovery_handler".to_string()),
+        on_recovery_times: None,
+        on_recovery_extra_args: None,
+        path: "f/system/failing_script_schedule".to_string(),
+        script_path: "f/system/failing_script".to_string(),
+        timezone: "UTC".to_string(),
+        schedule: format!("{} {} * * * *", then.second(), then.minute()).to_string(),
+    };
+
+    let _ = client.create_schedule("test-workspace", &schedule).await;
+
+    let mut str = listen_for_completed_jobs(&db).await;
+
+    let db2 = db.clone();
+    in_test_worker(
+        &db,
+        async move {
+            str.next().await; // completed error job
+
+            let uuid = timeout(Duration::from_millis(5000), str.next()).await; // error handler
+
+            if uuid.is_err() {
+                panic!("schedule error handler was not run within 5 s");
+            }
+
+            let uuid = uuid.unwrap().unwrap();
+
+            let completed_job =
+                sqlx::query_as::<_, CompletedJob>("SELECT * FROM completed_job WHERE id = $1")
+                    .bind(uuid)
+                    .fetch_one(&db2)
+                    .await
+                    .unwrap();
+
+            if completed_job.script_path.is_none()
+                || completed_job.script_path != Some("f/system/schedule_error_handler".to_string())
+            {
+                panic!(
+                    "a script was run after main job execution but was not schedule error handler"
+                );
+            }
+        },
+        port,
+    )
+    .await;
+
+    let mut args = std::collections::HashMap::new();
+    args.insert("fail".to_string(), json!(false));
+    let now = chrono::Utc::now();
+    let then = now
+        .checked_add_signed(chrono::Duration::seconds(5))
+        .unwrap();
+    client
+        .update_schedule(
+            "test-workspace",
+            "f/system/failing_script_schedule",
+            &EditSchedule {
+                args: ScriptArgs::from(args),
+                on_failure: Some("script/f/system/schedule_error_handler".to_string()),
+                on_failure_times: None,
+                on_failure_exact: None,
+                on_failure_extra_args: None,
+                on_recovery: Some("script/f/system/schedule_recovery_handler".to_string()),
+                on_recovery_times: None,
+                on_recovery_extra_args: None,
+                timezone: "UTC".to_string(),
+                schedule: format!("{} {} * * * *", then.second(), then.minute()).to_string(),
+            },
+        )
+        .await
+        .unwrap();
+
+    let mut str = listen_for_completed_jobs(&db).await;
+
+    let db2 = db.clone();
+    in_test_worker(
+        &db,
+        async move {
+            str.next().await; // completed working job
+            let uuid = timeout(Duration::from_millis(5000), str.next()).await; // recovery handler
+
+            if uuid.is_err() {
+                panic!("schedule recovery handler was not run within 5 s");
+            }
+
+            let uuid = uuid.unwrap().unwrap();
+            
+            let completed_job =
+                sqlx::query_as::<_, CompletedJob>("SELECT * FROM completed_job WHERE id = $1")
+                    .bind(uuid)
+                    .fetch_one(&db2)
+                    .await
+                    .unwrap();
+
+            if completed_job.script_path.is_none()
+                || completed_job.script_path
+                    != Some("f/system/schedule_recovery_handler".to_string())
+            {
+                panic!("a script was run after main job execution but was not schedule recovery handler");
+            }
+        },
+        port,
+    )
+    .await;
+}
+
+
+#[sqlx::test(fixtures("base"))]
+async fn test_flow_schedule_handlers(db: Pool<Postgres>) {
+    initialize_tracing().await;
+    let server = ApiServer::start(db.clone()).await;
+    let port = server.addr.port();
+
+    let client = windmill_api_client::create_client(
+        &format!("http://localhost:{port}"),
+        "SECRET_TOKEN".to_string(),
+    );
+
+    let mut args = std::collections::HashMap::new();
+    args.insert("fail".to_string(), json!(true));
+
+    let now = chrono::Utc::now();
+    // add 5 seconds to now
+    let then = now
+        .checked_add_signed(chrono::Duration::seconds(5))
+        .unwrap();
+
+    let schedule = NewSchedule {
+        args: ScriptArgs::from(args),
+        enabled: Some(true),
+        is_flow: true,
+        on_failure: Some("script/f/system/schedule_error_handler".to_string()),
+        on_failure_times: None,
+        on_failure_exact: None,
+        on_failure_extra_args: None,
+        on_recovery: Some("script/f/system/schedule_recovery_handler".to_string()),
+        on_recovery_times: None,
+        on_recovery_extra_args: None,
+        path: "f/system/failing_flow_schedule".to_string(),
+        script_path: "f/system/failing_flow".to_string(),
+        timezone: "UTC".to_string(),
+        schedule: format!("{} {} * * * *", then.second(), then.minute()).to_string(),
+    };
+
+    let _ = client.create_schedule("test-workspace", &schedule).await;
+
+    let mut str = listen_for_completed_jobs(&db).await;
+
+    let db2 = db.clone();
+    in_test_worker(
+        &db,
+        async move {
+            str.next().await; // completed error step
+            str.next().await; // completed error flow
+
+            let uuid = timeout(Duration::from_millis(5000), str.next()).await; // error handler
+
+            if uuid.is_err() {
+                panic!("schedule error handler was not run within 5 s");
+            }
+
+            let uuid = uuid.unwrap().unwrap();
+
+            let completed_job =
+                sqlx::query_as::<_, CompletedJob>("SELECT * FROM completed_job WHERE id = $1")
+                    .bind(uuid)
+                    .fetch_one(&db2)
+                    .await
+                    .unwrap();
+
+            if completed_job.script_path.is_none()
+                || completed_job.script_path != Some("f/system/schedule_error_handler".to_string())
+            {
+                panic!(
+                    "a script was run after main job execution but was not schedule error handler"
+                );
+            }
+        },
+        port,
+    )
+    .await;
+
+    let mut args = std::collections::HashMap::new();
+    args.insert("fail".to_string(), json!(false));
+    let now = chrono::Utc::now();
+    let then = now
+        .checked_add_signed(chrono::Duration::seconds(5))
+        .unwrap();
+    client
+        .update_schedule(
+            "test-workspace",
+            "f/system/failing_flow_schedule",
+            &EditSchedule {
+                args: ScriptArgs::from(args),
+                on_failure: Some("script/f/system/schedule_error_handler".to_string()),
+                on_failure_times: None,
+                on_failure_exact: None,
+                on_failure_extra_args: None,
+                on_recovery: Some("script/f/system/schedule_recovery_handler".to_string()),
+                on_recovery_times: None,
+                on_recovery_extra_args: None,
+                timezone: "UTC".to_string(),
+                schedule: format!("{} {} * * * *", then.second(), then.minute()).to_string(),
+            },
+        )
+        .await
+        .unwrap();
+
+    let mut str = listen_for_completed_jobs(&db).await;
+
+    let db2 = db.clone();
+    in_test_worker(
+        &db,
+        async move {
+            str.next().await; // completed working step
+            str.next().await; // completed working flow
+            let uuid = timeout(Duration::from_millis(5000), str.next()).await; // recovery handler
+
+            if uuid.is_err() {
+                panic!("schedule recovery handler was not run within 5 s");
+            }
+
+            let uuid = uuid.unwrap().unwrap();
+            
+            let completed_job =
+                sqlx::query_as::<_, CompletedJob>("SELECT * FROM completed_job WHERE id = $1")
+                    .bind(uuid)
+                    .fetch_one(&db2)
+                    .await
+                    .unwrap();
+
+            if completed_job.script_path.is_none()
+                || completed_job.script_path
+                    != Some("f/system/schedule_recovery_handler".to_string())
+            {
+                panic!("a script was run after main job execution but was not schedule recovery handler");
+            }
+        },
+        port,
+    )
+    .await;
+}
@@ -1873,8 +1873,7 @@ paths:

  /w/{workspace}/resources/get_value_interpolated/{path}:
    get:
-      summary:
-        get resource interpolated (variables and resources are fully unrolled)
+      summary: get resource interpolated (variables and resources are fully unrolled)
      operationId: getResourceValueInterpolated
      tags:
        - resource
@@ -2329,8 +2328,7 @@ paths:
          schema:
            type: string
        - name: first_parent_hash
-          description:
-            mask to filter scripts whom first direct parent has exact hash
+          description: mask to filter scripts whom first direct parent has exact hash
          in: query
          schema:
            type: string
@@ -2498,8 +2496,7 @@ paths:

  /workers/custom_tags:
    get:
-      summary:
-        get all instance custom tags (tags are used to dispatch jobs to
+      summary: get all instance custom tags (tags are used to dispatch jobs to
        different worker groups)
      operationId: getCustomTags
      tags:
@@ -2550,8 +2547,7 @@ paths:

  /w/{workspace}/scripts/delete/h/{hash}:
    post:
-      summary:
-        delete script by hash (erase content but keep hash, require admin)
+      summary: delete script by hash (erase content but keep hash, require admin)
      operationId: deleteScriptByHash
      tags:
        - script
@@ -2743,16 +2739,14 @@ paths:
            type: string
            format: date-time
        - name: scheduled_in_secs
-          description:
-            schedule the script to execute in the number of seconds starting now
+          description: schedule the script to execute in the number of seconds starting now
          in: query
          schema:
            type: integer
        - $ref: "#/components/parameters/ParentJob"
        - $ref: "#/components/parameters/NewJobId"
        - name: invisible_to_owner
-          description:
-            make the run invisible to the the script owner (default false)
+          description: make the run invisible to the the script owner (default false)
          in: query
          schema:
            type: boolean
@@ -3630,8 +3624,7 @@ paths:
            type: string
            format: date-time
        - name: scheduled_in_secs
-          description:
-            schedule the script to execute in the number of seconds starting now
+          description: schedule the script to execute in the number of seconds starting now
          in: query
          schema:
            type: integer
@@ -3639,8 +3632,7 @@ paths:
        - $ref: "#/components/parameters/NewJobId"
        - $ref: "#/components/parameters/IncludeHeader"
        - name: invisible_to_owner
-          description:
-            make the run invisible to the the flow owner (default false)
+          description: make the run invisible to the the flow owner (default false)
          in: query
          schema:
            type: boolean
@@ -3678,8 +3670,7 @@ paths:
            type: string
            format: date-time
        - name: scheduled_in_secs
-          description:
-            schedule the script to execute in the number of seconds starting now
+          description: schedule the script to execute in the number of seconds starting now
          in: query
          schema:
            type: integer
@@ -3687,8 +3678,7 @@ paths:
        - $ref: "#/components/parameters/NewJobId"
        - $ref: "#/components/parameters/IncludeHeader"
        - name: invisible_to_owner
-          description:
-            make the run invisible to the the script owner (default false)
+          description: make the run invisible to the the script owner (default false)
          in: query
          schema:
            type: boolean
@@ -3719,8 +3709,7 @@ paths:
        - $ref: "#/components/parameters/WorkspaceId"
        - $ref: "#/components/parameters/IncludeHeader"
        - name: invisible_to_owner
-          description:
-            make the run invisible to the the script owner (default false)
+          description: make the run invisible to the the script owner (default false)
          in: query
          schema:
            type: boolean
@@ -3753,8 +3742,7 @@ paths:
        - $ref: "#/components/parameters/WorkspaceId"
        - $ref: "#/components/parameters/IncludeHeader"
        - name: invisible_to_owner
-          description:
-            make the run invisible to the the script owner (default false)
+          description: make the run invisible to the the script owner (default false)
          in: query
          schema:
            type: boolean
@@ -3947,8 +3935,7 @@ paths:
        - job
      responses:
        "200":
-          description:
-            the timestamp of the db that can be used to compute the drift
+          description: the timestamp of the db that can be used to compute the drift
          content:
            application/json:
              schema:
@@ -4181,8 +4168,7 @@ paths:

  /w/{workspace}/jobs/resume_urls/{id}/{resume_id}:
    get:
-      summary:
-        get resume urls given a job_id, resume_id and a nonce to resume a flow
+      summary: get resume urls given a job_id, resume_id and a nonce to resume a flow
      operationId: getResumeUrls
      tags:
        - job
@@ -4695,8 +4681,7 @@ paths:
        - $ref: "#/components/parameters/WorkspaceId"
        - name: only_member_of
          in: query
-          description:
-            only list the groups the user is member of (default false)
+          description: only list the groups the user is member of (default false)
          schema:
            type: boolean
      responses:
@@ -4884,8 +4869,7 @@ paths:
        - $ref: "#/components/parameters/WorkspaceId"
        - name: only_member_of
          in: query
-          description:
-            only list the folders the user is member of (default false)
+          description: only list the folders the user is member of (default false)
          schema:
            type: boolean
      responses:
@@ -5523,8 +5507,7 @@ components:
        type: integer
    PerPage:
      name: per_page
-      description:
-        number of items to return for a given page (default 30, max 100)
+      description: number of items to return for a given page (default 30, max 100)
      in: query
      schema:
        type: integer
@@ -5658,8 +5641,7 @@ components:
        type: boolean
    ArgsFilter:
      name: args
-      description:
-        filter on jobs containing those args as a json subset (@> in postgres)
+      description: filter on jobs containing those args as a json subset (@> in postgres)
      in: query
      schema:
        type: string
@@ -5671,8 +5653,7 @@ components:
        type: string
    ResultFilter:
      name: result
-      description:
-        filter on jobs containing those result as a json subset (@> in postgres)
+      description: filter on jobs containing those result as a json subset (@> in postgres)
      in: query
      schema:
        type: string
@@ -6830,6 +6811,18 @@ components:
        on_failure:
          # a reference to a script path, flow path, or webhook (script/<path>, flow/<path>)
          type: string
+        on_failure_times:
+          type: number
+        on_failure_exact:
+          type: boolean
+        on_failure_extra_args:
+          $ref: "#/components/schemas/ScriptArgs"
+        on_recovery:
+          type: string
+        on_recovery_times:
+          type: number
+        on_recovery_extra_args:
+          $ref: "#/components/schemas/ScriptArgs"
      required:
        - path
        - edited_by
@@ -6883,6 +6876,18 @@ components:
        on_failure:
          # a reference to a script path, flow path, or webhook (script/<path>, flow/<path>)
          type: string
+        on_failure_times:
+          type: number
+        on_failure_exact:
+          type: boolean
+        on_failure_extra_args:
+          $ref: "#/components/schemas/ScriptArgs"
+        on_recovery:
+          type: string
+        on_recovery_times:
+          type: number
+        on_recovery_extra_args:
+          $ref: "#/components/schemas/ScriptArgs"
      required:
        - path
        - schedule
@@ -6903,6 +6908,18 @@ components:
        on_failure:
          # a reference to a script path, flow path, or webhook (script/<path>, flow/<path>)
          type: string
+        on_failure_times:
+          type: number
+        on_failure_exact:
+          type: boolean
+        on_failure_extra_args:
+          $ref: "#/components/schemas/ScriptArgs"
+        on_recovery:
+          type: string
+        on_recovery_times:
+          type: number
+        on_recovery_extra_args:
+          $ref: "#/components/schemas/ScriptArgs"
      required:
        - schedule
        - timezone
@@ -56,6 +56,12 @@ pub struct NewSchedule {
    pub args: Option<serde_json::Value>,
    pub enabled: Option<bool>,
    pub on_failure: Option<String>,
+    pub on_failure_times: Option<i32>,
+    pub on_failure_exact: Option<bool>,
+    pub on_failure_extra_args: Option<serde_json::Value>,
+    pub on_recovery: Option<String>,
+    pub on_recovery_times: Option<i32>,
+    pub on_recovery_extra_args: Option<serde_json::Value>,
 }

 async fn check_path_conflict<'c>(
@@ -105,7 +111,7 @@ async fn create_schedule(
    let schedule = sqlx::query_as!(
        Schedule,
        "INSERT INTO schedule (workspace_id, path, schedule, timezone, edited_by, script_path, \
-         is_flow, args, enabled, email, on_failure) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11) RETURNING *",
+         is_flow, args, enabled, email, on_failure, on_failure_times, on_failure_exact, on_failure_extra_args, on_recovery, on_recovery_times, on_recovery_extra_args) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17) RETURNING *",
        w_id,
        ns.path,
        ns.schedule,
@@ -116,7 +122,13 @@ async fn create_schedule(
        ns.args,
        ns.enabled.unwrap_or(false),
        &authed.email,
-        ns.on_failure
+        ns.on_failure,
+        ns.on_failure_times,
+        ns.on_failure_exact,
+        ns.on_failure_extra_args,
+        ns.on_recovery,
+        ns.on_recovery_times,
+        ns.on_recovery_extra_args,
    )
    .fetch_one(&mut tx)
    .await
@@ -178,12 +190,18 @@ async fn edit_schedule(
    clear_schedule(tx.transaction_mut(), path, is_flow, &w_id).await?;
    let schedule = sqlx::query_as!(
        Schedule,
-        "UPDATE schedule SET schedule = $1, timezone = $2, args = $3, on_failure = $4 WHERE path \
-         = $5 AND workspace_id = $6 RETURNING *",
+        "UPDATE schedule SET schedule = $1, timezone = $2, args = $3, on_failure = $4, on_failure_times = $5, on_failure_exact = $6, on_failure_extra_args = $7, on_recovery = $8, on_recovery_times = $9, on_recovery_extra_args = $10 WHERE path \
+         = $11 AND workspace_id = $12 RETURNING *",
        es.schedule,
        es.timezone,
        es.args,
        es.on_failure,
+        es.on_failure_times,
+        es.on_failure_exact,
+        es.on_failure_extra_args,
+        es.on_recovery,
+        es.on_recovery_times,
+        es.on_recovery_extra_args,
        path,
        w_id,
    )
@@ -268,6 +286,12 @@ pub struct ScheduleWJobs {
    pub email: String,
    pub error: Option<String>,
    pub on_failure: Option<String>,
+    pub on_failure_times: Option<i32>,
+    pub on_failure_exact: Option<bool>,
+    pub on_failure_extra_args: Option<serde_json::Value>,
+    pub on_recovery: Option<String>,
+    pub on_recovery_times: Option<i32>,
+    pub on_recovery_extra_args: Option<serde_json::Value>,
    pub jobs: Option<Vec<serde_json::Value>>,
 }

@@ -464,6 +488,12 @@ pub struct EditSchedule {
    pub timezone: String,
    pub args: Option<serde_json::Value>,
    pub on_failure: Option<String>,
+    pub on_failure_times: Option<i32>,
+    pub on_failure_exact: Option<bool>,
+    pub on_failure_extra_args: Option<serde_json::Value>,
+    pub on_recovery: Option<String>,
+    pub on_recovery_times: Option<i32>,
+    pub on_recovery_extra_args: Option<serde_json::Value>,
 }

 pub async fn clear_schedule<'c>(
@@ -27,6 +27,12 @@ pub struct Schedule {
    #[serde(skip_serializing_if = "Option::is_none")]
    pub error: Option<String>,
    pub on_failure: Option<String>,
+    pub on_failure_times: Option<i32>,
+    pub on_failure_exact: Option<bool>,
+    pub on_failure_extra_args: Option<serde_json::Value>,
+    pub on_recovery: Option<String>,
+    pub on_recovery_times: Option<i32>,
+    pub on_recovery_extra_args: Option<serde_json::Value>,
 }

 pub fn schedule_to_user(path: &str) -> String {
@@ -364,19 +364,35 @@ pub async fn add_completed_job<R: rsmq_async::RsmqConnection + Clone + Send>(

    tx = delete_job(tx, &queued_job.workspace_id, job_id).await?;
    if !queued_job.is_flow_step
-        && queued_job.job_kind != JobKind::Flow
-        && queued_job.job_kind != JobKind::FlowPreview
        && queued_job.schedule_path.is_some()
        && queued_job.script_path.is_some()
    {
-        tx = handle_maybe_scheduled_job(
+        tx = apply_schedule_handlers(
            tx,
            db,
            queued_job.schedule_path.as_ref().unwrap(),
            queued_job.script_path.as_ref().unwrap(),
            &queued_job.workspace_id,
            success,
-            if success { None } else { Some(&result) },
+            &result,
+            job_id,
+            queued_job.started_at.unwrap_or(chrono::Utc::now()),
+        )
+        .await?;
+    }
+    if !queued_job.is_flow_step
+        && queued_job.job_kind != JobKind::Flow
+        && queued_job.job_kind != JobKind::FlowPreview
+        && queued_job.schedule_path.is_some()
+        && queued_job.script_path.is_some()
+    {
+        // script only
+        tx = handle_maybe_scheduled_job(
+            tx,
+            db,
+            queued_job.schedule_path.as_ref().unwrap(),
+            queued_job.script_path.as_ref().unwrap(),
+            &queued_job.workspace_id,
        )
        .await?;
    }
@@ -543,8 +559,6 @@ pub async fn handle_maybe_scheduled_job<'c, R: rsmq_async::RsmqConnection + Clon
    schedule_path: &str,
    script_path: &str,
    w_id: &str,
-    success: bool,
-    result: Option<&serde_json::Value>,
 ) -> windmill_common::error::Result<QueueTransaction<'c, R>> {
    let schedule = get_schedule_opt(tx.transaction_mut(), w_id, schedule_path).await?;

@@ -558,46 +572,6 @@ pub async fn handle_maybe_scheduled_job<'c, R: rsmq_async::RsmqConnection + Clon
    let schedule = schedule.unwrap();

    if schedule.enabled && script_path == schedule.script_path {
-        if !success {
-            if let Some(on_failure_path) = schedule.on_failure.clone() {
-                let on_failure_result = handle_on_failure(
-                    db,
-                    tx,
-                    schedule_path,
-                    script_path,
-                    w_id,
-                    &on_failure_path,
-                    result,
-                    &schedule.email,
-                    &schedule_to_user(&schedule.path),
-                    username_to_permissioned_as(&schedule.edited_by),
-                )
-                .await;
-
-                match on_failure_result {
-                    Ok(ntx) => {
-                        tx = ntx;
-                    }
-                    Err(err) => {
-                        sqlx::query!(
-                        "UPDATE schedule SET enabled = false, error = $1 WHERE workspace_id = $2 AND path = $3",
-                        format!("Could not trigger error handler: {err}"),
-                        &schedule.workspace_id,
-                        &schedule.path
-                    )
-                    .execute(db)
-                    .await?;
-                        tracing::warn!(
-                            "Could not trigger error handler for {}: {}",
-                            schedule_path,
-                            err
-                        );
-                        return Err(err);
-                    }
-                }
-            }
-        }
-
        let res = push_scheduled_job(
            db,
            tx,
@@ -618,6 +592,12 @@ pub async fn handle_maybe_scheduled_job<'c, R: rsmq_async::RsmqConnection + Clon
                email: schedule.email,
                error: None,
                on_failure: schedule.on_failure,
+                on_failure_times: schedule.on_failure_times,
+                on_failure_exact: schedule.on_failure_exact,
+                on_failure_extra_args: schedule.on_failure_extra_args,
+                on_recovery: schedule.on_recovery,
+                on_recovery_times: schedule.on_recovery_times,
+                on_recovery_extra_args: schedule.on_recovery_extra_args,
            },
        )
        .await;
@@ -641,28 +621,211 @@ pub async fn handle_maybe_scheduled_job<'c, R: rsmq_async::RsmqConnection + Clon
    }
 }

+#[derive(Clone)]
+struct CompletedJobSubset {
+    success: bool,
+    result: Option<serde_json::Value>,
+    started_at: chrono::DateTime<chrono::Utc>,
+}
+async fn apply_schedule_handlers<'c, R: rsmq_async::RsmqConnection + Clone + Send + 'c>(
+    mut tx: QueueTransaction<'c, R>,
+    db: &Pool<Postgres>,
+    schedule_path: &str,
+    script_path: &str,
+    w_id: &str,
+    success: bool,
+    result: &serde_json::Value,
+    job_id: Uuid,
+    started_at: DateTime<Utc>,
+) -> windmill_common::error::Result<QueueTransaction<'c, R>> {
+    let schedule = get_schedule_opt(tx.transaction_mut(), w_id, schedule_path).await?;
+
+    if schedule.is_none() {
+        tracing::error!(
+            "Schedule {schedule_path} in {w_id} not found. Impossible to apply schedule handlers"
+        );
+        return Ok(tx);
+    }
+
+    let schedule = schedule.unwrap();
+
+    if !success {
+        if let Some(on_failure_path) = schedule.on_failure.clone() {
+            let times = schedule.on_failure_times.unwrap_or(1).max(1);
+            let exact = schedule.on_failure_exact.unwrap_or(false);
+            if times > 1 || exact {
+                let past_jobs = sqlx::query_as!(
+                    CompletedJobSubset,
+                    "SELECT success, result, started_at FROM completed_job WHERE workspace_id = $1 AND schedule_path = $2 AND script_path = $3 AND id != $4 ORDER BY created_at DESC LIMIT $5",
+                    &schedule.workspace_id,
+                    &schedule.path,
+                    &schedule.script_path,
+                    job_id,
+                    if exact { times } else { times - 1 } as i64,
+                ).fetch_all(&mut tx).await?;
+
+                let match_times = if exact {
+                    past_jobs.len() == times as usize
+                        && past_jobs[..(times - 1) as usize].iter().all(|j| !j.success)
+                        && past_jobs[(times - 1) as usize].success
+                } else {
+                    past_jobs.len() == ((times - 1) as usize)
+                        && past_jobs.iter().all(|j| !j.success)
+                };
+
+                if !match_times {
+                    return Ok(tx);
+                }
+            }
+
+            let on_failure_result = handle_on_failure(
+                db,
+                tx,
+                schedule_path,
+                script_path,
+                schedule.is_flow,
+                w_id,
+                &on_failure_path,
+                result,
+                times,
+                started_at,
+                schedule.on_failure_extra_args,
+                &schedule.email,
+                &schedule_to_user(&schedule.path),
+                username_to_permissioned_as(&schedule.edited_by),
+            )
+            .await;
+
+            match on_failure_result {
+                Ok(ntx) => {
+                    tx = ntx;
+                }
+                Err(err) => {
+                    sqlx::query!(
+                    "UPDATE schedule SET enabled = false, error = $1 WHERE workspace_id = $2 AND path = $3",
+                    format!("Could not trigger error handler: {err}"),
+                    &schedule.workspace_id,
+                    &schedule.path
+                )
+                .execute(db)
+                .await?;
+                    tracing::warn!(
+                        "Could not trigger error handler for {}: {}",
+                        schedule_path,
+                        err
+                    );
+                    return Err(err);
+                }
+            }
+        }
+    } else {
+        if let Some(on_recovery_path) = schedule.on_recovery.clone() {
+            let times = schedule.on_recovery_times.unwrap_or(1).max(1);
+            let past_jobs = sqlx::query_as!(
+                CompletedJobSubset,
+                "SELECT success, result, started_at FROM completed_job WHERE workspace_id = $1 AND schedule_path = $2 AND script_path = $3 AND id != $4 ORDER BY created_at DESC LIMIT $5",
+                &schedule.workspace_id,
+                &schedule.path,
+                &schedule.script_path,
+                job_id,
+                times as i64,
+            ).fetch_all(&mut tx).await?;
+
+            if past_jobs.len() < times as usize {
+                return Ok(tx);
+            }
+
+            let n_times_successful = past_jobs[..(times - 1) as usize].iter().all(|j| j.success);
+
+            if !n_times_successful {
+                return Ok(tx);
+            }
+
+            let failed_job = past_jobs[past_jobs.len() - 1].clone();
+
+            if !failed_job.success {
+                let on_recovery_result = handle_on_recovery(
+                    db,
+                    tx,
+                    schedule_path,
+                    script_path,
+                    schedule.is_flow,
+                    w_id,
+                    &on_recovery_path,
+                    failed_job,
+                    result,
+                    times,
+                    started_at,
+                    schedule.on_recovery_extra_args,
+                    &schedule.email,
+                    &schedule_to_user(&schedule.path),
+                    username_to_permissioned_as(&schedule.edited_by),
+                )
+                .await;
+
+                match on_recovery_result {
+                    Ok(ntx) => {
+                        tx = ntx;
+                    }
+                    Err(err) => {
+                        sqlx::query!(
+                            "UPDATE schedule SET enabled = false, error = $1 WHERE workspace_id = $2 AND path = $3",
+                            format!("Could not trigger recovery handler: {err}"),
+                            &schedule.workspace_id,
+                            &schedule.path
+                        )
+                        .execute(db)
+                        .await?;
+                        tracing::warn!(
+                            "Could not trigger recovery handler for {}: {}",
+                            schedule_path,
+                            err
+                        );
+                        return Err(err);
+                    }
+                }
+            }
+        }
+    }
+
+    Ok(tx)
+}
+
 async fn handle_on_failure<'c, R: rsmq_async::RsmqConnection + Clone + Send + 'c>(
    db: &Pool<Postgres>,
    tx: QueueTransaction<'c, R>,
    schedule_path: &str,
    script_path: &str,
+    is_flow: bool,
    w_id: &str,
    on_failure_path: &str,
-    result: Option<&serde_json::Value>,
+    result: &serde_json::Value,
+    failed_times: i32,
+    started_at: DateTime<Utc>,
+    extra_args: Option<serde_json::Value>,
    username: &str,
    email: &str,
    permissioned_as: String,
 ) -> windmill_common::error::Result<QueueTransaction<'c, R>> {
    let (payload, tag) = get_payload_tag_from_prefixed_path(on_failure_path, db, w_id).await?;

-    let mut args = result
-        .map(|x| x.clone())
-        .unwrap_or_else(|| json!({}))
-        .as_object()
-        .unwrap()
-        .clone();
+    let mut args = result.clone().as_object().unwrap().clone();
    args.insert("schedule_path".to_string(), json!(schedule_path));
    args.insert("path".to_string(), json!(script_path));
+    args.insert("is_flow".to_string(), json!(is_flow));
+    args.insert("started_at".to_string(), json!(started_at));
+    args.insert("failed_times".to_string(), json!(failed_times));
+
+    if let Some(args_v) = extra_args {
+        if let serde_json::Value::Object(args_m) = args_v {
+            args.extend(args_m);
+        } else {
+            return Err(error::Error::ExecutionErr(
+                "args of scripts needs to be dict".to_string(),
+            ));
+        }
+    }
+
    let tx = PushIsolationLevel::Transaction(tx);
    let (uuid, tx) = push(
        &db,
@@ -695,6 +858,82 @@ async fn handle_on_failure<'c, R: rsmq_async::RsmqConnection + Clone + Send + 'c
    return Ok(tx);
 }

+async fn handle_on_recovery<'c, R: rsmq_async::RsmqConnection + Clone + Send + 'c>(
+    db: &Pool<Postgres>,
+    tx: QueueTransaction<'c, R>,
+    schedule_path: &str,
+    script_path: &str,
+    is_flow: bool,
+    w_id: &str,
+    on_recovery_path: &str,
+    error_job: CompletedJobSubset,
+    successful_job_result: &serde_json::Value,
+    successful_times: i32,
+    successful_job_started_at: DateTime<Utc>,
+    extra_args: Option<serde_json::Value>,
+    username: &str,
+    email: &str,
+    permissioned_as: String,
+) -> windmill_common::error::Result<QueueTransaction<'c, R>> {
+    let (payload, tag) = get_payload_tag_from_prefixed_path(on_recovery_path, db, w_id).await?;
+
+    let mut args = error_job
+        .result
+        .unwrap_or(json!({}))
+        .as_object()
+        .unwrap()
+        .clone();
+    args.insert("error_started_at".to_string(), json!(error_job.started_at));
+    args.insert("schedule_path".to_string(), json!(schedule_path));
+    args.insert("path".to_string(), json!(script_path));
+    args.insert("is_flow".to_string(), json!(is_flow));
+    args.insert("success_result".to_string(), successful_job_result.clone());
+    args.insert("success_times".to_string(), json!(successful_times));
+    args.insert(
+        "success_started_at".to_string(),
+        json!(successful_job_started_at),
+    );
+    if let Some(args_v) = extra_args {
+        if let serde_json::Value::Object(args_m) = args_v {
+            args.extend(args_m);
+        } else {
+            return Err(error::Error::ExecutionErr(
+                "args of scripts needs to be dict".to_string(),
+            ));
+        }
+    }
+    let tx = PushIsolationLevel::Transaction(tx);
+    let (uuid, tx) = push(
+        &db,
+        tx,
+        w_id,
+        payload,
+        args,
+        username,
+        email,
+        permissioned_as,
+        None,
+        None,
+        None,
+        None,
+        None,
+        false,
+        false,
+        None,
+        true,
+        tag,
+        None,
+        None,
+    )
+    .await?;
+    tracing::info!(
+        "Pushed on_recovery job {} for {} to queue",
+        uuid,
+        schedule_path
+    );
+    return Ok(tx);
+}
+
 pub async fn pull<R: rsmq_async::RsmqConnection + Send + Clone>(
    db: &Pool<Postgres>,
    rsmq: Option<R>,
@@ -533,8 +533,6 @@ pub async fn update_flow_status_after_job_completion_internal<
                flow_job.schedule_path.as_ref().unwrap(),
                flow_job.script_path.as_ref().unwrap(),
                &w_id,
-                success,
-                if success { None } else { Some(&nresult) },
            )
            .await?;
        }
@@ -1,5 +1,5 @@
 <script lang="ts">
-	import { Alert, Button } from '$lib/components/common'
+	import { Alert, Button, Tab, Tabs } from '$lib/components/common'
 	import Drawer from '$lib/components/common/drawer/Drawer.svelte'
 	import DrawerContent from '$lib/components/common/drawer/DrawerContent.svelte'
 	import CronInput from '$lib/components/CronInput.svelte'
@@ -11,9 +11,14 @@
 	import Tooltip from '$lib/components/Tooltip.svelte'
 	import { FlowService, ScheduleService, Script, ScriptService, type Flow } from '$lib/gen'
 	import { userStore, workspaceStore } from '$lib/stores'
-	import { canWrite, emptyString, formatCron, sendUserToast } from '$lib/utils'
+	import { canWrite, emptySchema, emptyString, formatCron, sendUserToast } from '$lib/utils'
 	import { faList, faSave } from '@fortawesome/free-solid-svg-icons'
 	import { createEventDispatcher } from 'svelte'
+	import { inferArgs } from '$lib/infer'
+	import type { Schema, SupportedLanguage } from '$lib/common'
+
+	const slackErrorHandler = 'hub/2431/slack/schedule-error-handler-slack'
+	const slackRecoveryHandler = 'hub/2430/slack/schedule-recovery-handler-slack'

 	let initialPath = ''
 	let edit = true
@@ -23,6 +28,17 @@
 	let itemKind: 'flow' | 'script' = 'script'
 	let errorHandleritemKind: 'flow' | 'script' = 'script'
 	let errorHandlerPath: string | undefined = undefined
+	let recoveryHandlerPath: string | undefined = undefined
+	let errorHandlerSelected: 'custom' | 'slack' = 'custom'
+	let errorHandlerSchema: Schema | undefined = undefined
+	let errorHandlerExtraArgs: Record<string, any> = {}
+	let recoveryHandlerSelected: 'custom' | 'slack' = 'custom'
+	let recoveryHandlerItemKind: 'flow' | 'script' = 'script'
+	let recoveryHandlerSchema: Schema | undefined = undefined
+	let recoveryHandlerExtraArgs: Record<string, any> = {}
+	let failedTimes = 1
+	let failedExact = false
+	let recoveredTimes = 1

 	let script_path = ''
 	let initialScriptPath = ''
@@ -105,10 +121,29 @@
 				let splitted = s.on_failure.split('/')
 				errorHandleritemKind = splitted[0] as 'flow' | 'script'
 				errorHandlerPath = splitted.slice(1)?.join('/')
+				failedTimes = s.on_failure_times ?? 1
+				failedExact = s.on_failure_exact ?? false
+				errorHandlerExtraArgs = s.on_failure_extra_args ?? {}
+				if (errorHandlerPath == slackErrorHandler) {
+					errorHandlerSelected = 'slack'
+				}
 			} else {
 				errorHandlerPath = undefined
 				errorHandleritemKind = 'script'
 			}
+			if (s.on_recovery) {
+				let splitted = s.on_recovery.split('/')
+				recoveryHandlerItemKind = splitted[0] as 'flow' | 'script'
+				recoveryHandlerPath = splitted.slice(1)?.join('/')
+				recoveredTimes = s.on_recovery_times ?? 1
+				recoveryHandlerExtraArgs = s.on_recovery_extra_args ?? {}
+				if (recoveryHandlerPath == slackRecoveryHandler) {
+					recoveryHandlerSelected = 'slack'
+				}
+			} else {
+				recoveryHandlerPath = undefined
+				recoveryHandlerItemKind = 'script'
+			}
 			args = s.args ?? {}
 			can_write = canWrite(s.path, s.extra_perms, $userStore)
 		} catch (err) {
@@ -125,7 +160,15 @@
 					schedule: formatCron(schedule),
 					timezone,
 					args,
-					on_failure: errorHandlerPath ? `${errorHandleritemKind}/${errorHandlerPath}` : undefined
+					on_failure: errorHandlerPath ? `${errorHandleritemKind}/${errorHandlerPath}` : undefined,
+					on_failure_times: failedTimes,
+					on_failure_exact: failedExact,
+					on_failure_extra_args: errorHandlerExtraArgs,
+					on_recovery: recoveryHandlerPath
+						? `${recoveryHandlerItemKind}/${recoveryHandlerPath}`
+						: undefined,
+					on_recovery_times: recoveredTimes,
+					on_recovery_extra_args: recoveryHandlerExtraArgs
 				}
 			})
 			sendUserToast(`Schedule ${path} updated`)
@@ -140,7 +183,15 @@
 					is_flow,
 					args,
 					enabled: true,
-					on_failure: errorHandlerPath ? `${errorHandleritemKind}/${errorHandlerPath}` : undefined
+					on_failure: errorHandlerPath ? `${errorHandleritemKind}/${errorHandlerPath}` : undefined,
+					on_failure_times: failedTimes,
+					on_failure_exact: failedExact,
+					on_failure_extra_args: errorHandlerExtraArgs,
+					on_recovery: recoveryHandlerPath
+						? `${recoveryHandlerItemKind}/${recoveryHandlerPath}`
+						: undefined,
+					on_recovery_times: recoveredTimes,
+					on_recovery_extra_args: recoveryHandlerExtraArgs
 				}
 			})
 			sendUserToast(`Schedule ${path} created`)
@@ -149,6 +200,36 @@
 		drawer.closeDrawer()
 	}

+	async function loadHandlerScriptArgs(p: string, defaultArgs: string[] = []) {
+		try {
+			let schema: Schema | undefined = emptySchema()
+			if (p.startsWith('hub/')) {
+				const hubScript = await ScriptService.getHubScriptByPath({
+					path: p
+				})
+
+				if (hubScript.schema?.properties) {
+					schema = hubScript.schema
+				} else {
+					await inferArgs(hubScript.language as SupportedLanguage, hubScript.content ?? '', schema)
+				}
+			} else {
+				const script = await ScriptService.getScriptByPath({ workspace: $workspaceStore!, path: p })
+				schema = script.schema as Schema
+			}
+			if (schema && schema.properties) {
+				for (let key in schema.properties) {
+					if (defaultArgs.includes(key)) {
+						delete schema.properties[key]
+					}
+				}
+				return schema
+			}
+		} catch (err) {
+			sendUserToast(`Could not query handler schema: ${err}`, true)
+		}
+	}
+
 	$: {
 		if ($workspaceStore) {
 			if (edit && path != '') {
@@ -157,6 +238,32 @@
 		}
 	}

+	$: errorHandlerPath &&
+		loadHandlerScriptArgs(errorHandlerPath, [
+			'path',
+			'is_flow',
+			'schedule_path',
+			'error',
+			'failed_times',
+			'started_at'
+		]).then((schema) => (errorHandlerSchema = schema))
+	$: recoveryHandlerPath &&
+		loadHandlerScriptArgs(recoveryHandlerPath, [
+			'path',
+			'is_flow',
+			'schedule_path',
+			'error',
+			'error_started_at',
+			'success_times',
+			'success_result',
+			'success_started_at'
+		]).then((schema) => (recoveryHandlerSchema = schema))
+
+	$: errorHandlerSelected === 'slack' && (errorHandlerPath = slackErrorHandler)
+	$: errorHandlerSelected === 'custom' && (errorHandlerPath = undefined)
+	$: recoveryHandlerSelected === 'slack' && (recoveryHandlerPath = slackRecoveryHandler)
+	$: recoveryHandlerSelected === 'custom' && (recoveryHandlerPath = undefined)
+
 	let drawer: Drawer
 </script>

@@ -217,7 +324,7 @@

 			<div class="flex flex-row items-center mb-2 gap-1">
 				<div class="text-xl font-extrabold">Schedule</div>
-				<Tooltip>Schedules use CRON syntax. Seconds are mandatory.</Tooltip>
+				<Tooltip light>Schedules use CRON syntax. Seconds are mandatory.</Tooltip>
 			</div>

 			<CronInput disabled={!can_write} bind:schedule bind:timezone bind:validCRON />
@@ -264,32 +371,166 @@
 					</div>
 				{/if}
 			</div>
-			<h2 class="border-b pb-1 mt-8 mb-2">Error Handler</h2>
+			<h2 class="border-b pb-1 mt-8 mb-2"
+				>Error Handler <Tooltip light>
+					<div class="flex gap-20 items-start mt-3">
+						<div class="text-tertiary text-sm"
+							>The following args will be passed to the error handler:
+							<ul class="mt-1 ml-2">
+								<li><b>path</b>: The path of the script or flow that failed.</li>
+								<li><b>is_flow</b>: Whether the runnable is a flow.</li>
+								<li><b>schedule_path</b>: The path of the schedule.</li>
+								<li><b>error</b>: The error details.</li>
+								<li
+									><b>failed_times</b>: Minimum number of times the schedule failed before calling
+									the error handler.</li
+								>
+								<li><b>started_at</b>: The start datetime of the latest job that failed.</li>
+							</ul>
+						</div>
+					</div>
+				</Tooltip></h2
+			>

-			<ScriptPicker
-				disabled={!can_write}
-				initialPath={errorHandlerPath}
-				kind={Script.kind.SCRIPT}
-				allowFlow={true}
-				bind:scriptPath={errorHandlerPath}
-				bind:itemKind={errorHandleritemKind}
-				canRefresh
-			/>
-			<div class="flex gap-20 items-start mt-3">
-				<div class="text-tertiary text-sm"
-					>The following args will be passed to the error handler:
-					<ul class="mt-1 ml-2">
-						<li><b>path</b>: The path of the script or flow that errored.</li>
-						<li><b>schedule_path</b>: The path of the schedule.</li>
-						<li><b>error</b>: The error details.</li>
-					</ul>
-				</div>
-				<Button
-					wrapperClasses="mt-6"
-					href="/scripts/add?hub=hub%2F1087%2Fwindmill%2Fschedule_error_handler_template"
-					target="_blank">Use template</Button
-				>
+			<div>
+				<Tabs bind:selected={errorHandlerSelected} class="mt-2 mb-4">
+					<Tab value="custom">Custom</Tab>
+					<Tab value="slack">Slack</Tab>
+				</Tabs>
 			</div>
+
+			{#if errorHandlerSelected === 'custom'}
+				<div class="flex flex-row mb-2">
+					<ScriptPicker
+						disabled={!can_write}
+						initialPath={errorHandlerPath}
+						kind={Script.kind.SCRIPT}
+						allowFlow={true}
+						bind:scriptPath={errorHandlerPath}
+						bind:itemKind={errorHandleritemKind}
+						allowRefresh
+					/>
+
+					{#if errorHandlerPath === undefined}
+						<Button
+							btnClasses="ml-4 mt-2"
+							color="dark"
+							size="xs"
+							href="/scripts/add?hub=hub%2F2420%2Fwindmill%2Fschedule_error_handler_template"
+							target="_blank">Create from template</Button
+						>
+					{/if}
+				</div>
+			{:else if errorHandlerSelected === 'slack'}
+				<Alert type="info" title="Slack schedule error handler"
+					>You will receive a notification on the selected slack channel.
+				</Alert>
+			{/if}
+
+			<div class="flex flex-row items-center justify-between">
+				<div class="flex flex-row items-center mt-4 font-semibold text-sm gap-2">
+					<p>Triggered when schedule failed</p>
+					<select class="!w-14" bind:value={failedExact}>
+						<option value={false}>&gt;=</option>
+						<option value={true}>==</option>
+					</select>
+					<input type="number" class="!w-14 text-center" bind:value={failedTimes} min="1" />
+					<p>time{failedTimes > 1 ? 's in a row' : ''}</p>
+				</div>
+			</div>
+			{#if errorHandlerPath}
+				<p class="font-semibold text-sm mt-4 mb-2"
+					>{errorHandlerSelected !== 'custom' ? 'Configuration' : 'Extra arguments'}</p
+				>
+				<SchemaForm
+					disabled={!can_write}
+					schema={errorHandlerSchema}
+					bind:args={errorHandlerExtraArgs}
+					shouldHideNoInputs
+					class="text-xs"
+				/>
+				{#if errorHandlerSchema && errorHandlerSchema.properties && Object.keys(errorHandlerSchema.properties).length === 0}
+					<div class="text-xs texg-gray-700">This error handler takes no extra arguments</div>
+				{/if}
+			{/if}
+
+			<h2 class="border-b pb-1 mt-8 mb-2"
+				>Recovery Handler <Tooltip light
+					><div class="text-tertiary text-sm"
+						>The following args will be passed to the recovery handler:
+						<ul class="mt-1 ml-2">
+							<li><b>path</b>: The path of the script or flow that recovered.</li>
+							<li><b>is_flow</b>: Whether the runnable is a flow.</li>
+							<li><b>schedule_path</b>: The path of the schedule.</li>
+							<li><b>error</b>: The error of the last job that errored</li>
+							<li><b>error_started_at</b>: The start datetime of the last job that errored</li>
+							<li
+								><b>success_times</b>: The number of times the schedule succeeded before calling the
+								recovery handler.</li
+							>
+							<li><b>success_result</b>: The result of the latest successful job</li>
+							<li><b>success_started_at</b>: The start datetime of the latest successful job</li>
+						</ul>
+					</div></Tooltip
+				></h2
+			>
+
+			<Tabs bind:selected={recoveryHandlerSelected} class="mt-2 mb-4">
+				<Tab value="custom">Custom</Tab>
+				<Tab value="slack">Slack</Tab>
+			</Tabs>
+
+			{#if recoveryHandlerSelected === 'custom'}
+				<div class="flex flex-row mb-2">
+					<ScriptPicker
+						disabled={!can_write}
+						initialPath={recoveryHandlerPath}
+						kind={Script.kind.SCRIPT}
+						allowFlow={true}
+						bind:scriptPath={recoveryHandlerPath}
+						bind:itemKind={recoveryHandlerItemKind}
+						allowRefresh
+					/>
+
+					{#if recoveryHandlerPath === undefined}
+						<Button
+							btnClasses="ml-4 mt-2"
+							color="dark"
+							size="xs"
+							href="/scripts/add?hub=hub%2F2421%2Fwindmill%2Fschedule_recovery_handler_template"
+							target="_blank">Create from template</Button
+						>
+					{/if}
+				</div>
+			{:else if recoveryHandlerSelected === 'slack'}
+				<Alert type="info" title="Slack schedule recovery handler"
+					>You will receive a notification on the selected slack channel.
+				</Alert>
+			{/if}
+
+			<div class="flex flex-row items-center justify-between">
+				<div class="flex flex-row items-center mt-5 font-semibold text-sm">
+					<p>Triggered when schedule recovered</p>
+					<input type="number" class="!w-14 mx-2 text-center" bind:value={recoveredTimes} min="1" />
+					<p>time{recoveredTimes > 1 ? 's in a row' : ''}</p>
+				</div>
+			</div>
+
+			{#if recoveryHandlerPath}
+				<p class="font-semibold text-sm mt-4 mb-2"
+					>{recoveryHandlerSelected === 'custom' ? 'Extra arguments' : 'Configuration'}</p
+				>
+				<SchemaForm
+					disabled={!can_write}
+					schema={recoveryHandlerSchema}
+					bind:args={recoveryHandlerExtraArgs}
+					shouldHideNoInputs
+					class="text-xs"
+				/>
+				{#if recoveryHandlerSchema && recoveryHandlerSchema.properties && Object.keys(recoveryHandlerSchema.properties).length === 0}
+					<div class="text-xs texg-gray-700">This recovery handler takes no extra arguments</div>
+				{/if}
+			{/if}
 		</div>
 	</DrawerContent>
 </Drawer>
@@ -28,7 +28,7 @@
 	export let itemKind: 'hub' | 'script' | 'flow' = allowHub ? 'hub' : 'script'
 	export let kind: Script.kind = Script.kind.SCRIPT
 	export let disabled = false
-	export let canRefresh = false
+	export let allowRefresh = false

 	let items: { value: string; label: string }[] = []
 	let drawerViewer: Drawer
@@ -129,7 +129,7 @@
 		/>
 	{/if}

-	{#if canRefresh}
+	{#if allowRefresh}
 		<Button variant="border" color="light" wrapperClasses="self-stretch" on:click={loadItems}
 			><Icon scale={0.8} data={faRotateRight} /></Button
 		>
@@ -593,7 +593,7 @@
 				bind:scriptPath={errorHandlerScriptPath}
 				initialPath={errorHandlerInitialPath}
 				on:select={editErrorHandler}
-				canRefresh
+				allowRefresh
 			/>
 			<div class="flex flex-col gap-20 items-start mt-3">
 				<div class="w-2/3">