Improve and clarify large task error handling

Previously we changed this as an experiment [1] and forgot to check
if the new exception was appropriate. It is: testing with a large
task invocation ('process_job.apply_async(["a" * 200000])') gives...

Before (Celery 3, large-ish task):

    boto.exception.SQSError: SQSError: 400 Bad Request
    <?xml version="1.0"?><ErrorResponse xmlns="http://queue.amazonaws.com/doc/2012-11-05/"><Error><Type>Sender</Type><Code>InvalidParameterValue</Code><Message>One or more parameters are invalid. Reason: Message must be shorter than 262144 bytes.</Message><Detail/></Error><RequestId>96162552-cd96-5a14-b3a5-7f503300a662</RequestId></ErrorResponse>

Before (Celery 3, very large task):

    <hangs forever>

After (Celery 5, large-ish task):

    botocore.exceptions.ClientError: An error occurred (InvalidParameterValue) when calling the SendMessage operation: One or more parameters are invalid. Reason: Message must be shorter than 262144 bytes.

After (Celery 5, very large task):

    botocore.parsers.ResponseParserError: Unable to parse response (syntax error: line 1, column 0), invalid XML received. Further retries may succeed:
    b'HTTP content length exceeded 1662976 bytes.'

.pyup.yml

@@ -4,5 +4,5 @@ schedule: "every week on wednesday"
 
 search: False
 requirements:
-  - requirements.in
-  - requirements_for_test.txt
+  - requirements-app.txt
+  - requirements-dev.txt

Makefile

@@ -54,15 +54,29 @@ generate-version-file: ## Generates the app version file
 	@echo -e "__git_commit__ = \"${GIT_COMMIT}\"\n__time__ = \"${DATE}\"" > ${APP_VERSION_FILE}
 
 .PHONY: test
-test: ## Run tests
+test: test-requirements ## Run tests
 	flake8 .
 	isort --check-only ./app ./tests
 	pytest -n4 --maxfail=10
 
 .PHONY: freeze-requirements
 freeze-requirements: ## Pin all requirements including sub dependencies into requirements.txt
-	pip install --upgrade pip-tools
-	pip-compile requirements.in
+	rm -rf venv-freeze
+	virtualenv -p python3 venv-freeze
+	$$(pwd)/venv-freeze/bin/pip install -r requirements-app.txt
+	echo '# pyup: ignore file' > requirements.txt
+	echo '# This file is autogenerated. Do not edit it manually.' >> requirements.txt
+	cat requirements-app.txt >> requirements.txt
+	echo '' >> requirements.txt
+	$$(pwd)/venv-freeze/bin/pip freeze -r <(sed '/^--/d' requirements-app.txt) | sed -n '/The following requirements were added by pip freeze/,$$p' >> requirements.txt
+	rm -rf venv-freeze
+
+.PHONY: test-requirements
+test-requirements:
+	@diff requirements-app.txt requirements.txt | grep '<' \
+	    && { echo "requirements.txt doesn't match requirements-app.txt."; \
+	         echo "Run 'make freeze-requirements' to update."; exit 1; } \
+|| { echo "requirements.txt is up to date"; exit 0; }
 
 .PHONY: clean
 clean:

README.md

@@ -9,11 +9,7 @@ Contains:
 
 ### Python version
 
-We run python 3.9 both locally and in production.
-
-### pycurl
-
-See https://github.com/alphagov/notifications-manuals/wiki/Getting-started#pycurl
+At the moment we run Python 3.6 in production. You will run into problems if you try to use Python 3.5 or older, or Python 3.7 or newer.
 
 ### AWS credentials
 
@@ -92,9 +88,17 @@ make test
 
 ## To update application dependencies
 
-## To update application dependencies
+`requirements.txt` file is generated from the `requirements-app.txt` in order to pin
+versions of all nested dependencies. If `requirements-app.txt` has been changed (or
+we want to update the unpinned nested dependencies) `requirements.txt` should be
+regenerated with
+
+```
+make freeze-requirements
+```
+
+`requirements.txt` should be committed alongside `requirements-app.txt` changes.
 
-`requirements.txt` is generated from the `requirements.in` in order to pin versions of all nested dependencies. If `requirements.in` has been changed, run `make freeze-requirements` to regenerate it.
 
 ## To run one off tasks
 

app/__init__.py

@@ -20,7 +20,6 @@ from flask_sqlalchemy import SQLAlchemy as _SQLAlchemy
 from gds_metrics import GDSMetrics
 from gds_metrics.metrics import Gauge, Histogram
 from notifications_utils import logging, request_helper
-from notifications_utils.celery import NotifyCelery
 from notifications_utils.clients.encryption.encryption_client import Encryption
 from notifications_utils.clients.redis.redis_client import RedisClient
 from notifications_utils.clients.statsd.statsd_client import StatsdClient
@@ -29,6 +28,7 @@ from sqlalchemy import event
 from werkzeug.exceptions import HTTPException as WerkzeugHTTPException
 from werkzeug.local import LocalProxy
 
+from app.celery.celery import NotifyCelery
 from app.clients import NotificationProviderClients
 from app.clients.cbc_proxy import CBCProxyClient
 from app.clients.document_download import DocumentDownloadClient

app/broadcast_message/translators.py

@@ -29,5 +29,5 @@ def cap_xml_polygon_to_list(polygon_string):
         [
             float(coordinate) for coordinate in pair.split(',')
         ]
-        for pair in polygon_string.strip().split(' ')
+        for pair in polygon_string.split(' ')
     ]

app/celery/celery.py

@@ -0,0 +1,100 @@
+import time
+
+from celery import Celery, Task
+from celery.signals import worker_process_shutdown
+from flask import g, request
+from flask.ctx import has_app_context, has_request_context
+
+
+@worker_process_shutdown.connect
+def log_on_worker_shutdown(sender, signal, pid, exitcode, **kwargs):
+    # imported here to avoid circular imports
+    from app import notify_celery
+
+    # if the worker has already restarted at least once, then we no longer have app context and current_app won't work
+    # to create a new one. Instead we have to create a new app context from the original flask app and use that instead.
+    with notify_celery._app.app_context():
+        # if the worker has restarted
+        notify_celery._app.logger.info('worker shutdown: PID: {} Exitcode: {}'.format(pid, exitcode))
+
+
+def make_task(app):
+    class NotifyTask(Task):
+        abstract = True
+        start = None
+        typing = False
+
+        def on_success(self, retval, task_id, args, kwargs):
+            elapsed_time = time.monotonic() - self.start
+            delivery_info = self.request.delivery_info or {}
+            queue_name = delivery_info.get('routing_key', 'none')
+
+            app.logger.info(
+                "Celery task {task_name} (queue: {queue_name}) took {time}".format(
+                    task_name=self.name,
+                    queue_name=queue_name,
+                    time="{0:.4f}".format(elapsed_time)
+                )
+            )
+
+            app.statsd_client.timing(
+                "celery.{queue_name}.{task_name}.success".format(
+                    task_name=self.name,
+                    queue_name=queue_name
+                ), elapsed_time
+            )
+
+        def on_failure(self, exc, task_id, args, kwargs, einfo):
+            delivery_info = self.request.delivery_info or {}
+            queue_name = delivery_info.get('routing_key', 'none')
+
+            app.logger.exception(
+                "Celery task {task_name} (queue: {queue_name}) failed".format(
+                    task_name=self.name,
+                    queue_name=queue_name,
+                )
+            )
+
+            app.statsd_client.incr(
+                "celery.{queue_name}.{task_name}.failure".format(
+                    task_name=self.name,
+                    queue_name=queue_name
+                )
+            )
+
+            super().on_failure(exc, task_id, args, kwargs, einfo)
+
+        def __call__(self, *args, **kwargs):
+            # ensure task has flask context to access config, logger, etc
+            with app.app_context():
+                self.start = time.monotonic()
+                # Remove piggyback values from kwargs
+                # Add 'request_id' to 'g' so that it gets logged
+                g.request_id = kwargs.pop('request_id', None)
+
+                return super().__call__(*args, **kwargs)
+
+    return NotifyTask
+
+
+class NotifyCelery(Celery):
+
+    def init_app(self, app):
+        super().__init__(
+            app.import_name,
+            broker=app.config['CELERY']['broker_url'],
+            task_cls=make_task(app),
+        )
+
+        self.conf.update(app.config['CELERY'])
+        self._app = app
+
+    def send_task(self, name, args=None, kwargs=None, **other_kwargs):
+        kwargs = kwargs or {}
+
+        if has_request_context() and hasattr(request, 'request_id'):
+            kwargs['request_id'] = request.request_id
+        elif has_app_context() and 'request_id' in g:
+            kwargs['request_id'] = g.request_id
+
+        return super().send_task(name, args, kwargs, **other_kwargs)

app/celery/letters_pdf_tasks.py

@@ -31,7 +31,6 @@ from app.letters.utils import (
     generate_letter_pdf_filename,
     get_billable_units_for_letter_page_count,
     get_file_names_from_error_bucket,
-    get_folder_name,
     get_reference_from_filename,
     move_error_pdf_to_scan_bucket,
     move_failed_pdf,
@@ -126,6 +125,7 @@ def collate_letter_pdfs_to_be_sent():
     that have not yet been sent.
     If run after midnight, it will collect up letters created before 5:30pm the day before.
     """
+    current_app.logger.info("starting collate-letter-pdfs-to-be-sent")
     print_run_date = convert_utc_to_bst(datetime.utcnow())
     if print_run_date.time() < LETTER_PROCESSING_DEADLINE:
         print_run_date = print_run_date - timedelta(days=1)
@@ -524,37 +524,3 @@ def replay_letters_in_error(filename=None):
                     [filename],
                     queue=QueueNames.LETTERS
                 )
-
-
-@notify_celery.task(name='resanitise-pdf')
-def resanitise_pdf(notification_id):
-    """
-    `notification_id` is the notification id for a PDF letter which was either uploaded or sent using the API.
-
-    This task calls the `recreate_pdf_for_precompiled_letter` template preview task which recreates the
-    PDF for a letter which is already sanitised and in the letters-pdf bucket. The new file that is generated
-    will then overwrite the existing letter in the letters-pdf bucket.
-    """
-    notification = get_notification_by_id(notification_id)
-
-    # folder_name is the folder that the letter is in the letters-pdf bucket e.g. '2021-10-10/'
-    folder_name = get_folder_name(notification.created_at)
-
-    filename = generate_letter_pdf_filename(
-            reference=notification.reference,
-            created_at=notification.created_at,
-            ignore_folder=True,
-            postage=notification.postage
-        )
-
-    notify_celery.send_task(
-        name=TaskNames.RECREATE_PDF_FOR_PRECOMPILED_LETTER,
-        kwargs={
-            'notification_id': str(notification.id),
-            'file_location': f'{folder_name}{filename}',
-            'allow_international_letters': notification.service.has_permission(
-                INTERNATIONAL_LETTERS
-            ),
-        },
-        queue=QueueNames.SANITISE_LETTERS,
-    )

app/celery/nightly_tasks.py

@@ -136,7 +136,7 @@ def timeout_notifications():
     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
+        service_callback_api = get_service_delivery_status_callback_api_for_service(service_id=notification.service_id)
         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],

app/celery/reporting_tasks.py

@@ -20,6 +20,7 @@ from app.models import EMAIL_TYPE, LETTER_TYPE, SMS_TYPE
 @notify_celery.task(name="create-nightly-billing")
 @cronitor("create-nightly-billing")
 def create_nightly_billing(day_start=None):
+    current_app.logger.info("create-nightly-billing task: started")
     # day_start is a datetime.date() object. e.g.
     # up to 4 days of data counting back from day_start is consolidated
     if day_start is None:
@@ -66,6 +67,7 @@ def create_nightly_billing_for_day(process_day):
 @notify_celery.task(name="create-nightly-notification-status")
 @cronitor("create-nightly-notification-status")
 def create_nightly_notification_status():
+    current_app.logger.info("create-nightly-notification-status task: started")
     yesterday = convert_utc_to_bst(datetime.utcnow()).date() - timedelta(days=1)
 
     # email and sms

app/celery/scheduled_tasks.py

@@ -334,11 +334,3 @@ def auto_expire_broadcast_messages():
             name=TaskNames.PUBLISH_GOVUK_ALERTS,
             queue=QueueNames.GOVUK_ALERTS
         )
-
-
-@notify_celery.task(name='remove-yesterdays-planned-tests-on-govuk-alerts')
-def remove_yesterdays_planned_tests_on_govuk_alerts():
-    notify_celery.send_task(
-        name=TaskNames.PUBLISH_GOVUK_ALERTS,
-        queue=QueueNames.GOVUK_ALERTS
-    )

app/commands.py

@@ -19,10 +19,7 @@ from sqlalchemy.orm.exc import NoResultFound
 
 from app import db
 from app.aws import s3
-from app.celery.letters_pdf_tasks import (
-    get_pdf_for_templated_letter,
-    resanitise_pdf,
-)
+from app.celery.letters_pdf_tasks import get_pdf_for_templated_letter
 from app.celery.reporting_tasks import (
     create_nightly_notification_status_for_day,
 )
@@ -274,22 +271,14 @@ def insert_inbound_numbers_from_file(file_name):
     file.close()
 
 
-@notify_command(name='replay-create-pdf-for-templated-letter')
+@notify_command(name='replay-create-pdf-letters')
 @click.option('-n', '--notification_id', type=click.UUID, required=True,
               help="Notification id of the letter that needs the get_pdf_for_templated_letter task replayed")
-def replay_create_pdf_for_templated_letter(notification_id):
+def replay_create_pdf_letters(notification_id):
     print("Create task to get_pdf_for_templated_letter for notification: {}".format(notification_id))
     get_pdf_for_templated_letter.apply_async([str(notification_id)], queue=QueueNames.CREATE_LETTERS_PDF)
 
 
-@notify_command(name='recreate-pdf-for-precompiled-or-uploaded-letter')
-@click.option('-n', '--notification_id', type=click.UUID, required=True,
-              help="Notification ID of the precompiled or uploaded letter")
-def recreate_pdf_for_precompiled_or_uploaded_letter(notification_id):
-    print(f"Call resanitise_pdf task for notification: {notification_id}")
-    resanitise_pdf.apply_async([str(notification_id)], queue=QueueNames.LETTERS)
-
-
 @notify_command(name='replay-service-callbacks')
 @click.option('-f', '--file_name', required=True,
               help="""Full path of the file to upload, file is a contains client references of

app/config.py

@@ -77,7 +77,6 @@ class TaskNames(object):
     SANITISE_LETTER = 'sanitise-and-upload-letter'
     CREATE_PDF_FOR_TEMPLATED_LETTER = 'create-pdf-for-templated-letter'
     PUBLISH_GOVUK_ALERTS = 'publish-govuk-alerts'
-    RECREATE_PDF_FOR_PRECOMPILED_LETTER = 'recreate-pdf-for-precompiled-letter'
 
 
 class Config(object):
@@ -251,7 +250,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': {
@@ -332,11 +331,6 @@ class Config(object):
                 'schedule': timedelta(minutes=5),
                 'options': {'queue': QueueNames.PERIODIC}
             },
-            'remove-yesterdays-planned-tests-on-govuk-alerts': {
-                'task': 'remove-yesterdays-planned-tests-on-govuk-alerts',
-                'schedule': crontab(hour=00, minute=00),
-                'options': {'queue': QueueNames.PERIODIC}
-            },
         }
     }
 

app/cronitor.py

@@ -5,15 +5,12 @@ from flask import current_app
 
 
 def cronitor(task_name):
+    # check if task_name is in config
     def decorator(func):
         def ping_cronitor(command):
             if not current_app.config['CRONITOR_ENABLED']:
                 return
 
-            # it's useful to have a log that a periodic task has started in case it
-            # get stuck without generating any other logs - we know it got this far
-            current_app.logger.info(f'Pinging Cronitor for Celery task {task_name}')
-
             task_slug = current_app.config['CRONITOR_KEYS'].get(task_name)
             if not task_slug:
                 current_app.logger.error(

app/dao/notifications_dao.py

@@ -398,14 +398,14 @@ def insert_notification_history_delete_notifications(
     select_to_use = select_into_temp_table_for_letters if notification_type == 'letter' else select_into_temp_table
     db.session.execute(select_to_use, input_params)
 
-    result = db.session.execute("select count(*) from NOTIFICATION_ARCHIVE").fetchone()[0]
+    result = db.session.execute("select * from NOTIFICATION_ARCHIVE")
 
     db.session.execute(insert_query)
 
     db.session.execute(delete_query)
 
     db.session.execute("DROP TABLE NOTIFICATION_ARCHIVE")
-    return result
+    return result.rowcount
 
 
 def _move_notifications_to_notification_history(notification_type, service_id, day_to_delete_backwards_from, qry_limit):
@@ -470,11 +470,12 @@ def _timeout_notifications(current_statuses, new_status, timeout_start, updated_
     notifications = Notification.query.filter(
         Notification.created_at < timeout_start,
         Notification.status.in_(current_statuses),
-        Notification.notification_type.in_([SMS_TYPE, EMAIL_TYPE])
+        Notification.notification_type != LETTER_TYPE
     ).all()
-
     Notification.query.filter(
-        Notification.id.in_([n.id for n in notifications]),
+        Notification.created_at < timeout_start,
+        Notification.status.in_(current_statuses),
+        Notification.notification_type != LETTER_TYPE
     ).update(
         {'status': new_status, 'updated_at': updated_at},
         synchronize_session=False
@@ -827,7 +828,7 @@ def _duplicate_update_warning(notification, status):
     current_app.logger.info(
         (
             'Duplicate callback received. Notification id {id} received a status update to {new_status}'
-            ' from {old_status} for {type} sent by {sent_by}. This happened {time_diff} after being first set.'
+            '{time_diff} after being set to {old_status}. {type} sent by {sent_by}'
         ).format(
             id=notification.id,
             old_status=notification.status,

app/v2/broadcast/post_broadcast.py

@@ -44,17 +44,9 @@ def create_broadcast():
     _validate_template(broadcast_json)
 
     polygons = Polygons(list(chain.from_iterable((
-        [
-            [[y, x] for x, y in polygon]
-            for polygon in area['polygons']
-        ] for area in broadcast_json['areas']
+        area['polygons'] for area in broadcast_json['areas']
     ))))
 
-    if len(polygons) > 12 or polygons.point_count > 250:
-        simple_polygons = polygons.smooth.simplify
-    else:
-        simple_polygons = polygons
-
     broadcast_message = BroadcastMessage(
         service_id=authenticated_service.id,
         content=broadcast_json['content'],
@@ -64,7 +56,7 @@ def create_broadcast():
             'names': [
                 area['name'] for area in broadcast_json['areas']
             ],
-            'simple_polygons': simple_polygons.as_coordinate_pairs_lat_long,
+            'simple_polygons': polygons.smooth.simplify.as_coordinate_pairs_long_lat,
         },
         status=BroadcastStatusType.PENDING_APPROVAL,
         api_key_id=api_user.id,

requirements-app.txt

@@ -2,7 +2,7 @@
 # with package version changes made in requirements-app.txt
 
 cffi==1.14.5
-celery[sqs]==5.2.0
+celery[sqs]==5.1.2
 docopt==0.6.2
 Flask-Bcrypt==0.7.1
 flask-marshmallow==0.14.0
@@ -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==48.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

@@ -1,263 +1,110 @@
-#
-# This file is autogenerated by pip-compile with python 3.9
-# To update, run:
-#
-#    pip-compile requirements.in
-#
-alembic==1.7.4
-    # via flask-migrate
-amqp==5.0.6
-    # via kombu
-attrs==21.2.0
-    # via jsonschema
-awscli==1.21.4
-    # via
-    #   awscli-cwlogs
-    #   notifications-utils
-awscli-cwlogs==1.4.6
-    # via -r requirements.in
-bcrypt==3.2.0
-    # via flask-bcrypt
-beautifulsoup4==4.9.3
-    # via -r requirements.in
-billiard==3.6.4.0
-    # via celery
-bleach==4.1.0
-    # via notifications-utils
-blinker==1.4
-    # via gds-metrics
-boto3==1.19.4
-    # via notifications-utils
-botocore==1.22.4
-    # via
-    #   awscli
-    #   boto3
-    #   s3transfer
-cachetools==4.2.1
-    # via
-    #   -r requirements.in
-    #   notifications-utils
-celery[sqs]==5.2.0
-    # via -r requirements.in
-certifi==2021.10.8
-    # via
-    #   pyproj
-    #   requests
-cffi==1.14.5
-    # via
-    #   -r requirements.in
-    #   bcrypt
-    #   cryptography
-charset-normalizer==2.0.7
-    # via requests
-click==8.0.3
-    # via
-    #   celery
-    #   click-datetime
-    #   click-didyoumean
-    #   click-plugins
-    #   click-repl
-    #   flask
-click-datetime==0.2
-    # via -r requirements.in
-click-didyoumean==0.3.0
-    # via celery
-click-plugins==1.1.1
-    # via celery
-click-repl==0.2.0
-    # via celery
-colorama==0.4.3
-    # via awscli
-cryptography==3.3.2
-    # via -r requirements.in
-dnspython==1.16.0
-    # via eventlet
-docopt==0.6.2
-    # via
-    #   -r requirements.in
-    #   notifications-python-client
-docutils==0.15.2
-    # via awscli
-eventlet==0.30.2 # pyup: ignore
-    # via -r requirements.in
-flask==1.1.2
-    # via
-    #   -r requirements.in
-    #   flask-bcrypt
-    #   flask-marshmallow
-    #   flask-migrate
-    #   flask-redis
-    #   gds-metrics
-    #   notifications-utils
-flask-bcrypt==0.7.1
-    # via -r requirements.in
-flask-marshmallow==0.14.0
-    # via -r requirements.in
-flask-migrate==2.7.0
-    # via -r requirements.in
-flask-redis==0.4.0
-    # via notifications-utils
-flask-sqlalchemy @ git+https://github.com/mitsuhiko/flask-sqlalchemy.git@500e732dd1b975a56ab06a46bd1a20a21e682262
-    # via
-    #   -r requirements.in
-    #   flask-migrate
-gds-metrics==0.2.4
-    # via -r requirements.in
-geojson==2.5.0
-    # via notifications-utils
-govuk-bank-holidays==0.10
-    # via notifications-utils
-greenlet==1.1.2
-    # via
-    #   eventlet
-    #   sqlalchemy
-gunicorn==20.1.0
-    # via -r requirements.in
-idna==3.3
-    # via requests
-iso8601==0.1.14
-    # via -r requirements.in
-itsdangerous==1.1.0
-    # via
-    #   -r requirements.in
-    #   flask
-    #   notifications-utils
-jinja2==3.0.2
-    # via
-    #   flask
-    #   notifications-utils
-jmespath==0.10.0
-    # via
-    #   boto3
-    #   botocore
-jsonschema==3.2.0
-    # via -r requirements.in
-kombu==5.2.1
-    # via celery
-lxml==4.6.3
-    # via -r requirements.in
-mako==1.1.5
-    # via alembic
-markupsafe==2.0.1
-    # via
-    #   jinja2
-    #   mako
-marshmallow==2.21.0
-    # via
-    #   -r requirements.in
-    #   flask-marshmallow
-    #   marshmallow-sqlalchemy
-marshmallow-sqlalchemy==0.23.1
-    # via -r requirements.in
-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
-    # via -r requirements.in
-orderedset==2.0.3
-    # via notifications-utils
-packaging==21.0
-    # via bleach
-phonenumbers==8.12.36
-    # via notifications-utils
-prometheus-client==0.10.1
-    # via
-    #   -r requirements.in
-    #   gds-metrics
-prompt-toolkit==3.0.21
-    # via click-repl
-psycopg2-binary==2.8.6
-    # via -r requirements.in
-pyasn1==0.4.8
-    # via rsa
-pycparser==2.20
-    # via cffi
-pyjwt==2.0.1
-    # via
-    #   -r requirements.in
-    #   notifications-python-client
-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
-    # via
-    #   awscli-cwlogs
-    #   botocore
-python-json-logger==2.0.2
-    # via notifications-utils
-pytz==2021.3
-    # via
-    #   celery
-    #   notifications-utils
-pyyaml==5.4.1
-    # via
-    #   awscli
-    #   notifications-utils
-redis==3.5.3
-    # via flask-redis
-requests==2.26.0
-    # via
-    #   awscli-cwlogs
-    #   govuk-bank-holidays
-    #   notifications-python-client
-    #   notifications-utils
-rfc3987==1.3.8
-    # via -r requirements.in
-rsa==4.7.2
-    # via awscli
-s3transfer==0.5.0
-    # via
-    #   awscli
-    #   boto3
-shapely==1.8.0
-    # via notifications-utils
-six==1.16.0
-    # via
-    #   awscli-cwlogs
-    #   bcrypt
-    #   bleach
-    #   click-repl
-    #   cryptography
-    #   eventlet
-    #   flask-marshmallow
-    #   jsonschema
-    #   python-dateutil
-smartypants==2.0.1
-    # via notifications-utils
-soupsieve==2.2.1
-    # via beautifulsoup4
-sqlalchemy==1.4.10
-    # via
-    #   -r requirements.in
-    #   alembic
-    #   marshmallow-sqlalchemy
-statsd==3.3.0
-    # via notifications-utils
-strict-rfc3339==0.7
-    # via -r requirements.in
-urllib3==1.26.7
-    # via
-    #   botocore
-    #   requests
-vine==5.0.0
-    # via
-    #   amqp
-    #   celery
-    #   kombu
-wcwidth==0.2.5
-    # via prompt-toolkit
-webencodings==0.5.1
-    # via bleach
-werkzeug==2.0.2
-    # via
-    #   -r requirements.in
-    #   flask
+# pyup: ignore file
+# This file is autogenerated. Do not edit it manually.
+# Run `make freeze-requirements` to update requirements.txt
+# with package version changes made in requirements-app.txt
 
-# The following packages are considered to be unsafe in a requirements file:
-# setuptools
+cffi==1.14.5
+celery[sqs]==5.1.2
+docopt==0.6.2
+Flask-Bcrypt==0.7.1
+flask-marshmallow==0.14.0
+Flask-Migrate==2.7.0
+git+https://github.com/mitsuhiko/flask-sqlalchemy.git@500e732dd1b975a56ab06a46bd1a20a21e682262#egg=Flask-SQLAlchemy==2.3.2.dev20190108
+Flask==1.1.2
+click-datetime==0.2
+eventlet==0.30.2 # pyup: ignore # 0.31 breaks Gunicorn
+gunicorn==20.1.0
+iso8601==0.1.14
+itsdangerous==1.1.0
+jsonschema==3.2.0
+marshmallow-sqlalchemy==0.23.1 # pyup: <0.24.0 # marshmallow v3 throws errors
+marshmallow==2.21.0 # pyup: <3 # v3 throws errors
+psycopg2-binary==2.8.6
+PyJWT==2.0.1
+SQLAlchemy==1.4.10
+strict-rfc3339==0.7
+rfc3987==1.3.8
+cachetools==4.2.1
+beautifulsoup4==4.9.3
+lxml==4.6.3
+Werkzeug==2.0.2
+
+# higher version causes build to fail on PaaS due to lack of Rust
+# see https://github.com/pyca/cryptography/issues/5810
+cryptography<3.4 # pyup: <3.4
+
+notifications-python-client==6.0.2
+
+# PaaS
+awscli-cwlogs==1.4.6
+
+git+https://github.com/alphagov/notifications-utils.git@48.0.0#egg=notifications-utils==48.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
+gds-metrics==0.2.4
+
+## The following requirements were added by pip freeze:
+alembic==1.7.4
+amqp==5.0.6
+attrs==21.2.0
+awscli==1.21.4
+bcrypt==3.2.0
+billiard==3.6.4.0
+bleach==4.1.0
+blinker==1.4
+boto3==1.19.4
+botocore==1.22.4
+cached-property==1.5.2
+certifi==2021.10.8
+charset-normalizer==2.0.7
+click==7.1.2
+click-didyoumean==0.3.0
+click-plugins==1.1.1
+click-repl==0.2.0
+colorama==0.4.3
+dataclasses==0.8
+dnspython==1.16.0
+docutils==0.15.2
+flask-redis==0.4.0
+geojson==2.5.0
+govuk-bank-holidays==0.10
+greenlet==1.1.2
+idna==3.3
+importlib-metadata==4.8.1
+importlib-resources==5.3.0
+Jinja2==3.0.2
+jmespath==0.10.0
+kombu==5.1.0
+Mako==1.1.5
+MarkupSafe==2.0.1
+mistune==0.8.4
+orderedset==2.0.3
+packaging==21.0
+phonenumbers==8.12.36
+prompt-toolkit==3.0.21
+pyasn1==0.4.8
+pycparser==2.20
+pycurl==7.43.0.5
+pyparsing==3.0.1
+PyPDF2==1.26.0
+pyrsistent==0.18.0
+python-dateutil==2.8.2
+python-json-logger==2.0.2
+pytz==2021.3
+PyYAML==5.4.1
+redis==3.5.3
+requests==2.26.0
+rsa==4.7.2
+s3transfer==0.5.0
+Shapely==1.8.0
+six==1.16.0
+smartypants==2.0.1
+soupsieve==2.2.1
+statsd==3.3.0
+typing-extensions==3.10.0.2
+urllib3==1.26.7
+vine==5.0.0
+wcwidth==0.2.5
+webencodings==0.5.1
+zipp==3.6.0

runtime.txt

@@ -1 +1 @@
-python-3.9.x
+python-3.6.x

scripts/paas_app_wrapper.sh

@@ -17,7 +17,7 @@ case $NOTIFY_APP_NAME in
     -Q database-tasks,job-tasks 2> /dev/null
     ;;
   delivery-worker-research)
-    exec scripts/run_app_paas.sh celery -A run_celery.notify_celery worker --loglevel=INFO --concurrency=4 \
+    exec scripts/run_app_paas.sh celery -A run_celery.notify_celery worker --loglevel=INFO --concurrency=5 \
     -Q research-mode-tasks 2> /dev/null
     ;;
   delivery-worker-sender)
@@ -29,7 +29,7 @@ case $NOTIFY_APP_NAME in
     -Q periodic-tasks 2> /dev/null
     ;;
   delivery-worker-reporting)
-    exec scripts/run_app_paas.sh celery -A run_celery.notify_celery worker --loglevel=INFO --concurrency=2 \
+    exec scripts/run_app_paas.sh celery -A run_celery.notify_celery worker --loglevel=INFO --concurrency=2 -Ofair \
     -Q reporting-tasks 2> /dev/null
     ;;
   delivery-worker-priority)

tests/app/celery/test_celery.py

@@ -0,0 +1,117 @@
+import uuid
+
+import pytest
+from flask import g
+from freezegun import freeze_time
+
+from app import notify_celery
+
+
+# requiring notify_api ensures notify_celery.init_app has been called
+@pytest.fixture(scope='session')
+def celery_task(notify_api):
+    @notify_celery.task(name=uuid.uuid4(), base=notify_celery.task_cls)
+    def test_task(delivery_info=None): pass
+    return test_task
+
+
+@pytest.fixture
+def async_task(celery_task):
+    celery_task.push_request(delivery_info={'routing_key': 'test-queue'})
+    yield celery_task
+    celery_task.pop_request()
+
+
+def test_success_should_log_and_call_statsd(mocker, notify_api, async_task):
+    statsd = mocker.patch.object(notify_api.statsd_client, 'timing')
+    logger = mocker.patch.object(notify_api.logger, 'info')
+
+    with freeze_time() as frozen:
+        async_task()
+        frozen.tick(5)
+
+        async_task.on_success(
+            retval=None, task_id=1234, args=[], kwargs={}
+        )
+
+    statsd.assert_called_once_with(f'celery.test-queue.{async_task.name}.success', 5.0)
+    logger.assert_called_once_with(f'Celery task {async_task.name} (queue: test-queue) took 5.0000')
+
+
+def test_success_queue_when_applied_synchronously(mocker, notify_api, celery_task):
+    statsd = mocker.patch.object(notify_api.statsd_client, 'timing')
+    logger = mocker.patch.object(notify_api.logger, 'info')
+
+    with freeze_time() as frozen:
+        celery_task()
+        frozen.tick(5)
+
+        celery_task.on_success(
+            retval=None, task_id=1234, args=[], kwargs={}
+        )
+
+    statsd.assert_called_once_with(f'celery.none.{celery_task.name}.success', 5.0)
+    logger.assert_called_once_with(f'Celery task {celery_task.name} (queue: none) took 5.0000')
+
+
+def test_failure_should_log_and_call_statsd(mocker, notify_api, async_task):
+    statsd = mocker.patch.object(notify_api.statsd_client, 'incr')
+    logger = mocker.patch.object(notify_api.logger, 'exception')
+
+    async_task.on_failure(
+        exc=Exception, task_id=1234, args=[], kwargs={}, einfo=None
+    )
+
+    statsd.assert_called_once_with(f'celery.test-queue.{async_task.name}.failure')
+    logger.assert_called_once_with(f'Celery task {async_task.name} (queue: test-queue) failed')
+
+
+def test_failure_queue_when_applied_synchronously(mocker, notify_api, celery_task):
+    statsd = mocker.patch.object(notify_api.statsd_client, 'incr')
+    logger = mocker.patch.object(notify_api.logger, 'exception')
+
+    celery_task.on_failure(
+        exc=Exception, task_id=1234, args=[], kwargs={}, einfo=None
+    )
+
+    statsd.assert_called_once_with(f'celery.none.{celery_task.name}.failure')
+    logger.assert_called_once_with(f'Celery task {celery_task.name} (queue: none) failed')
+
+
+def test_call_exports_request_id_from_kwargs(mocker, celery_task):
+    g = mocker.patch('app.celery.celery.g')
+    # this would fail if the kwarg was passed through unexpectedly
+    celery_task(request_id='1234')
+    assert g.request_id == '1234'
+
+
+def test_send_task_injects_global_request_id_into_kwargs(mocker, notify_api):
+    super_apply = mocker.patch('celery.Celery.send_task')
+    g.request_id = '1234'
+    notify_celery.send_task('some-task')
+    super_apply.assert_called_with('some-task', None, {'request_id': '1234'})
+
+
+def test_send_task_injects_request_id_with_other_kwargs(mocker, notify_api):
+    super_apply = mocker.patch('celery.Celery.send_task')
+    g.request_id = '1234'
+    notify_celery.send_task('some-task', kwargs={'something': 'else'})
+    super_apply.assert_called_with('some-task', None, {'request_id': '1234', 'something': 'else'})
+
+
+def test_send_task_injects_request_id_with_positional_args(mocker, notify_api):
+    super_apply = mocker.patch('celery.Celery.send_task')
+    g.request_id = '1234'
+    notify_celery.send_task('some-task', ['args'], {'kw': 'args'})
+    super_apply.assert_called_with('some-task', ['args'], {'request_id': '1234', 'kw': 'args'})
+
+
+def test_send_task_injects_id_into_kwargs_from_request(mocker, notify_api):
+    super_apply = mocker.patch('celery.Celery.send_task')
+    request_id_header = notify_api.config['NOTIFY_TRACE_ID_HEADER']
+    request_headers = {request_id_header: '1234'}
+
+    with notify_api.test_request_context(headers=request_headers):
+        notify_celery.send_task('some-task')
+
+    super_apply.assert_called_with('some-task', None, {'request_id': '1234'})

tests/app/celery/test_letters_pdf_tasks.py

@@ -23,7 +23,6 @@ from app.celery.letters_pdf_tasks import (
     process_virus_scan_error,
     process_virus_scan_failed,
     replay_letters_in_error,
-    resanitise_pdf,
     sanitise_letter,
     send_letters_volume_email_to_dvla,
     update_billable_units_for_letter,
@@ -1099,33 +1098,3 @@ def test_replay_letters_in_error_for_one_file(notify_api, mocker):
     replay_letters_in_error("file_name")
     mock_move.assert_called_once_with('file_name')
     mock_celery.assert_called_once_with(name='scan-file', kwargs={'filename': 'file_name'}, queue='antivirus-tasks')
-
-
-@pytest.mark.parametrize('permissions, expected_international_letters_allowed', (
-    ([LETTER_TYPE], False),
-    ([LETTER_TYPE, INTERNATIONAL_LETTERS], True),
-))
-def test_resanitise_pdf_calls_template_preview_with_letter_details(
-    mocker,
-    sample_letter_notification,
-    permissions,
-    expected_international_letters_allowed,
-):
-    mock_celery = mocker.patch('app.celery.letters_pdf_tasks.notify_celery.send_task')
-
-    sample_letter_notification.created_at = datetime(2021, 2, 7, 12)
-    sample_letter_notification.service = create_service(
-        service_permissions=permissions
-    )
-
-    resanitise_pdf(sample_letter_notification.id)
-
-    mock_celery.assert_called_once_with(
-        name=TaskNames.RECREATE_PDF_FOR_PRECOMPILED_LETTER,
-        kwargs={
-            'notification_id': str(sample_letter_notification.id),
-            'file_location': '2021-02-07/NOTIFY.FOO.D.2.C.20210207120000.PDF',
-            'allow_international_letters': expected_international_letters_allowed,
-        },
-        queue=QueueNames.SANITISE_LETTERS,
-    )

tests/app/celery/test_scheduled_tasks.py

@@ -19,7 +19,6 @@ from app.celery.scheduled_tasks import (
     check_job_status,
     delete_invitations,
     delete_verify_codes,
-    remove_yesterdays_planned_tests_on_govuk_alerts,
     replay_created_notifications,
     run_scheduled_jobs,
     switch_current_sms_provider_on_slow_delivery,
@@ -701,16 +700,3 @@ def test_auto_expire_broadcast_messages(
         )
     else:
         assert not mock_celery.called
-
-
-def test_remove_yesterdays_planned_tests_on_govuk_alerts(
-    mocker
-):
-    mock_celery = mocker.patch('app.celery.scheduled_tasks.notify_celery.send_task')
-
-    remove_yesterdays_planned_tests_on_govuk_alerts()
-
-    mock_celery.assert_called_once_with(
-        name=TaskNames.PUBLISH_GOVUK_ALERTS,
-        queue=QueueNames.GOVUK_ALERTS
-    )

tests/app/test_cronitor.py

@@ -1,4 +1,3 @@
-import logging
 from urllib import parse
 
 import pytest
@@ -76,9 +75,6 @@ def test_cronitor_does_nothing_if_cronitor_not_enabled(notify_api, rmock):
 
 
 def test_cronitor_does_nothing_if_name_not_recognised(notify_api, rmock, caplog):
-    # ignore "INFO" log about task starting
-    caplog.set_level(logging.ERROR)
-
     with set_config_values(notify_api, {
         'CRONITOR_ENABLED': True,
         'CRONITOR_KEYS': {'not-hello': 'other'}

tests/app/v2/broadcast/test_post_broadcast.py

@@ -105,9 +105,9 @@ def test_valid_post_cap_xml_broadcast_returns_201(
     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['areas']['simple_polygons'][0]) == 22
+    assert response_json['areas']['simple_polygons'][0][0] == [53.10562, 0.244127]
+    assert response_json['areas']['simple_polygons'][0][-1] == [53.10562, 0.244127]
     assert response_json['areas']['names'] == ['River Steeping in Wainfleet All Saints']
     assert 'ids' not in response_json['areas']  # only for broadcasts created in Admin
 
@@ -119,27 +119,6 @@ def test_valid_post_cap_xml_broadcast_returns_201(
     assert response_json['updated_at'] is None
 
 
-def test_large_polygon_is_simplified(
-    client,
-    sample_broadcast_service,
-):
-    auth_header = create_service_authorization_header(service_id=sample_broadcast_service.id)
-    response = client.post(
-        path='/v2/broadcast',
-        data=sample_cap_xml_documents.WINDEMERE,
-        headers=[('Content-Type', 'application/cap+xml'), auth_header],
-    )
-    assert response.status_code == 201
-
-    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 response_json['areas']['simple_polygons'][0][0] == [54.419546, -2.988521]
-    assert response_json['areas']['simple_polygons'][0][-1] == [54.419546, -2.988521]
-
-
 @pytest.mark.parametrize("training_mode_service", [True, False])
 def test_valid_post_cap_xml_broadcast_sets_stubbed_to_true_for_training_mode_services(
     client,

tests/app/v2/notifications/test_post_notifications.py

@@ -1067,7 +1067,6 @@ def test_post_notifications_saves_email_or_sms_to_queue(client, notify_db_sessio
         assert not mock_send_task.called
         assert len(Notification.query.all()) == 0
 
-
 @pytest.mark.parametrize("exception", [
     botocore.exceptions.ClientError({'some': 'json'}, 'some opname'),
     botocore.parsers.ResponseParserError('exceeded max HTTP body length'),