Repository: RasaHQ/rasa
Pull request: Ticket lock
**Proposed changes**:
- fixes #3925

**Status (please check what you already did)**:
- [x] made PR ready for code review
- [x] added some tests for the functionality
- [x] 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:
RedisTrackerStore save data in non-binary format
The redis tracker store currently stores the conversation in a binary format since the tracker object is pickled. Storing this in a human readable format (json) would make the data

* easier reusable (for analyzing, extracting data)
* less error prone

Hints from issue discussion:

Commit message: lint

Patch to review:
diff --git a/rasa/core/agent.py b/rasa/core/agent.py
index 3783bd47092e..ac945f0dbdb1 100644
--- a/rasa/core/agent.py
+++ b/rasa/core/agent.py
@@ -4,10 +4,10 @@
 import tempfile
 import uuid
 from asyncio import CancelledError
-from sanic import Sanic
 from typing import Any, Callable, Dict, List, Optional, Text, Tuple, Union
 
 import aiohttp
+from sanic import Sanic
 
 import rasa
 import rasa.utils.io
@@ -23,15 +23,16 @@
 from rasa.core.domain import Domain, InvalidDomain
 from rasa.core.exceptions import AgentNotReady
 from rasa.core.interpreter import NaturalLanguageInterpreter, RegexInterpreter
+from rasa.core.lock_store import LockStore, InMemoryLockStore
 from rasa.core.nlg import NaturalLanguageGenerator
-from rasa.core.policies.policy import Policy
-from rasa.core.policies.form_policy import FormPolicy
 from rasa.core.policies.ensemble import PolicyEnsemble, SimplePolicyEnsemble
+from rasa.core.policies.form_policy import FormPolicy
 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.trackers import DialogueStateTracker
-from rasa.core.utils import LockCounter
+from rasa.exceptions import ModelNotFound
 from rasa.model import (
     get_model_subdirectories,
     get_latest_model,
@@ -41,7 +42,6 @@
 from rasa.nlu.utils import is_url
 from rasa.utils.common import update_sanic_log_level, set_log_level
 from rasa.utils.endpoints import EndpointConfig
-from rasa.exceptions import ModelNotFound
 
 logger = logging.getLogger(__name__)
 
@@ -224,6 +224,7 @@ async def load_agent(
     interpreter: Optional[NaturalLanguageInterpreter] = None,
     generator: Union[EndpointConfig, NaturalLanguageGenerator] = None,
     tracker_store: Optional[TrackerStore] = None,
+    lock_store: Optional[LockStore] = None,
     action_endpoint: Optional[EndpointConfig] = None,
 ):
     try:
@@ -233,6 +234,7 @@ async def load_agent(
                     interpreter=interpreter,
                     generator=generator,
                     tracker_store=tracker_store,
+                    lock_store=lock_store,
                     action_endpoint=action_endpoint,
                     model_server=model_server,
                     remote_storage=remote_storage,
@@ -285,6 +287,7 @@ def __init__(
         interpreter: Optional[NaturalLanguageInterpreter] = None,
         generator: Union[EndpointConfig, NaturalLanguageGenerator, None] = None,
         tracker_store: Optional[TrackerStore] = None,
+        lock_store: Optional[LockStore] = None,
         action_endpoint: Optional[EndpointConfig] = None,
         fingerprint: Optional[Text] = None,
         model_directory: Optional[Text] = None,
@@ -306,8 +309,8 @@ def __init__(
 
         self.nlg = NaturalLanguageGenerator.create(generator, self.domain)
         self.tracker_store = self.create_tracker_store(tracker_store, self.domain)
+        self.lock_store = self.create_lock_store(lock_store)
         self.action_endpoint = action_endpoint
-        self.conversations_in_processing = {}
 
         self._set_fingerprint(fingerprint)
         self.model_directory = model_directory
@@ -344,6 +347,7 @@ def load(
         interpreter: Optional[NaturalLanguageInterpreter] = None,
         generator: Union[EndpointConfig, NaturalLanguageGenerator] = None,
         tracker_store: Optional[TrackerStore] = None,
+        lock_store: Optional[LockStore] = None,
         action_endpoint: Optional[EndpointConfig] = None,
         model_server: Optional[EndpointConfig] = None,
         remote_storage: Optional[Text] = None,
@@ -386,6 +390,7 @@ def load(
             interpreter=interpreter,
             generator=generator,
             tracker_store=tracker_store,
+            lock_store=lock_store,
             action_endpoint=action_endpoint,
             model_directory=model_path,
             model_server=model_server,
@@ -427,34 +432,12 @@ def noop(_):
 
         processor = self.create_processor(message_preprocessor)
 
-        # get the lock for the current conversation
-        lock = self.conversations_in_processing.get(message.sender_id)
-        if not lock:
-            logger.debug(
-                "Created a new lock for conversation '{}'".format(message.sender_id)
-            )
-            lock = LockCounter()
-            self.conversations_in_processing[message.sender_id] = lock
-
+        ticket = self.lock_store.issue_ticket(message.sender_id)
         try:
-            async with lock:
-                # this makes sure that there can always only be one coroutine
-                # handling a conversation at any point in time
-                # Note: this doesn't support multi-processing, it just works
-                # for coroutines. If there are multiple processes handling
-                # messages, an external system needs to make sure messages
-                # for the same conversation are always processed by the same
-                # process.
+            async with await self.lock_store.lock(message.sender_id, ticket):
                 return await processor.handle_message(message)
         finally:
-            if not lock.is_someone_waiting():
-                # dispose of the lock if no one needs it to avoid
-                # accumulating locks
-                del self.conversations_in_processing[message.sender_id]
-                logger.debug(
-                    "Deleted lock for conversation '{}' (unused)"
-                    "".format(message.sender_id)
-                )
+            self.lock_store.cleanup(message.sender_id, ticket)
 
     # noinspection PyUnusedLocal
     def predict_next(self, sender_id: Text, **kwargs: Any) -> Optional[Dict[Text, Any]]:
@@ -842,6 +825,13 @@ def create_tracker_store(
         else:
             return InMemoryTrackerStore(domain)
 
+    @staticmethod
+    def create_lock_store(store: Optional[LockStore]) -> LockStore:
+        if store is not None:
+            return store
+        else:
+            return InMemoryLockStore()
+
     @staticmethod
     def _create_ensemble(
         policies: Union[List[Policy], PolicyEnsemble, None]
@@ -857,7 +847,7 @@ def _create_ensemble(
             raise ValueError(
                 "Invalid param `policies`. Passed object is "
                 "of type '{}', but should be policy, an array of "
-                "policies, or a policy ensemble".format(passed_type)
+                "policies, or a policy ensemble.".format(passed_type)
             )
 
     @staticmethod
@@ -866,6 +856,7 @@ def load_local_model(
         interpreter: Optional[NaturalLanguageInterpreter] = None,
         generator: Union[EndpointConfig, NaturalLanguageGenerator] = None,
         tracker_store: Optional[TrackerStore] = None,
+        lock_store: Optional[LockStore] = None,
         action_endpoint: Optional[EndpointConfig] = None,
         model_server: Optional[EndpointConfig] = None,
         remote_storage: Optional[Text] = None,
@@ -887,6 +878,7 @@ def load_local_model(
             interpreter=interpreter,
             generator=generator,
             tracker_store=tracker_store,
+            lock_store=lock_store,
             action_endpoint=action_endpoint,
             model_server=model_server,
             remote_storage=remote_storage,
diff --git a/rasa/core/lock.py b/rasa/core/lock.py
new file mode 100644
index 000000000000..2c43f0803a78
--- /dev/null
+++ b/rasa/core/lock.py
@@ -0,0 +1,141 @@
+import logging
+import typing
+from collections import deque
+from typing import Text, Optional, Union, Deque
+
+import time
+
+if typing.TYPE_CHECKING:
+    pass
+
+logger = logging.getLogger(__name__)
+
+
+class Ticket:
+    def __init__(self, number: int, expires: float):
+        self.number = number
+        self.expires = expires
+
+    def has_expired(self):
+        return time.time() > self.expires
+
+    def __repr__(self):
+        return "Ticket(number: {}, expires: {})".format(self.number, self.expires)
+
+
+class TicketLock(object):
+    def __init__(
+        self, conversation_id: Text, tickets: Optional[Deque[Ticket]] = None
+    ) -> None:
+        self.conversation_id = conversation_id
+        self.tickets = tickets or deque()
+
+    async def __aenter__(self):
+        return self
+
+    async def __aexit__(self, exc_type, exc, tb):
+        pass
+
+    def is_locked(self, ticket_number: int) -> bool:
+        """Return whether `ticket_number` is locked.
+
+        Returns False if lock has expired.
+        Otherwise returns True if `now_serving` is not equal to `ticket`.
+        """
+
+        return self.now_serving != ticket_number
+
+    def issue_ticket(self, lifetime: Union[float, int]) -> int:
+        """Issue a new ticket and return its number."""
+
+        self.remove_expired_tickets()
+        number = self.last_issued + 1
+        ticket = Ticket(number, time.time() + lifetime)
+        self.tickets.append(ticket)
+
+        return number
+
+    def remove_expired_tickets(self) -> None:
+        """Remove expired tickets."""
+
+        # iterate over copy of self.tickets so we can remove items
+        for ticket in list(self.tickets):
+            if ticket.has_expired():
+                self.tickets.remove(ticket)
+
+    @property
+    def last_issued(self) -> int:
+        """Return number of the ticket that was last added.
+
+        Return -1 if no tickets exist.
+        """
+
+        ticket_number = self._ticket_number_for_index(-1)
+        if ticket_number is not None:
+            return ticket_number
+
+        return -1
+
+    @property
+    def now_serving(self) -> Optional[int]:
+        """Return number of the ticket to be served next.
+
+        Return 0 if no tickets exists.
+        """
+
+        ticket_number = self._ticket_number_for_index(0)
+        if ticket_number is not None:
+            return ticket_number
+
+        return 0
+
+    def _ticket_number_for_index(self, idx: int) -> Optional[int]:
+        """Return ticket number for `idx`.
+
+        Return None if there are no tickets, or if `idx` is out of bounds of
+        `self.tickets`.
+        """
+
+        self.remove_expired_tickets()
+
+        if not self.tickets or len(self.tickets) < abs(idx):
+            return None
+
+        return self.tickets[idx].number
+
+    def _ticket_for_ticket_number(self, ticket_number: int) -> Optional[Ticket]:
+        """Return expiration time for `ticket_number`."""
+
+        self.remove_expired_tickets()
+
+        return next((t for t in self.tickets if t.number == ticket_number), None)
+
+    def is_someone_waiting(self) -> bool:
+        """Return whether someone is waiting for the lock to become available.
+
+        Returns True if the ticket queue has length greater than 0.
+        """
+
+        return len(self.tickets) > 0
+
+    def remove_ticket_for_ticket_number(self, ticket_number: int) -> None:
+        """Return whether ticket for `ticket_number` has expired.
+
+        Return True if ticket was not found.
+        """
+
+        ticket = self._ticket_for_ticket_number(ticket_number)
+        if ticket:
+            self.tickets.remove(ticket)
+
+    def has_lock_expired(self, ticket_number: int) -> Optional[bool]:
+        """Return whether ticket for `ticket_number` has expired.
+
+        Return True if ticket was not found.
+        """
+
+        ticket = self._ticket_for_ticket_number(ticket_number)
+        if ticket:
+            return ticket.has_expired()
+
+        return True
diff --git a/rasa/core/lock_store.py b/rasa/core/lock_store.py
new file mode 100644
index 000000000000..7c80a60fdb55
--- /dev/null
+++ b/rasa/core/lock_store.py
@@ -0,0 +1,210 @@
+import asyncio
+import logging
+import pickle
+import typing
+from typing import Text, Optional, Dict, Union
+
+from rasa.core.lock import TicketLock
+
+if typing.TYPE_CHECKING:
+    pass
+
+logger = logging.getLogger(__name__)
+
+ACCEPTED_LOCK_STORES = ["in_memory", "redis"]
+
+
+class LockError(Exception):
+    pass
+
+
+class LockStore(object):
+    def __init__(self, lifetime: int = 60) -> None:
+        self.lifetime = lifetime
+
+    @staticmethod
+    def find_lock_store(store=None) -> "LockStore":
+        if store is None or store.type is None or store.type == "in_memory":
+            lock_store = InMemoryLockStore()
+        elif store.type == "redis":
+            lock_store = RedisLockStore(host=store.url, **store.kwargs)
+        else:
+            raise ValueError(
+                "Cannot create `LockStore` of type '{}'. One of the following "
+                "`LockStore` types need to be specified: {}."
+                "".format(store.type, ", ".join(ACCEPTED_LOCK_STORES))
+            )
+
+        logger.debug(
+            "Connected to lock store {}.".format(lock_store.__class__.__name__)
+        )
+
+        return lock_store
+
+    def create_lock(self, conversation_id: Text) -> TicketLock:
+        """Create and save a new `TicketLock` for `conversation_id`."""
+
+        lock = TicketLock(conversation_id)
+        self.save_lock(lock)
+        return lock
+
+    def get_lock(self, conversation_id: Text) -> Optional[TicketLock]:
+        """Fetch lock for `conversation_id` from storage."""
+
+        raise NotImplementedError
+
+    def delete_lock(self, conversation_id: Text) -> None:
+        """Delete lock for `conversation_id` from storage."""
+
+        raise NotImplementedError
+
+    def save_lock(self, lock: TicketLock) -> None:
+        """Commit `lock` to storage."""
+
+        raise NotImplementedError
+
+    def issue_ticket(self, conversation_id: Text) -> int:
+        """Issue new ticket for lock associated with `conversation_id`.
+
+        Creates a new lock if none is found.
+        """
+
+        lock = self.get_or_create_lock(conversation_id)
+        ticket = lock.issue_ticket(self.lifetime)
+        self.save_lock(lock)
+        return ticket
+
+    async def lock(
+        self,
+        conversation_id: Text,
+        ticket: int,
+        attempts: int = 60,
+        wait: Union[int, float] = 1,
+    ) -> TicketLock:
+        """Acquire lock for `conversation_id` with `ticket`.
+
+        Perform `attempts` with a wait of `wait` seconds between them before
+        raising a `LockError`.
+        """
+
+        lock = self.get_lock(conversation_id)
+
+        while attempts > 0:
+            # acquire lock if it isn't locked
+            if not lock.is_locked(ticket):
+                return lock
+
+            # sleep and update lock
+            await asyncio.sleep(wait)
+            self.update_lock(conversation_id)
+
+            # fetch lock again because lock might no longer exist
+            lock = self.get_lock(conversation_id)
+
+            # exit loop if lock does not exist anymore (expired)
+            if not lock:
+                break
+
+            attempts -= 1
+
+        raise LockError(
+            "Could not acquire lock for conversation_id '{}'."
+            "".format(conversation_id)
+        )
+
+    def update_lock(self, conversation_id: Text) -> None:
+        """Fetch lock for `conversation_id`, remove expired tickets and save lock."""
+
+        lock = self.get_lock(conversation_id)
+        if lock:
+            lock.remove_expired_tickets()
+            self.save_lock(lock)
+
+    def get_or_create_lock(self, conversation_id: Text) -> TicketLock:
+        """Fetch existing lock for `conversation_id` or create a new one if
+        it doesn't exist."""
+
+        existing_lock = self.get_lock(conversation_id)
+
+        if existing_lock:
+            return existing_lock
+
+        return self.create_lock(conversation_id)
+
+    def is_someone_waiting(self, conversation_id: Text) -> bool:
+        """Return whether someone is waiting for lock associated with
+        `conversation_id`."""
+
+        lock = self.get_lock(conversation_id)
+        if lock:
+            return lock.is_someone_waiting()
+
+        return False
+
+    def finish_serving(self, conversation_id: Text, ticket_number: int) -> None:
+        """Finish serving ticket with `ticket_number` for `conversation_id`.
+
+        Removes ticket from lock and saves lock.
+        """
+
+        lock = self.get_lock(conversation_id)
+        if lock:
+            lock.remove_ticket_for_ticket_number(ticket_number)
+            self.save_lock(lock)
+
+    def cleanup(self, conversation_id: Text, ticket_number: int) -> None:
+        """Remove lock for `conversation_id` if no one is waiting."""
+        self.finish_serving(conversation_id, ticket_number)
+        if not self.is_someone_waiting(conversation_id):
+            self.delete_lock(conversation_id)
+            logger.debug(
+                "Deleted lock for conversation '{}' (unused)".format(conversation_id)
+            )
+
+
+class RedisLockStore(LockStore):
+    """Redis store for ticket locks."""
+
+    def __init__(
+        self,
+        host: Text = "localhost",
+        port: int = 6379,
+        db: int = 1,
+        password: Optional[Text] = None,
+        lifetime: int = 60,
+    ) -> None:
+        import redis
+
+        self.red = redis.StrictRedis(
+            host=host, port=int(port), db=int(db), password=password
+        )
+        super().__init__(lifetime)
+
+    def get_lock(self, conversation_id: Text) -> Optional[TicketLock]:
+        serialised_lock = self.red.get(conversation_id)
+        if serialised_lock:
+            return pickle.loads(serialised_lock)
+
+    def delete_lock(self, conversation_id: Text) -> None:
+        self.red.delete(conversation_id)
+
+    def save_lock(self, lock: TicketLock) -> None:
+        self.red.set(lock.conversation_id, pickle.dumps(lock))
+
+
+class InMemoryLockStore(LockStore):
+    """In-memory store for ticket locks."""
+
+    def __init__(self, lifetime: int = 60) -> None:
+        self.conversation_locks = {}  # type: Dict[Text, TicketLock]
+        super().__init__(lifetime)
+
+    def get_lock(self, conversation_id: Text) -> Optional[TicketLock]:
+        return self.conversation_locks.get(conversation_id)
+
+    def delete_lock(self, conversation_id: Text) -> None:
+        if conversation_id in self.conversation_locks:
+            del self.conversation_locks[conversation_id]
+
+    def save_lock(self, lock: TicketLock) -> None:
+        self.conversation_locks[lock.conversation_id] = lock
diff --git a/rasa/core/run.py b/rasa/core/run.py
index 7e6c20531324..86a7e8ad09f0 100644
--- a/rasa/core/run.py
+++ b/rasa/core/run.py
@@ -15,6 +15,7 @@
 from rasa.core.channels import BUILTIN_CHANNELS, console
 from rasa.core.channels.channel import InputChannel
 from rasa.core.interpreter import NaturalLanguageInterpreter
+from rasa.core.lock_store import LockStore
 from rasa.core.tracker_store import TrackerStore
 from rasa.core.utils import AvailableEndpoints, configure_file_logging
 from rasa.model import get_model_subdirectories, get_model
@@ -198,6 +199,7 @@ async def load_agent_on_start(
     _tracker_store = TrackerStore.find_tracker_store(
         None, endpoints.tracker_store, _broker
     )
+    _lock_store = LockStore.find_lock_store(endpoints.lock_store)
 
     model_server = endpoints.model if endpoints and endpoints.model else None
 
@@ -208,6 +210,7 @@ async def load_agent_on_start(
         interpreter=_interpreter,
         generator=endpoints.nlg,
         tracker_store=_tracker_store,
+        lock_store=_lock_store,
         action_endpoint=endpoints.action,
     )
 
diff --git a/rasa/core/utils.py b/rasa/core/utils.py
index ac8691b83610..6bc587080d40 100644
--- a/rasa/core/utils.py
+++ b/rasa/core/utils.py
@@ -1,16 +1,15 @@
 # -*- coding: utf-8 -*-
 import argparse
-import asyncio
 import json
 import logging
 import re
 import sys
-from pathlib import Path
-from typing import Union
 from asyncio import Future
 from hashlib import md5, sha1
 from io import StringIO
+from pathlib import Path
 from typing import Any, Dict, List, Optional, Set, TYPE_CHECKING, Text, Tuple, Callable
+from typing import Union
 
 import aiohttp
 from aiohttp import InvalidURL
@@ -18,12 +17,11 @@
 from sanic.views import CompositionView
 
 import rasa.utils.io as io_utils
-from rasa.utils.endpoints import read_endpoint_config
-
 
 # backwards compatibility 1.0.x
 # noinspection PyUnresolvedReferences
 from rasa.utils.endpoints import concat_url
+from rasa.utils.endpoints import read_endpoint_config
 
 logger = logging.getLogger(__name__)
 
@@ -393,9 +391,10 @@ def read_endpoints(cls, endpoint_file):
         tracker_store = read_endpoint_config(
             endpoint_file, endpoint_type="tracker_store"
         )
+        lock_store = read_endpoint_config(endpoint_file, endpoint_type="lock_store")
         event_broker = read_endpoint_config(endpoint_file, endpoint_type="event_broker")
 
-        return cls(nlg, nlu, action, model, tracker_store, event_broker)
+        return cls(nlg, nlu, action, model, tracker_store, lock_store, event_broker)
 
     def __init__(
         self,
@@ -404,6 +403,7 @@ def __init__(
         action=None,
         model=None,
         tracker_store=None,
+        lock_store=None,
         event_broker=None,
     ):
         self.model = model
@@ -411,6 +411,7 @@ def __init__(
         self.nlu = nlu
         self.nlg = nlg
         self.tracker_store = tracker_store
+        self.lock_store = lock_store
         self.event_broker = event_broker
 
 
@@ -452,29 +453,3 @@ def handler(fut: Future) -> None:
             )
 
     return handler
-
-
-class LockCounter(asyncio.Lock):
-    """Decorated asyncio lock that counts how many coroutines are waiting.
-
-    The counter can be used to discard the lock when there is no coroutine
-    waiting for it. For this to work, there should not be any execution yield
-    between retrieving the lock and acquiring it, otherwise there might be
-    race conditions."""
-
-    def __init__(self) -> None:
-        super().__init__()
-        self.wait_counter = 0
-
-    async def acquire(self) -> bool:
-        """Acquire the lock, makes sure only one coroutine can retrieve it."""
-
-        self.wait_counter += 1
-        try:
-            return await super(LockCounter, self).acquire()  # type: ignore
-        finally:
-            self.wait_counter -= 1
-
-    def is_someone_waiting(self) -> bool:
-        """Check if a coroutine is waiting for this lock to be freed."""
-        return self.wait_counter != 0
diff --git a/rasa/run.py b/rasa/run.py
index a8256e34784c..52fe47b23cdd 100644
--- a/rasa/run.py
+++ b/rasa/run.py
@@ -4,6 +4,7 @@
 
 from rasa.constants import DOCS_BASE_URL
 from rasa.cli.utils import minimal_kwargs, print_warning
+from rasa.core.lock_store import LockStore
 
 logger = logging.getLogger(__name__)
 
@@ -68,10 +69,12 @@ def create_agent(model: Text, endpoints: Text = None) -> "Agent":
     _tracker_store = TrackerStore.find_tracker_store(
         None, _endpoints.tracker_store, _broker
     )
+    _lock_store = LockStore.find_lock_store(_endpoints.lock_store)
 
     return Agent.load(
         model,
         generator=_endpoints.nlg,
         tracker_store=_tracker_store,
+        lock_store=_lock_store,
         action_endpoint=_endpoints.action,
     )
diff --git a/rasa/server.py b/rasa/server.py
index 1f01092793d9..3c3667bfe1b4 100644
--- a/rasa/server.py
+++ b/rasa/server.py
@@ -16,6 +16,7 @@
 import rasa.utils.endpoints
 import rasa.utils.io
 from rasa.core.domain import InvalidDomain
+from rasa.core.lock_store import LockStore
 from rasa.utils.endpoints import EndpointConfig
 from rasa.constants import (
     MINIMUM_COMPATIBLE_VERSION,
@@ -238,6 +239,7 @@ async def _load_agent(
 ) -> Agent:
     try:
         tracker_store = None
+        lock_store = None
         generator = None
         action_endpoint = None
 
@@ -246,6 +248,7 @@ async def _load_agent(
             tracker_store = TrackerStore.find_tracker_store(
                 None, endpoints.tracker_store, _broker
             )
+            lock_store = LockStore.find_lock_store(endpoints.lock_store)
             generator = endpoints.nlg
             action_endpoint = endpoints.action
 
@@ -255,6 +258,7 @@ async def _load_agent(
             remote_storage,
             generator=generator,
             tracker_store=tracker_store,
+            lock_store=lock_store,
             action_endpoint=action_endpoint,
         )
     except Exception as e:
diff --git a/tests/core/conftest.py b/tests/core/conftest.py
index fd023db2f5b5..975bb714883b 100644
--- a/tests/core/conftest.py
+++ b/tests/core/conftest.py
@@ -1,5 +1,4 @@
 import asyncio
-import logging
 import os
 from typing import Text
 
@@ -7,10 +6,8 @@
 import pytest
 
 import rasa.utils.io
-from rasa.core import train
 from rasa.core.agent import Agent
-from rasa.core.channels import channel
-from rasa.core.channels.channel import CollectingOutputChannel, RestInput
+from rasa.core.channels.channel import CollectingOutputChannel
 from rasa.core.domain import Domain
 from rasa.core.interpreter import RegexInterpreter
 from rasa.core.nlg import TemplatedNaturalLanguageGenerator
@@ -24,7 +21,6 @@
 from rasa.core.slots import Slot
 from rasa.core.tracker_store import InMemoryTrackerStore
 from rasa.core.trackers import DialogueStateTracker
-from rasa.utils.io import zip_folder
 from rasa.train import train_async
 
 matplotlib.use("Agg")
@@ -106,7 +102,7 @@ def default_domain():
 
 
 @pytest.fixture(scope="session")
-async def default_agent(default_domain):
+async def default_agent(default_domain) -> Agent:
     agent = Agent(
         default_domain,
         policies=[MemoizationPolicy()],
@@ -167,7 +163,6 @@ def moodbot_metadata(unpacked_trained_moodbot_path):
 async def trained_stack_model(
     default_domain_path, default_stack_config, default_nlu_data, default_stories_file
 ):
-
     trained_stack_model_path = await train_async(
         domain=default_domain_path,
         config=default_stack_config,
diff --git a/tests/core/test_lock_store.py b/tests/core/test_lock_store.py
new file mode 100644
index 000000000000..1d21bd0569d1
--- /dev/null
+++ b/tests/core/test_lock_store.py
@@ -0,0 +1,248 @@
+import asyncio
+from typing import Union
+from unittest.mock import patch
+
+import numpy as np
+import pytest
+import time
+from _pytest.tmpdir import TempdirFactory
+
+import rasa.utils.io
+from rasa.core.agent import Agent
+from rasa.core.channels import UserMessage
+from rasa.core.constants import INTENT_MESSAGE_PREFIX
+from rasa.core.lock import TicketLock
+from rasa.core.lock_store import InMemoryLockStore, LockError
+
+
+@pytest.fixture(scope="session")
+def loop():
+    loop = asyncio.new_event_loop()
+    asyncio.set_event_loop(loop)
+    loop = rasa.utils.io.enable_async_loop_debugging(loop)
+    yield loop
+    loop.close()
+
+
+def test_issue_ticket():
+    lock = TicketLock("random id 0")
+
+    # no lock issued
+    assert lock.last_issued == -1
+    assert lock.now_serving == 0
+
+    # no one is waiting
+    assert not lock.is_someone_waiting()
+
+    # issue ticket
+    ticket = lock.issue_ticket(1)
+    assert ticket == 0
+    assert lock.last_issued == 0
+    assert lock.now_serving == 0
+
+    # someone is waiting
+    assert lock.is_someone_waiting()
+
+
+def test_remove_expired_tickets():
+    lock = TicketLock("random id 1")
+
+    # issue one long- and one short-lived ticket
+    _ = list(map(lock.issue_ticket, [k for k in [0.01, 10]]))
+
+    # both tickets are there
+    assert len(lock.tickets) == 2
+
+    # sleep and only one ticket should be left
+    time.sleep(0.02)
+    lock.remove_expired_tickets()
+    assert len(lock.tickets) == 1
+
+
+def test_create_lock_store():
+    lock_store = InMemoryLockStore()
+    conversation_id = "my id 0"
+
+    # create lock
+    lock_store.create_lock(conversation_id)
+    lock = lock_store.get_lock(conversation_id)
+    assert lock
+    assert lock.conversation_id == conversation_id
+
+
+def test_serve_ticket():
+    lock_store = InMemoryLockStore()
+    conversation_id = "my id 1"
+
+    lock = lock_store.create_lock(conversation_id)
+
+    # issue ticket with long lifetime
+    ticket_0 = lock.issue_ticket(10)
+    assert ticket_0 == 0
+    assert lock.last_issued == ticket_0
+    assert lock.now_serving == ticket_0
+    assert lock.is_someone_waiting()
+
+    # issue another ticket
+    ticket_1 = lock.issue_ticket(10)
+
+    # finish serving ticket_0
+    lock_store.finish_serving(conversation_id, ticket_0)
+    lock = lock_store.get_lock(conversation_id)
+    assert lock.last_issued == ticket_1
+    assert lock.now_serving == ticket_1
+    assert lock.is_someone_waiting()
+
+    # serve second ticket and no one should be waiting
+    lock_store.finish_serving(conversation_id, ticket_1)
+    assert not lock.is_someone_waiting()
+
+
+def test_lock_expiration():
+    lock_store = InMemoryLockStore()
+    conversation_id = "my id 2"
+    lock = lock_store.create_lock(conversation_id)
+
+    # issue ticket with long lifetime
+    ticket = lock.issue_ticket(10)
+    assert ticket == 0
+    assert not lock.has_lock_expired(ticket)
+
+    # issue ticket with short lifetime
+    ticket = lock.issue_ticket(0.00001)
+    time.sleep(0.00002)
+    assert ticket == 1
+    assert lock.has_lock_expired(ticket)
+
+    # newly assigned ticket should get number 1 again
+    assert lock.issue_ticket(10) == 1
+
+
+async def test_multiple_conversation_ids(default_agent: Agent):
+    text = INTENT_MESSAGE_PREFIX + 'greet{"name":"Rasa"}'
+
+    conversation_ids = ["conversation {}".format(i) for i in range(2)]
+
+    # ensure conversations are processed in order
+    tasks = [default_agent.handle_text(text, sender_id=_id) for _id in conversation_ids]
+    results = await asyncio.gather(*tasks)
+
+    assert results
+    processed_ids = [result[0]["recipient_id"] for result in results]
+    assert processed_ids == conversation_ids
+
+
+async def test_message_order(tmpdir_factory: TempdirFactory, default_agent: Agent):
+    start_time = time.time()
+    n_messages = 10
+    lock_wait = 0.1
+
+    # let's write the incoming order of messages and the order of results to temp files
+    temp_path = tmpdir_factory.mktemp("message_order")
+    results_file = temp_path / "results_file"
+    incoming_order_file = temp_path / "incoming_order_file"
+
+    # We need to mock `Agent.handle_message()` so we can introduce an
+    # artificial holdup (`wait`). In the mocked method, we'll
+    # record messages as they come and and as they're processed in files so we
+    # can check the order later on. We don't need the return value of this method so
+    # we'll just return None.
+    async def mocked_handle_message(
+        self, message: UserMessage, wait: Union[int, float]
+    ) -> None:
+        ticket = self.lock_store.issue_ticket(message.sender_id)
+
+        # write incoming message to file
+        with open(str(incoming_order_file), "a+") as f_0:
+            f_0.write(message.text + "\n")
+
+        try:
+            async with await self.lock_store.lock(
+                message.sender_id, ticket, wait=lock_wait
+            ):
+
+                # hold up the message processing after the lock has been acquired
+                await asyncio.sleep(wait)
+
+                # write message to file as it's processed
+                with open(str(results_file), "a+") as f_1:
+                    f_1.write(message.text + "\n")
+
+                return None
+        finally:
+            self.lock_store.cleanup(message.sender_id, ticket)
+
+    # We'll send n_messages from the same sender_id with different blocking times
+    # after the lock has been acquired.
+    # We have to ensure that the messages are processed in the right order.
+    with patch.object(Agent, "handle_message", mocked_handle_message):
+        # use decreasing wait times so that every message after the first one
+        # does not acquire its lock immediately
+        wait_times = np.linspace(0.1, 0.05, n_messages)
+        tasks = [
+            default_agent.handle_message(
+                UserMessage("sender {0}".format(i), sender_id="some id"), wait=k
+            )
+            for i, k in enumerate(wait_times)
+        ]
+
+        # execute futures
+        await asyncio.gather(*(asyncio.ensure_future(t) for t in tasks))
+
+        expected_order = ["sender {0}".format(i) for i in range(len(wait_times))]
+
+        # ensure order of incoming messages is as expected
+        with open(str(incoming_order_file)) as f:
+            incoming_order = [l for l in f.read().split("\n") if l]
+            assert incoming_order == expected_order
+
+        # ensure results are processed in expected order
+        with open(str(results_file)) as f:
+            results_order = [l for l in f.read().split("\n") if l]
+            assert results_order == expected_order
+
+        # Every message after the first one will wait `lock_wait` seconds to acquire its
+        # lock (`wait` kwarg in `lock_store.lock()`). Let's make sure that this is not
+        # blocking and test that total test execution time is less than
+        # the sum of all wait times plus (n_messages - 1) * lock_wait
+        time_limit = np.sum(wait_times[1:])
+        time_limit += (n_messages - 1) * lock_wait
+        assert time.time() - start_time < time_limit
+
+
+async def test_lock_error(default_agent: Agent):
+    attempts = 10
+    wait_between_attempts = 0.001
+
+    # Mock message handler again to limit the number of attempts to acquire the lock
+    # and reduce the wait time between acquisition attempts.
+    async def mocked_handle_message(
+        self, message: UserMessage, wait: Union[int, float]
+    ) -> None:
+        ticket = self.lock_store.issue_ticket(message.sender_id)
+
+        try:
+            async with await self.lock_store.lock(
+                message.sender_id, ticket, attempts=attempts, wait=wait_between_attempts
+            ):
+
+                # hold up the message processing after the lock has been acquired
+                await asyncio.sleep(wait)
+
+                return None
+        finally:
+            self.lock_store.cleanup(message.sender_id, ticket)
+
+    with patch.object(Agent, "handle_message", mocked_handle_message):
+        # first message should block longer (attempts * wait_between_attempts),
+        # meaning the second message will not be able to acquire a lock
+        wait_times = [0.03, 0]
+        tasks = [
+            default_agent.handle_message(
+                UserMessage("sender {0}".format(i), sender_id="some id"), wait=k
+            )
+            for i, k in enumerate(wait_times)
+        ]
+
+        with pytest.raises(LockError):
+            await asyncio.gather(*(asyncio.ensure_future(t) for t in tasks))


Code review comments:
diff_hunk: @@ -0,0 +1,210 @@
+import asyncio
+import logging
+import pickle
+import typing
+from typing import Text, Optional, Dict, Union
+
+from rasa.core.lock import TicketLock
+
+if typing.TYPE_CHECKING:
+    pass
+
+logger = logging.getLogger(__name__)
+
+ACCEPTED_LOCK_STORES = ["in_memory", "redis"]
+
+
+class LockError(Exception):
+    pass
+
+
+class LockStore(object):
+    def __init__(self, lifetime: int = 60) -> None:
+        self.lifetime = lifetime
+
+    @staticmethod
+    def find_lock_store(store=None) -> "LockStore":
+        if store is None or store.type is None or store.type == "in_memory":
+            lock_store = InMemoryLockStore()
+        elif store.type == "redis":
+            lock_store = RedisLockStore(host=store.url, **store.kwargs)
+        else:
+            raise ValueError(
+                "Cannot create `LockStore` of type '{}'. One of the following "
+                "`LockStore` types need to be specified: {}."
+                "".format(store.type, ", ".join(ACCEPTED_LOCK_STORES))
+            )
+
+        logger.debug(
+            "Connected to lock store {}.".format(lock_store.__class__.__name__)
+        )
+
+        return lock_store
+
+    def create_lock(self, conversation_id: Text) -> TicketLock:
+        """Create and save a new `TicketLock` for `conversation_id`."""
+
+        lock = TicketLock(conversation_id)
+        self.save_lock(lock)
+        return lock
+
+    def get_lock(self, conversation_id: Text) -> Optional[TicketLock]:
+        """Fetch lock for `conversation_id` from storage."""
+
+        raise NotImplementedError
+
+    def delete_lock(self, conversation_id: Text) -> None:
+        """Delete lock for `conversation_id` from storage."""
+
+        raise NotImplementedError
+
+    def save_lock(self, lock: TicketLock) -> None:
+        """Commit `lock` to storage."""
+
+        raise NotImplementedError
+
+    def issue_ticket(self, conversation_id: Text) -> int:
+        """Issue new ticket for lock associated with `conversation_id`.
+
+        Creates a new lock if none is found.
+        """
+
+        lock = self.get_or_create_lock(conversation_id)
+        ticket = lock.issue_ticket(self.lifetime)
+        self.save_lock(lock)
+        return ticket
+
+    async def lock(
+        self,
+        conversation_id: Text,
+        ticket: int,
+        attempts: int = 60,
+        wait: Union[int, float] = 1,
+    ) -> TicketLock:
+        """Acquire lock for `conversation_id` with `ticket`.
+
+        Perform `attempts` with a wait of `wait` seconds between them before
+        raising a `LockError`.
+        """
+
+        lock = self.get_lock(conversation_id)
+
+        while attempts > 0:
+            # acquire lock if it isn't locked
+            if not lock.is_locked(ticket):
+                return lock
+
+            # sleep and update lock
+            await asyncio.sleep(wait)
+            self.update_lock(conversation_id)
+
+            # fetch lock again because lock might no longer exist
+            lock = self.get_lock(conversation_id)
+
+            # exit loop if lock does not exist anymore (expired)
+            if not lock:
+                break
+
+            attempts -= 1
+
+        raise LockError(
+            "Could not acquire lock for conversation_id '{}'."
+            "".format(conversation_id)
+        )
+
+    def update_lock(self, conversation_id: Text) -> None:
+        """Fetch lock for `conversation_id`, remove expired tickets and save lock."""
+
+        lock = self.get_lock(conversation_id)
+        if lock:
+            lock.remove_expired_tickets()
+            self.save_lock(lock)
+
+    def get_or_create_lock(self, conversation_id: Text) -> TicketLock:
+        """Fetch existing lock for `conversation_id` or create a new one if
+        it doesn't exist."""
+
+        existing_lock = self.get_lock(conversation_id)
+
+        if existing_lock:
+            return existing_lock
+
+        return self.create_lock(conversation_id)
+
+    def is_someone_waiting(self, conversation_id: Text) -> bool:
+        """Return whether someone is waiting for lock associated with
+        `conversation_id`."""
+
+        lock = self.get_lock(conversation_id)
+        if lock:
+            return lock.is_someone_waiting()
+
+        return False
+
+    def finish_serving(self, conversation_id: Text, ticket_number: int) -> None:
+        """Finish serving ticket with `ticket_number` for `conversation_id`.
+
+        Removes ticket from lock and saves lock.
+        """
+
+        lock = self.get_lock(conversation_id)
+        if lock:
+            lock.remove_ticket_for_ticket_number(ticket_number)
+            self.save_lock(lock)
+
+    def cleanup(self, conversation_id: Text, ticket_number: int) -> None:
+        """Remove lock for `conversation_id` if no one is waiting."""
+        self.finish_serving(conversation_id, ticket_number)
+        if not self.is_someone_waiting(conversation_id):
+            self.delete_lock(conversation_id)
+            logger.debug(
+                "Deleted lock for conversation '{}' (unused)".format(conversation_id)
+            )
+
+
+class RedisLockStore(LockStore):
+    """Redis store for ticket locks."""
+
+    def __init__(
+        self,
+        host: Text = "localhost",
+        port: int = 6379,
+        db: int = 1,
+        password: Optional[Text] = None,
+        lifetime: int = 60,
+    ) -> None:
+        import redis
+
+        self.red = redis.StrictRedis(
+            host=host, port=int(port), db=int(db), password=password
+        )
+        super().__init__(lifetime)
+
+    def get_lock(self, conversation_id: Text) -> Optional[TicketLock]:
+        serialised_lock = self.red.get(conversation_id)
+        if serialised_lock:
+            return pickle.loads(serialised_lock)
+
+    def delete_lock(self, conversation_id: Text) -> None:
+        self.red.delete(conversation_id)
+
+    def save_lock(self, lock: TicketLock) -> None:
+        self.red.set(lock.conversation_id, pickle.dumps(lock))
path: rasa/core/lock_store.py
text: @user1:
why pickle and not json? There is also an issue for the RedisTrackerStore regarding the storage format: https://github.com/RasaHQ/rasa/issues/3925

@author:
changed to json (also the redis tracker store)

diff_hunk: @@ -0,0 +1,210 @@
+import asyncio
+import logging
+import pickle
+import typing
+from typing import Text, Optional, Dict, Union
+
+from rasa.core.lock import TicketLock
+
+if typing.TYPE_CHECKING:
+    pass
+
+logger = logging.getLogger(__name__)
+
+ACCEPTED_LOCK_STORES = ["in_memory", "redis"]
+
+
+class LockError(Exception):
+    pass
+
+
+class LockStore(object):
+    def __init__(self, lifetime: int = 60) -> None:
+        self.lifetime = lifetime
+
+    @staticmethod
+    def find_lock_store(store=None) -> "LockStore":
+        if store is None or store.type is None or store.type == "in_memory":
+            lock_store = InMemoryLockStore()
+        elif store.type == "redis":
+            lock_store = RedisLockStore(host=store.url, **store.kwargs)
+        else:
+            raise ValueError(
+                "Cannot create `LockStore` of type '{}'. One of the following "
+                "`LockStore` types need to be specified: {}."
+                "".format(store.type, ", ".join(ACCEPTED_LOCK_STORES))
+            )
+
+        logger.debug(
+            "Connected to lock store {}.".format(lock_store.__class__.__name__)
+        )
+
+        return lock_store
+
+    def create_lock(self, conversation_id: Text) -> TicketLock:
+        """Create and save a new `TicketLock` for `conversation_id`."""
+
+        lock = TicketLock(conversation_id)
+        self.save_lock(lock)
+        return lock
+
+    def get_lock(self, conversation_id: Text) -> Optional[TicketLock]:
+        """Fetch lock for `conversation_id` from storage."""
+
+        raise NotImplementedError
+
+    def delete_lock(self, conversation_id: Text) -> None:
+        """Delete lock for `conversation_id` from storage."""
+
+        raise NotImplementedError
+
+    def save_lock(self, lock: TicketLock) -> None:
+        """Commit `lock` to storage."""
+
+        raise NotImplementedError
+
+    def issue_ticket(self, conversation_id: Text) -> int:
+        """Issue new ticket for lock associated with `conversation_id`.
+
+        Creates a new lock if none is found.
+        """
+
+        lock = self.get_or_create_lock(conversation_id)
+        ticket = lock.issue_ticket(self.lifetime)
+        self.save_lock(lock)
+        return ticket
+
+    async def lock(
+        self,
+        conversation_id: Text,
+        ticket: int,
+        attempts: int = 60,
+        wait: Union[int, float] = 1,
+    ) -> TicketLock:
+        """Acquire lock for `conversation_id` with `ticket`.
+
+        Perform `attempts` with a wait of `wait` seconds between them before
+        raising a `LockError`.
+        """
+
+        lock = self.get_lock(conversation_id)
+
+        while attempts > 0:
+            # acquire lock if it isn't locked
+            if not lock.is_locked(ticket):
+                return lock
+
+            # sleep and update lock
+            await asyncio.sleep(wait)
+            self.update_lock(conversation_id)
+
+            # fetch lock again because lock might no longer exist
+            lock = self.get_lock(conversation_id)
+
+            # exit loop if lock does not exist anymore (expired)
+            if not lock:
+                break
+
+            attempts -= 1
+
+        raise LockError(
+            "Could not acquire lock for conversation_id '{}'."
+            "".format(conversation_id)
+        )
+
+    def update_lock(self, conversation_id: Text) -> None:
+        """Fetch lock for `conversation_id`, remove expired tickets and save lock."""
+
+        lock = self.get_lock(conversation_id)
+        if lock:
+            lock.remove_expired_tickets()
+            self.save_lock(lock)
+
+    def get_or_create_lock(self, conversation_id: Text) -> TicketLock:
+        """Fetch existing lock for `conversation_id` or create a new one if
+        it doesn't exist."""
+
+        existing_lock = self.get_lock(conversation_id)
+
+        if existing_lock:
+            return existing_lock
+
+        return self.create_lock(conversation_id)
+
+    def is_someone_waiting(self, conversation_id: Text) -> bool:
+        """Return whether someone is waiting for lock associated with
+        `conversation_id`."""
+
+        lock = self.get_lock(conversation_id)
+        if lock:
+            return lock.is_someone_waiting()
+
+        return False
+
+    def finish_serving(self, conversation_id: Text, ticket_number: int) -> None:
+        """Finish serving ticket with `ticket_number` for `conversation_id`.
+
+        Removes ticket from lock and saves lock.
+        """
+
+        lock = self.get_lock(conversation_id)
+        if lock:
+            lock.remove_ticket_for_ticket_number(ticket_number)
+            self.save_lock(lock)
+
+    def cleanup(self, conversation_id: Text, ticket_number: int) -> None:
+        """Remove lock for `conversation_id` if no one is waiting."""
+        self.finish_serving(conversation_id, ticket_number)
+        if not self.is_someone_waiting(conversation_id):
+            self.delete_lock(conversation_id)
+            logger.debug(
+                "Deleted lock for conversation '{}' (unused)".format(conversation_id)
+            )
+
+
+class RedisLockStore(LockStore):
+    """Redis store for ticket locks."""
+
+    def __init__(
+        self,
+        host: Text = "localhost",
+        port: int = 6379,
+        db: int = 1,
+        password: Optional[Text] = None,
+        lifetime: int = 60,
+    ) -> None:
+        import redis
+
+        self.red = redis.StrictRedis(
+            host=host, port=int(port), db=int(db), password=password
+        )
+        super().__init__(lifetime)
+
+    def get_lock(self, conversation_id: Text) -> Optional[TicketLock]:
+        serialised_lock = self.red.get(conversation_id)
+        if serialised_lock:
+            return pickle.loads(serialised_lock)
+
+    def delete_lock(self, conversation_id: Text) -> None:
+        self.red.delete(conversation_id)
+
+    def save_lock(self, lock: TicketLock) -> None:
+        self.red.set(lock.conversation_id, pickle.dumps(lock))
+
+
+class InMemoryLockStore(LockStore):
+    """In-memory store for ticket locks."""
+
+    def __init__(self, lifetime: int = 60) -> None:
+        self.conversation_locks = {}  # type: Dict[Text, TicketLock]
path: rasa/core/lock_store.py
text: @user1:
I think we should make this at least safe for multiprocessing, shouldn't we?