From 6876c0c280f273f11e118079532f4e80ef42ece4 Mon Sep 17 00:00:00 2001 From: k Date: Fri, 29 May 2026 14:29:24 +0800 Subject: [PATCH] fix: de-risk launch_seed, monitor price feeds, enforce single-process MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three pre-launch audit findings: launch_seed.py — destructive cleanup is now OPT-IN Previously `launch_seed.py --yes` truncated KOL history (divergence / holdings / posts older than 30d) by default — an accidental run erased unrecoverable data, and abort_if_live_user_state() only guards on subscriptions/bindings, not KOL history. Now nothing is deleted unless --wipe is passed; the safe path is --seed-only (pure fetch). Bare/--yes without --wipe refuses and prints guidance. /api/health/deep — monitor the price feeds, not just scrapers The deep healthcheck only watched the (redundant) Trump scrapers + DB, so a dead Binance/HL price feed — which silently stops ALL tp_sl_monitor stop-loss / take-profit firing on live trades — left health green. Added per-feed liveness (binance.last_tick_at, hl_price_feed.last_tick_at) with a 180s boot grace so startup doesn't false-503. Body now includes price_feeds[]. Single-process enforcement (multi-worker safety) The backend is single-process by design (in-memory scheduler, replay cache, tp_sl table, price_store). systemd unit lacked the --workers 1 + rationale that supervisor.conf already had; added it. Added a runtime advisory file lock (app.main._acquire_singleton_lock): only the leader starts background tasks; extra workers serve HTTP reads only and log CRITICAL. health/deep now reports is_leader so the misconfig is visible to monitors. 72 tests pass. Co-Authored-By: Claude Opus 4.7 --- app/main.py | 101 ++++++++++++++++++++++++++++++++- app/services/binance.py | 10 ++++ app/services/hl_price_feed.py | 10 ++++ deploy/trumpsignal-api.service | 11 +++- scripts/launch_seed.py | 51 ++++++++++++----- 5 files changed, 167 insertions(+), 16 deletions(-) diff --git a/app/main.py b/app/main.py index 70ff960..2b84baa 100644 --- a/app/main.py +++ b/app/main.py @@ -1,5 +1,7 @@ import asyncio +import fcntl import logging +import os from contextlib import asynccontextmanager from apscheduler.schedulers.asyncio import AsyncIOScheduler @@ -41,16 +43,74 @@ logging.basicConfig( ) logger = logging.getLogger(__name__) +from datetime import datetime, timezone from typing import Optional _binance_task: Optional[asyncio.Task] = None _hl_price_task: Optional[asyncio.Task] = None _telegram_task: Optional[asyncio.Task] = None _scheduler: Optional[AsyncIOScheduler] = None +# Process boot time — used by /api/health/deep to grant feeds a startup grace +# period before "no tick yet" counts as a failure (avoids a false 503 at boot). +_PROCESS_START_AT = datetime.now(timezone.utc) +_FEED_BOOT_GRACE_SEC = 180 + +# Single-instance guard. The backend is single-process BY DESIGN (APScheduler, +# scanner_state, the replay cache, tp_sl_monitor's open-trade table and +# price_store are all in-memory). Running >1 worker silently duplicates the +# scheduler, splits TP/SL state, and lets two processes both long-poll the +# Telegram token. This advisory file lock lets only ONE process ("the leader") +# start background tasks; any extra worker serves HTTP reads only and logs loud. +_singleton_lock_fd = None +_is_leader: bool = False + + +def _acquire_singleton_lock() -> bool: + """Return True if this process won the advisory lock and may run background + tasks. False means another process/worker already holds it (multi-worker + misconfig). The OS releases the lock automatically when the holder exits, + so a crashed leader frees it for the next start.""" + global _singleton_lock_fd + path = os.environ.get("SINGLETON_LOCK_PATH", "/tmp/trumpsignal-backend.lock") + try: + fd = open(path, "w") + fcntl.flock(fd.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB) + try: + fd.write(str(os.getpid())) + fd.flush() + except OSError: + pass + _singleton_lock_fd = fd # keep the fd alive for the process lifetime + return True + except BlockingIOError: + # Lock held by another process — this is the multi-worker signal. + return False + except OSError as exc: + # Lock file itself is unusable (perms, missing dir). Fail OPEN so a + # lock-path problem never takes the whole service down — but log loudly. + logger.warning("Singleton lock unavailable (%s) — proceeding as leader.", exc) + return True + @asynccontextmanager async def lifespan(app: FastAPI): - global _binance_task, _hl_price_task, _telegram_task, _scheduler + global _binance_task, _hl_price_task, _telegram_task, _scheduler, _is_leader + + # 0. Single-instance guard. If another worker already holds the lock, this + # process must NOT start the scheduler / scrapers / price feeds / Telegram + # poller — otherwise scans double-fire, TP/SL state splits across processes, + # and two pollers fight over the Telegram token. Serve HTTP reads only. + _is_leader = _acquire_singleton_lock() + if not _is_leader: + logger.critical( + "SINGLETON LOCK NOT ACQUIRED — another backend process/worker is " + "already the leader. This process will serve HTTP reads ONLY and " + "start NO background tasks. You are almost certainly running with " + ">1 worker, which is UNSUPPORTED (see deploy/supervisor.conf). " + "Run uvicorn/gunicorn with --workers 1." + ) + yield + return # 1. Dev convenience only. Production should rely on Alembic so schema # ownership stays explicit and startup never mutates the DB implicitly. @@ -397,13 +457,52 @@ async def health_deep(): # 6× the 15s interval — both sources are silent problems.append(f"scrapers: all stale (freshest={freshest_age}s)") + # 3. Price-feed liveness — INDEPENDENT of the scrapers. A dead price feed + # means tp_sl_monitor never fires: stop-loss / take-profit / trailing stops + # silently stop protecting live trades. This is a real-money incident the + # scraper check above would completely miss, so it gets its own per-feed gate. + from app.services import binance as _binance + from app.services import hl_price_feed as _hl_feed + + boot_age = int((now - _PROCESS_START_AT).total_seconds()) + in_grace = boot_age < _FEED_BOOT_GRACE_SEC + # (feed name, last_tick_at, max age before stale). Binance ticks multiple + # times/sec for BTC/ETH; HL feed polls every 2s. Thresholds are generous. + feeds = [ + ("binance_ws", _binance.last_tick_at, 120), + ("hl_price_feed", _hl_feed.last_tick_at, 60), + ] + feed_status = [] + for name, last, max_age in feeds: + age = int((now - last).total_seconds()) if last else None + feed_status.append({ + "name": name, + "last_tick": last.isoformat() if last else None, + "age_sec": age, + "max_age_sec": max_age, + }) + if last is None: + # Never ticked: only a problem once past the boot grace window + # (otherwise we'd 503 for the first ~3 min of every restart). + if not in_grace: + problems.append(f"{name}: no tick since boot ({boot_age}s ago)") + elif age is not None and age > max_age: + problems.append(f"{name}: stale ({age}s > {max_age}s)") + + # A follower (lost the singleton lock) runs no background tasks — surface it + # explicitly so the 503 it already triggers has an obvious root cause. + if not _is_leader: + problems.append("instance: not the singleton leader (>1 worker?) — no background tasks running") + body = { "status": "ok" if not problems else "degraded", "now": now.isoformat(), + "is_leader": _is_leader, "db_ok": db_ok, "db_error": db_error, "scrapers": source_status, "freshest_age_sec": freshest_age, + "price_feeds": feed_status, "problems": problems, } diff --git a/app/services/binance.py b/app/services/binance.py index 9964b52..1f42e3f 100644 --- a/app/services/binance.py +++ b/app/services/binance.py @@ -1,6 +1,8 @@ import asyncio import json import logging +from datetime import datetime, timezone +from typing import Optional import httpx import websockets @@ -11,6 +13,11 @@ from app.ws.manager import manager logger = logging.getLogger(__name__) +# Liveness signal for /api/health/deep. Updated on every processed kline tick. +# A stale value means the price feed is down → TP/SL/trailing stops are NOT +# firing for live trades, which is a real-money incident, not a soft warning. +last_tick_at: Optional[datetime] = None + # symbol (lowercase Binance ticker) → asset (HL perp name, uppercase). # # WHY THIS MATTERS: tp_sl_monitor.on_price_tick(asset, price) is the sole @@ -68,6 +75,9 @@ async def _process_message(raw: str): } price_store.update(asset, candle) + global last_tick_at + last_tick_at = datetime.now(timezone.utc) + # TP/SL check on every tick (cheap no-op when no trades are being watched) from app.services.tp_sl_monitor import on_price_tick on_price_tick(asset, candle["close"]) diff --git a/app/services/hl_price_feed.py b/app/services/hl_price_feed.py index 8dd6e6d..4f26776 100644 --- a/app/services/hl_price_feed.py +++ b/app/services/hl_price_feed.py @@ -28,6 +28,7 @@ from __future__ import annotations import asyncio import logging import time +from datetime import datetime, timezone from typing import Optional import httpx @@ -42,6 +43,10 @@ HL_API_URL = "https://api.hyperliquid.xyz/info" # HL-native assets not covered by the Binance WS. Add new ones here as needed. HL_PRICE_ASSETS: frozenset[str] = frozenset({"HYPE", "PURR"}) +# Liveness signal for /api/health/deep. Set on every successful allMids poll. +# Stale → HYPE/PURR trades lose TP/SL just like a dead Binance feed would. +last_tick_at: Optional[datetime] = None + # Poll cadence — 2 s gives ~same freshness as a Binance 1-min kline close # without hammering the free HL REST endpoint (30 req/min well within limit). _POLL_INTERVAL = 2.0 @@ -96,6 +101,11 @@ async def _tick() -> None: r.raise_for_status() mids: dict = r.json() # {"BTC": "74541.0", "HYPE": "13.5", …} + # Feed is alive the moment we successfully fetch mids, even if a specific + # asset is momentarily absent from the response. + global last_tick_at + last_tick_at = datetime.now(timezone.utc) + for asset in HL_PRICE_ASSETS: raw: Optional[str] = mids.get(asset) if raw is None: diff --git a/deploy/trumpsignal-api.service b/deploy/trumpsignal-api.service index 5a0f087..614559f 100644 --- a/deploy/trumpsignal-api.service +++ b/deploy/trumpsignal-api.service @@ -9,7 +9,16 @@ User=trumpsignal Group=trumpsignal WorkingDirectory=/opt/trumpsignal/backend EnvironmentFile=/opt/trumpsignal/backend/.env -ExecStart=/opt/trumpsignal/backend/venv/bin/uvicorn app.main:app --host 0.0.0.0 --port 8000 +# MUST stay --workers 1. The backend is single-process BY DESIGN: APScheduler, +# scanner_state, the signed-request replay cache, tp_sl_monitor's open-trade +# table and price_store are all in-memory and NOT shared across processes. +# With >1 worker the scheduler runs in every worker (duplicate scans / ingest / +# trades), the scanner toggle split-brains, and TP/SL only sees same-worker +# trades. A runtime singleton lock in app.main now refuses to start background +# tasks in extra workers as a backstop, but the correct config is one worker. +# To scale horizontally, move scheduler leadership + scanner state + replay +# cache to Redis/DB FIRST. (Mirrors deploy/supervisor.conf.) +ExecStart=/opt/trumpsignal/backend/venv/bin/uvicorn app.main:app --host 0.0.0.0 --port 8000 --workers 1 # Auto-restart on any exit (crash, OOM, segfault, manual kill). # RestartSec=10 = wait 10s between restart attempts so we don't hammer a broken state. diff --git a/scripts/launch_seed.py b/scripts/launch_seed.py index 99bad4f..71d20d1 100755 --- a/scripts/launch_seed.py +++ b/scripts/launch_seed.py @@ -28,25 +28,29 @@ What it INTENTIONALLY DOES NOT touch: baseline for the next on-chain poll (without it, the next snapshot would diff against nothing and produce zero kol_holding_changes). -Usage: - # Preview (no DB writes): - DATABASE_URL='sqlite+aiosqlite:///./trumpsignal.db' \\ - venv/bin/python scripts/launch_seed.py --dry-run +DESTRUCTIVE STEPS ARE OPT-IN. By default this script only SEEDS (re-fetches +upstream data); it will NOT delete or truncate anything unless you explicitly +pass --wipe. This prevents an accidental `launch_seed.py --yes` from erasing +KOL history (divergence / holdings / posts) that cannot be recovered. - # Seed only (pure pre-launch warmup, no truncation / deletion): +Usage: + # Pure pre-launch warmup — fetch only, NOTHING deleted (the safe default): DATABASE_URL='sqlite+aiosqlite:///./trumpsignal.db' \\ venv/bin/python scripts/launch_seed.py --seed-only - # Execute: + # Preview what a wipe WOULD delete (no DB writes): DATABASE_URL='sqlite+aiosqlite:///./trumpsignal.db' \\ - venv/bin/python scripts/launch_seed.py --yes + venv/bin/python scripts/launch_seed.py --wipe --dry-run - # Skip the SEED step (just clean — useful if you'd rather let the - # scheduled poll jobs trigger naturally over the next 24h): - DATABASE_URL='...' venv/bin/python scripts/launch_seed.py --yes --no-seed + # Execute the destructive wipe + reseed (requires BOTH --wipe and --yes): + DATABASE_URL='sqlite+aiosqlite:///./trumpsignal.db' \\ + venv/bin/python scripts/launch_seed.py --wipe --yes -NOT safe to re-run after real users are on the platform — the KOL truncation -will erase divergence/alignment context that you can't recover. + # Wipe only, skip the reseed (let scheduled polls refill over 24h): + DATABASE_URL='...' venv/bin/python scripts/launch_seed.py --wipe --yes --no-seed + +The --wipe path is NOT safe to re-run after real users are on the platform — +the KOL truncation erases divergence/alignment context you can't recover. """ from __future__ import annotations @@ -283,12 +287,15 @@ async def seed_real_data(*, exercise_scanners: bool = False) -> None: async def main() -> int: p = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) + p.add_argument("--wipe", action="store_true", + help="OPT IN to destructive cleanup (drop obsolete-source posts + " + "truncate KOL window). Without this flag nothing is ever deleted.") p.add_argument("--dry-run", action="store_true", - help="show what would be deleted, no DB writes") + help="with --wipe: show what would be deleted, no DB writes") p.add_argument("--seed-only", action="store_true", help="run upstream fetch/warmup only; skip all wipe/truncation steps") p.add_argument("--yes", action="store_true", - help="actually perform the wipe (required without --dry-run)") + help="with --wipe: actually perform the wipe (required without --dry-run)") p.add_argument("--no-seed", action="store_true", help="skip the upstream re-fetch step") p.add_argument("--exercise-scanners", action="store_true", @@ -303,6 +310,22 @@ async def main() -> int: print(red("--seed-only conflicts with --no-seed.")) return 2 + if args.seed_only and args.wipe: + print(red("--seed-only conflicts with --wipe (seed-only never deletes).")) + return 2 + + # Default to the SAFE path. A bare invocation, or --yes/--dry-run/--no-seed + # WITHOUT --wipe, must never delete data. Steer the user to --seed-only. + if not args.wipe and not args.seed_only: + print(red("Refusing to run a destructive path without --wipe.")) + print(yellow( + "This script seeds by default and only deletes when --wipe is given.\n" + " • Pure pre-launch fetch (recommended): --seed-only\n" + " • Preview a destructive wipe: --wipe --dry-run\n" + " • Execute a destructive wipe + reseed: --wipe --yes" + )) + return 2 + if args.seed_only: before = await report_counts("BEFORE") await seed_real_data(exercise_scanners=args.exercise_scanners)