diff --git a/src/odoo_forge/resource_lifecycle/service.py b/src/odoo_forge/resource_lifecycle/service.py index 94166e6..e2e21ba 100644 --- a/src/odoo_forge/resource_lifecycle/service.py +++ b/src/odoo_forge/resource_lifecycle/service.py @@ -366,6 +366,7 @@ def _quarantine_history( evidence_digest=observation.evidence_digest, resource_class=observation.resource_class, quarantined_at=now or datetime.now(UTC), + last_activity=observation.last_activity, ) diff --git a/src/odoo_forge/resource_lifecycle/types.py b/src/odoo_forge/resource_lifecycle/types.py index df5f37f..95e43b6 100644 --- a/src/odoo_forge/resource_lifecycle/types.py +++ b/src/odoo_forge/resource_lifecycle/types.py @@ -98,6 +98,16 @@ def evaluate_expiration( def reset_activity_baseline( resource: LifecycleResource, occurred_at: datetime ) -> LifecycleResource: + """Model the effect of a qualifying use/renewal on the activity baseline. + + No production code path calls this: the design keeps the provider boundary + read-only (`DatabaseLifecycleGateway` exposes no renew/reset verb), so the + activity baseline is observed from the live provider (for example the + Docker `io.odoo-forge.last-activity` label) rather than written back by + this system. This pure function exists to prove the spec's "Configured + baseline prevents premature expiration" scenario at the domain-value + level, independent of how the provider records the renewal. + """ return resource.model_copy(update={"last_activity": occurred_at}) @@ -125,6 +135,7 @@ class QuarantineHistory(_LifecycleValue): evidence_digest: str resource_class: ResourceClass quarantined_at: AwareDatetime + last_activity: datetime | None = None class LifecycleJournalEvent(_LifecycleValue): diff --git a/src/odoo_forge_postgres_docker/authority.py b/src/odoo_forge_postgres_docker/authority.py index 5792334..198085f 100644 --- a/src/odoo_forge_postgres_docker/authority.py +++ b/src/odoo_forge_postgres_docker/authority.py @@ -381,6 +381,14 @@ def has_record(self, operation: str, name: str) -> bool: state = self.read() return self._latest(state, operation, name) is not None + def lifecycle_records(self) -> tuple[dict[str, object], ...]: + state = self.read() + latest = { + (str(record["operation"]), str(record["name"])): dict(record) + for record in state["records"] + } + return tuple(latest.values()) + def owns(self, operation: str, name: str, docker_id: str) -> bool: """Return whether a signed active local record proves this exact resource.""" state = self.read() @@ -521,7 +529,7 @@ def _validate_private_path(path: Path, mode: int, *, directory: bool) -> None: @staticmethod def _validate_record(record: Mapping[str, object]) -> None: if ( - set(record) != _REQUIRED_RECORD_FIELDS + not set(record) >= _REQUIRED_RECORD_FIELDS or not all(isinstance(value, str) for value in record.values()) or not all(record[key] for key in ("operation", "kind", "name", "state")) or record["state"] not in {"reserved", "active", "retired"} @@ -596,7 +604,7 @@ def _validate_state(self, payload: object) -> dict[str, Any]: for record in records: if not isinstance(record, dict): raise AuthorityStateError() - if set(record) != _STORED_RECORD_FIELDS: + if not set(record) >= _STORED_RECORD_FIELDS: raise AuthorityStateError() stored_generation = record.get("generation") key_id = record.get("key_id") diff --git a/src/odoo_forge_postgres_docker/lifecycle.py b/src/odoo_forge_postgres_docker/lifecycle.py new file mode 100644 index 0000000..67c33b5 --- /dev/null +++ b/src/odoo_forge_postgres_docker/lifecycle.py @@ -0,0 +1,277 @@ +from __future__ import annotations + +import fcntl +import json +import math +import os +import re +import subprocess +from collections.abc import Callable, Mapping, Sequence +from pathlib import Path +from typing import Any, cast + +from pydantic import AwareDatetime, BaseModel, ConfigDict + +from odoo_forge.database.types import ( + CleanupReport, + CreationReceipt, + DatabaseCreation, + DatabaseRef, + OperationIdentity, + ResourceOwnership, +) +from odoo_forge.durable_operations.types import DurableOperationIdentity +from odoo_forge.resource_lifecycle.types import ( + DatabaseObservation, + LifecycleJournalEvent, + ProviderPresence, + ResourceClass, +) +from odoo_forge.resource_ownership.types import OwnershipReceipt +from odoo_forge.tenancy.types import ProjectScope, TenantId + +MAX_DOCKER_TIMEOUT = 600.0 + +_IDENTIFIER = re.compile(r"^[a-z][a-z0-9-]{0,63}$") +_DOCKER_ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.:-]{0,127}$") +_OPERATION_LABEL = "io.odoo-forge.operation" +_CLASS_LABEL = "io.odoo-forge.resource-class" +_ACTIVITY_LABEL = "io.odoo-forge.last-activity" +_DIGEST_LABEL = "io.odoo-forge.evidence-digest" + + +class LifecycleAuthorityRecord(BaseModel): + model_config = ConfigDict(frozen=True, extra="forbid", hide_input_in_errors=True) + + resource: DatabaseRef + scope: ProjectScope + operation: DurableOperationIdentity + docker_id: str + resource_class: ResourceClass + last_activity: AwareDatetime + evidence_digest: str + + +def _run_docker(argv: Sequence[str], *, timeout: float) -> subprocess.CompletedProcess[str]: + return subprocess.run( + list(argv), capture_output=True, check=False, shell=False, text=True, timeout=timeout + ) + + +class PostgresDockerLifecycleAdapter: + def __init__( + self, + *, + provider: Any, + authority: Any, + runner: Callable[..., subprocess.CompletedProcess[str]] = _run_docker, + timeout: float = 10.0, + ) -> None: + if not math.isfinite(timeout) or not 0.0 < timeout <= MAX_DOCKER_TIMEOUT: + raise ValueError(f"timeout must be finite and within (0, {MAX_DOCKER_TIMEOUT}] seconds") + self.provider = provider + self.authority = authority + self._runner = runner + self._timeout = timeout + + def observe(self, scope: ProjectScope) -> tuple[DatabaseObservation, ...]: + records = tuple( + record + for raw in self.authority.lifecycle_records() + if (record := _coerce_record(raw)) is not None + ) + counts: dict[str, int] = {} + for record in records: + counts[record.resource.identifier] = counts.get(record.resource.identifier, 0) + 1 + observations: list[DatabaseObservation] = [] + for record in records: + if record.scope != scope: + continue + if counts[record.resource.identifier] != 1: + observations.append(self._observation(record, ProviderPresence.UNVERIFIABLE)) + continue + observations.append(self._observe_record(record)) + return tuple(observations) + + def _observe_record(self, record: LifecycleAuthorityRecord) -> DatabaseObservation: + try: + if _IDENTIFIER.fullmatch(record.resource.identifier) is None: + raise ValueError("unsafe resource identifier") + if _DOCKER_ID.fullmatch(record.docker_id) is None: + raise ValueError("unsafe Docker identity") + listed = self._run( + [ + "docker", + "ps", + "-a", + "--no-trunc", + "--filter", + f"id={record.docker_id}", + "--format", + "{{.ID}}", + ] + ) + lines = tuple(line for line in listed.stdout.splitlines() if line) + if not lines: + return self._observation(record, ProviderPresence.ABSENT) + if lines != (record.docker_id,): + raise ValueError("docker identity did not match authority") + inspected = self._run(["docker", "inspect", record.docker_id]) + payload = json.loads(inspected.stdout) + if ( + not isinstance(payload, list) + or len(payload) != 1 + or not isinstance(payload[0], dict) + ): + raise ValueError("malformed inspect result") + entry = payload[0] + if entry.get("Id") != record.docker_id: + raise ValueError("docker identity did not match inspect") + config = entry.get("Config") + state = entry.get("State") + labels = config.get("Labels") if isinstance(config, dict) else None + if not isinstance(labels, dict) or not isinstance(state, dict): + raise ValueError("missing live evidence") + if any( + labels.get(key) != value + for key, value in { + _OPERATION_LABEL: record.operation.operation_id, + _CLASS_LABEL: record.resource_class.value, + _ACTIVITY_LABEL: record.last_activity.isoformat(), + _DIGEST_LABEL: record.evidence_digest, + }.items() + ): + raise ValueError("live evidence contradicted authority") + dead = state.get("Dead") + if not isinstance(dead, bool): + raise ValueError("missing dead state") + return self._observation( + record, ProviderPresence.INVALID if dead else ProviderPresence.PRESENT + ) + except Exception: + return self._observation(record, ProviderPresence.UNVERIFIABLE) + + def _run(self, argv: Sequence[str]) -> subprocess.CompletedProcess[str]: + result = self._runner(argv, timeout=self._timeout) + if result.returncode != 0 or not isinstance(result.stdout, str): + raise ValueError("docker command failed") + return result + + @staticmethod + def _observation( + record: LifecycleAuthorityRecord, presence: ProviderPresence + ) -> DatabaseObservation: + receipt = OwnershipReceipt( + operation=record.operation, + owned_resource_ids=(record.resource.identifier,), + ) + return DatabaseObservation( + ref=record.resource, + scope=record.scope, + evidence_digest=record.evidence_digest, + ownership_valid=record.resource.ownership is not ResourceOwnership.EXTERNAL, + resource_class=record.resource_class, + last_activity=record.last_activity, + receipt=receipt, + presence=presence, + ) + + def quarantine(self, ref: DatabaseRef) -> DatabaseRef: + return cast(DatabaseRef, self.provider.quarantine(ref)) + + def adopt(self, ref: DatabaseRef) -> DatabaseRef: + return cast(DatabaseRef, self.provider.adopt(ref)) + + def reconcile(self, operation: OperationIdentity) -> DatabaseCreation: + return cast(DatabaseCreation, self.provider.reconcile(operation)) + + def delete(self, creation: DatabaseCreation) -> None: + self.provider.delete(creation) + + def cleanup(self, receipt: CreationReceipt) -> CleanupReport: + return cast(CleanupReport, self.provider.cleanup(receipt)) + + +class JsonlLifecycleJournal: + def __init__(self, path: Path) -> None: + self.path = path + + def append(self, event: LifecycleJournalEvent) -> LifecycleJournalEvent: + self.path.parent.mkdir(mode=0o700, parents=True, exist_ok=True) + descriptor = os.open( + self.path, os.O_WRONLY | os.O_APPEND | os.O_CREAT | os.O_NOFOLLOW, 0o600 + ) + try: + # A partial write can split one record across several os.write calls. + # Hold the lock across the whole record so a concurrent appender + # cannot interleave its own bytes and corrupt both JSONL lines. + fcntl.flock(descriptor, fcntl.LOCK_EX) + try: + encoded = (event.model_dump_json() + "\n").encode() + while encoded: + written = os.write(descriptor, encoded) + if written <= 0: + raise OSError("journal write made no progress") + encoded = encoded[written:] + os.fsync(descriptor) + finally: + fcntl.flock(descriptor, fcntl.LOCK_UN) + finally: + os.close(descriptor) + # Syncing the file alone leaves its directory entry unrecoverable after + # a crash, losing the whole audit trail. Every append pays this, not + # just the creating one: another process can open the new file and + # return successfully before the creator would have synced. + _fsync_directory(self.path.parent) + return event + + def events(self) -> tuple[LifecycleJournalEvent, ...]: + try: + descriptor = os.open(self.path, os.O_RDONLY | os.O_NOFOLLOW) + except FileNotFoundError: + return () + try: + # Without a shared lock a read can land inside an in-flight append + # and hand a truncated final line to the parser. + fcntl.flock(descriptor, fcntl.LOCK_SH) + try: + payload = os.fdopen(descriptor, encoding="utf-8", closefd=False).read() + finally: + fcntl.flock(descriptor, fcntl.LOCK_UN) + finally: + os.close(descriptor) + return tuple( + LifecycleJournalEvent.model_validate_json(line) for line in payload.splitlines() if line + ) + + +def _fsync_directory(directory: Path) -> None: + descriptor = os.open(directory, os.O_RDONLY) + try: + os.fsync(descriptor) + finally: + os.close(descriptor) + + +def _coerce_record(raw: object) -> LifecycleAuthorityRecord | None: + if isinstance(raw, LifecycleAuthorityRecord): + return raw + if not isinstance(raw, Mapping) or raw.get("state") != "active": + return None + try: + values = cast(Mapping[str, Any], raw) + return LifecycleAuthorityRecord( + resource=DatabaseRef(identifier=values["name"], ownership=ResourceOwnership.CREATED), + scope=ProjectScope( + tenant=TenantId(value=values["tenant_id"]), project_id=values["project_id"] + ), + operation=DurableOperationIdentity( + operation_id=values["operation"], request_digest=values["request_digest"] + ), + docker_id=values["docker_id"], + resource_class=ResourceClass(values["resource_class"]), + last_activity=values["last_activity"], + evidence_digest=values["evidence_digest"], + ) + except (KeyError, TypeError, ValueError): + return None diff --git a/tests/adapters/test_postgres_docker_lifecycle.py b/tests/adapters/test_postgres_docker_lifecycle.py new file mode 100644 index 0000000..1f1319d --- /dev/null +++ b/tests/adapters/test_postgres_docker_lifecycle.py @@ -0,0 +1,197 @@ +from __future__ import annotations + +import multiprocessing +import subprocess +from collections.abc import Callable +from datetime import timedelta +from pathlib import Path +from types import SimpleNamespace +from unittest.mock import Mock, call + +import pytest + +from odoo_forge.database.types import OperationIdentity +from odoo_forge.resource_lifecycle.types import ( + LifecycleAuthorization, + LifecycleEvidence, + LifecycleJournalEvent, + LifecycleOutcome, + LifecyclePolicy, + ProviderPresence, +) +from odoo_forge.tenancy import ProjectScope, TenantId +from odoo_forge_postgres_docker.authority import LocalOwnershipAuthority +from odoo_forge_postgres_docker.lifecycle import ( + MAX_DOCKER_TIMEOUT, + JsonlLifecycleJournal, + PostgresDockerLifecycleAdapter, + _run_docker, +) + +SCOPE = ProjectScope(tenant=TenantId(value="tenant"), project_id="project") +OPERATION = OperationIdentity(value="op") +_INSPECT = ( + '[{"Id":"id","Config":{"Labels":{"io.odoo-forge.operation":"op",' + '"io.odoo-forge.resource-class":"dev","io.odoo-forge.last-activity":"2026-01-01T00:00:00+00:00","io.odoo-forge.evidence-digest":"evidence-1"}},"State":{"Dead":false}}]' # noqa: E501 +) + + +def _authority(*records: object) -> SimpleNamespace: + return SimpleNamespace(lifecycle_records=lambda: records) + + +def _journal_event() -> LifecycleJournalEvent: + return LifecycleJournalEvent( + policy=LifecyclePolicy(ttl=timedelta(days=1), grace=timedelta(days=1)), + evidence=LifecycleEvidence(source="adapter", digest="evidence-1"), + authorization=LifecycleAuthorization(actor="operator", reason="approved"), + outcome=LifecycleOutcome.ALERTED, + kind="run", + ) + + +def _append_many(path: str, count: int) -> None: + journal = JsonlLifecycleJournal(Path(path)) + for _ in range(count): + journal.append(_journal_event()) + + +def _raw_record(full: bool = True) -> dict[str, str]: + value = { + "operation": "op", + "kind": "container", + "name": "database", + "docker_id": "id", + "state": "active", + } + if full: + value.update( + tenant_id=SCOPE.tenant.value, + project_id=SCOPE.project_id, + request_digest="request-1", + resource_class="dev", + last_activity="2026-01-01T00:00:00+00:00", + evidence_digest="evidence-1", + ) + return value + + +def _runner(mode: str = "present") -> Callable[..., subprocess.CompletedProcess[str]]: + def run(argv: list[str], *, timeout: float) -> subprocess.CompletedProcess[str]: + if mode == "timeout": + raise subprocess.TimeoutExpired(argv, timeout) + if mode == "nonzero": + return subprocess.CompletedProcess(argv, 1, "", "failure") + if argv[1:3] == ["ps", "-a"]: + output = {"absent": "", "identity": "other-id\n"}.get(mode, "id\n") + return subprocess.CompletedProcess(argv, 0, output, "") + if mode == "malformed": + return subprocess.CompletedProcess(argv, 0, "not-json", "") + output = _INSPECT.replace('"Dead":false', f'"Dead":{str(mode == "dead").lower()}') + if mode == "labels": + output = output.replace('"op"', '"other-operation"') + return subprocess.CompletedProcess(argv, 0, output, "") + + return run + + +def test_real_authority_and_legacy_rows(tmp_path: Path) -> None: + authority = LocalOwnershipAuthority(tmp_path / "authority") + authority.write(_raw_record()) + result = PostgresDockerLifecycleAdapter( + provider=Mock(), authority=authority, runner=_runner() + ).observe(SCOPE) + assert result[0].scope == SCOPE and result[0].presence is ProviderPresence.PRESENT + legacy = LocalOwnershipAuthority(tmp_path / "legacy") + legacy.write(_raw_record(False)) + assert not PostgresDockerLifecycleAdapter(provider=Mock(), authority=legacy).observe(SCOPE) + + +@pytest.mark.parametrize( + "mode, expected", + [ + ("absent", ProviderPresence.ABSENT), + ("dead", ProviderPresence.INVALID), + ("present", ProviderPresence.PRESENT), + ("malformed", ProviderPresence.UNVERIFIABLE), + ("nonzero", ProviderPresence.UNVERIFIABLE), + ("timeout", ProviderPresence.UNVERIFIABLE), + ("identity", ProviderPresence.UNVERIFIABLE), + ("labels", ProviderPresence.UNVERIFIABLE), + ], +) +def test_provider_evidence_maps_to_typed_presence(mode: str, expected: ProviderPresence) -> None: + record, provider = _raw_record(), Mock() + if mode == "identity": + record["name"] = "db;rm" + result = PostgresDockerLifecycleAdapter( + provider=provider, authority=_authority(record), runner=_runner(mode) + ).observe(SCOPE)[0] + assert result.presence is expected and provider.mock_calls == [] + + +def test_jsonl_journal_reloads_immutable_run_and_action_records(tmp_path: Path) -> None: + journal = JsonlLifecycleJournal(tmp_path / "lifecycle.jsonl") + event = _journal_event() + action = event.model_copy(update={"kind": "action", "outcome": LifecycleOutcome.QUARANTINED}) + journal.append(event) + journal.append(action) + assert JsonlLifecycleJournal(journal.path).events() == (event, action) + + +@pytest.mark.parametrize( + "timeout", [float("inf"), float("nan"), 0.0, -1.0, MAX_DOCKER_TIMEOUT + 1.0] +) +def test_adapter_rejects_timeouts_without_a_finite_positive_bound(timeout: float) -> None: + with pytest.raises(ValueError): + PostgresDockerLifecycleAdapter(provider=Mock(), authority=_authority(), timeout=timeout) + + +def test_journal_reloads_every_record_intact_after_concurrent_appends(tmp_path: Path) -> None: + path = tmp_path / "state" / "lifecycle.jsonl" + workers = [ + multiprocessing.get_context("spawn").Process(target=_append_many, args=(str(path), 25)) + for _ in range(4) + ] + for worker in workers: + worker.start() + for worker in workers: + worker.join(timeout=60) + assert [worker.exitcode for worker in workers] == [0, 0, 0, 0] + events = JsonlLifecycleJournal(path).events() + assert len(events) == 100 and set(events) == {_journal_event()} + + +def test_journal_reads_return_only_whole_records_while_appends_run(tmp_path: Path) -> None: + path = tmp_path / "state" / "lifecycle.jsonl" + worker = multiprocessing.get_context("spawn").Process(target=_append_many, args=(str(path), 60)) + worker.start() + try: + reader = JsonlLifecycleJournal(path) + for _ in range(40): + assert set(reader.events()) <= {_journal_event()} + finally: + worker.join(timeout=60) + assert worker.exitcode == 0 and len(JsonlLifecycleJournal(path).events()) == 60 + + +def test_default_runner_uses_fixed_argv_without_shell(monkeypatch: pytest.MonkeyPatch) -> None: + run = Mock(return_value=subprocess.CompletedProcess([], 0, "", "")) + monkeypatch.setattr(subprocess, "run", run) + _run_docker(["docker", "ps", "-a"], timeout=3.0) + assert run.call_count == 1 and run.call_args.args == (["docker", "ps", "-a"],) and run.call_args.kwargs["shell"] is False # noqa: E501 # fmt: skip + + +def test_recovery_verbs_delegate_without_provision_or_restore() -> None: + provider = Mock() + ref, creation, receipt = Mock(), Mock(), Mock() + adapter = PostgresDockerLifecycleAdapter(provider=provider, authority=_authority()) + adapter.quarantine(ref) + adapter.adopt(ref) + adapter.reconcile(OPERATION) + adapter.delete(creation) + report = adapter.cleanup(receipt) + expected = [call.quarantine(ref), call.adopt(ref), call.reconcile(OPERATION), call.delete(creation), call.cleanup(receipt)] # noqa: E501 # fmt: skip + assert report is provider.cleanup.return_value + assert provider.mock_calls == expected + assert provider.provision.call_count == provider.restore.call_count == 0 diff --git a/tests/resource_lifecycle/test_service.py b/tests/resource_lifecycle/test_service.py index 321a893..fb0c3e8 100644 --- a/tests/resource_lifecycle/test_service.py +++ b/tests/resource_lifecycle/test_service.py @@ -487,10 +487,24 @@ def test_quarantine_history_reuses_exact_pointer_and_preserves_lineage() -> None evidence_digest="digest-1", resource_class=ResourceClass.DEV, quarantined_at=NOW, + last_activity=NOW - timedelta(days=10), ) assert registry.get_calls[-1] == POINTER +def test_quarantine_records_the_applied_activity_baseline() -> None: + baseline = NOW - timedelta(days=30) + gateway = _ProviderOnlyGateway((_observation(last_activity=baseline),)) + journal = _AppendOnlyJournal() + service = _service(_Registry((RECORD,)), gateway, journal) + + service.run(SCOPE, POLICY, AUTHORIZATION, now=NOW) + + history = next(event.history for event in journal.events() if event.history is not None) + assert history is not None + assert history.last_activity == baseline + + @pytest.mark.parametrize("presence", [ProviderPresence.ABSENT, ProviderPresence.INVALID]) def test_confirmed_zombie_requires_registry_absence_and_provider_absent_or_invalid( presence: ProviderPresence,