Merge remote-tracking branch 'origin/main' into berndverst-cache-type-discovery-signature

Author: Bernd Verst

CHANGELOG.md

@@ -10,10 +10,26 @@ adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 ADDED
 
 - Added `FailureDetails.is_caused_by()` to check whether a task failure was caused by a given exception type, mirroring .NET's `TaskFailureDetails.IsCausedBy<T>()` and Java's `FailureDetails.isCausedBy()`. Passing an exception type performs a base-type-aware match (a failure caused by a subclass matches its base type) against exception classes already imported in the process; passing a string matches by qualified or unqualified name.
+- Added a per-invocation dependency-resolution API to `durabletask.extensions.history_export` so any hosting model can supply the export client and writer without a process-global. The export activities now resolve their `HistoryExportContext` from a resolver invoked once per activity execution; new public building blocks are the `HistoryExportContextResolver` type, the pure activity bodies `run_list_terminal_instances(context, input)` and `run_export_instance_history(context, input)`, and the `build_activities(resolver)` factory. This lets host-driven, multi-process models (such as Azure Functions, where the process that starts an export job is not the worker that runs an export activity) inject dependencies lazily per invocation.
 
 CHANGED
 
+- Importing `durabletask` no longer eagerly imports the worker implementation and its
+dependencies (gRPC, protobuf, entities, serialization, OpenTelemetry). The public names
+re-exported from the package — `ActivityWorkItemFilter`, `ConcurrencyOptions`,
+`EntityWorkItemFilter`, `GrpcChannelOptions`, `GrpcRetryPolicyOptions`,
+`LargePayloadStorageOptions`, `OrchestrationWorkItemFilter`, `PayloadStore`,
+`VersioningOptions`, and `WorkItemFilters` — are now resolved on first use, so
+`import durabletask` is substantially faster and loads far fewer modules. This
+measurably reduces cold-start time for client-only applications, including those using
+`durabletask.azuremanaged`, which shares the same `durabletask` namespace. All existing
+import paths, `__all__`, `dir()`, and star-imports behave exactly as before.
 - **Breaking:** `FailureDetails.error_type` — and the `errorType` value sent over the wire — is now the fully-qualified type name (`module.ClassName`, e.g. `builtins.ValueError`, `durabletask.task.TaskFailedError`) instead of the bare class name, matching the .NET and Java SDKs. Code that compared `error_type` against a bare name (for example `== "ValueError"`) must be updated to the qualified name or, preferably, switched to `FailureDetails.is_caused_by()`. Because this value is persisted and crosses the orchestration boundary, failures produced by older workers may still carry a bare name; `is_caused_by()` accepts both.
+- **Breaking:** Retired the process-global history-export context. `durabletask.extensions.history_export.bind_context()` and `clear_context()` have been removed; the export activities now resolve their `HistoryExportContext` per invocation via a resolver captured at registration. `ExportHistoryClient.register_worker()` continues to wire this up automatically, so most callers are unaffected. Code that called `bind_context(HistoryExportContext(client, writer))` directly should instead register the activities with `durabletask.extensions.history_export.activities.register(worker, lambda: HistoryExportContext(client, writer))` (or a lazier resolver), or use `ExportHistoryClient.register_worker()`.
+
+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.
 
 ## v1.8.0
 

azure-functions-durable/azure/durable_functions/internal/history_export_compat.py

@@ -14,8 +14,8 @@
 > This shim exists only because the Durable Functions host extension does not
 > yet implement ``ListInstanceIds``. Once it does, delete this module and have
 > :meth:`DFApp.configure_history_export` register the core
-> ``durabletask.extensions.history_export.activities.list_terminal_instances``
-> activity directly.
+> ``durabletask.extensions.history_export.activities.run_list_terminal_instances``
+> body (via a client-bound wrapper like the export one below) directly.
 """
 
 from __future__ import annotations
@@ -26,7 +26,6 @@
 from datetime import datetime, timezone
 from typing import Any, Optional, cast
 
-from durabletask import task
 from durabletask.client import (
     OrchestrationQuery,
     OrchestrationStatus,
@@ -38,9 +37,7 @@
 from durabletask.extensions.history_export.activities import (
     EXPORT_INSTANCE_HISTORY_ACTIVITY,
     HistoryExportContext,
-    _require_context,  # pyright: ignore[reportPrivateUsage]
-    bind_context,
-    export_instance_history,
+    run_export_instance_history,
 )
 from durabletask.extensions.history_export.entity import ExportJobEntity
 from durabletask.extensions.history_export.models import (
@@ -63,7 +60,7 @@
 
 
 def list_terminal_instances(
-        _: task.ActivityContext, input: Mapping[str, Any]) -> dict[str, Any]:
+        context: HistoryExportContext, input: Mapping[str, Any]) -> dict[str, Any]:
     """Enumerate terminal instances via ``QueryInstances``, one page at a time.
 
     Drop-in replacement for the core ``list_terminal_instances`` activity that
@@ -81,6 +78,10 @@ def list_terminal_instances(
     batch at a time (matching ``max_instances_per_batch``) instead of scheduling
     an export activity for every matching instance at once.
 
+    Like the core activity bodies, the resolved :class:`HistoryExportContext`
+    (client + writer) is passed in explicitly rather than read from a
+    process-global.
+
     > [!IMPORTANT]
     > This is experimental and intended for bounded, low-volume batch-export
     > windows. Because each batch re-scans and re-sorts the whole terminal
@@ -90,8 +91,6 @@ def list_terminal_instances(
     > API (e.g. ``ListInstanceIds``) that the Durable Functions host extension
     > does not yet implement.
     """
-    ctx = _require_context()
-
     raw_statuses = input.get("runtime_status")
     runtime_status_names: Optional[list[str]] = (
         list(raw_statuses) if raw_statuses is not None else None
@@ -110,7 +109,7 @@ def list_terminal_instances(
     if runtime_status_names is not None:
         runtime_status = [OrchestrationStatus[name] for name in runtime_status_names]
 
-    states = ctx.client.get_all_orchestration_states(
+    states = context.client.get_all_orchestration_states(
         OrchestrationQuery(runtime_status=runtime_status))
 
     instance_ids: list[str] = []
@@ -149,14 +148,19 @@ def list_terminal_instances(
 # a scaled-out, multi-worker deployment. The writer is user-supplied and not
 # host-injectable, so it is registered once at app configuration time (which
 # runs in every worker process on import) and reused.
+#
+# The core export activities take their resolved ``HistoryExportContext``
+# explicitly (per invocation), so this adapter simply builds that context from
+# the injected client + configured writer and passes it in -- no process-global
+# ``bind_context`` is involved. The built context is cached per process because
+# the endpoint and writer are stable for the app's lifetime.
 # ---------------------------------------------------------------------------
 
 _export_writer: Optional[HistoryWriter] = None
-_context_bound = False
-# The process-wide sync client bound into the export context, retained so it can
-# be closed at interpreter exit. Guarded by ``_context_lock`` together with the
-# first-bind race.
-_sync_export_client: Optional[TaskHubGrpcClient] = None
+# The per-process export context (sync client + writer), built lazily from the
+# first injected durable client and reused across every export activity. Guarded
+# by ``_context_lock`` for the first-build race; its client is closed at exit.
+_export_context: Optional[HistoryExportContext] = None
 _context_lock = threading.Lock()
 
 
@@ -197,62 +201,58 @@ def _build_sync_client(client: Any) -> TaskHubGrpcClient:
 def _close_sync_export_client() -> None:
     """Close the process-wide sync export client (registered via ``atexit``).
 
-    The client is bound once per worker process and reused across every export
+    The client is built once per worker process and reused across every export
     activity, so it lives for the app's lifetime; closing it at interpreter
     exit releases its gRPC channel on graceful shutdown. Idempotent and
     exception-safe -- shutdown must never surface an error from cleanup.
     """
-    global _sync_export_client
+    global _export_context
     with _context_lock:
-        client = _sync_export_client
-        _sync_export_client = None
-    if client is not None:
+        context = _export_context
+        _export_context = None
+    if context is not None:
         try:
-            client.close()
+            context.client.close()
         except Exception:
             pass
 
 
-def _ensure_context_bound(client: Any) -> None:
-    """Bind the export activity context once per worker process (lazily).
+def _context_for(client: Any) -> HistoryExportContext:
+    """Return the per-process export context, building it once from *client*.
 
     Uses the per-invocation injected client to build the synchronous client and
-    pairs it with the configured writer. Binding is idempotent per process: the
+    pairs it with the configured writer. The result is cached per process: the
     endpoint and writer are stable for the app's lifetime, so the first
     invocation in each process establishes the context for the rest. A lock
-    guards the first-bind race so concurrent fan-out activities do not each open
+    guards the first-build race so concurrent fan-out activities do not each open
     a channel; the built client is closed at process exit.
     """
-    global _context_bound, _sync_export_client
-    if _context_bound:
-        return
+    global _export_context
+    if _export_context is not None:
+        return _export_context
     with _context_lock:
-        if _context_bound:
-            return
-        if _export_writer is None:
-            raise RuntimeError(
-                "history export writer is not configured; pass a writer to "
-                "DFApp.configure_history_export(writer=...) at app startup")
-        sync_client = _build_sync_client(client)
-        _sync_export_client = sync_client
-        atexit.register(_close_sync_export_client)
-        bind_context(HistoryExportContext(
-            client=sync_client, writer=_export_writer))
-        _context_bound = True
+        if _export_context is None:
+            if _export_writer is None:
+                raise RuntimeError(
+                    "history export writer is not configured; pass a writer to "
+                    "DFApp.configure_history_export(writer=...) at app startup")
+            context = HistoryExportContext(
+                client=_build_sync_client(client), writer=_export_writer)
+            atexit.register(_close_sync_export_client)
+            _export_context = context
+    return _export_context
 
 
 def list_terminal_instances_client_bound(
         input: Mapping[str, Any], client: Any) -> dict[str, Any]:
     """``list_terminal_instances`` with the client injected per-invocation."""
-    _ensure_context_bound(client)
-    return list_terminal_instances(task.ActivityContext("", 0), input)
+    return list_terminal_instances(_context_for(client), input)
 
 
 def export_instance_history_client_bound(
         input: Mapping[str, Any], client: Any) -> dict[str, Any]:
     """``export_instance_history`` with the client injected per-invocation."""
-    _ensure_context_bound(client)
-    return export_instance_history(task.ActivityContext("", 0), input)
+    return run_export_instance_history(_context_for(client), input)
 
 
 # The host indexer rejects parameterized generics on trigger parameters and

durabletask-azuremanaged/CHANGELOG.md

@@ -13,6 +13,10 @@ adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
   result, credential failures (for example an unavailable managed identity or a misconfigured
   `DefaultAzureCredential`) now surface from the first request rather than from the constructor.
   The exception type and message are unchanged; only the timing differs.
+- Importing `durabletask.azuremanaged.preview.sandboxes` no longer loads the sandbox worker
+  runtime or `azure-identity` up front. The package's public names are now resolved on first
+  use, which roughly halves import cost for callers that only declare sandbox worker profiles
+  or use the sandbox client. Exported names, `__all__`, and import paths are unchanged.
 - `FailureDetails.error_type` now carries the fully-qualified type name (e.g. `durabletask.task.TaskFailedError`) instead of the bare class name, and the new `FailureDetails.is_caused_by()` helper is available (both inherited from durabletask). See the core `durabletask` changelog for details, including the breaking-change notes.
 - Improved async access token refresh concurrency handling to avoid duplicate
   refresh operations under concurrent access, matching the existing sync

durabletask-azuremanaged/durabletask/azuremanaged/internal/durabletask_grpc_interceptor.py

@@ -1,6 +1,7 @@
 # Copyright (c) Microsoft Corporation.
 # Licensed under the MIT License.
 
+from functools import lru_cache
 from importlib.metadata import version
 
 import grpc
@@ -17,6 +18,22 @@
 )
 
 
+@lru_cache(maxsize=1)
+def _get_sdk_version() -> str:
+    """Return the installed version of the azuremanaged package.
+
+    Resolving the version walks distribution metadata on disk, so the result is
+    cached and shared by every interceptor instance instead of being recomputed
+    on each client or worker construction. Falls back to ``"unknown"`` when the
+    version cannot be determined.
+    """
+    try:
+        return version('durabletask-azuremanaged')
+    except Exception:
+        # Fallback if version cannot be determined
+        return "unknown"
+
+
 class DTSDefaultClientInterceptorImpl (DefaultClientInterceptorImpl):
     """The class implements a UnaryUnaryClientInterceptor, UnaryStreamClientInterceptor,
     StreamUnaryClientInterceptor and StreamStreamClientInterceptor from grpc to add an
@@ -27,13 +44,7 @@ def __init__(
             token_credential: TokenCredential | None,
             taskhub_name: str,
             worker_id: str | None = None):
-        try:
-            # Get the version of the azuremanaged package
-            sdk_version = version('durabletask-azuremanaged')
-        except Exception:
-            # Fallback if version cannot be determined
-            sdk_version = "unknown"
-        user_agent = f"durabletask-python/{sdk_version}"
+        user_agent = f"durabletask-python/{_get_sdk_version()}"
         self._metadata = [
             ("taskhub", taskhub_name),
             ("x-user-agent", user_agent)]  # 'user-agent' is a reserved header; use 'x-user-agent'
@@ -81,13 +92,7 @@ class DTSAsyncDefaultClientInterceptorImpl(DefaultAsyncClientInterceptorImpl):
     (task hub name, user agent, and authentication token) to all async calls."""
 
     def __init__(self, token_credential: AsyncTokenCredential | None, taskhub_name: str):
-        try:
-            # Get the version of the azuremanaged package
-            sdk_version = version('durabletask-azuremanaged')
-        except Exception:
-            # Fallback if version cannot be determined
-            sdk_version = "unknown"
-        user_agent = f"durabletask-python/{sdk_version}"
+        user_agent = f"durabletask-python/{_get_sdk_version()}"
         self._metadata = [
             ("taskhub", taskhub_name),
             ("x-user-agent", user_agent)]

durabletask-azuremanaged/durabletask/azuremanaged/preview/sandboxes/__init__.py

@@ -13,14 +13,22 @@
         SandboxWorker,
         SandboxActivitiesClient,
     )
+
+The exports below are resolved lazily on first attribute access, so importing
+this package does not load the sandbox worker runtime (and its Azure Identity
+and gRPC dependencies) unless those APIs are actually used.
 """
 
-from durabletask.azuremanaged.preview.sandboxes.client import SandboxActivitiesClient
-from durabletask.azuremanaged.preview.sandboxes.helpers import SandboxActivity
-from durabletask.azuremanaged.preview.sandboxes.worker_profiles import SandboxWorkerProfile
-from durabletask.azuremanaged.preview.sandboxes.worker_profiles import SandboxWorkerProfileOptions
-from durabletask.azuremanaged.preview.sandboxes.worker_profiles import sandbox_worker_profile
-from durabletask.azuremanaged.preview.sandboxes.worker import SandboxWorker
+from importlib import import_module
+from typing import TYPE_CHECKING, Any
+
+if TYPE_CHECKING:
+    from durabletask.azuremanaged.preview.sandboxes.client import SandboxActivitiesClient
+    from durabletask.azuremanaged.preview.sandboxes.helpers import SandboxActivity
+    from durabletask.azuremanaged.preview.sandboxes.worker_profiles import SandboxWorkerProfile
+    from durabletask.azuremanaged.preview.sandboxes.worker_profiles import SandboxWorkerProfileOptions
+    from durabletask.azuremanaged.preview.sandboxes.worker_profiles import sandbox_worker_profile
+    from durabletask.azuremanaged.preview.sandboxes.worker import SandboxWorker
 
 __all__ = [
     "SandboxWorker",
@@ -30,3 +38,43 @@
     "SandboxActivitiesClient",
     "sandbox_worker_profile",
 ]
+
+# Public export name -> submodule of this package that defines it.
+_LAZY_EXPORTS: dict[str, str] = {
+    "SandboxWorker": "worker",
+    "SandboxActivity": "helpers",
+    "SandboxWorkerProfile": "worker_profiles",
+    "SandboxWorkerProfileOptions": "worker_profiles",
+    "SandboxActivitiesClient": "client",
+    "sandbox_worker_profile": "worker_profiles",
+}
+
+# Submodules that eager imports previously bound as attributes of this package.
+# They remain reachable through attribute access without an explicit import.
+_LAZY_SUBMODULES: frozenset[str] = frozenset({
+    "client",
+    "helpers",
+    "profile_builder",
+    "transport",
+    "worker",
+    "worker_messages",
+    "worker_profiles",
+})
+
+
+def __getattr__(name: str) -> Any:
+    """Import public sandbox exports on first access (PEP 562)."""
+    submodule = _LAZY_EXPORTS.get(name)
+    if submodule is not None:
+        value = getattr(import_module(f".{submodule}", __name__), name)
+    elif name in _LAZY_SUBMODULES:
+        value = import_module(f".{name}", __name__)
+    else:
+        raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
+
+    globals()[name] = value
+    return value
+
+
+def __dir__() -> list[str]:
+    return sorted(set(globals()) | set(__all__) | _LAZY_SUBMODULES)