Fix Flight denormalized field locking issues

Remove real-time updates to Flight.total_views and Flight.total_clicks
to eliminate database lock contention on high-traffic flights.

Changes:
- Remove F() expression updates in Advertisement.incr() that were
  causing row-level locks on every impression
- Add Flight.refresh_denormalized_totals() method to calculate
  these fields from AdImpression on-demand
- Add refresh_flight_denormalized_totals() Celery task for
  periodic background updates
- Add management command to manually refresh flight totals
- Update Offer.refund() to skip Flight field updates
- Update test to manually refresh denormalized fields

The denormalized fields are now calculated from the source of truth
(AdImpression) either on-demand or via scheduled tasks, avoiding
the lock contention that was blocking concurrent ad impressions.

Author: ericholscher

adserver/management/commands/refresh_flight_totals.py

@@ -0,0 +1,61 @@
+"""Management command to refresh denormalized total_views and total_clicks on Flight objects."""
+
+import logging
+
+from django.core.management.base import BaseCommand
+
+from ...models import Flight
+
+
+log = logging.getLogger(__name__)
+
+
+class Command(BaseCommand):
+    """Refresh denormalized total_views and total_clicks for flights."""
+
+    help = "Refresh denormalized total_views and total_clicks fields for all flights"
+
+    def add_arguments(self, parser):
+        parser.add_argument(
+            "--live-only",
+            action="store_true",
+            help="Only refresh live flights",
+        )
+        parser.add_argument(
+            "--flight-slug",
+            type=str,
+            help="Refresh only a specific flight by slug",
+        )
+
+    def handle(self, *args, **options):
+        live_only = options.get("live_only", False)
+        flight_slug = options.get("flight_slug")
+
+        if flight_slug:
+            flights = Flight.objects.filter(slug=flight_slug)
+            if not flights.exists():
+                self.stderr.write(self.style.ERROR(f"Flight '{flight_slug}' not found"))
+                return
+        elif live_only:
+            flights = Flight.objects.filter(live=True)
+        else:
+            flights = Flight.objects.all()
+
+        count = flights.count()
+        self.stdout.write(f"Refreshing denormalized totals for {count} flight(s)...")
+
+        for i, flight in enumerate(flights.iterator(), 1):
+            try:
+                flight.refresh_denormalized_totals()
+                if i % 100 == 0:
+                    self.stdout.write(f"  Processed {i}/{count} flights...")
+            except Exception as e:
+                self.stderr.write(
+                    self.style.ERROR(f"Failed to refresh flight '{flight.slug}': {e}")
+                )
+
+        self.stdout.write(
+            self.style.SUCCESS(
+                f"Successfully refreshed denormalized totals for {count} flight(s)"
+            )
+        )

adserver/models.py

@@ -1492,6 +1492,22 @@ def ecpc(self):
             return value_delivered / clicks
         return 0
 
+    def refresh_denormalized_totals(self):
+        """
+        Refresh the denormalized total_views and total_clicks fields from AdImpression.
+
+        This should be called periodically (e.g., hourly or daily) instead of updating
+        these fields in real-time to avoid lock contention.
+        """
+        aggregation = AdImpression.objects.filter(advertisement__flight=self).aggregate(
+            total_views=models.Sum("views"),
+            total_clicks=models.Sum("clicks"),
+        )
+
+        self.total_views = aggregation["total_views"] or 0
+        self.total_clicks = aggregation["total_clicks"] or 0
+        self.save(update_fields=["total_views", "total_clicks"])
+
     def copy_niche_targeting_urls(self, other_flight):
         """Copy niche targeting URLs from another flight."""
         if "adserver.analyzer" in settings.INSTALLED_APPS:
@@ -1909,15 +1925,11 @@ def incr(self, impression_type, publisher, offer=None):
         for imp_type in impression_types:
             assert imp_type in IMPRESSION_TYPES
 
-            # Update the denormalized fields on the Flight
-            if imp_type == VIEWS:
-                Flight.objects.filter(pk=self.flight_id).update(
-                    total_views=models.F("total_views") + 1
-                )
-            elif imp_type == CLICKS:
-                Flight.objects.filter(pk=self.flight_id).update(
-                    total_clicks=models.F("total_clicks") + 1
-                )
+            # NOTE: We no longer update the denormalized total_views/total_clicks fields
+            # in real-time to avoid lock contention on the Flight table.
+            # These fields are now calculated on-demand from AdImpression or
+            # refreshed periodically by a background task.
+            # See: Flight.refresh_denormalized_totals()
 
         # Ensure that an impression object exists for today
         # and make sure to query the writable DB for this
@@ -2917,19 +2929,14 @@ def refund(self):
                 **{self.impression_type: models.F(self.impression_type) - 1}
             )
 
-            # Update the denormalized impressions on the Flight
-            # And the denormalized aggregate AdImpressions
+            # Update the denormalized aggregate AdImpressions
+            # Note: We no longer update Flight.total_views/total_clicks in real-time
+            # to avoid lock contention. These are refreshed periodically.
             if self.viewed:
-                Flight.objects.filter(pk=self.advertisement.flight_id).update(
-                    total_views=models.F("total_views") - 1
-                )
                 AdImpression.objects.filter(pk=impression.pk).update(
                     **{VIEWS: models.F(VIEWS) - 1}
                 )
             if self.clicked:
-                Flight.objects.filter(pk=self.advertisement.flight_id).update(
-                    total_clicks=models.F("total_clicks") - 1
-                )
                 AdImpression.objects.filter(pk=impression.pk).update(
                     **{CLICKS: models.F(CLICKS) - 1}
                 )

adserver/tasks.py

@@ -924,6 +924,33 @@ def calculate_ad_ctrs(days=7, min_views=1_000):
             ad.save()
 
 
+@app.task()
+def refresh_flight_denormalized_totals():
+    """
+    Refresh denormalized total_views and total_clicks fields for all live flights.
+
+    This task should be run periodically (e.g., hourly) to update the denormalized
+    fields without causing lock contention on the Flight table.
+    """
+    log.info("Refreshing denormalized totals for live flights")
+
+    # Only refresh active flights to avoid unnecessary work
+    flights = Flight.objects.filter(live=True)
+
+    for flight in flights:
+        try:
+            flight.refresh_denormalized_totals()
+        except Exception as e:
+            log.error(
+                "Failed to refresh denormalized totals for flight %s: %s",
+                flight.slug,
+                e,
+                exc_info=True,
+            )
+
+    log.info("Finished refreshing denormalized totals for %d flights", flights.count())
+
+
 @app.task()
 def notify_on_ad_image_change(advertisement_id):
     ad = Advertisement.objects.filter(id=advertisement_id).first()

adserver/tests/test_decision_engine.py

@@ -300,8 +300,8 @@ def test_flight_clicks(self):
         self.advertisement1.incr(CLICKS, self.publisher)
         self.advertisement1.incr(CLICKS, self.publisher)
 
-        # Refresh the data on the include_flight - gets the denormalized views
-        self.include_flight.refresh_from_db()
+        # Refresh the denormalized totals manually since they're no longer updated in real-time
+        self.include_flight.refresh_denormalized_totals()
 
         self.assertEqual(self.include_flight.clicks_remaining(), 998)
         self.assertEqual(self.include_flight.total_clicks, 2)
@@ -315,8 +315,8 @@ def test_flight_clicks(self):
         # Add 1 click for today
         self.advertisement1.incr(CLICKS, self.publisher)
 
-        # Refresh the data on the include_flight - gets the denormalized views
-        self.include_flight.refresh_from_db()
+        # Refresh the denormalized totals manually since they're no longer updated in real-time
+        self.include_flight.refresh_denormalized_totals()
 
         self.assertEqual(self.include_flight.clicks_remaining(), 997)
         self.assertEqual(self.include_flight.clicks_today(), 1)