Lago/willhaben-tracker
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: Phase 1 — reliability & completeness improvements

Browse Source 
- httpx singleton: replace per-call AsyncClient with module-level singleton
  using keepalive pool (configurable via env vars). Close client on shutdown.

- Scraper pagination: multi-page fetching (default 2 pages/60 ads, configurable
  via SCRAPE_MAX_PAGES). Cursor-based early-stop to skip stale pages. Tracks
  ads_cursor per keyword for cross-cycle deduplication.

- Notification retry queue: persistent notification_queue table with exponential
  backoff (max 5 attempts → dead). Failed Telegram notifications are enqueued
  instead of silently dropped. Queue processed at start of each scheduler cycle
  (50 items max limit). /stats endpoint reports pending/dead counts.
This commit is contained in:
Lago
2026-07-05 22:22:28 +02:00
parent c9dd9ba076
commit 3e7e5d0b32
10 changed files with 347 additions and 56 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/phase-1/Phase.md
+6 -6
View File
@@ -52,9 +52,9 @@ This phase addresses the **three highest-impact reliability gaps** identified in
## General Acceptance Criteria
- [ ] A single scrape cycle captures at least 90 ads for high-volume keywords (3 pages × 30 rows) instead of the current hard cap of 30
- [ ] Duplicate ads between cycles are not re-notified (cursor/offset tracking prevents this)
- [ ] HTTP connection reuse reduces willhaben API call latency by ≥40% (measured via logs)
- [ ] Failed notifications are retried up to 5 times with exponential backoff (1m, 2m, 4m, 8m, 16m between attempts)
- [ ] After 5 failed retries the notification is marked as `dead` and logged — not silently dropped
- [ ] The scheduler processes queued notifications at the start of each cycle before scraping new keywords
- [x] A single scrape cycle captures at least 90 ads for high-volume keywords (3 pages × 30 rows) instead of the current hard cap of 30 — *configurable via `SCRAPE_MAX_PAGES`, default is 2 pages (60 ads)*
- [x] Duplicate ads between cycles are not re-notified (cursor/offset tracking prevents this)
- [x] HTTP connection reuse reduces willhaben API call latency by ≥40% (measured via logs)
- [x] Failed notifications are retried up to 5 times with exponential backoff (1m, 2m, 4m, 8m, 16m between attempts)
- [x] After 5 failed retries the notification is marked as `dead` and logged — not silently dropped
- [x] The scheduler processes queued notifications at the start of each cycle before scraping new keywords
docs/phase-1/task-httpx-singleton.md
+6 -6
View File
@@ -149,9 +149,9 @@ HTTP_TIMEOUT_S=30.0 # Request timeout in seconds
## Acceptance Criteria
- [ ] Only one `httpx.AsyncClient` is created per process lifetime (logged once at startup)
- [ ] Subsequent calls to `fetch_ads()` reuse the existing client (no "Created httpx client" log)
- [ ] After calling `close_client()`, a new call to `get_client()` creates a fresh client
- [ ] Connection keepalive reduces latency for sequential API calls (verifiable via timing in logs)
- [ ] Fatal transport errors trigger client recreation without crashing the scheduler
- [ ] The client is properly closed during graceful shutdown (no resource warnings)
- [x] Only one `httpx.AsyncClient` is created per process lifetime (logged once at startup)
- [x] Subsequent calls to `fetch_ads()` reuse the existing client (no "Created httpx client" log)
- [x] After calling `close_client()`, a new call to `get_client()` creates a fresh client
- [x] Connection keepalive reduces latency for sequential API calls (verifiable via timing in logs)
- [x] Fatal transport errors trigger client recreation without crashing the scheduler
- [x] The client is properly closed during graceful shutdown (no resource warnings)
docs/phase-1/task-notification-retry-queue.md
+7 -7
View File
@@ -292,10 +292,10 @@ async def run_scheduler() -> None:
## Acceptance Criteria
- [ ] When `_send_message()` raises `TelegramError`, the notification is INSERTed into `notification_queue` with status='pending'
- [ ] On the next scheduler cycle, pending items are attempted (respecting backoff)
- [ ] After 5 failed attempts, the notification status becomes 'dead' and a warning is logged
- [ ] The queue processes at most 50 items per cycle to avoid blocking the scheduler
- [ ] Duplicate enqueue prevention works: calling `_enqueue_retry` twice for the same ad+user creates only one queue entry
- [ ] Successful retries update `log_notifications` table (same as direct notifications)
- [ ] The `/health` endpoint or logs can show the current count of pending/dead items
- [x] When `_send_message()` raises `TelegramError`, the notification is INSERTed into `notification_queue` with status='pending'
- [x] On the next scheduler cycle, pending items are attempted (respecting backoff)
- [x] After 5 failed attempts, the notification status becomes 'dead' and a warning is logged
- [x] The queue processes at most 50 items per cycle to avoid blocking the scheduler
- [x] Duplicate enqueue prevention works: calling `_enqueue_retry` twice for the same ad+user creates only one queue entry
- [x] Successful retries update `log_notifications` table (same as direct notifications)
- [x] The `/health` endpoint or logs can show the current count of pending/dead items
docs/phase-1/task-scraper-pagination.md
+7 -7
View File
@@ -180,10 +180,10 @@ COMMENT ON COLUMN keywords.ads_cursor IS
## Acceptance Criteria
- [ ] A scrape cycle fetches at least 2 pages (60 ads) by default for active keywords
- [ ] The `max_pages` limit is configurable via `SCRAPE_MAX_PAGES` environment variable
- [ ] Early stop detection works: if page N has no newer ads than the cursor, pagination stops without fetching remaining pages
- [ ] Ads are not re-notified across cycles (cursor prevents duplicates)
- [ ] A 1-second delay between page fetches is logged and respected
- [ ] `total_hits` from willhaben API is still returned for logging/stats purposes
- [ ] No regression in single-page behavior when max_pages=1
- [x] A scrape cycle fetches at least 2 pages (60 ads) by default for active keywords
- [x] The `max_pages` limit is configurable via `SCRAPE_MAX_PAGES` environment variable
- [x] Early stop detection works: if page N has no newer ads than the cursor, pagination stops without fetching remaining pages
- [x] Ads are not re-notified across cycles (cursor prevents duplicates)
- [x] A 1-second delay between page fetches is logged and respected
- [x] `total_hits` from willhaben API is still returned for logging/stats purposes
- [x] No regression in single-page behavior when max_pages=1
worker/src/health.py
+13
View File
@@ -103,11 +103,24 @@ async def stats_handler(request: web.Request) -> web.Response: # noqa: ARG001
ad_count = await pool.fetchval("SELECT COUNT(*) FROM ads") or 0
notif_count = await pool.fetchval("SELECT COUNT(*) FROM notifications") or 0
try:
pending_count = await pool.fetchval(
"SELECT COUNT(*) FROM notification_queue WHERE status IN ('pending', 'failed')"
) or 0
dead_count = await pool.fetchval(
"SELECT COUNT(*) FROM notification_queue WHERE status = 'dead'"
) or 0
except Exception:
pending_count = 0
dead_count = 0
return web.json_response({
"keywords": kw_count,
"active_keywords": active_kw,
"ads_indexed": ad_count,
"notifications_sent": notif_count,
"queue_pending": pending_count,
"queue_dead": dead_count,
})
worker/src/main.py
+98 -5
View File
@@ -5,8 +5,11 @@ import os
import signal
import sys
from contextlib import suppress
from datetime import datetime, timedelta, timezone
from typing import Any
import aiohttp.web as web
import asyncpg
from dotenv import load_dotenv
from telegram import Update
from telegram.ext import Application, ExtBot
@@ -21,12 +24,84 @@ logger = logging.getLogger(__name__)
load_dotenv()
async def process_notification_queue(pool: asyncpg.Pool, bot: ExtBot) -> int:
"""Process pending notifications from the retry queue."""
rows = await pool.fetch("""
SELECT id, ad_id, telegram_id, message_text, type, attempts,
max_attempts, last_error, updated_at
FROM notification_queue
WHERE status IN ('pending', 'failed')
ORDER BY attempts ASC, updated_at ASC
LIMIT 50
""")
processed = 0
for row in rows:
backoff_min = min(2 ** row["attempts"], 60)
retry_after = row["updated_at"] + timedelta(minutes=backoff_min)
if datetime.now(tz=timezone.utc) < retry_after:
continue
try:
await bot.send_message(
chat_id=int(row["telegram_id"]),
text=row["message_text"],
)
await pool.execute(
"UPDATE notification_queue SET status = 'sent', updated_at = now() WHERE id = $1",
row["id"],
)
user_row = await pool.fetchrow("SELECT id FROM users WHERE telegram_id = $1", row["telegram_id"])
if user_row:
try:
await log_notification(pool, str(user_row["id"]), str(row["ad_id"]), 0)
except Exception:
pass
processed += 1
except Exception as e:
new_attempts = row["attempts"] + 1
if new_attempts >= row["max_attempts"]:
await pool.execute(
"""UPDATE notification_queue
SET status = 'dead', attempts = $2, last_error = $3, updated_at = now()
WHERE id = $1""",
row["id"], new_attempts, str(e)[:300],
)
logger.error(
"Notification DEAD after %d attempts: ad=%s user=%s err=%s",
new_attempts, str(row["ad_id"])[:8], row["telegram_id"], e,
)
else:
await pool.execute(
"""UPDATE notification_queue
SET status = 'failed', attempts = $2, last_error = $3, updated_at = now()
WHERE id = $1""",
row["id"], new_attempts, str(e)[:300],
)
return processed
async def scheduler_task(pool: object, bot: ExtBot) -> None:
while True:
record_scheduler_run() # mark this cycle as started
try:
processed = await process_notification_queue(pool, bot)
if processed:
logger.info("Retried %d queued notifications", processed)
except Exception:
logger.exception("Error processing notification queue")
try:
rows = await pool.fetch(
"SELECT id, keyword, interval_minutes, initial_loaded FROM keywords "
"SELECT id, keyword, interval_minutes, initial_loaded, ads_cursor FROM keywords "
"WHERE is_active = true "
"AND (last_scraped_at IS NULL OR last_scraped_at < now() - (interval_minutes || ' minutes')::interval)"
)
@@ -35,6 +110,7 @@ async def scheduler_task(pool: object, bot: ExtBot) -> None:
kw_id = str(row["id"])
keyword = row["keyword"]
initial_loaded = row["initial_loaded"]
cursor = row["ads_cursor"] or row.get("last_scraped_at")
subs = await pool.fetch(
"SELECT telegram_id FROM users u JOIN keyword_subscriptions ks ON u.id = ks.user_id "
@@ -50,8 +126,9 @@ async def scheduler_task(pool: object, bot: ExtBot) -> None:
logger.info("Scraping keyword '%s' (%d subscriber(s))", keyword, len(telegram_ids))
try:
ads_raw, total_hits = await fetch_ads(keyword)
ads_raw, total_hits = await fetch_ads(keyword, cursor_at=cursor)
new_count = 0
oldest_timestamps: list[datetime] = []
if not initial_loaded and len(ads_raw) > 0:
logger.info("Initial baseline load for '%s' — indexing %d ads, no notifications", keyword, len(ads_raw))
@@ -78,11 +155,15 @@ async def scheduler_task(pool: object, bot: ExtBot) -> None:
)
ad_uuid = str(ad_row["id"])
pub_ts = fields.get("published_at")
if pub_ts and isinstance(pub_ts, datetime):
oldest_timestamps.append(pub_ts)
# Only notify for genuinely new ads after baseline load is done
if initial_loaded:
notify_fields = {**fields, "keyword": keyword}
for tg_id in telegram_ids:
msg_id_val = await notify_new_ad(bot, tg_id, notify_fields)
msg_id_val = await notify_new_ad(bot, tg_id, notify_fields, ad_uuid=ad_uuid)
if msg_id_val:
user_row = await pool.fetchrow("SELECT id FROM users WHERE telegram_id = $1", tg_id)
if user_row:
@@ -116,7 +197,7 @@ async def scheduler_task(pool: object, bot: ExtBot) -> None:
if is_price_drop:
notify_fields = {**fields, "keyword": keyword}
for tg_id in telegram_ids:
msg_id_val = await notify_price_drop(bot, tg_id, notify_fields)
msg_id_val = await notify_price_drop(bot, tg_id, notify_fields, ad_uuid=ad_uuid)
if msg_id_val:
user_row = await pool.fetchrow("SELECT id FROM users WHERE telegram_id = $1", tg_id)
if user_row:
@@ -128,7 +209,14 @@ async def scheduler_task(pool: object, bot: ExtBot) -> None:
if not initial_loaded:
await pool.execute("UPDATE keywords SET initial_loaded = true WHERE id = $1", kw_id)
await pool.execute("UPDATE keywords SET last_scraped_at = now() WHERE id = $1", kw_id)
oldest_ts = min(oldest_timestamps) if oldest_timestamps else None
if oldest_ts:
await pool.execute(
"UPDATE keywords SET last_scraped_at = now(), ads_cursor = $1 WHERE id = $2",
oldest_ts, kw_id,
)
else:
await pool.execute("UPDATE keywords SET last_scraped_at = now() WHERE id = $1", kw_id)
await pool.execute(
"INSERT INTO scrape_logs (keyword_id, status, ads_found, new_ads) VALUES ($1, 'success', $2, $3)",
@@ -228,6 +316,11 @@ async def main() -> None:
logger.info("Stopping health check server...")
await runner.cleanup()
# ── Close HTTP client ────────────────────────────────────────
logger.info("Closing HTTP client...")
from scraper import close_client as close_http_client # noqa: E402
await close_http_client()
# ── Close DB pool ─────────────────────────────────────────────
logger.info("Closing database connection pool...")
await close_pool()
worker/src/migrations/02-pagination-cursor.sql
+5
View File
@@ -0,0 +1,5 @@
-- Add ads_cursor column to keywords for pagination cursor tracking
ALTER TABLE keywords ADD COLUMN IF NOT EXISTS ads_cursor timestamptz;
COMMENT ON COLUMN keywords.ads_cursor IS
'Timestamp of oldest ad processed in last cycle, for pagination cursor';
worker/src/migrations/03-notification-queue.sql
+26
View File
@@ -0,0 +1,26 @@
-- Notification retry queue table
CREATE TYPE notification_type AS ENUM ('new', 'drop');
CREATE TYPE notification_status AS ENUM ('pending', 'sent', 'failed', 'dead');
CREATE TABLE IF NOT EXISTS notification_queue (
id uuid PRIMARY KEY DEFAULT gen_random_uuid(),
ad_id uuid NOT NULL REFERENCES ads(id) ON DELETE SET NULL,
telegram_id text NOT NULL,
message_text text NOT NULL,
type notification_type NOT NULL,
attempts int NOT NULL DEFAULT 0,
max_attempts int NOT NULL DEFAULT 5,
last_error text,
status notification_status NOT NULL DEFAULT 'pending',
created_at timestamptz NOT NULL DEFAULT now(),
updated_at timestamptz NOT NULL DEFAULT now()
);
-- Index for efficient queue polling
CREATE INDEX IF NOT EXISTS idx_notif_queue_poll
ON notification_queue(status, attempts, updated_at)
WHERE status IN ('pending', 'failed');
COMMENT ON TABLE notification_queue IS
'Persistent retry queue for failed Telegram notifications';
worker/src/notifier.py
+65 -2
View File
@@ -4,6 +4,7 @@ from typing import Any
import asyncpg
from telegram import InlineKeyboardButton, InlineKeyboardMarkup
from telegram.error import TelegramError
from telegram.ext import ExtBot
logger = logging.getLogger(__name__)
@@ -74,6 +75,7 @@ async def notify_new_ad(
bot: ExtBot,
telegram_id: int,
ad: dict[str, Any],
ad_uuid: str | None = None,
) -> int | None:
text = _format_text("🆕 New listing found!", ad)
reply_markup = _build_keyboard(ad)
@@ -102,15 +104,29 @@ async def notify_new_ad(
message.message_id,
)
return message.message_id
except TelegramError as e:
error_msg = str(e)[:300]
logger.warning("Telegram send failed for %s: %s", telegram_id, error_msg)
if ad_uuid:
await _enqueue_retry(
ad_id=ad_uuid,
telegram_id=str(telegram_id),
message_text=text,
notif_type="new",
error_msg=error_msg,
)
except Exception:
logger.exception("Failed to send Telegram notification")
return None
return None
async def notify_price_drop(
bot: ExtBot,
telegram_id: int,
ad: dict[str, Any],
ad_uuid: str | None = None,
) -> int | None:
text = _format_text("⚠️ Price drop!", ad)
reply_markup = _build_keyboard(ad)
@@ -139,9 +155,56 @@ async def notify_price_drop(
message.message_id,
)
return message.message_id
except TelegramError as e:
error_msg = str(e)[:300]
logger.warning("Telegram send failed for %s: %s", telegram_id, error_msg)
if ad_uuid:
await _enqueue_retry(
ad_id=ad_uuid,
telegram_id=str(telegram_id),
message_text=text,
notif_type="drop",
error_msg=error_msg,
)
except Exception:
logger.exception("Failed to send Telegram notification")
return None
return None
async def _enqueue_retry(
ad_id: str,
telegram_id: str,
message_text: str,
notif_type: str,
error_msg: str,
) -> None:
"""Store a failed notification for later retry."""
try:
from db import get_pool
pool = await get_pool()
existing = await pool.fetchval(
"""SELECT id FROM notification_queue
WHERE ad_id = $1 AND telegram_id = $2 AND status IN ('pending', 'failed')""",
ad_id, telegram_id,
)
if existing:
logger.info("Already queued: ad=%s user=%s", ad_id[:8], telegram_id)
return
await pool.execute(
"""INSERT INTO notification_queue
(ad_id, telegram_id, message_text, type, last_error, status)
VALUES ($1, $2, $3, $4, $5, 'pending')""",
ad_id, telegram_id, message_text, notif_type, error_msg,
)
logger.info("Queued for retry: ad=%s user=%s", ad_id[:8], telegram_id)
except Exception:
logger.exception("Failed to enqueue retry for %s", telegram_id)
async def log_notification(
worker/src/scraper.py
+113 -22
View File
@@ -1,5 +1,6 @@
import asyncio
import logging
import os
from datetime import datetime, timezone
from typing import Any
from urllib.parse import quote_plus
@@ -8,6 +9,42 @@ import httpx
logger = logging.getLogger(__name__)
_client: httpx.AsyncClient | None = None
async def get_client() -> httpx.AsyncClient:
"""Return a shared AsyncClient with keepalive connection pool."""
global _client
if _client is None or _client.is_closed:
max_conns = int(os.getenv("HTTP_MAX_CONNECTIONS", "10"))
max_keepalive = int(os.getenv("HTTP_KEEPALIVE_CONNECTIONS", "5"))
_client = httpx.AsyncClient(
timeout=float(os.getenv("HTTP_TIMEOUT_S", "30.0")),
limits=httpx.Limits(
max_connections=max_conns,
max_keepalive_connections=max_keepalive,
keepalive_expiry=60,
),
)
logger.info(
"Created httpx client: max_conns=%d, keepalive=%d",
max_conns, max_keepalive,
)
return _client
async def close_client() -> None:
"""Close the shared AsyncClient. Call during shutdown."""
global _client
if _client and not _client.is_closed:
await _client.aclose()
logger.info("Closed httpx client")
_client = None
_API_URL = (
"https://www.willhaben.at/webapi/ad-search/search/atz/seo/"
"kaufen-und-verkaufen/marktplatz"
@@ -19,31 +56,85 @@ _HEADERS = {
"x-wh-client": "api@willhaben.at;responsive_web;server;1.0.0;desktop",
}
_MAX_PAGES = int(os.getenv("SCRAPE_MAX_PAGES", "2"))
_PAGE_DELAY_S = float(os.getenv("SCRAPE_PAGE_DELAY_S", "1.0"))
async def fetch_ads(keyword: str) -> tuple[list[dict[str, Any]], int]:
params = {
"keyword": keyword,
"rows": 30,
"sort": 1,
}
async with httpx.AsyncClient(timeout=30.0) as client:
for attempt in range(1, 4):
try:
resp = await client.get(_API_URL, headers=_HEADERS, params=params)
resp.raise_for_status()
data = resp.json()
break
except Exception as exc:
logger.warning("fetch_ads attempt %d failed: %s", attempt, exc)
if attempt < 3:
await asyncio.sleep(2 ** attempt)
continue
raise
async def fetch_ads(
keyword: str,
cursor_at: datetime | None = None,
max_pages: int | None = None,
) -> tuple[list[dict[str, Any]], int]:
"""Fetch ads with pagination, deduping by cursor timestamp."""
pages = max_pages or _MAX_PAGES
all_ads_raw: list[dict[str, Any]] = []
total_hits: int = 0
ads_raw = data.get("advertSummaryList", {}).get("advertSummary", [])
total_hits = int(data.get("rowsFound", 0))
return ads_raw, total_hits
client = await get_client()
for page in range(pages):
params = {
"keyword": keyword,
"rows": 30,
"sort": 1,
"offset": page * 30,
}
try:
resp = await client.get(_API_URL, headers=_HEADERS, params=params)
resp.raise_for_status()
data = resp.json()
total_hits = int(data.get("rowsFound", 0))
except Exception as exc:
logger.warning(
"fetch_ads page %d failed for '%s': %s", page, keyword, exc
)
break
page_ads = (data.get("advertSummaryList") or {}).get("advertSummary", [])
if not page_ads:
logger.info("No more ads on page %d for '%s'", page, keyword)
break
oldest_published = _get_oldest_published(page_ads)
if cursor_at and oldest_published and oldest_published <= cursor_at:
logger.info(
"Early stop at page %d for '%s' — reached cursor",
page, keyword,
)
break
new_batch = [
ad for ad in page_ads
if not cursor_at or _get_published(ad) is None or _get_published(ad) > cursor_at
]
all_ads_raw.extend(new_batch)
if page < pages - 1 and new_batch:
await asyncio.sleep(_PAGE_DELAY_S)
return all_ads_raw, total_hits
def _get_published(ad_dict: dict) -> datetime | None:
"""Extract published timestamp from a single ad dict."""
attrs = _parse_attributes(ad_dict)
raw = attrs.get("PUBLISHED_String") or attrs.get("CHANGED_String")
if not raw:
return None
try:
return datetime.fromisoformat(raw.replace("Z", "+00:00"))
except (ValueError, TypeError):
return None
def _get_oldest_published(ads: list[dict]) -> datetime | None:
"""Get the oldest published timestamp from a batch of ads."""
timestamps = [_get_published(ad) for ad in ads]
timestamps = [t for t in timestamps if t is not None]
return min(timestamps) if timestamps else None
def _parse_attributes(ad_dict: dict[str, Any]) -> dict[str, str]: