Schema back to 1.0, drop PENDING state

Author: andystaples

CHANGELOG.md

@@ -27,7 +27,7 @@ ADDED
   history, and re-fetches entity state at the top of every page loop so
   external delete or mark-failed signals stop the orchestrator cleanly.
   Job state lives in a durable entity with an explicit state-transition
-  matrix (PENDING / ACTIVE / COMPLETED / FAILED); invalid transitions raise
+  matrix (ACTIVE / COMPLETED / FAILED); invalid transitions raise
   `ExportJobInvalidTransitionError`. Persisted entity state uses a
   versioned, schema-stable JSON shape (`STATE_SCHEMA_VERSION`) with no
   embedded Python type metadata. Each export job's driving orchestrator

durabletask/extensions/history_export/client.py

@@ -136,11 +136,11 @@ def create_job(
     ) -> ExportJobDescription:
         """Create a new export job and start its driving orchestrator.
 
-        The entity is created in :attr:`ExportJobStatus.PENDING` and
-        immediately signalled with ``run``, which schedules the
+        The entity processes ``create`` by validating the transition,
+        persisting :attr:`ExportJobStatus.ACTIVE`, and scheduling the
         driving orchestrator from inside the entity using a
         deterministic instance ID (``export-job-{job_id}``).  This
-        matches the .NET ``ExportJob.Run`` pattern: callers can
+        matches the .NET ``ExportJob.Create`` pattern: callers can
         correlate a job with its orchestrator by ID alone and may
         safely re-create a previously-terminated job.
         """
@@ -150,10 +150,10 @@ def create_job(
         created_at = datetime.now(timezone.utc)
         config_dict = config.to_dict()
 
-        # Signal create first; the entity will validate the transition
-        # and persist PENDING.  Then signal run; the entity will
-        # schedule the orchestrator and transition to ACTIVE.  Both
-        # signals are processed in FIFO order by the entity dispatcher.
+        # A single ``create`` signal is enough: the entity validates
+        # the transition, persists ACTIVE, and schedules the
+        # orchestrator inline.  Mirrors the .NET ``ExportJob.Create``
+        # flow.
         self._client.signal_entity(
             entity_id,
             ExportJobEntity.OP_CREATE,
@@ -162,14 +162,13 @@ def create_job(
                 "created_at": created_at.isoformat(),
             },
         )
-        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),
         )
         return ExportJobDescription(
             job_id=resolved_job_id,
-            status=ExportJobStatus.PENDING,
+            status=ExportJobStatus.ACTIVE,
             created_at=created_at,
             last_modified_at=created_at,
             config=config,
@@ -285,7 +284,7 @@ def wait_for_job(
         """Poll until the job reaches a terminal status or *timeout* elapses.
 
         Raises:
-            TimeoutError: If the job is still pending/active after
+            TimeoutError: If the job is still active after
                 *timeout* seconds.
             ExportJobNotFoundError: If the job cannot be found at all.
         """

durabletask/extensions/history_export/entity.py

@@ -14,17 +14,16 @@
 Operations
 ----------
 ``create``
-    Initialise a fresh export job, or reset a terminal job back to
-    :attr:`ExportJobStatus.PENDING`.  Refuses to overwrite an active
-    job (raises :class:`ExportJobInvalidTransitionError`).
+    Initialise a fresh export job, or revive a terminal job, by
+    persisting :attr:`ExportJobStatus.ACTIVE` and scheduling the
+    driving orchestrator inline (with a deterministic instance ID
+    derived from the job ID).  Refuses to overwrite an active job
+    (raises :class:`ExportJobInvalidTransitionError`).  Mirrors the
+    .NET ``ExportJob.Create`` flow, so a single signal is enough to
+    launch a job.
 ``get``
     Returns the persisted state dict, or ``None`` if the entity has
     not been created (or has been deleted).
-``run``
-    Schedules the driving orchestrator (with a deterministic instance
-    ID derived from the job ID) and transitions the job to
-    :attr:`ExportJobStatus.ACTIVE`.  Idempotent so the client may
-    safely signal it more than once.
 ``commit_checkpoint``
     Applies an incremental update after a single export page.  When
     ``mark_failed_on_batch`` is true *and* ``failures`` is non-empty,
@@ -101,7 +100,6 @@ class ExportJobEntity(entities.DurableEntity):
 
     OP_CREATE = "create"
     OP_GET = "get"
-    OP_RUN = "run"
     OP_COMMIT_CHECKPOINT = "commit_checkpoint"
     OP_MARK_COMPLETED = "mark_completed"
     OP_MARK_FAILED = "mark_failed"
@@ -138,7 +136,7 @@ def create(self, payload: Mapping[str, Any]) -> dict[str, Any]:
         job_id = self._job_id()
         current = self._current_status()
         assert_valid_transition(
-            self.OP_CREATE, current, ExportJobStatus.PENDING, job_id=job_id,
+            self.OP_CREATE, current, ExportJobStatus.ACTIVE, job_id=job_id,
         )
 
         config_dict = payload.get("config")
@@ -165,69 +163,51 @@ def create(self, payload: Mapping[str, Any]) -> dict[str, Any]:
         # Matches the .NET ``ExportJob.Create`` revive semantics so a
         # re-created job starts from a clean slate.
         state = ExportJobState(
-            status=ExportJobStatus.PENDING,
+            status=ExportJobStatus.ACTIVE,
             config=config,
             created_at=created_at,
             last_modified_at=created_at,
         )
-        logger.info(
-            "Created export job %r in status %s", job_id, state.status.value,
-            extra={"job_id": job_id, "operation": "create"},
-        )
+
+        # The entity itself schedules the driving orchestrator inline,
+        # so a single ``create`` signal is enough to launch a job.
+        # Mirrors the .NET ``ExportJob.Create`` -> ``StartExportOrchestration``
+        # flow and avoids the client having to send a second ``run``
+        # signal (and the failure modes that come with it).
+        instance_id = orchestrator_instance_id_for(job_id)
+        try:
+            self.entity_context.schedule_new_orchestration(
+                ORCHESTRATOR_NAME,
+                input={"job_id": job_id, "config": state.config.to_dict()},
+                instance_id=instance_id,
+            )
+            state.orchestrator_instance_id = instance_id
+            logger.info(
+                "Created export job %r and scheduled orchestrator %s with "
+                "instance ID %s",
+                job_id, ORCHESTRATOR_NAME, instance_id,
+                extra={"job_id": job_id, "operation": "create"},
+            )
+        except Exception as ex:  # noqa: BLE001
+            # Mirror the .NET 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 with no error recorded.
+            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": "create"},
+            )
         return self._save(state)
 
     def get(self, _: Any = None) -> dict[str, Any] | None:
         state = self._load()
         return state.to_dict() if state is not None else None
 
-    def run(self, _: Any = None) -> dict[str, Any] | None:
-        state = self._load()
-        if state is None:
-            raise ValueError("Cannot run uninitialized export job")
-        job_id = self._job_id()
-        assert_valid_transition(
-            self.OP_RUN, state.status, ExportJobStatus.ACTIVE, job_id=job_id,
-        )
-
-        # The entity itself schedules the driving orchestrator.  The
-        # client is therefore decoupled from the orchestrator's name
-        # and input shape.
-        if state.status is ExportJobStatus.PENDING:
-            instance_id = orchestrator_instance_id_for(job_id)
-            try:
-                self.entity_context.schedule_new_orchestration(
-                    ORCHESTRATOR_NAME,
-                    input={"job_id": job_id, "config": state.config.to_dict()},
-                    instance_id=instance_id,
-                )
-                state.orchestrator_instance_id = instance_id
-                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"},
-                )
-                return self._save(state)
-
-        state.status = ExportJobStatus.ACTIVE
-        state.last_error = None
-        return self._save(state)
-
     def commit_checkpoint(self, payload: Mapping[str, Any]) -> dict[str, Any] | None:
         state = self._load()
         if state is None:

durabletask/extensions/history_export/models.py

@@ -63,24 +63,20 @@ class ExportJobStatus(Enum):
 
     Status meanings
     ---------------
-    ``PENDING``
-        The job has been created and persisted but the entity has not
-        yet kicked off its driving orchestrator.  Jobs sit in this
-        state briefly between the ``create`` and ``run`` signals
-        (the public client sends both in immediate succession), or
-        for longer if ``run`` is never invoked or if a caller revives
-        a previously terminal job via ``create``.
     ``ACTIVE``
-        The job is running and the driving orchestrator is making
-        progress through pages of terminal instances.
+        The job has been created and the driving orchestrator is
+        making progress through pages of terminal instances.  This is
+        the initial status after :meth:`ExportHistoryClient.create_job`
+        because the entity schedules the orchestrator inline as part
+        of its ``create`` operation (mirroring the .NET
+        ``ExportJob.Create`` flow).
     ``COMPLETED``
         The orchestrator finished a batch successfully.
     ``FAILED``
         The orchestrator threw, or a page of exports exhausted its
         retries.
     """
 
-    PENDING = "Pending"
     ACTIVE = "Active"
     COMPLETED = "Completed"
     FAILED = "Failed"
@@ -434,22 +430,11 @@ def to_configuration(self) -> ExportJobConfiguration:
 # replaced with a registry keyed by ``(entity_name, schema_version)`` without
 # changing the on-disk shape.
 
-STATE_SCHEMA_VERSION = "1.1"
+STATE_SCHEMA_VERSION = "1.0"
 """The schema version emitted by :meth:`ExportJobState.to_dict`.
 
 Increment this when the persisted shape changes in a non-backward-compatible
 way and add a new branch in :meth:`ExportJobState.from_dict`.
-
-Version history:
-
-``"1.0"``
-    Initial shape.  ``runtime_status`` filter values were persisted as
-    enum *names* (e.g. ``"COMPLETED"``), which broke if the core SDK
-    renamed an enum constant.  Read support retained.
-``"1.1"``
-    ``runtime_status`` filter values are persisted as the protobuf
-    enum *integer* (e.g. ``2`` for ``COMPLETED``).  Reads still accept
-    the legacy 1.0 string form for backward compatibility.
 """
 
 
@@ -502,10 +487,10 @@ def to_dict(self) -> dict[str, Any]:
     @classmethod
     def from_dict(cls, data: Mapping[str, Any]) -> "ExportJobState":
         version = data.get("schema_version", "1.0")
-        if version not in {"1.0", "1.1"}:
+        if version != STATE_SCHEMA_VERSION:
             raise ValueError(
                 f"Unsupported export job state schema_version={version!r}; "
-                f"expected one of: '1.0', '1.1' (current: {STATE_SCHEMA_VERSION!r})"
+                f"expected {STATE_SCHEMA_VERSION!r}"
             )
 
         config_data = data.get("config")

durabletask/extensions/history_export/transitions.py

@@ -28,14 +28,12 @@
 # Maps (operation_name, current_status_or_None) -> {valid target statuses}.
 TRANSITIONS: Mapping[tuple[str, ExportJobStatus | None], frozenset[ExportJobStatus]] = {
     # ``create`` initialises a fresh job and revives terminal jobs.
-    ("create", None): frozenset({ExportJobStatus.PENDING}),
-    ("create", ExportJobStatus.FAILED): frozenset({ExportJobStatus.PENDING}),
-    ("create", ExportJobStatus.COMPLETED): frozenset({ExportJobStatus.PENDING}),
-
-    # ``run`` flips the job from PENDING to ACTIVE.  Idempotent so the
-    # client may signal it more than once without crashing the entity.
-    ("run", ExportJobStatus.PENDING): frozenset({ExportJobStatus.ACTIVE}),
-    ("run", ExportJobStatus.ACTIVE): frozenset({ExportJobStatus.ACTIVE}),
+    # The entity schedules the driving orchestrator inline, so the job
+    # goes straight to ACTIVE without a separate ``run`` signal.
+    # Matches the .NET ``ExportJob.Create`` flow.
+    ("create", None): frozenset({ExportJobStatus.ACTIVE}),
+    ("create", ExportJobStatus.FAILED): frozenset({ExportJobStatus.ACTIVE}),
+    ("create", ExportJobStatus.COMPLETED): frozenset({ExportJobStatus.ACTIVE}),
 
     # ``commit_checkpoint`` is a no-op transition during normal runs.
     # When the orchestrator signals ``mark_failed_on_batch`` the entity
@@ -47,9 +45,6 @@
 
     ("mark_completed", ExportJobStatus.ACTIVE): frozenset({ExportJobStatus.COMPLETED}),
 
-    # ``mark_failed`` from PENDING covers the rare case of a failure
-    # happening between create and run.
-    ("mark_failed", ExportJobStatus.PENDING): frozenset({ExportJobStatus.FAILED}),
     ("mark_failed", ExportJobStatus.ACTIVE): frozenset({ExportJobStatus.FAILED}),
 }
 

tests/durabletask/extensions/history_export/test_client.py

@@ -127,7 +127,7 @@ def test_create_get_and_wait_for_job_end_to_end(
     )
 
     assert desc.job_id
-    assert desc.status == ExportJobStatus.PENDING
+    assert desc.status == ExportJobStatus.ACTIVE
     assert desc.config is not None
     assert desc.orchestrator_instance_id == f"export-job-{desc.job_id}"