diff --git a/src/sync/web/app.py b/src/sync/web/app.py index e96dc07..5d7087b 100644 --- a/src/sync/web/app.py +++ b/src/sync/web/app.py @@ -22,6 +22,11 @@ def create_app(state: ServiceState, cfg: SyncConfig) -> FastAPI: title="DataMacroSync", description="Access -> SQL Server incremental sync operational API", version="1.0.0", + # Keep all auto-generated docs under /api alongside the operational + # routes, so the whole HTTP surface shares one prefix. + docs_url="/api/docs", + redoc_url="/api/redoc", + openapi_url="/api/openapi.json", ) deps.bind(state, cfg) app.include_router(health_routes.router) diff --git a/src/sync/web/routes/health.py b/src/sync/web/routes/health.py index 7615f36..7e6aab2 100644 --- a/src/sync/web/routes/health.py +++ b/src/sync/web/routes/health.py @@ -1,29 +1,77 @@ -"""GET /health -- passive health-check endpoint. - -Returns a structured snapshot covering service health (process, cycle lag, -queue state) and data health (last compare result, drift traces, dead rows). - -HTTP status follows the body's ``status`` field so off-the-shelf monitors that -alert on status code work without parsing JSON: - -* ``200`` -- healthy -* ``503`` -- degraded or unhealthy - -This keeps "service unreachable" (network/down) distinguishable from "service -up but needs attention" (200 would conflate them). -""" +"""GET /api/health -- passive health-check endpoint.""" from __future__ import annotations from fastapi import APIRouter, Depends, Response from ..deps import get_checker +from ..schemas import HealthResponse from ...health import HealthChecker -router = APIRouter() +# All operational routes mount under /api (see app.py). The health route lives +# at /api/health so future endpoints (/api/compare, /api/queue, ...) share one +# prefix without per-router configuration. +router = APIRouter(prefix="/api") -@router.get("/health", summary="Sync service health check") -def health(response: Response, checker: HealthChecker = Depends(get_checker)) -> dict: +@router.get( + "/health", + response_model=HealthResponse, + summary="Get the sync service health snapshot", + description=( + "Returns a structured snapshot of the Access → SQL Server incremental " + "sync service. Use this for passive health probing (uptime monitors, " + "load-balancer checks, on-call dashboards).\n\n" + "**What it covers**\n\n" + "Two layers are reported:\n\n" + "1. **Service health** — whether the sync loop is alive and making " + "progress: the time since the last capture/apply/cleanup cycle " + "(`cycle_lag_seconds`), the current SyncQueue row counts by status, " + "and cumulative capture counters since process start.\n" + "2. **Data health** — lightweight signals about Access ↔ SQL Server " + "consistency: the result of the most recent scheduled full compare " + "(run daily ~05:08, parsed from its report file — NOT re-run on every " + "request), plus any abnormal audit traces (rows that aged out, " + "downgraded operations) and a sample of dead/error queue rows.\n\n" + "**Status derivation**\n\n" + "The top-level `status` field is derived from the signals above:\n\n" + "- `healthy` — the loop is pacing normally and no attention signals " + "are present.\n" + "- `degraded` — the loop is running but something needs attention: " + "queue rows stuck in `error`/`dead`, an Insert that aged out to dead, " + "or the last compare found mismatches.\n" + "- `unhealthy` — the sync loop appears stalled (cycle lag exceeds " + "3× the poll interval, e.g. wedged on an Access lock wait).\n\n" + "**HTTP status codes**\n\n" + "The HTTP status mirrors `status` so off-the-shelf monitors that alert " + "on status code work without parsing JSON:\n\n" + "- `200 OK` — healthy.\n" + "- `503 Service Unavailable` — degraded or unhealthy. This keeps " + "\"service unreachable\" (a network error / timeout, no response at " + "all) distinguishable from \"service up but needs attention\".\n\n" + "**Cost**\n\n" + "Each request performs one short SQL query (queue + audit) and reads " + "an in-memory state snapshot updated by the sync loop; it does not run " + "a full compare. Safe to poll at second-scale intervals." + ), + responses={ + 200: {"description": "Service is healthy."}, + 503: { + "description": ( + "Service is degraded or unhealthy. The body still contains the " + "full snapshot; inspect `status`, `service_health` and " + "`data_health` for the cause." + ), + "model": HealthResponse, + }, + }, +) +def health(response: Response, + checker: HealthChecker = Depends(get_checker)) -> dict: + """Compute and return the health snapshot. + + The heavy lifting lives in :class:`sync.health.HealthChecker`; this route + is a thin HTTP adapter that maps the resulting status to an HTTP code. + """ snap = checker.snapshot() # Map body status to HTTP code for code-based alerting. if snap["status"] != "healthy": diff --git a/src/sync/web/schemas.py b/src/sync/web/schemas.py new file mode 100644 index 0000000..4df5b85 --- /dev/null +++ b/src/sync/web/schemas.py @@ -0,0 +1,204 @@ +"""Pydantic response models for the API. + +Declaring these (and passing them as ``response_model`` on routes) makes the +auto-generated OpenAPI/Swagger docs at ``/api/docs`` show the full response +shape per field, rather than a bare ``object``. Keep them structurally aligned +with the dicts produced by :class:`sync.health.HealthChecker`. +""" +from __future__ import annotations + +from pydantic import BaseModel, Field + + +class QueueStatus(BaseModel): + """Live SyncQueue row counts grouped by Status.""" + + pending: int | None = Field( + default=None, + description="Rows staged but not yet applied by usp_SyncApply. " + "Briefly non-zero right after capture; should drain to 0 " + "each cycle.", + ) + error: int | None = Field( + default=None, + description="Rows whose apply failed and are under the retry budget. " + "Non-zero indicates a transient apply failure; they are " + "re-queued automatically. Drives the `degraded` status.", + ) + dead: int | None = Field( + default=None, + description="Rows that exhausted retries (or were parked as dead on " + "capture). These represent changes that exist in Access " + "but never reached SQL Server and need manual review. " + "Drives the `degraded` status.", + ) + cleaned: int | None = Field( + default=None, + description="Rows successfully applied and whose Access change-log " + "entry has been removed. Retained for a short audit " + "window then purged; not an error signal.", + ) + + +class CaptureSinceStart(BaseModel): + """Cumulative capture counters since the service process started.""" + + enqueued: int = Field( + default=0, + description="Total change-log rows successfully staged into SyncQueue " + "since process start.", + ) + deferred: int = Field( + default=0, + description="Total Insert/Update log rows whose source row was " + "momentarily unreadable (ACE visibility latency) and were " + "deferred for retry on a later cycle. Occasional non-zero " + "values are normal under bulk-insert bursts.", + ) + aged_out: int = Field( + default=0, + description="Total Insert/Update log rows still unreadable past the " + "defer window and parked as dead for manual review. " + "Non-zero drives the `degraded` status — these rows never " + "reached SQL Server.", + ) + + +class ServiceHealth(BaseModel): + """Live process and sync-loop metrics.""" + + last_cycle_at: str | None = Field( + default=None, + description="ISO 8601 timestamp of the most recent capture→apply→" + "cleanup cycle completion. Null until the first cycle " + "finishes.", + ) + last_cycle_duration_s: float | None = Field( + default=None, + description="Wall-clock duration of the last cycle, in seconds.", + ) + cycle_lag_seconds: float | None = Field( + default=None, + description="Seconds elapsed since `last_cycle_at`. The primary " + "liveness indicator: a healthy idle loop paces at " + "poll_interval (default 10s); exceeding 3× poll_interval " + "flips the status to `unhealthy`.", + ) + last_cycle_active: bool = Field( + default=False, + description="True if the last cycle actually processed changes " + "(captured/applied/cleaned anything), False if it was an " + "idle pass.", + ) + last_cycle_error: str | None = Field( + default=None, + description="Error message if the last cycle raised unexpectedly at " + "the top level (per-file failures are isolated and do not " + "set this). Null when the cycle completed normally.", + ) + queue: QueueStatus = Field( + description="Live SyncQueue row counts by status.", + ) + capture_since_start: CaptureSinceStart = Field( + description="Cumulative capture counters since process start.", + ) + + +class LastCompare(BaseModel): + """Result of the most recent scheduled full compare (parsed, not re-run).""" + + available: bool = Field( + default=False, + description="Whether a compare report file was found. False when no " + "report has been generated yet (e.g. before the first " + "daily run) or the logs directory is unreadable.", + ) + report: str | None = Field( + default=None, + description="File name of the parsed report (e.g. " + "compare_ids_2026-08-05.log).", + ) + at: str | None = Field( + default=None, + description="ISO 8601 timestamp the compare report was generated.", + ) + tables_match: int | None = Field( + default=None, + description="Number of compared tables whose Access and SQL row-ID " + "sets matched.", + ) + tables_mismatch: int | None = Field( + default=None, + description="Number of compared tables that diverged. Non-zero drives " + "the `degraded` status via `mismatch`.", + ) + mismatch: bool = Field( + default=False, + description="True if any table diverged in the last compare. Drives " + "the `degraded` status. Note this reflects the last " + "scheduled compare (daily), not a real-time check.", + ) + + +class DriftTable(BaseModel): + """A table with abnormal audit traces since process start.""" + + schema: str = Field(description="Target SQL Server schema name.") + table: str = Field(description="Target table name (with year suffix).") + kind: str = Field( + description="Trace kind: 'AgedOut' (Insert unreadable past defer " + "window) or the processed operate type when it diverged " + "from the original.", + ) + count: int = Field(description="Number of abnormal audit rows for this table.") + + +class DataHealth(BaseModel): + """Lightweight Access ↔ SQL Server consistency signals.""" + + last_compare: LastCompare = Field( + description="Result of the most recent scheduled full compare, parsed " + "from its report file. A full compare is NOT executed on " + "every health request (too heavy).", + ) + drift_tables: list[DriftTable] = Field( + default_factory=list, + description="Tables with abnormal audit traces (aged-out or " + "downgraded operations) recorded since process start. " + "Empty under normal operation.", + ) + dead_rows_sample: list[dict] = Field( + default_factory=list, + description="Up to 5 sample dead/error queue rows for triage " + "(table, record id, error message, ...). Empty when the " + "queue has no stuck rows.", + ) + + +class HealthResponse(BaseModel): + """Top-level health snapshot returned by GET /api/health.""" + + service: str = Field( + description="Service identifier (always 'DataMacroSync').", + ) + status: str = Field( + description="Derived overall status: 'healthy', 'degraded', or " + "'unhealthy'. Mirrored to the HTTP status code " + "(200 / 503).", + ) + checked_at: str = Field( + description="ISO 8601 timestamp the snapshot was generated.", + ) + pid: int = Field(description="OS process id of the sync service.") + started_at: str = Field( + description="ISO 8601 timestamp the service process started.", + ) + uptime_seconds: int = Field( + description="Seconds since the service process started.", + ) + service_health: ServiceHealth = Field( + description="Live process and sync-loop metrics.", + ) + data_health: DataHealth = Field( + description="Lightweight Access ↔ SQL Server consistency signals.", + ) diff --git a/tests/test_health.py b/tests/test_health.py index 43a3760..66b1fd7 100644 --- a/tests/test_health.py +++ b/tests/test_health.py @@ -143,7 +143,7 @@ def test_health_endpoint_returns_200_when_healthy(): return_value={"available": False}): app = create_app(state, cfg) client = TestClient(app) - r = client.get("/health") + r = client.get("/api/health") assert r.status_code == 200 body = r.json() assert body["status"] == "healthy" @@ -162,7 +162,7 @@ def test_health_endpoint_returns_503_when_unhealthy(): return_value={"available": False}): app = create_app(state, cfg) client = TestClient(app) - r = client.get("/health") + r = client.get("/api/health") assert r.status_code == 503 assert r.json()["status"] == "unhealthy" @@ -178,6 +178,6 @@ def test_health_endpoint_returns_503_when_degraded(): return_value={"available": False}): app = create_app(state, cfg) client = TestClient(app) - r = client.get("/health") + r = client.get("/api/health") assert r.status_code == 503 assert r.json()["status"] == "degraded"