Only aggregate status when necessary for a service

This takes a similar approach to the nightly deletion task so that
we only create sub-tasks when there are actually notifications to
aggregate for a given type and day [1].

We're making this change to stop the duplication errors we're getting
at the moment and ensure the task can scale to more messages and more
services. There are two parts to this:

- Each subtask should now run within the 5 minute visibility timeout.
However, they may still be duplicated if the parent task overruns [2].

- The parent task creates a mininal number of subtasks, and the query
to determine this is very fast for a normal process day (milliseconds).

Since all tasks will run quickly, there should be no more duplication.

In order to test this more nuanced task, I rewrote the tests:

- One test checks the subtask is called correctly.
- One test checks we create all the right subtasks.

[1]: https://github.com/alphagov/notifications-api/pull/3381
[2]: https://docs.google.com/document/d/1MaP6Nyy3nJKkuh_4lP1wuDm19X8LZITOLRd9n3Ax-xg/edit#heading=h.q3intzwqhfzl
This commit is contained in:
Ben Thorner
2022-01-25 11:29:57 +00:00
parent c8db58d0e8
commit 1213463b8e
3 changed files with 69 additions and 32 deletions

View File

@@ -3,7 +3,7 @@ from datetime import datetime, timedelta
from flask import current_app
from notifications_utils.timezones import convert_utc_to_bst
from app import db, notify_celery
from app import notify_celery
from app.config import QueueNames
from app.cronitor import cronitor
from app.dao.fact_billing_dao import (
@@ -14,7 +14,8 @@ from app.dao.fact_notification_status_dao import (
fetch_status_data_for_service_and_day,
update_fact_notification_status,
)
from app.models import EMAIL_TYPE, LETTER_TYPE, SMS_TYPE, Service
from app.dao.notifications_dao import get_service_ids_with_notifications_on_date
from app.models import EMAIL_TYPE, LETTER_TYPE, SMS_TYPE
@notify_celery.task(name="create-nightly-billing")
@@ -97,8 +98,11 @@ def create_nightly_notification_status():
for i in range(days):
process_day = yesterday - timedelta(days=i)
for (service_id,) in db.session.query(Service.id):
relevant_service_ids = get_service_ids_with_notifications_on_date(
notification_type, process_day
)
for service_id in relevant_service_ids:
create_nightly_notification_status_for_service_and_day.apply_async(
kwargs={
'process_day': process_day.isoformat(),