From a6cf7e811c7241dc5d9491209f8cd96a1e565490 Mon Sep 17 00:00:00 2001 From: Andy Staples Date: Tue, 14 Jul 2026 11:26:34 -0600 Subject: [PATCH 1/2] Apply durabletask fixes surfaced during Azure Functions investigation --- CHANGELOG.md | 9 ++++ .../extensions/history_export/__init__.py | 4 ++ durabletask/scheduled/models.py | 47 +++++++++++++++---- durabletask/scheduled/orchestrator.py | 30 ++++++++++-- durabletask/task.py | 5 ++ durabletask/worker.py | 21 +++++++-- 6 files changed, 98 insertions(+), 18 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index e9a7c40e..40d4c37b 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/durabletask/extensions/history_export/__init__.py b/durabletask/extensions/history_export/__init__.py index b7fdf0bf..1d3c846a 100644 --- a/durabletask/extensions/history_export/__init__.py +++ b/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", ] diff --git a/durabletask/scheduled/models.py b/durabletask/scheduled/models.py index b0ad5430..c0373f1c 100644 --- a/durabletask/scheduled/models.py +++ b/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) diff --git a/durabletask/scheduled/orchestrator.py b/durabletask/scheduled/orchestrator.py index 0493b744..6bf8b414 100644 --- a/durabletask/scheduled/orchestrator.py +++ b/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, diff --git a/durabletask/task.py b/durabletask/task.py index 8fc7d93e..8c5d495b 100644 --- a/durabletask/task.py +++ b/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: diff --git a/durabletask/worker.py b/durabletask/worker.py index 491fc48a..89fcd15c 100644 --- a/durabletask/worker.py +++ b/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,11 @@ 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. + if final_fire_at.tzinfo 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 +2354,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 +2917,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 +2995,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": }, # 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] From 2bf030e37150ca2fff337d985b0ec96aa5c9bbc3 Mon Sep 17 00:00:00 2001 From: Andy Staples Date: Tue, 14 Jul 2026 12:18:02 -0600 Subject: [PATCH 2/2] Use utcoffset() to detect truly-aware timer fire_at (PR review) --- durabletask/worker.py | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/durabletask/worker.py b/durabletask/worker.py index 89fcd15c..ce9851b6 100644 --- a/durabletask/worker.py +++ b/durabletask/worker.py @@ -1675,7 +1675,10 @@ def create_timer_internal( # Normalize timezone-aware datetimes to naive UTC so they can be safely # compared against and combined with the orchestration's naive UTC clock. - if final_fire_at.tzinfo is not None: + # 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