Skip to content

Commit 21aed7d

Browse files
committed
fix: preserve historical entity release order during replay
Retain legacy shared-list ordering while tracking pending lock release obligations independently. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 064f1284-f637-4ba9-8406-8c094ce9f845
1 parent e4cf01e commit 21aed7d

2 files changed

Lines changed: 163 additions & 4 deletions

File tree

‎durabletask/internal/orchestration_entity_context.py‎

Lines changed: 11 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,7 @@ def __init__(self, instance_id: str):
1919
self.critical_section_id: str | None = None
2020
self.critical_section_locks: list[EntityInstanceId] = []
2121
self.available_locks: list[EntityInstanceId] = []
22+
self._lock_release_obligations: list[EntityInstanceId] = []
2223

2324
@property
2425
def is_inside_critical_section(self) -> bool:
@@ -63,7 +64,13 @@ def recover_lock_after_call(self, target_instance_id: EntityInstanceId):
6364

6465
def get_lock_release_messages(self) -> Generator[pb.SendEntityMessageAction, None, None]:
6566
if self.is_inside_critical_section:
66-
for entity_id in self.critical_section_locks:
67+
# Preserve the historical shared-list order, then include held
68+
# locks removed from that list by calls which are still pending.
69+
release_order = self.critical_section_locks + [
70+
entity_id for entity_id in self._lock_release_obligations
71+
if entity_id not in self.critical_section_locks
72+
]
73+
for entity_id in release_order:
6774
unlock_event = pb.SendEntityMessageAction(entityUnlockSent=pb.EntityUnlockSentEvent(
6875
criticalSectionId=self.critical_section_id,
6976
targetInstanceId=get_string_value(str(entity_id)),
@@ -75,6 +82,7 @@ def emit_lock_release_messages(self) -> Generator[pb.SendEntityMessageAction, No
7582
yield from self.get_lock_release_messages()
7683
self.critical_section_locks = []
7784
self.available_locks = []
85+
self._lock_release_obligations = []
7886
self.critical_section_id = None
7987

8088
def emit_request_message(self, target: Any, operation_name: str, one_way: bool, operation_id: str,
@@ -108,14 +116,15 @@ def emit_acquire_message(
108116

109117
self.critical_section_id = critical_section_id
110118
self.critical_section_locks = entity_ids_dedup
119+
self._lock_release_obligations = list(entity_ids_dedup)
111120
self.lock_acquisition_pending = True
112121

113122
return request, target
114123

115124
def complete_acquire(self, critical_section_id: str) -> None:
116125
if self.critical_section_id != critical_section_id:
117126
raise RuntimeError(f"Unexpected lock acquire for critical section ID '{critical_section_id}' (expected '{self.critical_section_id}')")
118-
self.available_locks = list(self.critical_section_locks)
127+
self.available_locks = self.critical_section_locks
119128
self.lock_acquisition_pending = False
120129

121130
def adjust_outgoing_message(self, instance_id: str, request_message: Any, capped_time: datetime) -> str:

‎tests/durabletask/test_orchestrator_payload_failure.py‎

Lines changed: 152 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -4,6 +4,8 @@
44
"""Permanent payload failures must complete without stranding entity locks."""
55

66
import asyncio
7+
import json
8+
import logging
79
from threading import Event
810
from unittest.mock import MagicMock
911

@@ -13,6 +15,7 @@
1315
from durabletask.internal import helpers as ph
1416
from durabletask.internal import orchestrator_service_pb2 as pb
1517
from durabletask.internal.orchestration_entity_context import OrchestrationEntityContext
18+
from durabletask.serialization import JsonDataConverter
1619
from durabletask.testing import create_test_backend
1720
from tests.durabletask._port_utils import find_free_port
1821
from tests.durabletask.test_large_payload import FakePayloadStore
@@ -45,10 +48,138 @@ def _request(name: str) -> pb.OrchestratorRequest:
4548
)
4649

4750

51+
@pytest.mark.parametrize("call_failed", [False, True])
52+
def test_legacy_multi_lock_release_history_replays(call_failed):
53+
def orchestrator(ctx, _):
54+
with (yield ctx.lock_entities(ENTITY_IDS)):
55+
try:
56+
yield ctx.call_entity(ENTITY_IDS[0], "get")
57+
except task.TaskFailedError:
58+
pass
59+
yield ctx.wait_for_external_event("next")
60+
return "done"
61+
62+
registry = worker._Registry()
63+
name = registry.add_orchestrator(orchestrator)
64+
executor = worker._OrchestrationExecutor(registry, logging.getLogger("test"), JsonDataConverter())
65+
started = list(_request(name).newEvents)
66+
first = executor.execute("owner", [], started)
67+
section = first.actions[0].sendEntityMessage.entityLockRequested.criticalSectionId
68+
granted = [
69+
ph.new_event_sent_event(1, str(ENTITY_IDS[0]), json.dumps({"id": section})),
70+
ph.new_event_raised_event(section, None),
71+
]
72+
second = executor.execute("owner", started, granted)
73+
request_id = second.actions[0].sendEntityMessage.entityOperationCalled.requestId
74+
response = {"result": "42"}
75+
if call_failed:
76+
response["exceptionType"] = "ValueError"
77+
called = [
78+
ph.new_event_sent_event(2, str(ENTITY_IDS[0]), json.dumps({"id": request_id})),
79+
ph.new_event_raised_event(request_id, json.dumps(response)),
80+
]
81+
third = executor.execute("owner", started + granted, called)
82+
# Before acquired/available lock tracking was separated, completing a call
83+
# moved that entity to the end of the shared list used for releases.
84+
legacy_releases = [(3, str(ENTITY_IDS[1])), (4, str(ENTITY_IDS[0]))]
85+
assert [
86+
(action.id, action.sendEntityMessage.entityUnlockSent.targetInstanceId.value)
87+
for action in third.actions
88+
] == legacy_releases
89+
released = [ph.new_event_sent_event(action_id, target, None) for action_id, target in legacy_releases]
90+
history = started + granted + called + released
91+
replay = executor.execute("owner", history, [ph.new_orchestrator_started_event()])
92+
assert not replay.actions
93+
assert not replay.entity_lock_releases
94+
completed = executor.execute("owner", history, [ph.new_event_raised_event("next", None)])
95+
assert len(completed.actions) == 1
96+
assert completed.actions[0].id == 5
97+
assert completed.actions[0].completeOrchestration.orchestrationStatus == pb.ORCHESTRATION_STATUS_COMPLETED
98+
assert completed.actions[0].completeOrchestration.result.value == '"done"'
99+
100+
101+
def test_release_snapshot_retains_pending_locks_after_completed_calls():
102+
entity_ctx = OrchestrationEntityContext("owner")
103+
entity_ctx.emit_acquire_message("section", ENTITY_IDS)
104+
entity_ctx.complete_acquire("section")
105+
assert entity_ctx.validate_operation_transition(ENTITY_IDS[0], False) == (True, "")
106+
entity_ctx.recover_lock_after_call(ENTITY_IDS[0])
107+
assert entity_ctx.validate_operation_transition(ENTITY_IDS[1], False) == (True, "")
108+
releases = list(entity_ctx.get_lock_release_messages())
109+
assert [release.entityUnlockSent.targetInstanceId.value for release in releases] == [
110+
str(ENTITY_IDS[0]), str(ENTITY_IDS[1]),
111+
]
112+
assert all(release.entityUnlockSent.parentInstanceId.value == "owner" for release in releases)
113+
assert all(release.entityUnlockSent.criticalSectionId == "section" for release in releases)
114+
assert entity_ctx.available_locks == [ENTITY_IDS[0]]
115+
assert entity_ctx.critical_section_locks == [ENTITY_IDS[0]]
116+
assert entity_ctx.is_inside_critical_section
117+
118+
119+
@pytest.mark.parametrize("legacy_protocol", [False, True])
120+
@pytest.mark.parametrize("overlapping_call", [False, True])
121+
def test_pre_lock_call_can_complete_while_acquisition_is_pending(legacy_protocol, overlapping_call):
122+
outside_entity = ENTITY_IDS[0] if overlapping_call else entities.EntityInstanceId("counter", "outside")
123+
124+
def orchestrator(ctx, _):
125+
pending = ctx.call_entity(outside_entity, "get")
126+
with (yield ctx.lock_entities(ENTITY_IDS)):
127+
result = yield pending
128+
yield ctx.wait_for_external_event("next")
129+
return result
130+
131+
registry = worker._Registry()
132+
name = registry.add_orchestrator(orchestrator)
133+
executor = worker._OrchestrationExecutor(registry, logging.getLogger("test"), JsonDataConverter())
134+
started = list(_request(name).newEvents)
135+
first = executor.execute("owner", [], started)
136+
call = first.actions[0].sendEntityMessage.entityOperationCalled
137+
lock = first.actions[1].sendEntityMessage.entityLockRequested
138+
if legacy_protocol:
139+
called = [
140+
ph.new_event_sent_event(1, str(outside_entity), json.dumps({"id": call.requestId})),
141+
ph.new_event_sent_event(2, str(ENTITY_IDS[0]), json.dumps({"id": lock.criticalSectionId})),
142+
ph.new_event_raised_event(call.requestId, json.dumps({"result": "42"})),
143+
]
144+
granted = ph.new_event_raised_event(lock.criticalSectionId, None)
145+
else:
146+
called = [
147+
pb.HistoryEvent(eventId=1, entityOperationCalled=call),
148+
pb.HistoryEvent(eventId=2, entityLockRequested=lock),
149+
pb.HistoryEvent(entityOperationCompleted=pb.EntityOperationCompletedEvent(
150+
requestId=call.requestId, output=ph.get_string_value("42"))),
151+
]
152+
granted = pb.HistoryEvent(entityLockGranted=pb.EntityLockGrantedEvent(
153+
criticalSectionId=lock.criticalSectionId))
154+
waiting = executor.execute("owner", started, called)
155+
assert not waiting.actions
156+
unlocked = executor.execute("owner", started + called, [granted])
157+
assert [action.id for action in unlocked.actions] == [3, 4]
158+
assert [
159+
action.sendEntityMessage.entityUnlockSent.targetInstanceId.value for action in unlocked.actions
160+
] == [str(eid) for eid in ENTITY_IDS]
161+
if legacy_protocol:
162+
released = [ph.new_event_sent_event(action.id, str(eid), None)
163+
for action, eid in zip(unlocked.actions, ENTITY_IDS)]
164+
else:
165+
released = [pb.HistoryEvent(eventId=action.id, entityUnlockSent=action.sendEntityMessage.entityUnlockSent)
166+
for action in unlocked.actions]
167+
history = started + called + [granted] + released
168+
assert not executor.execute("owner", history, [ph.new_orchestrator_started_event()]).actions
169+
completed = executor.execute("owner", history, [ph.new_event_raised_event("next", None)])
170+
assert len(completed.actions) == 1
171+
assert completed.actions[-1].id == 5
172+
assert completed.actions[-1].completeOrchestration.orchestrationStatus == pb.ORCHESTRATION_STATUS_COMPLETED
173+
assert completed.actions[-1].completeOrchestration.result.value == "42"
174+
175+
48176
@pytest.mark.parametrize("mode", ["result", "status", "call", "continue"])
49-
def test_permanent_payload_failure_releases_acquired_locks(mode):
177+
@pytest.mark.parametrize("completed_call", [False, True])
178+
def test_permanent_payload_failure_releases_acquired_locks(mode, completed_call):
50179
def orchestrator(ctx, _):
51180
with (yield ctx.lock_entities(list(reversed(ENTITY_IDS)))):
181+
if completed_call:
182+
yield ctx.call_entity(ENTITY_IDS[0], "get")
52183
return (yield from _fail_payload(ctx, mode))
53184

54185
store = FakePayloadStore(threshold_bytes=100, max_stored_payload_bytes=200)
@@ -75,6 +206,24 @@ def orchestrator(ctx, _):
75206
criticalSectionId=lock_request.criticalSectionId)),
76207
],
77208
)
209+
if completed_call:
210+
w._execute_orchestrator(req, stub, "call-token")
211+
call_action = stub.CompleteOrchestratorTask.call_args.args[0].actions[0]
212+
call_message = call_action.sendEntityMessage.entityOperationCalled
213+
req = pb.OrchestratorRequest(
214+
instanceId="owner",
215+
pastEvents=[
216+
*req.pastEvents,
217+
*req.newEvents,
218+
pb.HistoryEvent(eventId=call_action.id, entityOperationCalled=call_message),
219+
],
220+
newEvents=[
221+
ph.new_orchestrator_started_event(),
222+
pb.HistoryEvent(entityOperationCompleted=pb.EntityOperationCompletedEvent(
223+
requestId=call_message.requestId, output=ph.get_string_value("42"))),
224+
],
225+
)
226+
stub.reset_mock()
78227
w._execute_orchestrator(req, stub, "failure-token")
79228

80229
stub.CompleteOrchestratorTask.assert_called_once()
@@ -86,7 +235,8 @@ def orchestrator(ctx, _):
86235
assert response.ByteSize() < 1024
87236
assert len(response.actions) == 3
88237
releases = [action.sendEntityMessage.entityUnlockSent for action in response.actions[:-1]]
89-
assert [release.targetInstanceId.value for release in releases] == [str(eid) for eid in ENTITY_IDS]
238+
release_order = list(reversed(ENTITY_IDS)) if completed_call or mode == "call" else ENTITY_IDS
239+
assert [release.targetInstanceId.value for release in releases] == [str(eid) for eid in release_order]
90240
assert all(release.parentInstanceId.value == "owner" for release in releases)
91241
assert all(release.criticalSectionId == lock_request.criticalSectionId for release in releases)
92242
assert [action.id for action in response.actions] == sorted({action.id for action in response.actions})

0 commit comments

Comments
 (0)