From ef9f579116ddcbdef5f08c85e67b47d0b0c5f00d Mon Sep 17 00:00:00 2001 From: Matt Bertrand Date: Sat, 18 Jul 2026 09:15:24 -0400 Subject: [PATCH 01/10] Embed only changed content files per ETL load Phase 2: make content-file embedding work proportional to change instead of re-embedding a run's entire file set on every load. - load_content_files snapshots {key: (checksum, published)} before the per-file loop and computes the changed subset (new, checksum differs, or unpublished->published republish) via _changed_content_file_ids. - The content_files_loaded hook gains content_file_ids and removed_unpublished params (both default None = legacy whole-run behavior, keeping the republish path and data migration working). - embed_run_content_files(run_id, content_file_ids=None) embeds only the named files when given a list, all published files when None. - Above CONTENT_FILE_EMBED_ID_CAP changed files, pass None so the task re-queries rather than serializing a huge id list into the broker. - removed_unpublished is tri-state in the plugin: True and None append the remove-unpublished task, only an explicit False skips it. Addresses https://github.com/mitodl/hq/issues/12454 Co-Authored-By: Claude Opus 4.8 --- learning_resources/etl/loaders.py | 60 ++++++++++++++++++++- learning_resources/etl/loaders_test.py | 65 +++++++++++++++++++++++ learning_resources/hooks.py | 13 ++++- learning_resources/utils.py | 21 ++++++-- learning_resources_search/plugins.py | 24 ++++++--- learning_resources_search/plugins_test.py | 36 +++++++++++-- main/settings.py | 8 +++ vector_search/tasks.py | 23 +++++--- vector_search/tasks_test.py | 51 ++++++++++++++++++ 9 files changed, 277 insertions(+), 24 deletions(-) diff --git a/learning_resources/etl/loaders.py b/learning_resources/etl/loaders.py index 62389d29ba..7676b63de3 100644 --- a/learning_resources/etl/loaders.py +++ b/learning_resources/etl/loaders.py @@ -1028,6 +1028,15 @@ def load_content_files( """ if course_run.learning_resource.resource_type == LearningResourceType.course.name: + # Snapshot existing files so we can embed only what actually changed this + # load, rather than re-embedding the whole run every time. + prior_files = { + key: (checksum, published) + for key, checksum, published in ContentFile.objects.filter( + run=course_run + ).values_list("key", "checksum", "published") + } + content_files_ids = [] content_tags = [] for content_file in content_files_data: @@ -1052,6 +1061,7 @@ def load_content_files( .values_list("direct_learning_resource_id", flat=True) .distinct() ) + removed_unpublished = stale_published_files.exists() stale_published_files.update(published=False) if stale_direct_resource_ids: LearningResource.objects.filter( @@ -1062,14 +1072,62 @@ def load_content_files( ): update_index(resource, newly_created=False) + changed_ids = _changed_content_file_ids(content_files_ids, prior_files) + # Emit the intra-run change ratio so the embedding-skip benefit is + # measurable in production (grep "content files changed" in the ETL logs). + log.info( + "run %s content files changed: %d of %d", + course_run.run_id, + len(changed_ids), + len(content_files_ids), + ) + # Past the cap, hand the task None (re-query published files itself) rather + # than serialize a large id list into the broker message. Empty stays empty + # (embed nothing); only large lists collapse to None. + embed_ids = ( + None + if len(changed_ids) > settings.CONTENT_FILE_EMBED_ID_CAP + else changed_ids + ) + if calc_completeness: calculate_completeness(course_run, content_tags=content_tags) - content_files_loaded_actions(run=course_run) + content_files_loaded_actions( + run=course_run, + content_file_ids=embed_ids, + removed_unpublished=removed_unpublished, + ) return content_files_ids return None +def _changed_content_file_ids( + content_files_ids: list[int], + prior_files: dict[str, tuple[str, bool]], +) -> list[int]: + """ + Return the subset of loaded content-file ids that need re-embedding. + + A file is "changed" when it is new, its checksum differs from the prior load, + or it transitioned from unpublished to published (an identical-checksum + republish still needs re-embedding because it was purged from Qdrant while + unpublished). + """ + changed_ids = [] + for cf_id, key, checksum, published in ContentFile.objects.filter( + id__in=content_files_ids + ).values_list("id", "key", "checksum", "published"): + prior = prior_files.get(key) + if ( + prior is None + or prior[0] != checksum + or (prior[1] is False and published is True) + ): + changed_ids.append(cf_id) + return changed_ids + + def load_learning_materials( course_run: LearningResourceRun, content_file_ids: list[int], diff --git a/learning_resources/etl/loaders_test.py b/learning_resources/etl/loaders_test.py index 9927387278..234569dad9 100644 --- a/learning_resources/etl/loaders_test.py +++ b/learning_resources/etl/loaders_test.py @@ -35,6 +35,7 @@ from learning_resources.etl.exceptions import ExtractException from learning_resources.etl.loaders import ( ProgramLoadResult, + _changed_content_file_ids, calculate_completeness, load_content_file, load_content_files, @@ -3806,3 +3807,67 @@ def test_load_learning_material(mocker, learning_material_exists): assert learning_material.url == content_file.url assert content_file.direct_learning_resource_id == learning_material.id + + +@pytest.mark.django_db +def test_changed_content_file_ids_detects_new_changed_and_republished(): + """ + Change detection flags new, checksum-changed, and republished (unpublished + -> published, identical checksum) files, and excludes unchanged files. + """ + run = LearningResourceRunFactory.create() + unchanged = ContentFileFactory.create(run=run, key="unchanged", published=True) + changed = ContentFileFactory.create(run=run, key="changed", published=True) + republished = ContentFileFactory.create(run=run, key="republished", published=True) + new_file = ContentFileFactory.create(run=run, key="new", published=True) + + prior_files = { + "unchanged": (unchanged.checksum, True), + "changed": ("stale-checksum", True), + # was unpublished last load, republished now with the same checksum + "republished": (republished.checksum, False), + # "new" absent from the prior snapshot entirely + } + + result = _changed_content_file_ids( + [unchanged.id, changed.id, republished.id, new_file.id], prior_files + ) + + assert sorted(result) == sorted([changed.id, republished.id, new_file.id]) + + +@pytest.mark.django_db +def test_load_content_files_caps_changed_ids_to_none(mocker, settings): + """ + When the changed-id list exceeds CONTENT_FILE_EMBED_ID_CAP, the hook is handed + None so embed_run_content_files re-queries instead of serializing a big list. + """ + settings.CONTENT_FILE_EMBED_ID_CAP = 2 + course = LearningResourceFactory.create(is_course=True, create_runs=False) + run = LearningResourceRunFactory.create(published=True, learning_resource=course) + mock_hook = mocker.patch( + "learning_resources.etl.loaders.content_files_loaded_actions", autospec=True + ) + payload = [{"key": f"f{i}", "content": f"content {i}"} for i in range(3)] + + load_content_files(run, payload) + + assert mock_hook.call_args.kwargs["content_file_ids"] is None + + +@pytest.mark.django_db +def test_load_content_files_passes_changed_ids_under_cap(mocker, settings): + """Under the cap, the exact changed ids are passed through to the hook.""" + settings.CONTENT_FILE_EMBED_ID_CAP = 100 + course = LearningResourceFactory.create(is_course=True, create_runs=False) + run = LearningResourceRunFactory.create(published=True, learning_resource=course) + mock_hook = mocker.patch( + "learning_resources.etl.loaders.content_files_loaded_actions", autospec=True + ) + payload = [{"key": f"f{i}", "content": f"content {i}"} for i in range(3)] + + ids = load_content_files(run, payload) + + passed = mock_hook.call_args.kwargs["content_file_ids"] + assert passed is not None + assert sorted(passed) == sorted(ids) diff --git a/learning_resources/hooks.py b/learning_resources/hooks.py index 36eb397094..0a12676e54 100644 --- a/learning_resources/hooks.py +++ b/learning_resources/hooks.py @@ -81,8 +81,17 @@ def offeror_delete(self, offeror): """Trigger actions to delete a learning resource offeror""" @hookspec - def content_files_loaded(self, run): - """Trigger actions after content files are loaded for a run""" + def content_files_loaded(self, run, content_file_ids, removed_unpublished): + """ + Trigger actions after content files are loaded for a run. + + Args: + run: the LearningResourceRun whose content files were loaded + content_file_ids: ids of the files that were created/changed this load, + or None to act on all of the run's files (backfill / republish) + removed_unpublished: whether any previously-published files were + unpublished this load, or None when unknown (legacy callers) + """ def get_plugin_manager(): diff --git a/learning_resources/utils.py b/learning_resources/utils.py index 01bc3947a7..63a92de070 100644 --- a/learning_resources/utils.py +++ b/learning_resources/utils.py @@ -435,13 +435,28 @@ def bulk_resources_unpublished_actions(resource_ids: list[int], resource_type: s ) -def content_files_loaded_actions(run: LearningResourceRun): +def content_files_loaded_actions( + run: LearningResourceRun, + content_file_ids: list[int] | None = None, + removed_unpublished: bool | None = None, # noqa: FBT001 +): """ - Trigger plugins when content files are loaded for a LearningResourceRun + Trigger plugins when content files are loaded for a LearningResourceRun. + + Args: + run: the LearningResourceRun whose content files were loaded + content_file_ids: ids of the files created/changed this load, or None to + act on all of the run's files (backfill / republish) + removed_unpublished: whether any previously-published files were + unpublished this load, or None when unknown (legacy callers) """ pm = get_plugin_manager() hook = pm.hook - hook.content_files_loaded(run=run) + hook.content_files_loaded( + run=run, + content_file_ids=content_file_ids, + removed_unpublished=removed_unpublished, + ) def resource_run_unpublished_actions(run: LearningResourceRun): diff --git a/learning_resources_search/plugins.py b/learning_resources_search/plugins.py index 7b62f42fcd..98254180df 100644 --- a/learning_resources_search/plugins.py +++ b/learning_resources_search/plugins.py @@ -257,16 +257,23 @@ def resource_run_delete(self, run): run.delete() @hookimpl - def content_files_loaded(self, run): + def content_files_loaded( + self, run, content_file_ids=None, removed_unpublished=None + ): """ Upsert a created/modified run's content files. - Qdrant: embed every loaded run (all runs of a published/test_mode course) - and drop stale files. OpenSearch: index only the best published non-B2B - run, or any published non-variant run of a test_mode course. + Qdrant: embed the changed files (or all files when content_file_ids is + None) and drop stale files. OpenSearch: index only the best published + non-B2B run, or any published non-variant run of a test_mode course. Args: run(LearningResourceRun): The LearningResourceRun that was upserted + content_file_ids: ids of the files created/changed this load, or None + to embed all of the run's published files (backfill / republish) + removed_unpublished: whether any previously-published files were + unpublished this load; None (unknown, legacy) still purges, only + an explicit False skips the removal task """ if not run.content_files.exists(): return @@ -283,10 +290,15 @@ def content_files_loaded(self, run): index_tasks.append(tasks.index_run_content_files.si(run.id)) if django_settings.QDRANT_ENABLE_INDEXING_PLUGIN_HOOKS: - index_tasks.append(vector_tasks.embed_run_content_files.si(run.id)) index_tasks.append( - vector_tasks.remove_unpublished_run_content_files.si(run.id) + vector_tasks.embed_run_content_files.si(run.id, content_file_ids) ) + # None (legacy/unknown) keeps the historical always-purge behavior; + # only load_content_files, which knows definitively, passes False. + if removed_unpublished is not False: + index_tasks.append( + vector_tasks.remove_unpublished_run_content_files.si(run.id) + ) if index_tasks: try_with_retry_as_task(chain(*index_tasks)) diff --git a/learning_resources_search/plugins_test.py b/learning_resources_search/plugins_test.py index 6ee40c540f..17b55605e3 100644 --- a/learning_resources_search/plugins_test.py +++ b/learning_resources_search/plugins_test.py @@ -387,7 +387,7 @@ def test_search_index_plugin_content_files_loaded_published_run_with_qdrant( SearchIndexPlugin().content_files_loaded(run) mock_search_index_helpers.mock_embed_run_contentfiles_immutable_signature.assert_called_once_with( - run.id + run.id, None ) mock_search_index_helpers.mock_upsert_contentfiles_immutable_signature.assert_called_once_with( run.id @@ -411,7 +411,7 @@ def test_content_files_loaded_unpublished_run_embeds_qdrant_only( SearchIndexPlugin().content_files_loaded(run) mock_search_index_helpers.mock_embed_run_contentfiles_immutable_signature.assert_called_once_with( - run.id + run.id, None ) mock_search_index_helpers.mock_remove_unpublished_run_contentfiles_immutable_signature.assert_called_once_with( run.id @@ -447,7 +447,7 @@ def test_content_files_loaded_non_best_published_run_skips_opensearch( SearchIndexPlugin().content_files_loaded(non_best) mock_search_index_helpers.mock_embed_run_contentfiles_immutable_signature.assert_called_once_with( - non_best.id + non_best.id, None ) mock_search_index_helpers.mock_upsert_contentfiles_immutable_signature.assert_not_called() @@ -470,7 +470,7 @@ def test_content_files_loaded_test_mode_published_run_indexes_opensearch( run.id ) mock_search_index_helpers.mock_embed_run_contentfiles_immutable_signature.assert_called_once_with( - run.id + run.id, None ) @@ -492,7 +492,7 @@ def test_content_files_loaded_variant_run_skips_opensearch( mock_search_index_helpers.mock_upsert_contentfiles_immutable_signature.assert_not_called() mock_search_index_helpers.mock_embed_run_contentfiles_immutable_signature.assert_called_once_with( - run.id + run.id, None ) mock_search_index_helpers.mock_remove_unpublished_run_contentfiles_immutable_signature.assert_called_once_with( run.id @@ -568,3 +568,29 @@ def test_search_index_plugin_resource_upserted_generate_embeddings( mock_search_index_helpers.mock_generate_embeddings_immutable_signature.assert_called_once_with( [resource.id], resource_type, overwrite=True ) + + +@pytest.mark.django_db +@pytest.mark.parametrize( + ("removed_unpublished", "expect_remove_called"), + [(True, True), (None, True), (False, False)], +) +def test_content_files_loaded_removed_unpublished_tristate( + mock_search_index_helpers, settings, removed_unpublished, expect_remove_called +): + """None (legacy) and True append the remove-unpublished task; only False skips.""" + settings.QDRANT_ENABLE_INDEXING_PLUGIN_HOOKS = True + run = LearningResourceRunFactory.create( + published=True, learning_resource__create_runs=False + ) + ContentFileFactory.create(run=run) + + SearchIndexPlugin().content_files_loaded( + run, content_file_ids=None, removed_unpublished=removed_unpublished + ) + + remove_mock = mock_search_index_helpers.mock_remove_unpublished_run_contentfiles_immutable_signature + if expect_remove_called: + remove_mock.assert_called_once_with(run.id) + else: + remove_mock.assert_not_called() diff --git a/main/settings.py b/main/settings.py index 604437bf01..1a02fb2f41 100644 --- a/main/settings.py +++ b/main/settings.py @@ -778,6 +778,14 @@ def get_all_config_keys(): default=10, ) +# Max changed content-file ids passed as an argument to embed_run_content_files. +# Above this, load_content_files passes None so the task re-queries the run's +# published files itself, keeping the broker message small on large re-ingests. +CONTENT_FILE_EMBED_ID_CAP = get_int( + name="CONTENT_FILE_EMBED_ID_CAP", + default=1000, +) + QDRANT_ENCODER = get_string( name="QDRANT_ENCODER", default="vector_search.encoders.gensim.GensimEncoder" ) diff --git a/vector_search/tasks.py b/vector_search/tasks.py index 4e842589bc..e7cd02a20b 100644 --- a/vector_search/tasks.py +++ b/vector_search/tasks.py @@ -463,15 +463,24 @@ def embed_new_content_files(self): @app.task(bind=True) -def embed_run_content_files(self, run_id): +def embed_run_content_files(self, run_id, content_file_ids=None): """ - Embed contentfiles associated with a run - """ - content_file_ids = list( - ContentFile.objects.filter(run__id=run_id).values_list("id", flat=True) - ) + Embed published content files associated with a run. - return _replace_with_finalized_chain(self, content_file_ids, overwrite=True) + Args: + run_id (int): the run whose content files to embed + content_file_ids (list of int or None): when provided, only these files are + embedded (the ETL change-detection path passes just the changed files); + when None, all published files for the run are embedded (backfill / + republish). + """ + content_files = ContentFile.objects.filter(run__id=run_id, published=True) + if content_file_ids is not None: + content_files = content_files.filter(id__in=content_file_ids) + ids = list(content_files.values_list("id", flat=True)) + if not ids: + return None + return _replace_with_finalized_chain(self, ids, overwrite=True) @app.task(bind=True) diff --git a/vector_search/tasks_test.py b/vector_search/tasks_test.py index 769339ba19..a07c388268 100644 --- a/vector_search/tasks_test.py +++ b/vector_search/tasks_test.py @@ -767,6 +767,57 @@ def test_embed_run_content_files_no_files_returns_none(mocker, mocked_celery): mocked_celery.chain.assert_not_called() +def test_embed_run_content_files_only_given_ids(mocker, mocked_celery, settings): + """When content_file_ids is passed, only those published files are embedded.""" + settings.QDRANT_CHUNK_SIZE = 50 + run = LearningResourceRunFactory.create() + files = ContentFileFactory.create_batch(3, run=run, published=True) + changed = [files[0].id, files[1].id] + generate_embeddings_mock = mocker.patch( + "vector_search.tasks.generate_embeddings", autospec=True + ) + + with pytest.raises(mocked_celery.replace_exception_class): + embed_run_content_files.delay(run.id, changed) + + embedded_ids = [ + content_file_id + for mock_call in generate_embeddings_mock.si.call_args_list + for content_file_id in mock_call.args[0] + ] + assert sorted(embedded_ids) == sorted(changed) + + +def test_embed_run_content_files_skips_unpublished(mocker, mocked_celery, settings): + """Unpublished files are never embedded, even if named in content_file_ids.""" + settings.QDRANT_CHUNK_SIZE = 50 + run = LearningResourceRunFactory.create() + published = ContentFileFactory.create(run=run, published=True) + unpublished = ContentFileFactory.create(run=run, published=False) + generate_embeddings_mock = mocker.patch( + "vector_search.tasks.generate_embeddings", autospec=True + ) + + with pytest.raises(mocked_celery.replace_exception_class): + embed_run_content_files.delay(run.id, [published.id, unpublished.id]) + + embedded_ids = [ + content_file_id + for mock_call in generate_embeddings_mock.si.call_args_list + for content_file_id in mock_call.args[0] + ] + assert embedded_ids == [published.id] + + +def test_embed_run_content_files_empty_ids_returns_none(mocker, mocked_celery): + """An explicit empty id list embeds nothing (no changed files this load).""" + run = LearningResourceRunFactory.create() + ContentFileFactory.create(run=run, published=True) + mocker.patch("vector_search.tasks.generate_embeddings", autospec=True) + assert embed_run_content_files(run.id, []) is None + mocked_celery.group.assert_not_called() + + def test_embeddings_healthcheck_no_missing_embeddings(mocker): """ Test embeddings_healthcheck when there are no missing embeddings From 1ffda7b1b5a80df79255c36ee520d9eac39f7f5d Mon Sep 17 00:00:00 2001 From: Matt Bertrand Date: Sun, 19 Jul 2026 14:06:34 -0400 Subject: [PATCH 02/10] Consolidate change-detection tests via parametrize and shared helper Merge the two CONTENT_FILE_EMBED_ID_CAP tests into one parametrized test and reuse the existing _embedded_content_file_ids helper in the embed_run_content_files id tests. No coverage change. Co-Authored-By: Claude Opus 4.8 (1M context) --- learning_resources/etl/loaders_test.py | 36 ++++++++++---------------- vector_search/tasks_test.py | 14 ++-------- 2 files changed, 16 insertions(+), 34 deletions(-) diff --git a/learning_resources/etl/loaders_test.py b/learning_resources/etl/loaders_test.py index 234569dad9..21794d3f6e 100644 --- a/learning_resources/etl/loaders_test.py +++ b/learning_resources/etl/loaders_test.py @@ -3837,28 +3837,18 @@ def test_changed_content_file_ids_detects_new_changed_and_republished(): @pytest.mark.django_db -def test_load_content_files_caps_changed_ids_to_none(mocker, settings): +@pytest.mark.parametrize( + ("cap", "expect_ids"), + # 3 changed files: cap=2 exceeds it → None; cap=100 → exact ids passed through + [(2, False), (100, True)], +) +def test_load_content_files_changed_id_cap(mocker, settings, cap, expect_ids): """ - When the changed-id list exceeds CONTENT_FILE_EMBED_ID_CAP, the hook is handed - None so embed_run_content_files re-queries instead of serializing a big list. + At/under CONTENT_FILE_EMBED_ID_CAP the exact changed ids reach the hook; over it + the hook is handed None so embed_run_content_files re-queries instead of + serializing a big list. """ - settings.CONTENT_FILE_EMBED_ID_CAP = 2 - course = LearningResourceFactory.create(is_course=True, create_runs=False) - run = LearningResourceRunFactory.create(published=True, learning_resource=course) - mock_hook = mocker.patch( - "learning_resources.etl.loaders.content_files_loaded_actions", autospec=True - ) - payload = [{"key": f"f{i}", "content": f"content {i}"} for i in range(3)] - - load_content_files(run, payload) - - assert mock_hook.call_args.kwargs["content_file_ids"] is None - - -@pytest.mark.django_db -def test_load_content_files_passes_changed_ids_under_cap(mocker, settings): - """Under the cap, the exact changed ids are passed through to the hook.""" - settings.CONTENT_FILE_EMBED_ID_CAP = 100 + settings.CONTENT_FILE_EMBED_ID_CAP = cap course = LearningResourceFactory.create(is_course=True, create_runs=False) run = LearningResourceRunFactory.create(published=True, learning_resource=course) mock_hook = mocker.patch( @@ -3869,5 +3859,7 @@ def test_load_content_files_passes_changed_ids_under_cap(mocker, settings): ids = load_content_files(run, payload) passed = mock_hook.call_args.kwargs["content_file_ids"] - assert passed is not None - assert sorted(passed) == sorted(ids) + if expect_ids: + assert sorted(passed) == sorted(ids) + else: + assert passed is None diff --git a/vector_search/tasks_test.py b/vector_search/tasks_test.py index a07c388268..00e8ba332b 100644 --- a/vector_search/tasks_test.py +++ b/vector_search/tasks_test.py @@ -780,12 +780,7 @@ def test_embed_run_content_files_only_given_ids(mocker, mocked_celery, settings) with pytest.raises(mocked_celery.replace_exception_class): embed_run_content_files.delay(run.id, changed) - embedded_ids = [ - content_file_id - for mock_call in generate_embeddings_mock.si.call_args_list - for content_file_id in mock_call.args[0] - ] - assert sorted(embedded_ids) == sorted(changed) + assert _embedded_content_file_ids(generate_embeddings_mock) == set(changed) def test_embed_run_content_files_skips_unpublished(mocker, mocked_celery, settings): @@ -801,12 +796,7 @@ def test_embed_run_content_files_skips_unpublished(mocker, mocked_celery, settin with pytest.raises(mocked_celery.replace_exception_class): embed_run_content_files.delay(run.id, [published.id, unpublished.id]) - embedded_ids = [ - content_file_id - for mock_call in generate_embeddings_mock.si.call_args_list - for content_file_id in mock_call.args[0] - ] - assert embedded_ids == [published.id] + assert _embedded_content_file_ids(generate_embeddings_mock) == {published.id} def test_embed_run_content_files_empty_ids_returns_none(mocker, mocked_celery): From 6ebebcdde8de668e5c01c6b1eea1c2c59414997c Mon Sep 17 00:00:00 2001 From: Matt Bertrand Date: Mon, 20 Jul 2026 15:52:51 -0400 Subject: [PATCH 03/10] Default content_files_loaded hookspec params to None Align the hookspec signature with its documented contract and every implementation, where content_file_ids/removed_unpublished are optional (None means backfill / unknown for legacy callers). Co-Authored-By: Claude Opus 4.8 (1M context) --- learning_resources/hooks.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/learning_resources/hooks.py b/learning_resources/hooks.py index 0a12676e54..dfbb3361c3 100644 --- a/learning_resources/hooks.py +++ b/learning_resources/hooks.py @@ -81,7 +81,9 @@ def offeror_delete(self, offeror): """Trigger actions to delete a learning resource offeror""" @hookspec - def content_files_loaded(self, run, content_file_ids, removed_unpublished): + def content_files_loaded( + self, run, content_file_ids=None, removed_unpublished=None + ): """ Trigger actions after content files are loaded for a run. From cdcc7659f0a3c3b4e75a0bf50a3b937d7231ebdc Mon Sep 17 00:00:00 2001 From: Matt Bertrand Date: Mon, 20 Jul 2026 16:24:43 -0400 Subject: [PATCH 04/10] Detect content-file metadata changes for Qdrant payload refresh checksum hashes only file content, so metadata-only changes (title, url, edx_module_id, etc.) to an identical-content file were skipped by change detection and their Qdrant payload went stale with no self-healing path. Snapshot the scalar payload columns and flag a file as changed when any differ; such files hit the cheap payload-refresh path without re-embedding. Co-Authored-By: Claude Opus 4.8 (1M context) --- learning_resources/etl/loaders.py | 43 +++++++++++++++++++------- learning_resources/etl/loaders_test.py | 32 ++++++++++++++----- 2 files changed, 56 insertions(+), 19 deletions(-) diff --git a/learning_resources/etl/loaders.py b/learning_resources/etl/loaders.py index 7676b63de3..457bcc4a26 100644 --- a/learning_resources/etl/loaders.py +++ b/learning_resources/etl/loaders.py @@ -1031,10 +1031,10 @@ def load_content_files( # Snapshot existing files so we can embed only what actually changed this # load, rather than re-embedding the whole run every time. prior_files = { - key: (checksum, published) - for key, checksum, published in ContentFile.objects.filter( + key: (checksum, published, tuple(metadata)) + for key, checksum, published, *metadata in ContentFile.objects.filter( run=course_run - ).values_list("key", "checksum", "published") + ).values_list("key", "checksum", "published", *_PAYLOAD_METADATA_FIELDS) } content_files_ids = [] @@ -1102,27 +1102,46 @@ def load_content_files( return None +# Scalar, ETL-settable ContentFile columns whose Qdrant payload isn't covered by +# `checksum` (content-only). A deliberate subset of QDRANT_CONTENT_FILE_PARAM_MAP +# (which also holds non-columns, run-level, and AI-generated fields). +_PAYLOAD_METADATA_FIELDS = ( + "title", + "description", + "url", + "file_type", + "file_extension", + "content_type", + "edx_module_id", +) + + def _changed_content_file_ids( content_files_ids: list[int], - prior_files: dict[str, tuple[str, bool]], + prior_files: dict[str, tuple[str, bool, tuple]], ) -> list[int]: """ Return the subset of loaded content-file ids that need re-embedding. A file is "changed" when it is new, its checksum differs from the prior load, - or it transitioned from unpublished to published (an identical-checksum - republish still needs re-embedding because it was purged from Qdrant while - unpublished). + it transitioned from unpublished to published (an identical-checksum republish + still needs re-embedding because it was purged from Qdrant while unpublished), + or a payload metadata field changed (needs a Qdrant payload refresh even + though the body text — and thus the checksum — is unchanged). """ changed_ids = [] - for cf_id, key, checksum, published in ContentFile.objects.filter( + for cf_id, key, checksum, published, *metadata in ContentFile.objects.filter( id__in=content_files_ids - ).values_list("id", "key", "checksum", "published"): + ).values_list("id", "key", "checksum", "published", *_PAYLOAD_METADATA_FIELDS): prior = prior_files.get(key) + if prior is None: + changed_ids.append(cf_id) + continue + prior_checksum, prior_published, prior_metadata = prior if ( - prior is None - or prior[0] != checksum - or (prior[1] is False and published is True) + prior_checksum != checksum + or (prior_published is False and published is True) + or prior_metadata != tuple(metadata) ): changed_ids.append(cf_id) return changed_ids diff --git a/learning_resources/etl/loaders_test.py b/learning_resources/etl/loaders_test.py index 21794d3f6e..b4aee27c90 100644 --- a/learning_resources/etl/loaders_test.py +++ b/learning_resources/etl/loaders_test.py @@ -34,6 +34,7 @@ from learning_resources.etl.edx_shared import sync_edx_course_files from learning_resources.etl.exceptions import ExtractException from learning_resources.etl.loaders import ( + _PAYLOAD_METADATA_FIELDS, ProgramLoadResult, _changed_content_file_ids, calculate_completeness, @@ -3812,28 +3813,45 @@ def test_load_learning_material(mocker, learning_material_exists): @pytest.mark.django_db def test_changed_content_file_ids_detects_new_changed_and_republished(): """ - Change detection flags new, checksum-changed, and republished (unpublished - -> published, identical checksum) files, and excludes unchanged files. + Change detection flags new, checksum-changed, republished (unpublished -> + published, identical checksum), and metadata-only-changed files, and excludes + unchanged files. """ run = LearningResourceRunFactory.create() unchanged = ContentFileFactory.create(run=run, key="unchanged", published=True) changed = ContentFileFactory.create(run=run, key="changed", published=True) republished = ContentFileFactory.create(run=run, key="republished", published=True) new_file = ContentFileFactory.create(run=run, key="new", published=True) + # identical content (checksum) but a payload metadata field (title) differs + metadata_only = ContentFileFactory.create( + run=run, key="metadata_only", published=True, title="new title" + ) + + def meta(cf): + return tuple(getattr(cf, field) for field in _PAYLOAD_METADATA_FIELDS) prior_files = { - "unchanged": (unchanged.checksum, True), - "changed": ("stale-checksum", True), + "unchanged": (unchanged.checksum, True, meta(unchanged)), + "changed": ("stale-checksum", True, meta(changed)), # was unpublished last load, republished now with the same checksum - "republished": (republished.checksum, False), + "republished": (republished.checksum, False, meta(republished)), + # same checksum, but title was "old title" last load (index 0 of meta) + "metadata_only": ( + metadata_only.checksum, + True, + ("old title", *meta(metadata_only)[1:]), + ), # "new" absent from the prior snapshot entirely } result = _changed_content_file_ids( - [unchanged.id, changed.id, republished.id, new_file.id], prior_files + [unchanged.id, changed.id, republished.id, new_file.id, metadata_only.id], + prior_files, ) - assert sorted(result) == sorted([changed.id, republished.id, new_file.id]) + assert sorted(result) == sorted( + [changed.id, republished.id, new_file.id, metadata_only.id] + ) @pytest.mark.django_db From 23e60c4853858fc511bfd7daa7028327b9235185 Mon Sep 17 00:00:00 2001 From: Matt Bertrand Date: Mon, 20 Jul 2026 16:40:58 -0400 Subject: [PATCH 05/10] Make content_files_loaded_actions new params keyword-only Co-Authored-By: Claude Opus 4.8 (1M context) --- learning_resources/utils.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/learning_resources/utils.py b/learning_resources/utils.py index 63a92de070..456d92bdd2 100644 --- a/learning_resources/utils.py +++ b/learning_resources/utils.py @@ -437,8 +437,9 @@ def bulk_resources_unpublished_actions(resource_ids: list[int], resource_type: s def content_files_loaded_actions( run: LearningResourceRun, + *, content_file_ids: list[int] | None = None, - removed_unpublished: bool | None = None, # noqa: FBT001 + removed_unpublished: bool | None = None, ): """ Trigger plugins when content files are loaded for a LearningResourceRun. From 712b20d587f2883fb72438298b8f555b8c796cbe Mon Sep 17 00:00:00 2001 From: Matt Bertrand Date: Tue, 21 Jul 2026 09:11:46 -0400 Subject: [PATCH 06/10] Some cleanup --- learning_resources/etl/constants.py | 14 ++++++++++++++ learning_resources/etl/loaders.py | 23 +++++++---------------- learning_resources/etl/loaders_test.py | 6 ++++-- learning_resources/hooks.py | 6 ++---- learning_resources/utils.py | 6 ++---- learning_resources_search/plugins.py | 12 +++++------- vector_search/tasks.py | 6 ++---- 7 files changed, 36 insertions(+), 37 deletions(-) diff --git a/learning_resources/etl/constants.py b/learning_resources/etl/constants.py index 7f6351e55e..98a9ae9db8 100644 --- a/learning_resources/etl/constants.py +++ b/learning_resources/etl/constants.py @@ -185,3 +185,17 @@ class CommitmentConfig: commitment: str = "" min_weekly_hours: int = None max_weekly_hours: int = None + + +# Scalar, ETL-settable ContentFile columns whose Qdrant payload isn't covered by +# `checksum` (content-only). A deliberate subset of QDRANT_CONTENT_FILE_PARAM_MAP +# (which also holds non-columns, run-level, and AI-generated fields). +CONTENT_FILE_PAYLOAD_METADATA_FIELDS = ( + "title", + "description", + "url", + "file_type", + "file_extension", + "content_type", + "edx_module_id", +) diff --git a/learning_resources/etl/loaders.py b/learning_resources/etl/loaders.py index 457bcc4a26..b6ee1ffa95 100644 --- a/learning_resources/etl/loaders.py +++ b/learning_resources/etl/loaders.py @@ -24,6 +24,7 @@ RunStatus, ) from learning_resources.etl.constants import ( + CONTENT_FILE_PAYLOAD_METADATA_FIELDS, CONTENT_TAG_CATEGORIES, READABLE_ID_FIELD, ContentTagCategory, @@ -1034,7 +1035,9 @@ def load_content_files( key: (checksum, published, tuple(metadata)) for key, checksum, published, *metadata in ContentFile.objects.filter( run=course_run - ).values_list("key", "checksum", "published", *_PAYLOAD_METADATA_FIELDS) + ).values_list( + "key", "checksum", "published", *CONTENT_FILE_PAYLOAD_METADATA_FIELDS + ) } content_files_ids = [] @@ -1102,20 +1105,6 @@ def load_content_files( return None -# Scalar, ETL-settable ContentFile columns whose Qdrant payload isn't covered by -# `checksum` (content-only). A deliberate subset of QDRANT_CONTENT_FILE_PARAM_MAP -# (which also holds non-columns, run-level, and AI-generated fields). -_PAYLOAD_METADATA_FIELDS = ( - "title", - "description", - "url", - "file_type", - "file_extension", - "content_type", - "edx_module_id", -) - - def _changed_content_file_ids( content_files_ids: list[int], prior_files: dict[str, tuple[str, bool, tuple]], @@ -1132,7 +1121,9 @@ def _changed_content_file_ids( changed_ids = [] for cf_id, key, checksum, published, *metadata in ContentFile.objects.filter( id__in=content_files_ids - ).values_list("id", "key", "checksum", "published", *_PAYLOAD_METADATA_FIELDS): + ).values_list( + "id", "key", "checksum", "published", *CONTENT_FILE_PAYLOAD_METADATA_FIELDS + ): prior = prior_files.get(key) if prior is None: changed_ids.append(cf_id) diff --git a/learning_resources/etl/loaders_test.py b/learning_resources/etl/loaders_test.py index b4aee27c90..bda7a20b33 100644 --- a/learning_resources/etl/loaders_test.py +++ b/learning_resources/etl/loaders_test.py @@ -27,6 +27,7 @@ ) from learning_resources.etl import loaders from learning_resources.etl.constants import ( + CONTENT_FILE_PAYLOAD_METADATA_FIELDS, CourseLoaderConfig, ETLSource, ProgramLoaderConfig, @@ -34,7 +35,6 @@ from learning_resources.etl.edx_shared import sync_edx_course_files from learning_resources.etl.exceptions import ExtractException from learning_resources.etl.loaders import ( - _PAYLOAD_METADATA_FIELDS, ProgramLoadResult, _changed_content_file_ids, calculate_completeness, @@ -3828,7 +3828,9 @@ def test_changed_content_file_ids_detects_new_changed_and_republished(): ) def meta(cf): - return tuple(getattr(cf, field) for field in _PAYLOAD_METADATA_FIELDS) + return tuple( + getattr(cf, field) for field in CONTENT_FILE_PAYLOAD_METADATA_FIELDS + ) prior_files = { "unchanged": (unchanged.checksum, True, meta(unchanged)), diff --git a/learning_resources/hooks.py b/learning_resources/hooks.py index dfbb3361c3..7d42fa53c5 100644 --- a/learning_resources/hooks.py +++ b/learning_resources/hooks.py @@ -89,10 +89,8 @@ def content_files_loaded( Args: run: the LearningResourceRun whose content files were loaded - content_file_ids: ids of the files that were created/changed this load, - or None to act on all of the run's files (backfill / republish) - removed_unpublished: whether any previously-published files were - unpublished this load, or None when unknown (legacy callers) + content_file_ids: ids of the changed files, or None for all files + removed_unpublished: whether files were unpublished, or None if unknown """ diff --git a/learning_resources/utils.py b/learning_resources/utils.py index 456d92bdd2..8cc2586788 100644 --- a/learning_resources/utils.py +++ b/learning_resources/utils.py @@ -446,10 +446,8 @@ def content_files_loaded_actions( Args: run: the LearningResourceRun whose content files were loaded - content_file_ids: ids of the files created/changed this load, or None to - act on all of the run's files (backfill / republish) - removed_unpublished: whether any previously-published files were - unpublished this load, or None when unknown (legacy callers) + content_file_ids: ids of the changed files, or None for all files + removed_unpublished: whether files were unpublished, or None if unknown """ pm = get_plugin_manager() hook = pm.hook diff --git a/learning_resources_search/plugins.py b/learning_resources_search/plugins.py index 98254180df..0d08f93363 100644 --- a/learning_resources_search/plugins.py +++ b/learning_resources_search/plugins.py @@ -267,13 +267,11 @@ def content_files_loaded( None) and drop stale files. OpenSearch: index only the best published non-B2B run, or any published non-variant run of a test_mode course. - Args: - run(LearningResourceRun): The LearningResourceRun that was upserted - content_file_ids: ids of the files created/changed this load, or None - to embed all of the run's published files (backfill / republish) - removed_unpublished: whether any previously-published files were - unpublished this load; None (unknown, legacy) still purges, only - an explicit False skips the removal task + Args: + run: the LearningResourceRun that was upserted + content_file_ids: ids of the changed files, or None for all files + removed_unpublished: whether files were unpublished; only an explicit + False skips the removal task (None still purges) """ if not run.content_files.exists(): return diff --git a/vector_search/tasks.py b/vector_search/tasks.py index e7cd02a20b..cadb40187d 100644 --- a/vector_search/tasks.py +++ b/vector_search/tasks.py @@ -469,10 +469,8 @@ def embed_run_content_files(self, run_id, content_file_ids=None): Args: run_id (int): the run whose content files to embed - content_file_ids (list of int or None): when provided, only these files are - embedded (the ETL change-detection path passes just the changed files); - when None, all published files for the run are embedded (backfill / - republish). + content_file_ids (list of int or None): only these files are embedded, or + all of the run's published files when None (backfill / republish) """ content_files = ContentFile.objects.filter(run__id=run_id, published=True) if content_file_ids is not None: From 7d3467852d781fa85bdc5b585c7d6a4dc5cbd361 Mon Sep 17 00:00:00 2001 From: Matt Bertrand Date: Tue, 21 Jul 2026 09:29:20 -0400 Subject: [PATCH 07/10] one more test --- learning_resources/etl/loaders_test.py | 39 ++++++++++++++++++++++++++ 1 file changed, 39 insertions(+) diff --git a/learning_resources/etl/loaders_test.py b/learning_resources/etl/loaders_test.py index bda7a20b33..8719e09738 100644 --- a/learning_resources/etl/loaders_test.py +++ b/learning_resources/etl/loaders_test.py @@ -3883,3 +3883,42 @@ def test_load_content_files_changed_id_cap(mocker, settings, cap, expect_ids): assert sorted(passed) == sorted(ids) else: assert passed is None + + +@pytest.mark.django_db +def test_load_content_files_reload_embeds_only_changed(mocker, settings): + """ + Reloading a run embeds only the file whose content changed: the snapshot is + taken before the update, unchanged files are excluded, and a file dropped from + the payload is flagged as removed_unpublished. + """ + settings.CONTENT_FILE_EMBED_ID_CAP = 100 + course = LearningResourceFactory.create(is_course=True, create_runs=False) + run = LearningResourceRunFactory.create(published=True, learning_resource=course) + mock_hook = mocker.patch( + "learning_resources.etl.loaders.content_files_loaded_actions", autospec=True + ) + + # First load establishes the prior snapshot (checksums) in the DB. + load_content_files( + run, + [ + {"key": "keep", "content": "same"}, + {"key": "change", "content": "v1"}, + {"key": "drop", "content": "gone"}, + ], + ) + + # Second load: "keep" identical, "change" has new content, "drop" is absent. + load_content_files( + run, + [ + {"key": "keep", "content": "same"}, + {"key": "change", "content": "v2"}, + ], + ) + + changed_id = ContentFile.objects.get(run=run, key="change").id + kwargs = mock_hook.call_args.kwargs + assert kwargs["content_file_ids"] == [changed_id] + assert kwargs["removed_unpublished"] is True From cad36237c262d7b0166648864c643be842359682 Mon Sep 17 00:00:00 2001 From: Matt Bertrand Date: Tue, 21 Jul 2026 14:02:22 -0400 Subject: [PATCH 08/10] Replace DB change detection with batched Qdrant checksum gating Per review: the DB-diff approach compared pre/post-load DB state, but the ETL doesn't wait for the async embed chain, so a permanently-failed embed left Qdrant stale forever (DB already showed the new checksum). Qdrant's stored payload checksum is the correct source of truth and self-heals: missing or stale points re-qualify on the next load. - Remove _changed_content_file_ids, the prior-files snapshot, the content_file_ids hook plumbing, and CONTENT_FILE_EMBED_ID_CAP; keep the removed_unpublished tri-state and published-only embed filtering. - Batch the Qdrant payload lookups (_stored_content_payloads): one retrieve per batch serves the existence filter, summary-change check, and embed gate, replacing the per-file retrieves. - Add a run-level pre-pass to embed_run_content_files: compare DB checksums and payload metadata columns (title, summary, flashcards, etc.) against the stored Qdrant payload from lightweight DB fields before serializing anything. Metadata-only drift is dispatched but exits via the cheap payload-only update path; a fully-unchanged 1,770-file run drops from ~31s (full serialization + per-file payload rewrites) to ~0.09s and dispatches no embedding tasks. Co-Authored-By: Claude Opus 4.8 (1M context) --- learning_resources/etl/constants.py | 14 -- learning_resources/etl/loaders.py | 64 -------- learning_resources/etl/loaders_test.py | 104 +----------- learning_resources/hooks.py | 5 +- learning_resources/utils.py | 8 +- learning_resources_search/plugins.py | 16 +- learning_resources_search/plugins_test.py | 12 +- main/settings.py | 8 - vector_search/constants.py | 16 ++ vector_search/tasks.py | 81 ++++++++-- vector_search/tasks_test.py | 150 +++++++++++++++--- vector_search/utils.py | 89 ++++++++--- vector_search/utils_test.py | 185 +++++++++++++--------- 13 files changed, 416 insertions(+), 336 deletions(-) diff --git a/learning_resources/etl/constants.py b/learning_resources/etl/constants.py index 98a9ae9db8..7f6351e55e 100644 --- a/learning_resources/etl/constants.py +++ b/learning_resources/etl/constants.py @@ -185,17 +185,3 @@ class CommitmentConfig: commitment: str = "" min_weekly_hours: int = None max_weekly_hours: int = None - - -# Scalar, ETL-settable ContentFile columns whose Qdrant payload isn't covered by -# `checksum` (content-only). A deliberate subset of QDRANT_CONTENT_FILE_PARAM_MAP -# (which also holds non-columns, run-level, and AI-generated fields). -CONTENT_FILE_PAYLOAD_METADATA_FIELDS = ( - "title", - "description", - "url", - "file_type", - "file_extension", - "content_type", - "edx_module_id", -) diff --git a/learning_resources/etl/loaders.py b/learning_resources/etl/loaders.py index b6ee1ffa95..5c40f23787 100644 --- a/learning_resources/etl/loaders.py +++ b/learning_resources/etl/loaders.py @@ -24,7 +24,6 @@ RunStatus, ) from learning_resources.etl.constants import ( - CONTENT_FILE_PAYLOAD_METADATA_FIELDS, CONTENT_TAG_CATEGORIES, READABLE_ID_FIELD, ContentTagCategory, @@ -1029,17 +1028,6 @@ def load_content_files( """ if course_run.learning_resource.resource_type == LearningResourceType.course.name: - # Snapshot existing files so we can embed only what actually changed this - # load, rather than re-embedding the whole run every time. - prior_files = { - key: (checksum, published, tuple(metadata)) - for key, checksum, published, *metadata in ContentFile.objects.filter( - run=course_run - ).values_list( - "key", "checksum", "published", *CONTENT_FILE_PAYLOAD_METADATA_FIELDS - ) - } - content_files_ids = [] content_tags = [] for content_file in content_files_data: @@ -1075,29 +1063,10 @@ def load_content_files( ): update_index(resource, newly_created=False) - changed_ids = _changed_content_file_ids(content_files_ids, prior_files) - # Emit the intra-run change ratio so the embedding-skip benefit is - # measurable in production (grep "content files changed" in the ETL logs). - log.info( - "run %s content files changed: %d of %d", - course_run.run_id, - len(changed_ids), - len(content_files_ids), - ) - # Past the cap, hand the task None (re-query published files itself) rather - # than serialize a large id list into the broker message. Empty stays empty - # (embed nothing); only large lists collapse to None. - embed_ids = ( - None - if len(changed_ids) > settings.CONTENT_FILE_EMBED_ID_CAP - else changed_ids - ) - if calc_completeness: calculate_completeness(course_run, content_tags=content_tags) content_files_loaded_actions( run=course_run, - content_file_ids=embed_ids, removed_unpublished=removed_unpublished, ) @@ -1105,39 +1074,6 @@ def load_content_files( return None -def _changed_content_file_ids( - content_files_ids: list[int], - prior_files: dict[str, tuple[str, bool, tuple]], -) -> list[int]: - """ - Return the subset of loaded content-file ids that need re-embedding. - - A file is "changed" when it is new, its checksum differs from the prior load, - it transitioned from unpublished to published (an identical-checksum republish - still needs re-embedding because it was purged from Qdrant while unpublished), - or a payload metadata field changed (needs a Qdrant payload refresh even - though the body text — and thus the checksum — is unchanged). - """ - changed_ids = [] - for cf_id, key, checksum, published, *metadata in ContentFile.objects.filter( - id__in=content_files_ids - ).values_list( - "id", "key", "checksum", "published", *CONTENT_FILE_PAYLOAD_METADATA_FIELDS - ): - prior = prior_files.get(key) - if prior is None: - changed_ids.append(cf_id) - continue - prior_checksum, prior_published, prior_metadata = prior - if ( - prior_checksum != checksum - or (prior_published is False and published is True) - or prior_metadata != tuple(metadata) - ): - changed_ids.append(cf_id) - return changed_ids - - def load_learning_materials( course_run: LearningResourceRun, content_file_ids: list[int], diff --git a/learning_resources/etl/loaders_test.py b/learning_resources/etl/loaders_test.py index 8719e09738..13832e664f 100644 --- a/learning_resources/etl/loaders_test.py +++ b/learning_resources/etl/loaders_test.py @@ -27,7 +27,6 @@ ) from learning_resources.etl import loaders from learning_resources.etl.constants import ( - CONTENT_FILE_PAYLOAD_METADATA_FIELDS, CourseLoaderConfig, ETLSource, ProgramLoaderConfig, @@ -36,7 +35,6 @@ from learning_resources.etl.exceptions import ExtractException from learning_resources.etl.loaders import ( ProgramLoadResult, - _changed_content_file_ids, calculate_completeness, load_content_file, load_content_files, @@ -3811,114 +3809,26 @@ def test_load_learning_material(mocker, learning_material_exists): @pytest.mark.django_db -def test_changed_content_file_ids_detects_new_changed_and_republished(): +def test_load_content_files_flags_removed_unpublished(mocker): """ - Change detection flags new, checksum-changed, republished (unpublished -> - published, identical checksum), and metadata-only-changed files, and excludes - unchanged files. + The hook receives removed_unpublished=False when every prior file is still + present, and True when a previously-published file is dropped from the payload. """ - run = LearningResourceRunFactory.create() - unchanged = ContentFileFactory.create(run=run, key="unchanged", published=True) - changed = ContentFileFactory.create(run=run, key="changed", published=True) - republished = ContentFileFactory.create(run=run, key="republished", published=True) - new_file = ContentFileFactory.create(run=run, key="new", published=True) - # identical content (checksum) but a payload metadata field (title) differs - metadata_only = ContentFileFactory.create( - run=run, key="metadata_only", published=True, title="new title" - ) - - def meta(cf): - return tuple( - getattr(cf, field) for field in CONTENT_FILE_PAYLOAD_METADATA_FIELDS - ) - - prior_files = { - "unchanged": (unchanged.checksum, True, meta(unchanged)), - "changed": ("stale-checksum", True, meta(changed)), - # was unpublished last load, republished now with the same checksum - "republished": (republished.checksum, False, meta(republished)), - # same checksum, but title was "old title" last load (index 0 of meta) - "metadata_only": ( - metadata_only.checksum, - True, - ("old title", *meta(metadata_only)[1:]), - ), - # "new" absent from the prior snapshot entirely - } - - result = _changed_content_file_ids( - [unchanged.id, changed.id, republished.id, new_file.id, metadata_only.id], - prior_files, - ) - - assert sorted(result) == sorted( - [changed.id, republished.id, new_file.id, metadata_only.id] - ) - - -@pytest.mark.django_db -@pytest.mark.parametrize( - ("cap", "expect_ids"), - # 3 changed files: cap=2 exceeds it → None; cap=100 → exact ids passed through - [(2, False), (100, True)], -) -def test_load_content_files_changed_id_cap(mocker, settings, cap, expect_ids): - """ - At/under CONTENT_FILE_EMBED_ID_CAP the exact changed ids reach the hook; over it - the hook is handed None so embed_run_content_files re-queries instead of - serializing a big list. - """ - settings.CONTENT_FILE_EMBED_ID_CAP = cap - course = LearningResourceFactory.create(is_course=True, create_runs=False) - run = LearningResourceRunFactory.create(published=True, learning_resource=course) - mock_hook = mocker.patch( - "learning_resources.etl.loaders.content_files_loaded_actions", autospec=True - ) - payload = [{"key": f"f{i}", "content": f"content {i}"} for i in range(3)] - - ids = load_content_files(run, payload) - - passed = mock_hook.call_args.kwargs["content_file_ids"] - if expect_ids: - assert sorted(passed) == sorted(ids) - else: - assert passed is None - - -@pytest.mark.django_db -def test_load_content_files_reload_embeds_only_changed(mocker, settings): - """ - Reloading a run embeds only the file whose content changed: the snapshot is - taken before the update, unchanged files are excluded, and a file dropped from - the payload is flagged as removed_unpublished. - """ - settings.CONTENT_FILE_EMBED_ID_CAP = 100 course = LearningResourceFactory.create(is_course=True, create_runs=False) run = LearningResourceRunFactory.create(published=True, learning_resource=course) mock_hook = mocker.patch( "learning_resources.etl.loaders.content_files_loaded_actions", autospec=True ) - # First load establishes the prior snapshot (checksums) in the DB. load_content_files( run, [ {"key": "keep", "content": "same"}, - {"key": "change", "content": "v1"}, {"key": "drop", "content": "gone"}, ], ) + assert mock_hook.call_args.kwargs["removed_unpublished"] is False - # Second load: "keep" identical, "change" has new content, "drop" is absent. - load_content_files( - run, - [ - {"key": "keep", "content": "same"}, - {"key": "change", "content": "v2"}, - ], - ) - - changed_id = ContentFile.objects.get(run=run, key="change").id - kwargs = mock_hook.call_args.kwargs - assert kwargs["content_file_ids"] == [changed_id] - assert kwargs["removed_unpublished"] is True + # Second load: "drop" is absent, so its file gets unpublished. + load_content_files(run, [{"key": "keep", "content": "same"}]) + assert mock_hook.call_args.kwargs["removed_unpublished"] is True diff --git a/learning_resources/hooks.py b/learning_resources/hooks.py index 7d42fa53c5..becb9c8534 100644 --- a/learning_resources/hooks.py +++ b/learning_resources/hooks.py @@ -81,15 +81,12 @@ def offeror_delete(self, offeror): """Trigger actions to delete a learning resource offeror""" @hookspec - def content_files_loaded( - self, run, content_file_ids=None, removed_unpublished=None - ): + def content_files_loaded(self, run, removed_unpublished=None): """ Trigger actions after content files are loaded for a run. Args: run: the LearningResourceRun whose content files were loaded - content_file_ids: ids of the changed files, or None for all files removed_unpublished: whether files were unpublished, or None if unknown """ diff --git a/learning_resources/utils.py b/learning_resources/utils.py index 8cc2586788..d082dcd077 100644 --- a/learning_resources/utils.py +++ b/learning_resources/utils.py @@ -438,7 +438,6 @@ def bulk_resources_unpublished_actions(resource_ids: list[int], resource_type: s def content_files_loaded_actions( run: LearningResourceRun, *, - content_file_ids: list[int] | None = None, removed_unpublished: bool | None = None, ): """ @@ -446,16 +445,11 @@ def content_files_loaded_actions( Args: run: the LearningResourceRun whose content files were loaded - content_file_ids: ids of the changed files, or None for all files removed_unpublished: whether files were unpublished, or None if unknown """ pm = get_plugin_manager() hook = pm.hook - hook.content_files_loaded( - run=run, - content_file_ids=content_file_ids, - removed_unpublished=removed_unpublished, - ) + hook.content_files_loaded(run=run, removed_unpublished=removed_unpublished) def resource_run_unpublished_actions(run: LearningResourceRun): diff --git a/learning_resources_search/plugins.py b/learning_resources_search/plugins.py index 0d08f93363..87100cc986 100644 --- a/learning_resources_search/plugins.py +++ b/learning_resources_search/plugins.py @@ -257,19 +257,17 @@ def resource_run_delete(self, run): run.delete() @hookimpl - def content_files_loaded( - self, run, content_file_ids=None, removed_unpublished=None - ): + def content_files_loaded(self, run, removed_unpublished=None): """ Upsert a created/modified run's content files. - Qdrant: embed the changed files (or all files when content_file_ids is - None) and drop stale files. OpenSearch: index only the best published - non-B2B run, or any published non-variant run of a test_mode course. + Qdrant: embed the run's published files (unchanged files exit via the + checksum gate in vector_search) and drop stale files. OpenSearch: index + only the best published non-B2B run, or any published non-variant run + of a test_mode course. Args: run: the LearningResourceRun that was upserted - content_file_ids: ids of the changed files, or None for all files removed_unpublished: whether files were unpublished; only an explicit False skips the removal task (None still purges) """ @@ -288,9 +286,7 @@ def content_files_loaded( index_tasks.append(tasks.index_run_content_files.si(run.id)) if django_settings.QDRANT_ENABLE_INDEXING_PLUGIN_HOOKS: - index_tasks.append( - vector_tasks.embed_run_content_files.si(run.id, content_file_ids) - ) + index_tasks.append(vector_tasks.embed_run_content_files.si(run.id)) # None (legacy/unknown) keeps the historical always-purge behavior; # only load_content_files, which knows definitively, passes False. if removed_unpublished is not False: diff --git a/learning_resources_search/plugins_test.py b/learning_resources_search/plugins_test.py index 17b55605e3..864862292f 100644 --- a/learning_resources_search/plugins_test.py +++ b/learning_resources_search/plugins_test.py @@ -387,7 +387,7 @@ def test_search_index_plugin_content_files_loaded_published_run_with_qdrant( SearchIndexPlugin().content_files_loaded(run) mock_search_index_helpers.mock_embed_run_contentfiles_immutable_signature.assert_called_once_with( - run.id, None + run.id ) mock_search_index_helpers.mock_upsert_contentfiles_immutable_signature.assert_called_once_with( run.id @@ -411,7 +411,7 @@ def test_content_files_loaded_unpublished_run_embeds_qdrant_only( SearchIndexPlugin().content_files_loaded(run) mock_search_index_helpers.mock_embed_run_contentfiles_immutable_signature.assert_called_once_with( - run.id, None + run.id ) mock_search_index_helpers.mock_remove_unpublished_run_contentfiles_immutable_signature.assert_called_once_with( run.id @@ -447,7 +447,7 @@ def test_content_files_loaded_non_best_published_run_skips_opensearch( SearchIndexPlugin().content_files_loaded(non_best) mock_search_index_helpers.mock_embed_run_contentfiles_immutable_signature.assert_called_once_with( - non_best.id, None + non_best.id ) mock_search_index_helpers.mock_upsert_contentfiles_immutable_signature.assert_not_called() @@ -470,7 +470,7 @@ def test_content_files_loaded_test_mode_published_run_indexes_opensearch( run.id ) mock_search_index_helpers.mock_embed_run_contentfiles_immutable_signature.assert_called_once_with( - run.id, None + run.id ) @@ -492,7 +492,7 @@ def test_content_files_loaded_variant_run_skips_opensearch( mock_search_index_helpers.mock_upsert_contentfiles_immutable_signature.assert_not_called() mock_search_index_helpers.mock_embed_run_contentfiles_immutable_signature.assert_called_once_with( - run.id, None + run.id ) mock_search_index_helpers.mock_remove_unpublished_run_contentfiles_immutable_signature.assert_called_once_with( run.id @@ -586,7 +586,7 @@ def test_content_files_loaded_removed_unpublished_tristate( ContentFileFactory.create(run=run) SearchIndexPlugin().content_files_loaded( - run, content_file_ids=None, removed_unpublished=removed_unpublished + run, removed_unpublished=removed_unpublished ) remove_mock = mock_search_index_helpers.mock_remove_unpublished_run_contentfiles_immutable_signature diff --git a/main/settings.py b/main/settings.py index 1a02fb2f41..604437bf01 100644 --- a/main/settings.py +++ b/main/settings.py @@ -778,14 +778,6 @@ def get_all_config_keys(): default=10, ) -# Max changed content-file ids passed as an argument to embed_run_content_files. -# Above this, load_content_files passes None so the task re-queries the run's -# published files itself, keeping the broker message small on large re-ingests. -CONTENT_FILE_EMBED_ID_CAP = get_int( - name="CONTENT_FILE_EMBED_ID_CAP", - default=1000, -) - QDRANT_ENCODER = get_string( name="QDRANT_ENCODER", default="vector_search.encoders.gensim.GensimEncoder" ) diff --git a/vector_search/constants.py b/vector_search/constants.py index 2302fae0bf..85ef1f42dc 100644 --- a/vector_search/constants.py +++ b/vector_search/constants.py @@ -5,6 +5,22 @@ CONTENT_FILES_COLLECTION_NAME = f"{settings.QDRANT_BASE_COLLECTION_NAME}.content_files" TOPICS_COLLECTION_NAME = f"{settings.QDRANT_BASE_COLLECTION_NAME}.topics" +# ContentFile columns (beyond checksum, which only covers content) compared by the +# embed_run_content_files pre-pass to detect stale Qdrant payloads. Every entry MUST +# be an exact serializer pass-through of a scalar/JSON ContentFile column: a field +# the serializer transforms would never converge, flagging every file on every load +# (test_content_file_prepass_fields_are_serializer_pass_through guards this). +CONTENT_FILE_PREPASS_PAYLOAD_FIELDS = ( + "title", + "description", + "url", + "file_type", + "file_extension", + "content_type", + "edx_module_id", + "summary", + "flashcards", +) QDRANT_CONTENT_FILE_PARAM_MAP = { "key": "key", diff --git a/vector_search/tasks.py b/vector_search/tasks.py index cadb40187d..4559594033 100644 --- a/vector_search/tasks.py +++ b/vector_search/tasks.py @@ -15,6 +15,7 @@ ContentFile, Course, LearningResource, + LearningResourceRun, ) from learning_resources.serializers import ( ContentFileSerializer, @@ -39,10 +40,12 @@ now_in_utc, ) from vector_search.constants import ( + CONTENT_FILE_PREPASS_PAYLOAD_FIELDS, CONTENT_FILES_COLLECTION_NAME, RESOURCES_COLLECTION_NAME, ) from vector_search.utils import ( + _stored_content_payloads, embed_learning_resources, embed_topics, filter_existing_qdrant_points_by_ids, @@ -463,19 +466,75 @@ def embed_new_content_files(self): @app.task(bind=True) -def embed_run_content_files(self, run_id, content_file_ids=None): +def embed_run_content_files(self, run_id): """ - Embed published content files associated with a run. - - Args: - run_id (int): the run whose content files to embed - content_file_ids (list of int or None): only these files are embedded, or - all of the run's published files when None (backfill / republish) + Embed the run's published content files whose Qdrant points are missing or + stale (checksum or a payload metadata field differs). + + A run-level pre-pass batch-compares each file's DB checksum and payload + metadata columns against the stored Qdrant payload, so a fully-unchanged + run costs one DB query plus a few batched retrieves instead of serializing + every file. A checksum-matching file with drifted metadata (edited title, + newly generated summary, ...) is dispatched but exits via the payload-only + update path downstream — no re-embedding. Failed or purged embeds show up + as missing/stale points, so they self-heal on the next load. """ - content_files = ContentFile.objects.filter(run__id=run_id, published=True) - if content_file_ids is not None: - content_files = content_files.filter(id__in=content_file_ids) - ids = list(content_files.values_list("id", flat=True)) + run = ( + LearningResourceRun.objects.select_related("learning_resource__platform") + .filter(id=run_id) + .first() + ) + if run is None: + return None + resource = run.learning_resource + platform_code = resource.platform.code if resource.platform else "" + + def chunk0_point_id(key): + # Mirrors the doc fields ContentFileSerializer emits for run files + return vector_point_id( + vector_point_key( + { + "platform": {"code": platform_code}, + "resource_readable_id": resource.readable_id, + "run_readable_id": run.run_id, + "key": key, + }, + chunk_number=0, + document_type="content_file", + ) + ) + + pid_rows = [ + (cf_id, chunk0_point_id(key), checksum, meta) + for cf_id, key, checksum, *meta in ContentFile.objects.filter( + run=run, published=True + ).values_list("id", "key", "checksum", *CONTENT_FILE_PREPASS_PAYLOAD_FIELDS) + ] + stored = _stored_content_payloads( + [pid for _, pid, _, _ in pid_rows], + fields=("checksum", *CONTENT_FILE_PREPASS_PAYLOAD_FIELDS), + ) + + def is_stale(pid, checksum, meta): + payload = stored.get(pid) + if payload is None or payload.get("checksum") != checksum: + return True + return any( + payload.get(field) != value + for field, value in zip(CONTENT_FILE_PREPASS_PAYLOAD_FIELDS, meta) + ) + + ids = [ + cf_id + for cf_id, pid, checksum, meta in pid_rows + if is_stale(pid, checksum, meta) + ] + log.info( + "embed_run_content_files run %s: %d of %d files need embedding", + run_id, + len(ids), + len(pid_rows), + ) if not ids: return None return _replace_with_finalized_chain(self, ids, overwrite=True) diff --git a/vector_search/tasks_test.py b/vector_search/tasks_test.py index 00e8ba332b..632d8c4b40 100644 --- a/vector_search/tasks_test.py +++ b/vector_search/tasks_test.py @@ -28,7 +28,9 @@ PROGRAM_TYPE, ) from learning_resources_search.exceptions import RetryError +from learning_resources_search.serializers import serialize_bulk_content_files from main.utils import now_in_utc +from vector_search.constants import CONTENT_FILE_PREPASS_PAYLOAD_FIELDS from vector_search.tasks import ( _record_embedding_failure, _retry_countdown, @@ -44,7 +46,7 @@ remove_unpublished_run_content_files, start_embed_resources, ) -from vector_search.utils import vector_point_id +from vector_search.utils import vector_point_id, vector_point_key pytestmark = pytest.mark.django_db @@ -767,45 +769,155 @@ def test_embed_run_content_files_no_files_returns_none(mocker, mocked_celery): mocked_celery.chain.assert_not_called() -def test_embed_run_content_files_only_given_ids(mocker, mocked_celery, settings): - """When content_file_ids is passed, only those published files are embedded.""" +def test_embed_run_content_files_skips_unpublished(mocker, mocked_celery, settings): + """Unpublished files are never embedded.""" settings.QDRANT_CHUNK_SIZE = 50 run = LearningResourceRunFactory.create() - files = ContentFileFactory.create_batch(3, run=run, published=True) - changed = [files[0].id, files[1].id] + published = ContentFileFactory.create(run=run, published=True) + ContentFileFactory.create(run=run, published=False) generate_embeddings_mock = mocker.patch( "vector_search.tasks.generate_embeddings", autospec=True ) with pytest.raises(mocked_celery.replace_exception_class): - embed_run_content_files.delay(run.id, changed) + embed_run_content_files.delay(run.id) - assert _embedded_content_file_ids(generate_embeddings_mock) == set(changed) + assert _embedded_content_file_ids(generate_embeddings_mock) == {published.id} -def test_embed_run_content_files_skips_unpublished(mocker, mocked_celery, settings): - """Unpublished files are never embedded, even if named in content_file_ids.""" +def _serializer_chunk0_pids(content_files): + """Chunk-0 point ids as the embed pipeline (serializer path) computes them""" + return { + doc["id"]: vector_point_id( + vector_point_key(doc, chunk_number=0, document_type="content_file") + ) + for doc in serialize_bulk_content_files([cf.id for cf in content_files]) + } + + +def _stored_payload_entry(content_file, **overrides): + """Build a stored-payload map entry matching the file's current DB state""" + return { + "checksum": content_file.checksum, + **{ + field: getattr(content_file, field) + for field in CONTENT_FILE_PREPASS_PAYLOAD_FIELDS + }, + **overrides, + } + + +def test_embed_run_content_files_pre_pass_skips_unchanged( + mocker, mocked_celery, settings +): + """ + Only files whose stored Qdrant payload is missing or stale are embedded. + + The stored-payload map is keyed by serializer-derived point ids, so the + unchanged file is skipped only if the task's pre-pass computes the same + point id as the embed pipeline. + """ settings.QDRANT_CHUNK_SIZE = 50 run = LearningResourceRunFactory.create() - published = ContentFileFactory.create(run=run, published=True) - unpublished = ContentFileFactory.create(run=run, published=False) + # ContentFile.save() computes checksum from content + unchanged = ContentFileFactory.create(run=run, published=True, content="aaa") + stale = ContentFileFactory.create(run=run, published=True, content="bbb") + missing = ContentFileFactory.create(run=run, published=True, content="ccc") + pids = _serializer_chunk0_pids([unchanged, stale, missing]) + stored_mock = mocker.patch( + "vector_search.tasks._stored_content_payloads", + return_value={ + pids[unchanged.id]: _stored_payload_entry(unchanged), + pids[stale.id]: _stored_payload_entry(stale, checksum="stale-checksum"), + }, + ) generate_embeddings_mock = mocker.patch( "vector_search.tasks.generate_embeddings", autospec=True ) with pytest.raises(mocked_celery.replace_exception_class): - embed_run_content_files.delay(run.id, [published.id, unpublished.id]) + embed_run_content_files.delay(run.id) - assert _embedded_content_file_ids(generate_embeddings_mock) == {published.id} + assert _embedded_content_file_ids(generate_embeddings_mock) == { + stale.id, + missing.id, + } + assert set(stored_mock.call_args.args[0]) == set(pids.values()) -def test_embed_run_content_files_empty_ids_returns_none(mocker, mocked_celery): - """An explicit empty id list embeds nothing (no changed files this load).""" +def test_embed_run_content_files_pre_pass_dispatches_metadata_only_change( + mocker, mocked_celery, settings +): + """ + A file with a matching checksum but drifted payload metadata (edited title, + newly generated summary) is dispatched so its Qdrant payload gets refreshed. + """ + settings.QDRANT_CHUNK_SIZE = 50 run = LearningResourceRunFactory.create() - ContentFileFactory.create(run=run, published=True) - mocker.patch("vector_search.tasks.generate_embeddings", autospec=True) - assert embed_run_content_files(run.id, []) is None - mocked_celery.group.assert_not_called() + retitled = ContentFileFactory.create(run=run, published=True, content="aaa") + summarized = ContentFileFactory.create(run=run, published=True, content="bbb") + current = ContentFileFactory.create(run=run, published=True, content="ccc") + pids = _serializer_chunk0_pids([retitled, summarized, current]) + mocker.patch( + "vector_search.tasks._stored_content_payloads", + return_value={ + pids[retitled.id]: _stored_payload_entry(retitled, title="old title"), + pids[summarized.id]: _stored_payload_entry(summarized, summary=""), + pids[current.id]: _stored_payload_entry(current), + }, + ) + summarized.summary = "a new summary" + summarized.save() + generate_embeddings_mock = mocker.patch( + "vector_search.tasks.generate_embeddings", autospec=True + ) + + with pytest.raises(mocked_celery.replace_exception_class): + embed_run_content_files.delay(run.id) + + assert _embedded_content_file_ids(generate_embeddings_mock) == { + retitled.id, + summarized.id, + } + + +def test_content_file_prepass_fields_are_serializer_pass_through(): + """ + Every pre-pass-compared field must be an exact serializer pass-through of + the ContentFile column: a transformed field would never converge with the + stored payload, flagging every file on every load. + """ + content_file = ContentFileFactory.create( + run=LearningResourceRunFactory.create(), + published=True, + content="some content", + summary="a summary", + flashcards=[{"question": "q", "answer": "a"}], + ) + doc = next(iter(serialize_bulk_content_files([content_file.id]))) + for field in ("checksum", *CONTENT_FILE_PREPASS_PAYLOAD_FIELDS): + assert doc[field] == getattr(content_file, field), field + + +def test_embed_run_content_files_all_unchanged_dispatches_nothing( + mocker, mocked_celery +): + """A fully-unchanged run embeds nothing and schedules no chain.""" + run = LearningResourceRunFactory.create() + files = ContentFileFactory.create_batch(2, run=run, published=True, content="x") + pids = _serializer_chunk0_pids(files) + mocker.patch( + "vector_search.tasks._stored_content_payloads", + return_value={pids[cf.id]: _stored_payload_entry(cf) for cf in files}, + ) + generate_embeddings_mock = mocker.patch( + "vector_search.tasks.generate_embeddings", autospec=True + ) + + assert embed_run_content_files(run.id) is None + + generate_embeddings_mock.si.assert_not_called() + mocked_celery.chain.assert_not_called() def test_embeddings_healthcheck_no_missing_embeddings(mocker): diff --git a/vector_search/utils.py b/vector_search/utils.py index 67aab4ee5f..0665a2dfc9 100644 --- a/vector_search/utils.py +++ b/vector_search/utils.py @@ -637,17 +637,29 @@ def _retrieve_content_file_point( return None -def _content_file_stored_checksum_changed(serialized_document: dict) -> bool: - point = _retrieve_content_file_point(serialized_document) - if not point: - return False - stored_checksum = (point.payload or {}).get("checksum") - # Missing checksums should not force an expensive summary rewrite by themselves. - # should_generate_content_embeddings still treats them as changed so embeddings - # can repair older Qdrant points without overwriting existing summaries. - return stored_checksum is not None and stored_checksum != serialized_document.get( - "checksum" - ) +def _stored_content_payloads( + point_ids: list[str], fields: tuple[str, ...] = ("checksum",) +) -> dict[str, dict]: + """ + Batch-retrieve stored payload fields for content-file points. + + Returns {point_id: partial payload dict} for points that exist in Qdrant; + absent points are absent from the map. One lookup per batch replaces the + per-file retrieves for the existence filter, summary-change check, and + embed gate. + """ + client = qdrant_client() + stored = {} + for id_batch in chunks( + point_ids, chunk_size=settings.QDRANT_POINT_UPLOAD_BATCH_SIZE + ): + for record in client.retrieve( + collection_name=CONTENT_FILES_COLLECTION_NAME, + ids=id_batch, + with_payload=list(fields), + ): + stored[record.id] = record.payload or {} + return stored def should_generate_content_embeddings( @@ -743,9 +755,13 @@ def _embed_course_metadata_as_contentfile(serialized_resources): client.upload_points(CONTENT_FILES_COLLECTION_NAME, points=points, wait=False) -def _generate_content_file_points(serialized_content): +def _generate_content_file_points(serialized_content, stored_payloads): """ - Chunk and embed content file documents, yielding PointStructs + Chunk and embed content file documents, yielding PointStructs. + + stored_payloads maps chunk-0 point ids to stored Qdrant payload fields + (see _stored_content_payloads); docs whose stored checksum matches get a + payload-only refresh instead of re-embedding. """ encoder_dense = dense_encoder() encoder_sparse = sparse_encoder() @@ -771,7 +787,17 @@ def _generate_content_file_points(serialized_content): embedding_context = _content_file_embedding_context(doc) if not embedding_context: continue - should_generate = should_generate_content_embeddings(doc) + # Point ids are content-key-derived and stable, so recompute per doc; + # summarization replaces the doc dicts between here and process_batch. + point_id = vector_point_id( + vector_point_key(doc, chunk_number=0, document_type="content_file") + ) + # Missing point or differing/missing stored checksum -> regenerate + # (self-heals failed or purged points on the next load). + should_generate = ( + point_id not in stored_payloads + or stored_payloads[point_id].get("checksum") != doc["checksum"] + ) if not should_generate: """ Just update the payload and continue @@ -947,7 +973,6 @@ def process_batch(docs_batch): fill_summary_content_ids = [] changed_summary_content_ids = [] - # Collect IDs for summarization contentfile_points = [ ( vector_point_id( @@ -959,17 +984,23 @@ def process_batch(docs_batch): ) for doc in docs_batch ] + # One batched lookup serves the existence filter, the summary-change + # check, and the embed gate in _generate_content_file_points. + stored_payloads = _stored_content_payloads( + [point[0] for point in contentfile_points] + ) if not overwrite: - filtered_point_ids = filter_existing_qdrant_points_by_ids( - [point[0] for point in contentfile_points], - collection_name=collection_name, - ) docs_batch = [ - point[1] + doc + for point_id, doc in contentfile_points + if point_id not in stored_payloads + ] + contentfile_points = [ + point for point in contentfile_points - if point[0] in filtered_point_ids + if point[0] not in stored_payloads ] - for resource in docs_batch: + for point_id, resource in contentfile_points: if ( resource.get("summary") or resource.get("require_summaries") @@ -977,7 +1008,15 @@ def process_batch(docs_batch): .filter(run__id=resource.get("run_id")) .exists() ): - if overwrite and _content_file_stored_checksum_changed(resource): + stored_checksum = stored_payloads.get(point_id, {}).get("checksum") + # A missing point or missing stored checksum must not force + # an expensive summary rewrite by itself; the embed gate + # still regenerates embeddings for those. + if ( + overwrite + and stored_checksum is not None + and stored_checksum != resource.get("checksum") + ): changed_summary_content_ids.append(resource["id"]) else: fill_summary_content_ids.append(resource["id"]) @@ -988,7 +1027,9 @@ def process_batch(docs_batch): changed_summary_content_ids, ) - points_generator_iter = _generate_content_file_points(docs_batch) + points_generator_iter = _generate_content_file_points( + docs_batch, stored_payloads + ) points_upload_batch = [] for point in points_generator_iter: diff --git a/vector_search/utils_test.py b/vector_search/utils_test.py index 28ed8b5226..af5a9cb346 100644 --- a/vector_search/utils_test.py +++ b/vector_search/utils_test.py @@ -226,16 +226,16 @@ def test_embed_learning_resources_no_overwrite(mocker, content_type): ], ) else: - # all contentfiles exist in qdrant - mocker.patch( - "vector_search.utils.filter_existing_qdrant_points_by_ids", - return_value=[ - vector_point_id( - f"{doc['platform']['code']}.{doc['resource_readable_id']}.{doc['run_readable_id']}.{doc['key']}.0" - ) - for doc in serialize_bulk_content_files([r.id for r in resources[0:3]]) - ], - ) + # the last 2 contentfiles already have points in qdrant; the first 3 don't + mock_qdrant.retrieve.return_value = [ + mocker.MagicMock( + id=vector_point_id( + vector_point_key(doc, chunk_number=0, document_type="content_file") + ), + payload={"checksum": doc["checksum"]}, + ) + for doc in serialize_bulk_content_files([r.id for r in resources[3:5]]) + ] mocker.patch( "learning_resources.content_summarizer.ContentSummarizer.summarize_content_files_by_ids" ) @@ -770,9 +770,6 @@ def test_generate_content_points_uses_markdown_chunking_for_marketing_pages(mock return_value=[Document(page_content="chunk1", metadata={"key": "k1"})], ) mock_chunk = mocker.patch("vector_search.utils._chunk_documents") - mocker.patch( - "vector_search.utils.should_generate_content_embeddings", return_value=True - ) mocker.patch("vector_search.utils.remove_points_matching_params") mock_dense = mocker.MagicMock() @@ -794,7 +791,7 @@ def test_generate_content_points_uses_markdown_chunking_for_marketing_pages(mock "key": "k1", } - list(_generate_content_file_points([doc])) + list(_generate_content_file_points([doc], {})) mock_md_chunk.assert_called_once() mock_chunk.assert_not_called() @@ -810,9 +807,6 @@ def test_generate_content_points_uses_standard_chunking_for_non_markdown(mocker) "vector_search.utils._chunk_documents", return_value=[Document(page_content="chunk1", metadata={"key": "k1"})], ) - mocker.patch( - "vector_search.utils.should_generate_content_embeddings", return_value=True - ) mocker.patch("vector_search.utils.remove_points_matching_params") mock_dense = mocker.MagicMock() @@ -834,7 +828,7 @@ def test_generate_content_points_uses_standard_chunking_for_non_markdown(mocker) "key": "k1", } - list(_generate_content_file_points([doc])) + list(_generate_content_file_points([doc], {})) mock_chunk.assert_called_once() mock_md_chunk.assert_not_called() @@ -857,9 +851,6 @@ def test_generate_content_points_leaves_headroom_under_token_limit(mocker): for i in range(num_chunks) ], ) - mocker.patch( - "vector_search.utils.should_generate_content_embeddings", return_value=True - ) mocker.patch("vector_search.utils.remove_points_matching_params") mock_dense = mocker.MagicMock() @@ -881,7 +872,7 @@ def test_generate_content_points_leaves_headroom_under_token_limit(mocker): "key": "k1", } - points = list(_generate_content_file_points([doc])) + points = list(_generate_content_file_points([doc], {})) batch_sizes = [ len(call.args[0]) for call in mock_dense.embed_documents.call_args_list @@ -906,9 +897,6 @@ def test_generate_content_points_request_chunk_size_never_zero(mocker): Document(page_content=f"chunk{i}", metadata={"key": "k1"}) for i in range(3) ], ) - mocker.patch( - "vector_search.utils.should_generate_content_embeddings", return_value=True - ) mocker.patch("vector_search.utils.remove_points_matching_params") mock_dense = mocker.MagicMock() @@ -930,7 +918,7 @@ def test_generate_content_points_request_chunk_size_never_zero(mocker): "key": "k1", } - points = list(_generate_content_file_points([doc])) + points = list(_generate_content_file_points([doc], {})) assert len(points) == 3 @@ -1127,43 +1115,81 @@ def test_should_generate_for_changed_content_file(mocker): assert result is True +def test_stored_content_payloads_batches_and_maps(mocker, settings): + """One retrieve per id-chunk; existing points map to their stored payload.""" + settings.QDRANT_POINT_UPLOAD_BATCH_SIZE = 2 + present = mocker.MagicMock(id="p1", payload={"checksum": "abc"}) + no_checksum = mocker.MagicMock(id="p2", payload={}) + mock_qdrant = mocker.MagicMock() + # p3 does not exist in Qdrant + mock_qdrant.retrieve.side_effect = [[present, no_checksum], []] + mocker.patch("vector_search.utils.qdrant_client", return_value=mock_qdrant) + + stored = vs_utils._stored_content_payloads( # noqa: SLF001 + ["p1", "p2", "p3"], fields=("checksum", "title") + ) + + assert stored == {"p1": {"checksum": "abc"}, "p2": {}} + assert mock_qdrant.retrieve.call_count == 2 # ceil(3 ids / batch size 2) + for call in mock_qdrant.retrieve.call_args_list: + assert call.kwargs["collection_name"] == CONTENT_FILES_COLLECTION_NAME + assert call.kwargs["with_payload"] == ["checksum", "title"] + + @pytest.mark.parametrize( - ("stored_payload", "expected"), + ("stored_entry", "expect_regenerate"), [ - ({"checksum": "previous-checksum"}, True), - ({"checksum": "current-checksum"}, False), - ({}, False), - (None, False), + ("missing", True), # no point in Qdrant (new file or failed prior embed) + (None, True), # point exists but has no stored checksum + ("stale-checksum", True), # stored checksum differs + ("current-checksum", False), # matches -> payload-only update ], -) -def test_content_file_stored_checksum_changed(mocker, stored_payload, expected): - """Only an existing, different stored checksum counts as changed for summaries.""" - serialized_document = { - "resource_readable_id": "resource-1", - "run_readable_id": "run-1", - "key": "transcript.txt", - "checksum": "current-checksum", - } - mock_qdrant = mocker.MagicMock() - if stored_payload is None: - mock_qdrant.retrieve.return_value = [] - else: - mock_point = mocker.MagicMock() - mock_point.payload = stored_payload - mock_qdrant.retrieve.return_value = [mock_point] - mocker.patch("vector_search.utils.qdrant_client", return_value=mock_qdrant) +) # stored_entry is the checksum in the stored payload dict +def test_generate_content_points_checksum_gate(mocker, stored_entry, expect_regenerate): + """Docs are re-embedded unless their stored Qdrant checksum matches.""" + settings.CONTENT_FILE_EMBEDDING_CHUNK_SIZE_OVERRIDE = 500 + settings.CONTENT_FILE_EMBEDDING_CHUNK_OVERLAP = 50 - assert ( - vs_utils._content_file_stored_checksum_changed( # noqa: SLF001 - serialized_document - ) - is expected + mocker.patch( + "vector_search.utils._chunk_documents", + return_value=[Document(page_content="chunk1", metadata={"key": "k1"})], ) - mock_qdrant.retrieve.assert_called_once() - assert ( - mock_qdrant.retrieve.call_args.kwargs["collection_name"] - == CONTENT_FILES_COLLECTION_NAME + mocker.patch("vector_search.utils.remove_points_matching_params") + update_payload_mock = mocker.patch( + "vector_search.utils.update_content_file_payload" + ) + mock_dense = mocker.MagicMock() + mock_dense.embed_documents.side_effect = lambda texts: [[0.1] for _ in texts] + mock_dense.model_short_name.return_value = "dense" + mock_sparse = mocker.MagicMock() + mock_sparse.embed_documents.side_effect = lambda texts: [[0.2] for _ in texts] + mock_sparse.model_short_name.return_value = "sparse" + mocker.patch("vector_search.utils.dense_encoder", return_value=mock_dense) + mocker.patch("vector_search.utils.sparse_encoder", return_value=mock_sparse) + + doc = { + "content": "Some plain text content", + "file_type": "page", + "file_extension": ".html", + "platform": {"code": "x"}, + "resource_readable_id": "r1", + "run_readable_id": "run1", + "key": "k1", + "checksum": "current-checksum", + } + point_id = vector_point_id( + vector_point_key(doc, chunk_number=0, document_type="content_file") ) + stored = {} if stored_entry == "missing" else {point_id: {"checksum": stored_entry}} + + points = list(_generate_content_file_points([doc], stored)) + + if expect_regenerate: + assert len(points) == 1 + update_payload_mock.assert_not_called() + else: + assert points == [] + update_payload_mock.assert_called_once_with(doc) def test_should_not_generate_for_unchanged_content_file(mocker): @@ -1246,11 +1272,9 @@ def test_embed_learning_resources_summarizes_only_contentfiles_with_summary(mock Test that embedding overwrites don't overwrite existing summaries. """ mock_qdrant = mocker.patch("qdrant_client.QdrantClient") + mock_qdrant.retrieve.return_value = [] mocker.patch("vector_search.utils.qdrant_client", return_value=mock_qdrant) mocker.patch("vector_search.utils.create_qdrant_collections") - mocker.patch( - "vector_search.utils.filter_existing_qdrant_points_by_ids", return_value=[] - ) mocker.patch("vector_search.utils.remove_qdrant_records") learning_resource = LearningResourceFactory.create( @@ -1285,9 +1309,6 @@ def test_embed_learning_resources_summarizes_only_contentfiles_with_summary(mock mocker.patch( "vector_search.utils.serialize_bulk_content_files", return_value=serialized ) - mocker.patch( - "vector_search.utils._content_file_stored_checksum_changed", return_value=False - ) summarize_mock = mocker.patch( "learning_resources.content_summarizer.ContentSummarizer.summarize_content_files_by_ids" @@ -1333,7 +1354,7 @@ def test_embed_learning_resources_overwrites_summaries_for_changed_content(mocke "key": cf.key, "summary": cf.summary, "content": cf.content, - "checksum": cf.checksum, + "checksum": f"current-{cf.id}", } for cf in all_contentfiles ] @@ -1341,10 +1362,22 @@ def test_embed_learning_resources_overwrites_summaries_for_changed_content(mocke mocker.patch( "vector_search.utils.serialize_bulk_content_files", return_value=serialized ) - mocker.patch( - "vector_search.utils._content_file_stored_checksum_changed", - side_effect=lambda resource: resource["id"] == changed_content_file.id, - ) + # The unchanged file takes the payload-only path (covered by its own tests) + mocker.patch("vector_search.utils.update_content_file_payload") + # Stored Qdrant checksum matches for the unchanged file, differs for the changed + mock_qdrant.retrieve.return_value = [ + mocker.MagicMock( + id=vector_point_id( + vector_point_key(doc, chunk_number=0, document_type="content_file") + ), + payload={ + "checksum": doc["checksum"] + if doc["id"] == unchanged_content_file.id + else "stale-checksum" + }, + ) + for doc in serialized + ] summarize_mock = mocker.patch( "learning_resources.content_summarizer.ContentSummarizer.summarize_content_files_by_ids" @@ -1392,15 +1425,23 @@ def test_embed_learning_resources_keeps_old_checksum_when_summary_fails(mocker): "key": content_file.key, "summary": content_file.summary, "content": content_file.content, - "checksum": content_file.checksum, + "checksum": "current-checksum", } ] mocker.patch( "vector_search.utils.serialize_bulk_content_files", return_value=serialized ) - mocker.patch( - "vector_search.utils._content_file_stored_checksum_changed", return_value=True - ) + # Stored Qdrant checksum differs, so the summary must be regenerated + mock_qdrant.retrieve.return_value = [ + mocker.MagicMock( + id=vector_point_id( + vector_point_key( + serialized[0], chunk_number=0, document_type="content_file" + ) + ), + payload={"checksum": "previous-checksum"}, + ) + ] summarize_mock = mocker.patch( "learning_resources.content_summarizer.ContentSummarizer.summarize_content_files_by_ids", return_value=[ From 9ce4a8e6bf946969e110f8d50e6719e6b5df84d3 Mon Sep 17 00:00:00 2001 From: Matt Bertrand Date: Wed, 22 Jul 2026 08:36:21 -0400 Subject: [PATCH 09/10] Derive removed_unpublished from update() row count Drops the redundant exists() query; the row count also reflects what was actually unpublished rather than what existed a moment earlier. Co-Authored-By: Claude Opus 4.8 (1M context) --- learning_resources/etl/loaders.py | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/learning_resources/etl/loaders.py b/learning_resources/etl/loaders.py index 5c40f23787..7812516da0 100644 --- a/learning_resources/etl/loaders.py +++ b/learning_resources/etl/loaders.py @@ -1052,8 +1052,7 @@ def load_content_files( .values_list("direct_learning_resource_id", flat=True) .distinct() ) - removed_unpublished = stale_published_files.exists() - stale_published_files.update(published=False) + removed_unpublished = stale_published_files.update(published=False) > 0 if stale_direct_resource_ids: LearningResource.objects.filter( id__in=stale_direct_resource_ids, published=True From 0eaef52b18fce8f5ce3d0ac91adf6b304d8f66c1 Mon Sep 17 00:00:00 2001 From: Matt Bertrand Date: Thu, 23 Jul 2026 14:48:11 -0400 Subject: [PATCH 10/10] Always purge unpublished content files from Qdrant Gating the removal task on this run's unpublish count meant a failed removal (Qdrant outage, failed embed task earlier in the chain) was never retried: the next load sees zero newly-unpublished rows and skips the purge, leaving orphaned points in Qdrant indefinitely. Restore the unconditional purge so removals self-heal, and drop the now-dead removed_unpublished plumbing. Co-Authored-By: Claude Fable 5 --- learning_resources/etl/loaders.py | 7 ++---- learning_resources/etl/loaders_test.py | 26 ----------------------- learning_resources/hooks.py | 3 +-- learning_resources/utils.py | 9 ++------ learning_resources_search/plugins.py | 15 ++++++------- learning_resources_search/plugins_test.py | 22 ++++++------------- 6 files changed, 18 insertions(+), 64 deletions(-) diff --git a/learning_resources/etl/loaders.py b/learning_resources/etl/loaders.py index 7812516da0..62389d29ba 100644 --- a/learning_resources/etl/loaders.py +++ b/learning_resources/etl/loaders.py @@ -1052,7 +1052,7 @@ def load_content_files( .values_list("direct_learning_resource_id", flat=True) .distinct() ) - removed_unpublished = stale_published_files.update(published=False) > 0 + stale_published_files.update(published=False) if stale_direct_resource_ids: LearningResource.objects.filter( id__in=stale_direct_resource_ids, published=True @@ -1064,10 +1064,7 @@ def load_content_files( if calc_completeness: calculate_completeness(course_run, content_tags=content_tags) - content_files_loaded_actions( - run=course_run, - removed_unpublished=removed_unpublished, - ) + content_files_loaded_actions(run=course_run) return content_files_ids return None diff --git a/learning_resources/etl/loaders_test.py b/learning_resources/etl/loaders_test.py index 13832e664f..9927387278 100644 --- a/learning_resources/etl/loaders_test.py +++ b/learning_resources/etl/loaders_test.py @@ -3806,29 +3806,3 @@ def test_load_learning_material(mocker, learning_material_exists): assert learning_material.url == content_file.url assert content_file.direct_learning_resource_id == learning_material.id - - -@pytest.mark.django_db -def test_load_content_files_flags_removed_unpublished(mocker): - """ - The hook receives removed_unpublished=False when every prior file is still - present, and True when a previously-published file is dropped from the payload. - """ - course = LearningResourceFactory.create(is_course=True, create_runs=False) - run = LearningResourceRunFactory.create(published=True, learning_resource=course) - mock_hook = mocker.patch( - "learning_resources.etl.loaders.content_files_loaded_actions", autospec=True - ) - - load_content_files( - run, - [ - {"key": "keep", "content": "same"}, - {"key": "drop", "content": "gone"}, - ], - ) - assert mock_hook.call_args.kwargs["removed_unpublished"] is False - - # Second load: "drop" is absent, so its file gets unpublished. - load_content_files(run, [{"key": "keep", "content": "same"}]) - assert mock_hook.call_args.kwargs["removed_unpublished"] is True diff --git a/learning_resources/hooks.py b/learning_resources/hooks.py index becb9c8534..e174bcf89b 100644 --- a/learning_resources/hooks.py +++ b/learning_resources/hooks.py @@ -81,13 +81,12 @@ def offeror_delete(self, offeror): """Trigger actions to delete a learning resource offeror""" @hookspec - def content_files_loaded(self, run, removed_unpublished=None): + def content_files_loaded(self, run): """ Trigger actions after content files are loaded for a run. Args: run: the LearningResourceRun whose content files were loaded - removed_unpublished: whether files were unpublished, or None if unknown """ diff --git a/learning_resources/utils.py b/learning_resources/utils.py index d082dcd077..3f0cd4752c 100644 --- a/learning_resources/utils.py +++ b/learning_resources/utils.py @@ -435,21 +435,16 @@ def bulk_resources_unpublished_actions(resource_ids: list[int], resource_type: s ) -def content_files_loaded_actions( - run: LearningResourceRun, - *, - removed_unpublished: bool | None = None, -): +def content_files_loaded_actions(run: LearningResourceRun): """ Trigger plugins when content files are loaded for a LearningResourceRun. Args: run: the LearningResourceRun whose content files were loaded - removed_unpublished: whether files were unpublished, or None if unknown """ pm = get_plugin_manager() hook = pm.hook - hook.content_files_loaded(run=run, removed_unpublished=removed_unpublished) + hook.content_files_loaded(run=run) def resource_run_unpublished_actions(run: LearningResourceRun): diff --git a/learning_resources_search/plugins.py b/learning_resources_search/plugins.py index 87100cc986..2a9ef9a223 100644 --- a/learning_resources_search/plugins.py +++ b/learning_resources_search/plugins.py @@ -257,7 +257,7 @@ def resource_run_delete(self, run): run.delete() @hookimpl - def content_files_loaded(self, run, removed_unpublished=None): + def content_files_loaded(self, run): """ Upsert a created/modified run's content files. @@ -268,8 +268,6 @@ def content_files_loaded(self, run, removed_unpublished=None): Args: run: the LearningResourceRun that was upserted - removed_unpublished: whether files were unpublished; only an explicit - False skips the removal task (None still purges) """ if not run.content_files.exists(): return @@ -287,12 +285,11 @@ def content_files_loaded(self, run, removed_unpublished=None): if django_settings.QDRANT_ENABLE_INDEXING_PLUGIN_HOOKS: index_tasks.append(vector_tasks.embed_run_content_files.si(run.id)) - # None (legacy/unknown) keeps the historical always-purge behavior; - # only load_content_files, which knows definitively, passes False. - if removed_unpublished is not False: - index_tasks.append( - vector_tasks.remove_unpublished_run_content_files.si(run.id) - ) + # Always purge unpublished files' points so a failed removal + # task self-heals on the next load. + index_tasks.append( + vector_tasks.remove_unpublished_run_content_files.si(run.id) + ) if index_tasks: try_with_retry_as_task(chain(*index_tasks)) diff --git a/learning_resources_search/plugins_test.py b/learning_resources_search/plugins_test.py index 864862292f..e6b5f185b7 100644 --- a/learning_resources_search/plugins_test.py +++ b/learning_resources_search/plugins_test.py @@ -571,26 +571,18 @@ def test_search_index_plugin_resource_upserted_generate_embeddings( @pytest.mark.django_db -@pytest.mark.parametrize( - ("removed_unpublished", "expect_remove_called"), - [(True, True), (None, True), (False, False)], -) -def test_content_files_loaded_removed_unpublished_tristate( - mock_search_index_helpers, settings, removed_unpublished, expect_remove_called +def test_content_files_loaded_always_purges_unpublished( + mock_search_index_helpers, settings ): - """None (legacy) and True append the remove-unpublished task; only False skips.""" + """The remove-unpublished task always runs so failed removals self-heal.""" settings.QDRANT_ENABLE_INDEXING_PLUGIN_HOOKS = True run = LearningResourceRunFactory.create( published=True, learning_resource__create_runs=False ) ContentFileFactory.create(run=run) - SearchIndexPlugin().content_files_loaded( - run, removed_unpublished=removed_unpublished - ) + SearchIndexPlugin().content_files_loaded(run) - remove_mock = mock_search_index_helpers.mock_remove_unpublished_run_contentfiles_immutable_signature - if expect_remove_called: - remove_mock.assert_called_once_with(run.id) - else: - remove_mock.assert_not_called() + mock_search_index_helpers.mock_remove_unpublished_run_contentfiles_immutable_signature.assert_called_once_with( + run.id + )