feat(edge): add offline read cache for allowlisted State Hub GET routes
The edge relay now persists successful GET responses and serves them with stale markers when upstream is unreachable. Extend Forgejo image workflow path filters so api changes trigger registry publishes.
This commit is contained in:
parent
138299293d
commit
1cf949bda4
5 changed files with 354 additions and 1 deletions
|
|
@ -12,6 +12,9 @@ on:
|
|||
paths:
|
||||
- ".forgejo/workflows/image.yaml"
|
||||
- "Dockerfile"
|
||||
- "api/**"
|
||||
- "pyproject.toml"
|
||||
- "uv.lock"
|
||||
workflow_dispatch:
|
||||
|
||||
env:
|
||||
|
|
|
|||
152
api/edge/read_cache.py
Normal file
152
api/edge/read_cache.py
Normal file
|
|
@ -0,0 +1,152 @@
|
|||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import sqlite3
|
||||
import stat
|
||||
from dataclasses import dataclass
|
||||
from datetime import datetime, timezone
|
||||
from pathlib import Path
|
||||
from urllib.parse import urlencode
|
||||
|
||||
DEFAULT_READ_CACHE_PATH = Path(
|
||||
os.environ.get("STATEHUB_READ_CACHE_PATH", "~/.statehub/edge-read-cache.sqlite3")
|
||||
).expanduser()
|
||||
MAX_CACHE_ENTRY_BYTES = 1024 * 1024
|
||||
|
||||
CACHEABLE_GET_PREFIXES = (
|
||||
"/state/",
|
||||
"/workplans/",
|
||||
"/messages/",
|
||||
"/decisions/",
|
||||
"/tasks/",
|
||||
"/progress/",
|
||||
"/sbom/",
|
||||
"/repos",
|
||||
"/legacy-meter/",
|
||||
)
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class ReadCacheEntry:
|
||||
cache_key: str
|
||||
status_code: int
|
||||
content: bytes
|
||||
content_type: str | None
|
||||
cached_at: str
|
||||
|
||||
|
||||
def default_read_cache_path() -> Path:
|
||||
return DEFAULT_READ_CACHE_PATH
|
||||
|
||||
|
||||
def utcnow() -> str:
|
||||
return datetime.now(tz=timezone.utc).isoformat()
|
||||
|
||||
|
||||
def build_cache_key(method: str, path: str, query_items: list[tuple[str, str]]) -> str:
|
||||
normalized_path = path if path.startswith("/") else f"/{path}"
|
||||
query = urlencode(sorted(query_items))
|
||||
if query:
|
||||
return f"{method.upper()}:{normalized_path}?{query}"
|
||||
return f"{method.upper()}:{normalized_path}"
|
||||
|
||||
|
||||
def is_cacheable_get(path: str) -> bool:
|
||||
normalized = path if path.startswith("/") else f"/{path}"
|
||||
if normalized.startswith("/edge/"):
|
||||
return False
|
||||
return any(normalized.startswith(prefix) for prefix in CACHEABLE_GET_PREFIXES)
|
||||
|
||||
|
||||
class ReadCacheStore:
|
||||
def __init__(self, path: str | Path | None = None) -> None:
|
||||
self.path = Path(path).expanduser() if path is not None else default_read_cache_path()
|
||||
self.path.parent.mkdir(parents=True, exist_ok=True)
|
||||
self._init_db()
|
||||
self._chmod_private()
|
||||
|
||||
def _connect(self) -> sqlite3.Connection:
|
||||
conn = sqlite3.connect(self.path)
|
||||
conn.row_factory = sqlite3.Row
|
||||
return conn
|
||||
|
||||
def _init_db(self) -> None:
|
||||
with self._connect() as conn:
|
||||
conn.execute(
|
||||
"""
|
||||
CREATE TABLE IF NOT EXISTS read_cache_entries (
|
||||
cache_key TEXT PRIMARY KEY,
|
||||
status_code INTEGER NOT NULL,
|
||||
content BLOB NOT NULL,
|
||||
content_type TEXT,
|
||||
cached_at TEXT NOT NULL
|
||||
)
|
||||
"""
|
||||
)
|
||||
conn.execute(
|
||||
"CREATE INDEX IF NOT EXISTS ix_read_cache_cached_at ON read_cache_entries(cached_at)"
|
||||
)
|
||||
conn.commit()
|
||||
|
||||
def _chmod_private(self) -> None:
|
||||
try:
|
||||
os.chmod(self.path, stat.S_IRUSR | stat.S_IWUSR)
|
||||
except OSError:
|
||||
pass
|
||||
|
||||
def put(
|
||||
self,
|
||||
*,
|
||||
cache_key: str,
|
||||
status_code: int,
|
||||
content: bytes,
|
||||
content_type: str | None,
|
||||
) -> None:
|
||||
if len(content) > MAX_CACHE_ENTRY_BYTES:
|
||||
return
|
||||
now = utcnow()
|
||||
with self._connect() as conn:
|
||||
conn.execute(
|
||||
"""
|
||||
INSERT INTO read_cache_entries (
|
||||
cache_key, status_code, content, content_type, cached_at
|
||||
) VALUES (?, ?, ?, ?, ?)
|
||||
ON CONFLICT(cache_key) DO UPDATE SET
|
||||
status_code = excluded.status_code,
|
||||
content = excluded.content,
|
||||
content_type = excluded.content_type,
|
||||
cached_at = excluded.cached_at
|
||||
""",
|
||||
(cache_key, status_code, content, content_type, now),
|
||||
)
|
||||
conn.commit()
|
||||
|
||||
def get(self, cache_key: str) -> ReadCacheEntry | None:
|
||||
with self._connect() as conn:
|
||||
row = conn.execute(
|
||||
"SELECT * FROM read_cache_entries WHERE cache_key = ?",
|
||||
(cache_key,),
|
||||
).fetchone()
|
||||
if row is None:
|
||||
return None
|
||||
return ReadCacheEntry(
|
||||
cache_key=row["cache_key"],
|
||||
status_code=row["status_code"],
|
||||
content=row["content"],
|
||||
content_type=row["content_type"],
|
||||
cached_at=row["cached_at"],
|
||||
)
|
||||
|
||||
def summary(self) -> dict[str, int | str | None]:
|
||||
with self._connect() as conn:
|
||||
row = conn.execute(
|
||||
"""
|
||||
SELECT COUNT(*) AS entry_count,
|
||||
MIN(cached_at) AS oldest_cached_at
|
||||
FROM read_cache_entries
|
||||
"""
|
||||
).fetchone()
|
||||
return {
|
||||
"entry_count": int(row["entry_count"] or 0),
|
||||
"oldest_cached_at": row["oldest_cached_at"],
|
||||
}
|
||||
|
|
@ -2,6 +2,7 @@ from __future__ import annotations
|
|||
|
||||
import os
|
||||
import socket
|
||||
from datetime import datetime, timezone
|
||||
from typing import Any
|
||||
|
||||
import httpx
|
||||
|
|
@ -9,6 +10,12 @@ from fastapi import FastAPI, Request
|
|||
from fastapi.responses import JSONResponse, Response
|
||||
|
||||
from api.edge.outbox import OutboxEnvelope, OutboxStore, PayloadRejected, default_outbox_path
|
||||
from api.edge.read_cache import (
|
||||
ReadCacheStore,
|
||||
build_cache_key,
|
||||
default_read_cache_path,
|
||||
is_cacheable_get,
|
||||
)
|
||||
from api.services.write_idempotency import route_class_for
|
||||
|
||||
HOP_BY_HOP_HEADERS = {
|
||||
|
|
@ -100,15 +107,35 @@ async def replay_pending(
|
|||
return counts
|
||||
|
||||
|
||||
def _cache_age_seconds(cached_at: str) -> int:
|
||||
try:
|
||||
cached = datetime.fromisoformat(cached_at)
|
||||
except ValueError:
|
||||
return 0
|
||||
if cached.tzinfo is None:
|
||||
cached = cached.replace(tzinfo=timezone.utc)
|
||||
return max(0, int((datetime.now(timezone.utc) - cached).total_seconds()))
|
||||
|
||||
|
||||
def _stale_cache_headers(cached_at: str) -> dict[str, str]:
|
||||
return {
|
||||
"X-StateHub-Edge-Cache": "stale",
|
||||
"X-StateHub-Edge-Cache-Age": str(_cache_age_seconds(cached_at)),
|
||||
}
|
||||
|
||||
|
||||
def create_app(
|
||||
*,
|
||||
upstream_url: str | None = None,
|
||||
outbox_path: str | None = None,
|
||||
read_cache_path: str | None = None,
|
||||
timeout: float = 10.0,
|
||||
) -> FastAPI:
|
||||
upstream = (upstream_url or os.environ.get("STATEHUB_UPSTREAM_URL") or os.environ.get("API_BASE") or "http://127.0.0.1:8000").rstrip("/")
|
||||
store_path = outbox_path or default_outbox_path()
|
||||
cache_path = read_cache_path or str(default_read_cache_path())
|
||||
store_instance: OutboxStore | None = None
|
||||
cache_instance: ReadCacheStore | None = None
|
||||
|
||||
def get_store() -> OutboxStore:
|
||||
nonlocal store_instance
|
||||
|
|
@ -116,6 +143,12 @@ def create_app(
|
|||
store_instance = OutboxStore(store_path)
|
||||
return store_instance
|
||||
|
||||
def get_cache() -> ReadCacheStore:
|
||||
nonlocal cache_instance
|
||||
if cache_instance is None:
|
||||
cache_instance = ReadCacheStore(cache_path)
|
||||
return cache_instance
|
||||
|
||||
app = FastAPI(title="State Hub Edge Relay", version="0.1.0")
|
||||
|
||||
@app.get("/edge/health")
|
||||
|
|
@ -134,6 +167,7 @@ def create_app(
|
|||
"upstream_reachable": reachable,
|
||||
"upstream_error": error,
|
||||
"outbox": get_store().summary(),
|
||||
"read_cache": get_cache().summary(),
|
||||
}
|
||||
|
||||
@app.post("/edge/replay")
|
||||
|
|
@ -156,6 +190,14 @@ def create_app(
|
|||
if request.headers.get("content-type"):
|
||||
headers["Content-Type"] = request.headers["content-type"]
|
||||
|
||||
cache_key = None
|
||||
if request.method == "GET" and is_cacheable_get(api_path):
|
||||
cache_key = build_cache_key(
|
||||
request.method,
|
||||
api_path,
|
||||
list(request.query_params.multi_items()),
|
||||
)
|
||||
|
||||
try:
|
||||
async with httpx.AsyncClient(base_url=upstream, timeout=timeout) as client:
|
||||
response = await client.request(
|
||||
|
|
@ -165,13 +207,36 @@ def create_app(
|
|||
json=body if body is not None else None,
|
||||
headers=headers,
|
||||
)
|
||||
if cache_key is not None and 200 <= response.status_code < 300:
|
||||
get_cache().put(
|
||||
cache_key=cache_key,
|
||||
status_code=response.status_code,
|
||||
content=response.content,
|
||||
content_type=response.headers.get("content-type"),
|
||||
)
|
||||
response_headers = _safe_response_headers(response.headers)
|
||||
if cache_key is not None:
|
||||
response_headers["X-StateHub-Edge-Cache"] = "hit"
|
||||
return Response(
|
||||
content=response.content,
|
||||
status_code=response.status_code,
|
||||
headers=_safe_response_headers(response.headers),
|
||||
headers=response_headers,
|
||||
media_type=response.headers.get("content-type"),
|
||||
)
|
||||
except httpx.HTTPError as exc:
|
||||
if cache_key is not None:
|
||||
cached = get_cache().get(cache_key)
|
||||
if cached is not None:
|
||||
stale_headers = _stale_cache_headers(cached.cached_at)
|
||||
if cached.content_type:
|
||||
stale_headers["Content-Type"] = cached.content_type
|
||||
return Response(
|
||||
content=cached.content,
|
||||
status_code=cached.status_code,
|
||||
headers=stale_headers,
|
||||
media_type=cached.content_type,
|
||||
)
|
||||
|
||||
route_class = route_class_for(request.method, api_path)
|
||||
if route_class is None or request.method not in {"POST", "PATCH"}:
|
||||
return JSONResponse(
|
||||
|
|
|
|||
|
|
@ -84,6 +84,25 @@ or secret-looking JSON fields. Payloads over 64 KiB are rejected.
|
|||
statehub outbox retry ENVELOPE_ID
|
||||
statehub outbox cancel ENVELOPE_ID
|
||||
|
||||
## Read Cache (Beachhead V1)
|
||||
|
||||
The edge relay also keeps a small SQLite read cache for allowlisted `GET`
|
||||
routes used by activity-core context resolution and daily triage:
|
||||
|
||||
- `/state/*`, `/workplans/*`, `/messages/*`, `/decisions/*`, `/tasks/*`,
|
||||
`/progress/*`, `/sbom/*`, `/repos*`, `/legacy-meter/*`
|
||||
|
||||
While upstream is reachable, successful `GET` responses are cached and marked
|
||||
`X-StateHub-Edge-Cache: hit`. When upstream is unreachable, the relay serves
|
||||
the last cached body with HTTP 200 and `X-StateHub-Edge-Cache: stale` plus
|
||||
`X-StateHub-Edge-Cache-Age`.
|
||||
|
||||
Default cache path: `STATEHUB_READ_CACHE_PATH`, recommended alongside the
|
||||
outbox at `~/.statehub/edge-read-cache.sqlite3`.
|
||||
|
||||
Uncached reads still return HTTP 503 during outage. Online-only `POST` routes
|
||||
remain queueable or rejected per the write allowlist above.
|
||||
|
||||
## Recovery Checklist
|
||||
|
||||
1. Confirm the central State Hub API is reachable.
|
||||
|
|
|
|||
114
tests/test_edge_read_cache.py
Normal file
114
tests/test_edge_read_cache.py
Normal file
|
|
@ -0,0 +1,114 @@
|
|||
import httpx
|
||||
import pytest
|
||||
from httpx import ASGITransport, AsyncClient
|
||||
|
||||
from api.edge.read_cache import ReadCacheStore, build_cache_key, is_cacheable_get
|
||||
from api.edge.relay import create_app
|
||||
|
||||
|
||||
class FailingAsyncClient:
|
||||
def __init__(self, *args, **kwargs):
|
||||
pass
|
||||
|
||||
async def __aenter__(self):
|
||||
return self
|
||||
|
||||
async def __aexit__(self, *exc_info):
|
||||
return False
|
||||
|
||||
async def request(self, *args, **kwargs):
|
||||
raise httpx.ConnectError("upstream down")
|
||||
|
||||
async def get(self, *args, **kwargs):
|
||||
raise httpx.ConnectError("upstream down")
|
||||
|
||||
|
||||
class SuccessAsyncClient:
|
||||
def __init__(self, *args, **kwargs):
|
||||
pass
|
||||
|
||||
async def __aenter__(self):
|
||||
return self
|
||||
|
||||
async def __aexit__(self, *exc_info):
|
||||
return False
|
||||
|
||||
async def request(self, method, path, **kwargs):
|
||||
request = httpx.Request(method, f"http://upstream{path}")
|
||||
return httpx.Response(
|
||||
200,
|
||||
json={"path": path, "ok": True},
|
||||
request=request,
|
||||
headers={"content-type": "application/json"},
|
||||
)
|
||||
|
||||
|
||||
def test_build_cache_key_sorts_query_params() -> None:
|
||||
key = build_cache_key("GET", "/state/summary", [("b", "2"), ("a", "1")])
|
||||
assert key == "GET:/state/summary?a=1&b=2"
|
||||
|
||||
|
||||
def test_is_cacheable_get_allows_activity_core_reads() -> None:
|
||||
assert is_cacheable_get("/state/summary")
|
||||
assert is_cacheable_get("/workplans/index")
|
||||
assert is_cacheable_get("/tasks/")
|
||||
assert is_cacheable_get("/edge/health") is False
|
||||
|
||||
|
||||
def test_read_cache_store_round_trip(tmp_path) -> None:
|
||||
store = ReadCacheStore(tmp_path / "cache.sqlite3")
|
||||
store.put(
|
||||
cache_key="GET:/state/summary",
|
||||
status_code=200,
|
||||
content=b'{"ok": true}',
|
||||
content_type="application/json",
|
||||
)
|
||||
entry = store.get("GET:/state/summary")
|
||||
assert entry is not None
|
||||
assert entry.status_code == 200
|
||||
assert entry.content == b'{"ok": true}'
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_relay_caches_successful_get_and_serves_stale_on_outage(tmp_path, monkeypatch):
|
||||
from api.edge import relay
|
||||
|
||||
monkeypatch.setattr(relay.httpx, "AsyncClient", SuccessAsyncClient)
|
||||
cache_path = tmp_path / "read-cache.sqlite3"
|
||||
app = create_app(
|
||||
upstream_url="http://upstream",
|
||||
outbox_path=str(tmp_path / "outbox.sqlite3"),
|
||||
read_cache_path=str(cache_path),
|
||||
)
|
||||
|
||||
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://edge") as client:
|
||||
warm = await client.get("/state/summary")
|
||||
assert warm.status_code == 200
|
||||
assert warm.headers.get("x-statehub-edge-cache") == "hit"
|
||||
|
||||
monkeypatch.setattr(relay.httpx, "AsyncClient", FailingAsyncClient)
|
||||
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://edge") as client:
|
||||
stale = await client.get("/state/summary")
|
||||
|
||||
assert stale.status_code == 200
|
||||
assert stale.headers.get("x-statehub-edge-cache") == "stale"
|
||||
assert stale.json()["ok"] is True
|
||||
assert int(stale.headers.get("x-statehub-edge-cache-age", "0")) >= 0
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_relay_returns_503_for_uncached_get_when_upstream_unreachable(tmp_path, monkeypatch):
|
||||
from api.edge import relay
|
||||
|
||||
monkeypatch.setattr(relay.httpx, "AsyncClient", FailingAsyncClient)
|
||||
app = create_app(
|
||||
upstream_url="http://upstream",
|
||||
outbox_path=str(tmp_path / "outbox.sqlite3"),
|
||||
read_cache_path=str(tmp_path / "read-cache.sqlite3"),
|
||||
)
|
||||
|
||||
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://edge") as client:
|
||||
response = await client.get("/state/summary")
|
||||
|
||||
assert response.status_code == 503
|
||||
assert "not queueable" in response.json()["error"]
|
||||
Loading…
Add table
Add a link
Reference in a new issue