Ausgangslage
Mein Watch-&-Alert-Setup sammelt aktuell 87 RSS- und Atom-Feeds, primär aus den Bereichen IT-Security, Open-Source-Releases und ein paar Branchendienste. Dazu kommen vier Webhooks von GitHub, einer von Sentry und ein eigener Heartbeat-Endpunkt. Im Schnitt laufen zwischen 12 und 40 Alerts pro Tag durch, an Spitzentagen bei CVE-Veröffentlichungen oder größeren Releases schnell auch mal 200 innerhalb weniger Minuten.
Der Versand läuft seit Monaten sauber. Klassifiziert wird lokal über Ollama mit einem kleinen qwen2.5:7b, nur wenn der Ollama-Host nicht erreichbar ist, fällt das System auf ein Cloud-Modell zurück. SQLite übernimmt die Persistenz, ein systemd-Timer feuert alle zwei Minuten den nächsten Lauf an. Der gesamte Versand an Telegram lief bislang synchron im Hauptprozess: Alert eingetroffen, LLM fertig, requests.post an die Telegram-API, fertig.
Dann kam der Dienstag. Ein großes Tech-Unternehmen veröffentlichte eine Tool-Suite, ein populäres Open-Source-Projekt ein Major-Release, parallel ein HSE-Newsletter mit Sammelmeldungen. Innerhalb von vier Minuten hagelte es 28 Alerts. Telegram meldete für einen Teil davon HTTP 429, der Rest ging durch. In meinem Kanal fehlten am Ende sieben Nachrichten — nicht weil das System sie nicht erkannt hatte, sondern weil der Versand im Feuer der Bursts an den Limits zerschellt war. Das System selbst hatte keinen Fehler. Es war schlecht entworfen für genau diesen Fall.
Anforderungen
Bevor ich Code schreibe, schreibe ich mir auf, was die Lösung können muss. Das spart Nacharbeit und verhindert, dass ich später strikt nach Lehrbuch etwas baue, was in meinem Setup gar nicht passt. In unserem Fall sind die Anforderungen an einen vernünftigen Versand konkret:
- Kein Alert darf verloren gehen. Wenn Telegram einen 429 liefert, muss die Nachricht warten und es später erneut versuchen, statt sie zu verwerfen.
- Reihenfolge muss nicht zwingend gewahrt bleiben. Wenn ein älterer Alert einen Moment warten muss, damit ein neuerer schneller raus kann, ist das im Monitoring in Ordnung.
- Backoff mit Jitter. Telegram schickt einen
Retry-After-Header, aber nicht immer zuverlässig. Ein exponentielles Backoff mit zufälligem Anteil verhindert, dass alle wartenden Nachrichten gleichzeitig wieder zuschlagen. - Persistente Queue. Wenn das System während eines Bursts crasht oder neu startet, dürfen bereits gesammelte Alerts nicht futsch sein. SQLite ist eh da, also nutze ich es.
- Heartbeat bleibt sichtbar. Das Watch-&-Alert-Heartbeat-Pattern soll weiter laufen, also muss der Versand so gestaltet sein, dass ein hängender Worker den Heartbeat nicht ausbremst.
- Kein neues Framework. Kein Celery, kein Redis, kein RabbitMQ. Wir bleiben bei
asynciound SQLite.
Die Liste ist ehrlich, nicht vollständig. Ich habe bewusst Punkte weggelassen, die in Enterprise-Setups stehen würden, etwa Multi-Region-Failover oder SLOs für die Versand-Latenz. Das ist ein Indie-Projekt, kein Zahlungsanbieter.
Drei Optionen im Vergleich
Bevor ich auf die Lösung eingehe, lohnt sich der Blick auf die Alternativen, die ich ernsthaft erwogen habe. Drei Optionen standen am Ende auf dem Zettel, eine ist geblieben.
Option 1: Naiv drosseln. Der einfachste Weg wäre, nach jedem sendMessage einfach eine Sekunde zu warten. Pro Nachricht eine Sekunde, fertig. Der Vorteil: kein Code, keine Queue, fünf Minuten Arbeit. Der Nachteil: Bei 28 Alerts in vier Minuten reden wir über 28 Sekunden Stau, und der nächste Burst verlängert das linear. In meinem Fall ist das System single-threaded — wenn der Versand 28 Sekunden blockiert, läuft der nächste systemd-Lauf ins Leere, der Heartbeat verspätet sich, und die LLM-Pipeline staut sich auf. Das ist keine Lösung, das ist ein workaroundter Workaround.
Option 2: Coalescing pro Channel. Eine andere Idee wäre, mehrere Alerts zu einer Sammelnachricht zusammenzufassen. Telegram hat ein Limit von 4096 Zeichen pro Nachricht, das reicht locker für zehn kurze Alerts. Das reduziert das Nachrichtenvolumen drastisch und ist ehrlich gesagt die eleganteste Lösung, wenn du viele kleine Meldungen hast. Der Nachteil: Du verlierst die Granularität. Wenn ein einzelner Alert klickbar sein soll, verschwindet diese Eigenschaft im Sammelposten. Außerdem funktioniert Coalescing schlecht für lange Markdown-Alerts, wie sie nach der LLM-Klassifikation entstehen. Ich habe Coalescing als zweite Verteidigungslinie behalten, aber nicht als primären Schutz.
Option 3: Dedizierte Versand-Queue mit Backoff. Die Lösung, die ich am Ende gebaut habe, ist ein eigener Versand-Worker, der Alerts aus einer SQLite-Tabelle abholt und mit Backoff an Telegram schickt. Pro Chat ein Token-Bucket-Limiter, der die Sendefrequenz konsequent einhält. Das ist mehr Code als Option 1, aber strukturell sauber. Der Hauptprozess schreibt nur noch in die Queue, kümmert sich nicht mehr um HTTP-Statuscodes und kann sofort zum nächsten Feed weitergehen. Das ist auch die Variante, die im Buch in Kapitel 13 unter „Versand entkoppeln" beschrieben wird.
Die Lösung
Der Umbau besteht aus drei Bausteinen: einer neuen SQLite-Tabelle, einem Worker, und einem schmalen Wrapper um die Telegram-API. Ich zeige dir die Teile, die wirklich zählen.
Zuerst die Tabelle. Sie ist bewusst einfach gehalten, weil SQLite keine Hochlast-Queue braucht.
import sqlite3
from pathlib import Path
DB_PATH = Path("/var/lib/watchalert/state.db")
SCHEMA = """
CREATE TABLE IF NOT EXISTS outbox (
id INTEGER PRIMARY KEY AUTOINCREMENT,
chat_id TEXT NOT NULL,
payload TEXT NOT NULL,
attempts INTEGER NOT NULL DEFAULT 0,
next_attempt REAL NOT NULL,
created_at REAL NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_outbox_ready
ON outbox(next_attempt);
"""
def init_outbox(db_path: Path = DB_PATH) -> None:
with sqlite3.connect(db_path) as conn:
conn.executescript(SCHEMA)
next_attempt speichert einen Unix-Timestamp. Der Index erlaubt dem Worker, in einem Schwung alle bereitstehenden Nachrichten zu holen. attempts zählt die Versuche für spätere Diagnose.
Der Enqueue-Schritt ersetzt den früheren direkten requests.post. Er läuft dort, wo vorher der synchrone Versand saß.
import json
import time
import sqlite3
from typing import Any
def enqueue_alert(
conn: sqlite3.Connection,
chat_id: str,
payload: dict[str, Any],
) -> None:
conn.execute(
"INSERT INTO outbox (chat_id, payload, next_attempt, created_at) "
"VALUES (?, ?, ?, ?)",
(chat_id, json.dumps(payload), time.time(), time.time()),
)
Der Worker ist das Herzstück. Er läuft als asyncio-Task, pollt die Outbox, schickt Nachrichten mit Limit und Backoff und respektiert Retry-After. Ich nutze aiohttp statt requests, damit die Wartezeiten andere Aufgaben nicht blockieren.
import asyncio
import json
import logging
import random
import sqlite3
import time
from dataclasses import dataclass
from pathlib import Path
import aiohttp
logger = logging.getLogger("watchalert.sender")
PER_CHAT_INTERVAL = 1.05 # Telegram: ~1 Nachricht/Sekunde/Chat
MAX_ATTEMPTS = 8
@dataclass
class Job:
id: int
chat_id: str
payload: dict
attempts: int
async def send_one(
session: aiohttp.ClientSession,
token: str,
job: Job,
) -> tuple[bool, float]:
"""Return (ok, retry_after_seconds)."""
url = f"https://api.telegram.org/bot{token}/sendMessage"
try:
async with session.post(url, json=job.payload, timeout=10) as resp:
if resp.status == 200:
return True, 0.0
if resp.status == 429:
data = await resp.json()
retry_after = float(data.get("parameters", {}).get(
"retry_after", 1.0
))
return False, retry_after
text = await resp.text()
logger.warning("telegram %s: %s", resp.status, text[:200])
return False, 5.0
except (aiohttp.ClientError, asyncio.TimeoutError) as exc:
logger.warning("telegram transport error: %s", exc)
return False, 5.0
def backoff_delay(attempts: int, base: float = 2.0) -> float:
"""Exponential backoff mit Jitter, gedeckelt auf 5 Minuten."""
raw = min(base ** attempts, 300.0)
return raw * (0.5 + random.random())
async def worker(
db_path: Path,
token: str,
last_seen: dict[str, float],
) -> None:
async with aiohttp.ClientSession() as session:
while True:
with sqlite3.connect(db_path) as conn:
rows = conn.execute(
"SELECT id, chat_id, payload, attempts "
"FROM outbox WHERE next_attempt <= ? "
"ORDER BY next_attempt LIMIT 20",
(time.time(),),
).fetchall()
for row in rows:
job = Job(
id=row[0],
chat_id=row[1],
payload=json.loads(row[2]),
attempts=row[3],
)
wait = last_seen.get(job.chat_id, 0.0)
delta = time.time() - wait
if delta < PER_CHAT_INTERVAL:
await asyncio.sleep(PER_CHAT_INTERVAL - delta)
if job.attempts >= MAX_ATTEMPTS:
logger.error(
"drop alert id=%s after %s attempts",
job.id, job.attempts,
)
with sqlite3.connect(db_path) as conn:
conn.execute(
"DELETE FROM outbox WHERE id = ?",
(job.id,),
)
continue
ok, retry_after = await send_one(session, token, job)
if ok:
with sqlite3.connect(db_path) as conn:
conn.execute(
"DELETE FROM outbox WHERE id = ?",
(job.id,),
)
last_seen[job.chat_id] = time.time()
continue
delay = max(retry_after, backoff_delay(job.attempts))
next_at = time.time() + delay
with sqlite3.connect(db_path) as conn:
conn.execute(
"UPDATE outbox "
"SET attempts = attempts + 1, next_attempt = ? "
"WHERE id = ?",
(next_at, job.id),
)
last_seen[job.chat_id] = time.time()
logger.info(
"retry alert id=%s in %.1fs (attempt %s)",
job.id, delay, job.attempts + 1,
)
await asyncio.sleep(1.0)
Die last_seen-Map ist der pro-Chat-Limiter. Sie sorgt dafür, dass selbst wenn 30 Alerts in der Queue stehen, immer nur einer pro Sekunde und Chat rausgeht. Der Jitter im Backoff verhindert, dass alle Nachrichten, die Telegram mit dem gleichen 429 abgewiesen hat, eine Sekunde später synchron wieder anklopfen. Das ist ein klassisches Anti-Thundering-Herd-Pattern.
Die Anbindung ans Watch-&-Alert-Hauptsystem ist ein Zweizeiler. Statt requests.post(...) ruft der Dispatcher jetzt enqueue_alert(...). In unserem Codepfad sieht das so aus:
def dispatch(
conn: sqlite3.Connection,
chat_id: str,
text: str,
) -> None:
enqueue_alert(
conn,
chat_id,
{"chat_id": chat_id, "text": text, "parse_mode": "Markdown"},
)Der Worker läuft als eigener systemd-Service. Das ist wichtig, weil er eine andere Lebensdauer hat als der Hauptloop. Der Hauptloop feuert alle zwei Minuten, der Sender läuft kontinuierlich. In der Praxis löse ich das mit zwei Timer-Units, aber das gehört eher in einen eigenen Artikel.
Was wir gelernt haben
Erkenntnis 1: HTTP 429 ist nicht optional. Wenn du einen Telegram-Bot für etwas anderes als reine Spielerei nutzt, wirst du das Limit treffen. Die Frage ist nicht ob, sondern wann. Bursts sind der Normalfall, nicht die Ausnahme.
Erkenntnis 2: Pro-Chat-Limiter schlagen globale Limiter. Mein erster Versuch war ein einfacher „eine Nachricht pro 1,2 Sekunden" ohne Chat-Bezug. Das hat funktioniert, bis ich einen zweiten Channel hinzugefügt habe. Dann wurden 0,6 Nachrichten pro Sekunde global drosselt, und der zweite Channel bekam Alerts mit ungewollter Verzögerung. Pro-Chat-State ist nicht nice to have, sondern Pflicht.
Erkenntnis 3: Coalescing hilft, aber ist kein Ersatz für eine Queue. Ich habe es als zusätzliche Schicht behalten. Pro Channel fasse ich jetzt alle 30 Sekunden ausstehende Alerts zu maximal drei Nachrichten zusammen. Das reduziert das Volumen, ohne die Queue zu ersetzen.
Erkenntnis 4: Persistenz schlägt In-Memory. Ich hatte kurz mit einer asyncio.Queue geliebäugelt. Das wäre schneller gewesen, aber ein Crash mitten im Burst hätte alle wartenden Alerts gekostet. SQLite hat hier den Vorteil, dass die Daten ohnehin schon da sind.
Erkenntnis 5: Jitter ist kein Kosmetik. Ohne den zufälligen Anteil im Backoff habe ich reproduzierbar beobachten können, dass Telegram nach einem 429 eine kurze Welle von 429-Antworten zurückgibt, wenn alle wartenden Nachrichten im selben 100-Millisekunden-Fenster erneut anfragen. Mit Jitter ist das verschwunden.
Bezug zum Buch
Kapitel 13 des Buches behandelt das Versenden der klassifizierten Daten über die drei Ausgabekanäle Telegram, ntfy.sh und E-Mail. Genau dort ist in der zweiten Auflage der Hinweis gewandert, dass der Versand asynchron und entkoppelt vom Klassifikations-Loop laufen muss — eben weil Telegram je nach Auslastung Limits setzt. Der Outbox-Mechanismus, den ich hier zeige, ist die direkte Umsetzung der Empfehlung aus dem Buch, ergänzt um die konkreten Zahlen aus meinem Setup. Wer das Buch liest, bekommt die Idee; wer den Code hier sieht, bekommt die Vorlage zum Umschreiben.
Wann du das nachbauen solltest — und wann nicht
Der Aufwand lohnt sich, sobald du regelmäßig Bursts hast. Konkret: Wenn dein Setup mehr als drei Alerts pro Minute an einen einzelnen Chat schickt, irgendwann an einem Tag. Wenn du mehrere Kanäle parallel bedienst, schon deutlich früher. Wenn du Webhooks von Drittanbietern einbindest, deren Volumen du nicht kontrollierst — bei mir ist das GitHub —, ist die Frage sowieso keine Frage mehr.
Wann nicht? Wenn du weniger als zehn Alerts pro Tag hast und ein einzelner Channel reicht, dann reicht ein time.sleep(1.1) zwischen den Nachrichten. Wenn du ein Multi-Tenant-Produkt mit Hunderten Kunden baust, ist aiohttp plus SQLite die falsche Basis, dann reden wir über echte Message-Broker. Für den typischen Indie-Stack aus RSS, ein paar Webhooks und SQLite ist die hier gezeigte Lösung aber genau das richtige Maß: ehrlich komplex, nicht übermotorisiert, gut testbar und vor allem robust gegen den Dienstag, an dem plötzlich alle gleichzeitig schreien.