retry sending broadcasts

Retry tasks if they fail to send a broadcast event. Note that each task
tries the regular proxy and the failover proxy for that provider. This
runs a bit differently than our other retries:

Retry with exponential backoff. Our other tasks retry with a fixed delay
of 5 minutes between tries. If we can't send a broadcast, we want to try
immediately. So instead, implement an exponential backoff (1, 2, 4, 8,
... seconds delay). We can't delay for longer than 310 seconds due to
visibility timeout settings in SQS, so cap the delay at that amount.

Normally we give up retrying after a set amount of retries (often 4
hours). As broadcast content is much more important than normal
notifications, we don't ever want to give up on sending them to phones...

...UNLESS WE DO!

Sometimes we do want to give up sending a broadcast though! Broadcasts
have an expiry time, when they stop showing up on peoples devices, so if
that has passed then we don't need to send the broadcast out.

Broadcast events can also be superceded by updates or cancels. Check
that the event is the most recent event for that broadcast message, if
not, give up, as we don't want to accidentally send out two conflicting
events for the same message.
This commit is contained in:
Leo Hemsted
2021-01-19 10:05:48 +00:00
parent d390eb2cac
commit ac34fb9c05
6 changed files with 251 additions and 52 deletions

View File

@@ -1,10 +1,13 @@
import uuid
from datetime import datetime
from flask import current_app
from notifications_utils.statsd_decorators import statsd
from sqlalchemy.schema import Sequence
from celery.exceptions import MaxRetriesExceededError
from app import cbc_proxy_client, db, notify_celery
from app.clients.cbc_proxy import CBCProxyFatalException, CBCProxyRetryableException
from app.config import QueueNames
from app.models import BroadcastEventMessageType, BroadcastProvider
from app.dao.broadcast_message_dao import dao_get_broadcast_event_by_id, create_broadcast_provider_message
@@ -12,6 +15,40 @@ from app.dao.broadcast_message_dao import dao_get_broadcast_event_by_id, create_
from app.utils import format_sequential_number
def get_retry_delay(retry_count):
"""
Given a count of retries so far, return a delay for the next one.
`retry_count` should be 0 the first time a task fails.
"""
# TODO: replace with celery's built in exponential backoff
# 2 to the power of x. 1, 2, 4, 8, 16, 32, ...
delay = 2**retry_count
# never wait longer than 5 minutes
return min(delay, 300)
def check_provider_message_should_retry(broadcast_provider_message):
this_event = broadcast_provider_message.broadcast_event
if this_event.transmitted_finishes_at < datetime.utcnow():
print(this_event.transmitted_finishes_at, datetime.utcnow(),)
raise MaxRetriesExceededError(
f'Given up sending broadcast_event {this_event.id} ' +
f'to provider {broadcast_provider_message.provider}: ' +
f'The expiry time of {this_event.transmitted_finishes_at} has already passed'
)
newest_event = max(this_event.broadcast_message.events, key=lambda x: x.sent_at)
if this_event != newest_event:
raise MaxRetriesExceededError(
f'Given up sending broadcast_event {this_event.id} ' +
f'to provider {broadcast_provider_message.provider}: ' +
f'This event has been superceeded by {newest_event.message_type} broadcast_event {newest_event.id}'
)
@notify_celery.task(name="send-broadcast-event")
@statsd(namespace="tasks")
def send_broadcast_event(broadcast_event_id):
@@ -27,9 +64,10 @@ def send_broadcast_event(broadcast_event_id):
)
@notify_celery.task(name="send-broadcast-provider-message")
# max_retries=None: retry forever
@notify_celery.task(bind=True, name="send-broadcast-provider-message", max_retries=None)
@statsd(namespace="tasks")
def send_broadcast_provider_message(broadcast_event_id, provider):
def send_broadcast_provider_message(self, broadcast_event_id, provider):
broadcast_event = dao_get_broadcast_event_by_id(broadcast_event_id)
broadcast_provider_message = create_broadcast_provider_message(broadcast_event, provider)
@@ -54,41 +92,52 @@ def send_broadcast_provider_message(broadcast_event_id, provider):
cbc_proxy_provider_client = cbc_proxy_client.get_proxy(provider)
if broadcast_event.message_type == BroadcastEventMessageType.ALERT:
cbc_proxy_provider_client.create_and_send_broadcast(
identifier=str(broadcast_provider_message.id),
message_number=formatted_message_number,
headline="GOV.UK Notify Broadcast",
description=broadcast_event.transmitted_content['body'],
areas=areas,
sent=broadcast_event.sent_at_as_cap_datetime_string,
expires=broadcast_event.transmitted_finishes_at_as_cap_datetime_string,
channel=channel
)
elif broadcast_event.message_type == BroadcastEventMessageType.UPDATE:
cbc_proxy_provider_client.update_and_send_broadcast(
identifier=str(broadcast_provider_message.id),
message_number=formatted_message_number,
headline="GOV.UK Notify Broadcast",
description=broadcast_event.transmitted_content['body'],
areas=areas,
previous_provider_messages=broadcast_event.get_earlier_provider_messages(provider),
sent=broadcast_event.sent_at_as_cap_datetime_string,
expires=broadcast_event.transmitted_finishes_at_as_cap_datetime_string,
# We think an alert update should always go out on the same channel that created the alert
# We recognise there is a small risk with this code here that if the services channel was
# changed between an alert being sent out and then updated, then something might go wrong
# but we are relying on service channels changing almost never, and not mid incident
# We may consider in the future, changing this such that we store the channel a broadcast was
# sent on on the broadcast message itself and pick the value from there instead of the service
channel=channel
)
elif broadcast_event.message_type == BroadcastEventMessageType.CANCEL:
cbc_proxy_provider_client.cancel_broadcast(
identifier=str(broadcast_provider_message.id),
message_number=formatted_message_number,
previous_provider_messages=broadcast_event.get_earlier_provider_messages(provider),
sent=broadcast_event.sent_at_as_cap_datetime_string,
try:
if broadcast_event.message_type == BroadcastEventMessageType.ALERT:
cbc_proxy_provider_client.create_and_send_broadcast(
identifier=str(broadcast_provider_message.id),
message_number=formatted_message_number,
headline="GOV.UK Notify Broadcast",
description=broadcast_event.transmitted_content['body'],
areas=areas,
sent=broadcast_event.sent_at_as_cap_datetime_string,
expires=broadcast_event.transmitted_finishes_at_as_cap_datetime_string,
channel=channel
)
elif broadcast_event.message_type == BroadcastEventMessageType.UPDATE:
cbc_proxy_provider_client.update_and_send_broadcast(
identifier=str(broadcast_provider_message.id),
message_number=formatted_message_number,
headline="GOV.UK Notify Broadcast",
description=broadcast_event.transmitted_content['body'],
areas=areas,
previous_provider_messages=broadcast_event.get_earlier_provider_messages(provider),
sent=broadcast_event.sent_at_as_cap_datetime_string,
expires=broadcast_event.transmitted_finishes_at_as_cap_datetime_string,
# We think an alert update should always go out on the same channel that created the alert
# We recognise there is a small risk with this code here that if the services channel was
# changed between an alert being sent out and then updated, then something might go wrong
# but we are relying on service channels changing almost never, and not mid incident
# We may consider in the future, changing this such that we store the channel a broadcast was
# sent on on the broadcast message itself and pick the value from there instead of the service
channel=channel
)
elif broadcast_event.message_type == BroadcastEventMessageType.CANCEL:
cbc_proxy_provider_client.cancel_broadcast(
identifier=str(broadcast_provider_message.id),
message_number=formatted_message_number,
previous_provider_messages=broadcast_event.get_earlier_provider_messages(provider),
sent=broadcast_event.sent_at_as_cap_datetime_string,
)
except CBCProxyRetryableException as exc:
# this will raise MaxRetriesExceededError if we no longer want to retry
# (because the message has expired)
check_provider_message_should_retry(broadcast_provider_message)
self.retry(
exc=exc,
countdown=get_retry_delay(self.request.retries),
queue=QueueNames.BROADCASTS,
)