diff --git a/CHANGELOG.md b/CHANGELOG.md index 2984747f..ed81716d 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/durabletask-azuremanaged/CHANGELOG.md b/durabletask-azuremanaged/CHANGELOG.md index 7b151143..cba95c64 100644 --- a/durabletask-azuremanaged/CHANGELOG.md +++ b/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) diff --git a/durabletask/internal/exceptions.py b/durabletask/internal/exceptions.py index 276cbe60..a04a304f 100644 --- a/durabletask/internal/exceptions.py +++ b/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.""" diff --git a/durabletask/internal/orchestration_entity_context.py b/durabletask/internal/orchestration_entity_context.py index e47d2ab5..3366590d 100644 --- a/durabletask/internal/orchestration_entity_context.py +++ b/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 diff --git a/durabletask/payload/helpers.py b/durabletask/payload/helpers.py index dbcd600a..4fe52ac1 100644 --- a/durabletask/payload/helpers.py +++ b/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." ) diff --git a/durabletask/worker.py b/durabletask/worker.py index 5b61a411..46229013 100644 --- a/durabletask/worker.py +++ b/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 diff --git a/tests/durabletask/test_large_payload.py b/tests/durabletask/test_large_payload.py index ed0214f9..ace3af61 100644 --- a/tests/durabletask/test_large_payload.py +++ b/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 diff --git a/tests/durabletask/test_orchestrator_payload_failure.py b/tests/durabletask/test_orchestrator_payload_failure.py new file mode 100644 index 00000000..1412e4c3 --- /dev/null +++ b/tests/durabletask/test_orchestrator_payload_failure.py @@ -0,0 +1,474 @@ +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT License. + +"""Permanent payload failures must complete without stranding entity locks.""" + +import asyncio +import json +import logging +from threading import Event +from unittest.mock import MagicMock + +import pytest + +from durabletask import client, entities, task, worker +from durabletask.internal import helpers as ph +from durabletask.internal import orchestrator_service_pb2 as pb +from durabletask.internal.orchestration_entity_context import OrchestrationEntityContext +from durabletask.serialization import JsonDataConverter +from durabletask.testing import create_test_backend +from tests.durabletask._port_utils import find_free_port +from tests.durabletask.test_large_payload import FakePayloadStore + + +ENTITY_IDS = [ + entities.EntityInstanceId("counter", "a"), + entities.EntityInstanceId("counter", "b"), +] +LARGE_PAYLOAD = "x" * 1024 + + +def _fail_payload(ctx: task.OrchestrationContext, mode: str): + if mode == "status": + ctx.set_custom_status(LARGE_PAYLOAD) + elif mode == "call": + yield ctx.call_entity(ENTITY_IDS[0], "add", LARGE_PAYLOAD) + elif mode == "continue": + ctx.continue_as_new(LARGE_PAYLOAD) + return LARGE_PAYLOAD if mode == "result" else "done" + + +def _request(name: str) -> pb.OrchestratorRequest: + return pb.OrchestratorRequest( + instanceId="owner", + newEvents=[ + ph.new_orchestrator_started_event(), + ph.new_execution_started_event(name, "owner"), + ], + ) + + +@pytest.mark.parametrize("call_failed", [False, True]) +def test_legacy_multi_lock_release_history_replays(call_failed): + def orchestrator(ctx, _): + with (yield ctx.lock_entities(ENTITY_IDS)): + try: + yield ctx.call_entity(ENTITY_IDS[0], "get") + except task.TaskFailedError: + # Handle the expected failure so replay can verify lock releases. + pass + yield ctx.wait_for_external_event("next") + return "done" + + registry = worker._Registry() + name = registry.add_orchestrator(orchestrator) + executor = worker._OrchestrationExecutor(registry, logging.getLogger("test"), JsonDataConverter()) + started = list(_request(name).newEvents) + first = executor.execute("owner", [], started) + section = first.actions[0].sendEntityMessage.entityLockRequested.criticalSectionId + granted = [ + ph.new_event_sent_event(1, str(ENTITY_IDS[0]), json.dumps({"id": section})), + ph.new_event_raised_event(section, None), + ] + second = executor.execute("owner", started, granted) + request_id = second.actions[0].sendEntityMessage.entityOperationCalled.requestId + response = {"result": "42"} + if call_failed: + response["exceptionType"] = "ValueError" + called = [ + ph.new_event_sent_event(2, str(ENTITY_IDS[0]), json.dumps({"id": request_id})), + ph.new_event_raised_event(request_id, json.dumps(response)), + ] + third = executor.execute("owner", started + granted, called) + # Before acquired/available lock tracking was separated, completing a call + # moved that entity to the end of the shared list used for releases. + legacy_releases = [(3, str(ENTITY_IDS[1])), (4, str(ENTITY_IDS[0]))] + assert [ + (action.id, action.sendEntityMessage.entityUnlockSent.targetInstanceId.value) + for action in third.actions + ] == legacy_releases + released = [ph.new_event_sent_event(action_id, target, None) for action_id, target in legacy_releases] + history = started + granted + called + released + replay = executor.execute("owner", history, [ph.new_orchestrator_started_event()]) + assert not replay.actions + assert not replay.entity_lock_releases + completed = executor.execute("owner", history, [ph.new_event_raised_event("next", None)]) + assert len(completed.actions) == 1 + assert completed.actions[0].id == 5 + assert completed.actions[0].completeOrchestration.orchestrationStatus == pb.ORCHESTRATION_STATUS_COMPLETED + assert completed.actions[0].completeOrchestration.result.value == '"done"' + + +def _timeout_history(executor, name): + started = list(_request(name).newEvents) + first = executor.execute("owner", [], started) + section = first.actions[0].sendEntityMessage.entityLockRequested.criticalSectionId + granted = [ + ph.new_event_sent_event(1, str(ENTITY_IDS[0]), json.dumps({"id": section})), + ph.new_event_raised_event(section, None), + ] + second = executor.execute("owner", started, granted) + call = second.actions[0].sendEntityMessage.entityOperationCalled + pending = ph.new_event_sent_event(2, str(ENTITY_IDS[0]), json.dumps({"id": call.requestId})) + return started + granted + [pending], section + + +def test_legacy_pending_call_timeout_history_replays_through_activity(): + def orchestrator(ctx, _): + with (yield ctx.lock_entities(ENTITY_IDS)): + call = ctx.call_entity(ENTITY_IDS[0], "get") + timeout = ctx.wait_for_external_event("timeout") + yield task.when_any([call, timeout]) + return (yield ctx.call_activity("after_timeout")) + + registry = worker._Registry() + name = registry.add_orchestrator(orchestrator) + executor = worker._OrchestrationExecutor(registry, logging.getLogger("test"), JsonDataConverter()) + history, _ = _timeout_history(executor, name) + timed_out = ph.new_event_raised_event("timeout", None) + result = executor.execute("owner", history, [timed_out]) + assert [action.id for action in result.actions] == [3, 4] + assert result.actions[0].sendEntityMessage.entityUnlockSent.targetInstanceId.value == str(ENTITY_IDS[1]) + assert result.actions[1].scheduleTask.name == "after_timeout" + assert [release.entityUnlockSent.targetInstanceId.value for release in result.entity_lock_releases] == [ + str(ENTITY_IDS[0]), + ] + + # Literal history from the old emitter: only b was released, so the next + # activity kept ID 4 even though the call to a was still outstanding. + history += [ + timed_out, + ph.new_event_sent_event(3, str(ENTITY_IDS[1]), None), + ph.new_task_scheduled_event(4, "after_timeout"), + ] + assert not executor.execute("owner", history, [ph.new_orchestrator_started_event()]).actions + completed = executor.execute("owner", history, [ph.new_task_completed_event(4, '"done"')]) + assert len(completed.actions) == 1 + assert completed.actions[0].id == 5 + assert completed.actions[0].completeOrchestration.orchestrationStatus == pb.ORCHESTRATION_STATUS_COMPLETED + assert completed.actions[0].completeOrchestration.result.value == '"done"' + + +@pytest.mark.parametrize("mode", ["result", "status", "activity"]) +@pytest.mark.parametrize("replay_exit", [False, True]) +def test_payload_failure_after_normal_pending_call_exit_releases_all_locks(mode, replay_exit): + def orchestrator(ctx, _): + with (yield ctx.lock_entities(ENTITY_IDS)): + call = ctx.call_entity(ENTITY_IDS[0], "get") + yield task.when_any([call, ctx.wait_for_external_event("timeout")]) + if replay_exit: + yield ctx.wait_for_external_event("fail") + if mode == "result": + return LARGE_PAYLOAD + if mode == "status": + ctx.set_custom_status(LARGE_PAYLOAD) + yield ctx.call_activity("after_timeout", input=LARGE_PAYLOAD if mode == "activity" else None) + return None + + store = FakePayloadStore(threshold_bytes=100, max_stored_payload_bytes=200) + stub = MagicMock() + with worker.TaskHubGrpcWorker(payload_store=store) as w: + w.add_orchestrator(orchestrator) + executor = worker._OrchestrationExecutor(w._registry, logging.getLogger("test"), JsonDataConverter()) + history, section = _timeout_history(executor, "orchestrator") + events = [ph.new_event_raised_event("timeout", None)] + if replay_exit: + exited = executor.execute("owner", history, events) + assert len(exited.actions) == 1 + assert exited.actions[0].id == 3 + history += events + [ph.new_event_sent_event(3, str(ENTITY_IDS[1]), None)] + events = [ph.new_event_raised_event("fail", None)] + w._execute_orchestrator(pb.OrchestratorRequest( + instanceId="owner", pastEvents=history, + newEvents=events, + ), stub, "failure-token") + + response = stub.CompleteOrchestratorTask.call_args.args[0] + assert response.completionToken == "failure-token" + assert len(response.actions) == (2 if replay_exit else 3) + if not replay_exit: + assert response.actions[0].id == 3 + releases = [action.sendEntityMessage.entityUnlockSent for action in response.actions[:-1]] + expected_targets = [str(ENTITY_IDS[0])] if replay_exit else [str(ENTITY_IDS[1]), str(ENTITY_IDS[0])] + assert [release.targetInstanceId.value for release in releases] == expected_targets + assert all(release.parentInstanceId.value == "owner" for release in releases) + assert all(release.criticalSectionId == section for release in releases) + completion = response.actions[-1].completeOrchestration + assert completion.orchestrationStatus == pb.ORCHESTRATION_STATUS_FAILED + assert completion.failureDetails.isNonRetriable + assert not response.HasField("customStatus") + + +def test_release_snapshot_retains_pending_locks_after_completed_calls(): + entity_ctx = OrchestrationEntityContext("owner") + entity_ctx.emit_acquire_message("section", ENTITY_IDS) + entity_ctx.complete_acquire("section") + assert entity_ctx.validate_operation_transition(ENTITY_IDS[0], False) == (True, "") + entity_ctx.recover_lock_after_call(ENTITY_IDS[0]) + assert entity_ctx.validate_operation_transition(ENTITY_IDS[1], False) == (True, "") + releases = list(entity_ctx.get_lock_release_messages()) + assert [release.entityUnlockSent.targetInstanceId.value for release in releases] == [ + str(ENTITY_IDS[0]), str(ENTITY_IDS[1]), + ] + assert all(release.entityUnlockSent.parentInstanceId.value == "owner" for release in releases) + assert all(release.entityUnlockSent.criticalSectionId == "section" for release in releases) + assert entity_ctx.available_locks == [ENTITY_IDS[0]] + assert entity_ctx.critical_section_locks == [ENTITY_IDS[0]] + assert entity_ctx.is_inside_critical_section + + +@pytest.mark.parametrize("legacy_protocol", [False, True]) +@pytest.mark.parametrize("overlapping_call", [False, True]) +def test_pre_lock_call_can_complete_while_acquisition_is_pending(legacy_protocol, overlapping_call): + outside_entity = ENTITY_IDS[0] if overlapping_call else entities.EntityInstanceId("counter", "outside") + + def orchestrator(ctx, _): + pending = ctx.call_entity(outside_entity, "get") + with (yield ctx.lock_entities(ENTITY_IDS)): + result = yield pending + yield ctx.wait_for_external_event("next") + return result + + registry = worker._Registry() + name = registry.add_orchestrator(orchestrator) + executor = worker._OrchestrationExecutor(registry, logging.getLogger("test"), JsonDataConverter()) + started = list(_request(name).newEvents) + first = executor.execute("owner", [], started) + call = first.actions[0].sendEntityMessage.entityOperationCalled + lock = first.actions[1].sendEntityMessage.entityLockRequested + if legacy_protocol: + called = [ + ph.new_event_sent_event(1, str(outside_entity), json.dumps({"id": call.requestId})), + ph.new_event_sent_event(2, str(ENTITY_IDS[0]), json.dumps({"id": lock.criticalSectionId})), + ph.new_event_raised_event(call.requestId, json.dumps({"result": "42"})), + ] + granted = ph.new_event_raised_event(lock.criticalSectionId, None) + else: + called = [ + pb.HistoryEvent(eventId=1, entityOperationCalled=call), + pb.HistoryEvent(eventId=2, entityLockRequested=lock), + pb.HistoryEvent(entityOperationCompleted=pb.EntityOperationCompletedEvent( + requestId=call.requestId, output=ph.get_string_value("42"))), + ] + granted = pb.HistoryEvent(entityLockGranted=pb.EntityLockGrantedEvent( + criticalSectionId=lock.criticalSectionId)) + waiting = executor.execute("owner", started, called) + assert not waiting.actions + unlocked = executor.execute("owner", started + called, [granted]) + assert [action.id for action in unlocked.actions] == [3, 4] + assert [ + action.sendEntityMessage.entityUnlockSent.targetInstanceId.value for action in unlocked.actions + ] == [str(eid) for eid in ENTITY_IDS] + if legacy_protocol: + released = [ph.new_event_sent_event(action.id, str(eid), None) + for action, eid in zip(unlocked.actions, ENTITY_IDS)] + else: + released = [pb.HistoryEvent(eventId=action.id, entityUnlockSent=action.sendEntityMessage.entityUnlockSent) + for action in unlocked.actions] + history = started + called + [granted] + released + assert not executor.execute("owner", history, [ph.new_orchestrator_started_event()]).actions + completed = executor.execute("owner", history, [ph.new_event_raised_event("next", None)]) + assert len(completed.actions) == 1 + assert completed.actions[-1].id == 5 + assert completed.actions[-1].completeOrchestration.orchestrationStatus == pb.ORCHESTRATION_STATUS_COMPLETED + assert completed.actions[-1].completeOrchestration.result.value == "42" + + +@pytest.mark.parametrize("mode", ["result", "status", "call", "continue"]) +@pytest.mark.parametrize("completed_call", [False, True]) +def test_permanent_payload_failure_releases_acquired_locks(mode, completed_call): + def orchestrator(ctx, _): + with (yield ctx.lock_entities(list(reversed(ENTITY_IDS)))): + if completed_call: + yield ctx.call_entity(ENTITY_IDS[0], "get") + return (yield from _fail_payload(ctx, mode)) + + store = FakePayloadStore(threshold_bytes=100, max_stored_payload_bytes=200) + provider = MagicMock() + stub = MagicMock() + with worker.TaskHubGrpcWorker(payload_store=store, exception_properties_provider=provider) as w: + w.add_orchestrator(orchestrator) + first = _request("orchestrator") + w._execute_orchestrator(first, stub, "first-token") + first_response = stub.CompleteOrchestratorTask.call_args.args[0] + lock_action = first_response.actions[0] + lock_request = lock_action.sendEntityMessage.entityLockRequested + stub.reset_mock() + + req = pb.OrchestratorRequest( + instanceId="owner", + pastEvents=[ + *first.newEvents, + pb.HistoryEvent(eventId=lock_action.id, entityLockRequested=lock_request), + ], + newEvents=[ + ph.new_orchestrator_started_event(), + pb.HistoryEvent(entityLockGranted=pb.EntityLockGrantedEvent( + criticalSectionId=lock_request.criticalSectionId)), + ], + ) + if completed_call: + w._execute_orchestrator(req, stub, "call-token") + call_action = stub.CompleteOrchestratorTask.call_args.args[0].actions[0] + call_message = call_action.sendEntityMessage.entityOperationCalled + req = pb.OrchestratorRequest( + instanceId="owner", + pastEvents=[ + *req.pastEvents, + *req.newEvents, + pb.HistoryEvent(eventId=call_action.id, entityOperationCalled=call_message), + ], + newEvents=[ + ph.new_orchestrator_started_event(), + pb.HistoryEvent(entityOperationCompleted=pb.EntityOperationCompletedEvent( + requestId=call_message.requestId, output=ph.get_string_value("42"))), + ], + ) + stub.reset_mock() + w._execute_orchestrator(req, stub, "failure-token") + + stub.CompleteOrchestratorTask.assert_called_once() + stub.AbandonTaskOrchestratorWorkItem.assert_not_called() + response = stub.CompleteOrchestratorTask.call_args.args[0] + assert response.instanceId == "owner" + assert response.completionToken == "failure-token" + assert not response.HasField("customStatus") + assert response.ByteSize() < 1024 + assert len(response.actions) == 3 + releases = [action.sendEntityMessage.entityUnlockSent for action in response.actions[:-1]] + release_order = list(reversed(ENTITY_IDS)) if completed_call or mode == "call" else ENTITY_IDS + assert [release.targetInstanceId.value for release in releases] == [str(eid) for eid in release_order] + assert all(release.parentInstanceId.value == "owner" for release in releases) + assert all(release.criticalSectionId == lock_request.criticalSectionId for release in releases) + assert [action.id for action in response.actions] == sorted({action.id for action in response.actions}) + completion = response.actions[-1].completeOrchestration + assert completion.orchestrationStatus == pb.ORCHESTRATION_STATUS_FAILED + assert not completion.HasField("result") + failure = completion.failureDetails + assert failure.isNonRetriable + assert failure.errorType == "durabletask.internal.exceptions.PayloadSizeExceededError" + assert "exceeds the maximum" in failure.errorMessage + assert not failure.HasField("stackTrace") + assert not failure.HasField("innerFailure") + assert not failure.properties + provider.get_exception_properties.assert_not_called() + + +@pytest.mark.parametrize("error", [OSError("unavailable"), ValueError("store error"), asyncio.CancelledError()]) +def test_storage_failure_and_cancellation_do_not_become_permanent(error): + def orchestrator(ctx, _): + return LARGE_PAYLOAD + + store = FakePayloadStore(threshold_bytes=100) + original_upload = store.upload + store.upload = MagicMock(side_effect=error) + stub = MagicMock() + with worker.TaskHubGrpcWorker(payload_store=store) as w: + w.add_orchestrator(orchestrator) + req = _request("orchestrator") + if isinstance(error, asyncio.CancelledError): + with pytest.raises(asyncio.CancelledError): + w._execute_orchestrator(req, stub, "token") + else: + w._execute_orchestrator(req, stub, "token") + stub.CompleteOrchestratorTask.assert_not_called() + + store.upload = original_upload + w._execute_orchestrator(req, stub, "retry-token") + + completion = stub.CompleteOrchestratorTask.call_args.args[0].actions[-1].completeOrchestration + assert completion.orchestrationStatus == pb.ORCHESTRATION_STATUS_COMPLETED + assert not completion.HasField("failureDetails") + + +def test_failure_preserves_emitted_releases_before_outstanding_releases(monkeypatch): + entity_ctx = OrchestrationEntityContext("owner") + entity_ctx.emit_acquire_message("old-section", ENTITY_IDS) + emitted = [ + pb.OrchestratorAction(id=index, sendEntityMessage=release) + for index, release in enumerate(entity_ctx.emit_lock_release_messages(), start=2) + ] + entity_ctx.emit_acquire_message("new-section", ENTITY_IDS) + entity_ctx.complete_acquire("new-section") + result = worker.ExecutionResults( + actions=[ + *emitted, + ph.new_call_entity_action(4, "owner", ENTITY_IDS[0], "add", LARGE_PAYLOAD, "request"), + ], + encoded_custom_status="status to discard", + entity_lock_releases=list(entity_ctx.get_lock_release_messages()), + ) + monkeypatch.setattr(worker._OrchestrationExecutor, "execute", lambda *args: result) + store = FakePayloadStore(threshold_bytes=1, max_stored_payload_bytes=200) + stub = MagicMock() + with worker.TaskHubGrpcWorker(payload_store=store) as w: + w._execute_orchestrator(_request("unused"), stub, "token") + + response = stub.CompleteOrchestratorTask.call_args.args[0] + assert response.actions[:2] == emitted + releases = [action.sendEntityMessage.entityUnlockSent for action in response.actions[:-1]] + assert [(release.criticalSectionId, release.targetInstanceId.value) for release in releases] == [ + (section, str(eid)) for section in ("old-section", "new-section") for eid in ENTITY_IDS + ] + assert all(release.parentInstanceId.value == "owner" for release in releases) + assert [action.id for action in response.actions] == [2, 3, 5, 6, 7] + assert response.actions[-1].completeOrchestration.orchestrationStatus == pb.ORCHESTRATION_STATUS_FAILED + assert not response.HasField("customStatus") + + +@pytest.mark.parametrize("mode", ["result", "status", "call"]) +def test_next_lock_caller_progresses_after_permanent_payload_failure(monkeypatch, mode): + locked = Event() + second_queued = Event() + + def failing(ctx, _): + with (yield ctx.lock_entities(ENTITY_IDS)): + locked.set() + yield ctx.wait_for_external_event("fail") + return (yield from _fail_payload(ctx, mode)) + + def next_caller(ctx, _): + with (yield ctx.lock_entities(ENTITY_IDS)): + return (yield ctx.call_entity(ENTITY_IDS[0], "get")) + + def counter(ctx: entities.EntityContext, _): + return 42 + + port = find_free_port() + host = f"localhost:{port}" + backend = create_test_backend(port=port) + original_grant = backend._try_grant_lock + + def observe_grant(pending): + granted = original_grant(pending) + if pending.parent_instance_id == "second" and not granted: + second_queued.set() + return granted + + monkeypatch.setattr(backend, "_try_grant_lock", observe_grant) + store = FakePayloadStore(threshold_bytes=100, max_stored_payload_bytes=200) + try: + with worker.TaskHubGrpcWorker(host_address=host, payload_store=store) as w: + w.add_orchestrator(failing) + w.add_orchestrator(next_caller) + w.add_entity(counter) + w.start() + with client.TaskHubGrpcClient(host_address=host) as c: + c.schedule_new_orchestration(failing, instance_id="first") + assert locked.wait(10), "First orchestration did not acquire both locks" + c.schedule_new_orchestration(next_caller, instance_id="second") + assert second_queued.wait(10), "Second caller was not blocked by the first" + c.raise_orchestration_event("first", "fail") + failed = c.wait_for_orchestration_completion("first", timeout=10) + completed = c.wait_for_orchestration_completion("second", timeout=10) + + assert failed is not None + assert failed.runtime_status == client.OrchestrationStatus.FAILED + assert failed.failure_details is not None + assert failed.failure_details.error_type == "durabletask.internal.exceptions.PayloadSizeExceededError" + assert completed is not None + assert completed.runtime_status == client.OrchestrationStatus.COMPLETED + assert completed.serialized_output == "42" + finally: + backend.stop() + backend.reset()