PR Feedback 3

Author: andystaples

CHANGELOG.md

@@ -20,12 +20,21 @@ ADDED
   (tail terminal instances indefinitely until stopped via `delete_job`).
   Exported blobs are self-describing: each blob carries an explicit
   `schema_version`, the orchestration's `OrchestrationState` metadata, and
-  the full ordered event list. The export workflow retries each instance up
+  the full ordered event list. Each exported blob also carries
+  `{"instance_id": <id>}` as destination-side metadata (the Azure writer
+  persists this as Azure Blob metadata) so consumers can scan a container
+  without parsing each blob body. The export workflow retries each instance up
   to 3 times with exponential backoff (15s/30s/60s), retries failed batches
   up to 3 times, caps in-flight exports via `max_parallel_exports`
   (default 32), continues-as-new every 5 page cycles to bound orchestrator
   history, and re-fetches entity state at the top of every page loop so
   external delete or mark-failed signals stop the orchestrator cleanly.
+  `delete_job` actively tears the job down: it clears the entity state,
+  terminates the driving orchestrator, waits briefly for it to settle, and
+  purges its orchestration history so a re-created job with the same ID
+  starts from a clean slate. Per-instance exports refuse to write a blob
+  when the target instance has been purged or has re-entered a non-terminal
+  state, surfacing the skipped instance as a per-batch failure.
   Job state lives in a durable entity with an explicit state-transition
   matrix (ACTIVE / COMPLETED / FAILED); invalid transitions raise
   `ExportJobInvalidTransitionError`. Persisted entity state uses a

durabletask/extensions/history_export/activities.py

@@ -48,6 +48,15 @@
 from durabletask.extensions.history_export.writer import HistoryWriter
 
 
+# The set of runtime statuses considered "terminal" by the export
+# activity's safety guard.  Matches the .NET ``IsCompleted`` helper.
+_TERMINAL_RUNTIME_STATUSES: frozenset[client_module.OrchestrationStatus] = frozenset({
+    client_module.OrchestrationStatus.COMPLETED,
+    client_module.OrchestrationStatus.FAILED,
+    client_module.OrchestrationStatus.TERMINATED,
+})
+
+
 # The activity name registered with the worker is simply ``fn.__name__``
 # (see :func:`durabletask.task.get_name`).  These constants exist so
 # downstream code (the orchestrator, tests) can refer to the names
@@ -174,13 +183,39 @@ def export_instance_history(
     prefix: str | None = str(prefix_raw) if prefix_raw is not None else None
 
     try:
-        events = ctx.client.get_orchestration_history(instance_id)
-        # Fetch the orchestration's terminal metadata too so the
-        # exported blob is self-describing (matches the .NET behavior).
+        # Resolve the instance's terminal metadata first.  If the
+        # instance was purged, deleted, or has somehow re-entered a
+        # non-terminal state between ``list_terminal_instances`` and
+        # 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(
             instance_id, fetch_payloads=True,
         )
-        metadata = orchestration_state_to_dict(state) if state is not None else None
+        if state is None:
+            return {
+                "instance_id": instance_id,
+                "success": False,
+                "error": (
+                    f"instance {instance_id!r} no longer exists or has been "
+                    "purged"
+                ),
+            }
+        if state.runtime_status not in _TERMINAL_RUNTIME_STATUSES:
+            return {
+                "instance_id": instance_id,
+                "success": False,
+                "error": (
+                    f"instance {instance_id!r} is no longer terminal "
+                    f"(runtime_status={state.runtime_status.name})"
+                ),
+            }
+
+        events = ctx.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.
+        metadata = orchestration_state_to_dict(state)
         payload = serialize_history(
             events,
             instance_id=instance_id,
@@ -195,6 +230,10 @@ def export_instance_history(
             payload=payload,
             content_type=content_type_for(fmt),
             content_encoding=content_encoding_for(fmt),
+            # Standard hook downstream consumers use to scan a
+            # container without parsing each blob body.  Matches the
+            # .NET writer's ``Metadata["instanceId"]`` convention.
+            metadata={"instance_id": instance_id},
         )
     except Exception as ex:  # noqa: BLE001 - reported back via return value
         return {

durabletask/extensions/history_export/azure_blob.py

@@ -17,6 +17,7 @@
 
 from __future__ import annotations
 
+from collections.abc import Mapping
 from dataclasses import dataclass, field
 from typing import Any
 
@@ -135,6 +136,7 @@ def write(
         payload: bytes,
         content_type: str,
         content_encoding: str | None,
+        metadata: Mapping[str, str] | None = None,
     ) -> None:
         del instance_id  # included by the protocol but not needed here
         # This writer pins to the container configured at construction
@@ -158,11 +160,19 @@ def write(
             if content_encoding
             else ContentSettings(content_type=content_type)
         )
+        # Azure Blob Storage requires the metadata dict to be a plain
+        # ``dict[str, str]`` (the SDK does its own validation).  Copy
+        # whatever the activity passed into the shape the underlying
+        # SDK expects, and pass ``None`` through unchanged so blobs
+        # written via :meth:`write` without metadata behave exactly
+        # the same as they did before this kwarg existed.
+        blob_metadata = dict(metadata) if metadata else None
         container_client.upload_blob(
             name=blob_name,
             data=payload,
             overwrite=self._options.overwrite,
             content_settings=content_settings,
+            metadata=blob_metadata,
         )
 
     # ------------------------------------------------------------------

durabletask/extensions/history_export/client.py

@@ -59,6 +59,8 @@
 from datetime import datetime, timezone
 from typing import Any, cast
 
+import grpc
+
 from durabletask import client as client_module
 from durabletask import entities
 from durabletask import worker as worker_module
@@ -93,6 +95,31 @@
 _TERMINAL_STATUSES = frozenset({ExportJobStatus.COMPLETED, ExportJobStatus.FAILED})
 _ENTITY_ID_PREFIX = f"@{ENTITY_NAME.lower()}@"
 
+# Max seconds :meth:`ExportHistoryClient.delete_job` waits for the
+# driving orchestrator to terminate before continuing on to purge.
+# Sized to be longer than a single ``commit_checkpoint`` round-trip
+# but short enough that a stuck orchestrator cannot block the caller
+# indefinitely.
+_DELETE_WAIT_TIMEOUT_SECONDS = 30.0
+
+
+def _grpc_status(ex: grpc.RpcError) -> grpc.StatusCode | None:
+    """Return the gRPC status code of *ex*, or ``None`` if it is not set.
+
+    The ``code()`` method is declared on the runtime ``grpc.Call``
+    mixin rather than on :class:`grpc.RpcError` itself, so we go
+    through ``getattr`` to keep both pyright and runtime happy when a
+    test backend raises a bare ``RpcError``.
+    """
+    code = getattr(ex, "code", None)
+    if not callable(code):
+        return None
+    try:
+        result = code()
+    except Exception:  # noqa: BLE001 - defensive, never re-raise here
+        return None
+    return result if isinstance(result, grpc.StatusCode) else None
+
 
 __all__ = ["ExportHistoryClient", "ExportHistoryJobClient"]
 
@@ -311,23 +338,72 @@ def wait_for_job(
             time.sleep(poll_interval)
 
     def delete_job(self, job_id: str) -> None:
-        """Request deletion of the export-job entity, clearing its state.
-
-        This call is **best-effort and fire-and-forget**: it enqueues a
-        ``delete`` signal on the entity but does not wait for the
-        entity dispatcher to process it.  Callers that need
-        confirmation should poll :meth:`get_job` and wait for it to
-        return ``None``.
-
-        The driving orchestrator will detect the deletion at its next
-        loop iteration (via :meth:`OrchestrationContext.call_entity`)
-        and exit cleanly without issuing further signals.
+        """Stop and delete an export job.
+
+        The call performs the full teardown sequence (matching the
+        .NET ``DefaultExportHistoryJobClient.DeleteAsync`` flow):
+
+        1.  Signal the entity to clear its persisted state
+            (``ExportJobEntity.OP_DELETE``).
+        2.  Terminate the driving orchestrator so it stops issuing
+            further activity calls and entity signals.
+        3.  Wait briefly for the orchestrator to actually reach a
+            terminal state.
+        4.  Purge the orchestration history so a re-created job with
+            the same ID can start from a clean slate.
+
+        Steps 2–4 are best-effort: each tolerates a missing
+        orchestrator (the job may never have run, or already been
+        purged) by swallowing gRPC ``NOT_FOUND`` errors.  Step 3
+        tolerates a slow termination by logging and continuing rather
+        than blocking the caller indefinitely.
 
         This does NOT delete blobs already written to the destination.
         """
         entity_id = entities.EntityInstanceId(ENTITY_NAME, job_id)
+        orch_instance_id = orchestrator_instance_id_for(job_id)
+
+        # Step 1: clear the persisted entity state.
         self._client.signal_entity(entity_id, ExportJobEntity.OP_DELETE)
 
+        # Step 2: terminate the driving orchestrator so it stops
+        # issuing activity calls and entity signals.
+        try:
+            self._client.terminate_orchestration(
+                orch_instance_id, recursive=True,
+            )
+        except grpc.RpcError as ex:
+            if _grpc_status(ex) != grpc.StatusCode.NOT_FOUND:
+                raise
+
+        # Step 3: wait briefly for the orchestration to settle so the
+        # subsequent purge actually removes its history.  Capped by
+        # ``_DELETE_WAIT_TIMEOUT_SECONDS`` so a stuck orchestrator
+        # cannot block the caller indefinitely; a slow termination is
+        # logged rather than re-raised.
+        try:
+            self._client.wait_for_orchestration_completion(
+                orch_instance_id, timeout=_DELETE_WAIT_TIMEOUT_SECONDS,
+            )
+        except TimeoutError:
+            logger.warning(
+                "Export job %r orchestrator %r did not terminate within %ss; "
+                "continuing with purge anyway",
+                job_id, orch_instance_id, _DELETE_WAIT_TIMEOUT_SECONDS,
+            )
+        except grpc.RpcError as ex:
+            if _grpc_status(ex) != grpc.StatusCode.NOT_FOUND:
+                raise
+
+        # Step 4: purge the orchestration history.
+        try:
+            self._client.purge_orchestration(
+                orch_instance_id, recursive=True,
+            )
+        except grpc.RpcError as ex:
+            if _grpc_status(ex) != grpc.StatusCode.NOT_FOUND:
+                raise
+
     def cancel_job(self, job_id: str) -> None:
         """Alias for :meth:`delete_job`.
 

durabletask/extensions/history_export/entity.py

@@ -211,7 +211,15 @@ def get(self, _: Any = None) -> dict[str, Any] | None:
     def commit_checkpoint(self, payload: Mapping[str, Any]) -> dict[str, Any] | None:
         state = self._load()
         if state is None:
-            raise ValueError("Cannot commit_checkpoint on uninitialized export job")
+            # The entity was deleted between the orchestrator's
+            # mid-loop ``get`` call and this checkpoint (the race
+            # window is small but real for CONTINUOUS jobs cancelled
+            # via :meth:`ExportHistoryClient.delete_job`).  Return
+            # ``None`` so the orchestrator's ``call_entity`` resolves
+            # cleanly and the loop exits via its normal
+            # "entity gone" path rather than raising and triggering
+            # an outer ``mark_failed`` on an already-deleted entity.
+            return None
         job_id = self._job_id()
 
         # commit_checkpoint may transition ACTIVE -> ACTIVE (no-op) or
@@ -266,7 +274,10 @@ def commit_checkpoint(self, payload: Mapping[str, Any]) -> dict[str, Any] | None
     def mark_completed(self, _: Any = None) -> dict[str, Any] | None:
         state = self._load()
         if state is None:
-            raise ValueError("Cannot mark_completed on uninitialized export job")
+            # Entity vanished mid-flight (see ``commit_checkpoint``
+            # for the race description); silently succeed so the
+            # orchestrator's final ``call_entity`` does not raise.
+            return None
         job_id = self._job_id()
         assert_valid_transition(
             self.OP_MARK_COMPLETED, state.status, ExportJobStatus.COMPLETED,
@@ -285,7 +296,11 @@ def mark_failed(
     ) -> dict[str, Any] | None:
         state = self._load()
         if state is None:
-            raise ValueError("Cannot mark_failed on uninitialized export job")
+            # Entity vanished mid-flight (see ``commit_checkpoint``
+            # for the race description); silently succeed so the
+            # orchestrator's best-effort failure report does not
+            # raise on an already-deleted entity.
+            return None
         job_id = self._job_id()
         assert_valid_transition(
             self.OP_MARK_FAILED, state.status, ExportJobStatus.FAILED, job_id=job_id,