diff --git a/.github/workflows/deploy.yml b/.github/workflows/deploy.yml index 5d934fa1..f1c350fa 100644 --- a/.github/workflows/deploy.yml +++ b/.github/workflows/deploy.yml @@ -344,7 +344,36 @@ jobs: curl -sf "$BASE/api/compilation-schemas" -H "Authorization: Bearer $TOKEN" | jq -e 'type == "array"' || { echo "::error::/api/compilation-schemas did not return array"; exit 1; } echo "PASS: compilation-schemas" - echo "Staging smoke tests complete (10/10 passed)" + # 11. Background job E2E — enqueue a ping job, wait for worker to process it. + # Requires auth (side-effecting endpoint). Retry with backoff to handle + # worker startup delays after deploy. + PING_OK=false + for attempt in 1 2 3 4 5; do + PING_RESULT=$(curl -sf "$BASE/health/job-ping" \ + -H "Authorization: Bearer $TOKEN") || { + echo "Attempt $attempt: /health/job-ping request failed, retrying in $((attempt * 5))s..." + sleep $((attempt * 5)) + continue + } + PING_STATUS=$(echo "$PING_RESULT" | jq -r '.status') + if [ "$PING_STATUS" = "ok" ]; then + PING_MS=$(echo "$PING_RESULT" | jq -r '.latency_ms') + echo "PASS: background job E2E (${PING_MS}ms, attempt $attempt)" + PING_OK=true + break + else + PING_ERR=$(echo "$PING_RESULT" | jq -r '.error // "unknown"') + echo "Attempt $attempt: job-ping status=$PING_STATUS error=$PING_ERR, retrying in $((attempt * 5))s..." + sleep $((attempt * 5)) + fi + done + if [ "$PING_OK" != "true" ]; then + echo "::error::Background job E2E failed after 5 attempts" + echo "$PING_RESULT" | jq . 2>/dev/null || true + exit 1 + fi + + echo "Staging smoke tests complete (11/11 passed)" - name: 🚨 Create alert on failure if: failure() @@ -540,6 +569,35 @@ jobs: ADMIN=$(curl -s -o /dev/null -w "%{http_code}" "$BASE/api/admin/stats" -H "Authorization: Bearer $TOKEN") echo "INFO: admin stats responded $ADMIN" + # 10. Background job E2E — enqueue a ping job, wait for worker to process it. + # Requires auth (side-effecting endpoint). Retry with backoff to handle + # worker startup delays after deploy. + PING_OK=false + for attempt in 1 2 3 4 5; do + PING_RESULT=$(curl -sf "$BASE/health/job-ping" \ + -H "Authorization: Bearer $TOKEN") || { + echo "Attempt $attempt: /health/job-ping request failed, retrying in $((attempt * 5))s..." + sleep $((attempt * 5)) + continue + } + PING_STATUS=$(echo "$PING_RESULT" | jq -r '.status') + if [ "$PING_STATUS" = "ok" ]; then + PING_MS=$(echo "$PING_RESULT" | jq -r '.latency_ms') + echo "PASS: background job E2E (${PING_MS}ms, attempt $attempt)" + PING_OK=true + break + else + PING_ERR=$(echo "$PING_RESULT" | jq -r '.error // "unknown"') + echo "Attempt $attempt: job-ping status=$PING_STATUS error=$PING_ERR, retrying in $((attempt * 5))s..." + sleep $((attempt * 5)) + fi + done + if [ "$PING_OK" != "true" ]; then + echo "::error::Background job E2E failed after 5 attempts" + echo "$PING_RESULT" | jq . 2>/dev/null || true + exit 1 + fi + echo "Production functional verification complete" # ----------------------------------------------------------- diff --git a/docs/openapi.yaml b/docs/openapi.yaml index 2bc05b9b..bf3cfa03 100644 --- a/docs/openapi.yaml +++ b/docs/openapi.yaml @@ -4064,6 +4064,38 @@ paths: additionalProperties: true type: object title: Response Deep Health Check Health Deep Get + /health/job-ping: + get: + tags: + - Admin + summary: Job Ping + description: 'Enqueue a lightweight ping job and wait for its result. + + + Proves the full background job pipeline works end-to-end: + + API -> Redis -> ARQ worker -> result. Requires authentication + + because it enqueues a real job (side effect). + + + Returns ``{"status": "ok", ...}`` when the job completes within the + + timeout, or ``{"status": "error", ...}`` on failure or timeout. + + In dev mode (no Redis), returns ``{"status": "skipped"}`` since + + there is no ARQ worker to verify.' + operationId: job_ping_health_job_ping_get + responses: + '200': + description: Successful Response + content: + application/json: + schema: + additionalProperties: true + type: object + title: Response Job Ping Health Job Ping Get /auth/login/{provider}: get: tags: @@ -7033,6 +7065,7 @@ components: - sync_push - sync_pull - poll_rss_feeds + - ping title: JobType description: Type of async job. LLMSettingsResponse: diff --git a/src/wikimind/api/routes/health.py b/src/wikimind/api/routes/health.py index 671ee33d..91cf604b 100644 --- a/src/wikimind/api/routes/health.py +++ b/src/wikimind/api/routes/health.py @@ -4,10 +4,15 @@ availability, stuck source processing jobs, Redis connectivity, and ARQ queue depth. Returns a structured JSON response with per-check status and overall system health. + +The ``/health/job-ping`` endpoint enqueues a lightweight ping job and +polls for its result to verify the full background job pipeline: +API -> Redis -> ARQ worker -> result. """ from __future__ import annotations +import asyncio import time from datetime import timedelta from typing import Any @@ -15,12 +20,15 @@ import structlog from alembic.config import Config as AlembicConfig from alembic.script import ScriptDirectory -from fastapi import APIRouter +from arq.jobs import Job as ArqJob +from arq.jobs import JobStatus as ArqJobStatus +from fastapi import APIRouter, Depends from redis.asyncio import Redis from sqlalchemy import text as sa_text from sqlmodel import select from wikimind._datetime import utcnow_naive +from wikimind.api.deps import get_current_user_id from wikimind.config import get_settings from wikimind.database import get_session_factory from wikimind.models import IngestStatus, Source @@ -186,3 +194,84 @@ async def deep_health_check() -> dict[str, Any]: "status": _overall_status(checks), "checks": checks, } + + +# --------------------------------------------------------------------------- +# Job ping — end-to-end background job verification +# --------------------------------------------------------------------------- + +# Maximum time to wait for the ping job to complete (seconds). +_JOB_PING_TIMEOUT = 30 +# Interval between result polls (seconds). +_JOB_PING_POLL_INTERVAL = 1 + + +# NOTE: This endpoint intentionally has side effects — it enqueues a real +# (lightweight) ARQ job to verify the job queue is functional end-to-end. +# Requires authentication to prevent unauthenticated abuse. +@router.get("/health/job-ping") +async def job_ping(_user_id: str = Depends(get_current_user_id)) -> dict[str, Any]: + """Enqueue a lightweight ping job and wait for its result. + + Proves the full background job pipeline works end-to-end: + API -> Redis -> ARQ worker -> result. Requires authentication + because it enqueues a real job (side effect). + + Returns ``{"status": "ok", ...}`` when the job completes within the + timeout, or ``{"status": "error", ...}`` on failure or timeout. + In dev mode (no Redis), returns ``{"status": "skipped"}`` since + there is no ARQ worker to verify. + """ + from wikimind.jobs.background import get_background_compiler # noqa: PLC0415 + + compiler = get_background_compiler() + + if not compiler.is_prod: + return {"status": "skipped", "reason": "in-process mode (no Redis)"} + + start = time.monotonic() + try: + job_id = await compiler.schedule_ping() + if job_id is None: + return {"status": "error", "error": "Failed to enqueue ping job"} + + # Poll ARQ for the job result + pool = await compiler._get_pool() + for _ in range(int(_JOB_PING_TIMEOUT / _JOB_PING_POLL_INTERVAL)): + arq_job = ArqJob(job_id, redis=pool) + status = await arq_job.status() + if status == ArqJobStatus.complete: + info = await arq_job.result_info() + latency_ms = round((time.monotonic() - start) * 1000) + return { + "status": "ok", + "job_id": job_id, + "result": info.result if info else None, + "latency_ms": latency_ms, + } + if status == ArqJobStatus.not_found: + # Job disappeared — worker may have crashed + latency_ms = round((time.monotonic() - start) * 1000) + return { + "status": "error", + "job_id": job_id, + "error": "Job not found — worker may not be running", + "latency_ms": latency_ms, + } + await asyncio.sleep(_JOB_PING_POLL_INTERVAL) + + latency_ms = round((time.monotonic() - start) * 1000) + return { + "status": "error", + "job_id": job_id, + "error": f"Ping job did not complete within {_JOB_PING_TIMEOUT}s", + "latency_ms": latency_ms, + } + except Exception as exc: + latency_ms = round((time.monotonic() - start) * 1000) + log.warning("health: job-ping failed", error=str(exc)) + return { + "status": "error", + "error": str(exc), + "latency_ms": latency_ms, + } diff --git a/src/wikimind/jobs/background.py b/src/wikimind/jobs/background.py index 252c567a..ca78d6c3 100644 --- a/src/wikimind/jobs/background.py +++ b/src/wikimind/jobs/background.py @@ -72,6 +72,35 @@ def is_prod(self) -> bool: """Return True when a real Redis URL is configured.""" return self._redis_url is not None + async def schedule_ping(self) -> str | None: + """Enqueue a lightweight ping job and return the ARQ job ID. + + Returns the ARQ job ID on success (prod mode) or ``None`` + in dev mode where no Redis is available. + + Returns: + The ARQ job ID string, or None in dev mode. + """ + if not self.is_prod: + return None + + pool = await self._get_pool() + job = await pool.enqueue_job("ping") + return job.job_id if job else None + + async def _get_pool(self) -> ArqRedis: + """Return the cached ARQ Redis pool, creating it lazily. + + Returns: + The ArqRedis connection pool. + """ + if self._arq_pool is None: + async with self._pool_lock: + if self._arq_pool is None: + settings = RedisSettings.from_dsn(self._redis_url) # type: ignore[arg-type] + self._arq_pool = await create_pool(settings) + return self._arq_pool + async def schedule_compile( self, source_id: str, @@ -198,12 +227,8 @@ async def _enqueue_arq(self, func_name: str, *args: object) -> None: overhead. An asyncio.Lock guards initialization to prevent duplicate pools under concurrent calls. """ - if self._arq_pool is None: - async with self._pool_lock: - if self._arq_pool is None: # double-check after acquiring lock - settings = RedisSettings.from_dsn(self._redis_url) # type: ignore[arg-type] - self._arq_pool = await create_pool(settings) - await self._arq_pool.enqueue_job(func_name, *args) + pool = await self._get_pool() + await pool.enqueue_job(func_name, *args) async def close(self) -> None: """Close the cached ARQ Redis pool, if any.""" diff --git a/src/wikimind/jobs/worker.py b/src/wikimind/jobs/worker.py index 0da7b075..560b1a43 100644 --- a/src/wikimind/jobs/worker.py +++ b/src/wikimind/jobs/worker.py @@ -226,6 +226,16 @@ async def _on_chunk_progress(message: str) -> None: await sweep_wikilinks(ctx, user_id=user_id) +async def ping(_ctx) -> str: + """Lightweight no-op job that proves the full ARQ pipeline works. + + Returns ``"pong"`` immediately. Used by the production deploy + verification step to confirm API -> Redis -> ARQ worker -> result. + """ + log.info("ping job executed") + return "pong" + + async def compile_source( ctx, source_id: str, @@ -705,7 +715,7 @@ def get_redis_settings() -> RedisSettings: class WorkerSettings: """ARQ worker configuration for production (requires Redis).""" - functions: ClassVar[list] = [compile_source, lint_wiki, recompile_article, sweep_wikilinks] + functions: ClassVar[list] = [ping, compile_source, lint_wiki, recompile_article, sweep_wikilinks] redis_settings = get_redis_settings() _worker_cfg = get_settings().worker max_jobs = _worker_cfg.max_jobs diff --git a/src/wikimind/models/enums.py b/src/wikimind/models/enums.py index f5961466..28f4a631 100644 --- a/src/wikimind/models/enums.py +++ b/src/wikimind/models/enums.py @@ -122,6 +122,7 @@ class JobType(StrEnum): SYNC_PUSH = "sync_push" SYNC_PULL = "sync_pull" POLL_RSS_FEEDS = "poll_rss_feeds" + PING = "ping" class JobStatus(StrEnum):