Fix:cascade subjob failure

documentation/api/change_log.rst

@@ -5,14 +5,13 @@ API change log
 
 .. note:: The FlexMeasures API follows its own versioning scheme. This is also reflected in the URL (e.g. `/api/v3_0`), allowing developers to upgrade at their own pace.
 
-v3.0-34 | September 2, 2026
-"""""""""""""""""""""""""""
-- Added ``POST /api/v3_0/assets/<id>/automations/<automation_id>/trigger``, to run one automation now, once, on top of its recurring runs. The response is the standard job response, extended with ``n_jobs``: how many jobs the run queued. An on-demand run does not affect the automation's recurrence, and inactive automations can be triggered, too. Triggering requires the same permission as writing data under the asset, and falls under the stricter rate limit that the other triggering endpoints share.
 
-v3.0-33 | September 1, 2026
+v3.0-33 | September 3, 2026
 """""""""""""""""""""""""""
+- Added ``POST /api/v3_0/assets/<id>/automations/<automation_id>/trigger``, to run one automation now, once, on top of its recurring runs. The response is the standard job response, extended with ``n_jobs``: how many jobs the run queued. An on-demand run does not affect the automation's recurrence, and inactive automations can be triggered, too. Triggering requires the same permission as writing data under the asset, and falls under the stricter rate limit that the other triggering endpoints share.
 - Added ``GET /api/v3_0/assets/<id>/automations`` and ``GET /api/v3_0/assets/<id>/automations/<automation_id>`` 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/<id>`` to show the full record of one data source, including the attributes in which data generators store their configuration.
+- Fixed: when a sequential schedule (triggered with ``"sequential": true`` on `/assets/(id)/schedules/trigger <../api/v3_0.html#post--api-v3_0-assets-id-schedules-trigger>`_ (POST)) cannot schedule one of its devices, and the scheduler defines no fallback scheduler, the job whose id was returned now reaches a terminal failed state, rather than staying deferred indefinitely. ``GET /api/v3_0/jobs/<uuid>`` answers such a job with ``422 Unprocessable Entity``, a ``FAILED`` status and a ``message`` naming the device that could not be scheduled (and the devices that were consequently not scheduled either); ``GET /sensors/<id>/schedules/<uuid>`` answers with ``UNKNOWN_SCHEDULE`` and the same reason.
 
 v3.0-32 | August 11, 2026
 """""""""""""""""""""""""

documentation/api/introduction.rst

@@ -244,6 +244,22 @@ This returns the current execution status and a human-readable result message. F
 
     Both of these endpoints will also return `202 Accepted` if the job is still being computed, so clients can continue to poll them directly if they prefer.
 
+**Retrying after a failed job:**
+
+Schedule trigger requests are de-duplicated: a request whose arguments match one that was sent before is answered with the id of the job that was already created for it, rather than with a new job.
+That holds for as long as the job cache remembers the request (see the ``FLEXMEASURES_JOB_CACHE_TTL`` config setting, one hour by default), and regardless of how that job ended.
+Re-sending a request whose job failed therefore hands back that same failed job, rather than starting a new attempt.
+
+To have FlexMeasures compute a new schedule within that hour, either change something about the request, or set ``force-new-job-creation``:
+
+.. code-block:: json
+
+    {
+        "start": "2015-06-02T10:00:00+00:00",
+        "duration": "PT12H",
+        "force-new-job-creation": true
+    }
+
 .. _api_deprecation:
 
 Deprecation and sunset

flexmeasures/api/v3_0/tests/test_asset_schedules_fresh_db.py

@@ -6,7 +6,7 @@
 
 from numpy.testing import assert_almost_equal
 import pandas as pd
-from rq.job import Job
+from rq.job import Job, JobStatus
 
 from flexmeasures import Sensor
 from flexmeasures.api.v3_0.tests.utils import message_for_trigger_schedule
@@ -18,6 +18,7 @@
     handle_scheduling_exception,
     get_data_source_for_job,
 )
+from flexmeasures.data.models.planning.storage import StorageScheduler
 from flexmeasures.data.services.utils import sort_jobs
 from flexmeasures.utils.unit_utils import ur
 
@@ -1098,3 +1099,116 @@ def test_asset_trigger_with_group_referencing_sensor_outside_asset_tree(
 
     # No scheduling job should have been queued
     assert len(app.queues["scheduling"]) == 0
+
+
+@pytest.mark.parametrize(
+    "requesting_user", ["test_prosumer_user@seita.nl"], indirect=True
+)
+def test_asset_sequential_schedule_without_fallback_fails_terminally(
+    app,
+    add_market_prices_fresh_db,
+    setup_roles_users_fresh_db,
+    add_charging_station_assets_fresh_db,
+    keep_scheduling_queue_empty,
+    requesting_user,
+):
+    """Trigger a sequential schedule whose first device is infeasible, using a scheduler without a fallback.
+
+    No scheduler defines a fallback since PR #2252, so the storage scheduler is used as it comes.
+    The job id handed to the client is the one of the wrap-up job. Polling it should yield a terminal failure,
+    with a reason naming the device that could not be scheduled, rather than a job that stays deferred forever.
+    Re-triggering the same request should not hand back a job that is still waiting on that chain, either.
+    """
+    price_sensor_id = add_market_prices_fresh_db["epex_da"].id
+
+    # The uni-directional charging station cannot discharge, and cannot charge faster than its power capacity,
+    # so a usage above that capacity cannot be met. SoC bounds and targets are relaxed by default since PR #2252,
+    # but a device's power capacity stays hard, so this is a genuine infeasibility rather than a priced breach.
+    charging_station = add_charging_station_assets_fresh_db["Test charging station"]
+    infeasible_sensor = charging_station.sensors[0]
+    bidirectional_charging_station = add_charging_station_assets_fresh_db[
+        "Test charging station (bidirectional)"
+    ]
+    feasible_sensor = bidirectional_charging_station.sensors[0]
+
+    message = {
+        "start": "2015-01-02T00:00:00+01:00",
+        "duration": "PT24H",
+        "resolution": "PT15M",
+        "sequential": True,
+        "flex-context": {
+            "consumption-price": {"sensor": price_sensor_id},
+            "production-price": {"sensor": price_sensor_id},
+            "site-power-capacity": "1 TW",
+        },
+        "flex-model": [
+            {
+                "sensor": infeasible_sensor.id,
+                "soc-at-start": 10,
+                "soc-min": 0,
+                "soc-max": 40,
+                "power-capacity": "1 MW",
+                "soc-usage": ["10 MW"],
+            },
+            {
+                "sensor": feasible_sensor.id,
+                "soc-at-start": 10,
+                "soc-min": 0,
+                "soc-max": 40,
+            },
+        ],
+    }
+    site_id = charging_station.parent_asset.id
+
+    deferred_registry = app.queues["scheduling"].deferred_job_registry
+    jobs_deferred_by_other_tests = set(deferred_registry.get_job_ids())
+
+    assert (
+        StorageScheduler.fallback_scheduler_class is None
+    ), "This test needs a scheduler without a fallback."
+
+    with app.test_client() as client:
+        trigger_schedule_response = client.post(
+            url_for("AssetAPI:trigger_schedule", id=site_id),
+            json=message,
+        )
+        assert trigger_schedule_response.status_code == 202
+        job_id = trigger_schedule_response.json["job"]
+
+        # The subjob for the second device, and the wrap-up job, wait for the first device to be scheduled
+        deferred_jobs_of_this_chain = (
+            set(deferred_registry.get_job_ids()) - jobs_deferred_by_other_tests
+        )
+        assert len(deferred_jobs_of_this_chain) == 2
+
+        work_on_rq(app.queues["scheduling"], exc_handler=handle_scheduling_exception)
+
+        # Polling the job we were handed gives a terminal failure, naming the device that could not be scheduled
+        job_status_response = client.get(url_for("JobAPI:get_job_status", uuid=job_id))
+        print("Server responded with:\n%s" % job_status_response.json)
+        assert job_status_response.status_code == 422
+        assert job_status_response.json["status"] == "FAILED"
+        message_to_client = job_status_response.json["message"]
+        assert (
+            f"sensor {infeasible_sensor.id} ({charging_station.name} - {infeasible_sensor.name})"
+            in message_to_client
+        )
+        assert "InfeasibleProblemException" in message_to_client
+
+        # No job is left waiting on a chain that will never complete
+        assert deferred_jobs_of_this_chain.isdisjoint(deferred_registry.get_job_ids())
+
+        # Re-triggering the same request does not hand back a job that is still waiting on that chain
+        retrigger_response = client.post(
+            url_for("AssetAPI:trigger_schedule", id=site_id),
+            json=message,
+        )
+        assert retrigger_response.status_code == 202
+        retriggered_job = Job.fetch(
+            retrigger_response.json["job"],
+            connection=app.queues["scheduling"].connection,
+        )
+        assert retriggered_job.get_status(refresh=True) not in (
+            JobStatus.DEFERRED,
+            JobStatus.SCHEDULED,
+        )

flexmeasures/data/models/planning/exceptions.py

@@ -24,3 +24,9 @@ class WrongTypeAttributeException(Exception):
 
 class InfeasibleProblemException(Exception):
     pass
+
+
+class UpstreamSchedulingFailure(Exception):
+    """A schedule could not be computed, because a scheduling job it depended on failed."""
+
+    pass