Skip to content

Commit 2bd9af5

Browse files
andystaplesCopilot
andcommitted
history_export: resolve activity dependencies per invocation
Retire the process-global `bind_context` from the history-export extension in favor of a per-invocation resolver/factory pattern so any hosting model (including multi-process ones like Azure Functions) can supply the export client and writer without a process-global. Core: - Add `HistoryExportContextResolver` plus pure activity bodies `run_list_terminal_instances`/`run_export_instance_history` that take an explicit `HistoryExportContext`. - Add `build_activities(resolver)` factory and change `register(worker, resolver)` to register resolver-bound activities. - Remove `bind_context`/`clear_context`/`_require_context` and the module-global context. `ExportHistoryClient.register_worker` now captures a context in the activity closures instead of installing a global. Azure Functions adapter: - Drop the `bind_context` shim; build the `HistoryExportContext` from the per-invocation injected client (cached per process) and call the core bodies explicitly. Fixes #182 Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: f91b67b0-d282-4744-b744-4b2a8e1294be
1 parent 0f316cc commit 2bd9af5

11 files changed

Lines changed: 365 additions & 276 deletions

File tree

CHANGELOG.md

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -10,10 +10,12 @@ adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
1010
ADDED
1111

1212
- 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.
13+
- 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.
1314

1415
CHANGED
1516

1617
- **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.
18+
- **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()`.
1719

1820
## v1.8.0
1921

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

Lines changed: 44 additions & 44 deletions
Original file line numberDiff line numberDiff line change
@@ -14,8 +14,8 @@
1414
> This shim exists only because the Durable Functions host extension does not
1515
> yet implement ``ListInstanceIds``. Once it does, delete this module and have
1616
> :meth:`DFApp.configure_history_export` register the core
17-
> ``durabletask.extensions.history_export.activities.list_terminal_instances``
18-
> activity directly.
17+
> ``durabletask.extensions.history_export.activities.run_list_terminal_instances``
18+
> body (via a client-bound wrapper like the export one below) directly.
1919
"""
2020

2121
from __future__ import annotations
@@ -26,7 +26,6 @@
2626
from datetime import datetime, timezone
2727
from typing import Any, Optional, cast
2828

29-
from durabletask import task
3029
from durabletask.client import (
3130
OrchestrationQuery,
3231
OrchestrationStatus,
@@ -38,9 +37,7 @@
3837
from durabletask.extensions.history_export.activities import (
3938
EXPORT_INSTANCE_HISTORY_ACTIVITY,
4039
HistoryExportContext,
41-
_require_context, # pyright: ignore[reportPrivateUsage]
42-
bind_context,
43-
export_instance_history,
40+
run_export_instance_history,
4441
)
4542
from durabletask.extensions.history_export.entity import ExportJobEntity
4643
from durabletask.extensions.history_export.models import (
@@ -63,7 +60,7 @@
6360

6461

6562
def list_terminal_instances(
66-
_: task.ActivityContext, input: Mapping[str, Any]) -> dict[str, Any]:
63+
context: HistoryExportContext, input: Mapping[str, Any]) -> dict[str, Any]:
6764
"""Enumerate terminal instances via ``QueryInstances``, one page at a time.
6865
6966
Drop-in replacement for the core ``list_terminal_instances`` activity that
@@ -81,6 +78,10 @@ def list_terminal_instances(
8178
batch at a time (matching ``max_instances_per_batch``) instead of scheduling
8279
an export activity for every matching instance at once.
8380
81+
Like the core activity bodies, the resolved :class:`HistoryExportContext`
82+
(client + writer) is passed in explicitly rather than read from a
83+
process-global.
84+
8485
> [!IMPORTANT]
8586
> This is experimental and intended for bounded, low-volume batch-export
8687
> windows. Because each batch re-scans and re-sorts the whole terminal
@@ -90,8 +91,6 @@ def list_terminal_instances(
9091
> API (e.g. ``ListInstanceIds``) that the Durable Functions host extension
9192
> does not yet implement.
9293
"""
93-
ctx = _require_context()
94-
9594
raw_statuses = input.get("runtime_status")
9695
runtime_status_names: Optional[list[str]] = (
9796
list(raw_statuses) if raw_statuses is not None else None
@@ -110,7 +109,7 @@ def list_terminal_instances(
110109
if runtime_status_names is not None:
111110
runtime_status = [OrchestrationStatus[name] for name in runtime_status_names]
112111

113-
states = ctx.client.get_all_orchestration_states(
112+
states = context.client.get_all_orchestration_states(
114113
OrchestrationQuery(runtime_status=runtime_status))
115114

116115
instance_ids: list[str] = []
@@ -149,14 +148,19 @@ def list_terminal_instances(
149148
# a scaled-out, multi-worker deployment. The writer is user-supplied and not
150149
# host-injectable, so it is registered once at app configuration time (which
151150
# runs in every worker process on import) and reused.
151+
#
152+
# The core export activities take their resolved ``HistoryExportContext``
153+
# explicitly (per invocation), so this adapter simply builds that context from
154+
# the injected client + configured writer and passes it in -- no process-global
155+
# ``bind_context`` is involved. The built context is cached per process because
156+
# the endpoint and writer are stable for the app's lifetime.
152157
# ---------------------------------------------------------------------------
153158

154159
_export_writer: Optional[HistoryWriter] = None
155-
_context_bound = False
156-
# The process-wide sync client bound into the export context, retained so it can
157-
# be closed at interpreter exit. Guarded by ``_context_lock`` together with the
158-
# first-bind race.
159-
_sync_export_client: Optional[TaskHubGrpcClient] = None
160+
# The per-process export context (sync client + writer), built lazily from the
161+
# first injected durable client and reused across every export activity. Guarded
162+
# by ``_context_lock`` for the first-build race; its client is closed at exit.
163+
_export_context: Optional[HistoryExportContext] = None
160164
_context_lock = threading.Lock()
161165

162166

@@ -197,62 +201,58 @@ def _build_sync_client(client: Any) -> TaskHubGrpcClient:
197201
def _close_sync_export_client() -> None:
198202
"""Close the process-wide sync export client (registered via ``atexit``).
199203
200-
The client is bound once per worker process and reused across every export
204+
The client is built once per worker process and reused across every export
201205
activity, so it lives for the app's lifetime; closing it at interpreter
202206
exit releases its gRPC channel on graceful shutdown. Idempotent and
203207
exception-safe -- shutdown must never surface an error from cleanup.
204208
"""
205-
global _sync_export_client
209+
global _export_context
206210
with _context_lock:
207-
client = _sync_export_client
208-
_sync_export_client = None
209-
if client is not None:
211+
context = _export_context
212+
_export_context = None
213+
if context is not None:
210214
try:
211-
client.close()
215+
context.client.close()
212216
except Exception:
213217
pass
214218

215219

216-
def _ensure_context_bound(client: Any) -> None:
217-
"""Bind the export activity context once per worker process (lazily).
220+
def _context_for(client: Any) -> HistoryExportContext:
221+
"""Return the per-process export context, building it once from *client*.
218222
219223
Uses the per-invocation injected client to build the synchronous client and
220-
pairs it with the configured writer. Binding is idempotent per process: the
224+
pairs it with the configured writer. The result is cached per process: the
221225
endpoint and writer are stable for the app's lifetime, so the first
222226
invocation in each process establishes the context for the rest. A lock
223-
guards the first-bind race so concurrent fan-out activities do not each open
227+
guards the first-build race so concurrent fan-out activities do not each open
224228
a channel; the built client is closed at process exit.
225229
"""
226-
global _context_bound, _sync_export_client
227-
if _context_bound:
228-
return
230+
global _export_context
231+
if _export_context is not None:
232+
return _export_context
229233
with _context_lock:
230-
if _context_bound:
231-
return
232-
if _export_writer is None:
233-
raise RuntimeError(
234-
"history export writer is not configured; pass a writer to "
235-
"DFApp.configure_history_export(writer=...) at app startup")
236-
sync_client = _build_sync_client(client)
237-
_sync_export_client = sync_client
238-
atexit.register(_close_sync_export_client)
239-
bind_context(HistoryExportContext(
240-
client=sync_client, writer=_export_writer))
241-
_context_bound = True
234+
if _export_context is None:
235+
if _export_writer is None:
236+
raise RuntimeError(
237+
"history export writer is not configured; pass a writer to "
238+
"DFApp.configure_history_export(writer=...) at app startup")
239+
context = HistoryExportContext(
240+
client=_build_sync_client(client), writer=_export_writer)
241+
atexit.register(_close_sync_export_client)
242+
_export_context = context
243+
return _export_context
242244

243245

244246
def list_terminal_instances_client_bound(
245247
input: Mapping[str, Any], client: Any) -> dict[str, Any]:
246248
"""``list_terminal_instances`` with the client injected per-invocation."""
247-
_ensure_context_bound(client)
248-
return list_terminal_instances(task.ActivityContext("", 0), input)
249+
return list_terminal_instances(_context_for(client), input)
249250

250251

251252
def export_instance_history_client_bound(
252253
input: Mapping[str, Any], client: Any) -> dict[str, Any]:
253254
"""``export_instance_history`` with the client injected per-invocation."""
254-
_ensure_context_bound(client)
255-
return export_instance_history(task.ActivityContext("", 0), input)
255+
return run_export_instance_history(_context_for(client), input)
256256

257257

258258
# The host indexer rejects parameterized generics on trigger parameters and

durabletask/extensions/history_export/__init__.py

Lines changed: 8 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -22,8 +22,10 @@
2222
)
2323
from durabletask.extensions.history_export.activities import (
2424
HistoryExportContext,
25-
bind_context,
26-
clear_context,
25+
HistoryExportContextResolver,
26+
build_activities,
27+
run_export_instance_history,
28+
run_list_terminal_instances,
2729
)
2830
from durabletask.extensions.history_export.client import (
2931
ExportHistoryClient,
@@ -77,8 +79,10 @@
7779
"ExportJobStatus",
7880
"ExportMode",
7981
"HistoryExportContext",
82+
"HistoryExportContextResolver",
8083
"HistoryWriter",
81-
"bind_context",
82-
"clear_context",
84+
"build_activities",
8385
"orchestrator_instance_id_for",
86+
"run_export_instance_history",
87+
"run_list_terminal_instances",
8488
]

0 commit comments

Comments
 (0)