We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
bug: State Machine references don't get resolved properly
### Is there an existing issue for this?

- [X] I have searched the existing issues

### Current Behavior

Lambda refs get lost

### Expected Behavior

Lambda refs work in state machines

### How are you starting LocalStack?

With a docker-compose file

### Steps To Reproduce

#### How are you starting localstack (e.g., `bin/localstack` command, arguments, or `docker-compose.yml`)

    docker run localstack/localstack

#### Client commands (e.g., AWS SDK code snippet, or sequence of "awslocal" commands)

    awslocal s3 mb s3://mybucket


### Environment

```markdown
- OS: 
- LocalStack:
```


### Anything else?

This is based on a conversation I had with @dominikschubert 

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `localstack/utils/generic/wait_utils.py`
Content:
```
1 import time
2 from typing import Callable
3 
4 from typing_extensions import Literal
5 
6 
7 def wait_until(
8     fn: Callable[[], bool],
9     wait: float = 1.0,
10     max_retries: int = 10,
11     strategy: Literal["exponential", "static", "linear"] = "exponential",
12     _retries: int = 0,
13     _max_wait: float = 240,
14 ) -> None:
15     """waits until a given condition is true, rechecking it periodically"""
16     if max_retries < _retries:
17         return
18     completed = fn()
19     if not completed:
20         if wait > _max_wait:
21             return
22         time.sleep(wait)
23         next_wait = wait  # default: static
24         if strategy == "linear":
25             next_wait = (wait / _retries) * (_retries + 1)
26         elif strategy == "exponential":
27             next_wait = wait ** 2
28         wait_until(fn, next_wait, max_retries, strategy, _retries + 1, _max_wait)
29
```

Path: `localstack/services/cloudformation/models/stepfunctions.py`
Content:
```
1 from localstack.services.cloudformation.deployment_utils import PLACEHOLDER_RESOURCE_NAME
2 from localstack.services.cloudformation.service_models import GenericBaseModel
3 from localstack.utils.aws import aws_stack
4 
5 
6 class SFNActivity(GenericBaseModel):
7     @staticmethod
8     def cloudformation_type():
9         return "AWS::StepFunctions::Activity"
10 
11     def fetch_state(self, stack_name, resources):
12         activity_arn = self.physical_resource_id
13         if not activity_arn:
14             return None
15         client = aws_stack.connect_to_service("stepfunctions")
16         result = client.describe_activity(activityArn=activity_arn)
17         return result
18 
19     @staticmethod
20     def get_deploy_templates():
21         return {
22             "create": {
23                 "function": "create_activity",
24                 "parameters": {"name": ["Name", PLACEHOLDER_RESOURCE_NAME], "tags": "Tags"},
25             },
26             "delete": {
27                 "function": "delete_activity",
28                 "parameters": {"activityArn": "PhysicalResourceId"},
29             },
30         }
31 
32 
33 class SFNStateMachine(GenericBaseModel):
34     @staticmethod
35     def cloudformation_type():
36         return "AWS::StepFunctions::StateMachine"
37 
38     def get_resource_name(self):
39         return self.props.get("StateMachineName")
40 
41     def get_physical_resource_id(self, attribute=None, **kwargs):
42         return self.props.get("stateMachineArn")
43 
44     def fetch_state(self, stack_name, resources):
45         sm_name = self.props.get("StateMachineName") or self.resource_id
46         sm_name = self.resolve_refs_recursively(stack_name, sm_name, resources)
47         sfn_client = aws_stack.connect_to_service("stepfunctions")
48         state_machines = sfn_client.list_state_machines()["stateMachines"]
49         sm_arn = [m["stateMachineArn"] for m in state_machines if m["name"] == sm_name]
50         if not sm_arn:
51             return None
52         result = sfn_client.describe_state_machine(stateMachineArn=sm_arn[0])
53         return result
54 
55     def update_resource(self, new_resource, stack_name, resources):
56         props = new_resource["Properties"]
57         client = aws_stack.connect_to_service("stepfunctions")
58         sm_arn = self.props.get("stateMachineArn")
59         if not sm_arn:
60             self.state = self.fetch_state(stack_name=stack_name, resources=resources)
61             sm_arn = self.state["stateMachineArn"]
62         kwargs = {
63             "stateMachineArn": sm_arn,
64             "definition": props["DefinitionString"],
65         }
66         return client.update_state_machine(**kwargs)
67 
68     @staticmethod
69     def get_deploy_templates():
70         return {
71             "create": {
72                 "function": "create_state_machine",
73                 "parameters": {
74                     "name": ["StateMachineName", PLACEHOLDER_RESOURCE_NAME],
75                     "definition": "DefinitionString",
76                     "roleArn": "RoleArn",
77                 },
78             },
79             "delete": {
80                 "function": "delete_state_machine",
81                 "parameters": {"stateMachineArn": "PhysicalResourceId"},
82             },
83         }
84
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/localstack/services/cloudformation/models/stepfunctions.py b/localstack/services/cloudformation/models/stepfunctions.py
--- a/localstack/services/cloudformation/models/stepfunctions.py
+++ b/localstack/services/cloudformation/models/stepfunctions.py
@@ -1,3 +1,6 @@
+import re
+from typing import Dict
+
 from localstack.services.cloudformation.deployment_utils import PLACEHOLDER_RESOURCE_NAME
 from localstack.services.cloudformation.service_models import GenericBaseModel
 from localstack.utils.aws import aws_stack
@@ -65,19 +68,43 @@
         }
         return client.update_state_machine(**kwargs)
 
-    @staticmethod
-    def get_deploy_templates():
+    @classmethod
+    def get_deploy_templates(cls):
+        def _create_params(params, **kwargs):
+            def _get_definition(params):
+                definition_str = params.get("DefinitionString")
+                substitutions = params.get("DefinitionSubstitutions")
+                if substitutions is not None:
+                    definition_str = _apply_substitutions(definition_str, substitutions)
+                return definition_str
+
+            return {
+                "name": params.get("StateMachineName", PLACEHOLDER_RESOURCE_NAME),
+                "definition": _get_definition(params),
+                "roleArn": params.get("RoleArn"),
+                "type": params.get("StateMachineTyp", None),
+            }
+
         return {
             "create": {
                 "function": "create_state_machine",
-                "parameters": {
-                    "name": ["StateMachineName", PLACEHOLDER_RESOURCE_NAME],
-                    "definition": "DefinitionString",
-                    "roleArn": "RoleArn",
-                },
+                "parameters": _create_params,
             },
             "delete": {
                 "function": "delete_state_machine",
                 "parameters": {"stateMachineArn": "PhysicalResourceId"},
             },
         }
+
+
+def _apply_substitutions(definition: str, substitutions: Dict[str, str]) -> str:
+    substitution_regex = re.compile("\\${[a-zA-Z0-9_]+}")  # might be a bit too strict in some cases
+    tokens = substitution_regex.findall(definition)
+    result = definition
+    for token in tokens:
+        raw_token = token[2:-1]  # strip ${ and }
+        if raw_token not in substitutions.keys():
+            raise
+        result = result.replace(token, substitutions[raw_token])
+
+    return result
diff --git a/localstack/utils/generic/wait_utils.py b/localstack/utils/generic/wait_utils.py
--- a/localstack/utils/generic/wait_utils.py
+++ b/localstack/utils/generic/wait_utils.py
@@ -24,5 +24,5 @@
         if strategy == "linear":
             next_wait = (wait / _retries) * (_retries + 1)
         elif strategy == "exponential":
-            next_wait = wait ** 2
+            next_wait = wait * 2
         wait_until(fn, next_wait, max_retries, strategy, _retries + 1, _max_wait)

{"golden_diff": "diff --git a/localstack/services/cloudformation/models/stepfunctions.py b/localstack/services/cloudformation/models/stepfunctions.py\n--- a/localstack/services/cloudformation/models/stepfunctions.py\n+++ b/localstack/services/cloudformation/models/stepfunctions.py\n@@ -1,3 +1,6 @@\n+import re\n+from typing import Dict\n+\n from localstack.services.cloudformation.deployment_utils import PLACEHOLDER_RESOURCE_NAME\n from localstack.services.cloudformation.service_models import GenericBaseModel\n from localstack.utils.aws import aws_stack\n@@ -65,19 +68,43 @@\n         }\n         return client.update_state_machine(**kwargs)\n \n-    @staticmethod\n-    def get_deploy_templates():\n+    @classmethod\n+    def get_deploy_templates(cls):\n+        def _create_params(params, **kwargs):\n+            def _get_definition(params):\n+                definition_str = params.get(\"DefinitionString\")\n+                substitutions = params.get(\"DefinitionSubstitutions\")\n+                if substitutions is not None:\n+                    definition_str = _apply_substitutions(definition_str, substitutions)\n+                return definition_str\n+\n+            return {\n+                \"name\": params.get(\"StateMachineName\", PLACEHOLDER_RESOURCE_NAME),\n+                \"definition\": _get_definition(params),\n+                \"roleArn\": params.get(\"RoleArn\"),\n+                \"type\": params.get(\"StateMachineTyp\", None),\n+            }\n+\n         return {\n             \"create\": {\n                 \"function\": \"create_state_machine\",\n-                \"parameters\": {\n-                    \"name\": [\"StateMachineName\", PLACEHOLDER_RESOURCE_NAME],\n-                    \"definition\": \"DefinitionString\",\n-                    \"roleArn\": \"RoleArn\",\n-                },\n+                \"parameters\": _create_params,\n             },\n             \"delete\": {\n                 \"function\": \"delete_state_machine\",\n                 \"parameters\": {\"stateMachineArn\": \"PhysicalResourceId\"},\n             },\n         }\n+\n+\n+def _apply_substitutions(definition: str, substitutions: Dict[str, str]) -> str:\n+    substitution_regex = re.compile(\"\\\\${[a-zA-Z0-9_]+}\")  # might be a bit too strict in some cases\n+    tokens = substitution_regex.findall(definition)\n+    result = definition\n+    for token in tokens:\n+        raw_token = token[2:-1]  # strip ${ and }\n+        if raw_token not in substitutions.keys():\n+            raise\n+        result = result.replace(token, substitutions[raw_token])\n+\n+    return result\ndiff --git a/localstack/utils/generic/wait_utils.py b/localstack/utils/generic/wait_utils.py\n--- a/localstack/utils/generic/wait_utils.py\n+++ b/localstack/utils/generic/wait_utils.py\n@@ -24,5 +24,5 @@\n         if strategy == \"linear\":\n             next_wait = (wait / _retries) * (_retries + 1)\n         elif strategy == \"exponential\":\n-            next_wait = wait ** 2\n+            next_wait = wait * 2\n         wait_until(fn, next_wait, max_retries, strategy, _retries + 1, _max_wait)\n", "issue": "bug: State Machine references don't get resolved properly\n### Is there an existing issue for this?\n\n- [X] I have searched the existing issues\n\n### Current Behavior\n\nLambda refs get lost\n\n### Expected Behavior\n\nLambda refs work in state machines\n\n### How are you starting LocalStack?\n\nWith a docker-compose file\n\n### Steps To Reproduce\n\n#### How are you starting localstack (e.g., `bin/localstack` command, arguments, or `docker-compose.yml`)\r\n\r\n    docker run localstack/localstack\r\n\r\n#### Client commands (e.g., AWS SDK code snippet, or sequence of \"awslocal\" commands)\r\n\r\n    awslocal s3 mb s3://mybucket\r\n\n\n### Environment\n\n```markdown\n- OS: \r\n- LocalStack:\n```\n\n\n### Anything else?\n\nThis is based on a conversation I had with @dominikschubert \n", "before_files": [{"content": "import time\nfrom typing import Callable\n\nfrom typing_extensions import Literal\n\n\ndef wait_until(\n    fn: Callable[[], bool],\n    wait: float = 1.0,\n    max_retries: int = 10,\n    strategy: Literal[\"exponential\", \"static\", \"linear\"] = \"exponential\",\n    _retries: int = 0,\n    _max_wait: float = 240,\n) -> None:\n    \"\"\"waits until a given condition is true, rechecking it periodically\"\"\"\n    if max_retries < _retries:\n        return\n    completed = fn()\n    if not completed:\n        if wait > _max_wait:\n            return\n        time.sleep(wait)\n        next_wait = wait  # default: static\n        if strategy == \"linear\":\n            next_wait = (wait / _retries) * (_retries + 1)\n        elif strategy == \"exponential\":\n            next_wait = wait ** 2\n        wait_until(fn, next_wait, max_retries, strategy, _retries + 1, _max_wait)\n", "path": "localstack/utils/generic/wait_utils.py"}, {"content": "from localstack.services.cloudformation.deployment_utils import PLACEHOLDER_RESOURCE_NAME\nfrom localstack.services.cloudformation.service_models import GenericBaseModel\nfrom localstack.utils.aws import aws_stack\n\n\nclass SFNActivity(GenericBaseModel):\n    @staticmethod\n    def cloudformation_type():\n        return \"AWS::StepFunctions::Activity\"\n\n    def fetch_state(self, stack_name, resources):\n        activity_arn = self.physical_resource_id\n        if not activity_arn:\n            return None\n        client = aws_stack.connect_to_service(\"stepfunctions\")\n        result = client.describe_activity(activityArn=activity_arn)\n        return result\n\n    @staticmethod\n    def get_deploy_templates():\n        return {\n            \"create\": {\n                \"function\": \"create_activity\",\n                \"parameters\": {\"name\": [\"Name\", PLACEHOLDER_RESOURCE_NAME], \"tags\": \"Tags\"},\n            },\n            \"delete\": {\n                \"function\": \"delete_activity\",\n                \"parameters\": {\"activityArn\": \"PhysicalResourceId\"},\n            },\n        }\n\n\nclass SFNStateMachine(GenericBaseModel):\n    @staticmethod\n    def cloudformation_type():\n        return \"AWS::StepFunctions::StateMachine\"\n\n    def get_resource_name(self):\n        return self.props.get(\"StateMachineName\")\n\n    def get_physical_resource_id(self, attribute=None, **kwargs):\n        return self.props.get(\"stateMachineArn\")\n\n    def fetch_state(self, stack_name, resources):\n        sm_name = self.props.get(\"StateMachineName\") or self.resource_id\n        sm_name = self.resolve_refs_recursively(stack_name, sm_name, resources)\n        sfn_client = aws_stack.connect_to_service(\"stepfunctions\")\n        state_machines = sfn_client.list_state_machines()[\"stateMachines\"]\n        sm_arn = [m[\"stateMachineArn\"] for m in state_machines if m[\"name\"] == sm_name]\n        if not sm_arn:\n            return None\n        result = sfn_client.describe_state_machine(stateMachineArn=sm_arn[0])\n        return result\n\n    def update_resource(self, new_resource, stack_name, resources):\n        props = new_resource[\"Properties\"]\n        client = aws_stack.connect_to_service(\"stepfunctions\")\n        sm_arn = self.props.get(\"stateMachineArn\")\n        if not sm_arn:\n            self.state = self.fetch_state(stack_name=stack_name, resources=resources)\n            sm_arn = self.state[\"stateMachineArn\"]\n        kwargs = {\n            \"stateMachineArn\": sm_arn,\n            \"definition\": props[\"DefinitionString\"],\n        }\n        return client.update_state_machine(**kwargs)\n\n    @staticmethod\n    def get_deploy_templates():\n        return {\n            \"create\": {\n                \"function\": \"create_state_machine\",\n                \"parameters\": {\n                    \"name\": [\"StateMachineName\", PLACEHOLDER_RESOURCE_NAME],\n                    \"definition\": \"DefinitionString\",\n                    \"roleArn\": \"RoleArn\",\n                },\n            },\n            \"delete\": {\n                \"function\": \"delete_state_machine\",\n                \"parameters\": {\"stateMachineArn\": \"PhysicalResourceId\"},\n            },\n        }\n", "path": "localstack/services/cloudformation/models/stepfunctions.py"}], "after_files": [{"content": "import time\nfrom typing import Callable\n\nfrom typing_extensions import Literal\n\n\ndef wait_until(\n    fn: Callable[[], bool],\n    wait: float = 1.0,\n    max_retries: int = 10,\n    strategy: Literal[\"exponential\", \"static\", \"linear\"] = \"exponential\",\n    _retries: int = 0,\n    _max_wait: float = 240,\n) -> None:\n    \"\"\"waits until a given condition is true, rechecking it periodically\"\"\"\n    if max_retries < _retries:\n        return\n    completed = fn()\n    if not completed:\n        if wait > _max_wait:\n            return\n        time.sleep(wait)\n        next_wait = wait  # default: static\n        if strategy == \"linear\":\n            next_wait = (wait / _retries) * (_retries + 1)\n        elif strategy == \"exponential\":\n            next_wait = wait * 2\n        wait_until(fn, next_wait, max_retries, strategy, _retries + 1, _max_wait)\n", "path": "localstack/utils/generic/wait_utils.py"}, {"content": "import re\nfrom typing import Dict\n\nfrom localstack.services.cloudformation.deployment_utils import PLACEHOLDER_RESOURCE_NAME\nfrom localstack.services.cloudformation.service_models import GenericBaseModel\nfrom localstack.utils.aws import aws_stack\n\n\nclass SFNActivity(GenericBaseModel):\n    @staticmethod\n    def cloudformation_type():\n        return \"AWS::StepFunctions::Activity\"\n\n    def fetch_state(self, stack_name, resources):\n        activity_arn = self.physical_resource_id\n        if not activity_arn:\n            return None\n        client = aws_stack.connect_to_service(\"stepfunctions\")\n        result = client.describe_activity(activityArn=activity_arn)\n        return result\n\n    @staticmethod\n    def get_deploy_templates():\n        return {\n            \"create\": {\n                \"function\": \"create_activity\",\n                \"parameters\": {\"name\": [\"Name\", PLACEHOLDER_RESOURCE_NAME], \"tags\": \"Tags\"},\n            },\n            \"delete\": {\n                \"function\": \"delete_activity\",\n                \"parameters\": {\"activityArn\": \"PhysicalResourceId\"},\n            },\n        }\n\n\nclass SFNStateMachine(GenericBaseModel):\n    @staticmethod\n    def cloudformation_type():\n        return \"AWS::StepFunctions::StateMachine\"\n\n    def get_resource_name(self):\n        return self.props.get(\"StateMachineName\")\n\n    def get_physical_resource_id(self, attribute=None, **kwargs):\n        return self.props.get(\"stateMachineArn\")\n\n    def fetch_state(self, stack_name, resources):\n        sm_name = self.props.get(\"StateMachineName\") or self.resource_id\n        sm_name = self.resolve_refs_recursively(stack_name, sm_name, resources)\n        sfn_client = aws_stack.connect_to_service(\"stepfunctions\")\n        state_machines = sfn_client.list_state_machines()[\"stateMachines\"]\n        sm_arn = [m[\"stateMachineArn\"] for m in state_machines if m[\"name\"] == sm_name]\n        if not sm_arn:\n            return None\n        result = sfn_client.describe_state_machine(stateMachineArn=sm_arn[0])\n        return result\n\n    def update_resource(self, new_resource, stack_name, resources):\n        props = new_resource[\"Properties\"]\n        client = aws_stack.connect_to_service(\"stepfunctions\")\n        sm_arn = self.props.get(\"stateMachineArn\")\n        if not sm_arn:\n            self.state = self.fetch_state(stack_name=stack_name, resources=resources)\n            sm_arn = self.state[\"stateMachineArn\"]\n        kwargs = {\n            \"stateMachineArn\": sm_arn,\n            \"definition\": props[\"DefinitionString\"],\n        }\n        return client.update_state_machine(**kwargs)\n\n    @classmethod\n    def get_deploy_templates(cls):\n        def _create_params(params, **kwargs):\n            def _get_definition(params):\n                definition_str = params.get(\"DefinitionString\")\n                substitutions = params.get(\"DefinitionSubstitutions\")\n                if substitutions is not None:\n                    definition_str = _apply_substitutions(definition_str, substitutions)\n                return definition_str\n\n            return {\n                \"name\": params.get(\"StateMachineName\", PLACEHOLDER_RESOURCE_NAME),\n                \"definition\": _get_definition(params),\n                \"roleArn\": params.get(\"RoleArn\"),\n                \"type\": params.get(\"StateMachineTyp\", None),\n            }\n\n        return {\n            \"create\": {\n                \"function\": \"create_state_machine\",\n                \"parameters\": _create_params,\n            },\n            \"delete\": {\n                \"function\": \"delete_state_machine\",\n                \"parameters\": {\"stateMachineArn\": \"PhysicalResourceId\"},\n            },\n        }\n\n\ndef _apply_substitutions(definition: str, substitutions: Dict[str, str]) -> str:\n    substitution_regex = re.compile(\"\\\\${[a-zA-Z0-9_]+}\")  # might be a bit too strict in some cases\n    tokens = substitution_regex.findall(definition)\n    result = definition\n    for token in tokens:\n        raw_token = token[2:-1]  # strip ${ and }\n        if raw_token not in substitutions.keys():\n            raise\n        result = result.replace(token, substitutions[raw_token])\n\n    return result\n", "path": "localstack/services/cloudformation/models/stepfunctions.py"}]}