Merge remote-tracking branch 'origin/main' into andystaples/add-functions-support

# Conflicts:
#	CHANGELOG.md
#	durabletask/worker.py

Author: andystaples

CHANGELOG.md

@@ -10,12 +10,23 @@ adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 ADDED
 
 - Added a `result` property to `Task` as a convenience alias for `get_result()`.
+- Added `OrchestrationContext.parent_instance_id`, which returns the instance
+  ID of the parent orchestration for a sub-orchestration, or `None` for a
+  top-level orchestration.
+
+CHANGED
+
+- Changed the default large-payload externalization threshold (`LargePayloadStorageOptions.threshold_bytes`) from 900,000 bytes to 262,144 bytes (256 KiB), matching the .NET SDK default. Behavioral change (not source/binary breaking): payloads larger than 256 KiB are now externalized by default.
 
 FIXED
 
 - `OrchestrationContext.create_timer` now accepts timezone-aware `datetime`
   values, normalizing them to UTC instead of raising when compared against the
   orchestration's internal clock.
+- Fixed `OrchestrationContext.call_entity` not propagating entity operation failures when running under the legacy entity protocol (used by the Azure Functions Durable extension). A failed entity operation now raises `TaskFailedError` in the calling orchestration instead of silently completing, matching the behavior of the current entity protocol and the .NET SDK.
+- Fixed `OrchestrationContext.call_entity` returning a double-encoded result under the legacy entity protocol. The entity's return value was left as a raw serialized JSON string (for example, a returned string arrived with extra quotes and dicts/lists arrived as strings); it is now fully deserialized and coerced to the requested `return_type`, matching the current entity protocol.
+- Fixed unbounded growth of the internal entity request/lock tracking maps when using entities over the legacy entity protocol. Entries are now released as each response is handled, reducing memory use in long-running orchestrations that call or lock entities.
+- Fixed schedules created with `durabletask.scheduled` not appearing in the Durable Task Scheduler dashboard. The schedule entity state is now persisted in a format compatible with the .NET SDK: the `status` is serialized as its numeric enum value, the `interval` as a .NET `TimeSpan` string, and the `orchestration_input` as a JSON-encoded string, all using .NET property names, so the dashboard can read it without a JSON deserialization error.
 
 ## v1.7.0
 

azure-functions-durable/azure/durable_functions/internal/compat/orchestration_context.py

@@ -87,14 +87,13 @@ def version(self) -> Optional[str]:
         return self._ctx.version
 
     @property
-    def parent_instance_id(self) -> str:
+    def parent_instance_id(self) -> Optional[str]:
         """Get the ID of the parent orchestration.
 
-        Not available: durabletask does not currently surface the parent
-        instance ID on the orchestration context.
+        Returns ``None`` for a top-level orchestration (i.e. one that was not
+        started as a sub-orchestration).
         """
-        raise NotImplementedError(
-            "parent_instance_id is not currently exposed by durabletask.")
+        return self._ctx.parent_instance_id
 
     @property
     def function_context(self) -> Any:

docs/features.md

@@ -285,7 +285,7 @@ pip install durabletask[azure-blob-payloads]
 #### How it works
 
 1. When the worker or client sends a payload that exceeds the
-   configured threshold (default 900 KB), the payload is
+   configured threshold (default 256 KiB), the payload is
    compressed (GZip, enabled by default) and uploaded to the
    external store.
 2. The original payload in the gRPC message is replaced with a
@@ -307,7 +307,7 @@ from durabletask.extensions.azure_blob_payloads import BlobPayloadStore, BlobPay
 store = BlobPayloadStore(BlobPayloadStoreOptions(
     connection_string="DefaultEndpointsProtocol=https;...",
     container_name="durabletask-payloads",  # default
-    threshold_bytes=900_000,                # default (900 KB)
+    threshold_bytes=262_144,                # default (256 KiB)
     max_stored_payload_bytes=10_485_760,    # default (10 MB)
     enable_compression=True,                # default
 ))
@@ -351,7 +351,7 @@ store = BlobPayloadStore(BlobPayloadStoreOptions(
 
 | Option | Default | Description |
 |---|---|---|
-| `threshold_bytes` | 900,000 (900 KB) | Payloads larger than this are externalized |
+| `threshold_bytes` | 262,144 (256 KiB) | Payloads larger than this are externalized |
 | `max_stored_payload_bytes` | 10,485,760 (10 MB) | Maximum size for externalized payloads |
 | `enable_compression` | `True` | GZip-compress payloads before uploading |
 | `container_name` | `"durabletask-payloads"` | Azure Blob container name |

docs/supported-patterns.md

@@ -217,7 +217,7 @@ with DurableTaskSchedulerWorker(
     state = c.wait_for_orchestration_completion(instance_id, timeout=60)
 ```
 
-In this example, any payload exceeding the threshold (default 900 KB) is compressed and uploaded to
+In this example, any payload exceeding the threshold (default 256 KiB) is compressed and uploaded to
 the configured Azure Blob container. When the worker or client reads the message, it downloads and
 decompresses the payload automatically.
 

durabletask/internal/helpers.py

@@ -3,6 +3,7 @@
 
 import traceback
 from datetime import datetime, timezone
+from typing import Any, cast
 
 from google.protobuf import timestamp_pb2, wrappers_pb2
 
@@ -136,6 +137,59 @@ def new_failure_details(ex: Exception, _visited: set[int] | None = None) -> pb.T
     )
 
 
+def _failure_details_from_core_dict(fd: dict[str, Any]) -> pb.TaskFailureDetails:
+    """Convert a serialized DurableTask.Core ``FailureDetails`` dict to protobuf."""
+    inner = fd.get("InnerFailure")
+    stack_trace = fd.get("StackTrace")
+    return pb.TaskFailureDetails(
+        errorType=str(fd.get("ErrorType") or ""),
+        errorMessage=str(fd.get("ErrorMessage") or ""),
+        stackTrace=get_string_value(str(stack_trace) if stack_trace is not None else None),
+        innerFailure=_failure_details_from_core_dict(cast(dict[str, Any], inner)) if isinstance(inner, dict) else None,
+        isNonRetriable=bool(fd.get("IsNonRetriable", False)),
+    )
+
+
+def entity_response_failure_details(
+        response: dict[str, Any],
+        error_content: Any = None) -> pb.TaskFailureDetails:
+    """Build failure details from a failed legacy-protocol entity ``ResponseMessage``.
+
+    Call this only for responses that :func:`is_entity_error_response` reports as
+    failures. In the WebJobs "old protocol" ``ResponseMessage`` (see
+    ``EntityScheduler/ResponseMessage.cs``), a failed operation serializes the
+    human-readable content into ``result`` while ``exceptionType`` carries only
+    the exception's type name (a presence marker) -- it is *not* the message.
+    This mirrors ``azure-functions-durable-python`` / ``-js``, which read the
+    message from ``result`` and ignore ``exceptionType``'s value.
+
+    Parameters
+    ----------
+    response:
+        The deserialized ``ResponseMessage`` dict.
+    error_content:
+        The already-deserialized ``result`` payload, used as the failure
+        message. A structured ``failureDetails`` object, if present, takes
+        precedence (current-protocol shape).
+    """
+    failure_details = response.get("failureDetails")
+    if isinstance(failure_details, dict):
+        return _failure_details_from_core_dict(cast(dict[str, Any], failure_details))
+    error_type = str(response.get("exceptionType") or "")
+    error_message = "" if error_content is None else str(error_content)
+    return pb.TaskFailureDetails(errorType=error_type, errorMessage=error_message)
+
+
+def is_entity_error_response(response: dict[str, Any]) -> bool:
+    """Return ``True`` if a legacy-protocol entity ``ResponseMessage`` is a failure.
+
+    In the WebJobs "old protocol" a failed operation is marked by the presence of
+    an ``exceptionType`` field (successful responses omit it). Current-protocol
+    payloads may instead carry a structured ``failureDetails`` object.
+    """
+    return "exceptionType" in response or isinstance(response.get("failureDetails"), dict)
+
+
 def new_event_sent_event(event_id: int, instance_id: str, input: str):
     return pb.HistoryEvent(
         eventId=event_id,

durabletask/payload/store.py

@@ -21,15 +21,15 @@ class LargePayloadStorageOptions:
 
     Attributes:
         threshold_bytes: Payloads larger than this value (in bytes) will
-            be externalized to the payload store.  Defaults to 900,000
-            (900 KB), matching the .NET SDK default.
+            be externalized to the payload store.  Defaults to 262,144
+            (256 KiB), matching the .NET SDK default.
         max_stored_payload_bytes: Maximum payload size (in bytes) that
             can be stored externally.  Payloads exceeding this limit
             will cause an error.  Defaults to 10,485,760 (10 MB).
         enable_compression: When ``True`` (the default), payloads are
             GZip-compressed before uploading.
     """
-    threshold_bytes: int = 900_000
+    threshold_bytes: int = 262_144
     max_stored_payload_bytes: int = 10 * 1024 * 1024  # 10 MB
     enable_compression: bool = True