From b18fab9df71710f44d5a44f2fc2d68283eb71950 Mon Sep 17 00:00:00 2001 From: Angel Parra <607418+aparragithub@users.noreply.github.com> Date: Fri, 7 Aug 2026 14:35:06 -0500 Subject: [PATCH 1/5] feat(lifecycle): add dormant PostgreSQL Docker lifecycle adapter Implement the PR4 slice of SP-RESOURCE-LIFECYCLE: a lifecycle adapter that observes PostgreSQL Docker resources through fixed-argv commands, resolves read-only authority records, and journals immutable evidence. - Fixed-argv Docker observation with no shell; malformed output, nonzero exit, timeout, and identifier mismatch all map to UNVERIFIABLE with zero mutation. - Read-only tenant/project authority lookup emitting typed presence evidence; caller scope is never copied into the result. - Fsynced append-only JSONL journal that reloads immutable run and action records; recovery verbs delegate without provision or restore. The module stays unimported by the server, so no runtime behavior changes until the composition slice wires it. --- src/odoo_forge_postgres_docker/authority.py | 12 +- src/odoo_forge_postgres_docker/lifecycle.py | 241 ++++++++++++++++++ .../test_postgres_docker_lifecycle.py | 147 +++++++++++ 3 files changed, 398 insertions(+), 2 deletions(-) create mode 100644 src/odoo_forge_postgres_docker/lifecycle.py create mode 100644 tests/adapters/test_postgres_docker_lifecycle.py 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..c43cfb0 --- /dev/null +++ b/src/odoo_forge_postgres_docker/lifecycle.py @@ -0,0 +1,241 @@ +from __future__ import annotations + +import json +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 + +_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: + 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: + 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: + os.close(descriptor) + return event + + def events(self) -> tuple[LifecycleJournalEvent, ...]: + if not self.path.exists(): + return () + return tuple( + LifecycleJournalEvent.model_validate_json(line) + for line in self.path.read_text(encoding="utf-8").splitlines() + if line + ) + + +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..f582048 --- /dev/null +++ b/tests/adapters/test_postgres_docker_lifecycle.py @@ -0,0 +1,147 @@ +from __future__ import annotations + +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 ( + 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 _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 = 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", + ) + action = event.model_copy(update={"kind": "action", "outcome": LifecycleOutcome.QUARANTINED}) + journal.append(event) + journal.append(action) + assert JsonlLifecycleJournal(journal.path).events() == (event, action) + + +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) + assert adapter.cleanup(receipt) is provider.cleanup.return_value + assert provider.mock_calls == [call.quarantine(ref), call.adopt(ref), call.reconcile(OPERATION), call.delete(creation), call.cleanup(receipt)] # noqa: E501 # fmt: skip + assert provider.provision.call_count == provider.restore.call_count == 0 From 153c9961b7499d417cd0a05ee62e765b45cd9c16 Mon Sep 17 00:00:00 2001 From: Angel Parra <607418+aparragithub@users.noreply.github.com> Date: Fri, 7 Aug 2026 14:39:17 -0500 Subject: [PATCH 2/5] test(lifecycle): hoist delegation calls out of assert statements CodeQL py/side-effect-in-assert flagged the mock_calls expectation. The preceding assert also invoked adapter.cleanup() inside the assert itself, so under python -O the call would be stripped and the delegation expectation would no longer hold. Bind both results before asserting. --- tests/adapters/test_postgres_docker_lifecycle.py | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/tests/adapters/test_postgres_docker_lifecycle.py b/tests/adapters/test_postgres_docker_lifecycle.py index f582048..e56826e 100644 --- a/tests/adapters/test_postgres_docker_lifecycle.py +++ b/tests/adapters/test_postgres_docker_lifecycle.py @@ -142,6 +142,8 @@ def test_recovery_verbs_delegate_without_provision_or_restore() -> None: adapter.adopt(ref) adapter.reconcile(OPERATION) adapter.delete(creation) - assert adapter.cleanup(receipt) is provider.cleanup.return_value - assert provider.mock_calls == [call.quarantine(ref), call.adopt(ref), call.reconcile(OPERATION), call.delete(creation), call.cleanup(receipt)] # noqa: E501 # fmt: skip + 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 From 31c577ebdff6648b64fb8cf477261a8b58987968 Mon Sep 17 00:00:00 2001 From: Angel Parra <607418+aparragithub@users.noreply.github.com> Date: Fri, 7 Aug 2026 14:46:08 -0500 Subject: [PATCH 3/5] fix(lifecycle): bound the Docker timeout and make journal appends durable Address two review findings on the lifecycle adapter. The adapter accepted any timeout, so float("inf") let subprocess.run wait without a deadline and defeated the bounded-timeout guarantee the Docker boundary relies on. Reject non-finite, non-positive, and over-limit values at construction, matching how LifecycleService already rejects a negative max_cleanup_retries. A journal record can span several os.write calls when a write is partial. Under O_APPEND another process could interleave its own record between those calls and corrupt both JSONL lines, so hold an exclusive lock across the whole record through fsync. Syncing the file also left a newly created journal's directory entry unrecoverable after a crash, losing the entire audit trail; fsync the parent directory when the append creates the file. --- src/odoo_forge_postgres_docker/lifecycle.py | 50 ++++++++++--- .../test_postgres_docker_lifecycle.py | 75 +++++++++++++++++-- 2 files changed, 108 insertions(+), 17 deletions(-) diff --git a/src/odoo_forge_postgres_docker/lifecycle.py b/src/odoo_forge_postgres_docker/lifecycle.py index c43cfb0..2cfb1aa 100644 --- a/src/odoo_forge_postgres_docker/lifecycle.py +++ b/src/odoo_forge_postgres_docker/lifecycle.py @@ -1,6 +1,8 @@ from __future__ import annotations +import fcntl import json +import math import os import re import subprocess @@ -28,6 +30,8 @@ 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" @@ -63,6 +67,8 @@ def __init__( 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 @@ -192,21 +198,37 @@ def __init__(self, path: Path) -> None: 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 - ) + descriptor, created = self._open() 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) + # 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) + if created: + # Syncing the file alone leaves its directory entry unrecoverable + # after a crash, which would lose the whole audit trail. + _fsync_directory(self.path.parent) return event + def _open(self) -> tuple[int, bool]: + flags = os.O_WRONLY | os.O_APPEND | os.O_NOFOLLOW + try: + return os.open(self.path, flags | os.O_CREAT | os.O_EXCL, 0o600), True + except FileExistsError: + return os.open(self.path, flags), False + def events(self) -> tuple[LifecycleJournalEvent, ...]: if not self.path.exists(): return () @@ -217,6 +239,14 @@ def events(self) -> tuple[LifecycleJournalEvent, ...]: ) +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 diff --git a/tests/adapters/test_postgres_docker_lifecycle.py b/tests/adapters/test_postgres_docker_lifecycle.py index e56826e..4b540fe 100644 --- a/tests/adapters/test_postgres_docker_lifecycle.py +++ b/tests/adapters/test_postgres_docker_lifecycle.py @@ -1,5 +1,8 @@ from __future__ import annotations +import fcntl +import os +import stat import subprocess from collections.abc import Callable from datetime import timedelta @@ -21,6 +24,7 @@ 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, @@ -38,6 +42,16 @@ 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 _raw_record(full: bool = True) -> dict[str, str]: value = { "operation": "op", @@ -114,19 +128,66 @@ def test_provider_evidence_maps_to_typed_presence(mode: str, expected: ProviderP def test_jsonl_journal_reloads_immutable_run_and_action_records(tmp_path: Path) -> None: journal = JsonlLifecycleJournal(tmp_path / "lifecycle.jsonl") - event = 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", - ) + 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_holds_an_exclusive_lock_from_first_write_through_fsync( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + order: list[str] = [] + real_flock, real_write, real_fsync = fcntl.flock, os.write, os.fsync + + def flock(descriptor: int, operation: int) -> None: + order.append("lock" if operation == fcntl.LOCK_EX else "unlock") + real_flock(descriptor, operation) + + def write(descriptor: int, data: bytes) -> int: + order.append("write") + return real_write(descriptor, data) + + def fsync(descriptor: int) -> None: + order.append("fsync") + real_fsync(descriptor) + + monkeypatch.setattr(fcntl, "flock", flock) + monkeypatch.setattr(os, "write", write) + monkeypatch.setattr(os, "fsync", fsync) + JsonlLifecycleJournal(tmp_path / "lifecycle.jsonl").append(_journal_event()) + assert order.index("lock") < order.index("write") < order.index("fsync") + assert order.index("fsync") < order.index("unlock") + + +def test_journal_fsyncs_parent_directory_only_when_it_creates_the_file( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + directories: list[bool] = [] + real_fsync = os.fsync + + def fsync(descriptor: int) -> None: + directories.append(stat.S_ISDIR(os.fstat(descriptor).st_mode)) + real_fsync(descriptor) + + monkeypatch.setattr(os, "fsync", fsync) + journal = JsonlLifecycleJournal(tmp_path / "state" / "lifecycle.jsonl") + journal.append(_journal_event()) + assert True in directories + directories.clear() + journal.append(_journal_event()) + assert True not in directories + + 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) From b090b910ecc5aab9fde5c3f8428eea1e04b6d6fa Mon Sep 17 00:00:00 2001 From: Angel Parra <607418+aparragithub@users.noreply.github.com> Date: Fri, 7 Aug 2026 20:58:23 -0500 Subject: [PATCH 4/5] fix(lifecycle): fsync the journal directory on every append and lock reads A non-creating append could return before the creating process synced the directory entry, leaving the audit trail unrecoverable after a crash. Sync on every append instead of only the first, and take a shared lock when reading so a concurrent append cannot hand back a truncated final line. --- src/odoo_forge_postgres_docker/lifecycle.py | 38 +++++---- .../test_postgres_docker_lifecycle.py | 79 ++++++++----------- 2 files changed, 55 insertions(+), 62 deletions(-) diff --git a/src/odoo_forge_postgres_docker/lifecycle.py b/src/odoo_forge_postgres_docker/lifecycle.py index 2cfb1aa..67c33b5 100644 --- a/src/odoo_forge_postgres_docker/lifecycle.py +++ b/src/odoo_forge_postgres_docker/lifecycle.py @@ -198,7 +198,9 @@ def __init__(self, path: Path) -> None: def append(self, event: LifecycleJournalEvent) -> LifecycleJournalEvent: self.path.parent.mkdir(mode=0o700, parents=True, exist_ok=True) - descriptor, created = self._open() + 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 @@ -216,26 +218,30 @@ def append(self, event: LifecycleJournalEvent) -> LifecycleJournalEvent: fcntl.flock(descriptor, fcntl.LOCK_UN) finally: os.close(descriptor) - if created: - # Syncing the file alone leaves its directory entry unrecoverable - # after a crash, which would lose the whole audit trail. - _fsync_directory(self.path.parent) + # 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 _open(self) -> tuple[int, bool]: - flags = os.O_WRONLY | os.O_APPEND | os.O_NOFOLLOW - try: - return os.open(self.path, flags | os.O_CREAT | os.O_EXCL, 0o600), True - except FileExistsError: - return os.open(self.path, flags), False - def events(self) -> tuple[LifecycleJournalEvent, ...]: - if not self.path.exists(): + 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 self.path.read_text(encoding="utf-8").splitlines() - if line + LifecycleJournalEvent.model_validate_json(line) for line in payload.splitlines() if line ) diff --git a/tests/adapters/test_postgres_docker_lifecycle.py b/tests/adapters/test_postgres_docker_lifecycle.py index 4b540fe..1f1319d 100644 --- a/tests/adapters/test_postgres_docker_lifecycle.py +++ b/tests/adapters/test_postgres_docker_lifecycle.py @@ -1,8 +1,6 @@ from __future__ import annotations -import fcntl -import os -import stat +import multiprocessing import subprocess from collections.abc import Callable from datetime import timedelta @@ -52,6 +50,12 @@ def _journal_event() -> LifecycleJournalEvent: ) +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", @@ -143,49 +147,32 @@ def test_adapter_rejects_timeouts_without_a_finite_positive_bound(timeout: float PostgresDockerLifecycleAdapter(provider=Mock(), authority=_authority(), timeout=timeout) -def test_journal_holds_an_exclusive_lock_from_first_write_through_fsync( - tmp_path: Path, monkeypatch: pytest.MonkeyPatch -) -> None: - order: list[str] = [] - real_flock, real_write, real_fsync = fcntl.flock, os.write, os.fsync - - def flock(descriptor: int, operation: int) -> None: - order.append("lock" if operation == fcntl.LOCK_EX else "unlock") - real_flock(descriptor, operation) - - def write(descriptor: int, data: bytes) -> int: - order.append("write") - return real_write(descriptor, data) - - def fsync(descriptor: int) -> None: - order.append("fsync") - real_fsync(descriptor) - - monkeypatch.setattr(fcntl, "flock", flock) - monkeypatch.setattr(os, "write", write) - monkeypatch.setattr(os, "fsync", fsync) - JsonlLifecycleJournal(tmp_path / "lifecycle.jsonl").append(_journal_event()) - assert order.index("lock") < order.index("write") < order.index("fsync") - assert order.index("fsync") < order.index("unlock") - - -def test_journal_fsyncs_parent_directory_only_when_it_creates_the_file( - tmp_path: Path, monkeypatch: pytest.MonkeyPatch -) -> None: - directories: list[bool] = [] - real_fsync = os.fsync - - def fsync(descriptor: int) -> None: - directories.append(stat.S_ISDIR(os.fstat(descriptor).st_mode)) - real_fsync(descriptor) - - monkeypatch.setattr(os, "fsync", fsync) - journal = JsonlLifecycleJournal(tmp_path / "state" / "lifecycle.jsonl") - journal.append(_journal_event()) - assert True in directories - directories.clear() - journal.append(_journal_event()) - assert True not in directories +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: From 91ebdd88b1feecb32b5fed92921fe831ee8a9ecd Mon Sep 17 00:00:00 2001 From: Angel Parra <607418+aparragithub@users.noreply.github.com> Date: Fri, 7 Aug 2026 20:58:23 -0500 Subject: [PATCH 5/5] feat(lifecycle): record the applied activity baseline in quarantine history Quarantine history carried the evidence digest and timestamp but not the baseline the expiration decision was made against, so the durable trail could not show why a resource was judged expired. --- src/odoo_forge/resource_lifecycle/service.py | 1 + src/odoo_forge/resource_lifecycle/types.py | 11 +++++++++++ tests/resource_lifecycle/test_service.py | 14 ++++++++++++++ 3 files changed, 26 insertions(+) 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/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,