"""CashPilot — FastAPI application. Self-hosted passive income dashboard: service catalog, Docker container management, and earnings tracking. """ from __future__ import annotations import asyncio import collections import contextlib import hmac import ipaddress import json import logging import math import os import re import secrets import socket import time import urllib.parse import weakref from contextlib import asynccontextmanager from datetime import UTC, datetime, timedelta from typing import Any import httpx from apscheduler.events import EVENT_JOB_ERROR, EVENT_JOB_MISSED from apscheduler.schedulers.asyncio import AsyncIOScheduler from fastapi import Depends, FastAPI, HTTPException, Request, Response from fastapi.encoders import jsonable_encoder from fastapi.exceptions import RequestValidationError from fastapi.responses import JSONResponse, PlainTextResponse from fastapi.staticfiles import StaticFiles from pydantic import BaseModel, Field, field_validator from starlette.middleware.base import BaseHTTPMiddleware from app import ( activity_ui, arch, auth, catalog, compose_generator, constants, credential_test, custom_services, database, disclosure, drift, egress, exchange_rates, fleet_key, ipgeo, lan_isolation, login_rate_limit, machine_economics, metrics, net_activity, notify, onchain, payout_registry, payouts, placement, power, preflight, producer_state, proxies, ratings, setup_token, update_check, usage, version, ) from app.collectors.base import REGISTER_PENDING from app.worker_proxy import _pin_url_to_ip, _validate_worker_url logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(levelname)s] %(name)s: %(message)s", ) logger = logging.getLogger(__name__) scheduler = AsyncIOScheduler() # In-memory store for the latest collector alerts (errors from last run) _collector_alerts: list[dict[str, str]] = [] # Activity alerts of the last day, rebuilt from the alerts table, so the bell # keeps them across the hourly wholesale rebuild of _collector_alerts. _activity_bell: list[dict[str, str]] = [] # Whether a collection has ever COMPLETED here. An empty alert list means # "nothing is wrong" only once something has looked; before that it means # "nothing has been checked", and the bell used to render both as "All # collectors healthy" (CashPilot-tb5). Restored on startup from durable state so # a restart does not reset the claim to "never ran" while data exists. _collection_has_run: bool = False # Monotonic time the collection machinery last COMPLETED a run (success or # failure — this measures aliveness, not outcome; the success timestamp is a # metric). Initialized to process start so boot gets a grace window, exactly # like the worker's heartbeat stamp. A dead scheduler, a wedged collection # lock, or a blocked event loop all freeze this — and they used to freeze # every other signal green with them. _last_collection_finished: float = time.monotonic() _collection_lock = asyncio.Lock() _collection_semaphore = asyncio.Semaphore(8) # Fire-and-forget background tasks (e.g. triggered collection runs). Keeping a # reference prevents the task from being garbage-collected mid-run and lets us # retrieve/log any exception it raised (bare `asyncio.create_task(...)` drops # the reference and silently swallows exceptions). _background_tasks: set[asyncio.Task] = set() def _spawn(coro) -> asyncio.Task: """Fire-and-forget a coroutine while keeping a reference and logging errors.""" task = asyncio.create_task(coro) _background_tasks.add(task) def _on_done(t: asyncio.Task) -> None: _background_tasks.discard(t) if not t.cancelled(): exc = t.exception() if exc is not None: logger.error("Background task failed: %s", exc, exc_info=exc) task.add_done_callback(_on_done) return task # Login rate limiting moved to app.login_rate_limit (bead sux) — it was the last # thing the routers genuinely needed from this module, and extracting it is what # breaks the main -> routers -> main import cycle. # # Re-exported by REFERENCE, not copied: tests patch app.main._check_login_rate # and a conftest fixture clears app.main._login_attempts between tests. Binding a # copy here would leave both pointing at an object nothing reads, and the fixture # would silently stop isolating tests. _login_attempts = login_rate_limit._login_attempts _LOGIN_MAX_ATTEMPTS = login_rate_limit.MAX_ATTEMPTS _LOGIN_WINDOW_SECONDS = login_rate_limit.WINDOW_SECONDS _check_login_rate = login_rate_limit.check_login_rate _record_failed_login = login_rate_limit.record_failed_login def _safe_json(raw: str, fallback: Any = None) -> Any: """Parse JSON with a fallback so one malformed DB row doesn't 500 the fleet.""" try: return json.loads(raw) except (json.JSONDecodeError, TypeError): return fallback if fallback is not None else [] def _decoded_worker(worker: dict[str, Any]) -> dict[str, Any]: """A worker row with its JSON-TEXT columns decoded into real structures. ``list_workers``/``get_worker`` return raw rows, so ``containers`` and ``system_info`` arrive as strings. Passing one of those straight to code expecting a mapping is an AttributeError at request time, so every caller that reads inside them goes through here. """ decoded = dict(worker) # Deliberately delegates rather than repeating the three _safe_json calls: # a second copy of the decoding is how one caller ends up forgetting it. _parse_worker_json(decoded) return decoded # Previous network counter reading per (worker, slug), for turning Docker's # CUMULATIVE totals into a rate. In memory on purpose: losing it on restart # costs one unknown reading, whereas persisting a stale baseline across a # restart risks pairing it with counters that reset in the meantime. _net_baselines: dict[tuple[Any, str], tuple[int, float]] = {} def _traffic_state(slug: str, containers: list[dict[str, Any]]) -> str | None: """MOVING/SILENT/UNKNOWN for a service, from two counter readings. Returns None when there is nothing to say at all, so the caller can leave the signal out entirely rather than assert "unknown" about a service whose containers were never seen. """ now = time.monotonic() rates: list[float] = [] measured = False for container in containers: total = net_activity.totals(container) if total is None: continue measured = True key = (container.get("_worker_id"), slug) previous = _net_baselines.get(key) _net_baselines[key] = (total, now) if previous is None: continue value = net_activity.rate(previous[0], total, now - previous[1]) if value is not None: rates.append(value) if not measured: return None if not rates: # Counters exist but no usable interval yet — first sight of this # container, or its counters reset. Saying "unknown" is the point. return net_activity.UNKNOWN # Any instance moving data means the service is not silent. return net_activity.classify(max(rates)) async def _get_all_worker_containers(workers: list[dict[str, Any]] | None = None) -> list[dict[str, Any]]: """Collect container/app data from all online workers' heartbeat data in DB. ``workers`` lets a caller that has ALREADY fetched the worker list hand it in rather than causing a second `SELECT *` and a second full JSON decode of every row in the same request. Everyone else keeps calling this with no arguments and gets the fetch for free. """ if workers is None: workers = await database.list_workers() result: list[dict[str, Any]] = [] for w in workers: if w.get("status") != "online": continue sys_info = _safe_json(w.get("system_info", "{}"), {}) worker_has_docker = sys_info.get("docker_available", False) is_android = placement.is_android(sys_info) worker_name = w.get("name", "worker") # Docker containers (from Docker-based workers only — skip for Android) if not is_android: containers = _safe_json(w.get("containers", "[]")) for c in containers: slug = c.get("slug", "") if slug: result.append( { "slug": slug, "name": c.get("name", slug), "status": c.get("status", "unknown"), "image": c.get("image", ""), "cpu_percent": c.get("cpu_percent", 0), "memory_mb": c.get("memory_mb", 0), "net_rx_bytes": c.get("net_rx_bytes"), "net_tx_bytes": c.get("net_tx_bytes"), "category": "", # The WORKER's answer, not the node name. The image # matcher sets "external" for a container it found by # image rather than by CashPilot's own label — one the # user started themselves. Overwriting it here meant # that container appeared as an ordinary managed # service with live Restart/Stop/Logs buttons, and # every one of them answered "404 Container not # found" for a row the same screen called Running. # The node name is already carried by _node. "deployed_by": c.get("deployed_by") or worker_name, # Which copy on that worker. Without it every copy # reads as #1 and its buttons act on the first one. "instance": c.get("instance") or 1, # The dial-back address the worker read from the ONE # env var the service's catalog entry declares. None # when undeclared/unreported — this dict is a key # allowlist, so without this line the field the # mismatch check depends on would be silently # dropped right here. "advertised_address": c.get("advertised_address"), # When and how a stopped container stopped, for the # container-down alert. Absent on a running one. "finished_at": c.get("finished_at"), "exit_code": c.get("exit_code"), "restart_policy": c.get("restart_policy"), # What the provider said when it refused to run the # client (an account_disabled health signal), read by the # worker from a stopped or crash-looping container. "fatal": c.get("fatal"), # When that refused run ended, to tell a refusal # older than the last balance read (stale) apart. "fatal_at": c.get("fatal_at"), # What the container was created with, for the drift # check (app/drift.py). None from a worker too old to send it. "shape": c.get("shape"), "_arch": sys_info.get("arch"), "_node": worker_name, "_worker_id": w.get("id"), "_has_docker": worker_has_docker, "_is_android": False, } ) # Android apps (from Android workers) if is_android: apps = _safe_json(w.get("apps", "[]")) for a in apps: slug = a.get("slug", "") if slug: result.append( { "slug": slug, "name": a.get("slug", slug), # Three-valued, matching what the Android client # now sends. `running` is None when the phone could # not determine it -- every detection signal degrades # to false when its permissions are denied -- and # `"running" if a.get("running") else "stopped"` # turned that into a confident "stopped", so the # fleet page stated the user's earning apps had died # when the device simply could not see them. "status": _android_app_status(a.get("running")), "image": "", "cpu_percent": 0, "memory_mb": 0, "category": "", # Android apps are enumerated by the worker itself and # are never CashPilot-managed containers, so there is # no external/managed distinction to preserve here. "deployed_by": worker_name, "_node": worker_name, "_worker_id": w.get("id"), "_has_docker": False, "_is_android": True, "_net_tx_24h": a.get("net_tx_24h", 0), "_net_rx_24h": a.get("net_rx_24h", 0), } ) return result async def _resolve_worker_id(worker_id: int | None) -> int: """Return a valid worker_id, auto-resolving when only one worker is online.""" if worker_id is not None: return worker_id workers = await database.list_workers() online = [w for w in workers if w["status"] == "online"] if len(online) == 1: return online[0]["id"] if len(online) == 0: raise HTTPException(status_code=503, detail="No workers online") raise HTTPException( status_code=400, detail="worker_id is required (multiple workers online)", ) # --------------------------------------------------------------------------- # Periodic collection job # --------------------------------------------------------------------------- async def _run_health_check() -> None: """Check health of all deployed containers and record events. Deduplicates by slug: if *any* instance of a service is running, record a single check_ok for that slug (avoids penalising services deployed on multiple nodes where one may be stopped). A service that vanishes from the heartbeat entirely (container removed outside CashPilot, crash-removed, worker's Docker daemon down) previously got no health event at all — it just stopped appearing, so its score stayed frozen wherever it last was (often green) forever. Any known Docker-backed deployment missing from every online worker's current data now gets an explicit check_down. Scoped to "at least one worker online" so a fully-offline fleet (no heartbeat data to trust either way) never triggers false check_downs; "external" deployments (e.g. Grass, Bytelixir) have no container and are excluded since no worker ever reports them. """ try: statuses = await _get_all_worker_containers() # Aggregate: slug -> best status (running wins) slug_best: dict[str, str] = {} for s in statuses: slug = s["slug"] status = s.get("status", "unknown") # A phone that cannot look (no usage or notification access) reports # its apps as unknown. That says nothing about the app, so it takes # no part here: it neither records downtime on its own nor hides a # stopped container another box reports for the same service. if s.get("_is_android") and status == "unknown": continue if slug_best.get(slug) != "running": slug_best[slug] = status # Collect every event for this cycle and write them in a single transaction # rather than one fsync'd commit per service (see database.record_health_events). events: list[tuple[str, str, str]] = [] for slug, status in slug_best.items(): if status == "running": events.append((slug, "check_ok", "")) else: events.append((slug, "check_down", status)) workers = await database.list_workers() online = [w for w in workers if w.get("status") == "online"] # "Missing from a heartbeat" is only evidence of being DOWN if every # online worker could actually look. A worker whose Docker socket became # unreadable — unmounted by a host update, permissions changed, daemon # erroring — keeps heartbeating happily and reports zero containers, so # this loop wrote a durable check_down for EVERY deployment every five # minutes while those containers were up and earning. The user watched # each service's Health column fall and its uptime read 0%, and the # events kept dragging the 7-day score long after the socket was fixed. # # The deployments table has no worker column, so a missing container # cannot be attributed to a particular host: if any online worker is # blind, the container might be on that one. Unknown is not down, so # nothing is recorded — the score is left untouched rather than # invented. # A phone has no Docker by design, so it cannot be hiding a missing # container and is not blind in this sense. Counting it switched the # missing-container check off for the whole fleet whenever a phone was # online, and warned about it every five minutes. # A paired Desktop reports its containers and sends no docker_available # either; counting it as blind did the same to every fleet with one. blind = [w for w in online if placement.is_blind(w)] _report_blind_workers(blind) if online and not blind: deployments = await database.get_deployments() for d in deployments: slug = d["slug"] if d.get("status") == "external" or slug in slug_best: continue events.append((slug, "check_down", "missing from heartbeat")) await database.record_health_events(events) await _container_down_check(statuses) await _account_disabled_check(statuses) await _missing_deployment_check(workers) except Exception as exc: logger.warning("Health check skipped: %s", exc) # The blind workers the last warning named. The health check runs every five # minutes, so warning on each run repeated one fact hundreds of times a day; it # now speaks when the set changes and once when every worker can see again. _blind_workers_logged: frozenset[str] | None = None def _report_blind_workers(blind: list[dict[str, Any]]) -> None: global _blind_workers_logged names = frozenset(str(w.get("name") or w.get("id")) for w in blind) if names == (_blind_workers_logged or frozenset()): return _blind_workers_logged = names if names: logger.warning( "Not recording missing-container downtime: %d online worker(s) cannot read Docker (%s). " "Their containers may be running; recording them as down would be a guess.", len(names), ", ".join(sorted(names)), ) else: logger.info("Every online worker can read Docker again; missing-container downtime is recorded again.") # How long a managed container may stay exited before it is an alert. Long # enough for a deploy, an update or a manual restart to finish; short enough # that a node which exited cleanly and stayed down (restart policy "no") is # news within the hour instead of after days of check_down rows nobody reads. CONTAINER_DOWN_GRACE = timedelta(minutes=30) # A stop recorded by CashPilot explains a container that finished within this # long after it; a container that finished later stopped for another reason. _STOP_EXPLAINS_WITHIN = timedelta(minutes=10) # When each stopped container was first seen down, for a worker too old to # report Docker's own FinishedAt. Keyed by (worker id, container name). _container_first_seen_down: dict[tuple[Any, str], datetime] = {} # The bell entries the check currently stands behind, re-added after every # collection, which replaces the bell list wholesale. _container_down_bell: dict[str, dict[str, str]] = {} _RESTARTING_POLICIES = ("unless-stopped", "always") def _docker_time(value: Any) -> datetime | None: """Docker's FinishedAt (RFC 3339 in UTC, nanoseconds) as a datetime.""" if not isinstance(value, str) or len(value) < 19: return None try: return datetime.fromisoformat(value[:19]).replace(tzinfo=UTC) except ValueError: return None async def _container_down_check(statuses: list[dict[str, Any]]) -> None: """Alert when a CashPilot-managed container has stayed exited past the grace period. The health check already wrote check_down for such a container every five minutes, and nothing read it: three urnetwork containers exited cleanly with restart policy "no" and sat down for two days with no alert. One alert per service, cleared when no managed copy of it is down any more. A stop the user made from CashPilot is not an alert. Never raises. """ global _collector_alerts try: now = datetime.now(UTC) down: dict[str, list[dict[str, Any]]] = {} seen_keys: set[tuple[Any, str]] = set() for s in statuses: if s.get("_is_android") or s.get("deployed_by") == "external": continue if s.get("status") not in ("exited", "dead"): continue # The provider refused it (account disabled): starting it again # cannot help, and the account-disabled alert says so instead. if s.get("fatal"): continue key = (s.get("_worker_id"), str(s.get("name") or s.get("slug"))) seen_keys.add(key) since = _docker_time(s.get("finished_at")) or _container_first_seen_down.setdefault(key, now) down.setdefault(s["slug"], []).append({**s, "_down_since": since}) for key in set(_container_first_seen_down) - seen_keys: _container_first_seen_down.pop(key, None) alerting: dict[str, str] = {} for slug, copies in down.items(): long_down = [c for c in copies if now - c["_down_since"] >= CONTAINER_DOWN_GRACE] if not long_down: continue last = await database.get_last_lifecycle_event(slug) if last and last[0] in ("stop", "remove"): stopped_at = _docker_time(last[1].replace(" ", "T")) if stopped_at is not None: long_down = [c for c in long_down if c["_down_since"] > stopped_at + _STOP_EXPLAINS_WITHIN] if long_down: alerting[slug] = _container_down_message(slug, long_down) # Recovery: nothing of that service is down any more. A copy that is # down but still inside the grace period (or explained by a stop) is # neither raised nor cleared, so a UI restart cannot clear an alert # just because it forgot when an old worker's container went down. for slug in await database.get_alert_subjects("container"): if slug not in down: await database.clear_alerts("container", slug) logger.info("%s is running again; cleared its container-down alert", slug) for slug in list(_container_down_bell): if slug not in alerting: _container_down_bell.pop(slug) for slug, message in alerting.items(): _container_down_bell[slug] = {"kind": "container", "platform": slug, "error": message} if alert_id := await database.record_alert("container", slug, message): _spawn(_push_alert("container", slug, f"CashPilot: {slug} has stopped running", message, alert_id)) _collector_alerts = [a for a in _collector_alerts if a.get("kind") != "container"] _collector_alerts.extend(_container_down_bell.values()) except Exception as exc: # noqa: BLE001 - a diagnostic must not break the health check logger.warning("Container-down check failed: %s", exc) def _container_down_message(slug: str, copies: list[dict[str, Any]]) -> str: name = (catalog.get_service(slug) or {}).get("name") or slug since = min(c["_down_since"] for c in copies).strftime("%Y-%m-%d %H:%M") nodes = ", ".join(sorted({str(c.get("_node") or "?") for c in copies})) codes = sorted({c["exit_code"] for c in copies if isinstance(c.get("exit_code"), int)}) how = f" (exit code {', '.join(str(code) for code in codes)})" if codes else "" message = f"{name} has not been running since {since} UTC on {nodes}{how}, so it is not earning." if catalog.fence_gated(catalog.get_service(slug)): # No restart policy on purpose: the worker starts it once the LAN fence is confirmed. return message + ( " The worker starts it as soon as the LAN fence is confirmed; while it stays down, the Fleet page " "says what holds it. If you stopped it on purpose, start it from the dashboard when you want it back." ) policies = sorted({str(c["restart_policy"]) for c in copies if c.get("restart_policy")} - set(_RESTARTING_POLICIES)) if policies: message += ( f' Its restart policy is "{", ".join(policies)}", so Docker will not start it again by itself;' " redeploying it from CashPilot sets unless-stopped." ) return message + " Start it from the dashboard, or remove it if you stopped it on purpose." # Services whose provider has disabled the account: slug -> the message shown. # One alert per service, whatever number of boxes crash-loop on it. _account_disabled: dict[str, str] = {} # Platforms whose collector said, on its last run, that the account is disabled. _collector_says_disabled: set[str] = set() # Platforms the containers' logs said were disabled on the last check that had # heartbeat data. Kept apart so a check without data keeps only what the logs # said, not a collector-only entry the collection loop has since dropped. _disabled_by_logs: set[str] = set() # Platforms whose collector read a balance on its last run. A balance read means # the account works again, whatever an old container log still says. _collected_ok_last_run: set[str] = set() # When the collector last read a balance for a service that declares an # account_disabled signal, kept in the config table so it survives a restart. # A refusal from a run that ended before it is history: a stopped copy keeps # the old refusal in its log after the account is restored, and without this a # later 429 or UI restart raised the alert and pushed again. _BALANCE_READ_KEY = "_balance_read_at:" # The alert stays one alert for as long as the account stays disabled: it is # cleared, and so armed again, only when the account comes back. _ACCOUNT_DISABLED_COOLDOWN_HOURS = 24 * 365 def _account_disabled_message(slug: str) -> str: name = (catalog.get_service(slug) or {}).get("name") or slug return ( f"{name} has disabled this account. Restarting will not help: only {name} can turn it back on, " f"so contact their support. CashPilot leaves its containers as they are." ) def _declares_account_disabled(slug: str) -> bool: signals = ((catalog.get_service(slug) or {}).get("docker") or {}).get("health_signals") return isinstance(signals, list) and any( isinstance(s, dict) and s.get("state") == catalog.ACCOUNT_DISABLED for s in signals ) async def _record_balance_reads(collected_ok: set[str]) -> None: """Remember when a balance was read, for the services whose refusal the worker can report.""" now = datetime.now(UTC).isoformat() for slug in sorted(collected_ok): if _declares_account_disabled(slug): try: await database.set_config(_BALANCE_READ_KEY + slug, now) except Exception as exc: # noqa: BLE001 - losing the stamp only keeps an alert up logger.warning("Could not record the %s balance read: %s", slug, exc) async def _balance_read_at(slug: str) -> datetime | None: try: raw = await database.get_config(_BALANCE_READ_KEY + slug) return datetime.fromisoformat(raw) if isinstance(raw, str) and raw else None except Exception: # noqa: BLE001 - unknown means no read to compare against return None async def _account_disabled_check(statuses: list[dict[str, Any]]) -> None: """Raise one alert per service whose provider disabled the account; clear it when the account is back. Two sources say so: a worker that matched an account_disabled health signal in a crash-looping or stopped container, and a collector whose login was refused as a disabled account. A balance read beats any refusal from a run that ended before it, also across restarts. Never stops or removes anything. Never raises. """ global _collector_alerts, _disabled_by_logs try: if statuses: from_logs: set[str] = set() read_at: dict[str, datetime | None] = {} for s in statuses: if s.get("fatal") != catalog.ACCOUNT_DISABLED: continue slug = s["slug"] if slug not in read_at: read_at[slug] = await _balance_read_at(slug) refused_at = _docker_time(s.get("fatal_at")) if read_at[slug] and refused_at and refused_at <= read_at[slug]: continue from_logs.add(slug) else: # No heartbeat data at all says nothing about the containers: keep # what they said last. from_logs = set(_disabled_by_logs) _disabled_by_logs = set(from_logs) from_logs -= _collected_ok_last_run disabled = from_logs | _collector_says_disabled for slug in list(_account_disabled): if slug not in disabled: _account_disabled.pop(slug) await database.clear_alerts("account_disabled", slug) logger.info("%s: the provider no longer refuses the account; cleared its alert", slug) for slug in sorted(disabled): message = _account_disabled_message(slug) _account_disabled[slug] = message if alert_id := await database.record_alert( "account_disabled", slug, message, cooldown_hours=_ACCOUNT_DISABLED_COOLDOWN_HOURS ): name = (catalog.get_service(slug) or {}).get("name") or slug _spawn( _push_alert("account_disabled", slug, f"CashPilot: {name} disabled your account", message, alert_id) ) _collector_alerts = _with_account_disabled(_collector_alerts) except Exception as exc: # noqa: BLE001 - a diagnostic must not break the health check logger.warning("Account-disabled check failed: %s", exc) def _with_account_disabled(alerts: list[dict[str, Any]]) -> list[dict[str, Any]]: """The bell list with one account-disabled entry per such service in place of its symptoms. A disabled account also fails the collector (often as a rate limit, from the crash loop's logins) and stops the container; those entries point at the wrong fixes, so they give way to the one that names the cause. """ kept = [ a for a in alerts if a.get("kind") != "account_disabled" and not (a.get("platform") in _account_disabled and a.get("kind", "collector") in ("collector", "container")) ] kept.extend({"kind": "account_disabled", "platform": slug, "error": msg} for slug, msg in _account_disabled.items()) return kept # When each recorded deployment was first seen missing from every heartbeat # (key: slug and worker id, or None for "no box at all"). See placement.missing. _deployment_first_missing: dict[tuple[str, Any], datetime] = {} # The bell entries the missing-deployment check stands behind, re-added after # every collection like the container-down ones. _missing_bell: dict[str, dict[str, Any]] = {} async def _missing_deployment_check(workers: list[dict[str, Any]] | None = None) -> None: """Alert when CashPilot records a deployment that no worker runs any more. The deployments table outlives a container removed by hand or a rebuilt box, and every screen that read it went on calling the service deployed. One alert per service, after placement.MISSING_GRACE, through the usual push path; it clears when a worker reports the service again, or when the record is forgotten (DELETE /api/deployments/{slug}). When the fleet cannot tell (no worker online, or a Docker host that cannot read its socket), nothing is raised or cleared. Never raises. """ global _collector_alerts try: if workers is None: workers = await database.list_workers() stored = await database.get_alert_subjects("missing") scopes = {slug: _missing_scope(slug) for slug in stored} found = placement.missing( await database.get_deployments(), await database.list_deployment_placements(), workers, now=datetime.now(UTC), first_missing=_deployment_first_missing, alerted=set().union(*scopes.values()) if scopes else set(), ) if found is None: return by_id = {w.get("id"): w for w in workers} where = placement.runs_on(workers) for slug in set(stored) | set(_missing_bell): if slug in found: continue # What the alert covered and is still absent, or sits on a box that # cannot be looked at right now: that much of it stands. scope = scopes.get(slug) or _missing_scope(slug) left = [wid for s, wid in _deployment_first_missing if s == slug and ((s, wid) in scope or s in scope)] if not left: if slug in stored: await database.clear_alerts("missing", slug) logger.info("%s is no longer recorded on a box that does not run it; cleared its alert", slug) _missing_bell.pop(slug, None) continue entry = _missing_bell.get(slug) if entry is not None and "everywhere" in entry: kept = { "everywhere": None in left, "boxes": [ {"worker_id": wid, "name": (by_id.get(wid) or {}).get("name") or str(wid)} for wid in sorted(w for w in left if w is not None) ], "still_on": [r["name"] for r in where.get(slug, [])], } entry.update( everywhere=kept["everywhere"], boxes=kept["boxes"], error=_missing_deployment_message(slug, kept), ) for slug, item in found.items(): message = _missing_deployment_message(slug, item) _missing_bell[slug] = { "kind": "missing", "platform": slug, "name": (catalog.get_service(slug) or {}).get("name") or slug, "error": message, # What Forget acts on: the whole record, or one box's. "everywhere": item["everywhere"], "boxes": item["boxes"], } if alert_id := await database.record_alert("missing", slug, message): name = (catalog.get_service(slug) or {}).get("name") or slug title = f"CashPilot: {name} " + ("runs on no box" if item["everywhere"] else "is gone from a box") _spawn(_push_alert("missing", slug, title, message, alert_id)) _collector_alerts = [a for a in _collector_alerts if a.get("kind") != "missing"] _collector_alerts.extend(_missing_bell.values()) except Exception as exc: # noqa: BLE001 - a diagnostic must not break the health check logger.warning("Missing-deployment check failed: %s", exc) def _missing_scope(slug: str) -> set[Any]: """What a stored missing alert of ``slug`` covers, as placement.missing keys. A bare slug means the whole service: an alert restored at startup, whose boxes are not known until the next check names them. """ entry = _missing_bell.get(slug) if entry is None or "everywhere" not in entry: return {slug} keys: set[Any] = {(slug, b["worker_id"]) for b in entry.get("boxes") or []} if entry["everywhere"]: keys.add((slug, None)) return keys def _missing_deployment_message(slug: str, item: dict[str, Any]) -> str: name = (catalog.get_service(slug) or {}).get("name") or slug boxes = ", ".join(b["name"] for b in item["boxes"]) if item["everywhere"]: where = f" (it was deployed on {boxes})" if boxes else "" return ( f"CashPilot records {name} as deployed{where}, but no box runs it, so it is not earning. " "Deploy it again, or forget the deployment if you removed it on purpose." ) still = ", ".join(item["still_on"]) return ( f"{name} was deployed on {boxes}, but {'that box no longer runs' if len(item['boxes']) == 1 else 'those boxes no longer run'} it" f"{f' (it still runs on {still})' if still else ''}. " "Deploy it there again, or forget it on that box if you removed it on purpose." ) async def _push_alert(kind: str, subject: str, title: str, message: str, alert_id: int) -> None: """Push an alert out-of-band, and make a total delivery failure retryable. The dedupe row is committed BEFORE this runs, so a failed delivery used to disarm the alert for the whole cooldown window: ntfy unreachable in the minute a collector broke meant one WARNING in a container log and 24 silent hours — the push path's own failure was the one failure it could not report. When no target accepts the message, the row THIS delivery owns (record_alert's returned id) is deleted so the next collection cycle records-and-pushes again — by id, never by (kind, subject), because a recovery and a fresh failure can insert a newer row while this push was in flight, and taking that row's dedupe with ours would double-notify. The outcome feeds a metric either way so dead delivery is visible to Prometheus instead of indistinguishable from health. """ try: delivered = await notify.send(title, message, kind=kind, subject=subject) except Exception as exc: # noqa: BLE001 - a notifier crash must never kill collection logger.warning("Alert push for %s/%s crashed: %s", kind, subject, notify.redact(str(exc))) delivered = 0 if not notify.is_enabled(): # Nothing is configured: the bell is the only channel, the row must # stand (clearing it would re-insert every cycle and churn the bell), # and delivery metrics would only count an absence. return metrics.record_notify_delivery(bool(delivered)) if not delivered: # The worker push floor is armed at spawn time, before the outcome is # known — a TOTAL delivery failure must release it, or the un-dedupe # retry below would be suppressed for the whole floor window and the # "will retry on the next cycle" promise becomes false. if kind == "worker": _worker_last_offline_push.pop(subject, None) try: undeduped = await database.delete_alert(alert_id) except Exception as exc: # noqa: BLE001 undeduped = False logger.warning("Could not un-dedupe alert %s after failed delivery: %s", alert_id, exc) if undeduped: logger.warning( "No notification target accepted the %s alert for %s — will retry on the next cycle", kind, subject, ) else: # Saying "will retry" when the dedupe row survived would be false. logger.warning( "No notification target accepted the %s alert for %s, and the dedupe row could " "not be released — the next push waits out the cooldown window", kind, subject, ) async def _detect_payout(result: Any) -> dict[str, str] | None: """Notice a balance drop that looks like a cashout, and ask. Never records income on its own. A balance also falls for a provider correction or a reset session, and an unconfirmed guess written as earnings corrupts lifetime-earned permanently and invisibly. Returns the alert to show the user, or None. It has to be RETURNED rather than filed and forgotten: the bell renders one in-memory list built during the collection run, and this function runs in the success branch that never touched that list — so a detected payout was written to the database and then never shown to anybody, which defeats the point of asking. """ try: previous = await database.get_latest_balance(result.platform) if previous is None: return None try: withdrawn = await database.withdraw_recovered_payout(result.platform, result.balance - previous) except Exception as exc: # noqa: BLE001 - must never block detecting a new payout withdrawn = False logger.warning("Could not check %s for a recovered payout: %s", result.platform, exc) if withdrawn: # The drop was one bad reading, not a cashout: the question is taken back. logger.info("%s balance recovered; withdrew the probable payout question", result.platform) return None probable = payouts.detect(previous, result.balance, catalog.get_service(result.platform)) if not probable: return None payout_id = await database.record_probable_payout( platform=result.platform, amount=probable["amount"], currency=result.currency, fx_rate_usd=exchange_rates.to_usd(1.0, result.currency), ) if payout_id is None: # One already pending for this platform; a second prompt for the # same event teaches the user to dismiss them. return None if alert_id := await database.record_alert("payout", result.platform, probable["reason"]): # The one alert that asks a QUESTION — unanswered, the payout never # counts toward lifetime earnings, so the headline number stays # wrong until a human replies. It was the only pushable kind that # never pushed: recorded, belled, and then waiting for someone to # happen to open the UI. The record_alert gate was already here, # its boolean just went unread. _spawn( _push_alert( "payout", result.platform, f"CashPilot: {result.platform} balance dropped — was this a payout?", probable["reason"], alert_id, ) ) return {"kind": "payout", "platform": result.platform, "error": probable["reason"]} except Exception as exc: # Earnings collection must not fail because payout detection did. logger.warning("Payout detection failed for %s: %s", getattr(result, "platform", "?"), exc) return None async def _pending_payout_alerts(seen: set[str] | None = None) -> list[dict[str, str]]: """One alert per payout still waiting for a yes or no. The question outlives the collection run that noticed it. Rebuilding the bell only from what was just detected made the prompt disappear on the next run, which is the one thing a prompt must not do — an unanswered payout is unanswered until the user answers it. """ seen = seen or set() try: rows = await database.get_payouts() except Exception as exc: logger.warning("Could not load pending payouts for the alert bell: %s", exc) return [] pending: list[dict[str, str]] = [] for row in rows: platform = row.get("platform", "") if row.get("confirmed") or platform in seen: continue seen.add(platform) amount = row.get("amount") currency = row.get("currency") or "" pending.append( { "kind": "payout", "platform": platform, "error": f"Balance dropped by {amount:.2f} {currency}".rstrip() + " — was this a payout? Confirm or reject it." if isinstance(amount, int | float) else "A balance drop looks like a payout — confirm or reject it.", } ) return pending async def _collect_bounded(collector) -> Any: """Run a single collector's `collect()` under the shared concurrency limit. A raised exception is converted into an EarningsResult HERE, while the collector -- and with it the platform name -- is still in hand. It used to propagate to the gather(), where the platform was unrecoverable: the failure was logged as an anonymous line and produced no alert, no bell entry and no metric. A collector that raised was invisible to the user (CashPilot-5bdm). """ from app.collectors import base as collectors_base async with _collection_semaphore: try: return await collector.collect() except Exception as exc: collectors_base.log_failure(logger, collector.platform or type(collector).__name__, exc) return collectors_base.EarningsResult( platform=collector.platform or type(collector).__name__, balance=0.0, error=collectors_base.describe_exception(exc), error_kind=collectors_base.classify_exception(exc), ) async def _offline_worker_alerts() -> list[dict[str, str]]: """Bell entries for every worker currently marked offline. Derived from the workers table on each rebuild rather than carried in memory: the collection loop replaces _collector_alerts wholesale every hour, and an entry that lived only in the list would be dropped there — silently un-reporting a machine that is still down. Recovery needs no bookkeeping either: the next heartbeat flips the row back to online and the entry stops being derived. """ try: workers = await database.list_workers() except Exception as exc: # noqa: BLE001 - the bell must not take down collection logger.warning("Could not derive offline-worker alerts: %s", exc) return [] entries: list[dict[str, str]] = [] for w in workers: if w.get("status") == "offline" and w.get("last_heartbeat"): entries.append( { "kind": "worker", "platform": w["name"], # Identity travels beside the display name: recovery and # dedupe match on client_id, because two hosts can share a # hostname and hostnames churn across recreates. "client_id": w.get("client_id") or w["name"], "error": ( f"Offline since {w['last_heartbeat']} UTC. Its containers keep running " "and earning, but CashPilot cannot see or manage them until it reconnects." ), } ) return entries def _quiet_days_by_platform() -> dict[str, int]: """Each catalog entry's collector.quiet_days, for the services that declare one.""" out: dict[str, int] = {} for svc in catalog.get_services(): raw = (svc.get("collector") or {}).get("quiet_days") with contextlib.suppress(TypeError, ValueError): if raw is not None and int(raw) > 0: out[svc["slug"]] = int(raw) return out def _flatline_message(flat: dict[str, Any]) -> str: name = (catalog.get_service(flat["platform"]) or {}).get("name") or flat["platform"] days = flat.get("days_flat") span = f" ({days} days)" if days else "" if flat.get("last_increase") is None: return ( f"{name} has not earned anything since CashPilot started reading it" f"{' on ' + str(flat['since']) if flat.get('since') else ''}{span}. " "Common causes: the device is not linked to your account, the provider has no demand for your " "IP or region yet, or the node cannot be reached from outside. " "This clears by itself when the balance grows." ) gap = flat.get("usual_gap_days") rhythm = ( f"it usually earns every {'day' if gap == 1 else f'{gap} days'}" if gap else "it earned before, though too rarely to say how often" ) return ( f"{name} hasn't earned since {flat['last_increase']}{span}; {rhythm}. " "Common causes: no demand for your IP or region right now, the device was unlinked from your " "account, or the node cannot be reached from outside. Services often go quiet for a while and " "pick up again; this clears by itself when the balance grows." ) # One push per quiet spell. The default 24 h cooldown would repeat the push # every day the balance stays flat; the alert row is cleared when the balance # grows, and that is what re-arms it. _FLATLINE_COOLDOWN_HOURS = 24 * 365 * 10 async def _flatline_check( collected: set[str] | None = None, deployments: list[dict[str, Any]] | None = None, statuses: list[dict[str, Any]] | None = None, unlinked: set[str] | None = None, ) -> list[dict[str, str]]: """Give a heads-up on a service whose balance has stopped growing. A container can be up and a collector can authenticate happily while the balance never moves. Every other view of the system looks healthy, so nothing else would surface it. Soft evidence, so it is held back unless everything else is fine: the collector succeeded this run (``collected``), and a service with a container has a copy running on an online worker (``statuses`` only lists online workers). A container that is down has its own alert, and an offline worker its own. When a precondition fails the alert is neither raised nor cleared; it clears only when the balance grows, so a payout or a restart cannot re-arm it into a second push. Never raises: this is a diagnostic, and a diagnostic must not be able to take down the collection run it is diagnosing. """ bell: list[dict[str, str]] = [] try: first_deployed = {d["slug"]: d.get("deployed_at") or "" for d in deployments or []} flat_services = await database.get_flatlined_services( quiet_days=_quiet_days_by_platform(), first_deployed=first_deployed ) flat_now = {f["platform"] for f in flat_services} # Clear the row for anything that has started earning again, so its # next quiet spell notifies instead of being deduped into silence. for recovered in await database.get_alert_subjects("flatline") - flat_now: await database.clear_alerts("flatline", recovered) logger.info("%s is earning again; cleared its flatline alert", recovered) has_container = {d["slug"] for d in deployments or [] if d.get("status") != "external"} running = {s["slug"] for s in statuses or [] if s.get("status") == "running"} for flat in flat_services: platform = flat["platform"] if collected is not None and platform not in collected: continue # A collector found a deployed device that is not on the account, # and its notice already names that cause; a vaguer heads-up for # the same service would only repeat it. if unlinked and platform in unlinked: continue if statuses is not None and platform in has_container and platform not in running: continue message = _flatline_message(flat) # Built for EVERY flatlined service, not only the ones that clear # record_alert's cooldown. The cooldown exists to stop repeat # NOTIFICATIONS; the bell is a standing statement of what is true # right now. bell.append({"kind": "flatline", "platform": platform, "error": message}) if alert_id := await database.record_alert( "flatline", platform, message, cooldown_hours=_FLATLINE_COOLDOWN_HOURS ): name = (catalog.get_service(platform) or {}).get("name") or platform _spawn(_push_alert("flatline", platform, f"CashPilot: {name} has gone quiet", message, alert_id)) except Exception as exc: logger.warning("Flatline check failed: %s", exc) # The check's own failure must not render as health: an empty return # here is indistinguishable from "nothing is flatlined", so a broken # query would silently retire the only detector this class of fault has. bell.append( { "kind": "flatline", "platform": "flatline check", "error": "The flatline check itself failed — earning anomalies are NOT being watched. See server logs.", } ) return bell async def _warm_collector_alerts() -> None: """Restore persisted collector alerts into the in-memory list the UI bell reads. Without this, a restart clears the bell while the collector is still broken, and the operator is told everything is fine until the next hourly run. """ global _collector_alerts try: stored = await database.list_alerts(limit=100) except Exception as exc: logger.warning("Could not restore persisted alerts: %s", exc) return # Newest-first, keeping only the most recent row per subject: a normal run puts # exactly one entry per failing platform in this list, and the restored bell must # look the same rather than showing one row per historical message. seen: set[str] = set() restored: list[dict[str, str]] = [] for alert in stored: # Payouts belong here too. Dropping every non-collector kind meant a # detected payout was recorded and then never shown, so the prompt the # user is supposed to answer never appeared. Dedup by kind AND subject: # a platform can legitimately have both a failing collector and a # pending payout, and they are different things to tell someone. # flatline belongs here for the same reason payout does: it is recorded, # and without this it is dropped on the way to the bell, so a restart # silently clears a warning about a service that is still not earning. # worker likewise: an offline machine must survive a UI restart. And # notice: #326 made notices durable and deduped, but a restart dropped # them from the bell while their 24h push-dedupe survived — so a # recovery inside the gap skipped clear_alerts and the stale row # swallowed the next genuine warning. if alert["kind"] not in ( "collector", "payout", "flatline", "worker", "notice", "container", "missing", "account_disabled", ): continue key = f"{alert['kind']}:{alert['subject']}" if key in seen: continue seen.add(key) entry = {"kind": alert["kind"], "platform": alert["subject"], "error": alert["message"]} if alert.get("category"): entry["category"] = alert["category"] restored.append(entry) # Worker alerts are stored under client_id (the identity); the bell shows # people a NAME. Resolve display names in one pass, keeping the client_id # beside it for recovery matching. A worker deleted since the alert was # stored keeps the raw id — still actionable, never wrong. worker_entries = [e for e in restored if e["kind"] == "worker"] if worker_entries: with contextlib.suppress(Exception): names = {w.get("client_id"): w["name"] for w in await database.list_workers()} for e in worker_entries: e["client_id"] = e["platform"] e["platform"] = names.get(e["platform"], e["platform"]) for e in restored: if e["kind"] == "account_disabled": _account_disabled.setdefault(e["platform"], e["error"]) # Which source raised it is not stored; until a heartbeat says, # treat it as the logs' so a check without data keeps it. _disabled_by_logs.add(e["platform"]) # A capped proxy is restored from the proxies table, which says whether the # cap still holds; the alerts table only says it once did. restored.extend(await _proxy_cap_alerts()) _collector_alerts = _with_account_disabled(restored) # The collection rebuild re-adds _missing_bell, not the restored list, so a # restored missing alert goes there too, with no scope: until the next # health check names its boxes, the bell offers no Forget for it. for e in restored: if e["kind"] == "missing": name = (catalog.get_service(e["platform"]) or {}).get("name") or e["platform"] _missing_bell.setdefault(e["platform"], {**e, "name": name}) # A restart must not make the bell claim nothing has ever been checked. Any # stored alert, or any earnings row, is proof that a collection ran. global _collection_has_run if restored: _collection_has_run = True else: try: _collection_has_run = bool(await database.get_earnings_summary()) except Exception as exc: # noqa: BLE001 - a warm-up must not block startup logger.warning("Could not tell whether a collection has run before: %s", exc) async def _track_fully_configured_services() -> int: """Give every fully-credentialled service a deployment row, so it is collected. Collection iterates DEPLOYMENT ROWS (``make_collectors``), so a service with no row is never collected however complete its credentials are. Saving credentials has created that row since the fix in #187 — but only at save time, and nothing has ever repaired a database written before it. That leaves the upgrade path in exactly the state the fix was meant to end: credentials stored under an older version show the green "Configured" badge, no collector is ever built for them, and no earnings arrive. Nothing prompts the user to re-save, because as far as the UI is concerned everything is already correct. So this runs once per start and is idempotent — a slug that already has a row of any status is left exactly as it is, because that row may carry a real container id and spec that must not be replaced with an empty placeholder. Returns the number of rows created, for the log line and the tests. """ from app.collectors import fully_configured_slugs try: config = await database.get_config() or {} if not isinstance(config, dict): return 0 existing = {d.get("slug") for d in await database.get_deployments()} # Removed from the whole fleet by the user: saved credentials are not a # reason to bring it back. Deploying it, or saving a changed value or a # retyped secret for it in Settings, lifts this (api_set_config, # database.save_deployment). removed = await database.removed_slugs() tracked = 0 for slug in sorted(fully_configured_slugs(config)): if slug in existing or slug in removed or not catalog.get_service(slug): continue await database.save_deployment(slug=slug, container_id="", status="external") tracked += 1 if tracked: logger.info( "Started tracking %d service(s) whose credentials were stored before " "saving them began tracking; their earnings will be collected from now on", tracked, ) return tracked except Exception as exc: # noqa: BLE001 - bookkeeping must never block startup logger.warning("Could not backfill tracking for stored credentials: %s", exc) return 0 #: A pending registration is retried this long after the first refusal, then #: twice as long each time, up to the cap; after the give-up it is the user's. DEVICE_LINK_FIRST_RETRY = timedelta(minutes=6) DEVICE_LINK_MAX_RETRY = timedelta(hours=4) DEVICE_LINK_GIVE_UP = timedelta(hours=24) def _db_time(when: datetime) -> str: """The format SQLite's datetime('now') writes, so stored times compare as text.""" return when.astimezone(UTC).strftime("%Y-%m-%d %H:%M:%S") def _device_name(device_id: str, env: dict[str, Any]) -> str: name = str(env.get("DEVICE_NAME") or "").strip() if not name or "${" in name: # an unfilled catalog default is not a name name = f"cashpilot-{device_id[:8]}" return name async def _try_register(slug: str, device_id: str, name: str) -> bool | str: """One registration attempt: True, False or REGISTER_PENDING. Never raises.""" from app.collectors import build_collector try: collector = build_collector(slug, await database.get_config() or {}) except Exception: return False if collector is None: return False try: result = await asyncio.wait_for(collector.register_device(device_id, name), 20) return result if result == REGISTER_PENDING else bool(result) except Exception as exc: logger.warning("Could not add the %s device to the account: %s", slug, notify.redact(str(exc))) return False finally: await collector.close() async def _register_device(slug: str, device_id: str, env: dict[str, Any]) -> bool | None: """Add a deployed device to the user's account with the stored collector credentials. Only where the provider's API can do it and the credentials are saved; anything else is False, and the deploy form shows the manual step instead. None when the provider has not seen the new device yet: the registration is stored and retried by the scheduler (_retry_device_links). Never fails the deploy, which has already happened. """ result = await _try_register(slug, device_id, _device_name(device_id, env)) if result == REGISTER_PENDING: now = datetime.now(UTC) try: await database.save_pending_device_link( slug, device_id, _device_name(device_id, env), _db_time(now + DEVICE_LINK_FIRST_RETRY) ) except Exception as exc: logger.warning("Could not store the %s registration to retry: %s", slug, exc) return False return None return bool(result) async def _device_still_deployed(slug: str, device_id: str) -> bool: """Whether ``device_id`` still runs: a deployment's id, or a proxied copy's derived one.""" link = (catalog.get_service(slug) or {}).get("device_link") or {} if not link.get("id_env"): return False deployed = await database.get_deployed_env_values(slug, link["id_env"]) if device_id in deployed: return True for copy in await database.list_proxied_copies(): if copy["slug"] != slug: continue for original in deployed: try: if proxies.copy_device_id(original, copy["proxy_id"]) == device_id: return True except ValueError: continue return False async def _retry_device_links() -> None: """Retry the registrations a provider asked to try again later, with backoff.""" now = datetime.now(UTC) for row in await database.list_pending_device_links(): if row["gave_up"] or row["next_at"] > _db_time(now): continue slug, device_id = row["slug"], row["device_id"] if not await _device_still_deployed(slug, device_id): await database.delete_pending_device_link(slug, device_id) continue result = await _try_register(slug, device_id, row["name"]) if result is True: await database.delete_pending_device_link(slug, device_id) await database.record_health_event(slug, "device_linked", f"{device_id} added to the account") logger.info( "Added the %s device %s to the account after %d retries", slug, device_id[:8], row["attempts"] + 1 ) continue attempts = row["attempts"] + 1 first = datetime.strptime(row["first_at"], "%Y-%m-%d %H:%M:%S").replace(tzinfo=UTC) if now - first >= DEVICE_LINK_GIVE_UP: await database.update_pending_device_link( slug, device_id, attempts=attempts, next_at=row["next_at"], gave_up=True ) svc = catalog.get_service(slug) or {} step = catalog.device_link_step(svc, {(svc.get("device_link") or {}).get("id_env"): device_id}) or {} message = ( f"{svc.get('name') or slug} never accepted the device {device_id} after a day of retries, " f"so it earns for nobody. {step.get('steps', '')} {step.get('url', '')}" ).strip() if alert_id := await database.record_alert("notice", slug, message): _spawn( _push_alert( "notice", slug, f"CashPilot: {svc.get('name') or slug} device not on your account", message, alert_id, ) ) continue wait = min(DEVICE_LINK_FIRST_RETRY * (2**attempts), DEVICE_LINK_MAX_RETRY) await database.update_pending_device_link( slug, device_id, attempts=attempts, next_at=_db_time(now + wait), gave_up=False ) async def _retrying_device_ids() -> dict[str, set[str]]: """Per slug, the device ids _retry_device_links is still adding to the account. Only inside the retry window: a row the retry job never got to mark as given up (the job failing every time) must not hide the notice for ever. A failed read answers {}: the plain notice then shows, as it did before. """ try: rows = await database.list_pending_device_links() except Exception as exc: logger.warning("Could not read the device registrations being retried: %s", exc) return {} window_start = _db_time(datetime.now(UTC) - DEVICE_LINK_GIVE_UP) retrying: dict[str, set[str]] = {} for row in rows: if not row["gave_up"] and str(row.get("first_at") or "") > window_start: retrying.setdefault(row["slug"], set()).add(row["device_id"]) return retrying def _unlinked_devices_warning(result: Any, retrying: set[str] | frozenset[str] = frozenset()) -> str: """The notice for deployed devices the provider says are not on the account. A device in ``retrying`` is left out: the app is still adding it, so asking the user to add it by hand is wrong, and the notice would take the day's push from the give-up message that follows if the retries fail. """ ids = sorted(set(getattr(result, "unlinked_devices", None) or []) - set(retrying)) if not ids: return "" svc = catalog.get_service(result.platform) or {} link = svc.get("device_link") or {} steps = [] for device_id in ids: step = catalog.device_link_step(svc, {link.get("id_env"): device_id}) or {} steps.append(f"{device_id} ({step['url']})" if step.get("url") else device_id) name = svc.get("name") or result.platform which = "A deployed device is" if len(ids) == 1 else f"{len(ids)} deployed devices are" earns = "earns" if len(ids) == 1 else "earn" return ( f"{which} not on your {name} account and {earns} for nobody: {'; '.join(steps)}. {link.get('steps', '')}" ).strip() async def _run_collection() -> None: """Collect earnings from all deployed services that have collectors.""" global _collector_alerts, _collected_ok_last_run, _collector_says_disabled if _collection_lock.locked(): logger.info("Collection already in progress, skipping") return async with _collection_lock: success = True start_time = 0.0 collectors: list = [] try: start_time = metrics.record_collection_start() deployments = await database.get_deployments() config = await database.get_config() or {} if not isinstance(config, dict): config = {} from app.collectors import _close_stale, make_collectors collectors = make_collectors(deployments, config) await _close_stale() for c in collectors: try: await _attach_node_readings(c) except Exception as exc: # noqa: BLE001 - the collector reports the empty read logger.warning("Could not read node figures for %s: %s", getattr(c, "platform", "?"), exc) # Platforms whose collector was asked to check deployed device ids. device_checked: set[str] = set() for c in collectors: slug = getattr(c, "platform", "") link = (catalog.get_service(slug) or {}).get("device_link") if slug else None if not isinstance(link, dict): continue try: c.device_ids = await database.get_deployed_env_values(slug, link["id_env"]) except Exception as exc: # Only this service's device check is lost, never the run. logger.warning("Could not read the deployed %s device ids: %s", slug, exc) continue if c.device_ids: device_checked.add(slug) results = await asyncio.gather(*(_collect_bounded(c) for c in collectors), return_exceptions=True) retrying = await _retrying_device_ids() alerts: list[dict[str, str]] = [] # Platforms that were already failing before this run: used to detect a # recovery, so a service that breaks again later notifies again rather # than being deduped into silence forever. previously_alerting = {a["platform"] for a in _collector_alerts} # Notices tracked separately: a warning-free success is a notice's # recovery, and conflating the two would clear the wrong kind. previously_noticing = {a["platform"] for a in _collector_alerts if a.get("kind") == "notice"} platforms_ok = 0 collected_ok: set[str] = set() disabled_now: set[str] = set() unlinked_ok: set[str] = set() for result in results: if isinstance(result, Exception): # Redacted BEFORE it is logged. A collector exception is # usually an httpx error that embeds the offending header, # which for several providers IS the live credential. logger.warning("Collector raised exception: %s", notify.redact(str(result))) success = False continue if result.error: # Redact FIRST. The comment below explains why the alert is # sanitised; this line used to log the raw string one step # earlier and put the credential in the container log # anyway, defeating the whole exercise. logger.warning("Collection error for %s: %s", result.platform, notify.redact(result.error)) # Redact ONCE, here, so the same sanitized string is what gets shown, # stored and sent. Collector errors are usually str(exc), and an httpx # exception embeds the offending header or URL — which for several # providers is a live credential. The alert is now durable and readable # by any authenticated role, so it must never hold a secret. safe_error = notify.redact(result.error) error_kind = getattr(result, "error_kind", None) says_disabled = credential_test.classify(safe_error) == credential_test.ACCOUNT_DISABLED if says_disabled: disabled_now.add(result.platform) entry = {"kind": "collector", "platform": result.platform, "error": safe_error} if error_kind: entry["category"] = error_kind alerts.append(entry) metrics.record_collection_error(result.platform) # Push out-of-band only the FIRST time this failure appears — a # collector broken for a week must not notify every single hour. # A disabled account has its own alert (_account_disabled_check); # a "collector failed" push beside it would point at the # credentials, which are fine. if says_disabled or result.platform in _account_disabled: pass elif alert_id := await database.record_alert( "collector", result.platform, safe_error, category=error_kind ): # The push is the only channel an unattended install # has, so the taxonomy must reach it too: "rotate your # cookie" and "provider hiccup" are different asks. cause = { "auth": " (credential rejected)", "transient": " (provider unreachable)", "shape": " (page changed)", }.get(error_kind or "", "") _spawn( _push_alert( "collector", result.platform, f"CashPilot: {result.platform} collector failed{cause}", safe_error, alert_id, ) ) else: # A caveat on a SUCCESSFUL reading. It is shown, because the # figure is partial and the user should know, but it is not # a failure: the balance below is stored exactly as any other. if unlinked := _unlinked_devices_warning(result, retrying.get(result.platform, set())): result.warning = f"{result.warning} {unlinked}" if result.warning else unlinked if getattr(result, "warning", None): safe_warning = notify.redact(result.warning) logger.info("Collector notice for %s: %s", result.platform, safe_warning) alerts.append({"kind": "notice", "platform": result.platform, "error": safe_warning}) # Push out-of-band with the same once-per-window dedupe # errors get. A notice used to reach ONLY the bell, # which is read by whoever happens to open the UI — and # "nobody looked for days" is the incident this class # of warning (an unreachable storj node) comes from. if alert_id := await database.record_alert("notice", result.platform, safe_warning): _spawn( _push_alert( "notice", result.platform, f"CashPilot: {result.platform} needs attention", safe_warning, alert_id, ) ) elif ( result.platform in previously_noticing and result.platform in device_checked and getattr(result, "unlinked_devices", None) is None ): # The device check could not finish (provider error, rate # limit): that proves nothing, so the notice stays rather # than being cleared now and pushed again next hour. alerts.extend( a for a in _collector_alerts if a.get("kind") == "notice" and a["platform"] == result.platform ) elif result.platform in previously_noticing: # A warning-free success is the notice's recovery: drop # the stored row so the NEXT warning pushes again rather # than being deduped into silence forever. await database.clear_alerts("notice", result.platform) earned = None if getattr(result, "nodes", None) is not None: # Read node by node (issue #488): the balance is every # known node's unsettled amount, and it falls by itself # each time a node settles, so it is never asked about # as a payout. The running total carries the earnings. totals = await database.record_node_figures(result.platform, result.nodes, result.currency) result.balance = totals["balance"] earned = totals["earned"] else: # Compare against the last reading BEFORE writing the new # one: a payout is only visible as the step between two # snapshots. payout_alert = await _detect_payout(result) if payout_alert: alerts.append(payout_alert) await database.upsert_earnings( platform=result.platform, balance=result.balance, currency=result.currency, # Snapshot the rate now. to_usd(1.0, cur) is the cur -> USD # rate (1.0 for USD, None if unknown); rates are only cached # live, so this is the only chance to record what this reading # was actually worth. fx_rate_usd=exchange_rates.to_usd(1.0, result.currency), earned=earned, ) logger.info("Collected %s: %.4f %s", result.platform, result.balance, result.currency) platforms_ok += 1 collected_ok.add(result.platform) if getattr(result, "unlinked_devices", None): unlinked_ok.add(result.platform) if result.platform in previously_alerting: # Recovered — drop the stored alert so a future failure counts # as new and notifies again. await database.clear_alerts("collector", result.platform) # Re-add every payout still awaiting an answer, not just ones # detected on THIS run. `record_probable_payout` returns None while # one is already pending, so a freshly detected payout produced an # alert once and then vanished from the bell on the very next # collection — before the user had a chance to confirm or reject # it. The pending rows in the database are the real source of # truth for "there is still a question outstanding", so the bell is # rebuilt from them rather than from what happened to be noticed in # the last few seconds. alerts.extend( await _pending_payout_alerts(seen={a["platform"] for a in alerts if a.get("kind") == "payout"}) ) # Flatline is the one failure mode nothing else can surface — the # container is up and the collector authenticates fine. It reached # the database and the notifier, but never this list, and on a # default install no notifier is configured (docker-compose.yml sets # none of the NTFY/WEBHOOK/TELEGRAM variables), so the only place the # user would ever see it was the one place it never went. The bell # said "All collectors healthy". try: fleet = await _get_all_worker_containers() except Exception as exc: # noqa: BLE001 - unknown fleet: hold the heads-up back logger.warning("Could not read the fleet for the flatline check: %s", exc) fleet = [] alerts.extend(await _flatline_check(collected_ok, deployments, fleet, unlinked=unlinked_ok)) _collected_ok_last_run = collected_ok await _record_balance_reads(collected_ok) # Sticky until a balance is read: a disabled account's later logins # often come back as a rate limit, which says nothing new. A service # the user removed has no collector to read that balance, so it # leaves here instead, or its bell entry stayed until a restart. deployed = {d.get("slug") for d in deployments} _collector_says_disabled = ((_collector_says_disabled | disabled_now) - collected_ok) & deployed await _account_disabled_check(fleet) # Offline workers re-derived every rebuild, or the hourly wholesale # replacement of this list would silently un-report a dead machine. alerts.extend(await _offline_worker_alerts()) # Kept by the health check, which runs every five minutes; without # this the hourly rebuild would drop a down container from the bell. alerts.extend(_container_down_bell.values()) alerts.extend(_missing_bell.values()) alerts.extend(_activity_bell) alerts.extend(await _proxy_cap_alerts()) _collector_alerts = _with_account_disabled(alerts) if collectors and platforms_ok == 0: # A run in which EVERY collector failed used to be recorded as # a SUCCESS: _collect_bounded converts exceptions into # EarningsResult.error, so the isinstance branch above never # fired and nothing else touched the flag. That refreshed # collection_last_success_timestamp hourly through a total # outage — both documented Prometheus alerts (CollectionFailing, # EarningsStale) were structurally unable to fire, proven by # execution. A PARTIAL failure stays a success: one chronically # broken collector must not freeze the staleness stamp and # false-alarm while fourteen others are collecting fine, and an # install with no collectors has nothing to be stale about. success = False except Exception as exc: logger.error("Collection run failed: %s", exc) success = False platforms_ok = 0 # Keep the per-platform entries: the next run derives "which platforms were # failing" from this list, so clobbering it would make every platform # permanently unrecoverable — its stored alert would never be cleared and # its next real failure would be deduped into silence. _collector_alerts = [ *(a for a in _collector_alerts if a.get("platform") != "collection"), {"platform": "collection", "error": "Collection run failed — see server logs"}, ] finally: metrics.record_collection_end(start_time, success, platforms_ok, len(collectors)) # Set even on a failed run: the bell's question is "has anything # looked", and a run that tried and failed HAS looked — its failure # is in the alert list the bell is about to show. global _collection_has_run, _last_collection_finished _collection_has_run = True # "The machinery ran", success or not — /api/health and the bell # measure staleness of THIS stamp. A run that wedges on the # collection lock returns early above and never reaches here, so # the age keeps growing through exactly the failure the skip line # used to disguise as routine. _last_collection_finished = time.monotonic() # --------------------------------------------------------------------------- # App lifecycle # --------------------------------------------------------------------------- async def _run_data_retention() -> None: """Purge data older than 400 days.""" try: deleted = await database.purge_old_data() if deleted: logger.info("Data retention: purged %d old rows", deleted) except Exception as exc: logger.warning("Data retention error: %s", exc) #: Config key holding when the last VACUUM finished (ISO 8601, UTC). _LAST_VACUUM_KEY = "_last_vacuum_at" _VACUUM_EVERY = timedelta(days=7) async def _run_vacuum() -> None: """Reclaim disk left by retention deletes (SQLite does not auto-shrink). Runs at startup and on a weekly timer, but only when the last VACUUM is more than a week old. The timer alone counts from process start, and every release recreates this container, so an install that restarts more than once a week never reached it. The stamp is what makes a startup run cheap: a restart soon after a VACUUM does not rewrite the whole file again. """ try: last = datetime.fromisoformat(str(await database.get_config(_LAST_VACUUM_KEY))) if datetime.now(UTC) - last < _VACUUM_EVERY: return except Exception: pass # no stamp yet, or an unreadable one: vacuum rather than never try: await database.vacuum_database() await database.set_config(_LAST_VACUUM_KEY, datetime.now(UTC).isoformat()) logger.info("Database VACUUM complete") except Exception as exc: logger.warning("Database VACUUM error: %s", exc) # A recovery re-arms the offline alert only after this many CONSECUTIVE sweeps # online. One heartbeat is not a recovery: an Android phone in Doze wakes every # ~12 minutes, heartbeats once and sleeps again — and when that single beat # cleared the alert row immediately, every nap became a "new" offline episode # with its own push. Seven identical Telegram messages in ~80 minutes, live. # Two sweeps (~4 minutes online) is enough to tell a recovery from a blip while # a genuinely recovered server re-arms almost immediately. In-memory on # purpose: a UI restart forgets streaks, which at worst re-arms one alert # early — the 24h record cooldown still caps the damage. _SUSTAINED_RECOVERY_SWEEPS = 2 #: Sentinel streak value: this recovery's durable clear SUCCEEDED — stop #: clearing until the next offline episode resets the streak. _STREAK_CLEARED = -1 _worker_online_streak: dict[str, int] = {} #: Floor between offline PUSHES for one worker, independent of episodes. #: Sustained-recovery damping alone was defeated by a phone whose NORMAL duty #: cycle is 7-10 minutes awake, 2-3 minutes dozing: every wake window passed #: the sustained bar, re-armed the alert, and the next nap pushed — eight #: pushes in two hours, live, AFTER the episode damping shipped. A worker #: whose ordinary operation is rhythmic flapping can never be rate-limited by #: episode detection, so the cap sits on the notification itself: the first #: offline push is immediate, repeats for the same worker wait out the floor. #: The bell and dashboard stay real-time; only the phone-buzz is capped. #: In-memory on purpose — a restart forgets the floor, which at worst costs #: one early push. _WORKER_OFFLINE_PUSH_COOLDOWN_S = 6 * 3600 _worker_last_offline_push: dict[str, float] = {} async def _check_stale_workers() -> None: """Mark workers as offline if stale, and purge never-enrolled workers offline > 1 hour. A worker that HAS enrolled a per-worker key (``api_key_enc`` set) is never auto-deleted here, even after a long outage: the host persists that same key locally and re-presents it on its next heartbeat. Deleting the row would leave no match for that key, and since a confirmed worker's shared bootstrap key is refused too, it would be rejected forever with no way to re-enroll — a permanent fleet lockout after a reboot/maintenance window. Only a worker that never completed enrollment is purged automatically; removing an enrolled worker is a deliberate action via the UI. This sweep is also the ONLY place the offline alert is cleared — see the sustained-recovery branch. The heartbeat route deliberately does not clear. """ try: workers = await database.list_workers() except Exception as exc: logger.warning("Stale worker check error: %s", exc) return now = datetime.now(UTC) purge_cutoff = now - timedelta(hours=1) global _collector_alerts for w in workers: try: last_hb = w.get("last_heartbeat") if not last_hb: continue cid = w.get("client_id") or w["name"] last = datetime.fromisoformat(last_hb).replace(tzinfo=UTC) cutoff = now - timedelta(seconds=_stale_after_seconds(w)) if w["status"] == "online" and last < cutoff: _worker_online_streak.pop(cid, None) # Conditional on the heartbeat the decision was based on: a # recovery heartbeat landing between this sweep's read and its # write must win, or the worker would be alerted offline at its # exact moment of coming back. if not await database.mark_worker_offline_if_unchanged(w["id"], last_hb): continue logger.info("Worker '%s' marked offline (last heartbeat: %s)", w["name"], last_hb) # The 42-hour incident, UI side: this transition used to be the # log line above and nothing else — the UI KNEW and told nobody. # Same alert/push/recovery pattern as a failing collector. await _raise_worker_offline_alert(w) elif w["status"] == "offline": _worker_online_streak.pop(cid, None) if last < purge_cutoff and not w.get("api_key_enc"): await database.delete_worker(w["id"]) logger.info("Purged stale unenrolled worker '%s' (offline since %s)", w["name"], last_hb) else: # Still offline: re-attempt the durable alert. record_alert # dedupes inside its window, so this is a no-op two minutes # after a SUCCESSFUL insert — but an insert or push that # FAILED at transition time (DB blip, notifier down) gets # retried instead of being lost to the already-offline # state, which used to make the miss permanent. await _raise_worker_offline_alert(w) elif w["status"] == "online": # Sustained recovery, and the ONLY place the offline alert is # cleared. Clearing re-arms the next push, so a clear on the # FIRST heartbeat turned every Doze nap of a phone worker into # a fresh alerted episode (~5 pushes/hour, live). The retry # state lives in the streak itself: at or past the threshold # the clear runs every sweep until it SUCCEEDS (a raise lands # in this loop's per-worker handler and the streak stays # armed), and only success parks the streak on the sentinel — # so a failed durable clear cannot be stranded by the hourly # bell rebuild dropping the in-memory entry. streak = _worker_online_streak.get(cid, 0) if streak == _STREAK_CLEARED: continue streak += 1 _worker_online_streak[cid] = streak if streak >= _SUSTAINED_RECOVERY_SWEEPS: await database.clear_alerts("worker", cid) had_bell = any(a.get("kind") == "worker" and a.get("client_id") == cid for a in _collector_alerts) _collector_alerts = [ a for a in _collector_alerts if not (a.get("kind") == "worker" and a.get("client_id") == cid) ] _worker_online_streak[cid] = _STREAK_CLEARED if had_bell: logger.info("Cleared offline alert for recovered worker '%s'", w["name"]) except Exception as exc: logger.warning("Stale worker check error for worker '%s': %s", w.get("name", w.get("id")), exc) #: Windowed health signals that currently hold, per running copy: #: (slug, worker_id, instance) -> the rendered ``means`` of each one. Rebuilt by #: every sweep, so a problem that went away clears on the next one. In memory on #: purpose: a restart forgets it for at most one sweep. _windowed_hints: dict[tuple[str, int, int], list[str]] = {} # The startup sweep and the scheduled one must not overlap: the older would # finish last and put back stale hints. _windowed_lock = asyncio.Lock() _DOCKER_POOLS = ipaddress.ip_network("172.16.0.0/12") def _private_ipv4(url: str) -> str | None: """The host of a worker URL when it looks like the box's LAN address, else None. A worker on Docker's bridge network auto-detects its CONTAINER address (172.17.x and the rest of Docker's 172.16.0.0/12 pools), which no router can forward to, so that range is never named. """ host = urllib.parse.urlparse(url or "").hostname or "" try: addr = ipaddress.ip_address(host) except ValueError: return None if addr.version != 4 or not addr.is_private or addr.is_loopback or addr in _DOCKER_POOLS: return None return host async def _render_signal_means(signal: dict[str, Any], slug: str, worker: dict[str, Any] | None) -> str: """The signal's ``means`` with ${VAR} filled from what this copy was deployed with. ${HOST_ADDRESS} is the box's LAN address when the worker reports one, so the hint can name the exact router rule to create. """ service = catalog.get_service(slug) or {} env = {var["key"]: str(var.get("default", "")) for var in (service.get("docker") or {}).get("env", [])} wid = (worker or {}).get("id") recorded = await database.get_deployment_spec(slug, worker_id=wid) if wid is not None else None recorded_env = {k: str(v) for k, v in ((recorded or {}).get("env") or {}).items() if str(v)} env.update(recorded_env) env["HOST_ADDRESS"] = _private_ipv4((worker or {}).get("url", "")) or "this machine's LAN address" text = " ".join(str(signal.get("means") or "").split()) # A copy deployed before a setting existed does not run with its default yet # (Mysterium without UDP_PORTS picks from 10000:60000), so naming the default # alone would send the operator to forward a range the node does not use. used = set(re.findall(r"\$\{(\w+)\}", text)) - {"HOST_ADDRESS"} if recorded is not None and any(key not in recorded_env for key in used): text = ( "This copy was deployed before CashPilot set these values, so it does not use them yet: " "redeploy it from the dashboard first. " + text ) return re.sub(r"\$\{(\w+)\}", lambda m: env.get(m.group(1), m.group(0)), text) async def _check_windowed_signals() -> None: """Judge every windowed health signal of every running copy (producer_state.is_windowed). A copy whose read fails keeps its last verdict rather than losing it to a blip; a copy no longer running, or no longer declaring the signal, drops it. """ async with _windowed_lock: await _sweep_windowed_signals() async def _sweep_windowed_signals() -> None: global _windowed_hints try: containers = await _get_all_worker_containers() except Exception as exc: logger.warning("Windowed signal check skipped: %s", exc) return hints: dict[tuple[str, int, int], list[str]] = {} workers: dict[int, dict[str, Any] | None] = {} for c in containers: slug = egress.container_slug(c) wid = c.get("_worker_id") if not slug or wid is None or str(c.get("status", "")).lower() != "running": continue signals = [s for s in producer_state.signals_for(catalog.get_service(slug)) if producer_state.is_windowed(s)] if not signals: continue instance = int(c.get("instance") or 1) key = (slug, wid, instance) found: list[str] = [] try: if wid not in workers: workers[wid] = await database.get_worker(wid) for signal in signals: params: dict[str, Any] = { "lines": 1000, "match": producer_state.window_filter(signal), "since_hours": float(signal["window_hours"]), } if instance > 1: params["instance"] = instance # Hours of a chatty node's logs: seconds on a fast box, more on a Pi. resp = await _proxy_to_worker(wid, "GET", f"/api/containers/{slug}/logs", params=params, timeout=120) # A worker older than the filter returns the plain tail, which is # not the window: no verdict from it. if not resp.get("filtered"): continue if producer_state.windowed_hit(resp.get("logs", ""), signal): found.append(await _render_signal_means(signal, slug, workers[wid])) except Exception as exc: logger.warning("Windowed signal check failed for %s on worker %s: %s", slug, wid, exc) if key in _windowed_hints: hints[key] = _windowed_hints[key] continue if found: hints[key] = found _windowed_hints = hints async def _raise_worker_offline_alert(w: dict[str, Any]) -> None: """Record + push + bell an offline worker, keyed by client_id. client_id is the identity, per this codebase's own rule: workers.name is the container hostname, cosmetic, mutable across recreates, and shareable by two hosts — keying alerts on it would let one worker suppress or clear another's. The display name travels in the message and title only. """ cid = w.get("client_id") or w["name"] msg = ( f"Worker '{w['name']}': no heartbeat for over {_stale_after_seconds(w) // 60} minutes. Its " "containers keep running and earning, but CashPilot cannot see or manage them until it reconnects." ) if alert_id := await database.record_alert("worker", cid, msg): # The push floor, checked only on a FRESH row (a deduped record never # gets here): a new episode inside the floor keeps its row and bell # entry — the dashboard tells the truth — but does not buzz the phone # again. Deliberately never reset on recovery; resetting there is # exactly the re-arm loop this exists to end. now = time.monotonic() last_push = _worker_last_offline_push.get(cid) if last_push is None or now - last_push >= _WORKER_OFFLINE_PUSH_COOLDOWN_S: _worker_last_offline_push[cid] = now _spawn( _push_alert( "worker", cid, f"CashPilot: worker '{w['name']}' went offline", msg, alert_id, ) ) else: logger.info( "Offline push for worker '%s' suppressed by the per-worker floor (%.0f min since last)", w["name"], (now - last_push) / 60, ) # Into the bell NOW (the sweep runs every 2 min); the hourly collection # rebuild re-derives it from the workers table. if not any(a.get("kind") == "worker" and a.get("client_id") == cid for a in _collector_alerts): _collector_alerts.append({"kind": "worker", "platform": w["name"], "client_id": cid, "error": msg}) FLEET_API_KEY = fleet_key.resolve_fleet_key() HOSTNAME_PREFIX = os.getenv("CASHPILOT_HOSTNAME_PREFIX", "cashpilot") COLLECT_INTERVAL_MIN = int(os.getenv("CASHPILOT_COLLECT_INTERVAL", "60")) STALE_WORKER_SECONDS = 180 # Mark worker offline after 3 missed heartbeats # A phone reports from Android's WorkManager, whose shortest period is 15 # minutes, so 180 s would show every phone offline between two reports. # Three missed periods, the same rule as the 60 s Docker workers. STALE_ANDROID_WORKER_SECONDS = 45 * 60 def _stale_after_seconds(w: dict[str, Any]) -> int: """How long this worker may stay silent before it is marked offline.""" info = w.get("system_info") or {} if not isinstance(info, dict): # raw DB rows carry it as JSON text info = _safe_json(info, {}) if placement.is_android(info): return STALE_ANDROID_WORKER_SECONDS return STALE_WORKER_SECONDS async def _warm_session_epochs() -> None: """Restore the in-memory per-user session-epoch cache from durable state. Two sources are merged, taking the later timestamp per user: - ``users.password_changed_at`` — a password change invalidates older sessions. - ``session_revocations.revoked_before`` — a delete/demote invalidates older sessions and survives the users row being deleted. This runs at startup so those invalidations survive a UI restart. Without the revocation half, a delete/demote (which only bumps the in-memory epoch) would be forgotten on the next restart and the account's still-valid cookie would be honored again with its old role — the bug this fixes. """ epochs: dict[int, float] = {} for _u in await database.list_users_with_pwd_epoch(): changed = _u.get("password_changed_at") or 0.0 if changed: epochs[_u["id"]] = changed for _r in await database.list_session_revocations(): uid = _r["user_id"] epochs[uid] = max(epochs.get(uid, 0.0), _r["revoked_before"] or 0.0) for uid, ts in epochs.items(): auth.set_user_pwd_epoch(uid, ts) @asynccontextmanager async def lifespan(app: FastAPI): # Startup # Before anything touches credentials: refuse to run when the encryption key # cannot survive a restart. Continuing would encrypt every credential entered # during this run under a key that dies with the process. database.verify_encryption_key_persisted() await database.init_db() await database.connect_shared() # Warm the per-user session-epoch cache (password changes + delete/demote # revocations) so invalidated sessions are rejected without a DB hit in the # request path — and, crucially, so those invalidations survive a restart. await _warm_session_epochs() # Restore the collector alerts persisted by earlier runs, so a restart doesn't # silently clear the notification bell while the underlying collector is still # broken (previously these lived only in memory). await _warm_collector_alerts() await _refresh_activity_bell() # Credentials saved before saving them began creating a tracking row are # otherwise inert forever, while Settings shows them as "Configured". await _track_fully_configured_services() # First-run setup token: while no users exist, require a one-time token # (printed below) for /register so a proxy-exposed instance cannot be seized # by the first public visitor. Persisted in config so it survives restarts; # cleared once the owner account is created. if not await database.has_any_users(): _tok = await database.get_config("_setup_token") if not _tok: _tok = setup_token.generate() await database.set_config("_setup_token", _tok) setup_token.set_active(_tok) logger.warning( "FIRST-RUN SETUP: no account exists yet. Open /register and enter this " "one-time setup token to create the owner account: %s (shown only here; " "not embedded in any URL so it stays out of proxy logs and browser history)", _tok, ) if not notify.is_enabled(): # The push path exists so problems reach someone when nobody is # watching the dashboard — and a default install has never armed it. # Say so once, loudly, in the same place the setup token announces # itself; anything quieter and this line is its own counterexample. logger.warning( "No notification channel configured — alerts will only appear in the UI bell, " "which somebody has to open. Set CASHPILOT_NTFY_URL, CASHPILOT_WEBHOOK_URL or " "the CASHPILOT_TELEGRAM_* variables to get collector failures, offline workers " "and payout prompts pushed out-of-band." ) catalog.load_services() catalog.register_sighup() def _on_job_event(event): logger.error("Scheduler job %s failed or missed", event.job_id, exc_info=getattr(event, "exception", None)) scheduler.add_listener(_on_job_event, EVENT_JOB_ERROR | EVENT_JOB_MISSED) scheduler.add_job( _run_collection, "interval", minutes=COLLECT_INTERVAL_MIN, id="collect", max_instances=1, coalesce=True, misfire_grace_time=300, ) scheduler.add_job( _retry_device_links, "interval", minutes=3, id="retry_device_links", max_instances=1, coalesce=True, misfire_grace_time=300, ) scheduler.add_job( _run_health_check, "interval", minutes=5, id="health_check", max_instances=1, coalesce=True, misfire_grace_time=300, ) scheduler.add_job( _check_stale_workers, "interval", minutes=2, id="stale_workers", max_instances=1, coalesce=True, misfire_grace_time=300, ) scheduler.add_job( _check_windowed_signals, "interval", minutes=30, id="windowed_signals", max_instances=1, coalesce=True, misfire_grace_time=300, ) scheduler.add_job( _run_data_retention, "interval", hours=24, id="data_retention", max_instances=1, coalesce=True, misfire_grace_time=300, ) # Activity (docs/activity.md): the raw-address purge, the spike check and # the network table refresh. Hourly, so no address outlives a day by more # than an hour even when no heartbeat arrives to purge on ingest. scheduler.add_job( _run_activity_hourly, "interval", hours=1, id="activity_hourly", max_instances=1, coalesce=True, misfire_grace_time=300, ) scheduler.add_job( _run_vacuum, "interval", weeks=1, id="db_vacuum", max_instances=1, coalesce=True, misfire_grace_time=300, ) # Frequent tick, daily fetch: refresh() asks GitHub once a day after a check # that worked and an hour after one that failed, so a rate-limited startup # check no longer leaves the release unknown for a whole day. A tick that # is not due fetches nothing. Every failure mode inside refresh() lands on # UNKNOWN, so a firewalled install produces no noise at all. scheduler.add_job( update_check.refresh, "interval", seconds=update_check.SCHEDULER_TICK_SECONDS, id="update_check", max_instances=1, coalesce=True, # 300 like every other job here rather than a carve-out. A missed daily # check simply happens tomorrow; that is not worth an exception to a # convention a test enforces uniformly. misfire_grace_time=300, ) # The Android client's own release track, on the same cadence and with # the same failure contract. Separate job rather than folded into the one # above so a GitHub hiccup fetching one cannot suppress the other. scheduler.add_job( update_check.refresh_android, "interval", seconds=update_check.SCHEDULER_TICK_SECONDS, id="update_check_android", max_instances=1, coalesce=True, misfire_grace_time=300, ) # Proxy traffic caps that reset on a day of the month (app/proxies.py). scheduler.add_job( _run_proxy_cap_resets, "interval", hours=1, id="proxy_cap_resets", max_instances=1, coalesce=True, misfire_grace_time=300, ) scheduler.add_job( exchange_rates.refresh, "interval", minutes=15, id="exchange_rates", max_instances=1, coalesce=True, misfire_grace_time=300, ) scheduler.start() await exchange_rates.refresh() # Kick both release checks NOW rather than waiting out the first 24-hour # interval. APScheduler's interval trigger does not fire on start, and this # container is recreated on every deploy -- so on an install that ships more # often than daily the check would never run at all, and the update banner # would sit permanently at known=false. Spawned, not awaited: a slow or # unreachable GitHub must not delay startup, and every failure inside these # already lands on UNKNOWN. _spawn(update_check.refresh()) _spawn(update_check.refresh_android()) # Retention too, for the same reason: its 24-hour interval counts from # process start, so an install that restarts more than once a day (every # release recreates this container) never purged anything at all. _spawn(_run_data_retention()) # And VACUUM, which has the same trap on a weekly interval; it checks its # own last run, so this rewrites the file at most once a week. _spawn(_run_vacuum()) # The Activity purge too: its hourly timer restarts with the process, so a # UI restarted more often than hourly would keep addresses past a day. _spawn(_run_activity_purge()) # And the proxy cap reset: its hourly timer restarts with the process too, so # a UI restarted more often than hourly would leave a capped proxy's copies # stopped past its reset day. _spawn(_run_proxy_cap_resets()) _spawn(_run_collection()) _spawn(_check_windowed_signals()) logger.info("CashPilot UI started (container ops via workers)") yield # Shutdown scheduler.shutdown(wait=False) await database.close_shared() from app.collectors import close_all_collectors await close_all_collectors() logger.info("CashPilot stopped") app = FastAPI( title="CashPilot", version=version.current(), lifespan=lifespan, # Off by default. FastAPI serves these unauthenticated, and this app's # schema is a map of its own admin surface — every route, parameter and # body shape, including the worker and payout endpoints. Nothing here # needs them in a self-hosted deployment. docs_url=None, redoc_url=None, openapi_url=None, ) metrics.setup(app) # Whether a trusted reverse proxy sits in front (opt-in). Only then do we believe an # X-Forwarded-Proto header for deciding HTTPS (matches app.deps._TRUST_PROXY semantics; # read here directly to avoid importing deps into the middleware). _HSTS_TRUST_PROXY = os.getenv("CASHPILOT_TRUSTED_PROXY", "").strip().lower() in ("1", "true", "yes", "on") # Security headers middleware class _SecurityHeadersMiddleware(BaseHTTPMiddleware): async def dispatch(self, request, call_next): # A fresh nonce per response. Templates read it via request.state so the # value in the header and the value in the markup can never disagree — # if they did, every inline script would be blocked and the UI would # simply stop working. nonce = secrets.token_urlsafe(16) request.state.csp_nonce = nonce response = await call_next(request) response.headers["X-Content-Type-Options"] = "nosniff" response.headers["X-Frame-Options"] = "DENY" response.headers["Referrer-Policy"] = "strict-origin-when-cross-origin" response.headers["Permissions-Policy"] = "camera=(), microphone=(), geolocation=()" # script-src no longer allows 'unsafe-inline' (bead guw). Inline event # handlers are gone — every control goes through one delegated listener — # and the few remaining inline