feat(worker): ручной пересчёт метрик через очередь, диагностика шагов и heartbeat

POST /metrics/refresh ставит задачу в sync_job (source=METRICS_JOB) и отвечает 202, пересчёт делает worker; GET /metrics/status отдаёт refreshing и consistent. metric_refresh_log хранит failed_step и step_timings. Источники с needs="tinvest_token" не попадают в расписание без токена, tinvest/moex добавлены в default_schedule. Воркер трогает heartbeat-файл для healthcheck.

backend/src/fintracker/api/routers/metrics.py

@@ -4,9 +4,9 @@ from fastapi import APIRouter, status
 from sqlalchemy import select
 
 from fintracker.api.deps import CurrentUser, SessionDep
-from fintracker.api.schemas.metrics import DataQualityRow, RefreshLogOut
-from fintracker.metrics.refresh import refresh_all
-from fintracker.models import MetricDataQuality, MetricRefreshLog
+from fintracker.api.schemas.metrics import DataQualityRow, MetricsStatusOut, RefreshLogOut
+from fintracker.api.schemas.sync import SyncJobOut
+from fintracker.models import METRICS_JOB, JobStatus, MetricDataQuality, MetricRefreshLog, SyncJob
 
 router = APIRouter(tags=["metrics"])
 
@@ -22,9 +22,10 @@ async def data_quality(session: SessionDep, _: CurrentUser) -> list[DataQualityR
 
 
 @router.get("/metrics/status", name="status")
-async def metrics_status(session: SessionDep, _: CurrentUser) -> RefreshLogOut | None:
-    """When the metric tables were last rebuilt, and whether it failed."""
-    row = (
+async def metrics_status(session: SessionDep, _: CurrentUser) -> MetricsStatusOut:
+    """When the metric tables were last rebuilt, whether they are one consistent snapshot,
+    and whether another rebuild is on its way."""
+    latest = (
         (
             await session.execute(
                 select(MetricRefreshLog).order_by(MetricRefreshLog.started_at.desc()).limit(1)
@@ -33,13 +34,54 @@ async def metrics_status(session: SessionDep, _: CurrentUser) -> RefreshLogOut |
         .scalars()
         .first()
     )
-    if row is None:
-        return None
-    return RefreshLogOut.model_validate(row, from_attributes=True)
+    last_finished = (
+        (
+            await session.execute(
+                select(MetricRefreshLog)
+                .where(MetricRefreshLog.finished_at.is_not(None))
+                .order_by(MetricRefreshLog.started_at.desc())
+                .limit(1)
+            )
+        )
+        .scalars()
+        .first()
+    )
+    queued = (
+        await session.execute(
+            select(SyncJob.id)
+            .where(
+                SyncJob.source == METRICS_JOB,
+                SyncJob.status.in_([JobStatus.queued, JobStatus.running]),
+            )
+            .limit(1)
+        )
+    ).first()
+    return MetricsStatusOut(
+        last_refresh=RefreshLogOut.model_validate(latest, from_attributes=True) if latest else None,
+        consistent=last_finished is None or last_finished.error is None,
+        refreshing=queued is not None or (latest is not None and latest.finished_at is None),
+    )
 
 
 @router.post("/metrics/refresh", name="refresh", status_code=status.HTTP_202_ACCEPTED)
-async def metrics_refresh(session: SessionDep, _: CurrentUser) -> RefreshLogOut:
-    """Rebuild every metric_* table inline (seconds at personal volumes)."""
-    entry = await refresh_all(session, trigger="manual")
-    return RefreshLogOut.model_validate(entry, from_attributes=True)
+async def metrics_refresh(session: SessionDep, _: CurrentUser) -> SyncJobOut:
+    """Queue a rebuild of every metric_* table; the worker runs it. Poll `/metrics/status`
+    until `refreshing` is false.
+
+    A request while one is already waiting shares it. One that arrives while a rebuild is
+    RUNNING queues another, because the running one may have read the data before the change
+    that prompted this call."""
+    waiting = (
+        await session.execute(
+            select(SyncJob)
+            .where(SyncJob.source == METRICS_JOB, SyncJob.status == JobStatus.queued)
+            .limit(1)
+        )
+    ).scalar_one_or_none()
+    if waiting is not None:
+        return SyncJobOut.model_validate(waiting, from_attributes=True)
+    job = SyncJob(source=METRICS_JOB, status=JobStatus.queued)
+    session.add(job)
+    await session.commit()
+    await session.refresh(job)
+    return SyncJobOut.model_validate(job, from_attributes=True)

backend/src/fintracker/api/schemas/metrics.py

@@ -23,3 +23,17 @@ class RefreshLogOut(BaseModel):
     finished_at: datetime | None
     trigger: str
     error: str | None
+    failed_step: str | None
+    """The step that raised; earlier steps committed, later ones did not run."""
+    step_timings: dict[str, float] | None
+    """Seconds per step that completed, in run order."""
+
+
+class MetricsStatusOut(BaseModel):
+    last_refresh: RefreshLogOut | None
+    """Latest run, finished or not; null before the first one."""
+    consistent: bool
+    """False when the latest FINISHED run failed part-way: some metric tables are from it and
+    the rest from an earlier run, so numbers on different screens may disagree."""
+    refreshing: bool
+    """A rebuild is queued or running, so `last_refresh` is about to be superseded."""

backend/src/fintracker/metrics/refresh.py

@@ -1,8 +1,8 @@
 """Rebuild every metric_* table after data changed (plan §3).
 
-Order (each step is a function taking an AsyncSession and committing its own tables):
-    fx -> classify -> net worth -> cash flow -> spending -> runway -> data quality
-Phase 2 inserts prices/lots/valuation/holdings/returns between fx and net worth.
+Each step is a function taking an AsyncSession and committing its own tables. The steps and
+their order live in `analytics.register_steps`, with the reason for each dependency — this
+module only runs whatever is registered, in registration order.
 
 `refresh_all` is what the worker calls after a sync that reported `changed=True`, what the
 CLI `fintracker metrics refresh` runs, and what `POST /metrics/refresh` queues.
@@ -16,6 +16,7 @@ connection serialises them: a refresh that arrives during a sync WAITS and then
 from __future__ import annotations
 
 import logging
+import time
 import traceback
 from collections.abc import Awaitable, Callable
 from datetime import UTC, datetime
@@ -52,15 +53,22 @@ async def refresh_all(session: AsyncSession, trigger: str) -> MetricRefreshLog:
         entry = MetricRefreshLog(trigger=trigger)
         session.add(entry)
         await session.commit()
+        timings: dict[str, float] = {}
+        current: str | None = None
         try:
             for name, step in STEPS:
+                current = name
                 log.info("metrics: %s", name)
+                started = time.monotonic()
                 await step(session)
                 await session.commit()
+                timings[name] = round(time.monotonic() - started, 3)
         except Exception as exc:
             await session.rollback()
             entry.error = f"{type(exc).__name__}: {exc}\n{traceback.format_exc()[-4000:]}"
-            log.exception("metrics refresh failed at step")
+            entry.failed_step = current
+            log.exception("metrics refresh failed at step %s", current)
+        entry.step_timings = timings
         entry.finished_at = datetime.now(UTC)
         await session.commit()
         return entry

backend/src/fintracker/models/__init__.py

@@ -72,6 +72,7 @@ from fintracker.models.reports import (
     ReportParseStatus,
 )
 from fintracker.models.sync import (
+    METRICS_JOB,
     JobStatus,
     RunStatus,
     SourceCredential,
@@ -101,6 +102,7 @@ from fintracker.models.zenmoney import (
 
 __all__ = [
     "EXTERNAL_FLOW_KINDS",
+    "METRICS_JOB",
     "POSITION_KINDS",
     "Account",
     "AccountKind",

backend/src/fintracker/models/metrics.py

@@ -135,6 +135,11 @@ class MetricRefreshLog(Base):
     trigger: Mapped[str] = mapped_column(String(32))
     """sync:<source> | manual | cli"""
     error: Mapped[str | None] = mapped_column(Text)
+    failed_step: Mapped[str | None] = mapped_column(String(64))
+    """Name of the step that raised. Steps before it committed and the ones after it did not
+    run, so the metric tables then come from two different runs."""
+    step_timings: Mapped[dict[str, Any] | None]
+    """Seconds per step that completed, in run order."""
 
 
 class MetricPortfolioValueDaily(Base):

backend/src/fintracker/models/sync.py

@@ -19,6 +19,11 @@ class RunStatus(enum.StrEnum):
     error = "error"
 
 
+METRICS_JOB = "metrics"
+"""`sync_job.source` of a queued metrics rebuild. Not a registered source: the worker
+routes it to `refresh_all` instead of `run_source`, and `POST /sync/{source}` answers 404."""
+
+
 class JobStatus(enum.StrEnum):
     queued = "queued"
     running = "running"
@@ -55,7 +60,8 @@ class SyncRun(Base):
 
 
 class SyncJob(Base):
-    """Manual trigger queue: API inserts, worker picks up (plan §2.4)."""
+    """Manual trigger queue: API inserts, worker picks up (plan §2.4). Also carries the metrics
+    rebuild requested by `POST /metrics/refresh`, as `source == METRICS_JOB`."""
 
     __tablename__ = "sync_job"
 

backend/src/fintracker/worker/jobs.py

@@ -9,6 +9,8 @@ from apscheduler.triggers.base import BaseTrigger
 from apscheduler.triggers.combining import OrTrigger
 from apscheduler.triggers.cron import CronTrigger
 
+from fintracker.config import Settings
+
 
 @dataclass(frozen=True)
 class JobSpec:
@@ -16,10 +18,13 @@ class JobSpec:
     trigger: BaseTrigger
     """A source appears at most once: `worker/scheduler.py` keys jobs by `sync:<source>`,
     so several times of day are one `OrTrigger`, not several specs."""
+    needs: str | None = None
+    """Name of a `Settings` field that must be set for the source to run at all. Without it
+    the job is left off the timetable: a run would only fail, hourly, and bury real errors."""
 
 
 MSK = ZoneInfo("Europe/Moscow")
-"""CBR publishes on Moscow time, so those jobs pin the zone instead of following settings."""
+"""CBR and MOEX publish on Moscow time, so those jobs pin the zone instead of following settings."""
 
 
 def default_schedule() -> list[JobSpec]:
@@ -36,4 +41,24 @@ def default_schedule() -> list[JobSpec]:
                 ]
             ),
         ),
+        # Broker operations and balances, every three hours. The MOEX run below is placed
+        # after a T-Invest one, because it prices whatever the ledger says is held.
+        JobSpec(
+            "tinvest", CronTrigger(hour="8-23/3", minute=10, timezone=MSK), needs="tinvest_token"
+        ),
+        # Prices: after the open, mid-session, after the close (main session ends 18:50 MSK),
+        # and once more for the evening session.
+        JobSpec("moex", CronTrigger(hour="10,14,19,23", minute=20, timezone=MSK)),
+        # Announced dividends and coupons change rarely: once a day, before the market opens.
+        JobSpec(
+            "tinvest_events", CronTrigger(hour=6, minute=30, timezone=MSK), needs="tinvest_token"
+        ),
+        JobSpec("moex_payouts", CronTrigger(hour=6, minute=45, timezone=MSK)),
     ]
+
+
+def scheduled(settings: Settings) -> tuple[list[JobSpec], list[JobSpec]]:
+    """(to run, left off because a required setting is missing)."""
+    specs = default_schedule()
+    ready = [s for s in specs if s.needs is None or getattr(settings, s.needs)]
+    return ready, [s for s in specs if s not in ready]

backend/src/fintracker/worker/scheduler.py

@@ -6,20 +6,30 @@ import asyncio
 import logging
 import signal
 from datetime import UTC, datetime
+from pathlib import Path
 
 from apscheduler.schedulers.asyncio import AsyncIOScheduler
 from sqlalchemy import select
 
 from fintracker.config import get_settings
 from fintracker.db import get_sessionmaker, reset_engine
-from fintracker.models import JobStatus, SyncJob
+from fintracker.metrics.refresh import refresh_all
+from fintracker.models import METRICS_JOB, JobStatus, SyncJob
 from fintracker.sources import registry
-from fintracker.worker.jobs import default_schedule
+from fintracker.worker.jobs import scheduled
 from fintracker.worker.runner import Skipped, run_source
 
 log = logging.getLogger(__name__)
 
 POLL_SECONDS = 5
+HEARTBEAT_SECONDS = 10
+HEARTBEAT_FILE = Path("/tmp/fintracker-worker-heartbeat")
+"""Touched by its own job, so a long sync (which blocks `_process_queued_jobs`, not the event
+loop) does not look like a dead worker. The compose healthcheck reads its mtime."""
+
+
+async def _heartbeat() -> None:
+    HEARTBEAT_FILE.touch()
 
 
 async def _scheduled(source: str) -> None:
@@ -43,6 +53,9 @@ async def _process_queued_jobs() -> None:
             .all()
         )
     for job in jobs:
+        if job.source == METRICS_JOB:
+            await _run_metrics_job(job.id)
+            continue
         if job.source not in registry.names():
             await _finish_job(job.id, JobStatus.error, error=f"unknown source {job.source}")
             continue
@@ -56,6 +69,20 @@ async def _process_queued_jobs() -> None:
         await _finish_job(job.id, status, run_id=run.id, error=run.error)
 
 
+async def _run_metrics_job(job_id) -> None:
+    await _mark_running(job_id)
+    try:
+        async with get_sessionmaker()() as session:
+            entry = await refresh_all(session, trigger="manual")
+    except Exception as exc:  # the lock connection or the log insert, not a step (those are caught)
+        log.exception("metrics job failed")
+        await _finish_job(job_id, JobStatus.error, error=f"{type(exc).__name__}: {exc}")
+        return
+    await _finish_job(
+        job_id, JobStatus.done if entry.error is None else JobStatus.error, error=entry.error
+    )
+
+
 async def _mark_running(job_id) -> None:
     async with get_sessionmaker()() as session:
         job = await session.get(SyncJob, job_id)
@@ -79,7 +106,10 @@ async def _finish_job(job_id, status: JobStatus, *, run_id=None, error: str | No
 def build_scheduler() -> AsyncIOScheduler:
     settings = get_settings()
     scheduler = AsyncIOScheduler(timezone=settings.timezone)
-    for spec in default_schedule():
+    specs, skipped = scheduled(settings)
+    for spec in skipped:
+        log.warning("not scheduling %s: %s is not set", spec.source, spec.needs)
+    for spec in specs:
         scheduler.add_job(
             _scheduled,
             spec.trigger,
@@ -89,6 +119,15 @@ def build_scheduler() -> AsyncIOScheduler:
             coalesce=True,
             misfire_grace_time=600,
         )
+    scheduler.add_job(
+        _heartbeat,
+        "interval",
+        seconds=HEARTBEAT_SECONDS,
+        id="heartbeat",
+        max_instances=1,
+        coalesce=True,
+        next_run_time=datetime.now(UTC),
+    )
     scheduler.add_job(
         _process_queued_jobs,
         "interval",