Skip embedding tasks and Qdrant writes for unchanged content files

learning_resources/hooks.py

@@ -82,7 +82,12 @@ def offeror_delete(self, offeror):
 
     @hookspec
     def content_files_loaded(self, run):
-        """Trigger actions after content files are loaded for a run"""
+        """
+        Trigger actions after content files are loaded for a run.
+
+        Args:
+            run: the LearningResourceRun whose content files were loaded
+        """
 
 
 def get_plugin_manager():

learning_resources/utils.py

@@ -437,7 +437,10 @@ def bulk_resources_unpublished_actions(resource_ids: list[int], resource_type: s
 
 def content_files_loaded_actions(run: LearningResourceRun):
     """
-    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
     """
     pm = get_plugin_manager()
     hook = pm.hook

learning_resources_search/plugins.py

@@ -261,12 +261,13 @@ def content_files_loaded(self, run):
         """
         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 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(LearningResourceRun): The LearningResourceRun that was upserted
+        Args:
+            run: the LearningResourceRun that was upserted
         """
         if not run.content_files.exists():
             return
@@ -284,6 +285,8 @@ def content_files_loaded(self, run):
 
             if django_settings.QDRANT_ENABLE_INDEXING_PLUGIN_HOOKS:
                 index_tasks.append(vector_tasks.embed_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)
                 )

learning_resources_search/plugins_test.py

@@ -568,3 +568,21 @@ 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
+def test_content_files_loaded_always_purges_unpublished(
+    mock_search_index_helpers, settings
+):
+    """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)
+
+    mock_search_index_helpers.mock_remove_unpublished_run_contentfiles_immutable_signature.assert_called_once_with(
+        run.id
+    )

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",

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,
@@ -465,13 +468,76 @@ def embed_new_content_files(self):
 @app.task(bind=True)
 def embed_run_content_files(self, run_id):
     """
-    Embed contentfiles associated with a run
+    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_file_ids = list(
-        ContentFile.objects.filter(run__id=run_id).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),
     )
 
-    return _replace_with_finalized_chain(self, content_file_ids, overwrite=True)
+    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)
 
 
 @app.task(bind=True)

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,6 +769,157 @@ 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_skips_unpublished(mocker, mocked_celery, settings):
+    """Unpublished files are never embedded."""
+    settings.QDRANT_CHUNK_SIZE = 50
+    run = LearningResourceRunFactory.create()
+    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)
+
+    assert _embedded_content_file_ids(generate_embeddings_mock) == {published.id}
+
+
+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()
+    # 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)
+
+    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_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()
+    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):
     """
     Test embeddings_healthcheck when there are no missing embeddings