Несколько cron-заданий автоматизации, развёрнутых на разных хостах, время от времени запускаются почти одновременно и конкурируют за один и тот же внешний ресурс, вызывая гонки и дублирование операций. Как спроектировать распределённую блокировку и идемпотентность на уровне платформы автоматизации?

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

Нужен внешний механизм распределённой блокировки с TTL (например, на базе Redis, Postgres advisory locks или etcd/Consul) плюс идемпотентные ключи операций на стороне вызываемого ресурса, чтобы даже при неудачном захвате блокировки или её истечении во время выполнения операция не выполнилась дважды с видимым эффектом; единственно верной реализации нет — выбор хранилища для блокировок и длительности TTL зависит от уже используемой инфраструктуры и допустимого времени простоя при сбое держателя блокировки.

Как это работает подробнее

наличие нескольких хостов с cron исключает использование локальных файловых блокировок (flock) как единственной защиты, так как они не видны между хостами; распределённая блокировка с TTL решает проблему "зависшего" держателя лока (если хост с блокировкой упал, она автоматически освободится по истечении TTL), но вводит риск двойного выполнения, если операция не уложилась в TTL, а блокировка освободилась раньше её завершения.

более длинный TTL снижает риск параллельного запуска, но увеличивает время простоя автоматизации при аварийном падении держателя блокировки до истечения TTL; добавление heartbeat/продления решает это, но усложняет реализацию и требует отдельной обработки сценария "держатель жив, но завис". Выбор между простым TTL-локом и более сложной схемой с heartbeat должен определяться тем, насколько критично время восстановления после сбоя конкретно для этой автоматизации.

Пример кода

example-489.pyКопировать

import redis
import uuid
r = redis.Redis(host="locks.internal", port=6379)
def run_with_distributed_lock(lock_key, idempotency_key, ttl_seconds, task_fn):
    token = str(uuid.uuid4())
    acquired = r.set(lock_key, token, nx=True, ex=ttl_seconds)
    if not acquired:
        print(f"Блокировка {lock_key} уже удержана другим узлом, пропускаем запуск")
        return
    try:
        if r.set(f"done:{idempotency_key}", "1", nx=True, ex=86400):
            task_fn()
        else:
            print(f"Операция {idempotency_key} уже была выполнена ранее, пропускаем")
    finally:
        if r.get(lock_key) == token.encode():
            r.delete(lock_key)

Что использовать на практике

• Рекомендация. надёжная схема комбинирует две независимые защиты — распределённую блокировку как первый, "быстрый" барьер против одновременного старта, и идемпотентный ключ операции (например, детерминированно вычисленный из бизнес-данных: дата + идентификатор задачи) как второй, окончательный барьер против дублирования эффекта, даже если блокировка почему-то не сработала. TTL блокировки стоит выбирать с запасом относительно максимального ожидаемого времени выполнения операции, и дополнительно продлевать его (heartbeat), если задача выполняется дольше ожидаемого.

Подводные камни

• Ошибка. полагаться только на flock локального файла при развёртывании задачи на нескольких хостах; выставлять TTL блокировки заведомо меньше реального времени выполнения операции без механизма продления, из-за чего блокировка освобождается до завершения работы и второй узел стартует параллельно; реализовывать идемпотентный ключ на основе времени запуска (например, текущей секунды) вместо стабильного бизнес-идентификатора, из-за чего он не защищает от повторного запуска.