localstack · MEPalma · Jul 14, 2023 · May 8, 2023 · May 10, 2023 · May 10, 2023
diff --git a/localstack/services/stepfunctions/asl/component/common/error_name/failure_event.py b/localstack/services/stepfunctions/asl/component/common/error_name/failure_event.py
@@ -1,6 +1,6 @@
 from typing import Final
 
-from localstack.aws.api.stepfunctions import HistoryEventType
+from localstack.aws.api.stepfunctions import ExecutionFailedEventDetails, HistoryEventType
 from localstack.services.stepfunctions.asl.component.common.error_name.error_name import ErrorName
 from localstack.services.stepfunctions.asl.eval.event.event_detail import EventDetails
 
@@ -16,3 +16,20 @@ def __init__(
         self.error_name = error_name
         self.event_type = event_type
         self.event_details = event_details
+
+
+class FailureEventException(Exception):
+    failure_event: Final[
8000
FailureEvent]
+
+    def __init__(self, failure_event: FailureEvent):
+        self.failure_event = failure_event
+
+    def get_execution_failed_event_details(self) -> ExecutionFailedEventDetails:
+        failure_event_spec = list(self.failure_event.event_details.values())[0]
+        execution_failed_event_details = ExecutionFailedEventDetails(
+            error=failure_event_spec.get("error")
+            or f"NoErrorSpecification in {failure_event_spec}",
+        )
+        if "cause" in failure_event_spec:
+            execution_failed_event_details["cause"] = failure_event_spec["cause"]
+        return execution_failed_event_details
diff --git a/localstack/services/stepfunctions/asl/component/program/program.py b/localstack/services/stepfunctions/asl/component/program/program.py
@@ -9,6 +9,9 @@
     HistoryEventType,
 )
 from localstack.services.stepfunctions.asl.component.common.comment import Comment
+from localstack.services.stepfunctions.asl.component.common.error_name.failure_event import (
+    FailureEventException,
+)
 from localstack.services.stepfunctions.asl.component.common.flow.start_at import StartAt
 from localstack.services.stepfunctions.asl.component.eval_component import EvalComponent
 from localstack.services.stepfunctions.asl.component.state.state import CommonStateField
@@ -46,8 +49,16 @@ def _eval_body(self, env: Environment) -> None:
             while env.is_running():
                 next_state: CommonStateField = self._get_state(env.next_state_name)
                 next_state.eval(env)
+        except FailureEventException as ex:
+            env.set_error(error=ex.get_execution_failed_event_details())
         except Exception as ex:
-            LOG.debug(f"Stepfunctions computation ended with exception '{ex}'.")
+            cause = f"{type(ex)}({str(ex)})"
+            LOG.error(f"Stepfunctions computation ended with exception '{cause}'.")
+            env.set_error(
+                ExecutionFailedEventDetails(
+                    error="Internal Error", cause=f"Internal Error due to '{cause}'"
+                )
+            )
 
         program_state: ProgramState = env.program_state()
         if isinstance(program_state, ProgramError):

diff --git a/localstack/services/stepfunctions/asl/component/state/state_execution/execute_state.py b/localstack/services/stepfunctions/asl/component/state/state_execution/execute_state.py
@@ -5,18 +5,15 @@
 from threading import Thread
 from typing import Any, Optional
 
-from localstack.aws.api.stepfunctions import (
-    ExecutionFailedEventDetails,
-    HistoryEventType,
-    TaskFailedEventDetails,
-)
+from localstack.aws.api.stepfunctions import HistoryEventType, TaskFailedEventDetails
 from localstack.services.stepfunctions.asl.component.common.catch.catch_decl import CatchDecl
 from localstack.services.stepfunctions.asl.component.common.catch.catch_outcome import (
     CatchOutcome,
     CatchOutcomeNotCaught,
 )
 from localstack.services.stepfunctions.asl.component.common.error_name.failure_event import (
     FailureEvent,
+    FailureEventException,
 )
 from localstack.services.stepfunctions.asl.component.common.error_name.states_error_name import (
     StatesErrorName,
@@ -126,13 +123,17 @@ def from_state_props(self, state_props: StateProps) -> None:
             self.heartbeat = heartbeat
 
     def _from_error(self, env: Environment, ex: Exception) -> FailureEvent:
-        LOG.warning("State Task executed generic failure event reporting logic.")
+        if isinstance(ex, FailureEventException):
+            return ex.failure_event
+        LOG.warning(
+            "State Task encountered an unhandled exception that lead to a State.Runtime error."
+        )
         return FailureEvent(
-            error_name=StatesErrorName(typ=StatesErrorNameType.StatesTaskFailed),
+            error_name=StatesErrorName(typ=StatesErrorNameType.StatesRuntime),
             event_type=HistoryEventType.TaskFailed,
             event_details=EventDetails(
                 taskFailedEventDetails=TaskFailedEventDetails(
-                    error="Unsupported Error Handling",
+                    error=StatesErrorNameType.StatesRuntime.to_name(),
                     cause=str(ex),
                 )
             ),
@@ -184,10 +185,7 @@ def _handle_uncaught(self, ex: Exception, env: Environment):
 
     @staticmethod
     def _terminate_with_event(failure_event: FailureEvent, env: Environment) -> None:
-        # Halt execution with the given failure event.
-        env.set_error(
-            ExecutionFailedEventDetails(**(list(failure_event.event_details.values())[0]))
-        )
+        raise FailureEventException(failure_event=failure_event)
 
     def _evaluate_with_timeout(self, env: Environment) -> None:
         self.timeout.eval(env=env)

diff --git a/...services/stepfunctions/asl/component/state/state_execution/state_task/service/resource.py b/...services/stepfunctions/asl/component/state/state_execution/state_task/service/resource.py
@@ -9,6 +9,7 @@
 
 class ResourceCondition(str):
     WaitForTaskToken = "waitForTaskToken"
+    Sync = "sync"
 
 
 class ResourceARN(TypedDict):
@@ -121,5 +122,7 @@ def __init__(
             match tail_parts[-1]:
                 case "waitForTaskToken":
                     self.condition = ResourceCondition.WaitForTaskToken
+                case "sync":
+                    self.condition = ResourceCondition.Sync
                 case unsupported:
                     raise RuntimeError(f"Unsupported condition '{unsupported}'.")
diff --git a/...tepfunctions/asl/component/state/state_execution/state_task/service/state_task_service.py b/...tepfunctions/asl/component/state/state_execution/state_task/service/state_task_service.py
@@ -65,6 +65,12 @@ def for_service(cls, service_name: str) -> StateTaskService:
                 )
 
                 return StateTaskServiceSqs()
+            case "states":
+                from localstack.services.stepfunctions.asl.component.state.state_execution.state_task.service.state_task_service_sfn import (
+                    StateTaskServiceSfn,
+                )
+
+                return StateTaskServiceSfn()
 
             case unknown:
                 raise NotImplementedError(f"Unsupported service: '{unknown}'.")  # noqa
diff --git a/...ions/asl/component/state/state_execution/state_task/service/state_task_service_aws_sdk.py b/...ions/asl/component/state/state_execution/state_task/service/state_task_service_aws_sdk.py
@@ -1,3 +1,4 @@
+from botocore.config import Config
 from botocore.exceptions import ClientError
 
 from localstack.aws.api.stepfunctions import HistoryEventType, TaskFailedEventDetails
@@ -15,7 +16,6 @@
     StateTaskServiceCallback,
 )
 from localstack.services.stepfunctions.asl.component.state.state_props import StateProps
-from localstack.services.stepfunctions.asl.eval.callback.callback import CallbackOutcomeFailureError
 from localstack.services.stepfunctions.asl.eval.environment import Environment
 from localstack.services.stepfunctions.asl.eval.event.event_detail import EventDetails
 from localstack.utils.aws import aws_stack
@@ -75,14 +75,10 @@ def _get_task_failure_event(self, error: str, cause: str) -> FailureEvent:
         )
 
     def _from_error(self, env: Environment, ex: Exception) -> FailureEvent:
-        if isinstance(ex, CallbackOutcomeFailureError):
-            return self._get_callback_outcome_failure_event(ex=ex)
-        if isinstance(ex, TimeoutError):
-            return self._get_timed_out_failure_event()
-
-        norm_service_name: str = self._normalise_service_name(self.resource.api_name)
-        error: str = self._normalise_exception_name(norm_service_name, ex)
         if isinstance(ex, ClientError):
+            norm_service_name: str = self._normalise_service_name(self.resource.api_name)
+            error: str = self._normalise_exception_name(norm_service_name, ex)
+
             error_message: str = ex.response["Error"]["Message"]
             cause_details = [
                 f"Service: {norm_service_name}",
@@ -97,14 +93,12 @@ def _from_error(self, env: Environment, ex: Exception) -> FailureEvent:
             cause: str = f"{error_message} ({', '.join(cause_details)})"
             failure_event = self._get_task_failure_event(error=error, cause=cause)
             return failure_event
-
-        failure_event = self._get_task_failure_event(
-            error=error, cause=str(ex)  # TODO: update cause decoration.
-        )
-        return failure_event
+        return super()._from_error(env=env, ex=ex)
 
     def _eval_service_task(self, env: Environment, parameters: dict) -> None:
-        api_client = aws_stack.create_external_boto_client(service_name=self._normalised_api_name)
+        api_client = aws_stack.create_external_boto_client(
+            service_name=self._normalised_api_name, config=Config(parameter_validation=False)
+        )
         response = getattr(api_client, self._normalised_api_action)(**parameters) or dict()
         if response:
             response.pop("ResponseMetadata", None)

diff --git a/...ons/asl/component/state/state_execution/state_task/service/state_task_service_callback.py b/...ons/asl/component/state/state_execution/state_task/service/state_task_service_callback.py
@@ -86,6 +86,11 @@ def _wait_for_task_token(self, env: Environment) -> None:  # noqa
         else:
             raise NotImplementedError(f"Unsupported CallbackOutcome type '{type(outcome)}'.")
 
+    def _sync(self, env: Environment) -> None:
+        raise RuntimeError(
+            f"Unsupported .sync callback procedure in resource {self.resource.resource_arn}"
+        )
+
     def _is_condition(self):
         return self.resource.condition is not None
 
@@ -105,6 +110,11 @@ def _get_callback_outcome_failure_event(self, ex: CallbackOutcomeFailureError) -
             ),
         )
 
+    def _from_error(self, env: Environment, ex: Exception) -> FailureEvent:
+        if isinstance(ex, CallbackOutcomeFailureError):
+            return self._get_callback_outcome_failure_event(ex=ex)
+        return super()._from_error(env=env, ex=ex)
+
     def _eval_execution(self, env: Environment) -> None:
         parameters = self._eval_parameters(env=env)
         parameters_str = to_json_str(parameters)
@@ -137,10 +147,11 @@ def _eval_execution(self, env: Environment) -> None:
             ),
         )
 
-        self._eval_service_task(env=env, parameters=parameters)
+        normalised_parameters = self._normalised_parameters_bindings(parameters)
+        self._eval_service_task(env=env, parameters=normalised_parameters)
 
         if self._is_condition():
-            output = env.stack.pop()
+            output = env.stack[-1]
             env.event_history.add_event(
                 hist_type_event=HistoryEventType.TaskSubmitted,
                 event_detail=EventDetails(
@@ -155,6 +166,8 @@ def _eval_execution(self, env: Environment) -> None:
             match self.resource.condition:
                 case ResourceCondition.WaitForTaskToken:
                     self._wait_for_task_token(env=env)
+                case ResourceCondition.Sync:
+                    self._sync(env=env)
                 case unsupported:
                     raise NotImplementedError(f"Unsupported callback type '{unsupported}'.")
 

diff --git a/...tions/asl/component/state/state_execution/state_task/service/state_task_service_lambda.py b/...tions/asl/component/state/state_execution/state_task/service/state_task_service_lambda.py
@@ -9,19 +9,12 @@
 from localstack.services.stepfunctions.asl.component.common.error_name.failure_event import (
     FailureEvent,
 )
-from localstack.services.stepfunctions.asl.component.common.error_name.states_error_name import (
-    StatesErrorName,
-)
-from localstack.services.stepfunctions.asl.component.common.error_name.states_error_name_type import (
-    StatesErrorNameType,
-)
 from localstack.services.stepfunctions.asl.component.state.state_execution.state_task import (
     lambda_eval_utils,
 )
 from localstack.services.stepfunctions.asl.component.state.state_execution.state_task.service.state_task_service_callback import (
     StateTaskServiceCallback,
 )
-from localstack.services.stepfunctions.asl.eval.callback.callback import CallbackOutcomeFailureError
 from localstack.services.stepfunctions.asl.eval.environment import Environment
 from localstack.services.stepfunctions.asl.eval.event.event_detail import EventDetails
 
@@ -60,11 +53,6 @@ def _error_cause_from_client_error(client_error: ClientError) -> tuple[str, str]
         return error, cause
 
     def _from_error(self, env: Environment, ex: Exception) -> FailureEvent:
-        if isinstance(ex, CallbackOutcomeFailureError):
-            return self._get_callback_outcome_failure_event(ex=ex)
-        if isinstance(ex, TimeoutError):
-            return self._get_timed_out_failure_event()
-
         if isinstance(ex, lambda_eval_utils.LambdaFunctionErrorException):
             error = "Exception"
             error_name = CustomErrorName(error)
@@ -73,10 +61,7 @@ def _from_error(self, env: Environment, ex: Exception) -> FailureEvent:
             error, cause = self._error_cause_from_client_error(ex)
             error_name = CustomErrorName(error)
         else:
-            error = "Exception"
-            error_name = StatesErrorName(typ=StatesErrorNameType.StatesTaskFailed)
-            cause = str(ex)
-
+            return super()._from_error(env=env, ex=ex)
         return FailureEvent(
             error_name=error_name,
             event_type=HistoryEventType.TaskFailed,