@@ -99,6 +99,105 @@ def orchestrator(ctx, _):
9999 assert completed .actions [0 ].completeOrchestration .result .value == '"done"'
100100
101101
102+ def _timeout_history (executor , name ):
103+ started = list (_request (name ).newEvents )
104+ first = executor .execute ("owner" , [], started )
105+ section = first .actions [0 ].sendEntityMessage .entityLockRequested .criticalSectionId
106+ granted = [
107+ ph .new_event_sent_event (1 , str (ENTITY_IDS [0 ]), json .dumps ({"id" : section })),
108+ ph .new_event_raised_event (section , None ),
109+ ]
110+ second = executor .execute ("owner" , started , granted )
111+ call = second .actions [0 ].sendEntityMessage .entityOperationCalled
112+ pending = ph .new_event_sent_event (2 , str (ENTITY_IDS [0 ]), json .dumps ({"id" : call .requestId }))
113+ return started + granted + [pending ], section
114+
115+
116+ def test_legacy_pending_call_timeout_history_replays_through_activity ():
117+ def orchestrator (ctx , _ ):
118+ with (yield ctx .lock_entities (ENTITY_IDS )):
119+ call = ctx .call_entity (ENTITY_IDS [0 ], "get" )
120+ timeout = ctx .wait_for_external_event ("timeout" )
121+ yield task .when_any ([call , timeout ])
122+ return (yield ctx .call_activity ("after_timeout" ))
123+
124+ registry = worker ._Registry ()
125+ name = registry .add_orchestrator (orchestrator )
126+ executor = worker ._OrchestrationExecutor (registry , logging .getLogger ("test" ), JsonDataConverter ())
127+ history , _ = _timeout_history (executor , name )
128+ timed_out = ph .new_event_raised_event ("timeout" , None )
129+ result = executor .execute ("owner" , history , [timed_out ])
130+ assert [action .id for action in result .actions ] == [3 , 4 ]
131+ assert result .actions [0 ].sendEntityMessage .entityUnlockSent .targetInstanceId .value == str (ENTITY_IDS [1 ])
132+ assert result .actions [1 ].scheduleTask .name == "after_timeout"
133+ assert [release .entityUnlockSent .targetInstanceId .value for release in result .entity_lock_releases ] == [
134+ str (ENTITY_IDS [0 ]),
135+ ]
136+
137+ # Literal history from the old emitter: only b was released, so the next
138+ # activity kept ID 4 even though the call to a was still outstanding.
139+ history += [
140+ timed_out ,
141+ ph .new_event_sent_event (3 , str (ENTITY_IDS [1 ]), None ),
142+ ph .new_task_scheduled_event (4 , "after_timeout" ),
143+ ]
144+ assert not executor .execute ("owner" , history , [ph .new_orchestrator_started_event ()]).actions
145+ completed = executor .execute ("owner" , history , [ph .new_task_completed_event (4 , '"done"' )])
146+ assert len (completed .actions ) == 1
147+ assert completed .actions [0 ].id == 5
148+ assert completed .actions [0 ].completeOrchestration .orchestrationStatus == pb .ORCHESTRATION_STATUS_COMPLETED
149+ assert completed .actions [0 ].completeOrchestration .result .value == '"done"'
150+
151+
152+ @pytest .mark .parametrize ("mode" , ["result" , "status" , "activity" ])
153+ @pytest .mark .parametrize ("replay_exit" , [False , True ])
154+ def test_payload_failure_after_normal_pending_call_exit_releases_all_locks (mode , replay_exit ):
155+ def orchestrator (ctx , _ ):
156+ with (yield ctx .lock_entities (ENTITY_IDS )):
157+ call = ctx .call_entity (ENTITY_IDS [0 ], "get" )
158+ yield task .when_any ([call , ctx .wait_for_external_event ("timeout" )])
159+ if replay_exit :
160+ yield ctx .wait_for_external_event ("fail" )
161+ if mode == "result" :
162+ return LARGE_PAYLOAD
163+ if mode == "status" :
164+ ctx .set_custom_status (LARGE_PAYLOAD )
165+ yield ctx .call_activity ("after_timeout" , input = LARGE_PAYLOAD if mode == "activity" else None )
166+
167+ store = FakePayloadStore (threshold_bytes = 100 , max_stored_payload_bytes = 200 )
168+ stub = MagicMock ()
169+ with worker .TaskHubGrpcWorker (payload_store = store ) as w :
170+ w .add_orchestrator (orchestrator )
171+ executor = worker ._OrchestrationExecutor (w ._registry , logging .getLogger ("test" ), JsonDataConverter ())
172+ history , section = _timeout_history (executor , "orchestrator" )
173+ events = [ph .new_event_raised_event ("timeout" , None )]
174+ if replay_exit :
175+ exited = executor .execute ("owner" , history , events )
176+ assert len (exited .actions ) == 1
177+ assert exited .actions [0 ].id == 3
178+ history += events + [ph .new_event_sent_event (3 , str (ENTITY_IDS [1 ]), None )]
179+ events = [ph .new_event_raised_event ("fail" , None )]
180+ w ._execute_orchestrator (pb .OrchestratorRequest (
181+ instanceId = "owner" , pastEvents = history ,
182+ newEvents = events ,
183+ ), stub , "failure-token" )
184+
185+ response = stub .CompleteOrchestratorTask .call_args .args [0 ]
186+ assert response .completionToken == "failure-token"
187+ assert len (response .actions ) == (2 if replay_exit else 3 )
188+ if not replay_exit :
189+ assert response .actions [0 ].id == 3
190+ releases = [action .sendEntityMessage .entityUnlockSent for action in response .actions [:- 1 ]]
191+ expected_targets = [str (ENTITY_IDS [0 ])] if replay_exit else [str (ENTITY_IDS [1 ]), str (ENTITY_IDS [0 ])]
192+ assert [release .targetInstanceId .value for release in releases ] == expected_targets
193+ assert all (release .parentInstanceId .value == "owner" for release in releases )
194+ assert all (release .criticalSectionId == section for release in releases )
195+ completion = response .actions [- 1 ].completeOrchestration
196+ assert completion .orchestrationStatus == pb .ORCHESTRATION_STATUS_FAILED
197+ assert completion .failureDetails .isNonRetriable
198+ assert not response .HasField ("customStatus" )
199+
200+
102201def test_release_snapshot_retains_pending_locks_after_completed_calls ():
103202 entity_ctx = OrchestrationEntityContext ("owner" )
104203 entity_ctx .emit_acquire_message ("section" , ENTITY_IDS )
0 commit comments