diff --git a/.forgejo/workflows/image.yaml b/.forgejo/workflows/image.yaml index 203f1a4..3dc9af2 100644 --- a/.forgejo/workflows/image.yaml +++ b/.forgejo/workflows/image.yaml @@ -12,6 +12,9 @@ on: paths: - ".forgejo/workflows/image.yaml" - "Dockerfile" + - "api/**" + - "pyproject.toml" + - "uv.lock" workflow_dispatch: env: diff --git a/api/edge/read_cache.py b/api/edge/read_cache.py new file mode 100644 index 0000000..817bbf1 --- /dev/null +++ b/api/edge/read_cache.py @@ -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"], + } \ No newline at end of file diff --git a/api/edge/relay.py b/api/edge/relay.py index 9277652..362e8e3 100644 --- a/api/edge/relay.py +++ b/api/edge/relay.py @@ -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( diff --git a/docs/offline-write-buffer.md b/docs/offline-write-buffer.md index de985ff..fd8fe14 100644 --- a/docs/offline-write-buffer.md +++ b/docs/offline-write-buffer.md @@ -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. diff --git a/tests/test_edge_read_cache.py b/tests/test_edge_read_cache.py new file mode 100644 index 0000000..8ab5f6f --- /dev/null +++ b/tests/test_edge_read_cache.py @@ -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"] \ No newline at end of file