Worker healthcheck

.pre-commit-config.yaml

@@ -77,7 +77,7 @@ repos:
     hooks:
       - id: uv-lock
   - repo: https://github.com/astral-sh/ruff-pre-commit
-    rev: v0.14.11
+    rev: v0.16.0
     hooks:
       - id: ruff-check
         types_or: [python, pyi]

docker/entrypoint.sh

@@ -12,8 +12,9 @@ fi
 
 set -e
 if [ "$#" -eq 0 ]; then
+  export WEB_CONCURRENCY="${WEB_CONCURRENCY:-4}"
   # FIXME: can we know the traefik/nginx internal docker ip easily ?
-  exec gunicorn "matrixrmapi.app:get_app()" --bind 0.0.0.0:8012 --forwarded-allow-ips='*' -w 4 -k uvicorn.workers.UvicornWorker
+  exec gunicorn "matrixrmapi.app:get_app()" --bind 0.0.0.0:8012 --forwarded-allow-ips='*' -w "$WEB_CONCURRENCY" -k uvicorn.workers.UvicornWorker
 else
   exec "$@"
 fi

src/matrixrmapi/api/__init__.py

@@ -2,17 +2,15 @@
 
 from fastapi.routing import APIRouter
 
-from .usercrud import router as usercrud_router
-from .clientinfo import router as clientinfo_router
 from .admininfo import router as admininfo_router
-from .healthcheck import router as healthcheck_router
+from .clientinfo import router as clientinfo_router
 from .description import router as description_router
-from .instructions import router as instructions_router
-
 from .description import router_v2 as description_router_v2
+from .healthcheck import router as healthcheck_router
+from .instructions import router as instructions_router
+from .usercrud import router as usercrud_router
 from .userinfo import router as userinfo_router
 
-
 all_routers = APIRouter()
 all_routers.include_router(usercrud_router, prefix="/users", tags=["users"])
 all_routers.include_router(clientinfo_router, prefix="/clients", tags=["clients"])

src/matrixrmapi/api/admininfo.py

@@ -3,9 +3,9 @@
 import logging
 
 from fastapi import APIRouter, Depends
+from jinja2 import Environment, FileSystemLoader
 from libpvarki.middleware import MTLSHeader
 from libpvarki.schemas.product import UserInstructionFragment
-from jinja2 import Environment, FileSystemLoader
 
 from ..config import TEMPLATES_PATH
 

src/matrixrmapi/api/clientinfo.py

@@ -1,10 +1,9 @@
 """Endpoints for information for the end-user"""
 
-from typing import List, Dict
-import logging
+import base64
 import io
+import logging
 import zipfile
-import base64
 
 from fastapi import APIRouter, Depends
 from libpvarki.middleware import MTLSHeader
@@ -24,7 +23,7 @@ def zip_pem(pem: str, filename: str) -> bytes:
 
 
 @router.post("/fragment", deprecated=True)
-async def client_instruction_fragment(user: UserCRUDRequest) -> List[Dict[str, str]]:
+async def client_instruction_fragment(user: UserCRUDRequest) -> list[dict[str, str]]:
     """Return user instructions, we use POST because the integration layer might not keep
     track of callsigns and certs by UUID and will probably need both for the instructions"""
     zip1_bytes = zip_pem(user.x509cert, f"{user.callsign}_1.pem")

src/matrixrmapi/api/description.py

@@ -1,12 +1,11 @@
 """Descriptions API"""
 
-from typing import Literal, Optional
 import logging
+from typing import Literal
 
 from fastapi import APIRouter, HTTPException
-from pydantic import BaseModel, Field, Extra
 from libpvarki.schemas.product import ProductDescription
-
+from pydantic import BaseModel, Extra, Field
 
 LOGGER = logging.getLogger(__name__)
 
@@ -31,7 +30,7 @@ class ProductDescriptionExtended(BaseModel):
         description="Short name for the product, used as slug/key in dicts and urls"
     )
     title: str = Field(description="Fancy name for the product")
-    icon: Optional[str] = Field(description="URL for icon")
+    icon: str | None = Field(description="URL for icon")
     description: str = Field(description="Short-ish description of the product")
     language: str = Field(description="Language of this response")
     docs: str = Field(description="Link to documentation")
@@ -49,7 +48,7 @@ class Config:
 )
 async def return_product_description(language: str) -> ProductDescription:
     """Fetch description from each product in manifest"""
-    LOGGER.debug("Got language: {}".format(language))
+    LOGGER.debug(f"Got language: {language}")
     if language == "fi":
         return ProductDescription(
             shortname=PRODUCT_SHORTNAME,

src/matrixrmapi/api/healthcheck.py

@@ -6,7 +6,8 @@
 from fastapi import APIRouter, Request
 from libpvarki.schemas.product import ProductHealthCheckResponse
 
-from ..config import MAS_HEALTH_URL, SYNAPSE_URL
+from ..config import MAS_HEALTH_URL, SYNAPSE_URL, WEB_CONCURRENCY
+from ..utils.startup import ready_workers
 
 LOGGER = logging.getLogger(__name__)
 
@@ -15,13 +16,21 @@
 
 @router.get("")
 async def request_healthcheck(request: Request) -> ProductHealthCheckResponse:
-    """Check that the Matrix integration is initialised and Synapse and MAS respond"""
+    """Check that the Matrix integration is initialised and Synapse and MAS respond
+
+    Ensures all workers are ready to handle UserCRUD (mas + synapse credentials )
+    """
     if getattr(request.app.state, "synapse", None) is None or not getattr(
         request.app.state, "rooms", None
     ):
         return ProductHealthCheckResponse(
             healthy=False, extra="Matrix integration not initialised"
         )
+    ready = ready_workers()
+    if ready < WEB_CONCURRENCY:
+        return ProductHealthCheckResponse(
+            healthy=False, extra=f"only {ready}/{WEB_CONCURRENCY} workers initialised"
+        )
     async with httpx.AsyncClient() as client:
         for name, url in (("Synapse", SYNAPSE_URL), ("MAS", MAS_HEALTH_URL)):
             try:

src/matrixrmapi/api/instructions.py

@@ -1,6 +1,5 @@
 """Instructions endpoints"""
 
-from typing import Dict
 import logging
 
 from fastapi import APIRouter, Depends
@@ -13,7 +12,7 @@
 
 
 @router.post("/{language}")
-async def user_intructions(user: UserCRUDRequest) -> Dict[str, str]:
+async def user_intructions(user: UserCRUDRequest) -> dict[str, str]:
     """return user instructions"""
     return {
         "callsign": user.callsign,

src/matrixrmapi/api/usercrud.py

@@ -3,7 +3,6 @@
 from __future__ import annotations
 
 import logging
-from typing import Dict, Optional
 
 import httpx
 from fastapi import APIRouter, Depends, HTTPException, Request
@@ -12,9 +11,9 @@
 from libpvarki.schemas.product import UserCRUDRequest
 
 from ..config import get_manifest, get_server_domain
+from ..types import AdminAction
 from ..utils.mas_admin import MasAdmin
 from ..utils.synapse_admin import SynapseAdmin, matrix_user_id
-from ..types import AdminAction
 
 LOGGER = logging.getLogger(__name__)
 
@@ -29,25 +28,25 @@ def comes_from_rm(request: Request) -> None:
         raise HTTPException(status_code=403)
 
 
-def get_synapse(request: Request) -> Optional[SynapseAdmin]:
+def get_synapse(request: Request) -> SynapseAdmin | None:
     """Return SynapseAdmin from app state, or None if not yet ready."""
-    val: Optional[SynapseAdmin] = getattr(request.app.state, "synapse", None)
+    val: SynapseAdmin | None = getattr(request.app.state, "synapse", None)
     return val
 
 
-def get_mas(request: Request) -> Optional[MasAdmin]:
+def get_mas(request: Request) -> MasAdmin | None:
     """Return MasAdmin from app state, or None if not yet ready."""
-    val: Optional[MasAdmin] = getattr(request.app.state, "mas", None)
+    val: MasAdmin | None = getattr(request.app.state, "mas", None)
     return val
 
 
-def get_rooms(request: Request) -> Optional[Dict[str, str]]:
+def get_rooms(request: Request) -> dict[str, str] | None:
     """Return room IDs dict from app state, or None if not yet ready."""
-    val: Optional[Dict[str, str]] = getattr(request.app.state, "rooms", None)
+    val: dict[str, str] | None = getattr(request.app.state, "rooms", None)
     return val
 
 
-def public_room_ids(rooms: Dict[str, str]) -> list[str]:
+def public_room_ids(rooms: dict[str, str]) -> list[str]:
     """Room IDs for the space + the three public rooms (not admin channel)."""
     return [
         rooms[k] for k in ("space", "general", "helpdesk", "offtopic") if k in rooms
@@ -132,7 +131,7 @@ async def apply_admin_action(
                 await synapse.force_join(admin_id, uid)
             else:
                 await synapse.kick(admin_id, uid)
-    except Exception as exc:
+    except Exception as exc:  # noqa: BLE001
         LOGGER.error("Failed to %s %s: %s", action.value, uid, exc)
         return OperationResultResponse(success=False)
     LOGGER.info("%sd %s (power level %d)", action.value.capitalize(), uid, level)

src/matrixrmapi/app.py

@@ -5,19 +5,20 @@
 import asyncio
 import contextlib
 import logging
+from collections.abc import AsyncGenerator
 from contextlib import asynccontextmanager
-from typing import AsyncGenerator, Optional
 
 from fastapi import FastAPI
 from fastapi.middleware.cors import CORSMiddleware
 from libpvarki.logging import init_logging
 
 from matrixrmapi import __version__
-from .config import LOG_LEVEL, get_manifest
+
 from .api import all_routers, all_routers_v2
+from .config import LOG_LEVEL, get_manifest
 from .utils.mas_admin import MasAdmin
-from .utils.synapse_admin import SynapseAdmin
 from .utils.startup import connect_to_matrix
+from .utils.synapse_admin import SynapseAdmin
 
 LOGGER = logging.getLogger(__name__)
 
@@ -32,10 +33,10 @@ async def app_lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
         task.cancel()
         with contextlib.suppress(asyncio.CancelledError):
             await task
-        synapse: Optional[SynapseAdmin] = getattr(app.state, "synapse", None)
+        synapse: SynapseAdmin | None = getattr(app.state, "synapse", None)
         if synapse:
             await synapse.close()
-        mas: Optional[MasAdmin] = getattr(app.state, "mas", None)
+        mas: MasAdmin | None = getattr(app.state, "mas", None)
         if mas:
             await mas.close()
 

src/matrixrmapi/config.py

@@ -1,10 +1,10 @@
 """Configurations with .env support"""
 
-from typing import Dict, Any, cast
-from pathlib import Path
-import json
 import functools
+import json
 import logging
+from pathlib import Path
+from typing import Any, cast
 
 from starlette.config import Config
 
@@ -26,10 +26,16 @@
 MAS_ADMIN_CLIENT_ID: str = cfg("MAS_ADMIN_CLIENT_ID", default="")
 MAS_ADMIN_CLIENT_SECRET: str = cfg("MAS_ADMIN_CLIENT_SECRET", default="")
 SYNAPSE_BOT_USERNAME: str = cfg("SYNAPSE_BOT_USERNAME", default="matrixrmapi-bot")
+WEB_CONCURRENCY: int = cfg("WEB_CONCURRENCY", default=1, cast=int)
+
+INIT_MAX_ATTEMPTS: int = cfg("INIT_MAX_ATTEMPTS", default=5, cast=int)
+INIT_RETRY_WAIT: float = cfg("INIT_RETRY_WAIT", default=5.0, cast=float)
+SERVICE_WAIT_RETRIES: int = cfg("SERVICE_WAIT_RETRIES", default=12, cast=int)
+SERVICE_WAIT_INTERVAL: float = cfg("SERVICE_WAIT_INTERVAL", default=5.0, cast=float)
 
 
 @functools.cache
-def get_manifest() -> Dict[str, Any]:
+def get_manifest() -> dict[str, Any]:
     """Get manifest contents"""
     pth = Path("/pvarki/kraftwerk-init.json")
     if not pth.exists():
@@ -50,7 +56,7 @@ def get_manifest() -> Dict[str, Any]:
             },
         }
     data = json.loads(pth.read_text(encoding="utf-8"))
-    return cast(Dict[str, Any], data)
+    return cast(dict[str, Any], data)
 
 
 def get_server_domain() -> str:

src/matrixrmapi/console.py

@@ -1,16 +1,15 @@
 """CLI entrypoints for matrix product integration api"""
 
 import asyncio
-import logging
 import json
+import logging
 
+import aiohttp
 import click
 from libadvian.logging import init_logging
-import aiohttp
 
 from matrixrmapi import __version__
 
-
 LOGGER = logging.getLogger(__name__)
 
 
@@ -54,16 +53,18 @@ async def doit() -> int:
         nonlocal host, port, timeout
         if "://" not in host:
             host = f"http://{host}"
-        async with aiohttp.ClientSession(
-            timeout=aiohttp.ClientTimeout(total=timeout)
-        ) as session:
-            async with session.get(f"{host}:{port}/api/v1/healthcheck") as resp:
-                if resp.status != 200:
-                    return resp.status
-                payload = await resp.json()
-                click.echo(json.dumps(payload))
-                if not payload["healthy"]:
-                    return 1
+        async with (
+            aiohttp.ClientSession(
+                timeout=aiohttp.ClientTimeout(total=timeout)
+            ) as session,
+            session.get(f"{host}:{port}/api/v1/healthcheck") as resp,
+        ):
+            if resp.status != 200:
+                return resp.status
+            payload = await resp.json()
+            click.echo(json.dumps(payload))
+            if not payload["healthy"]:
+                return 1
         return 0
 
     ctx.exit(asyncio.get_event_loop().run_until_complete(doit()))

src/matrixrmapi/types.py

@@ -1,7 +1,6 @@
 """Shared domain types and constants."""
 
 from enum import Enum
-from typing import Dict
 
 
 class AdminAction(Enum):
@@ -13,7 +12,7 @@ class AdminAction(Enum):
 
 # Call-related event types that regular users (power level 0) must be allowed to send.
 # Covers both legacy 1:1 calls and MSC3401 group calls (Element Call).
-CALL_EVENTS_DEFAULT_LEVEL: Dict[str, int] = {
+CALL_EVENTS_DEFAULT_LEVEL: dict[str, int] = {
     "m.call.invite": 0,
     "m.call.answer": 0,
     "m.call.hangup": 0,