Followup to #3646: skip contentless files in embed pre-pass, retry Qdrant blips, purge before embed

learning_resources_search/plugins.py

@@ -284,12 +284,13 @@ 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))
-                # Always purge unpublished files' points so a failed removal
-                # task self-heals on the next load.
+                # Purge before embedding so unpublished files' points are
+                # removed even if the embed task fails; the two tasks touch
+                # disjoint sets (published=False vs published=True files).
                 index_tasks.append(
                     vector_tasks.remove_unpublished_run_content_files.si(run.id)
                 )
+                index_tasks.append(vector_tasks.embed_run_content_files.si(run.id))
 
         if index_tasks:
             try_with_retry_as_task(chain(*index_tasks))

learning_resources_search/plugins_test.py

@@ -572,10 +572,14 @@ def test_search_index_plugin_resource_upserted_generate_embeddings(
 
 @pytest.mark.django_db
 def test_content_files_loaded_always_purges_unpublished(
-    mock_search_index_helpers, settings
+    mocker, mock_search_index_helpers, settings
 ):
-    """The remove-unpublished task always runs so failed removals self-heal."""
+    """
+    The remove-unpublished task always runs, and ahead of the embed task in the
+    chain, so a failed embed can't strand unpublished files' points in Qdrant.
+    """
     settings.QDRANT_ENABLE_INDEXING_PLUGIN_HOOKS = True
+    chain_mock = mocker.patch("learning_resources_search.plugins.chain")
     run = LearningResourceRunFactory.create(
         published=True, learning_resource__create_runs=False
     )
@@ -586,3 +590,7 @@ def test_content_files_loaded_always_purges_unpublished(
     mock_search_index_helpers.mock_remove_unpublished_run_contentfiles_immutable_signature.assert_called_once_with(
         run.id
     )
+    chained = list(chain_mock.call_args.args)
+    purge = mock_search_index_helpers.mock_remove_unpublished_run_contentfiles_immutable_signature.return_value
+    embed = mock_search_index_helpers.mock_embed_run_contentfiles_immutable_signature.return_value
+    assert chained.index(purge) < chained.index(embed)

vector_search/tasks.py

@@ -465,7 +465,7 @@ def embed_new_content_files(self):
     )
 
 
-@app.task(bind=True)
+@app.task(bind=True, max_retries=3)
 def embed_run_content_files(self, run_id):
     """
     Embed the run's published content files whose Qdrant points are missing or
@@ -478,6 +478,12 @@ def embed_run_content_files(self, run_id):
     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-less files are excluded: they never produce Qdrant points, so they
+    would otherwise be re-flagged on every load. Any point left over from when
+    such a file still had content is removed. Transient Qdrant errors during
+    the pre-pass retry with backoff so a blip doesn't defer the run's embedding
+    to the next load.
     """
     run = (
         LearningResourceRun.objects.select_related("learning_resource__platform")
@@ -489,8 +495,9 @@ def embed_run_content_files(self, run_id):
     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
+    def first_chunk_point_id(key):
+        # Returns the qdrant point id for the first chunk of the contentfile,
+        # mirroring the doc fields ContentFileSerializer emits for run files
         return vector_point_id(
             vector_point_key(
                 {
@@ -504,16 +511,33 @@ def chunk0_point_id(key):
             )
         )
 
+    contentless = Q(content__isnull=True) | Q(content="")
     pid_rows = [
-        (cf_id, chunk0_point_id(key), checksum, meta)
+        (cf_id, first_chunk_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)
+        )
+        .exclude(contentless)
+        .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),
-    )
+    contentless_rows = [
+        (cf_id, first_chunk_point_id(key))
+        for cf_id, key in ContentFile.objects.filter(
+            contentless, run=run, published=True
+        ).values_list("id", "key")
+    ]
+    try:
+        stored = _stored_content_payloads(
+            [pid for _, pid, _, _ in pid_rows] + [pid for _, pid in contentless_rows],
+            fields=("checksum", *CONTENT_FILE_PREPASS_PAYLOAD_FIELDS),
+        )
+    except grpc.RpcError as err:
+        if err.code() in (
+            grpc.StatusCode.DEADLINE_EXCEEDED,
+            grpc.StatusCode.UNAVAILABLE,
+        ):
+            raise self.retry(exc=err, countdown=_retry_countdown(self.request.retries))  # noqa: B904
+        raise
 
     def is_stale(pid, checksum, meta):
         payload = stored.get(pid)
@@ -529,12 +553,20 @@ def is_stale(pid, checksum, meta):
         for cf_id, pid, checksum, meta in pid_rows
         if is_stale(pid, checksum, meta)
     ]
+    # A stored point for a now-contentless file is a leftover from when the
+    # file had content — remove it. Inline rather than a chained task: leftovers
+    # are rare and few, and a failed delete self-heals on the next load.
+    leftover_ids = [cf_id for cf_id, pid in contentless_rows if pid in stored]
     log.info(
-        "embed_run_content_files run %s: %d of %d files need embedding",
+        "embed_run_content_files run %s: %d of %d files need embedding, "
+        "%d leftover contentless points to remove",
         run_id,
         len(ids),
         len(pid_rows),
+        len(leftover_ids),
     )
+    if leftover_ids:
+        remove_qdrant_records(leftover_ids, CONTENT_FILE_TYPE)
     if not ids:
         return None
     return _replace_with_finalized_chain(self, ids, overwrite=True)

vector_search/tasks_test.py

@@ -705,7 +705,8 @@ def test_embed_run_content_files(mocker, mocked_celery, settings):
     settings.QDRANT_CHUNK_SIZE = 2
     run = LearningResourceRunFactory.create()
     content_file_ids = [
-        content_file.id for content_file in ContentFileFactory.create_batch(3, run=run)
+        content_file.id
+        for content_file in ContentFileFactory.create_batch(3, run=run, content="text")
     ]
     ContentFileFactory.create()
     generate_embeddings_mock = mocker.patch(
@@ -773,8 +774,8 @@ def test_embed_run_content_files_skips_unpublished(mocker, mocked_celery, settin
     """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)
+    published = ContentFileFactory.create(run=run, published=True, content="aaa")
+    ContentFileFactory.create(run=run, published=False, content="bbb")
     generate_embeddings_mock = mocker.patch(
         "vector_search.tasks.generate_embeddings", autospec=True
     )
@@ -785,6 +786,83 @@ def test_embed_run_content_files_skips_unpublished(mocker, mocked_celery, settin
     assert _embedded_content_file_ids(generate_embeddings_mock) == {published.id}
 
 
+def test_embed_run_content_files_skips_contentless(mocker, mocked_celery, settings):
+    """
+    Files without content never produce Qdrant points, so the pre-pass must not
+    flag them as stale (they would otherwise be re-dispatched on every load).
+    """
+    settings.QDRANT_CHUNK_SIZE = 50
+    run = LearningResourceRunFactory.create()
+    with_content = ContentFileFactory.create(run=run, published=True, content="aaa")
+    ContentFileFactory.create(run=run, published=True, content="")
+    ContentFileFactory.create(run=run, published=True, content=None)
+    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) == {with_content.id}
+
+
+def test_embed_run_content_files_removes_leftover_contentless_points(mocker):
+    """
+    A published file whose content became empty keeps the point embedded from
+    its old content; the pre-pass detects and removes it. Contentless files
+    with no stored point trigger no removal.
+    """
+    run = LearningResourceRunFactory.create()
+    emptied = ContentFileFactory.create(run=run, published=True, content="")
+    ContentFileFactory.create(run=run, published=True, content=None)
+    pids = _serializer_chunk0_pids([emptied])
+    mocker.patch(
+        "vector_search.tasks._stored_content_payloads",
+        return_value={pids[emptied.id]: {"checksum": "from-old-content"}},
+    )
+    generate_embeddings_mock = mocker.patch(
+        "vector_search.tasks.generate_embeddings", autospec=True
+    )
+    remove_mock = mocker.patch(
+        "vector_search.tasks.remove_qdrant_records", autospec=True
+    )
+
+    assert embed_run_content_files(run.id) is None
+
+    generate_embeddings_mock.si.assert_not_called()
+    remove_mock.assert_called_once_with([emptied.id], CONTENT_FILE_TYPE)
+
+
+def test_embed_run_content_files_retries_transient_qdrant_errors(mocker):
+    """A transient Qdrant error in the pre-pass retries instead of failing the run."""
+    run = LearningResourceRunFactory.create()
+    ContentFileFactory.create(run=run, published=True, content="aaa")
+    mocker.patch(
+        "vector_search.tasks._stored_content_payloads",
+        side_effect=_rpc_error(grpc.StatusCode.UNAVAILABLE),
+    )
+    retry = mocker.patch.object(embed_run_content_files, "retry", side_effect=Retry())
+
+    with pytest.raises(Retry):
+        embed_run_content_files(run.id)
+
+    retry.assert_called_once()
+    assert retry.call_args.kwargs["countdown"] >= 0
+
+
+def test_embed_run_content_files_does_not_retry_terminal_errors(mocker):
+    """A non-transient Qdrant error in the pre-pass propagates without retry."""
+    run = LearningResourceRunFactory.create()
+    ContentFileFactory.create(run=run, published=True, content="aaa")
+    mocker.patch(
+        "vector_search.tasks._stored_content_payloads",
+        side_effect=_rpc_error(grpc.StatusCode.INVALID_ARGUMENT),
+    )
+
+    with pytest.raises(grpc.RpcError):
+        embed_run_content_files(run.id)
+
+
 def _serializer_chunk0_pids(content_files):
     """Chunk-0 point ids as the embed pipeline (serializer path) computes them"""
     return {