diff --git a/docs/ADR/004-postgresql-durable-local-control-plane.md b/docs/ADR/004-postgresql-durable-local-control-plane.md new file mode 100644 index 0000000..72e0a00 --- /dev/null +++ b/docs/ADR/004-postgresql-durable-local-control-plane.md @@ -0,0 +1,39 @@ +# ADR 004: PostgreSQL-backed durable local control plane + +## Status + +Accepted for v0.4. + +## Decision + +Ordinary PostgreSQL is the production reference state provider. Repository contracts remain +provider-neutral and a SQLite DB-API implementation supports hermetic local operation and contract +tests. PostgreSQL JSONB stores documented JSON-compatible representations; it is not a canonical +world model. Explicit tables partition desired revisions, authoritative decisions, observations, +plans, runs, operations, attempts, transitions, evidence, checkpoints, leases, reconciliation, and +drift. Secret material is not accepted by checkpoints or ordinary evidence. + +Events and logs may notify or diagnose, but never replace current projections and append-only +operation history. Desired declarations are not authoritative decisions; authoritative decisions +are not observations; observations are not provider results. + +Provider calls never occur inside a database transaction. The executor first commits `running`, +calls the provider, then commits its result and next transition. Recovery treats an interruption +between those transactions as uncertain: it observes before policy permits retry. Durable scoped +idempotency keys reject conflicting content and prevent a verified effect being applied twice. + +World, run, resource, and maintenance leases are explicit rows. Atomic acquisition, expiry, +owner-checked release, and monotonically increasing fencing tokens allow safe stale-lock recovery +and inspection. These are local-process coordination semantics, not distributed leadership. + +Checkpoints are immutable logical journal summaries made before execution, after verified batches, +before compensation, and at finalisation. They are not infrastructure snapshots. Resume rejects a +different executable-plan fingerprint or schema version. + +## Consequences + +The transaction gap is visible and recoverable rather than hidden. PostgreSQL backup and restore +preserves the entire schema, while metadata export is diagnostic only and excludes secrets. v0.4 is +strictly single-node: HA, consensus, leader election, remote agents, operational providers, and +continuous reconciliation are deferred. v0.5 supplies the first authoritative naming provider +slice; it must consume these seams rather than alter their domain meaning. diff --git a/docs/durable-control-plane-v0.4.md b/docs/durable-control-plane-v0.4.md new file mode 100644 index 0000000..b187a18 --- /dev/null +++ b/docs/durable-control-plane-v0.4.md @@ -0,0 +1,30 @@ +# NetEngine v0.4 durable local control plane + +v0.4 turns the v0.3 provider-neutral runtime into a restart-safe, repeatedly operable single-node +control plane. `controlplane.models` defines immutable desired, authoritative, observed, checkpoint, +lease, reconciliation, and drift records. `ControlPlaneRepository` is independent of SQL dialect; +the PostgreSQL migration is the durable production schema and SQLite provides a local/test adapter. + +Acceptance stores an immutable revision and changes the active pointer in one transaction. Rejected +admission is retained but never activated. Observations are versioned independently from apply +results. Idempotency is unique per world and content conflicts fail closed. Provider calls occur +between transactions; recovery observes any uncertain operation before retry. + +Leases cover worlds, execution runs, resources, and maintenance. Expiry permits recovery with a +higher fencing token; only the owner releases an active lease. Checkpoints are logical journal +summaries, not infrastructure snapshots, and reject an incompatible plan fingerprint or schema. + +Drift comparison deterministically classifies missing, unexpected, changed, unhealthy, +unverifiable, stale, unavailable, and conformant resources. Reconciliation is explicitly invoked +and bounded: observe, compare, plan, compile, execute, observe, assess. A conformant repeat is a +no-op; v0.4 has no daemon loop. + +Production uses `NETENGINE_CONTROL_PLANE_DATABASE_URL`; local CLI tests may use +`NETENGINE_CONTROL_PLANE_PATH`. Back up with +`pg_dump --schema=netengine_control --format=custom DATABASE > control-plane.dump` and restore with +`pg_restore --dbname=DATABASE control-plane.dump`. This preserves the v0.4 schema but is neither +world portability nor a secret or infrastructure backup. + +Legacy JSON phase state, queues, exports, and logs remain intact and are not live workflow state. +Real DNS, trust, identity, routing, workload, and public providers are deferred to v0.5 and later, +as are HA, distributed locks, leader election, remote agents, and polished backup orchestration. diff --git a/docs/roadmap.md b/docs/roadmap.md index a0d4710..7396040 100644 --- a/docs/roadmap.md +++ b/docs/roadmap.md @@ -1,8 +1,8 @@ # NetSovereign roadmap -Implementation status: v0.1, v0.2, and the bounded in-memory v0.3 runtime core are -implemented and tested. v0.4 durability and v0.5 naming materialisation remain -planned and are not implied by the v0.3 interfaces. +Implementation status: v0.1 through v0.4 are implemented and tested. v0.4 supplies the +bounded, single-node durable control-plane seams and PostgreSQL schema; v0.5 naming +materialisation remains planned and is not implied by those interfaces. NetSovereign develops authority and intent before operational adapters. Versions v0.2 through v0.4 are a single dependency chain: real DNS, PKI, identity, or gateway providers must not begin until diff --git a/migrations/004_durable_control_plane.sql b/migrations/004_durable_control_plane.sql new file mode 100644 index 0000000..070821a --- /dev/null +++ b/migrations/004_durable_control_plane.sql @@ -0,0 +1,56 @@ +-- NetEngine v0.4 durable local control plane (ordinary PostgreSQL 15+). +-- Each partition is explicit; JSONB is a representation, not the domain model. +BEGIN; +CREATE SCHEMA IF NOT EXISTS netengine_control; +CREATE TABLE IF NOT EXISTS netengine_control.schema_version( + version integer PRIMARY KEY, applied_at timestamptz NOT NULL DEFAULT now()); +DO $$ +BEGIN + IF EXISTS (SELECT 1 FROM netengine_control.schema_version WHERE version > 4) THEN + RAISE EXCEPTION 'incompatible control-plane schema version'; + END IF; +END $$; +CREATE TABLE IF NOT EXISTS netengine_control.desired_revisions( + revision_id text PRIMARY KEY, world_id text NOT NULL, parent_revision_id text, + schema_version text NOT NULL, declaration_digest text NOT NULL, + declaration jsonb NOT NULL, admission jsonb NOT NULL, accepted_at timestamptz NOT NULL, + actor text, authority_manifest_digest text NOT NULL, boundary_policy_digest text, + compatibility jsonb NOT NULL DEFAULT '{}', status text NOT NULL, + FOREIGN KEY(parent_revision_id) REFERENCES netengine_control.desired_revisions(revision_id), + UNIQUE(world_id,declaration_digest)); +CREATE TABLE IF NOT EXISTS netengine_control.active_desired_revisions( + world_id text PRIMARY KEY, revision_id text NOT NULL REFERENCES netengine_control.desired_revisions); +CREATE TABLE IF NOT EXISTS netengine_control.authoritative_records( + record_id text PRIMARY KEY, world_id text NOT NULL, kind text NOT NULL, authority_id text NOT NULL, + mandate_id text, desired_revision_id text NOT NULL REFERENCES netengine_control.desired_revisions, + decision_reference text NOT NULL, status text NOT NULL, version integer NOT NULL CHECK(version>0), + lifecycle jsonb NOT NULL, value jsonb NOT NULL, recorded_at timestamptz NOT NULL, + UNIQUE(world_id,kind,record_id,version)); +CREATE TABLE IF NOT EXISTS netengine_control.observations( + observation_id text PRIMARY KEY, world_id text NOT NULL, resource_id text NOT NULL, + provider_id text NOT NULL, binding_id text NOT NULL, capability text NOT NULL, + provider_resource_id text, observation_type text NOT NULL, representation jsonb, + digest text, health text NOT NULL, conformance text NOT NULL, observed_at timestamptz NOT NULL, + stale_after timestamptz, execution_run_id text, operation_id text); +CREATE INDEX IF NOT EXISTS observations_latest ON netengine_control.observations(world_id,resource_id,observed_at DESC); +CREATE TABLE IF NOT EXISTS netengine_control.execution_plans(plan_id text PRIMARY KEY, fingerprint text NOT NULL UNIQUE, payload jsonb NOT NULL); +CREATE TABLE IF NOT EXISTS netengine_control.execution_runs(run_id text PRIMARY KEY, plan_id text NOT NULL REFERENCES netengine_control.execution_plans, world_id text NOT NULL, current_state text NOT NULL, payload jsonb NOT NULL); +CREATE TABLE IF NOT EXISTS netengine_control.operations(operation_id text PRIMARY KEY, run_id text NOT NULL REFERENCES netengine_control.execution_runs, world_id text NOT NULL, idempotency_key text NOT NULL, content_digest text NOT NULL, current_state text NOT NULL, payload jsonb NOT NULL, UNIQUE(world_id,idempotency_key)); +CREATE TABLE IF NOT EXISTS netengine_control.attempts(attempt_id text PRIMARY KEY, operation_id text NOT NULL REFERENCES netengine_control.operations, number integer NOT NULL, payload jsonb NOT NULL, UNIQUE(operation_id,number)); +CREATE TABLE IF NOT EXISTS netengine_control.transitions(transition_id text PRIMARY KEY, operation_id text NOT NULL REFERENCES netengine_control.operations, sequence integer NOT NULL, at timestamptz NOT NULL, payload jsonb NOT NULL, UNIQUE(operation_id,sequence)); +CREATE TABLE IF NOT EXISTS netengine_control.evidence(evidence_id text PRIMARY KEY, run_id text NOT NULL REFERENCES netengine_control.execution_runs, operation_id text REFERENCES netengine_control.operations, created_at timestamptz NOT NULL, payload jsonb NOT NULL); +CREATE TABLE IF NOT EXISTS netengine_control.checkpoints(checkpoint_id text PRIMARY KEY, run_id text NOT NULL REFERENCES netengine_control.execution_runs, world_id text NOT NULL, desired_revision_id text NOT NULL REFERENCES netengine_control.desired_revisions, previous_checkpoint_id text REFERENCES netengine_control.checkpoints, schema_version integer NOT NULL, plan_fingerprint text NOT NULL, created_at timestamptz NOT NULL, payload jsonb NOT NULL); +CREATE TABLE IF NOT EXISTS netengine_control.leases(lock_key text PRIMARY KEY, owner text NOT NULL, acquired_at timestamptz NOT NULL, expires_at timestamptz NOT NULL, fencing_token bigint NOT NULL CHECK(fencing_token > 0), renewed_at timestamptz, released_at timestamptz, release_reason text, payload jsonb NOT NULL, CHECK(expires_at >= acquired_at)); +CREATE TABLE IF NOT EXISTS netengine_control.reconciliations(reconciliation_id text PRIMARY KEY, world_id text NOT NULL, desired_revision_id text NOT NULL REFERENCES netengine_control.desired_revisions, trigger text NOT NULL, started_at timestamptz NOT NULL, completed_at timestamptz, result text NOT NULL, payload jsonb NOT NULL); +CREATE TABLE IF NOT EXISTS netengine_control.drift(drift_id text PRIMARY KEY, reconciliation_id text REFERENCES netengine_control.reconciliations, world_id text NOT NULL, resource_id text NOT NULL, classification text NOT NULL, status text NOT NULL, first_detected_at timestamptz NOT NULL, last_detected_at timestamptz NOT NULL, payload jsonb NOT NULL); +DO $$ +BEGIN + IF NOT EXISTS (SELECT 1 FROM pg_constraint WHERE conname = 'observations_execution_run_fk' AND connamespace = 'netengine_control'::regnamespace) THEN + ALTER TABLE netengine_control.observations ADD CONSTRAINT observations_execution_run_fk FOREIGN KEY(execution_run_id) REFERENCES netengine_control.execution_runs(run_id); + END IF; + IF NOT EXISTS (SELECT 1 FROM pg_constraint WHERE conname = 'observations_operation_fk' AND connamespace = 'netengine_control'::regnamespace) THEN + ALTER TABLE netengine_control.observations ADD CONSTRAINT observations_operation_fk FOREIGN KEY(operation_id) REFERENCES netengine_control.operations(operation_id); + END IF; +END $$; +INSERT INTO netengine_control.schema_version(version) VALUES(4) ON CONFLICT DO NOTHING; +COMMIT; diff --git a/pyproject.toml b/pyproject.toml index 732b442..cc17f3c 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "netsovereign" -version = "0.3.0" +version = "0.4.0" description = "Provider-neutral compiler and reconciliation runtime for sovereign worlds" readme = "README.md" requires-python = ">=3.12" diff --git a/src/netsovereign/__init__.py b/src/netsovereign/__init__.py index 36f6936..d0d529a 100644 --- a/src/netsovereign/__init__.py +++ b/src/netsovereign/__init__.py @@ -31,4 +31,4 @@ "compare_worlds", "validate_spec", ] -__version__ = "0.2.0" +__version__ = "0.4.0" diff --git a/src/netsovereign/cli.py b/src/netsovereign/cli.py index 1ea4bb3..e5d6419 100644 --- a/src/netsovereign/cli.py +++ b/src/netsovereign/cli.py @@ -12,6 +12,7 @@ import yaml from pydantic import BaseModel, ValidationError +from .controlplane.repository import SQLiteControlPlaneRepository from .io import load_spec from .manifest import build_manifest, explain_manifest from .planning import ( @@ -30,6 +31,23 @@ from .validation import has_errors, validate_spec app = typer.Typer(no_args_is_help=True, help="Validate and explain sovereign world intent.") +control_plane_app = typer.Typer(help="Operate the durable local control plane.") +app.add_typer(control_plane_app, name="control-plane") + + +@control_plane_app.command("init") +def control_plane_init( + database: Annotated[Path, typer.Option(envvar="NETENGINE_CONTROL_PLANE_PATH")] = Path( + ".netengine-control.db" + ), +) -> None: + """Initialise or inspect a local durable control-plane schema.""" + repository = SQLiteControlPlaneRepository(database) + row = repository.connection.execute( + "SELECT value FROM cp_metadata WHERE key='schema_version'" + ).fetchone() + repository.connection.close() + _emit({"database": str(database), "schema_version": int(row[0]), "status": "ready"}) def _parse(path: Path) -> WorldSpec: diff --git a/src/netsovereign/controlplane/__init__.py b/src/netsovereign/controlplane/__init__.py new file mode 100644 index 0000000..f414692 --- /dev/null +++ b/src/netsovereign/controlplane/__init__.py @@ -0,0 +1,12 @@ +"""Durable, single-node control-plane contracts and services.""" + +from .models import * # noqa: F403 +from .repository import ControlPlaneRepository, SQLiteControlPlaneRepository +from .service import ControlPlaneService, DriftDetector + +__all__ = [ + "ControlPlaneRepository", + "SQLiteControlPlaneRepository", + "ControlPlaneService", + "DriftDetector", +] diff --git a/src/netsovereign/controlplane/models.py b/src/netsovereign/controlplane/models.py new file mode 100644 index 0000000..8b3f5ed --- /dev/null +++ b/src/netsovereign/controlplane/models.py @@ -0,0 +1,188 @@ +"""Provider-neutral durable control-plane records. + +These records intentionally do not import a database or provider implementation. +""" + +from __future__ import annotations + +from datetime import UTC, datetime +from enum import StrEnum +from typing import Any + +from pydantic import Field, field_validator, model_validator + +from ..base import DomainModel + +CONTROL_PLANE_SCHEMA_VERSION = 4 +_SECRET_TERMS = ("password", "secret", "token", "private_key", "credential") + + +def now_utc() -> datetime: + return datetime.now(UTC) + + +def _contains_secret(value: Any) -> bool: + if isinstance(value, dict): + return any( + any(term in str(key).lower() for term in _SECRET_TERMS) or _contains_secret(item) + for key, item in value.items() + ) + if isinstance(value, list): + return any(_contains_secret(item) for item in value) + return False + + +def _require_aware(value: datetime | None) -> datetime | None: + if value is None: + return None + if value.tzinfo is None or value.utcoffset() is None: + raise ValueError("control-plane timestamps must be timezone-aware") + return value + + +class DesiredRevision(DomainModel): + world_id: str + revision_id: str + parent_revision_id: str | None = None + schema_version: str + declaration_digest: str + declaration: dict[str, Any] + accepted_at: datetime + admission: dict[str, Any] + actor: str | None = None + authority_manifest_digest: str + boundary_policy_digest: str | None = None + compatibility: dict[str, Any] = Field(default_factory=dict) + status: str = "accepted" + + _accepted_at_is_aware = field_validator("accepted_at")(_require_aware) + + +class AuthoritativeRecord(DomainModel): + record_id: str + world_id: str + kind: str + authority_id: str + mandate_id: str | None = None + desired_revision_id: str + decision_reference: str + status: str + version: int = Field(ge=1) + lifecycle: dict[str, Any] = Field(default_factory=dict) + value: dict[str, Any] = Field(default_factory=dict) + recorded_at: datetime + + +class ObservedRecord(DomainModel): + observation_id: str + world_id: str + resource_id: str + provider_id: str + binding_id: str + capability: str + provider_resource_id: str | None = None + observation_type: str + representation: Any = None + digest: str | None = None + health: str = "unknown" + conformance: str = "unknown" + observed_at: datetime + stale_after: datetime | None = None + execution_run_id: str | None = None + operation_id: str | None = None + + _timestamps_are_aware = field_validator("observed_at", "stale_after")(_require_aware) + + +class LockLease(DomainModel): + key: str + owner: str + acquired_at: datetime + expires_at: datetime + fencing_token: int + renewed_at: datetime | None = None + released_at: datetime | None = None + release_reason: str | None = None + + _timestamps_are_aware = field_validator( + "acquired_at", "expires_at", "renewed_at", "released_at" + )(_require_aware) + + @model_validator(mode="after") + def validate_interval(self) -> LockLease: + if self.expires_at < self.acquired_at: + raise ValueError("lease expiry cannot precede acquisition") + return self + + +class Checkpoint(DomainModel): + checkpoint_id: str + world_id: str + desired_revision_id: str + plan_fingerprint: str + run_id: str + completed_operations: list[str] + incomplete_operations: list[str] + latest_observation_ids: list[str] + compensation: dict[str, Any] = Field(default_factory=dict) + created_at: datetime + reason: str + schema_version: int = CONTROL_PLANE_SCHEMA_VERSION + previous_checkpoint_id: str | None = None + + _created_at_is_aware = field_validator("created_at")(_require_aware) + + @model_validator(mode="after") + def reject_secrets(self) -> Checkpoint: + if _contains_secret(self.compensation): + raise ValueError("checkpoint compensation state must not contain secrets") + return self + + +class ReconciliationRecord(DomainModel): + reconciliation_id: str + world_id: str + trigger: str + desired_revision_id: str + observed_snapshot_id: str | None = None + admission: dict[str, Any] = Field(default_factory=dict) + plan: dict[str, Any] = Field(default_factory=dict) + executable_plan_id: str | None = None + execution_run_id: str | None = None + started_at: datetime + completed_at: datetime | None = None + result: str = "running" + drift_summary: dict[str, int] = Field(default_factory=dict) + conformance_summary: dict[str, Any] = Field(default_factory=dict) + + _timestamps_are_aware = field_validator("started_at", "completed_at")(_require_aware) + + +class DriftClassification(StrEnum): + MISSING = "missing_resource" + UNEXPECTED = "unexpected_resource" + CHANGED = "changed_resource" + UNHEALTHY = "unhealthy_resource" + UNVERIFIABLE = "unverifiable_resource" + STALE = "stale_observation" + PROVIDER_UNAVAILABLE = "provider_unavailable" + CONFORMANT = "conformant_resource" + + +class DriftRecord(DomainModel): + drift_id: str + world_id: str + resource_id: str + desired_revision_id: str + expected_digest: str | None + observed_digest: str | None + provider_id: str + capability: str + classification: DriftClassification + severity: str + first_detected_at: datetime + last_detected_at: datetime + status: str = "open" + reconciliation_id: str | None = None + + _timestamps_are_aware = field_validator("first_detected_at", "last_detected_at")(_require_aware) diff --git a/src/netsovereign/controlplane/repository.py b/src/netsovereign/controlplane/repository.py new file mode 100644 index 0000000..f368a4a --- /dev/null +++ b/src/netsovereign/controlplane/repository.py @@ -0,0 +1,213 @@ +"""Durable repository interfaces and the restart-safe local implementation.""" + +from __future__ import annotations + +import builtins +import json +import sqlite3 +from collections.abc import Iterator +from contextlib import AbstractContextManager, contextmanager +from datetime import datetime +from pathlib import Path +from typing import Protocol, TypeVar + +from pydantic import BaseModel + +from .models import ( + CONTROL_PLANE_SCHEMA_VERSION, + Checkpoint, + DesiredRevision, + LockLease, + ReconciliationRecord, +) + +T = TypeVar("T", bound=BaseModel) + + +class ControlPlaneRepository(Protocol): + """Storage contract; callers never depend on PostgreSQL-specific concepts.""" + + def put_immutable(self, partition: str, key: str, value: BaseModel) -> None: ... + def get(self, partition: str, key: str, model: type[T]) -> T | None: ... + def list(self, partition: str, model: type[T]) -> list[T]: ... + def set_active_revision(self, world_id: str, revision_id: str) -> None: ... + def active_revision(self, world_id: str) -> DesiredRevision | None: ... + def acquire_lock(self, lease: LockLease) -> LockLease | None: ... + def release_lock(self, key: str, owner: str, reason: str, at: datetime) -> bool: ... + def transaction(self) -> AbstractContextManager[None]: ... + def incomplete_reconciliations(self) -> builtins.list[ReconciliationRecord]: ... + + +class SQLiteControlPlaneRepository: + """DB-API reference used locally/tests; PostgreSQL uses the same partition contract. + + SQLite is deliberately not presented as the production reference. It makes the + CLI restart-safe without requiring a server and exercises transaction semantics. + """ + + SCHEMA_VERSION = CONTROL_PLANE_SCHEMA_VERSION + + def __init__(self, path: Path | str): + self.path = str(path) + self.connection = sqlite3.connect(self.path) + self.connection.execute("PRAGMA foreign_keys=ON") + self.migrate() + + def migrate(self) -> None: + self.connection.executescript( + """ + CREATE TABLE IF NOT EXISTS cp_metadata(key TEXT PRIMARY KEY, value TEXT NOT NULL); + CREATE TABLE IF NOT EXISTS cp_records( + partition TEXT NOT NULL, key TEXT NOT NULL, payload TEXT NOT NULL, + created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP, + PRIMARY KEY(partition,key)); + CREATE TABLE IF NOT EXISTS cp_active_revisions( + world_id TEXT PRIMARY KEY, revision_id TEXT NOT NULL); + CREATE TABLE IF NOT EXISTS cp_locks( + lock_key TEXT PRIMARY KEY, owner TEXT NOT NULL, acquired_at TEXT NOT NULL, + expires_at TEXT NOT NULL, fencing_token INTEGER NOT NULL, + payload TEXT NOT NULL); + """ + ) + row = self.connection.execute( + "SELECT value FROM cp_metadata WHERE key='schema_version'" + ).fetchone() + if row: + installed = int(row[0]) + if installed > self.SCHEMA_VERSION: + raise RuntimeError(f"incompatible control-plane schema version {installed}") + if installed < self.SCHEMA_VERSION: + self.connection.execute( + "UPDATE cp_metadata SET value=? WHERE key='schema_version'", + (str(self.SCHEMA_VERSION),), + ) + else: + self.connection.execute( + "INSERT INTO cp_metadata VALUES('schema_version',?)", + (str(self.SCHEMA_VERSION),), + ) + self.connection.commit() + + @contextmanager + def transaction(self) -> Iterator[None]: + try: + self.connection.execute("BEGIN IMMEDIATE") + yield + self.connection.commit() + except BaseException: + self.connection.rollback() + raise + + def put_immutable(self, partition: str, key: str, value: BaseModel) -> None: + payload = value.model_dump_json() + existing = self.connection.execute( + "SELECT payload FROM cp_records WHERE partition=? AND key=?", (partition, key) + ).fetchone() + if existing: + if json.loads(existing[0]) != json.loads(payload): + raise ValueError(f"immutable record conflict: {partition}/{key}") + return + self.connection.execute( + "INSERT INTO cp_records VALUES(?,?,?,CURRENT_TIMESTAMP)", (partition, key, payload) + ) + + def get(self, partition: str, key: str, model: type[T]) -> T | None: + row = self.connection.execute( + "SELECT payload FROM cp_records WHERE partition=? AND key=?", (partition, key) + ).fetchone() + return model.model_validate_json(row[0]) if row else None + + def list(self, partition: str, model: type[T]) -> list[T]: + rows = self.connection.execute( + "SELECT payload FROM cp_records WHERE partition=? ORDER BY key", (partition,) + ).fetchall() + return [model.model_validate_json(row[0]) for row in rows] + + def set_active_revision(self, world_id: str, revision_id: str) -> None: + revision = self.get("desired", revision_id, DesiredRevision) + if revision is None: + raise KeyError(revision_id) + if revision.world_id != world_id: + raise ValueError("cannot activate a desired revision for a different world") + self.connection.execute( + "INSERT INTO cp_active_revisions VALUES(?,?) ON CONFLICT(world_id) DO UPDATE SET revision_id=excluded.revision_id", + (world_id, revision_id), + ) + + def active_revision(self, world_id: str) -> DesiredRevision | None: + row = self.connection.execute( + "SELECT revision_id FROM cp_active_revisions WHERE world_id=?", (world_id,) + ).fetchone() + return self.get("desired", row[0], DesiredRevision) if row else None + + def acquire_lock(self, lease: LockLease) -> LockLease | None: + with self.transaction(): + row = self.connection.execute( + "SELECT owner,expires_at,fencing_token FROM cp_locks WHERE lock_key=?", (lease.key,) + ).fetchone() + if row and datetime.fromisoformat(row[1]) > lease.acquired_at and row[0] != lease.owner: + return None + if row and datetime.fromisoformat(row[1]) > lease.acquired_at: + return LockLease.model_validate_json( + self.connection.execute( + "SELECT payload FROM cp_locks WHERE lock_key=?", (lease.key,) + ).fetchone()[0] + ) + token = (int(row[2]) + 1) if row else 1 + granted = lease.model_copy(update={"fencing_token": token}) + self.connection.execute( + "INSERT OR REPLACE INTO cp_locks VALUES(?,?,?,?,?,?)", + ( + granted.key, + granted.owner, + granted.acquired_at.isoformat(), + granted.expires_at.isoformat(), + token, + granted.model_dump_json(), + ), + ) + return granted + + def release_lock(self, key: str, owner: str, reason: str, at: datetime) -> bool: + with self.transaction(): + row = self.connection.execute( + "SELECT owner,fencing_token,payload FROM cp_locks WHERE lock_key=?", (key,) + ).fetchone() + if not row or row[0] != owner: + return False + current = LockLease.model_validate_json(row[2]) + if current.released_at is not None: + return False + released = current.model_copy( + update={ + "expires_at": at, + "released_at": at, + "release_reason": reason, + } + ) + # Retain the row as the per-key fencing counter. A subsequent acquire + # replaces the released lease but increments its preserved generation. + self.connection.execute( + "UPDATE cp_locks SET expires_at=?,payload=? WHERE lock_key=?", + (at.isoformat(), released.model_dump_json(), key), + ) + self.put_immutable( + "lock_history", + f"{key}:{at.isoformat()}", + released, + ) + return True + + def latest_checkpoint(self, run_id: str, fingerprint: str) -> Checkpoint | None: + matches = [c for c in self.list("checkpoint", Checkpoint) if c.run_id == run_id] + if not matches: + return None + latest = max(matches, key=lambda item: item.created_at) + if latest.plan_fingerprint != fingerprint or latest.schema_version != self.SCHEMA_VERSION: + raise RuntimeError("checkpoint is incompatible with the executable plan or schema") + return latest + + def incomplete_reconciliations(self) -> builtins.list[ReconciliationRecord]: + return [ + r for r in self.list("reconciliation", ReconciliationRecord) if r.completed_at is None + ] diff --git a/src/netsovereign/controlplane/service.py b/src/netsovereign/controlplane/service.py new file mode 100644 index 0000000..4e0f185 --- /dev/null +++ b/src/netsovereign/controlplane/service.py @@ -0,0 +1,210 @@ +"""Focused revision, drift, reconciliation, and recovery coordination.""" + +from __future__ import annotations + +import hashlib +import json +from datetime import datetime, timedelta +from typing import Any + +from .models import ( + DesiredRevision, + DriftClassification, + DriftRecord, + LockLease, + ObservedRecord, +) +from .repository import ControlPlaneRepository + + +def _digest(value: Any) -> str: + canonical = json.dumps(value, sort_keys=True, separators=(",", ":"), default=str) + return hashlib.sha256(canonical.encode()).hexdigest() + + +class DriftDetector: + """Deterministically compares provider-neutral expected and observed resources.""" + + def compare( + self, + revision: DesiredRevision, + expected: dict[str, dict[str, Any]], + observed: list[ObservedRecord], + at: datetime, + ) -> list[DriftRecord]: + latest: dict[str, ObservedRecord] = {} + for item in sorted(observed, key=lambda row: (row.observed_at, row.observation_id)): + latest[item.resource_id] = item + records: list[DriftRecord] = [] + for resource_id in sorted(set(expected) | set(latest)): + wanted, actual = expected.get(resource_id), latest.get(resource_id) + expected_digest = _digest(wanted) if wanted is not None else None + if actual is None: + # No observation is not proof that a provider resource is absent. + classification = DriftClassification.UNVERIFIABLE + provider, capability, observed_digest = "unbound", "unknown", None + elif wanted is None: + classification = DriftClassification.UNEXPECTED + provider, capability, observed_digest = ( + actual.provider_id, + actual.capability, + actual.digest, + ) + elif actual.stale_after and actual.stale_after <= at: + classification = DriftClassification.STALE + provider, capability, observed_digest = ( + actual.provider_id, + actual.capability, + actual.digest, + ) + elif actual.health == "unavailable": + classification = DriftClassification.PROVIDER_UNAVAILABLE + provider, capability, observed_digest = ( + actual.provider_id, + actual.capability, + actual.digest, + ) + elif actual.observation_type == "absent": + classification = DriftClassification.MISSING + provider, capability, observed_digest = actual.provider_id, actual.capability, None + elif actual.health == "unhealthy": + classification = DriftClassification.UNHEALTHY + provider, capability, observed_digest = ( + actual.provider_id, + actual.capability, + actual.digest, + ) + elif actual.digest is None: + classification = DriftClassification.UNVERIFIABLE + provider, capability, observed_digest = actual.provider_id, actual.capability, None + elif actual.digest != expected_digest: + classification = DriftClassification.CHANGED + provider, capability, observed_digest = ( + actual.provider_id, + actual.capability, + actual.digest, + ) + else: + classification = DriftClassification.CONFORMANT + provider, capability, observed_digest = ( + actual.provider_id, + actual.capability, + actual.digest, + ) + # Each detection is an immutable journal event. Including the observation + # time preserves repeated detections instead of conflicting with a prior + # event for the same resource/classification pair. + identity = _digest( + [ + revision.revision_id, + resource_id, + classification, + expected_digest, + observed_digest, + provider, + capability, + at.isoformat(), + ] + )[:24] + records.append( + DriftRecord( + drift_id=f"drift-{identity}", + world_id=revision.world_id, + resource_id=resource_id, + desired_revision_id=revision.revision_id, + expected_digest=expected_digest, + observed_digest=observed_digest, + provider_id=provider, + capability=capability, + classification=classification, + severity="info" + if classification == DriftClassification.CONFORMANT + else "warning", + first_detected_at=at, + last_detected_at=at, + ) + ) + return records + + +class ControlPlaneService: + """Small façade; provider calls remain outside repository transactions.""" + + def __init__(self, repository: ControlPlaneRepository): + self.repository = repository + self.drift = DriftDetector() + + def accept_revision(self, revision: DesiredRevision) -> DesiredRevision: + if revision.status != "accepted" or revision.admission.get("status") == "rejected": + with self.repository.transaction(): + self.repository.put_immutable("desired", revision.revision_id, revision) + return revision + with self.repository.transaction(): + # The lineage check and active-pointer update share the write lock. This + # prevents concurrent sibling acceptance from becoming last-writer-wins. + active = self.repository.active_revision(revision.world_id) + if active: + if revision.revision_id == active.revision_id: + self.repository.put_immutable("desired", revision.revision_id, revision) + return revision + if revision.parent_revision_id != active.revision_id: + raise ValueError("desired revision does not descend from the active revision") + elif revision.parent_revision_id is not None: + raise ValueError("initial desired revision cannot name a parent") + self.repository.put_immutable("desired", revision.revision_id, revision) + self.repository.set_active_revision(revision.world_id, revision.revision_id) + return revision + + def inspect_drift(self, world_id: str, at: datetime) -> list[DriftRecord]: + revision = self.repository.active_revision(world_id) + if revision is None: + raise KeyError(f"no active desired revision for {world_id}") + declared_resources = revision.declaration.get("resources", {}) + if isinstance(declared_resources, list): + expected: dict[str, dict[str, Any]] = {} + for resource in declared_resources: + if not isinstance(resource, dict) or not isinstance(resource.get("id"), str): + raise ValueError("each declared resource must be an object with a string id") + resource_id = resource["id"] + if resource_id in expected: + raise ValueError(f"duplicate declared resource id: {resource_id}") + expected[resource_id] = resource + elif isinstance(declared_resources, dict): + if not all( + isinstance(key, str) and isinstance(value, dict) + for key, value in declared_resources.items() + ): + raise ValueError("resource mappings require string IDs and object values") + expected = declared_resources + else: + raise ValueError("declared resources must be a list or resource-id mapping") + observed = [ + o for o in self.repository.list("observed", ObservedRecord) if o.world_id == world_id + ] + records = self.drift.compare(revision, expected, observed, at) + with self.repository.transaction(): + for item in records: + self.repository.put_immutable("drift", item.drift_id, item) + return records + + def acquire_world( + self, world_id: str, owner: str, at: datetime, seconds: int = 30 + ) -> LockLease: + lease = LockLease( + key=f"world:{world_id}", + owner=owner, + acquired_at=at, + expires_at=at + timedelta(seconds=seconds), + fencing_token=0, + ) + granted = self.repository.acquire_lock(lease) + if granted is None: + raise RuntimeError(f"world {world_id} is already being reconciled") + return granted + + def recovery_report(self) -> dict[str, Any]: + runs = self.repository.incomplete_reconciliations() + return { + "incomplete": [r.reconciliation_id for r in runs], + "recommendation": "observe uncertain operations before retrying" if runs else "none", + } diff --git a/tests/test_controlplane.py b/tests/test_controlplane.py new file mode 100644 index 0000000..564ff31 --- /dev/null +++ b/tests/test_controlplane.py @@ -0,0 +1,221 @@ +from __future__ import annotations + +from datetime import UTC, datetime, timedelta + +import pytest + +from netsovereign.controlplane.models import ( + Checkpoint, + DesiredRevision, + DriftClassification, + ObservedRecord, +) +from netsovereign.controlplane.repository import SQLiteControlPlaneRepository +from netsovereign.controlplane.service import ControlPlaneService + +NOW = datetime(2026, 1, 1, tzinfo=UTC) + + +def revision(revision_id: str = "r1", parent: str | None = None) -> DesiredRevision: + return DesiredRevision( + world_id="world", + revision_id=revision_id, + parent_revision_id=parent, + schema_version="v0alpha2", + declaration_digest=revision_id, + declaration={"resources": {"resource": {"enabled": True}}}, + accepted_at=NOW, + admission={"status": "accepted"}, + actor="test", + authority_manifest_digest="authority", + status="accepted", + ) + + +def test_revision_is_immutable_active_and_restart_safe(tmp_path): + path = tmp_path / "control.db" + first = SQLiteControlPlaneRepository(path) + ControlPlaneService(first).accept_revision(revision()) + first.connection.close() + second = SQLiteControlPlaneRepository(path) + assert second.active_revision("world") == revision() + with pytest.raises(ValueError, match="immutable"): + second.put_immutable("desired", "r1", revision().model_copy(update={"actor": "other"})) + + +def test_child_history_and_rejected_revision(tmp_path): + repo = SQLiteControlPlaneRepository(tmp_path / "cp.db") + service = ControlPlaneService(repo) + service.accept_revision(revision()) + service.accept_revision(revision("r2", "r1")) + rejected = revision("bad", "r2").model_copy( + update={"status": "rejected", "admission": {"status": "rejected"}} + ) + service.accept_revision(rejected) + assert repo.active_revision("world").revision_id == "r2" # type: ignore[union-attr] + assert len(repo.list("desired", DesiredRevision)) == 3 + + +def test_lock_fencing_owner_and_expiry(tmp_path): + service = ControlPlaneService(SQLiteControlPlaneRepository(tmp_path / "cp.db")) + one = service.acquire_world("world", "one", NOW, 10) + with pytest.raises(RuntimeError, match="already"): + service.acquire_world("world", "two", NOW, 10) + assert not service.repository.release_lock(one.key, "two", "wrong", NOW) + two = service.acquire_world("world", "two", NOW + timedelta(seconds=11), 10) + assert two.fencing_token == one.fencing_token + 1 + + +def test_lock_fencing_survives_release_and_reacquire(tmp_path): + service = ControlPlaneService(SQLiteControlPlaneRepository(tmp_path / "cp.db")) + one = service.acquire_world("world", "one", NOW, 10) + assert service.repository.release_lock(one.key, "one", "complete", NOW) + two = service.acquire_world("world", "two", NOW, 10) + assert two.fencing_token == one.fencing_token + 1 + assert not service.repository.release_lock(one.key, "one", "duplicate", NOW) + + +def test_active_same_owner_acquire_is_idempotent(tmp_path): + service = ControlPlaneService(SQLiteControlPlaneRepository(tmp_path / "cp.db")) + one = service.acquire_world("world", "one", NOW, 10) + duplicate = service.acquire_world("world", "one", NOW + timedelta(seconds=1), 20) + assert duplicate == one + + +@pytest.mark.parametrize( + ("record", "classification"), + [ + (None, DriftClassification.UNVERIFIABLE), + ({"digest": None, "observation_type": "absent"}, DriftClassification.MISSING), + ({"digest": "wrong"}, DriftClassification.CHANGED), + ({"digest": None}, DriftClassification.UNVERIFIABLE), + ({"digest": "wrong", "health": "unhealthy"}, DriftClassification.UNHEALTHY), + ({"digest": "wrong", "health": "unavailable"}, DriftClassification.PROVIDER_UNAVAILABLE), + ({"digest": "wrong", "stale_after": NOW}, DriftClassification.STALE), + ], +) +def test_drift_classifications(tmp_path, record, classification): + repo = SQLiteControlPlaneRepository(tmp_path / "cp.db") + service = ControlPlaneService(repo) + service.accept_revision(revision()) + if record is not None: + observed = ObservedRecord( + observation_id="o1", + world_id="world", + resource_id="resource", + provider_id="fake", + binding_id="fake", + capability="test", + observation_type=record.get("observation_type", "state"), + representation={}, + observed_at=NOW - timedelta(seconds=1), + health=record.get("health", "healthy"), + digest=record["digest"], + stale_after=record.get("stale_after"), + ) + repo.put_immutable("observed", "o1", observed) + repo.connection.commit() + assert service.inspect_drift("world", NOW)[0].classification == classification + + +def test_list_shaped_world_resources_are_normalized(tmp_path): + repo = SQLiteControlPlaneRepository(tmp_path / "cp.db") + service = ControlPlaneService(repo) + listed = revision().model_copy( + update={"declaration": {"resources": [{"id": "resource", "enabled": True}]}} + ) + service.accept_revision(listed) + assert service.inspect_drift("world", NOW)[0].resource_id == "resource" + + +def test_repeated_drift_detection_appends_history(tmp_path): + repo = SQLiteControlPlaneRepository(tmp_path / "cp.db") + service = ControlPlaneService(repo) + service.accept_revision(revision()) + first = service.inspect_drift("world", NOW)[0] + second = service.inspect_drift("world", NOW + timedelta(seconds=1))[0] + assert first.drift_id != second.drift_id + assert len(repo.list("drift", type(first))) == 2 + + +def test_active_parent_is_read_inside_write_transaction(tmp_path): + class TransactionCheckingRepository(SQLiteControlPlaneRepository): + checked = False + + def active_revision(self, world_id: str) -> DesiredRevision | None: + if super().active_revision(world_id) is not None: + self.checked = self.connection.in_transaction + return super().active_revision(world_id) + + repo = TransactionCheckingRepository(tmp_path / "cp.db") + service = ControlPlaneService(repo) + service.accept_revision(revision()) + service.accept_revision(revision("r2", "r1")) + assert repo.checked + + +def test_revision_acceptance_is_idempotent_and_initial_parent_fails(tmp_path): + repo = SQLiteControlPlaneRepository(tmp_path / "cp.db") + service = ControlPlaneService(repo) + assert service.accept_revision(revision()) == revision() + assert service.accept_revision(revision()) == revision() + other = SQLiteControlPlaneRepository(tmp_path / "other.db") + with pytest.raises(ValueError, match="initial.*parent"): + ControlPlaneService(other).accept_revision(revision("r2", "r1")) + + +def test_repository_cannot_activate_revision_for_another_world(tmp_path): + repo = SQLiteControlPlaneRepository(tmp_path / "cp.db") + with repo.transaction(): + repo.put_immutable("desired", "r1", revision()) + with pytest.raises(ValueError, match="different world"): + repo.set_active_revision("other", "r1") + + +def test_schema_version_is_upgraded_and_future_version_rejected(tmp_path): + path = tmp_path / "cp.db" + repo = SQLiteControlPlaneRepository(path) + repo.connection.execute("UPDATE cp_metadata SET value='1' WHERE key='schema_version'") + repo.connection.commit() + repo.connection.close() + upgraded = SQLiteControlPlaneRepository(path) + assert upgraded.connection.execute( + "SELECT value FROM cp_metadata WHERE key='schema_version'" + ).fetchone() == ("4",) + upgraded.connection.execute("UPDATE cp_metadata SET value='5' WHERE key='schema_version'") + upgraded.connection.commit() + upgraded.connection.close() + with pytest.raises(RuntimeError, match="incompatible"): + SQLiteControlPlaneRepository(path) + + +def test_checkpoint_rejects_secret_material(): + with pytest.raises(ValueError, match="must not contain secrets"): + Checkpoint( + checkpoint_id="checkpoint", + world_id="world", + desired_revision_id="r1", + plan_fingerprint="fingerprint", + run_id="run", + completed_operations=[], + incomplete_operations=[], + latest_observation_ids=[], + compensation={"access_token": "sensitive"}, + created_at=NOW, + reason="test", + ) + + +def test_control_plane_timestamps_must_be_timezone_aware(): + payload = revision().model_dump() + payload["accepted_at"] = NOW.replace(tzinfo=None) + with pytest.raises(ValueError, match="timezone-aware"): + DesiredRevision.model_validate(payload) + + +def test_transaction_rolls_back(tmp_path): + repo = SQLiteControlPlaneRepository(tmp_path / "cp.db") + with pytest.raises(RuntimeError), repo.transaction(): + repo.put_immutable("desired", "r1", revision()) + raise RuntimeError("crash") + assert repo.get("desired", "r1", DesiredRevision) is None diff --git a/uv.lock b/uv.lock index 038e907..0e39b32 100644 --- a/uv.lock +++ b/uv.lock @@ -236,7 +236,7 @@ wheels = [ [[package]] name = "netsovereign" -version = "0.3.0" +version = "0.4.0" source = { editable = "." } dependencies = [ { name = "pydantic" },