Complete permanent payload failures with entity-lock cleanup

CHANGELOG.md

@@ -14,6 +14,9 @@ size before uploading or replacing any protobuf values, including carryover
 events and entity payloads, for both synchronous and asynchronous calls.
 Oversized messages remain unchanged and cause no uploads; their size error
 now takes precedence over hypothetical earlier storage failures.
+- Orchestrations that exceed the configured external payload storage limit now
+fail without retrying and release their entity locks, allowing other callers to
+progress. Transient payload-store failures remain retryable.
 - Orchestrator generators are now closed at the end of each dispatch, including
 while waiting for durable work, releasing resources held by async bridges.
 `finally` blocks and context-manager exits can run before orchestration completion

durabletask-azuremanaged/CHANGELOG.md

@@ -12,6 +12,9 @@ FIXED
 - With the corresponding core `durabletask` SDK fix, oversized messages fail
 large-payload validation before any uploads or payload replacements. Size
 errors now take precedence over hypothetical earlier storage failures.
+- With the corresponding core `durabletask` SDK fix, orchestrations that exceed
+the configured external payload storage limit fail without retrying and release
+their entity locks. Transient payload-store failures remain retryable.
 - With the corresponding core `durabletask` SDK fix, orchestrator generators
 are closed at the end of each dispatch, including while waiting for durable work.
 See [cleanup semantics](../docs/supported-patterns.md#cleanup-in-orchestrator-finally-blocks)

durabletask/internal/exceptions.py

@@ -12,3 +12,7 @@ def __init__(self, error_details: pb.TaskFailureDetails) -> None:
 
 class AbandonOrchestrationError(Exception):
     pass
+
+
+class PayloadSizeExceededError(ValueError):
+    """A raw payload exceeds the configured external storage limit."""

durabletask/internal/orchestration_entity_context.py

@@ -19,6 +19,8 @@ def __init__(self, instance_id: str):
         self.critical_section_id: str | None = None
         self.critical_section_locks: list[EntityInstanceId] = []
         self.available_locks: list[EntityInstanceId] = []
+        self._lock_release_obligations: list[EntityInstanceId] = []
+        self._pending_lock_releases: list[pb.SendEntityMessageAction] = []
 
     @property
     def is_inside_critical_section(self) -> bool:
@@ -61,19 +63,40 @@ def recover_lock_after_call(self, target_instance_id: EntityInstanceId):
         if self.is_inside_critical_section:
             self.available_locks.append(target_instance_id)
 
+    def get_lock_release_messages(self) -> Generator[pb.SendEntityMessageAction, None, None]:
+        yield from self._pending_lock_releases
+        if self.is_inside_critical_section:
+            # Preserve the historical shared-list order, then include held
+            # locks removed from that list by calls which are still pending.
+            release_order = self.critical_section_locks + [
+                entity_id for entity_id in self._lock_release_obligations
+                if entity_id not in self.critical_section_locks
+            ]
+            for entity_id in release_order:
+                yield self._new_lock_release_message(entity_id)
+
+    def _new_lock_release_message(self, entity_id: EntityInstanceId) -> pb.SendEntityMessageAction:
+        return pb.SendEntityMessageAction(entityUnlockSent=pb.EntityUnlockSentEvent(
+            criticalSectionId=self.critical_section_id,
+            targetInstanceId=get_string_value(str(entity_id)),
+            parentInstanceId=get_string_value(self.instance_id)
+        ))
+
     def emit_lock_release_messages(self) -> Generator[pb.SendEntityMessageAction, None, None]:
         if self.is_inside_critical_section:
+            # Normal emission must retain historical action IDs. Missing held
+            # locks are kept only for a later terminal-failure snapshot.
+            self._pending_lock_releases.extend(
+                self._new_lock_release_message(entity_id)
+                for entity_id in self._lock_release_obligations
+                if entity_id not in self.critical_section_locks
+            )
             for entity_id in self.critical_section_locks:
-                unlock_event = pb.SendEntityMessageAction(entityUnlockSent=pb.EntityUnlockSentEvent(
-                    criticalSectionId=self.critical_section_id,
-                    targetInstanceId=get_string_value(str(entity_id)),
-                    parentInstanceId=get_string_value(self.instance_id)
-                ))
-                yield unlock_event
-
-            self.critical_section_locks = []
-            self.available_locks = []
-            self.critical_section_id = None
+                yield self._new_lock_release_message(entity_id)
+        self.critical_section_locks = []
+        self.available_locks = []
+        self._lock_release_obligations = []
+        self.critical_section_id = None
 
     def emit_request_message(self, target: Any, operation_name: str, one_way: bool, operation_id: str,
                              scheduled_time_utc: datetime, input: str | None,
@@ -106,6 +129,7 @@ def emit_acquire_message(
 
         self.critical_section_id = critical_section_id
         self.critical_section_locks = entity_ids_dedup
+        self._lock_release_obligations = list(entity_ids_dedup)
         self.lock_acquisition_pending = True
 
         return request, target

durabletask/payload/helpers.py

@@ -18,6 +18,8 @@
 from google.protobuf import message as proto_message
 from google.protobuf import wrappers_pb2
 
+from durabletask.internal.exceptions import PayloadSizeExceededError
+
 if TYPE_CHECKING:
     from durabletask.payload.store import PayloadStore
 
@@ -150,7 +152,7 @@ def _collect_payloads_to_externalize(
         if len(payload_bytes) <= threshold:
             continue
         if len(payload_bytes) > max_bytes:
-            raise ValueError(
+            raise PayloadSizeExceededError(
                 f"Payload size {len(payload_bytes)} bytes exceeds the maximum "
                 f"allowed size of {max_bytes} bytes."
             )

durabletask/worker.py

@@ -1179,6 +1179,7 @@ def _execute_orchestrator(
         if (req.HasField("orchestrationTraceContext") and req.orchestrationTraceContext.HasField("spanID") and req.orchestrationTraceContext.spanID.value):
             persisted_orch_span_id = req.orchestrationTraceContext.spanID.value
 
+        result: ExecutionResults | None = None
         try:
             executor = _OrchestrationExecutor(
                 self._registry, self._logger,
@@ -1192,30 +1193,6 @@ def _execute_orchestrator(
                     result = executor.execute(
                         instance_id, req.pastEvents, req.newEvents)
 
-            # Determine completion status for span
-            is_complete = False
-            is_failed = False
-            failure_details = None
-            for action in result.actions:
-                if action.HasField("completeOrchestration"):
-                    is_complete = True
-                    orch_status = action.completeOrchestration.orchestrationStatus
-                    if orch_status == pb.ORCHESTRATION_STATUS_FAILED:
-                        is_failed = True
-                        failure_details = action.completeOrchestration.failureDetails
-
-            if is_complete and self._emit_trace_spans:
-                # Orchestration finished — emit a single span covering its lifetime
-                tracing.emit_orchestration_span(
-                    orchestration_name,
-                    instance_id,
-                    start_time_ns,
-                    is_failed,
-                    failure_details=failure_details,
-                    parent_trace_context=parent_trace_ctx,
-                    orchestration_trace_context=result._orchestration_trace_context,  # pyright: ignore[reportPrivateUsage]
-                )
-
             # Include the span ID in the orchestration trace context
             # so it persists across dispatches.
             orch_span_id = None
@@ -1278,10 +1255,54 @@ def _execute_orchestrator(
 
         try:
             # Externalize any large payloads in the response
-            if self._payload_store is not None:
-                payload_helpers.externalize_payloads(
-                    res, self._payload_store, instance_id=instance_id,
+            try:
+                if self._payload_store is not None:
+                    payload_helpers.externalize_payloads(
+                        res, self._payload_store, instance_id=instance_id,
+                    )
+            except pe.PayloadSizeExceededError as ex:
+                self._logger.warning(
+                    f"Orchestration '{instance_id}' exceeded the payload storage limit: {ex}"
                 )
+                # Use the original actions, not a partially externalized response.
+                releases = [
+                    action for action in result.actions
+                    if action.HasField("sendEntityMessage")
+                    and action.sendEntityMessage.HasField("entityUnlockSent")
+                ] if result is not None else []
+                next_id = max((action.id for action in res.actions), default=0) + 1
+                if result is not None:
+                    for release in result.entity_lock_releases:
+                        releases.append(pb.OrchestratorAction(id=next_id, sendEntityMessage=release))
+                        next_id += 1
+                releases.append(ph.new_complete_orchestration_action(
+                    next_id,
+                    pb.ORCHESTRATION_STATUS_FAILED,
+                    None,
+                    pb.TaskFailureDetails(
+                        errorType=ph.get_qualified_name(type(ex)),
+                        errorMessage=str(ex)[:1024],
+                        isNonRetriable=True,
+                    ),
+                ))
+                res.ClearField("actions")
+                res.actions.extend(releases)
+                res.ClearField("customStatus")
+                # Keep this fallback inline: it must not need the failing store.
+
+            if result is not None and self._emit_trace_spans:
+                for action in res.actions:
+                    if action.HasField("completeOrchestration"):
+                        completion = action.completeOrchestration
+                        tracing.emit_orchestration_span(
+                            orchestration_name,
+                            instance_id,
+                            start_time_ns,
+                            completion.orchestrationStatus == pb.ORCHESTRATION_STATUS_FAILED,
+                            failure_details=completion.failureDetails if completion.HasField("failureDetails") else None,
+                            parent_trace_context=parent_trace_ctx,
+                            orchestration_trace_context=result._orchestration_trace_context,  # pyright: ignore[reportPrivateUsage]
+                        )
             stub.CompleteOrchestratorTask(res)
         except Exception as ex:
             self._logger.exception(
@@ -2194,14 +2215,17 @@ class ExecutionResults:
     actions: list[pb.OrchestratorAction]
     encoded_custom_status: str | None
     _orchestration_trace_context: pb.TraceContext | None
+    entity_lock_releases: list[pb.SendEntityMessageAction]
 
     def __init__(
             self, actions: list[pb.OrchestratorAction], encoded_custom_status: str | None,
             orchestration_trace_context: pb.TraceContext | None = None,
+            entity_lock_releases: list[pb.SendEntityMessageAction] | None = None,
     ):
         self.actions = actions
         self.encoded_custom_status = encoded_custom_status
         self._orchestration_trace_context = orchestration_trace_context
+        self.entity_lock_releases = entity_lock_releases if entity_lock_releases is not None else []
 
 
 class _OrchestrationExecutor:
@@ -2347,6 +2371,8 @@ def execute(
             return ExecutionResults(
                 actions=actions, encoded_custom_status=ctx._encoded_custom_status,  # pyright: ignore[reportPrivateUsage]
                 orchestration_trace_context=ctx._orchestration_trace_context,  # pyright: ignore[reportPrivateUsage]
+                # Snapshot before generator teardown can exit a critical section.
+                entity_lock_releases=list(ctx._entity_context.get_lock_release_messages()),  # pyright: ignore[reportPrivateUsage]
             )
         finally:
             # Close after building the response, so the orchestrator's finally

tests/durabletask/test_large_payload.py

@@ -12,6 +12,7 @@
 from google.protobuf import wrappers_pb2
 
 import durabletask.internal.orchestrator_service_pb2 as pb
+from durabletask.internal.exceptions import PayloadSizeExceededError
 from durabletask.payload.helpers import (
     deexternalize_payloads,
     deexternalize_payloads_async,
@@ -152,8 +153,9 @@ def test_exceeds_max_raises_error(self):
             name="MyOrch",
             input=sv(large_data),
         )
-        with pytest.raises(ValueError, match="exceeds the maximum"):
+        with pytest.raises(PayloadSizeExceededError, match="exceeds the maximum") as failure:
             externalize_payloads(req, store, instance_id="test-1")
+        assert isinstance(failure.value, ValueError)
 
     def test_empty_value_not_externalized(self):
         """Empty StringValue should not be externalized."""
@@ -540,6 +542,40 @@ async def test_async_externalize_and_deexternalize(self):
         await deexternalize_payloads_async(req, store)
         assert req.input.value == original
 
+    @pytest.mark.asyncio
+    async def test_async_cap_failure_is_a_value_error(self):
+        store = FakePayloadStore(threshold_bytes=1, max_stored_payload_bytes=10)
+        req = pb.CreateInstanceRequest(input=sv("x" * 11))
+        with pytest.raises(PayloadSizeExceededError) as failure:
+            await externalize_payloads_async(req, store)
+        assert isinstance(failure.value, ValueError)
+        assert not store._blobs
+
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize("use_async", [False, True])
+    async def test_token_larger_than_cap_is_not_reuploaded(self, use_async):
+        store = FakePayloadStore(threshold_bytes=1, max_stored_payload_bytes=10)
+        token = f"{FakePayloadStore.TOKEN_PREFIX}existing-blob"
+        assert len(token.encode("utf-8")) > store.options.max_stored_payload_bytes
+        req = pb.CreateInstanceRequest(input=sv(token))
+        if use_async:
+            await externalize_payloads_async(req, store)
+        else:
+            externalize_payloads(req, store)
+        assert req.input.value == token
+        assert not store._blobs
+
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize("error", [OSError("unavailable"), ValueError("store error"), asyncio.CancelledError()])
+    async def test_async_store_error_propagates_unchanged(self, error):
+        store = FakePayloadStore(threshold_bytes=1)
+        store.upload_async = AsyncMock(side_effect=error)
+        req = pb.CreateInstanceRequest(input=sv("large enough"))
+        with pytest.raises(type(error)) as failure:
+            await externalize_payloads_async(req, store)
+        assert failure.value is error
+        assert req.input.value == "large enough"
+
 
 # ------------------------------------------------------------------
 # Tests: worker capability flag