PR Feedback 2

Author: andystaples

docs/plan-orchestrationHistoryExport.prompt.md

@@ -27,6 +27,7 @@
     ExportHistoryClient,
     ExportHistoryJobClient,
 )
+from durabletask.extensions.history_export.entity import ExportJobEntity
 from durabletask.extensions.history_export.exceptions import (
     ExportJobError,
     ExportJobInvalidTransitionError,
@@ -65,6 +66,7 @@
     "ExportJobConfiguration",
     "ExportJobCreationOptions",
     "ExportJobDescription",
+    "ExportJobEntity",
     "ExportJobError",
     "ExportJobInvalidTransitionError",
     "ExportJobNotFoundError",
@@ -76,5 +78,3 @@
     "HistoryWriter",
     "orchestrator_instance_id_for",
 ]
-
-PACKAGE_NAME = "durabletask.extensions.history_export"

durabletask/extensions/history_export/__init__.py

@@ -68,8 +68,25 @@ class HistoryExportContext:
 
 
 def bind_context(context: HistoryExportContext) -> None:
-    """Install the runtime dependencies for the history-export activities."""
+    """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
 
 

durabletask/extensions/history_export/activities.py

@@ -53,6 +53,11 @@ class AzureBlobHistoryExportWriterOptions:
             (useful for Azurite compatibility).
         create_container_if_not_exists: When ``True`` (the default),
             ensure the container exists on the first write.
+        overwrite: When ``True`` (the default), each blob upload
+            replaces any existing blob of the same name.  Set to
+            ``False`` for compliance setups that require
+            write-once / immutable exports; in that mode the writer
+            raises if a blob already exists at the target path.
     """
 
     container_name: str
@@ -61,6 +66,7 @@ class AzureBlobHistoryExportWriterOptions:
     credential: Any = field(default=None, repr=False)
     api_version: str | None = None
     create_container_if_not_exists: bool = True
+    overwrite: bool = True
 
     def __post_init__(self) -> None:
         if not self.container_name:
@@ -155,7 +161,7 @@ def write(
         container_client.upload_blob(
             name=blob_name,
             data=payload,
-            overwrite=True,
+            overwrite=self._options.overwrite,
             content_settings=content_settings,
         )
 

durabletask/extensions/history_export/azure_blob.py

@@ -145,7 +145,7 @@ def create_job(
         safely re-create a previously-terminated job.
         """
         config = options.to_configuration()
-        resolved_job_id = job_id or options.job_id or uuid.uuid4().hex
+        resolved_job_id = job_id or uuid.uuid4().hex
         entity_id = entities.EntityInstanceId(ENTITY_NAME, resolved_job_id)
         created_at = datetime.now(timezone.utc)
         config_dict = config.to_dict()
@@ -156,14 +156,13 @@ def create_job(
         # signals are processed in FIFO order by the entity dispatcher.
         self._client.signal_entity(
             entity_id,
-            "create",
+            ExportJobEntity.OP_CREATE,
             input={
                 "config": config_dict,
                 "created_at": created_at.isoformat(),
             },
         )
-        self._client.signal_entity(entity_id, "run")
-
+        self._client.signal_entity(entity_id, ExportJobEntity.OP_RUN)
         logger.info(
             "Submitted export job %r; orchestrator instance ID will be %s",
             resolved_job_id, orchestrator_instance_id_for(resolved_job_id),
@@ -184,7 +183,14 @@ def create_job(
         )
 
     def get_job(self, job_id: str) -> ExportJobDescription | None:
-        """Look up an export job by ID.  Returns ``None`` if not found."""
+        """Look up an export job by ID.  Returns ``None`` if not found.
+
+        Note that the lookup-miss contract differs from
+        :meth:`wait_for_job`: ``get_job`` is a passive read that
+        returns ``None`` when the entity does not exist, while
+        ``wait_for_job`` raises :class:`ExportJobNotFoundError` after
+        its timeout if the entity never appears.
+        """
         entity_id = entities.EntityInstanceId(ENTITY_NAME, job_id)
         meta = self._client.get_entity(entity_id, include_state=True)
         if meta is None:
@@ -235,18 +241,35 @@ def list_jobs(
                 continue
             raw = meta.get_state(str)
             if not raw:
+                logger.warning(
+                    "list_jobs: skipping export-job entity %r with no "
+                    "persisted state", meta.id.key,
+                )
                 continue
             try:
                 state = json.loads(raw)
-            except (TypeError, ValueError):
+            except (TypeError, ValueError) as ex:
+                logger.warning(
+                    "list_jobs: skipping export-job entity %r; failed to "
+                    "parse state JSON (%s)", meta.id.key, ex,
+                )
                 continue
             if not isinstance(state, dict):
+                logger.warning(
+                    "list_jobs: skipping export-job entity %r; persisted "
+                    "state is not a JSON object (got %s)",
+                    meta.id.key, type(state).__name__,
+                )
                 continue
             try:
                 desc = ExportJobDescription.from_state_dict(
                     meta.id.key, cast("dict[str, Any]", state),
                 )
-            except (KeyError, ValueError):
+            except (KeyError, ValueError) as ex:
+                logger.warning(
+                    "list_jobs: skipping export-job entity %r; state did "
+                    "not match the current schema (%s)", meta.id.key, ex,
+                )
                 continue
             if status_filter is not None and desc.status not in status_filter:
                 continue
@@ -289,7 +312,13 @@ def wait_for_job(
             time.sleep(poll_interval)
 
     def delete_job(self, job_id: str) -> None:
-        """Delete the export-job entity, clearing its state.
+        """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`)
@@ -298,7 +327,17 @@ def delete_job(self, job_id: str) -> None:
         This does NOT delete blobs already written to the destination.
         """
         entity_id = entities.EntityInstanceId(ENTITY_NAME, job_id)
-        self._client.signal_entity(entity_id, "delete")
+        self._client.signal_entity(entity_id, ExportJobEntity.OP_DELETE)
+
+    def cancel_job(self, job_id: str) -> None:
+        """Alias for :meth:`delete_job`.
+
+        ``CONTINUOUS`` mode has no natural completion, so users
+        looking to stop a tailing export are likely to look for
+        ``cancel_job`` rather than ``delete_job``.  Provided as a thin
+        alias to make either name discoverable.
+        """
+        self.delete_job(job_id)
 
     # ------------------------------------------------------------------
     # Convenience

durabletask/extensions/history_export/client.py

@@ -92,6 +92,21 @@ def _summarize_failures(failures: list[ExportFailure], *, limit: int = 10) -> st
 class ExportJobEntity(entities.DurableEntity):
     """Durable entity that owns the lifecycle state of one export job."""
 
+    # ----- operation names ------------------------------------------
+    #
+    # Single source of truth for the wire-level entity operation
+    # names.  Clients, the orchestrator, and the transitions matrix
+    # all import these so a typo in any one call site is impossible.
+    # Mirrors the .NET ``nameof(this.Create)`` pattern.
+
+    OP_CREATE = "create"
+    OP_GET = "get"
+    OP_RUN = "run"
+    OP_COMMIT_CHECKPOINT = "commit_checkpoint"
+    OP_MARK_COMPLETED = "mark_completed"
+    OP_MARK_FAILED = "mark_failed"
+    OP_DELETE = "delete"
+
     # ----- state helpers --------------------------------------------
 
     def _load(self) -> ExportJobState | None:
@@ -123,18 +138,32 @@ def create(self, payload: Mapping[str, Any]) -> dict[str, Any]:
         job_id = self._job_id()
         current = self._current_status()
         assert_valid_transition(
-            "create", current, ExportJobStatus.PENDING, job_id=job_id,
+            self.OP_CREATE, current, ExportJobStatus.PENDING, job_id=job_id,
         )
 
         config_dict = payload.get("config")
-        if not config_dict:
+        if config_dict is None:
             raise ValueError("create payload requires 'config'")
-        config = ExportJobConfiguration.from_dict(config_dict)
+        if not isinstance(config_dict, Mapping) or not config_dict:
+            raise ValueError(
+                "create payload 'config' must be a non-empty mapping"
+            )
+        config = ExportJobConfiguration.from_dict(
+            cast("Mapping[str, Any]", config_dict),
+        )
 
         created_at_raw = payload.get("created_at")
         created_at = dt_from_iso(created_at_raw) if created_at_raw else _utcnow()
         assert created_at is not None
 
+        # Reviving a terminal job (COMPLETED / FAILED) constructs a
+        # *fresh* ExportJobState here.  That intentionally resets
+        # every progress field — ``scanned_instances``,
+        # ``exported_instances``, ``failed_instances``,
+        # ``checkpoint.last_instance_key``, ``last_checkpoint_time``,
+        # ``last_error``, and the accumulated ``failures`` list.
+        # Matches the .NET ``ExportJob.Create`` revive semantics so a
+        # re-created job starts from a clean slate.
         state = ExportJobState(
             status=ExportJobStatus.PENDING,
             config=config,
@@ -143,6 +172,7 @@ def create(self, payload: Mapping[str, Any]) -> dict[str, Any]:
         )
         logger.info(
             "Created export job %r in status %s", job_id, state.status.value,
+            extra={"job_id": job_id, "operation": "create"},
         )
         return self._save(state)
 
@@ -156,7 +186,7 @@ def run(self, _: Any = None) -> dict[str, Any] | None:
             raise ValueError("Cannot run uninitialized export job")
         job_id = self._job_id()
         assert_valid_transition(
-            "run", state.status, ExportJobStatus.ACTIVE, job_id=job_id,
+            self.OP_RUN, state.status, ExportJobStatus.ACTIVE, job_id=job_id,
         )
 
         # The entity itself schedules the driving orchestrator.  The
@@ -174,17 +204,25 @@ def run(self, _: Any = None) -> dict[str, Any] | None:
                 logger.info(
                     "Scheduled orchestrator %s for job %r with instance ID %s",
                     ORCHESTRATOR_NAME, job_id, instance_id,
+                    extra={"job_id": job_id, "operation": "run"},
                 )
             except Exception as ex:  # noqa: BLE001
+                # Mirror the .NET ExportJob.StartExportOrchestration pattern:
+                # record the failure on persisted state and return, rather
+                # than re-raising.  Re-raising inside an entity operation
+                # can cause some entity backends to discard the in-flight
+                # state mutations, leaving the job stuck in PENDING with no
+                # error recorded.  Returning ensures FAILED + last_error
+                # actually persist.
                 state.status = ExportJobStatus.FAILED
                 state.last_error = (
                     f"Failed to schedule orchestrator: {type(ex).__name__}: {ex}"
                 )
                 logger.exception(
                     "Failed to schedule orchestrator for export job %r", job_id,
+                    extra={"job_id": job_id, "operation": "run"},
                 )
-                self._save(state)
-                raise
+                return self._save(state)
 
         state.status = ExportJobStatus.ACTIVE
         state.last_error = None
@@ -210,7 +248,7 @@ def commit_checkpoint(self, payload: Mapping[str, Any]) -> dict[str, Any] | None
         will_fail = bool(payload.get("mark_failed_on_batch")) and bool(new_failures)
         target = ExportJobStatus.FAILED if will_fail else ExportJobStatus.ACTIVE
         assert_valid_transition(
-            "commit_checkpoint", state.status, target, job_id=job_id,
+            self.OP_COMMIT_CHECKPOINT, state.status, target, job_id=job_id,
         )
 
         state.scanned_instances += scanned_delta
@@ -240,6 +278,7 @@ def commit_checkpoint(self, payload: Mapping[str, Any]) -> dict[str, Any] | None
             logger.warning(
                 "Export job %r marked FAILED after batch retries (%d failures)",
                 job_id, len(new_failures),
+                extra={"job_id": job_id, "operation": "commit_checkpoint"},
             )
 
         return self._save(state)
@@ -250,12 +289,15 @@ def mark_completed(self, _: Any = None) -> dict[str, Any] | None:
             raise ValueError("Cannot mark_completed on uninitialized export job")
         job_id = self._job_id()
         assert_valid_transition(
-            "mark_completed", state.status, ExportJobStatus.COMPLETED,
+            self.OP_MARK_COMPLETED, state.status, ExportJobStatus.COMPLETED,
             job_id=job_id,
         )
         state.status = ExportJobStatus.COMPLETED
         state.last_error = None
-        logger.info("Export job %r marked COMPLETED", job_id)
+        logger.info(
+            "Export job %r marked COMPLETED", job_id,
+            extra={"job_id": job_id, "operation": "mark_completed"},
+        )
         return self._save(state)
 
     def mark_failed(
@@ -266,21 +308,28 @@ def mark_failed(
             raise ValueError("Cannot mark_failed on uninitialized export job")
         job_id = self._job_id()
         assert_valid_transition(
-            "mark_failed", state.status, ExportJobStatus.FAILED, job_id=job_id,
+            self.OP_MARK_FAILED, state.status, ExportJobStatus.FAILED, job_id=job_id,
         )
         reason = ""
         if payload is not None:
             reason = str(payload.get("reason", ""))
         state.status = ExportJobStatus.FAILED
         state.last_error = reason or None
-        logger.info("Export job %r marked FAILED: %s", job_id, reason or "(no reason)")
+        logger.info(
+            "Export job %r marked FAILED: %s", job_id, reason or "(no reason)",
+            extra={"job_id": job_id, "operation": "mark_failed"},
+        )
         return self._save(state)
 
     def delete(self, _: Any = None) -> None:  # type: ignore[override]
         # The base class's delete() calls set_state(None) which is
         # exactly what we want for export-job cleanup.  ``delete`` is
         # always valid regardless of current status.
-        logger.info("Export job %r deleted", self._job_id())
+        job_id = self._job_id()
+        logger.info(
+            "Export job %r deleted", job_id,
+            extra={"job_id": job_id, "operation": "delete"},
+        )
         super().delete()