diff --git a/documentation/api/change_log.rst b/documentation/api/change_log.rst index bb98ba154a..72007b2c9b 100644 --- a/documentation/api/change_log.rst +++ b/documentation/api/change_log.rst @@ -5,14 +5,13 @@ API change log .. note:: The FlexMeasures API follows its own versioning scheme. This is also reflected in the URL (e.g. `/api/v3_0`), allowing developers to upgrade at their own pace. -v3.0-34 | September 2, 2026 -""""""""""""""""""""""""""" -- Added ``POST /api/v3_0/assets//automations//trigger``, to run one automation now, once, on top of its recurring runs. The response is the standard job response, extended with ``n_jobs``: how many jobs the run queued. An on-demand run does not affect the automation's recurrence, and inactive automations can be triggered, too. Triggering requires the same permission as writing data under the asset, and falls under the stricter rate limit that the other triggering endpoints share. -v3.0-33 | September 1, 2026 +v3.0-33 | September 3, 2026 """"""""""""""""""""""""""" +- Added ``POST /api/v3_0/assets//automations//trigger``, to run one automation now, once, on top of its recurring runs. The response is the standard job response, extended with ``n_jobs``: how many jobs the run queued. An on-demand run does not affect the automation's recurrence, and inactive automations can be triggered, too. Triggering requires the same permission as writing data under the asset, and falls under the stricter rate limit that the other triggering endpoints share. - Added ``GET /api/v3_0/assets//automations`` and ``GET /api/v3_0/assets//automations/`` for listing and inspecting forecast automations, including the sensors an automation reads from and writes to. Each automation shows the IANA ``timezone`` in which its cron expression is interpreted, and a ``cursor``: the offset-aware UTC time of the most recent run it committed to. The cursor advances just before queueing, so it does not indicate that queueing or the forecast itself succeeded. Asset job entries now include ``created_via`` provenance; automation identity is included only when the caller may read that automation. - Added ``GET /api/v3_0/sources/`` to show the full record of one data source, including the attributes in which data generators store their configuration. +- 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-32 | August 11, 2026 """"""""""""""""""""""""" diff --git a/documentation/api/introduction.rst b/documentation/api/introduction.rst index f13bfc69f4..fd2ccc72b4 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 8100b4af8a..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 @@ -6,7 +6,7 @@ from numpy.testing import assert_almost_equal import pandas as pd -from rq.job import Job +from rq.job import Job, JobStatus from flexmeasures import Sensor from flexmeasures.api.v3_0.tests.utils import message_for_trigger_schedule @@ -18,6 +18,7 @@ handle_scheduling_exception, get_data_source_for_job, ) +from flexmeasures.data.models.planning.storage import StorageScheduler from flexmeasures.data.services.utils import sort_jobs from flexmeasures.utils.unit_utils import ur @@ -1098,3 +1099,116 @@ def test_asset_trigger_with_group_referencing_sensor_outside_asset_tree( # No scheduling job should have been queued assert len(app.queues["scheduling"]) == 0 + + +@pytest.mark.parametrize( + "requesting_user", ["test_prosumer_user@seita.nl"], indirect=True +) +def test_asset_sequential_schedule_without_fallback_fails_terminally( + app, + add_market_prices_fresh_db, + setup_roles_users_fresh_db, + add_charging_station_assets_fresh_db, + keep_scheduling_queue_empty, + requesting_user, +): + """Trigger a sequential schedule whose first device is infeasible, using a scheduler without a fallback. + + No scheduler defines a fallback since PR #2252, so the storage scheduler is used as it comes. + The job id handed to the client is the one of the wrap-up job. Polling it should yield a terminal failure, + with a reason naming the device that could not be scheduled, rather than a job that stays deferred forever. + Re-triggering the same request should not hand back a job that is still waiting on that chain, either. + """ + price_sensor_id = add_market_prices_fresh_db["epex_da"].id + + # The uni-directional charging station cannot discharge, and cannot charge faster than its power capacity, + # so a usage above that capacity cannot be met. SoC bounds and targets are relaxed by default since PR #2252, + # but a device's power capacity stays hard, so this is a genuine infeasibility rather than a priced breach. + charging_station = add_charging_station_assets_fresh_db["Test charging station"] + infeasible_sensor = charging_station.sensors[0] + bidirectional_charging_station = add_charging_station_assets_fresh_db[ + "Test charging station (bidirectional)" + ] + feasible_sensor = bidirectional_charging_station.sensors[0] + + message = { + "start": "2015-01-02T00:00:00+01:00", + "duration": "PT24H", + "resolution": "PT15M", + "sequential": True, + "flex-context": { + "consumption-price": {"sensor": price_sensor_id}, + "production-price": {"sensor": price_sensor_id}, + "site-power-capacity": "1 TW", + }, + "flex-model": [ + { + "sensor": infeasible_sensor.id, + "soc-at-start": 10, + "soc-min": 0, + "soc-max": 40, + "power-capacity": "1 MW", + "soc-usage": ["10 MW"], + }, + { + "sensor": feasible_sensor.id, + "soc-at-start": 10, + "soc-min": 0, + "soc-max": 40, + }, + ], + } + site_id = charging_station.parent_asset.id + + deferred_registry = app.queues["scheduling"].deferred_job_registry + jobs_deferred_by_other_tests = set(deferred_registry.get_job_ids()) + + assert ( + StorageScheduler.fallback_scheduler_class is None + ), "This test needs a scheduler without a fallback." + + with app.test_client() as client: + trigger_schedule_response = client.post( + url_for("AssetAPI:trigger_schedule", id=site_id), + json=message, + ) + assert trigger_schedule_response.status_code == 202 + job_id = trigger_schedule_response.json["job"] + + # The subjob for the second device, and the wrap-up job, wait for the first device to be scheduled + deferred_jobs_of_this_chain = ( + set(deferred_registry.get_job_ids()) - jobs_deferred_by_other_tests + ) + assert len(deferred_jobs_of_this_chain) == 2 + + work_on_rq(app.queues["scheduling"], exc_handler=handle_scheduling_exception) + + # Polling the job we were handed gives a terminal failure, naming the device that could not be scheduled + job_status_response = client.get(url_for("JobAPI:get_job_status", uuid=job_id)) + print("Server responded with:\n%s" % job_status_response.json) + assert job_status_response.status_code == 422 + assert job_status_response.json["status"] == "FAILED" + message_to_client = job_status_response.json["message"] + assert ( + f"sensor {infeasible_sensor.id} ({charging_station.name} - {infeasible_sensor.name})" + in message_to_client + ) + assert "InfeasibleProblemException" in message_to_client + + # No job is left waiting on a chain that will never complete + assert deferred_jobs_of_this_chain.isdisjoint(deferred_registry.get_job_ids()) + + # Re-triggering the same request does not hand back a job that is still waiting on that chain + retrigger_response = client.post( + url_for("AssetAPI:trigger_schedule", id=site_id), + json=message, + ) + assert retrigger_response.status_code == 202 + retriggered_job = Job.fetch( + retrigger_response.json["job"], + connection=app.queues["scheduling"].connection, + ) + assert retriggered_job.get_status(refresh=True) not in ( + JobStatus.DEFERRED, + JobStatus.SCHEDULED, + ) 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 d169924255..65f1c318c8 100644 --- a/flexmeasures/data/services/scheduling.py +++ b/flexmeasures/data/services/scheduling.py @@ -19,11 +19,12 @@ 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 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 @@ -32,7 +33,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 +46,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 +139,193 @@ 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), - **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, + ) + + # 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. + + 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) - job.meta["fallback_job_id"] = fallback_job.id - job.save_meta() - current_app.queues["scheduling"].enqueue_job(fallback_job) + +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 +456,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. """ - 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: @@ -466,6 +651,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: diff --git a/flexmeasures/data/tests/test_scheduling_sequential.py b/flexmeasures/data/tests/test_scheduling_sequential.py index f219c62b60..386cfe14d7 100644 --- a/flexmeasures/data/tests/test_scheduling_sequential.py +++ b/flexmeasures/data/tests/test_scheduling_sequential.py @@ -3,9 +3,15 @@ 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 +from flexmeasures.data.models.planning.storage import StorageScheduler from flexmeasures.data.models.time_series import Sensor @@ -156,13 +162,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"] @@ -180,10 +211,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"], @@ -193,48 +313,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(