Performance [P2]: Cache type-discovery signature structure

CHANGELOG.md

@@ -30,6 +30,7 @@ import paths, `__all__`, `dir()`, and star-imports behave exactly as before.
 FIXED
 
 - Fixed the worker allocating one `asyncio` task per queued work item before applying the concurrency limit, which made memory use and event-loop scheduling overhead grow with the queue backlog during bursts. In-flight work item tasks are now bounded by the configured `ConcurrencyOptions` limits.
+- Fixed input/output type discovery raising `TypeError: unhashable type` for handlers that are unhashable callables. A callable object registered through `add_named_activity()`, `add_named_orchestrator()`, or `add_named_entity()` is unhashable whenever its class defines `__eq__` without `__hash__` — most commonly a `@dataclass` with a `__call__` method, since dataclasses default to `eq=True`. Annotations on such handlers are now discovered normally instead of failing.
 
 ## v1.8.0
 

durabletask/internal/type_discovery.py

@@ -27,7 +27,7 @@
 import functools
 import inspect
 import typing
-from typing import Any, Callable
+from typing import Any, Callable, NamedTuple
 
 from durabletask.serialization import DEFAULT_DATA_CONVERTER, DataConverter
 
@@ -37,22 +37,116 @@ def _resolve_converter(converter: DataConverter | None) -> DataConverter:
     return converter if converter is not None else DEFAULT_DATA_CONVERTER
 
 
+def _resolve_hints(fn: Callable[..., Any]) -> dict[str, Any] | None:
+    """Resolve a function's type hints, honoring postponed annotations."""
+    try:
+        return typing.get_type_hints(fn)
+    except Exception:
+        return None
+
+
 # Bounded so a worker that registers dynamically-created functions or closures
 # cannot accumulate cache entries unboundedly over the process lifetime. The
 # common case (a fixed set of module-level orchestrators/activities) fits well
 # within this bound.
 @functools.lru_cache(maxsize=2048)
 def _resolved_hints(fn: Callable[..., Any]) -> dict[str, Any] | None:
-    """Resolve a function's type hints, honoring postponed annotations.
+    """Memoized :func:`_resolve_hints`.
 
     Results are memoized per function because discovery runs on every
     orchestrator/activity/entity execution (including replay).
     """
+    return _resolve_hints(fn)
+
+
+# Sentinel for "there is no annotation here worth asking the converter about"
+# (parameter absent, unannotated, ``Any``, or an unresolvable string
+# annotation). It is deliberately distinct from ``None`` because the two
+# annotation paths treat a literal ``None`` differently, and both behaviours
+# predate this cache:
+#   * parameters -- a ``None`` annotation is a real annotation and is still
+#     passed to the converter, so it must not collapse into the sentinel;
+#   * return values -- ``activity_output_type()`` has always short-circuited on
+#     ``annotation is None``, so ``_build_signature_info()`` normalizes a
+#     ``-> None`` return annotation to this sentinel and it never reaches the
+#     converter.
+_NO_ANNOTATION: Any = object()
+
+
+class _SignatureInfo(NamedTuple):
+    """The converter-independent shape of a callable's signature.
+
+    ``positional`` holds the resolved annotation of each positional parameter in
+    declaration order, and ``return_annotation`` the resolved return annotation.
+    Entries are :data:`_NO_ANNOTATION` when there is nothing to reconstruct.
+
+    This depends only on the callable, so it is safe to memoize. The
+    converter-dependent decision (:meth:`DataConverter.can_reconstruct`) is
+    deliberately *not* part of it and stays at call time, so the same function
+    discovered through two different converters still gets two answers.
+    """
+
+    positional: tuple[Any, ...]
+    return_annotation: Any
+
+
+def _resolve_annotation(raw: Any, name: str, hints: dict[str, Any] | None) -> Any:
+    """Resolve one raw annotation, or return :data:`_NO_ANNOTATION`."""
+    annotation: Any = raw
+    if hints is not None and name in hints:
+        annotation = hints[name]
+    elif isinstance(annotation, str):
+        # Could not resolve a postponed (string) annotation -- give up.
+        return _NO_ANNOTATION
+
+    if annotation is inspect.Parameter.empty or annotation is Any:
+        return _NO_ANNOTATION
+    return annotation
+
+
+def _build_signature_info(fn: Any, *, memoized: bool) -> _SignatureInfo | None:
+    """Inspect ``fn`` and resolve its annotations, or return ``None``."""
     try:
-        return typing.get_type_hints(fn)
-    except Exception:
+        sig = inspect.signature(fn)
+    except (TypeError, ValueError):
         return None
 
+    hints = _resolved_hints(fn) if memoized else _resolve_hints(fn)
+    positional = tuple(
+        _resolve_annotation(p.annotation, p.name, hints)
+        for p in sig.parameters.values()
+        if p.kind in (inspect.Parameter.POSITIONAL_ONLY,
+                      inspect.Parameter.POSITIONAL_OR_KEYWORD)
+    )
+    return_annotation = _resolve_annotation(sig.return_annotation, "return", hints)
+    if return_annotation is None:
+        # ``activity_output_type()`` has always treated an explicit ``-> None``
+        # as "nothing to reconstruct". Fold it into the sentinel here so the
+        # special case stays out of the per-call path.
+        return_annotation = _NO_ANNOTATION
+    return _SignatureInfo(positional, return_annotation)
+
+
+@functools.lru_cache(maxsize=2048)
+def _memoized_signature_info(fn: Any) -> _SignatureInfo | None:
+    return _build_signature_info(fn, memoized=True)
+
+
+def _signature_info(fn: Any) -> _SignatureInfo | None:
+    """Return ``fn``'s resolved signature shape, or ``None`` when unavailable.
+
+    ``inspect.signature()`` and annotation resolution are comparatively
+    expensive and their result depends only on the callable, so the result is
+    memoized: discovery runs once per work item, and a worker executes the same
+    registered functions over and over.
+    """
+    try:
+        return _memoized_signature_info(fn)
+    except TypeError:
+        # Unhashable callable, so it cannot be used as a cache key. Fall back to
+        # computing the shape on every call rather than failing.
+        return _build_signature_info(fn, memoized=False)
+
 
 def _input_annotation(fn: Callable[..., Any], position: int,
                       converter: DataConverter | None = None) -> Any | None:
@@ -64,29 +158,12 @@ def _input_annotation(fn: Callable[..., Any], position: int,
     position 1). Returns ``None`` when the parameter is absent, unannotated, or
     its annotation is not reconstructable by ``converter``.
     """
-    try:
-        sig = inspect.signature(fn)
-    except (TypeError, ValueError):
-        return None
-
-    positional = [
-        p for p in sig.parameters.values()
-        if p.kind in (inspect.Parameter.POSITIONAL_ONLY,
-                      inspect.Parameter.POSITIONAL_OR_KEYWORD)
-    ]
-    if position >= len(positional):
-        return None
-    param = positional[position]
-
-    annotation: Any = param.annotation
-    hints = _resolved_hints(fn)
-    if hints is not None and param.name in hints:
-        annotation = hints[param.name]
-    elif isinstance(annotation, str):
-        # Could not resolve a postponed (string) annotation -- give up.
+    info = _signature_info(fn)
+    if info is None or position >= len(info.positional):
         return None
 
-    if annotation is inspect.Parameter.empty or annotation is Any:
+    annotation = info.positional[position]
+    if annotation is _NO_ANNOTATION:
         return None
     return annotation if _resolve_converter(converter).can_reconstruct(annotation) else None
 
@@ -114,20 +191,13 @@ def activity_output_type(fn: Any, converter: DataConverter | None = None) -> Any
     """
     if not callable(fn):
         return None
-    try:
-        sig = inspect.signature(fn)
-    except (TypeError, ValueError):
-        return None
 
-    annotation: Any = sig.return_annotation
-    hints = _resolved_hints(fn)
-    if hints is not None and "return" in hints:
-        annotation = hints["return"]
-    elif isinstance(annotation, str):
-        # Could not resolve a postponed (string) annotation -- give up.
+    info = _signature_info(fn)
+    if info is None:
         return None
 
-    if annotation is inspect.Signature.empty or annotation is Any or annotation is None:
+    annotation = info.return_annotation
+    if annotation is _NO_ANNOTATION:
         return None
     return annotation if _resolve_converter(converter).can_reconstruct(annotation) else None
 

tests/durabletask/test_type_discovery.py

@@ -3,10 +3,14 @@
 
 """Tests for annotation-based input type discovery and inbound coercion."""
 
+import inspect
 import json
 import logging
 from dataclasses import dataclass
 from typing import Any, Optional
+from unittest.mock import patch
+
+import pytest
 
 from durabletask import entities, task, worker
 from durabletask.internal import type_discovery
@@ -199,6 +203,178 @@ def test_string_name_returns_none(self):
         assert type_discovery.activity_output_type("some_activity_name") is None
 
 
+# ----- signature-structure caching -----
+
+
+class _CountingConverter(JsonDataConverter):
+    """Records every ``can_reconstruct`` call so per-call evaluation is visible."""
+
+    def __init__(self):
+        self.seen: list[Any] = []
+
+    def can_reconstruct(self, target_type: Any) -> bool:
+        self.seen.append(target_type)
+        return super().can_reconstruct(target_type)
+
+
+class TestSignatureCaching:
+    """The signature *structure* is cached per callable, but the converter's
+    reconstructability decision is still made on every call."""
+
+    def test_signature_inspected_once_per_function(self):
+        def act(ctx, order: Order) -> Money:
+            ...
+
+        real_signature = inspect.signature
+        calls: list[Any] = []
+
+        def counting_signature(obj, *args, **kwargs):
+            calls.append(obj)
+            return real_signature(obj, *args, **kwargs)
+
+        with patch.object(inspect, "signature", counting_signature):
+            assert type_discovery.activity_input_type(act) is Order
+            after_first = len(calls)
+            for _ in range(5):
+                assert type_discovery.activity_input_type(act) is Order
+            # The input and output helpers share one cached structure, so the
+            # return annotation costs no additional inspection either.
+            assert type_discovery.activity_output_type(act) is Money
+            assert type_discovery.orchestrator_input_type(act) is Order
+
+        assert after_first >= 1
+        assert len(calls) == after_first, "signature() was re-inspected for a cached function"
+
+    def test_entity_operation_signature_inspected_once(self):
+        class Store(entities.DurableEntity):
+            def add(self, order: Order):
+                ...
+
+        real_signature = inspect.signature
+        calls: list[Any] = []
+
+        def counting_signature(obj, *args, **kwargs):
+            calls.append(obj)
+            return real_signature(obj, *args, **kwargs)
+
+        with patch.object(inspect, "signature", counting_signature):
+            assert type_discovery.entity_input_type(Store, "add") is Order
+            after_first = len(calls)
+            for _ in range(5):
+                assert type_discovery.entity_input_type(Store, "add") is Order
+
+        assert after_first >= 1
+        assert len(calls) == after_first
+
+    def test_converter_decision_is_made_per_call(self):
+        class Widget:
+            pass
+
+        class WidgetConverter(JsonDataConverter):
+            def can_reconstruct(self, target_type: Any) -> bool:
+                if isinstance(target_type, type) and issubclass(target_type, Widget):
+                    return True
+                return super().can_reconstruct(target_type)
+
+        def act(ctx, w: Widget) -> Widget:
+            ...
+
+        # Prime the cache with the converter that *does* recognize Widget, then
+        # switch back: the cached structure must not bake in the first answer.
+        assert type_discovery.activity_input_type(act, WidgetConverter()) is Widget
+        assert type_discovery.activity_input_type(act) is None
+        assert type_discovery.activity_input_type(act, WidgetConverter()) is Widget
+
+        assert type_discovery.activity_output_type(act, WidgetConverter()) is Widget
+        assert type_discovery.activity_output_type(act) is None
+        assert type_discovery.activity_output_type(act, WidgetConverter()) is Widget
+
+    def test_stateful_converter_answer_is_not_cached(self):
+        class Widget:
+            pass
+
+        class ToggleConverter(JsonDataConverter):
+            def __init__(self):
+                self.enabled = False
+
+            def can_reconstruct(self, target_type: Any) -> bool:
+                if self.enabled and target_type is Widget:
+                    return True
+                return super().can_reconstruct(target_type)
+
+        def act(ctx, w: Widget):
+            ...
+
+        converter = ToggleConverter()
+        assert type_discovery.activity_input_type(act, converter) is None
+        converter.enabled = True
+        assert type_discovery.activity_input_type(act, converter) is Widget
+        converter.enabled = False
+        assert type_discovery.activity_input_type(act, converter) is None
+
+    def test_converter_is_consulted_on_every_call(self):
+        def act(ctx, order: Order) -> Order:
+            ...
+
+        converter = _CountingConverter()
+        for _ in range(3):
+            assert type_discovery.activity_input_type(act, converter) is Order
+        assert converter.seen == [Order, Order, Order]
+
+        converter.seen.clear()
+        for _ in range(3):
+            assert type_discovery.activity_output_type(act, converter) is Order
+        assert converter.seen == [Order, Order, Order]
+
+    def test_unannotated_parameters_are_not_offered_to_the_converter(self):
+        def act(ctx, value, *args, keyword_only: Order = None, **kwargs):
+            ...
+
+        converter = _CountingConverter()
+        assert type_discovery.activity_input_type(act, converter) is None
+        # *args/**kwargs and keyword-only parameters are not positional inputs,
+        # and an unannotated parameter never reaches the converter at all.
+        assert converter.seen == []
+
+    def test_unhashable_callable_still_resolves(self):
+        class Callable_:
+            # Defining __eq__ without __hash__ makes instances unhashable, so
+            # they cannot be used as a cache key.
+            def __eq__(self, other):
+                return self is other
+
+            def __call__(self, ctx, order: Order) -> Order:
+                ...
+
+        act = Callable_()
+        assert type_discovery.activity_input_type(act) is Order
+        assert type_discovery.activity_input_type(act) is Order
+        assert type_discovery.activity_output_type(act) is Order
+
+    def test_unhashable_dataclass_handler_registered_by_name_resolves(self):
+        # A ``@dataclass`` with ``__call__`` is the realistic shape of this:
+        # dataclasses default to ``eq=True``, which sets ``__hash__ = None``.
+        # Such a handler has no ``__name__``, so it reaches the worker through
+        # the explicit-name registration API rather than ``add_activity()``.
+        @dataclass
+        class ConfiguredActivity:
+            retries: int = 3
+
+            def __call__(self, ctx, order: Order) -> Order:
+                ...
+
+        handler = ConfiguredActivity()
+        with pytest.raises(TypeError):
+            hash(handler)
+
+        registry = worker._Registry()
+        registry.add_named_activity("process_order", handler)
+        registered = registry.get_activity("process_order")
+
+        assert type_discovery.activity_input_type(registered) is Order
+        assert type_discovery.activity_output_type(registered) is Order
+
+
 # ----- activity executor inbound coercion -----