Fix durabletask issues surfaced during Azure Functions investigation (#174)

* Apply durabletask fixes surfaced during Azure Functions investigation

Author: andystaples

CHANGELOG.md

@@ -10,6 +10,15 @@ adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 ADDED
 
 - Added `TaskHubGrpcClient.rewind_orchestration()` to rewind a failed orchestration instance to its last known good state. Failed activity and sub-orchestration results are removed from the history and the orchestration replays from the last successful checkpoint, retrying only the failed work. The in-memory testing backend supports rewind as well.
+- Added a `Task.result` property as a read-only alias for `Task.get_result()`.
+- Exported `bind_context` and `clear_context` from `durabletask.extensions.history_export` for managing the ambient history-export context.
+
+FIXED
+
+- Fixed timers created with a timezone-aware `fire_at` datetime being compared against the orchestration's naive UTC clock, which could cause timers to fire at the wrong time. Timezone-aware fire times are now normalized to naive UTC before scheduling.
+- Fixed the orchestration version being set from an unset `executionStarted.version` field. Presence is now checked with `HasField`, so `OrchestrationContext` no longer reports a version when none was provided.
+- Fixed orchestrations failing with `OrchestrationStateError` when a `genericEvent` history event was replayed (for example, the marker the Durable Functions extension appends when rewinding an orchestration). Such informational events are now ignored during replay, matching the .NET worker.
+- Fixed a lock-granted entity response over the legacy entity protocol raising while trying to deserialize an empty operation result. Lock-granted events no longer attempt result deserialization.
 
 ## v1.7.2
 

durabletask/extensions/history_export/__init__.py

@@ -22,6 +22,8 @@
 )
 from durabletask.extensions.history_export.activities import (
     HistoryExportContext,
+    bind_context,
+    clear_context,
 )
 from durabletask.extensions.history_export.client import (
     ExportHistoryClient,
@@ -76,5 +78,7 @@
     "ExportMode",
     "HistoryExportContext",
     "HistoryWriter",
+    "bind_context",
+    "clear_context",
     "orchestrator_instance_id_for",
 ]

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/task.py

@@ -508,6 +508,11 @@ def is_failed(self) -> bool:
         """Returns True if the task has failed, False otherwise."""
         return self._exception is not None
 
+    @property
+    def result(self) -> T:
+        """Returns the result of the task (alias for :meth:`get_result`)."""
+        return self.get_result()
+
     def get_result(self) -> T:
         """Returns the result of the task."""
         if not self._is_complete:

durabletask/worker.py

@@ -1434,8 +1434,7 @@ def _execute_entity_batch(
             stub.CompleteEntityTask(batch_result)
         except Exception as ex:
             self._logger.exception(
-                f"Failed to deliver entity response for '{entity_instance_id}' of orchestration ID '{instance_id}' to sidecar: {ex}"
-            )
+                f"Failed to deliver entity response for '{entity_instance_id}' of orchestration ID '{instance_id}' to sidecar: {ex}")
 
         # TODO: Reset context
 
@@ -1674,6 +1673,14 @@ def create_timer_internal(
         else:
             final_fire_at = fire_at
 
+        # Normalize timezone-aware datetimes to naive UTC so they can be safely
+        # compared against and combined with the orchestration's naive UTC clock.
+        # A datetime is only truly aware when utcoffset() returns a value; a
+        # tzinfo whose utcoffset() is None is still naive and must be left as-is
+        # (calling astimezone() on it would raise ValueError).
+        if final_fire_at.utcoffset() is not None:
+            final_fire_at = final_fire_at.astimezone(timezone.utc).replace(tzinfo=None)
+
         next_fire_at: datetime = final_fire_at
 
         if (
@@ -2350,7 +2357,7 @@ def process_event(
                         f"A '{event.executionStarted.name}' orchestrator was not registered."
                     )
 
-                if event.executionStarted.version:
+                if event.executionStarted.HasField("version"):
                     ctx._version = event.executionStarted.version.value  # pyright: ignore[reportPrivateUsage]
 
                 # Store the parent orchestration instance ID (set for
@@ -2913,6 +2920,12 @@ def _cancel_timer() -> None:
             elif event.HasField("executionRewound"):
                 # Informational event added when an orchestration is rewound. No action needed.
                 pass
+            elif event.HasField("genericEvent"):
+                # Informational history event with no execution semantics (for
+                # example, the marker the Durable Functions extension appends
+                # when rewinding an orchestration). Ignored during replay,
+                # matching the .NET worker.
+                pass
             elif event.HasField("eventSent"):
                 # Check if this eventSent corresponds to an entity operation call after being translated to the old
                 # entity protocol by the Durable WebJobs extension. If so, treat this message similarly to
@@ -2985,12 +2998,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]