feat: isolate readiness probes from Gitea traffic (Closes #687)
This commit is contained in:
parent
9ba03613d4
commit
d2b2c7bada
110
src/main.py
110
src/main.py
|
|
@ -10,6 +10,7 @@ import sqlite3
|
||||||
import time
|
import time
|
||||||
from collections.abc import Awaitable, Coroutine
|
from collections.abc import Awaitable, Coroutine
|
||||||
from contextlib import asynccontextmanager
|
from contextlib import asynccontextmanager
|
||||||
|
from dataclasses import dataclass
|
||||||
from datetime import datetime
|
from datetime import datetime
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
from typing import Any, Literal
|
from typing import Any, Literal
|
||||||
|
|
@ -59,6 +60,9 @@ from src.views import FRONTEND_BUILD, router as frontend_router
|
||||||
|
|
||||||
|
|
||||||
async def _drain_authored_action_operations() -> None:
|
async def _drain_authored_action_operations() -> None:
|
||||||
|
for key, operation in list(_authored_action_operations.items()):
|
||||||
|
if operation[1].done():
|
||||||
|
_authored_action_operations.pop(key, None)
|
||||||
tasks = {
|
tasks = {
|
||||||
operation[1]
|
operation[1]
|
||||||
for operation in _authored_action_operations.values()
|
for operation in _authored_action_operations.values()
|
||||||
|
|
@ -116,10 +120,18 @@ async def _push_poll_loop() -> None:
|
||||||
continue
|
continue
|
||||||
|
|
||||||
|
|
||||||
|
async def _readiness_monitor() -> None:
|
||||||
|
while True:
|
||||||
|
await _check_readiness()
|
||||||
|
await asyncio.sleep(READINESS_INTERVAL_SECONDS)
|
||||||
|
|
||||||
|
|
||||||
@asynccontextmanager
|
@asynccontextmanager
|
||||||
async def lifespan(_app: FastAPI):
|
async def lifespan(_app: FastAPI):
|
||||||
global _live_snapshot_task, _available_issue_snapshot_task, _push_poll_task
|
global _live_snapshot_task, _available_issue_snapshot_task, _push_poll_task
|
||||||
|
global _readiness_task
|
||||||
gitea_proxy.start_client()
|
gitea_proxy.start_client()
|
||||||
|
_readiness_task = asyncio.create_task(_readiness_monitor())
|
||||||
if _push_configuration().enabled:
|
if _push_configuration().enabled:
|
||||||
_push_poll_task = asyncio.create_task(_push_poll_loop())
|
_push_poll_task = asyncio.create_task(_push_poll_loop())
|
||||||
try:
|
try:
|
||||||
|
|
@ -128,7 +140,8 @@ async def lifespan(_app: FastAPI):
|
||||||
live_task = _live_snapshot_task
|
live_task = _live_snapshot_task
|
||||||
available_task = _available_issue_snapshot_task
|
available_task = _available_issue_snapshot_task
|
||||||
push_task = _push_poll_task
|
push_task = _push_poll_task
|
||||||
for task in (live_task, available_task, push_task):
|
readiness_task = _readiness_task
|
||||||
|
for task in (live_task, available_task, push_task, readiness_task):
|
||||||
if task is not None and not task.done():
|
if task is not None and not task.done():
|
||||||
task.cancel()
|
task.cancel()
|
||||||
try:
|
try:
|
||||||
|
|
@ -145,6 +158,8 @@ async def lifespan(_app: FastAPI):
|
||||||
_available_issue_snapshot_task = None
|
_available_issue_snapshot_task = None
|
||||||
if _push_poll_task is push_task:
|
if _push_poll_task is push_task:
|
||||||
_push_poll_task = None
|
_push_poll_task = None
|
||||||
|
if _readiness_task is readiness_task:
|
||||||
|
_readiness_task = None
|
||||||
|
|
||||||
|
|
||||||
app = FastAPI(title="Stackchain Dashboard", lifespan=lifespan)
|
app = FastAPI(title="Stackchain Dashboard", lifespan=lifespan)
|
||||||
|
|
@ -153,6 +168,60 @@ app.add_middleware(NegotiatedGZipMiddleware, minimum_size=1_024)
|
||||||
CONTEXT_TIMEOUT_SECONDS = 5.0
|
CONTEXT_TIMEOUT_SECONDS = 5.0
|
||||||
EVENT_STREAM_TIMEOUT_SECONDS = 5.0
|
EVENT_STREAM_TIMEOUT_SECONDS = 5.0
|
||||||
READINESS_TIMEOUT_SECONDS = 5.0
|
READINESS_TIMEOUT_SECONDS = 5.0
|
||||||
|
READINESS_INTERVAL_SECONDS = max(
|
||||||
|
READINESS_TIMEOUT_SECONDS,
|
||||||
|
float(os.getenv("STACKCHAIN_READINESS_INTERVAL_SECONDS", "30")),
|
||||||
|
)
|
||||||
|
READINESS_MAX_AGE_SECONDS = max(
|
||||||
|
READINESS_INTERVAL_SECONDS + READINESS_TIMEOUT_SECONDS + 5.0,
|
||||||
|
float(os.getenv("STACKCHAIN_READINESS_MAX_AGE_SECONDS", "65")),
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
@dataclass(frozen=True)
|
||||||
|
class ReadinessState:
|
||||||
|
checked_at: float
|
||||||
|
ready: bool
|
||||||
|
login: str | None = None
|
||||||
|
error: str | None = None
|
||||||
|
timed_out: bool = False
|
||||||
|
|
||||||
|
|
||||||
|
_readiness_state: ReadinessState | None = None
|
||||||
|
|
||||||
|
|
||||||
|
async def _check_readiness() -> None:
|
||||||
|
global _readiness_state
|
||||||
|
checked_at = time.monotonic()
|
||||||
|
try:
|
||||||
|
user = await asyncio.wait_for(current_user(), timeout=READINESS_TIMEOUT_SECONDS)
|
||||||
|
if not isinstance(user, dict) or not user.get("login"):
|
||||||
|
raise ReadinessPayloadError(
|
||||||
|
"Gitea current-user response did not include a login"
|
||||||
|
)
|
||||||
|
except Exception as exc:
|
||||||
|
timed_out = isinstance(exc, TimeoutError)
|
||||||
|
error = (
|
||||||
|
f"Gitea readiness check timed out after {READINESS_TIMEOUT_SECONDS:g}s"
|
||||||
|
if timed_out
|
||||||
|
else (
|
||||||
|
str(exc)
|
||||||
|
if isinstance(exc, ReadinessPayloadError)
|
||||||
|
else "Gitea readiness check is temporarily unavailable"
|
||||||
|
)
|
||||||
|
)
|
||||||
|
_readiness_state = ReadinessState(
|
||||||
|
checked_at=checked_at,
|
||||||
|
ready=False,
|
||||||
|
error=error,
|
||||||
|
timed_out=timed_out,
|
||||||
|
)
|
||||||
|
return
|
||||||
|
_readiness_state = ReadinessState(
|
||||||
|
checked_at=checked_at,
|
||||||
|
ready=True,
|
||||||
|
login=user["login"],
|
||||||
|
)
|
||||||
REVIEW_DETAIL_TIMEOUT_SECONDS = 5.0
|
REVIEW_DETAIL_TIMEOUT_SECONDS = 5.0
|
||||||
ISSUE_ACTION_TIMEOUT_SECONDS = 5.0
|
ISSUE_ACTION_TIMEOUT_SECONDS = 5.0
|
||||||
ISSUE_CREATION_IDEMPOTENCY_TTL_SECONDS = 600.0
|
ISSUE_CREATION_IDEMPOTENCY_TTL_SECONDS = 600.0
|
||||||
|
|
@ -181,6 +250,7 @@ AVAILABLE_ISSUE_SNAPSHOT_LEASE_SECONDS = WORK_PAGE_TIMEOUT_SECONDS + 1.0
|
||||||
FRONTEND_DIR = Path(__file__).resolve().parent.parent / "frontend"
|
FRONTEND_DIR = Path(__file__).resolve().parent.parent / "frontend"
|
||||||
_live_snapshot_task: asyncio.Task | None = None
|
_live_snapshot_task: asyncio.Task | None = None
|
||||||
_push_poll_task: asyncio.Task | None = None
|
_push_poll_task: asyncio.Task | None = None
|
||||||
|
_readiness_task: asyncio.Task | None = None
|
||||||
_live_snapshot_value: dict | None = None
|
_live_snapshot_value: dict | None = None
|
||||||
_live_snapshot_created_at: float | None = None
|
_live_snapshot_created_at: float | None = None
|
||||||
LIVE_SNAPSHOT_SECTIONS = ("context", "events", "notifications")
|
LIVE_SNAPSHOT_SECTIONS = ("context", "events", "notifications")
|
||||||
|
|
@ -2358,43 +2428,35 @@ async def sign_out_all_devices(
|
||||||
|
|
||||||
@app.get("/readyz")
|
@app.get("/readyz")
|
||||||
async def readiness():
|
async def readiness():
|
||||||
"""Return readiness after verifying the configured Gitea connection."""
|
"""Return the latest readiness snapshot without contacting Gitea."""
|
||||||
try:
|
state = _readiness_state
|
||||||
user = await asyncio.wait_for(
|
if state is None or time.monotonic() - state.checked_at > READINESS_MAX_AGE_SECONDS:
|
||||||
current_user(), timeout=READINESS_TIMEOUT_SECONDS
|
|
||||||
)
|
|
||||||
if not isinstance(user, dict) or not user.get("login"):
|
|
||||||
raise ReadinessPayloadError(
|
|
||||||
"Gitea current-user response did not include a login"
|
|
||||||
)
|
|
||||||
except Exception as exc:
|
|
||||||
timed_out = isinstance(exc, TimeoutError)
|
|
||||||
error_message = (
|
|
||||||
f"Gitea readiness check timed out after {READINESS_TIMEOUT_SECONDS:g}s"
|
|
||||||
if timed_out
|
|
||||||
else (
|
|
||||||
str(exc)
|
|
||||||
if isinstance(exc, ReadinessPayloadError)
|
|
||||||
else "Gitea readiness check is temporarily unavailable"
|
|
||||||
)
|
|
||||||
)
|
|
||||||
return JSONResponse(
|
return JSONResponse(
|
||||||
{
|
{
|
||||||
"status": "not_ready",
|
"status": "not_ready",
|
||||||
"service": "stackchain-dashboard",
|
"service": "stackchain-dashboard",
|
||||||
"error": error_message,
|
"error": "Gitea readiness status is not available",
|
||||||
|
},
|
||||||
|
status_code=503,
|
||||||
|
)
|
||||||
|
if not state.ready:
|
||||||
|
return JSONResponse(
|
||||||
|
{
|
||||||
|
"status": "not_ready",
|
||||||
|
"service": "stackchain-dashboard",
|
||||||
|
"error": state.error,
|
||||||
},
|
},
|
||||||
status_code=503,
|
status_code=503,
|
||||||
headers={
|
headers={
|
||||||
"Retry-After": str(max(1, math.ceil(READINESS_TIMEOUT_SECONDS)))
|
"Retry-After": str(max(1, math.ceil(READINESS_TIMEOUT_SECONDS)))
|
||||||
} if timed_out else None,
|
} if state.timed_out else None,
|
||||||
)
|
)
|
||||||
payload = {
|
payload = {
|
||||||
"status": "ready",
|
"status": "ready",
|
||||||
"service": "stackchain-dashboard",
|
"service": "stackchain-dashboard",
|
||||||
}
|
}
|
||||||
if not dashboard_auth.enabled():
|
if not dashboard_auth.enabled():
|
||||||
payload["gitea_user"] = user["login"]
|
payload["gitea_user"] = state.login
|
||||||
return payload
|
return payload
|
||||||
|
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -2046,6 +2046,7 @@ async def test_public_routes_remain_available_and_readiness_hides_identity(acces
|
||||||
return {"id": 1, "login": "timmy"}
|
return {"id": 1, "login": "timmy"}
|
||||||
|
|
||||||
monkeypatch.setattr(main, "current_user", user)
|
monkeypatch.setattr(main, "current_user", user)
|
||||||
|
await main._check_readiness()
|
||||||
transport = httpx.ASGITransport(app=main.app)
|
transport = httpx.ASGITransport(app=main.app)
|
||||||
async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
|
async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
|
||||||
health = await client.get("/healthz")
|
health = await client.get("/healthz")
|
||||||
|
|
|
||||||
|
|
@ -1,4 +1,5 @@
|
||||||
import asyncio
|
import asyncio
|
||||||
|
from unittest.mock import AsyncMock
|
||||||
|
|
||||||
import httpx
|
import httpx
|
||||||
import pytest
|
import pytest
|
||||||
|
|
@ -245,6 +246,7 @@ async def test_application_lifespan_opens_and_closes_gitea_transport(monkeypatch
|
||||||
calls = []
|
calls = []
|
||||||
|
|
||||||
monkeypatch.setattr(gitea_proxy, "start_client", lambda: calls.append("start"))
|
monkeypatch.setattr(gitea_proxy, "start_client", lambda: calls.append("start"))
|
||||||
|
monkeypatch.setattr(main, "_check_readiness", AsyncMock())
|
||||||
|
|
||||||
async def stop_client():
|
async def stop_client():
|
||||||
calls.append("stop")
|
calls.append("stop")
|
||||||
|
|
@ -270,6 +272,7 @@ async def test_application_shutdown_finishes_snapshot_before_closing_transport(m
|
||||||
calls.append("snapshot cancelled")
|
calls.append("snapshot cancelled")
|
||||||
|
|
||||||
monkeypatch.setattr(gitea_proxy, "start_client", lambda: calls.append("start"))
|
monkeypatch.setattr(gitea_proxy, "start_client", lambda: calls.append("start"))
|
||||||
|
monkeypatch.setattr(main, "_check_readiness", AsyncMock())
|
||||||
|
|
||||||
async def stop_client():
|
async def stop_client():
|
||||||
assert main._live_snapshot_task is not None
|
assert main._live_snapshot_task is not None
|
||||||
|
|
@ -306,6 +309,7 @@ async def test_application_shutdown_cancels_available_work_scan_before_transport
|
||||||
calls.append("available scan cancelled")
|
calls.append("available scan cancelled")
|
||||||
|
|
||||||
monkeypatch.setattr(gitea_proxy, "start_client", lambda: calls.append("start"))
|
monkeypatch.setattr(gitea_proxy, "start_client", lambda: calls.append("start"))
|
||||||
|
monkeypatch.setattr(main, "_check_readiness", AsyncMock())
|
||||||
|
|
||||||
async def stop_client():
|
async def stop_client():
|
||||||
task = main._available_issue_snapshot_task
|
task = main._available_issue_snapshot_task
|
||||||
|
|
@ -341,6 +345,7 @@ async def test_application_shutdown_drains_authored_mutation_before_transport(mo
|
||||||
return {"id": 461}
|
return {"id": 461}
|
||||||
|
|
||||||
monkeypatch.setattr(gitea_proxy, "start_client", lambda: calls.append("start"))
|
monkeypatch.setattr(gitea_proxy, "start_client", lambda: calls.append("start"))
|
||||||
|
monkeypatch.setattr(main, "_check_readiness", AsyncMock())
|
||||||
|
|
||||||
async def stop_client():
|
async def stop_client():
|
||||||
calls.append("stop")
|
calls.append("stop")
|
||||||
|
|
@ -382,6 +387,7 @@ async def test_application_shutdown_bounds_and_settles_stalled_authored_mutation
|
||||||
|
|
||||||
monkeypatch.setattr(main, "AUTHORED_ACTION_SHUTDOWN_GRACE_SECONDS", 0.01)
|
monkeypatch.setattr(main, "AUTHORED_ACTION_SHUTDOWN_GRACE_SECONDS", 0.01)
|
||||||
monkeypatch.setattr(gitea_proxy, "start_client", lambda: calls.append("start"))
|
monkeypatch.setattr(gitea_proxy, "start_client", lambda: calls.append("start"))
|
||||||
|
monkeypatch.setattr(main, "_check_readiness", AsyncMock())
|
||||||
|
|
||||||
async def stop_client():
|
async def stop_client():
|
||||||
calls.append("stop")
|
calls.append("stop")
|
||||||
|
|
|
||||||
|
|
@ -11,6 +11,49 @@ def test_health_endpoint_reports_service_liveness_without_gitea_access():
|
||||||
assert main.health() == {"status": "ok", "service": "stackchain-dashboard"}
|
assert main.health() == {"status": "ok", "service": "stackchain-dashboard"}
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.anyio
|
||||||
|
async def test_readiness_endpoint_never_contacts_gitea_for_a_request(monkeypatch):
|
||||||
|
calls = 0
|
||||||
|
|
||||||
|
async def unexpected_user_lookup():
|
||||||
|
nonlocal calls
|
||||||
|
calls += 1
|
||||||
|
raise AssertionError("request-time readiness must not contact Gitea")
|
||||||
|
|
||||||
|
monkeypatch.setattr(main, "_readiness_state", None, raising=False)
|
||||||
|
monkeypatch.setattr(main, "current_user", unexpected_user_lookup)
|
||||||
|
|
||||||
|
first = await main.readiness()
|
||||||
|
second = await main.readiness()
|
||||||
|
|
||||||
|
assert first.status_code == 503
|
||||||
|
assert second.status_code == 503
|
||||||
|
assert calls == 0
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.anyio
|
||||||
|
async def test_readiness_monitor_refreshes_on_an_interval_and_cancels(monkeypatch):
|
||||||
|
calls = 0
|
||||||
|
refreshed_twice = asyncio.Event()
|
||||||
|
|
||||||
|
async def check():
|
||||||
|
nonlocal calls
|
||||||
|
calls += 1
|
||||||
|
if calls == 2:
|
||||||
|
refreshed_twice.set()
|
||||||
|
|
||||||
|
monkeypatch.setattr(main, "_check_readiness", check)
|
||||||
|
monkeypatch.setattr(main, "READINESS_INTERVAL_SECONDS", 0.01, raising=False)
|
||||||
|
|
||||||
|
task = asyncio.create_task(main._readiness_monitor())
|
||||||
|
await asyncio.wait_for(refreshed_twice.wait(), timeout=0.2)
|
||||||
|
task.cancel()
|
||||||
|
with pytest.raises(asyncio.CancelledError):
|
||||||
|
await task
|
||||||
|
|
||||||
|
assert calls == 2
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.anyio
|
@pytest.mark.anyio
|
||||||
async def test_readiness_endpoint_reports_connected_gitea_user(monkeypatch):
|
async def test_readiness_endpoint_reports_connected_gitea_user(monkeypatch):
|
||||||
async def connected_user():
|
async def connected_user():
|
||||||
|
|
@ -18,6 +61,8 @@ async def test_readiness_endpoint_reports_connected_gitea_user(monkeypatch):
|
||||||
|
|
||||||
monkeypatch.setattr(main, "current_user", connected_user)
|
monkeypatch.setattr(main, "current_user", connected_user)
|
||||||
|
|
||||||
|
await main._check_readiness()
|
||||||
|
|
||||||
response = await main.readiness()
|
response = await main.readiness()
|
||||||
|
|
||||||
assert any(getattr(route, "path", None) == "/readyz" for route in main.app.routes)
|
assert any(getattr(route, "path", None) == "/readyz" for route in main.app.routes)
|
||||||
|
|
@ -35,6 +80,8 @@ async def test_readiness_endpoint_hides_unexpected_upstream_error_details(monkey
|
||||||
|
|
||||||
monkeypatch.setattr(main, "current_user", unavailable_user)
|
monkeypatch.setattr(main, "current_user", unavailable_user)
|
||||||
|
|
||||||
|
await main._check_readiness()
|
||||||
|
|
||||||
response = await main.readiness()
|
response = await main.readiness()
|
||||||
|
|
||||||
assert response.status_code == 503
|
assert response.status_code == 503
|
||||||
|
|
@ -53,6 +100,8 @@ async def test_readiness_endpoint_returns_503_for_null_current_user_payload(monk
|
||||||
|
|
||||||
monkeypatch.setattr(main, "current_user", null_user)
|
monkeypatch.setattr(main, "current_user", null_user)
|
||||||
|
|
||||||
|
await main._check_readiness()
|
||||||
|
|
||||||
response = await main.readiness()
|
response = await main.readiness()
|
||||||
|
|
||||||
assert response.status_code == 503
|
assert response.status_code == 503
|
||||||
|
|
@ -77,6 +126,8 @@ async def test_readiness_endpoint_times_out_and_cancels_stalled_gitea_check(monk
|
||||||
monkeypatch.setattr(main, "READINESS_TIMEOUT_SECONDS", 0.01, raising=False)
|
monkeypatch.setattr(main, "READINESS_TIMEOUT_SECONDS", 0.01, raising=False)
|
||||||
monkeypatch.setattr(main, "current_user", hanging_user)
|
monkeypatch.setattr(main, "current_user", hanging_user)
|
||||||
|
|
||||||
|
await main._check_readiness()
|
||||||
|
|
||||||
response = await main.readiness()
|
response = await main.readiness()
|
||||||
|
|
||||||
assert response.status_code == 503
|
assert response.status_code == 503
|
||||||
|
|
@ -87,3 +138,18 @@ async def test_readiness_endpoint_times_out_and_cancels_stalled_gitea_check(monk
|
||||||
"error": "Gitea readiness check timed out after 0.01s",
|
"error": "Gitea readiness check timed out after 0.01s",
|
||||||
}
|
}
|
||||||
assert cancelled.is_set()
|
assert cancelled.is_set()
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.anyio
|
||||||
|
async def test_readiness_endpoint_expires_a_stale_healthy_snapshot(monkeypatch):
|
||||||
|
monkeypatch.setattr(
|
||||||
|
main,
|
||||||
|
"_readiness_state",
|
||||||
|
main.ReadinessState(checked_at=10.0, ready=True, login="timmy"),
|
||||||
|
)
|
||||||
|
monkeypatch.setattr(main.time, "monotonic", lambda: 76.0)
|
||||||
|
|
||||||
|
response = await main.readiness()
|
||||||
|
|
||||||
|
assert response.status_code == 503
|
||||||
|
assert json.loads(response.body)["error"] == "Gitea readiness status is not available"
|
||||||
|
|
|
||||||
Loading…
Reference in New Issue
Block a user