Merge pull request #8105 from RasaHQ/improvement/lock-store-acquire-log

improve Lockstore queuing log message

changelog/8100.improvement.md

@@ -0,0 +1,2 @@
+Improved the [lock store](lock-stores.mdx) debug log message when the process has to
+queue because other messages have to be processed before this item.

rasa/core/lock_store.py

@@ -123,9 +123,13 @@ async def _acquire_lock(
                 logger.debug(f"Acquired lock for conversation '{conversation_id}'.")
                 return lock
 
+            items_before_this = ticket - (lock.now_serving or 0)
+
             logger.debug(
-                f"Failed to acquire lock for conversation ID '{conversation_id}'. "
-                f"Retrying..."
+                f"Failed to acquire lock for conversation ID '{conversation_id}' "
+                f"because {items_before_this} other item(s) for this "
+                f"conversation ID have to be finished processing first. "
+                f"Retrying in {wait_time_in_seconds} seconds ..."
             )
 
             # sleep and update lock

tests/core/test_lock_store.py

@@ -1,4 +1,5 @@
 import asyncio
+import logging
 import os
 import sys
 from pathlib import Path
@@ -7,6 +8,7 @@
 import pytest
 import time
 
+from _pytest.logging import LogCaptureFixture
 from _pytest.monkeypatch import MonkeyPatch
 from _pytest.tmpdir import TempdirFactory
 from unittest.mock import patch, Mock
@@ -282,6 +284,41 @@ async def test_lock_lifetime_environment_variable(monkeypatch: MonkeyPatch):
     assert rasa.core.lock_store._get_lock_lifetime() == new_lock_lifetime
 
 
+@pytest.mark.parametrize("lock_store", [InMemoryLockStore(), FakeRedisLockStore()])
+async def test_acquire_lock_debug_message(
+    lock_store: LockStore, caplog: LogCaptureFixture
+):
+    conversation_id = "test_acquire_lock_debug_message"
+    wait_time_in_seconds = 0.01
+
+    async def locking_task() -> None:
+        async with lock_store.lock(
+            conversation_id, wait_time_in_seconds=wait_time_in_seconds
+        ):
+            # Do a very short sleep so that the other tasks can try to acquire the lock
+            # in the meantime
+            await asyncio.sleep(0.0)
+
+    with caplog.at_level(logging.DEBUG):
+        await asyncio.gather(
+            locking_task(),  # Gets served immediately
+            locking_task(),  # Gets served second
+            locking_task(),  # Gets served last
+        )
+
+    assert any(
+        f"because 1 other item(s) for this conversation ID have to be finished "
+        f"processing first. Retrying in {wait_time_in_seconds} seconds ..." in message
+        for message in caplog.messages
+    )
+
+    assert any(
+        f"because 2 other item(s) for this conversation ID have to be finished "
+        f"processing first. Retrying in {wait_time_in_seconds} seconds ..." in message
+        for message in caplog.messages
+    )
+
+
 async def test_redis_lock_store_timeout(monkeypatch: MonkeyPatch):
     import redis.exceptions