Enable replay V2 (#291)

Author: davidmrdavid

azure/durable_functions/models/DurableOrchestrationContext.py

@@ -1,3 +1,4 @@
+from azure.durable_functions.models.ReplaySchema import ReplaySchema
 import json
 import datetime
 import inspect
@@ -30,7 +31,7 @@ class DurableOrchestrationContext:
     # parameter names are as defined by JSON schema and do not conform to PEP8 naming conventions
     def __init__(self,
                  history: List[Dict[Any, Any]], instanceId: str, isReplaying: bool,
-                 parentInstanceId: str, input: Any = None, **kwargs):
+                 parentInstanceId: str, input: Any = None, upperSchemaVersion: int = 0, **kwargs):
         self._histories: List[HistoryEvent] = [HistoryEvent(**he) for he in history]
         self._instance_id: str = instanceId
         self._is_replaying: bool = isReplaying
@@ -45,8 +46,11 @@ def __init__(self,
         self._current_utc_datetime: datetime.datetime = \
             self.decision_started_event.timestamp
         self._new_uuid_counter = 0
-        self.actions: List[List[Action]] = []
         self._function_context: FunctionContext = FunctionContext(**kwargs)
+        self._replay_schema = ReplaySchema(upperSchemaVersion)
+        self.actions: List[List[Action]] = []
+        if self._replay_schema == ReplaySchema.V2:
+            self.actions.append([])
 
         # make _input always a string
         # (consistent with Python Functions generic trigger/input bindings)
@@ -240,7 +244,7 @@ def task_all(self, activities: List[Task]) -> TaskSet:
         TaskSet
             The results of all activities.
         """
-        return task_all(tasks=activities)
+        return task_all(tasks=activities, replay_schema=self._replay_schema)
 
     def task_any(self, activities: List[Task]) -> TaskSet:
         """Schedule the execution of all activities.
@@ -260,7 +264,7 @@ def task_any(self, activities: List[Task]) -> TaskSet:
         TaskSet
             The first [[Task]] instance to complete.
         """
-        return task_any(tasks=activities)
+        return task_any(tasks=activities, replay_schema=self._replay_schema)
 
     def set_custom_status(self, status: Any):
         """Set the customized orchestration status for your orchestrator function.

azure/durable_functions/models/OrchestratorState.py

@@ -1,6 +1,8 @@
 import json
 from typing import List, Any, Dict, Optional, Union
 
+from azure.durable_functions.models.ReplaySchema import ReplaySchema
+
 from .utils.json_utils import add_attrib
 from azure.durable_functions.models.actions.Action import Action
 
@@ -16,13 +18,15 @@ def __init__(self,
                  is_done: bool,
                  actions: List[List[Action]],
                  output: Any,
+                 replay_schema: ReplaySchema,
                  error: str = None,
                  custom_status: Any = None):
         self._is_done: bool = is_done
         self._actions: List[List[Action]] = actions
         self._output: Any = output
         self._error: Optional[str] = error
         self._custom_status: Any = custom_status
+        self._replay_schema: ReplaySchema = replay_schema
 
     @property
     def actions(self) -> List[List[Action]]:
@@ -66,6 +70,11 @@ def custom_status(self):
         """Get the JSON-serializable value used by DurableOrchestrationContext.SetCustomStatus."""
         return self._custom_status
 
+    @property
+    def schema_version(self):
+        """Get the Replay Schema represented in this OrchestratorState payload."""
+        return self._replay_schema.value
+
     def to_json(self) -> Dict[str, Union[str, int]]:
         """Convert object into a json dictionary.
 
@@ -76,6 +85,8 @@ def to_json(self) -> Dict[str, Union[str, int]]:
         """
         json_dict: Dict[str, Union[str, int]] = {}
         add_attrib(json_dict, self, '_is_done', 'isDone')
+        if self._replay_schema != ReplaySchema.V1:
+            add_attrib(json_dict, self, 'schema_version', 'schemaVersion')
         self._add_actions(json_dict)
         if not (self._output is None):
             json_dict['output'] = self._output

azure/durable_functions/models/ReplaySchema.py

@@ -0,0 +1,8 @@
+from enum import Enum
+
+
+class ReplaySchema(Enum):
+    """Enum representing the ReplaySchemas supported by this SDK version."""
+
+    V1 = 0
+    V2 = 1

azure/durable_functions/models/actions/ActionType.py

@@ -14,3 +14,5 @@ class ActionType(IntEnum):
     CALL_ENTITY = 7
     CALL_HTTP: int = 8
     SIGNAL_ENTITY: int = 9
+    WHEN_ANY = 11
+    WHEN_ALL = 12

azure/durable_functions/models/actions/CompoundAction.py

@@ -0,0 +1,35 @@
+from typing import Dict, Union
+
+from .Action import Action
+from ..utils.json_utils import add_attrib
+from typing import List
+from abc import abstractmethod
+
+
+class CompoundAction(Action):
+    """Defines the structure of the WhenAll Action object.
+
+    Provides the information needed by the durable extension to be able to invoke WhenAll tasks.
+    """
+
+    def __init__(self, compoundTasks: List[Action]):
+        self.compound_actions = list(map(lambda x: x.to_json(), compoundTasks))
+
+    @property
+    @abstractmethod
+    def action_type(self) -> int:
+        """Get this object's action type as an integer."""
+        ...
+
+    def to_json(self) -> Dict[str, Union[str, int]]:
+        """Convert object into a json dictionary.
+
+        Returns
+        -------
+        Dict[str, Union[str, int]]
+            The instance of the class converted into a json dictionary
+        """
+        json_dict: Dict[str, Union[str, int]] = {}
+        add_attrib(json_dict, self, 'action_type', 'actionType')
+        add_attrib(json_dict, self, 'compound_actions', 'compoundActions')
+        return json_dict

azure/durable_functions/models/actions/WhenAllAction.py

@@ -0,0 +1,14 @@
+from .ActionType import ActionType
+from azure.durable_functions.models.actions.CompoundAction import CompoundAction
+
+
+class WhenAllAction(CompoundAction):
+    """Defines the structure of the WhenAll Action object.
+
+    Provides the information needed by the durable extension to be able to invoke WhenAll tasks.
+    """
+
+    @property
+    def action_type(self) -> int:
+        """Get the type of action this class represents."""
+        return ActionType.WHEN_ALL

azure/durable_functions/models/actions/WhenAnyAction.py

@@ -0,0 +1,14 @@
+from azure.durable_functions.models.actions.CompoundAction import CompoundAction
+from .ActionType import ActionType
+
+
+class WhenAnyAction(CompoundAction):
+    """Defines the structure of the WhenAll Action object.
+
+    Provides the information needed by the durable extension to be able to invoke WhenAll tasks.
+    """
+
+    @property
+    def action_type(self) -> int:
+        """Get the type of action this class represents."""
+        return ActionType.WHEN_ANY

azure/durable_functions/orchestrator.py

@@ -5,6 +5,8 @@
 """
 from typing import Callable, Iterator, Any, Generator
 
+from azure.durable_functions.models.ReplaySchema import ReplaySchema
+
 from .models import (
     DurableOrchestrationContext,
     Task,
@@ -55,6 +57,7 @@ def handle(self, context: DurableOrchestrationContext):
         # `fn_output` is the return value instead of a generator
         if not isinstance(fn_output, Iterator):
             orchestration_state = OrchestratorState(
+                replay_schema=self.durable_context._replay_schema,
                 is_done=True,
                 output=fn_output,
                 actions=self.durable_context.actions,
@@ -75,6 +78,7 @@ def handle(self, context: DurableOrchestrationContext):
                         # `will_continue_as_new` essentially "tracks"
                         # whether or not the orchestration is done.
                         orchestration_state = OrchestratorState(
+                            replay_schema=self.durable_context._replay_schema,
                             is_done=self.durable_context.will_continue_as_new,
                             output=None,
                             actions=self.durable_context.actions,
@@ -95,13 +99,15 @@ def handle(self, context: DurableOrchestrationContext):
 
             except StopIteration as sie:
                 orchestration_state = OrchestratorState(
+                    replay_schema=self.durable_context._replay_schema,
                     is_done=True,
                     output=sie.value,
                     actions=self.durable_context.actions,
                     custom_status=self.durable_context.custom_status)
             except Exception as e:
                 exception_str = str(e)
                 orchestration_state = OrchestratorState(
+                    replay_schema=self.durable_context._replay_schema,
                     is_done=False,
                     output=None,  # Should have no output, after generation range
                     actions=self.durable_context.actions,
@@ -135,12 +141,17 @@ def _add_to_actions(self, generation_state):
         if self.durable_context.will_continue_as_new:
             return
         if not generation_state._is_yielded:
-            if (isinstance(generation_state, Task)
-                    and hasattr(generation_state, "action")):
-                self.durable_context.actions.append([generation_state.action])
-            elif (isinstance(generation_state, TaskSet)
-                    and hasattr(generation_state, "actions")):
-                self.durable_context.actions.append(generation_state.actions)
+            if isinstance(generation_state, Task):
+                if self.durable_context._replay_schema == ReplaySchema.V1:
+                    self.durable_context.actions.append([generation_state.action])
+                else:
+                    self.durable_context.actions[0].append(generation_state.action)
+
+            elif isinstance(generation_state, TaskSet):
+                if self.durable_context._replay_schema == ReplaySchema.V1:
+                    self.durable_context.actions.append(generation_state.actions)
+                else:
+                    self.durable_context.actions[0].append(generation_state.actions)
             generation_state._is_yielded = True
 
     def _update_timestamp(self):

azure/durable_functions/tasks/task_all.py

@@ -1,3 +1,5 @@
+from azure.durable_functions.models.actions.WhenAllAction import WhenAllAction
+from azure.durable_functions.models.ReplaySchema import ReplaySchema
 from datetime import datetime
 from typing import List, Optional, Any
 
@@ -6,7 +8,7 @@
 from ..models.actions import Action
 
 
-def task_all(tasks: List[Task]):
+def task_all(tasks: List[Task], replay_schema: ReplaySchema):
     """Determine the state of scheduling the activities for execution with retry options.
 
     Parameters
@@ -33,7 +35,10 @@ def task_all(tasks: List[Task]):
     for task in tasks:
         # Add actions and results
         if isinstance(task, TaskSet):
-            actions.extend(task.actions)
+            if replay_schema == ReplaySchema.V1:
+                actions.extend(task.actions)
+            else:
+                actions.append(task.actions)
         else:
             # We know it's an atomic Task
             actions.append(task.action)
@@ -62,6 +67,9 @@ def task_all(tasks: List[Task]):
         results = []
         end_time = None
 
+    if replay_schema == ReplaySchema.V2:
+        actions = WhenAllAction(actions)
+
     # Construct TaskSet
     taskset = TaskSet(
         is_completed=is_completed,

azure/durable_functions/tasks/task_any.py

@@ -1,7 +1,9 @@
+from azure.durable_functions.models.actions.WhenAnyAction import WhenAnyAction
+from azure.durable_functions.models.ReplaySchema import ReplaySchema
 from ..models.TaskSet import TaskSet
 
 
-def task_any(tasks):
+def task_any(tasks, replay_schema: ReplaySchema):
     """Determine whether any of the given tasks is completed.
 
     Parameters
@@ -22,8 +24,10 @@ def task_any(tasks):
     error_message = []
     for task in tasks:
         if isinstance(task, TaskSet):
-            for action in task.actions:
-                all_actions.append(action)
+            if replay_schema == ReplaySchema.V1:
+                all_actions.extend(task.actions)
+            else:
+                all_actions.append(task.actions)
         else:
             all_actions.append(task.action)
 
@@ -35,6 +39,9 @@ def task_any(tasks):
 
     completed_tasks.sort(key=lambda t: t.timestamp)
 
+    if replay_schema == ReplaySchema.V2:
+        all_actions = WhenAnyAction(all_actions)
+
     if len(faulted_tasks) == len(tasks):
         return TaskSet(True, all_actions, None, is_faulted=True, exception=Exception(
             f"All tasks have failed, errors messages in all tasks:{error_message}"))