From 981ef3bd6aaba4fbacf99e9e00dad737841ca452 Mon Sep 17 00:00:00 2001 From: Ahtisahm Shahid Date: Wed, 2 Sep 2026 19:47:33 +0500 Subject: [PATCH 1/2] fix: bound the synchronous Meilisearch wait when indexing library blocks Creating or editing a v2 library block indexes it synchronously inside the request, via `upsert_library_block_index_doc.apply()`. That call ends in `_wait_for_meili_task()`, which polls until Meilisearch reports the task finished with no timeout, using a client built with no socket timeout either. So the HTTP response is held open for as long as the search backend takes. The write itself has already committed by then: `LibraryBlocksView` is `non_atomic_requests` and the content-library write happens in its own `transaction.atomic()`, and the events that trigger indexing only fire after that commit. So a slow index cannot roll the write back - it only delays the response. When a gateway in front of Studio times out first, the user sees a 5xx for a request that actually succeeded, and retrying creates another orphaned block. Bound the wait instead: - `_wait_for_meili_task()` takes an optional `timeout` and returns whether the task completed. On timeout it logs and returns rather than polling forever. The backoff is also clamped so a 2s sleep cannot overshoot a shorter deadline. - The Meilisearch client is constructed with a socket timeout, so a backend that accepts the connection and then stops responding can no longer pin a worker. - `_update_index_docs()` and `api.upsert_library_block_index_doc()` thread the timeout through, and the two library-block handlers pass `SYNC_INDEX_WAIT_TIMEOUT`. Giving up on the wait does not lose the update. Once `update_documents()` returns a task uid, Meilisearch applies the documents regardless; waiting only told us when. The default stays `None` (wait indefinitely), so reindex and the management commands are unchanged - only the two interactive paths are bounded. Measured against a Meilisearch proxied to defer applying writes: healthy, timeout=3 -> 0.10s slow 20s, timeout=3 -> 3.01s slow 20s, timeout=None -> 21.27s (previous behaviour) documents present in all three cases End to end through Studio with indexing deferred 100s, this changes `POST /api/libraries/v2/{lib}/blocks/` from no response at 60s to a 200 in ~3.2s. Note this bounds how long the *request* waits, not how long indexing takes. The Authoring MFE lists components from the search index, so on a slow backend a new component can take a few seconds to appear in the listing. That is a better failure mode than a gateway error plus an orphaned block, but fully closing the gap needs the frontend to render the new component from the create response. Refs: openedx/openedx-platform#38993 --- openedx/core/djangoapps/content/search/api.py | 56 +++++++++++-- .../djangoapps/content/search/handlers.py | 23 ++++-- .../core/djangoapps/content/search/tasks.py | 7 +- .../content/search/tests/test_api.py | 79 ++++++++++++++++++- 4 files changed, 147 insertions(+), 18 deletions(-) diff --git a/openedx/core/djangoapps/content/search/api.py b/openedx/core/djangoapps/content/search/api.py index 6d6ce6148cd2..6cc28c96064a 100644 --- a/openedx/core/djangoapps/content/search/api.py +++ b/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,17 +163,24 @@ 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. @@ -165,6 +188,16 @@ def _wait_for_meili_task(info: TaskInfo) -> None: 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,9 +888,12 @@ 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) @@ -861,7 +901,7 @@ def upsert_library_block_index_doc(usage_key: UsageKey) -> None: 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: diff --git a/openedx/core/djangoapps/content/search/handlers.py b/openedx/core/djangoapps/content/search/handlers.py index 0fe93292d8fa..9e793ba6f954 100644 --- a/openedx/core/djangoapps/content/search/handlers.py +++ b/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) diff --git a/openedx/core/djangoapps/content/search/tasks.py b/openedx/core/djangoapps/content/search/tasks.py index a95bedb062db..14bf0afa7352 100644 --- a/openedx/core/djangoapps/content/search/tasks.py +++ b/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)) diff --git a/openedx/core/djangoapps/content/search/tests/test_api.py b/openedx/core/djangoapps/content/search/tests/test_api.py index 3fd6859cafd8..ef70e58074f4 100644 --- a/openedx/core/djangoapps/content/search/tests/test_api.py +++ b/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,77 @@ def test_get_all_blocks_from_context(self, mock_meilisearch): "attributesToRetrieve": ["usage_key", "display_name"], } ) + + +@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 = api._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() + api._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 api._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): + api._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 api._wait_for_meili_task(MagicMock(task_uid=5)) is True + assert client.get_task.call_count == 2 From 7ee768f3691e57f6d37434d2225a00b081a625af Mon Sep 17 00:00:00 2001 From: Ahtisahm Shahid Date: Wed, 2 Sep 2026 20:11:25 +0500 Subject: [PATCH 2/2] fix: satisfy pylint protected-access in the bounded-wait tests Wrap the call to the module-private helper in a single thin function with one waiver, instead of repeating the disable at five call sites. Deliberately a function rather than a module-level alias: `api` is imported inside a try/except in this module because the import raises in the LMS, so resolving `api._wait_for_meili_task` at import time would break collection on LMS test runs even though these tests are CMS-only. Verified LMS collection still succeeds. --- .../content/search/tests/test_api.py | 21 ++++++++++++++----- 1 file changed, 16 insertions(+), 5 deletions(-) diff --git a/openedx/core/djangoapps/content/search/tests/test_api.py b/openedx/core/djangoapps/content/search/tests/test_api.py index ef70e58074f4..7c9976a8aa47 100644 --- a/openedx/core/djangoapps/content/search/tests/test_api.py +++ b/openedx/core/djangoapps/content/search/tests/test_api.py @@ -1439,6 +1439,17 @@ def test_get_all_blocks_from_context(self, mock_meilisearch): ) +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): @@ -1470,7 +1481,7 @@ def test_returns_false_once_the_timeout_elapses(self): """ with patch.object(api, "_MEILI_CLIENT", self._client_with_status("processing")): started = time.monotonic() - completed = api._wait_for_meili_task(MagicMock(task_uid=1), timeout=0.5) + completed = _wait_for_meili_task(MagicMock(task_uid=1), timeout=0.5) elapsed = time.monotonic() - started assert completed is False @@ -1484,19 +1495,19 @@ def test_does_not_sleep_past_the_deadline(self): """ with patch.object(api, "_MEILI_CLIENT", self._client_with_status("enqueued")): started = time.monotonic() - api._wait_for_meili_task(MagicMock(task_uid=2), timeout=0.1) + _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 api._wait_for_meili_task(MagicMock(task_uid=3), timeout=5) is True + 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): - api._wait_for_meili_task(MagicMock(task_uid=4), timeout=5) + _wait_for_meili_task(MagicMock(task_uid=4), timeout=5) def test_waits_indefinitely_by_default(self): """ @@ -1509,5 +1520,5 @@ def test_waits_indefinitely_by_default(self): MagicMock(status="succeeded", error=None), ] with patch.object(api, "_MEILI_CLIENT", client): - assert api._wait_for_meili_task(MagicMock(task_uid=5)) is True + assert _wait_for_meili_task(MagicMock(task_uid=5)) is True assert client.get_task.call_count == 2