Build a Notification Service from Scratch — Part 3: Retries, Rate Limiting, and Monitoring

# app/rate_limiter.py
from app.core.cache import redis_client
import time

CHANNEL_LIMITS = {
    "email": 50,   # per second
    "sms": 10,     # per second
    "push": 100    # per second
}

def check_rate_limit(channel: str) -> bool:
    """Returns True if within rate limit, False if exceeded."""
    limit = CHANNEL_LIMITS.get(channel, 10)
    key = f"rate_limit:{channel}:{int(time.time())}"
    current = redis_client.incr(key)
    redis_client.expire(key, 2)  # Auto-cleanup after 2 seconds
    return current <= limit

@celery_app.task(bind=True, max_retries=3)
def deliver_notification(self, notification_id: int):
    # ... load notification ...
    if not check_rate_limit(notification.channel):
        raise self.retry(countdown=5)  # Wait and retry
    # ... dispatch ...

def move_to_dlq(notification_id: int, error: str):
    """Move failed notification to dead letter queue."""
    redis_client.lpush("dlq", json.dumps({
        "notification_id": notification_id,
        "error": str(error),
        "moved_at": datetime.now(timezone.utc).isoformat()
    }))

def process_dlq():
    """View dead letter queue items for manual inspection."""
    items = redis_client.lrange("dlq", 0, -1)
    return [json.loads(item) for item in items]

def retry_from_dlq(notification_id: int):
    """Manually retry a notification from the DLQ."""
    # Remove from DLQ and re-enqueue
    redis_client.lrem("dlq", 0, f'"notification_id":{notification_id}')
    enqueue_notification(notification_id)

# app/metrics.py
from prometheus_client import Counter, Histogram, Gauge, generate_latest
from fastapi import Response

notifications_sent = Counter(
    "notifications_sent_total",
    "Total notifications sent",
    ["channel", "status"]
)

delivery_duration = Histogram(
    "notification_delivery_seconds",
    "Time to deliver a notification",
    ["channel"]
)

queue_size = Gauge(
    "notification_queue_size",
    "Number of notifications in queue",
    ["channel"]
)

@app.get("/metrics")
def metrics():
    return Response(content=generate_latest(), media_type="text/plain")

@celery_app.task(bind=True, max_retries=3)
def deliver_notification(self, notification_id: int):
    start = time.time()
    # ... deliver ...
    duration = time.time() - start

    notifications_sent.labels(
        channel=notification.channel,
        status=notification.status
    ).inc()
    delivery_duration.labels(channel=notification.channel).observe(duration)

# prometheus/rules.yml
groups:
  - name: notification_alerts
    rules:
      - alert: HighFailureRate
        expr: rate(notifications_sent_total{status="failed"}[5m]) > 0.1
        annotations:
          summary: "Notification failure rate > 10%"

      - alert: DLQGrowing
        expr: dlq_size > 100
        annotations:
          summary: "Dead letter queue has {{ $value }} items"

      - alert: DeliveryLatency
        expr: histogram_quantile(0.95, notification_delivery_seconds) > 5
        annotations:
          summary: "P95 delivery latency > 5 seconds"