From c8ebc229f96c17861de178b6f44cb748d137bc8d Mon Sep 17 00:00:00 2001 From: Mohamed Belhsan Hmida Date: Fri, 7 Aug 2026 13:47:12 +0100 Subject: [PATCH 01/11] data/scheduling: cascade a subjob failure to its dependents Context: - Issue #2404: when a subjob in a sequential scheduling chain fails and its scheduler defines no fallback, the chain never reaches a terminal state. RQ only enqueues the dependents of a job that succeeded, so the remaining subjobs and the wrap-up job stay deferred forever. The wrap-up job's id is what POST /assets//schedules/trigger hands to the client, so that client never learns what went wrong. - This already happens for every scheduler without a fallback (the base Scheduler defaults fallback_scheduler_class to None), such as the ProcessScheduler and any custom scheduler. Change: - trigger_optional_fallback now handles any failure without a fallback job by failing the deferred jobs that depend on the failed job, recording an UpstreamSchedulingFailure that names the device that could not be scheduled. A failing fallback job cascades from the original job, whose dependents it was standing in for. - The wrap-up job of a sequential chain now tolerates a failing dependency, so it runs and fails with a message listing the devices that failed and the devices that were consequently never scheduled. - Split the fallback creation out of trigger_optional_fallback into _trigger_fallback_job, which reports whether a fallback job was created. Signed-off-by: Mohamed Belhsan Hmida --- .../data/models/planning/exceptions.py | 6 + flexmeasures/data/services/scheduling.py | 287 ++++++++++++++---- 2 files changed, 230 insertions(+), 63 deletions(-) diff --git a/flexmeasures/data/models/planning/exceptions.py b/flexmeasures/data/models/planning/exceptions.py index f25c0f62b4..f7e5f70d9a 100644 --- a/flexmeasures/data/models/planning/exceptions.py +++ b/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 diff --git a/flexmeasures/data/services/scheduling.py b/flexmeasures/data/services/scheduling.py index 188ea959c2..1507a1c2ab 100644 --- a/flexmeasures/data/services/scheduling.py +++ b/flexmeasures/data/services/scheduling.py @@ -19,8 +19,8 @@ from flask import current_app from isodate import duration_isoformat from rq import get_current_job, Callback -from rq.exceptions import InvalidJobOperation -from rq.job import Job +from rq.exceptions import InvalidJobOperation, NoSuchJobError +from rq.job import Dependency, Job, JobStatus import timely_beliefs as tb import pandas as pd from sqlalchemy import select @@ -32,7 +32,10 @@ SCHEDULING_RESULT_KEY, ) from flexmeasures.data.models.planning.devices import INFLEXIBLE_DEVICE_KEYS -from flexmeasures.data.models.planning.exceptions import InfeasibleProblemException +from flexmeasures.data.models.planning.exceptions import ( + InfeasibleProblemException, + UpstreamSchedulingFailure, +) from flexmeasures.data.models.planning.process import ProcessScheduler from flexmeasures.data.services.scheduling_result import SchedulingJobResult from flexmeasures.data.models.time_series import Sensor, TimedBelief @@ -42,6 +45,7 @@ from flexmeasures.data.utils import get_data_source, save_to_db from flexmeasures.utils.time_utils import server_now from flexmeasures.data.services.utils import ( + failed_job_reason, job_cache, get_asset_or_sensor_ref, get_asset_or_sensor_from_ref, @@ -134,73 +138,165 @@ def success_callback(job, connection, result, *args, **kwargs): queue.deferred_job_registry.requeue(dependent_job_ids) +def _describe_scheduled_device(asset_or_sensor_ref: dict | None) -> str: + """Describe the device that a scheduling job was scheduling, for use in a failure message. + + :param asset_or_sensor_ref: Serialized reference to an Asset or Sensor, as stored in a job's meta data. + """ + if not asset_or_sensor_ref: + return "an unknown device" + asset_or_sensor = get_asset_or_sensor_from_ref(asset_or_sensor_ref) + kind = asset_or_sensor_ref["class"].lower() + if asset_or_sensor is None: + return f"{kind} {asset_or_sensor_ref['id']}" + if isinstance(asset_or_sensor, Sensor): + return f"{kind} {asset_or_sensor.id} ({asset_or_sensor.generic_asset.name} - {asset_or_sensor.name})" + return f"{kind} {asset_or_sensor.id} ({asset_or_sensor.name})" + + def trigger_optional_fallback(job, connection, type, value, traceback): - """Create a fallback schedule job when the error is of type InfeasibleProblemException""" + """Handle a failed scheduling job. + + A fallback schedule job is created when the error is of type InfeasibleProblemException, + and the scheduler that failed defines a fallback scheduler. + + Schedulers are not required to define a fallback, though. Without one, the failure is cascaded to the jobs that depend on the failed job, + so that a client polling one of them (such as the wrap-up job of a sequential schedule, whose id is what the trigger endpoint returns) + reaches a terminal state with a reason, rather than waiting on a job that stays deferred forever. + """ job.meta["exception"] = value job.save_meta() - if type is InfeasibleProblemException: - asset_or_sensor = get_asset_or_sensor_from_ref(job.meta.get("asset_or_sensor")) + if type is InfeasibleProblemException and _trigger_fallback_job(job): + return + + # A failing fallback job leaves the dependents of the original job deferred, so cascade from that job instead. + job_with_dependents = job + original_job_id = job.meta.get("original_job_id") + if original_job_id is not None: + try: + job_with_dependents = Job.fetch(original_job_id, connection=connection) + except NoSuchJobError: + current_app.logger.error( + f"Original job with ID={original_job_id} (fallback Job ID={job.id}) not found, so its dependents cannot be failed." + ) + return + + if not job_with_dependents.dependent_ids: + return + device = _describe_scheduled_device(job.meta.get("asset_or_sensor")) + _cascade_failure_to_dependents( + job_with_dependents, + connection, + reason=f"Scheduling {device} failed with {type.__name__}: {value}, so this schedule could not be computed either.", + ) - scheduler_kwargs = job.meta["scheduler_kwargs"] - # Deserialize start, end, resolution and belief_time - # Workaround for https://github.com/Parallels/rq-dashboard/issues/510 - timezone = "UTC" - if hasattr(asset_or_sensor, "timezone"): - timezone = asset_or_sensor.timezone - scheduler_kwargs["start"] = pd.Timestamp(scheduler_kwargs["start"]).tz_convert( - timezone - ) - scheduler_kwargs["end"] = pd.Timestamp(scheduler_kwargs["end"]).tz_convert( - timezone +def _trigger_fallback_job(job) -> bool: + """Create and enqueue a fallback schedule job for a failed scheduling job, if its scheduler defines a fallback. + + :param job: The failed scheduling job. + :returns: True if a fallback job was created, and False if the scheduler has no fallback. + """ + asset_or_sensor = get_asset_or_sensor_from_ref(job.meta.get("asset_or_sensor")) + + scheduler_kwargs = job.meta["scheduler_kwargs"] + + # Deserialize start, end, resolution and belief_time + # Workaround for https://github.com/Parallels/rq-dashboard/issues/510 + timezone = "UTC" + if hasattr(asset_or_sensor, "timezone"): + timezone = asset_or_sensor.timezone + scheduler_kwargs["start"] = pd.Timestamp(scheduler_kwargs["start"]).tz_convert( + timezone + ) + scheduler_kwargs["end"] = pd.Timestamp(scheduler_kwargs["end"]).tz_convert(timezone) + if isinstance(scheduler_kwargs.get("belief_time"), str): + scheduler_kwargs["belief_time"] = pd.Timestamp( + scheduler_kwargs["belief_time"] + ).tz_convert(timezone) + if isinstance(scheduler_kwargs.get("resolution"), str): + scheduler_kwargs["resolution"] = pd.Timedelta(scheduler_kwargs["resolution"]) + + if ("scheduler_specs" in job.kwargs) and ( + job.kwargs["scheduler_specs"] is not None + ): + scheduler_class: Type[Scheduler] = load_custom_scheduler( + job.kwargs["scheduler_specs"] ) - if isinstance(scheduler_kwargs.get("belief_time"), str): - scheduler_kwargs["belief_time"] = pd.Timestamp( - scheduler_kwargs["belief_time"] - ).tz_convert(timezone) - if isinstance(scheduler_kwargs.get("resolution"), str): - scheduler_kwargs["resolution"] = pd.Timedelta( - scheduler_kwargs["resolution"] - ) + else: + scheduler_class: Type[Scheduler] = find_scheduler_class(asset_or_sensor) - if ("scheduler_specs" in job.kwargs) and ( - job.kwargs["scheduler_specs"] is not None - ): - scheduler_class: Type[Scheduler] = load_custom_scheduler( - job.kwargs["scheduler_specs"] - ) - else: - scheduler_class: Type[Scheduler] = find_scheduler_class(asset_or_sensor) - - # only schedule a fallback schedule job if the original job has a fallback - # mechanism - if scheduler_class.fallback_scheduler_class is not None: - scheduler_class = scheduler_class.fallback_scheduler_class - scheduler_specs = { - "class": scheduler_class.__name__, - "module": inspect.getmodule(scheduler_class).__name__, - } + # only schedule a fallback schedule job if the original job has a fallback + # mechanism + if scheduler_class.fallback_scheduler_class is None: + return False - fallback_job = create_scheduling_job( - asset_or_sensor, - force_new_job_creation=True, - enqueue=False, - scheduler_specs=scheduler_specs, - success_callback=Callback(success_callback), - **scheduler_kwargs, - ) + scheduler_class = scheduler_class.fallback_scheduler_class + scheduler_specs = { + "class": scheduler_class.__name__, + "module": inspect.getmodule(scheduler_class).__name__, + } - # keep track of the id of the original (non-fallback) job - fallback_job.meta["original_job_id"] = job.meta.get( - "original_job_id", job.id - ) - fallback_job.save_meta() + fallback_job = create_scheduling_job( + asset_or_sensor, + force_new_job_creation=True, + enqueue=False, + scheduler_specs=scheduler_specs, + success_callback=Callback(success_callback), + **scheduler_kwargs, + ) - job.meta["fallback_job_id"] = fallback_job.id - job.save_meta() - current_app.queues["scheduling"].enqueue_job(fallback_job) + # keep track of the id of the original (non-fallback) job + fallback_job.meta["original_job_id"] = job.meta.get("original_job_id", job.id) + fallback_job.save_meta() + + job.meta["fallback_job_id"] = fallback_job.id + job.save_meta() + current_app.queues["scheduling"].enqueue_job(fallback_job) + return True + + +def _cascade_failure_to_dependents(job: Job, connection, reason: str) -> None: + """Put the jobs that depend on a failed scheduling job into a terminal state, too. + + RQ only enqueues the dependents of a job that succeeded, so a failed job without a fallback would otherwise leave its dependents deferred forever, + which leaves a client polling such a job (in particular the wrap-up job of a sequential schedule) without a terminal state or a reason. + + Dependent jobs that were set up to tolerate a failing dependency are enqueued rather than failed, so they can run and report on the failure. + RQ itself does that for the dependents of the job that just failed, so here we only need to do it for the jobs that we fail ourselves. + + :param job: The failed job whose dependents should be dealt with. + :param connection: Redis connection. + :param reason: Why the schedule could not be computed, naming the device that failed to be scheduled. + """ + queue = current_app.queues["scheduling"] + dependent_ids = list(job.dependent_ids) + if not dependent_ids: + return + for dependent in Job.fetch_many(dependent_ids, connection=connection): + if dependent is None or dependent.allow_dependency_failures: + continue + if dependent.get_status(refresh=True) != JobStatus.DEFERRED: + continue + _fail_deferred_job(dependent, reason) + _cascade_failure_to_dependents(dependent, connection, reason) + queue.enqueue_dependents(dependent) + + +def _fail_deferred_job(job: Job, reason: str) -> None: + """Move a deferred job that will never run to a terminal failed state, recording why. + + :param job: The deferred job. + :param reason: Why the schedule could not be computed, naming the device that failed to be scheduled. + """ + queue = current_app.queues["scheduling"] + job.meta["exception"] = UpstreamSchedulingFailure(reason) + job.save_meta() + job.set_status(JobStatus.FAILED) + queue.deferred_job_registry.remove(job) + queue.failed_job_registry.add(job, ttl=job.failure_ttl, exc_string=reason) @job_cache("scheduling") @@ -331,11 +427,71 @@ def create_scheduling_job( def cb_done_sequential_scheduling_job(jobs_ids: list[str]): - """ + """Wrap up a chain of sequential scheduling (sub)jobs. + + This job also runs when one of the subjobs failed (see the Dependency set up in create_sequential_scheduling_job), + in which case it fails, too, naming the devices that could not be scheduled. + Its id is what the trigger endpoint hands to the client, so this is what gives that client a terminal state and a reason. + TODO: maybe check if any of the subjobs used a fallback scheduler or accrued a relaxation penalty. + + :param jobs_ids: Ids of the scheduling subjobs in the chain. + :raises UpstreamSchedulingFailure: When any of the subjobs did not produce a schedule. """ - current_app.logger.info("Sequential scheduling job finished its chain of subjobs.") - # jobs = [Job.fetch(job_id) for job_id in jobs_ids] + connection = current_app.queues["scheduling"].connection + failed_devices, skipped_devices = [], [] + for job_id in jobs_ids: + if _scheduling_job_succeeded(job_id, connection): + continue + try: + job = Job.fetch(job_id, connection=connection) + except NoSuchJobError: + failed_devices.append( + f"an unknown device (scheduling job {job_id} is no longer available)" + ) + continue + device = _describe_scheduled_device(job.meta.get("asset_or_sensor")) + if isinstance(job.meta.get("exception"), UpstreamSchedulingFailure): + # This device was never scheduled, because a device earlier in the chain failed. + skipped_devices.append(device) + else: + reason = failed_job_reason(job) or f"job status is {job.get_status()}" + failed_devices.append(f"{device}: {reason}") + + if not failed_devices and not skipped_devices: + current_app.logger.info( + "Sequential scheduling job finished its chain of subjobs." + ) + return + + complaints = [] + if failed_devices: + complaints.append( + f"Sequential scheduling failed for {'; '.join(failed_devices)}." + ) + if skipped_devices: + complaints.append( + f"As a result, no schedule was computed for {', '.join(skipped_devices)}." + ) + raise UpstreamSchedulingFailure(" ".join(complaints)) + + +def _scheduling_job_succeeded(job_id: str, connection) -> bool: + """Tell whether a scheduling job produced a schedule, either by itself or through its fallback job. + + :param job_id: Id of the scheduling job. + :param connection: Redis connection. + """ + try: + job = Job.fetch(job_id, connection=connection) + except NoSuchJobError: + return False + if job.get_status(refresh=True) == JobStatus.FINISHED: + return True + fallback_job_id = job.meta.get("fallback_job_id") + if fallback_job_id is None: + return False + return _scheduling_job_succeeded(fallback_job_id, connection) def _add_inflexible_devices(flex_context: dict, sensors: list[Sensor]) -> None: @@ -447,11 +603,16 @@ def create_sequential_scheduling_job( previous_sensors.append(sensor) previous_job = job - # create job that triggers when the last job is done + # create job that triggers when the last job is done, or failed: + # tolerating a failing dependency lets the wrap-up job report which devices could not be scheduled, + # rather than staying deferred forever (see cb_done_sequential_scheduling_job) + depends_on_last_job = ( + Dependency(previous_job, allow_failure=True) if previous_job else None + ) job = Job.create( func=cb_done_sequential_scheduling_job, args=([j.id for j in jobs],), - depends_on=previous_job, + depends_on=depends_on_last_job, ttl=int( current_app.config.get( "FLEXMEASURES_JOB_TTL", timedelta(-1) From fd0ec1445dda16b534224d61669d8dae44c4f019 Mon Sep 17 00:00:00 2001 From: Mohamed Belhsan Hmida Date: Fri, 7 Aug 2026 13:47:55 +0100 Subject: [PATCH 02/11] tests/scheduling: cover the failure cascade in sequential chains Context: - Issue #2404: a subjob failing under a scheduler without a fallback used to leave the rest of the chain deferred forever, so nothing proved that a client ever reaches a terminal state. Change: - test_create_sequential_jobs_without_fallback retires the storage fallback and makes the first device infeasible, then asserts that the second subjob and the wrap-up job both end up failed, that the wrap-up job's reason names the device that could not be scheduled, and that no job is left deferred. - test_asset_sequential_schedule_without_fallback_fails_terminally covers the same end-to-end: it triggers a sequential schedule over the API with a genuinely infeasible first device, polls the returned job id, and asserts a 422 whose message names that device. It also re-triggers the same request and asserts the job it gets back is not waiting on a chain that will never complete. Signed-off-by: Mohamed Belhsan Hmida --- .../tests/test_asset_schedules_fresh_db.py | 117 +++++++++++++++++- .../data/tests/test_scheduling_sequential.py | 75 +++++++++++ 2 files changed, 191 insertions(+), 1 deletion(-) diff --git a/flexmeasures/api/v3_0/tests/test_asset_schedules_fresh_db.py b/flexmeasures/api/v3_0/tests/test_asset_schedules_fresh_db.py index 8100b4af8a..772ecdb8c4 100644 --- a/flexmeasures/api/v3_0/tests/test_asset_schedules_fresh_db.py +++ b/flexmeasures/api/v3_0/tests/test_asset_schedules_fresh_db.py @@ -1,12 +1,14 @@ from __future__ import annotations +from unittest.mock import patch + from flask import url_for import pytest from isodate import parse_datetime, parse_duration 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 @@ -1098,3 +1100,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. + + 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, so it cannot reach a target below its initial SoC + 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, + "soc-targets": [{"value": 9, "datetime": "2015-01-02T02:00:00+01:00"}], + }, + { + "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()) + + storage_module = "flexmeasures.data.models.planning.storage" + with patch(f"{storage_module}.StorageScheduler.fallback_scheduler_class", None): + 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, + ) diff --git a/flexmeasures/data/tests/test_scheduling_sequential.py b/flexmeasures/data/tests/test_scheduling_sequential.py index aea1d5317f..b22f37cd2b 100644 --- a/flexmeasures/data/tests/test_scheduling_sequential.py +++ b/flexmeasures/data/tests/test_scheduling_sequential.py @@ -6,6 +6,7 @@ from flexmeasures.data.services.scheduling import create_sequential_scheduling_job from flexmeasures.utils.job_utils import work_on_rq from flexmeasures.data.services.scheduling import handle_scheduling_exception +from flexmeasures.data.services.utils import failed_job_reason, sort_jobs from flexmeasures.data.models.time_series import Sensor @@ -233,6 +234,80 @@ def test_create_sequential_jobs_fallback( assert deferred_jobs[1].id in finished_jobs +def test_create_sequential_jobs_without_fallback( + db, app, flex_description_sequential, smart_building +): + """Test that a failing subjob without a fallback scheduler does not wedge the chain. + + The first device is infeasible, and its scheduler has no fallback. The remaining subjobs can then never run, + so they should be failed rather than left deferred, and the wrap-up job — whose id is what the trigger endpoint hands to the client — + should reach a terminal failed state naming the device that could not be scheduled. + """ + assets, sensors, _ = smart_building + queue = app.queues["scheduling"] + + start = pd.Timestamp("2015-01-03").tz_localize("Europe/Amsterdam") + end = pd.Timestamp("2015-01-04").tz_localize("Europe/Amsterdam") + + scheduler_specs = { + "module": "flexmeasures.data.models.planning.storage", + "class": "StorageScheduler", + } + + flex_description_sequential["start"] = start + flex_description_sequential["end"] = end + + storage_module = "flexmeasures.data.models.planning.storage" + + with patch(f"{storage_module}.StorageScheduler.persist_flex_model"): + # Retire the fallback scheduler, like ProcessScheduler and custom schedulers do by default + with patch(f"{storage_module}.StorageScheduler.fallback_scheduler_class", None): + with patch( + f"{storage_module}.StorageScheduler.compute", + side_effect=iter([InfeasibleProblemException(), [], []]), + ): + create_sequential_scheduling_job( + asset=assets["Test Site"], + scheduler_specs=scheduler_specs, + enqueue=True, + force_new_job_creation=True, # otherwise the cache might kick in due to sub-jobs already created in other tests + **flex_description_sequential, + ) + + queued_jobs = queue.jobs + deferred_jobs = sort_jobs( + queue, queue.deferred_job_registry.get_job_ids() + ) + assert len(queued_jobs) == 1 + assert len(deferred_jobs) == 2 + ev_job = queued_jobs[0] + battery_job, wrapup_job = deferred_jobs + + # Work on jobs + work_on_rq(queue, exc_handler=handle_scheduling_exception) + + failed_jobs = queue.failed_job_registry.get_job_ids() + + # The EV subjob failed, and had no fallback to fall back on + assert ev_job.id in failed_jobs + ev_job.refresh() + assert "fallback_job_id" not in ev_job.meta + + # The battery subjob can never run, so it was failed rather than left deferred + assert battery_job.id in failed_jobs + assert battery_job.get_status() == "failed" + + # The wrap-up job ran, and failed while naming the device that could not be scheduled + assert wrapup_job.id in failed_jobs + assert wrapup_job.get_status() == "failed" + reason = failed_job_reason(Job.fetch(wrapup_job.id, connection=queue.connection)) + assert f"sensor {sensors['Test EV'].id} (Test EV - power)" in reason + assert "InfeasibleProblemException" in reason + + # No job is left waiting on a chain that will never complete + assert queue.deferred_job_registry.get_job_ids() == [] + + def test_create_sequential_jobs_with_sign_explicit_context( db, app, flex_description_sequential, smart_building ): From de9294347bf9e36f7bbef2e0cbc04aefad42a08a Mon Sep 17 00:00:00 2001 From: Mohamed Belhsan Hmida Date: Fri, 7 Aug 2026 13:48:55 +0100 Subject: [PATCH 03/11] docs/changelog: note the terminal state for failed sequential schedules Context: - Issue #2404: the client-visible outcome of a failing sequential schedule changed, so both the general and the API change log need an entry. Change: - Added a bugfix entry describing the cascade, and an API change log entry stating the status codes and message a client now gets when polling a sequential schedule whose device could not be scheduled. Signed-off-by: Mohamed Belhsan Hmida --- documentation/api/change_log.rst | 1 + documentation/changelog.rst | 1 + 2 files changed, 2 insertions(+) diff --git a/documentation/api/change_log.rst b/documentation/api/change_log.rst index 58681e6c43..9b20e129a6 100644 --- a/documentation/api/change_log.rst +++ b/documentation/api/change_log.rst @@ -7,6 +7,7 @@ API change log v3.0-32 | July XX, 2026 """""""""""""""""""""""" +- 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/`` 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//schedules/`` answers with ``UNKNOWN_SCHEDULE`` and the same reason. - API endpoints are now rate-limited. A request which exceeds a limit is answered with a ``429 (Too Many Requests)`` status code and a ``Retry-After`` header stating how many seconds to wait. Responses also carry ``X-RateLimit-*`` headers, describing the limit that applied, how much of it is left, and when it resets. A stricter limit applies to ``POST /assets//schedules/trigger``, ``POST /sensors//schedules/trigger`` and ``POST /sensors//forecasts/trigger`` than to other endpoints; the health endpoints are exempt. Per-account overrides are set by assigning the account a plan (a ``Plan`` database row), rather than through an account attribute. - Introduced the ``inflexible-consumption`` and ``inflexible-production`` flex-context fields, which make explicit how the sign of each inflexible device's power data should be read: positive values denote consumption resp. production. Each entry is a sensor reference (``{"sensor": }``), optionally with source filters (``source-types``, ``exclude-source-types``, ``sources``, ``source-account``). Deprecated the ``inflexible-device-sensors`` field (a list of bare sensor IDs, whose sign convention is read from each sensor's ``consumption_is_positive`` attribute); it remains supported, but cannot be combined with the new fields in one flex-context. - Added a ``role`` query parameter to ``GET /api/v3_0/accounts`` for filtering accessible organisations by account role. diff --git a/documentation/changelog.rst b/documentation/changelog.rst index df0181b484..57d007b323 100644 --- a/documentation/changelog.rst +++ b/documentation/changelog.rst @@ -87,6 +87,7 @@ Infrastructure / Support Bugfixes ----------- +* A sequential schedule no longer gets stuck when one of its devices fails to be scheduled by a scheduler without a fallback (such as the ``ProcessScheduler``, or a custom scheduler): the failure now cascades to the remaining subjobs and to the wrap-up job whose id the trigger endpoint returns, so a client polling that job gets a terminal failure naming the device that could not be scheduled, instead of a job that stays deferred forever [see `PR #2406 `_ and `issue #2404 `_] * Show icons for more asset types in the UI's asset structure view, which previously fell back to a question mark: the ``wind``, ``process`` and ``heat-storage`` types that FlexMeasures seeds by default, and EV infrastructure under its various names (such as ``one-way_evse``, ``two-way_evse``, ``evse``, ``charging_station`` and ``charging_hub``) and building services equipment (``hvac``, ``ahu``, ``dhw``, ``heatpump``, ``chiller``, ``lighting`` and ``other-loads``). Asset type names are now matched ignoring case and separators, so an asset type named ``charge-point`` gets the same icon as ``chargepoint`` [see `PR #2391 `_] * Replaying a chart for a past window no longer shows annotations that were only recorded later; annotation searches and the ``chart_annotations`` endpoints can now be scoped by recording (belief) time [see `PR #2367 `_] * Continuing the query-parameter cleanup started in PR #2352: the chart-related endpoints now use ``prior``, ``start``, ``end`` and hyphenated field names, with a new ``duration`` field to derive a missing ``start``/``end``; old spellings keep working as legacy aliases [see `PR #2367 `_] From bcac4c8ea8985c48931049fcc2dd6381fe4b9ce6 Mon Sep 17 00:00:00 2001 From: Mohamed Belhsan Hmida Date: Fri, 7 Aug 2026 14:03:32 +0100 Subject: [PATCH 04/11] docs/api: explain how to retry after a failed scheduling job Context: - The background-job section walks a client through triggering and polling, including the failure path, but never says what happens when that client re-sends the request: schedule triggers are de-duplicated, so within FLEXMEASURES_JOB_CACHE_TTL the same failed job is handed back instead of a new attempt. - Issue #2404 makes this reachable in practice, since a failing sequential schedule now ends up in a terminal failed state that a client will want to retry. Change: - Added a "Retrying after a failed job" paragraph to the background job monitoring section, naming the config setting that governs the cache and showing the force-new-job-creation field that bypasses it. Signed-off-by: Mohamed Belhsan Hmida --- documentation/api/introduction.rst | 16 ++++++++++++++++ 1 file changed, 16 insertions(+) diff --git a/documentation/api/introduction.rst b/documentation/api/introduction.rst index 08712d9020..9b0feb3bae 100644 --- a/documentation/api/introduction.rst +++ b/documentation/api/introduction.rst @@ -243,6 +243,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 From d5dbe9bf6d946517f008df456c0ca5634b5b15fd Mon Sep 17 00:00:00 2001 From: Mohamed Belhsan Hmida Date: Fri, 7 Aug 2026 14:19:33 +0100 Subject: [PATCH 05/11] data/scheduling: do not let the wrap-up job outrun a pending fallback job Context: - Marking the wrap-up job's dependency with Dependency(allow_failure=True) made RQ enqueue that job the moment the last subjob failed. When that subjob has a fallback scheduler, its fallback job is enqueued by the same failure, so both sit in the queue at once: with more than one scheduling worker, the wrap-up job can run while the fallback is still pending, see a subjob that has not (yet) produced a schedule, and report the chain as failed just before the fallback schedules the device after all. - A single worker pops the queue in order and happens to run the fallback first, which is why this did not show up in the tests. Change: - The wrap-up job depends on the last subjob plainly again, so RQ never enqueues it on failure and the fallback path behaves as it did before this branch. - A job that should run anyway is now marked in its meta data with RUNS_ON_CHAIN_FAILURE, and the cascade queues those jobs itself, after the rest of the chain has reached a terminal state. Signed-off-by: Mohamed Belhsan Hmida --- flexmeasures/data/services/scheduling.py | 40 ++++++++++++++++-------- 1 file changed, 27 insertions(+), 13 deletions(-) diff --git a/flexmeasures/data/services/scheduling.py b/flexmeasures/data/services/scheduling.py index 1507a1c2ab..9d6e10a7d9 100644 --- a/flexmeasures/data/services/scheduling.py +++ b/flexmeasures/data/services/scheduling.py @@ -20,7 +20,7 @@ from isodate import duration_isoformat from rq import get_current_job, Callback from rq.exceptions import InvalidJobOperation, NoSuchJobError -from rq.job import Dependency, Job, JobStatus +from rq.job import Job, JobStatus import timely_beliefs as tb import pandas as pd from sqlalchemy import select @@ -138,6 +138,13 @@ def success_callback(job, connection, result, *args, **kwargs): queue.deferred_job_registry.requeue(dependent_job_ids) +# Meta flag marking a job that should still run when a job it depends on failed, so it can report on that failure. +# We deliberately do not use RQ's Dependency(allow_failure=True) for this: RQ enqueues such a job the moment its +# dependency fails, which would let a wrap-up job run while the failed subjob's fallback job is still pending, +# and report the chain as failed just before the fallback schedules the device after all. +RUNS_ON_CHAIN_FAILURE = "runs_on_chain_failure" + + def _describe_scheduled_device(asset_or_sensor_ref: dict | None) -> str: """Describe the device that a scheduling job was scheduling, for use in a failure message. @@ -264,8 +271,7 @@ def _cascade_failure_to_dependents(job: Job, connection, reason: str) -> None: RQ only enqueues the dependents of a job that succeeded, so a failed job without a fallback would otherwise leave its dependents deferred forever, which leaves a client polling such a job (in particular the wrap-up job of a sequential schedule) without a terminal state or a reason. - Dependent jobs that were set up to tolerate a failing dependency are enqueued rather than failed, so they can run and report on the failure. - RQ itself does that for the dependents of the job that just failed, so here we only need to do it for the jobs that we fail ourselves. + Jobs marked with RUNS_ON_CHAIN_FAILURE are queued rather than failed, so they can run and report on the failure. :param job: The failed job whose dependents should be dealt with. :param connection: Redis connection. @@ -275,14 +281,24 @@ def _cascade_failure_to_dependents(job: Job, connection, reason: str) -> None: dependent_ids = list(job.dependent_ids) if not dependent_ids: return + jobs_that_report_on_the_failure = [] for dependent in Job.fetch_many(dependent_ids, connection=connection): - if dependent is None or dependent.allow_dependency_failures: + if dependent is None: continue if dependent.get_status(refresh=True) != JobStatus.DEFERRED: continue + if dependent.allow_dependency_failures: + continue # RQ enqueues a job that tolerates a failing dependency by itself + if dependent.meta.get(RUNS_ON_CHAIN_FAILURE): + jobs_that_report_on_the_failure.append(dependent) + continue _fail_deferred_job(dependent, reason) _cascade_failure_to_dependents(dependent, connection, reason) - queue.enqueue_dependents(dependent) + + # Only once the rest of the chain has reached a terminal state, let the reporting jobs run, + # so that they see every subjob they report on in its final state. + for dependent in jobs_that_report_on_the_failure: + queue.deferred_job_registry.requeue(dependent.id) def _fail_deferred_job(job: Job, reason: str) -> None: @@ -429,7 +445,7 @@ def create_scheduling_job( def cb_done_sequential_scheduling_job(jobs_ids: list[str]): """Wrap up a chain of sequential scheduling (sub)jobs. - This job also runs when one of the subjobs failed (see the Dependency set up in create_sequential_scheduling_job), + This job also runs when one of the subjobs failed without being rescued by a fallback (see RUNS_ON_CHAIN_FAILURE), in which case it fails, too, naming the devices that could not be scheduled. Its id is what the trigger endpoint hands to the client, so this is what gives that client a terminal state and a reason. @@ -603,16 +619,11 @@ def create_sequential_scheduling_job( previous_sensors.append(sensor) previous_job = job - # create job that triggers when the last job is done, or failed: - # tolerating a failing dependency lets the wrap-up job report which devices could not be scheduled, - # rather than staying deferred forever (see cb_done_sequential_scheduling_job) - depends_on_last_job = ( - Dependency(previous_job, allow_failure=True) if previous_job else None - ) + # create job that triggers when the last job is done job = Job.create( func=cb_done_sequential_scheduling_job, args=([j.id for j in jobs],), - depends_on=depends_on_last_job, + depends_on=previous_job, ttl=int( current_app.config.get( "FLEXMEASURES_JOB_TTL", timedelta(-1) @@ -627,6 +638,9 @@ def create_sequential_scheduling_job( connection=current_app.queues["scheduling"].connection, ) job.meta["asset_or_sensor"] = get_asset_or_sensor_ref(asset) + # This job should also run when a subjob failed, so it can report which devices could not be scheduled + # (see _cascade_failure_to_dependents), instead of staying deferred forever. + job.meta[RUNS_ON_CHAIN_FAILURE] = True job.save_meta() try: From fbfd2bb0474d5256b16298e77018e992c80a05aa Mon Sep 17 00:00:00 2001 From: Mohamed Belhsan Hmida Date: Fri, 7 Aug 2026 14:19:55 +0100 Subject: [PATCH 06/11] tests/scheduling: cover a fallback for the last device in a chain Context: - The existing fallback test fails the first device, whose wrap-up job sits two dependencies away. Nothing covered a fallback for the last device, where the wrap-up job depends on the failing subjob directly, and where an eagerly queued wrap-up job would report the chain as failed while the fallback was still pending. Change: - Added test_create_sequential_jobs_fallback_for_last_device, asserting that the wrap-up job waits for the fallback job and finishes, rather than reporting a failure. Signed-off-by: Mohamed Belhsan Hmida --- .../data/tests/test_scheduling_sequential.py | 63 +++++++++++++++++++ 1 file changed, 63 insertions(+) diff --git a/flexmeasures/data/tests/test_scheduling_sequential.py b/flexmeasures/data/tests/test_scheduling_sequential.py index b22f37cd2b..007669eece 100644 --- a/flexmeasures/data/tests/test_scheduling_sequential.py +++ b/flexmeasures/data/tests/test_scheduling_sequential.py @@ -234,6 +234,69 @@ def test_create_sequential_jobs_fallback( assert deferred_jobs[1].id in finished_jobs +def test_create_sequential_jobs_fallback_for_last_device( + db, app, flex_description_sequential, smart_building +): + """Test the fallback scheduler kicking in for the last device in a chain of sequential scheduling (sub)jobs. + + The wrap-up job depends on that last subjob, so it must wait for the fallback to finish, + rather than concluding that the chain failed while the fallback is still queued. + """ + assets, sensors, _ = smart_building + queue = app.queues["scheduling"] + + start = pd.Timestamp("2015-01-03").tz_localize("Europe/Amsterdam") + end = pd.Timestamp("2015-01-04").tz_localize("Europe/Amsterdam") + + scheduler_specs = { + "module": "flexmeasures.data.models.planning.storage", + "class": "StorageScheduler", + } + + flex_description_sequential["start"] = start + flex_description_sequential["end"] = end + + storage_module = "flexmeasures.data.models.planning.storage" + + with patch(f"{storage_module}.StorageScheduler.persist_flex_model"): + with patch(f"{storage_module}.StorageFallbackScheduler.persist_flex_model"): + # The first device is scheduled fine, the last one is infeasible and falls back + with patch( + f"{storage_module}.StorageScheduler.compute", + side_effect=iter([[], InfeasibleProblemException(), []]), + ): + create_sequential_scheduling_job( + asset=assets["Test Site"], + scheduler_specs=scheduler_specs, + enqueue=True, + force_new_job_creation=True, # otherwise the cache might kick in due to sub-jobs already created in other tests + **flex_description_sequential, + ) + + queued_jobs = queue.jobs + deferred_jobs = sort_jobs( + queue, queue.deferred_job_registry.get_job_ids() + ) + assert len(queued_jobs) == 1 + assert len(deferred_jobs) == 2 + battery_job, wrapup_job = deferred_jobs + + work_on_rq(queue, exc_handler=handle_scheduling_exception) + + finished_jobs = queue.finished_job_registry.get_job_ids() + + # The last subjob failed, but its fallback scheduled the device after all + battery_job.refresh() + assert battery_job.get_status() == "failed" + assert battery_job.meta["fallback_job_id"] in finished_jobs + + # So the chain succeeded, and the wrap-up job should not report a failure + assert wrapup_job.id in finished_jobs, ( + "The wrap-up job should have waited for the fallback job to finish, " + f"but it is {wrapup_job.get_status()}: {failed_job_reason(Job.fetch(wrapup_job.id, connection=queue.connection))}" + ) + + def test_create_sequential_jobs_without_fallback( db, app, flex_description_sequential, smart_building ): From eb3297f52763fe4a9aca3e4b0b5c1a29ebd4954b Mon Sep 17 00:00:00 2001 From: Mohamed Belhsan Hmida Date: Fri, 7 Aug 2026 14:34:25 +0100 Subject: [PATCH 07/11] data/scheduling: keep a failure reportable when the session is unusable Context: - Naming the device that failed costs a database look-up, and the cascade now runs for every failed scheduling job, not just for an infeasible problem. A job that failed on a database error leaves the session needing a rollback, and nothing rolls it back between jobs, so that look-up raises. The failure callback would then abort before cascading, and the chain would wedge again -- precisely for the failures where it matters most. Change: - _describe_scheduled_device falls back to naming the device by its bare reference when the look-up raises a SQLAlchemyError, and logs a warning. Losing the device's name is a small price for still reporting the failure. Signed-off-by: Mohamed Belhsan Hmida --- flexmeasures/data/services/scheduling.py | 13 ++++++++++++- 1 file changed, 12 insertions(+), 1 deletion(-) diff --git a/flexmeasures/data/services/scheduling.py b/flexmeasures/data/services/scheduling.py index 9d6e10a7d9..df7f2a44d5 100644 --- a/flexmeasures/data/services/scheduling.py +++ b/flexmeasures/data/services/scheduling.py @@ -24,6 +24,7 @@ import timely_beliefs as tb import pandas as pd from sqlalchemy import select +from sqlalchemy.exc import SQLAlchemyError from flexmeasures.data import db from flexmeasures.data.models.planning import Scheduler, SchedulerOutputType @@ -148,12 +149,22 @@ def success_callback(job, connection, result, *args, **kwargs): def _describe_scheduled_device(asset_or_sensor_ref: dict | None) -> str: """Describe the device that a scheduling job was scheduling, for use in a failure message. + Naming the device costs a database look-up, which is not something we can count on while handling a failure: + a job that failed on a database error leaves the session needing a rollback, and every query on it raises. + We therefore fall back to naming the device by its bare reference, so that a failure is still reported. + :param asset_or_sensor_ref: Serialized reference to an Asset or Sensor, as stored in a job's meta data. """ if not asset_or_sensor_ref: return "an unknown device" - asset_or_sensor = get_asset_or_sensor_from_ref(asset_or_sensor_ref) kind = asset_or_sensor_ref["class"].lower() + try: + asset_or_sensor = get_asset_or_sensor_from_ref(asset_or_sensor_ref) + except SQLAlchemyError as e: + current_app.logger.warning( + f"Could not look up {kind} {asset_or_sensor_ref['id']} to name it in a scheduling failure message: {e}" + ) + return f"{kind} {asset_or_sensor_ref['id']}" if asset_or_sensor is None: return f"{kind} {asset_or_sensor_ref['id']}" if isinstance(asset_or_sensor, Sensor): From d69dcb9a2b5ab6476652c9c155439411996477a1 Mon Sep 17 00:00:00 2001 From: Mohamed Belhsan Hmida Date: Fri, 7 Aug 2026 14:34:49 +0100 Subject: [PATCH 08/11] tests/scheduling: assert the wrap-up job waits for a pending fallback Context: - The first version of this test only checked the end state, which a single worker reaches correctly even when the wrap-up job is queued too early, because it pops the queue in order and runs the fallback job first. It therefore passed with the bug it was meant to catch still in place. Change: - The test now stops the worker right after the last subjob failed, and asserts that the wrap-up job is still deferred while its fallback job is queued. That is the invariant a second worker would break, and it does fail when the wrap-up job's dependency tolerates failure. - Also added a unit test for _describe_scheduled_device falling back to the bare reference when the session is unusable. Signed-off-by: Mohamed Belhsan Hmida --- .../data/tests/test_scheduling_sequential.py | 55 +++++++++++++++++-- 1 file changed, 49 insertions(+), 6 deletions(-) diff --git a/flexmeasures/data/tests/test_scheduling_sequential.py b/flexmeasures/data/tests/test_scheduling_sequential.py index 007669eece..685f3fd4f0 100644 --- a/flexmeasures/data/tests/test_scheduling_sequential.py +++ b/flexmeasures/data/tests/test_scheduling_sequential.py @@ -3,7 +3,11 @@ import pandas as pd from rq.job import Job -from flexmeasures.data.services.scheduling import create_sequential_scheduling_job +from sqlalchemy.exc import PendingRollbackError +from flexmeasures.data.services.scheduling import ( + _describe_scheduled_device, + create_sequential_scheduling_job, +) from flexmeasures.utils.job_utils import work_on_rq from flexmeasures.data.services.scheduling import handle_scheduling_exception from flexmeasures.data.services.utils import failed_job_reason, sort_jobs @@ -234,13 +238,38 @@ def test_create_sequential_jobs_fallback( assert deferred_jobs[1].id in finished_jobs +def test_describe_scheduled_device_survives_an_unusable_session( + db, app, smart_building +): + """Naming the device must not raise when the database session is unusable. + + A job that failed on a database error leaves the session needing a rollback. If naming its device raised, + the failure would never be cascaded to the dependent jobs, and the chain would wedge after all. + """ + _, sensors, _ = smart_building + sensor = sensors["Test EV"] + reference = {"id": sensor.id, "class": "Sensor"} + + assert ( + _describe_scheduled_device(reference) + == f"sensor {sensor.id} ({sensor.generic_asset.name} - {sensor.name})" + ) + + with patch( + "flexmeasures.data.services.scheduling.get_asset_or_sensor_from_ref", + side_effect=PendingRollbackError("session needs rollback", None, None), + ): + assert _describe_scheduled_device(reference) == f"sensor {sensor.id}" + + def test_create_sequential_jobs_fallback_for_last_device( db, app, flex_description_sequential, smart_building ): """Test the fallback scheduler kicking in for the last device in a chain of sequential scheduling (sub)jobs. - The wrap-up job depends on that last subjob, so it must wait for the fallback to finish, - rather than concluding that the chain failed while the fallback is still queued. + The wrap-up job depends on that last subjob directly, so it must stay deferred while the fallback job is pending. + Were it queued alongside the fallback job, a second worker could run it right away, find a device without a schedule, + and report the chain as failed just before the fallback schedules that device after all. """ assets, sensors, _ = smart_building queue = app.queues["scheduling"] @@ -281,14 +310,28 @@ def test_create_sequential_jobs_fallback_for_last_device( assert len(deferred_jobs) == 2 battery_job, wrapup_job = deferred_jobs + # Work until the last subjob has failed and triggered its fallback, but no further + work_on_rq(queue, exc_handler=handle_scheduling_exception, max_jobs=2) + + battery_job.refresh() + wrapup_job.refresh() + fallback_job_id = battery_job.meta["fallback_job_id"] + assert battery_job.get_status() == "failed" + assert fallback_job_id in [job.id for job in queue.jobs] + + # The wrap-up job must not be runnable while the fallback job is still pending + assert wrapup_job.get_status() == "deferred", ( + "The wrap-up job should still be waiting for the fallback job, " + f"but it is {wrapup_job.get_status()}." + ) + + # Now let the fallback job (and, after it, the wrap-up job) run work_on_rq(queue, exc_handler=handle_scheduling_exception) finished_jobs = queue.finished_job_registry.get_job_ids() # The last subjob failed, but its fallback scheduled the device after all - battery_job.refresh() - assert battery_job.get_status() == "failed" - assert battery_job.meta["fallback_job_id"] in finished_jobs + assert fallback_job_id in finished_jobs # So the chain succeeded, and the wrap-up job should not report a failure assert wrapup_job.id in finished_jobs, ( From c790cb92b82bab39aae1282a2a1d67320eee5e8f Mon Sep 17 00:00:00 2001 From: Mohamed Belhsan Hmida Date: Fri, 7 Aug 2026 14:45:52 +0100 Subject: [PATCH 09/11] docs/changelog: reference PR #2409 in the changelog entry Context: - The entry was added before the PR existed, so it carried a placeholder number and also linked the issue, where entries in this changelog normally reference the PR only. Change: - Corrected the link to PR #2409, and dropped the issue reference. Signed-off-by: Mohamed Belhsan Hmida --- documentation/changelog.rst | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/documentation/changelog.rst b/documentation/changelog.rst index 57d007b323..2dfab52dfd 100644 --- a/documentation/changelog.rst +++ b/documentation/changelog.rst @@ -87,7 +87,7 @@ Infrastructure / Support Bugfixes ----------- -* A sequential schedule no longer gets stuck when one of its devices fails to be scheduled by a scheduler without a fallback (such as the ``ProcessScheduler``, or a custom scheduler): the failure now cascades to the remaining subjobs and to the wrap-up job whose id the trigger endpoint returns, so a client polling that job gets a terminal failure naming the device that could not be scheduled, instead of a job that stays deferred forever [see `PR #2406 `_ and `issue #2404 `_] +* A sequential schedule no longer gets stuck when one of its devices fails to be scheduled by a scheduler without a fallback (such as the ``ProcessScheduler``, or a custom scheduler): the failure now cascades to the remaining subjobs and to the wrap-up job whose id the trigger endpoint returns, so a client polling that job gets a terminal failure naming the device that could not be scheduled, instead of a job that stays deferred forever [see `PR #2409 `_] * Show icons for more asset types in the UI's asset structure view, which previously fell back to a question mark: the ``wind``, ``process`` and ``heat-storage`` types that FlexMeasures seeds by default, and EV infrastructure under its various names (such as ``one-way_evse``, ``two-way_evse``, ``evse``, ``charging_station`` and ``charging_hub``) and building services equipment (``hvac``, ``ahu``, ``dhw``, ``heatpump``, ``chiller``, ``lighting`` and ``other-loads``). Asset type names are now matched ignoring case and separators, so an asset type named ``charge-point`` gets the same icon as ``chargepoint`` [see `PR #2391 `_] * Replaying a chart for a past window no longer shows annotations that were only recorded later; annotation searches and the ``chart_annotations`` endpoints can now be scoped by recording (belief) time [see `PR #2367 `_] * Continuing the query-parameter cleanup started in PR #2352: the chart-related endpoints now use ``prior``, ``start``, ``end`` and hyphenated field names, with a new ``duration`` field to derive a missing ``start``/``end``; old spellings keep working as legacy aliases [see `PR #2367 `_] From f643bf6381d2ac18efee6d6fc93ad58f3e1322cf Mon Sep 17 00:00:00 2001 From: Mohamed Belhsan Hmida Date: Fri, 7 Aug 2026 20:03:59 +0100 Subject: [PATCH 10/11] tests/scheduling: adapt the cascade tests to the retired storage fallback Context: - Merging main brought in PR #2252, which retires the storage fallback scheduler. Three things broke or went stale as a result. - test_create_sequential_jobs_without_storage_fallback, added by #2252, asserts that the deferred subjobs stay deferred, and clears them so they do not leak into the next test. It kept passing here only because its assertions are "not in finished_jobs", which a cascaded (failed) job also satisfies, so it silently asserted the opposite of what this branch establishes. - The fallback race test patched StorageFallbackScheduler, which #2252 deleted, so it failed on import. - The end-to-end API test relied on a soc-target the device could not reach, which #2252 turns into a priced breach rather than a failure, so the chain succeeded and the test no longer exercised a failure at all. Change: - Removed test_create_sequential_jobs_without_storage_fallback, which test_create_sequential_jobs_without_fallback supersedes with stronger assertions, along with the cleanup block the cascade makes unnecessary. - The fallback race test now makes the storage scheduler stand in as its own fallback, which is the situation a custom scheduler that defines one is still in. - The API test now provokes a genuine infeasibility with a soc-usage above the device's power-capacity, which stays a hard constraint, and both tests assert up front that the scheduler really has no fallback rather than patching one away. Signed-off-by: Mohamed Belhsan Hmida --- documentation/changelog.rst | 2 +- .../tests/test_asset_schedules_fresh_db.py | 113 +++++++-------- .../data/tests/test_scheduling_sequential.py | 137 ++++-------------- 3 files changed, 88 insertions(+), 164 deletions(-) diff --git a/documentation/changelog.rst b/documentation/changelog.rst index 3fc2296f90..6a37a9013f 100644 --- a/documentation/changelog.rst +++ b/documentation/changelog.rst @@ -103,7 +103,7 @@ Infrastructure / Support Bugfixes ----------- -* A sequential schedule no longer gets stuck when one of its devices fails to be scheduled by a scheduler without a fallback (such as the ``ProcessScheduler``, or a custom scheduler): the failure now cascades to the remaining subjobs and to the wrap-up job whose id the trigger endpoint returns, so a client polling that job gets a terminal failure naming the device that could not be scheduled, instead of a job that stays deferred forever [see `PR #2409 `_] +* A sequential schedule no longer gets stuck when one of its devices fails to be scheduled by a scheduler without a fallback, which since the retirement of the storage fallback scheduler is every scheduler that does not define one itself: the failure now cascades to the remaining subjobs and to the wrap-up job whose id the trigger endpoint returns, so a client polling that job gets a terminal failure naming the device that could not be scheduled, instead of a job that stays deferred forever [see `PR #2409 `_] * In a multi-device flex-model, a device without a stock (e.g. a converter port or curtailable generator) silently disabled constraint validation for all devices after it; validation now covers every device, and also newly checks that each device's power bounds do not contradict each other, so a contradictory hard bound fails with a clear per-time-step message instead of a bare solver infeasibility [see `PR #2252 `_] * The scheduler now rejects a commitment that no constraint would bind — a stock commitment naming no device or known stock group, or a commodity commitment for a commodity that no commitment maps devices to — instead of silently dropping it from the problem, or letting a favourably priced deviation make the problem unbounded; the error names the commitment [see `PR #2410 `_ and `PR #2413 `_] * Show icons for more asset types in the UI's asset structure view, which previously fell back to a question mark: the ``wind``, ``process`` and ``heat-storage`` types that FlexMeasures seeds by default, and EV infrastructure under its various names (such as ``one-way_evse``, ``two-way_evse``, ``evse``, ``charging_station`` and ``charging_hub``) and building services equipment (``hvac``, ``ahu``, ``dhw``, ``heatpump``, ``chiller``, ``lighting`` and ``other-loads``). Asset type names are now matched ignoring case and separators, so an asset type named ``charge-point`` gets the same icon as ``chargepoint`` [see `PR #2391 `_] diff --git a/flexmeasures/api/v3_0/tests/test_asset_schedules_fresh_db.py b/flexmeasures/api/v3_0/tests/test_asset_schedules_fresh_db.py index 772ecdb8c4..933c87497d 100644 --- a/flexmeasures/api/v3_0/tests/test_asset_schedules_fresh_db.py +++ b/flexmeasures/api/v3_0/tests/test_asset_schedules_fresh_db.py @@ -1,7 +1,5 @@ from __future__ import annotations -from unittest.mock import patch - from flask import url_for import pytest from isodate import parse_datetime, parse_duration @@ -20,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 @@ -1115,13 +1114,16 @@ def test_asset_sequential_schedule_without_fallback_fails_terminally( ): """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, so it cannot reach a target below its initial SoC + # 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[ @@ -1145,7 +1147,8 @@ def test_asset_sequential_schedule_without_fallback_fails_terminally( "soc-at-start": 10, "soc-min": 0, "soc-max": 40, - "soc-targets": [{"value": 9, "datetime": "2015-01-02T02:00:00+01:00"}], + "power-capacity": "1 MW", + "soc-usage": ["10 MW"], }, { "sensor": feasible_sensor.id, @@ -1160,56 +1163,52 @@ def test_asset_sequential_schedule_without_fallback_fails_terminally( deferred_registry = app.queues["scheduling"].deferred_job_registry jobs_deferred_by_other_tests = set(deferred_registry.get_job_ids()) - storage_module = "flexmeasures.data.models.planning.storage" - with patch(f"{storage_module}.StorageScheduler.fallback_scheduler_class", None): - 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, - ) + 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, + ) diff --git a/flexmeasures/data/tests/test_scheduling_sequential.py b/flexmeasures/data/tests/test_scheduling_sequential.py index f69686bc79..1f8d42ed0b 100644 --- a/flexmeasures/data/tests/test_scheduling_sequential.py +++ b/flexmeasures/data/tests/test_scheduling_sequential.py @@ -11,6 +11,7 @@ from flexmeasures.utils.job_utils import work_on_rq from flexmeasures.data.services.scheduling import handle_scheduling_exception from flexmeasures.data.services.utils import failed_job_reason, sort_jobs +from flexmeasures.data.models.planning.storage import StorageScheduler from flexmeasures.data.models.time_series import Sensor @@ -161,87 +162,6 @@ def test_create_sequential_jobs(db, app, flex_description_sequential, smart_buil # ) -def test_create_sequential_jobs_without_storage_fallback( - db, app, flex_description_sequential, smart_building -): - """Test an infeasible first subjob in a chain of sequential scheduling jobs. - - Checks that no storage fallback job is created. The deferred subjobs should remain - deferred because the first subjob failed. - """ - assets, sensors, _ = smart_building - queue = app.queues["scheduling"] - - start = pd.Timestamp("2015-01-03").tz_localize("Europe/Amsterdam") - end = pd.Timestamp("2015-01-04").tz_localize("Europe/Amsterdam") - - scheduler_specs = { - "module": "flexmeasures.data.models.planning.storage", - "class": "StorageScheduler", - } - - flex_description_sequential["start"] = start - flex_description_sequential["end"] = end - - storage_module = "flexmeasures.data.models.planning.storage" - - with patch(f"{storage_module}.StorageScheduler.persist_flex_model"): - with patch( - f"{storage_module}.StorageScheduler.compute", - side_effect=InfeasibleProblemException(), - ): - create_sequential_scheduling_job( - asset=assets["Test Site"], - scheduler_specs=scheduler_specs, - enqueue=True, - force_new_job_creation=True, # otherwise the cache might kick in due to sub-jobs already created in other tests - **flex_description_sequential, - ) - - # There should be 3 jobs: - # 2 jobs scheduling the 2 flexible devices in the flex-model, plus 1 'done job' to wrap things up - queued_jobs = app.queues["scheduling"].jobs - deferred_jobs = [ - Job.fetch(job_id, connection=queue.connection) - for job_id in app.queues[ - "scheduling" - ].deferred_job_registry.get_job_ids() - ] - # Sort deferred_jobs by their created_at attribute - deferred_jobs = sorted(deferred_jobs, key=lambda job: job.created_at) - assert ( - len(queued_jobs) == 1 - ), "Only the job for scheduling the first device sequentially should be queued." - assert ( - len(deferred_jobs) == 2 - ), "The job for scheduling the second device, and the wrap-up job, should be deferred." - - # Work on jobs - work_on_rq(queue, exc_handler=handle_scheduling_exception) - - for job in queued_jobs: - job.refresh() - for job in deferred_jobs: - job.refresh() - - finished_jobs = queue.finished_job_registry.get_job_ids() - failed_jobs = queue.failed_job_registry.get_job_ids() - - # Original job failed and no fallback job was created - assert queued_jobs[0].id in failed_jobs - assert queued_jobs[0].meta.get("fallback_job_id") is None - - # The deferred jobs should not run when their dependency fails without fallback - assert deferred_jobs[0].id not in finished_jobs - assert deferred_jobs[1].id not in finished_jobs - - # Without a fallback to unblock the chain, the deferred subjobs stay deferred - # for good, so clear them here rather than leaking them into the next test. - for deferred_job_id in queue.deferred_job_registry.get_job_ids(): - queue.deferred_job_registry.remove(deferred_job_id) - queue.empty() - - def test_describe_scheduled_device_survives_an_unusable_session( db, app, smart_building ): @@ -292,7 +212,12 @@ def test_create_sequential_jobs_fallback_for_last_device( storage_module = "flexmeasures.data.models.planning.storage" with patch(f"{storage_module}.StorageScheduler.persist_flex_model"): - with patch(f"{storage_module}.StorageFallbackScheduler.persist_flex_model"): + # No scheduler ships with a fallback since PR #2252, so let the storage scheduler stand in as its own, + # which is the situation a custom scheduler that does define a fallback is still in. + with patch( + f"{storage_module}.StorageScheduler.fallback_scheduler_class", + StorageScheduler, + ): # The first device is scheduled fine, the last one is infeasible and falls back with patch( f"{storage_module}.StorageScheduler.compute", @@ -349,7 +274,7 @@ def test_create_sequential_jobs_without_fallback( ): """Test that a failing subjob without a fallback scheduler does not wedge the chain. - The first device is infeasible, and its scheduler has no fallback. The remaining subjobs can then never run, + The first device is infeasible, and no scheduler defines a fallback since PR #2252. The remaining subjobs can then never run, so they should be failed rather than left deferred, and the wrap-up job — whose id is what the trigger endpoint hands to the client — should reach a terminal failed state naming the device that could not be scheduled. """ @@ -369,32 +294,32 @@ def test_create_sequential_jobs_without_fallback( storage_module = "flexmeasures.data.models.planning.storage" + assert ( + StorageScheduler.fallback_scheduler_class is None + ), "This test needs a scheduler without a fallback." + with patch(f"{storage_module}.StorageScheduler.persist_flex_model"): - # Retire the fallback scheduler, like ProcessScheduler and custom schedulers do by default - with patch(f"{storage_module}.StorageScheduler.fallback_scheduler_class", None): - with patch( - f"{storage_module}.StorageScheduler.compute", - side_effect=iter([InfeasibleProblemException(), [], []]), - ): - create_sequential_scheduling_job( - asset=assets["Test Site"], - scheduler_specs=scheduler_specs, - enqueue=True, - force_new_job_creation=True, # otherwise the cache might kick in due to sub-jobs already created in other tests - **flex_description_sequential, - ) + with patch( + f"{storage_module}.StorageScheduler.compute", + side_effect=iter([InfeasibleProblemException(), [], []]), + ): + create_sequential_scheduling_job( + asset=assets["Test Site"], + scheduler_specs=scheduler_specs, + enqueue=True, + force_new_job_creation=True, # otherwise the cache might kick in due to sub-jobs already created in other tests + **flex_description_sequential, + ) - queued_jobs = queue.jobs - deferred_jobs = sort_jobs( - queue, queue.deferred_job_registry.get_job_ids() - ) - assert len(queued_jobs) == 1 - assert len(deferred_jobs) == 2 - ev_job = queued_jobs[0] - battery_job, wrapup_job = deferred_jobs + queued_jobs = queue.jobs + deferred_jobs = sort_jobs(queue, queue.deferred_job_registry.get_job_ids()) + assert len(queued_jobs) == 1 + assert len(deferred_jobs) == 2 + ev_job = queued_jobs[0] + battery_job, wrapup_job = deferred_jobs - # Work on jobs - work_on_rq(queue, exc_handler=handle_scheduling_exception) + # Work on jobs + work_on_rq(queue, exc_handler=handle_scheduling_exception) failed_jobs = queue.failed_job_registry.get_job_ids() From db1f099e925349da3b4a6fa43eeea226331db4a5 Mon Sep 17 00:00:00 2001 From: Felix Claessen <30658763+Flix6x@users.noreply.github.com> Date: Thu, 3 Sep 2026 16:57:42 +0200 Subject: [PATCH 11/11] Apply batched suggestions from code review Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com> Signed-off-by: Felix Claessen <30658763+Flix6x@users.noreply.github.com> --- flexmeasures/data/services/scheduling.py | 8 +++++--- flexmeasures/data/tests/test_scheduling_sequential.py | 6 ++++-- 2 files changed, 9 insertions(+), 5 deletions(-) diff --git a/flexmeasures/data/services/scheduling.py b/flexmeasures/data/services/scheduling.py index 3c7bb7d273..65f1c318c8 100644 --- a/flexmeasures/data/services/scheduling.py +++ b/flexmeasures/data/services/scheduling.py @@ -139,9 +139,11 @@ def success_callback(job, connection, result, *args, **kwargs): queue.deferred_job_registry.requeue(dependent_job_ids) -# Meta flag marking a job that should still run when a job it depends on failed, so it can report on that failure. -# We deliberately do not use RQ's Dependency(allow_failure=True) for this: RQ enqueues such a job the moment its -# dependency fails, which would let a wrap-up job run while the failed subjob's fallback job is still pending, +# Meta flag marking a job that should still run when a job it depends on failed, +# so it can report on that failure. +# We deliberately do not use RQ's Dependency(allow_failure=True) for this. +# RQ enqueues such a job the moment its dependency fails, +# which would let a wrap-up job run while the failed subjob's fallback job is still pending, # and report the chain as failed just before the fallback schedules the device after all. RUNS_ON_CHAIN_FAILURE = "runs_on_chain_failure" diff --git a/flexmeasures/data/tests/test_scheduling_sequential.py b/flexmeasures/data/tests/test_scheduling_sequential.py index 1f8d42ed0b..386cfe14d7 100644 --- a/flexmeasures/data/tests/test_scheduling_sequential.py +++ b/flexmeasures/data/tests/test_scheduling_sequential.py @@ -274,8 +274,10 @@ def test_create_sequential_jobs_without_fallback( ): """Test that a failing subjob without a fallback scheduler does not wedge the chain. - The first device is infeasible, and no scheduler defines a fallback since PR #2252. The remaining subjobs can then never run, - so they should be failed rather than left deferred, and the wrap-up job — whose id is what the trigger endpoint hands to the client — + The first device is infeasible, and no scheduler defines a fallback since PR #2252. + The remaining subjobs can then never run, + so they should be failed rather than left deferred. + The wrap-up job, whose id is what the trigger endpoint hands to the client, should reach a terminal failed state naming the device that could not be scheduled. """ assets, sensors, _ = smart_building