delete expired results in batches to avoid running OOM

Author: saz

django_celery_results/managers.py

@@ -84,10 +84,16 @@ def get_all_expired(self, expires):
         """Get all expired results."""
         return self.filter(date_done__lt=now() - maybe_timedelta(expires))
 
-    def delete_expired(self, expires):
+    def delete_expired(self, expires, batch_size=100000):
         """Delete all expired results."""
-        with transaction.atomic(using=self.db):
-            self.get_all_expired(expires).delete()
+        qs = self.get_all_expired(expires).order_by("id")
+
+        while True:
+            ids = list(qs.values_list("id", flat=True)[:batch_size])
+            if not ids:
+                break
+            with transaction.atomic(using=self.db):
+                self.filter(id__in=ids).delete()
 
 
 class TaskResultManager(ResultManager):

t/integration/benchmark_models.py

@@ -67,4 +67,4 @@ def test_taskresult_delete_expired(self):
             'setup time: {setup:.2f}\n'
             'bench time: {bench:.2f}\n'
         ).format(setup=after_setup - start, bench=done - after_setup))
-        assert self.benchmark.stats.stats.max < 1
+        assert self.benchmark.stats.stats.max < 5

t/unit/test_models.py

@@ -210,6 +210,22 @@ class TransactionError(Exception):
         except TransactionError:
             pass
 
+    def test_result_batch_deletion(self):
+        # Create 200 expired records
+        TaskResult.objects.bulk_create(
+            [
+                TaskResult(task_id=uuid(), date_done=now() - timedelta(days=1))
+                for i in range(200)
+            ]
+        )
+        assert TaskResult.objects.get_all_expired(0).count() == 200
+
+        # Run deletion in small batches
+        TaskResult.objects.delete_expired(0, batch_size=25)
+
+        # All expired records should be gone
+        assert TaskResult.objects.get_all_expired(0).count() == 0
+
 
 @pytest.mark.usefixtures('depends_on_current_app')
 class test_ModelsWithoutDefaultDB(TransactionTestCase):