feat: Add replica to step creation (and step-by-name query)

Author: Alan Christie

tests/wapi_adapter.py

@@ -136,8 +136,12 @@ def create_running_workflow_step(
         *,
         running_workflow_id: str,
         step: str,
+        replica: int = 0,
         prior_running_workflow_step_id: str | None = None,
     ) -> dict[str, Any]:
+        if replica:
+            assert replica > 0
+
         UnitTestWorkflowAPIAdapter.lock.acquire()
         with open(_RUNNING_WORKFLOW_STEP_PICKLE_FILE, "rb") as pickle_file:
             running_workflow_step = Unpickler(pickle_file).load()
@@ -150,6 +154,7 @@ def create_running_workflow_step(
             "name": step,
             "done": False,
             "success": False,
+            "replica": replica,
             "variables": {},
             "running_workflow": {"id": running_workflow_id},
         }
@@ -177,24 +182,28 @@ def get_running_workflow_step(
             return {}, 0
         response = running_workflow_step[running_workflow_step_id]
         response["id"] = running_workflow_step_id
+        if response["replica"] == 0:
+            _ = response.pop("replica")
         return response, 0
 
     def get_running_workflow_step_by_name(
-        self, *, name: str, running_workflow_id: str
+        self, *, name: str, running_workflow_id: str, replica: int = 0
     ) -> dict[str, Any]:
+        if replica:
+            assert replica > 0
         UnitTestWorkflowAPIAdapter.lock.acquire()
         with open(_RUNNING_WORKFLOW_STEP_PICKLE_FILE, "rb") as pickle_file:
             running_workflow_step = Unpickler(pickle_file).load()
         UnitTestWorkflowAPIAdapter.lock.release()
 
-        print(f"name={name} running_workflow_id={running_workflow_id}")
         for rwfs_id, record in running_workflow_step.items():
-            print(f"rwfs_id={rwfs_id} record={record}")
             if record["running_workflow"]["id"] != running_workflow_id:
                 continue
-            if record["name"] == name:
+            if record["name"] == name and record["replica"] == replica:
                 response = record
                 response["id"] = rwfs_id
+                if record["replica"] == 0:
+                    _ = response.pop("replica")
                 return response, 0
         return {}, 0
 

workflow/workflow_abc.py

@@ -205,9 +205,14 @@ def create_running_workflow_step(
         *,
         running_workflow_id: str,
         step: str,
+        replica: int = 0,
         prior_running_workflow_step_id: str | None = None,
     ) -> tuple[dict[str, Any], int]:
-        """Create a RunningWorkflowStep Record (from a RunningWorkflow)"""
+        """Create a RunningWorkflowStep Record (from a RunningWorkflow).
+        If this is a replica (concurrent execution) of a step the replica
+        value must be set to a value greater than 0. The replica is unique
+        for a given step and is used to distinguish between running workflow steps
+        generated from the same step name."""
         # Should return:
         # {
         #    "id": "r-workflow-step-00000000-0000-0000-0000-000000000001",
@@ -244,10 +249,12 @@ def get_running_workflow_step(
 
     @abstractmethod
     def get_running_workflow_step_by_name(
-        self, *, name: str, running_workflow_id: str
+        self, *, name: str, running_workflow_id: str, replica: int = 0
     ) -> tuple[dict[str, Any], int]:
         """Get a RunningWorkflowStep Record given a step name
-        (and its RUnningWorkflow ID)"""
+        (and its RunningWorkflow ID). For steps that may be replicated
+        the replica, a value of 1 or higher, is used to identify the specific replica.
+        """
         # Should return:
         # {
         #       "id": "r-workflow-step-00000000-0000-0000-0000-000000000001",