Tighten on_error handler type to OrchestratorHookError (#23575)

* Fix on_error type to OrchestratorHookError and remove None return types from tests

* Add changelog

* Make on_error and _apply_error_policy generic over wrapped error type

Codex flagged that _apply_error_policy is also called from _task_wrapper
with MessageProcessingError and ProcessorHookError, so narrowing it to
OrchestratorHookError broke mypy. Make it generic on E: Exception and
turn ErrorHandler into a generic alias so each scope keeps the
relationship between the wrapped-error type and its handler. Tighten
BaseProcessor.on_error to MessageProcessingError | ProcessorHookError to
match its actual contract.

Author: AAraKKe

ddev/changelog.d/23575.fixed

@@ -0,0 +1 @@
+Fix type annotation of on_error in the EventBusOrchestrator by narrowing it down to OrchestratorHookError. This better represents the actual error passed to the method.

ddev/src/ddev/event_bus/orchestrator.py

@@ -10,7 +10,7 @@
 from collections.abc import Awaitable, Callable
 from concurrent.futures import Executor, ThreadPoolExecutor
 from dataclasses import dataclass
-from typing import assert_never
+from typing import assert_never, cast
 
 from .exceptions import (
     FatalProcessingError,
@@ -22,7 +22,7 @@
     SkipMessageError,
 )
 
-ErrorHandler = Callable[[Exception], Awaitable[None]]
+type ErrorHandler[E: Exception] = Callable[[E], Awaitable[None]]
 
 
 @dataclass
@@ -44,7 +44,7 @@ def __init__(self, name: str):
     async def on_success(self, message: T) -> None:
         pass
 
-    async def on_error(self, error: Exception) -> None:
+    async def on_error(self, error: MessageProcessingError | ProcessorHookError) -> None:
         """
         Handle a processor-scoped failure.
 
@@ -232,7 +232,7 @@ async def on_message_received(self, message: BaseMessage):  # pragma: no cover
         """
         pass
 
-    async def on_error(self, error: Exception) -> None:
+    async def on_error(self, error: OrchestratorHookError) -> None:
         """
         Handle an orchestrator-scoped failure.
 
@@ -251,7 +251,7 @@ async def on_error(self, error: Exception) -> None:
         """
         raise error
 
-    async def _apply_error_policy(self, wrapped_error: Exception, handler: ErrorHandler) -> None:
+    async def _apply_error_policy[E: Exception](self, wrapped_error: E, handler: ErrorHandler[E]) -> None:
         """
         Routes ``wrapped_error`` through ``handler`` and applies the orchestrator's policy.
 
@@ -460,7 +460,7 @@ async def _task_wrapper(self, processor: Processor, message: BaseMessage):
         try:
             match processor:
                 case AsyncProcessor():
-                    await processor.process_message(message)
+                    await cast(AsyncProcessor, processor).process_message(message)
                 case SyncProcessor():
                     await asyncio.get_running_loop().run_in_executor(self._executor, processor.process_message, message)
                 case _: