fix: bound how long content/search waits on Meilisearch document writes

cms/envs/common.py

@@ -1358,6 +1358,18 @@ def _should_send_xblock_events(settings):
 # See https://www.meilisearch.com/docs/learn/security/tenant_tokens
 MEILISEARCH_INDEX_PREFIX = ""
 MEILISEARCH_API_KEY = "devkey"
+# .. setting_name: MEILISEARCH_WRITE_WAIT_TIMEOUT
+# .. setting_default: None
+# .. setting_description: Seconds to wait for Meilisearch to apply each incremental document write (the writes made
+#   when course or library content changes, not index rebuilds) before giving up on waiting. None waits until
+#   Meilisearch finishes. Meilisearch applies tasks one at a time in the order they were enqueued, so a write made
+#   during a large reindex can wait minutes behind it, and the library handlers make these writes inline in the Studio
+#   request. The limit is per write, and one library change can make several inline writes (one per affected
+#   component, unit, subsection or section), so keep the timeout times that count below your gateway's upstream
+#   timeout. Each wait can also overrun by up to one 2s poll interval. A write that is not waited for is still applied
+#   later, but until then the Authoring MFE's refetch of the library listing will not show it, and if it fails the
+#   failure is not raised (so a Celery task making it is not retried).
+MEILISEARCH_WRITE_WAIT_TIMEOUT = None
 
 # .. setting_name: LIBRARY_ENABLED_BLOCKS
 # .. setting_default: ['problem', 'video', 'html', 'drag-and-drop-v2']

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

@@ -165,12 +165,16 @@ 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) -> None:
     """
     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.
 
+    If `timeout` (in seconds) is given and the task is still enqueued or processing when it runs out, this logs a
+    warning and returns instead of raising. The task stays in Meilisearch's queue and will still be applied; only the
+    caller stops waiting for it.
+
     ✨ 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.
@@ -180,9 +184,20 @@ def _wait_for_meili_task(info: TaskInfo) -> None:
     # 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
+    deadline = None if timeout is None else time.monotonic() + timeout
     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 on index %s after %ss; it is still %s and will be applied "
+                "when Meilisearch reaches it in the queue.",
+                info.task_uid,
+                info.index_uid,
+                timeout,
+                current_status.status,
+            )
+            return
         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)
@@ -194,6 +209,14 @@ def _wait_for_meili_task(info: TaskInfo) -> None:
         raise MeilisearchError(err_reason)
 
 
+def _write_wait_timeout() -> float | None:
+    """
+    How long the incremental document write helpers wait for Meilisearch to apply a write. See
+    MEILISEARCH_WRITE_WAIT_TIMEOUT.
+    """
+    return getattr(settings, "MEILISEARCH_WRITE_WAIT_TIMEOUT", None)
+
+
 def _index_exists(index_name: str) -> bool:
     """
     Check if an index exists
@@ -350,7 +373,7 @@ def _update_index_docs(index_name: str, 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(index_name).update_documents(docs))
+    _wait_for_meili_task(client.index(index_name).update_documents(docs), timeout=_write_wait_timeout())
 
 
 def only_if_meilisearch_enabled(f):
@@ -920,7 +943,7 @@ def _delete_documents(index_name: str, filter_query: str) -> None:
     if current_rebuild_index_name:
         # If there is a rebuild in progress, the document will also be removed from the new index.
         client.index(current_rebuild_index_name).delete_documents(filter=filter_query)
-    _wait_for_meili_task(client.index(index_name).delete_documents(filter=filter_query))
+    _wait_for_meili_task(client.index(index_name).delete_documents(filter=filter_query), timeout=_write_wait_timeout())
 
 
 def _delete_index_doc(index_name: str, doc_id) -> None:
@@ -939,7 +962,7 @@ def _delete_index_doc(index_name: str, doc_id) -> None:
         # If there is a rebuild in progress, the document will also be removed from the new index.
         client.index(current_rebuild_index_name).delete_document(doc_id)
 
-    _wait_for_meili_task(client.index(index_name).delete_document(doc_id))
+    _wait_for_meili_task(client.index(index_name).delete_document(doc_id), timeout=_write_wait_timeout())
 
 
 def upsert_library_block_index_doc(usage_key: UsageKey) -> None:

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

@@ -12,7 +12,7 @@
 import ddt
 import pytest
 from django.apps import apps
-from django.test import override_settings
+from django.test import TestCase, override_settings
 from freezegun import freeze_time
 from meilisearch.errors import MeilisearchApiError, MeilisearchError
 from opaque_keys.edx.keys import CourseKey, UsageKey
@@ -1599,3 +1599,83 @@ def test_delete_library_docs_from_missing_course_index(self, mock_meilisearch) -
         api.delete_library_docs_from_course_index()
 
         mock_meilisearch.return_value.index.assert_not_called()
+
+
+@skip_unless_cms
+@override_settings(MEILISEARCH_ENABLED=True)
+@patch("openedx.core.djangoapps.content.search.api.time")
+@patch("openedx.core.djangoapps.content.search.api.MeilisearchClient")
+class TestWaitForMeiliTask(TestCase):
+    """
+    Tests for how long the API waits on Meilisearch tasks.
+    """
+    # pylint: disable=protected-access
+
+    def setUp(self) -> None:
+        super().setUp()
+        api.clear_meilisearch_client()
+        self.addCleanup(api.clear_meilisearch_client)
+
+    def _statuses(self, mock_meilisearch, *statuses: str) -> None:
+        mock_meilisearch.return_value.get_task.side_effect = [Mock(status=status, error=None) for status in statuses]
+
+    def test_waits_until_done_without_timeout(self, mock_meilisearch, mock_time) -> None:
+        """
+        With no timeout, it polls until the task leaves the queue, however long that takes.
+        """
+        self._statuses(mock_meilisearch, "enqueued", "processing", "processing", "succeeded")
+
+        api._wait_for_meili_task(Mock(task_uid=1))
+
+        assert mock_meilisearch.return_value.get_task.call_count == 4
+        mock_time.monotonic.assert_not_called()
+
+    def test_stops_waiting_at_timeout(self, mock_meilisearch, mock_time) -> None:
+        """
+        A task still queued when the timeout runs out is logged and left to Meilisearch, not raised.
+        """
+        self._statuses(mock_meilisearch, "enqueued", "enqueued", "processing", "succeeded")
+        mock_time.monotonic.side_effect = [100.0, 102.0, 106.0]
+
+        with self.assertLogs("openedx.core.djangoapps.content.search.api", level="WARNING") as logs:
+            api._wait_for_meili_task(Mock(task_uid=7, index_uid="studio_library_content"), timeout=5)
+
+        assert mock_meilisearch.return_value.get_task.call_count == 2
+        assert "Stopped waiting for Meilisearch task 7 on index studio_library_content after 5s" in logs.output[0]
+
+    def test_raises_on_failed_task_within_timeout(self, mock_meilisearch, mock_time) -> None:
+        """
+        A task that fails before the timeout still raises.
+        """
+        mock_meilisearch.return_value.get_task.side_effect = [
+            Mock(status="processing"),
+            Mock(status="failed", error={"message": "index_not_found"}),
+        ]
+        mock_time.monotonic.side_effect = [100.0, 101.0]
+
+        with pytest.raises(MeilisearchError, match="index_not_found"):
+            api._wait_for_meili_task(Mock(task_uid=1), timeout=5)
+
+    @override_settings(MEILISEARCH_WRITE_WAIT_TIMEOUT=5)
+    def test_document_writes_use_write_wait_timeout(self, mock_meilisearch, mock_time) -> None:
+        """
+        Incremental document writes and deletes wait at most MEILISEARCH_WRITE_WAIT_TIMEOUT.
+        """
+        with patch("openedx.core.djangoapps.content.search.api._wait_for_meili_task") as mock_wait:
+            api._update_index_docs("idx", [{"id": "a"}])
+            api._delete_index_doc("idx", "a")
+            api._delete_documents("idx", 'context_key = "x"')
+
+        assert mock_wait.call_count == 3
+        for wait_call in mock_wait.call_args_list:
+            assert wait_call.kwargs == {"timeout": 5}
+
+    def test_document_writes_wait_indefinitely_by_default(self, mock_meilisearch, mock_time) -> None:
+        """
+        Without MEILISEARCH_WRITE_WAIT_TIMEOUT, document writes wait as long as Meilisearch takes.
+        """
+        with patch("openedx.core.djangoapps.content.search.api._wait_for_meili_task") as mock_wait:
+            api._update_index_docs("idx", [{"id": "a"}])
+
+        mock_wait.assert_called_once()
+        assert mock_wait.call_args.kwargs == {"timeout": None}