Task added to update 'sending' notifications after 72 hours, set task to temporary-failure.

app/celery/tasks.py

@@ -1,10 +1,11 @@
 import itertools
-from datetime import datetime
+from datetime import (datetime, timedelta)
 
 from flask import current_app
 from monotonic import monotonic
 from sqlalchemy.exc import SQLAlchemyError
 from app import clients, statsd_client
+from app.clients import STATISTICS_FAILURE
 from app.clients.email import EmailClientException
 from app.clients.sms import SmsClientException
 from app.dao.services_dao import dao_fetch_service_by_id
@@ -37,7 +38,9 @@ from app.dao.notifications_dao import (
     dao_update_notification,
     delete_notifications_created_more_than_a_week_ago,
     dao_get_notification_statistics_for_service_and_day,
-    update_provider_stats
+    update_provider_stats,
+    get_notifications,
+    update_notification_status_by_id
 )
 
 from app.dao.jobs_dao import (
@@ -524,3 +527,23 @@ def provider_to_use(notification_type, notification_id):
         raise Exception("No active {} providers".format(notification_type))
 
     return clients.get_client_by_name_and_type(active_providers_in_order[0].identifier, notification_type)
+
+
+@notify_celery.task(name='timeout-sending-notifications')
+def timeout_notifications():
+    notifications = get_notifications(filter_dict={'status': 'sending'})
+    now = datetime.utcnow()
+    for noti in notifications:
+        try:
+            if (now - noti.created_at) > timedelta(
+                seconds=current_app.config.get('SENDING_NOTIFICATIONS_TIMEOUT_PERIOD')
+            ):
+                update_notification_status_by_id(noti.id, 'temporary-failure', STATISTICS_FAILURE)
+                current_app.logger.info((
+                    "Timeout period reached for notification ({})"
+                    ", status has been updated.").format(noti.id))
+        except Exception as e:
+            current_app.logger.exception(e)
+            current_app.logger.error((
+                "Exception raised trying to timeout notification ({})"
+                ", skipping notification update.").format(noti.id))

app/dao/notifications_dao.py

@@ -338,6 +338,10 @@ def get_notification_by_id(notification_id):
     return Notification.query.filter_by(id=notification_id).first()
 
 
+def get_notifications(filter_dict=None):
+    return _filter_query(Notification.query, filter_dict=filter_dict)
+
+
 def get_notifications_for_service(service_id,
                                   filter_dict=None,
                                   page=1,

config.py

@@ -67,6 +67,11 @@ class Config(object):
             'task': 'delete-successful-notifications',
             'schedule': crontab(minute=0, hour='0,1,2'),
             'options': {'queue': 'periodic'}
+        },
+        'timeout-sending-notifications': {
+            'task': 'timeout-sending-notifications',
+            'schedule': crontab(minute=0, hour='0,1,2'),
+            'options': {'queue': 'periodic'}
         }
     }
     CELERY_QUEUES = [
@@ -99,6 +104,8 @@ class Config(object):
     STATSD_PORT = None
     STATSD_PREFIX = None
 
+    SENDING_NOTIFICATIONS_TIMEOUT_PERIOD = 259200
+
 
 class Development(Config):
     DEBUG = True

tests/app/celery/test_tasks.py

@@ -15,7 +15,8 @@ from app.celery.tasks import (
     delete_invitations,
     delete_failed_notifications,
     delete_successful_notifications,
-    provider_to_use
+    provider_to_use,
+    timeout_notifications
 )
 from app.celery.research_mode_tasks import (
     send_email_response,
@@ -1184,3 +1185,41 @@ def _notification_json(template, to, personalisation=None, job_id=None, row_numb
     if row_number:
         notification['row_number'] = row_number
     return notification
+
+
+def test_update_status_of_notifications_after_timeout(notify_api,
+                                                      notify_db,
+                                                      notify_db_session,
+                                                      sample_service,
+                                                      sample_template,
+                                                      mmg_provider):
+    with notify_api.test_request_context():
+        not1 = sample_notification(
+            notify_db,
+            notify_db_session,
+            service=sample_service,
+            template=sample_template,
+            status='sending',
+            created_at=datetime.utcnow() - timedelta(
+                seconds=current_app.config.get('SENDING_NOTIFICATIONS_TIMEOUT_PERIOD') + 10))
+        timeout_notifications()
+        assert not1.status == 'temporary-failure'
+
+
+def test_not_update_status_of_notification_before_timeout(notify_api,
+                                                          notify_db,
+                                                          notify_db_session,
+                                                          sample_service,
+                                                          sample_template,
+                                                          mmg_provider):
+    with notify_api.test_request_context():
+        not1 = sample_notification(
+            notify_db,
+            notify_db_session,
+            service=sample_service,
+            template=sample_template,
+            status='sending',
+            created_at=datetime.utcnow() - timedelta(
+                seconds=current_app.config.get('SENDING_NOTIFICATIONS_TIMEOUT_PERIOD') - 10))
+        timeout_notifications()
+        assert not1.status == 'sending'