fix email batch retry loop and recover stale jobs (ea3ba9c1) · Commits · Samorząd Uczniowski / Wybory

mails/admin.py

+18 −3

Changes for mails/admin.py: 18 added lines, 3 removed lines.
Original line number	Diff line number	Diff line
		from functools import partial

		from django.contrib import admin, messages
		from django.db import transaction
		from django_no_queryset_admin_actions import (
		NoQuerySetAdminActionsMixin,
		no_queryset_action,
		@@ -6,7 +9,7 @@ from django_no_queryset_admin_actions import (
		from solo.admin import SingletonModelAdmin

		from .models import Mail, MailServerConfig, MailStatus
		from .tasks import send_mail_batch_task
		from .tasks import requeue_stale_processing_mails, send_mail_batch_task


		class MailAdmin(NoQuerySetAdminActionsMixin):
		@@ -23,11 +26,23 @@ class MailAdmin(NoQuerySetAdminActionsMixin):

		@no_queryset_action(description="Ponów nieudane", permissions=["manage_mails"])
		def retry_failed(self, request):
		send_mail_batch_task.enqueue(base_status=MailStatus.ERROR)
		with transaction.atomic():
		stale_count = requeue_stale_processing_mails()
		error_count = Mail.objects.filter(status=MailStatus.ERROR).update(
		status=MailStatus.PENDING,
		error_message=None,
		)
		retried_count = stale_count + error_count
		if retried_count:
		transaction.on_commit(partial(send_mail_batch_task.enqueue))

		if retried_count:
		messages.info(
		request,
		"Rozpoczęto wysyłanie ponowne wysyłanie niedostarczonych wiadomości.",
		f"Rozpoczęto ponowne wysyłanie {retried_count} wiadomości.",
		)
		else:
		messages.info(request, "Brak nieudanych wiadomości do ponowienia.")

		def has_add_permission(self, request):
		return False

mails/tasks.py

+78 −36

Changes for mails/tasks.py: 78 added lines, 36 removed lines.
Original line number	Diff line number	Diff line
		import logging
		from datetime import timedelta
		from functools import partial

		from django.core import mail
		from django.core.mail import EmailMultiAlternatives
		from django.db import transaction
		from django.db.models import Q
		from django.tasks import task
		from django.template.loader import render_to_string
		from django.utils import timezone
		@@ -10,15 +13,33 @@ from django.utils import timezone
		from mails.models import Mail, MailServerConfig, MailStatus
		from voters.models import Voter

		logger = logging.getLogger(__name__)
		PROCESSING_LEASE_TIMEOUT = timedelta(minutes=15)


		def requeue_stale_processing_mails(now=None):
		stale_before = (now or timezone.now()) - PROCESSING_LEASE_TIMEOUT
		return (
		Mail.objects.filter(status=MailStatus.PROCESSING)
		.filter(Q(last_try_date__isnull=True) \| Q(last_try_date__lt=stale_before))
		.update(status=MailStatus.PENDING, error_message=None)
		)


		@task(queue_name="email_batch")
		def send_mail_batch_task(base_status=MailStatus.PENDING):
		def send_mail_batch_task(base_status=None):
		if base_status not in (None, MailStatus.PENDING):
		logger.warning("Pominięto nieaktualny job wysyłki maili")
		return

		config = MailServerConfig.get_solo()

		with transaction.atomic():
		requeue_stale_processing_mails()
		ids_to_update = list(
		Mail.objects.filter(status=base_status)
		.select_for_update(skip_locked=True)[
		: MailServerConfig.get_solo().batch_size
		]
		Mail.objects.filter(status=MailStatus.PENDING)
		.select_for_update(skip_locked=True)
		.order_by("pk")[: config.batch_size]
		.values_list("pk", flat=True)
		)

		@@ -29,67 +50,88 @@ def send_mail_batch_task(base_status=MailStatus.PENDING):
		status=MailStatus.PROCESSING, last_try_date=timezone.now()
		)

		batch = Mail.objects.filter(pk__in=ids_to_update)[
		: MailServerConfig.get_solo().batch_size
		]

		connection = mail.get_connection()
		batch = (
		Mail.objects.filter(pk__in=ids_to_update).select_related("voter").order_by("pk")
		)

		connection = None
		try:
		connection = mail.get_connection()
		connection.open()
		except Exception:
		logger.exception("Nie udało się nawiązać połączenia z serwerem SMTP")
		Mail.objects.filter(pk__in=ids_to_update).update(
		status=MailStatus.ERROR,
		error_message="Nie udało się nawiązać połączenia z serwerem SMTP",
		)
		if Mail.objects.filter(status=base_status).exists():
		send_mail_batch_task.enqueue(base_status=base_status)
		if connection is not None:
		try:
		connection.close()
		except Exception:
		logger.exception("Nie udało się zamknąć połączenia z serwerem SMTP")
		return

		emails_to_send = []
		continue_chain = True
		try:
		for queued_email in batch:
		context = {"voter": queued_email.voter}
		if queued_email.voter.token is None:
		queued_email.status = MailStatus.ERROR
		queued_email.error_message = "Wyborca nie ma wygenerowanego tokenu"
		queued_email.save(update_fields=["status", "error_message"])
		continue

		try:
		context = {"voter": queued_email.voter}
		html_content = render_to_string("mails/voting_link_mail.html", context)
		text_content = render_to_string("mails/voting_link_mail.txt", context=context)

		text_content = render_to_string(
		"mails/voting_link_mail.txt", context=context
		)
		email = EmailMultiAlternatives(
		subject="Dostęp do Samorządowego Systemu Wyborczego",
		body=text_content,
		to=[queued_email.voter.email],
		connection=connection,
		)

		email.attach_alternative(html_content, "text/html")

		emails_to_send.append((queued_email, email))

		for queued_email, email in emails_to_send:
		if queued_email.voter.token is None:
		queued_email.status = MailStatus.ERROR
		queued_email.error_message = "Wyborca nie ma wygenerowanego tokenu"
		queued_email.save()
		continue

		try:
		sending_status = email.send(fail_silently=False)
		except Exception as exc:
		logger.exception(
		"Nie udało się przygotować lub wysłać wiadomości %s",
		queued_email.pk,
		)
		queued_email.status = MailStatus.ERROR
		queued_email.error_message = str(exc)
		else:
		if sending_status:
		queued_email.status = MailStatus.SENT
		queued_email.error_message = None
		queued_email.voter.last_email_date = timezone.now()
		queued_email.voter.save()
		queued_email.voter.save(update_fields=["last_email_date"])
		else:
		queued_email.status = MailStatus.ERROR
		except Exception as e:
		queued_email.status = MailStatus.ERROR
		queued_email.error_message = e

		queued_email.save()
		queued_email.error_message = (
		"Serwer SMTP nie potwierdził wysłania wiadomości"
		)

		queued_email.save(update_fields=["status", "error_message"])
		except Exception:
		continue_chain = False
		logger.exception("Nieoczekiwany błąd podczas przetwarzania pakietu maili")
		Mail.objects.filter(
		pk__in=ids_to_update,
		status=MailStatus.PROCESSING,
		).update(
		status=MailStatus.ERROR,
		error_message="Nieoczekiwany błąd podczas przetwarzania wiadomości",
		)
		finally:
		try:
		connection.close()
		except Exception:
		logger.exception("Nie udało się zamknąć połączenia z serwerem SMTP")

		if Mail.objects.filter(status=MailStatus.PENDING).exists():
		send_mail_batch_task.enqueue(base_status=base_status)
		if continue_chain and Mail.objects.filter(status=MailStatus.PENDING).exists():
		send_mail_batch_task.enqueue()


		@task