diff --git a/CHANGELOG.md b/CHANGELOG.md index 99372609..03a7ff22 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -10,10 +10,12 @@ 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()` 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 - **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()`. ## v1.8.0 diff --git a/azure-functions-durable/azure/durable_functions/internal/history_export_compat.py b/azure-functions-durable/azure/durable_functions/internal/history_export_compat.py index 86a96b2c..b56ddcae 100644 --- a/azure-functions-durable/azure/durable_functions/internal/history_export_compat.py +++ b/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 diff --git a/durabletask/extensions/history_export/__init__.py b/durabletask/extensions/history_export/__init__.py index 1d3c846a..9cd191d1 100644 --- a/durabletask/extensions/history_export/__init__.py +++ b/durabletask/extensions/history_export/__init__.py @@ -22,8 +22,10 @@ ) from durabletask.extensions.history_export.activities import ( HistoryExportContext, - bind_context, - clear_context, + HistoryExportContextResolver, + build_activities, + run_export_instance_history, + run_list_terminal_instances, ) from durabletask.extensions.history_export.client import ( ExportHistoryClient, @@ -77,8 +79,10 @@ "ExportJobStatus", "ExportMode", "HistoryExportContext", + "HistoryExportContextResolver", "HistoryWriter", - "bind_context", - "clear_context", + "build_activities", "orchestrator_instance_id_for", + "run_export_instance_history", + "run_list_terminal_instances", ] diff --git a/durabletask/extensions/history_export/activities.py b/durabletask/extensions/history_export/activities.py index 5a08c334..b119afae 100644 --- a/durabletask/extensions/history_export/activities.py +++ b/durabletask/extensions/history_export/activities.py @@ -15,17 +15,27 @@ blob through a :class:`HistoryWriter`. The client and writer are not serializable, so they cannot be passed -through orchestrator inputs. Instead, the public client registers a -module-level :class:`HistoryExportContext` once at worker startup. -The activities resolve their dependencies from that context at -execution time. This is acceptable because activities run in-process -within the worker that registered them. +through orchestrator inputs. Instead, each activity resolves its +dependencies from a :class:`HistoryExportContext` supplied *per +invocation* by a resolver callable. The resolver is captured in the +activity closure at registration time (see :func:`build_activities` +and :func:`register`), so the dependencies never live in a +process-global. This lets any hosting model — including host-driven, +multi-process models such as Azure Functions, where the process that +registers the export job is not the worker that runs an export +activity — supply the client and writer lazily at execution time. + +The pure activity bodies (:func:`run_list_terminal_instances` and +:func:`run_export_instance_history`) take the resolved context +explicitly, so a host with its own per-invocation dependency injection +can call them directly without going through the resolver-based +registration helpers. """ from __future__ import annotations import hashlib -from collections.abc import Mapping +from collections.abc import Callable, Mapping from dataclasses import dataclass from datetime import datetime, timezone from typing import Any, cast @@ -74,57 +84,29 @@ class HistoryExportContext: writer: HistoryWriter -_context: HistoryExportContext | None = None - - -def bind_context(context: HistoryExportContext) -> None: - """Install the runtime dependencies for the history-export activities. - - The bound context is process-wide. Calling this more than once in - the same process — for example by constructing two - :class:`ExportHistoryClient` instances with different writers — - silently replaces the previously-bound writer for *all* in-flight - activities. Such a rebind emits a logger warning so the - misconfiguration is visible at runtime. - """ - global _context - if _context is not None and _context is not context: - from durabletask.extensions.history_export._logging import logger - logger.warning( - "history_export.bind_context() replacing an existing bound " - "context (writer=%r); only one writer can be active per process. " - "Run a separate worker process per writer if you need multiple " - "destinations.", - type(context.writer).__name__, - ) - _context = context - - -def clear_context() -> None: - """Remove the bound context. Useful for tests.""" - global _context - _context = None - - -def _require_context() -> HistoryExportContext: - if _context is None: - raise RuntimeError( - "history-export activities invoked without a bound context; " - "call bind_context(HistoryExportContext(...)) before starting the worker" - ) - return _context +# A resolver produces the :class:`HistoryExportContext` to use for a +# single activity invocation. It is invoked once per activity +# execution, so a host can build (or look up) the client and writer +# lazily — for example from a per-invocation binding — instead of +# relying on a process-global installed at startup. +HistoryExportContextResolver = Callable[[], HistoryExportContext] # ---------------------------------------------------------------------- # Activity bodies # ---------------------------------------------------------------------- -def list_terminal_instances( - _: task.ActivityContext, input: Mapping[str, Any], +def run_list_terminal_instances( + context: HistoryExportContext, input: Mapping[str, Any], ) -> dict[str, Any]: - """Activity: fetch one page of terminal instance IDs.""" - ctx = _require_context() + """Fetch one page of terminal instance IDs using *context*. + Pure activity body: the caller supplies the resolved + :class:`HistoryExportContext` explicitly. :func:`build_activities` + wraps this into a worker-registrable activity that resolves the + context per invocation; hosts with their own dependency injection + can call it directly instead. + """ raw_statuses = input.get("runtime_status") runtime_status_names: list[str] | None = ( list(raw_statuses) if raw_statuses is not None else None @@ -147,7 +129,7 @@ def list_terminal_instances( client_module.OrchestrationStatus[name] for name in runtime_status_names ] - page = ctx.client.list_instance_ids( + page = context.client.list_instance_ids( runtime_status=runtime_status, completed_time_from=completed_time_from, completed_time_to=completed_time_to, @@ -161,12 +143,16 @@ def list_terminal_instances( } -def export_instance_history( - _: task.ActivityContext, input: Mapping[str, Any], +def run_export_instance_history( + context: HistoryExportContext, input: Mapping[str, Any], ) -> dict[str, Any]: - """Activity: serialize and write one instance's history.""" - ctx = _require_context() + """Serialize and write one instance's history using *context*. + Pure activity body: the caller supplies the resolved + :class:`HistoryExportContext` explicitly. See + :func:`run_list_terminal_instances` for how this relates to + :func:`build_activities`. + """ instance_id = str(input["instance_id"]) fmt_input = input.get("format") or { "kind": ExportFormatKind.JSONL_GZIP.value, @@ -190,7 +176,7 @@ def export_instance_history( # now, we refuse to write a partial/empty blob and surface a # specific failure to the orchestrator. Matches the .NET # ``ExportInstanceHistoryActivity`` guard. - state = ctx.client.get_orchestration_state( + state = context.client.get_orchestration_state( instance_id, fetch_payloads=True, ) if state is None: @@ -212,7 +198,7 @@ def export_instance_history( ), } - events = ctx.client.get_orchestration_history(instance_id) + events = context.client.get_orchestration_history(instance_id) # The exported blob is self-describing: it carries the # serialized ``OrchestrationState`` metadata alongside the # event list. Matches the .NET behavior. @@ -238,7 +224,7 @@ def export_instance_history( prefix=prefix, fmt=fmt, ) - ctx.writer.write( + context.writer.write( instance_id=instance_id, container=container, blob_name=blob_name, @@ -316,10 +302,51 @@ def _dotnet_o_format(dt: datetime) -> str: return f"{base}.{fractional}{offset_str}" -def register(worker_instance: worker_module.TaskHubGrpcWorker) -> None: - """Convenience helper to register both activities on *worker*.""" - worker_instance.add_activity(list_terminal_instances) - worker_instance.add_activity(export_instance_history) +def build_activities( + resolver: HistoryExportContextResolver, +) -> tuple[task.Activity[Any, Any], task.Activity[Any, Any]]: + """Build the two history-export activities bound to *resolver*. + + Returns ``(list_terminal_instances, export_instance_history)`` — a + pair of activity callables with the canonical activity names + (:data:`LIST_TERMINAL_INSTANCES_ACTIVITY` and + :data:`EXPORT_INSTANCE_HISTORY_ACTIVITY`). *resolver* is invoked + once per activity execution to obtain the + :class:`HistoryExportContext`, so the client and writer are + captured in the returned closures rather than in a process-global. + """ + def list_terminal_instances( + _: task.ActivityContext, input: Mapping[str, Any], + ) -> dict[str, Any]: + return run_list_terminal_instances(resolver(), input) + + def export_instance_history( + _: task.ActivityContext, input: Mapping[str, Any], + ) -> dict[str, Any]: + return run_export_instance_history(resolver(), input) + + # The activity name registered with the worker is ``fn.__name__`` + # (see :func:`durabletask.task.get_name`); pin both to the + # canonical names the orchestrator calls. + list_terminal_instances.__name__ = LIST_TERMINAL_INSTANCES_ACTIVITY + export_instance_history.__name__ = EXPORT_INSTANCE_HISTORY_ACTIVITY + return list_terminal_instances, export_instance_history + + +def register( + worker_instance: worker_module.TaskHubGrpcWorker, + resolver: HistoryExportContextResolver, +) -> None: + """Register both activities on *worker*, resolving deps via *resolver*. + + *resolver* is invoked per activity execution to obtain the + :class:`HistoryExportContext` (client + writer). Pass + ``lambda: HistoryExportContext(client, writer)`` for a fixed + client/writer, or a factory that builds them lazily per invocation. + """ + list_activity, export_activity = build_activities(resolver) + worker_instance.add_activity(list_activity) + worker_instance.add_activity(export_activity) # Used by the orchestrator to build a fresh activity input from the diff --git a/durabletask/extensions/history_export/client.py b/durabletask/extensions/history_export/client.py index fdd302ab..e9dbffe0 100644 --- a/durabletask/extensions/history_export/client.py +++ b/durabletask/extensions/history_export/client.py @@ -72,7 +72,6 @@ from durabletask.extensions.history_export._logging import logger from durabletask.extensions.history_export.activities import ( HistoryExportContext, - bind_context, register as _register_activities, ) from durabletask.extensions.history_export.writer import HistoryWriter @@ -141,14 +140,16 @@ def __init__( def register_worker(self, worker_instance: worker_module.TaskHubGrpcWorker) -> None: """Register the entity, activities, and orchestrator on *worker*. - Also binds the activity execution context so the activities - can find the underlying client and writer at runtime. Call - this once per worker before :meth:`start`. + The activities resolve their runtime dependencies (this + client and the writer) from a :class:`HistoryExportContext` + supplied per invocation by a resolver captured in the activity + closures — no process-global is installed. Call this once per + worker before :meth:`start`. """ worker_instance.add_entity(ExportJobEntity, name=ENTITY_NAME) - _register_activities(worker_instance) + context = HistoryExportContext(client=self._client, writer=self._writer) + _register_activities(worker_instance, lambda: context) worker_instance.add_orchestrator(export_job_orchestrator) - bind_context(HistoryExportContext(client=self._client, writer=self._writer)) # ------------------------------------------------------------------ # Job lifecycle diff --git a/tests/azure-functions-durable/e2e/apps/dtask_style/history_export_routes.py b/tests/azure-functions-durable/e2e/apps/dtask_style/history_export_routes.py index fd8e98b0..6e324877 100644 --- a/tests/azure-functions-durable/e2e/apps/dtask_style/history_export_routes.py +++ b/tests/azure-functions-durable/e2e/apps/dtask_style/history_export_routes.py @@ -8,12 +8,13 @@ activities (registered via ``app.configure_history_export()``), plus the public ``ExportHistoryClient`` surface (create / get job). -The export activities need a durabletask client and a ``HistoryWriter`` bound -into the worker process via ``bind_context``. The client is only available at -request time (from the durable-client binding), so the context is bound here on -each request -- the route handler shares the worker process with the activities, -so the binding is visible to them. A local-filesystem writer sends exported -history to ``/_export_output`` where the test can read it back. +The export activities need a durabletask client and a ``HistoryWriter``. They +resolve their client per invocation from a durable-client binding (host-supplied +in whatever worker runs them) and use the writer registered at +``configure_history_export`` time, so this route no longer binds a process-wide +context. The route only builds an ``ExportHistoryClient`` for the job-management +surface (create / get job). A local-filesystem writer sends exported history to +``/_export_output`` where the test can read it back. """ import json diff --git a/tests/azure-functions-durable/test_history_export_compat.py b/tests/azure-functions-durable/test_history_export_compat.py index 8988a1bf..74521cfa 100644 --- a/tests/azure-functions-durable/test_history_export_compat.py +++ b/tests/azure-functions-durable/test_history_export_compat.py @@ -36,59 +36,64 @@ def _state(instance_id: str, completed_at: datetime = _COMPLETED) -> SimpleNames return SimpleNamespace(instance_id=instance_id, last_updated_at=completed_at) -def _bind_states(monkeypatch, states) -> MagicMock: +def _context_with_states(states) -> SimpleNamespace: + """Build an explicit history-export context whose client returns *states*. + + The Functions ``list_terminal_instances`` body takes the resolved context + per invocation and only reads ``context.client``, so a lightweight + namespace suffices. + """ client = MagicMock() client.get_all_orchestration_states.return_value = states - monkeypatch.setattr(hec, "_require_context", lambda: SimpleNamespace(client=client)) - return client + return SimpleNamespace(client=client) -def test_single_page_when_all_fit(monkeypatch): - _bind_states(monkeypatch, [_state("id-1"), _state("id-0")]) +def test_single_page_when_all_fit(): + ctx = _context_with_states([_state("id-1"), _state("id-0")]) page = hec.list_terminal_instances( - None, {"completed_time_from": _FROM, "page_size": 10}) + ctx, {"completed_time_from": _FROM, "page_size": 10}) # Sorted deterministically, single page, no further pages. assert page["instance_ids"] == ["id-0", "id-1"] assert page["continuation_token"] is None -def test_pages_by_page_size_with_keyset_cursor(monkeypatch): +def test_pages_by_page_size_with_keyset_cursor(): states = [_state(f"id-{i}") for i in (3, 1, 4, 0, 2)] - _bind_states(monkeypatch, states) + ctx = _context_with_states(states) base = {"completed_time_from": _FROM, "page_size": 2} - p1 = hec.list_terminal_instances(None, dict(base)) + p1 = hec.list_terminal_instances(ctx, dict(base)) assert p1["instance_ids"] == ["id-0", "id-1"] assert p1["continuation_token"] == "id-1" p2 = hec.list_terminal_instances( - None, {**base, "continuation_token": p1["continuation_token"]}) + ctx, {**base, "continuation_token": p1["continuation_token"]}) assert p2["instance_ids"] == ["id-2", "id-3"] assert p2["continuation_token"] == "id-3" # Final page has fewer than page_size items -> no continuation token. p3 = hec.list_terminal_instances( - None, {**base, "continuation_token": p2["continuation_token"]}) + ctx, {**base, "continuation_token": p2["continuation_token"]}) assert p3["instance_ids"] == ["id-4"] assert p3["continuation_token"] is None -def test_exact_multiple_of_page_size_terminates(monkeypatch): - _bind_states(monkeypatch, [_state("id-0"), _state("id-1")]) +def test_exact_multiple_of_page_size_terminates(): + ctx = _context_with_states([_state("id-0"), _state("id-1")]) base = {"completed_time_from": _FROM, "page_size": 2} - p1 = hec.list_terminal_instances(None, dict(base)) + p1 = hec.list_terminal_instances(ctx, dict(base)) assert p1["instance_ids"] == ["id-0", "id-1"] # Exactly page_size items remained, so this is the last page. assert p1["continuation_token"] is None -def test_completed_time_window_is_applied(monkeypatch): +def test_completed_time_window_is_applied(): early = datetime(2020, 1, 1, tzinfo=timezone.utc) late = datetime(2030, 1, 1, tzinfo=timezone.utc) - _bind_states(monkeypatch, [ + ctx = _context_with_states([ _state("early", early), _state("in-window", _COMPLETED), _state("late", late)]) - page = hec.list_terminal_instances(None, { + page = hec.list_terminal_instances(ctx, { "completed_time_from": _FROM, "completed_time_to": "2027-01-01T00:00:00+00:00", "page_size": 10, @@ -97,16 +102,16 @@ def test_completed_time_window_is_applied(monkeypatch): assert page["continuation_token"] is None -def test_requires_completed_time_from(monkeypatch): - _bind_states(monkeypatch, []) +def test_requires_completed_time_from(): + ctx = _context_with_states([]) with pytest.raises(ValueError, match="completed_time_from"): - hec.list_terminal_instances(None, {"page_size": 10}) + hec.list_terminal_instances(ctx, {"page_size": 10}) -def test_empty_result(monkeypatch): - _bind_states(monkeypatch, []) +def test_empty_result(): + ctx = _context_with_states([]) page = hec.list_terminal_instances( - None, {"completed_time_from": _FROM, "page_size": 10}) + ctx, {"completed_time_from": _FROM, "page_size": 10}) assert page["instance_ids"] == [] assert page["continuation_token"] is None @@ -180,47 +185,57 @@ def test_continuous_create_on_fresh_entity_is_marked_failed(): # --------------------------------------------------------------------------- -# Sync export client lifecycle -- the process-wide client bound into the export -# context is closed at interpreter exit. +# Sync export client lifecycle -- the per-process context (and its sync client) +# is built once from the injected client and closed at interpreter exit. # --------------------------------------------------------------------------- -def test_ensure_context_bound_registers_and_closes_sync_client(monkeypatch): +def test_context_for_builds_caches_and_closes_sync_client(monkeypatch): fake_client = MagicMock() monkeypatch.setattr(hec, "_build_sync_client", lambda _c: fake_client) - monkeypatch.setattr(hec, "_export_writer", MagicMock()) - monkeypatch.setattr(hec, "_context_bound", False) - monkeypatch.setattr(hec, "_sync_export_client", None) + writer = MagicMock() + monkeypatch.setattr(hec, "_export_writer", writer) + monkeypatch.setattr(hec, "_export_context", None) - bind_calls = [] - monkeypatch.setattr(hec, "bind_context", lambda ctx: bind_calls.append(ctx)) registered = [] monkeypatch.setattr(hec.atexit, "register", lambda fn: registered.append(fn)) - hec._ensure_context_bound(object()) + context = hec._context_for(object()) - assert hec._sync_export_client is fake_client - assert hec._context_bound is True - assert len(bind_calls) == 1 + # The context pairs the built sync client with the configured writer and is + # cached for the process. + assert context.client is fake_client + assert context.writer is writer + assert hec._export_context is context assert hec._close_sync_export_client in registered - # A second bind is a no-op: no new client, no duplicate registration. - hec._ensure_context_bound(object()) - assert len(bind_calls) == 1 + # A second resolution is a no-op: same cached context, no duplicate + # registration and no new client build. + assert hec._context_for(object()) is context assert registered.count(hec._close_sync_export_client) == 1 # Closing releases the client, resets state, and is idempotent + safe. hec._close_sync_export_client() fake_client.close.assert_called_once() - assert hec._sync_export_client is None + assert hec._export_context is None hec._close_sync_export_client() fake_client.close.assert_called_once() +def test_context_for_requires_configured_writer(monkeypatch): + monkeypatch.setattr(hec, "_build_sync_client", lambda _c: MagicMock()) + monkeypatch.setattr(hec, "_export_writer", None) + monkeypatch.setattr(hec, "_export_context", None) + + with pytest.raises(RuntimeError, match="writer is not configured"): + hec._context_for(object()) + + def test_close_sync_export_client_swallows_errors(monkeypatch): fake_client = MagicMock() fake_client.close.side_effect = RuntimeError("boom") - monkeypatch.setattr(hec, "_sync_export_client", fake_client) + context = hec.HistoryExportContext(client=fake_client, writer=MagicMock()) + monkeypatch.setattr(hec, "_export_context", context) # Cleanup at shutdown must never raise. hec._close_sync_export_client() - assert hec._sync_export_client is None + assert hec._export_context is None diff --git a/tests/durabletask/extensions/history_export/_test_helpers.py b/tests/durabletask/extensions/history_export/_test_helpers.py index 3f01a646..29eed4a6 100644 --- a/tests/durabletask/extensions/history_export/_test_helpers.py +++ b/tests/durabletask/extensions/history_export/_test_helpers.py @@ -16,9 +16,34 @@ import time from typing import Any, Callable, Optional, TypeVar +from durabletask.extensions.history_export.activities import HistoryExportContext + T = TypeVar("T") +class MutableContextHolder: + """A per-invocation context resolver whose context tests can swap. + + History-export activities resolve their :class:`HistoryExportContext` + per invocation via a resolver captured at registration time. A + module worker is registered once with :meth:`resolve` as that + resolver, and each test assigns :attr:`context` (or ``None`` to make + the resolver raise) so a single registration can exercise many + different client/writer pairs. + """ + + def __init__(self) -> None: + self.context: HistoryExportContext | None = None + + def resolve(self) -> HistoryExportContext: + if self.context is None: + raise RuntimeError( + "history-export test context not set; assign .context before " + "scheduling the orchestration" + ) + return self.context + + def wait_until( predicate: Callable[[], Optional[T]], *, diff --git a/tests/durabletask/extensions/history_export/test_activities.py b/tests/durabletask/extensions/history_export/test_activities.py index 13251049..4a2dfb1b 100644 --- a/tests/durabletask/extensions/history_export/test_activities.py +++ b/tests/durabletask/extensions/history_export/test_activities.py @@ -4,11 +4,11 @@ """E2E tests for history-export activities against the in-memory backend. Tests share a single backend + worker per module to avoid paying the -worker start/shutdown cost on every test. The bound activity context -is module-global, so individual tests swap their own -:class:`HistoryExportContext` (writer + client) in via ``bind_context`` -at the top of the test, and restore a clean slate via ``clear_context`` -on teardown. +worker start/shutdown cost on every test. The activities are +registered once with a resolver that reads from a module-level holder; +individual tests swap their own :class:`HistoryExportContext` (writer + +client) into the holder, and the autouse fixture clears it between +tests. """ from __future__ import annotations @@ -30,18 +30,20 @@ EXPORT_INSTANCE_HISTORY_ACTIVITY, LIST_TERMINAL_INSTANCES_ACTIVITY, HistoryExportContext, - bind_context, - clear_context, register as register_activities, ) from durabletask.testing import create_test_backend +from ._test_helpers import MutableContextHolder from tests.durabletask._port_utils import find_free_port PORT = find_free_port() HOST = f"localhost:{PORT}" +_holder = MutableContextHolder() + + class _InMemoryWriter: def __init__(self) -> None: self._lock = threading.Lock() @@ -90,7 +92,7 @@ def w(backend): w_ = worker.TaskHubGrpcWorker(host_address=HOST) w_.add_orchestrator(_echo_orchestrator) w_.add_orchestrator(_list_then_export) - register_activities(w_) + register_activities(w_, _holder.resolve) w_.start() yield w_ w_.stop() @@ -103,10 +105,10 @@ def c(w): @pytest.fixture(autouse=True) def _isolate_context(): - """Each test must explicitly bind its own context.""" - clear_context() + """Each test must explicitly set its own context in the holder.""" + _holder.context = None yield - clear_context() + _holder.context = None @pytest.fixture(scope="module") @@ -129,7 +131,7 @@ def seeded_ids(c, w): def test_activities_list_and_export_to_in_memory_writer(c, seeded_ids): writer = _InMemoryWriter() - bind_context(HistoryExportContext(client=c, writer=writer)) + _holder.context = HistoryExportContext(client=c, writer=writer) now = datetime.now(timezone.utc) fmt = ExportFormat(kind=ExportFormatKind.JSONL_GZIP) @@ -182,7 +184,7 @@ class FailingWriter: def write(self, **_): raise RuntimeError("disk full") - bind_context(HistoryExportContext(client=c, writer=FailingWriter())) + _holder.context = HistoryExportContext(client=c, writer=FailingWriter()) now = datetime.now(timezone.utc) fmt = ExportFormat(kind=ExportFormatKind.JSON) @@ -211,8 +213,9 @@ def write(self, **_): assert all("disk full" in r["error"] for r in failures) -def test_activities_require_bound_context(c): - # Do NOT bind a context. The activities should raise. +def test_activities_fail_when_context_resolution_raises(c): + # Do NOT set a context in the holder. The resolver should raise, + # and the activity failure must surface on the orchestration. now = datetime.now(timezone.utc) fmt = ExportFormat(kind=ExportFormatKind.JSON) dest = ExportDestination(container="exports") @@ -235,7 +238,7 @@ def test_activities_require_bound_context(c): assert state is not None assert state.runtime_status == client.OrchestrationStatus.FAILED assert state.failure_details is not None - assert "without a bound context" in (state.failure_details.message or "") + assert "test context not set" in (state.failure_details.message or "") # --------------------------------------------------------------------- @@ -286,14 +289,14 @@ def _basic_activity_input() -> dict: def test_export_activity_skips_when_instance_no_longer_exists(): """N-2: instance purged between list and export -> failure without write.""" from durabletask.extensions.history_export.activities import ( - export_instance_history, + run_export_instance_history, ) stub_client = _StubGetStateClient(state=None) writer = _CountingWriter() - bind_context(HistoryExportContext(client=stub_client, writer=writer)) + context = HistoryExportContext(client=stub_client, writer=writer) - result = export_instance_history(None, _basic_activity_input()) + result = run_export_instance_history(context, _basic_activity_input()) assert result["success"] is False assert "no longer exists" in result["error"] @@ -304,7 +307,7 @@ def test_export_activity_skips_when_instance_no_longer_exists(): def test_export_activity_skips_when_instance_is_not_terminal(): """N-2: instance has re-entered a running state -> failure without write.""" from durabletask.extensions.history_export.activities import ( - export_instance_history, + run_export_instance_history, ) class _State: @@ -312,9 +315,9 @@ class _State: stub_client = _StubGetStateClient(state=_State()) writer = _CountingWriter() - bind_context(HistoryExportContext(client=stub_client, writer=writer)) + context = HistoryExportContext(client=stub_client, writer=writer) - result = export_instance_history(None, _basic_activity_input()) + result = run_export_instance_history(context, _basic_activity_input()) assert result["success"] is False assert "no longer terminal" in result["error"] diff --git a/tests/durabletask/extensions/history_export/test_client.py b/tests/durabletask/extensions/history_export/test_client.py index 456d8836..3275b50f 100644 --- a/tests/durabletask/extensions/history_export/test_client.py +++ b/tests/durabletask/extensions/history_export/test_client.py @@ -31,7 +31,6 @@ ExportMode, orchestrator_instance_id_for, ) -from durabletask.extensions.history_export.activities import clear_context from durabletask.testing import create_test_backend from ._test_helpers import wait_until @@ -66,7 +65,6 @@ def backend(): yield b b.stop() b.reset() - clear_context() @pytest.fixture(scope="module") diff --git a/tests/durabletask/extensions/history_export/test_orchestrator.py b/tests/durabletask/extensions/history_export/test_orchestrator.py index dc81118d..5298e176 100644 --- a/tests/durabletask/extensions/history_export/test_orchestrator.py +++ b/tests/durabletask/extensions/history_export/test_orchestrator.py @@ -28,16 +28,29 @@ ExportJobStatus, ExportMode, ) -from durabletask.extensions.history_export.activities import clear_context +from durabletask.extensions.history_export._constants import ENTITY_NAME +from durabletask.extensions.history_export.activities import ( + HistoryExportContext, + register as register_activities, +) +from durabletask.extensions.history_export.entity import ExportJobEntity +from durabletask.extensions.history_export.orchestrator import ( + export_job_orchestrator, +) from durabletask.extensions.history_export import orchestrator as orch_mod from durabletask.testing import create_test_backend -from ._test_helpers import wait_until +from ._test_helpers import MutableContextHolder, wait_until from tests.durabletask._port_utils import find_free_port PORT = find_free_port() HOST = f"localhost:{PORT}" +# The module worker registers the export activities once with this holder as +# their per-invocation context resolver; tests swap ``_holder.context`` to +# exercise different writers (or ``None`` to force a resolution failure). +_holder = MutableContextHolder() + class _InMemoryWriter: def __init__(self) -> None: @@ -94,15 +107,35 @@ def export_client(dt_client, writer): @pytest.fixture(scope="module") -def w(backend, export_client): +def w(backend, dt_client, writer): w_ = worker.TaskHubGrpcWorker(host_address=HOST) w_.add_orchestrator(_echo) - export_client.register_worker(w_) + # Register the export built-ins manually (rather than via + # ``export_client.register_worker``) so the activities resolve their + # context from ``_holder`` — this lets individual tests swap the writer + # (or force a resolution failure) without a process-global. + w_.add_entity(ExportJobEntity, name=ENTITY_NAME) + register_activities(w_, _holder.resolve) + w_.add_orchestrator(export_job_orchestrator) w_.start() yield w_ w_.stop() +@pytest.fixture(autouse=True) +def _default_context(w, dt_client, writer): + """Arm the holder with the good client + module writer for each test. + + Depends on ``w`` so the shared worker is always running for a test. + Tests that need a different context (a failing writer, or none at all) + overwrite ``_holder.context`` in their body; this fixture restores the + default before and after every test so they stay isolated. + """ + _holder.context = HistoryExportContext(client=dt_client, writer=writer) + yield + _holder.context = HistoryExportContext(client=dt_client, writer=writer) + + @pytest.fixture(scope="module") def seeded_ids(dt_client, w): ids: list[str] = [] @@ -215,40 +248,28 @@ def _orchestration_is_done() -> bool: ) -def test_orchestrator_records_failure_when_no_context_bound( - dt_client, export_client, +def test_orchestrator_records_failure_when_context_resolution_fails( + export_client, ): - """An orchestrator that cannot reach its activity context fails the job.""" - # The shared module worker has a bound context. Clear it so the - # next orchestrator's activities raise, then restore it afterwards - # so subsequent tests are unaffected. - from durabletask.extensions.history_export.activities import ( - HistoryExportContext, - bind_context, - ) - clear_context() - try: - now = datetime.now(timezone.utc) - desc = export_client.create_job( - ExportJobCreationOptions( - mode=ExportMode.BATCH, - completed_time_from=now - timedelta(hours=1), - completed_time_to=now + timedelta(hours=1), - destination=ExportDestination(container="exports"), - format=ExportFormat(kind=ExportFormatKind.JSON), - max_instances_per_batch=10, - ) - ) - final = export_client.wait_for_job(desc.job_id, timeout=15, poll_interval=0.1) - assert final.status == ExportJobStatus.FAILED - assert final.last_error is not None - finally: - # Re-arm the context for any subsequent tests. - bind_context( - HistoryExportContext( - client=dt_client, writer=export_client.writer, - ) + """An orchestrator that cannot resolve its activity context fails the job.""" + # Clear the holder so the resolver raises when the next + # orchestrator's activities run; the autouse ``_default_context`` + # fixture restores the good context afterwards. + _holder.context = None + now = datetime.now(timezone.utc) + desc = export_client.create_job( + ExportJobCreationOptions( + mode=ExportMode.BATCH, + completed_time_from=now - timedelta(hours=1), + completed_time_to=now + timedelta(hours=1), + destination=ExportDestination(container="exports"), + format=ExportFormat(kind=ExportFormatKind.JSON), + max_instances_per_batch=10, ) + ) + final = export_client.wait_for_job(desc.job_id, timeout=15, poll_interval=0.1) + assert final.status == ExportJobStatus.FAILED + assert final.last_error is not None class _AlwaysFailingWriter: @@ -268,62 +289,54 @@ def test_batch_failure_marks_job_failed_without_invalid_transition( already driven the entity to FAILED, which the transitions matrix would reject and log as an invalid-transition error. """ - from durabletask.extensions.history_export.activities import ( - HistoryExportContext, - bind_context, + # Swap in a permanently-failing writer for this test only via the + # holder; the autouse ``_default_context`` fixture restores the good + # context afterwards so the shared module worker stays consistent. + _holder.context = HistoryExportContext( + client=dt_client, writer=_AlwaysFailingWriter(), ) - # Swap in a permanently-failing writer for this test only; restore - # the original writer in the finally block so the shared module - # fixtures stay consistent. - original_writer = export_client.writer - bind_context(HistoryExportContext(client=dt_client, writer=_AlwaysFailingWriter())) - try: - with caplog.at_level("WARNING", logger="durabletask.extensions.history_export"): - now = datetime.now(timezone.utc) - desc = export_client.create_job( - ExportJobCreationOptions( - mode=ExportMode.BATCH, - completed_time_from=now - timedelta(hours=1), - completed_time_to=now + timedelta(hours=1), - destination=ExportDestination( - container="exports", prefix="batch-fail-test", - ), - format=ExportFormat(kind=ExportFormatKind.JSON), - max_instances_per_batch=10, - ) + with caplog.at_level("WARNING", logger="durabletask.extensions.history_export"): + now = datetime.now(timezone.utc) + desc = export_client.create_job( + ExportJobCreationOptions( + mode=ExportMode.BATCH, + completed_time_from=now - timedelta(hours=1), + completed_time_to=now + timedelta(hours=1), + destination=ExportDestination( + container="exports", prefix="batch-fail-test", + ), + format=ExportFormat(kind=ExportFormatKind.JSON), + max_instances_per_batch=10, ) - # Generous timeout because the orchestrator does 3 batch x - # 3 activity retries against the (overridden, fast) backoff. - final = export_client.wait_for_job(desc.job_id, timeout=60, poll_interval=0.1) - - assert final.status == ExportJobStatus.FAILED - assert final.last_error is not None - # last_error summary mentions the writer's failure reason. - assert "simulated permanent write failure" in (final.last_error or "") - # The failures list is populated and at least one entry - # carries the reason text propagated up from the writer. - # (Each failure's ``instance_id`` is whatever terminal - # orchestration was in the export window — which may include - # prior tests' export orchestrators, not just the seeded - # sample workload. Reasons can also vary if prior writers - # left blobs behind, etc.) - assert len(final.failures) >= 1 - assert any( - "simulated permanent write failure" in f.reason - for f in final.failures ) + # Generous timeout because the orchestrator does 3 batch x + # 3 activity retries against the (overridden, fast) backoff. + final = export_client.wait_for_job(desc.job_id, timeout=60, poll_interval=0.1) + + assert final.status == ExportJobStatus.FAILED + assert final.last_error is not None + # last_error summary mentions the writer's failure reason. + assert "simulated permanent write failure" in (final.last_error or "") + # The failures list is populated and at least one entry + # carries the reason text propagated up from the writer. + # (Each failure's ``instance_id`` is whatever terminal + # orchestration was in the export window — which may include + # prior tests' export orchestrators, not just the seeded + # sample workload. Reasons can also vary if prior writers + # left blobs behind, etc.) + assert len(final.failures) >= 1 + assert any( + "simulated permanent write failure" in f.reason + for f in final.failures + ) - # Critical regression check: the orchestrator must not have - # issued a second ``mark_failed`` signal after - # ``commit_checkpoint`` already transitioned the entity to - # FAILED. If it had, the entity would have raised - # ExportJobInvalidTransitionError; the SDK logs that into - # caplog at WARNING/ERROR severity. - for record in caplog.records: - assert "ExportJobInvalidTransitionError" not in record.getMessage(), ( - f"Found invalid-transition log: {record.getMessage()!r}" - ) - finally: - bind_context( - HistoryExportContext(client=dt_client, writer=original_writer) + # Critical regression check: the orchestrator must not have + # issued a second ``mark_failed`` signal after + # ``commit_checkpoint`` already transitioned the entity to + # FAILED. If it had, the entity would have raised + # ExportJobInvalidTransitionError; the SDK logs that into + # caplog at WARNING/ERROR severity. + for record in caplog.records: + assert "ExportJobInvalidTransitionError" not in record.getMessage(), ( + f"Found invalid-transition log: {record.getMessage()!r}" )