Release: fix stale search results after publishing

api/signals/collaborative_signals.py

@@ -1,16 +1,14 @@
 from typing import Any
 
 import structlog
-from django.core.cache import cache
 from django.db.models.signals import pre_save
 from django.dispatch import receiver
 
 from api.models.Collaborative import Collaborative
 from api.utils.enums import CollaborativeStatus
+from api.utils.search_cache import invalidate_search_cache
 from search.documents import CollaborativeDocument
 
-from .dataset_signals import SEARCH_CACHE_VERSION_KEY
-
 logger = structlog.get_logger(__name__)
 
 
@@ -37,30 +35,30 @@ def handle_collaborative_publication(sender: Any, instance: Collaborative, **kwa
             and instance.status == CollaborativeStatus.PUBLISHED
         )
 
-        if status_changing_to_published or status_changing_from_published:
-            version = cache.get(SEARCH_CACHE_VERSION_KEY, 0)
-            cache.set(SEARCH_CACHE_VERSION_KEY, version + 1)
-            logger.info("Invalidated search cache for collaborative", collaborative_id=instance.id)
-
         if status_changing_from_published:
             document = CollaborativeDocument.get(id=instance.id, ignore=404)
             if document:
-                document.delete()
+                document.delete(refresh=True)
                 logger.info(
                     "Removed collaborative from Elasticsearch index",
                     collaborative_id=instance.id,
                 )
         elif status_changing_to_published or remains_published:
             document = CollaborativeDocument.get(id=instance.id, ignore=404)
             if document:
-                document.update(instance)
+                document.update(instance, refresh=True)
             else:
-                CollaborativeDocument().update(instance)
+                CollaborativeDocument().update(instance, refresh=True)
             logger.info(
                 "Synced collaborative to Elasticsearch index",
                 collaborative_id=instance.id,
             )
 
+        # After the index write, never before it: see invalidate_search_cache.
+        if status_changing_to_published or status_changing_from_published:
+            invalidate_search_cache()
+            logger.info("Invalidated search cache for collaborative", collaborative_id=instance.id)
+
     except Exception as exc:  # pragma: no cover - logging only
         logger.error(
             "Error in collaborative publication signal handler",

api/signals/dataset_signals.py

@@ -2,19 +2,16 @@
 
 import structlog
 from django.conf import settings
-from django.core.cache import cache
 from django.db.models.signals import pre_save
 from django.dispatch import receiver
 
 from api.managers.dvc_manager import DVCManager
 from api.models.Dataset import Dataset
 from api.models.Resource import Resource, ResourceVersion, _increment_version
 from api.utils.enums import DatasetStatus
+from api.utils.search_cache import invalidate_search_cache
 from search.documents.dataset_document import DatasetDocument
 
-# Cache version key for search results
-SEARCH_CACHE_VERSION_KEY = "search_results_version"
-
 logger = structlog.getLogger(__name__)
 
 
@@ -43,19 +40,11 @@ def handle_dataset_publication(sender: Any, instance: Dataset, **kwargs: Any) ->
                 and instance.status != DatasetStatus.PUBLISHED
             )
 
-            # Invalidate search results cache by incrementing version
-            try:
-                version = cache.get(SEARCH_CACHE_VERSION_KEY, 0)
-                cache.set(SEARCH_CACHE_VERSION_KEY, version + 1)
-                logger.info(f"Invalidated search cache for dataset {instance.title}")
-            except Exception as e:
-                logger.error(f"Failed to invalidate search cache: {str(e)}")
-
             # Update Elasticsearch index
             if status_changing_from_published:
                 try:
                     document = DatasetDocument.get(id=instance.id, ignore=404)
-                    document.delete()
+                    document.delete(refresh=True)
                 except Exception as e:
                     logger.error(
                         f"Failed to delete Elasticsearch document for dataset {instance.title}: {str(e)}"
@@ -64,14 +53,21 @@ def handle_dataset_publication(sender: Any, instance: Dataset, **kwargs: Any) ->
                 try:
                     document = DatasetDocument.get(id=instance.id, ignore=404)
                     if document:
-                        document.update(instance)
+                        document.update(instance, refresh=True)
                     else:
-                        DatasetDocument().update(instance)
+                        DatasetDocument().update(instance, refresh=True)
                 except Exception as e:
                     logger.error(
                         f"Failed to update Elasticsearch document for dataset {instance.title}: {str(e)}"
                     )
 
+            # After the index write, never before it: see invalidate_search_cache.
+            try:
+                invalidate_search_cache()
+                logger.info(f"Invalidated search cache for dataset {instance.title}")
+            except Exception as e:
+                logger.error(f"Failed to invalidate search cache: {str(e)}")
+
             # Handle resource version increments for publication
             if status_changing_to_published:
                 logger.info(

api/signals/usecase_signals.py

@@ -1,17 +1,14 @@
 from typing import Any
 
 import structlog
-from django.core.cache import cache
 from django.db.models.signals import pre_save
 from django.dispatch import receiver
 
 from api.models.UseCase import UseCase
 from api.utils.enums import UseCaseStatus
+from api.utils.search_cache import invalidate_search_cache
 from search.documents.usecase_document import UseCaseDocument
 
-# Cache version key for search results
-SEARCH_CACHE_VERSION_KEY = "search_results_version"
-
 logger = structlog.getLogger(__name__)
 
 
@@ -42,23 +39,13 @@ def handle_usecase_publication(sender: Any, instance: UseCase, **kwargs: Any) ->
 
             # Only proceed if status is actually changing
             if status_changing_to_published or status_changing_from_published:
-                # Invalidate search results cache by incrementing version
-                try:
-                    version = cache.get(SEARCH_CACHE_VERSION_KEY, 0)
-                    cache.set(SEARCH_CACHE_VERSION_KEY, version + 1)
-                    logger.info(
-                        f"Invalidated search cache for usecase {instance.title}"
-                    )
-                except Exception as e:
-                    logger.error(f"Failed to invalidate search cache: {str(e)}")
-
                 # Update Elasticsearch index
                 if status_changing_from_published:
                     # Remove from index when unpublished
                     try:
                         document = UseCaseDocument.get(id=instance.id, ignore=404)
                         if document:
-                            document.delete()
+                            document.delete(refresh=True)
                             logger.info(
                                 f"Removed usecase {instance.title} from Elasticsearch index"
                             )
@@ -70,7 +57,7 @@ def handle_usecase_publication(sender: Any, instance: UseCase, **kwargs: Any) ->
                     # Add to index when published
                     try:
                         document = UseCaseDocument()
-                        document.update(instance)
+                        document.update(instance, refresh=True)
                         logger.info(
                             f"Added usecase {instance.title} to Elasticsearch index"
                         )
@@ -79,6 +66,13 @@ def handle_usecase_publication(sender: Any, instance: UseCase, **kwargs: Any) ->
                             f"Failed to add Elasticsearch document for usecase {instance.title}: {str(e)}"
                         )
 
+                # After the index write, never before it: see invalidate_search_cache.
+                try:
+                    invalidate_search_cache()
+                    logger.info(f"Invalidated search cache for usecase {instance.title}")
+                except Exception as e:
+                    logger.error(f"Failed to invalidate search cache: {str(e)}")
+
     except Exception as e:
         logger.error(f"Error in usecase publication signal handler: {str(e)}")
         # Don't raise the exception to avoid blocking the save operation

api/utils/search_cache.py

@@ -0,0 +1,19 @@
+"""Versioned invalidation for cached search results."""
+
+from django.core.cache import cache
+
+SEARCH_CACHE_VERSION_KEY = "search_results_version"
+
+
+def invalidate_search_cache() -> None:
+    """Move cached search results on to a new version.
+
+    Call only after the Elasticsearch write has been refreshed. Search totals are
+    read from the index, so bumping any earlier lets a request in the gap cache
+    the old results under the new version for the full result TTL.
+
+    The key never expires. Under the cache's default 300s timeout it fell back
+    to 0, making results cached under an earlier version 0 reachable again.
+    """
+    version = cache.get(SEARCH_CACHE_VERSION_KEY, 0)
+    cache.set(SEARCH_CACHE_VERSION_KEY, version + 1, timeout=None)

api/views/paginated_elastic_view.py

@@ -10,7 +10,7 @@
 from rest_framework.serializers import Serializer
 from rest_framework.views import APIView
 
-from api.signals.dataset_signals import SEARCH_CACHE_VERSION_KEY
+from api.utils.search_cache import SEARCH_CACHE_VERSION_KEY
 
 def as_plain_data(value: Any) -> Any:
     """Convert Elasticsearch wrapper objects into plain Python containers.

api/views/search_unified.py

@@ -14,7 +14,7 @@
 from api.models import Dataset, Geography, Metadata, UseCase
 from api.models.AIModel import AIModel
 from api.models.Collaborative import Collaborative
-from api.signals.dataset_signals import SEARCH_CACHE_VERSION_KEY
+from api.utils.search_cache import SEARCH_CACHE_VERSION_KEY
 from api.utils.telemetry_utils import trace_method
 from DataSpace import settings
 from search.documents import (

tests/test_search_cache_invalidation.py

@@ -0,0 +1,59 @@
+"""The search cache must be invalidated after the index write, with a key that never expires."""
+
+from types import SimpleNamespace
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from api.signals import collaborative_signals, dataset_signals, usecase_signals
+from api.utils import search_cache
+from api.utils.enums import CollaborativeStatus, DatasetStatus, UseCaseStatus
+
+
+def test_version_key_never_expires() -> None:
+    with patch.object(search_cache, "cache") as cache:
+        cache.get.return_value = 4
+        search_cache.invalidate_search_cache()
+    cache.set.assert_called_once_with(search_cache.SEARCH_CACHE_VERSION_KEY, 5, timeout=None)
+
+
+CASES = [
+    (dataset_signals, "handle_dataset_publication", "Dataset", "DatasetDocument", DatasetStatus),
+    (usecase_signals, "handle_usecase_publication", "UseCase", "UseCaseDocument", UseCaseStatus),
+    (
+        collaborative_signals,
+        "handle_collaborative_publication",
+        "Collaborative",
+        "CollaborativeDocument",
+        CollaborativeStatus,
+    ),
+]
+
+
+@pytest.mark.parametrize("module,handler,model,document,status", CASES)
+@pytest.mark.parametrize("publishing", [True, False])
+def test_invalidates_only_after_the_refreshed_index_write(
+    module, handler, model, document, status, publishing
+) -> None:
+    """A search between the bump and the index write would re-cache stale results."""
+    calls = []
+    before, after = (status.DRAFT, status.PUBLISHED) if publishing else (status.PUBLISHED, status.DRAFT)
+    instance = SimpleNamespace(pk=1, id=1, title="t", status=after)
+
+    doc = MagicMock()
+    doc.update.side_effect = lambda *a, **kw: calls.append(("index", kw.get("refresh")))
+    doc.delete.side_effect = lambda *a, **kw: calls.append(("index", kw.get("refresh")))
+    document_class = MagicMock(return_value=doc)
+    document_class.get.return_value = doc
+
+    model_class = MagicMock()
+    model_class.objects.get.return_value = SimpleNamespace(status=before)
+
+    with patch.object(module, model, model_class), patch.object(
+        module, document, document_class
+    ), patch.object(
+        module, "invalidate_search_cache", side_effect=lambda: calls.append(("invalidate", None))
+    ):
+        getattr(module, handler)(sender=None, instance=instance)
+
+    assert calls == [("index", True), ("invalidate", None)]