Catch up on missed automation runs

documentation/cli/change_log.rst

@@ -11,6 +11,7 @@ since v1.0.0 | July XX, 2026
 * Add ``flexmeasures delete secret`` to remove an encrypted secret from an account or asset.
 * Add ``flexmeasures add automation``, ``flexmeasures edit automation`` and ``flexmeasures delete automation`` to manage automations (recurring tasks on an asset; for now, computing forecasts).
 * Add ``flexmeasures jobs run-automations`` to queue jobs for all automations that are due to run this minute (run this once per minute, e.g. via cron).
+* Let ``flexmeasures jobs run-automations`` catch up on minutes missed by previous invocations (e.g. due to host downtime), bounded by a new ``--max-catchup`` option (60 minutes by default); an automation due in several missed minutes still runs only once per invocation.
 
 since v0.33.0 | June 01, 2026
 =================================

documentation/cli/commands.rst

@@ -117,7 +117,7 @@ of which some are referred to in this documentation.
 ``flexmeasures jobs run-job``                     Run a single job (useful for debugging it)
 ``flexmeasures jobs inspect-job``                 Inspect a background job and print its current status, result and metadata.
 ``flexmeasures jobs stats``                       Show estimated live statistics of the queueing system.
-``flexmeasures jobs run-automations``             Queue jobs for all automations that are due to run this minute (run this once per minute, e.g. via cron).
+``flexmeasures jobs run-automations``             Queue jobs for all automations that are due to run this minute (run this once per minute, e.g. via cron), catching up on missed minutes.
 ================================================= =======================================
 
 

documentation/features/forecasting.rst

@@ -203,4 +203,10 @@ For automations to actually run, let a cron job execute the following command on
 Each due automation then queues its forecasting jobs.
 The jobs record how they were created, which is shown on the asset's status page (UI), where recent jobs are listed.
 
+The runner remembers (in Redis) the last minute it processed.
+If your cron job misses some minutes (e.g. due to host downtime or overload), the next invocation catches up:
+it also matches automations against the missed minutes, up to a maximum catch-up window (``--max-catchup``, 60 minutes by default; set it to 0 to disable catching up).
+An automation that was due in several missed minutes still runs only *once per invocation* — its timing parameters (such as the forecast start) are resolved against the current time, so queueing multiple identical jobs would be wasteful.
+Minutes missed for longer than the catch-up window are skipped.
+
 Automations defined on an asset can be viewed on the asset's *Automations* page in the UI, and listed with the API endpoint `[GET] /assets/(id)/automations <../api/v3_0.html>`_.

flexmeasures/cli/jobs.py

@@ -34,7 +34,7 @@
 from flexmeasures.data.schemas import AssetIdField, SensorIdField
 from flexmeasures.data.services.automations import (
     floor_to_minute,
-    get_due_automations,
+    get_due_automations_in_window,
     run_automation,
 )
 from flexmeasures.data.services.scheduling import handle_scheduling_exception
@@ -61,9 +61,22 @@ def fm_jobs():
     """FlexMeasures: Job queueing."""
 
 
+LAST_TICK_KEY = "automation-runner:last-tick"
+
+
 @fm_jobs.command("run-automations")
 @with_appcontext
-def run_automations():
+@click.option(
+    "--max-catchup",
+    "max_catchup",
+    type=click.IntRange(min=0),
+    default=60,
+    show_default=True,
+    help="Maximum catch-up window, in minutes. If previous invocations missed some minutes"
+    " (e.g. due to host downtime), also process those missed minutes, up to this many"
+    " minutes back. Set to 0 to disable catching up.",
+)
+def run_automations(max_catchup: int):
     """
     Queue jobs for all automations that are due to run this minute.
 
@@ -73,28 +86,69 @@ def run_automations():
     \b
         * * * * * flexmeasures jobs run-automations
 
+    The runner remembers (in Redis) the last minute it processed. If some minutes were
+    missed since then (e.g. due to host downtime or overload), it catches up by also
+    matching automations against the missed minutes, up to --max-catchup minutes back.
+    An automation that was due in several missed minutes still runs only once per
+    invocation, as its (time-related) parameters are resolved against the current time,
+    so queueing multiple identical jobs would be wasteful.
+
     A Redis-based guard prevents queueing jobs twice if the command happens to run
-    more than once within the same minute.
+    more than once within the same minute (or over overlapping catch-up windows).
     """
     now = floor_to_minute(server_now())
-    due_automations = get_due_automations(now)
+    connection = app.queues["forecasting"].connection
+
+    # Determine the window of minutes to process: from just after the last processed
+    # minute (bounded by the maximum catch-up window) through the current minute.
+    start = now
+    if max_catchup > 0:
+        last_tick = connection.get(LAST_TICK_KEY)
+        if last_tick is not None:
+            if isinstance(last_tick, bytes):
+                last_tick = last_tick.decode()
+            start = max(
+                datetime.fromisoformat(last_tick) + timedelta(minutes=1),
+                now - timedelta(minutes=max_catchup),
+            )
+            # always process the current minute, even if the last tick lies in the future
+            start = min(floor_to_minute(start), now)
+    if start < now:
+        click.secho(
+            f"Catching up on missed minutes: processing {start} through {now}.",
+            **MsgStyle.WARN,
+        )
+
+    due_automations = get_due_automations_in_window(start, now)
     if not due_automations:
         click.secho(f"No automations due at {now}.", **MsgStyle.SUCCESS)
+        connection.set(LAST_TICK_KEY, now.isoformat())
         return
 
-    connection = app.queues["forecasting"].connection
     n_run = 0
     n_failed = 0
-    for automation in due_automations:
-        # guard against running the same automation twice within the same minute
-        guard_key = f"automation-run:{automation.id}:{now.isoformat()}"
-        if not connection.set(guard_key, 1, nx=True, ex=120):
+    for automation, due_minutes in due_automations:
+        # Guard against running the same automation twice for the same minute
+        # (by concurrent invocations, or by invocations with overlapping catch-up windows).
+        guard_keys = [
+            f"automation-run:{automation.id}:{minute.isoformat()}"
+            for minute in due_minutes
+        ]
+        acquired_keys = [
+            guard_key
+            for guard_key in guard_keys
+            if connection.set(guard_key, 1, nx=True, ex=120)
+        ]
+        if not acquired_keys:
             click.secho(
-                f"Automation {automation.id} ('{automation.name}') already ran at {now}. Skipping.",
+                f"Automation {automation.id} ('{automation.name}') already ran at {due_minutes[-1]}. Skipping.",
                 **MsgStyle.WARN,
             )
             continue
         try:
+            # Even if the automation was due in several minutes within the window,
+            # we run it only once: its (time-related) parameters are resolved against
+            # the current time, so queueing multiple identical jobs would be wasteful.
             returns = run_automation(automation)
             n_jobs = returns.get("n_jobs") if returns else 0
             click.secho(
@@ -104,13 +158,16 @@ def run_automations():
             n_run += 1
         except Exception as e:
             db.session.rollback()
-            # release the guard, so a retry within the same minute can still queue jobs
-            connection.delete(guard_key)
+            # release the guards, so a retry can still queue jobs
+            connection.delete(*acquired_keys)
             click.secho(
                 f"Automation {automation.id} ('{automation.name}') failed to queue jobs: {e}",
                 **MsgStyle.ERROR,
             )
             n_failed += 1
+    # Remember the last minute we processed, so the next invocation can catch up
+    # in case it misses some minutes.
+    connection.set(LAST_TICK_KEY, now.isoformat())
     if n_failed:
         click.secho(f"{n_run} automation(s) ran, {n_failed} failed.", **MsgStyle.ERROR)
         raise click.exceptions.Exit(1)

flexmeasures/cli/tests/test_automations.py

@@ -1,11 +1,19 @@
+from __future__ import annotations
+
+from datetime import datetime, timedelta
+
 import pytest
 
 from sqlalchemy import select
 
 from flexmeasures.data.models.audit_log import AssetAuditLog
 from flexmeasures.data.models.automations import Automation
+from flexmeasures.data.services.automations import floor_to_minute
+from flexmeasures.utils.time_utils import server_now
 from flexmeasures.cli.tests.utils import to_flags
 
+LAST_TICK_KEY = "automation-runner:last-tick"
+
 
 @pytest.fixture(scope="function")
 def clean_redis(app):
@@ -14,6 +22,32 @@ def clean_redis(app):
     app.redis_connection.flushdb()
 
 
+def add_automation(app, name: str, cron: str, sensor_id: int):
+    """Create an automation on asset 1 through the CLI."""
+    from flexmeasures.cli.data_add import add_automation as add_automation_command
+
+    runner = app.test_cli_runner()
+    cli_input = {"asset": 1, "name": name, "cron": cron, "sensor": sensor_id}
+    result = runner.invoke(add_automation_command, to_flags(cli_input))
+    assert "Successfully created" in result.output, result.output
+
+
+def get_last_tick(app) -> datetime | None:
+    """Read the runner's last processed minute from Redis."""
+    value = app.redis_connection.get(LAST_TICK_KEY)
+    if value is None:
+        return None
+    if isinstance(value, bytes):
+        value = value.decode()
+    return datetime.fromisoformat(value)
+
+
+def daily_cron(minute_in_time: datetime) -> str:
+    """Cron string matching (daily) the given minute, in the FLEXMEASURES_TIMEZONE."""
+    minute_in_time = floor_to_minute(minute_in_time)
+    return f"{minute_in_time.minute} {minute_in_time.hour} * * *"
+
+
 def test_add_edit_delete_automation(app, fresh_db, setup_dummy_data):
     """Roundtrip: create an automation, edit it, then delete it, checking the audit log along the way."""
     from flexmeasures.cli.data_add import add_automation
@@ -136,3 +170,109 @@ def test_run_automations(app, fresh_db, setup_dummy_data, clean_redis):
     app.redis_connection.flushdb()
     result = runner.invoke(run_automations)
     assert "No automations due" in result.output, result.output
+
+
+def test_run_automations_catchup(app, fresh_db, setup_dummy_data, clean_redis):
+    """An automation due in a missed minute (but not this minute) runs when catching up."""
+    from flexmeasures.cli.jobs import run_automations
+
+    now = floor_to_minute(server_now())
+    # due 5 minutes ago, i.e. within the missed window, but not due this minute
+    add_automation(
+        app, "Catch me up", daily_cron(now - timedelta(minutes=5)), setup_dummy_data[0]
+    )
+    app.redis_connection.set(
+        LAST_TICK_KEY, floor_to_minute(now - timedelta(minutes=10)).isoformat()
+    )
+
+    runner = app.test_cli_runner()
+    result = runner.invoke(run_automations)
+    assert result.exit_code == 0, result.output
+    assert "Catching up" in result.output, result.output
+    assert result.output.count("queued") == 1, result.output
+    n_jobs = len(app.queues["forecasting"].jobs)
+    assert n_jobs > 0
+
+    # the last tick advanced to the current minute, so the next invocation has nothing to do
+    assert get_last_tick(app) >= now
+    result = runner.invoke(run_automations)
+    assert "No automations due" in result.output, result.output
+    assert len(app.queues["forecasting"].jobs) == n_jobs
+
+
+def test_run_automations_catchup_once_per_invocation(
+    app, fresh_db, setup_dummy_data, clean_redis
+):
+    """An automation due in every missed minute still runs only once per invocation."""
+    from flexmeasures.cli.jobs import run_automations
+
+    now = floor_to_minute(server_now())
+    add_automation(app, "Every minute", "* * * * *", setup_dummy_data[0])
+    app.redis_connection.set(
+        LAST_TICK_KEY, floor_to_minute(now - timedelta(minutes=10)).isoformat()
+    )
+
+    runner = app.test_cli_runner()
+    result = runner.invoke(run_automations)
+    assert result.exit_code == 0, result.output
+    # due in all 11 minutes of the window, but queued only once
+    assert result.output.count("queued") == 1, result.output
+
+
+def test_run_automations_max_catchup(app, fresh_db, setup_dummy_data, clean_redis):
+    """Minutes missed longer ago than --max-catchup are ignored."""
+    from flexmeasures.cli.jobs import run_automations
+
+    now = floor_to_minute(server_now())
+    add_automation(
+        app,
+        "Too long ago",
+        daily_cron(now - timedelta(minutes=20)),
+        setup_dummy_data[0],
+    )
+    add_automation(
+        app,
+        "Recently missed",
+        daily_cron(now - timedelta(minutes=5)),
+        setup_dummy_data[1],
+    )
+    app.redis_connection.set(
+        LAST_TICK_KEY, floor_to_minute(now - timedelta(minutes=30)).isoformat()
+    )
+
+    runner = app.test_cli_runner()
+    result = runner.invoke(run_automations, ["--max-catchup", "10"])
+    assert result.exit_code == 0, result.output
+    assert result.output.count("queued") == 1, result.output
+    assert "Recently missed" in result.output, result.output
+    assert "Too long ago" not in result.output, result.output
+
+    # with catch-up disabled, only the current minute is processed
+    app.redis_connection.flushdb()
+    app.redis_connection.set(
+        LAST_TICK_KEY, floor_to_minute(now - timedelta(minutes=30)).isoformat()
+    )
+    result = runner.invoke(run_automations, ["--max-catchup", "0"])
+    assert "No automations due" in result.output, result.output
+
+
+def test_run_automations_first_run(app, fresh_db, setup_dummy_data, clean_redis):
+    """Without a last tick recorded, only the current minute is processed."""
+    from flexmeasures.cli.jobs import run_automations
+
+    now = floor_to_minute(server_now())
+    # due 5 minutes ago, but there is no last tick, so there is no window to catch up on
+    add_automation(
+        app,
+        "Missed before first run",
+        daily_cron(now - timedelta(minutes=5)),
+        setup_dummy_data[0],
+    )
+    assert app.redis_connection.get(LAST_TICK_KEY) is None
+
+    runner = app.test_cli_runner()
+    result = runner.invoke(run_automations)
+    assert result.exit_code == 0, result.output
+    assert "No automations due" in result.output, result.output
+    # the first run recorded a last tick, so future invocations can catch up
+    assert get_last_tick(app) >= now

flexmeasures/data/services/automations.py

@@ -4,7 +4,7 @@
 
 from __future__ import annotations
 
-from datetime import datetime
+from datetime import datetime, timedelta
 from typing import Any
 
 from cron_descriptor import get_description, Options
@@ -44,14 +44,36 @@ def get_due_automations(now: datetime | None = None) -> list[Automation]:
     if now is None:
         now = server_now()
     now = floor_to_minute(now)
+    return [automation for automation, _ in get_due_automations_in_window(now, now)]
+
+
+def get_due_automations_in_window(
+    start: datetime, end: datetime
+) -> list[tuple[Automation, list[datetime]]]:
+    """Return active automations due at any minute from start through end (both inclusive),
+    together with the minutes at which they are due.
+
+    Each automation is listed at most once, even if its cron string matches several
+    minutes within the window (callers are expected to run it only once per invocation).
+    Cron strings are interpreted in the FLEXMEASURES_TIMEZONE.
+    """
+    start = floor_to_minute(start)
+    end = floor_to_minute(end)
     active_automations = (
         db.session.scalars(select(Automation).filter_by(active=True)).unique().all()
     )
-    return [
-        automation
-        for automation in active_automations
-        if croniter.match(automation.cronstr, now)
-    ]
+    results = []
+    for automation in active_automations:
+        due_minutes = []
+        minute = start
+        while minute <= end:
+            if croniter.match(automation.cronstr, minute):
+                due_minutes.append(minute)
+            # floor again so DST transitions are handled correctly
+            minute = floor_to_minute(minute + timedelta(minutes=1))
+        if due_minutes:
+            results.append((automation, due_minutes))
+    return results
 
 
 def get_automation_job_stats(automation: Automation) -> dict[str, int]: