Address second review: stampede lock, cache key consistency, error handling, tests

- Add cache.add lock to prevent duplicate task dispatch on concurrent requests
- Pass combined_cache_key from view to task to avoid key mismatch from
  race conditions between dispatch and execution
- Handle Conference.DoesNotExist in task for deleted conferences
- Clean up computing lock in finally block
- Align frontend poll timeout (3min) with error cache TTL (2min)
- Add integration tests: task cache population, error caching, missing conference
- Add stampede prevention test (cache.add returns False)

Author: patrick91

backend/reviews/admin.py

@@ -470,13 +470,15 @@ def review_recap_compute_analysis_view(self, request, review_session_id):
             if cached_result is not None:
                 return JsonResponse(cached_result)
 
-        # Dispatch the Celery task to the heavy_processing queue
-        compute_recap_analysis.apply_async(
-            args=[conference.id],
-            kwargs={"force_recompute": force_recompute},
-            queue="heavy_processing",
-        )
-        check_pending_heavy_processing_work.delay()
+        # Use cache.add as a lock to prevent duplicate task dispatch
+        computing_key = f"{combined_cache_key}:computing"
+        if cache.add(computing_key, True, timeout=600):
+            compute_recap_analysis.apply_async(
+                args=[conference.id, combined_cache_key],
+                kwargs={"force_recompute": force_recompute},
+                queue="heavy_processing",
+            )
+            check_pending_heavy_processing_work.delay()
 
         return JsonResponse({"status": "processing"})
 

backend/reviews/tasks.py

@@ -6,25 +6,26 @@
 
 
 @app.task
-def compute_recap_analysis(conference_id, force_recompute=False):
+def compute_recap_analysis(conference_id, combined_cache_key, force_recompute=False):
     from django.core.cache import cache
 
+    from conferences.models import Conference
     from reviews.admin import get_accepted_submissions
     from reviews.similar_talks import (
-        _get_cache_key,
         compute_similar_talks,
         compute_topic_clusters,
     )
 
-    from conferences.models import Conference
+    try:
+        conference = Conference.objects.get(id=conference_id)
+    except Conference.DoesNotExist:
+        logger.error(
+            "Conference %s not found for recap analysis", conference_id
+        )
+        return
 
-    conference = Conference.objects.get(id=conference_id)
     accepted_submissions = list(get_accepted_submissions(conference))
 
-    combined_cache_key = _get_cache_key(
-        "recap_analysis", conference_id, accepted_submissions
-    )
-
     try:
         similar_talks = compute_similar_talks(
             accepted_submissions,
@@ -72,6 +73,8 @@ def compute_recap_analysis(conference_id, force_recompute=False):
         cache.set(
             combined_cache_key,
             {"status": "error", "message": "Analysis failed. Please try again."},
-            60 * 5,
+            60 * 2,
         )
         raise
+    finally:
+        cache.delete(f"{combined_cache_key}:computing")

backend/reviews/templates/reviews-recap.html

@@ -573,7 +573,7 @@ <h2 class="recap-section-title">🔗 Similar Talks</h2>
   var pollTimer = null;
   var pollStartTime = null;
   var pollAttempt = 0;
-  var POLL_TIMEOUT = 120000;
+  var POLL_TIMEOUT = 180000;
 
   function getNextPollInterval() {
     // Exponential backoff: 1s, 2s, 3s, 5s, 5s, 5s...

backend/reviews/tests/test_recap.py

@@ -3,6 +3,7 @@
 import pytest
 from django.contrib.admin import AdminSite
 from django.core.exceptions import PermissionDenied
+from django.test import override_settings
 
 from conferences.tests.factories import ConferenceFactory
 from reviews.admin import ReviewSessionAdmin
@@ -182,16 +183,24 @@ def test_recap_view_redirects_when_shortlist_not_visible(rf, mocker):
 # --- review_recap_compute_analysis_view tests ---
 
 
+FAKE_CACHE_KEY = "recap_analysis:conf_test:abc123"
+
+
 def _mock_analysis_deps(mocker, cache_return=None):
     """Mock the lazy-imported dependencies used in the compute analysis view."""
-    mock_cache = mocker.patch("django.core.cache.cache.get", return_value=cache_return)
+    mock_cache_get = mocker.patch(
+        "django.core.cache.cache.get", return_value=cache_return
+    )
+    mock_cache_add = mocker.patch(
+        "django.core.cache.cache.add", return_value=True
+    )
     mock_task = mocker.patch("reviews.tasks.compute_recap_analysis.apply_async")
     mock_check = mocker.patch("pycon.tasks.check_pending_heavy_processing_work.delay")
     mocker.patch(
         "reviews.similar_talks._get_cache_key",
-        return_value="recap_analysis:conf_test:abc123",
+        return_value=FAKE_CACHE_KEY,
     )
-    return mock_cache, mock_task, mock_check
+    return mock_cache_get, mock_cache_add, mock_task, mock_check
 
 
 def test_compute_analysis_view_returns_cached_result(rf, mocker):
@@ -226,7 +235,9 @@ def test_compute_analysis_view_returns_cached_result(rf, mocker):
         },
     }
 
-    mock_cache, mock_task, _ = _mock_analysis_deps(mocker, cache_return=cached_data)
+    mock_cache_get, _, mock_task, _ = _mock_analysis_deps(
+        mocker, cache_return=cached_data
+    )
 
     request = rf.get("/")
     request.user = user
@@ -248,7 +259,7 @@ def test_compute_analysis_view_returns_cached_result(rf, mocker):
 def test_compute_analysis_view_dispatches_task_on_cache_miss(rf, mocker):
     user, conference, review_session, submissions = _create_recap_setup()
 
-    _, mock_task, mock_check = _mock_analysis_deps(mocker, cache_return=None)
+    _, _, mock_task, mock_check = _mock_analysis_deps(mocker, cache_return=None)
 
     request = rf.get("/")
     request.user = user
@@ -261,7 +272,7 @@ def test_compute_analysis_view_dispatches_task_on_cache_miss(rf, mocker):
     assert data == {"status": "processing"}
 
     mock_task.assert_called_once_with(
-        args=[conference.id],
+        args=[conference.id, FAKE_CACHE_KEY],
         kwargs={"force_recompute": False},
         queue="heavy_processing",
     )
@@ -272,7 +283,7 @@ def test_compute_analysis_view_dispatches_task_on_cache_miss(rf, mocker):
 def test_compute_analysis_view_dispatches_task_with_recompute(rf, mocker):
     user, conference, review_session, submissions = _create_recap_setup()
 
-    _, mock_task, _ = _mock_analysis_deps(mocker, cache_return=None)
+    _, _, mock_task, _ = _mock_analysis_deps(mocker, cache_return=None)
 
     request = rf.get("/?recompute=1")
     request.user = user
@@ -293,7 +304,9 @@ def test_compute_analysis_view_recompute_skips_cache(rf, mocker):
     user, conference, review_session, submissions = _create_recap_setup()
 
     cached_data = {"submissions_list": [], "topic_clusters": {"topics": []}}
-    mock_cache, mock_task, _ = _mock_analysis_deps(mocker, cache_return=cached_data)
+    mock_cache_get, _, mock_task, _ = _mock_analysis_deps(
+        mocker, cache_return=cached_data
+    )
 
     request = rf.get("/?recompute=1")
     request.user = user
@@ -305,10 +318,33 @@ def test_compute_analysis_view_recompute_skips_cache(rf, mocker):
     assert data == {"status": "processing"}
 
     # Cache should NOT have been checked when recompute=1
-    mock_cache.assert_not_called()
+    mock_cache_get.assert_not_called()
     mock_task.assert_called_once()
 
 
+def test_compute_analysis_view_skips_dispatch_when_already_computing(rf, mocker):
+    user, conference, review_session, submissions = _create_recap_setup()
+
+    mock_cache_get, mock_cache_add, mock_task, mock_check = _mock_analysis_deps(
+        mocker, cache_return=None
+    )
+    # Simulate lock already held — cache.add returns False
+    mock_cache_add.return_value = False
+
+    request = rf.get("/")
+    request.user = user
+
+    admin = ReviewSessionAdmin(ReviewSession, AdminSite())
+    response = admin.review_recap_compute_analysis_view(request, review_session.id)
+
+    data = json.loads(response.content)
+    assert data == {"status": "processing"}
+
+    # Task should NOT be dispatched since lock was already held
+    mock_task.assert_not_called()
+    mock_check.assert_not_called()
+
+
 def test_compute_analysis_view_permission_denied_for_non_reviewer(rf):
     user = UserFactory(is_staff=True, is_superuser=False)
     conference = ConferenceFactory()
@@ -342,3 +378,101 @@ def test_compute_analysis_view_permission_denied_when_shortlist_not_visible(rf):
 
     with pytest.raises(PermissionDenied):
         admin.review_recap_compute_analysis_view(request, review_session.id)
+
+
+# --- compute_recap_analysis task tests ---
+
+
+LOCMEM_CACHE = {
+    "default": {
+        "BACKEND": "django.core.cache.backends.locmem.LocMemCache",
+        "LOCATION": "test-recap-analysis",
+    }
+}
+
+
+@pytest.mark.django_db
+@override_settings(CACHES=LOCMEM_CACHE)
+def test_task_populates_cache_with_results(mocker):
+    from django.core.cache import cache
+
+    from reviews.tasks import compute_recap_analysis
+
+    user, conference, review_session, submissions = _create_recap_setup()
+    sub1, sub2 = submissions
+
+    mocker.patch(
+        "reviews.similar_talks.compute_similar_talks",
+        return_value={
+            sub1.id: [{"id": sub2.id, "title": str(sub2.title), "similarity": 75.0}],
+            sub2.id: [],
+        },
+    )
+    mocker.patch(
+        "reviews.similar_talks.compute_topic_clusters",
+        return_value={
+            "topics": [
+                {"name": "ML", "count": 2, "keywords": ["ml"], "submissions": []}
+            ],
+            "outliers": [],
+            "submission_topics": {},
+        },
+    )
+
+    cache_key = "recap_analysis:conf_test:integration"
+    # Set computing lock to verify it gets cleaned up
+    cache.set(f"{cache_key}:computing", True)
+
+    result = compute_recap_analysis(conference.id, cache_key)
+
+    assert len(result["submissions_list"]) == 2
+    assert result["submissions_list"][0]["id"] == sub1.id
+    assert result["submissions_list"][0]["similar"][0]["similarity"] == 75.0
+    assert result["topic_clusters"]["topics"][0]["name"] == "ML"
+
+    # Verify cache was populated
+    cached = cache.get(cache_key)
+    assert cached == result
+
+    # Verify computing lock was cleaned up
+    assert cache.get(f"{cache_key}:computing") is None
+
+
+@pytest.mark.django_db
+@override_settings(CACHES=LOCMEM_CACHE)
+def test_task_caches_error_on_failure(mocker):
+    from django.core.cache import cache
+
+    from reviews.tasks import compute_recap_analysis
+
+    user, conference, review_session, submissions = _create_recap_setup()
+
+    mocker.patch(
+        "reviews.similar_talks.compute_similar_talks",
+        side_effect=RuntimeError("ML model failed"),
+    )
+
+    cache_key = "recap_analysis:conf_test:error"
+    cache.set(f"{cache_key}:computing", True)
+
+    with pytest.raises(RuntimeError, match="ML model failed"):
+        compute_recap_analysis(conference.id, cache_key)
+
+    # Verify error was cached
+    cached = cache.get(cache_key)
+    assert cached["status"] == "error"
+    assert "failed" in cached["message"].lower()
+
+    # Verify computing lock was cleaned up
+    assert cache.get(f"{cache_key}:computing") is None
+
+
+def test_task_handles_missing_conference(mocker):
+    from reviews.tasks import compute_recap_analysis
+
+    mock_similar = mocker.patch("reviews.similar_talks.compute_similar_talks")
+
+    result = compute_recap_analysis(999999, "recap_analysis:conf_999999:key")
+
+    assert result is None
+    mock_similar.assert_not_called()