Handle each failed job with its own queue's exception handler in run-worker

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 <https://www.github.com/FlexMeasures/flexmeasures/pull/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 <https://www.github.com/FlexMeasures/flexmeasures/pull/2850>`_]
+
 
 v1.1.0 | November 5, 2026
 ============================

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.
 

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