From afdc7a2f49425d5baeadfbd1c90c2c68754fde24 Mon Sep 17 00:00:00 2001 From: Sankalp Thakur Date: Sun, 8 Nov 2026 03:02:36 +0400 Subject: [PATCH 1/2] Handle each failed job with its own queue's exception handler in run-worker run-worker looked up the exception handler with the whole --queue option, so a worker on "scheduling|forecasting", as the hosting docs set it up, used the generic fallback handler for every job. Failed scheduling jobs lost their traceback in the worker log, and failed forecasting jobs were not counted in their meta. Give the worker one handler that passes each job on to the handler of the queue the job came from (job.origin), as run-job already does since #2480. Signed-off-by: Sankalp Thakur --- flexmeasures/cli/jobs.py | 17 ++++++- flexmeasures/cli/tests/test_jobs.py | 74 +++++++++++++++++++++++++++++ 2 files changed, 90 insertions(+), 1 deletion(-) 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 From 1e1e5a0dbf6cc18c68bdbdd6ad6ce50b7735b1a2 Mon Sep 17 00:00:00 2001 From: Sankalp Thakur Date: Mon, 9 Nov 2026 19:42:29 +0400 Subject: [PATCH 2/2] Add the changelog entry Signed-off-by: Sankalp Thakur --- documentation/changelog.rst | 2 ++ 1 file changed, 2 insertions(+) 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 ============================