Add history export to afd

Author: andystaples

.gitignore

@@ -114,6 +114,9 @@ venv.bak/
 # Azure Functions E2E test host logs (written by the test harness)
 _func_host.log
 
+# Azure Functions E2E history-export output (written by the sample app writer)
+_export_output/
+
 # Spyder project settings
 .spyderproject
 .spyproject

CHANGELOG.md

@@ -13,13 +13,25 @@ ADDED
 - Added `OrchestrationContext.parent_instance_id`, which returns the instance
   ID of the parent orchestration for a sub-orchestration, or `None` for a
   top-level orchestration.
+- Exported `bind_context` and `clear_context` from
+  `durabletask.extensions.history_export` so hosts that register the export
+  functions themselves (rather than via `ExportHistoryClient.register_worker`)
+  can supply the activities' runtime dependencies.
 
 CHANGED
 
 - Changed the default large-payload externalization threshold (`LargePayloadStorageOptions.threshold_bytes`) from 900,000 bytes to 262,144 bytes (256 KiB), matching the .NET SDK default. Behavioral change (not source/binary breaking): payloads larger than 256 KiB are now externalized by default.
 
 FIXED
 
+- Fixed durabletask scheduled tasks (`durabletask.scheduled`) failing under
+  data converters that reconstruct nested custom-object envelopes bottom-up
+  (such as the Azure Functions Durable `df` codec). `ScheduleState.from_json`
+  now tolerates an already-reconstructed nested `ScheduleConfiguration` (and
+  accepts the active `DataConverter` for nested reconstruction), and
+  `ScheduleOperationRequest` gained `to_json`/`from_json` hooks so it can be
+  serialized by converters that require them. The default JSON converter's
+  behavior is unchanged.
 - Fixed `OrchestrationContext.lock_entities` failing when used over the legacy
   entity protocol (used by the Azure Functions Durable extension). Acquiring an
   entity lock raised a `JSONDecodeError` because the worker tried to deserialize

azure-functions-durable/CHANGELOG.md

@@ -9,6 +9,19 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- `DFApp.configure_scheduled_tasks()` opts an app in to durabletask scheduled
+  tasks by registering the schedule entity and operation orchestrator. Once
+  enabled, schedules are managed from a client via
+  `durabletask.scheduled.ScheduledTaskClient`. Scheduled tasks are not
+  registered unless this method is called.
+- `DFApp.configure_history_export()` opts an app in to durabletask history
+  export by registering the export-job entity, driving orchestrator, and
+  activities. Once enabled, export jobs are driven from a client via
+  `durabletask.extensions.history_export.ExportHistoryClient`; supply the
+  activities' runtime dependencies with `history_export.bind_context(...)`.
+  The instance-enumeration activity uses a Functions-specific implementation
+  based on `QueryInstances` because the Durable Functions host extension does
+  not implement the `ListInstanceIds` gRPC call the core activity relies on.
 - `DurableOrchestrationContext.call_http(...)` for making durable HTTP calls
   from orchestrators, restoring the v1 API. The request is executed by a
   built-in activity and, when the endpoint responds with `202 Accepted` and a

azure-functions-durable/azure/durable_functions/decorators/durable_app.py

@@ -90,6 +90,69 @@ def configure_scheduled_tasks(self) -> None:
         self.orchestration_trigger(
             context_name="context")(execute_schedule_operation_orchestrator)
 
+    def configure_history_export(self) -> None:
+        """Opt in to durabletask history export by registering its built-ins.
+
+        Like scheduled tasks, history export is opt-in: its export-job entity,
+        driving orchestrator, and two activities are only registered when this
+        method is called. After calling it, drive export jobs from the client
+        with :class:`durabletask.extensions.history_export.ExportHistoryClient`.
+
+        The runtime dependencies the activities need (a durabletask client and a
+        :class:`~durabletask.extensions.history_export.writer.HistoryWriter`) are
+        supplied separately via
+        :func:`durabletask.extensions.history_export.bind_context`, since the
+        client is only available at request time in the host-driven Functions
+        model.
+
+        The durabletask export activities take a ``(context, input)`` signature;
+        they ignore the context, so they are wrapped as single-input Functions
+        activities to match the host's activity calling convention.
+
+        The enumeration activity uses a Functions-specific implementation
+        (:mod:`azure.durable_functions.internal.history_export_compat`) that
+        queries terminal instances via ``QueryInstances`` instead of the core
+        ``ListInstanceIds`` call, which the Durable Functions host extension
+        does not implement.
+
+        NOTE: We need to consider whether this will perform well on Azure 
+        Functions' distributed architecture - later.
+        """
+        from durabletask.extensions.history_export._constants import (
+            ENTITY_NAME as EXPORT_ENTITY_NAME,
+        )
+        from durabletask.extensions.history_export.activities import (
+            EXPORT_INSTANCE_HISTORY_ACTIVITY,
+            LIST_TERMINAL_INSTANCES_ACTIVITY,
+            export_instance_history,
+        )
+        from durabletask.extensions.history_export.entity import ExportJobEntity
+        from durabletask.extensions.history_export.orchestrator import (
+            export_job_orchestrator,
+        )
+        from ..internal.history_export_compat import (
+            list_terminal_instances,
+        )
+
+        self.entity_trigger(
+            context_name="context", entity_name=EXPORT_ENTITY_NAME)(ExportJobEntity)
+        self.orchestration_trigger(context_name="context")(export_job_orchestrator)
+
+        def _list_terminal_instances(input: dict) -> dict:
+            return list_terminal_instances(None, input)  # type: ignore[arg-type]
+
+        def _export_instance_history(input: dict) -> dict:
+            return export_instance_history(None, input)  # type: ignore[arg-type]
+
+        _list_terminal_instances.__name__ = LIST_TERMINAL_INSTANCES_ACTIVITY
+        _export_instance_history.__name__ = EXPORT_INSTANCE_HISTORY_ACTIVITY
+        self.activity_trigger(
+            input_name="input",
+            activity=LIST_TERMINAL_INSTANCES_ACTIVITY)(_list_terminal_instances)
+        self.activity_trigger(
+            input_name="input",
+            activity=EXPORT_INSTANCE_HISTORY_ACTIVITY)(_export_instance_history)
+
     def _configure_orchestrator_callable(
             self,
             wrap: Callable[[Callable[..., Any]], FunctionBuilder],

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

@@ -0,0 +1,83 @@
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+
+"""Compatibility override for the history-export enumeration activity.
+
+The core durabletask ``list_terminal_instances`` activity enumerates terminal
+instances via the ``ListInstanceIds`` gRPC call. The Azure Functions Durable
+extension's gRPC endpoint does not implement that method (it returns
+``UNIMPLEMENTED``), so this module provides a drop-in replacement that
+enumerates via ``QueryInstances`` (:meth:`get_all_orchestration_states`)
+instead -- a method the extension does implement.
+
+> [!NOTE]
+> 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.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Mapping
+from typing import Any, Optional
+
+from durabletask import task
+from durabletask.client import OrchestrationQuery, OrchestrationStatus
+from durabletask.internal.helpers import ensure_aware
+
+from durabletask.extensions.history_export._internal import dt_from_iso
+from durabletask.extensions.history_export.activities import _require_context
+
+# The activity registers under the same name the export orchestrator calls, so
+# it transparently replaces the core activity.
+LIST_TERMINAL_INSTANCES_ACTIVITY = "list_terminal_instances"
+
+
+def list_terminal_instances(
+        _: task.ActivityContext, input: Mapping[str, Any]) -> dict[str, Any]:
+    """Enumerate terminal instances via ``QueryInstances``.
+
+    Drop-in replacement for the core ``list_terminal_instances`` activity that
+    avoids the unimplemented ``ListInstanceIds`` call. ``QueryInstances`` filters
+    by *created* time whereas the export filters by *completed* time, so the
+    completed-time window is applied client-side against each instance's last
+    update time. Every match is returned in a single page
+    (``continuation_token`` is always ``None``) because
+    ``get_all_orchestration_states`` paginates internally.
+    """
+    ctx = _require_context()
+
+    # A continuation token means the first call already returned everything;
+    # there is no second page under this enumeration strategy.
+    if input.get("continuation_token"):
+        return {"instance_ids": [], "continuation_token": None}
+
+    raw_statuses = input.get("runtime_status")
+    runtime_status_names: Optional[list[str]] = (
+        list(raw_statuses) if raw_statuses is not None else None
+    )
+    completed_time_from = dt_from_iso(input.get("completed_time_from"))
+    completed_time_to = dt_from_iso(input.get("completed_time_to"))
+    if completed_time_from is None:
+        raise ValueError("list_terminal_instances requires 'completed_time_from'")
+
+    runtime_status: Optional[list[OrchestrationStatus]] = None
+    if runtime_status_names is not None:
+        runtime_status = [OrchestrationStatus[name] for name in runtime_status_names]
+
+    states = ctx.client.get_all_orchestration_states(
+        OrchestrationQuery(runtime_status=runtime_status))
+
+    instance_ids: list[str] = []
+    for state in states:
+        completed_at = ensure_aware(state.last_updated_at)
+        if completed_at is not None:
+            if completed_at < completed_time_from:
+                continue
+            if completed_time_to is not None and completed_at > completed_time_to:
+                continue
+        instance_ids.append(state.instance_id)
+
+    return {"instance_ids": instance_ids, "continuation_token": None}

durabletask/extensions/history_export/__init__.py

@@ -22,6 +22,8 @@
 )
 from durabletask.extensions.history_export.activities import (
     HistoryExportContext,
+    bind_context,
+    clear_context,
 )
 from durabletask.extensions.history_export.client import (
     ExportHistoryClient,
@@ -76,5 +78,7 @@
     "ExportMode",
     "HistoryExportContext",
     "HistoryWriter",
+    "bind_context",
+    "clear_context",
     "orchestrator_instance_id_for",
 ]

tests/azure-functions-durable/e2e/_harness.py

@@ -419,6 +419,36 @@ def delete_schedule(self, schedule_id: str) -> None:
             "POST", f"{self.base_url}/api/schedule/{schedule_id}/delete", timeout=90)
         assert result.status == 202, f"delete schedule failed: {result.status} {result.body}"
 
+    def start_export(self, container: str = "exports",
+                     job_id: Optional[str] = None,
+                     completed_from: Optional[str] = None) -> dict[str, Any]:
+        """Start a history-export job via the app's ``/api/export/start`` route.
+
+        ``completed_from`` (an ISO-8601 timestamp) narrows the export window so
+        it only covers instances completed at/after that time.
+        """
+        data: dict[str, Any] = {"container": container}
+        if job_id is not None:
+            data["job_id"] = job_id
+        if completed_from is not None:
+            data["completed_from"] = completed_from
+        result = http_request("POST", f"{self.base_url}/api/export/start", data=data, timeout=90)
+        assert result.status == 200, f"start export failed: {result.status} {result.body}"
+        return result.json()
+
+    def wait_for_export(self, job_id: str, timeout: float = 90) -> dict[str, Any]:
+        """Poll ``/api/export/status/{id}`` until the export job is terminal."""
+        deadline = time.time() + timeout
+        payload: dict[str, Any] = {}
+        while time.time() < deadline:
+            result = http_request("GET", f"{self.base_url}/api/export/status/{job_id}")
+            assert result.status == 200, f"export status failed: {result.status} {result.body}"
+            payload = result.json()
+            if (payload.get("status") or "") in ("Completed", "Failed"):
+                return payload
+            time.sleep(0.5)
+        raise TimeoutError(f"export job {job_id} did not finish within {timeout}s; last: {payload}")
+
     def signal_entity(self, name: str, key: str, op: str, input: Any = None,
                       delay_seconds: Optional[float] = None) -> None:
         """Signal an entity via the app's ``/api/signal/{name}/{key}/{op}`` route.

tests/azure-functions-durable/e2e/apps/dtask_style/function_app.py

@@ -21,6 +21,7 @@
 import activities
 import client_routes
 import entities
+import history_export_routes
 import orchestrators
 
 app = df.DFApp(http_auth_level=func.AuthLevel.ANONYMOUS)
@@ -29,7 +30,12 @@
 app.register_functions(entities.bp)
 app.register_functions(orchestrators.bp)
 app.register_functions(client_routes.bp)
+app.register_functions(history_export_routes.bp)
 
 # Opt in to durabletask scheduled tasks: registers the schedule entity and
 # operation orchestrator so schedules can be managed via ScheduledTaskClient.
 app.configure_scheduled_tasks()
+
+# Opt in to durabletask history export: registers the export-job entity, driving
+# orchestrator, and activities so export jobs can be driven via ExportHistoryClient.
+app.configure_history_export()