add expire times for redis objects

app/celery/process_ses_receipts_tasks.py

@@ -12,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
@@ -65,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"
                 )
-                self.retry(queue=QueueNames.RETRY)
+                self.retry(
+                    queue=QueueNames.RETRY, expires=Config.DEFAULT_REDIS_EXPIRE_TIME
+                )
             else:
                 current_app.logger.warning(
                     f"Notification not found for reference: {reference} "
@@ -115,7 +117,7 @@ def process_ses_results(self, response):
 
     except Exception:
         current_app.logger.exception("Error processing SES results")
-        self.retry(queue=QueueNames.RETRY)
+        self.retry(queue=QueueNames.RETRY, expires=Config.DEFAULT_REDIS_EXPIRE_TIME)
 
 
 def determine_notification_bounce_type(ses_message):

app/celery/provider_tasks.py

@@ -10,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,
@@ -152,9 +152,15 @@ def deliver_sms(self, notification_id):
 
         try:
             if self.request.retries == 0:
-                self.retry(queue=QueueNames.RETRY, countdown=0)
+                self.retry(
+                    queue=QueueNames.RETRY,
+                    countdown=0,
+                    expires=Config.DEFAULT_REDIS_EXPIRE_TIME,
+                )
             else:
-                self.retry(queue=QueueNames.RETRY)
+                self.retry(
+                    queue=QueueNames.RETRY, expires=Config.DEFAULT_REDIS_EXPIRE_TIME
+                )
         except self.MaxRetriesExceededError:
             message = (
                 "RETRY FAILED: Max retries reached. The task send_sms_to_provider failed for notification {}. "
@@ -203,7 +209,7 @@ def deliver_email(self, notification_id):
                     f"RETRY: Email notification {notification_id} failed"
                 )
 
-            self.retry(queue=QueueNames.RETRY)
+            self.retry(queue=QueueNames.RETRY, expires=Config.DEFAULT_REDIS_EXPIRE_TIME)
         except self.MaxRetriesExceededError:
             message = (
                 "RETRY FAILED: Max retries reached. "

app/celery/tasks.py

@@ -7,7 +7,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 (
@@ -146,6 +146,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
 
@@ -369,7 +370,7 @@ def save_api_email_or_sms(self, encrypted_notification):
 
     except SQLAlchemyError:
         try:
-            self.retry(queue=QueueNames.RETRY)
+            self.retry(queue=QueueNames.RETRY, expires=Config.DEFAULT_REDIS_EXPIRE_TIME)
         except self.MaxRetriesExceededError:
             current_app.logger.exception(
                 f"Max retry failed Failed to persist notification {notification['id']}",
@@ -390,7 +391,11 @@ def handle_exception(task, notification, notification_id, exc):
         # 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)
+            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)
 
@@ -439,7 +444,9 @@ def send_inbound_sms_to_service(self, inbound_sms_id, service_id):
         )
         if not isinstance(e, HTTPError) or e.response.status_code >= 500:
             try:
-                self.retry(queue=QueueNames.RETRY)
+                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"

app/config.py

@@ -53,6 +53,7 @@ class TaskNames(object):
 
 class Config(object):
     NOTIFY_APP_NAME = "api"
+    DEFAULT_REDIS_EXPIRE_TIME = 8 * 24 * 60 * 60
     NOTIFY_ENVIRONMENT = getenv("NOTIFY_ENVIRONMENT", "development")
     # URL of admin app
     ADMIN_BASE_URL = getenv("ADMIN_BASE_URL", "http://localhost:6012")