Pass polygons through if they’re small already

If a polygon is smaller than the largest polygon in our dataset of
simplified polygons then we’re only throwing away useful detail by
simplifying it.

We should still simplify larger polygons as a fallback, to avoid sending
anything to the CBC that we’re not sure it will like.

The thresholds here are low: we can raise them as we test and experiment
more.

Here’s some data about the Flood Warning Service polygons

Percentile | 80% | 90%   | 95%    | 98%     | 99%     | 99.9%
-----------|-----|-------|--------|---------|---------|---------
Point count| 226 | 401.9 | 640.45 | 1015.38 | 1389.07 | 3008.609

Percentile    | 80% | 90%   | 95%    | 98%     | 99%     | 99.9%
--------------|-----|-------|--------|---------|---------|---------
Polygon count |2----|3------|5-------|8--------|10-------|40.469

app/celery/nightly_tasks.py

@@ -130,25 +130,35 @@ def delete_letter_notifications_older_than_retention():
 @notify_celery.task(name='timeout-sending-notifications')
 @cronitor('timeout-sending-notifications')
 def timeout_notifications():
-    technical_failure_notifications, temporary_failure_notifications = \
-        dao_timeout_notifications(current_app.config.get('SENDING_NOTIFICATIONS_TIMEOUT_PERIOD'))
+    # TEMPORARY: re-run the following code over small batches of notifications
+    # so that we can cope with a high volume that need processing. We've changed
+    # dao_timeout_notifications to return up to 100K notifications, so this task
+    # will operate on up to 500K - normally we only get around 20K.
+    for _ in range(0, 5):
+        technical_failure_notifications, temporary_failure_notifications = \
+            dao_timeout_notifications(current_app.config.get('SENDING_NOTIFICATIONS_TIMEOUT_PERIOD'))
 
-    notifications = technical_failure_notifications + temporary_failure_notifications
-    for notification in notifications:
-        # queue callback task only if the service_callback_api exists
-        service_callback_api = get_service_delivery_status_callback_api_for_service(service_id=notification.service_id)  # noqa: E501
-        if service_callback_api:
-            encrypted_notification = create_delivery_status_callback_data(notification, service_callback_api)
-            send_delivery_status_to_service.apply_async([str(notification.id), encrypted_notification],
-                                                        queue=QueueNames.CALLBACKS)
+        notifications = technical_failure_notifications + temporary_failure_notifications
+        for notification in notifications:
+            # queue callback task only if the service_callback_api exists
+            service_callback_api = get_service_delivery_status_callback_api_for_service(service_id=notification.service_id)  # noqa: E501
+            if service_callback_api:
+                encrypted_notification = create_delivery_status_callback_data(notification, service_callback_api)
+                send_delivery_status_to_service.apply_async([str(notification.id), encrypted_notification],
+                                                            queue=QueueNames.CALLBACKS)
 
-    current_app.logger.info(
-        "Timeout period reached for {} notifications, status has been updated.".format(len(notifications)))
-    if technical_failure_notifications:
-        message = "{} notifications have been updated to technical-failure because they " \
-                  "have timed out and are still in created.Notification ids: {}".format(
-                      len(technical_failure_notifications), [str(x.id) for x in technical_failure_notifications])
-        raise NotificationTechnicalFailureException(message)
+        current_app.logger.info(
+            "Timeout period reached for {} notifications, status has been updated.".format(len(notifications)))
+        if technical_failure_notifications:
+            message = "{} notifications have been updated to technical-failure because they " \
+                      "have timed out and are still in created.Notification ids: {}".format(
+                          len(technical_failure_notifications), [str(x.id) for x in technical_failure_notifications])
+            raise NotificationTechnicalFailureException(message)
+
+        if len(notifications) < 100000:
+            return
+
+    raise RuntimeError("Some notifications may still be in sending.")
 
 
 @notify_celery.task(name="delete-inbound-sms")

app/config.py

@@ -251,7 +251,7 @@ class Config(object):
             # app/celery/nightly_tasks.py
             'timeout-sending-notifications': {
                 'task': 'timeout-sending-notifications',
-                'schedule': crontab(minute=5),
+                'schedule': crontab(hour=0, minute=5),
                 'options': {'queue': QueueNames.PERIODIC}
             },
             'create-nightly-billing': {

app/dao/notifications_dao.py

@@ -467,13 +467,20 @@ def dao_delete_notifications_by_id(notification_id):
 
 
 def _timeout_notifications(current_statuses, new_status, timeout_start, updated_at):
+    # TEMPORARY: limit the notifications to 100K as otherwise we
+    # see an issues where the task vanishes after it starts executing
+    # - we believe this is a OOM error but there are no logs. From
+    # experimentation we've found we can safely process up to 100K.
     notifications = Notification.query.filter(
         Notification.created_at < timeout_start,
         Notification.status.in_(current_statuses),
         Notification.notification_type.in_([SMS_TYPE, EMAIL_TYPE])
-    ).all()
+    ).limit(100000).all()
 
     Notification.query.filter(
+        Notification.created_at < timeout_start,
+        Notification.status.in_(current_statuses),
+        Notification.notification_type.in_([SMS_TYPE, EMAIL_TYPE]),
         Notification.id.in_([n.id for n in notifications]),
     ).update(
         {'status': new_status, 'updated_at': updated_at},

requirements.in

@@ -36,7 +36,7 @@ notifications-python-client==6.0.2
 # PaaS
 awscli-cwlogs==1.4.6
 
-git+https://github.com/alphagov/notifications-utils.git@49.0.0#egg=notifications-utils==49.0.0
+git+https://github.com/alphagov/notifications-utils.git@48.0.0#egg=notifications-utils==49.0.0
 
 # gds-metrics requires prometheseus 0.2.0, override that requirement as 0.7.1 brings significant performance gains
 prometheus-client==0.10.1

requirements.txt

@@ -40,9 +40,7 @@ cachetools==4.2.1
 celery[sqs]==5.2.0
     # via -r requirements.in
 certifi==2021.10.8
-    # via
-    #   pyproj
-    #   requests
+    # via requests
 cffi==1.14.5
     # via
     #   -r requirements.in
@@ -153,7 +151,7 @@ mistune==0.8.4
     # via notifications-utils
 notifications-python-client==6.0.2
     # via -r requirements.in
-notifications-utils @ git+https://github.com/alphagov/notifications-utils.git@49.0.0
+notifications-utils @ git+https://github.com/alphagov/notifications-utils.git@48.1.0
     # via -r requirements.in
 orderedset==2.0.3
     # via notifications-utils
@@ -181,8 +179,6 @@ pyparsing==3.0.1
     # via packaging
 pypdf2==1.26.0
     # via notifications-utils
-pyproj==3.2.1
-    # via notifications-utils
 pyrsistent==0.18.0
     # via jsonschema
 python-dateutil==2.8.2

tests/app/v2/broadcast/test_post_broadcast.py

@@ -104,10 +104,10 @@ def test_valid_post_cap_xml_broadcast_returns_201(
     assert response_json['personalisation'] is None
     assert response_json['service_id'] == str(sample_broadcast_service.id)
 
-    assert len(response_json['areas']['simple_polygons']) == 1
-    assert len(response_json['areas']['simple_polygons'][0]) == 29
-    assert response_json['areas']['simple_polygons'][0][0] == [53.10569, 0.24453]
-    assert response_json['areas']['simple_polygons'][0][-1] == [53.10569, 0.24453]
+    assert len(response_json['simple_polygons']) == 1
+    assert len(response_json['simple_polygons'][0]) == 29
+    assert response_json['simple_polygons'][0][0] == [53.10569, 0.24453]
+    assert response_json['simple_polygons'][0][-1] == [53.10569, 0.24453]
     assert response_json['areas']['names'] == ['River Steeping in Wainfleet All Saints']
     assert 'ids' not in response_json['areas']  # only for broadcasts created in Admin
 
@@ -133,11 +133,11 @@ def test_large_polygon_is_simplified(
 
     response_json = json.loads(response.get_data(as_text=True))
 
-    assert len(response_json['areas']['simple_polygons']) == 1
-    assert len(response_json['areas']['simple_polygons'][0]) == 110
+    assert len(response_json['simple_polygons']) == 1
+    assert len(response_json['simple_polygons'][0]) == 110
 
-    assert response_json['areas']['simple_polygons'][0][0] == [54.419546, -2.988521]
-    assert response_json['areas']['simple_polygons'][0][-1] == [54.419546, -2.988521]
+    assert response_json['simple_polygons'][0][0] == [54.419546, -2.988521]
+    assert response_json['simple_polygons'][0][-1] == [54.419546, -2.988521]
 
 
 @pytest.mark.parametrize("training_mode_service", [True, False])