Расписание и повторные попытки
Оглавление серии «Агрегатор лент» · Урок 6 из 10.
После предыдущего урока «Сигнал» сохраняет записи и состояние источника в SQLite. Теперь сбор нужно повторять, сохраняя разумный интервал и не перегружая источник после отказа. Заменим ручной запуск одним worker. Он читает время следующей попытки из базы, обрабатывает только наступившие задания и сохраняет задержку перед повтором. Перезапуск процесса не должен превращать все источники в немедленные новые запросы.
В этом состоянии проект рассчитан на Linux или macOS: для единственного worker используется модуль fcntl. Сеть по умолчанию выключена, а учебные RSS и Atom читаются из собственных fixture. Реальный источник github-blog обрабатывается лишь с явным --network. Мы рассмотрим ожидаемое поведение подготовленного кода; worker, база и сетевые запросы при написании урока не выполнялись.
Время следующей попытки
Поле next_due хранит момент, начиная с которого источник можно снова обработать. Worker выбирает строки с next_due <= now и сортирует их по времени и идентификатору. Таким образом, расписание принадлежит базе, а не счётчику в памяти Python. Если процесс остановится после успешной фиксации, следующий запуск прочитает сохранённое будущее время и не начнёт сбор раньше него.
Успешный проход устанавливает next_due через обычный интервал из config.py, равный тридцати минутам. Это интервал после завершённой попытки, а не обещание обращения ровно в определённую минуту календаря. Длительность чтения, остановка worker и время обработки других источников могут сдвинуть реальный момент. Для небольшой локальной подборки такое расписание проще и честнее, чем утверждение о точном времени публикации каждого материала.
Флаг --once означает один проход по источникам, срок которых уже наступил. Он не означает «принудительно обновить всё». Если сразу повторить такую команду после успеха, источники окажутся в будущем и не будут читаться снова. Это важное различие между однократным запуском процесса и обходом расписания. Принудительную операцию, если она понадобится, следует проектировать отдельно, сохраняя ограничения частоты и право пользователя её вызвать.
Обычный длительный режим делает проход, ждёт пять секунд и снова проверяет SQLite. Пять секунд — частота проверки расписания, а не период сетевых обращений. Реальный HTTP-источник по-прежнему имеет получасовой next_due либо срок backoff. Смешение этих двух интервалов привело бы к ошибочному объяснению, будто worker опрашивает редакционный сайт каждые пять секунд.
Полный файл worker.py состояния шестого урока:
import argparse, fcntl, random, time
from config import BASE, SOURCES
from store import initialize, connect, transaction, collect_one
def run_due(con, network=False):
now = int(time.time())
due = con.execute("SELECT id,failures FROM sources WHERE next_due<=? ORDER BY next_due,id", (now,)).fetchall()
for state in due:
source_id = state["id"]
if "url" in SOURCES[source_id] and not network: continue
try:
count, status = collect_one(con, source_id, network)
print(source_id, status, count)
except Exception as error:
failures = min(state["failures"] + 1, 12)
delay = min(3600, 60 * 2 ** min(failures - 1, 6)) + random.randint(0, 30)
now = int(time.time())
with transaction(con):
con.execute("UPDATE sources SET failures=?,next_due=?,last_error=? WHERE id=?",
(failures,now+delay,type(error).__name__,source_id))
print(source_id, "failed", type(error).__name__)
if __name__ == "__main__":
cli = argparse.ArgumentParser()
cli.add_argument("--once", action="store_true")
cli.add_argument("--network", action="store_true")
args = cli.parse_args()
with (BASE / "signal.lock").open("a") as lock:
fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
initialize()
con = connect()
try:
while True:
run_due(con, args.network)
if args.once: break
time.sleep(5)
finally:
con.close()Повтор после отказа
Если сбор одного источника завершился исключением, worker увеличивает его failures и назначает новый next_due. Первая задержка имеет основу шестьдесят секунд, следующая — сто двадцать, затем двести сорок. Это экспоненциальный backoff: повтор становится реже при продолжающейся проблеме. Такая политика даёт источнику время восстановиться и не превращает техническую ошибку в непрерывный поток одинаковых запросов.
Основа ограничена одним часом, а случайная добавка составляет от нуля до тридцати секунд включительно. Поэтому фактическая верхняя задержка после вычисления может быть больше часа на величину добавки. Счётчик отказов ограничен двенадцатью, а степень растёт лишь до заданного предела. Эти ограничения относятся к вычислению интервала, а не к числу разрешённых попыток: worker продолжает возвращаться к отказавшему источнику позднее.
Случайный разброс называют jitter. Если несколько независимых установок потеряли доступ одновременно, строго одинаковая задержка заставила бы их снова обратиться одновременно. Разброс уменьшает такой эффект. Здесь не требуется криптографическая случайность: число не является токеном или идентификатором права доступа. Для прибавки используется обычный random.randint, который включает обе границы диапазона. Документация Python уточняет этот контракт.
Успех сбрасывает failures и last_error, обновляет last_ok и возвращает обычный интервал. Условный HTTP-ответ 304 также может быть успешным проходом, когда сохранённое представление существует. Неизменившаяся лента — нормальный результат, а не технический отказ. Число новых записей при таком проходе может быть нулевым, но проверка свежести источника всё равно состоялась.
В last_error записывается название класса исключения. Это ограничивает объём и не копирует URL, body или случайный текст библиотечной ошибки в базу диагностики. Одновременно такая простая запись не всегда объясняет причину: ValueError может обозначать HTTP-отказ, неожиданную ленту или превышение лимита. Для расширения диагностики нужен заранее определённый словарь причин, а не безусловная запись любого содержимого исключения.
Один владелец расписания
Перед инициализацией базы worker открывает signal.lock и пытается получить LOCK_EX | LOCK_NB. Эксклюзивная блокировка допускает одного сотрудничающего владельца, а неблокирующий режим приводит к отказу второго процесса вместо ожидания. Файл открыт на протяжении всего цикла, поэтому блокировка сохраняется между проходами. Python fcntl описывает платформенную доступность и значения этих флагов.
Оставшийся после завершения файл не доказывает, что worker ещё работает. Важна активная блокировка открытого дескриптора, а не наличие имени на диске. Не нужно удалять signal.lock, пытаясь «исправить» занятый worker: удаление пути может нарушить согласованный способ блокировки. Сначала выясняют, какой процесс держит работающий экземпляр и почему был запущен второй.
Механизм относится к одной локальной папке и одной согласованной точке запуска. Он не является распределённым lease для нескольких машин. Если каждый процесс использует свой lock-файл, защита между ними отсутствует. Поэтому с шестого состояния прежний collect.py удалён: нельзя оставить независимый сборщик, который меняет те же источники в обход выбранного владельца расписания. SQLite-транзакция обеспечивает согласованность записи, но не заменяет запрет параллельных HTTP-сборов.
Редакторская очередь следующего урока и read-only сервер могут работать рядом с worker по собственным правилам доступа к SQLite. Они не получают право собирать источники. При одновременном изменении база может заставить операцию ждать в рамках своего timeout; это другой механизм, чем файловая блокировка расписания. Не следует называть fcntl общей блокировкой всех пользователей базы.
Сохранённый результат и следующий проход
Для источника, уже обработанного успешно, ожидается сохранённый будущий next_due. Для отказавшего — увеличенный failures, класс причины и более близкая повторная попытка. Для HTTP-источника при выключенной сети проход пропускается без фиктивной ошибки и без утверждения об успешном обновлении. Локальные fixture остаются доступными для изучения нормализации и состояния без обращения к издателю.
При аварии во время сетевого чтения успешная транзакция не фиксируется. Новый worker после получения блокировки увидит прежнее состояние и сможет повторить попытку по расписанию. Если успешная фиксация успела завершиться, сохранённые записи и validators остаются согласованными. Именно база определяет результат перехода, а не последняя строка, которую процесс успел вывести в терминал перед остановкой.
Теперь «Сигнал» умеет управлять повторением: обычный интервал, backoff и один локальный владелец образуют связный процесс. Но автоматический сбор ещё не разрешает публикацию найденной записи. В следующем уроке мы свяжем редакторское одобрение с конкретным fingerprint, чтобы изменившийся заголовок или ссылка не унаследовали решение, принятое для прежней версии.
Полное состояние шага — папка lesson-06 архива «Сигнал». Код, сеть, база данных и учебный сервер при подготовке не запускались; описанное поведение ожидается из разбора исходников.