Overriding retry policy for SMS high priority (#2008)

* Overriding retry policy for SMS high priority

* Added comments + format

* Fixed ordering

* mypy: Added typing to the untyped dict

* mypy again

* Fix existing test with new changes

* Fix existing test with new changes

* Refactoring: rename method

* Refactoring: granular params

* Added test around params building for delivery tasks

* Changed test for correct value of retry period for sms high

* Introducing feature flag for celery retry policies

* Changed FF name for custom task celery params

* Adding extension to center editor on kb cursor

* Forgot the test config to activate env var

* Use constants in tests

* Reversing env var config enabling logic (but same config)

* Arf got trick by Python conditional logic

* Print that env var because it aint working

* Format

* Removing print debug statement

* Apply low retry period for the sms high priority lane on celery retry send

* Refactoring retry periods for sms high priority

* Fix test

* Fix sort

* Added timonwong.shellcheck shell checker to devcontainer

* Removed manually raised exception only meant for testing

.devcontainer/devcontainer.json

@@ -26,6 +26,7 @@
 				"GitHub.copilot",
 				"GitHub.copilot-labs",
 				"googlecloudtools.cloudcode",
+				"kaiwood.center-editor-window",
 				"matangover.mypy",
 				"ms-azuretools.vscode-docker",
 				"ms-ossdata.vscode-postgresql",
@@ -35,6 +36,7 @@
 				"mtxr.sqltools",
 				"mtxr.sqltools-driver-pg",
 				"pmbenjamin.vscode-snyk",
+				"timonwong.shellcheck",
 				"usernamehw.errorlens",
 				"visualstudioexptteam.vscodeintellicode",
 				"wenfangdu.jump",

.vscode/launch.json

@@ -43,7 +43,7 @@
                 "-l",
                 "DEBUG",
                 "-Q",
-                "database-tasks,-priority-database-tasks.fifo,-normal-database-tasks,-bulk-database-tasks,job-tasks,notify-internal-tasks,periodic-tasks,priority-tasks,bulk-tasks,reporting-tasks,research-mode-tasks,retry-tasks,send-sms-tasks,send-email-tasks,service-callbacks,delivery-receipts",
+                "database-tasks,-priority-database-tasks.fifo,-normal-database-tasks,-bulk-database-tasks,job-tasks,notify-internal-tasks,periodic-tasks,priority-tasks,normal-tasks,bulk-tasks,reporting-tasks,research-mode-tasks,retry-tasks,send-sms-tasks,send-sms-high,send-sms-medium,send-sms-low,send-throttled-sms-tasks,send-email-high,send-email-medium,send-email-low,send-email-tasks,service-callbacks,delivery-receipts",
             ]
         },
         {

app/celery/provider_tasks.py

@@ -15,6 +15,7 @@
     NotificationTechnicalFailureException,
 )
 from app.models import NOTIFICATION_TECHNICAL_FAILURE
+from app.notifications import build_retry_task_params
 from app.notifications.callbacks import _check_and_queue_callback_task
 
 
@@ -127,7 +128,7 @@ def _deliver_sms(self, notification_id):
                 # Once the previous retry failed, log the exception and this time,
                 # retry with the default delay.
                 current_app.logger.exception("SMS notification delivery for id: {} failed".format(notification_id))
-                self.retry(queue=QueueNames.RETRY)
+                self.retry(**build_retry_task_params(notification.notification_type, notification.template.process_type))
         except self.MaxRetriesExceededError:
             message = (
                 "RETRY FAILED: Max retries reached. The task send_sms_to_provider failed for notification {}. "

app/config.py

@@ -574,7 +574,6 @@ class Config(object):
     CSV_BULK_REDIRECT_THRESHOLD = os.getenv("CSV_BULK_REDIRECT_THRESHOLD", 200)
 
     # Endpoint of Cloudwatch agent running as a side car in EKS listening for embedded metrics
-    FF_CLOUDWATCH_METRICS_ENABLED = env.bool("FF_CLOUDWATCH_METRICS_ENABLED", False)
     CLOUDWATCH_AGENT_EMF_PORT = 25888
     CLOUDWATCH_AGENT_ENDPOINT = os.getenv("CLOUDWATCH_AGENT_ENDPOINT", f"tcp://{STATSD_HOST}:{CLOUDWATCH_AGENT_EMF_PORT}")
 
@@ -583,14 +582,15 @@ class Config(object):
     BR_WARNING_PERCENTAGE = 0.05
     BR_CRITICAL_PERCENTAGE = 0.1
 
-    FF_SALESFORCE_CONTACT = env.bool("FF_SALESFORCE_CONTACT", False)
-
     # Feature flags for bounce rate
     # Timestamp in epoch milliseconds to seed the bounce rate. We will seed data for (24, the below config) included.
     FF_BOUNCE_RATE_SEED_EPOCH_MS = os.getenv("FF_BOUNCE_RATE_SEED_EPOCH_MS", False)
-
+    # Feature flag to enable custom retry policies such as lowering retry period for certain priority lanes.
+    FF_CELERY_CUSTOM_TASK_PARAMS = env.bool("FF_CELERY_CUSTOM_TASK_PARAMS", True)
+    FF_CLOUDWATCH_METRICS_ENABLED = env.bool("FF_CLOUDWATCH_METRICS_ENABLED", False)
     # Feature flags for email_daily_limit
     FF_EMAIL_DAILY_LIMIT = env.bool("FF_EMAIL_DAILY_LIMIT", False)
+    FF_SALESFORCE_CONTACT = env.bool("FF_SALESFORCE_CONTACT", False)
 
     @classmethod
     def get_sensitive_config(cls) -> list[str]:
@@ -687,11 +687,12 @@ class Test(Development):
     API_HOST_NAME = "http://localhost:6011"
 
     TEMPLATE_PREVIEW_API_HOST = "http://localhost:9999"
-    FF_EMAIL_DAILY_LIMIT = False
     CRM_GITHUB_PERSONAL_ACCESS_TOKEN = "test-token"
     CRM_ORG_LIST_URL = "https://test-url.com"
     FAILED_LOGIN_LIMIT = 0
 
+    FF_EMAIL_DAILY_LIMIT = False
+
 
 class Production(Config):
     NOTIFY_EMAIL_DOMAIN = os.getenv("NOTIFY_EMAIL_DOMAIN", "notification.canada.ca")
@@ -708,6 +709,7 @@ class Production(Config):
     API_RATE_LIMIT_ENABLED = True
     CHECK_PROXY_HEADER = False
     CRONITOR_ENABLED = False
+    FF_CELERY_CUSTOM_TASK_PARAMS = False
 
 
 class Staging(Production):

app/delivery/send_to_providers.py

@@ -60,6 +60,7 @@ def send_sms_to_provider(notification):
         inactive_service_failure(notification=notification)
         return
 
+    # If the notification was not sent already, the status should be created.
     if notification.status == "created":
         provider = provider_to_use(
             SMS_TYPE,
@@ -215,6 +216,8 @@ def send_email_to_provider(notification: Notification):
     if not service.active:
         inactive_service_failure(notification=notification)
         return
+
+    # If the notification was not sent already, the status should be created.
     if notification.status == "created":
         provider = provider_to_use(EMAIL_TYPE, notification.id)
 

app/notifications/__init__.py

@@ -0,0 +1,36 @@
+from typing import Any, Dict
+
+from flask import current_app
+
+from app.config import QueueNames
+from app.models import BULK, NORMAL, PRIORITY, SMS_TYPE
+
+# Default retry periods for sending notifications.
+RETRY_DEFAULT = 300
+RETRY_HIGH = 25
+
+RETRY_PERIODS = {
+    BULK: RETRY_DEFAULT,
+    NORMAL: RETRY_DEFAULT,
+    PRIORITY: RETRY_HIGH,
+}
+
+
+def build_retry_task_params(notification_type: str, notification_process_type: str) -> Dict[str, Any]:
+    """
+    Build task params for the sending parameter retry tasks.
+
+    If the notification is a high priority SMS, set the retry policy to retry every 25 seconds
+    else fall back to the default retry policy of retrying every 5 minutes.
+    """
+    params: dict[str, Any] = {"queue": QueueNames.RETRY}
+    if current_app.config["FF_CELERY_CUSTOM_TASK_PARAMS"] is False:
+        return params
+
+    # Overring the retry policy is only supported for SMS for now;
+    # email support coming later.
+    if notification_type == SMS_TYPE:
+        params["countdown"] = RETRY_PERIODS[notification_process_type]
+    else:
+        params["countdown"] = RETRY_DEFAULT
+    return params

app/notifications/process_notifications.py

@@ -210,7 +210,10 @@ def db_save_and_send_notification(notification: Notification):
 
     deliver_task = choose_deliver_task(notification)
     try:
-        deliver_task.apply_async([str(notification.id)], queue=notification.queue_name)
+        deliver_task.apply_async(
+            [str(notification.id)],
+            queue=notification.queue_name,
+        )
     except Exception:
         dao_delete_notifications_by_id(notification.id)
         raise

pytest.ini

@@ -15,6 +15,7 @@ env =
     ASSET_DOMAIN=assets.notification.canada.ca
     NOTIFY_EMAIL_DOMAIN=notification.canada.ca
     AWS_EMF_ENVIRONMENT=local
+    D:FF_CELERY_CUSTOM_TASK_PARAMS=True
     D:FF_SALESFORCE_CONTACT=True
     D:FF_CLOUDWATCH_METRICS_ENABLED=True
     D:REDIS_URL=redis://localhost:6380

tests/app/notifications/test_process_notification.py

@@ -1,5 +1,6 @@
 import datetime
 import uuid
+from typing import Any, Dict
 from unittest.mock import call
 
 import pytest
@@ -14,12 +15,18 @@
 from app.config import QueueNames
 from app.dao.service_sms_sender_dao import dao_update_service_sms_sender
 from app.models import (
+    BULK,
+    EMAIL_TYPE,
     LETTER_TYPE,
+    NORMAL,
+    PRIORITY,
+    SMS_TYPE,
     Notification,
     NotificationHistory,
     ScheduledNotification,
     Template,
 )
+from app.notifications import RETRY_PERIODS, build_retry_task_params
 from app.notifications.process_notifications import (
     choose_queue,
     create_content_for_notification,
@@ -1085,6 +1092,24 @@ def test_db_save_and_send_notification_throws_exception_deletes_notification(
         assert Notification.query.count() == 0
         assert NotificationHistory.query.count() == 0
 
+    @pytest.mark.parametrize(
+        ("notification_type, process_type, expected_retry_period"),
+        [
+            (EMAIL_TYPE, BULK, RETRY_PERIODS[BULK]),
+            (EMAIL_TYPE, NORMAL, RETRY_PERIODS[NORMAL]),
+            (EMAIL_TYPE, PRIORITY, RETRY_PERIODS[NORMAL]),
+            (SMS_TYPE, BULK, RETRY_PERIODS[BULK]),
+            (SMS_TYPE, NORMAL, RETRY_PERIODS[NORMAL]),
+            (SMS_TYPE, PRIORITY, RETRY_PERIODS[PRIORITY]),
+        ],
+    )
+    def test_retry_task_parameters(self, notify_api, notification_type, process_type, expected_retry_period):
+        with notify_api.app_context():
+            params: Dict[str, Any] = build_retry_task_params(notification_type, process_type)
+
+        assert params["queue"] == QueueNames.RETRY
+        assert params["countdown"] == expected_retry_period
+
     def test_db_save_and_send_notification_throws_exception_when_missing_template(self, sample_api_key, mocker):
         mocker.patch("app.celery.provider_tasks.deliver_sms.apply_async")
         assert Notification.query.count() == 0