Task Framework

Practical recipes for common task operations.

Create a New Task

Prerequisites

• Python function that can be serialised (JSON-compatible arguments)

• Understanding of what work the task will do

Steps

1. Define the task function in gyrinx/core/tasks.py:

from django.tasks import task

@task
def send_notification(user_id: str, message: str):
    """Send a notification to a user."""
    from gyrinx.core.models import User

    user = User.objects.get(pk=user_id)
    # ... send notification logic

1. Register the task in gyrinx/tasks/registry.py:

from gyrinx.core.tasks import send_notification

def _get_tasks() -> list[TaskRoute]:
    global _tasks
    if _tasks is None:
        from gyrinx.core.tasks import (
            # ... existing imports
            send_notification,
        )

        _tasks = [
            # ... existing tasks
            TaskRoute(send_notification),
        ]
    return _tasks

1. Enqueue the task from your application code:

from gyrinx.core.tasks import send_notification

# Enqueue for async execution
send_notification.enqueue(user_id=str(user.pk), message="Welcome!")

Notes

• All arguments must be JSON-serialisable (strings, numbers, lists, dicts)

• Use string UUIDs rather than UUID objects

• Import the task lazily in _get_tasks() to avoid circular imports

Schedule a Task to Run Periodically

Prerequisites

• An existing registered task

• Understanding of cron syntax

Steps

1. Add a schedule to the task registration:

TaskRoute(
    cleanup_expired_sessions,
    schedule="0 3 * * *",  # Daily at 3am UTC
)

1. Optionally specify a timezone:

TaskRoute(
    send_daily_digest,
    schedule="0 9 * * *",      # Daily at 9am
    schedule_timezone="Europe/London",
)

1. Deploy - The Cloud Scheduler job is created automatically on startup.

Common Schedules

Schedule	Expression
Every 5 minutes	*/5 * * * *
Every hour	0 * * * *
Daily at midnight	0 0 * * *
Weekly on Monday	0 0 * * 1
Monthly on the 1st	0 0 1 * *

Add a Kill Switch to a Task

Kill switches let you disable tasks at runtime without redeploying.

Steps

1. Add a setting in settings.py that reads from an environment variable:

# settings.py
ENABLE_BATCH_JOB = env.bool("ENABLE_BATCH_JOB", default=True)

1. Check the setting at the start of your task:

from django.conf import settings
from django.tasks import task

@task
def expensive_batch_job():
    if not settings.ENABLE_BATCH_JOB:
        logger.info("Batch job disabled via ENABLE_BATCH_JOB")
        return

    # ... rest of task

1. Document the setting in docs/deployment-environment-variables.md.

2. Disable in production by setting the environment variable:

# In Cloud Run environment variables
ENABLE_BATCH_JOB=false

Configure Retry Behaviour

Adjust how Pub/Sub retries failed tasks.

Steps

1. Set retry parameters in the task registration:

TaskRoute(
    process_large_file,
    ack_deadline=600,      # 10 minutes to complete before retry
    min_retry_delay=60,    # Wait at least 1 minute before retry
    max_retry_delay=1800,  # Cap backoff at 30 minutes
)

Guidelines

Scenario	ack_deadline	min_retry_delay	max_retry_delay
Quick task (<10s)	60	10	300
Normal task (<1m)	300	10	600
Long-running task	600	60	1800
Database-intensive	300	30	600

Notes

• ack_deadline is the time before Pub/Sub assumes the task failed

• Retry delay uses exponential backoff between min and max

• For database-intensive tasks, longer delays help avoid overwhelming the database

Make a Task Idempotent

Tasks may be delivered more than once. Design for idempotency.

Pattern 1: Check Before Acting

@task
def backfill_user_data(user_id: str):
    from django.db import transaction

    with transaction.atomic():
        # Lock the row to prevent races
        user = User.objects.select_for_update().get(pk=user_id)

        # Check if already processed
        if user.data_backfilled:
            logger.info(f"User {user_id} already backfilled, skipping")
            return

        # Do the work
        user.backfill_data()
        user.data_backfilled = True
        user.save()

Pattern 2: Upsert Operations

@task
def sync_external_data(external_id: str):
    data = fetch_from_external_api(external_id)

    # update_or_create is naturally idempotent
    ExternalRecord.objects.update_or_create(
        external_id=external_id,
        defaults={"data": data, "synced_at": timezone.now()},
    )

Pattern 3: Idempotency Keys

@task
def send_email(email_id: str):
    # Use the email_id as an idempotency key
    if SentEmail.objects.filter(email_id=email_id).exists():
        logger.info(f"Email {email_id} already sent")
        return

    # Send and record
    send_email_to_user(...)
    SentEmail.objects.create(email_id=email_id)

Test a Task Locally

Prerequisites

• Development environment running

• Task registered in registry

Steps

1. In development, tasks run immediately (no Pub/Sub):

# In Django shell
from gyrinx.core.tasks import hello_world

# This runs synchronously in development
hello_world.enqueue(name="Developer")

1. To test the actual task function:

# Call the underlying function directly
from gyrinx.core.tasks import refresh_list_facts

refresh_list_facts("list-uuid-here")

1. To test with Pub/Sub locally, you would need to:

   • Set up a local Pub/Sub emulator

   • Configure the backend to use it

   • This is not typically necessary for development

Remove a Scheduled Task

Steps

1. Remove the task from gyrinx/tasks/registry.py

2. Deploy - The orphan cleanup will automatically delete the Cloud Scheduler job

Notes

• The provisioning system detects and removes orphaned scheduler jobs

• Orphan detection uses the {env}--gyrinx-scheduler-- prefix to identify managed jobs

• Jobs are only deleted if they match the current environment

Debug a Failed Task

Steps

1. Check Cloud Logging for the task execution:

   • Filter by task_name or task_id

   • Look for task_started, task_failed, task_completed events

2. Check Pub/Sub dead letter queue (if configured) for messages that exceeded retry limits

3. Reproduce locally:

# Get the arguments from the failed message
from gyrinx.core.tasks import my_task

# Call directly to see the full traceback
my_task("arg1", "arg2")

1. Common failure causes:

   • 429: Database connection pool exhausted (check CONN_MAX_AGE, connection limits)

   • 500: Unhandled exception (check task code)

   • 400: Message format issues (check enqueue arguments)