fix(schedules): re-arm enabled schedules left with no queued occurrence (#10174)

* fix(schedules): re-arm enabled schedules left with no queued occurrence

* fix(schedules): lock schedule row while re-arming and report outcome

* fix(schedules): make reconcile lock cancellation-safe, re-check armed under lock

Address review feedback on the schedule reconciler:

- Use a transaction-scoped advisory lock (pg_try_advisory_xact_lock) instead of
  a session-scoped one. monitor_db runs under a 600s timeout; on cancellation a
  session lock on a pooled connection would be stranded, wedging reconciliation
  on every replica. An xact lock releases when its transaction is dropped.
- rearm_schedule re-checks for a queued occurrence under the row lock and
  returns NoOp if already armed, closing the scan→lock window that could
  double-push across a cron boundary. Add a regression test.
- Make reconcile_unarmed_schedules private (its only caller is in monitor.rs)
  and document its system-only contract.
- Log the disable only after the guarded UPDATE actually disables the schedule.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

* fix(schedules): never disable from reconciliation and cap re-arms per pass

---------

Co-authored-by: claude[bot] <41898282+claude[bot]@users.noreply.github.com>
Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com>

backend/.sqlx/query-2756fab586489be33a3031fabfdc87be642091a2f8a6973d001e59be892a984d.json

@@ -0,0 +1,26 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT s.workspace_id, s.path\n         FROM schedule s JOIN workspace w ON w.id = s.workspace_id AND NOT w.deleted\n         WHERE s.enabled IS TRUE\n             AND NOT EXISTS (\n                 SELECT 1 FROM v2_job_queue q JOIN v2_job j USING (id)\n                 WHERE j.workspace_id = s.workspace_id\n                     AND j.trigger_kind = 'schedule'\n                     AND j.trigger = s.path\n                     AND j.runnable_path = s.script_path\n                     AND j.parent_job IS NULL\n             )",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "workspace_id",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 1,
+        "name": "path",
+        "type_info": "Varchar"
+      }
+    ],
+    "parameters": {
+      "Left": []
+    },
+    "nullable": [
+      false,
+      false
+    ]
+  },
+  "hash": "2756fab586489be33a3031fabfdc87be642091a2f8a6973d001e59be892a984d"
+}

backend/src/monitor.rs

@@ -108,7 +108,11 @@ use windmill_common::{
 };
 #[cfg(feature = "parquet")]
 use windmill_object_store::reload_object_store_setting;
-use windmill_queue::{cancel_job, get_queued_job_v2, SameWorkerPayload};
+use windmill_queue::{
+    cancel_job, get_queued_job_v2,
+    schedule::{find_unarmed_schedules, rearm_schedule, RearmOutcome},
+    SameWorkerPayload,
+};
 use windmill_worker::{
     result_processor::handle_job_error, JobCompletedSender, JobIsolationLevel,
     OtelTracingProxySettings, SameWorkerSender, WorkspaceRegistryMap, BUNFIG_INSTALL_SCOPES,
@@ -3286,6 +3290,15 @@ pub async fn monitor_db(
         }
     };
 
+    // run every 30 iterations (~5min at the default LISTEN_NEW_EVENTS_INTERVAL_SEC).
+    let reconcile_unarmed_schedules_f = async {
+        if server_mode && iteration.is_some() && iteration.as_ref().unwrap().should_run(30) {
+            if let Some(db) = conn.as_sql() {
+                reconcile_unarmed_schedules(&db).await;
+            }
+        }
+    };
+
     // Poll git-sync repositories for new commits and pull them into the
     // workspace (repo → Windmill auto-pull). Runs every 2 iterations.
     let git_auto_pull_f = async {
@@ -3349,9 +3362,144 @@ pub async fn monitor_db(
         cleanup_scheduled_job_deletions_f,
         git_auto_pull_f,
         pipeline_freshness_watchdog_f,
+        reconcile_unarmed_schedules_f,
     );
 }
 
+/// Advisory lock id ensuring only one server replica reconciles schedules at a
+/// time (adjacent to GIT_AUTO_PULL_LOCK_ID).
+const SCHEDULE_RECONCILE_LOCK_ID: i64 = 737_483_922;
+
+/// Consecutive reconciliation passes an enabled schedule must be observed with no
+/// queued occurrence before it is re-armed. The next occurrence is pushed in the
+/// same transaction that completes the previous one (or, for flows, on entry to
+/// step 0), so an unarmed schedule is normally only ever a mid-flight push or a
+/// push being retried. Requiring two passes keeps the reconciler from racing
+/// those and double-pushing an occurrence.
+const SCHEDULE_RECONCILE_STRIKES: u8 = 2;
+
+/// Most schedules re-armed in one pass, so a large first-pass backlog is drained
+/// over several passes instead of enqueuing every occurrence at once.
+const SCHEDULE_RECONCILE_MAX_PER_PASS: usize = 50;
+
+lazy_static::lazy_static! {
+    /// `(workspace_id, path)` -> consecutive passes seen with no queued occurrence.
+    /// Bounded by the number of enabled schedules; entries drop as soon as a
+    /// schedule is seen armed again.
+    static ref UNARMED_SCHEDULES: Mutex<std::collections::HashMap<(String, String), u8>> =
+        Mutex::new(std::collections::HashMap::new());
+}
+
+/// Re-arm enabled schedules that have no queued occurrence.
+///
+/// Every path that completes a scheduled job is supposed to push the next
+/// occurrence atomically, but a run that dies through an abnormal path (a flow
+/// whose status update fails and is later force-completed by zombie detection,
+/// say) can skip that push and leave the schedule enabled yet dead forever. This
+/// is the backstop: without it the only recovery is a manual disable/enable.
+///
+/// Replicas each run their own passes (staggered by `rd_shift`) and each keep
+/// their own strike tally, so the scan cost is per-replica. That is deliberate:
+/// scanning inside the advisory lock is what makes a double-push impossible —
+/// whoever holds it re-reads the unarmed set, so a schedule another replica just
+/// re-armed is seen armed and its tally dropped, rather than pushed twice.
+///
+/// Not an authorization boundary: it re-arms schedules across every workspace, so
+/// this is a system caller (the monitor loop) only.
+async fn reconcile_unarmed_schedules(db: &Pool<Postgres>) {
+    // Transaction-scoped advisory lock, not session-scoped: monitor_db runs under a
+    // 600s timeout, and if it fires the whole future is dropped mid-pass. A session
+    // lock taken on a pooled connection would then ride that connection back into the
+    // pool still held, wedging reconciliation on every replica until the process
+    // restarts. An xact lock is released when its transaction ends — including the
+    // rollback a dropped `Transaction` performs — so cancellation can't strand it.
+    // The tx is held open only to own the lock; the scan and re-arm run on separate
+    // pool connections.
+    let mut lock_tx = match db.begin().await {
+        Ok(tx) => tx,
+        Err(e) => {
+            tracing::error!("schedule reconcile: failed to begin lock tx: {e:#}");
+            return;
+        }
+    };
+    let locked: bool = match sqlx::query_scalar("SELECT pg_try_advisory_xact_lock($1)")
+        .bind(SCHEDULE_RECONCILE_LOCK_ID)
+        .fetch_one(&mut *lock_tx)
+        .await
+    {
+        Ok(v) => v,
+        Err(e) => {
+            tracing::error!("schedule reconcile: advisory lock failed: {e:#}");
+            return;
+        }
+    };
+    if !locked {
+        // Another replica is already reconciling this tick.
+        return;
+    }
+
+    if let Err(e) = reconcile_unarmed_schedules_inner(db).await {
+        tracing::error!("schedule reconcile: {e:#}");
+    }
+
+    // Ends the transaction and releases the xact lock; a plain drop would too.
+    if let Err(e) = lock_tx.rollback().await {
+        tracing::error!("schedule reconcile: releasing lock failed: {e:#}");
+    }
+}
+
+/// Record this pass's unarmed schedules against `seen` and return those that have
+/// now struck out. An armed observation drops the schedule's tally entirely, so
+/// the strikes a re-arm rests on are always consecutive.
+fn strike_unarmed(
+    seen: &mut std::collections::HashMap<(String, String), u8>,
+    current: std::collections::HashSet<(String, String)>,
+) -> Vec<(String, String)> {
+    seen.retain(|k, _| current.contains(k));
+    current
+        .into_iter()
+        .filter(|k| {
+            let strikes = seen.entry(k.clone()).or_insert(0);
+            *strikes = strikes.saturating_add(1);
+            *strikes >= SCHEDULE_RECONCILE_STRIKES
+        })
+        .collect()
+}
+
+async fn reconcile_unarmed_schedules_inner(db: &Pool<Postgres>) -> error::Result<()> {
+    let current = find_unarmed_schedules(db).await?.into_iter().collect();
+    let mut to_rearm = strike_unarmed(&mut UNARMED_SCHEDULES.lock().unwrap(), current);
+
+    // The first pass on an instance that has never been swept can find a large
+    // backlog; re-arming it all at once would enqueue that whole backlog in one
+    // go. The overflow keeps its tally and is picked up next pass.
+    if to_rearm.len() > SCHEDULE_RECONCILE_MAX_PER_PASS {
+        tracing::warn!(
+            "schedule reconcile: {} schedules have no queued occurrence, re-arming {} this pass and the rest on later passes",
+            to_rearm.len(),
+            SCHEDULE_RECONCILE_MAX_PER_PASS
+        );
+        to_rearm.truncate(SCHEDULE_RECONCILE_MAX_PER_PASS);
+    }
+
+    for (w_id, path) in to_rearm {
+        match rearm_schedule(db, &w_id, &path).await {
+            Ok(outcome) => {
+                if outcome == RearmOutcome::Rearmed {
+                    tracing::warn!(
+                        "schedule reconcile: re-armed enabled schedule {path} in {w_id}, which had no queued occurrence"
+                    );
+                }
+                UNARMED_SCHEDULES.lock().unwrap().remove(&(w_id, path));
+            }
+            Err(e) => tracing::error!(
+                "schedule reconcile: could not re-arm schedule {path} in {w_id}: {e:#}"
+            ),
+        }
+    }
+    Ok(())
+}
+
 /// Advisory lock id ensuring only one server replica runs the git auto-pull
 /// poll at a time (adjacent to RESTART_LOCK_ID used for restart coordination).
 #[cfg(feature = "private")]
@@ -5390,3 +5538,46 @@ mod retention_overrides_tests {
         assert!(parse_retention_overrides(over_cap).is_err());
     }
 }
+
+#[cfg(test)]
+mod strike_unarmed_tests {
+    use super::strike_unarmed;
+    use std::collections::{HashMap, HashSet};
+
+    fn key(path: &str) -> (String, String) {
+        ("ws".to_string(), path.to_string())
+    }
+
+    fn set(paths: &[&str]) -> HashSet<(String, String)> {
+        paths.iter().map(|p| key(p)).collect()
+    }
+
+    /// The strike threshold is the only thing keeping the reconciler from racing
+    /// an in-flight push: `push_scheduled_job`'s own `already_exists` guard keys
+    /// on the same columns as the scan, so it is false by construction whenever a
+    /// schedule is found unarmed.
+    #[test]
+    fn rearms_only_after_consecutive_unarmed_passes() {
+        let mut seen = HashMap::new();
+        assert!(strike_unarmed(&mut seen, set(&["a"])).is_empty());
+        assert_eq!(strike_unarmed(&mut seen, set(&["a"])), vec![key("a")]);
+    }
+
+    #[test]
+    fn armed_observation_resets_the_tally() {
+        let mut seen = HashMap::new();
+        assert!(strike_unarmed(&mut seen, set(&["a"])).is_empty());
+        // `a` is armed again on this pass, so its strike must not carry over.
+        assert!(strike_unarmed(&mut seen, set(&[])).is_empty());
+        assert!(strike_unarmed(&mut seen, set(&["a"])).is_empty());
+        assert_eq!(strike_unarmed(&mut seen, set(&["a"])), vec![key("a")]);
+    }
+
+    #[test]
+    fn tallies_are_per_schedule() {
+        let mut seen = HashMap::new();
+        assert!(strike_unarmed(&mut seen, set(&["a"])).is_empty());
+        assert_eq!(strike_unarmed(&mut seen, set(&["a", "b"])), vec![key("a")]);
+        assert_eq!(strike_unarmed(&mut seen, set(&["b"])), vec![key("b")]);
+    }
+}

backend/windmill-queue/src/schedule.rs

@@ -597,6 +597,121 @@ pub async fn push_scheduled_job<'c>(
     Ok(tx) // TODO: Bubble up pushed UUID from here
 }
 
+/// Enabled schedules with no occurrence in the queue, as `(workspace_id, path)`.
+///
+/// Every path that completes a scheduled job pushes the next occurrence in the
+/// same transaction (for flows, on entry to step 0), so an enabled schedule
+/// always has a queued occurrence — a run in progress is itself one. A run that
+/// dies through an abnormal path can skip that push though, leaving the schedule
+/// enabled yet dead until it is manually disabled and re-enabled. This is how the
+/// monitor spots that state; see `rearm_schedule` for the recovery.
+///
+/// Not an authorization boundary: it reports schedules across every workspace, so
+/// this is for system callers (the monitor's reconciliation pass) only and its
+/// result must never be returned to a user unfiltered.
+pub async fn find_unarmed_schedules(db: &DB) -> Result<Vec<(String, String)>> {
+    let rows = sqlx::query!(
+        // Query plan: the anti-join builds from `v2_job_queue` (only pending and
+        // running jobs) rather than probing `v2_job` once per schedule.
+        "SELECT s.workspace_id, s.path
+         FROM schedule s JOIN workspace w ON w.id = s.workspace_id AND NOT w.deleted
+         WHERE s.enabled IS TRUE
+             AND NOT EXISTS (
+                 SELECT 1 FROM v2_job_queue q JOIN v2_job j USING (id)
+                 WHERE j.workspace_id = s.workspace_id
+                     AND j.trigger_kind = 'schedule'
+                     AND j.trigger = s.path
+                     AND j.runnable_path = s.script_path
+                     AND j.parent_job IS NULL
+             )"
+    )
+    .fetch_all(db)
+    .await?;
+    Ok(rows.into_iter().map(|r| (r.workspace_id, r.path)).collect())
+}
+
+#[derive(Debug, PartialEq, Eq)]
+pub enum RearmOutcome {
+    /// The next occurrence was pushed.
+    Rearmed,
+    /// Nothing to do: the schedule was deleted or disabled since it was found.
+    NoOp,
+}
+
+/// Push the next occurrence of a schedule that has none queued.
+///
+/// Only ever starts a schedule, never stops one: re-arming something that did not
+/// need it costs one extra run, whereas wrongly disabling one is the silent
+/// permanent stoppage this whole mechanism exists to prevent. So an occurrence
+/// that cannot be pushed is logged and left alone — the schedule is already not
+/// running, and `try_schedule_next_job` still disables on the completion path,
+/// where the population is limited to actively-cycling schedules. Keep it that
+/// way: this sweeps *every* enabled schedule, including ones broken long before
+/// this code existed and never swept before.
+///
+/// Not an authorization boundary: it pushes under the schedule's own
+/// `permissioned_as` identity for any `(w_id, path)`, so this is for system
+/// callers (the monitor's reconciliation pass) only. A caller acting for a user
+/// MUST already have enforced their permissions on `w_id` and `path`.
+pub async fn rearm_schedule(db: &DB, w_id: &str, path: &str) -> Result<RearmOutcome> {
+    let mut tx = db.begin().await?;
+    // Lock the row for the whole push: an edit or a disable committing between the
+    // read and the push would otherwise leave a queued occurrence for a schedule
+    // that is disabled, or one built from superseded settings.
+    let schedule = sqlx::query_as::<_, Schedule>(
+        "SELECT workspace_id, path, edited_by, edited_at, schedule, timezone, enabled, script_path, is_flow, args, extra_perms, email, permissioned_as, error, on_failure, on_failure_times, on_failure_exact, on_failure_extra_args, on_recovery, on_recovery_times, on_recovery_extra_args, on_success, on_success_extra_args, ws_error_handler_muted, retry, no_flow_overlap, summary, description, tag, paused_until, cron_version, dynamic_skip, labels FROM schedule WHERE path = $1 AND workspace_id = $2 FOR UPDATE",
+    )
+    .bind(path)
+    .bind(w_id)
+    .fetch_optional(&mut *tx)
+    .await?;
+    let Some(schedule) = schedule else {
+        return Ok(RearmOutcome::NoOp);
+    };
+    if !schedule.enabled {
+        return Ok(RearmOutcome::NoOp);
+    }
+    // Re-check for a queued occurrence now that the row is locked: a normal
+    // completion, an edit, or a re-enable could have pushed one between the unarmed
+    // scan and this lock. push_scheduled_job only dedups the exact computed
+    // scheduled_for, so re-arming a schedule that has since become armed and crossed a
+    // cron boundary would queue a second root occurrence. Mirrors the anti-join in
+    // find_unarmed_schedules.
+    let already_armed: bool = sqlx::query_scalar(
+        "SELECT EXISTS (
+             SELECT 1 FROM v2_job_queue q JOIN v2_job j USING (id)
+             WHERE j.workspace_id = $1
+                 AND j.trigger_kind = 'schedule'
+                 AND j.trigger = $2
+                 AND j.runnable_path = $3
+                 AND j.parent_job IS NULL
+         )",
+    )
+    .bind(w_id)
+    .bind(path)
+    .bind(&schedule.script_path)
+    .fetch_one(&mut *tx)
+    .await?;
+    if already_armed {
+        return Ok(RearmOutcome::NoOp);
+    }
+    match push_scheduled_job(db, tx, &schedule, None, None).await {
+        Ok(tx) => {
+            tx.commit().await?;
+            Ok(RearmOutcome::Rearmed)
+        }
+        // An occurrence that can never be pushed (runnable gone, quota blown) is
+        // reported, not acted on — see the note above on why this never disables.
+        Err(err @ (error::Error::NotFound(_) | error::Error::QuotaExceeded(_))) => {
+            tracing::error!(
+                "Could not re-arm schedule {path} in {w_id}: {err}. Leaving it enabled; it will not run until the cause is fixed."
+            );
+            Ok(RearmOutcome::NoOp)
+        }
+        Err(err) => Err(err),
+    }
+}
+
 pub async fn get_schedule_opt<'c>(
     e: impl PgExecutor<'c>,
     w_id: &str,

backend/windmill-queue/tests/schedule_push.rs

@@ -10,7 +10,9 @@ mod schedule_push {
     use windmill_common::scripts::ScriptHash;
     use windmill_common::users::username_to_permissioned_as;
     use windmill_queue::jobs::{try_schedule_next_job, MiniCompletedJob};
-    use windmill_queue::schedule::push_scheduled_job;
+    use windmill_queue::schedule::{
+        find_unarmed_schedules, push_scheduled_job, rearm_schedule, RearmOutcome,
+    };
 
     fn make_schedule(overrides: impl FnOnce(&mut Schedule)) -> Schedule {
         let mut s = Schedule {
@@ -1762,4 +1764,121 @@ mod schedule_push {
         assert!(!row_exists, "managed schedule row must be deleted");
         Ok(())
     }
+
+    // -----------------------------------------------------------------------
+    // find_unarmed_schedules / rearm_schedule: recovery for a schedule left
+    // enabled with no queued occurrence (a run that died on an abnormal path
+    // skipped its next-occurrence push). Without this the chain stays dead
+    // until the schedule is manually disabled and re-enabled.
+    // -----------------------------------------------------------------------
+
+    async fn insert_schedule(db: &Pool<Postgres>, path: &str, script_path: &str, enabled: bool) {
+        sqlx::query(
+            "INSERT INTO schedule (workspace_id, path, edited_by, edited_at, schedule, timezone, enabled, script_path, is_flow, email, extra_perms, ws_error_handler_muted, no_flow_overlap, permissioned_as)
+             VALUES ('test-workspace', $1, 'test-user', now(), '0 0 */5 * * *', 'UTC', $3, $2, false, 'test@windmill.dev', '{}', false, true, 'u/test-user')",
+        )
+        .bind(path)
+        .bind(script_path)
+        .bind(enabled)
+        .execute(db)
+        .await
+        .unwrap();
+    }
+
+    #[sqlx::test(migrations = "../migrations", fixtures("base", "schedule_push"))]
+    async fn test_find_unarmed_schedules(db: Pool<Postgres>) -> anyhow::Result<()> {
+        insert_schedule(&db, "f/system/test_schedule", "f/system/test_script", true).await;
+        insert_schedule(&db, "f/system/disabled", "f/system/test_script", false).await;
+
+        // No occurrence queued yet: the enabled schedule is unarmed, the disabled one is ignored.
+        assert_eq!(
+            find_unarmed_schedules(&db).await?,
+            vec![(
+                "test-workspace".to_string(),
+                "f/system/test_schedule".to_string()
+            )]
+        );
+
+        // Once an occurrence is queued it is armed and must not be reported —
+        // re-arming it would double-push the occurrence.
+        let tx = db.begin().await?;
+        let tx = push_scheduled_job(&db, tx, &make_schedule(|_| {}), None, None).await?;
+        tx.commit().await?;
+        assert_eq!(count_queued_jobs(&db).await, 1);
+        assert!(find_unarmed_schedules(&db).await?.is_empty());
+        Ok(())
+    }
+
+    #[sqlx::test(migrations = "../migrations", fixtures("base", "schedule_push"))]
+    async fn test_rearm_schedule_pushes_next_occurrence(db: Pool<Postgres>) -> anyhow::Result<()> {
+        insert_schedule(&db, "f/system/test_schedule", "f/system/test_script", true).await;
+
+        assert_eq!(
+            rearm_schedule(&db, "test-workspace", "f/system/test_schedule").await?,
+            RearmOutcome::Rearmed
+        );
+
+        assert_eq!(count_queued_jobs(&db).await, 1);
+        assert!(find_unarmed_schedules(&db).await?.is_empty());
+        Ok(())
+    }
+
+    #[sqlx::test(migrations = "../migrations", fixtures("base", "schedule_push"))]
+    async fn test_rearm_schedule_skips_disabled(db: Pool<Postgres>) -> anyhow::Result<()> {
+        // A disable that lands between the scan and the re-arm must win: pushing an
+        // occurrence for a disabled schedule would resurrect a schedule the user
+        // just turned off.
+        insert_schedule(&db, "f/system/test_schedule", "f/system/test_script", false).await;
+
+        assert_eq!(
+            rearm_schedule(&db, "test-workspace", "f/system/test_schedule").await?,
+            RearmOutcome::NoOp
+        );
+        assert_eq!(count_queued_jobs(&db).await, 0);
+        Ok(())
+    }
+
+    #[sqlx::test(migrations = "../migrations", fixtures("base", "schedule_push"))]
+    async fn test_rearm_schedule_never_disables(db: Pool<Postgres>) -> anyhow::Result<()> {
+        insert_schedule(&db, "f/system/bad_schedule", "f/system/nonexistent", true).await;
+
+        // Reconciliation only ever starts a schedule. An unpushable occurrence is
+        // reported and left alone: this sweeps every enabled schedule in the
+        // instance, so disabling here would turn a wrong invariant into the exact
+        // silent stoppage the reconciler exists to undo.
+        assert_eq!(
+            rearm_schedule(&db, "test-workspace", "f/system/bad_schedule").await?,
+            RearmOutcome::NoOp
+        );
+
+        assert_eq!(count_queued_jobs(&db).await, 0);
+        let (enabled, error): (bool, Option<String>) = sqlx::query_as(
+            "SELECT enabled, error FROM schedule WHERE workspace_id = 'test-workspace' AND path = 'f/system/bad_schedule'",
+        )
+        .fetch_one(&db)
+        .await?;
+        assert!(enabled, "reconciliation must never disable a schedule");
+        assert!(error.is_none());
+        Ok(())
+    }
+
+    #[sqlx::test(migrations = "../migrations", fixtures("base", "schedule_push"))]
+    async fn test_rearm_schedule_skips_already_armed(db: Pool<Postgres>) -> anyhow::Result<()> {
+        // An occurrence can be queued (a normal completion, an edit, a re-enable)
+        // between the unarmed scan and rearm_schedule acquiring the row lock. Re-arming
+        // then would double-push, since push_scheduled_job only dedups the exact
+        // computed scheduled_for.
+        insert_schedule(&db, "f/system/test_schedule", "f/system/test_script", true).await;
+        let tx = db.begin().await?;
+        let tx = push_scheduled_job(&db, tx, &make_schedule(|_| {}), None, None).await?;
+        tx.commit().await?;
+        assert_eq!(count_queued_jobs(&db).await, 1);
+
+        assert_eq!(
+            rearm_schedule(&db, "test-workspace", "f/system/test_schedule").await?,
+            RearmOutcome::NoOp
+        );
+        assert_eq!(count_queued_jobs(&db).await, 1);
+        Ok(())
+    }
 }