fix(cohorts): guard cohort recalculation Redis calls

products/cohorts/backend/models/dependencies.py

@@ -315,22 +315,33 @@ def warm_team_cohort_dependency_cache(team_id: int, batch_size: int = 1000):
 
 
 def _on_cohort_changed(cohort: Cohort, always_invalidate: bool = False):
-    new_dependencies = extract_cohort_dependencies(cohort)
-    existing_dependencies = cache.get(_cohort_dependencies_key(cohort.id))
-    dependencies_changed = existing_dependencies is None or set(existing_dependencies) != new_dependencies
+    # Under autocommit this runs synchronously inside Cohort.save(), so a Redis error here escapes the
+    # save and can leave the cohort stuck with is_calculating=True. The dependency caches rebuild on the
+    # next change or warm pass, so a failed refresh is safe to log and continue past.
+    try:
+        new_dependencies = extract_cohort_dependencies(cohort)
+        existing_dependencies = cache.get(_cohort_dependencies_key(cohort.id))
+        dependencies_changed = existing_dependencies is None or set(existing_dependencies) != new_dependencies
 
-    # If the dependencies haven't changed, no need to refresh the cache
-    if not always_invalidate and not cohort.deleted and not dependencies_changed:
-        return
+        # If the dependencies haven't changed, no need to refresh the cache
+        if not always_invalidate and not cohort.deleted and not dependencies_changed:
+            return
 
-    cache.delete(_cohort_dependencies_key(cohort.id))
-    cache.delete(_cohort_dependents_key(cohort.id))
+        cache.delete(_cohort_dependencies_key(cohort.id))
+        cache.delete(_cohort_dependents_key(cohort.id))
 
-    if existing_dependencies:
-        for dep_id in existing_dependencies:
-            cache.delete(_cohort_dependents_key(dep_id))
+        if existing_dependencies:
+            for dep_id in existing_dependencies:
+                cache.delete(_cohort_dependents_key(dep_id))
 
-    warm_team_cohort_dependency_cache(cohort.team_id)
+        warm_team_cohort_dependency_cache(cohort.team_id)
+    except Exception as error:
+        logger.exception(
+            "failed_to_refresh_cohort_dependency_cache",
+            cohort_id=cohort.pk,
+            team_id=cohort.team_id,
+            error=str(error),
+        )
 
 
 def _has_backfillable_filters(cohort: Cohort, kind: CohortBackfillKind) -> bool:

products/cohorts/backend/models/test/test_dependencies.py

@@ -258,6 +258,20 @@ def test_warm_team_cohort_dependency_cache_refreshes_ttl(self) -> None:
             args, kwargs = mock_set_many.call_args
             self.assertEqual(kwargs.get("timeout"), DEPENDENCY_CACHE_TIMEOUT)
 
+    def test_cohort_save_survives_redis_error_in_dependency_refresh(self) -> None:
+        # A non-recalculation save fires _on_cohort_changed synchronously (see mock_transaction).
+        # A Redis error there must not escape Cohort.save() and leave the cohort stuck calculating.
+        cohort = self._create_cohort(name="Test Cohort")
+        with mock.patch(
+            "products.cohorts.backend.models.dependencies.cache.get",
+            side_effect=Exception("redis unavailable"),
+        ):
+            cohort.name = "renamed"
+            cohort.save()
+
+        cohort.refresh_from_db()
+        self.assertEqual(cohort.name, "renamed")
+
     def test_cache_miss_get_cohort_dependencies(self) -> None:
         cohort_a = self._create_cohort(name="Test Cohort A")
         cohort_b = self._create_cohort(

products/cohorts/backend/models/test/test_util.py

@@ -1360,10 +1360,17 @@ def test_original_error_surfaces_when_recovery_history_save_fails(self):
         assert mock_history.save.call_count == 2
         mock_connections[DEFAULT_DB_ALIAS].close.assert_called_once()
 
-    def test_original_error_surfaces_when_reset_calculating_save_fails(self):
-        # calculate_people_ch's finally block resets is_calculating on the same connection the recovery
-        # bookkeeping save uses. If that reset write hits the dropped connection it must not raise out of
-        # finally and mask the real calculation error - it runs through the same reconnect-and-retry.
+    @parameterized.expand(
+        [
+            # A dropped Postgres connection, and a Redis error escaping the save's post_save signals.
+            ("connection_error", OperationalError("the connection is closed")),
+            ("redis_error", Exception("redis unavailable")),
+        ]
+    )
+    def test_original_error_surfaces_when_reset_calculating_save_fails(self, _name, bookkeeping_error):
+        # calculate_people_ch's finally block resets is_calculating through the recovery bookkeeping.
+        # If that reset write fails - a dropped connection, or a Redis error from the save's signals -
+        # it must not raise out of finally and mask the real calculation error.
         cohort = _create_cohort(
             team=self.team,
             name="c",
@@ -1376,11 +1383,11 @@ def test_original_error_surfaces_when_reset_calculating_save_fails(self):
                 "products.cohorts.backend.models.util.recalculate_cohortpeople",
                 side_effect=real_error,
             ),
-            # The is_calculating reset write fails on the dropped connection, initial attempt and retry.
+            # The is_calculating reset write fails, initial attempt and retry.
             patch.object(
                 Cohort,
                 "_safe_reset_calculating_state",
-                side_effect=OperationalError("the connection is closed"),
+                side_effect=bookkeeping_error,
             ) as mock_reset,
             # Patch connections so the reconnect doesn't close the real test transaction's connection.
             patch("products.cohorts.backend.models.util.connections") as mock_connections,

products/cohorts/backend/models/util.py

@@ -8,7 +8,7 @@
 from typing import TYPE_CHECKING, Any, Optional, Union, cast
 
 from django.conf import settings
-from django.db import DEFAULT_DB_ALIAS, InterfaceError, OperationalError, connections
+from django.db import DEFAULT_DB_ALIAS, connections
 from django.utils import timezone
 
 import structlog
@@ -156,15 +156,16 @@ def save_recovery_bookkeeping(save_fn: Callable[[], None], *, cohort_id: int, te
     """Persist post-calculation bookkeeping, surviving a Postgres connection dropped mid-recalculation.
 
     A long recalculation can outlive its connection (the server closes it unexpectedly); the first
-    write afterwards then raises a connection error. Left unguarded on the error/finally recovery
-    path, that cascades into "the connection is closed" - burying the real root-cause error and
-    leaving the cohort stuck with is_calculating=True. Reconnect and retry once so the bookkeeping
-    still lands and the original error is what propagates; if the retry fails too, swallow it (a
-    recovery write must never mask the failure it is recording).
+    write afterwards then raises a connection error. The save also fires post_save signal handlers
+    that touch Redis, so a Redis blip can escape here too. Left unguarded on the error/finally
+    recovery path, either failure buries the real root-cause error and leaves the cohort stuck with
+    is_calculating=True. Reconnect and retry once so the bookkeeping still lands and the original
+    error is what propagates; if the retry fails too, swallow it (a recovery write must never mask
+    the failure it is recording).
     """
     try:
         save_fn()
-    except (InterfaceError, OperationalError):
+    except Exception:
         connections[DEFAULT_DB_ALIAS].close()  # next query opens a fresh connection
         try:
             save_fn()
@@ -224,14 +225,20 @@ def run_cohort_query(
         # If calculation succeeded and we scheduled a delayed task, cancel it and run immediately
         # This avoids waiting the full timeout when the query completed quickly
         if delayed_task and history and query and not settings.TEST:
-            if delayed_task.state in ["PENDING", "RECEIVED"]:
-                delayed_task.revoke()  # Cancel the delayed task
+            # The recalculation result is already computed. This rescheduling is best-effort stats
+            # bookkeeping that talks to the Celery result backend and broker, so a Redis blip must
+            # not discard the result. The delayed task still runs after its original countdown.
+            try:
+                if delayed_task.state in ["PENDING", "RECEIVED"]:
+                    delayed_task.revoke()  # Cancel the delayed task
 
-            # Run immediately since the query already completed
-            collect_cohort_query_stats.apply_async(
-                args=[cohort_tag, cohort_id, start_time.isoformat(), history.id, query],
-                countdown=COHORT_STATS_COLLECTION_DELAY_SECONDS,
-            )
+                # Run immediately since the query already completed
+                collect_cohort_query_stats.apply_async(
+                    args=[cohort_tag, cohort_id, start_time.isoformat(), history.id, query],
+                    countdown=COHORT_STATS_COLLECTION_DELAY_SECONDS,
+                )
+            except Exception as error:
+                logger.warning("cohort_stats_collection_scheduling_failed", cohort_id=cohort_id, error=str(error))
 
         return result, end_time