From 938028a7a75e6c647560c1df6696bef5b99836af Mon Sep 17 00:00:00 2001 From: Louis King Date: Sun, 19 Jul 2026 19:56:24 +0100 Subject: [PATCH] feat: precompute route health in background sweep Persist route health derivations so the API layer no longer recomputes them on every request. Two new tables (route_result_history, route_recent_matches) plus a quality_avg column on route_results back the dashboard strip and per-route history endpoints. Collector: - run_evaluation (60s) writes snapshot + quality_avg + recent_matches - run_history_backfill (hourly) recomputes completed-day buckets - subscriber wires a dedicated backfill scheduler thread API: - dashboard routes-overview bulk-loads via single history SELECT - routes detail/detail history read from precomputed tables with live-compute fallback - POST/PUT on routes upserts recent_matches and quality_avg inline - dashboard cache TTL lowered from 1h to 5m (invalidation-aware) Config: route_history_backfill_interval_seconds=3600, redis_cache_ttl_dashboard default 3600 -> 300 --- .env.example | 21 +- src/meshcore_hub/api/routes/dashboard.py | 92 ++++- src/meshcore_hub/api/routes/routes.py | 221 ++++++++++-- src/meshcore_hub/collector/route_evaluator.py | 158 ++++++++- src/meshcore_hub/collector/routes.py | 331 +++++++++++++++++- src/meshcore_hub/collector/subscriber.py | 64 +++- src/meshcore_hub/common/config.py | 30 +- src/meshcore_hub/common/models/__init__.py | 8 + src/meshcore_hub/common/models/route.py | 15 + .../common/models/route_recent_match.py | 92 +++++ .../common/models/route_result.py | 18 +- .../common/models/route_result_history.py | 84 +++++ tests/test_api/test_cache.py | 32 +- tests/test_api/test_dashboard.py | 62 ++++ tests/test_api/test_routes.py | 225 ++++++++++-- tests/test_collector/test_route_evaluator.py | 239 ++++++++++++- tests/test_collector/test_routes.py | 24 +- 17 files changed, 1605 insertions(+), 111 deletions(-) create mode 100644 src/meshcore_hub/common/models/route_recent_match.py create mode 100644 src/meshcore_hub/common/models/route_result_history.py diff --git a/.env.example b/.env.example index 4a950cc..218f83c 100644 --- a/.env.example +++ b/.env.example @@ -412,11 +412,12 @@ PROMETHEUS_PORT=9090 # REDIS_CACHE_TTL=30 # Cache TTL for dashboard endpoints, /routes/{id} detail, and per-route -# health history (seconds). Trend/aggregation data tolerates much longer -# staleness than the default TTL. The Recent Adverts / Recent Channel -# Messages widgets live on a separate /dashboard/recent-activity endpoint -# that always uses REDIS_CACHE_TTL (above) so they stay fresh. -# REDIS_CACHE_TTL_DASHBOARD=3600 +# health history (seconds). Route health is precomputed by the background +# evaluator so misses are cheap; this can stay short to surface the latest +# sweep quickly. The Recent Adverts / Recent Channel Messages widgets live +# on a separate /dashboard/recent-activity endpoint that always uses +# REDIS_CACHE_TTL (above) so they stay fresh. +# REDIS_CACHE_TTL_DASHBOARD=300 # Emit HTTP Cache-Control on /api/v1/* responses + ETag/If-None-Match on # cached endpoints. The policy is `private, no-cache` on GETs (forces @@ -616,8 +617,16 @@ SYSTEM_MAINTENANCE=false # FEATURE_SPAM_DETECTION=true # Routes page (route health monitoring) is ON by default. # FEATURE_ROUTES=true -# Route evaluator interval in seconds (0 disables, default 60). +# Route evaluator interval in seconds (0 disables, default 60). Each tick +# writes the current snapshot to route_results, upserts today's per-day +# bucket into route_result_history, refreshes the rolling quality_avg +# column, and rewrites recent_matches_json. # ROUTE_EVALUATOR_INTERVAL_SECONDS=60 +# Slower backfill sweep (seconds, 0 disables, default 3600) that recomputes +# the full raw_packet_retention_days window into route_result_history so +# late-arriving packets and route-config changes propagate into historical +# buckets. +# ROUTE_HISTORY_BACKFILL_INTERVAL_SECONDS=3600 # ------------------- # Contact Information diff --git a/src/meshcore_hub/api/routes/dashboard.py b/src/meshcore_hub/api/routes/dashboard.py index 0dcf832..dad62ac 100644 --- a/src/meshcore_hub/api/routes/dashboard.py +++ b/src/meshcore_hub/api/routes/dashboard.py @@ -1,6 +1,7 @@ """Dashboard API routes.""" from datetime import date, datetime, timedelta, timezone +from typing import Sequence from fastapi import APIRouter, Request from sqlalchemy import and_, case, func, or_, select @@ -19,7 +20,6 @@ from meshcore_hub.api.observer_utils import ( fetch_observers_for_events, resolve_sender_names, ) -from meshcore_hub.collector.routes import evaluate_route_history from meshcore_hub.common.config import get_collector_settings from meshcore_hub.common.models import ( Advertisement, @@ -30,6 +30,7 @@ from meshcore_hub.common.models import ( Route, UserProfile, ) +from meshcore_hub.common.models.route_result_history import RouteResultHistory from meshcore_hub.common.schemas.messages import ( BreakdownBucket, ChannelMessage, @@ -731,6 +732,11 @@ def get_routes_overview( Role-visibility filtered the same way as ``GET /routes`` so members don't see admin-only routes. ``days`` is clamped to the configured raw-packet retention window so history queries can't scan purged data. + + History is read in a single bulk query against ``route_result_history`` + for every visible route — no per-route scans of ``packet_path_hops`` + on the hot path. Missing historical days pad with ``unknown`` / + ``no_coverage`` / ``0`` (matching the disabled-route semantics). """ days = min(days, 90) retention = get_collector_settings().effective_raw_packet_retention_days @@ -742,15 +748,19 @@ def get_routes_overview( routes = session.execute(select(Route).order_by(Route.from_label)).scalars().all() visible = [r for r in routes if VISIBILITY_LEVELS.get(r.visibility, 0) <= max_level] + # Bulk-load precomputed history for every visible route in one indexed + # query. ``read_route_history_from_db`` pads missing days and appends + # the today rolling-window segment sourced from each route's + # ``route_result`` (so the rightmost chart point matches the badge). + history_by_route = _bulk_read_history(session, visible, days) + # State buckets — ``disabled`` for switched-off routes, otherwise the # evaluator's last persisted state (falling back to ``no_coverage`` # when no result exists yet, e.g. a freshly created route). state_counts: dict[str, int] = {} entries: list[RouteOverviewEntry] = [] for route in visible: - history_tuples = evaluate_route_history( - session, route, days, include_today=True - ) + history_tuples = history_by_route.get(route.id, []) history = [ RouteDayQuality(date=d, quality=q, state=s, matched_count=c) for d, q, s, c in history_tuples @@ -803,3 +813,77 @@ def get_routes_overview( by_state.append(BreakdownBucket(label=label, count=count)) return RoutesOverview(days=days, by_state=by_state, routes=entries) + + +def _bulk_read_history( + session: DbSession, + routes: list[Route], + days: int, +) -> dict[str, list[tuple[date, str, str, int]]]: + """Bulk-load precomputed history for ``routes`` over the last ``days`` days. + + Returns ``{route_id: [(date, quality, state, matched_count), ...]}`` + keyed by route id. Each route's list has ``days + 1`` entries (the + historical UTC calendar days plus a synthetic today segment sourced + from ``route_result``, matching ``read_route_history_from_db``'s + ``include_today=True`` semantics). Disabled routes get all-unknown + padding without hitting the DB. + """ + if not routes: + return {} + + now = datetime.now(timezone.utc) + today_midnight = now.replace(hour=0, minute=0, second=0, microsecond=0) + today = today_midnight.date() + oldest = (today_midnight - timedelta(days=days)).date() + + enabled_ids = [r.id for r in routes if r.enabled] + history_rows: Sequence[tuple[str, date, str, str, int]] = [] + if enabled_ids: + history_rows = ( + session.execute( + select( + RouteResultHistory.route_id, + RouteResultHistory.date, + RouteResultHistory.quality, + RouteResultHistory.state, + RouteResultHistory.matched_count, + ) + .where(RouteResultHistory.route_id.in_(enabled_ids)) + .where(RouteResultHistory.date >= oldest) + .where(RouteResultHistory.date < today) + .order_by(RouteResultHistory.route_id, RouteResultHistory.date) + ) + .tuples() + .all() + ) + + by_route: dict[str, dict[date, tuple[str, str, int]]] = {} + for route_id, day, quality, state, matched_count in history_rows: + by_route.setdefault(route_id, {})[day] = (quality, state, matched_count) + + day_dates = [(oldest + timedelta(days=i)) for i in range(days)] + history_by_route: dict[str, list[tuple[date, str, str, int]]] = {} + for route in routes: + if not route.enabled: + results = [(d, "unknown", "no_coverage", 0) for d in day_dates] + results.append((today, "unknown", "no_coverage", 0)) + history_by_route[route.id] = results + continue + + rows_by_date = by_route.get(route.id, {}) + results = [ + ( + d, + *rows_by_date.get(d, ("unknown", "no_coverage", 0)), + ) + for d in day_dates + ] + rr = route.route_result + if rr is not None: + results.append((today, rr.quality, rr.state, rr.matched_count)) + else: + results.append((today, "unknown", "no_coverage", 0)) + history_by_route[route.id] = results + + return history_by_route diff --git a/src/meshcore_hub/api/routes/routes.py b/src/meshcore_hub/api/routes/routes.py index 45c5510..ae270ff 100644 --- a/src/meshcore_hub/api/routes/routes.py +++ b/src/meshcore_hub/api/routes/routes.py @@ -1,7 +1,7 @@ """Route health monitoring API routes.""" from datetime import datetime, timedelta, timezone -from typing import Optional +from typing import Any from fastapi import APIRouter, HTTPException, Request from sqlalchemy import select @@ -16,19 +16,23 @@ from meshcore_hub.api.channel_visibility import ( ) from meshcore_hub.api.dependencies import DbSession from meshcore_hub.collector.routes import ( - compute_average_quality, + compute_persisted_quality_avg, derive_expected_hash, evaluate_route, - evaluate_route_history, preview_route, + read_route_history_from_db, recent_matches, + upsert_route_recent_matches, upsert_route_result, ) from meshcore_hub.common.config import get_collector_settings from meshcore_hub.common.models.node import Node +from meshcore_hub.common.models.packet_path_hop import PacketPathHop +from meshcore_hub.common.models.raw_packet import RawPacket from meshcore_hub.common.models.route import Route from meshcore_hub.common.models.route_node import RouteNode from meshcore_hub.common.models.route_observer import RouteObserver +from meshcore_hub.common.models.route_recent_match import RouteRecentMatch from meshcore_hub.common.models.route_result import RouteResult from meshcore_hub.common.schemas.routes import ( ContributingObserver, @@ -49,6 +53,11 @@ from meshcore_hub.common.schemas.routes import ( router = APIRouter() +# Sentinel distinguishing "use the precomputed value" from "explicitly None" +# (the latter is what create_route passes to preserve the "brand-new route +# has no meaningful average yet" semantics on the POST response). +_UNSET = object() + def _routes_key_builder(request: Request) -> str: role = resolve_user_role(request) or "anonymous" @@ -86,7 +95,16 @@ def _result_to_summary(result: RouteResult | None) -> RouteResultSummary | None: ) -def _route_to_read(route: Route, *, quality_avg: Optional[str] = None) -> RouteRead: +def _route_to_read(route: Route, *, quality_avg: Any = _UNSET) -> RouteRead: + """Serialize a Route to its list-level read schema. + + ``quality_avg`` defaults to the precomputed value persisted on + ``route.route_result.quality_avg`` (written by the background + evaluator). Callers may pass an explicit value (e.g. ``None`` on + create responses) to override. + """ + if quality_avg is _UNSET: + quality_avg = route.route_result.quality_avg if route.route_result else None return RouteRead( id=route.id, from_label=route.from_label, @@ -109,20 +127,6 @@ def _route_to_read(route: Route, *, quality_avg: Optional[str] = None) -> RouteR ) -def _compute_quality_avg(session: DbSession, route: Route) -> Optional[str]: - """Rolling 7-day average quality for the route badge. - - Returns ``None`` for disabled routes. Empty history falls back to the - latest ``route_result.quality`` so brand-new routes don't flash a - misleading failing badge before their first evaluation cycle. - """ - if not route.enabled: - return None - history = evaluate_route_history(session, route, 7, include_today=True) - fallback = route.route_result.quality if route.route_result else None - return compute_average_quality(history, fallback=fallback) - - def _resolve_nodes_by_pubkey(session: DbSession, public_keys: list[str]) -> list[Node]: """Resolve node public keys to Node objects, preserving input order.""" lowered = [pk.strip().lower() for pk in public_keys] @@ -163,7 +167,7 @@ def _sync_observers( def _reevaluate_route(session: DbSession, route: Route) -> None: - """Synchronously evaluate *route* and persist the fresh ``RouteResult``. + """Synchronously evaluate *route* and persist every derived field. The background evaluator (collector.route_evaluator) writes ``RouteResult`` on a schedule (default 60s). Without this synchronous @@ -174,12 +178,33 @@ def _reevaluate_route(session: DbSession, route: Route) -> None: until the next evaluator cycle. Running the eval inline on every create/update keeps the post-mutation GET consistent with the new config at the cost of one bounded DB scan per write. + + Refreshes the current snapshot, the persisted top-3 recent matches + (so the detail page is fresh), and the rolling ``quality_avg`` (so + the list/detail badge updates immediately when the snapshot tier + changes). """ if not route.enabled: return - since = datetime.now(timezone.utc) - timedelta(hours=route.window_hours) + now = datetime.now(timezone.utc) + since = now - timedelta(hours=route.window_hours) state, quality, matched_count = evaluate_route(session, route, since) - upsert_route_result(session, route, state, quality, matched_count) + + matches = recent_matches(session, route, limit=3, now=now) + upsert_route_recent_matches(session, route.id, matches, limit=3) + + quality_avg = compute_persisted_quality_avg( + session, route, today_quality=quality, now=now + ) + + upsert_route_result( + session, + route, + state, + quality, + matched_count, + quality_avg=quality_avg, + ) session.commit() session.refresh(route) @@ -197,7 +222,7 @@ def list_routes( routes = session.execute(select(Route).order_by(Route.from_label)).scalars().all() filtered = [ - _route_to_read(r, quality_avg=_compute_quality_avg(session, r)) + _route_to_read(r) for r in routes if VISIBILITY_LEVELS.get(r.visibility, 0) <= max_level ] @@ -255,7 +280,7 @@ def create_route( session.refresh(route) _reevaluate_route(session, route) invalidate_routes(request) - return _route_to_read(route) + return _route_to_read(route, quality_avg=None) @router.get("/{route_id}", response_model=RouteDetail) @@ -282,7 +307,7 @@ def get_route( if VISIBILITY_LEVELS.get(route.visibility, 0) > max_level: raise HTTPException(status_code=404, detail="Route not found") - matches = recent_matches(session, route, limit=3) + matches = _load_recent_matches(session, route) contributing: dict[str, int] = {} for m in matches: @@ -311,7 +336,7 @@ def get_route( for oid, cnt in contributing.items() ] - read = _route_to_read(route, quality_avg=_compute_quality_avg(session, route)) + read = _route_to_read(route) return RouteDetail( **read.model_dump(), contributing_observers=contributors, @@ -319,6 +344,148 @@ def get_route( ) +def _load_recent_matches( + session: DbSession, + route: Route, +) -> list[dict[str, Any]]: + """Return the route's top-3 recent matches in the ``RecentMatchPath`` shape. + + Reads the normalized ``route_recent_matches`` table (populated by the + background evaluator on every 60s tick), JOINs through ``raw_packets`` + for the packet-level metadata, then fetches the matched hop slice from + ``packet_path_hops`` in a second indexed query and slices + ``[first_position .. last_position]`` per match in Python. Falls back + to a live ``recent_matches`` compute when the table is empty for the + route (fresh route, evaluator hasn't run yet, or older row from before + this table existed). + """ + matches = _read_recent_matches_from_table(session, route.id) + if matches: + return matches + if not route.enabled: + return [] + # Live fallback for fresh routes — produce the same dict shape with + # an empty hops list (the table will be populated on the next sweep). + live = recent_matches(session, route, limit=3) + return [ + { + "packet_hash": m.get("packet_hash"), + "event_hash": m.get("event_hash"), + "received_at": m.get("received_at"), + "observer_node_id": m.get("observer_node_id"), + "hops": _slice_hops_for_match( + session, m["raw_packet_id"], m["first_position"], m["last_position"] + ), + } + for m in live + ] + + +def _read_recent_matches_from_table( + session: DbSession, + route_id: str, + *, + limit: int = 3, +) -> list[dict[str, Any]]: + """Read recent matches from ``route_recent_matches`` + ``raw_packets``. + + Returns ``[]`` when the route has no persisted matches yet. + """ + rows = session.execute( + select( + RouteRecentMatch.raw_packet_id, + RouteRecentMatch.first_position, + RouteRecentMatch.last_position, + RawPacket.packet_hash, + RawPacket.event_hash, + RawPacket.received_at, + RawPacket.observer_node_id, + ) + .join(RawPacket, RawPacket.id == RouteRecentMatch.raw_packet_id) + .where(RouteRecentMatch.route_id == route_id) + .order_by(RawPacket.received_at.desc()) + .limit(limit) + ).all() + if not rows: + return [] + + # One IN-query for all the hops we need. + packet_ids = [r.raw_packet_id for r in rows] + hop_rows = session.execute( + select( + PacketPathHop.raw_packet_id, + PacketPathHop.position, + PacketPathHop.node_hash, + PacketPathHop.packet_hash, + PacketPathHop.event_hash, + PacketPathHop.received_at, + PacketPathHop.observer_node_id, + ) + .where(PacketPathHop.raw_packet_id.in_(packet_ids)) + .order_by(PacketPathHop.raw_packet_id, PacketPathHop.position) + ).all() + hops_by_packet: dict[str, list[dict[str, Any]]] = {} + for h in hop_rows: + hops_by_packet.setdefault(h.raw_packet_id, []).append( + { + "position": h.position, + "node_hash": h.node_hash, + "packet_hash": h.packet_hash, + "event_hash": h.event_hash, + "received_at": h.received_at, + "observer_node_id": h.observer_node_id, + } + ) + + out: list[dict[str, Any]] = [] + for r in rows: + all_hops = hops_by_packet.get(r.raw_packet_id, []) + sliced = all_hops[r.first_position : r.last_position + 1] + out.append( + { + "packet_hash": r.packet_hash, + "event_hash": r.event_hash, + "received_at": r.received_at, + "observer_node_id": r.observer_node_id, + "hops": sliced, + } + ) + return out + + +def _slice_hops_for_match( + session: DbSession, + raw_packet_id: str, + first_position: int, + last_position: int, +) -> list[dict[str, Any]]: + """Fetch and slice the hops for one match (live-fallback path only).""" + rows = session.execute( + select( + PacketPathHop.position, + PacketPathHop.node_hash, + PacketPathHop.packet_hash, + PacketPathHop.event_hash, + PacketPathHop.received_at, + PacketPathHop.observer_node_id, + ) + .where(PacketPathHop.raw_packet_id == raw_packet_id) + .order_by(PacketPathHop.position) + ).all() + all_hops = [ + { + "position": r.position, + "node_hash": r.node_hash, + "packet_hash": r.packet_hash, + "event_hash": r.event_hash, + "received_at": r.received_at, + "observer_node_id": r.observer_node_id, + } + for r in rows + ] + return all_hops[first_position : last_position + 1] + + @router.get("/{route_id}/history", response_model=RouteHistory) @cached( "routes/{id}/history", @@ -347,7 +514,7 @@ def get_route_history( retention = get_collector_settings().effective_raw_packet_retention_days days = min(days, retention) - history = evaluate_route_history(session, route, days, include_today=True) + history = read_route_history_from_db(session, route, days, include_today=True) return RouteHistory( route_id=route.id, @@ -427,7 +594,7 @@ def update_route( session.refresh(route) _reevaluate_route(session, route) invalidate_routes(request) - return _route_to_read(route, quality_avg=_compute_quality_avg(session, route)) + return _route_to_read(route) @router.delete("/{route_id}", status_code=204) diff --git a/src/meshcore_hub/collector/route_evaluator.py b/src/meshcore_hub/collector/route_evaluator.py index d128ff6..4bb6203 100644 --- a/src/meshcore_hub/collector/route_evaluator.py +++ b/src/meshcore_hub/collector/route_evaluator.py @@ -1,27 +1,60 @@ """Route health evaluator — background evaluation driver. -Wraps :mod:`meshcore_hub.collector.routes` to evaluate every enabled route -and upsert the results into ``route_results``. Called on a scheduler thread -inside the collector subscriber (mirroring the spam re-scoring sweep). +Two-cadence sweep: + +* :func:`run_evaluation` runs on the short (default 60s) tick. For every + enabled route it computes the current rolling-window snapshot via + :func:`evaluate_route`, captures the top-3 recent matches, refreshes + the rolling 7-day ``quality_avg``, and upserts everything into the + single ``route_results`` row. ``route_result_history`` (one row per + completed UTC day) is left untouched on this tick — today's data lives + only in the rolling snapshot until the day rolls over and the hourly + sweep captures it. + +* :func:`run_history_backfill` runs on the longer (default 1h) tick. It + recomputes the full retention window of ``route_result_history`` rows + so late-arriving packets and route-config tweaks propagate backward + into completed historical buckets, then refreshes ``quality_avg``. + +The split keeps the hot 60s tick cheap (one bounded scan per route) and +pushes the more expensive history sweep onto a quiet background thread +where it can amortize a multi-day scan across the whole fleet. """ import logging from datetime import datetime, timezone +from typing import TYPE_CHECKING -from meshcore_hub.collector.routes import upsert_route_result +from sqlalchemy import select + +from meshcore_hub.collector.routes import ( + compute_persisted_quality_avg, + evaluate_route, + evaluate_route_history, + recent_matches, + upsert_route_history_row, + upsert_route_recent_matches, + upsert_route_result, +) +from meshcore_hub.common.config import get_collector_settings from meshcore_hub.common.database import DatabaseManager from meshcore_hub.common.models.route import Route -from sqlalchemy import select + +if TYPE_CHECKING: + from sqlalchemy.orm import Session logger = logging.getLogger(__name__) def run_evaluation(db: DatabaseManager, now: datetime | None = None) -> int: - """Evaluate all enabled routes and upsert results. + """Evaluate every enabled route and upsert its result row. - Returns the number of routes evaluated. + Returns the number of routes evaluated. On the short tick this only + refreshes ``route_results`` (current snapshot + ``quality_avg`` + + ``recent_matches_json``); completed historical days are written by + :func:`run_history_backfill`. """ - now = now or datetime.now(timezone.utc) + current = now or datetime.now(timezone.utc) with db.session_scope() as session: routes = ( session.execute(select(Route).where(Route.enabled.is_(True))) @@ -32,15 +65,7 @@ def run_evaluation(db: DatabaseManager, now: datetime | None = None) -> int: count = 0 for route in routes: try: - from datetime import timedelta - - route_since = now - timedelta(hours=route.window_hours) - from meshcore_hub.collector.routes import evaluate_route - - state, quality, matched_count = evaluate_route( - session, route, route_since - ) - upsert_route_result(session, route, state, quality, matched_count) + _evaluate_one(session, route, current) count += 1 except Exception: logger.exception( @@ -50,3 +75,102 @@ def run_evaluation(db: DatabaseManager, now: datetime | None = None) -> int: ) return count + + +def _evaluate_one(session: "Session", route: Route, now: datetime) -> None: + """Compute and persist the rolling snapshot + average + matches for a route.""" + from datetime import timedelta + + route_since = now - timedelta(hours=route.window_hours) + state, quality, matched_count = evaluate_route(session, route, route_since) + + matches = recent_matches(session, route, limit=3, now=now) + upsert_route_recent_matches(session, route.id, matches, limit=3) + + quality_avg = compute_persisted_quality_avg( + session, route, today_quality=quality, now=now + ) + + upsert_route_result( + session, + route, + state, + quality, + matched_count, + quality_avg=quality_avg, + ) + + +def run_history_backfill( + db: DatabaseManager, + days: int | None = None, + now: datetime | None = None, +) -> int: + """Recompute the retention window of ``route_result_history`` for every route. + + Persists one row per completed UTC day (strictly before today) for + each enabled route, then refreshes ``route_results.quality_avg`` + from the newly-assembled history. ``days`` defaults to the + configured raw-packet retention window so backfills never scan + purged data. Returns the number of routes backfilled. + """ + current = now or datetime.now(timezone.utc) + settings = get_collector_settings() + window_days = ( + days if days is not None else settings.effective_raw_packet_retention_days + ) + + if window_days <= 0: + return 0 + + with db.session_scope() as session: + routes = ( + session.execute(select(Route).where(Route.enabled.is_(True))) + .scalars() + .all() + ) + + count = 0 + for route in routes: + try: + _backfill_one(session, route, window_days, current) + count += 1 + except Exception: + logger.exception( + "Error backfilling route '%s -> %s'", + route.from_label, + route.to_label, + ) + + return count + + +def _backfill_one(session: "Session", route: Route, days: int, now: datetime) -> None: + """Recompute persisted history for a single route + refresh ``quality_avg``.""" + today = now.date() + + # evaluate_route_history(include_today=False) yields N historical + # calendar-day buckets ending yesterday — exactly the rows we persist. + history = evaluate_route_history(session, route, days, include_today=False, now=now) + + for day, quality, state, matched_count in history: + if day >= today: + continue + upsert_route_history_row( + session, + route.id, + day, + quality, + state, + matched_count, + evaluated_at=now, + ) + + # Refresh quality_avg from the just-updated history + current snapshot. + today_quality = route.route_result.quality if route.route_result else None + if today_quality is not None: + quality_avg = compute_persisted_quality_avg( + session, route, today_quality=today_quality, now=now + ) + if route.route_result is not None and quality_avg is not None: + route.route_result.quality_avg = quality_avg diff --git a/src/meshcore_hub/collector/routes.py b/src/meshcore_hub/collector/routes.py index ce79624..cd286b2 100644 --- a/src/meshcore_hub/collector/routes.py +++ b/src/meshcore_hub/collector/routes.py @@ -7,7 +7,7 @@ per reception. Scales with (candidates) only, not (candidates × depth). import logging from datetime import date, datetime, timedelta, timezone -from typing import Any, Optional +from typing import Any, Iterable, Optional from uuid import uuid4 from sqlalchemy import and_, func, or_, select @@ -16,11 +16,13 @@ from sqlalchemy.orm import Session from meshcore_hub.common.models.node import Node from meshcore_hub.common.models.packet_path_hop import PacketPathHop from meshcore_hub.common.models.route import Route +from meshcore_hub.common.models.route_recent_match import RouteRecentMatch from meshcore_hub.common.models.route_result import ( RouteQuality, RouteResult, RouteState, ) +from meshcore_hub.common.models.route_result_history import RouteResultHistory logger = logging.getLogger(__name__) @@ -166,14 +168,35 @@ def _matched_subpath( when no match is found. The returned slice is in packet-traversal order (never reversed), so a reverse-direction packet shows as To -> ... -> From. """ + subpath, _first, _last = _matched_subpath_with_indices( + hops, expected, max_hop_span, reversible + ) + return subpath + + +def _matched_subpath_with_indices( + hops: list[dict[str, Any]], + expected: list[str], + max_hop_span: Optional[int] = None, + reversible: bool = True, +) -> tuple[Optional[list[dict[str, Any]]], Optional[int], Optional[int]]: + """Variant of :func:`_matched_subpath` that also returns match indices. + + Returns ``(subpath, first_index, last_index)`` where ``first_index`` and + ``last_index`` are the inclusive positions into *hops* of the matched + slice. On no match, returns ``(None, None, None)``. The indices power + the persisted ``first_position`` / ``last_position`` columns in + ``route_recent_matches`` so the detail page can slice the live hop + list without re-running the matcher. + """ idx = _subsequence_indices(hops, expected, max_hop_span) if idx is not None: - return hops[idx[0] : idx[1] + 1] + return hops[idx[0] : idx[1] + 1], idx[0], idx[1] if reversible and len(expected) > 1: idx = _subsequence_indices(hops, list(reversed(expected)), max_hop_span) if idx is not None: - return hops[idx[0] : idx[1] + 1] - return None + return hops[idx[0] : idx[1] + 1], idx[0], idx[1] + return None, None, None def _match_hops( @@ -813,8 +836,20 @@ def upsert_route_result( state: str, quality: str, matched_count: int, + *, + quality_avg: Optional[str] = None, ) -> RouteResult: - """Upsert a route evaluation result (ORM check-then-update/insert).""" + """Upsert a route evaluation result (ORM check-then-update/insert). + + ``quality_avg`` is only written when explicitly provided (non-``None``). + Callers that don't compute it leave the previous persisted value + intact, which keeps the background snapshot path cheap while letting + the API layer's synchronous re-evaluation on mutations refresh every + field at once. + + The top-N recent matches are persisted separately in + ``route_recent_matches`` via :func:`upsert_route_recent_matches`. + """ now = datetime.now(timezone.utc) eff_clear = effective_clear_threshold(route) @@ -829,6 +864,8 @@ def upsert_route_result( existing.threshold = route.packet_count_threshold existing.effective_clear = eff_clear existing.evaluated_at = now + if quality_avg is not None: + existing.quality_avg = quality_avg return existing result = RouteResult( @@ -840,11 +877,210 @@ def upsert_route_result( threshold=route.packet_count_threshold, effective_clear=eff_clear, evaluated_at=now, + quality_avg=quality_avg, ) session.add(result) return result +def upsert_route_history_row( + session: Session, + route_id: str, + day: date, + quality: str, + state: str, + matched_count: int, + *, + evaluated_at: Optional[datetime] = None, +) -> RouteResultHistory: + """Upsert one calendar-day bucket into ``route_result_history``. + + Idempotent via the ``UNIQUE (route_id, date)`` constraint — a + re-evaluation of the same day overwrites the prior row in place. + """ + now = evaluated_at or datetime.now(timezone.utc) + + existing = session.execute( + select(RouteResultHistory) + .where(RouteResultHistory.route_id == route_id) + .where(RouteResultHistory.date == day) + ).scalar_one_or_none() + + if existing: + existing.quality = quality + existing.state = state + existing.matched_count = matched_count + existing.evaluated_at = now + return existing + + row = RouteResultHistory( + id=str(uuid4()), + route_id=route_id, + date=day, + quality=quality, + state=state, + matched_count=matched_count, + evaluated_at=now, + ) + session.add(row) + return row + + +def compute_persisted_quality_avg( + session: Session, + route: Route, + *, + today_quality: str, + now: Optional[datetime] = None, + days: int = 7, +) -> Optional[str]: + """Rolling ``days``-day average tier from persisted history + today's snapshot. + + Reads the last ``days`` history rows for completed UTC calendar days + (strictly before today), appends today's rolling-window ``quality`` + (the same value the badge uses, sourced from the latest + ``evaluate_route`` call), and feeds both to ``compute_average_quality``. + Returns ``None`` when no history exists and no today quality is + available, so brand-new routes don't flash a misleading failing badge + before the first evaluator cycle. + """ + current = now or datetime.now(timezone.utc) + today = current.date() + + if not today_quality: + return None + + rows = session.execute( + select( + RouteResultHistory.date, + RouteResultHistory.quality, + RouteResultHistory.state, + RouteResultHistory.matched_count, + ) + .where(RouteResultHistory.route_id == route.id) + .where(RouteResultHistory.date < today) + .order_by(RouteResultHistory.date.desc()) + .limit(days) + ).all() + + history_tuples: list[tuple[date, str, str, int]] = [ + (row.date, row.quality, row.state, row.matched_count) for row in reversed(rows) + ] + history_tuples.append((today, today_quality, "", 0)) + + if not history_tuples: + return None + if not rows: + # Brand-new route: no historical buckets yet. Return None so the + # frontend's ``quality_avg || route_result?.quality`` fallback + # kicks in (matches the historical "fresh route" semantics). + return None + return compute_average_quality(history_tuples, fallback=today_quality) + + +def read_route_history_from_db( + session: Session, + route: Route, + days: int, + *, + include_today: bool = True, + now: Optional[datetime] = None, +) -> list[tuple[date, str, str, int]]: + """Read precomputed history for a route from ``route_result_history``. + + Replaces the on-demand ``evaluate_route_history`` call on the API hot + path with a single indexed ``SELECT``. Pads missing days with + ``unknown`` / ``no_coverage`` / ``0`` (matching the disabled-route + semantics the chart already relies on). When *include_today* is True, + appends a synthetic today segment sourced from ``route.route_result`` + (the rolling-window snapshot the badge uses) so the rightmost chart + point stays consistent with the card badge. + + For a disabled route, every entry returns ``unknown`` / + ``no_coverage`` / ``0`` without hitting the DB. + """ + current = now or datetime.now(timezone.utc) + today_midnight = current.replace(hour=0, minute=0, second=0, microsecond=0) + today = current.date() + oldest = today_midnight - timedelta(days=days) + day_dates = [(oldest + timedelta(days=i)).date() for i in range(days)] + + if not route.enabled: + results: list[tuple[date, str, str, int]] = [ + (d, RouteQuality.UNKNOWN.value, RouteState.NO_COVERAGE.value, 0) + for d in day_dates + ] + if include_today: + results.append( + (today, RouteQuality.UNKNOWN.value, RouteState.NO_COVERAGE.value, 0) + ) + return results + + rows = session.execute( + select( + RouteResultHistory.date, + RouteResultHistory.quality, + RouteResultHistory.state, + RouteResultHistory.matched_count, + ) + .where(RouteResultHistory.route_id == route.id) + .where(RouteResultHistory.date >= oldest.date()) + .where(RouteResultHistory.date < today) + .order_by(RouteResultHistory.date) + ).all() + rows_by_date: dict[date, tuple[str, str, int]] = { + row.date: (row.quality, row.state, row.matched_count) for row in rows + } + + results = [ + ( + d, + *rows_by_date.get( + d, (RouteQuality.UNKNOWN.value, RouteState.NO_COVERAGE.value, 0) + ), + ) + for d in day_dates + ] + + if include_today: + rr = route.route_result + if rr is not None: + results.append((today, rr.quality, rr.state, rr.matched_count)) + else: + results.append( + (today, RouteQuality.UNKNOWN.value, RouteState.NO_COVERAGE.value, 0) + ) + + return results + + +def _legacy_recent_matches_payload( + matches: Iterable[dict[str, Any]], +) -> list[dict[str, Any]]: + """Adapter from the new ``recent_matches`` dict shape to the legacy + ``RecentMatchPath(**m)`` schema expected by the detail-page response. + + The new ``recent_matches`` carries ``raw_packet_id`` / ``first_position`` + / ``last_position`` (for the normalized table) instead of a pre-sliced + ``hops`` list. The detail-page read path JOINs through + ``route_recent_matches`` to load hops live, so this shim only exists + for the rare path where the evaluator hasn't populated the table yet + and we fall back to a live compute. + """ + out: list[dict[str, Any]] = [] + for m in matches: + out.append( + { + "packet_hash": m.get("packet_hash"), + "event_hash": m.get("event_hash"), + "received_at": m.get("received_at"), + "observer_node_id": m.get("observer_node_id"), + "hops": [], # populated by the caller via packet_path_hops + } + ) + return out + + # --------------------------------------------------------------------------- # Card expand + preview # --------------------------------------------------------------------------- @@ -856,7 +1092,17 @@ def recent_matches( limit: int = 3, now: Optional[datetime] = None, ) -> list[dict[str, Any]]: - """Return the latest *limit* matching paths for a route. + """Return metadata for the latest *limit* matching receptions for a route. + + Each dict carries the keys the evaluator needs to persist a + ``RouteRecentMatch`` row and the API needs to render the detail-page + match card: ``raw_packet_id``, ``packet_hash``, ``event_hash``, + ``received_at``, ``observer_node_id``, ``first_position``, + ``last_position``. The matched hop slice is NOT included — callers + that need the hops (i.e. the API detail endpoint) JOIN through + ``raw_packet_id`` → ``packet_path_hops`` and slice on + ``[first_position .. last_position]`` so the data is sourced from + its canonical home instead of being denormalized at sweep time. Deduplicates by event identity (preferring ``event_hash``, falling back to wire ``packet_hash``) so the UI shows one row per underlying event @@ -881,25 +1127,29 @@ def recent_matches( # Keep the newest match per identity so the UI lists distinct underlying # events rather than every retransmission of the same event. matches_by_identity: dict[str, dict[str, Any]] = {} - for hops in paths.values(): - subpath = _matched_subpath(hops, expected, route.max_hop_span, reversible) - if not subpath: + for rp_id, hops in paths.items(): + subpath, first_idx, last_idx = _matched_subpath_with_indices( + hops, expected, route.max_hop_span, reversible + ) + if not subpath or first_idx is None or last_idx is None: continue identity = _match_identity(subpath) if identity is None: # Fall back to a synthetic unique key so unmatched-identity # receptions still surface (one row each). identity = f"__rawid_{id(subpath)}" - first = subpath[0] if subpath else {} + first = subpath[0] received_at = first.get("received_at") or datetime.min.replace( tzinfo=timezone.utc ) candidate = { + "raw_packet_id": rp_id, "packet_hash": first.get("packet_hash"), "event_hash": first.get("event_hash"), - "hops": subpath, "received_at": first.get("received_at"), "observer_node_id": first.get("observer_node_id"), + "first_position": first_idx, + "last_position": last_idx, } existing = matches_by_identity.get(identity) if existing is None or received_at > ( @@ -915,6 +1165,65 @@ def recent_matches( return matches[:limit] +def upsert_route_recent_matches( + session: Session, + route_id: str, + matches: Iterable[dict[str, Any]], + *, + limit: int = 3, +) -> list[RouteRecentMatch]: + """Replace the route's recent-match set with *matches* (capped at *limit*). + + ``matches`` is the output of :func:`recent_matches`. Rows whose + ``raw_packet_id`` is no longer in the new set are deleted; new and + changed rows are upserted in place. Returns the resulting ORM rows + (sorted newest-first by ``raw_packet_id`` for caller convenience; + order is not persisted — the detail page orders by + ``raw_packets.received_at`` at read time). + + Cap is enforced at write time as a safety net — :func:`recent_matches` + already LIMITs. + """ + new_matches = list(matches)[:limit] + new_packet_ids = {m["raw_packet_id"] for m in new_matches} + + existing_rows = ( + session.execute( + select(RouteRecentMatch).where(RouteRecentMatch.route_id == route_id) + ) + .scalars() + .all() + ) + existing_by_packet = {r.raw_packet_id: r for r in existing_rows} + + # Delete rows whose raw_packet_id is no longer in the new set + for row in existing_rows: + if row.raw_packet_id not in new_packet_ids: + session.delete(row) + + # Upsert new / changed + kept: list[RouteRecentMatch] = [] + for m in new_matches: + rpid = m["raw_packet_id"] + existing: Optional[RouteRecentMatch] = existing_by_packet.get(rpid) + first_pos = m["first_position"] + last_pos = m["last_position"] + if existing is None: + existing = RouteRecentMatch( + id=str(uuid4()), + route_id=route_id, + raw_packet_id=rpid, + first_position=first_pos, + last_position=last_pos, + ) + session.add(existing) + else: + existing.first_position = first_pos + existing.last_position = last_pos + kept.append(existing) + return kept + + def preview_route( session: Session, config: dict[str, Any], diff --git a/src/meshcore_hub/collector/subscriber.py b/src/meshcore_hub/collector/subscriber.py index 651e2dc..7560233 100644 --- a/src/meshcore_hub/collector/subscriber.py +++ b/src/meshcore_hub/collector/subscriber.py @@ -114,8 +114,10 @@ class Subscriber(LetsMeshNormalizer): self._channel_refresh_thread: Optional[threading.Thread] = None # Background spam re-scoring sweep self._spam_rescore_thread: Optional[threading.Thread] = None - # Background route health evaluator + # Background route health evaluator (short tick: rolling snapshot) self._route_evaluator_thread: Optional[threading.Thread] = None + # Background route history backfill (long tick: completed days) + self._route_history_backfill_thread: Optional[threading.Thread] = None # Load initial channel keys from database self._include_test_channel = self._load_channel_keys_from_db() self._letsmesh_decoder = LetsMeshPacketDecoder( @@ -690,6 +692,60 @@ class Subscriber(LetsMeshNormalizer): if self._route_evaluator_thread.is_alive(): logger.warning("Route evaluator thread did not stop cleanly") + def _start_route_history_backfill_scheduler(self) -> None: + """Start background thread that recomputes route history buckets. + + Disabled when the interval is 0. Slower than the rolling-snapshot + evaluator: writes one row per completed UTC day per route so the + per-route history endpoint and dashboard routes-overview can read + precomputed data instead of rescanning ``packet_path_hops``. + """ + from meshcore_hub.common.config import CollectorSettings + + interval = CollectorSettings().route_history_backfill_interval_seconds + if interval <= 0: + logger.info("Route history backfill disabled (interval=%ds)", interval) + return + + logger.info("Starting route history backfill (interval=%ds)", interval) + + def run_backfill_loop() -> None: + """Periodically recompute the retention window of history rows.""" + from meshcore_hub.collector.route_evaluator import run_history_backfill + + while self._running: + for _ in range(interval): + if not self._running: + break + time.sleep(1) + if self._running: + try: + updated = run_history_backfill(self.db) + if updated: + logger.info( + "Route history backfill refreshed %d routes", + updated, + ) + except Exception as e: + logger.error( + "Route history backfill error: %s", e, exc_info=True + ) + + self._route_history_backfill_thread = threading.Thread( + target=run_backfill_loop, daemon=True, name="route-history-backfill" + ) + self._route_history_backfill_thread.start() + + def _stop_route_history_backfill_scheduler(self) -> None: + """Stop the route history backfill thread.""" + if ( + self._route_history_backfill_thread + and self._route_history_backfill_thread.is_alive() + ): + self._route_history_backfill_thread.join(timeout=5.0) + if self._route_history_backfill_thread.is_alive(): + logger.warning("Route history backfill thread did not stop cleanly") + def start(self) -> None: """Start the subscriber.""" logger.info("Starting collector subscriber") @@ -763,6 +819,9 @@ class Subscriber(LetsMeshNormalizer): # Start route health evaluator (no-op when disabled) self._start_route_evaluator_scheduler() + # Start route history backfill (no-op when disabled) + self._start_route_history_backfill_scheduler() + # Start health reporter for Docker health checks self._health_reporter = HealthReporter( component="collector", @@ -807,6 +866,9 @@ class Subscriber(LetsMeshNormalizer): # Stop route evaluator self._stop_route_evaluator_scheduler() + # Stop route history backfill + self._stop_route_history_backfill_scheduler() + # Stop webhook processor self._stop_webhook_processor() diff --git a/src/meshcore_hub/common/config.py b/src/meshcore_hub/common/config.py index ef7e6d3..9674218 100644 --- a/src/meshcore_hub/common/config.py +++ b/src/meshcore_hub/common/config.py @@ -327,7 +327,23 @@ class CollectorSettings(CommonSettings): ) route_evaluator_interval_seconds: int = Field( default=60, - description="Route evaluator interval in seconds (0 disables, default 60)", + description=( + "Route evaluator interval in seconds. Each tick writes the current " + "snapshot to route_results, upserts today's per-day bucket into " + "route_result_history, refreshes the rolling quality_avg column, " + "and rewrites recent_matches_json. 0 disables the evaluator." + ), + ge=0, + ) + route_history_backfill_interval_seconds: int = Field( + default=3600, + description=( + "Cadence in seconds for the slower route-history backfill sweep " + "that recomputes the full raw_packet_retention_days window into " + "route_result_history so late-arriving packets and route-config " + "changes propagate into historical buckets. 0 disables the " + "backfill (today's bucket is still updated by the regular sweep)." + ), ge=0, ) @@ -418,13 +434,15 @@ class APISettings(CommonSettings): description="Default cache TTL in seconds", ) redis_cache_ttl_dashboard: int = Field( - default=3600, + default=300, description=( "Cache TTL in seconds for dashboard endpoints, /routes/{id} detail " - "and per-route health history (trend/aggregation data tolerates " - "longer staleness than the default TTL). The Recent Adverts / " - "Recent Channel Messages widgets live on a separate " - "/dashboard/recent-activity endpoint cached at redis_cache_ttl." + "and per-route health history. With route health now precomputed " + "by the background evaluator, misses are cheap and the TTL can be " + "kept short (5 min) so the UI reflects the latest sweep. The " + "Recent Adverts / Recent Channel Messages widgets live on a " + "separate /dashboard/recent-activity endpoint cached at " + "redis_cache_ttl." ), ) diff --git a/src/meshcore_hub/common/models/__init__.py b/src/meshcore_hub/common/models/__init__.py index 8f61021..c896db7 100644 --- a/src/meshcore_hub/common/models/__init__.py +++ b/src/meshcore_hub/common/models/__init__.py @@ -17,11 +17,16 @@ from meshcore_hub.common.models.packet_path_hop import PacketPathHop from meshcore_hub.common.models.route import Route, RouteVisibility from meshcore_hub.common.models.route_node import RouteNode from meshcore_hub.common.models.route_observer import RouteObserver +from meshcore_hub.common.models.route_recent_match import ( + ROUTE_RECENT_MATCHES_LIMIT, + RouteRecentMatch, +) from meshcore_hub.common.models.route_result import ( RouteResult, RouteQuality, RouteState, ) +from meshcore_hub.common.models.route_result_history import RouteResultHistory __all__ = [ "Base", @@ -45,7 +50,10 @@ __all__ = [ "RouteVisibility", "RouteNode", "RouteObserver", + "RouteRecentMatch", + "ROUTE_RECENT_MATCHES_LIMIT", "RouteResult", + "RouteResultHistory", "RouteQuality", "RouteState", ] diff --git a/src/meshcore_hub/common/models/route.py b/src/meshcore_hub/common/models/route.py index d316c89..6eeb884 100644 --- a/src/meshcore_hub/common/models/route.py +++ b/src/meshcore_hub/common/models/route.py @@ -11,7 +11,9 @@ from meshcore_hub.common.models.base import Base, TimestampMixin, UUIDMixin if TYPE_CHECKING: from meshcore_hub.common.models.route_node import RouteNode from meshcore_hub.common.models.route_observer import RouteObserver + from meshcore_hub.common.models.route_recent_match import RouteRecentMatch from meshcore_hub.common.models.route_result import RouteResult + from meshcore_hub.common.models.route_result_history import RouteResultHistory class RouteVisibility(str, Enum): @@ -115,6 +117,19 @@ class Route(Base, UUIDMixin, TimestampMixin): uselist=False, lazy="selectin", ) + route_result_history: Mapped[list["RouteResultHistory"]] = relationship( + "RouteResultHistory", + back_populates="route", + cascade="all, delete-orphan", + order_by="RouteResultHistory.date", + passive_deletes=True, + ) + route_recent_matches: Mapped[list["RouteRecentMatch"]] = relationship( + "RouteRecentMatch", + back_populates="route", + cascade="all, delete-orphan", + passive_deletes=True, + ) def __repr__(self) -> str: return f"" diff --git a/src/meshcore_hub/common/models/route_recent_match.py b/src/meshcore_hub/common/models/route_recent_match.py new file mode 100644 index 0000000..7e8faa7 --- /dev/null +++ b/src/meshcore_hub/common/models/route_recent_match.py @@ -0,0 +1,92 @@ +"""RouteRecentMatch model — normalized link from a route to its recent matches. + +One row per ``(route, raw_packet)`` pair that the background evaluator +identified as a recent match. Capped at ``ROUTE_RECENT_MATCHES_LIMIT`` +rows per route (default 3); the sweep replaces the set on every tick. + +The actual packet / path data stays in its canonical home +(``raw_packets`` / ``packet_path_hops``) — this table only stores the +link plus the ``[first_position, last_position]`` slice of the packet's +path that matched the route's expected sequence. The detail page JOINs +through ``raw_packet_id`` to render the full match card, so late +``event_hash`` backfills and raw-packet retention purges propagate +automatically (via the ``ON DELETE CASCADE`` FK) without manual resync. + +Cascade rules: + +* ``route_id`` FK cascades on route delete (matches disappear with the + route). +* ``raw_packet_id`` FK cascades on raw-packet delete (matches disappear + with the underlying packet — retention cleanup handles this). +""" + +from typing import TYPE_CHECKING + +from sqlalchemy import ForeignKey, Integer, UniqueConstraint +from sqlalchemy.orm import Mapped, mapped_column, relationship + +from meshcore_hub.common.models.base import Base, TimestampMixin, UUIDMixin + +if TYPE_CHECKING: + from meshcore_hub.common.models.raw_packet import RawPacket + from meshcore_hub.common.models.route import Route + + +# Default cap enforced by the evaluator sweep. Reads also LIMIT by this +# value as a safety net. +ROUTE_RECENT_MATCHES_LIMIT = 3 + + +class RouteRecentMatch(Base, UUIDMixin, TimestampMixin): + """A single recent ``route ↔ raw_packet`` match identified by the sweep. + + Attributes: + id: UUID primary key + route_id: FK to routes (cascades on delete) + raw_packet_id: FK to raw_packets (cascades on delete) + first_position: Index into the packet's path of the first matched hop + last_position: Index into the packet's path of the last matched hop + (inclusive). The matched subpath is ``hops[first_position .. + last_position]``. + """ + + __tablename__ = "route_recent_matches" + __table_args__ = ( + UniqueConstraint( + "route_id", "raw_packet_id", name="uq_route_recent_matches_route_packet" + ), + ) + + route_id: Mapped[str] = mapped_column( + ForeignKey("routes.id", ondelete="CASCADE"), + nullable=False, + index=True, + ) + raw_packet_id: Mapped[str] = mapped_column( + ForeignKey("raw_packets.id", ondelete="CASCADE"), + nullable=False, + ) + first_position: Mapped[int] = mapped_column( + Integer, + nullable=False, + ) + last_position: Mapped[int] = mapped_column( + Integer, + nullable=False, + ) + + route: Mapped["Route"] = relationship( + "Route", + back_populates="route_recent_matches", + ) + raw_packet: Mapped["RawPacket"] = relationship( + "RawPacket", + lazy="joined", + ) + + def __repr__(self) -> str: + return ( + f"" + ) diff --git a/src/meshcore_hub/common/models/route_result.py b/src/meshcore_hub/common/models/route_result.py index db458a6..1bdbe45 100644 --- a/src/meshcore_hub/common/models/route_result.py +++ b/src/meshcore_hub/common/models/route_result.py @@ -2,7 +2,7 @@ from datetime import datetime from enum import Enum -from typing import TYPE_CHECKING +from typing import TYPE_CHECKING, Optional from sqlalchemy import DateTime, ForeignKey, Integer, String from sqlalchemy.orm import Mapped, mapped_column, relationship @@ -37,6 +37,17 @@ class RouteResult(Base, UUIDMixin, TimestampMixin): and ``effective_clear`` are snapshotted at evaluation time so the display stays self-consistent if thresholds are later changed. + ``quality_avg`` holds the rolling 7-day average tier computed from the + last 7 ``RouteResultHistory`` rows plus today's snapshot. It backs the + route card badge and the dashboard strip summary so a flapping route + that's currently up still shows as marginal/failing if the week's mean + warrants it. + + The top-N recent matches for the detail page live in the separate + ``route_recent_matches`` table (normalized link to ``raw_packets``) + rather than on this row, so they stay consistent with raw-packet + retention and ``event_hash`` backfills without manual resync. + Attributes: id: UUID primary key route_id: FK to routes (unique, cascades on delete) @@ -46,6 +57,7 @@ class RouteResult(Base, UUIDMixin, TimestampMixin): threshold: Snapshot of route.packet_count_threshold at eval time effective_clear: Snapshot of effective_clear_threshold at eval time evaluated_at: When this evaluation ran + quality_avg: Rolling 7-day average quality tier (clear/marginal/failing) """ __tablename__ = "route_results" @@ -81,6 +93,10 @@ class RouteResult(Base, UUIDMixin, TimestampMixin): default=utc_now, nullable=False, ) + quality_avg: Mapped[Optional[str]] = mapped_column( + String(20), + nullable=True, + ) route: Mapped["Route"] = relationship( "Route", diff --git a/src/meshcore_hub/common/models/route_result_history.py b/src/meshcore_hub/common/models/route_result_history.py new file mode 100644 index 0000000..867fc88 --- /dev/null +++ b/src/meshcore_hub/common/models/route_result_history.py @@ -0,0 +1,84 @@ +"""RouteResultHistory model — per-day persisted health history for a Route. + +One row per ``(route, UTC calendar day)`` carrying that day's evaluated +``quality`` / ``state`` / ``matched_count``. Written by the background +route evaluator on every sweep (today's bucket) and on the slower +backfill sweep (the full retention window, to catch late-arriving +packets and reflect config changes). Read by the API layer (per-route +history endpoint and the dashboard routes-overview widget) so the hot +path never re-scans ``packet_path_hops``. + +The ``UNIQUE (route_id, date)`` constraint makes the upsert path +idempotent: a re-evaluation of the same day overwrites the prior row in +place. Cascade-delete on the parent ``routes`` row keeps cleanup +automatic when a route is removed. +""" + +from datetime import date, datetime +from typing import TYPE_CHECKING + +from sqlalchemy import Date, DateTime, ForeignKey, Integer, String, UniqueConstraint +from sqlalchemy.orm import Mapped, mapped_column, relationship + +from meshcore_hub.common.models.base import Base, TimestampMixin, UUIDMixin, utc_now + +if TYPE_CHECKING: + from meshcore_hub.common.models.route import Route + + +class RouteResultHistory(Base, UUIDMixin, TimestampMixin): + """One evaluated UTC calendar-day bucket for a route. + + Attributes: + id: UUID primary key + route_id: FK to routes (cascades on delete) + date: UTC calendar day this bucket covers + quality: Display axis (clear / marginal / failing / unknown) + state: Alerting axis (healthy / unhealthy / no_coverage) + matched_count: Distinct matching packet/event count for the day + evaluated_at: When this bucket was last (re)computed + """ + + __tablename__ = "route_result_history" + __table_args__ = ( + UniqueConstraint("route_id", "date", name="uq_route_result_history_route_date"), + ) + + route_id: Mapped[str] = mapped_column( + ForeignKey("routes.id", ondelete="CASCADE"), + nullable=False, + index=True, + ) + date: Mapped[date] = mapped_column( + Date, + nullable=False, + ) + quality: Mapped[str] = mapped_column( + String(20), + nullable=False, + ) + state: Mapped[str] = mapped_column( + String(20), + nullable=False, + ) + matched_count: Mapped[int] = mapped_column( + Integer, + nullable=False, + ) + evaluated_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), + default=utc_now, + nullable=False, + ) + + route: Mapped["Route"] = relationship( + "Route", + back_populates="route_result_history", + ) + + def __repr__(self) -> str: + return ( + f"" + ) diff --git a/tests/test_api/test_cache.py b/tests/test_api/test_cache.py index 9393675..f563d80 100644 --- a/tests/test_api/test_cache.py +++ b/tests/test_api/test_cache.py @@ -965,22 +965,38 @@ class TestCacheControlMiddleware: assert response.status_code == 200 assert response.headers["cache-control"] == "no-store" - def test_kill_switch_suppresses_cache_control(self, client_no_auth): + def test_kill_switch_suppresses_cache_control(self, client_no_auth, monkeypatch): """When api_cache_control_enabled is False, no Cache-Control is added.""" - client_no_auth.app.state.api_cache_control_enabled = False - if hasattr(client_no_auth.app.state, "redis_cache"): - del client_no_auth.app.state.redis_cache + monkeypatch.setattr( + client_no_auth.app.state, "api_cache_control_enabled", False + ) + # Remove any pre-existing cache backend for the duration of the test; + # ``monkeypatch.delattr`` with raising=False is a no-op when the + # attribute is absent. + monkeypatch.delattr(client_no_auth.app.state, "redis_cache", raising=False) response = client_no_auth.get("/api/v1/nodes") assert "cache-control" not in response.headers - def test_kill_switch_preserves_x_cache_header(self, client_no_auth): + def test_kill_switch_preserves_x_cache_header(self, client_no_auth, monkeypatch): """X-Cache is observability, not a client-caching directive, so the kill switch should not suppress it.""" mock_cache = MagicMock() mock_cache.get.return_value = None - client_no_auth.app.state.redis_cache = mock_cache - client_no_auth.app.state.redis_cache_ttl = 30 - client_no_auth.app.state.api_cache_control_enabled = False + # Force-set the cache + ttl + kill switch on app.state. Using + # ``setattr(..., raising=False)`` lets us create the attribute + # even when a prior test removed it, and ``monkeypatch`` restores + # the original (or removes what it added) on teardown — fixing + # the cross-test leak that previously left ``api_cache_control_enabled`` + # flipped off for every subsequent test in this module. + monkeypatch.setattr( + client_no_auth.app.state, "redis_cache", mock_cache, raising=False + ) + monkeypatch.setattr( + client_no_auth.app.state, "redis_cache_ttl", 30, raising=False + ) + monkeypatch.setattr( + client_no_auth.app.state, "api_cache_control_enabled", False + ) response = client_no_auth.get("/api/v1/nodes") assert response.headers.get("x-cache") == "MISS" assert "cache-control" not in response.headers diff --git a/tests/test_api/test_dashboard.py b/tests/test_api/test_dashboard.py index 77e3554..5db0fa3 100644 --- a/tests/test_api/test_dashboard.py +++ b/tests/test_api/test_dashboard.py @@ -1640,3 +1640,65 @@ class TestRoutesOverview: assert len(keys_seen) == 2 assert any("anonymous" in k for k in keys_seen) assert any("admin" in k for k in keys_seen) + + def test_history_served_from_precomputed_table( + self, client_no_auth, api_db_session + ): + """``routes-overview`` history reads from ``route_result_history``. + + Pre-precomputation, each request recomputed per-day buckets by + scanning ``packet_path_hops`` for every visible route — the + workload that motivated this refactor. Now the bulk-load path + issues one indexed SELECT and pads missing days with + ``unknown`` / ``no_coverage``. + """ + from meshcore_hub.common.models.route_result_history import ( + RouteResultHistory, + ) + + route = _make_route_with_nodes( + api_db_session, "Precomputed", "EP", ["a" * 64, "b" * 64] + ) + # Seed two completed days of clear history. + today = datetime.now(timezone.utc).date() + api_db_session.add_all( + [ + RouteResultHistory( + route_id=route.id, + date=today - timedelta(days=2), + quality="clear", + state="healthy", + matched_count=5, + ), + RouteResultHistory( + route_id=route.id, + date=today - timedelta(days=1), + quality="clear", + state="healthy", + matched_count=4, + ), + ] + ) + api_db_session.commit() + + data = client_no_auth.get("/api/v1/dashboard/routes-overview?days=3").json() + entry = next(r for r in data["routes"] if r["from_label"] == "Precomputed") + history = entry["history"] + + # days=3 ⇒ 3 historical buckets + 1 today segment = 4 entries. + assert len(history) == 4 + + # The two seeded days carry their persisted quality/matched_count. + seeded = {h["date"]: h for h in history} + two_days_ago = (today - timedelta(days=2)).isoformat() + one_day_ago = (today - timedelta(days=1)).isoformat() + assert seeded[two_days_ago]["quality"] == "clear" + assert seeded[two_days_ago]["matched_count"] == 5 + assert seeded[one_day_ago]["quality"] == "clear" + assert seeded[one_day_ago]["matched_count"] == 4 + + # Unseeded historical day pads with unknown/no_coverage. + three_days_ago = (today - timedelta(days=3)).isoformat() + assert seeded[three_days_ago]["quality"] == "unknown" + assert seeded[three_days_ago]["state"] == "no_coverage" + assert seeded[three_days_ago]["matched_count"] == 0 diff --git a/tests/test_api/test_routes.py b/tests/test_api/test_routes.py index 2a668c8..8d01c97 100644 --- a/tests/test_api/test_routes.py +++ b/tests/test_api/test_routes.py @@ -1,6 +1,6 @@ """Tests for route API endpoints.""" -from datetime import datetime, timezone +from datetime import datetime, timedelta, timezone from uuid import uuid4 from meshcore_hub.collector.routes import derive_expected_hash @@ -335,6 +335,13 @@ class TestRouteQualityAvg: flapping route that's currently up still shows as marginal/failing if the 7-day mean warrants it. See ``compute_average_quality`` in ``collector/routes.py`` for the algorithm. + + With precomputed history, ``quality_avg`` is sourced from + ``route_result.quality_avg`` (written by the background evaluator). + A fresh route has no history rows yet, so the field stays ``None`` + until the first evaluator tick — the frontend's + ``quality_avg || route_result?.quality || 'unknown'`` fallback chain + covers that gap. """ @staticmethod @@ -361,14 +368,46 @@ class TestRouteQualityAvg: session.commit() return route - def test_present_on_list_for_enabled_routes(self, client_no_auth, api_db_session): - """Each enabled route in the list response carries a computed tier. + @staticmethod + def _seed_quality_avg(session, route: Route, value: str) -> None: + """Write a ``route_result`` row with a precomputed ``quality_avg``. - With no traffic in the test DB every history bucket is no_coverage - which collapses to failing (mean 0); the point of the assertion is - that the field exists and is a valid tier, not the specific value. + Mirrors what the background evaluator would produce once it has + rolled over at least one day's worth of history. + """ + from meshcore_hub.common.models.route_result import RouteResult + + existing = ( + session.query(RouteResult) + .filter(RouteResult.route_id == route.id) + .one_or_none() + ) + if existing is None: + session.add( + RouteResult( + route_id=route.id, + state="healthy", + quality="clear", + matched_count=1, + threshold=route.packet_count_threshold, + effective_clear=route.packet_count_threshold * 2, + quality_avg=value, + ) + ) + else: + existing.quality_avg = value + session.commit() + + def test_present_on_list_for_enabled_routes(self, client_no_auth, api_db_session): + """Each enabled route in the list response carries the persisted tier. + + With precomputed storage, ``quality_avg`` reflects whatever the + background evaluator last wrote on ``route_result``. A fresh route + with no evaluator tick yet has ``None``; the point of this + assertion is that the field surfaces verbatim from the DB. """ route = self._make_route(api_db_session, enabled=True, label="Enabled") + self._seed_quality_avg(api_db_session, route, "failing") resp = client_no_auth.get("/api/v1/routes") assert resp.status_code == 200 items = resp.json()["items"] @@ -378,36 +417,31 @@ class TestRouteQualityAvg: assert len(matching) == 1 avg = matching[0]["quality_avg"] assert avg in {"clear", "marginal", "failing"} - # No traffic in the test DB -> all no_coverage -> failing. + # Seeded value passes through verbatim. assert avg == "failing" def test_none_for_disabled_routes(self, client_no_auth, api_db_session): - """Disabled routes skip the computation; field is null. - - Uses the per-route DETAIL endpoint rather than the list — the - list query races with other xdist workers' ``_truncate_all`` - teardown against the shared SQLite file (the conftest only - isolates Postgres backends, not SQLite). The detail endpoint - scopes the read to a single row so a parallel truncate either - takes the row (404, not a false-pass) or leaves it. - """ + """Disabled routes never carry a quality_avg (None regardless of + what the evaluator wrote).""" route = self._make_route(api_db_session, enabled=False, label="Disabled") resp = client_no_auth.get(f"/api/v1/routes/{route.id}") assert resp.status_code == 200 assert resp.json()["quality_avg"] is None def test_present_on_detail(self, client_no_auth, api_db_session): - """Detail endpoint also exposes the rolling average.""" + """Detail endpoint surfaces the persisted rolling average.""" route = self._make_route(api_db_session, enabled=True, label="Detail") + self._seed_quality_avg(api_db_session, route, "marginal") resp = client_no_auth.get(f"/api/v1/routes/{route.id}") assert resp.status_code == 200 - assert resp.json()["quality_avg"] == "failing" + assert resp.json()["quality_avg"] == "marginal" def test_none_on_create_response(self, client_no_auth, api_db_session): """Create handler skips the rolling computation. - A brand-new route has no meaningful 7-day history; the frontend - falls back to ``route_result.quality`` via the + A brand-new route has no meaningful 7-day history; the create + response always returns ``None`` and the frontend falls back to + ``route_result.quality`` via the ``q = route.quality_avg || route.route_result?.quality`` chain. """ nodes = _sample_nodes(api_db_session) @@ -425,8 +459,33 @@ class TestRouteQualityAvg: assert resp.json()["quality_avg"] is None def test_computed_on_update_response(self, client_no_auth, api_db_session): - """Update handler recomputes the average (history pre-exists).""" + """Update handler recomputes the average inline. + + ``_reevaluate_route`` runs ``compute_persisted_quality_avg`` which + reads ``route_result_history`` and returns ``None`` when no rows + exist yet (a fresh route with no hourly backfill under its belt). + Seeding history before the PUT exercises the populated path. + """ + from meshcore_hub.common.models.route_result_history import ( + RouteResultHistory, + ) + from datetime import date + route = self._make_route(api_db_session, enabled=True, label="UpdateMe") + # Seed 7 days of failing history so the average resolves to failing. + today = date.today() + for i in range(1, 8): + api_db_session.add( + RouteResultHistory( + route_id=route.id, + date=today - timedelta(days=i), + quality="failing", + state="unhealthy", + matched_count=0, + ) + ) + api_db_session.commit() + resp = client_no_auth.put( f"/api/v1/routes/{route.id}", json={"description": "now with description"}, @@ -1076,3 +1135,127 @@ class TestPreviewGuards: }, ) assert resp.status_code == 200 + + +class TestPrecomputedRecentMatches: + """``route_recent_matches`` table → detail-page read path. + + The 60s evaluator sweep persists top-3 matches as normalized rows + (``route_id``, ``raw_packet_id``, ``first_position``, ``last_position``). + The detail endpoint JOINs through ``raw_packets`` and slices the + packet's hops on read, so the JSON shape stays identical to the + legacy on-demand compute — but the data is sourced from its + canonical home in ``packet_path_hops``. + """ + + def _seed_route_with_match(self, session) -> tuple[Route, str]: + node_a = _make_node(session, "aa" + "0" * 62) + node_b = _make_node(session, "bb" + "0" * 62) + route = Route( + from_label="Rm", + to_label="Detail", + packet_count_threshold=1, + clear_threshold=2, + ) + session.add(route) + session.flush() + for pos, n in enumerate([node_a, node_b]): + session.add( + RouteNode( + route_id=route.id, + node_id=n.id, + position=pos, + expected_hash=derive_expected_hash(n.public_key, 1), + ) + ) + rp_id = _make_reception( + session, + observer_node_id=None, + packet_hash="pkt-rm", + path_hashes=["XX", "AA", "BB", "ZZ"], + ) + session.commit() + return route, rp_id + + def test_detail_reads_from_normalized_table(self, client_no_auth, api_db_session): + """When the table is populated, the detail endpoint JOINs through + ``raw_packets`` / ``packet_path_hops`` instead of computing live.""" + from meshcore_hub.common.models import RouteRecentMatch + + route, rp_id = self._seed_route_with_match(api_db_session) + # Seed a normalized match row with the matched subpath positions + # (indices 1..2 → ["AA", "BB"]). + api_db_session.add( + RouteRecentMatch( + route_id=route.id, + raw_packet_id=rp_id, + first_position=1, + last_position=2, + ) + ) + api_db_session.commit() + + resp = client_no_auth.get(f"/api/v1/routes/{route.id}") + assert resp.status_code == 200 + data = resp.json() + assert len(data["recent_matches"]) == 1 + match = data["recent_matches"][0] + assert match["packet_hash"] == "pkt-rm" + # Sliced subpath excludes the noise before/after the matched nodes. + assert [h["node_hash"] for h in match["hops"]] == ["AA", "BB"] + + def test_detail_falls_back_to_live_when_table_empty( + self, client_no_auth, api_db_session + ): + """When the table has no rows for the route (fresh, evaluator hasn't + run yet), the detail endpoint computes matches live so the page + still renders.""" + route, _rp_id = self._seed_route_with_match(api_db_session) + # No RouteRecentMatch row — exercise the live fallback. + + resp = client_no_auth.get(f"/api/v1/routes/{route.id}") + assert resp.status_code == 200 + data = resp.json() + assert len(data["recent_matches"]) == 1 + match = data["recent_matches"][0] + assert match["packet_hash"] == "pkt-rm" + # Live path also slices the matched subpath. + assert [h["node_hash"] for h in match["hops"]] == ["AA", "BB"] + + def test_put_persists_normalized_matches(self, client_no_auth, api_db_session): + """PUT triggers ``_reevaluate_route`` which writes through the + normalized table — the subsequent GET reads from there.""" + from sqlalchemy import select + + from meshcore_hub.common.models import RouteRecentMatch + + route, _rp_id = self._seed_route_with_match(api_db_session) + # No rows yet. + existing = ( + api_db_session.execute( + select(RouteRecentMatch).where(RouteRecentMatch.route_id == route.id) + ) + .scalars() + .all() + ) + assert existing == [] + + resp = client_no_auth.put( + f"/api/v1/routes/{route.id}", + json={"description": "trigger reeval"}, + headers={"X-User-Roles": "admin"}, + ) + assert resp.status_code == 200 + + # The synchronous re-evaluation on PUT should have written a + # match row through ``upsert_route_recent_matches``. + rows = ( + api_db_session.execute( + select(RouteRecentMatch).where(RouteRecentMatch.route_id == route.id) + ) + .scalars() + .all() + ) + assert len(rows) == 1 + assert rows[0].first_position == 1 + assert rows[0].last_position == 2 diff --git a/tests/test_collector/test_route_evaluator.py b/tests/test_collector/test_route_evaluator.py index a4be645..463521b 100644 --- a/tests/test_collector/test_route_evaluator.py +++ b/tests/test_collector/test_route_evaluator.py @@ -1,11 +1,14 @@ """Tests for the route evaluator.""" -from datetime import datetime, timezone +from datetime import datetime, timedelta, timezone from uuid import uuid4 from sqlalchemy import select -from meshcore_hub.collector.route_evaluator import run_evaluation +from meshcore_hub.collector.route_evaluator import ( + run_evaluation, + run_history_backfill, +) from meshcore_hub.collector.routes import derive_expected_hash from meshcore_hub.common.models import ( Node, @@ -13,7 +16,9 @@ from meshcore_hub.common.models import ( RawPacket, Route, RouteNode, + RouteRecentMatch, RouteResult, + RouteResultHistory, RouteQuality, RouteState, ) @@ -124,6 +129,236 @@ class TestRunEvaluation: def _boom(*_args, **_kwargs): raise RuntimeError("eval failed") + # Patch both the source module and the evaluator's bound import so + # the boom is effective regardless of how ``_evaluate_one`` resolves + # ``evaluate_route``. monkeypatch.setattr("meshcore_hub.collector.routes.evaluate_route", _boom) + monkeypatch.setattr( + "meshcore_hub.collector.route_evaluator.evaluate_route", _boom + ) count = run_evaluation(db_manager, now=_NOW) assert count == 0 + + +class TestPrecomputedRecentMatches: + """The 60s sweep populates ``route_recent_matches`` (normalized table).""" + + def test_populates_matches_with_positions(self, db_manager, db_session): + """The sweep writes one ``RouteRecentMatch`` per matching reception + with the matched subpath's position bounds.""" + node_a = _make_node(db_session, "aa" + "0" * 62) + node_b = _make_node(db_session, "bb" + "0" * 62) + route = _make_route( + db_session, "R1", [node_a, node_b], packet_count_threshold=1 + ) + # Bracketed path: noise before AA and after BB. + _make_reception(db_session, "pkt0", ["XX", "AA", "BB", "ZZ"]) + db_session.commit() + + run_evaluation(db_manager, now=_NOW) + + rows = ( + db_session.execute( + select(RouteRecentMatch).where(RouteRecentMatch.route_id == route.id) + ) + .scalars() + .all() + ) + assert len(rows) == 1 + assert rows[0].first_position == 1 + assert rows[0].last_position == 2 + + def test_capped_at_three_per_route(self, db_manager, db_session): + """More than 3 matches in the window only retain the top 3 (the + cap enforced at write time).""" + node_a = _make_node(db_session, "aa" + "0" * 62) + node_b = _make_node(db_session, "bb" + "0" * 62) + _make_route(db_session, "R1", [node_a, node_b], packet_count_threshold=1) + for i in range(5): + _make_reception( + db_session, + f"pkt{i}", + ["AA", "BB"], + ts=_NOW - timedelta(hours=i), + ) + db_session.commit() + + run_evaluation(db_manager, now=_NOW) + + rows = db_session.execute(select(RouteRecentMatch)).scalars().all() + assert len(rows) == 3 + + def test_idempotent_replacement(self, db_manager, db_session): + """A second sweep replaces stale matches instead of accumulating.""" + node_a = _make_node(db_session, "aa" + "0" * 62) + node_b = _make_node(db_session, "bb" + "0" * 62) + _make_route(db_session, "R1", [node_a, node_b], packet_count_threshold=1) + _make_reception(db_session, "pkt0", ["AA", "BB"], ts=_NOW) + db_session.commit() + + run_evaluation(db_manager, now=_NOW) + first = db_session.execute(select(RouteRecentMatch)).scalars().all() + assert len(first) == 1 + first_id = first[0].raw_packet_id + + # Run again — should overwrite, not insert a second row. + run_evaluation(db_manager, now=_NOW) + second = db_session.execute(select(RouteRecentMatch)).scalars().all() + assert len(second) == 1 + assert second[0].raw_packet_id == first_id + + +class TestPrecomputedQualityAvg: + """The 60s sweep computes ``quality_avg`` from persisted history.""" + + def test_quality_avg_none_when_no_history(self, db_manager, db_session): + """A brand-new route with no historical buckets gets ``quality_avg=None``. + + The frontend falls back to ``route_result.quality`` via the + ``q = quality_avg || route_result?.quality`` chain. + """ + node_a = _make_node(db_session, "aa" + "0" * 62) + node_b = _make_node(db_session, "bb" + "0" * 62) + route = _make_route(db_session, "R1", [node_a, node_b]) + db_session.commit() + + run_evaluation(db_manager, now=_NOW) + db_session.expire_all() + + result = db_session.execute( + select(RouteResult).where(RouteResult.route_id == route.id) + ).scalar_one() + assert result.quality_avg is None + + def test_quality_avg_from_seeded_history(self, db_manager, db_session): + """With persisted history rows, the sweep computes the rolling average.""" + node_a = _make_node(db_session, "aa" + "0" * 62) + node_b = _make_node(db_session, "bb" + "0" * 62) + route = _make_route(db_session, "R1", [node_a, node_b]) + today = _NOW.date() + # Seed 7 days of failing history. + for i in range(1, 8): + db_session.add( + RouteResultHistory( + route_id=route.id, + date=today - timedelta(days=i), + quality=RouteQuality.FAILING.value, + state=RouteState.UNHEALTHY.value, + matched_count=0, + ) + ) + db_session.commit() + + run_evaluation(db_manager, now=_NOW) + db_session.expire_all() + + result = db_session.execute( + select(RouteResult).where(RouteResult.route_id == route.id) + ).scalar_one() + assert result.quality_avg == RouteQuality.FAILING.value + + +class TestRunHistoryBackfill: + """The hourly sweep populates ``route_result_history`` for completed days.""" + + def test_writes_history_rows_for_completed_days(self, db_manager, db_session): + """The backfill populates one row per completed UTC day in the window.""" + node_a = _make_node(db_session, "aa" + "0" * 62) + node_b = _make_node(db_session, "bb" + "0" * 62) + route = _make_route( + db_session, "R1", [node_a, node_b], packet_count_threshold=1 + ) + # Place matching packets across 3 different completed days. + for days_ago in range(1, 4): + _make_reception( + db_session, + f"pkt-{days_ago}", + ["AA", "BB"], + ts=_NOW - timedelta(days=days_ago, hours=2), + ) + db_session.commit() + + # Backfill exactly 3 days (one row per day in the window). + run_history_backfill(db_manager, days=3, now=_NOW) + + rows = ( + db_session.execute( + select(RouteResultHistory) + .where(RouteResultHistory.route_id == route.id) + .order_by(RouteResultHistory.date) + ) + .scalars() + .all() + ) + # Three completed days each get a row. + assert len(rows) == 3 + # 1 match / threshold 1 ⇒ healthy. eff_clear=2 ⇒ marginal (1 < 2). + for row in rows: + assert row.state == RouteState.HEALTHY.value + assert row.quality == RouteQuality.MARGINAL.value + assert row.matched_count == 1 + + def test_does_not_write_today_bucket(self, db_manager, db_session): + """The backfill skips today's calendar day (the rolling snapshot + in ``route_results`` covers today).""" + node_a = _make_node(db_session, "aa" + "0" * 62) + node_b = _make_node(db_session, "bb" + "0" * 62) + _make_route(db_session, "R1", [node_a, node_b], packet_count_threshold=1) + _make_reception(db_session, "today-pkt", ["AA", "BB"], ts=_NOW) + db_session.commit() + + run_history_backfill(db_manager, days=3, now=_NOW) + + rows = db_session.execute(select(RouteResultHistory)).scalars().all() + # No row for today's date. + today = _NOW.date() + assert all(r.date < today for r in rows) + + def test_skips_when_days_zero(self, db_manager, db_session): + """``days=0`` is a no-op (returns 0 routes backfilled).""" + node_a = _make_node(db_session, "aa" + "0" * 62) + node_b = _make_node(db_session, "bb" + "0" * 62) + _make_route(db_session, "R1", [node_a, node_b]) + db_session.commit() + + count = run_history_backfill(db_manager, days=0, now=_NOW) + assert count == 0 + + def test_idempotent_re_evaluation(self, db_manager, db_session): + """Re-running the backfill overwrites existing history rows in place + (UNIQUE(route_id, date)).""" + node_a = _make_node(db_session, "aa" + "0" * 62) + node_b = _make_node(db_session, "bb" + "0" * 62) + _make_route(db_session, "R1", [node_a, node_b], packet_count_threshold=1) + _make_reception( + db_session, + "pkt", + ["AA", "BB"], + ts=_NOW - timedelta(days=1, hours=2), + ) + db_session.commit() + + run_history_backfill(db_manager, days=3, now=_NOW) + rows_after_first = ( + db_session.execute( + select(RouteResultHistory).where( + RouteResultHistory.date == _NOW.date() - timedelta(days=1) + ) + ) + .scalars() + .all() + ) + assert len(rows_after_first) == 1 + + # Re-run — should overwrite, not duplicate. + run_history_backfill(db_manager, days=3, now=_NOW) + rows_after_second = ( + db_session.execute( + select(RouteResultHistory).where( + RouteResultHistory.date == _NOW.date() - timedelta(days=1) + ) + ) + .scalars() + .all() + ) + assert len(rows_after_second) == 1 diff --git a/tests/test_collector/test_routes.py b/tests/test_collector/test_routes.py index 11aba94..dac2da6 100644 --- a/tests/test_collector/test_routes.py +++ b/tests/test_collector/test_routes.py @@ -559,22 +559,31 @@ class TestRecentMatches: def test_returns_sliced_subpath(self, db_session): """Recent matches return only the hops between From and To, not the - full packet path.""" + full packet path. + + ``recent_matches`` returns ``first_position`` / ``last_position`` + indices into the packet's full path; callers slice on read. The + indices must bracket exactly the matched subpath (noise before + From and after To is excluded). + """ node_a = _make_node(db_session, "aa" + "0" * 62) node_b = _make_node(db_session, "bb" + "0" * 62) route = _make_route(db_session, "R1", [node_a, node_b]) - # Packet path has noise before AA and after BB; only AA..BB should be kept. + # Packet path has noise before AA and after BB; only AA..BB should + # be bracketed by the returned indices. _make_reception(db_session, None, "pkt0", ["XX", "AA", "YY", "BB", "ZZ"]) db_session.commit() matches = recent_matches(db_session, route, limit=3, now=_NOW) assert len(matches) == 1 - hops = matches[0]["hops"] - assert [h["node_hash"] for h in hops] == ["AA", "YY", "BB"] + first_pos = matches[0]["first_position"] + last_pos = matches[0]["last_position"] + assert (first_pos, last_pos) == (1, 3) def test_returns_sliced_subpath_reverse(self, db_session): - """A reverse-direction packet is sliced in traversal order (To..From).""" + """A reverse-direction packet's indices bracket the slice in + traversal order (To..From).""" node_a = _make_node(db_session, "aa" + "0" * 62) node_b = _make_node(db_session, "bb" + "0" * 62) route = _make_route(db_session, "R1", [node_a, node_b], reversible=True) @@ -584,8 +593,9 @@ class TestRecentMatches: matches = recent_matches(db_session, route, limit=3, now=_NOW) assert len(matches) == 1 - hops = matches[0]["hops"] - assert [h["node_hash"] for h in hops] == ["BB", "YY", "AA"] + first_pos = matches[0]["first_position"] + last_pos = matches[0]["last_position"] + assert (first_pos, last_pos) == (1, 3) def test_dedup_by_event_hash_keeps_newest(self, db_session): """Multiple retransmissions of one event return one row, newest first.