Repository: RasaHQ/rasa
Pull request: Tracker fallback
**Proposed changes**:
- fix https://github.com/RasaHQ/rasa/issues/3243
- add `FailSafeTrackerStore` which wraps the actual tracker store and falls back to a `InMemoryTrackerStore` in case there is any problem
- use `InMemoryTrackerStore` in case there is a problem during the initialization of the tracker store

**Status (please check what you already did)**:
- [x] made PR ready for code review
- [x] added some tests for the functionality
- [ ] updated the documentation
- [x] updated the changelog
- [x] reformat files using `black` (please check [Readme](https://github.com/RasaHQ/rasa_nlu#code-style) for instructions)


Problem statement:
Fallback to InMemory tracker store if mongo / sql / redis is not reachable
If you defined an external tracker store, and the tracker store is not reachable, Rasa should automatically fallback to use the InmemoryTrackerStore. Otherwise this leads to weird behavior.

Hints from issue discussion:
This issue has been automatically marked as stale because it has not had recent activity. It will be closed if no further activity occurs. Thank you for your contributions.

@akelad Should we just do this at the initial connection attempt or also when the tracker store crashes somewhere in between? As far as I remember this should also handle the second case.
Never mind, nearly through with the production code impl

Commit message: add types

Patch to review:
diff --git a/CHANGELOG.rst b/CHANGELOG.rst
index 183835a6d36b..f69a0c3f21f9 100644
--- a/CHANGELOG.rst
+++ b/CHANGELOG.rst
@@ -14,6 +14,8 @@ This project adheres to `Semantic Versioning`_ starting with version 1.0.
 Added
 -----
 - Added the KeywordIntentClassifier
+- Fall back to ``InMemoryTrackerStore`` in case there is any problem with the current
+  tracker store
 
 Changed
 -------
diff --git a/rasa/cli/x.py b/rasa/cli/x.py
index d49ab2224895..d24ea01fc6ef 100644
--- a/rasa/cli/x.py
+++ b/rasa/cli/x.py
@@ -347,7 +347,7 @@ def run_in_production(args: argparse.Namespace):
 
 
 def _get_credentials_and_endpoints_paths(
-    args: argparse.Namespace
+    args: argparse.Namespace,
 ) -> Tuple[Optional[Text], Optional[Text]]:
     config_endpoint = args.config_endpoint
     if config_endpoint:
diff --git a/rasa/core/agent.py b/rasa/core/agent.py
index 4d42593adef9..1e40f3d3b688 100644
--- a/rasa/core/agent.py
+++ b/rasa/core/agent.py
@@ -25,7 +25,11 @@
 from rasa.core.policies.memoization import MemoizationPolicy
 from rasa.core.policies.policy import Policy
 from rasa.core.processor import MessageProcessor
-from rasa.core.tracker_store import InMemoryTrackerStore, TrackerStore
+from rasa.core.tracker_store import (
+    InMemoryTrackerStore,
+    TrackerStore,
+    FailSafeTrackerStore,
+)
 from rasa.core.trackers import DialogueStateTracker
 from rasa.exceptions import ModelNotFound
 from rasa.importers.importer import TrainingDataImporter
@@ -827,6 +831,14 @@ def create_processor(
             message_preprocessor=preprocessor,
         )
 
+    def on_tracker_store_error(self, error: Exception) -> None:
+        logger.error(
+            f"Error happened when trying to save conversation tracker to "
+            f"'{self.tracker_store.__class__.__name__}'. Falling back to use "
+            f"the '{InMemoryTrackerStore.__name__}'. Please "
+            f"investigate the following error: {error}."
+        )
+
     @staticmethod
     def _create_domain(domain: Union[Domain, Text]) -> Domain:
 
@@ -845,15 +857,16 @@ def _create_domain(domain: Union[Domain, Text]) -> Domain:
                 "type '{}' with value '{}'".format(type(domain), domain)
             )
 
-    @staticmethod
     def create_tracker_store(
-        store: Optional[TrackerStore], domain: Domain
+        self, store: Optional[TrackerStore], domain: Domain
     ) -> TrackerStore:
         if store is not None:
             store.domain = domain
-            return store
+            tracker_store = store
         else:
-            return InMemoryTrackerStore(domain)
+            tracker_store = InMemoryTrackerStore(domain)
+
+        return FailSafeTrackerStore(tracker_store, self.on_tracker_store_error)
 
     @staticmethod
     def _create_lock_store(store: Optional[LockStore]) -> LockStore:
diff --git a/rasa/core/processor.py b/rasa/core/processor.py
index fcb41505ee12..7f759d9a2f88 100644
--- a/rasa/core/processor.py
+++ b/rasa/core/processor.py
@@ -578,7 +578,7 @@ def _get_tracker(self, sender_id: Text) -> Optional[DialogueStateTracker]:
         sender_id = sender_id or UserMessage.DEFAULT_SENDER_ID
         return self.tracker_store.get_or_create_tracker(sender_id)
 
-    def _save_tracker(self, tracker):
+    def _save_tracker(self, tracker: DialogueStateTracker) -> None:
         self.tracker_store.save(tracker)
 
     def _prob_array_for_action(
diff --git a/rasa/core/test.py b/rasa/core/test.py
index ef57b3645744..b4c7f1e9d441 100644
--- a/rasa/core/test.py
+++ b/rasa/core/test.py
@@ -587,7 +587,6 @@ async def compare_models_in_dir(
     model_dir: Text, stories_file: Text, output: Text
 ) -> None:
     """Evaluates multiple trained models in a directory on a test set."""
-    from rasa.core import utils
     import rasa.utils.io as io_utils
 
     number_correct = defaultdict(list)
@@ -615,7 +614,6 @@ async def compare_models_in_dir(
 
 async def compare_models(models: List[Text], stories_file: Text, output: Text) -> None:
     """Evaluates provided trained models on a test set."""
-    from rasa.core import utils
 
     number_correct = defaultdict(list)
 
diff --git a/rasa/core/tracker_store.py b/rasa/core/tracker_store.py
index 5840091bf8f8..942b39ae8dd0 100644
--- a/rasa/core/tracker_store.py
+++ b/rasa/core/tracker_store.py
@@ -5,7 +5,7 @@
 import pickle
 import typing
 from datetime import datetime, timezone
-from typing import Iterator, Optional, Text, Iterable, Union, Dict
+from typing import Iterator, Optional, Text, Iterable, Union, Dict, Callable
 
 import itertools
 from boto3.dynamodb.conditions import Key
@@ -48,9 +48,33 @@ def find_tracker_store(
         event_broker: Optional[EventChannel] = None,
     ) -> "TrackerStore":
         """Returns the tracker_store type"""
-        if store is None or store.type is None:
-            tracker_store = InMemoryTrackerStore(domain, event_broker=event_broker)
-        elif store.type.lower() == "redis":
+
+        tracker_store = None
+        if store is not None and store.type is not None:
+            try:
+                tracker_store = TrackerStore._create_tracker_store(
+                    domain, store, event_broker
+                )
+            except Exception as e:
+                logger.error(
+                    f"Error when trying to connect to '{store.type}' "
+                    f"tracker store. Using "
+                    f"'{InMemoryTrackerStore.__name__}'' instead. "
+                    f"The causing error was: {e}."
+                )
+
+        if not tracker_store:
+            tracker_store = InMemoryTrackerStore(domain, event_broker)
+
+        logger.debug("Connected to {}.".format(tracker_store.__class__.__name__))
+
+        return tracker_store
+
+    @staticmethod
+    def _create_tracker_store(
+        domain: Domain, store: EndpointConfig, event_broker: Optional[EventChannel]
+    ) -> "TrackerStore":
+        if store.type.lower() == "redis":
             tracker_store = RedisTrackerStore(
                 domain=domain, host=store.url, event_broker=event_broker, **store.kwargs
             )
@@ -71,7 +95,6 @@ def find_tracker_store(
                 domain, store, event_broker
             )
 
-        logger.debug("Connected to {}.".format(tracker_store.__class__.__name__))
         return tracker_store
 
     @staticmethod
@@ -751,3 +774,58 @@ def _additional_events(
         )
 
         return itertools.islice(tracker.events, n_events, len(tracker.events))
+
+
+class FailSafeTrackerStore(TrackerStore):
+    """Wraps a tracker store so that we can fallback to a different tracker store in
+    case of errors."""
+
+    def __init__(
+        self,
+        tracker_store: TrackerStore,
+        on_tracker_store_error: Callable[[Exception], None],
+        fallback_tracker_store: Optional[TrackerStore] = None,
+    ) -> None:
+        """Create a `FailSafeTrackerStore`.
+
+        Args:
+            tracker_store: Primary tracker store.
+            on_tracker_store_error: Callback which is called when there is an error
+                in the primary tracker store.
+        """
+
+        super().__init__(tracker_store.domain, tracker_store.event_broker)
+
+        self._fallback_tracker_store: Optional[TrackerStore] = fallback_tracker_store
+        self._tracker_store = tracker_store
+        self.on_tracker_store_error = on_tracker_store_error
+
+    @property
+    def fallback_tracker_store(self) -> TrackerStore:
+        if not self._fallback_tracker_store:
+            self._fallback_tracker_store = InMemoryTrackerStore(
+                self._tracker_store.domain, self._tracker_store.event_broker
+            )
+
+        return self._fallback_tracker_store
+
+    def retrieve(self, sender_id: Text) -> Optional[DialogueStateTracker]:
+        try:
+            return self._tracker_store.retrieve(sender_id)
+        except Exception as e:
+            self.on_tracker_store_error(e)
+            return None
+
+    def keys(self) -> Iterable[Text]:
+        try:
+            return self._tracker_store.keys()
+        except Exception as e:
+            self.on_tracker_store_error(e)
+            return []
+
+    def save(self, tracker: DialogueStateTracker) -> None:
+        try:
+            self._tracker_store.save(tracker)
+        except Exception as e:
+            self.on_tracker_store_error(e)
+            self.fallback_tracker_store.save(tracker)
diff --git a/tests/core/test_tracker_stores.py b/tests/core/test_tracker_stores.py
index df927d83dce2..98f3c138febf 100644
--- a/tests/core/test_tracker_stores.py
+++ b/tests/core/test_tracker_stores.py
@@ -1,9 +1,11 @@
 import logging
 import tempfile
-from typing import Tuple
+from typing import Tuple, Text
+from unittest.mock import Mock
 
 import pytest
 from _pytest.logging import LogCaptureFixture
+from _pytest.monkeypatch import MonkeyPatch
 from moto import mock_dynamodb2
 
 from rasa.core.channels.channel import UserMessage
@@ -15,6 +17,7 @@
     RedisTrackerStore,
     SQLTrackerStore,
     DynamoTrackerStore,
+    FailSafeTrackerStore,
 )
 import rasa.core.tracker_store
 from rasa.core.trackers import DialogueStateTracker
@@ -50,7 +53,7 @@ def test_dynamo_get_or_create():
     get_or_create_tracker_store(DynamoTrackerStore(domain))
 
 
-def test_restart_after_retrieval_from_tracker_store(default_domain):
+def test_restart_after_retrieval_from_tracker_store(default_domain: Domain):
     store = InMemoryTrackerStore(default_domain)
     tr = store.get_or_create_tracker("myuser")
     synth = [ActionExecuted("action_listen") for _ in range(4)]
@@ -67,7 +70,7 @@ def test_restart_after_retrieval_from_tracker_store(default_domain):
     assert latest_restart == latest_restart_after_loading
 
 
-def test_tracker_store_remembers_max_history(default_domain):
+def test_tracker_store_remembers_max_history(default_domain: Domain):
     store = InMemoryTrackerStore(default_domain)
     tr = store.get_or_create_tracker("myuser", max_event_history=42)
     tr.update(Restarted())
@@ -92,7 +95,7 @@ def test_tracker_store_endpoint_config_loading():
     )
 
 
-def test_find_tracker_store(default_domain):
+def test_find_tracker_store(default_domain: Domain):
     store = read_endpoint_config(DEFAULT_ENDPOINTS_FILE, "tracker_store")
     tracker_store = RedisTrackerStore(
         domain=default_domain,
@@ -108,6 +111,17 @@ def test_find_tracker_store(default_domain):
     )
 
 
+def test_find_tracker_store(default_domain: Domain, monkeypatch: MonkeyPatch):
+    store = read_endpoint_config(DEFAULT_ENDPOINTS_FILE, "tracker_store")
+    mock = Mock(side_effect=Exception("ignore this"))
+    monkeypatch.setattr(rasa.core.tracker_store, "RedisTrackerStore", mock)
+
+    assert isinstance(
+        InMemoryTrackerStore(domain),
+        type(TrackerStore.find_tracker_store(default_domain, store)),
+    )
+
+
 class ExampleTrackerStore(RedisTrackerStore):
     def __init__(self, domain, url, port, db, password, record_exp, event_broker=None):
         super(ExampleTrackerStore, self).__init__(
@@ -121,7 +135,7 @@ def __init__(self, domain, url, port, db, password, record_exp, event_broker=Non
         )
 
 
-def test_tracker_store_from_string(default_domain):
+def test_tracker_store_from_string(default_domain: Domain):
     endpoints_path = "data/test_endpoints/custom_tracker_endpoints.yml"
     store_config = read_endpoint_config(endpoints_path, "tracker_store")
 
@@ -130,7 +144,7 @@ def test_tracker_store_from_string(default_domain):
     assert isinstance(tracker_store, ExampleTrackerStore)
 
 
-def test_tracker_store_from_invalid_module(default_domain):
+def test_tracker_store_from_invalid_module(default_domain: Domain):
     endpoints_path = "data/test_endpoints/custom_tracker_endpoints.yml"
     store_config = read_endpoint_config(endpoints_path, "tracker_store")
     store_config.type = "a.module.which.cannot.be.found"
@@ -140,7 +154,7 @@ def test_tracker_store_from_invalid_module(default_domain):
     assert isinstance(tracker_store, InMemoryTrackerStore)
 
 
-def test_tracker_store_from_invalid_string(default_domain):
+def test_tracker_store_from_invalid_string(default_domain: Domain):
     endpoints_path = "data/test_endpoints/custom_tracker_endpoints.yml"
     store_config = read_endpoint_config(endpoints_path, "tracker_store")
     store_config.type = "any string"
@@ -206,7 +220,7 @@ def pickle_serialise_tracker(_tracker):
         "postgresql://user:secret@localhost",
     ],
 )
-def test_get_db_url_with_fully_specified_url(full_url):
+def test_get_db_url_with_fully_specified_url(full_url: Text):
     assert SQLTrackerStore.get_db_url(host=full_url) == full_url
 
 
@@ -284,3 +298,71 @@ def test_db_url_with_query_from_endpoint_config():
             itertools.permutations(("another=query", "driver=my-driver"))
         )
     )
+
+
+def test_fail_safe_tracker_store_if_no_errors():
+    mocked_tracker_store = Mock()
+
+    tracker_store = FailSafeTrackerStore(mocked_tracker_store, None)
+
+    # test save
+    mocked_tracker_store.save = Mock()
+    tracker_store.save(None)
+    mocked_tracker_store.save.assert_called_once()
+
+    # test retrieve
+    expected = [1]
+    mocked_tracker_store.retrieve = Mock(return_value=expected)
+    sender_id = "10"
+    assert tracker_store.retrieve(sender_id) == expected
+    mocked_tracker_store.retrieve.assert_called_once_with(sender_id)
+
+    # test keys
+    expected = ["sender 1", "sender 2"]
+    mocked_tracker_store.keys = Mock(return_value=expected)
+    assert tracker_store.keys() == expected
+    mocked_tracker_store.keys.assert_called_once()
+
+
+def test_fail_safe_tracker_store_with_save_error():
+    mocked_tracker_store = Mock()
+    mocked_tracker_store.save = Mock(side_effect=Exception())
+
+    fallback_tracker_store = Mock()
+    fallback_tracker_store.save = Mock()
+
+    on_error_callback = Mock()
+
+    tracker_store = FailSafeTrackerStore(
+        mocked_tracker_store, on_error_callback, fallback_tracker_store
+    )
+    tracker_store.save(None)
+
+    fallback_tracker_store.save.assert_called_once()
+    on_error_callback.assert_called_once()
+
+
+def test_fail_safe_tracker_store_with_keys_error():
+    mocked_tracker_store = Mock()
+    mocked_tracker_store.keys = Mock(side_effect=Exception())
+
+    on_error_callback = Mock()
+
+    tracker_store = FailSafeTrackerStore(mocked_tracker_store, on_error_callback)
+    assert tracker_store.keys() == []
+    on_error_callback.assert_called_once()
+
+
+def test_fail_safe_tracker_store_with_retrieve_error():
+    mocked_tracker_store = Mock()
+    mocked_tracker_store.retrieve = Mock(side_effect=Exception())
+
+    fallback_tracker_store = Mock()
+    on_error_callback = Mock()
+
+    tracker_store = FailSafeTrackerStore(
+        mocked_tracker_store, on_error_callback, fallback_tracker_store
+    )
+
+    assert tracker_store.retrieve("sender_id") is None
+    on_error_callback.assert_called_once()


Code review comments:
diff_hunk: @@ -827,6 +831,14 @@ def create_processor(
             message_preprocessor=preprocessor,
         )
 
+    def on_tracker_store_error(self, error: Exception) -> None:
path: rasa/core/agent.py
text: @user1:
Could this be placed in `FailSafeTrackerStore` instead? And instead of making `FailSafeTrackerStore.__init__` receive an `on_tracker_store_error` parameter, simply allow users to subclass it and override the method with their own error callbacks.

@user1:
I think doing that will allow us to reduce the number of changes to `Agent`, and simplify `FailSafeTrackerStore` a bit.

@author:
I prefer having the option to change the behavior on runtime. However, how about providing a default value for it?