executor.py

from __future__ import annotations

import copy
import hashlib
import inspect
import json
import re
import traceback
from typing import TYPE_CHECKING, Any, Callable

import multiprocess

import simpleflow.task as base_task
import simpleflow.swf.mapper.exceptions
import simpleflow.swf.mapper.models
import simpleflow.swf.mapper.models.decision
from simpleflow import exceptions, executor, format, futures, logger, task
from simpleflow.activity import PRIORITY_NOT_SET, Activity
from simpleflow.base import Submittable
from simpleflow.history import History
from simpleflow.marker import Marker
from simpleflow.signal import WaitForSignal
from simpleflow.swf import constants
from simpleflow.swf.helpers import swf_identity
from simpleflow.swf.task import (
    ActivityTask,
    CancelTimerTask,
    ContinueAsNewWorkflowTask,
    MarkerTask,
    SignalTask,
    SwfTask,
    TimerTask,
    WorkflowTask,
)
from simpleflow.swf.utils import DecisionsAndContext
from simpleflow.utils import hex_hash, issubclass_, json_dumps, retry
from simpleflow.workflow import Workflow
from simpleflow.swf.mapper.core import ConnectedSWFObject

if TYPE_CHECKING:
    from simpleflow.swf.mapper.models.domain import Domain

__all__ = ["Executor"]


# if "poll_for_activity_task" doesn't contain a "taskToken"
# key, then retry ; it happens (not often) that the decider
# doesn't get the scheduled task while it should...
@retry.with_delay(nb_times=3, delay=retry.exponential, on_exceptions=KeyError)
def run_fake_activity_task(domain, task_list, result):
    obj = ConnectedSWFObject()
    resp = obj.poll_for_activity_task(
        domain,
        task_list,
        identity=swf_identity(),
    )
    obj.respond_activity_task_completed(
        resp["taskToken"],
        result,
    )


# Same retry condition as run_fake_activity_task
@retry.with_delay(nb_times=3, delay=retry.exponential, on_exceptions=KeyError)
def run_fake_child_workflow_task(domain, task_list, result=None):
    obj = ConnectedSWFObject()
    resp = obj.poll_for_decision_task(
        domain,
        task_list,
        identity=swf_identity(),
    )
    obj.respond_decision_task_completed(
        resp["taskToken"],
        decisions=[
            {
                "decisionType": "CompleteWorkflowExecution",
                "completeWorkflowExecutionDecisionAttributes": {
                    "result": result,
                },
            }
        ],
    )


def run_fake_task_worker(domain, task_list, former_event):
    if former_event["type"] == "activity":
        worker_proc = multiprocess.Process(
            target=run_fake_activity_task,
            args=(
                domain,
                task_list,
                former_event["result"],
            ),
        )
    elif former_event["type"] == "child_workflow":
        worker_proc = multiprocess.Process(
            target=run_fake_child_workflow_task,
            args=(
                domain,
                task_list,
            ),
            kwargs={
                "result": former_event["result"],
            },
        )
    else:
        raise Exception(f"Wrong event type {former_event['type']}")

    worker_proc.start()


class TaskRegistry(dict):
    """This registry tracks tasks and assign them an integer identifier."""

    def add(self, a_task):
        """
        ID's are assigned sequentially by incrementing an integer. They start
        from 1.

        :type a_task: ActivityTask | WorkflowTask
        :returns:
            :rtype: int.

        """
        name = a_task.name
        self[name] = self.setdefault(name, 0) + 1

        return self[name]


class Executor(executor.Executor):
    """
    Manage a workflow's execution with Amazon SWF. It replays the workflow's
    definition from the start until it blocks (i.e. raises
    :py:class:`exceptions.ExecutionBlocked`).

    SWF stores the history of all events that occurred in the workflow and
    passes it to the executor. Only one executor handles a workflow at a time.
    It means the history is consistent and there is no concurrent modifications
    on the execution of the workflow.
    """

    def __init__(
        self,
        domain: Domain,
        workflow_class: type[Workflow],
        task_list: str | None = None,
        repair_with: History | None = None,
        force_activities: str | None = None,
        repair_workflow_id: str | None = None,
        repair_run_id: str | None = None,
    ):
        super().__init__(workflow_class)
        self._history: History | None = None
        self._run_context = {}
        self.domain = domain
        self.task_list = task_list
        self.repair_with = repair_with
        self._repair_workflow_id = repair_workflow_id
        self._repair_run_id = repair_run_id
        if force_activities:
            self.force_activities = re.compile(force_activities)
        else:
            self.force_activities = None
        self._open_activity_count = 0
        self._decisions_and_context = DecisionsAndContext()
        self._append_timer = False  # Append an immediate timer decision
        self._tasks = TaskRegistry()
        self._idempotent_tasks_to_submit = set()
        self._execution = None
        self.current_priority = None
        self.handled_failures = {}
        self.created_activity_types = set()

    def reset(self):
        """
        Clears the state of the execution.

        It is required to ensure the id of the tasks are assigned the same way
        on each replay.

        """
        self._open_activity_count = 0
        self._decisions_and_context = DecisionsAndContext()
        self._append_timer = False  # Append an immediate timer decision
        self._tasks = TaskRegistry()
        self._idempotent_tasks_to_submit = set()
        self._execution = None
        self.current_priority = None
        self.handled_failures = {}
        self.created_activity_types = set()
        self.create_workflow()

    @property
    def history(self) -> History | None:
        return self._history

    def _make_task_id(self, a_task: ActivityTask | WorkflowTask, workflow_id: str, run_id: str, *args, **kwargs) -> str:
        """
        Assign a new ID to *a_task*.

        :returns:
            String with at most 256 characters.
        """
        if isinstance(a_task, ActivityTask) and hasattr(a_task.activity.callable, "get_task_id"):
            suffix = a_task.activity.callable.get_task_id(self.workflow, *args, **kwargs)
        elif not a_task.idempotent:
            # If idempotency is False or unknown, let's generate a task id by
            # incrementing an id after the a_task name.
            # (default strategy, backwards compatible with previous versions)
            suffix = self._tasks.add(a_task)
        else:
            # If a_task is idempotent, we can do better and hash arguments.
            # It makes the workflow resistant to retries or variations on the
            # same task name (see #11).
            arguments = json_dumps({"args": args, "kwargs": kwargs})
            suffix = hashlib.md5(arguments.encode("utf-8")).hexdigest()  # nosec

        if isinstance(a_task, WorkflowTask):
            # Some task types must have globally unique names.
            suffix = f"{workflow_id}--{hex_hash(run_id)}--{suffix}"

        task_id = f"{a_task.name}-{suffix}"
        if len(task_id) > 256:  # Better safe than sorry...
            task_id = task_id[0:223] + "-" + hashlib.md5(task_id.encode("utf-8")).hexdigest()  # nosec
        return task_id

    def _get_future_from_activity_event(self, event: dict[str, Any]) -> futures.Future | None:
        """Maps an activity event to a Future with the corresponding state.

        :param event: activity event
        """
        future = futures.Future()  # state is PENDING.
        state = event["state"]

        if state == "scheduled":
            pass
        elif state == "schedule_failed":
            name = event["activity_type"]["name"]
            version = event["activity_type"]["version"]
            if event["cause"] == "ACTIVITY_TYPE_DOES_NOT_EXIST" and (name, version) not in self.created_activity_types:
                self.created_activity_types.add((name, version))
                activity_type = simpleflow.swf.mapper.models.ActivityType(self.domain, name=name, version=version)
                logger.info(f"creating activity type {activity_type.name} in domain {self.domain.name}")
                try:
                    activity_type.save()
                except simpleflow.swf.mapper.exceptions.AlreadyExistsError:
                    logger.info(
                        f"oops: Activity type {activity_type.name} in domain {self.domain.name} already exists,"
                        f" creation failed, continuing..."
                    )
                return None
            logger.info(f"failed to schedule {name}: {event['cause']}")
            return None
        elif state == "started":
            future.set_running()
        elif state == "completed":
            result = event["result"]
            future.set_finished(format.decode(result))
        elif state == "canceled":
            future.set_cancelled()
        elif state == "failed":
            exception = exceptions.TaskFailed(name=event["id"], reason=event["reason"], details=event.get("details"))
            future.set_exception(exception)
        elif state == "timed_out":
            exception = exceptions.TimeoutError(event["timeout_type"], event["timeout_value"])
            future.set_exception(exception)
        else:
            logger.info(
                "unhandled state for activity %s: %s",
                event.get("name", f"#{event['id']}"),
                state,
            )

        return future

    def _get_future_from_child_workflow_event(self, event: dict[str, Any]) -> futures.Future | None:
        """Maps a child workflow event to a Future with the corresponding
        state.

        :param event: child workflow event
        :type  event: dict[str, Any]
        """
        future = futures.Future()
        state = event["state"]

        if state == "start_initiated":
            pass  # future._state = futures.PENDING
        elif state == "start_failed":
            if event["cause"] == "WORKFLOW_TYPE_DOES_NOT_EXIST":
                workflow_type = simpleflow.swf.mapper.models.WorkflowType(
                    self.domain,
                    name=event["name"],
                    version=event["version"],
                )
                logger.info(f"Creating workflow type {workflow_type.name} in domain {self.domain.name}")
                try:
                    workflow_type.save()
                except simpleflow.swf.mapper.exceptions.AlreadyExistsError:
                    # Could have be created by a concurrent workflow execution.
                    pass
                return None
            future.set_exception(
                exceptions.TaskFailed(
                    name=event["id"],
                    reason=event["cause"],
                    details=event.get("details"),
                )
            )
        elif state == "started":
            future.set_running()
        elif state == "completed":
            future.set_finished(format.decode(event["result"]))
        elif state == "failed":
            future.set_exception(
                exceptions.TaskFailed(
                    name=event["id"],
                    reason=event["reason"],
                    details=event.get("details"),
                )
            )
        elif state == "timed_out":
            future.set_exception(
                exceptions.TimeoutError(
                    event["timeout_type"],
                    None,
                )
            )
        elif state == "canceled":
            future.set_exception(
                exceptions.TaskCanceled(
                    event.get("details"),
                )
            )
        elif state == "terminated":
            future.set_exception(exceptions.TaskTerminated())
        else:
            logger.info(
                "unhandled state for workflow %s: %s",
                event.get("name", f"#{event['id']}"),
                state,
            )

        return future

    def _get_future_from_marker_event(self, a_task, event: dict[str, Any]) -> futures.Future:
        """Maps a marker event to a Future with the corresponding
        state.

        :param a_task: currently unused
        :param event: marker event
        """
        future = futures.Future()
        if not event:
            return future
        state = event["state"]
        if state == "recorded":
            future.set_finished(event["details"])
        elif state == "failed":
            future.set_exception(
                exceptions.TaskFailed(
                    name=event["name"],
                    reason=event["cause"],
                )
            )

        return future

    def get_future_from_signal_event(self, a_task: SignalTask | None, event: dict[str, Any]) -> futures.Future:
        """Maps a signal event to a Future with the corresponding
        state.
        a_task: currently unused
        """
        future = futures.Future()
        if not event:
            return future
        state = event["state"]
        if state == "signaled":
            future.set_finished(event["input"])

        return future

    def get_future_from_external_workflow_event(self, a_task: WorkflowTask, event: dict[str, Any]) -> futures.Future:
        """Maps an external workflow event to a Future with the corresponding
        state.

        :param a_task: currently unused
        :param event: external workflow event
        """
        future = futures.Future()
        if not event:
            return future
        state = event["state"]
        if state == "signal_execution_initiated":
            # Don't re-initiate signal sending
            future.set_running()
        elif state == "execution_signaled":
            future.set_finished(event["input"])
        elif state == "signal_execution_failed":
            future.set_exception(
                exceptions.TaskFailed(
                    name=event["name"],
                    reason=event["cause"],
                )
            )

        return future

    def _get_future_from_timer_event(self, a_task: TimerTask, event: dict[str, Any]) -> futures.Future:
        """
        Maps a timer event to a Future with the corresponding state.

        :param a_task: Timer task; unused.
        :param event: Timer event
        """
        future = futures.Future()
        if not event:
            return future
        state = event["state"]
        if state == "started":
            future.set_running()
        elif state == "fired":
            future.set_finished(None)
        elif state == "canceled":
            future.set_cancelled()
        elif state in ("start_failed", "cancel_failed"):
            future.set_exception(
                exceptions.TaskFailed(
                    name=event["timer_id"],
                    reason=event["cause"],
                )
            )

        return future

    def get_future_from_signal(self, signal_name: str) -> futures.Future:
        event = self._history.signals.get(signal_name)
        return self.get_future_from_signal_event(None, event)

    def find_activity_event(self, a_task: ActivityTask, history: History) -> dict[str, Any] | None:
        """
        Get the event corresponding to an activity task, if any.
        """
        activity = history.activities.get(a_task.id)
        return activity

    def find_child_workflow_event(self, a_task: WorkflowTask, history: History) -> dict[str, Any] | None:
        """
        Get the event corresponding to a child workflow, if any.
        """
        return history.child_workflows.get(a_task.id)

    def find_signal_event(self, a_task: SignalTask, history: History) -> dict[str, Any] | None:
        """
        Get the event corresponding to a signal, if any.
        """
        # FIXME could look directly in signaled_workflows?
        event = history.signals.get(a_task.name)
        if not event:
            if a_task.workflow_id is None:  # Broadcast, should be in signals
                return None
            signaled_workflows = history.signaled_workflows.get(a_task.name, [])
            for w in signaled_workflows:
                if w["workflow_id"] == a_task.workflow_id and (a_task.run_id is None or w["run_id"] == a_task.run_id):
                    event = w
                    break
        return event

    def find_marker_event(self, a_task: MarkerTask, history: History) -> dict[str, Any] | None:
        """
        Get the event corresponding to a marker, if any.
        """
        json_details = json_dumps(a_task.details) if a_task.details is not None else None
        marker_list = history.markers.get(a_task.name)
        if not marker_list:
            return None
        marker_list = [m for m in marker_list if m["state"] == "recorded" and m["details"] == json_details]
        return marker_list[-1] if marker_list else None

    def find_timer_event(self, a_task: TimerTask | CancelTimerTask, history: History) -> dict[str, Any] | None:
        """
        Get the event corresponding to a timer or timer cancellation, if any.
        """
        event = history.timers.get(a_task.id)
        if not event:
            return None
        if isinstance(a_task, CancelTimerTask):
            if "canceled_event_id" not in event and "cancel_failed_event_id" not in event:
                # Timer not yet cancelled: no future returned
                return None
        return event

    TASK_TYPE_TO_EVENT_FINDER: dict[type, callable] = {
        ActivityTask: find_activity_event,
        WorkflowTask: find_child_workflow_event,
        SignalTask: find_signal_event,
        MarkerTask: find_marker_event,
        TimerTask: find_timer_event,
        CancelTimerTask: find_timer_event,
    }

    def find_event(self, a_task: SwfTask, history: History) -> dict[str, Any] | None:
        """
        Get the event corresponding to a "task", if any.
        """
        for typ in inspect.getmro(type(a_task)):
            finder = self.TASK_TYPE_TO_EVENT_FINDER.get(typ)
            if finder:
                return finder(self, a_task, history)
        raise TypeError(f"invalid type {type(a_task)} for task {a_task}")

    def resume_activity(self, a_task: ActivityTask, event: dict[str, Any]) -> futures.Future | None:
        """
        Resume an activity task.
        """
        future = self._get_future_from_activity_event(event)
        if not future:  # schedule failed, maybe OK later.
            return None

        if not future.finished:  # Still pending or running...
            return future

        if future.exception is None:  # Result available!
            return future

        return self.handle_failure(event, future, a_task, exceptions.TaskException)

    def resume_child_workflow(self, a_workflow: WorkflowTask, event: dict[str, Any]) -> futures.Future | None:
        """
        Resume a child workflow.
        """
        future = self._get_future_from_child_workflow_event(event)

        if not future:
            # WORKFLOW_TYPE_DOES_NOT_EXIST, will be created
            return None

        if not future.finished:  # Still pending or running...
            return future

        if future.exception is None:  # Result available!
            return future

        return self.handle_failure(event, future, a_workflow, exceptions.WorkflowException)

    def handle_failure(
        self,
        event: dict[str, Any],
        future: futures.Future,
        swf_task: ActivityTask | WorkflowTask,
        exception_class: type[Exception],
    ) -> futures.Future | tuple[futures.Future | None, SwfTask] | None:
        """
        Call the workflow's on_task_failure method if it exists.
        If no retry/abort/ignore decision, use the default strategy (using retry count and raises_on_failure).

        on_task_failure can:
        * abort the task
        * ignore the error, and set the future's result as wanted
        * cancel the task (the future will be marked "cancelled")
        * retry as many times as wanted, immediately or with a wait period
        * do nothing: the default error handling is used
        """
        event_id = History.get_event_id(event)
        if event_id in self.handled_failures:  # don't call workflow method multiple times
            return self.handled_failures[event_id]

        logger.debug("handle_failure: failed_id=%s", event_id)
        rc = self.do_handle_failure(event, future, swf_task, exception_class)
        self.handled_failures[event_id] = rc
        return rc

    def do_handle_failure(
        self,
        event: dict,
        future: futures.Future,
        swf_task: ActivityTask | WorkflowTask,
        exception_class: type[Exception],
    ) -> futures.Future | tuple[futures.Future | None, ActivityTask | WorkflowTask | TimerTask] | None:
        timer = self.find_timer_associated_with(event, swf_task)
        if timer:
            if isinstance(timer["control"], str):  # FIXME unconditional?
                control = format.decode(timer["control"])
            else:
                control = timer["control"]
            if not isinstance(control, dict):
                control = {}
            if timer["state"] == "started":
                logger.debug(f'handle_failure: timer {timer["id"]} started, "pending" future')
                return futures.Future(), swf_task  # mark as pending
            elif timer["state"] in ("fired", "canceled"):
                logger.debug(f"handle_failure: timer {timer['id']} fired or canceled, retrying")
                swf_task.args = control.get("args", ())
                swf_task.kwargs = control.get("kwargs", {})
                return None, swf_task
            elif timer["state"] == "start_failed":
                raise exceptions.TaskFailed("timer", timer["id"], timer["cause"])
            else:  # TODO: handle
                logger.warning(f'Unexpected timer state for timer "{timer["id"]}": {timer["state"]}')

        failure_context = base_task.TaskFailureContext(
            a_task=swf_task, event=event, future=future, exception_class=exception_class, history=self._history
        )
        if hasattr(self.workflow, "on_task_failure"):
            new_failure_context: base_task.TaskFailureContext = self.workflow.on_task_failure(failure_context)
            if new_failure_context:
                failure_context = new_failure_context
            future, swf_task, event = (
                failure_context.future,
                failure_context.a_task,
                failure_context.event,
            )  # updatable
            if failure_context.decision == base_task.TaskFailureContext.Decision.abort:
                if swf_task.payload.raises_on_failure:
                    raise exception_class(swf_task, future.exception)
                return future, swf_task
            elif failure_context.decision == base_task.TaskFailureContext.Decision.ignore:
                future.set_exception(None)
                return future, swf_task
            elif failure_context.decision == base_task.TaskFailureContext.Decision.cancel:
                future.set_cancelled()
                return future, swf_task
            elif failure_context.decision == base_task.TaskFailureContext.Decision.retry_now or (
                failure_context.decision == base_task.TaskFailureContext.Decision.retry_later
                and not failure_context.retry_wait_timeout
            ):
                return None, swf_task
            elif failure_context.decision == base_task.TaskFailureContext.Decision.retry_later:
                return (
                    None,
                    TimerTask(
                        self.get_retry_task_timer_id(swf_task),
                        failure_context.retry_wait_timeout,
                        swf_task.get_input(),
                    ),
                )
            elif failure_context.decision == base_task.TaskFailureContext.Decision.handled:
                return future, swf_task
            if failure_context.decision != base_task.TaskFailureContext.Decision.none:
                raise ValueError(f"Unexpected TaskFailureValue decision: {failure_context.decision}")

        new_failure_context = self.default_failure_handling(failure_context)
        return new_failure_context.future

    @staticmethod
    def default_failure_handling(
        failure_context: base_task.TaskFailureContext,
    ) -> base_task.TaskFailureContext:
        # Compare number of retries in history with configured max retries
        # NB: we used to do a strict comparison (==), but that can lead to
        # infinite retries in case the code is redeployed with a decreased
        # retry limit and a workflow has a already crossed the new limit. So
        # ">=" is better there.
        if failure_context.event.get("retry", 0) >= failure_context.a_task.payload.retry:
            if failure_context.a_task.payload.raises_on_failure:
                raise failure_context.exception_class(failure_context.a_task, failure_context.future.exception)
        else:
            # Otherwise retry the workflow by scheduling it again.
            failure_context.future = None  # means it is not in SWF.
        failure_context.decision = base_task.TaskFailureContext.Decision.handled
        return failure_context

    def find_timer_associated_with(self, event: dict, swf_task: ActivityTask | WorkflowTask) -> dict | None:
        """
        Return a potential timer "associated with" an event, i.e.
        * with a related name
        * launched in a decision completed after the event's decision
        :param event:
        :param swf_task:
        :return:
        """
        timer = self._history.timers.get(self.get_retry_task_timer_id(swf_task))
        if timer and timer["decision_task_completed_event_id"] > event["decision_task_completed_event_id"]:
            return timer
        return None

    @staticmethod
    def get_retry_task_timer_id(swf_task):
        return f"__simpleflow_task_{str(swf_task.id)}"

    def schedule_task(
        self, a_task: ActivityTask | WorkflowTask | SignalTask | MarkerTask, task_list: str | None = None
    ) -> None:
        """
        Let a task schedule itself.
        If too many decisions are in flight, add a timer decision and raise ExecutionBlocked.
        :raise: exceptions.ExecutionBlocked if too many decisions waiting
        """

        if a_task.idempotent:
            task_identifier = (type(a_task), self.domain, a_task.id)
            if task_identifier in self._idempotent_tasks_to_submit:
                logger.debug(f"Not resubmitting task {a_task.name}")
                return
            self._idempotent_tasks_to_submit.add(task_identifier)

        # NB: ``decisions`` contains a single decision.
        decisions = a_task.schedule(self.domain, task_list, priority=self.current_priority, executor=self)

        # Ready to schedule
        if isinstance(a_task, ActivityTask):
            self._open_activity_count += 1
        elif isinstance(a_task, (MarkerTask, CancelTimerTask)):
            self._append_timer = True  # Marker and CancelTimer don't generate decisions, so force a wake-up timer

        # Check if we won't violate the 1MB limit on API requests ; if so, do NOT
        # schedule the requested task and block execution instead, with a timer
        # to wake up the workflow immediately after completing these decisions.
        # See: http://docs.aws.amazon.com/amazonswf/latest/developerguide/swf-dg-limits.html
        # NB: here we use json.dumps, not json_dumps, since the serialization will
        # happen inside boto.swf and is out of our control.
        request_size = len(json.dumps(self._decisions_and_context.decisions + decisions))
        # We keep a 5kB of error margin for headers, json structure, and the
        # timer decision, and 32kB for the context, even if we don't use it now.
        if request_size > constants.MAX_REQUEST_SIZE - 5000 - 32000:
            # TODO: at this point we may check that self._decisions is not empty
            # If it's the case, it means that a single decision was weighting
            # more than 900kB, so we have bigger problems.
            self._append_timer = True
            raise exceptions.ExecutionBlocked()

        self._decisions_and_context.extend_decision(decisions)

        # Check if we won't exceed max decisions -1
        # TODO: if we had exactly MAX_DECISIONS - 1 to take, this will wake up
        # the workflow for no reason. Evaluate if we can do better.
        if len(self._decisions_and_context.decisions) == constants.MAX_DECISIONS - 1:
            # We add a timer to wake up the workflow immediately after
            # completing these decisions.
            self._append_timer = True
            raise exceptions.ExecutionBlocked()

    def _add_start_timer_decision(self, id, timeout=0):
        timer = simpleflow.swf.mapper.models.decision.TimerDecision("start", id=id, start_to_fire_timeout=str(timeout))
        self._decisions_and_context.append_decision(timer)

    EVENT_TYPE_TO_FUTURE: dict[
        str, Callable[[ActivityTask | WorkflowTask | SignalTask | MarkerTask, dict[str, Any]], futures.Future | None]
    ] = {
        "activity": resume_activity,
        "child_workflow": resume_child_workflow,
        "signal": get_future_from_signal_event,
        "external_workflow": get_future_from_external_workflow_event,
        "marker": _get_future_from_marker_event,
        "timer": _get_future_from_timer_event,
    }

    def resume(self, a_task: ActivityTask | WorkflowTask | SignalTask | MarkerTask, *args, **kwargs) -> futures.Future:
        """Resume the execution of a task.
        Called by `submit`.

        If the task was scheduled, returns a future that wraps its state,
        otherwise schedules it.
        If in repair mode, we may fake the task to repair from the previous history.
        :raise: exceptions.ExecutionBlocked if open activities limit reached
        """

        self.make_task_id(a_task, *args, **kwargs)
        event = self.find_event(a_task, self._history)
        logger.debug(f"executor: resume {a_task}, event={event}")
        future = None

        # in repair mode, check if we absolutely want to re-execute this task
        is_repair = bool(self.repair_with)
        force_execution = self.force_activities and self.force_activities.search(a_task.id)

        # try to fill in the blanks with the workflow we're trying to repair if any
        if not event and is_repair and not force_execution:
            # try to find a former event matching this task
            former_event = self.find_event(a_task, self.repair_with)
            # ... but only keep the event if the task was successful
            if former_event and former_event["state"] == "completed":
                logger.info(f"faking task completed successfully in previous workflow: {former_event['id']}")
                json_hash = hashlib.md5(json_dumps(former_event).encode("utf-8")).hexdigest()  # nosec
                fake_task_list = "FAKE-" + json_hash

                # schedule task on a fake task list
                self.schedule_task(a_task, task_list=fake_task_list)
                future = futures.Future()

                # start a dedicated process to handle the fake activity
                run_fake_task_worker(self.domain.name, fake_task_list, former_event)

        # back to normal execution flow
        if event:
            ttf = self.EVENT_TYPE_TO_FUTURE.get(event["type"])
            if ttf:
                future_and_more = ttf(self, a_task, event)
                if isinstance(future_and_more, tuple):
                    future, a_task = future_and_more
                else:
                    future = future_and_more
            if event["type"] == "activity":
                if future and future.state in (futures.PENDING, futures.RUNNING):
                    self._open_activity_count += 1

        if not future:
            self.schedule_task(a_task, task_list=self.task_list)
            future = futures.Future()  # return a pending future.

        if self._open_activity_count == constants.MAX_OPEN_ACTIVITY_COUNT:
            logger.warning(f"limit of {constants.MAX_OPEN_ACTIVITY_COUNT} open activities reached")
            raise exceptions.ExecutionBlocked

        return future

    def make_task_id(self, a_task: ActivityTask | WorkflowTask | SignalTask | MarkerTask, *args, **kwargs) -> None:
        if a_task.id:  # Can be already set (WorkflowTask)
            return

        is_repair = bool(self.repair_with)
        if is_repair:
            workflow_id, run_id = self._repair_workflow_id, self._repair_run_id
        else:
            workflow_id, run_id = self._workflow_id, self._run_id
        a_task.id = self._make_task_id(a_task, workflow_id, run_id, *args, **kwargs)

    def _compute_priority(self, priority_set_on_submit, a_task):
        """
        Computes the correct task priority, with the following precedence (first
        is better/preferred):
        - priority set with self.submit(..., __priority=<N>)
        - priority set on the activity task decorator if any
        - priority set on the workflow execution
        - None otherwise

        :param priority_set_on_submit:
        :type  priority_set_on_submit: str|int|PRIORITY_NOT_SET

        :param a_task:
        :type  a_task: ActivityTask|WorkflowTask

        :returns: the priority for this task
        :rtype: str|int|None
        """
        if priority_set_on_submit is not PRIORITY_NOT_SET:
            return priority_set_on_submit
        elif isinstance(a_task, ActivityTask) and a_task.activity.task_priority is not PRIORITY_NOT_SET:
            return a_task.activity.task_priority
        elif self._workflow.task_priority is not PRIORITY_NOT_SET:
            return self._workflow.task_priority
        return None

    def submit(self, func, *args, **kwargs):
        """Register a function and its arguments for asynchronous execution.

        ``*args`` and ``**kwargs`` must be serializable in JSON.
        :type func: simpleflow.base.Submittable | Activity | Workflow

        """
        # NB: we don't set self.current_priority here directly, because we need
        # to extract it from the underlying Activity() if it's not passed to
        # self.submit() ; we DO need to pop the "__priority" kwarg though, so it
        # doesn't pollute the rest of the code.
        priority_set_on_submit = kwargs.pop("__priority", PRIORITY_NOT_SET)

        # casts simpleflow.task.*Task to their equivalent in simpleflow.swf.task
        if not isinstance(func, SwfTask):
            if isinstance(func, base_task.ActivityTask):
                func = ActivityTask.from_generic_task(func)
            elif isinstance(func, base_task.WorkflowTask):
                func = WorkflowTask.from_generic_task(func)
            elif isinstance(func, base_task.SignalTask):
                func = SignalTask.from_generic_task(func, self._workflow_id, self._run_id, None, None)
            elif isinstance(func, base_task.MarkerTask):
                func = MarkerTask.from_generic_task(func)
            elif isinstance(func, base_task.TimerTask):
                func = TimerTask.from_generic_task(func)
            elif isinstance(func, base_task.CancelTimerTask):
                func = CancelTimerTask.from_generic_task(func)

        try:
            # do not use directly "Submittable" here because we want to catch if
            # we don't have an instance from a class known to work under simpleflow.swf
            if isinstance(func, SwfTask):
                # no need to wrap it, already wrapped in the correct format
                a_task = func
            elif isinstance(func, Activity):
                a_task = ActivityTask(func, *args, **kwargs)
            elif issubclass_(func, Workflow):
                a_task = WorkflowTask(self, func, *args, **kwargs)
            elif isinstance(func, WaitForSignal):
                future = self.get_future_from_signal(func.signal_name)
                logger.debug(f"submitted WaitForSignalTask({func.signal_name}): future={future}")
                if not future.done:
                    self._decisions_and_context.append_kv_to_set_context("waiting_signals", func.signal_name)
                return future
            elif isinstance(func, Submittable):
                raise TypeError(
                    f"invalid type Submittable {type(func)} for {func}"
                    f" (you probably wanted a simpleflow.swf.task.*Task)"
                )
            else:
                raise TypeError(f"invalid type {type(func)} for {func}")
        except exceptions.ExecutionBlocked:
            return futures.Future()

        # extract priority now that we have a *Task
        self.current_priority = self._compute_priority(priority_set_on_submit, a_task)

        # finally resume task
        return self.resume(a_task, *a_task.args, **a_task.kwargs)

    # TODO: check if really used or remove it
    def map(self, callable, iterable):
        """Submit *callable* with each of the items in ``*iterables``.

        All items in ``*iterables`` must be serializable in JSON.

        """
        iterable = task.get_actual_value(iterable)
        return super().map(callable, iterable)

    # TODO: check if really used or remove it
    def starmap(self, callable, iterable):
        iterable = task.get_actual_value(iterable)
        return super().starmap(callable, iterable)

    def replay(
        self, decision_response: simpleflow.swf.mapper.responses.Response, decref_workflow: bool = True
    ) -> DecisionsAndContext:
        """Replay the workflow from the start until it blocks.
        Called by the DeciderWorker.

        :param decision_response: an object wrapping the PollForDecisionTask response
        :param decref_workflow : Decref workflow once replay is done (to save memory)

        :returns: a list of decision with an optional context
        """
        self.reset()

        # noinspection PyUnresolvedReferences
        history = decision_response.history
        self._history = History(history)
        self._history.parse()
        self.build_run_context(decision_response)
        # noinspection PyUnresolvedReferences
        self._execution = decision_response.execution

        workflow_started_event = history[0]
        input = workflow_started_event.input
        if input is None:
            input = {}
        args = input.get("args", ())
        kwargs = input.get("kwargs", {})

        self.before_replay()

        try:
            if self._history.cancel_requested:
                decisions = self.handle_cancel_requested()
                if decisions is not None:
                    self.after_replay()
                    self.after_closed()
                    if decref_workflow:
                        self.decref_workflow()
                    return DecisionsAndContext(decisions)
            self.propagate_signals()
            result = self.run_workflow(*args, **kwargs)
        except exceptions.ExecutionBlocked:
            logger.info(
                f"{self._open_activity_count} open activities ({len(self._decisions_and_context.decisions)} decisions)"
            )
            self.after_replay()
            if decref_workflow:
                self.decref_workflow()
            if self._append_timer:
                self._add_start_timer_decision("_simpleflow_wake_up_timer")

            if not self._decisions_and_context.execution_context:
                self.maybe_clear_execution_context()

            return self._decisions_and_context
        except (exceptions.TaskException, exceptions.WorkflowException) as err:
            decision = self.handle_replay_swf_exception(err)
            self.after_closed()
            if decref_workflow:
                self.decref_workflow()
            return DecisionsAndContext([decision])

        except Exception as err:
            reason = f"Cannot replay the workflow: {err.__class__.__name__}({err})"

            tb = traceback.format_exc()
            details = f"Traceback:\n{tb}"
            logger.exception("%s", reason + "\n" + details)  # Don't let logger try to interpolate the message

            self.on_failure(reason)

            decision = simpleflow.swf.mapper.models.decision.WorkflowExecutionDecision()
            decision.fail(
                reason=reason,
                details=details,
            )
            self.after_closed()
            if decref_workflow:
                self.decref_workflow()
            return DecisionsAndContext([decision])

        self.after_replay()
        decision = simpleflow.swf.mapper.models.decision.WorkflowExecutionDecision()
        decision.complete(result=result)
        self.on_completed()
        self.after_closed()
        if decref_workflow:
            self.decref_workflow()
        return DecisionsAndContext([decision])

    def handle_replay_swf_exception(
        self, err: exceptions.TaskException | exceptions.WorkflowException
    ) -> simpleflow.swf.mapper.models.decision.WorkflowExecutionDecision:
        def _extract_reason():
            if hasattr(err.exception, "reason"):
                raw = err.exception.reason
                # don't parse a potential json object here, since we will cast
                # the result to a string anyway, better keep a json representation
                return format.decode(raw, parse_json=False, use_proxy=False)
            return repr(err.exception)

        reason = f'Workflow execution error in {err.payload.name}: "{_extract_reason()}"'
        logger.exception("%s", reason)  # Don't let logger try to interpolate the message
        details = getattr(err.exception, "details", None)
        self.on_failure(reason, details)
        decision = simpleflow.swf.mapper.models.decision.WorkflowExecutionDecision()
        decision.fail(
            reason=reason,
            details=details,
        )
        return decision

    def maybe_clear_execution_context(self):
        """
        Replace a null execution_context with an empty string if the preceding one was set.
        This is to clear latestExecutionContext.
        :return:
        """
        events = self._history.events
        last_completed_decision = next(
            # next((generator), default) to prevent StopIteration. Python is fun :-)
            (e for e in reversed(events) if e.type == "DecisionTask" and e.state == "completed"),
            None,
        )
        last_decision_had_context = (
            last_completed_decision
            and hasattr(last_completed_decision, "execution_context")
            and last_completed_decision.execution_context
        )
        if last_decision_had_context:
            self._decisions_and_context.execution_context = ""

    def decref_workflow(self):
        """
        Set the `_workflow` ivar to None in the hope of reducing memory consumption.
        """
        self._workflow = None

    def before_replay(self):
        return self._workflow.before_replay(self._history)

    def after_replay(self):
        return self._workflow.after_replay(self._history)

    def after_closed(self):
        return self._workflow.after_closed(self._history)

    def on_failure(self, reason, details=None):
        try:
            self._workflow.on_failure(self._history, reason, details)
        except NotImplementedError:
            pass

    def on_completed(self):
        try:
            self._workflow.on_completed(self._history)
        except NotImplementedError:
            pass

    def on_canceled(self):
        self._workflow.on_canceled(self._history)

    def fail(self, reason, details=None):
        self.on_failure(reason, details)

        decision = simpleflow.swf.mapper.models.decision.WorkflowExecutionDecision()
        decision.fail(
            reason=f"Workflow execution failed: {reason}",
            details=details,
        )

        self._decisions_and_context.append_decision(decision)
        raise exceptions.ExecutionBlocked("workflow execution failed")

    def run(self, decision_response):
        return self.replay(decision_response)

    def get_run_context(self):
        return self._run_context

    def build_run_context(self, decision_response):
        """
        Extract data from the execution and history.
        :param decision_response:
        :type  decision_response:  simpleflow.swf.mapper.responses.Response
        """
        # noinspection PyUnresolvedReferences
        execution = decision_response.execution
        if not execution:
            # For tests that don't provide an execution object.
            return

        # noinspection PyUnresolvedReferences
        history = decision_response.history
        workflow_started_event = history[0]
        self._run_context = {
            "name": execution.workflow_type.name,
            "version": execution.workflow_type.version,
            "domain_name": self.domain.name,
            "workflow_id": execution.workflow_id,
            "run_id": execution.run_id,
            "tag_list": getattr(workflow_started_event, "tag_list", None) or [],
            "continued_execution_run_id": getattr(workflow_started_event, "continued_execution_run_id", None),
            "parent_workflow_id": getattr(workflow_started_event, "parent_workflow_execution", {}).get("workflowId"),
            "parent_run_id": getattr(workflow_started_event, "parent_workflow_execution", {}).get("runId"),
        }

    @property
    def _workflow_id(self):
        return self._run_context.get("workflow_id")

    @property
    def _run_id(self):
        return self._run_context.get("run_id")

    def signal(self, name, *args, workflow_id: str = None, run_id: str = None, propagate: bool = True, **kwargs):
        """
        Send a signal.
        Pop workflow_id, run_id and propagate (default: True) from the kwargs.
        If workflow_id is not set or falsy, use the current workflow_id/run_id.
        """
        logger.debug(
            f"signal: name={name}, workflow_id={workflow_id if workflow_id else self._workflow_id},"
            f" run_id={run_id if workflow_id else self._run_id}, propagate={propagate}"
        )

        extra_input = {"__propagate": propagate if isinstance(propagate, bool) else str(propagate)}
        return SignalTask(
            name,
            workflow_id=workflow_id if workflow_id else self._workflow_id,
            run_id=run_id if workflow_id else self._run_id,
            extra_input=extra_input,
            *args,
            **kwargs,
        )

    def wait_signal(self, name):
        logger.debug(f"{self._workflow_id} - wait_signal({name})")
        return WaitForSignal(name)

    def propagate_signals(self):
        """
        Send every signals we got to our parent and children.
        Don't send to workflows present in history.signaled_workflows.
        """
        history = self._history
        if not history.signals:
            return

        known_workflows_ids = []
        if self._run_context.get("parent_workflow_id") and getattr(self.workflow, "propagate_signals_to_parent", True):
            known_workflows_ids.append(
                (
                    self._run_context["parent_workflow_id"],
                    self._run_context["parent_run_id"],
                )
            )
        known_workflows_ids.extend(
            (w["workflow_id"], w["run_id"]) for w in history.child_workflows.values() if w["state"] == "started"
        )

        known_workflows_ids = frozenset(known_workflows_ids)

        signals_scheduled = False

        for signal in history.signals.values():
            input = signal["input"]
            if not isinstance(input, dict):  # foreign signal: don't try processing it
                continue
            propagate = input.get("__propagate", False)
            if not propagate:
                continue
            name = signal["name"]

            args = input.get("args", ())
            kwargs = input.get("kwargs", {})
            sender = (signal["external_workflow_id"], signal["external_run_id"])
            signaled_workflows_ids = {(w["workflow_id"], w["run_id"]) for w in history.signaled_workflows[name]}
            not_signaled_workflows_ids = list(known_workflows_ids - signaled_workflows_ids - {sender})
            extra_input = {"__propagate": propagate}
            for workflow_id, run_id in not_signaled_workflows_ids:
                self.schedule_task(SignalTask(name, workflow_id, run_id, None, extra_input, *args, **kwargs))
                signals_scheduled = True
        if signals_scheduled:
            raise exceptions.ExecutionBlocked()

    def record_marker(self, name, details=None):
        return MarkerTask(name, details)

    def list_markers(self, all=False):
        if all:
            return [Marker(m["name"], format.decode(m["details"])) for ml in self._history.markers.values() for m in ml]
        rc = []
        for ml in self._history.markers.values():
            m = ml[-1]
            if m["state"] == "recorded":
                rc.append(Marker(m["name"], format.decode(m["details"])))
        return rc

    def get_event_details(self, event_type, event_name):
        if event_type == "signal":
            return self._history.signals.get(event_name)
        elif event_type == "marker":
            marker_list = self._history.markers.get(event_name)
            if not marker_list:
                return None
            marker_list = [m for m in marker_list if m["state"] == "recorded"]
            if not marker_list:
                return None
            # Make pleasing details
            marker = copy.copy(marker_list[-1])
            marker["details"] = format.decode(marker["details"])
            return marker
        elif event_type == "timer":
            return self._history.timers.get(event_name)
        else:
            raise ValueError(f"Unimplemented type {event_type!r} for get_event_details")

    def handle_cancel_requested(self) -> list[simpleflow.swf.mapper.models.decision.WorkflowExecutionDecision] | None:
        decision = simpleflow.swf.mapper.models.decision.WorkflowExecutionDecision()
        is_current_decision = self._history.completed_decision_id < self._history.cancel_requested_id
        should_cancel = self._workflow.should_cancel(self._history)
        if not should_cancel:
            return None  # ignore cancel
        if is_current_decision:
            self.on_canceled()
            decision.cancel()
            return [decision]
        if self._history.cancel_failed:
            logger.warning("failed: %s", self._history.cancel_failed)
        if (
            self._history.cancel_failed
            and self._history.cancel_failed_decision_task_completed_event_id == self._history.completed_decision_id
        ):
            # Per http://docs.aws.amazon.com/amazonswf/latest/apireference/API_Decision.html,
            # we should call RespondDecisionTaskCompleted without any decisions; however this hangs the workflow...

            # <1 WorkflowExecution : started>, <2 DecisionTask : scheduled>, <3 DecisionTask : started>,
            # <4 DecisionTask : completed>, <5 ActivityTask : scheduled>, <6 ActivityTask : started>,
            # <7 WorkflowExecution : cancel_requested>, <8 DecisionTask : scheduled>, <9 DecisionTask : started>,
            # <10 ActivityTask : completed>, <11 DecisionTask : scheduled>, <12 DecisionTask : completed>,
            # <13 WorkflowExecution : cancel_failed>, <14 DecisionTask : started>

            # return []
            pass
        decision.cancel()
        return [decision]

    def continue_as_new(self, workflow: type[Workflow], *args, **kwargs):
        return ContinueAsNewWorkflowTask(executor=self, workflow=workflow, *args, **kwargs)