def _refresh_task_state(task_ex_id):
    with db_api.transaction():
        task_ex = db_api.load_task_execution(task_ex_id)

        if not task_ex:
            return

        if (states.is_completed(task_ex.state)
                or task_ex.state == states.RUNNING):
            return

        wf_ex = task_ex.workflow_execution

        if states.is_completed(wf_ex.state):
            return

        wf_spec = spec_parser.get_workflow_spec_by_execution_id(
            task_ex.workflow_execution_id
        )

        wf_ctrl = wf_base.get_controller(wf_ex, wf_spec)

        with db_api.named_lock(task_ex.id):
            # NOTE: we have to use this lock to prevent two (or more) such
            # methods from changing task state and starting its action or
            # workflow. Checking task state outside of this section is a
            # performance optimization because locking is pretty expensive.
            db_api.refresh(task_ex)

            if (states.is_completed(task_ex.state)
                    or task_ex.state == states.RUNNING):
                return

            log_state = wf_ctrl.get_logical_task_state(task_ex)

            state = log_state.state
            state_info = log_state.state_info

            # Update 'triggered_by' because it could have changed.
            task_ex.runtime_context['triggered_by'] = log_state.triggered_by

            if state == states.RUNNING:
                continue_task(task_ex)
            elif state == states.ERROR:
                complete_task(task_ex, state, state_info)
            elif state == states.WAITING:
                LOG.info(
                    "Task execution is still in WAITING state"
                    " [task_ex_id=%s, task_name=%s]",
                    task_ex_id,
                    task_ex.name
                )
            else:
                # Must never get here.
                raise RuntimeError(
                    'Unexpected logical task state [task_ex_id=%s, '
                    'task_name=%s, state=%s]' %
                    (task_ex_id, task_ex.name, state)
                )

    def _dispatch_workflow_commands(self, wf_ex, wf_cmds):
        if not wf_cmds:
            return

        for cmd in wf_cmds:
            if isinstance(cmd, commands.RunTask) and cmd.is_waiting():
                task_handler.defer_task(cmd)
            elif isinstance(cmd, commands.RunTask):
                task_handler.run_new_task(cmd)
            elif isinstance(cmd, commands.RunExistingTask):
                task_handler.run_existing_task(
                    cmd.task_ex.id,
                    reset=cmd.reset
                )
            elif isinstance(cmd, commands.SetWorkflowState):
                if states.is_completed(cmd.new_state):
                    self._stop_workflow(cmd.wf_ex, cmd.new_state, cmd.msg)
                else:
                    wf_handler.set_execution_state(wf_ex, cmd.new_state)
            elif isinstance(cmd, commands.Noop):
                # Do nothing.
                pass
            else:
                raise RuntimeError('Unsupported workflow command: %s' % cmd)

            if wf_ex.state != states.RUNNING:
                break

    def delete(self, id):
        """Delete the specified action_execution.

        :param id: UUID of action execution to delete
        """
        acl.enforce('action_executions:delete', context.ctx())

        LOG.debug("Delete action_execution [id=%s]", id)

        if not cfg.CONF.api.allow_action_execution_deletion:
            raise exc.NotAllowedException("Action execution deletion is not "
                                          "allowed.")

        with db_api.transaction():
            action_ex = db_api.get_action_execution(id)

            if action_ex.task_execution_id:
                raise exc.NotAllowedException(
                    "Only ad-hoc action execution can be deleted."
                )

            if not states.is_completed(action_ex.state):
                raise exc.NotAllowedException(
                    "Only completed action execution can be deleted."
                )

            return db_api.delete_action_execution(id)

    def _find_next_task_names(self, task_ex):
        t_state = task_ex.state
        t_name = task_ex.name

        ctx = data_flow.evaluate_task_outbound_context(task_ex)

        t_names = []

        if states.is_completed(t_state):
            t_names += self._find_next_task_names_for_clause(
                self.wf_spec.get_on_complete_clause(t_name),
                ctx
            )

        if t_state == states.ERROR:
            t_names += self._find_next_task_names_for_clause(
                self.wf_spec.get_on_error_clause(t_name),
                ctx
            )

        elif t_state == states.SUCCESS:
            t_names += self._find_next_task_names_for_clause(
                self.wf_spec.get_on_success_clause(t_name),
                ctx
            )

        return t_names

    def delete(self, id, force=False):
        """Delete the specified Execution.

        :param id: UUID of execution to delete.
        :param force: Optional. Force the deletion of unfinished executions.
                      Default: false. While the api is backward compatible
                      the behaviour is not the same. The new default is the
                      safer option
        """
        acl.enforce('executions:delete', context.ctx())

        LOG.debug("Delete execution [id=%s]", id)

        if not force:
            state = db_api.get_workflow_execution(
                id,
                fields=(db_models.WorkflowExecution.state,)
            )[0]

            if not states.is_completed(state):
                raise exc.NotAllowedException(
                    "Only completed executions can be deleted. "
                    "Use --force to override this. "
                    "Execution {} is in {} state".format(id, state)
                )

        return rest_utils.rest_retry_on_db_error(
            db_api.delete_workflow_execution
        )(id)

def dispatch_workflow_commands(wf_ex, wf_cmds):
    # TODO(rakhmerov): I don't like these imports but otherwise we have
    # import cycles.
    from mistral.engine import task_handler
    from mistral.engine import workflow_handler as wf_handler

    if not wf_cmds:
        return

    for cmd in wf_cmds:
        if isinstance(cmd, (commands.RunTask, commands.RunExistingTask)):
            task_handler.run_task(cmd)
        elif isinstance(cmd, commands.SetWorkflowState):
            # TODO(rakhmerov): Make just a single call to workflow_handler
            if states.is_completed(cmd.new_state):
                wf_handler.stop_workflow(cmd.wf_ex, cmd.new_state, cmd.msg)
            else:
                wf_handler.set_workflow_state(wf_ex, cmd.new_state, cmd.msg)
        elif isinstance(cmd, commands.Noop):
            # Do nothing.
            pass
        else:
            raise exc.MistralError('Unsupported workflow command: %s' % cmd)

        if wf_ex.state != states.RUNNING:
            break

    def _continue_workflow(self, task_ex=None, reset=True, env=None):
        wf_ctrl = wf_base.get_controller(self.wf_ex)

        # Calculate commands to process next.
        cmds = wf_ctrl.continue_workflow(task_ex=task_ex, reset=reset, env=env)

        # When resuming a workflow we need to ignore all 'pause'
        # commands because workflow controller takes tasks that
        # completed within the period when the workflow was paused.
        cmds = list(
            filter(lambda c: not isinstance(c, commands.PauseWorkflow), cmds)
        )

        # Since there's no explicit task causing the operation
        # we need to mark all not processed tasks as processed
        # because workflow controller takes only completed tasks
        # with flag 'processed' equal to False.
        for t_ex in self.wf_ex.task_executions:
            if states.is_completed(t_ex.state) and not t_ex.processed:
                t_ex.processed = True

        dispatcher.dispatch_workflow_commands(self.wf_ex, cmds)

        if not cmds:
            self._check_and_complete()

    def _get_induced_join_state(self, in_task_spec, in_task_ex,
                                join_task_spec, t_execs_cache):
        join_task_name = join_task_spec.get_name()

        if not in_task_ex:
            possible, depth = self._possible_route(
                in_task_spec,
                t_execs_cache
            )

            if possible:
                return states.WAITING, depth, None
            else:
                return states.ERROR, depth, 'impossible route'

        if not states.is_completed(in_task_ex.state):
            return states.WAITING, 1, None

        if self._is_conditional_transition(in_task_ex, in_task_spec) and \
                not hasattr(in_task_ex, "in_context"):
            in_task_ex = db_api.get_task_execution(in_task_ex.id)

        # [(task name, params, event name), ...]
        next_tasks_tuples = self._find_next_tasks(in_task_ex)

        next_tasks_dict = {tup[0]: tup[2] for tup in next_tasks_tuples}

        if join_task_name not in next_tasks_dict:
            return states.ERROR, 1, "not triggered"

        return states.RUNNING, 1, next_tasks_dict[join_task_name]

    def _find_next_task_names(self, task_ex, ctx):
        t_state = task_ex.state
        t_name = task_ex.name

        t_names = []

        if states.is_completed(t_state):
            t_names += self._find_next_task_names_for_clause(
                self.get_on_complete_clause(t_name),
                ctx
            )

        if t_state == states.ERROR:
            t_names += self._find_next_task_names_for_clause(
                self.get_on_error_clause(t_name),
                ctx
            )

        elif t_state == states.SUCCESS:
            t_names += self._find_next_task_names_for_clause(
                self.get_on_success_clause(t_name),
                ctx
            )

        return t_names

    def _continue_workflow(self, wf_ex, task_ex=None, reset=True):
        wf_handler.set_execution_state(wf_ex, states.RUNNING)

        wf_ctrl = wf_base.WorkflowController.get_controller(wf_ex)

        # Calculate commands to process next.
        cmds = wf_ctrl.continue_workflow(task_ex=task_ex, reset=reset)

        # When resuming a workflow we need to ignore all 'pause'
        # commands because workflow controller takes tasks that
        # completed within the period when the workflow was pause.
        cmds = filter(
            lambda c: not isinstance(c, commands.PauseWorkflow),
            cmds
        )

        # Since there's no explicit task causing the operation
        # we need to mark all not processed tasks as processed
        # because workflow controller takes only completed tasks
        # with flag 'processed' equal to False.
        for t_ex in wf_ex.task_executions:
            if states.is_completed(t_ex.state) and not t_ex.processed:
                t_ex.processed = True

        self._dispatch_workflow_commands(wf_ex, cmds)

        if not cmds:
            if not wf_utils.find_incomplete_task_executions(wf_ex):
                wf_handler.succeed_workflow(
                    wf_ex,
                    wf_ctrl.evaluate_workflow_final_context()
                )

        return wf_ex.get_clone()

def find_task_executions_by_name(wf_ex_id, task_name):
    """Finds task executions by workflow execution id and task name.

    :param wf_ex_id: Workflow execution id.
    :param task_name: Task name.
    :return: Task executions (possibly a cached value).
    """
    cache_key = (wf_ex_id, task_name)

    with _TASK_EXECUTIONS_CACHE_LOCK:
        t_execs = _TASK_EXECUTIONS_CACHE.get(cache_key)

    if t_execs:
        return t_execs

    t_execs = db_api.get_task_executions(
        workflow_execution_id=wf_ex_id,
        name=task_name
    )

    # We can cache only finished tasks because they won't change.
    all_finished = (
        t_execs and
        all([states.is_completed(t_ex.state) for t_ex in t_execs])
    )

    if all_finished:
        with _TASK_EXECUTIONS_CACHE_LOCK:
            _TASK_EXECUTIONS_CACHE[cache_key] = t_execs

    return t_execs

    def after_task_complete(self, task_ex, task_spec):
        """Possible Cases:

        1. state = SUCCESS
           if continue_on is not specified,
           no need to move to next iteration;
           if current:count achieve retry:count then policy
           breaks the loop (regardless on continue-on condition);
           otherwise - check continue_on condition and if
           it is True - schedule the next iteration,
           otherwise policy breaks the loop.
        2. retry:count = 5, current:count = 2, state = ERROR,
           state = IDLE/DELAYED, current:count = 3
        3. retry:count = 5, current:count = 4, state = ERROR
        Iterations complete therefore state = #{state}, current:count = 4.
        """
        super(RetryPolicy, self).after_task_complete(task_ex, task_spec)

        context_key = "retry_task_policy"

        runtime_context = _ensure_context_has_key(task_ex.runtime_context, context_key)

        continue_on_evaluation = expressions.evaluate(
            self._continue_on_clause, data_flow.evaluate_task_outbound_context(task_ex)
        )

        task_ex.runtime_context = runtime_context

        state = task_ex.state

        if not states.is_completed(state):
            return

        policy_context = runtime_context[context_key]

        retry_no = 0

        if "retry_no" in policy_context:
            retry_no = policy_context["retry_no"]
            del policy_context["retry_no"]

        retries_remain = retry_no + 1 < self.count

        stop_continue_flag = task_ex.state == states.SUCCESS and not self._continue_on_clause
        stop_continue_flag = stop_continue_flag or (self._continue_on_clause and not continue_on_evaluation)
        break_triggered = task_ex.state == states.ERROR and self.break_on

        if not retries_remain or break_triggered or stop_continue_flag:
            return

        _log_task_delay(task_ex, self.delay)

        data_flow.invalidate_task_execution_result(task_ex)
        task_ex.state = states.DELAYED

        policy_context["retry_no"] = retry_no + 1
        runtime_context[context_key] = policy_context

        scheduler.schedule_call(None, _RUN_EXISTING_TASK_PATH, self.delay, task_ex_id=task_ex.id)

def _get_unaccepted_act_exs(task_ex):
    # Choose only if not accepted but completed.
    return list(
        filter(
            lambda x: not x.accepted and states.is_completed(x.state),
            task_ex.executions
        )
    )

    def _is_upstream_task_execution(self, t_spec, t_ex_candidate):
        if not states.is_completed(t_ex_candidate.state):
            return False

        if not t_spec.get_join():
            return not t_ex_candidate.processed

        return self._triggers_join(t_spec, self.wf_spec.get_tasks()[t_ex_candidate.name])

def set_workflow_state(wf_ex, state, msg=None):
    if states.is_completed(state):
        stop_workflow(wf_ex, state, msg)
    elif states.is_paused(state):
        pause_workflow(wf_ex, msg)
    else:
        raise exc.MistralError(
            'Invalid workflow state [wf_ex=%s, state=%s]' % (wf_ex, state)
        )

def _fail_task_if_incomplete(task_ex_id, timeout):
    from mistral.engine import task_handler

    with db_api.transaction():
        task_ex = db_api.get_task_execution(task_ex_id)

        if not states.is_completed(task_ex.state):
            msg = 'Task timed out [timeout(s)=%s].' % timeout

            task_handler.complete_task(task_ex, states.ERROR, msg)

    def _triggers_join(self, join_task_spec, inbound_task_spec):
        in_t_ex = wf_utils.find_task_execution(self.wf_ex, inbound_task_spec)

        if not in_t_ex or not states.is_completed(in_t_ex.state):
            return False

        return filter(
            lambda t_name: join_task_spec.get_name() == t_name,
            self._find_next_task_names(in_t_ex, data_flow.evaluate_task_outbound_context(in_t_ex)),
        )

def set_workflow_state(wf_ex, state, msg=None):
    if states.is_completed(state):
        stop_workflow(wf_ex, state, msg)
    elif states.is_paused(state):
        pause_workflow(wf_ex, msg)
    else:
        raise exc.MistralError(
            'Invalid workflow execution state [wf_ex_id=%s, wf_name=%s, '
            'state=%s]' % (wf_ex.id, wf_ex.name, state)
        )

    def set_state(self, state, state_info=None):
        assert self.wf_ex

        cur_state = self.wf_ex.state

        if states.is_valid_transition(cur_state, state):
            wf_ex = db_api.update_workflow_execution_state(
                id=self.wf_ex.id,
                cur_state=cur_state,
                state=state
            )

            if wf_ex is None:
                # Do nothing because the state was updated previously.
                return False

            self.wf_ex = wf_ex
            self.wf_ex.state_info = json.dumps(state_info) \
                if isinstance(state_info, dict) else state_info

            wf_trace.info(
                self.wf_ex,
                "Workflow '%s' [%s -> %s, msg=%s]" %
                (self.wf_ex.workflow_name,
                 cur_state,
                 state,
                 self.wf_ex.state_info)
            )
        else:
            msg = ("Can't change workflow execution state from %s to %s. "
                   "[workflow=%s, execution_id=%s]" %
                   (cur_state, state, self.wf_ex.name, self.wf_ex.id))

            raise exc.WorkflowException(msg)

        # Workflow result should be accepted by parent workflows (if any)
        # only if it completed successfully or failed.
        self.wf_ex.accepted = states.is_completed(state)

        if states.is_completed(state):
            triggers.on_workflow_complete(self.wf_ex)

        return True

def fail_task_if_incomplete(task_ex_id, timeout):
    task_ex = db_api.get_task_execution(task_ex_id)

    if not states.is_completed(task_ex.state):
        msg = "Task timed out [id=%s, timeout(s)=%s]." % (task_ex_id, timeout)

        wf_trace.info(task_ex, msg)

        wf_trace.info(task_ex, "Task '%s' [%s -> ERROR]" % (task_ex.name, task_ex.state))

        rpc.get_engine_client().on_task_state_change(task_ex_id, states.ERROR)