More E2E test edge cases, fixes, add scheduled tasks

Author: andystaples

CHANGELOG.md

@@ -20,6 +20,11 @@ CHANGED
 
 FIXED
 
+- 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
+  an operation result from the lock-granted event, which carries none; the
+  result is now only read for entity operation calls, not lock acquisitions.
 - Fixed `OrchestrationContext.version` returning an empty string (`''`) instead
   of `None` for orchestrations started without an explicit version. The version
   field is a protobuf wrapper (a singular message that is always truthy), so it

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

@@ -67,6 +67,29 @@ def _register_builtin_http_functions(self) -> None:
             context_name="context",
             orchestration=BUILTIN_HTTP_POLL_ORCHESTRATOR_NAME)(builtin_http_poll_orchestrator)
 
+    def configure_scheduled_tasks(self) -> None:
+        """Opt in to durabletask scheduled tasks by registering their built-ins.
+
+        Unlike durable HTTP (which is always available), scheduled tasks are
+        opt-in: most apps don't use them, so their schedule entity and
+        operation orchestrator are only registered when this method is called.
+        After calling it, manage schedules from the client with
+        :class:`durabletask.scheduled.ScheduledTaskClient`.
+
+        The schedule entity is self-driving (it re-arms itself with delayed
+        self-signals), so no additional worker configuration is required in the
+        host-driven Functions model.
+        """
+        from durabletask.scheduled.orchestrator import (
+            execute_schedule_operation_orchestrator,
+        )
+        from durabletask.scheduled.schedule_entity import ENTITY_NAME, Schedule
+
+        self.entity_trigger(
+            context_name="context", entity_name=ENTITY_NAME)(Schedule)
+        self.orchestration_trigger(
+            context_name="context")(execute_schedule_operation_orchestrator)
+
     def _configure_orchestrator_callable(
             self,
             wrap: Callable[[Callable[..., Any]], FunctionBuilder],

durabletask/scheduled/models.py

@@ -8,7 +8,7 @@
 
 from durabletask.internal.helpers import ensure_aware
 from durabletask.scheduled.schedule_status import ScheduleStatus
-from durabletask.serialization import JsonDataConverter
+from durabletask.serialization import DataConverter, JsonDataConverter
 
 MINIMUM_INTERVAL = timedelta(seconds=1)
 
@@ -425,13 +425,23 @@ def to_json(self) -> dict[str, Any]:
         }
 
     @classmethod
-    def from_json(cls, data: dict[str, Any]) -> "ScheduleState":
-        # The nested configuration is reconstructed by calling its own
-        # ``from_json`` hook directly. ``ScheduleConfiguration`` is an internal
-        # type, so there is no need to route it through a (possibly custom)
-        # converter -- keeping this hook converter-free means it round-trips
-        # under any code path, not only the worker's threaded converter. Reads
-        # accept both the .NET-compatible and legacy snake_case shapes.
+    def from_json(cls, data: dict[str, Any],
+                  converter: DataConverter | None = None) -> "ScheduleState":
+        # Reads accept both the .NET-compatible and legacy snake_case shapes.
+        #
+        # The nested ``ScheduleConfiguration`` must round-trip under any
+        # conforming converter. Converters differ in how they hand nested
+        # custom objects to this hook:
+        #   * The default JSON converter leaves nested values as plain dicts and
+        #     expects the parent hook to rebuild them; it passes itself as
+        #     ``converter`` (a hook that declares the parameter opts in) so the
+        #     reconstruction can defer to ``converter.coerce``.
+        #   * The Azure Functions ``df`` codec rebuilds nested ``to_json`` /
+        #     ``from_json`` envelopes bottom-up, so it invokes this hook with the
+        #     configuration *already* reconstructed (and without a converter).
+        # Handle both: skip reconstruction when it is already a
+        # ``ScheduleConfiguration``, otherwise route a raw dict through the
+        # converter when one was supplied, falling back to the hook directly.
         state = cls()
         state.status = ScheduleStatus.from_dotnet(_get(data, "Status", "status"))
         # Preserve the token generated by ``__init__`` when the field is absent;
@@ -446,8 +456,7 @@ def from_json(cls, data: dict[str, Any]) -> "ScheduleState":
         state.schedule_last_modified_at = _from_iso(
             _get(data, "ScheduleLastModifiedAt", "schedule_last_modified_at"))
         config_data = _get(data, "ScheduleConfiguration", "schedule_configuration")
-        state.schedule_configuration = (
-            ScheduleConfiguration.from_json(config_data) if config_data is not None else None)
+        state.schedule_configuration = _rebuild_configuration(config_data, converter)
         return state
 
     def to_description(self) -> ScheduleDescription:
@@ -470,3 +479,21 @@ def to_description(self) -> ScheduleDescription:
 
 def _new_token() -> str:
     return uuid.uuid4().hex
+
+
+def _rebuild_configuration(
+        config_data: Any, converter: DataConverter | None) -> "ScheduleConfiguration | None":
+    """Reconstruct a nested ``ScheduleConfiguration`` for any converter.
+
+    ``config_data`` may be ``None``, an already-reconstructed
+    ``ScheduleConfiguration`` (codecs that rebuild nested envelopes bottom-up,
+    e.g. the Azure Functions ``df`` codec), or a plain mapping (the default JSON
+    converter, which leaves nested values as dicts). When a ``converter`` is
+    supplied its ``coerce`` drives reconstruction; otherwise the hook is called
+    directly.
+    """
+    if config_data is None or isinstance(config_data, ScheduleConfiguration):
+        return config_data
+    if converter is not None:
+        return converter.coerce(config_data, ScheduleConfiguration)
+    return ScheduleConfiguration.from_json(config_data)

durabletask/scheduled/orchestrator.py

@@ -13,16 +13,38 @@
 class ScheduleOperationRequest:
     """Request describing an operation to execute against a schedule entity.
 
-    A plain dataclass: the serializer round-trips it automatically. ``input`` is
-    typed ``Any``, so it is reconstructed as the raw deserialized payload; the
-    concrete options type is rebuilt later, at the entity-method boundary, from
-    that method's parameter annotation.
+    ``input`` is typed ``Any``, so it is reconstructed as the raw deserialized
+    payload; the concrete options type is rebuilt later, at the entity-method
+    boundary, from that method's parameter annotation.
+
+    The ``to_json`` / ``from_json`` hooks mirror the plain dataclass field
+    mapping so the wire format is unchanged for the default JSON converter,
+    while also making the type serializable by converters that require an
+    explicit hook (for example the Azure Functions ``df`` codec, which cannot
+    serialize a bare dataclass). This matches the sibling schedule models
+    (``ScheduleState``, ``ScheduleCreationOptions``), which already define these
+    hooks.
     """
 
     entity_id: str
     operation_name: str
     input: Any | None = None
 
+    def to_json(self) -> dict[str, Any]:
+        return {
+            "entity_id": self.entity_id,
+            "operation_name": self.operation_name,
+            "input": self.input,
+        }
+
+    @classmethod
+    def from_json(cls, data: dict[str, Any]) -> "ScheduleOperationRequest":
+        return cls(
+            entity_id=data["entity_id"],
+            operation_name=data["operation_name"],
+            input=data.get("input"),
+        )
+
 
 def execute_schedule_operation_orchestrator(
         ctx: task.OrchestrationContext,

durabletask/worker.py

@@ -2857,12 +2857,15 @@ def _handle_entity_event_raised(self,
             return
 
         result = None
-        if response is not None:
+        if response is not None and not is_lock_event:
             # The legacy protocol wraps the result as {"result": <serialized>},
             # where the value is a serialized JSON string (like the new protocol's
             # entityOperationCompleted.output). Deserialize it -- not coerce -- so
             # the value is fully parsed and the expected type applied; coercing
             # would skip JSON parsing and leave it double-encoded (e.g. '"done"').
+            # Skipped for lock-granted events: those carry no operation result
+            # (the payload's "result" is empty), and the lock branch below
+            # ignores ``result`` entirely, so deserializing it would only raise.
             result = self._data_converter.deserialize(
                 response["result"],
                 entity_task._expected_type,  # pyright: ignore[reportPrivateUsage]

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

@@ -379,10 +379,58 @@ def read_entity(self, name: str, key: str) -> dict[str, Any]:
         assert result.status == 200, f"entity read failed: {result.status} {result.body}"
         return result.json()
 
-    def signal_entity(self, name: str, key: str, op: str, input: Any = None) -> None:
-        """Signal an entity via the app's ``/api/signal/{name}/{key}/{op}`` route."""
+    def list_entities(self, starts_with: Optional[str] = None) -> dict[str, Any]:
+        """List entities via the app's ``/api/entities`` route.
+
+        An optional ``starts_with`` filters by the entity instance-id prefix
+        (entity IDs are formatted ``@name@key``).
+        """
+        url = f"{self.base_url}/api/entities"
+        if starts_with is not None:
+            url += f"?starts_with={urllib.parse.quote(starts_with)}"
+        result = http_request("GET", url)
+        assert result.status == 200, f"list entities failed: {result.status} {result.body}"
+        return result.json()
+
+    def clean_entity_storage(self) -> dict[str, Any]:
+        """Trigger entity storage cleanup via the app's ``/api/clean-entities`` route."""
+        result = http_request("POST", f"{self.base_url}/api/clean-entities")
+        assert result.status == 200, f"clean entities failed: {result.status} {result.body}"
+        return result.json()
+
+    def create_schedule(self, schedule_id: str, interval_seconds: float = 2,
+                        input: Any = None) -> dict[str, Any]:
+        """Create a scheduled task via the app's ``/api/schedule/{id}`` route."""
+        result = http_request(
+            "POST", f"{self.base_url}/api/schedule/{schedule_id}",
+            data={"interval_seconds": interval_seconds, "input": input}, timeout=90)
+        assert result.status == 200, f"create schedule failed: {result.status} {result.body}"
+        return result.json()
+
+    def describe_schedule(self, schedule_id: str) -> dict[str, Any]:
+        """Describe a scheduled task via the app's ``/api/schedule/{id}`` route."""
+        result = http_request("GET", f"{self.base_url}/api/schedule/{schedule_id}", timeout=90)
+        assert result.status == 200, f"describe schedule failed: {result.status} {result.body}"
+        return result.json()
+
+    def delete_schedule(self, schedule_id: str) -> None:
+        """Delete a scheduled task via the app's ``/api/schedule/{id}/delete`` route."""
+        result = http_request(
+            "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 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.
+
+        When ``delay_seconds`` is provided, the app schedules the signal for
+        future delivery (a delayed/scheduled entity signal).
+        """
+        data: dict[str, Any] = {"input": input}
+        if delay_seconds is not None:
+            data["delay_seconds"] = delay_seconds
         result = http_request(
-            "POST", f"{self.base_url}/api/signal/{name}/{key}/{op}", data={"input": input})
+            "POST", f"{self.base_url}/api/signal/{name}/{key}/{op}", data=data)
         assert result.status == 202, f"signal failed: {result.status} {result.body}"
 
     def wait_for_entity(