diff --git a/api/signals/collaborative_signals.py b/api/signals/collaborative_signals.py index 2556048..23a30eb 100644 --- a/api/signals/collaborative_signals.py +++ b/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,15 +35,10 @@ 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, @@ -53,14 +46,19 @@ def handle_collaborative_publication(sender: Any, instance: Collaborative, **kwa 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", diff --git a/api/signals/dataset_signals.py b/api/signals/dataset_signals.py index 08517e6..3d4ac98 100644 --- a/api/signals/dataset_signals.py +++ b/api/signals/dataset_signals.py @@ -2,7 +2,6 @@ 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 @@ -10,11 +9,9 @@ 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( diff --git a/api/signals/usecase_signals.py b/api/signals/usecase_signals.py index f3547ec..324bf94 100644 --- a/api/signals/usecase_signals.py +++ b/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 diff --git a/api/utils/search_cache.py b/api/utils/search_cache.py new file mode 100644 index 0000000..9dd8aa5 --- /dev/null +++ b/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) diff --git a/api/views/paginated_elastic_view.py b/api/views/paginated_elastic_view.py index c8b77a8..9f81f09 100644 --- a/api/views/paginated_elastic_view.py +++ b/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. diff --git a/api/views/search_unified.py b/api/views/search_unified.py index b9c86a0..e718d58 100644 --- a/api/views/search_unified.py +++ b/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 ( diff --git a/tests/test_search_cache_invalidation.py b/tests/test_search_cache_invalidation.py new file mode 100644 index 0000000..3d737b3 --- /dev/null +++ b/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)]