Repository: saleor/saleor
Pull request: Delete event deliveries in batches
This is a port of changes from #13246 since we decided to merge it to main instead of 3.9.

I want to merge this change because it adds batches to event deliveries delete task. Fixes https://github.com/saleor/saleor/issues/13235
I'm switching back to using django delete() instead od raw SQL query (which was introduced in https://github.com/saleor/saleor/pull/9806 ) and use the logic that is used in zero-downtime migrations to make the task more optimal.
It will now delete all the EventDelivery objects and by CASCADE all EventDeliveryAttempts (max 5 attempts per delivery) and when it is done it will take care of unused EventPayloads

PERFORMANCE NOTE: 
`BATCH=1000` - was tested on large DB with multiple attempts per delivery and big JSON payloads. Tasks took less than 1s and no significant raise in memory usage (around 1%), tested also with multiple workers at once with no crashes. 

<!-- Please mention all relevant issue numbers. -->

# Impact

* [ ] New migrations
* [ ] New/Updated API fields or mutations
* [ ] Deprecated API fields or mutations
* [ ] Removed API types, fields, or mutations
* [ ] Documentation needs to be updated

# Pull Request Checklist

<!-- Please keep this section. It will make the maintainer's life easier. -->

* [ ] Privileged queries and mutations are either absent or guarded by proper permission checks
* [ ] Database queries are optimized and the number of queries is constant
* [ ] Database migrations are either absent or optimized for zero downtime
* [ ] The changes are covered by test cases


Problem statement:
Handle event deliveries delete in batches
Event deliveries, event delivery attempts and payloads are deleted in a `delete_event_payloads_task` without using batches.
This should be done in batches as it can lead to timeouts on db in case of large number of event delivery fails in two weeks period.

Hints from issue discussion:

Commit message: fix linters

Patch to review:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 2b870aec097..0416847b34e 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -98,6 +98,7 @@ All notable, unreleased changes to this project will be documented in this file.
 - Add missing descriptions to Product module. - #13259 by @FremahA
 - Add missing description for Invoice module - #13265 by @DevilsAutumn
 - Add missing descriptions to Discount module - #13261 by @devilsautumn
+- Fix deleting event payloads by adding batches - #13368 by @jakubkuc
 
 # 3.14.0
 
diff --git a/saleor/core/tasks.py b/saleor/core/tasks.py
index 59401971c2b..b1c47854229 100644
--- a/saleor/core/tasks.py
+++ b/saleor/core/tasks.py
@@ -1,11 +1,16 @@
 from botocore.exceptions import ClientError
 from django.conf import settings
 from django.core.files.storage import default_storage
-from django.db.models import Exists, OuterRef, Q
+from django.db.models import Exists, OuterRef
 from django.utils import timezone
 
 from ..celeryconf import app
-from .models import EventDelivery, EventDeliveryAttempt, EventPayload
+from .models import EventDelivery, EventPayload
+
+# Batch size was tested on db with 1mln payloads and deliveries, each delivery
+# had multiple attempts. One task took less than 0,5 second, memory usage didn't raise
+# more than 1 percent.
+BATCH_SIZE = 1000
 
 
 @app.task
@@ -16,18 +21,15 @@ def delete_from_storage_task(path):
 @app.task
 def delete_event_payloads_task():
     delete_period = timezone.now() - settings.EVENT_PAYLOAD_DELETE_PERIOD
-    deliveries = EventDelivery.objects.filter(created_at__lte=delete_period)
-    attempts = EventDeliveryAttempt.objects.filter(
-        Q(Exists(deliveries.filter(id=OuterRef("delivery_id"))))
-        | Q(delivery__isnull=True, created_at__lte=delete_period)
-    )
-    payloads = EventPayload.objects.filter(
-        deliveries__isnull=True, created_at__lte=delete_period
-    )
-
-    attempts._raw_delete(attempts.db)  # type: ignore[attr-defined] # raw access # noqa: E501
-    deliveries._raw_delete(deliveries.db)  # type: ignore[attr-defined] # raw access # noqa: E501
-    payloads._raw_delete(payloads.db)  # type: ignore[attr-defined] # raw access # noqa: E501
+    valid_deliveries = EventDelivery.objects.filter(created_at__gt=delete_period)
+    payloads_to_delete = EventPayload.objects.filter(
+        ~Exists(valid_deliveries.filter(payload_id=OuterRef("id")))
+    ).order_by("-pk")
+    ids = payloads_to_delete.values_list("pk", flat=True)[:BATCH_SIZE]
+    qs = EventPayload.objects.filter(pk__in=ids)
+    if ids:
+        qs.delete()
+        delete_event_payloads_task.delay()
 
 
 @app.task(


Code review comments:
diff_hunk: @@ -16,18 +21,15 @@ def delete_from_storage_task(path):
 @app.task
 def delete_event_payloads_task():
     delete_period = timezone.now() - settings.EVENT_PAYLOAD_DELETE_PERIOD
-    deliveries = EventDelivery.objects.filter(created_at__lte=delete_period)
-    attempts = EventDeliveryAttempt.objects.filter(
-        Q(Exists(deliveries.filter(id=OuterRef("delivery_id"))))
-        | Q(delivery__isnull=True, created_at__lte=delete_period)
-    )
-    payloads = EventPayload.objects.filter(
-        deliveries__isnull=True, created_at__lte=delete_period
-    )
-
-    attempts._raw_delete(attempts.db)  # type: ignore[attr-defined] # raw access # noqa: E501
-    deliveries._raw_delete(deliveries.db)  # type: ignore[attr-defined] # raw access # noqa: E501
-    payloads._raw_delete(payloads.db)  # type: ignore[attr-defined] # raw access # noqa: E501
+    valid_deliveries = EventDelivery.objects.filter(created_at__gt=delete_period)
+    payloads_to_delete = EventPayload.objects.filter(
+        ~Exists(valid_deliveries.filter(payload_id=OuterRef("id")))
+    ).order_by("-pk")
+    ids = payloads_to_delete.values_list("pk", flat=True)[:BATCH_SIZE]
+    qs = EventPayload.objects.filter(pk__in=ids)
+    if ids:
+        qs.delete()
+        delete_event_payloads_task.delay()
path: saleor/core/tasks.py
text: @user1:
Same comment as here: https://github.com/saleor/saleor/pull/13246#discussion_r1258298416

@author:
We have decided that we should check all our tasks in search of such probability and make a separate PR with fix that will prevent that from happening in all of them. So that will be out of scope of this PR.

@user1:
Why not fix it now instead of introducing technical debt in a bug fix? This would be highly problematic if we start queuing tons of concurrent writes on the same rows as it is a "recursive" function.

You can do something similar as here: https://github.com/saleor/saleor/blob/5e7c57dad9e64b09477ebdcee53f0277359bc598/saleor/checkout/tasks.py#L86-L96


@author:
I don't get it to be honest. This would only limit the number of this task invocations. And we don't know how many will be needed to clean up all expired payloads/deliveries so we can't just set number of invocations to some fixed number.  Or am I missing something here? 

@user1:
You could also set an expiration date/timestamp and pass it to the task. If null then you need to give one to the sub tasks, otherwise you just check the value is lower than current time and pass it along the value.

@author:
And what do you think would be an appropriate time to timeout this task? 1 hour? (while testing 1m of payloads and deliveries took 5 min. to finish) 

@author:
@user1 I've made changes. What do you think about it?

@user1:
Looks good! 👌