From 2c5b83268b7063085d3c0f15160f2b499068d94e Mon Sep 17 00:00:00 2001 From: bot-backend Date: Sun, 31 May 2026 19:50:49 +0300 Subject: [PATCH] =?UTF-8?q?feat(tradein):=20systemd=20timer=20scheduler=20?= =?UTF-8?q?=E2=80=94=20trigger=20script=20+=20units=20+=20SCHEDULER=5FENAB?= =?UTF-8?q?LE=20flag=20(#581)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- tradein-mvp/backend/app/core/config.py | 6 + tradein-mvp/backend/app/main.py | 26 ++- tradein-mvp/ops/systemd/README.md | 57 ++++++ tradein-mvp/ops/systemd/install.sh | 12 ++ .../systemd/tradein-scrape-trigger.service | 15 ++ .../ops/systemd/tradein-scrape-trigger.timer | 11 + tradein-mvp/scripts/trigger-schedules.py | 188 ++++++++++++++++++ 7 files changed, 305 insertions(+), 10 deletions(-) create mode 100644 tradein-mvp/ops/systemd/README.md create mode 100644 tradein-mvp/ops/systemd/install.sh create mode 100644 tradein-mvp/ops/systemd/tradein-scrape-trigger.service create mode 100644 tradein-mvp/ops/systemd/tradein-scrape-trigger.timer create mode 100644 tradein-mvp/scripts/trigger-schedules.py diff --git a/tradein-mvp/backend/app/core/config.py b/tradein-mvp/backend/app/core/config.py index 6b15943c..b1b92443 100644 --- a/tradein-mvp/backend/app/core/config.py +++ b/tradein-mvp/backend/app/core/config.py @@ -12,6 +12,12 @@ class Settings(BaseSettings): # required — задаётся через env DATABASE_URL. Нет дефолта: fail-fast при старте # если переменная не задана (C-3 security audit). database_url: str + + # In-app asyncio scheduler enable flag (#581). + # Set SCHEDULER_ENABLE=false when using the systemd timer trigger instead. + # Default true preserves existing behaviour. + scheduler_enable: bool = Field(default=True, validation_alias="SCHEDULER_ENABLE") + cors_origins: list[str] = ["http://localhost", "http://localhost:3000", "http://localhost:8080"] environment: str = "dev" diff --git a/tradein-mvp/backend/app/main.py b/tradein-mvp/backend/app/main.py index c46176e1..ef797fcc 100644 --- a/tradein-mvp/backend/app/main.py +++ b/tradein-mvp/backend/app/main.py @@ -74,17 +74,23 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]: except Exception: logger.exception("FDW user mapping bootstrap failed — cadastral queries may fail") - # Startup: launch scheduler background task - scheduler_task = asyncio.create_task(scheduler_loop()) - logger.info("FastAPI lifespan: scheduler task spawned") + # Startup: launch scheduler background task (disabled when SCHEDULER_ENABLE=false, + # e.g. when using the systemd timer trigger — see tradein-mvp/ops/systemd/ #581). + scheduler_task = None + if settings.scheduler_enable: + scheduler_task = asyncio.create_task(scheduler_loop()) + logger.info("FastAPI lifespan: scheduler task spawned") + else: + logger.info("FastAPI lifespan: in-app scheduler disabled (SCHEDULER_ENABLE=false)") yield - # Shutdown: cancel scheduler - scheduler_task.cancel() - try: - await scheduler_task - except asyncio.CancelledError: - pass - logger.info("FastAPI lifespan: scheduler cancelled cleanly") + # Shutdown: cancel scheduler if it was started + if scheduler_task is not None: + scheduler_task.cancel() + try: + await scheduler_task + except asyncio.CancelledError: + pass + logger.info("FastAPI lifespan: scheduler cancelled cleanly") app = FastAPI( diff --git a/tradein-mvp/ops/systemd/README.md b/tradein-mvp/ops/systemd/README.md new file mode 100644 index 00000000..36268ffc --- /dev/null +++ b/tradein-mvp/ops/systemd/README.md @@ -0,0 +1,57 @@ +# TradeIn systemd scrape trigger + +Replaces the in-app asyncio scheduler with a systemd timer that calls the admin API every 60s. + +## Install + +Run as root on the VPS after deploy: + +```bash +cd /opt/gendesign/tradein-mvp/ops/systemd +bash install.sh +``` + +`install.sh` copies the unit files to `/etc/systemd/system/`, reloads systemd, and +enables + starts the timer. + +## Environment variables + +Create `/etc/tradein/trigger.env` before enabling the timer: + +```ini +# psycopg v3 connection string (postgresql://... or postgresql+psycopg://...) +TRADEIN_DATABASE_URL=postgresql://tradein:@localhost:5432/tradein + +# Admin username from roles.yaml with role=admin — passed as X-Authenticated-User header +TRADEIN_ADMIN_TOKEN=admin + +# Backend base URL (no trailing slash). Default: http://localhost:8000 +TRADEIN_API_BASE=http://localhost:8000 +``` + +## Diagnostic commands + +```bash +# Follow live timer logs +journalctl -u tradein-scrape-trigger -f + +# List all active timers and next trigger times +systemctl list-timers + +# Check timer status +systemctl status tradein-scrape-trigger.timer + +# Check last service run +systemctl status tradein-scrape-trigger.service + +# Manual one-shot trigger (for testing) +systemctl start tradein-scrape-trigger.service +``` + +## Notes + +- The script uses `flock` on `/var/run/tradein-trigger.lock` to prevent overlapping + runs if the previous invocation is still running. +- Sources without a dedicated admin endpoint (e.g. `rosreestr_dkp_import`, + `listing_source_snapshot`) are logged as warnings and skipped. Set + `SCHEDULER_ENABLE=true` in the backend env to handle them via the in-app scheduler. diff --git a/tradein-mvp/ops/systemd/install.sh b/tradein-mvp/ops/systemd/install.sh new file mode 100644 index 00000000..3c318346 --- /dev/null +++ b/tradein-mvp/ops/systemd/install.sh @@ -0,0 +1,12 @@ +#!/bin/bash +# Run as root on VPS after deploy. +# Installs (or updates) the systemd timer that triggers scrape schedules every 60s. +set -euo pipefail + +SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" + +cp "$SCRIPT_DIR/tradein-scrape-trigger.service" /etc/systemd/system/ +cp "$SCRIPT_DIR/tradein-scrape-trigger.timer" /etc/systemd/system/ +systemctl daemon-reload +systemctl enable --now tradein-scrape-trigger.timer +systemctl status tradein-scrape-trigger.timer diff --git a/tradein-mvp/ops/systemd/tradein-scrape-trigger.service b/tradein-mvp/ops/systemd/tradein-scrape-trigger.service new file mode 100644 index 00000000..15ea6fef --- /dev/null +++ b/tradein-mvp/ops/systemd/tradein-scrape-trigger.service @@ -0,0 +1,15 @@ +[Unit] +Description=TradeIn Scrape Schedule Trigger +After=network-online.target docker.service + +[Service] +Type=oneshot +User=root +EnvironmentFile=/etc/tradein/trigger.env +ExecStart=/usr/bin/python3 /opt/gendesign/tradein-mvp/scripts/trigger-schedules.py +StandardOutput=journal +StandardError=journal +TimeoutStartSec=55 + +[Install] +WantedBy=multi-user.target diff --git a/tradein-mvp/ops/systemd/tradein-scrape-trigger.timer b/tradein-mvp/ops/systemd/tradein-scrape-trigger.timer new file mode 100644 index 00000000..62c1a08c --- /dev/null +++ b/tradein-mvp/ops/systemd/tradein-scrape-trigger.timer @@ -0,0 +1,11 @@ +[Unit] +Description=TradeIn Scrape Trigger — every 60s +Requires=tradein-scrape-trigger.service + +[Timer] +OnBootSec=30s +OnUnitActiveSec=60s +AccuracySec=5s + +[Install] +WantedBy=timers.target diff --git a/tradein-mvp/scripts/trigger-schedules.py b/tradein-mvp/scripts/trigger-schedules.py new file mode 100644 index 00000000..7b325e03 --- /dev/null +++ b/tradein-mvp/scripts/trigger-schedules.py @@ -0,0 +1,188 @@ +#!/usr/bin/env python3 +"""Trigger due scrape schedules via admin API. Called by systemd timer every 60s.""" + +from __future__ import annotations + +import fcntl +import logging +import os +import random +import sys +import time + +import httpx +import psycopg + +LOCK_FILE = "/var/run/tradein-trigger.lock" +DB_URL = os.environ["TRADEIN_DATABASE_URL"] +API_BASE = os.environ.get("TRADEIN_API_BASE", "http://localhost:8000") +# Username that maps to role=admin in roles.yaml — passed as X-Authenticated-User. +ADMIN_TOKEN = os.environ["TRADEIN_ADMIN_TOKEN"] + +logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s") +log = logging.getLogger(__name__) + +# Mapping from scrape_schedules.source → admin API path. +# Sources handled entirely inside the scheduler loop (no separate admin endpoint) +# are listed in _NO_ENDPOINT_SOURCES: they are skipped when the systemd trigger +# runs, and require SCHEDULER_ENABLE=true to fire. +_SOURCE_TO_PATH: dict[str, str] = { + "avito_city_sweep": "/api/v1/admin/scrape/avito-city-sweep", + "yandex_city_sweep": "/api/v1/admin/scrape/yandex-city-sweep", + "n1_city_sweep": "/api/v1/admin/scrape/n1", + "cian_history_backfill": "/api/v1/admin/scrape/cian-backfill-history", + "yandex_address_backfill": "/api/v1/admin/scrape/yandex-address-backfill", +} + +# Sources with no dedicated admin trigger endpoint. +_NO_ENDPOINT_SOURCES = frozenset( + { + "rosreestr_dkp_import", + "listing_source_snapshot", + "asking_to_sold_ratio_refresh", + "refresh_search_matview", + "deactivate_stale_avito", + "sber_index_pull", + "rosreestr_quarter_poll", + } +) + +_MAX_RETRIES = 3 +_RETRY_BASE_S = 2.0 + + +def _normalize_db_url(url: str) -> str: + """Strip SQLAlchemy driver prefix so psycopg v3 accepts the connection string.""" + if url.startswith("postgresql+psycopg://"): + return "postgresql://" + url[len("postgresql+psycopg://"):] + return url + + +def get_due_schedules() -> list[dict[str, str]]: + """Fetch scrape_schedules rows that are due for execution.""" + conn_url = _normalize_db_url(DB_URL) + with psycopg.connect(conn_url) as conn: + rows = conn.execute( + """ + SELECT id, source, schedule_type + FROM scrape_schedules + WHERE enabled = true + AND (next_run_at IS NULL OR next_run_at <= NOW()) + """, + ).fetchall() + description = conn.execute( + "SELECT id, source, schedule_type FROM scrape_schedules LIMIT 0" + ).description or [] + col_names = [desc.name for desc in description] + if not col_names: + col_names = ["id", "source", "schedule_type"] + return [ + {col_names[i]: (str(row[i]) if row[i] is not None else "") for i in range(len(col_names))} + for row in rows + ] + + +def trigger_source(client: httpx.Client, source: str) -> None: + """POST to the admin trigger endpoint for a given source. Retries on connection error.""" + path = _SOURCE_TO_PATH.get(source) + if path is None: + if source in _NO_ENDPOINT_SOURCES: + log.warning( + "trigger: source=%s has no admin API endpoint — " + "ensure SCHEDULER_ENABLE=true or add a dedicated admin endpoint", + source, + ) + else: + log.warning("trigger: unknown source=%s, skipping", source) + return + + url = f"{API_BASE}{path}" + last_exc: Exception | None = None + + for attempt in range(1, _MAX_RETRIES + 1): + try: + resp = client.post( + url, + headers={"X-Authenticated-User": ADMIN_TOKEN}, + timeout=50.0, + ) + log.info( + "trigger: source=%s url=%s status=%d attempt=%d", + source, + url, + resp.status_code, + attempt, + ) + if resp.status_code >= 500: + log.warning( + "trigger: source=%s got HTTP %d — body: %s", + source, + resp.status_code, + resp.text[:200], + ) + return + except httpx.ConnectError as exc: + last_exc = exc + wait = _RETRY_BASE_S * (2 ** (attempt - 1)) + log.warning( + "trigger: source=%s connection error (attempt %d/%d), " + "retrying in %.1fs: %s", + source, + attempt, + _MAX_RETRIES, + wait, + exc, + ) + if attempt < _MAX_RETRIES: + time.sleep(wait) + + log.error( + "trigger: source=%s failed after %d attempts: %s", + source, + _MAX_RETRIES, + last_exc, + ) + + +def main() -> int: + # Exclusive non-blocking flock: if a previous run is still active, exit immediately + # with code 0 so systemd does not log a failure. + lock_fd = open(LOCK_FILE, "w") + try: + fcntl.flock(lock_fd.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB) + except BlockingIOError: + log.info("trigger: lock busy — previous run still active, exiting") + lock_fd.close() + return 0 + + try: + due = get_due_schedules() + if not due: + log.info("trigger: no schedules due") + return 0 + + log.info("trigger: %d schedule(s) due", len(due)) + + with httpx.Client() as client: + for i, row in enumerate(due): + source = row.get("source", "") + trigger_source(client, source) + if i < len(due) - 1: + # Stagger requests to avoid thundering herd on the API + time.sleep(random.uniform(1, 4)) + + return 0 + + except Exception: + log.exception("trigger: unexpected error") + return 1 + finally: + try: + fcntl.flock(lock_fd.fileno(), fcntl.LOCK_UN) + except Exception: + pass + lock_fd.close() + + +if __name__ == "__main__": + sys.exit(main())