@@ -970,7 +970,6 @@ def execute(
970
970
)
971
971
972
972
ctx = _RuntimeOrchestrationContext (instance_id , self ._registry )
973
- version_failure = None
974
973
try :
975
974
# Rebuild local state by replaying old history into the orchestrator function
976
975
self ._logger .debug (
@@ -980,23 +979,6 @@ def execute(
980
979
for old_event in old_events :
981
980
self .process_event (ctx , old_event )
982
981
983
- # Process versioning if applicable
984
- execution_started_events = [e .executionStarted for e in old_events if e .HasField ("executionStarted" )]
985
- # We only check versioning if there are executionStarted events - otherwise, on the first replay when
986
- # ctx.version will be Null, we may invalidate orchestrations early depending on the versioning strategy.
987
- if self ._registry .versioning and len (execution_started_events ) > 0 :
988
- version_failure = self .evaluate_orchestration_versioning (
989
- self ._registry .versioning ,
990
- ctx .version
991
- )
992
- if version_failure :
993
- self ._logger .warning (
994
- f"Orchestration version did not meet worker versioning requirements. "
995
- f"Error action = '{ self ._registry .versioning .failure_strategy } '. "
996
- f"Version error = '{ version_failure } '"
997
- )
998
- raise pe .VersionFailureException
999
-
1000
982
# Get new actions by executing newly received events into the orchestrator function
1001
983
if self ._logger .level <= logging .DEBUG :
1002
984
summary = _get_new_event_summary (new_events )
@@ -1009,8 +991,8 @@ def execute(
1009
991
1010
992
except pe .VersionFailureException as ex :
1011
993
if self ._registry .versioning and self ._registry .versioning .failure_strategy == VersionFailureStrategy .FAIL :
1012
- if version_failure :
1013
- ctx .set_failed (version_failure )
994
+ if ex . error_details :
995
+ ctx .set_failed (ex . error_details )
1014
996
else :
1015
997
ctx .set_failed (ex )
1016
998
elif self ._registry .versioning and self ._registry .versioning .failure_strategy == VersionFailureStrategy .REJECT :
@@ -1068,6 +1050,19 @@ def process_event(
1068
1050
if event .executionStarted .version :
1069
1051
ctx ._version = event .executionStarted .version .value
1070
1052
1053
+ if self ._registry .versioning :
1054
+ version_failure = self .evaluate_orchestration_versioning (
1055
+ self ._registry .versioning ,
1056
+ ctx .version
1057
+ )
1058
+ if version_failure :
1059
+ self ._logger .warning (
1060
+ f"Orchestration version did not meet worker versioning requirements. "
1061
+ f"Error action = '{ self ._registry .versioning .failure_strategy } '. "
1062
+ f"Version error = '{ version_failure } '"
1063
+ )
1064
+ raise pe .VersionFailureException (version_failure )
1065
+
1071
1066
# deserialize the input, if any
1072
1067
input = None
1073
1068
if (
0 commit comments