diff --git a/documentation/api/change_log.rst b/documentation/api/change_log.rst index 686db20ff0..f35c2ded62 100644 --- a/documentation/api/change_log.rst +++ b/documentation/api/change_log.rst @@ -8,6 +8,7 @@ API change log v3.0-35 | September 9, 2026 """"""""""""""""""""""""""" - The ``resolution`` field is now rejected with a ``422 (Unprocessable Entity)`` response unless it spans a positive amount of time. This applies wherever the API accepts one: as a query parameter on ``GET /api/v3_0/sensors//data`` and on the ``chart_data`` endpoints under ``api/dev``, and in the request body of the ``POST`` schedule trigger endpoints. Previously, a zero resolution (such as ``PT0S``) either crashed the request with a ``500`` or was silently ignored, and a negative resolution returned an empty set of values. +- 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. v3.0-34 | September 2, 2026 """"""""""""""""""""""""""" diff --git a/documentation/api/introduction.rst b/documentation/api/introduction.rst index c2e19f4fdb..775a2172de 100644 --- a/documentation/api/introduction.rst +++ b/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 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 9b47df2491..e6520cba79 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 @@ -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 @@ -1156,3 +1157,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, + ) 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 a2499732bf..b9c7cafb9f 100644 --- a/flexmeasures/data/services/scheduling.py +++ b/flexmeasures/data/services/scheduling.py @@ -20,11 +20,12 @@ from isodate import duration_isoformat from marshmallow import ValidationError 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 Job, JobStatus 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 @@ -33,7 +34,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 @@ -43,6 +47,7 @@ from flexmeasures.data.utils import 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, @@ -135,81 +140,201 @@ 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. + + 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" + 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): + 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 - scheduler_kwargs = job.meta["scheduler_kwargs"] + # 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 - # 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 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.", + ) + + +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), - trigger=job.meta.get("trigger"), - **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() - - job.meta["fallback_job_id"] = fallback_job.id - job.save_meta() - current_app.queues["scheduling"].enqueue_job(fallback_job) - asset_or_sensor_ref = get_asset_or_sensor_ref(asset_or_sensor) - current_app.job_cache.add( - asset_or_sensor_ref["id"], - fallback_job.id, - queue="scheduling", - asset_or_sensor_type=asset_or_sensor_ref["class"].lower(), - ) + fallback_job = create_scheduling_job( + asset_or_sensor, + force_new_job_creation=True, + enqueue=False, + scheduler_specs=scheduler_specs, + success_callback=Callback(success_callback), + trigger=job.meta.get("trigger"), + **scheduler_kwargs, + ) + + # 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) + asset_or_sensor_ref = get_asset_or_sensor_ref(asset_or_sensor) + current_app.job_cache.add( + asset_or_sensor_ref["id"], + fallback_job.id, + queue="scheduling", + asset_or_sensor_type=asset_or_sensor_ref["class"].lower(), + ) + 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. + + 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. + :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 + jobs_that_report_on_the_failure = [] + for dependent in Job.fetch_many(dependent_ids, connection=connection): + 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) + + # 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: + """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") @@ -348,11 +473,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 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. + 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. + """ + 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. """ - current_app.logger.info("Sequential scheduling job finished its chain of subjobs.") - # jobs = [Job.fetch(job_id) for job_id in jobs_ids] + 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: @@ -540,6 +725,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 if trigger: job.meta["trigger"] = trigger job.save_meta() diff --git a/flexmeasures/data/tests/test_scheduling_sequential.py b/flexmeasures/data/tests/test_scheduling_sequential.py index ba3051eccd..0296d2901b 100644 --- a/flexmeasures/data/tests/test_scheduling_sequential.py +++ b/flexmeasures/data/tests/test_scheduling_sequential.py @@ -4,11 +4,17 @@ import pandas as pd from rq.job import Job from sqlalchemy import select +from sqlalchemy.exc import PendingRollbackError from flexmeasures.data.models.data_sources import DataSource -from flexmeasures.data.services.scheduling import create_sequential_scheduling_job +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 +from flexmeasures.data.models.planning.storage import StorageScheduler from flexmeasures.data.models.time_series import Sensor @@ -216,13 +222,38 @@ def test_create_sequential_jobs(db, app, flex_description_sequential, smart_buil # ) -def test_create_sequential_jobs_without_storage_fallback( +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 an infeasible first subjob in a chain of sequential scheduling jobs. + """Test the fallback scheduler kicking in for the last device in a chain of sequential scheduling (sub)jobs. - Checks that no storage fallback job is created. The deferred subjobs should remain - deferred because the first subjob failed. + 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"] @@ -240,10 +271,99 @@ def test_create_sequential_jobs_without_storage_fallback( storage_module = "flexmeasures.data.models.planning.storage" + with patch(f"{storage_module}.StorageScheduler.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", + 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 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 + 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, ( + "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 +): + """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. + 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" + + assert ( + StorageScheduler.fallback_scheduler_class is None + ), "This test needs a scheduler without a fallback." + with patch(f"{storage_module}.StorageScheduler.persist_flex_model"): with patch( f"{storage_module}.StorageScheduler.compute", - side_effect=InfeasibleProblemException(), + side_effect=iter([InfeasibleProblemException(), [], []]), ): create_sequential_scheduling_job( asset=assets["Test Site"], @@ -253,48 +373,36 @@ def test_create_sequential_jobs_without_storage_fallback( **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." + 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) - for job in queued_jobs: - job.refresh() - for job in deferred_jobs: - job.refresh() + failed_jobs = queue.failed_job_registry.get_job_ids() - finished_jobs = queue.finished_job_registry.get_job_ids() - 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 - # 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 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 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 + # 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 - # 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() + # 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(