diff --git a/documentation/api/change_log.rst b/documentation/api/change_log.rst index 1e34a3079a..d6c901dbcd 100644 --- a/documentation/api/change_log.rst +++ b/documentation/api/change_log.rst @@ -26,6 +26,7 @@ v3.0-33 | September 1, 2026 - Added ``POST /api/v3_0/assets//reports/trigger`` to queue a one-off report as a background job. It returns ``202 Accepted`` with the canonical ``job`` and ``job-url`` fields, and shares the trigger rate limit with forecast and schedule endpoints. - Added ``GET /api/v3_0/assets//automations`` and ``GET /api/v3_0/assets//automations/`` for listing and inspecting forecast automations, including the sensors an automation reads from and writes to. Each automation shows the IANA ``timezone`` in which its cron expression is interpreted, and a ``cursor``: the offset-aware UTC time of the most recent run it committed to. The cursor advances just before queueing, so it does not indicate that queueing or the forecast itself succeeded. Asset job entries now include ``created_via`` provenance; automation identity is included only when the caller may read that automation. - Added ``GET /api/v3_0/sources/`` to show the full record of one data source, including the attributes in which data generators store their configuration. +- Automation responses now also include ``schedule_revision``, which counts the execution-affecting edits made to the automation's schedule, and automation detail responses gained a ``run_stats`` object. It summarizes the durable runs of that automation and describes the most recent ones: their scheduled time, dispatch state (``pending``, ``claimed``, ``partially_queued``, ``queued`` or ``failed``), execution state (``pending``, ``running``, ``succeeded``, ``failed`` or ``canceled``), attempt count, intended and queued job counts, timestamps, last error, latest attempt, and the individual jobs they created. Both additions are backward compatible: no existing field changed. - ``GET /api/v3_0/sensors//stats`` now reports an ``All sources`` entry summarising every data source, whenever more than one recorded. Its mean divides by the values that were summed, not by ``Number of values``, which also counts rows holding NaN. v3.0-32 | August 11, 2026 diff --git a/documentation/changelog.rst b/documentation/changelog.rst index 95ca999b9e..747dbc9bc5 100644 --- a/documentation/changelog.rst +++ b/documentation/changelog.rst @@ -27,6 +27,7 @@ New features * A forecaster can now be told which data sources hold the truth about the sensor it forecasts, the way its regressors already could, so that a sensor several sources report on is trained on the ones you trust [see `PR #2542 `_] * Try out a forecast without recording it, using ``flexmeasures add forecasts --dry-run``, which computes the forecast in full and reports the sensor, data source, number of beliefs and event range it would have saved [see `PR #2483 `_] * Run one-off reports as background jobs from the CLI or the asset API, with sensor-level authorization and a dedicated reporting worker queue [see `PR #2298 `_] + * The asset's status page now splits its sensor data and its jobs over two tabs, of which only the opened one loads its data, and it opens the tab you last looked at [see `PR #2470 `_] * Both tabs of an asset's status page now name the asset each row belongs to, and the jobs tab also lists the jobs of the asset's sub-assets, so a site asset shows what happened anywhere below it, which you can switch off per session [see `PR #2500 `_] * Changing the selected time range on an asset or sensor chart now only loads the data that is actually new, instead of reloading the whole range, which makes stepping through or extending a long period much faster; reloading the page, or leaving it open for five minutes, still fetches everything afresh [see `PR #2433 `_] @@ -74,6 +75,7 @@ Automations arrived over several pull requests. This is what each of them contri * Reports as well as forecasts and schedules: a report automation stores report parameters, with its reporter and the reporter's configuration on a data source, and reports on a period resolved afresh on each run, either from ``start-offset`` and ``end-offset`` applied to the run time in the automation's timezone, or since the last successful report ended, while a fixed ``start`` or ``end`` is refused; a report job records only on the sensors the automation was checked against [see `PR #2297 `_] * Every automation times its runs the same way: a fixed ``start``, ``end`` or ``prior`` in its parameters is refused, as every run would share that moment, and two of ``start-offset``, ``end-offset`` and ``duration`` describe the period each run covers instead, with the offsets applied to the time the run was due on the automation's own clock, so that, for instance, a schedule automation can plan the whole of the next day [see `PR #2551 `_] * Look up automations from the command line with ``flexmeasures show automations``, which lists them all (inactive ones included) with the IDs that the edit, delete and run commands expect, and, with ``--id``, shows a single automation's recurrence, cursor, parameters and the sensors it reads from and writes to [see `PR #2533 `_] +* Automations now keep a durable record of every scheduled run, so a forecast run which failed before queueing any work is simply picked up again, while one which failed halfway only queues the jobs it still owes; an automation's details show, per run, what it queued, how many attempts that took, and, for its forecast jobs, how they ended [see `PR #2457 `_] v1.0.1 | September 9, 2026 diff --git a/documentation/cli/change_log.rst b/documentation/cli/change_log.rst index c6ead6fb55..2a8a3983a6 100644 --- a/documentation/cli/change_log.rst +++ b/documentation/cli/change_log.rst @@ -14,9 +14,11 @@ since v1.1.0 | September XX, 2026 * Add ``flexmeasures add automation``, ``flexmeasures edit automation`` and ``flexmeasures delete automation`` to manage automations (recurring tasks on an asset, with ``--type forecasting``, ``--type scheduling`` or ``--type reporting`` saying which task to automate). Each automation carries its own IANA timezone (``--timezone``), in which its cron expression is interpreted. * ``flexmeasures add automation --type scheduling`` refuses a flex config field which fixes a moment in time, such as ``soc-at-start`` or a ``soc-targets`` entry with a ``datetime``, naming the field: a recurring schedule automation computes a fresh schedule on every run, so such a value would be stale on the next one. * ``flexmeasures add automation`` refuses a fixed ``start``, ``end`` or ``prior`` in the parameters of any automation, as every run would share that moment. The parameters take ``start-offset`` and ``end-offset`` instead, or either one with a ``duration``, applied to the time each run was due, on the automation's own clock, and so do the new ``--start-offset`` and ``--end-offset`` options, next to ``--duration``. -* Add ``flexmeasures jobs run-automations`` to queue jobs for all automations that are due to run this minute from standard five-field cron expressions. Run this command once per minute. It makes at most one queueing attempt per automation per minute, including when an attempt fails after partially queueing jobs. Runs missed while the runner was down are caught up once, with several missed forecast runs coalesced into the latest useful forecast, and a run at a skipped or repeated daylight-saving-time hour happens exactly once. +* Add ``flexmeasures jobs run-automations`` to queue jobs for all automations that are due to run this minute from standard five-field cron expressions. Run this command once per minute. Each scheduled run is claimed durably, so its jobs are queued exactly once even when several runners overlap. Runs missed while the runner was down are caught up once, with several missed forecast runs coalesced into the latest useful forecast, and a run at a skipped or repeated daylight-saving-time hour happens exactly once. * Add ``flexmeasures jobs run-automation --automation `` to queue the jobs for a single run of one automation, now, on top of its recurring runs. This leaves the automation's cursor alone, so its next recurring run still happens as scheduled, and inactive automations can be run this way, too. * ``flexmeasures delete sensor`` now warns which automations read from or write to a sensor before it is deleted, as an automation refers to its sensors by ID and would fail on its next run. +* ``flexmeasures jobs run-automations`` now records a durable run for each scheduled run it claims, and retries the forecast runs whose queueing did not finish. A run which failed before queueing anything is dispatched again in full, and one which queued only part of its jobs resumes from its stored plan, reusing the job IDs it already queued. Each attempt is recorded with its owner, outcome and error, and the command reports the run and attempt it is working on. A run is only picked up by another runner once the claim lease of the runner holding it has expired, which is how a runner that died mid-queueing hands its work over. +* ``flexmeasures edit automation`` now counts up the automation's schedule revision whenever it rebases the cursor (on a changed cron string or timezone, or on reactivation), which keeps the durable runs of the old and the new schedule apart, even at the same scheduled UTC time. since v1.0.1 | September 9, 2026 ================================= diff --git a/documentation/features/automations.rst b/documentation/features/automations.rst index de6d73c718..152327f975 100644 --- a/documentation/features/automations.rst +++ b/documentation/features/automations.rst @@ -185,11 +185,37 @@ Each due automation then queues its jobs. If the runner misses runs, because it was down or overloaded, it catches up when it resumes: it queues only the latest missed run of each automation, rather than replaying stale ones. Timing parameters that default to the run time are resolved when that catch-up run is queued, so it produces a current forecast, schedule or report. -Each scheduled run receives at most one automatic queueing attempt. -If the process crashes, or queueing fails after creating some jobs, that run is not retried automatically, because a retry could duplicate partial work. +Each scheduled run a runner picks up is recorded durably, so a queueing attempt which fails can be retried without duplicating the jobs it already created. +See :ref:`automation_runs`. The jobs record how they were created, which is shown on the asset's status page (UI), where recent jobs are listed. +.. _automation_runs: + +Runs and retries +---------------- + +Every scheduled run a runner picks up gets a record in the database, which outlives the jobs it creates (jobs in Redis expire). +The runner claims the run before doing any work, and the database allows only one record per automation, scheduled time and schedule revision, so two runners started in the same minute cannot both execute it. +A claim comes with a lease: while one runner holds a live lease on a run, no other runner touches it, and once that lease expires the run is up for grabs again, which is how a runner that died mid-queueing hands its work over. + +Before queueing anything, the runner writes down the plan for the run: the parameters it will use and the individual jobs it intends to create, each with its own logical name and a job ID derived from the run. +This is what makes a retry safe. +A run which failed before queueing anything is dispatched again in full. +A run which queued only some of its jobs resumes from the same plan, recognizes the jobs already in Redis by their IDs, and queues only the ones still missing, so a retry never duplicates work, and never silently drops it either. +Because the plan is stored, a retry hours later still uses the parameters the run was planned with, even if the automation has been edited since. +Timings the automation left to the run time are not part of those parameters, so they are resolved afresh on each attempt: a resumed run's jobs can therefore cover a later window than the ones its first attempt queued. + +Retrying a failed dispatch this way is what a forecast run does. +A schedule run is recorded, claimed and reported in just the same way, but is left where it failed rather than dispatched again, because its jobs get a fresh ID on every dispatch, so a retry could not tell an already queued schedule from a missing one. + +A run tracks two things separately: how far its *dispatch* got (``pending``, ``claimed``, ``partially_queued``, ``queued`` or ``failed``), and how its *execution* by the workers ended (``pending``, ``running``, ``succeeded``, ``failed`` or ``canceled``). +Each attempt to dispatch a run is recorded too, with the runner which made it, what it queued, and why it failed if it did. +This is what an operator needs to tell a run which failed before queueing anything, one which queued half its work, and one which queued everything but then failed while computing, apart from each other. + +Editing an automation's cron string or timezone, or reactivating it, counts up its schedule revision. +Runs of the old and the new schedule therefore stay distinct, even when they fall on the same scheduled UTC time. + Running one automation on demand -------------------------------- @@ -213,7 +239,7 @@ Viewing automations 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#get--api-v3_0-assets-id-automations>`_. The page shows the next scheduled run for each automation (excluding any pending catch-up run). -An automation's details show the sensors it reads from and writes to, linking to each sensor's page. +An automation's details show the sensors it reads from and writes to, linking to each sensor's page, and summarize its recent runs and their outcomes. Conversely, a sensor's page lists the automations that write data to it. .. _automation_cursor: @@ -230,7 +256,9 @@ Runs at or before the cursor are never queued again. Before queueing any jobs, the runner advances the cursor to the run it is about to queue, and saves it. The cursor therefore records that a run was claimed, not that queueing or the task itself succeeded. -Keeping a single moving timestamp, rather than a record per run, is what makes the behaviour above fall out: a runner that has been down catches up by moving the cursor straight to the latest due run, and two runners started in the same minute cannot queue the same run twice, because the cursor is advanced with a conditional update that only one of them can win. +Keeping a single moving timestamp is what makes the catch-up behaviour above fall out: a runner that has been down catches up by moving the cursor straight to the latest due run, rather than replaying every run it missed. +The cursor also decides who may claim a newly due run, because it is advanced with a conditional update which only one of two runners started in the same minute can win. +What happened to a run once it is claimed is kept in its own record instead (see :ref:`automation_runs`), which is why the cursor alone says nothing about whether queueing or the task succeeded. A new automation starts from its creation minute and does not replay runs from before it existed. Changing its cron expression or timezone, or reactivating it, restarts from the time of that change. diff --git a/flexmeasures/api/v3_0/assets.py b/flexmeasures/api/v3_0/assets.py index b0dfac91f7..14ebf736e2 100644 --- a/flexmeasures/api/v3_0/assets.py +++ b/flexmeasures/api/v3_0/assets.py @@ -59,6 +59,7 @@ delete_automation as remove_automation, describe_cronstr, get_automation_job_stats, + get_automation_run_stats, resolve_automation_sensors, run_automation, update_automation, @@ -1503,6 +1504,7 @@ def get_automations(self, id: int, asset: GenericAsset): cursor: "2026-07-11T06:00:00+02:00" next-run: "2026-07-12T06:00:00+02:00" recurrence-description: "At 06:00" + schedule-revision: 1 active: true 401: description: UNAUTHORIZED @@ -1544,8 +1546,8 @@ def get_automation(self, id: int, automation_id: int, asset: GenericAsset): the automation's parameters (forecast parameters or a schedule trigger message), the data source it records under, as its `source` (null for schedule automations), the sensors it reads from and writes to, - and counts of recently created jobs, per job status. - Note that jobs in Redis have a limited TTL, so not all past jobs will be counted. + durable run status, and counts of recently created jobs, per job status. + Note that jobs in Redis have a limited TTL, so not all past jobs will be counted, while durable run status records queueing attempts and outcomes even after those jobs expire. The cursor is the time of the most recent run the automation committed to, in the automation's own timezone; runs at or before it are never queued again. It advances just before queueing, so it does not indicate that queueing or the forecast itself succeeded. security: @@ -1582,6 +1584,7 @@ def get_automation(self, id: int, automation_id: int, asset: GenericAsset): cursor: "2026-07-11T06:00:00+02:00" next-run: "2026-07-12T06:00:00+02:00" recurrence-description: "At 06:00" + schedule-revision: 1 active: true parameters: sensor: 2092 @@ -1599,6 +1602,23 @@ def get_automation(self, id: int, automation_id: int, asset: GenericAsset): job-stats: finished: 3 failed: 1 + run-stats: + total: 1 + dispatch: + queued: 1 + execution: + succeeded: 1 + latest-run: + id: 12 + scheduled-at: "2026-07-11T04:00:00+00:00" + schedule-revision: 1 + dispatch-state: queued + execution-state: succeeded + attempt-count: 1 + intended-job-count: 2 + queued-job-count: 2 + last-error: null + recent-runs: [] redis-connection-err: null 401: description: UNAUTHORIZED @@ -1652,6 +1672,7 @@ def get_automation(self, id: int, automation_id: int, asset: GenericAsset): except NoRedisConfigured as e: automation_data["job-stats"] = {} redis_connection_err = e.args[0] + automation_data["run-stats"] = get_automation_run_stats(automation) automation_data["redis-connection-err"] = redis_connection_err return automation_data, 200 diff --git a/flexmeasures/api/v3_0/tests/test_automations_api.py b/flexmeasures/api/v3_0/tests/test_automations_api.py index 695159334e..8e25caea80 100644 --- a/flexmeasures/api/v3_0/tests/test_automations_api.py +++ b/flexmeasures/api/v3_0/tests/test_automations_api.py @@ -8,7 +8,12 @@ from flask import url_for from sqlalchemy import select -from flexmeasures.data.models.automations import Automation +from flexmeasures.data.models.automations import ( + Automation, + AutomationRun, + AutomationRunAttempt, + AutomationRunJob, +) from flexmeasures.data.models.data_sources import DataSource from flexmeasures.data.models.time_series import Sensor @@ -47,6 +52,129 @@ def add_automations(fresh_db, add_battery_assets_fresh_db): ] fresh_db.session.add_all(automations) fresh_db.session.flush() + run = AutomationRun( + automation=automations[0], + scheduled_at=datetime(2026, 7, 11, 4, 0, tzinfo=timezone.utc), + schedule_revision=automations[0].schedule_revision, + automation_type="forecasting", + generator_id=generator.id, + dispatch_state="partially_queued", + execution_state="pending", + attempt_count=2, + first_enqueued_at=datetime(2026, 7, 11, 4, 1, tzinfo=timezone.utc), + parameters=dict(automations[0].parameters), + plan={"cronstr": automations[0].cronstr, "timezone": automations[0].timezone}, + last_error_type="ConnectionError", + last_error_message="lost Redis connection", + ) + fresh_db.session.add(run) + fresh_db.session.flush() + fresh_db.session.add_all( + [ + AutomationRunJob( + run=run, + logical_job_key="cycle-001", + rq_job_id=f"automation-run-{run.id}-cycle-001", + queue="forecasting", + kind="forecast-cycle", + status="queued", + depends_on=[], + payload={}, + ), + AutomationRunJob( + run=run, + logical_job_key="wrap-up", + rq_job_id=f"automation-run-{run.id}-wrap-up", + queue="forecasting", + kind="forecast-wrap-up", + status="pending", + depends_on=["cycle-001"], + payload={}, + ), + ] + ) + # The second automation shows the other two outcomes an operator needs to tell apart: + # an occurrence which failed before queueing anything, and one which queued and then ran to completion. + failed_before_queueing = AutomationRun( + automation=automations[1], + scheduled_at=datetime(2026, 7, 11, 5, 0, tzinfo=timezone.utc), + schedule_revision=automations[1].schedule_revision, + automation_type="forecasting", + generator_id=generator.id, + dispatch_state="failed", + execution_state="pending", + attempt_count=1, + parameters=dict(automations[1].parameters), + plan={"cronstr": automations[1].cronstr, "timezone": automations[1].timezone}, + last_error_type="ValidationError", + last_error_message="forecast output sensor no longer exists", + ) + fully_queued_and_succeeded = AutomationRun( + automation=automations[1], + scheduled_at=datetime(2026, 7, 11, 4, 0, tzinfo=timezone.utc), + schedule_revision=automations[1].schedule_revision, + automation_type="forecasting", + generator_id=generator.id, + dispatch_state="queued", + execution_state="succeeded", + attempt_count=2, + first_enqueued_at=datetime(2026, 7, 11, 4, 1, tzinfo=timezone.utc), + dispatch_completed_at=datetime(2026, 7, 11, 4, 2, tzinfo=timezone.utc), + execution_started_at=datetime(2026, 7, 11, 4, 3, tzinfo=timezone.utc), + execution_completed_at=datetime(2026, 7, 11, 4, 9, tzinfo=timezone.utc), + parameters=dict(automations[1].parameters), + plan={"cronstr": automations[1].cronstr, "timezone": automations[1].timezone}, + ) + fresh_db.session.add_all([failed_before_queueing, fully_queued_and_succeeded]) + fresh_db.session.flush() + fresh_db.session.add_all( + [ + AutomationRunAttempt( + run=failed_before_queueing, + attempt_no=1, + owner="runner-a:1", + started_at=datetime(2026, 7, 11, 5, 0, tzinfo=timezone.utc), + finished_at=datetime(2026, 7, 11, 5, 0, tzinfo=timezone.utc), + outcome="failed", + queued_job_count=0, + error_type="ValidationError", + error_message="forecast output sensor no longer exists", + ), + AutomationRunAttempt( + run=fully_queued_and_succeeded, + attempt_no=1, + owner="runner-a:1", + started_at=datetime(2026, 7, 11, 4, 0, tzinfo=timezone.utc), + finished_at=datetime(2026, 7, 11, 4, 0, tzinfo=timezone.utc), + outcome="failed", + queued_job_count=0, + error_type="ConnectionError", + error_message="lost Redis connection", + ), + AutomationRunAttempt( + run=fully_queued_and_succeeded, + attempt_no=2, + owner="runner-b:2", + started_at=datetime(2026, 7, 11, 4, 1, tzinfo=timezone.utc), + finished_at=datetime(2026, 7, 11, 4, 2, tzinfo=timezone.utc), + outcome="queued", + queued_job_count=1, + ), + AutomationRunJob( + run=fully_queued_and_succeeded, + logical_job_key="cycle-001", + rq_job_id=f"automation-run-{fully_queued_and_succeeded.id}-cycle-001", + queue="forecasting", + kind="forecast-cycle", + status="succeeded", + depends_on=[], + payload={}, + enqueued_at=datetime(2026, 7, 11, 4, 1, tzinfo=timezone.utc), + started_at=datetime(2026, 7, 11, 4, 3, tzinfo=timezone.utc), + finished_at=datetime(2026, 7, 11, 4, 9, tzinfo=timezone.utc), + ), + ] + ) return automations @@ -103,6 +231,7 @@ def test_get_automations( assert day_ahead["cursor"] == "2026-07-11T06:00:00+02:00" assert day_ahead["next-run"] == "2026-07-11T06:00:00+02:00" assert day_ahead["recurrence-description"] == "At 06:00" + assert day_ahead["schedule-revision"] == 1 assert day_ahead["active"] is True assert day_ahead["created-at"] is not None intraday = next(a for a in automations if a["name"] == "Intraday forecasts") @@ -142,7 +271,23 @@ def test_get_automation_details( assert response.json["timezone"] == "Europe/Amsterdam" assert response.json["cursor"] == "2026-07-11T06:00:00+02:00" assert response.json["next-run"] == "2026-07-11T06:00:00+02:00" + assert response.json["schedule-revision"] == 1 assert response.json["parameters"] == {"sensor": battery.sensors[0].id} + run_stats = response.json["run-stats"] + assert run_stats["total"] == 1 + assert run_stats["dispatch"] == {"partially_queued": 1} + assert run_stats["execution"] == {"pending": 1} + assert run_stats["latest-run"]["dispatch-state"] == "partially_queued" + assert run_stats["latest-run"]["attempt-count"] == 2 + assert run_stats["latest-run"]["queued-job-count"] == 1 + assert run_stats["latest-run"]["last-error"] == { + "type": "ConnectionError", + "message": "lost Redis connection", + } + assert [job["logical-job-key"] for job in run_stats["latest-run"]["jobs"]] == [ + "cycle-001", + "wrap-up", + ] assert response.json["job-stats"] == {} # this automation has not queued any jobs # the sensor to forecast is both read from (its history) and written to sensor = {"id": battery.sensors[0].id, "name": battery.sensors[0].name} @@ -150,6 +295,61 @@ def test_get_automation_details( assert response.json["output-sensors"] == [sensor] +@pytest.mark.parametrize( + "requesting_user", ["test_prosumer_user@seita.nl"], indirect=True +) +def test_get_automation_details_distinguishes_run_outcomes( + app, + add_battery_assets_fresh_db, + add_automations, + requesting_user, +): + """An operator can tell a pre-queue failure, a completed dispatch and its execution outcome apart.""" + battery = add_battery_assets_fresh_db["Test battery"] + automation = add_automations[1] + with app.test_client() as client: + response = client.get( + url_for( + "AssetAPI:get_automation", + id=battery.id, + automation_id=automation.id, + ), + ) + assert response.status_code == 200 + run_stats = response.json["run-stats"] + assert run_stats["total"] == 2 + assert run_stats["dispatch"] == {"failed": 1, "queued": 1} + assert run_stats["execution"] == {"pending": 1, "succeeded": 1} + + # The most recent occurrence failed before it queued anything, so it can be retried in full. + latest_run = run_stats["latest-run"] + assert latest_run["scheduled-at"] == "2026-07-11T05:00:00+00:00" + assert latest_run["dispatch-state"] == "failed" + assert latest_run["intended-job-count"] == 0 + assert latest_run["queued-job-count"] == 0 + assert latest_run["first-enqueued-at"] is None + assert latest_run["last-error"] == { + "type": "ValidationError", + "message": "forecast output sensor no longer exists", + } + assert latest_run["latest-attempt"]["attempt-no"] == 1 + assert latest_run["latest-attempt"]["outcome"] == "failed" + + # The earlier occurrence needed a retry, finished queueing, and its jobs then succeeded. + retried_run = run_stats["recent-runs"][1] + assert retried_run["scheduled-at"] == "2026-07-11T04:00:00+00:00" + assert retried_run["dispatch-state"] == "queued" + assert retried_run["execution-state"] == "succeeded" + assert retried_run["attempt-count"] == 2 + assert retried_run["dispatch-completed-at"] == "2026-07-11T04:02:00+00:00" + assert retried_run["execution-completed-at"] == "2026-07-11T04:09:00+00:00" + assert retried_run["latest-attempt"]["attempt-no"] == 2 + assert retried_run["latest-attempt"]["owner"] == "runner-b:2" + assert retried_run["latest-attempt"]["outcome"] == "queued" + assert retried_run["latest-attempt"]["error"] is None + assert [job["status"] for job in retried_run["jobs"]] == ["succeeded"] + + @pytest.mark.parametrize( "requesting_user", ["test_prosumer_user@seita.nl"], indirect=True ) diff --git a/flexmeasures/api/v3_0/tests/test_sensor_data.py b/flexmeasures/api/v3_0/tests/test_sensor_data.py index 2a6fdd54cf..b4230cd70f 100644 --- a/flexmeasures/api/v3_0/tests/test_sensor_data.py +++ b/flexmeasures/api/v3_0/tests/test_sensor_data.py @@ -637,35 +637,39 @@ def receive_handle_error(exception_context): # If the assert failed, we would get a 500 status code assert error_info.__class__.__name__ == "IntegrityError" - # Check that 1st time posting the data succeeds - response = client.post( - url_for("SensorAPI:post_data", id=sensor.id), - json=post_data, - ) - print(response.json) - assert response.status_code == 200 - - # Check that 2nd time posting the same data succeeds informatively - response = client.post( - url_for("SensorAPI:post_data", id=sensor.id), - json=post_data, - ) - print(response.json) - assert response.status_code == 200 - assert "data has already been received" in response.json["message"] - - # Check that replacing data fails informatively - post_data["values"][0] = 100 - response = client.post( - url_for("SensorAPI:post_data", id=sensor.id), - json=post_data, - ) - print(response.json) - assert response.status_code == 403 - assert "data represents a replacement" in response.json["message"] + try: + # Check that 1st time posting the data succeeds + response = client.post( + url_for("SensorAPI:post_data", id=sensor.id), + json=post_data, + ) + print(response.json) + assert response.status_code == 200 - # at this point, the transaction has failed and needs to be rolled back. - db.session.rollback() + # Check that 2nd time posting the same data succeeds informatively + response = client.post( + url_for("SensorAPI:post_data", id=sensor.id), + json=post_data, + ) + print(response.json) + assert response.status_code == 200 + assert "data has already been received" in response.json["message"] + + # Check that replacing data fails informatively + post_data["values"][0] = 100 + response = client.post( + url_for("SensorAPI:post_data", id=sensor.id), + json=post_data, + ) + print(response.json) + assert response.status_code == 403 + assert "data represents a replacement" in response.json["message"] + + # at this point, the transaction has failed and needs to be rolled back. + db.session.rollback() + finally: + # Without this, the listener would outlive the test and assert on every later database error in the process. + event.remove(Engine, "handle_error", receive_handle_error) @pytest.mark.parametrize( diff --git a/flexmeasures/cli/jobs.py b/flexmeasures/cli/jobs.py index 296228bf7a..801bc91fce 100644 --- a/flexmeasures/cli/jobs.py +++ b/flexmeasures/cli/jobs.py @@ -37,9 +37,9 @@ from flexmeasures.data.schemas import AssetIdField, SensorIdField from flexmeasures.data.schemas.automations import AutomationIdField from flexmeasures.data.services.automations import ( - claim_due_automation, + dispatch_automation_run, floor_to_minute, - get_due_automations, + get_dispatchable_automation_runs, run_automation, ) from flexmeasures.data.services.scheduling import handle_scheduling_exception @@ -84,53 +84,31 @@ def run_automations(): \b * * * * * flexmeasures jobs run-automations - A Redis-based guard allows at most one queueing attempt per scheduled run. - A failed attempt is not retried automatically, because it may already have queued some jobs. + Durable automation run records claim each scheduled run and make queueing resumable. + Failed dispatch attempts are retried safely by reusing the original run plan and deterministic job IDs. """ now = floor_to_minute(server_now()) - due_automations = get_due_automations(now) - if not due_automations: + claimed_runs = get_dispatchable_automation_runs(now) + if not claimed_runs: click.secho(f"No automations due at {now}.", **MsgStyle.SUCCESS) return - connection = app.queues["forecasting"].connection n_run = 0 n_failed = 0 - for due_automation in due_automations: - automation = due_automation.automation - # Guard the canonical run, including catch-ups and repeated wall times. - guard_key = ( - f"automation-run:{automation.id}:{due_automation.scheduled_at.isoformat()}" - ) - if not connection.set(guard_key, 1, nx=True, ex=120): - click.secho( - f"Automation {automation.id} ('{automation.name}') was already attempted for {due_automation.scheduled_at}. " - "Skipping to avoid duplicate jobs.", - **MsgStyle.WARN, - ) - continue - if not claim_due_automation(due_automation): - click.secho( - f"Automation {automation.id} ('{automation.name}') run {due_automation.scheduled_at} was already claimed. Skipping to avoid duplicate jobs.", - **MsgStyle.WARN, - ) - continue + for claimed_run in claimed_runs: + automation = claimed_run.run.automation try: - returns = run_automation( - automation, scheduled_at=due_automation.scheduled_at - ) - n_jobs = returns.get("n_jobs") if returns else 0 + returns = dispatch_automation_run(claimed_run) + n_jobs = returns["n_jobs"] click.secho( - f"Automation {automation.id} ('{automation.name}') queued {n_jobs} {automation.type} job(s) for asset {automation.asset_id}.", + f"Automation {automation.id} ('{automation.name}') run {claimed_run.run.id} queued {n_jobs} {automation.type} job(s) for asset {automation.asset_id}, scheduled for {claimed_run.run.scheduled_at}.", **MsgStyle.SUCCESS, ) n_run += 1 except Exception as e: db.session.rollback() - # Queueing a multi-cycle forecast is not transactional. Keep the guard - # because this attempt may have queued some jobs before failing. click.secho( - f"Automation {automation.id} ('{automation.name}') failed to queue jobs: {e}", + f"Automation {automation.id} ('{automation.name}') run {claimed_run.run.id} failed while dispatching attempt {claimed_run.attempt.attempt_no}: {e}", **MsgStyle.ERROR, ) n_failed += 1 diff --git a/flexmeasures/cli/tests/test_automations.py b/flexmeasures/cli/tests/test_automations.py index de6b05d706..31c6f5582a 100644 --- a/flexmeasures/cli/tests/test_automations.py +++ b/flexmeasures/cli/tests/test_automations.py @@ -170,7 +170,7 @@ def test_add_automation_default_cron( """Without --cron, an automation recurs daily.""" from flexmeasures.cli.data_add import add_automation from flexmeasures.data.services.automations import ( - claim_due_automation, + claim_due_automation_run, get_due_automations, ) @@ -196,7 +196,7 @@ def test_add_automation_default_cron( assert [d.automation.id for d in due] == [automation.id] # and, once claimed, not handed out again an hour later - assert claim_due_automation(due[0]) + assert claim_due_automation_run(due[0]) is not None assert get_due_automations(midnight + timedelta(hours=1)) == [] @@ -1651,12 +1651,16 @@ def test_run_report_automation( assert result.exit_code == 0, result.output assert "queued 1 reporting job(s)" in result.output, result.output - # the queued job recorded how it was created + # the queued job recorded how it was created, including the durable run it belongs to + from flexmeasures.data.models.automations import AutomationRun + + run = fresh_db.session.execute(select(AutomationRun)).scalar_one() jobs = app.queues["reporting"].jobs assert len(jobs) == 1 assert jobs[0].meta["trigger"] == { "origin": "automation", "automation_id": automation.id, + "automation_run_id": run.id, } # the covered-until anchor is only recorded once the job succeeds @@ -1897,41 +1901,35 @@ def test_run_automations_catches_up_once_after_downtime( assert automation.cursor == datetime(2026, 1, 15, 9, 0, tzinfo=timezone.utc) -def test_failed_automation_attempt_is_not_retried(app, clean_redis, mocker): - """A failure after partial queueing must not duplicate that work on retry.""" +def test_run_automations_reports_durable_run_status(app, clean_redis, mocker): + """The automation runner reports durable run and retry-attempt identifiers.""" from flexmeasures.cli.jobs import run_automations - from flexmeasures.data.services.automations import DueAutomation - automation = SimpleNamespace(id=42, name="Partial run", asset_id=1) - due_automation = DueAutomation( + automation = SimpleNamespace( + id=42, name="Partial run", asset_id=1, type="scheduling" + ) + run = SimpleNamespace( + id=7, automation=automation, scheduled_at=datetime(2026, 8, 5, 1, 0, tzinfo=timezone.utc), - expected_cursor=datetime(2026, 8, 5, 0, 0, tzinfo=timezone.utc), - expected_cronstr="0 * * * *", - expected_timezone="UTC", ) + attempt = SimpleNamespace(attempt_no=2) + claimed_run = SimpleNamespace(run=run, attempt=attempt) mocker.patch( - "flexmeasures.cli.jobs.get_due_automations", return_value=[due_automation] + "flexmeasures.cli.jobs.get_dispatchable_automation_runs", + return_value=[claimed_run], + ) + mocker.patch( + "flexmeasures.cli.jobs.dispatch_automation_run", + return_value={"run_id": 7, "job_id": "job-1", "n_jobs": 3}, ) - mocker.patch("flexmeasures.cli.jobs.claim_due_automation", return_value=True) - - def queue_then_fail(_automation, **_kwargs): - app.queues["forecasting"].enqueue("flexmeasures.utils.time_utils.server_now") - raise RuntimeError("failed after queueing") - - mocker.patch("flexmeasures.cli.jobs.run_automation", side_effect=queue_then_fail) runner = app.test_cli_runner() - first_result = runner.invoke(run_automations) - assert first_result.exit_code == 1, first_result.output - assert "failed after queueing" in first_result.output - assert app.queues["forecasting"].count == 1 - - retry_result = runner.invoke(run_automations) - assert retry_result.exit_code == 0, retry_result.output - assert "already attempted" in retry_result.output - assert "Skipping to avoid duplicate jobs" in retry_result.output - assert app.queues["forecasting"].count == 1 + result = runner.invoke(run_automations) + + assert result.exit_code == 0, result.output + assert "run 7 queued 3 scheduling job(s) for asset 1" in result.output + assert "scheduled for 2026-08-05 01:00:00+00:00" in result.output def test_run_automation_revalidates_output_scope( diff --git a/flexmeasures/data/migrations/versions/f3d8e2c9a741_add_durable_automation_runs.py b/flexmeasures/data/migrations/versions/f3d8e2c9a741_add_durable_automation_runs.py new file mode 100644 index 0000000000..c7580d7e18 --- /dev/null +++ b/flexmeasures/data/migrations/versions/f3d8e2c9a741_add_durable_automation_runs.py @@ -0,0 +1,162 @@ +"""add durable automation runs + +Revision ID: f3d8e2c9a741 +Revises: c7a2f13b9e04 +Create Date: 2026-08-28 13:10:00.000000 + +""" + +from alembic import op +import sqlalchemy as sa +from sqlalchemy.dialects import postgresql + +# revision identifiers, used by Alembic. +revision = "f3d8e2c9a741" +down_revision = "c7a2f13b9e04" +branch_labels = None +depends_on = None + + +def upgrade(): + op.add_column( + "automation", + sa.Column( + "schedule_revision", sa.Integer(), nullable=False, server_default="1" + ), + ) + op.alter_column("automation", "schedule_revision", server_default=None) + + op.create_table( + "automation_run", + sa.Column("id", sa.Integer(), autoincrement=True, nullable=False), + sa.Column("automation_id", sa.Integer(), nullable=False), + sa.Column("created_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("scheduled_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("schedule_revision", sa.Integer(), nullable=False), + sa.Column("automation_type", sa.String(length=80), nullable=False), + sa.Column("generator_id", sa.Integer(), nullable=True), + sa.Column("dispatch_state", sa.String(length=32), nullable=False), + sa.Column("execution_state", sa.String(length=32), nullable=False), + sa.Column("claim_owner", sa.String(length=128), nullable=True), + sa.Column("claimed_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("claim_expires_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("attempt_count", sa.Integer(), nullable=False), + sa.Column("first_enqueued_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("dispatch_completed_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("execution_started_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("execution_completed_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("last_error_type", sa.String(length=160), nullable=True), + sa.Column("last_error_message", sa.Text(), nullable=True), + sa.Column( + "parameters", postgresql.JSONB(astext_type=sa.Text()), nullable=False + ), + sa.Column("plan", postgresql.JSONB(astext_type=sa.Text()), nullable=False), + sa.CheckConstraint( + "dispatch_state IN ('pending', 'claimed', 'partially_queued', 'queued', 'failed')", + name=op.f("automation_run_automation_run_dispatch_state_ck"), + ), + sa.CheckConstraint( + "execution_state IN ('pending', 'running', 'succeeded', 'failed', 'canceled')", + name=op.f("automation_run_automation_run_execution_state_ck"), + ), + sa.ForeignKeyConstraint( + ["automation_id"], + ["automation.id"], + name=op.f("automation_run_automation_id_automation_fkey"), + ondelete="CASCADE", + ), + sa.PrimaryKeyConstraint("id", name=op.f("automation_run_pkey")), + sa.UniqueConstraint( + "automation_id", + "scheduled_at", + "schedule_revision", + name="automation_run_occurrence_uq", + ), + ) + op.create_index( + "automation_run_dispatch_state_idx", + "automation_run", + ["dispatch_state", "claim_expires_at"], + ) + op.create_index( + "automation_run_automation_scheduled_at_idx", + "automation_run", + ["automation_id", "scheduled_at"], + ) + + op.create_table( + "automation_run_attempt", + sa.Column("id", sa.Integer(), autoincrement=True, nullable=False), + sa.Column("run_id", sa.Integer(), nullable=False), + sa.Column("attempt_no", sa.Integer(), nullable=False), + sa.Column("owner", sa.String(length=128), nullable=False), + sa.Column("started_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("finished_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("outcome", sa.String(length=64), nullable=True), + sa.Column("queued_job_count", sa.Integer(), nullable=False), + sa.Column("error_type", sa.String(length=160), nullable=True), + sa.Column("error_message", sa.Text(), nullable=True), + sa.ForeignKeyConstraint( + ["run_id"], + ["automation_run.id"], + name=op.f("automation_run_attempt_run_id_automation_run_fkey"), + ondelete="CASCADE", + ), + sa.PrimaryKeyConstraint("id", name=op.f("automation_run_attempt_pkey")), + sa.UniqueConstraint( + "run_id", "attempt_no", name="automation_run_attempt_no_uq" + ), + ) + + op.create_table( + "automation_run_job", + sa.Column("id", sa.Integer(), autoincrement=True, nullable=False), + sa.Column("run_id", sa.Integer(), nullable=False), + sa.Column("logical_job_key", sa.String(length=128), nullable=False), + sa.Column("rq_job_id", sa.String(length=191), nullable=False), + sa.Column("queue", sa.String(length=80), nullable=False), + sa.Column("kind", sa.String(length=80), nullable=False), + sa.Column("status", sa.String(length=32), nullable=False), + sa.Column("enqueued_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("started_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("finished_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("last_error_type", sa.String(length=160), nullable=True), + sa.Column("last_error_message", sa.Text(), nullable=True), + sa.Column( + "depends_on", postgresql.JSONB(astext_type=sa.Text()), nullable=False + ), + sa.Column("payload", postgresql.JSONB(astext_type=sa.Text()), nullable=False), + sa.CheckConstraint( + "status IN ('pending', 'queued', 'running', 'succeeded', 'failed', 'canceled')", + name=op.f("automation_run_job_automation_run_job_status_ck"), + ), + sa.ForeignKeyConstraint( + ["run_id"], + ["automation_run.id"], + name=op.f("automation_run_job_run_id_automation_run_fkey"), + ondelete="CASCADE", + ), + sa.PrimaryKeyConstraint("id", name=op.f("automation_run_job_pkey")), + sa.UniqueConstraint("rq_job_id", name="automation_run_job_rq_job_uq"), + sa.UniqueConstraint( + "run_id", "logical_job_key", name="automation_run_job_logical_uq" + ), + ) + op.create_index( + "automation_run_job_run_status_idx", + "automation_run_job", + ["run_id", "status"], + ) + + +def downgrade(): + op.drop_index("automation_run_job_run_status_idx", table_name="automation_run_job") + op.drop_table("automation_run_job") + op.drop_table("automation_run_attempt") + op.drop_index( + "automation_run_automation_scheduled_at_idx", table_name="automation_run" + ) + op.drop_index("automation_run_dispatch_state_idx", table_name="automation_run") + op.drop_table("automation_run") + op.drop_column("automation", "schedule_revision") diff --git a/flexmeasures/data/models/automations.py b/flexmeasures/data/models/automations.py index 54a96de65f..89db693aad 100644 --- a/flexmeasures/data/models/automations.py +++ b/flexmeasures/data/models/automations.py @@ -7,7 +7,7 @@ from flask import current_app from pytz import all_timezones_set from sqlalchemy.dialects.postgresql import JSONB -from sqlalchemy.ext.mutable import MutableDict +from sqlalchemy.ext.mutable import MutableDict, MutableList from sqlalchemy.orm import validates from flexmeasures.auth.policy import AuthModelMixin @@ -93,6 +93,7 @@ class Automation(db.Model, AuthModelMixin): nullable=False, default=get_initial_cursor, ) + schedule_revision = db.Column(db.Integer, nullable=False, default=1) active = db.Column(db.Boolean, nullable=False, default=True) generator_id = db.Column( db.Integer, db.ForeignKey("data_source.id"), nullable=False @@ -107,6 +108,14 @@ class Automation(db.Model, AuthModelMixin): ), ) generator = db.relationship("DataSource", foreign_keys=[generator_id]) + runs = db.relationship( + "AutomationRun", + back_populates="automation", + lazy=True, + cascade="all, delete-orphan", + passive_deletes=True, + order_by="desc(AutomationRun.scheduled_at)", + ) @validates("timezone") def validate_timezone(self, key: str, timezone: str) -> str: @@ -163,3 +172,209 @@ def output_sensors(self) -> list: from flexmeasures.data.services.automations import get_automation_sensors return get_automation_sensors(self)["output_sensors"] + + +# A job intent counts as dispatched from this status onwards: it is in Redis, whatever became of it since. +AUTOMATION_RUN_JOB_QUEUED_OR_LATER = ( + "queued", + "running", + "succeeded", + "failed", + "canceled", +) + + +class AutomationRun(db.Model): + """Durable execution record for one scheduled automation occurrence.""" + + __tablename__ = "automation_run" + __table_args__ = ( + db.UniqueConstraint( + "automation_id", + "scheduled_at", + "schedule_revision", + name="automation_run_occurrence_uq", + ), + db.CheckConstraint( + "dispatch_state IN ('pending', 'claimed', 'partially_queued', 'queued', 'failed')", + name="automation_run_dispatch_state_ck", + ), + db.CheckConstraint( + "execution_state IN ('pending', 'running', 'succeeded', 'failed', 'canceled')", + name="automation_run_execution_state_ck", + ), + ) + + id = db.Column(db.Integer, autoincrement=True, primary_key=True) + automation_id = db.Column( + db.Integer, + db.ForeignKey("automation.id", ondelete="CASCADE"), + nullable=False, + ) + created_at = db.Column( + db.DateTime(timezone=True), nullable=False, default=server_now + ) + updated_at = db.Column( + db.DateTime(timezone=True), + nullable=False, + default=server_now, + onupdate=server_now, + ) + scheduled_at = db.Column(db.DateTime(timezone=True), nullable=False) + schedule_revision = db.Column(db.Integer, nullable=False) + automation_type = db.Column(db.String(80), nullable=False) + generator_id = db.Column(db.Integer, nullable=True) + dispatch_state = db.Column(db.String(32), nullable=False, default="pending") + execution_state = db.Column(db.String(32), nullable=False, default="pending") + claim_owner = db.Column(db.String(128), nullable=True) + claimed_at = db.Column(db.DateTime(timezone=True), nullable=True) + claim_expires_at = db.Column(db.DateTime(timezone=True), nullable=True) + attempt_count = db.Column(db.Integer, nullable=False, default=0) + first_enqueued_at = db.Column(db.DateTime(timezone=True), nullable=True) + dispatch_completed_at = db.Column(db.DateTime(timezone=True), nullable=True) + execution_started_at = db.Column(db.DateTime(timezone=True), nullable=True) + execution_completed_at = db.Column(db.DateTime(timezone=True), nullable=True) + last_error_type = db.Column(db.String(160), nullable=True) + last_error_message = db.Column(db.Text, nullable=True) + parameters = db.Column(MutableDict.as_mutable(JSONB), nullable=False, default=dict) + plan = db.Column(MutableDict.as_mutable(JSONB), nullable=False, default=dict) + + automation = db.relationship("Automation", back_populates="runs") + attempts = db.relationship( + "AutomationRunAttempt", + back_populates="run", + cascade="all, delete-orphan", + passive_deletes=True, + order_by="AutomationRunAttempt.attempt_no", + ) + job_intents = db.relationship( + "AutomationRunJob", + back_populates="run", + cascade="all, delete-orphan", + passive_deletes=True, + order_by="AutomationRunJob.logical_job_key", + ) + + @validates( + "scheduled_at", + "created_at", + "updated_at", + "claimed_at", + "claim_expires_at", + "first_enqueued_at", + "dispatch_completed_at", + "execution_started_at", + "execution_completed_at", + ) + def validate_datetime_is_aware( + self, key: str, value: datetime | None + ) -> datetime | None: + """Store all automation run timestamps as timezone-aware UTC datetimes.""" + if value is None: + return None + if value.tzinfo is None or value.utcoffset() is None: + raise ValueError(f"Automation run {key} must be timezone-aware.") + return value.astimezone(timezone.utc) + + @property + def intended_job_count(self) -> int: + """Return the number of persisted logical job intents.""" + return len(self.job_intents) + + @property + def queued_job_count(self) -> int: + """Return the number of logical jobs durably marked as queued or later.""" + return sum( + 1 + for intent in self.job_intents + if intent.status in AUTOMATION_RUN_JOB_QUEUED_OR_LATER + ) + + +class AutomationRunAttempt(db.Model): + """One durable attempt to claim and dispatch an automation run.""" + + __tablename__ = "automation_run_attempt" + __table_args__ = ( + db.UniqueConstraint( + "run_id", "attempt_no", name="automation_run_attempt_no_uq" + ), + ) + + id = db.Column(db.Integer, autoincrement=True, primary_key=True) + run_id = db.Column( + db.Integer, + db.ForeignKey("automation_run.id", ondelete="CASCADE"), + nullable=False, + ) + attempt_no = db.Column(db.Integer, nullable=False) + owner = db.Column(db.String(128), nullable=False) + started_at = db.Column( + db.DateTime(timezone=True), nullable=False, default=server_now + ) + finished_at = db.Column(db.DateTime(timezone=True), nullable=True) + outcome = db.Column(db.String(64), nullable=True) + queued_job_count = db.Column(db.Integer, nullable=False, default=0) + error_type = db.Column(db.String(160), nullable=True) + error_message = db.Column(db.Text, nullable=True) + + run = db.relationship("AutomationRun", back_populates="attempts") + + @validates("started_at", "finished_at") + def validate_datetime_is_aware( + self, key: str, value: datetime | None + ) -> datetime | None: + """Store all automation run attempt timestamps as timezone-aware UTC datetimes.""" + if value is None: + return None + if value.tzinfo is None or value.utcoffset() is None: + raise ValueError(f"Automation run attempt {key} must be timezone-aware.") + return value.astimezone(timezone.utc) + + +class AutomationRunJob(db.Model): + """Durable outbox record for one logical job in an automation run.""" + + __tablename__ = "automation_run_job" + __table_args__ = ( + db.UniqueConstraint( + "run_id", "logical_job_key", name="automation_run_job_logical_uq" + ), + db.UniqueConstraint("rq_job_id", name="automation_run_job_rq_job_uq"), + db.CheckConstraint( + "status IN ('pending', 'queued', 'running', 'succeeded', 'failed', 'canceled')", + name="automation_run_job_status_ck", + ), + ) + + id = db.Column(db.Integer, autoincrement=True, primary_key=True) + run_id = db.Column( + db.Integer, + db.ForeignKey("automation_run.id", ondelete="CASCADE"), + nullable=False, + ) + logical_job_key = db.Column(db.String(128), nullable=False) + rq_job_id = db.Column(db.String(191), nullable=False) + queue = db.Column(db.String(80), nullable=False, default="forecasting") + kind = db.Column(db.String(80), nullable=False) + status = db.Column(db.String(32), nullable=False, default="pending") + enqueued_at = db.Column(db.DateTime(timezone=True), nullable=True) + started_at = db.Column(db.DateTime(timezone=True), nullable=True) + finished_at = db.Column(db.DateTime(timezone=True), nullable=True) + last_error_type = db.Column(db.String(160), nullable=True) + last_error_message = db.Column(db.Text, nullable=True) + depends_on = db.Column(MutableList.as_mutable(JSONB), nullable=False, default=list) + payload = db.Column(MutableDict.as_mutable(JSONB), nullable=False, default=dict) + + run = db.relationship("AutomationRun", back_populates="job_intents") + + @validates("enqueued_at", "started_at", "finished_at") + def validate_datetime_is_aware( + self, key: str, value: datetime | None + ) -> datetime | None: + """Store all automation run job timestamps as timezone-aware UTC datetimes.""" + if value is None: + return None + if value.tzinfo is None or value.utcoffset() is None: + raise ValueError(f"Automation run job {key} must be timezone-aware.") + return value.astimezone(timezone.utc) diff --git a/flexmeasures/data/models/data_sources.py b/flexmeasures/data/models/data_sources.py index 9f1a9b7aab..e2fd0213c1 100644 --- a/flexmeasures/data/models/data_sources.py +++ b/flexmeasures/data/models/data_sources.py @@ -100,7 +100,12 @@ def __init__( elif len(kwargs) == 0: self._config = self._config_schema.load({}) - def set_job_trigger(self, origin: str, automation_id: int | None = None): + def set_job_trigger( + self, + origin: str, + automation_id: int | None = None, + automation_run_id: int | None = None, + ): """Record how any queued jobs got created (e.g. via the CLI, the API or an automation). This information is stored on the jobs themselves (as job meta data). @@ -108,6 +113,8 @@ def set_job_trigger(self, origin: str, automation_id: int | None = None): self._job_trigger = {"origin": origin} if automation_id is not None: self._job_trigger["automation_id"] = automation_id + if automation_run_id is not None: + self._job_trigger["automation_run_id"] = automation_run_id @property def input_sensors(self) -> list: diff --git a/flexmeasures/data/models/forecasting/pipelines/train_predict.py b/flexmeasures/data/models/forecasting/pipelines/train_predict.py index 62be0da3dd..2151120fc3 100644 --- a/flexmeasures/data/models/forecasting/pipelines/train_predict.py +++ b/flexmeasures/data/models/forecasting/pipelines/train_predict.py @@ -170,30 +170,66 @@ def _load_job_parameters_payload(payload: dict[str, Any]) -> dict[str, Any]: return parameters +# Logical name of the job which reports on all cycle jobs of one pipeline run. +WRAP_UP_LOGICAL_JOB_KEY = "wrap-up" + + def run_train_predict_cycle_job( config: dict, parameters: dict, data_source_id: int, delete_model: bool, + automation_run_id: int | None = None, + logical_job_key: str | None = None, **cycle_params, ): """Run one train-predict cycle after reconstructing worker-local ORM state.""" + from flexmeasures.data.services.automations import ( + record_automation_job_failed, + record_automation_job_started, + record_automation_job_succeeded, + ) + + record_automation_job_started(automation_run_id, logical_job_key) pipeline = TrainPredictPipeline(delete_model=delete_model) pipeline._config = _load_job_config_payload(config) for key, value in pipeline._config.items(): setattr(pipeline, key, value) pipeline._parameters = _load_job_parameters_payload(parameters) pipeline._data_source = _get_attached_data_source(data_source_id) - return pipeline.run_cycle(**cycle_params) - - -def run_train_predict_wrap_up_job(cycle_job_ids: list[str], queue: str = "forecasting"): + try: + result = pipeline.run_cycle(**cycle_params) + except Exception as exc: + record_automation_job_failed(automation_run_id, logical_job_key, exc) + raise + record_automation_job_succeeded(automation_run_id, logical_job_key) + return result + + +def run_train_predict_wrap_up_job( + cycle_job_ids: list[str], + queue: str = "forecasting", + automation_run_id: int | None = None, + logical_job_key: str | None = None, +): """Log the status of all cycle jobs after completion.""" + from flexmeasures.data.services.automations import ( + record_automation_job_failed, + record_automation_job_started, + record_automation_job_succeeded, + ) + + record_automation_job_started(automation_run_id, logical_job_key) connection = current_app.queues[queue].connection - for index, job_id in enumerate(cycle_job_ids): - status = Job.fetch(job_id, connection=connection).get_status() - logging.info(f"{queue} job-{index}: {job_id} status: {status}") + try: + for index, job_id in enumerate(cycle_job_ids): + status = Job.fetch(job_id, connection=connection).get_status() + logging.info(f"{queue} job-{index}: {job_id} status: {status}") + except Exception as exc: + record_automation_job_failed(automation_run_id, logical_job_key, exc) + raise + record_automation_job_succeeded(automation_run_id, logical_job_key) class TrainPredictPipeline(Forecaster): @@ -448,99 +484,250 @@ def run( ) if as_job: - cycle_job_ids = [] - - job_config = _make_job_config_payload(self._config) - job_parameters = _make_job_parameters_payload(self._parameters) - sensor_id = job_parameters["sensor_id"] - sensor_to_save_id = job_parameters["sensor_to_save_id"] - - # Ensure the data source ID is available in the database when the job runs. - self._data_source = db.session.merge(self.data_source) - db.session.commit() - data_source_id = self._data_source.id - - # job metadata for tracking - # Serialize start and end to ISO format strings - # Workaround for https://github.com/Parallels/rq-dashboard/issues/510 - job_metadata = { - "data_source_info": {"id": data_source_id}, - "start": self._parameters["predict_start"].isoformat(), - "end": self._parameters["end_date"].isoformat(), - "sensor_id": sensor_to_save_id, + return self._queue_cycle_jobs(cycles_job_params, queue, connection) + + return self.return_values + + def _persist_data_source_id(self) -> int: + """Make sure this pipeline's data source is in the database, so that the workers can look it up.""" + self._data_source = db.session.merge(self.data_source) + db.session.commit() + data_source_id = self._data_source.id + return data_source_id + + def _job_ttls(self) -> tuple[int, int]: + """Return the time-to-live of a job and of its result, in seconds. + + NB job.cleanup docs say that a negative number of seconds means persisting forever. + """ + return ( + int( + current_app.config.get( + "FLEXMEASURES_JOB_TTL", timedelta(-1) + ).total_seconds() + ), + int( + current_app.config.get( + "FLEXMEASURES_PLANNING_TTL", timedelta(-1) + ).total_seconds() + ), + ) + + def _job_meta( + self, + job_metadata: dict, + job_spec: dict, + automation_run_id: int | None, + ) -> dict: + """Return the metadata to store on one job, identifying its automation run where there is one.""" + meta = dict(job_metadata) + if automation_run_id is not None: + meta["automation_run_id"] = automation_run_id + meta["logical_job_key"] = job_spec["logical_job_key"] + return meta + + def _plan_cycle_jobs( + self, + cycles_job_params: list[dict], + queue: str, + data_source_id: int, + job_metadata: dict, + automation_run_id: int | None, + ) -> list[dict]: + """Describe every job this run intends to create, before any of them is queued. + + Each job gets a logical key which stays the same across retries of an automation run, and a job ID derived from it, + so that a retry recognises the jobs it already queued instead of queueing them a second time. + Outside an automation run there is nothing to retry, so RQ is left to make up the job IDs. + """ + job_config = _make_job_config_payload(self._config) + job_parameters = _make_job_parameters_payload(self._parameters) + + def rq_job_id_for(logical_job_key: str) -> str | None: + if automation_run_id is None: + return None + return f"automation-run-{automation_run_id}-{logical_job_key}" + + cycle_specs = [] + for cycle_params in cycles_job_params: + logical_job_key = f"cycle-{cycle_params['counter']:03d}" + job_kwargs = { + "config": job_config, + "parameters": job_parameters, + "data_source_id": data_source_id, + "delete_model": self.delete_model, + "automation_run_id": automation_run_id, + "logical_job_key": logical_job_key, + **cycle_params, } - if self._job_trigger: - job_metadata["trigger"] = self._job_trigger - for cycle_params in cycles_job_params: - job_kwargs = { - "config": job_config, - "parameters": job_parameters, - "data_source_id": data_source_id, - "delete_model": self.delete_model, - **cycle_params, + _assert_no_orm_objects(job_kwargs) + cycle_specs.append( + { + "logical_job_key": logical_job_key, + "rq_job_id": rq_job_id_for(logical_job_key), + "queue": queue, + "kind": "forecast-cycle", + "depends_on": [], + "payload": {"kwargs": job_kwargs, "meta": job_metadata}, } - _assert_no_orm_objects(job_kwargs) + ) + wrap_up_spec = { + "logical_job_key": WRAP_UP_LOGICAL_JOB_KEY, + "rq_job_id": rq_job_id_for(WRAP_UP_LOGICAL_JOB_KEY), + "queue": queue, + "kind": "forecast-wrap-up", + "depends_on": [spec["logical_job_key"] for spec in cycle_specs], + "payload": { + "kwargs": { + "cycle_job_ids": [spec["rq_job_id"] for spec in cycle_specs], + "queue": queue, + "automation_run_id": automation_run_id, + "logical_job_key": WRAP_UP_LOGICAL_JOB_KEY, + }, + "meta": job_metadata, + }, + } + return cycle_specs + [wrap_up_spec] - job = Job.create( - run_train_predict_cycle_job, - kwargs=job_kwargs, - connection=connection, - ttl=int( - current_app.config.get( - "FLEXMEASURES_JOB_TTL", timedelta(-1) - ).total_seconds() - ), - result_ttl=int( - current_app.config.get( - "FLEXMEASURES_PLANNING_TTL", timedelta(-1) - ).total_seconds() - ), # NB job.cleanup docs says a negative number of seconds means persisting forever - meta=job_metadata, - ) + def _queue_cycle_jobs( + self, cycles_job_params: list[dict], queue: str, connection + ) -> dict: + """Queue one job per training cycle, plus a wrap-up job which waits for all of them. - # Store the job ID for this cycle - cycle_job_ids.append(job.id) + When this pipeline runs for an automation, the jobs it intends to create are written down first, + so that an attempt which fails halfway can be resumed without queueing the same work twice. + """ + automation_run_id = (self._job_trigger or {}).get("automation_run_id") + data_source_id = self._persist_data_source_id() + job_parameters = _make_job_parameters_payload(self._parameters) + sensor_id = job_parameters["sensor_id"] + + # job metadata for tracking + # Serialize start and end to ISO format strings + # Workaround for https://github.com/Parallels/rq-dashboard/issues/510 + job_metadata = { + "data_source_info": {"id": data_source_id}, + "start": self._parameters["predict_start"].isoformat(), + "end": self._parameters["end_date"].isoformat(), + "sensor_id": job_parameters["sensor_to_save_id"], + } + if self._job_trigger: + job_metadata["trigger"] = self._job_trigger + + job_specs = self._plan_cycle_jobs( + cycles_job_params, queue, data_source_id, job_metadata, automation_run_id + ) + intents = {} + if automation_run_id is not None: + from flexmeasures.data.services.automations import ( + ensure_automation_run_job_intents, + ) - current_app.queues[queue].enqueue_job(job) - current_app.job_cache.add( - sensor_id, - job_id=job.id, - queue=queue, - asset_or_sensor_type="sensor", + intents = { + intent.logical_job_key: intent + for intent in ensure_automation_run_job_intents( + automation_run_id, job_specs ) + } - wrap_up_job = Job.create( - run_train_predict_wrap_up_job, - kwargs={ - "cycle_job_ids": cycle_job_ids, - "queue": queue, - }, # cycles jobs IDs to wait for - connection=connection, - depends_on=cycle_job_ids, # wrap-up job depends on all cycle jobs - ttl=int( - current_app.config.get( - "FLEXMEASURES_JOB_TTL", timedelta(-1) - ).total_seconds() - ), - result_ttl=int( - current_app.config.get( - "FLEXMEASURES_PLANNING_TTL", timedelta(-1) - ).total_seconds() - ), # NB job.cleanup docs says a negative number of seconds means persisting forever - meta=job_metadata, + cycle_job_ids = [] + for job_spec in job_specs: + if job_spec["kind"] != "forecast-cycle": + continue + cycle_job_ids.append( + self._queue_planned_job( + run_train_predict_cycle_job, + job_spec, + intents, + queue, + connection, + job_metadata, + automation_run_id, + cache_for_sensor_id=sensor_id, + ) ) - current_app.queues[queue].enqueue_job(wrap_up_job) - if len(cycle_job_ids) > 1: - # Return the wrap-up job ID if multiple cycle jobs are queued - return {"job_id": wrap_up_job.id, "n_jobs": len(cycle_job_ids)} - else: - # Return the single cycle job ID if only one job is queued - return { - "job_id": ( - cycle_job_ids[0] if len(cycle_job_ids) == 1 else wrap_up_job.id - ), - "n_jobs": 1, - } + wrap_up_spec = job_specs[-1] + # The wrap-up job reports on the cycle jobs, whose IDs are only known now when this is not an automation run. + wrap_up_spec["payload"]["kwargs"]["cycle_job_ids"] = cycle_job_ids + wrap_up_job_id = self._queue_planned_job( + run_train_predict_wrap_up_job, + wrap_up_spec, + intents, + queue, + connection, + job_metadata, + automation_run_id, + depends_on=cycle_job_ids, + ) - return self.return_values + if len(cycle_job_ids) > 1: + # Point at the wrap-up job, as it is the one that completes last. + job_id = wrap_up_job_id + else: + job_id = cycle_job_ids[0] if cycle_job_ids else wrap_up_job_id + if automation_run_id is not None: + # An automation run is accounted for in full, wrap-up job included. + n_jobs = len(cycle_job_ids) + 1 + else: + n_jobs = len(cycle_job_ids) if len(cycle_job_ids) > 1 else 1 + return {"job_id": job_id, "n_jobs": n_jobs} + + def _queue_planned_job( + self, + func, + job_spec: dict, + intents: dict, + queue: str, + connection, + job_metadata: dict, + automation_run_id: int | None, + cache_for_sensor_id: int | None = None, + depends_on: list[str] | None = None, + ) -> str: + """Queue one planned job, unless an earlier attempt already put it in Redis.""" + intent = intents.get(job_spec["logical_job_key"]) + if intent is not None: + from flexmeasures.data.services.automations import ( + reconcile_automation_job_intent, + ) + + if reconcile_automation_job_intent(intent): + # This job survived an earlier attempt at this run, so leave it be. + if cache_for_sensor_id is not None: + current_app.job_cache.add( + cache_for_sensor_id, + job_id=intent.rq_job_id, + queue=queue, + asset_or_sensor_type="sensor", + ) + return intent.rq_job_id + + ttl, result_ttl = self._job_ttls() + job = Job.create( + func, + kwargs=job_spec["payload"]["kwargs"], + connection=connection, + id=job_spec["rq_job_id"], + depends_on=depends_on, + ttl=ttl, + result_ttl=result_ttl, + meta=self._job_meta(job_metadata, job_spec, automation_run_id), + ) + current_app.queues[queue].enqueue_job(job) + if automation_run_id is not None: + from flexmeasures.data.services.automations import ( + mark_automation_job_queued, + ) + + mark_automation_job_queued( + automation_run_id, job_spec["logical_job_key"], job.id + ) + if cache_for_sensor_id is not None: + current_app.job_cache.add( + cache_for_sensor_id, + job_id=job.id, + queue=queue, + asset_or_sensor_type="sensor", + ) + return job.id diff --git a/flexmeasures/data/schemas/automations.py b/flexmeasures/data/schemas/automations.py index 4a070c9641..5f398cb08d 100644 --- a/flexmeasures/data/schemas/automations.py +++ b/flexmeasures/data/schemas/automations.py @@ -163,6 +163,14 @@ class Meta: "example": "2026-08-05T08:00:00+02:00", }, ) + schedule_revision = ma.auto_field( + data_key="schedule-revision", + dump_only=True, + metadata={ + "description": "Execution-affecting schedule/configuration revision used to distinguish durable runs around automation edits and reactivation.", + "example": 2, + }, + ) active = ma.auto_field() @staticmethod diff --git a/flexmeasures/data/services/automations.py b/flexmeasures/data/services/automations.py index 120e2fcf5d..fa28ba942a 100644 --- a/flexmeasures/data/services/automations.py +++ b/flexmeasures/data/services/automations.py @@ -7,6 +7,8 @@ from copy import copy, deepcopy from dataclasses import dataclass from datetime import datetime, timedelta, timezone +import os +import socket from typing import Any from zoneinfo import ZoneInfo, ZoneInfoNotFoundError @@ -18,12 +20,18 @@ from isodate.isoerror import ISO8601Error from flask import current_app from marshmallow import ValidationError -from sqlalchemy import select, update +from rq.job import Job +from sqlalchemy import func, or_, select, update +from sqlalchemy.exc import IntegrityError +from sqlalchemy.orm import selectinload from flexmeasures import Forecaster, Reporter from flexmeasures.data import db from flexmeasures.data.models.automations import ( Automation, + AutomationRun, + AutomationRunAttempt, + AutomationRunJob, get_default_automation_timezone, get_initial_cursor, ) @@ -40,6 +48,19 @@ ) from flexmeasures.utils.time_utils import apply_offset_chain, get_timezone, server_now +AUTOMATION_RUN_CLAIM_LEASE = timedelta(minutes=10) +# How many of an automation's most recent runs its status summary describes in full. +AUTOMATION_RUN_STATS_RECENT_LIMIT = 10 +# Dispatch is only finished once `dispatch_completed_at` is set, so every other dispatch state is resumable. +# A run in one of these states is nevertheless off limits while another runner still holds a live claim on it. +AUTOMATION_RUN_RESUMABLE_DISPATCH_STATES = ( + "pending", + "claimed", + "partially_queued", + "queued", + "failed", +) + @dataclass(frozen=True) class DueAutomation: @@ -52,6 +73,446 @@ class DueAutomation: expected_timezone: str +@dataclass(frozen=True) +class ClaimedAutomationRun: + """An automation run and the attempt which currently owns its dispatch.""" + + run: AutomationRun + attempt: AutomationRunAttempt + + +class AutomationRunClaimError(Exception): + """Raised when an automation occurrence cannot be claimed.""" + + +def _runner_owner() -> str: + """Return a short owner string for an automation-run claim lease.""" + return f"{socket.gethostname()}:{os.getpid()}" + + +def _now_utc() -> datetime: + """Return the current database-facing time as timezone-aware UTC.""" + return server_now().astimezone(timezone.utc) + + +def _claim_expires_at(now: datetime, lease: timedelta) -> datetime: + """Return the UTC timestamp at which a claim becomes stale.""" + return now + lease + + +def _claim_is_available(now: datetime): + """Return the criterion for a run whose claim is free to take at ``now``. + + A claim is free when no runner holds it, or when the runner holding it let its lease expire, + which is how a runner that died mid-dispatch releases its occurrence. + """ + return or_( + AutomationRun.claim_expires_at.is_(None), + AutomationRun.claim_expires_at <= now, + ) + + +def _json_safe(value: Any) -> Any: + """Convert values from an RQ job payload to JSON-compatible diagnostics.""" + if isinstance(value, datetime): + return value.astimezone(timezone.utc).isoformat() + if isinstance(value, timedelta): + return isodate.duration_isoformat(value) + if isinstance(value, dict): + return {str(k): _json_safe(v) for k, v in value.items()} + if isinstance(value, (list, tuple, set)): + return [_json_safe(item) for item in value] + if isinstance(value, (str, int, float, bool)) or value is None: + return value + return repr(value) + + +def _run_snapshot(automation: Automation, scheduled_at: datetime) -> dict[str, Any]: + """Snapshot automation configuration for an immutable run plan.""" + return { + "automation_id": automation.id, + "automation_type": automation.type, + "automation_name": automation.name, + "asset_id": automation.asset_id, + "scheduled_at": scheduled_at.astimezone(timezone.utc).isoformat(), + "schedule_revision": automation.schedule_revision, + "cronstr": automation.cronstr, + "timezone": automation.timezone, + "generator_id": automation.generator_id, + } + + +def _new_attempt(run: AutomationRun, owner: str, now: datetime) -> AutomationRunAttempt: + """Append a durable dispatch attempt to a claimed automation run.""" + attempt = AutomationRunAttempt( + run=run, + attempt_no=run.attempt_count, + owner=owner, + started_at=now, + queued_job_count=run.queued_job_count, + ) + db.session.add(attempt) + return attempt + + +def _finish_attempt( + attempt: AutomationRunAttempt | None, + outcome: str, + queued_job_count: int, + error: BaseException | None = None, +) -> None: + """Record the result of a dispatch attempt.""" + if attempt is None: + return + attempt.finished_at = _now_utc() + attempt.outcome = outcome + attempt.queued_job_count = queued_job_count + if error is not None: + attempt.error_type = error.__class__.__name__ + attempt.error_message = str(error) + + +def claim_due_automation_run( + due_automation: DueAutomation, + owner: str | None = None, + lease: timedelta = AUTOMATION_RUN_CLAIM_LEASE, +) -> ClaimedAutomationRun | None: + """Atomically claim a newly due occurrence and create its durable run.""" + owner = owner or _runner_owner() + now = _now_utc() + if due_automation.expected_cursor is None: + cursor_matches = Automation.cursor.is_(None) + else: + cursor_matches = Automation.cursor == due_automation.expected_cursor + result = db.session.execute( + update(Automation) + .where( + Automation.id == due_automation.automation.id, + Automation.active.is_(True), + Automation.cronstr == due_automation.expected_cronstr, + Automation.timezone == due_automation.expected_timezone, + Automation.schedule_revision == due_automation.automation.schedule_revision, + cursor_matches, + ) + .values(cursor=due_automation.scheduled_at) + .execution_options(synchronize_session=False) + ) + if result.rowcount != 1: + db.session.rollback() + return None + + automation = due_automation.automation + run = AutomationRun( + automation=automation, + scheduled_at=due_automation.scheduled_at, + schedule_revision=automation.schedule_revision, + automation_type=automation.type, + generator_id=automation.generator_id, + dispatch_state="claimed", + execution_state="pending", + claim_owner=owner, + claimed_at=now, + claim_expires_at=_claim_expires_at(now, lease), + attempt_count=1, + parameters=dict(automation.parameters or {}), + plan=_run_snapshot(automation, due_automation.scheduled_at), + ) + db.session.add(run) + db.session.flush() + attempt = _new_attempt(run, owner, now) + try: + db.session.commit() + except IntegrityError: + db.session.rollback() + return None + return ClaimedAutomationRun(run=run, attempt=attempt) + + +def claim_existing_automation_run( + run: AutomationRun, + owner: str | None = None, + lease: timedelta = AUTOMATION_RUN_CLAIM_LEASE, +) -> ClaimedAutomationRun | None: + """Claim a durable automation run whose dispatch is unfinished and unclaimed. + + A run is only up for grabs once no other runner holds a live claim on it, because the dispatch state turns to + 'partially_queued' while the owning runner is still queueing the rest of its jobs. + A runner which fails releases its own claim, so its run is immediately retryable. + """ + owner = owner or _runner_owner() + now = _now_utc() + result = db.session.execute( + update(AutomationRun) + .where( + AutomationRun.id == run.id, + AutomationRun.dispatch_state.in_(AUTOMATION_RUN_RESUMABLE_DISPATCH_STATES), + AutomationRun.dispatch_completed_at.is_(None), + _claim_is_available(now), + ) + .values( + dispatch_state="claimed", + claim_owner=owner, + claimed_at=now, + claim_expires_at=_claim_expires_at(now, lease), + attempt_count=AutomationRun.attempt_count + 1, + ) + .execution_options(synchronize_session=False) + ) + if result.rowcount != 1: + db.session.rollback() + return None + db.session.flush() + claimed_run = db.session.get(AutomationRun, run.id) + assert claimed_run is not None + db.session.refresh(claimed_run) + attempt = _new_attempt(claimed_run, owner, now) + db.session.commit() + return ClaimedAutomationRun(run=claimed_run, attempt=attempt) + + +def get_dispatchable_automation_runs( + now: datetime | None = None, + owner: str | None = None, +) -> list[ClaimedAutomationRun]: + """Claim new due occurrences and resumable durable runs for dispatch.""" + if now is None: + now = _now_utc() + now = floor_to_minute(now) + claimed_runs: list[ClaimedAutomationRun] = [] + for due_automation in get_due_automations(now): + claimed = claim_due_automation_run(due_automation, owner=owner) + if claimed is not None: + claimed_runs.append(claimed) + + resumable_runs = db.session.scalars( + select(AutomationRun) + .join(Automation) + .where( + Automation.active.is_(True), + # Only a forecast run can be dispatched a second time safely. + # Its jobs carry IDs derived from the run, so a retry recognizes the ones it already queued. + # A schedule run's jobs get a fresh ID on every dispatch, so retrying one would duplicate its schedules, + # which is why such a run is recorded and reported, but left where it failed. + AutomationRun.automation_type == "forecasting", + AutomationRun.dispatch_state.in_(AUTOMATION_RUN_RESUMABLE_DISPATCH_STATES), + AutomationRun.dispatch_completed_at.is_(None), + _claim_is_available(now), + ) + .order_by(AutomationRun.scheduled_at, AutomationRun.id) + ).all() + claimed_ids = {claimed.run.id for claimed in claimed_runs} + for run in resumable_runs: + if run.id in claimed_ids: + continue + claimed = claim_existing_automation_run(run, owner=owner) + if claimed is not None: + claimed_runs.append(claimed) + return claimed_runs + + +def ensure_automation_run_job_intents( + run_id: int, job_specs: list[dict[str, Any]] +) -> list[AutomationRunJob]: + """Persist immutable logical job intents before any Redis enqueue.""" + run = db.session.get(AutomationRun, run_id) + if run is None: + raise ValueError(f"Automation run {run_id} does not exist.") + existing_intents = {intent.logical_job_key: intent for intent in run.job_intents} + if existing_intents: + return [existing_intents[spec["logical_job_key"]] for spec in job_specs] + + run.plan = { + **dict(run.plan or {}), + "jobs": [_json_safe(spec) for spec in job_specs], + } + intents = [] + for spec in job_specs: + intent = AutomationRunJob( + run=run, + logical_job_key=spec["logical_job_key"], + rq_job_id=spec["rq_job_id"], + queue=spec.get("queue", "forecasting"), + kind=spec["kind"], + status="pending", + depends_on=list(spec.get("depends_on", [])), + payload=_json_safe(spec.get("payload", {})), + ) + db.session.add(intent) + intents.append(intent) + db.session.commit() + return intents + + +def mark_automation_job_queued( + run_id: int, logical_job_key: str, rq_job_id: str +) -> None: + """Mark one logical job intent as queued in Redis.""" + now = _now_utc() + intent = db.session.scalars( + select(AutomationRunJob).filter_by( + run_id=run_id, logical_job_key=logical_job_key + ) + ).one() + intent.status = "queued" + intent.rq_job_id = rq_job_id + intent.enqueued_at = intent.enqueued_at or now + run = intent.run + run.first_enqueued_at = run.first_enqueued_at or now + queued_count = run.queued_job_count + run.dispatch_state = ( + "queued" if queued_count == run.intended_job_count else "partially_queued" + ) + db.session.commit() + + +def mark_automation_run_dispatch_queued( + run_id: int, attempt: AutomationRunAttempt | None = None +) -> None: + """Mark an automation run as fully queued and release its dispatch claim.""" + now = _now_utc() + run = db.session.get(AutomationRun, run_id) + if run is None: + raise ValueError(f"Automation run {run_id} does not exist.") + run.dispatch_state = "queued" + run.dispatch_completed_at = now + run.claim_owner = None + run.claim_expires_at = None + _finish_attempt(attempt, "queued", run.queued_job_count) + db.session.commit() + + +def mark_automation_run_dispatch_failed( + run_id: int, + attempt: AutomationRunAttempt | None, + error: BaseException, +) -> None: + """Record a failed dispatch attempt and release the claim, so the run stays retryable. + + The failure may have come from the database itself, so roll back first to get a usable session, + then re-read the run and the attempt through it. + """ + db.session.rollback() + run = db.session.get(AutomationRun, run_id) + if run is None: + raise ValueError(f"Automation run {run_id} does not exist.") + if attempt is not None: + attempt = db.session.get(AutomationRunAttempt, attempt.id) + queued_count = run.queued_job_count + run.dispatch_state = "partially_queued" if queued_count else "failed" + run.last_error_type = error.__class__.__name__ + run.last_error_message = str(error) + # Hand the occurrence back rather than making the next runner wait out this attempt's lease. + run.claim_owner = None + run.claim_expires_at = None + _finish_attempt(attempt, run.dispatch_state, queued_count, error) + db.session.commit() + + +def record_automation_job_started( + run_id: int | None, logical_job_key: str | None +) -> None: + """Record that a worker started an automation-created job.""" + if run_id is None or logical_job_key is None: + return + now = _now_utc() + intent = db.session.scalars( + select(AutomationRunJob).filter_by( + run_id=run_id, logical_job_key=logical_job_key + ) + ).one_or_none() + if intent is None: + return + intent.status = "running" + intent.started_at = intent.started_at or now + if intent.run.execution_state != "failed": + intent.run.execution_state = "running" + intent.run.execution_started_at = intent.run.execution_started_at or now + db.session.commit() + + +def _refresh_run_execution_state(run: AutomationRun, now: datetime) -> None: + """Derive a run's execution state from the state of all the jobs it created. + + A failed job keeps the whole run failed: a later job succeeding, as the wrap-up job does whatever became of the + cycle jobs it reports on, must not put the run back to 'running' and bury the failure. + """ + statuses = [job.status for job in run.job_intents] + finished = all(status in ("succeeded", "failed", "canceled") for status in statuses) + if "failed" in statuses: + run.execution_state = "failed" + elif finished and all(status == "succeeded" for status in statuses): + run.execution_state = "succeeded" + else: + run.execution_state = "running" + if finished or run.execution_state == "failed": + run.execution_completed_at = run.execution_completed_at or now + + +def record_automation_job_succeeded( + run_id: int | None, logical_job_key: str | None +) -> None: + """Record that a worker finished an automation-created job successfully.""" + if run_id is None or logical_job_key is None: + return + now = _now_utc() + intent = db.session.scalars( + select(AutomationRunJob).filter_by( + run_id=run_id, logical_job_key=logical_job_key + ) + ).one_or_none() + if intent is None: + return + intent.status = "succeeded" + intent.finished_at = now + _refresh_run_execution_state(intent.run, now) + db.session.commit() + + +def record_automation_job_failed( + run_id: int | None, + logical_job_key: str | None, + error: BaseException, +) -> None: + """Record that a worker failed an automation-created job. + + The job may well have failed on the database itself, which leaves the session in an aborted transaction where + every further statement is refused. Roll back first, so that the failure is still recorded. The job's own + uncommitted work is lost either way, since it is failing. + """ + if run_id is None or logical_job_key is None: + return + db.session.rollback() + now = _now_utc() + intent = db.session.scalars( + select(AutomationRunJob).filter_by( + run_id=run_id, logical_job_key=logical_job_key + ) + ).one_or_none() + if intent is None: + return + intent.status = "failed" + intent.finished_at = now + intent.last_error_type = error.__class__.__name__ + intent.last_error_message = str(error) + run = intent.run + _refresh_run_execution_state(run, now) + run.last_error_type = error.__class__.__name__ + run.last_error_message = str(error) + db.session.commit() + + +def reconcile_automation_job_intent(intent: AutomationRunJob) -> bool: + """Return whether Redis already has the deterministic job for an intent.""" + connection = current_app.queues[intent.queue].connection + if Job.exists(intent.rq_job_id, connection=connection): + if intent.status == "pending": + mark_automation_job_queued( + intent.run_id, intent.logical_job_key, intent.rq_job_id + ) + return True + return False + + # Fields naming a sensor that a scheduler records its results on, rather than reads from. # A scheduler hands its results to `make_schedule` as (sensor, data) pairs, and these are the fields that decide which sensors those are: # besides the power sensor of each device in the flex-model, its state of charge and its consumption and production sensors, @@ -1392,6 +1853,9 @@ def update_automation( reset_cursor = True automation.active = active if reset_cursor: + # A new revision keeps the durable runs of the old and the new schedule apart, + # even where they fall on the same scheduled UTC time. + automation.schedule_revision += 1 automation.cursor = get_initial_cursor() if changes: AssetAuditLog.add_record( @@ -1412,6 +1876,142 @@ def delete_automation(automation: Automation, origin: str = "API"): db.session.delete(automation) +def serialize_automation_run(run: AutomationRun) -> dict[str, Any]: + """Return operator-facing durable status for one automation run.""" + latest_attempt = run.attempts[-1] if run.attempts else None + return { + "id": run.id, + "scheduled-at": run.scheduled_at.isoformat(), + "schedule-revision": run.schedule_revision, + "dispatch-state": run.dispatch_state, + "execution-state": run.execution_state, + "attempt-count": run.attempt_count, + "intended-job-count": run.intended_job_count, + "queued-job-count": run.queued_job_count, + "first-enqueued-at": ( + run.first_enqueued_at.isoformat() if run.first_enqueued_at else None + ), + "dispatch-completed-at": ( + run.dispatch_completed_at.isoformat() if run.dispatch_completed_at else None + ), + "execution-completed-at": ( + run.execution_completed_at.isoformat() + if run.execution_completed_at + else None + ), + "claim-owner": run.claim_owner, + "claim-expires-at": ( + run.claim_expires_at.isoformat() if run.claim_expires_at else None + ), + "last-error": ( + { + "type": run.last_error_type, + "message": run.last_error_message, + } + if run.last_error_type or run.last_error_message + else None + ), + "latest-attempt": ( + { + "attempt-no": latest_attempt.attempt_no, + "owner": latest_attempt.owner, + "started-at": latest_attempt.started_at.isoformat(), + "finished-at": ( + latest_attempt.finished_at.isoformat() + if latest_attempt.finished_at + else None + ), + "outcome": latest_attempt.outcome, + "queued-job-count": latest_attempt.queued_job_count, + "error": ( + { + "type": latest_attempt.error_type, + "message": latest_attempt.error_message, + } + if latest_attempt.error_type or latest_attempt.error_message + else None + ), + } + if latest_attempt is not None + else None + ), + "jobs": [ + { + "logical-job-key": intent.logical_job_key, + "rq-job-id": intent.rq_job_id, + "queue": intent.queue, + "kind": intent.kind, + "status": intent.status, + "depends-on": list(intent.depends_on or []), + "enqueued-at": ( + intent.enqueued_at.isoformat() if intent.enqueued_at else None + ), + "started-at": ( + intent.started_at.isoformat() if intent.started_at else None + ), + "finished-at": ( + intent.finished_at.isoformat() if intent.finished_at else None + ), + "last-error": ( + { + "type": intent.last_error_type, + "message": intent.last_error_message, + } + if intent.last_error_type or intent.last_error_message + else None + ), + } + for intent in run.job_intents + ], + } + + +def _count_automation_runs_per_state( + automation_id: int, state_column +) -> dict[str, int]: + """Count an automation's runs per value of one state column, in the database.""" + rows = db.session.execute( + select(state_column, func.count()) + .where(AutomationRun.automation_id == automation_id) + .group_by(state_column) + ).all() + return {state: count for state, count in rows} + + +def get_automation_run_stats(automation: Automation) -> dict[str, Any]: + """Summarize durable automation runs for API and UI status displays. + + An automation keeps a run record per scheduled run, so its history grows without bound, + while this summary only ever shows counts and the most recent few. + Count in the database and read only those few in full, rather than loading a year of runs to render a panel. + """ + dispatch_counts = _count_automation_runs_per_state( + automation.id, AutomationRun.dispatch_state + ) + execution_counts = _count_automation_runs_per_state( + automation.id, AutomationRun.execution_state + ) + recent_runs = db.session.scalars( + select(AutomationRun) + .where(AutomationRun.automation_id == automation.id) + .order_by(AutomationRun.scheduled_at.desc(), AutomationRun.id.desc()) + .limit(AUTOMATION_RUN_STATS_RECENT_LIMIT) + .options( + # The serialization reads both of these for every run, so fetch them in one query each, not per run. + selectinload(AutomationRun.attempts), + selectinload(AutomationRun.job_intents), + ) + ).all() + serialized_runs = [serialize_automation_run(run) for run in recent_runs] + return { + "total": sum(dispatch_counts.values()), + "dispatch": dispatch_counts, + "execution": execution_counts, + "latest-run": serialized_runs[0] if serialized_runs else None, + "recent-runs": serialized_runs, + } + + def get_forecast_output_sensor(parameters: dict[str, Any]) -> Sensor: """Resolve the sensor on which a forecast automation registers beliefs.""" sensor_reference = parameters.get("sensor-to-save") @@ -1445,28 +2045,59 @@ def validate_automation_output_scope( ) +def dispatch_automation_run( + claimed_run: ClaimedAutomationRun, +) -> dict[str, Any]: + """Dispatch an already claimed automation run and record its attempt outcome.""" + run = claimed_run.run + try: + returns = run_automation(run.automation, automation_run=run) + except Exception as exc: + mark_automation_run_dispatch_failed(run.id, claimed_run.attempt, exc) + raise + mark_automation_run_dispatch_queued(run.id, claimed_run.attempt) + return { + "run_id": run.id, + "job_id": returns.get("job_id") if returns else None, + "n_jobs": returns.get("n_jobs") if returns else 0, + "dispatch_state": "queued", + } + + def run_automation( - automation: Automation, scheduled_at: datetime | None = None + automation: Automation, + automation_run: AutomationRun | None = None, + scheduled_at: datetime | None = None, ) -> dict[str, Any] | None: """Queue the jobs for one run of an automation. + A durable run (see `dispatch_automation_run`) carries the time it was scheduled for; a run on demand has none. + :returns: a dict like {"job_id": , "n_jobs": }. """ + if scheduled_at is None and automation_run is not None: + scheduled_at = automation_run.scheduled_at if automation.type == "forecasting": - return _run_forecast_automation(automation, scheduled_at=scheduled_at) + return _run_forecast_automation( + automation, automation_run, scheduled_at=scheduled_at + ) elif automation.type == "scheduling": - return _run_schedule_automation(automation, scheduled_at=scheduled_at) + return _run_schedule_automation( + automation, automation_run, scheduled_at=scheduled_at + ) elif automation.type == "reporting": # The reporting job records how far the reports reach once it succeeds (see run_report_job), # so a failed job leaves no gap for the next run to skip over. - return _run_report_automation(automation, scheduled_at=scheduled_at) + return _run_report_automation(automation, automation_run, scheduled_at) raise NotImplementedError( f"Automations of type '{automation.type}' cannot be run yet." ) def _run_forecast_automation( - automation: Automation, scheduled_at: datetime | None = None + automation: Automation, + automation_run: AutomationRun | None = None, + scheduled_at: datetime | None = None, ) -> dict[str, Any] | None: if automation.generator is None: raise ValueError( @@ -1479,17 +2110,26 @@ def _run_forecast_automation( raise ValueError( f"Data source {automation.generator_id} of automation {automation.id} does not store a Forecaster." ) - output_sensor = get_forecast_output_sensor(automation.parameters or {}) + parameters = ( + dict(automation_run.parameters) + if automation_run is not None + else dict(automation.parameters) + ) + output_sensor = get_forecast_output_sensor(parameters) validate_automation_output_scope( automation.asset_id, output_sensor, automation.type ) # Wipe any parameter state the copy inherited from a previous run. forecaster._parameters = None - forecaster.set_job_trigger("automation", automation_id=automation.id) + forecaster.set_job_trigger( + "automation", + automation_id=automation.id, + automation_run_id=automation_run.id if automation_run is not None else None, + ) return forecaster.compute( as_job=True, parameters=resolve_automation_window( - dict(automation.parameters), + parameters, automation.type, automation.timezone, scheduled_at, @@ -1498,7 +2138,9 @@ def _run_forecast_automation( def _run_report_automation( - automation: Automation, scheduled_at: datetime | None = None + automation: Automation, + automation_run: AutomationRun | None = None, + scheduled_at: datetime | None = None, ) -> dict[str, Any] | None: if automation.generator is None: raise ValueError( @@ -1509,8 +2151,13 @@ def _run_report_automation( raise ValueError( f"Data source {automation.generator_id} of automation {automation.id} does not store a Reporter." ) + # A retried run reports with the parameters it was planned with, rather than whatever the automation says now. parameters = prepare_report_parameters( - dict(automation.parameters), + ( + dict(automation_run.parameters) + if automation_run is not None + else dict(automation.parameters) + ), automation.cronstr, automation.timezone, automation_id=automation.id, @@ -1537,12 +2184,18 @@ def _run_report_automation( # The data generator instance is cached on the data source, which may be shared by several automations, # so wipe any parameter state from a previous run. reporter._parameters = None - reporter.set_job_trigger("automation", automation_id=automation.id) + reporter.set_job_trigger( + "automation", + automation_id=automation.id, + automation_run_id=automation_run.id if automation_run is not None else None, + ) return reporter.compute(as_job=True, parameters=parameters) def _run_schedule_automation( - automation: Automation, scheduled_at: datetime | None = None + automation: Automation, + automation_run: AutomationRun | None = None, + scheduled_at: datetime | None = None, ) -> dict[str, Any]: from flexmeasures.data.schemas.scheduling import AssetTriggerSchema from flexmeasures.data.services.scheduling import ( @@ -1550,20 +2203,23 @@ def _run_schedule_automation( create_simultaneous_scheduling_job, ) + # A retried run re-queues the jobs it was planned with, rather than whatever the automation says now. + parameters = ( + dict(automation_run.parameters) + if automation_run is not None + else dict(automation.parameters) + ) # The scheduler and the flex config it merges in can both change between runs, # so record which data source this run actually computes under. generator = resolve_schedule_generator( - automation.asset_id, automation.parameters, automation.timezone, scheduled_at + automation.asset_id, parameters, automation.timezone, scheduled_at ) if automation.generator_id != generator.id: automation.generator_id = generator.id db.session.commit() message = prepare_schedule_trigger_message( - dict(automation.parameters), - automation.asset_id, - automation.timezone, - scheduled_at, + parameters, automation.asset_id, automation.timezone, scheduled_at ) trigger_data = AssetTriggerSchema().load(message) start = trigger_data["start_of_schedule"] @@ -1576,6 +2232,9 @@ def _run_schedule_automation( ) if trigger_data.get("resolution") is not None: scheduler_kwargs["resolution"] = trigger_data["resolution"] + trigger = {"origin": "automation", "automation_id": automation.id} + if automation_run is not None: + trigger["automation_run_id"] = automation_run.id if trigger_data["sequential"]: f = create_sequential_scheduling_job else: @@ -1584,7 +2243,7 @@ def _run_schedule_automation( asset=trigger_data["asset"], enqueue=True, force_new_job_creation=trigger_data.get("force_new_job_creation", False), - trigger={"origin": "automation", "automation_id": automation.id}, + trigger=trigger, **scheduler_kwargs, ) n_jobs = len(job.args[0]) + 1 if trigger_data["sequential"] else 1 diff --git a/flexmeasures/data/services/forecasting.py b/flexmeasures/data/services/forecasting.py index 212bed42b4..ea663e029e 100644 --- a/flexmeasures/data/services/forecasting.py +++ b/flexmeasures/data/services/forecasting.py @@ -53,3 +53,13 @@ def handle_forecasting_exception(job, exc_type, exc_value, traceback): job.meta["exception"] = exception job.save_meta() + + trigger = job.meta.get("trigger", {}) + automation_run_id = job.meta.get("automation_run_id") or trigger.get( + "automation_run_id" + ) + logical_job_key = job.meta.get("logical_job_key") + if automation_run_id is not None and logical_job_key is not None: + from flexmeasures.data.services.automations import record_automation_job_failed + + record_automation_job_failed(automation_run_id, logical_job_key, exc_value) diff --git a/flexmeasures/data/tests/test_automation_runs_fresh_db.py b/flexmeasures/data/tests/test_automation_runs_fresh_db.py new file mode 100644 index 0000000000..05fce6cc70 --- /dev/null +++ b/flexmeasures/data/tests/test_automation_runs_fresh_db.py @@ -0,0 +1,855 @@ +"""Regression tests for durable automation run dispatch.""" + +from __future__ import annotations + +from datetime import datetime, timedelta, timezone + +import pytest +from sqlalchemy import event, select, text +from sqlalchemy.engine import Engine +from sqlalchemy.exc import DatabaseError, IntegrityError + +from flexmeasures.cli.tests.utils import to_flags +from flexmeasures.data.models.automations import ( + Automation, + AutomationRun, + AutomationRunAttempt, + AutomationRunJob, +) + + +@pytest.fixture(scope="function") +def clean_redis(app): + app.redis_connection.flushdb() + yield + app.redis_connection.flushdb() + + +@pytest.fixture() +def due_forecast_automation( + app, fresh_db, setup_fresh_test_forecast_data, freeze_server_now +): + """Create a persisted forecast automation due at the frozen minute.""" + from flexmeasures.cli.data_add import add_automation + + freeze_server_now(datetime(2026, 8, 5, 0, 58, tzinfo=timezone.utc)) + sensor = setup_fresh_test_forecast_data["solar-sensor"] + runner = app.test_cli_runner() + result = runner.invoke( + add_automation, + to_flags( + { + "asset": sensor.generic_asset_id, + "name": "Durable forecasts", + "cron": "0 1 * * *", + "timezone": "UTC", + "sensor": sensor.id, + # A duration alone starts the forecast at the time of each run, which the frozen clock puts at 01:00. + "duration": "PT2H", + "forecast-frequency": "PT1H", + "max-forecast-horizon": "PT2H", + "retrain-frequency": "PT1H", + } + ), + ) + assert result.exit_code == 0, result.output + automation = fresh_db.session.scalars(select(Automation)).one() + freeze_server_now(datetime(2026, 8, 5, 1, 0, tzinfo=timezone.utc)) + return automation + + +def test_automation_run_unique_per_revision(fresh_db, due_forecast_automation): + """A scheduled occurrence is unique for one automation revision.""" + scheduled_at = datetime(2026, 8, 5, 1, 0, tzinfo=timezone.utc) + run = AutomationRun( + automation=due_forecast_automation, + scheduled_at=scheduled_at, + schedule_revision=due_forecast_automation.schedule_revision, + automation_type="forecasting", + generator_id=due_forecast_automation.generator_id, + dispatch_state="pending", + execution_state="pending", + parameters=dict(due_forecast_automation.parameters), + plan={}, + ) + fresh_db.session.add(run) + fresh_db.session.commit() + + duplicate = AutomationRun( + automation=due_forecast_automation, + scheduled_at=scheduled_at, + schedule_revision=due_forecast_automation.schedule_revision, + automation_type="forecasting", + generator_id=due_forecast_automation.generator_id, + dispatch_state="pending", + execution_state="pending", + parameters=dict(due_forecast_automation.parameters), + plan={}, + ) + fresh_db.session.add(duplicate) + + with pytest.raises(IntegrityError): + fresh_db.session.commit() + fresh_db.session.rollback() + + same_occurrence_new_revision = AutomationRun( + automation=due_forecast_automation, + scheduled_at=scheduled_at, + schedule_revision=due_forecast_automation.schedule_revision + 1, + automation_type="forecasting", + generator_id=due_forecast_automation.generator_id, + dispatch_state="pending", + execution_state="pending", + parameters=dict(due_forecast_automation.parameters), + plan={}, + ) + fresh_db.session.add(same_occurrence_new_revision) + fresh_db.session.commit() + + +def test_job_intents_are_unique_per_run(fresh_db, due_forecast_automation): + """The database rejects duplicate logical job keys for one run.""" + run = AutomationRun( + automation=due_forecast_automation, + scheduled_at=datetime(2026, 8, 5, 1, 0, tzinfo=timezone.utc), + schedule_revision=due_forecast_automation.schedule_revision, + automation_type="forecasting", + generator_id=due_forecast_automation.generator_id, + dispatch_state="pending", + execution_state="pending", + parameters=dict(due_forecast_automation.parameters), + plan={}, + ) + fresh_db.session.add(run) + fresh_db.session.flush() + fresh_db.session.add_all( + [ + AutomationRunJob( + run=run, + logical_job_key="cycle-001", + rq_job_id="automation-run-test-cycle-001", + queue="forecasting", + kind="forecast-cycle", + status="pending", + depends_on=[], + payload={}, + ), + AutomationRunJob( + run=run, + logical_job_key="cycle-001", + rq_job_id="automation-run-test-cycle-duplicate", + queue="forecasting", + kind="forecast-cycle", + status="pending", + depends_on=[], + payload={}, + ), + ] + ) + + with pytest.raises(IntegrityError): + fresh_db.session.commit() + + +def test_failed_before_first_enqueue_can_be_retried( + app, fresh_db, clean_redis, due_forecast_automation, mocker +): + """A pre-enqueue failure leaves no Redis job and a retryable durable run.""" + from flexmeasures.cli.jobs import run_automations + + queue = app.queues["forecasting"] + original_enqueue_job = queue.enqueue_job + patched_enqueue_job = mocker.patch.object( + queue, "enqueue_job", side_effect=RuntimeError("redis unavailable") + ) + runner = app.test_cli_runner() + + first_result = runner.invoke(run_automations) + + assert first_result.exit_code == 1, first_result.output + assert queue.count == 0 + run = fresh_db.session.scalars(select(AutomationRun)).one() + assert run.dispatch_state == "failed" + assert run.queued_job_count == 0 + assert run.attempt_count == 1 + + patched_enqueue_job.side_effect = lambda job: original_enqueue_job(job) + retry_result = runner.invoke(run_automations) + + assert retry_result.exit_code == 0, retry_result.output + fresh_db.session.refresh(run) + assert run.dispatch_state == "queued" + assert run.attempt_count == 2 + assert run.queued_job_count == run.intended_job_count + assert queue.count > 0 + + +def test_partial_enqueue_retry_queues_only_missing_jobs( + app, fresh_db, clean_redis, due_forecast_automation, mocker +): + """A partial dispatch retry keeps queued job IDs and only enqueues missing intents.""" + from flexmeasures.cli.jobs import run_automations + + queue = app.queues["forecasting"] + original_enqueue_job = queue.enqueue_job + calls = [] + + def enqueue_once_then_fail(job): + calls.append(job.id) + if len(calls) == 1: + return original_enqueue_job(job) + raise RuntimeError("lost connection after first job") + + patched_enqueue_job = mocker.patch.object( + queue, "enqueue_job", side_effect=enqueue_once_then_fail + ) + runner = app.test_cli_runner() + + first_result = runner.invoke(run_automations) + + assert first_result.exit_code == 1, first_result.output + run = fresh_db.session.scalars(select(AutomationRun)).one() + first_job_ids = [intent.rq_job_id for intent in run.job_intents] + assert run.dispatch_state == "partially_queued" + assert run.queued_job_count == 1 + + patched_enqueue_job.side_effect = lambda job: original_enqueue_job(job) + retry_result = runner.invoke(run_automations) + + assert retry_result.exit_code == 0, retry_result.output + fresh_db.session.refresh(run) + assert [intent.rq_job_id for intent in run.job_intents] == first_job_ids + assert run.dispatch_state == "queued" + assert run.queued_job_count == run.intended_job_count + assert queue.fetch_job(first_job_ids[0]) is not None + + +def test_stale_claim_is_adopted_after_restart( + fresh_db, due_forecast_automation, freeze_server_now +): + """A stale SQL claim is reused by a later runner without creating a new run.""" + from flexmeasures.data.services.automations import ( + claim_existing_automation_run, + get_due_automations, + claim_due_automation_run, + ) + + due = get_due_automations(datetime(2026, 8, 5, 1, 0, tzinfo=timezone.utc))[0] + claimed = claim_due_automation_run(due, owner="first-runner") + assert claimed is not None + run_id = claimed.run.id + + claimed.run.claim_expires_at = datetime(2026, 8, 5, 1, 4, tzinfo=timezone.utc) + fresh_db.session.commit() + freeze_server_now(datetime(2026, 8, 5, 1, 5, tzinfo=timezone.utc)) + fresh_db.session.remove() + + run = fresh_db.session.get(AutomationRun, run_id) + adopted = claim_existing_automation_run(run, owner="second-runner") + + assert adopted is not None + assert adopted.run.id == run_id + assert adopted.run.claim_owner == "second-runner" + assert adopted.run.attempt_count == 2 + + +def test_fresh_claim_blocks_second_runner(fresh_db, due_forecast_automation): + """A non-stale SQL claim cannot be adopted by another runner.""" + from flexmeasures.data.services.automations import ( + claim_existing_automation_run, + get_due_automations, + claim_due_automation_run, + ) + + due = get_due_automations(datetime(2026, 8, 5, 1, 0, tzinfo=timezone.utc))[0] + claimed = claim_due_automation_run(due, owner="first-runner") + assert claimed is not None + + adopted = claim_existing_automation_run(claimed.run, owner="second-runner") + + assert adopted is None + fresh_db.session.refresh(claimed.run) + assert claimed.run.claim_owner == "first-runner" + + +def test_run_plan_snapshot_is_immutable_after_automation_edit( + app, fresh_db, clean_redis, due_forecast_automation +): + """Retries use the original run parameters even after automation edits.""" + from flexmeasures.cli.jobs import run_automations + + runner = app.test_cli_runner() + first_result = runner.invoke(run_automations) + assert first_result.exit_code == 0, first_result.output + run = fresh_db.session.scalars(select(AutomationRun)).one() + original_parameters = dict(run.parameters) + original_revision = run.schedule_revision + + due_forecast_automation.parameters["duration"] = "PT4H" + due_forecast_automation.cronstr = "30 1 * * *" + due_forecast_automation.schedule_revision += 1 + fresh_db.session.commit() + fresh_db.session.refresh(run) + + assert run.parameters == original_parameters + assert run.schedule_revision == original_revision + assert run.plan["cronstr"] == "0 1 * * *" + + +def test_live_partial_dispatch_claim_is_not_stolen(fresh_db, due_forecast_automation): + """A runner which is still queueing keeps its claim, even while partially queued. + + The dispatch state turns to 'partially_queued' as soon as the first job is queued, so a second runner must fall back on the claim lease to decide whether the first runner is gone. + """ + from flexmeasures.data.services.automations import ( + claim_due_automation_run, + claim_existing_automation_run, + get_due_automations, + ) + + due = get_due_automations(datetime(2026, 8, 5, 1, 0, tzinfo=timezone.utc))[0] + claimed = claim_due_automation_run(due, owner="first-runner") + assert claimed is not None + # The first runner queued one of its jobs and is still working on the rest. + claimed.run.dispatch_state = "partially_queued" + fresh_db.session.commit() + + adopted = claim_existing_automation_run(claimed.run, owner="second-runner") + + assert adopted is None + fresh_db.session.refresh(claimed.run) + assert claimed.run.claim_owner == "first-runner" + assert claimed.run.attempt_count == 1 + + +def test_crash_between_last_enqueue_and_dispatch_completion_is_finalized( + app, fresh_db, clean_redis, due_forecast_automation, mocker, freeze_server_now +): + """A crash after the last enqueue, but before dispatch is marked complete, still gets finalized. + + All jobs are already in Redis, so a later runner must adopt the abandoned claim, reconcile the durable + intents against Redis, and complete the dispatch without queueing anything again. + """ + from flexmeasures.cli.jobs import run_automations + from flexmeasures.data.services import automations as automations_service + + real_mark_queued = automations_service.mark_automation_run_dispatch_queued + marks: list[int] = [] + + def crash_on_first_completion(run_id, attempt=None): + marks.append(run_id) + if len(marks) == 1: + raise RuntimeError("died before recording dispatch completion") + return real_mark_queued(run_id, attempt) + + mocker.patch( + "flexmeasures.data.services.automations.mark_automation_run_dispatch_queued", + side_effect=crash_on_first_completion, + ) + runner = app.test_cli_runner() + + first_result = runner.invoke(run_automations) + + assert first_result.exit_code == 1, first_result.output + run = fresh_db.session.scalars(select(AutomationRun)).one() + queued_job_ids = [intent.rq_job_id for intent in run.job_intents] + assert run.queued_job_count == run.intended_job_count + assert run.dispatch_completed_at is None + queue = app.queues["forecasting"] + jobs_after_crash = queue.count + + # The abandoned claim only becomes adoptable once its lease has expired. + freeze_server_now(datetime(2026, 8, 5, 1, 30, tzinfo=timezone.utc)) + fresh_db.session.remove() + + retry_result = runner.invoke(run_automations) + + assert retry_result.exit_code == 0, retry_result.output + run = fresh_db.session.scalars(select(AutomationRun)).one() + assert run.dispatch_state == "queued" + assert run.dispatch_completed_at is not None + assert run.claim_owner is None + # Reconciliation recognised the existing Redis jobs, so nothing was queued twice. + assert [intent.rq_job_id for intent in run.job_intents] == queued_job_ids + assert queue.count == jobs_after_crash + + +def test_death_after_claim_is_recovered_once_the_lease_expires( + app, fresh_db, clean_redis, due_forecast_automation, freeze_server_now +): + """A runner which dies right after claiming an occurrence queues nothing and blocks nothing. + + The occurrence stays claimed until the lease runs out, after which a later runner adopts the same durable run + instead of creating a second one. + """ + from flexmeasures.cli.jobs import run_automations + from flexmeasures.data.services.automations import ( + claim_due_automation_run, + get_due_automations, + ) + + due = get_due_automations(datetime(2026, 8, 5, 1, 0, tzinfo=timezone.utc))[0] + claimed = claim_due_automation_run(due, owner="runner-that-dies") + assert claimed is not None + run_id = claimed.run.id + queue = app.queues["forecasting"] + assert claimed.run.job_intents == [] + assert queue.count == 0 + + # While the lease is live, nothing else touches the occurrence. + runner = app.test_cli_runner() + blocked_result = runner.invoke(run_automations) + assert blocked_result.exit_code == 0, blocked_result.output + assert queue.count == 0 + + freeze_server_now(datetime(2026, 8, 5, 1, 30, tzinfo=timezone.utc)) + fresh_db.session.remove() + + recovered_result = runner.invoke(run_automations) + + assert recovered_result.exit_code == 0, recovered_result.output + run = fresh_db.session.scalars(select(AutomationRun)).one() + assert run.id == run_id + assert run.dispatch_state == "queued" + assert run.attempt_count == 2 + assert run.queued_job_count == run.intended_job_count + # The abandoned attempt is still visible as unfinished, which is how an operator spots a dead runner. + assert [(a.attempt_no, a.owner, a.outcome) for a in run.attempts] == [ + (1, "runner-that-dies", None), + (2, run.attempts[1].owner, "queued"), + ] + + +def test_two_independent_sessions_claim_one_occurrence( + app, fresh_db, clean_redis, due_forecast_automation +): + """Exactly one of two concurrent runner sessions wins the same occurrence.""" + import threading + + from flexmeasures.data import db + from flexmeasures.data.services.automations import ( + claim_due_automation_run, + get_due_automations, + ) + + now = datetime(2026, 8, 5, 1, 0, tzinfo=timezone.utc) + start_together = threading.Barrier(2, timeout=30) + outcomes: dict[str, int | None] = {} + lock = threading.Lock() + + def claim_as(owner: str) -> None: + # Each thread gets its own app context, and with it its own database session. + with app.app_context(): + try: + due_automations = get_due_automations(now) + start_together.wait() + claimed = ( + claim_due_automation_run(due_automations[0], owner=owner) + if due_automations + else None + ) + with lock: + outcomes[owner] = claimed.run.id if claimed is not None else None + finally: + db.session.remove() + + threads = [ + threading.Thread(target=claim_as, args=(owner,)) + for owner in ("runner-a", "runner-b") + ] + for thread in threads: + thread.start() + for thread in threads: + thread.join(timeout=60) + assert not thread.is_alive(), "a claiming thread did not finish" + + assert sorted(outcomes) == ["runner-a", "runner-b"] + winners = [owner for owner, run_id in outcomes.items() if run_id is not None] + assert len(winners) == 1, f"expected exactly one winner, got {outcomes}" + runs = fresh_db.session.scalars(select(AutomationRun)).all() + assert len(runs) == 1 + assert runs[0].id == outcomes[winners[0]] + assert runs[0].claim_owner == winners[0] + + +def test_completed_dispatch_is_not_redone_after_redis_is_flushed( + app, fresh_db, clean_redis, due_forecast_automation +): + """Losing the Redis jobs does not make a completed occurrence run a second time. + + The durable run record, not any Redis key, is what says the occurrence was already dispatched. + """ + from flexmeasures.cli.jobs import run_automations + + runner = app.test_cli_runner() + first_result = runner.invoke(run_automations) + assert first_result.exit_code == 0, first_result.output + run = fresh_db.session.scalars(select(AutomationRun)).one() + dispatch_completed_at = run.dispatch_completed_at + assert dispatch_completed_at is not None + + app.redis_connection.flushdb() + assert app.queues["forecasting"].count == 0 + + second_result = runner.invoke(run_automations) + + assert second_result.exit_code == 0, second_result.output + assert app.queues["forecasting"].count == 0 + runs = fresh_db.session.scalars(select(AutomationRun)).all() + assert len(runs) == 1 + fresh_db.session.refresh(run) + assert run.dispatch_completed_at == dispatch_completed_at + assert run.attempt_count == 1 + + +def test_multi_cycle_run_records_its_wrap_up_dependencies( + app, fresh_db, clean_redis, due_forecast_automation +): + """Every cycle job and the wrap-up job carry the run identity, and the wrap-up waits for the cycles.""" + from flexmeasures.cli.jobs import run_automations + + runner = app.test_cli_runner() + result = runner.invoke(run_automations) + assert result.exit_code == 0, result.output + run = fresh_db.session.scalars(select(AutomationRun)).one() + + cycle_intents = [i for i in run.job_intents if i.kind == "forecast-cycle"] + wrap_up_intents = [i for i in run.job_intents if i.kind == "forecast-wrap-up"] + assert len(cycle_intents) > 1, "this automation is expected to need several cycles" + assert len(wrap_up_intents) == 1 + wrap_up = wrap_up_intents[0] + assert wrap_up.depends_on == [i.logical_job_key for i in cycle_intents] + + queue = app.queues["forecasting"] + # The cycle jobs are ready to run, while the wrap-up job waits for them in the deferred registry. + assert sorted(queue.job_ids) == sorted(i.rq_job_id for i in cycle_intents) + assert list(queue.deferred_job_registry.get_job_ids()) == [wrap_up.rq_job_id] + wrap_up_job = queue.fetch_job(wrap_up.rq_job_id) + assert sorted(wrap_up_job._dependency_ids) == sorted( + i.rq_job_id for i in cycle_intents + ) + for intent in run.job_intents: + job = queue.fetch_job(intent.rq_job_id) + assert job is not None, f"{intent.logical_job_key} is not in Redis" + assert job.meta["automation_run_id"] == run.id + assert job.meta["logical_job_key"] == intent.logical_job_key + assert job.meta["trigger"]["automation_run_id"] == run.id + + +def test_worker_success_is_recorded_durably( + app, fresh_db, clean_redis, due_forecast_automation +): + """A job which a worker completes is marked succeeded on its durable intent.""" + from rq import SimpleWorker + + from flexmeasures.cli.jobs import run_automations + + runner = app.test_cli_runner() + assert runner.invoke(run_automations).exit_code == 0 + run = fresh_db.session.scalars(select(AutomationRun)).one() + queue = app.queues["forecasting"] + wrap_up = next(i for i in run.job_intents if i.kind == "forecast-wrap-up") + + worker = SimpleWorker([queue], connection=queue.connection) + worker.perform_job(queue.fetch_job(wrap_up.rq_job_id), queue) + + fresh_db.session.refresh(run) + fresh_db.session.refresh(wrap_up) + assert wrap_up.status == "succeeded" + assert wrap_up.started_at is not None + assert wrap_up.finished_at is not None + # The cycle jobs have not run yet, so the run as a whole is still in progress. + assert run.execution_state == "running" + assert run.execution_started_at is not None + assert run.execution_completed_at is None + + +def test_worker_failure_is_recorded_durably( + app, fresh_db, clean_redis, due_forecast_automation +): + """A job which a worker fails is marked failed on its durable intent, with the error kept for diagnosis.""" + from rq import SimpleWorker + + from flexmeasures.cli.jobs import run_automations + + runner = app.test_cli_runner() + assert runner.invoke(run_automations).exit_code == 0 + run = fresh_db.session.scalars(select(AutomationRun)).one() + queue = app.queues["forecasting"] + wrap_up = next(i for i in run.job_intents if i.kind == "forecast-wrap-up") + cycle = next(i for i in run.job_intents if i.kind == "forecast-cycle") + + # Make the wrap-up job fail by taking away one of the cycle jobs it reports on. + queue.fetch_job(cycle.rq_job_id).delete() + worker = SimpleWorker([queue], connection=queue.connection) + worker.perform_job(queue.fetch_job(wrap_up.rq_job_id), queue) + + fresh_db.session.refresh(run) + fresh_db.session.refresh(wrap_up) + assert wrap_up.status == "failed" + assert wrap_up.last_error_type is not None + assert run.execution_state == "failed" + assert run.execution_completed_at is not None + assert run.last_error_type == wrap_up.last_error_type + + +def test_run_history_survives_a_new_session( + app, fresh_db, clean_redis, due_forecast_automation +): + """Run, attempt and job records are readable again after the application session is thrown away.""" + from flexmeasures.cli.jobs import run_automations + + queue = app.queues["forecasting"] + original_enqueue_job = queue.enqueue_job + runner = app.test_cli_runner() + + def fail_before_queueing(job): + raise RuntimeError("redis unavailable") + + queue.enqueue_job = fail_before_queueing # type: ignore[method-assign] + try: + assert runner.invoke(run_automations).exit_code == 1 + finally: + queue.enqueue_job = original_enqueue_job # type: ignore[method-assign] + assert runner.invoke(run_automations).exit_code == 0 + + run_id = fresh_db.session.scalars(select(AutomationRun)).one().id + fresh_db.session.remove() + + run = fresh_db.session.get(AutomationRun, run_id) + assert run.attempt_count == 2 + assert run.dispatch_state == "queued" + assert [(a.attempt_no, a.outcome) for a in run.attempts] == [ + (1, "failed"), + (2, "queued"), + ] + assert run.attempts[0].error_type == "RuntimeError" + assert run.attempts[0].error_message == "redis unavailable" + assert run.queued_job_count == run.intended_job_count + assert all(intent.status == "queued" for intent in run.job_intents) + + +def test_job_failure_is_recorded_even_after_a_database_error( + app, fresh_db, clean_redis, due_forecast_automation +): + """A job which fails on a database error still gets its failure recorded. + + The failing statement leaves the session in an aborted transaction, in which every further statement is refused, + so recording the failure has to start by putting the session back in a usable state. + """ + from flexmeasures.cli.jobs import run_automations + from flexmeasures.data.services.automations import record_automation_job_failed + + runner = app.test_cli_runner() + assert runner.invoke(run_automations).exit_code == 0 + run = fresh_db.session.scalars(select(AutomationRun)).one() + run_id = run.id + logical_job_key = next( + i.logical_job_key for i in run.job_intents if i.kind == "forecast-cycle" + ) + + # Break the transaction the way a failing statement inside the job would. + with pytest.raises(DatabaseError): + fresh_db.session.execute(text("SELECT no_such_function_2393()")) + + record_automation_job_failed( + run_id, logical_job_key, RuntimeError("the job hit a database error") + ) + + fresh_db.session.remove() + run = fresh_db.session.scalars(select(AutomationRun)).one() + failed_intent = next( + i for i in run.job_intents if i.logical_job_key == logical_job_key + ) + assert failed_intent.status == "failed" + assert failed_intent.last_error_type == "RuntimeError" + assert failed_intent.last_error_message == "the job hit a database error" + assert run.execution_state == "failed" + assert run.execution_completed_at is not None + + +def test_a_later_success_does_not_hide_an_earlier_job_failure( + app, fresh_db, clean_redis, due_forecast_automation +): + """A run whose job failed stays failed, even when its remaining jobs go on to succeed. + + The wrap-up job succeeds whatever became of the cycle jobs it reports on, so it must not report the run as + merely still running and bury the failure an operator needs to see. + """ + from flexmeasures.cli.jobs import run_automations + from flexmeasures.data.services.automations import ( + record_automation_job_failed, + record_automation_job_succeeded, + ) + + runner = app.test_cli_runner() + assert runner.invoke(run_automations).exit_code == 0 + run = fresh_db.session.scalars(select(AutomationRun)).one() + cycles = [i for i in run.job_intents if i.kind == "forecast-cycle"] + wrap_up = next(i for i in run.job_intents if i.kind == "forecast-wrap-up") + + record_automation_job_failed( + run.id, cycles[0].logical_job_key, RuntimeError("the cycle blew up") + ) + for cycle in cycles[1:]: + record_automation_job_succeeded(run.id, cycle.logical_job_key) + record_automation_job_succeeded(run.id, wrap_up.logical_job_key) + + fresh_db.session.remove() + run = fresh_db.session.scalars(select(AutomationRun)).one() + assert run.execution_state == "failed" + assert run.execution_completed_at is not None + assert run.last_error_type == "RuntimeError" + assert sorted(i.status for i in run.job_intents) == sorted( + ["failed"] + ["succeeded"] * len(run.job_intents[1:]) + ) + + +def _add_finished_runs(db, automation: Automation, count: int) -> None: + """Give an automation a run history, each run with one attempt and three jobs.""" + first_scheduled_at = datetime(2026, 8, 5, 1, 0, tzinfo=timezone.utc) + for index in range(count): + run = AutomationRun( + automation=automation, + scheduled_at=first_scheduled_at + timedelta(minutes=index), + schedule_revision=automation.schedule_revision, + automation_type="forecasting", + generator_id=automation.generator_id, + dispatch_state="queued" if index % 2 else "failed", + execution_state="succeeded" if index % 2 else "pending", + attempt_count=1, + parameters=dict(automation.parameters), + plan={}, + ) + db.session.add(run) + db.session.flush() + db.session.add( + AutomationRunAttempt( + run=run, + attempt_no=1, + owner="runner:1", + outcome="queued", + queued_job_count=3, + ) + ) + for logical_job_key in ("cycle-001", "cycle-002", "wrap-up"): + db.session.add( + AutomationRunJob( + run=run, + logical_job_key=logical_job_key, + rq_job_id=f"automation-run-{run.id}-{logical_job_key}", + queue="forecasting", + kind="forecast-cycle", + status="succeeded", + depends_on=[], + payload={}, + ) + ) + db.session.commit() + + +def test_run_stats_do_not_load_the_whole_run_history(fresh_db, due_forecast_automation): + """The status summary counts runs in the database and reads only the most recent ones. + + An automation keeps one run record per scheduled run, so its history grows without bound, + and loading all of it to render a panel would get slower for the automations that run most often. + """ + from flexmeasures.data.services.automations import ( + AUTOMATION_RUN_STATS_RECENT_LIMIT, + get_automation_run_stats, + ) + + history_size = AUTOMATION_RUN_STATS_RECENT_LIMIT * 5 + _add_finished_runs(fresh_db, due_forecast_automation, history_size) + fresh_db.session.remove() + automation = fresh_db.session.scalars(select(Automation)).one() + + statements: list[str] = [] + + def record_statement(conn, cursor, statement, parameters, context, executemany): + statements.append(" ".join(statement.split())) + + event.listen(Engine, "before_cursor_execute", record_statement) + try: + stats = get_automation_run_stats(automation) + finally: + event.remove(Engine, "before_cursor_execute", record_statement) + + # The counts cover the whole history, even though it was never all loaded. + assert stats["total"] == history_size + assert stats["dispatch"] == { + "queued": history_size // 2, + "failed": history_size // 2, + } + assert stats["execution"] == { + "succeeded": history_size // 2, + "pending": history_size // 2, + } + assert len(stats["recent-runs"]) == AUTOMATION_RUN_STATS_RECENT_LIMIT + assert stats["latest-run"] == stats["recent-runs"][0] + # The most recent runs are the ones described, newest first. + scheduled_times = [run["scheduled-at"] for run in stats["recent-runs"]] + assert scheduled_times == sorted(scheduled_times, reverse=True) + + # Counting happens in the database, and the runs that are read are limited, + # so no query may select whole run rows without a limit on how many. + run_selects = [s for s in statements if "FROM automation_run " in s] + unbounded = [s for s in run_selects if "count(" not in s and "LIMIT" not in s] + assert not unbounded, f"a query reads the whole run history: {unbounded}" + # Two aggregates, the limited read of recent runs, and one eager load per child relationship. + assert len(statements) <= 6, f"{len(statements)} queries: {statements}" + + +def _add_partially_queued_run( + db, automation: Automation, automation_type: str, scheduled_at: datetime +) -> AutomationRun: + """Give an automation a run whose previous dispatch stopped halfway, with its claim released.""" + run = AutomationRun( + automation=automation, + scheduled_at=scheduled_at, + schedule_revision=automation.schedule_revision, + automation_type=automation_type, + generator_id=automation.generator_id, + dispatch_state="partially_queued", + execution_state="pending", + attempt_count=1, + parameters=dict(automation.parameters), + plan={}, + ) + db.session.add(run) + db.session.commit() + return run + + +def test_only_a_forecast_run_is_dispatched_a_second_time( + fresh_db, due_forecast_automation, freeze_server_now +): + """A forecast run resumes where it stopped, while a schedule run is left where it failed. + + A forecast run's jobs carry IDs derived from the run, so a retry can tell which of them it already queued, + whereas a schedule run's jobs get a fresh ID on every dispatch, so retrying one would duplicate its schedules. + """ + from flexmeasures.data.services.automations import ( + get_dispatchable_automation_runs, + ) + + freeze_server_now(datetime(2026, 8, 5, 2, 0, tzinfo=timezone.utc)) + forecast_run = _add_partially_queued_run( + fresh_db, + due_forecast_automation, + "forecasting", + datetime(2026, 8, 5, 1, 15, tzinfo=timezone.utc), + ) + schedule_run = _add_partially_queued_run( + fresh_db, + due_forecast_automation, + "scheduling", + datetime(2026, 8, 5, 1, 30, tzinfo=timezone.utc), + ) + + claimed_ids = { + claimed.run.id for claimed in get_dispatchable_automation_runs(owner="runner:1") + } + + assert forecast_run.id in claimed_ids + assert schedule_run.id not in claimed_ids diff --git a/flexmeasures/data/tests/test_automation_scheduling_fresh_db.py b/flexmeasures/data/tests/test_automation_scheduling_fresh_db.py index c5c78392c7..2db083bb5b 100644 --- a/flexmeasures/data/tests/test_automation_scheduling_fresh_db.py +++ b/flexmeasures/data/tests/test_automation_scheduling_fresh_db.py @@ -11,7 +11,7 @@ from flexmeasures.data.models.data_sources import DataSource from flexmeasures.data.models.generic_assets import GenericAsset, GenericAssetType from flexmeasures.data.services.automations import ( - claim_due_automation, + claim_due_automation_run, get_due_automations, ) @@ -77,7 +77,7 @@ def test_automations_use_independent_timezones(fresh_db, automation_factory): assert [(item.automation.id, item.scheduled_at) for item in due] == [ (amsterdam.id, datetime(2026, 1, 15, 6, 0, tzinfo=timezone.utc)) ] - assert claim_due_automation(due[0]) is True + assert claim_due_automation_run(due[0]) is not None due = get_due_automations(datetime(2026, 1, 15, 12, 0, tzinfo=timezone.utc)) @@ -133,7 +133,7 @@ def test_spring_forward_run_happens_at_transition_boundary( assert [(item.automation.id, item.scheduled_at) for item in due] == [ (automation.id, datetime(2026, 3, 29, 1, 0, tzinfo=timezone.utc)) ] - assert claim_due_automation(due[0]) is True + assert claim_due_automation_run(due[0]) is not None assert get_due_automations(datetime(2026, 3, 29, 1, 1, tzinfo=timezone.utc)) == [] @@ -151,7 +151,7 @@ def test_fall_back_wall_time_runs_only_once(fresh_db, automation_factory): assert [(item.automation.id, item.scheduled_at) for item in first_fold_due] == [ (automation.id, datetime(2026, 10, 25, 0, 30, tzinfo=timezone.utc)) ] - assert claim_due_automation(first_fold_due[0]) is True + assert claim_due_automation_run(first_fold_due[0]) is not None fresh_db.session.remove() second_fold_due = get_due_automations( @@ -187,7 +187,7 @@ def test_persisted_cursor_survives_restart(fresh_db, automation_factory): ) now = datetime(2026, 2, 1, 10, 0, tzinfo=timezone.utc) due = get_due_automations(now) - assert claim_due_automation(due[0]) is True + assert claim_due_automation_run(due[0]) is not None automation_id = automation.id fresh_db.session.remove() @@ -254,7 +254,7 @@ def test_claim_rejects_automation_deactivated_after_discovery( automation.active = False fresh_db.session.commit() - assert claim_due_automation(due) is False + assert claim_due_automation_run(due) is None assert automation.cursor == cursor @@ -277,7 +277,7 @@ def test_claim_rejects_recurrence_edited_after_discovery( setattr(automation, field, new_value) fresh_db.session.commit() - assert claim_due_automation(due) is False + assert claim_due_automation_run(due) is None assert automation.cursor == cursor @@ -295,7 +295,7 @@ def test_claim_rejects_cursor_changed_after_discovery(fresh_db, automation_facto automation.cursor = newer_cursor fresh_db.session.commit() - assert claim_due_automation(due) is False + assert claim_due_automation_run(due) is None assert automation.cursor == newer_cursor @@ -311,7 +311,7 @@ def test_claim_allows_name_edit_after_discovery(fresh_db, automation_factory): automation.name = "New display name" fresh_db.session.commit() - assert claim_due_automation(due) is True + assert claim_due_automation_run(due) is not None def test_claim_rejects_automation_deleted_after_discovery(fresh_db, automation_factory): @@ -326,4 +326,4 @@ def test_claim_rejects_automation_deleted_after_discovery(fresh_db, automation_f fresh_db.session.delete(automation) fresh_db.session.commit() - assert claim_due_automation(due) is False + assert claim_due_automation_run(due) is None diff --git a/flexmeasures/data/tests/test_automations_fresh_db.py b/flexmeasures/data/tests/test_automations_fresh_db.py index 4fb1779eff..e017311f37 100644 --- a/flexmeasures/data/tests/test_automations_fresh_db.py +++ b/flexmeasures/data/tests/test_automations_fresh_db.py @@ -1,13 +1,14 @@ from __future__ import annotations -from datetime import timedelta, timezone +from datetime import datetime, timedelta, timezone +import isodate import pytest from rq.job import Job from sqlalchemy.exc import IntegrityError from flexmeasures.api.v3_0.tests.utils import message_for_trigger_schedule -from flexmeasures.data.models.automations import Automation +from flexmeasures.data.models.automations import Automation, AutomationRun from flexmeasures.data.services.automations import resolve_schedule_generator from flexmeasures.data.models.data_sources import DataSource from flexmeasures.data.models.generic_assets import GenericAsset, GenericAssetType @@ -166,6 +167,62 @@ def test_run_schedule_automation( } +def test_a_durable_run_schedules_what_it_was_planned_with( + fresh_db, + app, + add_battery_assets_fresh_db, + add_market_prices_fresh_db, + clean_scheduling_redis, +): + """A run dispatched from its own record schedules its stored parameters, and names itself on the job. + + The automation's parameters may have moved on since the run was planned, + so a retry must re-queue the run's plan rather than today's settings. + """ + battery = add_battery_assets_fresh_db["Test battery"] + message = message_for_trigger_schedule() + flex_model = message.pop("flex-model") + flex_model["sensor"] = battery.sensors[0].id + planned_parameters = {**message, "flex-model": [flex_model]} + + automation = build_schedule_automation( + battery, + name="Nightly schedules", + cronstr="0 0 * * *", + parameters=planned_parameters, + ) + fresh_db.session.add(automation) + fresh_db.session.flush() + run = AutomationRun( + automation=automation, + scheduled_at=datetime(2026, 8, 5, 1, 0, tzinfo=timezone.utc), + schedule_revision=automation.schedule_revision, + automation_type=automation.type, + generator_id=automation.generator_id, + dispatch_state="claimed", + execution_state="pending", + parameters=planned_parameters, + plan={}, + ) + fresh_db.session.add(run) + fresh_db.session.flush() + # The automation is edited after the run was planned, to a duration the run must not pick up. + automation.parameters = {**planned_parameters, "duration": "PT6H"} + fresh_db.session.flush() + + returns = run_automation(automation, automation_run=run) + + job = Job.fetch(returns["job_id"], connection=app.queues["scheduling"].connection) + assert job.meta["trigger"] == { + "origin": "automation", + "automation_id": automation.id, + "automation_run_id": run.id, + } + assert job.kwargs["end"] - job.kwargs["start"] == isodate.parse_duration( + planned_parameters["duration"] + ) + + def test_run_day_ahead_schedule_automation( fresh_db, app, diff --git a/flexmeasures/ui/static/openapi-specs.json b/flexmeasures/ui/static/openapi-specs.json index 6e312ecd58..09ec23540f 100644 --- a/flexmeasures/ui/static/openapi-specs.json +++ b/flexmeasures/ui/static/openapi-specs.json @@ -3419,7 +3419,7 @@ }, "get": { "summary": "Get details of one automation defined on an asset.", - "description": "In addition to the fields shown when listing automations, the response shows\nthe automation's parameters (forecast parameters or a schedule trigger message),\nthe data source it records under, as its `source` (null for schedule automations),\nthe sensors it reads from and writes to,\nand counts of recently created jobs, per job status.\nNote that jobs in Redis have a limited TTL, so not all past jobs will be counted.\nThe cursor is the time of the most recent run the automation committed to, in the automation's own timezone; runs at or before it are never queued again.\nIt advances just before queueing, so it does not indicate that queueing or the forecast itself succeeded.\n", + "description": "In addition to the fields shown when listing automations, the response shows\nthe automation's parameters (forecast parameters or a schedule trigger message),\nthe data source it records under, as its `source` (null for schedule automations),\nthe sensors it reads from and writes to,\ndurable run status, and counts of recently created jobs, per job status.\nNote that jobs in Redis have a limited TTL, so not all past jobs will be counted, while durable run status records queueing attempts and outcomes even after those jobs expire.\nThe cursor is the time of the most recent run the automation committed to, in the automation's own timezone; runs at or before it are never queued again.\nIt advances just before queueing, so it does not indicate that queueing or the forecast itself succeeded.\n", "security": [ { "ApiKeyAuth": [] @@ -3464,6 +3464,7 @@ "cursor": "2026-07-11T06:00:00+02:00", "next-run": "2026-07-12T06:00:00+02:00", "recurrence-description": "At 06:00", + "schedule-revision": 1, "active": true, "parameters": { "sensor": 2092 @@ -3492,6 +3493,27 @@ "finished": 3, "failed": 1 }, + "run-stats": { + "total": 1, + "dispatch": { + "queued": 1 + }, + "execution": { + "succeeded": 1 + }, + "latest-run": { + "id": 12, + "scheduled-at": "2026-07-11T04:00:00+00:00", + "schedule-revision": 1, + "dispatch-state": "queued", + "execution-state": "succeeded", + "attempt-count": 1, + "intended-job-count": 2, + "queued-job-count": 2, + "last-error": null + }, + "recent-runs": [] + }, "redis-connection-err": null } } @@ -3630,6 +3652,7 @@ "cursor": "2026-07-11T06:00:00+02:00", "next-run": "2026-07-12T06:00:00+02:00", "recurrence-description": "At 06:00", + "schedule-revision": 1, "active": true } ] diff --git a/flexmeasures/ui/templates/assets/asset_automations.html b/flexmeasures/ui/templates/assets/asset_automations.html index 2262e9192f..9a8fad4141 100644 --- a/flexmeasures/ui/templates/assets/asset_automations.html +++ b/flexmeasures/ui/templates/assets/asset_automations.html @@ -399,11 +399,27 @@
Automation '${esc(automation.name)}' (ID: ${automation.i const generator = res.source ? `${res.source.description} (data source ${res.source.id})` : "None"; + const runStats = res["run-stats"] || {}; + const latestRun = runStats["latest-run"]; + const latestRunSummary = latestRun + ? { + id: latestRun.id, + "scheduled-at": latestRun["scheduled-at"], + "dispatch-state": latestRun["dispatch-state"], + "execution-state": latestRun["execution-state"], + "attempt-count": latestRun["attempt-count"], + "intended-job-count": latestRun["intended-job-count"], + "queued-job-count": latestRun["queued-job-count"], + "last-error": latestRun["last-error"], + } + : null; $(`#AutomationDetailsBody-${automationId}`).html(`
Timezone

${esc(res.timezone)}

Cursor (UTC)

${esc(res.cursor || "Not initialized yet")}

+
Schedule revision
+

${esc(res["schedule-revision"])}

Data generator

${esc(generator)}

Reads from
@@ -412,6 +428,8 @@
Writes to

${sensorLinks(res["output-sensors"], res.source)}

Parameters
${esc(JSON.stringify(res.parameters, null, 4))}
+
Durable runs
+ 
${esc(JSON.stringify({ total: runStats.total || 0, dispatch: runStats.dispatch || {}, execution: runStats.execution || {}, "latest-run": latestRunSummary }, null, 4))}
Recently created jobs
${esc(JSON.stringify(jobStats, null, 4))}
`);