diff --git a/README.md b/README.md index afe7e81..5e24fbd 100644 --- a/README.md +++ b/README.md @@ -67,7 +67,10 @@ response, state, task, error, and permission dimensions. Database migrations are ordered SQL files applied transactionally and recorded with SHA-256 checksums. Re-running `make db-migrate` is safe; changing an already -applied migration is rejected instead of silently drifting the schema. +applied migration is rejected instead of silently drifting the schema. Readiness +stays unavailable until the latest packaged migration is present; task workers +retry claims and recover automatically when migrations are applied after process +startup. Seed profiles are deterministic and replace the previously seeded simulation state atomically. `small` creates one node, two QEMU guests, one LXC, two storages, an administrator, and completed task history. `medium` creates three diff --git a/app/db/pool.py b/app/db/pool.py index cef40b7..e0d04bf 100644 --- a/app/db/pool.py +++ b/app/db/pool.py @@ -8,6 +8,9 @@ import asyncpg # type: ignore[import-untyped] from asyncpg import Pool from app.config import Settings +from app.db.migrations import load_migrations + +LATEST_SCHEMA_VERSION = max(migration.version for migration in load_migrations()) class Database(Protocol): @@ -62,12 +65,18 @@ class AsyncpgDatabase: await pool.close() async def is_ready(self) -> bool: - """Check that PostgreSQL accepts a trivial query.""" + """Check connectivity and that all packaged migrations are applied.""" if self._pool is None: return False try: - return bool(await self._pool.fetchval("SELECT 1") == 1) + return bool( + await self._pool.fetchval( + """SELECT COALESCE(max(version), 0) >= $1 + FROM schema_migrations""", + LATEST_SCHEMA_VERSION, + ) + ) except asyncpg.PostgresError: return False diff --git a/app/db/repositories/__init__.py b/app/db/repositories/__init__.py new file mode 100644 index 0000000..3d9e325 --- /dev/null +++ b/app/db/repositories/__init__.py @@ -0,0 +1 @@ +"""Typed PostgreSQL repositories for simulation domain state.""" diff --git a/app/db/repositories/resources.py b/app/db/repositories/resources.py new file mode 100644 index 0000000..a293695 --- /dev/null +++ b/app/db/repositories/resources.py @@ -0,0 +1,97 @@ +"""Typed resource persistence with explicit optimistic locking.""" + +from __future__ import annotations + +import json +import uuid +from collections.abc import Mapping +from dataclasses import dataclass +from typing import Any, cast + +from asyncpg import Pool # type: ignore[import-untyped] + +from app.db.primitives import ConflictError, transaction + + +@dataclass(frozen=True, slots=True) +class ResourceRecord: + id: uuid.UUID + cluster_id: uuid.UUID + node: str + kind: str + external_id: str + state: dict[str, Any] + metadata: dict[str, Any] + version: int + + +def _json_object(value: object) -> dict[str, Any]: + if isinstance(value, str): + return cast(dict[str, Any], json.loads(value)) + return dict(cast(Mapping[str, Any], value)) + + +def _record(row: Mapping[str, object]) -> ResourceRecord: + return ResourceRecord( + id=cast(uuid.UUID, row["id"]), + cluster_id=cast(uuid.UUID, row["cluster_id"]), + node=str(row["node"]), + kind=str(row["kind"]), + external_id=str(row["external_id"]), + state=_json_object(row["state"]), + metadata=_json_object(row["metadata"]), + version=int(cast(int, row["version"])), + ) + + +class ResourceRepository: + def __init__(self, pool: Pool) -> None: + self._pool = pool + + async def list( + self, *, kind: str | None = None, node: str | None = None + ) -> list[ResourceRecord]: + rows = await self._pool.fetch( + """SELECT r.id, r.cluster_id, n.name AS node, r.kind, r.external_id, + r.state, r.metadata, r.version + FROM resources r JOIN nodes n ON n.id=r.node_id + WHERE ($1::text IS NULL OR r.kind=$1) + AND ($2::text IS NULL OR n.name=$2) + ORDER BY r.kind, r.external_id""", + kind, + node, + ) + return [_record(row) for row in rows] + + async def get(self, *, kind: str, external_id: str) -> ResourceRecord | None: + row = await self._pool.fetchrow( + """SELECT r.id, r.cluster_id, n.name AS node, r.kind, r.external_id, + r.state, r.metadata, r.version + FROM resources r JOIN nodes n ON n.id=r.node_id + WHERE r.kind=$1 AND r.external_id=$2""", + kind, + external_id, + ) + return None if row is None else _record(row) + + async def update_state( + self, + resource_id: uuid.UUID, + *, + expected_version: int, + state: Mapping[str, object], + ) -> ResourceRecord: + async with transaction(self._pool) as connection: + row = await connection.fetchrow( + """UPDATE resources SET state=$3::jsonb, version=version+1, + updated_at=now() WHERE id=$1 AND version=$2 + RETURNING id, cluster_id, + (SELECT name FROM nodes WHERE id=resources.node_id) AS node, + kind, external_id, state, metadata, version""", + resource_id, + expected_version, + json.dumps(dict(state), sort_keys=True), + ) + if row is None: + raise ConflictError("resource version conflict or resource missing") + return _record(row) diff --git a/docker-compose.yml b/docker-compose.yml index 8bc7a9d..1236883 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -46,7 +46,7 @@ services: image: nginx:1.28.0-alpine depends_on: simulator: - condition: service_healthy + condition: service_started ports: - "8007:8443" volumes: diff --git a/docs/architecture.md b/docs/architecture.md index 11da393..e2ee417 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -197,6 +197,12 @@ The first vertical release deliberately supports a small set of endpoints with complete stateful semantics. All other imported endpoints remain visibly unsupported until their handlers and compatibility tests exist. +Database readiness includes the latest packaged migration version, not merely a +successful connectivity query. Workers tolerate the documented container-first +startup sequence by retrying failed claims until migration tables exist. +Normalized resource writes use compare-and-swap version updates through a typed +repository, so stale writers receive a domain conflict. + ## Deployment model One Uvicorn process runs per container. Horizontal replicas coordinate through diff --git a/docs/original-prompt-gap-plan.md b/docs/original-prompt-gap-plan.md index ddb0c9b..5d5dd8d 100644 --- a/docs/original-prompt-gap-plan.md +++ b/docs/original-prompt-gap-plan.md @@ -29,7 +29,7 @@ live admin report renders the same evidence deterministically. LXC, disks, NICs, snapshots, backups, pools, users/groups/roles/ACLs/tokens and observed contracts. Preserve version, metadata, timestamps, and cluster-wide VMID uniqueness. -- [ ] Make migration readiness explicit so workers cannot become permanently +- [x] Make migration readiness explicit so workers cannot become permanently unhealthy before schema creation. - [x] Match the required `small` profile (one node, two QEMU, one LXC, two storages, administrator, completed tasks). diff --git a/tests/integration/test_migrations.py b/tests/integration/test_migrations.py index fdd215a..c63a166 100644 --- a/tests/integration/test_migrations.py +++ b/tests/integration/test_migrations.py @@ -6,7 +6,11 @@ import uuid import asyncpg # type: ignore[import-untyped] import pytest +from app.config import Settings from app.db.migrations import migrate +from app.db.pool import AsyncpgDatabase +from app.db.primitives import ConflictError +from app.db.repositories.resources import ResourceRepository from app.simulation.seed import apply_seed, small_profile pytestmark = [ @@ -58,3 +62,34 @@ async def test_small_seed_is_idempotent() -> None: assert await connection.fetchval("SELECT count(*) FROM storage_contents") == 4 finally: await connection.close() + + +async def test_schema_readiness_and_optimistic_resource_repository() -> None: + url = os.environ["TEST_DATABASE_URL"] + connection = await asyncpg.connect(url) + database = AsyncpgDatabase(Settings(database_url=url)) + try: + await migrate(connection) + await apply_seed(connection, small_profile()) + await database.connect() + assert await database.is_ready() + + repository = ResourceRepository(database.pool) + resource = await repository.get(kind="qemu", external_id="101") + assert resource is not None + updated = await repository.update_state( + resource.id, + expected_version=resource.version, + state={**resource.state, "status": "running"}, + ) + assert updated.version == resource.version + 1 + assert updated.state["status"] == "running" + with pytest.raises(ConflictError): + await repository.update_state( + resource.id, + expected_version=resource.version, + state=resource.state, + ) + finally: + await database.close() + await connection.close()