diff --git a/documentation/changelog.rst b/documentation/changelog.rst index b51fcfe824..a51cc054d0 100644 --- a/documentation/changelog.rst +++ b/documentation/changelog.rst @@ -22,6 +22,8 @@ Bugfixes * Hovering anywhere over a session in the Charge Points chart shows its tooltip again, listing all of the session's timestamps, as it did before the move to ECharts, without highlighting another charger's session; also, a chart's tooltip no longer shows a point from a neighbouring subplot [see `PR #2829 `_] +* Handle each failed job of a worker on several queues, such as one started with ``flexmeasures jobs run-worker --queue "scheduling|forecasting"``, with the exception handler of the job's own queue, so a failed scheduling job shows its traceback in the worker log [see `PR #2850 `_] + v1.1.0 | November 5, 2026 ============================ diff --git a/flexmeasures/cli/jobs.py b/flexmeasures/cli/jobs.py index 6b57cd73e7..96e732d84a 100644 --- a/flexmeasures/cli/jobs.py +++ b/flexmeasures/cli/jobs.py @@ -553,7 +553,8 @@ def run_worker(queue: str, name: str | None, with_scheduler: bool): while used_name in worker_names: used_name = f"{name}-{next(name_suffixes)}" - error_handler = get_exception_handler(queue) + # A worker can work on several queues, so each failed job goes to the handler of its own queue. + error_handler = handle_exception_by_queue # On macOS: RQ's fork-based Worker triggers a known OpenSSL/psycopg2 # segmentation fault due to reinitialization of SSL state in forked children. @@ -1008,6 +1009,20 @@ def get_exception_handler(queue_name: str) -> Callable: return QUEUE_EXCEPTION_HANDLERS.get(queue_name, handle_worker_exception) +def handle_exception_by_queue( + job: Job, + exc_type: Type[Exception], + exc_value: Exception, + traceback: TracebackType, +) -> bool | None: + """Handle a failed job with the exception handler of the queue the job came from. + + A worker that works on several queues would otherwise use one handler for the jobs of all of them. + RQ reads the return value to decide whether to call the next handler, so we pass it on. + """ + return get_exception_handler(job.origin)(job, exc_type, exc_value, traceback) + + def parse_queue_list(queue_names_str: str) -> list[Queue]: """Parse a | separated string of queue names against the app.queues dict. diff --git a/flexmeasures/cli/tests/test_jobs.py b/flexmeasures/cli/tests/test_jobs.py index dc1f1ee5f8..07f7dfe5f8 100644 --- a/flexmeasures/cli/tests/test_jobs.py +++ b/flexmeasures/cli/tests/test_jobs.py @@ -169,6 +169,80 @@ def test_run_job_errors_on_a_missing_job(app, clean_job_redis): assert "not found" in result.output.lower() +def test_run_worker_handles_a_job_with_the_handler_of_its_own_queue( + app, clean_job_redis, monkeypatch +): + """A worker on several queues should handle a failing job with the exception handler of the job's queue (#2526). + + The forecasting queue's handler counts the job's failures in its meta, whereas the generic fallback handler does not. + The forecasting queue comes second in the option, so taking the first queue's handler would fail this test, too. + """ + from flexmeasures.cli import jobs as jobs_cli + from rq.job import Job + + class BurstWorker(SimpleWorker): + """Work in this process until the queues are empty, so that the command returns.""" + + def work(self, **kwargs): + return super().work(burst=True, with_scheduler=False) + + # The command uses SimpleWorker on macOS and the forking Worker elsewhere. + monkeypatch.setattr(jobs_cli, "Worker", BurstWorker) + monkeypatch.setattr(jobs_cli, "SimpleWorker", BurstWorker) + + with app.app_context(): + job = app.queues["forecasting"].enqueue("math.sqrt", -1) + + runner = app.test_cli_runner() + result = runner.invoke( + jobs_cli.fm_jobs, ["run-worker", "--queue", "scheduling|forecasting"] + ) + assert result.exit_code == 0, result.output + + with app.app_context(): + job = Job.fetch(job.id, connection=app.redis_connection) + assert job.is_failed + assert ( + job.meta.get("failures") == 1 + ), f"Expected the forecasting queue's exception handler to count the failure, got meta: {job.meta!r}" + + +def test_handle_exception_by_queue_falls_back_and_passes_on_the_result( + app, monkeypatch +): + """A job from a queue without a handler of its own should get the generic handler. + + RQ stops calling further exception handlers when one returns False, so the result of the handler should be passed on. + """ + from types import SimpleNamespace + + from flexmeasures.cli import jobs as jobs_cli + + calls = [] + + def own_handler(job, exc_type, exc_value, traceback): + calls.append(("own", job.origin)) + return False + + def generic_handler(job, exc_type, exc_value, traceback): + calls.append(("generic", job.origin)) + + monkeypatch.setitem(jobs_cli.QUEUE_EXCEPTION_HANDLERS, "scheduling", own_handler) + monkeypatch.setattr(jobs_cli, "handle_worker_exception", generic_handler) + + error = ValueError("boom") + stop = jobs_cli.handle_exception_by_queue( + SimpleNamespace(origin="scheduling"), ValueError, error, None + ) + go_on = jobs_cli.handle_exception_by_queue( + SimpleNamespace(origin="reporting"), ValueError, error, None + ) + + assert calls == [("own", "scheduling"), ("generic", "reporting")] + assert stop is False + assert go_on is None + + def test_check_source_references(app, db, clean_job_redis, monkeypatch): """Waiting jobs referring to a data source that does not exist, or that a database upgrade will merge, are listed; other jobs are not.""" from datetime import timedelta