diff --git a/api/docs/adr/008-local-mode-composition.md b/api/docs/adr/008-local-mode-composition.md new file mode 100644 index 00000000..033f00a1 --- /dev/null +++ b/api/docs/adr/008-local-mode-composition.md @@ -0,0 +1,44 @@ +--- +date: 2026-07-08 +--- + +# ADR-008 - Local mode via composition, not conditionals + +## Context and Problem Statement + +The API supports a local development mode (`--path ` / `DW_API_DAMNIT_PATH`): no OAuth, no MyMdC, a synthetic development user, and a single fixed proposal directory instead of GPFS discovery. + +A mode can be implemented in two ways. One branches on `settings.is_local` at every affected call site. The other selects implementations once, at composition time. Scattered branching means every new code path must remember both modes, local behaviour is defined piecemeal, and the two modes can drift apart silently. Selecting implementations once means local behaviour is defined in one reviewable place, and both modes share the same interfaces by construction. + +## Considered Options + +- Branch on `settings.is_local` at each affected call site. +- Select implementations once, in the composition root. + +## Decision Outcome + +Chosen option: "select implementations once, in the composition root", because it keeps local behaviour in one place and makes the two modes share interfaces by construction. + +- `settings.is_local` is read only in the composition root (`main.py` / `state.py`) and in the `Settings` property that defines it ([ADR-003](003-injected-settings.md)). +- It selects implementations: a path locator ([ADR-004](004-proposal-path-locator.md)), a MyMdC client, an auth controller, and a store backend. +- Feature code depends on the selected collaborator through injection, never on the mode. + +### Consequences + +- Good: adding a code path cannot silently break local mode, because there is no second branch to forget. +- Good: local and production can differ only in implementation, never in interface shape; each collaborator is independently testable and tests get the local composition for free. +- Bad: a few more small interfaces exist (user provider, authorisation policy, proposals provider) rather than inline conditionals. + +## Details + +The composition maps each mode-dependent concern to a production and a local implementation. Some concerns are still branched inline rather than composed. + +| Concern | Production | Local | +|---|---|---| +| Proposal path resolution | GPFS path locator ([ADR-004](004-proposal-path-locator.md)) | fixed path locator for the given directory | +| MyMdC | HTTP client | mock client synthesising the single local proposal | +| Auth routes | OAuth controller | no-auth controller | +| Current user | session-derived user info | synthetic development user | +| Authorisation | proposal-membership policy | allow-all policy | +| Proposal metadata | MyMdC fetch plus app-DB cache | synthetic metadata for the local directory | +| Stores | file-backed | in-memory | diff --git a/api/docs/adr/009-channels-subscriptions.md b/api/docs/adr/009-channels-subscriptions.md new file mode 100644 index 00000000..9cf300d9 --- /dev/null +++ b/api/docs/adr/009-channels-subscriptions.md @@ -0,0 +1,40 @@ +--- +date: 2026-07-08 +--- + +# ADR-009 - Subscriptions: channel consumers, composition-selected publisher + +## Context and Problem Statement + +The frontend's live table updates use a GraphQL subscription (`latest_data`) over `graphql-transport-ws`. DAMNIT proposal databases are plain SQLite files written by an external process. There is no change feed to subscribe to today, so the server must produce one by polling. + +That will not stay true. Candidate event sources exist on the horizon: the facility Kafka bus, or Postgres `LISTEN/NOTIFY` should the application database move to Postgres. The subscription design must not weld resolvers to any one change-detection mechanism. + +Two forces shape the design. Naive per-client polling multiplies identical reads: N subscribers to one proposal would issue N queries per tick against a GPFS-hosted SQLite file. And any process-local coordination state is a deployment constraint that must be a recorded decision with a retirement path, not an accident. + +## Considered Options + +- Per-client polling loops inside the resolver. +- Subscribers consume a channel; a composition-selected publisher produces events. + +## Decision Outcome + +Chosen option: "subscribers consume a channel; a composition-selected publisher produces events", because it keeps subscribers stable while the event source changes and coalesces reads structurally. + +- Subscribers consume a per-proposal channel on Litestar's `ChannelsPlugin` and fan events to their client. Resolvers hold no polling loops, no cursors, and no knowledge of how an event was produced. +- Exactly one publisher per deployment is chosen in the composition root, the same composition-root selection as [ADR-008](008-local-mode-composition.md). Today it is a SQLite poller; a Postgres `LISTEN/NOTIFY` consumer or a Kafka bridge can replace it behind the same contract without touching subscribers. +- The poller reads each watched proposal once per tick through the repository ([ADR-005](005-repository-pattern.md)) and publishes new rows, so one poll serves every subscriber. Its per-proposal high-water mark is publisher-internal state, deleted with the poller when a push publisher lands. +- Each client passes its own `timestamp`; the resolver filters the shared event per client, so late joiners receive neither stale rows nor duplicates. +- Persistent publisher failures publish an error event and terminate affected subscriptions with a typed error ([ADR-001](001-error-classes.md)), so clients resubscribe deliberately rather than silently receiving nothing. + +### Consequences + +- Good: subscriber code is stable across the polling to push migration; the Postgres-versus-Kafka choice is deferred without accruing rewrite cost. +- Good: under polling, load scales with the number of watched proposals, not connected clients; under a push publisher the per-tick GPFS reads disappear. +- Bad: the process-local channels backend forces single-worker deployment for now, a constraint that must stay loud until retired. + +## Details + +The single-worker retirement plan, in order: move sessions and tokens to a store-backed server-side session; move the channels backend to a shared backend; make the publisher push-based (deleting the poller and its cursors) or, if polling must persist, run it once per deployment with store-backed cursors. After those steps nothing process-local remains that is not a cache (the per-repository caches of [ADR-005](005-repository-pattern.md)), and the startup guard is removed. + +The event contract is that every new run is published to its proposal's channel exactly once, in order. Push publishers satisfy it natively. The polling publisher satisfies it with the high-water mark, initialised from the proposal's current max timestamp and advanced as new rows are seen. diff --git a/api/docs/architecture.md b/api/docs/architecture.md index ff00ee13..c6a05626 100644 --- a/api/docs/architecture.md +++ b/api/docs/architecture.md @@ -58,7 +58,7 @@ The key rules are: 3. **Composition root reads settings:** everything else receives configuration as parameters (see [ADR-003](adr/003-injected-settings.md)). 4. **Authorisation applied at the edge:** routes and resolvers use dependencies and permission classes. - This means that services should not apply authorisation rules themselves. -5. **No `if settings.is_local:` outside the composition root:** Local mode is selected by composition, not conditionals throughout the codebase. +5. **No `if settings.is_local:` outside the composition root:** Local mode is selected by composition, not conditionals throughout the codebase (see [ADR-008](adr/008-local-mode-composition.md)). Note that these are currently only enforced by convention/review. Import linter/archetecture check tool is planned to be added. diff --git a/api/pyproject.toml b/api/pyproject.toml index 813a1bf1..d3848627 100644 --- a/api/pyproject.toml +++ b/api/pyproject.toml @@ -79,8 +79,8 @@ dev = [ default-groups = [] [tool.pytest.ini_options] -# Session-scoped event loop so alru_cache instances bind once and never -# see a loop change between tests (no AlruCacheLoopResetWarning). +# Session-scoped event loop so async fixtures (e.g. the channels plugin) and +# tests share one loop across the suite, with no loop change between tests. asyncio_default_fixture_loop_scope = "session" asyncio_default_test_loop_scope = "session" diff --git a/api/src/damnit_api/graphql/dependencies.py b/api/src/damnit_api/graphql/dependencies.py index 7a0c69c8..311b3cea 100644 --- a/api/src/damnit_api/graphql/dependencies.py +++ b/api/src/damnit_api/graphql/dependencies.py @@ -1,15 +1,23 @@ """Litestar dependency helpers for GraphQL subscription state.""" +from litestar.channels import ChannelsPlugin from litestar.datastructures import State -from .subscriptions import SubscriptionCursors +from .publisher import RunUpdatePublisher -def get_subscription_cursors(state: State) -> SubscriptionCursors: - """Provide the subscription cursors from the application state.""" - return state.app_state.subscription_cursors # type: ignore[attr-defined] +def get_channels(state: State) -> ChannelsPlugin: + """Provide the channels plugin from the application state.""" + return state.app_state.channels # type: ignore[attr-defined] -# Plain type alias; Litestar injects by the parameter name `subscription_cursors`. -SubscriptionCursorsDep = SubscriptionCursors -"""Type alias for the subscription cursors dependency.""" +def get_run_update_publisher(state: State) -> RunUpdatePublisher: + """Provide the run-update publisher from the application state.""" + return state.app_state.run_update_publisher # type: ignore[attr-defined] + + +# Plain type aliases; Litestar injects by the parameter name. +ChannelsDep = ChannelsPlugin +"""Type alias for the channels-plugin dependency.""" +RunUpdatePublisherDep = RunUpdatePublisher +"""Type alias for the run-update-publisher dependency.""" diff --git a/api/src/damnit_api/graphql/publisher.py b/api/src/damnit_api/graphql/publisher.py new file mode 100644 index 00000000..ba8d6833 --- /dev/null +++ b/api/src/damnit_api/graphql/publisher.py @@ -0,0 +1,165 @@ +"""Run-update publishers: composition-selected producers of channel events. + +Subscription resolvers consume per-proposal channels (ADR-009); exactly one +publisher per deployment produces the events. The SQLite poller below is the +default; a Postgres LISTEN/NOTIFY consumer or a Kafka bridge slot in behind +the same contract in the composition root (ADR-008) without touching +subscribers. +""" + +import asyncio +import dataclasses +from typing import TYPE_CHECKING, Any, Protocol, runtime_checkable + +from .. import get_logger +from ..runs.types import DamnitRun +from ..shared.models import ProposalNumber + +if TYPE_CHECKING: + from litestar.channels import ChannelsPlugin + + from ..runs.repository import DamnitRepositoryRegistry + +logger = get_logger() + +POLLING_INTERVAL = 1.0 # seconds +MAX_CONSECUTIVE_FAILURES = 5 + + +def proposal_channel(proposal_number: ProposalNumber) -> str: + """Name of the run-updates channel for a proposal.""" + return f"proposal:{proposal_number}" + + +# `runtime_checkable` so Litestar's msgspec-based signature validation +# (`isinstance` on injected Protocols) doesn't raise on every request. +@runtime_checkable +class RunUpdatePublisher(Protocol): + """Produce run-update events on per-proposal channels. + + The event payload is the snapshot dict consumed by `latest_data` + (`runs`/`run_timestamps`/`max_timestamp`/`metadata`), or + `{"error": {...}}` when the publisher gives up on a proposal. + """ + + def watch(self, proposal_number: ProposalNumber) -> None: ... + async def aclose(self) -> None: ... + + +class SqlitePollingRunUpdatePublisher: + """Polls DAMNIT databases and publishes new-run snapshots per proposal. + + One poll task per watched proposal regardless of subscriber count — + coalescing is structural (one publisher, N channel subscribers). The + per-proposal high-water mark keeps poll cost proportional to new data; + it is an implementation detail of *this* publisher and is deleted with + it under push-based backends (ADR-009). + """ + + def __init__( + self, + channels: "ChannelsPlugin", + repositories: "DamnitRepositoryRegistry", + *, + interval: float = POLLING_INTERVAL, + max_consecutive_failures: int = MAX_CONSECUTIVE_FAILURES, + ) -> None: + self._channels = channels + self._repositories = repositories + self._interval = interval + self._max_consecutive_failures = max_consecutive_failures + self._cursors: dict[ProposalNumber, float] = {} + self._tasks: dict[ProposalNumber, asyncio.Task] = {} + + def watch(self, proposal_number: ProposalNumber) -> None: + """Ensure a poll task is running for the proposal.""" + task = self._tasks.get(proposal_number) + if task is None or task.done(): + self._tasks[proposal_number] = asyncio.create_task( + self._poll_loop(proposal_number) + ) + + async def aclose(self) -> None: + for task in self._tasks.values(): + task.cancel() + await asyncio.gather(*self._tasks.values(), return_exceptions=True) + self._tasks.clear() + + async def _poll_loop(self, proposal_number: ProposalNumber) -> None: + failures = 0 + while True: + await asyncio.sleep(self._interval) + try: + snapshot = await self._poll(proposal_number) + except Exception: + failures += 1 + logger.exception( + "Run-update poll failed", + proposal=proposal_number, + consecutive_failures=failures, + ) + if failures >= self._max_consecutive_failures: + # Persistent failure: tell subscribers to terminate with + # a typed error (ADR-001) instead of silently going quiet. + self._channels.publish( + { + "error": { + "message": ( + "Run updates unavailable after " + f"{failures} consecutive poll failures" + ), + } + }, + proposal_channel(proposal_number), + ) + self._tasks.pop(proposal_number, None) + return + continue + + failures = 0 + if snapshot is not None: + self._channels.publish(snapshot, proposal_channel(proposal_number)) + + async def _poll(self, proposal_number: ProposalNumber) -> dict[str, Any] | None: + """One poll tick: return a snapshot of rows newer than the cursor.""" + repo = self._repositories.get(proposal_number) + + # Initialize cursor from the current max timestamp on first visit + if proposal_number not in self._cursors: + metadata = await repo.get_metadata() + self._cursors[proposal_number] = metadata.timestamp + + start_at = self._cursors[proposal_number] + records = await repo.get_latest_runs(start_at=start_at) + if not records: + return None + + runs: dict[int, Any] = {} + run_timestamps: dict[int, float] = {} + for record in records: + runs[record.run] = DamnitRun.resolve_record(record) + if record.variables: + run_timestamps[record.run] = max( + vv.timestamp for vv in record.variables.values() + ) + + if not runs: + return None + + latest_ts = max(run_timestamps.values(), default=start_at) + + repo.invalidate_metadata_cache() + metadata = await repo.get_metadata() + self._cursors[proposal_number] = latest_ts + meta_dict = dataclasses.asdict(metadata) + + return { + "runs": runs, + "run_timestamps": run_timestamps, + "max_timestamp": latest_ts, + "metadata": { + "runs": sorted(set(metadata.runs) | set(runs.keys())), + "variables": meta_dict["variables"], + "timestamp": latest_ts * 1000, # ms for JS + }, + } diff --git a/api/src/damnit_api/graphql/subscriptions.py b/api/src/damnit_api/graphql/subscriptions.py index 1b5f79c8..01d6cf82 100644 --- a/api/src/damnit_api/graphql/subscriptions.py +++ b/api/src/damnit_api/graphql/subscriptions.py @@ -1,99 +1,24 @@ -import asyncio -import dataclasses +import json from collections.abc import AsyncGenerator -from typing import Any +from typing import TYPE_CHECKING, Any import strawberry -from async_lru import alru_cache from strawberry.scalars import JSON from strawberry.types import Info from .. import get_logger from ..auth.permissions import PROPOSAL_PERMISSIONS -from ..runs.repository import DamnitRepository -from ..runs.types import DamnitRun, Timestamp -from ..shared.models import ProposalNumber +from ..runs.types import Timestamp +from ..shared.errors import DataUnavailableError +from .publisher import proposal_channel from .utils import DatabaseInput -logger = get_logger() - -POLLING_INTERVAL = 1 # seconds - - -class SubscriptionCursors: - """Server-side high-water mark per proposal so each tick only fetches rows - newer than what the previous tick already shipped. Hashable by identity - for alru_cache.""" - - def __init__(self) -> None: - self._data: dict[ProposalNumber, float] = {} - - def __contains__(self, proposal_number: ProposalNumber) -> bool: - return proposal_number in self._data - - def __getitem__(self, proposal_number: ProposalNumber) -> float: - return self._data[proposal_number] - - def __setitem__(self, proposal_number: ProposalNumber, value: float) -> None: - self._data[proposal_number] = value - - def clear(self) -> None: - self._data.clear() - - -# Per-client cursor is deliberately omitted from the cache key so that -# concurrent subscribers coalesce into a single DB read per tick. -@alru_cache(maxsize=32, ttl=POLLING_INTERVAL) -async def poll_proposal( - proposal_number: ProposalNumber, - cursors: SubscriptionCursors, - repo: DamnitRepository, -) -> dict[str, Any] | None: - # Initialise cursor from the current max timestamp on first visit. - if proposal_number not in cursors: - try: - metadata = await repo.get_metadata() - cursors[proposal_number] = metadata.timestamp - except Exception: - return None +if TYPE_CHECKING: + from litestar.channels import ChannelsPlugin - start_at = cursors[proposal_number] - records = await repo.get_latest_runs(start_at=start_at) - if not records: - return None - - runs: dict[int, Any] = {} - run_timestamps: dict[int, float] = {} - for record in records: - runs[record.run] = DamnitRun.resolve_record(record) - if record.variables: - run_timestamps[record.run] = max( - vv.timestamp for vv in record.variables.values() - ) - - if not runs: - return None - - latest_ts = max(run_timestamps.values(), default=start_at) + from .publisher import RunUpdatePublisher - repo.invalidate_metadata_cache() - try: - metadata = await repo.get_metadata() - except Exception: - return None - cursors[proposal_number] = latest_ts - meta_dict = dataclasses.asdict(metadata) - - return { - "runs": runs, - "run_timestamps": run_timestamps, - "max_timestamp": latest_ts, - "metadata": { - "runs": sorted(set(metadata.runs) | set(runs.keys())), - "variables": meta_dict["variables"], - "timestamp": latest_ts * 1000, # ms for JS - }, - } +logger = get_logger() def filter_for_client( @@ -121,22 +46,32 @@ async def latest_data( database: DatabaseInput, timestamp: Timestamp, ) -> AsyncGenerator[JSON]: - cursors: SubscriptionCursors = info.context.subscription_cursors - while True: - await asyncio.sleep(POLLING_INTERVAL) - - proposal = database.proposal - try: - repo = info.context.repositories.get(proposal) - snapshot = await poll_proposal( - proposal_number=proposal, - cursors=cursors, - repo=repo, - ) - except Exception: - logger.exception("Subscription poll failed", proposal=proposal) - raise - - result = filter_for_client(snapshot, timestamp) - if result is not None: - yield result # ty: ignore[invalid-yield] + """Stream new-run snapshots for a proposal. + + Consumes the proposal's run-updates channel; how events are produced + (SQLite poller today, pg-notify/Kafka later) is the publisher's + concern (ADR-009). Late joiners are handled by the per-client + `timestamp` filter, not by the channel. + """ + proposal_number = database.proposal + channels: ChannelsPlugin = info.context.channels + publisher: RunUpdatePublisher = info.context.run_update_publisher + + publisher.watch(proposal_number) + + async with channels.start_subscription( + proposal_channel(proposal_number) + ) as subscriber: + async for raw in subscriber.iter_events(): + event = json.loads(raw) + if error := event.get("error"): + raise DataUnavailableError(error["message"]) + # Channel events are JSON-encoded; restore the integer run + # keys the payload contract uses. + event["runs"] = {int(k): v for k, v in event["runs"].items()} + event["run_timestamps"] = { + int(k): v for k, v in event["run_timestamps"].items() + } + result = filter_for_client(event, timestamp) + if result is not None: + yield result # ty: ignore[invalid-yield] diff --git a/api/src/damnit_api/main.py b/api/src/damnit_api/main.py index 529c9ce5..0025e86f 100644 --- a/api/src/damnit_api/main.py +++ b/api/src/damnit_api/main.py @@ -15,6 +15,8 @@ def create_app(): from litestar import Request, Response + from litestar.channels import ChannelsPlugin + from litestar.channels.backends.memory import MemoryChannelsBackend from litestar.middleware.session.server_side import ServerSideSessionConfig from litestar.openapi import OpenAPIConfig from litestar.response import Redirect @@ -24,7 +26,8 @@ def create_app(): from . import _logging, auth, get_logger from .auth.oauth import SESSION_COOKIE_KEY, create_oauth_client - from .graphql.dependencies import get_subscription_cursors + from .graphql.dependencies import get_channels, get_run_update_publisher + from .graphql.publisher import SqlitePollingRunUpdatePublisher from .runs.dependencies import get_repositories from .shared.errors import DamnitWebError from .shared.gql import get_gql_controller @@ -35,7 +38,6 @@ def create_app(): create_db_sessionmaker, create_mymdc_client, create_repositories, - create_subscription_cursors, provide_app_state, ) @@ -48,6 +50,32 @@ def create_app(): # (in-memory locally, file-backed otherwise) and selected below. session_config = ServerSideSessionConfig(key=SESSION_COOKIE_KEY) + # ── Channels (run-update pub/sub) ───────────────────────────────────────── + # Subscribers consume per-proposal channels; the composition-selected + # publisher (built in the lifespan) produces the events. The in-memory + # backend is process-local (ADR-009). + channels_plugin = ChannelsPlugin( + backend=MemoryChannelsBackend(), + arbitrary_channels_allowed=True, + ) + + # ── Multi-worker guard ──────────────────────────────────────────────────── + # The channels backend (and, in local mode, the session store) are + # process-local, so run-update events and sessions are not shared across + # workers. Refuse to start multi-worker until shared backends are wired + # (ADR-009). + process_local = ["channels: MemoryChannelsBackend"] + if settings.is_local: + process_local.append("stores: MemoryStore") + workers = getattr(settings.uvicorn, "workers", None) or 1 + if workers > 1: + msg = ( + f"uvicorn workers={workers} requires shared backends, but these " + f"are process-local: {', '.join(process_local)}. Run a single " + "worker, or wire shared channels/store backends (ADR-009)." + ) + raise RuntimeError(msg) + # ── Exception handlers ──────────────────────────────────────────────────── def dw_error_handler(request: Request, exc: DamnitWebError) -> Response: code = getattr(exc, "code", None) or 500 @@ -92,18 +120,26 @@ async def lifespan(app: Litestar): if oauth_client is not None: await oauth_client.load_server_metadata() + repositories = create_repositories() + run_update_publisher = SqlitePollingRunUpdatePublisher( + channels=channels_plugin, + repositories=repositories, + ) + app.state.app_state = AppState( db_engine=engine, db_sessionmaker=create_db_sessionmaker(engine), mymdc_client=create_mymdc_client(settings), oauth_client=oauth_client, - repositories=create_repositories(), - subscription_cursors=create_subscription_cursors(), + repositories=repositories, + channels=channels_plugin, + run_update_publisher=run_update_publisher, ) try: yield finally: + await run_update_publisher.aclose() await engine.dispose() def _file_store(name: str) -> FileStore: @@ -144,11 +180,13 @@ def _file_store(name: str) -> FileStore: "mymdc": Provide(get_mymdc_client, sync_to_thread=False), "user": Provide(get_user), "oauth_user": Provide(get_oauth_user_info, sync_to_thread=False), - "subscription_cursors": Provide( - get_subscription_cursors, sync_to_thread=False + "channels": Provide(get_channels, sync_to_thread=False), + "run_update_publisher": Provide( + get_run_update_publisher, sync_to_thread=False ), "repositories": Provide(get_repositories, sync_to_thread=False), }, + plugins=[channels_plugin], stores=stores, middleware=[ session_config.middleware, diff --git a/api/src/damnit_api/shared/gql.py b/api/src/damnit_api/shared/gql.py index edbbd139..3c19fe59 100644 --- a/api/src/damnit_api/shared/gql.py +++ b/api/src/damnit_api/shared/gql.py @@ -12,7 +12,7 @@ from ..auth import gql as auth from ..auth.dependencies import OAuthUserInfo from ..auth.models import User -from ..graphql.dependencies import SubscriptionCursorsDep +from ..graphql.dependencies import ChannelsDep, RunUpdatePublisherDep from ..metadata import gql as metadata from ..runs import types as run_types from ..runs.dependencies import Repositories @@ -40,7 +40,8 @@ class Context(BaseContext): oauth_user: OAuthUserInfo session: DBSession repositories: Repositories - subscription_cursors: SubscriptionCursorsDep + channels: ChannelsDep + run_update_publisher: RunUpdatePublisherDep _user: User | None = None async def get_user(self) -> User: @@ -57,14 +58,16 @@ async def get_context( # noqa: RUF029 mymdc: MyMdCClient, session: DBSession, repositories: Repositories, - subscription_cursors: SubscriptionCursorsDep, + channels: ChannelsDep, + run_update_publisher: RunUpdatePublisherDep, ) -> Context: return Context( oauth_user=oauth_user, mymdc=mymdc, session=session, repositories=repositories, - subscription_cursors=subscription_cursors, + channels=channels, + run_update_publisher=run_update_publisher, ) diff --git a/api/src/damnit_api/shared/settings.py b/api/src/damnit_api/shared/settings.py index 36bdc434..6e89b984 100644 --- a/api/src/damnit_api/shared/settings.py +++ b/api/src/damnit_api/shared/settings.py @@ -30,6 +30,12 @@ class UvicornSettings(BaseModel): ssl_keyfile: FilePath | None = None ssl_certfile: FilePath | None = None + # NOTE (ops): `workers` may be set here (extra fields are allowed and + # passed straight to uvicorn), but workers > 1 requires shared channels + # and store backends. With the current process-local backends + # (MemoryChannelsBackend; MemoryStore in local mode) the app refuses to + # start multi-worker; see the guard in main.create_app (ADR-009). + @field_validator("factory", mode="after") @classmethod def factory_must_be_true(cls, v, values): diff --git a/api/src/damnit_api/state.py b/api/src/damnit_api/state.py index a73b7f5d..38112cfe 100644 --- a/api/src/damnit_api/state.py +++ b/api/src/damnit_api/state.py @@ -18,9 +18,11 @@ from sqlmodel.ext.asyncio.session import AsyncSession if TYPE_CHECKING: + from litestar.channels import ChannelsPlugin + from ._mymdc.clients import MyMdCClient from .auth.oauth import OAuthClient - from .graphql.subscriptions import SubscriptionCursors + from .graphql.publisher import RunUpdatePublisher from .runs.repository import DamnitRepositoryRegistry from .shared.settings import Settings @@ -32,7 +34,8 @@ class AppState: mymdc_client: MyMdCClient oauth_client: OAuthClient | None # None when auth is disabled repositories: DamnitRepositoryRegistry - subscription_cursors: SubscriptionCursors + channels: ChannelsPlugin + run_update_publisher: RunUpdatePublisher def create_db_engine(settings: Settings) -> AsyncEngine: @@ -71,12 +74,6 @@ def create_repositories() -> DamnitRepositoryRegistry: return DamnitRepositoryRegistry(SQLiteDamnitRepository) -def create_subscription_cursors() -> SubscriptionCursors: - from .graphql.subscriptions import SubscriptionCursors - - return SubscriptionCursors() - - def provide_app_state(state: LitestarState) -> AppState: """Litestar dependency: the application's :class:`AppState`.""" return state.app_state # type: ignore[attr-defined] diff --git a/api/tests/graphql/conftest.py b/api/tests/graphql/conftest.py index 9977d43c..726bd026 100644 --- a/api/tests/graphql/conftest.py +++ b/api/tests/graphql/conftest.py @@ -2,16 +2,14 @@ from types import SimpleNamespace import pytest +import pytest_asyncio import strawberry from strawberry.schema.config import StrawberryConfig from damnit_api.graphql.directives import lightweight +from damnit_api.graphql.publisher import SqlitePollingRunUpdatePublisher from damnit_api.graphql.queries import Query -from damnit_api.graphql.subscriptions import ( - Subscription, - SubscriptionCursors, - poll_proposal, -) +from damnit_api.graphql.subscriptions import Subscription from damnit_api.runs.csv import CsvDamnitRepository from damnit_api.runs.repository import DamnitRepositoryRegistry from damnit_api.runs.types import SCALAR_MAP, DamnitVariable @@ -40,14 +38,24 @@ async def subscribe(self, query, *, context_value=None, variable_values=None): ) -@pytest.fixture -def subscription_cursors() -> SubscriptionCursors: - return SubscriptionCursors() +@pytest_asyncio.fixture +async def channels_plugin(): + from litestar.channels import ChannelsPlugin + from litestar.channels.backends.memory import MemoryChannelsBackend + + plugin = ChannelsPlugin( + backend=MemoryChannelsBackend(), arbitrary_channels_allowed=True + ) + async with plugin: + yield plugin -@pytest.fixture(autouse=True) -def reset_caches(): - poll_proposal.cache_clear() +def make_publisher(channels_plugin, repositories, **kwargs): + """A fast-ticking SQLite polling publisher for subscription tests.""" + kwargs.setdefault("interval", 0.01) + return SqlitePollingRunUpdatePublisher( + channels=channels_plugin, repositories=repositories, **kwargs + ) def _patch_permissions(mocker, *, authenticated: bool, member: bool) -> None: @@ -96,7 +104,6 @@ def mock_repositories(csv_fixture_dir): def graphql_context(mock_repositories): return SimpleNamespace( repositories=mock_repositories, - subscription_cursors=SubscriptionCursors(), oauth_user=None, ) diff --git a/api/tests/graphql/test_publisher.py b/api/tests/graphql/test_publisher.py new file mode 100644 index 00000000..1e2deaae --- /dev/null +++ b/api/tests/graphql/test_publisher.py @@ -0,0 +1,180 @@ +"""Unit tests for SqlitePollingRunUpdatePublisher. + +Cursor semantics, structural coalescing (one poll task per proposal) and the +give-up-with-an-error-event failure path are exercised directly against a +mocked repository. The channels plugin is a plain mock; publish payloads are +asserted, not delivered. +""" + +import asyncio +from datetime import UTC, datetime +from unittest.mock import AsyncMock, MagicMock + +import pytest + +from damnit_api.graphql.publisher import ( + SqlitePollingRunUpdatePublisher, + proposal_channel, +) +from damnit_api.runs.models import MetadataSnapshot, RunRecord, VariableValue +from damnit_api.runs.repository import DamnitRepositoryRegistry +from damnit_api.runs.types import DamnitRun +from damnit_api.shared.models import ProposalNumber + +from .const import KNOWN_DATA, NEW_DATA, PROPOSAL + +NEW_RUN = 400 +PROPOSAL_NO = ProposalNumber(PROPOSAL) + + +@pytest.fixture +def current_timestamp(): + return datetime.now(tz=UTC).timestamp() + + +def _new_record(current_timestamp) -> RunRecord: + return RunRecord( + proposal=PROPOSAL_NO, + run=NEW_RUN, + start_time=KNOWN_DATA["start_time"].value, # ty: ignore[invalid-argument-type] + added_at=KNOWN_DATA["added_at"].value, # ty: ignore[invalid-argument-type] + variables={ + name: VariableValue( + value=data.value, + summary_type=data.summary_type, + timestamp=current_timestamp, + ) + for name, data in NEW_DATA.items() + }, + ) + + +def _publisher_for(repo, **kwargs) -> SqlitePollingRunUpdatePublisher: + registry = DamnitRepositoryRegistry(lambda _p: repo) + return SqlitePollingRunUpdatePublisher( + channels=MagicMock(), repositories=registry, **kwargs + ) + + +# ----------------------------------------------------------------------------- +# Cursor semantics + + +@pytest.mark.asyncio +async def test_poll_uses_existing_cursor(mocker, current_timestamp): + """On the second poll, the cursor advanced by the first poll is used.""" + proposal = PROPOSAL_NO + repo = MagicMock() + repo.get_metadata = AsyncMock( + return_value=MetadataSnapshot( + runs=(), variables={}, tags={}, timestamp=current_timestamp - 10 + ) + ) + repo.get_latest_runs = AsyncMock(return_value=[_new_record(current_timestamp)]) + repo.invalidate_metadata_cache = MagicMock() + mocker.patch.object( + DamnitRun, "resolve_record", side_effect=lambda r: {"run": r.run} + ) + publisher = _publisher_for(repo) + + await publisher._poll(proposal) + assert (call := repo.get_latest_runs.await_args) is not None + assert call.kwargs["start_at"] == current_timestamp - 10 + + await publisher._poll(proposal) + assert (call := repo.get_latest_runs.await_args) is not None + assert call.kwargs["start_at"] == current_timestamp + + +@pytest.mark.asyncio +async def test_poll_cursor_unchanged_on_empty_result(current_timestamp): + """When get_latest_runs returns [], the cursor is not advanced.""" + proposal = PROPOSAL_NO + repo = MagicMock() + repo.get_metadata = AsyncMock( + return_value=MetadataSnapshot( + runs=(), variables={}, tags={}, timestamp=current_timestamp + ) + ) + repo.get_latest_runs = AsyncMock(return_value=[]) + publisher = _publisher_for(repo) + + result = await publisher._poll(proposal) + assert result is None + assert publisher._cursors[proposal] == current_timestamp + + +@pytest.mark.asyncio +async def test_poll_empty_variables_cursor_stays_pinned(mocker, current_timestamp): + """Runs without variables don't advance the cursor past start_at.""" + proposal = PROPOSAL_NO + record = RunRecord( + proposal=proposal, + run=NEW_RUN, + start_time=KNOWN_DATA["start_time"].value, # ty: ignore[invalid-argument-type] + added_at=KNOWN_DATA["added_at"].value, # ty: ignore[invalid-argument-type] + variables={}, + ) + repo = MagicMock() + repo.get_metadata = AsyncMock( + return_value=MetadataSnapshot( + runs=(), variables={}, tags={}, timestamp=current_timestamp + ) + ) + repo.get_latest_runs = AsyncMock(return_value=[record]) + repo.invalidate_metadata_cache = MagicMock() + mocker.patch.object( + DamnitRun, "resolve_record", side_effect=lambda r: {"run": r.run} + ) + publisher = _publisher_for(repo) + + snapshot = await publisher._poll(proposal) + assert snapshot is not None + assert snapshot["max_timestamp"] == current_timestamp + assert publisher._cursors[proposal] == current_timestamp + + +# ----------------------------------------------------------------------------- +# Structural coalescing: one poll task per proposal + + +@pytest.mark.asyncio +async def test_watch_is_idempotent_per_proposal(): + """Repeated watch() calls for one proposal share a single poll task.""" + proposal = PROPOSAL_NO + publisher = _publisher_for(MagicMock()) + + publisher.watch(proposal) + publisher.watch(proposal) + try: + assert len(publisher._tasks) == 1 + finally: + await publisher.aclose() + assert publisher._tasks == {} + + +# ----------------------------------------------------------------------------- +# Failure behaviour + + +@pytest.mark.asyncio +async def test_poll_loop_publishes_error_event_after_max_failures(current_timestamp): + """The poll loop gives up after N consecutive failures with an error event.""" + proposal = PROPOSAL_NO + repo = MagicMock() + repo.get_metadata = AsyncMock(side_effect=OSError("gpfs down")) + channels = MagicMock() + registry = DamnitRepositoryRegistry(lambda _p: repo) + publisher = SqlitePollingRunUpdatePublisher( + channels=channels, + repositories=registry, + interval=0, + max_consecutive_failures=3, + ) + + await asyncio.wait_for(publisher._poll_loop(proposal), timeout=2) + + channels.publish.assert_called_once() + event, channel = channels.publish.call_args.args + assert channel == proposal_channel(proposal) + assert "error" in event diff --git a/api/tests/graphql/test_subscriptions.py b/api/tests/graphql/test_subscriptions.py index b0f93fc1..75c274f2 100644 --- a/api/tests/graphql/test_subscriptions.py +++ b/api/tests/graphql/test_subscriptions.py @@ -1,95 +1,276 @@ -"""Tests for the latest_data subscription polling logic. +"""Tests for the latest_data subscription. -`poll_proposal` and `filter_for_client` are exercised directly against a -`CsvDamnitRepository` (see conftest). The CSV fixtures put run 348's variables -at timestamp 1000.0, so a cursor seeded below that surfaces the run. +Subscribers consume a per-proposal channel fed by the composition-selected +publisher; `filter_for_client` narrows each snapshot to what a given client +has not yet seen. The CSV fixtures put run 348's variables at timestamp +1000.0, and `subscription_repo` injects one fresh run on top. """ +import asyncio +from datetime import UTC, datetime +from types import SimpleNamespace +from unittest.mock import AsyncMock, MagicMock + import pytest -from damnit_api.graphql.subscriptions import ( - SubscriptionCursors, - filter_for_client, - poll_proposal, -) +from damnit_api.graphql.subscriptions import filter_for_client +from damnit_api.runs.csv import CsvDamnitRepository +from damnit_api.runs.models import RunRecord, VariableValue +from damnit_api.runs.repository import DamnitRepositoryRegistry +from damnit_api.runs.types import DamnitRun +from damnit_api.shared.const import DamnitType from damnit_api.shared.models import ProposalNumber -from .const import PROPOSAL +from .conftest import make_publisher +from .const import KNOWN_DATA, NEW_DATA, PROPOSAL, RUNS, DatabaseVariable -_PROPOSAL = ProposalNumber(PROPOSAL) +NEW_RUN = 400 +PROPOSAL_NO = ProposalNumber(PROPOSAL) -@pytest.mark.asyncio -async def test_poll_proposal_surfaces_new_runs(mock_repositories): - cursors = SubscriptionCursors() - cursors[_PROPOSAL] = 0.0 # seed below the fixture timestamps - repo = mock_repositories.get(_PROPOSAL) - - snapshot = await poll_proposal(_PROPOSAL, cursors, repo) - - assert snapshot is not None - assert set(snapshot.keys()) == { - "runs", - "run_timestamps", - "max_timestamp", - "metadata", +@pytest.fixture(scope="module") +def current_timestamp(): + return datetime.now(tz=UTC).timestamp() + + +def _new_record(current_timestamp) -> RunRecord: + return RunRecord( + proposal=PROPOSAL_NO, + run=NEW_RUN, + start_time=KNOWN_DATA["start_time"].value, # ty: ignore[invalid-argument-type] + added_at=KNOWN_DATA["added_at"].value, # ty: ignore[invalid-argument-type] + variables={ + name: VariableValue( + value=data.value, + summary_type=data.summary_type, + timestamp=current_timestamp, + ) + for name, data in NEW_DATA.items() + }, + ) + + +@pytest.fixture +def subscription_repo(mocker, current_timestamp, csv_fixture_dir): + """A repository registry whose get_latest_runs returns one fresh run.""" + repo = CsvDamnitRepository(PROPOSAL_NO, csv_fixture_dir) + mock_get_latest = mocker.AsyncMock(return_value=[_new_record(current_timestamp)]) + mocker.patch.object(repo, "get_latest_runs", mock_get_latest) + return DamnitRepositoryRegistry(lambda _p: repo) + + +def _sub_ctx(channels_plugin, publisher, repositories): + return SimpleNamespace( + channels=channels_plugin, + run_update_publisher=publisher, + repositories=repositories, + ) + + +_QUERY = """ + subscription LatestDataSubscription( + $proposal: ProposalNo!, + $timestamp: Timestamp!) { + latest_data( + database: { proposal: $proposal }, + timestamp: $timestamp + ) } - assert 348 in snapshot["runs"] - assert snapshot["max_timestamp"] == pytest.approx(1000.0) - run_payload = snapshot["runs"][348] - for variable in run_payload.values(): - assert set(variable.keys()) == {"value", "dtype"} +""" @pytest.mark.asyncio -async def test_poll_proposal_metadata_shape(mock_repositories): - cursors = SubscriptionCursors() - cursors[_PROPOSAL] = 0.0 - repo = mock_repositories.get(_PROPOSAL) +async def test_latest_data( + graphql_schema, + current_timestamp, + subscription_repo, + channels_plugin, +): + publisher = make_publisher(channels_plugin, subscription_repo) + ctx = _sub_ctx(channels_plugin, publisher, subscription_repo) - snapshot = await poll_proposal(_PROPOSAL, cursors, repo) + subscription = await graphql_schema.subscribe( + _QUERY, + variable_values={ + "proposal": PROPOSAL, + "timestamp": (current_timestamp - 1) * 1000, # before the new row + }, + context_value=ctx, + ) - assert snapshot is not None - metadata = snapshot["metadata"] - assert set(metadata.keys()) == {"runs", "variables", "timestamp"} - assert 348 in metadata["runs"] - # ms-timestamp, matching the metadata query's serialization. - assert metadata["timestamp"] == pytest.approx(1000.0 * 1000) + try: + result = await asyncio.wait_for(anext(subscription), timeout=2) + assert not result.errors + data = { + **KNOWN_DATA, + **NEW_DATA, + "run": DatabaseVariable( + value=NEW_RUN, + damnit_dtype=DamnitType.NUMBER, + ), + } -@pytest.mark.asyncio -async def test_poll_proposal_no_new_data_returns_none(mock_repositories): - """A cursor at/above the newest row yields nothing.""" - cursors = SubscriptionCursors() - cursors[_PROPOSAL] = 1000.0 - repo = mock_repositories.get(_PROPOSAL) + latest_data = result.data["latest_data"] + assert set(latest_data.keys()) == {"runs", "metadata"} + assert latest_data["runs"] == { + NEW_RUN: { + name: { + "value": var.damnit_value, + "dtype": var.damnit_dtype.value, + } + for name, var in data.items() + } + } + + metadata = latest_data["metadata"] + assert set(metadata.keys()) == {"runs", "timestamp", "variables"} + assert metadata["runs"] == [*RUNS, NEW_RUN] + assert metadata["timestamp"] == current_timestamp * 1000 - assert await poll_proposal(_PROPOSAL, cursors, repo) is None + # Variable names, titles and tags come through the metadata snapshot. + for name, var_data in DamnitRun.known_variables().items(): + assert name in metadata["variables"] + assert metadata["variables"][name]["title"] == var_data["title"] + assert metadata["variables"]["etof_settings.ret0"]["tags"] == ["eTOF setting"] + assert metadata["variables"]["etof.eTOF_calibration"]["tags"] == ["eTOF"] + finally: + await subscription.aclose() + await publisher.aclose() -def test_filter_for_client_drops_stale_snapshot(): - snapshot = { - "runs": {348: {"x": {"value": 1, "dtype": "number"}}}, - "run_timestamps": {348: 1000.0}, - "max_timestamp": 1000.0, - "metadata": {"runs": [348], "variables": {}, "timestamp": 1_000_000.0}, +@pytest.mark.asyncio +async def test_concurrent_subscriptions_share_one_poll_task( + graphql_schema, + current_timestamp, + subscription_repo, + channels_plugin, +): + """Coalescing is structural: N channel subscribers, one poll task.""" + publisher = make_publisher(channels_plugin, subscription_repo) + ctx = _sub_ctx(channels_plugin, publisher, subscription_repo) + + variables = { + "proposal": PROPOSAL, + "timestamp": (current_timestamp - 1) * 1000, } - # since >= max_timestamp -> nothing new - assert filter_for_client(snapshot, since=1000.0) is None + first_sub = await graphql_schema.subscribe( + _QUERY, variable_values=variables, context_value=ctx + ) + second_sub = await graphql_schema.subscribe( + _QUERY, variable_values=variables, context_value=ctx + ) + + try: + first = await asyncio.wait_for(anext(first_sub), timeout=2) + second = await asyncio.wait_for(anext(second_sub), timeout=2) + assert not first.errors + assert not second.errors + + # Both subscribers are fed by a single poll task for the proposal. + assert len(publisher._tasks) == 1 + finally: + await first_sub.aclose() + await second_sub.aclose() + await publisher.aclose() + + +@pytest.mark.asyncio +async def test_publisher_failure_terminates_subscription_with_typed_error( + graphql_schema, + current_timestamp, + channels_plugin, +): + """A persistent publisher failure surfaces as a typed error (ADR-001).""" + failing_repo = MagicMock() + failing_repo.get_metadata = AsyncMock(side_effect=OSError("gpfs down")) + registry = DamnitRepositoryRegistry(lambda _p: failing_repo) + publisher = make_publisher(channels_plugin, registry, max_consecutive_failures=2) + ctx = _sub_ctx(channels_plugin, publisher, registry) + subscription = await graphql_schema.subscribe( + _QUERY, + variable_values={ + "proposal": PROPOSAL, + "timestamp": (current_timestamp - 1) * 1000, + }, + context_value=ctx, + ) -def test_filter_for_client_returns_fresh_runs(): - snapshot = { - "runs": {348: {"x": {"value": 1, "dtype": "number"}}}, - "run_timestamps": {348: 1000.0}, - "max_timestamp": 1000.0, - "metadata": {"runs": [348], "variables": {}, "timestamp": 1_000_000.0}, + try: + result = await asyncio.wait_for(anext(subscription), timeout=2) + assert result.errors + assert "unavailable" in result.errors[0].message.lower() + finally: + await subscription.aclose() + await publisher.aclose() + + +# ----------------------------------------------------------------------------- +# filter_for_client + + +def _snapshot(run_timestamps) -> dict: + return { + "runs": {run: {"v": run} for run in run_timestamps}, + "run_timestamps": dict(run_timestamps), + "max_timestamp": max(run_timestamps.values()), + "metadata": {"runs": sorted(run_timestamps), "variables": {}}, } - result = filter_for_client(snapshot, since=500.0) - assert result is not None - assert set(result.keys()) == {"runs", "metadata"} - assert 348 in result["runs"] def test_filter_for_client_none_snapshot(): - assert filter_for_client(None, since=500.0) is None + assert filter_for_client(None, since=0) is None + + +def test_filter_for_client_since_zero_returns_none(): + snapshot = _snapshot({1: 100.0, 2: 200.0}) + assert filter_for_client(snapshot, since=0) is None + + +def test_filter_for_client_excludes_equal_timestamp(): + snapshot = _snapshot({1: 100.0, 2: 200.0}) + result = filter_for_client(snapshot, since=100.0) + assert result is not None + assert set(result["runs"]) == {2} + + +def test_filter_for_client_since_above_all_returns_none(): + snapshot = _snapshot({1: 100.0, 2: 200.0}) + assert filter_for_client(snapshot, since=300.0) is None + + +# ----------------------------------------------------------------------------- +# Authorization + + +@pytest.mark.asyncio +async def test_latest_data_unauthorized(graphql_schema_no_auth, current_timestamp): + gen = await graphql_schema_no_auth.subscribe( + """ + subscription { + latest_data(database: { proposal: 999999 }, timestamp: 0) + } + """, + ) + # Subscription permission failures surface as a PreExecutionError on the + # first iteration rather than immediately from subscribe(). + first = await gen.__anext__() + assert first.errors is not None + assert first.errors[0].message == "Authentication required." + + +@pytest.mark.asyncio +async def test_latest_data_forbidden(graphql_schema_authenticated_non_member): + gen = await graphql_schema_authenticated_non_member.subscribe( + f""" + subscription {{ + latest_data(database: {{ proposal: {PROPOSAL} }}, timestamp: 0) + }} + """, + ) + # Subscription permission failures surface as a PreExecutionError on the + # first iteration rather than immediately from subscribe(). + first = await gen.__anext__() + assert first.errors is not None + assert first.errors[0].message == "Access to this proposal is forbidden." diff --git a/api/tests/refactor/conftest.py b/api/tests/refactor/conftest.py index 5d9e8c98..35fd2ce1 100644 --- a/api/tests/refactor/conftest.py +++ b/api/tests/refactor/conftest.py @@ -12,5 +12,4 @@ graphql_schema_no_auth, mock_repositories, mocked_ensure_damnit_path, - reset_caches, ) diff --git a/api/tests/refactor/test_gql_parity.py b/api/tests/refactor/test_gql_parity.py index ad9d9840..4ee6146c 100644 --- a/api/tests/refactor/test_gql_parity.py +++ b/api/tests/refactor/test_gql_parity.py @@ -132,20 +132,21 @@ async def test_metadata_query_wire_shape_unchanged(graphql_schema): @pytest.mark.asyncio async def test_latest_data_subscription_wire_shape_unchanged(mock_repositories): """The subscription's client payload shape, exercised through the same - poll_proposal/filter_for_client path the resolver drives.""" - from damnit_api.graphql.subscriptions import ( - SubscriptionCursors, - filter_for_client, - poll_proposal, - ) + publisher `_poll`/`filter_for_client` path that feeds the resolver's + channel.""" + from unittest.mock import MagicMock + + from damnit_api.graphql.publisher import SqlitePollingRunUpdatePublisher + from damnit_api.graphql.subscriptions import filter_for_client from damnit_api.shared.models import ProposalNumber proposal = ProposalNumber(PROPOSAL) - cursors = SubscriptionCursors() - cursors[proposal] = 0.0 # seed below the fixture timestamps so a run surfaces - repo = mock_repositories.get(proposal) + publisher = SqlitePollingRunUpdatePublisher( + channels=MagicMock(), repositories=mock_repositories + ) + publisher._cursors[proposal] = 0.0 # seed below the fixture timestamps - snapshot = await poll_proposal(proposal, cursors, repo) + snapshot = await publisher._poll(proposal) result = filter_for_client(snapshot, since=500.0) assert result is not None diff --git a/api/tests/test_state.py b/api/tests/test_state.py index e6bca79b..11019f58 100644 --- a/api/tests/test_state.py +++ b/api/tests/test_state.py @@ -65,6 +65,19 @@ def test_create_mymdc_client_builds_mock_client(tmp_path): assert create_mymdc_client(settings) is not None +def test_create_app_refuses_multiworker_with_process_local_backends(monkeypatch): + """Process-local channels/store backends cannot be shared across workers, + so the composition root refuses to start multi-worker (ADR-009).""" + from damnit_api.main import create_app + from damnit_api.shared import settings as settings_module + + monkeypatch.setattr( + settings_module.settings.uvicorn, "workers", 2, raising=False + ) + with pytest.raises(RuntimeError, match="process-local"): + create_app() + + def test_repository_registry_memoizes_per_proposal(): created = [] diff --git a/api/zensical.toml b/api/zensical.toml index ca4c34aa..a8b28bd9 100644 --- a/api/zensical.toml +++ b/api/zensical.toml @@ -12,6 +12,15 @@ nav = [ {"ADRs" = [ "adr/README.md", "adr/000-vertical-slice-architecture.md", + "adr/001-error-classes.md", + "adr/002-no-global-mutable-state.md", + "adr/003-injected-settings.md", + "adr/004-proposal-path-locator.md", + "adr/005-repository-pattern.md", + "adr/006-litestar.md", + "adr/007-graphql-transport-only.md", + "adr/008-local-mode-composition.md", + "adr/009-channels-subscriptions.md", ]} ] }, { "Development" = [