email: Fix race conditions with concurrent ScheduledEmail handling.

The main race conditions, which actually happened in production was with
concurrent execution of deliver_email and clear_scheduled_emails.
clear_scheduled_emails could delete all email.users in the middle of
deliver_email execution, causing it to pass empty to_user_ids list to
send_email. We mitigate this by getting the list of user ids in a single
query and moving forward with that snapshot, not having to worry about
database data being mutated anymore.

clear_scheduled_emails had potential race conditions with concurrent
execution of itself due to not locking the appropriate rows upon
selecting them for the purpose of potentially deleting them. FOR UPDATE
locks need to be acquired to prevent simultaneous mutation.

Tested manually with some print+sleep debugging to make some races
happen.

fixes #zulip-2k (sentry)

zerver/tests/test_users.py

@@ -1284,6 +1284,25 @@ class ActivateTest(ZulipTestCase):
             )
         self.assertEqual(ScheduledEmail.objects.count(), 0)
 
+    def test_deliver_email_no_addressees(self) -> None:
+        iago = self.example_user('iago')
+        hamlet = self.example_user('hamlet')
+        to_user_ids = [hamlet.id, iago.id]
+        send_future_email('zerver/emails/followup_day1', iago.realm,
+                          to_user_ids=to_user_ids, delay=datetime.timedelta(hours=1))
+        self.assertEqual(ScheduledEmail.objects.count(), 1)
+        email = ScheduledEmail.objects.all().first()
+        email.users.remove(*to_user_ids)
+
+        with self.assertLogs('zulip.send_email', level='INFO') as info_log:
+            deliver_email(email)
+        from django.core.mail import outbox
+        self.assertEqual(len(outbox), 0)
+        self.assertEqual(ScheduledEmail.objects.count(), 1)
+        self.assertEqual(info_log.output, [
+            f'WARNING:zulip.send_email:ScheduledEmail id {email.id} has empty users and address attributes.'
+        ])
+
 class RecipientInfoTest(ZulipTestCase):
     def test_stream_recipient_info(self) -> None:
         hamlet = self.example_user('hamlet')