Drop APScheduler, convert to native asyncio

Author: suecharo

docs/architecture.md

@@ -5,16 +5,38 @@
 The sapporo-service is a FastAPI application that accepts WES API requests, prepares a run directory for each workflow execution, and delegates the actual workflow engine invocation to a shell script (`run.sh`). Each workflow engine runs inside its own Docker container, spawned as a sibling container via the host's Docker socket. See [Installation - Volume Mounts](installation.md#volume-mounts-docker-in-docker) for details on the DinD volume mount requirements.
 
 ```text
-+--------+     +---------+     +--------+     +---------------------+
-| Client | --> | FastAPI | --> | run.py | --> | run.sh (subprocess) |
-+--------+     +---------+     +--------+     +---------------------+
-                                                   |
-                                              docker run
-                                                   |
-                                             +------------+
-                                             | Engine     |
-                                             | Container  |
-                                             +------------+
+         +-----------+
+         |  Client   |
+         +-----------+
+               |
+          HTTP request
+               |
+               v
+         +-----------+
+         |  FastAPI  |
+         +-----------+
+               |
+         Python call
+               |
+               v
+         +-----------+
+         |  run.py   |
+         +-----------+
+               |
+          subprocess
+               |
+               v
+         +-----------+
+         |  run.sh   |
+         +-----------+
+               |
+          docker run
+               |
+               v
+         +-----------+
+         |  Engine   |
+         | Container |
+         +-----------+
 ```
 
 The Python side (`run.py`) never calls a workflow engine directly. It prepares the run directory, writes all input files, then forks `run.sh` as a subprocess. All run data is persisted to the filesystem, with a SQLite index for fast listing.
@@ -114,41 +136,23 @@ runs/
 | `system_logs.json` | System-level logs |
 | `workflow_engine_params.txt` | Engine-specific parameters |
 
-## Orphan Recovery
+## Reconciliation
 
-When the sapporo process restarts (e.g., container recreation), any `run.sh` subprocesses from the previous instance are dead. Runs that were in a non-terminal state are now orphans — their `state.txt` still says `RUNNING` or `QUEUED`, but no process is driving them forward.
+Detects runs stuck in `RUNNING`/`QUEUED` after a process restart and marks them as `SYSTEM_ERROR`.
 
-At startup, **before** the SQLite index is built, `recover_orphaned_runs()` scans all run directories and transitions orphaned runs to `SYSTEM_ERROR`.
+`reconcile_runs()` runs at startup (before `init_db()`) and periodically in the background (at the snapshot interval, default: 30 minutes). For each run in a non-terminal state, it reads `run.pid` and checks process liveness via `os.kill(pid, 0)`:
 
-### Target States
+| PID file | Process alive | Action |
+|---|---|---|
+| Present | Yes | Skip (running normally) |
+| Present | No | Set `SYSTEM_ERROR` (reason: "process vanished") |
+| Absent | N/A | Set `SYSTEM_ERROR` (reason: "no pid file") |
 
-Runs in the following non-terminal states are recovered:
-
-- `INITIALIZING`
-- `QUEUED`
-- `RUNNING`
-- `PAUSED`
-- `PREEMPTED`
-- `CANCELING`
-- `DELETING`
-
-Runs in terminal states (`COMPLETE`, `EXECUTOR_ERROR`, `SYSTEM_ERROR`, `CANCELED`, `DELETED`) and `UNKNOWN` are left unchanged.
-
-### Recovery Actions
-
-For each orphaned run, the recovery process:
-
-1. Sets `state.txt` to `SYSTEM_ERROR`
-2. Writes the current timestamp to `end_time.txt`
-3. Appends a descriptive message to `system_logs.json`
-
-### Ordering
-
-`recover_orphaned_runs()` runs before `init_db()` in the application lifespan, so the SQLite index reflects the corrected states from its first build.
+Runs in terminal states (`COMPLETE`, `EXECUTOR_ERROR`, `SYSTEM_ERROR`, `CANCELED`, `DELETED`) and `UNKNOWN` are skipped. For each reconciled run, `state.txt` is set to `SYSTEM_ERROR`, the current timestamp is written to `end_time.txt`, and the reason is logged to `system_logs.json`.
 
 ## SQLite Index
 
-The SQLite database (`sapporo.db`) is an **index**, not a data store. It is rebuilt at a configurable interval (default: 30 minutes) by scanning the run directories and can be deleted at any time without data loss. It exists solely to make `GET /runs` (list all runs) fast. Individual run queries (`GET /runs/{run_id}`) always read from the filesystem.
+The SQLite database (`sapporo.db`) is an **index**, not a data store. It is rebuilt at a configurable interval (default: 30 minutes) by a background asyncio task that scans the run directories, and can be deleted at any time without data loss. It exists solely to make `GET /runs` (list all runs) fast. Individual run queries (`GET /runs/{run_id}`) always read from the filesystem.
 
 ## RO-Crate
 

pyproject.toml

@@ -26,7 +26,6 @@ classifiers = [
     "Topic :: Scientific/Engineering :: Bio-Informatics",
 ]
 dependencies = [
-    "apscheduler>=3,<4",
     "argon2-cffi>=21,<26",
     "cachetools>=5,<8",
     "fastapi>=0.100,<1",
@@ -53,6 +52,7 @@ tests = [
     "hypothesis",
     "mutmut",
     "mypy",
+    "pytest-asyncio",
     "pytest-mock",
     "pytest-randomly",
     "pytest",
@@ -79,6 +79,7 @@ packages = ["sapporo"]
 
 [tool.pytest.ini_options]
 addopts = "-v --tb=short --strict-markers"
+asyncio_mode = "strict"
 testpaths = ["tests/unit"]
 markers = [
     "slow: marks tests as slow (deselect with '-m \"not slow\"')",

sapporo/app.py

@@ -1,10 +1,10 @@
+import asyncio
 import logging
 import logging.config
 from collections.abc import AsyncGenerator
 from contextlib import asynccontextmanager
 
 import uvicorn
-from apscheduler.schedulers.background import BackgroundScheduler
 from fastapi import FastAPI, Request
 from fastapi.exceptions import RequestValidationError
 from fastapi.middleware.cors import CORSMiddleware
@@ -18,7 +18,7 @@
 from sapporo.database import init_db
 from sapporo.factory import create_executable_wfs, create_service_info
 from sapporo.routers import router
-from sapporo.run import recover_orphaned_runs, remove_old_runs
+from sapporo.run import reconcile_runs, remove_old_runs
 from sapporo.schemas import ErrorResponse
 from sapporo.utils import mask_sensitive
 
@@ -158,21 +158,34 @@ def init_app_state() -> None:
 
 @asynccontextmanager
 async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]:
-    recover_orphaned_runs()
+    reconcile_runs()
     init_db()
 
     snapshot_interval = get_config().snapshot_interval
-    scheduler = BackgroundScheduler()
-    scheduler.add_job(init_db, "interval", minutes=snapshot_interval)
-    scheduler.add_job(remove_old_runs, "interval", minutes=snapshot_interval)
-    scheduler.start()
-    LOGGER.info("DB snapshot scheduler started")
+    tasks: set[asyncio.Task[None]] = set()
+
+    async def _reconciliation_loop() -> None:
+        while True:
+            await asyncio.sleep(snapshot_interval * 60)
+            await asyncio.to_thread(reconcile_runs)
+
+    async def _db_sync_loop() -> None:
+        while True:
+            await asyncio.sleep(snapshot_interval * 60)
+            await asyncio.to_thread(init_db)
+            await asyncio.to_thread(remove_old_runs)
+
+    tasks.add(asyncio.create_task(_reconciliation_loop()))
+    tasks.add(asyncio.create_task(_db_sync_loop()))
+    LOGGER.info("Background loops started (interval=%d min)", snapshot_interval)
 
     try:
         yield
     finally:
-        scheduler.shutdown()
-        LOGGER.info("DB snapshot scheduler stopped")
+        for task in tasks:
+            task.cancel()
+        await asyncio.gather(*tasks, return_exceptions=True)
+        LOGGER.info("Background loops stopped")
 
 
 def create_app() -> FastAPI:

sapporo/auth.py

@@ -4,7 +4,6 @@
 import logging
 import os
 import re
-import time
 from functools import cache
 from threading import RLock
 from typing import Any, Literal
@@ -182,13 +181,24 @@ async def create_access_token(username: str, password: str) -> str:
     return await external_create_access_token(username, password)
 
 
-def decode_token(token: str) -> TokenPayload:
+async def decode_token(token: str) -> TokenPayload:
     auth_config = get_auth_config()
     if auth_config.idp_provider == "sapporo":
         payload = spr_decode_token(token)
         check_valid_username(payload.sub)
+
         return payload
-    return external_decode_token(token)
+
+    return await external_decode_token(token)
+
+
+async def resolve_username(token: str | None) -> str | None:
+    """Extract username from a token, or return None if no token."""
+    if token is None:
+        return None
+    payload = await decode_token(token)
+
+    return extract_username(payload)
 
 
 def sanitize_username(username: str) -> str:
@@ -365,38 +375,39 @@ def _validate_https_url(url: str, context: str) -> None:
         )
 
 
-def _fetch_once(url: str) -> httpx.Response:
+async def _fetch_once(url: str) -> httpx.Response:
     """Execute a single HTTP GET request with timeout."""
-    with httpx.Client(timeout=HTTPX_TIMEOUT) as client:
-        res = client.get(url, follow_redirects=True, headers={"User-Agent": user_agent()})
+    async with httpx.AsyncClient(timeout=HTTPX_TIMEOUT) as client:
+        res = await client.get(url, follow_redirects=True, headers={"User-Agent": user_agent()})
         res.raise_for_status()
+
         return res
 
 
-def _fetch_with_retry(url: str, context: str) -> httpx.Response:
+async def _fetch_with_retry(url: str, context: str) -> httpx.Response:
     """Fetch a URL with retry and exponential backoff."""
     last_exc: Exception | None = None
     for attempt in range(_MAX_RETRIES):
         try:
-            return _fetch_once(url)
+            return await _fetch_once(url)
         except httpx.HTTPError as exc:  # noqa: PERF203
             last_exc = exc
             LOGGER.warning("Attempt %d/%d to fetch %s failed: %s", attempt + 1, _MAX_RETRIES, context, exc)
             if attempt < _MAX_RETRIES - 1:
-                time.sleep(_RETRY_BASE_DELAY * (2**attempt))
+                await asyncio.sleep(_RETRY_BASE_DELAY * (2**attempt))
     msg = f"Failed to fetch {context} after {_MAX_RETRIES} retries: {last_exc}"
     raise HTTPException(status_code=500, detail=msg)
 
 
-def _fetch_endpoint_metadata_sync() -> ExternalEndpointMetadata:
-    """Fetch endpoint metadata synchronously."""
+async def _fetch_endpoint_metadata_impl() -> ExternalEndpointMetadata:
+    """Fetch endpoint metadata from the IdP."""
     auth_config = get_auth_config()
     idp_url = auth_config.external_config.idp_url
     _validate_https_url(idp_url, "External IdP URL")
 
     well_known_url = f"{idp_url}/.well-known/openid-configuration"
     try:
-        res = _fetch_with_retry(well_known_url, "IdP metadata")
+        res = await _fetch_with_retry(well_known_url, "IdP metadata")
         metadata = ExternalEndpointMetadata.model_validate(res.json())
 
         _validate_https_url(metadata.token_endpoint, "Token endpoint")
@@ -407,21 +418,22 @@ def _fetch_endpoint_metadata_sync() -> ExternalEndpointMetadata:
     else:
         with _cache_lock:
             _metadata_cache["metadata"] = metadata
+
         return metadata
 
 
-def fetch_endpoint_metadata() -> ExternalEndpointMetadata:
+async def fetch_endpoint_metadata() -> ExternalEndpointMetadata:
     with _cache_lock:
         cached: ExternalEndpointMetadata | None = _metadata_cache.get("metadata")
         if cached is not None:
             return cached
 
-    return _fetch_endpoint_metadata_sync()
+    return await _fetch_endpoint_metadata_impl()
 
 
 async def external_create_access_token(username: str, password: str) -> str:
     auth_config = get_auth_config()
-    metadata = await asyncio.to_thread(fetch_endpoint_metadata)
+    metadata = await fetch_endpoint_metadata()
     token_url = metadata.token_endpoint
     data: dict[str, str | None] = {
         "grant_type": "password",
@@ -444,7 +456,7 @@ async def external_create_access_token(username: str, password: str) -> str:
         raise_invalid_credentials()
 
 
-def external_decode_token(token: str) -> TokenPayload:
+async def external_decode_token(token: str) -> TokenPayload:
     try:
         unverified_header = jwt.get_unverified_header(token)
     except jwt.exceptions.DecodeError:
@@ -458,13 +470,13 @@ def external_decode_token(token: str) -> TokenPayload:
     if alg not in ALLOWED_ALGORITHMS:
         raise_invalid_token()
 
-    metadata = fetch_endpoint_metadata()
-    jwks = fetch_jwks()
+    metadata = await fetch_endpoint_metadata()
+    jwks = await fetch_jwks()
 
     jwk_key = next((k.key for k in jwks.keys if k.key_id == kid), None)
     if jwk_key is None:
         # Key rotation: re-fetch JWKS and try again
-        jwks = fetch_jwks(force_refresh=True)
+        jwks = await fetch_jwks(force_refresh=True)
         jwk_key = next((k.key for k in jwks.keys if k.key_id == kid), None)
         if jwk_key is None:
             raise_invalid_token()
@@ -486,25 +498,27 @@ def external_decode_token(token: str) -> TokenPayload:
         raise_invalid_token()
 
 
-def _fetch_jwks_sync(force_refresh: bool = False) -> PyJWKSet:
-    """Fetch JWKS synchronously."""
-    jwks_uri = fetch_endpoint_metadata().jwks_uri
+async def _fetch_jwks_impl(force_refresh: bool = False) -> PyJWKSet:
+    """Fetch JWKS from the IdP."""
+    metadata = await fetch_endpoint_metadata()
+    jwks_uri = metadata.jwks_uri
     try:
-        res = _fetch_with_retry(jwks_uri, "JWKS")
+        res = await _fetch_with_retry(jwks_uri, "JWKS")
         jwk_set = PyJWKSet.from_dict(res.json())
     except (ValueError, KeyError):
         raise_internal_error("Failed to fetch JWKS from the IdP")
     else:
         with _cache_lock:
             _jwks_cache["jwks"] = jwk_set
+
         return jwk_set
 
 
-def fetch_jwks(force_refresh: bool = False) -> PyJWKSet:
+async def fetch_jwks(force_refresh: bool = False) -> PyJWKSet:
     if not force_refresh:
         with _cache_lock:
             cached: PyJWKSet | None = _jwks_cache.get("jwks")
             if cached is not None:
                 return cached
 
-    return _fetch_jwks_sync(force_refresh)
+    return await _fetch_jwks_impl(force_refresh)