diff --git a/docs/contributions/async-library-indexing.md b/docs/contributions/async-library-indexing.md new file mode 100644 index 000000000000..53205eaea4b0 --- /dev/null +++ b/docs/contributions/async-library-indexing.md @@ -0,0 +1,50 @@ +# Durable library-wide indexing slice + +## Baseline and boundary + +Base: `7944dc908dffff5ff21754859fc0ca554005b313` of openedx/edx-platform. +Content search already has Celery tasks. Library create and rename handlers call their library-wide task with `.apply`, which executes synchronously. Some source library events are themselves emitted by workers; this change decouples that event worker from the engine wait, rather than asserting every event originates in an HTTP publication request. + +## Implemented contract + +`LIBRARY_SEARCH_ASYNC_INDEXING=True` opts only library create/rename into a durable database intent. Default behavior stays compatible with the current frontend. SearchAccess creation remains immediate. Library block, container, collection, association and course flows retain their existing behavior. + +Within the caller's transaction, request_library_index increments a per-library requested revision. A post-commit callback dispatches a Celery task. Broker operational failures are logged; the database request survives. A rolled-back caller transaction leaves neither intent nor dispatch. + +The worker takes the library request row lock, reads current authoritative content rather than an old event payload, streams block, collection and container documents through the existing document constructors and `_update_index_docs` API boundary (including their breadcrumbs), then advances completion only after every batch's engine task wait returns. Source querysets use `iterator(chunk_size=batch_size)` rather than populating Django's full result cache. The default batch size is 100 documents, configurable with the positive integer `LIBRARY_SEARCH_INDEX_BATCH_SIZE`. The synchronous baseline API remains unchanged. Each batch retains the existing current-index wait and concurrent-rebuild dual-write behavior; the existing rebuild shadow write is not newly given a completion wait. Duplicate deliveries are no-ops; multiple pending requests coalesce. Failures roll back completion. Celery retries connection/Meilisearch errors five times with backoff and jitter, and late acknowledgment plus worker-loss rejection allow redelivery. A bounded recovery command redispatches unresolved rows, including requests lost between database commit and broker dispatch. Run it periodically through the deployment's existing scheduler; no scheduler is silently installed. + +Library deletion cancels the durable row. A missing-library exception during delivery also removes the request, avoiding a permanent poison item. This cancellation does not introduce deletion of existing search documents; library lifecycle cleanup remains the existing upstream responsibility. + +## Lock and ordering limits + +The row lock spans the engine call deliberately: queued library-wide writes cannot overtake one another, and enqueue revisions cannot be acknowledged accidentally by an older worker. A crash after an engine write but before the database commit causes replay of current content, giving idempotent convergence rather than exactly-once delivery. + +This conservative design means a subsequent create/rename enqueue for the same library can wait for an active worker's engine call. It moves the ordinary wait out of the event path but does not guarantee bounded authoring latency under lock contention. Pending revisions are ordered by transaction arrival; they are not source content versions. They always reread current content, so delayed notifications converge on current state. Existing per-item writes do not acquire this lock: their concurrent changes can race library-wide indexing. Therefore global stale-write prevention is not claimed. + +SQLite does not enforce select_for_update. The isolated tests prove persistence/rollback/coalescing, not production row-lock ordering. Before activation, run concurrent-worker/rename/deletion fault tests against the supported production database and the real engine. For nonblocking enqueue, a future immutable-intent table with a separate per-library worker lock is preferable; it needs explicit FK-lock and deletion semantics on the deployment database. + +## Rollout and outstanding work + +1. Obtain maintainer agreement for the bounded scope and setting name; run CMS integration tests and migration checks with pinned dependencies. +2. Deploy the additive migration with the setting false. Deploy workers before enabling the setting. +3. Configure periodic bounded recovery and alert on age/revision lag; validate engine timeout and broker retry settings. +4. Add an author-visible pending/failed indicator and refetch behavior before opt-in production use. This slice exposes durable internal state, not a new public status endpoint or frontend. +5. Run production-database concurrency tests and a real Meilisearch fault-injection scenario. Redesign the long lock or unify per-item writes before claiming comprehensive order safety. + +Document batches are bounded, but document byte size, nested metadata construction, and database-driver buffering are not. End-to-end process memory therefore requires measurement on the production database. Engine failure after an accepted batch leaves the revision pending; recovery rebuilds batches from current content and replays accepted earlier batches. Library disappearance after a partial write cancels the request, but does not clean up partially indexed documents. Library deletion/reconciliation must handle that existing lifecycle gap. + +Per-item durable intents, source versions, cross-engine support, telemetry, job dead-letter UI, library-wide deletion repair, retention cleanup, and deployment benchmarks remain open. This is an initial contribution for library-wide indexing requests. + +Disabling the setting restores synchronous future events but does not cancel already pending requests. Drain/recover pending rows before rollback; workers can continue draining with the setting false. Do not drop the table with pending work. + +## Local validation evidence + +17 focused tests passed with complete CMS dependencies. Django reported no missing +search migrations. The actual additive migration applied and rolled back on a +dedicated MySQL 8.4.11 database using the existing migration dependency state. +That same database preserved four concurrent same-library enqueues, serialized +duplicate deliveries into one indexing call, and recovered a pending revision +after its task process was killed. A real Redis broker connection outage left +the intent pending; a Celery worker consumed it after recovery and ignored a +duplicate. Those fault scenarios mock the indexing/engine boundary. They prove +outbox and broker behavior, not complete source-to-engine concurrency safety. diff --git a/openedx/core/djangoapps/content/search/handlers.py b/openedx/core/djangoapps/content/search/handlers.py index c60c67eb7115..0b13f6e71abd 100644 --- a/openedx/core/djangoapps/content/search/handlers.py +++ b/openedx/core/djangoapps/content/search/handlers.py @@ -4,6 +4,7 @@ import logging +from django.conf import settings from django.db.models.signals import post_delete from django.dispatch import receiver from opaque_keys import InvalidKeyError @@ -42,7 +43,7 @@ ) from openedx.core.djangoapps.content.course_overviews.models import CourseOverview -from openedx.core.djangoapps.content.search.models import SearchAccess +from openedx.core.djangoapps.content.search.models import LibraryIndexRequest, SearchAccess from openedx.core.djangoapps.content_libraries import api as lib_api from xmodule.modulestore.django import SignalHandler @@ -54,6 +55,7 @@ upsert_item_collections_index_docs, upsert_item_containers_index_docs, ) +from .library_indexing import request_library_index from .tasks import ( delete_course_index_docs, delete_library_block_index_doc, @@ -111,6 +113,7 @@ def delete_course_search_access(sender, instance, **kwargs): # pylint: disable= @receiver(CONTENT_LIBRARY_DELETED) def delete_library_search_access(content_library: ContentLibraryData, **kwargs): """Deletes the SearchAccess instance for deleted content libraries""" + LibraryIndexRequest.objects.filter(library_key=str(content_library.library_key)).delete() SearchAccess.objects.filter(context_key=content_library.library_key).delete() @@ -237,7 +240,10 @@ def content_library_created_handler(**kwargs) -> None: # right after creation. Without this, the JWT token won't include the new library's # access_id until it's added by the document indexing process or the page is refreshed. SearchAccess.objects.get_or_create(context_key=library_key) - update_content_library_index_docs.apply(args=[str(library_key), True]) + if getattr(settings, "LIBRARY_SEARCH_ASYNC_INDEXING", False): + request_library_index(library_key) + else: + update_content_library_index_docs.apply(args=[str(library_key), True]) @receiver(CONTENT_LIBRARY_UPDATED) @@ -257,7 +263,10 @@ def content_library_updated_handler(**kwargs) -> None: # Update ALL items in the library, because their breadcrumbs will be outdated. # TODO: just patch the "breadcrumbs" field? It's the same on every one. # TODO: check if the library display_name has actually changed before updating all items? - update_content_library_index_docs.apply(args=[str(library_key)]) + if getattr(settings, "LIBRARY_SEARCH_ASYNC_INDEXING", False): + request_library_index(library_key) + else: + update_content_library_index_docs.apply(args=[str(library_key)]) @receiver(LIBRARY_COLLECTION_CREATED) diff --git a/openedx/core/djangoapps/content/search/library_indexing.py b/openedx/core/djangoapps/content/search/library_indexing.py new file mode 100644 index 000000000000..f62d5ad79b6c --- /dev/null +++ b/openedx/core/djangoapps/content/search/library_indexing.py @@ -0,0 +1,71 @@ +"""Durable library-wide indexing with an explicit, recoverable broker boundary.""" + +import logging + +from celery import shared_task +from django.conf import settings +from django.db import transaction +from kombu.exceptions import OperationalError +from meilisearch.errors import MeilisearchError +from opaque_keys.edx.locator import LibraryLocatorV2 + +from .library_indexing_batches import index_library_in_batches +from .models import LibraryIndexRequest + +log = logging.getLogger(__name__) + + +def _dispatch(library_key): + """A broker outage must not undo a committed content change.""" + try: + process_library_index_request.delay(library_key) + except OperationalError: + log.exception("Library indexing dispatch failed; durable request remains pending for %s", library_key) + + +def request_library_index(library_key): + """Record intent in the caller's transaction and dispatch only after commit.""" + library_key = str(LibraryLocatorV2.from_string(str(library_key))) + with transaction.atomic(): + request, _ = LibraryIndexRequest.objects.get_or_create(library_key=library_key) + request = LibraryIndexRequest.objects.select_for_update().get(pk=request.pk) + request.requested_revision += 1 + request.save(update_fields=["requested_revision", "updated_at"]) + transaction.on_commit(lambda: _dispatch(library_key)) + + +@shared_task( + autoretry_for=(MeilisearchError, ConnectionError), + retry_backoff=True, + retry_jitter=True, + max_retries=5, + acks_late=True, + reject_on_worker_lost=True, +) +def process_library_index_request(library_key): + """Serialize writes for one library and acknowledge only completed engine tasks. + + The conservative row lock spans the engine call. Duplicate broker deliveries + become no-ops; a crashed worker leaves the revision pending. This lock requires + a database supporting row locks (the production MySQL/PostgreSQL backends). + """ + from openedx.core.djangoapps.content_libraries.api import ( # pylint: disable=import-outside-toplevel + ContentLibraryNotFound, + ) + + + with transaction.atomic(): + request = LibraryIndexRequest.objects.select_for_update().filter(library_key=library_key).first() + if request is None or request.completed_revision == request.requested_revision: + return + try: + index_library_in_batches( + LibraryLocatorV2.from_string(library_key), + batch_size=getattr(settings, "LIBRARY_SEARCH_INDEX_BATCH_SIZE", 100), + ) + except ContentLibraryNotFound: + # Deletion may precede delivery even if its cancellation signal was lost. + request.delete() + return + request.completed_revision = request.requested_revision + request.save(update_fields=["completed_revision", "updated_at"]) diff --git a/openedx/core/djangoapps/content/search/library_indexing_batches.py b/openedx/core/djangoapps/content/search/library_indexing_batches.py new file mode 100644 index 000000000000..157c574bd677 --- /dev/null +++ b/openedx/core/djangoapps/content/search/library_indexing_batches.py @@ -0,0 +1,36 @@ +"""Stream full library documents into bounded, individually awaited engine batches.""" + +from openedx.core.djangoapps.content_libraries import api as lib_api + + +def index_library_in_batches(library_key, batch_size=100): + """Reuse existing document construction and dual-index write behavior. + + Each source queryset uses iterator() to avoid Django's full result cache. + No persistent per-batch checkpoint is used: interrupted runs replay current + authoritative content, including batches the engine may already have accepted. + """ + if not isinstance(batch_size, int) or isinstance(batch_size, bool) or batch_size <= 0: + raise ValueError("batch_size must be a positive integer") + + from . import api # pylint: disable=import-outside-toplevel + + def documents(): + for component in lib_api.get_library_components(library_key).iterator(chunk_size=batch_size): + metadata = lib_api.LibraryXBlockMetadata.from_component(library_key, component) + yield api.searchable_doc_for_library_block(metadata) + for container in lib_api.get_library_containers(library_key).iterator(chunk_size=batch_size): + container_key = lib_api.library_container_locator(library_key, container) + yield api.searchable_doc_for_container(container_key) + for collection in lib_api.get_library_collections(library_key).iterator(chunk_size=batch_size): + collection_key = lib_api.library_collection_locator(library_key, collection.collection_code) + yield api.searchable_doc_for_collection(collection_key, collection=collection) + + batch = [] + for doc in documents(): + batch.append(doc) + if len(batch) == batch_size: + api._update_index_docs(api.STUDIO_LIBRARY_INDEX_NAME, batch) # pylint: disable=protected-access + batch = [] + if batch: + api._update_index_docs(api.STUDIO_LIBRARY_INDEX_NAME, batch) # pylint: disable=protected-access diff --git a/openedx/core/djangoapps/content/search/management/commands/retry_library_index_requests.py b/openedx/core/djangoapps/content/search/management/commands/retry_library_index_requests.py new file mode 100644 index 000000000000..5bb9fa113f1b --- /dev/null +++ b/openedx/core/djangoapps/content/search/management/commands/retry_library_index_requests.py @@ -0,0 +1,27 @@ +"""Redispatch durable requests after a broker outage or exhausted worker retries.""" + +from django.core.management.base import BaseCommand +from django.db.models import F + +from ...library_indexing import process_library_index_request +from ...models import LibraryIndexRequest + + +class Command(BaseCommand): + """Dispatch only the requested bounded number of pending revisions.""" + help = "Redispatch a bounded batch of pending library indexing requests." + + def add_arguments(self, parser): + parser.add_argument("--limit", type=int, default=100) + + def handle(self, *args, **options): + if options["limit"] <= 0: + raise ValueError("--limit must be positive") + pending = LibraryIndexRequest.objects.filter( + requested_revision__gt=F("completed_revision"), + ).order_by("updated_at").values_list("library_key", flat=True)[:options["limit"]] + count = 0 + for library_key in pending: + process_library_index_request.delay(library_key) + count += 1 + self.stdout.write(f"Dispatched {count} pending library indexing requests.") diff --git a/openedx/core/djangoapps/content/search/migrations/0004_libraryindexrequest.py b/openedx/core/djangoapps/content/search/migrations/0004_libraryindexrequest.py new file mode 100644 index 000000000000..9b0204b01c4c --- /dev/null +++ b/openedx/core/djangoapps/content/search/migrations/0004_libraryindexrequest.py @@ -0,0 +1,18 @@ +from django.db import migrations, models + + +class Migration(migrations.Migration): + dependencies = [("search", "0003_clear_library_incremental_index_checkpoints")] + + operations = [ + migrations.CreateModel( + name="LibraryIndexRequest", + fields=[ + ("id", models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name="ID")), + ("library_key", models.CharField(max_length=255, unique=True)), + ("requested_revision", models.PositiveBigIntegerField(default=0)), + ("completed_revision", models.PositiveBigIntegerField(default=0)), + ("updated_at", models.DateTimeField(auto_now=True)), + ], + ), + ] diff --git a/openedx/core/djangoapps/content/search/models.py b/openedx/core/djangoapps/content/search/models.py index d726f1ead057..52cc9d4ee61f 100644 --- a/openedx/core/djangoapps/content/search/models.py +++ b/openedx/core/djangoapps/content/search/models.py @@ -81,3 +81,15 @@ class IncrementalIndexCompleted(models.Model): # noqa: DJ008 unique=True, null=False, ) + + +class LibraryIndexRequest(models.Model): # noqa: DJ008 + """Durable, coalesced library-wide indexing intent. Contains no learner data. + + .. no_pii: + """ + + library_key = models.CharField(max_length=255, unique=True) + requested_revision = models.PositiveBigIntegerField(default=0) + completed_revision = models.PositiveBigIntegerField(default=0) + updated_at = models.DateTimeField(auto_now=True) diff --git a/openedx/core/djangoapps/content/search/tasks.py b/openedx/core/djangoapps/content/search/tasks.py index 3a94bd362129..b4f2037865e1 100644 --- a/openedx/core/djangoapps/content/search/tasks.py +++ b/openedx/core/djangoapps/content/search/tasks.py @@ -18,6 +18,7 @@ ) from . import api +from .library_indexing import process_library_index_request # noqa: F401 # pylint: disable=unused-import log = logging.getLogger(__name__) diff --git a/openedx/core/djangoapps/content/search/tests/test_library_indexing.py b/openedx/core/djangoapps/content/search/tests/test_library_indexing.py new file mode 100644 index 000000000000..b08424a31f8b --- /dev/null +++ b/openedx/core/djangoapps/content/search/tests/test_library_indexing.py @@ -0,0 +1,112 @@ +"""Durable outbox tests; engine calls are mocked at the existing API boundary.""" + +from unittest.mock import patch + +import pytest +from django.core.management import call_command +from django.db import transaction +from django.test import TestCase +from kombu.exceptions import OperationalError + +from openedx.core.djangoapps.content_libraries.api import ContentLibraryNotFound + +from ..library_indexing import process_library_index_request, request_library_index +from ..models import LibraryIndexRequest + + +class LibraryIndexingTests(TestCase): + """Verify durable revisions and dispatch across transaction boundaries.""" + library_key = "lib:org:library" + + def request(self): + with patch("openedx.core.djangoapps.content.search.library_indexing.process_library_index_request.delay"): + with self.captureOnCommitCallbacks(execute=True): + request_library_index(self.library_key) + + def test_dispatch_only_after_commit(self): + with patch( + "openedx.core.djangoapps.content.search.library_indexing.process_library_index_request.delay", + ) as delay: + with self.captureOnCommitCallbacks(execute=True): + request_library_index(self.library_key) + delay.assert_not_called() + delay.assert_called_once_with(self.library_key) + + def test_rollback_discards_intent_and_dispatch(self): + with patch( + "openedx.core.djangoapps.content.search.library_indexing.process_library_index_request.delay", + ) as delay: + with self.captureOnCommitCallbacks(execute=True): + with pytest.raises(RuntimeError), transaction.atomic(): # noqa: PT012 + request_library_index(self.library_key) + raise RuntimeError("publication rolled back") + delay.assert_not_called() + assert not LibraryIndexRequest.objects.exists() + + def test_broker_failure_preserves_pending_request(self): + with patch("openedx.core.djangoapps.content.search.library_indexing.process_library_index_request.delay", + side_effect=OperationalError("broker unavailable")): + with self.captureOnCommitCallbacks(execute=True): + request_library_index(self.library_key) + request = LibraryIndexRequest.objects.get() + assert (request.requested_revision, request.completed_revision) == (1, 0) + + @patch("openedx.core.djangoapps.content.search.library_indexing.index_library_in_batches") + def test_coalesce_and_duplicate_delivery(self, upsert): + self.request() + self.request() + process_library_index_request.run(self.library_key) + process_library_index_request.run(self.library_key) + upsert.assert_called_once() + assert upsert.call_args.kwargs["batch_size"] == 100 + request = LibraryIndexRequest.objects.get() + assert (request.requested_revision, request.completed_revision) == (2, 2) + + @patch("openedx.core.djangoapps.content.search.library_indexing.index_library_in_batches") + def test_failed_engine_call_remains_pending_and_can_retry(self, upsert): + self.request() + upsert.side_effect = ConnectionError("engine unavailable") + # Call the original task body so this assertion tests persistence, not Celery retry machinery. + with pytest.raises(ConnectionError): + process_library_index_request._orig_run(self.library_key) # pylint: disable=protected-access + assert LibraryIndexRequest.objects.get().completed_revision == 0 + upsert.side_effect = None + process_library_index_request.run(self.library_key) + assert LibraryIndexRequest.objects.get().completed_revision == 1 + + @patch("openedx.core.djangoapps.content.search.library_indexing.index_library_in_batches") + def test_new_request_after_completion_runs_again(self, upsert): + self.request() + process_library_index_request.run(self.library_key) + self.request() + process_library_index_request.run(self.library_key) + assert upsert.call_count == 2 + assert LibraryIndexRequest.objects.get().completed_revision == 2 + + @patch("openedx.core.djangoapps.content.search.library_indexing.index_library_in_batches") + def test_cancelled_request_is_noop(self, upsert): + self.request() + LibraryIndexRequest.objects.all().delete() + process_library_index_request.run(self.library_key) + upsert.assert_not_called() + + @patch("openedx.core.djangoapps.content.search.library_indexing.index_library_in_batches") + def test_missing_library_cancels_pending_request(self, upsert): + self.request() + upsert.side_effect = ContentLibraryNotFound() + process_library_index_request.run(self.library_key) + assert not LibraryIndexRequest.objects.exists() + + def test_recovery_command_only_dispatches_pending_with_limit(self): + self.request() + LibraryIndexRequest.objects.create(library_key="lib:org:complete", requested_revision=1, completed_revision=1) + LibraryIndexRequest.objects.create(library_key="lib:org:other", requested_revision=1) + with patch( + "openedx.core.djangoapps.content.search.library_indexing.process_library_index_request.delay", + ) as delay: + call_command("retry_library_index_requests", limit=1) + assert delay.call_count == 1 + + def test_recovery_rejects_unbounded_negative_limit(self): + with pytest.raises(ValueError, match="limit must be positive"): + call_command("retry_library_index_requests", limit=-1) diff --git a/openedx/core/djangoapps/content/search/tests/test_library_indexing_batches.py b/openedx/core/djangoapps/content/search/tests/test_library_indexing_batches.py new file mode 100644 index 000000000000..77ec84fd8d96 --- /dev/null +++ b/openedx/core/djangoapps/content/search/tests/test_library_indexing_batches.py @@ -0,0 +1,120 @@ +"""Verify real streaming batches and durable replay through the worker boundary.""" + +from types import SimpleNamespace +from unittest.mock import Mock, patch + +import pytest +from django.test import TestCase + +from openedx.core.djangoapps.content_libraries.api import ContentLibraryNotFound + +from ..library_indexing import process_library_index_request, request_library_index +from ..library_indexing_batches import index_library_in_batches +from ..models import LibraryIndexRequest + + +class LibraryIndexBatchTests(TestCase): + """Exercise lazy batches and replay after partial engine failures.""" + library_key = "lib:org:library" + + def setUp(self): + super().setUp() + self.querysets = [Mock(), Mock(), Mock()] + self.items = [ + [SimpleNamespace(id=f"block-{number}") for number in range(3)], + [SimpleNamespace(id=f"container-{number}") for number in range(2)], + [SimpleNamespace(id=f"collection-{number}", collection_code=str(number)) for number in range(2)], + ] + # Fresh iterator each invocation: retries must read authoritative source again. + for queryset, items in zip(self.querysets, self.items, strict=True): + queryset.iterator.side_effect = lambda chunk_size, items=items: iter(items) + + def document(item, **kwargs): + return {"id": item.id} + + lib_patcher = patch.multiple( + "openedx.core.djangoapps.content.search.library_indexing_batches.lib_api", + create=True, + get_library_components=Mock(return_value=self.querysets[0]), + get_library_containers=Mock(return_value=self.querysets[1]), + get_library_collections=Mock(return_value=self.querysets[2]), + LibraryXBlockMetadata=SimpleNamespace(from_component=lambda key, component: component), + library_container_locator=Mock(side_effect=lambda key, container: container), + library_collection_locator=Mock(side_effect=lambda key, code: self.items[2][int(code)]), + ) + self.write = Mock() + api_patcher = patch.multiple( + "openedx.core.djangoapps.content.search.api", + create=True, + searchable_doc_for_library_block=Mock(side_effect=document), + searchable_doc_for_container=Mock(side_effect=document), + searchable_doc_for_collection=Mock(side_effect=document), + STUDIO_LIBRARY_INDEX_NAME="studio_library", + _update_index_docs=self.write, + ) + lib_patcher.start() + api_patcher.start() + self.addCleanup(lib_patcher.stop) + self.addCleanup(api_patcher.stop) + + def request(self): + with patch("openedx.core.djangoapps.content.search.library_indexing.process_library_index_request.delay"): + request_library_index(self.library_key) + + def test_bounded_batches_include_all_document_types(self): + index_library_in_batches(self.library_key, batch_size=2) + calls = self.write.call_args_list + assert [len(call.args[1]) for call in calls] == [2, 2, 2, 1] + assert [doc["id"] for call in calls for doc in call.args[1]] == [ + item.id for items in self.items for item in items + ] + for queryset in self.querysets: + queryset.iterator.assert_called_once_with(chunk_size=2) + + def test_empty_library_submits_no_batch(self): + for queryset in self.querysets: + queryset.iterator.side_effect = lambda chunk_size: iter(()) + index_library_in_batches(self.library_key, batch_size=2) + self.write.assert_not_called() + + def test_exact_boundary_submits_no_empty_tail(self): + self.querysets[2].iterator.side_effect = lambda chunk_size: iter(self.items[2][:1]) + index_library_in_batches(self.library_key, batch_size=2) + assert [len(call.args[1]) for call in self.write.call_args_list] == [2, 2, 2] + + def test_invalid_batch_sizes_fail_before_source_reads(self): + for batch_size in (0, -1, True, 2.5, "2"): + with pytest.raises(ValueError, match="positive integer"): + index_library_in_batches(self.library_key, batch_size=batch_size) + for queryset in self.querysets: + queryset.iterator.assert_not_called() + + def test_failed_batch_stops_later_reads(self): + self.write.side_effect = ConnectionError("batch failed") + with pytest.raises(ConnectionError): + index_library_in_batches(self.library_key, batch_size=2) + assert self.write.call_count == 1 + self.querysets[1].iterator.assert_not_called() + self.querysets[2].iterator.assert_not_called() + + def test_partial_failure_leaves_revision_pending_and_replays(self): + self.request() + self.write.side_effect = [None, ConnectionError("second batch failed")] + with self.settings(LIBRARY_SEARCH_INDEX_BATCH_SIZE=2): + with pytest.raises(ConnectionError): + process_library_index_request._orig_run(self.library_key) # pylint: disable=protected-access + assert LibraryIndexRequest.objects.get().completed_revision == 0 + first_batch = self.write.call_args_list[0].args[1] + self.write.reset_mock(side_effect=True) + process_library_index_request.run(self.library_key) + assert self.write.call_args_list[0].args[1] == first_batch + assert self.write.call_count == 4 + assert LibraryIndexRequest.objects.get().completed_revision == 1 + + def test_missing_library_after_partial_write_cancels_intent(self): + self.request() + self.write.side_effect = [None, ContentLibraryNotFound()] + with self.settings(LIBRARY_SEARCH_INDEX_BATCH_SIZE=2): + process_library_index_request.run(self.library_key) + assert not LibraryIndexRequest.objects.exists() + assert self.write.call_count == 2 diff --git a/openedx/core/djangoapps/content/search/tests/test_library_indexing_cms.py b/openedx/core/djangoapps/content/search/tests/test_library_indexing_cms.py new file mode 100644 index 000000000000..ab998c8d2f79 --- /dev/null +++ b/openedx/core/djangoapps/content/search/tests/test_library_indexing_cms.py @@ -0,0 +1,34 @@ +"""Bounded indexing with real CMS source models and serializers.""" +from unittest.mock import patch + +from django.test import TestCase, override_settings +from organizations.models import Organization + +from openedx.core.djangoapps.content_libraries import api as library_api +from openedx.core.djangolib.testing.utils import skip_unless_cms + +from .. import api +from ..library_indexing_batches import index_library_in_batches +from ..models import SearchAccess + + +@skip_unless_cms +class LibraryIndexingCMSIntegrationTests(TestCase): + """Keep source reads real and replace only the external write boundary.""" + + def test_real_library_documents_are_batched(self): + with override_settings(MEILISEARCH_ENABLED=False): + org = Organization.objects.create(name="Async validation", short_name="asyncvalidation") + library = library_api.create_library(org=org, slug="bounded", title="Bounded library") + blocks = [library_api.create_library_block(library.key, "html", f"block{n}") for n in range(3)] + library_api.create_library_collection( + library.key, collection_key="collection", title="A collection", created_by=None, + ) + SearchAccess.objects.get_or_create(context_key=library.key) + with patch("openedx.core.djangoapps.content.search.api._update_index_docs") as write: + index_library_in_batches(library.key, batch_size=2) + assert [len(call.args[1]) for call in write.call_args_list] == [2, 2] + documents = [doc for call in write.call_args_list for doc in call.args[1]] + for block in blocks: + assert api.searchable_doc_for_library_block(block) in documents + assert all(doc["context_key"] == str(library.key) for doc in documents)