diff --git a/cms/envs/common.py b/cms/envs/common.py index 87ee13f3e0a6..c8a997f5b3c5 100644 --- a/cms/envs/common.py +++ b/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'] diff --git a/openedx/core/djangoapps/content/search/api.py b/openedx/core/djangoapps/content/search/api.py index da68dcacc297..a2fe52fc169e 100644 --- a/openedx/core/djangoapps/content/search/api.py +++ b/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: diff --git a/openedx/core/djangoapps/content/search/tests/test_api.py b/openedx/core/djangoapps/content/search/tests/test_api.py index c81a4521faff..11e3c615ca00 100644 --- a/openedx/core/djangoapps/content/search/tests/test_api.py +++ b/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}