Fix task_any bug (Azure#538)

* Fix long timer bug
* Also fix task_any scheduling issue

Author: andystaples

azure/durable_functions/models/DurableOrchestrationContext.py

@@ -722,7 +722,8 @@ def _add_to_open_tasks(self, task: TaskBase):
                 task.id = self._sequence_number
                 self._sequence_number += 1
                 self.open_tasks[task.id] = task
-            elif task.id != -1:
+            elif task.id != -1 and self.open_tasks[task.id] != task:
+                # Case when returning task_any with multiple external events having the same ID
                 self.open_tasks[task.id].append(task)
 
             if task.id in self.deferred_tasks:

azure/durable_functions/models/TaskOrchestrationExecutor.py

@@ -245,14 +245,14 @@ def resume_user_code(self):
 
         self.current_task = new_task
         if not (new_task is None):
+            if not (self.current_task._is_scheduled):
+                # new task is received. it needs to be resolved to a value
+                self.context._add_to_actions(self.current_task.action_repr)
+                self._mark_as_scheduled(self.current_task)
             if not (new_task.state is TaskState.RUNNING):
                 # user yielded the same task multiple times, continue executing code
                 # until a new/not-previously-yielded task is encountered
                 self.resume_user_code()
-            elif not (self.current_task._is_scheduled):
-                # new task is received. it needs to be resolved to a value
-                self.context._add_to_actions(self.current_task.action_repr)
-                self._mark_as_scheduled(self.current_task)
 
     def _mark_as_scheduled(self, task: TaskBase):
         if isinstance(task, CompoundTask):