Merge pull request #1492 from GSA/notify-api-1487

remove tasks from redis after they complete

.github/workflows/restage-apps.yml

@@ -19,18 +19,18 @@ jobs:
         app: ["api", "admin"]
     steps:
       - name: Restage ${{matrix.app}}
-        uses: 18f/cg-deploy-action@main
+        uses: cloud-gov/cg-cli-tools@main
         with:
           cf_username: ${{ secrets.CLOUDGOV_USERNAME }}
           cf_password: ${{ secrets.CLOUDGOV_PASSWORD }}
           cf_org: gsa-tts-benefits-studio
           cf_space: notify-${{ inputs.environment }}
-          full_command: "cf restage --strategy rolling notify-${{matrix.app}}-${{inputs.environment}}"
+          command: "cf restage --strategy rolling notify-${{matrix.app}}-${{inputs.environment}}"
       - name: Restage ${{matrix.app}} egress
-        uses: 18f/cg-deploy-action@main
+        uses: cloud-gov/cg-cli-tools@main
         with:
           cf_username: ${{ secrets.CLOUDGOV_USERNAME }}
           cf_password: ${{ secrets.CLOUDGOV_PASSWORD }}
           cf_org: gsa-tts-benefits-studio
           cf_space: notify-${{ inputs.environment }}-egress
-          full_command: "cf restage --strategy rolling egress-proxy-notify-${{matrix.app}}-${{inputs.environment}}"
+          command: "cf restage --strategy rolling egress-proxy-notify-${{matrix.app}}-${{inputs.environment}}"

app/__init__.py

@@ -1,3 +1,4 @@
+import logging as real_logging
 import os
 import secrets
 import string
@@ -36,6 +37,9 @@ class NotifyCelery(Celery):
 
         # Configure Celery app with options from the main app config.
         self.config_from_object(app.config["CELERY"])
+        self.conf.worker_hijack_root_logger = False
+        logger = real_logging.getLogger("celery")
+        logger.propagate = False
 
     def send_task(self, name, args=None, kwargs=None, **other_kwargs):
         other_kwargs["headers"] = other_kwargs.get("headers") or {}

app/aws/s3.py

@@ -402,7 +402,12 @@ def extract_phones(job):
     phone_index = 0
     for item in first_row:
         # Note: may contain a BOM and look like \ufeffphone number
-        if item.lower() in ["phone number", "\\ufeffphone number"]:
+        if item.lower() in [
+            "phone number",
+            "\\ufeffphone number",
+            "\\ufeffphone number\n",
+            "phone number\n",
+        ]:
             break
         phone_index = phone_index + 1
 

app/celery/process_ses_receipts_tasks.py

@@ -1,6 +1,7 @@
 from datetime import timedelta
 
 import iso8601
+from celery.exceptions import Retry
 from flask import current_app, json
 from sqlalchemy.orm.exc import NoResultFound
 
@@ -11,7 +12,7 @@ from app.celery.service_callback_tasks import (
     send_complaint_to_service,
     send_delivery_status_to_service,
 )
-from app.config import QueueNames
+from app.config import Config, QueueNames
 from app.dao import notifications_dao
 from app.dao.complaint_dao import save_complaint
 from app.dao.notifications_dao import dao_get_notification_history_by_reference
@@ -25,12 +26,7 @@ from app.utils import utc_now
 
 
 @notify_celery.task(
-    bind=True,
-    name="process-ses-result",
-    autoretry_for=(Exception,),
-    # throws=(Exception,),  # Been attempted, did nothing.
-    max_retries=5,
-    default_retry_delay=300,
+    bind=True, name="process-ses-result", max_retries=5, default_retry_delay=300
 )
 def process_ses_results(self, response):
     try:
@@ -69,7 +65,9 @@ def process_ses_results(self, response):
                     f"Callback may have arrived before notification was"
                     f"persisted to the DB. Adding task to retry queue"
                 )
-                raise
+                self.retry(
+                    queue=QueueNames.RETRY, expires=Config.DEFAULT_REDIS_EXPIRE_TIME
+                )
             else:
                 current_app.logger.warning(
                     f"Notification not found for reference: {reference} "
@@ -99,26 +97,28 @@ def process_ses_results(self, response):
 
         if not aws_response_dict["success"]:
             current_app.logger.info(
-                f"SES delivery failed: notification id {notification.id} and reference "
-                f"{reference} has error found. Status {aws_response_dict['message']}"
+                "SES delivery failed: notification id {} and reference {} has error found. Status {}".format(
+                    notification.id, reference, aws_response_dict["message"]
+                )
             )
         else:
             current_app.logger.info(
-                f"SES callback return status of {notification_status} "
-                f"for notification: {notification.id}"
+                "SES callback return status of {} for notification: {}".format(
+                    notification_status, notification.id
+                )
             )
 
         check_and_queue_callback_task(notification)
 
         return True
 
-    except Exception as e:
-        print("Exception REACHED")
-        print(type(e))
-        print(e)
-        current_app.logger.exception("Error processing SES results")
+    except Retry:
         raise
 
+    except Exception:
+        current_app.logger.exception("Error processing SES results")
+        self.retry(queue=QueueNames.RETRY, expires=Config.DEFAULT_REDIS_EXPIRE_TIME)
+
 
 def determine_notification_bounce_type(ses_message):
     notification_type = ses_message["notificationType"]

app/celery/provider_tasks.py

@@ -1,8 +1,6 @@
 import json
-import logging
 import os
 from datetime import timedelta
-from functools import wraps
 
 from botocore.exceptions import ClientError
 from flask import current_app
@@ -12,7 +10,7 @@ from app import aws_cloudwatch_client, notify_celery, redis_store
 from app.clients.email import EmailClientNonRetryableException
 from app.clients.email.aws_ses import AwsSesClientThrottlingSendRateException
 from app.clients.sms import SmsClientResponseException
-from app.config import QueueNames
+from app.config import Config, QueueNames
 from app.dao import notifications_dao
 from app.dao.notifications_dao import (
     sanitize_successful_notification_by_id,
@@ -33,7 +31,6 @@ DELIVERY_RECEIPT_DELAY_IN_SECONDS = 30
     name="check_sms_delivery_receipt",
     max_retries=48,
     default_retry_delay=300,
-    autoretry_for=(NotificationTechnicalFailureException, ClientError),
 )
 def check_sms_delivery_receipt(self, message_id, notification_id, sent_at):
     """
@@ -55,7 +52,7 @@ def check_sms_delivery_receipt(self, message_id, notification_id, sent_at):
             status, provider_response, carrier = aws_cloudwatch_client.check_sms(
                 message_id, notification_id, sent_at
             )
-        except NotificationTechnicalFailureException:
+        except NotificationTechnicalFailureException as ntfe:
             provider_response = "Unable to find carrier response -- still looking"
             status = "pending"
             carrier = ""
@@ -65,7 +62,7 @@ def check_sms_delivery_receipt(self, message_id, notification_id, sent_at):
                 carrier=carrier,
                 provider_response=provider_response,
             )
-            raise
+            raise self.retry(exc=ntfe)
         except ClientError as err:
             # Probably a ThrottlingException but could be something else
             error_code = err.response["Error"]["Code"]
@@ -80,7 +77,7 @@ def check_sms_delivery_receipt(self, message_id, notification_id, sent_at):
                 carrier=carrier,
                 provider_response=provider_response,
             )
-            raise
+            raise self.retry(exc=err)
 
     if status == "success":
         status = NotificationStatus.DELIVERED
@@ -107,40 +104,8 @@ def check_sms_delivery_receipt(self, message_id, notification_id, sent_at):
         )
 
 
-def _deliver_sms_task_handler(cls):
-    """Handle the max retries exceeded error case for delivering sms notifications."""
-
-    func = cls.__call__
-
-    @wraps(func)
-    def deliver_sms_task_wrapper(self, notification_id):
-        try:
-            return func(self, notification_id)
-        except self.MaxRetriesExceededError:
-            message = (
-                "RETRY FAILED: Max retries reached. The task send_sms_to_provider failed for notification {}. "
-                "Notification has been updated to technical-failure".format(
-                    notification_id
-                )
-            )
-            update_notification_status_by_id(
-                notification_id,
-                NotificationStatus.TECHNICAL_FAILURE,
-            )
-            raise NotificationTechnicalFailureException(message)
-
-    cls.__call__ = deliver_sms_task_wrapper
-
-    return cls
-
-
-@_deliver_sms_task_handler
 @notify_celery.task(
-    bind=True,
-    name="deliver_sms",
-    max_retries=48,
-    default_retry_delay=300,
-    autoretry_for=(Exception,),
+    bind=True, name="deliver_sms", max_retries=48, default_retry_delay=300
 )
 def deliver_sms(self, notification_id):
     """Branch off to the final step in delivering the notification to sns and get delivery receipts."""
@@ -163,6 +128,8 @@ def deliver_sms(self, notification_id):
             )
         # Code branches off to send_to_providers.py
         message_id = send_to_providers.send_sms_to_provider(notification)
+
+        # DEPRECATED
         # We have to put it in UTC.  For other timezones, the delay
         # will be ignored and it will fire immediately (although this probably only affects developer testing)
         my_eta = utc_now() + timedelta(seconds=DELIVERY_RECEIPT_DELAY_IN_SECONDS)
@@ -176,37 +143,32 @@ def deliver_sms(self, notification_id):
             notification_id,
             NotificationStatus.TEMPORARY_FAILURE,
         )
-
         if isinstance(e, SmsClientResponseException):
-            log_lvl = logging.WARNING
-            log_exc_info = False
+            current_app.logger.warning(
+                "SMS notification delivery for id: {} failed".format(notification_id),
+            )
         else:
-            log_lvl = logging.ERROR
-            log_exc_info = True
+            current_app.logger.exception(
+                "SMS notification delivery for id: {} failed".format(notification_id),
+            )
 
-        current_app.logger.log(
-            level=log_lvl,
-            msg=f"SMS notification delivery for id: {notification_id} failed",
-            exc_info=log_exc_info,
-        )
-
-        raise
-
-
-def _deliver_email_task_handler(cls):
-    """Handle the max retries exceeded error case for delivering email notifications."""
-
-    func = cls.__call__
-
-    @wraps(func)
-    def deliver_email_task_wrapper(self, notification_id):
         try:
-            return func(self, notification_id)
+            if self.request.retries == 0:
+                self.retry(
+                    queue=QueueNames.RETRY,
+                    countdown=0,
+                    expires=Config.DEFAULT_REDIS_EXPIRE_TIME,
+                )
+            else:
+                self.retry(
+                    queue=QueueNames.RETRY, expires=Config.DEFAULT_REDIS_EXPIRE_TIME
+                )
         except self.MaxRetriesExceededError:
             message = (
-                "RETRY FAILED: Max retries reached. "
-                f"The task send_email_to_provider failed for notification {notification_id}. "
-                "Notification has been updated to technical-failure"
+                "RETRY FAILED: Max retries reached. The task send_sms_to_provider failed for notification {}. "
+                "Notification has been updated to technical-failure".format(
+                    notification_id
+                )
             )
             update_notification_status_by_id(
                 notification_id,
@@ -214,19 +176,9 @@ def _deliver_email_task_handler(cls):
             )
             raise NotificationTechnicalFailureException(message)
 
-    cls.__call__ = deliver_email_task_wrapper
 
-    return cls
-
-
-@_deliver_email_task_handler
 @notify_celery.task(
-    bind=True,
-    name="deliver_email",
-    max_retries=48,
-    default_retry_delay=30,
-    autoretry_for=(Exception,),
-    dont_autoretry_for=(EmailClientNonRetryableException,),
+    bind=True, name="deliver_email", max_retries=48, default_retry_delay=30
 )
 def deliver_email(self, notification_id):
     try:
@@ -247,19 +199,29 @@ def deliver_email(self, notification_id):
         send_to_providers.send_email_to_provider(notification)
     except EmailClientNonRetryableException:
         current_app.logger.exception(f"Email notification {notification_id} failed")
-        update_notification_status_by_id(
-            notification_id,
-            NotificationStatus.TECHNICAL_FAILURE,
-        )
-        raise
+        update_notification_status_by_id(notification_id, "technical-failure")
     except Exception as e:
-        if isinstance(e, AwsSesClientThrottlingSendRateException):
-            current_app.logger.warning(
-                f"RETRY: Email notification {notification_id} was rate limited by SES"
-            )
-        else:
-            current_app.logger.exception(
-                f"RETRY: Email notification {notification_id} failed"
-            )
+        try:
+            if isinstance(e, AwsSesClientThrottlingSendRateException):
+                current_app.logger.warning(
+                    f"RETRY: Email notification {notification_id} was rate limited by SES"
+                )
+            else:
+                current_app.logger.exception(
+                    f"RETRY: Email notification {notification_id} failed"
+                )
 
-        raise
+            self.retry(queue=QueueNames.RETRY, expires=Config.DEFAULT_REDIS_EXPIRE_TIME)
+        except self.MaxRetriesExceededError:
+            message = (
+                "RETRY FAILED: Max retries reached. "
+                "The task send_email_to_provider failed for notification {}. "
+                "Notification has been updated to technical-failure".format(
+                    notification_id
+                )
+            )
+            update_notification_status_by_id(
+                notification_id,
+                NotificationStatus.TECHNICAL_FAILURE,
+            )
+            raise NotificationTechnicalFailureException(message)

app/celery/service_callback_tasks.py

@@ -1,68 +1,15 @@
 import json
-from functools import wraps
-from inspect import signature
 
 from flask import current_app
 from requests import HTTPError, RequestException, request
 
 from app import encryption, notify_celery
+from app.config import QueueNames
 from app.utils import DATETIME_FORMAT
 
 
-def _send_to_service_task_handler(cls):
-
-    func = cls.__call__
-
-    @wraps(func)
-    def send_to_service_task_wrapper(*args, **kwargs):
-        sig = signature(func)
-        bargs = sig.bind(*args, **kwargs)
-        bargs.apply_defaults()
-
-        function_name = func.__name__
-
-        if function_name == "send_delivery_status_to_service":
-            encrypted_status_update = bargs.arguments["encrypted_status_update"]
-
-            status_update = encryption.decrypt(encrypted_status_update)
-            service_callback_url = status_update["service_callback_api_url"]
-
-            notification_id = bargs.arguments["notification_id"]
-
-        elif function_name == "send_complaint_to_service":
-            complaint_data = bargs.arguments["complaint_data"]
-
-            notification_id = complaint_data["notification_id"]
-            service_callback_url = complaint_data["service_callback_api_url"]
-
-        else:
-            raise ValueError(
-                f"Incorrect send to service function name found: {function_name}"
-            )
-
-        self_ = bargs.arguments["self"]
-
-        try:
-            return func(*args, **kwargs)
-        except self_.MaxRetriesExceededError:
-            current_app.logger.warning(
-                f"Retry: {function_name} has retried the max num of times for callback url "
-                f"{service_callback_url} and notification_id: {notification_id}"
-            )
-            raise
-
-    cls.__call__ = send_to_service_task_wrapper
-
-    return cls
-
-
-@_send_to_service_task_handler
 @notify_celery.task(
-    bind=True,
-    name="send-delivery-status",
-    max_retries=5,
-    default_retry_delay=300,
-    autoretry_for=(HTTPError,),
+    bind=True, name="send-delivery-status", max_retries=5, default_retry_delay=300
 )
 def send_delivery_status_to_service(self, notification_id, encrypted_status_update):
     status_update = encryption.decrypt(encrypted_status_update)
@@ -89,13 +36,8 @@ def send_delivery_status_to_service(self, notification_id, encrypted_status_upda
     )
 
 
-@_send_to_service_task_handler
 @notify_celery.task(
-    bind=True,
-    name="send-complaint",
-    max_retries=5,
-    default_retry_delay=300,
-    autoretry_for=(HTTPError,),
+    bind=True, name="send-complaint", max_retries=5, default_retry_delay=300
 )
 def send_complaint_to_service(self, complaint_data):
     complaint = encryption.decrypt(complaint_data)
@@ -130,29 +72,43 @@ def _send_data_to_service_callback_api(
             data=json.dumps(data),
             headers={
                 "Content-Type": "application/json",
-                "Authorization": f"Bearer {token}",
+                "Authorization": "Bearer {}".format(token),
             },
             timeout=5,
         )
         current_app.logger.info(
-            f"{function_name} sending {notification_id} to {service_callback_url}, response {response.status_code}"
+            "{} sending {} to {}, response {}".format(
+                function_name,
+                notification_id,
+                service_callback_url,
+                response.status_code,
+            )
         )
         response.raise_for_status()
     except RequestException as e:
         current_app.logger.warning(
-            f"{function_name} request failed for notification_id: {notification_id} and "
-            f"url: {service_callback_url}. exception: {e}"
+            "{} request failed for notification_id: {} and url: {}. exception: {}".format(
+                function_name, notification_id, service_callback_url, e
+            )
         )
         if (
             not isinstance(e, HTTPError)
             or e.response.status_code >= 500
             or e.response.status_code == 429
         ):
-            raise
+            try:
+                self.retry(queue=QueueNames.CALLBACKS_RETRY)
+            except self.MaxRetriesExceededError:
+                current_app.logger.warning(
+                    "Retry: {} has retried the max num of times for callback url {} and notification_id: {}".format(
+                        function_name, service_callback_url, notification_id
+                    )
+                )
         else:
             current_app.logger.warning(
-                f"{function_name} callback is not being retried for notification_id: "
-                f"{notification_id} and url: {service_callback_url}. exception: {e}"
+                "{} callback is not being retried for notification_id: {} and url: {}. exception: {}".format(
+                    function_name, notification_id, service_callback_url, e
+                )
             )
 
 

app/celery/tasks.py

@@ -1,7 +1,6 @@
 import json
-from functools import wraps
-from inspect import signature
 
+from celery.signals import task_postrun
 from flask import current_app
 from requests import HTTPError, RequestException, request
 from sqlalchemy.exc import IntegrityError, SQLAlchemyError
@@ -9,7 +8,7 @@ from sqlalchemy.exc import IntegrityError, SQLAlchemyError
 from app import create_uuid, encryption, notify_celery
 from app.aws import s3
 from app.celery import provider_tasks
-from app.config import QueueNames
+from app.config import Config, QueueNames
 from app.dao.inbound_sms_dao import dao_get_inbound_sms_by_id
 from app.dao.jobs_dao import dao_get_job_by_id, dao_update_job
 from app.dao.notifications_dao import (
@@ -22,7 +21,10 @@ from app.dao.service_sms_sender_dao import dao_get_service_sms_senders_by_id
 from app.dao.templates_dao import dao_get_template_by_id
 from app.enums import JobStatus, KeyType, NotificationType
 from app.errors import TotalRequestsError
-from app.notifications.process_notifications import persist_notification
+from app.notifications.process_notifications import (
+    get_notification,
+    persist_notification,
+)
 from app.notifications.validators import check_service_over_total_message_limit
 from app.serialised_models import SerialisedService, SerialisedTemplate
 from app.service.utils import service_allowed_to_send_to
@@ -145,6 +147,7 @@ def process_row(row, template, job, service, sender_id=None):
         ),
         task_kwargs,
         queue=QueueNames.DATABASE,
+        expires=Config.DEFAULT_REDIS_EXPIRE_TIME,
     )
     return notification_id
 
@@ -168,38 +171,14 @@ def __total_sending_limits_for_job_exceeded(service, job, job_id):
         return True
 
 
-def _save_task_hander(func):
-    @wraps(func)
-    def save_task_wrapper(*args, **kwargs):
-        sig = signature(func)
-        bargs = sig.bind(*args, **kwargs)
-        bargs.apply_defaults()
-
-        task = bargs.arguments["self"]
-        notification_id = bargs.arguments["notification_id"]
-        notification = encryption.decrypt(bargs.arguments["encrypted_notification"])
-
-        try:
-            return func(*args, **kwargs)
-        except task.MaxRetriesExceededError:
-            retry_msg = (
-                f"{task.__name__} notification for job {notification.get("job", None)} "
-            )
-            f"row number {notification.get("row_number", None)} and notification id {notification_id}"
-            current_app.logger.exception("Max retry failed" + retry_msg)
-            raise
-
-    return save_task_wrapper
+@task_postrun.connect
+def log_task_ejection(sender=None, task_id=None, **kwargs):
+    current_app.logger.info(
+        f"Task {task_id} ({sender.name if sender else 'unknown_task'}) has been completed and removed"
+    )
 
 
-@_save_task_hander
-@notify_celery.task(
-    bind=True,
-    name="save-sms",
-    max_retries=5,
-    default_retry_delay=300,
-    autoretry_for=(SQLAlchemyError,),
-)
+@notify_celery.task(bind=True, name="save-sms", max_retries=2, default_retry_delay=600)
 def save_sms(self, service_id, notification_id, encrypted_notification, sender_id=None):
     """Persist notification to db and place notification in queue to send to sns."""
     notification = encryption.decrypt(encrypted_notification)
@@ -227,7 +206,9 @@ def save_sms(self, service_id, notification_id, encrypted_notification, sender_i
                 f"service not allowed to send for job_id {notification.get('job', None)}, aborting"
             )
         )
-        current_app.logger.debug(f"SMS {notification_id} failed as restricted service")
+        current_app.logger.debug(
+            "SMS {} failed as restricted service".format(notification_id)
+        )
         return
 
     try:
@@ -266,27 +247,19 @@ def save_sms(self, service_id, notification_id, encrypted_notification, sender_i
         )
 
         current_app.logger.debug(
-            f"SMS {saved_notification.id} created at {saved_notification.created_at} for job "
-            f"{notification.get('job', None)}"
+            "SMS {} created at {} for job {}".format(
+                saved_notification.id,
+                saved_notification.created_at,
+                notification.get("job", None),
+            )
         )
 
-    except SQLAlchemyError:
-        if not get_notification_by_id(notification_id):
-            retry_msg = (
-                f"{self.__name__} notification for job {notification.get("job", None)} "
-            )
-            f"row number {notification.get("row_number", None)} and notification id {notification_id}"
-            current_app.logger.exception(retry_msg)
-            raise
+    except SQLAlchemyError as e:
+        handle_exception(self, notification, notification_id, e)
 
 
-@_save_task_hander
 @notify_celery.task(
-    bind=True,
-    name="save-email",
-    max_retries=5,
-    default_retry_delay=300,
-    autoretry_for=(SQLAlchemyError,),
+    bind=True, name="save-email", max_retries=5, default_retry_delay=300
 )
 def save_email(
     self, service_id, notification_id, encrypted_notification, sender_id=None
@@ -306,9 +279,11 @@ def save_email(
         reply_to_text = template.reply_to_text
 
     if not service_allowed_to_send_to(notification["to"], service, KeyType.NORMAL):
-        current_app.logger.info(f"Email {notification_id} failed as restricted service")
+        current_app.logger.info(
+            "Email {} failed as restricted service".format(notification_id)
+        )
         return
-
+    original_notification = get_notification(notification_id)
     try:
         saved_notification = persist_notification(
             template_id=notification["template"],
@@ -325,63 +300,30 @@ def save_email(
             notification_id=notification_id,
             reply_to_text=reply_to_text,
         )
-
-        provider_tasks.deliver_email.apply_async(
-            [str(saved_notification.id)], queue=QueueNames.SEND_EMAIL
-        )
+        # we only want to send once
+        if original_notification is None:
+            provider_tasks.deliver_email.apply_async(
+                [str(saved_notification.id)], queue=QueueNames.SEND_EMAIL
+            )
 
         current_app.logger.debug(
-            f"Email {saved_notification.id} created at {saved_notification.created_at}"
+            "Email {} created at {}".format(
+                saved_notification.id, saved_notification.created_at
+            )
         )
-    except SQLAlchemyError:
-        if not get_notification_by_id(notification_id):
-            retry_msg = (
-                f"{self.__name__} notification for job {notification.get("job", None)} "
-                f"row number {notification.get("row_number", None)} and notification id {notification_id}"
-            )
-            current_app.logger.exception(retry_msg)
-            raise
+    except SQLAlchemyError as e:
+        handle_exception(self, notification, notification_id, e)
 
 
-def _save_api_task_handler(func):
-    @wraps(func)
-    def save_api_task_wrapper(*args, **kwargs):
-        sig = signature(func)
-        bargs = sig.bind(*args, **kwargs)
-        bargs.apply_defaults()
-
-        self_ = bargs.argument["self"]
-        notification = encryption.decrypt[bargs.arguments["encrypted_notification"]]
-        try:
-            return func(*args, **kwargs)
-        except self_.MaxRetriesExceededError:
-            current_app.logger.exception(
-                f"Max retry failed Failed to persist notification {notification['id']}",
-            )
-            raise
-
-    return save_api_task_wrapper
-
-
-@_save_api_task_handler
 @notify_celery.task(
-    bind=True,
-    name="save-api-email",
-    max_retries=5,
-    default_retry_delay=300,
-    autoretry_for=(SQLAlchemyError,),
+    bind=True, name="save-api-email", max_retries=5, default_retry_delay=300
 )
 def save_api_email(self, encrypted_notification):
     save_api_email_or_sms(self, encrypted_notification)
 
 
-@_save_api_task_handler
 @notify_celery.task(
-    bind=True,
-    name="save-api-sms",
-    max_retries=5,
-    default_retry_delay=300,
-    autoretry_for=(SQLAlchemyError,),
+    bind=True, name="save-api-sms", max_retries=2, default_retry_delay=600
 )
 def save_api_sms(self, encrypted_notification):
     save_api_email_or_sms(self, encrypted_notification)
@@ -400,6 +342,8 @@ def save_api_email_or_sms(self, encrypted_notification):
         if notification["notification_type"] == NotificationType.EMAIL
         else provider_tasks.deliver_sms
     )
+
+    original_notification = get_notification(notification["id"])
     try:
         persist_notification(
             notification_id=notification["id"],
@@ -418,10 +362,11 @@ def save_api_email_or_sms(self, encrypted_notification):
             document_download_count=notification["document_download_count"],
         )
         # Only get here if save to the db was successful (i.e. first time)
-        provider_task.apply_async([notification["id"]], queue=q)
-        current_app.logger.debug(
-            f"{notification['notification_type']} {notification['id']} has been persisted and sent to delivery queue."
-        )
+        if original_notification is None:
+            provider_task.apply_async([notification["id"]], queue=q)
+            current_app.logger.debug(
+                f"{notification['id']} has been persisted and sent to delivery queue."
+            )
 
     except IntegrityError:
         current_app.logger.warning(
@@ -431,29 +376,40 @@ def save_api_email_or_sms(self, encrypted_notification):
         # up retrying because IntegrityError is a subclass of SQLAlchemyError
         return
 
-
-def _send_inbound_sms_to_service_handler(func):
-    @wraps(func)
-    def send_inbound_sms_to_service_wrapper(self, inbound_sms_id, service_id):
+    except SQLAlchemyError:
         try:
-            return func(self, inbound_sms_id, service_id)
+            self.retry(queue=QueueNames.RETRY, expires=Config.DEFAULT_REDIS_EXPIRE_TIME)
         except self.MaxRetriesExceededError:
             current_app.logger.exception(
-                "Retry: send_inbound_sms_to_service has retried the max number of"
-                + f"times for service: {service_id} and inbound_sms {inbound_sms_id}"
+                f"Max retry failed Failed to persist notification {notification['id']}",
             )
-            raise
-
-    return send_inbound_sms_to_service_wrapper
 
 
-@_send_inbound_sms_to_service_handler
+def handle_exception(task, notification, notification_id, exc):
+    if not get_notification_by_id(notification_id):
+        retry_msg = "{task} notification for job {job} row number {row} and notification id {noti}".format(
+            task=task.__name__,
+            job=notification.get("job", None),
+            row=notification.get("row_number", None),
+            noti=notification_id,
+        )
+        # Sometimes, SQS plays the same message twice. We should be able to catch an IntegrityError, but it seems
+        # SQLAlchemy is throwing a FlushError. So we check if the notification id already exists then do not
+        # send to the retry queue.
+        # This probably (hopefully) is not an issue with Redis as the celery backing store
+        current_app.logger.exception("Retry" + retry_msg)
+        try:
+            task.retry(
+                queue=QueueNames.RETRY,
+                exc=exc,
+                expires=Config.DEFAULT_REDIS_EXPIRE_TIME,
+            )
+        except task.MaxRetriesExceededError:
+            current_app.logger.exception("Max retry failed" + retry_msg)
+
+
 @notify_celery.task(
-    bind=True,
-    name="send-inbound-sms",
-    max_retries=5,
-    default_retry_delay=300,
-    autoretry_for=(RequestException,),
+    bind=True, name="send-inbound-sms", max_retries=5, default_retry_delay=300
 )
 def send_inbound_sms_to_service(self, inbound_sms_id, service_id):
     inbound_api = get_service_inbound_api_for_service(service_id=service_id)
@@ -495,7 +451,15 @@ def send_inbound_sms_to_service(self, inbound_sms_id, service_id):
             + f"and url: {inbound_api.url}. exception: {e}"
         )
         if not isinstance(e, HTTPError) or e.response.status_code >= 500:
-            raise
+            try:
+                self.retry(
+                    queue=QueueNames.RETRY, expires=Config.DEFAULT_REDIS_EXPIRE_TIME
+                )
+            except self.MaxRetriesExceededError:
+                current_app.logger.exception(
+                    "Retry: send_inbound_sms_to_service has retried the max number of"
+                    + f"times for service: {service_id} and inbound_sms {inbound_sms_id}"
+                )
         else:
             current_app.logger.warning(
                 f"send_inbound_sms_to_service is not being retried for service_id: {service_id} for "

app/config.py

@@ -53,6 +53,7 @@ class TaskNames(object):
 
 class Config(object):
     NOTIFY_APP_NAME = "api"
+    DEFAULT_REDIS_EXPIRE_TIME = 4 * 24 * 60 * 60
     NOTIFY_ENVIRONMENT = getenv("NOTIFY_ENVIRONMENT", "development")
     # URL of admin app
     ADMIN_BASE_URL = getenv("ADMIN_BASE_URL", "http://localhost:6012")
@@ -167,6 +168,8 @@ class Config(object):
 
     CELERY = {
         "worker_max_tasks_per_child": 500,
+        "task_ignore_result": True,
+        "result_persistent": False,
         "broker_url": REDIS_URL,
         "broker_transport_options": {
             "visibility_timeout": 310,

app/dao/notifications_dao.py

@@ -71,7 +71,12 @@ def dao_create_notification(notification):
     # notify-api-742 remove phone numbers from db
     notification.to = "1"
     notification.normalised_to = "1"
-    db.session.add(notification)
+
+    # notify-api-1454 insert only if it doesn't exist
+    stmt = select(Notification).where(Notification.id == notification.id)
+    result = db.session.execute(stmt).scalar()
+    if result is None:
+        db.session.add(notification)
 
 
 def country_records_delivery(phone_prefix):
@@ -105,6 +110,16 @@ def _update_notification_status(
     return notification
 
 
+def update_notification_message_id(notification_id, message_id):
+    stmt = (
+        update(Notification)
+        .where(Notification.id == notification_id)
+        .values(message_id=message_id)
+    )
+    db.session.execute(stmt)
+    db.session.commit()
+
+
 @autocommit
 def update_notification_status_by_id(
     notification_id, status, sent_by=None, provider_response=None, carrier=None

app/delivery/send_to_providers.py

@@ -16,7 +16,10 @@ from app import (
 from app.aws.s3 import get_personalisation_from_s3, get_phone_number_from_s3
 from app.celery.test_key_tasks import send_email_response, send_sms_response
 from app.dao.email_branding_dao import dao_get_email_branding_by_id
-from app.dao.notifications_dao import dao_update_notification
+from app.dao.notifications_dao import (
+    dao_update_notification,
+    update_notification_message_id,
+)
 from app.dao.provider_details_dao import get_provider_details_by_notification_type
 from app.dao.service_sms_sender_dao import dao_get_sms_senders_by_service_id
 from app.enums import BrandType, KeyType, NotificationStatus, NotificationType
@@ -117,6 +120,7 @@ def send_sms_to_provider(notification):
 
                 message_id = provider.send_sms(**send_sms_kwargs)
                 current_app.logger.info(f"got message_id {message_id}")
+                update_notification_message_id(notification.id, message_id)
             except Exception as e:
                 n = notification
                 msg = f"FAILED send to sms, job_id: {n.job_id} row_number {n.job_row_number} message_id {message_id}"

app/models.py

@@ -1532,6 +1532,7 @@ class Notification(db.Model):
 
     provider_response = db.Column(db.Text, nullable=True)
     carrier = db.Column(db.Text, nullable=True)
+    message_id = db.Column(db.Text, nullable=True)
 
     # queue_name = db.Column(db.Text, nullable=True)
 

app/notifications/process_notifications.py

@@ -8,6 +8,7 @@ from app.config import QueueNames
 from app.dao.notifications_dao import (
     dao_create_notification,
     dao_delete_notifications_by_id,
+    get_notification_by_id,
 )
 from app.enums import KeyType, NotificationStatus, NotificationType
 from app.errors import BadRequestError
@@ -53,6 +54,10 @@ def check_placeholders(template_object):
         raise BadRequestError(fields=[{"template": message}], message=message)
 
 
+def get_notification(notification_id):
+    return get_notification_by_id(notification_id)
+
+
 def persist_notification(
     *,
     template_id,

migrations/versions/0413_add_message_id.py

@@ -0,0 +1,28 @@
+"""
+
+Revision ID: 0413_add_message_id
+Revises: 412_remove_priority
+Create Date: 2023-12-11 11:35:22.873930
+
+"""
+
+import sqlalchemy as sa
+from alembic import op
+
+revision = "0413_add_message_id"
+down_revision = "0412_remove_priority"
+
+
+def upgrade():
+    op.add_column("notifications", sa.Column("message_id", sa.Text))
+    op.create_index(
+        "ix_notifications_message_id",
+        "notifications",
+        ["message_id"],
+        unique=False,
+    )
+
+
+def downgrade():
+    op.drop_index("ix_notifications_message_id", table_name="notifications")
+    op.drop_column("notifications", "message_id")

notifications_utils/logging.py

@@ -70,10 +70,12 @@ def init_app(app):
     for logger_instance, handler in product(loggers, handlers):
         logger_instance.addHandler(handler)
         logger_instance.setLevel(loglevel)
+        logger_instance.propagate = False
     warning_loggers = [logging.getLogger("boto3"), logging.getLogger("s3transfer")]
     for logger_instance, handler in product(warning_loggers, handlers):
         logger_instance.addHandler(handler)
         logger_instance.setLevel(logging.WARNING)
+        logger_instance.propagate = False
 
     # Suppress specific loggers to prevent leaking sensitive info
     logging.getLogger("boto3").setLevel(logging.ERROR)

tests/app/aws/test_s3.py

@@ -219,6 +219,20 @@ def test_get_s3_file_makes_correct_call(notify_api, mocker):
             2,
             "5555555552",
         ),
+        (
+            # simulate file saved with utf8withbom
+            "\\ufeffPHONE NUMBER\n",
+            "eee",
+            2,
+            "5555555552",
+        ),
+        (
+            # simulate file saved without utf8withbom
+            "\\PHONE NUMBER\n",
+            "eee",
+            2,
+            "5555555552",
+        ),
     ],
 )
 def test_get_phone_number_from_s3(

tests/app/celery/test_process_ses_receipts_tasks.py

@@ -1,7 +1,6 @@
 import json
 from unittest.mock import ANY
 
-import pytest
 from freezegun import freeze_time
 
 from app import encryption
@@ -158,10 +157,7 @@ def test_process_ses_results_retry_called(sample_email_template, mocker):
     mocked = mocker.patch(
         "app.celery.process_ses_receipts_tasks.process_ses_results.retry"
     )
-    with pytest.raises(Exception):  # noqa: B017
-        # In order to make this work, we have to suppress the flake8 warning about
-        # pytest.raises(Exception), which is usually considered a bad thing.
-        process_ses_results(response=ses_notification_callback(reference="ref1"))
+    process_ses_results(response=ses_notification_callback(reference="ref1"))
     assert mocked.call_count != 0
 
 
@@ -244,29 +240,17 @@ def test_ses_callback_should_not_update_notification_status_if_already_delivered
     assert mock_upd.call_count == 0
 
 
-def test_ses_callback_should_retry_if_notification_is_new(client, _notify_db, mocker):
-    # mock_retry = mocker.patch(
-    #     "app.celery.process_ses_receipts_tasks.process_ses_results.retry"
-    # )
+def test_ses_callback_should_retry_if_notification_is_new(mocker):
+    mock_retry = mocker.patch(
+        "app.celery.process_ses_receipts_tasks.process_ses_results.retry"
+    )
     mock_logger = mocker.patch(
-        "app.celery.process_ses_receipts_tasks.current_app.logger.exception"
+        "app.celery.process_ses_receipts_tasks.current_app.logger.error"
     )
     with freeze_time("2017-11-17T12:14:03.646Z"):
-        try:
-            assert (
-                process_ses_results(ses_notification_callback(reference="ref")) is None
-            )
-        except Exception as e:
-            import traceback
-            print(type(e))
-            print("*" * 80)
-            print(e)
-            print("-" * 80)
-            print(traceback.format_exc())
-            print("-" * 80)
-            raise
+        assert process_ses_results(ses_notification_callback(reference="ref")) is None
         assert mock_logger.call_count == 0
-        # assert mock_retry.call_count == 1
+        assert mock_retry.call_count == 1
 
 
 def test_ses_callback_should_log_if_notification_is_missing(client, _notify_db, mocker):

tests/app/celery/test_provider_tasks.py

@@ -1,4 +1,5 @@
 import json
+from unittest.mock import ANY
 
 import pytest
 from botocore.exceptions import ClientError
@@ -148,7 +149,7 @@ def test_should_add_to_retry_queue_if_notification_not_found_in_deliver_sms_task
     deliver_sms(notification_id)
     app.delivery.send_to_providers.send_sms_to_provider.assert_not_called()
     app.celery.provider_tasks.deliver_sms.retry.assert_called_with(
-        queue="retry-tasks", countdown=0
+        queue="retry-tasks", countdown=0, expires=ANY
     )
 
 
@@ -208,7 +209,7 @@ def test_should_go_into_technical_error_if_exceeds_retries_on_deliver_sms_task(
     assert str(sample_notification.id) in str(e.value)
 
     provider_tasks.deliver_sms.retry.assert_called_with(
-        queue="retry-tasks", countdown=0
+        queue="retry-tasks", countdown=0, expires=ANY
     )
 
     assert sample_notification.status == NotificationStatus.TEMPORARY_FAILURE
@@ -240,7 +241,7 @@ def test_should_add_to_retry_queue_if_notification_not_found_in_deliver_email_ta
     deliver_email(notification_id)
     app.delivery.send_to_providers.send_email_to_provider.assert_not_called()
     app.celery.provider_tasks.deliver_email.retry.assert_called_with(
-        queue="retry-tasks"
+        queue="retry-tasks", expires=ANY
     )
 
 
@@ -268,7 +269,9 @@ def test_should_go_into_technical_error_if_exceeds_retries_on_deliver_email_task
         deliver_email(sample_notification.id)
     assert str(sample_notification.id) in str(e.value)
 
-    provider_tasks.deliver_email.retry.assert_called_with(queue="retry-tasks")
+    provider_tasks.deliver_email.retry.assert_called_with(
+        queue="retry-tasks", expires=ANY
+    )
     assert sample_notification.status == NotificationStatus.TECHNICAL_FAILURE
 
 

tests/app/celery/test_scheduled_tasks.py

@@ -415,6 +415,7 @@ def test_check_for_missing_rows_in_completed_jobs_calls_save_email(
         ),
         {},
         queue="database-tasks",
+        expires=ANY,
     )
 
 

tests/app/celery/test_tasks.py

@@ -1,7 +1,7 @@
 import json
 import uuid
 from datetime import datetime, timedelta
-from unittest.mock import Mock, call
+from unittest.mock import ANY, Mock, call
 
 import pytest
 import requests_mock
@@ -115,6 +115,7 @@ def test_should_process_sms_job(sample_job, mocker):
         (str(sample_job.service_id), "uuid", "something_encrypted"),
         {},
         queue="database-tasks",
+        expires=ANY,
     )
     job = jobs_dao.dao_get_job_by_id(sample_job.id)
     assert job.job_status == JobStatus.FINISHED
@@ -135,6 +136,7 @@ def test_should_process_sms_job_with_sender_id(sample_job, mocker, fake_uuid):
         (str(sample_job.service_id), "uuid", "something_encrypted"),
         {"sender_id": fake_uuid},
         queue="database-tasks",
+        expires=ANY,
     )
 
 
@@ -179,6 +181,7 @@ def test_should_process_job_if_send_limits_are_not_exceeded(
         ),
         {},
         queue="database-tasks",
+        expires=ANY,
     )
 
 
@@ -237,6 +240,7 @@ def test_should_process_email_job(email_job_with_placeholders, mocker):
         ),
         {},
         queue="database-tasks",
+        expires=ANY,
     )
     job = jobs_dao.dao_get_job_by_id(email_job_with_placeholders.id)
     assert job.job_status == JobStatus.FINISHED
@@ -262,6 +266,7 @@ def test_should_process_email_job_with_sender_id(
         (str(email_job_with_placeholders.service_id), "uuid", "something_encrypted"),
         {"sender_id": fake_uuid},
         queue="database-tasks",
+        expires=ANY,
     )
 
 
@@ -351,6 +356,7 @@ def test_process_row_sends_letter_task(
         ),
         {},
         queue=expected_queue,
+        expires=ANY,
     )
 
 
@@ -387,6 +393,7 @@ def test_process_row_when_sender_id_is_provided(mocker, fake_uuid):
         ),
         {"sender_id": fake_uuid},
         queue="database-tasks",
+        expires=ANY,
     )
 
 
@@ -839,7 +846,9 @@ def test_save_sms_should_go_to_retry_queue_if_database_errors(sample_template, m
             encryption.encrypt(notification),
         )
     assert provider_tasks.deliver_sms.apply_async.called is False
-    tasks.save_sms.retry.assert_called_with(exc=expected_exception, queue="retry-tasks")
+    tasks.save_sms.retry.assert_called_with(
+        exc=expected_exception, queue="retry-tasks", expires=ANY
+    )
 
     assert _get_notification_query_count() == 0
 
@@ -868,7 +877,7 @@ def test_save_email_should_go_to_retry_queue_if_database_errors(
         )
     assert not provider_tasks.deliver_email.apply_async.called
     tasks.save_email.retry.assert_called_with(
-        exc=expected_exception, queue="retry-tasks"
+        exc=expected_exception, queue="retry-tasks", expires=ANY
     )
 
     assert _get_notification_query_count() == 0

tests/app/delivery/test_send_to_providers.py

@@ -93,6 +93,7 @@ def test_should_send_personalised_template_to_correct_sms_provider_and_persist(
     mock_s3 = mocker.patch("app.delivery.send_to_providers.get_phone_number_from_s3")
     mock_s3.return_value = "2028675309"
 
+    mocker.patch("app.delivery.send_to_providers.update_notification_message_id")
     mock_personalisation = mocker.patch(
         "app.delivery.send_to_providers.get_personalisation_from_s3"
     )
@@ -233,6 +234,7 @@ def test_send_sms_should_use_template_version_from_notification_not_latest(
     mock_s3 = mocker.patch("app.delivery.send_to_providers.get_phone_number_from_s3")
     mock_s3.return_value = "2028675309"
 
+    mocker.patch("app.delivery.send_to_providers.update_notification_message_id")
     mock_s3_p = mocker.patch(
         "app.delivery.send_to_providers.get_personalisation_from_s3"
     )
@@ -327,6 +329,7 @@ def test_should_send_sms_with_downgraded_content(notify_db_session, mocker):
     # ī, grapes, tabs, zero width space and ellipsis are not
     # ó isn't in GSM, but it is in the welsh alphabet so will still be sent
 
+    mocker.patch("app.delivery.send_to_providers.update_notification_message_id")
     mocker.patch("app.delivery.send_to_providers.redis_store", return_value=None)
     mocker.patch(
         "app.delivery.send_to_providers.get_sender_numbers", return_value=["testing"]
@@ -365,6 +368,7 @@ def test_send_sms_should_use_service_sms_sender(
 
     mocker.patch("app.delivery.send_to_providers.redis_store", return_value=None)
     mocker.patch("app.aws_sns_client.send_sms")
+    mocker.patch("app.delivery.send_to_providers.update_notification_message_id")
 
     sms_sender = create_service_sms_sender(
         service=sample_service, sms_sender="123456", is_default=False
@@ -405,6 +409,8 @@ def test_send_email_to_provider_should_not_send_to_provider_when_status_is_not_c
     )
     mocker.patch("app.aws_ses_client.send_email")
     mocker.patch("app.delivery.send_to_providers.send_email_response")
+
+    mocker.patch("app.delivery.send_to_providers.update_notification_message_id")
     mock_phone = mocker.patch("app.delivery.send_to_providers.get_phone_number_from_s3")
     mock_phone.return_value = "15555555555"
 
@@ -627,6 +633,10 @@ def test_should_update_billable_units_and_status_according_to_research_mode_and_
 ):
 
     mocker.patch("app.delivery.send_to_providers.redis_store", return_value=None)
+    mocker.patch(
+        "app.delivery.send_to_providers.update_notification_message_id",
+        return_value=None,
+    )
     mocker.patch(
         "app.delivery.send_to_providers.get_sender_numbers", return_value=["testing"]
     )
@@ -637,6 +647,11 @@ def test_should_update_billable_units_and_status_according_to_research_mode_and_
         key_type=key_type,
         reply_to_text="testing",
     )
+
+    mocker.patch(
+        "app.delivery.send_to_providers.update_notification_message_id",
+        return_value=None,
+    )
     mocker.patch("app.aws_sns_client.send_sms")
     mocker.patch(
         "app.delivery.send_to_providers.send_sms_response",
@@ -647,6 +662,8 @@ def test_should_update_billable_units_and_status_according_to_research_mode_and_
         sample_template.service.research_mode = True
 
     mock_phone = mocker.patch("app.delivery.send_to_providers.get_phone_number_from_s3")
+
+    mocker.patch("app.delivery.send_to_providers.update_notification_message_id")
     mock_phone.return_value = "15555555555"
 
     mock_personalisation = mocker.patch(
@@ -670,6 +687,8 @@ def test_should_set_notification_billable_units_and_reduces_provider_priority_if
     assert sample_notification.sent_by is None
 
     mock_phone = mocker.patch("app.delivery.send_to_providers.get_phone_number_from_s3")
+
+    mocker.patch("app.delivery.send_to_providers.update_notification_message_id")
     mock_phone.return_value = "15555555555"
 
     mock_personalisation = mocker.patch(
@@ -705,8 +724,14 @@ def test_should_send_sms_to_international_providers(
     )
 
     mock_s3 = mocker.patch("app.delivery.send_to_providers.get_phone_number_from_s3")
+
+    mocker.patch("app.delivery.send_to_providers.update_notification_message_id")
     mock_s3.return_value = "601117224412"
 
+    mocker.patch(
+        "app.delivery.send_to_providers.update_notification_message_id",
+        return_value=None,
+    )
     mock_personalisation = mocker.patch(
         "app.delivery.send_to_providers.get_personalisation_from_s3"
     )
@@ -744,6 +769,11 @@ def test_should_handle_sms_sender_and_prefix_message(
 
     mocker.patch("app.delivery.send_to_providers.redis_store", return_value=None)
     mocker.patch("app.aws_sns_client.send_sms")
+
+    mocker.patch(
+        "app.delivery.send_to_providers.update_notification_message_id",
+        return_value=None,
+    )
     service = create_service_with_defined_sms_sender(
         sms_sender_value=sms_sender, prefix_sms=prefix_sms
     )
@@ -803,6 +833,11 @@ def test_send_sms_to_provider_should_use_normalised_to(mocker, client, sample_te
     mocker.patch(
         "app.delivery.send_to_providers.get_sender_numbers", return_value=["testing"]
     )
+
+    mocker.patch(
+        "app.delivery.send_to_providers.update_notification_message_id",
+        return_value=None,
+    )
     send_mock = mocker.patch("app.aws_sns_client.send_sms")
     notification = create_notification(
         template=sample_template,
@@ -866,6 +901,11 @@ def test_send_sms_to_provider_should_return_template_if_found_in_redis(
     mocker.patch(
         "app.delivery.send_to_providers.get_sender_numbers", return_value=["testing"]
     )
+
+    mocker.patch(
+        "app.delivery.send_to_providers.update_notification_message_id",
+        return_value=None,
+    )
     from app.schemas import service_schema, template_schema
 
     service_dict = service_schema.dump(sample_template.service)