Don't keep important data in your Celery queue

The Celery library (previous posts) makes it as easy to schedule a task to run later as calling a function. Just change:

send_welcome_email('dan@example.com')

to:

send_welcome_email.apply_async('dan@example.com')

and rely on Celery to run it later.

But this introduces a new point of failure for your application -- if Celery loses track of that task and never runs it, then a user will not get an email that we wanted to send them.

This could happen, for example, if the server hosting your Celery queue crashed. You could set up a hot backup server for your queue server, but it would be a lot of work.

It's simpler if you treat your Celery queue like a cache -- it's helpful to keep around, but if you lose it, your work will still get done.

[Ed: the code snippets in this post are just intended as little illustrations to go with the ideas in this post, and by no means should they be taken as examples of how to implement these ideas in production. ]

We can do that by changing our pattern for doing things in the background. The key idea is to keep the information about the work that needs to be done in the database, with the rest of our crucial data.

For example, instead of:

send_welcome_email.apply_async('dan@example.com')

we might add a needs_welcome_email Boolean field to our model, and write:

user.needs_welcome_email = True
user.save()

Now we know from our database that this user needs to get a welcome email, independently of Celery's queue.

Then we set up a periodic task to send any emails that need sending:

@task
def send_background_emails():
    for user in User.objects.filter(needs_welcome_email=True):
        send_welcome_email(user.email)
        user.needs_welcome_email = False
        user.save()

We can run that every 5 minutes, it'll be quite cheap to run if there's no work to do, and it'll take care of all the "queued" welcome emails that need sending.

If we want the user to get their email faster, we can just schedule another run of the background task immediately:

user.needs_welcome_email = True
user.save()
send_background_emails.apply_async()

And the user will get their email as fast as they would have before.

We will still want to run the background task periodically in case our queued task gets lost (the whole point of this), but it doesn't have to run as frequently since it will rarely have any work to do.

By the way, I learned this while doing a code review of some of my co-worker Karen's code. This ability to continue learning is one of my favorite benefits of working on a team.

CELERYBEAT_SCHEDULE = {
    'process_new_work': {
        'task': 'tasks.send_background_emails',
        'schedule': timedelta(minutes=15),
    },
}

CELERYBEAT_SCHEDULE = {
    'process_new_work': {
        'task': 'tasks.send_background_emails',
        'schedule': timedelta(minutes=15),
        'options': {
            'expires': 10*60,  # 10 minutes
        }
    },
}