async-processing

Async Processing Skill

Expertise: Task queues (Celery, ARQ, Dramatiq), Kafka/NATS consumers, background jobs, retry strategies, idempotency, dead-letter queues.

When to Use Async Processing

Use async when:
✅ Operation takes > 200ms (send email, resize image, call slow 3rd party)
✅ Work can be retried independently (payment webhook, notification)
✅ Decoupling producers from consumers is required
✅ Fan-out to multiple consumers needed

Keep synchronous:
❌ Response depends on the result (user sees outcome immediately)
❌ Must be transactional with the triggering DB write

Task Queue (Celery + Redis)

# tasks.py
from celery import Celery
from celery.utils.log import get_task_logger

app = Celery("myapp", broker="redis://localhost:6379/1", backend="redis://localhost:6379/2")
app.conf.update(
    task_serializer="json",
    result_expires=3600,
    task_acks_late=True,          # Ack after completion, not on receive
    task_reject_on_worker_lost=True,
    task_default_retry_delay=60,  # 1 min base delay
    task_max_retries=5,
)

logger = get_task_logger(__name__)

@app.task(bind=True, max_retries=5, default_retry_delay=30)
def send_order_confirmation(self, order_id: int) -> None:
    try:
        order = Order.objects.get(id=order_id)
        email_service.send_confirmation(order)
        logger.info("email.sent", extra={"order_id": order_id})
    except EmailServiceError as exc:
        # Exponential backoff: 30s, 60s, 120s, 240s, 480s
        delay = 30 * (2 ** self.request.retries)
        raise self.retry(exc=exc, countdown=delay)
    except Order.DoesNotExist:
        logger.error("order.not_found", extra={"order_id": order_id})
        # Don't retry — data issue, not transient

Message Consumer (Kafka / aiokafka)

from aiokafka import AIOKafkaConsumer
import asyncio, json

async def consume_order_events():
    consumer = AIOKafkaConsumer(
        "orders.events",
        bootstrap_servers="kafka:9092",
        group_id="notification-service",
        auto_offset_reset="earliest",
        enable_auto_commit=False,    # Manual commit — control exactly-once
        value_deserializer=lambda v: json.loads(v.decode()),
    )
    await consumer.start()
    try:
        async for msg in consumer:
            event = msg.value
            try:
                await handle_event(event)
                await consumer.commit()           # Only commit on success
            except TransientError as e:
                logger.warning("event.retry", event_type=event["type"], error=str(e))
                await asyncio.sleep(5)            # Back off, do NOT commit
            except PermanentError as e:
                logger.error("event.dead_letter", event=event, error=str(e))
                await dead_letter_queue.publish(event)
                await consumer.commit()           # Commit to move past poison message
    finally:
        await consumer.stop()

# Idempotency — always check before processing
async def handle_event(event: dict) -> None:
    event_id = event["event_id"]
    if await redis.exists(f"processed:{event_id}"):
        return  # Already handled — skip

    await process(event)
    await redis.setex(f"processed:{event_id}", 86400, "1")