Repository: localstack/localstack
Pull request: [SFN] Enhanced Support for Exception Handling in Parallel States
<!-- Please refer to the contribution guidelines before raising a PR: https://github.com/localstack/localstack/blob/master/CONTRIBUTING.md -->

<!-- Why am I raising this PR? Add context such as related issues, PRs, or documentation. -->
## Motivation
The SFN v2 interpreter is currently unable to propagate Parallel states' Branch failures to the Parallel state. This masks state failures, and prevents exception handling. The PR resolves such issues, ensuring that Branch failures are relayed orderly to the state's thread.
Closes: #9863 

<!-- What notable changes does this PR make? -->
## Changes
Reworked the role of the Branches evaluation component to dispatch to a branch worker pool, which intercepts through callbacks workers' terminations. Workers' failures are cached and other workers are stopped. It then signals the state's thread to resume computation whenever a stopping condition is met (all branches evaluated, or at least one branch failed).
Reworked program workers into branch workers, which integrate better with branches objects.
Reworked the eval state of parallel states to orderly handle retry and/or catch declarations.
Added a few relevant tests. 

<!-- The following sections are optional, but can be useful! 

## Testing

Description of how to test the changes

## TODO

What's left to do:

- [ ] ...
- [ ] ...

-->



Problem statement:
bug: Step Functions error not propagated outside of Parallel state
### Is there an existing issue for this?

- [X] I have searched the existing issues

### Current Behavior

I'm executing a state machine using the following definition:

```json
{
  "StartAt": "Parallel",
  "States": {
    "Parallel": {
      "Type": "Parallel",
      "Next": "Unreachable",
      "Branches": [
        {
          "StartAt": "Fail",
          "States": {
            "Fail": {
              "Type": "Fail"
            }
          }
        }
      ]
    },
    "Unreachable": {
      "Type": "Pass",
      "Parameters": {
        "UnreachableOutput": true
      },
      "End": true
    }
  }
}
```

The execution is a success with the following output:

```json
{
   "UnreachableOutput": true
}
```

### Expected Behavior

The execution should fail.

On AWS executing a state machine using the provided definition results in an error. The execution should fail before reaching the state named `Unreachable` because an error occurred in one of the branches of the preceding Parallel state.

### How are you starting LocalStack?

With a docker-compose file

### Steps To Reproduce

Create and execute a state machine using the provided definition.

### Environment

```markdown
- LocalStack: 3.0.3.dev20231207105008
```


### Anything else?

_No response_

Hints from issue discussion:

Commit message: Merge branch 'master' into MEP-sfn-9863

Patch to review:
diff --git a/localstack/services/stepfunctions/asl/component/state/state_execution/state_parallel/branch_worker.py b/localstack/services/stepfunctions/asl/component/state/state_execution/state_parallel/branch_worker.py
new file mode 100644
index 0000000000000..0ca3a14bc0eca
--- /dev/null
+++ b/localstack/services/stepfunctions/asl/component/state/state_execution/state_parallel/branch_worker.py
@@ -0,0 +1,52 @@
+import abc
+import logging
+import threading
+from typing import Final, Optional
+
+from localstack.aws.api.stepfunctions import Timestamp
+from localstack.services.stepfunctions.asl.component.program.program import Program
+from localstack.services.stepfunctions.asl.eval.environment import Environment
+from localstack.utils.threads import TMP_THREADS
+
+LOG = logging.getLogger(__name__)
+
+
+class BranchWorker:
+    class BranchWorkerComm(abc.ABC):
+        @abc.abstractmethod
+        def on_terminated(self, env: Environment):
+            ...
+
+    _branch_worker_comm: Final[BranchWorkerComm]
+    _program: Final[Program]
+    _worker_thread: Optional[threading.Thread]
+    env: Final[Environment]
+
+    def __init__(self, branch_worker_comm: BranchWorkerComm, program: Program, env: Environment):
+        self._branch_worker_comm = branch_worker_comm
+        self._program = program
+        self.env = env
+        self._worker_thread = None
+
+    def _thread_routine(self) -> None:
+        LOG.info(f"[BranchWorker] [launched] [id: {self._worker_thread.native_id}]")
+        self._program.eval(self.env)
+        LOG.info(f"[BranchWorker] [terminated] [id: {self._worker_thread.native_id}]")
+        self._branch_worker_comm.on_terminated(env=self.env)
+
+    def start(self):
+        if self._worker_thread is not None:
+            raise RuntimeError(f"Attempted to rerun BranchWorker for program ${self._program}.")
+
+        self._worker_thread = threading.Thread(target=self._thread_routine)
+        TMP_THREADS.append(self._worker_thread)
+        self._worker_thread.start()
+
+    def stop(self, stop_date: Timestamp, cause: Optional[str], error: Optional[str]) -> None:
+        env = self.env
+        if env:
+            try:
+                env.set_stop(stop_date, cause, error)
+            except Exception:
+                # Ignore closing exceptions, this method attempts to release resources earlier.
+                pass
diff --git a/localstack/services/stepfunctions/asl/component/state/state_execution/state_parallel/branches_decl.py b/localstack/services/stepfunctions/asl/component/state/state_execution/state_parallel/branches_decl.py
index d23b4574d9986..33e75171903f1 100644
--- a/localstack/services/stepfunctions/asl/component/state/state_execution/state_parallel/branches_decl.py
+++ b/localstack/services/stepfunctions/asl/component/state/state_execution/state_parallel/branches_decl.py
@@ -1,10 +1,65 @@
-from typing import Final
+import copy
+import datetime
+import threading
+from typing import Final, Optional
 
+from localstack.aws.api.stepfunctions import ExecutionFailedEventDetails, HistoryEventType
+from localstack.services.stepfunctions.asl.component.common.error_name.custom_error_name import (
+    CustomErrorName,
+)
+from localstack.services.stepfunctions.asl.component.common.error_name.failure_event import (
+    FailureEvent,
+    FailureEventException,
+)
 from localstack.services.stepfunctions.asl.component.eval_component import EvalComponent
 from localstack.services.stepfunctions.asl.component.program.program import Program
-from localstack.services.stepfunctions.asl.eval.count_down_latch import CountDownLatch
+from localstack.services.stepfunctions.asl.component.state.state_execution.state_parallel.branch_worker import (
+    BranchWorker,
+)
 from localstack.services.stepfunctions.asl.eval.environment import Environment
-from localstack.services.stepfunctions.asl.eval.program_worker import ProgramWorker
+from localstack.services.stepfunctions.asl.eval.event.event_detail import EventDetails
+from localstack.services.stepfunctions.asl.eval.program_state import ProgramError, ProgramState
+from localstack.utils.collections import select_from_typed_dict
+
+
+class BranchWorkerPool(BranchWorker.BranchWorkerComm):
+    _mutex: Final[threading.Lock]
+    _termination_event: Final[threading.Event]
+    _active_workers_num: int
+
+    _terminated: bool
+    _terminated_with_error: Optional[ExecutionFailedEventDetails]
+
+    def __init__(self, workers_num: int):
+        self._mutex = threading.Lock()
+        self._termination_event = threading.Event()
+        self._active_workers_num = workers_num
+
+        self._terminated = False
+        self._terminated_with_error = None
+
+    def on_terminated(self, env: Environment):
+        if self._terminated:
+            return
+
+        with self._mutex:
+            end_program_state: ProgramState = env.program_state()
+            if isinstance(end_program_state, ProgramError):
+                self._terminated_with_error = select_from_typed_dict(
+                    typed_dict=ExecutionFailedEventDetails, obj=end_program_state.error or dict()
+                )
+                self._terminated = True
+                self._termination_event.set()
+            else:
+                self._active_workers_num -= 1
+                if self._active_workers_num == 0:
+                    self._termination_event.set()
+
+    def wait(self):
+        self._termination_event.wait()
+
+    def get_exit_event_details(self) -> Optional[ExecutionFailedEventDetails]:
+        return self._terminated_with_error
 
 
 class BranchesDecl(EvalComponent):
@@ -12,30 +67,50 @@ def __init__(self, programs: list[Program]):
         self.programs: Final[list[Program]] = programs
 
     def _eval_body(self, env: Environment) -> None:
-        # CountDownLatch to detect halting of all worker threads.
-        latch: Final[CountDownLatch] = CountDownLatch(len(self.programs))
-
         # Input value for every state_parallel process.
         input_val = env.stack.pop()
 
-        worker_pool: list[ProgramWorker] = list()
+        branch_worker_pool = BranchWorkerPool(workers_num=len(self.programs))
+
+        branch_workers: list[BranchWorker] = list()
         for program in self.programs:
             # Environment frame for this sub process.
             env_frame: Environment = env.open_frame()
-            env_frame.inp = input_val
+            env_frame.inp = copy.deepcopy(input_val)
 
             # Launch the worker.
-            worker = ProgramWorker()
-            worker.eval(program=program, env_frame=env_frame, latch=latch)
+            worker = BranchWorker(
+                branch_worker_comm=branch_worker_pool, program=program, env=env_frame
+            )
+            branch_workers.append(worker)
+
+            worker.start()
+
+        branch_worker_pool.wait()
 
-            worker_pool.append(worker)
+        # Propagate exception if parallel task failed.
+        exit_event_details: Optional[
+            ExecutionFailedEventDetails
+        ] = branch_worker_pool.get_exit_event_details()
+        if exit_event_details is not None:
+            for branch_worker in branch_workers:
+                branch_worker.stop(stop_date=datetime.datetime.now(), cause=None, error=None)
+                env.close_frame(branch_worker.env)
 
-        latch.wait()
+            exit_error_name = exit_event_details.get("error")
+            raise FailureEventException(
+                failure_event=FailureEvent(
+                    error_name=CustomErrorName(error_name=exit_error_name),
+                    event_type=HistoryEventType.ExecutionFailed,
+                    event_details=EventDetails(executionFailedEventDetails=exit_event_details),
+                )
+            )
 
+        # Collect the results and return.
         result_list = list()
 
-        for worker in worker_pool:
-            env_frame = worker.env_frame
+        for worker in branch_workers:
+            env_frame = worker.env
             result_list.append(env_frame.inp)
             env.close_frame(env_frame)
 
diff --git a/localstack/services/stepfunctions/asl/component/state/state_execution/state_parallel/state_parallel.py b/localstack/services/stepfunctions/asl/component/state/state_execution/state_parallel/state_parallel.py
index de3e5ef791c2c..07d4540bb58fa 100644
--- a/localstack/services/stepfunctions/asl/component/state/state_execution/state_parallel/state_parallel.py
+++ b/localstack/services/stepfunctions/asl/component/state/state_execution/state_parallel/state_parallel.py
@@ -1,4 +1,12 @@
+import copy
+
 from localstack.aws.api.stepfunctions import HistoryEventType
+from localstack.services.stepfunctions.asl.component.common.catch.catch_outcome import CatchOutcome
+from localstack.services.stepfunctions.asl.component.common.error_name.failure_event import (
+    FailureEvent,
+    FailureEventException,
+)
+from localstack.services.stepfunctions.asl.component.common.retry.retry_outcome import RetryOutcome
 from localstack.services.stepfunctions.asl.component.state.state_execution.execute_state import (
     ExecutionState,
 )
@@ -40,3 +48,40 @@ def _eval_execution(self, env: Environment) -> None:
             hist_type_event=HistoryEventType.ParallelStateSucceeded,
             update_source_event_id=False,
         )
+
+    def _eval_state(self, env: Environment) -> None:
+        # Initialise the retry counter for execution states.
+        env.context_object_manager.context_object["State"]["RetryCount"] = 0
+
+        # Cache the input, so it can be resubmitted in case of failure.
+        input_value = copy.deepcopy(env.stack.pop())
+
+        # Attempt to evaluate the state's logic through until it's successful, caught, or retries have run out.
+        while True:
+            try:
+                env.stack.append(input_value)
+                self._evaluate_with_timeout(env)
+                break
+            except FailureEventException as failure_event_ex:
+                failure_event: FailureEvent = failure_event_ex.failure_event
+
+                if self.retry is not None:
+                    retry_outcome: RetryOutcome = self._handle_retry(
+                        env=env, failure_event=failure_event
+                    )
+                    if retry_outcome == RetryOutcome.CanRetry:
+                        continue
+
+                env.event_history.add_event(
+                    context=env.event_history_context,
+                    hist_type_event=HistoryEventType.ParallelStateFailed,
+                )
+
+                if self.catch is not None:
+                    catch_outcome: CatchOutcome = self._handle_catch(
+                        env=env, failure_event=failure_event
+                    )
+                    if catch_outcome == CatchOutcome.Caught:
+                        break
+
+                self._handle_uncaught(env=env, failure_event=failure_event)
diff --git a/localstack/services/stepfunctions/asl/eval/program_worker.py b/localstack/services/stepfunctions/asl/eval/program_worker.py
deleted file mode 100644
index 507d5b875842b..0000000000000
--- a/localstack/services/stepfunctions/asl/eval/program_worker.py
+++ /dev/null
@@ -1,44 +0,0 @@
-import logging
-import threading
-from typing import Optional
-
-from localstack.aws.api.stepfunctions import Timestamp
-from localstack.services.stepfunctions.asl.component.program.program import Program
-from localstack.services.stepfunctions.asl.eval.count_down_latch import CountDownLatch
-from localstack.services.stepfunctions.asl.eval.environment import Environment
-from localstack.utils.threads import TMP_THREADS
-
-LOG = logging.getLogger(__name__)
-
-
-class ProgramWorker:
-    def __init__(self):
-        self._worker_thread: Optional[threading.Thread] = None
-        self.env_frame: Optional[Environment] = None
-
-    def _worker_routine(self, program: Program, latch: Optional[CountDownLatch]) -> None:
-        # TODO: exception handling.
-        LOG.info(f"[ProgramWorker] [launched] [id: {self._worker_thread.native_id}]")
-
-        program.eval(self.env_frame)
-
-        LOG.info(f"[ProgramWorker] [terminated] [id: {self._worker_thread.native_id}]")
-
-        self._worker_thread = None
-        if latch is not None:
-            latch.count_down()
-
-    def eval(
-        self, program: Program, env_frame: Environment, latch: Optional[CountDownLatch] = None
-    ):
-        if self._worker_thread is not None:
-            raise RuntimeError("Call to ProgramWorker.eval whilst another job is running.")
-        self.env_frame = env_frame
-        self._worker_thread = threading.Thread(target=self._worker_routine, args=(program, latch))
-        TMP_THREADS.append(self._worker_thread)
-        self._worker_thread.start()
-
-    def stop(self, stop_date: Timestamp, cause: Optional[str], error: Optional[str]) -> None:
-        env = self.env_frame
-        if env:
-            env.set_stop(stop_date, cause, error)
diff --git a/tests/aws/services/stepfunctions/templates/scenarios/scenarios_templates.py b/tests/aws/services/stepfunctions/templates/scenarios/scenarios_templates.py
index 738e55c8adda2..25cfbcd4190b4 100644
--- a/tests/aws/services/stepfunctions/templates/scenarios/scenarios_templates.py
+++ b/tests/aws/services/stepfunctions/templates/scenarios/scenarios_templates.py
@@ -12,6 +12,15 @@ class ScenariosTemplate(TemplateLoader):
         _THIS_FOLDER, "statemachines/catch_states_runtime.json5"
     )
     PARALLEL_STATE: Final[str] = os.path.join(_THIS_FOLDER, "statemachines/parallel_state.json5")
+    PARALLEL_STATE_FAIL: Final[str] = os.path.join(
+        _THIS_FOLDER, "statemachines/parallel_state_fail.json5"
+    )
+    PARALLEL_STATE_CATCH: Final[str] = os.path.join(
+        _THIS_FOLDER, "statemachines/parallel_state_catch.json5"
+    )
+    PARALLEL_STATE_RETRY: Final[str] = os.path.join(
+        _THIS_FOLDER, "statemachines/parallel_state_retry.json5"
+    )
     PARALLEL_STATE_ORDER: Final[str] = os.path.join(
         _THIS_FOLDER, "statemachines/parallel_state_order.json5"
     )
diff --git a/tests/aws/services/stepfunctions/templates/scenarios/statemachines/parallel_state_catch.json5 b/tests/aws/services/stepfunctions/templates/scenarios/statemachines/parallel_state_catch.json5
new file mode 100644
index 0000000000000..a05c8147d2862
--- /dev/null
+++ b/tests/aws/services/stepfunctions/templates/scenarios/statemachines/parallel_state_catch.json5
@@ -0,0 +1,38 @@
+{
+  "Comment": "PARALLEL_STATE_CATCH",
+  "StartAt": "ParallelState",
+  "States": {
+    "ParallelState": {
+      "Type": "Parallel",
+      "Branches": [
+        {
+          "StartAt": "Branch1",
+          "States": {
+            "Branch1": {
+              "Type": "Fail",
+              "Error": "FailureType",
+              "Cause": "Cause description"
+            }
+          }
+        },
+      ],
+      "Catch": [
+        {
+          "ErrorEquals": [
+            "FailureType"
+          ],
+          "Next": "FinalCaught",
+        },
+      ],
+      "Next": "Final"
+    },
+    "FinalCaught": {
+      "Type": "Pass",
+      "End": true
+    },
+    "Final": {
+      "Type": "Pass",
+      "End": true
+    }
+  }
+}
diff --git a/tests/aws/services/stepfunctions/templates/scenarios/statemachines/parallel_state_fail.json5 b/tests/aws/services/stepfunctions/templates/scenarios/statemachines/parallel_state_fail.json5
new file mode 100644
index 0000000000000..4f9e9851e0110
--- /dev/null
+++ b/tests/aws/services/stepfunctions/templates/scenarios/statemachines/parallel_state_fail.json5
@@ -0,0 +1,21 @@
+{
+  "Comment": "PARALLEL_STATE_FAIL",
+  "StartAt": "ParallelState",
+  "States": {
+    "ParallelState": {
+      "Type": "Parallel",
+      "End": true,
+      "Branches": [
+        {
+          "StartAt": "Branch1",
+          "States": {
+            "Branch1": {
+              "Type": "Fail",
+              "Error": "FailureType"
+            }
+          }
+        },
+      ]
+    }
+  }
+}
diff --git a/tests/aws/services/stepfunctions/templates/scenarios/statemachines/parallel_state_retry.json5 b/tests/aws/services/stepfunctions/templates/scenarios/statemachines/parallel_state_retry.json5
new file mode 100644
index 0000000000000..d2671863f1041
--- /dev/null
+++ b/tests/aws/services/stepfunctions/templates/scenarios/statemachines/parallel_state_retry.json5
@@ -0,0 +1,34 @@
+{
+  "Comment": "PARALLEL_STATE_RETRY",
+  "StartAt": "ParallelState",
+  "States": {
+    "ParallelState": {
+      "Type": "Parallel",
+      "Branches": [
+        {
+          "StartAt": "Branch1",
+          "States": {
+            "Branch1": {
+              "Type": "Fail",
+              "Error": "FailureType",
+              "Cause": "Cause description"
+            }
+          }
+        },
+      ],
+      "Retry": [
+        {
+          "ErrorEquals": [
+            "FailureType"
+          ],
+          "MaxAttempts": 2,
+        },
+      ],
+      "Next": "Final",
+    },
+    "Final": {
+      "Type": "Pass",
+      "End": true
+    }
+  }
+}
diff --git a/tests/aws/services/stepfunctions/v2/scenarios/test_base_scenarios.py b/tests/aws/services/stepfunctions/v2/scenarios/test_base_scenarios.py
index 3878c76f45979..efb5ce30eebbe 100644
--- a/tests/aws/services/stepfunctions/v2/scenarios/test_base_scenarios.py
+++ b/tests/aws/services/stepfunctions/v2/scenarios/test_base_scenarios.py
@@ -135,6 +135,69 @@ def test_parallel_state_order(
             exec_input,
         )
 
+    @markers.aws.validated
+    def test_parallel_state_fail(
+        self,
+        aws_client,
+        create_iam_role_for_sfn,
+        create_state_machine,
+        sfn_snapshot,
+    ):
+        template = ST.load_sfn_template(ST.PARALLEL_STATE_FAIL)
+        definition = json.dumps(template)
+
+        exec_input = json.dumps({})
+        create_and_record_execution(
+            aws_client.stepfunctions,
+            create_iam_role_for_sfn,
+            create_state_machine,
+            sfn_snapshot,
+            definition,
+            exec_input,
+        )
+
+    @markers.aws.validated
+    def test_parallel_state_catch(
+        self,
+        aws_client,
+        create_iam_role_for_sfn,
+        create_state_machine,
+        sfn_snapshot,
+    ):
+        template = ST.load_sfn_template(ST.PARALLEL_STATE_CATCH)
+        definition = json.dumps(template)
+
+        exec_input = json.dumps({})
+        create_and_record_execution(
+            aws_client.stepfunctions,
+            create_iam_role_for_sfn,
+            create_state_machine,
+            sfn_snapshot,
+            definition,
+            exec_input,
+        )
+
+    @markers.aws.validated
+    def test_parallel_state_retry(
+        self,
+        aws_client,
+        create_iam_role_for_sfn,
+        create_state_machine,
+        sfn_snapshot,
+    ):
+        template = ST.load_sfn_template(ST.PARALLEL_STATE_RETRY)
+        definition = json.dumps(template)
+
+        exec_input = json.dumps({})
+        create_and_record_execution(
+            aws_client.stepfunctions,
+            create_iam_role_for_sfn,
+            create_state_machine,
+            sfn_snapshot,
+            definition,
+            exec_input,
+        )
+
     @markers.aws.validated
     def test_map_state(
         self,
diff --git a/tests/aws/services/stepfunctions/v2/scenarios/test_base_scenarios.snapshot.json b/tests/aws/services/stepfunctions/v2/scenarios/test_base_scenarios.snapshot.json
index f279d633e8718..616f29eaf51d3 100644
--- a/tests/aws/services/stepfunctions/v2/scenarios/test_base_scenarios.snapshot.json
+++ b/tests/aws/services/stepfunctions/v2/scenarios/test_base_scenarios.snapshot.json
@@ -11750,5 +11750,317 @@
         }
       }
     }
+  },
+  "tests/aws/services/stepfunctions/v2/scenarios/test_base_scenarios.py::TestBaseScenarios::test_parallel_state_fail": {
+    "recorded-date": "16-12-2023, 18:52:06",
+    "recorded-content": {
+      "get_execution_history": {
+        "events": [
+          {
+            "executionStartedEventDetails": {
+              "input": {},
+              "inputDetails": {
+                "truncated": false
+              },
+              "roleArn": "snf_role_arn"
+            },
+            "id": 1,
+            "previousEventId": 0,
+            "timestamp": "timestamp",
+            "type": "ExecutionStarted"
+          },
+          {
+            "id": 2,
+            "previousEventId": 0,
+            "stateEnteredEventDetails": {
+              "input": {},
+              "inputDetails": {
+                "truncated": false
+              },
+              "name": "ParallelState"
+            },
+            "timestamp": "timestamp",
+            "type": "ParallelStateEntered"
+          },
+          {
+            "id": 3,
+            "previousEventId": 2,
+            "timestamp": "timestamp",
+            "type": "ParallelStateStarted"
+          },
+          {
+            "id": 4,
+            "previousEventId": 3,
+            "stateEnteredEventDetails": {
+              "input": {},
+              "inputDetails": {
+                "truncated": false
+              },
+              "name": "Branch1"
+            },
+            "timestamp": "timestamp",
+            "type": "FailStateEntered"
+          },
+          {
+            "id": 5,
+            "previousEventId": 4,
+            "timestamp": "timestamp",
+            "type": "ParallelStateFailed"
+          },
+          {
+            "executionFailedEventDetails": {
+              "error": "FailureType"
+            },
+            "id": 6,
+            "previousEventId": 5,
+            "timestamp": "timestamp",
+            "type": "ExecutionFailed"
+          }
+        ],
+        "ResponseMetadata": {
+          "HTTPHeaders": {},
+          "HTTPStatusCode": 200
+        }
+      }
+    }
+  },
+  "tests/aws/services/stepfunctions/v2/scenarios/test_base_scenarios.py::TestBaseScenarios::test_parallel_state_catch": {
+    "recorded-date": "16-12-2023, 21:45:29",
+    "recorded-content": {
+      "get_execution_history": {
+        "events": [
+          {
+            "executionStartedEventDetails": {
+              "input": {},
+              "inputDetails": {
+                "truncated": false
+              },
+              "roleArn": "snf_role_arn"
+            },
+            "id": 1,
+            "previousEventId": 0,
+            "timestamp": "timestamp",
+            "type": "ExecutionStarted"
+          },
+          {
+            "id": 2,
+            "previousEventId": 0,
+            "stateEnteredEventDetails": {
+              "input": {},
+              "inputDetails": {
+                "truncated": false
+              },
+              "name": "ParallelState"
+            },
+            "timestamp": "timestamp",
+            "type": "ParallelStateEntered"
+          },
+          {
+            "id": 3,
+            "previousEventId": 2,
+            "timestamp": "timestamp",
+            "type": "ParallelStateStarted"
+          },
+          {
+            "id": 4,
+            "previousEventId": 3,
+            "stateEnteredEventDetails": {
+              "input": {},
+              "inputDetails": {
+                "truncated": false
+              },
+              "name": "Branch1"
+            },
+            "timestamp": "timestamp",
+            "type": "FailStateEntered"
+          },
+          {
+            "id": 5,
+            "previousEventId": 4,
+            "timestamp": "timestamp",
+            "type": "ParallelStateFailed"
+          },
+          {
+            "id": 6,
+            "previousEventId": 5,
+            "stateExitedEventDetails": {
+              "name": "ParallelState",
+              "output": {
+                "Error": "FailureType",
+                "Cause": "Cause description"
+              },
+              "outputDetails": {
+                "truncated": false
+              }
+            },
+            "timestamp": "timestamp",
+            "type": "ParallelStateExited"
+          },
+          {
+            "id": 7,
+            "previousEventId": 6,
+            "stateEnteredEventDetails": {
+              "input": {
+                "Error": "FailureType",
+                "Cause": "Cause description"
+              },
+              "inputDetails": {
+                "truncated": false
+              },
+              "name": "FinalCaught"
+            },
+            "timestamp": "timestamp",
+            "type": "PassStateEntered"
+          },
+          {
+            "id": 8,
+            "previousEventId": 7,
+            "stateExitedEventDetails": {
+              "name": "FinalCaught",
+              "output": {
+                "Error": "FailureType",
+                "Cause": "Cause description"
+              },
+              "outputDetails": {
+                "truncated": false
+              }
+            },
+            "timestamp": "timestamp",
+            "type": "PassStateExited"
+          },
+          {
+            "executionSucceededEventDetails": {
+              "output": {
+                "Error": "FailureType",
+                "Cause": "Cause description"
+              },
+              "outputDetails": {
+                "truncated": false
+              }
+            },
+            "id": 9,
+            "previousEventId": 8,
+            "timestamp": "timestamp",
+            "type": "ExecutionSucceeded"
+          }
+        ],
+        "ResponseMetadata": {
+          "HTTPHeaders": {},
+          "HTTPStatusCode": 200
+        }
+      }
+    }
+  },
+  "tests/aws/services/stepfunctions/v2/scenarios/test_base_scenarios.py::TestBaseScenarios::test_parallel_state_retry": {
+    "recorded-date": "16-12-2023, 21:58:47",
+    "recorded-content": {
+      "get_execution_history": {
+        "events": [
+          {
+            "executionStartedEventDetails": {
+              "input": {},
+              "inputDetails": {
+                "truncated": false
+              },
+              "roleArn": "snf_role_arn"
+            },
+            "id": 1,
+            "previousEventId": 0,
+            "timestamp": "timestamp",
+            "type": "ExecutionStarted"
+          },
+          {
+            "id": 2,
+            "previousEventId": 0,
+            "stateEnteredEventDetails": {
+              "input": {},
+              "inputDetails": {
+                "truncated": false
+              },
+              "name": "ParallelState"
+            },
+            "timestamp": "timestamp",
+            "type": "ParallelStateEntered"
+          },
+          {
+            "id": 3,
+            "previousEventId": 2,
+            "timestamp": "timestamp",
+            "type": "ParallelStateStarted"
+          },
+          {
+            "id": 4,
+            "previousEventId": 3,
+            "stateEnteredEventDetails": {
+              "input": {},
+              "inputDetails": {
+                "truncated": false
+              },
+              "name": "Branch1"
+            },
+            "timestamp": "timestamp",
+            "type": "FailStateEntered"
+          },
+          {
+            "id": 5,
+            "previousEventId": 4,
+            "timestamp": "timestamp",
+            "type": "ParallelStateStarted"
+          },
+          {
+            "id": 6,
+            "previousEventId": 5,
+            "stateEnteredEventDetails": {
+              "input": {},
+              "inputDetails": {
+                "truncated": false
+              },
+              "name": "Branch1"
+            },
+            "timestamp": "timestamp",
+            "type": "FailStateEntered"
+          },
+          {
+            "id": 7,
+            "previousEventId": 6,
+            "timestamp": "timestamp",
+            "type": "ParallelStateStarted"
+          },
+          {
+            "id": 8,
+            "previousEventId": 7,
+            "stateEnteredEventDetails": {
+              "input": {},
+              "inputDetails": {
+                "truncated": false
+              },
+              "name": "Branch1"
+            },
+            "timestamp": "timestamp",
+            "type": "FailStateEntered"
+          },
+          {
+            "id": 9,
+            "previousEventId": 8,
+            "timestamp": "timestamp",
+            "type": "ParallelStateFailed"
+          },
+          {
+            "executionFailedEventDetails": {
+              "cause": "Cause description",
+              "error": "FailureType"
+            },
+            "id": 10,
+            "previousEventId": 9,
+            "timestamp": "timestamp",
+            "type": "ExecutionFailed"
+          }
+        ],
+        "ResponseMetadata": {
+          "HTTPHeaders": {},
+          "HTTPStatusCode": 200
+        }
+      }
+    }
   }
 }


Code review comments:
diff_hunk: @@ -1,41 +1,116 @@
-from typing import Final
+import copy
+import datetime
+import threading
+from typing import Final, Optional
 
+from localstack.aws.api.stepfunctions import ExecutionFailedEventDetails, HistoryEventType
+from localstack.services.stepfunctions.asl.component.common.error_name.custom_error_name import (
+    CustomErrorName,
+)
+from localstack.services.stepfunctions.asl.component.common.error_name.failure_event import (
+    FailureEvent,
+    FailureEventException,
+)
 from localstack.services.stepfunctions.asl.component.eval_component import EvalComponent
 from localstack.services.stepfunctions.asl.component.program.program import Program
-from localstack.services.stepfunctions.asl.eval.count_down_latch import CountDownLatch
+from localstack.services.stepfunctions.asl.component.state.state_execution.state_parallel.branch_worker import (
+    BranchWorker,
+)
 from localstack.services.stepfunctions.asl.eval.environment import Environment
-from localstack.services.stepfunctions.asl.eval.program_worker import ProgramWorker
+from localstack.services.stepfunctions.asl.eval.event.event_detail import EventDetails
+from localstack.services.stepfunctions.asl.eval.program_state import ProgramError, ProgramState
+from localstack.utils.collections import select_from_typed_dict
+
+
+class BranchWorkerPool(BranchWorker.BranchWorkerComm):
+    _mutex: Final[threading.Lock]
+    _termination_event: Final[threading.Event]
+    _active_workers_num: int
+
+    _terminated: bool
path: localstack/services/stepfunctions/asl/component/state/state_execution/state_parallel/branches_decl.py
text: @user1:
isn't this equivalent to checking if the `_termination_event` is set?

@author:
Indeed, thank you!

diff_hunk: @@ -0,0 +1,52 @@
+import abc
+import logging
+import threading
+from typing import Final, Optional
+
+from localstack.aws.api.stepfunctions import Timestamp
+from localstack.services.stepfunctions.asl.component.program.program import Program
+from localstack.services.stepfunctions.asl.eval.environment import Environment
+from localstack.utils.threads import TMP_THREADS
+
+LOG = logging.getLogger(__name__)
+
+
+class BranchWorker:
+    class BranchWorkerComm(abc.ABC):
+        @abc.abstractmethod
+        def on_terminated(self, env: Environment):
+            ...
+
+    _branch_worker_comm: Final[BranchWorkerComm]
+    _program: Final[Program]
+    _worker_thread: Optional[threading.Thread]
+    env: Final[Environment]
+
+    def __init__(self, branch_worker_comm: BranchWorkerComm, program: Program, env: Environment):
+        self._branch_worker_comm = branch_worker_comm
+        self._program = program
+        self.env = env
+        self._worker_thread = None
+
+    def _thread_routine(self) -> None:
+        LOG.info(f"[BranchWorker] [launched] [id: {self._worker_thread.native_id}]")
+        self._program.eval(self.env)
+        LOG.info(f"[BranchWorker] [terminated] [id: {self._worker_thread.native_id}]")
+        self._branch_worker_comm.on_terminated(env=self.env)
+
+    def start(self):
+        if self._worker_thread is not None:
+            raise RuntimeError(f"Attempted to rerun BranchWorker for program ${self._program}.")
+
+        self._worker_thread = threading.Thread(target=self._thread_routine)
path: localstack/services/stepfunctions/asl/component/state/state_execution/state_parallel/branch_worker.py
text: @user1:
Please make sure to always name your threads and thread pools. This practice helps us identify any potential leaks