Перезапуск бота больше не повторяет рассылку

Расписание common.check в 00:35 совпало с выкаткой: старый процесс успел отправить цепочку, новый после старта отправил её ещё раз. Теперь каждая отработанная минута атомарно отмечается в bot.schedule_runs; если отметка уже есть, рассылка пропускается. Старые отметки чистятся через 7 дней.

README.md

@@ -73,7 +73,9 @@ HttpOnly-cookie (`ADMIN_SESSION_SECRET`), все формы защищены CSR
 
 Раздел «Расписания» — CRUD рассылок (`bot.schedules` + история): время, дни недели,
 таймзона, шаблон и вкл/выкл. Бот раз в минуту читает активные расписания из БД; если
-база недоступна — использует расписание из кода (совпадает с сидом `V16`).
+база недоступна — использует расписание из кода (совпадает с сидом `V16`). Каждая
+отправленная минута фиксируется в `bot.schedule_runs`, поэтому перезапуск бота в ту же
+минуту не повторяет рассылку.
 
 Раздел «Пользователи» — список с поиском по нику/username/имени/Telegram-ID и карточка:
 уровень, опыт и серия, баланс (`SUM(money.amount)`), яички, сообщения, очищения,
@@ -173,7 +175,7 @@ make backup E=test     # на домашнем сервере: ./scripts/clickho
 
 ## Версионирование
 
-- Единый источник версии — `common/version.py` (`__version__`), текущая — `0.6.8`.
+- Единый источник версии — `common/version.py` (`__version__`), текущая — `0.6.9`.
 - Схема — SemVer-подобная (`MAJOR.MINOR.PATCH`); git-тег = `v` + версия (`v0.1`).
 - Docker-образы тегируются версией (Блок 2), события ClickHouse получают `bot_version` (Блок 3).
 - Выпуск: изменить `common/version.py` → коммит → аннотированный тег `vX.Y` → `git push --tags`.

bot/services/notifications.py

@@ -12,7 +12,7 @@ import asyncio
 import logging
 import time
 from dataclasses import dataclass
-from datetime import datetime
+from datetime import UTC, datetime
 from zoneinfo import ZoneInfo
 
 from aiogram import Bot
@@ -45,6 +45,7 @@ class ScheduledMessage:
     template_key: str
     weekdays: tuple[int, ...] = ()
     timezone: str = "Europe/Moscow"
+    schedule_id: int | None = None
 
     def matches(self, now: datetime) -> bool:
         """Совпадает ли момент ``now`` (aware) с этим расписанием."""
@@ -76,6 +77,7 @@ def _from_row(row: dict) -> ScheduledMessage:
         template_key=str(row["template_key"]),
         weekdays=tuple(int(day) for day in (row["weekdays"] or ())),
         timezone=str(row["timezone"] or "Europe/Moscow"),
+        schedule_id=int(row["id"]) if row.get("id") is not None else None,
     )
 
 
@@ -114,6 +116,14 @@ async def _tick(bot: Bot, now: datetime) -> None:
     for schedule in await _load_schedules():
         if not schedule.matches(now):
             continue
+
+        # Защита от повторной отправки после перезапуска в ту же минуту.
+        if schedule.schedule_id is not None:
+            run_at = now.astimezone(UTC).replace(second=0, microsecond=0)
+            if not await schedules_repo.claim_run(schedule.schedule_id, run_at):
+                logger.info("Рассылка %s за %s уже отправлена — пропускаю", schedule.schedule_id, run_at)
+                continue
+
         subscribers = await users_service.list_subscribers()
         await _send_broadcast(bot, schedule.template_key, subscribers)
 

common/repositories/schedules_repo.py

@@ -50,6 +50,27 @@ async def list_active() -> list[dict[str, Any]]:
         return await cur.fetchall()
 
 
+async def claim_run(schedule_id: int, run_at: Any) -> bool:
+    """Атомарно отметить минуту как отработанную.
+
+    ``True`` — эту минуту ещё никто не отправлял; ``False`` — уже была отправка
+    (например, до перезапуска бота).
+    """
+    async with get_pool().connection() as conn:
+        cur = await conn.execute(
+            """
+            INSERT INTO bot.schedule_runs (schedule_id, run_at)
+            VALUES (%s, %s)
+            ON CONFLICT DO NOTHING
+            """,
+            (schedule_id, run_at),
+        )
+        claimed = bool(cur.rowcount)
+        if claimed:
+            await conn.execute("DELETE FROM bot.schedule_runs WHERE run_at < now() - interval '7 days'")
+        return claimed
+
+
 async def get_schedule(schedule_id: int) -> dict[str, Any] | None:
     async with get_pool().connection() as conn:
         cur = await conn.execute(

common/version.py

@@ -7,4 +7,4 @@
 
 from __future__ import annotations
 
-__version__ = "0.6.8"
+__version__ = "0.6.9"

sql/V25__schedule_runs.sql

@@ -0,0 +1,11 @@
+-- Отметки отправленных минут: защита от повторной рассылки после перезапуска бота.
+CREATE TABLE IF NOT EXISTS bot.schedule_runs (
+    schedule_id bigint NOT NULL,
+    run_at      timestamptz NOT NULL,
+    PRIMARY KEY (schedule_id, run_at)
+);
+
+CREATE INDEX IF NOT EXISTS idx_schedule_runs_run_at
+    ON bot.schedule_runs (run_at);
+
+COMMENT ON TABLE bot.schedule_runs IS 'Какие минуты рассылок уже отработаны (атомарная защита от дублей)';

tests/test_notifications_schedule.py

@@ -6,6 +6,7 @@ import re
 from datetime import UTC, datetime
 from pathlib import Path
 
+from bot.services import notifications
 from bot.services.notifications import DEFAULT_SCHEDULES, ScheduledMessage, _from_row
 
 SEED_FILE = Path(__file__).resolve().parents[1] / "sql" / "V16__seed_schedules.sql"
@@ -66,3 +67,58 @@ def test_default_schedules_match_seed() -> None:
         )
 
     assert seeded == list(DEFAULT_SCHEDULES)
+
+
+async def test_tick_skips_already_sent_minute(monkeypatch) -> None:
+    schedule = ScheduledMessage(12, 0, "notify.blago", schedule_id=1)
+    sent: list[str] = []
+
+    async def load() -> tuple[ScheduledMessage, ...]:
+        return (schedule,)
+
+    async def claim(schedule_id: int, run_at: datetime) -> bool:
+        return False
+
+    async def subscribers() -> list[dict]:
+        return []
+
+    async def broadcast(bot, key: str, rows: list[dict]) -> None:
+        sent.append(key)
+
+    monkeypatch.setattr(notifications, "_load_schedules", load)
+    monkeypatch.setattr(notifications.schedules_repo, "claim_run", claim)
+    monkeypatch.setattr(notifications.users_service, "list_subscribers", subscribers)
+    monkeypatch.setattr(notifications, "_send_broadcast", broadcast)
+
+    await notifications._tick(None, datetime(2026, 1, 1, 9, 0, tzinfo=UTC))
+
+    assert sent == []
+
+
+async def test_tick_sends_once_when_claimed(monkeypatch) -> None:
+    schedule = ScheduledMessage(12, 0, "notify.blago", schedule_id=1)
+    sent: list[str] = []
+    claims: list[tuple[int, datetime]] = []
+
+    async def load() -> tuple[ScheduledMessage, ...]:
+        return (schedule,)
+
+    async def claim(schedule_id: int, run_at: datetime) -> bool:
+        claims.append((schedule_id, run_at))
+        return True
+
+    async def subscribers() -> list[dict]:
+        return []
+
+    async def broadcast(bot, key: str, rows: list[dict]) -> None:
+        sent.append(key)
+
+    monkeypatch.setattr(notifications, "_load_schedules", load)
+    monkeypatch.setattr(notifications.schedules_repo, "claim_run", claim)
+    monkeypatch.setattr(notifications.users_service, "list_subscribers", subscribers)
+    monkeypatch.setattr(notifications, "_send_broadcast", broadcast)
+
+    await notifications._tick(None, datetime(2026, 1, 1, 9, 0, tzinfo=UTC))
+
+    assert sent == ["notify.blago"]
+    assert claims == [(1, datetime(2026, 1, 1, 9, 0, tzinfo=UTC))]

tests/test_version.py

@@ -7,4 +7,4 @@ from common import __version__
 
 def test_version_is_set():
     assert __version__
-    assert __version__ == "0.6.8"
+    assert __version__ == "0.6.9"