Merge pull request #397 from Azure/dev

Promote dev to main for release

Author: davidmrdavid

SECURITY.md

@@ -0,0 +1,41 @@
+<!-- BEGIN MICROSOFT SECURITY.MD V0.0.7 BLOCK -->
+
+## Security
+
+Microsoft takes the security of our software products and services seriously, which includes all source code repositories managed through our GitHub organizations, which include [Microsoft](https://github.com/Microsoft), [Azure](https://github.com/Azure), [DotNet](https://github.com/dotnet), [AspNet](https://github.com/aspnet), [Xamarin](https://github.com/xamarin), and [our GitHub organizations](https://opensource.microsoft.com/).
+
+If you believe you have found a security vulnerability in any Microsoft-owned repository that meets [Microsoft's definition of a security vulnerability](https://aka.ms/opensource/security/definition), please report it to us as described below.
+
+## Reporting Security Issues
+
+**Please do not report security vulnerabilities through public GitHub issues.**
+
+Instead, please report them to the Microsoft Security Response Center (MSRC) at [https://msrc.microsoft.com/create-report](https://aka.ms/opensource/security/create-report).
+
+If you prefer to submit without logging in, send email to [[email protected]](mailto:[email protected]).  If possible, encrypt your message with our PGP key; please download it from the [Microsoft Security Response Center PGP Key page](https://aka.ms/opensource/security/pgpkey).
+
+You should receive a response within 24 hours. If for some reason you do not, please follow up via email to ensure we received your original message. Additional information can be found at [microsoft.com/msrc](https://aka.ms/opensource/security/msrc). 
+
+Please include the requested information listed below (as much as you can provide) to help us better understand the nature and scope of the possible issue:
+
+  * Type of issue (e.g. buffer overflow, SQL injection, cross-site scripting, etc.)
+  * Full paths of source file(s) related to the manifestation of the issue
+  * The location of the affected source code (tag/branch/commit or direct URL)
+  * Any special configuration required to reproduce the issue
+  * Step-by-step instructions to reproduce the issue
+  * Proof-of-concept or exploit code (if possible)
+  * Impact of the issue, including how an attacker might exploit the issue
+
+This information will help us triage your report more quickly.
+
+If you are reporting for a bug bounty, more complete reports can contribute to a higher bounty award. Please visit our [Microsoft Bug Bounty Program](https://aka.ms/opensource/security/bounty) page for more details about our active programs.
+
+## Preferred Languages
+
+We prefer all communications to be in English.
+
+## Policy
+
+Microsoft follows the principle of [Coordinated Vulnerability Disclosure](https://aka.ms/opensource/security/cvd).
+
+<!-- END MICROSOFT SECURITY.MD BLOCK -->

azure/durable_functions/models/Task.py

@@ -171,6 +171,11 @@ def __init__(self, tasks: List[TaskBase], compound_action_constructor=None):
         if len(self.children) == 0:
             self.state = TaskState.SUCCEEDED
 
+        # Sub-tasks may have already completed, so we process them
+        for child in self.children:
+            if not(child.state is TaskState.RUNNING):
+                self.handle_completion(child)
+
     def handle_completion(self, child: TaskBase):
         """Manage sub-task completion events.
 

azure/durable_functions/models/TaskOrchestrationExecutor.py

@@ -68,6 +68,22 @@ def execute(self, context: DurableOrchestrationContext,
         self.context = context
         evaluated_user_code = fn(context)
 
+        # The minimum History size is 2, in the shape: [OrchestratorStarted, ExecutionStarted].
+        # At the start of replay, the `is_replaying` flag is determined from the
+        # ExecutionStarted event.
+        # For some reason, OrchestratorStarted does not update its `isPlayed` field.
+        if len(history) < 2:
+            err_message = "Internal Durable Functions error: "\
+                + f"received History array of size {len(history)} "\
+                + "when a minimum size of 2 is expected. "\
+                + "Please report this issue at "\
+                + "https://github.com/Azure/azure-functions-durable-python/issues."
+            raise Exception(err_message)
+
+        # Set initial is_replaing state.
+        execution_started_event = history[1]
+        self.current_task.is_played = execution_started_event.is_played
+
         # If user code is a generator, then it uses `yield` statements (the DF API)
         # and so we iterate through the DF history, generating tasks and populating
         # them with values when the history provides them

tests/orchestrator/test_is_replaying_flag.py

@@ -47,7 +47,7 @@ def add_timer_action(state: OrchestratorState, fire_at: datetime):
 
 def test_is_replaying_initial_value():
 
-    context_builder = ContextBuilder("")
+    context_builder = ContextBuilder("", is_replaying=False)
     result = get_orchestration_property(
         context_builder, generator_function, "durable_context")
 

tests/orchestrator/test_sequential_orchestrator.py

@@ -24,6 +24,26 @@ def generator_function(context):
 
     return outputs
 
+def generator_function_multi_yield_when_all(context):
+    outputs = []
+
+    task1 = context.call_activity("Hello", "Tokyo")
+    yield context.task_all([task1])
+    result = yield context.task_all([task1])
+
+    return result
+
+def generator_function_is_replaying(context):
+    outputs = []
+
+    outputs.append(context.is_replaying)
+    yield context.call_activity("Hello", "Tokyo")
+    outputs.append(context.is_replaying)
+    yield context.call_activity("Hello", "Seattle")
+    outputs.append(context.is_replaying)
+    yield context.call_activity("Hello", "London")
+    return outputs
+
 def generator_function_no_yield(context):
     outputs = []
 
@@ -150,11 +170,11 @@ def add_hello_action(state: OrchestratorState, input_: str):
     state.actions.append([action])
 
 def add_hello_completed_events(
-        context_builder: ContextBuilder, id_: int, result: str):
+        context_builder: ContextBuilder, id_: int, result: str, is_played=False):
     context_builder.add_task_scheduled_event(name='Hello', id_=id_)
     context_builder.add_orchestrator_completed_event()
     context_builder.add_orchestrator_started_event()
-    context_builder.add_task_completed_event(id_=id_, result=result)
+    context_builder.add_task_completed_event(id_=id_, result=result, is_played=is_played)
 
 
 def add_hello_failed_events(
@@ -286,6 +306,42 @@ def test_tokyo_and_seattle_and_london_state():
     assert_valid_schema(result)
     assert_orchestration_state_equals(expected, result)
 
+def test_multi_when_all_yield():
+    context_builder = ContextBuilder('test_simple_function')
+    add_hello_completed_events(context_builder, 0, "\"Hello Tokyo!\"")
+
+    result = get_orchestration_state_result(
+        context_builder, generator_function_multi_yield_when_all)
+
+    expected_state = base_expected_state(
+        ['Hello Tokyo!'])
+    add_hello_action(expected_state, 'Tokyo')
+    expected_state._is_done = True
+    expected = expected_state.to_json()
+
+    assert_valid_schema(result)
+    assert_orchestration_state_equals(expected, result)
+
+def test_sequential_is_replaying():
+    context_builder = ContextBuilder('test_simple_function', is_replaying=True)
+    add_hello_completed_events(context_builder, 0, "\"Hello Tokyo!\"", True)
+    add_hello_completed_events(context_builder, 1, "\"Hello Seattle!\"", True)
+    add_hello_completed_events(context_builder, 2, "\"Hello London!\"", True)
+
+    result = get_orchestration_state_result(
+        context_builder, generator_function_is_replaying)
+
+    expected_state = base_expected_state(
+        [True, True, True])
+    add_hello_action(expected_state, 'Tokyo')
+    add_hello_action(expected_state, 'Seattle')
+    add_hello_action(expected_state, 'London')
+    expected_state._is_done = True
+    expected = expected_state.to_json()
+
+    assert_valid_schema(result)
+    assert_orchestration_state_equals(expected, result)
+
 def test_sequential_orchestration_no_yield():
     context_builder = ContextBuilder('test_simple_function')
     add_hello_completed_events(context_builder, 0, "\"Hello London!\"")

tests/test_utils/ContextBuilder.py

@@ -14,7 +14,7 @@
 
 
 class ContextBuilder:
-    def __init__(self, name: str="", increase_time: bool = True, starting_time: Optional[datetime] = None, replay_schema: ReplaySchema = ReplaySchema.V1):
+    def __init__(self, name: str="", increase_time: bool = True, starting_time: Optional[datetime] = None, is_replaying=False, replay_schema: ReplaySchema = ReplaySchema.V1):
         self.increase_time = increase_time
         self.instance_id = uuid.uuid4()
         self.is_replaying: bool = False
@@ -28,7 +28,7 @@ def __init__(self, name: str="", increase_time: bool = True, starting_time: Opti
         self.upperSchemaVersion = replay_schema.value
 
         self.add_orchestrator_started_event()
-        self.add_execution_started_event(name)
+        self.add_execution_started_event(name, is_played=is_replaying)
 
     def get_base_event(
             self, event_type: HistoryEventType, id_: int = -1,
@@ -87,8 +87,8 @@ def add_task_scheduled_event(
         event.Input_ = input_
         self.history_events.append(event)
 
-    def add_task_completed_event(self, id_: int, result):
-        event = self.get_base_event(HistoryEventType.TASK_COMPLETED)
+    def add_task_completed_event(self, id_: int, result, is_played=False):
+        event = self.get_base_event(HistoryEventType.TASK_COMPLETED, is_played=is_played)
         event.Result = result
         event.TaskScheduledId = id_
         self.history_events.append(event)
@@ -116,8 +116,8 @@ def add_timer_fired_event(self, id_: int, fire_at: str, is_played: bool = True):
         self.history_events.append(event)
 
     def add_execution_started_event(
-            self, name: str, version: str = '', input_=None):
-        event = self.get_base_event(HistoryEventType.EXECUTION_STARTED, is_played=True)
+            self, name: str, version: str = '', input_=None, is_played=True):
+        event = self.get_base_event(HistoryEventType.EXECUTION_STARTED, is_played=is_played)
         event.orchestration_instance = OrchestrationInstance()
         self.instance_id = event.orchestration_instance.instance_id
         event.Name = name