Merge pull request #371 from Azure/dev

Promote dev to main for 1.1.4 release

Author: davidmrdavid

.github/ISSUE_TEMPLATE/---bug-report.md

@@ -7,6 +7,13 @@ assignees: ''
 
 ---
 
+<!-- 
+Please read.
+
+Before posting, please be sure to review whether your issue matches the description of a known regression by using [this](https://github.com/Azure/azure-functions-durable-python/issues?q=is%3Aissue+is%3Aopen+label%3Aknown-regression) query.
+
+If your issue corresponds to a known regression, please try the suggested workarounds listed in the issue before filing an issue. Thanks!
+-->
 🐛 **Describe the bug**
 > A clear and concise description of what the bug is. 
 

.github/workflows/submodule-sync.yml

@@ -0,0 +1,39 @@
+name: Sync submodule pipeline
+
+on:
+  push:
+    branches: [ submodule ]
+jobs:
+  build:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v3
+        with:
+          repository: azure-functions-python-library
+          submodules: true
+      - id: Go to submodule
+        run: |
+          cd azure/functions/durable
+          git submodule update --remote --merge
+          git add .
+      - name: Create Pull Request
+        id: createPullRequest
+        uses: peter-evans/create-pull-request@v4
+        with:
+          commit-message: Update durable submodule
+          committer: GitHub <[email protected]>
+          branch: submodule-sync
+          delete-branch: true
+          title: 'Update durable submodule'
+          body: |
+            Updated submodule
+
+            [1]: https://github.com/peter-evans/create-pull-request
+          labels: |
+            required submodule update
+            automated pr
+          reviewers: vameru
+      - name: Check outputs
+        run: |
+            echo "Pull Request Number - ${{ steps.createPullRequest.outputs.pull-request-number }}"
+            echo "Pull Request URL - ${{ steps.createPullRequest.outputs.pull-request-url }}"

MANIFEST.in

@@ -0,0 +1 @@
+include ./_manifest/*

azure-pipelines-release.yml

@@ -0,0 +1,29 @@
+trigger: none
+pr: none
+
+resources:
+    pipelines:
+        - pipeline: DurablePyCI
+          project: "Azure Functions"
+          source: Azure.azure-functions-durable-python
+          branch: main
+
+jobs:
+    - job: Release
+      pool:
+        name: "1ES-Hosted-AzFunc"
+        demands:
+            - ImageOverride -equals MMSUbuntu20.04TLS
+      steps:
+          - task: UsePythonVersion@0
+            inputs:
+                versionSpec: '3.6'
+          - download: DurablePyCI
+          - script: "rm -r ./azure_functions_durable/_manifest"
+            displayName: 'Remove _manifest folder'
+            workingDirectory: "$(Pipeline.Workspace)/DurablePyCI"
+          - script: 'pip install twine'
+            displayName: "Install twine"
+          - script: "twine upload azure_functions_durable/*"
+            displayName: "Publish to PyPI"
+            workingDirectory: "$(Pipeline.Workspace)/DurablePyCI"

azure-pipelines.yml

@@ -13,7 +13,6 @@ trigger:
     - v*
 
 variables:     
-  agentPool: 'ubuntu-latest' # 'Ubuntu-16.04'
   python.version: '3.6'
   baseFolder: .
   componentArtifactName: 'azure_functions_durable'
@@ -28,7 +27,9 @@ stages:
   - job: Build_Durable_Functions
     displayName: Build Python Package
     pool: 
-      vmImage: $(agentPool)
+      name: "1ES-Hosted-AzFunc"
+      demands:
+          - ImageOverride -equals MMSUbuntu20.04TLS
     steps:
     - task: UsePythonVersion@0
       inputs:
@@ -49,7 +50,12 @@ stages:
         pip install pytest pytest-azurepipelines
         pytest
       displayName: 'pytest'
-
+    - task: ManifestGeneratorTask@0
+      displayName: "SBOM Generation Task"
+      inputs:
+          BuildComponentPath: '$(baseFolder)'
+          BuildDropPath: $(baseFolder)
+          Verbosity: "Information"
     - script: |
         python setup.py sdist bdist_wheel
       workingDirectory: $(baseFolder) 

azure/durable_functions/models/Task.py

@@ -347,6 +347,7 @@ def __init__(self, child: TaskBase, retry_options: RetryOptions, context):
         self.context = context
         self.actions = child.action_repr
         self.is_waiting_on_timer = False
+        self.error = None
 
     @property
     def id_(self):
@@ -373,10 +374,21 @@ def try_set_value(self, child: TaskBase):
         if self.is_waiting_on_timer:
             # timer fired, re-scheduling original task
             self.is_waiting_on_timer = False
-            rescheduled_task = self.context._generate_task(
-                action=NoOpAction("rescheduled task"), parent=self)
-            self.pending_tasks.add(rescheduled_task)
-            self.context._add_to_open_tasks(rescheduled_task)
+            # As per DTFx semantics: we need to check the number of retires only after the final
+            # timer has fired. This means we essentially have to wait for one "extra" timer after
+            # the maximum number of attempts has been reached. Removing this extra timer will cause
+            # stuck orchestrators as we need to be "in sync" with the replay logic of DTFx.
+            if self.num_attempts >= self.retry_options.max_number_of_attempts:
+                self.is_waiting_on_timer = True
+                # we have reached the maximum number of attempts, set error
+                self.set_value(is_error=True, value=self.error)
+            else:
+                rescheduled_task = self.context._generate_task(
+                    action=NoOpAction("rescheduled task"), parent=self)
+                self.pending_tasks.add(rescheduled_task)
+                self.context._add_to_open_tasks(rescheduled_task)
+                self.num_attempts += 1
+
             return
         if child.state is TaskState.SUCCEEDED:
             if len(self.pending_tasks) == 0:
@@ -386,17 +398,11 @@ def try_set_value(self, child: TaskBase):
                 self.set_value(is_error=False, value=child.result)
 
         else:  # child.state is TaskState.FAILED:
-            if self.num_attempts >= self.retry_options.max_number_of_attempts:
-                # we have reached the maximum number of attempts, set error
-                self.set_value(is_error=True, value=child.result)
-            else:
-                # still have some retries left.
-                # increase size of pending tasks by adding a timer task
-                # when it completes, we'll retry the original task
-                timer_task = self.context._generate_task(
-                    action=NoOpAction("-WithRetry timer"), parent=self)
-                self.pending_tasks.add(timer_task)
-                self.context._add_to_open_tasks(timer_task)
-                self.is_waiting_on_timer = True
-
-            self.num_attempts += 1
+            # increase size of pending tasks by adding a timer task
+            # when it completes, we'll retry the original task
+            timer_task = self.context._generate_task(
+                action=NoOpAction("-WithRetry timer"), parent=self)
+            self.pending_tasks.add(timer_task)
+            self.context._add_to_open_tasks(timer_task)
+            self.is_waiting_on_timer = True
+            self.error = child.result

azure/durable_functions/models/actions/CompoundAction.py

@@ -12,8 +12,8 @@ class CompoundAction(Action):
     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))
+    def __init__(self, compound_tasks: List[Action]):
+        self.compound_tasks = compound_tasks
 
     @property
     @abstractmethod
@@ -31,5 +31,5 @@ def to_json(self) -> Dict[str, Union[str, int]]:
         """
         json_dict: Dict[str, Union[str, int]] = {}
         add_attrib(json_dict, self, 'action_type', 'actionType')
-        add_attrib(json_dict, self, 'compound_actions', 'compoundActions')
+        json_dict['compoundActions'] = list(map(lambda x: x.to_json(), self.compound_tasks))
         return json_dict

setup.py

@@ -4,8 +4,8 @@
 import shutil
 import subprocess
 import sys
-import glob
 
+from glob import glob
 from setuptools import setup, find_packages
 from distutils.command import build
 
@@ -65,6 +65,9 @@ def run(self, *args, **kwargs):
         'pytest-asyncio==0.10.0'
     ],
     include_package_data=True,
+    data_files= [
+        ('_manifest', list(filter(os.path.isfile, glob('_manifest/**/*', recursive=True)))),
+    ],
     cmdclass={
         'build': BuildModule
     },

tests/orchestrator/test_create_timer.py

@@ -1,4 +1,5 @@
 from azure.durable_functions.models.ReplaySchema import ReplaySchema
+from azure.durable_functions.models.actions.WhenAnyAction import WhenAnyAction
 from tests.test_utils.ContextBuilder import ContextBuilder
 from .orchestrator_test_utils \
     import get_orchestration_state_result, assert_orchestration_state_equals, assert_valid_schema
@@ -9,7 +10,7 @@
 
 
 def base_expected_state(output=None, replay_schema: ReplaySchema = ReplaySchema.V1) -> OrchestratorState:
-    return OrchestratorState(is_done=False, actions=[], output=output, replay_schema=replay_schema.value)
+    return OrchestratorState(is_done=False, actions=[], output=output, replay_schema=replay_schema)
 
 def add_timer_fired_events(context_builder: ContextBuilder, id_: int, timestamp: str):
     fire_at: str = context_builder.add_timer_created_event(id_, timestamp)
@@ -99,4 +100,28 @@ def test_timers_can_be_cancelled():
     expected = expected_state.to_json()
 
     assert_orchestration_state_equals(expected, result)
-    assert result["actions"][0][1]["isCanceled"]
+    assert result["actions"][0][1]["isCanceled"]
+
+def test_timers_can_be_cancelled_replayV2():
+
+    context_builder = ContextBuilder("test_timers_can_be_cancelled", replay_schema=ReplaySchema.V2)
+    fire_at1 = context_builder.current_datetime + timedelta(minutes=5)
+    fire_at2 = context_builder.current_datetime + timedelta(minutes=10)
+    add_timer_fired_events(context_builder, 0, str(fire_at1))
+    add_timer_fired_events(context_builder, 1, str(fire_at2))
+
+    result = get_orchestration_state_result(
+        context_builder, generator_function_timer_can_be_cancelled)
+
+    expected_state = base_expected_state(output='Done!', replay_schema=ReplaySchema.V2)
+    expected_state._actions = [
+        [WhenAnyAction(
+            [CreateTimerAction(fire_at=fire_at1), CreateTimerAction(fire_at=fire_at2, is_cancelled=True)]
+        )]
+    ]
+
+    expected_state._is_done = True
+    expected = expected_state.to_json()
+
+    assert_orchestration_state_equals(expected, result)
+    assert result["actions"][0][0]['compoundActions'][1]["isCanceled"]

tests/orchestrator/test_sequential_orchestrator_with_retry.py

@@ -29,6 +29,19 @@ def generator_function(context):
 
     return outputs
 
+def generator_function_try_catch(context):
+    outputs = []
+
+    retry_options = RETRY_OPTIONS
+    result = None
+    try:
+        result = yield context.call_activity_with_retry(
+        "Hello", retry_options, "Tokyo")
+    except:
+        result = yield context.call_activity_with_retry(
+        "Hello",  retry_options, "Seattle")
+    return result
+
 def generator_function_concurrent_retries(context):
     outputs = []
 
@@ -305,6 +318,37 @@ def test_failed_tokyo_hit_max_attempts():
         expected_error_str = f"{error_msg}{error_label}{state_str}"
         assert expected_error_str == error_str
 
+def test_failed_tokyo_hit_max_attempts_in_try_catch():
+    # This test ensures that APIs can still be invoked after a failed CallActivityWithRetry invocation
+    failed_reason = 'Reasons'
+    failed_details = 'Stuff and Things'
+    context_builder = ContextBuilder('test_simple_function')
+    
+    # events for first task: "Hello Tokyo"
+    add_hello_failed_events(context_builder, 0, failed_reason, failed_details)
+    add_retry_timer_events(context_builder, 1)
+    add_hello_failed_events(context_builder, 2, failed_reason, failed_details)
+    add_retry_timer_events(context_builder, 3)
+    add_hello_failed_events(context_builder, 4, failed_reason, failed_details)
+    # we have an "extra" timer to wait for, due to legacy behavior in DTFx.
+    add_retry_timer_events(context_builder, 5)
+
+    # events to task in except block
+    add_hello_completed_events(context_builder, 6, "\"Hello Seattle!\"")
+
+    result = get_orchestration_state_result(
+        context_builder, generator_function_try_catch)
+
+    expected_state = base_expected_state()
+    add_hello_action(expected_state, 'Tokyo')
+    add_hello_action(expected_state, 'Seattle')
+    expected_state._output = "Hello Seattle!"
+    expected_state._is_done = True
+    expected = expected_state.to_json()
+
+    assert_valid_schema(result)
+    assert_orchestration_state_equals(expected, result)
+
 def test_concurrent_retriable_results():
     failed_reason = 'Reasons'
     failed_details = 'Stuff and Things'