> Как работает периодический таск, который читает таблицу уведомлений и отправляет их? (Python)

Короткий ответ

Периодический таск - это фоновый процесс, который запускается по расписанию (через Celery beat, APScheduler или cron), выбирает из таблицы уведомлений записи со статусом "pending" и отправляет их через нужный канал (email, push, SMS). Ключевые моменты: атомарная выборка с блокировкой строк, обработка ошибок с ретраями, идемпотентность отправки и защита от повторной обработки при параллельных инстансах.

Подробное объяснение

Основная задача такого таска - надёжно доставить уведомления, не потеряв их и не отправив дважды. Архитектурно это выглядит так:

1. Планировщик - Celery beat или cron запускает задачу с фиксированным интервалом (например, каждые 30 секунд).
2. Выборка - таск выполняет SQL-запрос с условием WHERE status = 'pending' и LIMIT N, где N - batch size. Важно использовать SELECT ... FOR UPDATE SKIP LOCKED (PostgreSQL) или атомарный UPDATE ... RETURNING, чтобы несколько параллельных воркеров не взяли одни и те же строки.
3. Обновление статуса - перед отправкой строки помечаются как processing (или сразу sent после успеха). Это защищает от повторной обработки при падении процесса.
4. Отправка - для каждой записи вызывается соответствующий провайдер (SMTP, push-сервис, SMS-шлюз). Ошибки сети или таймауты обрабатываются отдельно.
5. Финализация - при успехе статус меняется на sent, при неудаче - на failed или retry с счётчиком попыток и временем следующей попытки.
6. Ретраи - для временных ошибок используется exponential backoff: next_attempt_at = now + base_delay * 2^attempts.

Важный trade-off: частота запуска таска vs задержка доставки. Если интервал 1 минута, уведомление может уйти с задержкой до 60 секунд. Для критичных уведомлений лучше использовать очередь (RabbitMQ, SQS) вместо polling, но для простых систем периодический таск - достаточное решение.

На практике

На практике нужно учитывать несколько вещей:

• Batch size - не стоит тянуть все pending-записи разом, особенно при большом объёме. Лучше брать порциями по 100-500 штук.
• Мониторинг - важно логировать количество обработанных, успешных и упавших уведомлений, а также время выполнения таска.
• Dead letter - записи, превысившие максимальное число попыток, должны уходить в отдельную таблицу или помечаться как dead, чтобы не блокировать очередь.
• Индексы - обязателен индекс по (status, next_attempt_at), иначе выборка будет сканировать всю таблицу.
• Транзакции - статус processing и сама отправка не должны быть в одной транзакции, иначе при долгой отправке блокируются строки и растёт число конфликтов.

Пример кода

PYTHON
from celery import shared_task
from sqlalchemy import text
from sqlalchemy.orm import Session

BATCH_SIZE = 100
MAX_ATTEMPTS = 5

@shared_task(bind=True, max_retries=3, default_retry_delay=60)
def process_notifications(self):
    db: Session = get_db_session()
    
    # Атомарно забираем batch строк в статусе pending
    rows = db.execute(
        text("""
            UPDATE notifications
            SET status = 'processing'
            WHERE id IN (
                SELECT id FROM notifications
                WHERE status = 'pending'
                  AND next_attempt_at <= now()
                ORDER BY created_at
                LIMIT :batch_size
                FOR UPDATE SKIP LOCKED
            )
            RETURNING id, channel, payload
        """),
        {"batch_size": BATCH_SIZE}
    ).fetchall()
    
    for row in rows:
        try:
            send_notification(row.channel, row.payload)
            db.execute(
                text("UPDATE notifications SET status = 'sent', sent_at = now() WHERE id = :id"),
                {"id": row.id}
            )
        except Exception as exc:
            db.execute(
                text("""
                    UPDATE notifications
                    SET status = 'retry',
                        attempts = attempts + 1,
                        next_attempt_at = now() + make_interval(secs => power(2, attempts) * 30)
                    WHERE id = :id
                """),
                {"id": row.id}
            )
            if row.attempts >= MAX_ATTEMPTS:
                db.execute(
                    text("UPDATE notifications SET status = 'dead' WHERE id = :id"),
                    {"id": row.id}
                )
    
    db.commit()

Как отвечать на собеседовании

Начните с краткого описания схемы: планировщик → выборка → отправка → обновление статуса. Затем обязательно упомяните проблему конкурентности и способы её решения - это главный сигнал senior-уровня. Расскажите про FOR UPDATE SKIP LOCKED или атомарный UPDATE ... RETURNING, объясните, зачем нужен статус processing. Потом переходите к обработке ошибок: ретраи, exponential backoff, dead letter. В конце упомяните мониторинг и индексы - это показывает, что вы думаете о продакшене. Если спросят про альтернативы - скажите, что для низкой задержки лучше очередь, но polling проще и надёжнее для небольших объёмов.

Что проверяет интервьюер

Интервьюер оценивает:

• Понимание проблем конкурентного доступа к базе данных.
• Умение проектировать надёжные фоновые процессы с учётом отказов.
• Знание паттернов обработки ошибок (retry, backoff, dead letter).
• Понимание trade-off между polling и event-driven подходами.
• Внимание к деталям: индексы, batch size, транзакции, мониторинг.

Типичные ошибки

• Выборка без блокировки строк - при нескольких воркерах одно уведомление отправляется дважды.
• Отправка внутри транзакции с обновлением статуса - при долгой отправке блокируются строки и растёт нагрузка на БД.
• Отсутствие ограничения на число попыток - битые записи вечно висят в очереди.
• Нет индекса по (status, next_attempt_at) - таск деградирует при росте таблицы.
• Слишком частый запуск таска без batch - лишняя нагрузка на БД.
• Игнорирование идемпотентности - повторная отправка при ретрае без проверки, было ли уже отправлено.