deliver_scheduled_*: SELECT FOR UPDATE the relevant rows.

`deliver_scheduled_emails` and `deliver_scheduled_messages` use their
respective tables like a queue, but do not have guarantees that there
was only one consumer (besides the EMAIL_DELIVERER_DISABLED setting),
and could send duplicate messages if multiple consumers raced in
reading rows.

Use database locking to ensure that the database only feeds a given
ScheduledMessage or ScheduledEmail row to a single consumer.  A second
consumer, if it exists, will block until the first consumer commits
the transaction.

zerver/management/commands/deliver_scheduled_emails.py

@@ -15,6 +15,7 @@ from typing import Any
 
 from django.conf import settings
 from django.core.management.base import BaseCommand
+from django.db import transaction
 from django.utils.timezone import now as timezone_now
 
 from zerver.lib.logging_util import log_to_file
@@ -42,17 +43,20 @@ Usage: ./manage.py deliver_scheduled_emails
             sleep_forever()
 
         while True:
-            email_jobs_to_deliver = ScheduledEmail.objects.filter(
-                scheduled_timestamp__lte=timezone_now()
-            )
-            if email_jobs_to_deliver:
-                for job in email_jobs_to_deliver:
-                    try:
-                        deliver_scheduled_emails(job)
-                    except EmailNotDeliveredException:
-                        logger.warning("%r not delivered", job)
+            found_rows = False
+            with transaction.atomic():
+                email_jobs_to_deliver = ScheduledEmail.objects.filter(
+                    scheduled_timestamp__lte=timezone_now()
+                ).select_for_update()
+                if email_jobs_to_deliver:
+                    for job in email_jobs_to_deliver:
+                        try:
+                            deliver_scheduled_emails(job)
+                        except EmailNotDeliveredException:
+                            logger.warning("%r not delivered", job)
+            # Less load on the db during times of activity,
+            # and more responsiveness when the load is low
+            if found_rows:
                 time.sleep(10)
             else:
-                # Less load on the db during times of activity,
-                # and more responsiveness when the load is low
                 time.sleep(2)

zerver/management/commands/deliver_scheduled_messages.py

@@ -5,6 +5,7 @@ from typing import Any
 
 from django.conf import settings
 from django.core.management.base import BaseCommand
+from django.db import transaction
 from django.utils.timezone import now as timezone_now
 
 from zerver.lib.actions import build_message_send_dict, do_send_messages
@@ -65,13 +66,14 @@ Usage: ./manage.py deliver_scheduled_messages
                 sleep_forever()
 
             while True:
-                messages_to_deliver = ScheduledMessage.objects.filter(
-                    scheduled_timestamp__lte=timezone_now(), delivered=False
-                )
-                for message in messages_to_deliver:
-                    do_send_messages([self.construct_message(message)])
-                    message.delivered = True
-                    message.save(update_fields=["delivered"])
+                with transaction.atomic():
+                    messages_to_deliver = ScheduledMessage.objects.filter(
+                        scheduled_timestamp__lte=timezone_now(), delivered=False
+                    ).select_for_update()
+                    for message in messages_to_deliver:
+                        do_send_messages([self.construct_message(message)])
+                        message.delivered = True
+                        message.save(update_fields=["delivered"])
 
                 cur_time = timezone_now()
                 time_next_min = (cur_time + timedelta(minutes=1)).replace(second=0, microsecond=0)