fix: bound the synchronous Meilisearch wait when indexing library blocks

openedx/core/djangoapps/content/search/api.py

@@ -58,6 +58,16 @@
 
 log = logging.getLogger(__name__)
 
+# Socket timeout, in seconds, for HTTP calls to Meilisearch. The Meilisearch client defaults to no
+# timeout, which lets an unresponsive search backend pin a web worker indefinitely.
+MEILISEARCH_HTTP_TIMEOUT = 30
+
+# How long a request may block waiting for Meilisearch to finish applying an index update before
+# giving up and returning. Giving up does not lose the update: Meilisearch has already accepted the
+# documents and will apply them regardless, so this only bounds how long the user waits for the
+# index to become consistent.
+SYNC_INDEX_WAIT_TIMEOUT = 3.0
+
 User = get_user_model()
 
 STUDIO_INDEX_SUFFIX = "studio_content"
@@ -122,7 +132,13 @@ def _get_meilisearch_client():
     if _MEILI_CLIENT is not None:
         return _MEILI_CLIENT
 
-    _MEILI_CLIENT = MeilisearchClient(settings.MEILISEARCH_URL, settings.MEILISEARCH_API_KEY)
+    _MEILI_CLIENT = MeilisearchClient(
+        settings.MEILISEARCH_URL,
+        settings.MEILISEARCH_API_KEY,
+        # Without this the client has no socket timeout at all, so a Meilisearch that accepts
+        # the connection and then stops responding blocks the calling thread forever.
+        timeout=MEILISEARCH_HTTP_TIMEOUT,
+    )
     try:
         _MEILI_CLIENT.health()
     except MeilisearchError as err:
@@ -147,24 +163,41 @@ def _get_meili_api_key_uid():
     return _MEILI_API_KEY_UID
 
 
-def _wait_for_meili_task(info: TaskInfo) -> None:
+def _wait_for_meili_task(info: TaskInfo, timeout: float | None = None) -> bool:
     """
     Simple helper method to wait for a Meilisearch task to complete
-    This method will block until the task is completed, so it should only be used in celery tasks
-    or management commands.
+
+    By default this blocks until the task is completed, so the default should only be used in celery
+    tasks or management commands. Pass ``timeout`` (seconds) to bound the wait when calling from a
+    request thread.
+
+    Returns True if the task completed, False if we stopped waiting because ``timeout`` elapsed.
+    Giving up does not lose the update: Meilisearch has already accepted the documents and will apply
+    them regardless. Waiting only tells us *when* the index became consistent.
 
     ✨ Note: "Meilisearch processes tasks in the order they were added to the queue."
        per https://www.meilisearch.com/docs/capabilities/indexing/tasks_and_batches/monitor_tasks#monitoring-task-status
        so if you need to wait for multiple tasks, simply wait for the final (last) task.
     """
     client = _get_meilisearch_client()
+    deadline = None if timeout is None else time.monotonic() + timeout
     # This function almost always gets called immediately after enqueing a task, and from experiments, an initial wait
     # of at least 15ms is warranted, as the task is almost never done in less than 10ms. We are using 20ms which seems
     # to work well without requiring an additional wait in most cases.
     sleep_delay = 0.020  # Initial wait is only 20ms but we will back off exponentially
     time.sleep(sleep_delay)
     current_status = client.get_task(info.task_uid)
     while current_status.status in ("enqueued", "processing"):
+        if deadline is not None and time.monotonic() >= deadline:
+            log.warning(
+                "Stopped waiting for Meilisearch task %s after %.1fs (status=%s). The update has been "
+                "accepted by Meilisearch and will still be applied; the search index is briefly stale.",
+                info.task_uid, timeout, current_status.status,
+            )
+            return False
+        # Never sleep past the deadline - otherwise a 2s backoff can overshoot a short timeout.
+        if deadline is not None:
+            sleep_delay = min(sleep_delay, max(deadline - time.monotonic(), 0.005))
         time.sleep(sleep_delay)
         sleep_delay = min(sleep_delay * 1.5, 2.0)  # Increase delay up to 2s
         current_status = client.get_task(info.task_uid)
@@ -174,6 +207,7 @@ def _wait_for_meili_task(info: TaskInfo) -> None:
         except (TypeError, KeyError):
             err_reason = "Unknown error"
         raise MeilisearchError(err_reason)
+    return True
 
 
 def _index_exists(index_name: str) -> bool:
@@ -313,11 +347,14 @@ def _recurse_children(block, fn, status_cb: Callable[[str], None] | None = None)
                 fn(child)
 
 
-def _update_index_docs(docs) -> None:
+def _update_index_docs(docs, wait_timeout: float | None = None) -> None:
     """
     Helper function that updates the documents in the search index
 
     If there is a rebuild in progress, the document will also be added to the new index.
+
+    ``wait_timeout`` bounds how long we wait for Meilisearch to finish applying the update. The
+    documents are applied either way; see ``_wait_for_meili_task``.
     """
     if not docs:
         return
@@ -328,7 +365,7 @@ def _update_index_docs(docs) -> None:
     if current_rebuild_index_name:
         # If there is a rebuild in progress, the document will also be added to the new index.
         client.index(current_rebuild_index_name).update_documents(docs)
-    _wait_for_meili_task(client.index(STUDIO_INDEX_NAME).update_documents(docs))
+    _wait_for_meili_task(client.index(STUDIO_INDEX_NAME).update_documents(docs), timeout=wait_timeout)
 
 
 def only_if_meilisearch_enabled(f):
@@ -851,17 +888,20 @@ def _delete_index_doc(doc_id) -> None:
     _wait_for_meili_task(client.index(STUDIO_INDEX_NAME).delete_document(doc_id))
 
 
-def upsert_library_block_index_doc(usage_key: UsageKey) -> None:
+def upsert_library_block_index_doc(usage_key: UsageKey, wait_timeout: float | None = None) -> None:
     """
     Creates or updates the document for the given Library Block in the search index
+
+    ``wait_timeout`` bounds how long to wait for Meilisearch to apply the update; pass it when
+    calling from a request thread so a slow search backend cannot hold the response open.
     """
 
     library_block = lib_api.get_component_from_usage_key(usage_key)
     library_block_metadata = lib_api.LibraryXBlockMetadata.from_component(usage_key.context_key, library_block)
 
     docs = [searchable_doc_for_library_block(library_block_metadata)]
 
-    _update_index_docs(docs)
+    _update_index_docs(docs, wait_timeout=wait_timeout)
 
 
 def _get_document_from_index(document_id: str) -> dict:

openedx/core/djangoapps/content/search/handlers.py

@@ -47,6 +47,7 @@
 from xmodule.modulestore.django import SignalHandler
 
 from .api import (
+    SYNC_INDEX_WAIT_TIMEOUT,
     is_meilisearch_enabled,
     only_if_meilisearch_enabled,
     reconcile_index,
@@ -174,9 +175,14 @@ def library_block_updated_handler(**kwargs) -> None:
         log.error("Received null or incorrect data for event")
         return
 
-    # Update content library index synchronously to make sure that search index is updated before
-    # the frontend invalidates/refetches results. This is only a single document update so is very fast.
-    upsert_library_block_index_doc.apply(args=[str(library_block_data.usage_key)])
+    # Update the content library index synchronously so that the search index is usually fresh before
+    # the frontend invalidates/refetches results. The wait is *bounded*: when the search backend is
+    # slow we return without it rather than holding the response open, and Meilisearch still applies
+    # the update in the background. See SYNC_INDEX_WAIT_TIMEOUT.
+    upsert_library_block_index_doc.apply(
+        args=[str(library_block_data.usage_key)],
+        kwargs={"wait_timeout": SYNC_INDEX_WAIT_TIMEOUT},
+    )
 
 
 @receiver(LIBRARY_BLOCK_PUBLISHED)
@@ -200,9 +206,14 @@ def library_block_published_handler(**kwargs) -> None:
         # via the DELETED handler, so there's nothing to do now.
         return
 
-    # Update content library index synchronously to make sure that search index is updated before
-    # the frontend invalidates/refetches results. This is only a single document update so is very fast.
-    upsert_library_block_index_doc.apply(args=[str(library_block_data.usage_key)])
+    # Update the content library index synchronously so that the search index is usually fresh before
+    # the frontend invalidates/refetches results. The wait is *bounded*: when the search backend is
+    # slow we return without it rather than holding the response open, and Meilisearch still applies
+    # the update in the background. See SYNC_INDEX_WAIT_TIMEOUT.
+    upsert_library_block_index_doc.apply(
+        args=[str(library_block_data.usage_key)],
+        kwargs={"wait_timeout": SYNC_INDEX_WAIT_TIMEOUT},
+    )
 
 
 @receiver(LIBRARY_BLOCK_DELETED)

openedx/core/djangoapps/content/search/tasks.py

@@ -65,15 +65,18 @@ def delete_xblock_index_doc(usage_key_str: str) -> None:
 
 @shared_task(base=LoggedTask, autoretry_for=(MeilisearchError, ConnectionError))
 @set_code_owner_attribute
-def upsert_library_block_index_doc(usage_key_str: str) -> None:
+def upsert_library_block_index_doc(usage_key_str: str, wait_timeout: float | None = None) -> None:
     """
     Celery task to update the content index document for a library block
+
+    ``wait_timeout`` bounds how long to wait for Meilisearch to apply the update. Callers running
+    this eagerly inside a request should pass it; a real celery worker can leave it as None.
     """
     usage_key = LibraryUsageLocatorV2.from_string(usage_key_str)
 
     log.info("Updating content index document for library block with id: %s", usage_key)
 
-    api.upsert_library_block_index_doc(usage_key)
+    api.upsert_library_block_index_doc(usage_key, wait_timeout=wait_timeout)
 
 
 @shared_task(base=LoggedTask, autoretry_for=(MeilisearchError, ConnectionError))

openedx/core/djangoapps/content/search/tests/test_api.py

@@ -4,14 +4,15 @@
 from __future__ import annotations
 
 import copy
+import time
 from datetime import UTC, datetime
 from unittest.mock import MagicMock, Mock, call, patch
 
 import ddt
 import pytest
-from django.test import override_settings
+from django.test import TestCase, override_settings
 from freezegun import freeze_time
-from meilisearch.errors import MeilisearchApiError
+from meilisearch.errors import MeilisearchApiError, MeilisearchError
 from opaque_keys.edx.keys import UsageKey
 from opaque_keys.edx.locator import LibraryCollectionLocator, LibraryContainerLocator
 from openedx_content import api as content_api
@@ -1436,3 +1437,88 @@ def test_get_all_blocks_from_context(self, mock_meilisearch):
                 "attributesToRetrieve": ["usage_key", "display_name"],
             }
         )
+
+
+def _wait_for_meili_task(*args, **kwargs):
+    """
+    Call the module-private helper under test.
+
+    Wrapped rather than aliased at module level: `api` is imported inside a try/except above
+    because it raises in the LMS, so resolving the attribute at import time would break
+    collection on LMS test runs even though these tests are CMS-only.
+    """
+    return api._wait_for_meili_task(*args, **kwargs)  # pylint: disable=protected-access
+
+
+@override_settings(MEILISEARCH_ENABLED=True)
+@skip_unless_cms
+class TestWaitForMeiliTaskTimeout(TestCase):
+    """
+    Tests for the bounded wait in ``_wait_for_meili_task``.
+    """
+
+    def setUp(self):
+        super().setUp()
+        api.clear_meilisearch_client()
+
+    def tearDown(self):
+        super().tearDown()
+        api.clear_meilisearch_client()
+
+    @staticmethod
+    def _client_with_status(status):
+        """A Meilisearch client whose tasks never leave ``status``."""
+        client = MagicMock()
+        client.get_task.return_value = MagicMock(status=status, error=None)
+        return client
+
+    def test_returns_false_once_the_timeout_elapses(self):
+        """
+        A task that never finishes must not block the caller past the timeout.
+
+        This is the whole point of the bound: the documents have already been accepted by
+        Meilisearch and will be applied regardless, so there is nothing to gain by waiting.
+        """
+        with patch.object(api, "_MEILI_CLIENT", self._client_with_status("processing")):
+            started = time.monotonic()
+            completed = _wait_for_meili_task(MagicMock(task_uid=1), timeout=0.5)
+            elapsed = time.monotonic() - started
+
+        assert completed is False
+        # Generous upper bound so this does not flake on a loaded CI worker, but tight enough to
+        # fail if the timeout is ignored (without it, this call never returns).
+        assert elapsed < 5
+
+    def test_does_not_sleep_past_the_deadline(self):
+        """
+        The backoff caps at 2s, so a naive implementation would overshoot a shorter timeout.
+        """
+        with patch.object(api, "_MEILI_CLIENT", self._client_with_status("enqueued")):
+            started = time.monotonic()
+            _wait_for_meili_task(MagicMock(task_uid=2), timeout=0.1)
+            elapsed = time.monotonic() - started
+
+        assert elapsed < 2
+
+    def test_returns_true_when_the_task_succeeds(self):
+        with patch.object(api, "_MEILI_CLIENT", self._client_with_status("succeeded")):
+            assert _wait_for_meili_task(MagicMock(task_uid=3), timeout=5) is True
+
+    def test_raises_on_failure_rather_than_reporting_completion(self):
+        with patch.object(api, "_MEILI_CLIENT", self._client_with_status("failed")):
+            with pytest.raises(MeilisearchError):
+                _wait_for_meili_task(MagicMock(task_uid=4), timeout=5)
+
+    def test_waits_indefinitely_by_default(self):
+        """
+        Omitting ``timeout`` must keep the historical behaviour, which reindex and the management
+        commands rely on. Verified by letting the task complete on the second poll.
+        """
+        client = MagicMock()
+        client.get_task.side_effect = [
+            MagicMock(status="processing", error=None),
+            MagicMock(status="succeeded", error=None),
+        ]
+        with patch.object(api, "_MEILI_CLIENT", client):
+            assert _wait_for_meili_task(MagicMock(task_uid=5)) is True
+        assert client.get_task.call_count == 2