Fixed: get_version() should always return the same result per workflow execution.

Author: firdaus

cadence/clock_decision_context.py

@@ -9,7 +9,7 @@
 from cadence.conversions import args_to_json
 from cadence.decision_loop import ReplayDecider, DecisionContext
 from cadence.exceptions import CancellationException
-from cadence.marker import MarkerHandler
+from cadence.marker import MarkerHandler, MarkerInterface, MarkerResult
 from cadence.util import OpenRequestInfo
 
 logger = logging.getLogger(__name__)
@@ -86,15 +86,14 @@ def handle_timer_canceled(self, event: HistoryEvent):
             self.timer_cancelled(started_event_id, None)
 
     def get_version(self, change_id: str, min_supported: int, max_supported) -> int:
-        def func(stored):
-            if stored:
-                return None
-            else:
-                return json.dumps(max_supported)
+        def func():
+            return json.dumps(max_supported)
 
         result: bytes = self.version_handler.handle(change_id, func)
-        if not result:
-            return DEFAULT_VERSION
+        if result is None:
+            result = json.dumps(DEFAULT_VERSION)
+            self.version_handler.set_data(change_id, result)
+            self.version_handler.mark_replayed(change_id)  # so that we don't ever emit a MarkerRecorded for this
 
         version: int = json.loads(result)
         self.validate_version(change_id, version, min_supported, max_supported)
@@ -106,6 +105,9 @@ def validate_version(self, change_id: str, version: int, min_supported: int, max
                             f"Supported version is between {min_supported} and {max_supported}.")
 
     def handle_marker_recorded(self, event: HistoryEvent):
+        """
+        Will be executed more than once for the same event.
+        """
         attributes = event.marker_recorded_event_attributes
         name: str = attributes.marker_name
         if SIDE_EFFECT_MARKER_NAME == name:
@@ -116,7 +118,12 @@ def handle_marker_recorded(self, event: HistoryEvent):
             # TODO
             # handleLocalActivityMarker(attributes);
             pass
-        elif MUTABLE_SIDE_EFFECT_MARKER_NAME != name and VERSION_MARKER_NAME != name:
+        elif VERSION_MARKER_NAME == name:
+            marker_data = MarkerInterface.from_event_attributes(attributes)
+            change_id: str = marker_data.get_id()
+            data: bytes = marker_data.get_data()
+            self.version_handler.mutable_marker_results[change_id] = MarkerResult(data=data)
+        elif MUTABLE_SIDE_EFFECT_MARKER_NAME != name:
             # TODO
             # if (log.isWarnEnabled()) {
             #       log.warn("Unexpected marker: " + event);

cadence/marker.py

@@ -70,6 +70,7 @@ def get_id(self) -> str:
 class MarkerResult:
     data: bytes = None
     access_count: int = 0
+    replayed = False
 
 
 @dataclass
@@ -80,29 +81,42 @@ class MarkerHandler:
 
     def record_mutable_marker(self, id: str, event_id: int, data: bytes, access_count: int):
         marker = MarkerData.create(id=id, event_id=event_id, data=data, access_count=access_count)
-        self.mutable_marker_results[id] = MarkerResult(data=data)
+        if id in self.mutable_marker_results:
+            self.mutable_marker_results[id].replayed = True
+        else:
+            self.mutable_marker_results[id] = MarkerResult(data=data)
         self.decision_context.record_marker(self.marker_name, marker.get_header(), data)
 
-    def handle(self, id: str, func) -> bytes:
-        result: MarkerResult = self.mutable_marker_results.get(id)
-        stored: bytes = None
-        if result:
-            stored = result.data
-        event_id = self.decision_context.decider.next_decision_event_id
-        access_count = 0 if result is None else result.access_count
-        if self.decision_context.is_replaying():
-            data: bytes = self.get_marker_data_from_history(event_id, id, access_count)
-            if data:
-                self.record_mutable_marker(id, event_id, data, access_count)
-                return data
-            return stored
-        to_store = func(stored)
-        if to_store:
-            data = to_store
-            self.record_mutable_marker(id, event_id, data, access_count)
-            return to_store
-        return stored
+    # Sets data without creating a decision - used when DEFAULT_VERSION is the implicit current version
+    def set_data(self, id, data: bytes):
+        self.mutable_marker_results[id] = MarkerResult(data=data)
 
+    def mark_replayed(self, id):
+        self.mutable_marker_results[id].replayed = True
+
+    def handle(self, id: str, func) -> Optional[bytes]:
+        event_id = self.decision_context.decider.next_decision_event_id
+        result: MarkerResult = self.mutable_marker_results.get(id)
+        if result or self.decision_context.is_replaying():
+            if result:
+                if self.decision_context.is_replaying() and not result.replayed:
+                    # Need to insert marker to ensure that event_id is incremented
+                    self.record_mutable_marker(id, event_id, result.data, 0)
+                return result.data
+            else:
+                return None
+        else:
+            to_store = func()
+            if to_store:
+                data = to_store
+                self.record_mutable_marker(id, event_id, data, 0)
+                return to_store
+            else:
+                # TODO: Should this ever happen? - at least for version it will never happen
+                pass
+
+    # This method is currently not being used - after adopting the version logic from the
+    # Golang client
     def get_marker_data_from_history(self, event_id: int, marker_id: str, expected_access_count: int) -> \
             Optional[bytes]:
         event: HistoryEvent = self.decision_context.decider.get_optional_decision_event(event_id)

cadence/tests/test_marker.py

@@ -151,8 +151,8 @@ def callback(stored):
     handler = MarkerHandler(decision_context=decision_context, marker_name="the-marker-name")
     handler.mutable_marker_results["the-id"] = MarkerResult(data=b'123', access_count=35)
     ret = handler.handle("the-id", callback)
-    assert ret == b'blah-blah'
-    assert len(decision_context.decider.decisions) == 1
+    assert ret == b'123'
+    assert len(decision_context.decider.decisions) == 0
 
 
 def test_handle_replaying_no_history(decision_context):
@@ -175,8 +175,8 @@ def callback(stored):
     handler = MarkerHandler(decision_context=decision_context, marker_name="the-marker-name")
     handler.mutable_marker_results["the-id"] = MarkerResult(data=b'123', access_count=35)
     ret = handler.handle("the-id", callback)
-    assert ret == b'456'
-    assert len(decision_context.decider.decisions) == 1
+    assert ret == b'123'
+    assert len(decision_context.decider.decisions) == 0
 
 
 def test_handle_not_replaying_callback_returns_none(decision_context):

cadence/tests/test_version.py

@@ -88,8 +88,8 @@ def version_decision_context(version_marker_recorded_event):
 def test_clock_decision_context_from_replay(version_decision_context):
     version_decision_context.workflow_clock.set_replaying(True)
     version = version_decision_context.workflow_clock.get_version("abc", 1, 5)
-    assert version == 4
-    assert len(version_decision_context.decider.decisions) == 1
+    assert version == -1
+    assert len(version_decision_context.decider.decisions) == 0
 
 
 def test_validate_version(version_decision_context):

cadence/tests/test_workflow_get_version.py

@@ -11,8 +11,10 @@
 v1_hits = 0
 v2_hits = 0
 
-version_found_in_v2_step_1 = None
-version_found_in_v2_step_2 = None
+version_found_in_v2_step_1_0 = None
+version_found_in_v2_step_1_1 = None
+version_found_in_v2_step_2_0 = None
+version_found_in_v2_step_2_1 = None
 
 v2_done = False
 
@@ -41,13 +43,16 @@ def __init__(self):
 
     async def get_greetings(self):
         global v2_hits
-        global version_found_in_v2_step_1, version_found_in_v2_step_2
+        global version_found_in_v2_step_1_0, version_found_in_v2_step_1_1
+        global version_found_in_v2_step_2_0, version_found_in_v2_step_2_1
         global v2_done
         v2_hits += 1
 
-        version_found_in_v2_step_1 = Workflow.get_version("first-item", DEFAULT_VERSION, 2)
+        version_found_in_v2_step_1_0 = Workflow.get_version("first-item", DEFAULT_VERSION, 2)
+        version_found_in_v2_step_1_1 = Workflow.get_version("first-item", DEFAULT_VERSION, 2)
         await Workflow.sleep(60)
-        version_found_in_v2_step_2 = Workflow.get_version("first-item", DEFAULT_VERSION, 2)
+        version_found_in_v2_step_2_0 = Workflow.get_version("first-item", DEFAULT_VERSION, 2)
+        version_found_in_v2_step_2_1 = Workflow.get_version("first-item", DEFAULT_VERSION, 2)
         v2_done = True
 
 
@@ -74,8 +79,12 @@ def test_workflow_workflow_get_version():
 
     assert v1_hits == 1
     assert v2_hits == 1
-    assert version_found_in_v2_step_1 == DEFAULT_VERSION
-    assert version_found_in_v2_step_2 == 2
+    assert version_found_in_v2_step_1_0 == DEFAULT_VERSION
+    assert version_found_in_v2_step_1_1 == DEFAULT_VERSION
+    assert version_found_in_v2_step_2_0 == DEFAULT_VERSION
+    assert version_found_in_v2_step_2_1 == DEFAULT_VERSION
+
+    # TODO: Assert that there are no markers recorded
 
     print("Stopping workers")
     worker.stop(background=True)

cadence/tests/test_workflow_get_version_single.py

@@ -8,8 +8,10 @@
 TASK_LIST = "TestWorkflowGetVersionSingle"
 DOMAIN = "sample"
 
-version_found_in_step_1 = None
-version_found_in_step_2 = None
+version_found_in_step_1_0 = None
+version_found_in_step_1_1 = None
+version_found_in_step_2_0 = None
+version_found_in_step_2_1 = None
 
 
 class TestWorkflowGetVersionSingle:
@@ -24,10 +26,13 @@ def __init__(self):
         pass
 
     async def get_greetings(self):
-        global version_found_in_step_1, version_found_in_step_2
-        version_found_in_step_1 = Workflow.get_version("first-item", DEFAULT_VERSION, 2)
+        global version_found_in_step_1_0, version_found_in_step_1_1
+        global version_found_in_step_2_0, version_found_in_step_2_1
+        version_found_in_step_1_0 = Workflow.get_version("first-item", DEFAULT_VERSION, 2)
+        version_found_in_step_1_1 = Workflow.get_version("first-item", DEFAULT_VERSION, 2)
         await Workflow.sleep(60)
-        version_found_in_step_2 = Workflow.get_version("first-item", DEFAULT_VERSION, 2)
+        version_found_in_step_2_0 = Workflow.get_version("first-item", DEFAULT_VERSION, 2)
+        version_found_in_step_2_1 = Workflow.get_version("first-item", DEFAULT_VERSION, 2)
 
 
 def test_workflow_workflow_get_version_single():
@@ -40,8 +45,12 @@ def test_workflow_workflow_get_version_single():
     workflow: TestWorkflowGetVersionSingle = client.new_workflow_stub(TestWorkflowGetVersionSingle)
     workflow.get_greetings()
 
-    assert version_found_in_step_1 == 2
-    assert version_found_in_step_2 == 2
+    assert version_found_in_step_1_0 == 2
+    assert version_found_in_step_1_1 == 2
+    assert version_found_in_step_2_0 == 2
+    assert version_found_in_step_2_1 == 2
+
+    # TODO: Assert that there is only a single marker recorded
 
     print("Stopping workers")
     worker.stop(background=True)