diff --git a/openedx/core/djangoapps/content/search/tasks.py b/openedx/core/djangoapps/content/search/tasks.py index cd8519d6c5bc..cfd6cd23e1a5 100644 --- a/openedx/core/djangoapps/content/search/tasks.py +++ b/openedx/core/djangoapps/content/search/tasks.py @@ -1,5 +1,23 @@ """ Defines asynchronous celery task for content indexing + +Every task here talks to Meilisearch, so every task inherits whatever latency the +search backend is having. Three of the knobs below bound that: + +``max_retries`` bounds the number of attempts. It is the only one of the three that +also applies on the inline path: the signal handlers in ``handlers.py`` run several +of these with ``.apply()`` so that the index is updated before the Authoring MFE +refetches, which puts the whole attempt sequence inside a Studio HTTP request. Celery +runs an eager retry immediately, so the backoff below does not slow that path down; +only the attempt count limits it. + +``retry_backoff`` / ``retry_backoff_max`` / ``retry_jitter`` bound how hard a queued +retry storm hits a struggling Meilisearch. Celery's default is a flat 180s, which is +both too long for the request-path tasks and unjittered for the rest. + +``soft_time_limit`` / ``time_limit`` bound a single attempt. These are enforced by the +worker's prefork pool and are therefore ignored under ``.apply()``; they exist so a +task that blocks on Meilisearch cannot occupy a worker slot indefinitely. """ from __future__ import annotations @@ -21,8 +39,48 @@ log = logging.getLogger(__name__) - -@shared_task(base=LoggedTask, autoretry_for=(MeilisearchError, ConnectionError)) +# Tasks the handlers run inline with .apply(), so their attempts land inside a Studio +# HTTP request. Each is a single-document write; two extra attempts is the useful +# amount before the caller is better off failing fast. +_REQUEST_PATH_TASK = { + "base": LoggedTask, + "autoretry_for": (MeilisearchError, ConnectionError), + "max_retries": 2, + "retry_backoff": 1, + "retry_backoff_max": 8, + "retry_jitter": True, + "soft_time_limit": 10, + "time_limit": 15, +} + +# Tasks only ever reached through .delay(). Nothing is waiting on them, so they get +# more attempts and a longer per-attempt budget. +_BACKGROUND_TASK = { + "base": LoggedTask, + "autoretry_for": (MeilisearchError, ConnectionError), + "max_retries": 3, + "retry_backoff": True, + "retry_backoff_max": 60, + "retry_jitter": True, + "soft_time_limit": 60, + "time_limit": 90, +} + +# Whole-context walks: these load every block in a course or library from the +# modulestore before writing, which legitimately takes minutes on a large context. +_BULK_TASK = { + "base": LoggedTask, + "autoretry_for": (MeilisearchError, ConnectionError), + "max_retries": 2, + "retry_backoff": True, + "retry_backoff_max": 300, + "retry_jitter": True, + "soft_time_limit": 30 * 60, + "time_limit": 31 * 60, +} + + +@shared_task(**_BACKGROUND_TASK) def upsert_xblock_index_doc(usage_key_str: str, recursive: bool) -> None: """ Celery task to update the content index document for an XBlock @@ -34,7 +92,7 @@ def upsert_xblock_index_doc(usage_key_str: str, recursive: bool) -> None: api.upsert_xblock_index_doc(usage_key, recursive) -@shared_task(base=LoggedTask, autoretry_for=(MeilisearchError, ConnectionError)) +@shared_task(**_BULK_TASK) def upsert_course_blocks_docs(course_key_str: str) -> None: """ Celery task to update the content index document for all XBlocks in a course. @@ -46,7 +104,7 @@ def upsert_course_blocks_docs(course_key_str: str) -> None: api.index_course(course_key) -@shared_task(base=LoggedTask, autoretry_for=(MeilisearchError, ConnectionError)) +@shared_task(**_BACKGROUND_TASK) def delete_xblock_index_doc(usage_key_str: str) -> None: """ Celery task to delete the content index document for an XBlock @@ -59,7 +117,7 @@ def delete_xblock_index_doc(usage_key_str: str) -> None: api.delete_index_doc(usage_key, delete_children=True) -@shared_task(base=LoggedTask, autoretry_for=(MeilisearchError, ConnectionError)) +@shared_task(**_REQUEST_PATH_TASK) 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 @@ -74,7 +132,7 @@ def upsert_library_block_index_doc(usage_key_str: str, wait_timeout: float | Non api.upsert_library_block_index_doc(usage_key, wait_timeout=wait_timeout) -@shared_task(base=LoggedTask, autoretry_for=(MeilisearchError, ConnectionError)) +@shared_task(**_REQUEST_PATH_TASK) def delete_library_block_index_doc(usage_key_str: str) -> None: """ Celery task to delete the content index document for a library block @@ -86,7 +144,7 @@ def delete_library_block_index_doc(usage_key_str: str) -> None: api.delete_index_doc(usage_key) -@shared_task(base=LoggedTask, autoretry_for=(MeilisearchError, ConnectionError)) +@shared_task(**_BULK_TASK) def update_content_library_index_docs(library_key_str: str, full_index: bool = False) -> None: """ Celery task to update the content index documents for all library blocks in a library @@ -99,7 +157,7 @@ def update_content_library_index_docs(library_key_str: str, full_index: bool = F api.upsert_content_library_index_docs(library_key, full_index=full_index) -@shared_task(base=LoggedTask, autoretry_for=(MeilisearchError, ConnectionError)) +@shared_task(**_REQUEST_PATH_TASK) def update_library_collection_index_doc(collection_key_str: str) -> None: """ Celery task to update the content index document for a library collection @@ -112,7 +170,7 @@ def update_library_collection_index_doc(collection_key_str: str) -> None: api.upsert_library_collection_index_doc(collection_key) -@shared_task(base=LoggedTask, autoretry_for=(MeilisearchError, ConnectionError)) +@shared_task(**_BACKGROUND_TASK) def update_library_components_collections(collection_key_str: str) -> None: """ Celery task to update the "collections" field for components in the given content library collection. @@ -125,7 +183,7 @@ def update_library_components_collections(collection_key_str: str) -> None: api.update_library_components_collections(collection_key) -@shared_task(base=LoggedTask, autoretry_for=(MeilisearchError, ConnectionError)) +@shared_task(**_BACKGROUND_TASK) def update_library_containers_collections(collection_key_str: str) -> None: """ Celery task to update the "collections" field for containers in the given content library collection. @@ -138,7 +196,7 @@ def update_library_containers_collections(collection_key_str: str) -> None: api.update_library_containers_collections(collection_key) -@shared_task(base=LoggedTask, autoretry_for=(MeilisearchError, ConnectionError)) +@shared_task(**_REQUEST_PATH_TASK) def update_library_container_index_doc(container_key_str: str) -> None: """ Celery task to update the content index document for a library container @@ -151,7 +209,7 @@ def update_library_container_index_doc(container_key_str: str) -> None: api.upsert_library_container_index_doc(container_key) -@shared_task(base=LoggedTask, autoretry_for=(MeilisearchError, ConnectionError)) +@shared_task(**_REQUEST_PATH_TASK) def delete_library_container_index_doc(container_key_str: str) -> None: """ Celery task to delete the content index document for a library block @@ -163,7 +221,7 @@ def delete_library_container_index_doc(container_key_str: str) -> None: api.delete_index_doc(container_key) -@shared_task(base=LoggedTask, autoretry_for=(MeilisearchError, ConnectionError)) +@shared_task(**_BACKGROUND_TASK) def delete_course_index_docs(course_key_str: str) -> None: """ Celery task to delete the content index documents for a Course @@ -176,11 +234,16 @@ def delete_course_index_docs(course_key_str: str) -> None: api.delete_docs_with_context_key(course_key) +# No time limit: a full rebuild walks every course and library and legitimately runs +# for hours. It is bounded by the index rebuild lock instead, and by resuming from +# IncrementalIndexCompleted rather than starting over. @shared_task( base=LoggedTask, autoretry_for=(MeilisearchError, ConnectionError), max_retries=3, retry_backoff=True, + retry_backoff_max=600, + retry_jitter=True, ) def rebuild_index_incremental() -> None: """