Address latest PR review comments from berndverst

Author: andystaples

CHANGELOG.md

@@ -32,8 +32,12 @@ ADDED
   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
+  history while preserving cumulative totals across continue-as-new segments,
+  and re-fetches entity state at the top of every page loop so
   external delete or mark-failed signals stop the orchestrator cleanly.
+  Empty-page BATCH checkpoints no longer reset the persisted resume cursor,
+  and duplicate `mark_failed` signals are now idempotent no-ops when a job
+  is already failed to reduce transition-noise logs.
   `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

durabletask/extensions/history_export/entity.py

@@ -305,6 +305,12 @@ def mark_failed(
         assert_valid_transition(
             self.OP_MARK_FAILED, state.status, ExportJobStatus.FAILED, job_id=job_id,
         )
+        if state.status is ExportJobStatus.FAILED:
+            # Idempotent no-op: if the job is already FAILED, do not
+            # mutate persisted state (especially ``last_error``) and
+            # avoid noisy invalid-transition logs from duplicate
+            # best-effort mark_failed signals.
+            return state.to_dict()
         reason = ""
         if payload is not None:
             reason = str(payload.get("reason", ""))

durabletask/extensions/history_export/orchestrator.py

@@ -145,12 +145,22 @@ def export_job_orchestrator(
     config = ExportJobConfiguration.from_dict(config_mapping)
     initial_checkpoint = input.get("checkpoint") or {"last_instance_key": None}
     processed_cycles = int(input.get("processed_cycles", 0))
+    input_totals_raw = input.get("totals")
+    input_totals: Mapping[str, Any] = (
+        cast("Mapping[str, Any]", input_totals_raw)
+        if isinstance(input_totals_raw, Mapping)
+        else {}
+    )
 
     entity_id = task.EntityInstanceId(ENTITY_NAME, job_id)
     runtime_status_names = [s.name for s in config.filter.effective_runtime_status()]
     continuation_token: str | None = initial_checkpoint.get("last_instance_key")
 
-    totals: dict[str, int] = {"scanned": 0, "exported": 0, "failed": 0}
+    totals: dict[str, int] = {
+        "scanned": int(input_totals.get("scanned", 0)),
+        "exported": int(input_totals.get("exported", 0)),
+        "failed": int(input_totals.get("failed", 0)),
+    }
 
     try:
         while True:
@@ -160,6 +170,7 @@ def export_job_orchestrator(
                     "job_id": job_id,
                     "config": dict(config_mapping),
                     "checkpoint": {"last_instance_key": continuation_token},
+                    "totals": dict(totals),
                     "processed_cycles": 0,
                 })
                 return None
@@ -227,7 +238,6 @@ def export_job_orchestrator(
                         "scanned_delta": 0,
                         "exported_delta": 0,
                         "failed_delta": 0,
-                        "last_instance_key": None,
                     },
                 )
                 break

durabletask/extensions/history_export/transitions.py

@@ -46,6 +46,7 @@
     ("mark_completed", ExportJobStatus.ACTIVE): frozenset({ExportJobStatus.COMPLETED}),
 
     ("mark_failed", ExportJobStatus.ACTIVE): frozenset({ExportJobStatus.FAILED}),
+    ("mark_failed", ExportJobStatus.FAILED): frozenset({ExportJobStatus.FAILED}),
 }
 
 

tests/durabletask/extensions/history_export/test_transitions_and_exceptions.py

@@ -57,10 +57,13 @@ def test_mark_completed_requires_active(self) -> None:
             "mark_completed", ExportJobStatus.FAILED, ExportJobStatus.COMPLETED,
         )
 
-    def test_mark_failed_allowed_from_active(self) -> None:
+    def test_mark_failed_allowed_from_active_or_failed(self) -> None:
         assert is_valid_transition(
             "mark_failed", ExportJobStatus.ACTIVE, ExportJobStatus.FAILED,
         )
+        assert is_valid_transition(
+            "mark_failed", ExportJobStatus.FAILED, ExportJobStatus.FAILED,
+        )
 
     def test_unknown_operation_rejected(self) -> None:
         assert not is_valid_transition(